Add typed crsql changes extraction

AGENTS.md

@@ -120,8 +120,9 @@ Roadmap items should be actionable and checkable:
   `cas cleanup` evicts unpinned blobs while retaining pinned blobs.
 - DB resources can be registered locally and report local-only status plus a
   read-only SQLite schema summary/hash and `crsql_changes` metadata when
-  present. cr-sqlite loading, typed change extraction, and sync are still
+  present. The DB crate can extract typed read-only `crsql_changes` batches for
-  roadmap work.
+  future sync messages. cr-sqlite loading, applying remote changes, and sync are
+  still roadmap work.
 - KV stores support local SQLite-backed create/set/get. Iroh Documents
   replication and command-level prefix-capability enforcement are still roadmap
   work. The auth evaluator already understands `kv.write_prefix:<prefix>`

README.md

@@ -88,7 +88,8 @@ The bootstrap implementation provides:
 - local filesystem CAS commands: `add`, `get`, `hash`, `has`, `pin`, `unpin`,
   `cleanup`, `list`
 - local DB resource registration: `geth db add <name> <path>` and
-  `geth db status <name>` with schema and `crsql_changes` metadata
+  `geth db status <name>` with schema and `crsql_changes` metadata; the DB
+  crate can extract typed local `crsql_changes` batches for future sync
 - local SQLite-backed KV commands: `geth kv create/set/get`
 - local JSON document commands: `geth document create/status/set/get`
 - local daemon-lifetime pubsub snapshots: `geth pubsub pub/sub`

crates/geth-db/src/lib.rs

@@ -1,4 +1,5 @@
 use geth_types::{DbId, ResourceId};
+use rusqlite::types::Value;
 use rusqlite::{Connection, OpenFlags};
 use serde::{Deserialize, Serialize};
 use std::path::Path;
@@ -25,6 +26,36 @@ pub struct CrSqliteChangeMetadata {
     pub error: Option<String>,
 }
 
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(tag = "type", content = "value", rename_all = "kebab-case")]
+pub enum SqliteValue {
+    Null,
+    Integer(i64),
+    Real(f64),
+    Text(String),
+    Blob(Vec<u8>),
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct CrSqliteChange {
+    pub table_name: String,
+    pub pk: SqliteValue,
+    pub column_id: String,
+    pub value: SqliteValue,
+    pub column_version: i64,
+    pub db_version: i64,
+    pub site_id: Option<Vec<u8>>,
+    pub causal_length: Option<i64>,
+    pub sequence: Option<i64>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct CrSqliteChangeBatch {
+    pub schema_metadata: String,
+    pub max_db_version: Option<i64>,
+    pub changes: Vec<CrSqliteChange>,
+}
+
 impl CrSqliteChangeMetadata {
     #[must_use]
     pub fn unavailable() -> Self {
@@ -55,6 +86,10 @@ pub enum DbError {
     InvalidName(String),
     #[error("sqlite error: {0}")]
     Sqlite(#[from] rusqlite::Error),
+    #[error("crsql_changes table or view is missing")]
+    MissingCrSqliteChanges,
+    #[error("crsql_changes is missing required column: {0}")]
+    MissingCrSqliteColumn(&'static str),
 }
 
 pub fn validate_db_name(name: &str) -> Result<(), DbError> {
@@ -149,6 +184,102 @@ pub fn crsqlite_change_metadata(path: &Path) -> Result<CrSqliteChangeMetadata, D
     })
 }
 
+pub fn extract_crsqlite_changes(
+    path: &Path,
+    after_db_version: Option<i64>,
+    limit: u32,
+) -> Result<CrSqliteChangeBatch, DbError> {
+    let schema_metadata = schema_metadata(path)?;
+    let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
+    let columns = crsqlite_change_columns(&conn)?;
+    require_crsqlite_column(&columns, "table_name")?;
+    require_crsqlite_column(&columns, "pk")?;
+    require_crsqlite_column(&columns, "cid")?;
+    require_crsqlite_column(&columns, "val")?;
+    require_crsqlite_column(&columns, "col_version")?;
+    require_crsqlite_column(&columns, "db_version")?;
+
+    let site_id_expr = optional_crsqlite_column_expr(&columns, "site_id");
+    let causal_length_expr = optional_crsqlite_column_expr(&columns, "cl");
+    let sequence_expr = optional_crsqlite_column_expr(&columns, "seq");
+    let bounded_limit = limit.max(1);
+    let sql = format!(
+        r#"SELECT table_name, pk, cid, val, col_version, db_version, {site_id_expr}, {causal_length_expr}, {sequence_expr}
+           FROM crsql_changes
+           WHERE (?1 IS NULL OR db_version > ?1)
+           ORDER BY db_version, table_name, pk, cid
+           LIMIT ?2"#
+    );
+    let mut stmt = conn.prepare(&sql)?;
+    let rows = stmt.query_map((after_db_version, i64::from(bounded_limit)), |row| {
+        Ok(CrSqliteChange {
+            table_name: row.get(0)?,
+            pk: sqlite_value(row.get(1)?),
+            column_id: row.get(2)?,
+            value: sqlite_value(row.get(3)?),
+            column_version: row.get(4)?,
+            db_version: row.get(5)?,
+            site_id: row.get(6)?,
+            causal_length: row.get(7)?,
+            sequence: row.get(8)?,
+        })
+    })?;
+    let changes = rows.collect::<Result<Vec<_>, _>>()?;
+    let max_db_version = changes.iter().map(|change| change.db_version).max();
+
+    Ok(CrSqliteChangeBatch {
+        schema_metadata,
+        max_db_version,
+        changes,
+    })
+}
+
+fn crsqlite_change_columns(conn: &Connection) -> Result<Vec<String>, DbError> {
+    let available: bool = conn.query_row(
+        r#"SELECT EXISTS(
+            SELECT 1 FROM sqlite_master
+            WHERE name = 'crsql_changes' AND type IN ('table', 'view')
+        )"#,
+        [],
+        |row| row.get(0),
+    )?;
+    if !available {
+        return Err(DbError::MissingCrSqliteChanges);
+    }
+
+    let mut columns_stmt = conn.prepare("PRAGMA table_info('crsql_changes')")?;
+    columns_stmt
+        .query_map([], |row| row.get::<_, String>(1))?
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(DbError::from)
+}
+
+fn require_crsqlite_column(columns: &[String], column: &'static str) -> Result<(), DbError> {
+    if columns.iter().any(|candidate| candidate == column) {
+        Ok(())
+    } else {
+        Err(DbError::MissingCrSqliteColumn(column))
+    }
+}
+
+fn optional_crsqlite_column_expr(columns: &[String], column: &'static str) -> &'static str {
+    if columns.iter().any(|candidate| candidate == column) {
+        column
+    } else {
+        "NULL"
+    }
+}
+
+fn sqlite_value(value: Value) -> SqliteValue {
+    match value {
+        Value::Null => SqliteValue::Null,
+        Value::Integer(value) => SqliteValue::Integer(value),
+        Value::Real(value) => SqliteValue::Real(value),
+        Value::Text(value) => SqliteValue::Text(value),
+        Value::Blob(value) => SqliteValue::Blob(value),
+    }
+}
+
 #[must_use]
 pub fn crsqlite_sync_roadmap() -> &'static str {
     "future db sync reads crsql_changes, exchanges changes over Iroh, and applies through crsql_changes"
@@ -225,4 +356,87 @@ mod tests {
         assert_eq!(metadata.max_db_version, Some(7));
         assert!(metadata.columns.contains(&"db_version".to_owned()));
     }
+
+    #[test]
+    fn extract_crsqlite_changes_returns_typed_batch_with_schema_metadata() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join("notes.sqlite");
+        let conn = Connection::open(&path).expect("open sqlite");
+        conn.execute("CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT)", [])
+            .expect("create notes");
+        conn.execute(
+            r#"CREATE TABLE crsql_changes(
+                table_name TEXT NOT NULL,
+                pk BLOB NOT NULL,
+                cid TEXT NOT NULL,
+                val BLOB,
+                col_version INTEGER NOT NULL,
+                db_version INTEGER NOT NULL,
+                site_id BLOB,
+                cl INTEGER,
+                seq INTEGER
+            )"#,
+            [],
+        )
+        .expect("create crsql_changes table");
+        conn.execute(
+            "INSERT INTO crsql_changes(table_name, pk, cid, val, col_version, db_version, site_id, cl, seq) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
+            (
+                "notes",
+                vec![1_u8],
+                "body",
+                Vec::from("hello".as_bytes()),
+                1_i64,
+                1_i64,
+                vec![9_u8],
+                10_i64,
+                11_i64,
+            ),
+        )
+        .expect("insert first change");
+        conn.execute(
+            "INSERT INTO crsql_changes(table_name, pk, cid, val, col_version, db_version, site_id, cl, seq) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
+            (
+                "notes",
+                vec![2_u8],
+                "body",
+                Vec::from("world".as_bytes()),
+                1_i64,
+                2_i64,
+                vec![9_u8],
+                12_i64,
+                13_i64,
+            ),
+        )
+        .expect("insert second change");
+        drop(conn);
+
+        let batch = extract_crsqlite_changes(&path, Some(1), 10).expect("extract batch");
+
+        assert!(batch.schema_metadata.contains("schema_hash="));
+        assert_eq!(batch.max_db_version, Some(2));
+        assert_eq!(batch.changes.len(), 1);
+        assert_eq!(batch.changes[0].table_name, "notes");
+        assert_eq!(batch.changes[0].pk, SqliteValue::Blob(vec![2]));
+        assert_eq!(batch.changes[0].column_id, "body");
+        assert_eq!(
+            batch.changes[0].value,
+            SqliteValue::Blob(Vec::from("world".as_bytes()))
+        );
+        assert_eq!(batch.changes[0].site_id, Some(vec![9]));
+        assert_eq!(batch.changes[0].causal_length, Some(12));
+        assert_eq!(batch.changes[0].sequence, Some(13));
+    }
+
+    #[test]
+    fn extract_crsqlite_changes_requires_change_table() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join("notes.sqlite");
+        Connection::open(&path).expect("open sqlite");
+
+        assert!(matches!(
+            extract_crsqlite_changes(&path, None, 10),
+            Err(DbError::MissingCrSqliteChanges)
+        ));
+    }
 }

docs/architecture.md

@@ -96,8 +96,10 @@ and file sync trees are future work.
 `geth-db` currently registers local SQLite paths as DB resources and reports
 local-only sync status plus a read-only SQLite schema summary/hash. It also
 inspects `crsql_changes` metadata when that table or view exists, reporting
-change count, columns, and max `db_version`. Loading cr-sqlite, extracting
+change count, columns, and max `db_version`. The crate can extract read-only
-change batches, applying changes, and DB sync are future work.
+typed change batches from `crsql_changes` with schema metadata for future sync
+messages. Loading cr-sqlite, applying remote changes, and DB sync are future
+work.
 
 `geth-kv` currently provides a SQLite-backed local fallback for named KV stores
 through `kv create/set/get`. Iroh Documents namespaces, prefix authorization

docs/roadmap.md

@@ -312,8 +312,10 @@ Automerge documents.
   - `[x]` DB status reports `crsql_changes` row count, columns, and max
     `db_version` when available.
   - `[x]` Tests use a temp SQLite DB and deterministic fixture changes.
-  - `[ ]` The module can extract typed change batches from `crsql_changes`.
+  - `[x]` The module can extract typed change batches from `crsql_changes`.
-  - `[ ]` Schema hash/version metadata is included in sync batches.
+  - `[x]` Schema hash/version metadata is included in extracted change batches.
+  - `[ ]` Extracted batches are exposed through a daemon control command or sync
+    protocol.
 
 - `[ ]` DB sync over Iroh.
   Acceptance criteria: