eventuary_sqlite/
dedupe.rs

1//! SQLite [`DedupeStore`] implementation.
2//!
3//! Keyed by event id (stored as TEXT). `mark_if_new` overrides the
4//! default exists+mark path with a single
5//! `INSERT ... ON CONFLICT DO NOTHING RETURNING` so concurrent dedupe
6//! checks converge without a race. All SQLite work runs in
7//! `spawn_blocking`.
8
9use std::sync::Arc;
10
11use eventuary_core::io::reader::DedupeStore;
12use eventuary_core::{Error, Event, Result};
13
14use crate::database::SqliteConn;
15use crate::relation::SqliteRelationName;
16use crate::schema::{Migration, RelationReplacement};
17
18const DEDUPE_STORE_0001_INIT_SQL: &str = r#"
19CREATE TABLE IF NOT EXISTS {dedupe_keys} (
20    event_id     TEXT NOT NULL PRIMARY KEY,
21    processed_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
22);
23"#;
24
25const DEDUPE_STORE_MIGRATIONS: &[Migration] = &[Migration {
26    name: "0001_init",
27    sql: DEDUPE_STORE_0001_INIT_SQL,
28}];
29
30#[derive(Debug, Clone)]
31pub struct SqliteDedupeStoreConfig {
32    pub relation: SqliteRelationName,
33}
34
35impl Default for SqliteDedupeStoreConfig {
36    fn default() -> Self {
37        Self {
38            relation: SqliteRelationName::new("dedupe_keys").expect("default dedupe relation"),
39        }
40    }
41}
42
43#[derive(Clone)]
44pub struct SqliteDedupeStore {
45    conn: SqliteConn,
46    relation: Arc<String>,
47}
48
49impl SqliteDedupeStore {
50    pub fn new(conn: SqliteConn, config: SqliteDedupeStoreConfig) -> Self {
51        Self {
52            conn,
53            relation: Arc::new(config.relation.render()),
54        }
55    }
56
57    pub fn connect(conn: SqliteConn, config: SqliteDedupeStoreConfig) -> Result<Self> {
58        Self::prepare_schema(&conn, &config)?;
59        Ok(Self::new(conn, config))
60    }
61
62    pub fn prepare_schema(conn: &SqliteConn, config: &SqliteDedupeStoreConfig) -> Result<()> {
63        let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
64        crate::schema::apply_schema(
65            &guard,
66            DEDUPE_STORE_MIGRATIONS,
67            &[RelationReplacement {
68                token: "{dedupe_keys}",
69                relation: &config.relation,
70            }],
71        )
72    }
73
74    pub fn schema_sql(config: &SqliteDedupeStoreConfig) -> String {
75        crate::schema::render_schema_sql(
76            DEDUPE_STORE_MIGRATIONS,
77            &[RelationReplacement {
78                token: "{dedupe_keys}",
79                relation: &config.relation,
80            }],
81        )
82    }
83}
84
85impl DedupeStore for SqliteDedupeStore {
86    async fn exists(&self, event: &Event) -> Result<bool> {
87        let conn = Arc::clone(&self.conn);
88        let relation = Arc::clone(&self.relation);
89        let event_id = event.id().to_string();
90        tokio::task::spawn_blocking(move || {
91            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
92            let sql = format!("SELECT 1 FROM {relation} WHERE event_id = ?1");
93            let found = guard
94                .query_row(&sql, rusqlite::params![event_id], |_| Ok(()))
95                .map(|_| true)
96                .or_else(|e| match e {
97                    rusqlite::Error::QueryReturnedNoRows => Ok(false),
98                    other => Err(other),
99                })
100                .map_err(|e| Error::Store(e.to_string()))?;
101            Ok(found)
102        })
103        .await
104        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
105    }
106
107    async fn mark_processed(&self, event: &Event) -> Result<()> {
108        let conn = Arc::clone(&self.conn);
109        let relation = Arc::clone(&self.relation);
110        let event_id = event.id().to_string();
111        tokio::task::spawn_blocking(move || {
112            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
113            let sql = format!(
114                "INSERT INTO {relation} (event_id) VALUES (?1) \
115                 ON CONFLICT (event_id) DO NOTHING"
116            );
117            guard
118                .execute(&sql, rusqlite::params![event_id])
119                .map_err(|e| Error::Store(e.to_string()))?;
120            Ok(())
121        })
122        .await
123        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
124    }
125
126    async fn mark_if_new(&self, event: &Event) -> Result<bool> {
127        let conn = Arc::clone(&self.conn);
128        let relation = Arc::clone(&self.relation);
129        let event_id = event.id().to_string();
130        tokio::task::spawn_blocking(move || {
131            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
132            let sql = format!(
133                "INSERT INTO {relation} (event_id) VALUES (?1) \
134                 ON CONFLICT (event_id) DO NOTHING \
135                 RETURNING event_id"
136            );
137            let inserted = guard
138                .query_row(&sql, rusqlite::params![event_id], |r| r.get::<_, String>(0))
139                .map(|_| true)
140                .or_else(|e| match e {
141                    rusqlite::Error::QueryReturnedNoRows => Ok(false),
142                    other => Err(other),
143                })
144                .map_err(|e| Error::Store(e.to_string()))?;
145            Ok(inserted)
146        })
147        .await
148        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
149    }
150}
151
152#[cfg(test)]
153mod schema_tests {
154    use super::*;
155
156    #[test]
157    fn schema_sql_contains_expected_table() {
158        let sql = SqliteDedupeStore::schema_sql(&SqliteDedupeStoreConfig::default());
159        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"dedupe_keys\""));
160    }
161}
eventuary_sqlite/dedupe.rs

eventuary_sqlite/
dedupe.rs
