Expose db change batches over control

crates/geth-cli/src/lib.rs

@@ -248,8 +248,20 @@ pub enum PipeCommand {
 
 #[derive(Debug, Subcommand)]
 pub enum DbCommand {
-    Add { name: String, path: PathBuf },
-    Status { name: String },
+    Add {
+        name: String,
+        path: PathBuf,
+    },
+    Status {
+        name: String,
+    },
+    Changes {
+        name: String,
+        #[arg(long)]
+        after_db_version: Option<i64>,
+        #[arg(long, default_value_t = 100)]
+        limit: u32,
+    },
 }
 
 #[derive(Debug, Subcommand)]
@@ -469,6 +481,15 @@ fn request_for_command(command: Command) -> Result<ControlRequest> {
         Command::Db { command } => match command {
             DbCommand::Add { name, path } => ControlRequest::DbAdd { name, path },
             DbCommand::Status { name } => ControlRequest::DbStatus { name },
+            DbCommand::Changes {
+                name,
+                after_db_version,
+                limit,
+            } => ControlRequest::DbChanges {
+                name,
+                after_db_version,
+                limit,
+            },
         },
         Command::Document { command } => match command {
             DocumentCommand::Create { name } => ControlRequest::DocumentCreate { name },
@@ -893,6 +914,24 @@ fn print_response(response: ControlResponse, json: bool) -> Result<()> {
             }
             println!("sync_status: {}", db.sync_status);
         }
+        ControlResponse::DbChanges { db, batch } => {
+            println!("db: {}", db.name);
+            println!("changes: {}", batch.changes.len());
+            println!(
+                "max_db_version: {}",
+                batch
+                    .max_db_version
+                    .map(|version| version.to_string())
+                    .unwrap_or_else(|| "none".to_owned())
+            );
+            println!("schema_metadata: {}", batch.schema_metadata);
+            for change in batch.changes {
+                println!(
+                    "{}\t{}\t{}",
+                    change.db_version, change.table_name, change.column_id
+                );
+            }
+        }
         ControlResponse::KvCreated { kv } => {
             println!("created kv: {}", kv.name);
             println!("id: {}", kv.id);

crates/geth-control/src/lib.rs

@@ -1,5 +1,5 @@
 use geth_auth::{AuthExplanation, AuthOp};
-use geth_db::DbResource;
+use geth_db::{CrSqliteChangeBatch, DbResource};
 use geth_document::{DocumentResource, DocumentState};
 use geth_keychain::KeychainOp;
 use geth_kv::{KvEntry, KvResource};
@@ -121,6 +121,11 @@ pub enum ControlRequest {
     DbStatus {
         name: String,
     },
+    DbChanges {
+        name: String,
+        after_db_version: Option<i64>,
+        limit: u32,
+    },
     KvCreate {
         name: String,
     },
@@ -165,7 +170,7 @@ pub enum ControlRequest {
     },
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "kebab-case")]
 pub enum ControlResponse {
     Status(StatusResponse),
@@ -262,6 +267,10 @@ pub enum ControlResponse {
     DbStatus {
         db: DbResource,
     },
+    DbChanges {
+        db: DbResource,
+        batch: CrSqliteChangeBatch,
+    },
     KvCreated {
         kv: KvResource,
     },
@@ -427,5 +436,15 @@ mod tests {
             decode_request(&encode_request(&request).expect("encode")).expect("decode"),
             request
         );
+
+        let request = ControlRequest::DbChanges {
+            name: "notes".to_owned(),
+            after_db_version: Some(7),
+            limit: 10,
+        };
+        assert_eq!(
+            decode_request(&encode_request(&request).expect("encode")).expect("decode"),
+            request
+        );
     }
 }

crates/geth-node/src/lib.rs

@@ -758,6 +758,25 @@ pub fn handle_request(
                 db: db_resource_from_stored(&stored)?,
             })
         }
+        ControlRequest::DbChanges {
+            name,
+            after_db_version,
+            limit,
+        } => {
+            geth_db::validate_db_name(&name).map_err(|_| NodeError::InvalidDbName(name.clone()))?;
+            let stored = store
+                .get_db_resource_by_name(&name)?
+                .ok_or_else(|| NodeError::DbNotFound(name.clone()))?;
+            let batch = geth_db::extract_crsqlite_changes(
+                Path::new(&stored.path),
+                after_db_version,
+                limit,
+            )?;
+            Ok(ControlResponse::DbChanges {
+                db: db_resource_from_stored(&stored)?,
+                batch,
+            })
+        }
         ControlRequest::KvCreate { name } => {
             geth_kv::validate_kv_name(&name).map_err(|_| NodeError::InvalidKvName(name.clone()))?;
             let resource_id = format!("resource:kv:{name}");

crates/geth/tests/bootstrap.rs

@@ -463,6 +463,44 @@ fn db_add_and_status_register_local_db_metadata() {
         other => panic!("unexpected response: {other:?}"),
     }
 
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::DbChanges {
+            name: "notes".to_owned(),
+            after_db_version: None,
+            limit: 10,
+        },
+    )
+    .expect("db changes");
+    match response {
+        geth_control::ControlResponse::DbChanges { db, batch } => {
+            assert_eq!(db.name, "notes");
+            assert_eq!(batch.changes.len(), 1);
+            assert_eq!(batch.max_db_version, Some(3));
+            assert_eq!(batch.changes[0].table_name, "notes");
+            assert_eq!(batch.changes[0].column_id, "body");
+            assert!(batch.schema_metadata.contains("schema_hash="));
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
+    let response = geth_node::handle_request(
+        &node,
+        geth_control::ControlRequest::DbChanges {
+            name: "notes".to_owned(),
+            after_db_version: Some(3),
+            limit: 10,
+        },
+    )
+    .expect("db changes after latest");
+    match response {
+        geth_control::ControlResponse::DbChanges { batch, .. } => {
+            assert!(batch.changes.is_empty());
+            assert_eq!(batch.max_db_version, None);
+        }
+        other => panic!("unexpected response: {other:?}"),
+    }
+
     assert!(
         geth_node::handle_request(
             &node,