kcode_k1_persons_store/
lib.rs

1use rusqlite::ffi::ErrorCode::{
2    ConstraintViolation, DatabaseCorrupt, NotADatabase, SchemaChanged, TypeMismatch, Unknown,
3};
4use rusqlite::{Connection, Error, TransactionBehavior, params};
5use std::{
6    collections::HashMap,
7    fs,
8    path::{Path, PathBuf},
9    time::Duration,
10};
11
12pub use kcode_k1_person_types::PersonId;
13use kcode_k1_txn_ordering::K1TxnOrdering;
14pub use kcode_k1_txn_ordering::TxId;
15
16const APPLICATION_ID: i64 = 0x4b31_5050;
17const SCHEMA_VERSION: i64 = 1;
18const META_SCHEMA: &str = "CREATE TABLE meta(singleton INTEGER PRIMARY KEY CHECK(singleton = 1), checkpoint BLOB CHECK(checkpoint IS NULL OR length(checkpoint) = 12)) STRICT";
19const PERSONS_SCHEMA: &str = "CREATE TABLE persons(id BLOB PRIMARY KEY CHECK(length(id) = 12), root BLOB NOT NULL REFERENCES persons(id), name TEXT, CHECK((id = root AND name IS NOT NULL) OR (id <> root AND name IS NULL))) STRICT, WITHOUT ROWID";
20const ROOT_INDEX_SCHEMA: &str = "CREATE INDEX persons_root ON persons(root)";
21
22#[derive(Clone, Debug, Eq, PartialEq)]
23pub struct StoredPerson {
24    id: PersonId,
25    root: PersonId,
26    name: Option<String>,
27}
28
29impl StoredPerson {
30    pub fn id(&self) -> PersonId {
31        self.id
32    }
33
34    pub fn root(&self) -> PersonId {
35        self.root
36    }
37
38    pub fn name(&self) -> Option<&str> {
39        self.name.as_deref()
40    }
41}
42
43#[derive(Clone, Debug, Eq, PartialEq)]
44pub struct StoredSnapshot {
45    checkpoint: Option<TxId>,
46    persons: Vec<StoredPerson>,
47}
48
49impl StoredSnapshot {
50    pub fn checkpoint(&self) -> Option<TxId> {
51        self.checkpoint
52    }
53
54    pub fn persons(&self) -> &[StoredPerson] {
55        &self.persons
56    }
57}
58
59#[derive(Clone, Debug, Eq, PartialEq)]
60pub struct StoreChange(Change);
61
62#[derive(Clone, Debug, Eq, PartialEq)]
63enum Change {
64    Create(PersonId, String),
65    Update(PersonId, String),
66    Resolve(PersonId, PersonId, usize),
67}
68
69impl StoreChange {
70    pub fn create(id: PersonId, name: String) -> Self {
71        Self(Change::Create(id, name))
72    }
73
74    pub fn update(id: PersonId, name: String) -> Self {
75        Self(Change::Update(id, name))
76    }
77
78    pub fn resolve(from: PersonId, to: PersonId, expected_class_size: usize) -> Self {
79        Self(Change::Resolve(from, to, expected_class_size))
80    }
81}
82
83pub struct Store {
84    connection: Connection,
85}
86
87impl Store {
88    pub fn open(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, StoredSnapshot), String> {
89        fs::create_dir_all(root).map_err(|error| format!("create store root: {error}"))?;
90        let path = root.join("persons.sqlite3");
91        let exists = match fs::symlink_metadata(&path) {
92            Ok(metadata) if metadata.file_type().is_file() => true,
93            Ok(_) => return Err("persons.sqlite3 is a symlink or non-file".to_owned()),
94            Err(error) if error.kind() == std::io::ErrorKind::NotFound => false,
95            Err(error) => return Err(format!("inspect persons.sqlite3: {error}")),
96        };
97        if !exists {
98            return empty_store(create_database(&path)?);
99        }
100        let connection =
101            Connection::open(&path).map_err(|error| format!("open database: {error}"))?;
102        if let Err(error) = configure(&connection) {
103            let failure = load_error("configure database", error);
104            drop(connection);
105            return match failure {
106                LoadFailure::Recoverable => replace_database(&path),
107                LoadFailure::Fatal(message) => Err(message),
108            };
109        }
110        match load_database(&connection, ordering) {
111            Ok(snapshot) => Ok((Self { connection }, snapshot)),
112            Err(LoadFailure::Fatal(message)) => Err(message),
113            Err(LoadFailure::Recoverable) => {
114                drop(connection);
115                replace_database(&path)
116            }
117        }
118    }
119
120    pub fn commit(&mut self, callback: TxId, change: Option<&StoreChange>) -> Result<(), String> {
121        let transaction = self
122            .connection
123            .transaction_with_behavior(TransactionBehavior::Immediate)
124            .map_err(|error| format!("begin commit: {error}"))?;
125        if let Some(StoreChange(change)) = change {
126            apply_change(&transaction, change)?;
127        }
128        let changed = transaction
129            .execute(
130                "UPDATE meta SET checkpoint = ?1 WHERE singleton = 1",
131                params![&callback.as_bytes()[..]],
132            )
133            .map_err(|error| format!("write checkpoint: {error}"))?;
134        if changed != 1 {
135            return Err(format!("checkpoint update affected {changed} rows"));
136        }
137        transaction
138            .commit()
139            .map_err(|error| format!("commit database: {error}"))
140    }
141
142    pub fn clear(&mut self) -> Result<(), String> {
143        let transaction = self
144            .connection
145            .transaction_with_behavior(TransactionBehavior::Immediate)
146            .map_err(|error| format!("begin clear: {error}"))?;
147        transaction
148            .execute("DELETE FROM persons", [])
149            .map_err(|error| format!("delete persons: {error}"))?;
150        let changed = transaction
151            .execute("UPDATE meta SET checkpoint = NULL WHERE singleton = 1", [])
152            .map_err(|error| format!("clear checkpoint: {error}"))?;
153        if changed != 1 {
154            return Err(format!("checkpoint clear affected {changed} rows"));
155        }
156        transaction
157            .commit()
158            .map_err(|error| format!("commit clear: {error}"))
159    }
160}
161
162fn apply_change(transaction: &rusqlite::Transaction<'_>, change: &Change) -> Result<(), String> {
163    let changed = match change {
164        Change::Create(id, name) => {
165            let id = id.as_tx_id().into_bytes();
166            transaction.execute(
167                "INSERT INTO persons(id, root, name) VALUES(?1, ?1, ?2)",
168                params![&id[..], name],
169            )
170        }
171        Change::Update(id, name) => {
172            let id = id.as_tx_id().into_bytes();
173            transaction.execute(
174                "UPDATE persons SET name = ?1 WHERE id = ?2 AND root = id AND name IS NOT NULL",
175                params![name, &id[..]],
176            )
177        }
178        Change::Resolve(from, to, expected) => {
179            let from = from.as_tx_id().into_bytes();
180            let to = to.as_tx_id().into_bytes();
181            let changed = transaction
182                .execute(
183                    "UPDATE persons SET root = ?1, name = CASE WHEN id = ?2 THEN NULL ELSE name END WHERE root = ?2",
184                    params![&to[..], &from[..]],
185                )
186                .map_err(|error| format!("apply resolve: {error}"))?;
187            if changed == 0 || changed != *expected {
188                return Err(format!(
189                    "resolve affected {changed} rows, expected {expected}"
190                ));
191            }
192            return Ok(());
193        }
194    }
195    .map_err(|error| format!("apply store change: {error}"))?;
196    if changed != 1 {
197        return Err(format!("store change affected {changed} rows"));
198    }
199    Ok(())
200}
201
202fn configure(connection: &Connection) -> rusqlite::Result<()> {
203    connection.busy_timeout(Duration::from_secs(5))?;
204    connection.pragma_update(None, "journal_mode", "WAL")?;
205    connection.pragma_update(None, "synchronous", "FULL")?;
206    connection.pragma_update(None, "foreign_keys", "ON")
207}
208
209fn create_database(path: &Path) -> Result<Connection, String> {
210    let connection = Connection::open(path).map_err(|error| format!("create database: {error}"))?;
211    configure(&connection).map_err(|error| format!("configure new database: {error}"))?;
212    let setup = format!(
213        "BEGIN IMMEDIATE;PRAGMA application_id={APPLICATION_ID};PRAGMA user_version={SCHEMA_VERSION};{META_SCHEMA};{PERSONS_SCHEMA};{ROOT_INDEX_SCHEMA};INSERT INTO meta(singleton, checkpoint) VALUES(1, NULL);COMMIT;"
214    );
215    connection
216        .execute_batch(&setup)
217        .map_err(|error| format!("create schema: {error}"))?;
218    Ok(connection)
219}
220
221fn replace_database(path: &Path) -> Result<(Store, StoredSnapshot), String> {
222    for candidate in [
223        sidecar(path, "-wal"),
224        sidecar(path, "-shm"),
225        path.to_path_buf(),
226    ] {
227        match fs::remove_file(&candidate) {
228            Ok(()) => {}
229            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
230            Err(error) => return Err(format!("remove {}: {error}", candidate.display())),
231        }
232    }
233    empty_store(create_database(path)?)
234}
235
236fn sidecar(path: &Path, suffix: &str) -> PathBuf {
237    let mut value = path.as_os_str().to_os_string();
238    value.push(suffix);
239    PathBuf::from(value)
240}
241
242fn empty_store(connection: Connection) -> Result<(Store, StoredSnapshot), String> {
243    Ok((
244        Store { connection },
245        StoredSnapshot {
246            checkpoint: None,
247            persons: Vec::new(),
248        },
249    ))
250}
251
252enum LoadFailure {
253    Recoverable,
254    Fatal(String),
255}
256
257fn load_error(context: &str, error: Error) -> LoadFailure {
258    let recoverable = match &error {
259        Error::SqliteFailure(failure, _) => matches!(
260            failure.code,
261            DatabaseCorrupt
262                | NotADatabase
263                | SchemaChanged
264                | ConstraintViolation
265                | TypeMismatch
266                | Unknown
267        ),
268        _ => true,
269    };
270    if recoverable {
271        LoadFailure::Recoverable
272    } else {
273        LoadFailure::Fatal(format!("{context}: {error}"))
274    }
275}
276
277fn load_database(
278    connection: &Connection,
279    ordering: &K1TxnOrdering,
280) -> Result<StoredSnapshot, LoadFailure> {
281    let application_id: i64 = connection
282        .query_row("PRAGMA application_id", [], |row| row.get(0))
283        .map_err(|error| load_error("read application id", error))?;
284    let version: i64 = connection
285        .query_row("PRAGMA user_version", [], |row| row.get(0))
286        .map_err(|error| load_error("read schema version", error))?;
287    let integrity: String = connection
288        .query_row("PRAGMA integrity_check", [], |row| row.get(0))
289        .map_err(|error| load_error("check database integrity", error))?;
290    if application_id != APPLICATION_ID || version != SCHEMA_VERSION || integrity != "ok" {
291        return Err(LoadFailure::Recoverable);
292    }
293    let schema = schema_rows(connection).map_err(|error| load_error("read schema", error))?;
294    let expected = vec![
295        (
296            "index".to_owned(),
297            "persons_root".to_owned(),
298            "persons".to_owned(),
299            Some(ROOT_INDEX_SCHEMA.to_owned()),
300        ),
301        (
302            "table".to_owned(),
303            "meta".to_owned(),
304            "meta".to_owned(),
305            Some(META_SCHEMA.to_owned()),
306        ),
307        (
308            "table".to_owned(),
309            "persons".to_owned(),
310            "persons".to_owned(),
311            Some(PERSONS_SCHEMA.to_owned()),
312        ),
313    ];
314    if schema != expected || foreign_key_error(connection)? {
315        return Err(LoadFailure::Recoverable);
316    }
317    let meta = meta_rows(connection).map_err(|error| load_error("read metadata", error))?;
318    if meta.len() != 1 || meta[0].0 != 1 {
319        return Err(LoadFailure::Recoverable);
320    }
321    let checkpoint = match &meta[0].1 {
322        Some(bytes) => Some(TxId::from_bytes(fixed_bytes(bytes)?)),
323        None => None,
324    };
325    if checkpoint.is_some_and(|id| !ordering.contains(id)) {
326        return Err(LoadFailure::Recoverable);
327    }
328    let rows = person_rows(connection).map_err(|error| load_error("read persons", error))?;
329    if !rows.is_empty() && checkpoint.is_none() {
330        return Err(LoadFailure::Recoverable);
331    }
332    let mut persons = Vec::with_capacity(rows.len());
333    for (id, root, name) in rows {
334        let id = PersonId::from_tx_id(TxId::from_bytes(fixed_bytes(&id)?));
335        let root = PersonId::from_tx_id(TxId::from_bytes(fixed_bytes(&root)?));
336        if !ordering.contains(id.as_tx_id())
337            || (id == root && name.is_none())
338            || (id != root && name.is_some())
339        {
340            return Err(LoadFailure::Recoverable);
341        }
342        persons.push(StoredPerson { id, root, name });
343    }
344    let by_id: HashMap<PersonId, &StoredPerson> =
345        persons.iter().map(|person| (person.id, person)).collect();
346    for person in &persons {
347        let Some(root) = by_id.get(&person.root) else {
348            return Err(LoadFailure::Recoverable);
349        };
350        if root.id != root.root || root.name.is_none() {
351            return Err(LoadFailure::Recoverable);
352        }
353    }
354    Ok(StoredSnapshot {
355        checkpoint,
356        persons,
357    })
358}
359
360fn fixed_bytes(bytes: &[u8]) -> Result<[u8; 12], LoadFailure> {
361    bytes.try_into().map_err(|_| LoadFailure::Recoverable)
362}
363
364type SchemaRow = (String, String, String, Option<String>);
365type PersonRow = (Vec<u8>, Vec<u8>, Option<String>);
366
367fn schema_rows(connection: &Connection) -> rusqlite::Result<Vec<SchemaRow>> {
368    let mut statement = connection.prepare(
369        "SELECT type, name, tbl_name, sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%' ORDER BY type, name",
370    )?;
371    statement
372        .query_map([], |row| {
373            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
374        })?
375        .collect()
376}
377
378fn meta_rows(connection: &Connection) -> rusqlite::Result<Vec<(i64, Option<Vec<u8>>)>> {
379    let mut statement = connection.prepare("SELECT singleton, checkpoint FROM meta")?;
380    statement
381        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?
382        .collect()
383}
384
385fn person_rows(connection: &Connection) -> rusqlite::Result<Vec<PersonRow>> {
386    let mut statement = connection.prepare("SELECT id, root, name FROM persons ORDER BY id")?;
387    statement
388        .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))?
389        .collect()
390}
391
392fn foreign_key_error(connection: &Connection) -> Result<bool, LoadFailure> {
393    let mut statement = connection
394        .prepare("PRAGMA foreign_key_check")
395        .map_err(|error| load_error("prepare foreign key check", error))?;
396    let mut rows = statement
397        .query([])
398        .map_err(|error| load_error("run foreign key check", error))?;
399    rows.next()
400        .map(|row| row.is_some())
401        .map_err(|error| load_error("read foreign key check", error))
402}
kcode_k1_persons_store/lib.rs

kcode_k1_persons_store/
lib.rs
