sz-orm-core 1.1.0

//! 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::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use sz_orm_core::dialect::{get_dialect, ColumnDef};
use sz_orm_core::DbType;
use sz_orm_core::Value;

/// 默认 PostgreSQL 连接 URL（本机）；可通过环境变量 `SZ_ORM_PG_URL` 覆盖以指向真实云数据库。
const PG_URL_DEFAULT: &str = "postgres://postgres:<your-password>@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;
}