use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use crate::column::OrderSpec;
use crate::condition::Condition;
use crate::error::{Error, LimitContext, Result};
use crate::model::{value_to_pg_type, values_to_array_for_postgres, Executor, FromRow};
use crate::types::{DriverType, LimitConfig};
use crate::values::Value;

// ============================================
// RelationBatch — shared batch loading context
// ============================================

/// Shared batch context for lazy-loading relations.
/// Created by `find()` and shared (via `Arc`) among all records in the result set.
/// The first `load()` call triggers a single batch query; subsequent calls return cached data.
pub struct RelationBatch<T> {
    loader: Box<
        dyn Fn() -> Pin<Box<dyn Future<Output = Result<HashMap<String, T>>> + Send>>
            + Send
            + Sync,
    >,
    data: tokio::sync::OnceCell<HashMap<String, T>>,
}

impl<T: Send + Sync + 'static> RelationBatch<T> {
    pub fn new(
        loader: Box<
            dyn Fn() -> Pin<Box<dyn Future<Output = Result<HashMap<String, T>>> + Send>>
                + Send
                + Sync,
        >,
    ) -> Self {
        Self {
            loader,
            data: tokio::sync::OnceCell::new(),
        }
    }

    /// Get the batch-loaded data, initializing on first call.
    pub async fn get_or_load(&self) -> Result<&HashMap<String, T>> {
        self.data.get_or_try_init(|| (self.loader)()).await
    }
}

// ============================================
// Relation<T> — lazy-loaded relation field
// ============================================

/// A lazy-loaded relation field.
///
/// Records returned by `find()` automatically get batch contexts.
/// Access via `load().await?` triggers transparent batch loading on first access.
///
/// ```ignore
/// let users = db.find::<User>(&[], None).await?;
/// for user in &users {
///     let posts = user.posts.load().await?;    // batch-loads for ALL users on first call
///     let org = user.organization.load().await?; // same: one query for all users
/// }
/// ```
///
/// Sync access via `get()` / `get_many()` is available after `load()` or `set()`.
pub struct Relation<T> {
    loaded: Option<T>,
    batch: Option<Arc<RelationBatch<T>>>,
    my_key: Option<String>,
}

impl<T: std::fmt::Debug> std::fmt::Debug for Relation<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Relation")
            .field("loaded", &self.loaded)
            .field("has_batch", &self.batch.is_some())
            .finish()
    }
}

impl<T: Clone> Clone for Relation<T> {
    fn clone(&self) -> Self {
        Self {
            loaded: self.loaded.clone(),
            batch: self.batch.clone(),
            my_key: self.my_key.clone(),
        }
    }
}

impl<T> Default for Relation<T> {
    fn default() -> Self {
        Self {
            loaded: None,
            batch: None,
            my_key: None,
        }
    }
}

impl<T: Send + Sync + 'static> Relation<T> {
    /// Lazy-load the relation data. Triggers batch loading on first access.
    ///
    /// - If data was already loaded (via a prior `load()` or `set()`), returns it immediately.
    /// - If a batch context exists (from `find()`), runs a single batch query
    ///   for all sibling records, then returns this record's data.
    /// - Panics if no batch context is available
    ///   (e.g., record was manually constructed without `find()`).
    pub async fn load(&self) -> Result<&T> {
        if let Some(ref val) = self.loaded {
            return Ok(val);
        }
        if let Some(ref batch) = self.batch {
            let map = batch.get_or_load().await?;
            if let Some(key) = &self.my_key {
                if let Some(data) = map.get(key) {
                    return Ok(data);
                }
            }
        }
        panic!(
            "Relation not loaded. Use find() for automatic batch loading, \
             or call set() to assign data manually."
        );
    }

    /// Try to lazy-load the relation data. Returns `None` if no batch context is available.
    pub async fn try_load(&self) -> Result<Option<&T>> {
        if let Some(ref val) = self.loaded {
            return Ok(Some(val));
        }
        if let Some(ref batch) = self.batch {
            let map = batch.get_or_load().await?;
            if let Some(key) = &self.my_key {
                return Ok(map.get(key));
            }
        }
        Ok(None)
    }
}

impl<T> Relation<T> {
    pub fn new() -> Self {
        Self::default()
    }

    /// Create a relation with data already loaded.
    pub fn with_data(data: T) -> Self {
        Self {
            loaded: Some(data),
            batch: None,
            my_key: None,
        }
    }

    /// Check if the relation data has been loaded.
    pub fn is_loaded(&self) -> bool {
        self.loaded.is_some() || self.batch.as_ref().is_some_and(|b| b.data.initialized())
    }

    /// Alias for `is_loaded()`.
    pub fn loaded(&self) -> bool {
        self.is_loaded()
    }

    /// Get a reference to the loaded data (sync).
    ///
    /// # Panics
    /// Panics if the relation has not been loaded.
    /// Use `try_get()` for a non-panicking alternative,
    /// or `load().await?` for automatic lazy loading.
    pub fn get(&self) -> &T {
        self.loaded
            .as_ref()
            .expect("Relation not loaded. Use load().await? for lazy loading.")
    }

    /// Get a reference to the loaded data, or `None` if not yet loaded.
    pub fn try_get(&self) -> Option<&T> {
        self.loaded.as_ref()
    }

    /// Set the loaded data directly.
    pub fn set(&mut self, data: T) {
        self.loaded = Some(data);
        self.batch = None;
        self.my_key = None;
    }

    /// Attach a batch context for lazy loading (called by `setup_batch_loading`).
    pub fn set_batch(&mut self, batch: Arc<RelationBatch<T>>, key: String) {
        self.batch = Some(batch);
        self.my_key = Some(key);
    }
}

impl<C> Relation<Vec<C>> {
    /// Get a slice of loaded items (for has_many relations).
    ///
    /// # Panics
    /// Panics if the relation has not been loaded.
    /// Use `try_get_many()` for a non-panicking alternative,
    /// or `load().await?` for automatic lazy loading.
    pub fn get_many(&self) -> &[C] {
        self.loaded
            .as_ref()
            .expect("Relation not loaded. Use load().await? for lazy loading.")
            .as_slice()
    }

    /// Get a slice of loaded items, or `None` if not yet loaded.
    pub fn try_get_many(&self) -> Option<&[C]> {
        self.loaded.as_deref()
    }
}

/// Relation type for model definitions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RelationType {
    HasMany,
    HasOne,
    BelongsTo,
}

/// Configuration for a relation definition.
#[derive(Debug, Clone)]
pub struct RelationConfig {
    pub relation_type: RelationType,
    pub source_columns: Vec<String>,
    pub target_columns: Vec<String>,
    pub target_table: String,
    pub order: Option<Vec<OrderSpec>>,
    pub order_raw: Option<String>,
    pub filter: Option<Vec<Condition>>,
    /// Per-parent limit for hasMany: uses ROW_NUMBER() to limit rows per parent key.
    pub limit_per_parent: Option<usize>,
    /// Per-relation hard limit: throws if exceeded (overrides global hasManyHardLimit).
    pub hard_limit: Option<usize>,
}

/// Metadata for a relation, stored in the model.
#[derive(Debug, Clone)]
pub struct RelationMeta {
    pub name: &'static str,
    pub relation_type: RelationType,
    /// Key pairs: (source_column, target_column). Single or composite.
    pub key_pairs: &'static [(&'static str, &'static str)],
    pub target_table: &'static str,
    pub order: Option<&'static str>,
    pub limit_per_parent: Option<usize>,
    pub hard_limit: Option<usize>,
}

// ============================================
// Value key helper for hash-based grouping
// ============================================

pub fn value_to_key(v: &Value) -> String {
    match v {
        Value::Null => "\0null".to_string(),
        Value::Bool(b) => format!("b:{}", b),
        Value::Int(i) => format!("i:{}", i),
        Value::Float(f) => format!("f:{:?}", f),
        Value::Text(s) => format!("t:{}", s),
        Value::Uuid(u) => format!("u:{}", u),
        Value::Timestamp(t) => format!("ts:{}", t),
        Value::Date(d) => format!("d:{}", d),
        _ => format!("{:?}", v),
    }
}

pub fn composite_value_to_key(vals: &[Value]) -> String {
    vals.iter()
        .map(value_to_key)
        .collect::<Vec<_>>()
        .join("\x00")
}

// ============================================
// Utility Functions
// ============================================

/// Extract column values from a slice of records.
pub fn column_list<T>(records: &[T], f: impl Fn(&T) -> Value) -> Vec<Value> {
    records.iter().map(f).collect()
}

/// Map records by a key function into a HashMap (last wins on duplicates).
pub fn hash_by_key<T, K: Eq + std::hash::Hash>(
    records: Vec<T>,
    f: impl Fn(&T) -> K,
) -> HashMap<K, T> {
    let mut map = HashMap::new();
    for r in records {
        let key = f(&r);
        map.insert(key, r);
    }
    map
}

/// Group records by a key function into a HashMap of Vecs.
pub fn group_by_key<T, K: Eq + std::hash::Hash>(
    records: Vec<T>,
    f: impl Fn(&T) -> K,
) -> HashMap<K, Vec<T>> {
    let mut map: HashMap<K, Vec<T>> = HashMap::new();
    for r in records {
        let key = f(&r);
        map.entry(key).or_default().push(r);
    }
    map
}

// ============================================
// Batch Load Functions (return HashMap for lazy loading)
// ============================================

/// Batch-load a has_many relation, returning a HashMap keyed by source key string.
/// Used by the lazy loading system (`Relation::load()`).
pub async fn batch_load_has_many<C: FromRow + Send>(
    executor: &dyn Executor,
    source_keys: &[Value],
    target_table: &str,
    target_col: &str,
    order: Option<&str>,
    limit_per_parent: Option<usize>,
    hard_limit: Option<usize>,
    limit_config: Option<&LimitConfig>,
) -> Result<HashMap<String, Vec<C>>> {
    if source_keys.is_empty() {
        return Ok(HashMap::new());
    }

    let order_clause = order.unwrap_or(target_col);
    let effective_hard_limit =
        hard_limit.or_else(|| limit_config.and_then(|c| c.has_many_hard_limit));
    let is_postgres = executor.driver_type() == DriverType::Postgres;

    let (sql, params) = if let Some(limit) = limit_per_parent {
        if is_postgres {
            let pg_type = value_to_pg_type(&source_keys[0]);
            let inner_sql = format!(
                "SELECT *, ROW_NUMBER() OVER (PARTITION BY {} ORDER BY {}) AS _rn FROM {} WHERE {} = ANY(?::{}[])",
                target_col, order_clause, target_table, target_col, pg_type
            );
            let sql = format!(
                "SELECT * FROM ({}) AS _ranked WHERE _ranked._rn <= {}",
                inner_sql, limit
            );
            (sql, vec![values_to_array_for_postgres(source_keys)])
        } else {
            let placeholders: Vec<&str> = source_keys.iter().map(|_| "?").collect();
            let inner_sql = format!(
                "SELECT *, ROW_NUMBER() OVER (PARTITION BY {} ORDER BY {}) AS _rn FROM {} WHERE {} IN ({})",
                target_col, order_clause, target_table, target_col, placeholders.join(", ")
            );
            let sql = format!(
                "SELECT * FROM ({}) AS _ranked WHERE _ranked._rn <= {}",
                inner_sql, limit
            );
            (sql, source_keys.to_vec())
        }
    } else if is_postgres {
        let pg_type = value_to_pg_type(&source_keys[0]);
        let mut sql = format!(
            "SELECT * FROM {} WHERE {} = ANY(?::{}[])",
            target_table, target_col, pg_type
        );
        if let Some(order) = order {
            sql.push_str(&format!(" ORDER BY {}", order));
        }
        if let Some(limit) = effective_hard_limit {
            sql.push_str(&format!(" LIMIT {}", limit + 1));
        }
        (sql, vec![values_to_array_for_postgres(source_keys)])
    } else {
        let placeholders: Vec<&str> = source_keys.iter().map(|_| "?").collect();
        let mut sql = format!(
            "SELECT * FROM {} WHERE {} IN ({})",
            target_table, target_col, placeholders.join(", ")
        );
        if let Some(order) = order {
            sql.push_str(&format!(" ORDER BY {}", order));
        }
        if let Some(limit) = effective_hard_limit {
            sql.push_str(&format!(" LIMIT {}", limit + 1));
        }
        (sql, source_keys.to_vec())
    };

    let result = executor.execute_query(&sql, &params).await?;

    if let Some(limit) = effective_hard_limit {
        if result.rows.len() > limit {
            return Err(Error::LimitExceeded {
                limit,
                actual_count: result.rows.len(),
                context: LimitContext::Relation,
                model: target_table.to_string(),
            });
        }
    }

    let mut grouped: HashMap<String, Vec<C>> = HashMap::new();
    for row in &result.rows {
        let child = C::from_row(row)?;
        let key_val = row.get(target_col).cloned().unwrap_or(Value::Null);
        let key_str = value_to_key(&key_val);
        grouped.entry(key_str).or_default().push(child);
    }

    for key in source_keys {
        let key_str = value_to_key(key);
        grouped.entry(key_str).or_default();
    }

    Ok(grouped)
}

/// Batch-load a has_one or belongs_to relation, returning a HashMap keyed by source key string.
/// Used by the lazy loading system (`Relation::load()`).
pub async fn batch_load_one<C: FromRow + Send + Clone>(
    executor: &dyn Executor,
    source_keys: &[Value],
    target_table: &str,
    target_col: &str,
) -> Result<HashMap<String, Option<C>>> {
    if source_keys.is_empty() {
        return Ok(HashMap::new());
    }

    let is_postgres = executor.driver_type() == DriverType::Postgres;
    let (sql, params) = if is_postgres {
        let pg_type = value_to_pg_type(&source_keys[0]);
        let sql = format!(
            "SELECT * FROM {} WHERE {} = ANY(?::{}[])",
            target_table, target_col, pg_type
        );
        (sql, vec![values_to_array_for_postgres(source_keys)])
    } else {
        let placeholders: Vec<&str> = source_keys.iter().map(|_| "?").collect();
        let sql = format!(
            "SELECT * FROM {} WHERE {} IN ({})",
            target_table, target_col, placeholders.join(", ")
        );
        (sql, source_keys.to_vec())
    };

    let result = executor.execute_query(&sql, &params).await?;

    let mut mapped: HashMap<String, Option<C>> = HashMap::new();
    for row in &result.rows {
        let child = C::from_row(row)?;
        let key_val = row.get(target_col).cloned().unwrap_or(Value::Null);
        let key_str = value_to_key(&key_val);
        mapped.insert(key_str, Some(child));
    }

    for key in source_keys {
        let key_str = value_to_key(key);
        mapped.entry(key_str).or_insert(None);
    }

    Ok(mapped)
}

/// Batch-load a has_many relation with composite keys, returning a HashMap.
pub async fn batch_load_has_many_composite<C: FromRow + Send>(
    executor: &dyn Executor,
    source_key_tuples: &[Vec<Value>],
    target_table: &str,
    target_columns: &[&str],
    order: Option<&str>,
    limit_per_parent: Option<usize>,
    hard_limit: Option<usize>,
    limit_config: Option<&LimitConfig>,
) -> Result<HashMap<String, Vec<C>>> {
    if source_key_tuples.is_empty() || target_columns.is_empty() {
        return Ok(HashMap::new());
    }

    let order_clause = order.unwrap_or(target_columns[0]);
    let effective_hard_limit =
        hard_limit.or_else(|| limit_config.and_then(|c| c.has_many_hard_limit));
    let cols = target_columns.join(", ");
    let is_postgres = executor.driver_type() == DriverType::Postgres;

    let (sql, params) = if is_postgres {
        let col_arrays: Vec<Value> = (0..target_columns.len())
            .map(|i| {
                let col_vals: Vec<Value> =
                    source_key_tuples.iter().map(|t| t[i].clone()).collect();
                values_to_array_for_postgres(&col_vals)
            })
            .collect();
        let unnest_parts: Vec<String> = source_key_tuples[0]
            .iter()
            .map(|v| format!("?::{}[]", value_to_pg_type(v)))
            .collect();

        if let Some(limit) = limit_per_parent {
            let partition_by = target_columns.join(", ");
            let inner_sql = format!(
                "SELECT *, ROW_NUMBER() OVER (PARTITION BY {} ORDER BY {}) AS _rn FROM {} WHERE ({}) IN (SELECT * FROM UNNEST({}))",
                partition_by, order_clause, target_table, cols, unnest_parts.join(", ")
            );
            let sql = format!(
                "SELECT * FROM ({}) AS _ranked WHERE _ranked._rn <= {}",
                inner_sql, limit
            );
            (sql, col_arrays)
        } else {
            let mut sql = format!(
                "SELECT * FROM {} WHERE ({}) IN (SELECT * FROM UNNEST({}))",
                target_table, cols, unnest_parts.join(", ")
            );
            if let Some(o) = order {
                sql.push_str(&format!(" ORDER BY {}", o));
            }
            if let Some(limit) = effective_hard_limit {
                sql.push_str(&format!(" LIMIT {}", limit + 1));
            }
            (sql, col_arrays)
        }
    } else {
        let tuple_placeholders: Vec<String> = source_key_tuples
            .iter()
            .map(|_| {
                let p = target_columns
                    .iter()
                    .map(|_| "?")
                    .collect::<Vec<_>>()
                    .join(", ");
                format!("({})", p)
            })
            .collect();
        let in_clause = tuple_placeholders.join(", ");
        let params: Vec<Value> = source_key_tuples.iter().flatten().cloned().collect();

        if let Some(limit) = limit_per_parent {
            let partition_by = target_columns.join(", ");
            let inner_sql = format!(
                "SELECT *, ROW_NUMBER() OVER (PARTITION BY {} ORDER BY {}) AS _rn FROM {} WHERE ({}) IN ({})",
                partition_by, order_clause, target_table, cols, in_clause
            );
            let sql = format!(
                "SELECT * FROM ({}) AS _ranked WHERE _ranked._rn <= {}",
                inner_sql, limit
            );
            (sql, params)
        } else {
            let mut sql = format!(
                "SELECT * FROM {} WHERE ({}) IN ({})",
                target_table, cols, in_clause
            );
            if let Some(o) = order {
                sql.push_str(&format!(" ORDER BY {}", o));
            }
            if let Some(limit) = effective_hard_limit {
                sql.push_str(&format!(" LIMIT {}", limit + 1));
            }
            (sql, params)
        }
    };

    let result = executor.execute_query(&sql, &params).await?;

    if let Some(limit) = effective_hard_limit {
        if result.rows.len() > limit {
            return Err(Error::LimitExceeded {
                limit,
                actual_count: result.rows.len(),
                context: LimitContext::Relation,
                model: target_table.to_string(),
            });
        }
    }

    let mut grouped: HashMap<String, Vec<C>> = HashMap::new();
    for row in &result.rows {
        let child = C::from_row(row)?;
        let key_vals: Vec<Value> = target_columns
            .iter()
            .filter_map(|c| row.get(*c).cloned())
            .collect();
        let key_str = composite_value_to_key(&key_vals);
        grouped.entry(key_str).or_default().push(child);
    }

    for tuple in source_key_tuples {
        let key_str = composite_value_to_key(tuple);
        grouped.entry(key_str).or_default();
    }

    Ok(grouped)
}

/// Batch-load a has_one or belongs_to relation with composite keys, returning a HashMap.
pub async fn batch_load_one_composite<C: FromRow + Send + Clone>(
    executor: &dyn Executor,
    source_key_tuples: &[Vec<Value>],
    target_table: &str,
    target_columns: &[&str],
) -> Result<HashMap<String, Option<C>>> {
    if source_key_tuples.is_empty() || target_columns.is_empty() {
        return Ok(HashMap::new());
    }

    let cols = target_columns.join(", ");
    let is_postgres = executor.driver_type() == DriverType::Postgres;

    let (sql, params) = if is_postgres {
        let col_arrays: Vec<Value> = (0..target_columns.len())
            .map(|i| {
                let col_vals: Vec<Value> =
                    source_key_tuples.iter().map(|t| t[i].clone()).collect();
                values_to_array_for_postgres(&col_vals)
            })
            .collect();
        let unnest_parts: Vec<String> = source_key_tuples[0]
            .iter()
            .map(|v| format!("?::{}[]", value_to_pg_type(v)))
            .collect();
        let sql = format!(
            "SELECT * FROM {} WHERE ({}) IN (SELECT * FROM UNNEST({}))",
            target_table, cols, unnest_parts.join(", ")
        );
        (sql, col_arrays)
    } else {
        let tuple_placeholders: Vec<String> = source_key_tuples
            .iter()
            .map(|_| {
                let p = target_columns
                    .iter()
                    .map(|_| "?")
                    .collect::<Vec<_>>()
                    .join(", ");
                format!("({})", p)
            })
            .collect();
        let in_clause = tuple_placeholders.join(", ");
        let params: Vec<Value> = source_key_tuples.iter().flatten().cloned().collect();
        let sql = format!(
            "SELECT * FROM {} WHERE ({}) IN ({})",
            target_table, cols, in_clause
        );
        (sql, params)
    };

    let result = executor.execute_query(&sql, &params).await?;

    let mut mapped: HashMap<String, Option<C>> = HashMap::new();
    for row in &result.rows {
        let child = C::from_row(row)?;
        let key_vals: Vec<Value> = target_columns
            .iter()
            .filter_map(|c| row.get(*c).cloned())
            .collect();
        let key_str = composite_value_to_key(&key_vals);
        mapped.insert(key_str, Some(child));
    }

    for tuple in source_key_tuples {
        let key_str = composite_value_to_key(tuple);
        mapped.entry(key_str).or_insert(None);
    }

    Ok(mapped)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::driver::{QueryResult, Row};
    use crate::model::Executor;
    use crate::types::DriverType;
    use std::sync::{Arc, Mutex};

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

    impl MockExecutor {
        fn new(rows: Vec<Row>) -> Self {
            Self {
                result_rows: rows,
                last_sql: Arc::new(Mutex::new(String::new())),
            }
        }
        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>>
        {
            Box::pin(async {
                Ok(QueryResult {
                    rows: vec![],
                    rows_affected: 0,
                })
            })
        }
        fn driver_type(&self) -> DriverType {
            DriverType::Sqlite
        }
    }

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

    #[derive(Debug, PartialEq, Clone)]
    struct ChildRecord {
        user_id: i64,
        name: String,
    }

    impl crate::model::FromRow for ChildRecord {
        fn from_row(row: &Row) -> Result<Self> {
            let user_id = match row.get("user_id") {
                Some(Value::Int(i)) => *i,
                _ => 0,
            };
            let name = match row.get("name") {
                Some(Value::Text(s)) => s.clone(),
                _ => String::new(),
            };
            Ok(ChildRecord { user_id, name })
        }
    }

    #[tokio::test]
    async fn test_batch_load_has_many_basic() {
        let rows = vec![make_row(1, "a"), make_row(1, "b"), make_row(2, "c")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1), Value::Int(2)];
        let result: HashMap<String, Vec<ChildRecord>> =
            batch_load_has_many(&executor, &keys, "children", "user_id", None, None, None, None)
                .await
                .unwrap();
        assert_eq!(result.get("i:1").unwrap().len(), 2);
        assert_eq!(result.get("i:1").unwrap()[0].name, "a");
        assert_eq!(result.get("i:1").unwrap()[1].name, "b");
        assert_eq!(result.get("i:2").unwrap().len(), 1);
        assert_eq!(result.get("i:2").unwrap()[0].name, "c");
    }

    #[tokio::test]
    async fn test_batch_load_has_many_empty() {
        let executor = MockExecutor::new(vec![]);
        let keys: Vec<Value> = vec![];
        let result: HashMap<String, Vec<ChildRecord>> =
            batch_load_has_many(&executor, &keys, "children", "user_id", None, None, None, None)
                .await
                .unwrap();
        assert!(result.is_empty());
        assert!(executor.last_sql().is_empty());
    }

    #[tokio::test]
    async fn test_batch_load_one_basic() {
        let rows = vec![make_row(1, "profile_a"), make_row(2, "profile_b")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1), Value::Int(2)];
        let result: HashMap<String, Option<ChildRecord>> =
            batch_load_one(&executor, &keys, "profiles", "user_id")
                .await
                .unwrap();
        assert_eq!(result.get("i:1").unwrap().as_ref().unwrap().name, "profile_a");
        assert_eq!(result.get("i:2").unwrap().as_ref().unwrap().name, "profile_b");
    }

    #[tokio::test]
    async fn test_batch_load_one_missing_child() {
        let rows = vec![make_row(1, "only_one")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1), Value::Int(2)];
        let result: HashMap<String, Option<ChildRecord>> =
            batch_load_one(&executor, &keys, "profiles", "user_id")
                .await
                .unwrap();
        assert_eq!(result.get("i:1").unwrap().as_ref().unwrap().name, "only_one");
        assert!(result.get("i:2").unwrap().is_none());
    }

    #[tokio::test]
    async fn test_batch_load_has_many_with_limit_enforced() {
        let rows = vec![make_row(1, "a"), make_row(1, "b"), make_row(1, "c")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1)];
        let limit_config = LimitConfig {
            has_many_hard_limit: Some(2),
            ..Default::default()
        };
        let err =
            batch_load_has_many::<ChildRecord>(&executor, &keys, "children", "user_id", None, None, None, Some(&limit_config))
                .await
                .unwrap_err();
        match &err {
            Error::LimitExceeded {
                limit,
                actual_count,
                context,
                model,
            } => {
                assert_eq!(*limit, 2);
                assert_eq!(*actual_count, 3);
                assert_eq!(*context, LimitContext::Relation);
                assert_eq!(model, "children");
            }
            _ => panic!("Expected LimitExceeded, got {:?}", err),
        }
        assert!(executor.last_sql().contains("LIMIT 3"));
    }

    #[tokio::test]
    async fn test_batch_load_has_many_limit_per_parent() {
        let rows = vec![make_row(1, "a"), make_row(1, "b"), make_row(2, "e")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1), Value::Int(2)];
        let result: HashMap<String, Vec<ChildRecord>> =
            batch_load_has_many(&executor, &keys, "children", "user_id", Some("name ASC"), Some(2), None, None)
                .await
                .unwrap();
        assert_eq!(result.get("i:1").unwrap().len(), 2);
        assert_eq!(result.get("i:1").unwrap()[0].name, "a");
        assert_eq!(result.get("i:2").unwrap().len(), 1);
        let sql = executor.last_sql();
        assert!(sql.contains("ROW_NUMBER()"));
        assert!(sql.contains("PARTITION BY user_id"));
        assert!(sql.contains("ORDER BY name ASC"));
        assert!(sql.contains("_rn <= 2"));
    }

    #[tokio::test]
    async fn test_batch_load_has_many_within_limit_succeeds() {
        let rows = vec![make_row(1, "a"), make_row(1, "b")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1)];
        let limit_config = LimitConfig {
            has_many_hard_limit: Some(5),
            ..Default::default()
        };
        let result: HashMap<String, Vec<ChildRecord>> =
            batch_load_has_many(&executor, &keys, "children", "user_id", None, None, None, Some(&limit_config))
                .await
                .unwrap();
        assert_eq!(result.get("i:1").unwrap().len(), 2);
        assert!(executor.last_sql().contains("LIMIT 6"));
    }

    #[tokio::test]
    async fn test_batch_load_has_many_hard_limit_overrides_global() {
        let rows = vec![make_row(1, "a"), make_row(1, "b"), make_row(1, "c")];
        let executor = MockExecutor::new(rows);
        let keys = vec![Value::Int(1)];
        let limit_config = LimitConfig {
            has_many_hard_limit: Some(10),
            ..Default::default()
        };
        let err =
            batch_load_has_many::<ChildRecord>(&executor, &keys, "children", "user_id", None, None, Some(2), Some(&limit_config))
                .await
                .unwrap_err();
        match &err {
            Error::LimitExceeded {
                limit,
                actual_count,
                ..
            } => {
                assert_eq!(*limit, 2);
                assert_eq!(*actual_count, 3);
            }
            _ => panic!("Expected LimitExceeded, got {:?}", err),
        }
    }

    #[test]
    fn test_relation_default() {
        let rel: Relation<Vec<String>> = Relation::default();
        assert!(!rel.is_loaded());
        assert!(rel.try_get().is_none());
    }

    #[test]
    fn test_relation_with_data() {
        let rel = Relation::with_data(vec!["a".to_string(), "b".to_string()]);
        assert!(rel.is_loaded());
        assert_eq!(rel.get().len(), 2);
    }

    #[test]
    fn test_relation_set() {
        let mut rel: Relation<i64> = Relation::new();
        assert!(!rel.is_loaded());
        rel.set(42);
        assert!(rel.is_loaded());
        assert_eq!(*rel.get(), 42);
    }

    #[test]
    fn test_relation_loaded_alias() {
        let rel: Relation<Vec<String>> = Relation::default();
        assert!(!rel.loaded());
        let rel = Relation::with_data(vec!["a".to_string()]);
        assert!(rel.loaded());
    }

    #[test]
    fn test_relation_get_many() {
        let mut rel = Relation::with_data(vec![1i64, 2, 3]);
        assert_eq!(rel.get_many(), &[1, 2, 3]);
        rel.set(vec![10, 20]);
        assert_eq!(rel.get_many(), &[10, 20]);
    }

    #[test]
    #[should_panic(expected = "Relation not loaded")]
    fn test_relation_get_panics_when_not_loaded() {
        let rel: Relation<String> = Relation::new();
        let _ = rel.get();
    }

    #[test]
    #[should_panic(expected = "Relation not loaded")]
    fn test_relation_get_many_panics_when_not_loaded() {
        let rel: Relation<Vec<i64>> = Relation::new();
        let _ = rel.get_many();
    }

    #[test]
    fn test_relation_try_get() {
        let rel: Relation<String> = Relation::new();
        assert!(rel.try_get().is_none());
        let rel = Relation::with_data("hello".to_string());
        assert_eq!(rel.try_get(), Some(&"hello".to_string()));
    }

    #[test]
    fn test_relation_try_get_many() {
        let rel: Relation<Vec<i64>> = Relation::new();
        assert!(rel.try_get_many().is_none());
        let rel = Relation::with_data(vec![1i64, 2, 3]);
        assert_eq!(rel.try_get_many(), Some([1, 2, 3].as_slice()));
    }

    #[test]
    fn test_column_list() {
        let records = vec![("alice", 1i64), ("bob", 2), ("carol", 3)];
        let ids = column_list(&records, |r| Value::Int(r.1));
        assert_eq!(ids, vec![Value::Int(1), Value::Int(2), Value::Int(3)]);
    }

    #[test]
    fn test_hash_by_key() {
        let records = vec![("alice", 1), ("bob", 2), ("carol", 3)];
        let map = hash_by_key(records, |r| r.1);
        assert_eq!(map.get(&1), Some(&("alice", 1)));
        assert_eq!(map.get(&2), Some(&("bob", 2)));
        assert_eq!(map.len(), 3);
    }

    #[test]
    fn test_group_by_key() {
        let records = vec![
            ("alice", "team_a"),
            ("bob", "team_b"),
            ("carol", "team_a"),
            ("dave", "team_b"),
        ];
        let map = group_by_key(records, |r| r.1);
        assert_eq!(map.get("team_a").unwrap().len(), 2);
        assert_eq!(map.get("team_b").unwrap().len(), 2);
    }

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

    #[derive(Debug, PartialEq, Clone)]
    struct CompositeChildRecord {
        tenant_id: i64,
        user_id: i64,
        name: String,
    }

    impl crate::model::FromRow for CompositeChildRecord {
        fn from_row(row: &Row) -> Result<Self> {
            let tenant_id = match row.get("tenant_id") {
                Some(Value::Int(i)) => *i,
                _ => 0,
            };
            let user_id = match row.get("user_id") {
                Some(Value::Int(i)) => *i,
                _ => 0,
            };
            let name = match row.get("name") {
                Some(Value::Text(s)) => s.clone(),
                _ => String::new(),
            };
            Ok(CompositeChildRecord {
                tenant_id,
                user_id,
                name,
            })
        }
    }

    #[tokio::test]
    async fn test_batch_load_has_many_composite() {
        let rows = vec![
            make_row_composite(1, 10, "a"),
            make_row_composite(1, 10, "b"),
            make_row_composite(1, 20, "c"),
        ];
        let executor = MockExecutor::new(rows);
        let tuples = vec![
            vec![Value::Int(1), Value::Int(10)],
            vec![Value::Int(1), Value::Int(20)],
        ];
        let result: HashMap<String, Vec<CompositeChildRecord>> =
            batch_load_has_many_composite(
                &executor, &tuples, "children", &["tenant_id", "user_id"],
                None, None, None, None,
            )
            .await
            .unwrap();
        let key1 = composite_value_to_key(&[Value::Int(1), Value::Int(10)]);
        let key2 = composite_value_to_key(&[Value::Int(1), Value::Int(20)]);
        assert_eq!(result.get(&key1).unwrap().len(), 2);
        assert_eq!(result.get(&key1).unwrap()[0].name, "a");
        assert_eq!(result.get(&key2).unwrap().len(), 1);
        assert_eq!(result.get(&key2).unwrap()[0].name, "c");
        let sql = executor.last_sql();
        assert!(sql.contains("(tenant_id, user_id) IN"));
    }

    #[tokio::test]
    async fn test_batch_load_one_composite() {
        let rows = vec![
            make_row_composite(1, 10, "profile_a"),
            make_row_composite(1, 20, "profile_b"),
        ];
        let executor = MockExecutor::new(rows);
        let tuples = vec![
            vec![Value::Int(1), Value::Int(10)],
            vec![Value::Int(1), Value::Int(20)],
        ];
        let result: HashMap<String, Option<CompositeChildRecord>> =
            batch_load_one_composite(
                &executor, &tuples, "profiles", &["tenant_id", "user_id"],
            )
            .await
            .unwrap();
        let key1 = composite_value_to_key(&[Value::Int(1), Value::Int(10)]);
        let key2 = composite_value_to_key(&[Value::Int(1), Value::Int(20)]);
        assert_eq!(result.get(&key1).unwrap().as_ref().unwrap().name, "profile_a");
        assert_eq!(result.get(&key2).unwrap().as_ref().unwrap().name, "profile_b");
    }

    #[test]
    fn test_value_to_key() {
        assert_eq!(value_to_key(&Value::Int(42)), "i:42");
        assert_eq!(value_to_key(&Value::Text("hello".into())), "t:hello");
        assert_eq!(value_to_key(&Value::Bool(true)), "b:true");
        assert_eq!(value_to_key(&Value::Null), "\0null");
        assert_ne!(
            value_to_key(&Value::Int(1)),
            value_to_key(&Value::Text("1".into()))
        );
    }
}
