use crate::values::Value;

/// A query condition that compiles to a WHERE clause fragment.
#[derive(Debug, Clone, PartialEq)]
pub enum Condition {
    /// `column = ?`
    Eq {
        column: String,
        table: Option<String>,
        value: Value,
    },
    /// Raw SQL expression with bind parameters.
    /// Produced by the `sql!()` macro.
    Raw { sql: String, params: Vec<Value> },
    /// `column IN (subquery)` or `(col1, col2) IN (subquery)` for composite keys
    InSubquery {
        columns: Vec<String>,
        sql: String,
        params: Vec<Value>,
    },
    /// `column NOT IN (subquery)` or `(col1, col2) NOT IN (subquery)` for composite keys
    NotInSubquery {
        columns: Vec<String>,
        sql: String,
        params: Vec<Value>,
    },
    /// `EXISTS (subquery)`
    Exists { sql: String, params: Vec<Value> },
    /// `NOT EXISTS (subquery)`
    NotExists { sql: String, params: Vec<Value> },
    /// `cond1 AND cond2 AND ...`
    And(Vec<Condition>),
    /// `cond1 OR cond2 OR ...`
    Or(Vec<Condition>),
}

// ============================================
// WHERE clause compilation
// ============================================

/// Compiled WHERE clause with parameterized SQL.
#[derive(Debug, Clone)]
pub struct CompiledCondition {
    pub sql: String,
    pub params: Vec<Value>,
}

impl Condition {
    /// Compile this condition into a SQL WHERE fragment and parameter list.
    /// Uses `?` as the placeholder (driver layer converts to `$1` etc).
    pub fn compile(&self) -> CompiledCondition {
        let mut params = Vec::new();
        let sql = self.compile_inner(&mut params);
        CompiledCondition { sql, params }
    }

    fn compile_inner(&self, params: &mut Vec<Value>) -> String {
        match self {
            Condition::Eq {
                column,
                table: _,
                value,
            } => match value {
                Value::Null => format!("{} IS NULL", column),
                Value::NotNull => format!("{} IS NOT NULL", column),
                Value::IntArray(arr) => {
                    let placeholders: Vec<&str> = arr.iter().map(|_| "?").collect();
                    for v in arr {
                        params.push(Value::Int(*v));
                    }
                    format!("{} IN ({})", column, placeholders.join(", "))
                }
                Value::TextArray(arr) => {
                    let placeholders: Vec<&str> = arr.iter().map(|_| "?").collect();
                    for v in arr {
                        params.push(Value::Text(v.clone()));
                    }
                    format!("{} IN ({})", column, placeholders.join(", "))
                }
                Value::ValueArray(arr) => {
                    if arr.is_empty() {
                        return "1 = 0".to_string();
                    }
                    let placeholders: Vec<&str> = arr.iter().map(|_| "?").collect();
                    for v in arr {
                        params.push(v.clone());
                    }
                    format!("{} IN ({})", column, placeholders.join(", "))
                }
                Value::Dynamic {
                    sql: dyn_sql,
                    params: dyn_params,
                } => {
                    params.extend(dyn_params.iter().cloned());
                    format!("{} = {}", column, dyn_sql)
                }
                Value::Cast {
                    value: inner,
                    sql_type,
                } => {
                    params.push(*inner.clone());
                    format!("{} = ?::{}", column, sql_type)
                }
                Value::CastArray { values, sql_type } => {
                    if values.is_empty() {
                        return "1 = 0".to_string();
                    }
                    let placeholders: Vec<String> = values
                        .iter()
                        .map(|v| {
                            params.push(v.clone());
                            format!("?::{}", sql_type)
                        })
                        .collect();
                    format!("{} IN ({})", column, placeholders.join(", "))
                }
                _ => {
                    params.push(value.clone());
                    format!("{} = ?", column)
                }
            },
            Condition::Raw {
                sql,
                params: raw_params,
            } => {
                params.extend(raw_params.iter().cloned());
                sql.clone()
            }
            Condition::InSubquery {
                columns,
                sql: sub_sql,
                params: sub_params,
            } => {
                params.extend(sub_params.iter().cloned());
                let col_expr = if columns.len() == 1 {
                    columns[0].clone()
                } else {
                    format!("({})", columns.join(", "))
                };
                format!("{} IN ({})", col_expr, sub_sql)
            }
            Condition::NotInSubquery {
                columns,
                sql: sub_sql,
                params: sub_params,
            } => {
                params.extend(sub_params.iter().cloned());
                let col_expr = if columns.len() == 1 {
                    columns[0].clone()
                } else {
                    format!("({})", columns.join(", "))
                };
                format!("{} NOT IN ({})", col_expr, sub_sql)
            }
            Condition::Exists {
                sql: sub_sql,
                params: sub_params,
            } => {
                params.extend(sub_params.iter().cloned());
                format!("EXISTS ({})", sub_sql)
            }
            Condition::NotExists {
                sql: sub_sql,
                params: sub_params,
            } => {
                params.extend(sub_params.iter().cloned());
                format!("NOT EXISTS ({})", sub_sql)
            }
            Condition::And(conditions) => {
                if conditions.is_empty() {
                    return "1=1".to_string();
                }
                let parts: Vec<String> =
                    conditions.iter().map(|c| c.compile_inner(params)).collect();
                if parts.len() == 1 {
                    parts.into_iter().next().unwrap()
                } else {
                    parts.join(" AND ")
                }
            }
            Condition::Or(conditions) => {
                if conditions.is_empty() {
                    return "1=0".to_string();
                }
                let parts: Vec<String> = conditions
                    .iter()
                    .map(|c| {
                        let compiled = c.compile_inner(params);
                        if matches!(c, Condition::And(v) if v.len() > 1) {
                            format!("({})", compiled)
                        } else {
                            compiled
                        }
                    })
                    .collect();
                if parts.len() == 1 {
                    parts.into_iter().next().unwrap()
                } else {
                    format!("({})", parts.join(" OR "))
                }
            }
        }
    }
}

/// Compile a slice of conditions into a single AND-joined WHERE clause.
pub fn compile_conditions(conditions: &[Condition]) -> CompiledCondition {
    if conditions.is_empty() {
        return CompiledCondition {
            sql: String::new(),
            params: Vec::new(),
        };
    }
    if conditions.len() == 1 {
        return conditions[0].compile();
    }
    Condition::And(conditions.to_vec()).compile()
}

/// Create `column IN (subquery)` condition.
/// For composite keys, use `model_in_subquery` with multiple key pairs.
pub fn in_subquery(column: &str, sql: &str, params: Vec<Value>) -> Condition {
    Condition::InSubquery {
        columns: vec![column.to_string()],
        sql: sql.to_string(),
        params,
    }
}

/// Create `column NOT IN (subquery)` condition.
/// For composite keys, use `model_not_in_subquery` with multiple key pairs.
pub fn not_in_subquery(column: &str, sql: &str, params: Vec<Value>) -> Condition {
    Condition::NotInSubquery {
        columns: vec![column.to_string()],
        sql: sql.to_string(),
        params,
    }
}

/// Creates an IN subquery condition from model column mappings.
/// key_pairs: pairs of (outer_column_name, inner_column_name)
/// inner_table: the subquery table
/// inner_conditions: conditions applied inside the subquery
///
/// Single key: `outer_col IN (SELECT inner_col FROM inner_table WHERE ...)`
/// Multiple keys: `(outer_col1, outer_col2) IN (SELECT inner_col1, inner_col2 FROM inner_table WHERE ...)`
pub fn model_in_subquery(
    key_pairs: &[(&str, &str)],
    inner_table: &str,
    inner_conditions: &[Condition],
) -> Condition {
    if key_pairs.is_empty() {
        panic!("model_in_subquery requires at least one key pair");
    }
    let outer_columns: Vec<String> = key_pairs.iter().map(|(o, _)| (*o).to_string()).collect();
    let inner_columns: Vec<String> = key_pairs.iter().map(|(_, i)| (*i).to_string()).collect();
    let compiled = compile_conditions(inner_conditions);
    let where_clause = if compiled.sql.is_empty() {
        String::new()
    } else {
        format!(" WHERE {}", compiled.sql)
    };
    let select_clause = inner_columns.join(", ");
    let sub_sql = format!(
        "SELECT {} FROM {}{}",
        select_clause, inner_table, where_clause
    );
    Condition::InSubquery {
        columns: outer_columns,
        sql: sub_sql,
        params: compiled.params,
    }
}

/// Creates a NOT IN subquery condition from model column mappings.
/// Same as `model_in_subquery` but with NOT IN.
pub fn model_not_in_subquery(
    key_pairs: &[(&str, &str)],
    inner_table: &str,
    inner_conditions: &[Condition],
) -> Condition {
    if key_pairs.is_empty() {
        panic!("model_not_in_subquery requires at least one key pair");
    }
    let outer_columns: Vec<String> = key_pairs.iter().map(|(o, _)| (*o).to_string()).collect();
    let inner_columns: Vec<String> = key_pairs.iter().map(|(_, i)| (*i).to_string()).collect();
    let compiled = compile_conditions(inner_conditions);
    let where_clause = if compiled.sql.is_empty() {
        String::new()
    } else {
        format!(" WHERE {}", compiled.sql)
    };
    let select_clause = inner_columns.join(", ");
    let sub_sql = format!(
        "SELECT {} FROM {}{}",
        select_clause, inner_table, where_clause
    );
    Condition::NotInSubquery {
        columns: outer_columns,
        sql: sub_sql,
        params: compiled.params,
    }
}

/// Creates a correlated EXISTS subquery.
/// outer_table: table name of the outer query (for qualified reference)
/// outer_column: column from the outer query
/// inner_column: column from the inner table
/// inner_table: the subquery table
/// inner_conditions: extra conditions
///
/// Generates: `EXISTS (SELECT 1 FROM inner_table WHERE inner_column = outer_table.outer_column AND ...)`
pub fn correlated_exists(
    outer_table: &str,
    outer_column: &str,
    inner_column: &str,
    inner_table: &str,
    inner_conditions: &[Condition],
) -> Condition {
    let outer_ref = format!("{}.{}", outer_table, outer_column);
    let mut conds = vec![Condition::Raw {
        sql: format!("{} = {}", inner_column, outer_ref),
        params: vec![],
    }];
    conds.extend(inner_conditions.iter().cloned());
    let compiled = compile_conditions(&conds);
    let where_clause = if compiled.sql.is_empty() {
        String::new()
    } else {
        format!(" WHERE {}", compiled.sql)
    };
    let sub_sql = format!("SELECT 1 FROM {}{}", inner_table, where_clause);
    Condition::Exists {
        sql: sub_sql,
        params: compiled.params,
    }
}

/// Create `EXISTS (subquery)` condition.
pub fn exists(sql: &str, params: Vec<Value>) -> Condition {
    Condition::Exists {
        sql: sql.to_string(),
        params,
    }
}

/// Create `NOT EXISTS (subquery)` condition.
pub fn not_exists(sql: &str, params: Vec<Value>) -> Condition {
    Condition::NotExists {
        sql: sql.to_string(),
        params,
    }
}

/// Macro to create OR condition from multiple condition groups without `vec![]` wrapper.
///
/// ```ignore
/// or![conds![(c.role, "admin")], conds![(c.role, "moderator")]]
/// ```
#[macro_export]
macro_rules! or {
    [ $($group:expr),* $(,)? ] => {
        $crate::condition::or(vec![$($group),*])
    };
}

/// Helper: create OR condition from multiple condition groups.
pub fn or(groups: Vec<Vec<Condition>>) -> Condition {
    Condition::Or(
        groups
            .into_iter()
            .map(|g| {
                if g.len() == 1 {
                    g.into_iter().next().unwrap()
                } else {
                    Condition::And(g)
                }
            })
            .collect(),
    )
}

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

    #[test]
    fn test_eq_compile() {
        let cond = Condition::Eq {
            column: "name".to_string(),
            table: Some("users".to_string()),
            value: Value::Text("John".to_string()),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "name = ?");
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::Text("John".to_string()));
    }

    #[test]
    fn test_eq_null_compile() {
        let cond = Condition::Eq {
            column: "deleted_at".to_string(),
            table: None,
            value: Value::Null,
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "deleted_at IS NULL");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_eq_int_array_compile() {
        let cond = Condition::Eq {
            column: "id".to_string(),
            table: None,
            value: Value::IntArray(vec![1, 2, 3]),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "id IN (?, ?, ?)");
        assert_eq!(compiled.params.len(), 3);
    }

    #[test]
    fn test_raw_compile() {
        let cond = Condition::Raw {
            sql: "age > ? AND age < ?".to_string(),
            params: vec![Value::Int(18), Value::Int(65)],
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "age > ? AND age < ?");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_and_compile() {
        let cond = Condition::And(vec![
            Condition::Eq {
                column: "is_active".to_string(),
                table: None,
                value: Value::Bool(true),
            },
            Condition::Eq {
                column: "role".to_string(),
                table: None,
                value: Value::Text("admin".to_string()),
            },
        ]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "is_active = ? AND role = ?");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_or_compile() {
        let cond = or(vec![
            vec![Condition::Eq {
                column: "role".to_string(),
                table: None,
                value: Value::Text("admin".to_string()),
            }],
            vec![Condition::Eq {
                column: "role".to_string(),
                table: None,
                value: Value::Text("editor".to_string()),
            }],
        ]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "(role = ? OR role = ?)");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_or_macro() {
        let cond = or![
            vec![Condition::Eq {
                column: "role".to_string(),
                table: None,
                value: Value::Text("admin".to_string()),
            }],
            vec![Condition::Eq {
                column: "role".to_string(),
                table: None,
                value: Value::Text("moderator".to_string()),
            }],
        ];
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "(role = ? OR role = ?)");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_compile_conditions_empty() {
        let compiled = compile_conditions(&[]);
        assert_eq!(compiled.sql, "");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_compile_conditions_multiple() {
        let conditions = vec![
            Condition::Eq {
                column: "is_active".to_string(),
                table: None,
                value: Value::Bool(true),
            },
            Condition::Raw {
                sql: "age > ?".to_string(),
                params: vec![Value::Int(18)],
            },
        ];
        let compiled = compile_conditions(&conditions);
        assert_eq!(compiled.sql, "is_active = ? AND age > ?");
        assert_eq!(compiled.params.len(), 2);
    }

    // ========== Deeply nested And/Or ==========
    #[test]
    fn test_deeply_nested_and_or() {
        let cond = Condition::And(vec![
            Condition::Or(vec![
                Condition::Eq {
                    column: "a".to_string(),
                    table: None,
                    value: Value::Int(1),
                },
                Condition::Eq {
                    column: "b".to_string(),
                    table: None,
                    value: Value::Int(2),
                },
            ]),
            Condition::Eq {
                column: "c".to_string(),
                table: None,
                value: Value::Int(3),
            },
        ]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "(a = ? OR b = ?) AND c = ?");
        assert_eq!(compiled.params.len(), 3);
    }

    // ========== Empty conditions ==========
    #[test]
    fn test_and_empty() {
        let cond = Condition::And(vec![]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "1=1");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_or_empty() {
        let cond = Condition::Or(vec![]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "1=0");
        assert!(compiled.params.is_empty());
    }

    // ========== Single-element And/Or ==========
    #[test]
    fn test_and_single_element() {
        let cond = Condition::And(vec![Condition::Eq {
            column: "x".to_string(),
            table: None,
            value: Value::Int(1),
        }]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "x = ?");
        assert_eq!(compiled.params.len(), 1);
    }

    #[test]
    fn test_or_single_element() {
        let cond = Condition::Or(vec![Condition::Eq {
            column: "y".to_string(),
            table: None,
            value: Value::Text("z".to_string()),
        }]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "y = ?");
        assert_eq!(compiled.params.len(), 1);
    }

    // ========== Value variants in Eq ==========
    #[test]
    fn test_eq_value_not_null() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::NotNull,
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col IS NOT NULL");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_eq_value_raw() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::Raw("NOW()".to_string()),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col = ?");
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::Raw("NOW()".to_string()));
    }

    #[test]
    fn test_eq_value_immediate() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::Immediate("42".to_string()),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col = ?");
        assert_eq!(compiled.params.len(), 1);
    }

    #[test]
    fn test_eq_value_dynamic() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::Dynamic {
                sql: "LOWER(?)".to_string(),
                params: vec![Value::Text("X".to_string())],
            },
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col = LOWER(?)");
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::Text("X".to_string()));
    }

    #[test]
    fn test_eq_value_array_empty() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::ValueArray(vec![]),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "1 = 0");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_eq_value_array_mixed() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::ValueArray(vec![Value::Int(1), Value::Text("a".to_string())]),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col IN (?, ?)");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_eq_text_array() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::TextArray(vec!["a".to_string(), "b".to_string()]),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col IN (?, ?)");
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_eq_bool_array() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::BoolArray(vec![true, false]),
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col = ?");
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::BoolArray(vec![true, false]));
    }

    #[test]
    fn test_eq_cast() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::Cast {
                value: Box::new(Value::Text("uuid-here".to_string())),
                sql_type: "uuid".to_string(),
            },
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col = ?::uuid");
        assert_eq!(compiled.params.len(), 1);
    }

    #[test]
    fn test_eq_cast_array_empty() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::CastArray {
                values: vec![],
                sql_type: "uuid".to_string(),
            },
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "1 = 0");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_eq_cast_array() {
        let cond = Condition::Eq {
            column: "col".to_string(),
            table: None,
            value: Value::CastArray {
                values: vec![Value::Text("a".to_string()), Value::Text("b".to_string())],
                sql_type: "uuid".to_string(),
            },
        };
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "col IN (?::uuid, ?::uuid)");
        assert_eq!(compiled.params.len(), 2);
    }

    // ========== Subquery conditions ==========
    #[test]
    fn test_in_subquery_compile() {
        let cond = in_subquery(
            "user_id",
            "SELECT id FROM orders WHERE status = ?",
            vec![Value::Text("active".to_string())],
        );
        let compiled = cond.compile();
        assert_eq!(
            compiled.sql,
            "user_id IN (SELECT id FROM orders WHERE status = ?)"
        );
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::Text("active".to_string()));
    }

    #[test]
    fn test_not_in_subquery_compile() {
        let cond = not_in_subquery("id", "SELECT user_id FROM banned_users", vec![]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "id NOT IN (SELECT user_id FROM banned_users)");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_exists_compile() {
        let cond = exists(
            "SELECT 1 FROM orders WHERE orders.user_id = users.id",
            vec![],
        );
        let compiled = cond.compile();
        assert_eq!(
            compiled.sql,
            "EXISTS (SELECT 1 FROM orders WHERE orders.user_id = users.id)"
        );
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_not_exists_compile() {
        let cond = not_exists(
            "SELECT 1 FROM deleted_users WHERE deleted_users.id = users.id",
            vec![],
        );
        let compiled = cond.compile();
        assert_eq!(
            compiled.sql,
            "NOT EXISTS (SELECT 1 FROM deleted_users WHERE deleted_users.id = users.id)"
        );
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_in_subquery_with_params() {
        let cond = in_subquery(
            "id",
            "SELECT parent_id FROM nodes WHERE level = ? AND active = ?",
            vec![Value::Int(1), Value::Bool(true)],
        );
        let compiled = cond.compile();
        assert!(compiled.sql.contains("id IN (SELECT parent_id"));
        assert_eq!(compiled.params.len(), 2);
        assert_eq!(compiled.params[0], Value::Int(1));
        assert_eq!(compiled.params[1], Value::Bool(true));
    }

    #[test]
    fn test_subquery_in_and_condition() {
        let cond = Condition::And(vec![
            in_subquery("id", "SELECT user_id FROM orders", vec![]),
            Condition::Eq {
                column: "active".to_string(),
                table: None,
                value: Value::Bool(true),
            },
        ]);
        let compiled = cond.compile();
        assert!(compiled.sql.contains("id IN (SELECT user_id FROM orders)"));
        assert!(compiled.sql.contains("active = ?"));
        assert_eq!(compiled.params.len(), 1);
    }

    #[test]
    fn test_exists_with_params() {
        let cond = exists(
            "SELECT 1 FROM logs WHERE user_id = ? AND created_at > ?",
            vec![Value::Int(42), Value::Text("2024-01-01".to_string())],
        );
        let compiled = cond.compile();
        assert!(compiled.sql.starts_with("EXISTS ("));
        assert_eq!(compiled.params.len(), 2);
    }

    #[test]
    fn test_not_in_subquery_with_params() {
        let cond = not_in_subquery(
            "email",
            "SELECT email FROM blacklist WHERE reason = ?",
            vec![Value::Text("spam".to_string())],
        );
        let compiled = cond.compile();
        assert!(compiled.sql.contains("email NOT IN ("));
        assert_eq!(compiled.params.len(), 1);
    }

    // ========== model_in_subquery, model_not_in_subquery, correlated_exists ==========
    #[test]
    fn test_model_in_subquery_single_key() {
        let cond = model_in_subquery(
            &[("id", "user_id")],
            "orders",
            &[Condition::Eq {
                column: "amount".to_string(),
                table: None,
                value: Value::Int(100),
            }],
        );
        let compiled = cond.compile();
        assert_eq!(
            compiled.sql,
            "id IN (SELECT user_id FROM orders WHERE amount = ?)"
        );
        assert_eq!(compiled.params.len(), 1);
        assert_eq!(compiled.params[0], Value::Int(100));
    }

    #[test]
    fn test_model_in_subquery_composite_key() {
        let cond = model_in_subquery(
            &[("tenant_id", "tenant_id"), ("user_id", "user_id")],
            "orders",
            &[Condition::Eq {
                column: "status".to_string(),
                table: None,
                value: Value::Text("paid".to_string()),
            }],
        );
        let compiled = cond.compile();
        assert_eq!(
            compiled.sql,
            "(tenant_id, user_id) IN (SELECT tenant_id, user_id FROM orders WHERE status = ?)"
        );
        assert_eq!(compiled.params.len(), 1);
    }

    #[test]
    fn test_model_not_in_subquery() {
        let cond = model_not_in_subquery(&[("id", "user_id")], "banned_users", &[]);
        let compiled = cond.compile();
        assert_eq!(compiled.sql, "id NOT IN (SELECT user_id FROM banned_users)");
        assert!(compiled.params.is_empty());
    }

    #[test]
    fn test_correlated_exists() {
        let cond = correlated_exists(
            "users",
            "id",
            "user_id",
            "orders",
            &[Condition::Eq {
                column: "status".to_string(),
                table: None,
                value: Value::Text("paid".to_string()),
            }],
        );
        let compiled = cond.compile();
        assert!(compiled
            .sql
            .starts_with("EXISTS (SELECT 1 FROM orders WHERE "));
        assert!(compiled.sql.contains("user_id = users.id"));
        assert!(compiled.sql.contains("status = ?"));
        assert_eq!(compiled.params.len(), 1);
    }
}
