//! Database types — sqlx access through Mad Hatter.
//!
//! Engine code uses `Db` for all database operations. The internal `SqlitePool`
//! is never exposed — all queries return `Critical<T>`.
//!
//! # Usage
//!
//! ```ignore
//! // Fetch typed rows
//! let users = db.fetch_all(
//!     sqlx::query_as::<_, User>("SELECT * FROM users WHERE active = ?").bind(true)
//! ).await.suppress(token, "reason").unwrap();
//!
//! // Execute write
//! db.execute(
//!     sqlx::query("INSERT INTO users (name) VALUES (?)").bind(&name)
//! ).await.suppress(token, "reason").unwrap();
//!
//! // Transaction
//! let mut tx = db.begin().await.suppress(token, "reason").unwrap();
//! tx.execute(sqlx::query("UPDATE ...").bind(val)).await.suppress(token, "reason").unwrap();
//! tx.commit().await.suppress(token, "reason").unwrap();
//! ```

use sqlx::sqlite::{SqliteArguments, SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
use sqlx::{FromRow, Sqlite, SqlitePool};

use crate::critical::{Critical, IntoCritical};

use std::any::TypeId;
use std::collections::HashMap;
use std::sync::{Arc, OnceLock};
use tokio::sync::Mutex as TokioMutex;

// ─── Dynamic Data Source ───

/// Data source configuration — abstraction over database backends.
///
/// Currently supports SQLite; designed for future MySQL/Postgres extension.
#[derive(Debug, Clone)]
pub enum DsConfig {
    /// SQLite database file.
    Sqlite { path: String },
    // Future: MySql { url: String, pool_size: Option<u32> },
}

impl DsConfig {
    /// Create a SQLite data source configuration.
    pub fn sqlite(path: impl Into<String>) -> Self {
        DsConfig::Sqlite { path: path.into() }
    }
}

/// Trait for dynamic (per-instance) data sources.
///
/// Implemented by `dao!` macro with `resolver = ...` syntax.
/// The resolver maps a business parameter (e.g. instance_id) to a `DsConfig`.
pub trait DynamicDao: Send + Sync + 'static {
    /// Resolve a business parameter to a data source configuration.
    fn resolve(param: &str) -> DsConfig;

    /// Optional schema JSON for auto-reconciliation on first connect.
    fn schema() -> Option<&'static str> { None }

    /// Optional DaoConfig for pool tuning.
    fn dao_config() -> crate::dao::DaoConfig { crate::dao::DaoConfig::default() }
}

/// Global cache of dynamically-created connection pools.
///
/// Keyed by (TypeId of DynamicDao impl, parameter string).
/// No eviction policy — pools are lightweight and long-lived.
static DYNAMIC_POOL_CACHE: OnceLock<TokioMutex<HashMap<(TypeId, String), Db>>> = OnceLock::new();

fn pool_cache() -> &'static TokioMutex<HashMap<(TypeId, String), Db>> {
    DYNAMIC_POOL_CACHE.get_or_init(|| TokioMutex::new(HashMap::new()))
}

/// Get or create a connection pool for a dynamic data source.
///
/// Thread-safe: uses async mutex to prevent duplicate pool creation.
///
/// Double-check locking pattern:
/// - Fast path: check cache under lock (zero contention after first init)
/// - Slow path: hold lock for the entire create+reconcile+insert sequence
///   (only happens once per data source at startup; subsequent calls hit fast path)
pub async fn get_or_create_dynamic_pool<D: DynamicDao>(param: &str) -> Result<Db, String> {
    let key = (TypeId::of::<D>(), param.to_string());

    // Fast path: check cache under lock (all calls after first init take this path)
    {
        let cache = pool_cache().lock().await;
        if let Some(db) = cache.get(&key) {
            return Ok(db.clone());
        }
    }

    // Slow path: hold lock for entire create+reconcile+insert sequence.
    // Only one task can be here per key; others block until cache is populated.
    let mut cache = pool_cache().lock().await;

    // Double-check: another task may have created it while we waited for the lock
    if let Some(db) = cache.get(&key) {
        return Ok(db.clone());
    }

    // Create pool (holding lock — ensures only one pool per key is ever created)
    let config = D::resolve(param);
    let dao_config = D::dao_config();

    let db = match config {
        DsConfig::Sqlite { ref path } => {
            // Ensure parent directory exists
            if let Some(parent) = std::path::Path::new(path).parent() {
                if !parent.as_os_str().is_empty() {
                    std::fs::create_dir_all(parent)
                        .map_err(|e| format!("failed to create dir for {}: {}", path, e))?;
                }
            }
            Db::open(path, dao_config.pool_size, dao_config.busy_timeout_ms, dao_config.journal_mode_wal)
                .await
                .map_err(|e| format!("failed to open dynamic db {}: {}", path, e))?
        }
    };

    // Reconcile schema if provided (holding lock — prevents concurrent reconcile on same db)
    if let Some(schema_json) = D::schema() {
        crate::dao::reconcile_schema(&db, schema_json)
            .await
            .map_err(|e| format!("schema reconcile failed: {}", e))?;
    }

    cache.insert(key, db.clone());
    Ok(db)
}


/// Database handle wrapping an sqlx connection pool.
///
/// Engine cannot access the internal `SqlitePool`. All operations return
/// `Critical<T>` — engine must use `.propagate(scope, token)` or `.suppress(token, reason)`.
pub struct Db(SqlitePool);

impl Db {
    /// Create a Db from a pool. Called by `dao!` macro only.
    #[doc(hidden)]
    pub fn from_pool(pool: SqlitePool) -> Self {
        Db(pool)
    }

    /// Clone the internal pool. Used by TxScope for lazy BEGIN.
    pub(crate) fn pool_clone(&self) -> SqlitePool {
        self.0.clone()
    }

    /// Open a SQLite database with pool configuration.
    ///
    /// Called by `dao!` macro. Not part of the public API.
    #[doc(hidden)]
    pub async fn open(
        path: &str,
        pool_size: u32,
        busy_timeout_ms: u32,
        journal_mode_wal: bool,
    ) -> Result<Self, String> {
        use std::str::FromStr;

        let uri = format!("sqlite:{}", path);
        let mut options = SqliteConnectOptions::from_str(&uri)
            .map_err(|e| format!("invalid db path: {}", e))?
            .create_if_missing(true)
            .pragma("busy_timeout", busy_timeout_ms.to_string());

        if journal_mode_wal {
            options = options.pragma("journal_mode", "WAL");
        }

        let pool = SqlitePoolOptions::new()
            .max_connections(pool_size)
            .connect_with(options)
            .await
            .map_err(|e| format!("pool connection failed: {}", e))?;

        Ok(Db(pool))
    }

    /// Execute raw SQL (may contain multiple semicolon-separated statements).
    ///
    /// Used internally by `dao!` macro for init SQL and reconcile_schema.
    #[doc(hidden)]
    pub async fn execute_raw(&self, sql: &str) -> Result<(), String> {
        for stmt in sql.split(';') {
            let trimmed = stmt.trim();
            if !trimmed.is_empty() {
                sqlx::query(trimmed)
                    .execute(&self.0)
                    .await
                    .map_err(|e| format!("{}: {}", e, trimmed))?;
            }
        }
        Ok(())
    }

    /// Fetch scalar String values from a raw SQL query.
    ///
    /// Used internally by reconcile_schema for PRAGMA queries.
    #[doc(hidden)]
    pub async fn fetch_scalars(&self, sql: &str) -> Result<Vec<String>, String> {
        use sqlx::Row as _;
        let rows = sqlx::query(sql)
            .fetch_all(&self.0)
            .await
            .map_err(|e| e.to_string())?;
        let mut results = Vec::new();
        for row in rows {
            let val: String = row.try_get::<String, _>(0)
                .or_else(|_| row.try_get::<i64, _>(0).map(|v| v.to_string()))
                .or_else(|_| row.try_get::<f64, _>(0).map(|v| v.to_string()))
                .map_err(|e| e.to_string())?;
            results.push(val);
        }
        Ok(results)
    }

    /// Fetch all rows matching a typed query.
    ///
    /// Build the query with `sqlx::query_as()`:
    /// ```ignore
    /// db.fetch_all(sqlx::query_as("SELECT ...").bind(val)).await
    /// ```
    pub async fn fetch_all<'q, O>(
        &self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Vec<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_all(&self.0)
            .await
            .into_critical("db_fetch_all")
    }

    /// Fetch exactly one row.
    ///
    /// Returns a critical error if zero or more than one row matches.
    pub async fn fetch_one<'q, O>(
        &self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<O>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_one(&self.0)
            .await
            .into_critical("db_fetch_one")
    }

    /// Fetch zero or one row.
    pub async fn fetch_optional<'q, O>(
        &self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Option<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_optional(&self.0)
            .await
            .into_critical("db_fetch_optional")
    }

    /// Execute a write query, returning the number of rows affected.
    ///
    /// Build the query with `sqlx::query()`:
    /// ```ignore
    /// db.execute(sqlx::query("INSERT ...").bind(val)).await
    /// ```
    pub async fn execute<'q>(
        &self,
        query: sqlx::query::Query<'q, Sqlite, SqliteArguments<'q>>,
    ) -> Critical<u64> {
        query
            .execute(&self.0)
            .await
            .map(|r| r.rows_affected())
            .into_critical("db_execute")
    }

    /// Begin a transaction.
    ///
    /// The returned `Tx` auto-rolls back on drop if not committed.
    pub async fn begin(&self) -> Critical<Tx> {
        self.0
            .begin()
            .await
            .map(Tx)
            .into_critical("db_begin")
    }

    /// Begin a transaction (framework-internal). Returns Result instead of Critical.
    /// Begin a transaction, returning Result instead of Critical.
    /// For internal framework use only.
    #[doc(hidden)]
    pub async fn begin_inner(&self) -> Result<Tx, String> {
        self.0
            .begin()
            .await
            .map(Tx)
            .map_err(|e| e.to_string())
    }
}

impl Clone for Db {
    fn clone(&self) -> Self {
        Db(self.0.clone())
    }
}

/// Transaction handle. Automatically rolls back on drop if not committed.
///
/// All query methods take `&mut self` because sqlx transactions require
/// mutable access to the underlying connection.
pub struct Tx(sqlx::Transaction<'static, Sqlite>);

impl Tx {
    /// Create Tx from raw sqlx transaction (framework-internal).
    pub(crate) fn from_sqlx(inner: sqlx::Transaction<'static, Sqlite>) -> Self {
        Tx(inner)
    }

    /// Fetch all rows matching a typed query within this transaction.
    pub async fn fetch_all<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Vec<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_all(&mut *self.0)
            .await
            .into_critical("tx_fetch_all")
    }

    /// Fetch exactly one row within this transaction.
    pub async fn fetch_one<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<O>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_one(&mut *self.0)
            .await
            .into_critical("tx_fetch_one")
    }

    /// Fetch zero or one row within this transaction.
    pub async fn fetch_optional<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Option<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        query
            .fetch_optional(&mut *self.0)
            .await
            .into_critical("tx_fetch_optional")
    }

    /// Execute a write query within this transaction.
    pub async fn execute<'q>(
        &mut self,
        query: sqlx::query::Query<'q, Sqlite, SqliteArguments<'q>>,
    ) -> Critical<u64> {
        query
            .execute(&mut *self.0)
            .await
            .map(|r| r.rows_affected())
            .into_critical("tx_execute")
    }

    /// Commit the transaction. Consumes self.
    ///
    /// If this method is not called, the transaction is rolled back on drop.
    pub async fn commit(self) -> Critical<()> {
        self.0.commit().await.into_critical("tx_commit")
    }

    /// Raw commit for framework internal use. Returns Result instead of Critical.
    /// Commit transaction, returning Result instead of Critical.
    /// For internal framework use only.
    #[doc(hidden)]
    pub async fn commit_inner(self) -> Result<(), String> {
        self.0.commit().await.map_err(|e| e.to_string())
    }

    /// Execute SAVEPOINT within this transaction (framework-internal).
    pub(crate) async fn savepoint(&mut self, name: &str) -> Result<(), String> {
        sqlx::query(&format!("SAVEPOINT {}", name))
            .execute(&mut *self.0)
            .await
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Release a savepoint (framework-internal).
    pub(crate) async fn release_savepoint(&mut self, name: &str) -> Result<(), String> {
        sqlx::query(&format!("RELEASE SAVEPOINT {}", name))
            .execute(&mut *self.0)
            .await
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Rollback to a savepoint (framework-internal).
    pub(crate) async fn rollback_to_savepoint(&mut self, name: &str) -> Result<(), String> {
        sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", name))
            .execute(&mut *self.0)
            .await
            .map(|_| ())
            .map_err(|e| e.to_string())
    }
}

// ─── ScopedDb ───

/// Scoped database handle bound to a transaction scope.
///
/// Obtained via `scope.db().await`. Provides the same query methods as `Tx`
/// but tied to the scope's lifecycle. The DB transaction is lazily begun
/// on first access and auto-committed when the scope completes.
///
/// `ScopedDb` does NOT expose `begin()` — sub-transactions are managed
/// exclusively through `TxScope::suppress()` / `TxScope::propagate()`.
pub struct ScopedDb(pub(crate) tokio::sync::OwnedMutexGuard<Option<Tx>>);

impl ScopedDb {
    /// Get mutable reference to the active transaction.
    fn tx(&mut self) -> &mut Tx {
        self.0.as_mut().expect("ScopedDb: transaction not active")
    }

    /// Fetch all rows matching a typed query.
    pub async fn fetch_all<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Vec<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        self.tx().fetch_all(query).await
    }

    /// Fetch exactly one row.
    pub async fn fetch_one<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<O>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        self.tx().fetch_one(query).await
    }

    /// Fetch zero or one row.
    pub async fn fetch_optional<'q, O>(
        &mut self,
        query: sqlx::query::QueryAs<'q, Sqlite, O, SqliteArguments<'q>>,
    ) -> Critical<Option<O>>
    where
        O: for<'r> FromRow<'r, SqliteRow> + Send + Unpin,
    {
        self.tx().fetch_optional(query).await
    }

    /// Execute a write query, returning the number of rows affected.
    pub async fn execute<'q>(
        &mut self,
        query: sqlx::query::Query<'q, Sqlite, SqliteArguments<'q>>,
    ) -> Critical<u64> {
        self.tx().execute(query).await
    }
}

/// Test helper: open a SQLite database at `path`, optionally apply JSON schema,
/// begin a transaction, and return `(Db, ScopedDb)`.
///
/// Caller must keep `Db` alive for the duration of `ScopedDb` usage.
/// Bypasses the DynamicDao resolver — intended for unit tests only.
#[doc(hidden)]
pub async fn open_test_db(path: &str, schema_json: Option<&str>) -> (Db, ScopedDb) {
    // Ensure parent directory exists
    if let Some(parent) = std::path::Path::new(path).parent() {
        std::fs::create_dir_all(parent).expect("open_test_db: create dir");
    }
    let db = Db::open(path, 5, 5000, true).await.expect("open_test_db: open failed");
    if let Some(json) = schema_json {
        crate::dao::reconcile_schema(&db, json).await.expect("open_test_db: schema reconcile failed");
    }
    let tx = db.begin_inner().await.expect("open_test_db: begin failed");
    let arc = std::sync::Arc::new(tokio::sync::Mutex::new(Some(tx)));
    let guard = arc.clone().lock_owned().await;
    (db, ScopedDb(guard))
}

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

    #[derive(Debug, sqlx::FromRow, PartialEq)]
    struct TestRow {
        id: i64,
        name: String,
    }

    // Token for testing
    #[derive(Debug, Clone, Copy)]
    enum TestToken {
        DbOp,
    }
    impl crate::critical::ConceptMapToken for TestToken {
        fn operation_name(&self) -> &'static str {
            "test::db_op"
        }
    }

    impl crate::critical::SuppressibleToken for TestToken {}

    async fn setup_db() -> Db {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();

        sqlx::query("CREATE TABLE test_rows (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
            .execute(&pool)
            .await
            .unwrap();

        Db::from_pool(pool)
    }

    #[tokio::test]
    async fn test_execute_and_fetch_all() {
        let db = setup_db().await;

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(2i64)
                .bind("bob"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows ORDER BY id"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].name, "alice");
        assert_eq!(rows[1].name, "bob");
    }

    #[tokio::test]
    async fn test_fetch_one() {
        let db = setup_db().await;

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        let row: TestRow = db
            .fetch_one(sqlx::query_as("SELECT id, name FROM test_rows WHERE id = ?").bind(1i64))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(row.name, "alice");
    }

    #[tokio::test]
    async fn test_fetch_optional() {
        let db = setup_db().await;

        let row: Option<TestRow> = db
            .fetch_optional(
                sqlx::query_as("SELECT id, name FROM test_rows WHERE id = ?").bind(999i64),
            )
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert!(row.is_none());
    }

    #[tokio::test]
    async fn test_transaction_commit() {
        let db = setup_db().await;

        let mut tx = db.begin().await.suppress(TestToken::DbOp, "test").unwrap();

        tx.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        tx.commit().await.suppress(TestToken::DbOp, "test").unwrap();

        // Verify data persisted
        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 1);
    }

    #[tokio::test]
    async fn test_transaction_rollback_on_drop() {
        let db = setup_db().await;

        {
            let mut tx = db.begin().await.suppress(TestToken::DbOp, "test").unwrap();

            tx.execute(
                sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                    .bind(1i64)
                    .bind("alice"),
            )
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

            // tx dropped without commit — should rollback
        }

        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 0);
    }

    #[tokio::test]
    async fn test_clone() {
        let db = setup_db().await;
        let db2 = db.clone();

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        let rows: Vec<TestRow> = db2
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 1);
    }

    #[tokio::test]
    async fn test_execute_returns_rows_affected() {
        let db = setup_db().await;

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        db.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(2i64)
                .bind("bob"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        let affected = db
            .execute(sqlx::query("DELETE FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(affected, 2);
    }

    #[tokio::test]
    async fn test_open() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let path_str = db_path.to_str().unwrap();

        let db = Db::open(path_str, 2, 5000, true).await.unwrap();

        db.execute_raw("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
            .await
            .unwrap();

        db.execute(
            sqlx::query("INSERT INTO t (id, name) VALUES (?, ?)").bind(1i64).bind("test"),
        )
        .await
        .suppress(TestToken::DbOp, "test")
        .unwrap();

        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM t"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].name, "test");
    }

    #[tokio::test]
    async fn test_execute_raw_multi_statement() {
        let db = setup_db().await;

        db.execute_raw(
            "INSERT INTO test_rows (id, name) VALUES (1, 'a');
             INSERT INTO test_rows (id, name) VALUES (2, 'b')"
        )
        .await
        .unwrap();

        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows ORDER BY id"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();

        assert_eq!(rows.len(), 2);
    }

    #[tokio::test]
    async fn test_fetch_scalars() {
        let db = setup_db().await;

        db.execute_raw(
            "INSERT INTO test_rows (id, name) VALUES (1, 'alice');
             INSERT INTO test_rows (id, name) VALUES (2, 'bob')"
        )
        .await
        .unwrap();

        let names: Vec<String> = db
            .fetch_scalars("SELECT name FROM test_rows ORDER BY id")
            .await
            .unwrap();

        assert_eq!(names, vec!["alice", "bob"]);
    }

    #[tokio::test]
    async fn test_scoped_db_fetch_and_execute() {
        let db = setup_db().await;
        // Begin a transaction manually
        let tx = db.begin_inner().await.unwrap();
        let mutex = std::sync::Arc::new(tokio::sync::Mutex::new(Some(tx)));
        let guard = mutex.clone().lock_owned().await;
        let mut scoped = ScopedDb(guard);

        scoped.execute(
            sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                .bind(1i64)
                .bind("alice"),
        ).await.suppress(TestToken::DbOp, "test").unwrap();

        let rows: Vec<TestRow> = scoped
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].name, "alice");

        let one: TestRow = scoped
            .fetch_one(sqlx::query_as("SELECT id, name FROM test_rows WHERE id = 1"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();
        assert_eq!(one.name, "alice");

        let opt: Option<TestRow> = scoped
            .fetch_optional(sqlx::query_as("SELECT id, name FROM test_rows WHERE id = 99"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();
        assert!(opt.is_none());
    }

    #[tokio::test]
    async fn test_scoped_db_commit_inner() {
        let db = setup_db().await;
        // Insert via transaction, commit, verify persisted
        let tx = db.begin_inner().await.unwrap();
        let mutex = std::sync::Arc::new(tokio::sync::Mutex::new(Some(tx)));
        {
            let guard = mutex.clone().lock_owned().await;
            let mut scoped = ScopedDb(guard);
            scoped.execute(
                sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                    .bind(1i64)
                    .bind("committed"),
            ).await.suppress(TestToken::DbOp, "test").unwrap();
            // Drop ScopedDb (releases MutexGuard)
        }
        // Take the Tx out and commit
        let tx = std::sync::Arc::try_unwrap(mutex).ok().expect("Arc still shared").into_inner().unwrap();
        tx.commit_inner().await.unwrap();

        // Verify committed
        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].name, "committed");
    }

    #[tokio::test]
    async fn test_scoped_db_rollback_on_drop() {
        let db = setup_db().await;
        // Insert via transaction, drop without commit → rollback
        {
            let tx = db.begin_inner().await.unwrap();
            let mutex = std::sync::Arc::new(tokio::sync::Mutex::new(Some(tx)));
            let guard = mutex.clone().lock_owned().await;
            let mut scoped = ScopedDb(guard);
            scoped.execute(
                sqlx::query("INSERT INTO test_rows (id, name) VALUES (?, ?)")
                    .bind(1i64)
                    .bind("rolled_back"),
            ).await.suppress(TestToken::DbOp, "test").unwrap();
            // Drop scoped → drop guard → drop mutex → drop Tx → auto rollback
        }

        // Verify nothing persisted
        let rows: Vec<TestRow> = db
            .fetch_all(sqlx::query_as("SELECT id, name FROM test_rows"))
            .await
            .suppress(TestToken::DbOp, "test")
            .unwrap();
        assert_eq!(rows.len(), 0);
    }

}

