use geth_types::{DbId, ResourceId}; use rusqlite::types::Value; use rusqlite::{Connection, OpenFlags, params_from_iter}; use serde::{Deserialize, Serialize}; use std::path::Path; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct DbResource { pub id: DbId, pub resource: ResourceId, pub name: String, pub path: String, pub path_exists: bool, pub size_bytes: Option, pub schema_metadata: String, pub crsqlite_changes: CrSqliteChangeMetadata, pub sync_status: String, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct CrSqliteChangeMetadata { pub available: bool, pub change_count: Option, pub max_db_version: Option, pub columns: Vec, 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 { Self { available: false, change_count: None, max_db_version: None, columns: Vec::new(), error: None, } } #[must_use] pub fn error(error: impl Into) -> Self { Self { available: false, change_count: None, max_db_version: None, columns: Vec::new(), error: Some(error.into()), } } } #[derive(Debug, thiserror::Error)] pub enum DbError { #[error("invalid db resource name: {0}")] 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> { if name.is_empty() || !name .bytes() .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) { return Err(DbError::InvalidName(name.to_owned())); } Ok(()) } pub fn schema_metadata(path: &Path) -> Result { let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?; let mut stmt = conn.prepare( r#"SELECT type, name, COALESCE(sql, '') FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' ORDER BY type, name"#, )?; let rows = stmt.query_map([], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, )) })?; let mut canonical = String::new(); let mut table_count = 0_usize; let mut index_count = 0_usize; let mut view_count = 0_usize; let mut trigger_count = 0_usize; for row in rows { let (kind, name, sql) = row?; match kind.as_str() { "table" => table_count += 1, "index" => index_count += 1, "view" => view_count += 1, "trigger" => trigger_count += 1, _ => {} } canonical.push_str(&kind); canonical.push('\0'); canonical.push_str(&name); canonical.push('\0'); canonical.push_str(&sql); canonical.push('\n'); } Ok(format!( "tables={table_count} indexes={index_count} views={view_count} triggers={trigger_count} schema_hash={}", blake3::hash(canonical.as_bytes()) )) } pub fn crsqlite_change_metadata(path: &Path) -> Result { let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?; 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 Ok(CrSqliteChangeMetadata::unavailable()); } let mut columns_stmt = conn.prepare("PRAGMA table_info('crsql_changes')")?; let columns = columns_stmt .query_map([], |row| row.get::<_, String>(1))? .collect::, _>>()?; let change_count: u64 = conn.query_row("SELECT COUNT(*) FROM crsql_changes", [], |row| row.get(0))?; let max_db_version = if columns.iter().any(|column| column == "db_version") { conn.query_row("SELECT MAX(db_version) FROM crsql_changes", [], |row| { row.get(0) })? } else { None }; Ok(CrSqliteChangeMetadata { available: true, change_count: Some(change_count), max_db_version, columns, error: None, }) } pub fn extract_crsqlite_changes( path: &Path, after_db_version: Option, 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, }) } 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( 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), } } 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" } #[cfg(test)] mod tests { use super::*; #[test] fn db_name_validation_rejects_paths_and_empty_names() { assert!(validate_db_name("notes").is_ok()); assert!(validate_db_name("notes.v1").is_ok()); assert!(validate_db_name("").is_err()); assert!(validate_db_name("../notes").is_err()); assert!(validate_db_name("notes/main").is_err()); assert!(validate_db_name("notes main").is_err()); } #[test] fn schema_metadata_reports_counts_and_stable_hash() { 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 NOT NULL)", [], ) .expect("create table"); conn.execute("CREATE INDEX notes_body ON notes(body)", []) .expect("create index"); drop(conn); let first = schema_metadata(&path).expect("schema metadata"); let second = schema_metadata(&path).expect("schema metadata again"); assert_eq!(first, second); assert!(first.contains("tables=1")); assert!(first.contains("indexes=1")); assert!(first.contains("schema_hash=")); } #[test] fn crsqlite_change_metadata_reports_mock_change_table() { 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 TEXT 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) VALUES ('notes', '1', 'body', 'hello', 1, 7)", [], ) .expect("insert change"); drop(conn); let metadata = crsqlite_change_metadata(&path).expect("change metadata"); assert!(metadata.available); assert_eq!(metadata.change_count, Some(1)); 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) )); } #[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]); } }