feat: add incremental semantic projection families

src/bench_support/derived_access/adapter.rs

@@ -5,16 +5,23 @@ use std::path::{Path, PathBuf};
 
 use super::sqlite_cursor::{CursorLedgerError, CursorLedgerIdentity, SqliteCursorLedger};
 use super::sqlite_locator::{LocatorInventory, SqliteLocator, SqliteLocatorError};
+use super::sqlite_semantic::{SemanticInventory, SqliteSemantic, SqliteSemanticError};
 use crate::canonical_hash::sha256_bytes_hex;
+use crate::model::RevisionId;
 use crate::session::EventStore;
-use crate::session::derived_access::cursor::{AppendResolution, TruthCursor, TruthHead};
+use crate::session::derived_access::cursor::{
+    AppendResolution, CursorDelta, TruthCursor, TruthHead,
+};
 use crate::session::derived_access::locator::{
     ChronologicalWindowRequest, HydratedWindow, LocatorModelError, LocatorRead, LocatorRow,
 };
 use crate::session::derived_access::semantic::state::{
     DerivedAccessFreshness, FreshnessModelError,
 };
-use crate::session::event::ShoreEvent;
+use crate::session::derived_access::semantic::{
+    HydratedRevisionDetail, SemanticFact, SemanticModelError, SemanticSnapshot,
+};
+use crate::session::event::{ShoreEvent, WorkObjectProposal, WorkObjectProposedPayload};
 
 const DEFAULT_DELTA_LIMIT: usize = 512;
 
@@ -23,6 +30,7 @@ pub(crate) struct QualificationDerivedAccessAdapter {
     store_root: PathBuf,
     cursor: SqliteCursorLedger,
     locator: SqliteLocator,
+    semantic: SqliteSemantic,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -32,13 +40,19 @@ pub(crate) enum DerivedAccessAdapterError {
     #[error(transparent)]
     Locator(#[from] SqliteLocatorError),
     #[error(transparent)]
+    Semantic(#[from] SqliteSemanticError),
+    #[error(transparent)]
+    SemanticModel(#[from] SemanticModelError),
+    #[error(transparent)]
     LocatorModel(#[from] LocatorModelError),
     #[error(transparent)]
     Freshness(#[from] FreshnessModelError),
     #[error("authoritative truth read failed: {0}")]
     Truth(String),
     #[error("authoritative event does not match locator row at {0:?}")]
     LocatorMismatch(TruthCursor),
+    #[error("authoritative event does not match semantic fact at {0:?}")]
+    SemanticMismatch(TruthCursor),
     #[error("derived catch-up batch limit must be greater than zero")]
     ZeroBatchLimit,
     #[error("derived catch-up returned no receipts before observed head {0:?}")]
@@ -55,10 +69,12 @@ impl QualificationDerivedAccessAdapter {
             .map_err(|error| DerivedAccessAdapterError::Truth(error.to_string()))?;
         let cursor = SqliteCursorLedger::open(&store_root, identity)?;
         let locator = SqliteLocator::open(&store_root)?;
+        let semantic = SqliteSemantic::open(locator.clone())?;
         Ok(Self {
             store_root,
             cursor,
             locator,
+            semantic,
         })
     }
 
@@ -91,20 +107,8 @@ impl QualificationDerivedAccessAdapter {
                     delta.observed_head,
                 ));
             }
-            let rows = delta
-                .receipts
-                .iter()
-                .map(|receipt| {
-                    let event = self.read_authoritative(&receipt.logical_reread_key)?;
-                    LocatorRow::from_event(
-                        receipt.cursor,
-                        &event,
-                        receipt.validation_witness.clone(),
-                    )
-                    .map_err(Into::into)
-                })
-                .collect::<Result<Vec<_>, DerivedAccessAdapterError>>()?;
-            let applied = self.locator.apply_delta(&delta, &rows)?.applied;
+            let (rows, semantic_facts) = self.derived_rows(&delta)?;
+            let applied = self.semantic.apply_delta(&delta, &rows, &semantic_facts)?;
             if delta.complete {
                 return Ok(applied);
             }
@@ -174,6 +178,126 @@ impl QualificationDerivedAccessAdapter {
         Ok(self.locator.inventory()?)
     }
 
+    pub(crate) fn semantic_inventory(
+        &self,
+    ) -> Result<SemanticInventory, DerivedAccessAdapterError> {
+        Ok(self.semantic.inventory()?)
+    }
+
+    pub(crate) fn semantic_audit_snapshot(
+        &self,
+    ) -> Result<LocatorRead<SemanticSnapshot>, DerivedAccessAdapterError> {
+        let observed = self.cursor.head()?.cursor;
+        Ok(self.semantic.audit_snapshot(observed)?)
+    }
+
+    pub(crate) fn semantic_materialized_audit_snapshot(
+        &self,
+    ) -> Result<LocatorRead<SemanticSnapshot>, DerivedAccessAdapterError> {
+        let observed = self.cursor.head()?.cursor;
+        Ok(self.semantic.materialized_audit_snapshot(observed)?)
+    }
+
+    pub(crate) fn semantic_materialized_engagement_snapshot(
+        &self,
+        engagement_id: &str,
+    ) -> Result<LocatorRead<SemanticSnapshot>, DerivedAccessAdapterError> {
+        let observed = self.cursor.head()?.cursor;
+        Ok(self
+            .semantic
+            .materialized_engagement_snapshot(engagement_id, observed)?)
+    }
+
+    pub(crate) fn revision_detail(
+        &self,
+        revision_id: &RevisionId,
+    ) -> Result<LocatorRead<Option<HydratedRevisionDetail>>, DerivedAccessAdapterError> {
+        let observed = self.cursor.head()?.cursor;
+        let facts = match self
+            .semantic
+            .facts_for_revision(revision_id.as_str(), observed)?
+        {
+            LocatorRead::Ready(facts) => facts,
+            LocatorRead::CatchUpRequired { applied, observed } => {
+                return Ok(LocatorRead::CatchUpRequired { applied, observed });
+            }
+        };
+        if facts.is_empty() {
+            return Ok(LocatorRead::Ready(None));
+        }
+
+        let mut authoritative_events = facts
+            .iter()
+            .map(|fact| self.hydrate_semantic(fact))
+            .collect::<Result<Vec<_>, _>>()?;
+        authoritative_events.sort_by(|left, right| left.event_id.cmp(&right.event_id));
+        let capture = authoritative_events
+            .iter()
+            .find(|event| event.event_type == crate::session::event::EventType::WorkObjectProposed)
+            .and_then(|event| {
+                serde_json::from_value::<WorkObjectProposedPayload>(event.payload.clone()).ok()
+            })
+            .and_then(|payload| match payload.work_object {
+                WorkObjectProposal::Revision {
+                    revision,
+                    object_artifact_content_hash,
+                    ..
+                } if revision.id == *revision_id => Some(object_artifact_content_hash),
+                _ => None,
+            });
+        let Some(object_content_hash) = capture else {
+            return Ok(LocatorRead::Ready(None));
+        };
+        let object_content_removed = self
+            .semantic
+            .content_is_removed(&object_content_hash, observed)?;
+        Ok(LocatorRead::Ready(Some(HydratedRevisionDetail {
+            as_of: observed,
+            revision_id: revision_id.clone(),
+            object_content_hash,
+            object_content_removed,
+            authoritative_events,
+        })))
+    }
+
+    #[cfg(test)]
+    pub(crate) fn catch_up_with_semantic_failure_for_test(
+        &self,
+        batch_limit: usize,
+    ) -> Result<TruthCursor, DerivedAccessAdapterError> {
+        if batch_limit == 0 {
+            return Err(DerivedAccessAdapterError::ZeroBatchLimit);
+        }
+        let checkpoint = self.locator.checkpoint()?;
+        let delta = self.cursor.events_after(checkpoint.applied, batch_limit)?;
+        let (rows, semantic_facts) = self.derived_rows(&delta)?;
+        Ok(self
+            .semantic
+            .apply_delta_with_failure(&delta, &rows, &semantic_facts)?)
+    }
+
+    fn derived_rows(
+        &self,
+        delta: &CursorDelta,
+    ) -> Result<(Vec<LocatorRow>, Vec<SemanticFact>), DerivedAccessAdapterError> {
+        let mut locator_rows = Vec::with_capacity(delta.receipts.len());
+        let mut semantic_facts = Vec::with_capacity(delta.receipts.len());
+        for receipt in &delta.receipts {
+            let event = self.read_authoritative(&receipt.logical_reread_key)?;
+            locator_rows.push(LocatorRow::from_event(
+                receipt.cursor,
+                &event,
+                receipt.validation_witness.clone(),
+            )?);
+            semantic_facts.push(SemanticFact::from_event(
+                receipt.cursor,
+                &event,
+                receipt.validation_witness.clone(),
+            )?);
+        }
+        Ok((locator_rows, semantic_facts))
+    }
+
     fn hydrate(&self, row: &LocatorRow) -> Result<ShoreEvent, DerivedAccessAdapterError> {
         let event = self.read_authoritative(&row.logical_reread_key)?;
         let witness = sha256_bytes_hex(
@@ -187,6 +311,22 @@ impl QualificationDerivedAccessAdapter {
         Ok(event)
     }
 
+    fn hydrate_semantic(
+        &self,
+        fact: &SemanticFact,
+    ) -> Result<ShoreEvent, DerivedAccessAdapterError> {
+        let event = self.read_authoritative(&fact.logical_reread_key)?;
+        let witness = sha256_bytes_hex(
+            &serde_json::to_vec(&event)
+                .map_err(|error| DerivedAccessAdapterError::Truth(error.to_string()))?,
+        );
+        let observed = SemanticFact::from_event(fact.cursor, &event, witness)?;
+        if &observed != fact {
+            return Err(DerivedAccessAdapterError::SemanticMismatch(fact.cursor));
+        }
+        Ok(event)
+    }
+
     fn read_authoritative(
         &self,
         logical_reread_key: &str,

src/bench_support/derived_access/mod.rs

@@ -2,6 +2,7 @@ mod adapter;
 mod contract;
 mod sqlite_cursor;
 mod sqlite_locator;
+mod sqlite_semantic;
 mod writer_lock;
 
 pub use contract::*;
@@ -10,3 +11,5 @@ pub use contract::*;
 mod sqlite_cursor_tests;
 #[cfg(test)]
 mod sqlite_locator_tests;
+#[cfg(test)]
+mod sqlite_semantic_tests;

src/bench_support/derived_access/sqlite_locator.rs

@@ -96,10 +96,11 @@ impl SqliteLocator {
         read_locator_checkpoint(&connection)
     }
 
-    pub(crate) fn apply_delta(
+    pub(super) fn apply_delta_with(
         &self,
         delta: &CursorDelta,
         rows: &[LocatorRow],
+        apply_semantic: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<(), SqliteLocatorError>,
     ) -> Result<LocatorCheckpoint, SqliteLocatorError> {
         if rows.len() != delta.receipts.len() {
             return Err(SqliteLocatorError::Delta(format!(
@@ -151,6 +152,7 @@ impl SqliteLocator {
         for row in rows {
             insert_locator_row(&transaction, row)?;
         }
+        apply_semantic(&transaction)?;
         let updated = transaction
             .execute(
                 "UPDATE locator_checkpoint
@@ -458,7 +460,7 @@ impl SqliteLocator {
         Ok(connection)
     }
 
-    fn validated_connection(&self) -> Result<Connection, SqliteLocatorError> {
+    pub(super) fn validated_connection(&self) -> Result<Connection, SqliteLocatorError> {
         let connection = self.connection()?;
         let cursor = validate_cursor_metadata(&connection)?;
         validate_locator_checkpoint(&connection, &cursor)?;
@@ -612,7 +614,7 @@ fn validate_locator_checkpoint(
     Ok(())
 }
 
-fn read_locator_checkpoint(
+pub(super) fn read_locator_checkpoint(
     connection: &Connection,
 ) -> Result<LocatorCheckpoint, SqliteLocatorError> {
     connection