use std::collections::{HashMap, HashSet};

use crate::column::{order_to_sql, ColumnValue, ColumnValueKind, OrderSpec};
use crate::condition::{compile_conditions, Condition};
use crate::driver::mysql::MysqlSqlBuilder;
use crate::driver::postgres::PostgresSqlBuilder;
use crate::driver::sqlite::SqliteSqlBuilder;
use crate::driver::{OnConflict, OnConflictAction, QueryResult, Row, SqlBuilder};
use crate::error::{Error, LimitContext, Result};
use crate::middleware::OperationType;
use crate::types::{
    ConflictUpdate, DeleteOptions, DriverType, InsertOptions, LimitConfig, PkeyResult,
    SelectOptions, UpdateManyOptions, UpdateOptions,
};
use crate::values::Value;

/// Metadata for a column in a model.
#[derive(Debug, Clone)]
pub struct ColumnMeta {
    pub column_name: &'static str,
    pub property_name: &'static str,
    pub primary_key: bool,
    pub auto_increment: bool,
    pub sql_cast: Option<&'static str>,
    pub type_cast: Option<&'static str>,
    /// When true, the column has a database-level default. SKIP on create omits it (DB default used).
    pub has_default: bool,
}

/// Core trait implemented by all model structs via `#[derive(Model)]`.
pub trait Model: Sized + Send + Sync + FromRow + 'static {
    type Columns;

    const TABLE_NAME: &'static str;

    fn table_name() -> &'static str {
        Self::TABLE_NAME
    }

    fn update_table_name() -> &'static str {
        Self::table_name()
    }

    fn column_meta() -> &'static [ColumnMeta];

    fn primary_key_columns() -> &'static [&'static str];

    fn columns() -> Self::Columns;

    fn default_order() -> Option<Vec<OrderSpec>> {
        None
    }

    /// Raw ORDER BY clause (e.g. "created_at DESC").
    /// When set via #[model(order = "...")], used when no order is specified in SelectOptions.
    fn default_order_raw() -> Option<&'static str> {
        None
    }

    fn default_filter() -> Option<Vec<Condition>> {
        None
    }

    fn select_columns() -> &'static str {
        "*"
    }

    /// SQL query for query-based models (view models, aggregations, etc.).
    /// When defined, the model uses this query as a CTE instead of a table.
    /// Use `#[model(query = "...")]` in the derive macro.
    fn query_sql() -> Option<&'static str> {
        None
    }

    /// Pre-built SELECT SQL for the common case (no user conditions, no options).
    /// Generated by `#[derive(Model)]` at compile time. Includes default_filter and order
    /// when defined. Returns `None` for manually implemented models (falls back to dynamic).
    fn static_select_sql() -> Option<&'static str> {
        None
    }

    /// Pre-built COUNT SQL for the common case (no user conditions).
    /// Generated by `#[derive(Model)]` at compile time. Includes default_filter when defined.
    fn static_count_sql() -> Option<&'static str> {
        None
    }

    /// Returns a function to deserialize a `tokio_postgres::Row` directly into this model,
    /// bypassing the intermediate HashMap. Override via `#[derive(Model)]` for the fast path.
    #[cfg(feature = "postgres")]
    fn pg_row_converter() -> Option<fn(&tokio_postgres::Row) -> Option<Self>> {
        None
    }

    /// Set up lazy batch loading contexts for all records from the same query.
    /// Called by `find()` after constructing records.
    /// Override generated by `#[derive(Model)]` for models with relations.
    fn setup_batch_loading(
        _records: &mut [Self],
        _executor: &dyn Executor,
        _limit_config: Option<&LimitConfig>,
    ) {
        // Default: no-op for models without relations
    }
}

/// Trait for converting a database row into a model instance.
pub trait FromRow: Sized {
    fn from_row(row: &Row) -> Result<Self>;
}

/// Trait for converting a model instance into column-value pairs.
pub trait IntoColumnValues {
    fn into_column_values(self) -> Vec<ColumnValue>;
}

// ============================================
// PostgreSQL direct deserialization support
// ============================================

/// Trait for converting a PostgreSQL row directly into a model instance,
/// bypassing the intermediate HashMap<String, Value> representation.
#[cfg(feature = "postgres")]
pub trait FromPgRow: Sized {
    fn from_pg_row(row: &tokio_postgres::Row) -> Option<Self>;
}

/// Helper trait for extracting typed values directly from a `tokio_postgres::Row`.
/// Used by the `#[derive(Model)]` macro to generate direct PG deserialization.
#[cfg(feature = "postgres")]
pub trait PgGet: Sized {
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self>;
}

#[cfg(feature = "postgres")]
impl PgGet for bool {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for i32 {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for i64 {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get::<_, i64>(col)
            .ok()
            .or_else(|| row.try_get::<_, i32>(col).ok().map(|v| v as i64))
    }
}

#[cfg(feature = "postgres")]
impl PgGet for f32 {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for f64 {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get::<_, f64>(col)
            .ok()
            .or_else(|| row.try_get::<_, f32>(col).ok().map(|v| v as f64))
    }
}

#[cfg(feature = "postgres")]
impl PgGet for String {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for Vec<u8> {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for serde_json::Value {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for uuid::Uuid {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl PgGet for chrono::NaiveDateTime {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get::<_, chrono::NaiveDateTime>(col)
            .ok()
            .or_else(|| {
                row.try_get::<_, chrono::DateTime<chrono::Utc>>(col)
                    .ok()
                    .map(|dt| dt.naive_utc())
            })
    }
}

#[cfg(feature = "postgres")]
impl PgGet for chrono::NaiveDate {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        row.try_get(col).ok()
    }
}

#[cfg(feature = "postgres")]
impl<T: PgGet> PgGet for Option<T> {
    #[inline]
    fn pg_get(row: &tokio_postgres::Row, col: &str) -> Option<Self> {
        Some(T::pg_get(row, col))
    }
}

// ============================================
// Executor trait -- abstracts over DbHandler and Transaction
// ============================================

/// Trait for anything that can execute SQL queries (pool or transaction).
/// Implemented by `DbHandler`, `TransactionExecutor`, and `WriterExecutor`.
pub trait Executor: Send + Sync {
    fn execute_query<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>>;

    /// Execute query with operation type for method-level middleware dispatch.
    /// Default implementation calls execute_query.
    fn execute_query_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        _op: OperationType,
        _table: &'a str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_query(sql, params)
    }

    fn execute_write<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>>;

    /// Execute write with operation type for method-level middleware dispatch.
    /// Default implementation calls execute_write.
    fn execute_write_with_op<'a>(
        &'a self,
        sql: &'a str,
        params: &'a [Value],
        _op: OperationType,
        _table: &'a str,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>> {
        self.execute_write(sql, params)
    }

    fn driver_type(&self) -> crate::types::DriverType;

    /// Returns true if this executor is bound to a transaction.
    fn is_transaction(&self) -> bool {
        false
    }

    /// Returns true if this executor is in a read-only context (e.g., withWriter).
    fn is_read_only(&self) -> bool {
        false
    }

    /// Execute a query and return raw `tokio_postgres::Row` values for direct deserialization.
    /// Returns `None` when the fast path is unavailable (non-PG driver, middleware present, etc.).
    #[cfg(feature = "postgres")]
    fn execute_pg_rows<'a>(
        &'a self,
        _sql: &'a str,
        _params: &'a [Value],
    ) -> Option<
        std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<Vec<tokio_postgres::Row>>> + Send + 'a>,
        >,
    > {
        None
    }

    /// Create a shared clone of this executor for lazy relation loading.
    /// Called by `find()` to attach a batch-loading context to returned records.
    fn clone_arc(&self) -> std::sync::Arc<dyn Executor> {
        panic!(
            "clone_arc() not implemented for this Executor. \
             Required for automatic batch loading of relations."
        )
    }
}

// ============================================
// SQL Building Functions
// ============================================

/// Build a SELECT SQL statement.
pub fn build_select_sql<M: Model>(
    conditions: &[Condition],
    options: Option<&SelectOptions>,
) -> (String, Vec<Value>) {
    if conditions.is_empty() && options.is_none() {
        if let Some(sql) = M::static_select_sql() {
            return (sql.to_string(), vec![]);
        }
    }

    let is_query_based = M::query_sql().is_some();
    let cte_alias = M::table_name();
    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(if is_query_based {
            cte_alias
        } else {
            M::table_name()
        });

    let select_cols = options
        .and_then(|o| o.select.as_deref())
        .unwrap_or(M::select_columns());

    let mut all_conditions = Vec::new();
    if let Some(filter) = M::default_filter() {
        all_conditions.extend(filter);
    }
    all_conditions.extend_from_slice(conditions);

    let compiled = compile_conditions(&all_conditions);
    let mut params = Vec::new();

    let mut sql = String::new();

    // CTE: custom CTE from options, then query-based CTE from model
    let mut cte_parts: Vec<String> = Vec::new();
    if let Some(opts) = options {
        if let Some(ctes) = &opts.cte {
            for c in ctes {
                cte_parts.push(format!("{} AS ({})", c.name, c.sql));
                params.extend(c.params.iter().cloned());
            }
        }
    }
    if let Some(query_sql) = M::query_sql() {
        cte_parts.push(format!("{} AS ({})", cte_alias, query_sql));
    }
    if !cte_parts.is_empty() {
        sql.push_str(&format!("WITH {} ", cte_parts.join(", ")));
    }

    sql.push_str(&format!("SELECT {} FROM {}", select_cols, table));

    // JOIN
    if let Some(opts) = options {
        if let Some(joins) = &opts.join {
            for j in joins {
                sql.push(' ');
                sql.push_str(j);
            }
        }
        if let Some(left_joins) = &opts.left_join {
            for j in left_joins {
                sql.push(' ');
                sql.push_str(j);
            }
        }
        if opts.join.is_some() || opts.left_join.is_some() {
            params.extend(opts.join_params.iter().cloned());
        }
    }

    // WHERE
    if !compiled.sql.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(&compiled.sql);
        params.extend(compiled.params);
    }

    // GROUP BY
    if let Some(opts) = options {
        if let Some(group) = &opts.group {
            sql.push_str(&format!(" GROUP BY {}", group));
        }
    }

    // HAVING
    if let Some(opts) = options {
        if let Some(having) = &opts.having {
            sql.push_str(&format!(" HAVING {}", having));
        }
    }

    // ORDER BY: options.order overrides default, but if options has no order, fall back to default
    let order_sql = if let Some(opts) = options {
        if let Some(ref raw) = opts.order_raw {
            Some(raw.clone())
        } else if let Some(ref o) = opts.order {
            Some(order_to_sql(o))
        } else {
            M::default_order_raw()
                .map(|s| s.to_string())
                .or_else(|| M::default_order().map(|o| order_to_sql(&o)))
        }
    } else {
        M::default_order_raw()
            .map(|s| s.to_string())
            .or_else(|| M::default_order().map(|o| order_to_sql(&o)))
    };
    if let Some(order) = order_sql {
        sql.push_str(&format!(" ORDER BY {}", order));
    }

    // LIMIT
    if let Some(opts) = options {
        if let Some(limit) = opts.limit {
            sql.push_str(&format!(" LIMIT {}", limit));
        }
    }

    // OFFSET
    if let Some(opts) = options {
        if let Some(offset) = opts.offset {
            sql.push_str(&format!(" OFFSET {}", offset));
        }
    }

    // FOR UPDATE / FOR SHARE
    // NOTE: FOR SHARE is a Rust-only extension not present in the TS version.
    // The TS version only supports forUpdate.
    if options.is_some_and(|o| o.for_update == Some(true)) {
        sql.push_str(" FOR UPDATE");
    } else if options.is_some_and(|o| o.for_share == Some(true)) {
        sql.push_str(" FOR SHARE");
    }

    // APPEND
    if let Some(opts) = options {
        if let Some(append) = &opts.append {
            sql.push(' ');
            sql.push_str(append);
        }
    }

    (sql, params)
}

/// Build an INSERT SQL statement.
/// `driver_type` determines ON CONFLICT syntax: PostgreSQL uses ON CONFLICT,
/// MySQL uses INSERT IGNORE/ON DUPLICATE KEY UPDATE, SQLite uses INSERT OR IGNORE/ON CONFLICT.
pub fn build_insert_sql(
    table: &str,
    column_values: &[ColumnValue],
    options: Option<&InsertOptions>,
    driver_type: DriverType,
) -> (String, Vec<Value>) {
    let active: Vec<&ColumnValue> = column_values.iter().filter(|cv| !cv.is_skip()).collect();

    let columns: Vec<&str> = active.iter().map(|cv| cv.column_name).collect();

    let mut params = Vec::new();
    let placeholders: Vec<String> = active
        .iter()
        .map(|cv| {
            if let ColumnValueKind::Value(v) = &cv.value {
                match v {
                    Value::Raw(expr) => expr.clone(),
                    Value::Immediate(expr) => expr.clone(),
                    _ => {
                        params.push(v.clone());
                        "?".to_string()
                    }
                }
            } else {
                unreachable!()
            }
        })
        .collect();

    let write_table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(table);

    let cols_str = columns.join(", ");
    let vals_str = placeholders.join(", ");

    // Determine INSERT prefix and ON CONFLICT handling per driver
    let (insert_prefix, conflict_suffix) = if let Some(opts) = options {
        if opts.on_conflict_columns.is_empty() {
            ("INSERT INTO".to_string(), String::new())
        } else {
            let mut update_cols: Vec<&str> = match &opts.on_conflict_update {
                Some(ConflictUpdate::All) => columns.to_vec(),
                Some(ConflictUpdate::Columns(cols)) => cols.iter().map(|s| s.as_str()).collect(),
                None if opts.on_conflict_ignore => vec![],
                None => vec![],
            };
            // Fallback: when DoUpdate has empty columns, use all inserted columns (matches PG driver)
            if !opts.on_conflict_ignore
                && opts.on_conflict_update.is_some()
                && update_cols.is_empty()
            {
                update_cols = columns.to_vec();
            }

            match driver_type {
                DriverType::Postgres => {
                    let suffix = if opts.on_conflict_ignore {
                        format!(
                            " ON CONFLICT ({}) DO NOTHING",
                            opts.on_conflict_columns.join(", ")
                        )
                    } else if !update_cols.is_empty() {
                        let set_parts: Vec<String> = update_cols
                            .iter()
                            .map(|c| format!("{} = EXCLUDED.{}", c, c))
                            .collect();
                        format!(
                            " ON CONFLICT ({}) DO UPDATE SET {}",
                            opts.on_conflict_columns.join(", "),
                            set_parts.join(", ")
                        )
                    } else {
                        String::new()
                    };
                    ("INSERT INTO".to_string(), suffix)
                }
                DriverType::Mysql => {
                    if opts.on_conflict_ignore {
                        ("INSERT IGNORE INTO".to_string(), String::new())
                    } else if !update_cols.is_empty() {
                        let set_parts: Vec<String> = update_cols
                            .iter()
                            .map(|c| format!("{} = VALUES({})", c, c))
                            .collect();
                        (
                            "INSERT INTO".to_string(),
                            format!(" ON DUPLICATE KEY UPDATE {}", set_parts.join(", ")),
                        )
                    } else {
                        ("INSERT INTO".to_string(), String::new())
                    }
                }
                DriverType::Sqlite => {
                    if opts.on_conflict_ignore {
                        ("INSERT OR IGNORE INTO".to_string(), String::new())
                    } else if !update_cols.is_empty() {
                        let set_parts: Vec<String> = update_cols
                            .iter()
                            .map(|c| format!("{} = excluded.{}", c, c))
                            .collect();
                        (
                            "INSERT INTO".to_string(),
                            format!(
                                " ON CONFLICT ({}) DO UPDATE SET {}",
                                opts.on_conflict_columns.join(", "),
                                set_parts.join(", ")
                            ),
                        )
                    } else {
                        ("INSERT INTO".to_string(), String::new())
                    }
                }
            }
        }
    } else {
        ("INSERT INTO".to_string(), String::new())
    };

    let mut sql = format!(
        "{} {} ({}) VALUES ({}){}",
        insert_prefix, write_table, cols_str, vals_str, conflict_suffix,
    );

    // RETURNING
    if options.is_some_and(|o| o.returning) {
        // will be post-processed by driver
        sql.push_str(" --RETURNING--");
    }

    (sql, params)
}

/// Build an UPDATE SQL statement.
pub fn build_update_sql(
    table: &str,
    conditions: &[Condition],
    column_values: &[ColumnValue],
    options: Option<&UpdateOptions>,
) -> (String, Vec<Value>) {
    let active: Vec<&ColumnValue> = column_values.iter().filter(|cv| !cv.is_skip()).collect();

    let mut params = Vec::new();
    let set_clause: Vec<String> = active
        .iter()
        .map(|cv| {
            if let ColumnValueKind::Value(v) = &cv.value {
                match v {
                    Value::Raw(expr) => format!("{} = {}", cv.column_name, expr),
                    Value::Immediate(expr) => format!("{} = {}", cv.column_name, expr),
                    _ => {
                        params.push(v.clone());
                        format!("{} = ?", cv.column_name)
                    }
                }
            } else {
                unreachable!()
            }
        })
        .collect();

    let compiled = compile_conditions(conditions);
    params.extend(compiled.params);

    let write_table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(table);

    let mut sql = format!("UPDATE {} SET {}", write_table, set_clause.join(", "));

    if !compiled.sql.is_empty() {
        sql.push_str(&format!(" WHERE {}", compiled.sql));
    }

    if options.is_some_and(|o| o.returning) {
        sql.push_str(" --RETURNING--");
    }

    (sql, params)
}

/// Build a DELETE SQL statement.
pub fn build_delete_sql(
    table: &str,
    conditions: &[Condition],
    options: Option<&DeleteOptions>,
) -> (String, Vec<Value>) {
    let compiled = compile_conditions(conditions);

    let write_table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(table);

    let mut sql = format!("DELETE FROM {}", write_table);

    if !compiled.sql.is_empty() {
        sql.push_str(&format!(" WHERE {}", compiled.sql));
    }

    if options.is_some_and(|o| o.returning) {
        sql.push_str(" --RETURNING--");
    }

    (sql, compiled.params)
}

/// Build a COUNT SQL statement.
pub fn build_count_sql<M: Model>(conditions: &[Condition]) -> (String, Vec<Value>) {
    if conditions.is_empty() {
        if let Some(sql) = M::static_count_sql() {
            return (sql.to_string(), vec![]);
        }
    }

    let is_query_based = M::query_sql().is_some();
    let cte_alias = M::table_name();
    let table = if is_query_based {
        cte_alias
    } else {
        M::table_name()
    };

    let mut all_conditions = Vec::new();
    if let Some(filter) = M::default_filter() {
        all_conditions.extend(filter);
    }
    all_conditions.extend_from_slice(conditions);

    let compiled = compile_conditions(&all_conditions);

    let mut sql = String::new();
    if let Some(query_sql) = M::query_sql() {
        sql.push_str(&format!("WITH {} AS ({}) ", cte_alias, query_sql));
    }
    sql.push_str(&format!("SELECT COUNT(*) AS count FROM {}", table));
    if !compiled.sql.is_empty() {
        sql.push_str(&format!(" WHERE {}", compiled.sql));
    }

    (sql, compiled.params)
}

/// Return the appropriate SqlBuilder for the given driver type.
fn get_sql_builder(driver_type: DriverType) -> Box<dyn SqlBuilder> {
    match driver_type {
        DriverType::Postgres => Box::new(PostgresSqlBuilder),
        DriverType::Mysql => Box::new(MysqlSqlBuilder),
        DriverType::Sqlite => Box::new(SqliteSqlBuilder),
    }
}

/// Infer the PostgreSQL SQL type name for UNNEST casts based on a Value variant.
/// Must match what `values_to_array_for_postgres` produces as the array type.
pub fn value_to_pg_type(v: &Value) -> &'static str {
    match v {
        Value::Int(_) => "int8",
        Value::Float(_) => "float8",
        Value::Bool(_) => "bool",
        _ => "text",
    }
}

/// Convert a column of Values (one per row) to the appropriate array Value for Postgres.
pub fn values_to_array_for_postgres(values: &[Value]) -> Value {
    if values.is_empty() {
        return Value::IntArray(vec![]);
    }
    match &values[0] {
        Value::Int(_) => {
            let arr: Vec<i64> = values
                .iter()
                .filter_map(|v| {
                    if let Value::Int(n) = v {
                        Some(*n)
                    } else {
                        None
                    }
                })
                .collect();
            Value::IntArray(arr)
        }
        Value::Float(_) => {
            let arr: Vec<f64> = values
                .iter()
                .filter_map(|v| {
                    if let Value::Float(f) = v {
                        Some(*f)
                    } else {
                        None
                    }
                })
                .collect();
            Value::FloatArray(arr)
        }
        Value::Bool(_) => {
            let arr: Vec<bool> = values
                .iter()
                .filter_map(|v| {
                    if let Value::Bool(b) = v {
                        Some(*b)
                    } else {
                        None
                    }
                })
                .collect();
            Value::BoolArray(arr)
        }
        _ => {
            let arr: Vec<String> = values
                .iter()
                .map(|v| match v {
                    Value::Text(s) => s.clone(),
                    Value::Int(n) => n.to_string(),
                    Value::Float(f) => f.to_string(),
                    Value::Bool(b) => b.to_string(),
                    Value::Null => "".to_string(),
                    _ => format!("{:?}", v),
                })
                .collect();
            Value::TextArray(arr)
        }
    }
}

// ============================================
// CRUD Functions
// ============================================

/// Find records matching conditions, with optional hard limit enforcement.
pub async fn find<M: Model, E: Executor>(
    executor: &E,
    conditions: &[Condition],
    options: Option<&SelectOptions>,
    limit_config: Option<&LimitConfig>,
) -> Result<Vec<M>> {
    let hard_limit = limit_config.and_then(|lc| lc.find_hard_limit);
    let should_check = hard_limit.is_some() && options.is_none_or(|o| o.limit.is_none());

    let (sql, params) = if should_check {
        let mut opts = options.cloned().unwrap_or_default();
        opts.limit = Some(hard_limit.unwrap() + 1);
        build_select_sql::<M>(conditions, Some(&opts))
    } else {
        build_select_sql::<M>(conditions, options)
    };

    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::table_name());

    #[cfg(feature = "postgres")]
    {
        if let Some(converter) = M::pg_row_converter() {
            if let Some(pg_fut) = executor.execute_pg_rows(&sql, &params) {
                let pg_rows = pg_fut.await?;
                let mut rows: Vec<M> = pg_rows.iter().filter_map(converter).collect();

                if should_check {
                    let limit = hard_limit.unwrap();
                    if rows.len() > limit {
                        return Err(Error::LimitExceeded {
                            limit,
                            actual_count: rows.len(),
                            context: LimitContext::Find,
                            model: M::table_name().to_string(),
                        });
                    }
                }
                M::setup_batch_loading(&mut rows, executor, limit_config);
                return Ok(rows);
            }
        }
    }

    let result = executor
        .execute_query_with_op(&sql, &params, OperationType::Find, table)
        .await?;
    let mut rows: Vec<M> = result
        .rows
        .iter()
        .map(|row| M::from_row(row))
        .collect::<Result<Vec<_>>>()?;

    if should_check {
        let limit = hard_limit.unwrap();
        if rows.len() > limit {
            return Err(Error::LimitExceeded {
                limit,
                actual_count: rows.len(),
                context: LimitContext::Find,
                model: M::table_name().to_string(),
            });
        }
    }

    M::setup_batch_loading(&mut rows, executor, limit_config);
    Ok(rows)
}

/// Find records by primary key values.
/// `pk_values`: each inner Vec is one row's PK values (in order of primary_key_columns).
pub async fn find_by_id<M: Model, E: Executor>(
    executor: &E,
    pk_values: &[Vec<Value>],
    options: Option<&SelectOptions>,
) -> Result<Vec<M>> {
    if pk_values.is_empty() {
        return Ok(vec![]);
    }

    let pkey_columns = M::primary_key_columns();
    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::table_name());
    let select_column = options
        .and_then(|o| o.select.as_deref())
        .unwrap_or(M::select_columns());
    let pkey_count = pk_values.len();

    let builder = get_sql_builder(executor.driver_type());
    let pkey_cols: Vec<&str> = pkey_columns.to_vec();
    let column_types_vec: Vec<&str> = if executor.driver_type() == DriverType::Postgres {
        pk_values[0].iter().map(|v| value_to_pg_type(v)).collect()
    } else {
        vec![]
    };
    let column_types: Option<Vec<&str>> = if column_types_vec.is_empty() {
        None
    } else {
        Some(column_types_vec)
    };
    let (sql, param_count) = builder.build_find_by_pkeys(
        table,
        &pkey_cols,
        pkey_count,
        select_column,
        column_types.as_deref(),
    );

    let params =
        build_find_by_pkeys_params(executor.driver_type(), pk_values, pkey_columns, param_count);

    let mut final_sql = sql;
    if let Some(opts) = options {
        if let Some(ref raw) = opts.order_raw {
            final_sql.push_str(&format!(" ORDER BY {}", raw));
        } else if let Some(ref order) = opts.order {
            final_sql.push_str(&format!(" ORDER BY {}", order_to_sql(order)));
        }
    }

    #[cfg(feature = "postgres")]
    {
        if let Some(converter) = M::pg_row_converter() {
            if let Some(pg_fut) = executor.execute_pg_rows(&final_sql, &params) {
                let pg_rows = pg_fut.await?;
                let mut rows: Vec<M> = pg_rows.iter().filter_map(converter).collect();
                M::setup_batch_loading(&mut rows, executor, None);
                return Ok(rows);
            }
        }
    }

    let result = executor
        .execute_query_with_op(&final_sql, &params, OperationType::Find, table)
        .await?;
    let mut rows: Vec<M> = result.rows.iter().map(|row| M::from_row(row)).collect::<Result<Vec<_>>>()?;
    M::setup_batch_loading(&mut rows, executor, None);
    Ok(rows)
}

/// Build params for find_by_pkeys based on driver type.
fn build_find_by_pkeys_params(
    driver_type: DriverType,
    pk_values: &[Vec<Value>],
    pkey_columns: &[&str],
    _param_count: usize,
) -> Vec<Value> {
    match driver_type {
        DriverType::Postgres => {
            if pkey_columns.len() == 1 {
                let col_values: Vec<Value> = pk_values
                    .iter()
                    .filter_map(|row| row.first().cloned())
                    .collect();
                vec![values_to_array_for_postgres(&col_values)]
            } else {
                (0..pkey_columns.len())
                    .map(|col_idx| {
                        let col_values: Vec<Value> = pk_values
                            .iter()
                            .filter_map(|row| row.get(col_idx).cloned())
                            .collect();
                        values_to_array_for_postgres(&col_values)
                    })
                    .collect()
            }
        }
        DriverType::Mysql | DriverType::Sqlite => {
            let mut params = Vec::new();
            for row in pk_values {
                for v in row {
                    params.push(v.clone());
                }
            }
            params
        }
    }
}

/// Find a single record matching conditions.
pub async fn find_one<M: Model, E: Executor>(
    executor: &E,
    conditions: &[Condition],
    options: Option<&SelectOptions>,
) -> Result<Option<M>> {
    let mut opts = options.cloned().unwrap_or_default();
    opts.limit = Some(1);
    let (sql, params) = build_select_sql::<M>(conditions, Some(&opts));
    let table = opts.table_name.as_deref().unwrap_or(M::table_name());

    #[cfg(feature = "postgres")]
    {
        if let Some(converter) = M::pg_row_converter() {
            if let Some(pg_fut) = executor.execute_pg_rows(&sql, &params) {
                let pg_rows = pg_fut.await?;
                let mut record = pg_rows.first().and_then(converter);
                if let Some(ref mut r) = record {
                    M::setup_batch_loading(std::slice::from_mut(r), executor, None);
                }
                return Ok(record);
            }
        }
    }

    let result = executor
        .execute_query_with_op(&sql, &params, OperationType::Find, table)
        .await?;
    if result.rows.is_empty() {
        Ok(None)
    } else {
        let mut record = M::from_row(&result.rows[0])?;
        M::setup_batch_loading(std::slice::from_mut(&mut record), executor, None);
        Ok(Some(record))
    }
}

/// Count records matching conditions.
pub async fn count<M: Model, E: Executor>(executor: &E, conditions: &[Condition]) -> Result<i64> {
    let (sql, params) = build_count_sql::<M>(conditions);
    let table = M::table_name();
    let result = executor
        .execute_query_with_op(&sql, &params, OperationType::Find, table)
        .await?;
    if let Some(row) = result.rows.first() {
        if let Some(Value::Int(n)) = row.get("count") {
            return Ok(*n);
        }
    }
    Ok(0)
}

/// Insert a record. Requires a transaction context.
pub async fn create<M: Model, E: Executor>(
    executor: &E,
    values: &[ColumnValue],
    options: Option<&InsertOptions>,
) -> Result<QueryResult> {
    if executor.is_read_only() {
        return Err(Error::WriteInReadOnlyContext {
            operation: "INSERT".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if !executor.is_transaction() {
        return Err(Error::WriteOutsideTransaction {
            operation: "INSERT".to_string(),
            model: M::table_name().to_string(),
        });
    }
    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::update_table_name());
    let (sql, params) = build_insert_sql(table, values, options, executor.driver_type());
    executor
        .execute_write_with_op(&sql, &params, OperationType::Create, table)
        .await
}

/// Insert multiple records. Requires a transaction context.
/// `rows`: each inner Vec is one row's column values (ColumnValue).
/// Returns PkeyResult if options.returning is true, otherwise None.
///
/// # SKIP (ColumnValueKind::Skip) limitation
///
/// All rows must have the **same set of non-SKIP columns**. Per-row SKIP (where some rows
/// have a value for a column and others have SKIP) is not supported. With PostgreSQL's
/// UNNEST-based bulk insert, each column becomes an array; SKIP would require DEFAULT
/// per cell, which UNNEST cannot express. Use consistent columns across all rows, or
/// split into separate `create_many` calls by column pattern.
pub async fn create_many<M: Model, E: Executor>(
    executor: &E,
    rows: &[Vec<ColumnValue>],
    options: Option<&InsertOptions>,
) -> Result<Option<PkeyResult>> {
    if executor.is_read_only() {
        return Err(Error::WriteInReadOnlyContext {
            operation: "INSERT".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if !executor.is_transaction() {
        return Err(Error::WriteOutsideTransaction {
            operation: "INSERT".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if rows.is_empty() {
        let pkey_columns = M::primary_key_columns();
        return Ok(options.and_then(|o| {
            if o.returning {
                Some(PkeyResult {
                    key_columns: pkey_columns.iter().map(|s| (*s).to_string()).collect(),
                    values: vec![],
                })
            } else {
                None
            }
        }));
    }

    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::update_table_name());
    let row_count = rows.len();

    // Extract columns from first row; validate all rows have same non-SKIP column set
    let first_row: Vec<&ColumnValue> = rows[0].iter().filter(|cv| !cv.is_skip()).collect();
    let columns: Vec<&str> = first_row.iter().map(|cv| cv.column_name).collect();
    let first_set: std::collections::HashSet<&str> = columns.iter().copied().collect();
    for (i, row) in rows.iter().enumerate().skip(1) {
        let row_set: std::collections::HashSet<&str> = row
            .iter()
            .filter(|cv| !cv.is_skip())
            .map(|cv| cv.column_name)
            .collect();
        if row_set != first_set {
            return Err(Error::CreateManyInconsistentColumns {
                row_index: i,
                expected: columns.iter().map(|s| (*s).to_string()).collect(),
                actual: row_set.into_iter().map(String::from).collect(),
            });
        }
    }

    let builder = get_sql_builder(executor.driver_type());
    let returning_cols = if options.is_some_and(|o| o.returning) {
        Some(M::primary_key_columns())
    } else {
        None
    };

    let (conflict_cols, update_cols): (Vec<&str>, Vec<&str>) = if let Some(o) = options {
        if o.on_conflict_columns.is_empty() {
            (vec![], vec![])
        } else {
            let c: Vec<&str> = o.on_conflict_columns.iter().map(|s| s.as_str()).collect();
            let u: Vec<&str> = match &o.on_conflict_update {
                Some(ConflictUpdate::All) => columns.to_vec(),
                Some(ConflictUpdate::Columns(cols)) => cols.iter().map(|s| s.as_str()).collect(),
                None => vec![],
            };
            (c, u)
        }
    } else {
        (vec![], vec![])
    };

    let on_conflict = if conflict_cols.is_empty() {
        None
    } else {
        let action = if let Some(o) = options {
            if o.on_conflict_ignore || update_cols.is_empty() {
                OnConflictAction::DoNothing
            } else {
                OnConflictAction::DoUpdate(update_cols.as_slice())
            }
        } else {
            OnConflictAction::DoNothing
        };
        Some(OnConflict {
            columns: conflict_cols.as_slice(),
            action,
        })
    };

    let column_types_vec: Vec<&str> =
        if executor.driver_type() == DriverType::Postgres && row_count > 1 {
            columns
                .iter()
                .map(|col_name| {
                    first_row
                        .iter()
                        .find(|cv| cv.column_name == *col_name)
                        .map(|cv| {
                            if let ColumnValueKind::Value(v) = &cv.value {
                                value_to_pg_type(v)
                            } else {
                                "text"
                            }
                        })
                        .unwrap_or("text")
                })
                .collect()
        } else {
            vec![]
        };
    let column_types: Option<Vec<&str>> = if column_types_vec.is_empty() {
        None
    } else {
        Some(column_types_vec)
    };

    let (sql, param_count) = builder.build_insert(
        table,
        &columns,
        row_count,
        on_conflict.as_ref(),
        returning_cols,
        column_types.as_deref(),
    );

    let params = build_insert_many_params(executor.driver_type(), rows, &columns, param_count);

    let result = executor
        .execute_write_with_op(&sql, &params, OperationType::Create, table)
        .await?;

    if options.is_some_and(|o| o.returning) {
        let pkey_columns = M::primary_key_columns();
        let pkey_names: Vec<String> = pkey_columns.iter().map(|s| (*s).to_string()).collect();
        let values: Vec<Vec<Value>> = result
            .rows
            .iter()
            .map(|row| {
                pkey_names
                    .iter()
                    .filter_map(|name| row.get(name).cloned())
                    .collect()
            })
            .collect();
        Ok(Some(PkeyResult {
            key_columns: pkey_names,
            values,
        }))
    } else {
        Ok(None)
    }
}

/// Build params for create_many based on driver type.
fn build_insert_many_params(
    driver_type: DriverType,
    rows: &[Vec<ColumnValue>],
    columns: &[&str],
    param_count: usize,
) -> Vec<Value> {
    match driver_type {
        DriverType::Postgres if rows.len() > 1 => {
            // UNNEST: one array per column
            (0..columns.len())
                .map(|col_idx| {
                    let col_values: Vec<Value> = rows
                        .iter()
                        .filter_map(|row| {
                            row.iter()
                                .find(|cv| cv.column_name == columns[col_idx])
                                .and_then(|cv| {
                                    if let ColumnValueKind::Value(v) = &cv.value {
                                        Some(v.clone())
                                    } else {
                                        None
                                    }
                                })
                        })
                        .collect();
                    values_to_array_for_postgres(&col_values)
                })
                .collect()
        }
        _ => {
            // Flat params: MySQL, SQLite, or Postgres single row
            let mut params = Vec::with_capacity(param_count);
            for row in rows {
                for col_name in columns.iter() {
                    if let Some(cv) = row.iter().find(|cv| cv.column_name == *col_name) {
                        if !cv.is_skip() {
                            if let ColumnValueKind::Value(v) = &cv.value {
                                if !matches!(v, Value::Raw(_) | Value::Immediate(_)) {
                                    params.push(v.clone());
                                }
                            }
                        }
                    }
                }
            }
            params
        }
    }
}

/// Update multiple records with different values per row.
/// `rows`: each inner Vec is one row's column values (must include key columns).
/// `options.key_columns`: columns that identify each row (e.g. ["id"]).
pub async fn update_many<M: Model, E: Executor>(
    executor: &E,
    rows: &[Vec<ColumnValue>],
    options: &UpdateManyOptions,
) -> Result<Option<PkeyResult>> {
    if executor.is_read_only() {
        return Err(Error::WriteInReadOnlyContext {
            operation: "UPDATE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if !executor.is_transaction() {
        return Err(Error::WriteOutsideTransaction {
            operation: "UPDATE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if rows.is_empty() {
        let pkey_columns = M::primary_key_columns();
        return Ok(if options.returning {
            Some(PkeyResult {
                key_columns: pkey_columns.iter().map(|s| (*s).to_string()).collect(),
                values: vec![],
            })
        } else {
            None
        });
    }

    let key_columns: Vec<&str> = options.key_columns.iter().map(|s| s.as_str()).collect();
    let table = M::update_table_name();
    let row_count = rows.len();

    // Build value_columns: all non-key columns that have at least one non-skip value
    // Build skip_map: per-row set of columns to skip (keep existing value)
    let mut value_col_set = HashSet::new();
    let mut skip_map: HashMap<usize, HashSet<String>> = HashMap::new();
    for (row_idx, row) in rows.iter().enumerate() {
        for cv in row.iter() {
            if key_columns.contains(&cv.column_name) {
                continue;
            }
            if cv.is_skip() {
                skip_map
                    .entry(row_idx)
                    .or_default()
                    .insert(cv.column_name.to_string());
            } else {
                value_col_set.insert(cv.column_name.to_string());
            }
        }
    }
    let value_columns: Vec<&str> = value_col_set.iter().map(|s| s.as_str()).collect();

    if value_columns.is_empty() {
        let pkey_columns = M::primary_key_columns();
        return Ok(if options.returning {
            Some(PkeyResult {
                key_columns: pkey_columns.iter().map(|s| (*s).to_string()).collect(),
                values: vec![],
            })
        } else {
            None
        });
    }

    let builder = get_sql_builder(executor.driver_type());
    let returning_cols = if options.returning {
        Some(M::primary_key_columns())
    } else {
        None
    };

    let skip_map_ref = if skip_map.is_empty() {
        None
    } else {
        Some(&skip_map)
    };

    let all_columns: Vec<&str> = key_columns
        .iter()
        .chain(value_columns.iter())
        .copied()
        .collect();
    let column_types_vec: Vec<&str> = if executor.driver_type() == DriverType::Postgres {
        all_columns
            .iter()
            .map(|col_name| {
                rows[0]
                    .iter()
                    .find(|cv| cv.column_name == *col_name)
                    .map(|cv| {
                        if let ColumnValueKind::Value(v) = &cv.value {
                            value_to_pg_type(v)
                        } else {
                            "text"
                        }
                    })
                    .unwrap_or("text")
            })
            .collect()
    } else {
        vec![]
    };
    let column_types: Option<Vec<&str>> = if column_types_vec.is_empty() {
        None
    } else {
        Some(column_types_vec)
    };

    let (sql, _param_count) = builder.build_update_many(
        table,
        &key_columns,
        &value_columns,
        row_count,
        returning_cols,
        column_types.as_deref(),
        skip_map_ref,
    );

    let params = build_update_many_params(
        executor.driver_type(),
        rows,
        &key_columns,
        &value_columns,
        skip_map_ref,
    );

    let result = executor
        .execute_write_with_op(&sql, &params, OperationType::Update, table)
        .await?;

    if options.returning {
        let pkey_columns = M::primary_key_columns();
        let pkey_names: Vec<String> = pkey_columns.iter().map(|s| (*s).to_string()).collect();
        let values: Vec<Vec<Value>> = result
            .rows
            .iter()
            .map(|row| {
                pkey_names
                    .iter()
                    .filter_map(|name| row.get(name).cloned())
                    .collect()
            })
            .collect();
        Ok(Some(PkeyResult {
            key_columns: pkey_names,
            values,
        }))
    } else {
        Ok(None)
    }
}

/// Build params for update_many based on driver type.
fn build_update_many_params(
    driver_type: DriverType,
    rows: &[Vec<ColumnValue>],
    key_columns: &[&str],
    value_columns: &[&str],
    skip_map: Option<&HashMap<usize, HashSet<String>>>,
) -> Vec<Value> {
    match driver_type {
        DriverType::Postgres => {
            // UNNEST: one array per column (key_columns first, then value_columns)
            let all_columns: Vec<&str> = key_columns
                .iter()
                .chain(value_columns.iter())
                .copied()
                .collect();
            let mut params: Vec<Value> = all_columns
                .iter()
                .map(|col_name| {
                    let col_values: Vec<Value> = rows
                        .iter()
                        .enumerate()
                        .map(|(idx, row)| {
                            let is_skipped = skip_map
                                .and_then(|m| m.get(&idx))
                                .map(|s| s.contains(*col_name))
                                .unwrap_or(false);
                            if is_skipped {
                                Value::Null
                            } else if let Some(cv) =
                                row.iter().find(|cv| cv.column_name == *col_name)
                            {
                                if let ColumnValueKind::Value(v) = &cv.value {
                                    v.clone()
                                } else {
                                    Value::Null
                                }
                            } else {
                                Value::Null
                            }
                        })
                        .collect();
                    values_to_array_for_postgres(&col_values)
                })
                .collect();
            // Add skip flag arrays for Postgres
            if let Some(skip_map) = skip_map {
                let skip_columns: Vec<String> = {
                    let mut set = HashSet::new();
                    for s in skip_map.values() {
                        for col in s {
                            set.insert(col.clone());
                        }
                    }
                    let mut v: Vec<String> = set.into_iter().collect();
                    v.sort();
                    v
                };
                for col in skip_columns {
                    let flags: Vec<Value> = rows
                        .iter()
                        .enumerate()
                        .map(|(idx, _)| {
                            Value::Bool(
                                skip_map
                                    .get(&idx)
                                    .map(|s| s.contains(&col))
                                    .unwrap_or(false),
                            )
                        })
                        .collect();
                    params.push(values_to_array_for_postgres(&flags));
                }
            }
            params
        }
        DriverType::Mysql => {
            // Flat per row: key_columns, value_columns (null when skipped), then _skip_col for each skip column
            let skip_columns: Vec<String> = skip_map
                .map(|m| {
                    let mut set = HashSet::new();
                    for s in m.values() {
                        for col in s {
                            set.insert(col.clone());
                        }
                    }
                    let mut v: Vec<String> = set.into_iter().collect();
                    v.sort();
                    v
                })
                .unwrap_or_default();

            let all_columns: Vec<&str> = key_columns
                .iter()
                .chain(value_columns.iter())
                .copied()
                .collect();

            let mut params = Vec::new();
            for (row_idx, row) in rows.iter().enumerate() {
                for col_name in all_columns.iter() {
                    let is_skipped = skip_map
                        .and_then(|m| m.get(&row_idx))
                        .map(|s| s.contains(*col_name))
                        .unwrap_or(false);
                    if is_skipped {
                        params.push(Value::Null);
                    } else if let Some(cv) = row.iter().find(|cv| cv.column_name == *col_name) {
                        if let ColumnValueKind::Value(v) = &cv.value {
                            params.push(v.clone());
                        } else {
                            params.push(Value::Null);
                        }
                    } else {
                        params.push(Value::Null);
                    }
                }
                for col in &skip_columns {
                    let is_skipped = skip_map
                        .and_then(|m| m.get(&row_idx))
                        .map(|s| s.contains(col))
                        .unwrap_or(false);
                    params.push(Value::Bool(is_skipped));
                }
            }
            params
        }
        DriverType::Sqlite => {
            // Order: for each value_column, for each row: key params, then value param (if not skipped)
            let mut params = Vec::new();
            for col_name in value_columns.iter() {
                for (row_idx, row) in rows.iter().enumerate() {
                    for k in key_columns.iter() {
                        if let Some(cv) = row.iter().find(|cv| cv.column_name == *k) {
                            if let ColumnValueKind::Value(v) = &cv.value {
                                params.push(v.clone());
                            }
                        }
                    }
                    let is_skipped = skip_map
                        .and_then(|m| m.get(&row_idx))
                        .map(|s| s.contains(*col_name))
                        .unwrap_or(false);
                    if !is_skipped {
                        if let Some(cv) = row.iter().find(|cv| cv.column_name == *col_name) {
                            if let ColumnValueKind::Value(v) = &cv.value {
                                params.push(v.clone());
                            }
                        }
                    }
                }
            }
            // WHERE IN params
            for row in rows {
                for k in key_columns.iter() {
                    if let Some(cv) = row.iter().find(|cv| cv.column_name == *k) {
                        if let ColumnValueKind::Value(v) = &cv.value {
                            params.push(v.clone());
                        }
                    }
                }
            }
            params
        }
    }
}

/// Update records matching conditions. Requires a transaction context.
pub async fn update<M: Model, E: Executor>(
    executor: &E,
    conditions: &[Condition],
    values: &[ColumnValue],
    options: Option<&UpdateOptions>,
) -> Result<QueryResult> {
    if executor.is_read_only() {
        return Err(Error::WriteInReadOnlyContext {
            operation: "UPDATE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if !executor.is_transaction() {
        return Err(Error::WriteOutsideTransaction {
            operation: "UPDATE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    let mut all_conditions = Vec::new();
    if let Some(filter) = M::default_filter() {
        all_conditions.extend(filter);
    }
    all_conditions.extend_from_slice(conditions);
    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::update_table_name());
    let (sql, params) = build_update_sql(table, &all_conditions, values, options);
    executor
        .execute_write_with_op(&sql, &params, OperationType::Update, table)
        .await
}

/// Delete records matching conditions. Requires a transaction context.
pub async fn delete<M: Model, E: Executor>(
    executor: &E,
    conditions: &[Condition],
    options: Option<&DeleteOptions>,
) -> Result<QueryResult> {
    if executor.is_read_only() {
        return Err(Error::WriteInReadOnlyContext {
            operation: "DELETE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    if !executor.is_transaction() {
        return Err(Error::WriteOutsideTransaction {
            operation: "DELETE".to_string(),
            model: M::table_name().to_string(),
        });
    }
    let mut all_conditions = Vec::new();
    if let Some(filter) = M::default_filter() {
        all_conditions.extend(filter);
    }
    all_conditions.extend_from_slice(conditions);
    let table = options
        .and_then(|o| o.table_name.as_deref())
        .unwrap_or(M::update_table_name());
    let (sql, params) = build_delete_sql(table, &all_conditions, options);
    executor
        .execute_write_with_op(&sql, &params, OperationType::Delete, table)
        .await
}

/// Execute raw SQL and map results to model instances.
pub async fn query<M: Model, E: Executor>(
    executor: &E,
    sql: &str,
    params: &[Value],
) -> Result<Vec<M>> {
    #[cfg(feature = "postgres")]
    {
        if let Some(converter) = M::pg_row_converter() {
            if let Some(pg_fut) = executor.execute_pg_rows(sql, params) {
                let pg_rows = pg_fut.await?;
                return Ok(pg_rows.iter().filter_map(converter).collect());
            }
        }
    }

    let result = executor.execute_query(sql, params).await?;
    result.rows.iter().map(|row| M::from_row(row)).collect()
}

/// Execute raw SQL and return rows without mapping to a model.
pub async fn query_raw<E: Executor>(
    executor: &E,
    sql: &str,
    params: &[Value],
) -> Result<Vec<crate::driver::Row>> {
    let result = executor.execute_query(sql, params).await?;
    Ok(result.rows)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::column::{Column, ColumnValueKind};
    use crate::condition::Condition;
    use crate::values::Value;

    // Minimal test model (without derive macro, to test SQL building directly)
    struct TestUser;

    impl FromRow for TestUser {
        fn from_row(_row: &Row) -> Result<Self> {
            Ok(TestUser)
        }
    }

    impl Model for TestUser {
        type Columns = ();
        const TABLE_NAME: &'static str = "users";
        fn column_meta() -> &'static [ColumnMeta] {
            &[]
        }
        fn primary_key_columns() -> &'static [&'static str] {
            &["id"]
        }
        fn columns() -> Self::Columns {}
    }

    #[test]
    fn test_build_select_simple() {
        let conds = vec![Condition::Eq {
            column: "is_active".into(),
            table: None,
            value: Value::Bool(true),
        }];
        let (sql, params) = build_select_sql::<TestUser>(&conds, None);
        assert_eq!(sql, "SELECT * FROM users WHERE is_active = ?");
        assert_eq!(params, vec![Value::Bool(true)]);
    }

    #[test]
    fn test_build_select_with_options() {
        let conds = vec![];
        let opts = SelectOptions {
            limit: Some(10),
            offset: Some(20),
            order_raw: Some("name ASC".to_string()),
            for_update: Some(true),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert_eq!(
            sql,
            "SELECT * FROM users ORDER BY name ASC LIMIT 10 OFFSET 20 FOR UPDATE"
        );
    }

    #[test]
    fn test_build_select_with_group_and_having() {
        let conds = vec![];
        let opts = SelectOptions {
            group: Some("status".to_string()),
            having: Some("COUNT(*) > 1".to_string()),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.contains("GROUP BY status"));
        assert!(sql.contains("HAVING COUNT(*) > 1"));
    }

    #[test]
    fn test_build_select_with_having_only() {
        let conds = vec![];
        let opts = SelectOptions {
            having: Some("SUM(amount) > 100".to_string()),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.contains("HAVING SUM(amount) > 100"));
    }

    #[test]
    fn test_build_insert() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![
            col_name.set("John".to_string()),
            col_email.set("john@example.com".to_string()),
        ];
        let (sql, params) = build_insert_sql("users", &cvs, None, DriverType::Postgres);
        assert_eq!(sql, "INSERT INTO users (name, email) VALUES (?, ?)");
        assert_eq!(params.len(), 2);
    }

    #[test]
    fn test_build_insert_with_skip() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![col_name.set("John".to_string()), col_email.skip()];
        let (sql, params) = build_insert_sql("users", &cvs, None, DriverType::Postgres);
        assert_eq!(sql, "INSERT INTO users (name) VALUES (?)");
        assert_eq!(params.len(), 1);
    }

    #[test]
    fn test_build_insert_on_conflict() {
        let col_name: Column<String> = Column::new("users", "name");
        let cvs = vec![col_name.set("John".to_string())];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: true,
            ..Default::default()
        };
        let (sql, _) = build_insert_sql("users", &cvs, Some(&opts), DriverType::Postgres);
        assert!(sql.contains("ON CONFLICT (email) DO NOTHING"));
    }

    #[test]
    fn test_build_insert_on_conflict_mysql_ignore() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![
            col_name.set("John".to_string()),
            col_email.set("john@example.com".to_string()),
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: true,
            ..Default::default()
        };
        let (sql, _) = build_insert_sql("users", &cvs, Some(&opts), DriverType::Mysql);
        assert!(sql.contains("INSERT IGNORE INTO"));
        assert!(!sql.contains("ON CONFLICT"));
    }

    #[test]
    fn test_build_insert_on_conflict_mysql_update() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![
            col_name.set("John".to_string()),
            col_email.set("john@example.com".to_string()),
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: false,
            on_conflict_update: Some(ConflictUpdate::Columns(vec!["name".to_string()])),
            ..Default::default()
        };
        let (sql, _) = build_insert_sql("users", &cvs, Some(&opts), DriverType::Mysql);
        assert!(sql.contains("ON DUPLICATE KEY UPDATE"));
        assert!(sql.contains("name = VALUES(name)"));
    }

    #[test]
    fn test_build_insert_on_conflict_sqlite_ignore() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![
            col_name.set("John".to_string()),
            col_email.set("john@example.com".to_string()),
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: true,
            ..Default::default()
        };
        let (sql, _) = build_insert_sql("users", &cvs, Some(&opts), DriverType::Sqlite);
        assert!(sql.contains("INSERT OR IGNORE INTO"));
        assert!(!sql.contains("ON CONFLICT"));
    }

    #[test]
    fn test_build_insert_on_conflict_sqlite_update() {
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let cvs = vec![
            col_name.set("John".to_string()),
            col_email.set("john@example.com".to_string()),
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: false,
            on_conflict_update: Some(ConflictUpdate::Columns(vec!["name".to_string()])),
            ..Default::default()
        };
        let (sql, _) = build_insert_sql("users", &cvs, Some(&opts), DriverType::Sqlite);
        assert!(sql.contains("ON CONFLICT (email) DO UPDATE SET"));
        assert!(sql.contains("name = excluded.name"));
        assert!(!sql.contains("EXCLUDED"));
    }

    #[test]
    fn test_build_update() {
        let col_name: Column<String> = Column::new("users", "name");
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let cvs = vec![col_name.set("Jane".to_string())];
        let (sql, params) = build_update_sql("users", &conds, &cvs, None);
        assert_eq!(sql, "UPDATE users SET name = ? WHERE id = ?");
        assert_eq!(params.len(), 2);
    }

    #[test]
    fn test_build_delete() {
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let (sql, params) = build_delete_sql("users", &conds, None);
        assert_eq!(sql, "DELETE FROM users WHERE id = ?");
        assert_eq!(params, vec![Value::Int(1)]);
    }

    #[test]
    fn test_build_count() {
        let conds = vec![Condition::Eq {
            column: "is_active".into(),
            table: None,
            value: Value::Bool(true),
        }];
        let (sql, params) = build_count_sql::<TestUser>(&conds);
        assert_eq!(
            sql,
            "SELECT COUNT(*) AS count FROM users WHERE is_active = ?"
        );
        assert_eq!(params, vec![Value::Bool(true)]);
    }

    #[test]
    fn test_build_select_with_join() {
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let opts = SelectOptions {
            join: Some(vec![
                "JOIN profiles ON users.id = profiles.user_id".to_string()
            ]),
            ..Default::default()
        };
        let (sql, params) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.contains("SELECT * FROM users"));
        assert!(sql.contains("JOIN profiles ON users.id = profiles.user_id"));
        assert!(sql.contains("WHERE id = ?"));
        assert_eq!(params, vec![Value::Int(1)]);
    }

    #[test]
    fn test_build_select_with_left_join() {
        let conds = vec![];
        let opts = SelectOptions {
            left_join: Some(vec![
                "LEFT JOIN other ON users.id = other.user_id".to_string()
            ]),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.contains("SELECT * FROM users"));
        assert!(sql.contains("LEFT JOIN other ON users.id = other.user_id"));
    }

    #[test]
    fn test_build_select_query_based_model_wraps_in_cte() {
        struct QueryBasedReport;
        impl FromRow for QueryBasedReport {
            fn from_row(_row: &Row) -> Result<Self> {
                Ok(QueryBasedReport)
            }
        }
        impl Model for QueryBasedReport {
            type Columns = ();
            const TABLE_NAME: &'static str = "monthly_report";
            fn column_meta() -> &'static [ColumnMeta] {
                &[]
            }
            fn primary_key_columns() -> &'static [&'static str] {
                &["user_id"]
            }
            fn columns() -> Self::Columns {}
            fn query_sql() -> Option<&'static str> {
                Some("SELECT user_id, SUM(amount) as total FROM orders GROUP BY user_id")
            }
        }
        let conds: Vec<Condition> = vec![];
        let (sql, params) = build_select_sql::<QueryBasedReport>(&conds, None);
        assert!(sql.starts_with("WITH monthly_report AS (SELECT user_id, SUM(amount) as total FROM orders GROUP BY user_id) "));
        assert!(sql.contains("SELECT * FROM monthly_report"));
        assert!(params.is_empty());
    }

    #[test]
    fn test_build_select_with_cte() {
        let conds = vec![];
        let opts = SelectOptions {
            cte: Some(vec![crate::types::CteClause {
                name: "ranked".to_string(),
                sql: "SELECT *, ROW_NUMBER() OVER () AS rn FROM users".to_string(),
                params: vec![],
            }]),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(
            sql.starts_with("WITH ranked AS (SELECT *, ROW_NUMBER() OVER () AS rn FROM users) ")
        );
        assert!(sql.contains("SELECT * FROM users"));
    }

    #[test]
    fn test_build_select_with_multiple_ctes() {
        let conds = vec![];
        let opts = SelectOptions {
            cte: Some(vec![
                crate::types::CteClause {
                    name: "cte1".to_string(),
                    sql: "SELECT 1".to_string(),
                    params: vec![],
                },
                crate::types::CteClause {
                    name: "cte2".to_string(),
                    sql: "SELECT 2".to_string(),
                    params: vec![],
                },
            ]),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.contains("WITH cte1 AS (SELECT 1), cte2 AS (SELECT 2) "));
    }

    #[test]
    fn test_build_select_for_share() {
        let conds = vec![];
        let opts = SelectOptions {
            for_share: Some(true),
            ..Default::default()
        };
        let (sql, _) = build_select_sql::<TestUser>(&conds, Some(&opts));
        assert!(sql.ends_with("FOR SHARE"));
    }

    #[test]
    fn test_build_insert_with_raw_value() {
        let cvs = vec![
            ColumnValue {
                column_name: "name",
                table_name: "users",
                value: ColumnValueKind::Value(Value::Text("John".to_string())),
            },
            ColumnValue {
                column_name: "updated_at",
                table_name: "users",
                value: ColumnValueKind::Value(Value::Immediate("NOW()".to_string())),
            },
        ];
        let (sql, params) = build_insert_sql("users", &cvs, None, DriverType::Postgres);
        assert_eq!(
            sql,
            "INSERT INTO users (name, updated_at) VALUES (?, NOW())"
        );
        assert_eq!(params.len(), 1);
    }

    // ============================================
    // Mock Executor for CRUD function tests
    // ============================================

    use crate::driver::QueryResult;
    use crate::types::{DriverType, LimitConfig, UpdateManyOptions};
    use std::sync::{Arc, Mutex};

    struct MockExecutor {
        is_tx: bool,
        read_only: bool,
        last_sql: Arc<Mutex<String>>,
        result_rows: Vec<Row>,
    }

    impl MockExecutor {
        fn new_reader(result_rows: Vec<Row>) -> Self {
            Self {
                is_tx: false,
                read_only: false,
                last_sql: Arc::new(Mutex::new(String::new())),
                result_rows,
            }
        }

        fn new_transaction(result_rows: Vec<Row>) -> Self {
            Self {
                is_tx: true,
                read_only: false,
                last_sql: Arc::new(Mutex::new(String::new())),
                result_rows,
            }
        }

        fn new_read_only() -> Self {
            Self {
                is_tx: false,
                read_only: true,
                last_sql: Arc::new(Mutex::new(String::new())),
                result_rows: vec![],
            }
        }

        fn last_sql(&self) -> String {
            self.last_sql.lock().unwrap().clone()
        }
    }

    impl Executor for MockExecutor {
        fn execute_query<'a>(
            &'a self,
            sql: &'a str,
            _params: &'a [Value],
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>>
        {
            *self.last_sql.lock().unwrap() = sql.to_string();
            let rows = self.result_rows.clone();
            Box::pin(async move {
                Ok(QueryResult {
                    rows,
                    rows_affected: 0,
                })
            })
        }

        fn execute_write<'a>(
            &'a self,
            sql: &'a str,
            _params: &'a [Value],
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<QueryResult>> + Send + 'a>>
        {
            *self.last_sql.lock().unwrap() = sql.to_string();
            let rows = self.result_rows.clone();
            Box::pin(async move {
                Ok(QueryResult {
                    rows,
                    rows_affected: 1,
                })
            })
        }

        fn driver_type(&self) -> DriverType {
            DriverType::Postgres
        }
        fn is_transaction(&self) -> bool {
            self.is_tx
        }
        fn is_read_only(&self) -> bool {
            self.read_only
        }
    }

    // Test model with actual fields for FromRow testing
    #[derive(Debug, PartialEq)]
    struct UserRecord {
        name: String,
        age: i64,
    }

    impl FromRow for UserRecord {
        fn from_row(row: &Row) -> Result<Self> {
            let name = match row.get("name") {
                Some(Value::Text(s)) => s.clone(),
                _ => return Err(Error::ColumnNotFound("name".to_string())),
            };
            let age = match row.get("age") {
                Some(Value::Int(n)) => *n,
                _ => return Err(Error::ColumnNotFound("age".to_string())),
            };
            Ok(UserRecord { name, age })
        }
    }

    impl Model for UserRecord {
        type Columns = ();
        const TABLE_NAME: &'static str = "users";
        fn column_meta() -> &'static [ColumnMeta] {
            &[]
        }
        fn primary_key_columns() -> &'static [&'static str] {
            &["id"]
        }
        fn columns() -> Self::Columns {}
    }

    fn make_user_row(name: &str, age: i64) -> Row {
        let mut row = Row::new();
        row.insert("name".to_string(), Value::Text(name.to_string()));
        row.insert("age".to_string(), Value::Int(age));
        row
    }

    #[tokio::test]
    async fn test_find_from_row_error_propagation() {
        let mut row = Row::new();
        row.insert("wrong_col".to_string(), Value::Int(1));
        let executor = MockExecutor::new_reader(vec![row]);
        let result = find::<UserRecord, _>(&executor, &[], None, None).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::ColumnNotFound(name) => assert_eq!(name, "name"),
            other => panic!("Expected ColumnNotFound, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_count_with_empty_results() {
        let executor = MockExecutor::new_reader(vec![]);
        let n = count::<TestUser, _>(&executor, &[]).await.unwrap();
        assert_eq!(n, 0);
    }

    #[tokio::test]
    async fn test_count_with_malformed_results() {
        let mut row = Row::new();
        row.insert("other_key".to_string(), Value::Int(99));
        let executor = MockExecutor::new_reader(vec![row]);
        let n = count::<TestUser, _>(&executor, &[]).await.unwrap();
        assert_eq!(n, 0);
    }

    #[tokio::test]
    async fn test_find_returns_rows() {
        let rows = vec![make_user_row("Alice", 30), make_user_row("Bob", 25)];
        let executor = MockExecutor::new_reader(rows);
        let conds = vec![Condition::Eq {
            column: "is_active".into(),
            table: None,
            value: Value::Bool(true),
        }];
        let result: Vec<UserRecord> = find(&executor, &conds, None, None).await.unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result[0].name, "Alice");
        assert_eq!(result[1].age, 25);
        assert!(executor.last_sql().contains("SELECT * FROM users"));
    }

    #[tokio::test]
    async fn test_find_one_returns_first() {
        let rows = vec![make_user_row("Alice", 30)];
        let executor = MockExecutor::new_reader(rows);
        let result: Option<UserRecord> = find_one(&executor, &[], None).await.unwrap();
        assert!(result.is_some());
        assert_eq!(result.unwrap().name, "Alice");
        assert!(executor.last_sql().contains("LIMIT 1"));
    }

    #[tokio::test]
    async fn test_find_one_returns_none() {
        let executor = MockExecutor::new_reader(vec![]);
        let result: Option<UserRecord> = find_one(&executor, &[], None).await.unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_count_returns_value() {
        let mut row = Row::new();
        row.insert("count".to_string(), Value::Int(42));
        let executor = MockExecutor::new_reader(vec![row]);
        let n = count::<TestUser, _>(&executor, &[]).await.unwrap();
        assert_eq!(n, 42);
    }

    #[tokio::test]
    async fn test_find_hard_limit_enforced() {
        let rows: Vec<Row> = (0..6)
            .map(|i| make_user_row(&format!("User{}", i), i))
            .collect();
        let executor = MockExecutor::new_reader(rows);
        let limit_config = LimitConfig {
            find_hard_limit: Some(5),
            has_many_hard_limit: None,
        };
        let result = find::<UserRecord, _>(&executor, &[], None, Some(&limit_config)).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::LimitExceeded { limit, context, .. } => {
                assert_eq!(limit, 5);
                assert_eq!(context, LimitContext::Find);
            }
            other => panic!("Expected LimitExceeded, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_find_hard_limit_not_triggered_with_explicit_limit() {
        let rows: Vec<Row> = (0..3)
            .map(|i| make_user_row(&format!("User{}", i), i))
            .collect();
        let executor = MockExecutor::new_reader(rows);
        let limit_config = LimitConfig {
            find_hard_limit: Some(5),
            has_many_hard_limit: None,
        };
        let opts = SelectOptions {
            limit: Some(3),
            ..Default::default()
        };
        let result = find::<UserRecord, _>(&executor, &[], Some(&opts), Some(&limit_config)).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap().len(), 3);
    }

    #[tokio::test]
    async fn test_create_requires_transaction() {
        let executor = MockExecutor::new_reader(vec![]);
        let col: Column<String> = Column::new("users", "name");
        let cvs = vec![col.set("John".to_string())];
        let result = create::<TestUser, _>(&executor, &cvs, None).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::WriteOutsideTransaction { operation, model } => {
                assert_eq!(operation, "INSERT");
                assert_eq!(model, "users");
            }
            other => panic!("Expected WriteOutsideTransaction, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_create_in_transaction_succeeds() {
        let executor = MockExecutor::new_transaction(vec![]);
        let col: Column<String> = Column::new("users", "name");
        let cvs = vec![col.set("John".to_string())];
        let result = create::<TestUser, _>(&executor, &cvs, None).await;
        assert!(result.is_ok());
        assert!(executor.last_sql().contains("INSERT INTO users"));
    }

    #[tokio::test]
    async fn test_update_requires_transaction() {
        let executor = MockExecutor::new_reader(vec![]);
        let col: Column<String> = Column::new("users", "name");
        let cvs = vec![col.set("Jane".to_string())];
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let result = update::<TestUser, _>(&executor, &conds, &cvs, None).await;
        match result.unwrap_err() {
            Error::WriteOutsideTransaction { operation, .. } => assert_eq!(operation, "UPDATE"),
            other => panic!("Expected WriteOutsideTransaction, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_delete_requires_transaction() {
        let executor = MockExecutor::new_reader(vec![]);
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let result = delete::<TestUser, _>(&executor, &conds, None).await;
        match result.unwrap_err() {
            Error::WriteOutsideTransaction { operation, .. } => assert_eq!(operation, "DELETE"),
            other => panic!("Expected WriteOutsideTransaction, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_write_blocked_in_read_only_context() {
        let executor = MockExecutor::new_read_only();
        let col: Column<String> = Column::new("users", "name");
        let cvs = vec![col.set("John".to_string())];
        let result = create::<TestUser, _>(&executor, &cvs, None).await;
        match result.unwrap_err() {
            Error::WriteInReadOnlyContext { operation, model } => {
                assert_eq!(operation, "INSERT");
                assert_eq!(model, "users");
            }
            other => panic!("Expected WriteInReadOnlyContext, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_update_blocked_in_read_only_context() {
        let executor = MockExecutor::new_read_only();
        let col: Column<String> = Column::new("users", "name");
        let cvs = vec![col.set("Jane".to_string())];
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let result = update::<TestUser, _>(&executor, &conds, &cvs, None).await;
        match result.unwrap_err() {
            Error::WriteInReadOnlyContext { operation, model } => {
                assert_eq!(operation, "UPDATE");
                assert_eq!(model, "users");
            }
            other => panic!("Expected WriteInReadOnlyContext, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_delete_blocked_in_read_only_context() {
        let executor = MockExecutor::new_read_only();
        let conds = vec![Condition::Eq {
            column: "id".into(),
            table: None,
            value: Value::Int(1),
        }];
        let result = delete::<TestUser, _>(&executor, &conds, None).await;
        match result.unwrap_err() {
            Error::WriteInReadOnlyContext { operation, model } => {
                assert_eq!(operation, "DELETE");
                assert_eq!(model, "users");
            }
            other => panic!("Expected WriteInReadOnlyContext, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_find_on_query_based_model() {
        struct QueryBasedReport {
            name: String,
            age: i64,
        }
        impl FromRow for QueryBasedReport {
            fn from_row(row: &Row) -> Result<Self> {
                let name = match row.get("name") {
                    Some(Value::Text(s)) => s.clone(),
                    _ => return Err(Error::ColumnNotFound("name".to_string())),
                };
                let age = match row.get("age") {
                    Some(Value::Int(n)) => *n,
                    _ => return Err(Error::ColumnNotFound("age".to_string())),
                };
                Ok(QueryBasedReport { name, age })
            }
        }
        impl Model for QueryBasedReport {
            type Columns = ();
            const TABLE_NAME: &'static str = "report";
            fn column_meta() -> &'static [ColumnMeta] {
                &[]
            }
            fn primary_key_columns() -> &'static [&'static str] {
                &["user_id"]
            }
            fn columns() -> Self::Columns {}
            fn query_sql() -> Option<&'static str> {
                Some("SELECT user_id, name, age FROM users GROUP BY user_id")
            }
        }
        let rows = vec![make_user_row("Alice", 30)];
        let executor = MockExecutor::new_reader(rows);
        let conds: Vec<Condition> = vec![];
        let result: Vec<QueryBasedReport> = find(&executor, &conds, None, None).await.unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].name, "Alice");
        assert_eq!(result[0].age, 30);
        let sql = executor.last_sql();
        assert!(
            sql.contains("WITH report AS (SELECT user_id, name, age FROM users GROUP BY user_id)")
        );
        assert!(sql.contains("SELECT * FROM report"));
    }

    #[tokio::test]
    async fn test_query_raw_sql() {
        let rows = vec![make_user_row("Alice", 30)];
        let executor = MockExecutor::new_reader(rows);
        let result: Vec<UserRecord> = query(
            &executor,
            "SELECT * FROM users WHERE id = ?",
            &[Value::Int(1)],
        )
        .await
        .unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].name, "Alice");
    }

    #[tokio::test]
    async fn test_query_raw_returns_rows() {
        let rows = vec![make_user_row("Alice", 30), make_user_row("Bob", 25)];
        let executor = MockExecutor::new_reader(rows.clone());
        let result = query_raw(&executor, "SELECT * FROM users", &[])
            .await
            .unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result[0].get("name"), rows[0].get("name"));
        assert_eq!(result[1].get("age"), rows[1].get("age"));
    }

    // ============================================
    // find_by_id tests
    // ============================================

    #[tokio::test]
    async fn test_find_by_id_empty_returns_empty() {
        let executor = MockExecutor::new_reader(vec![]);
        let result: Vec<UserRecord> = find_by_id::<UserRecord, _>(&executor, &[], None)
            .await
            .unwrap();
        assert!(result.is_empty());
        assert!(executor.last_sql().is_empty());
    }

    #[tokio::test]
    async fn test_find_by_id_single_pk_returns_rows() {
        let rows = vec![make_user_row("Alice", 30), make_user_row("Bob", 25)];
        let executor = MockExecutor::new_reader(rows);
        let pk_values = vec![vec![Value::Int(1)], vec![Value::Int(2)]];
        let result: Vec<UserRecord> = find_by_id::<UserRecord, _>(&executor, &pk_values, None)
            .await
            .unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(result[0].name, "Alice");
        assert_eq!(result[1].name, "Bob");
        // Postgres uses = ANY($1::text[])
        assert!(executor.last_sql().contains("SELECT * FROM users"));
        assert!(executor.last_sql().contains("ANY") || executor.last_sql().contains("IN"));
    }

    #[tokio::test]
    async fn test_find_by_id_with_select_options() {
        let rows = vec![make_user_row("Alice", 30)];
        let executor = MockExecutor::new_reader(rows);
        let pk_values = vec![vec![Value::Int(1)]];
        let opts = SelectOptions {
            select: Some("id, name".to_string()),
            ..Default::default()
        };
        let result: Vec<UserRecord> =
            find_by_id::<UserRecord, _>(&executor, &pk_values, Some(&opts))
                .await
                .unwrap();
        assert_eq!(result.len(), 1);
        assert!(executor.last_sql().contains("id, name"));
    }

    // ============================================
    // create_many tests
    // ============================================

    #[tokio::test]
    async fn test_create_many_requires_transaction() {
        let executor = MockExecutor::new_reader(vec![]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![
                col_name.set("Jane".to_string()),
                col_email.set("jane@example.com".to_string()),
            ],
        ];
        let result = create_many::<TestUser, _>(&executor, &rows, None).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::WriteOutsideTransaction { operation, .. } => assert_eq!(operation, "INSERT"),
            other => panic!("Expected WriteOutsideTransaction, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_create_many_in_transaction_succeeds() {
        let mut id_row = Row::new();
        id_row.insert("id".to_string(), Value::Int(1));
        let executor = MockExecutor::new_transaction(vec![id_row]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![
                col_name.set("Jane".to_string()),
                col_email.set("jane@example.com".to_string()),
            ],
        ];
        let result = create_many::<TestUser, _>(&executor, &rows, None).await;
        assert!(result.is_ok());
        assert!(result.unwrap().is_none());
        assert!(executor.last_sql().contains("INSERT"));
        assert!(executor.last_sql().contains("UNNEST") || executor.last_sql().contains("VALUES"));
    }

    #[tokio::test]
    async fn test_create_many_empty_returns_none() {
        let executor = MockExecutor::new_transaction(vec![]);
        let result = create_many::<TestUser, _>(&executor, &[], None)
            .await
            .unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_create_many_with_returning_returns_pkey_result() {
        let mut row1 = Row::new();
        row1.insert("id".to_string(), Value::Int(1));
        let mut row2 = Row::new();
        row2.insert("id".to_string(), Value::Int(2));
        let executor = MockExecutor::new_transaction(vec![row1, row2]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![
                col_name.set("Jane".to_string()),
                col_email.set("jane@example.com".to_string()),
            ],
        ];
        let opts = InsertOptions {
            returning: true,
            ..Default::default()
        };
        let result = create_many::<TestUser, _>(&executor, &rows, Some(&opts))
            .await
            .unwrap();
        assert!(result.is_some());
        let pkey = result.unwrap();
        assert_eq!(pkey.key_columns, vec!["id"]);
        assert_eq!(pkey.values.len(), 2);
    }

    #[tokio::test]
    async fn test_create_many_with_on_conflict_ignore() {
        let executor = MockExecutor::new_transaction(vec![]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![
                col_name.set("Jane".to_string()),
                col_email.set("jane@example.com".to_string()),
            ],
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: true,
            ..Default::default()
        };
        let result = create_many::<TestUser, _>(&executor, &rows, Some(&opts))
            .await
            .unwrap();
        assert!(result.is_none());
        let sql = executor.last_sql();
        assert!(
            sql.contains("ON CONFLICT")
                || sql.contains("INSERT IGNORE")
                || sql.contains("OR IGNORE")
        );
    }

    #[tokio::test]
    async fn test_create_many_inconsistent_columns_returns_error() {
        let executor = MockExecutor::new_transaction(vec![]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![col_name.set("Jane".to_string()), col_email.skip()],
        ];
        let result = create_many::<TestUser, _>(&executor, &rows, None).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::CreateManyInconsistentColumns { row_index, .. } => assert_eq!(row_index, 1),
            other => panic!("Expected CreateManyInconsistentColumns, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_create_many_with_on_conflict_update() {
        let executor = MockExecutor::new_transaction(vec![]);
        let col_name: Column<String> = Column::new("users", "name");
        let col_email: Column<String> = Column::new("users", "email");
        let rows = vec![
            vec![
                col_name.set("John".to_string()),
                col_email.set("john@example.com".to_string()),
            ],
            vec![
                col_name.set("Jane".to_string()),
                col_email.set("jane@example.com".to_string()),
            ],
        ];
        let opts = InsertOptions {
            on_conflict_columns: vec!["email".to_string()],
            on_conflict_ignore: false,
            on_conflict_update: Some(ConflictUpdate::Columns(vec!["name".to_string()])),
            ..Default::default()
        };
        let result = create_many::<TestUser, _>(&executor, &rows, Some(&opts))
            .await
            .unwrap();
        assert!(result.is_none());
        let sql = executor.last_sql();
        assert!(sql.contains("ON CONFLICT") || sql.contains("ON DUPLICATE KEY UPDATE"));
        assert!(sql.contains("name"));
    }

    // ============================================
    // update_many tests
    // ============================================

    #[tokio::test]
    async fn test_update_many_requires_transaction() {
        let executor = MockExecutor::new_reader(vec![]);
        let col_id: Column<i64> = Column::new("users", "id");
        let col_name: Column<String> = Column::new("users", "name");
        let rows = vec![
            vec![col_id.set(1i64), col_name.set("John".to_string())],
            vec![col_id.set(2i64), col_name.set("Jane".to_string())],
        ];
        let opts = UpdateManyOptions {
            key_columns: vec!["id".to_string()],
            returning: false,
        };
        let result = update_many::<TestUser, _>(&executor, &rows, &opts).await;
        assert!(result.is_err());
        match result.unwrap_err() {
            Error::WriteOutsideTransaction { operation, .. } => assert_eq!(operation, "UPDATE"),
            other => panic!("Expected WriteOutsideTransaction, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_update_many_in_transaction_succeeds() {
        let executor = MockExecutor::new_transaction(vec![]);
        let col_id: Column<i64> = Column::new("users", "id");
        let col_name: Column<String> = Column::new("users", "name");
        let rows = vec![
            vec![col_id.set(1i64), col_name.set("John".to_string())],
            vec![col_id.set(2i64), col_name.set("Jane".to_string())],
        ];
        let opts = UpdateManyOptions {
            key_columns: vec!["id".to_string()],
            returning: false,
        };
        let result = update_many::<TestUser, _>(&executor, &rows, &opts).await;
        assert!(result.is_ok());
        assert!(result.unwrap().is_none());
        assert!(executor.last_sql().contains("UPDATE"));
        assert!(executor.last_sql().contains("users"));
    }

    #[tokio::test]
    async fn test_update_many_empty_returns_none() {
        let executor = MockExecutor::new_transaction(vec![]);
        let opts = UpdateManyOptions {
            key_columns: vec!["id".to_string()],
            returning: false,
        };
        let result = update_many::<TestUser, _>(&executor, &[], &opts)
            .await
            .unwrap();
        assert!(result.is_none());
    }

    #[tokio::test]
    async fn test_update_many_with_returning_returns_pkey_result() {
        let mut row1 = Row::new();
        row1.insert("id".to_string(), Value::Int(1));
        row1.insert("name".to_string(), Value::Text("John".to_string()));
        let mut row2 = Row::new();
        row2.insert("id".to_string(), Value::Int(2));
        row2.insert("name".to_string(), Value::Text("Jane".to_string()));
        let executor = MockExecutor::new_transaction(vec![row1, row2]);
        let col_id: Column<i64> = Column::new("users", "id");
        let col_name: Column<String> = Column::new("users", "name");
        let rows = vec![
            vec![col_id.set(1i64), col_name.set("John".to_string())],
            vec![col_id.set(2i64), col_name.set("Jane".to_string())],
        ];
        let opts = UpdateManyOptions {
            key_columns: vec!["id".to_string()],
            returning: true,
        };
        let result = update_many::<TestUser, _>(&executor, &rows, &opts)
            .await
            .unwrap();
        assert!(result.is_some());
        let pkey = result.unwrap();
        assert_eq!(pkey.key_columns, vec!["id"]);
        assert_eq!(pkey.values.len(), 2);
    }
}
