fraiseql-core 2.19.1

//! Aggregate Query Parser
//!
//! Parses GraphQL aggregate queries into `AggregationRequest` for execution.
//!
//! # GraphQL Query Format
//!
//! ```graphql
//! query {
//!   sales_aggregate(
//!     where: { customer_id: { _eq: "uuid-123" } }
//!     groupBy: { category: true, occurred_at_day: true }
//!     having: { revenue_sum_gt: 1000 }
//!     orderBy: { revenue_sum: DESC }
//!     limit: 10
//!   ) {
//!     category
//!     occurred_at_day
//!     count
//!     revenue_sum
//!     revenue_avg
//!   }
//! }
//! ```
//!
//! # Parsed Result
//!
//! ```text
//! // Illustrative output structure only — not executable code.
//! AggregationRequest {
//!     table_name: "tf_sales",
//!     where_clause: Some(...),
//!     group_by: vec![
//!         GroupBySelection::Dimension { path: "category", alias: "category" },
//!         GroupBySelection::TemporalBucket { column: "occurred_at", bucket: Day, alias: "occurred_at_day" },
//!     ],
//!     aggregates: vec![
//!         AggregateSelection::Count { alias: "count" },
//!         AggregateSelection::MeasureAggregate { measure: "revenue", function: Sum, alias: "revenue_sum" },
//!         AggregateSelection::MeasureAggregate { measure: "revenue", function: Avg, alias: "revenue_avg" },
//!     ],
//!     having: vec![...],
//!     order_by: vec![...],
//!     limit: Some(10),
//!     offset: None,
//! }
//! ```

use serde_json::Value;

use crate::{
    backend::where_clause::{WhereClause, WhereOperator},
    compiler::{
        aggregate_types::{AggregateFunction, HavingOperator, TemporalBucket},
        aggregation::{
            AggregateSelection, AggregationRequest, GroupBySelection, HavingCondition,
            OrderByClause, OrderDirection, TreeGrouping,
        },
        fact_table::FactTableMetadata,
    },
    error::{FraiseQLError, Result},
};

/// The refusal of a `table` argument, inline or as a variable: an aggregate or window root
/// names its fact table, so the argument is no longer read (#1800).
pub fn table_argument_refusal(table_name: &str) -> FraiseQLError {
    FraiseQLError::Validation {
        message: format!(
            "`table` is no longer an argument: the root reads fact table `{table_name}`; \
             remove it"
        ),
        path:    Some("table".to_string()),
    }
}

/// Aggregate query parser
pub struct AggregateQueryParser;

impl AggregateQueryParser {
    /// Parse a simplified aggregate query into `AggregationRequest`.
    ///
    /// For we'll accept a JSON structure that represents the query:
    /// ```json
    /// {
    ///   "groupBy": {
    ///     "category": true,
    ///     "occurred_at_day": true
    ///   },
    ///   "aggregates": [
    ///     {"count": true},
    ///     {"revenue_sum": true}
    ///   ],
    ///   "having": {
    ///     "revenue_sum_gt": 1000
    ///   },
    ///   "orderBy": {
    ///     "revenue_sum": "DESC"
    ///   },
    ///   "limit": 10
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns error if the query structure is invalid or references non-existent measures.
    ///
    /// The `native_columns` map (column name → PostgreSQL cast string) controls whether
    /// WHERE filters and GROUP BY fields are emitted as direct column references
    /// (`WhereClause::NativeField`, `GroupBySelection::NativeDimension`) rather than
    /// JSONB extraction variants.  Pass an empty map to get the old behaviour.
    pub fn parse(
        query_json: &Value,
        metadata: &FactTableMetadata,
        native_columns: &std::collections::HashMap<String, String>,
    ) -> Result<AggregationRequest> {
        // The root names its fact table: a `table` key is refused, not checked.
        if query_json.get("table").is_some() {
            return Err(table_argument_refusal(&metadata.table_name));
        }
        let table_name = metadata.table_name.clone();

        // Parse WHERE clause (if present)
        let where_clause = if let Some(where_obj) = query_json.get("where") {
            Some(Self::parse_where_clause(where_obj, metadata, native_columns)?)
        } else {
            None
        };

        // Parse GROUP BY selections
        let group_by = if let Some(group_by_obj) = query_json.get("groupBy") {
            Self::parse_group_by(group_by_obj, metadata, native_columns)?
        } else {
            vec![]
        };

        // Parse aggregate selections from requested fields
        let aggregates = if let Some(agg_array) = query_json.get("aggregates") {
            Self::parse_aggregates(agg_array, metadata)?
        } else {
            vec![]
        };

        // Parse HAVING conditions
        let having = if let Some(having_obj) = query_json.get("having") {
            Self::parse_having(having_obj, &aggregates, metadata)?
        } else {
            vec![]
        };

        // Parse ORDER BY clauses
        let order_by = if let Some(order_obj) = query_json.get("orderBy") {
            Self::parse_order_by(order_obj)?
        } else {
            vec![]
        };

        // Parse LIMIT
        // #1532: read as a list's are. A value that is not a non-negative Int is refused;
        // `as_u64()` read `-1`, `1.5` and `"10"` as no limit and saturated `2^32`.
        let limit =
            crate::runtime::executor::coerce_pagination_arg("limit", query_json.get("limit"))?;

        // Parse OFFSET
        let offset =
            crate::runtime::executor::coerce_pagination_arg("offset", query_json.get("offset"))?;

        Ok(AggregationRequest {
            table_name,
            where_clause,
            group_by,
            aggregates,
            having,
            order_by,
            limit,
            offset,
        })
    }

    /// Parse WHERE clause from JSON.
    ///
    /// For aggregate queries, WHERE works on denormalized filter columns, on dimensions a
    /// `native_dimension_mapping` maps to a column (read from that column), and on
    /// top-level JSONB dimension keys.
    /// Expected format: `{ "field_operator": value }`
    /// Example: `{ "customer_id_eq": "123", "occurred_at_gte": "2024-01-01" }`
    ///
    /// When a field name matches an entry in `native_columns`, the clause is emitted
    /// as [`WhereClause::NativeField`] (direct column reference) rather than
    /// [`WhereClause::Field`] (JSONB extraction).
    fn parse_where_clause(
        where_obj: &Value,
        metadata: &FactTableMetadata,
        native_columns: &std::collections::HashMap<String, String>,
    ) -> Result<WhereClause> {
        let Some(obj) = where_obj.as_object() else {
            return Ok(WhereClause::And(vec![]));
        };

        let mut conditions = Vec::new();

        for (key, value) in obj {
            conditions.push(where_condition(key, value, metadata, native_columns)?);
        }

        Ok(WhereClause::And(conditions))
    }

    /// Parse GROUP BY selections.
    ///
    /// Each key takes the form the root's published `groupBy` type gives it (#1800):
    /// - `true` groups by a dimension, a filter column or a native mapping (`category`), by a
    ///   temporal bucket (`occurred_at_day`), or by an ltree column's depth (`org_path_depth`);
    /// - a positive `Int` rolls an ltree column up to that level (`org_path_level: 2`);
    /// - `false` does not group by the key.
    ///
    /// Any other value is refused, naming the key that carries it.
    ///
    /// When a field name matches an entry in `native_columns`, the selection is emitted
    /// as [`GroupBySelection::NativeDimension`] rather than [`GroupBySelection::Dimension`].
    fn parse_group_by(
        group_by_obj: &Value,
        metadata: &FactTableMetadata,
        native_columns: &std::collections::HashMap<String, String>,
    ) -> Result<Vec<GroupBySelection>> {
        let mut selections = Vec::new();

        let Some(obj) = group_by_obj.as_object() else {
            return Ok(selections);
        };
        for (key, value) in obj {
            if let Some(column) = Self::tree_column(key, "_level", metadata) {
                let level = value
                    .as_u64()
                    .filter(|n| *n >= 1)
                    .and_then(|n| u32::try_from(n).ok())
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: format!("groupBy `{key}` takes a positive Int level, got {value}"),
                        path:    None,
                    })?;
                selections.push(Self::tree_level(column, TreeGrouping::Level(level), metadata)?);
                continue;
            }
            match value {
                Value::Bool(false) => {},
                Value::Bool(true) => {
                    selections.push(Self::group_by_key(key, metadata, native_columns)?);
                },
                other => return Err(Self::unpublished_group_by_form(key, other)),
            }
        }

        Ok(selections)
    }

    /// The selection a `groupBy` key set to `true` names.
    fn group_by_key(
        key: &str,
        metadata: &FactTableMetadata,
        native_columns: &std::collections::HashMap<String, String>,
    ) -> Result<GroupBySelection> {
        if let Some(column) = Self::tree_column(key, "_depth", metadata) {
            return Self::tree_level(column, TreeGrouping::Depth, metadata);
        }
        // Priority 1: a calendar dimension (highest performance)
        if let Some(calendar_sel) = Self::try_parse_calendar_bucket(key, metadata)? {
            return Ok(calendar_sel);
        }
        // Priority 2: DATE_TRUNC when no calendar dimension serves the bucket
        if let Some(bucket_sel) = Self::parse_temporal_bucket(key, metadata)? {
            return Ok(bucket_sel);
        }
        // Priority 3: a deep JSONB path mapped to a native column
        if let Some(mapped_col) = mapped_column(metadata, key) {
            return Ok(GroupBySelection::NativeDimension {
                column:  mapped_col.clone(),
                pg_cast: native_columns.get(mapped_col.as_str()).cloned().unwrap_or_default(),
                alias:   key.to_string(),
            });
        }
        // Priority 4: a native SQL column (filter-derived), referenced directly
        if let Some(pg_cast) = native_columns.get(key) {
            return Ok(GroupBySelection::NativeDimension {
                column:  key.to_string(),
                pg_cast: pg_cast.clone(),
                alias:   key.to_string(),
            });
        }
        // Priority 5: a regular JSONB dimension. Unlike priorities 1-4 (each gated by a
        // lookup against server-defined calendar/temporal/native metadata), this fallback
        // echoes the raw GraphQL key as the SELECT alias (quoted since #1516; it was once
        // interpolated bare). Reject keys that are not safe identifiers so an attacker
        // cannot inject an extra SELECT column (H1) — e.g. `a, (SELECT …) AS leak`. This
        // fires at parse time, independent of the compiler's dimension allowlist (which is
        // skipped when `dimensions.paths` is empty).
        validate_dimension_key(key)?;
        Ok(GroupBySelection::Dimension {
            // The `path` is the JSONB extraction key → recase to snake_case so a camelCase
            // dimension (`machineStatus`) groups by `data->>'machine_status'` (#486). The
            // `alias` is the result/response column name, consumed verbatim by
            // `AggregationProjector`, so it keeps the camel surface name (the #418/#410
            // "output key is the alias" rule).
            path:  crate::utils::to_snake_case(key),
            alias: key.to_string(),
        })
    }

    /// The denormalized filter column `key` names with `suffix` (`org_path_level` →
    /// `org_path`), when there is one.
    fn tree_column<'k>(
        key: &'k str,
        suffix: &str,
        metadata: &FactTableMetadata,
    ) -> Option<&'k str> {
        key.strip_suffix(suffix)
            .filter(|column| metadata.denormalized_filters.iter().any(|f| f.name == *column))
    }

    /// The refusal of a `groupBy` value the published type cannot carry, naming the key that
    /// carries it where there is one.
    fn unpublished_group_by_form(key: &str, value: &Value) -> FraiseQLError {
        let replacement = match value {
            Value::String(form) if form == "depth" => Some(format!("`{key}_depth: true`")),
            Value::String(form) => TemporalBucket::from_str(form)
                .ok()
                .map(|bucket| format!("`{key}_{}: true`", bucket.postgres_arg())),
            Value::Object(form) => form.get("level").map(|level| format!("`{key}_level: {level}`")),
            _ => None,
        };
        let message = replacement.map_or_else(
            || format!("groupBy `{key}` takes `true` (grouped) or `false`, got {value}"),
            |replacement| {
                format!("groupBy `{key}: {value}` is no longer accepted: send {replacement}")
            },
        );
        FraiseQLError::Validation {
            message,
            path: None,
        }
    }

    /// A tree grouping of the denormalized filter column `key` (#1498). The column's
    /// type is checked when the request is planned.
    fn tree_level(
        key: &str,
        grouping: TreeGrouping,
        metadata: &FactTableMetadata,
    ) -> Result<GroupBySelection> {
        if !metadata.denormalized_filters.iter().any(|f| f.name == key) {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "groupBy '{key}': a tree level or depth needs an ltree denormalized filter \
                     column, and '{key}' is not a denormalized filter of '{}'",
                    metadata.table_name
                ),
                path:    None,
            });
        }
        let alias = match grouping {
            TreeGrouping::Depth => format!("{key}_depth"),
            TreeGrouping::Level(_) => format!("{key}_level"),
        };
        Ok(GroupBySelection::TreeLevel {
            column: key.to_string(),
            grouping,
            alias,
        })
    }

    /// Parse temporal bucket if the key matches pattern
    fn parse_temporal_bucket(
        key: &str,
        metadata: &FactTableMetadata,
    ) -> Result<Option<GroupBySelection>> {
        // Check for temporal bucket patterns: column_day, column_week, etc.
        for filter_col in &metadata.denormalized_filters {
            for bucket in &[
                ("_second", TemporalBucket::Second),
                ("_minute", TemporalBucket::Minute),
                ("_hour", TemporalBucket::Hour),
                ("_day", TemporalBucket::Day),
                ("_week", TemporalBucket::Week),
                ("_month", TemporalBucket::Month),
                ("_quarter", TemporalBucket::Quarter),
                ("_year", TemporalBucket::Year),
            ] {
                let expected_key = format!("{}{}", filter_col.name, bucket.0);
                if key == expected_key {
                    return Ok(Some(GroupBySelection::TemporalBucket {
                        column: filter_col.name.clone(),
                        bucket: bucket.1,
                        alias:  key.to_string(),
                    }));
                }
            }
        }

        Ok(None)
    }

    /// Try to parse calendar dimension from key pattern (e.g., "`occurred_at_day`")
    ///
    /// Checks if the key matches a calendar dimension pattern and returns
    /// a `CalendarDimension` selection if available, otherwise None.
    fn try_parse_calendar_bucket(
        key: &str,
        metadata: &FactTableMetadata,
    ) -> Result<Option<GroupBySelection>> {
        for calendar_dim in &metadata.calendar_dimensions {
            // Check all temporal bucket suffixes
            for (suffix, bucket_type) in &[
                ("_second", TemporalBucket::Second),
                ("_minute", TemporalBucket::Minute),
                ("_hour", TemporalBucket::Hour),
                ("_day", TemporalBucket::Day),
                ("_week", TemporalBucket::Week),
                ("_month", TemporalBucket::Month),
                ("_quarter", TemporalBucket::Quarter),
                ("_year", TemporalBucket::Year),
            ] {
                let expected_key = format!("{}{}", calendar_dim.source_column, suffix);
                if key == expected_key {
                    // Find matching calendar bucket
                    if let Some((gran, bucket)) =
                        Self::find_calendar_bucket(calendar_dim, *bucket_type)
                    {
                        return Ok(Some(GroupBySelection::CalendarDimension {
                            source_column:   calendar_dim.source_column.clone(),
                            calendar_column: gran.column_name.clone(),
                            json_key:        bucket.json_key.clone(),
                            bucket:          bucket.bucket_type,
                            alias:           key.to_string(),
                        }));
                    }
                }
            }
        }
        Ok(None)
    }

    /// Find calendar bucket in available granularities
    ///
    /// Searches through calendar dimension granularities to find a matching bucket type.
    /// Returns the granularity and bucket if found.
    fn find_calendar_bucket(
        calendar_dim: &crate::compiler::fact_table::CalendarDimension,
        bucket: TemporalBucket,
    ) -> Option<(
        &crate::compiler::fact_table::CalendarGranularity,
        &crate::compiler::fact_table::CalendarBucket,
    )> {
        for granularity in &calendar_dim.granularities {
            for cal_bucket in &granularity.buckets {
                if cal_bucket.bucket_type == bucket {
                    return Some((granularity, cal_bucket));
                }
            }
        }
        None
    }

    /// Parse aggregate selections
    fn parse_aggregates(
        agg_array: &Value,
        metadata: &FactTableMetadata,
    ) -> Result<Vec<AggregateSelection>> {
        let mut aggregates = Vec::new();

        if let Some(arr) = agg_array.as_array() {
            for item in arr {
                if let Some(obj) = item.as_object() {
                    // Each object should have one key (the aggregate name), set to the
                    // published `Boolean`: `true` computes it, `false` does not.
                    for (agg_name, value) in obj {
                        match value {
                            Value::Bool(true) => aggregates
                                .push(Self::parse_aggregate_selection(agg_name, metadata)?),
                            Value::Bool(false) => {},
                            other => {
                                return Err(FraiseQLError::Validation {
                                    message: format!(
                                        "aggregates item `{agg_name}` takes `true` (computed) \
                                         or `false`, got {other}"
                                    ),
                                    path:    None,
                                });
                            },
                        }
                    }
                }
            }
        }

        Ok(aggregates)
    }

    /// Parse a single aggregate selection
    fn parse_aggregate_selection(
        agg_name: &str,
        metadata: &FactTableMetadata,
    ) -> Result<AggregateSelection> {
        // Handle COUNT
        if agg_name == "count" {
            return Ok(AggregateSelection::Count {
                alias: "count".to_string(),
            });
        }

        // Handle COUNT_DISTINCT: supports both "count_distinct" (defaults to first dimension)
        // and "field_count_distinct" pattern (e.g., "product_id_count_distinct")
        if agg_name == "count_distinct" {
            // Default to first dimension path, or "id" if none available
            let default_field = Self::extract_dimension_paths(metadata)
                .first()
                .cloned()
                .unwrap_or_else(|| "id".to_string());
            return Ok(AggregateSelection::CountDistinct {
                field: default_field,
                alias: "count_distinct".to_string(),
            });
        }

        // Handle field_count_distinct pattern (e.g., "customer_id_count_distinct")
        if let Some(stripped) = agg_name.strip_suffix("_count_distinct") {
            // Check if the stripped part matches a dimension path
            let dimension_paths = Self::extract_dimension_paths(metadata);
            if dimension_paths.iter().any(|p| p == stripped) {
                return Ok(AggregateSelection::CountDistinct {
                    field: stripped.to_string(),
                    alias: agg_name.to_string(),
                });
            }
            // Also allow count distinct on measures
            if metadata.measures.iter().any(|m| m.name == stripped) {
                return Ok(AggregateSelection::CountDistinct {
                    field: stripped.to_string(),
                    alias: agg_name.to_string(),
                });
            }
            // If no match found, return error with helpful message
            return Err(FraiseQLError::Validation {
                message: format!(
                    "COUNT DISTINCT field '{}' not found in dimensions or measures. Available: {:?}",
                    stripped, dimension_paths
                ),
                path:    None,
            });
        }

        // Handle boolean aggregates (BOOL_AND, BOOL_OR)
        // e.g., "is_active_bool_and", "has_discount_bool_or"
        for dimension_path in Self::extract_dimension_paths(metadata) {
            if let Some(stripped) = agg_name.strip_suffix("_bool_and") {
                if stripped == dimension_path {
                    return Ok(AggregateSelection::BoolAggregate {
                        field:    dimension_path,
                        function: crate::compiler::aggregate_types::BoolAggregateFunction::And,
                        alias:    agg_name.to_string(),
                    });
                }
            }
            if let Some(stripped) = agg_name.strip_suffix("_bool_or") {
                if stripped == dimension_path {
                    return Ok(AggregateSelection::BoolAggregate {
                        field:    dimension_path,
                        function: crate::compiler::aggregate_types::BoolAggregateFunction::Or,
                        alias:    agg_name.to_string(),
                    });
                }
            }
        }

        // Handle measure aggregates: revenue_sum, revenue_avg, etc.
        for measure in &metadata.measures {
            for func in &[
                ("_sum", AggregateFunction::Sum),
                ("_avg", AggregateFunction::Avg),
                ("_min", AggregateFunction::Min),
                ("_max", AggregateFunction::Max),
                ("_stddev", AggregateFunction::Stddev),
                ("_variance", AggregateFunction::Variance),
                // Advanced aggregates
                ("_array_agg", AggregateFunction::ArrayAgg),
                ("_json_agg", AggregateFunction::JsonAgg),
                ("_jsonb_agg", AggregateFunction::JsonbAgg),
                ("_string_agg", AggregateFunction::StringAgg),
            ] {
                let expected_name = format!("{}{}", measure.name, func.0);
                if agg_name == expected_name {
                    return Ok(AggregateSelection::MeasureAggregate {
                        measure:  measure.name.clone(),
                        function: func.1,
                        alias:    agg_name.to_string(),
                    });
                }
            }
        }

        // Handle native measure aggregates: measures.volume_sum, measures.cost_avg, etc.
        // These are JSONB measure paths mapped to flat SQL columns via native_measures.
        for jsonb_path in metadata.native_measures.keys() {
            for func in &[
                ("_sum", AggregateFunction::Sum),
                ("_avg", AggregateFunction::Avg),
                ("_min", AggregateFunction::Min),
                ("_max", AggregateFunction::Max),
                ("_stddev", AggregateFunction::Stddev),
                ("_variance", AggregateFunction::Variance),
                ("_array_agg", AggregateFunction::ArrayAgg),
                ("_json_agg", AggregateFunction::JsonAgg),
                ("_jsonb_agg", AggregateFunction::JsonbAgg),
                ("_string_agg", AggregateFunction::StringAgg),
            ] {
                let expected_name = format!("{}{}", jsonb_path, func.0);
                if agg_name == expected_name {
                    // Store the JSONB path as the measure name; the planner resolves
                    // it to the native column via FactTableMetadata.native_measures.
                    return Ok(AggregateSelection::MeasureAggregate {
                        measure:  jsonb_path.clone(),
                        function: func.1,
                        alias:    agg_name.to_string(),
                    });
                }
            }
        }

        // Check for dimension-level advanced aggregates
        // e.g., "product_id_array_agg", "product_name_string_agg"
        for dimension_path in Self::extract_dimension_paths(metadata) {
            for func in &[
                ("_array_agg", AggregateFunction::ArrayAgg),
                ("_json_agg", AggregateFunction::JsonAgg),
                ("_jsonb_agg", AggregateFunction::JsonbAgg),
                ("_string_agg", AggregateFunction::StringAgg),
            ] {
                let expected_name = format!("{}{}", dimension_path, func.0);
                if agg_name == expected_name {
                    // For dimension aggregates, store the path as the "measure"
                    return Ok(AggregateSelection::MeasureAggregate {
                        measure:  dimension_path,
                        function: func.1,
                        alias:    agg_name.to_string(),
                    });
                }
            }
        }

        Err(FraiseQLError::Validation {
            message: format!("Unknown aggregate selection: {agg_name}"),
            path:    None,
        })
    }

    /// Extract dimension paths from metadata for advanced aggregate parsing
    fn extract_dimension_paths(metadata: &FactTableMetadata) -> Vec<String> {
        let mut paths = Vec::new();

        // Add dimension paths from JSONB column
        for dim_path in &metadata.dimensions.paths {
            paths.push(dim_path.name.clone());
        }

        // Add denormalized filter columns (these can also be aggregated)
        for filter in &metadata.denormalized_filters {
            paths.push(filter.name.clone());
        }

        paths
    }

    /// Parse HAVING conditions
    fn parse_having(
        having_obj: &Value,
        aggregates: &[AggregateSelection],
        _metadata: &FactTableMetadata,
    ) -> Result<Vec<HavingCondition>> {
        let mut conditions = Vec::new();

        if let Some(obj) = having_obj.as_object() {
            for (key, value) in obj {
                // Parse condition: revenue_sum_gt: 1000
                if let Some((agg_name, operator)) = Self::parse_having_key(key) {
                    // Find the aggregate
                    let aggregate = aggregates
                        .iter()
                        .find(|a| a.alias() == agg_name)
                        .ok_or_else(|| FraiseQLError::Validation {
                            message: format!(
                                "HAVING condition references non-selected aggregate: {agg_name}"
                            ),
                            path:    None,
                        })?
                        .clone();

                    conditions.push(HavingCondition {
                        aggregate,
                        operator,
                        value: value.clone(),
                    });
                }
            }
        }

        Ok(conditions)
    }

    /// Parse HAVING key to extract aggregate name and operator
    fn parse_having_key(key: &str) -> Option<(&str, HavingOperator)> {
        for (suffix, op) in &[
            ("_gt", HavingOperator::Gt),
            ("_gte", HavingOperator::Gte),
            ("_lt", HavingOperator::Lt),
            ("_lte", HavingOperator::Lte),
            ("_eq", HavingOperator::Eq),
            ("_neq", HavingOperator::Neq),
        ] {
            if let Some(agg_name) = key.strip_suffix(suffix) {
                return Some((agg_name, *op));
            }
        }
        None
    }

    /// Parse ORDER BY clauses: each key takes the published `SortDirection`, `ASC` or `DESC`
    /// (#1800). Any other value is refused rather than read as `ASC`.
    fn parse_order_by(order_obj: &Value) -> Result<Vec<OrderByClause>> {
        let mut clauses = Vec::new();

        if let Some(obj) = order_obj.as_object() {
            for (field, value) in obj {
                let direction = match value.as_str() {
                    Some("ASC") => OrderDirection::Asc,
                    Some("DESC") => OrderDirection::Desc,
                    _ => {
                        return Err(FraiseQLError::Validation {
                            message: format!(
                                "orderBy `{field}` takes `ASC` or `DESC`, got {value}"
                            ),
                            path:    None,
                        });
                    },
                };

                clauses.push(OrderByClause::new(field.clone(), direction));
            }
        }

        Ok(clauses)
    }
}

/// The condition one fact-table `where` entry, `<field>_<operator>: value`, makes.
///
/// One resolution for aggregate and window queries (#1517; the window parser used to send
/// every key to the JSONB column): a dimension mapped to a native column is read from the
/// column, as its `groupBy` is (#1231), cast as the column's declared filter type is; a
/// filter column is read from itself; anything else is a JSONB dimension
/// ([`dimension_location`]).
pub(super) fn where_condition(
    key: &str,
    value: &Value,
    metadata: &FactTableMetadata,
    native_columns: &std::collections::HashMap<String, String>,
) -> Result<WhereClause> {
    let (field, operator) = split_where_key(key, metadata, native_columns)?;
    Ok(if let Some(column) = mapped_column(metadata, field) {
        WhereClause::NativeField {
            column: column.clone(),
            pg_cast: native_columns.get(column.as_str()).cloned().unwrap_or_default(),
            operator,
            value: value.clone(),
        }
    } else if let Some(pg_cast) = native_columns.get(field) {
        WhereClause::NativeField {
            column: field.to_string(),
            pg_cast: pg_cast.clone(),
            operator,
            value: value.clone(),
        }
    } else {
        WhereClause::Field {
            path: dimension_location(metadata, field)?,
            operator,
            value: value.clone(),
        }
    })
}

/// The JSON keys a `where` on the JSONB dimension `field` reads (#1517).
///
/// A declared dimension path is read at its declared location, whatever its depth, as
/// `groupBy` reads it. Any other key is a top-level JSONB key, recased so a camelCase
/// filter (`organizationId_eq`) builds `data->>'organization_id'` rather than a
/// never-matching `organizationId` key (#486). Shared by the aggregate and window parsers.
pub(super) fn dimension_location(metadata: &FactTableMetadata, field: &str) -> Result<Vec<String>> {
    metadata
        .declared_dimension_segments(field)
        .unwrap_or_else(|| Ok(vec![crate::utils::to_snake_case(field)]))
}

/// Validate that a regular-JSONB-dimension GraphQL key is a safe SQL identifier
/// (`[_A-Za-z][_0-9A-Za-z]*`).
///
/// Regular dimensions echo this key as the SELECT-list alias and as the JSONB key read.
/// The alias is identifier-quoted since #1516, so the response key keeps its case; this
/// charset check stays as the first layer (H1: a key containing commas, parentheses or a
/// subquery once injected an extra SELECT column when the alias was emitted bare).
/// Calendar/temporal/native dimensions are gated by lookups against server-defined
/// metadata, so only this fallback path needs the charset guard.
///
/// # Errors
///
/// Returns [`FraiseQLError::Validation`] if `key` is empty or contains a
/// character outside `[_A-Za-z][_0-9A-Za-z]*`.
fn validate_dimension_key(key: &str) -> Result<()> {
    let mut chars = key.chars();
    let first_ok = chars.next().is_some_and(|c| c.is_ascii_alphabetic() || c == '_');
    let rest_ok = chars.all(|c| c.is_ascii_alphanumeric() || c == '_');
    if first_ok && rest_ok {
        Ok(())
    } else {
        Err(FraiseQLError::Validation {
            message: format!(
                "groupBy dimension '{key}' contains invalid characters; \
                 only [_A-Za-z][_0-9A-Za-z]* is allowed"
            ),
            path:    None,
        })
    }
}

/// Split a fact-table `where` key, `<field>_<operator>`, against the operator table.
///
/// Shared by the aggregate and window parsers. Splitting at the last underscore could not
/// reach an operator spelled with one (`descendant_of`, `is_not_null`, `depth_eq`); the
/// aggregate parser refused such keys and the window parser dropped them, widening the row
/// set (#1460, #1499). Every split whose suffix names an operator is a candidate. When
/// several are (`path_depth_eq` reads as `path` + `depth_eq` or as `path_depth` + `eq`),
/// the one whose field the fact table declares (a denormalized filter, a native column or a
/// dimension path) wins, and a key that stays ambiguous is refused rather than guessed.
///
/// # Errors
///
/// `FraiseQLError::Validation` naming the key when no split names an operator, or when the
/// key stays ambiguous.
/// The native column a `native_dimension_mapping` maps the dimension `key` to (#1231).
///
/// Keys compare in `snake_case`, segment by segment: a mapping declared `itemCategory` and
/// one declared `item_category` both serve a request for either spelling. Looking the raw
/// request key up instead made a mapping in the other casing never fire, silently.
pub(super) fn mapped_column<'m>(metadata: &'m FactTableMetadata, key: &str) -> Option<&'m String> {
    let wanted = crate::compiler::fact_table::dimension_key(key);
    metadata
        .native_dimension_mapping
        .iter()
        .find(|(declared, _)| crate::compiler::fact_table::dimension_key(declared) == wanted)
        .map(|(_, column)| column)
}

pub(super) fn split_where_key<'k>(
    key: &'k str,
    metadata: &FactTableMetadata,
    native_columns: &std::collections::HashMap<String, String>,
) -> Result<(&'k str, WhereOperator)> {
    let is_declared = |field: &str| {
        let snake = crate::utils::to_snake_case(field);
        native_columns.contains_key(field)
            || metadata.denormalized_filters.iter().any(|f| f.name == field || f.name == snake)
            || metadata.dimensions.paths.iter().any(|p| p.name == field || p.name == snake)
    };
    let candidates: Vec<(&str, &str, WhereOperator)> = key
        .match_indices('_')
        .filter_map(|(i, _)| {
            let (field, suffix) = (&key[..i], &key[i + 1..]);
            let operator = WhereOperator::from_str(suffix).ok()?;
            (!field.is_empty()).then_some((field, suffix, operator))
        })
        .collect();
    let declared: Vec<_> = candidates.iter().filter(|(f, ..)| is_declared(f)).collect();
    let chosen = match (candidates.as_slice(), declared.as_slice()) {
        ([], _) => {
            return Err(FraiseQLError::validation(format!(
                "`where` key `{key}` is not `<field>_<operator>` with a supported operator"
            )));
        },
        ([only], _) => only,
        (_, [only]) => *only,
        (many, _) => {
            let readings: Vec<String> =
                many.iter().map(|(f, op, _)| format!("`{f}` `{op}`")).collect();
            return Err(FraiseQLError::validation(format!(
                "`where` key `{key}` is ambiguous: it reads as {}; declare the field as a \
                 filter column or a dimension path",
                readings.join(" or ")
            )));
        },
    };
    let (field, _, operator) = chosen;
    Ok((field, operator.clone()))
}

#[cfg(test)]
mod alias_injection_tests {
    #![allow(clippy::unwrap_used, clippy::panic)] // Reason: test code, panics are acceptable
    use std::collections::HashMap;

    use super::*;

    fn empty_metadata() -> FactTableMetadata {
        // `dimensions.paths` is empty on purpose: the compiler's dimension
        // allowlist is skipped in that state (the realistic empty-table-at-
        // compile-time default), so parse-time validation is the live guard.
        serde_json::from_value(serde_json::json!({
            "table_name": "tf_sales",
            "measures": [],
            "dimensions": { "name": "dimensions", "paths": [] },
            "denormalized_filters": []
        }))
        .expect("valid empty fact-table metadata")
    }

    #[test]
    fn validate_dimension_key_accepts_plain_identifiers() {
        assert!(validate_dimension_key("category").is_ok());
        assert!(validate_dimension_key("occurred_at_day").is_ok());
        assert!(validate_dimension_key("_private").is_ok());
    }

    #[test]
    fn validate_dimension_key_rejects_injection_shapes() {
        assert!(validate_dimension_key("a, (SELECT 1) AS x").is_err());
        assert!(validate_dimension_key("x; DROP TABLE t").is_err());
        assert!(validate_dimension_key("1leading_digit").is_err());
        assert!(validate_dimension_key("dotted.path").is_err());
        assert!(validate_dimension_key("").is_err());
    }

    #[test]
    fn parse_rejects_hostile_group_by_key_with_empty_allowlist() {
        // The H1 exploit: an alias that appends a scalar-subquery column. With
        // `dimensions.paths` empty the compiler allowlist never fires, so parse-
        // time validation must reject this on its own.
        let query = serde_json::json!({
            "groupBy": {
                "a, (SELECT string_agg(rolpassword, ',') FROM pg_authid) AS leak": true
            }
        });
        let err = AggregateQueryParser::parse(&query, &empty_metadata(), &HashMap::new())
            .expect_err("hostile groupBy key must be rejected at parse time");
        assert!(matches!(err, FraiseQLError::Validation { .. }), "got {err:?}");
    }

    #[test]
    fn parse_accepts_plain_group_by_key() {
        let query = serde_json::json!({
            "groupBy": { "category": true }
        });
        let req = AggregateQueryParser::parse(&query, &empty_metadata(), &HashMap::new())
            .expect("plain groupBy key must parse");
        assert_eq!(req.group_by.len(), 1);
    }

    #[test]
    fn camel_group_by_dimension_recases_path_but_keeps_alias() {
        // #486: a camelCase groupBy dimension must extract the snake_case JSONB key
        // (`data->>'machine_status'`) so grouping actually buckets rows, while the
        // alias stays the camel surface name because `AggregationProjector` uses it
        // verbatim as the GraphQL response key (#418/#410 "output key is the alias").
        let query = serde_json::json!({
            "groupBy": { "machineStatus": true }
        });
        let req = AggregateQueryParser::parse(&query, &empty_metadata(), &HashMap::new())
            .expect("camel groupBy key must parse");
        match req.group_by.as_slice() {
            [GroupBySelection::Dimension { path, alias }] => {
                assert_eq!(path, "machine_status", "path is the snake JSONB key");
                assert_eq!(alias, "machineStatus", "alias is the camel response key");
            },
            other => panic!("expected a single Dimension, got {other:?}"),
        }
    }
}