sz-orm-core 1.2.2

//! PostgreSQL 真实数据库集成测试
//!
//! 使用 sqlx (PostgreSQL 18) 验证 sz-orm-core 的 PostgreSQL 方言、值转换、
//! SQL 转义、事务、连接池语义、分页、JSON 操作、SQL 注入防护等核心功能。
//!
//! 超大数据量场景：10 万条记录 CRUD、8 任务并发读写、批量插入性能基线。
//!
//! 测试数据库：postgres://postgres:<your-password>@127.0.0.1:5432/sz_orm_test
//!
//! 运行方式：cargo test --package sz-orm-core --test integration_pg -- --ignored --nocapture

use sqlx::postgres::{PgPool, PgPoolOptions};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use sz_orm_core::dialect::{get_dialect, ColumnDef};
use sz_orm_core::{DbType, Model, ModelExt, QueryBuilder, Value};

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

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

/// 全局唯一表名计数器
static TABLE_COUNTER: AtomicU64 = AtomicU64::new(0);

fn unique_table(prefix: &str) -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    let counter = TABLE_COUNTER.fetch_add(1, Ordering::Relaxed);
    // PG 标识符最长 63 字节，使用简短前缀
    format!("{}_{}_{}", prefix, nanos % 1_000_000, counter)
}

async fn setup_pool() -> PgPool {
    PgPoolOptions::new()
        .max_connections(8)
        .acquire_timeout(Duration::from_secs(30))
        .connect(&pg_url())
        .await
        .expect("pg connect failed - is PostgreSQL 18 running?")
}

/// 用方言生成 CREATE TABLE 并执行（PG 使用 SERIAL/BIGSERIAL 自动递增）
async fn create_test_table(pool: &PgPool, table: &str) {
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let columns = vec![
        ColumnDef {
            name: "id".to_string(),
            sql_type: "BIGSERIAL".to_string(),
            nullable: false,
            default: None,
            auto_increment: true,
            primary_key: true,
        },
        ColumnDef {
            name: "name".to_string(),
            sql_type: "VARCHAR(255)".to_string(),
            nullable: false,
            default: None,
            auto_increment: false,
            primary_key: false,
        },
        ColumnDef {
            name: "value".to_string(),
            sql_type: "BIGINT".to_string(),
            nullable: true,
            default: None,
            auto_increment: false,
            primary_key: false,
        },
        ColumnDef {
            name: "data".to_string(),
            sql_type: "TEXT".to_string(),
            nullable: true,
            default: None,
            auto_increment: false,
            primary_key: false,
        },
        ColumnDef {
            name: "meta".to_string(),
            sql_type: "JSONB".to_string(),
            nullable: true,
            default: None,
            auto_increment: false,
            primary_key: false,
        },
    ];
    let sql = dialect.build_create_table(table, &columns);
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .execute(pool)
        .await
        .expect("create table");
}

async fn drop_table(pool: &PgPool, table: &str) {
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let sql = dialect.build_drop_table(table, true);
    let _ = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .execute(pool)
        .await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_dialect_basics() {
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    assert_eq!(dialect.quote("user"), "\"user\"");
    assert_eq!(dialect.quote("with\"quote"), "\"with\"\"quote\"");
    assert_eq!(dialect.escape_string("it's"), "it''s");
    assert_eq!(dialect.escape_string("back\\slash"), "back\\slash");
    assert!(dialect.supports_returning());
    // sz-orm-core PG 方言使用 IDENTITY 列（PG 10+ 标准方式）
    assert_eq!(
        dialect.auto_increment_keyword(),
        "GENERATED BY DEFAULT AS IDENTITY"
    );
    assert_eq!(dialect.json_type(), "JSONB");
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_create_insert_select() {
    let pool = setup_pool().await;
    let table = unique_table("t1");
    create_test_table(&pool, &table).await;

    // PG 使用 $1, $2 参数化
    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("alice")
        .bind(100i64)
        .bind("data1")
        .execute(&pool)
        .await
        .expect("insert 1");
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("bob")
        .bind(200i64)
        .bind("data2")
        .execute(&pool)
        .await
        .expect("insert 2");
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("carol")
        .bind(300i64)
        .bind("data3")
        .execute(&pool)
        .await
        .expect("insert 3");

    let select_sql = format!("SELECT name, value FROM \"{}\" ORDER BY id", table);
    let rows: Vec<(String, i64)> = sqlx::query_as(sqlx::AssertSqlSafe(select_sql.as_str()))
        .fetch_all(&pool)
        .await
        .expect("select");
    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].0, "alice");
    assert_eq!(rows[2].0, "carol");

    // Value 类型转换验证
    let v = Value::String("alice".to_string());
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let escaped = dialect.escape_string(v.as_str().unwrap());
    let sql = format!("SELECT value FROM \"{}\" WHERE name = '{}'", table, escaped);
    let row: (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(sql.as_str()))
        .fetch_one(&pool)
        .await
        .expect("query row");
    assert_eq!(row.0, 100);

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_bulk_insert_100k() {
    let pool = setup_pool().await;
    let table = unique_table("t_bulk");
    create_test_table(&pool, &table).await;

    let total: usize = 100_000;
    let start = Instant::now();

    // PG 使用 UNNEST 批量插入性能更优；这里为简单使用 batched VALUES
    let mut tx = pool.begin().await.expect("begin");
    let batch_size = 1000;
    let mut total_inserted = 0usize;
    for batch_start in (0..total).step_by(batch_size) {
        let batch_end = (batch_start + batch_size).min(total);
        let placeholders: Vec<String> = (batch_start..batch_end)
            .enumerate()
            .map(|(i, _)| {
                let base = i * 3;
                format!("(${}, ${}, ${})", base + 1, base + 2, base + 3)
            })
            .collect();
        let sql = format!(
            "INSERT INTO \"{}\" (name, value, data) VALUES {}",
            table,
            placeholders.join(", ")
        );
        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
        for i in batch_start..batch_end {
            q = q
                .bind(format!("user_{}", i))
                .bind(i as i64)
                .bind(format!("data_{}", i % 1000));
        }
        q.execute(&mut *tx).await.expect("batch insert");
        total_inserted += batch_end - batch_start;
    }
    tx.commit().await.expect("commit");
    let elapsed = start.elapsed();
    println!(
        "pg bulk insert {} rows in {:?} ({:.0} rows/s)",
        total,
        elapsed,
        total as f64 / elapsed.as_secs_f64()
    );
    assert_eq!(total_inserted, total);

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .expect("count");
    assert_eq!(count as usize, total);

    let last_sql = format!("SELECT name FROM \"{}\" WHERE value = $1", table);
    let (last_name,): (String,) = sqlx::query_as(sqlx::AssertSqlSafe(last_sql.as_str()))
        .bind((total - 1) as i64)
        .fetch_one(&pool)
        .await
        .expect("query last");
    assert_eq!(last_name, format!("user_{}", total - 1));

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_update_delete() {
    let pool = setup_pool().await;
    let table = unique_table("t_ud");
    create_test_table(&pool, &table).await;

    let mut tx = pool.begin().await.expect("begin");
    for i in 0..1000i64 {
        let sql = format!(
            "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
            table
        );
        sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
            .bind(format!("n_{}", i))
            .bind(i)
            .bind("x")
            .execute(&mut *tx)
            .await
            .expect("insert");
    }
    tx.commit().await.expect("commit");

    let upd = format!(
        "UPDATE \"{}\" SET value = value + 1000 WHERE value < 100",
        table
    );
    let result = sqlx::query(sqlx::AssertSqlSafe(upd.as_str()))
        .execute(&pool)
        .await
        .expect("update");
    assert_eq!(result.rows_affected(), 100);

    let del = format!("DELETE FROM \"{}\" WHERE value >= 1000", table);
    let result = sqlx::query(sqlx::AssertSqlSafe(del.as_str()))
        .execute(&pool)
        .await
        .expect("delete");
    assert_eq!(result.rows_affected(), 100);

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 900);

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_transaction_commit() {
    let pool = setup_pool().await;
    let table = unique_table("t_tc");
    create_test_table(&pool, &table).await;

    let mut tx = pool.begin().await.expect("begin");
    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("commit_row")
        .bind(1i64)
        .bind("c")
        .execute(&mut *tx)
        .await
        .expect("insert");
    tx.commit().await.expect("commit");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 1);

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_transaction_rollback() {
    let pool = setup_pool().await;
    let table = unique_table("t_tr");
    create_test_table(&pool, &table).await;

    let mut tx = pool.begin().await.expect("begin");
    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("rollback_row")
        .bind(1i64)
        .bind("r")
        .execute(&mut *tx)
        .await
        .expect("insert");
    tx.rollback().await.expect("rollback");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 0, "rollback should leave table empty");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_pagination() {
    let pool = setup_pool().await;
    let table = unique_table("t_page");
    create_test_table(&pool, &table).await;

    let mut tx = pool.begin().await.expect("begin");
    for i in 0..1000i64 {
        let sql = format!(
            "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
            table
        );
        sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
            .bind(format!("p_{}", i))
            .bind(i)
            .bind("p")
            .execute(&mut *tx)
            .await
            .expect("insert");
    }
    tx.commit().await.expect("commit");

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let page_size = 50u64;
    let mut total_fetched = 0u64;
    let mut last_value = -1i64;
    for page in 1..=20 {
        let sql = dialect.build_pagination(
            &format!("SELECT value FROM \"{}\" ORDER BY value", table),
            page,
            page_size,
        );
        let rows: Vec<(i64,)> = sqlx::query_as(sqlx::AssertSqlSafe(sql.as_str()))
            .fetch_all(&pool)
            .await
            .expect("page query");
        assert_eq!(rows.len() as u64, page_size, "page {} size mismatch", page);
        for (v,) in rows {
            assert!(
                v > last_value,
                "pagination order violated: {} <= {}",
                v,
                last_value
            );
            last_value = v;
            total_fetched += 1;
        }
    }
    assert_eq!(total_fetched, 1000);

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_sql_injection_protection() {
    let pool = setup_pool().await;
    let table = unique_table("t_inj");
    create_test_table(&pool, &table).await;

    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("alice")
        .bind(1i64)
        .bind("x")
        .execute(&pool)
        .await
        .expect("insert");

    let malicious = "alice' OR '1'='1";
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let escaped = dialect.escape_string(malicious);

    let sql = format!(
        "SELECT COUNT(*) FROM \"{}\" WHERE name = '{}'",
        table, escaped
    );
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 0, "escaped malicious input should match nothing");

    let unescaped_sql = format!(
        "SELECT COUNT(*) FROM \"{}\" WHERE name = '{}'",
        table, malicious
    );
    let (count_unescaped,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(unescaped_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count_unescaped, 1, "unescaped input should be injectable");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_json_operations() {
    let pool = setup_pool().await;
    let table = unique_table("t_json");
    create_test_table(&pool, &table).await;

    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data, meta) VALUES ($1, $2, $3, $4)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("alice")
        .bind(1i64)
        .bind("d1")
        .bind(serde_json::json!({"age": 30, "city": "shanghai"}))
        .execute(&pool)
        .await
        .expect("insert 1");
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("bob")
        .bind(2i64)
        .bind("d2")
        .bind(serde_json::json!({"age": 25, "city": "beijing"}))
        .execute(&pool)
        .await
        .expect("insert 2");

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let extract_expr = dialect.json_extract("meta", "$.age");
    let sql = format!(
        "SELECT name FROM \"{}\" WHERE ({})::int > 26 ORDER BY name",
        table, extract_expr
    );
    let rows: Vec<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(sql.as_str()))
        .fetch_all(&pool)
        .await
        .expect("json query");
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].0, "alice");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_concurrent_8tasks_10k_ops() {
    let pool = setup_pool().await;
    let table = unique_table("t_conc");
    create_test_table(&pool, &table).await;

    // 预填充 10000 条
    let mut tx = pool.begin().await.expect("begin");
    let batch_size = 1000;
    for batch_start in (0..10_000).step_by(batch_size) {
        let batch_end = (batch_start + batch_size).min(10_000);
        let placeholders: Vec<String> = (batch_start..batch_end)
            .enumerate()
            .map(|(i, _)| {
                let base = i * 3;
                format!("(${}, ${}, ${})", base + 1, base + 2, base + 3)
            })
            .collect();
        let sql = format!(
            "INSERT INTO \"{}\" (name, value, data) VALUES {}",
            table,
            placeholders.join(", ")
        );
        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
        for i in batch_start..batch_end {
            q = q.bind(format!("u_{}", i)).bind(i as i64).bind("init");
        }
        q.execute(&mut *tx).await.expect("batch insert");
    }
    tx.commit().await.expect("commit");

    let pool_arc = std::sync::Arc::new(pool);
    let table_arc = std::sync::Arc::new(table);
    let ops_per_task: u64 = 10_000;
    let mut handles = vec![];

    for task_id in 0..8u64 {
        let pool_clone = pool_arc.clone();
        let table_clone = table_arc.clone();
        handles.push(tokio::spawn(async move {
            let mut success = 0u64;
            let mut errors = 0u64;
            for op in 0..ops_per_task {
                let key = (task_id * ops_per_task + op) as i64;
                let sql = format!("UPDATE \"{}\" SET data = $1 WHERE value = $2", table_clone);
                let res = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
                    .bind(format!("task_{}_op_{}", task_id, op))
                    .bind(key)
                    .execute(&*pool_clone)
                    .await;
                match res {
                    Ok(_) => success += 1,
                    Err(e) => {
                        errors += 1;
                        eprintln!("task {} op {} error: {}", task_id, op, e);
                    }
                }
            }
            (task_id, success, errors)
        }));
    }

    let mut total_success = 0u64;
    let mut total_errors = 0u64;
    for h in handles {
        let (task_id, success, errors) = h.await.expect("task join");
        println!("task {} success={} errors={}", task_id, success, errors);
        total_success += success;
        total_errors += errors;
    }
    assert_eq!(
        total_success,
        8 * ops_per_task,
        "all 8 tasks * 10k ops should succeed"
    );
    assert_eq!(total_errors, 0);

    drop_table(&pool_arc, &table_arc).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_savepoint_nested() {
    let pool = setup_pool().await;
    let table = unique_table("t_sp");
    create_test_table(&pool, &table).await;

    let mut tx = pool.begin().await.expect("begin");
    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("outer")
        .bind(1i64)
        .bind("o")
        .execute(&mut *tx)
        .await
        .expect("outer insert");

    sqlx::query(sqlx::AssertSqlSafe("SAVEPOINT sp1"))
        .execute(&mut *tx)
        .await
        .expect("sp1");
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("inner1")
        .bind(2i64)
        .bind("i1")
        .execute(&mut *tx)
        .await
        .expect("inner1 insert");
    sqlx::query(sqlx::AssertSqlSafe("ROLLBACK TO SAVEPOINT sp1"))
        .execute(&mut *tx)
        .await
        .expect("rollback sp1");
    sqlx::query(sqlx::AssertSqlSafe("RELEASE SAVEPOINT sp1"))
        .execute(&mut *tx)
        .await
        .expect("release sp1");

    sqlx::query(sqlx::AssertSqlSafe("SAVEPOINT sp2"))
        .execute(&mut *tx)
        .await
        .expect("sp2");
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("inner2")
        .bind(3i64)
        .bind("i2")
        .execute(&mut *tx)
        .await
        .expect("inner2 insert");
    sqlx::query(sqlx::AssertSqlSafe("RELEASE SAVEPOINT sp2"))
        .execute(&mut *tx)
        .await
        .expect("release sp2");

    tx.commit().await.expect("commit");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 2, "should have outer + inner2 (inner1 rolled back)");

    let names_sql = format!("SELECT name FROM \"{}\" ORDER BY id", table);
    let names: Vec<(String,)> = sqlx::query_as(sqlx::AssertSqlSafe(names_sql.as_str()))
        .fetch_all(&pool)
        .await
        .unwrap();
    let names: Vec<String> = names.into_iter().map(|(n,)| n).collect();
    assert_eq!(names, vec!["outer".to_string(), "inner2".to_string()]);

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_returning_clause() {
    let pool = setup_pool().await;
    let table = unique_table("t_ret");
    create_test_table(&pool, &table).await;

    // PG 支持 RETURNING，验证方言声明
    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    assert!(dialect.supports_returning());

    let sql = format!(
        "INSERT INTO \"{}\" (name, value, data) VALUES ($1, $2, $3) RETURNING id",
        table
    );
    let row: (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(sql.as_str()))
        .bind("returning_test")
        .bind(1i64)
        .bind("rt")
        .fetch_one(&pool)
        .await
        .expect("insert returning");
    assert!(row.0 > 0, "RETURNING should return generated id");

    drop_table(&pool, &table).await;
}

// =============================================================================
// Upsert 集成测试（PostgreSQL ON CONFLICT DO UPDATE）
// =============================================================================

struct DummyModel;

impl Model for DummyModel {
    type PrimaryKey = i64;
    fn table_name() -> &'static str {
        "dummy"
    }
    fn pk(&self) -> Self::PrimaryKey {
        0
    }
    fn set_pk(&mut self, _pk: Self::PrimaryKey) {}
}

impl ModelExt for DummyModel {
    fn columns() -> Vec<&'static str> {
        vec!["id", "name", "age", "email"]
    }
    fn fillable() -> Vec<&'static str> {
        vec!["name", "age", "email"]
    }
    fn guarded() -> Vec<&'static str> {
        vec!["id"]
    }
    fn hidden() -> Vec<&'static str> {
        vec![]
    }
    fn relations() -> HashMap<&'static str, sz_orm_core::Relation> {
        HashMap::new()
    }
    fn fill(&mut self, _data: HashMap<String, Value>) {}
    fn to_json(&self) -> serde_json::Value {
        serde_json::json!({})
    }
}

fn row_for_upsert(id: i64, name: &str, age: i32, email: &str) -> HashMap<String, Value> {
    let mut row = HashMap::new();
    row.insert("id".to_string(), Value::I64(id));
    row.insert("name".to_string(), Value::String(name.to_string()));
    row.insert("age".to_string(), Value::I32(age));
    row.insert("email".to_string(), Value::String(email.to_string()));
    row
}

async fn create_upsert_table(pool: &PgPool, table: &str) {
    let sql = format!(
        "CREATE TABLE \"{}\" (
            id    BIGINT NOT NULL PRIMARY KEY,
            name  VARCHAR(255) NOT NULL UNIQUE,
            age   INTEGER NOT NULL,
            email VARCHAR(255)
        )",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(sql.as_str()))
        .execute(pool)
        .await
        .expect("create upsert table");
}

fn bind_value_pg<'q>(
    q: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
    v: &'q Value,
) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
    match v {
        Value::I32(n) => q.bind(n),
        Value::I64(n) => q.bind(n),
        Value::U32(n) => q.bind(*n as i64),
        Value::U64(n) => q.bind(*n as i64),
        Value::F64(f) => q.bind(f),
        Value::Bool(b) => q.bind(b),
        Value::String(s) => q.bind(s.as_str()),
        Value::Null => q.bind(None::<String>),
        _ => q.bind(v.to_param().to_string()),
    }
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_basic_insert_path() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_bi");
    create_upsert_table(&pool, &table).await;

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let rows = vec![row_for_upsert(1, "Alice", 30, "alice@t.com")];
    let (sql, params) = builder
        .build_batch_upsert_with_params(&rows, &["id"], &[])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool).await.expect("execute upsert insert");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 1, "首次 upsert 应插入 1 行");

    let sel = format!("SELECT name, age, email FROM \"{}\" WHERE id = 1", table);
    let (name, age, email): (String, i64, String) =
        sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(name, "Alice");
    assert_eq!(age, 30);
    assert_eq!(email, "alice@t.com");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_conflict_update_path() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_cu");
    create_upsert_table(&pool, &table).await;

    let ins = format!(
        "INSERT INTO \"{}\" (id, name, age, email) VALUES ($1, $2, $3, $4)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(ins.as_str()))
        .bind(1i64)
        .bind("Alice")
        .bind(30i32)
        .bind("alice@old.com")
        .execute(&pool)
        .await
        .expect("seed insert");

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let rows = vec![row_for_upsert(1, "Alice", 31, "alice@new.com")];
    let (sql, params) = builder
        .build_batch_upsert_with_params(&rows, &["id"], &[])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool).await.expect("execute upsert update");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 1, "冲突时应更新而非插入新行");

    let sel = format!("SELECT age, email FROM \"{}\" WHERE id = 1", table);
    let (age, email): (i64, String) = sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(age, 31, "age 应被更新");
    assert_eq!(email, "alice@new.com", "email 应被更新");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_batch_mixed_insert_update() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_mx");
    create_upsert_table(&pool, &table).await;

    let ins = format!(
        "INSERT INTO \"{}\" (id, name, age, email) VALUES ($1, $2, $3, $4)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(ins.as_str()))
        .bind(1i64)
        .bind("Alice")
        .bind(30i32)
        .bind("alice@old.com")
        .execute(&pool)
        .await
        .expect("seed");

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let rows = vec![
        row_for_upsert(1, "Alice", 31, "alice@new.com"),
        row_for_upsert(2, "Bob", 25, "bob@t.com"),
        row_for_upsert(3, "Carol", 28, "carol@t.com"),
    ];
    let (sql, params) = builder
        .build_batch_upsert_with_params(&rows, &["id"], &[])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool).await.expect("execute batch upsert");

    let count_sql = format!("SELECT COUNT(*) FROM \"{}\"", table);
    let (count,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(count_sql.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 3, "应有 3 行（1 更新 + 2 新增）");

    let sel = format!("SELECT age FROM \"{}\" WHERE id = 1", table);
    let (age,): (i64,) = sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(age, 31, "id=1 的 age 应被更新为 31");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_specific_update_columns_only() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_sc");
    create_upsert_table(&pool, &table).await;

    let ins = format!(
        "INSERT INTO \"{}\" (id, name, age, email) VALUES ($1, $2, $3, $4)",
        table
    );
    sqlx::query(sqlx::AssertSqlSafe(ins.as_str()))
        .bind(1i64)
        .bind("Alice")
        .bind(30i32)
        .bind("alice@old.com")
        .execute(&pool)
        .await
        .expect("seed");

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let rows = vec![row_for_upsert(1, "Alice-Changed", 99, "alice@ignored.com")];
    let (sql, params) = builder
        .build_batch_upsert_with_params(&rows, &["id"], &["age"])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool)
        .await
        .expect("execute upsert specific cols");

    let sel = format!("SELECT name, age, email FROM \"{}\" WHERE id = 1", table);
    let (name, age, email): (String, i64, String) =
        sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
            .fetch_one(&pool)
            .await
            .unwrap();
    assert_eq!(name, "Alice", "name 不应被更新（不在 update_columns 中）");
    assert_eq!(age, 99, "age 应被更新");
    assert_eq!(
        email, "alice@old.com",
        "email 不应被更新（不在 update_columns 中）"
    );

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_null_value_handling() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_nl");
    create_upsert_table(&pool, &table).await;

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let mut row = row_for_upsert(1, "Alice", 30, "");
    row.insert("email".to_string(), Value::Null);
    let (sql, params) = builder
        .build_batch_upsert_with_params(&[row], &["id"], &[])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool).await.expect("execute upsert with null");

    let sel = format!("SELECT email FROM \"{}\" WHERE id = 1", table);
    let (email,): (Option<String>,) = sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert!(email.is_none(), "email 应为 NULL");

    drop_table(&pool, &table).await;
}

#[tokio::test]
#[ignore = "需要 PostgreSQL 18 运行于 127.0.0.1:5432"]
async fn test_pg_upsert_unicode_and_special_chars() {
    let pool = setup_pool().await;
    let table = unique_table("t_ups_un");
    create_upsert_table(&pool, &table).await;

    let dialect = get_dialect(DbType::PostgreSQL).expect("pg dialect");
    let builder = QueryBuilder::<DummyModel>::new(dialect).table(&table);

    let rows = vec![row_for_upsert(
        1,
        "张三-汉字",
        30,
        "test'with`special\"chars",
    )];
    let (sql, params) = builder
        .build_batch_upsert_with_params(&rows, &["id"], &[])
        .expect("build upsert sql");

    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql.as_str()));
    for v in &params {
        q = bind_value_pg(q, v);
    }
    q.execute(&pool)
        .await
        .expect("execute upsert unicode/special");

    let sel = format!("SELECT name, email FROM \"{}\" WHERE id = 1", table);
    let (name, email): (String, String) = sqlx::query_as(sqlx::AssertSqlSafe(sel.as_str()))
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(name, "张三-汉字", "Unicode 名字应正确存储");
    assert_eq!(
        email, "test'with`special\"chars",
        "特殊字符应通过参数化查询正确存储"
    );

    drop_table(&pool, &table).await;
}