Added decent logging for event processing.

Cargo.lock

@@ -765,6 +765,7 @@ dependencies = [
  "simple_logger",
  "tempfile",
  "tokio",
+ "uuid",
 ]
 
 [[package]]
@@ -2211,9 +2212,9 @@ dependencies = [
 
 [[package]]
 name = "uuid"
-version = "1.2.2"
+version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "422ee0de9031b5b948b97a8fc04e3aa35230001a722ddd27943e0be31564ce4c"
+checksum = "1674845326ee10d37ca60470760d4288a6f80f304007d92e5c53bab78c9cfd79"
 dependencies = [
  "serde",
 ]

Cargo.toml

@@ -33,6 +33,7 @@ async-channel = "1.8.0"
 base64 = "0.21.0"
 r2d2_sqlite = "0.21.0"
 r2d2 = "0.8.10"
+uuid = "1.3.0"
 
 [dependencies.apache-avro]
 version = "0.14.0"

resources/migrations/sql/create_tables_001.sql

@@ -8,7 +8,8 @@ CREATE TABLE IF NOT EXISTS ledger_internal
     cookie_secret TEXT NOT NULL
 );
 
-INSERT INTO ledger_internal(cookie_secret) VALUES ('aaaabbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk');
+INSERT INTO ledger_internal(cookie_secret)
+VALUES ('aaaabbbbccccddddeeeeffffgggghhhhiiiijjjjkkkk');
 
 CREATE TABLE IF NOT EXISTS users
 (
@@ -104,3 +105,25 @@ CREATE TABLE IF NOT EXISTS cashed_cheques
     FOREIGN KEY (payee_account_id) REFERENCES accounts (id),
     CONSTRAINT double_spending UNIQUE (drawer_user_id, cheque_number) -- no double spending
 );
+
+CREATE TABLE IF NOT EXISTS event_processing_statuses
+(
+    value TEXT PRIMARY KEY
+);
+
+INSERT INTO event_processing_statuses(value)
+VALUES ('processing'),
+       ('completed'),
+       ('error');
+
+
+CREATE TABLE IF NOT EXISTS event_processing_log
+(
+    id     TEXT PRIMARY KEY,
+    event  JSON,
+    status TEXT NOT NULL,
+    received_time DATETIME NOT NULL,
+    completed_time DATETIME,
+    error_message TEXT NOT NULL DEFAULT '',
+    FOREIGN KEY (status) REFERENCES event_processing_statuses (value)
+)

src/events.rs

@@ -1,8 +1,8 @@
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 use crate::db::securities::{NewSecurity, SecurityType};
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Serialize, Debug)]
 pub struct CreateNewSecurity {
     pub security_type: String,
     pub symbol: String,
@@ -30,13 +30,13 @@ impl CreateNewSecurity {
     }
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Serialize, Debug)]
 pub struct CreateNewUser {
     pub name: String,
     pub public_key: String,
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Serialize, Debug)]
 pub enum Event {
     CreateNewSecurity(CreateNewSecurity),
     CreateNewUser(CreateNewUser),

src/webui/mod.rs

@@ -8,16 +8,19 @@ use actix_web::{
     web, App, FromRequest, HttpResponse, HttpServer, Responder, Result, Scope,
 };
 
+use crate::event_processor::process_event;
 use async_channel;
 use futures::{pin_mut, select, FutureExt};
 use minijinja;
 use minijinja_autoreload::AutoReloader;
+use rusqlite::Connection;
+use uuid::Uuid;
 
 use crate::events::Event;
 use crate::util::{CustomResult, Error, SQLiteConnectionFactory};
 use crate::webui::services::{creat_db_pool, Services};
 use crate::webui::templates::MiniJinjaRenderer;
-use crate::webui::worker::event_processor_loop;
+use crate::webui::worker::{event_processor_loop, EventProcessor};
 
 mod index_views;
 mod services;
@@ -25,6 +28,14 @@ mod templates;
 mod views;
 mod worker;
 
+struct ProductionEventProcessor;
+
+impl EventProcessor for ProductionEventProcessor {
+    fn process_event(&self, conn: &mut Connection, event: Event) -> CustomResult<()> {
+        process_event(conn, event)
+    }
+}
+
 fn create_auto_reloader(enable_template_autoreload: bool) -> AutoReloader {
     AutoReloader::new(move |notifier| {
         let mut env: minijinja::Environment<'static> = minijinja::Environment::new();
@@ -67,12 +78,7 @@ pub async fn web_ui_main(
     port: u16,
     enable_template_autoreload: bool,
 ) -> CustomResult<()> {
-    let (event_sender, event_receiver) = async_channel::unbounded::<Event>();
-
-    // use r2d2 for connection pools (https://github.com/sfackler/r2d2 &
-    // https://github.com/ivanceras/r2d2-sqlite) use
-    // https://docs.rs/actix-web/4.3.1/actix_web/web/fn.block.html
-    // in handlers.
+    let (event_sender, event_receiver) = async_channel::unbounded::<(Uuid, Event)>();
     let pool = creat_db_pool(&conn_fact.path().unwrap());
     let services = Services::new(pool);
 
@@ -93,7 +99,9 @@ pub async fn web_ui_main(
     .unwrap();
 
     let t1 = server.run().fuse();
-    let t2 = event_processor_loop(conn_fact, event_receiver).fuse();
+
+    let mut conn = conn_fact.connection().unwrap();
+    let t2 = event_processor_loop(&mut conn, &ProductionEventProcessor {}, event_receiver).fuse();
     pin_mut!(t1, t2);
 
     select! {

src/webui/worker.rs

@@ -1,21 +1,219 @@
-use crate::event_processor::process_event;
 use crate::events::Event;
-use crate::util::Error::FuckingError;
-use crate::util::{CustomResult, SQLiteConnectionFactory};
 
-// todo, all the "?" and handle the exceptions for real
+use crate::util::{CustomResult, Error};
+
+use rusqlite::Error::QueryReturnedNoRows;
+use rusqlite::{params, Connection, Result, Row};
+
+use uuid::Uuid;
+
+struct EventLog {
+    id: Uuid,
+    event: Event,
+    status: String,
+}
+
+impl EventLog {
+    pub fn from_row(row: &Row) -> Result<EventLog> {
+        let id: String = row.get("id")?;
+        let event: String = row.get("event")?;
+
+        // todo, do not unwrap
+        Ok(EventLog {
+            id: Uuid::parse_str(id.as_str()).unwrap(),
+            event: serde_json::from_str(event.as_str()).unwrap(),
+            status: row.get("status")?,
+        })
+    }
+}
+
+pub trait EventProcessor {
+    fn process_event(&self, conn: &mut Connection, event: Event) -> CustomResult<()>;
+}
+
+fn log_event_start_of_processing(conn: &mut Connection, id: &Uuid, event: &Event) {
+    conn.execute(
+        "INSERT INTO event_processing_log(id, event, status, received_time)
+                      VALUES (?1, ?2, 'processing', CURRENT_TIMESTAMP)",
+        params![id.to_string(), serde_json::to_string(event).unwrap()],
+    )
+    .unwrap();
+}
+
+fn log_event_end_of_processing(conn: &mut Connection, id: &Uuid, ret: &CustomResult<()>) {
+    match ret {
+        Ok(_) => {
+            conn.execute(
+                "UPDATE event_processing_log SET completed_time = CURRENT_TIMESTAMP,
+                                                              status = 'completed'
+                              WHERE id = ?1",
+                params![id.to_string()],
+            )
+            .unwrap();
+        }
+        Err(e) => {
+            conn.execute(
+                "UPDATE event_processing_log SET completed_time = CURRENT_TIMESTAMP,
+                                                              status = 'error',
+                                                              error_message = ?1
+                              WHERE id = ?2",
+                params![format!("{:?}", e), id.to_string()],
+            )
+            .unwrap();
+        }
+    }
+}
+
+fn _process_event(
+    conn: &mut Connection,
+    event_processor: &dyn EventProcessor,
+    id: Uuid,
+    event: Event,
+) {
+    log_event_start_of_processing(conn, &id, &event);
+    let ret = event_processor.process_event(conn, event);
+    log_event_end_of_processing(conn, &id, &ret);
+}
+
+fn get_event_log(conn: &Connection, id: Uuid) -> CustomResult<Option<EventLog>> {
+    let q = "SELECT * FROM event_processing_log WHERE id = ?";
+    let p = params![id.as_hyphenated().to_string()];
+    let res = conn.query_row(q, p, EventLog::from_row);
+    match res {
+        Ok(event_log) => Ok(Some(event_log)),
+        Err(e) => match e {
+            QueryReturnedNoRows => Ok(None),
+            e => Err(Error::SqlError(e)),
+        },
+    }
+}
+
 pub async fn event_processor_loop(
-    conn_fact: SQLiteConnectionFactory,
-    r: async_channel::Receiver<Event>,
+    conn: &mut Connection,
+    event_processor: &dyn EventProcessor,
+    r: async_channel::Receiver<(Uuid, Event)>,
 ) -> CustomResult<()> {
     loop {
         let res = r.recv().await;
-        let mut conn = conn_fact.connection()?;
         match res {
-            Ok(event) => process_event(&mut conn, event)?,
-            Err(e) => {
-                return Err(FuckingError(e.to_string()));
+            Ok((id, event)) => _process_event(conn, event_processor, id, event),
+            Err(_e) => {
+                return Ok(());
             }
         }
     }
 }
+
+#[cfg(test)]
+pub mod test_worker {
+    use super::*;
+    use crate::events::CreateNewUser;
+    use crate::migration;
+    use futures::{future, TryFutureExt};
+    use std::sync::Mutex;
+
+    struct StatefulEventProcessor {
+        m: Mutex<i32>,
+    }
+
+    impl StatefulEventProcessor {
+        fn new() -> StatefulEventProcessor {
+            return StatefulEventProcessor { m: Mutex::new(0) };
+        }
+
+        fn call_count(&self) -> i32 {
+            let mut m = self.m.lock().unwrap();
+            m.clone()
+        }
+    }
+
+    impl EventProcessor for StatefulEventProcessor {
+        fn process_event(&self, conn: &mut Connection, event: Event) -> CustomResult<()> {
+            let mut m = self.m.lock().unwrap();
+            *m += 1;
+            get_event_processing_result(&event)
+        }
+    }
+
+    fn get_events() -> Vec<Event> {
+        vec![
+            Event::CreateNewUser(CreateNewUser {
+                name: "1".to_string(),
+                public_key: "ok".to_string(),
+            }),
+            Event::CreateNewUser(CreateNewUser {
+                name: "2".to_string(),
+                public_key: "error".to_string(),
+            }),
+            Event::CreateNewUser(CreateNewUser {
+                name: "3".to_string(),
+                public_key: "ok".to_string(),
+            }),
+        ]
+    }
+
+    fn get_event_uuid(e: &Event) -> Uuid {
+        match e {
+            Event::CreateNewUser(nu) => Uuid::from_u128(nu.name.parse().unwrap()),
+            _ => panic!("Test must only use CreateNewUser."),
+        }
+    }
+
+    fn get_event_processing_result(e: &Event) -> CustomResult<()> {
+        match e {
+            Event::CreateNewUser(nu) => match nu.public_key.as_str() {
+                "ok" => Ok(()),
+                e => Err(Error::Testing(e.to_string())),
+            },
+            _ => panic!("Test must only use CreateNewUser."),
+        }
+    }
+
+    async fn send_events(event_sender: async_channel::Sender<(Uuid, Event)>, events: Vec<Event>) {
+        for event in events {
+            event_sender
+                .send((get_event_uuid(&event), event))
+                .await
+                .unwrap();
+        }
+    }
+
+    fn get_connection() -> Connection {
+        let mut conn = Connection::open_in_memory().unwrap();
+        migration::apply_migrations(&mut conn).unwrap();
+        conn
+    }
+
+    #[tokio::test]
+    async fn test_on_close_channel() {
+        let mut ep = StatefulEventProcessor::new();
+        let (event_sender, event_receiver) = async_channel::unbounded::<(Uuid, Event)>();
+        event_sender.close();
+        let r = event_processor_loop(&mut get_connection(), &ep, event_receiver).await;
+        assert!(r.is_ok());
+        assert_eq!(0, ep.call_count());
+    }
+
+    #[tokio::test]
+    async fn test_basic() -> CustomResult<()> {
+        let mut ep = StatefulEventProcessor::new();
+        let mut conn = get_connection();
+        let (event_sender, event_receiver) = async_channel::unbounded::<(Uuid, Event)>();
+        let f1 = send_events(event_sender, get_events());
+        let f2 = event_processor_loop(&mut conn, &ep, event_receiver);
+
+        let (_, r) = future::join(f1, f2).await;
+        assert!(r.is_ok());
+        assert_eq!(3, ep.call_count());
+
+        let event1 = get_event_log(&conn, get_event_uuid(get_events().get(0).unwrap()))?;
+        assert!(event1.is_some());
+        assert_eq!(event1.unwrap().status, "completed");
+
+        let event2 = get_event_log(&conn, get_event_uuid(get_events().get(1).unwrap()))?;
+        assert!(event2.is_some());
+        assert_eq!(event2.unwrap().status, "error");
+
+        Ok(())
+    }
+}