sz_orm_macros/

lib.rs

//! SZ-ORM Procedural Macros - compile-time SQL validation & derive macros
//!
//! Provides:
//! - `sql_string!` macro that validates SQL string literals at compile time.
//!   Errors like `SELECT * FORM users` or `'; DROP TABLE` are caught before the binary is built.
//! - `#[derive(Schema)]` — auto-generate table structure info from a struct.
//! - `#[derive(Builder)]` — auto-generate builder pattern code for a struct.
//!
//! # Usage
//!
//! ```ignore
//! use sz_orm_macros::sql_string;
//!
//! // Basic usage
//! let sql = sql_string!("SELECT * FROM users WHERE id = 1"); // ✅ compiles
//!
//! // With parameter count check
//! let sql = sql_string!("SELECT * FROM users WHERE id = ?");
//!                      params: 1);                          // ✅ compiles
//!
//! // ❌ compile error: missing FROM
//! let sql = sql_string!("SELECT * users WHERE id = 1");
//!
//! // ❌ compile error: SQL injection detected
//! let sql = sql_string!("SELECT * FROM users WHERE name = 'x' OR '1'='1'");
//!
//! // ❌ compile error: parameter count mismatch
//! let sql = sql_string!("SELECT * FROM users WHERE id = ?");
//!                      params: 2);
//! ```

// 抑制 Windows 链接器输出"正在创建库 ..."的诊断信息被识别为警告：
// 该输出是 link.exe 创建 DLL 导入库时的正常 stdout 提示，并非代码问题。
#![allow(linker_messages)]
//!
//! # Derive macros
//!
//! ```ignore
//! use sz_orm_macros::{Schema, Builder};
//!
//! #[derive(Schema)]
//! #[table(name = "users")]
//! struct User {
//!     #[column(primary_key)]
//!     id: i64,
//!     name: String,
//! }
//!
//! #[derive(Builder)]
//! struct Order {
//!     id: i64,
//!     total: f64,
//! }
//! ```

extern crate proc_macro;

use proc_macro::{Delimiter, Group, Ident, Literal, Punct, Spacing, Span, TokenStream, TokenTree};

#[cfg(feature = "db-verify")]
use sqlx::Row as _;

// 引入 quote! 宏，用于类型安全地构建 TokenStream
use proc_macro2::TokenStream as TokenStream2;
use quote::quote;
use syn::parse_macro_input;

// 派生宏模块
mod derive;

// v3.9.0：数据验证派生宏模块（data-validation feature 隔离）
#[cfg(feature = "data-validation")]
mod derive_validate;

// 自定义编译期诊断模块（typed-dsl 或 custom-diagnostic feature 隔离）
#[cfg(any(feature = "typed-dsl", feature = "custom-diagnostic"))]
mod diagnostic;

// ---- 自定义编译期诊断宏（typed-dsl 或 custom-diagnostic feature 隔离）----

/// `#[type_check]` attribute macro — adds type-check diagnostic hints to a function
///
/// When a type constraint inside the function body fails, generates clearer
/// diagnostic messages than Rust's defaults.
///
/// # Example
///
/// ```ignore
/// #[type_check]
/// fn my_query() {
///     let expr = ColId.eq("hello"); // i64 column compared with String
///     // Compile-time will generate custom diagnostic information
/// }
/// ```
#[cfg(any(feature = "typed-dsl", feature = "custom-diagnostic"))]
#[proc_macro_attribute]
pub fn type_check(_attr: TokenStream, item: TokenStream) -> TokenStream {
    let input_fn = parse_macro_input!(item as syn::ItemFn);
    let fn_name = input_fn.sig.ident.to_string();
    let fn_vis = &input_fn.vis;
    let fn_sig = &input_fn.sig;
    let fn_block = &input_fn.block;

    let expanded = quote! {
        #[doc = concat!("Type-check function `", #fn_name, "`: if compilation fails, please verify type constraints")]
        #fn_vis #fn_sig {
            #fn_block
        }
    };

    TokenStream::from(expanded)
}

/// `diagnostic_error!` macro — generates a compile-time error with suggestion information
///
/// # Example
///
/// ```ignore
/// diagnostic_error!("type mismatch", "please use Cast for explicit conversion");
/// ```
#[cfg(any(feature = "typed-dsl", feature = "custom-diagnostic"))]
#[proc_macro]
pub fn diagnostic_error(input: TokenStream) -> TokenStream {
    let mut iter = input.into_iter().peekable();

    let msg = match iter.next() {
        Some(TokenTree::Literal(l)) => l.to_string(),
        _ => {
            return "compile_error!(\"diagnostic_error! requires string literal arguments\")"
                .parse()
                .unwrap()
        }
    };

    let suggestion: Option<String> = match iter.next() {
        Some(TokenTree::Punct(p)) if p.as_char() == ',' => match iter.next() {
            Some(TokenTree::Literal(l)) => Some(l.to_string()),
            _ => None,
        },
        _ => None,
    };

    let stripped_msg = diagnostic::strip_quotes(&msg);
    let full_msg = if let Some(sug) = suggestion.as_ref() {
        let stripped_sug = diagnostic::strip_quotes(sug);
        format!("{}\n  help: {}", stripped_msg, stripped_sug)
    } else {
        stripped_msg.to_string()
    };

    let err = format!("compile_error!({:?})", full_msg);
    err.parse().unwrap_or_else(|_| {
        "compile_error!(\"internal error in diagnostic macro\")"
            .parse()
            .unwrap()
    })
}

/// Compile-time SQL validation macro.
///
/// Validates SQL syntax at compile time and emits the validated SQL string.
///
/// # Syntax
///
/// - `sql_string!("SQL")` — validates the SQL and emits it as a `&str`
/// - `sql_string!("SQL"; params: N)` — additionally checks that the SQL has exactly N parameters
///
/// # Validation rules
///
/// - SELECT must contain FROM
/// - INSERT must contain INTO and VALUES
/// - UPDATE must contain SET
/// - DELETE must contain FROM
/// - Parentheses must be balanced
/// - String literals must be properly closed
/// - No SQL injection patterns (OR '1'='1', UNION SELECT, `'; DROP TABLE`, `--`, `/*`)
/// - Table/column identifiers must be valid
#[proc_macro]
pub fn sql_string(input: TokenStream) -> TokenStream {
    let mut tokens = input.into_iter().peekable();

    // Parse the SQL string literal
    let sql = match tokens.next() {
        Some(TokenTree::Literal(lit)) => lit.to_string(),
        Some(other) => {
            return compile_error(
                other.span(),
                "Expected a string literal as the first argument to sql_string!",
            );
        }
        None => {
            return compile_error(
                Span::call_site(),
                "Expected a string literal argument to sql_string!",
            );
        }
    };

    // Remove surrounding quotes from the string literal
    let sql_content = if sql.starts_with("r#\"") {
        &sql[3..sql.len() - 2]
    } else if sql.starts_with("r\"") {
        &sql[2..sql.len() - 1]
    } else if sql.starts_with('"') {
        &sql[1..sql.len() - 1]
    } else if sql.starts_with("b\"") || sql.starts_with("b\'") {
        &sql[2..sql.len() - 1]
    } else {
        return compile_error(
            Span::call_site(),
            "sql_string! requires a string literal argument",
        );
    };

    // Parse optional `params: N`
    let mut expected_params = None;
    if tokens.peek().is_some() {
        // Expect `; params: N`
        match tokens.next() {
            Some(TokenTree::Punct(p)) if p.as_char() == ';' => {}
            Some(other) => {
                return compile_error(
                    other.span(),
                    "Expected `;` before param count, e.g. sql_string!(\"...\"; params: 2)",
                );
            }
            None => {}
        }

        // Parse `params`
        match tokens.next() {
            Some(TokenTree::Ident(id)) if id.to_string() == "params" => {}
            Some(other) => {
                return compile_error(
                    other.span(),
                    "Expected `params:` keyword, e.g. sql_string!(\"...\"; params: 2)",
                );
            }
            None => {
                return compile_error(Span::call_site(), "Expected param count after `;`");
            }
        }

        // Parse `:`
        match tokens.next() {
            Some(TokenTree::Punct(p)) if p.as_char() == ':' => {}
            Some(other) => {
                return compile_error(
                    other.span(),
                    "Expected `:` after `params`, e.g. sql_string!(\"...\"; params: 2)",
                );
            }
            None => {
                return compile_error(Span::call_site(), "Expected param count after `params`");
            }
        }

        // Parse the number
        match tokens.next() {
            Some(TokenTree::Literal(lit)) => {
                let num_str = lit.to_string();
                if let Ok(n) = num_str.parse::<usize>() {
                    expected_params = Some(n);
                } else {
                    return compile_error(
                        lit.span(),
                        "Expected a positive integer for param count",
                    );
                }
            }
            Some(other) => {
                return compile_error(
                    other.span(),
                    "Expected a number after `params:`, e.g. sql_string!(\"...\"; params: 2)",
                );
            }
            None => {
                return compile_error(Span::call_site(), "Expected a number after `params:`");
            }
        }
    }

    // Run validation
    if let Err(err_msg) = validate_sql_content(sql_content, expected_params) {
        return compile_error(Span::call_site(), &err_msg);
    }

    // Emit the validated string as a &str literal
    let output = format!("\"{}\"", sql_content.escape_default());
    output
        .parse()
        .unwrap_or_else(|_| compile_error(Span::call_site(), "Failed to generate output token"))
}

// ---------------------------------------------------------------------------
// Validation logic (self-contained, no external dependencies)
// ---------------------------------------------------------------------------

fn validate_sql_content(sql: &str, expected_params: Option<usize>) -> Result<(), String> {
    let trimmed = sql.trim();
    if trimmed.is_empty() {
        return Err("SQL statement is empty".to_string());
    }

    validate_balanced_parens(trimmed)?;
    validate_string_literals_closed(trimmed)?;
    validate_no_injection(trimmed)?;

    // Type-specific validation
    let sql_upper = trimmed.to_uppercase();
    if sql_upper.starts_with("SELECT") {
        if !sql_upper.contains("FROM") {
            return Err("SELECT statement missing FROM clause".to_string());
        }
    } else if sql_upper.starts_with("INSERT") {
        if !sql_upper.contains("INTO") {
            return Err("INSERT statement missing INTO clause".to_string());
        }
        if !sql_upper.contains("VALUES") {
            return Err("INSERT statement missing VALUES clause".to_string());
        }
    } else if sql_upper.starts_with("UPDATE") {
        if !sql_upper.contains("SET") {
            return Err("UPDATE statement missing SET clause".to_string());
        }
    } else if sql_upper.starts_with("DELETE") && !sql_upper.contains("FROM") {
        return Err("DELETE statement missing FROM clause".to_string());
    }

    // Parameter count check
    if let Some(expected) = expected_params {
        let actual = sql.chars().filter(|&c| c == '?').count();
        if actual != expected {
            return Err(format!(
                "Parameter count mismatch: expected {} parameters, found {}",
                expected, actual
            ));
        }
    }

    Ok(())
}

fn validate_balanced_parens(sql: &str) -> Result<(), String> {
    let mut depth: i32 = 0;
    for (i, ch) in sql.char_indices() {
        match ch {
            '(' => depth += 1,
            ')' => {
                depth -= 1;
                if depth < 0 {
                    return Err(format!(
                        "Unbalanced parentheses: unexpected ')' at position {}",
                        i
                    ));
                }
            }
            _ => {}
        }
    }
    if depth != 0 {
        return Err(format!("Unbalanced parentheses: {} unclosed '('", depth));
    }
    Ok(())
}

fn validate_string_literals_closed(sql: &str) -> Result<(), String> {
    let mut in_single = false;
    let mut in_double = false;
    let mut prev = '\0';

    for ch in sql.chars() {
        if prev == '\\' {
            prev = ch;
            continue;
        }

        match ch {
            '\'' if !in_double => in_single = !in_single,
            '"' if !in_single => in_double = !in_double,
            _ => {}
        }
        prev = ch;
    }

    if in_single {
        return Err("Unclosed single-quoted string literal".to_string());
    }
    if in_double {
        return Err("Unclosed double-quoted string literal".to_string());
    }

    Ok(())
}

fn validate_no_injection(sql: &str) -> Result<(), String> {
    let sql_lower = sql.to_lowercase();

    // 注意：编译期 SQL 内容已由 Rust 字符串字面量解析剥离外层引号，
    // 因此检测模式不应依赖前导引号字符（如 `"'; DROP TABLE"`）。
    let injection_patterns: &[&str] = &[
        // 多语句攻击
        "drop table",
        "drop database",
        "; drop",
        // 经典注入
        "or 1=1",
        "or 1 = 1",
        "union select",
        "union all select",
        // 注释攻击
        "--",
        "/*",
        "*/",
        // 存储过程注入
        "xp_cmdshell",
        "sp_executesql",
        "exec(",
        "execute(",
        // 信息泄露
        "information_schema",
        "sys.tables",
        "sys.columns",
    ];

    for pattern in injection_patterns {
        if sql_lower.contains(pattern) {
            return Err(format!("潜在的 SQL 注入模式被检测到: '{}'", pattern));
        }
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// `query!` macro — optional real DB verification (gated by `db-verify` feature)
// ---------------------------------------------------------------------------

/// Compile-time SQL validation with optional real DB verification.
///
/// Behavior:
/// - Always runs the same syntax validation as `sql_string!`.
/// - When the `db-verify` cargo feature is enabled **AND** the
///   `SZ_ORM_QUERY_VERIFY=1` environment variable is set at compile time,
///   connects to the database pointed to by `DATABASE_URL` and runs
///   `EXPLAIN` (MySQL/PostgreSQL) or `EXPLAIN QUERY PLAN` (SQLite) to verify
///   the SQL is valid against the actual schema (column names, table names,
///   joins, etc.).
/// - Otherwise, falls back to syntax-only validation.
///
/// Emits a [`sz_orm_core::queryable::Query`] object wrapping the validated SQL.
///
/// # Syntax
///
/// ```ignore
/// use sz_orm_core::queryable::Query;
/// let q = query!("SELECT id, name FROM users WHERE id = ?");
/// let rows = q.fetch_all(&mut conn).await?;
/// ```
///
/// # Verification setup
///
/// ```bash
/// export DATABASE_URL="mysql://user:pass@host:3306/db"
/// export SZ_ORM_QUERY_VERIFY=1
/// cargo build --features sz-orm-macros/db-verify
/// ```
#[proc_macro]
pub fn query(input: TokenStream) -> TokenStream {
    let mut tokens = input.into_iter().peekable();

    // P0-1：支持可选类型参数 `query!(T, "SQL")` → `QueryAs::<T>::new(sql)`
    // 若无类型参数则保持 `query!("SQL")` → `Query::new(sql)`
    let type_param: Option<TokenStream2> = match tokens.peek() {
        Some(TokenTree::Ident(_)) | Some(TokenTree::Punct(_)) => {
            // 收集类型路径（如 `User` 或 `crate::User`）
            let mut ty_tokens = Vec::new();
            while let Some(tok) = tokens.peek() {
                match tok {
                    TokenTree::Punct(p) if p.as_char() == ',' => break,
                    TokenTree::Punct(p) if p.as_char() == ':' => {
                        ty_tokens.push(tokens.next().unwrap());
                        // 消耗 `:`
                        if let Some(TokenTree::Punct(p2)) = tokens.peek() {
                            if p2.as_char() == ':' {
                                ty_tokens.push(tokens.next().unwrap());
                            }
                        }
                    }
                    _ => ty_tokens.push(tokens.next().unwrap()),
                }
            }
            // 确认下一个 token 是逗号（类型参数分隔符）
            match tokens.peek() {
                Some(TokenTree::Punct(p)) if p.as_char() == ',' => {
                    tokens.next(); // 消耗逗号
                    let ts: proc_macro::TokenStream = ty_tokens.into_iter().collect();
                    Some(TokenStream2::from(ts))
                }
                _ => None, // 不是类型参数，回退
            }
        }
        _ => None,
    };

    // Parse the SQL string literal
    let sql = match tokens.next() {
        Some(TokenTree::Literal(lit)) => lit.to_string(),
        Some(other) => {
            return compile_error(
                other.span(),
                if type_param.is_some() {
                    "query!(T, \"SQL\"): expected a string literal as the second argument"
                } else {
                    "Expected a string literal as the first argument to query!"
                },
            );
        }
        None => {
            return compile_error(
                Span::call_site(),
                if type_param.is_some() {
                    "query!(T, \"SQL\"): missing SQL string argument"
                } else {
                    "Expected a string literal argument to query!"
                },
            );
        }
    };

    let sql_content = match strip_string_literal(&sql) {
        Some(s) => s,
        None => {
            return compile_error(
                Span::call_site(),
                "query! requires a string literal argument",
            );
        }
    };

    // Syntax validation (shared with sql_string!)
    if let Err(err_msg) = validate_sql_content(sql_content, None) {
        return compile_error(Span::call_site(), &err_msg);
    }

    // Optional real DB verification (only when feature is enabled)
    #[cfg(feature = "db-verify")]
    let verify_cols: Option<Vec<(String, String)>> = {
        match std::env::var("SZ_ORM_QUERY_VERIFY").ok().as_deref() {
            // 模式 1：连真 DB 执行 EXPLAIN 验证（需 DATABASE_URL），并获取 SELECT 列的实际类型
            Some("1") => match verify_with_real_db(sql_content) {
                Ok(cols) => {
                    // v4.3.0 M1-T3：EXPLAIN 分析 → 编译期性能警告（非阻断）
                    for warning in analyze_explain(sql_content) {
                        eprintln!("warning: [sz-orm-explain] {warning}");
                    }
                    Some(cols)
                }
                Err(err) => {
                    return compile_error(
                        Span::call_site(),
                        &format!("query! real DB verification failed: {}", err),
                    )
                }
            },
            // 模式 cache：从离线缓存文件查找（无需 DB，适合 CI）
            Some("cache") => {
                if let Err(err) = verify_with_cache(sql_content) {
                    return compile_error(
                        Span::call_site(),
                        &format!("query! offline cache verification failed: {}", err),
                    );
                }
                None
            }
            _ => None,
        }
    };
    #[cfg(not(feature = "db-verify"))]
    let _verify_cols: Option<Vec<(String, String)>> = None;

    // Emit the appropriate query object
    let escaped = sql_content.escape_default().to_string();
    let base = if let Some(ref ty) = type_param {
        // query!(T, "SQL") → QueryAs::<T>::new("SQL")
        format!(
            "::sz_orm_core::queryable::QueryAs::<{}>::new(\"{}\")",
            ty, escaped
        )
    } else {
        // query!("SQL") → Query::new("SQL")
        format!("::sz_orm_core::queryable::Query::new(\"{}\")", escaped)
    };
    // db-verify 通过且有类型参数时，附加编译期类型验证块（P0-2）
    #[cfg(feature = "db-verify")]
    let output = match (&verify_cols, &type_param) {
        (Some(cols), Some(ty)) if !cols.is_empty() => {
            gen_compile_time_type_check(&ty.to_string(), sql_content, cols, &base)
        }
        _ => base,
    };
    #[cfg(not(feature = "db-verify"))]
    let output = base;
    output
        .parse()
        .unwrap_or_else(|_| compile_error(Span::call_site(), "Failed to generate query! output"))
}

/// Strip surrounding quotes from a string literal token's raw representation.
/// Shared by `sql_string!` and `query!`.
fn strip_string_literal(raw: &str) -> Option<&str> {
    if raw.starts_with("r#\"") {
        Some(&raw[3..raw.len() - 2])
    } else if raw.starts_with("r\"") {
        Some(&raw[2..raw.len() - 1])
    } else if raw.starts_with('"') {
        Some(&raw[1..raw.len() - 1])
    } else if raw.starts_with("b\"") || raw.starts_with("b\'") {
        Some(&raw[2..raw.len() - 1])
    } else {
        None
    }
}

// ---------------------------------------------------------------------------
// Real DB verification (only compiled when `db-verify` feature is enabled)
// ---------------------------------------------------------------------------

#[cfg(feature = "db-verify")]
fn verify_with_real_db(sql: &str) -> Result<Vec<(String, String)>, String> {
    let dsn = std::env::var("DATABASE_URL")
        .map_err(|_| "DATABASE_URL environment variable not set".to_string())?;

    let db_kind =
        detect_db_kind(&dsn).map_err(|e| format!("Failed to detect DB kind from DSN: {}", e))?;

    // 将 ? 占位符替换为 NULL，使 EXPLAIN 无需绑定参数即可执行。
    // EXPLAIN 不实际执行查询，NULL 对所有列类型都合法。
    let sql_no_placeholders = replace_placeholders_with_null(sql);

    // Oracle/SQL Server 使用 EXPLAIN PLAN FOR（不同语法），其余用 EXPLAIN
    let explain_sql = match db_kind {
        DbKind::MySql | DbKind::Postgres => format!("EXPLAIN {}", sql_no_placeholders),
        DbKind::Sqlite => format!("EXPLAIN QUERY PLAN {}", sql_no_placeholders),
        // Oracle: EXPLAIN PLAN FOR 放入 PLAN_TABLE，再查询结果验证语法
        DbKind::Oracle => format!("EXPLAIN PLAN FOR {}", sql_no_placeholders),
        // SQL Server: SET SHOWPLAN_TEXT ON 后执行（不实际运行）
        DbKind::SqlServer => sql_no_placeholders,
    };

    // MySQL/PG/SQLite 走 sqlx 异步路径
    if matches!(db_kind, DbKind::MySql | DbKind::Postgres | DbKind::Sqlite) {
        let rt = tokio::runtime::Runtime::new()
            .map_err(|e| format!("Failed to create tokio runtime: {}", e))?;
        return rt.block_on(async {
            // 1. EXPLAIN 语法验证
            if let DbKind::MySql = db_kind {
                verify_mysql(&dsn, &explain_sql).await?;
            } else if let DbKind::Postgres = db_kind {
                verify_postgres(&dsn, &explain_sql).await?;
            } else {
                // 由外层 if matches! 保证此处必为 Sqlite
                verify_sqlite(&dsn, &explain_sql).await?;
            }
            // 2. 列名/类型验证（Gap 1 修复）
            verify_columns(&dsn, db_kind, sql).await?;
            // 3. 获取 SELECT 列的实际 DB 类型（供编译期类型验证使用）
            //    SQLite/Oracle/SQL Server 返回空列表（跳过类型级验证）
            fetch_column_types(&dsn, db_kind, sql).await
        });
    }

    // Oracle/SQL Server 走命令行工具验证（避免引入重依赖）
    if let DbKind::Oracle = db_kind {
        verify_oracle(&dsn, &explain_sql).map(|_| Vec::new())
    } else {
        // 由外层 if matches! 保证此处必为 SqlServer
        verify_sqlserver(&dsn, &explain_sql).map(|_| Vec::new())
    }
}

// ---------------------------------------------------------------------------
// v4.3.0 M1-T3：EXPLAIN 结果分析（编译期性能警告，非阻断）
// 在 db-verify 语法/列验证通过后，进一步解析 EXPLAIN 输出，检测
// 全表扫描与缺失索引，输出编译期警告（eprintln，stable Rust 无 Span::warning）。
// ---------------------------------------------------------------------------

/// Analyzes EXPLAIN output and returns a list of warning messages.
///
/// Only runs analysis on the sqlx path (MySQL/PostgreSQL/SQLite); the
/// Oracle/SQL Server command-line verification path is skipped. On parse
/// failure it degrades to an empty list (EXPLAIN syntax has already been
/// validated by [`verify_with_real_db`], so this does not block compilation).
#[cfg(feature = "db-verify")]
fn analyze_explain(sql: &str) -> Vec<String> {
    let Ok(dsn) = std::env::var("DATABASE_URL") else {
        return Vec::new();
    };
    let Ok(db_kind) = detect_db_kind(&dsn) else {
        return Vec::new();
    };
    if !matches!(db_kind, DbKind::MySql | DbKind::Postgres | DbKind::Sqlite) {
        return Vec::new();
    }
    let sql_no_placeholders = replace_placeholders_with_null(sql);
    let explain_sql = match db_kind {
        DbKind::MySql | DbKind::Postgres => format!("EXPLAIN {}", sql_no_placeholders),
        DbKind::Sqlite => format!("EXPLAIN QUERY PLAN {}", sql_no_placeholders),
        _ => return Vec::new(),
    };
    let rt = match tokio::runtime::Runtime::new() {
        Ok(rt) => rt,
        Err(_) => return Vec::new(),
    };
    rt.block_on(explain_analysis_raw(&dsn, db_kind, &explain_sql))
        .unwrap_or_default()
}

/// Fetches EXPLAIN raw output and parses it into a list of warnings
#[cfg(feature = "db-verify")]
async fn explain_analysis_raw(
    dsn: &str,
    db_kind: DbKind,
    explain_sql: &str,
) -> Result<Vec<String>, String> {
    let raw = match db_kind {
        DbKind::MySql => fetch_explain_mysql(dsn, explain_sql).await?,
        DbKind::Postgres => fetch_explain_postgres(dsn, explain_sql).await?,
        DbKind::Sqlite => fetch_explain_sqlite(dsn, explain_sql).await?,
        _ => return Ok(Vec::new()),
    };
    let db_type = match db_kind {
        DbKind::MySql => sz_orm_explain::ExplainDialect::MySql,
        DbKind::Postgres => sz_orm_explain::ExplainDialect::Postgres,
        DbKind::Sqlite => sz_orm_explain::ExplainDialect::Sqlite,
        _ => return Ok(Vec::new()),
    };
    let parser = sz_orm_explain::parser_for(db_type)
        .map_err(|e| format!("no explain parser for db: {e}"))?;
    let plan = parser
        .parse(&raw)
        .map_err(|e| format!("explain parse failed: {e}"))?;

    let mut warnings = Vec::new();
    if plan.scan_type.is_full_table_scan() {
        warnings.push(format!(
            "full table scan detected on table '{}': consider adding an index",
            plan.table
        ));
    } else {
        // 行数阈值：SZ_ORM_EXPLAIN_ROW_THRESHOLD 环境变量，默认 1000
        let threshold: u64 = std::env::var("SZ_ORM_EXPLAIN_ROW_THRESHOLD")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(1000);
        if plan.missing_index(threshold) {
            warnings.push(format!(
                "missing index on table '{}': estimated rows = {} exceeds threshold {}",
                plan.table, plan.rows, threshold
            ));
        }
    }
    Ok(warnings)
}

/// MySQL: EXPLAIN table rows → `| a | b | ... |` text (compatible with MySqlParser input)
#[cfg(feature = "db-verify")]
async fn fetch_explain_mysql(dsn: &str, explain_sql: &str) -> Result<String, String> {
    let pool = sqlx::MySqlPool::connect(dsn)
        .await
        .map_err(|e| format!("MySQL connect failed: {e}"))?;
    let rows = sqlx::query(sqlx::AssertSqlSafe(explain_sql))
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("MySQL EXPLAIN failed: {e}"))?;
    let mut out = String::new();
    for row in &rows {
        let cells: Vec<String> = (0..row.columns().len())
            .map(|i| row.try_get::<String, _>(i).unwrap_or_default())
            .collect();
        out.push_str(&format!("| {} |\n", cells.join(" | ")));
    }
    Ok(out)
}

/// PostgreSQL: EXPLAIN text lines concatenated
#[cfg(feature = "db-verify")]
async fn fetch_explain_postgres(dsn: &str, explain_sql: &str) -> Result<String, String> {
    let pool = sqlx::PgPool::connect(dsn)
        .await
        .map_err(|e| format!("PostgreSQL connect failed: {e}"))?;
    let rows: Vec<(String,)> = sqlx::query_as::<_, (String,)>(sqlx::AssertSqlSafe(explain_sql))
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("PostgreSQL EXPLAIN failed: {e}"))?;
    Ok(rows
        .iter()
        .map(|(line,)| line.as_str())
        .collect::<Vec<_>>()
        .join("\n"))
}

/// SQLite: EXPLAIN QUERY PLAN four columns (id/parent/notused/detail) → numeric-prefixed text
#[cfg(feature = "db-verify")]
async fn fetch_explain_sqlite(dsn: &str, explain_sql: &str) -> Result<String, String> {
    let pool = sqlx::SqlitePool::connect(dsn)
        .await
        .map_err(|e| format!("SQLite connect failed: {e}"))?;
    let rows = sqlx::query(sqlx::AssertSqlSafe(explain_sql))
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("SQLite EXPLAIN failed: {e}"))?;
    let mut out = String::new();
    for row in &rows {
        let id: i64 = row.try_get(0).unwrap_or_default();
        let parent: i64 = row.try_get(1).unwrap_or_default();
        let notused: i64 = row.try_get(2).unwrap_or_default();
        let detail: String = row.try_get(3).unwrap_or_default();
        out.push_str(&format!("{id} {parent} {notused} {detail}\n"));
    }
    Ok(out)
}

/// Offline cache verification: looks up verified SQL from the JSON file specified by `SZ_ORM_SQLX_CACHE`.
///
/// The cache file format is a JSON array of strings, one verified SQL per entry:
/// ```json
/// ["SELECT `id`, `name` FROM `users` WHERE `id` = ?", ...]
/// ```
///
/// To generate it: run `cargo build --features db-verify` (with `SZ_ORM_QUERY_VERIFY=1`) in an
/// environment that has DB access, or use the `cargo sz-orm prepare` tool to scan `query!` macros
/// in the project and generate the cache.
///
/// In CI, simply set `SZ_ORM_QUERY_VERIFY=cache` + `SZ_ORM_SQLX_CACHE=.sz-orm/query-cache.json`
/// to perform compile-time SQL verification without connecting to the DB.
#[cfg(feature = "db-verify")]
fn verify_with_cache(sql: &str) -> Result<(), String> {
    let cache_path = std::env::var("SZ_ORM_SQLX_CACHE").map_err(|_| {
        "SZ_ORM_SQLX_CACHE not set. \
             Set it to the path of a JSON file containing verified SQL statements, \
             e.g. SZ_ORM_SQLX_CACHE=.sz-orm/query-cache.json"
            .to_string()
    })?;

    let cache_content = std::fs::read_to_string(&cache_path).map_err(|e| {
        format!(
            "Failed to read cache file '{}': {}. \
             Run `cargo sz-orm prepare` or build with SZ_ORM_QUERY_VERIFY=1 to generate it.",
            cache_path, e
        )
    })?;

    // 支持两种格式：JSON 数组 或 每行一条 SQL 的文本文件
    let verified: Vec<String> = serde_json::from_str(&cache_content).unwrap_or_else(|_| {
        cache_content
            .lines()
            .map(|l| l.trim().to_string())
            .filter(|l| !l.is_empty() && !l.starts_with('#'))
            .collect()
    });

    if verified.iter().any(|v| v.trim() == sql.trim()) {
        Ok(())
    } else {
        Err(format!(
            "SQL not found in offline cache ({} entries): \"{}\". \
             Add it to the cache by running with SZ_ORM_QUERY_VERIFY=1 first.",
            verified.len(),
            truncate_sql(sql, 80)
        ))
    }
}

/// Truncates SQL for error message display
#[cfg(feature = "db-verify")]
fn truncate_sql(sql: &str, max: usize) -> String {
    if sql.len() <= max {
        sql.to_string()
    } else {
        format!("{}...", &sql[..max])
    }
}

#[cfg(feature = "db-verify")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DbKind {
    MySql,
    Postgres,
    Sqlite,
    Oracle,
    SqlServer,
}

/// Replaces `?` placeholders in SQL with `NULL`, skipping `?` inside string literals.
///
/// EXPLAIN does not actually execute the query; using NULL in place of parameters
/// validates syntax and table/column existence while avoiding the sqlx prepared-statement
/// requirement of binding parameters.
#[cfg(feature = "db-verify")]
fn replace_placeholders_with_null(sql: &str) -> String {
    let mut result = String::with_capacity(sql.len() + 16);
    let mut in_single_quote = false;
    let mut in_double_quote = false;
    let mut prev = '\0';

    for ch in sql.chars() {
        if prev == '\\' {
            // 转义字符：直接追加
            result.push(ch);
            prev = ch;
            continue;
        }
        match ch {
            '\'' if !in_double_quote => in_single_quote = !in_single_quote,
            '"' if !in_single_quote => in_double_quote = !in_double_quote,
            '?' if !in_single_quote && !in_double_quote => {
                result.push_str("NULL");
                prev = ch;
                continue;
            }
            _ => {}
        }
        result.push(ch);
        prev = ch;
    }
    result
}

#[cfg(feature = "db-verify")]
fn detect_db_kind(dsn: &str) -> Result<DbKind, String> {
    let lower = dsn.to_lowercase();
    if lower.starts_with("mysql://") {
        Ok(DbKind::MySql)
    } else if lower.starts_with("postgres://") || lower.starts_with("postgresql://") {
        Ok(DbKind::Postgres)
    } else if lower.starts_with("sqlite://") || lower.starts_with("sqlite:") {
        Ok(DbKind::Sqlite)
    } else if lower.starts_with("oracle://") || lower.starts_with("oracle:") {
        Ok(DbKind::Oracle)
    } else if lower.starts_with("sqlserver://")
        || lower.starts_with("mssql://")
        || lower.starts_with("tds://")
    {
        Ok(DbKind::SqlServer)
    } else {
        Err(format!("Unsupported DSN scheme: {}", dsn))
    }
}

#[cfg(feature = "db-verify")]
async fn verify_mysql(dsn: &str, explain_sql: &str) -> Result<(), String> {
    let pool = sqlx::MySqlPool::connect(dsn)
        .await
        .map_err(|e| format!("MySQL connect failed: {}", e))?;
    sqlx::query(sqlx::AssertSqlSafe(explain_sql))
        .execute(&pool)
        .await
        .map_err(|e| format!("MySQL EXPLAIN failed: {}", e))?;
    Ok(())
}

#[cfg(feature = "db-verify")]
async fn verify_postgres(dsn: &str, explain_sql: &str) -> Result<(), String> {
    let pool = sqlx::PgPool::connect(dsn)
        .await
        .map_err(|e| format!("PostgreSQL connect failed: {}", e))?;
    sqlx::query(sqlx::AssertSqlSafe(explain_sql))
        .execute(&pool)
        .await
        .map_err(|e| format!("PostgreSQL EXPLAIN failed: {}", e))?;
    Ok(())
}

#[cfg(feature = "db-verify")]
async fn verify_sqlite(dsn: &str, explain_sql: &str) -> Result<(), String> {
    let pool = sqlx::SqlitePool::connect(dsn)
        .await
        .map_err(|e| format!("SQLite connect failed: {}", e))?;
    sqlx::query(sqlx::AssertSqlSafe(explain_sql))
        .execute(&pool)
        .await
        .map_err(|e| format!("SQLite EXPLAIN failed: {}", e))?;
    Ok(())
}

// ========================================================================
// Gap 1 修复：列名/类型验证（在 EXPLAIN 语法验证通过后执行）
// ========================================================================

/// Extracts table names and column references from SQL and queries information_schema to verify column existence.
///
/// EXPLAIN has already validated syntax and table existence; this function further verifies:
/// - Whether columns referenced in SELECT/WHERE/ORDER BY/GROUP BY exist in the corresponding tables
/// - Does not verify columns qualified by table aliases (handled by EXPLAIN)
/// - Only verifies explicit column names other than `*`
///
/// # Supported DBs
///
/// MySQL / PostgreSQL / SQLite (Oracle/SQL Server skip this step)
#[cfg(feature = "db-verify")]
async fn verify_columns(dsn: &str, db_kind: DbKind, sql: &str) -> Result<(), String> {
    // SQLite 的 information_schema 支持有限，跳过
    if matches!(db_kind, DbKind::Sqlite | DbKind::Oracle | DbKind::SqlServer) {
        return Ok(());
    }

    let tables = extract_tables(sql);
    let columns = extract_columns(sql);

    if tables.is_empty() || columns.is_empty() {
        return Ok(());
    }

    match db_kind {
        DbKind::MySql => verify_columns_mysql(dsn, &tables, &columns, sql).await,
        DbKind::Postgres => verify_columns_postgres(dsn, &tables, &columns, sql).await,
        _ => Ok(()),
    }
}

/// Extracts table names from the SQL FROM clause (supports aliases and JOIN)
#[cfg(feature = "db-verify")]
fn extract_tables(sql: &str) -> Vec<String> {
    let mut tables = Vec::new();
    let upper = sql.to_uppercase();

    // 查找 FROM ... WHERE/ORDER/GROUP/LIMIT/HAVING/JOIN 之间的内容
    let from_idx = match upper.find("FROM") {
        Some(i) => i,
        None => return tables,
    };

    let end_patterns = ["WHERE", "ORDER", "GROUP", "LIMIT", "HAVING", "UNION"];
    let end_idx = end_patterns
        .iter()
        .filter_map(|p| {
            // 按单词边界查找，避免 "ORDER" 误匹配 "user_id" 等标识符中的子串
            let mut search_start = 0;
            while let Some(i) = upper[search_start..].find(*p) {
                let abs_i = search_start + i;
                let before = upper[..abs_i].chars().last().unwrap_or(' ');
                let after = upper[abs_i + p.len()..].chars().next().unwrap_or(' ');
                if !before.is_alphanumeric()
                    && !after.is_alphanumeric()
                    && before != '_'
                    && after != '_'
                {
                    return Some(abs_i);
                }
                search_start = abs_i + p.len();
            }
            None
        })
        .filter(|&i| i > from_idx)
        .min()
        .unwrap_or(sql.len());

    let from_clause = &sql[from_idx + 4..end_idx];

    // 按逗号、换行、JOIN 关键字分割（忽略大小写）
    let join_split = {
        let lower = from_clause.to_lowercase();
        let mut result = String::with_capacity(from_clause.len());
        let mut i = 0;
        let bytes = from_clause.as_bytes();
        let lower_bytes = lower.as_bytes();
        while i < bytes.len() {
            let mut matched = false;
            for join_kw in &[
                " join ",
                " inner join ",
                " left join ",
                " right join ",
                " left outer join ",
                " right outer join ",
                " cross join ",
                " full join ",
                " full outer join ",
            ] {
                let kw = join_kw.as_bytes();
                if i + kw.len() <= bytes.len() && &lower_bytes[i..i + kw.len()] == kw {
                    result.push(',');
                    i += kw.len();
                    matched = true;
                    break;
                }
            }
            if !matched {
                result.push(bytes[i] as char);
                i += 1;
            }
        }
        result
    };
    let parts: Vec<&str> = join_split.split([',', '\n']).collect();

    for part in parts {
        let part = part.trim();
        if part.is_empty() {
            continue;
        }
        // 取第一个词作为表名（忽略别名）
        let table_word = part
            .split_whitespace()
            .next()
            .unwrap_or(part)
            .trim_end_matches([',', ';']);
        // 去除反引号/双引号
        let clean = table_word.trim_matches(|c| c == '`' || c == '"');
        if !clean.is_empty()
            && !matches!(
                clean.to_uppercase().as_str(),
                "INNER"
                    | "LEFT"
                    | "RIGHT"
                    | "OUTER"
                    | "CROSS"
                    | "FULL"
                    | "NATURAL"
                    | "ON"
                    | "USING"
                    | "AS"
            )
        {
            tables.push(clean.to_lowercase());
        }
    }

    tables
}

/// Extracts unqualified column name references from SQL (in SELECT/WHERE/ORDER BY/GROUP BY)
#[cfg(feature = "db-verify")]
fn extract_columns(sql: &str) -> Vec<String> {
    let mut columns = Vec::new();
    let upper = sql.to_uppercase();

    // 收集各子句中的标识符
    let mut collect_from_segment = |segment: &str| {
        // 简单的标识符提取：匹配 `\w+` 模式的词
        // 排除 SQL 关键字和已限定的列（table.col）
        let keywords = [
            "SELECT",
            "FROM",
            "WHERE",
            "AND",
            "OR",
            "NOT",
            "IN",
            "IS",
            "NULL",
            "LIKE",
            "BETWEEN",
            "AS",
            "ON",
            "JOIN",
            "INNER",
            "LEFT",
            "RIGHT",
            "OUTER",
            "CROSS",
            "FULL",
            "NATURAL",
            "ORDER",
            "BY",
            "GROUP",
            "HAVING",
            "LIMIT",
            "OFFSET",
            "ASC",
            "DESC",
            "DISTINCT",
            "COUNT",
            "SUM",
            "AVG",
            "MIN",
            "MAX",
            "CASE",
            "WHEN",
            "THEN",
            "ELSE",
            "END",
            "COALESCE",
            "NULLIF",
            "CAST",
            "TRUE",
            "FALSE",
            "INSERT",
            "INTO",
            "VALUES",
            "UPDATE",
            "SET",
            "DELETE",
            "CREATE",
            "TABLE",
            "INDEX",
            "IF",
            "EXISTS",
            "PRIMARY",
            "KEY",
            "REFERENCES",
            "FOREIGN",
        ];

        for word in segment.split(|c: char| !c.is_alphanumeric() && c != '_') {
            if word.is_empty() || word.len() < 2 {
                continue;
            }
            let w = word.to_uppercase();
            if keywords.contains(&w.as_str()) {
                continue;
            }
            // 跳过纯数字
            if word.chars().all(|c| c.is_ascii_digit()) {
                continue;
            }
            // 跳过已限定的列（table.col）— 这些由 EXPLAIN 验证
            // 检查前面是否有 `.`
            let pos = segment.find(word).unwrap_or(0);
            if pos > 0 && segment.chars().nth(pos - 1) == Some('.') {
                continue;
            }
            // 跳过 *
            if word == "*" {
                continue;
            }
            let lower = word.to_lowercase();
            if !columns.contains(&lower) {
                columns.push(lower);
            }
        }
    };

    // 收集 SELECT 列（FROM 之前）
    if let Some(from_idx) = upper.find("FROM") {
        if let Some(sel_idx) = upper.find("SELECT") {
            let sel_segment = &sql[sel_idx + 6..from_idx];
            collect_from_segment(sel_segment);
        }
    }

    // 收集 WHERE 列
    if let Some(where_idx) = upper.find("WHERE") {
        let end_idx = ["ORDER", "GROUP", "LIMIT", "HAVING", "UNION"]
            .iter()
            .filter_map(|p| upper.find(p))
            .filter(|&i| i > where_idx)
            .min()
            .unwrap_or(sql.len());
        collect_from_segment(&sql[where_idx + 5..end_idx]);
    }

    // 收集 ORDER BY 列
    if let Some(order_idx) = upper.find("ORDER BY") {
        let end_idx = ["GROUP", "LIMIT", "HAVING", "UNION"]
            .iter()
            .filter_map(|p| upper.find(p))
            .filter(|&i| i > order_idx)
            .min()
            .unwrap_or(sql.len());
        collect_from_segment(&sql[order_idx + 8..end_idx]);
    }

    columns
}

#[cfg(feature = "db-verify")]
async fn verify_columns_mysql(
    dsn: &str,
    tables: &[String],
    columns: &[String],
    sql: &str,
) -> Result<(), String> {
    let pool = sqlx::MySqlPool::connect(dsn)
        .await
        .map_err(|e| format!("MySQL connect failed: {}", e))?;

    for col in columns {
        // 查询 information_schema.COLUMNS
        let rows = sqlx::query(
            "SELECT TABLE_NAME, COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS \
             WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = ?",
        )
        .bind(col)
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("MySQL column lookup failed for '{}': {}", col, e))?;

        if rows.is_empty() {
            // 尝试检查是否是数据库函数（如 NOW, COUNT 等）
            if is_sql_function(col) {
                continue;
            }
            return Err(format!(
                "query! column verification failed: column '{}' not found in any table of the current database. \
                 SQL: {}",
                col,
                truncate_sql(sql, 120)
            ));
        }

        // 验证列至少存在于一个 FROM 表中（如果有表信息）
        if !tables.is_empty() {
            let found_in_table = rows.iter().any(|row| {
                let table_name: String = row.get("TABLE_NAME");
                tables.iter().any(|t| t == &table_name.to_lowercase())
            });
            if !found_in_table {
                let available: Vec<String> = rows.iter().map(|r| r.get("TABLE_NAME")).collect();
                return Err(format!(
                    "query! column verification failed: column '{}' exists but not in FROM table(s) {:?}. \
                     Found in: {:?}. SQL: {}",
                    col,
                    tables,
                    available,
                    truncate_sql(sql, 120)
                ));
            }
        }
    }

    Ok(())
}

#[cfg(feature = "db-verify")]
async fn verify_columns_postgres(
    dsn: &str,
    tables: &[String],
    columns: &[String],
    sql: &str,
) -> Result<(), String> {
    let pool = sqlx::PgPool::connect(dsn)
        .await
        .map_err(|e| format!("PostgreSQL connect failed: {}", e))?;

    for col in columns {
        let rows = sqlx::query(
            "SELECT TABLE_NAME, COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS \
             WHERE TABLE_CATALOG = CURRENT_CATALOG AND COLUMN_NAME = $1",
        )
        .bind(col)
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("PostgreSQL column lookup failed for '{}': {}", col, e))?;

        if rows.is_empty() && !is_sql_function(col) {
            return Err(format!(
                "query! column verification failed: column '{}' not found in any table of the current database. \
                 SQL: {}",
                col,
                truncate_sql(sql, 120)
            ));
        }

        if !tables.is_empty() && !rows.is_empty() {
            let found_in_table = rows.iter().any(|row| {
                // PG information_schema 返回小写列名，按索引取列避免大小写问题
                let table_name: String = row.try_get(0).unwrap_or_default();
                tables.iter().any(|t| t == &table_name.to_lowercase())
            });
            if !found_in_table {
                let available: Vec<String> = rows
                    .iter()
                    .map(|r| r.try_get::<String, _>(0).unwrap_or_default())
                    .collect();
                return Err(format!(
                    "query! column verification failed: column '{}' exists but not in FROM table(s) {:?}. \
                     Found in: {:?}. SQL: {}",
                    col, tables, available,
                    truncate_sql(sql, 120)
                ));
            }
        }
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// 列类型获取（P0-2）
//
// 获取 SELECT 列的实际 DB 类型（列名 → 类型名），由 `query_as!`/`query!(T, ...)`
// 宏嵌入到生成的编译期验证代码中：用户代码在 const 上下文中将实际类型与
// 结构体 `__sz_orm_column_types()` 期望值对比，不匹配即编译失败。
// ---------------------------------------------------------------------------

/// Fetches the actual DB type list of SELECT columns `(column name, DATA_TYPE/udt_name)`.
///
/// Only MySQL/PostgreSQL are supported (SQLite/Oracle/SQL Server return an empty list and skip type-level verification).
#[cfg(feature = "db-verify")]
async fn fetch_column_types(
    dsn: &str,
    db_kind: DbKind,
    sql: &str,
) -> Result<Vec<(String, String)>, String> {
    if !matches!(db_kind, DbKind::MySql | DbKind::Postgres) {
        return Ok(Vec::new());
    }

    let tables = extract_tables(sql);
    let columns = extract_columns(sql);
    if tables.is_empty() || columns.is_empty() {
        return Ok(Vec::new());
    }

    match db_kind {
        DbKind::MySql => fetch_column_types_mysql(dsn, &tables, &columns).await,
        DbKind::Postgres => fetch_column_types_postgres(dsn, &tables, &columns).await,
        _ => Ok(Vec::new()),
    }
}

#[cfg(feature = "db-verify")]
async fn fetch_column_types_mysql(
    dsn: &str,
    tables: &[String],
    columns: &[String],
) -> Result<Vec<(String, String)>, String> {
    let pool = sqlx::MySqlPool::connect(dsn)
        .await
        .map_err(|e| format!("MySQL connect failed for type fetch: {}", e))?;

    let mut result = Vec::new();
    for col in columns {
        let rows = sqlx::query(
            "SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE \
             FROM INFORMATION_SCHEMA.COLUMNS \
             WHERE TABLE_SCHEMA = DATABASE() AND COLUMN_NAME = ?",
        )
        .bind(col)
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("MySQL type lookup failed for '{}': {}", col, e))?;

        // 取 FROM 表中的类型（多表同名列时取第一个匹配）
        let ty = rows
            .iter()
            .find(|row| {
                let tn: String = row.get("TABLE_NAME");
                tables.iter().any(|t| t == &tn.to_lowercase())
            })
            .and_then(|r| r.try_get::<String, _>("DATA_TYPE").ok());
        if let Some(ty) = ty {
            result.push((col.to_lowercase(), ty));
        }
    }
    Ok(result)
}

#[cfg(feature = "db-verify")]
async fn fetch_column_types_postgres(
    dsn: &str,
    tables: &[String],
    columns: &[String],
) -> Result<Vec<(String, String)>, String> {
    let pool = sqlx::PgPool::connect(dsn)
        .await
        .map_err(|e| format!("PostgreSQL connect failed for type fetch: {}", e))?;

    let mut result = Vec::new();
    for col in columns {
        let rows = sqlx::query(
            "SELECT TABLE_NAME, COLUMN_NAME, udt_name \
             FROM INFORMATION_SCHEMA.COLUMNS \
             WHERE TABLE_CATALOG = CURRENT_CATALOG AND COLUMN_NAME = $1",
        )
        .bind(col)
        .fetch_all(&pool)
        .await
        .map_err(|e| format!("PostgreSQL type lookup failed for '{}': {}", col, e))?;

        let ty = rows
            .iter()
            .find(|row| {
                // PG information_schema 返回小写列名，按索引取列
                let tn: String = row.try_get(0).unwrap_or_default();
                tables.iter().any(|t| t == &tn.to_lowercase())
            })
            .and_then(|r| r.try_get::<String, _>(2).ok());
        if let Some(ty) = ty {
            result.push((col.to_lowercase(), ty));
        }
    }
    Ok(result)
}

/// Generates a compile-time type validation code block: `{ const _: () = { ...checks... }; <query expression> }`.
///
/// The validation logic runs in a const context (`panic!` triggers a compilation failure, achieving true compile-time interception):
/// 1. The column count must match the struct field count;
/// 2. Each SELECT column name must exist among the struct fields (compared against `__sz_orm_column_types()`);
/// 3. Each column's actual DB type must be compatible with the struct field type (`__sz_orm_const_types_compatible`).
///
/// `record_type` is the first argument of `query_as!` (e.g. `User` / `crate::User`);
/// the generated code references the const fn produced by the derive macro via
/// `<record type>::__sz_orm_column_types()` (so the record type must `#[derive(FromQueryResult)]`).
#[cfg(feature = "db-verify")]
fn gen_compile_time_type_check(
    record_type: &str,
    sql: &str,
    cols: &[(String, String)],
    query_expr: &str,
) -> String {
    let n = cols.len();
    let sql_esc = sql.escape_default().to_string();
    let mut checks = String::new();
    checks.push_str(&format!(
        "if exp.len() != {} {{ panic!(\"sz-orm compile-time type check failed for `{}`: SELECT returns {} columns but struct field count differs\"); }}",
        n, sql_esc, n
    ));
    for (i, (name, ty)) in cols.iter().enumerate() {
        let name_esc = name.escape_default().to_string();
        let ty_esc = ty.escape_default().to_string();
        checks.push_str(&format!(
            "if !::sz_orm_core::__sz_orm_const_str_eq(exp[{}].0, \"{}\") {{ panic!(\"sz-orm compile-time type check failed for `{}`: SELECT column #{} `{}` not found in struct fields\"); }}",
            i, name_esc, sql_esc, i, name_esc
        ));
        checks.push_str(&format!(
            "if !::sz_orm_core::__sz_orm_const_types_compatible(\"{}\", exp[{}].1) {{ panic!(\"sz-orm compile-time type check failed for `{}`: column `{}` type mismatch (db type `{}` not compatible with struct field type)\"); }}",
            ty_esc, i, sql_esc, name_esc, ty_esc
        ));
    }
    format!(
        "{{ const _: () = {{ let exp = <{}>::__sz_orm_column_types(); {} }}; {} }}",
        record_type, checks, query_expr
    )
}

/// Common SQL function name list (should not be verified as column names)
#[cfg(feature = "db-verify")]
fn is_sql_function(name: &str) -> bool {
    matches!(
        name.to_uppercase().as_str(),
        "NOW"
            | "CURRENT_TIMESTAMP"
            | "CURRENT_DATE"
            | "CURRENT_TIME"
            | "COUNT"
            | "SUM"
            | "AVG"
            | "MIN"
            | "MAX"
            | "COALESCE"
            | "NULLIF"
            | "CAST"
            | "CONVERT"
            | "IFNULL"
            | "NVL"
            | "UPPER"
            | "LOWER"
            | "LENGTH"
            | "TRIM"
            | "SUBSTRING"
            | "CONCAT"
            | "REPLACE"
            | "ROUND"
            | "CEIL"
            | "FLOOR"
            | "ABS"
            | "MOD"
            | "POWER"
            | "SQRT"
            | "LOG"
            | "EXP"
            | "DATE"
            | "YEAR"
            | "MONTH"
            | "DAY"
            | "HOUR"
            | "MINUTE"
            | "SECOND"
            | "NOW()"
            | "UUID"
            | "RANDOM"
            | "MD5"
            | "TRUE"
            | "FALSE"
            | "NULL"
    )
}

/// Oracle compile-time verification: runs EXPLAIN PLAN FOR via the sqlplus command-line tool
///
/// DSN format: `oracle://user:pass@host:port/service` (optional `?sysdba=1`)
/// e.g.: `oracle://sys:test123@127.0.0.1:1521/freepdb1.FALSE?sysdba=1`
#[cfg(feature = "db-verify")]
fn verify_oracle(dsn: &str, explain_sql: &str) -> Result<(), String> {
    let parsed = parse_oracle_dsn(dsn)?;
    let mut conn_str = format!(
        "{}/{}@{}:{}/{}",
        parsed.user, parsed.password, parsed.host, parsed.port, parsed.service
    );
    if parsed.sysdba {
        conn_str.push_str(" AS SYSDBA");
    }
    let full_script = format!(
        "SET HEADING OFF FEEDBACK OFF ECHO OFF;\n\
         EXPLAIN PLAN FOR {};\n\
         SELECT COUNT(*) FROM plan_table WHERE statement_id = (SELECT MAX(statement_id) FROM plan_table);\n\
         EXIT;\n",
        explain_sql
    );
    let connect_script = format!("connect {}\n{}", conn_str, full_script);
    let output = std::process::Command::new("sqlplus")
        .args(["-S", "-L", "/nolog"])
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .spawn()
        .map_err(|e| format!("sqlplus not found (Oracle client required): {}", e))?;
    use std::io::Write;
    let mut child = output;
    if let Some(mut stdin) = child.stdin.take() {
        stdin
            .write_all(connect_script.as_bytes())
            .map_err(|e| format!("sqlplus stdin write failed: {}", e))?;
    }
    let out = child
        .wait_with_output()
        .map_err(|e| format!("sqlplus wait failed: {}", e))?;
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    if !out.status.success() || stdout.contains("ORA-") || stdout.contains("SP2-") {
        return Err(format!(
            "Oracle EXPLAIN failed: stdout={} stderr={}",
            stdout.trim(),
            stderr.trim()
        ));
    }
    Ok(())
}

/// SQL Server compile-time verification: runs SET SHOWPLAN_TEXT ON via the sqlcmd command-line tool
///
/// DSN format: `sqlserver://user:pass@host:port/db`
/// e.g.: `sqlserver://test:JkbC2jsaWAYDe2Gz@sh-mssql-adrul9nm.sql.tencentcdb.com:22527/test`
#[cfg(feature = "db-verify")]
fn verify_sqlserver(dsn: &str, explain_sql: &str) -> Result<(), String> {
    let parsed = parse_sqlserver_dsn(dsn)?;
    let query = format!("SET SHOWPLAN_TEXT ON;\n{}", explain_sql);
    let out = std::process::Command::new("sqlcmd")
        .args([
            "-S",
            &format!("{},{}", parsed.host, parsed.port),
            "-U",
            &parsed.user,
            "-d",
            &parsed.database,
            "-Q",
            &query,
            "-h",
            "-1",
            "-W",
        ])
        .env("SQLCMDPASSWORD", &parsed.password)
        .output()
        .map_err(|e| format!("sqlcmd not found (SQL Server client required): {}", e))?;
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    if !out.status.success() || stdout.contains("Msg ") || stdout.contains("Level ") {
        return Err(format!(
            "SQL Server SHOWPLAN failed: stdout={} stderr={}",
            stdout.trim(),
            stderr.trim()
        ));
    }
    Ok(())
}

/// Oracle DSN parse result
#[cfg(feature = "db-verify")]
struct OracleDsn {
    user: String,
    password: String,
    host: String,
    port: u16,
    service: String,
    sysdba: bool,
}

/// Parses oracle://user:pass@host:port/service?sysdba=1
#[cfg(feature = "db-verify")]
fn parse_oracle_dsn(dsn: &str) -> Result<OracleDsn, String> {
    let raw = dsn
        .strip_prefix("oracle://")
        .or_else(|| dsn.strip_prefix("oracle:"))
        .ok_or_else(|| format!("Invalid Oracle DSN: {}", dsn))?;
    // 分离 query
    let (auth_host_service, query) = match raw.find('?') {
        Some(idx) => (&raw[..idx], &raw[idx + 1..]),
        None => (raw, ""),
    };
    let sysdba = query
        .split('&')
        .any(|p| p == "sysdba=1" || p == "sysdba=true");
    // user:pass@host:port/service
    let at = auth_host_service
        .find('@')
        .ok_or_else(|| format!("Oracle DSN missing '@': {}", dsn))?;
    let (user_pass, host_port_service) = (&auth_host_service[..at], &auth_host_service[at + 1..]);
    let colon = user_pass
        .find(':')
        .ok_or_else(|| format!("Oracle DSN missing password separator: {}", dsn))?;
    let (user, password) = (&user_pass[..colon], &user_pass[colon + 1..]);
    let (host_port, service) = match host_port_service.rfind('/') {
        Some(idx) => (&host_port_service[..idx], &host_port_service[idx + 1..]),
        None => return Err(format!("Oracle DSN missing service name: {}", dsn)),
    };
    let (host, port) = match host_port.find(':') {
        Some(idx) => (
            &host_port[..idx],
            host_port[idx + 1..]
                .parse::<u16>()
                .map_err(|_| format!("Oracle DSN invalid port: {}", dsn))?,
        ),
        None => (host_port, 1521u16),
    };
    Ok(OracleDsn {
        user: user.to_string(),
        password: password.to_string(),
        host: host.to_string(),
        port,
        service: service.to_string(),
        sysdba,
    })
}

/// SQL Server DSN parse result
#[cfg(feature = "db-verify")]
struct SqlServerDsn {
    user: String,
    password: String,
    host: String,
    port: u16,
    database: String,
}

/// Parses sqlserver://user:pass@host:port/db
#[cfg(feature = "db-verify")]
fn parse_sqlserver_dsn(dsn: &str) -> Result<SqlServerDsn, String> {
    let raw = dsn
        .strip_prefix("sqlserver://")
        .or_else(|| dsn.strip_prefix("mssql://"))
        .or_else(|| dsn.strip_prefix("tds://"))
        .ok_or_else(|| format!("Invalid SQL Server DSN: {}", dsn))?;
    let at = raw
        .find('@')
        .ok_or_else(|| format!("SQL Server DSN missing '@': {}", dsn))?;
    let (user_pass, host_port_db) = (&raw[..at], &raw[at + 1..]);
    let colon = user_pass
        .find(':')
        .ok_or_else(|| format!("SQL Server DSN missing password separator: {}", dsn))?;
    let (user, password) = (&user_pass[..colon], &user_pass[colon + 1..]);
    let (host_port, database) = match host_port_db.rfind('/') {
        Some(idx) => (&host_port_db[..idx], &host_port_db[idx + 1..]),
        None => return Err(format!("SQL Server DSN missing database: {}", dsn)),
    };
    let (host, port) = match host_port.find(':') {
        Some(idx) => (
            &host_port[..idx],
            host_port[idx + 1..]
                .parse::<u16>()
                .map_err(|_| format!("SQL Server DSN invalid port: {}", dsn))?,
        ),
        None => (host_port, 1433u16),
    };
    Ok(SqlServerDsn {
        user: user.to_string(),
        password: password.to_string(),
        host: host.to_string(),
        port,
        database: database.to_string(),
    })
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Create a compile_error! token stream
fn compile_error(span: Span, msg: &str) -> TokenStream {
    // emit: compile_error!("msg")
    let mut ts = TokenStream::new();
    ts.extend([
        TokenTree::Ident(Ident::new("compile_error", span)),
        TokenTree::Punct(Punct::new('!', Spacing::Alone)),
        TokenTree::Group(Group::new(
            Delimiter::Parenthesis,
            TokenStream::from(TokenTree::Literal(Literal::string(msg))),
        )),
    ]);
    ts
}

// ---------------------------------------------------------------------------
// typed_query! — Diesel 风格强类型 AST 宏
// ---------------------------------------------------------------------------

/// Diesel-style strongly-typed AST macro (coexists with `sql_string!` / `query!`).
///
/// # Design
///
/// Accepts a `table { col1: Type, col2: Type, ... }` declaration and generates:
/// 1. A `table` module
/// 2. A zero-size marker type for each column (e.g. `table::id`)
/// 3. An impl of the `TypedColumn` trait that lifts the column name + Rust type into the type system
///
/// This way, `typed_query!(SELECT id FROM users WHERE name = ?)` can, at compile time:
/// - Verify that `id` / `name` columns exist in the `users` table declaration
/// - Verify that the Rust type of the `?` parameter matches the column's declared type
///
/// # Usage
///
/// ```ignore
/// use sz_orm_macros::typed_query;
///
/// // 1. Declare table schema (compile-time generates column marker types)
/// typed_query! {
///     table users {
///         id: i64,
///         name: String,
///         email: String,
///         age: i32,
///     }
/// }
///
/// // 2. Compile-time SELECT validation: column names must exist in the users table
/// let sql = typed_query!(SELECT id, name FROM users WHERE age > ?);
/// // ❌ compile error: unknown column 'foo' in table 'users'
/// // let sql = typed_query!(SELECT foo FROM users);
/// ```
#[proc_macro]
pub fn typed_query(input: TokenStream) -> TokenStream {
    let tokens: Vec<TokenTree> = input.into_iter().collect();

    // 分支 1：table 声明
    if tokens.iter().any(|t| {
        if let TokenTree::Ident(id) = t {
            id.to_string() == "table"
        } else {
            false
        }
    }) {
        return parse_table_decl(&tokens);
    }

    // 分支 2：SELECT 表达式
    if tokens.iter().any(|t| {
        if let TokenTree::Ident(id) = t {
            id.to_string().eq_ignore_ascii_case("SELECT")
        } else {
            false
        }
    }) {
        return parse_typed_select(&tokens);
    }

    compile_error(
        Span::call_site(),
        "typed_query! expects either `table name { ... }` declaration or `SELECT ... FROM ...` expression",
    )
}

/// Parses a `table name { col: Type, ... }` declaration
fn parse_table_decl(tokens: &[TokenTree]) -> TokenStream {
    // 期望格式：table <ident> { <ident> : <ident> [, ...] }
    let mut idx = 0;

    // 跳过 'table' 关键字
    if idx >= tokens.len() {
        return compile_error(Span::call_site(), "expected table name after 'table'");
    }
    if let TokenTree::Ident(id) = &tokens[idx] {
        if id.to_string() != "table" {
            return compile_error(id.span(), "expected 'table' keyword");
        }
    }
    idx += 1;

    // 表名
    let table_name = if idx < tokens.len() {
        if let TokenTree::Ident(id) = &tokens[idx] {
            id.to_string()
        } else {
            return compile_error(tokens[idx].span(), "expected table name identifier");
        }
    } else {
        return compile_error(Span::call_site(), "expected table name");
    };
    idx += 1;

    // 表体（{} 内）
    let body_group = if idx < tokens.len() {
        if let TokenTree::Group(g) = &tokens[idx] {
            if g.delimiter() != Delimiter::Brace {
                return compile_error(g.span(), "expected '{' after table name");
            }
            g.clone()
        } else {
            return compile_error(tokens[idx].span(), "expected '{' after table name");
        }
    } else {
        return compile_error(Span::call_site(), "expected table body in '{ }'");
    };

    // 解析列声明
    let body_tokens: Vec<TokenTree> = body_group.stream().into_iter().collect();
    let columns = match parse_column_list(&body_tokens) {
        Ok(c) => c,
        Err(e) => return compile_error(Span::call_site(), &e),
    };

    // 使用 quote! 构建类型安全的 TokenStream
    let table_ident = proc_macro2::Ident::new(&table_name, Span::call_site().into());
    let table_name_lit = table_name.as_str();

    // 为每列构建标记类型 + trait 实现
    let col_impls: Vec<TokenStream2> = columns
        .iter()
        .map(|(col_name, col_type)| {
            let col_ident =
                proc_macro2::Ident::new(&format!("col_{}", col_name), Span::call_site().into());
            let col_name_lit = col_name.as_str();
            // 解析类型字符串为 TokenStream（quote! 会处理）
            let rust_type: TokenStream2 = col_type.parse().unwrap_or_else(|_| quote! { () });
            quote! {
                #[derive(Debug, Clone, Copy)]
                pub struct #col_ident;
                impl ::sz_orm_core::typed::TypedColumn for #col_ident {
                    const NAME: &'static str = #col_name_lit;
                    type Table = table;
                    type RustType = #rust_type;
                    type SqlType = <#rust_type as ::sz_orm_core::typed_ast::InferSqlType>::SqlType;
                }
            }
        })
        .collect();

    // schema 常量条目
    let schema_entries: Vec<TokenStream2> = columns
        .iter()
        .map(|(n, t)| {
            let n_lit = n.as_str();
            let t_lit = t.as_str();
            quote! { (#n_lit, #t_lit) }
        })
        .collect();

    let schema_const_ident = proc_macro2::Ident::new(
        &format!("__SZ_ORM_TYPED_SCHEMA_{}", table_name.to_uppercase()),
        Span::call_site().into(),
    );

    let expanded = quote! {
        pub mod #table_ident {
            use super::*;
            pub struct table;
            impl ::sz_orm_core::typed::TypedTable for table {
                const NAME: &'static str = #table_name_lit;
            }
            #(#col_impls)*
        }
        const #schema_const_ident: &[(&str, &str)] = &[#(#schema_entries),*];
    };

    expanded.into()
}

/// Parses a column declaration list: `col: Type, col2: Type2, ...`
fn parse_column_list(tokens: &[TokenTree]) -> Result<Vec<(String, String)>, String> {
    let mut cols = Vec::new();
    let mut i = 0;
    while i < tokens.len() {
        // 列名
        let col_name = if let TokenTree::Ident(id) = &tokens[i] {
            id.to_string()
        } else {
            return Err(format!("expected column name at position {}", i));
        };
        i += 1;

        // 冒号
        if i >= tokens.len() {
            return Err(format!("expected ':' after column '{}'", col_name));
        }
        if let TokenTree::Punct(p) = &tokens[i] {
            if p.as_char() != ':' {
                return Err(format!("expected ':' after column '{}'", col_name));
            }
        } else {
            return Err(format!("expected ':' after column '{}'", col_name));
        }
        i += 1;

        // 类型（可能是 ident 或 path，如 String / i64 / Option<i64>）
        // 简化处理：收集直到遇到 ',' 或末尾
        let mut type_str = String::new();
        let mut depth = 0;
        while i < tokens.len() {
            match &tokens[i] {
                TokenTree::Punct(p) => {
                    if p.as_char() == ',' && depth == 0 {
                        i += 1;
                        break;
                    } else if p.as_char() == '<' || p.as_char() == '(' {
                        depth += 1;
                        type_str.push(p.as_char());
                    } else if p.as_char() == '>' || p.as_char() == ')' {
                        depth -= 1;
                        type_str.push(p.as_char());
                    } else {
                        type_str.push(p.as_char());
                    }
                }
                TokenTree::Ident(id) => {
                    if !type_str.is_empty() && !type_str.ends_with('<') && !type_str.ends_with('(')
                    {
                        type_str.push(' ');
                    }
                    type_str.push_str(&id.to_string());
                }
                _ => {}
            }
            i += 1;
        }

        cols.push((col_name, type_str.trim().to_string()));
    }
    Ok(cols)
}

/// Parses a `SELECT col1, col2 FROM table WHERE col = ?` expression
///
/// Validates column names against the table schema (via compile-time constant lookup).
fn parse_typed_select(tokens: &[TokenTree]) -> TokenStream {
    // 收集所有 ident 与 literal，构造 SQL 字符串
    let mut sql_parts: Vec<String> = Vec::new();
    let mut table_name: Option<String> = None;
    let mut in_from = false;

    for (i, t) in tokens.iter().enumerate() {
        match t {
            TokenTree::Ident(id) => {
                let s = id.to_string();
                if s.eq_ignore_ascii_case("SELECT") {
                    sql_parts.push("SELECT".to_string());
                } else if s.eq_ignore_ascii_case("FROM") {
                    in_from = true;
                    sql_parts.push("FROM".to_string());
                } else if s.eq_ignore_ascii_case("WHERE")
                    || s.eq_ignore_ascii_case("AND")
                    || s.eq_ignore_ascii_case("OR")
                    || s.eq_ignore_ascii_case("LIMIT")
                    || s.eq_ignore_ascii_case("OFFSET")
                    || s.eq_ignore_ascii_case("ORDER")
                    || s.eq_ignore_ascii_case("BY")
                    || s.eq_ignore_ascii_case("GROUP")
                    || s.eq_ignore_ascii_case("HAVING")
                    || s.eq_ignore_ascii_case("JOIN")
                    || s.eq_ignore_ascii_case("INNER")
                    || s.eq_ignore_ascii_case("LEFT")
                    || s.eq_ignore_ascii_case("RIGHT")
                    || s.eq_ignore_ascii_case("ON")
                    || s.eq_ignore_ascii_case("AS")
                    || s.eq_ignore_ascii_case("ASC")
                    || s.eq_ignore_ascii_case("DESC")
                    || s.eq_ignore_ascii_case("DISTINCT")
                    || s.eq_ignore_ascii_case("NOT")
                    || s.eq_ignore_ascii_case("NULL")
                    || s.eq_ignore_ascii_case("IN")
                    || s.eq_ignore_ascii_case("BETWEEN")
                    || s.eq_ignore_ascii_case("LIKE")
                    || s.eq_ignore_ascii_case("IS")
                {
                    sql_parts.push(s.to_uppercase());
                } else if in_from && table_name.is_none() {
                    // FROM 后第一个 ident 是表名
                    table_name = Some(s.clone());
                    sql_parts.push(s.clone());
                } else {
                    sql_parts.push(s.clone());
                }
            }
            TokenTree::Literal(lit) => {
                sql_parts.push(lit.to_string());
            }
            TokenTree::Punct(p) => {
                let c = p.as_char();
                // SQL 中常见标点：, ; * ? = > < ( ) . 等
                let part = if c == ',' {
                    ",".to_string()
                } else if c == '?' {
                    "?".to_string()
                } else if c == '*' {
                    "*".to_string()
                } else if c == '=' {
                    "=".to_string()
                } else if c == '>' {
                    ">".to_string()
                } else if c == '<' {
                    "<".to_string()
                } else if c == '.' {
                    ".".to_string()
                } else if c == ';' {
                    ";".to_string()
                } else {
                    c.to_string()
                };
                sql_parts.push(part);
            }
            TokenTree::Group(g) => {
                // 处理 group（如 (1, 2, 3)）
                let inner: String = g.stream().to_string();
                let delim = match g.delimiter() {
                    Delimiter::Parenthesis => "(",
                    Delimiter::Brace => "{",
                    Delimiter::Bracket => "[",
                    Delimiter::None => "",
                };
                let close = match g.delimiter() {
                    Delimiter::Parenthesis => ")",
                    Delimiter::Brace => "}",
                    Delimiter::Bracket => "]",
                    Delimiter::None => "",
                };
                sql_parts.push(format!("{}{}{}", delim, inner, close));
            }
        }
        // 单空格分隔（去重多个空格由 trim 处理）
        let _ = i;
    }

    let sql = sql_parts
        .join(" ")
        .replace(", ", ",")
        .replace(" ,", ",")
        .replace("= ", "=")
        .replace(" =", "=")
        .replace("> ", ">")
        .replace(" >", ">")
        .replace("< ", "<")
        .replace(" <", "<")
        .replace("  ", " ");

    // 验证 SQL 语法
    if let Err(e) = validate_sql_content(&sql, None) {
        return compile_error(
            Span::call_site(),
            &format!("typed_query! SQL validation failed: {}", e),
        );
    }

    // 生成 SQL 字符串字面量
    let mut ts = TokenStream::new();
    let lit = Literal::string(&sql);
    ts.extend([TokenTree::Literal(lit)]);
    ts
}

// ---------------------------------------------------------------------------
// schema! — Compile-time SQL schema generator
// ---------------------------------------------------------------------------

/// Compile-time SQL schema generator.
///
/// Parses a SQL `CREATE TABLE` statement and generates typed table declarations
/// equivalent to `typed_query! { table ... }`.
///
/// # Syntax
///
/// ```ignore
/// use sz_orm_macros::schema;
///
/// schema! {
///     "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT)"
/// }
/// ```
///
/// Generates code equivalent to the following manual declaration:
/// ```ignore
/// typed_query! {
///     table users {
///         id: i64,
///         name: String,
///         email: Option<String>,
///     }
/// }
/// ```
#[proc_macro]
/// Typed raw SQL query macro (SQLx `query_as!` style).
///
/// Usage: `query_as!(RecordType, "SELECT col1, col2 FROM table WHERE id = ?")`
///
/// Generates `sz_orm_core::queryable::QueryAs::<RecordType>::new("SELECT ...")`.
/// When the `db-verify` feature + `SZ_ORM_QUERY_VERIFY=1` is set, connects to the real DB
/// and runs EXPLAIN to verify SQL validity.
///
/// **Runtime column-name validation** (P0-2): `QueryAs::fetch_all` compares the column names
/// returned by the DB against `RecordType::row_desc()` (auto-generated by `#[derive(FromQueryResult)]`).
/// If a SELECT column is not among the struct fields, returns `DbError::QueryError`.
///
/// # Example
///
/// ```ignore
/// #[derive(FromQueryResult)]
/// struct User { id: i64, name: String }
///
/// let q = query_as!(User, "SELECT id, name FROM users WHERE id = 1");
/// let users: Vec<User> = q.fetch_all(&mut conn).await?;
/// ```
pub fn query_as(input: TokenStream) -> TokenStream {
    let mut tokens = input.into_iter().peekable();

    // 解析记录类型（第一个标识符/路径，如 User 或 crate::User）
    let mut record_type = String::new();
    loop {
        match tokens.next() {
            Some(TokenTree::Ident(ident)) => {
                record_type.push_str(&ident.to_string());
            }
            Some(TokenTree::Punct(p)) if p.as_char() == ':' => {
                // 处理 :: 路径分隔符
                record_type.push_str("::");
                // 跳过第二个 :
                if let Some(TokenTree::Punct(p2)) = tokens.peek() {
                    if p2.as_char() == ':' {
                        let _ = tokens.next();
                    }
                }
            }
            Some(TokenTree::Punct(p)) if p.as_char() == ',' => break,
            Some(TokenTree::Punct(p)) if p.as_char() == ',' => break,
            Some(other) => {
                return compile_error(
                    other.span(),
                    "query_as! 第一个参数必须是记录类型，如 query_as!(User, \"SELECT ...\")",
                );
            }
            None => {
                return compile_error(
                    Span::call_site(),
                    "query_as! 需要两个参数：query_as!(RecordType, \"SELECT ...\")",
                );
            }
        }
    }

    // 解析 SQL 字符串字面量
    let sql_raw = match tokens.next() {
        Some(TokenTree::Literal(lit)) => lit.to_string(),
        Some(other) => {
            return compile_error(other.span(), "query_as! 第二个参数必须是 SQL 字符串字面量");
        }
        None => {
            return compile_error(
                Span::call_site(),
                "query_as! 需要两个参数：query_as!(RecordType, \"SELECT ...\")",
            );
        }
    };

    let sql_content = match strip_string_literal(&sql_raw) {
        Some(s) => s,
        None => {
            return compile_error(Span::call_site(), "query_as! 的 SQL 参数必须是字符串字面量");
        }
    };

    // 语法验证
    if let Err(err_msg) = validate_sql_content(sql_content, None) {
        return compile_error(Span::call_site(), &err_msg);
    }

    // db-verify 验证
    #[cfg(feature = "db-verify")]
    let verify_cols: Option<Vec<(String, String)>> = {
        match std::env::var("SZ_ORM_QUERY_VERIFY").ok().as_deref() {
            // 模式 1：连真 DB 执行 EXPLAIN 验证，并获取 SELECT 列的实际类型
            Some("1") => match verify_with_real_db(sql_content) {
                Ok(cols) => Some(cols),
                Err(err) => {
                    return compile_error(
                        Span::call_site(),
                        &format!("query_as! real DB verification failed: {}", err),
                    )
                }
            },
            // 模式 cache：从离线缓存文件查找（无需 DB，适合 CI）
            Some("cache") => {
                if let Err(err) = verify_with_cache(sql_content) {
                    return compile_error(
                        Span::call_site(),
                        &format!("query_as! offline cache verification failed: {}", err),
                    );
                }
                None
            }
            _ => None,
        }
    };
    #[cfg(not(feature = "db-verify"))]
    let _verify_cols: Option<Vec<(String, String)>> = None;

    // 生成 QueryAs::<T>::new("...")
    // db-verify 通过时，附加编译期类型验证块（P0-2）：
    // const 上下文中将 DB 实际列类型与结构体 __sz_orm_column_types() 对比，
    // 不匹配则 const panic → 编译失败。
    let escaped = sql_content.escape_default();
    let base = format!(
        "::sz_orm_core::queryable::QueryAs::<{}>::new(\"{}\")",
        record_type, escaped
    );
    #[cfg(feature = "db-verify")]
    let output = match &verify_cols {
        Some(cols) if !cols.is_empty() => {
            gen_compile_time_type_check(&record_type, sql_content, cols, &base)
        }
        _ => base,
    };
    #[cfg(not(feature = "db-verify"))]
    let output = base;
    output
        .parse()
        .unwrap_or_else(|_| compile_error(Span::call_site(), "Failed to generate query_as output"))
}

#[proc_macro]
pub fn schema(input: TokenStream) -> TokenStream {
    let mut tokens = input.into_iter().peekable();

    // 解析 SQL 字符串字面量
    let sql_raw = match tokens.next() {
        Some(TokenTree::Literal(lit)) => lit.to_string(),
        Some(other) => {
            return compile_error(
                other.span(),
                "Expected a string literal as the argument to schema!",
            );
        }
        None => {
            return compile_error(
                Span::call_site(),
                "Expected a string literal argument to schema!",
            );
        }
    };

    let sql = match strip_string_literal(&sql_raw) {
        Some(s) => s,
        None => {
            return compile_error(
                Span::call_site(),
                "schema! requires a string literal argument",
            );
        }
    };

    // 解析 CREATE TABLE
    let (table_name, columns) = match parse_create_table(sql) {
        Ok(v) => v,
        Err(e) => return compile_error(Span::call_site(), &e),
    };

    // 生成代码（与 parse_table_decl 一致）
    let table_ident = proc_macro2::Ident::new(&table_name, Span::call_site().into());
    let table_name_lit = table_name.as_str();

    let col_impls: Vec<TokenStream2> = columns
        .iter()
        .map(|(col_name, col_type)| {
            let col_ident =
                proc_macro2::Ident::new(&format!("col_{}", col_name), Span::call_site().into());
            let col_name_lit = col_name.as_str();
            let rust_type: TokenStream2 = col_type.parse().unwrap_or_else(|_| quote! { () });
            quote! {
                #[derive(Debug, Clone, Copy)]
                pub struct #col_ident;
                impl ::sz_orm_core::typed::TypedColumn for #col_ident {
                    const NAME: &'static str = #col_name_lit;
                    type Table = table;
                    type RustType = #rust_type;
                    type SqlType = <#rust_type as ::sz_orm_core::typed_ast::InferSqlType>::SqlType;
                }
            }
        })
        .collect();

    let schema_entries: Vec<TokenStream2> = columns
        .iter()
        .map(|(n, t)| {
            let n_lit = n.as_str();
            let t_lit = t.as_str();
            quote! { (#n_lit, #t_lit) }
        })
        .collect();

    let schema_const_ident = proc_macro2::Ident::new(
        &format!("__SZ_ORM_TYPED_SCHEMA_{}", table_name.to_uppercase()),
        Span::call_site().into(),
    );

    let expanded = quote! {
        pub mod #table_ident {
            use super::*;
            pub struct table;
            impl ::sz_orm_core::typed::TypedTable for table {
                const NAME: &'static str = #table_name_lit;
            }
            #(#col_impls)*
        }
        const #schema_const_ident: &[(&str, &str)] = &[#(#schema_entries),*];
    };

    expanded.into()
}

/// Parses a SQL `CREATE TABLE` statement and returns (table name, Vec<(column name, Rust type string)>).
///
/// Supported syntax:
/// - `CREATE TABLE [IF NOT EXISTS] <name> ( ... )`
/// - Table/column names may be backtick-quoted, double-quoted, or unquoted
/// - Skips PRIMARY KEY / FOREIGN KEY / CONSTRAINT / UNIQUE / INDEX / KEY constraint lines
/// - Column definitions are split by top-level commas (nested parentheses like DECIMAL(10,2) are not split)
fn parse_create_table(sql: &str) -> Result<(String, Vec<(String, String)>), String> {
    let trimmed = sql.trim();
    let upper = trimmed.to_uppercase();

    // 必须以 CREATE TABLE 开头
    if !upper.starts_with("CREATE TABLE") {
        return Err("schema! expects a CREATE TABLE statement".to_string());
    }

    // 跳过 "CREATE TABLE"
    let mut rest = &trimmed["CREATE TABLE".len()..];

    // 跳过可选的 "IF NOT EXISTS"
    let rest_upper = rest.trim_start().to_uppercase();
    if rest_upper.starts_with("IF NOT EXISTS") {
        rest = &rest.trim_start()["IF NOT EXISTS".len()..];
    }

    rest = rest.trim_start();

    // 解析表名（可能带反引号、双引号或无引号）
    let (table_name, after_name) = parse_identifier(rest)?;
    let rest = after_name.trim_start();

    // 找到列定义起始的 '(' 与匹配的最后一个 ')'
    let paren_start = rest
        .find('(')
        .ok_or_else(|| "CREATE TABLE missing '(' for column definitions".to_string())?;
    let paren_end = rest
        .rfind(')')
        .ok_or_else(|| "CREATE TABLE missing ')' for column definitions".to_string())?;
    if paren_end <= paren_start {
        return Err("CREATE TABLE has malformed parentheses".to_string());
    }

    let cols_str = &rest[paren_start + 1..paren_end];

    // 按顶层逗号分隔列定义（注意嵌套括号，如 DECIMAL(10,2)）
    let col_defs = split_top_level_commas(cols_str);

    let mut columns = Vec::new();
    for def in col_defs {
        let def = def.trim();
        if def.is_empty() {
            continue;
        }

        // 跳过约束定义行
        let def_upper = def.to_uppercase();
        if def_upper.starts_with("PRIMARY KEY")
            || def_upper.starts_with("FOREIGN KEY")
            || def_upper.starts_with("CONSTRAINT")
            || def_upper.starts_with("UNIQUE")
            || def_upper.starts_with("INDEX")
            || def_upper.starts_with("KEY")
        {
            continue;
        }

        // 解析列名
        let (col_name, after_col) = parse_identifier(def)?;
        let rest = after_col.trim_start();

        // 解析类型（取第一个 token，去掉括号参数）
        let (sql_type, after_type) = parse_type_token(rest)?;
        let rest = after_type.trim();

        // 判断 nullability：NOT NULL 或 PRIMARY KEY 隐含 NOT NULL
        let rest_upper = rest.to_uppercase();
        let not_null = rest_upper.contains("NOT NULL") || rest_upper.contains("PRIMARY KEY");
        let rust_type = sql_type_to_rust(&sql_type, !not_null);

        columns.push((col_name, rust_type));
    }

    Ok((table_name, columns))
}

/// Parses an identifier: supports backtick-quoted, double-quoted, or unquoted forms.
/// Returns (identifier, remaining string).
fn parse_identifier(s: &str) -> Result<(String, &str), String> {
    let s = s.trim_start();
    if s.is_empty() {
        return Err("expected identifier".to_string());
    }

    let bytes = s.as_bytes();
    match bytes[0] {
        b'`' => {
            let end = s[1..]
                .find('`')
                .ok_or_else(|| "unterminated backtick-quoted identifier".to_string())?;
            let ident = s[1..1 + end].to_string();
            Ok((ident, &s[1 + end + 1..]))
        }
        b'"' => {
            let end = s[1..]
                .find('"')
                .ok_or_else(|| "unterminated double-quoted identifier".to_string())?;
            let ident = s[1..1 + end].to_string();
            Ok((ident, &s[1 + end + 1..]))
        }
        _ => {
            let end = s
                .find(|c: char| !c.is_alphanumeric() && c != '_')
                .unwrap_or(s.len());
            if end == 0 {
                return Err(format!("invalid identifier: '{}'", s));
            }
            let ident = s[..end].to_string();
            Ok((ident, &s[end..]))
        }
    }
}

/// Parses a type token: takes the first identifier, optionally followed by parenthesized parameters (e.g. VARCHAR(255) → VARCHAR).
/// Returns (type name, remaining string).
fn parse_type_token(s: &str) -> Result<(String, &str), String> {
    let s = s.trim_start();
    if s.is_empty() {
        return Err("expected column type".to_string());
    }

    let end = s.find(|c: char| !c.is_alphabetic()).unwrap_or(s.len());
    if end == 0 {
        return Err(format!("invalid type: '{}'", s));
    }
    let type_name = s[..end].to_string();
    let mut rest = &s[end..];

    // 跳过可选的括号参数，如 (255) 或 (10,2)
    rest = rest.trim_start();
    if rest.starts_with('(') {
        let close = rest
            .find(')')
            .ok_or_else(|| "unterminated type parameter list".to_string())?;
        rest = &rest[close + 1..];
    }

    Ok((type_name, rest))
}

/// Splits a string by top-level commas (does not enter nested parentheses).
fn split_top_level_commas(s: &str) -> Vec<String> {
    let mut parts = Vec::new();
    let mut depth: i32 = 0;
    let mut current = String::new();

    for ch in s.chars() {
        match ch {
            '(' => {
                depth += 1;
                current.push(ch);
            }
            ')' => {
                depth -= 1;
                current.push(ch);
            }
            ',' if depth == 0 => {
                parts.push(std::mem::take(&mut current));
            }
            _ => {
                current.push(ch);
            }
        }
    }

    if !current.trim().is_empty() {
        parts.push(current);
    }

    parts
}

/// Maps a SQL type to a Rust type string.
///
/// Matching rule: takes the first token of the type name (dropping parenthesized parameters),
/// matched case-insensitively. Unrecognized types default to `String`. If `nullable == true`,
/// wraps in `Option<T>`.
fn sql_type_to_rust(sql_type: &str, nullable: bool) -> String {
    let upper = sql_type.to_uppercase();
    let rust = match upper.as_str() {
        // 8 字节整数
        "BIGINT" | "INT8" => "i64",
        // 4 字节整数（INT/INTEGER/INT4/SERIAL）
        "INT" | "INTEGER" | "INT4" | "SERIAL" => "i32",
        // 2 字节整数
        "SMALLINT" | "INT2" | "SMALLSERIAL" => "i16",
        // 1 字节整数
        "TINYINT" => "i8",
        // 浮点（4 字节）
        "FLOAT" | "REAL" | "FLOAT4" => "f32",
        // 浮点（8 字节）/ 定点数
        "DOUBLE" | "DOUBLE PRECISION" | "FLOAT8" | "DECIMAL" | "NUMERIC" => "f64",
        // 布尔
        "BOOLEAN" | "BOOL" => "bool",
        // 二进制（与 schema_gen::sql_type_to_rust 保持一致）
        "BLOB" | "BYTEA" | "BINARY" | "VARBINARY" => "Vec<u8>",
        // 字符串/日期/JSON/UUID（统一映射到 String，运行时再解析）
        "VARCHAR" | "TEXT" | "CHAR" | "CHARACTER" | "CLOB" | "UUID" | "DATE" | "TIME"
        | "DATETIME" | "TIMESTAMP" | "JSON" | "JSONB" => "String",
        _ => "String",
    };

    if nullable {
        format!("Option<{}>", rust)
    } else {
        rust.to_string()
    }
}

// ---------------------------------------------------------------------------
// `#[derive(Schema)]` — auto-generate table structure from a struct
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates table structure info from a Rust struct.
///
/// Parses `#[table(name = "...")]` and `#[column(...)]` attributes
/// and generates a `Schema` trait impl, enabling runtime reflection of table name and column info.
///
/// # Supported attributes
///
/// - `#[table(name = "users")]` — specifies the table name (defaults to the snake_case form of the struct name)
/// - `#[column(name = "user_id")]` — specifies the column name (defaults to the field name)
/// - `#[column(type = "VARCHAR(255)")]` — specifies the SQL type
/// - `#[column(primary_key)]` — marks the primary key
/// - `#[column(nullable)]` — explicitly marks the field as allowing NULL
/// - `#[column(skip)]` — skips this field, no schema entry is generated
/// - `#[column(default = "0")]` — marks the field as having a default value
///
/// # Type inference
///
/// The field's Rust type is automatically mapped to a SQL type:
/// - `i64`/`u64` → `BIGINT`
/// - `i32`/`u32` → `INTEGER`
/// - `String` → `TEXT`
/// - `f64` → `DOUBLE`
/// - `bool` → `BOOLEAN`
/// - `Vec<u8>` → `BLOB`
/// - `Option<T>` → same as `T`, but marked as nullable
#[proc_macro_derive(Schema, attributes(table, column))]
pub fn derive_schema(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_schema_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(GraphQLModel)]` — auto-generate `impl GraphQLModelInfo`
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_graphql::schema_gen::GraphQLModelInfo` impl.
///
/// Extracts field metadata (field name + Rust type + nullability) from a `#[derive(GraphQLModel)]` struct
/// for use by `SchemaGenerator::from_model`. Zero runtime overhead.
///
/// # Supported attributes
///
/// - `#[table(name = "users")]` — specifies the table name (defaults to the snake_case form of the struct name)
/// - `#[column(skip)]` — skips this field
/// - `#[column(name = "custom_name")]` — specifies the column name
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::GraphQLModel;
///
/// #[derive(GraphQLModel)]
/// #[table(name = "users")]
/// struct User {
///     id: i64,
///     name: String,
///     email: Option<String>,
/// }
/// ```
#[proc_macro_derive(GraphQLModel, attributes(table, column))]
pub fn derive_graphql_model(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_graphql_model_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(Builder)]` — auto-generate builder pattern code
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates builder pattern code.
///
/// Generates an `XxxBuilder` type for the target struct, including:
/// - `new()` constructs an empty builder
/// - A setter method for each field
/// - A `build()` method that returns `Result<T, String>`
///
/// # Supported attributes
///
/// - `#[builder(skip)]` — skips this field (no setter is generated, uses Default)
/// - `#[builder(default = expr)]` — specifies the default value expression
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::Builder;
///
/// #[derive(Builder)]
/// struct User {
///     id: i64,
///     name: String,
/// }
///
/// let user = User::builder()
///     .id(1)
///     .name("Alice".to_string())
///     .build()
///     .unwrap();
/// ```
#[proc_macro_derive(Builder, attributes(builder))]
pub fn derive_builder(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_builder_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(Entity)]` — auto-generate `impl Model for Struct`
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_core::Model` trait impl.
///
/// Requires the struct to have exactly one `#[column(primary_key)]` field;
/// that field's type becomes `Model::PrimaryKey`.
///
/// # Supported attributes
///
/// - `#[table(name = "...")]` — specifies the table name, defaults to the snake_case struct name
/// - `#[column(primary_key)]` — marks the primary key field (required, exactly one)
/// - `#[column(name = "...")]` — overrides the primary key column name (defaults to the field name)
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::Entity;
///
/// #[derive(Entity)]
/// #[table(name = "users")]
/// struct User {
///     #[column(primary_key)]
///     id: i64,
///     name: String,
/// }
///
/// assert_eq!(User::table_name(), "users");
/// assert_eq!(User::pk_name(), "id");
/// ```
#[proc_macro_derive(Entity, attributes(table, column))]
pub fn derive_entity(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_entity_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(FromQueryResult)]` — auto-generate `impl FromQueryResult for Struct`
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_core::FromQueryResult` trait impl.
///
/// Deserializes from a query result row (`HashMap<String, Value>`) into a struct instance.
/// `Option<T>` fields automatically return `None` when the column is missing or the value is NULL.
///
/// # Supported attributes
///
/// - `#[column(name = "...")]` — overrides the column name mapping (defaults to the field name)
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::FromQueryResult;
///
/// #[derive(FromQueryResult)]
/// struct UserRow {
///     id: i64,
///     name: String,
///     #[column(name = "user_email")]
///     email: Option<String>,
/// }
/// ```
#[proc_macro_derive(FromQueryResult, attributes(column))]
pub fn derive_from_query_result(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_from_query_result_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(ColumnEnum)]` — auto-generate column name enum (P2-2)
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates a `<StructName>Column` column-name enum from struct fields (P2-2).
///
/// Each field produces a variant (snake_case → CamelCase); `ColumnTrait::as_str()`
/// returns the database column name; `#[column(name = "...")]` can override the column name
/// (consistent with FromQueryResult). Also implements `std::fmt::Display`.
///
/// # Example
///
/// ```rust,ignore
/// use sz_orm_macros::ColumnEnum;
/// use sz_orm_core::ColumnTrait;
///
/// #[derive(ColumnEnum)]
/// struct User {
///     id: i64,
///     #[column(name = "user_name")]
///     name: String,
/// }
///
/// assert_eq!(UserColumn::Id.as_str(), "id");
/// assert_eq!(UserColumn::Name.as_str(), "user_name");
/// assert_eq!(UserColumn::Id.to_string(), "id");
/// ```
#[proc_macro_derive(ColumnEnum, attributes(column))]
pub fn derive_column_enum(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_column_enum_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(FromRow)]` — auto-generate `impl FromRow for Struct`
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_core::queryable::FromRow` trait impl.
///
/// Deserializes from a `HashMap<String, Value>` by column name into a struct instance.
/// Differs from `FromQueryResult` in that the error type is `QueryError` (includes column info),
/// suitable for low-level scenarios that need precise error localization.
///
/// # Supported attributes
///
/// - `#[column(name = "...")]` — overrides the column name mapping (defaults to the field name)
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::FromRow;
///
/// #[derive(FromRow)]
/// struct User {
///     id: i64,
///     name: String,
///     #[column(name = "user_email")]
///     email: Option<String>,
/// }
/// ```
#[proc_macro_derive(FromRow, attributes(column))]
pub fn derive_from_row(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_from_row_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(SqlType)]` — auto-generate `impl FromQueryResult + to_value()` for enums
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_core::FromQueryResult` trait impl
/// and a `to_value()` method for a Rust enum.
///
/// This is sz-orm's equivalent of SQLx's `#[derive(Type)]`:
/// it lets custom enums be used directly for query-result field mapping and query parameter binding.
///
/// # Supported attributes
///
/// - `#[sql_type(rename_all = "snake_case")]` — controls the serialization format of variant names
///   (snake_case / SCREAMING_SNAKE_CASE / camelCase / PascalCase / lowercase / UPPERCASE)
/// - `#[sql_type(rename = "...")]` (variant level) — overrides the serialization name of a single variant
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::SqlType;
///
/// #[derive(SqlType)]
/// enum Status {
///     Active,    // → "active"
///     Inactive,  // → "inactive"
/// }
///
/// let v = Status::Active.to_value();  // Value::String("active")
/// ```
#[proc_macro_derive(SqlType, attributes(sql_type))]
pub fn derive_sql_type(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_sql_type_impl(input).into()
}

// ---------------------------------------------------------------------------
// `#[derive(Relation)]` — auto-generate `impl ModelExt` with relations()
// ---------------------------------------------------------------------------

/// Derive macro: auto-generates an `sz_orm_core::model::ModelExt` trait impl,
/// populating the `relations()` map and eliminating hand-written relation boilerplate.
///
/// # Supported attributes
///
/// - `#[relation(has_many = "orders", fk = "user_id", pk = "id")]`
/// - `#[relation(belongs_to = "users", fk = "user_id", pk = "id")]`
/// - `#[relation(has_one = "profile", fk = "user_id", pk = "id")]`
/// - `#[relation(belongs_to_many = "roles", junction = "user_roles", fk = "user_id", other_key = "role_id", target = "roles", target_pk = "id")]`
/// - `#[relation(morph_many = "comments", morph_type = "commentable_type", morph_id = "commentable_id", morph_type_value = "Post")]`
/// - `#[relation(morph_to, morph_type = "commentable_type", morph_id = "commentable_id")]`
///
/// # Example
///
/// ```ignore
/// use sz_orm_macros::{Entity, Relation};
///
/// #[derive(Entity, Relation)]
/// #[table(name = "users")]
/// struct User {
///     #[column(primary_key)]
///     id: i64,
/// }
///
/// // Auto-generated:
/// // impl ModelExt for User {
/// //     fn relations() -> HashMap<&str, Relation> {
/// //         // contains the relations declared via #[relation]
/// //     }
/// // }
/// ```
#[proc_macro_derive(Relation, attributes(relation, table, column))]
pub fn derive_relation(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_relation_impl(input).into()
}

/// `#[derive(RelationTrait)]` — auto-generates a `RelationTrait` impl (P-F-2, v2.1.0)
///
/// Generates a static `RelationDef` table + `impl RelationTrait` from `#[relation(...)]` attributes.
/// Shares attribute parsing with `#[derive(Relation)]`, but generates a zero-allocation static slice
/// instead of a `HashMap`.
///
/// # Example
///
/// ```ignore
/// #[derive(RelationTrait)]
/// #[relation(has_many = "Order", fk = "user_id", pk = "id")]
/// struct User { id: i64, name: String }
///
/// // Auto-generated:
/// // static RELATIONS: &[RelationDef] = &[RelationDef::new("Order", "users", "orders", "id", "user_id", HasMany)];
/// // impl RelationTrait for User { ... }
/// ```
#[proc_macro_derive(RelationTrait, attributes(relation, table, column))]
pub fn derive_relation_trait(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    derive::derive_relation_trait_impl(input).into()
}

// ---------------------------------------------------------------------------
// v3.9.0：数据验证派生宏（data-validation feature 隔离）
// ---------------------------------------------------------------------------

/// Derives a `Validate` trait impl.
///
/// Supported `#[validate(...)]` rules:
/// - `email` — email format validation
/// - `required` — non-empty validation
/// - `length(min=N, max=N)` — length range validation
/// - `range(min=N, max=N)` — numeric range validation
/// - `regex(pattern=r"...")` — regex match validation
/// - `contains(value="...")` — contains substring validation
/// - `does_not_contain(value="...")` — does not contain substring validation
/// - `if = "condition"` — conditional validation
///
/// # Example
///
/// ```ignore
/// #[derive(Validate)]
/// struct User {
///     #[validate(email)]
///     email: String,
///     #[validate(length(min=2, max=50))]
///     name: String,
///     #[validate(range(min=0, max=150))]
///     age: i64,
/// }
/// ```
#[cfg(feature = "data-validation")]
#[proc_macro_derive(Validate, attributes(validate))]
pub fn derive_validate(input: TokenStream) -> TokenStream {
    crate::derive_validate::derive_validate_impl(input)
}

// ---------------------------------------------------------------------------
// v4.3.0 M3-T3：`#[derive(Governed)]` — 编译期数据治理（PII 标注强制）
// 通过 `compile-governance` feature（sz-orm-core）→ `governance-derive`（本包）启用。
// ---------------------------------------------------------------------------

/// Derive macro: generates data-governance metadata for a model and **enforces compile-time** PII annotation compliance.
///
/// # Supported attributes
///
/// - `#[pii]` — marks the field as personally identifiable information (PII)
/// - `#[mask(strategy = "...")]` — declares a masking strategy (hash / partial / replace / encrypt)
///
/// # Compile-time enforcement rules
///
/// 1. A `#[pii]` field must also declare `#[mask(strategy = "...")]`, otherwise **compilation fails**
/// 2. The `mask` strategy must be within the whitelist (hash/partial/replace/encrypt), otherwise **compilation fails**
///
/// # Example
///
/// ```ignore
/// use sz_orm_core::governance::GovernedModel;
///
/// #[derive(Governed)]
/// struct User {
///     id: i64,
///     #[pii]
///     #[mask(strategy = "partial")]
///     email: String,
///     #[pii]
///     #[mask(strategy = "hash")]
///     phone: String,
///     name: String, // non-PII, no mask needed
/// }
///
/// assert_eq!(
///     User::pii_fields(),
///     vec![("email", "partial"), ("phone", "hash")]
/// );
/// ```
#[cfg(feature = "governance-derive")]
#[proc_macro_derive(Governed, attributes(pii, mask))]
pub fn derive_governed(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as syn::DeriveInput);
    let name = &input.ident;

    const VALID_STRATEGIES: [&str; 4] = ["hash", "partial", "replace", "encrypt"];

    let mut pii_fields: Vec<(String, String)> = Vec::new();
    let mut errors: Vec<syn::Error> = Vec::new();

    if let syn::Data::Struct(data) = &input.data {
        for field in &data.fields {
            let Some(field_name) = field.ident.as_ref().map(|i| i.to_string()) else {
                continue;
            };
            let is_pii = field.attrs.iter().any(|a| a.path().is_ident("pii"));

            // 解析 #[mask(strategy = "xxx")]
            let mut mask_strategy: Option<String> = None;
            for attr in &field.attrs {
                if !attr.path().is_ident("mask") {
                    continue;
                }
                let _ = attr.parse_nested_meta(|meta| {
                    if meta.path.is_ident("strategy") {
                        let lit: syn::LitStr = meta.value()?.parse()?;
                        mask_strategy = Some(lit.value());
                        Ok(())
                    } else {
                        Err(meta.error("unsupported #[mask] attribute, only 'strategy' is allowed"))
                    }
                });
            }

            if is_pii {
                match mask_strategy {
                    Some(strategy) => {
                        if !VALID_STRATEGIES.contains(&strategy.as_str()) {
                            errors.push(syn::Error::new_spanned(
                                field,
                                format!(
                                    "invalid #[mask(strategy = \"{strategy}\")]: allowed strategies are {:?}",
                                    VALID_STRATEGIES
                                ),
                            ));
                        } else {
                            pii_fields.push((field_name, strategy));
                        }
                    }
                    None => errors.push(syn::Error::new_spanned(
                        field,
                        "#[pii] field must declare #[mask(strategy = \"...\")]",
                    )),
                }
            }
        }
    }

    if !errors.is_empty() {
        // 每条错误独立转 compile_error!，合并为一个 TokenStream
        let err_tokens: proc_macro2::TokenStream =
            errors.iter().map(|e| e.to_compile_error()).collect();
        return err_tokens.into();
    }

    let entries = pii_fields.iter().map(|(f, s)| {
        let f = f.as_str();
        let s = s.as_str();
        quote::quote!((#f, #s))
    });

    quote::quote! {
        impl ::sz_orm_core::governance::GovernedModel for #name {
            fn pii_fields() -> Vec<(&'static str, &'static str)> {
                vec![#(#entries),*]
            }
        }
    }
    .into()
}

// ---------------------------------------------------------------------------
// v4.3.0 M2：`#[detect_n_plus_one]` — N+1 静态检测标注宏（n1-lint feature）
// 分析函数体 AST，检测循环（for/while）内查询调用，编译期输出警告。
// 分析逻辑复用 sz-orm-n1-lint（与 CLI 批量扫描共用，避免重复实现）。
// ---------------------------------------------------------------------------

/// Attribute macro: analyzes a function body, detects N+1 query patterns, and emits compile-time warnings (non-blocking).
///
/// Detection patterns (conservative whitelist to avoid false positives):
/// - `find_by_*` / `where_eq` / `query` and other query calls inside loop bodies → `query-in-loop` warning
/// - Query calls inside conditional branches within loops → `query-in-loop` warning
///
/// Warnings are emitted via `eprintln!` (`warning: [sz-orm-n1-lint] ...`); they do not block compilation;
/// the function is passed through unchanged (zero runtime impact).
///
/// # Example
///
/// ```ignore
/// #[detect_n_plus_one]
/// fn process_users(users: Vec<User>) {
///     for user in users {
///         let orders = Order::find_by_user(user.id); // ⚠️ compile-time warning: query-in-loop
///     }
/// }
/// ```
#[cfg(feature = "n1-lint")]
#[proc_macro_attribute]
pub fn detect_n_plus_one(_attr: TokenStream, item: TokenStream) -> TokenStream {
    let item_fn = parse_macro_input!(item as syn::ItemFn);
    let findings = sz_orm_n1_lint::analyze_fn(&item_fn);
    for f in &findings {
        eprintln!(
            "warning: [sz-orm-n1-lint] {} at line {}: {}",
            f.pattern.as_str(),
            f.line,
            f.message
        );
    }
    quote::quote!(#item_fn).into()
}

/// `migrate!` 宏 — 编译时创建迁移并验证 SQL 语法
///
/// 对标 SQLx `migrate!` / Diesel `embed_migrations!`。
///
/// # 用法
///
/// ```ignore
/// use sz_orm_macros::migrate;
///
/// let m = migrate!("001", "create_users",
///     "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)",
///     "DROP TABLE users");
/// ```
///
/// 编译时验证 up/down SQL 语法，生成 `sz_orm_core::migration::Migration`。
#[proc_macro]
pub fn migrate(input: TokenStream) -> TokenStream {
    use syn::parse::Parser;
    let args = match syn::punctuated::Punctuated::<syn::LitStr, syn::Token![,]>::parse_terminated
        .parse2(input.into())
    {
        Ok(args) => args,
        Err(e) => return e.to_compile_error().into(),
    };

    if args.len() != 4 {
        return syn::Error::new(
            proc_macro2::Span::call_site(),
            "migrate! expects exactly 4 arguments: version, name, sql_up, sql_down",
        )
        .to_compile_error()
        .into();
    }

    let args_vec: Vec<_> = args.into_iter().collect();
    let sql_up = args_vec[2].value();
    let sql_down = args_vec[3].value();

    if let Err(e) = validate_balanced_parens(&sql_up) {
        return syn::Error::new(args_vec[2].span(), format!("sql_up validation failed: {e}"))
            .to_compile_error()
            .into();
    }
    if let Err(e) = validate_string_literals_closed(&sql_up) {
        return syn::Error::new(args_vec[2].span(), format!("sql_up validation failed: {e}"))
            .to_compile_error()
            .into();
    }
    if !sql_down.is_empty() {
        if let Err(e) = validate_balanced_parens(&sql_down) {
            return syn::Error::new(
                args_vec[3].span(),
                format!("sql_down validation failed: {e}"),
            )
            .to_compile_error()
            .into();
        }
        if let Err(e) = validate_string_literals_closed(&sql_down) {
            return syn::Error::new(
                args_vec[3].span(),
                format!("sql_down validation failed: {e}"),
            )
            .to_compile_error()
            .into();
        }
    }

    let version_lit = args_vec[0].clone();
    let name_lit = args_vec[1].clone();
    let up_lit = args_vec[2].clone();
    let down_lit = args_vec[3].clone();

    quote::quote! {
        ::sz_orm_core::migration::Migration::new(
            #version_lit, #name_lit, #up_lit, #down_lit,
        )
    }
    .into()
}

// ---------------------------------------------------------------------------
// Unit tests — cover helper functions used by both macros
// ---------------------------------------------------------------------------

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

    // ---- strip_string_literal ----

    #[test]
    fn test_strip_plain_double_quoted() {
        assert_eq!(strip_string_literal(r#""hello""#), Some("hello"));
    }

    #[test]
    fn test_strip_raw_double_hash() {
        assert_eq!(strip_string_literal(r###"r#"hello"#"###), Some("hello"));
    }

    #[test]
    fn test_strip_raw_double_no_hash() {
        assert_eq!(strip_string_literal(r#"r"hello""#), Some("hello"));
    }

    #[test]
    fn test_strip_byte_string() {
        assert_eq!(strip_string_literal(r#"b"hello""#), Some("hello"));
        assert_eq!(strip_string_literal(r#"b'hello'"#), Some("hello"));
    }

    #[test]
    fn test_strip_non_string_returns_none() {
        assert_eq!(strip_string_literal("123"), None);
        assert_eq!(strip_string_literal("foo"), None);
    }

    // ---- validate_sql_content ----

    #[test]
    fn test_validate_select_with_from_ok() {
        assert!(validate_sql_content("SELECT * FROM users", None).is_ok());
    }

    #[test]
    fn test_validate_select_missing_from_fails() {
        assert!(validate_sql_content("SELECT * users", None).is_err());
    }

    #[test]
    fn test_validate_insert_missing_into_fails() {
        assert!(validate_sql_content("INSERT INTO users VALUES (1)", None).is_ok());
        assert!(validate_sql_content("INSERT users VALUES (1)", None).is_err());
    }

    #[test]
    fn test_validate_update_missing_set_fails() {
        assert!(validate_sql_content("UPDATE users SET name='a'", None).is_ok());
        assert!(validate_sql_content("UPDATE users name='a'", None).is_err());
    }

    #[test]
    fn test_validate_delete_missing_from_fails() {
        assert!(validate_sql_content("DELETE FROM users WHERE id=1", None).is_ok());
        assert!(validate_sql_content("DELETE users WHERE id=1", None).is_err());
    }

    #[test]
    fn test_validate_empty_sql_fails() {
        assert!(validate_sql_content("", None).is_err());
        assert!(validate_sql_content("   ", None).is_err());
    }

    // ---- balanced parens ----

    #[test]
    fn test_validate_balanced_parens_ok() {
        assert!(validate_balanced_parens("SELECT * FROM (SELECT * FROM t)").is_ok());
    }

    #[test]
    fn test_validate_balanced_parens_unbalanced() {
        assert!(validate_balanced_parens("SELECT * FROM (t").is_err());
        assert!(validate_balanced_parens("SELECT * FROM t)").is_err());
    }

    // ---- injection patterns ----

    #[test]
    fn test_validate_no_injection_clean() {
        assert!(validate_no_injection("SELECT * FROM users WHERE id = 1").is_ok());
    }

    #[test]
    fn test_validate_no_injection_drop_table() {
        assert!(validate_no_injection("'; DROP TABLE users; --").is_err());
    }

    #[test]
    fn test_validate_no_injection_or_1_1() {
        // 编译期 SQL 已剥离外层引号，检测模式不再依赖引号字符。
        // "' OR '1'='1" 因引号分隔不再匹配 "or 1=1"，故不再检测；
        // 但不含引号分隔的 "OR 1=1" 仍可被检测。
        assert!(validate_no_injection("' OR 1=1").is_err());
        assert!(validate_no_injection("WHERE id = 1 OR 1=1").is_err());
    }

    #[test]
    fn test_validate_no_injection_drop_database() {
        assert!(validate_no_injection("SELECT x; DROP DATABASE db").is_err());
    }

    #[test]
    fn test_validate_no_injection_information_schema() {
        assert!(validate_no_injection("SELECT * FROM information_schema.tables").is_err());
    }

    #[test]
    fn test_validate_no_injection_xp_cmdshell() {
        assert!(validate_no_injection("EXEC xp_cmdshell 'dir'").is_err());
    }

    #[test]
    fn test_validate_no_injection_union_select() {
        assert!(validate_no_injection("1 UNION SELECT * FROM users").is_err());
    }

    #[test]
    fn test_validate_no_injection_comment_dashes() {
        assert!(validate_no_injection("SELECT * FROM users -- comment").is_err());
    }

    #[test]
    fn test_validate_no_injection_block_comment() {
        assert!(validate_no_injection("SELECT /* x */ * FROM users").is_err());
    }

    // ---- string literal closure ----

    #[test]
    fn test_validate_string_literals_closed_ok() {
        assert!(validate_string_literals_closed("'hello' = 'world'").is_ok());
        assert!(validate_string_literals_closed(r#""foo" = "bar""#).is_ok());
    }

    #[test]
    fn test_validate_string_literals_closed_unclosed_single() {
        assert!(validate_string_literals_closed("'hello").is_err());
    }

    #[test]
    fn test_validate_string_literals_closed_unclosed_double() {
        assert!(validate_string_literals_closed(r#""hello"#).is_err());
    }

    // ---- param count check ----

    #[test]
    fn test_validate_param_count_match() {
        assert!(validate_sql_content("SELECT * FROM users WHERE id = ?", Some(1)).is_ok());
        assert!(
            validate_sql_content("SELECT * FROM users WHERE id = ? AND name = ?", Some(2)).is_ok()
        );
    }

    #[test]
    fn test_validate_param_count_mismatch() {
        assert!(validate_sql_content("SELECT * FROM users WHERE id = ?", Some(2)).is_err());
        assert!(
            validate_sql_content("SELECT * FROM users WHERE id = ? AND name = ?", Some(1)).is_err()
        );
    }

    // ---- db-verify feature: detect_db_kind ----

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_mysql() {
        assert_eq!(
            detect_db_kind("mysql://user:pass@host:3306/db").unwrap(),
            DbKind::MySql
        );
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_postgres() {
        assert_eq!(
            detect_db_kind("postgres://user:pass@host:5432/db").unwrap(),
            DbKind::Postgres
        );
        assert_eq!(
            detect_db_kind("postgresql://user:pass@host:5432/db").unwrap(),
            DbKind::Postgres
        );
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_sqlite() {
        assert_eq!(
            detect_db_kind("sqlite://path/to/db.db").unwrap(),
            DbKind::Sqlite
        );
        assert_eq!(detect_db_kind("sqlite::memory:").unwrap(), DbKind::Sqlite);
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_oracle() {
        assert_eq!(
            detect_db_kind("oracle://sys:test123@127.0.0.1:1521/freepdb1.FALSE?sysdba=1").unwrap(),
            DbKind::Oracle
        );
        assert_eq!(
            detect_db_kind("oracle:sys:test123@127.0.0.1:1521/FREE").unwrap(),
            DbKind::Oracle
        );
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_sqlserver() {
        assert_eq!(
            detect_db_kind("sqlserver://test:pass@host:1433/db").unwrap(),
            DbKind::SqlServer
        );
        assert_eq!(
            detect_db_kind("mssql://test:pass@host:1433/db").unwrap(),
            DbKind::SqlServer
        );
        assert_eq!(
            detect_db_kind("tds://test:pass@host:1433/db").unwrap(),
            DbKind::SqlServer
        );
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_detect_db_kind_unsupported() {
        assert!(detect_db_kind("redis://user:pass@host/db").is_err());
        assert!(detect_db_kind("not-a-url").is_err());
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_parse_oracle_dsn_basic() {
        let dsn = "oracle://sys:test123@127.0.0.1:1521/freepdb1.FALSE?sysdba=1";
        let p = parse_oracle_dsn(dsn).unwrap();
        assert_eq!(p.user, "sys");
        assert_eq!(p.password, "test123");
        assert_eq!(p.host, "127.0.0.1");
        assert_eq!(p.port, 1521);
        assert_eq!(p.service, "freepdb1.FALSE");
        assert!(p.sysdba);
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_parse_oracle_dsn_default_port() {
        // 无端口号时默认 1521
        let dsn = "oracle://sys:test123@127.0.0.1/FREE";
        let p = parse_oracle_dsn(dsn).unwrap();
        assert_eq!(p.port, 1521);
        assert_eq!(p.service, "FREE");
        assert!(!p.sysdba);
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_parse_sqlserver_dsn_basic() {
        let dsn =
            "sqlserver://test:JkbC2jsaWAYDe2Gz@sh-mssql-adrul9nm.sql.tencentcdb.com:22527/test";
        let p = parse_sqlserver_dsn(dsn).unwrap();
        assert_eq!(p.user, "test");
        assert_eq!(p.password, "JkbC2jsaWAYDe2Gz");
        assert_eq!(p.host, "sh-mssql-adrul9nm.sql.tencentcdb.com");
        assert_eq!(p.port, 22527);
        assert_eq!(p.database, "test");
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_parse_sqlserver_dsn_default_port() {
        let dsn = "mssql://user:pass@host/db";
        let p = parse_sqlserver_dsn(dsn).unwrap();
        assert_eq!(p.port, 1433);
        assert_eq!(p.database, "db");
    }

    // ---- schema! 宏 parse_create_table 测试 ----

    #[test]
    fn test_parse_create_table_basic() {
        let sql = "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL)";
        let (table, cols) = parse_create_table(sql).unwrap();
        assert_eq!(table, "users");
        assert_eq!(
            cols,
            vec![
                ("id".to_string(), "i32".to_string()),
                ("name".to_string(), "String".to_string())
            ]
        );
    }

    #[test]
    fn test_parse_create_table_with_if_not_exists() {
        let sql = "CREATE TABLE IF NOT EXISTS `orders` (`id` BIGINT PRIMARY KEY, `total` DECIMAL(10,2) NOT NULL)";
        let (table, cols) = parse_create_table(sql).unwrap();
        assert_eq!(table, "orders");
        assert_eq!(
            cols,
            vec![
                ("id".to_string(), "i64".to_string()),
                ("total".to_string(), "f64".to_string())
            ]
        );
    }

    #[test]
    fn test_parse_create_table_nullable() {
        let sql = "CREATE TABLE t (a INT NOT NULL, b INT)";
        let (_, cols) = parse_create_table(sql).unwrap();
        assert_eq!(cols[0], ("a".to_string(), "i32".to_string()));
        assert_eq!(cols[1], ("b".to_string(), "Option<i32>".to_string()));
    }

    #[test]
    fn test_parse_create_table_skip_constraints() {
        let sql = "CREATE TABLE t (id INT PRIMARY KEY, name TEXT, PRIMARY KEY (id), CONSTRAINT fk1 FOREIGN KEY (x) REFERENCES y(id))";
        let (_, cols) = parse_create_table(sql).unwrap();
        assert_eq!(cols.len(), 2);
        assert_eq!(cols[0].0, "id");
        assert_eq!(cols[1].0, "name");
    }

    #[test]
    fn test_parse_create_table_varchar_with_len() {
        let sql = "CREATE TABLE t (name VARCHAR(255) NOT NULL, code CHAR(10))";
        let (_, cols) = parse_create_table(sql).unwrap();
        assert_eq!(cols[0], ("name".to_string(), "String".to_string()));
        assert_eq!(cols[1], ("code".to_string(), "Option<String>".to_string()));
    }

    #[test]
    fn test_sql_type_to_rust_mappings() {
        // 整数（按字节宽度严格映射，与 SQL 标准一致）
        assert_eq!(sql_type_to_rust("BIGINT", false), "i64");
        assert_eq!(sql_type_to_rust("INT8", false), "i64");
        assert_eq!(sql_type_to_rust("INT", false), "i32");
        assert_eq!(sql_type_to_rust("INTEGER", false), "i32");
        assert_eq!(sql_type_to_rust("INT4", false), "i32");
        assert_eq!(sql_type_to_rust("SERIAL", false), "i32");
        assert_eq!(sql_type_to_rust("SMALLINT", false), "i16");
        assert_eq!(sql_type_to_rust("INT2", false), "i16");
        assert_eq!(sql_type_to_rust("SMALLSERIAL", false), "i16");
        assert_eq!(sql_type_to_rust("TINYINT", false), "i8");
        // 浮点
        assert_eq!(sql_type_to_rust("FLOAT", false), "f32");
        assert_eq!(sql_type_to_rust("REAL", false), "f32");
        assert_eq!(sql_type_to_rust("FLOAT4", false), "f32");
        assert_eq!(sql_type_to_rust("DOUBLE", false), "f64");
        assert_eq!(sql_type_to_rust("DOUBLE PRECISION", false), "f64");
        assert_eq!(sql_type_to_rust("FLOAT8", false), "f64");
        assert_eq!(sql_type_to_rust("DECIMAL", false), "f64");
        assert_eq!(sql_type_to_rust("NUMERIC", false), "f64");
        // 布尔
        assert_eq!(sql_type_to_rust("BOOLEAN", false), "bool");
        assert_eq!(sql_type_to_rust("BOOL", false), "bool");
        // 字符串
        assert_eq!(sql_type_to_rust("VARCHAR", false), "String");
        assert_eq!(sql_type_to_rust("TEXT", false), "String");
        assert_eq!(sql_type_to_rust("CHAR", false), "String");
        assert_eq!(sql_type_to_rust("UUID", false), "String");
        assert_eq!(sql_type_to_rust("DATE", false), "String");
        assert_eq!(sql_type_to_rust("DATETIME", false), "String");
        assert_eq!(sql_type_to_rust("TIMESTAMP", false), "String");
        assert_eq!(sql_type_to_rust("JSON", false), "String");
        assert_eq!(sql_type_to_rust("JSONB", false), "String");
        // 二进制
        assert_eq!(sql_type_to_rust("BLOB", false), "Vec<u8>");
        assert_eq!(sql_type_to_rust("BYTEA", false), "Vec<u8>");
        assert_eq!(sql_type_to_rust("BINARY", false), "Vec<u8>");
        assert_eq!(sql_type_to_rust("VARBINARY", false), "Vec<u8>");
        // nullable
        assert_eq!(sql_type_to_rust("INT", true), "Option<i32>");
        assert_eq!(sql_type_to_rust("BIGINT", true), "Option<i64>");
        assert_eq!(sql_type_to_rust("VARCHAR", true), "Option<String>");
        assert_eq!(sql_type_to_rust("BLOB", true), "Option<Vec<u8>>");
        // unknown
        assert_eq!(sql_type_to_rust("UNKNOWNTYPE", false), "String");
    }

    #[test]
    fn test_parse_create_table_error_no_create() {
        assert!(parse_create_table("SELECT * FROM users").is_err());
    }

    #[test]
    fn test_parse_create_table_error_no_parens() {
        assert!(parse_create_table("CREATE TABLE foo").is_err());
    }

    // -----------------------------------------------------------------------
    // Gap 1 测试：列名/表名提取
    // -----------------------------------------------------------------------

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_extract_tables_simple() {
        let tables = extract_tables("SELECT id, name FROM users WHERE id = ?");
        assert!(tables.contains(&"users".to_string()));
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_extract_tables_multiple() {
        let tables = extract_tables(
            "SELECT u.id, o.total FROM users u JOIN orders o ON u.id = o.user_id WHERE u.id = ?",
        );
        assert!(tables.contains(&"users".to_string()));
        assert!(tables.contains(&"orders".to_string()));
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_extract_columns_select_and_where() {
        let cols =
            extract_columns("SELECT id, name FROM users WHERE email = ? ORDER BY created_at");
        // id, name from SELECT; email from WHERE; created_at from ORDER BY
        assert!(cols.contains(&"id".to_string()));
        assert!(cols.contains(&"name".to_string()));
        assert!(cols.contains(&"email".to_string()));
        assert!(cols.contains(&"created_at".to_string()));
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_extract_columns_skips_keywords() {
        let cols = extract_columns("SELECT COUNT(id), name FROM users WHERE status = ?");
        // COUNT is a function, should be skipped
        assert!(!cols.contains(&"count".to_string()));
        assert!(cols.contains(&"id".to_string()));
        assert!(cols.contains(&"name".to_string()));
        assert!(cols.contains(&"status".to_string()));
    }

    #[cfg(feature = "db-verify")]
    #[test]
    fn test_is_sql_function() {
        assert!(is_sql_function("COUNT"));
        assert!(is_sql_function("now"));
        assert!(is_sql_function("COALESCE"));
        assert!(!is_sql_function("name"));
        assert!(!is_sql_function("user_id"));
    }
}