Skip to main content

prolly_store_sqlite/
lib.rs

1//! SQLite storage backend implementation
2
3use std::collections::{hash_map::Entry, HashMap};
4use std::path::Path;
5use std::sync::{Mutex, MutexGuard};
6use std::time::Duration;
7
8use rusqlite::{params, Connection, OptionalExtension};
9
10use prolly::{
11    BatchOp, Cid, Error, ManifestStore, ManifestStoreScan, ManifestUpdate, NamedRootManifest,
12    NodeStoreScan, RootCondition, RootManifest, RootWrite, Store, TransactionConflict,
13    TransactionNodeWrite, TransactionUpdate, TransactionalStore,
14};
15
16struct OrderedBatchReadPlan<'a> {
17    unique_keys: Vec<&'a [u8]>,
18    positions: Option<Vec<usize>>,
19}
20
21impl<'a> OrderedBatchReadPlan<'a> {
22    fn new(keys: &[&'a [u8]]) -> Self {
23        let mut unique_indexes = HashMap::with_capacity(keys.len());
24        let mut unique_keys = Vec::with_capacity(keys.len());
25        let mut positions = None;
26        for key in keys {
27            match unique_indexes.entry(*key) {
28                Entry::Occupied(entry) => positions
29                    .get_or_insert_with(|| (0..unique_keys.len()).collect::<Vec<_>>())
30                    .push(*entry.get()),
31                Entry::Vacant(entry) => {
32                    let index = unique_keys.len();
33                    unique_keys.push(*key);
34                    if let Some(positions) = positions.as_mut() {
35                        positions.push(index);
36                    }
37                    entry.insert(index);
38                }
39            }
40        }
41        Self {
42            unique_keys,
43            positions,
44        }
45    }
46
47    fn unique_keys(&self) -> &[&'a [u8]] {
48        &self.unique_keys
49    }
50
51    fn expand_owned<T: Clone>(&self, values: Vec<Option<T>>) -> Vec<Option<T>> {
52        match &self.positions {
53            Some(positions) => positions
54                .iter()
55                .map(|&index| values[index].clone())
56                .collect(),
57            None => values,
58        }
59    }
60}
61
62fn cid_from_store_key(key: &[u8], context: &str) -> Result<Cid, String> {
63    let bytes: [u8; 32] = key.try_into().map_err(|_| {
64        format!(
65            "{context} key has invalid CID length {}, expected 32",
66            key.len()
67        )
68    })?;
69    Ok(Cid(bytes))
70}
71
72fn sort_cids(cids: &mut [Cid]) {
73    cids.sort_by(|left, right| left.as_bytes().cmp(right.as_bytes()));
74}
75
76fn sort_named_root_manifests(roots: &mut [NamedRootManifest]) {
77    roots.sort_by(|left, right| left.name.cmp(&right.name));
78}
79
80const CREATE_TABLE_SQL: &str = "\
81CREATE TABLE IF NOT EXISTS prolly_nodes (
82    cid  BLOB PRIMARY KEY NOT NULL,
83    node BLOB NOT NULL
84) WITHOUT ROWID;";
85
86const CREATE_HINTS_TABLE_SQL: &str = "\
87CREATE TABLE IF NOT EXISTS prolly_hints (
88    namespace BLOB NOT NULL,
89    key       BLOB NOT NULL,
90    value     BLOB NOT NULL,
91    PRIMARY KEY (namespace, key)
92) WITHOUT ROWID;";
93
94const CREATE_ROOTS_TABLE_SQL: &str = "\
95CREATE TABLE IF NOT EXISTS prolly_roots (
96    name     BLOB PRIMARY KEY NOT NULL,
97    manifest BLOB NOT NULL
98) WITHOUT ROWID;";
99
100const SELECT_SQL: &str = "SELECT node FROM prolly_nodes WHERE cid = ?1";
101const SELECT_NODE_CIDS_SQL: &str = "SELECT cid FROM prolly_nodes ORDER BY cid";
102const UPSERT_SQL: &str = "\
103INSERT INTO prolly_nodes (cid, node)
104VALUES (?1, ?2)
105ON CONFLICT(cid) DO UPDATE SET node = excluded.node";
106const DELETE_SQL: &str = "DELETE FROM prolly_nodes WHERE cid = ?1";
107const SELECT_ROOT_SQL: &str = "SELECT manifest FROM prolly_roots WHERE name = ?1";
108const SELECT_ROOTS_SQL: &str = "SELECT name, manifest FROM prolly_roots ORDER BY name";
109const UPSERT_ROOT_SQL: &str = "\
110INSERT INTO prolly_roots (name, manifest)
111VALUES (?1, ?2)
112ON CONFLICT(name) DO UPDATE SET manifest = excluded.manifest";
113const DELETE_ROOT_SQL: &str = "DELETE FROM prolly_roots WHERE name = ?1";
114
115/// Configuration options for [`SqliteStore`].
116#[derive(Debug, Clone)]
117pub struct SqliteStoreConfig {
118    /// Busy timeout in milliseconds for contended SQLite locks.
119    pub busy_timeout_ms: u64,
120    /// Enable WAL journaling for file-backed databases.
121    pub enable_wal: bool,
122    /// Set SQLite synchronous mode to NORMAL when applying default pragmas.
123    pub synchronous_normal: bool,
124}
125
126impl Default for SqliteStoreConfig {
127    fn default() -> Self {
128        Self {
129            busy_timeout_ms: 5_000,
130            enable_wal: true,
131            synchronous_normal: true,
132        }
133    }
134}
135
136/// Error type for SQLite store operations.
137#[derive(Debug)]
138pub struct SqliteStoreError {
139    message: String,
140    source: Option<rusqlite::Error>,
141}
142
143impl SqliteStoreError {
144    /// Create a new error with a message.
145    pub fn new(message: impl Into<String>) -> Self {
146        Self {
147            message: message.into(),
148            source: None,
149        }
150    }
151
152    /// Create a new error from a rusqlite error.
153    pub fn from_sqlite(err: rusqlite::Error, context: impl Into<String>) -> Self {
154        Self {
155            message: format!("{}: {}", context.into(), err),
156            source: Some(err),
157        }
158    }
159}
160
161impl std::fmt::Display for SqliteStoreError {
162    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163        write!(f, "SQLite error: {}", self.message)
164    }
165}
166
167impl std::error::Error for SqliteStoreError {
168    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
169        self.source
170            .as_ref()
171            .map(|e| e as &(dyn std::error::Error + 'static))
172    }
173}
174
175impl From<rusqlite::Error> for SqliteStoreError {
176    fn from(err: rusqlite::Error) -> Self {
177        Self {
178            message: err.to_string(),
179            source: Some(err),
180        }
181    }
182}
183
184/// SQLite-backed storage backend for Prolly Trees.
185///
186/// This store persists content-addressed nodes in a single SQLite table and
187/// supports atomic batch operations through transactions.
188pub struct SqliteStore {
189    conn: Mutex<Connection>,
190}
191
192impl SqliteStore {
193    /// Open or create a SQLite database at the given path with default config.
194    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, SqliteStoreError> {
195        Self::open_with_config(path, SqliteStoreConfig::default())
196    }
197
198    /// Open or create a SQLite database with custom configuration.
199    pub fn open_with_config<P: AsRef<Path>>(
200        path: P,
201        config: SqliteStoreConfig,
202    ) -> Result<Self, SqliteStoreError> {
203        let conn = Connection::open(path.as_ref()).map_err(|e| {
204            SqliteStoreError::from_sqlite(
205                e,
206                format!("Failed to open database at {:?}", path.as_ref()),
207            )
208        })?;
209        Self::from_connection(conn, config)
210    }
211
212    /// Create an in-memory SQLite store.
213    pub fn open_in_memory() -> Result<Self, SqliteStoreError> {
214        let conn = Connection::open_in_memory()
215            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to open in-memory database"))?;
216        Self::from_connection(conn, SqliteStoreConfig::default())
217    }
218
219    fn from_connection(
220        conn: Connection,
221        config: SqliteStoreConfig,
222    ) -> Result<Self, SqliteStoreError> {
223        conn.busy_timeout(Duration::from_millis(config.busy_timeout_ms))
224            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to set busy timeout"))?;
225
226        if config.enable_wal {
227            conn.pragma_update(None, "journal_mode", "WAL")
228                .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to enable WAL mode"))?;
229        }
230        if config.synchronous_normal {
231            conn.pragma_update(None, "synchronous", "NORMAL")
232                .map_err(|e| {
233                    SqliteStoreError::from_sqlite(e, "Failed to set synchronous=NORMAL")
234                })?;
235        }
236        conn.pragma_update(None, "temp_store", "MEMORY")
237            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to set temp_store=MEMORY"))?;
238        conn.execute_batch(CREATE_TABLE_SQL)
239            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize schema"))?;
240        conn.execute_batch(CREATE_HINTS_TABLE_SQL)
241            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize hint schema"))?;
242        conn.execute_batch(CREATE_ROOTS_TABLE_SQL)
243            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to initialize root schema"))?;
244
245        Ok(Self {
246            conn: Mutex::new(conn),
247        })
248    }
249
250    fn connection(&self) -> Result<MutexGuard<'_, Connection>, SqliteStoreError> {
251        self.conn
252            .lock()
253            .map_err(|e| SqliteStoreError::new(format!("lock poisoned: {}", e)))
254    }
255}
256
257impl Store for SqliteStore {
258    type Error = SqliteStoreError;
259
260    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
261        let conn = self.connection()?;
262        conn.query_row(SELECT_SQL, params![key], |row| row.get(0))
263            .optional()
264            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read key"))
265    }
266
267    fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Self::Error> {
268        let conn = self.connection()?;
269        conn.execute(UPSERT_SQL, params![key, value])
270            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write key"))?;
271        Ok(())
272    }
273
274    fn delete(&self, key: &[u8]) -> Result<(), Self::Error> {
275        let conn = self.connection()?;
276        conn.execute(DELETE_SQL, params![key])
277            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to delete key"))?;
278        Ok(())
279    }
280
281    fn batch(&self, ops: &[BatchOp]) -> Result<(), Self::Error> {
282        let mut conn = self.connection()?;
283        let tx = conn
284            .transaction()
285            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
286
287        {
288            let mut upsert = tx
289                .prepare_cached(UPSERT_SQL)
290                .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch write"))?;
291            let mut delete = tx
292                .prepare_cached(DELETE_SQL)
293                .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch delete"))?;
294
295            for op in ops {
296                match op {
297                    BatchOp::Upsert { key, value } => {
298                        upsert.execute(params![key, value]).map_err(|e| {
299                            SqliteStoreError::from_sqlite(e, "Failed to write key in batch")
300                        })?;
301                    }
302                    BatchOp::Delete { key } => {
303                        delete.execute(params![key]).map_err(|e| {
304                            SqliteStoreError::from_sqlite(e, "Failed to delete key in batch")
305                        })?;
306                    }
307                }
308            }
309        }
310
311        tx.commit()
312            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
313    }
314
315    fn batch_get(&self, keys: &[&[u8]]) -> Result<HashMap<Vec<u8>, Vec<u8>>, Self::Error> {
316        let conn = self.connection()?;
317        let mut stmt = conn
318            .prepare_cached(SELECT_SQL)
319            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare batch read"))?;
320        let plan = OrderedBatchReadPlan::new(keys);
321        let mut results = HashMap::with_capacity(plan.unique_keys().len());
322
323        for key in plan.unique_keys() {
324            if let Some(value) = stmt
325                .query_row(params![key], |row| row.get(0))
326                .optional()
327                .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read key in batch"))?
328            {
329                results.insert(key.to_vec(), value);
330            }
331        }
332
333        Ok(results)
334    }
335
336    fn batch_get_ordered(&self, keys: &[&[u8]]) -> Result<Vec<Option<Vec<u8>>>, Self::Error> {
337        let conn = self.connection()?;
338        let mut stmt = conn.prepare_cached(SELECT_SQL).map_err(|e| {
339            SqliteStoreError::from_sqlite(e, "Failed to prepare ordered batch read")
340        })?;
341        let plan = OrderedBatchReadPlan::new(keys);
342        let mut unique_values = Vec::with_capacity(plan.unique_keys().len());
343
344        for key in plan.unique_keys() {
345            let value = stmt
346                .query_row(params![key], |row| row.get(0))
347                .optional()
348                .map_err(|e| {
349                    SqliteStoreError::from_sqlite(e, "Failed to read key in ordered batch")
350                })?;
351            unique_values.push(value);
352        }
353
354        Ok(plan.expand_owned(unique_values))
355    }
356
357    fn batch_get_ordered_unique(
358        &self,
359        keys: &[&[u8]],
360    ) -> Result<Vec<Option<Vec<u8>>>, Self::Error> {
361        let conn = self.connection()?;
362        let mut stmt = conn.prepare_cached(SELECT_SQL).map_err(|e| {
363            SqliteStoreError::from_sqlite(e, "Failed to prepare unique ordered batch read")
364        })?;
365        let mut values = Vec::with_capacity(keys.len());
366
367        for key in keys {
368            let value = stmt
369                .query_row(params![key], |row| row.get(0))
370                .optional()
371                .map_err(|e| {
372                    SqliteStoreError::from_sqlite(e, "Failed to read key in unique ordered batch")
373                })?;
374            values.push(value);
375        }
376
377        Ok(values)
378    }
379
380    fn batch_put(&self, entries: &[(&[u8], &[u8])]) -> Result<(), Self::Error> {
381        let mut conn = self.connection()?;
382        let tx = conn
383            .transaction()
384            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
385
386        {
387            let mut stmt = tx.prepare_cached(UPSERT_SQL).map_err(|e| {
388                SqliteStoreError::from_sqlite(e, "Failed to prepare batch_put write")
389            })?;
390            for (key, value) in entries {
391                stmt.execute(params![key, value]).map_err(|e| {
392                    SqliteStoreError::from_sqlite(e, "Failed to write key in batch_put")
393                })?;
394            }
395        }
396
397        tx.commit()
398            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
399    }
400
401    fn supports_hints(&self) -> bool {
402        true
403    }
404
405    fn get_hint(&self, namespace: &[u8], key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
406        let conn = self.connection()?;
407        conn.query_row(
408            "SELECT value FROM prolly_hints WHERE namespace = ?1 AND key = ?2",
409            params![namespace, key],
410            |row| row.get(0),
411        )
412        .optional()
413        .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read hint"))
414    }
415
416    fn put_hint(&self, namespace: &[u8], key: &[u8], value: &[u8]) -> Result<(), Self::Error> {
417        let conn = self.connection()?;
418        conn.execute(
419            "\
420            INSERT INTO prolly_hints (namespace, key, value) \
421            VALUES (?1, ?2, ?3) \
422            ON CONFLICT(namespace, key) DO UPDATE SET value = excluded.value",
423            params![namespace, key, value],
424        )
425        .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write hint"))?;
426        Ok(())
427    }
428
429    fn batch_put_with_hint(
430        &self,
431        entries: &[(&[u8], &[u8])],
432        namespace: &[u8],
433        key: &[u8],
434        value: &[u8],
435    ) -> Result<(), Self::Error> {
436        let mut conn = self.connection()?;
437        let tx = conn
438            .transaction()
439            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start transaction"))?;
440
441        {
442            let mut upsert_node = tx.prepare_cached(UPSERT_SQL).map_err(|e| {
443                SqliteStoreError::from_sqlite(e, "Failed to prepare batch_put write")
444            })?;
445            for (key, value) in entries {
446                upsert_node.execute(params![key, value]).map_err(|e| {
447                    SqliteStoreError::from_sqlite(e, "Failed to write key in batch_put")
448                })?;
449            }
450        }
451
452        tx.execute(
453            "\
454            INSERT INTO prolly_hints (namespace, key, value) \
455            VALUES (?1, ?2, ?3) \
456            ON CONFLICT(namespace, key) DO UPDATE SET value = excluded.value",
457            params![namespace, key, value],
458        )
459        .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write hint in batch_put"))?;
460
461        tx.commit()
462            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit transaction"))
463    }
464}
465
466impl NodeStoreScan for SqliteStore {
467    type Error = SqliteStoreError;
468
469    fn list_node_cids(&self) -> Result<Vec<Cid>, Self::Error> {
470        let conn = self.connection()?;
471        let mut stmt = conn
472            .prepare_cached(SELECT_NODE_CIDS_SQL)
473            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to prepare node CID listing"))?;
474        let rows = stmt
475            .query_map([], |row| row.get::<_, Vec<u8>>(0))
476            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to list node CIDs"))?;
477
478        let mut cids = Vec::new();
479        for row in rows {
480            let key = row
481                .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read listed node CID"))?;
482            cids.push(cid_from_store_key(&key, "SQLite node").map_err(SqliteStoreError::new)?);
483        }
484        sort_cids(&mut cids);
485        Ok(cids)
486    }
487}
488
489impl ManifestStore for SqliteStore {
490    type Error = SqliteStoreError;
491
492    fn get_root(&self, name: &[u8]) -> Result<Option<RootManifest>, Self::Error> {
493        let conn = self.connection()?;
494        let bytes = conn
495            .query_row(SELECT_ROOT_SQL, params![name], |row| row.get(0))
496            .optional()
497            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read root manifest"))?;
498        decode_root_manifest(bytes)
499    }
500
501    fn put_root(&self, name: &[u8], manifest: &RootManifest) -> Result<(), Self::Error> {
502        let conn = self.connection()?;
503        let bytes = encode_root_manifest(manifest)?;
504        conn.execute(UPSERT_ROOT_SQL, params![name, bytes])
505            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to write root manifest"))?;
506        Ok(())
507    }
508
509    fn delete_root(&self, name: &[u8]) -> Result<(), Self::Error> {
510        let conn = self.connection()?;
511        conn.execute(DELETE_ROOT_SQL, params![name])
512            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to delete root manifest"))?;
513        Ok(())
514    }
515
516    fn compare_and_swap_root(
517        &self,
518        name: &[u8],
519        expected: Option<&RootManifest>,
520        new: Option<&RootManifest>,
521    ) -> Result<ManifestUpdate, Self::Error> {
522        let expected_bytes = expected.map(encode_root_manifest).transpose()?;
523        let new_bytes = new.map(encode_root_manifest).transpose()?;
524
525        let mut conn = self.connection()?;
526        let tx = conn
527            .transaction()
528            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to start root transaction"))?;
529
530        let current_bytes = tx
531            .query_row(SELECT_ROOT_SQL, params![name], |row| row.get(0))
532            .optional()
533            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to read root manifest"))?;
534
535        if current_bytes.as_deref() != expected_bytes.as_deref() {
536            return Ok(ManifestUpdate::Conflict {
537                current: decode_root_manifest(current_bytes)?,
538            });
539        }
540
541        match new_bytes {
542            Some(bytes) => {
543                tx.execute(UPSERT_ROOT_SQL, params![name, bytes])
544                    .map_err(|e| {
545                        SqliteStoreError::from_sqlite(e, "Failed to write root manifest")
546                    })?;
547            }
548            None => {
549                tx.execute(DELETE_ROOT_SQL, params![name]).map_err(|e| {
550                    SqliteStoreError::from_sqlite(e, "Failed to delete root manifest")
551                })?;
552            }
553        }
554
555        tx.commit()
556            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to commit root transaction"))?;
557        Ok(ManifestUpdate::Applied)
558    }
559}
560
561impl ManifestStoreScan for SqliteStore {
562    fn list_roots(&self) -> Result<Vec<NamedRootManifest>, Self::Error> {
563        let conn = self.connection()?;
564        let mut stmt = conn.prepare_cached(SELECT_ROOTS_SQL).map_err(|e| {
565            SqliteStoreError::from_sqlite(e, "Failed to prepare root manifest listing")
566        })?;
567        let rows = stmt
568            .query_map([], |row| {
569                Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, Vec<u8>>(1)?))
570            })
571            .map_err(|e| SqliteStoreError::from_sqlite(e, "Failed to list root manifests"))?;
572
573        let mut roots = Vec::new();
574        for row in rows {
575            let (name, bytes) = row.map_err(|e| {
576                SqliteStoreError::from_sqlite(e, "Failed to read listed root manifest")
577            })?;
578            let manifest = RootManifest::from_bytes(&bytes)
579                .map_err(|err| SqliteStoreError::new(err.to_string()))?;
580            roots.push(NamedRootManifest::new(name, manifest));
581        }
582        sort_named_root_manifests(&mut roots);
583        Ok(roots)
584    }
585}
586
587impl TransactionalStore for SqliteStore {
588    fn supports_transactions(&self) -> bool {
589        true
590    }
591
592    fn commit_transaction(
593        &self,
594        node_writes: &[TransactionNodeWrite],
595        root_conditions: &[RootCondition],
596        root_writes: &[RootWrite],
597    ) -> Result<TransactionUpdate, Error> {
598        let mut conn = self
599            .connection()
600            .map_err(|err| Error::Store(Box::new(err)))?;
601        let tx = conn.transaction().map_err(|err| {
602            Error::Store(Box::new(SqliteStoreError::from_sqlite(
603                err,
604                "Failed to start transaction commit",
605            )))
606        })?;
607
608        for condition in root_conditions {
609            let current_bytes = tx
610                .query_row(SELECT_ROOT_SQL, params![condition.name], |row| row.get(0))
611                .optional()
612                .map_err(|err| {
613                    Error::Store(Box::new(SqliteStoreError::from_sqlite(
614                        err,
615                        "Failed to read root manifest during transaction commit",
616                    )))
617                })?;
618            let current =
619                decode_root_manifest(current_bytes).map_err(|err| Error::Store(Box::new(err)))?;
620            if current != condition.expected {
621                return Ok(TransactionUpdate::Conflict(TransactionConflict::new(
622                    condition.name.clone(),
623                    condition.expected.clone(),
624                    current,
625                )));
626            }
627        }
628
629        {
630            let mut upsert_node = tx.prepare_cached(UPSERT_SQL).map_err(|err| {
631                Error::Store(Box::new(SqliteStoreError::from_sqlite(
632                    err,
633                    "Failed to prepare transaction node write",
634                )))
635            })?;
636            let mut delete_node = tx.prepare_cached(DELETE_SQL).map_err(|err| {
637                Error::Store(Box::new(SqliteStoreError::from_sqlite(
638                    err,
639                    "Failed to prepare transaction node delete",
640                )))
641            })?;
642            for write in node_writes {
643                match write {
644                    TransactionNodeWrite::Upsert { key, value } => {
645                        upsert_node.execute(params![key, value]).map_err(|err| {
646                            Error::Store(Box::new(SqliteStoreError::from_sqlite(
647                                err,
648                                "Failed to write node during transaction commit",
649                            )))
650                        })?;
651                    }
652                    TransactionNodeWrite::Delete { key } => {
653                        delete_node.execute(params![key]).map_err(|err| {
654                            Error::Store(Box::new(SqliteStoreError::from_sqlite(
655                                err,
656                                "Failed to delete node during transaction commit",
657                            )))
658                        })?;
659                    }
660                }
661            }
662        }
663
664        {
665            let mut upsert_root = tx.prepare_cached(UPSERT_ROOT_SQL).map_err(|err| {
666                Error::Store(Box::new(SqliteStoreError::from_sqlite(
667                    err,
668                    "Failed to prepare transaction root write",
669                )))
670            })?;
671            let mut delete_root = tx.prepare_cached(DELETE_ROOT_SQL).map_err(|err| {
672                Error::Store(Box::new(SqliteStoreError::from_sqlite(
673                    err,
674                    "Failed to prepare transaction root delete",
675                )))
676            })?;
677            for write in root_writes {
678                match write {
679                    RootWrite::Put { name, manifest } => {
680                        let bytes = encode_root_manifest(manifest)
681                            .map_err(|err| Error::Store(Box::new(err)))?;
682                        upsert_root.execute(params![name, bytes]).map_err(|err| {
683                            Error::Store(Box::new(SqliteStoreError::from_sqlite(
684                                err,
685                                "Failed to write root during transaction commit",
686                            )))
687                        })?;
688                    }
689                    RootWrite::Delete { name } => {
690                        delete_root.execute(params![name]).map_err(|err| {
691                            Error::Store(Box::new(SqliteStoreError::from_sqlite(
692                                err,
693                                "Failed to delete root during transaction commit",
694                            )))
695                        })?;
696                    }
697                }
698            }
699        }
700
701        tx.commit().map_err(|err| {
702            Error::Store(Box::new(SqliteStoreError::from_sqlite(
703                err,
704                "Failed to commit transaction",
705            )))
706        })?;
707        Ok(TransactionUpdate::Applied {
708            nodes_written: node_writes.len(),
709            roots_written: root_writes.len(),
710        })
711    }
712}
713
714fn encode_root_manifest(manifest: &RootManifest) -> Result<Vec<u8>, SqliteStoreError> {
715    manifest
716        .to_bytes()
717        .map_err(|e| SqliteStoreError::new(format!("failed to encode root manifest: {e}")))
718}
719
720fn decode_root_manifest(bytes: Option<Vec<u8>>) -> Result<Option<RootManifest>, SqliteStoreError> {
721    bytes
722        .as_deref()
723        .map(RootManifest::from_bytes)
724        .transpose()
725        .map_err(|e| SqliteStoreError::new(format!("failed to decode root manifest: {e}")))
726}
727
728#[cfg(test)]
729mod tests {
730    use super::*;
731
732    #[test]
733    fn sqlite_store_put_get_delete() {
734        let store = SqliteStore::open_in_memory().unwrap();
735
736        store.put(b"key", b"value").unwrap();
737        assert_eq!(store.get(b"key").unwrap(), Some(b"value".to_vec()));
738
739        store.delete(b"key").unwrap();
740        assert_eq!(store.get(b"key").unwrap(), None);
741    }
742
743    #[test]
744    fn sqlite_store_batch_is_order_preserving_for_reads() {
745        let store = SqliteStore::open_in_memory().unwrap();
746        let ops = vec![
747            BatchOp::Upsert {
748                key: b"a",
749                value: b"1",
750            },
751            BatchOp::Upsert {
752                key: b"b",
753                value: b"2",
754            },
755            BatchOp::Upsert {
756                key: b"c",
757                value: b"3",
758            },
759        ];
760
761        store.batch(&ops).unwrap();
762
763        let keys: Vec<&[u8]> = vec![b"c", b"missing", b"a", b"c", b"missing", b"b"];
764        assert_eq!(
765            store.batch_get_ordered(&keys).unwrap(),
766            vec![
767                Some(b"3".to_vec()),
768                None,
769                Some(b"1".to_vec()),
770                Some(b"3".to_vec()),
771                None,
772                Some(b"2".to_vec())
773            ]
774        );
775    }
776
777    #[test]
778    fn sqlite_store_batch_put_updates_existing_keys() {
779        let store = SqliteStore::open_in_memory().unwrap();
780
781        store.put(b"a", b"old").unwrap();
782        store
783            .batch_put(&[(b"a".as_slice(), b"new".as_slice()), (b"b", b"2")])
784            .unwrap();
785
786        assert_eq!(store.get(b"a").unwrap(), Some(b"new".to_vec()));
787        assert_eq!(store.get(b"b").unwrap(), Some(b"2".to_vec()));
788    }
789
790    #[test]
791    fn sqlite_store_persists_hints_separately_from_nodes() {
792        let store = SqliteStore::open_in_memory().unwrap();
793
794        store.put_hint(b"rightmost", b"root", b"hint-v1").unwrap();
795        assert_eq!(
796            store.get_hint(b"rightmost", b"root").unwrap(),
797            Some(b"hint-v1".to_vec())
798        );
799        assert_eq!(store.get_hint(b"rightmost", b"missing").unwrap(), None);
800        assert_eq!(store.get(b"root").unwrap(), None);
801
802        store.put_hint(b"rightmost", b"root", b"hint-v2").unwrap();
803        assert_eq!(
804            store.get_hint(b"rightmost", b"root").unwrap(),
805            Some(b"hint-v2".to_vec())
806        );
807    }
808}