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

use rusqlite::{Connection, params};
use std::path::Path;

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("sqlite error: {0}")]
    Sqlite(#[from] rusqlite::Error),
    #[error("json error: {0}")]
    Json(#[from] serde_json::Error),
}

pub struct Store {
    conn: Connection,
}

impl Store {
    pub fn open(path: &Path) -> Result<Self, StoreError> {
        let conn = Connection::open(path)?;
        let store = Self { conn };
        store.migrate()?;
        Ok(store)
    }

    pub fn open_memory() -> Result<Self, StoreError> {
        let conn = Connection::open_in_memory()?;
        let store = Self { conn };
        store.migrate()?;
        Ok(store)
    }

    pub fn migrate(&self) -> Result<(), StoreError> {
        self.conn.execute_batch(
            r#"
            PRAGMA foreign_keys = ON;
            CREATE TABLE IF NOT EXISTS meta (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS agents (
                agent_id TEXT PRIMARY KEY,
                public_key TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS nodes (
                node_id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                agent_id TEXT,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS node_endpoints (
                endpoint_id TEXT PRIMARY KEY,
                node_id TEXT NOT NULL,
                agent_id TEXT NOT NULL,
                transport TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS resources (
                resource_id TEXT PRIMARY KEY,
                kind TEXT NOT NULL,
                name TEXT NOT NULL,
                authority_ref TEXT NOT NULL,
                local_role TEXT NOT NULL,
                replication_policy TEXT NOT NULL,
                retention_policy TEXT NOT NULL,
                status TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS keychain_ops (
                op_id TEXT PRIMARY KEY,
                op_json TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS keychain_signatures (
                op_id TEXT NOT NULL,
                signer TEXT NOT NULL,
                signer_public_key TEXT NOT NULL DEFAULT '',
                namespace TEXT NOT NULL,
                signature BLOB NOT NULL,
                created_at_ms INTEGER NOT NULL,
                PRIMARY KEY (op_id, signer, namespace)
            );
            CREATE TABLE IF NOT EXISTS auth_ops (
                op_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                op_json TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS grants (
                grant_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                principal_id TEXT NOT NULL,
                capability TEXT NOT NULL,
                revoked INTEGER NOT NULL DEFAULT 0
            );
            CREATE TABLE IF NOT EXISTS resource_secrets (
                secret_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                epoch INTEGER NOT NULL,
                status TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS cas_objects (
                hash TEXT PRIMARY KEY,
                size_bytes INTEGER NOT NULL,
                path TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS cas_pins (
                hash TEXT PRIMARY KEY,
                pinned_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS cas_providers (
                hash TEXT NOT NULL,
                peer_node_id TEXT NOT NULL,
                endpoint_id TEXT NOT NULL,
                last_seen_ms INTEGER NOT NULL,
                PRIMARY KEY(hash, peer_node_id)
            );
            CREATE TABLE IF NOT EXISTS kv_stores (
                kv_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                name TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS kv_entries (
                kv_id TEXT NOT NULL,
                key TEXT NOT NULL,
                value TEXT NOT NULL,
                updated_at_ms INTEGER NOT NULL,
                PRIMARY KEY(kv_id, key)
            );
            CREATE TABLE IF NOT EXISTS db_resources (
                db_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                path TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS document_resources (
                document_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                name TEXT NOT NULL,
                state_json TEXT NOT NULL DEFAULT '{}',
                updated_at_ms INTEGER NOT NULL DEFAULT 0
            );
            CREATE TABLE IF NOT EXISTS file_roots (
                root_id TEXT PRIMARY KEY,
                resource_id TEXT NOT NULL,
                name TEXT NOT NULL UNIQUE,
                path TEXT NOT NULL,
                latest_tree_hash TEXT,
                latest_tree_json TEXT,
                updated_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS file_conflicts (
                conflict_id TEXT PRIMARY KEY,
                root_name TEXT NOT NULL,
                resource_id TEXT NOT NULL,
                path TEXT NOT NULL,
                kind TEXT NOT NULL,
                status TEXT NOT NULL,
                base_tree_hash TEXT,
                local_tree_hash TEXT,
                remote_tree_hash TEXT,
                detail TEXT NOT NULL,
                resolution TEXT,
                resolution_note TEXT,
                created_at_ms INTEGER NOT NULL,
                resolved_at_ms INTEGER
            );
            CREATE TABLE IF NOT EXISTS peer_cards (
                peer_id TEXT PRIMARY KEY,
                card_json TEXT NOT NULL,
                updated_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS module_state (
                module TEXT PRIMARY KEY,
                state_json TEXT NOT NULL,
                updated_at_ms INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS ssh_cert_requests (
                request_id TEXT PRIMARY KEY,
                requester_node TEXT NOT NULL,
                public_key TEXT NOT NULL,
                public_key_fingerprint TEXT NOT NULL,
                cert_kind TEXT NOT NULL,
                principals_json TEXT NOT NULL,
                requested_validity TEXT,
                renewal_of TEXT,
                reason TEXT,
                status TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL,
                provenance_json TEXT
            );
            CREATE TABLE IF NOT EXISTS ssh_certificates (
                cert_id TEXT PRIMARY KEY,
                request_id TEXT NOT NULL,
                certificate TEXT NOT NULL,
                certificate_fingerprint TEXT NOT NULL,
                imported_at_ms INTEGER NOT NULL,
                provenance_json TEXT
            );
            CREATE TABLE IF NOT EXISTS ssh_revocations (
                revocation_id TEXT PRIMARY KEY,
                kind TEXT NOT NULL,
                target TEXT NOT NULL,
                reason TEXT,
                created_at_ms INTEGER NOT NULL,
                published INTEGER NOT NULL DEFAULT 0,
                provenance_json TEXT
            );
            INSERT OR IGNORE INTO meta(key, value) VALUES ('schema_version', '1');
            "#,
        )?;
        self.add_column_if_missing(
            "keychain_signatures",
            "signer_public_key",
            "TEXT NOT NULL DEFAULT ''",
        )?;
        self.add_column_if_missing("ssh_cert_requests", "provenance_json", "TEXT")?;
        self.add_column_if_missing("ssh_certificates", "provenance_json", "TEXT")?;
        self.add_column_if_missing("ssh_revocations", "provenance_json", "TEXT")?;
        Ok(())
    }

    fn add_column_if_missing(
        &self,
        table: &str,
        column: &str,
        definition: &str,
    ) -> Result<(), StoreError> {
        let mut stmt = self.conn.prepare(&format!("PRAGMA table_info({table})"))?;
        let columns = stmt
            .query_map([], |row| row.get::<_, String>(1))?
            .collect::<Result<Vec<_>, _>>()?;
        if !columns.iter().any(|name| name == column) {
            self.conn.execute(
                &format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
                [],
            )?;
        }
        Ok(())
    }

    pub fn upsert_agent(&self, agent_id: &str, public_key: &str) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR IGNORE INTO agents(agent_id, public_key, created_at_ms) VALUES (?1, ?2, ?3)",
            params![agent_id, public_key, now_ms()],
        )?;
        Ok(())
    }

    pub fn upsert_node(&self, node_id: &str, name: &str, agent_id: &str) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR IGNORE INTO nodes(node_id, name, agent_id, created_at_ms) VALUES (?1, ?2, ?3, ?4)",
            params![node_id, name, agent_id, now_ms()],
        )?;
        Ok(())
    }

    pub fn upsert_node_endpoint(
        &self,
        endpoint_id: &str,
        node_id: &str,
        agent_id: &str,
        transport: &str,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR IGNORE INTO node_endpoints(endpoint_id, node_id, agent_id, transport, created_at_ms) VALUES (?1, ?2, ?3, ?4, ?5)",
            params![endpoint_id, node_id, agent_id, transport, now_ms()],
        )?;
        Ok(())
    }

    pub fn list_resources(&self) -> Result<Vec<StoredResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT resource_id, kind, name, status FROM resources ORDER BY kind, name, resource_id",
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredResource {
                resource_id: row.get(0)?,
                kind: row.get(1)?,
                name: row.get(2)?,
                status: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_resource(&self, resource: &StoredResource) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR IGNORE INTO resources(
                resource_id, kind, name, authority_ref, local_role, replication_policy,
                retention_policy, status, created_at_ms
            ) VALUES (?1, ?2, ?3, 'local', 'owner', 'local-only', 'keep', ?4, ?5)"#,
            params![
                resource.resource_id,
                resource.kind,
                resource.name,
                resource.status,
                now_ms()
            ],
        )?;
        Ok(())
    }

    pub fn get_resource_by_kind_name(
        &self,
        kind: &str,
        name: &str,
    ) -> Result<Option<StoredResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT resource_id, kind, name, status FROM resources WHERE kind = ?1 AND name = ?2",
        )?;
        let mut rows = stmt.query(params![kind, name])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredResource {
                resource_id: row.get(0)?,
                kind: row.get(1)?,
                name: row.get(2)?,
                status: row.get(3)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn insert_db_resource(&self, db: &StoredDbResource) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO db_resources(db_id, resource_id, path)
               VALUES (?1, ?2, ?3)"#,
            params![db.db_id, db.resource_id, db.path],
        )?;
        Ok(())
    }

    pub fn get_db_resource_by_name(
        &self,
        name: &str,
    ) -> Result<Option<StoredDbResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT db.db_id, db.resource_id, resources.name, db.path
               FROM db_resources db
               JOIN resources ON resources.resource_id = db.resource_id
               WHERE resources.kind = 'db' AND resources.name = ?1"#,
        )?;
        let mut rows = stmt.query(params![name])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredDbResource {
                db_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                path: row.get(3)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_db_resources(&self) -> Result<Vec<StoredDbResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT db.db_id, db.resource_id, resources.name, db.path
               FROM db_resources db
               JOIN resources ON resources.resource_id = db.resource_id
               WHERE resources.kind = 'db'
               ORDER BY resources.name"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredDbResource {
                db_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                path: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_kv_store(&self, kv: &StoredKvStore) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO kv_stores(kv_id, resource_id, name)
               VALUES (?1, ?2, ?3)"#,
            params![kv.kv_id, kv.resource_id, kv.name],
        )?;
        Ok(())
    }

    pub fn get_kv_store_by_name(&self, name: &str) -> Result<Option<StoredKvStore>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT kv_id, resource_id, name
               FROM kv_stores WHERE name = ?1"#,
        )?;
        let mut rows = stmt.query(params![name])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredKvStore {
                kv_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_kv_stores(&self) -> Result<Vec<StoredKvStore>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT kv_id, resource_id, name FROM kv_stores ORDER BY name")?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredKvStore {
                kv_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn set_kv_entry(&self, entry: &StoredKvEntry) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO kv_entries(kv_id, key, value, updated_at_ms)
               VALUES (?1, ?2, ?3, ?4)"#,
            params![entry.kv_id, entry.key, entry.value, entry.updated_at_ms],
        )?;
        Ok(())
    }

    pub fn get_kv_entry(
        &self,
        kv_id: &str,
        key: &str,
    ) -> Result<Option<StoredKvEntry>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT kv_id, key, value, updated_at_ms
               FROM kv_entries WHERE kv_id = ?1 AND key = ?2"#,
        )?;
        let mut rows = stmt.query(params![kv_id, key])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredKvEntry {
                kv_id: row.get(0)?,
                key: row.get(1)?,
                value: row.get(2)?,
                updated_at_ms: row.get(3)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_kv_entries_since(
        &self,
        kv_id: &str,
        since_ms: i64,
    ) -> Result<Vec<StoredKvEntry>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT kv_id, key, value, updated_at_ms
               FROM kv_entries
               WHERE kv_id = ?1 AND updated_at_ms >= ?2
               ORDER BY updated_at_ms, key"#,
        )?;
        let rows = stmt.query_map(params![kv_id, since_ms], |row| {
            Ok(StoredKvEntry {
                kv_id: row.get(0)?,
                key: row.get(1)?,
                value: row.get(2)?,
                updated_at_ms: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_document_resource(
        &self,
        document: &StoredDocumentResource,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO document_resources(
                document_id, resource_id, name, state_json, updated_at_ms
            ) VALUES (?1, ?2, ?3, ?4, ?5)"#,
            params![
                document.document_id,
                document.resource_id,
                document.name,
                document.state_json,
                document.updated_at_ms
            ],
        )?;
        Ok(())
    }

    pub fn get_document_resource_by_name(
        &self,
        name: &str,
    ) -> Result<Option<StoredDocumentResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT document_id, resource_id, name, state_json, updated_at_ms
               FROM document_resources WHERE name = ?1"#,
        )?;
        let mut rows = stmt.query(params![name])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredDocumentResource {
                document_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                state_json: row.get(3)?,
                updated_at_ms: row.get(4)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_document_resources(&self) -> Result<Vec<StoredDocumentResource>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT document_id, resource_id, name, state_json, updated_at_ms
               FROM document_resources ORDER BY name"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredDocumentResource {
                document_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                state_json: row.get(3)?,
                updated_at_ms: row.get(4)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn upsert_file_root(&self, root: &StoredFileRoot) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO file_roots(
                root_id, resource_id, name, path, latest_tree_hash, latest_tree_json, updated_at_ms
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
            params![
                root.root_id,
                root.resource_id,
                root.name,
                root.path,
                root.latest_tree_hash,
                root.latest_tree_json,
                root.updated_at_ms
            ],
        )?;
        Ok(())
    }

    pub fn get_file_root_by_name(&self, name: &str) -> Result<Option<StoredFileRoot>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT root_id, resource_id, name, path, latest_tree_hash, latest_tree_json, updated_at_ms
               FROM file_roots WHERE name = ?1"#,
        )?;
        let mut rows = stmt.query(params![name])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredFileRoot {
                root_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                path: row.get(3)?,
                latest_tree_hash: row.get(4)?,
                latest_tree_json: row.get(5)?,
                updated_at_ms: row.get(6)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_file_roots(&self) -> Result<Vec<StoredFileRoot>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT root_id, resource_id, name, path, latest_tree_hash, latest_tree_json, updated_at_ms
               FROM file_roots ORDER BY name"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredFileRoot {
                root_id: row.get(0)?,
                resource_id: row.get(1)?,
                name: row.get(2)?,
                path: row.get(3)?,
                latest_tree_hash: row.get(4)?,
                latest_tree_json: row.get(5)?,
                updated_at_ms: row.get(6)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn upsert_file_conflict(&self, conflict: &StoredFileConflict) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO file_conflicts(
                conflict_id, root_name, resource_id, path, kind, status, base_tree_hash,
                local_tree_hash, remote_tree_hash, detail, resolution, resolution_note,
                created_at_ms, resolved_at_ms
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#,
            params![
                conflict.conflict_id,
                conflict.root_name,
                conflict.resource_id,
                conflict.path,
                conflict.kind,
                conflict.status,
                conflict.base_tree_hash,
                conflict.local_tree_hash,
                conflict.remote_tree_hash,
                conflict.detail,
                conflict.resolution,
                conflict.resolution_note,
                conflict.created_at_ms,
                conflict.resolved_at_ms
            ],
        )?;
        Ok(())
    }

    pub fn get_file_conflict(
        &self,
        conflict_id: &str,
    ) -> Result<Option<StoredFileConflict>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT conflict_id, root_name, resource_id, path, kind, status, base_tree_hash,
               local_tree_hash, remote_tree_hash, detail, resolution, resolution_note,
               created_at_ms, resolved_at_ms
               FROM file_conflicts WHERE conflict_id = ?1"#,
        )?;
        let mut rows = stmt.query(params![conflict_id])?;
        if let Some(row) = rows.next()? {
            Ok(Some(stored_file_conflict_from_row(row)?))
        } else {
            Ok(None)
        }
    }

    pub fn list_file_conflicts(
        &self,
        root_name: Option<&str>,
    ) -> Result<Vec<StoredFileConflict>, StoreError> {
        let sql = if root_name.is_some() {
            r#"SELECT conflict_id, root_name, resource_id, path, kind, status, base_tree_hash,
               local_tree_hash, remote_tree_hash, detail, resolution, resolution_note,
               created_at_ms, resolved_at_ms
               FROM file_conflicts WHERE root_name = ?1
               ORDER BY status, created_at_ms, conflict_id"#
        } else {
            r#"SELECT conflict_id, root_name, resource_id, path, kind, status, base_tree_hash,
               local_tree_hash, remote_tree_hash, detail, resolution, resolution_note,
               created_at_ms, resolved_at_ms
               FROM file_conflicts
               ORDER BY status, root_name, created_at_ms, conflict_id"#
        };
        let mut stmt = self.conn.prepare(sql)?;
        let rows = if let Some(root_name) = root_name {
            stmt.query_map(params![root_name], stored_file_conflict_from_row)?
        } else {
            stmt.query_map([], stored_file_conflict_from_row)?
        };
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_resource_secret(&self, secret: &StoredResourceSecret) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO resource_secrets(
                secret_id, resource_id, epoch, status, created_at_ms
            ) VALUES (?1, ?2, ?3, ?4, ?5)"#,
            params![
                secret.secret_id,
                secret.resource_id,
                secret.epoch as i64,
                secret.status,
                secret.created_at_ms
            ],
        )?;
        Ok(())
    }

    pub fn list_resource_secrets(&self) -> Result<Vec<StoredResourceSecret>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT secret_id, resource_id, epoch, status, created_at_ms
               FROM resource_secrets ORDER BY resource_id, epoch, secret_id"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredResourceSecret {
                secret_id: row.get(0)?,
                resource_id: row.get(1)?,
                epoch: row.get::<_, i64>(2)? as u64,
                status: row.get(3)?,
                created_at_ms: row.get(4)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn latest_resource_secret(
        &self,
        resource_id: &str,
    ) -> Result<Option<StoredResourceSecret>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT secret_id, resource_id, epoch, status, created_at_ms
               FROM resource_secrets
               WHERE resource_id = ?1
               ORDER BY epoch DESC, created_at_ms DESC, secret_id DESC
               LIMIT 1"#,
        )?;
        let mut rows = stmt.query(params![resource_id])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredResourceSecret {
                secret_id: row.get(0)?,
                resource_id: row.get(1)?,
                epoch: row.get::<_, i64>(2)? as u64,
                status: row.get(3)?,
                created_at_ms: row.get(4)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn record_cas_object(
        &self,
        hash: &str,
        size_bytes: u64,
        path: &str,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR REPLACE INTO cas_objects(hash, size_bytes, path, created_at_ms) VALUES (?1, ?2, ?3, ?4)",
            params![hash, size_bytes as i64, path, now_ms()],
        )?;
        Ok(())
    }

    pub fn delete_cas_object(&self, hash: &str) -> Result<(), StoreError> {
        self.conn
            .execute("DELETE FROM cas_objects WHERE hash = ?1", params![hash])?;
        self.conn
            .execute("DELETE FROM cas_pins WHERE hash = ?1", params![hash])?;
        Ok(())
    }

    pub fn list_cas_objects(&self) -> Result<Vec<CasObject>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT hash, size_bytes, path FROM cas_objects ORDER BY created_at_ms, hash",
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(CasObject {
                hash: row.get(0)?,
                size_bytes: row.get::<_, i64>(1)? as u64,
                path: row.get(2)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn pin_cas_object(&self, hash: &str) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR REPLACE INTO cas_pins(hash, pinned_at_ms) VALUES (?1, ?2)",
            params![hash, now_ms()],
        )?;
        Ok(())
    }

    pub fn unpin_cas_object(&self, hash: &str) -> Result<(), StoreError> {
        self.conn
            .execute("DELETE FROM cas_pins WHERE hash = ?1", params![hash])?;
        Ok(())
    }

    pub fn is_cas_object_pinned(&self, hash: &str) -> Result<bool, StoreError> {
        let count: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM cas_pins WHERE hash = ?1",
            params![hash],
            |row| row.get(0),
        )?;
        Ok(count > 0)
    }

    pub fn list_cas_pins(&self) -> Result<Vec<CasPin>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT hash, pinned_at_ms FROM cas_pins ORDER BY pinned_at_ms, hash")?;
        let rows = stmt.query_map([], |row| {
            Ok(CasPin {
                hash: row.get(0)?,
                pinned_at_ms: row.get(1)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn record_cas_provider(
        &self,
        hash: &str,
        peer_node_id: &str,
        endpoint_id: &str,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO cas_providers(hash, peer_node_id, endpoint_id, last_seen_ms)
               VALUES (?1, ?2, ?3, ?4)"#,
            params![hash, peer_node_id, endpoint_id, now_ms()],
        )?;
        Ok(())
    }

    pub fn list_cas_providers(&self, hash: &str) -> Result<Vec<CasProvider>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT hash, peer_node_id, endpoint_id, last_seen_ms
               FROM cas_providers
               WHERE hash = ?1
               ORDER BY last_seen_ms DESC, peer_node_id"#,
        )?;
        let rows = stmt.query_map(params![hash], |row| {
            Ok(CasProvider {
                hash: row.get(0)?,
                peer_node_id: row.get(1)?,
                endpoint_id: row.get(2)?,
                last_seen_ms: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn upsert_peer_card(&self, peer_card: &StoredPeerCard) -> Result<(), StoreError> {
        self.conn.execute(
            "INSERT OR REPLACE INTO peer_cards(peer_id, card_json, updated_at_ms) VALUES (?1, ?2, ?3)",
            params![peer_card.peer_id, peer_card.card_json, peer_card.updated_at_ms],
        )?;
        Ok(())
    }

    pub fn get_peer_card(&self, peer_id: &str) -> Result<Option<StoredPeerCard>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT peer_id, card_json, updated_at_ms FROM peer_cards WHERE peer_id = ?1",
        )?;
        let mut rows = stmt.query(params![peer_id])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredPeerCard {
                peer_id: row.get(0)?,
                card_json: row.get(1)?,
                updated_at_ms: row.get(2)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn list_peer_cards(&self) -> Result<Vec<StoredPeerCard>, StoreError> {
        let mut stmt = self
            .conn
            .prepare("SELECT peer_id, card_json, updated_at_ms FROM peer_cards ORDER BY peer_id")?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredPeerCard {
                peer_id: row.get(0)?,
                card_json: row.get(1)?,
                updated_at_ms: row.get(2)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn put_module_state(&self, state: &StoredModuleState) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO module_state(module, state_json, updated_at_ms)
               VALUES (?1, ?2, ?3)"#,
            params![state.module, state.state_json, state.updated_at_ms],
        )?;
        Ok(())
    }

    pub fn get_module_state(&self, module: &str) -> Result<Option<StoredModuleState>, StoreError> {
        let mut stmt = self.conn.prepare(
            "SELECT module, state_json, updated_at_ms FROM module_state WHERE module = ?1",
        )?;
        let mut rows = stmt.query(params![module])?;
        if let Some(row) = rows.next()? {
            Ok(Some(StoredModuleState {
                module: row.get(0)?,
                state_json: row.get(1)?,
                updated_at_ms: row.get(2)?,
            }))
        } else {
            Ok(None)
        }
    }

    pub fn insert_auth_op(&self, op: &StoredAuthOp) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO auth_ops(op_id, resource_id, op_json, created_at_ms)
               VALUES (?1, ?2, ?3, ?4)"#,
            params![op.op_id, op.resource_id, op.op_json, op.created_at_ms],
        )?;
        Ok(())
    }

    pub fn list_auth_ops_for_resource(
        &self,
        resource_id: &str,
    ) -> Result<Vec<StoredAuthOp>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT op_id, resource_id, op_json, created_at_ms
               FROM auth_ops WHERE resource_id = ?1 ORDER BY created_at_ms, op_id"#,
        )?;
        let rows = stmt.query_map(params![resource_id], |row| {
            Ok(StoredAuthOp {
                op_id: row.get(0)?,
                resource_id: row.get(1)?,
                op_json: row.get(2)?,
                created_at_ms: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn list_auth_ops(&self) -> Result<Vec<StoredAuthOp>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT op_id, resource_id, op_json, created_at_ms
               FROM auth_ops ORDER BY resource_id, created_at_ms, op_id"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredAuthOp {
                op_id: row.get(0)?,
                resource_id: row.get(1)?,
                op_json: row.get(2)?,
                created_at_ms: row.get(3)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_keychain_op(&self, op: &StoredKeychainOp) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO keychain_ops(op_id, op_json, created_at_ms)
               VALUES (?1, ?2, ?3)"#,
            params![op.op_id, op.op_json, op.created_at_ms],
        )?;
        Ok(())
    }

    pub fn list_keychain_ops(&self) -> Result<Vec<StoredKeychainOp>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT op_id, op_json, created_at_ms
               FROM keychain_ops ORDER BY created_at_ms, op_id"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredKeychainOp {
                op_id: row.get(0)?,
                op_json: row.get(1)?,
                created_at_ms: row.get(2)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_keychain_signature(
        &self,
        signature: &StoredKeychainSignature,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO keychain_signatures(
               op_id, signer, signer_public_key, namespace, signature, created_at_ms
               )
               VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#,
            params![
                signature.op_id,
                signature.signer,
                signature.signer_public_key,
                signature.namespace,
                signature.signature,
                signature.created_at_ms
            ],
        )?;
        Ok(())
    }

    pub fn list_keychain_signatures(&self) -> Result<Vec<StoredKeychainSignature>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT op_id, signer, signer_public_key, namespace, signature, created_at_ms
               FROM keychain_signatures ORDER BY created_at_ms, op_id, signer, namespace"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredKeychainSignature {
                op_id: row.get(0)?,
                signer: row.get(1)?,
                signer_public_key: row.get(2)?,
                namespace: row.get(3)?,
                signature: row.get(4)?,
                created_at_ms: row.get(5)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_ssh_cert_request(
        &self,
        request: &StoredSshCertRequest,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO ssh_cert_requests(
                request_id, requester_node, public_key, public_key_fingerprint, cert_kind,
                principals_json, requested_validity, renewal_of, reason, status, created_at_ms,
                provenance_json
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)"#,
            params![
                request.request_id,
                request.requester_node,
                request.public_key,
                request.public_key_fingerprint,
                request.cert_kind,
                serde_json::to_string(&request.principals)?,
                request.requested_validity,
                request.renewal_of,
                request.reason,
                request.status,
                request.created_at_ms,
                request.provenance_json
            ],
        )?;
        Ok(())
    }

    pub fn get_ssh_cert_request(
        &self,
        request_id: &str,
    ) -> Result<Option<StoredSshCertRequest>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT request_id, requester_node, public_key, public_key_fingerprint, cert_kind,
               principals_json, requested_validity, renewal_of, reason, status, created_at_ms,
               provenance_json
               FROM ssh_cert_requests WHERE request_id = ?1"#,
        )?;
        let mut rows = stmt.query(params![request_id])?;
        if let Some(row) = rows.next()? {
            Ok(Some(stored_ssh_cert_request_from_row(row)?))
        } else {
            Ok(None)
        }
    }

    pub fn update_ssh_cert_request_status(
        &self,
        request_id: &str,
        status: &str,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            "UPDATE ssh_cert_requests SET status = ?2 WHERE request_id = ?1",
            params![request_id, status],
        )?;
        Ok(())
    }

    pub fn list_ssh_cert_requests(&self) -> Result<Vec<StoredSshCertRequest>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT request_id, requester_node, public_key, public_key_fingerprint, cert_kind,
               principals_json, requested_validity, renewal_of, reason, status, created_at_ms,
               provenance_json
               FROM ssh_cert_requests ORDER BY created_at_ms, request_id"#,
        )?;
        let rows = stmt.query_map([], stored_ssh_cert_request_from_row)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn list_ssh_cert_requests_since(
        &self,
        since_ms: i64,
    ) -> Result<Vec<StoredSshCertRequest>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT request_id, requester_node, public_key, public_key_fingerprint, cert_kind,
               principals_json, requested_validity, renewal_of, reason, status, created_at_ms,
               provenance_json
               FROM ssh_cert_requests WHERE created_at_ms >= ?1 ORDER BY created_at_ms, request_id"#,
        )?;
        let rows = stmt.query_map(params![since_ms], stored_ssh_cert_request_from_row)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_ssh_certificate(
        &self,
        certificate: &StoredSshCertificate,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO ssh_certificates(
                cert_id, request_id, certificate, certificate_fingerprint, imported_at_ms,
                provenance_json
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)"#,
            params![
                certificate.cert_id,
                certificate.request_id,
                certificate.certificate,
                certificate.certificate_fingerprint,
                certificate.imported_at_ms,
                certificate.provenance_json
            ],
        )?;
        Ok(())
    }

    pub fn list_ssh_certificates(&self) -> Result<Vec<StoredSshCertificate>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT cert_id, request_id, certificate, certificate_fingerprint, imported_at_ms
               , provenance_json
               FROM ssh_certificates ORDER BY imported_at_ms, cert_id"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredSshCertificate {
                cert_id: row.get(0)?,
                request_id: row.get(1)?,
                certificate: row.get(2)?,
                certificate_fingerprint: row.get(3)?,
                imported_at_ms: row.get(4)?,
                provenance_json: row.get(5)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn list_ssh_certificates_since(
        &self,
        since_ms: i64,
    ) -> Result<Vec<StoredSshCertificate>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT cert_id, request_id, certificate, certificate_fingerprint, imported_at_ms
               , provenance_json
               FROM ssh_certificates WHERE imported_at_ms >= ?1 ORDER BY imported_at_ms, cert_id"#,
        )?;
        let rows = stmt.query_map(params![since_ms], |row| {
            Ok(StoredSshCertificate {
                cert_id: row.get(0)?,
                request_id: row.get(1)?,
                certificate: row.get(2)?,
                certificate_fingerprint: row.get(3)?,
                imported_at_ms: row.get(4)?,
                provenance_json: row.get(5)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn insert_ssh_revocation(
        &self,
        revocation: &StoredSshRevocation,
    ) -> Result<(), StoreError> {
        self.conn.execute(
            r#"INSERT OR REPLACE INTO ssh_revocations(
                revocation_id, kind, target, reason, created_at_ms, published, provenance_json
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
            params![
                revocation.revocation_id,
                revocation.kind,
                revocation.target,
                revocation.reason,
                revocation.created_at_ms,
                if revocation.published { 1_i64 } else { 0_i64 },
                revocation.provenance_json
            ],
        )?;
        Ok(())
    }

    pub fn list_ssh_revocations(&self) -> Result<Vec<StoredSshRevocation>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT revocation_id, kind, target, reason, created_at_ms, published
               , provenance_json
               FROM ssh_revocations ORDER BY created_at_ms, revocation_id"#,
        )?;
        let rows = stmt.query_map([], |row| {
            Ok(StoredSshRevocation {
                revocation_id: row.get(0)?,
                kind: row.get(1)?,
                target: row.get(2)?,
                reason: row.get(3)?,
                created_at_ms: row.get(4)?,
                published: row.get::<_, i64>(5)? != 0,
                provenance_json: row.get(6)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }

    pub fn list_ssh_revocations_since(
        &self,
        since_ms: i64,
    ) -> Result<Vec<StoredSshRevocation>, StoreError> {
        let mut stmt = self.conn.prepare(
            r#"SELECT revocation_id, kind, target, reason, created_at_ms, published
               , provenance_json
               FROM ssh_revocations WHERE created_at_ms >= ?1 ORDER BY created_at_ms, revocation_id"#,
        )?;
        let rows = stmt.query_map(params![since_ms], |row| {
            Ok(StoredSshRevocation {
                revocation_id: row.get(0)?,
                kind: row.get(1)?,
                target: row.get(2)?,
                reason: row.get(3)?,
                created_at_ms: row.get(4)?,
                published: row.get::<_, i64>(5)? != 0,
                provenance_json: row.get(6)?,
            })
        })?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(StoreError::from)
    }
}

fn stored_ssh_cert_request_from_row(
    row: &rusqlite::Row<'_>,
) -> Result<StoredSshCertRequest, rusqlite::Error> {
    let principals_json: String = row.get(5)?;
    let principals = serde_json::from_str(&principals_json).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(5, rusqlite::types::Type::Text, Box::new(error))
    })?;
    Ok(StoredSshCertRequest {
        request_id: row.get(0)?,
        requester_node: row.get(1)?,
        public_key: row.get(2)?,
        public_key_fingerprint: row.get(3)?,
        cert_kind: row.get(4)?,
        principals,
        requested_validity: row.get(6)?,
        renewal_of: row.get(7)?,
        reason: row.get(8)?,
        status: row.get(9)?,
        created_at_ms: row.get(10)?,
        provenance_json: row.get(11)?,
    })
}

fn stored_file_conflict_from_row(
    row: &rusqlite::Row<'_>,
) -> Result<StoredFileConflict, rusqlite::Error> {
    Ok(StoredFileConflict {
        conflict_id: row.get(0)?,
        root_name: row.get(1)?,
        resource_id: row.get(2)?,
        path: row.get(3)?,
        kind: row.get(4)?,
        status: row.get(5)?,
        base_tree_hash: row.get(6)?,
        local_tree_hash: row.get(7)?,
        remote_tree_hash: row.get(8)?,
        detail: row.get(9)?,
        resolution: row.get(10)?,
        resolution_note: row.get(11)?,
        created_at_ms: row.get(12)?,
        resolved_at_ms: row.get(13)?,
    })
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredResource {
    pub resource_id: String,
    pub kind: String,
    pub name: String,
    pub status: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CasProvider {
    pub hash: String,
    pub peer_node_id: String,
    pub endpoint_id: String,
    pub last_seen_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredDbResource {
    pub db_id: String,
    pub resource_id: String,
    pub name: String,
    pub path: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKvStore {
    pub kv_id: String,
    pub resource_id: String,
    pub name: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKvEntry {
    pub kv_id: String,
    pub key: String,
    pub value: String,
    pub updated_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredDocumentResource {
    pub document_id: String,
    pub resource_id: String,
    pub name: String,
    pub state_json: String,
    pub updated_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredFileRoot {
    pub root_id: String,
    pub resource_id: String,
    pub name: String,
    pub path: String,
    pub latest_tree_hash: Option<String>,
    pub latest_tree_json: Option<String>,
    pub updated_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredFileConflict {
    pub conflict_id: String,
    pub root_name: String,
    pub resource_id: String,
    pub path: String,
    pub kind: String,
    pub status: String,
    pub base_tree_hash: Option<String>,
    pub local_tree_hash: Option<String>,
    pub remote_tree_hash: Option<String>,
    pub detail: String,
    pub resolution: Option<String>,
    pub resolution_note: Option<String>,
    pub created_at_ms: i64,
    pub resolved_at_ms: Option<i64>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredResourceSecret {
    pub secret_id: String,
    pub resource_id: String,
    pub epoch: u64,
    pub status: String,
    pub created_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CasObject {
    pub hash: String,
    pub size_bytes: u64,
    pub path: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CasPin {
    pub hash: String,
    pub pinned_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredPeerCard {
    pub peer_id: String,
    pub card_json: String,
    pub updated_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredModuleState {
    pub module: String,
    pub state_json: String,
    pub updated_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredAuthOp {
    pub op_id: String,
    pub resource_id: String,
    pub op_json: String,
    pub created_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKeychainOp {
    pub op_id: String,
    pub op_json: String,
    pub created_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredKeychainSignature {
    pub op_id: String,
    pub signer: String,
    pub signer_public_key: String,
    pub namespace: String,
    pub signature: Vec<u8>,
    pub created_at_ms: i64,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredSshCertRequest {
    pub request_id: String,
    pub requester_node: String,
    pub public_key: String,
    pub public_key_fingerprint: String,
    pub cert_kind: String,
    pub principals: Vec<String>,
    pub requested_validity: Option<String>,
    pub renewal_of: Option<String>,
    pub reason: Option<String>,
    pub status: String,
    pub created_at_ms: i64,
    pub provenance_json: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredSshCertificate {
    pub cert_id: String,
    pub request_id: String,
    pub certificate: String,
    pub certificate_fingerprint: String,
    pub imported_at_ms: i64,
    pub provenance_json: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StoredSshRevocation {
    pub revocation_id: String,
    pub kind: String,
    pub target: String,
    pub reason: Option<String>,
    pub created_at_ms: i64,
    pub published: bool,
    pub provenance_json: Option<String>,
}

#[must_use]
pub fn now_ms() -> i64 {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default();
    i64::try_from(now.as_millis()).unwrap_or(i64::MAX)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn migrations_are_idempotent() {
        let store = Store::open_memory().expect("open");
        store.migrate().expect("migrate again");
        store.migrate().expect("migrate third time");
        let resources = store.list_resources().expect("resources");
        assert!(resources.is_empty());
    }

    #[test]
    fn ssh_cert_request_and_revocation_roundtrip() {
        let store = Store::open_memory().expect("open");
        let request = StoredSshCertRequest {
            request_id: "ssh-cert-request:1".to_owned(),
            requester_node: "node:laptop".to_owned(),
            public_key: "ssh-ed25519 AAAA test".to_owned(),
            public_key_fingerprint: "ssh:blake3:test".to_owned(),
            cert_kind: "user".to_owned(),
            principals: vec!["eric".to_owned()],
            requested_validity: Some("+52w".to_owned()),
            renewal_of: None,
            reason: Some("renewal".to_owned()),
            status: "pending".to_owned(),
            created_at_ms: 1,
            provenance_json: Some(r#"{"test":true}"#.to_owned()),
        };
        store
            .insert_ssh_cert_request(&request)
            .expect("insert request");
        assert_eq!(
            store
                .get_ssh_cert_request("ssh-cert-request:1")
                .expect("get request"),
            Some(request.clone())
        );

        let revocation = StoredSshRevocation {
            revocation_id: "ssh-revocation:1".to_owned(),
            kind: "public-key".to_owned(),
            target: "ssh:blake3:test".to_owned(),
            reason: Some("lost key".to_owned()),
            created_at_ms: 2,
            published: true,
            provenance_json: Some(r#"{"test":true}"#.to_owned()),
        };
        store
            .insert_ssh_revocation(&revocation)
            .expect("insert revocation");
        assert_eq!(
            store.list_ssh_revocations().expect("list revocations"),
            vec![revocation.clone()]
        );
        assert_eq!(
            store
                .list_ssh_cert_requests_since(0)
                .expect("list requests since"),
            vec![request]
        );
        assert_eq!(
            store
                .list_ssh_revocations_since(1)
                .expect("list revocations since"),
            vec![revocation]
        );
    }

    #[test]
    fn peer_card_roundtrip() {
        let store = Store::open_memory().expect("open");
        let peer_card = StoredPeerCard {
            peer_id: "node:laptop".to_owned(),
            card_json: r#"{"node_id":"node:laptop"}"#.to_owned(),
            updated_at_ms: 10,
        };
        store.upsert_peer_card(&peer_card).expect("insert");
        assert_eq!(
            store.get_peer_card("node:laptop").expect("get"),
            Some(peer_card.clone())
        );
        assert_eq!(store.list_peer_cards().expect("list"), vec![peer_card]);
    }

    #[test]
    fn module_state_roundtrip() {
        let store = Store::open_memory().expect("open");
        let state = StoredModuleState {
            module: "live-sync:node:laptop:ssh-certs".to_owned(),
            state_json: r#"{"cursor_ms":42}"#.to_owned(),
            updated_at_ms: 43,
        };
        store.put_module_state(&state).expect("put state");
        assert_eq!(
            store
                .get_module_state("live-sync:node:laptop:ssh-certs")
                .expect("get state"),
            Some(state)
        );
    }

    #[test]
    fn auth_ops_roundtrip_by_resource() {
        let store = Store::open_memory().expect("open");
        let first = StoredAuthOp {
            op_id: "op:auth:1".to_owned(),
            resource_id: "resource:notes".to_owned(),
            op_json: r#"{"id":"op:auth:1"}"#.to_owned(),
            created_at_ms: 1,
        };
        let second = StoredAuthOp {
            op_id: "op:auth:2".to_owned(),
            resource_id: "resource:other".to_owned(),
            op_json: r#"{"id":"op:auth:2"}"#.to_owned(),
            created_at_ms: 2,
        };

        store.insert_auth_op(&second).expect("insert second");
        store.insert_auth_op(&first).expect("insert first");

        assert_eq!(
            store
                .list_auth_ops_for_resource("resource:notes")
                .expect("list auth ops"),
            vec![first]
        );
        assert_eq!(store.list_auth_ops().expect("list all auth ops").len(), 2);
    }

    #[test]
    fn keychain_ops_roundtrip() {
        let store = Store::open_memory().expect("open");
        let first = StoredKeychainOp {
            op_id: "op:keychain:1".to_owned(),
            op_json: r#"{"id":"op:keychain:1"}"#.to_owned(),
            created_at_ms: 1,
        };
        let second = StoredKeychainOp {
            op_id: "op:keychain:2".to_owned(),
            op_json: r#"{"id":"op:keychain:2"}"#.to_owned(),
            created_at_ms: 2,
        };

        store.insert_keychain_op(&second).expect("insert second");
        store.insert_keychain_op(&first).expect("insert first");

        assert_eq!(
            store.list_keychain_ops().expect("list keychain ops"),
            vec![first, second]
        );
    }

    #[test]
    fn keychain_signatures_roundtrip() {
        let store = Store::open_memory().expect("open");
        let signature = StoredKeychainSignature {
            op_id: "op:keychain:1".to_owned(),
            signer: "ssh:blake3:admin".to_owned(),
            signer_public_key: "ssh-ed25519 AAAAADMIN eric@geth".to_owned(),
            namespace: "geth.keychain.v1@geth.local".to_owned(),
            signature: b"-----BEGIN SSH SIGNATURE-----".to_vec(),
            created_at_ms: 3,
        };

        store
            .insert_keychain_signature(&signature)
            .expect("insert signature");

        assert_eq!(
            store
                .list_keychain_signatures()
                .expect("list keychain signatures"),
            vec![signature]
        );
    }

    #[test]
    fn cas_pins_roundtrip() {
        let store = Store::open_memory().expect("open");
        let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";

        assert!(!store.is_cas_object_pinned(hash).expect("unpinned"));
        store.pin_cas_object(hash).expect("pin");
        assert!(store.is_cas_object_pinned(hash).expect("pinned"));
        assert_eq!(store.list_cas_pins().expect("pins").len(), 1);
        store.unpin_cas_object(hash).expect("unpin");
        assert!(!store.is_cas_object_pinned(hash).expect("unpinned again"));
        assert!(store.list_cas_pins().expect("pins").is_empty());
    }

    #[test]
    fn cas_providers_roundtrip_by_hash() {
        let store = Store::open_memory().expect("open");
        let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
        store
            .record_cas_provider(hash, "node:laptop", "endpoint:laptop")
            .expect("record provider");

        let providers = store.list_cas_providers(hash).expect("list providers");

        assert_eq!(providers.len(), 1);
        assert_eq!(providers[0].hash, hash);
        assert_eq!(providers[0].peer_node_id, "node:laptop");
        assert_eq!(providers[0].endpoint_id, "endpoint:laptop");
        assert!(
            store
                .list_cas_providers(
                    "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
                )
                .expect("list missing")
                .is_empty()
        );
    }

    #[test]
    fn deleting_cas_object_removes_pin_metadata() {
        let store = Store::open_memory().expect("open");
        let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
        store
            .record_cas_object(hash, 10, "/tmp/blob")
            .expect("record object");
        store.pin_cas_object(hash).expect("pin");

        store.delete_cas_object(hash).expect("delete object");

        assert!(store.list_cas_objects().expect("objects").is_empty());
        assert!(store.list_cas_pins().expect("pins").is_empty());
    }

    #[test]
    fn db_resource_roundtrip_by_name() {
        let store = Store::open_memory().expect("open");
        let resource = StoredResource {
            resource_id: "resource:db:notes".to_owned(),
            kind: "db".to_owned(),
            name: "notes".to_owned(),
            status: "active".to_owned(),
        };
        store.insert_resource(&resource).expect("insert resource");
        store
            .insert_db_resource(&StoredDbResource {
                db_id: "db:notes".to_owned(),
                resource_id: resource.resource_id.clone(),
                name: resource.name.clone(),
                path: "/tmp/notes.sqlite".to_owned(),
            })
            .expect("insert db");

        assert_eq!(
            store
                .get_resource_by_kind_name("db", "notes")
                .expect("get resource"),
            Some(resource)
        );
        assert_eq!(
            store
                .get_db_resource_by_name("notes")
                .expect("get db")
                .map(|db| db.path),
            Some("/tmp/notes.sqlite".to_owned())
        );
        assert_eq!(
            store
                .list_db_resources()
                .expect("list db resources")
                .into_iter()
                .map(|db| db.name)
                .collect::<Vec<_>>(),
            vec!["notes".to_owned()]
        );
    }

    #[test]
    fn file_roots_roundtrip_by_name() {
        let store = Store::open_memory().expect("open");
        let root = StoredFileRoot {
            root_id: "file-root:notes".to_owned(),
            resource_id: "resource:cas-tree:notes".to_owned(),
            name: "notes".to_owned(),
            path: "/tmp/notes".to_owned(),
            latest_tree_hash: Some(
                "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".to_owned(),
            ),
            latest_tree_json: Some("{}".to_owned()),
            updated_at_ms: 10,
        };

        store.upsert_file_root(&root).expect("insert root");

        assert_eq!(
            store.get_file_root_by_name("notes").expect("get file root"),
            Some(root.clone())
        );
        assert_eq!(store.list_file_roots().expect("list roots"), vec![root]);
    }

    #[test]
    fn file_conflicts_roundtrip_and_resolve() {
        let store = Store::open_memory().expect("open");
        let conflict = StoredFileConflict {
            conflict_id: "file-conflict:1".to_owned(),
            root_name: "notes".to_owned(),
            resource_id: "resource:cas-tree:notes".to_owned(),
            path: "notes/todo.md".to_owned(),
            kind: "concurrent-edit".to_owned(),
            status: "open".to_owned(),
            base_tree_hash: None,
            local_tree_hash: Some(
                "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".to_owned(),
            ),
            remote_tree_hash: Some(
                "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210".to_owned(),
            ),
            detail: "local and remote edits touched the same path".to_owned(),
            resolution: None,
            resolution_note: None,
            created_at_ms: 10,
            resolved_at_ms: None,
        };
        store
            .upsert_file_conflict(&conflict)
            .expect("insert conflict");
        assert_eq!(
            store
                .get_file_conflict("file-conflict:1")
                .expect("get conflict"),
            Some(conflict.clone())
        );
        assert_eq!(
            store
                .list_file_conflicts(Some("notes"))
                .expect("list by root"),
            vec![conflict.clone()]
        );

        let resolved = StoredFileConflict {
            status: "resolved".to_owned(),
            resolution: Some("keep-local".to_owned()),
            resolution_note: Some("local file is authoritative".to_owned()),
            resolved_at_ms: Some(20),
            ..conflict
        };
        store
            .upsert_file_conflict(&resolved)
            .expect("resolve conflict");

        assert_eq!(
            store.list_file_conflicts(None).expect("list all"),
            vec![resolved]
        );
    }

    #[test]
    fn kv_store_and_entries_roundtrip() {
        let store = Store::open_memory().expect("open");
        let kv = StoredKvStore {
            kv_id: "kv:prefs".to_owned(),
            resource_id: "resource:kv:prefs".to_owned(),
            name: "prefs".to_owned(),
        };
        store.insert_kv_store(&kv).expect("insert kv");
        assert_eq!(
            store.get_kv_store_by_name("prefs").expect("get kv"),
            Some(kv.clone())
        );
        assert_eq!(store.list_kv_stores().expect("list kv"), vec![kv.clone()]);

        let entry = StoredKvEntry {
            kv_id: kv.kv_id.clone(),
            key: "apps/foo/theme".to_owned(),
            value: "dark".to_owned(),
            updated_at_ms: 1,
        };
        store.set_kv_entry(&entry).expect("set entry");
        assert_eq!(
            store
                .get_kv_entry("kv:prefs", "apps/foo/theme")
                .expect("get entry"),
            Some(entry.clone())
        );
        assert_eq!(
            store
                .list_kv_entries_since("kv:prefs", 1)
                .expect("list entries since"),
            vec![entry]
        );
    }

    #[test]
    fn document_resource_roundtrip_by_name() {
        let store = Store::open_memory().expect("open");
        let resource = StoredResource {
            resource_id: "resource:document:notes".to_owned(),
            kind: "document".to_owned(),
            name: "notes".to_owned(),
            status: "active".to_owned(),
        };
        store.insert_resource(&resource).expect("insert resource");
        let document = StoredDocumentResource {
            document_id: "document:notes".to_owned(),
            resource_id: resource.resource_id,
            name: resource.name,
            state_json: "{}".to_owned(),
            updated_at_ms: 1,
        };
        store
            .insert_document_resource(&document)
            .expect("insert document");

        assert_eq!(
            store
                .get_document_resource_by_name("notes")
                .expect("get document"),
            Some(document)
        );
    }

    #[test]
    fn resource_secret_epochs_roundtrip() {
        let store = Store::open_memory().expect("open");
        let first = StoredResourceSecret {
            secret_id: "secret:1".to_owned(),
            resource_id: "resource:kv:prefs".to_owned(),
            epoch: 1,
            status: "active".to_owned(),
            created_at_ms: 1,
        };
        let second = StoredResourceSecret {
            secret_id: "secret:2".to_owned(),
            resource_id: "resource:kv:prefs".to_owned(),
            epoch: 2,
            status: "active".to_owned(),
            created_at_ms: 2,
        };

        store.insert_resource_secret(&first).expect("insert first");
        store
            .insert_resource_secret(&second)
            .expect("insert second");

        assert_eq!(
            store
                .latest_resource_secret("resource:kv:prefs")
                .expect("latest"),
            Some(second.clone())
        );
        assert_eq!(
            store.list_resource_secrets().expect("list"),
            vec![first, second]
        );
    }
}