cratestack_client_store_sqlite/
lib.rs

1use std::fs;
2use std::path::{Path, PathBuf};
3use std::sync::Mutex;
4
5use chrono::{DateTime, Utc};
6use cratestack_client_rust::{
7    ClientError, ClientStateStore, PersistedClientState, RequestJournalEntry,
8};
9use rusqlite::{Connection, OptionalExtension, params};
10
11const SQLITE_SCHEMA_VERSION: u32 = 1;
12
13pub struct SqliteStateStore {
14    connection: Mutex<Connection>,
15    path: PathBuf,
16}
17
18impl SqliteStateStore {
19    pub fn open(path: impl Into<PathBuf>) -> Result<Self, ClientError> {
20        let path = path.into();
21        if let Some(parent) = path.parent() {
22            fs::create_dir_all(parent).map_err(|error| {
23                ClientError::State(format!(
24                    "failed to create SQLite state directory {}: {error}",
25                    parent.display()
26                ))
27            })?;
28        }
29        let connection = Connection::open(&path).map_err(|error| {
30            ClientError::State(format!(
31                "failed to open SQLite state store {}: {error}",
32                path.display()
33            ))
34        })?;
35        bootstrap(&connection, SQLITE_SCHEMA_VERSION)?;
36
37        Ok(Self {
38            connection: Mutex::new(connection),
39            path,
40        })
41    }
42
43    pub fn path(&self) -> &Path {
44        &self.path
45    }
46}
47
48impl ClientStateStore for SqliteStateStore {
49    fn load(&self) -> Result<PersistedClientState, ClientError> {
50        let connection = self.connection.lock().map_err(|error| {
51            ClientError::State(format!("failed to lock SQLite state store: {error}"))
52        })?;
53        load_state(&connection)
54    }
55
56    fn save(&self, state: &PersistedClientState) -> Result<(), ClientError> {
57        let mut connection = self.connection.lock().map_err(|error| {
58            ClientError::State(format!("failed to lock SQLite state store: {error}"))
59        })?;
60        save_state(&mut connection, state)
61    }
62
63    fn append_request_journal(&self, entry: &RequestJournalEntry) -> Result<(), ClientError> {
64        let mut connection = self.connection.lock().map_err(|error| {
65            ClientError::State(format!("failed to lock SQLite state store: {error}"))
66        })?;
67        let transaction = connection.transaction().map_err(sqlite_error)?;
68        transaction
69            .execute(
70                "INSERT INTO request_journal (method, path, status_code, content_type, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5)",
71                params![
72                    &entry.method,
73                    &entry.path,
74                    entry.status_code,
75                    &entry.content_type,
76                    entry.recorded_at.to_rfc3339(),
77                ],
78            )
79            .map_err(sqlite_error)?;
80        transaction
81            .execute(
82                "UPDATE state_meta SET state_version = state_version + 1, updated_at = ?1 WHERE singleton = 1",
83                params![Utc::now().to_rfc3339()],
84            )
85            .map_err(sqlite_error)?;
86        transaction.commit().map_err(sqlite_error)
87    }
88}
89
90fn bootstrap(connection: &Connection, schema_version: u32) -> Result<(), ClientError> {
91    connection
92        .execute_batch(
93            "
94            CREATE TABLE IF NOT EXISTS state_meta (
95              singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
96              schema_version INTEGER NOT NULL,
97              state_version INTEGER NOT NULL,
98              updated_at TEXT NOT NULL
99            );
100
101            CREATE TABLE IF NOT EXISTS request_journal (
102              seq INTEGER PRIMARY KEY AUTOINCREMENT,
103              method TEXT NOT NULL,
104              path TEXT NOT NULL,
105              status_code INTEGER NOT NULL,
106              content_type TEXT,
107              recorded_at TEXT NOT NULL
108            );
109
110            CREATE INDEX IF NOT EXISTS request_journal_recorded_at_idx
111              ON request_journal(recorded_at);
112            ",
113        )
114        .map_err(sqlite_error)?;
115    let exists = connection
116        .query_row("SELECT 1 FROM state_meta WHERE singleton = 1", [], |row| {
117            row.get::<_, i64>(0)
118        })
119        .optional()
120        .map_err(sqlite_error)?
121        .is_some();
122    if !exists {
123        connection
124            .execute(
125                "INSERT INTO state_meta (singleton, schema_version, state_version, updated_at) VALUES (1, ?1, 0, ?2)",
126                params![schema_version, Utc::now().to_rfc3339()],
127            )
128            .map_err(sqlite_error)?;
129    }
130    Ok(())
131}
132
133fn load_state(connection: &Connection) -> Result<PersistedClientState, ClientError> {
134    let (schema_version, state_version) = connection
135        .query_row(
136            "SELECT schema_version, state_version FROM state_meta WHERE singleton = 1",
137            [],
138            |row| Ok((row.get::<_, u32>(0)?, row.get::<_, u64>(1)?)),
139        )
140        .map_err(sqlite_error)?;
141
142    let mut statement = connection
143        .prepare(
144            "SELECT method, path, status_code, content_type, recorded_at FROM request_journal ORDER BY seq ASC",
145        )
146        .map_err(sqlite_error)?;
147    let rows = statement
148        .query_map([], |row| {
149            let recorded_at = row.get::<_, String>(4)?;
150            let recorded_at = DateTime::parse_from_rfc3339(&recorded_at)
151                .map(|value| value.with_timezone(&Utc))
152                .map_err(|error| {
153                    rusqlite::Error::FromSqlConversionFailure(
154                        4,
155                        rusqlite::types::Type::Text,
156                        Box::new(error),
157                    )
158                })?;
159            Ok(RequestJournalEntry {
160                method: row.get(0)?,
161                path: row.get(1)?,
162                status_code: row.get(2)?,
163                content_type: row.get(3)?,
164                recorded_at,
165            })
166        })
167        .map_err(sqlite_error)?;
168    let request_journal = rows.collect::<Result<Vec<_>, _>>().map_err(sqlite_error)?;
169
170    Ok(PersistedClientState {
171        schema_version,
172        state_version,
173        request_journal,
174    })
175}
176
177fn save_state(
178    connection: &mut Connection,
179    state: &PersistedClientState,
180) -> Result<(), ClientError> {
181    let transaction = connection.transaction().map_err(sqlite_error)?;
182    transaction
183        .execute("DELETE FROM request_journal", [])
184        .map_err(sqlite_error)?;
185    transaction
186        .execute(
187            "UPDATE state_meta SET schema_version = ?1, state_version = ?2, updated_at = ?3 WHERE singleton = 1",
188            params![state.schema_version, state.state_version, Utc::now().to_rfc3339()],
189        )
190        .map_err(sqlite_error)?;
191
192    {
193        let mut statement = transaction
194            .prepare(
195                "INSERT INTO request_journal (method, path, status_code, content_type, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5)",
196            )
197            .map_err(sqlite_error)?;
198        for entry in &state.request_journal {
199            statement
200                .execute(params![
201                    &entry.method,
202                    &entry.path,
203                    entry.status_code,
204                    &entry.content_type,
205                    entry.recorded_at.to_rfc3339(),
206                ])
207                .map_err(sqlite_error)?;
208        }
209    }
210
211    transaction.commit().map_err(sqlite_error)
212}
213
214fn sqlite_error(error: rusqlite::Error) -> ClientError {
215    ClientError::State(format!("SQLite state store error: {error}"))
216}
217
218#[cfg(test)]
219mod tests {
220    use std::path::{Path, PathBuf};
221    use std::time::{SystemTime, UNIX_EPOCH};
222
223    use chrono::Utc;
224    use cratestack_client_rust::{ClientStateStore, RequestJournalEntry};
225
226    use super::SqliteStateStore;
227
228    #[test]
229    fn bootstrap_loads_default_state() {
230        let path = project_tmp_path("bootstrap");
231        cleanup(&path);
232
233        let store = SqliteStateStore::open(&path).expect("store should open");
234        let state = store.load().expect("state should load");
235
236        assert_eq!(state.schema_version, 1);
237        assert_eq!(state.state_version, 0);
238        assert!(state.request_journal.is_empty());
239
240        cleanup(&path);
241    }
242
243    #[test]
244    fn append_round_trips_and_increments_state_version() {
245        let path = project_tmp_path("append");
246        cleanup(&path);
247
248        let store = SqliteStateStore::open(&path).expect("store should open");
249        store
250            .append_request_journal(&RequestJournalEntry {
251                method: "POST".to_owned(),
252                path: "/$procs/getFeed".to_owned(),
253                status_code: 200,
254                content_type: Some("application/cbor".to_owned()),
255                recorded_at: Utc::now(),
256            })
257            .expect("journal entry should append");
258
259        let state = store.load().expect("state should load");
260        assert_eq!(state.state_version, 1);
261        assert_eq!(state.request_journal.len(), 1);
262
263        cleanup(&path);
264    }
265
266    fn project_tmp_path(label: &str) -> PathBuf {
267        let suffix = SystemTime::now()
268            .duration_since(UNIX_EPOCH)
269            .expect("time should move forward")
270            .as_nanos();
271        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
272            .join("../../tmp/client-store-sqlite-tests")
273            .join(format!("{label}-{suffix}.sqlite"))
274    }
275
276    fn cleanup(path: &Path) {
277        if path.exists() {
278            std::fs::remove_file(path).expect("tmp file should be removable");
279        }
280    }
281}
cratestack_client_store_sqlite/lib.rs

cratestack_client_store_sqlite/
lib.rs
