//! SQLite database wrapper with versioned migrations for samples, VFS, tags, and analysis tables.

mod migrations;
mod stats;

use std::path::Path;

use rusqlite::Connection;
use synckit_config::{ConfigError, ConfigStore};
use thiserror::Error;
use tracing::instrument;

use crate::config_key::{CONFIG, ConfigKey};
use migrations::{MIGRATIONS, register_hash_row_id};

pub use migrations::SCHEMA_VERSION;

#[derive(Error, Debug)]
pub enum DbError {
    #[error("SQLite error: {0}")]
    Sqlite(#[from] rusqlite::Error),

    /// The vault's schema is ahead of what this build knows how to read, so
    /// opening it would mean querying a shape this code has never seen. See
    /// [`Database::migrate`] for why that is refused rather than tolerated.
    #[error(
        "this vault was written by a newer version of audiofiles \
         (vault schema {found}, this build understands {supported}). \
         Update audiofiles to open it."
    )]
    VaultTooNew {
        /// `PRAGMA user_version` read off the vault.
        found: i32,
        /// The highest version this build can produce, [`SCHEMA_VERSION`].
        supported: i32,
    },
}

/// The config store wraps rusqlite; its one failure mode is the database, so it
/// folds into [`DbError::Sqlite`] rather than carrying a second SQL error type.
impl From<ConfigError> for DbError {
    fn from(error: ConfigError) -> Self {
        match error {
            ConfigError::Db(error) => DbError::Sqlite(error),
        }
    }
}

/// Core database wrapper. All access is synchronous, no async runtime needed,
/// safe to use from a CLAP plugin host thread.
pub struct Database {
    conn: Connection,
    /// The shared config store over `user_config`. Attached, not opened: the
    /// table and its sync triggers are stood up by this crate's migrations, so
    /// the store drives the existing table rather than creating its own.
    config: ConfigStore,
}

/// Compile-time proof that a write transaction is open on the connection.
///
/// Constructed only by [`Database::transaction`], and required by the row-write
/// functions that must run inside a batched transaction rather than as
/// standalone autocommits in a loop (e.g. [`crate::analysis::save_analysis`],
/// [`crate::rules::apply_tag_sourced`]). Holding a `&Tx` is the only way to call
/// those functions, so "bulk loop of per-row autocommits", the chronic
/// per-row-commit pattern, does not compile: there is no `Tx` to pass except
/// inside a `transaction` closure that already batches the whole loop.
///
/// The token is a zero-sized marker; the SQL still runs on `db.conn()`, which
/// participates in the ambient `BEGIN IMMEDIATE` opened by `transaction`.
pub struct Tx(());

impl Database {
    /// Open (or create) the database at the given path and run migrations.
    #[instrument(skip_all)]
    pub fn open(path: impl AsRef<Path>) -> Result<Self, DbError> {
        let conn = Connection::open(path)?;
        conn.execute_batch(
            // WAL + synchronous=NORMAL is the standard durable-but-fast pairing:
            // commits no longer fsync individually (only at checkpoint), which is
            // what made the import path ~2 fsyncs/file. NORMAL under WAL can lose
            // only the last few committed transactions on power loss, never
            // corruption, acceptable for a local sample library. The cache /
            // mmap / temp_store pragmas cut page churn on large scans and the
            // import write batch.
            "PRAGMA journal_mode=WAL;\
             PRAGMA synchronous=NORMAL;\
             PRAGMA foreign_keys=ON;\
             PRAGMA busy_timeout=5000;\
             PRAGMA cache_size=-16000;\
             PRAGMA mmap_size=268435456;\
             PRAGMA temp_store=MEMORY;\
             PRAGMA wal_checkpoint(TRUNCATE);",
        )?;
        register_hash_row_id(&conn)?;
        let mut db = Self {
            conn,
            config: ConfigStore::attached(&CONFIG),
        };
        db.migrate()?;
        db.seed_config_key_policy()?;
        Ok(db)
    }

    /// Flush the WAL back into the main database file and remove the -shm file.
    ///
    /// Call after large write batches (e.g. import completion) to keep the
    /// WAL index fresh and avoid stale memory-mapped state on macOS.
    pub fn wal_checkpoint(&self) -> Result<(), DbError> {
        self.conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")?;
        Ok(())
    }

    /// Open an in-memory database (for tests).
    #[instrument(skip_all)]
    pub fn open_in_memory() -> Result<Self, DbError> {
        let conn = Connection::open_in_memory()?;
        conn.execute_batch("PRAGMA foreign_keys=ON;")?;
        register_hash_row_id(&conn)?;
        let mut db = Self {
            conn,
            config: ConfigStore::attached(&CONFIG),
        };
        db.migrate()?;
        db.seed_config_key_policy()?;
        Ok(db)
    }

    /// Seed `config_key_policy` from the [`CONFIG`] spec, the single source of
    /// truth for which `user_config` keys may sync. Run at every open so the SQL
    /// export triggers always reflect the current registry: adding a key in Rust
    /// is enough, no migration needed. Idempotent (clear + reinsert the closed
    /// set); the rows are the spec's [`policy_rows`](synckit_config::ConfigSpec::policy_rows),
    /// so an undeclared key has no row and the export filter cannot admit it.
    fn seed_config_key_policy(&self) -> Result<(), DbError> {
        self.conn.execute("DELETE FROM config_key_policy", [])?;
        let mut stmt = self
            .conn
            .prepare("INSERT INTO config_key_policy (key, replicated) VALUES (?1, ?2)")?;
        for row in CONFIG.policy_rows() {
            stmt.execute(rusqlite::params![row.key, i64::from(row.replicated)])?;
        }
        Ok(())
    }

    /// Read a `user_config` value through the shared config store, `None` when
    /// unset. The store is attached to the `user_config` table this crate's
    /// migrations own.
    pub fn get_config(&self, key: ConfigKey) -> Result<Option<String>, DbError> {
        Ok(self.config.get(&self.conn, key.as_str())?)
    }

    /// Write a `user_config` value through the shared config store.
    ///
    /// An upsert on the key: writing a key that already exists fires the table's
    /// UPDATE trigger once, not a DELETE followed by an INSERT the way the old
    /// `INSERT OR REPLACE` did, so a synced key enqueues one changelog row per
    /// edit rather than a spurious delete-then-insert pair.
    pub fn set_config(&self, key: ConfigKey, value: &str) -> Result<(), DbError> {
        self.config.set(&self.conn, key.as_str(), value)?;
        Ok(())
    }

    /// Remove a `user_config` key through the shared config store. Absent
    /// already is not an error.
    pub fn delete_config(&self, key: ConfigKey) -> Result<(), DbError> {
        self.config.unset(&self.conn, key.as_str())?;
        Ok(())
    }

    /// Apply pending migrations using PRAGMA user_version as the version tracker.
    ///
    /// Each migration step runs inside a transaction so the schema change and
    /// version bump are atomic, a crash between the two can no longer leave the
    /// database in an inconsistent state.
    ///
    /// The runner is bounded on both sides. Forward is the ordinary case.
    /// Backward is not possible and must not be attempted silently: a vault
    /// carrying a version this build has never heard of was written by a newer
    /// audiofiles, and every query past this point assumes a schema this code
    /// has seen. Applying nothing and returning `Ok` reads as success and then
    /// queries a shape it does not understand, survivable for an added column
    /// and silent data loss for a dropped one, a rename, or a NOT NULL the
    /// older code never populates.
    ///
    /// Nothing exotic is needed to reach it: a rollback to an older release
    /// after a bad update, a restored backup, or one vault opened from two
    /// machines running different versions, which is the shape the
    /// vault-per-detachable-drive workflow is made of.
    #[instrument(skip_all)]
    fn migrate(&mut self) -> Result<(), DbError> {
        let version: i32 = self
            .conn
            .query_row("PRAGMA user_version", [], |row| row.get(0))?;

        if version > SCHEMA_VERSION {
            return Err(DbError::VaultTooNew {
                found: version,
                supported: SCHEMA_VERSION,
            });
        }

        for (i, sql) in MIGRATIONS.iter().enumerate() {
            let target = (i + 1) as i32;
            if version < target {
                let batch = format!("BEGIN;\n{sql}\nPRAGMA user_version = {target};\nCOMMIT;");
                match self.conn.execute_batch(&batch) {
                    Ok(()) => {}
                    Err(e) if e.to_string().contains("duplicate column") => {
                        // Recovery path: a prior partial migration committed
                        // some ALTERs before crashing. Re-run the migration in
                        // pieces, tolerating "duplicate column" on ALTERs and
                        // "already exists" on CREATEs (both mean: the prior
                        // partial run got there already; the desired final
                        // state is still reachable). Any OTHER error here is
                        // a real failure, we roll back and surface it,
                        // because silently bumping user_version on a partially
                        // applied schema is the worst possible outcome.
                        let _ = self.conn.execute_batch("ROLLBACK");
                        self.conn.execute_batch("BEGIN")?;

                        // ALTER TABLEs first, individually, tolerating duplicates.
                        for line in sql.lines() {
                            let trimmed = line.trim();
                            if trimmed.to_uppercase().starts_with("ALTER TABLE")
                                && trimmed.to_uppercase().contains("ADD COLUMN")
                                && let Err(alter_err) = self.conn.execute_batch(trimmed)
                                && !alter_err.to_string().contains("duplicate column")
                            {
                                let _ = self.conn.execute_batch("ROLLBACK");
                                return Err(DbError::Sqlite(alter_err));
                            }
                        }

                        // Non-ALTER statements (CREATE TABLE / INDEX /
                        // TRIGGER, DROP IF EXISTS, INSERT OR IGNORE, plain
                        // INSERT / UPDATE / DELETE). After M018, every
                        // migration from M003 onward is replay-safe by
                        // construction (verified by the
                        // migration_replay_from_version_two_against_full_schema
                        // regression test), so this batch should succeed
                        // cleanly even against a populated schema. "already
                        // exists" stays tolerable as a belt-and-braces guard
                        // for pre-idempotent migration bodies. Anything else
                        // is a real failure, fail fast, don't bump.
                        let non_alter: String = sql
                            .lines()
                            .filter(|l| {
                                let t = l.trim().to_uppercase();
                                !(t.starts_with("ALTER TABLE") && t.contains("ADD COLUMN"))
                            })
                            .collect::<Vec<_>>()
                            .join("\n");
                        if !non_alter.trim().is_empty()
                            && let Err(e) = self.conn.execute_batch(&non_alter)
                            && !e.to_string().contains("already exists")
                        {
                            let _ = self.conn.execute_batch("ROLLBACK");
                            return Err(DbError::Sqlite(e));
                        }

                        self.conn
                            .execute_batch(&format!("PRAGMA user_version = {target};\nCOMMIT;"))?;
                    }
                    Err(e) => return Err(DbError::Sqlite(e)),
                }
            }
        }

        Ok(())
    }

    /// Run a closure inside a SQLite transaction.
    ///
    /// Uses `BEGIN IMMEDIATE` to acquire a write lock upfront, preventing
    /// deadlocks when the closure issues writes. The closure receives a [`Tx`]
    /// token proving a transaction is open; pass it to row-write functions that
    /// require batching (the token cannot be constructed any other way, so those
    /// functions cannot be called in an un-batched per-row loop). The closure
    /// accesses the same `Database` through the shared `Mutex<Database>`, which is
    /// safe because the caller already holds the lock.
    #[instrument(skip_all)]
    pub fn transaction<T, F>(&self, f: F) -> Result<T, DbError>
    where
        F: FnOnce(&Tx) -> Result<T, DbError>,
    {
        self.conn.execute_batch("BEGIN IMMEDIATE")?;
        match f(&Tx(())) {
            Ok(val) => {
                self.conn.execute_batch("COMMIT")?;
                Ok(val)
            }
            Err(e) => {
                if let Err(rb_err) = self.conn.execute_batch("ROLLBACK") {
                    tracing::warn!("ROLLBACK failed after transaction error: {rb_err}");
                }
                Err(e)
            }
        }
    }

    /// [`transaction`](Self::transaction) for closures that produce a
    /// [`CoreError`](crate::error::CoreError), i.e. the row-write batchers in
    /// `analysis`, `rules`, and `harvest`, which call functions returning
    /// `CoreError`. Same `BEGIN IMMEDIATE` / commit / rollback semantics and the
    /// same [`Tx`] proof token; only the closure's error type differs.
    #[instrument(skip_all)]
    pub fn transaction_core<T, F>(&self, f: F) -> Result<T, crate::error::CoreError>
    where
        F: FnOnce(&Tx) -> Result<T, crate::error::CoreError>,
    {
        self.conn.execute_batch("BEGIN IMMEDIATE")?;
        match f(&Tx(())) {
            Ok(val) => {
                self.conn.execute_batch("COMMIT")?;
                Ok(val)
            }
            Err(e) => {
                if let Err(rb_err) = self.conn.execute_batch("ROLLBACK") {
                    tracing::warn!("ROLLBACK failed after transaction error: {rb_err}");
                }
                Err(e)
            }
        }
    }

    /// Borrow the underlying connection for queries.
    pub fn conn(&self) -> &Connection {
        &self.conn
    }
}

#[cfg(test)]
mod tests;