use rusqlite::{Connection, OptionalExtension, Transaction, params}; use std::path::Path; pub const CURRENT_SCHEMA_VERSION: i64 = 2; pub const FILE_JOURNAL_MODE: &str = "wal"; pub const SYNCHRONOUS_MODE: &str = "normal"; #[derive(Debug, thiserror::Error)] pub enum StoreError { #[error("sqlite error: {0}")] Sqlite(#[from] rusqlite::Error), #[error("json error: {0}")] Json(#[from] serde_json::Error), #[error("invalid schema version value: {0}")] InvalidSchemaVersion(String), } pub struct Store { conn: Connection, } fn configure_file_connection(conn: &Connection) -> Result<(), StoreError> { conn.execute_batch( r#" PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON; "#, )?; Ok(()) } fn configure_memory_connection(conn: &Connection) -> Result<(), StoreError> { conn.execute_batch( r#" PRAGMA synchronous = NORMAL; PRAGMA foreign_keys = ON; "#, )?; Ok(()) } impl Store { pub fn open(path: &Path) -> Result { let conn = Connection::open(path)?; configure_file_connection(&conn)?; let store = Self { conn }; store.migrate()?; Ok(store) } pub fn open_memory() -> Result { let conn = Connection::open_in_memory()?; configure_memory_connection(&conn)?; 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 ); "#, )?; let mut version = self.schema_version()?; if version == 0 { self.apply_schema_v1()?; version = 1; } while version < CURRENT_SCHEMA_VERSION { let next = version + 1; match next { 2 => self.apply_migration_v2()?, _ => unreachable!("missing migration for schema version {next}"), } version = next; } Ok(()) } pub fn schema_version(&self) -> Result { let version = self .conn .query_row( "SELECT value FROM meta WHERE key = 'schema_version'", [], |row| row.get::<_, String>(0), ) .optional()?; let Some(version) = version else { return Ok(0); }; version .parse::() .map_err(|_| StoreError::InvalidSchemaVersion(version)) } pub fn journal_mode(&self) -> Result { self.conn .query_row("PRAGMA journal_mode", [], |row| row.get::<_, String>(0)) .map(|mode| mode.to_ascii_lowercase()) .map_err(StoreError::from) } pub fn synchronous_mode(&self) -> Result { let mode = self .conn .query_row("PRAGMA synchronous", [], |row| row.get::<_, i64>(0))?; Ok(match mode { 0 => "off".to_owned(), 1 => "normal".to_owned(), 2 => "full".to_owned(), 3 => "extra".to_owned(), other => format!("unknown-{other}"), }) } fn apply_schema_v1(&self) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; tx.execute_batch( r#" 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, 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 node_enrollment_requests ( request_id TEXT PRIMARY KEY, requester_node TEXT NOT NULL, request_json TEXT NOT NULL, status TEXT NOT NULL, created_at_ms INTEGER NOT NULL ); 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 auth_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 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 ); 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 ); 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 ); "#, )?; self.set_schema_version_tx(&tx, 1)?; tx.commit()?; Ok(()) } fn apply_migration_v2(&self) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; Self::add_column_if_missing_tx( &tx, "keychain_signatures", "signer_public_key", "TEXT NOT NULL DEFAULT ''", )?; Self::add_column_if_missing_tx(&tx, "ssh_cert_requests", "provenance_json", "TEXT")?; Self::add_column_if_missing_tx(&tx, "ssh_certificates", "provenance_json", "TEXT")?; Self::add_column_if_missing_tx(&tx, "ssh_revocations", "provenance_json", "TEXT")?; self.set_schema_version_tx(&tx, 2)?; tx.commit()?; Ok(()) } fn set_schema_version_tx(&self, tx: &Transaction<'_>, version: i64) -> Result<(), StoreError> { tx.execute( "INSERT INTO meta(key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", [version.to_string()], )?; Ok(()) } fn add_column_if_missing_tx( tx: &Transaction<'_>, table: &str, column: &str, definition: &str, ) -> Result<(), StoreError> { let mut stmt = tx.prepare(&format!("PRAGMA table_info({table})"))?; let columns = stmt .query_map([], |row| row.get::<_, String>(1))? .collect::, _>>()?; if !columns.iter().any(|name| name == column) { tx.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, 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::, _>>() .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, 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, 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, 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::, _>>() .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, 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, 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::, _>>() .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, 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, 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::, _>>() .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, 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, 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::, _>>() .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, 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, 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::, _>>() .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, 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, 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::, _>>() .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, 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::, _>>() .map_err(StoreError::from) } pub fn latest_resource_secret( &self, resource_id: &str, ) -> Result, 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, 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::, _>>() .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 { 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, 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::, _>>() .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, 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::, _>>() .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, 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, 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::, _>>() .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, 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 delete_module_state(&self, module: &str) -> Result<(), StoreError> { self.conn.execute( "DELETE FROM module_state WHERE module = ?1", params![module], )?; Ok(()) } pub fn list_module_states_with_prefix( &self, prefix: &str, ) -> Result, StoreError> { let like = format!("{prefix}%"); let mut stmt = self.conn.prepare( r#"SELECT module, state_json, updated_at_ms FROM module_state WHERE module LIKE ?1 ORDER BY module"#, )?; let rows = stmt.query_map(params![like], |row| { Ok(StoredModuleState { module: row.get(0)?, state_json: row.get(1)?, updated_at_ms: row.get(2)?, }) })?; rows.collect::, _>>() .map_err(StoreError::from) } 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 insert_auth_op_with_signatures( &self, op: &StoredAuthOp, signatures: &[StoredAuthSignature], ) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?; tx.commit()?; Ok(()) } pub fn insert_auth_ops_with_signatures( &self, ops: &[(StoredAuthOp, Vec)], ) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; for (op, signatures) in ops { Self::insert_auth_op_with_signatures_tx(&tx, op, signatures)?; } tx.commit()?; Ok(()) } pub fn list_auth_ops_for_resource( &self, resource_id: &str, ) -> Result, 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::, _>>() .map_err(StoreError::from) } pub fn list_auth_ops(&self) -> Result, 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::, _>>() .map_err(StoreError::from) } pub fn insert_auth_signature(&self, signature: &StoredAuthSignature) -> Result<(), StoreError> { self.conn.execute( r#"INSERT OR REPLACE INTO auth_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_auth_signatures(&self) -> Result, StoreError> { let mut stmt = self.conn.prepare( r#"SELECT op_id, signer, signer_public_key, namespace, signature, created_at_ms FROM auth_signatures ORDER BY created_at_ms, op_id, signer, namespace"#, )?; let rows = stmt.query_map([], |row| { Ok(StoredAuthSignature { 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::, _>>() .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 insert_keychain_op_with_signatures( &self, op: &StoredKeychainOp, signatures: &[StoredKeychainSignature], ) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?; tx.commit()?; Ok(()) } pub fn insert_keychain_ops_with_signatures( &self, ops: &[(StoredKeychainOp, Vec)], ) -> Result<(), StoreError> { let tx = self.conn.unchecked_transaction()?; for (op, signatures) in ops { Self::insert_keychain_op_with_signatures_tx(&tx, op, signatures)?; } tx.commit()?; Ok(()) } pub fn list_keychain_ops(&self) -> Result, 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::, _>>() .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, 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::, _>>() .map_err(StoreError::from) } fn insert_auth_op_with_signatures_tx( tx: &Transaction<'_>, op: &StoredAuthOp, signatures: &[StoredAuthSignature], ) -> Result<(), StoreError> { tx.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], )?; for signature in signatures { tx.execute( r#"INSERT OR REPLACE INTO auth_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(()) } fn insert_keychain_op_with_signatures_tx( tx: &Transaction<'_>, op: &StoredKeychainOp, signatures: &[StoredKeychainSignature], ) -> Result<(), StoreError> { tx.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], )?; for signature in signatures { tx.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 insert_node_enrollment_request( &self, request: &StoredNodeEnrollmentRequest, ) -> Result<(), StoreError> { self.conn.execute( r#"INSERT OR REPLACE INTO node_enrollment_requests( request_id, requester_node, request_json, status, created_at_ms ) VALUES (?1, ?2, ?3, ?4, ?5)"#, params![ request.request_id, request.requester_node, request.request_json, request.status, request.created_at_ms ], )?; Ok(()) } pub fn get_node_enrollment_request( &self, request_id: &str, ) -> Result, StoreError> { let mut stmt = self.conn.prepare( r#"SELECT request_id, requester_node, request_json, status, created_at_ms FROM node_enrollment_requests WHERE request_id = ?1"#, )?; let mut rows = stmt.query(params![request_id])?; if let Some(row) = rows.next()? { Ok(Some(StoredNodeEnrollmentRequest { request_id: row.get(0)?, requester_node: row.get(1)?, request_json: row.get(2)?, status: row.get(3)?, created_at_ms: row.get(4)?, })) } else { Ok(None) } } pub fn list_node_enrollment_requests( &self, ) -> Result, StoreError> { let mut stmt = self.conn.prepare( r#"SELECT request_id, requester_node, request_json, status, created_at_ms FROM node_enrollment_requests ORDER BY created_at_ms, request_id"#, )?; let rows = stmt.query_map([], |row| { Ok(StoredNodeEnrollmentRequest { request_id: row.get(0)?, requester_node: row.get(1)?, request_json: row.get(2)?, status: row.get(3)?, created_at_ms: row.get(4)?, }) })?; rows.collect::, _>>() .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, 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, 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::, _>>() .map_err(StoreError::from) } pub fn list_ssh_cert_requests_since( &self, since_ms: i64, ) -> Result, 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::, _>>() .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, 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::, _>>() .map_err(StoreError::from) } pub fn list_ssh_certificates_since( &self, since_ms: i64, ) -> Result, 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::, _>>() .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, 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::, _>>() .map_err(StoreError::from) } pub fn list_ssh_revocations_since( &self, since_ms: i64, ) -> Result, 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::, _>>() .map_err(StoreError::from) } } fn stored_ssh_cert_request_from_row( row: &rusqlite::Row<'_>, ) -> Result { 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 { 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, pub latest_tree_json: Option, 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, pub local_tree_hash: Option, pub remote_tree_hash: Option, pub detail: String, pub resolution: Option, pub resolution_note: Option, pub created_at_ms: i64, pub resolved_at_ms: Option, } #[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 StoredAuthSignature { pub op_id: String, pub signer: String, pub signer_public_key: String, pub namespace: String, pub signature: Vec, 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, pub created_at_ms: i64, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct StoredNodeEnrollmentRequest { pub request_id: String, pub requester_node: String, pub request_json: String, pub status: String, 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, pub requested_validity: Option, pub renewal_of: Option, pub reason: Option, pub status: String, pub created_at_ms: i64, pub provenance_json: Option, } #[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, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct StoredSshRevocation { pub revocation_id: String, pub kind: String, pub target: String, pub reason: Option, pub created_at_ms: i64, pub published: bool, pub provenance_json: Option, } #[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"); assert_eq!( store.schema_version().expect("schema version"), CURRENT_SCHEMA_VERSION ); let resources = store.list_resources().expect("resources"); assert!(resources.is_empty()); } #[test] fn schema_v1_fixture_migrates_to_current_schema() { let path = std::env::temp_dir().join(format!("geth-store-v1-migration-{}.sqlite", now_ms())); { let conn = Connection::open(&path).expect("open fixture"); conn.execute_batch( r#" CREATE TABLE meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); INSERT INTO meta(key, value) VALUES ('schema_version', '1'); CREATE TABLE keychain_signatures ( op_id TEXT NOT NULL, signer TEXT NOT NULL, namespace TEXT NOT NULL, signature BLOB NOT NULL, created_at_ms INTEGER NOT NULL, PRIMARY KEY (op_id, signer, namespace) ); CREATE TABLE 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 ); CREATE TABLE 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 ); CREATE TABLE 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 ); "#, ) .expect("create v1 fixture"); } let store = Store::open(&path).expect("migrate fixture"); assert_eq!( store.schema_version().expect("schema version"), CURRENT_SCHEMA_VERSION ); assert!(column_exists( &store, "keychain_signatures", "signer_public_key" )); assert!(column_exists( &store, "ssh_cert_requests", "provenance_json" )); assert!(column_exists(&store, "ssh_certificates", "provenance_json")); assert!(column_exists(&store, "ssh_revocations", "provenance_json")); drop(store); let _ = std::fs::remove_file(path); } #[test] fn file_store_uses_documented_durability_pragmas() { let path = std::env::temp_dir().join(format!("geth-store-durability-{}.sqlite", now_ms())); let store = Store::open(&path).expect("open file store"); assert_eq!( store.journal_mode().expect("journal mode"), FILE_JOURNAL_MODE ); assert_eq!( store.synchronous_mode().expect("synchronous mode"), SYNCHRONOUS_MODE ); drop(store); let _ = std::fs::remove_file(&path); let _ = std::fs::remove_file(path.with_extension("sqlite-wal")); let _ = std::fs::remove_file(path.with_extension("sqlite-shm")); } fn column_exists(store: &Store, table: &str, column: &str) -> bool { let mut stmt = store .conn .prepare(&format!("PRAGMA table_info({table})")) .expect("table info"); let columns = stmt .query_map([], |row| row.get::<_, String>(1)) .expect("columns") .collect::, _>>() .expect("column names"); columns.iter().any(|name| name == column) } #[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"); store .put_module_state(&StoredModuleState { module: "live-sync-status:node:laptop:ssh-certs".to_owned(), state_json: r#"{"last_attempt_ms":44}"#.to_owned(), updated_at_ms: 44, }) .expect("put status state"); assert_eq!( store .get_module_state("live-sync:node:laptop:ssh-certs") .expect("get state"), Some(state) ); assert_eq!( store .list_module_states_with_prefix("live-sync-status:node:laptop:") .expect("list prefix") .len(), 1 ); store .delete_module_state("live-sync:node:laptop:ssh-certs") .expect("delete state"); assert_eq!( store .get_module_state("live-sync:node:laptop:ssh-certs") .expect("get deleted state"), None ); } #[test] fn node_enrollment_requests_roundtrip() { let store = Store::open_memory().expect("open"); let request = StoredNodeEnrollmentRequest { request_id: "node-enrollment:1".to_owned(), requester_node: "node:laptop".to_owned(), request_json: r#"{"id":"node-enrollment:1"}"#.to_owned(), status: "pending".to_owned(), created_at_ms: 7, }; store .insert_node_enrollment_request(&request) .expect("insert enrollment request"); assert_eq!( store .get_node_enrollment_request("node-enrollment:1") .expect("get enrollment request"), Some(request.clone()) ); assert_eq!( store .list_node_enrollment_requests() .expect("list enrollment requests"), vec![request] ); } #[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 auth_signatures_roundtrip() { let store = Store::open_memory().expect("open"); let signature = StoredAuthSignature { op_id: "op:auth:1".to_owned(), signer: "ssh:blake3:admin".to_owned(), signer_public_key: "ssh-ed25519 AAAAADMIN eric@geth".to_owned(), namespace: "geth.auth-op.v1@geth.local".to_owned(), signature: b"-----BEGIN SSH SIGNATURE-----".to_vec(), created_at_ms: 3, }; store .insert_auth_signature(&signature) .expect("insert auth signature"); assert_eq!( store.list_auth_signatures().expect("list auth signatures"), vec![signature] ); } #[test] fn auth_op_with_signatures_rolls_back_when_signature_insert_fails() { let store = Store::open_memory().expect("open"); let op = StoredAuthOp { op_id: "auth-op:atomic".to_owned(), resource_id: "resource:test".to_owned(), op_json: "{}".to_owned(), created_at_ms: 1, }; let signature = StoredAuthSignature { op_id: op.op_id.clone(), signer: "key:admin".to_owned(), signer_public_key: "ssh-ed25519 AAAA test".to_owned(), namespace: "geth.auth.v1@geth.local".to_owned(), signature: b"sig".to_vec(), created_at_ms: 2, }; store .conn .execute("DROP TABLE auth_signatures", []) .expect("drop signatures table"); assert!( store .insert_auth_op_with_signatures(&op, &[signature]) .is_err() ); let count: i64 = store .conn .query_row( "SELECT COUNT(*) FROM auth_ops WHERE op_id = ?1", [&op.op_id], |row| row.get(0), ) .expect("count auth ops"); assert_eq!(count, 0); } #[test] fn auth_ops_with_signatures_batch_commits_multiple_ops() { let store = Store::open_memory().expect("open"); let first = StoredAuthOp { op_id: "auth-op:batch:1".to_owned(), resource_id: "resource:test".to_owned(), op_json: r#"{"id":"auth-op:batch:1"}"#.to_owned(), created_at_ms: 1, }; let second = StoredAuthOp { op_id: "auth-op:batch:2".to_owned(), resource_id: "resource:test".to_owned(), op_json: r#"{"id":"auth-op:batch:2"}"#.to_owned(), created_at_ms: 2, }; let signature = |op: &StoredAuthOp| StoredAuthSignature { op_id: op.op_id.clone(), signer: "key:admin".to_owned(), signer_public_key: "ssh-ed25519 AAAA test".to_owned(), namespace: "geth.auth.v1@geth.local".to_owned(), signature: b"sig".to_vec(), created_at_ms: op.created_at_ms, }; store .insert_auth_ops_with_signatures(&[ (first.clone(), vec![signature(&first)]), (second.clone(), vec![signature(&second)]), ]) .expect("batch insert auth ops"); assert_eq!( store .list_auth_ops_for_resource("resource:test") .expect("list auth ops"), vec![first, second] ); assert_eq!( store .list_auth_signatures() .expect("list auth signatures") .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_ops_with_signatures_batch_commits_multiple_ops() { let store = Store::open_memory().expect("open"); let first = StoredKeychainOp { op_id: "keychain-op:batch:1".to_owned(), op_json: r#"{"id":"keychain-op:batch:1"}"#.to_owned(), created_at_ms: 1, }; let second = StoredKeychainOp { op_id: "keychain-op:batch:2".to_owned(), op_json: r#"{"id":"keychain-op:batch:2"}"#.to_owned(), created_at_ms: 2, }; let signature = |op: &StoredKeychainOp| StoredKeychainSignature { op_id: op.op_id.clone(), signer: "key:admin".to_owned(), signer_public_key: "ssh-ed25519 AAAA test".to_owned(), namespace: "geth.keychain.v1@geth.local".to_owned(), signature: b"sig".to_vec(), created_at_ms: op.created_at_ms, }; store .insert_keychain_ops_with_signatures(&[ (first.clone(), vec![signature(&first)]), (second.clone(), vec![signature(&second)]), ]) .expect("batch insert keychain ops"); assert_eq!( store.list_keychain_ops().expect("list keychain ops"), vec![first, second] ); assert_eq!( store .list_keychain_signatures() .expect("list keychain signatures") .len(), 2 ); } #[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!["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] ); } }