athena-gateway 3.18.0

//! Portable PostgREST compatibility parsing for Athena's legacy REST shim.
//!
//! The parser operates on plain query-string and header values so the grammar
//! can be reused without pulling `actix_web::HttpRequest` into this crate.

use athena_query::postgres_types::where_cast_for_column;
use athena_query::query_builder::{Condition, ConditionOperator};
use serde_json::Value;
use serde_urlencoded::from_str;
use std::collections::HashMap;

use crate::normalize_column_name;

/// Normalized PostgREST query components derived from a request URL.
#[derive(Debug, Clone, PartialEq)]
pub struct PostgrestQuery {
    pub columns: Vec<String>,
    pub filters: Vec<PostgrestFilter>,
    pub or_filters: Vec<Vec<PostgrestFilter>>,
    pub limit: Option<i64>,
    pub offset: Option<i64>,
    pub order: Option<OrderSpec>,
}

/// One parsed PostgREST filter expression.
#[derive(Debug, Clone, PartialEq)]
pub struct PostgrestFilter {
    pub column: String,
    pub operator: PostgrestFilterOperator,
    pub values: Vec<Value>,
    pub negated: bool,
}

/// PostgREST filter operators supported by the legacy REST shim.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PostgrestFilterOperator {
    Eq,
    Neq,
    Gt,
    Lt,
    Gte,
    Lte,
    Like,
    ILike,
    Is,
    In,
    Contains,
    Contained,
}

/// Parsed `order` query parameter.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrderSpec {
    pub column: String,
    pub ascending: bool,
}

/// Parses a Supabase/PostgREST query into a normalized representation that can
/// be mapped to Athena/Postgres operations.
pub fn parse_postgrest_query(
    query_string: &str,
    range_header: Option<&str>,
    force_snake_case: bool,
) -> Result<PostgrestQuery, String> {
    let mut columns: Vec<String> = vec!["*".to_string()];
    let mut filters: Vec<PostgrestFilter> = Vec::new();
    let mut or_filters: Vec<Vec<PostgrestFilter>> = Vec::new();
    let mut limit: Option<i64> = None;
    let mut offset: Option<i64> = None;
    let mut order: Option<OrderSpec> = None;

    let query_pairs: Vec<(String, String)> = from_str::<Vec<(String, String)>>(query_string)
        .map_err(|err| format!("failed to parse query string: {err}"))?;

    for (key, value) in query_pairs {
        match key.as_str() {
            "select" => {
                let parsed: Vec<String> = value
                    .split(',')
                    .map(|part| part.trim().to_string())
                    .filter(|part| !part.is_empty())
                    .collect();
                if !parsed.is_empty() {
                    columns = parsed;
                }
            }
            "limit" => {
                if let Ok(parsed) = value.parse::<i64>() {
                    limit = Some(parsed);
                }
            }
            "offset" => {
                if let Ok(parsed) = value.parse::<i64>() {
                    offset = Some(parsed);
                }
            }
            "order" => {
                if let Some(spec) = parse_order(&value, force_snake_case) {
                    order = Some(spec);
                }
            }
            "or" => {
                if let Some(parsed) = parse_or_filter(&value, force_snake_case)
                    && !parsed.is_empty()
                {
                    or_filters.push(parsed);
                }
            }
            other => {
                if let Some(filter) = parse_filter(other, &value, force_snake_case) {
                    filters.push(filter);
                }
            }
        }
    }

    if let Some((start, end)) = range_header.and_then(parse_range_header_value)
        && end >= start
    {
        offset = Some(start);
        limit = Some(end - start + 1);
    }

    Ok(PostgrestQuery {
        columns,
        filters,
        or_filters,
        limit,
        offset,
        order,
    })
}

/// Converts parsed PostgREST filters into Athena query-builder conditions.
pub fn convert_postgrest_filters_to_conditions(
    filters: &[PostgrestFilter],
    auto_cast_uuid_filter_values_to_text: bool,
    column_types: Option<&HashMap<String, String>>,
) -> Vec<Condition> {
    filters
        .iter()
        .filter_map(|filter| {
            convert_postgrest_filter_to_condition(
                filter,
                auto_cast_uuid_filter_values_to_text,
                column_types,
            )
        })
        .collect()
}

/// Converts OR filter groups into Athena query-builder condition groups.
pub fn convert_postgrest_or_filter_groups_to_conditions(
    or_groups: &[Vec<PostgrestFilter>],
    auto_cast_uuid_filter_values_to_text: bool,
    column_types: Option<&HashMap<String, String>>,
) -> Vec<Vec<Condition>> {
    or_groups
        .iter()
        .map(|group| {
            convert_postgrest_filters_to_conditions(
                group,
                auto_cast_uuid_filter_values_to_text,
                column_types,
            )
        })
        .filter(|group| !group.is_empty())
        .collect()
}

/// Converts one parsed PostgREST filter into an Athena query condition.
pub fn convert_postgrest_filter_to_condition(
    filter: &PostgrestFilter,
    auto_cast_uuid_filter_values_to_text: bool,
    column_types: Option<&HashMap<String, String>>,
) -> Option<Condition> {
    let cast: Option<&'static str> = where_cast_for_column(&filter.column, column_types);
    Some(
        Condition::new(
            filter.column.clone(),
            map_filter_operator(filter.operator),
            filter.values.clone(),
            filter.negated,
        )
        .with_uuid_value_text_cast(auto_cast_uuid_filter_values_to_text)
        .with_pg_cast(cast),
    )
}

fn map_filter_operator(op: PostgrestFilterOperator) -> ConditionOperator {
    match op {
        PostgrestFilterOperator::Eq => ConditionOperator::Eq,
        PostgrestFilterOperator::Neq => ConditionOperator::Neq,
        PostgrestFilterOperator::Gt => ConditionOperator::Gt,
        PostgrestFilterOperator::Lt => ConditionOperator::Lt,
        PostgrestFilterOperator::Gte => ConditionOperator::Gte,
        PostgrestFilterOperator::Lte => ConditionOperator::Lte,
        PostgrestFilterOperator::Like => ConditionOperator::Like,
        PostgrestFilterOperator::ILike => ConditionOperator::ILike,
        PostgrestFilterOperator::Is => ConditionOperator::Is,
        PostgrestFilterOperator::In => ConditionOperator::In,
        PostgrestFilterOperator::Contains => ConditionOperator::Contains,
        PostgrestFilterOperator::Contained => ConditionOperator::Contained,
    }
}

fn parse_order(value: &str, force_snake_case: bool) -> Option<OrderSpec> {
    let trimmed = value.trim();
    let (raw_column, ascending) = if let Some(column) = trimmed.strip_suffix(".asc") {
        (column, true)
    } else if let Some(column) = trimmed.strip_suffix(".desc") {
        (column, false)
    } else {
        (trimmed, true)
    };
    let column: String = if force_snake_case {
        normalize_column_name(raw_column, true)
    } else {
        raw_column.to_ascii_lowercase()
    };
    if column.is_empty() {
        return None;
    }
    Some(OrderSpec { column, ascending })
}

fn parse_or_filter(value: &str, force_snake_case: bool) -> Option<Vec<PostgrestFilter>> {
    let trimmed: &str = value.trim();
    let inner: &str = trimmed.trim_start_matches('(').trim_end_matches(')').trim();

    if inner.is_empty() {
        return None;
    }

    let mut filters: Vec<PostgrestFilter> = Vec::new();
    for expression in inner.split(',') {
        let expression: &str = expression.trim();
        if expression.is_empty() {
            continue;
        }
        if let Some((column, _remainder)) = expression.split_once('.')
            && let Some(filter) = parse_filter(
                column,
                expression
                    .strip_prefix(&format!("{}.", column))
                    .unwrap_or(""),
                force_snake_case,
            )
        {
            filters.push(filter);
        }
    }

    if filters.is_empty() {
        None
    } else {
        Some(filters)
    }
}

fn parse_filter(column: &str, expression: &str, force_snake_case: bool) -> Option<PostgrestFilter> {
    let normalized_column: String = if force_snake_case {
        normalize_column_name(column, true)
    } else {
        column.to_string()
    };

    if normalized_column.is_empty() {
        return None;
    }

    let (negated, expr) = if let Some(stripped) = expression.strip_prefix("not.") {
        (true, stripped)
    } else {
        (false, expression)
    };

    let (operator_str, value_str) = if let Some((op, rest)) = expr.split_once('.') {
        (op, rest)
    } else {
        return None;
    };

    let operator: PostgrestFilterOperator = match operator_str.to_lowercase().as_str() {
        "eq" => PostgrestFilterOperator::Eq,
        "neq" => PostgrestFilterOperator::Neq,
        "gt" => PostgrestFilterOperator::Gt,
        "lt" => PostgrestFilterOperator::Lt,
        "gte" => PostgrestFilterOperator::Gte,
        "lte" => PostgrestFilterOperator::Lte,
        "like" => PostgrestFilterOperator::Like,
        "ilike" => PostgrestFilterOperator::ILike,
        "is" => PostgrestFilterOperator::Is,
        "in" => PostgrestFilterOperator::In,
        "cs" => PostgrestFilterOperator::Contains,
        "cd" => PostgrestFilterOperator::Contained,
        other => {
            if let Some(stripped) = other.strip_prefix("array_") {
                match stripped {
                    "contains" => PostgrestFilterOperator::Contains,
                    "contained" => PostgrestFilterOperator::Contained,
                    _ => PostgrestFilterOperator::Eq,
                }
            } else {
                PostgrestFilterOperator::Eq
            }
        }
    };

    let values: Vec<Value> = match operator {
        PostgrestFilterOperator::In => parse_in_values(value_str),
        PostgrestFilterOperator::Contains | PostgrestFilterOperator::Contained => {
            vec![parse_array_filter(value_str)]
        }
        PostgrestFilterOperator::Is => vec![parse_scalar_value(value_str)],
        _ => vec![parse_scalar_value(value_str)],
    };

    Some(PostgrestFilter {
        column: normalized_column,
        operator,
        values,
        negated,
    })
}

fn parse_in_values(value: &str) -> Vec<Value> {
    let trimmed: &str = value.trim().trim_start_matches('(').trim_end_matches(')');
    trimmed
        .split(',')
        .map(|part| parse_scalar_value(part.trim()))
        .collect()
}

fn parse_array_filter(value: &str) -> Value {
    let trimmed: &str = value.trim().trim_start_matches('.').trim();
    let inner: &str = trimmed.trim_start_matches('{').trim_end_matches('}').trim();
    let elements: Vec<Value> = inner
        .split(',')
        .map(|part| parse_scalar_value(part.trim()))
        .collect();
    Value::Array(elements)
}

fn parse_scalar_value(value: &str) -> Value {
    let lowered: String = value.to_lowercase();
    if lowered.is_empty() {
        return Value::String(String::new());
    }

    if lowered == "null" {
        return Value::Null;
    }

    if lowered == "true" {
        return Value::Bool(true);
    }

    if lowered == "false" {
        return Value::Bool(false);
    }

    if let Ok(int_value) = value.parse::<i64>() {
        return Value::Number(int_value.into());
    }

    if let Ok(float_value) = value.parse::<f64>()
        && let Some(number) = serde_json::Number::from_f64(float_value)
    {
        return Value::Number(number);
    }

    Value::String(value.replace('*', "%"))
}

fn parse_range_header_value(header_value: &str) -> Option<(i64, i64)> {
    let cleaned: &str = header_value.trim().trim_start_matches("items=");
    let mut parts = cleaned.split('-');
    let start: i64 = parts.next()?.trim().parse::<i64>().ok()?;
    let end: i64 = parts.next()?.trim().parse::<i64>().ok()?;
    Some((start, end))
}

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

    #[test]
    fn parse_postgrest_query_parses_select_filters_order_and_range() {
        let parsed = parse_postgrest_query(
            "select=id,name&active=eq.true&score=gte.42&order=created_at.desc",
            Some("items=10-24"),
            false,
        )
        .expect("query");

        assert_eq!(parsed.columns, vec!["id", "name"]);
        assert_eq!(parsed.limit, Some(15));
        assert_eq!(parsed.offset, Some(10));
        assert_eq!(
            parsed.order,
            Some(OrderSpec {
                column: "created_at".to_string(),
                ascending: false,
            })
        );
        assert_eq!(parsed.filters.len(), 2);
        assert_eq!(parsed.filters[0].column, "active");
        assert_eq!(parsed.filters[1].values[0], Value::Number(42.into()));
    }

    #[test]
    fn parse_postgrest_query_normalizes_columns_when_forced() {
        let parsed = parse_postgrest_query("organizationId=eq.123&order=createdAt.asc", None, true)
            .expect("query");

        assert_eq!(parsed.filters[0].column, "organization_id");
        assert_eq!(
            parsed.order,
            Some(OrderSpec {
                column: "created_at".to_string(),
                ascending: true,
            })
        );
    }

    #[test]
    fn parse_postgrest_query_supports_or_and_array_filters() {
        let parsed = parse_postgrest_query(
            "or=(status.eq.active,status.eq.pending)&tags=cs.{alpha,beta}",
            None,
            false,
        )
        .expect("query");

        assert_eq!(parsed.or_filters.len(), 1);
        assert_eq!(parsed.or_filters[0].len(), 2);
        assert_eq!(
            parsed.filters[0].operator,
            PostgrestFilterOperator::Contains
        );
        assert_eq!(
            parsed.filters[0].values,
            vec![Value::Array(vec![
                Value::String("alpha".to_string()),
                Value::String("beta".to_string()),
            ])]
        );
    }

    #[test]
    fn convert_postgrest_filters_to_conditions_preserves_operator_and_pg_cast() {
        let filters = vec![PostgrestFilter {
            column: "cache_hit_ratio".to_string(),
            operator: PostgrestFilterOperator::Gte,
            values: vec![Value::String("0.75".to_string())],
            negated: false,
        }];
        let column_types = HashMap::from([(
            "cache_hit_ratio".to_string(),
            "double precision|float8".to_string(),
        )]);

        let conditions =
            convert_postgrest_filters_to_conditions(&filters, true, Some(&column_types));

        assert_eq!(conditions.len(), 1);
        assert_eq!(conditions[0].operator, ConditionOperator::Gte);
        assert_eq!(conditions[0].pg_cast.as_deref(), Some("float8"));
        assert!(conditions[0].auto_cast_uuid_value_to_text);
    }
}