eventuary_sqlite/
buffer_store.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;
18
19#[derive(Debug, Clone)]
20pub struct SqliteBufferStoreConfig {
21    pub relation: SqliteRelationName,
22}
23
24impl Default for SqliteBufferStoreConfig {
25    fn default() -> Self {
26        Self {
27            relation: SqliteRelationName::new("buffer_entries").expect("default buffer relation"),
28        }
29    }
30}
31
32#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
33pub struct SqliteBufferStoreId(i64);
34
35impl SqliteBufferStoreId {
36    pub fn as_i64(&self) -> i64 {
37        self.0
38    }
39}
40
41pub struct SqliteBufferStore<C> {
42    conn: SqliteConn,
43    relation: Arc<String>,
44    _cursor: PhantomData<C>,
45}
46
47impl<C> Clone for SqliteBufferStore<C> {
48    fn clone(&self) -> Self {
49        Self {
50            conn: Arc::clone(&self.conn),
51            relation: Arc::clone(&self.relation),
52            _cursor: PhantomData,
53        }
54    }
55}
56
57impl<C> SqliteBufferStore<C> {
58    pub fn new(conn: SqliteConn, config: SqliteBufferStoreConfig) -> Self {
59        Self {
60            conn,
61            relation: Arc::new(config.relation.render()),
62            _cursor: PhantomData,
63        }
64    }
65}
66
67fn encode_cursor<C: Serialize>(cursor: &C) -> Result<String> {
68    serde_json::to_string(cursor).map_err(|e| Error::Serialization(format!("buffer encode: {e}")))
69}
70
71fn decode_cursor<C: DeserializeOwned>(value: &str) -> Result<C> {
72    serde_json::from_str(value)
73        .map_err(|e| Error::Serialization(format!("buffer decode cursor: {e}")))
74}
75
76fn encode_event(event: &Event) -> Result<String> {
77    let serialized = SerializedEvent::from_event(event)?;
78    serde_json::to_string(&serialized)
79        .map_err(|e| Error::Serialization(format!("buffer encode event: {e}")))
80}
81
82fn decode_event(value: &str) -> Result<Event> {
83    let serialized: SerializedEvent = serde_json::from_str(value)
84        .map_err(|e| Error::Serialization(format!("buffer decode event: {e}")))?;
85    serialized.to_event()
86}
87
88impl<C> BufferStore<C> for SqliteBufferStore<C>
89where
90    C: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
91{
92    type Id = SqliteBufferStoreId;
93
94    async fn push(&self, event: &Event, cursor: &C) -> Result<Self::Id> {
95        let conn = Arc::clone(&self.conn);
96        let relation = Arc::clone(&self.relation);
97        let event_json = encode_event(event)?;
98        let cursor_json = encode_cursor(cursor)?;
99        tokio::task::spawn_blocking(move || {
100            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
101            let sql =
102                format!("INSERT INTO {relation} (event, cursor) VALUES (?1, ?2) RETURNING id");
103            let id: i64 = guard
104                .query_row(&sql, rusqlite::params![event_json, cursor_json], |r| {
105                    r.get(0)
106                })
107                .map_err(|e| Error::Store(e.to_string()))?;
108            Ok(SqliteBufferStoreId(id))
109        })
110        .await
111        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
112    }
113
114    async fn pending(&self) -> Result<Vec<BufferEntry<C, Self::Id>>> {
115        let conn = Arc::clone(&self.conn);
116        let relation = Arc::clone(&self.relation);
117        tokio::task::spawn_blocking(move || {
118            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
119            let sql = format!("SELECT id, event, cursor FROM {relation} ORDER BY id");
120            let mut stmt = guard
121                .prepare(&sql)
122                .map_err(|e| Error::Store(e.to_string()))?;
123            let rows = stmt
124                .query_map([], |r| {
125                    Ok((
126                        r.get::<_, i64>(0)?,
127                        r.get::<_, String>(1)?,
128                        r.get::<_, String>(2)?,
129                    ))
130                })
131                .map_err(|e| Error::Store(e.to_string()))?;
132            let mut out = Vec::new();
133            for row in rows {
134                let (id, event_json, cursor_json) = row.map_err(|e| Error::Store(e.to_string()))?;
135                out.push(BufferEntry {
136                    id: SqliteBufferStoreId(id),
137                    event: decode_event(&event_json)?,
138                    cursor: decode_cursor::<C>(&cursor_json)?,
139                });
140            }
141            Ok(out)
142        })
143        .await
144        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
145    }
146
147    async fn ack(&self, id: &Self::Id) -> Result<()> {
148        let conn = Arc::clone(&self.conn);
149        let relation = Arc::clone(&self.relation);
150        let id_value = id.0;
151        tokio::task::spawn_blocking(move || {
152            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
153            let sql = format!("DELETE FROM {relation} WHERE id = ?1");
154            guard
155                .execute(&sql, rusqlite::params![id_value])
156                .map_err(|e| Error::Store(e.to_string()))?;
157            Ok(())
158        })
159        .await
160        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
161    }
162
163    async fn nack(&self, _id: &Self::Id) -> Result<()> {
164        Ok(())
165    }
166}
eventuary_sqlite/buffer_store.rs

eventuary_sqlite/
buffer_store.rs
