athena-gateway 3.18.0

//! Portable structured-fetch plan and parser support for `/gateway/fetch`.
//!
//! This module owns the request-domain half of structured fetch:
//!
//! - the parsed query/field/filter/order model,
//! - request-body validation,
//! - compatibility `select` string parsing,
//! - structured cache-key generation.
//!
//! SQL compilation, catalog loading, and backend execution stay in `athena_rs`.

use crate::{
    normalize_column_name, normalize_gateway_schema_name, qualify_gateway_table_name,
    sanitize_identifier, schema_name_from_body,
};
use serde::Serialize;
use serde_json::{Map, Value, json};
use std::collections::BTreeSet;
use std::iter::Peekable;

/// Parsed structured gateway fetch request plus the normalized root selector.
#[derive(Debug, Clone)]
pub struct StructuredGatewayFetchPlan {
    /// Normalized root table selector in `table` or `schema.table` form.
    pub table_name: String,
    /// Optional explicit root schema.
    pub schema_name: Option<String>,
    /// Parsed structured query model.
    pub query: StructuredSelectQuery,
}

impl StructuredGatewayFetchPlan {
    /// Returns the top-level limit, if present.
    pub fn limit(&self) -> Option<i64> {
        self.query.limit
    }

    /// Returns the unique resource names referenced by the root query and nested relations.
    pub fn resource_names(&self) -> Vec<String> {
        let mut resources = BTreeSet::new();
        collect_query_resource_names(&self.query, &mut resources);
        resources.into_iter().collect()
    }
}

/// Structured-fetch operation kinds currently accepted by the public contract.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub enum StructuredSelectOperation {
    Select,
}

/// Join behavior for nested structured relation selections.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub enum StructuredJoinKind {
    Left,
    Inner,
}

/// Supported filter operators in structured fetch requests.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub enum StructuredFilterOperator {
    Eq,
    Neq,
    Gt,
    Lt,
    In,
}

/// Supported sort directions in structured fetch requests.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub enum StructuredSortDirection {
    Asc,
    Desc,
}

impl StructuredSortDirection {
    /// Returns the SQL keyword form used by the server-side emitter.
    pub fn sql_keyword(&self) -> &'static str {
        match self {
            Self::Asc => "ASC",
            Self::Desc => "DESC",
        }
    }
}

/// Parsed structured fetch query.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct StructuredSelectQuery {
    pub operation: StructuredSelectOperation,
    pub from: String,
    pub fields: Vec<StructuredSelectField>,
    pub filters: Vec<StructuredFilter>,
    pub order_by: Vec<StructuredOrderBy>,
    pub limit: Option<i64>,
    pub offset: Option<i64>,
}

/// Field projection entry in a structured fetch query.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub enum StructuredSelectField {
    Column(StructuredColumnField),
    Relation(StructuredRelationField),
}

/// Column projection inside a structured fetch query.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct StructuredColumnField {
    pub name: String,
    pub alias: Option<String>,
}

/// Nested relation projection inside a structured fetch query.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct StructuredRelationField {
    pub name: String,
    pub alias: Option<String>,
    pub join: StructuredJoinKind,
    pub foreign_key: Option<String>,
    pub query: StructuredSelectQuery,
}

impl StructuredRelationField {
    /// Returns the response field name for this relation.
    pub fn display_name(&self) -> &str {
        self.alias.as_deref().unwrap_or(&self.name)
    }
}

/// Structured filter clause.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct StructuredFilter {
    pub column: String,
    pub operator: StructuredFilterOperator,
    pub values: Vec<Value>,
    pub column_cast: Option<String>,
    pub value_cast: Option<String>,
}

/// Structured ordering clause.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct StructuredOrderBy {
    pub column: String,
    pub direction: StructuredSortDirection,
}

/// Builds a parsed structured fetch plan when the request body uses the structured read contract.
pub fn build_structured_fetch_plan(
    body: &Value,
    force_camel_case_to_snake_case: bool,
) -> Result<Option<StructuredGatewayFetchPlan>, String> {
    if uses_top_level_direct_ast_payload(body) {
        return Err(
            "first-class direct AST request bodies are not supported on /gateway/fetch in this Athena release; use table_name plus select"
                .to_string(),
        );
    }

    if !is_structured_fetch_body(body) {
        return Ok(None);
    }

    let schema_name = normalize_gateway_schema_name(schema_name_from_body(body).as_deref())?;
    let from = body
        .get("table_name")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| "table_name is required for structured gateway fetch".to_string())?;

    let qualified_table_name = qualify_gateway_table_name(from, schema_name.as_deref())?;
    let query = parse_find_many_query(body, from, force_camel_case_to_snake_case)?;

    Ok(Some(StructuredGatewayFetchPlan {
        table_name: qualified_table_name,
        schema_name,
        query,
    }))
}

/// Builds the structured fetch cache key from the normalized plan and compiled SQL.
pub fn build_structured_fetch_cache_key(
    plan: &StructuredGatewayFetchPlan,
    sql: &str,
    strip_nulls: bool,
    client_name: &str,
) -> String {
    let hash_input = json!({
        "table_name": plan.table_name,
        "schema_name": plan.schema_name,
        "sql": sql,
        "strip_nulls": strip_nulls,
        "client": client_name,
    });
    let digest = sha256::digest(serde_json::to_string(&hash_input).unwrap_or_default());
    let short_hash = &digest[..16.min(digest.len())];
    format!(
        "{}:structured:{}:{}:{}",
        plan.table_name, client_name, strip_nulls, short_hash
    )
}

fn is_structured_fetch_body(body: &Value) -> bool {
    body.get("select").is_some() || uses_top_level_direct_ast_payload(body)
}

fn uses_top_level_direct_ast_payload(body: &Value) -> bool {
    body.get("operation").is_some()
        || body.get("fields").is_some()
        || (body.get("from").is_some() && body.get("table_name").is_none())
}

fn parse_find_many_query(
    body: &Value,
    from: &str,
    force_camel_case_to_snake_case: bool,
) -> Result<StructuredSelectQuery, String> {
    let select_value = body
        .get("select")
        .ok_or_else(|| "select is required for structured gateway fetch".to_string())?;
    let fields = parse_select_input(
        select_value,
        force_camel_case_to_snake_case,
        Some(normalize_query_table_name(from)),
    )?;

    Ok(StructuredSelectQuery {
        operation: StructuredSelectOperation::Select,
        from: from.to_string(),
        fields,
        filters: parse_where_clause(
            body.get("where_filters")
                .or_else(|| body.get("where"))
                .or_else(|| body.get("where_clause")),
            force_camel_case_to_snake_case,
        )?,
        order_by: parse_order_by_clause(
            body.get("orderBy").or_else(|| body.get("order_by")),
            force_camel_case_to_snake_case,
        )?,
        limit: parse_optional_i64(body.get("limit"), "limit")?,
        offset: parse_optional_i64(body.get("offset"), "offset")?,
    })
}

fn parse_ast_query(
    value: &Value,
    fallback_from: Option<&str>,
    force_camel_case_to_snake_case: bool,
) -> Result<StructuredSelectQuery, String> {
    let object = value
        .as_object()
        .ok_or_else(|| "structured gateway fetch body must be a JSON object".to_string())?;

    let operation = match object.get("operation").and_then(Value::as_str) {
        None | Some("select") => StructuredSelectOperation::Select,
        Some(other) => {
            return Err(format!(
                "unsupported structured gateway fetch operation '{other}'"
            ));
        }
    };

    let from = object
        .get("from")
        .and_then(Value::as_str)
        .or(fallback_from)
        .map(str::trim)
        .filter(|candidate| !candidate.is_empty())
        .ok_or_else(|| "from is required for structured gateway fetch query".to_string())?
        .to_string();

    let fields = if let Some(fields_value) = object.get("fields") {
        parse_ast_fields(
            fields_value,
            force_camel_case_to_snake_case,
            Some(normalize_query_table_name(&from)),
        )?
    } else if let Some(select_value) = object.get("select") {
        parse_select_input(
            select_value,
            force_camel_case_to_snake_case,
            Some(normalize_query_table_name(&from)),
        )?
    } else {
        return Err("fields or select is required for structured gateway fetch query".to_string());
    };

    Ok(StructuredSelectQuery {
        operation,
        from,
        fields,
        filters: parse_where_clause(
            object
                .get("where_filters")
                .or_else(|| object.get("where"))
                .or_else(|| object.get("where_clause")),
            force_camel_case_to_snake_case,
        )?,
        order_by: parse_order_by_clause(
            object.get("orderBy").or_else(|| object.get("order_by")),
            force_camel_case_to_snake_case,
        )?,
        limit: parse_optional_i64(object.get("limit"), "limit")?,
        offset: parse_optional_i64(object.get("offset"), "offset")?,
    })
}

fn parse_select_input(
    value: &Value,
    force_camel_case_to_snake_case: bool,
    parent_table: Option<String>,
) -> Result<Vec<StructuredSelectField>, String> {
    if let Some(select_object) = value.as_object() {
        return parse_select_object(select_object, force_camel_case_to_snake_case);
    }
    if let Some(select_string) = value.as_str() {
        return parse_select_string(select_string, parent_table, force_camel_case_to_snake_case);
    }
    if value.is_array() {
        return parse_ast_fields(value, force_camel_case_to_snake_case, parent_table);
    }
    Err("select must be an object, string, or fields array".to_string())
}

fn parse_select_object(
    object: &Map<String, Value>,
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredSelectField>, String> {
    let mut fields = Vec::new();

    for (raw_key, value) in object {
        let key = normalize_path(raw_key, force_camel_case_to_snake_case)?;
        if matches!(value, Value::Bool(true)) {
            fields.push(StructuredSelectField::Column(StructuredColumnField {
                name: key,
                alias: None,
            }));
            continue;
        }

        let relation_object = value.as_object().ok_or_else(|| {
            format!("select['{raw_key}'] must be true for a column or an object with nested select")
        })?;
        if relation_object.get("select").is_none() && relation_object.get("fields").is_none() {
            return Err(format!(
                "select['{raw_key}'] is missing nested select/fields for relation projection"
            ));
        }

        let nested_query = parse_ast_query(value, Some(&key), force_camel_case_to_snake_case)?;
        fields.push(StructuredSelectField::Relation(StructuredRelationField {
            name: key,
            alias: None,
            join: StructuredJoinKind::Left,
            foreign_key: None,
            query: nested_query,
        }));
    }

    if fields.is_empty() {
        return Err("select must include at least one column or relation".to_string());
    }

    Ok(fields)
}

fn parse_ast_fields(
    value: &Value,
    force_camel_case_to_snake_case: bool,
    parent_table: Option<String>,
) -> Result<Vec<StructuredSelectField>, String> {
    let array = value
        .as_array()
        .ok_or_else(|| "fields must be an array".to_string())?;
    let mut fields = Vec::new();

    for field in array {
        let object = field
            .as_object()
            .ok_or_else(|| "each fields entry must be an object".to_string())?;
        let kind = object
            .get("kind")
            .and_then(Value::as_str)
            .ok_or_else(|| "each fields entry must include kind".to_string())?;

        match kind {
            "column" => {
                let raw_name = object
                    .get("name")
                    .and_then(Value::as_str)
                    .ok_or_else(|| "column fields require name".to_string())?;
                let alias = parse_optional_alias(object.get("alias"))?;
                fields.push(StructuredSelectField::Column(StructuredColumnField {
                    name: normalize_path(raw_name, force_camel_case_to_snake_case)?,
                    alias,
                }));
            }
            "relation" => {
                let raw_name = object
                    .get("name")
                    .and_then(Value::as_str)
                    .ok_or_else(|| "relation fields require name".to_string())?;
                let name = normalize_path(raw_name, force_camel_case_to_snake_case)?;
                let alias = parse_optional_alias(object.get("alias"))?;
                let join = parse_join_kind(object.get("join").or_else(|| object.get("join_type")))?;
                let foreign_key = parse_optional_foreign_key(object.get("foreign_key"))?;
                let nested_query = parse_ast_query(
                    object
                        .get("query")
                        .ok_or_else(|| "relation fields require query".to_string())?,
                    Some(&name),
                    force_camel_case_to_snake_case,
                )?;

                fields.push(StructuredSelectField::Relation(StructuredRelationField {
                    name,
                    alias,
                    join,
                    foreign_key,
                    query: nested_query,
                }));
            }
            other => {
                return Err(format!("unsupported field kind '{other}'"));
            }
        }
    }

    if fields.is_empty() {
        return Err("fields must include at least one entry".to_string());
    }

    if let Some(parent_table) = parent_table
        && fields
            .iter()
            .all(|field| matches!(field, StructuredSelectField::Relation(_)))
    {
        return Err(format!(
            "structured select for '{parent_table}' must include at least one column"
        ));
    }

    Ok(fields)
}

fn parse_where_clause(
    value: Option<&Value>,
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredFilter>, String> {
    let Some(value) = value else {
        return Ok(Vec::new());
    };

    if let Some(array) = value.as_array() {
        return parse_where_filters_array(array, force_camel_case_to_snake_case);
    }

    let object = value
        .as_object()
        .ok_or_else(|| "where/where_clause must be an object".to_string())?;
    let mut filters = Vec::new();

    for (raw_column, raw_filter) in object {
        let column = normalize_path(raw_column, force_camel_case_to_snake_case)?;
        match raw_filter {
            Value::Object(filter_object) => {
                let mut pushed = false;
                for (raw_operator, operand) in filter_object {
                    let operator = match raw_operator.as_str() {
                        "eq" => StructuredFilterOperator::Eq,
                        "neq" => StructuredFilterOperator::Neq,
                        "gt" => StructuredFilterOperator::Gt,
                        "lt" => StructuredFilterOperator::Lt,
                        "in" => StructuredFilterOperator::In,
                        other => {
                            return Err(format!(
                                "unsupported where operator '{other}' for column '{raw_column}'"
                            ));
                        }
                    };
                    let values = if operator == StructuredFilterOperator::In {
                        operand.as_array().cloned().ok_or_else(|| {
                            format!("where.{raw_column}.in must be an array of values")
                        })?
                    } else {
                        vec![operand.clone()]
                    };
                    filters.push(StructuredFilter {
                        column: column.clone(),
                        operator,
                        values,
                        column_cast: None,
                        value_cast: None,
                    });
                    pushed = true;
                }
                if !pushed {
                    return Err(format!(
                        "where.{raw_column} must include at least one supported operator"
                    ));
                }
            }
            Value::Array(values) => {
                filters.push(StructuredFilter {
                    column,
                    operator: StructuredFilterOperator::In,
                    values: values.clone(),
                    column_cast: None,
                    value_cast: None,
                });
            }
            scalar => {
                filters.push(StructuredFilter {
                    column,
                    operator: StructuredFilterOperator::Eq,
                    values: vec![scalar.clone()],
                    column_cast: None,
                    value_cast: None,
                });
            }
        }
    }

    Ok(filters)
}

fn parse_where_filters_array(
    array: &[Value],
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredFilter>, String> {
    let mut filters = Vec::new();

    for entry in array {
        let object = entry
            .as_object()
            .ok_or_else(|| "where/where_clause array entries must be objects".to_string())?;
        let raw_column = object
            .get("column")
            .and_then(Value::as_str)
            .ok_or_else(|| "where/where_clause array entries require column".to_string())?;
        let raw_operator = object
            .get("operator")
            .and_then(Value::as_str)
            .ok_or_else(|| "where/where_clause array entries require operator".to_string())?;
        let operator = parse_structured_filter_operator(raw_operator, raw_column)?;
        let column = normalize_path(raw_column, force_camel_case_to_snake_case)?;
        let column_cast = parse_optional_type_cast(
            object
                .get("column_cast")
                .or_else(|| object.get("columnCast")),
            "column_cast",
        )?;
        let value_cast = parse_optional_type_cast(
            object.get("value_cast").or_else(|| object.get("valueCast")),
            "value_cast",
        )?;
        let values = if operator == StructuredFilterOperator::In {
            object
                .get("values")
                .or_else(|| object.get("value"))
                .and_then(Value::as_array)
                .cloned()
                .ok_or_else(|| {
                    "where/where_clause array entry with operator 'in' requires values array"
                        .to_string()
                })?
        } else {
            vec![object.get("value").cloned().ok_or_else(|| {
                "where/where_clause array entries require value for scalar operators".to_string()
            })?]
        };

        filters.push(StructuredFilter {
            column,
            operator,
            values,
            column_cast,
            value_cast,
        });
    }

    Ok(filters)
}

fn parse_structured_filter_operator(
    raw_operator: &str,
    raw_column: &str,
) -> Result<StructuredFilterOperator, String> {
    match raw_operator {
        "eq" => Ok(StructuredFilterOperator::Eq),
        "neq" => Ok(StructuredFilterOperator::Neq),
        "gt" => Ok(StructuredFilterOperator::Gt),
        "lt" => Ok(StructuredFilterOperator::Lt),
        "in" => Ok(StructuredFilterOperator::In),
        other => Err(format!(
            "unsupported where operator '{other}' for column '{raw_column}'"
        )),
    }
}

fn parse_order_by_clause(
    value: Option<&Value>,
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredOrderBy>, String> {
    let Some(value) = value else {
        return Ok(Vec::new());
    };

    if let Some(object) = value.as_object() {
        let mut order_by = Vec::new();
        for (raw_column, raw_direction) in object {
            let direction = parse_sort_direction(raw_direction, raw_column)?;
            order_by.push(StructuredOrderBy {
                column: normalize_path(raw_column, force_camel_case_to_snake_case)?,
                direction,
            });
        }
        return Ok(order_by);
    }

    if let Some(array) = value.as_array() {
        let mut order_by = Vec::new();
        for entry in array {
            let object = entry
                .as_object()
                .ok_or_else(|| "orderBy/order_by array entries must be objects".to_string())?;
            let raw_column = object
                .get("column")
                .or_else(|| object.get("field"))
                .and_then(Value::as_str)
                .ok_or_else(|| "orderBy/order_by array entries require column/field".to_string())?;
            let direction = parse_sort_direction(
                object
                    .get("direction")
                    .or_else(|| object.get("order"))
                    .unwrap_or(&Value::String("asc".to_string())),
                raw_column,
            )?;
            order_by.push(StructuredOrderBy {
                column: normalize_path(raw_column, force_camel_case_to_snake_case)?,
                direction,
            });
        }
        return Ok(order_by);
    }

    Err("orderBy/order_by must be an object or array".to_string())
}

fn parse_optional_i64(value: Option<&Value>, label: &str) -> Result<Option<i64>, String> {
    let Some(value) = value else {
        return Ok(None);
    };
    let parsed = value
        .as_i64()
        .ok_or_else(|| format!("{label} must be an integer"))?;
    if parsed < 0 {
        return Err(format!("{label} must be greater than or equal to 0"));
    }
    Ok(Some(parsed))
}

fn parse_optional_alias(value: Option<&Value>) -> Result<Option<String>, String> {
    let Some(value) = value else {
        return Ok(None);
    };
    let alias = value
        .as_str()
        .map(str::trim)
        .filter(|candidate| !candidate.is_empty())
        .ok_or_else(|| "alias must be a non-empty string".to_string())?;
    validate_identifier(alias, "alias")?;
    Ok(Some(alias.to_string()))
}

fn parse_optional_type_cast(value: Option<&Value>, label: &str) -> Result<Option<String>, String> {
    let Some(value) = value else {
        return Ok(None);
    };
    let raw_cast = value
        .as_str()
        .map(str::trim)
        .filter(|candidate| !candidate.is_empty())
        .ok_or_else(|| format!("{label} must be a non-empty string"))?;
    sanitize_qualified_identifier(raw_cast)
        .ok_or_else(|| format!("{label} '{raw_cast}' must be a valid SQL type identifier"))?;
    Ok(Some(raw_cast.to_string()))
}

fn parse_optional_foreign_key(value: Option<&Value>) -> Result<Option<String>, String> {
    let Some(value) = value else {
        return Ok(None);
    };
    let raw = value
        .as_str()
        .map(str::trim)
        .filter(|candidate| !candidate.is_empty())
        .ok_or_else(|| "foreign_key must be a non-empty string".to_string())?;
    validate_foreign_key_hint(raw)?;
    Ok(Some(raw.to_string()))
}

fn parse_join_kind(value: Option<&Value>) -> Result<StructuredJoinKind, String> {
    match value.and_then(Value::as_str) {
        None | Some("left") => Ok(StructuredJoinKind::Left),
        Some("inner") => Ok(StructuredJoinKind::Inner),
        Some(other) => Err(format!("unsupported relation join kind '{other}'")),
    }
}

fn parse_sort_direction(value: &Value, context: &str) -> Result<StructuredSortDirection, String> {
    let raw = value
        .as_str()
        .ok_or_else(|| format!("sort direction for '{context}' must be a string"))?;
    match raw.to_ascii_lowercase().as_str() {
        "asc" | "ascending" => Ok(StructuredSortDirection::Asc),
        "desc" | "descending" => Ok(StructuredSortDirection::Desc),
        other => Err(format!(
            "unsupported sort direction '{other}' for '{context}'"
        )),
    }
}

fn validate_foreign_key_hint(raw: &str) -> Result<(), String> {
    if let Some(stripped) = raw.strip_prefix("parent.") {
        validate_identifier(stripped, "foreign_key")?;
        return Ok(());
    }
    if let Some(stripped) = raw.strip_prefix("child.") {
        validate_identifier(stripped, "foreign_key")?;
        return Ok(());
    }
    validate_identifier(raw, "foreign_key")
}

fn normalize_path(raw: &str, force_camel_case_to_snake_case: bool) -> Result<String, String> {
    let mut normalized_segments = Vec::new();
    for segment in raw.split('.') {
        let trimmed = segment.trim();
        if trimmed.is_empty() {
            return Err(format!("invalid identifier path '{raw}'"));
        }
        let normalized = normalize_column_name(trimmed, force_camel_case_to_snake_case);
        validate_identifier(&normalized, "identifier")?;
        normalized_segments.push(normalized);
    }
    Ok(normalized_segments.join("."))
}

fn validate_identifier(identifier: &str, label: &str) -> Result<(), String> {
    sanitize_identifier(identifier)
        .map(|_| ())
        .ok_or_else(|| format!("{label} '{identifier}' must be a valid SQL identifier"))
}

fn normalize_query_table_name(raw: &str) -> String {
    raw.rsplit('.').next().unwrap_or(raw).trim().to_string()
}

fn collect_query_resource_names(query: &StructuredSelectQuery, resources: &mut BTreeSet<String>) {
    resources.insert(normalize_query_table_name(&query.from));

    for field in &query.fields {
        if let StructuredSelectField::Relation(relation) = field {
            collect_query_resource_names(&relation.query, resources);
        }
    }
}

fn parse_select_string(
    input: &str,
    parent_table: Option<String>,
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredSelectField>, String> {
    let mut iter = input.chars().peekable();
    let fields = parse_string_nodes(&mut iter, parent_table, force_camel_case_to_snake_case)?;
    skip_whitespace(&mut iter);
    if iter.peek().is_some() {
        return Err("invalid select string: trailing tokens".to_string());
    }
    Ok(fields)
}

fn parse_string_nodes<I>(
    iter: &mut Peekable<I>,
    parent_table: Option<String>,
    force_camel_case_to_snake_case: bool,
) -> Result<Vec<StructuredSelectField>, String>
where
    I: Iterator<Item = char>,
{
    let mut fields = Vec::new();
    loop {
        skip_whitespace(iter);
        if iter.peek().is_none() || matches!(iter.peek(), Some(')')) {
            break;
        }
        fields.push(parse_string_node(
            iter,
            parent_table.clone(),
            force_camel_case_to_snake_case,
        )?);
        skip_whitespace(iter);
        if matches!(iter.peek(), Some(',')) {
            iter.next();
        }
    }
    Ok(fields)
}

fn parse_string_node<I>(
    iter: &mut Peekable<I>,
    parent_table: Option<String>,
    force_camel_case_to_snake_case: bool,
) -> Result<StructuredSelectField, String>
where
    I: Iterator<Item = char>,
{
    let token = read_string_token(iter)?;
    let mut alias = None;
    let mut name = token.clone();

    if matches!(iter.peek(), Some(':')) {
        iter.next();
        alias = Some(token);
        name = read_string_token(iter)?;
    }

    let mut join = StructuredJoinKind::Left;
    let mut foreign_key = None;

    if matches!(iter.peek(), Some('!')) {
        iter.next();
        let modifier = read_modifier_token(iter)?;
        if modifier.eq_ignore_ascii_case("inner") {
            join = StructuredJoinKind::Inner;
        } else if !modifier.is_empty() {
            validate_foreign_key_hint(&modifier)?;
            foreign_key = Some(modifier);
        }
    }

    if matches!(iter.peek(), Some('(')) {
        iter.next();
        let relation_name = normalize_path(&name, force_camel_case_to_snake_case)?;
        let nested_fields = parse_string_nodes(
            iter,
            Some(relation_name.clone()),
            force_camel_case_to_snake_case,
        )?;
        if iter.next() != Some(')') {
            return Err("invalid select string: missing ')'".to_string());
        }
        let relation_alias = match alias {
            Some(raw_alias) => Some(normalize_path(&raw_alias, false)?),
            None => None,
        };
        Ok(StructuredSelectField::Relation(StructuredRelationField {
            name: relation_name.clone(),
            alias: relation_alias,
            join,
            foreign_key,
            query: StructuredSelectQuery {
                operation: StructuredSelectOperation::Select,
                from: relation_name,
                fields: nested_fields,
                filters: Vec::new(),
                order_by: Vec::new(),
                limit: None,
                offset: None,
            },
        }))
    } else {
        let column_name = normalize_path(&name, force_camel_case_to_snake_case)?;
        let column_alias = match alias {
            Some(raw_alias) => Some(normalize_path(&raw_alias, false)?),
            None => None,
        };
        if column_name == "*" && parent_table.is_some() {
            return Err(
                "'*' is not supported inside structured nested select projections".to_string(),
            );
        }
        Ok(StructuredSelectField::Column(StructuredColumnField {
            name: column_name,
            alias: column_alias,
        }))
    }
}

fn read_string_token<I>(iter: &mut Peekable<I>) -> Result<String, String>
where
    I: Iterator<Item = char>,
{
    let mut token = String::new();
    while let Some(&ch) = iter.peek() {
        if matches!(ch, '!' | ':' | '(' | ')' | ',') {
            break;
        }
        token.push(ch);
        iter.next();
    }
    let trimmed = token.trim();
    if trimmed.is_empty() {
        return Err("invalid select string: empty token".to_string());
    }
    Ok(trimmed.to_string())
}

fn read_modifier_token<I>(iter: &mut Peekable<I>) -> Result<String, String>
where
    I: Iterator<Item = char>,
{
    let mut token = String::new();
    while let Some(&ch) = iter.peek() {
        if matches!(ch, '(' | ')' | ',') {
            break;
        }
        token.push(ch);
        iter.next();
    }
    let trimmed = token.trim();
    if trimmed.is_empty() {
        return Err("invalid select string: empty relation modifier".to_string());
    }
    Ok(trimmed.to_string())
}

fn skip_whitespace<I>(iter: &mut Peekable<I>)
where
    I: Iterator<Item = char>,
{
    while matches!(iter.peek(), Some(ch) if ch.is_whitespace()) {
        iter.next();
    }
}

fn sanitize_qualified_identifier(identifier: &str) -> Option<String> {
    let mut parts = Vec::new();
    for segment in identifier.split('.') {
        let trimmed = segment.trim().trim_matches('"');
        if trimmed.is_empty() {
            return None;
        }
        parts.push(sanitize_identifier(trimmed)?);
    }
    if parts.is_empty() {
        return None;
    }
    Some(parts.join("."))
}

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

    fn build_plan(
        body: &Value,
        force_camel_case_to_snake_case: bool,
    ) -> StructuredGatewayFetchPlan {
        build_structured_fetch_plan(body, force_camel_case_to_snake_case)
            .expect("plan ok")
            .expect("structured plan")
    }

    #[test]
    fn structured_fetch_rejects_top_level_direct_ast_body() {
        let body = json!({
            "operation": "select",
            "from": "orchestral_sections",
            "fields": [
                {
                    "kind": "column",
                    "name": "name"
                }
            ]
        });

        let err = build_structured_fetch_plan(&body, false).expect_err("ast should be rejected");
        assert!(err.contains("first-class direct AST request bodies are not supported"));
    }

    #[test]
    fn structured_select_collects_nested_resource_names() {
        let body = json!({
            "table_name": "orchestral_sections",
            "select": {
                "name": true,
                "instruments": {
                    "select": {
                        "name": true,
                        "players": {
                            "select": {
                                "display_name": true
                            }
                        }
                    }
                }
            }
        });

        let plan = build_plan(&body, false);

        assert_eq!(
            plan.resource_names(),
            vec![
                "instruments".to_string(),
                "orchestral_sections".to_string(),
                "players".to_string(),
            ]
        );
    }

    #[test]
    fn structured_select_where_only_body_stays_legacy() {
        let body = json!({
            "table_name": "users",
            "where": {
                "id": { "eq": 1 }
            }
        });

        assert!(
            build_structured_fetch_plan(&body, false)
                .expect("plan ok")
                .is_none()
        );
    }

    #[test]
    fn structured_select_string_alias_and_join_parse() {
        let body = json!({
            "table_name": "public.chat_subscriptions",
            "select": "user_id,users:athena.users!inner(id,username)"
        });

        let plan = build_plan(&body, false);
        assert_eq!(plan.table_name, "public.chat_subscriptions");
        assert_eq!(plan.query.fields.len(), 2);

        match &plan.query.fields[1] {
            StructuredSelectField::Relation(relation) => {
                assert_eq!(relation.name, "athena.users");
                assert_eq!(relation.alias.as_deref(), Some("users"));
                assert_eq!(relation.join, StructuredJoinKind::Inner);
                assert_eq!(relation.display_name(), "users");
            }
            other => panic!("expected relation field, got {other:?}"),
        }
    }
}