diff --git a/AGENTS.md b/AGENTS.md index bd467d5..b62330a 100644 --- a/AGENTS.md +++ b/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 - roadmap work. + present. The DB crate can extract typed read-only `crsql_changes` batches for + 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:` diff --git a/README.md b/README.md index 6d89a56..17a69e3 100644 --- a/README.md +++ b/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 ` and - `geth db status ` with schema and `crsql_changes` metadata + `geth db status ` 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` diff --git a/crates/geth-db/src/lib.rs b/crates/geth-db/src/lib.rs index c3be119..bf24d66 100644 --- a/crates/geth-db/src/lib.rs +++ b/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, } +#[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), +} + +#[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>, + pub causal_length: Option, + pub sequence: Option, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct CrSqliteChangeBatch { + pub schema_metadata: String, + pub max_db_version: Option, + pub changes: Vec, +} + 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, + limit: u32, +) -> Result { + 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::, _>>()?; + 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, 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::, _>>() + .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) + )); + } } diff --git a/docs/architecture.md b/docs/architecture.md index b2e86ec..dcc91a3 100644 --- a/docs/architecture.md +++ b/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 batches, applying changes, and DB sync are future work. +change count, columns, and max `db_version`. The crate can extract read-only +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 diff --git a/docs/roadmap.md b/docs/roadmap.md index aca8a30..94af58d 100644 --- a/docs/roadmap.md +++ b/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`. - - `[ ]` Schema hash/version metadata is included in sync batches. + - `[x]` The module can extract typed change batches from `crsql_changes`. + - `[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: