eventuary_sqlite/
writer.rs

1use std::sync::Arc;
2
3use eventuary_core::io::Writer;
4use eventuary_core::{Error, Event, Result, SerializedEvent};
5
6use crate::database::SqliteConn;
7use crate::relation::SqliteRelationName;
8
9#[derive(Debug, Clone)]
10pub struct SqliteWriterConfig {
11    pub events_relation: SqliteRelationName,
12}
13
14impl Default for SqliteWriterConfig {
15    fn default() -> Self {
16        Self {
17            events_relation: SqliteRelationName::new("events").expect("default events relation"),
18        }
19    }
20}
21
22pub struct SqliteWriter {
23    conn: SqliteConn,
24    insert_sql: Arc<String>,
25}
26
27impl SqliteWriter {
28    pub fn new(conn: SqliteConn) -> Self {
29        Self::new_with_config(conn, SqliteWriterConfig::default())
30    }
31
32    pub fn new_with_config(conn: SqliteConn, config: SqliteWriterConfig) -> Self {
33        let insert_sql = format!(
34            "INSERT INTO {events} (id, organization, namespace, topic, event_key, payload, content_type, metadata, timestamp, version, parent_id, correlation_id, causation_id)
35             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
36            events = config.events_relation.render(),
37        );
38        Self {
39            conn,
40            insert_sql: Arc::new(insert_sql),
41        }
42    }
43}
44
45impl Writer for SqliteWriter {
46    async fn write(&self, event: &Event) -> Result<()> {
47        let conn = Arc::clone(&self.conn);
48        let event = event.clone();
49        let sql = Arc::clone(&self.insert_sql);
50        tokio::task::spawn_blocking(move || {
51            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
52            insert_event(&guard, &sql, &event)
53        })
54        .await
55        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
56    }
57
58    async fn write_all(&self, events: &[Event]) -> Result<()> {
59        if events.is_empty() {
60            return Ok(());
61        }
62        let conn = Arc::clone(&self.conn);
63        let events = events.to_vec();
64        let sql = Arc::clone(&self.insert_sql);
65        tokio::task::spawn_blocking(move || {
66            let mut guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
67            let tx = guard
68                .transaction()
69                .map_err(|e| Error::Store(e.to_string()))?;
70            for event in &events {
71                insert_event(&tx, &sql, event)?;
72            }
73            tx.commit().map_err(|e| Error::Store(e.to_string()))?;
74            Ok(())
75        })
76        .await
77        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
78    }
79}
80
81fn insert_event(conn: &rusqlite::Connection, sql: &str, event: &Event) -> Result<()> {
82    let serialized = SerializedEvent::from_event(event)?;
83    let content_type = serialized.payload.content_type().to_string();
84    let payload = serde_json::to_string(&serialized.payload)
85        .map_err(|e| Error::Store(format!("encode payload: {e}")))?;
86    let metadata = serde_json::to_string(&serialized.metadata)
87        .map_err(|e| Error::Store(format!("encode metadata: {e}")))?;
88    conn.execute(
89        sql,
90        rusqlite::params![
91            serialized.id.to_string(),
92            serialized.organization,
93            serialized.namespace,
94            serialized.topic,
95            serialized.key,
96            payload,
97            content_type,
98            metadata,
99            serialized.timestamp.to_rfc3339(),
100            serialized.version as i64,
101            serialized.parent_id.map(|id| id.to_string()),
102            serialized.correlation_id,
103            serialized.causation_id,
104        ],
105    )
106    .map_err(|e| Error::Store(e.to_string()))?;
107    Ok(())
108}
eventuary_sqlite/writer.rs

eventuary_sqlite/
writer.rs
