eventuary_sqlite/
watermark_store.rs

1//! SQLite [`WatermarkStore`] implementation.
2//!
3//! Persists per-key high-water timestamps. Timestamps round-trip
4//! through RFC3339 TEXT. `save_watermark` upserts so redelivered or
5//! out-of-order saves converge. All SQLite work runs in
6//! `spawn_blocking`.
7
8use std::sync::Arc;
9
10use chrono::{DateTime, Utc};
11
12use eventuary_core::io::reader::WatermarkStore;
13use eventuary_core::{Error, Result};
14
15use crate::database::SqliteConn;
16use crate::relation::SqliteRelationName;
17use crate::schema::{Migration, RelationReplacement};
18
19const WATERMARK_STORE_0001_INIT_SQL: &str = r#"
20CREATE TABLE IF NOT EXISTS {watermarks} (
21    key        TEXT NOT NULL PRIMARY KEY,
22    ts         TEXT NOT NULL,
23    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
24);
25"#;
26
27const WATERMARK_STORE_MIGRATIONS: &[Migration] = &[Migration {
28    name: "0001_init",
29    sql: WATERMARK_STORE_0001_INIT_SQL,
30}];
31
32#[derive(Debug, Clone)]
33pub struct SqliteWatermarkStoreConfig {
34    pub relation: SqliteRelationName,
35}
36
37impl Default for SqliteWatermarkStoreConfig {
38    fn default() -> Self {
39        Self {
40            relation: SqliteRelationName::new("watermarks").expect("default watermarks relation"),
41        }
42    }
43}
44
45#[derive(Clone)]
46pub struct SqliteWatermarkStore {
47    conn: SqliteConn,
48    relation: Arc<String>,
49}
50
51impl SqliteWatermarkStore {
52    pub fn new(conn: SqliteConn, config: SqliteWatermarkStoreConfig) -> Self {
53        Self {
54            conn,
55            relation: Arc::new(config.relation.render()),
56        }
57    }
58
59    pub fn connect(conn: SqliteConn, config: SqliteWatermarkStoreConfig) -> Result<Self> {
60        Self::prepare_schema(&conn, &config)?;
61        Ok(Self::new(conn, config))
62    }
63
64    pub fn prepare_schema(conn: &SqliteConn, config: &SqliteWatermarkStoreConfig) -> Result<()> {
65        let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
66        crate::schema::apply_schema(
67            &guard,
68            WATERMARK_STORE_MIGRATIONS,
69            &[RelationReplacement {
70                token: "{watermarks}",
71                relation: &config.relation,
72            }],
73        )
74    }
75
76    pub fn schema_sql(config: &SqliteWatermarkStoreConfig) -> String {
77        crate::schema::render_schema_sql(
78            WATERMARK_STORE_MIGRATIONS,
79            &[RelationReplacement {
80                token: "{watermarks}",
81                relation: &config.relation,
82            }],
83        )
84    }
85}
86
87impl WatermarkStore for SqliteWatermarkStore {
88    async fn load_watermark(&self, key: &str) -> Result<Option<DateTime<Utc>>> {
89        let conn = Arc::clone(&self.conn);
90        let relation = Arc::clone(&self.relation);
91        let key = key.to_owned();
92        tokio::task::spawn_blocking(move || {
93            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
94            let sql = format!("SELECT ts FROM {relation} WHERE key = ?1");
95            let ts_str = guard
96                .query_row(&sql, rusqlite::params![key], |r| r.get::<_, String>(0))
97                .map(Some)
98                .or_else(|e| match e {
99                    rusqlite::Error::QueryReturnedNoRows => Ok(None),
100                    other => Err(other),
101                })
102                .map_err(|e| Error::Store(e.to_string()))?;
103            match ts_str {
104                Some(s) => {
105                    let ts = DateTime::parse_from_rfc3339(&s)
106                        .map_err(|e| Error::Serialization(format!("watermark decode: {e}")))?
107                        .with_timezone(&Utc);
108                    Ok(Some(ts))
109                }
110                None => Ok(None),
111            }
112        })
113        .await
114        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
115    }
116
117    async fn save_watermark(&self, key: &str, ts: DateTime<Utc>) -> Result<()> {
118        let conn = Arc::clone(&self.conn);
119        let relation = Arc::clone(&self.relation);
120        let key = key.to_owned();
121        let ts_str = ts.to_rfc3339();
122        tokio::task::spawn_blocking(move || {
123            let guard = conn.lock().map_err(|e| Error::Store(e.to_string()))?;
124            let sql = format!(
125                "INSERT INTO {relation} (key, ts) VALUES (?1, ?2) \
126                 ON CONFLICT (key) DO UPDATE SET ts = excluded.ts, updated_at = CURRENT_TIMESTAMP"
127            );
128            guard
129                .execute(&sql, rusqlite::params![key, ts_str])
130                .map_err(|e| Error::Store(e.to_string()))?;
131            Ok(())
132        })
133        .await
134        .map_err(|e| Error::Store(format!("blocking task panicked: {e}")))?
135    }
136}
137
138#[cfg(test)]
139mod schema_tests {
140    use super::*;
141
142    #[test]
143    fn schema_sql_contains_expected_table() {
144        let sql = SqliteWatermarkStore::schema_sql(&SqliteWatermarkStoreConfig::default());
145        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"watermarks\""));
146    }
147}
eventuary_sqlite/watermark_store.rs

eventuary_sqlite/
watermark_store.rs
