eventuary_sqlite/
database.rs

1use std::path::Path;
2use std::sync::{Arc, Mutex};
3
4use rusqlite::Connection;
5
6use eventuary_core::{Error, Result};
7
8use crate::relation::SqliteRelationName;
9
10pub type SqliteConn = Arc<Mutex<Connection>>;
11
12#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13pub struct Migration {
14    pub filename: &'static str,
15    pub template: &'static str,
16}
17
18impl Migration {
19    pub fn version(&self) -> i64 {
20        migration_version(self.filename)
21    }
22}
23
24const MIGRATION_TEMPLATES: &[Migration] = &[
25    Migration {
26        filename: "0001_init.sql",
27        template: include_str!("../migrations/0001_init.sql"),
28    },
29    Migration {
30        filename: "0002_stores.sql",
31        template: include_str!("../migrations/0002_stores.sql"),
32    },
33];
34
35pub fn migrations() -> &'static [Migration] {
36    MIGRATION_TEMPLATES
37}
38
39#[derive(Debug, Clone)]
40pub struct SqliteDatabaseConfig {
41    pub events_relation: SqliteRelationName,
42    pub offsets_relation: SqliteRelationName,
43    pub multiplexer_completions_relation: SqliteRelationName,
44    pub dedupe_keys_relation: SqliteRelationName,
45    pub buffer_entries_relation: SqliteRelationName,
46    pub watermarks_relation: SqliteRelationName,
47}
48
49impl Default for SqliteDatabaseConfig {
50    fn default() -> Self {
51        Self {
52            events_relation: SqliteRelationName::new("events").expect("default events relation"),
53            offsets_relation: SqliteRelationName::new("consumer_offsets")
54                .expect("default offsets relation"),
55            multiplexer_completions_relation: SqliteRelationName::new("multiplexer_completions")
56                .expect("default multiplexer relation"),
57            dedupe_keys_relation: SqliteRelationName::new("dedupe_keys")
58                .expect("default dedupe relation"),
59            buffer_entries_relation: SqliteRelationName::new("buffer_entries")
60                .expect("default buffer relation"),
61            watermarks_relation: SqliteRelationName::new("watermarks")
62                .expect("default watermarks relation"),
63        }
64    }
65}
66
67pub fn render_migration_sql(migration: &Migration, config: &SqliteDatabaseConfig) -> String {
68    migration
69        .template
70        .replace("{events}", &config.events_relation.render())
71        .replace("{offsets}", &config.offsets_relation.render())
72        .replace(
73            "{multiplexer_completions}",
74            &config.multiplexer_completions_relation.render(),
75        )
76        .replace("{dedupe_keys}", &config.dedupe_keys_relation.render())
77        .replace("{buffer_entries}", &config.buffer_entries_relation.render())
78        .replace("{watermarks}", &config.watermarks_relation.render())
79}
80
81pub fn render_schema_sql(config: &SqliteDatabaseConfig) -> String {
82    let mut sql = String::new();
83    for migration in migrations() {
84        sql.push_str(&render_migration_sql(migration, config));
85        if !sql.ends_with('\n') {
86            sql.push('\n');
87        }
88    }
89    sql
90}
91
92pub fn schema_sql() -> String {
93    render_schema_sql(&SqliteDatabaseConfig::default())
94}
95
96pub struct SqliteDatabase {
97    conn: SqliteConn,
98    config: SqliteDatabaseConfig,
99}
100
101impl SqliteDatabase {
102    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
103        let conn = Connection::open(path).map_err(|e| Error::Store(e.to_string()))?;
104        Self::init(conn, SqliteDatabaseConfig::default())
105    }
106
107    pub fn open_in_memory() -> Result<Self> {
108        let conn = Connection::open_in_memory().map_err(|e| Error::Store(e.to_string()))?;
109        Self::init(conn, SqliteDatabaseConfig::default())
110    }
111
112    pub fn open_with_config(path: impl AsRef<Path>, config: SqliteDatabaseConfig) -> Result<Self> {
113        let conn = Connection::open(path).map_err(|e| Error::Store(e.to_string()))?;
114        Self::init(conn, config)
115    }
116
117    pub fn open_in_memory_with_config(config: SqliteDatabaseConfig) -> Result<Self> {
118        let conn = Connection::open_in_memory().map_err(|e| Error::Store(e.to_string()))?;
119        Self::init(conn, config)
120    }
121
122    fn init(conn: Connection, config: SqliteDatabaseConfig) -> Result<Self> {
123        let _: String = conn
124            .pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get(0))
125            .map_err(|e| Error::Store(e.to_string()))?;
126        apply_migrations(&conn, &config)?;
127
128        Ok(Self {
129            conn: Arc::new(Mutex::new(conn)),
130            config,
131        })
132    }
133
134    pub fn conn(&self) -> SqliteConn {
135        Arc::clone(&self.conn)
136    }
137
138    pub fn config(&self) -> &SqliteDatabaseConfig {
139        &self.config
140    }
141}
142
143fn apply_migrations(conn: &Connection, config: &SqliteDatabaseConfig) -> Result<()> {
144    // Every statement in the rendered migration is `CREATE TABLE IF NOT
145    // EXISTS` / `CREATE INDEX IF NOT EXISTS`, so applying every migration
146    // on every open is safe and avoids skipping configured-relation
147    // creation when the previous opener wrote `PRAGMA user_version`
148    // against a different relation pair.
149    for migration in migrations() {
150        let sql = render_migration_sql(migration, config);
151        conn.execute_batch(&sql)
152            .map_err(|e| Error::Store(e.to_string()))?;
153    }
154    Ok(())
155}
156
157fn migration_version(filename: &str) -> i64 {
158    let Some((version, _)) = filename.split_once('_') else {
159        panic!("migration filename must start with a numeric version prefix")
160    };
161    version
162        .parse()
163        .expect("migration filename version prefix must be numeric")
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    #[test]
171    fn open_in_memory_creates_schema() {
172        let db = SqliteDatabase::open_in_memory().unwrap();
173        let conn = db.conn();
174        let guard = conn.lock().unwrap();
175        let count: i64 = guard
176            .query_row(
177                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name IN ('events', 'consumer_offsets')",
178                [],
179                |row| row.get(0),
180            )
181            .unwrap();
182        assert_eq!(count, 2);
183    }
184
185    #[test]
186    fn schema_idempotent() {
187        let db = SqliteDatabase::open_in_memory().unwrap();
188        let conn = db.conn();
189        let guard = conn.lock().unwrap();
190        let sql = render_migration_sql(&migrations()[0], &SqliteDatabaseConfig::default());
191        guard.execute_batch(&sql).unwrap();
192    }
193
194    #[test]
195    fn schema_sql_is_available_for_manual_migrations() {
196        let sql = schema_sql();
197        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"events\""));
198        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"consumer_offsets\""));
199    }
200
201    #[test]
202    fn opens_idempotently_with_alternate_relations_after_default_relations() {
203        // First, open with default relations so `events` + `consumer_offsets` exist.
204        let db = SqliteDatabase::open_in_memory().unwrap();
205        let conn = db.conn();
206
207        // Reuse the same connection with a custom relation pair. The
208        // alternate tables must be created, not skipped by a global
209        // migration version.
210        let alt = SqliteDatabaseConfig {
211            events_relation: SqliteRelationName::new("alt_events").unwrap(),
212            offsets_relation: SqliteRelationName::new("alt_offsets").unwrap(),
213            ..SqliteDatabaseConfig::default()
214        };
215        let sql = render_migration_sql(&migrations()[0], &alt);
216        {
217            let guard = conn.lock().unwrap();
218            guard.execute_batch(&sql).unwrap();
219        }
220        let guard = conn.lock().unwrap();
221        let alt_count: i64 = guard
222            .query_row(
223                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name IN ('alt_events', 'alt_offsets')",
224                [],
225                |row| row.get(0),
226            )
227            .unwrap();
228        assert_eq!(alt_count, 2, "alternate relations must be created");
229    }
230
231    #[test]
232    fn schema_sql_uses_configured_relations() {
233        let config = SqliteDatabaseConfig {
234            events_relation: SqliteRelationName::new("custom_events").unwrap(),
235            offsets_relation: SqliteRelationName::new("custom_offsets").unwrap(),
236            ..SqliteDatabaseConfig::default()
237        };
238        let sql = render_schema_sql(&config);
239        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"custom_events\""));
240        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"custom_offsets\""));
241    }
242}
eventuary_sqlite/database.rs

eventuary_sqlite/
database.rs
