eventuary_sqlite/
buffer.rs

1//! SQLite [`BufferStore`] implementation.
2//!
3//! Persists buffered events plus their cursor as JSON TEXT. The
4//! generic cursor `C` must round-trip via `serde_json`. `pending`
5//! returns the current snapshot ordered by id; `nack` is a no-op. All
6//! SQLite work runs in `spawn_blocking`.
7
8use std::marker::PhantomData;
9use std::sync::Arc;
10
11use serde::{Serialize, de::DeserializeOwned};
12
13use eventuary_core::io::reader::{BufferEntry, BufferStore};
14use eventuary_core::{Error, Event, Result, SerializedEvent};
15
16use crate::database::SqliteConn;
17use crate::relation::SqliteRelationName;
18use crate::schema::{Migration, RelationReplacement};
19
20const BUFFER_STORE_0001_INIT_SQL: &str = r#"
21CREATE TABLE IF NOT EXISTS {buffer_entries} (
22    id        INTEGER PRIMARY KEY AUTOINCREMENT,
23    event     TEXT    NOT NULL,
24    cursor    TEXT    NOT NULL,
25    pushed_at TEXT    NOT NULL DEFAULT CURRENT_TIMESTAMP
26);
27
28CREATE INDEX IF NOT EXISTS idx_buffer_entries_pushed_at ON {buffer_entries} (pushed_at);
29"#;
30
31const BUFFER_STORE_MIGRATIONS: &[Migration] = &[Migration {
32    name: "0001_init",
33    sql: BUFFER_STORE_0001_INIT_SQL,
34}];
35
36#[derive(Debug, Clone)]
37pub struct SqliteBufferStoreConfig {
38    pub relation: SqliteRelationName,
39}
40
41impl Default for SqliteBufferStoreConfig {
42    fn default() -> Self {
43        Self {
44            relation: SqliteRelationName::new("buffer_entries").expect("default buffer relation"),
45        }
46    }
47}
48
49#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
50pub struct SqliteBufferStoreId(i64);
51
52impl SqliteBufferStoreId {
53    pub fn as_i64(&self) -> i64 {
54        self.0
55    }
56}
57
58pub struct SqliteBufferStore<C> {
59    conn: SqliteConn,
60    relation: Arc<String>,
61    _cursor: PhantomData<C>,
62}
63
64impl<C> Clone for SqliteBufferStore<C> {
65    fn clone(&self) -> Self {
66        Self {
67            conn: Arc::clone(&self.conn),
68            relation: Arc::clone(&self.relation),
69            _cursor: PhantomData,
70        }
71    }
72}
73
74impl<C> SqliteBufferStore<C> {
75    pub fn new(conn: SqliteConn, config: SqliteBufferStoreConfig) -> Self {
76        Self {
77            conn,
78            relation: Arc::new(config.relation.render()),
79            _cursor: PhantomData,
80        }
81    }
82
83    pub fn connect(conn: SqliteConn, config: SqliteBufferStoreConfig) -> Result<Self> {
84        Self::prepare_schema(&conn, &config)?;
85        Ok(Self::new(conn, config))
86    }
87
88    pub fn prepare_schema(conn: &SqliteConn, config: &SqliteBufferStoreConfig) -> Result<()> {
89        let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
90        crate::schema::apply_schema(
91            &guard,
92            BUFFER_STORE_MIGRATIONS,
93            &[RelationReplacement {
94                token: "{buffer_entries}",
95                relation: &config.relation,
96            }],
97        )
98    }
99
100    pub fn schema_sql(config: &SqliteBufferStoreConfig) -> String {
101        crate::schema::render_schema_sql(
102            BUFFER_STORE_MIGRATIONS,
103            &[RelationReplacement {
104                token: "{buffer_entries}",
105                relation: &config.relation,
106            }],
107        )
108    }
109}
110
111fn encode_cursor<C: Serialize>(cursor: &C) -> Result<String> {
112    serde_json::to_string(cursor).map_err(|e| Error::Serialization(format!("buffer encode: {e}")))
113}
114
115fn decode_cursor<C: DeserializeOwned>(value: &str) -> Result<C> {
116    serde_json::from_str(value)
117        .map_err(|e| Error::Serialization(format!("buffer decode cursor: {e}")))
118}
119
120fn encode_event(event: &Event) -> Result<String> {
121    let serialized = SerializedEvent::from_event(event)?;
122    serde_json::to_string(&serialized)
123        .map_err(|e| Error::Serialization(format!("buffer encode event: {e}")))
124}
125
126fn decode_event(value: &str) -> Result<Event> {
127    let serialized: SerializedEvent = serde_json::from_str(value)
128        .map_err(|e| Error::Serialization(format!("buffer decode event: {e}")))?;
129    serialized.to_event()
130}
131
132impl<C> BufferStore<C> for SqliteBufferStore<C>
133where
134    C: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
135{
136    type Id = SqliteBufferStoreId;
137
138    async fn push(&self, event: &Event, cursor: &C) -> Result<Self::Id> {
139        let conn = Arc::clone(&self.conn);
140        let relation = Arc::clone(&self.relation);
141        let event_json = encode_event(event)?;
142        let cursor_json = encode_cursor(cursor)?;
143        tokio::task::spawn_blocking(move || {
144            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
145            let sql =
146                format!("INSERT INTO {relation} (event, cursor) VALUES (?1, ?2) RETURNING id");
147            let id: i64 = guard
148                .query_row(&sql, rusqlite::params![event_json, cursor_json], |r| {
149                    r.get(0)
150                })
151                .map_err(|e| Error::Store(e.to_string()))?;
152            Ok(SqliteBufferStoreId(id))
153        })
154        .await
155        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
156    }
157
158    async fn pending(&self) -> Result<Vec<BufferEntry<C, Self::Id>>> {
159        let conn = Arc::clone(&self.conn);
160        let relation = Arc::clone(&self.relation);
161        tokio::task::spawn_blocking(move || {
162            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
163            let sql = format!("SELECT id, event, cursor FROM {relation} ORDER BY id");
164            let mut stmt = guard
165                .prepare(&sql)
166                .map_err(|e| Error::Store(e.to_string()))?;
167            let rows = stmt
168                .query_map([], |r| {
169                    Ok((
170                        r.get::<_, i64>(0)?,
171                        r.get::<_, String>(1)?,
172                        r.get::<_, String>(2)?,
173                    ))
174                })
175                .map_err(|e| Error::Store(e.to_string()))?;
176            let mut out = Vec::new();
177            for row in rows {
178                let (id, event_json, cursor_json) = row.map_err(|e| Error::Store(e.to_string()))?;
179                out.push(BufferEntry {
180                    id: SqliteBufferStoreId(id),
181                    event: decode_event(&event_json)?,
182                    cursor: decode_cursor::<C>(&cursor_json)?,
183                });
184            }
185            Ok(out)
186        })
187        .await
188        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
189    }
190
191    async fn ack(&self, id: &Self::Id) -> Result<()> {
192        let conn = Arc::clone(&self.conn);
193        let relation = Arc::clone(&self.relation);
194        let id_value = id.0;
195        tokio::task::spawn_blocking(move || {
196            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
197            let sql = format!("DELETE FROM {relation} WHERE id = ?1");
198            guard
199                .execute(&sql, rusqlite::params![id_value])
200                .map_err(|e| Error::Store(e.to_string()))?;
201            Ok(())
202        })
203        .await
204        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
205    }
206
207    async fn nack(&self, _id: &Self::Id) -> Result<()> {
208        Ok(())
209    }
210}
211
212#[cfg(test)]
213mod schema_tests {
214    use super::*;
215
216    #[test]
217    fn schema_sql_contains_expected_table() {
218        let sql = SqliteBufferStore::<crate::reader::SqliteCursor>::schema_sql(
219            &SqliteBufferStoreConfig::default(),
220        );
221        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"buffer_entries\""));
222    }
223}
eventuary_sqlite/buffer.rs

eventuary_sqlite/
buffer.rs
