use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, OnceLock};

use std::time::Duration;

use crate::column::ColumnValue;
use crate::condition::Condition;
use crate::driver::QueryResult;
use crate::error::{Error, Result};
use crate::middleware::{MiddlewareChain, NextExecute, OperationType, RequestContext};
use crate::model::{
    count, create, create_many, delete, find, find_by_id, find_one, query, update, update_many,
    Executor, Model,
};
use crate::types::{
    DbConfig, DbConfigOptions, DeleteOptions, DriverType, InsertOptions, LimitConfig, PkeyResult,
    SelectOptions, TransactionOptions, UpdateManyOptions, UpdateOptions,
};
use crate::values::Value;

static GLOBAL_HANDLER: OnceLock<DbHandler> = OnceLock::new();

#[derive(Clone)]
pub struct DbHandler {
    inner: Arc<DbHandlerInner>,
}

struct DbHandlerInner {
    config: DbConfig,
    limit_config: LimitConfig,
    pool: Pool,
    writer_pool: Option<Pool>,
    middleware_chain: MiddlewareChain,
    /// Fallback context used when DbHandler is used directly (not via DbScope).
    default_ctx: RequestContext,
}

/// Abstract pool that wraps database connection pools behind a driver-agnostic interface.
#[derive(Clone)]
pub enum Pool {
    #[cfg(feature = "postgres")]
    Postgres(deadpool_postgres::Pool),
    #[cfg(feature = "mysql")]
    Mysql(sqlx::MySqlPool),
    #[cfg(feature = "sqlite")]
    Sqlite(sqlx::SqlitePool),
    /// Placeholder for when no driver is compiled in or for testing.
    None,
}

/// Real transaction, wrapping the driver-specific transaction type.
pub(crate) enum PoolTransaction {
    #[cfg(feature = "postgres")]
    Postgres(deadpool_postgres::Object),
    #[cfg(feature = "mysql")]
    Mysql(sqlx::Transaction<'static, sqlx::MySql>),
    #[cfg(feature = "sqlite")]
    Sqlite(sqlx::Transaction<'static, sqlx::Sqlite>),
    #[allow(dead_code)]
    None,
}

impl PoolTransaction {
    async fn execute(&mut self, sql: &str, params: &[Value]) -> Result<QueryResult> {
        match self {
            #[cfg(feature = "postgres")]
            Self::Postgres(conn) => run_pg_query(conn, sql, params).await,
            #[cfg(feature = "mysql")]
            Self::Mysql(tx) => run_mysql_query(&mut **tx, sql, params).await,
            #[cfg(feature = "sqlite")]
            Self::Sqlite(tx) => run_sqlite_query(&mut **tx, sql, params).await,
            Self::None => Err(Error::Driver("No database driver configured".to_string())),
        }
    }

    async fn commit(self) -> Result<()> {
        match self {
            #[cfg(feature = "postgres")]
            Self::Postgres(conn) => {
                conn.batch_execute("COMMIT")
                    .await
                    .map_err(Error::Postgres)?;
                Ok(())
            }
            #[cfg(feature = "mysql")]
            Self::Mysql(tx) => tx.commit().await.map_err(Error::Database),
            #[cfg(feature = "sqlite")]
            Self::Sqlite(tx) => tx.commit().await.map_err(Error::Database),
            Self::None => Ok(()),
        }
    }

    async fn rollback(self) -> Result<()> {
        match self {
            #[cfg(feature = "postgres")]
            Self::Postgres(conn) => {
                conn.batch_execute("ROLLBACK")
                    .await
                    .map_err(Error::Postgres)?;
                Ok(())
            }
            #[cfg(feature = "mysql")]
            Self::Mysql(tx) => tx.rollback().await.map_err(Error::Database),
            #[cfg(feature = "sqlite")]
            Self::Sqlite(tx) => tx.rollback().await.map_err(Error::Database),
            Self::None => Ok(()),
        }
    }
}

async fn begin_transaction(pool: &Pool) -> Result<PoolTransaction> {
    match pool {
        #[cfg(feature = "postgres")]
        Pool::Postgres(p) => {
            let conn = p.get().await.map_err(|e| Error::PoolError(e.to_string()))?;
            conn.batch_execute("BEGIN").await.map_err(Error::Postgres)?;
            Ok(PoolTransaction::Postgres(conn))
        }
        #[cfg(feature = "mysql")]
        Pool::Mysql(p) => {
            let tx = p.begin().await.map_err(Error::Database)?;
            Ok(PoolTransaction::Mysql(tx))
        }
        #[cfg(feature = "sqlite")]
        Pool::Sqlite(p) => {
            let tx = p.begin().await.map_err(Error::Database)?;
            Ok(PoolTransaction::Sqlite(tx))
        }
        Pool::None => Err(Error::Driver("No database driver configured".to_string())),
    }
}

// ============================================
// DbHandler methods
// ============================================

impl DbHandler {
    pub fn config(&self) -> &DbConfig {
        &self.inner.config
    }

    pub fn limit_config(&self) -> &LimitConfig {
        &self.inner.limit_config
    }

    pub fn driver_type(&self) -> DriverType {
        self.inner.config.driver
    }

    pub fn pool(&self) -> &Pool {
        &self.inner.pool
    }

    pub fn writer_pool(&self) -> Option<&Pool> {
        self.inner.writer_pool.as_ref()
    }

    pub fn middleware_chain(&self) -> &MiddlewareChain {
        &self.inner.middleware_chain
    }

    /// Set the global handler (called once at startup).
    pub fn set_global(handler: DbHandler) {
        GLOBAL_HANDLER.set(handler).ok();
    }

    /// Get the global handler.
    pub fn global() -> &'static DbHandler {
        GLOBAL_HANDLER
            .get()
            .expect("DbHandler not initialized. Call DbHandler::set_global() first.")
    }

    /// Run a function inside a real database transaction.
    /// Issues BEGIN, runs the closure, then COMMIT on success or ROLLBACK on error.
    /// Creates a fresh `RequestContext`. For per-request state, use `DbScope::transaction()`.
    pub async fn transaction<F, Fut, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(TransactionExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let writer_pool = self.inner.writer_pool.as_ref().unwrap_or(&self.inner.pool);
        let pool_tx = begin_transaction(writer_pool).await?;
        let tx_arc = Arc::new(tokio::sync::Mutex::new(Some(pool_tx)));

        let tx_executor = TransactionExecutor {
            tx: Arc::clone(&tx_arc),
            handler: self.clone(),
            rollback_only: false,
            depth: 0,
            ctx: Arc::new(RequestContext::new()),
        };

        match f(tx_executor).await {
            Ok(result) => {
                let mut guard = tx_arc.lock().await;
                if let Some(pool_tx) = guard.take() {
                    pool_tx.commit().await?;
                }
                Ok(result)
            }
            Err(e) => {
                let mut guard = tx_arc.lock().await;
                if let Some(pool_tx) = guard.take() {
                    let _ = pool_tx.rollback().await;
                }
                Err(e)
            }
        }
    }

    /// Run a function inside a transaction with retry support.
    /// The closure must be `Fn` (not `FnOnce`) to allow retries.
    /// Creates a fresh `RequestContext`. For per-request state, use `DbScope::transaction_with_options()`.
    pub async fn transaction_with_options<F, Fut, R>(
        &self,
        f: F,
        options: TransactionOptions,
    ) -> Result<R>
    where
        F: Fn(TransactionExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let retry_limit = if options.retry_on_error {
            options.retry_limit
        } else {
            0
        };
        let mut attempts = 0;

        loop {
            let writer_pool = self.inner.writer_pool.as_ref().unwrap_or(&self.inner.pool);
            let pool_tx = begin_transaction(writer_pool).await?;
            let tx_arc = Arc::new(tokio::sync::Mutex::new(Some(pool_tx)));

            let tx_executor = TransactionExecutor {
                tx: Arc::clone(&tx_arc),
                handler: self.clone(),
                rollback_only: options.rollback_only,
                depth: 0,
                ctx: Arc::new(RequestContext::new()),
            };

            match f(tx_executor).await {
                Ok(result) => {
                    let mut guard = tx_arc.lock().await;
                    if let Some(pool_tx) = guard.take() {
                        if options.rollback_only {
                            let _ = pool_tx.rollback().await;
                        } else {
                            pool_tx.commit().await?;
                        }
                    }
                    return Ok(result);
                }
                Err(e) => {
                    let mut guard = tx_arc.lock().await;
                    if let Some(pool_tx) = guard.take() {
                        let _ = pool_tx.rollback().await;
                    }
                    attempts += 1;
                    if attempts > retry_limit {
                        return Err(e);
                    }
                    tracing::warn!("Transaction attempt {} failed, retrying: {}", attempts, e);
                    tokio::time::sleep(std::time::Duration::from_millis(options.retry_duration_ms))
                        .await;
                }
            }
        }
    }

    /// Execute a function with explicit writer connection access for reads.
    /// Write operations are NOT allowed in this context (read-only).
    /// Creates a fresh `RequestContext`. For per-request state, use `DbScope::with_writer()`.
    pub async fn with_writer<F, Fut, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(WriterExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let executor = WriterExecutor {
            handler: self.clone(),
            ctx: Arc::new(RequestContext::new()),
        };
        f(executor).await
    }

    /// Create a DbScope for this handler (convenience for multi-database patterns).
    pub fn scope(&self) -> DbScope {
        DbScope::new(self.clone())
    }

    /// Close all connection pools.
    pub async fn close(&self) {
        match &self.inner.pool {
            #[cfg(feature = "postgres")]
            Pool::Postgres(pool) => pool.close(),
            #[cfg(feature = "mysql")]
            Pool::Mysql(pool) => pool.close().await,
            #[cfg(feature = "sqlite")]
            Pool::Sqlite(pool) => pool.close().await,
            Pool::None => {}
        }
        if let Some(writer) = &self.inner.writer_pool {
            match writer {
                #[cfg(feature = "postgres")]
                Pool::Postgres(pool) => pool.close(),
                #[cfg(feature = "mysql")]
                Pool::Mysql(pool) => pool.close().await,
                #[cfg(feature = "sqlite")]
                Pool::Sqlite(pool) => pool.close().await,
                Pool::None => {}
            }
        }
    }
}

// ============================================
// DbScope - Multi-database convenience
// ============================================

/// Associates a DbHandler with models for independent database scopes.
/// Use when connecting to multiple databases (e.g. main DB vs analytics DB).
///
/// # Example
/// ```ignore
/// let main_db = DbScope::new(main_handler);
/// let analytics_db = DbScope::new(analytics_handler);
/// let users = main_db.find::<User>(&[], None).await?;
/// let reports = analytics_db.find::<Report>(&[], None).await?;
/// ```
#[derive(Clone)]
pub struct DbScope {
    handler: DbHandler,
    ctx: Arc<RequestContext>,
}

impl DbScope {
    pub fn new(handler: DbHandler) -> Self {
        Self {
            handler,
            ctx: Arc::new(RequestContext::new()),
        }
    }

    pub fn handler(&self) -> &DbHandler {
        &self.handler
    }

    /// Access the per-request context (type-map).
    pub fn ctx(&self) -> &RequestContext {
        &self.ctx
    }

    pub async fn find<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find(
            &self.handler,
            conditions,
            options,
            Some(self.handler.limit_config()),
        )
        .await
    }

    pub async fn find_one<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Option<M>> {
        find_one(&self.handler, conditions, options).await
    }

    pub async fn find_by_id<M: Model>(
        &self,
        pk_values: &[Vec<Value>],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find_by_id(&self.handler, pk_values, options).await
    }

    pub async fn count<M: Model>(&self, conditions: &[Condition]) -> Result<i64> {
        count::<M, _>(&self.handler, conditions).await
    }

    pub async fn query<M: Model>(&self, sql: &str, params: &[Value]) -> Result<Vec<M>> {
        query::<M, _>(&self.handler, sql, params).await
    }

    /// Run a function inside a transaction, inheriting this scope's context.
    pub async fn transaction<F, Fut, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(TransactionExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let writer_pool = self
            .handler
            .inner
            .writer_pool
            .as_ref()
            .unwrap_or(&self.handler.inner.pool);
        let pool_tx = begin_transaction(writer_pool).await?;
        let tx_arc = Arc::new(tokio::sync::Mutex::new(Some(pool_tx)));

        let tx_executor = TransactionExecutor {
            tx: Arc::clone(&tx_arc),
            handler: self.handler.clone(),
            rollback_only: false,
            depth: 0,
            ctx: Arc::clone(&self.ctx),
        };

        match f(tx_executor).await {
            Ok(result) => {
                let mut guard = tx_arc.lock().await;
                if let Some(pool_tx) = guard.take() {
                    pool_tx.commit().await?;
                }
                Ok(result)
            }
            Err(e) => {
                let mut guard = tx_arc.lock().await;
                if let Some(pool_tx) = guard.take() {
                    let _ = pool_tx.rollback().await;
                }
                Err(e)
            }
        }
    }

    /// Run a function inside a transaction with retry support, inheriting this scope's context.
    pub async fn transaction_with_options<F, Fut, R>(
        &self,
        f: F,
        options: TransactionOptions,
    ) -> Result<R>
    where
        F: Fn(TransactionExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let retry_limit = if options.retry_on_error {
            options.retry_limit
        } else {
            0
        };
        let mut attempts = 0;

        loop {
            let writer_pool = self
                .handler
                .inner
                .writer_pool
                .as_ref()
                .unwrap_or(&self.handler.inner.pool);
            let pool_tx = begin_transaction(writer_pool).await?;
            let tx_arc = Arc::new(tokio::sync::Mutex::new(Some(pool_tx)));

            let tx_executor = TransactionExecutor {
                tx: Arc::clone(&tx_arc),
                handler: self.handler.clone(),
                rollback_only: options.rollback_only,
                depth: 0,
                ctx: Arc::clone(&self.ctx),
            };

            match f(tx_executor).await {
                Ok(result) => {
                    let mut guard = tx_arc.lock().await;
                    if let Some(pool_tx) = guard.take() {
                        if options.rollback_only {
                            let _ = pool_tx.rollback().await;
                        } else {
                            pool_tx.commit().await?;
                        }
                    }
                    return Ok(result);
                }
                Err(e) => {
                    let mut guard = tx_arc.lock().await;
                    if let Some(pool_tx) = guard.take() {
                        let _ = pool_tx.rollback().await;
                    }
                    attempts += 1;
                    if attempts > retry_limit {
                        return Err(e);
                    }
                    tracing::warn!("Transaction attempt {} failed, retrying: {}", attempts, e);
                    tokio::time::sleep(std::time::Duration::from_millis(options.retry_duration_ms))
                        .await;
                }
            }
        }
    }

    /// Execute a function with writer connection access, inheriting this scope's context.
    pub async fn with_writer<F, Fut, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(WriterExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let executor = WriterExecutor {
            handler: self.handler.clone(),
            ctx: Arc::clone(&self.ctx),
        };
        f(executor).await
    }

    pub async fn create<M: Model>(
        &self,
        values: &[ColumnValue],
        options: Option<&InsertOptions>,
    ) -> Result<QueryResult> {
        self.transaction(|tx| async move { create::<M, _>(&tx, values, options).await })
            .await
    }

    pub async fn create_many<M: Model>(
        &self,
        rows: &[Vec<ColumnValue>],
        options: Option<&InsertOptions>,
    ) -> Result<Option<PkeyResult>> {
        self.transaction(|tx| async move { create_many::<M, _>(&tx, rows, options).await })
            .await
    }

    pub async fn update<M: Model>(
        &self,
        conditions: &[Condition],
        values: &[ColumnValue],
        options: Option<&UpdateOptions>,
    ) -> Result<QueryResult> {
        self.transaction(|tx| async move { update::<M, _>(&tx, conditions, values, options).await })
            .await
    }

    pub async fn update_many<M: Model>(
        &self,
        rows: &[Vec<ColumnValue>],
        options: &UpdateManyOptions,
    ) -> Result<Option<PkeyResult>> {
        self.transaction(|tx| async move { update_many::<M, _>(&tx, rows, options).await })
            .await
    }

    pub async fn delete<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&DeleteOptions>,
    ) -> Result<QueryResult> {
        self.transaction(|tx| async move { delete::<M, _>(&tx, conditions, options).await })
            .await
    }
}

// ============================================
// task_local for request-scoped DbScope
// ============================================

tokio::task_local! {
    static DB_SCOPE: DbScope;
}

impl DbScope {
    /// Run an async block with this DbScope available via `task_local`.
    /// Inside the block, use `DbScope::current()` to retrieve it.
    pub async fn run<F, Fut, R>(self, f: F) -> R
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = R>,
    {
        DB_SCOPE.scope(self, f()).await
    }

    /// Get the current request-scoped DbScope from `task_local`.
    /// Panics if not inside a `DbScope::run()` block.
    pub fn current() -> DbScope {
        DB_SCOPE.with(|db| db.clone())
    }

    /// Try to get the current DbScope. Returns `None` if not in scope.
    pub fn try_current() -> Option<DbScope> {
        DB_SCOPE.try_with(|db| db.clone()).ok()
    }
}

// ============================================
// Middleware-wrapped execution helper
// ============================================

fn execute_with_middleware_op<'a>(
    chain: &'a MiddlewareChain,
    sql: &'a str,
    params: &'a [Value],
    core: NextExecute<'a>,
    ctx: &'a RequestContext,
    op: OperationType,
    table: &'a str,
) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
    if chain.is_empty() {
        return core.run(sql, params);
    }

    Box::pin(run_middleware_chain_op(
        chain.list(),
        0,
        sql,
        params,
        core,
        ctx,
        op,
        table,
    ))
}

fn run_middleware_chain_op<'a>(
    middlewares: &'a [Arc<dyn crate::middleware::Middleware>],
    index: usize,
    sql: &'a str,
    params: &'a [Value],
    core: NextExecute<'a>,
    ctx: &'a RequestContext,
    op: OperationType,
    table: &'a str,
) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
    if index >= middlewares.len() {
        return core.run(sql, params);
    }
    let mw = &middlewares[index];
    let remaining = middlewares;
    let idx = index + 1;
    let next = NextExecute::new(move |s: &'a str, p: &'a [Value]| {
        let core = core.clone();
        run_middleware_chain_op(remaining, idx, s, p, core, ctx, op, table)
    });
    match op {
        OperationType::Find => mw.on_find(ctx, table, sql, params, next),
        OperationType::Create => mw.on_create(ctx, table, sql, params, next),
        OperationType::Update => mw.on_update(ctx, table, sql, params, next),
        OperationType::Delete => mw.on_delete(ctx, table, sql, params, next),
        OperationType::Other => mw.on_execute(ctx, sql, params, next),
    }
}

// ============================================
// Executor impl for DbHandler
// ============================================

impl Executor for DbHandler {
    fn execute_query<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_query_with_op(sql, params, OperationType::Other, "")
    }

    fn execute_query_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        op: OperationType,
        table: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        Box::pin(async move {
            let core = NextExecute::new(|s: &'a str, p: &'a [Value]| {
                Box::pin(execute_on_pool(&self.inner.pool, s, p))
                    as Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>>
            });
            execute_with_middleware_op(
                &self.inner.middleware_chain,
                sql,
                params,
                core,
                &self.inner.default_ctx,
                op,
                table,
            )
            .await
        })
    }

    fn execute_write<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_write_with_op(sql, params, OperationType::Other, "")
    }

    fn execute_write_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        op: OperationType,
        table: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        Box::pin(async move {
            let pool = self.inner.writer_pool.as_ref().unwrap_or(&self.inner.pool);
            let core = NextExecute::new(move |s: &'a str, p: &'a [Value]| {
                Box::pin(execute_on_pool(pool, s, p))
                    as Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>>
            });
            execute_with_middleware_op(
                &self.inner.middleware_chain,
                sql,
                params,
                core,
                &self.inner.default_ctx,
                op,
                table,
            )
            .await
        })
    }

    fn driver_type(&self) -> DriverType {
        self.inner.config.driver
    }

    fn is_transaction(&self) -> bool {
        false
    }

    #[cfg(feature = "postgres")]
    fn execute_pg_rows<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Option<Pin<Box<dyn Future<Output = Result<Vec<tokio_postgres::Row>>> + Send + 'a>>> {
        if !self.inner.middleware_chain.is_empty() {
            return None;
        }
        match &self.inner.pool {
            Pool::Postgres(pg_pool) => Some(Box::pin(async move {
                let conn = pg_pool
                    .get()
                    .await
                    .map_err(|e| Error::PoolError(e.to_string()))?;
                run_pg_rows(&conn, sql, params).await
            })),
            _ => None,
        }
    }

    fn clone_arc(&self) -> Arc<dyn Executor> {
        Arc::new(self.clone())
    }
}

// ============================================
// TransactionExecutor -- backed by a real database transaction
// ============================================

/// Executor bound to a real database transaction.
/// All queries go through the same connection that issued BEGIN.
/// Supports nested transactions via savepoints (depth > 0).
pub struct TransactionExecutor {
    tx: Arc<tokio::sync::Mutex<Option<PoolTransaction>>>,
    handler: DbHandler,
    rollback_only: bool,
    /// Nesting depth: 0 = outermost (real tx), 1+ = savepoint level.
    depth: u32,
    ctx: Arc<RequestContext>,
}

impl Clone for TransactionExecutor {
    fn clone(&self) -> Self {
        Self {
            tx: Arc::clone(&self.tx),
            handler: self.handler.clone(),
            rollback_only: self.rollback_only,
            depth: self.depth,
            ctx: Arc::clone(&self.ctx),
        }
    }
}

impl TransactionExecutor {
    pub fn is_rollback_only(&self) -> bool {
        self.rollback_only
    }

    pub fn handler(&self) -> &DbHandler {
        &self.handler
    }

    /// Access the per-request context (type-map), inherited from the parent scope.
    pub fn ctx(&self) -> &RequestContext {
        &self.ctx
    }

    /// Run a nested transaction using a savepoint.
    /// When called inside an existing transaction:
    /// - Issues SAVEPOINT sp_N before the closure
    /// - On success: RELEASE SAVEPOINT sp_N
    /// - On error: ROLLBACK TO SAVEPOINT sp_N
    pub async fn transaction<F, Fut, R>(&self, f: F) -> Result<R>
    where
        F: FnOnce(TransactionExecutor) -> Fut,
        Fut: Future<Output = Result<R>>,
    {
        let next_depth = self.depth + 1;
        let savepoint_name = format!("sp_{}", next_depth);

        // Issue SAVEPOINT
        {
            let mut guard = self.tx.lock().await;
            let pool_tx = guard
                .as_mut()
                .ok_or_else(|| Error::Transaction("Transaction already completed".to_string()))?;
            let sql = format!("SAVEPOINT {}", savepoint_name);
            pool_tx.execute(&sql, &[]).await?;
        }

        let nested = TransactionExecutor {
            tx: Arc::clone(&self.tx),
            handler: self.handler.clone(),
            rollback_only: self.rollback_only,
            depth: next_depth,
            ctx: Arc::clone(&self.ctx),
        };

        match f(nested).await {
            Ok(result) => {
                let mut guard = self.tx.lock().await;
                if let Some(pool_tx) = guard.as_mut() {
                    let sql = format!("RELEASE SAVEPOINT {}", savepoint_name);
                    pool_tx.execute(&sql, &[]).await?;
                }
                Ok(result)
            }
            Err(e) => {
                let mut guard = self.tx.lock().await;
                if let Some(pool_tx) = guard.as_mut() {
                    let sql = format!("ROLLBACK TO SAVEPOINT {}", savepoint_name);
                    let _ = pool_tx.execute(&sql, &[]).await;
                }
                Err(e)
            }
        }
    }

    pub async fn find<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find(self, conditions, options, Some(self.handler.limit_config())).await
    }

    pub async fn find_one<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Option<M>> {
        find_one(self, conditions, options).await
    }

    pub async fn find_by_id<M: Model>(
        &self,
        pk_values: &[Vec<Value>],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find_by_id(self, pk_values, options).await
    }

    pub async fn count<M: Model>(&self, conditions: &[Condition]) -> Result<i64> {
        count::<M, _>(self, conditions).await
    }

    pub async fn query<M: Model>(&self, sql: &str, params: &[Value]) -> Result<Vec<M>> {
        query::<M, _>(self, sql, params).await
    }

    pub async fn create<M: Model>(
        &self,
        values: &[ColumnValue],
        options: Option<&InsertOptions>,
    ) -> Result<QueryResult> {
        create::<M, _>(self, values, options).await
    }

    pub async fn create_many<M: Model>(
        &self,
        rows: &[Vec<ColumnValue>],
        options: Option<&InsertOptions>,
    ) -> Result<Option<PkeyResult>> {
        create_many::<M, _>(self, rows, options).await
    }

    pub async fn update<M: Model>(
        &self,
        conditions: &[Condition],
        values: &[ColumnValue],
        options: Option<&UpdateOptions>,
    ) -> Result<QueryResult> {
        update::<M, _>(self, conditions, values, options).await
    }

    pub async fn update_many<M: Model>(
        &self,
        rows: &[Vec<ColumnValue>],
        options: &UpdateManyOptions,
    ) -> Result<Option<PkeyResult>> {
        update_many::<M, _>(self, rows, options).await
    }

    pub async fn delete<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&DeleteOptions>,
    ) -> Result<QueryResult> {
        delete::<M, _>(self, conditions, options).await
    }
}

impl Executor for TransactionExecutor {
    fn execute_query<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_query_with_op(sql, params, OperationType::Other, "")
    }

    fn execute_query_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        op: OperationType,
        table: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        Box::pin(async move {
            let core = NextExecute::new(|s: &'a str, p: &'a [Value]| {
                let tx = Arc::clone(&self.tx);
                Box::pin(async move {
                    let mut guard = tx.lock().await;
                    let pool_tx = guard.as_mut().ok_or_else(|| {
                        Error::Transaction("Transaction already completed".to_string())
                    })?;
                    pool_tx.execute(s, p).await
                }) as Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>>
            });
            execute_with_middleware_op(
                &self.handler.inner.middleware_chain,
                sql,
                params,
                core,
                &self.ctx,
                op,
                table,
            )
            .await
        })
    }

    fn execute_write<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_write_with_op(sql, params, OperationType::Other, "")
    }

    fn execute_write_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        op: OperationType,
        table: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_query_with_op(sql, params, op, table)
    }

    fn driver_type(&self) -> DriverType {
        self.handler.driver_type()
    }

    fn is_transaction(&self) -> bool {
        true
    }

    #[cfg(feature = "postgres")]
    fn execute_pg_rows<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Option<Pin<Box<dyn Future<Output = Result<Vec<tokio_postgres::Row>>> + Send + 'a>>> {
        if self.handler.driver_type() != DriverType::Postgres {
            return None;
        }
        if !self.handler.inner.middleware_chain.is_empty() {
            return None;
        }
        Some(Box::pin(async move {
            let mut guard = self.tx.lock().await;
            let pool_tx = guard
                .as_mut()
                .ok_or_else(|| Error::Transaction("Transaction already completed".to_string()))?;
            match pool_tx {
                #[cfg(feature = "postgres")]
                PoolTransaction::Postgres(conn) => run_pg_rows(conn, sql, params).await,
                _ => Err(Error::Driver("Not PostgreSQL".to_string())),
            }
        }))
    }

    fn clone_arc(&self) -> Arc<dyn Executor> {
        Arc::new(self.clone())
    }
}

// ============================================
// WriterExecutor -- routes reads to writer pool
// ============================================

/// Executor that routes reads to the writer pool.
/// Write operations are rejected (read-only context).
#[derive(Clone)]
pub struct WriterExecutor {
    handler: DbHandler,
    ctx: Arc<RequestContext>,
}

impl WriterExecutor {
    pub fn handler(&self) -> &DbHandler {
        &self.handler
    }

    /// Access the per-request context (type-map), inherited from the parent scope.
    pub fn ctx(&self) -> &RequestContext {
        &self.ctx
    }

    pub async fn find<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find(self, conditions, options, Some(self.handler.limit_config())).await
    }

    pub async fn find_one<M: Model>(
        &self,
        conditions: &[Condition],
        options: Option<&SelectOptions>,
    ) -> Result<Option<M>> {
        find_one(self, conditions, options).await
    }

    pub async fn find_by_id<M: Model>(
        &self,
        pk_values: &[Vec<Value>],
        options: Option<&SelectOptions>,
    ) -> Result<Vec<M>> {
        find_by_id(self, pk_values, options).await
    }

    pub async fn count<M: Model>(&self, conditions: &[Condition]) -> Result<i64> {
        count::<M, _>(self, conditions).await
    }

    pub async fn query<M: Model>(&self, sql: &str, params: &[Value]) -> Result<Vec<M>> {
        query::<M, _>(self, sql, params).await
    }
}

impl Executor for WriterExecutor {
    fn execute_query<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_query_with_op(sql, params, OperationType::Other, "")
    }

    fn execute_query_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        op: OperationType,
        table: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        Box::pin(async move {
            let pool = self.handler.writer_pool().unwrap_or(self.handler.pool());
            let core = NextExecute::new(move |s: &'a str, p: &'a [Value]| {
                Box::pin(execute_on_pool(pool, s, p))
                    as Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>>
            });
            execute_with_middleware_op(
                &self.handler.inner.middleware_chain,
                sql,
                params,
                core,
                &self.ctx,
                op,
                table,
            )
            .await
        })
    }

    fn execute_write<'a>(
        &'a self,
        sql: &'a str,
        _params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryResult>> + Send + 'a>> {
        Box::pin(async move {
            Err(Error::WriteInReadOnlyContext {
                operation: sql.split_whitespace().next().unwrap_or("WRITE").to_string(),
                model: "unknown".to_string(),
            })
        })
    }

    fn driver_type(&self) -> DriverType {
        self.handler.driver_type()
    }

    fn is_read_only(&self) -> bool {
        true
    }

    #[cfg(feature = "postgres")]
    fn execute_pg_rows<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Option<Pin<Box<dyn Future<Output = Result<Vec<tokio_postgres::Row>>> + Send + 'a>>> {
        if !self.handler.inner.middleware_chain.is_empty() {
            return None;
        }
        let pool = self.handler.writer_pool().unwrap_or(self.handler.pool());
        match pool {
            Pool::Postgres(pg_pool) => Some(Box::pin(async move {
                let conn = pg_pool
                    .get()
                    .await
                    .map_err(|e| Error::PoolError(e.to_string()))?;
                run_pg_rows(&conn, sql, params).await
            })),
            _ => None,
        }
    }

    fn clone_arc(&self) -> Arc<dyn Executor> {
        Arc::new(self.clone())
    }
}

// ============================================
// Pool execution dispatch
// ============================================

async fn execute_on_pool(pool: &Pool, sql: &str, params: &[Value]) -> Result<QueryResult> {
    match pool {
        Pool::None => Err(Error::Driver("No database driver configured".to_string())),
        #[cfg(feature = "postgres")]
        Pool::Postgres(pg_pool) => {
            let conn = pg_pool
                .get()
                .await
                .map_err(|e| Error::PoolError(e.to_string()))?;
            run_pg_query(&conn, sql, params).await
        }
        #[cfg(feature = "mysql")]
        Pool::Mysql(mysql_pool) => run_mysql_query(mysql_pool, sql, params).await,
        #[cfg(feature = "sqlite")]
        Pool::Sqlite(sqlite_pool) => run_sqlite_query(sqlite_pool, sql, params).await,
    }
}

// ============================================
// PostgreSQL execution (tokio-postgres + deadpool)
// ============================================

#[cfg(feature = "postgres")]
async fn run_pg_query(
    client: &deadpool_postgres::ClientWrapper,
    sql: &str,
    params: &[Value],
) -> Result<QueryResult> {
    use crate::driver::postgres::PostgresSqlBuilder;
    use crate::driver::SqlBuilder;

    let builder = PostgresSqlBuilder;
    let rewritten = builder.rewrite_placeholders(sql);

    let pg_params: Vec<PgParam> = params.iter().map(value_to_pg_param).collect();
    let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
        .iter()
        .map(|p| p as &(dyn tokio_postgres::types::ToSql + Sync))
        .collect();

    let stmt = client
        .prepare_cached(&rewritten)
        .await
        .map_err(Error::Postgres)?;
    let rows = client
        .query(&stmt, &param_refs)
        .await
        .map_err(Error::Postgres)?;

    let result_rows: Vec<crate::driver::Row> = rows.iter().map(pg_row_to_map).collect();

    Ok(QueryResult {
        rows_affected: result_rows.len() as u64,
        rows: result_rows,
    })
}

/// Execute a PG query and return raw `tokio_postgres::Row` values without converting to HashMap.
/// This is the fast path for direct deserialization via `FromPgRow`/`PgGet`.
#[cfg(feature = "postgres")]
async fn run_pg_rows(
    client: &deadpool_postgres::ClientWrapper,
    sql: &str,
    params: &[Value],
) -> Result<Vec<tokio_postgres::Row>> {
    use crate::driver::postgres::PostgresSqlBuilder;
    use crate::driver::SqlBuilder;

    let builder = PostgresSqlBuilder;
    let rewritten = builder.rewrite_placeholders(sql);

    let pg_params: Vec<PgParam> = params.iter().map(value_to_pg_param).collect();
    let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
        .iter()
        .map(|p| p as &(dyn tokio_postgres::types::ToSql + Sync))
        .collect();

    let stmt = client
        .prepare_cached(&rewritten)
        .await
        .map_err(Error::Postgres)?;
    client
        .query(&stmt, &param_refs)
        .await
        .map_err(Error::Postgres)
}

/// Stack-allocated parameter enum for PostgreSQL.
/// Eliminates per-parameter Box heap allocations by consolidating all
/// Value-to-ToSql conversion into a single enum that lives on the stack.
#[cfg(feature = "postgres")]
#[derive(Debug)]
enum PgParam {
    Null,
    Bool(bool),
    Int(i64),
    Float(f64),
    Text(String),
    Bytes(Vec<u8>),
    Json(serde_json::Value),
    Uuid(uuid::Uuid),
    Timestamp(chrono::NaiveDateTime),
    Date(chrono::NaiveDate),
    IntArray(Vec<i64>),
    FloatArray(Vec<f64>),
    TextArray(Vec<String>),
    BoolArray(Vec<bool>),
}

#[cfg(feature = "postgres")]
impl postgres_types::ToSql for PgParam {
    fn to_sql(
        &self,
        ty: &postgres_types::Type,
        out: &mut bytes::BytesMut,
    ) -> std::result::Result<postgres_types::IsNull, Box<dyn std::error::Error + Sync + Send>> {
        match self {
            PgParam::Null => Ok(postgres_types::IsNull::Yes),
            PgParam::Bool(b) => b.to_sql(ty, out),
            PgParam::Int(i) => match *ty {
                postgres_types::Type::INT2 => (*i as i16).to_sql(ty, out),
                postgres_types::Type::INT4 => (*i as i32).to_sql(ty, out),
                _ => i.to_sql(ty, out),
            },
            PgParam::Float(f) => match *ty {
                postgres_types::Type::FLOAT4 => (*f as f32).to_sql(ty, out),
                _ => f.to_sql(ty, out),
            },
            PgParam::Text(s) => s.to_sql(ty, out),
            PgParam::Bytes(b) => b.as_slice().to_sql(ty, out),
            PgParam::Json(j) => j.to_sql(ty, out),
            PgParam::Uuid(u) => u.to_sql(ty, out),
            PgParam::Timestamp(t) => t.to_sql(ty, out),
            PgParam::Date(d) => d.to_sql(ty, out),
            PgParam::IntArray(a) => a.to_sql(ty, out),
            PgParam::FloatArray(a) => a.to_sql(ty, out),
            PgParam::TextArray(a) => a.to_sql(ty, out),
            PgParam::BoolArray(a) => a.to_sql(ty, out),
        }
    }

    fn accepts(_ty: &postgres_types::Type) -> bool {
        true
    }

    postgres_types::to_sql_checked!();
}

#[cfg(feature = "postgres")]
fn value_to_pg_param(v: &Value) -> PgParam {
    match v {
        Value::Null => PgParam::Null,
        Value::Bool(b) => PgParam::Bool(*b),
        Value::Int(i) => PgParam::Int(*i),
        Value::Float(f) => PgParam::Float(*f),
        Value::Text(s) => PgParam::Text(s.clone()),
        Value::Bytes(b) => PgParam::Bytes(b.clone()),
        Value::Json(j) => PgParam::Json(j.clone()),
        Value::Uuid(u) => PgParam::Uuid(*u),
        Value::Timestamp(t) => PgParam::Timestamp(*t),
        Value::Date(d) => PgParam::Date(*d),
        Value::IntArray(a) => PgParam::IntArray(a.clone()),
        Value::FloatArray(a) => PgParam::FloatArray(a.clone()),
        Value::TextArray(a) => PgParam::TextArray(a.clone()),
        Value::BoolArray(a) => PgParam::BoolArray(a.clone()),
        Value::Cast { value: inner, .. } => value_to_pg_param(inner),
        Value::Raw(_) | Value::Immediate(_) | Value::NotNull => PgParam::Null,
        Value::ValueArray(_) | Value::Dynamic { .. } | Value::CastArray { .. } => PgParam::Null,
    }
}

#[cfg(feature = "postgres")]
fn pg_row_to_map(row: &tokio_postgres::Row) -> crate::driver::Row {
    let mut map = crate::driver::Row::new();
    for (idx, col) in row.columns().iter().enumerate() {
        let name = col.name().to_string();
        let type_name = col.type_().name();
        let value = match type_name {
            "bool" => row.try_get::<_, bool>(idx).ok().map(Value::Bool),
            "int2" | "int4" => row
                .try_get::<_, i32>(idx)
                .ok()
                .map(|v| Value::Int(v as i64)),
            "int8" => row.try_get::<_, i64>(idx).ok().map(Value::Int),
            "float4" => row
                .try_get::<_, f32>(idx)
                .ok()
                .map(|v| Value::Float(v as f64)),
            "float8" => row.try_get::<_, f64>(idx).ok().map(Value::Float),
            "text" | "varchar" | "bpchar" | "name" => {
                row.try_get::<_, String>(idx).ok().map(Value::Text)
            }
            "bytea" => row.try_get::<_, Vec<u8>>(idx).ok().map(Value::Bytes),
            "uuid" => row.try_get::<_, uuid::Uuid>(idx).ok().map(Value::Uuid),
            "timestamp" => row
                .try_get::<_, chrono::NaiveDateTime>(idx)
                .ok()
                .map(Value::Timestamp),
            "timestamptz" => row
                .try_get::<_, chrono::DateTime<chrono::Utc>>(idx)
                .ok()
                .map(|dt| Value::Timestamp(dt.naive_utc())),
            "date" => row
                .try_get::<_, chrono::NaiveDate>(idx)
                .ok()
                .map(Value::Date),
            "json" | "jsonb" => row
                .try_get::<_, serde_json::Value>(idx)
                .ok()
                .map(Value::Json),
            _ => row.try_get::<_, String>(idx).ok().map(Value::Text),
        };
        map.insert(name, value.unwrap_or(Value::Null));
    }
    map
}

// ============================================
// MySQL execution
// ============================================

#[cfg(feature = "mysql")]
async fn run_mysql_query<'c>(
    executor: impl sqlx::Executor<'c, Database = sqlx::MySql>,
    sql: &str,
    params: &[Value],
) -> Result<QueryResult> {
    let mut query = sqlx::query(sql);
    for param in params {
        query = bind_mysql_value(query, param);
    }

    let rows = query.fetch_all(executor).await.map_err(Error::Database)?;

    let result_rows: Vec<crate::driver::Row> =
        rows.iter().map(|row| mysql_row_to_map(row)).collect();

    Ok(QueryResult {
        rows_affected: result_rows.len() as u64,
        rows: result_rows,
    })
}

#[cfg(feature = "mysql")]
fn bind_mysql_value<'q>(
    query: sqlx::query::Query<'q, sqlx::MySql, sqlx::mysql::MySqlArguments>,
    value: &'q Value,
) -> sqlx::query::Query<'q, sqlx::MySql, sqlx::mysql::MySqlArguments> {
    match value {
        Value::Null => query.bind(None::<String>),
        Value::Bool(v) => query.bind(v),
        Value::Int(v) => query.bind(v),
        Value::Float(v) => query.bind(v),
        Value::Text(v) => query.bind(v.as_str()),
        Value::Bytes(v) => query.bind(v.as_slice()),
        Value::Json(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::Uuid(v) => query.bind(v.to_string()),
        Value::Timestamp(v) => query.bind(v),
        Value::Date(v) => query.bind(v),
        Value::IntArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::FloatArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::TextArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::BoolArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::Raw(_) | Value::Immediate(_) | Value::NotNull => query,
        Value::ValueArray(_) | Value::Dynamic { .. } | Value::CastArray { .. } => query,
        Value::Cast { value: inner, .. } => bind_mysql_value(query, inner),
    }
}

#[cfg(feature = "mysql")]
fn mysql_row_to_map(row: &sqlx::mysql::MySqlRow) -> crate::driver::Row {
    use sqlx::Column as SqlxColumn;
    use sqlx::Row as SqlxRow;
    use sqlx::TypeInfo;

    let mut map = crate::driver::Row::new();
    for col in row.columns() {
        let name = col.name().to_string();
        let type_name = col.type_info().name();
        let value = match type_name {
            "BOOLEAN" | "TINYINT(1)" => row.try_get::<bool, _>(col.ordinal()).ok().map(Value::Bool),
            "TINYINT" | "SMALLINT" | "INT" | "MEDIUMINT" => row
                .try_get::<i32, _>(col.ordinal())
                .ok()
                .map(|v| Value::Int(v as i64)),
            "BIGINT" => row.try_get::<i64, _>(col.ordinal()).ok().map(Value::Int),
            "FLOAT" => row
                .try_get::<f32, _>(col.ordinal())
                .ok()
                .map(|v| Value::Float(v as f64)),
            "DOUBLE" | "DECIMAL" => row.try_get::<f64, _>(col.ordinal()).ok().map(Value::Float),
            "VARCHAR" | "CHAR" | "TEXT" | "TINYTEXT" | "MEDIUMTEXT" | "LONGTEXT" => row
                .try_get::<String, _>(col.ordinal())
                .ok()
                .map(Value::Text),
            "BLOB" | "TINYBLOB" | "MEDIUMBLOB" | "LONGBLOB" | "BINARY" | "VARBINARY" => row
                .try_get::<Vec<u8>, _>(col.ordinal())
                .ok()
                .map(Value::Bytes),
            "DATETIME" | "TIMESTAMP" => row
                .try_get::<chrono::NaiveDateTime, _>(col.ordinal())
                .ok()
                .map(Value::Timestamp),
            "DATE" => row
                .try_get::<chrono::NaiveDate, _>(col.ordinal())
                .ok()
                .map(Value::Date),
            "JSON" => row
                .try_get::<serde_json::Value, _>(col.ordinal())
                .ok()
                .map(Value::Json),
            _ => row
                .try_get::<String, _>(col.ordinal())
                .ok()
                .map(Value::Text),
        };
        map.insert(name, value.unwrap_or(Value::Null));
    }
    map
}

// ============================================
// SQLite execution
// ============================================

#[cfg(feature = "sqlite")]
async fn run_sqlite_query<'c>(
    executor: impl sqlx::Executor<'c, Database = sqlx::Sqlite>,
    sql: &str,
    params: &[Value],
) -> Result<QueryResult> {
    let mut query = sqlx::query(sql);
    for param in params {
        query = bind_sqlite_value(query, param);
    }

    let rows = query.fetch_all(executor).await.map_err(Error::Database)?;

    let result_rows: Vec<crate::driver::Row> =
        rows.iter().map(|row| sqlite_row_to_map(row)).collect();

    Ok(QueryResult {
        rows_affected: result_rows.len() as u64,
        rows: result_rows,
    })
}

#[cfg(feature = "sqlite")]
fn bind_sqlite_value<'q>(
    query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
    value: &'q Value,
) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {
    match value {
        Value::Null => query.bind(None::<String>),
        Value::Bool(v) => query.bind(v),
        Value::Int(v) => query.bind(v),
        Value::Float(v) => query.bind(v),
        Value::Text(v) => query.bind(v.as_str()),
        Value::Bytes(v) => query.bind(v.as_slice()),
        Value::Json(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::Uuid(v) => query.bind(v.to_string()),
        Value::Timestamp(v) => query.bind(v.and_utc().timestamp()),
        Value::Date(v) => query.bind(v.to_string()),
        Value::IntArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::FloatArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::TextArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::BoolArray(v) => query.bind(serde_json::to_string(v).unwrap_or_default()),
        Value::Raw(_) | Value::Immediate(_) | Value::NotNull => query,
        Value::ValueArray(_) | Value::Dynamic { .. } | Value::CastArray { .. } => query,
        Value::Cast { value: inner, .. } => bind_sqlite_value(query, inner),
    }
}

#[cfg(feature = "sqlite")]
fn sqlite_row_to_map(row: &sqlx::sqlite::SqliteRow) -> crate::driver::Row {
    use sqlx::Column as SqlxColumn;
    use sqlx::Row as SqlxRow;
    use sqlx::TypeInfo;

    let mut map = crate::driver::Row::new();
    for col in row.columns() {
        let name = col.name().to_string();
        let type_name = col.type_info().name();
        let value = match type_name {
            "BOOLEAN" => row.try_get::<bool, _>(col.ordinal()).ok().map(Value::Bool),
            "INTEGER" | "INT" => row.try_get::<i64, _>(col.ordinal()).ok().map(Value::Int),
            "REAL" => row.try_get::<f64, _>(col.ordinal()).ok().map(Value::Float),
            "TEXT" => row
                .try_get::<String, _>(col.ordinal())
                .ok()
                .map(Value::Text),
            "BLOB" => row
                .try_get::<Vec<u8>, _>(col.ordinal())
                .ok()
                .map(Value::Bytes),
            _ => row
                .try_get::<i64, _>(col.ordinal())
                .ok()
                .map(Value::Int)
                .or_else(|| {
                    row.try_get::<String, _>(col.ordinal())
                        .ok()
                        .map(Value::Text)
                }),
        };
        map.insert(name, value.unwrap_or(Value::Null));
    }
    map
}

// ============================================
// Builder
// ============================================

pub struct DbHandlerBuilder {
    config: DbConfig,
    options: DbConfigOptions,
    middleware_chain: MiddlewareChain,
}

impl DbHandlerBuilder {
    pub fn new(config: DbConfig) -> Self {
        Self {
            config,
            options: DbConfigOptions::default(),
            middleware_chain: MiddlewareChain::new(),
        }
    }

    /// Get the current config (for introspection).
    pub fn config(&self) -> &DbConfig {
        &self.config
    }

    pub fn with_writer(mut self, config: DbConfig) -> Self {
        self.options.writer_config = Some(config);
        self
    }

    pub fn with_writer_sticky(mut self, duration_ms: u64) -> Self {
        self.options.writer_sticky_duration_ms = duration_ms;
        self
    }

    pub fn with_find_hard_limit(mut self, limit: usize) -> Self {
        self.options.find_hard_limit = Some(limit);
        self
    }

    /// Alias for `with_find_hard_limit` for API consistency.
    pub fn find_hard_limit(mut self, limit: usize) -> Self {
        self.options.find_hard_limit = Some(limit);
        self
    }

    /// Set maximum number of connections in the pool.
    pub fn max_connections(mut self, n: u32) -> Self {
        self.config.max_connections = Some(n);
        self
    }

    /// Set minimum number of connections to maintain in the pool.
    pub fn min_connections(mut self, n: u32) -> Self {
        self.config.min_connections = Some(n);
        self
    }

    /// Set maximum idle duration for connections in the pool.
    pub fn idle_timeout(mut self, duration: Duration) -> Self {
        self.config.idle_timeout = Some(duration);
        self
    }

    /// Set maximum lifetime of connections in the pool.
    pub fn max_lifetime(mut self, duration: Duration) -> Self {
        self.config.max_lifetime = Some(duration);
        self
    }

    pub fn with_has_many_hard_limit(mut self, limit: usize) -> Self {
        self.options.has_many_hard_limit = Some(limit);
        self
    }

    pub fn with_middleware(mut self, middleware: impl crate::middleware::Middleware) -> Self {
        self.middleware_chain.add(middleware);
        self
    }

    /// Build the handler. Creates connection pools for the configured driver.
    pub async fn build(self) -> Result<DbHandler> {
        let pool = create_pool(&self.config).await?;
        let writer_pool = match &self.options.writer_config {
            Some(wc) => Some(create_pool(wc).await?),
            None => None,
        };

        Ok(DbHandler {
            inner: Arc::new(DbHandlerInner {
                limit_config: LimitConfig {
                    find_hard_limit: self.options.find_hard_limit,
                    has_many_hard_limit: self.options.has_many_hard_limit,
                },
                config: self.config,
                pool,
                writer_pool,
                middleware_chain: self.middleware_chain,
                default_ctx: RequestContext::new(),
            }),
        })
    }
}

/// Create a connection pool for the given config.
async fn create_pool(config: &DbConfig) -> Result<Pool> {
    let conn_str = config.connection_string();
    let max = config.max_connections.unwrap_or(10);

    match config.driver {
        #[cfg(feature = "postgres")]
        DriverType::Postgres => {
            let pg_config: tokio_postgres::Config = conn_str
                .parse()
                .map_err(|e: tokio_postgres::Error| Error::Config(e.to_string()))?;
            let mgr = deadpool_postgres::Manager::new(pg_config, tokio_postgres::NoTls);
            let pool = deadpool_postgres::Pool::builder(mgr)
                .max_size(max as usize)
                .runtime(deadpool_postgres::Runtime::Tokio1)
                .build()
                .map_err(|e| Error::Config(format!("Failed to create pool: {}", e)))?;
            Ok(Pool::Postgres(pool))
        }
        #[cfg(feature = "mysql")]
        DriverType::Mysql => {
            let mut opts = sqlx::mysql::MySqlPoolOptions::new().max_connections(max);
            if let Some(min) = config.min_connections {
                opts = opts.min_connections(min);
            }
            if let Some(ref timeout) = config.idle_timeout {
                opts = opts.idle_timeout(Some(timeout.clone()));
            }
            if let Some(ref lifetime) = config.max_lifetime {
                opts = opts.max_lifetime(Some(lifetime.clone()));
            }
            let pool = opts.connect(&conn_str).await.map_err(Error::Database)?;
            Ok(Pool::Mysql(pool))
        }
        #[cfg(feature = "sqlite")]
        DriverType::Sqlite => {
            let mut opts = sqlx::sqlite::SqlitePoolOptions::new().max_connections(max);
            if let Some(min) = config.min_connections {
                opts = opts.min_connections(min);
            }
            if let Some(ref timeout) = config.idle_timeout {
                opts = opts.idle_timeout(Some(timeout.clone()));
            }
            if let Some(ref lifetime) = config.max_lifetime {
                opts = opts.max_lifetime(Some(lifetime.clone()));
            }
            let pool = opts.connect(&conn_str).await.map_err(Error::Database)?;
            Ok(Pool::Sqlite(pool))
        }
        #[allow(unreachable_patterns)]
        _ => Err(Error::Config(format!(
            "Driver {:?} is not enabled. Enable the corresponding feature flag.",
            config.driver
        ))),
    }
}

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

    #[test]
    fn test_builder_pool_config_stores_values() {
        let config = DbConfig::sqlite(":memory:");
        let builder = DbHandlerBuilder::new(config)
            .max_connections(20)
            .min_connections(2)
            .idle_timeout(Duration::from_secs(300))
            .max_lifetime(Duration::from_secs(3600))
            .find_hard_limit(1000);

        let cfg = builder.config();
        assert_eq!(cfg.max_connections, Some(20));
        assert_eq!(cfg.min_connections, Some(2));
        assert_eq!(cfg.idle_timeout, Some(Duration::from_secs(300)));
        assert_eq!(cfg.max_lifetime, Some(Duration::from_secs(3600)));
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_builder_builds_with_pool_config() {
        use crate::types::DriverType;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(5)
            .min_connections(0)
            .find_hard_limit(500)
            .build()
            .await
            .unwrap();
        assert_eq!(handler.config().driver, DriverType::Sqlite);
        assert_eq!(handler.limit_config().find_hard_limit, Some(500));
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_simple_transaction_sqlite() {
        use crate::values::Value;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        let count: i64 = handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE simple_test (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                tx.execute_write("INSERT INTO simple_test (id, x) VALUES (1, 10)", &[])
                    .await?;
                let r = tx
                    .execute_query("SELECT COUNT(*) AS count FROM simple_test", &[])
                    .await?;
                let c = r
                    .rows
                    .first()
                    .and_then(|row| row.get("count").or_else(|| row.get("COUNT(*)")))
                    .and_then(|v| {
                        if let Value::Int(n) = v {
                            Some(*n)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                Ok(c)
            })
            .await
            .unwrap();
        assert_eq!(count, 1, "simple transaction should see 1 row");
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_nested_transaction_commit() {
        use crate::values::Value;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        let count: i64 = handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE nest_test (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                tx.execute_write("INSERT INTO nest_test (id, x) VALUES (1, 10)", &[])
                    .await?;
                tx.transaction(|tx2| async move {
                    tx2.execute_write("INSERT INTO nest_test (id, x) VALUES (2, 20)", &[])
                        .await?;
                    Ok(())
                })
                .await?;
                let r = tx
                    .execute_query("SELECT COUNT(*) AS count FROM nest_test", &[])
                    .await?;
                let c = r
                    .rows
                    .first()
                    .and_then(|row| row.get("count").or_else(|| row.get("COUNT(*)")))
                    .and_then(|v| {
                        if let Value::Int(n) = v {
                            Some(*n)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                Ok(c)
            })
            .await
            .unwrap();
        assert_eq!(count, 2);
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_nested_transaction_rollback_outer_commits() {
        use crate::values::Value;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        let count: i64 = handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE nest_rollback (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                tx.execute_write("INSERT INTO nest_rollback (id, x) VALUES (1, 10)", &[])
                    .await?;
                let _: Result<()> = tx
                    .transaction(|tx2| async move {
                        tx2.execute_write("INSERT INTO nest_rollback (id, x) VALUES (2, 20)", &[])
                            .await?;
                        Err(crate::error::Error::Transaction(
                            "intentional rollback".to_string(),
                        ))
                    })
                    .await;
                let r = tx
                    .execute_query("SELECT COUNT(*) AS count FROM nest_rollback", &[])
                    .await?;
                let c = r
                    .rows
                    .first()
                    .and_then(|row| row.get("count").or_else(|| row.get("COUNT(*)")))
                    .and_then(|v| {
                        if let Value::Int(n) = v {
                            Some(*n)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                Ok(c)
            })
            .await
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_double_nested_transaction() {
        use crate::values::Value;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        let count: i64 = handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE double_nest (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                tx.execute_write("INSERT INTO double_nest (id, x) VALUES (1, 10)", &[])
                    .await?;
                tx.transaction(|tx2| async move {
                    tx2.execute_write("INSERT INTO double_nest (id, x) VALUES (2, 20)", &[])
                        .await?;
                    tx2.transaction(|tx3| async move {
                        tx3.execute_write("INSERT INTO double_nest (id, x) VALUES (3, 30)", &[])
                            .await?;
                        Ok(())
                    })
                    .await?;
                    Ok(())
                })
                .await?;
                let r = tx
                    .execute_query("SELECT COUNT(*) AS count FROM double_nest", &[])
                    .await?;
                let c = r
                    .rows
                    .first()
                    .and_then(|row| row.get("count").or_else(|| row.get("COUNT(*)")))
                    .and_then(|v| {
                        if let Value::Int(n) = v {
                            Some(*n)
                        } else {
                            None
                        }
                    })
                    .unwrap_or(0);
                Ok(c)
            })
            .await
            .unwrap();
        assert_eq!(count, 3);
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_transaction_commit() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE tx_commit_test (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                tx.execute_write("INSERT INTO tx_commit_test (id, x) VALUES (1, 100)", &[])
                    .await?;
                Ok(())
            })
            .await
            .unwrap();

        let result = handler
            .execute_query("SELECT * FROM tx_commit_test WHERE id = 1", &[])
            .await
            .unwrap();
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0].get("x"), Some(&Value::Int(100)));
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_transaction_rollback() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE tx_rollback_test (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                Ok(())
            })
            .await
            .unwrap();

        let _: Result<()> = handler
            .transaction(|tx| async move {
                tx.execute_write("INSERT INTO tx_rollback_test (id, x) VALUES (1, 200)", &[])
                    .await?;
                Err(crate::error::Error::Transaction(
                    "intentional rollback".to_string(),
                ))
            })
            .await;

        let result = handler
            .execute_query("SELECT * FROM tx_rollback_test", &[])
            .await
            .unwrap();
        assert!(
            result.rows.is_empty(),
            "row should not exist after rollback"
        );
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_transaction_with_options_retry() {
        use crate::types::TransactionOptions;
        use std::sync::Mutex;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        let attempts = std::sync::Arc::new(Mutex::new(0u32));
        let result = handler
            .transaction_with_options(
                |tx| {
                    let attempts = std::sync::Arc::clone(&attempts);
                    async move {
                        let mut a = attempts.lock().unwrap();
                        *a += 1;
                        if *a < 2 {
                            Err(crate::error::Error::Transaction("fail first".to_string()))
                        } else {
                            tx.execute_write(
                                "CREATE TABLE retry_test (id INTEGER PRIMARY KEY)",
                                &[],
                            )
                            .await?;
                            tx.execute_write("INSERT INTO retry_test (id) VALUES (1)", &[])
                                .await?;
                            Ok(42i64)
                        }
                    }
                },
                TransactionOptions {
                    retry_on_error: true,
                    retry_limit: 2,
                    retry_duration_ms: 1,
                    ..Default::default()
                },
            )
            .await
            .unwrap();
        assert_eq!(result, 42);
        assert_eq!(*attempts.lock().unwrap(), 2);
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_rollback_only() {
        use crate::types::TransactionOptions;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE rollback_only_test (id INTEGER PRIMARY KEY, x INTEGER)",
                    &[],
                )
                .await?;
                Ok(())
            })
            .await
            .unwrap();

        let result: Result<i64> = handler
            .transaction_with_options(
                |tx| async move {
                    tx.execute_write(
                        "INSERT INTO rollback_only_test (id, x) VALUES (1, 999)",
                        &[],
                    )
                    .await?;
                    Ok(1i64)
                },
                TransactionOptions {
                    rollback_only: true,
                    ..Default::default()
                },
            )
            .await;
        assert!(result.is_ok());

        let rows = handler
            .execute_query("SELECT * FROM rollback_only_test", &[])
            .await
            .unwrap();
        assert!(
            rows.rows.is_empty(),
            "row should not exist due to forced rollback"
        );
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_writer_executor_properties() {
        use crate::model::Executor;
        use crate::types::DriverType;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .with_writer(|exec| async move {
                assert!(exec.is_read_only(), "WriterExecutor should be read-only");
                assert!(
                    !exec.is_transaction(),
                    "WriterExecutor should not be in transaction"
                );
                assert_eq!(exec.driver_type(), DriverType::Sqlite);
                Ok(())
            })
            .await
            .unwrap();
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_transaction_executor_properties() {
        use crate::model::Executor;
        use crate::types::DriverType;

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .transaction(|tx| async move {
                assert!(
                    tx.is_transaction(),
                    "TransactionExecutor should be in transaction"
                );
                assert_eq!(tx.driver_type(), DriverType::Sqlite);
                Ok(())
            })
            .await
            .unwrap();
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_db_scope_find_and_transaction() {
        use crate::condition::Condition;
        use crate::driver::Row;
        use crate::model::{ColumnMeta, FromRow, Model};

        #[derive(Debug)]
        struct ScopeTestUser {
            pub id: i64,
            pub name: String,
        }
        impl FromRow for ScopeTestUser {
            fn from_row(row: &Row) -> Result<Self> {
                use crate::values::Value;
                let id = match row.get("id") {
                    Some(Value::Int(n)) => *n,
                    _ => return Err(Error::ColumnNotFound("id".to_string())),
                };
                let name = match row.get("name") {
                    Some(Value::Text(s)) => s.clone(),
                    _ => return Err(Error::ColumnNotFound("name".to_string())),
                };
                Ok(ScopeTestUser { id, name })
            }
        }
        impl Model for ScopeTestUser {
            type Columns = ();
            const TABLE_NAME: &'static str = "scope_test_users";
            fn column_meta() -> &'static [ColumnMeta] {
                &[]
            }
            fn primary_key_columns() -> &'static [&'static str] {
                &["id"]
            }
            fn columns() -> Self::Columns {
                ()
            }
        }

        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        handler
            .transaction(|tx| async move {
                tx.execute_write(
                    "CREATE TABLE scope_test_users (id INTEGER PRIMARY KEY, name TEXT)",
                    &[],
                )
                .await?;
                tx.execute_write(
                    "INSERT INTO scope_test_users (id, name) VALUES (1, 'Alice'), (2, 'Bob')",
                    &[],
                )
                .await?;
                Ok(())
            })
            .await
            .unwrap();

        let scope = DbScope::new(handler.clone());
        let conds: Vec<Condition> = vec![];
        let users: Vec<ScopeTestUser> = scope.find(&conds, None).await.unwrap();
        assert_eq!(users.len(), 2);
        assert_eq!(users[0].name, "Alice");
        assert_eq!(users[1].name, "Bob");

        let one: Option<ScopeTestUser> = scope.find_one(&conds, None).await.unwrap();
        assert!(one.is_some());
        assert_eq!(one.unwrap().name, "Alice");

        let n = scope.count::<ScopeTestUser>(&[]).await.unwrap();
        assert_eq!(n, 2);

        scope
            .transaction(|tx| async move {
                let col = crate::column::Column::<String>::new("scope_test_users", "name");
                let cvs = vec![col.set("Charlie".to_string())];
                let conds = vec![Condition::Eq {
                    column: "id".into(),
                    table: None,
                    value: Value::Int(1),
                }];
                update::<ScopeTestUser, _>(&tx, &conds, &cvs, None).await?;
                Ok(())
            })
            .await
            .unwrap();

        let users: Vec<ScopeTestUser> = scope.find(&conds, None).await.unwrap();
        assert_eq!(users[0].name, "Charlie");
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_db_handler_close() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();
        handler.close().await;
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_dbscope_ctx_available() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        #[derive(Clone, Debug, PartialEq)]
        struct TenantId(String);

        let scope = handler.scope();
        scope.ctx().set(TenantId("abc".into()));
        let got = scope.ctx().get::<TenantId>();
        assert_eq!(got, Some(TenantId("abc".into())));
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_transaction_inherits_ctx() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        #[derive(Clone, Debug, PartialEq)]
        struct RequestId(u64);

        let scope = handler.scope();
        scope.ctx().set(RequestId(42));

        scope
            .transaction(|tx| async move {
                let id = tx.ctx().get::<RequestId>();
                assert_eq!(id, Some(RequestId(42)), "tx should inherit ctx from scope");
                Ok(())
            })
            .await
            .unwrap();
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_nested_transaction_inherits_ctx() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        #[derive(Clone, Debug, PartialEq)]
        struct TraceId(String);

        let scope = handler.scope();
        scope.ctx().set(TraceId("trace-1".into()));

        scope
            .transaction(|tx| async move {
                tx.execute_write("CREATE TABLE ctx_test (id INTEGER PRIMARY KEY)", &[])
                    .await?;

                tx.transaction(|tx2| async move {
                    let tid = tx2.ctx().get::<TraceId>();
                    assert_eq!(
                        tid,
                        Some(TraceId("trace-1".into())),
                        "nested tx should inherit ctx"
                    );
                    Ok(())
                })
                .await?;
                Ok(())
            })
            .await
            .unwrap();
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_task_local_run_and_current() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        #[derive(Clone, Debug, PartialEq)]
        struct UserId(i64);

        let scope = handler.scope();
        scope.ctx().set(UserId(99));

        scope
            .run(|| async {
                let db = DbScope::current();
                let uid = db.ctx().get::<UserId>();
                assert_eq!(uid, Some(UserId(99)));
            })
            .await;
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_try_current_outside_scope() {
        assert!(DbScope::try_current().is_none());
    }

    #[tokio::test]
    #[cfg(feature = "sqlite")]
    async fn test_writer_executor_inherits_ctx() {
        let config = DbConfig::sqlite(":memory:");
        let handler = DbHandlerBuilder::new(config)
            .max_connections(2)
            .min_connections(0)
            .build()
            .await
            .unwrap();

        #[derive(Clone, Debug, PartialEq)]
        struct Role(String);

        let scope = handler.scope();
        scope.ctx().set(Role("admin".into()));

        scope
            .with_writer(|w| async move {
                let role = w.ctx().get::<Role>();
                assert_eq!(role, Some(Role("admin".into())));
                Ok(())
            })
            .await
            .unwrap();
    }
}
