solo-storage 0.10.0

// SPDX-License-Identifier: Apache-2.0

//! Tenant registry for v0.8.0 multi-tenancy.
//!
//! This module owns the in-process abstraction for `<data_dir>/tenants_index.db`
//! — the SQLCipher-encrypted registry that tracks every tenant present in a
//! Solo data dir, their on-disk filenames, lifecycle status, and admin audit
//! trail.
//!
//! See `docs/dev-log/0090-v0.8.0-implementation-plan.md` §2 Priority 1 for
//! the design rationale; ADR-0004 (added in v0.8.0 P7) carries the canonical
//! statement of the per-tenant isolation invariants.
//!
//! ## Layout
//!
//!   * [`TenantId`]           — re-exported from `solo-core::types`. Newtype
//!     with validation rules (lowercase a-z + 0-9 + `-`/`_`, ≤ 64 bytes).
//!   * [`TenantStatus`]       — lifecycle enum mirrored on the SQL
//!     `status` column's CHECK domain.
//!   * [`TenantRecord`]       — one row from the `tenants` table, in
//!     Rust form.
//!   * [`TenantsIndex`]       — handle to the open SQLCipher connection,
//!     CRUD methods.
//!   * [`migrate`] (sibling)  — the v0.7.1 → v0.8.0 mass-data-move
//!     helper. Re-exported as `solo_storage::tenants::migrate_v071_to_v080`.
//!
//! Per-tenant `TenantHandle` (writer + readers + HNSW + audit) lands in
//! v0.8.0 P2 — it lives next to this module but is intentionally NOT
//! introduced in P1 to keep the schema-foundation commit reviewable in
//! isolation.

pub mod handle;
pub mod migrate;
pub mod registry;
#[cfg(test)]
mod handle_registry_tests;

pub use handle::{TenantHandle, TenantOpenParams};
pub use migrate::migrate_v071_to_v080;
pub use registry::{TenantRegistry, TenantRegistryParams};

use rusqlite::{Connection, OptionalExtension, params};
use solo_core::{Error, Result, TenantId};
use std::path::Path;

use crate::init::open_sqlcipher;
use crate::key_material::KeyMaterial;
use crate::migration;

/// File name of the per-data-dir tenant registry, relative to the data dir.
pub const TENANTS_INDEX_FILENAME: &str = "tenants_index.db";

/// Subdirectory holding per-tenant DB files, relative to the data dir.
pub const TENANTS_SUBDIR: &str = "tenants";

/// Lifecycle status of a tenant in the registry.
///
/// `Active` is the steady state. `PendingMigration` is set transiently
/// during the v0.7.1 → v0.8.0 mass-data-move helper between the moment the
/// registry row is written and the moment all the data files have been
/// relocated; if the daemon crashes in that window, the next boot
/// observes this status and resumes the move. `PendingDelete` is set
/// transiently during `solo tenants delete` (P6) for the same reason.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TenantStatus {
    Active,
    PendingMigration,
    PendingDelete,
}

impl TenantStatus {
    /// SQL-compatible string form (matches the CHECK domain on
    /// `tenants.status`).
    pub fn as_sql_str(&self) -> &'static str {
        match self {
            Self::Active => "active",
            Self::PendingMigration => "pending_migration",
            Self::PendingDelete => "pending_delete",
        }
    }

    fn parse(s: &str) -> Result<Self> {
        match s {
            "active" => Ok(Self::Active),
            "pending_migration" => Ok(Self::PendingMigration),
            "pending_delete" => Ok(Self::PendingDelete),
            other => Err(Error::storage(format!(
                "unknown tenant status from registry: {other:?}"
            ))),
        }
    }
}

/// One tenant's metadata as stored in tenants_index.db.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct TenantRecord {
    pub tenant_id: TenantId,
    pub db_filename: String,
    pub display_name: Option<String>,
    pub created_at_ms: i64,
    pub status: TenantStatus,
    /// v0.8.1 P3: per-tenant byte quota. `None` means unlimited (the
    /// default for tenants migrated from v0.8.0 and for tenants created
    /// without `--quota-bytes`). When `Some(q)`, the writer-actor
    /// rejects writes that would push `tenants/<id>.db` size above `q`.
    #[serde(default)]
    pub quota_bytes: Option<u64>,
    /// v0.9.0 P1: epoch ms of the last time this tenant was opened (cache
    /// miss OR cache hit) via `TenantRegistry::get_or_open`. `None` means
    /// the tenant has not been opened since migration 0009 ran (or has
    /// not been opened at all). Closes the v0.8.0 doc-vs-code gap
    /// (lesson #39 second incident).
    #[serde(default)]
    pub last_accessed_ms: Option<i64>,
}

/// Handle to the tenants_index.db registry.
///
/// Holds a single open SQLCipher connection. Not concurrency-safe by itself
/// — wrap in `Mutex` / `Arc<Mutex<...>>` at the caller layer if multiple
/// threads need to mutate the registry. The TenantHandle abstraction in P2
/// will own the canonical wrapper; for P1 the registry is only touched by
/// `init` and the v0.7.1 → v0.8.0 migration helper, both single-threaded
/// at startup.
pub struct TenantsIndex {
    conn: Connection,
}

impl TenantsIndex {
    /// Open (or create) the tenants_index.db file at `<data_dir>/tenants_index.db`.
    ///
    /// Applies migration 0004 if needed. Idempotent — calling twice on the
    /// same data dir is fine; the second call just sees the schema is
    /// already up to date.
    ///
    /// The data dir itself must already exist (created by the caller, since
    /// `init::init()` owns the data-dir creation contract).
    pub fn open(data_dir: &Path, key: &KeyMaterial) -> Result<Self> {
        let path = data_dir.join(TENANTS_INDEX_FILENAME);
        let mut conn = open_sqlcipher(&path, key)?;
        migration::run_tenants_index_migrations(&mut conn)?;
        Ok(Self { conn })
    }

    /// Register a new tenant with no byte quota (unlimited).
    ///
    /// Inserts a row with `status='active'` and `quota_bytes=NULL`.
    /// Errors with [`Error::Conflict`] if `tenant_id` is already present.
    ///
    /// Use [`Self::register_with_status`] when the caller needs to set a
    /// transient status (e.g., `PendingMigration` from the mass-data-move
    /// helper before the files have been moved into place). Use
    /// [`Self::register_with_quota`] to set an initial quota on a fresh
    /// tenant (v0.8.1 P3).
    pub fn register(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
    ) -> Result<()> {
        self.register_with_quota_and_status(
            tenant_id,
            db_filename,
            display_name,
            None,
            TenantStatus::Active,
        )
    }

    /// Variant of [`Self::register`] that sets a specific initial status
    /// (used by the mass-data-move helper to register tenants in
    /// `PendingMigration`).
    pub fn register_with_status(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
        status: TenantStatus,
    ) -> Result<()> {
        self.register_with_quota_and_status(
            tenant_id,
            db_filename,
            display_name,
            None,
            status,
        )
    }

    /// v0.8.1 P3: variant of [`Self::register`] that also persists an
    /// initial byte quota. `quota_bytes = None` means unlimited (matches
    /// the v0.8.0 behavior); `Some(q)` enables enforcement on the
    /// writer-actor side.
    pub fn register_with_quota(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
        quota_bytes: Option<u64>,
    ) -> Result<()> {
        self.register_with_quota_and_status(
            tenant_id,
            db_filename,
            display_name,
            quota_bytes,
            TenantStatus::Active,
        )
    }

    /// Full INSERT path. Internal — every other `register*` method
    /// delegates here.
    fn register_with_quota_and_status(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
        quota_bytes: Option<u64>,
        status: TenantStatus,
    ) -> Result<()> {
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        // SQLite bind for `Option<u64>`: rusqlite serialises `u64` as
        // INTEGER (signed i64). Quotas above `i64::MAX` clamp; for human-
        // scale Solo deployments that's not a concern, but we cast
        // explicitly so a future change to a u64-clean SQL backend
        // doesn't silently truncate.
        let quota_i64: Option<i64> = quota_bytes.map(|q| q.min(i64::MAX as u64) as i64);
        let res = self.conn.execute(
            "INSERT INTO tenants (tenant_id, db_filename, display_name, created_at_ms, status, quota_bytes)
             VALUES (?, ?, ?, ?, ?, ?)",
            params![
                tenant_id.as_str(),
                db_filename,
                display_name,
                now_ms,
                status.as_sql_str(),
                quota_i64,
            ],
        );
        match res {
            Ok(_) => Ok(()),
            Err(rusqlite::Error::SqliteFailure(err, msg))
                if err.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_PRIMARYKEY
                    || err.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE =>
            {
                Err(Error::conflict(format!(
                    "tenant already exists: {} ({})",
                    tenant_id,
                    msg.as_deref().unwrap_or("UNIQUE/PK violation")
                )))
            }
            Err(e) => Err(Error::storage(format!("register tenant {tenant_id}: {e}"))),
        }
    }

    /// Update an existing tenant's status. No-ops if the row is absent —
    /// the caller should call [`Self::lookup`] first if presence matters.
    pub fn set_status(
        &mut self,
        tenant_id: &TenantId,
        status: TenantStatus,
    ) -> Result<()> {
        self.conn
            .execute(
                "UPDATE tenants SET status = ? WHERE tenant_id = ?",
                params![status.as_sql_str(), tenant_id.as_str()],
            )
            .map_err(|e| Error::storage(format!("set_status({tenant_id}): {e}")))?;
        Ok(())
    }

    /// v0.8.1 P3: update a tenant's byte quota. `None` clears the quota
    /// (unlimited). `Some(0)` is normalized to `None` by the CLI's
    /// `--unlimited` flag but persisted as-is here — the caller is
    /// responsible for the normalisation. Returns the number of rows
    /// affected (0 if the tenant id wasn't present).
    ///
    /// Refusing-to-shrink-below-current-size is enforced at the CLI
    /// layer, not here — keep storage-layer setters total so the CLI
    /// has flexibility around `--confirm`-gated overrides.
    pub fn set_quota(
        &mut self,
        tenant_id: &TenantId,
        quota_bytes: Option<u64>,
    ) -> Result<usize> {
        let quota_i64: Option<i64> = quota_bytes.map(|q| q.min(i64::MAX as u64) as i64);
        let updated = self
            .conn
            .execute(
                "UPDATE tenants SET quota_bytes = ? WHERE tenant_id = ?",
                params![quota_i64, tenant_id.as_str()],
            )
            .map_err(|e| Error::storage(format!("set_quota({tenant_id}): {e}")))?;
        Ok(updated)
    }

    /// List every tenant, ordered by `created_at_ms` (oldest first).
    pub fn list(&self) -> Result<Vec<TenantRecord>> {
        let mut stmt = self
            .conn
            .prepare(
                "SELECT tenant_id, db_filename, display_name, created_at_ms, status, quota_bytes, last_accessed
                 FROM tenants
                 ORDER BY created_at_ms ASC, tenant_id ASC",
            )
            .map_err(|e| Error::storage(format!("prepare list tenants: {e}")))?;
        let rows = stmt
            .query_map([], row_to_tenant_record)
            .map_err(|e| Error::storage(format!("query list tenants: {e}")))?;
        let mut out = Vec::new();
        for r in rows {
            out.push(r.map_err(|e| Error::storage(format!("scan tenant row: {e}")))?);
        }
        Ok(out)
    }

    /// Lookup a single tenant by id. Returns `Ok(None)` if absent.
    pub fn lookup(&self, tenant_id: &TenantId) -> Result<Option<TenantRecord>> {
        self.conn
            .query_row(
                "SELECT tenant_id, db_filename, display_name, created_at_ms, status, quota_bytes, last_accessed
                 FROM tenants WHERE tenant_id = ?",
                params![tenant_id.as_str()],
                row_to_tenant_record,
            )
            .optional()
            .map_err(|e| Error::storage(format!("lookup tenant {tenant_id}: {e}")))
    }

    /// v0.9.0 P1: stamp `last_accessed = now_ms` on the tenant row. Called
    /// by `TenantRegistry::get_or_open` on every cache miss AND cache hit.
    /// No-op if the tenant id is absent (returns Ok(0)) — matches the
    /// `set_quota` shape; the caller decides whether absence is an error.
    ///
    /// Single-statement UPDATE. The lookup is by primary key, and the
    /// write is bounded — fits comfortably inside the existing
    /// `RegistryDeps.index: Mutex<TenantsIndex>` serialization.
    pub fn touch_last_accessed(
        &mut self,
        tenant_id: &TenantId,
        now_ms: i64,
    ) -> Result<usize> {
        let updated = self
            .conn
            .execute(
                "UPDATE tenants SET last_accessed = ? WHERE tenant_id = ?",
                params![now_ms, tenant_id.as_str()],
            )
            .map_err(|e| {
                Error::storage(format!(
                    "touch_last_accessed({tenant_id}): {e}"
                ))
            })?;
        Ok(updated)
    }

    /// Remove a tenant row. Idempotent — removing a non-existent tenant is
    /// not an error. Called by the hard-delete sequence in P6 AFTER the
    /// on-disk DB file has been removed.
    pub fn remove(&mut self, tenant_id: &TenantId) -> Result<()> {
        self.conn
            .execute(
                "DELETE FROM tenants WHERE tenant_id = ?",
                params![tenant_id.as_str()],
            )
            .map_err(|e| Error::storage(format!("remove tenant {tenant_id}: {e}")))?;
        Ok(())
    }

    /// Borrow the underlying connection, e.g. for the admin-audit emitter
    /// in P4/P6 that needs to write to `audit_events_admin` in the same
    /// SQLCipher file. NOT part of the long-term public surface; we
    /// expose it now to keep the P1 → P2 handoff straightforward and will
    /// fold the audit write into a dedicated method when P4 lands.
    pub(crate) fn connection(&self) -> &Connection {
        &self.conn
    }

    /// Test-only constructor: wrap an existing SQLite connection (used
    /// by `TenantRegistry::for_tests_with_single_tenant` to build an
    /// in-memory index stub for transport-layer test harnesses).
    #[cfg(any(test, feature = "test-support"))]
    pub fn from_connection_for_tests(conn: Connection) -> Self {
        Self { conn }
    }
}

fn row_to_tenant_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<TenantRecord> {
    let tenant_id_str: String = row.get(0)?;
    let tenant_id = TenantId::new(tenant_id_str.clone()).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(
            0,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("invalid tenant_id in registry: {tenant_id_str}: {e}"),
            )),
        )
    })?;
    let db_filename: String = row.get(1)?;
    let display_name: Option<String> = row.get(2)?;
    let created_at_ms: i64 = row.get(3)?;
    let status_str: String = row.get(4)?;
    let status = TenantStatus::parse(&status_str).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(
            4,
            rusqlite::types::Type::Text,
            Box::new(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                format!("{e}"),
            )),
        )
    })?;
    // v0.8.1 P3: quota_bytes column added in migration 0008. Stored as
    // INTEGER (i64); we read as Option<i64> and convert to Option<u64>
    // (clamping negative-by-error to None so a manual SQL surgery
    // can't poison the read).
    let quota_bytes: Option<u64> = row
        .get::<_, Option<i64>>(5)?
        .and_then(|v| if v < 0 { None } else { Some(v as u64) });
    // v0.9.0 P1: last_accessed column added in migration 0009. Stored
    // as INTEGER ms-since-epoch (i64), nullable — NULL for tenants
    // that have not been opened since the migration ran.
    let last_accessed_ms: Option<i64> = row.get::<_, Option<i64>>(6)?;
    Ok(TenantRecord {
        tenant_id,
        db_filename,
        display_name,
        created_at_ms,
        status,
        quota_bytes,
        last_accessed_ms,
    })
}

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

    /// Argon2id is the slow path; use a cheap throwaway key for unit tests.
    /// We're testing registry semantics, not key derivation.
    fn fast_test_key() -> KeyMaterial {
        let salt = [7u8; 16];
        // Use the production KeyMaterial::derive — the cost is dwarfed by
        // SQLite open + first migration. Tests stay well under 1s each.
        KeyMaterial::derive("registry-test-passphrase", &salt)
            .expect("derive test key")
    }

    fn open_fresh(tmp: &TempDir) -> TenantsIndex {
        let key = fast_test_key();
        TenantsIndex::open(tmp.path(), &key).expect("open tenants_index")
    }

    #[test]
    fn open_creates_schema() {
        let tmp = TempDir::new().unwrap();
        let idx = open_fresh(&tmp);
        // Schema present: registry table exists.
        let n: i64 = idx
            .conn
            .query_row("SELECT COUNT(*) FROM tenants", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
        // Migration tracker row should have been inserted.
        let v: u32 = idx
            .conn
            .query_row(
                "SELECT MAX(version) FROM schema_migrations_tenants_index",
                [],
                |r| r.get(0),
            )
            .unwrap();
        // v0.9.0 P1 advanced the tenants_index chain to 9
        // (last_accessed, closing v0.8.0 doc-vs-code gap). Previously:
        // 8 (quota_bytes, v0.8.1 P3); 4 (initial registry, v0.8.0 P1).
        assert_eq!(v, 9);
    }

    #[test]
    fn register_then_lookup() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("acme").unwrap();
        idx.register(&t, "acme.db", Some("ACME Corp")).unwrap();

        let rec = idx.lookup(&t).unwrap().expect("tenant must be present");
        assert_eq!(rec.tenant_id, t);
        assert_eq!(rec.db_filename, "acme.db");
        assert_eq!(rec.display_name.as_deref(), Some("ACME Corp"));
        assert_eq!(rec.status, TenantStatus::Active);
    }

    #[test]
    fn register_duplicate_errors() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("dup").unwrap();
        idx.register(&t, "dup.db", None).unwrap();
        let err = idx
            .register(&t, "dup-other.db", None)
            .expect_err("duplicate tenant_id must error");
        assert!(
            matches!(err, Error::Conflict(_)),
            "expected Conflict, got {err:?}"
        );
    }

    #[test]
    fn list_returns_in_creation_order() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let ids = ["alpha", "beta", "gamma"];
        for id in ids {
            idx.register(
                &TenantId::new(id).unwrap(),
                &format!("{id}.db"),
                None,
            )
            .unwrap();
            // Ensure created_at_ms diverges so the ASC sort is testable.
            std::thread::sleep(std::time::Duration::from_millis(2));
        }
        let listed = idx.list().unwrap();
        assert_eq!(listed.len(), 3);
        let listed_ids: Vec<&str> =
            listed.iter().map(|r| r.tenant_id.as_str()).collect();
        assert_eq!(listed_ids, ["alpha", "beta", "gamma"]);
    }

    #[test]
    fn set_status_persists() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("statushost").unwrap();
        idx.register(&t, "statushost.db", None).unwrap();

        idx.set_status(&t, TenantStatus::PendingMigration).unwrap();
        let rec = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(rec.status, TenantStatus::PendingMigration);

        idx.set_status(&t, TenantStatus::PendingDelete).unwrap();
        let rec = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(rec.status, TenantStatus::PendingDelete);
    }

    #[test]
    fn remove_idempotent_on_missing() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let absent = TenantId::new("ghost").unwrap();
        // Removing a tenant that never existed is a no-op, not an error.
        idx.remove(&absent).expect("idempotent remove");
        // Still empty.
        assert_eq!(idx.list().unwrap().len(), 0);
    }

    // ---- v0.8.1 P3: quota_bytes wiring ----

    #[test]
    fn register_default_persists_no_quota() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("noquota").unwrap();
        idx.register(&t, "noquota.db", None).unwrap();
        let rec = idx.lookup(&t).unwrap().expect("registered");
        assert_eq!(rec.quota_bytes, None);
    }

    #[test]
    fn register_with_quota_persists_value() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("limited").unwrap();
        idx.register_with_quota(&t, "limited.db", None, Some(1_048_576))
            .unwrap();
        let rec = idx.lookup(&t).unwrap().expect("registered");
        assert_eq!(rec.quota_bytes, Some(1_048_576));
    }

    #[test]
    fn set_quota_updates_and_clears() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("quotachange").unwrap();
        idx.register_with_quota(&t, "qc.db", None, Some(1_000_000)).unwrap();
        // Bump up.
        let n = idx.set_quota(&t, Some(5_000_000)).unwrap();
        assert_eq!(n, 1);
        let rec = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(rec.quota_bytes, Some(5_000_000));
        // Clear.
        let n2 = idx.set_quota(&t, None).unwrap();
        assert_eq!(n2, 1);
        let rec2 = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(rec2.quota_bytes, None);
    }

    #[test]
    fn set_quota_on_missing_tenant_returns_zero_rows() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let n = idx
            .set_quota(&TenantId::new("ghost").unwrap(), Some(123))
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn list_surfaces_quota_bytes() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("listed").unwrap();
        idx.register_with_quota(&t, "listed.db", None, Some(42))
            .unwrap();
        let listed = idx.list().unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].quota_bytes, Some(42));
    }

    // ---- v0.9.0 P1: last_accessed wiring ----

    #[test]
    fn register_leaves_last_accessed_null_until_first_touch() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("fresh").unwrap();
        idx.register(&t, "fresh.db", None).unwrap();
        let rec = idx.lookup(&t).unwrap().expect("registered");
        // Newly-registered tenant has no access stamp until first
        // `touch_last_accessed` call — matches the v0.8.0 doc claim that
        // the value reflects real opens, not registrations.
        assert_eq!(rec.last_accessed_ms, None);
    }

    #[test]
    fn touch_last_accessed_persists_ms() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("touched").unwrap();
        idx.register(&t, "touched.db", None).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let n = idx.touch_last_accessed(&t, now_ms).unwrap();
        assert_eq!(n, 1, "expected exactly one row updated");
        let rec = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(rec.last_accessed_ms, Some(now_ms));
    }

    #[test]
    fn touch_last_accessed_overwrites_prior_stamp() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("rebumped").unwrap();
        idx.register(&t, "rebumped.db", None).unwrap();
        let first_ms: i64 = 1_000_000_000_000;
        idx.touch_last_accessed(&t, first_ms).unwrap();
        let second_ms: i64 = 2_000_000_000_000;
        idx.touch_last_accessed(&t, second_ms).unwrap();
        let rec = idx.lookup(&t).unwrap().unwrap();
        assert_eq!(
            rec.last_accessed_ms,
            Some(second_ms),
            "second touch must overwrite first"
        );
    }

    #[test]
    fn touch_last_accessed_on_missing_tenant_returns_zero_rows() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let n = idx
            .touch_last_accessed(&TenantId::new("ghost").unwrap(), 123)
            .unwrap();
        assert_eq!(n, 0, "missing tenant must not error and must report 0 rows");
    }

    #[test]
    fn list_surfaces_last_accessed_ms() {
        let tmp = TempDir::new().unwrap();
        let mut idx = open_fresh(&tmp);
        let t = TenantId::new("listed-stamp").unwrap();
        idx.register(&t, "ls.db", None).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        idx.touch_last_accessed(&t, now_ms).unwrap();
        let listed = idx.list().unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].last_accessed_ms, Some(now_ms));
    }
}