mahbot 0.4.1

use anyhow::Context;
use chrono::{DateTime, Utc};
use futures_util::FutureExt;
use std::fs::File;
use std::panic::AssertUnwindSafe;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::OnceCell;
use tracing::{info, warn};
use turso::Builder;
pub(crate) use turso::{IntoParams, Row, Value, params};

use crate::util::UnwrapPoison;

/// Read `N` bytes at `offset` via a positional read (does not move the shared
/// file offset and never opens/closes the file).
#[cfg(unix)]
pub(crate) fn pread_at<const N: usize>(file: &File, offset: u64) -> Option<[u8; N]> {
    use std::os::unix::fs::FileExt;
    let mut buf = [0u8; N];
    file.read_exact_at(&mut buf, offset).ok()?;
    Some(buf)
}

#[cfg(windows)]
pub(crate) fn pread_at<const N: usize>(file: &File, offset: u64) -> Option<[u8; N]> {
    use std::os::windows::fs::FileExt;
    let mut buf = [0u8; N];
    file.seek_read(&mut buf, offset).ok()?;
    Some(buf)
}

/// Positional read from offset 0 — convenience for header-sized reads.
pub(crate) fn pread<const N: usize>(file: &File) -> Option<[u8; N]> {
    pread_at(file, 0)
}

// ── Timestamp helper ────────────────────────────────────────────────

/// Current UTC timestamp in RFC 3339 format for database columns.
#[must_use]
pub fn now() -> String {
    Utc::now().to_rfc3339()
}

/// Parse an RFC 3339 timestamp string into a UTC DateTime.
///
/// All database timestamps are generated by [`now`] and are stored as RFC 3339
/// strings with a timezone offset (e.g., `2026-07-02T14:20:40+00:00`). This
/// function normalises them to [`DateTime<Utc>`] regardless of the offset
/// embedded in the string.
///
/// # Errors
///
/// Returns [`chrono::ParseError`] when the input is not a valid RFC 3339
/// timestamp.
pub(crate) fn parse_utc_timestamp(s: &str) -> Result<DateTime<Utc>, chrono::ParseError> {
    DateTime::parse_from_rfc3339(s).map(|dt| dt.with_timezone(&Utc))
}

/// Feature names for experimental Turso database features that must be enabled
/// consistently by the main daemon and the debug CLI.
///
/// These correspond to the `with_*()` methods / public fields of
/// [`turso::core::DatabaseOpts`] — see [`experimental_database_opts`] for
/// the canonical construction.
///
/// [`Connection::open`] derives its `experimental_*()` builder calls from
/// [`experimental_database_opts`], so this constant and [`experimental_database_opts`]
/// are the two places that define which features are active. The
/// `experimental_features_are_consistent` test verifies they match.
///
/// # API naming asymmetries
///
/// The `turso::Builder` (used by [`Connection::open`]) and `turso::core::DatabaseOpts`
/// (used by [`experimental_database_opts`]) have slightly different APIs for the same
/// underlying features. Known mismatches:
///
/// | DatabaseOpts field / `with_*()` | Builder `experimental_*()` | Feature string |
/// |---|---:|---|
/// | `enable_views` / `with_views()` | `experimental_materialized_views()` | `"views"` |
///
/// Some fields exist on only one side:
/// - `enable_autovacuum` / `with_autovacuum()` — DatabaseOpts only, no Builder equivalent.
/// - `unsafe_testing` / `with_unsafe_testing()` — DatabaseOpts only, no Builder equivalent.
/// - `experimental_triggers()` — Builder only (no-op for backwards compatibility).
/// - `experimental_strict()` — Builder only (no-op for backwards compatibility).
///
/// See the test `builder_mapping_matches_experimental_features` which verifies that
/// the field-by-field mapping in [`Connection::open`] enables exactly the features
/// listed here.
#[cfg_attr(
    not(test),
    expect(
        dead_code,
        reason = "Referenced only by assertion tests; kept for documentation"
    )
)]
pub(crate) const EXPERIMENTAL_FEATURES: &[&str] = &["index_method", "multiprocess_wal"];

/// Return an iterator over all checkpointable database stores.
///
/// Each item is `(name, Option<&'static Connection>)` where `None` means the
/// store has not been initialized yet.
///
/// This is the **single source of truth** for which stores exist and are
/// checkpointed.  [`store_names`] derives the name list from this iterator,
/// guaranteeing no drift between the name list and the checkpoint list.
pub(crate) fn iter_checkpoint_stores()
-> impl Iterator<Item = (&'static str, Option<&'static crate::turso::Connection>)> {
    [
        ("board", crate::board::BOARD.get().map(|s| &s.conn)),
        (
            "chat_history",
            crate::chat_history::CHAT_HISTORY.get().map(|s| &s.conn),
        ),
        (
            "config",
            crate::config_db::CONFIG_STORE.get().map(|s| &s.conn),
        ),
        ("logs", crate::logs::LOG_STORE.get().map(|s| &s.conn)),
        ("sessions", crate::session::SESSIONS.get().map(|s| &s.conn)),
        ("users", crate::users::USER_STORE.get().map(|s| &s.conn)),
        (
            "workspaces",
            crate::workspace::WORKSPACES.get().map(|s| &s.conn),
        ),
    ]
    .into_iter()
}

/// Return all canonical store names, derived from [`iter_checkpoint_stores`].
///
/// This replaces the former `ALL_STORE_NAMES` constant — the name list is now
/// derived from the same single-source-of-truth iterator that drives
/// checkpointing, so no drift is possible.
///
/// Used by:
/// - `mahbot debug` — validates `--db` argument values.
/// - Callers that previously referenced `ALL_STORE_NAMES`.
pub(crate) fn store_names() -> Vec<&'static str> {
    iter_checkpoint_stores().map(|(name, _)| name).collect()
}

/// Initialize all database stores concurrently.
///
/// This is the canonical initialization path for the 6 data stores (board,
/// session, workspace, users, chat_history, config).  The `logs` store
/// is **not** included here because it must be initialized earlier via
/// [`crate::logs::init_tracing`], which requires the log store before any
/// other subsystem is ready. The stats tables (tool_calls, llm_requests)
/// live in the logs store.
///
/// > **Keep this list in sync with [`iter_checkpoint_stores`]** — every store
/// > listed here must also appear in that iterator.  The converse is not strictly
/// > required because `logs` (and any future store initialized outside this path)
/// > lives only in the checkpoint iterator.
///
/// # Real parallelism (mahbot-1709 decision 5)
///
/// Each store is spawned onto the runtime rather than joined via `try_join!`.
/// turso's async query API performs the actual scan synchronously inside
/// `poll` (a `step` only returns `Pending` on async I/O; for page-cache hits it
/// returns `Row`/`Done` immediately), so `try_join!` polls the six
/// opens — each ending in a full-DB `quick_check` — sequentially on a single
/// task. Spawning gives each store its own worker, so the integrity scans run
/// in parallel (the win scales with DB size).
///
/// # Error semantics
///
/// Unlike `try_join!` (fail-fast: cancels sibling opens on the first error,
/// propagates panics), the spawned tasks all run to completion — every store
/// gets its integrity verified even when a sibling fails — and the first error
/// (in completion order) is reported. A panic inside a store init surfaces as
/// a `JoinError` and is mapped back to an error so the boot failure surfaces
/// through the binary's `bootstrap_mahbot_safe` catch_unwind instead of
/// vanishing into the runtime.
pub async fn init_all_stores() -> anyhow::Result<()> {
    let mut set = tokio::task::JoinSet::new();
    set.spawn(crate::session::init_global());
    set.spawn(crate::workspace::init_global());
    set.spawn(crate::users::init_global());
    set.spawn(crate::board::init_global());
    set.spawn(crate::chat_history::init_global());
    set.spawn(crate::config_db::init_global());

    let mut first_error: Option<anyhow::Error> = None;
    while let Some(result) = set.join_next().await {
        let outcome = match result {
            Ok(Ok(())) => None,
            Ok(Err(e)) => Some(e),
            Err(join_err) => {
                // A store-init panic surfaces as JoinError — preserve the
                // boot-failure UX (previously panics propagated through
                // try_join! into bootstrap_mahbot_safe's catch_unwind).
                let message = join_err.try_into_panic().map_or_else(
                    |_| "store init task failed to join".to_string(),
                    |p| crate::util::panic_message(&*p),
                );
                Some(anyhow::anyhow!("store init task panicked: {message}"))
            }
        };
        if let Some(e) = outcome
            && first_error.is_none()
        {
            first_error = Some(e);
        }
    }

    match first_error {
        Some(e) => Err(e),
        None => Ok(()),
    }
}

/// Create [`turso::core::DatabaseOpts`] with all experimental features enabled.
///
/// This is the **single source of truth** for which experimental features are active.
/// [`Connection::open`] reads its builder calls from this function, and
/// [`EXPERIMENTAL_FEATURES`] lists the feature names for test verification.
///
/// Used by `mahbot debug` to open databases with the same feature set as the
/// main daemon, preventing `.tshm` WAL coordination file inconsistencies.
///
/// # Adding a new feature
///
/// 1. Add the `with_*()` call here.
/// 2. Add the `experimental_*()` mapping in [`Connection::open`].
/// 3. Add the feature name string to [`EXPERIMENTAL_FEATURES`].
///
/// See [`EXPERIMENTAL_FEATURES`] for known API naming asymmetries between
/// `DatabaseOpts::with_*()` and `Builder::experimental_*()`.
#[must_use]
pub(crate) fn experimental_database_opts() -> turso::core::DatabaseOpts {
    turso::core::DatabaseOpts::new()
        .with_multiprocess_wal(true)
        .with_index_method(true)
}

/// Engine opts for forensic-family reads (`mahbot debug --family`): the live
/// feature set minus `multiprocess_wal` — the legacy read-only WAL path reads
/// `db` + `-wal` directly. Note it still probes a present `.tshm`
/// (`reject_live_multiprocess_wal_for_legacy_open`), so tshm-bearing families
/// are opened from a temp copy that omits the coordination file; the
/// no-touch guarantee lives in that copy, not in these opts.
#[must_use]
pub(crate) fn family_database_opts() -> turso::core::DatabaseOpts {
    experimental_database_opts().with_multiprocess_wal(false)
}

/// Register a global singleton store.
///
/// This is the canonical init pattern for all DB-backed global stores. Each module
/// calls this from its `init_global()` function with its `OnceCell`, a name for
/// error messages, and an async open function (typically a closure that captures
/// the storage root and calls the store's `open` method).
///
/// # Errors
/// Returns an error if the store fails to open, or if the cell is already set.
pub(crate) async fn register_global_store<T, F, Fut>(
    cell: &OnceCell<T>,
    name: &str,
    open_fn: F,
) -> anyhow::Result<()>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = anyhow::Result<T>> + Send,
{
    let store = open_fn().await?;
    cell.set(store)
        .map_err(|_| anyhow::anyhow!("{name} already initialized"))?;
    Ok(())
}

/// Declare a global `OnceCell`-backed store with `init_global()` and `store()`.
///
/// Generates three items:
/// - `pub static $NAME: OnceCell<$Type>` — the underlying cell.
/// - `pub async fn init_global()` — calls `register_global_store` with the
///   constructor function invoked on `CONFIG.global_storage_root()`.
/// - `#[must_use] pub fn store()` — returns `&'static $Type`, panicking if
///   not yet initialized.
///
/// # Syntax
///
/// Invocation with a required custom expect message:
/// ```ignore
/// global_store! {
///     /// Doc comment for the static.
///     pub static $NAME: $Type,
///     constructor = $constructor_expr,
///     expect = $expect_message,
/// }
/// ```
#[macro_export]
macro_rules! global_store {
    // Custom expect form.
    (
        $(#[$attr:meta])*
        $vis:vis static $name:ident: $ty:ty,
        constructor = $constructor:expr,
        expect = $expect:expr,
    ) => {
        $(#[$attr])*
        $vis static $name: ::tokio::sync::OnceCell<$ty> =
            ::tokio::sync::OnceCell::const_new();

        #[doc = concat!("Initialize the global ", stringify!($name), " store.")]
        $vis async fn init_global() -> ::anyhow::Result<()> {
            let root = $crate::config::CONFIG.global_storage_root();
            $crate::turso::register_global_store(
                &$name,
                stringify!($name),
                || $constructor(&root),
            )
            .await
        }

        #[must_use]
        #[doc = concat!(
            "Get a reference to the global ",
            stringify!($name),
            " store.\n\n# Panics\n\nPanics if the store has not been initialized.",
        )]
        $vis fn store() -> &'static $ty {
            $name.get().expect($expect)
        }
    };
}

/// Remove characters that cause FTS query parser errors.
///
/// Tantivy special characters that act as syntax operators in query terms.
/// Characters not listed here (`.`, `@`, `#`, `_`, `/`, `$`, `%`, `!`, `,`,
/// `?`, etc.) are safe word characters in Tantivy's query grammar and are
/// preserved to improve search precision.
///
/// Source: Tantivy's `query_grammar.rs`.
static TANTIVY_SPECIAL: &[char] = &[
    '+', // Must (term must be present)
    '^', // Boost modifier
    '~', // Proximity / fuzzy — conservative guard; Tantivy only treats ~ as
    // a modifier when it appears after a term/phrase, but stripping it
    // in other positions is harmless.
    ':', // Field specifier
    '{', '}', // Exclusive range
    '"', '\'', // Phrase query delimiters
    '`',  // ESCAPE_IN_WORD — strict parser rejects backtick at any position
    '[', ']', // Inclusive range
    '(', ')',  // Grouping / sub-query
    '\\', // Escape character
    '*',  // Wildcard / prefix operator
    '-',  // MustNot (negation) at word start; word boundary elsewhere
];

/// Sanitize a user-supplied query string for Tantivy FTS, stripping syntax
/// operators that would cause parse errors.
///
/// Required because turso's FTS uses Tantivy, whose query parser treats
/// certain characters as syntax operators, causing parse errors on
/// user-generated queries. This is a Tantivy design decision, not a turso
/// version bug — upgrading turso will not eliminate this requirement.
///
/// The `ngram` tokenizer indexes character substrings, so removing syntax
/// operators from queries does not reduce search quality. Non-special
/// punctuation is preserved to improve search precision — e.g. email
/// addresses like `user@example.com`, identifiers like `my_function`, or
/// paths like `feature/x` are kept intact.
///
/// # Edge cases
///
/// * Queries starting with `/` have the leading slash stripped to prevent
///   Tantivy's lenient parser from interpreting them as regex queries.
/// * Queries consisting entirely of Tantivy special characters (e.g. `+-~`)
///   produce an empty string, allowing callers to short-circuit.
#[must_use]
pub(crate) fn sanitize_fts_query(query: &str) -> String {
    query
        .split(|c: char| c.is_whitespace() || TANTIVY_SPECIAL.contains(&c))
        .map(|word| word.trim_start_matches('/'))
        .filter(|word| !word.is_empty())
        .collect::<Vec<_>>()
        .join(" ")
}

/// Build a comma-separated list of `?` placeholders for SQL IN-clauses.
///
/// Returns an empty string for `count == 0`. Callers MUST guard against
/// empty lists to avoid producing invalid SQL like `WHERE id IN ()`.
///
/// # Example
///
/// ```ignore
/// // Internal utility — use `sql_in_placeholders(3)` from crate::turso
/// assert_eq!("?, ?, ?", vec!["?"; 3].join(", "));
/// assert_eq!("", vec!["?"; 0].join(", "));
/// ```
///
/// Note: libSQL/SQLite binds `Vec<Value>` positionally regardless of whether
/// the SQL uses `?` or `?N`, so numbered placeholders (`?1, ?2, ...`) are
/// never necessary — use this helper everywhere.
#[must_use]
pub(crate) fn sql_in_placeholders(count: usize) -> String {
    vec!["?"; count].join(", ")
}

/// One sidecar file's identity check result (see [`PersistentSidecarFd::identity`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SidecarIdentity {
    /// fd held and its inode/device match a fresh `stat` of the path.
    Matches,
    /// No fd and no file (fresh store — the sidecar appears lazily on first
    /// write and is then opened once via [`PersistentSidecarFd::ensure_open`]).
    Absent,
    /// The path no longer exists while an fd is held (unlinked by an external
    /// process). The coordination predicate detects any resulting orphaned-WAL
    /// state; checkpoints via the daemon's own fd are unaffected.
    Deleted,
    /// The path resolves to a different inode than the held fd — the file was
    /// replaced by an external process. Callers suspend checkpoints/TRUNCATE
    /// for the store (the identity result itself is the signal).
    Replaced,
}

/// Persistent read-only fd to a store sidecar (`-tshm`/`-wal`), opened once
/// when the file first exists and never closed.
///
/// macOS fcntl locks are process-scoped: closing **any** fd to a file drops
/// all of the process's locks on it. The daemon must therefore never open+
/// close `.tshm` files after startup — that would silently release the byte-0
/// lifetime lock and let a second process classify its open as `Exclusive`
/// (triggering repair that wipes live reader slots). All coordination reads go
/// through this fd via positional `pread`.
///
/// The fd's inode/device identity is compared against the path on each read
/// (stat by path never opens the file); a mismatch means an external process
/// replaced the file — consumers suspend destructive operations (checkpoints)
/// on the store. When the store is recreated (logs quarantine path) the whole
/// `Connection` is rebuilt, so the fd is naturally re-pointed at the new file.
///
/// A sidecar that does not exist yet (turso creates `-wal`/`-tshm` lazily on
/// first write) is re-pointed once by [`ensure_open`](Self::ensure_open) when
/// it appears; until then reads fall back to path-based access (counted by the
/// daemon's open+close regression counter).
#[derive(Debug)]
pub(crate) struct PersistentSidecarFd {
    path: std::path::PathBuf,
    /// Opened on first appearance, never closed (see struct docs).
    file: std::sync::OnceLock<File>,
    /// Serializes the one-time lazy open — a second transient fd on a
    /// coordination file must never be created and dropped (that drop would
    /// release the process's fcntl locks on it).
    open_lock: std::sync::Mutex<()>,
}

impl PersistentSidecarFd {
    /// Open the sidecar `db_path + suffix` read-only. No fd when the file does
    /// not exist yet (fresh store before first write; [`ensure_open`] re-points
    /// once it appears).
    fn open(db_path: &Path, suffix: &str) -> Self {
        let path = std::path::PathBuf::from(format!("{}{suffix}", db_path.display()));
        let file = std::sync::OnceLock::new();
        if let Ok(f) = std::fs::File::open(&path) {
            let _ = file.set(f);
        }
        Self {
            path,
            file,
            open_lock: std::sync::Mutex::new(()),
        }
    }

    /// Lazy one-time re-point: open the persistent fd when the sidecar file
    /// first appears (serialized — see [`PersistentSidecarFd`]).
    fn ensure_open(&self) {
        if self.file.get().is_some() {
            return;
        }
        let _guard = self.open_lock.lock().unwrap_poison();
        if self.file.get().is_some() {
            return; // another thread re-pointed first
        }
        if let Ok(f) = std::fs::File::open(&self.path) {
            let _ = self.file.set(f); // cannot fail under the lock
        }
    }

    /// The persistent fd, when the sidecar file exists.
    pub(crate) fn file(&self) -> Option<&File> {
        self.file.get()
    }

    /// Re-check the persistent fd's inode/device against a fresh `stat` of the
    /// path. `Absent` is silent (fresh store — no fd to compare); `Deleted`
    /// and `Replaced` are external-interference conditions (only `Replaced`
    /// means the fd reads a stale inode — the caller suspends checkpoints).
    #[must_use]
    pub(crate) fn identity(&self) -> SidecarIdentity {
        self.ensure_open();
        let Some(file) = self.file.get() else {
            return SidecarIdentity::Absent;
        };
        #[cfg(unix)]
        {
            use std::os::unix::fs::MetadataExt;
            let Ok(fd_meta) = file.metadata() else {
                return SidecarIdentity::Deleted;
            };
            let Ok(path_meta) = std::fs::metadata(&self.path) else {
                return SidecarIdentity::Deleted; // path unlinked
            };
            if fd_meta.dev() == path_meta.dev() && fd_meta.ino() == path_meta.ino() {
                return SidecarIdentity::Matches;
            }
            SidecarIdentity::Replaced
        }
        #[cfg(not(unix))]
        {
            // The inode/device replacement detector guards the macOS
            // process-scoped fcntl-lock vector; other platforms have no such
            // lock-drop mechanism.
            SidecarIdentity::Matches
        }
    }
}

/// One store's coordination-file read sources for wal-guard inspection.
pub(crate) type StoreFds<'a> = crate::wal_guard::StoreFds<'a>;

/// A serialized handle to a turso connection with persistent sidecar fds.
///
/// Mutex-serializes concurrent access (turso connections do not support
/// concurrent operations) and tracks a dangling transaction: when a
/// [`TxGuard`] is dropped without explicit commit/rollback, the flag is set
/// and the next write operation rolls it back first (mirrors the upstream
/// `turso::Connection::dangling_tx` pattern at wrapper level, avoiding async
/// in `Drop`). The persistent sidecar fds (see [`PersistentSidecarFd`]) are
/// opened once at store open and never closed — the daemon-side open+close
/// of `.tshm`/`-wal` is what drops the macOS process-scoped fcntl locks.
#[derive(Clone, Debug)]
pub(crate) struct Connection {
    /// Persistent turso connection — reused for all execute/query calls.
    /// Mutex serializes concurrent access since libsql connections
    /// do not support concurrent operations.
    conn: Arc<tokio::sync::Mutex<turso::Connection>>,
    /// Set when a TxGuard is dropped without explicit commit/rollback.
    /// Checked at the start of every write operation (execute, begin_tx).
    /// Mirrors the upstream `turso::Connection::dangling_tx` pattern but
    /// works at our wrapper level so we don't need async in Drop.
    has_dangling_tx: Arc<AtomicBool>,
    /// Persistent read-only fds to the store's `-tshm`/`-wal` sidecars —
    /// opened once, never closed (see [`PersistentSidecarFd`]).
    tshm_fd: Arc<PersistentSidecarFd>,
    wal_fd: Arc<PersistentSidecarFd>,
}

impl Connection {
    /// The persistent sidecar fds for wal-guard inspection (lazily re-pointed
    /// when a sidecar appears after the store open).
    #[must_use]
    pub(crate) fn store_fds(&self) -> StoreFds<'_> {
        self.tshm_fd.ensure_open();
        self.wal_fd.ensure_open();
        StoreFds {
            tshm: self.tshm_fd.file(),
            wal: self.wal_fd.file(),
        }
    }

    /// Re-check the sidecar identities and return the worst non-matching
    /// condition (`Replaced` > `Deleted`), or `None` when both sidecars match
    /// or are absent. The caller throttles the announcement (the wal-guard
    /// loop re-announces on the class-warning schedule); `Replaced` means the
    /// persistent fds read stale inodes and checkpoints must be suspended.
    #[must_use]
    pub(crate) fn check_coordination_identity(&self) -> Option<SidecarIdentity> {
        let mut worst = None;
        for fd in [&self.tshm_fd, &self.wal_fd] {
            match fd.identity() {
                SidecarIdentity::Matches | SidecarIdentity::Absent => {}
                SidecarIdentity::Deleted => {
                    if worst != Some(SidecarIdentity::Replaced) {
                        worst = Some(SidecarIdentity::Deleted);
                    }
                }
                SidecarIdentity::Replaced => worst = Some(SidecarIdentity::Replaced),
            }
        }
        worst
    }
}

/// Message prefix of a known Limbo quick_check false positive.
///
/// Limbo's FTS keeps tantivy chunks in an internal backing index while the
/// dir table stays empty, so the index-cardinality comparison always
/// mismatches once the index has content (upstream tursodatabase/turso#7611,
/// unfixed through 0.7.x). Only this exact count-mismatch message is masked;
/// other messages naming the same internal index (missing/non-unique entries)
/// signal real corruption and stay reported. Re-check if upstream lands a fix
/// or renames internals.
const KNOWN_FTS_DIR_COUNT_FALSE_POSITIVE: &str =
    "wrong # of entries in index __turso_internal_fts_dir_";

/// Name prefix of the internal FTS backing index (never a class-B repair
/// target — FTS-rebuild is out of scope; the repair guard matches the
/// extracted index name against this, not the full message prefix).
const FTS_INTERNAL_INDEX_PREFIX: &str = "__turso_internal_fts_dir_";

/// `sqlite_master` filter for user-owned objects: excludes the engine's own
/// tables (`sqlite_%`) and turso's protected `__turso_internal_%`
/// (AUTOINCREMENT seq backing, FTS dir) — those reject user writes and are
/// recreated by the DDL replay. Shared by the rebuild's counts enumeration,
/// DDL replay, and the debug CLI's schema dump so the three cannot drift.
pub(crate) const USER_OBJECT_FILTER: &str =
    "name NOT LIKE 'sqlite_%' AND name NOT LIKE '__turso_internal_%'";

/// Best-effort removal of a rebuild temp family (main + sidecars).
fn remove_rebuild_temp(temp: &Path) {
    let _ = std::fs::remove_file(temp);
    let _ = std::fs::remove_file(format!("{}-wal", temp.display()));
    let _ = std::fs::remove_file(format!("{}-shm", temp.display()));
    let _ = std::fs::remove_file(format!("{}-tshm", temp.display()));
}

/// RAII cleanup of the rebuild temp family: every exit from the migration —
/// including a turso panic mid-copy (the documented pager-OOB class) —
/// removes the temp main + sidecars. The panic fallback in open_store
/// (recreate on the Healthy arm, propagate/retry on the other arms) discards
/// the fully-read migrated data either way — the accepted panic trade-off.
struct TempCleanup<'a>(&'a Path);
impl Drop for TempCleanup<'_> {
    fn drop(&mut self) {
        remove_rebuild_temp(self.0);
    }
}

/// True when a `quick_check`/`integrity_check` message names the internal FTS
/// backing index — either the known count-mismatch false positive or any
/// other problem row about that index (the row-level scan masks only the exact
/// count-mismatch message; this is the broader never-REINDEX-the-FTS guard).
#[must_use]
pub(crate) fn known_fts_dir_false_positive(message: &str) -> bool {
    message.contains(FTS_INTERNAL_INDEX_PREFIX)
}

/// Map each row in a slice through a fallible closure, collecting into a
/// `Vec<turso::Result<T>>` with per-row error conversion.
///
/// Used by `Connection::query_map`.
fn map_rows<T, E>(
    rows: &[Row],
    mut map: impl FnMut(&Row) -> std::result::Result<T, E>,
) -> Vec<turso::Result<T>>
where
    E: std::fmt::Display,
{
    rows.iter()
        .map(|row| map(row).map_err(|e| turso::Error::Error(e.to_string())))
        .collect()
}

impl Connection {
    pub async fn open(path: &Path) -> anyhow::Result<Self> {
        let path_str = path
            .to_str()
            .with_context(|| format!("database path must be UTF-8: {}", path.display()))?;
        // Derive experimental features from the canonical opts function.
        // If you need to add/remove an experimental feature, change
        // experimental_database_opts() and EXPERIMENTAL_FEATURES — not here.
        let opts = experimental_database_opts();
        let db = Builder::new_local(path_str)
            .experimental_index_method(opts.enable_index_method)
            .experimental_multiprocess_wal(opts.enable_multiprocess_wal)
            .build()
            .await
            .context("failed to open local database")?;
        let conn = db.connect()?;
        conn.busy_timeout(Duration::from_mins(1))?;
        // In-memory temp storage for this connection (turso's
        // TempStore::Memory): every intermediate query structure — RETURNING
        // buffers, ORDER BY/LIMIT heap sorts, DISTINCT, IN-subqueries,
        // compound SELECTs, window functions, CREATE INDEX, sorter/hash-join
        // spills — stays in RAM and never touches a temp dir. turso_core's
        // tempdir creation has no parent-creation and no fallback chain, and
        // resolves through $TMPDIR, which the daemon pins to its private
        // root: a missing root used to fail every eager-temp statement with
        // "I/O error (tempdir): entity not found". The PRAGMA maps to the
        // per-connection set_temp_store (turso_core translate/pragma.rs);
        // if a future engine rejects it, this open fails loudly instead of
        // silently regressing to disk-backed temp storage.
        conn.execute("PRAGMA temp_store = MEMORY;", ())
            .await
            .context("failed to set in-memory temp storage (PRAGMA temp_store = MEMORY)")?;
        // Open the persistent sidecar fds AFTER turso's open (which creates
        // the .tshm on first use). These fds are never closed for the
        // process lifetime — see PersistentSidecarFd.
        let tshm_fd = Arc::new(PersistentSidecarFd::open(path, "-tshm"));
        let wal_fd = Arc::new(PersistentSidecarFd::open(path, "-wal"));
        Ok(Self {
            conn: Arc::new(tokio::sync::Mutex::new(conn)),
            has_dangling_tx: Arc::new(AtomicBool::new(false)),
            tshm_fd,
            wal_fd,
        })
    }

    /// Lock the inner connection and rollback any dangling transaction.
    /// Both read and write callers should use this method — dangling
    /// transactions can affect read visibility in WAL mode if left active.
    async fn lock_and_cleanup(&self) -> tokio::sync::MutexGuard<'_, turso::Connection> {
        let conn = self.conn.lock().await;
        if self.has_dangling_tx.swap(false, Ordering::SeqCst) {
            let _ = conn.execute("ROLLBACK", ()).await;
        }
        conn
    }

    pub async fn execute(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<u64> {
        let conn = self.lock_and_cleanup().await;
        conn.execute(sql, params).await
    }

    pub(crate) async fn execute_batch(&self, sql: &str) -> turso::Result<()> {
        let conn = self.lock_and_cleanup().await;
        conn.execute_batch(sql).await
    }

    /// Begin a transaction and return a guard that keeps the connection locked
    /// until the transaction is committed or rolled back.
    pub async fn begin_tx(&self) -> turso::Result<TxGuard<'_>> {
        let conn = self.lock_and_cleanup().await;
        conn.execute("BEGIN", ()).await?;
        Ok(TxGuard {
            conn,
            has_dangling_tx: Some(self.has_dangling_tx.clone()),
        })
    }

    /// Execute a read-only query, returning all matching rows.
    /// Acquires the mutex so reads are serialized with writes — this eliminates
    /// page-cache races that occurred when reads used `db.connect()` to spawn
    /// a fresh connection sharing a cache with the write connection.
    pub async fn query(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<Vec<Row>> {
        let conn = self.lock_and_cleanup().await;
        Self::query_impl(&conn, sql, params).await
    }

    /// Core query logic shared by [`Connection::query`] and [`TxGuard::query`].
    /// Operates on an already-locked connection. Collects all rows into a `Vec`.
    async fn query_impl(
        conn: &turso::Connection,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<Vec<Row>> {
        let mut rows = conn.query(sql, params).await?;
        let mut result = Vec::new();
        while let Some(row) = rows.next().await? {
            result.push(row);
        }
        Ok(result)
    }

    /// Execute a read-only query, mapping each row through a closure.
    /// Returns a Vec of results so callers can handle per-row errors
    /// individually.
    pub async fn query_map<T, E>(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnMut(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> turso::Result<Vec<turso::Result<T>>>
    where
        T: Send + 'static,
        E: std::fmt::Display + Send + Sync + 'static,
    {
        let rows = self.query(sql, params).await?;
        Ok(map_rows(&rows, map))
    }

    /// Map every row, failing on the first per-row error (unlike
    /// [`Self::query_map`], which returns per-row results).
    pub async fn query_map_strict<T, E>(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnMut(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> anyhow::Result<Vec<T>>
    where
        T: Send + 'static,
        E: std::fmt::Display + Send + Sync + 'static,
    {
        let rows = self.query_map(sql, params, map).await?;
        rows.into_iter()
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(Into::into)
    }

    /// Execute a query that returns exactly one row.
    /// Acquires the mutex so reads are serialized with writes.
    pub async fn query_row<T, E>(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnOnce(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> turso::Result<T>
    where
        E: std::fmt::Display + Send + Sync + 'static,
    {
        let conn = self.lock_and_cleanup().await;
        Self::query_row_impl(&conn, sql, params, map).await
    }

    /// Execute a query that returns zero or one row.
    ///
    /// Returns `Ok(Some(val))` if a row is found, `Ok(None)` when no row
    /// matches (i.e. [`turso::Error::QueryReturnedNoRows`] is caught), or
    /// `Err` if the query fails for another reason.
    ///
    /// This is a convenience wrapper around [`Self::query_row`] that
    /// eliminates the common `match { Ok(val) => Ok(Some(val)),
    /// Err(QueryReturnedNoRows) => Ok(None), Err(e) => Err(e.into()) }`
    /// boilerplate.
    pub async fn query_optional<T, E>(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnOnce(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> anyhow::Result<Option<T>>
    where
        E: std::fmt::Display + Send + Sync + 'static,
    {
        match self.query_row(sql, params, map).await {
            Ok(val) => Ok(Some(val)),
            Err(::turso::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// Core query_row logic shared by [`Connection::query_row`] and
    /// [`TxGuard::query_row`].  Operates on an already-locked connection.
    /// Returns [`turso::Error::QueryReturnedNoRows`] when no row matches.
    async fn query_row_impl<T, E>(
        conn: &turso::Connection,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnOnce(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> turso::Result<T>
    where
        E: std::fmt::Display + Send + Sync + 'static,
    {
        let mut rows = conn.query(sql, params).await?;
        let row = rows
            .next()
            .await?
            .ok_or(turso::Error::QueryReturnedNoRows)?;
        map(&row).map_err(|e| turso::Error::Error(e.to_string()))
    }

    /// Force a WAL checkpoint with TRUNCATE mode.
    ///
    /// Writes all pending WAL content to the main database file and truncates
    /// the WAL. Run before hard process termination (e.g.,
    /// [`std::process::exit`] in self-update) for a clean store handoff;
    /// committed data is already fsync-durable at COMMIT.
    ///
    /// Safe to call even if the database is not in WAL mode — non-WAL databases
    /// treat this as a no-op (the result row reports `log == checkpointed == -1`).
    ///
    /// The result row is parsed into a [`CheckpointOutcome`] so callers can
    /// distinguish a complete checkpoint from a busy or partial one — Limbo
    /// never reports a busy result in the row's first column on its success
    /// path, so incompleteness must be derived from `log > checkpointed`.
    pub async fn checkpoint(&self) -> anyhow::Result<CheckpointOutcome> {
        self.run_checkpoint(CheckpointMode::Truncate).await
    }

    /// Run a non-truncating (PASSIVE) WAL checkpoint.
    ///
    /// Backfills as many WAL frames as possible into the main database file
    /// without blocking readers or writers and without truncating the WAL —
    /// it never resets the shared WAL frame index, so it is safe to run while
    /// other connections are live. The WAL file keeps growing until a
    /// TRUNCATE checkpoint runs; callers bound that growth with a size cap.
    pub async fn checkpoint_passive(&self) -> anyhow::Result<CheckpointOutcome> {
        self.run_checkpoint(CheckpointMode::Passive).await
    }

    async fn run_checkpoint(&self, mode: CheckpointMode) -> anyhow::Result<CheckpointOutcome> {
        let rows = self
            .query(&format!("PRAGMA wal_checkpoint({});", mode.label()), ())
            .await
            .context("Failed to checkpoint WAL")?;
        let row = rows
            .first()
            .context("PRAGMA wal_checkpoint returned no result row")?;
        Ok(CheckpointOutcome {
            busy: match row.get_value(0)? {
                Value::Integer(n) => n != 0,
                _ => anyhow::bail!("Unexpected result from PRAGMA wal_checkpoint"),
            },
            log_frames: int_column(row, 1)?,
            checkpointed_frames: int_column(row, 2)?,
        })
    }

    /// Run PRAGMA quick_check to verify database integrity.
    ///
    /// Checks b-tree page structure, NOT NULL and CHECK constraints, and
    /// index cardinality. Returns `Ok(())` on success, or an error with the
    /// first corruption message if any corruption is detected.
    ///
    /// Lightweight (~10ms on a healthy store) and read-only — safe to call
    /// periodically while the system is running.
    pub async fn quick_check(&self) -> anyhow::Result<()> {
        if let Some(problem) = self.quick_check_problems().await?.into_iter().next() {
            anyhow::bail!("Database integrity check failed: {problem}");
        }
        Ok(())
    }

    /// All `quick_check` problem rows (the known FTS false positive filtered
    /// out) — unlike [`Self::quick_check`], which bails on the first problem.
    /// The class-B repair needs the full list: an overflow-aliasing row
    /// anywhere in the scan must veto the REINDEX even when an earlier row
    /// names an index desync.
    pub(crate) async fn quick_check_problems(&self) -> anyhow::Result<Vec<String>> {
        let rows = self
            .query("PRAGMA quick_check;", ())
            .await
            .context("Failed to execute PRAGMA quick_check")?;
        scan_integrity_rows(&rows)
    }
}

/// Collect problem rows from `PRAGMA quick_check` (skipping `"ok"` and the
/// known FTS index-cardinality false positive). Used by
/// [`Connection::quick_check_problems`].
fn scan_integrity_rows(rows: &[Row]) -> anyhow::Result<Vec<String>> {
    let mut problems: Vec<String> = Vec::new();
    for row in rows {
        match row.get_value(0)? {
            Value::Text(s) if s == "ok" => {}
            Value::Text(s) if s.contains(KNOWN_FTS_DIR_COUNT_FALSE_POSITIVE) => {}
            Value::Text(s) => problems.push(s),
            _ => anyhow::bail!("Unexpected result from PRAGMA quick_check"),
        }
    }
    Ok(problems)
}

/// Outcome of `PRAGMA wal_checkpoint`, parsed from its 3-column result row
/// (`busy`, `log`, `checkpointed`).
///
/// Limbo's `op_checkpoint` writes `busy == 0` on its success path; `busy == 1`
/// is set only on the pager checkpoint error path. A partial checkpoint is
/// therefore detected via `log > checkpointed` (frames remaining in the WAL).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CheckpointOutcome {
    /// True when the checkpoint returned SQLITE_BUSY (locks held by another
    /// writer/checkpointer).
    pub busy: bool,
    /// WAL frame high-water mark after the checkpoint (`-1` for non-WAL
    /// databases). After a complete TRUNCATE the frame index resets, so this
    /// equals `checkpointed_frames`; `is_complete` relies on that reset.
    pub log_frames: i64,
    /// Frames backfilled into the database by the checkpoint (`-1` for
    /// non-WAL databases).
    pub checkpointed_frames: i64,
}

impl CheckpointOutcome {
    /// True when every WAL frame was backfilled and no busy condition was hit.
    #[must_use]
    pub fn is_complete(&self) -> bool {
        !self.busy && self.log_frames <= self.checkpointed_frames
    }
}

/// Read an integer column from a checkpoint result row.
fn int_column(row: &Row, idx: usize) -> anyhow::Result<i64> {
    match row.get_value(idx)? {
        Value::Integer(n) => Ok(n),
        _ => anyhow::bail!("Unexpected result from PRAGMA wal_checkpoint"),
    }
}

/// A locked connection handle scoped to a single transaction.
/// Holds the mutex guard for the entire duration — dropped guard triggers rollback.
pub(crate) struct TxGuard<'a> {
    conn: tokio::sync::MutexGuard<'a, turso::Connection>,
    /// Shared flag on the parent Connection; set in Drop to signal a deferred
    /// rollback on the next write operation. Set to None when the transaction
    /// has been explicitly committed or rolled back, preventing Drop from
    /// flagging a dangling transaction.
    has_dangling_tx: Option<Arc<AtomicBool>>,
}

impl TxGuard<'_> {
    pub async fn execute(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<u64> {
        self.conn.execute(sql, params).await
    }

    /// Execute multiple SQL statements (e.g. a schema) inside the transaction.
    pub async fn execute_batch(&self, sql: &str) -> turso::Result<()> {
        self.conn.execute_batch(sql).await
    }

    /// Execute a query that returns exactly one row.
    pub async fn query_row<T, E>(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        map: impl FnOnce(&Row) -> std::result::Result<T, E> + Send + 'static,
    ) -> turso::Result<T>
    where
        E: std::fmt::Display + Send + Sync + 'static,
    {
        Connection::query_row_impl(&self.conn, sql, params, map).await
    }

    /// Execute a query returning zero or more rows.
    /// Returns an empty Vec when no rows match.
    pub async fn query(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<Vec<Row>> {
        Connection::query_impl(&self.conn, sql, params).await
    }

    /// Commit the transaction and release the lock.
    pub async fn commit(mut self) -> turso::Result<()> {
        self.conn.execute("COMMIT", ()).await?;
        // Clear flag so Drop doesn't try to roll back an already-committed tx
        self.has_dangling_tx = None;
        Ok(())
    }

    /// Rollback the transaction and release the lock.
    pub async fn rollback(mut self) -> turso::Result<()> {
        self.conn.execute("ROLLBACK", ()).await?;
        // Clear flag so Drop doesn't try to roll back again
        self.has_dangling_tx = None;
        Ok(())
    }
}

impl Drop for TxGuard<'_> {
    fn drop(&mut self) {
        // Set the dangling_tx flag on the parent Connection so the next
        // write operation (execute, begin_tx) will issue a ROLLBACK first.
        // This is a deferred pattern — we can't call async methods from Drop,
        // but the lock isn't released yet (MutexGuard is still alive), so
        // no other task can execute a write before the flag is set.
        if let Some(flag) = &self.has_dangling_tx {
            flag.store(true, Ordering::SeqCst);
        }
    }
}

// Schema / index management

/// Ensure a full-text search index exists with the correct tokenizer.
/// Drops and recreates if the existing index has a different tokenizer.
pub(crate) async fn ensure_fts_index(
    conn: &Connection,
    index_name: &str,
    tokenizer: &str,
    ddl: &str,
) -> anyhow::Result<()> {
    let existing_sql: Option<String> = conn
        .query_optional(
            "SELECT sql FROM sqlite_master WHERE type='index' AND name=?1 LIMIT 1",
            params![index_name],
            |row| match row.get_value(0)? {
                Value::Text(s) => Ok::<_, ::turso::Error>(s),
                _ => Ok::<_, ::turso::Error>(String::new()),
            },
        )
        .await?
        .filter(|s| !s.is_empty());

    let needs_rebuild = existing_sql
        .as_deref()
        .is_none_or(|sql| !sql.to_lowercase().contains(&tokenizer.to_lowercase()));

    if needs_rebuild {
        conn.execute(&format!("DROP INDEX IF EXISTS {index_name}"), ())
            .await?;
        // Race-safe: `CREATE INDEX IF NOT EXISTS` prevents failure when two
        // connections run schema init concurrently (e.g. parallel tests).
        conn.execute(ddl, ()).await?;
    }

    Ok(())
}

// ── Database migrations ────────────────────────────────────────────────
//
// Centralized migration machinery. This design tracks applied migrations in a
// `schema_migrations` table — the historical PRAGMA user_version approach was
// deliberately removed and must NOT be resurrected (it is not
// transaction-atomic with DDL). Each migration is a ready-made SQL
// statement/script applied exactly once, in order, at store initialization.

/// One ordered, ready-made schema migration.
#[derive(Debug, Clone, Copy)]
pub(crate) struct Migration {
    /// Stable unique id recorded in `schema_migrations` — a migration with
    /// this id never runs twice, even across store recreations.
    pub(crate) id: &'static str,
    /// The SQL statement/script applied when the migration is pending.
    pub(crate) sql: &'static str,
    /// Optional existence guard `(table, column)`: when the column already
    /// exists, the SQL is skipped but the migration is still recorded as
    /// applied. This is what makes the wipe-and-recreate operational
    /// sequence safe — a fresh-DB SCHEMA already contains the migrated shape
    /// (the column exists), so a duplicate-column ALTER must never fire; the
    /// guard turns it into a no-op recorded as applied. The upgrade-in-place
    /// path (column missing) runs the SQL and records it. Both paths
    /// converge to the same final state.
    pub(crate) guard: Option<(&'static str, &'static str)>,
}

/// Check whether `table` has a column named `column`.
///
/// `PRAGMA table_info` reports one row per column with the column name at
/// position 1.
pub(crate) async fn column_exists(
    conn: &Connection,
    table: &str,
    column: &str,
) -> anyhow::Result<bool> {
    let rows = conn
        .query(&format!("PRAGMA table_info({table})"), ())
        .await
        .context("Failed to read table schema (PRAGMA table_info)")?;
    Ok(rows
        .iter()
        .any(|row| row.get::<String>(1).ok().as_deref() == Some(column)))
}

/// Apply pending migrations for a store, in order, exactly once each.
///
/// Called at store initialization (after the SCHEMA batch ran) via the
/// store's `post_open` hook (see [`crate::define_store!`]). Migrations are
/// tracked in a per-store `schema_migrations` table:
///
/// - The tracking table is created if missing (fresh databases).
/// - Each pending migration runs inside its own transaction — the schema
///   change and its tracking row commit atomically (turso supports
///   transactional DDL), so a failed migration never leaves a half-applied
///   state or a false "applied" record.
/// - When a migration's guard already holds (e.g. a fresh-DB SCHEMA that
///   already contains the migrated shape), the SQL is skipped and only the
///   tracking row is written.
/// - Already-applied migrations are skipped entirely (never re-run).
///
/// The `logs` store and the read-only `mahbot debug` path never call this —
/// the debug CLI opens with `OpenFlags::ReadOnly|NoLock` and migrations only
/// run through the store `open` methods.
pub(crate) async fn run_pending_migrations(
    conn: &Connection,
    store_name: &str,
    migrations: &[Migration],
) -> anyhow::Result<()> {
    conn.execute(
        "CREATE TABLE IF NOT EXISTS schema_migrations (\
             id         TEXT PRIMARY KEY,\
             applied_at TEXT NOT NULL\
         )",
        (),
    )
    .await
    .context("Failed to create schema_migrations tracking table")?;

    let applied: std::collections::HashSet<String> = conn
        .query("SELECT id FROM schema_migrations", ())
        .await
        .context("Failed to read applied migrations")?
        .into_iter()
        .filter_map(|row| row.get::<String>(0).ok())
        .collect();

    for migration in migrations {
        if applied.contains(migration.id) {
            continue;
        }
        let guard_holds = match migration.guard {
            Some((table, column)) => {
                column_exists(conn, table, column).await.with_context(|| {
                    format!(
                        "Migration '{}' guard check failed on {table}.{column}",
                        migration.id
                    )
                })?
            }
            None => false,
        };
        if guard_holds {
            // Fresh-DB path: the SCHEMA already produced the migrated shape —
            // record the migration as applied without running its SQL.
            conn.execute(
                "INSERT INTO schema_migrations (id, applied_at) VALUES (?1, ?2)",
                params![migration.id, now()],
            )
            .await
            .with_context(|| format!("Failed to record migration '{}' as applied", migration.id))?;
        } else {
            let tx = conn.begin_tx().await.with_context(|| {
                format!("Migration '{}': failed to begin transaction", migration.id)
            })?;
            tx.execute_batch(migration.sql)
                .await
                .with_context(|| format!("Migration '{}' failed", migration.id))?;
            tx.execute(
                "INSERT INTO schema_migrations (id, applied_at) VALUES (?1, ?2)",
                params![migration.id, now()],
            )
            .await
            .with_context(|| format!("Failed to record migration '{}' as applied", migration.id))?;
            tx.commit()
                .await
                .with_context(|| format!("Migration '{}': failed to commit", migration.id))?;
        }
        tracing::info!(
            store = store_name,
            migration = migration.id,
            skipped_sql = guard_holds,
            "Applied database migration",
        );
    }
    Ok(())
}

/// Open a database, create parent directories if needed, and run schema init.
///
/// `schema` is executed via `execute_batch` (multiple DDL statements).
pub(crate) async fn open_with_schema(db_path: &Path, schema: &str) -> anyhow::Result<Connection> {
    if let Some(parent) = db_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("Failed to create directory: {}", parent.display()))?;
    }

    let conn = Connection::open(db_path)
        .await
        .with_context(|| format!("Failed to open database: {}", db_path.display()))?;

    conn.execute("PRAGMA foreign_keys = ON;", ())
        .await
        .context("Failed to enable foreign key enforcement")?;

    conn.execute_batch(schema)
        .await
        .context(format!("Failed to run schema {schema}"))?;

    Ok(conn)
}

/// Absolute path to `<root>/db/<name>.db` — single source of truth for store
/// file naming across the debug CLI, WAL guard, and logs quarantine.
#[must_use]
pub(crate) fn store_db_path(root: &Path, name: &str) -> std::path::PathBuf {
    root.join("db").join(format!("{name}.db"))
}

/// Sidecar files (`-wal`, `-shm`, `-tshm`) beside a store's database file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct StoreSidecars {
    pub wal: std::path::PathBuf,
    pub shm: std::path::PathBuf,
    pub tshm: std::path::PathBuf,
}

#[must_use]
pub(crate) fn store_sidecars(db_path: &Path) -> StoreSidecars {
    let base = db_path.display().to_string();
    StoreSidecars {
        wal: std::path::PathBuf::from(format!("{base}-wal")),
        shm: std::path::PathBuf::from(format!("{base}-shm")),
        tshm: std::path::PathBuf::from(format!("{base}-tshm")),
    }
}

/// Resource/permission keywords shared by the error classifiers: actionable
/// signals (never corruption) — ENOSPC/EMFILE/OOM/permission must not trigger
/// a quarantine. One base predicate keeps the two classifiers from drifting.
const RESOURCE_SIGNAL_KEYWORDS: [&str; 4] = [
    "no space left on device",
    "too many open files",
    "out of memory",
    "permission denied",
];

fn has_resource_signal(lower: &str) -> bool {
    RESOURCE_SIGNAL_KEYWORDS.iter().any(|k| lower.contains(k))
}

/// True when a `quick_check`/open failure is corruption-class rather than a
/// busy/locked/IO failure of the PRAGMA itself.
///
/// Two forms qualify: the PRAGMA returned a non-`ok` row (our
/// `Database integrity check failed` bail), or the PRAGMA failed to execute
/// with a message that is not a lock/busy, I/O, or resource condition (e.g. a
/// page-level error reading a zeroed page — `Invalid page type`). Only
/// corruption-class failures quarantine — a busy or locked store must never
/// trigger the rename path, and ENOSPC/EMFILE are actionable signals, never
/// corruption (the unified guarded-open path must not quarantine on them).
/// Unknown messages classify as corruption (fail-closed): a novel turso error
/// string at boot could trigger a recreate the matrix does not justify.
/// Accepted trade-off — genuine corruption must not pass as an actionable
/// signal; reclassify new strings as they surface.
pub(crate) fn is_corruption_class(e: &anyhow::Error) -> bool {
    let msg = format!("{e:#}");
    if msg.contains("Database integrity check failed") {
        return true;
    }
    let lower = msg.to_lowercase();
    !(has_resource_signal(&lower)
        || lower.contains("busy")
        || lower.contains("locked")
        || lower.contains("i/o error")
        || lower.contains("no such file"))
}

/// True when a heal-phase failure is an actionable resource/permission
/// condition: ENOSPC/EMFILE/OOM must never trigger a quarantine (they are
/// signals, not corruption — the boot-heal path propagates them instead of
/// recreating). Persistent busy, I/O errors, and unreadable states are
/// recreate candidates, so only this narrow set propagates.
fn is_actionable_signal(e: &anyhow::Error) -> bool {
    has_resource_signal(&format!("{e:#}").to_lowercase())
}

/// Marker wrapping a fresh-store open failure after the boot-heal path already
/// quarantined the original family. The outer quarantine (logs path) must not
/// run again on this error — the fresh store is not corrupt, the recreate
/// itself failed.
#[derive(Debug)]
pub(crate) struct RecreateFailed(pub anyhow::Error);

impl std::fmt::Display for RecreateFailed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "fresh store open failed after quarantine: {}", self.0)
    }
}

impl std::error::Error for RecreateFailed {}

/// Open a store database under `<root>/db/<name>.db`.
///
/// Creates parent directories if needed and runs the provided `schema` via
/// [`open_with_schema`].  This is a convenience helper for the near-identical
/// [`open`] methods on each store module, keeping the DB filename and path
/// construction centralised in one place.
///
/// During daemon boot (when a pre-flight diagnosis exists for `name`), the
/// open runs the per-store heal strategy derived by the pre-flight:
/// TRUNCATE-first for stale-tail, PASSIVE-first + TRUNCATE for durable-B,
/// quarantine + recreate for structural damage, and class-B btree-index repair
/// after the open. Every diagnosed open is panic-absorbed (turso's own reopen
/// of a damaged-coordination store can panic; boot must never fail because of
/// a damaged store): heal-phase panics and errors recreate, while post-heal
/// standard-open panics propagate loudly (recreate is not justified for a
/// store whose data was just preserved). Outside the boot flow (tests, CLI)
/// the diagnosis map is empty and this behaves exactly like
/// [`open_with_schema`].
pub(crate) async fn open_store(
    root: &Path,
    name: &str,
    schema: &str,
) -> anyhow::Result<Connection> {
    let db_path = store_db_path(root, name);
    let Some(diagnosis) = crate::wal_guard::take_boot_diagnosis(&db_path) else {
        return open_with_schema(&db_path, schema).await;
    };

    match diagnosis {
        crate::wal_guard::BootDiagnosis::BlockedCoordination => {
            // Pre-open intent: the coordination state blocks a safe reopen
            // (orphaned/foreign/truncated WAL) — no heal; the open proceeds
            // panic-absorbed and turso's own reopen rebuilds the `.tshm` from
            // the WAL, after which the predicate sees Healthy again (the
            // "blocked" state applies to the checkpoint loop only while the
            // store is closed). turso's reopen of a truncated/foreign WAL can
            // panic (wal.rs monotonicity, pager OOB — the documented prod
            // class). Detect-and-continue on the intact main DB: move only
            // the broken coordination sidecars aside and retry; full recreate
            // is the last resort.
            //
            // Error asymmetry vs the healable arm: non-corruption-class open
            // errors (busy/I-O/resource) here propagate raw (failing boot)
            // rather than recreating — corruption-class errors already
            // recreate inside `open_and_repair`, and a plain failure would
            // discard the intact main DB's data.
            let result = AssertUnwindSafe(open_and_repair(&db_path, name, schema))
                .catch_unwind()
                .await;
            match result {
                Ok(Ok(conn)) => Ok(conn),
                Ok(Err(e)) => Err(e),
                Err(payload) => {
                    retry_after_coordination_panic(&db_path, name, schema, &panic_err(&*payload))
                        .await
                }
            }
        }
        crate::wal_guard::BootDiagnosis::Healthy => {
            // Healthy coordination + open panic is a main-DB content issue
            // (pager OOB), not a coordination one — the sidecar-only
            // quarantine path would discard -wal durability while the
            // "intact main DB" premise is contradicted by the panic itself.
            // Recreate (the incurable path).
            let result = AssertUnwindSafe(open_and_repair(&db_path, name, schema))
                .catch_unwind()
                .await;
            match result {
                Ok(Ok(conn)) => Ok(conn),
                Ok(Err(e)) => Err(e),
                Err(payload) => {
                    recreate_after_failed_heal(&db_path, name, schema, &panic_err(&*payload)).await
                }
            }
        }
        healable @ (crate::wal_guard::BootDiagnosis::StaleTail
        | crate::wal_guard::BootDiagnosis::DurableB) => {
            // Phase 1: the heal connection's own open — the risky reopen of
            // damaged coordination. A panic or persistent error (e.g. busy)
            // here is a recreate candidate — but only for corruption-
            // class failures: ENOSPC/EMFILE are actionable signals and must
            // never trigger a quarantine.
            let healed = AssertUnwindSafe(heal_checkpoint_sequence(&db_path, name, healable))
                .catch_unwind()
                .await;
            match healed {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    // Persistent heal failure (busy after retries, I/O,
                    // unreadable) is a recreate candidate per the heal
                    // fallback; only actionable resource conditions
                    // (ENOSPC/EMFILE/OOM/permission) propagate as signals.
                    if is_actionable_signal(&e) {
                        return Err(e);
                    }
                    return recreate_after_failed_heal(&db_path, name, schema, &e).await;
                }
                Err(payload) => {
                    return recreate_after_failed_heal(
                        &db_path,
                        name,
                        schema,
                        &panic_err(&*payload),
                    )
                    .await;
                }
            }
            // Phase 2: standard open + schema + class-B repair. The heal
            // succeeded — a panic here is not a heal failure and recreate is
            // not justified (recreate only for heal-phase failures and
            // structural/unreadable states); propagate loudly.
            let result = AssertUnwindSafe(open_and_repair(&db_path, name, schema))
                .catch_unwind()
                .await;
            match result {
                Ok(Ok(conn)) => Ok(conn),
                Ok(Err(e)) => Err(e),
                Err(payload) => Err(anyhow::anyhow!(
                    "post-heal open of store '{name}' panicked: {}",
                    crate::util::panic_message(&*payload)
                )),
            }
        }
        crate::wal_guard::BootDiagnosis::Structural => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' structurally damaged (pre-flight diagnosis) — \
                 quarantining and recreating",
            ));
            // The recreate path tolerates a partial quarantine (un-moved
            // files stay in place) — the bool is intentionally ignored.
            let _ = quarantine_store_artifacts(&db_path);
            // Fresh store: open_with_schema directly — open_and_repair's
            // schema-error path would quarantine the fresh store a second
            // time (opposite policy of the logs path). Marker: the original
            // family is already quarantined — a further quarantine (logs
            // outer path) would move the fresh store.
            let result = AssertUnwindSafe(open_with_schema(&db_path, schema))
                .catch_unwind()
                .await;
            match result {
                Ok(Ok(conn)) => Ok(conn),
                Ok(Err(e)) => Err(anyhow::anyhow!(RecreateFailed(e))),
                Err(payload) => Err(anyhow::anyhow!(RecreateFailed(anyhow::anyhow!(
                    "recreated store '{name}' open panicked: {}",
                    crate::util::panic_message(&*payload)
                )))),
            }
        }
    }
}

/// Format a catch_unwind panic payload into a store-open error.
fn panic_err(payload: &(dyn std::any::Any + Send)) -> anyhow::Error {
    anyhow::anyhow!(
        "store open panicked: {}",
        crate::util::panic_message(payload)
    )
}

/// Detect-and-continue fallback after a BlockedCoordination open panic:
/// turso's reopen of a truncated/foreign WAL can panic (wal.rs monotonicity,
/// pager OOB — the documented prod class). Move only the broken coordination
/// sidecars aside and retry on the intact main DB; full recreate is the last
/// resort.
///
/// The whole `-wal` is moved aside, including for the Oversized class where it
/// holds valid committed frames up to max_frame that a PASSIVE-first heal
/// would preserve — asymmetric with the healable arms, but only reachable via
/// a reopen panic (which signals deeper damage than a plain oversized tail).
///
/// The sidecar quarantine happens before the retry: an actionable-signal
/// failure of the retry open (ENOSPC/EMFILE/permission) propagates with the
/// coordination sidecars already quarantined while the main DB is preserved —
/// the `-wal`'s uncheckpointed frames are lost in that scenario (inherent to
/// the panic path that precedes it).
async fn retry_after_coordination_panic(
    db_path: &Path,
    name: &str,
    schema: &str,
    reason: &anyhow::Error,
) -> anyhow::Result<Connection> {
    crate::boot::boot_diagnostic(format!(
        "store '{name}' open panicked ({reason}) — quarantining coordination \
         sidecars and retrying on the intact main DB",
    ));
    // The recreate path tolerates a partial quarantine — the bool is
    // intentionally ignored.
    let _ = quarantine_coordination_sidecars(db_path);
    let retry = AssertUnwindSafe(open_with_schema(db_path, schema))
        .catch_unwind()
        .await;
    match retry {
        Ok(Ok(conn)) => Ok(conn),
        Ok(Err(e)) => {
            // Actionable resource conditions (ENOSPC/EMFILE/permission) must
            // never quarantine the family — the main DB is intact here, so
            // propagate the signal; anything else (busy after retries, I/O,
            // unreadable, corruption) recreates.
            if is_actionable_signal(&e) {
                return Err(e);
            }
            recreate_after_failed_heal(db_path, name, schema, &e).await
        }
        Err(p) => recreate_after_failed_heal(db_path, name, schema, &panic_err(&*p)).await,
    }
}

/// Outcome of the class-B btree-index repair.
enum RepairOutcome {
    /// No recreate needed: quick_check passed, or a report-only condition
    /// (unknown signature / failed in-place repair / aborted migration left
    /// for operator review). The store opens normally either way.
    NoRepair,
    /// REINDEX (or the DROP+CREATE fallback) cleared the desync.
    Repaired,
    /// Overflow-aliasing: the store's data was rebuilt in a fresh file and
    /// swapped into place (the in-file DROP of the aliased b-trees would
    /// double-free the shared overflow page and re-corrupt the freelist).
    /// The returned connection is the reopened, verified store.
    Migrated,
    /// The table is unreadable (quick_check scan failure) — recreate is
    /// justified per the recreate matrix.
    Unreadable,
}

/// Open with the schema, then run the class-B btree-index repair (boot path
/// only — single-writer, exclusive access, wal-guard not yet started). This
/// runs a full quick_check on every store at every boot (7× full-DB scans,
/// plus a verification scan after a repair) — the fixed boot cost of the
/// repair-at-init design.
async fn open_and_repair(db_path: &Path, name: &str, schema: &str) -> anyhow::Result<Connection> {
    let conn = match open_with_schema(db_path, schema).await {
        Ok(conn) => conn,
        Err(e) if is_corruption_class(&e) => {
            // Schema application choked on a corrupt table (e.g. CREATE INDEX
            // scans it and hits "Invalid page type") — unreadable table,
            // recreate justified.
            return recreate_after_failed_heal(db_path, name, schema, &e).await;
        }
        Err(e) => return Err(e),
    };
    match repair_btree_index_if_desynced(conn, db_path, name, schema).await {
        Ok((RepairOutcome::Unreadable, conn)) => {
            drop(conn); // release the fds before the family rename
            recreate_after_failed_heal(
                db_path,
                name,
                schema,
                &anyhow::anyhow!("class-B unreadable table — recreate justified"),
            )
            .await
        }
        Ok((_, conn)) => Ok(conn),
        // An actionable quick_check failure (ENOSPC/EMFILE/permission)
        // propagated from the repair — never a recreate trigger.
        Err(e) => Err(e),
    }
}

/// The baked overflow-aliasing signature: a prior REINDEX/DROP+CREATE on a
/// store whose index leaves shared an overflow page left the page doubly on
/// the freelist, so a later allocation reuses it as both a b-tree page and an
/// overflow chain; reading that chain yields this exact short-read on every
/// scan. The table stays readable there — table-rebuild in place applies,
/// recreate (data loss) is not justified.
const BAKED_OVERFLOW_ALIASING_READ: &str = "short read on page 167772160";

/// What the class-B repair should do given the full quick_check problem list.
enum RepairTarget {
    /// The table cannot be scanned ("Invalid page type"/"short read") —
    /// recreate is justified.
    Unreadable,
    /// Shared overflow pages between index leaves (fresh "Page N referenced
    /// multiple times" or the baked short-read) — never REINDEX (bakes a
    /// worse error); rebuild the store data-preservingly.
    OverflowAliasing,
    /// A specific non-FTS btree index is desynced — in-place repair.
    Index(String),
    /// No recognizable signature — report only.
    Unknown,
}

/// Classify the full quick_check problem list for the class-B repair.
///
/// Ordering is deliberate and structural: unreadable-table signatures win
/// over everything (the table cannot be scanned at all); overflow-aliasing
/// anywhere in the list vetoes the REINDEX even when an earlier row names an
/// index desync (quick_check can surface both in one scan — the veto must not
/// depend on row order). The baked short-read (page 0x0A000000) is an
/// overflow-aliasing marker, not a table-read failure — the generic short-read
/// check comes after it.
fn classify_repair_target(problems: &[String]) -> RepairTarget {
    if problems.iter().any(|p| p.contains("Invalid page type")) {
        RepairTarget::Unreadable
    } else if problems.iter().any(|p| {
        p.contains("referenced multiple times") || p.contains(BAKED_OVERFLOW_ALIASING_READ)
    }) {
        RepairTarget::OverflowAliasing
    } else if problems.iter().any(|p| p.contains("short read")) {
        RepairTarget::Unreadable
    } else {
        match problems.iter().find_map(|p| desynced_index_name(p)) {
            Some(index) => RepairTarget::Index(index.to_string()),
            None => RepairTarget::Unknown,
        }
    }
}

/// Pre-repair forensic snapshot path (db + wal, no tshm):
/// `{db}.pre-reindex-{stamp}-{pid}`. The base name must stay parseable by
/// `debug::parse_family_name` — the writer round-trip test locks the coupling.
#[must_use]
pub(crate) fn pre_reindex_snapshot_path(db_path: &Path) -> std::path::PathBuf {
    std::path::PathBuf::from(format!(
        "{}.pre-reindex-{}",
        db_path.display(),
        family_stamp()
    ))
}

/// Class-B btree-index desync repair: `quick_check` names a specific
/// non-FTS index → REINDEX in place, DROP+CREATE as the fallback (both
/// validated on the 324MB prod sessions copy — both M>N and M<N desync forms,
/// survives reopen and TRUNCATE). Returns `Ok(([`RepairOutcome`], conn))`; an
/// actionable quick_check failure (ENOSPC/EMFILE/permission) propagates as
/// `Err` — it is a signal, never a recreate trigger.
///
/// Preconditions are satisfied by construction at boot: single-writer,
/// exclusive access, the WAL frame index already healed/reset (fi_len==maxf),
/// and a snapshot copy (db + wal, no tshm) taken before the REINDEX. The
/// overflow-aliasing signature ("referenced multiple times" / shared overflow
/// pages, fresh or baked as "short read on page 167772160") is **never**
/// REINDEXed — that bakes a worse error; the store is rebuilt
/// data-preservingly instead (see [`migrate_overflow_aliased_store`]).
/// Unreadable tables surface as quick_check scan failures ("Invalid page
/// type"/"short read") and fall through to the recreate path.
async fn repair_btree_index_if_desynced(
    conn: Connection,
    db_path: &Path,
    name: &str,
    schema: &str,
) -> anyhow::Result<(RepairOutcome, Connection)> {
    // All problem rows, not just the first: quick_check can surface a named
    // index desync in one row and overflow-aliasing in another — the
    // overflow-aliasing veto must hold regardless of row order.
    let problems = match conn.quick_check_problems().await {
        Ok(p) if p.is_empty() => return Ok((RepairOutcome::NoRepair, conn)),
        Ok(p) => p,
        Err(e) => {
            // The quick_check itself failed to run — an actionable resource
            // condition (ENOSPC/EMFILE/permission) is a signal, never a
            // recreate trigger; propagate it like the heal-phase arm.
            if is_actionable_signal(&e) {
                return Err(e);
            }
            crate::boot::boot_diagnostic(format!(
                "store '{name}' quick_check could not run: {e} — unreadable table, \
                 recreate justified",
            ));
            return Ok((RepairOutcome::Unreadable, conn));
        }
    };
    let index = match classify_repair_target(&problems) {
        RepairTarget::Unreadable => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' quick_check cannot scan a table ({}) — unreadable table, \
                 recreate justified",
                problems.join("; "),
            ));
            return Ok((RepairOutcome::Unreadable, conn));
        }
        // Overflow-aliasing (shared overflow pages between index leaves) must
        // NEVER be REINDEXed — that bakes a worse error ("short read on
        // page …"). Structural veto: checked across the whole problem list.
        RepairTarget::OverflowAliasing => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' quick_check reports overflow-aliasing ({}) — rebuilding \
                 the store data-preservingly in a fresh file",
                problems.join("; "),
            ));
            return migrate_overflow_aliased_store(conn, db_path, name, schema).await;
        }
        RepairTarget::Unknown => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' quick_check flagged an unknown condition: {}",
                problems.join("; "),
            ));
            return Ok((RepairOutcome::NoRepair, conn));
        }
        RepairTarget::Index(index) => {
            // Belt-and-braces: unreachable today (`scan_integrity_rows` filters
            // the exact FTS count-mismatch row before this list is built), but
            // never REINDEX the out-of-scope FTS internal index if a future
            // filter change lets it through.
            if known_fts_dir_false_positive(&index) {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' quick_check flagged the known FTS false positive — \
                     not repairing",
                ));
                return Ok((RepairOutcome::NoRepair, conn));
            }
            index
        }
    };
    // Forensic snapshot (db + wal, no tshm) before the in-place repair.
    let snap = pre_reindex_snapshot_path(db_path);
    let sidecars = store_sidecars(db_path);
    for (src, suffix) in [(db_path, ""), (&sidecars.wal, "-wal")] {
        if let Err(e) = std::fs::copy(
            src,
            std::path::PathBuf::from(format!("{}{suffix}", snap.display())),
        ) {
            warn!(
                error = %e,
                from = %src.display(),
                "Failed to copy pre-reindex snapshot",
            );
        }
    }
    // REINDEX is planner-safe under multiprocess_wal (it rewrites the index
    // btree in place). Quoting: identifiers may contain special characters.
    let quoted = index.replace('"', "\"\"");
    if conn
        .execute_batch(&format!("REINDEX \"{quoted}\";"))
        .await
        .is_ok()
        && conn.quick_check().await.is_ok()
    {
        info!(
            db = %name,
            index = %index,
            "class-B btree index desync repaired in place (REINDEX)",
        );
        return Ok((RepairOutcome::Repaired, conn));
    }
    crate::boot::boot_diagnostic(format!(
        "store '{name}' REINDEX of '{index}' did not clear the quick_check desync — \
         falling back to DROP+CREATE",
    ));
    Ok((
        drop_create_index_fallback(&conn, name, &index, &quoted).await,
        conn,
    ))
}

/// Overflow-aliasing repair: rebuild the store's data in a fresh sibling file,
/// verify it, then swap it into place. The in-file alternative (DROP the
/// aliased b-trees and recreate) double-frees the shared overflow page — the
/// free-walk frees it once per referencing leaf — corrupting the freelist, so
/// the rebuilt store re-aliases on the next allocation (the "baked short read"
/// mechanism). A fresh file is the only path that verifies clean (index valid,
/// data intact).
///
/// Preconditions hold at boot: single-writer, exclusive access, wal-guard and
/// the periodic checkpoint loop not yet started. The original family is
/// quarantined (renamed aside, never deleted — the forensic record) before the
/// swap, satisfying the recreate-path preservation guarantee. Returns
/// `Ok(([`RepairOutcome::Migrated`], reopened))` on success; aborts as
/// `NoRepair` (report-only) on a constraint violation during the data copy —
/// a data-integrity finding, not corruption, and never a silent recreate — or
/// on any other non-actionable failure (the original store is preserved).
/// An unreadable table surfaces as `Unreadable` (recreate justified).
/// Actionable resource conditions (ENOSPC/EMFILE/permission) propagate as
/// `Err` — they must never trigger a quarantine.
#[expect(clippy::too_many_lines)] // one linear boot-repair flow, split across helpers
async fn migrate_overflow_aliased_store(
    conn: Connection,
    db_path: &Path,
    name: &str,
    schema: &str,
) -> anyhow::Result<(RepairOutcome, Connection)> {
    // ── Precondition: fi_len == maxf, else TRUNCATE-checkpoint first ──
    let status = crate::wal_guard::inspect_store_at(db_path, conn.store_fds());
    if let Some(h) = status.tshm
        && u64::from(h.frame_index_len) != h.max_frame
    {
        crate::boot::boot_diagnostic(format!(
            "store '{name}' WAL frame index (len={}) does not match max_frame ({}) — \
             TRUNCATE-checkpoint first in the single-writer window",
            h.frame_index_len, h.max_frame,
        ));
        match conn.checkpoint().await {
            Ok(o) if o.is_complete() => {}
            // Incomplete (busy / frames remaining) is the same precondition
            // failure as an error — the WAL cannot be reset in the
            // single-writer window.
            Ok(o) => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' pre-rebuild TRUNCATE checkpoint incomplete \
                     (busy={}, {} of {} frames) — recreate justified",
                    o.busy, o.checkpointed_frames, o.log_frames,
                ));
                return Ok((RepairOutcome::Unreadable, conn));
            }
            Err(e) if is_actionable_signal(&e) => return Err(e),
            // The WAL precondition cannot be met — recreate is justified here
            // (unlike a table-enumeration failure): quarantine + fresh open
            // clears a stuck WAL, so this is a recovery, not data loss.
            Err(e) => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' pre-rebuild TRUNCATE checkpoint failed: {e} — \
                     recreate justified",
                ));
                return Ok((RepairOutcome::Unreadable, conn));
            }
        }
    }

    // ── Readability + row-count per user table (quick_check does not name
    // the aliased index, so every table's read is gated; unreadable →
    // recreate). `__turso_internal_%` tables reject user writes and are
    // recreated by the DDL replay — excluded from the copy + verification.
    let mut counts: Vec<(String, i64)> = Vec::new();
    let tables = match conn
        .query(
            &format!(
                "SELECT name FROM sqlite_master WHERE type='table' \
                 AND {USER_OBJECT_FILTER} ORDER BY rowid"
            ),
            (),
        )
        .await
    {
        Ok(t) => t,
        // A table-enumeration failure is an engine-level read problem, not
        // proof the data is unreadable — report-only (conservative, no data
        // loss), unlike the checkpoint precondition below which recreate
        // clears.
        Err(e) => {
            let err = anyhow::anyhow!(e);
            if is_actionable_signal(&err) {
                return Err(err);
            }
            crate::boot::boot_diagnostic(format!(
                "store '{name}' cannot enumerate tables for the rebuild: {err} — left \
                 for operator review",
            ));
            return Ok((RepairOutcome::NoRepair, conn));
        }
    };
    for t in &tables {
        let tbl = match t.get::<String>(0) {
            Ok(tbl) => tbl,
            // sqlite_master.name is always TEXT — a non-string here is an
            // engine anomaly; fail closed rather than silently dropping the
            // table from the copy and verification.
            Err(e) => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' table name is not text ({e}) — rebuild aborted; \
                     left for operator review",
                ));
                return Ok((RepairOutcome::NoRepair, conn));
            }
        };
        let quoted = tbl.replace('"', "\"\"");
        match conn
            .query_row(&format!("SELECT COUNT(*) FROM \"{quoted}\""), (), |r| {
                r.get::<i64>(0)
            })
            .await
        {
            Ok(count) => counts.push((tbl, count)),
            Err(e) => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' table '{tbl}' is unreadable ({e}) — the data cannot \
                     be preserved by the rebuild; recreate justified",
                ));
                return Ok((RepairOutcome::Unreadable, conn));
            }
        }
    }

    // ── Build the fresh store at a sibling temp path ──
    let temp =
        std::path::PathBuf::from(format!("{}.rebuild-{}", db_path.display(), family_stamp()));
    // Guard removes the temp family on every exit, including a turso panic
    // mid-copy (absorbed by open_store's catch_unwind).
    let _temp_guard = TempCleanup(&temp);
    let fresh = match Connection::open(&temp).await {
        Ok(f) => f,
        Err(e) if is_actionable_signal(&e) => return Err(e),
        Err(e) => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' rebuild store open failed: {e} — left for operator review",
            ));
            return Ok((RepairOutcome::NoRepair, conn));
        }
    };
    // Schema replay (old creation order) + data copy.
    let migrated = migrate_schema_and_data(&conn, &fresh, &counts).await;
    let outcome = match migrated {
        Ok(()) => match fresh.quick_check().await {
            // TRUNCATE-checkpoint the fresh store before the swap so the temp
            // main file holds every frame — a crash between the main and wal
            // renames then cannot leave a valid-but-empty store (turso does
            // not checkpoint on close).
            Ok(()) => match fresh.checkpoint().await {
                Ok(o) if o.is_complete() => Ok(()),
                Ok(o) => Err(MigrateFailure::Finding(format!(
                    "rebuilt store checkpoint incomplete (busy={}, {} of {} frames)",
                    o.busy, o.checkpointed_frames, o.log_frames,
                ))),
                Err(e) => Err(classify_migrate(e)),
            },
            Err(e) => Err(classify_migrate(e)),
        },
        Err(f) => Err(f),
    };
    drop(fresh); // release the fds before the temp cleanup / family swap
    if let Err(failure) = outcome {
        return match failure {
            MigrateFailure::Actionable(e) => Err(e),
            // The aliasing persists: every boot re-runs this full rebuild
            // attempt (fresh-store open + DDL replay + data copy) before
            // falling back to NoRepair — the operator-facing signal to
            // intervene (e.g. restore from backup).
            MigrateFailure::Finding(msg) => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' rebuild aborted: {msg} — original store preserved \
                     for operator review (no data changed)",
                ));
                Ok((RepairOutcome::NoRepair, conn))
            }
        };
    }

    // ── Swap: quarantine the original family (forensic, never deleted), move
    // the fresh family into place, reopen with the module schema. ──
    drop(conn); // release the fds before the family rename
    if !quarantine_store_artifacts(db_path) {
        // A clobbering swap would destroy part of the original family
        // without a forensic record — never swap onto a partial quarantine.
        // The temp guard cleans the migrated family. The main renames
        // first; sibling renames normally fail together, so main-in-place
        // means the original main is preserved (the reopen assumes it is
        // complete — the precondition checkpoint normally guarantees that)
        // and main-moved means the data is not at its path.
        crate::boot::boot_diagnostic(format!(
            "store '{name}' original family could not be fully quarantined — the \
             rebuild swap would clobber it without a forensic record; rebuild aborted, \
             migrated data discarded",
        ));
        return if db_path.exists() {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' original family remains in place — left for operator \
                 review",
            ));
            open_with_schema(db_path, schema)
                .await
                .map(|reopened| (RepairOutcome::NoRepair, reopened))
        } else {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' original family is in the quarantine and the store \
                 path is empty — boot aborted; recover from the quarantine and retry",
            ));
            Err(anyhow::anyhow!(
                "store '{name}' rebuild aborted: partial quarantine with the main file \
                 already moved — the store path was not rebuilt"
            ))
        };
    }
    // Main-file rename first; the sidecars move only when it succeeds. The
    // fresh store was TRUNCATE-checkpointed, so its main file alone holds
    // every row — a failed sidecar rename is benign, and a crash between
    // the renames cannot leave a valid-but-empty store.
    let sidecars = store_sidecars(db_path);
    let temp_sidecars = store_sidecars(&temp);
    if let Err(e) = std::fs::rename(&temp, db_path) {
        // The swap cannot proceed: the store path is absent and the original
        // family is in the quarantine. Bail loudly (the store's boot fails)
        // instead of opening a fresh empty store over the abandoned migrated
        // data; the temp guard discards the migrated family.
        return Err(anyhow::anyhow!(e).context(format!(
            "store '{name}' rebuild swap main-file rename failed — migrated temp \
             family discarded; original family quarantined for recovery"
        )));
    }
    for (src, dst) in [
        (&temp_sidecars.wal, &sidecars.wal),
        (&temp_sidecars.shm, &sidecars.shm),
        (&temp_sidecars.tshm, &sidecars.tshm),
    ] {
        if src.exists()
            && let Err(e) = std::fs::rename(src, dst)
        {
            warn!(
                error = %e,
                from = %src.display(),
                to = %dst.display(),
                "rebuild swap sidecar rename failed",
            );
        }
    }
    let reopened = match open_with_schema(db_path, schema).await {
        Ok(reopened) => reopened,
        Err(e) if is_actionable_signal(&e) => return Err(e),
        Err(e) => {
            // The reopened store failed — the quarantine holds the original.
            crate::boot::boot_diagnostic(format!(
                "store '{name}' rebuilt store reopen failed: {e} — original family is \
                 quarantined for recovery",
            ));
            return Err(e);
        }
    };
    // Post-swap verification: a genuine finding (integrity message or count
    // mismatch) reports loudly with its own diagnostic; a transient read
    // error on the healthy swapped store stays at warn level — "lost data"
    // must not fire on the latter (it could prompt restoring the corrupt
    // original over a healthy rebuild).
    let mut verified = true;
    let mut finding_reported = false;
    match reopened.quick_check().await {
        Ok(()) => {}
        Err(e) if is_actionable_signal(&e) => {
            warn!(error = %e, db = %name, "post-swap verification quick_check hit a resource signal");
            verified = false;
        }
        Err(e) => {
            crate::boot::boot_diagnostic(format!(
                "store '{name}' rebuilt store failed post-swap quick_check ({e}) — \
                 original family is quarantined for recovery",
            ));
            verified = false;
            finding_reported = true;
        }
    }
    for (tbl, expected) in &counts {
        let quoted = tbl.replace('"', "\"\"");
        match reopened
            .query_row(&format!("SELECT COUNT(*) FROM \"{quoted}\""), (), |r| {
                r.get::<i64>(0)
            })
            .await
        {
            // A read error is not proof of data loss — actionable signals
            // warn transiently, anything else is a genuine finding (the
            // "lost data" diagnostic must not fire on a read failure).
            Err(e) => {
                let err = anyhow::anyhow!(e);
                if is_actionable_signal(&err) {
                    verified = false;
                    warn!(
                        error = %err,
                        db = %name,
                        table = %tbl,
                        "post-swap counts verification query hit a resource signal",
                    );
                } else {
                    crate::boot::boot_diagnostic(format!(
                        "store '{name}' post-swap verification counts query for table \
                         '{tbl}' failed ({err}) — verification incomplete; original \
                         family is quarantined for recovery",
                    ));
                    verified = false;
                    finding_reported = true;
                }
            }
            Ok(c) if c != *expected => {
                crate::boot::boot_diagnostic(format!(
                    "store '{name}' post-swap verification failed for table '{tbl}' \
                     (expected {expected} rows, found {c}) — the swap lost data; \
                     original family is quarantined for recovery",
                ));
                verified = false;
                finding_reported = true;
            }
            Ok(_) => {}
        }
    }
    if verified {
        info!(db = %name, "overflow-aliasing repaired: store rebuilt data-preservingly");
        Ok((RepairOutcome::Migrated, reopened))
    } else {
        // Transient read errors only (no finding reported above): the swap
        // succeeded and the store is in place — the next boot re-verifies.
        if !finding_reported {
            warn!(db = %name, "post-swap verification incomplete (transient read errors) — store is in place; re-verified on the next boot");
        }
        Ok((RepairOutcome::NoRepair, reopened))
    }
}

/// Failures of the overflow-aliasing rebuild, classified for the caller.
enum MigrateFailure {
    /// Actionable resource condition (ENOSPC/EMFILE/permission) — propagate.
    Actionable(anyhow::Error),
    /// Data-integrity finding (constraint violation) or any other
    /// non-actionable failure — report only; the original store is preserved.
    Finding(String),
}

/// Classify a rebuild-path error: actionable resource conditions propagate,
/// everything else is a report-only finding (covers both `turso::Error` from
/// the wrapper's query/execute and `anyhow::Error` from checkpoint/quick_check).
fn classify_migrate<E: Into<anyhow::Error>>(e: E) -> MigrateFailure {
    let err = e.into();
    if is_actionable_signal(&err) {
        MigrateFailure::Actionable(err)
    } else {
        MigrateFailure::Finding(format!("{err:#}"))
    }
}

/// Build `INSERT INTO "…" VALUES (…)` for a row copy using positional
/// placeholders ([`sql_in_placeholders`]); `table_ref` is a bare identifier —
/// the helper adds the surrounding double quotes.
fn row_insert_sql(table_ref: &str, row: &Row) -> Result<(String, Vec<Value>), MigrateFailure> {
    let ncols = row.column_count();
    let mut vals = Vec::with_capacity(ncols);
    for c in 0..ncols {
        vals.push(row.get_value(c).map_err(classify_migrate)?);
    }
    Ok((
        format!(
            "INSERT INTO \"{table_ref}\" VALUES ({})",
            sql_in_placeholders(ncols)
        ),
        vals,
    ))
}

/// Replay the old store's schema (tables + indexes, creation order) into the
/// fresh store and copy every table's data row by row (two connections — the
/// SQL engine cannot reference the old store from the fresh one). Indexes are
/// created before the data copy and maintained incrementally by the inserts.
/// Counts are verified per table; a mismatch or a constraint violation is a
/// data-integrity finding, not corruption.
async fn migrate_schema_and_data(
    old: &Connection,
    fresh: &Connection,
    counts: &[(String, i64)],
) -> Result<(), MigrateFailure> {
    let ddl_rows = old
        .query(
            &format!(
                "SELECT sql FROM sqlite_master \
                 WHERE sql IS NOT NULL \
                   AND {USER_OBJECT_FILTER} \
                 ORDER BY rowid"
            ),
            (),
        )
        .await
        .map_err(classify_migrate)?;
    for row in &ddl_rows {
        // sqlite_master.sql is always TEXT — a non-string here is an engine
        // anomaly; fail closed rather than silently skipping the object
        // (mirrors the counts loop's fail-closed name handling).
        let ddl = row
            .get::<String>(0)
            .map_err(|e| MigrateFailure::Finding(format!("DDL replay row is not text ({e})")))?;
        fresh.execute(&ddl, ()).await.map_err(classify_migrate)?;
    }
    // sqlite_sequence (AUTOINCREMENT bookkeeping) is auto-created by the
    // replayed DDLs; copy its rows so the old watermark survives the
    // explicit-id copy (WAL-mode explicit-rowid inserts advance, never
    // clobber) — deleted high ids must not be re-issued after migration
    // (surviving-rows tables; a fully-purged table restarts at 1, harmless).
    let has_sequence: i64 = old
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='sqlite_sequence'",
            (),
            |r| r.get::<i64>(0),
        )
        .await
        .map_err(classify_migrate)?;
    if has_sequence > 0 {
        let seq_rows = old
            .query("SELECT * FROM sqlite_sequence", ())
            .await
            .map_err(classify_migrate)?;
        for row in &seq_rows {
            let (sql, vals) = row_insert_sql("sqlite_sequence", row)?;
            fresh.execute(&sql, vals).await.map_err(classify_migrate)?;
        }
    }
    for (tbl, expected) in counts {
        let quoted = tbl.replace('"', "\"\"");
        // Full-table materialization: `SELECT *` collects every row before
        // the copy loop — a boot-time memory spike on large stores (the
        // 324 MB-class sessions store), accepted for this rare repair path.
        let rows = old
            .query(&format!("SELECT * FROM \"{quoted}\""), ())
            .await
            .map_err(classify_migrate)?;
        let tx = fresh.begin_tx().await.map_err(classify_migrate)?;
        let mut copied = 0i64;
        for row in &rows {
            let (sql, vals) = row_insert_sql(&quoted, row)?;
            tx.execute(&sql, vals).await.map_err(classify_migrate)?;
            copied += 1;
        }
        tx.commit().await.map_err(classify_migrate)?;
        if copied != *expected {
            return Err(MigrateFailure::Finding(format!(
                "table '{tbl}' copy count {copied} != pre-migration {expected}",
            )));
        }
    }
    Ok(())
}

/// DROP+CREATE fallback for a REINDEX that could not clear the desync:
/// reproduce the index's original DDL from `sqlite_master` (validated on the
/// prod copy; clears desyncs REINDEX cannot).
async fn drop_create_index_fallback(
    conn: &Connection,
    name: &str,
    index: &str,
    quoted: &str,
) -> RepairOutcome {
    let ddl = conn
        .query_optional(
            "SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?",
            (index.to_string(),),
            |row| row.get::<String>(0),
        )
        .await
        .ok()
        .flatten();
    let Some(ddl) = ddl else {
        crate::boot::boot_diagnostic(format!(
            "store '{name}' index '{index}' DDL not found in sqlite_master — cannot \
             DROP+CREATE; left for operator review",
        ));
        return RepairOutcome::NoRepair;
    };
    // Transactional: `execute_batch` runs statements in autocommit, so a
    // failed rebuild (e.g. a UNIQUE-constraint violation on table data)
    // would leave the index permanently dropped. DROP+CREATE inside a tx
    // rolls the DROP back when the CREATE fails — the original (desynced)
    // index survives for operator review. The rollback is deferred: the
    // dropped guard flags the connection and the ROLLBACK fires at the next
    // lock_and_cleanup (all wrapper methods route through it), so the
    // uncommitted DROP is undone before any further DB access.
    let outcome = async {
        let tx = conn.begin_tx().await?;
        tx.execute_batch(&format!("DROP INDEX \"{quoted}\"; {ddl};"))
            .await?;
        tx.commit().await
    }
    .await;
    match outcome {
        Ok(()) if conn.quick_check().await.is_ok() => {
            info!(
                db = %name,
                index = %index,
                "class-B btree index desync repaired in place (DROP+CREATE)",
            );
            RepairOutcome::Repaired
        }
        Ok(()) => {
            // The batch committed but the post-rebuild quick_check is not
            // clean (desync report or the scan itself failed to run) —
            // reported only; the store opens normally either way.
            crate::boot::boot_diagnostic(format!(
                "store '{name}' DROP+CREATE of '{index}' post-rebuild quick_check \
                 not clean — left for operator review",
            ));
            RepairOutcome::NoRepair
        }
        Err(repair_err) => {
            // The DROP is rolled back on the next connection operation
            // (deferred) — the original index is preserved, not dropped.
            crate::boot::boot_diagnostic(format!(
                "store '{name}' DROP+CREATE of '{index}' failed: {repair_err} — DROP \
                 rolled back on the next connection op, original index preserved; left \
                 for operator review",
            ));
            RepairOutcome::NoRepair
        }
    }
}

/// Extract the specific btree index name from a quick_check desync message
/// (`wrong # of entries in index <name>`). Returns `None` for any other
/// message shape (overflow-aliasing, unreadable table — the FTS false
/// positive is already filtered by [`scan_integrity_rows`] before quick_check
/// surfaces an error).
fn desynced_index_name(msg: &str) -> Option<&str> {
    const PREFIX: &str = "wrong # of entries in index ";
    let rest = msg.strip_prefix(PREFIX)?;
    let end = rest.find(['\n', ';', '\r']).unwrap_or(rest.len());
    let name = &rest[..end];
    (!name.is_empty()).then_some(name)
}

/// WAL checkpoint mode — replaces the stringly-typed mode literals.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CheckpointMode {
    Passive,
    Truncate,
}

impl CheckpointMode {
    fn label(self) -> &'static str {
        match self {
            Self::Passive => "PASSIVE",
            Self::Truncate => "TRUNCATE",
        }
    }
}

/// Run the heal strategy on a dedicated connection per mode (each call opens
/// and fully closes its own connection — durable-B's PASSIVE-first requires a
/// reopen before the TRUNCATE). The heal connection's own open is the risky
/// turso reopen of damaged coordination; panics there are handled by the
/// caller (recreate).
async fn heal_checkpoint_sequence(
    db_path: &Path,
    name: &str,
    diagnosis: crate::wal_guard::BootDiagnosis,
) -> anyhow::Result<()> {
    let sidecars = store_sidecars(db_path);
    // PASSIVE-first for durable-B (0-byte main DB + live WAL): backfill
    // without truncating, then a reopen + TRUNCATE compacts. Defensively
    // re-checked in the StaleTail arm — TRUNCATE-first on a 0-byte main DB
    // destroys committed frames.
    let durable_b = matches!(diagnosis, crate::wal_guard::BootDiagnosis::DurableB)
        || (std::fs::metadata(db_path).is_ok_and(|m| m.len() == 0)
            && std::fs::metadata(&sidecars.wal).is_ok_and(|m| m.len() > 0));
    if durable_b {
        heal_checkpoint(db_path, CheckpointMode::Passive, name).await?;
        heal_checkpoint(db_path, CheckpointMode::Truncate, name).await?;
        return Ok(());
    }
    if matches!(diagnosis, crate::wal_guard::BootDiagnosis::StaleTail) {
        // Defensive orphan guard: the WAL bytes are gone (a genuine stale
        // tail always has frames on disk, wal ≥ 32) — TRUNCATE-first has
        // nothing to heal and the heal connection's reopen could panic on the
        // empty WAL; the panic-absorbed phase-2 open handles that state.
        if std::fs::metadata(&sidecars.wal).is_ok_and(|m| m.len() < 32) {
            return Ok(());
        }
        // TRUNCATE-first resets the shared frame index and max_frame while
        // keeping committed frames (validated empirically).
        heal_checkpoint(db_path, CheckpointMode::Truncate, name).await?;
    }
    Ok(())
}

/// Quarantine the whole artifact family and recreate the store fresh.
/// The quarantine copy is never deleted — it is the forensic record.
async fn recreate_after_failed_heal(
    db_path: &Path,
    name: &str,
    schema: &str,
    reason: &anyhow::Error,
) -> anyhow::Result<Connection> {
    crate::boot::boot_diagnostic(format!(
        "store '{name}' heal failed: {reason} — quarantining artifact family and \
         recreating a fresh store",
    ));
    // The recreate path tolerates a partial quarantine — the bool is
    // intentionally ignored.
    let _ = quarantine_store_artifacts(db_path);
    match open_with_schema(db_path, schema).await {
        Ok(conn) => Ok(conn),
        // Marker: the original family is already quarantined — a further
        // quarantine (logs outer path) would move the fresh store.
        Err(e) => Err(anyhow::anyhow!(RecreateFailed(e))),
    }
}

/// Retries for a heal checkpoint that returns busy (1s interval — the LLM
/// retry policies are deliberately not reused).
const HEAL_CHECKPOINT_RETRIES: usize = 3;

/// Run a single-mode WAL checkpoint on a dedicated heal connection, retrying
/// busy outcomes ~3 times with a 1-second interval. The connection is closed
/// before the next mode's open (or the standard store open).
///
/// The temporary connection's fds are dropped on close — on macOS that
/// releases the process's fcntl locks on that `.tshm` (the process-scoped
/// fcntl mechanism the persistent fds exist to avoid).
/// Benign here: the heal runs in the single-writer boot window before the
/// main connection exists, and the instance flock already excludes a second
/// daemon (the open+close regression counter does not observe turso-internal
/// opens by design).
async fn heal_checkpoint(db_path: &Path, mode: CheckpointMode, name: &str) -> anyhow::Result<()> {
    let conn = Connection::open(db_path)
        .await
        .with_context(|| format!("heal connection open failed for {name}"))?;
    let mut attempts_left = HEAL_CHECKPOINT_RETRIES;
    loop {
        attempts_left -= 1;
        let outcome = conn.run_checkpoint(mode).await;
        match outcome {
            Ok(o) if o.is_complete() => return Ok(()),
            Ok(o) if attempts_left > 0 => {
                // is_complete() is also false for a non-busy partial backfill
                // (log > checkpointed); the frame counts make that visible.
                warn!(
                    db = %name,
                    busy = o.busy,
                    checkpointed = o.checkpointed_frames,
                    log = o.log_frames,
                    "heal checkpoint incomplete — retrying",
                );
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
            Ok(o) => anyhow::bail!(
                "heal checkpoint {} incomplete after {HEAL_CHECKPOINT_RETRIES} attempts \
                 (busy={}, checkpointed={}/{})",
                mode.label(),
                o.busy,
                o.checkpointed_frames,
                o.log_frames,
            ),
            Err(e) if attempts_left > 0 => {
                warn!(db = %name, error = %e, "heal checkpoint failed — retrying");
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
            Err(e) => {
                return Err(e).with_context(|| {
                    format!("heal checkpoint {} failed for {name}", mode.label())
                });
            }
        }
    }
}

/// Move a store's whole artifact family (`db`/`-wal`/`-shm`/`-tshm`) aside to
/// a timestamped quarantine name. Best-effort: a rename failure is logged,
/// never fatal. The quarantine copy is never deleted — it is the forensic
/// record for the recreate decision. Returns false when any existing file
/// could not be moved (a partial quarantine — the recreate path proceeds with
/// the un-moved files in place; the migration path must abort instead).
#[must_use]
pub(crate) fn quarantine_store_artifacts(db_path: &Path) -> bool {
    let sidecars = store_sidecars(db_path);
    quarantine_family(
        db_path,
        &[
            (db_path, ""),
            (&sidecars.wal, "-wal"),
            (&sidecars.shm, "-shm"),
            (&sidecars.tshm, "-tshm"),
        ],
    )
}

/// Move only the coordination sidecars (`-wal`/`-shm`/`-tshm`) aside, keeping
/// the main DB in place — detect-and-continue on an intact main DB whose
/// broken sidecars made turso's reopen panic. Same best-effort,
/// never-deleted quarantine semantics as [`quarantine_store_artifacts`].
#[must_use]
fn quarantine_coordination_sidecars(db_path: &Path) -> bool {
    let sidecars = store_sidecars(db_path);
    quarantine_family(
        db_path,
        &[
            (&sidecars.wal, "-wal"),
            (&sidecars.shm, "-shm"),
            (&sidecars.tshm, "-tshm"),
        ],
    )
}

/// Shared quarantine-family rename (timestamped, PID + seq-suffixed to never
/// clobber an earlier quarantine — the seq counter is process-local, so the
/// PID disambiguates same-second renames from other processes, where POSIX
/// rename would silently overwrite the earlier forensic copy).
///
/// Returns true when every existing source was moved aside — a partial
/// quarantine means a subsequent clobbering rename would destroy part of the
/// forensic record; the recreate path tolerates that (the un-moved files stay
/// in place), the migration path does not.
#[must_use]
fn quarantine_family(db_path: &Path, sources: &[(&Path, &str)]) -> bool {
    static QUARANTINE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let stamp = family_stamp();
    let seq = QUARANTINE_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let base = if seq == 0 {
        format!(
            "{}.quarantine-{stamp}",
            db_path.file_name().unwrap_or_default().to_string_lossy()
        )
    } else {
        format!(
            "{}.quarantine-{stamp}-{seq}",
            db_path.file_name().unwrap_or_default().to_string_lossy()
        )
    };
    let mut complete = true;
    for (src, suffix) in sources {
        if !src.exists() {
            continue;
        }
        let dst = db_path.with_file_name(format!("{base}{suffix}"));
        if let Err(e) = std::fs::rename(src, &dst) {
            warn!(
                error = %e,
                from = %src.display(),
                to = %dst.display(),
                "Failed to quarantine store artifact",
            );
            complete = false;
        }
    }
    complete
}

/// Execute `work` within a transaction on `conn`, committing on success.
///
/// `action_label` accepts any `&str` including dynamic temporaries from
/// `format!` — it is intentionally not `&'static str` to allow callers to
/// include dynamic context (e.g. phase transitions, short hashes) in log
/// messages. Use a verb phrase for natural reading (e.g. "add comment",
/// "record sanitation failure") rather than a bare noun.
///
/// Uses `ticket_id` (or any identifying label) and `action_label` for
/// structured warn-level logging on failure.
///
/// Thin adapter over [`with_tx_outcome`]: commits whenever `work` returns
/// `Ok`, inheriting its transaction lifecycle and warning semantics.
pub(crate) async fn with_tx(
    conn: &Connection,
    ticket_id: &str,
    action_label: &str,
    work: impl AsyncFnOnce(&TxGuard<'_>) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
    with_tx_outcome(conn, ticket_id, action_label, async |tx| {
        work(tx).await?;
        Ok(true)
    })
    .await
    .map(|_| ())
}

/// Execute `work` within a transaction, letting the work choose the outcome
/// via its return value, and report it.
///
/// `work` returns `anyhow::Result<bool>`:
/// - `Ok(true)` — commit the transaction.
/// - `Ok(false)` — roll back **without** a warning: the work deliberately
///   aborted and nothing must be written. This follows the board layer's
///   claim convention for expected no-ops (e.g. a CAS phase-guard miss when
///   the ticket moved externally while a stage was finishing).
/// - `Err(e)` — roll back and log the standard warn-level messages,
///   including `"{action_label}: transaction rolled back"`.
///
/// Returns `Ok(true)` when the transaction committed, `Ok(false)` when it
/// was rolled back silently, and the wrapped error on failure.
///
/// This is the canonical transaction helper; [`with_tx`] delegates to it.
pub(crate) async fn with_tx_outcome(
    conn: &Connection,
    ticket_id: &str,
    action_label: &str,
    work: impl AsyncFnOnce(&TxGuard<'_>) -> anyhow::Result<bool>,
) -> anyhow::Result<bool> {
    let tx = conn
        .begin_tx()
        .await
        .map_err(|e| {
            warn!(
                ticket = %ticket_id,
                error = %e,
                "Failed to begin transaction for {action_label}",
            );
            e
        })
        .with_context(|| format!("Failed to begin transaction for {action_label}"))?;

    match work(&tx).await {
        Ok(true) => {
            tx.commit()
                .await
                .map_err(|e| {
                    warn!(
                        ticket = %ticket_id,
                        error = %e,
                        "Failed to commit transaction for {action_label}",
                    );
                    e
                })
                .with_context(|| format!("Failed to commit transaction for {action_label}"))?;
            Ok(true)
        }
        Ok(false) => {
            tx.rollback()
                .await
                .map_err(|e| {
                    warn!(
                        ticket = %ticket_id,
                        error = %e,
                        "Transaction rollback also failed for {action_label}",
                    );
                    e
                })
                .with_context(|| format!("Failed to roll back transaction for {action_label}"))?;
            Ok(false)
        }
        Err(e) => {
            if let Err(rollback_err) = tx.rollback().await {
                warn!(
                    ticket = %ticket_id,
                    error = %rollback_err,
                    "Transaction rollback also failed for {action_label}",
                );
            }
            warn!(
                ticket = %ticket_id,
                error = %e,
                "{action_label}: transaction rolled back",
            );
            Err(e.context(format!("{action_label}: transaction rolled back")))
        }
    }
}

/// Forensic-family name suffix `{stamp}-{pid}` (stamp = `%Y%m%dT%H%M%SZ`, pid = process
/// id). Parse side shape-validates it via `debug::is_family_stamp` — the writer round-trip
/// test locks the coupling. Bind once per family: two calls could straddle a second boundary.
#[must_use]
fn family_stamp() -> String {
    format!(
        "{}-{}",
        Utc::now().format("%Y%m%dT%H%M%SZ"),
        std::process::id()
    )
}

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

    #[test]
    fn experimental_features_are_consistent() {
        // Verify that experimental_database_opts() enables exactly
        // the features listed in EXPERIMENTAL_FEATURES, and no other
        // known experimental features.
        let opts = experimental_database_opts();

        // All features listed in the const must be enabled.
        for feature in EXPERIMENTAL_FEATURES {
            match *feature {
                "index_method" => assert!(
                    opts.enable_index_method,
                    "index_method should be enabled per EXPERIMENTAL_FEATURES"
                ),
                "multiprocess_wal" => assert!(
                    opts.enable_multiprocess_wal,
                    "multiprocess_wal should be enabled per EXPERIMENTAL_FEATURES"
                ),
                other => panic!("unknown experimental feature: {other}"),
            }
        }

        // No other known DatabaseOpts experimental features should be enabled.
        // If turso_core adds a new field here, add a check below to keep the
        // test honest — every field should be accounted for.
        assert!(
            !opts.enable_views,
            "views is not an active experimental feature"
        );
        assert!(
            !opts.enable_custom_types,
            "custom_types is not an active experimental feature"
        );
        assert!(
            !opts.enable_encryption,
            "encryption is not an active experimental feature"
        );
        assert!(
            !opts.enable_autovacuum,
            "autovacuum is not an active experimental feature"
        );
        assert!(
            !opts.enable_vacuum,
            "vacuum is not an active experimental feature"
        );
        assert!(
            !opts.enable_attach,
            "attach is not an active experimental feature"
        );
        assert!(
            !opts.enable_generated_columns,
            "generated_columns is not an active experimental feature"
        );
        assert!(
            !opts.enable_without_rowid,
            "without_rowid is not an active experimental feature"
        );
        assert!(
            !opts.unsafe_testing,
            "unsafe_testing is not an active experimental feature"
        );

        // Family (forensic-snapshot) reads derive from the live opts minus
        // multiprocess_wal (legacy read-only WAL path); the no-touch guarantee
        // comes from the debug CLI's temp copy that omits the `.tshm`, since
        // the legacy path still probes a present coordination file.
        let fam = family_database_opts();
        assert!(
            fam.enable_index_method,
            "family reads need index_method (stores are created with it)"
        );
        assert!(
            !fam.enable_multiprocess_wal,
            "family reads must disable multiprocess_wal (snapshot semantics)"
        );
    }

    /// The forensic-family name writers (`quarantine_family`, the pre-reindex
    /// snapshot helper) must produce names the debug CLI's `--family` parser
    /// accepts — a writer change that breaks the listing contract fails here,
    /// not against a stale literal lock.
    #[test]
    fn family_writer_names_parse_via_debug_parser() {
        let dir = tempfile::TempDir::new().unwrap();
        let db_path = dir.path().join("board.db");
        let wal = dir.path().join("board.db-wal");
        let tshm = dir.path().join("board.db-tshm");
        for f in [&db_path, &wal, &tshm] {
            std::fs::write(f, b"x").unwrap();
        }

        // Quarantine: move db + wal + tshm aside; every moved base name must
        // be a valid `--family` id (the listing/query contract).
        assert!(
            quarantine_family(
                &db_path,
                &[(&db_path, ""), (&wal, "-wal"), (&tshm, "-tshm")],
            ),
            "quarantine writer must move every existing source"
        );
        let moved: Vec<String> = std::fs::read_dir(dir.path())
            .unwrap()
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .filter(|n| n.contains("board.db.quarantine-"))
            .collect();
        assert_eq!(moved.len(), 3, "db + wal + tshm moved: {moved:?}");
        for name in moved {
            let base = ["-wal", "-tshm"]
                .iter()
                .find_map(|s| name.strip_suffix(s))
                .unwrap_or(&name);
            let meta = crate::debug::parse_family_name(base).unwrap_or_else(|| {
                panic!("quarantine writer name must parse as a family id: {name}")
            });
            assert_eq!(meta.store, "board");
            assert_eq!(meta.kind, crate::debug::FamilyKind::Quarantine);
        }

        // Pre-reindex snapshot: the writer helper's base name must parse too.
        let snap = pre_reindex_snapshot_path(&db_path);
        let meta = crate::debug::parse_family_name(snap.file_name().unwrap().to_str().unwrap())
            .expect("pre-reindex snapshot name must parse as a family id");
        assert_eq!(meta.kind, crate::debug::FamilyKind::PreReindex);
    }

    #[test]
    fn builder_mapping_matches_experimental_features() {
        // Verify that the field-by-field mapping in Connection::open (simulated
        // below) enables exactly the features listed in EXPERIMENTAL_FEATURES.
        //
        // This catches reverse-drift: if someone adds a feature to
        // experimental_database_opts() and EXPERIMENTAL_FEATURES but forgets
        // the experimental_*() mapping line in Connection::open, the test
        // fails because the simulated mapping doesn't enable it.
        //
        // To add a new feature:
        //   1. Add with_*() to experimental_database_opts()
        //   2. Add experimental_*() to Connection::open
        //   3. Add feature name to EXPERIMENTAL_FEATURES
        //   4. Add the if-guard below to this test's mapping table
        let opts = experimental_database_opts();

        // ── Simulate the builder mapping from Connection::open ────────────
        // Must stay in sync with the .experimental_*() calls there.
        let mut mapped: Vec<&str> = Vec::new();
        if opts.enable_index_method {
            mapped.push("index_method");
        }
        if opts.enable_multiprocess_wal {
            mapped.push("multiprocess_wal");
        }
        // Note: enable_views maps to experimental_materialized_views() but
        // is not an active feature — no if-guard needed.
        // Note: enable_autovacuum and unsafe_testing have no Builder
        // experimental_*() equivalent — they cannot be mapped here.
        // ─────────────────────────────────────────────────────────────────

        mapped.sort_unstable();
        let mut expected: Vec<&str> = EXPERIMENTAL_FEATURES.to_vec();
        expected.sort_unstable();

        assert_eq!(
            mapped, expected,
            "Connection::open builder mapping enables features that differ from \
             EXPERIMENTAL_FEATURES.\n\
             If you added a feature: add the experimental_*() guard above AND \
             add it to Connection::open.\n\
             If you removed a feature: remove it from both places.\n\
             See EXPERIMENTAL_FEATURES docs for naming asymmetries."
        );
    }

    /// Guard: raw `turso::Builder` usage is confined to the persistence module
    /// (`src/turso.rs`) and the debug CLI (`src/debug.rs`).
    ///
    /// All database access must go through [`crate::turso::Connection`], which
    /// centralises the experimental feature flags, busy timeout, mutex
    /// serialization, and dangling-transaction handling. The debug CLI is the
    /// documented exception: it needs the upstream lazy `Rows` iterator for
    /// `mahbot debug` queries.
    ///
    /// This is a source-scanning tripwire, not a security boundary: any new
    /// module that opens a database via `turso::Builder` / `Builder::new_local`
    /// directly fails this test with a pointer to the canonical path. (The
    /// cross-process repro bench in `benches/` is outside this scan and is a
    /// documented exception — it needs raw builder access for the two-writer
    /// harness.)
    #[test]
    fn raw_builder_usage_is_confined_to_persistence_and_debug() {
        const PATTERNS: [&str; 2] = ["turso::Builder", "Builder::new_local"];
        const ALLOWED: [&str; 2] = ["src/turso.rs", "src/debug.rs"];

        fn collect_rs_files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
            for entry in std::fs::read_dir(dir).expect("read src directory") {
                let path = entry.expect("read directory entry").path();
                if path.is_dir() {
                    collect_rs_files(&path, out);
                } else if path.extension().is_some_and(|e| e == "rs") {
                    out.push(path);
                }
            }
        }

        let manifest_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let mut files = Vec::new();
        collect_rs_files(&manifest_dir.join("src"), &mut files);

        let mut violations: Vec<(String, &'static str)> = Vec::new();
        for file in files {
            let rel = file
                .strip_prefix(manifest_dir)
                .expect("source files live under the manifest dir")
                .to_string_lossy()
                .to_string();
            if ALLOWED.contains(&rel.as_str()) {
                continue;
            }
            let content = std::fs::read_to_string(&file).expect("read source file");
            // Skip line comments (// and ///) so doc comments mentioning the
            // builder don't false-positive — the tripwire targets actual code
            // usage. Block comments are not handled (best-effort by design).
            let code_only: String = content
                .lines()
                .filter(|line| !line.trim_start().starts_with("//"))
                .collect::<Vec<_>>()
                .join("\n");
            for pattern in PATTERNS {
                if code_only.contains(pattern) {
                    violations.push((rel.clone(), pattern));
                }
            }
        }

        assert!(
            violations.is_empty(),
            "raw turso::Builder usage outside the persistence module and debug CLI.\n\
             All database access must go through crate::turso::Connection (turso.rs);\n\
             the debug CLI (debug.rs) is the only documented exception.\n\
             Violations: {violations:#?}"
        );
    }

    /// Guard: the in-memory temp-store setting is applied on BOTH turso
    /// opening paths.
    ///
    /// P0 incident (2026-08-17): a missing pinned temp root failed every
    /// statement whose plan contains eager temp-file opcodes (RETURNING
    /// buffers, ORDER BY+LIMIT heap sorts, DISTINCT, IN-subqueries, compound
    /// SELECTs, window functions, CREATE INDEX) and every overflowing
    /// sorter/hash-join spill with "I/O error (tempdir): entity not found" —
    /// turso_core's tempdir creation has no parent-creation and no fallback
    /// chain. The fix: every application connection runs with
    /// `PRAGMA temp_store = MEMORY` (the engine's `TempStore::Memory`), so
    /// intermediate query structures stay in RAM and a missing temp root
    /// cannot affect any turso operation.
    ///
    /// The service connection factory (`Connection::open` in src/turso.rs)
    /// and the debug CLI's separate read-only path (`connect_readonly` in
    /// src/debug.rs) do NOT share an opening path — both must carry the
    /// PRAGMA. If either regresses, the guarantee silently breaks; this
    /// source-scanning tripwire fails with a pointer to the requirement.
    /// The bench harnesses in `benches/` are outside this scan (documented
    /// repro exceptions, not application code).
    #[test]
    fn in_memory_temp_store_applied_on_both_opening_paths() {
        const PRAGMA: &str = "PRAGMA temp_store = MEMORY";
        // (file, symbol that must carry the PRAGMA)
        const EXPECTED: [(&str, &str); 2] = [
            ("src/turso.rs", "impl Connection"),
            ("src/debug.rs", "fn connect_readonly"),
        ];

        let manifest_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let mut violations: Vec<String> = Vec::new();
        for (file, carrier) in EXPECTED {
            let path = manifest_dir.join(file);
            let content =
                std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {file}: {e}"));
            // Skip line comments so doc comments mentioning the PRAGMA don't
            // false-positive — the tripwire targets actual code usage (same
            // best-effort approach as the raw-builder guard).
            let code_only: String = content
                .lines()
                .filter(|line| !line.trim_start().starts_with("//"))
                .collect::<Vec<_>>()
                .join("\n");
            if !code_only.contains(PRAGMA) {
                violations.push(format!(
                    "{file} ({carrier}) no longer applies the in-memory temp-store \
                     setting (missing `{PRAGMA}` in code).\n\
                     The missing-temp-root guarantee requires EVERY turso connection \
                     the application opens — service factory AND debug CLI — to run \
                     with in-memory temp storage."
                ));
            }
        }

        assert!(
            violations.is_empty(),
            "in-memory temp-store regression guard failed:\n{}",
            violations.join("\n\n")
        );
    }

    #[test]
    fn test_sanitize_fts_query() {
        let cases = [
            // Basic cases
            ("hello world", "hello world"),
            // All Tantivy special chars → empty string (triggers caller short-circuit)
            ("+-~", ""),
            // Curly braces and backticks stripped, $ preserved
            ("`Hello ${name}`", "Hello $ name"),
            // Email preserved
            (
                "contact user@example.com now",
                "contact user@example.com now",
            ),
            // Punctuation preserved except apostrophe (phrase delimiter)
            (
                "hello, world! How's it going?",
                "hello, world! How s it going?",
            ),
            // Non-special punctuation all preserved
            ("!@#$%", "!@#$%"),
            // Identifiers with underscores preserved
            ("my_function", "my_function"),
            // Tags preserved
            ("#381", "#381"),
            // Version strings preserved
            ("v1.2.3", "v1.2.3"),
            // Paths preserved (no leading slash)
            ("feature/x", "feature/x"),
            // Leading slash stripped to prevent Tantivy regex parsing
            ("/something", "something"),
            // Leading - (MustNot) stripped via special-char → space → split
            ("-hello", "hello"),
            // Hyphen in middle becomes space (word boundary)
            ("hello-world", "hello world"),
            // Leading + (Must) stripped
            ("+term", "term"),
            // Apostrophe in word stripped (prevents phrase parsing)
            ("don't", "don t"),
        ];
        for (input, expected) in cases {
            assert_eq!(sanitize_fts_query(input), expected, "input: {input:?}");
        }
    }

    // ── parse_utc_timestamp tests ─────────────────────────────────────

    #[test]
    fn test_parse_utc_timestamp() {
        let valid_cases = [
            ("2024-01-15T10:30:00Z", "2024-01-15T10:30:00+00:00"),
            ("2024-06-15T14:30:00+05:00", "2024-06-15T09:30:00+00:00"),
            ("2024-12-25T20:00:00-08:00", "2024-12-26T04:00:00+00:00"),
        ];
        for (input, expected) in valid_cases {
            let ts = parse_utc_timestamp(input)
                .unwrap_or_else(|e| panic!("parse_utc_timestamp({input:?}) failed: {e}"));
            assert_eq!(ts.to_rfc3339(), expected, "input: {input:?}");
        }

        for invalid in ["garbage", "", "2024-01-15"] {
            assert!(
                parse_utc_timestamp(invalid).is_err(),
                "expected error for: {invalid:?}",
            );
        }
    }

    // ── quick_check tests ────────────────────────────────────────────

    /// Verify that checkpoint reports a complete outcome on a healthy database
    /// for both TRUNCATE and PASSIVE modes, and that the outcome is parsed from
    /// the result row (not silently discarded as success).
    #[tokio::test]
    async fn test_checkpoint_reports_complete_outcome() {
        let tmp = tempfile::TempDir::new().expect("temp dir for test");
        let conn = Connection::open(tmp.path().join("test.db").as_path())
            .await
            .expect("open test database");
        conn.execute(
            "CREATE TABLE IF NOT EXISTS _test (id INTEGER PRIMARY KEY, val TEXT NOT NULL)",
            (),
        )
        .await
        .expect("create test table");
        conn.execute("INSERT INTO _test (id, val) VALUES (1, 'hello')", ())
            .await
            .expect("insert test row");

        for mode in ["checkpoint", "checkpoint_passive"] {
            let outcome = match mode {
                "checkpoint" => conn.checkpoint().await,
                _ => conn.checkpoint_passive().await,
            }
            .expect("checkpoint should succeed on a healthy database");
            assert!(
                outcome.is_complete(),
                "{mode} outcome must be complete: {outcome:?}"
            );
        }
    }

    /// Exercise the busy/partial detection directly: `is_complete` is false for
    /// a busy condition or when WAL frames remain uncheckpointed (`log >
    /// checkpointed`), and true only for a fully backfilled, non-busy outcome.
    #[test]
    fn checkpoint_outcome_completeness_predicate() {
        let complete = CheckpointOutcome {
            busy: false,
            log_frames: 0,
            checkpointed_frames: 0,
        };
        assert!(complete.is_complete());

        let busy = CheckpointOutcome {
            busy: true,
            log_frames: 0,
            checkpointed_frames: 0,
        };
        assert!(!busy.is_complete(), "busy outcome must be incomplete");

        let partial = CheckpointOutcome {
            busy: false,
            log_frames: 12,
            checkpointed_frames: 5,
        };
        assert!(
            !partial.is_complete(),
            "uncheckpointed WAL frames must be incomplete"
        );
    }

    /// Verify that quick_check returns Ok on a healthy (empty) database.
    #[tokio::test]
    async fn test_quick_check_passes_on_healthy_db() {
        let tmp = tempfile::TempDir::new().expect("temp dir for test");
        let conn = Connection::open(tmp.path().join("test.db").as_path())
            .await
            .expect("open test database");
        conn.quick_check()
            .await
            .expect("quick_check should pass on a healthy empty database");
    }

    /// The known FTS-dir count-mismatch false positive is masked, while other
    /// messages naming the same internal index (genuine corruption) are not.
    #[test]
    fn known_fts_dir_false_positive_classification() {
        let fp = "wrong # of entries in index __turso_internal_fts_dir_idx_tickets_title_fts_key";
        assert!(fp.contains(KNOWN_FTS_DIR_COUNT_FALSE_POSITIVE));

        let genuine_missing =
            "row 5 missing from index __turso_internal_fts_dir_idx_tickets_title_fts_key";
        let genuine_unique =
            "non-unique entry in index __turso_internal_fts_dir_idx_tickets_title_fts_key";
        assert!(!genuine_missing.contains(KNOWN_FTS_DIR_COUNT_FALSE_POSITIVE));
        assert!(!genuine_unique.contains(KNOWN_FTS_DIR_COUNT_FALSE_POSITIVE));
    }

    /// A Structural boot diagnosis makes `open_store` quarantine the artifact
    /// family and recreate a fresh store (the unified recreate path).
    #[tokio::test]
    async fn open_store_recreates_structural_store() {
        let tmp = tempfile::TempDir::new().expect("temp dir for test");
        let root = tmp.path();
        let db_path = store_db_path(root, "board");
        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
        // Truncated main DB header → structural.
        std::fs::write(&db_path, [0u8; 64]).unwrap();
        std::fs::write(format!("{}-tshm", db_path.display()), [0u8; 32]).unwrap();
        crate::wal_guard::set_boot_diagnosis(&db_path, crate::wal_guard::BootDiagnosis::Structural);

        let conn = open_store(root, "board", "CREATE TABLE IF NOT EXISTS t (id INTEGER);")
            .await
            .expect("structural store must be recreated, not fail boot");
        let rows: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='t'",
                (),
                |r| r.get::<i64>(0),
            )
            .await
            .expect("schema applied on the recreated store");
        assert_eq!(rows, 1, "recreated store must carry the schema");
        let quarantined = std::fs::read_dir(db_path.parent().unwrap())
            .unwrap()
            .filter_map(std::result::Result::ok)
            .any(|e| e.file_name().to_string_lossy().contains("quarantine-"));
        assert!(
            quarantined,
            "artifact family must be quarantined (forensic copy)"
        );
    }

    /// The quick_check desync parser names exactly the non-FTS btree index,
    /// and refuses the overflow-aliasing / unreadable shapes (never REINDEX).
    #[test]
    fn desynced_index_name_parses_quick_check_shapes() {
        assert_eq!(
            desynced_index_name("wrong # of entries in index idx_tickets_phase"),
            Some("idx_tickets_phase")
        );
        assert_eq!(
            desynced_index_name("wrong # of entries in index a; row 5 missing"),
            Some("a")
        );
        // FTS false positive — masked upstream, never REINDEXed here.
        assert!(
            desynced_index_name(
                "wrong # of entries in index __turso_internal_fts_dir_idx_tickets_title_fts_key"
            )
            .is_some()
        );
        // Overflow-aliasing and unreadable-table signatures.
        assert!(desynced_index_name("Page referenced multiple times: page 167772160").is_none());
        assert!(desynced_index_name("short read on page 12").is_none());
        assert!(desynced_index_name("row 5 missing from index idx_x").is_none());
        assert!(desynced_index_name("ok").is_none());
    }

    /// The overflow-aliasing veto is structural: a named index desync in an
    /// earlier row must NOT lead to a REINDEX when a later row reports
    /// overflow-aliasing (the operation that bakes the worse error). The
    /// matcher covers turso's real message shape ("Page N referenced multiple
    /// times (references=[...], page_category=...)") and the baked short-read
    /// (page 0x0A000000) — both route to the data-preserving rebuild, while a
    /// genuine short read on a real page stays unreadable (recreate).
    #[test]
    fn overflow_aliasing_vetoes_reindex_across_rows() {
        assert!(matches!(
            classify_repair_target(&[
                "wrong # of entries in index idx_phase".to_string(),
                "Page referenced multiple times: page 167772160".to_string(),
            ]),
            RepairTarget::OverflowAliasing
        ));
        // turso's actual message shape (index leaves sharing an overflow page).
        assert!(matches!(
            classify_repair_target(&[
                "*** in database main ***\nPage 871 referenced multiple times \
                 (references=[3, 3], page_category=Normal)"
                    .to_string(),
                "wrong # of entries in index idx_t_v".to_string(),
            ]),
            RepairTarget::OverflowAliasing
        ));
        assert!(matches!(
            classify_repair_target(&[
                "Page referenced multiple times: page 5".to_string(),
                "wrong # of entries in index idx_phase".to_string(),
            ]),
            RepairTarget::OverflowAliasing
        ));
        // The baked overflow-aliasing signature routes to the rebuild, not
        // to the recreate-whole path.
        assert!(matches!(
            classify_repair_target(&["short read on page 167772160".to_string()]),
            RepairTarget::OverflowAliasing
        ));
        assert!(matches!(
            classify_repair_target(&[
                "wrong # of entries in index idx_phase".to_string(),
                "short read on page 12".to_string(),
            ]),
            RepairTarget::Unreadable
        ));
        assert!(matches!(
            classify_repair_target(&["short read on page 12".to_string()]),
            RepairTarget::Unreadable
        ));
        assert!(matches!(
            classify_repair_target(&["wrong # of entries in index idx_phase".to_string()]),
            RepairTarget::Index(name) if name == "idx_phase"
        ));
        assert!(matches!(
            classify_repair_target(&["row 5 missing from index idx_x".to_string()]),
            RepairTarget::Unknown
        ));
    }

    /// turso rolls back DDL: the class-B DROP+CREATE fallback wraps the pair
    /// in a transaction because `execute_batch` is autocommit per statement —
    /// a failed rebuild (constraint violation) must leave the original index
    /// in place, not silently dropped. Covers both rollback paths: the
    /// explicit one and the production failure path (guard dropped without
    /// commit → deferred ROLLBACK on the next connection op).
    #[tokio::test]
    async fn transactional_ddl_rollback_preserves_dropped_index() {
        let tmp = tempfile::TempDir::new().unwrap();
        let conn = open_with_schema(
            &tmp.path().join("t.db"),
            "CREATE TABLE t (a TEXT, b TEXT); \
             INSERT INTO t VALUES ('1', 'x'), ('1', 'y'); \
             CREATE UNIQUE INDEX u ON t(b);",
        )
        .await
        .expect("open test store");
        // The failed rebuild: DROP the existing index, then CREATE a UNIQUE
        // index on the duplicated column — constraint violation aborts the
        // batch; the rollback must restore `u`.
        let tx = conn.begin_tx().await.expect("begin tx");
        let err = tx
            .execute_batch("DROP INDEX u; CREATE UNIQUE INDEX u2 ON t(a);")
            .await
            .expect_err("CREATE UNIQUE on duplicated data must fail");
        assert!(
            format!("{err}").to_lowercase().contains("unique"),
            "expected a constraint violation, got: {err}"
        );
        tx.rollback().await.expect("rollback");
        // Production path: the guard is dropped without commit/rollback after
        // the failure — the dangling flag makes the next connection op
        // (query → lock_and_cleanup) issue the deferred ROLLBACK.
        {
            let tx = conn.begin_tx().await.expect("begin tx");
            let err = tx
                .execute_batch("DROP INDEX u; CREATE UNIQUE INDEX u2 ON t(a);")
                .await
                .expect_err("CREATE UNIQUE on duplicated data must fail");
            assert!(
                format!("{err}").to_lowercase().contains("unique"),
                "expected a constraint violation, got: {err}"
            );
        }
        let names: Vec<String> = conn
            .query_map_strict(
                "SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('u', 'u2')",
                (),
                |row| row.get::<String>(0),
            )
            .await
            .expect("query sqlite_master");
        assert_eq!(names, vec!["u".to_string()], "original index must survive");
        // The restored index is functional and the store is consistent.
        conn.quick_check()
            .await
            .expect("store must be consistent after the rollback");
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM t INDEXED BY u", (), |row| {
                row.get::<i64>(0)
            })
            .await
            .expect("INDEXED BY must resolve");
        assert_eq!(n, 2);
    }

    // ── Overflow-aliasing rebuild (data-preserving migration) ─────────

    /// Synthesize the overflow-aliasing corruption on a closed store: patch
    /// one index leaf cell's overflow pointer to another cell's overflow page
    /// so two index entries share an overflow chain (turso's
    /// "Page N referenced multiple times"). The cells are located by their
    /// value prefix preceded by the record header `[hdr][text serial][rowid
    /// serial]` — the 2007-byte text serial is `[0x9f, 0x3b]`, and turso
    /// spills index payloads at the min threshold (`((usable-12)*32/255)-23`
    /// = 489 bytes for a 4096 page), so the 4-byte overflow pointer sits
    /// `local - 4` bytes after the value prefix.
    fn synthesize_overflow_aliasing(db_path: &Path) -> u32 {
        let file = std::fs::read(db_path).unwrap();
        let page_size = u16::from_be_bytes([file[16], file[17]]) as usize;
        let local = ((page_size - 12) * 32 / 255) - 23;
        let find = |file: &[u8], prefix: &[u8], rowid_serial: u8| -> (usize, u32) {
            for i in 0..file.len().saturating_sub(prefix.len() + 12) {
                // The index record header is [hdr=0x04][text serial 0x9f 0x3b]
                // [rowid serial] — the table record's header differs (more
                // columns → different hdr byte and serial order), so the
                // 4-byte pattern pins the index cell (leaf or interior
                // separator, both use the same record layout).
                if &file[i..i + prefix.len()] == prefix
                    && file[i - 4..i] == [0x04, 0x9f, 0x3b, rowid_serial]
                {
                    let ptr =
                        u32::from_be_bytes(file[i + local - 4..i + local].try_into().unwrap());
                    return (i, ptr);
                }
            }
            panic!(
                "index cell for {:?} not found",
                String::from_utf8_lossy(prefix)
            );
        };
        let (_, shared) = find(&file, b"v04000-", 0x02); // rowid 4001 → 16-bit serial
        let (patch_at, _) = find(&file, b"v00000-", 0x09); // rowid 1 → constant serial 9
        let mut file = file;
        file[patch_at + local - 4..patch_at + local].copy_from_slice(&shared.to_be_bytes());
        std::fs::write(db_path, &file).unwrap();
        shared
    }

    /// Build a store whose index carries overflow pages (long values), then
    /// checkpoint via a fresh connection (the writer connection's own
    /// checkpoint does not flush its last frames) so the main file holds
    /// every page the surgery will patch.
    async fn build_aliasing_candidate(
        db_path: &Path,
        schema: &str,
        extra_row: Option<(&str, i64)>,
    ) {
        {
            let conn = open_with_schema(db_path, schema).await.unwrap();
            for i in 0..5000i32 {
                let mut big = format!("v{i:05}-");
                big.push_str(&"q".repeat(2000));
                conn.execute(
                    "INSERT INTO t (v, n) VALUES (?1, 1);",
                    turso::params![big.clone()],
                )
                .await
                .unwrap();
            }
            if let Some((v, n)) = extra_row {
                conn.execute("PRAGMA ignore_check_constraints = ON;", ())
                    .await
                    .unwrap();
                conn.execute(
                    "INSERT INTO t (v, n) VALUES (?1, ?2);",
                    turso::params![v, n],
                )
                .await
                .unwrap();
                conn.execute("PRAGMA ignore_check_constraints = OFF;", ())
                    .await
                    .unwrap();
            }
            drop(conn);
        }
        {
            let conn = Connection::open(db_path).await.unwrap();
            conn.checkpoint().await.unwrap();
            drop(conn);
        }
    }

    /// The rebuild works on the FTS-bearing store shape (board's
    /// `CREATE INDEX ... USING fts`): turso's protected
    /// `__turso_internal_fts_dir_*` backing tables are excluded from the copy
    /// and post-swap verification, the DDL replay rebuilds the FTS index, and
    /// the fresh store's quick_check masks the known FTS count-mismatch false
    /// positive. The FTS index lives on a separate short-value table (real
    /// titles), not the overflow-aliased long-value column.
    ///
    /// This test is `#[ignore]` by default because it performs real overflow-aliasing page surgery on a multi-MB database fixture (~4 s). Run it
    /// explicitly with:
    ///
    /// ```sh
    /// cargo test overflow_aliasing_rebuild_preserves_fts_store -- --ignored --nocapture
    /// ```
    #[ignore = "performs real overflow-page DB surgery on a multi-MB fixture (~4 s); runs only when explicitly invoked"]
    #[tokio::test]
    async fn overflow_aliasing_rebuild_preserves_fts_store() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path();
        let db_path = store_db_path(root, "board");
        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
        let schema = "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v TEXT NOT NULL, \
                      n INTEGER NOT NULL DEFAULT 1); \
                      CREATE INDEX IF NOT EXISTS idx_t_v ON t(v); \
                      CREATE TABLE IF NOT EXISTS ft (title TEXT NOT NULL); \
                      CREATE INDEX IF NOT EXISTS idx_ft_fts ON ft USING fts (title) \
                      WITH (tokenizer = 'ngram');";
        build_aliasing_candidate(&db_path, schema, None).await;
        // FTS content (the count-mismatch false positive only appears once
        // the index has rows), checkpointed into the main file before the
        // page surgery.
        {
            let conn = open_with_schema(&db_path, schema).await.unwrap();
            for i in 0..5 {
                conn.execute(
                    "INSERT INTO ft (title) VALUES (?1);",
                    turso::params![format!("ticket title {i}")],
                )
                .await
                .unwrap();
            }
            drop(conn);
        }
        {
            let conn = Connection::open(&db_path).await.unwrap();
            conn.checkpoint().await.unwrap();
            drop(conn);
        }
        let shared = synthesize_overflow_aliasing(&db_path);
        assert!(shared > 0, "surgery must reference a real overflow page");

        let conn = open_and_repair(&db_path, "board", schema)
            .await
            .expect("overflow-aliasing repair must succeed on an FTS store");
        conn.quick_check()
            .await
            .expect("rebuilt store must pass quick_check");
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM t", (), |r| r.get::<i64>(0))
            .await
            .unwrap();
        assert_eq!(count, 5000, "all rows must survive the rebuild");
        let idx: i64 = conn
            .query_row("SELECT COUNT(*) FROM t INDEXED BY idx_t_v", (), |r| {
                r.get::<i64>(0)
            })
            .await
            .unwrap();
        assert_eq!(idx, 5000, "rebuilt index must be valid and complete");
        let ft: i64 = conn
            .query_row("SELECT COUNT(*) FROM ft", (), |r| r.get::<i64>(0))
            .await
            .unwrap();
        assert_eq!(ft, 5, "FTS table data must survive the rebuild");
        // The rebuilt FTS dir must answer MATCH queries — quick_check masks
        // the __turso_internal_fts_dir_ count row for a broken index too, so
        // the MATCH assertion is the only check that pins the rebuild.
        let matched: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM ft WHERE title MATCH 'title'",
                (),
                |r| r.get::<i64>(0),
            )
            .await
            .unwrap();
        assert_eq!(matched, 5, "rebuilt FTS index must answer MATCH queries");
        let quarantined = std::fs::read_dir(db_path.parent().unwrap())
            .unwrap()
            .filter_map(std::result::Result::ok)
            .any(|e| e.file_name().to_string_lossy().contains("quarantine-"));
        assert!(
            quarantined,
            "original family must be quarantined (forensic record)"
        );
        // A write on the rebuilt store stays clean (no freelist landmine).
        conn.execute("INSERT INTO t (v, n) VALUES ('post-rebuild', 1);", ())
            .await
            .unwrap();
        conn.quick_check()
            .await
            .expect("rebuilt store must stay clean after a write");
    }

    /// A constraint violation during the data copy is a data-integrity
    /// finding, not corruption: the rebuild aborts report-only (no silent
    /// recreate, no quarantine) and the original store — with its readable
    /// data — is preserved for operator review.
    ///
    /// This test is `#[ignore]` by default because it performs real overflow-aliasing page surgery on a multi-MB database fixture (~4 s). Run it
    /// explicitly with:
    ///
    /// ```sh
    /// cargo test overflow_aliasing_rebuild_constraint_finding_aborts -- --ignored --nocapture
    /// ```
    #[ignore = "performs real overflow-page DB surgery on a multi-MB fixture (~4 s); runs only when explicitly invoked"]
    #[tokio::test]
    async fn overflow_aliasing_rebuild_constraint_finding_aborts() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path();
        let db_path = store_db_path(root, "board");
        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
        let schema = "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, v TEXT NOT NULL, \
                      n INTEGER NOT NULL CHECK (n < 50)); \
                      CREATE INDEX IF NOT EXISTS idx_t_v ON t(v);";
        // Row with n=99 violates the CHECK (written with constraints ignored).
        build_aliasing_candidate(&db_path, schema, Some(("v09999x", 99))).await;
        synthesize_overflow_aliasing(&db_path);

        let conn = open_and_repair(&db_path, "board", schema)
            .await
            .expect("aborted rebuild must still open the store (report-only)");
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM t", (), |r| r.get::<i64>(0))
            .await
            .unwrap();
        assert_eq!(
            count, 5001,
            "original data must be preserved, not recreated"
        );
        let quarantined = std::fs::read_dir(db_path.parent().unwrap())
            .unwrap()
            .filter_map(std::result::Result::ok)
            .any(|e| e.file_name().to_string_lossy().contains("quarantine-"));
        assert!(
            !quarantined,
            "a constraint finding must not quarantine/recreate the store"
        );
    }

    /// The rebuild works on the real AUTOINCREMENT store shape (logs,
    /// chat_history, sessions): turso's protected `__turso_internal_seq_*`
    /// backing tables are excluded from the copy and the post-swap
    /// verification, `sqlite_sequence` is carried over, and the fresh
    /// watermark advances past the copied max ids via the explicit-id copy.
    /// Rows 4001..5000 are deleted before the repair (retention) so the old
    /// watermark (5000) exceeds the surviving max id (4000) — the only shape
    /// where the carry-over matters: without it, the next auto-id would
    /// re-issue 4001.
    ///
    /// This test is `#[ignore]` by default because it performs real overflow-aliasing page surgery on a multi-MB database fixture (~4 s). Run it
    /// explicitly with:
    ///
    /// ```sh
    /// cargo test overflow_aliasing_rebuild_preserves_autoincrement_store -- --ignored --nocapture
    /// ```
    #[ignore = "performs real overflow-page DB surgery on a multi-MB fixture (~4 s); runs only when explicitly invoked"]
    #[tokio::test]
    async fn overflow_aliasing_rebuild_preserves_autoincrement_store() {
        let tmp = tempfile::TempDir::new().unwrap();
        let root = tmp.path();
        let db_path = store_db_path(root, "board");
        std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
        let schema = "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY AUTOINCREMENT, \
                      v TEXT NOT NULL, n INTEGER NOT NULL DEFAULT 1); \
                      CREATE INDEX IF NOT EXISTS idx_t_v ON t(v);";
        build_aliasing_candidate(&db_path, schema, None).await;
        let shared = synthesize_overflow_aliasing(&db_path);
        assert!(shared > 0, "surgery must reference a real overflow page");
        {
            let conn = Connection::open(&db_path).await.unwrap();
            conn.execute("DELETE FROM t WHERE id > 4000", ())
                .await
                .unwrap();
            drop(conn);
        }

        let conn = open_and_repair(&db_path, "board", schema)
            .await
            .expect("overflow-aliasing repair must succeed on an AUTOINCREMENT store");
        conn.quick_check()
            .await
            .expect("rebuilt store must pass quick_check");
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM t", (), |r| r.get::<i64>(0))
            .await
            .unwrap();
        assert_eq!(count, 4000, "all surviving rows must be preserved");
        let idx: i64 = conn
            .query_row("SELECT COUNT(*) FROM t INDEXED BY idx_t_v", (), |r| {
                r.get::<i64>(0)
            })
            .await
            .unwrap();
        assert_eq!(idx, 4000, "rebuilt index must be valid and complete");
        // The backing seq table is recreated by the DDL replay and advanced
        // by the explicit-id copy + carry-over — the next implicit-id insert
        // must be 5001, not a reused 4001.
        conn.execute("INSERT INTO t (v, n) VALUES ('auto-next', 1);", ())
            .await
            .unwrap();
        let max_id: i64 = conn
            .query_row("SELECT MAX(id) FROM t", (), |r| r.get::<i64>(0))
            .await
            .unwrap();
        assert_eq!(
            max_id, 5001,
            "AUTOINCREMENT must advance past the old watermark, never re-issue ids"
        );
        // The sqlite_sequence mirror is carried over from the old store and
        // follows the advanced watermark.
        let seq: i64 = conn
            .query_row(
                "SELECT seq FROM sqlite_sequence WHERE name = 't'",
                (),
                |r| r.get::<i64>(0),
            )
            .await
            .unwrap();
        assert_eq!(
            seq, 5001,
            "sqlite_sequence watermark must be carried over and advanced"
        );
        conn.quick_check()
            .await
            .expect("rebuilt store must stay clean after a write");
        let quarantined = std::fs::read_dir(db_path.parent().unwrap())
            .unwrap()
            .filter_map(std::result::Result::ok)
            .any(|e| e.file_name().to_string_lossy().contains("quarantine-"));
        assert!(
            quarantined,
            "original family must be quarantined (forensic record)"
        );
    }
}