diff --git a/AGENTS.md b/AGENTS.md index a4b71ff..f7ddfe9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -136,8 +136,11 @@ Roadmap items should be actionable and checkable: - 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. The DB crate and daemon can extract typed read-only `crsql_changes` - batches through `geth db changes` for future sync messages. cr-sqlite loading, - applying remote changes, and sync are still roadmap work. + 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. - 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 @@ -170,6 +173,6 @@ Roadmap items should be actionable and checkable: `ssh_revocation.sync` on `resource:ssh:revocations`. The daemon live-syncs known peers every 30 seconds using per-peer cursors; this is pull-only metadata sync, not yet a CRDT/resource-log replication model. -- cr-sqlite, iroh-docs, iroh-blobs provider/fetch, Automerge sync, broader auth - enforcement, and Keyhive/BeeKEM-style authorization are future roadmap items - unless implemented later. +- cr-sqlite apply, iroh-docs, iroh-blobs provider/fetch, Automerge sync, + broader auth enforcement, and Keyhive/BeeKEM-style authorization are future + roadmap items unless implemented later. diff --git a/Cargo.lock b/Cargo.lock index 9b57c3d..88ae890 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1249,6 +1249,7 @@ dependencies = [ "geth-store", "geth-types", "iroh", + "rusqlite", "serde", "serde_json", "swarm-discovery", diff --git a/README.md b/README.md index b5be07d..7c9f7fe 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,8 @@ The bootstrap implementation provides: - local DB resource registration: `geth db add ` and `geth db status ` with schema and `crsql_changes` metadata; the DB crate and daemon can extract typed local `crsql_changes` batches through - `geth db changes ` for future sync + `geth db changes ` and exchange authorized remote batches with + `geth db sync ` - local SQLite-backed KV commands: `geth kv create/set/get`; `kv set` accepts `--subject ` to exercise local capability checks for non-local callers; `geth kv sync ` pulls authorized remote updates @@ -161,6 +162,13 @@ Document sync is a bootstrap JSON last-writer-wins path before Automerge: manual `geth document sync ` and background live-sync require `document.read` on `resource:document:` and import only state that is not 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. 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 2982698..5a15601 100644 --- a/crates/geth-cli/src/lib.rs +++ b/crates/geth-cli/src/lib.rs @@ -351,6 +351,12 @@ pub enum DbCommand { #[arg(long, default_value_t = 100)] limit: u32, }, + Sync { + node: String, + name: String, + #[arg(long, default_value_t = 100)] + limit: u32, + }, } #[derive(Debug, Subcommand)] @@ -665,6 +671,7 @@ fn request_for_command(command: Command) -> Result { after_db_version, limit, }, + DbCommand::Sync { node, name, limit } => ControlRequest::DbSync { node, name, limit }, }, Command::Document { command } => match command { DocumentCommand::Create { name } => ControlRequest::DocumentCreate { name }, @@ -1321,6 +1328,37 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> { ); } } + ControlResponse::DbSynced { + peer_node_id, + peer_agent_id, + endpoint_id, + name, + changes_received, + max_db_version, + schema_match, + allowed, + reason, + note, + } => { + if allowed { + println!("synced db changes for {name} from {peer_node_id}"); + } else { + println!("db sync denied by {peer_node_id}"); + } + println!("agent: {peer_agent_id}"); + println!("endpoint: {endpoint_id}"); + println!("changes_received: {changes_received}"); + println!( + "max_db_version: {}", + max_db_version + .map(|version| version.to_string()) + .unwrap_or_else(|| "none".to_owned()) + ); + println!("schema_match: {schema_match}"); + println!("allowed: {allowed}"); + println!("reason: {reason}"); + println!("note: {note}"); + } ControlResponse::KvCreated { kv } => { println!("created kv: {}", kv.name); println!("id: {}", kv.id); diff --git a/crates/geth-control/src/lib.rs b/crates/geth-control/src/lib.rs index 22609e2..daf7c6f 100644 --- a/crates/geth-control/src/lib.rs +++ b/crates/geth-control/src/lib.rs @@ -183,6 +183,11 @@ pub enum ControlRequest { after_db_version: Option, limit: u32, }, + DbSync { + node: String, + name: String, + limit: u32, + }, KvCreate { name: String, }, @@ -423,6 +428,18 @@ pub enum ControlResponse { db: DbResource, batch: CrSqliteChangeBatch, }, + DbSynced { + peer_node_id: String, + peer_agent_id: String, + endpoint_id: String, + name: String, + changes_received: usize, + max_db_version: Option, + schema_match: bool, + allowed: bool, + reason: String, + note: String, + }, KvCreated { kv: KvResource, }, @@ -592,6 +609,13 @@ pub enum PeerControlRequest { since_ms: i64, nonce: String, }, + DbSync { + peer_card: PeerCard, + name: String, + after_db_version: Option, + limit: u32, + nonce: String, + }, } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] @@ -712,6 +736,20 @@ pub enum PeerControlResponse { nonce: String, note: String, }, + DbSynced { + node_id: String, + agent_id: String, + endpoint_id: String, + remote_endpoint_id: String, + name: String, + batch: Option, + high_water_db_version: Option, + allowed: bool, + reason: String, + evaluated_ops: usize, + nonce: String, + note: String, + }, Error { message: String, }, @@ -977,6 +1015,33 @@ mod tests { request ); + let request = ControlRequest::DbSync { + node: "node:peer".to_owned(), + name: "notes".to_owned(), + limit: 50, + }; + assert_eq!( + decode_request(&encode_request(&request).expect("encode")).expect("decode"), + request + ); + + let response = ControlResponse::DbSynced { + peer_node_id: "node:peer".to_owned(), + peer_agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + name: "notes".to_owned(), + changes_received: 1, + max_db_version: Some(7), + schema_match: true, + allowed: true, + reason: "direct grant".to_owned(), + note: "exchange only".to_owned(), + }; + assert_eq!( + decode_response(&encode_response(&response).expect("encode")).expect("decode"), + response + ); + let request = ControlRequest::CasRootScan { name: "notes".to_owned(), }; @@ -1073,6 +1138,29 @@ mod tests { response ); + let request = PeerControlRequest::DbSync { + peer_card: PeerCard { + node_id: "node:caller".into(), + agent_id: "agent:caller".into(), + endpoints: Vec::new(), + issued_at: geth_types::UnixMillis(1), + signature: geth_discovery::SignatureMetadata { + namespace: "geth.peer-card.v1@geth.local".to_owned(), + signer: "agent:caller".to_owned(), + public_key: "key".to_owned(), + signature: "sig".to_owned(), + }, + }, + name: "notes".to_owned(), + after_db_version: Some(7), + limit: 10, + nonce: "nonce".to_owned(), + }; + assert_eq!( + decode_peer_request(&encode_peer_request(&request).expect("encode")).expect("decode"), + request + ); + let response = PeerControlResponse::SshCertSynced { node_id: "node:peer".to_owned(), agent_id: "agent:peer".to_owned(), @@ -1180,5 +1268,29 @@ mod tests { .expect("decode"), response ); + + let response = PeerControlResponse::DbSynced { + node_id: "node:peer".to_owned(), + agent_id: "agent:peer".to_owned(), + endpoint_id: "endpoint:peer".to_owned(), + remote_endpoint_id: "endpoint:caller".to_owned(), + name: "notes".to_owned(), + batch: Some(CrSqliteChangeBatch { + schema_metadata: "tables=1 schema_hash=abc".to_owned(), + max_db_version: Some(7), + changes: Vec::new(), + }), + high_water_db_version: Some(7), + allowed: true, + reason: "direct grant".to_owned(), + evaluated_ops: 1, + nonce: "nonce".to_owned(), + note: "db sync".to_owned(), + }; + assert_eq!( + decode_peer_response(&encode_peer_response(&response).expect("encode")) + .expect("decode"), + response + ); } } diff --git a/crates/geth-node/Cargo.toml b/crates/geth-node/Cargo.toml index f399055..9022749 100644 --- a/crates/geth-node/Cargo.toml +++ b/crates/geth-node/Cargo.toml @@ -34,4 +34,5 @@ iroh.workspace = true swarm-discovery.workspace = true [dev-dependencies] +rusqlite.workspace = true tempfile.workspace = true diff --git a/crates/geth-node/src/lib.rs b/crates/geth-node/src/lib.rs index 7cbb6c4..7f64058 100644 --- a/crates/geth-node/src/lib.rs +++ b/crates/geth-node/src/lib.rs @@ -263,6 +263,11 @@ pub async fn handle_request_async( node: peer_node, name, } => kv_sync_from_peer(node, &peer_node, &name).await, + ControlRequest::DbSync { + node: peer_node, + name, + limit, + } => db_sync_from_peer(node, &peer_node, &name, limit).await, ControlRequest::PubsubPub { topic, message, @@ -555,7 +560,8 @@ async fn peer_ping(node: &LocalNode, peer_node: &str) -> Result Err(NodeError::IrohPeer( + | PeerControlResponse::DocumentSynced { .. } + | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to ping request".to_owned(), )), } @@ -666,7 +672,8 @@ async fn peer_auth_check( | PeerControlResponse::KvSynced { .. } | PeerControlResponse::PubsubPublished { .. } | PeerControlResponse::PipeConnected { .. } - | PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer( + | PeerControlResponse::DocumentSynced { .. } + | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to auth-check request".to_owned(), )), } @@ -806,7 +813,8 @@ async fn cas_fetch_from_peer( | PeerControlResponse::KvSynced { .. } | PeerControlResponse::PubsubPublished { .. } | PeerControlResponse::PipeConnected { .. } - | PeerControlResponse::DocumentSynced { .. } => Err(NodeError::IrohPeer( + | PeerControlResponse::DocumentSynced { .. } + | PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( "peer returned wrong response type to CAS fetch".to_owned(), )), } @@ -1212,6 +1220,94 @@ async fn document_sync_from_peer( } } +async fn db_sync_from_peer( + node: &LocalNode, + peer_node: &str, + name: &str, + limit: u32, +) -> Result { + geth_db::validate_db_name(name).map_err(|_| NodeError::InvalidDbName(name.to_owned()))?; + let store = Store::open(&node.paths.metadata_db())?; + let local = store + .get_db_resource_by_name(name)? + .ok_or_else(|| NodeError::DbNotFound(name.to_owned()))?; + let stream = format!("db:{name}"); + let cursor = load_live_sync_cursor(&store, peer_node, &stream)?; + let after_db_version = (cursor > 0).then_some(cursor); + let response = request_peer_control(node, peer_node, "db-sync", |peer_card, nonce| { + PeerControlRequest::DbSync { + peer_card, + name: name.to_owned(), + after_db_version, + limit, + nonce, + } + }) + .await?; + match response { + PeerControlResponse::DbSynced { + node_id, + agent_id, + endpoint_id, + name: response_name, + batch, + high_water_db_version, + allowed, + reason, + note, + .. + } if response_name == name => { + if !allowed { + return Ok(ControlResponse::DbSynced { + peer_node_id: node_id, + peer_agent_id: agent_id, + endpoint_id, + name: response_name, + changes_received: 0, + max_db_version: None, + schema_match: false, + allowed, + reason, + note, + }); + } + 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)?; + } + } + (changes_received, max_db_version, schema_match) + } else { + (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, + max_db_version, + schema_match, + allowed, + reason, + note, + }) + } + PeerControlResponse::DbSynced { .. } => Err(NodeError::IrohPeer( + "peer DB sync response did not match request".to_owned(), + )), + PeerControlResponse::Error { message } => Err(NodeError::IrohPeer(message)), + _ => Err(NodeError::IrohPeer( + "peer returned wrong response type to DB sync".to_owned(), + )), + } +} + fn live_sync_cursor_key(peer_node: &str, stream: &str) -> String { format!("live-sync:{peer_node}:{stream}") } @@ -1320,6 +1416,10 @@ async fn request_peer_control( | PeerControlResponse::DocumentSynced { nonce: response_nonce, .. + } + | PeerControlResponse::DbSynced { + nonce: response_nonce, + .. } if response_nonce == &nonce => Ok(response), PeerControlResponse::Error { .. } => Ok(response), _ => Err(NodeError::IrohPeer(format!( @@ -1369,6 +1469,7 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> { let peers = Store::open(&node.paths.metadata_db())?.list_peer_cards()?; let kv_stores = Store::open(&node.paths.metadata_db())?.list_kv_stores()?; let documents = Store::open(&node.paths.metadata_db())?.list_document_resources()?; + let dbs = Store::open(&node.paths.metadata_db())?.list_db_resources()?; for peer in peers { if let Err(error) = ssh_cert_sync_from_peer(node, &peer.peer_id).await { tracing::debug!(peer = %peer.peer_id, %error, "SSH cert live sync failed"); @@ -1386,6 +1487,11 @@ async fn run_live_sync_once(node: &LocalNode) -> Result<(), NodeError> { tracing::debug!(peer = %peer.peer_id, document = %document.name, %error, "document live sync failed"); } } + for db in &dbs { + if let Err(error) = db_sync_from_peer(node, &peer.peer_id, &db.name, 100).await { + tracing::debug!(peer = %peer.peer_id, db = %db.name, %error, "DB live sync failed"); + } + } } Ok(()) } @@ -1868,6 +1974,78 @@ async fn handle_iroh_control_connection( } } } + PeerControlRequest::DbSync { + peer_card, + name, + after_db_version, + limit, + nonce, + } => { + geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?; + peer_card.validate_candidate()?; + ensure_peer_card_matches_endpoint(&peer_card, &remote_endpoint_id)?; + let discovered = DiscoveredPeer::candidate( + peer_card.clone(), + UnixMillis(geth_store::now_ms()), + DiscoverySource::PeerExchange, + )?; + let store = Store::open(&node.paths.metadata_db())?; + store.upsert_peer_card(&StoredPeerCard { + peer_id: peer_card.node_id.to_string(), + card_json: serde_json::to_string(&peer_card)?, + updated_at_ms: discovered.discovered_at.0, + })?; + if let Some(db) = store.get_db_resource_by_name(&name)? { + let capability = "db.sync".to_owned(); + let explanation = geth_auth::explain_auth_ops( + &load_auth_ops_for_resource(&store, &db.resource_id)?, + PrincipalId::new(peer_card.node_id.to_string()), + ResourceId::new(db.resource_id.clone()), + Capability::new(capability), + ); + let sync_result: Result<_, String> = if explanation.allowed { + let path = Path::new(&db.path); + match geth_db::extract_crsqlite_changes(path, after_db_version, limit) { + Ok(batch) => match geth_db::crsqlite_change_metadata(path) { + Ok(metadata) => { + let high_water_db_version = + metadata.max_db_version.or(batch.max_db_version); + Ok((Some(batch), high_water_db_version)) + } + Err(error) => Err(format!( + "db sync cannot read crsql_changes metadata for {name}: {error}" + )), + }, + Err(error) => Err(format!( + "db sync cannot read crsql_changes for {name}: {error}" + )), + } + } else { + Ok((None, None)) + }; + match sync_result { + Ok((batch, high_water_db_version)) => PeerControlResponse::DbSynced { + node_id: node.node_id.clone(), + agent_id: node.agent_id.clone(), + endpoint_id: node.iroh_status.endpoint_id.clone().unwrap_or_default(), + remote_endpoint_id, + name, + batch, + high_water_db_version, + allowed: explanation.allowed, + 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(), + }, + Err(message) => PeerControlResponse::Error { message }, + } + } else { + PeerControlResponse::Error { + message: format!("db resource not found: {name}"), + } + } + } }; send.write_all(geth_control::encode_peer_response(&response)?.as_bytes()) .await @@ -2007,6 +2185,7 @@ pub fn handle_request( ControlRequest::SshCertSync { .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::SshRevocationSync { .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::KvSync { .. } => Err(NodeError::IrohEndpointUnavailable), + ControlRequest::DbSync { .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::DocumentSync { .. } => Err(NodeError::IrohEndpointUnavailable), ControlRequest::ResourceList => Ok(ControlResponse::ResourceList { resources: store @@ -3471,6 +3650,63 @@ mod tests { .expect("write config"); } + fn create_mock_crsqlite_db(path: &Path, change_db_version: Option) { + let conn = rusqlite::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"); + if let Some(db_version) = change_db_version { + 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![db_version as u8], + "body", + Vec::from("hello".as_bytes()), + 1_i64, + db_version, + vec![1_u8], + 1_i64, + db_version, + ), + ) + .expect("insert initial mock crsqlite change"); + } + } + + fn insert_mock_crsqlite_change(path: &Path, db_version: i64, body: &str) { + let conn = rusqlite::Connection::open(path).expect("open sqlite"); + 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![db_version as u8], + "body", + body.as_bytes().to_vec(), + 1_i64, + db_version, + vec![1_u8], + 1_i64, + db_version, + ), + ) + .expect("insert mock crsqlite change"); + } + #[test] fn lan_discovery_address_selection_uses_iroh_direct_addresses() { let key = AgentKey::generate(); @@ -3618,6 +3854,26 @@ mod tests { }, ) .expect("right document set"); + let left_db_path = left_home.path().join("notes.sqlite"); + let right_db_path = right_home.path().join("notes.sqlite"); + create_mock_crsqlite_db(&left_db_path, None); + create_mock_crsqlite_db(&right_db_path, Some(7)); + handle_request( + &left, + ControlRequest::DbAdd { + name: "notes".to_owned(), + path: left_db_path.clone(), + }, + ) + .expect("left db add"); + handle_request( + &right, + ControlRequest::DbAdd { + name: "notes".to_owned(), + path: right_db_path.clone(), + }, + ) + .expect("right db add"); let ping = handle_request_async( &left, @@ -3789,6 +4045,32 @@ mod tests { other => panic!("unexpected denied document sync response: {other:?}"), } + let denied_db_sync = handle_request_async( + &left, + ControlRequest::DbSync { + node: right_card.node_id.to_string(), + name: "notes".to_owned(), + limit: 10, + }, + ) + .await + .expect("denied db sync"); + match denied_db_sync { + ControlResponse::DbSynced { + allowed, + changes_received, + schema_match, + reason, + .. + } => { + assert!(!allowed); + assert_eq!(changes_received, 0); + assert!(!schema_match); + assert!(reason.contains("no active direct or group grant")); + } + other => panic!("unexpected denied DB sync response: {other:?}"), + } + handle_request( &right, ControlRequest::AuthGrant { @@ -3839,6 +4121,16 @@ mod tests { }, ) .expect("grant left document read"); + handle_request( + &right, + ControlRequest::AuthGrant { + subject: left.node_id.clone(), + resource: "resource:db:notes".to_owned(), + capability: "db.sync".to_owned(), + grant_id: Some("grant:left-db-sync".to_owned()), + }, + ) + .expect("grant left db sync"); let allowed = handle_request_async( &left, @@ -4040,6 +4332,37 @@ mod tests { other => panic!("unexpected synced document get response: {other:?}"), } + let db_sync = handle_request_async( + &left, + ControlRequest::DbSync { + node: right_card.node_id.to_string(), + name: "notes".to_owned(), + limit: 10, + }, + ) + .await + .expect("allowed db sync"); + match db_sync { + ControlResponse::DbSynced { + allowed, + changes_received, + max_db_version, + schema_match, + reason, + note, + .. + } => { + assert!(allowed); + assert_eq!(changes_received, 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")); + } + other => panic!("unexpected allowed DB sync response: {other:?}"), + } + let denied_cert_sync = handle_request_async( &left, ControlRequest::SshCertSync { @@ -4207,6 +4530,7 @@ mod tests { }, ) .expect("right live document set"); + insert_mock_crsqlite_change(&right_db_path, 8, "live"); run_live_sync_once(&left) .await @@ -4240,6 +4564,19 @@ mod tests { .map(|document| document.state_json), Some(r#"{"title":"live"}"#.to_owned()) ); + let db_cursor = left_store + .get_module_state(&live_sync_cursor_key( + right_card.node_id.as_str(), + "db:notes", + )) + .expect("get live-synced db cursor") + .expect("db cursor exists"); + assert_eq!( + serde_json::from_str::(&db_cursor.state_json) + .expect("parse db cursor") + .cursor_ms, + 8 + ); left_endpoint.shutdown().await; right_endpoint.shutdown().await; diff --git a/crates/geth-store/src/lib.rs b/crates/geth-store/src/lib.rs index ab3f1c9..5ffd354 100644 --- a/crates/geth-store/src/lib.rs +++ b/crates/geth-store/src/lib.rs @@ -309,6 +309,26 @@ impl Store { } } + pub fn list_db_resources(&self) -> Result, StoreError> { + let mut stmt = self.conn.prepare( + r#"SELECT db.db_id, db.resource_id, resources.name, db.path + FROM db_resources db + JOIN resources ON resources.resource_id = db.resource_id + WHERE resources.kind = 'db' + ORDER BY resources.name"#, + )?; + let rows = stmt.query_map([], |row| { + Ok(StoredDbResource { + db_id: row.get(0)?, + resource_id: row.get(1)?, + name: row.get(2)?, + path: row.get(3)?, + }) + })?; + rows.collect::, _>>() + .map_err(StoreError::from) + } + pub fn insert_kv_store(&self, kv: &StoredKvStore) -> Result<(), StoreError> { self.conn.execute( r#"INSERT OR REPLACE INTO kv_stores(kv_id, resource_id, name) @@ -1479,6 +1499,15 @@ mod tests { .map(|db| db.path), Some("/tmp/notes.sqlite".to_owned()) ); + assert_eq!( + store + .list_db_resources() + .expect("list db resources") + .into_iter() + .map(|db| db.name) + .collect::>(), + vec!["notes".to_owned()] + ); } #[test] diff --git a/docs/architecture.md b/docs/architecture.md index c8b81d6..f781c01 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -134,8 +134,13 @@ 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`. The crate and local daemon can extract read-only typed change batches from `crsql_changes` with schema metadata -through `db changes`. Loading cr-sqlite, applying remote changes, and DB sync -are future work. +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. `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 4db9464..8fb7dfd 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -376,11 +376,25 @@ Automerge documents. - `[x]` Schema hash/version metadata is included in extracted change batches. - `[x]` Extracted batches are exposed through `geth db changes`. -- `[ ]` DB sync over Iroh. +- `[~]` DB sync over Iroh. Acceptance criteria: - - Two local test nodes can exchange and apply DB changes. - - Schema mismatch is detected before applying changes. - - Optional CAS-backed snapshots or batches are documented if used. + - `[x]` `geth db sync ` exists and talks to the daemon. + - `[x]` Remote DB sync uses the protected Iroh control ALPN. + - `[x]` The serving peer validates the caller's signed peer card against the + observed Iroh EndpointID before considering authorization. + - `[x]` Remote DB sync requires `db.sync` on the remote + `resource:db:`. + - `[x]` The response carries typed `crsql_changes` batches plus schema + metadata. + - `[x]` The requester detects schema mismatch before advancing the sync + cursor. + - `[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. + - `[ ]` 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. - `[~]` Automerge document resource. Acceptance criteria: