diff --git a/AGENTS.md b/AGENTS.md index edb9036..9d2e267 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,8 +147,10 @@ Roadmap items should be actionable and checkable: batches through `geth db changes`. `geth db sync ` can pull authorized typed `crsql_changes` batches over the protected Iroh control ALPN with `db.sync` on the remote `resource:db:`, compare schema metadata, - and advance per-peer/per-DB cursors during manual or background live-sync. - cr-sqlite loading and applying remote changes are still roadmap work. + apply compatible batches into local `crsql_changes`, and advance + per-peer/per-DB cursors during manual or background live-sync. Loading and + configuring cr-sqlite for real application databases is still outside the + bootstrap. - KV stores support local SQLite-backed create/set/get plus authorized `geth kv sync ` over Iroh. Background live-sync refreshes local KV stores from known peers with per-peer/per-KV cursors. Iroh Documents diff --git a/README.md b/README.md index 667ce8c..b18ac52 100644 --- a/README.md +++ b/README.md @@ -188,10 +188,10 @@ older than the local document timestamp. DB sync is a staged cr-sqlite path: manual `geth db sync ` and background live-sync require `db.sync` on `resource:db:`, exchange typed `crsql_changes` batches over the protected Iroh control path, and check remote -schema metadata against the local DB before advancing the per-peer cursor. -Applying remote changes through cr-sqlite is still future work, so the current -prototype is useful for validating auth, schema gating, and live change -exchange without mutating the local application database. +schema metadata against the local DB before applying and advancing the per-peer +cursor. Compatible batches are inserted into the local `crsql_changes` table or +view; for real cr-sqlite databases, loading/configuring cr-sqlite remains the +database owner's responsibility. Importing or pinging a peer card never grants capabilities by itself. When `[iroh].local_discovery = true`, the daemon also advertises and discovers signed peer cards on LAN using a geth-specific mDNS TXT payload. That payload is diff --git a/crates/geth-cli/src/lib.rs b/crates/geth-cli/src/lib.rs index 7fbffa2..ffab433 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -1413,6 +1413,7 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { endpoint_id, name, changes_received, + changes_applied, max_db_version, schema_match, allowed, @@ -1427,6 +1428,7 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { println!("agent: {peer_agent_id}"); println!("endpoint: {endpoint_id}"); println!("changes_received: {changes_received}"); + println!("changes_applied: {changes_applied}"); println!( "max_db_version: {}", max_db_version diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 539e50e..df9f17e 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -461,6 +461,7 @@ pub enum ControlResponse { endpoint_id: String, name: String, changes_received: usize, + changes_applied: usize, max_db_version: Option, schema_match: bool, allowed: bool, @@ -1228,6 +1229,7 @@ mod tests { endpoint_id: "endpoint:peer".to_owned(), name: "notes".to_owned(), changes_received: 1, + changes_applied: 1, max_db_version: Some(7), schema_match: true, allowed: true, diff --git a/crates/geth-db/src/lib.rs b/crates/geth-db/src/lib.rs index bf24d66..46780ba 100644 --- a/crates/geth-db/src/lib.rs +++ b/crates/geth-db/src/lib.rs @@ -1,6 +1,6 @@ use geth_types::{DbId, ResourceId}; use rusqlite::types::Value; -use rusqlite::{Connection, OpenFlags}; +use rusqlite::{Connection, OpenFlags, params_from_iter}; use serde::{Deserialize, Serialize}; use std::path::Path; @@ -234,6 +234,78 @@ pub fn extract_crsqlite_changes( }) } +pub fn apply_crsqlite_changes(path: &Path, batch: &CrSqliteChangeBatch) -> Result { + if batch.changes.is_empty() { + return Ok(0); + } + let conn = Connection::open(path)?; + 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 has_site_id = columns.iter().any(|column| column == "site_id"); + let has_causal_length = columns.iter().any(|column| column == "cl"); + let has_sequence = columns.iter().any(|column| column == "seq"); + let mut insert_columns = vec![ + "table_name", + "pk", + "cid", + "val", + "col_version", + "db_version", + ]; + if has_site_id { + insert_columns.push("site_id"); + } + if has_causal_length { + insert_columns.push("cl"); + } + if has_sequence { + insert_columns.push("seq"); + } + let placeholders = (1..=insert_columns.len()) + .map(|index| format!("?{index}")) + .collect::>() + .join(", "); + let sql = format!( + "INSERT INTO crsql_changes({}) VALUES ({placeholders})", + insert_columns.join(", ") + ); + + let tx = conn.unchecked_transaction()?; + let mut applied = 0; + { + let mut stmt = tx.prepare(&sql)?; + for change in &batch.changes { + let mut values = vec![ + Value::Text(change.table_name.clone()), + sqlite_value_to_rusqlite(&change.pk), + Value::Text(change.column_id.clone()), + sqlite_value_to_rusqlite(&change.value), + Value::Integer(change.column_version), + Value::Integer(change.db_version), + ]; + if has_site_id { + values.push(change.site_id.clone().map_or(Value::Null, Value::Blob)); + } + if has_causal_length { + values.push(change.causal_length.map_or(Value::Null, Value::Integer)); + } + if has_sequence { + values.push(change.sequence.map_or(Value::Null, Value::Integer)); + } + stmt.execute(params_from_iter(values.iter()))?; + applied += 1; + } + } + tx.commit()?; + Ok(applied) +} + fn crsqlite_change_columns(conn: &Connection) -> Result, DbError> { let available: bool = conn.query_row( r#"SELECT EXISTS( @@ -280,6 +352,16 @@ fn sqlite_value(value: Value) -> SqliteValue { } } +fn sqlite_value_to_rusqlite(value: &SqliteValue) -> Value { + match value { + SqliteValue::Null => Value::Null, + SqliteValue::Integer(value) => Value::Integer(*value), + SqliteValue::Real(value) => Value::Real(*value), + SqliteValue::Text(value) => Value::Text(value.clone()), + SqliteValue::Blob(value) => Value::Blob(value.clone()), + } +} + #[must_use] pub fn crsqlite_sync_roadmap() -> &'static str { "future db sync reads crsql_changes, exchanges changes over Iroh, and applies through crsql_changes" @@ -439,4 +521,48 @@ mod tests { Err(DbError::MissingCrSqliteChanges) )); } + + #[test] + fn apply_crsqlite_changes_inserts_typed_batch() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("notes.sqlite"); + let conn = Connection::open(&path).expect("open sqlite"); + 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"); + drop(conn); + let batch = CrSqliteChangeBatch { + schema_metadata: "test-schema".to_owned(), + max_db_version: Some(9), + changes: vec![CrSqliteChange { + table_name: "notes".to_owned(), + pk: SqliteValue::Blob(vec![1]), + column_id: "body".to_owned(), + value: SqliteValue::Text("hello".to_owned()), + column_version: 2, + db_version: 9, + site_id: Some(vec![7]), + causal_length: Some(8), + sequence: Some(9), + }], + }; + + assert_eq!(apply_crsqlite_changes(&path, &batch).expect("apply"), 1); + + let extracted = extract_crsqlite_changes(&path, None, 10).expect("extract"); + assert_eq!(extracted.changes.len(), 1); + assert_eq!(extracted.changes[0], batch.changes[0]); + } } diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 29d2a81..0475c1f 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -1386,6 +1386,7 @@ async fn db_sync_from_peer( endpoint_id, name: response_name, changes_received: 0, + changes_applied: 0, max_db_version: None, schema_match: false, allowed, @@ -1394,25 +1395,37 @@ async fn db_sync_from_peer( }); } let local_schema = geth_db::schema_metadata(Path::new(&local.path))?; - let (changes_received, max_db_version, schema_match) = if let Some(batch) = batch { - let schema_match = batch.schema_metadata == local_schema; - let changes_received = batch.changes.len(); - let max_db_version = batch.max_db_version; - if schema_match { - if let Some(next_cursor) = high_water_db_version.or(max_db_version) { - store_live_sync_cursor(&store, peer_node, &stream, next_cursor)?; + let (changes_received, changes_applied, max_db_version, schema_match) = + if let Some(batch) = batch { + let schema_match = batch.schema_metadata == local_schema; + let changes_received = batch.changes.len(); + let max_db_version = batch.max_db_version; + let changes_applied = if schema_match { + geth_db::apply_crsqlite_changes(Path::new(&local.path), &batch)? + } else { + 0 + }; + if schema_match { + if let Some(next_cursor) = high_water_db_version.or(max_db_version) { + store_live_sync_cursor(&store, peer_node, &stream, next_cursor)?; + } } - } - (changes_received, max_db_version, schema_match) - } else { - (0, high_water_db_version, false) - }; + ( + changes_received, + changes_applied, + max_db_version, + schema_match, + ) + } else { + (0, 0, high_water_db_version, false) + }; Ok(ControlResponse::DbSynced { peer_node_id: node_id, peer_agent_id: agent_id, endpoint_id, name: response_name, changes_received, + changes_applied, max_db_version, schema_match, allowed, @@ -2445,7 +2458,7 @@ async fn handle_iroh_control_connection( reason: explanation.reason, evaluated_ops: explanation.evaluated_ops, nonce, - note: "DB sync authenticated endpoint/card binding and required db.sync on the remote DB resource; bootstrap exchanges typed crsql_changes and cursors them, but applying remote changes is not implemented yet".to_owned(), + note: "DB sync authenticated endpoint/card binding and required db.sync on the remote DB resource; bootstrap exchanges typed crsql_changes and applies compatible batches through local crsql_changes".to_owned(), }, Err(message) => PeerControlResponse::Error { message }, } @@ -4953,12 +4966,14 @@ mod tests { ControlResponse::DbSynced { allowed, changes_received, + changes_applied, schema_match, reason, .. } => { assert!(!allowed); assert_eq!(changes_received, 0); + assert_eq!(changes_applied, 0); assert!(!schema_match); assert!(reason.contains("no active direct or group grant")); } @@ -5322,6 +5337,7 @@ mod tests { ControlResponse::DbSynced { allowed, changes_received, + changes_applied, max_db_version, schema_match, reason, @@ -5330,14 +5346,19 @@ mod tests { } => { assert!(allowed); assert_eq!(changes_received, 1); + assert_eq!(changes_applied, 1); assert_eq!(max_db_version, Some(7)); assert!(schema_match); assert!(reason.contains("direct grant")); assert!(note.contains("db.sync")); - assert!(note.contains("applying remote changes is not implemented yet")); + assert!(note.contains("applies compatible batches")); } other => panic!("unexpected allowed DB sync response: {other:?}"), } + let left_changes = geth_db::extract_crsqlite_changes(&left_db_path, None, 10) + .expect("left db changes after apply"); + assert_eq!(left_changes.changes.len(), 1); + assert_eq!(left_changes.max_db_version, Some(7)); let denied_cert_sync = handle_request_async( &left, diff --git a/docs/architecture.md b/docs/architecture.md index aaf06ce..d2f4a2b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -151,10 +151,12 @@ extract read-only typed change batches from `crsql_changes` with schema metadata through `db changes`. As a staged network path, `geth db sync ` uses the protected Iroh control ALPN to request remote typed change batches when the caller has `db.sync` on the remote `resource:db:`. The requester -checks remote schema metadata against its local DB before advancing its -per-peer/per-DB cursor. Loading cr-sqlite and applying remote changes through -`crsql_changes` are future work; the current path exchanges and cursors changes -but does not mutate the local application database. +checks remote schema metadata against its local DB before applying changes and +advancing its per-peer/per-DB cursor. Compatible remote batches are inserted +into the local `crsql_changes` table or view before the cursor advances. +Loading/configuring the cr-sqlite extension for real application databases +remains the database owner's responsibility; the bootstrap tests use +deterministic fixture tables. `geth-kv` currently provides a SQLite-backed local fallback for named KV stores through `kv create/set/get`. `kv set --subject ` evaluates local auth diff --git a/docs/roadmap.md b/docs/roadmap.md index fa4aa9a..7ca3d9c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -438,7 +438,11 @@ Automerge documents. - `[x]` Background live-sync runs DB sync for local DB resources and known peers. - `[x]` Per-peer/per-DB high-water cursors are stored in `module_state`. - - `[ ]` Apply compatible remote changes through cr-sqlite. + - `[x]` Extracted batches are exposed through `geth db changes`. + - `[x]` Compatible remote batches are inserted into the local + `crsql_changes` table or view before advancing the cursor. + - `[x]` Tests cover typed batch application into deterministic fixture + `crsql_changes` tables. - `[ ]` Add an integration test with a real cr-sqlite-enabled SQLite DB that proves two local test nodes exchange and apply changes. - `[ ]` Optional CAS-backed snapshots or batches are documented if used.