Apply compatible DB sync batches

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<usize, DbError> {
+    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::<Vec<_>>()
+        .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<Vec<String>, 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]);
+    }
 }