sz-orm-sqlx 10.13.2

//! 真实 DB Pool/Transaction 行为验证测试（P0-3）
//!
//! 这些测试需要真实 MySQL/PG 实例，默认 ignored
//! 运行方式：cargo test -p sz-orm-sqlx --test real_db_pool_tests -- --ignored --nocapture

use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use sz_orm_core::{Pool, PoolConfigBuilder};
use sz_orm_sqlx::{
    MySqlPoolHandle, PgPoolHandle, SqlxMySqlConnectionFactory, SqlxPgConnectionFactory,
};

/// 默认 MySQL 连接 URL（本机）；可通过环境变量 `SZ_ORM_MYSQL_URL` 覆盖以指向真实云数据库。
const MYSQL_URL_DEFAULT: &str = "mysql://root:test123@127.0.0.1:3306/sz_orm_test";
/// 默认 PostgreSQL 连接 URL（本机）；可通过环境变量 `SZ_ORM_PG_URL` 覆盖以指向真实云数据库。
const PG_URL_DEFAULT: &str = "postgres://postgres:test123@127.0.0.1:5432/sz_orm_test";

fn mysql_url() -> String {
    std::env::var("SZ_ORM_MYSQL_URL").unwrap_or_else(|_| MYSQL_URL_DEFAULT.to_string())
}

fn pg_url() -> String {
    std::env::var("SZ_ORM_PG_URL").unwrap_or_else(|_| PG_URL_DEFAULT.to_string())
}

fn unique_table(prefix: &str) -> String {
    let pid = std::process::id();
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("{}_{}_{}", prefix, pid, nanos % 1_000_000)
}

// ===================== MySQL 真实 DB 测试 =====================

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_pool_acquire_release_roundtrip() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(5)
        .min_idle(0)
        .acquire_timeout(10)
        .build()
        .unwrap();
    let pool = Pool::new(config, factory).unwrap();

    // 多次 acquire/release 应该重用连接
    for i in 0..20u32 {
        let mut conn = pool.acquire().await.unwrap();
        let sql = format!("SELECT {} as n", i);
        let rows = conn.query(&sql).await.unwrap();
        assert_eq!(rows.len(), 1);
        pool.release(conn).await;
    }

    let status = pool.status().await;
    assert!(
        status.idle > 0,
        "should have idle connections after release"
    );
    assert!(status.active <= 5, "active should not exceed max_size");
}

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_pool_concurrent_8_tasks() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_conc");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id INT PRIMARY KEY, value INT)",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(8)
        .min_idle(0)
        .acquire_timeout(15)
        .build()
        .unwrap();
    let pool: &'static Pool = Box::leak(Box::new(Pool::new(config, factory).unwrap()));

    let success = Arc::new(AtomicU32::new(0));
    let mut handles = Vec::new();
    for i in 0..8u32 {
        let sc = success.clone();
        let table = table.clone();
        handles.push(tokio::spawn(async move {
            for j in 0..50u32 {
                if let Ok(mut conn) = pool.acquire().await {
                    let id = i * 1000 + j;
                    let sql = format!("INSERT INTO {} (id, value) VALUES ({}, {})", table, id, id);
                    if conn.execute(&sql).await.is_ok() {
                        sc.fetch_add(1, Ordering::SeqCst);
                    }
                    pool.release(conn).await;
                }
            }
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    assert_eq!(
        success.load(Ordering::SeqCst),
        400,
        "all 8*50 inserts should succeed"
    );

    // 清理
    let mut conn = pool.acquire().await.unwrap();
    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_transaction_commit_rollback() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_tx");
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id INT PRIMARY KEY, name VARCHAR(100))",
        table
    ))
    .await
    .unwrap();

    // commit 测试
    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (1, 'commit_test')",
        table
    ))
    .await
    .unwrap();
    conn.commit().await.unwrap();

    // rollback 测试
    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (2, 'rollback_test')",
        table
    ))
    .await
    .unwrap();
    conn.rollback().await.unwrap();

    let rows = conn
        .query(&format!("SELECT * FROM {}", table))
        .await
        .unwrap();
    assert_eq!(rows.len(), 1, "only committed row should remain");

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

/// 回归测试（2026-09-25 下游 sz-rust 挂死缺陷）：事务中途 drop 连接，归还时自动回滚
///
/// 缺陷：release 路径不回滚未提交事务，开放事务随连接回池；MySQL 会话持续
/// 持有行锁与 MDL 元数据锁，同池其他连接的 DROP TABLE 被永久阻塞。
/// 验证：未提交即 drop → （1）泄漏行被自动回滚；（2）同池连接的 DDL 不再被阻塞。
#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_drop_uncommitted_tx_auto_rollback_on_release() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_tx_leak");
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id INT PRIMARY KEY, val VARCHAR(50))",
        table
    ))
    .await
    .unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, val) VALUES (1, 'kept')",
        table
    ))
    .await
    .unwrap();

    // 事务中途 drop（不 commit/rollback），连接经 Drop spawn 异步归还
    {
        let mut c2 = pool.acquire().await.unwrap();
        c2.begin_transaction().await.unwrap();
        c2.execute(&format!(
            "INSERT INTO {} (id, val) VALUES (2, 'leaked')",
            table
        ))
        .await
        .unwrap();
        // 不 commit 直接 drop
    }

    // 给 Drop spawn 的异步归还（含自动回滚）执行窗口
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    // 断言 1：泄漏行已被归还路径自动回滚
    let rows = conn
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0]
        .get("cnt")
        .and_then(|v| v.as_i64())
        .expect("应包含 cnt 字段");
    assert_eq!(cnt, 1, "未提交事务应在连接归还时被自动回滚");

    // 断言 2（缺陷核心症状）：同池连接的 DDL 不再被 MDL 阻塞；
    // 修复前此 DROP 永久挂起（MySQL lock_wait_timeout 默认 1 年）
    // 30s：高负载（如 CI 并行编译）下 10s 窗口会时序误报；回归语义是
    // "永久挂死"，30s 依然充分（修复后实测 <0.3s）
    let dropped = tokio::time::timeout(
        std::time::Duration::from_secs(30),
        conn.execute(&format!("DROP TABLE IF EXISTS {}", table)),
    )
    .await;
    assert!(
        dropped.is_ok(),
        "DROP TABLE 被 MDL 阻塞超过 10s：归还路径未回滚未提交事务"
    );

    pool.release(conn).await;
}

/// 回归测试（2026-09-25 深度检查发现）：流式查询提前弃读 → 毒化连接不得回池
///
/// 缺陷：query_stream 将 PoolConnection 从包装器 take() 出去（self.conn = None），
/// 提前 drop 流后包装器毒化但 connected 仍为 true，回池后下次 acquire 借出即报
/// "connection already closed"。修复后 is_connected 纳入 conn.is_some()，
/// acquire 路径过滤并关闭毒化连接（自愈）。
#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_stream_abandon_self_heals_on_reacquire() {
    use futures::StreamExt;
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_stream_leak");
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(1).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!("CREATE TABLE {} (id INT PRIMARY KEY)", table))
        .await
        .unwrap();
    for i in 0..50u32 {
        conn.execute(&format!("INSERT INTO {} (id) VALUES ({})", table, i))
            .await
            .unwrap();
    }

    // 读 1 行后提前弃读：PoolConnection 已被 take 出包装器
    {
        let select_sql = format!("SELECT id FROM {}", table);
        let mut stream = conn.query_stream(&select_sql);
        let first = stream.next().await.unwrap().unwrap();
        assert_eq!(first.get("id").and_then(|v| v.as_i64()), Some(0));
        // drop(stream) —— 弃读
    }
    drop(conn); // 毒化包装器经 release 回池

    // 重新 acquire：毒化连接应被过滤关闭，借出的是新建连接
    let mut conn2 = pool.acquire().await.unwrap();
    let rows = conn2
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0].get("cnt").and_then(|v| v.as_i64()).unwrap_or(-1);
    assert_eq!(cnt, 50, "重借连接应可正常查询（毒化连接已被过滤关闭）");
    conn2
        .execute(&format!("DROP TABLE IF EXISTS {}", table))
        .await
        .ok();
    pool.release(conn2).await;
}

/// 回归测试（2026-09-25 深度检查发现的设计缺口）：into_inner 移交不得泄漏池容量
///
/// 缺陷：into_inner 不递减 total_count 且无归还 API——Transaction 流程
/// （acquire → into_inner → Transaction → 丢弃连接）每执行一次永久损失一格
/// 容量，max_size 小的池很快 acquire Timeout。
/// 修复：into_inner 移交时递减 total_count；新增 Pool::return_raw 归还 API。
#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_into_inner_capacity_roundtrip() {
    use sz_orm_core::{TransactOptions, Transaction};
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    // 泄漏回归：acquire → into_inner → 丢弃（修复前：active 永久占用）
    for _ in 0..4 {
        let conn = pool.acquire().await.unwrap();
        let raw = conn.into_inner();
        assert_eq!(
            pool.status().await.active,
            0,
            "移交后槽位应立即释放（修复前：active 恒为 1，容量永久泄漏）"
        );
        drop(raw); // 不归还——连接随之关闭
        assert_eq!(pool.status().await.active, 0, "丢弃后不得占用池容量");
    }

    // Transaction 流程闭环（stress.rs 模式）：acquire → into_inner → Transaction
    // → commit → take_connection → return_raw
    let table = unique_table("mysql_inner_tx");
    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!("CREATE TABLE {} (id INT PRIMARY KEY)", table))
        .await
        .unwrap();
    let mut raw = conn.into_inner();
    raw.begin_transaction().await.unwrap();
    raw.execute(&format!("INSERT INTO {} (id) VALUES (1)", table))
        .await
        .unwrap();
    let mut tx = Transaction::new(raw, TransactOptions::default());
    tx.commit().await.unwrap();
    let raw2 = tx.take_connection().await.unwrap();
    pool.return_raw(raw2).await.expect("归还应成功");
    let st = pool.status().await;
    assert_eq!(st.active, 1, "归还后槽位重新占用");
    assert_eq!(st.idle, 1, "归还的连接应进入空闲队列");

    // 归还的连接可复用且数据已提交可见
    let mut conn2 = pool.acquire().await.unwrap();
    let rows = conn2
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0].get("cnt").and_then(|v| v.as_i64()).unwrap_or(-1);
    assert_eq!(cnt, 1, "复用归还连接应健康且事务已提交");
    conn2
        .execute(&format!("DROP TABLE IF EXISTS {}", table))
        .await
        .ok();
    pool.release(conn2).await;
}

/// return_raw 卫生检查：带未提交事务的裸连接归还时自动回滚（与 release 同规则）
#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_return_raw_rolls_back_open_transaction() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_raw_tx");
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!("CREATE TABLE {} (id INT PRIMARY KEY)", table))
        .await
        .unwrap();
    let mut raw = conn.into_inner();
    raw.begin_transaction().await.unwrap();
    raw.execute(&format!("INSERT INTO {} (id) VALUES (1)", table))
        .await
        .unwrap();
    assert!(raw.in_transaction(), "归还前应处于未提交事务中");
    pool.return_raw(raw).await.expect("归还应成功（自动回滚）");

    let mut conn2 = pool.acquire().await.unwrap();
    assert!(!conn2.in_transaction(), "归还时未提交事务应被自动回滚");
    let rows = conn2
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0].get("cnt").and_then(|v| v.as_i64()).unwrap_or(-1);
    assert_eq!(cnt, 0, "未提交行应已回滚");
    conn2
        .execute(&format!("DROP TABLE IF EXISTS {}", table))
        .await
        .ok();
    pool.release(conn2).await;
}

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_savepoint_nested() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_sp");
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id INT PRIMARY KEY, name VARCHAR(100))",
        table
    ))
    .await
    .unwrap();

    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (1, 'outer')",
        table
    ))
    .await
    .unwrap();
    conn.execute("SAVEPOINT sp1").await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (2, 'inner')",
        table
    ))
    .await
    .unwrap();
    conn.execute("ROLLBACK TO SAVEPOINT sp1").await.unwrap();
    conn.execute("RELEASE SAVEPOINT sp1").await.unwrap();
    conn.commit().await.unwrap();

    let rows = conn
        .query(&format!("SELECT * FROM {} ORDER BY id", table))
        .await
        .unwrap();
    assert_eq!(
        rows.len(),
        1,
        "only outer row should remain after savepoint rollback"
    );

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_pool_stress_10k_ops() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_stress");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id INT PRIMARY KEY, value INT)",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(10)
        .min_idle(2)
        .acquire_timeout(30)
        .build()
        .unwrap();
    let pool: &'static Pool = Box::leak(Box::new(Pool::new(config, factory).unwrap()));

    let success = Arc::new(AtomicU32::new(0));
    let mut handles = Vec::new();
    for i in 0..10u32 {
        let sc = success.clone();
        let table = table.clone();
        handles.push(tokio::spawn(async move {
            for j in 0..1000u32 {
                let id = i * 10000 + j;
                if let Ok(mut conn) = pool.acquire().await {
                    let sql = format!("INSERT INTO {} (id, value) VALUES ({}, {})", table, id, id);
                    if conn.execute(&sql).await.is_ok() {
                        sc.fetch_add(1, Ordering::SeqCst);
                    }
                    pool.release(conn).await;
                }
            }
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    assert_eq!(
        success.load(Ordering::SeqCst),
        10000,
        "all 10*1000 = 10k inserts should succeed"
    );

    // 清理
    let mut conn = pool.acquire().await.unwrap();
    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

// ===================== PostgreSQL 真实 DB 测试 =====================

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_pool_acquire_release_roundtrip() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(5)
        .min_idle(0)
        .acquire_timeout(10)
        .build()
        .unwrap();
    let pool = Pool::new(config, factory).unwrap();

    for i in 0..20u32 {
        let mut conn = pool.acquire().await.unwrap();
        let sql = format!("SELECT {} as n", i);
        let rows = conn.query(&sql).await.unwrap();
        assert_eq!(rows.len(), 1);
        pool.release(conn).await;
    }
    let status = pool.status().await;
    assert!(status.idle > 0, "should have idle connections");
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_pool_concurrent_8_tasks() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_conc");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id BIGINT PRIMARY KEY, value BIGINT)",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(8)
        .min_idle(0)
        .acquire_timeout(15)
        .build()
        .unwrap();
    let pool: &'static Pool = Box::leak(Box::new(Pool::new(config, factory).unwrap()));

    let success = Arc::new(AtomicU32::new(0));
    let mut handles = Vec::new();
    for i in 0..8u32 {
        let sc = success.clone();
        let table = table.clone();
        handles.push(tokio::spawn(async move {
            for j in 0..50u32 {
                let id: i64 = (i as i64) * 1000 + j as i64;
                if let Ok(mut conn) = pool.acquire().await {
                    let sql = format!("INSERT INTO {} (id, value) VALUES ({}, {})", table, id, id);
                    if conn.execute(&sql).await.is_ok() {
                        sc.fetch_add(1, Ordering::SeqCst);
                    }
                    pool.release(conn).await;
                }
            }
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    assert_eq!(
        success.load(Ordering::SeqCst),
        400,
        "all 8*50 inserts should succeed"
    );

    let mut conn = pool.acquire().await.unwrap();
    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

/// PG 回归测试（与 MySQL 版对齐）：事务中途 drop 连接，归还时自动回滚。
/// PG 侧开放事务持有 ACCESS SHARE 锁，同池连接的 DROP TABLE（需 ACCESS EXCLUSIVE）
/// 在修复前同样被永久阻塞。
#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_drop_uncommitted_tx_auto_rollback_on_release() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_tx_leak");
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id BIGINT PRIMARY KEY, val TEXT)",
        table
    ))
    .await
    .unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, val) VALUES (1, 'kept')",
        table
    ))
    .await
    .unwrap();

    {
        let mut c2 = pool.acquire().await.unwrap();
        c2.begin_transaction().await.unwrap();
        c2.execute(&format!(
            "INSERT INTO {} (id, val) VALUES (2, 'leaked')",
            table
        ))
        .await
        .unwrap();
        // 不 commit 直接 drop
    }

    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let rows = conn
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0]
        .get("cnt")
        .and_then(|v| v.as_i64())
        .expect("应包含 cnt 字段");
    assert_eq!(cnt, 1, "未提交事务应在连接归还时被自动回滚");

    // 30s：高负载（如 CI 并行编译）下 10s 窗口会时序误报；回归语义是
    // "永久挂死"，30s 依然充分（修复后实测 <0.3s）
    let dropped = tokio::time::timeout(
        std::time::Duration::from_secs(30),
        conn.execute(&format!("DROP TABLE IF EXISTS {}", table)),
    )
    .await;
    assert!(
        dropped.is_ok(),
        "DROP TABLE 被锁阻塞超过 10s：归还路径未回滚未提交事务"
    );

    pool.release(conn).await;
}

/// PG 回归测试（与 MySQL 弃读流测试对齐）：三后端 query_stream 同构 take() 模式，
/// 毒化连接经 is_connected 修复后由 acquire 路径过滤自愈。
#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_stream_abandon_self_heals_on_reacquire() {
    use futures::StreamExt;
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_stream_leak");
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(1).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!("CREATE TABLE {} (id BIGINT PRIMARY KEY)", table))
        .await
        .unwrap();
    for i in 0..50i64 {
        conn.execute(&format!("INSERT INTO {} (id) VALUES ({})", table, i))
            .await
            .unwrap();
    }

    {
        let select_sql = format!("SELECT id FROM {}", table);
        let mut stream = conn.query_stream(&select_sql);
        let first = stream.next().await.unwrap().unwrap();
        assert_eq!(first.get("id").and_then(|v| v.as_i64()), Some(0));
        // drop(stream) —— 弃读
    }
    drop(conn);

    let mut conn2 = pool.acquire().await.unwrap();
    let rows = conn2
        .query(&format!("SELECT COUNT(*) AS cnt FROM {}", table))
        .await
        .unwrap();
    let cnt = rows[0].get("cnt").and_then(|v| v.as_i64()).unwrap_or(-1);
    assert_eq!(cnt, 50, "重借连接应可正常查询（毒化连接已被过滤关闭）");
    conn2
        .execute(&format!("DROP TABLE IF EXISTS {}", table))
        .await
        .ok();
    pool.release(conn2).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_transaction_commit_rollback() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_tx");
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id BIGINT PRIMARY KEY, name TEXT)",
        table
    ))
    .await
    .unwrap();

    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (1, 'commit_test')",
        table
    ))
    .await
    .unwrap();
    conn.commit().await.unwrap();

    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (2, 'rollback_test')",
        table
    ))
    .await
    .unwrap();
    conn.rollback().await.unwrap();

    let rows = conn
        .query(&format!("SELECT * FROM {}", table))
        .await
        .unwrap();
    assert_eq!(rows.len(), 1, "only committed row should remain");

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_savepoint_nested() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_sp");
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle.clone()));
    let config = PoolConfigBuilder::new().max_size(3).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute(&format!(
        "CREATE TABLE {} (id BIGINT PRIMARY KEY, name TEXT)",
        table
    ))
    .await
    .unwrap();

    conn.begin_transaction().await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (1, 'outer')",
        table
    ))
    .await
    .unwrap();
    conn.execute("SAVEPOINT sp1").await.unwrap();
    conn.execute(&format!(
        "INSERT INTO {} (id, name) VALUES (2, 'inner')",
        table
    ))
    .await
    .unwrap();
    conn.execute("ROLLBACK TO SAVEPOINT sp1").await.unwrap();
    conn.execute("RELEASE SAVEPOINT sp1").await.unwrap();
    conn.commit().await.unwrap();

    let rows = conn
        .query(&format!("SELECT * FROM {} ORDER BY id", table))
        .await
        .unwrap();
    assert_eq!(rows.len(), 1, "only outer row should remain");

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_pool_stress_10k_ops() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_stress");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id BIGINT PRIMARY KEY, value BIGINT)",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new()
        .max_size(10)
        .min_idle(2)
        .acquire_timeout(30)
        .build()
        .unwrap();
    let pool: &'static Pool = Box::leak(Box::new(Pool::new(config, factory).unwrap()));

    let success = Arc::new(AtomicU32::new(0));
    let mut handles = Vec::new();
    for i in 0..10u32 {
        let sc = success.clone();
        let table = table.clone();
        handles.push(tokio::spawn(async move {
            for j in 0..1000u32 {
                let id: i64 = (i as i64) * 10000 + j as i64;
                if let Ok(mut conn) = pool.acquire().await {
                    let sql = format!("INSERT INTO {} (id, value) VALUES ({}, {})", table, id, id);
                    if conn.execute(&sql).await.is_ok() {
                        sc.fetch_add(1, Ordering::SeqCst);
                    }
                    pool.release(conn).await;
                }
            }
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    assert_eq!(
        success.load(Ordering::SeqCst),
        10000,
        "all 10k inserts should succeed"
    );

    let mut conn = pool.acquire().await.unwrap();
    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

// ===================== 超大数据量测试 =====================

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0，10万行批量插入"]
async fn mysql_bulk_insert_100k_via_pool() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let table = unique_table("mysql_bulk");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id BIGINT PRIMARY KEY, value BIGINT, name VARCHAR(50))",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new().max_size(5).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let start = std::time::Instant::now();
    let mut conn = pool.acquire().await.unwrap();
    conn.begin_transaction().await.unwrap();
    for i in 0..100_000u32 {
        let sql = format!(
            "INSERT INTO {} (id, value, name) VALUES ({}, {}, 'name_{}')",
            table, i, i, i
        );
        conn.execute(&sql).await.unwrap();
    }
    conn.commit().await.unwrap();
    let elapsed = start.elapsed();

    let rows = conn
        .query(&format!("SELECT COUNT(*) as c FROM {}", table))
        .await
        .unwrap();
    let count = rows[0].get("c").and_then(|v| v.as_i64()).unwrap_or(0);
    assert_eq!(count, 100_000, "should have 100k rows");

    println!("MySQL 10万行插入 via sz-orm-core Pool: {:?}", elapsed);

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18，10万行批量插入"]
async fn pg_bulk_insert_100k_via_pool() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let table = unique_table("pg_bulk");
    {
        let mut conn = pool_handle.pool().acquire().await.unwrap();
        sqlx::query(sqlx::AssertSqlSafe(&*format!(
            "CREATE TABLE {} (id BIGINT PRIMARY KEY, value BIGINT, name TEXT)",
            table
        )))
        .execute(&mut *conn)
        .await
        .unwrap();
    }

    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new().max_size(5).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let start = std::time::Instant::now();
    let mut conn = pool.acquire().await.unwrap();
    conn.begin_transaction().await.unwrap();
    for i in 0..100_000u32 {
        let sql = format!(
            "INSERT INTO {} (id, value, name) VALUES ({}, {}, 'name_{}')",
            table, i, i, i
        );
        conn.execute(&sql).await.unwrap();
    }
    conn.commit().await.unwrap();
    let elapsed = start.elapsed();

    let rows = conn
        .query(&format!("SELECT COUNT(*) as c FROM {}", table))
        .await
        .unwrap();
    let count = rows[0].get("c").and_then(|v| v.as_i64()).unwrap_or(0);
    assert_eq!(count, 100_000, "should have 100k rows");

    println!("PG 10万行插入 via sz-orm-core Pool: {:?}", elapsed);

    let _ = conn.execute(&format!("DROP TABLE {}", table)).await;
    pool.release(conn).await;
}

// ===================== R3：隔离级别在 BEGIN 前后正确下发（真库回归） =====================

/// MySQL：`SET TRANSACTION ISOLATION LEVEL` 必须在 BEGIN 之前生效。
///
/// 回归背景（2026-10-28 DBA 评测 R3）：旧实现把 `SET TRANSACTION` 放在 BEGIN 之后，
/// MySQL 语义是“作用于下一个事务”，导致显式隔离级别的**当前**事务 100% 失败。
/// 会话变量 `@@transaction_isolation` 不随一次性 `SET TRANSACTION` 变化，故用可观察
/// 行为断言：READ COMMITTED 事务内第二次读必须看到外部已提交的新行；默认
/// REPEATABLE READ 则必须保持快照。两条路径构成差分。
async fn iso_count(
    tx: &mut sz_orm_core::Transaction,
    table: &str,
) -> Result<i64, sz_orm_core::TxError> {
    let rows = tx
        .query(&format!("SELECT COUNT(*) AS c FROM {}", table))
        .await?;
    Ok(rows[0].get("c").and_then(|v| v.as_i64()).unwrap_or(-1))
}

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_isolation_level_applied_before_begin() {
    use sz_orm_core::{run_transaction, IsolationLevel, TransactOptions};
    use tokio::sync::Mutex;

    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new().max_size(4).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let table = unique_table("mysql_iso_rc");
    let writer = Arc::new(Mutex::new(pool.acquire().await.unwrap()));
    {
        let mut w = writer.lock().await;
        w.execute(&format!("CREATE TABLE {} (id INT PRIMARY KEY)", table))
            .await
            .unwrap();
        w.execute(&format!("INSERT INTO {} (id) VALUES (1)", table))
            .await
            .unwrap();
    }

    // 默认（REPEATABLE READ）：事务内第二次读仍是首读快照 (1, 1)
    let reader = pool.acquire().await.unwrap();
    let default_pair = {
        let writer = writer.clone();
        let table = table.clone();
        run_transaction(reader.into_inner(), TransactOptions::default(), move |tx| {
            Box::pin(async move {
                let first = iso_count(tx, &table).await?;
                {
                    let mut w = writer.lock().await;
                    w.execute(&format!("INSERT INTO {} (id) VALUES (2)", table))
                        .await
                        .map_err(|e| sz_orm_core::TxError::CommitFailed(e.to_string()))?;
                }
                let second = iso_count(tx, &table).await?;
                Ok((first, second))
            })
        })
        .await
        .expect("默认事务应成功")
    };
    assert_eq!(
        default_pair,
        (1, 1),
        "默认 REPEATABLE READ 下事务内快照不应看到外部新提交"
    );

    // READ COMMITTED：BEGIN 前下发后，第二次读必须看到外部新提交 (2, 3)
    let reader = pool.acquire().await.unwrap();
    let rc_pair = {
        let writer = writer.clone();
        let table = table.clone();
        run_transaction(
            reader.into_inner(),
            TransactOptions::default().with_isolation(IsolationLevel::ReadCommitted),
            move |tx| {
                Box::pin(async move {
                    let first = iso_count(tx, &table).await?;
                    {
                        let mut w = writer.lock().await;
                        w.execute(&format!("INSERT INTO {} (id) VALUES (3)", table))
                            .await
                            .map_err(|e| sz_orm_core::TxError::CommitFailed(e.to_string()))?;
                    }
                    let second = iso_count(tx, &table).await?;
                    Ok((first, second))
                })
            },
        )
        .await
        .expect("READ COMMITTED 事务应成功")
    };
    assert_eq!(
        rc_pair,
        (2, 3),
        "READ COMMITTED 事务内第二次读必须看到外部新提交（SET 在 BEGIN 之前下发）"
    );

    let mut w = writer.lock().await;
    w.execute(&format!("DROP TABLE IF EXISTS {}", table))
        .await
        .ok();
}

/// PostgreSQL：`SET TRANSACTION ISOLATION LEVEL` 必须在事务内部下发（PG 语义要求）。
#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_isolation_level_applied_inside_transaction() {
    use sz_orm_core::{run_transaction, IsolationLevel, TransactOptions};

    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new().max_size(2).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let conn = pool.acquire().await.unwrap();
    let level = run_transaction(
        conn.into_inner(),
        TransactOptions::default().with_isolation(IsolationLevel::Serializable),
        |tx| {
            Box::pin(async move {
                let rows = tx
                    .query("SELECT current_setting('transaction_isolation') AS lvl")
                    .await?;
                let lvl = rows[0]
                    .get("lvl")
                    .and_then(|v| v.as_str())
                    .unwrap_or("")
                    .to_string();
                Ok(lvl)
            })
        },
    )
    .await
    .expect("显式隔离级别事务应成功");

    assert_eq!(
        level.to_ascii_lowercase(),
        "serializable",
        "事务内隔离级别必须是 serializable（SET TRANSACTION 在 BEGIN 之后下发）"
    );
}

// ===================== R5：归还连接时会话状态复位（真库回归） =====================

#[tokio::test]
#[ignore = "需要 MySQL 9.6.0 在 127.0.0.1:3306"]
async fn mysql_release_resets_session_state() {
    let pool_handle = Arc::new(MySqlPoolHandle::connect(&mysql_url()).await.unwrap());
    let factory = Arc::new(SqlxMySqlConnectionFactory::new(pool_handle));
    // max_size(1)：确保归还后重新借出的是同一条物理连接
    let config = PoolConfigBuilder::new().max_size(1).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute("SET SESSION sql_mode = 'ANSI_QUOTES'")
        .await
        .unwrap();
    let dirty = conn.query("SELECT @@SESSION.sql_mode AS m").await.unwrap();
    let dirty_mode = dirty[0]
        .get("m")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_uppercase();
    assert!(dirty_mode.contains("ANSI_QUOTES"), "前置条件：会话已被污染");
    pool.release(conn).await;

    let mut reused = pool.acquire().await.unwrap();
    let clean = reused
        .query("SELECT @@SESSION.sql_mode AS m")
        .await
        .unwrap();
    let clean_mode = clean[0]
        .get("m")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_uppercase();
    assert!(
        !clean_mode.contains("ANSI_QUOTES"),
        "归还时必须复位 sql_mode（R5），实际仍为 {clean_mode}"
    );
    pool.release(reused).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 在 127.0.0.1:5432"]
async fn pg_release_resets_session_state() {
    let pool_handle = Arc::new(PgPoolHandle::connect(&pg_url()).await.unwrap());
    let factory = Arc::new(SqlxPgConnectionFactory::new(pool_handle));
    let config = PoolConfigBuilder::new().max_size(1).build().unwrap();
    let pool = Pool::new(config, factory).unwrap();

    let mut conn = pool.acquire().await.unwrap();
    conn.execute("SET search_path TO pg_catalog").await.unwrap();
    let dirty = conn.query("SHOW search_path").await.unwrap();
    assert_eq!(
        dirty[0].get("search_path").and_then(|v| v.as_str()),
        Some("pg_catalog"),
        "前置条件：会话已被污染"
    );
    pool.release(conn).await;

    let mut reused = pool.acquire().await.unwrap();
    let clean = reused.query("SHOW search_path").await.unwrap();
    let path = clean[0]
        .get("search_path")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    assert_ne!(
        path, "pg_catalog",
        "归还时必须执行 DISCARD ALL 复位 search_path（R5）"
    );
    pool.release(reused).await;
}