kmp-adapter-embedded 0.18.1

//! The storage seam ([historical ADR-018](https://github.com/underpass-ai/kmp/blob/v0.5.0/archive/docs/adr/ADR-018-multi-process-embedded-store.md)).
//!
//! Everything the kernel ports need from a storage engine, and nothing an
//! engine would need to know about the kernel: fourteen key-to-bytes maps,
//! transactions over them, and four key shapes. Port logic — graph
//! traversal, revision checks, idempotency — is written once against this
//! and never sees an engine type.
//!
//! The seam is deliberately narrow. Every method here corresponds to an
//! operation the port code performs against SQLite, and no more:
//! point get, the stored byte length of one value, insert, remove, a full
//! ordered scan, a scan of one first key component, the last row of a
//! `u64`-keyed table, a row count, and a table clear. Rows come back in ascending key order, compared component by
//! component and byte-wise within a component; the neighborhood output
//! depends on it.
//!
//! Scans return `Vec` rather than an iterator. Every caller collected before
//! this seam existed, so nothing is lost, and it keeps the trait object-safe
//! without a lifetime tying a row to its transaction.

use std::fmt;

use kmp_domain::PortError;

mod snapshot_revision_observer;
pub(crate) mod sqlite;
mod sqlite_snapshot;
mod sqlite_snapshot_read;

/// The tables a kernel store consists of. Every engine materializes all of
/// them; the key shape of each is fixed and recorded on the variant.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum Table {
    /// Graph nodes: `node_id -> NodeRecord`.
    Nodes,
    /// Outgoing adjacency: `(source, target, relation_type) -> explanation`.
    Relations,
    /// Incoming adjacency index: `(target, source, relation_type) -> ()`.
    RelationsByTarget,
    /// Node details: `node_id -> DetailRecord`.
    Details,
    /// Canonical body headers: `node_id -> DetailHeaderRecord`. Written in the
    /// same transaction as `Details`, so a detail without its header is an
    /// inconsistent projection, never a body that does not exist.
    DetailHeaders,
    /// Reader-authored compact cards: `(node_id, language) -> CardRecord`.
    /// A derived view beside the canonical body, never inside it.
    Cards,
    /// Memory anchor index: `node_id -> ()`.
    Anchors,
    /// Append-only context event log: `sequence -> ContextUpdatedEvent`.
    EventLog,
    /// Aggregate heads: `"root\u{1f}role" -> AggregateRecord`.
    Aggregates,
    /// Idempotency outcomes: `key -> IdempotentOutcome`.
    Idempotency,
    /// Projection-consumer dedup: `(consumer, event_id) -> ()`.
    Processed,
    /// Projection checkpoints: `(consumer, stream) -> CheckpointRecord`.
    Checkpoints,
    /// Snapshot audit records: `(root, role) -> snapshot summary`.
    Snapshots,
    /// Historical migration receipts retained for data-file compatibility.
    Migrations,
}

impl Table {
    /// The key shape this table is defined with. A call carrying a key of a
    /// different shape is a programming error inside this crate, and the
    /// engine reports it as one instead of guessing.
    pub(crate) const fn key_shape(self) -> KeyShape {
        match self {
            Table::Nodes
            | Table::Details
            | Table::DetailHeaders
            | Table::Anchors
            | Table::Aggregates
            | Table::Idempotency
            | Table::Migrations => KeyShape::Str,
            Table::Cards | Table::Processed | Table::Checkpoints | Table::Snapshots => {
                KeyShape::Str2
            }
            Table::Relations | Table::RelationsByTarget => KeyShape::Str3,
            Table::EventLog => KeyShape::U64,
        }
    }
}

impl fmt::Display for Table {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Table::Nodes => "nodes",
            Table::Relations => "relations_by_source",
            Table::RelationsByTarget => "relations_by_target",
            Table::Details => "details",
            Table::DetailHeaders => "detail_headers",
            Table::Cards => "node_cards",
            Table::Anchors => "memory_anchors",
            Table::EventLog => "event_log",
            Table::Aggregates => "aggregates",
            Table::Idempotency => "idempotency",
            Table::Processed => "processed_events",
            Table::Checkpoints => "projection_checkpoints",
            Table::Snapshots => "snapshots",
            Table::Migrations => "store_migrations",
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum KeyShape {
    Str,
    Str2,
    Str3,
    U64,
}

/// A borrowed key in one of the four shapes the tables use.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Key<'a> {
    Str(&'a str),
    Str2(&'a str, &'a str),
    Str3(&'a str, &'a str, &'a str),
    U64(u64),
}

impl Key<'_> {
    pub(crate) const fn shape(&self) -> KeyShape {
        match self {
            Key::Str(_) => KeyShape::Str,
            Key::Str2(..) => KeyShape::Str2,
            Key::Str3(..) => KeyShape::Str3,
            Key::U64(_) => KeyShape::U64,
        }
    }
}

/// A row read back from a `Str`-keyed table.
pub(crate) type StrRow = (String, Vec<u8>);
/// A row read back from a `Str3`-keyed table.
pub(crate) type Str3Row = ((String, String, String), Vec<u8>);
/// A row read back from a `U64`-keyed table.
pub(crate) type U64Row = (u64, Vec<u8>);

/// A read transaction: a consistent snapshot of every table.
pub(crate) trait ReadTx {
    /// The value at `key`, if any. Unit-valued tables answer `Some(vec![])`
    /// for a present key.
    fn get(&self, table: Table, key: Key<'_>) -> Result<Option<Vec<u8>>, PortError>;

    /// Project named paths from a JSON object without returning discarded
    /// fields. Paths consist of dot-separated object keys. This is an internal
    /// storage operation; field selection belongs to the adapter, not SQL.
    fn project_str_json(
        &self,
        table: Table,
        key: &str,
        fields: &[&str],
    ) -> Result<Option<Vec<u8>>, PortError> {
        let Some(raw) = self.get(table, Key::Str(key))? else {
            return Ok(None);
        };
        let value: serde_json::Value = serde_json::from_slice(&raw)
            .map_err(|e| PortError::InvalidState(format!("invalid JSON record: {e}")))?;
        let mut output = serde_json::Map::new();
        for field in fields {
            let mut selected = &value;
            for component in field.split('.') {
                selected = selected.get(component).unwrap_or(&serde_json::Value::Null);
            }
            output.insert((*field).to_string(), selected.clone());
        }
        serde_json::to_vec(&output)
            .map(Some)
            .map_err(|e| PortError::InvalidState(e.to_string()))
    }

    /// Stored byte length of the value at `key`, without loading or decoding
    /// it. `None` means the key is absent, which is not the same answer as
    /// `Some(0)` for a present unit value. This measures the stored record,
    /// including whatever envelope the record carries, never a field inside
    /// it. A value that is not stored as a blob is reported as an error
    /// rather than measured, because a text value would be counted in
    /// characters and silently under-report its bytes.
    fn value_len(&self, table: Table, key: Key<'_>) -> Result<Option<u64>, PortError>;

    /// Every row of a `Str`-keyed table, ascending.
    fn scan_str(&self, table: Table) -> Result<Vec<StrRow>, PortError>;

    /// Every row of a `Str3`-keyed table whose first component equals
    /// `first`, ascending by the remaining components. This is the adjacency
    /// query: all edges out of (or into) one node.
    fn scan_str3_by_first(&self, table: Table, first: &str) -> Result<Vec<Str3Row>, PortError>;

    /// Keyset page of an adjacency index. Reads at most limit values; never
    /// loads the complete adjacency or probes an extra row to report its end.
    fn scan_str3_page(
        &self,
        table: Table,
        first: &str,
        after: Option<(&str, &str)>,
        limit: u32,
        relation_type: Option<&str>,
    ) -> Result<Vec<Str3Row>, PortError>;

    /// Every row of a `U64`-keyed table, ascending.
    fn scan_u64(&self, table: Table) -> Result<Vec<U64Row>, PortError>;

    /// The highest-keyed row of a `U64`-keyed table.
    fn last_u64(&self, table: Table) -> Result<Option<U64Row>, PortError>;

    /// Number of rows in `table`.
    fn count(&self, table: Table) -> Result<u64, PortError>;
}

/// A write transaction. Reads see this transaction's own writes; nothing is
/// durable until [`commit`](WriteTx::commit) returns, and dropping the
/// transaction discards everything it did.
pub(crate) trait WriteTx: ReadTx {
    /// Upsert. Unit-valued tables ignore `value`.
    fn insert(&mut self, table: Table, key: Key<'_>, value: &[u8]) -> Result<(), PortError>;

    /// Remove `key` if present. Removing an absent key is not an error.
    fn remove(&mut self, table: Table, key: Key<'_>) -> Result<(), PortError>;

    /// Empty `table`. The table still exists afterwards: a reader following
    /// this commit finds it empty, never missing.
    fn clear(&mut self, table: Table) -> Result<(), PortError>;

    /// Make every write in this transaction durable. On return the data has
    /// reached the disk with the durability the engine was opened with.
    fn commit(self: Box<Self>) -> Result<(), PortError>;
}

/// A storage engine: one opened kernel store, shareable across tasks.
pub(crate) trait Engine: fmt::Debug + Send + Sync {
    fn graph_read_revision(&self) -> Option<kmp_domain::GraphReadRevision> {
        None
    }

    fn read_snapshot(&self) -> Result<std::sync::Arc<dyn Engine>, PortError> {
        Err(PortError::Unavailable(
            "read snapshots are not supported by this engine".into(),
        ))
    }
    fn begin_read(&self) -> Result<Box<dyn ReadTx + '_>, PortError>;
    fn begin_write(&self) -> Result<Box<dyn WriteTx + '_>, PortError>;
}

pub(crate) fn key_shape_mismatch(table: Table, key: KeyShape) -> PortError {
    PortError::InvalidState(format!(
        "embedded engine: table `{table}` is keyed by {:?}, called with a {key:?} key",
        table.key_shape()
    ))
}

pub(crate) fn scan_shape_mismatch(table: Table, wanted: KeyShape) -> PortError {
    PortError::InvalidState(format!(
        "embedded engine: table `{table}` is keyed by {:?}, scanned as {wanted:?}",
        table.key_shape()
    ))
}