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;
16
17#[derive(Debug, Clone)]
18pub struct SqliteMultiplexerStoreConfig {
19    pub relation: SqliteRelationName,
20}
21
22impl Default for SqliteMultiplexerStoreConfig {
23    fn default() -> Self {
24        Self {
25            relation: SqliteRelationName::new("multiplexer_completions")
26                .expect("default multiplexer relation"),
27        }
28    }
29}
30
31#[derive(Clone)]
32pub struct SqliteMultiplexerStore {
33    conn: SqliteConn,
34    relation: Arc<String>,
35}
36
37impl SqliteMultiplexerStore {
38    pub fn new(conn: SqliteConn, config: SqliteMultiplexerStoreConfig) -> Self {
39        Self {
40            conn,
41            relation: Arc::new(config.relation.render()),
42        }
43    }
44}
45
46impl MultiplexerStore for SqliteMultiplexerStore {
47    async fn is_completed(&self, key: &MultiplexerKey) -> Result<bool> {
48        let conn = Arc::clone(&self.conn);
49        let relation = Arc::clone(&self.relation);
50        let event_id = key.event_id.to_string();
51        let subscriber_id = key.subscriber_id.as_str().to_owned();
52        tokio::task::spawn_blocking(move || {
53            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
54            let sql = format!(
55                "SELECT 1 FROM {relation} \
56                 WHERE event_id = ?1 AND subscriber_id = ?2"
57            );
58            let row = guard
59                .query_row(&sql, rusqlite::params![event_id, subscriber_id], |_| Ok(()))
60                .map(|_| true)
61                .or_else(|e| match e {
62                    rusqlite::Error::QueryReturnedNoRows => Ok(false),
63                    other => Err(other),
64                })
65                .map_err(|e| Error::Store(e.to_string()))?;
66            Ok(row)
67        })
68        .await
69        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
70    }
71
72    async fn mark_completed(&self, key: &MultiplexerKey) -> Result<()> {
73        let conn = Arc::clone(&self.conn);
74        let relation = Arc::clone(&self.relation);
75        let event_id = key.event_id.to_string();
76        let subscriber_id = key.subscriber_id.as_str().to_owned();
77        tokio::task::spawn_blocking(move || {
78            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
79            let sql = format!(
80                "INSERT INTO {relation} (event_id, subscriber_id) \
81                 VALUES (?1, ?2) \
82                 ON CONFLICT (event_id, subscriber_id) DO NOTHING"
83            );
84            guard
85                .execute(&sql, rusqlite::params![event_id, subscriber_id])
86                .map_err(|e| Error::Store(e.to_string()))?;
87            Ok(())
88        })
89        .await
90        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
91    }
92}
eventuary_sqlite/multiplexer_store.rs

eventuary_sqlite/
multiplexer_store.rs
