mahbot 0.5.0

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

pub(crate) mod cdc;
pub mod checkpoint;
pub mod debug;
pub mod ipc;
pub(crate) mod migrations;
pub mod wal_guard;

// ── 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"];

/// Name (file stem) of the single consolidated domain database file that backs
/// the board, sessions, workspaces, users, config, chat_history, and alarms stores.
///
/// The domain stores were historically split across separate files; they
/// now share one file (and one shared [`Connection`]) so that cross-store
/// integrity is enforceable with real foreign keys and phase+stage updates are
/// atomic within a single transaction domain. The logs store remains a
/// separate file ([`LOG_DB_NAME`]).
///
/// `core` distinguishes this domain database from the separate logs store and
/// reads as the canonical core-domain file (`~/.mahbot/db/core.db`).
pub(crate) const CONSOLIDATED_DB_NAME: &str = "core";

/// Logical store names consolidated into [`CONSOLIDATED_DB_NAME`].
///
/// These are the names accepted by `mahbot debug --db <name>` and used to
/// resolve the consolidated file path; each maps to the single physical
/// [`CONSOLIDATED_DB_NAME`] file. The logs store is NOT in this list — it
/// remains its own file.
pub(crate) const DOMAIN_STORE_NAMES: &[&str] = &[
    "board",
    "sessions",
    "workspaces",
    "users",
    "config",
    "chat_history",
    "alarms",
];

/// Name (file stem) of the logs store — the one store that stays a separate
/// file/connection. Opened manually in [`crate::logs::init_tracing`] before the
/// other stores; its schema is owned by the catalog ([`crate::db::migrations`]).
pub(crate) const LOG_DB_NAME: &str = "logs";

/// The single shared consolidated domain connection, set once by
/// [`init_all_stores`]. Each domain store struct holds a clone of this
/// connection (same underlying turso connection — the shared transaction
/// domain). The logs store does NOT use this connection.
pub(crate) static DOMAIN_CONN: tokio::sync::OnceCell<Connection> =
    tokio::sync::OnceCell::const_new();

/// Return an iterator over all **physical, 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 **physical files** exist and
/// are checkpointed. After consolidation all the domain stores share ONE file
/// ([`CONSOLIDATED_DB_NAME`]) plus the separate logs file, so this yields
/// exactly two units. [`store_names`] returns the LOGICAL store names accepted
/// by `mahbot debug --db <name>` — a deliberately separate list, because
/// several logical names map to the same physical file.
pub(crate) fn iter_checkpoint_stores()
-> impl Iterator<Item = (&'static str, Option<&'static crate::db::Connection>)> {
    [
        (CONSOLIDATED_DB_NAME, DOMAIN_CONN.get()),
        (LOG_DB_NAME, crate::logs::LOG_STORE.get().map(|s| &s.conn)),
    ]
    .into_iter()
}

/// Return all canonical **logical** store names accepted by `mahbot debug --db`.
///
/// These are the names users address stores by (`board`, `sessions`, …,
/// `chat_history`) plus `logs`. Several map to the single consolidated file via
/// [`store_db_path`]. This is **not** derived from
/// [`iter_checkpoint_stores`] (which enumerates the *physical* files): the two
/// lists are intentionally separate — several logical names share one physical
/// file — so both must be updated when a store is added or consolidated.
///
/// Used by:
/// - `mahbot debug` — validates `--db` argument values.
/// - Callers that previously referenced `ALL_STORE_NAMES`.
///
/// # Consolidated layout
///
/// The domain stores (board, sessions, workspaces, users, config,
/// chat_history, alarms) are consolidated into ONE physical file
/// ([`CONSOLIDATED_DB_NAME`]); only the logs store is a separate file. This
/// function returns the LOGICAL store names so `--db board`, `--db sessions`,
/// etc. resolve to the single consolidated file via [`store_db_path`].
pub(crate) fn store_names() -> Vec<&'static str> {
    let mut names: Vec<&'static str> = DOMAIN_STORE_NAMES.to_vec();
    names.push(LOG_DB_NAME);
    names
}

/// All accepted database selector values for the read-only debug/query surface
/// (the CLI's `--db` argument and the Analyst `mahbot_debug` tool's `db`
/// parameter): the logical store names ([`store_names`]) plus the physical
/// consolidated file name ([`CONSOLIDATED_DB_NAME`]).
///
/// This is the single derivation of that accepted set — [`crate::db::debug`]'s
/// validator, CLI usage text, and the tool's model-facing schema all use it so
/// the three cannot drift.
pub(crate) fn debug_db_names() -> Vec<&'static str> {
    let mut names = store_names();
    names.push(CONSOLIDATED_DB_NAME);
    names
}

/// Initialize all database stores concurrently.
///
/// This is the canonical initialization path for the domain stores (board,
/// session, workspace, users, chat_history, config, alarms).  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.
///
/// # Consolidated layout (decision 1)
///
/// The domain stores ([`DOMAIN_STORE_NAMES`]) now share ONE file
/// ([`CONSOLIDATED_DB_NAME`]) and ONE shared [`Connection`] ([`DOMAIN_CONN`]) —
/// every store struct holds a clone of the same connection, so `begin_tx` on
/// any store acquires the same mutex and phase+stage updates are atomic within
/// a single transaction domain. The
/// consolidated schema + migrations are applied once via
/// [`open_consolidated_store`], and per-store post-open hooks (tickets FTS
/// index, admin-user seed) run idempotently before the cells are set.
pub async fn init_all_stores() -> anyhow::Result<()> {
    // Idempotent: the consolidated connection is set once. Re-entering (e.g.
    // a test that calls init_test_stores twice) must not re-open / re-migrate.
    if DOMAIN_CONN.get().is_some() {
        return Ok(());
    }

    let root = crate::config::CONFIG.global_storage_root();
    let conn = open_consolidated_store(&root).await?;

    // Run the idempotent per-store post-open hooks on the shared connection,
    // in a defined order, before exposing the store cells.
    let board = crate::pipeline::board::BoardStore { conn: conn.clone() };
    board.after_open().await?;
    let users = crate::users::UserStore { conn: conn.clone() };
    users.ensure_admin_user().await?;

    let session = crate::session::SessionStore { conn: conn.clone() };
    let workspace = crate::workspace::WorkspaceStore { conn: conn.clone() };
    let config = crate::config_db::ConfigStore { conn: conn.clone() };
    let chat_history = crate::channels::chat_history::ChatHistoryStore { conn: conn.clone() };
    let alarms = crate::alarms::AlarmStore { conn: conn.clone() };

    // Record the shared connection before setting any cell so a concurrent
    // re-entry observes it and short-circuits.
    let _ = DOMAIN_CONN.set(conn.clone());

    // Start the CDC drain loop once. It broadcasts ticket change events to the
    // GUI and the chronicle subscriber; resuming after a heal/recreate re-enables
    // capture in open_consolidated_store (above).
    crate::db::cdc::spawn_drainer(conn.clone());

    crate::pipeline::board::BOARD
        .set(board)
        .map_err(|_| anyhow::anyhow!("BOARD already initialized"))?;
    crate::session::SESSIONS
        .set(session)
        .map_err(|_| anyhow::anyhow!("SESSIONS already initialized"))?;
    crate::workspace::WORKSPACES
        .set(workspace)
        .map_err(|_| anyhow::anyhow!("WORKSPACES already initialized"))?;
    crate::users::USER_STORE
        .set(users)
        .map_err(|_| anyhow::anyhow!("USER_STORE already initialized"))?;
    crate::config_db::CONFIG_STORE
        .set(config)
        .map_err(|_| anyhow::anyhow!("CONFIG_STORE already initialized"))?;
    crate::channels::chat_history::CHAT_HISTORY
        .set(chat_history)
        .map_err(|_| anyhow::anyhow!("CHAT_HISTORY already initialized"))?;
    crate::alarms::ALARMS
        .set(alarms)
        .map_err(|_| anyhow::anyhow!("ALARMS already initialized"))?;

    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.
///
/// The daemon runs in **default single-process mode** — there is no
/// `multiprocess_wal` coordination; `-tshm` is a stale leftover from a
/// pre-removal multiprocess run and is never created in normal operation.
///
/// Forensic-family reads (`mahbot debug --family`, via `open_readonly` in
/// `debug.rs`) use this same opts: family snapshots are copied from live stores
/// (which are created with `index_method`), so they read with the same
/// single-process feature set.
///
/// # 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_index_method(true)
}

/// 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 ()`.
///
/// 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(", ")
}

/// A serialized handle to a turso connection.
///
/// 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 daemon runs in **default single-process mode**: there is no `.tshm`
/// coordination file, so no persistent sidecar fds are held.
///
/// # Shared-connection discipline (consolidated layout)
///
/// After consolidation all the domain stores hold a **clone of one** connection
/// (the same [`Mutex`]), so the mutex is NOT reentrant and is never shared
/// between independent locks. A code path that holds a [`TxGuard`] (which keeps
/// the mutex locked for the transaction's lifetime) MUST use `tx.query()` /
/// `tx.execute()` / `tx.begin_tx()` for any further access to that connection —
/// calling another store's `conn.query()` / `conn.execute()` / `conn.begin_tx()`
/// while holding the guard deadlocks (the guard's mutex is already held). Before
/// consolidation each store had its own mutex, so cross-store calls were safe;
/// they are now a single lock and must be routed through the guard.
#[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>,
}

/// Result of a read-only query: column names, row values, and whether the
/// row limit was hit (more rows exist).
pub(crate) struct ReadonlyRows {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<Value>>,
    pub truncated: bool,
}

/// 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_%`), turso's protected `__turso_internal_%`
/// (AUTOINCREMENT seq backing, FTS dir), and the engine-managed CDC tables
/// (`turso_cdc`, `turso_cdc_version`) — 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_%' \
     AND name NOT IN ('turso_cdc', 'turso_cdc_version')";

/// Best-effort removal of a rebuild temp family (main + sidecars). Single-process
/// mode never creates a `-tshm` sidecar, so only `-wal`/`-shm` are removed.
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()));
}

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

/// Run a read-only query against an already-locked turso connection, bounded
/// to `row_limit` rows using LIMIT+1 semantics. Caller is responsible for the
/// `PRAGMA query_only` guard and its reset.
async fn run_readonly_query(
    conn: &turso::Connection,
    sql: &str,
    params: impl IntoParams + Send + 'static,
    row_limit: usize,
) -> turso::Result<ReadonlyRows> {
    let mut rows = conn.query(sql, params).await?;
    let columns = rows.column_names();
    let col_count = columns.len();
    let mut result_rows = Vec::new();
    let mut count = 0usize;
    let mut truncated = false;
    while let Some(row) = rows.next().await? {
        if count >= row_limit {
            truncated = true;
            break;
        }
        let mut values = Vec::with_capacity(col_count);
        for idx in 0..col_count {
            values.push(row.get_value(idx)?);
        }
        result_rows.push(values);
        count += 1;
    }
    Ok(ReadonlyRows {
        columns,
        rows: result_rows,
        truncated,
    })
}

/// Number of retry attempts for the open-time WAL lock held by another process.
const OPEN_LOCK_RETRY_ATTEMPTS: usize = 10;

/// Delay between open-time WAL lock retries.
const OPEN_LOCK_RETRY_DELAY: Duration = Duration::from_millis(500);

/// Whether an error is the transient open-time WAL lock from [turso_core].
///
/// Turso's [`turso::Builder::build`] takes the fcntl `F_SETLK` lock on the
/// `-wal` file while the database is opened. When another process already
/// holds that lock (e.g. the previous daemon process during a self-update
/// restart), the open fails with a `WouldBlock` that [turso_core] folds into
/// [`turso::Error::Error`] via the catch-all `From` impl, surfacing as a
/// `"Locking error: Failed locking file '...'. File is locked by another
/// process"` message. [`Connection::open`]'s `busy_timeout` does **not** cover
/// this open-time lock — it only governs per-statement waits after the handle
/// is connected.
///
/// The predicate is deliberately narrow: it requires both the `"Locking
/// error:"` and `"File is locked by another process"` substrings, so it never
/// treats generic `"database is locked"` / "table is locked" busy errors
/// (which have their own retry/backoff semantics) or unrelated corruption or
/// I/O messages as this transient condition.
///
/// [turso_core]: https://github.com/tursodatabase/turso
fn is_open_time_lock_error(err: &turso::Error) -> bool {
    matches!(
        err,
        turso::Error::Error(msg)
            if msg.contains("Locking error:") && msg.contains("File is locked by another process")
    )
}

/// Retries `attempt` only for the transient open-time WAL lock.
///
/// The previous process still holds the `-wal` lock briefly during a
/// self-update restart; this bounded retry (a few seconds total) absorbs that
/// race without masking any other open failure. `attempts` is the number of
/// retries *after* the first try, so the maximum number of total tries is
/// `attempts + 1`. A non-[open-time lock] error is returned on the first
/// encounter and never retried.
///
/// [open-time lock]: is_open_time_lock_error
async fn with_open_lock_retry<T, F, Fut>(
    attempts: usize,
    delay: Duration,
    mut attempt: F,
) -> Result<T, turso::Error>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, turso::Error>>,
{
    let mut remaining = attempts;
    loop {
        match attempt().await {
            Ok(value) => return Ok(value),
            Err(err) if is_open_time_lock_error(&err) => {
                warn!(
                    attempt = attempts - remaining + 1,
                    total_attempts = attempts + 1,
                    error = ?err,
                    "database open blocked by another process (open-time WAL lock)"
                );
                if remaining == 0 {
                    return Err(err);
                }
                tokio::time::sleep(delay).await;
                remaining -= 1;
            }
            Err(err) => return Err(err),
        }
    }
}

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 = with_open_lock_retry(OPEN_LOCK_RETRY_ATTEMPTS, OPEN_LOCK_RETRY_DELAY, || async {
            Builder::new_local(path_str)
                .experimental_index_method(opts.enable_index_method)
                .build()
                .await
        })
        .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)")?;
        Ok(Self {
            conn: Arc::new(tokio::sync::Mutex::new(conn)),
            has_dangling_tx: Arc::new(AtomicBool::new(false)),
        })
    }

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

    /// Execute a write via the engine's prepared-statement cache (turso's
    /// `execute` re-parses SQL on every call; hot loops use this instead). See
    /// [`Self::query_cached`] for the cache contract.
    pub(crate) async fn execute_cached(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<u64> {
        let conn = self.lock_and_cleanup().await;
        let mut stmt = conn.prepare_cached(sql).await?;
        stmt.execute(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
    }

    /// Execute a read-only query with engine-level write enforcement.
    ///
    /// Sets `PRAGMA query_only=1`, runs the query, and resets it to `0` — all
    /// under a single hold of the connection mutex. The reset is attempted on
    /// every outcome (including a query-engine panic) and is itself
    /// panic-guarded, so a reset-phase never unwinds the caller. A reset-phase
    /// panic, or a failure to roll a leaked transaction back, flags
    /// `has_dangling_tx` so the next access attempts a rollback; a non-panic
    /// `query_only=0` failure only surfaces an error. Either way a failed
    /// `query_only=0` leaves `query_only=1` (rejecting daemon writes) — loud,
    /// with no automatic recovery. Bounds the result set to `row_limit` rows
    /// using LIMIT+1 semantics (a result of exactly `row_limit` rows is NOT
    /// flagged truncated; the `row_limit + 1`-th row flags it). Returns column
    /// names + rows.
    pub(crate) async fn query_readonly(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
        row_limit: usize,
    ) -> turso::Result<ReadonlyRows> {
        let conn = self.lock_and_cleanup().await;
        conn.execute("PRAGMA query_only = 1;", ()).await?;
        let result = AssertUnwindSafe(run_readonly_query(&conn, sql, params, row_limit))
            .catch_unwind()
            .await;
        // Always restore the shared connection to its normal state (query_only
        // off, autocommit) so daemon writes are never left disabled. The
        // `query_only` reset happens first; then a transaction-control statement
        // (BEGIN/SAVEPOINT) that a raw IPC client could send — not re-validated
        // daemon-side — is rolled back to autocommit, since an open transaction
        // on the shared connection would poison subsequent daemon writes. The
        // whole reset is also panic-guarded: a turso engine panic here must not
        // unwind (leaving query_only=1), so the caller gets an error instead.
        let (reset, tx_reset) = match AssertUnwindSafe(async {
            let reset = conn.execute("PRAGMA query_only = 0;", ()).await;
            let tx_reset = match conn.is_autocommit() {
                Ok(true) => Ok(()),
                _ => conn.execute("ROLLBACK;", ()).await.map(|_| ()),
            };
            (reset, tx_reset)
        })
        .catch_unwind()
        .await
        {
            Ok(ok) => ok,
            Err(payload) => {
                // The reset panicked: the connection state is unknown (`query_only`
                // may still be 1 and/or a transaction may be open). Flag it so the
                // next `lock_and_cleanup` attempts a rollback, and surface the error
                // rather than leaving daemon writes silently blocked.
                self.has_dangling_tx.store(true, Ordering::SeqCst);
                return Err(turso::Error::Error(format!(
                    "the database engine panicked while resetting the read-only query state: {}",
                    crate::util::panic_message(&*payload)
                )));
            }
        };
        if let Err(e) = &reset {
            tracing::error!(
                error = %e,
                "query_readonly: failed to reset PRAGMA query_only — daemon writes may \
                 be left disabled on the shared connection"
            );
        }
        if let Err(e) = &tx_reset {
            // Flag the leaked transaction so the next `lock_and_cleanup` retries
            // the rollback before a daemon write reuses the connection.
            self.has_dangling_tx.store(true, Ordering::SeqCst);
            tracing::error!(
                error = %e,
                "query_readonly: failed to roll back a leaked transaction — daemon writes may \
                 be left disabled on the shared connection"
            );
        }
        match result {
            Ok(Ok(rows)) => {
                reset?;
                tx_reset?;
                Ok(rows)
            }
            Ok(Err(e)) => {
                reset?;
                tx_reset?;
                Err(e)
            }
            Err(payload) => {
                // Always attempt the reset above; report the panic as a query error.
                reset?;
                tx_reset?;
                Err(turso::Error::Error(format!(
                    "the database engine panicked while executing the query: {}",
                    crate::util::panic_message(&*payload)
                )))
            }
        }
    }

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

    /// Execute a query, returning all matching rows, using the connection's
    /// prepared-statement cache instead of re-preparing the SQL on every call
    /// (this includes `UPDATE..RETURNING`, which reads rows back).
    ///
    /// # Cache contract
    ///
    /// Only SQL whose full text is invariant across calls — call-site string
    /// literals or compile-time constants — may be routed through this method.
    /// turso caches compiled programs per connection keyed by the exact SQL
    /// text and never evicts them, so per-call-composed fragments (dynamic
    /// IN-lists, interpolated limits, arbitrary user SQL) must keep the plain
    /// prepare-per-call path instead. All `_cached` methods below share this
    /// contract.
    pub async fn query_cached(
        &self,
        sql: &str,
        params: impl IntoParams + Send + 'static,
    ) -> turso::Result<Vec<Row>> {
        let conn = self.lock_and_cleanup().await;
        Self::query_cached_impl(&conn, sql, params).await
    }

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

    /// Map every row, failing on the first per-row error (unlike
    /// [`Self::query_map`], which returns per-row results), using the cached
    /// prepared-statement path (see [`Self::query_cached`] for the cache
    /// contract).
    pub async fn query_map_strict_cached<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_cached(sql, params).await?;
        map_rows(&rows, map)
            .into_iter()
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(Into::into)
    }

    /// Execute a query that returns exactly one row, using the cached
    /// prepared-statement path (see [`Self::query_cached`] for the cache
    /// contract).
    async fn query_row_cached<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;
        let mut stmt = conn.prepare_cached(sql).await?;
        let mut rows = stmt.query(params).await?;
        let row = rows
            .next()
            .await?
            .ok_or(turso::Error::QueryReturnedNoRows)?;
        map(&row).map_err(|e| turso::Error::Error(e.to_string()))
    }

    /// Execute a query that returns zero or one row, using the cached
    /// prepared-statement path (see [`Self::query_cached`] for the cache
    /// contract).
    ///
    /// 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_cached`] 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_cached<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_cached(sql, params, map).await {
            Ok(val) => Ok(Some(val)),
            Err(::turso::Error::QueryReturnedNoRows) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    /// 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(crate) 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(crate) 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(crate) 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(())
}

// Any schema creation, change, or removal is an entry in the append-only
// catalog (see [`crate::db::migrations`]); the only tracking mechanism is the
// id-based applied check in [`crate::db::migrations::run_migrations`].

/// Check whether a table currently exists under `name`.
pub(crate) async fn table_exists(conn: &Connection, table: &str) -> anyhow::Result<bool> {
    let rows = conn
        .query(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=?1",
            params![table],
        )
        .await
        .context("Failed to read table schema (sqlite_master)")?;
    Ok(!rows.is_empty())
}

/// 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")?;

    // Never embed the DDL in the error: a multi-kilobyte schema string hides
    // the engine's actual reason (e.g. `no such column: ticket_id`) and was
    // how a quarantine of a readable store went undiagnosed.
    execute_schema_ddl(&conn, schema).await?;

    Ok(conn)
}

/// Run a schema batch. The source Turso error is the context's cause — do not
/// interpolate `sql` into the message (see [`open_with_schema`]).
async fn execute_schema_ddl(conn: &Connection, sql: &str) -> anyhow::Result<()> {
    if sql.trim().is_empty() {
        return Ok(());
    }
    conn.execute_batch(sql)
        .await
        .context("Failed to run schema DDL")?;
    Ok(())
}

/// Resolve a logical store name to its physical database file path.
///
/// # Consolidated layout (decision 1)
///
/// The domain stores ([`DOMAIN_STORE_NAMES`]) are consolidated into ONE file
/// ([`CONSOLIDATED_DB_NAME`]); the logs store remains its own file. So this
/// function maps every domain name (and the canonical
/// [`CONSOLIDATED_DB_NAME`] itself) to `root/db/{CONSOLIDATED_DB_NAME}.db`, and
/// [`LOG_DB_NAME`] to `root/db/{LOG_DB_NAME}.db`.
#[must_use]
pub(crate) fn store_db_path(root: &Path, name: &str) -> std::path::PathBuf {
    let stem = if name == LOG_DB_NAME {
        LOG_DB_NAME
    } else {
        CONSOLIDATED_DB_NAME
    };
    root.join("db").join(format!("{stem}.db"))
}

/// Resolve a logical store name to its **legacy** per-store file path
/// (`root/db/{name}.db`), as it existed before consolidation.
///
/// Used only by the one-time consolidation import (reading pre-consolidation
/// store files) and the fresh-install probe, so the upgrade path can detect an
/// existing install that still uses the legacy layout.
pub(crate) fn legacy_store_db_path(root: &Path, name: &str) -> std::path::PathBuf {
    root.join("db").join(format!("{name}.db"))
}

/// Sidecar files (`-wal`, `-shm`) beside a store's database file.
///
/// The `-tshm` field is retained for stale-leftover detection: it is a
/// leftover from a pre-removal multiprocess run and is **never** created in
/// normal operation (single-process mode uses the standard `-shm`).
#[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.
///
/// Schema-mismatch / DDL-shape errors are **never** corruption — they are a
/// code bug (a catalog migration ordering error, a typo, …). Those must fail
/// boot with the store intact; quarantining them destroyed readable installs.
/// See [`is_schema_mismatch`].
pub(crate) fn is_corruption_class(e: &anyhow::Error) -> bool {
    if is_schema_mismatch(e) {
        return false;
    }
    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 schema/`execute_batch` failure is a DDL shape error, not
/// page-level damage.
///
/// `CREATE TABLE IF NOT EXISTS` is a no-op on an existing table even when the
/// live columns differ from the expected shape. A later `CREATE INDEX … ON
/// t(new_col)` then fails with `no such column` — that is a catalog ordering
/// bug, not an unreadable file. Keep this list in lockstep with Turso/SQLite
/// wording as new strings surface; do **not** fold these into the fail-closed
/// corruption path.
fn is_schema_mismatch(e: &anyhow::Error) -> bool {
    const DDL_SHAPE: [&str; 8] = [
        "no such column",
        "no such table",
        "no such index",
        "has no column",
        "duplicate column",
        "already exists",
        "syntax error",
        "parse error",
    ];
    let lower = format!("{e:#}").to_lowercase();
    DDL_SHAPE.iter().any(|k| lower.contains(k))
}

/// 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:
/// PASSIVE-first + TRUNCATE for durable-B (0-byte main DB + non-empty WAL),
/// 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 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::db::wal_guard::take_boot_diagnosis(&db_path) else {
        return open_with_schema(&db_path, schema).await;
    };

    match diagnosis {
        crate::db::wal_guard::BootDiagnosis::Healthy => {
            // Healthy main-DB header + 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
                }
            }
        }
        crate::db::wal_guard::BootDiagnosis::DurableB => {
            // Phase 1: the heal connection's own open — the risky reopen of a
            // 0-byte main DB with a non-empty WAL. 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,
                crate::db::wal_guard::BootDiagnosis::DurableB,
            ))
            .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::db::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)
                )))),
            }
        }
    }
}

/// Open (or create) the single consolidated domain database file.
///
/// This is the canonical open for the domain stores: it opens the one
/// consolidated file ([`store_db_path`] mapping all domain names to
/// [`CONSOLIDATED_DB_NAME`]) and applies the append-only schema catalog
/// ([`crate::db::migrations`]) — the single owner of every schema creation,
/// change, and removal for core.db. It does NOT run the per-store post-open
/// hooks (tickets FTS index, admin-user seed) — those are idempotent and run
/// in [`init_all_stores`] (production) and each store's `open` (isolated
/// tests).
///
/// # Catalog, not SCHEMA
///
/// The catalog is a strictly linear history anchored at the 0.4.2 baseline:
/// the starter entries create the 0.4.2 table shapes, then every later entry
/// brings a database to the current shape. The only control flow is the
/// id-based applied check (skip if the id is already recorded, otherwise run
/// and record). A catalog failure is a hard boot failure — it is never
/// classified as corruption and never triggers a quarantine/recreate.
///
/// # Why tables → ALTER → indexes is baked into the catalog order
///
/// The historical incident was real: SCHEMA gained `CREATE UNIQUE INDEX … ON
/// jobs(kind, ticket_id)` while `ticket_id` was only added by a later ALTER.
/// `CREATE TABLE IF NOT EXISTS jobs` left the old table unchanged; the index
/// failed with `no such column` and the store was misclassified as corruption
/// and recreated empty. In the catalog, `jobs.ticket_id` is ADDed
/// (consolidate_003) strictly before the `idx_jobs_phase_ticket` unique index
/// (consolidate_007) — the linear order preserves the invariant.
///
/// # Catalog ordering around the import
///
/// The FTS/board indexes run **before** the one-time consolidation import (a
/// Rust-function catalog entry) so imported tickets are already FTS-searchable.
/// The import copies the legacy user tables **verbatim** (FK off → bulk load →
/// FK on), then two post-import entries shape the migrated rows:
///
/// - `003_reset_nonterminal_tickets` — non-terminal legacy tickets are reset to
///   `backlog` (it is a DATA reset, so it must run after the import fills the
///   consolidated tickets table; on an already-current DB it is recorded and
///   skipped, never touching live in-progress tickets).
/// - the `13` post-import cleanup DELETE — pre-rework `kind='ticket_stage'`,
///   `'ticket_analysis'`, `'ticket_implementation'` jobs the import rolled in
///   are discarded, and a `PRAGMA foreign_key_check` reports (never fails on)
///   any orphan rows.
///
/// So the consolidation contract is *migrate without losing data* (rows that
/// should survive are preserved) while the deliberate resets/drops are applied
/// to the imported rows, and no destructive operation re-fires on an
/// already-current DB.
///
/// Returns a **fresh** connection. Production shares it across all domain
/// stores via [`DOMAIN_CONN`]; isolated test opens (`open_test_store!`) drop it
/// with the store. There is deliberately no process-global cache here so test
/// isolation never leaks a connection across roots.
pub(crate) async fn open_consolidated_store(root: &Path) -> anyhow::Result<Connection> {
    // The catalog owns the ENTIRE schema. Open the raw connection (no DDL) and
    // run the catalog; a catalog failure is a hard boot error — never healed
    // or quarantined. A fresh/recreated store is empty until the catalog builds
    // the full current shape.
    let conn = open_store(root, CONSOLIDATED_DB_NAME, "").await?;
    migrations::run_migrations(&conn, migrations::TargetDb::Core, root).await?;

    // Enable CDC on the FINAL connection, after all migrations/import, so
    // boot-time schema + import writes are not captured. This single chokepoint
    // covers every connection-recreation path (open_store's heal/recreate arms
    // all funnel back through here) and the test-store opens (open_test_store!
    // and init_test_stores both call open_consolidated_store).
    cdc::enable_capture(&conn).await?;

    Ok(conn)
}

/// Copy user tables from the 6 legacy per-store files into the consolidated
/// database, preserving the AUTOINCREMENT watermark (`sqlite_sequence`).
///
/// *FK handling*: FK enforcement is OFF during the bulk load (children may be
/// copied before their parents, and the target schema's FKs must not fire
/// mid-copy). It is restored to ON afterwards; the post-import cleanup DELETE
/// and `PRAGMA foreign_key_check` live in the catalog's own cleanup migration
/// ([`crate::db::migrations`]).
pub(crate) async fn import_legacy_stores(conn: &Connection, root: &Path) -> anyhow::Result<()> {
    conn.execute("PRAGMA foreign_keys = OFF;", ())
        .await
        .context("Failed to disable FK enforcement during consolidation import")?;

    let result = import_legacy_stores_inner(conn, root).await;

    // Restore FK enforcement regardless of the import outcome so the connection
    // is left in a sane state, but surface the import (root-cause) error
    // preferentially over a secondary FK-re-enable failure.
    let reenabled = conn
        .execute("PRAGMA foreign_keys = ON;", ())
        .await
        .context("Failed to re-enable FK enforcement after consolidation import");
    match (result, reenabled) {
        (Ok(()), Ok(_)) => Ok(()),
        (Err(e), _) | (Ok(()), Err(e)) => Err(e),
    }
}

async fn import_legacy_stores_inner(conn: &Connection, root: &Path) -> anyhow::Result<()> {
    for name in DOMAIN_STORE_NAMES {
        let legacy_path = legacy_store_db_path(root, name);
        if !legacy_path.exists() {
            continue;
        }
        if let Err(e) = import_one_legacy_store(conn, name, &legacy_path).await {
            return Err(e).with_context(|| format!("consolidation import failed for '{name}'"));
        }
    }
    Ok(())
}

/// Copy one legacy store's user tables into the consolidated database.
///
/// The source is opened READ-ONLY (`ReadOnly | NoLock`) with the same
/// experimental feature set as the daemon so the WAL is read correctly. Turso
/// internals (`sqlite_%`, `__turso_internal_%`) are never
/// copied — the consolidated schema recreates them. A source table that has no
/// matching target table (schema drift) is skipped; the removed rework-child
/// tables (`ticket_jobs`, `ticket_stage_jobs`) fall through to `None` and are
/// dropped rather than imported.
async fn import_one_legacy_store(
    conn: &Connection,
    store_name: &str,
    legacy_path: &Path,
) -> anyhow::Result<()> {
    use std::sync::Arc;

    let io: Arc<dyn turso::core::IO> = Arc::new(turso::core::PlatformIO::new()?);
    let path_str = legacy_path
        .to_str()
        .with_context(|| format!("legacy store path must be UTF-8: {}", legacy_path.display()))?;
    let db = turso::core::Database::open_file_with_flags(
        io.clone(),
        path_str,
        turso::core::OpenFlags::ReadOnly | turso::core::OpenFlags::NoLock,
        experimental_database_opts(),
        None,
    )
    .with_context(|| format!("failed to open legacy store '{store_name}' read-only"))?;
    let src = db
        .connect()
        .with_context(|| format!("failed to connect to legacy store '{store_name}'"))?;
    // NB: `turso::core::Connection` (the source) is SYNCHRONOUS — rows are
    // read via `stmt.step()`; the consolidated target `conn` is our async
    // wrapper. The source is read fully into memory per table so the sync
    // statement is never held across an await (keeps the future `Send`).

    // Enumerate source user tables.
    let src_tables = read_sync_rows(
        &io,
        &src,
        &format!("SELECT name FROM sqlite_master WHERE type='table' AND {USER_OBJECT_FILTER}"),
        1,
    )?;

    for table_row in src_tables {
        let Some(src_table) = table_row.first().and_then(|v| match v {
            Value::Text(s) => Some(s.as_str()),
            _ => None,
        }) else {
            continue;
        };
        let Some(target_table) = consolidate_table_name(src_table) else {
            continue;
        };
        if !table_exists(conn, target_table).await? {
            continue;
        }
        copy_table(conn, &io, &src, src_table, target_table).await?;
    }

    // Preserve the AUTOINCREMENT watermark (sqlite_sequence) so id counters
    // keep incrementing correctly after consolidation.
    copy_sqlite_sequence(conn, &io, &src).await?;

    Ok(())
}

/// Map a legacy source table name to its consolidated target name. Returns
/// `None` for tables that have no consolidated counterpart (schema drift).
///
/// There are no renames across the consolidation — every legacy table either
/// keeps its name or maps to `None`.
///
/// **Dormant legacy tables are deliberately dropped** (mapped to `None`): the
/// consolidated schema catalog never recreates them and no production code
/// reads them. These include `config_role` — historically created for the
/// (removed) per-role config override rows (no code references, not part of the
/// config tables) — and the old `ticket_jobs` child table, which the
/// job-per-phase model removed (it stores `ticket_id` directly on `jobs`).
/// The consolidated `schema_migrations`/fresh-install probe paths do not depend
/// on them.
///
/// `user_roles` is an explicit `None` arm (a deliberate, documented drop — the
/// role pool is permission-derived now) even though the wildcard already maps
/// unknown tables to `None`; `#[allow]` on the function documents the intent
/// without a wildcard-only match.
#[allow(clippy::match_same_arms)]
fn consolidate_table_name(src: &str) -> Option<&'static str> {
    match src {
        // All other current user tables keep their name.
        "tickets" => Some("tickets"),
        "ticket_comments" => Some("ticket_comments"),
        "ticket_counters" => Some("ticket_counters"),
        "sessions" => Some("sessions"),
        "session_metadata" => Some("session_metadata"),
        "jobs" => Some("jobs"),
        "agents" => Some("agents"),
        "pending_jobs" => Some("pending_jobs"),
        "research_jobs" => Some("research_jobs"),
        "workspaces" => Some("workspaces"),
        "workspace_contexts" => Some("workspace_contexts"),
        "editor_tabs" => Some("editor_tabs"),
        "users" => Some("users"),
        "user_channels" => Some("user_channels"),
        // `user_roles` is dropped: role pools are permission-derived now, so
        // no production code reads it and the consolidated schema never
        // recreates it. Rows are discarded during the import.
        "user_roles" => None,
        "config_kv" => Some("config_kv"),
        "config_model_routing" => Some("config_model_routing"),
        "chat_history" => Some("chat_history"),
        // Anything else — including the dormant `config_role` table — is a
        // pre-consolidation artifact with no consolidated counterpart; drop it.
        _ => None,
    }
}

/// Copy one table's rows (intersecting columns) from the read-only source to
/// the consolidated target.
///
/// The clear + bulk insert run inside **one transaction** (FK enforcement is
/// already OFF during the import): a large table is copied with a single fsync
/// rather than one per row, and a crash mid-copy rolls back the whole table so
/// a re-run leaves no partial rows.
///
/// The source is read in BOUNDED batches via `LIMIT/OFFSET` pagination, ordered
/// by `rowid` so each batch is deterministic and does not depend on an
/// unspecified scan order — a large legacy table (e.g. the ~251MB sessions
/// store) is never fully materialized in memory, bounding the deploy-time OOM
/// risk that a whole-table `Vec<Vec<Value>>` would pose. Each batch statement is
/// fully consumed before the next await, keeping the future `Send`. (The source
/// is opened read-only and unmodified during the import; every copied user table
/// is a rowid table, so `ORDER BY rowid` is always valid.)
///
/// *Batch size*: [`IMPORT_READ_BATCH`]. Offset scans re-scan from the start, so
/// the cost is one full rescan per batch — acceptable for a one-time migration
/// versus unbounded memory.
const IMPORT_READ_BATCH: usize = 10_000;

async fn copy_table(
    conn: &Connection,
    io: &std::sync::Arc<dyn turso::core::IO>,
    src: &std::sync::Arc<turso::core::Connection>,
    src_table: &str,
    target_table: &str,
) -> anyhow::Result<()> {
    let src_cols = column_names_sync(io, src, src_table)?;
    let target_cols = column_names_async(conn, target_table).await?;

    // Intersect on column name, preserving the target's column order.
    let common: Vec<String> = target_cols
        .iter()
        .filter(|c| src_cols.contains(*c))
        .cloned()
        .collect();
    if common.is_empty() {
        return Ok(());
    }
    let col_list = common.join(", ");
    let placeholders = vec!["?"; common.len()].join(", ");
    let insert_sql = format!("INSERT INTO {target_table} ({col_list}) VALUES ({placeholders})");

    // Empty the target table before re-copy so a re-run (crash mid-import) does
    // not leave duplicate rows — atomically with the inserts so a failure rolls
    // back both.
    let tx = conn
        .begin_tx()
        .await
        .with_context(|| format!("failed to begin import for '{target_table}'"))?;
    tx.execute(&format!("DELETE FROM {target_table}"), ())
        .await
        .with_context(|| format!("failed to clear target table '{target_table}'"))?;

    // Read bounded batches from the (read-only, unmodified) source and insert
    // each into the same per-table transaction, bounding peak memory to one
    // batch.
    let mut offset = 0usize;
    loop {
        let batch_select = format!(
            "SELECT {col_list} FROM {src_table} ORDER BY rowid LIMIT {IMPORT_READ_BATCH} OFFSET {offset}"
        );
        let rows = read_sync_rows(io, src, &batch_select, common.len())?;
        if rows.is_empty() {
            break;
        }
        for vals in rows {
            tx.execute(&insert_sql, vals).await.with_context(|| {
                format!("failed to insert row into '{target_table}' during import")
            })?;
        }
        offset += IMPORT_READ_BATCH;
    }

    tx.commit()
        .await
        .with_context(|| format!("failed to commit import for '{target_table}'"))?;
    Ok(())
}

/// Run a read-only SQL that returns `ncols` columns and collect every row's
/// values as owned [`Value`]s. The source connection is synchronous; the
/// statement is fully consumed here so no non-`Send` statement is held across
/// an await in the caller.
fn read_sync_rows(
    io: &std::sync::Arc<dyn turso::core::IO>,
    src: &std::sync::Arc<turso::core::Connection>,
    sql: &str,
    ncols: usize,
) -> anyhow::Result<Vec<Vec<Value>>> {
    use turso::core::StepResult;
    let Some(mut stmt) = src
        .query(sql)
        .with_context(|| format!("failed to prepare read: {sql}"))?
    else {
        return Ok(Vec::new());
    };
    let mut out = Vec::new();
    loop {
        match stmt.step() {
            Ok(StepResult::Row) => {
                let row = stmt.row().context("row missing after StepResult::Row")?;
                let mut vals = Vec::with_capacity(ncols);
                for idx in 0..ncols {
                    vals.push(row_to_value(row.get_value(idx)));
                }
                out.push(vals);
            }
            Ok(StepResult::Done) => break,
            Ok(StepResult::IO | StepResult::Yield) => {
                io.step().context("I/O step during read-only import")?;
            }
            Ok(StepResult::Interrupt) => {
                return Err(anyhow::anyhow!("read-only import query interrupted"));
            }
            Ok(StepResult::Busy) => {
                return Err(anyhow::anyhow!(
                    "read-only import query busy; try again later"
                ));
            }
            Err(e) => return Err(anyhow::anyhow!("read-only import query failed: {e}")),
        }
    }
    Ok(out)
}

/// Enumerate a source table's column names (synchronous core connection).
fn column_names_sync(
    io: &std::sync::Arc<dyn turso::core::IO>,
    src: &std::sync::Arc<turso::core::Connection>,
    table: &str,
) -> anyhow::Result<Vec<String>> {
    let rows = read_sync_rows(io, src, &format!("PRAGMA table_info({table})"), 2)?;
    Ok(rows
        .iter()
        .filter_map(|r| match r.get(1) {
            Some(Value::Text(s)) => Some(s.as_str().to_string()),
            _ => None,
        })
        .collect())
}

/// Enumerate a target table's column names (our async wrapper).
async fn column_names_async(conn: &Connection, table: &str) -> anyhow::Result<Vec<String>> {
    let rows = conn
        .query(&format!("PRAGMA table_info({table})"), ())
        .await
        .with_context(|| format!("failed to read columns of target '{table}'"))?;
    Ok(rows
        .iter()
        .filter_map(|r| r.get::<String>(1).ok())
        .collect())
}

/// Convert a `turso::core::Value` into a `turso::Value` for the async insert.
fn row_to_value(v: &turso::core::Value) -> Value {
    match v {
        turso::core::Value::Null => Value::Null,
        turso::core::Value::Numeric(turso::core::Numeric::Integer(i)) => Value::Integer(*i),
        turso::core::Value::Numeric(turso::core::Numeric::Float(f)) => {
            let f64_val: f64 = (*f).into();
            Value::Real(f64_val)
        }
        turso::core::Value::Text(t) => Value::Text(t.as_str().to_string()),
        turso::core::Value::Blob(b) => Value::Blob(b.clone()),
    }
}

/// Copy the `sqlite_sequence` AUTOINCREMENT watermark (only for tables present
/// in both source and target) so `sessions.id` / `chat_history.id` keep
/// incrementing correctly after consolidation.
///
/// `sqlite_sequence` is created as `CREATE TABLE sqlite_sequence(name,seq)` —
/// no PRIMARY KEY or UNIQUE on `name`, and Turso rejects indexes on it — so
/// `ON CONFLICT(name)` is invalid SQL. Delete-then-insert is the portable
/// upsert (the engine's AUTOINCREMENT path maintains one row per name).
async fn copy_sqlite_sequence(
    conn: &Connection,
    io: &std::sync::Arc<dyn turso::core::IO>,
    src: &std::sync::Arc<turso::core::Connection>,
) -> anyhow::Result<()> {
    // `sqlite_sequence` only exists when the source has AUTOINCREMENT tables;
    // a missing table makes the query return an error, which we treat as empty.
    let Ok(rows) = read_sync_rows(io, src, "SELECT name, seq FROM sqlite_sequence", 2) else {
        return Ok(());
    };
    for row in rows {
        let (Some(name), Some(seq)) = (
            row.first().and_then(|v| match v {
                Value::Text(s) => Some(s.as_str().to_string()),
                _ => None,
            }),
            row.get(1).and_then(|v| match v {
                Value::Integer(i) => Some(*i),
                _ => None,
            }),
        ) else {
            continue;
        };
        if !table_exists(conn, &name).await? {
            continue;
        }
        conn.execute(
            "DELETE FROM sqlite_sequence WHERE name = ?1",
            params![name.clone()],
        )
        .await
        .with_context(|| format!("failed to clear AUTOINCREMENT watermark for '{name}'"))?;
        conn.execute(
            "INSERT INTO sqlite_sequence (name, seq) VALUES (?1, ?2)",
            params![name.clone(), seq],
        )
        .await
        .with_context(|| format!("failed to preserve AUTOINCREMENT watermark for '{name}'"))?;
    }
    Ok(())
}

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

/// 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 before any live traffic). 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. Schema-mismatch (`no such column`, …) is
            // excluded from [`is_corruption_class`]: fail boot, keep the file.
            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 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 before any
/// live traffic. 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)> {
    // The old `.tshm` frame-index precondition (`frame_index_len == max_fr`)
    // no longer applies — multiprocess_wal was removed and the daemon runs in
    // default single-process mode, so there is no shared WAL frame index to
    // reset before a rebuild.

    // ── 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::db::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 here — TRUNCATE-first on a 0-byte main DB destroys
    // committed frames.
    let durable_b = matches!(diagnosis, crate::db::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?;
    }
    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).
///
/// 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.
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"),
        ],
    )
}

/// 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::*;
    use std::sync::atomic::AtomicUsize;

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

        // Forensic-family reads use the same live single-process feature set.
        let fam = experimental_database_opts();
        assert!(
            fam.enable_index_method,
            "family reads need index_method (stores are created with it)"
        );
    }

    #[test]
    fn schema_mismatch_is_not_corruption() {
        for msg in [
            "no such column: ticket_id",
            "no such table: jobs",
            "no such index: idx_jobs_phase_ticket",
            "table jobs has no column named ticket_id",
            "duplicate column name: ticket_id",
            "index idx_jobs_phase_ticket already exists",
            "syntax error",
            "parse error at line 1",
        ] {
            let e = anyhow::anyhow!("{msg}");
            assert!(is_schema_mismatch(&e), "mismatch: {msg}");
            assert!(
                !is_corruption_class(&e),
                "schema-mismatch must not quarantine: {msg}"
            );
            let wrapped = e.context("Failed to run schema DDL");
            assert!(
                !is_corruption_class(&wrapped),
                "wrapped schema-mismatch must not quarantine: {msg}"
            );
        }
        let page = anyhow::anyhow!("Invalid page type");
        assert!(!is_schema_mismatch(&page));
        assert!(
            is_corruption_class(&page),
            "page-level damage stays corruption-class"
        );
    }

    /// An existing, readable store whose open-time DDL indexes a column the
    /// live table does not have must fail boot — never quarantine. This is
    /// the job-per-phase incident in miniature.
    #[tokio::test]
    async fn schema_mismatch_does_not_quarantine_existing_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 _conn = open_with_schema(
                &db_path,
                "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY);",
            )
            .await
            .unwrap();
        }
        crate::db::wal_guard::set_boot_diagnosis(
            &db_path,
            crate::db::wal_guard::BootDiagnosis::Healthy,
        );
        let err = open_store(
            root,
            "board",
            "CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY);\
             CREATE INDEX IF NOT EXISTS idx_t_missing ON t(missing_col);",
        )
        .await
        .expect_err("schema mismatch must fail boot, not succeed");
        assert!(
            is_schema_mismatch(&err) || format!("{err:#}").to_lowercase().contains("no such"),
            "expected a DDL-shape error, got: {err:#}"
        );
        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,
            "readable store must not be quarantined on schema mismatch"
        );
        assert!(
            db_path.exists(),
            "original main file must still be at the live path"
        );
    }

    /// 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::db::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::db::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::db::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::db::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");
        }
        // 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/db/mod.rs`) and the debug CLI (`src/db/debug.rs`).
    ///
    /// All database access must go through [`crate::db::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.
    #[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/db/mod.rs", "src/db/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::db::Connection (db/mod.rs);\n\
             the debug CLI (db/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/db/mod.rs)
    /// and the debug CLI's separate read-only path (`connect_readonly` in
    /// src/db/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.
    #[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/db/mod.rs", "impl Connection"),
            ("src/db/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::db::wal_guard::set_boot_diagnosis(
            &db_path,
            crate::db::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)"
        );
    }

    #[test]
    fn open_time_lock_error_predicate() {
        let lock = |msg: &str| turso::Error::Error(msg.to_string());
        // The real open-time fcntl `F_SETLK` WouldBlock folded into the
        // generic Error variant by turso_core's catch-all From impl.
        assert!(is_open_time_lock_error(&lock(
            "Locking error: Failed locking file '/x/logs.db-wal'. File is locked by another process"
        )));
        // The shared-WAL coordination file variant of the same open-time lock.
        assert!(is_open_time_lock_error(&lock(
            "Locking error: Failed locking shared WAL coordination file. File is locked by another process"
        )));
        // A statement-level busy error is NOT the open-time lock.
        assert!(!is_open_time_lock_error(&turso::Error::Busy(
            "database is locked".into()
        )));
        // A table-level lock surfaced as a generic Error is NOT the open-time lock.
        assert!(!is_open_time_lock_error(&lock(
            "Runtime error: database table is locked"
        )));
        // Corruption and I/O-class messages must not be treated as this race.
        assert!(!is_open_time_lock_error(&lock(
            "Database file is corrupted"
        )));
    }

    #[tokio::test]
    async fn open_lock_retry_succeeds_after_transient_lock() {
        let calls = AtomicUsize::new(0);
        let result = with_open_lock_retry(3, Duration::from_millis(1), || {
            let n = calls.fetch_add(1, Ordering::SeqCst);
            async move {
                if n < 2 {
                    Err(turso::Error::Error(
                        "Locking error: Failed locking file '/x/logs.db-wal'. File is locked by another process".to_string(),
                    ))
                } else {
                    Ok("opened")
                }
            }
        })
        .await;
        assert!(
            result.is_ok(),
            "open must succeed after the transient lock clears"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            3,
            "closure must be called twice before success"
        );
    }

    #[tokio::test]
    async fn open_lock_retry_exhausts_and_preserves_error() {
        let calls = AtomicUsize::new(0);
        let err = with_open_lock_retry(2, Duration::from_millis(1), || {
            calls.fetch_add(1, Ordering::SeqCst);
            async {
                Err::<(), turso::Error>(turso::Error::Error(
                    "Locking error: Failed locking file '/x/logs.db-wal'. File is locked by another process".to_string(),
                ))
            }
        })
        .await
        .expect_err("a persistent open-time lock must eventually surface");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            3,
            "max total tries must be attempts + 1"
        );
        assert!(
            is_open_time_lock_error(&err),
            "the original lock error must be preserved"
        );
    }

    #[tokio::test]
    async fn open_lock_no_retry_on_non_lock_error() {
        let calls = AtomicUsize::new(0);
        let err = with_open_lock_retry(2, Duration::from_millis(1), || {
            calls.fetch_add(1, Ordering::SeqCst);
            async { Err::<(), turso::Error>(turso::Error::Error("Disk I/O error".to_string())) }
        })
        .await
        .expect_err("a non-lock open failure must not be retried");
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "non-lock errors must not trigger retries"
        );
        assert!(
            matches!(&err, turso::Error::Error(msg) if msg == "Disk I/O error"),
            "the original non-lock error must be returned"
        );
    }
}