escarmouche/src/webui/worker.rs

use crate::events::Event;

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: &mut Connection,
    event_processor: &dyn EventProcessor,
    r: async_channel::Receiver<(Uuid, Event)>,
) -> CustomResult<()> {
    loop {
        let res = r.recv().await;
        match res {
            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;
    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 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 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 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(())
    }
}