diff --git a/Cargo.lock b/Cargo.lock index 7ece1c7..8973967 100644 --- a/Cargo.lock +++ b/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", ] diff --git a/Cargo.toml b/Cargo.toml index f06d52e..9257bb2 100644 --- a/Cargo.toml +++ b/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" diff --git a/resources/migrations/sql/create_tables_001.sql b/resources/migrations/sql/create_tables_001.sql index 6fb4169..1069eb3 100644 --- a/resources/migrations/sql/create_tables_001.sql +++ b/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) +) \ No newline at end of file diff --git a/src/events.rs b/src/events.rs index f3137e9..7391d0d 100644 --- a/src/events.rs +++ b/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), diff --git a/src/webui/mod.rs b/src/webui/mod.rs index 6017864..69130cf 100644 --- a/src/webui/mod.rs +++ b/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::(); - - // 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! { diff --git a/src/webui/worker.rs b/src/webui/worker.rs index b595929..31177d3 100644 --- a/src/webui/worker.rs +++ b/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 { + 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> { + 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, + 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, + } + + 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 { + 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) { + 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(()) + } +}