kestrove-core/crates/kv-store/src/lib.rs

//! Local encrypted replica (milestone M1, kestrove-spec/docs/07).
//!
//! [`SqliteStore`] is native only; on wasm32 the web vault (`kv-wasm`) implements [`Store`] over IndexedDB.
//!
//! The store holds the same `Envelope` bytes the server holds, plus sync
//! bookkeeping. It never sees plaintext. Schema (rusqlite, bundled):
//!
//! ```sql
//! CREATE TABLE meta    (key TEXT PRIMARY KEY, value BLOB NOT NULL);
//! CREATE TABLE vault   (vault_id BLOB PRIMARY KEY, sealed_key TEXT, meta TEXT, cursor INTEGER,
//!                       manifest_version INTEGER, manifest_hash BLOB, integrity_alarm INTEGER);
//! CREATE TABLE item    (vault_id BLOB, item_id BLOB, revision INTEGER, deleted INTEGER,
//!                       wrapped_item_key TEXT, payload TEXT, updated_by BLOB, updated_at TEXT,
//!                       state TEXT CHECK(state IN ('clean','pending')),
//!                       base_revision INTEGER, base_deleted INTEGER, base_wrapped_item_key TEXT, base_payload TEXT,
//!                       PRIMARY KEY (vault_id, item_id));
//! CREATE TABLE history (seq INTEGER PRIMARY KEY, vault_id BLOB, item_id BLOB, revision INTEGER,
//!                       wrapped_item_key TEXT, payload TEXT, saved_at INTEGER);
//! ```
//!
//! Envelopes are stored as their JSON form (the same bytes the server stores).
//! The `base_*` columns hold the last version of the item the server confirmed;
//! they are the common ancestor for three-way merges and the input to the manifest's Merkle root.

use std::{
    collections::{BTreeMap, HashMap},
    sync::Mutex,
};

use kv_proto::{Envelope, ItemRecord};
use uuid::Uuid;

#[cfg(not(target_arch = "wasm32"))]
mod sqlite;
#[cfg(not(target_arch = "wasm32"))]
pub use sqlite::SqliteStore;

/// Versions of an item kept in history.
pub const HISTORY_LIMIT: usize = 10;

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("storage I/O: {0}")]
    Io(String),
    #[error("not found")]
    NotFound,
}

impl From<serde_json::Error> for StoreError {
    fn from(e: serde_json::Error) -> Self {
        StoreError::Io(format!("corrupt envelope: {e}"))
    }
}

pub type Result<T> = std::result::Result<T, StoreError>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncState {
    /// Matches the server's version (`base`).
    Clean,
    /// Edited locally; not yet accepted by the server. The edit was made against `base`.
    Pending,
}

/// One item as replicated on this device.
#[derive(Debug, Clone)]
pub struct LocalItem {
    /// The working copy. For pending items `record.revision` equals the base revision.
    pub record: ItemRecord,
    pub state: SyncState,
    /// Last version confirmed by the server. `None` for items created here and never pushed.
    pub base: Option<ItemRecord>,
}

impl LocalItem {
    pub fn base_revision(&self) -> u64 {
        self.base.as_ref().map_or(0, |b| b.revision)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VaultRow {
    pub vault_id: Uuid,
    pub sealed_key: Envelope,
    pub meta: Option<Envelope>,
    /// Highest server revision pulled.
    pub cursor: u64,
    /// Highest manifest version seen and the blake3 hash of its body.
    pub manifest_version: u64,
    pub manifest_hash: Option<[u8; 32]>,
    /// Set on a manifest rollback or tampering; syncing this vault stops.
    pub integrity_alarm: bool,
}

impl VaultRow {
    pub fn new(vault_id: Uuid, sealed_key: Envelope, meta: Option<Envelope>) -> Self {
        Self { vault_id, sealed_key, meta, cursor: 0, manifest_version: 0, manifest_hash: None, integrity_alarm: false }
    }
}

/// An earlier version of an item (local edit or merge loser).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HistoryEntry {
    pub revision: u64,
    pub wrapped_item_key: Envelope,
    pub payload: Envelope,
    pub saved_at: i64,
}

/// Storage backend: [`SqliteStore`] on devices, [`MemoryStore`] in tests.
pub trait Store: Send + Sync {
    fn get_meta(&self, key: &str) -> Result<Option<Vec<u8>>>;
    fn put_meta(&self, key: &str, value: &[u8]) -> Result<()>;
    fn delete_meta(&self, key: &str) -> Result<()>;

    /// Inserts or replaces a vault row.
    fn put_vault(&self, vault: &VaultRow) -> Result<()>;
    fn vault(&self, vault: Uuid) -> Result<Option<VaultRow>>;
    fn vaults(&self) -> Result<Vec<VaultRow>>;

    /// Inserts or replaces an item.
    fn put(&self, vault: Uuid, item: &LocalItem) -> Result<()>;
    fn get(&self, vault: Uuid, item: Uuid) -> Result<Option<LocalItem>>;
    /// All items, including tombstones.
    fn list(&self, vault: Uuid) -> Result<Vec<LocalItem>>;
    fn remove(&self, vault: Uuid, item: Uuid) -> Result<()>;

    /// Appends a version and keeps only the newest [`HISTORY_LIMIT`].
    fn push_history(&self, vault: Uuid, item: Uuid, entry: &HistoryEntry) -> Result<()>;
    /// Newest first.
    fn history(&self, vault: Uuid, item: Uuid) -> Result<Vec<HistoryEntry>>;

    /// Deletes everything (sign-out / reset).
    fn wipe(&self) -> Result<()>;

    // ---- convenience ----

    fn cursor(&self, vault: Uuid) -> Result<u64> {
        Ok(self.vault(vault)?.ok_or(StoreError::NotFound)?.cursor)
    }

    fn set_cursor(&self, vault: Uuid, revision: u64) -> Result<()> {
        let mut v = self.vault(vault)?.ok_or(StoreError::NotFound)?;
        v.cursor = revision;
        self.put_vault(&v)
    }

    /// Stores a server-confirmed record as clean (it becomes its own base).
    fn put_remote(&self, vault: Uuid, record: &ItemRecord) -> Result<()> {
        self.put(vault, &LocalItem { record: record.clone(), state: SyncState::Clean, base: Some(record.clone()) })
    }

    /// Stores a local edit as pending against the current base.
    fn put_local(&self, vault: Uuid, record: &ItemRecord) -> Result<()> {
        let base = self.get(vault, record.item_id)?.and_then(|i| i.base);
        let mut record = record.clone();
        record.revision = base.as_ref().map_or(0, |b| b.revision);
        self.put(vault, &LocalItem { record, state: SyncState::Pending, base })
    }

    fn pending(&self, vault: Uuid) -> Result<Vec<LocalItem>> {
        Ok(self.list(vault)?.into_iter().filter(|i| i.state == SyncState::Pending).collect())
    }
}

// ---------------------------------------------------------------- in-memory

#[derive(Default)]
struct Mem {
    meta: HashMap<String, Vec<u8>>,
    vaults: Vec<VaultRow>,
    items: HashMap<Uuid, BTreeMap<Uuid, LocalItem>>,
    history: HashMap<(Uuid, Uuid), Vec<HistoryEntry>>,
}

/// Non-persistent store for tests and `VaultCore::new_in_memory`.
#[derive(Default)]
pub struct MemoryStore {
    inner: Mutex<Mem>,
}

impl MemoryStore {
    pub fn new() -> Self {
        Self::default()
    }

    fn mem(&self) -> std::sync::MutexGuard<'_, Mem> {
        self.inner.lock().unwrap_or_else(|p| p.into_inner())
    }
}

impl Store for MemoryStore {
    fn get_meta(&self, key: &str) -> Result<Option<Vec<u8>>> {
        Ok(self.mem().meta.get(key).cloned())
    }

    fn put_meta(&self, key: &str, value: &[u8]) -> Result<()> {
        self.mem().meta.insert(key.into(), value.to_vec());
        Ok(())
    }

    fn delete_meta(&self, key: &str) -> Result<()> {
        self.mem().meta.remove(key);
        Ok(())
    }

    fn put_vault(&self, vault: &VaultRow) -> Result<()> {
        let mut m = self.mem();
        match m.vaults.iter_mut().find(|v| v.vault_id == vault.vault_id) {
            Some(v) => *v = vault.clone(),
            None => m.vaults.push(vault.clone()),
        }
        Ok(())
    }

    fn vault(&self, vault: Uuid) -> Result<Option<VaultRow>> {
        Ok(self.mem().vaults.iter().find(|v| v.vault_id == vault).cloned())
    }

    fn vaults(&self) -> Result<Vec<VaultRow>> {
        Ok(self.mem().vaults.clone())
    }

    fn put(&self, vault: Uuid, item: &LocalItem) -> Result<()> {
        self.mem().items.entry(vault).or_default().insert(item.record.item_id, item.clone());
        Ok(())
    }

    fn get(&self, vault: Uuid, item: Uuid) -> Result<Option<LocalItem>> {
        Ok(self.mem().items.get(&vault).and_then(|v| v.get(&item)).cloned())
    }

    fn list(&self, vault: Uuid) -> Result<Vec<LocalItem>> {
        Ok(self.mem().items.get(&vault).map(|v| v.values().cloned().collect()).unwrap_or_default())
    }

    fn remove(&self, vault: Uuid, item: Uuid) -> Result<()> {
        if let Some(v) = self.mem().items.get_mut(&vault) {
            v.remove(&item);
        }
        Ok(())
    }

    fn push_history(&self, vault: Uuid, item: Uuid, entry: &HistoryEntry) -> Result<()> {
        let mut m = self.mem();
        let h = m.history.entry((vault, item)).or_default();
        h.insert(0, entry.clone());
        h.truncate(HISTORY_LIMIT);
        Ok(())
    }

    fn history(&self, vault: Uuid, item: Uuid) -> Result<Vec<HistoryEntry>> {
        Ok(self.mem().history.get(&(vault, item)).cloned().unwrap_or_default())
    }

    fn wipe(&self) -> Result<()> {
        *self.mem() = Mem::default();
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use kv_proto::{Alg, FORMAT_VERSION};

    use super::*;

    fn env(tag: u8) -> Envelope {
        Envelope {
            v: FORMAT_VERSION,
            alg: Alg::XChaCha20Poly1305,
            kid: Uuid::nil(),
            n: vec![tag; 24],
            ct: vec![tag; 20],
            enc: vec![],
        }
    }

    fn record(id: Uuid, rev: u64, tag: u8) -> ItemRecord {
        ItemRecord {
            item_id: id,
            revision: rev,
            deleted: false,
            wrapped_item_key: Some(env(tag)),
            payload: Some(env(tag + 1)),
            updated_by: Uuid::nil(),
            updated_at: "2026-01-01T00:00:00Z".into(),
        }
    }

    fn exercise(store: &dyn Store) {
        let vid = Uuid::new_v4();
        assert!(store.vault(vid).unwrap().is_none());
        store.put_vault(&VaultRow::new(vid, env(1), Some(env(2)))).unwrap();
        store.set_cursor(vid, 7).unwrap();
        let mut v = store.vault(vid).unwrap().unwrap();
        assert_eq!(v.cursor, 7);
        v.manifest_version = 3;
        v.manifest_hash = Some([5; 32]);
        store.put_vault(&v).unwrap();
        assert_eq!(store.vaults().unwrap(), vec![v]);

        store.put_meta("account", b"{}").unwrap();
        assert_eq!(store.get_meta("account").unwrap().as_deref(), Some(&b"{}"[..]));
        store.delete_meta("account").unwrap();
        assert!(store.get_meta("account").unwrap().is_none());

        // Remote record, then a local edit keeps the base for three-way merge.
        let id = Uuid::new_v4();
        store.put_remote(vid, &record(id, 4, 10)).unwrap();
        assert!(store.pending(vid).unwrap().is_empty());
        let mut edit = record(id, 999, 20);
        edit.updated_at = "later".into();
        store.put_local(vid, &edit).unwrap();
        let got = store.get(vid, id).unwrap().unwrap();
        assert_eq!(got.state, SyncState::Pending);
        assert_eq!(got.record.revision, 4);
        assert_eq!(got.record.payload, Some(env(21)));
        assert_eq!(got.base.as_ref().unwrap().payload, Some(env(11)));
        assert_eq!(got.base_revision(), 4);
        assert_eq!(store.pending(vid).unwrap().len(), 1);

        // A brand-new local item has no base.
        let new_id = Uuid::new_v4();
        store.put_local(vid, &record(new_id, 0, 30)).unwrap();
        assert!(store.get(vid, new_id).unwrap().unwrap().base.is_none());

        // Tombstones round-trip with no envelopes.
        let tomb = ItemRecord { deleted: true, wrapped_item_key: None, payload: None, ..record(id, 9, 0) };
        store.put_remote(vid, &tomb).unwrap();
        let got = store.get(vid, id).unwrap().unwrap();
        assert!(got.record.deleted && got.record.payload.is_none() && got.base.unwrap().deleted);
        assert_eq!(store.list(vid).unwrap().len(), 2);
        store.remove(vid, new_id).unwrap();
        assert_eq!(store.list(vid).unwrap().len(), 1);

        for i in 0..15u64 {
            store
                .push_history(
                    vid,
                    id,
                    &HistoryEntry { revision: i, wrapped_item_key: env(1), payload: env(2), saved_at: 0 },
                )
                .unwrap();
        }
        let h = store.history(vid, id).unwrap();
        assert_eq!(h.len(), HISTORY_LIMIT);
        assert_eq!(h[0].revision, 14);
        assert_eq!(h[9].revision, 5);
        assert!(store.history(vid, Uuid::new_v4()).unwrap().is_empty());

        store.wipe().unwrap();
        assert!(store.vaults().unwrap().is_empty());
        assert!(store.list(vid).unwrap().is_empty());
    }

    #[test]
    fn memory_store() {
        exercise(&MemoryStore::new());
    }

    #[test]
    fn sqlite_store() {
        exercise(&SqliteStore::open_in_memory().unwrap());
    }

    #[test]
    fn sqlite_persists_across_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("vault.db");
        let vid = Uuid::new_v4();
        let id = Uuid::new_v4();
        {
            let s = SqliteStore::open(&path).unwrap();
            s.put_vault(&VaultRow::new(vid, env(1), None)).unwrap();
            s.put_remote(vid, &record(id, 2, 3)).unwrap();
        }
        let s = SqliteStore::open(&path).unwrap();
        assert_eq!(s.vault(vid).unwrap().unwrap().sealed_key, env(1));
        assert_eq!(s.get(vid, id).unwrap().unwrap().record.revision, 2);
    }

    #[test]
    fn sqlite_never_stores_plaintext_columns() {
        // Guard against accidentally adding a plaintext column: only ciphertext/ids/bookkeeping exist.
        let s = SqliteStore::open_in_memory().unwrap();
        let conn = s.conn();
        let mut stmt = conn.prepare("SELECT name FROM pragma_table_info('item')").unwrap();
        let cols: Vec<String> = stmt.query_map([], |r| r.get(0)).unwrap().map(|c| c.unwrap()).collect();
        for c in &cols {
            assert!(!["title", "username", "password", "url", "notes"].contains(&c.as_str()), "{c}");
        }
    }
}