mahbot 0.3.0

use anyhow::Context;
use chrono::{DateTime, Utc};
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::OnceCell;
use tracing::warn;
use turso::Builder;
pub use turso::{IntoParams, Row, Value, params, params_from_iter};

// ── 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`] (or `turso::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 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.
pub const EXPERIMENTAL_FEATURES: &[&str] = &["index_method", "multiprocess_wal"];

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

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

/// Initialize all database stores concurrently.
///
/// This is the canonical initialization path for the 7 data stores (board,
/// session, workspace, users, stats, chat_history, config).  The `logs` store
/// is **not** included here because it must be initialized earlier via
/// [`crate::logs::init_tracing`], which requires the log store before any
/// other subsystem is ready.
///
/// > **Keep this list in sync with [`iter_checkpoint_stores`]** — every store
/// > listed here must also appear in that iterator.  The converse is not strictly
/// > required because `logs` (and any future store initialized outside this path)
/// > lives only in the checkpoint iterator.
pub async fn init_all_stores() -> anyhow::Result<()> {
    tokio::try_join!(
        crate::session::init_global(),
        crate::workspace::init_global(),
        crate::users::init_global(),
        crate::board::init_global(),
        crate::stats::init_global(),
        crate::chat_history::init_global(),
        crate::config_db::init_global(),
    )?;
    Ok(())
}

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

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

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

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

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

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

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

    sanitized
        .split_whitespace()
        .map(|word| word.trim_start_matches('/'))
        .filter(|word| !word.is_empty())
        .collect::<Vec<_>>()
        .join(" ")
}

/// Build a comma-separated list of `?` placeholders for SQL IN-clauses.
///
/// Returns an empty string for `count == 0`. Callers MUST guard against
/// empty lists to avoid producing invalid SQL like `WHERE id IN ()`.
///
/// # Example
///
/// ```
/// # use mahbot::turso::sql_in_placeholders;
/// assert_eq!(sql_in_placeholders(3), "?, ?, ?");
/// assert_eq!(sql_in_placeholders(0), "");
/// ```
///
/// 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 fn sql_in_placeholders(count: usize) -> String {
    vec!["?"; count].join(", ")
}

/// Compatibility layer over Turso with a persistent connection.
///
/// Every `execute` / `execute_batch` / `query` / `query_map` call reuses a
/// single cached connection — eliminating the per-call overhead of
/// `db.connect()` and preventing page-cache races that occur when concurrent
/// read+write operations share a Limbo page cache.
///
/// All operations are serialized through an internal mutex to support
/// concurrent access from multiple tasks (required for parallel tests).
///
/// Dangling transactions (TxGuard dropped without commit/rollback) are handled
/// via a deferred rollback pattern: the flag is set in Drop and the actual
/// ROLLBACK is executed at the start of the next write operation on any clone
/// of this Connection.
#[derive(Clone, Debug)]
pub 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>,
}

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

impl Connection {
    pub async fn open(path: &Path) -> anyhow::Result<Self> {
        let path_str = path
            .to_str()
            .with_context(|| format!("database path must be UTF-8: {}", path.display()))?;
        // Derive experimental features from the canonical opts function.
        // If you need to add/remove an experimental feature, change
        // experimental_database_opts() and EXPERIMENTAL_FEATURES — not here.
        let opts = experimental_database_opts();
        let db = Builder::new_local(path_str)
            .experimental_index_method(opts.enable_index_method)
            .experimental_multiprocess_wal(opts.enable_multiprocess_wal)
            .build()
            .await
            .context("failed to open local database")?;
        let conn = db.connect()?;
        conn.busy_timeout(Duration::from_mins(1))?;
        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
    }

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

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

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

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

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

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

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

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

    /// Force a WAL checkpoint with TRUNCATE mode.
    ///
    /// Writes all pending WAL content to the main database file and truncates
    /// the WAL. This is critical before hard process termination
    /// (e.g., [`std::process::exit`] in self-update) to prevent data loss from
    /// unwritten WAL pages.
    ///
    /// Safe to call even if the database is not in WAL mode — non-WAL databases
    /// treat this as a no-op.
    pub async fn checkpoint(&self) -> anyhow::Result<()> {
        self.query("PRAGMA wal_checkpoint(TRUNCATE);", ())
            .await
            .context("Failed to checkpoint WAL")?;
        Ok(())
    }

    /// 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<()> {
        let rows = self
            .query("PRAGMA quick_check;", ())
            .await
            .context("Failed to execute PRAGMA quick_check")?;

        if let Some(row) = rows.first() {
            match row.get_value(0)? {
                Value::Text(s) if s == "ok" => {}
                Value::Text(s) => anyhow::bail!("Database integrity check failed: {s}"),
                _ => anyhow::bail!("Unexpected result from PRAGMA quick_check"),
            }
        }
        Ok(())
    }

    /// Run PRAGMA integrity_check to produce a complete diagnostic report.
    ///
    /// Performs a thorough database integrity scan (b-tree tree structure,
    /// index consistency, constraint checks, etc.) and returns every problem
    /// description as a separate `String` in the result vector.  Returns an
    /// empty `Vec` when the database is healthy — the only row from
    /// `PRAGMA integrity_check` is `"ok"`.
    ///
    /// This is more comprehensive and slower than [`quick_check`]; it should
    /// only be run when [`quick_check`] has already indicated corruption, so
    /// that operators get the full diagnostic picture without the overhead on
    /// every 5-minute periodic scan.
    ///
    /// # Errors
    ///
    /// Fails if the SQL statement itself fails (e.g., severe corruption that
    /// prevents even diagnostic queries from executing), or if any row value
    /// has an unexpected type.
    pub async fn integrity_check(&self) -> anyhow::Result<Vec<String>> {
        let rows = self
            .query("PRAGMA integrity_check;", ())
            .await
            .context("Failed to execute PRAGMA integrity_check")?;

        let mut problems: Vec<String> = Vec::new();
        for row in &rows {
            match row.get_value(0)? {
                Value::Text(s) if s == "ok" => { /* healthy, skip */ }
                Value::Text(s) => problems.push(s),
                _ => anyhow::bail!("Unexpected result from PRAGMA integrity_check"),
            }
        }
        Ok(problems)
    }
}

/// A locked connection handle scoped to a single transaction.
/// Holds the mutex guard for the entire duration — dropped guard triggers rollback.
pub 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 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 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(())
}

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

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

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

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

    Ok(conn)
}

/// 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.
pub(crate) async fn open_store(
    root: &Path,
    name: &str,
    schema: &str,
) -> anyhow::Result<Connection> {
    let db_path = root.join("db").join(format!("{name}.db"));
    open_with_schema(&db_path, schema).await
}

/// 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.
///
/// This is the canonical implementation; callers in `board.rs` and
/// `management.rs` delegate to it rather than duplicating the logic.
pub(crate) async fn with_tx(
    conn: &Connection,
    ticket_id: &str,
    action_label: &str,
    work: impl AsyncFnOnce(&TxGuard<'_>) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
    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}"))?;

    if let Err(e) = work(&tx).await {
        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",
        );
        return Err(e.context(format!("{action_label}: transaction rolled back")));
    }

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

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

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

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

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

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

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

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

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

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

    /// Verify that integrity_check returns an empty Vec on a healthy database.
    #[tokio::test]
    async fn test_integrity_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");

        // Create a simple table and insert data so the database has real
        // schema — integrity_check on a truly empty DB works, but testing
        // against a minimally populated one is more representative.
        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");

        let problems = conn
            .integrity_check()
            .await
            .expect("integrity_check should pass on a healthy database");
        assert!(
            problems.is_empty(),
            "expected no integrity problems, got: {problems:?}",
        );
    }
}