churn/crates/churn-server/src/event.rs

use std::sync::Arc;

use axum::async_trait;

use churn_domain::LogEvent;
use itertools::Itertools;

use churn_capnp::CapnpPackExt;

use crate::db::Db;

#[derive(Clone)]
pub struct EventService(Arc<dyn EventServiceTrait + Send + Sync + 'static>);

impl EventService {
    pub fn new(db: Db) -> Self {
        Self(Arc::new(DefaultEventService::new(db)))
    }
}

impl std::ops::Deref for EventService {
    type Target = Arc<dyn EventServiceTrait + Send + Sync + 'static>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

struct DefaultEventService {
    db: Db,
}

impl DefaultEventService {
    pub fn new(db: Db) -> Self {
        Self { db }
    }
}

#[async_trait]
pub trait EventServiceTrait {
    async fn append(&self, req: LogEvent) -> anyhow::Result<()>;
    async fn get_from_cursor(&self, cursor: uuid::Uuid) -> anyhow::Result<Vec<LogEvent>>;
    async fn get_from_beginning(&self) -> anyhow::Result<Vec<LogEvent>>;
    async fn get_latest_cursor(&self) -> anyhow::Result<uuid::Uuid>;
}

#[async_trait]
impl EventServiceTrait for DefaultEventService {
    async fn append(&self, req: LogEvent) -> anyhow::Result<()> {
        self.db
            .insert("events_log", &req.id.to_string(), &req.serialize_capnp())
            .await?;

        Ok(())
    }
    async fn get_from_cursor(&self, cursor: uuid::Uuid) -> anyhow::Result<Vec<LogEvent>> {
        let events = self.db.get_all("events_log").await?;

        let events = events
            .iter()
            .flat_map(|e| match LogEvent::deserialize_capnp(e) {
                Ok(o) => Ok(o),
                Err(e) => {
                    tracing::error!("failed to deserialize capnp: {e}");
                    Err(e)
                }
            })
            .sorted_by_key(|i| i.timestamp)
            .skip_while(|item| item.id != cursor)
            .skip(1)
            .collect();

        Ok(events)
    }
    async fn get_from_beginning(&self) -> anyhow::Result<Vec<LogEvent>> {
        let events = self.db.get_all("events_log").await?;

        let events = events
            .iter()
            .map(|x| x.as_slice())
            .flat_map(LogEvent::deserialize_capnp)
            .sorted_by_key(|i| i.timestamp)
            .collect();

        Ok(events)
    }

    async fn get_latest_cursor(&self) -> anyhow::Result<uuid::Uuid> {
        let events = self.db.get_all("events_log").await?;

        let event = events
            .iter()
            .flat_map(|e| match LogEvent::deserialize_capnp(e) {
                Ok(o) => Ok(o),
                Err(e) => {
                    tracing::error!("failed to deserialize capnp: {e}");
                    Err(e)
                }
            })
            .sorted_by_key(|i| i.timestamp)
            .last();

        match event {
            Some(x) => Ok(x.id),
            None => anyhow::bail!("no events found"),
        }
    }
}