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

109 lines
2.8 KiB
Rust
Raw Normal View History

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"),
}
}
}