ckg-storage 1.1.3

//! DB lifecycle: open, schema-version check, rebuild, path-swap, debris sweep.
//!
//! Contains `Storage::open_at`, `open_unverified`, `open_inner`, plus all
//! the supporting free functions for the build-then-swap rebuild pattern and
//! `root_path` collision-detection (I2).

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use ckg_core::{Error, RepoId, Result};
use cozo::{DataValue, DbInstance, ScriptMutability};

use crate::schema::PER_REPO_DDL;

use super::meta::{read_meta_bool, stamp_meta_bool, stamp_needs_reindex};
use super::{Storage, map_err};

/// Cozo engine label for the `storage-new-rocksdb` feature.
pub(super) const ENGINE: &str = "newrocksdb";

/// Result of probing Meta for the recorded `root_path`.
///
/// **NI1 fix**: the original implementation used `.ok()?` which collapsed
/// every error class — Meta missing (legit fresh DB), Cozo IO failure,
/// transient lock — into `None`. Three explicit states let `open_inner`
/// distinguish "no row → stamp it" from "could not read → refuse open".
#[derive(Debug)]
pub(super) enum RootPathProbe {
    /// Meta has a `root_path` row with this value.
    Recorded(String),
    /// Query succeeded with zero rows — fresh DB or pre-I2 DB. Safe to stamp.
    NoRow,
    /// Query itself failed. Hard error: do NOT stamp.
    ReadFailed(String),
}

fn read_root_path(db: &DbInstance) -> RootPathProbe {
    match db.run_script(
        "?[v] := *Meta{key: \"root_path\", value: v}",
        BTreeMap::new(),
        ScriptMutability::Immutable,
    ) {
        Ok(rows) => match rows.rows.first().and_then(|r| r.first()) {
            Some(DataValue::Str(s)) => RootPathProbe::Recorded(s.to_string()),
            // No row at all — fresh DB or pre-I2 DB. Safe to stamp.
            None => RootPathProbe::NoRow,
            // Row exists but value is not a string (e.g. Null or Int from a
            // corrupted write). Do NOT treat this as NoRow — that would
            // silently overwrite corrupted data and mask the problem.
            Some(other) => RootPathProbe::ReadFailed(format!(
                "non-string root_path in Meta: {:?}",
                other
            )),
        },
        Err(e) => RootPathProbe::ReadFailed(e.to_string()),
    }
}

/// Write the `schema_version` row into `Meta` via parameter binding so a
/// non-numeric / quote-bearing version string cannot escape the literal.
fn stamp_schema_version(db: &DbInstance, schema_version: &str) -> Result<()> {
    let mut params = BTreeMap::new();
    params.insert("v".into(), DataValue::from(schema_version));
    db.run_script(
        "?[key, value] <- [[\"schema_version\", $v]] :put Meta {key => value}",
        params,
        ScriptMutability::Mutable,
    )
    .map_err(map_err)?;
    Ok(())
}

/// Stamp the canonical `root_path` for this repo into `Meta` (I2).
fn stamp_root_path(db: &DbInstance, root_path: &str) -> Result<()> {
    let mut params = BTreeMap::new();
    params.insert("v".into(), DataValue::from(root_path));
    db.run_script(
        "?[key, value] <- [[\"root_path\", $v]] :put Meta {key => value}",
        params,
        ScriptMutability::Mutable,
    )
    .map_err(map_err)?;
    Ok(())
}

/// CR-M-7: returns true iff two path-shaped strings resolve to the same
/// absolute path after `Path::canonicalize`. Used to tolerate
/// symlink-equivalent recorded vs caller paths (macOS `/var ↔
/// /private/var`, Linux bind mounts). Conservative: if EITHER side
/// fails to canonicalize (path no longer exists, broken symlink, etc.)
/// returns false — we can't prove equivalence, so the caller's
/// collision-detection error fires (matches pre-M-7 fail-closed
/// behavior). True hash collisions between two distinct repos
/// canonicalize to different absolutes and also return false.
fn paths_canonicalize_equal(a: &str, b: &str) -> bool {
    // M10: cache the canonicalize result — each is a syscall.
    let Ok(pa) = std::path::Path::new(a).canonicalize() else {
        return false;
    };
    let Ok(pb) = std::path::Path::new(b).canonicalize() else {
        return false;
    };
    pa == pb
}

/// CR-review-#3: returns the canonical absolute string for `path` if
/// canonicalize succeeds, else falls back to the input. Used by M-7
/// auto-migrate to stamp a STABLE shape regardless of which alias the
/// caller passed — without this, two callers passing different
/// symlink shapes would flap the recorded value back and forth on
/// every alternating open.
fn canonical_or_input(path: &str) -> String {
    std::path::Path::new(path)
        .canonicalize()
        .ok()
        .map(|p| p.display().to_string())
        .unwrap_or_else(|| path.to_string())
}

/// Run a script, swallowing "relation already exists" errors so DDL is
/// idempotent across re-opens.
pub(super) fn run_idempotent(db: &DbInstance, script: &str) -> Result<()> {
    match db.run_script(script, BTreeMap::new(), ScriptMutability::Mutable) {
        Ok(_) => Ok(()),
        Err(e) => {
            let msg = e.to_string();
            if msg.contains("conflicts with an existing one")
                || msg.contains("already exists")
                || msg.contains("EvalRelationConflict")
            {
                Ok(())
            } else {
                Err(map_err(e))
            }
        }
    }
}

/// Schema-mismatch rebuild using a build-then-swap pattern:
///   1. Build a fresh DB at `<path>.new` (DDL + schema-stamp). On failure, old DB untouched.
///   2. Drop the new handle. Sweep stale `<path>.old`.
///   3. `rename(path → path.old)` + `rename(path.new → path)` — two atomic renames.
///   4. Reopen at `path`. Cleanup `path.old`.
fn rebuild_at_path(path: &Path, schema_version: &str) -> Result<DbInstance> {
    let parent = path.parent().ok_or_else(|| {
        Error::Storage(format!("rebuild target has no parent: {}", path.display()))
    })?;
    let file = path.file_name().and_then(|s| s.to_str()).ok_or_else(|| {
        Error::Storage(format!(
            "rebuild target has no file name: {}",
            path.display()
        ))
    })?;
    let new_path = parent.join(format!("{file}.new"));
    let old_path = parent.join(format!("{file}.old"));

    // Sweep any debris from a prior crashed rebuild before claiming names.
    sweep_rebuild_debris(path);

    // 1. Build the fresh DB at <path>.new. Original is untouched on error.
    std::fs::create_dir_all(&new_path)?;
    let new_db = DbInstance::new(ENGINE, &new_path, "{}").map_err(map_err)?;
    for stmt in PER_REPO_DDL {
        run_idempotent(&new_db, stmt)?;
    }
    stamp_schema_version(&new_db, schema_version)?;
    // Drop the handle so the rename can move the directory cleanly.
    drop(new_db);

    // 2. Swap directories (atomic on Linux, two-rename on others).
    swap_directories(path, &new_path, &old_path)?;

    // 3. Reopen at the canonical path.
    let installed = DbInstance::new(ENGINE, path, "{}").map_err(map_err)?;

    // 4. Cleanup the old DB — best-effort, log on failure.
    if let Err(cleanup_err) = std::fs::remove_dir_all(&old_path) {
        tracing::error!(
            "rebuild succeeded but old-DB cleanup failed at {}: {cleanup_err}; \
             will be swept on next rebuild",
            old_path.display()
        );
    }

    Ok(installed)
}

/// Linux: atomically swap `<path>` and `<new_path>` via
/// `renameat2(RENAME_EXCHANGE)`, then rename the displaced original to
/// `<old_path>` for cleanup. No missing-path window for concurrent readers.
///
/// Other platforms: fall back to two `std::fs::rename` calls. Brief window
/// where `<path>` doesn't exist (~one syscall); concurrent readers get ENOENT.
#[cfg(target_os = "linux")]
fn swap_directories(path: &Path, new_path: &Path, old_path: &Path) -> Result<()> {
    use std::ffi::CString;
    use std::os::unix::ffi::OsStrExt;

    let c_path = CString::new(path.as_os_str().as_bytes())
        .map_err(|e| Error::wrap("rebuild path has interior NUL", e))?;
    let c_new = CString::new(new_path.as_os_str().as_bytes())
        .map_err(|e| Error::wrap("rebuild path has interior NUL", e))?;
    // SAFETY: passing C strings owned by `c_path` / `c_new`; AT_FDCWD matches
    // std::fs semantics. Syscall returns 0 on success, -1 on error.
    let rc = unsafe {
        libc::renameat2(
            libc::AT_FDCWD,
            c_path.as_ptr(),
            libc::AT_FDCWD,
            c_new.as_ptr(),
            libc::RENAME_EXCHANGE,
        )
    };
    if rc != 0 {
        // Fallback: kernel < 3.15 or filesystem doesn't support RENAME_EXCHANGE.
        let errno = std::io::Error::last_os_error();
        tracing::warn!(
            "renameat2(RENAME_EXCHANGE) failed ({errno}); falling back to two-rename swap"
        );
        return swap_directories_two_rename(path, new_path, old_path);
    }
    // After EXCHANGE: `<path>` holds the new DB; `<new_path>` holds the old.
    if let Err(rename_err) = std::fs::rename(new_path, old_path) {
        tracing::error!(
            "renameat2 swap succeeded but old-DB rename {} → {} failed: {rename_err}",
            new_path.display(),
            old_path.display()
        );
    }
    Ok(())
}

#[cfg(not(target_os = "linux"))]
fn swap_directories(path: &Path, new_path: &Path, old_path: &Path) -> Result<()> {
    swap_directories_two_rename(path, new_path, old_path)
}

/// Portable fallback: rename original out of the way, then move new in.
/// Brief missing-path window between the two renames.
fn swap_directories_two_rename(path: &Path, new_path: &Path, old_path: &Path) -> Result<()> {
    // M7: absolute paths only — relative paths with rename() are CWD-dependent.
    debug_assert!(path.is_absolute(), "swap_directories: path must be absolute: {path:?}");
    std::fs::rename(path, old_path)?;
    if let Err(swap_err) = std::fs::rename(new_path, path) {
        // Restore the original so the user isn't worse off.
        if let Err(rb) = std::fs::rename(old_path, path) {
            // CRITICAL: original is now orphaned at <old_path> and the
            // canonical path is empty. Surface a distinct hard error so
            // an operator can `mv <old_path> <path>` manually.
            tracing::error!(
                "rebuild swap failed AND rollback failed: could not restore {} → {}: {rb}",
                old_path.display(),
                path.display()
            );
            // STORAGE-C1: remove the orphaned .new directory so
            // `sweep_rebuild_debris` on next open doesn't delete healthy
            // DDL that an operator might want to inspect.
            let _ = std::fs::remove_dir_all(new_path);
            return Err(Error::Storage(format!(
                "DB CORRUPTED: rebuild swap failed (swap_err: {swap_err}) and rollback also \
                 failed (rb: {rb}). Original DB is at {} (manual recovery: \
                 `mv {} {}`).",
                old_path.display(),
                old_path.display(),
                path.display(),
            )));
        }
        return Err(swap_err.into());
    }
    Ok(())
}

/// Sweep stale rebuild debris (`<path>.new` / `<path>.old`).
///
/// **Recovery preservation (R1):** if `<path>` itself doesn't exist or lacks
/// a `CURRENT` file (i.e. is not a healthy RocksDB), `<path>.old` is the
/// manual-recovery copy of a prior failed swap — leave it in place.
/// We sweep `.new` unconditionally — that's always partial-build debris.
///
/// ## Concurrency (L2)
///
/// This function is called at `open_at()` entry, BEFORE acquiring the repo
/// lock. Concurrent openers may race through sweep — that's benign because
/// `remove_dir_all` of a non-existent path returns `NotFound` which we log
/// and ignore. The worst case is a duplicated `tracing::warn` line.
///
/// ## Visibility (L9)
///
/// `pub(super)` — called from `open_at()` and `open_unverified()` in this
/// module. Not exposed to downstream crates.
pub(super) fn sweep_rebuild_debris(path: &Path) {
    let parent = match path.parent() {
        Some(p) => p,
        None => return,
    };
    let file = match path.file_name().and_then(|s| s.to_str()) {
        Some(f) => f,
        None => return,
    };
    let new_path = parent.join(format!("{file}.new"));
    if new_path.exists()
        && let Err(e) = std::fs::remove_dir_all(&new_path)
    {
        tracing::warn!(
            "could not sweep rebuild debris at {}: {e}",
            new_path.display()
        );
    }
    // RocksDB always writes a `CURRENT` file pointing at the active MANIFEST.
    // Its presence is the closest cheap proxy for "this dir holds a usable DB".
    let path_has_healthy_db = path.join("CURRENT").is_file();
    if !path_has_healthy_db {
        let old_path = parent.join(format!("{file}.old"));
        if old_path.exists() {
            tracing::warn!(
                "preserving recovery dir at {} because {} is empty/missing — \
                 manual recovery: `mv {} {}`",
                old_path.display(),
                path.display(),
                old_path.display(),
                path.display()
            );
        }
        return;
    }
    let old_path = parent.join(format!("{file}.old"));
    if old_path.exists()
        && let Err(e) = std::fs::remove_dir_all(&old_path)
    {
        tracing::warn!(
            "could not sweep rebuild debris at {}: {e}",
            old_path.display()
        );
    }
}

impl Storage {
    /// **Escape hatch** — open without recording / verifying a canonical
    /// root_path. Use ONLY when the caller has no canonical root available
    /// (tests, ad-hoc CLI exploration of a known `repo_id`, benchmarks).
    pub fn open_unverified(repo_id: RepoId, base: &Path) -> Result<Self> {
        Self::open_inner(repo_id, base, None)
    }

    /// Open or create with canonical root-path verification (I2).
    ///
    /// On first init, the canonical path is stamped into Meta. Subsequent opens
    /// read it back; if the recorded path doesn't match we **refuse** rather
    /// than silently merge two repos that happen to hash to the same 64-bit RepoId.
    pub fn open_at(repo_id: RepoId, root_path: &Path, base: &Path) -> Result<Self> {
        // CR-review-#7: this is the caller's path, lossy-decoded — NOT
        // canonicalized (the variable used to be called `canonical`
        // which was misleading). The actual canonicalize happens
        // inside `paths_canonicalize_equal` / `canonical_or_input`.
        let caller_root = root_path.to_string_lossy().into_owned();
        Self::open_inner(repo_id, base, Some(caller_root))
    }

    pub(super) fn open_inner(
        repo_id: RepoId,
        base: &Path,
        root_path: Option<String>,
    ) -> Result<Self> {
        // CR-storage-H1: defense-in-depth at the FS boundary. RepoId's
        // private field + try_new validator ensures only 24-hex strings
        // reach here from production code, but a debug-mode assert
        // catches accidental construction via raw struct expressions
        // in tests / future internal code paths. Production is fail-
        // closed too — Cozo would reject the path, but the error
        // surface is opaque.
        debug_assert!(
            RepoId::is_valid(repo_id.as_str()),
            "RepoId fed into open_inner is not 24-hex: {:?}",
            repo_id.as_str()
        );
        let db_path: PathBuf = base.join("workspace_folders").join(repo_id.as_str());
        sweep_rebuild_debris(&db_path);
        std::fs::create_dir_all(&db_path)?;
        let db = DbInstance::new(ENGINE, &db_path, "{}").map_err(map_err)?;
        for stmt in PER_REPO_DDL {
            run_idempotent(&db, stmt)?;
        }

        // Schema-version check: on mismatch, auto-rebuild.
        // L7: use the pre-computed string constant to avoid repeated `.to_string()`.
        let current = crate::schema::SCHEMA_VERSION_STR;
        let read = match db.run_script(
            "?[v] := *Meta{key: \"schema_version\", value: v}",
            BTreeMap::new(),
            ScriptMutability::Immutable,
        ) {
            Ok(r) => Some(r),
            Err(e) => {
                tracing::warn!(
                    "could not read schema_version from {} ({e}); treating as fresh DB",
                    db_path.display()
                );
                None
            }
        };

        let mismatch = read
            .as_ref()
            .and_then(|r| r.rows.first())
            .and_then(|row| row.first())
            .and_then(|v| match v {
                DataValue::Str(s) if s.as_str() != current => Some(s.to_string()),
                _ => None,
            });
        if let Some(stored) = mismatch {
            // H4: parse().unwrap_or(0) would silently treat a corrupted
            // non-numeric schema_version as 0, then trigger rebuild and
            // destroy the DB. Hard-error instead so the operator can inspect.
            let stored_v: u32 = stored.parse().map_err(|_| {
                Error::Storage(format!(
                    "schema version meta is corrupted at {}: \
                     stored value {stored:?} is not a valid u32. \
                     Manual recovery: inspect Meta{{key:\"schema_version\"}} \
                     and either stamp the correct value or delete the DB.",
                    db_path.display()
                ))
            })?;
            // M9: refuse downgrade — stored schema is newer than this binary.
            // Policy: if the binary is older than the on-disk schema, the
            // operator must upgrade ckg. We deliberately do NOT offer an
            // --allow-downgrade escape hatch — the on-disk format may have
            // changed in ways an older binary cannot read safely, and a
            // silent downgrade would corrupt the DB.
            let current_v: u32 = current.parse().unwrap_or(0);
            if stored_v > current_v {
                return Err(map_err(format!(
                    "DB at {} has schema version {stored} which is newer than \
                     this binary's version {current}. Upgrade ckg to >= v{stored} \
                     or delete the DB and re-index.",
                    db_path.display()
                )));
            }
            tracing::warn!(
                "schema version mismatch (on-disk: {stored}, binary: {current}); \
                 auto-rebuilding {}",
                db_path.display()
            );
            drop(db);
            let new_db = rebuild_at_path(&db_path, &current)?;
            if let Some(rp) = &root_path {
                stamp_root_path(&new_db, rp)?;
            }
            stamp_needs_reindex(&new_db, true)?;
            return Ok(Self {
                repo_id,
                db_path,
                db: new_db,
            });
        }

        // Stamp current schema version (idempotent — `:put` overwrites).
        stamp_schema_version(&db, &current)?;

        // I2: root_path verification.
        // CR-M-7: tolerate symlink-equivalent paths. macOS `/var ↔
        // /private/var` and Linux bind-mount aliases produced spurious
        // "RepoId collision" errors when the caller passed one shape
        // and the DB recorded the other. The fix: if a byte-compare
        // mismatch occurs but BOTH sides canonicalize to the same
        // absolute path, accept and re-stamp the recorded value with
        // the canonical form (one-shot transparent migration). True
        // collisions (two distinct repos that happen to hash to the
        // same RepoId) still differ post-canonicalize and continue to
        // fail closed.
        if let Some(caller_root) = root_path {
            match read_root_path(&db) {
                RootPathProbe::Recorded(stored) if stored != caller_root => {
                    if paths_canonicalize_equal(&stored, &caller_root) {
                        // CR-review-#3: stamp the canonical absolute shape
                        // (not the caller's), so any future caller passing
                        // an alias canonicalizes to the same recorded
                        // value and triggers no further migration writes.
                        // Prevents flap under concurrent multi-shape opens.
                        let canon = canonical_or_input(&caller_root);
                        if stored != canon {
                            tracing::warn!(
                                "{}: recorded root_path {stored:?} differs from caller \
                                 {caller_root:?} byte-for-byte but canonicalizes equal — \
                                 re-stamping with canonical shape {canon:?} (CR-M-7 auto-migrate)",
                                db_path.display()
                            );
                            stamp_root_path(&db, &canon)?;
                        }
                        // Note: registry's `Repo.root_path` (stamped by
                        // put_repo at index time) is NOT updated here —
                        // it's the manifest-key authority and changing
                        // it could break ckg-resolve lookup. The Meta-vs-
                        // registry asymmetry is intentional; callers
                        // matching by registry shape will hit auto-
                        // migrate exactly once per (alias, repo) pair.
                    } else {
                        return Err(Error::Storage(format!(
                            "RepoId collision detected at {}: DB recorded root_path \
                             {stored:?} but caller passed {caller_root:?}. Refusing to \
                             silently merge two repos. Recover by removing the per-repo \
                             DB and re-indexing the correct repo.",
                            db_path.display()
                        )));
                    }
                }
                RootPathProbe::Recorded(_) => { /* matches: confirmed identity */ }
                RootPathProbe::NoRow => {
                    stamp_root_path(&db, &caller_root)?;
                }
                RootPathProbe::ReadFailed(e) => {
                    return Err(Error::Storage(format!(
                        "could not verify recorded root_path at {} (Meta read failed: {e}); \
                         refusing to open under collision-detection contract. Re-run with \
                         `Storage::open_unverified` if you intend to bypass I2 protection.",
                        db_path.display()
                    )));
                }
            }
        }

        // CR-I-2: Crash-recovery probe. If `index_in_progress` was set but
        // never cleared, promote `needs_reindex=true` and clear the flag.
        if read_meta_bool(&db, "index_in_progress") {
            tracing::warn!(
                "{}: previous index run did not complete cleanly — \
                 stamping needs_reindex=true. The :put indexer upserts \
                 rows by id, so plain `ckg index` would NOT evict \
                 orphan rows from the partial run. Re-run \
                 `ckg index --clean` to repopulate from a clean state.",
                db_path.display()
            );
            stamp_needs_reindex(&db, true)?;
            stamp_meta_bool(&db, "index_in_progress", false)?;
        }
        Ok(Self {
            repo_id,
            db_path,
            db,
        })
    }
}

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

    /// H3: Opening a DB stamped with SCHEMA_VERSION + 1 must be refused.
    /// Policy: forced upgrade — no --allow-downgrade escape hatch.
    #[test]
    fn schema_downgrade_is_refused() {
        let base = tempdir().unwrap();
        let repo_id = RepoId::try_new("aabbccddeeff00112233aabb").unwrap();

        // First open — creates the DB at current schema version.
        let storage = Storage::open_unverified(repo_id.clone(), base.path()).unwrap();
        let future_version = (crate::schema::SCHEMA_VERSION + 1).to_string();
        // Stamp a future version to simulate a newer binary having written this DB.
        stamp_schema_version(storage.db(), &future_version).unwrap();
        drop(storage);

        // Second open — must refuse (stored_v > current_v).
        let result = Storage::open_unverified(repo_id, base.path());
        assert!(result.is_err(), "open must refuse when stored schema version > binary version");
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("newer than") || msg.contains("schema version"),
            "error must reference schema version mismatch; got: {msg}"
        );
        assert!(
            msg.contains(&future_version),
            "error must include stored version number; got: {msg}"
        );
    }

    /// STORAGE-M-2: `read_root_path` on a fresh DB (no row stamped yet) must
    /// return `NoRow`, not `ReadFailed`. Verifies the `None => NoRow` arm is
    /// reachable and correctly distinct from an error.
    ///
    /// Note: Cozo enforces `Meta {key: String => value: String}` at the schema
    /// level, so it is impossible to stamp a Null or Int into Meta via the
    /// standard API. The `Some(other) => ReadFailed(...)` arm in `read_root_path`
    /// is a defensive catch for hypothetical future Cozo API changes that relax
    /// type enforcement (e.g. expose a union DataValue in projection). It cannot
    /// be exercised via a live DB in the current Cozo version.
    #[test]
    fn absent_root_path_returns_no_row() {
        let base = tempdir().unwrap();
        let repo_id = RepoId::try_new("cafebabe0011223344556677").unwrap();
        // open_unverified creates the schema but does not stamp root_path.
        let storage = Storage::open_unverified(repo_id, base.path()).unwrap();
        let db = storage.db();

        let probe = read_root_path(db);
        assert!(
            matches!(probe, RootPathProbe::NoRow),
            "fresh DB must return NoRow before root_path is stamped"
        );
    }

    /// STORAGE-M-2b: once root_path is stamped, `read_root_path` must return
    /// `Recorded` with the correct value — confirming the `Some(Str)` arm.
    #[test]
    fn stamped_root_path_returns_recorded() {
        let base = tempdir().unwrap();
        let repo_id = RepoId::try_new("aabb00112233445566778899").unwrap();
        let storage = Storage::open_unverified(repo_id, base.path()).unwrap();
        let db = storage.db();

        stamp_root_path(db, "/my/repo").unwrap();
        let probe = read_root_path(db);
        match probe {
            RootPathProbe::Recorded(s) => assert_eq!(s, "/my/repo"),
            other => panic!("expected Recorded, got {other:?}"),
        }
    }

    /// H4: A non-numeric schema_version must hard-error, not silently rebuild.
    /// Previously `parse().unwrap_or(0)` would trigger rebuild and destroy data.
    #[test]
    fn corrupted_schema_version_is_refused() {
        let base = tempdir().unwrap();
        let repo_id = RepoId::try_new("deadbeefcafe001122334455").unwrap();

        // First open — creates the DB normally.
        let storage = Storage::open_unverified(repo_id.clone(), base.path()).unwrap();
        // Stamp a non-numeric value to simulate on-disk corruption.
        stamp_schema_version(storage.db(), "not_a_number").unwrap();
        drop(storage);

        // Second open — must error, NOT silently rebuild.
        let result = Storage::open_unverified(repo_id, base.path());
        assert!(result.is_err(), "open must refuse on corrupted (non-numeric) schema_version");
        let msg = result.err().unwrap().to_string();
        assert!(
            msg.contains("corrupted") || msg.contains("not a valid"),
            "error must mention schema version corruption; got: {msg}"
        );
        assert!(
            msg.contains("not_a_number"),
            "error must echo the bad stored value; got: {msg}"
        );
    }
}