//! Generic mapping from type-safe `FilterNode`<F> to `SeaORM` Conditions.
//!
//! This gear provides the core logic for converting DTO-level filter expressions
//! into `SeaORM` conditions. Concrete gears only need to provide a mapping from
//! their DTO field enum to `SeaORM` Column types via the `FieldToColumn` trait.

use crate::secure::{Scoped, SecureSelect};
use bigdecimal::ToPrimitive;
use chrono::SecondsFormat;
use sea_orm::{
    Condition, EntityTrait, QueryFilter, QueryOrder, QuerySelect,
    sea_query::{Expr, Order},
};
use toolkit_odata::filter::{
    FieldKind, FilterField, FilterNode, FilterOp, ODataValue, convert_expr_to_filter_node,
};
use toolkit_odata::{CursorV1, Error as ODataError, ODataOrderBy, Page, PageInfo, SortDir};

use crate::secure::{DBRunner, DBRunnerInternal, SeaOrmRunner};

/// Trait for mapping DTO filter fields to `SeaORM` columns.
///
/// This trait abstracts the mapping from a type-safe field enum (generated by
/// `#[derive(ODataFilterable)]`) to the corresponding `SeaORM` Column enum.
///
/// # Example
///
/// ```ignore
/// use toolkit_db::odata::sea_orm_filter::FieldToColumn;
/// use crate::api::rest::dto::UserDtoFilterField;
/// use crate::testing::storage::entity::Column;
///
/// struct UserFieldMapper;
///
/// impl FieldToColumn<UserDtoFilterField> for UserFieldMapper {
///     type Column = Column;
///
///     fn map_field(field: UserDtoFilterField) -> Column {
///         match field {
///             UserDtoFilterField::Id => Column::Id,
///             UserDtoFilterField::Email => Column::Email,
///             UserDtoFilterField::CreatedAt => Column::CreatedAt,
///         }
///     }
/// }
/// ```
pub trait FieldToColumn<F: FilterField> {
    /// The `SeaORM` Column type for the entity
    type Column: Clone + sea_orm::Iden + sea_orm::ColumnTrait + sea_orm::IntoSimpleExpr + 'static;

    /// Map a DTO filter field to a `SeaORM` column
    fn map_field(field: F) -> Self::Column;

    /// Optional per-field value normalisation. Default returns the
    /// value unchanged. Implementors override this when the wire
    /// representation of a field differs from its storage
    /// representation — e.g. an SDK that exposes a lifecycle column
    /// as a string enum (`"active"` / `"deleted"`) backed by a
    /// `SMALLINT` column in the database. The framework calls this
    /// hook inside [`filter_node_to_condition`] AFTER the validated
    /// `FilterNode` is built and BEFORE the value is bound into the
    /// `SeaORM` predicate, so the returned [`ODataValue`] becomes
    /// the effective value the database sees.
    ///
    /// `op` is the operator the value is being bound for:
    /// [`FilterOp::Eq`] / [`FilterOp::Ne`] / [`FilterOp::Gt`] /
    /// [`FilterOp::Ge`] / [`FilterOp::Lt`] / [`FilterOp::Le`] /
    /// [`FilterOp::Contains`] / [`FilterOp::StartsWith`] /
    /// [`FilterOp::EndsWith`] for binary nodes, and
    /// [`FilterOp::In`] for membership (`$filter=col in (...)`)
    /// nodes. Implementors that translate wire shape to storage shape
    /// SHOULD reject operators whose semantics differ between the two
    /// shapes — typically the ordered comparisons (`Gt`/`Ge`/`Lt`/`Le`)
    /// on a categorical column whose numeric storage encoding is not a
    /// meaningful order over the wire string values.
    ///
    /// # Errors
    ///
    /// Returns a human-readable error string when the wire value is
    /// invalid for this field (typically: enum string outside the
    /// allowed set) or when `op` is not supported on the translated
    /// column. The framework wraps the message as
    /// `ODataError::InvalidFilter`; the call site (e.g.
    /// [`paginate_odata_try`]) maps that to the gear's validation
    /// category.
    fn map_value(_field: F, _op: FilterOp, value: &ODataValue) -> Result<ODataValue, String> {
        Ok(value.clone())
    }

    /// Whether `field` is admissible in `$orderby` (and therefore as
    /// a key in cursor-based pagination). Default `true`. Implementors
    /// override with `false` when the wire shape of the field differs
    /// from its storage shape — there is no consistent ordering
    /// between the two shapes (alphabetical strings vs ordinal
    /// integers), and the cursor encoding path relies on
    /// `field.kind()` to read tokens back, which would fail to decode
    /// a storage-encoded value. The framework checks this in
    /// `paginate_odata_collect` before composing the effective order
    /// and rejects the `$orderby` clause as
    /// [`ODataError::InvalidOrderByField`] — keeping the diagnostic
    /// at the same site as unknown-field rejections.
    fn is_orderable(_field: F) -> bool {
        true
    }
}

/// Extended trait for `OData` field mapping including cursor extraction.
///
/// This trait extends `FieldToColumn` with the ability to extract cursor values
/// from entity models, enabling type-safe pagination with ordering.
///
/// # Example
///
/// ```ignore
/// struct UserODataMapper;
///
/// impl FieldToColumn<UserDtoFilterField> for UserODataMapper {
///     type Column = Column;
///     fn map_field(field: UserDtoFilterField) -> Column { /* ... */ }
/// }
///
/// impl ODataFieldMapping<UserDtoFilterField> for UserODataMapper {
///     type Entity = Entity;
///
///     fn extract_cursor_value(model: &Model, field: UserDtoFilterField) -> sea_orm::Value {
///         match field {
///             UserDtoFilterField::Id => sea_orm::Value::Uuid(Some(Box::new(model.id))),
///             UserDtoFilterField::Email => sea_orm::Value::String(Some(Box::new(model.email.clone()))),
///             UserDtoFilterField::CreatedAt => sea_orm::Value::ChronoDateTimeUtc(Some(Box::new(model.created_at))),
///         }
///     }
/// }
/// ```
pub trait ODataFieldMapping<F: FilterField>: FieldToColumn<F> {
    /// The `SeaORM` Entity type
    type Entity: EntityTrait;

    /// Extract a cursor value from a model for a specific field.
    ///
    /// This is used for cursor-based pagination to encode the values of
    /// ordering fields into the cursor.
    fn extract_cursor_value(
        model: &<Self::Entity as EntityTrait>::Model,
        field: F,
    ) -> sea_orm::Value;

    /// Extract cursor values for all fields in an order.
    ///
    /// This is a convenience method that can be overridden for optimization,
    /// but has a default implementation.
    ///
    /// # Errors
    /// Returns an error string if an unknown orderby field is encountered.
    fn extract_cursor_values(
        model: &<Self::Entity as EntityTrait>::Model,
        order: &ODataOrderBy,
    ) -> Result<Vec<(F, sea_orm::Value)>, String> {
        let mut values = Vec::new();
        for order_key in &order.0 {
            let field = F::from_name(&order_key.field)
                .ok_or_else(|| format!("Unknown orderby field: {}", order_key.field))?;
            let value = Self::extract_cursor_value(model, field);
            values.push((field, value));
        }
        Ok(values)
    }
}

/// Convert a `FilterNode`<F> to a `SeaORM` Condition using a `FieldToColumn` mapping.
///
/// This function provides generic traversal of the `FilterNode` AST and handles
/// all standard `OData` operations. Concrete gears only need to implement
/// `FieldToColumn` to map their DTO fields to database columns.
///
/// # Type Parameters
///
/// - `F`: The `FilterField` implementation (generated by `#[derive(ODataFilterable)]`)
/// - `M`: The `FieldToColumn` mapper that maps F to `SeaORM` columns
///
/// # Arguments
///
/// - `filter`: The type-safe filter node to convert
///
/// # Returns
///
/// A `SeaORM` Condition that can be applied to a query, or an error string if
/// the conversion fails.
///
/// # Example
///
/// ```ignore
/// let condition = filter_node_to_condition::<UserDtoFilterField, UserFieldMapper>(&filter)?;
/// let query = Entity::find().filter(condition);
/// ```
///
/// # Errors
/// Returns an error string if the filter contains unsupported operations or invalid values.
pub fn filter_node_to_condition<F, M>(filter: &FilterNode<F>) -> Result<Condition, String>
where
    F: FilterField,
    M: FieldToColumn<F>,
{
    match filter {
        FilterNode::Binary { field, op, value } => {
            // Map DTO field to database column. The `map_value` hook
            // runs per-field-instance and per-operator so a mapper can
            // translate wire-shape values (e.g. enum-as-string
            // `"deleted"`) into storage-shape values (e.g. SMALLINT
            // `3`) without the framework having to know about the
            // enum encoding, AND reject operators whose semantics
            // differ between the two shapes (typically the ordered
            // comparisons). Default impl is identity and accepts every
            // operator, so existing call sites are unaffected.
            let column = M::map_field(*field);
            let mapped = M::map_value(*field, *op, value)?;
            build_binary_condition(column, *op, &mapped)
        }
        FilterNode::InList { field, values } => {
            if values.is_empty() {
                return Err("IN list must not be empty".to_owned());
            }
            let column = M::map_field(*field);
            let kind = field.kind();
            let sea_values: Vec<sea_orm::Value> = values
                .iter()
                .map(|v| {
                    let mapped = M::map_value(*field, FilterOp::In, v)?;
                    odata_value_to_sea_value_for_kind(&mapped, kind)
                })
                .collect::<Result<_, _>>()?;
            Ok(Condition::all().add(Expr::col(column).is_in(sea_values)))
        }
        FilterNode::Composite { op, children } => {
            // Combine child conditions with AND or OR
            let base = match op {
                FilterOp::And => Condition::all(),
                FilterOp::Or => Condition::any(),
                _ => return Err(format!("Invalid composite operator: {op:?}")),
            };

            children.iter().try_fold(base, |acc, child| {
                let child_cond = filter_node_to_condition::<F, M>(child)?;
                Ok(acc.add(child_cond))
            })
        }
        FilterNode::Not(inner) => {
            // FIXED: Call .not() AFTER adding the inner condition
            let inner_cond = filter_node_to_condition::<F, M>(inner)?;
            Ok(Condition::all().add(inner_cond).not())
        }
    }
}

/// Build a binary condition (field op value) for `SeaORM`.
///
/// This handles all comparison and string function operations.
fn build_binary_condition<C>(
    column: C,
    op: FilterOp,
    value: &ODataValue,
) -> Result<Condition, String>
where
    C: sea_orm::Iden + sea_orm::ColumnTrait + sea_orm::IntoSimpleExpr + Clone + 'static,
{
    // Convert ODataValue to sea_orm::Value
    let sea_value = odata_value_to_sea_value(value)?;

    // Handle NULL specially
    if matches!(value, ODataValue::Null) {
        return Ok(match op {
            FilterOp::Eq => Condition::all().add(Expr::col(column).is_null()),
            FilterOp::Ne => Condition::all().add(Expr::col(column).is_not_null()),
            _ => return Err(format!("Unsupported operator for NULL: {op:?}")),
        });
    }

    // Build the expression based on the operator
    let expr = match op {
        FilterOp::Eq => Expr::col(column).eq(sea_value),
        FilterOp::Ne => Expr::col(column).ne(sea_value),
        FilterOp::Gt => Expr::col(column).gt(sea_value),
        FilterOp::Ge => Expr::col(column).gte(sea_value),
        FilterOp::Lt => Expr::col(column).lt(sea_value),
        FilterOp::Le => Expr::col(column).lte(sea_value),
        FilterOp::Contains => {
            let s = extract_string(value)?;
            Expr::col(column).like(format!("%{}%", escape_like(&s)))
        }
        FilterOp::StartsWith => {
            let s = extract_string(value)?;
            Expr::col(column).like(format!("{}%", escape_like(&s)))
        }
        FilterOp::EndsWith => {
            let s = extract_string(value)?;
            Expr::col(column).like(format!("%{}", escape_like(&s)))
        }
        FilterOp::In | FilterOp::And | FilterOp::Or => {
            return Err(format!("Operator {op:?} not valid in binary context"));
        }
    };

    Ok(Condition::all().add(expr))
}

/// Convert an `ODataValue` to a `sea_orm::Value`.
fn odata_value_to_sea_value(value: &ODataValue) -> Result<sea_orm::Value, String> {
    Ok(match value {
        ODataValue::String(s) => sea_orm::Value::String(Some(Box::new(s.clone()))),
        ODataValue::Number(n) => {
            // Try to convert to i64 first, then f64
            if let Some(i) = n.to_i64() {
                sea_orm::Value::BigInt(Some(i))
            } else if let Some(f) = n.to_f64() {
                sea_orm::Value::Double(Some(f))
            } else {
                return Err("Number value out of range".to_owned());
            }
        }
        ODataValue::Bool(b) => sea_orm::Value::Bool(Some(*b)),
        ODataValue::Uuid(u) => sea_orm::Value::Uuid(Some(Box::new(*u))),
        ODataValue::DateTime(dt) => sea_orm::Value::ChronoDateTimeUtc(Some(Box::new(*dt))),
        ODataValue::Date(d) => sea_orm::Value::ChronoDate(Some(Box::new(*d))),
        ODataValue::Time(t) => sea_orm::Value::ChronoTime(Some(Box::new(*t))),
        ODataValue::Null => {
            return Err("NULL values should be handled separately".to_owned());
        }
    })
}

/// Convert an `ODataValue` to a `sea_orm::Value`, using the field's `FieldKind`
/// to coerce numeric values to the correct SQL type (especially Decimal).
fn odata_value_to_sea_value_for_kind(
    value: &ODataValue,
    kind: FieldKind,
) -> Result<sea_orm::Value, String> {
    match (kind, value) {
        (FieldKind::I64, ODataValue::Number(n)) => n
            .to_i64()
            .map(|i| sea_orm::Value::BigInt(Some(i)))
            .ok_or_else(|| "Number value out of range for i64".to_owned()),
        (FieldKind::F64, ODataValue::Number(n)) => n
            .to_f64()
            .map(|f| sea_orm::Value::Double(Some(f)))
            .ok_or_else(|| "Number value out of range for f64".to_owned()),
        (FieldKind::Decimal, ODataValue::Number(n)) => {
            let d = n
                .to_string()
                .parse::<rust_decimal::Decimal>()
                .map_err(|_| "Invalid decimal value".to_owned())?;
            Ok(sea_orm::Value::Decimal(Some(Box::new(d))))
        }
        _ => odata_value_to_sea_value(value),
    }
}

/// Extract a string from an `ODataValue`.
fn extract_string(value: &ODataValue) -> Result<String, String> {
    match value {
        ODataValue::String(s) => Ok(s.clone()),
        _ => Err(format!("Expected string value, got {value:?}")),
    }
}

/// Escape special characters in LIKE patterns.
///
/// This escapes `%`, `_`, and `\` to prevent them from being interpreted
/// as wildcards in SQL LIKE patterns.
#[must_use]
pub fn escape_like(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for ch in s.chars() {
        match ch {
            '%' | '_' | '\\' => {
                out.push('\\');
                out.push(ch);
            }
            c => out.push(c),
        }
    }
    out
}

/// Encode a `sea_orm::Value` to a string for cursor storage.
///
/// This converts `SeaORM` values to strings that can be embedded in cursors.
///
/// # Errors
/// Returns an error string if the value type doesn't match the expected kind.
pub fn encode_cursor_value(value: &sea_orm::Value, kind: FieldKind) -> Result<String, String> {
    use sea_orm::Value as V;

    let result: Result<String, String> = match (kind, value) {
        (FieldKind::String, V::String(Some(s))) => Ok(s.to_string()),
        (FieldKind::I64, V::BigInt(Some(i))) => Ok(i.to_string()),
        (FieldKind::F64, V::Double(Some(f))) => Ok(ryu::Buffer::new().format(*f).to_owned()),
        (FieldKind::Bool, V::Bool(Some(b))) => Ok(b.to_string()),
        (FieldKind::Uuid, V::Uuid(Some(u))) => Ok(u.to_string()),
        (FieldKind::DateTimeUtc, V::ChronoDateTimeUtc(Some(dt))) => {
            Ok(dt.to_rfc3339_opts(SecondsFormat::Nanos, true))
        }
        (FieldKind::DateTimeUtc, V::TimeDateTimeWithTimeZone(Some(dt))) => {
            let fmt = time::format_description::parse_borrowed::<2>(
                "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:9]Z",
            )
            .map_err(|_| "invalid datetime format".to_owned())?;
            dt.to_offset(time::UtcOffset::UTC)
                .format(&fmt)
                .map_err(|_| "failed to format datetime".to_owned())
        }
        (FieldKind::Date, V::ChronoDate(Some(d))) => Ok(d.to_string()),
        (FieldKind::Time, V::ChronoTime(Some(t))) => Ok(t.to_string()),
        (FieldKind::Decimal, V::Decimal(Some(d))) => Ok(d.to_string()),
        _ => Err("Unsupported or mismatched cursor value type".to_owned()),
    };

    result
}

/// Parse a cursor value from string based on field kind.
///
/// This is the inverse of `encode_cursor_value`, converting cursor strings
/// back to typed `SeaORM` values for comparison in queries.
///
/// # Errors
/// Returns an error string if the string cannot be parsed as the expected type.
pub fn parse_cursor_value(kind: FieldKind, s: &str) -> Result<sea_orm::Value, String> {
    use sea_orm::Value as V;

    let result = match kind {
        FieldKind::String => V::String(Some(Box::new(s.to_owned()))),
        FieldKind::I64 => {
            let i = s
                .parse::<i64>()
                .map_err(|_| "invalid i64 in cursor".to_owned())?;
            V::BigInt(Some(i))
        }
        FieldKind::F64 => {
            let f = s
                .parse::<f64>()
                .map_err(|_| "invalid f64 in cursor".to_owned())?;
            V::Double(Some(f))
        }
        FieldKind::Bool => {
            let b = s
                .parse::<bool>()
                .map_err(|_| "invalid bool in cursor".to_owned())?;
            V::Bool(Some(b))
        }
        FieldKind::Uuid => {
            let u = s
                .parse::<uuid::Uuid>()
                .map_err(|_| "invalid uuid in cursor".to_owned())?;
            V::Uuid(Some(Box::new(u)))
        }
        FieldKind::DateTimeUtc => {
            if let Ok(dt) =
                time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
            {
                V::TimeDateTimeWithTimeZone(Some(Box::new(dt)))
            } else {
                let dt = chrono::DateTime::parse_from_rfc3339(s)
                    .map_err(|_| "invalid datetime in cursor".to_owned())?
                    .with_timezone(&chrono::Utc);
                V::ChronoDateTimeUtc(Some(Box::new(dt)))
            }
        }
        FieldKind::Date => {
            let d = s
                .parse::<chrono::NaiveDate>()
                .map_err(|_| "invalid date in cursor".to_owned())?;
            V::ChronoDate(Some(Box::new(d)))
        }
        FieldKind::Time => {
            let t = s
                .parse::<chrono::NaiveTime>()
                .map_err(|_| "invalid time in cursor".to_owned())?;
            V::ChronoTime(Some(Box::new(t)))
        }
        FieldKind::Decimal => {
            let d = s
                .parse::<rust_decimal::Decimal>()
                .map_err(|_| "invalid decimal in cursor".to_owned())?;
            V::Decimal(Some(Box::new(d)))
        }
    };

    Ok(result)
}

/// Pagination limit configuration
#[derive(Clone, Copy)]
pub struct LimitCfg {
    pub default: u64,
    pub max: u64,
}

/// Clamp the requested limit to configured bounds
fn clamp_limit(req: Option<u64>, cfg: LimitCfg) -> u64 {
    let mut l = req.unwrap_or(cfg.default);
    if l == 0 {
        l = 1;
    }
    if l > cfg.max {
        l = cfg.max;
    }
    l
}

/// Type-safe `OData` pagination with filters, ordering, and cursors.
///
/// This function provides complete cursor-based pagination using the type-safe
/// `FilterField` and `ODataFieldMapping` traits. It replaces the legacy FieldMap-based
/// pagination.
///
/// # Type Parameters
///
/// - `F`: `FilterField` implementation (generated by `#[derive(ODataFilterable)]`)
/// - `M`: `ODataFieldMapping` implementation that maps F to `SeaORM` columns and extracts cursor values
/// - `E`: `SeaORM` Entity type
/// - `D`: Domain/DTO type to map models to
/// - `Mapper`: Function to map `Entity::Model` to D
/// - `C`: Database connection type
///
/// # Arguments
///
/// - `select`: Base `SeaORM` select query (already security-scoped if needed)
/// - `conn`: Database connection
/// - `query`: `OData` query with filter, order, cursor, and limit
/// - `tiebreaker`: Default orderby field and direction for stable pagination
/// - `limit_cfg`: Default and maximum page sizes
/// - `model_to_domain`: Function to convert entity models to domain types
///
/// # Returns
///
/// A Page containing the results and pagination metadata (next/prev cursors)
///
/// # Example
///
/// ```ignore
/// let page = paginate_odata::<UserDtoFilterField, UserODataMapper, _, _, _, _>(
///     base_query,
///     db,
///     &odata_query,
///     ("id", SortDir::Desc),
///     LimitCfg { default: 25, max: 1000 },
///     |model| model.into(),
/// ).await?;
/// ```
///
/// # Errors
/// Returns `ODataError` if filter application, cursor validation, or database query fails.
pub async fn paginate_odata<F, M, E, D, Mapper, C>(
    select: SecureSelect<E, Scoped>,
    conn: &C,
    query: &toolkit_odata::ODataQuery,
    tiebreaker: (&str, SortDir),
    limit_cfg: LimitCfg,
    model_to_domain: Mapper,
) -> Result<Page<D>, ODataError>
where
    F: FilterField,
    M: ODataFieldMapping<F, Entity = E>,
    E: EntityTrait,
    Mapper: Fn(E::Model) -> D,
    C: DBRunner,
{
    let (rows, page_info) =
        paginate_odata_collect::<F, M, E, C>(select, conn, query, tiebreaker, limit_cfg).await?;
    let items = rows.into_iter().map(model_to_domain).collect();
    Ok(Page { items, page_info })
}

/// Sibling of [`paginate_odata`] for callers whose `model -> domain`
/// mapping is **fallible** — e.g. when a domain enum is reconstructed
/// from a `SMALLINT` column whose values are pinned by a DDL `CHECK`
/// constraint but could in theory drift on legacy / manually-repaired
/// databases. Instead of panicking in the mapper, the error is
/// propagated through the typed [`PaginateOdataTryError`] wrapper so
/// the caller decides whether the row-shape drift surfaces as 500
/// (`Internal`) or some other category.
///
/// Existing infallible callers should keep using [`paginate_odata`];
/// this variant is additive and the two share the internal
/// pagination machinery (`paginate_odata_collect`) so cursor / filter
/// / ordering semantics stay identical.
///
/// # Errors
/// * [`PaginateOdataTryError::OData`] — same set of failures
///   [`paginate_odata`] surfaces (filter / cursor / DB).
/// * [`PaginateOdataTryError::MapError`] — first mapper failure
///   short-circuits the page assembly with the caller's domain error
///   verbatim.
pub async fn paginate_odata_try<F, M, E, D, Mapper, MapErr, C>(
    select: SecureSelect<E, Scoped>,
    conn: &C,
    query: &toolkit_odata::ODataQuery,
    tiebreaker: (&str, SortDir),
    limit_cfg: LimitCfg,
    model_to_domain: Mapper,
) -> Result<Page<D>, PaginateOdataTryError<MapErr>>
where
    F: FilterField,
    M: ODataFieldMapping<F, Entity = E>,
    E: EntityTrait,
    Mapper: Fn(E::Model) -> Result<D, MapErr>,
    C: DBRunner,
{
    let (rows, page_info) =
        paginate_odata_collect::<F, M, E, C>(select, conn, query, tiebreaker, limit_cfg)
            .await
            .map_err(PaginateOdataTryError::OData)?;
    let items = rows
        .into_iter()
        .map(model_to_domain)
        .collect::<Result<Vec<_>, _>>()
        .map_err(PaginateOdataTryError::MapError)?;
    Ok(Page { items, page_info })
}

/// Error envelope for [`paginate_odata_try`]. Carries the
/// caller's `MapErr` verbatim so the AM domain-error mapping
/// (`DomainError::Internal { diagnostic, cause }`) is preserved
/// across the helper boundary.
#[derive(thiserror::Error, Debug)]
pub enum PaginateOdataTryError<MapErr> {
    /// `OData` / filter / cursor / DB failure — identical category set to
    /// what plain [`paginate_odata`] would surface.
    #[error("OData pagination failed: {0}")]
    OData(#[from] ODataError),
    /// Caller's `model -> domain` mapping failed on a row. The first
    /// failure short-circuits page assembly.
    #[error("model→domain mapping failed: {0}")]
    MapError(MapErr),
}

/// Internal page-collector shared by [`paginate_odata`] and
/// [`paginate_odata_try`]. Returns the raw `Vec<E::Model>` rows plus
/// the assembled [`PageInfo`] (cursor tokens + clamped limit) so the
/// public wrappers can apply either an infallible or a fallible
/// `model -> domain` projection on top.
async fn paginate_odata_collect<F, M, E, C>(
    select: SecureSelect<E, Scoped>,
    conn: &C,
    query: &toolkit_odata::ODataQuery,
    tiebreaker: (&str, SortDir),
    limit_cfg: LimitCfg,
) -> Result<(Vec<E::Model>, PageInfo), ODataError>
where
    F: FilterField,
    M: ODataFieldMapping<F, Entity = E>,
    E: EntityTrait,
    C: DBRunner,
{
    let limit = clamp_limit(query.limit, limit_cfg);
    let fetch = limit + 1;

    // Effective order derivation
    let effective_order = if let Some(cur) = &query.cursor {
        ODataOrderBy::from_signed_tokens(&cur.s).map_err(|_| ODataError::InvalidCursor)?
    } else {
        query
            .order
            .clone()
            .ensure_tiebreaker(tiebreaker.0, tiebreaker.1)
    };

    // Reject `$orderby` keys against fields the mapper has flagged as
    // non-orderable (typically: filter columns whose wire shape
    // differs from their storage shape — e.g. a SMALLINT-backed enum
    // exposed as a string on the API surface). Without this guard the
    // cursor codec would round-trip through `field.kind()` and either
    // misinterpret the encoded token or hand `SeaORM` a value with
    // the wrong SQL type. Tiebreakers go through the same gate so a
    // caller cannot smuggle a non-orderable column in via
    // `ensure_tiebreaker`.
    for order_key in &effective_order.0 {
        let field = F::from_name(&order_key.field)
            .ok_or_else(|| ODataError::InvalidOrderByField(order_key.field.clone()))?;
        if !M::is_orderable(field) {
            return Err(ODataError::InvalidOrderByField(order_key.field.clone()));
        }
    }

    // Validate cursor consistency (filter hash only)
    if let Some(cur) = &query.cursor
        && let (Some(h), Some(cf)) = (query.filter_hash.as_deref(), cur.f.as_deref())
        && h != cf
    {
        return Err(ODataError::FilterMismatch);
    }

    let mut s = select.inner;

    // Apply filter using type-safe FilterNode
    if let Some(ast) = query.filter.as_deref() {
        let filter_node = convert_expr_to_filter_node::<F>(ast)
            .map_err(|e| ODataError::InvalidFilter(e.to_string()))?;

        s = s.filter(
            filter_node_to_condition::<F, M>(&filter_node).map_err(ODataError::InvalidFilter)?,
        );
    }

    let is_backward = query.cursor.as_ref().is_some_and(|c| c.d == "bwd");

    // Apply cursor predicate
    if let Some(cursor) = &query.cursor {
        let cursor_cond = build_cursor_predicate::<F, M>(cursor, &effective_order)?;
        s = s.filter(cursor_cond);
    }

    // Apply ordering
    let query_order = if is_backward {
        effective_order.clone().reverse_directions()
    } else {
        effective_order.clone()
    };

    for order_key in &query_order.0 {
        let field = F::from_name(&order_key.field)
            .ok_or_else(|| ODataError::InvalidOrderByField(order_key.field.clone()))?;
        let column = M::map_field(field);
        let sea_order = match order_key.dir {
            SortDir::Asc => Order::Asc,
            SortDir::Desc => Order::Desc,
        };
        s = s.order_by(column, sea_order);
    }

    s = s.limit(fetch);

    #[allow(clippy::disallowed_methods)]
    let mut rows = match DBRunnerInternal::as_seaorm(conn) {
        SeaOrmRunner::Conn(db) => s.all(db).await,
        SeaOrmRunner::Tx(tx) => s.all(tx).await,
    }
    .map_err(|e| ODataError::Db(e.to_string()))?;

    let has_more = (rows.len() as u64) > limit;

    // Handle backward pagination reversal
    if is_backward {
        if has_more {
            rows.pop();
        }
        rows.reverse();
    } else if has_more {
        rows.truncate(usize::try_from(limit).unwrap_or(usize::MAX));
    }

    // Build cursors
    let next_cursor = if is_backward || has_more {
        build_cursor_from_rows::<E, F, M>(
            &rows,
            &effective_order,
            query.filter_hash.as_deref(),
            "fwd",
            true,
        )?
    } else {
        None
    };

    let prev_cursor = if is_backward {
        if has_more {
            build_cursor_from_rows::<E, F, M>(
                &rows,
                &effective_order,
                query.filter_hash.as_deref(),
                "bwd",
                false,
            )?
        } else {
            None
        }
    } else if query.cursor.is_some() {
        build_cursor_from_rows::<E, F, M>(
            &rows,
            &effective_order,
            query.filter_hash.as_deref(),
            "bwd",
            false,
        )?
    } else {
        None
    };

    Ok((
        rows,
        PageInfo {
            next_cursor,
            prev_cursor,
            limit,
        },
    ))
}

/// Build a cursor from rows, using either the first or last row
fn build_cursor_from_rows<E, F, M: ODataFieldMapping<F, Entity = E>>(
    rows: &[<E as EntityTrait>::Model],
    effective_order: &ODataOrderBy,
    filter_hash: Option<&str>,
    direction: &str,
    use_last: bool,
) -> Result<Option<String>, ODataError>
where
    F: FilterField,
    E: EntityTrait,
{
    let row = if use_last { rows.last() } else { rows.first() };

    row.map(|m| build_cursor_from_model::<F, M>(m, effective_order, filter_hash, direction))
        .transpose()
        .and_then(|opt| match opt {
            Some(c) => c.encode().map(Some).map_err(|_| ODataError::InvalidCursor),
            None => Ok(None),
        })
}

/// Build a cursor predicate for pagination
fn build_cursor_predicate<F, M>(
    cursor: &CursorV1,
    order: &ODataOrderBy,
) -> Result<Condition, ODataError>
where
    F: FilterField,
    M: ODataFieldMapping<F>,
{
    if cursor.k.len() != order.0.len() {
        return Err(ODataError::InvalidCursor);
    }

    // Parse all cursor values first
    let mut cursor_values: Vec<(F, M::Column, sea_orm::Value, SortDir)> = Vec::new();
    for (i, key_str) in cursor.k.iter().enumerate() {
        let order_key = &order.0[i];
        let field = F::from_name(&order_key.field)
            .ok_or(ODataError::InvalidOrderByField(order_key.field.clone()))?;
        let column = M::map_field(field);
        let kind = field.kind();
        let value = parse_cursor_value(kind, key_str).map_err(|_| ODataError::InvalidCursor)?;
        cursor_values.push((field, column, value, order_key.dir));
    }

    let is_backward = cursor.d == "bwd";
    let mut main_condition = Condition::any();

    for i in 0..cursor_values.len() {
        let mut prefix_condition = Condition::all();

        // Add equality conditions for all previous fields
        for (_field, column, value, _dir) in cursor_values.iter().take(i) {
            prefix_condition = prefix_condition.add(Expr::col(*column).eq(value.clone()));
        }

        // Add comparison for current field
        let (_field, column, value, dir) = &cursor_values[i];
        let comparison = if is_backward {
            // Backward: reverse the comparison
            match dir {
                SortDir::Asc => Expr::col(*column).lt(value.clone()),
                SortDir::Desc => Expr::col(*column).gt(value.clone()),
            }
        } else {
            // Forward: normal comparison
            match dir {
                SortDir::Asc => Expr::col(*column).gt(value.clone()),
                SortDir::Desc => Expr::col(*column).lt(value.clone()),
            }
        };

        prefix_condition = prefix_condition.add(comparison);
        main_condition = main_condition.add(prefix_condition);
    }

    Ok(main_condition)
}

/// Build a cursor from an entity model
fn build_cursor_from_model<F, M>(
    model: &<M::Entity as EntityTrait>::Model,
    order: &ODataOrderBy,
    filter_hash: Option<&str>,
    direction: &str,
) -> Result<CursorV1, ODataError>
where
    F: FilterField,
    M: ODataFieldMapping<F>,
{
    let field_values =
        M::extract_cursor_values(model, order).map_err(|_e| ODataError::InvalidCursor)?;

    let mut cursor_keys = Vec::new();
    for (field, value) in field_values {
        let kind = field.kind();
        let key_str = encode_cursor_value(&value, kind).map_err(|_| ODataError::InvalidCursor)?;
        cursor_keys.push(key_str);
    }

    // Determine primary sort direction from first order key
    let primary_dir = order.0.first().map_or(SortDir::Desc, |k| k.dir);

    Ok(CursorV1 {
        k: cursor_keys,
        o: primary_dir,
        s: order.to_signed_tokens(),
        f: filter_hash.map(str::to_owned),
        d: direction.to_owned(),
    })
}
