solo-storage 0.8.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,
}

/// 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.
    ///
    /// Inserts a row with `status='active'`. 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).
    pub fn register(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
    ) -> Result<()> {
        self.register_with_status(tenant_id, db_filename, display_name, TenantStatus::Active)
    }

    /// Variant of [`Self::register`] that sets a specific initial status.
    pub fn register_with_status(
        &mut self,
        tenant_id: &TenantId,
        db_filename: &str,
        display_name: Option<&str>,
        status: TenantStatus,
    ) -> Result<()> {
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let res = self.conn.execute(
            "INSERT INTO tenants (tenant_id, db_filename, display_name, created_at_ms, status)
             VALUES (?, ?, ?, ?, ?)",
            params![
                tenant_id.as_str(),
                db_filename,
                display_name,
                now_ms,
                status.as_sql_str(),
            ],
        );
        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(())
    }

    /// 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
                 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
                 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}")))
    }

    /// 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}"),
            )),
        )
    })?;
    Ok(TenantRecord {
        tenant_id,
        db_filename,
        display_name,
        created_at_ms,
        status,
    })
}

#[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();
        assert_eq!(v, 4);
    }

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