sz_orm_sqlx/

any.rs

//! sqlx 后端适配器实现
//!
//! 为 MySQL、PostgreSQL、SQLite 分别实现 Connection 和 ConnectionFactory。
//! 不使用 sqlx::Any 以避免其类型限制和生命周期问题。
//!
//! 关键设计：
//! Connection trait 已手动解糖（不使用 `#[async_trait]`），所有 async 方法
//! 使用单一生命周期 `'a`（绑定 `&'a mut self` 和 `&'a str`），而非 HRTB。
//! 这样 sqlx::Executor 对 `&'c mut XxxConnection` 的 impl（针对具体 `'c`）
//! 即可满足约束，避免 "implementation of Executor is not general enough" 错误。

use async_trait::async_trait;
use futures::StreamExt;
use sqlx::{Column, Executor, Row, TypeInfo};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::Arc;
use sz_orm_core::{
    ColType, Connection, ConnectionFactory, DbError, DbType, QueryRows, QueryValues, Value,
};

use crate::error::map_sqlx_error;

/// 判断 SQL 是否需要走 raw_sql 路径
/// MySQL prepared statement 协议不支持 BEGIN/COMMIT/ROLLBACK/SAVEPOINT 等命令
fn needs_raw_sql(sql: &str) -> bool {
    let trimmed = sql.trim_start();
    let upper = trimmed.to_uppercase();
    upper.starts_with("BEGIN")
        || upper.starts_with("COMMIT")
        || upper.starts_with("ROLLBACK")
        || upper.starts_with("SAVEPOINT")
        || upper.starts_with("RELEASE")
        || upper.starts_with("SET ")
        || upper.starts_with("USE ")
        || upper.starts_with("START TRANSACTION")
}

// ===================== SQLite 适配器 =====================

// 注：sqlx 0.9 起 Executor trait 要求 'static lifetime，
// 原先的 execute_sqlite_boxed / query_sqlite_boxed 已内联到调用点。
// 见 SqlxSqliteConnection::execute / query 实现。

// 注：原 row_to_value_sqlite（按 type_info().name() 字符串 match）已被
// row_to_value_with_coltype_sqlite（按预解析 ColType 枚举分派）取代，
// 性能更优且避免每行字符串比较。详见该函数。

/// SQLite: 使用预解析的 ColType 进行类型分派（避免每行字符串 match）
///
/// 调用方预先通过 [`ColType::parse_sqlite`] 解析列类型，
/// 后续行直接用枚举分派（编译器优化为跳转表），避免每行每列的字符串比较。
///
/// **注意**：必须使用 `parse_sqlite` 而非通用 `from_type_name`：SQLite 的 INTEGER 类型
/// 实际为 64 位动态存储，通用映射会错误地归类为 I32，导致数值截断。
fn row_to_value_with_coltype_sqlite(
    row: &sqlx::sqlite::SqliteRow,
    ordinal: usize,
    col_type: ColType,
) -> Value {
    match col_type {
        ColType::Bool => match row.try_get::<Option<bool>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bool).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I8 => match row.try_get::<Option<i8>, usize>(ordinal) {
            Ok(v) => v.map(Value::I8).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I16 => match row.try_get::<Option<i16>, usize>(ordinal) {
            Ok(v) => v.map(Value::I16).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I32 => match row.try_get::<Option<i32>, usize>(ordinal) {
            Ok(v) => v.map(Value::I32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I64 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U8 => match row.try_get::<Option<u8>, usize>(ordinal) {
            Ok(v) => v.map(Value::U8).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U16 => match row.try_get::<Option<u16>, usize>(ordinal) {
            Ok(v) => v.map(Value::U16).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U32 => match row.try_get::<Option<u32>, usize>(ordinal) {
            Ok(v) => v.map(Value::U32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // SQLite 无原生 u64，按 i64 解码
        ColType::U64 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F32 => match row.try_get::<Option<f32>, usize>(ordinal) {
            Ok(v) => v.map(Value::F32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F64 => match row.try_get::<Option<f64>, usize>(ordinal) {
            Ok(v) => v.map(Value::F64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::Decimal => match row.try_get::<Option<String>, usize>(ordinal) {
            Ok(v) => v.map(Value::Decimal).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::String => match row.try_get::<Option<String>, usize>(ordinal) {
            Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::Bytes => match row.try_get::<Option<Vec<u8>>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bytes).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // SQLite 中 DATE/DATETIME/TIME/JSON/UUID 通常以 TEXT 存储
        ColType::Date | ColType::DateTime | ColType::Time | ColType::Json | ColType::Uuid => {
            match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            }
        }
        ColType::Unknown => {
            // 未知类型（表达式/聚合列，SQLite type_info 返回空串），按
            // i64 → f64 → bool → String 顺序回退（与 MySQL/PG 臂一致）。
            // 修复前 bool 优先：任意非零整数被解码为 Bool(true)，下游
            // as_i64() 恒为 1 —— COUNT(*)=50 被读成 1（2026-09-25 实测
            // 复现的静默读损坏）。表达式布尔值读为 I64(0/1)，as_bool()
            // 对整数类型语义不变。
            if let Ok(v) = row.try_get::<Option<i64>, usize>(ordinal) {
                return v.map(Value::I64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<f64>, usize>(ordinal) {
                return v.map(Value::F64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<bool>, usize>(ordinal) {
                return v.map(Value::Bool).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<String>, usize>(ordinal) {
                return v.map(Value::String).unwrap_or(Value::Null);
            }
            Value::Null
        }
        // ColType 标记为 #[non_exhaustive]，未来新增变体按 Unknown 处理
        _ => Value::Null,
    }
}

pub struct SqlitePoolHandle {
    pool: sqlx::SqlitePool,
}

impl SqlitePoolHandle {
    pub async fn connect(url: &str) -> Result<Self, DbError> {
        // 默认容量 10（向后兼容）；上层应用 `connect_with_max` 对齐外层 PoolConfig::max_size
        Self::connect_with_max(url, 10).await
    }

    /// 容量显式版 connect（2026-10-28 P3-4，PERFV2-03）。
    ///
    /// sz-orm 外层 `PoolConfig::max_size` 与内层 sqlx 池容量不一致时，
    /// 实际并发 = min(max_size, 内层容量) 而外层按 max_size 记账——
    /// `max_size > 内层容量` 时会确定性出现 acquire 超时。
    /// 创建 handle 时用本方法把内层容量对齐到外层 max_size。
    pub async fn connect_with_max(url: &str, max_connections: u32) -> Result<Self, DbError> {
        // SQLite 生产配置：WAL + NORMAL 同步 + 5s 忙等 + 256MB mmap
        // - WAL：写不阻塞读，崩溃后通过 WAL 恢复，远比 DELETE 模式安全
        // - Synchronous=Normal：WAL 模式下仅在 checkpoint 时 fsync，性能 ~2x 于 Full
        // - busy_timeout=5s：避免 "database is locked" 误报
        // - mmap_size=256MB：大结果集减少 syscall，提升读取吞吐
        let opts = sqlx::sqlite::SqliteConnectOptions::from_str(url)
            .map_err(map_sqlx_error)?
            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
            .synchronous(sqlx::sqlite::SqliteSynchronous::Normal)
            .busy_timeout(std::time::Duration::from_secs(5))
            .pragma("mmap_size", "268435456");
        let pool = sqlx::sqlite::SqlitePoolOptions::new()
            .max_connections(max_connections)
            .acquire_timeout(std::time::Duration::from_secs(30))
            .idle_timeout(Some(std::time::Duration::from_secs(600)))
            .max_lifetime(Some(std::time::Duration::from_secs(1800)))
            .connect_with(opts)
            .await
            .map_err(map_sqlx_error)?;
        Ok(Self { pool })
    }

    pub fn from_pool(pool: sqlx::SqlitePool) -> Self {
        Self { pool }
    }

    pub fn pool(&self) -> &sqlx::SqlitePool {
        &self.pool
    }
}

pub struct SqlxSqliteConnectionFactory {
    pool: Arc<SqlitePoolHandle>,
}

impl SqlxSqliteConnectionFactory {
    pub fn new(pool: Arc<SqlitePoolHandle>) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl ConnectionFactory for SqlxSqliteConnectionFactory {
    async fn create(&self) -> Result<Box<dyn Connection>, DbError> {
        let conn = self.pool.pool.acquire().await.map_err(map_sqlx_error)?;
        // 2026-10-28 P1-3：MetricsConn 包装统一上报查询指标（ops/latency/errors）
        Ok(Box::new(crate::metrics_conn::MetricsConn::new(Box::new(
            SqlxSqliteConnection {
                conn: Some(conn),
                connected: true,
                in_transaction: false,
            },
        ))))
    }
}

pub struct SqlxSqliteConnection {
    conn: Option<sqlx::pool::PoolConnection<sqlx::Sqlite>>,
    connected: bool,
    in_transaction: bool,
}

impl Connection for SqlxSqliteConnection {
    fn db_type(&self) -> Option<DbType> {
        Some(DbType::Sqlite)
    }

    fn execute<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let result = if needs_raw_sql(sql) {
                (&mut *pool_conn)
                    .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
                    .await
            } else {
                (&mut *pool_conn).execute(sqlx::AssertSqlSafe(sql)).await
            };
            self.conn = Some(pool_conn);

            match result {
                Ok(r) => Ok(r.rows_affected()),
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    fn query<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, DbError>> + Send + 'a>>
    {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);

            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            // 预解析列类型（只在第一行解析，后续行复用，避免每行字符串 match）
            // SQLite 专用解析：INTEGER 映射为 I64（SQLite 整数动态存储，可容纳 64 位）
            let col_types: Vec<ColType> = rows[0]
                .columns()
                .iter()
                .map(|col| ColType::parse_sqlite(col.type_info().name()))
                .collect();
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_types.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_sqlite(row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    fn begin_transaction<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                return Err(DbError::Internal("transaction already started".to_string()));
            }
            self.execute("BEGIN").await?;
            self.in_transaction = true;
            Ok(())
        })
    }

    fn commit<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                self.execute("COMMIT").await?;
                self.in_transaction = false;
            }
            Ok(())
        })
    }

    fn rollback<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                // 回滚失败时保留事务标志：非连接类错误下服务端事务可能仍开启，
                // 若此刻清标志，连接带着开放事务回池将绕过归还路径的自动回滚
                // （事务泄漏）。失败后由归还路径重试回滚或直接关闭连接。
                // 语义与 Oracle/MSSQL 适配器一致。
                let result = self.execute("ROLLBACK").await;
                if result.is_ok() {
                    self.in_transaction = false;
                }
                result.map(|_| ())
            } else {
                Ok(())
            }
        })
    }

    fn is_connected(&self) -> bool {
        // 纳入 conn.is_some()：query_stream 提前弃读时 PoolConnection 已从包装器
        // take() 出去（self.conn = None），毒化状态不得被判定为可用连接
        // （否则回池后下次 acquire 借出即报 "connection already closed"）
        self.connected && self.conn.is_some()
    }

    fn in_transaction(&self) -> bool {
        self.in_transaction
    }

    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
        Box::pin(async move {
            match self.execute("SELECT 1").await {
                Ok(_) => true,
                Err(_) => {
                    self.connected = false;
                    false
                }
            }
        })
    }

    fn close<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if let Some(conn) = self.conn.take() {
                drop(conn);
            }
            self.connected = false;
            self.in_transaction = false;
            Ok(())
        })
    }

    /// SQLite 参数绑定执行（INSERT/UPDATE/DELETE）
    ///
    /// 使用 sqlx prepared statement 绑定参数，避免 SQL 注入与字符串转义开销。
    /// 对 `Value::Bool`/`I8`..=`I64`/`U8`..=`U64`/`F32`/`F64`/`String`/`Bytes` 直接 bind；
    /// 其他类型（Date/DateTime/Json/Array/Object）回退为 `to_string()` 后以 TEXT 绑定。
    fn execute_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if needs_raw_sql(sql) || params.is_empty() {
                return self.execute(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let result = q.execute(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            match result {
                Ok(r) => Ok(r.rows_affected()),
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    /// SQLite 参数绑定查询（SELECT，HashMap 映射）
    ///
    /// 使用 sqlx prepared statement 绑定参数，结果按预解析 ColType 解码为
    /// `HashMap<String, Value>`。与 `query` 的区别仅在于参数绑定路径。
    fn query_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            // 预解析列类型（只在第一行解析，后续行复用）
            // SQLite 专用解析：INTEGER → I64
            let col_types: Vec<ColType> = rows[0]
                .columns()
                .iter()
                .map(|col| ColType::parse_sqlite(col.type_info().name()))
                .collect();
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_types.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_sqlite(row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    /// SQLite 位置式查询（SELECT，无参数）
    ///
    /// 绕过 `HashMap<String, Value>` 行映射，返回 `(列名列表, 按列序号的值矩阵)`。
    /// 列名与 ColType 仅 `to_string`/解析一次，后续行复用；每行值按列序号直接 `Vec::push`，
    /// 无哈希计算与字符串克隆。适用于 SELECT ALL 大结果集场景。
    fn query_values<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            let mut col_types: Vec<ColType> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
                col_types.push(ColType::parse_sqlite(col.type_info().name()));
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    row_values.push(row_to_value_with_coltype_sqlite(row, idx, col_types[idx]));
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// SQLite 参数绑定位置式查询（SELECT）
    ///
    /// 叠加 prepared statement + 位置式映射 + ColType 预解析三重优化。
    fn query_values_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query_values(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            let mut col_types: Vec<ColType> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
                col_types.push(ColType::parse_sqlite(col.type_info().name()));
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    row_values.push(row_to_value_with_coltype_sqlite(row, idx, col_types[idx]));
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// SQLite 流式查询：逐行返回结果，避免大结果集 `fetch_all` 内存峰值
    ///
    /// 使用 `sqlx::query::fetch` 获取行流，逐行映射为 `HashMap<String, Value>`。
    /// 流正常消费完毕后连接归还 `self.conn`；若提前 drop 流，连接通过
    /// `PoolConnection::drop` 归还到 sqlx 池，但 `self.conn` 变为 `None`。
    fn query_stream<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn futures::Stream<Item = Result<HashMap<String, Value>, DbError>> + Send + 'a>>
    {
        Box::pin(async_stream::try_stream! {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut row_stream = sqlx::query(sqlx::AssertSqlSafe(sql)).fetch(&mut *pool_conn);
            let mut col_types: Vec<ColType> = Vec::new();
            let mut col_names: Vec<String> = Vec::new();
            let mut first_row = true;
            while let Some(row_result) = row_stream.next().await {
                let row = row_result.map_err(map_sqlx_error)?;
                if first_row {
                    for col in row.columns() {
                        col_names.push(col.name().to_string());
                        col_types.push(ColType::parse_sqlite(col.type_info().name()));
                    }
                    first_row = false;
                }
                let mut record = HashMap::with_capacity(col_names.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_sqlite(&row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                yield record;
            }
            // 显式释放 row_stream 对 pool_conn 的可变借用，否则借用检查器
            // 无法证明 row_stream 在移动 pool_conn 前已被销毁（E0505）
            drop(row_stream);
            self.conn = Some(pool_conn);
        })
    }
}

impl Drop for SqlxSqliteConnection {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            drop(conn);
        }
    }
}

/// SQLite 在线备份（使用 VACUUM INTO，要求 SQLite 3.27+）
///
/// 将当前数据库完整复制到目标路径，原数据库可继续读写（在线备份）。
/// 目标文件若已存在会报错。路径中的单引号会被转义以防 SQL 注入。
pub async fn sqlite_backup(
    conn: &mut SqlxSqliteConnection,
    dest_path: &str,
) -> Result<(), DbError> {
    let escaped_path = dest_path.replace('\'', "''");
    let sql = format!("VACUUM INTO '{}'", escaped_path);
    conn.execute(&sql).await?;
    Ok(())
}

// ===================== MySQL 适配器 =====================

// 注：sqlx 0.9 起 Executor trait 要求 'static lifetime，
// 原先的 execute_mysql_boxed / query_mysql_boxed 已内联到调用点。

fn row_to_value_mysql(row: &sqlx::mysql::MySqlRow, ordinal: usize) -> Value {
    use sqlx::TypeInfo;
    let type_name = row.columns()[ordinal].type_info().name();
    match type_name {
        "BOOLEAN" => match row.try_get::<Option<bool>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bool).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        "TINYINT" | "TINYINT UNSIGNED" => match row.try_get::<Option<i8>, usize>(ordinal) {
            Ok(v) => v.map(Value::I8).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u8>, usize>(ordinal) {
                Ok(v) => v.map(Value::U8).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        "SMALLINT" | "SMALLINT UNSIGNED" => match row.try_get::<Option<i16>, usize>(ordinal) {
            Ok(v) => v.map(Value::I16).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u16>, usize>(ordinal) {
                Ok(v) => v.map(Value::U16).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        "INT" | "INT UNSIGNED" | "MEDIUMINT" | "MEDIUMINT UNSIGNED" => {
            match row.try_get::<Option<i32>, usize>(ordinal) {
                Ok(v) => v.map(Value::I32).unwrap_or(Value::Null),
                Err(_) => match row.try_get::<Option<u32>, usize>(ordinal) {
                    Ok(v) => v.map(Value::U32).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        "BIGINT" | "BIGINT UNSIGNED" => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u64>, usize>(ordinal) {
                Ok(v) => v.map(Value::U64).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        "FLOAT" => match row.try_get::<Option<f32>, usize>(ordinal) {
            Ok(v) => v.map(Value::F32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        "DOUBLE" => match row.try_get::<Option<f64>, usize>(ordinal) {
            Ok(v) => v.map(Value::F64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        "VARCHAR" | "TEXT" | "CHAR" | "TINYTEXT" | "MEDIUMTEXT" | "LONGTEXT" | "ENUM" => {
            match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            }
        }
        "BLOB" | "TINYBLOB" | "MEDIUMBLOB" | "LONGBLOB" | "BINARY" | "VARBINARY" => {
            match row.try_get::<Option<Vec<u8>>, usize>(ordinal) {
                Ok(v) => v.map(Value::Bytes).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            }
        }
        // DECIMAL/NUMERIC 使用任意精度 BigDecimal 解码，38 位不截断
        "DECIMAL" | "NUMERIC" | "NEWDECIMAL" => {
            match row.try_get::<Option<bigdecimal::BigDecimal>, usize>(ordinal) {
                Ok(Some(v)) => Value::Decimal(v.to_string()),
                Ok(None) => Value::Null,
                Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                    Ok(v) => v.map(Value::Decimal).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        // 时间类型必须按 chrono 解码：MySQL 二进制协议下按 String 解码必然失败 → 静默 NULL
        "DATE" => match row.try_get::<Option<sqlx::types::chrono::NaiveDate>, usize>(ordinal) {
            Ok(v) => v.map(|d| Value::Date(d.to_string())).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        "DATETIME" | "TIMESTAMP" => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveDateTime>, usize>(ordinal) {
                Ok(v) => v
                    .map(|d| Value::DateTime(d.to_string()))
                    .unwrap_or(Value::Null),
                Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                    Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        "TIME" => match row.try_get::<Option<sqlx::types::chrono::NaiveTime>, usize>(ordinal) {
            Ok(v) => v.map(|t| Value::Time(t.to_string())).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        _ => {
            // 未知类型回退：i64 → f64 → bool → String
            if let Ok(v) = row.try_get::<Option<i64>, usize>(ordinal) {
                return v.map(Value::I64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<f64>, usize>(ordinal) {
                return v.map(Value::F64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<bool>, usize>(ordinal) {
                return v.map(Value::Bool).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<String>, usize>(ordinal) {
                return v.map(Value::String).unwrap_or(Value::Null);
            }
            Value::Null
        }
    }
}

/// MySQL: 使用预解析的 ColType 进行类型分派（避免每行字符串 match）
///
/// 与 [`row_to_value_mysql`] 的区别：调用方预先通过 [`ColType::parse_mysql`] 解析列类型，
/// 后续行直接用枚举分派（编译器优化为跳转表），避免每行每列的字符串比较。
///
/// **注意**：必须使用 `parse_mysql` 而非通用 `from_type_name`：MySQL 协议报告的类型名
/// 包含 NEWDECIMAL/YEAR/ENUM/SET 等特有类型，通用映射无法识别。
fn row_to_value_with_coltype_mysql(
    row: &sqlx::mysql::MySqlRow,
    ordinal: usize,
    col_type: ColType,
) -> Value {
    match col_type {
        ColType::Bool => match row.try_get::<Option<bool>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bool).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I8 => match row.try_get::<Option<i8>, usize>(ordinal) {
            Ok(v) => v.map(Value::I8).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u8>, usize>(ordinal) {
                Ok(v) => v.map(Value::U8).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::I16 => match row.try_get::<Option<i16>, usize>(ordinal) {
            Ok(v) => v.map(Value::I16).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u16>, usize>(ordinal) {
                Ok(v) => v.map(Value::U16).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::I32 => match row.try_get::<Option<i32>, usize>(ordinal) {
            Ok(v) => v.map(Value::I32).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u32>, usize>(ordinal) {
                Ok(v) => v.map(Value::U32).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::I64 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<u64>, usize>(ordinal) {
                Ok(v) => v.map(Value::U64).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::U8 => match row.try_get::<Option<u8>, usize>(ordinal) {
            Ok(v) => v.map(Value::U8).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U16 => match row.try_get::<Option<u16>, usize>(ordinal) {
            Ok(v) => v.map(Value::U16).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U32 => match row.try_get::<Option<u32>, usize>(ordinal) {
            Ok(v) => v.map(Value::U32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U64 => match row.try_get::<Option<u64>, usize>(ordinal) {
            Ok(v) => v.map(Value::U64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F32 => match row.try_get::<Option<f32>, usize>(ordinal) {
            Ok(v) => v.map(Value::F32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F64 => match row.try_get::<Option<f64>, usize>(ordinal) {
            Ok(v) => v.map(Value::F64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // DECIMAL/NUMERIC 使用任意精度 BigDecimal 解码，38 位不截断
        ColType::Decimal => match row.try_get::<Option<bigdecimal::BigDecimal>, usize>(ordinal) {
            Ok(Some(v)) => Value::Decimal(v.to_string()),
            Ok(None) => Value::Null,
            Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::Decimal).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::String => match row.try_get::<Option<String>, usize>(ordinal) {
            Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::Bytes => match row.try_get::<Option<Vec<u8>>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bytes).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // 时间类型必须按 chrono 解码：MySQL 二进制协议下按 String 解码必然失败 → 静默 NULL
        ColType::Date => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveDate>, usize>(ordinal) {
                Ok(v) => v.map(|d| Value::Date(d.to_string())).unwrap_or(Value::Null),
                Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                    Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        ColType::DateTime => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveDateTime>, usize>(ordinal) {
                Ok(v) => v
                    .map(|d| Value::DateTime(d.to_string()))
                    .unwrap_or(Value::Null),
                Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                    Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        ColType::Time => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveTime>, usize>(ordinal) {
                Ok(v) => v.map(|t| Value::Time(t.to_string())).unwrap_or(Value::Null),
                Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                    Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                    Err(_) => Value::Null,
                },
            }
        }
        // JSON/UUID 在 MySQL 协议下以文本传输
        ColType::Json | ColType::Uuid => match row.try_get::<Option<String>, usize>(ordinal) {
            Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::Unknown => {
            // 未知类型回退：i64 → f64 → bool → String
            if let Ok(v) = row.try_get::<Option<i64>, usize>(ordinal) {
                return v.map(Value::I64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<f64>, usize>(ordinal) {
                return v.map(Value::F64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<bool>, usize>(ordinal) {
                return v.map(Value::Bool).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<String>, usize>(ordinal) {
                return v.map(Value::String).unwrap_or(Value::Null);
            }
            Value::Null
        }
        // ColType 标记为 #[non_exhaustive]，未来新增变体按 Unknown 处理
        _ => Value::Null,
    }
}

pub struct MySqlPoolHandle {
    pool: sqlx::MySqlPool,
}

impl MySqlPoolHandle {
    pub async fn connect(url: &str) -> Result<Self, DbError> {
        // 默认容量 10（向后兼容）；上层应用 `connect_with_max` 对齐外层 PoolConfig::max_size
        Self::connect_with_max(url, 10).await
    }

    /// 容量显式版 connect（2026-10-28 P3-4，PERFV2-03）：与外层 `PoolConfig::max_size` 对齐，
    /// 避免"外层按 max_size 记账、内层硬编码 10"的双层上限不一致（详见 SqlitePoolHandle::connect_with_max）。
    pub async fn connect_with_max(url: &str, max_connections: u32) -> Result<Self, DbError> {
        let pool = sqlx::pool::PoolOptions::<sqlx::MySql>::new()
            .max_connections(max_connections)
            .acquire_timeout(std::time::Duration::from_secs(30))
            .idle_timeout(Some(std::time::Duration::from_secs(600)))
            .max_lifetime(Some(std::time::Duration::from_secs(1800)))
            .connect(url)
            .await
            .map_err(map_sqlx_error)?;
        Ok(Self { pool })
    }

    pub fn from_pool(pool: sqlx::MySqlPool) -> Self {
        Self { pool }
    }

    pub fn pool(&self) -> &sqlx::MySqlPool {
        &self.pool
    }
}

pub struct SqlxMySqlConnectionFactory {
    pool: Arc<MySqlPoolHandle>,
}

impl SqlxMySqlConnectionFactory {
    pub fn new(pool: Arc<MySqlPoolHandle>) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl ConnectionFactory for SqlxMySqlConnectionFactory {
    async fn create(&self) -> Result<Box<dyn Connection>, DbError> {
        let conn = self.pool.pool.acquire().await.map_err(map_sqlx_error)?;
        // 2026-10-28 P1-3：MetricsConn 包装统一上报查询指标（ops/latency/errors）
        Ok(Box::new(crate::metrics_conn::MetricsConn::new(Box::new(
            SqlxMySqlConnection {
                conn: Some(conn),
                connected: true,
                in_transaction: false,
                // 新连接会话天然处于默认状态：干净（R5 脏标记跳过复位）
                session_dirty: false,
            },
        ))))
    }
}

pub struct SqlxMySqlConnection {
    conn: Option<sqlx::pool::PoolConnection<sqlx::MySql>>,
    connected: bool,
    in_transaction: bool,
    /// R5 优化（v10.13.1）：本次租用是否执行过会话级语句（SET/USE）。
    /// 连接池归还时据此跳过无条件复位（省 1 条语句 + 1 个往返/借用）。
    session_dirty: bool,
}

/// 判断一条 SQL 是否可能在成功执行后改变会话级状态（R5 脏标记依据）。
///
/// 置脏：`SET`（含 `SET SESSION`/`SET NAMES`/`SET autocommit`/用户变量）与 `USE`。
/// 不置脏：`SET TRANSACTION`（按 R3 仅作用下一事务，事务结束不残留）、
/// 只读语句（`SELECT`/`SHOW`/`EXPLAIN`/`DESC`）、事务语句
/// （`START TRANSACTION`/`BEGIN`/`COMMIT`/`ROLLBACK`——autocommit 语义自动恢复，
/// 且池在归还时本就回滚未提交事务）。
pub(crate) fn marks_session_dirty(sql: &str) -> bool {
    // 大小写不敏感地截取语句头部比较（32 字符足够覆盖前两个词）
    let trimmed = sql.trim_start();
    let mut upper = String::with_capacity(32);
    for ch in trimmed.chars() {
        if upper.len() >= 32 {
            break;
        }
        upper.push(ch.to_ascii_uppercase());
    }
    if upper.starts_with("USE ") || upper.starts_with("USE\t") || upper == "USE" {
        return true;
    }
    if !upper.starts_with("SET") {
        return false;
    }
    // "SET" 后必须跟词边界（排除 "SETUP..." 这类误判）
    let after_set = &upper[3..];
    if !(after_set.is_empty() || after_set.starts_with(' ') || after_set.starts_with('\t')) {
        return false;
    }
    let rest = after_set.trim_start();
    // `SET TRANSACTION ...` 仅作用下一事务；`SET TRANSACTIONX` 是标识符语境，置脏
    !(rest.starts_with("TRANSACTION")
        && rest["TRANSACTION".len()..]
            .chars()
            .next()
            .map(|c| c.is_whitespace() || c == ';')
            .unwrap_or(true))
}

impl Connection for SqlxMySqlConnection {
    fn db_type(&self) -> Option<DbType> {
        Some(DbType::MySQL)
    }

    /// MySQL 的 `COM_RESET_CONNECTION` 是协议命令而非 SQL 语句（sqlx 未暴露），
    /// 归还连接时用单条 `SET SESSION` 复位关键会话变量。必须保持单语句且
    /// 版本无关：`transaction_isolation`/`max_execution_time` 在 MySQL 5.7、
    /// MariaDB 上名称为 `tx_isolation`/`max_statement_time`，故不在此复位；
    /// 隔离级别由事务层按“下一事务”作用域下发（R3），不会残留到会话。
    fn session_reset_sql(&self) -> Option<&'static str> {
        Some(
            "SET SESSION sql_mode = DEFAULT, \
             SESSION character_set_client = @@GLOBAL.character_set_client, \
             SESSION character_set_results = @@GLOBAL.character_set_results, \
             SESSION character_set_connection = @@GLOBAL.character_set_connection, \
             SESSION time_zone = @@GLOBAL.time_zone, \
             SESSION autocommit = 1",
        )
    }

    /// R5 脏标记（v10.13.1）：读取并清除脏标记（take 语义）。
    /// 返回 false 时池跳过复位语句——连接从未执行过 SET/USE，会话仍是默认状态。
    fn take_session_dirty(&mut self) -> bool {
        std::mem::take(&mut self.session_dirty)
    }

    /// 复位语句执行成功后由池调用；复位 SQL 是 SET，执行路径会置脏，这里统一清零
    fn mark_session_clean(&mut self) {
        self.session_dirty = false;
    }

    fn execute<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let result = if needs_raw_sql(sql) {
                (&mut *pool_conn)
                    .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
                    .await
            } else {
                (&mut *pool_conn).execute(sqlx::AssertSqlSafe(sql)).await
            };
            self.conn = Some(pool_conn);

            match result {
                Ok(r) => {
                    // R5 脏标记（v10.13.1）：仅执行成功的语句才可能改变会话状态
                    if marks_session_dirty(sql) {
                        self.session_dirty = true;
                    }
                    Ok(r.rows_affected())
                }
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    fn query<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, DbError>> + Send + 'a>>
    {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);

            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            // 预解析列类型（只在第一行解析，后续行复用，避免每行字符串 match）
            // MySQL 专用解析：覆盖 MySQL 协议特有类型名（NEWDECIMAL/YEAR/ENUM/SET 等）
            let col_types: Vec<ColType> = rows[0]
                .columns()
                .iter()
                .map(|col| ColType::parse_mysql(col.type_info().name()))
                .collect();
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_types.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_mysql(row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    fn begin_transaction<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                return Err(DbError::Internal("transaction already started".to_string()));
            }
            self.execute("BEGIN").await?;
            self.in_transaction = true;
            Ok(())
        })
    }

    fn commit<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                self.execute("COMMIT").await?;
                self.in_transaction = false;
            }
            Ok(())
        })
    }

    fn rollback<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                // 回滚失败时保留事务标志：非连接类错误下服务端事务可能仍开启，
                // 若此刻清标志，连接带着开放事务回池将绕过归还路径的自动回滚
                // （事务泄漏）。失败后由归还路径重试回滚或直接关闭连接。
                // 语义与 Oracle/MSSQL 适配器一致。
                let result = self.execute("ROLLBACK").await;
                if result.is_ok() {
                    self.in_transaction = false;
                }
                result.map(|_| ())
            } else {
                Ok(())
            }
        })
    }

    fn is_connected(&self) -> bool {
        // 纳入 conn.is_some()：query_stream 提前弃读时 PoolConnection 已从包装器
        // take() 出去（self.conn = None），毒化状态不得被判定为可用连接
        // （否则回池后下次 acquire 借出即报 "connection already closed"）
        self.connected && self.conn.is_some()
    }

    fn in_transaction(&self) -> bool {
        self.in_transaction
    }

    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
        Box::pin(async move {
            match self.execute("SELECT 1").await {
                Ok(_) => true,
                Err(_) => {
                    self.connected = false;
                    false
                }
            }
        })
    }

    fn close<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if let Some(conn) = self.conn.take() {
                drop(conn);
            }
            self.connected = false;
            self.in_transaction = false;
            Ok(())
        })
    }

    /// MySQL 参数绑定执行（INSERT/UPDATE/DELETE）
    ///
    /// 使用 sqlx prepared statement 绑定参数。MySQL 协议原生支持 `?` 占位符。
    /// 对 `Value::Bool`/`I8`..=`I64`/`U8`..=`U64`/`F32`/`F64`/`String`/`Bytes` 直接 bind；
    /// 其他类型回退为 `to_string()` 后以 TEXT 绑定。
    fn execute_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if needs_raw_sql(sql) || params.is_empty() {
                return self.execute(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let result = q.execute(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            match result {
                Ok(r) => {
                    // R5 脏标记（v10.13.1）：参数化路径同样可能执行 SET（如 SET x = ?）
                    if marks_session_dirty(sql) {
                        self.session_dirty = true;
                    }
                    Ok(r.rows_affected())
                }
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    /// MySQL 参数绑定查询（SELECT，HashMap 映射）
    fn query_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_names.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_mysql(row, i);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    /// MySQL 位置式查询（SELECT，无参数）
    ///
    /// 绕过 HashMap 行映射，返回 `(列名, 按列序号的值矩阵)`。适用于 SELECT ALL 大结果集。
    fn query_values<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            let mut col_types: Vec<ColType> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
                col_types.push(ColType::parse_mysql(col.type_info().name()));
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    row_values.push(row_to_value_with_coltype_mysql(row, idx, col_types[idx]));
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// MySQL 参数绑定位置式查询（SELECT）
    fn query_values_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query_values(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut q = sqlx::query(sqlx::AssertSqlSafe(sql));
            for v in params {
                q = match v {
                    Value::Null => q.bind(None::<i64>),
                    Value::Bool(b) => q.bind(*b),
                    Value::I8(n) => q.bind(*n),
                    Value::I16(n) => q.bind(*n),
                    Value::I32(n) => q.bind(*n),
                    Value::I64(n) => q.bind(*n),
                    Value::U8(n) => q.bind(*n),
                    Value::U16(n) => q.bind(*n),
                    Value::U32(n) => q.bind(*n),
                    Value::U64(n) => q.bind(*n as i64),
                    Value::F32(f) => q.bind(*f),
                    Value::F64(f) => q.bind(*f),
                    Value::String(s) => q.bind(s.as_str()),
                    Value::Decimal(s) => q.bind(s.as_str()),
                    Value::Bytes(b) => q.bind(b.as_slice()),
                    // 字符串类变体直接按文本绑定：`Value::Display` 会给字符串加引号，
                    // 走 `to_string()` 会把 '2026-01-01 00:00:00' 连同引号一起写入
                    Value::Uuid(s)
                    | Value::Date(s)
                    | Value::DateTime(s)
                    | Value::Time(s)
                    | Value::Json(s) => q.bind(s.as_str()),
                    other => q.bind(other.to_string()),
                };
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            let mut col_types: Vec<ColType> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
                col_types.push(ColType::parse_mysql(col.type_info().name()));
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    row_values.push(row_to_value_with_coltype_mysql(row, idx, col_types[idx]));
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// MySQL 流式查询：逐行返回结果，避免大结果集 `fetch_all` 内存峰值
    ///
    /// 使用 `sqlx::query::fetch` 获取行流，逐行映射为 `HashMap<String, Value>`。
    /// 流正常消费完毕后连接归还 `self.conn`；若提前 drop 流，连接通过
    /// `PoolConnection::drop` 归还到 sqlx 池，但 `self.conn` 变为 `None`。
    fn query_stream<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn futures::Stream<Item = Result<HashMap<String, Value>, DbError>> + Send + 'a>>
    {
        Box::pin(async_stream::try_stream! {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut row_stream = sqlx::query(sqlx::AssertSqlSafe(sql)).fetch(&mut *pool_conn);
            let mut col_types: Vec<ColType> = Vec::new();
            let mut col_names: Vec<String> = Vec::new();
            let mut first_row = true;
            while let Some(row_result) = row_stream.next().await {
                let row = row_result.map_err(map_sqlx_error)?;
                if first_row {
                    for col in row.columns() {
                        col_names.push(col.name().to_string());
                        col_types.push(ColType::parse_mysql(col.type_info().name()));
                    }
                    first_row = false;
                }
                let mut record = HashMap::with_capacity(col_names.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_mysql(&row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                yield record;
            }
            // 显式释放 row_stream 对 pool_conn 的可变借用，否则借用检查器
            // 无法证明 row_stream 在移动 pool_conn 前已被销毁（E0505）
            drop(row_stream);
            self.conn = Some(pool_conn);
        })
    }
}

impl Drop for SqlxMySqlConnection {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            drop(conn);
        }
    }
}

/// MySQL 批量导入（多行 INSERT，作为 LOAD DATA LOCAL INFILE 的安全替代）
///
/// 构建多行 INSERT 语句：`INSERT INTO t (c1, c2) VALUES (?, ?), (?, ?), ...`
/// 使用参数绑定避免 SQL 注入。MySQL 原生支持 `?` 占位符。
/// 当数据量极大时建议分批调用（每批 1000~10000 行），避免 SQL 过长或超出
/// `max_allowed_packet` 限制。
pub async fn mysql_bulk_insert(
    conn: &mut SqlxMySqlConnection,
    table: &str,
    columns: &[&str],
    rows: &[Vec<Value>],
) -> Result<u64, DbError> {
    if rows.is_empty() {
        return Ok(0);
    }
    let col_list = columns.join(", ");
    let cols_per_row = columns.len();
    // MySQL 占位符 ? 每列一个，跨行复用相同模式
    let row_placeholder = format!("({})", vec!["?"; cols_per_row].join(", "));
    let placeholders = vec![row_placeholder; rows.len()].join(", ");
    let sql = format!(
        "INSERT INTO {} ({}) VALUES {}",
        table, col_list, placeholders
    );
    // 展平所有行的值为单一参数数组
    let mut params: Vec<Value> = Vec::with_capacity(rows.len() * cols_per_row);
    for row in rows {
        for v in row {
            params.push(v.clone());
        }
    }
    conn.execute_with_params(&sql, &params).await
}

// ===================== PostgreSQL 适配器 =====================

// 注：sqlx 0.9 起 Executor trait 要求 'static lifetime，
// 原先的 execute_pg_boxed / query_pg_boxed 已内联到调用点。

/// 构造带列名的解码错误：时间类型解码失败必须上抛，不得静默变成 NULL
fn pg_decode_err(
    row: &sqlx::postgres::PgRow,
    ordinal: usize,
    type_name: &str,
    err: sqlx::Error,
) -> DbError {
    let column = row
        .columns()
        .get(ordinal)
        .map(|c| c.name().to_string())
        .unwrap_or_else(|| format!("#{ordinal}"));
    DbError::QueryError(format!(
        "PostgreSQL 列 `{column}`（类型 {type_name}，序号 {ordinal}）解码失败：{err}；\
         已拒绝以 NULL 代替真实值"
    ))
}

fn row_to_value_pg(row: &sqlx::postgres::PgRow, ordinal: usize) -> Result<Value, DbError> {
    use sqlx::TypeInfo;
    let type_name = row.columns()[ordinal].type_info().name();
    macro_rules! dec {
        ($ty:ty) => {
            row.try_get::<Option<$ty>, usize>(ordinal)
                .map_err(|e| pg_decode_err(row, ordinal, type_name, e))
        };
    }
    match type_name {
        "BOOL" => Ok(dec!(bool)?.map(Value::Bool).unwrap_or(Value::Null)),
        "INT2" => Ok(dec!(i16)?.map(Value::I16).unwrap_or(Value::Null)),
        "INT4" | "OID" => Ok(dec!(i32)?.map(Value::I32).unwrap_or(Value::Null)),
        "INT8" => Ok(dec!(i64)?.map(Value::I64).unwrap_or(Value::Null)),
        "FLOAT4" => Ok(dec!(f32)?.map(Value::F32).unwrap_or(Value::Null)),
        "FLOAT8" => Ok(dec!(f64)?.map(Value::F64).unwrap_or(Value::Null)),
        "TEXT" | "VARCHAR" | "CHAR" | "NAME" | "BPCHAR" => {
            Ok(dec!(String)?.map(Value::String).unwrap_or(Value::Null))
        }
        "BYTEA" => Ok(dec!(Vec<u8>)?.map(Value::Bytes).unwrap_or(Value::Null)),
        // R2：NUMERIC 用任意精度 BigDecimal 解码，38 位不截断
        "NUMERIC" => Ok(dec!(bigdecimal::BigDecimal)?
            .map(|v| Value::Decimal(v.to_string()))
            .unwrap_or(Value::Null)),
        "UUID" => Ok(dec!(sqlx::types::Uuid)?
            .map(|uuid| Value::String(uuid.to_string()))
            .unwrap_or(Value::Null)),
        // R1：时间类型此前落入未知分支 → 静默 NULL
        "TIMESTAMP" => Ok(dec!(sqlx::types::chrono::NaiveDateTime)?
            .map(|v| Value::DateTime(v.to_string()))
            .unwrap_or(Value::Null)),
        "TIMESTAMPTZ" => Ok(
            dec!(sqlx::types::chrono::DateTime<sqlx::types::chrono::Utc>)?
                .map(|v| Value::DateTime(v.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()))
                .unwrap_or(Value::Null),
        ),
        "DATE" => Ok(dec!(sqlx::types::chrono::NaiveDate)?
            .map(|v| Value::Date(v.to_string()))
            .unwrap_or(Value::Null)),
        "TIME" => Ok(dec!(sqlx::types::chrono::NaiveTime)?
            .map(|v| Value::Time(v.to_string()))
            .unwrap_or(Value::Null)),
        "JSON" | "JSONB" => Ok(dec!(serde_json::Value)?
            .map(|v| Value::Json(v.to_string()))
            .unwrap_or(Value::Null)),
        _ => {
            // 未知类型回退：i64 → f64 → bool → String；全部失败则报错（不再静默 NULL）
            if let Ok(v) = row.try_get::<Option<i64>, usize>(ordinal) {
                return Ok(v.map(Value::I64).unwrap_or(Value::Null));
            }
            if let Ok(v) = row.try_get::<Option<f64>, usize>(ordinal) {
                return Ok(v.map(Value::F64).unwrap_or(Value::Null));
            }
            if let Ok(v) = row.try_get::<Option<bool>, usize>(ordinal) {
                return Ok(v.map(Value::Bool).unwrap_or(Value::Null));
            }
            match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => Ok(v.map(Value::String).unwrap_or(Value::Null)),
                Err(e) => Err(pg_decode_err(row, ordinal, type_name, e)),
            }
        }
    }
}

/// PostgreSQL: 使用预解析的 ColType 进行类型分派（避免每行字符串 match）
///
/// 与 [`row_to_value_pg`] 的区别：调用方预先通过 [`ColType::parse_postgres`] 解析列类型，
/// 后续行直接用枚举分派（编译器优化为跳转表），避免每行每列的字符串比较。
///
/// **注意**：必须使用 `parse_postgres` 而非通用 `from_type_name`：PostgreSQL 使用
/// PG 内部类型名（INT4/INT8/FLOAT8/BPCHAR/JSONB/TIMESTAMPTZ 等），通用映射无法识别。
fn row_to_value_with_coltype_pg(
    row: &sqlx::postgres::PgRow,
    ordinal: usize,
    col_type: ColType,
) -> Value {
    match col_type {
        ColType::Bool => match row.try_get::<Option<bool>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bool).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I8 => match row.try_get::<Option<i8>, usize>(ordinal) {
            Ok(v) => v.map(Value::I8).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I16 => match row.try_get::<Option<i16>, usize>(ordinal) {
            Ok(v) => v.map(Value::I16).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // PostgreSQL OID 在 parse_postgres 中归为 I32
        ColType::I32 => match row.try_get::<Option<i32>, usize>(ordinal) {
            Ok(v) => v.map(Value::I32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::I64 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // PostgreSQL 无原生无符号类型（u8/u16/u32）：
        // - ColType::U8/U16/U32 仅在 parse_postgres 误判时出现（PG 无 UNSIGNED 关键字）
        // - 退化为最小可容纳的有符号类型：U8→i16、U16→i32、U32→i64
        // - 数值正确性保持，仅在 Value 枚举上为有符号变体（与 sqlx Type<Postgres> 实现一致）
        ColType::U8 => match row.try_get::<Option<i16>, usize>(ordinal) {
            Ok(v) => v.map(Value::I16).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U16 => match row.try_get::<Option<i32>, usize>(ordinal) {
            Ok(v) => v.map(Value::I32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::U32 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // PostgreSQL 无原生 U64，按 i64 解码
        ColType::U64 => match row.try_get::<Option<i64>, usize>(ordinal) {
            Ok(v) => v.map(Value::I64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F32 => match row.try_get::<Option<f32>, usize>(ordinal) {
            Ok(v) => v.map(Value::F32).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::F64 => match row.try_get::<Option<f64>, usize>(ordinal) {
            Ok(v) => v.map(Value::F64).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // NUMERIC/DECIMAL/MONEY 使用任意精度 BigDecimal 解码，避免 38 位精度截断
        ColType::Decimal => match row.try_get::<Option<bigdecimal::BigDecimal>, usize>(ordinal) {
            Ok(Some(v)) => Value::Decimal(v.to_string()),
            Ok(None) => Value::Null,
            Err(_) => match row.try_get::<Option<rust_decimal::Decimal>, usize>(ordinal) {
                Ok(Some(v)) => Value::Decimal(v.to_string()),
                Ok(None) => Value::Null,
                Err(_) => Value::Null,
            },
        },
        ColType::String => match row.try_get::<Option<String>, usize>(ordinal) {
            Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        ColType::Bytes => match row.try_get::<Option<Vec<u8>>, usize>(ordinal) {
            Ok(v) => v.map(Value::Bytes).unwrap_or(Value::Null),
            Err(_) => Value::Null,
        },
        // 时间类型必须按 chrono 解码：PG 二进制协议下按 String 解码必然失败
        ColType::Date => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveDate>, usize>(ordinal) {
                Ok(v) => v.map(|d| Value::Date(d.to_string())).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            }
        }
        ColType::DateTime => match row
            .try_get::<Option<sqlx::types::chrono::NaiveDateTime>, usize>(ordinal)
        {
            Ok(v) => v
                .map(|d| Value::DateTime(d.to_string()))
                .unwrap_or(Value::Null),
            Err(_) => match row
                .try_get::<Option<sqlx::types::chrono::DateTime<sqlx::types::chrono::Utc>>, usize>(
                    ordinal,
                ) {
                Ok(v) => v
                    .map(|d| Value::DateTime(d.format("%Y-%m-%dT%H:%M:%S%.6fZ").to_string()))
                    .unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::Time => {
            match row.try_get::<Option<sqlx::types::chrono::NaiveTime>, usize>(ordinal) {
                Ok(v) => v.map(|t| Value::Time(t.to_string())).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            }
        }
        ColType::Json => match row.try_get::<Option<serde_json::Value>, usize>(ordinal) {
            Ok(v) => v.map(|j| Value::Json(j.to_string())).unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        // UUID：使用 sqlx::types::Uuid（16 字节）解码，避免 36 字符字符串的内存浪费
        ColType::Uuid => match row.try_get::<Option<sqlx::types::Uuid>, usize>(ordinal) {
            Ok(v) => v
                .map(|uuid| Value::String(uuid.to_string()))
                .unwrap_or(Value::Null),
            Err(_) => match row.try_get::<Option<String>, usize>(ordinal) {
                Ok(v) => v.map(Value::String).unwrap_or(Value::Null),
                Err(_) => Value::Null,
            },
        },
        ColType::Unknown => {
            // 未知类型回退：i64 → f64 → bool → String
            if let Ok(v) = row.try_get::<Option<i64>, usize>(ordinal) {
                return v.map(Value::I64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<f64>, usize>(ordinal) {
                return v.map(Value::F64).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<bool>, usize>(ordinal) {
                return v.map(Value::Bool).unwrap_or(Value::Null);
            }
            if let Ok(v) = row.try_get::<Option<String>, usize>(ordinal) {
                return v.map(Value::String).unwrap_or(Value::Null);
            }
            Value::Null
        }
        // ColType 标记为 #[non_exhaustive]，未来新增变体按 Unknown 处理
        _ => Value::Null,
    }
}

/// 将 SQL 中的 `?` 占位符重写为 PostgreSQL 原生 `$1..$n`
///
/// 仅改写字符串字面量、标识符引号、注释、dollar-quote（`$$...$$` / `$tag$...$tag$`）之外，
/// 且不属于 JSON 运算符（`?`、`?|`、`?&`、`@?`）的 `?`。
///
/// 若 SQL 在字符串/注释/dollar-quote 之外已包含 `$n` 形式原生占位符，则原样返回，
/// 避免混用导致编号错乱；字符串字面量内部的 `$1`（如 `'$1'`）不视为原生占位符，
/// 真实的 `?` 仍会被重写。
fn pg_rewrite_placeholders(sql: &str) -> std::borrow::Cow<'_, str> {
    if !sql.as_bytes().contains(&b'?') {
        return std::borrow::Cow::Borrowed(sql);
    }
    let mut out = String::with_capacity(sql.len() + 8);
    let mut chars = sql.chars().peekable();
    let mut in_single = false;
    let mut in_double = false;
    let mut in_line_comment = false;
    let mut in_block_comment = false;
    let mut dollar_tag: Option<String> = None;
    let mut prev = '\0';
    let mut n = 0usize;
    let mut saw_native = false;
    while let Some(c) = chars.next() {
        if in_line_comment {
            out.push(c);
            if c == '\n' {
                in_line_comment = false;
            }
            prev = c;
            continue;
        }
        if in_block_comment {
            out.push(c);
            if c == '*' && chars.peek() == Some(&'/') {
                out.push(chars.next().unwrap_or('/'));
                in_block_comment = false;
            }
            prev = c;
            continue;
        }
        if in_single {
            out.push(c);
            if c == '\'' {
                if chars.peek() == Some(&'\'') {
                    out.push(chars.next().unwrap_or('\''));
                } else {
                    in_single = false;
                }
            }
            prev = c;
            continue;
        }
        if in_double {
            out.push(c);
            if c == '"' {
                if chars.peek() == Some(&'"') {
                    out.push(chars.next().unwrap_or('"'));
                } else {
                    in_double = false;
                }
            }
            prev = c;
            continue;
        }
        if let Some(tag) = dollar_tag.as_deref() {
            // dollar-quote 内部：除结束定界符 `$tag$` 外原样输出
            if c == '$' {
                if let Some(extra) = pg_try_close_dollar_quote(&chars, tag) {
                    out.push(c);
                    for _ in 0..extra {
                        out.push(chars.next().unwrap_or('$'));
                    }
                    dollar_tag = None;
                } else {
                    out.push(c);
                }
            } else {
                out.push(c);
            }
            prev = c;
            continue;
        }
        match c {
            '\'' => {
                in_single = true;
                out.push(c);
            }
            '"' => {
                in_double = true;
                out.push(c);
            }
            '-' if chars.peek() == Some(&'-') => {
                out.push(c);
                out.push(chars.next().unwrap_or('-'));
                in_line_comment = true;
            }
            '/' if chars.peek() == Some(&'*') => {
                out.push(c);
                out.push(chars.next().unwrap_or('*'));
                in_block_comment = true;
            }
            '$' => {
                // 尝试解析 dollar-quote 开始定界符（`$$` 或 `$tag$`）
                if let Some(rest) = pg_peek_dollar_quote_rest(&chars) {
                    out.push('$');
                    for ch in rest.chars() {
                        out.push(ch);
                        chars.next();
                    }
                    dollar_tag = Some(rest[..rest.len() - 1].to_string());
                } else if chars.peek().is_some_and(|next| next.is_ascii_digit()) {
                    // 字符串外的 `$数字` 是 PG 原生占位符：整体不再重写任何 `?`
                    saw_native = true;
                    out.push(c);
                } else {
                    out.push(c);
                }
            }
            '?' if prev != '@'
                && !matches!(chars.peek(), Some('|') | Some('&'))
                && !pg_question_is_json_operator(&chars) =>
            {
                n += 1;
                out.push('$');
                out.push_str(&n.to_string());
            }
            _ => out.push(c),
        }
        prev = c;
    }
    if saw_native {
        return std::borrow::Cow::Borrowed(sql);
    }
    std::borrow::Cow::Owned(out)
}

/// 判断当前 `?` 是否 PostgreSQL JSON 存在运算符（如 `doc ? 'k'`）而非占位符。
///
/// JSON 运算符右侧必然是字符串字面量/标识符，占位符右侧不会是引号。
fn pg_question_is_json_operator(chars: &std::iter::Peekable<std::str::Chars<'_>>) -> bool {
    let mut lookahead = chars.clone();
    while let Some(&c) = lookahead.peek() {
        match c {
            ' ' | '\t' => {
                lookahead.next();
            }
            '\'' | '"' => return true,
            _ => return false,
        }
    }
    false
}

/// 解析 dollar-quote 开始定界符：`chars` 指向当前 `$` 之后的字符。
///
/// 返回 `Some(rest)` 表示命中，`rest` 为 `$` 之后的定界符片段：
/// - `$$` → `Some("$")`（空 tag）
/// - `$tag$` → `Some("tag$")`
///
/// 返回 `None` 表示不是 dollar-quote（如 `$1` 原生占位符、普通 `$`）。
fn pg_peek_dollar_quote_rest(chars: &std::iter::Peekable<std::str::Chars<'_>>) -> Option<String> {
    let mut lookahead = chars.clone();
    if lookahead.peek() == Some(&'$') {
        return Some("$".to_string());
    }
    let first = *lookahead.peek()?;
    if !(first.is_ascii_alphabetic() || first == '_') {
        return None;
    }
    let mut rest = String::new();
    while let Some(&c) = lookahead.peek() {
        match c {
            '$' => {
                rest.push('$');
                return Some(rest);
            }
            c if c.is_ascii_alphanumeric() || c == '_' => {
                rest.push(c);
                lookahead.next();
            }
            _ => return None,
        }
    }
    None
}

/// 在 dollar-quote 内部遇到 `$` 时，检查是否为 `$tag$` 结束定界符。
///
/// `chars` 指向当前 `$` 之后的字符；返回需要额外消费的字符数（tag 字符加结束 `$`），
/// 不是结束定界符时返回 `None`。
fn pg_try_close_dollar_quote(
    chars: &std::iter::Peekable<std::str::Chars<'_>>,
    tag: &str,
) -> Option<usize> {
    let mut lookahead = chars.clone();
    for expected in tag.chars() {
        if lookahead.next() != Some(expected) {
            return None;
        }
    }
    if lookahead.next() == Some('$') {
        Some(tag.chars().count() + 1)
    } else {
        None
    }
}

/// PostgreSQL 参数化查询语句类型别名
type PgQuery<'q> = sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>;

/// 按 `Value` 变体绑定 PostgreSQL 参数
///
/// 字符串类变体优先转换为对应的原生类型（DECIMAL/DATE/TIMESTAMP/TIME/UUID/JSONB），
/// 转换失败时回退为 TEXT 绑定，避免参数类型不匹配导致整条语句失败。
fn bind_pg_value<'q>(q: PgQuery<'q>, v: &'q Value) -> PgQuery<'q> {
    match v {
        Value::Null => q.bind(None::<i64>),
        Value::Bool(b) => q.bind(*b),
        Value::I8(n) => q.bind(*n),
        Value::I16(n) => q.bind(*n),
        Value::I32(n) => q.bind(*n),
        Value::I64(n) => q.bind(*n),
        // PostgreSQL 无 u8/u16/u32/u64：按最小可容纳有符号类型转换
        Value::U8(n) => q.bind(*n as i16),
        Value::U16(n) => q.bind(*n as i32),
        Value::U32(n) => q.bind(*n as i64),
        Value::U64(n) => q.bind(*n as i64),
        Value::F32(f) => q.bind(*f),
        Value::F64(f) => q.bind(*f),
        Value::String(s) => q.bind(s.as_str()),
        // R2：以任意精度 BigDecimal 绑定，避免 38 位 DECIMAL 被 rust_decimal 截断
        Value::Decimal(s) => match s.parse::<bigdecimal::BigDecimal>() {
            Ok(d) => q.bind(d),
            Err(_) => q.bind(s.as_str()),
        },
        Value::Uuid(s) => match sqlx::types::Uuid::parse_str(s) {
            Ok(u) => q.bind(u),
            Err(_) => q.bind(s.as_str()),
        },
        Value::Date(s) => match parse_pg_date(s) {
            Some(d) => q.bind(d),
            None => q.bind(s.as_str()),
        },
        Value::DateTime(s) => match parse_pg_datetime(s) {
            // 带时区偏移的输入必须绑定为 TIMESTAMPTZ，否则 PG 会按会话时区解释
            Some(PgDateTime::Zoned(d)) => q.bind(d),
            Some(PgDateTime::Naive(d)) => q.bind(d),
            None => q.bind(s.as_str()),
        },
        Value::Time(s) => match parse_pg_time(s) {
            Some(t) => q.bind(t),
            None => q.bind(s.as_str()),
        },
        Value::Json(s) => match serde_json::from_str::<serde_json::Value>(s) {
            Ok(j) => q.bind(j),
            Err(_) => q.bind(s.as_str()),
        },
        Value::Bytes(b) => q.bind(b.as_slice()),
        other => q.bind(other.to_string()),
    }
}

/// 解析 DATE：`YYYY-MM-DD`
fn parse_pg_date(s: &str) -> Option<sqlx::types::chrono::NaiveDate> {
    use sqlx::types::chrono::NaiveDate;
    let head = s.get(..10)?;
    NaiveDate::parse_from_str(head, "%Y-%m-%d").ok()
}

/// TIMESTAMP/TIMESTAMPTZ 的绑定形态
enum PgDateTime {
    /// 无时区：绑定 TIMESTAMP
    Naive(sqlx::types::chrono::NaiveDateTime),
    /// 带时区偏移：绑定 TIMESTAMPTZ（UTC 时刻）
    Zoned(sqlx::types::chrono::DateTime<sqlx::types::chrono::Utc>),
}

/// 解析 TIMESTAMP/TIMESTAMPTZ：
/// - 带 `Z` 或 ±HH:MM 偏移 → [`PgDateTime::Zoned`]
/// - 其余 → [`PgDateTime::Naive`]
fn parse_pg_datetime(s: &str) -> Option<PgDateTime> {
    use sqlx::types::chrono::{DateTime, NaiveDateTime, Utc};
    let has_offset = s.ends_with('Z')
        || s.ends_with('z')
        || s.get(10..).is_some_and(|rest| {
            rest.char_indices().any(|(i, c)| {
                (c == '+' || c == '-')
                    && rest[..i].chars().any(|p| p == ':')
                    // 排除日期部分本身的 '-'（位置 4/7 已在 get(10..) 之外）
                    && i > 0
            })
        });
    if has_offset {
        if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
            return Some(PgDateTime::Zoned(dt.with_timezone(&Utc)));
        }
        for fmt in [
            "%Y-%m-%d %H:%M:%S%.f%:z",
            "%Y-%m-%d %H:%M:%S%:z",
            "%Y-%m-%dT%H:%M:%S%.f%:z",
            "%Y-%m-%dT%H:%M:%S%:z",
            "%Y-%m-%d %H:%M:%S%.f%z",
        ] {
            if let Ok(dt) = DateTime::parse_from_str(s, fmt) {
                return Some(PgDateTime::Zoned(dt.with_timezone(&Utc)));
            }
        }
    }
    for fmt in ["%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"] {
        if let Ok(dt) = NaiveDateTime::parse_from_str(s, fmt) {
            return Some(PgDateTime::Naive(dt));
        }
        let with_frac = format!("{fmt}%.f");
        if let Ok(dt) = NaiveDateTime::parse_from_str(s, &with_frac) {
            return Some(PgDateTime::Naive(dt));
        }
    }
    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
        return Some(PgDateTime::Zoned(dt.with_timezone(&Utc)));
    }
    None
}

/// 解析 TIME：支持 `HH:MM:SS` 与 `HH:MM:SS.f`
fn parse_pg_time(s: &str) -> Option<sqlx::types::chrono::NaiveTime> {
    use sqlx::types::chrono::NaiveTime;
    NaiveTime::parse_from_str(s, "%H:%M:%S")
        .or_else(|_| NaiveTime::parse_from_str(s, "%H:%M:%S%.f"))
        .ok()
}

pub struct PgPoolHandle {
    pool: sqlx::PgPool,
}

impl PgPoolHandle {
    pub async fn connect(url: &str) -> Result<Self, DbError> {
        // 默认容量 10（向后兼容）；上层应用 `connect_with_max` 对齐外层 PoolConfig::max_size
        Self::connect_with_max(url, 10).await
    }

    /// 容量显式版 connect（2026-10-28 P3-4，PERFV2-03）：与外层 `PoolConfig::max_size` 对齐，
    /// 避免"外层按 max_size 记账、内层硬编码 10"的双层上限不一致（详见 SqlitePoolHandle::connect_with_max）。
    pub async fn connect_with_max(url: &str, max_connections: u32) -> Result<Self, DbError> {
        let pool = sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
            .max_connections(max_connections)
            .acquire_timeout(std::time::Duration::from_secs(30))
            .idle_timeout(Some(std::time::Duration::from_secs(600)))
            .max_lifetime(Some(std::time::Duration::from_secs(1800)))
            .connect(url)
            .await
            .map_err(map_sqlx_error)?;
        Ok(Self { pool })
    }

    pub fn from_pool(pool: sqlx::PgPool) -> Self {
        Self { pool }
    }

    pub fn pool(&self) -> &sqlx::PgPool {
        &self.pool
    }
}

pub struct SqlxPgConnectionFactory {
    pool: Arc<PgPoolHandle>,
}

impl SqlxPgConnectionFactory {
    pub fn new(pool: Arc<PgPoolHandle>) -> Self {
        Self { pool }
    }
}

#[async_trait]
impl ConnectionFactory for SqlxPgConnectionFactory {
    async fn create(&self) -> Result<Box<dyn Connection>, DbError> {
        let conn = self.pool.pool.acquire().await.map_err(map_sqlx_error)?;
        // 2026-10-28 P1-3：MetricsConn 包装统一上报查询指标（ops/latency/errors）
        Ok(Box::new(crate::metrics_conn::MetricsConn::new(Box::new(
            SqlxPgConnection {
                conn: Some(conn),
                connected: true,
                in_transaction: false,
            },
        ))))
    }
}

pub struct SqlxPgConnection {
    conn: Option<sqlx::pool::PoolConnection<sqlx::Postgres>>,
    connected: bool,
    in_transaction: bool,
}

impl Connection for SqlxPgConnection {
    fn db_type(&self) -> Option<DbType> {
        Some(DbType::PostgreSQL)
    }

    /// PostgreSQL 的 `DISCARD ALL` 复位全部会话状态：`SET` 参数、
    /// `search_path`、临时表、预备语句、游标、监听频道、序列缓存。
    /// 必须在事务外执行（连接池已在复位前回滚未提交事务）。
    fn session_reset_sql(&self) -> Option<&'static str> {
        Some("DISCARD ALL")
    }

    fn execute<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let result = if needs_raw_sql(sql) {
                (&mut *pool_conn)
                    .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
                    .await
            } else {
                (&mut *pool_conn).execute(sqlx::AssertSqlSafe(sql)).await
            };
            self.conn = Some(pool_conn);

            match result {
                Ok(r) => Ok(r.rows_affected()),
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    fn query<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, DbError>> + Send + 'a>>
    {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            // sqlx 0.9: PoolConnection 不再实现 Executor，需通过 DerefMut 解引用到内部连接
            // sqlx 0.9: SqlSafeStr 只对 &'static str 直接实现，非 'static 的 &str 需用 AssertSqlSafe 包装
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);

            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            // 预解析列类型（只在第一行解析，后续行复用，避免每行字符串 match）
            // PostgreSQL 专用解析：覆盖 PG 内部类型名（INT4/INT8/FLOAT8/BPCHAR/JSONB 等）
            let col_types: Vec<ColType> = rows[0]
                .columns()
                .iter()
                .map(|col| ColType::parse_postgres(col.type_info().name()))
                .collect();
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_types.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_pg(row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    fn begin_transaction<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                return Err(DbError::Internal("transaction already started".to_string()));
            }
            self.execute("BEGIN").await?;
            self.in_transaction = true;
            Ok(())
        })
    }

    fn commit<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                self.execute("COMMIT").await?;
                self.in_transaction = false;
            }
            Ok(())
        })
    }

    fn rollback<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if self.in_transaction {
                // 回滚失败时保留事务标志：非连接类错误下服务端事务可能仍开启，
                // 若此刻清标志，连接带着开放事务回池将绕过归还路径的自动回滚
                // （事务泄漏）。失败后由归还路径重试回滚或直接关闭连接。
                // 语义与 Oracle/MSSQL 适配器一致。
                let result = self.execute("ROLLBACK").await;
                if result.is_ok() {
                    self.in_transaction = false;
                }
                result.map(|_| ())
            } else {
                Ok(())
            }
        })
    }

    fn is_connected(&self) -> bool {
        // 纳入 conn.is_some()：query_stream 提前弃读时 PoolConnection 已从包装器
        // take() 出去（self.conn = None），毒化状态不得被判定为可用连接
        // （否则回池后下次 acquire 借出即报 "connection already closed"）
        self.connected && self.conn.is_some()
    }

    fn in_transaction(&self) -> bool {
        self.in_transaction
    }

    fn ping<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = bool> + Send + 'a>> {
        Box::pin(async move {
            match self.execute("SELECT 1").await {
                Ok(_) => true,
                Err(_) => {
                    self.connected = false;
                    false
                }
            }
        })
    }

    fn close<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), DbError>> + Send + 'a>> {
        Box::pin(async move {
            if let Some(conn) = self.conn.take() {
                drop(conn);
            }
            self.connected = false;
            self.in_transaction = false;
            Ok(())
        })
    }

    /// PostgreSQL 参数绑定执行（INSERT/UPDATE/DELETE）
    ///
    /// 使用 sqlx prepared statement 绑定参数。PostgreSQL 协议使用 `$1, $2, ...` 占位符，
    /// 调用方传入的 `?` 占位符会由 `pg_rewrite_placeholders` 自动改写；
    /// 已使用 `$n` 的 SQL 不会被二次改写。
    ///
    /// # 类型映射说明
    ///
    /// PostgreSQL 不支持无符号整数类型（u8/u16/u32/u64），因此无符号 Value 绑定
    /// 时按"最小可容纳的有符号类型"转换：
    /// - `U8`  → `i16`（PostgreSQL `SMALLINT`）
    /// - `U16` → `i32`（PostgreSQL `INTEGER`）
    /// - `U32` → `i64`（PostgreSQL `BIGINT`）
    /// - `U64` → `i64`（可能截断，仅适用于 < `i64::MAX` 的值）
    ///
    /// `Decimal`/`Uuid`/`Date`/`DateTime`/`Time`/`Json` 会按各自原生类型绑定，
    /// 解析失败时回退为 TEXT 绑定；未知类型回退为 `to_string()` 后以 TEXT 绑定。
    fn execute_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<u64, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if needs_raw_sql(sql) || params.is_empty() {
                return self.execute(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rewritten = pg_rewrite_placeholders(sql);
            let mut q = sqlx::query(sqlx::AssertSqlSafe(rewritten.as_ref()));
            for v in params {
                q = bind_pg_value(q, v);
            }
            let result = q.execute(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            match result {
                Ok(r) => Ok(r.rows_affected()),
                Err(e) => {
                    let db_err = map_sqlx_error(e);
                    if matches!(db_err, DbError::ConnectionError(_) | DbError::IoError(_)) {
                        self.connected = false;
                    }
                    Err(db_err)
                }
            }
        })
    }

    /// PostgreSQL 参数绑定查询（SELECT，HashMap 映射）
    ///
    /// 类型映射规则与 [`Self::execute_with_params`] 相同。
    fn query_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryRows, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rewritten = pg_rewrite_placeholders(sql);
            let mut q = sqlx::query(sqlx::AssertSqlSafe(rewritten.as_ref()));
            for v in params {
                q = bind_pg_value(q, v);
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok(Vec::new());
            }
            // 预解析列类型（只在第一行解析，后续行复用，避免每行字符串 match）
            let col_types: Vec<ColType> = rows[0]
                .columns()
                .iter()
                .map(|col| ColType::parse_postgres(col.type_info().name()))
                .collect();
            let col_names: Vec<String> = rows[0]
                .columns()
                .iter()
                .map(|col| col.name().to_string())
                .collect();
            let mut result = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut record = HashMap::with_capacity(col_types.len());
                for (i, name) in col_names.iter().enumerate() {
                    let value = row_to_value_with_coltype_pg(row, i, col_types[i]);
                    record.insert(name.clone(), value);
                }
                result.push(record);
            }
            Ok(result)
        })
    }

    /// PostgreSQL 位置式查询（SELECT，无参数）
    ///
    /// 绕过 HashMap 行映射，返回 `(列名, 按列序号的值矩阵)`。适用于 SELECT ALL 大结果集。
    fn query_values<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rows_result = (&mut *pool_conn).fetch_all(sqlx::AssertSqlSafe(sql)).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            let mut col_types: Vec<ColType> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
                col_types.push(ColType::parse_postgres(col.type_info().name()));
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in &rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    row_values.push(row_to_value_with_coltype_pg(row, idx, col_types[idx]));
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// PostgreSQL 参数绑定位置式查询（SELECT）
    ///
    /// 类型映射规则与 [`Self::execute_with_params`] 相同。
    fn query_values_with_params<'a>(
        &'a mut self,
        sql: &'a str,
        params: &'a [Value],
    ) -> Pin<Box<dyn Future<Output = Result<QueryValues, DbError>> + Send + 'a>> {
        Box::pin(async move {
            if params.is_empty() {
                return self.query_values(sql).await;
            }
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let rewritten = pg_rewrite_placeholders(sql);
            let mut q = sqlx::query(sqlx::AssertSqlSafe(rewritten.as_ref()));
            for v in params {
                q = bind_pg_value(q, v);
            }
            let rows_result = q.fetch_all(&mut *pool_conn).await;
            self.conn = Some(pool_conn);
            let rows = rows_result.map_err(map_sqlx_error)?;
            if rows.is_empty() {
                return Ok((Vec::new(), Vec::new()));
            }
            let cols = rows[0].columns();
            let mut col_names: Vec<String> = Vec::with_capacity(cols.len());
            for col in cols {
                col_names.push(col.name().to_string());
            }
            let mut result_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
            for row in rows {
                let mut row_values: Vec<Value> = Vec::with_capacity(col_names.len());
                for (idx, _) in col_names.iter().enumerate() {
                    let ordinal = row.columns()[idx].ordinal();
                    row_values.push(row_to_value_pg(&row, ordinal)?);
                }
                result_rows.push(row_values);
            }
            Ok((col_names, result_rows))
        })
    }

    /// PostgreSQL 流式查询：逐行返回结果，避免大结果集 `fetch_all` 内存峰值
    ///
    /// 使用 `sqlx::query::fetch` 获取行流，逐行映射为 `HashMap<String, Value>`。
    /// 流正常消费完毕后连接归还 `self.conn`；若提前 drop 流，连接通过
    /// `PoolConnection::drop` 归还到 sqlx 池，但 `self.conn` 变为 `None`。
    ///
    /// 注意：PostgreSQL 的 `fetch` 返回 `Pin<Box<dyn Stream<...>>>`（boxed trait object），
    /// 其析构函数可能持有 `pool_conn` 的借用，导致无法直接 `self.conn = Some(pool_conn)`。
    /// 通过显式 `drop(row_stream)` 提前结束借用，再归还连接。
    fn query_stream<'a>(
        &'a mut self,
        sql: &'a str,
    ) -> Pin<Box<dyn futures::Stream<Item = Result<HashMap<String, Value>, DbError>> + Send + 'a>>
    {
        Box::pin(async_stream::try_stream! {
            let mut pool_conn = self
                .conn
                .take()
                .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
            let mut row_stream = sqlx::query(sqlx::AssertSqlSafe(sql)).fetch(&mut *pool_conn);
            // 流式查询无法预解析第一行列类型（流首行未知），使用每行 columns() 长度预分配 HashMap
            // 仍然使用 row_to_value_pg（按列类型字符串 match）以保证正确性
            while let Some(row_result) = row_stream.next().await {
                let row = row_result.map_err(map_sqlx_error)?;
                let cols = row.columns();
                let mut record = HashMap::with_capacity(cols.len());
                for (i, col) in cols.iter().enumerate() {
                    let name = col.name().to_string();
                    // 流式场景下每行都解析 ColType 反而增加开销，直接使用 row_to_value_pg
                    let value = row_to_value_pg(&row, i)?;
                    record.insert(name, value);
                }
                yield record;
            }
            // 显式 drop row_stream 以释放对 pool_conn 的借用
            // PostgreSQL 的 fetch 流是 boxed trait object，析构函数可能持有借用
            drop(row_stream);
            self.conn = Some(pool_conn);
        })
    }
}

impl Drop for SqlxPgConnection {
    fn drop(&mut self) {
        if let Some(conn) = self.conn.take() {
            drop(conn);
        }
    }
}

// ===================== PostgreSQL 扩展功能 =====================

/// PostgreSQL 扩展功能 trait
///
/// 提供 PostgreSQL 特有的 LISTEN/NOTIFY 通道通信与 COPY FROM STDIN 批量导入功能。
/// 这些功能在其他数据库（MySQL/SQLite）中没有对应实现，因此单独定义为扩展 trait，
/// 仅由 [`SqlxPgConnection`] 实现。
///
/// # LISTEN/NOTIFY
///
/// PostgreSQL 的轻量级进程间通信机制：
/// - `LISTEN channel`：订阅通道
/// - `NOTIFY channel, payload`：向通道发送通知
///
/// 接收通知需配合 `PgListener` 或轮询 `pg_notification` 系列函数。
///
/// # COPY FROM STDIN
///
/// PostgreSQL 的高性能批量导入协议，比逐行 INSERT 快 10~100 倍。
/// 数据通过专用协议流式传输，绕过 SQL 解析器。
#[async_trait]
pub trait PgExtensions: Send + Sync {
    /// LISTEN 通道：订阅指定通道的通知
    ///
    /// # 参数
    ///
    /// - `channel`: 通道名（仅允许字母、数字、下划线，防止 SQL 注入）
    ///
    /// # 错误
    ///
    /// - 通道名包含非法字符时返回 `DbError::Internal`
    /// - 连接已关闭时返回 `DbError::Internal`
    async fn listen(&mut self, channel: &str) -> Result<(), DbError>;

    /// NOTIFY 通道：向指定通道发送通知
    ///
    /// # 参数
    ///
    /// - `channel`: 通道名（仅允许字母、数字、下划线）
    /// - `payload`: 通知载荷字符串（单引号自动转义）
    ///
    /// # 错误
    ///
    /// - 通道名包含非法字符时返回 `DbError::Internal`
    async fn notify(&mut self, channel: &str, payload: &str) -> Result<(), DbError>;

    /// COPY FROM STDIN：批量导入数据
    ///
    /// # 参数
    ///
    /// - `sql`: COPY 语句，如 `COPY mytable (col1, col2) FROM STDIN WITH (FORMAT csv, HEADER true)`
    /// - `data`: 完整的导入数据字节流
    ///
    /// # 返回
    ///
    /// 返回受影响的行数。
    ///
    /// # 性能
    ///
    /// 比逐行 INSERT 快 10~100 倍，适用于大批量数据导入（10 万行以上）。
    async fn copy_from_stdin(&mut self, sql: &str, data: &[u8]) -> Result<u64, DbError>;
}

/// 校验 PostgreSQL 通道名合法性（仅允许字母、数字、下划线）
///
/// LISTEN/NOTIFY 的通道名不能通过参数化绑定，必须拼接 SQL，
/// 因此需严格校验防止 SQL 注入。
fn validate_pg_channel_name(channel: &str) -> Result<(), DbError> {
    if channel.is_empty() {
        return Err(DbError::Internal(
            "PG channel name must not be empty".to_string(),
        ));
    }
    if !channel
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_')
    {
        return Err(DbError::Internal(format!(
            "invalid PG channel name: {} (only alphanumeric and underscore allowed)",
            channel
        )));
    }
    Ok(())
}

#[async_trait]
impl PgExtensions for SqlxPgConnection {
    async fn listen(&mut self, channel: &str) -> Result<(), DbError> {
        validate_pg_channel_name(channel)?;
        // LISTEN 是简单 SQL 命令，channel 已校验，可安全拼接
        self.execute(&format!("LISTEN {}", channel)).await?;
        Ok(())
    }

    async fn notify(&mut self, channel: &str, payload: &str) -> Result<(), DbError> {
        validate_pg_channel_name(channel)?;
        // payload 通过 PostgreSQL 字符串字面量转义（单引号加倍）防止注入
        let escaped_payload = payload.replace('\'', "''");
        self.execute(&format!("NOTIFY {}, '{}'", channel, escaped_payload))
            .await?;
        Ok(())
    }

    async fn copy_from_stdin(&mut self, sql: &str, data: &[u8]) -> Result<u64, DbError> {
        let mut pool_conn = self
            .conn
            .take()
            .ok_or_else(|| DbError::Internal("connection already closed".to_string()))?;
        // sqlx 0.9: 通过 DerefMut 访问 PgConnection 的 copy_in_raw 方法
        // 启动 COPY 协议，返回 PgCopyIn 流式写入句柄
        // 注意：直接使用 *pool_conn 解引用，避免 clippy::needless_borrow 警告
        let mut copy = (*pool_conn)
            .copy_in_raw(sql)
            .await
            .map_err(map_sqlx_error)?;
        // 一次性发送全部数据（非流式，适用于数据可完整载入内存的场景）
        copy.send(data).await.map_err(map_sqlx_error)?;
        // 完成 COPY 操作，sqlx 0.9 的 PgCopyIn::finish() 直接返回 u64（受影响行数）
        let result = copy.finish().await.map_err(map_sqlx_error)?;
        self.conn = Some(pool_conn);
        Ok(result)
    }
}

/// PostgreSQL 批量导入（多行 INSERT，作为 COPY 协议的简化替代）
///
/// 构建多行 INSERT 语句：`INSERT INTO t (c1, c2) VALUES ($1, $2), ($3, $4), ...`
/// 使用参数绑定避免 SQL 注入。PostgreSQL 占位符 `$N` 跨行递增。
///
/// 当数据量极大（>10 万行）时，建议分批调用或使用
/// [`PgExtensions::copy_from_stdin`] 走 COPY 协议以获得更高吞吐。
pub async fn pg_bulk_insert(
    conn: &mut SqlxPgConnection,
    table: &str,
    columns: &[&str],
    rows: &[Vec<Value>],
) -> Result<u64, DbError> {
    if rows.is_empty() {
        return Ok(0);
    }
    let col_list = columns.join(", ");
    let cols_per_row = columns.len();
    // PostgreSQL 占位符 $1, $2, ... 跨行递增
    let placeholders: Vec<String> = rows
        .iter()
        .enumerate()
        .map(|(row_idx, _)| {
            let base = row_idx * cols_per_row;
            let ph: Vec<String> = (0..cols_per_row)
                .map(|i| format!("${}", base + i + 1))
                .collect();
            format!("({})", ph.join(", "))
        })
        .collect();
    let sql = format!(
        "INSERT INTO {} ({}) VALUES {}",
        table,
        col_list,
        placeholders.join(", ")
    );
    // 展平所有行的值为单一参数数组
    let mut params: Vec<Value> = Vec::with_capacity(rows.len() * cols_per_row);
    for row in rows {
        for v in row {
            params.push(v.clone());
        }
    }
    conn.execute_with_params(&sql, &params).await
}
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_pg_rewrite_placeholders_basic() {
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE a = ? AND b = ?"),
            "SELECT * FROM t WHERE a = $1 AND b = $2"
        );
    }

    #[test]
    fn test_pg_rewrite_placeholders_skips_native() {
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE a = $1"),
            "SELECT * FROM t WHERE a = $1"
        );
    }

    #[test]
    fn test_pg_rewrite_placeholders_skips_literals_and_comments() {
        assert_eq!(
            pg_rewrite_placeholders("SELECT '?' AS q -- ?\nFROM t WHERE id = ?"),
            "SELECT '?' AS q -- ?\nFROM t WHERE id = $1"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT /* ? */ \"a?\" FROM t WHERE id = ?"),
            "SELECT /* ? */ \"a?\" FROM t WHERE id = $1"
        );
    }

    #[test]
    fn test_pg_rewrite_placeholders_keeps_json_operators() {
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE doc ? 'k' AND b = ?"),
            "SELECT * FROM t WHERE doc ? 'k' AND b = $1"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE doc ?| array['a'] AND b = ?"),
            "SELECT * FROM t WHERE doc ?| array['a'] AND b = $1"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE doc ?& array['a', 'b'] AND b = ?"),
            "SELECT * FROM t WHERE doc ?& array['a', 'b'] AND b = $1"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE doc @? '$.a' AND b = ?"),
            "SELECT * FROM t WHERE doc @? '$.a' AND b = $1"
        );
    }

    #[test]
    fn test_pg_rewrite_placeholders_dollar_in_literal_still_rewrites() {
        // 字符串字面量里的 `$1` 不是原生占位符，真实的 `?` 必须被重写
        assert_eq!(
            pg_rewrite_placeholders("SELECT '$1' AS lit, ? AS val"),
            "SELECT '$1' AS lit, $1 AS val"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT 'price: $2' AS note FROM t WHERE id = ? AND code = ?"),
            "SELECT 'price: $2' AS note FROM t WHERE id = $1 AND code = $2"
        );
    }

    #[test]
    fn test_pg_rewrite_placeholders_skips_dollar_quoted_strings() {
        // dollar-quote 内的 `?` 是字面量，不得改写
        assert_eq!(
            pg_rewrite_placeholders("SELECT $$?$$ AS lit, ? AS val"),
            "SELECT $$?$$ AS lit, $1 AS val"
        );
        assert_eq!(
            pg_rewrite_placeholders("SELECT $tag$a ? b$tag$ AS lit, ? AS val"),
            "SELECT $tag$a ? b$tag$ AS lit, $1 AS val"
        );
        // 函数体（plpgsql）使用 dollar-quote，体内的 `?` 保持原样
        let func = "CREATE FUNCTION f() RETURNS int AS $$ BEGIN RETURN (SELECT count(*) FROM t WHERE x = ?); END; $$ LANGUAGE plpgsql";
        assert_eq!(pg_rewrite_placeholders(func), func);
    }

    #[test]
    fn test_pg_rewrite_placeholders_native_skips_even_with_question() {
        // 字符串外存在真实 `$n` 时整体不重写，`?` 保持原样（避免编号错乱）
        assert_eq!(
            pg_rewrite_placeholders("SELECT * FROM t WHERE a = $1 AND doc ? 'k'"),
            "SELECT * FROM t WHERE a = $1 AND doc ? 'k'"
        );
    }

    #[test]
    fn test_parse_pg_date_and_datetime() {
        assert!(parse_pg_date("2026-10-28").is_some());
        assert!(parse_pg_date("2026-10-28 12:00:00").is_some());
        assert!(parse_pg_date("not-a-date").is_none());
        assert!(parse_pg_datetime("2026-10-28 12:00:00").is_some());
        assert!(parse_pg_datetime("2026-10-28T12:00:00").is_some());
        assert!(parse_pg_datetime("2026-10-28T12:00:00Z").is_some());
        assert!(parse_pg_datetime("bad").is_none());
    }

    #[test]
    fn test_needs_raw_sql_begin() {
        assert!(needs_raw_sql("BEGIN"));
        assert!(needs_raw_sql("begin"));
        assert!(needs_raw_sql("  BEGIN"));
        assert!(needs_raw_sql("\tbegin"));
    }

    #[test]
    fn test_needs_raw_sql_commit() {
        assert!(needs_raw_sql("COMMIT"));
        assert!(needs_raw_sql("commit"));
        assert!(needs_raw_sql(" COMMIT WORK"));
    }

    #[test]
    fn test_needs_raw_sql_rollback() {
        assert!(needs_raw_sql("ROLLBACK"));
        assert!(needs_raw_sql("rollback"));
        assert!(needs_raw_sql(" ROLLBACK TO SAVEPOINT sp1"));
    }

    #[test]
    fn test_needs_raw_sql_savepoint() {
        assert!(needs_raw_sql("SAVEPOINT sp1"));
        assert!(needs_raw_sql("savepoint s1"));
    }

    #[test]
    fn test_needs_raw_sql_release() {
        assert!(needs_raw_sql("RELEASE sp1"));
        assert!(needs_raw_sql("release savepoint s1"));
    }

    #[test]
    fn test_needs_raw_sql_set() {
        assert!(needs_raw_sql("SET autocommit = 0"));
        assert!(needs_raw_sql("set names utf8"));
    }

    #[test]
    fn test_needs_raw_sql_use() {
        assert!(needs_raw_sql("USE mydb"));
        assert!(needs_raw_sql("use mydb"));
    }

    #[test]
    fn test_needs_raw_sql_start_transaction() {
        assert!(needs_raw_sql("START TRANSACTION"));
        assert!(needs_raw_sql("start transaction"));
        assert!(needs_raw_sql("  START TRANSACTION READ ONLY"));
    }

    #[test]
    fn test_needs_raw_sql_normal_queries() {
        assert!(!needs_raw_sql("SELECT * FROM users"));
        assert!(!needs_raw_sql("INSERT INTO t VALUES (1)"));
        assert!(!needs_raw_sql("UPDATE t SET x = 1"));
        assert!(!needs_raw_sql("DELETE FROM t"));
        assert!(!needs_raw_sql("CREATE TABLE t (id INT)"));
        assert!(!needs_raw_sql("DROP TABLE t"));
        assert!(!needs_raw_sql("ALTER TABLE t ADD COLUMN x INT"));
    }

    #[test]
    fn test_needs_raw_sql_edge_cases() {
        assert!(!needs_raw_sql(""));
        assert!(!needs_raw_sql("   "));
        assert!(!needs_raw_sql("SELECT"));
        assert!(needs_raw_sql("BEGINNING"));
        assert!(needs_raw_sql("COMMITTEE"));
        assert!(!needs_raw_sql("INSERT"));
        assert!(!needs_raw_sql("DELETE"));
    }

    // ---- v9.2.0 M18: validate_pg_channel_name 纯字符串校验 ----

    #[test]
    fn test_validate_pg_channel_name_valid() {
        assert!(validate_pg_channel_name("my_channel").is_ok());
        assert!(validate_pg_channel_name("abc123").is_ok());
        assert!(validate_pg_channel_name("_test").is_ok());
        assert!(validate_pg_channel_name("a_b_c").is_ok());
    }

    #[test]
    fn test_validate_pg_channel_name_empty() {
        let err = validate_pg_channel_name("").unwrap_err();
        assert!(matches!(err, DbError::Internal(ref s) if s.contains("empty")));
    }

    #[test]
    fn test_validate_pg_channel_name_invalid_chars() {
        let err = validate_pg_channel_name("my-channel").unwrap_err();
        assert!(matches!(err, DbError::Internal(ref s) if s.contains("invalid")));
        assert!(validate_pg_channel_name("my.channel").is_err());
        assert!(validate_pg_channel_name("my channel").is_err());
        assert!(validate_pg_channel_name("my;channel").is_err());
        assert!(validate_pg_channel_name("my'channel").is_err());
    }

    // ---- v9.2.0 M18: SqlxSqliteConnection conn=None 错误分支 ----

    fn closed_sqlite() -> SqlxSqliteConnection {
        SqlxSqliteConnection {
            conn: None,
            connected: false,
            in_transaction: false,
        }
    }

    #[tokio::test]
    async fn test_sqlite_execute_closed() {
        let mut c = closed_sqlite();
        let result = c.execute("SELECT 1").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("closed"));
    }

    #[tokio::test]
    async fn test_sqlite_query_closed() {
        let mut c = closed_sqlite();
        let result = c.query("SELECT 1").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_begin_tx_already_started() {
        let mut c = SqlxSqliteConnection {
            conn: None,
            connected: false,
            in_transaction: true,
        };
        let result = c.begin_transaction().await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("already started"));
    }

    #[tokio::test]
    async fn test_sqlite_begin_tx_closed() {
        let mut c = closed_sqlite();
        let result = c.begin_transaction().await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_commit_no_tx() {
        let mut c = closed_sqlite();
        let result = c.commit().await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_sqlite_rollback_no_tx() {
        let mut c = closed_sqlite();
        let result = c.rollback().await;
        assert!(result.is_ok());
    }

    #[test]
    fn test_sqlite_is_connected_false_when_conn_none() {
        let c = closed_sqlite();
        assert!(!c.is_connected());
    }

    #[test]
    fn test_sqlite_is_connected_false_when_connected_false() {
        let c = SqlxSqliteConnection {
            conn: None,
            connected: true,
            in_transaction: false,
        };
        assert!(!c.is_connected());
    }

    #[test]
    fn test_sqlite_in_transaction_flag() {
        let c = SqlxSqliteConnection {
            conn: None,
            connected: false,
            in_transaction: true,
        };
        assert!(c.in_transaction());
        let c2 = closed_sqlite();
        assert!(!c2.in_transaction());
    }

    #[tokio::test]
    async fn test_sqlite_ping_closed() {
        let mut c = closed_sqlite();
        let result = c.ping().await;
        assert!(!result);
        assert!(!c.is_connected());
    }

    #[tokio::test]
    async fn test_sqlite_close_closed() {
        let mut c = closed_sqlite();
        let result = c.close().await;
        assert!(result.is_ok());
        assert!(!c.is_connected());
        assert!(!c.in_transaction());
    }

    #[tokio::test]
    async fn test_sqlite_execute_with_params_closed() {
        let mut c = closed_sqlite();
        let result = c
            .execute_with_params("INSERT INTO t VALUES (?)", &[Value::I64(1)])
            .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_execute_with_params_empty_delegates() {
        let mut c = closed_sqlite();
        let result = c.execute_with_params("INSERT INTO t VALUES (1)", &[]).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_query_with_params_empty_delegates() {
        let mut c = closed_sqlite();
        let result = c.query_with_params("SELECT * FROM t", &[]).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_query_with_params_closed() {
        let mut c = closed_sqlite();
        let result = c
            .query_with_params("SELECT * FROM t WHERE id = ?", &[Value::I64(1)])
            .await;
        assert!(result.is_err());
    }

    // ---- v9.2.0 M18: SqlxMySqlConnection conn=None 错误分支 ----

    fn closed_mysql() -> SqlxMySqlConnection {
        SqlxMySqlConnection {
            conn: None,
            connected: false,
            in_transaction: false,
            session_dirty: false,
        }
    }

    #[tokio::test]
    async fn test_mysql_execute_closed() {
        let mut c = closed_mysql();
        assert!(c.execute("SELECT 1").await.is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_closed() {
        let mut c = closed_mysql();
        assert!(c.query("SELECT 1").await.is_err());
    }

    #[test]
    fn test_mysql_is_connected_false() {
        let c = closed_mysql();
        assert!(!c.is_connected());
    }

    #[tokio::test]
    async fn test_mysql_close_closed() {
        let mut c = closed_mysql();
        assert!(c.close().await.is_ok());
        assert!(!c.is_connected());
    }

    #[tokio::test]
    async fn test_mysql_commit_no_tx() {
        let mut c = closed_mysql();
        assert!(c.commit().await.is_ok());
    }

    #[tokio::test]
    async fn test_mysql_rollback_no_tx() {
        let mut c = closed_mysql();
        assert!(c.rollback().await.is_ok());
    }

    #[tokio::test]
    async fn test_mysql_begin_tx_already_started() {
        let mut c = SqlxMySqlConnection {
            conn: None,
            connected: false,
            in_transaction: true,
            session_dirty: false,
        };
        let result = c.begin_transaction().await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_mysql_ping_closed() {
        let mut c = closed_mysql();
        assert!(!c.ping().await);
    }

    // ---- v9.2.0 M18: SqlxPgConnection conn=None 错误分支 ----

    fn closed_pg() -> SqlxPgConnection {
        SqlxPgConnection {
            conn: None,
            connected: false,
            in_transaction: false,
        }
    }

    #[tokio::test]
    async fn test_pg_execute_closed() {
        let mut c = closed_pg();
        assert!(c.execute("SELECT 1").await.is_err());
    }

    #[tokio::test]
    async fn test_pg_query_closed() {
        let mut c = closed_pg();
        assert!(c.query("SELECT 1").await.is_err());
    }

    #[test]
    fn test_pg_is_connected_false() {
        let c = closed_pg();
        assert!(!c.is_connected());
    }

    #[tokio::test]
    async fn test_pg_close_closed() {
        let mut c = closed_pg();
        assert!(c.close().await.is_ok());
        assert!(!c.is_connected());
    }

    #[tokio::test]
    async fn test_pg_commit_no_tx() {
        let mut c = closed_pg();
        assert!(c.commit().await.is_ok());
    }

    #[tokio::test]
    async fn test_pg_rollback_no_tx() {
        let mut c = closed_pg();
        assert!(c.rollback().await.is_ok());
    }

    #[tokio::test]
    async fn test_pg_begin_tx_already_started() {
        let mut c = SqlxPgConnection {
            conn: None,
            connected: false,
            in_transaction: true,
        };
        assert!(c.begin_transaction().await.is_err());
    }

    #[tokio::test]
    async fn test_pg_ping_closed() {
        let mut c = closed_pg();
        assert!(!c.ping().await);
    }

    // ---- v9.2.0 M18: MySQL 扩展方法 conn=None ----

    #[tokio::test]
    async fn test_mysql_execute_with_params_closed() {
        let mut c = closed_mysql();
        assert!(c
            .execute_with_params("INSERT INTO t VALUES (?)", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_mysql_execute_with_params_empty_delegates() {
        let mut c = closed_mysql();
        assert!(c
            .execute_with_params("INSERT INTO t VALUES (1)", &[])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_with_params_closed() {
        let mut c = closed_mysql();
        assert!(c
            .query_with_params("SELECT * FROM t WHERE id = ?", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_with_params_empty_delegates() {
        let mut c = closed_mysql();
        assert!(c.query_with_params("SELECT * FROM t", &[]).await.is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_values_closed() {
        let mut c = closed_mysql();
        assert!(c.query_values("SELECT * FROM t").await.is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_values_with_params_closed() {
        let mut c = closed_mysql();
        assert!(c
            .query_values_with_params("SELECT * FROM t WHERE id = ?", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_mysql_query_values_with_params_empty_delegates() {
        let mut c = closed_mysql();
        assert!(c
            .query_values_with_params("SELECT * FROM t", &[])
            .await
            .is_err());
    }

    // ---- v9.2.0 M18: PG 扩展方法 conn=None ----

    #[tokio::test]
    async fn test_pg_execute_with_params_closed() {
        let mut c = closed_pg();
        assert!(c
            .execute_with_params("INSERT INTO t VALUES ($1)", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_pg_execute_with_params_empty_delegates() {
        let mut c = closed_pg();
        assert!(c
            .execute_with_params("INSERT INTO t VALUES (1)", &[])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_pg_query_with_params_closed() {
        let mut c = closed_pg();
        assert!(c
            .query_with_params("SELECT * FROM t WHERE id = $1", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_pg_query_with_params_empty_delegates() {
        let mut c = closed_pg();
        assert!(c.query_with_params("SELECT * FROM t", &[]).await.is_err());
    }

    #[tokio::test]
    async fn test_pg_query_values_closed() {
        let mut c = closed_pg();
        assert!(c.query_values("SELECT * FROM t").await.is_err());
    }

    #[tokio::test]
    async fn test_pg_query_values_with_params_closed() {
        let mut c = closed_pg();
        assert!(c
            .query_values_with_params("SELECT * FROM t WHERE id = $1", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_pg_query_values_with_params_empty_delegates() {
        let mut c = closed_pg();
        assert!(c
            .query_values_with_params("SELECT * FROM t", &[])
            .await
            .is_err());
    }

    // ---- v9.2.0 M18: sqlite_backup conn=None ----

    #[tokio::test]
    async fn test_sqlite_backup_closed_conn() {
        let mut c = closed_sqlite();
        let result = sqlite_backup(&mut c, "/tmp/backup.db").await;
        assert!(result.is_err());
    }

    // ---- v9.2.0 M18: SQLite query_values conn=None ----

    #[tokio::test]
    async fn test_sqlite_query_values_closed() {
        let mut c = closed_sqlite();
        assert!(c.query_values("SELECT * FROM t").await.is_err());
    }

    #[tokio::test]
    async fn test_sqlite_query_values_with_params_closed() {
        let mut c = closed_sqlite();
        assert!(c
            .query_values_with_params("SELECT * FROM t WHERE id = ?", &[Value::I64(1)])
            .await
            .is_err());
    }

    #[tokio::test]
    async fn test_sqlite_query_values_with_params_empty_delegates() {
        let mut c = closed_sqlite();
        assert!(c
            .query_values_with_params("SELECT * FROM t", &[])
            .await
            .is_err());
    }

    // ---- v9.2.0 M18: bulk_insert 空行 + conn=None ----

    #[tokio::test]
    async fn test_mysql_bulk_insert_empty_rows() {
        let mut c = closed_mysql();
        let result = mysql_bulk_insert(&mut c, "t", &["a", "b"], &[]).await;
        assert_eq!(result.unwrap(), 0);
    }

    #[tokio::test]
    async fn test_mysql_bulk_insert_closed_conn() {
        let mut c = closed_mysql();
        let rows = vec![vec![Value::I64(1), Value::String("x".to_string())]];
        let result = mysql_bulk_insert(&mut c, "t", &["a", "b"], &rows).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_pg_bulk_insert_empty_rows() {
        let mut c = closed_pg();
        let result = pg_bulk_insert(&mut c, "t", &["a", "b"], &[]).await;
        assert_eq!(result.unwrap(), 0);
    }

    #[tokio::test]
    async fn test_pg_bulk_insert_closed_conn() {
        let mut c = closed_pg();
        let rows = vec![vec![Value::I64(1), Value::String("x".to_string())]];
        let result = pg_bulk_insert(&mut c, "t", &["a", "b"], &rows).await;
        assert!(result.is_err());
    }

    // ---- R1：TIMESTAMP/TIMESTAMPTZ 绑定解析 ----

    #[test]
    fn test_parse_pg_datetime_naive() {
        match parse_pg_datetime("2026-10-28 13:14:15.123456") {
            Some(PgDateTime::Naive(dt)) => {
                assert_eq!(
                    dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string(),
                    "2026-10-28 13:14:15.123456"
                );
            }
            _ => panic!("无偏移输入必须解析为 Naive"),
        }
    }

    #[test]
    fn test_parse_pg_datetime_zoned_forms() {
        // RFC3339 / 空格分隔 + 偏移 / 无冒号偏移 都必须识别为 Zoned
        for s in [
            "2026-10-28T13:14:15+08:00",
            "2026-10-28 13:14:15+08:00",
            "2026-10-28 13:14:15 +0800",
            "2026-10-28T05:14:15Z",
        ] {
            match parse_pg_datetime(s) {
                Some(PgDateTime::Zoned(dt)) => {
                    // 同一时刻的 UTC 表示
                    assert_eq!(
                        dt.format("%Y-%m-%d %H:%M:%S").to_string(),
                        "2026-10-28 05:14:15",
                        "输入 {s}"
                    );
                }
                other => panic!("{s} 必须解析为 Zoned，实际 {:?}", other.is_some()),
            }
        }
    }

    #[test]
    fn test_parse_pg_time_fractional() {
        assert!(parse_pg_time("13:14:15").is_some());
        assert!(parse_pg_time("13:14:15.123456").is_some());
        assert!(parse_pg_time("13:14").is_none());
    }

    // ---- R2：DECIMAL 38 位无损 ----

    #[test]
    fn test_bigdecimal_38_digits_no_truncation() {
        let s = "1234567890123456789012345678.9012345678"; // 38 位有效数字
        let d: bigdecimal::BigDecimal = s.parse().unwrap();
        assert_eq!(d.to_string(), s, "BigDecimal 往返必须逐位一致");
        // 反证：rust_decimal 在 96 位尾数处截断（记录回归动机）
        assert!(
            s.parse::<rust_decimal::Decimal>().is_err() || {
                let rd: rust_decimal::Decimal = s.parse().unwrap();
                rd.to_string() != s
            }
        );
    }
}

#[cfg(test)]
mod session_dirty_tests {
    use super::marks_session_dirty;

    #[test]
    fn set_statements_mark_dirty() {
        assert!(marks_session_dirty("SET SESSION sql_mode = DEFAULT"));
        assert!(marks_session_dirty("set names utf8mb4"));
        assert!(marks_session_dirty("SET autocommit = 0"));
        assert!(marks_session_dirty("SET @my_var = 1"));
        assert!(marks_session_dirty("  set\tsql_mode='x'"));
    }

    #[test]
    fn use_marks_dirty() {
        assert!(marks_session_dirty("USE sz_orm_test"));
        assert!(marks_session_dirty("use other_db;"));
    }

    #[test]
    fn set_transaction_not_dirty() {
        assert!(!marks_session_dirty(
            "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"
        ));
        assert!(!marks_session_dirty("set transaction read only"));
        assert!(!marks_session_dirty("SET TRANSACTION;"));
    }

    #[test]
    fn set_transaction_identifier_still_dirty() {
        // `SET TRANSACTIONX` 不是 SET TRANSACTION 语句
        assert!(marks_session_dirty("SET TRANSACTIONX = 1"));
    }

    #[test]
    fn read_and_tx_statements_not_dirty() {
        assert!(!marks_session_dirty("SELECT 1"));
        assert!(!marks_session_dirty("select * from t where a=1"));
        assert!(!marks_session_dirty("SHOW TABLES"));
        assert!(!marks_session_dirty("EXPLAIN SELECT 1"));
        assert!(!marks_session_dirty("START TRANSACTION"));
        assert!(!marks_session_dirty("BEGIN"));
        assert!(!marks_session_dirty("COMMIT"));
        assert!(!marks_session_dirty("ROLLBACK"));
        assert!(!marks_session_dirty("INSERT INTO t VALUES (1)"));
        assert!(!marks_session_dirty("UPDATE t SET a = 1 WHERE id = 1"));
        assert!(!marks_session_dirty("DELETE FROM t WHERE id = 1"));
    }

    #[test]
    fn word_boundary_no_false_positive() {
        // "SETUP..." 不是 SET 语句
        assert!(!marks_session_dirty("SETUP_CHECK = 1"));
        // "USER" 表名查询不是 USE 语句
        assert!(!marks_session_dirty("SELECT * FROM USER"));
    }
}