geth/crates/geth-db/src/lib.rs

use geth_types::{DbId, ResourceId};
use rusqlite::types::Value;
use rusqlite::{Connection, OpenFlags};
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,
    })
}

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),
    }
}

#[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)
        ));
    }
}