eventuary_sqlite/
multiplexer_store.rs

1//! SQLite [`MultiplexerStore`] implementation.
2//!
3//! Records `(event_id, subscriber_id)` completion rows in the
4//! configured relation. Event ids are stored as TEXT (UUID string)
5//! since SQLite has no native UUID type. `mark_completed` uses
6//! `INSERT ... ON CONFLICT DO NOTHING` so concurrent or redelivered
7//! calls converge. All SQLite work runs in `spawn_blocking`.
8
9use std::sync::Arc;
10
11use eventuary_core::io::handler::{MultiplexerKey, MultiplexerStore};
12use eventuary_core::{Error, Result};
13
14use crate::database::SqliteConn;
15use crate::relation::SqliteRelationName;
16use crate::schema::{Migration, RelationReplacement};
17
18const MULTIPLEXER_STORE_0001_INIT_SQL: &str = r#"
19CREATE TABLE IF NOT EXISTS {multiplexer_completions} (
20    event_id      TEXT NOT NULL,
21    subscriber_id TEXT NOT NULL,
22    completed_at  TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
23    PRIMARY KEY (event_id, subscriber_id)
24);
25"#;
26
27const MULTIPLEXER_STORE_MIGRATIONS: &[Migration] = &[Migration {
28    name: "0001_init",
29    sql: MULTIPLEXER_STORE_0001_INIT_SQL,
30}];
31
32#[derive(Debug, Clone)]
33pub struct SqliteMultiplexerStoreConfig {
34    pub relation: SqliteRelationName,
35}
36
37impl Default for SqliteMultiplexerStoreConfig {
38    fn default() -> Self {
39        Self {
40            relation: SqliteRelationName::new("multiplexer_completions")
41                .expect("default multiplexer relation"),
42        }
43    }
44}
45
46#[derive(Clone)]
47pub struct SqliteMultiplexerStore {
48    conn: SqliteConn,
49    relation: Arc<String>,
50}
51
52impl SqliteMultiplexerStore {
53    pub fn new(conn: SqliteConn, config: SqliteMultiplexerStoreConfig) -> Self {
54        Self {
55            conn,
56            relation: Arc::new(config.relation.render()),
57        }
58    }
59
60    pub fn connect(conn: SqliteConn, config: SqliteMultiplexerStoreConfig) -> Result<Self> {
61        Self::prepare_schema(&conn, &config)?;
62        Ok(Self::new(conn, config))
63    }
64
65    pub fn prepare_schema(conn: &SqliteConn, config: &SqliteMultiplexerStoreConfig) -> Result<()> {
66        let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
67        crate::schema::apply_schema(
68            &guard,
69            MULTIPLEXER_STORE_MIGRATIONS,
70            &[RelationReplacement {
71                token: "{multiplexer_completions}",
72                relation: &config.relation,
73            }],
74        )
75    }
76
77    pub fn schema_sql(config: &SqliteMultiplexerStoreConfig) -> String {
78        crate::schema::render_schema_sql(
79            MULTIPLEXER_STORE_MIGRATIONS,
80            &[RelationReplacement {
81                token: "{multiplexer_completions}",
82                relation: &config.relation,
83            }],
84        )
85    }
86}
87
88impl MultiplexerStore for SqliteMultiplexerStore {
89    async fn is_completed(&self, key: &MultiplexerKey) -> Result<bool> {
90        let conn = Arc::clone(&self.conn);
91        let relation = Arc::clone(&self.relation);
92        let event_id = key.event_id.to_string();
93        let subscriber_id = key.subscriber_id.as_str().to_owned();
94        tokio::task::spawn_blocking(move || {
95            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
96            let sql = format!(
97                "SELECT 1 FROM {relation} \
98                 WHERE event_id = ?1 AND subscriber_id = ?2"
99            );
100            let row = guard
101                .query_row(&sql, rusqlite::params![event_id, subscriber_id], |_| Ok(()))
102                .map(|_| true)
103                .or_else(|e| match e {
104                    rusqlite::Error::QueryReturnedNoRows => Ok(false),
105                    other => Err(other),
106                })
107                .map_err(|e| Error::Store(e.to_string()))?;
108            Ok(row)
109        })
110        .await
111        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
112    }
113
114    async fn mark_completed(&self, key: &MultiplexerKey) -> Result<()> {
115        let conn = Arc::clone(&self.conn);
116        let relation = Arc::clone(&self.relation);
117        let event_id = key.event_id.to_string();
118        let subscriber_id = key.subscriber_id.as_str().to_owned();
119        tokio::task::spawn_blocking(move || {
120            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
121            let sql = format!(
122                "INSERT INTO {relation} (event_id, subscriber_id) \
123                 VALUES (?1, ?2) \
124                 ON CONFLICT (event_id, subscriber_id) DO NOTHING"
125            );
126            guard
127                .execute(&sql, rusqlite::params![event_id, subscriber_id])
128                .map_err(|e| Error::Store(e.to_string()))?;
129            Ok(())
130        })
131        .await
132        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
133    }
134}
135
136#[cfg(test)]
137mod schema_tests {
138    use super::*;
139
140    #[test]
141    fn schema_sql_contains_expected_table() {
142        let sql = SqliteMultiplexerStore::schema_sql(&SqliteMultiplexerStoreConfig::default());
143        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"multiplexer_completions\""));
144    }
145}
eventuary_sqlite/multiplexer_store.rs

eventuary_sqlite/
multiplexer_store.rs
