568 lines
18 KiB
Rust
568 lines
18 KiB
Rust
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<u64>,
|
|
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<u64>,
|
|
pub max_db_version: Option<i64>,
|
|
pub columns: Vec<String>,
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
#[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<u8>),
|
|
}
|
|
|
|
#[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<Vec<u8>>,
|
|
pub causal_length: Option<i64>,
|
|
pub sequence: Option<i64>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
|
|
pub struct CrSqliteChangeBatch {
|
|
pub schema_metadata: String,
|
|
pub max_db_version: Option<i64>,
|
|
pub changes: Vec<CrSqliteChange>,
|
|
}
|
|
|
|
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<String>) -> 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<String, DbError> {
|
|
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<CrSqliteChangeMetadata, DbError> {
|
|
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::<Result<Vec<_>, _>>()?;
|
|
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<i64>,
|
|
limit: u32,
|
|
) -> Result<CrSqliteChangeBatch, DbError> {
|
|
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::<Result<Vec<_>, _>>()?;
|
|
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<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(
|
|
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::<Result<Vec<_>, _>>()
|
|
.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]);
|
|
}
|
|
}
|