sim_lib_doc_store/
store.rs

1//! Provider-neutral document projection storage.
2
3use crate::codec::{CodecError, decode_doc, decode_edit, encode_doc, encode_edit};
4use sim_kernel::{Datum, NumberLiteral, Symbol};
5use sim_lib_doc_core::{Doc, DocId, Edit};
6use sim_platform_sqlite::{PreopenedStores, SqliteDriver};
7use sim_relation_core::{
8    BaseDomain, BindingName, Cell, ColumnName, DomainCatalog, FieldName, FieldType, IndexName,
9    ProviderName, RevisionName, Row, RowType, SchemaName, SourceName, StorageRepr, TableName,
10};
11use sim_relation_migrate::AdoptionManifest;
12use sim_relation_plan::{
13    AdmissionLimits, ConflictAction, ConflictTarget, FieldRef, Mutation, NamedScalar,
14    OrderDirection, OrderKey, Rel, Scalar, ScalarOp, admit_mutation, admit_query,
15};
16use sim_relation_schema::{
17    AcceptAllValues, ColumnBuilder, Constraint, ForeignKey, Index, PhysicalColumn, PhysicalIndex,
18    PhysicalSchema, PhysicalTable, PrimaryKey, Schema, SchemaBuilder, TableBuilder,
19};
20use sim_relation_site::{Bindings, Driver, Limits, Session, SiteError, StorageAccess, VecRowSink};
21use std::{cell::RefCell, fmt, path::Path};
22
23const EMPTY_STORE: &[u8] = include_bytes!("../fixtures/empty-doc-store-v1.sqlite");
24
25/// Stable document-store failure categories.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub enum StoreError {
28    /// Invalid caller or schema value.
29    Invalid(String),
30    /// Persisted codec failure.
31    Codec(String),
32    /// Relational provider failure.
33    Storage(SiteError),
34    /// Host materialization failure.
35    Host(String),
36}
37impl fmt::Display for StoreError {
38    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
39        write!(f, "{self:?}")
40    }
41}
42impl std::error::Error for StoreError {}
43impl From<SiteError> for StoreError {
44    fn from(v: SiteError) -> Self {
45        Self::Storage(v)
46    }
47}
48impl From<CodecError> for StoreError {
49    fn from(v: CodecError) -> Self {
50        Self::Codec(v.to_string())
51    }
52}
53/// Document-store result.
54pub type StoreResult<T> = Result<T, StoreError>;
55
56/// A document store with one private provider-neutral session.
57pub struct DocStore {
58    session: RefCell<Box<dyn Session>>,
59    schema: Schema,
60    domains: DomainCatalog,
61    limits: Limits,
62}
63impl DocStore {
64    /// Opens or creates an exactly compatible store.
65    pub fn create(path: &Path) -> StoreResult<Self> {
66        if !path.exists() {
67            std::fs::write(path, EMPTY_STORE).map_err(|e| StoreError::Host(e.to_string()))?;
68        }
69        Self::open(path, StorageAccess::ReadWrite)
70    }
71    /// Opens a legacy store without write authority or stamping it.
72    pub fn open_read_only(path: &Path) -> StoreResult<Self> {
73        Self::open(path, StorageAccess::ReadOnly)
74    }
75    fn open(path: &Path, access: StorageAccess) -> StoreResult<Self> {
76        let domains = domains()?;
77        let schema = document_schema(&domains)?;
78        let reference = Symbol::new("office-doc-store");
79        let driver = SqliteDriver::new(
80            domains.clone(),
81            PreopenedStores::new([(reference.clone(), path.to_path_buf())]),
82        );
83        let limits = Limits::new(100_000, 1_000_000, 256 * 1024 * 1024, 1_000_000)?;
84        let locator = Datum::Node {
85            tag: Symbol::qualified("relation", "preopened"),
86            fields: vec![
87                (Symbol::new("ref"), Datum::Symbol(reference)),
88                (
89                    Symbol::new("access"),
90                    Datum::Symbol(Symbol::new(if access == StorageAccess::ReadOnly {
91                        "read-only"
92                    } else {
93                        "read-write"
94                    })),
95                ),
96            ],
97        };
98        Ok(Self {
99            session: RefCell::new(driver.connect(&locator, &limits)?),
100            schema,
101            domains,
102            limits,
103        })
104    }
105    /// Saves a document snapshot.
106    pub fn save_doc(&self, doc: &Doc) -> StoreResult<()> {
107        self.upsert(
108            "docs",
109            &["id", "kind", "body"],
110            vec![
111                text(doc.id.as_str()),
112                text(doc.kind.as_str()),
113                text(encode_doc(doc)?),
114            ],
115        )
116    }
117    /// Loads a document snapshot by id.
118    pub fn load_doc(&self, id: &DocId) -> StoreResult<Option<Doc>> {
119        self.select(
120            "docs",
121            &["body"],
122            &[("id", text(id.as_str()))],
123            &[],
124            Some(1),
125        )?
126        .into_iter()
127        .next()
128        .map(|r| decode_doc(cell_text(&r, 0)?).map_err(StoreError::from))
129        .transpose()
130    }
131    /// Records a projected ledger commit.
132    pub fn project_commit(&self, doc: &DocId, edit: &Edit, seq: u64) -> StoreResult<u64> {
133        if &edit.doc != doc {
134            return Err(StoreError::Invalid(format!(
135                "edit targets {}, not {}",
136                edit.doc.as_str(),
137                doc.as_str()
138            )));
139        }
140        let signed = sequence(seq)?;
141        self.insert(
142            "edit_projection",
143            &["seq", "doc", "edit", "inverse"],
144            vec![
145                integer(signed),
146                text(doc.as_str()),
147                text(encode_edit(edit)?),
148                text(encode_edit(&edit.inverted())?),
149            ],
150        )?;
151        Ok(seq)
152    }
153    /// Returns the inverse edit for the latest projected ledger sequence.
154    pub fn undo_last(&self, doc: &DocId) -> StoreResult<Option<Edit>> {
155        self.select(
156            "edit_projection",
157            &["inverse"],
158            &[("doc", text(doc.as_str()))],
159            &[("seq", OrderDirection::Desc)],
160            Some(1),
161        )?
162        .into_iter()
163        .next()
164        .map(|r| decode_edit(cell_text(&r, 0)?).map_err(StoreError::from))
165        .transpose()
166    }
167    pub(crate) fn insert(
168        &self,
169        table: &str,
170        columns: &[&str],
171        cells: Vec<Cell>,
172    ) -> StoreResult<()> {
173        self.mutate(table, columns, cells, ConflictAction::Fail)
174    }
175    pub(crate) fn upsert(
176        &self,
177        table: &str,
178        columns: &[&str],
179        cells: Vec<Cell>,
180    ) -> StoreResult<()> {
181        let assignments = columns
182            .iter()
183            .zip(&cells)
184            .skip(1)
185            .map(|(column_name, value)| (column(column_name), Scalar::Literal(value.clone())))
186            .collect();
187        self.mutate(
188            table,
189            columns,
190            cells,
191            ConflictAction::DoUpdate {
192                target: ConflictTarget::PrimaryKey,
193                assignments,
194                predicate: None,
195            },
196        )
197    }
198    fn mutate(
199        &self,
200        table: &str,
201        columns: &[&str],
202        cells: Vec<Cell>,
203        conflict: ConflictAction,
204    ) -> StoreResult<()> {
205        let ty = row_type(columns, &cells)?;
206        let row = Row::new(ty.clone(), cells).map_err(|e| StoreError::Invalid(e.to_string()))?;
207        let raw = Mutation::Insert {
208            table: table_name(table),
209            columns: columns.iter().map(|c| column(c)).collect(),
210            input: Box::new(Rel::Values {
211                bind: binding("input"),
212                row_type: ty,
213                rows: vec![row],
214            }),
215            conflict,
216            returning: vec![],
217        };
218        let plan = admit_mutation(
219            raw,
220            &self.schema,
221            &self.domains,
222            RowType::new([]).unwrap(),
223            AdmissionLimits::default(),
224        )
225        .map_err(|e| StoreError::Invalid(e.to_string()))?;
226        let bindings = Bindings::new(&RowType::new([]).unwrap(), [])?;
227        self.session.borrow_mut().transaction(&mut |transaction| {
228            transaction.mutate(&plan, &bindings, &self.limits, &mut VecRowSink::default())?;
229            Ok(())
230        })?;
231        Ok(())
232    }
233    pub(crate) fn select(
234        &self,
235        table: &str,
236        columns: &[&str],
237        filters: &[(&str, Cell)],
238        order: &[(&str, OrderDirection)],
239        limit: Option<u64>,
240    ) -> StoreResult<Vec<Row>> {
241        let bind = "row";
242        let mut rel = Rel::Scan {
243            source: source("main"),
244            table: table_name(table),
245            bind: binding(bind),
246        };
247        for (name, value) in filters {
248            rel = Rel::Filter {
249                input: Box::new(rel),
250                predicate: Scalar::Call(
251                    ScalarOp::Eq,
252                    vec![field(bind, name), Scalar::Literal(value.clone())],
253                ),
254            };
255        }
256        if !order.is_empty() {
257            rel = Rel::Order {
258                input: Box::new(rel),
259                keys: order
260                    .iter()
261                    .map(|(n, d)| OrderKey {
262                        scalar: field(bind, n),
263                        direction: *d,
264                    })
265                    .collect(),
266            };
267        }
268        if limit.is_some() {
269            rel = Rel::Limit {
270                input: Box::new(rel),
271                count: limit,
272                offset: 0,
273            };
274        }
275        rel = Rel::Project {
276            input: Box::new(rel),
277            bind: binding("output"),
278            fields: columns
279                .iter()
280                .map(|n| NamedScalar {
281                    name: field_name(n),
282                    scalar: field(bind, n),
283                })
284                .collect(),
285        };
286        let plan = admit_query(
287            rel,
288            &self.schema,
289            &self.domains,
290            RowType::new([]).unwrap(),
291            AdmissionLimits::default(),
292        )
293        .map_err(|e| StoreError::Invalid(e.to_string()))?;
294        let bindings = Bindings::new(&RowType::new([]).unwrap(), [])?;
295        let mut sink = VecRowSink::default();
296        self.session
297            .borrow_mut()
298            .query(&plan, &bindings, &self.limits, &mut sink)?;
299        Ok(sink.into_rows())
300    }
301}
302pub(crate) fn text(v: impl Into<String>) -> Cell {
303    Cell::new(BaseDomain::Text.id(), Some(Datum::String(v.into())))
304}
305pub(crate) fn bytes(v: impl Into<Vec<u8>>) -> Cell {
306    Cell::new(BaseDomain::Bytes.id(), Some(Datum::Bytes(v.into())))
307}
308pub(crate) fn integer(v: i64) -> Cell {
309    Cell::new(
310        BaseDomain::I64.id(),
311        Some(Datum::Number(NumberLiteral {
312            domain: Symbol::qualified("core", "i64"),
313            canonical: v.to_string(),
314        })),
315    )
316}
317pub(crate) fn nullable_text(v: Option<String>) -> Cell {
318    v.map(text)
319        .unwrap_or_else(|| Cell::null(BaseDomain::Text.id()))
320}
321pub(crate) fn cell_text(r: &Row, i: usize) -> StoreResult<&str> {
322    match r.cells().get(i).and_then(Cell::value) {
323        Some(Datum::String(v)) => Ok(v),
324        _ => Err(StoreError::Storage(SiteError::Conversion)),
325    }
326}
327pub(crate) fn cell_bytes(r: &Row, i: usize) -> StoreResult<&[u8]> {
328    match r.cells().get(i).and_then(Cell::value) {
329        Some(Datum::Bytes(v)) => Ok(v),
330        _ => Err(StoreError::Storage(SiteError::Conversion)),
331    }
332}
333pub(crate) fn cell_optional_text(r: &Row, i: usize) -> StoreResult<Option<&str>> {
334    match r.cells().get(i).and_then(Cell::value) {
335        Some(Datum::String(v)) => Ok(Some(v)),
336        None => Ok(None),
337        _ => Err(StoreError::Storage(SiteError::Conversion)),
338    }
339}
340pub(crate) fn cell_i64(r: &Row, i: usize) -> StoreResult<i64> {
341    match r.cells().get(i).and_then(Cell::value) {
342        Some(Datum::Number(v)) => v
343            .canonical
344            .parse()
345            .map_err(|_| StoreError::Storage(SiteError::Conversion)),
346        _ => Err(StoreError::Storage(SiteError::Conversion)),
347    }
348}
349fn sequence(v: u64) -> StoreResult<i64> {
350    i64::try_from(v)
351        .map_err(|_| StoreError::Invalid(format!("sequence {v} exceeds signed relational domain")))
352}
353fn name<T: TryFrom<Symbol>>(v: &str) -> T
354where
355    T::Error: fmt::Debug,
356{
357    T::try_from(Symbol::new(v)).expect("static relation name")
358}
359fn table_name(v: &str) -> TableName {
360    name(v)
361}
362fn column(v: &str) -> ColumnName {
363    name(v)
364}
365fn field_name(v: &str) -> FieldName {
366    name(v)
367}
368fn binding(v: &str) -> BindingName {
369    name(v)
370}
371fn source(v: &str) -> SourceName {
372    name(v)
373}
374fn field(b: &str, n: &str) -> Scalar {
375    Scalar::Field(FieldRef {
376        binding: binding(b),
377        field: field_name(n),
378    })
379}
380fn row_type(ns: &[&str], cs: &[Cell]) -> StoreResult<RowType> {
381    RowType::new(ns.iter().zip(cs).map(|(n, c)| FieldType {
382        name: field_name(n),
383        domain: c.domain().clone(),
384        nullable: c.value().is_none(),
385    }))
386    .map_err(|e| StoreError::Invalid(e.to_string()))
387}
388fn domains() -> StoreResult<DomainCatalog> {
389    DomainCatalog::new([
390        BaseDomain::I64.spec(),
391        BaseDomain::Text.spec(),
392        BaseDomain::Bytes.spec(),
393    ])
394    .map_err(|e| StoreError::Invalid(e.to_string()))
395}
396
397/// Exact logical schema corresponding to the legacy DDL and normalized fixture.
398pub fn document_schema(domains: &DomainCatalog) -> StoreResult<Schema> {
399    let req = |n, d| ColumnBuilder::required(column(n), d).build();
400    let nul = |n, d| ColumnBuilder::nullable(column(n), d).build();
401    let pk = |t: &str, cs: &[&str]| {
402        Constraint::Primary(PrimaryKey {
403            name: name(&format!("{t}_pk")),
404            columns: cs.iter().map(|c| column(c)).collect(),
405        })
406    };
407    let ix = |n: &str, cs: &[&str]| Index {
408        name: name(n),
409        columns: cs.iter().map(|c| column(c)).collect(),
410        unique: false,
411    };
412    let docs = TableBuilder::new(table_name("docs"))
413        .column(req("id", BaseDomain::Text.id()))
414        .column(req("kind", BaseDomain::Text.id()))
415        .column(req("body", BaseDomain::Text.id()))
416        .constraint(pk("docs", &["id"]))
417        .build();
418    let edits = TableBuilder::new(table_name("edit_projection"))
419        .column(req("seq", BaseDomain::I64.id()))
420        .column(req("doc", BaseDomain::Text.id()))
421        .column(req("edit", BaseDomain::Text.id()))
422        .column(req("inverse", BaseDomain::Text.id()))
423        .constraint(pk("edit_projection", &["seq"]))
424        .index(ix("edit_projection_doc_seq", &["doc", "seq"]))
425        .build();
426    let ev = TableBuilder::new(table_name("evidence_facts"))
427        .column(req("subject", BaseDomain::Text.id()))
428        .column(req("predicate", BaseDomain::Text.id()))
429        .column(req("object", BaseDomain::Text.id()))
430        .column(req("captured_at_seq", BaseDomain::I64.id()))
431        .column(nul("immutable_hint", BaseDomain::Text.id()))
432        .constraint(pk(
433            "evidence_facts",
434            &["subject", "predicate", "object", "captured_at_seq"],
435        ))
436        .index(ix(
437            "evidence_facts_subject_seq",
438            &["subject", "captured_at_seq", "predicate", "object"],
439        ))
440        .build();
441    let cap = TableBuilder::new(table_name("web_captures"))
442        .column(req("capture_id", BaseDomain::Text.id()))
443        .column(req("source_uri", BaseDomain::Text.id()))
444        .column(req("body", BaseDomain::Bytes.id()))
445        .column(req("exchange_json", BaseDomain::Text.id()))
446        .constraint(pk("web_captures", &["capture_id"]))
447        .build();
448    let rep = TableBuilder::new(table_name("web_representations"))
449        .column(req("representation_id", BaseDomain::Text.id()))
450        .column(req("capture_id", BaseDomain::Text.id()))
451        .column(req("text", BaseDomain::Text.id()))
452        .column(req("metadata_json", BaseDomain::Text.id()))
453        .constraint(pk("web_representations", &["representation_id"]))
454        .constraint(Constraint::Foreign(ForeignKey {
455            name: name("web_representations_capture_fk"),
456            columns: vec![column("capture_id")],
457            target_table: table_name("web_captures"),
458            target_columns: vec![column("capture_id")],
459        }))
460        .build();
461    let anc = TableBuilder::new(table_name("web_evidence_anchors"))
462        .column(req("anchor_id", BaseDomain::Text.id()))
463        .column(req("subject", BaseDomain::Text.id()))
464        .column(req("representation_id", BaseDomain::Text.id()))
465        .column(req("record_json", BaseDomain::Text.id()))
466        .constraint(pk("web_evidence_anchors", &["anchor_id"]))
467        .constraint(Constraint::Foreign(ForeignKey {
468            name: name("web_evidence_anchors_representation_fk"),
469            columns: vec![column("representation_id")],
470            target_table: table_name("web_representations"),
471            target_columns: vec![column("representation_id")],
472        }))
473        .index(ix("web_evidence_anchor_subject", &["subject", "anchor_id"]))
474        .build();
475    SchemaBuilder::new(SchemaName::new(Symbol::new("main")).unwrap())
476        .table(docs)
477        .table(edits)
478        .table(ev)
479        .table(cap)
480        .table(rep)
481        .table(anc)
482        .build(domains, &AcceptAllValues)
483        .map_err(|e| StoreError::Invalid(e.to_string()))
484}
485
486/// Exact manifest required to adopt an unstamped v1 document-store file.
487pub fn legacy_adoption_manifest() -> StoreResult<AdoptionManifest> {
488    let domains = domains()?;
489    let logical_schema = document_schema(&domains)?
490        .id()
491        .map_err(|e| StoreError::Invalid(e.to_string()))?;
492    let physical_schema = legacy_physical_schema()?
493        .id()
494        .map_err(|e| StoreError::Invalid(e.to_string()))?;
495    Ok(AdoptionManifest {
496        logical_schema,
497        physical_schema,
498    })
499}
500
501fn legacy_physical_schema() -> StoreResult<PhysicalSchema> {
502    let col = |name_: &str, storage, nullable, ordinal| PhysicalColumn {
503        name: column(name_),
504        domain: match storage {
505            StorageRepr::I64 => BaseDomain::I64.id(),
506            StorageRepr::Text => BaseDomain::Text.id(),
507            _ => BaseDomain::Bytes.id(),
508        },
509        storage,
510        nullable,
511        ordinal,
512    };
513    let index = |name_: &str, columns: &[&str]| PhysicalIndex {
514        name: name::<IndexName>(name_),
515        columns: columns.iter().map(|v| column(v)).collect(),
516        unique: false,
517    };
518    let table = |name_: &str, columns, indexes| PhysicalTable {
519        name: table_name(name_),
520        columns,
521        indexes,
522    };
523    let tables = vec![
524        table(
525            "docs",
526            vec![
527                col("id", StorageRepr::Text, false, 0),
528                col("kind", StorageRepr::Text, false, 1),
529                col("body", StorageRepr::Text, false, 2),
530            ],
531            vec![],
532        ),
533        table(
534            "edit_projection",
535            vec![
536                col("seq", StorageRepr::I64, true, 0),
537                col("doc", StorageRepr::Text, false, 1),
538                col("edit", StorageRepr::Text, false, 2),
539                col("inverse", StorageRepr::Text, false, 3),
540            ],
541            vec![index("edit_projection_doc_seq", &["doc", "seq"])],
542        ),
543        table(
544            "evidence_facts",
545            vec![
546                col("subject", StorageRepr::Text, false, 0),
547                col("predicate", StorageRepr::Text, false, 1),
548                col("object", StorageRepr::Text, false, 2),
549                col("captured_at_seq", StorageRepr::I64, false, 3),
550                col("immutable_hint", StorageRepr::Text, true, 4),
551            ],
552            vec![index(
553                "evidence_facts_subject_seq",
554                &["subject", "captured_at_seq", "predicate", "object"],
555            )],
556        ),
557        table(
558            "web_captures",
559            vec![
560                col("capture_id", StorageRepr::Text, false, 0),
561                col("source_uri", StorageRepr::Text, false, 1),
562                col("body", StorageRepr::Bytes, false, 2),
563                col("exchange_json", StorageRepr::Text, false, 3),
564            ],
565            vec![],
566        ),
567        table(
568            "web_evidence_anchors",
569            vec![
570                col("anchor_id", StorageRepr::Text, false, 0),
571                col("subject", StorageRepr::Text, false, 1),
572                col("representation_id", StorageRepr::Text, false, 2),
573                col("record_json", StorageRepr::Text, false, 3),
574            ],
575            vec![index(
576                "web_evidence_anchor_subject",
577                &["subject", "anchor_id"],
578            )],
579        ),
580        table(
581            "web_representations",
582            vec![
583                col("representation_id", StorageRepr::Text, false, 0),
584                col("capture_id", StorageRepr::Text, false, 1),
585                col("text", StorageRepr::Text, false, 2),
586                col("metadata_json", StorageRepr::Text, false, 3),
587            ],
588            vec![],
589        ),
590    ];
591    PhysicalSchema::normalize(
592        ProviderName::new(Symbol::qualified("relation/provider", "sqlite")).unwrap(),
593        name::<SchemaName>("main"),
594        name::<RevisionName>("doc-store-v1"),
595        tables,
596    )
597    .map_err(|e| StoreError::Invalid(e.to_string()))
598}
sim_lib_doc_store/store.rs

sim_lib_doc_store/
store.rs
