athena_rs 3.3.0

//! Translators that turn query builders into backend SQL/CQL.
use crate::client::backend::{BackendError, BackendResult, PostgrestMethod, TranslatedQuery};
use crate::client::gateway_api::{
    GatewayDeleteRequest, GatewayFetchRequest, GatewayInsertRequest, GatewayRequest,
    GatewayRequestPayload, GatewayUpdateRequest,
};
use crate::client::query_builder::{Condition, ConditionOperator, OrderDirection};
use serde_json::Value;
use std::fmt::Write;
use std::iter::Peekable;
use supabase_rs::query::Query;

/// ## `QueryTranslator`    
/// Translator trait that converts builder metadata into SQL/CQL and parameters.
///
/// # Arguments
///
/// * `query` - The query to translate.
///
/// # Returns
///
/// A `BackendResult` containing the translated query.
pub(crate) trait QueryTranslator {
    /// Translate a typed gateway request to a backend-specific query shape.
    fn translate_request(&self, request: &GatewayRequest) -> BackendResult<TranslatedQuery>;
}

/// SQL translator suitable for PostgreSQL/Supabase/Neon.
///
/// # Arguments
///
/// * `query` - The query to translate.
///
/// # Returns
///
/// A `BackendResult` containing the translated query.
pub struct SqlTranslator;

/// PostgREST translator suitable for Supabase backend (REST).
///
/// # Arguments
///
/// * `query` - The query to translate.
///
/// # Returns
///
/// A `BackendResult` containing the translated query.
pub struct PostgrestTranslator;

/// CQL translator suitable for ScyllaDB.
///
/// # Arguments
///
/// * `query` - The query to translate.
///
/// # Returns
///
/// A `BackendResult` containing the translated query.
pub struct CqlTranslator;

impl QueryTranslator for SqlTranslator {
    fn translate_request(&self, request: &GatewayRequest) -> BackendResult<TranslatedQuery> {
        match request.payload() {
            GatewayRequestPayload::Fetch(query) => translate_sql_fetch(query),
            GatewayRequestPayload::Insert(query) => translate_sql_insert(query),
            GatewayRequestPayload::Update(query) => translate_sql_update(query),
            GatewayRequestPayload::Delete(query) => translate_sql_delete(query),
            GatewayRequestPayload::Sql(query) => {
                Ok(TranslatedQuery::sql(&query.query, Vec::new(), None))
            }
            GatewayRequestPayload::Rpc(_) => Err(BackendError::Generic(
                "SQL translator does not support RPC requests".to_string(),
            )),
        }
    }
}

impl QueryTranslator for CqlTranslator {
    fn translate_request(&self, request: &GatewayRequest) -> BackendResult<TranslatedQuery> {
        match request.payload() {
            GatewayRequestPayload::Fetch(query) => translate_cql_fetch(query),
            GatewayRequestPayload::Insert(query) => translate_cql_insert(query),
            GatewayRequestPayload::Update(query) => translate_cql_update(query),
            GatewayRequestPayload::Delete(query) => translate_cql_delete(query),
            GatewayRequestPayload::Sql(query) => {
                Ok(TranslatedQuery::cql(&query.query, Vec::new(), None))
            }
            GatewayRequestPayload::Rpc(_) => Err(BackendError::Generic(
                "CQL translator does not support RPC requests".to_string(),
            )),
        }
    }
}

impl QueryTranslator for PostgrestTranslator {
    fn translate_request(&self, request: &GatewayRequest) -> BackendResult<TranslatedQuery> {
        match request.payload() {
            GatewayRequestPayload::Fetch(query) => translate_postgrest_fetch(query),
            GatewayRequestPayload::Insert(query) => translate_postgrest_insert(query),
            GatewayRequestPayload::Update(query) => translate_postgrest_update(query),
            GatewayRequestPayload::Delete(query) => translate_postgrest_delete(query),
            GatewayRequestPayload::Sql(_) => Err(BackendError::Generic(
                "PostgREST translator does not support raw SQL requests".to_string(),
            )),
            GatewayRequestPayload::Rpc(_) => Err(BackendError::Generic(
                "PostgREST translator does not support RPC requests".to_string(),
            )),
        }
    }
}

fn translate_sql_fetch(query: &GatewayFetchRequest) -> BackendResult<TranslatedQuery> {
    let select_nodes: Vec<SelectNode> = parse_select(&query.raw_select, &query.columns)?;
    let mut emitter: SqlEmitter = SqlEmitter::default();
    emitter.emit_root(&query.table, select_nodes, query)
}

fn translate_sql_insert(query: &GatewayInsertRequest) -> BackendResult<TranslatedQuery> {
    let (columns, values, params) = build_insert_fragments(&query.payload);

    let sql: String = format!(
        "INSERT INTO {} ({}) VALUES ({})",
        query.table,
        columns.join(", "),
        values.join(", ")
    );

    Ok(TranslatedQuery::sql(sql, params, Some(query.table.clone())))
}

fn translate_sql_update(query: &GatewayUpdateRequest) -> BackendResult<TranslatedQuery> {
    let (assignments, mut params) = build_update_fragments(&query.payload);
    let mut sql: String = format!("UPDATE {} SET {}", query.table, assignments.join(", "));
    let conditions: Vec<Condition> =
        combined_conditions(&query.scope.conditions, &query.scope.row_id);
    let (where_clause, where_params) = build_where_clause(&conditions);
    sql.push_str(&where_clause);
    params.extend(where_params);

    Ok(TranslatedQuery::sql(sql, params, Some(query.table.clone())))
}

fn translate_sql_delete(query: &GatewayDeleteRequest) -> BackendResult<TranslatedQuery> {
    let mut sql: String = format!("DELETE FROM {}", query.table);
    let conditions: Vec<Condition> = combined_conditions(&query.conditions, &query.row_id);
    let (where_clause, params) = build_where_clause(&conditions);
    sql.push_str(&where_clause);

    Ok(TranslatedQuery::sql(sql, params, Some(query.table.clone())))
}

fn translate_cql_fetch(query: &GatewayFetchRequest) -> BackendResult<TranslatedQuery> {
    let columns: String = if query.columns.is_empty() {
        "*".to_string()
    } else {
        query.columns.join(", ")
    };

    let mut sql: String = format!("SELECT {} FROM {}", columns, query.table);
    let (where_clause, params) = build_where_clause(&query.conditions);
    sql.push_str(&where_clause);

    if let Some(limit) = query.limit {
        sql.push_str(&format!(" LIMIT {}", limit));
    }

    Ok(TranslatedQuery::cql(sql, params, Some(query.table.clone())))
}

fn translate_cql_insert(query: &GatewayInsertRequest) -> BackendResult<TranslatedQuery> {
    let (columns, values, params) = build_insert_fragments(&query.payload);

    let sql: String = format!(
        "INSERT INTO {} ({}) VALUES ({})",
        query.table,
        columns.join(", "),
        values.join(", ")
    );

    Ok(TranslatedQuery::cql(sql, params, Some(query.table.clone())))
}

fn translate_cql_update(query: &GatewayUpdateRequest) -> BackendResult<TranslatedQuery> {
    let (assignments, mut params) = build_update_fragments(&query.payload);
    let mut sql: String = format!("UPDATE {} SET {}", query.table, assignments.join(", "));
    let conditions: Vec<Condition> =
        combined_conditions(&query.scope.conditions, &query.scope.row_id);
    let (where_clause, where_params) = build_where_clause(&conditions);
    sql.push_str(&where_clause);
    params.extend(where_params);

    Ok(TranslatedQuery::cql(sql, params, Some(query.table.clone())))
}

fn translate_cql_delete(query: &GatewayDeleteRequest) -> BackendResult<TranslatedQuery> {
    let mut sql: String = format!("DELETE FROM {}", query.table);
    let conditions: Vec<Condition> = combined_conditions(&query.conditions, &query.row_id);
    let (where_clause, params) = build_where_clause(&conditions);
    sql.push_str(&where_clause);

    Ok(TranslatedQuery::cql(sql, params, Some(query.table.clone())))
}

fn translate_postgrest_fetch(query: &GatewayFetchRequest) -> BackendResult<TranslatedQuery> {
    let mut q: Query = Query::new();
    let mut params: Vec<Value> = Vec::new();
    let select: String = if let Some(raw) = &query.raw_select {
        raw.clone()
    } else if query.columns.is_empty() {
        "*".to_string()
    } else {
        query.columns.join(",")
    };
    q.add_param("select", &select);

    for condition in &query.conditions {
        let encoded: String = postgrest_encoded_condition(condition)?;
        q.add_param(&condition.column, &encoded);
        params.push(postgrest_condition_param_value(condition)?);
    }

    for (column, direction) in &query.order_by {
        let dir_str: &str = match direction {
            OrderDirection::Asc => "asc",
            OrderDirection::Desc => "desc",
        };
        q.add_param("order", &format!("{column}.{dir_str}"));
    }

    if let Some(limit) = query.limit {
        q.add_param("limit", &limit.to_string());
    }

    if let Some(offset) = query.offset {
        q.add_param("offset", &offset.to_string());
    }

    Ok(
        TranslatedQuery::postgrest(q.build(), params, Some(query.table.clone()))
            .with_postgrest_method(PostgrestMethod::Get),
    )
}

fn translate_postgrest_insert(query: &GatewayInsertRequest) -> BackendResult<TranslatedQuery> {
    Ok(
        TranslatedQuery::postgrest("", Vec::new(), Some(query.table.clone()))
            .with_postgrest_method(PostgrestMethod::Post)
            .with_payload(query.payload.clone()),
    )
}

fn translate_postgrest_update(query: &GatewayUpdateRequest) -> BackendResult<TranslatedQuery> {
    let conditions: Vec<Condition> =
        combined_conditions(&query.scope.conditions, &query.scope.row_id);
    if conditions.is_empty() {
        return Err(BackendError::Generic(
            "PostgREST update requires at least one filter condition".to_string(),
        ));
    }

    let mut q: Query = Query::new();
    let mut params: Vec<Value> = Vec::new();
    for condition in &conditions {
        let encoded: String = postgrest_encoded_condition(condition)?;
        q.add_param(&condition.column, &encoded);
        params.push(postgrest_condition_param_value(condition)?);
    }

    Ok(
        TranslatedQuery::postgrest(q.build(), params, Some(query.table.clone()))
            .with_postgrest_method(PostgrestMethod::Patch)
            .with_payload(query.payload.clone()),
    )
}

fn translate_postgrest_delete(query: &GatewayDeleteRequest) -> BackendResult<TranslatedQuery> {
    let conditions: Vec<Condition> = combined_conditions(&query.conditions, &query.row_id);
    if conditions.is_empty() {
        return Err(BackendError::Generic(
            "PostgREST delete requires at least one filter condition".to_string(),
        ));
    }

    let mut q: Query = Query::new();
    let mut params: Vec<Value> = Vec::new();
    for condition in &conditions {
        let encoded: String = postgrest_encoded_condition(condition)?;
        q.add_param(&condition.column, &encoded);
        params.push(postgrest_condition_param_value(condition)?);
    }

    Ok(
        TranslatedQuery::postgrest(q.build(), params, Some(query.table.clone()))
            .with_postgrest_method(PostgrestMethod::Delete),
    )
}

/// ## `build_where_clause`
/// Build a where clause.
///
/// # Arguments
///
/// * `conditions` - The conditions.
///
/// # Returns
///
/// A `(String, Vec<Value>)` containing the where clause and parameters.
///
fn combined_conditions(conditions: &[Condition], row_id: &Option<String>) -> Vec<Condition> {
    let mut merged: Vec<Condition> = conditions.to_vec();
    if let Some(row_id) = row_id {
        merged.push(Condition::new(
            "id",
            ConditionOperator::Eq,
            vec![Value::String(row_id.clone())],
        ));
    }
    merged
}

fn build_where_clause(conditions: &[Condition]) -> (String, Vec<Value>) {
    if conditions.is_empty() {
        return (String::new(), Vec::new());
    }

    let mut clauses: Vec<String> = Vec::new();
    let mut params: Vec<Value> = Vec::new();

    for condition in conditions {
        match condition.operator {
            ConditionOperator::In => {
                let formatted: Vec<String> = condition.values.iter().map(format_value).collect();
                clauses.push(format!(
                    "{} IN ({})",
                    condition.column,
                    formatted.join(", ")
                ));
                params.extend(condition.values.clone());
            }
            _ => {
                if let Some(value) = condition.values.first() {
                    let operator = match condition.operator {
                        ConditionOperator::Eq => "=",
                        ConditionOperator::Neq => "<>",
                        ConditionOperator::Gt => ">",
                        ConditionOperator::Lt => "<",
                        _ => "=",
                    };

                    clauses.push(format!(
                        "{} {} {}",
                        condition.column,
                        operator,
                        format_value(value)
                    ));
                    params.extend(condition.values.clone());
                }
            }
        }
    }

    (format!(" WHERE {}", clauses.join(" AND ")), params)
}

/// ## `build_insert_fragments`
/// Build insert fragments.
///
/// # Arguments
///
/// * `payload` - The payload.
///
/// # Returns
///
/// A `(Vec<String>, Vec<String>, Vec<Value>)` containing the columns, values and parameters.
///
fn build_insert_fragments(payload: &Value) -> (Vec<String>, Vec<String>, Vec<Value>) {
    if let Value::Object(map) = payload {
        let mut columns: Vec<String> = Vec::new();
        let mut values: Vec<String> = Vec::new();
        let mut params: Vec<Value> = Vec::new();

        for (column, value) in map {
            columns.push(column.clone());
            values.push(format_value(value));
            params.push(value.clone());
        }

        (columns, values, params)
    } else {
        (Vec::new(), Vec::new(), Vec::new())
    }
}

/// ## `build_update_fragments`
/// Build update fragments.
///
/// # Arguments
///
/// * `payload` - The payload.
///
/// # Returns
///
/// A `(Vec<String>, Vec<Value>)` containing the assignments and parameters.
///
fn build_update_fragments(payload: &Value) -> (Vec<String>, Vec<Value>) {
    if let Value::Object(map) = payload {
        let mut assignments: Vec<String> = Vec::new();
        let mut params: Vec<Value> = Vec::new();

        for (column, value) in map {
            assignments.push(format!("{} = {}", column, format_value(value)));
            params.push(value.clone());
        }

        (assignments, params)
    } else {
        (Vec::new(), Vec::new())
    }
}

/// ## `format_value`
/// Format a value.
///
/// # Arguments
///
/// * `value` - The value.
///
/// # Returns
///
/// A `String` containing the formatted value.
///
fn format_value(value: &Value) -> String {
    match value {
        Value::String(text) => format!("'{}'", text.replace('\'', "''")),
        Value::Number(num) => num.to_string(),
        Value::Bool(flag) => flag.to_string(),
        Value::Null => "NULL".to_string(),
        other => serde_json::to_string(other).unwrap_or_else(|_| "NULL".to_string()),
    }
}

/// ## `postgrest_operator`
/// PostgREST operator.
///
/// # Arguments
///
/// * `op` - The operator.
/// * `value` - The value.
///
/// # Returns
///
/// A `BackendResult` containing the encoded operator.
///
///
fn postgrest_operator(op: ConditionOperator, value: &Value) -> BackendResult<String> {
    let val: String = match value {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    };
    let encoded: String = match op {
        ConditionOperator::Eq => format!("eq.{val}"),
        ConditionOperator::Neq => format!("neq.{val}"),
        ConditionOperator::Gt => format!("gt.{val}"),
        ConditionOperator::Lt => format!("lt.{val}"),
        ConditionOperator::In => format!("in.({val})"),
    };
    Ok(encoded)
}

/// ## `postgrest_encoded_condition`
/// PostgREST encoded condition.
///
/// # Arguments
///
/// * `condition` - The condition.
///
/// # Returns
///
/// A `BackendResult` containing the encoded condition.
///
fn postgrest_encoded_condition(condition: &Condition) -> BackendResult<String> {
    match condition.operator {
        ConditionOperator::In => {
            let joined: String = condition
                .values
                .iter()
                .map(|value| match value {
                    Value::String(s) => s.clone(),
                    other => other.to_string(),
                })
                .collect::<Vec<String>>()
                .join(",");
            postgrest_operator(ConditionOperator::In, &Value::String(joined))
        }
        _ => {
            let value: &Value = condition.values.first().ok_or_else(|| {
                BackendError::Generic(format!(
                    "missing value for condition on column '{}'",
                    condition.column
                ))
            })?;
            postgrest_operator(condition.operator, value)
        }
    }
}

/// ## `postgrest_condition_param_value`
/// PostgREST condition parameter value.
///
/// # Arguments
///
/// * `condition` - The condition.
///
/// # Returns
///
/// A `BackendResult` containing the condition parameter value.
///
fn postgrest_condition_param_value(condition: &Condition) -> BackendResult<Value> {
    match condition.operator {
        ConditionOperator::In => Ok(Value::Array(condition.values.clone())),
        _ => condition.values.first().cloned().ok_or_else(|| {
            BackendError::Generic(format!(
                "missing value for condition on column '{}'",
                condition.column
            ))
        }),
    }
}

/// ## `JoinKind`
/// Join kind.
///
/// # Arguments
///
/// * `Left` - The left join.
/// * `Inner` - The inner join.
///
/// # Returns
#[derive(Debug, Clone, PartialEq, Eq)]
enum JoinKind {
    Left,
    Inner,
}

/// ## `SelectNode`
/// Select node.
///
/// # Arguments
///
/// * `Column` - The column.
/// * `Relation` - The relation.
///
/// # Returns
#[derive(Debug, Clone)]
enum SelectNode {
    Column(String),
    Relation(RelationNode),
}

/// ## `RelationNode`
/// Relation node.
///
/// # Arguments
///
/// * `name` - The name.
/// * `alias` - The alias.
/// * `join` - The join.
/// * `foreign_key` - The foreign key.
/// # Returns
///
/// A `RelationNode` containing the relation node.
///
#[derive(Debug, Clone)]
struct RelationNode {
    name: String,
    alias: Option<String>,
    join: JoinKind,
    foreign_key: Option<String>,
    children: Vec<SelectNode>,
    conditions: Vec<Condition>,
}

impl RelationNode {
    /// ## `display_name`
    /// Display name.
    ///
    /// # Arguments
    ///
    /// * `self` - The relation node.
    ///
    /// # Returns
    ///
    /// A `&str` containing the display name.
    ///
    fn display_name(&self) -> &str {
        self.alias.as_deref().unwrap_or(&self.name)
    }
}

/// ## `parse_select`
/// Parse a select.
///
/// # Arguments
///
/// * `raw` - The raw select.
/// * `columns` - The columns.
///
/// # Returns
///
/// A `BackendResult` containing the parsed select.
///
fn parse_select(raw: &Option<String>, columns: &[String]) -> BackendResult<Vec<SelectNode>> {
    if let Some(raw) = raw {
        parse_select_string(raw)
    } else {
        Ok(columns
            .iter()
            .map(|c| SelectNode::Column(c.clone()))
            .collect())
    }
}

/// ## `parse_select_string`
/// Parse a select string.
///
/// # Arguments
///
/// * `input` - The input.
///
/// # Returns
///
/// A `BackendResult` containing the parsed select.
///
fn parse_select_string(input: &str) -> BackendResult<Vec<SelectNode>> {
    let mut chars = input.chars().peekable();
    parse_nodes(&mut chars)
}

/// ## `parse_nodes`
/// Parse nodes.
///
/// # Arguments
///
/// * `iter` - The iterator.
///
/// # Returns
///
/// A `BackendResult` containing the parsed nodes.
///
fn parse_nodes<I>(iter: &mut Peekable<I>) -> BackendResult<Vec<SelectNode>>
where
    I: Iterator<Item = char>,
{
    let mut nodes: Vec<SelectNode> = Vec::new();
    loop {
        skip_ws(iter);
        if matches!(iter.peek(), Some(')')) || iter.peek().is_none() {
            break;
        }
        nodes.push(parse_node(iter)?);
        skip_ws(iter);
        if matches!(iter.peek(), Some(',')) {
            iter.next();
            continue;
        }
        if matches!(iter.peek(), Some(')')) || iter.peek().is_none() {
            break;
        }
    }
    Ok(nodes)
}

/// ## `parse_node`
/// Parse a node.
///
/// # Arguments
///
/// * `iter` - The iterator.
///
/// # Returns
///
/// A `BackendResult` containing the parsed node.
///
fn parse_node<I>(iter: &mut Peekable<I>) -> BackendResult<SelectNode>
where
    I: Iterator<Item = char>,
{
    let mut token: String = String::new();
    while let Some(&ch) = iter.peek() {
        if matches!(ch, '!' | ':' | '(' | ')' | ',') {
            break;
        }
        token.push(ch);
        iter.next();
    }
    let token: String = token.trim().to_string();
    if token.is_empty() {
        return Err(BackendError::Generic(
            "invalid select syntax: empty token".to_string(),
        ));
    }

    let mut alias: Option<String> = None;
    let mut name_owned: String = token.clone();

    if matches!(iter.peek(), Some(':')) {
        iter.next();
        let mut relation: String = String::new();
        while let Some(&ch) = iter.peek() {
            if matches!(ch, '!' | '(' | ')' | ',') {
                break;
            }
            relation.push(ch);
            iter.next();
        }
        alias = Some(token.clone());
        name_owned = relation;
    }
    let name: &str = name_owned.as_str();

    let mut join: JoinKind = JoinKind::Left;
    let mut foreign_key: Option<String> = None;

    if matches!(iter.peek(), Some('!')) {
        iter.next();
        let mut modifier: String = String::new();
        while let Some(&ch) = iter.peek() {
            if matches!(ch, '(' | ')' | ',') {
                break;
            }
            modifier.push(ch);
            iter.next();
        }
        let modifier: &str = modifier.trim();
        if modifier.eq_ignore_ascii_case("inner") {
            join = JoinKind::Inner;
        } else if !modifier.is_empty() {
            foreign_key = Some(modifier.to_string());
        }
    }

    if matches!(iter.peek(), Some('(')) {
        iter.next();
        let children: Vec<SelectNode> = parse_nodes(iter)?;
        if iter.next() != Some(')') {
            return Err(BackendError::Generic(
                "invalid select syntax: missing ')'".to_string(),
            ));
        }
        Ok(SelectNode::Relation(RelationNode {
            name: name.to_string(),
            alias,
            join,
            foreign_key,
            children,
            conditions: Vec::new(),
        }))
    } else {
        Ok(SelectNode::Column(name.to_string()))
    }
}

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

/// Attach dotted conditions (`relation.column`) to the corresponding relation node tree.
fn distribute_conditions(nodes: &mut [SelectNode], conditions: &[Condition]) -> Vec<Condition> {
    let mut remaining: Vec<Condition> = Vec::new();
    for cond in conditions {
        if let Some((head, rest)) = cond.column.split_once('.')
            && let Some(node) = nodes.iter_mut().find_map(|n| match n {
                SelectNode::Relation(r) if r.display_name() == head => Some(r),
                _ => None,
            })
        {
            push_condition(node, rest, cond.clone());
            continue;
        }
        remaining.push(cond.clone());
    }
    remaining
}

/// ## `push_condition`
/// Push a condition.
///
/// # Arguments
///
/// * `target` - The target.
/// * `path` - The path.
/// * `condition` - The condition.
///
/// # Returns
///
/// A `BackendResult` containing the pushed condition.
///
fn push_condition(target: &mut RelationNode, path: &str, condition: Condition) {
    if let Some((head, rest)) = path.split_once('.') {
        if let Some(child) = target.children.iter_mut().find_map(|n| match n {
            SelectNode::Relation(r) if r.display_name() == head => Some(r),
            _ => None,
        }) {
            push_condition(child, rest, condition);
        }
    } else {
        target.conditions.push(condition);
    }
}

/// ## `SqlEmitter`
/// SQL emitter.
///
/// # Arguments
///
/// * `alias_counter` - The alias counter.
///
/// # Returns
///
/// A `SqlEmitter` containing the SQL emitter.
///
#[derive(Default)]
struct SqlEmitter {
    alias_counter: usize,
}

impl SqlEmitter {
    /// ## `emit_root`
    /// Emit the root.
    ///
    /// # Arguments
    ///
    /// * `table` - The table.
    /// * `nodes` - The nodes.
    /// * `query` - The query.
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the emitted root.
    ///
    fn emit_root(
        &mut self,
        table: &str,
        mut nodes: Vec<SelectNode>,
        query: &GatewayFetchRequest,
    ) -> BackendResult<TranslatedQuery> {
        let base_alias: String = "t0".to_string();
        let base_conditions: Vec<Condition> = distribute_conditions(&mut nodes, &query.conditions);

        let mut select_exprs: Vec<String> = Vec::new();
        let mut joins: Vec<String> = Vec::new();

        for node in nodes {
            self.emit_node(&node, &base_alias, table, &mut select_exprs, &mut joins)?;
        }

        if select_exprs.is_empty() {
            select_exprs.push("*".to_string());
        }

        let mut sql: String = format!(
            "SELECT {select} FROM {table} {alias}",
            select = select_exprs.join(", "),
            table = table,
            alias = base_alias
        );

        if !joins.is_empty() {
            sql.push('\n');
            sql.push_str(&joins.join("\n"));
        }

        if !base_conditions.is_empty() {
            let adapted: Vec<Condition> = base_conditions
                .iter()
                .cloned()
                .map(|mut c| {
                    c.column = format!("{base_alias}.{}", c.column);
                    c
                })
                .collect();
            let (where_clause, _) = build_where_clause(&adapted);
            sql.push_str(&where_clause);
        }

        if !query.order_by.is_empty() {
            let ordering: Vec<String> = query
                .order_by
                .iter()
                .map(|(column, direction)| {
                    let dir_str: &str = match direction {
                        OrderDirection::Asc => "ASC",
                        OrderDirection::Desc => "DESC",
                    };

                    if column.contains('.') {
                        format!("{column} {dir_str}")
                    } else {
                        format!("{base_alias}.{column} {dir_str}")
                    }
                })
                .collect();
            sql.push_str(&format!(" ORDER BY {}", ordering.join(", ")));
        }

        if let Some(limit) = query.limit {
            sql.push_str(&format!(" LIMIT {}", limit));
        }

        if let Some(offset) = query.offset {
            sql.push_str(&format!(" OFFSET {}", offset));
        }

        Ok(TranslatedQuery::sql(
            sql,
            Vec::new(),
            Some(table.to_string()),
        ))
    }

    /// ## `emit_node`
    /// Emit a node.
    ///
    /// # Arguments
    ///
    /// * `node` - The node.
    /// * `parent_alias` - The parent alias.
    /// * `parent_table` - The parent table.
    /// * `select_exprs` - The select expressions.
    /// * `joins` - The joins.
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the emitted node.
    ///
    fn emit_node(
        &mut self,
        node: &SelectNode,
        parent_alias: &str,
        parent_table: &str,
        select_exprs: &mut Vec<String>,
        joins: &mut Vec<String>,
    ) -> BackendResult<()> {
        match node {
            SelectNode::Column(col) => {
                if col == "*" {
                    select_exprs.push(format!("{parent_alias}.*"));
                } else {
                    select_exprs.push(format!("{parent_alias}.{col}"));
                }
            }
            SelectNode::Relation(rel) => {
                let alias: String = self.next_alias(&rel.name)?;
                let agg_alias: String = format!("{alias}_agg");

                let mut relation_clone: RelationNode = rel.clone();
                let _ = distribute_conditions(&mut relation_clone.children, &rel.conditions);
                let (row_expr, child_joins) = self.emit_relation_row(&relation_clone, &alias)?;

                let (join_condition, many_to_one) =
                    relation_join_condition(&relation_clone, parent_alias, parent_table, &alias);

                let mut subquery: String = String::new();

                if many_to_one {
                    writeln!(
                        subquery,
                        "SELECT {row_expr} AS data FROM {table} {alias}",
                        table = relation_clone.name,
                        alias = alias,
                        row_expr = row_expr
                    )
                    .unwrap();
                } else {
                    write!(
                        subquery,
                        "SELECT COALESCE(jsonb_agg({row_expr} ORDER BY {alias}.id) FILTER (WHERE {alias}.id IS NOT NULL), '[]'::jsonb) AS data\nFROM {table} {alias}\n",
                        row_expr = row_expr,
                        alias = alias,
                        table = relation_clone.name
                    )
                    .unwrap();
                }

                if !child_joins.is_empty() {
                    subquery.push_str(&child_joins.join("\n"));
                    subquery.push('\n');
                }

                write!(subquery, "WHERE {join_condition}").unwrap();

                if !relation_clone.conditions.is_empty() {
                    let adapted: Vec<Condition> = relation_clone
                        .conditions
                        .iter()
                        .cloned()
                        .map(|mut c| {
                            c.column = format!("{alias}.{}", c.column);
                            c
                        })
                        .collect();
                    let (where_clause, _) = build_where_clause(&adapted);
                    subquery.push_str(&where_clause);
                }

                if many_to_one {
                    subquery.push_str(" LIMIT 1");
                }

                let join_type: &str = match rel.join {
                    JoinKind::Left => "LEFT JOIN",
                    JoinKind::Inner => "JOIN",
                };

                let mut join_sql: String =
                    format!("{join_type} LATERAL (\n{subquery}\n) {agg_alias} ON ");

                if rel.join == JoinKind::Inner {
                    join_sql.push_str(&format!("{agg_alias}.data IS NOT NULL"));
                } else {
                    join_sql.push_str("TRUE");
                }

                joins.push(join_sql);
                select_exprs.push(format!("{} as {}", agg_alias, rel.display_name()));
            }
        }
        Ok(())
    }

    /// ## `emit_relation_row`
    /// Emit a relation row.
    ///
    /// # Arguments
    ///
    /// * `relation` - The relation.
    /// * `alias` - The alias.
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the emitted relation row.
    ///
    fn emit_relation_row(
        &mut self,
        relation: &RelationNode,
        alias: &str,
    ) -> BackendResult<(String, Vec<String>)> {
        let mut field_pairs: Vec<String> = Vec::new();
        let mut joins: Vec<String> = Vec::new();

        for child in &relation.children {
            match child {
                SelectNode::Column(col) => {
                    field_pairs.push(format!("'{}', {}.{}", col, alias, col));
                }

                SelectNode::Relation(rel) => {
                    let child_alias: String = self.next_alias(&rel.name)?;
                    let mut rel_clone: RelationNode = rel.clone();
                    let _ = distribute_conditions(&mut rel_clone.children, &rel.conditions);
                    let (row_expr, nested_joins) =
                        self.emit_relation_row(&rel_clone, &child_alias)?;

                    let (join_condition, many_to_one) =
                        relation_join_condition(&rel_clone, alias, &relation.name, &child_alias);

                    let mut subquery: String = String::new();
                    if many_to_one {
                        writeln!(
                            subquery,
                            "SELECT {row_expr} AS data FROM {table} {child_alias}",
                            row_expr = row_expr,
                            table = rel_clone.name,
                            child_alias = child_alias
                        )
                        .unwrap();
                    } else {
                        write!(
                            subquery,
                            "SELECT COALESCE(jsonb_agg({row_expr} ORDER BY {child_alias}.id) FILTER (WHERE {child_alias}.id IS NOT NULL), '[]'::jsonb) AS data\nFROM {table} {child_alias}\n",
                            row_expr = row_expr,
                            child_alias = child_alias,
                            table = rel_clone.name
                        )
                        .unwrap();
                    }

                    if !nested_joins.is_empty() {
                        subquery.push_str(&nested_joins.join("\n"));
                        subquery.push('\n');
                    }

                    write!(subquery, "WHERE {join_condition}").unwrap();

                    if !rel_clone.conditions.is_empty() {
                        let adapted: Vec<Condition> = rel_clone
                            .conditions
                            .iter()
                            .cloned()
                            .map(|mut c| {
                                c.column = format!("{child_alias}.{}", c.column);
                                c
                            })
                            .collect();
                        let (where_clause, _) = build_where_clause(&adapted);
                        subquery.push_str(&where_clause);
                    }

                    if many_to_one {
                        subquery.push_str(" LIMIT 1");
                    }

                    let join_type: &str = match rel.join {
                        JoinKind::Left => "LEFT JOIN",
                        JoinKind::Inner => "JOIN",
                    };

                    let agg_alias: String = format!("{child_alias}_agg");
                    let mut join_sql: String =
                        format!("{join_type} LATERAL (\n{subquery}\n) {agg_alias} ON ");
                    if rel.join == JoinKind::Inner {
                        join_sql.push_str(&format!("{agg_alias}.data IS NOT NULL"));
                    } else {
                        join_sql.push_str("TRUE");
                    }
                    joins.push(join_sql);

                    field_pairs.push(format!("'{}', {agg_alias}.data", rel.display_name()));
                }
            }
        }

        let row_expr: String = if field_pairs.is_empty() {
            format!("to_jsonb({alias})")
        } else {
            format!("jsonb_build_object({})", field_pairs.join(", "))
        };

        Ok((row_expr, joins))
    }

    /// ## `next_alias`
    /// Next alias.
    ///
    /// # Arguments
    ///
    /// * `base` - The base.
    ///
    /// # Returns
    ///
    /// A `String` containing the next alias.
    ///
    /// # Examples
    ///
    /// ```
    /// let mut emitter = SqlEmitter::default();
    /// let alias: String = emitter.next_alias("t");
    /// assert_eq!(alias, "t_0");
    /// ```
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the next alias.
    ///
    /// # Errors
    ///
    /// Returns a `BackendError` if the alias counter exceeds the maximum value.
    ///
    /// # Examples
    ///
    /// ```
    /// let mut emitter = SqlEmitter::default();
    /// let alias: String = emitter.next_alias("t").expect("next alias ok");
    /// assert_eq!(alias, "t_0");
    /// ```
    ///
    fn next_alias(&mut self, base: &str) -> BackendResult<String> {
        let alias: String = format!("{}_{}", base, self.alias_counter);
        self.alias_counter += 1;
        Ok(alias)
    }
}

/// ## `relation_join_condition`
/// Determine join condition and whether the relationship should be treated as many-to-one.
///
/// # Arguments
///
/// * `relation` - The relation.
/// * `parent_alias` - The parent alias.
/// * `parent_table` - The parent table.
/// * `child_alias` - The child alias.
/// `child_alias` is the alias used for the relation table in the subquery (e.g. "posts_0").
///
/// # Returns
///
/// A `(String, bool)` containing the join condition and whether the relationship should be treated as many-to-one.
///
fn relation_join_condition(
    relation: &RelationNode,
    parent_alias: &str,
    parent_table: &str,
    child_alias: &str,
) -> (String, bool) {
    if let Some(fk) = &relation.foreign_key {
        let fk_lower: String = fk.to_lowercase();
        if let Some(stripped) = fk_lower.strip_prefix("parent.") {
            return (
                format!("{parent_alias}.{stripped} = {child_alias}.id"),
                true,
            );
        }

        if let Some(stripped) = fk_lower.strip_prefix("child.") {
            return (
                format!("{child_alias}.{stripped} = {parent_alias}.id"),
                false,
            );
        }

        // Explicit FK without prefix: infer placement from naming.
        // If FK name contains the relation (child) table name, assume parent-side FK;
        // otherwise assume child-side FK (e.g. posts!author_id where author_id is on posts).
        let rel_singular: String = relation.name.trim_end_matches('s').to_lowercase();
        let rel_lower: String = relation.name.to_lowercase();
        let fk_suggests_parent_side: bool =
            fk_lower.contains(&rel_lower) || fk_lower.contains(&rel_singular);

        return if fk_suggests_parent_side {
            (format!("{parent_alias}.{fk} = {child_alias}.id"), true)
        } else {
            (format!("{child_alias}.{fk} = {parent_alias}.id"), false)
        };
    }

    let child_fk: String = format!("{}_id", parent_table.trim_end_matches('s'));
    let parent_fk: String = format!("{}_id", relation.name.trim_end_matches('s'));

    // Heuristic: when the relation name is longer (often parent tables), prefer many-to-one.
    let many_to_one: bool = relation.name.len() >= parent_table.len();

    if many_to_one {
        (
            format!("{parent_alias}.{parent_fk} = {child_alias}.id"),
            true,
        )
    } else {
        (
            format!("{child_alias}.{child_fk} = {parent_alias}.id"),
            false,
        )
    }
}

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

    /// ## `translate`
    /// Translate a request.
    ///
    /// # Arguments
    ///
    /// * `translator` - The translator.
    /// * `request` - The request.
    ///
    /// # Returns
    /// A `BackendResult` containing the translated query.
    ///
    /// # Examples
    ///
    /// ```
    /// let translator = SqlTranslator;
    /// let request = GatewayFetchRequest::new("orchestral_sections").raw_select("id,name,instruments(id,name)");
    /// let translated = translate(&translator, request).expect("translate ok");
    /// assert_eq!(translated.language, QueryLanguage::Sql);
    /// assert_eq!(translated.sql, "SELECT id,name,instruments(id,name) FROM orchestral_sections");
    /// ```
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the translated query.
    ///
    /// # Errors
    ///
    /// Returns a `BackendError` if the request cannot be translated.
    ///
    fn translate<T>(
        translator: &T,
        request: impl Into<GatewayRequest>,
    ) -> BackendResult<TranslatedQuery>
    where
        T: QueryTranslator,
    {
        let request: GatewayRequest = request.into();
        translator.translate_request(&request)
    }

    /// ## `parses_nested_supabase_select`
    /// Parses a nested Supabase select.
    ///
    /// # Arguments
    ///
    /// * `select` - The select.
    ///
    /// # Returns
    /// A `Vec<SelectNode>` containing the parsed select nodes.
    ///
    /// # Examples
    ///
    /// ```
    /// let select: &str = "id,name,instruments!inner(id,name),start_scan:scans!scan_id_start(id)";
    /// let nodes: Vec<SelectNode> = parse_select(&Some(select.to_string()), &[]).expect("parse ok");
    /// assert_eq!(nodes.len(), 4);
    /// ```
    ///
    #[test]
    fn parses_nested_supabase_select() {
        let select: &str = "id,name,instruments!inner(id,name),start_scan:scans!scan_id_start(id)";
        let nodes: Vec<SelectNode> =
            parse_select(&Some(select.to_string()), &[]).expect("parse ok");
        assert_eq!(nodes.len(), 4);
        match &nodes[2] {
            SelectNode::Relation(rel) => {
                assert_eq!(rel.name, "instruments");
                assert!(matches!(rel.join, JoinKind::Inner));
                assert_eq!(rel.children.len(), 2);
            }
            _ => panic!("expected relation"),
        }
    }

    /// ## `postgrest_translator_builds_query_string`
    /// PostgREST translator builds query string.
    ///
    /// # Arguments
    ///
    /// * `translator` - The translator.
    /// * `request` - The request.
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the translated query.
    ///
    #[test]
    fn postgrest_translator_builds_query_string() {
        let translator: PostgrestTranslator = PostgrestTranslator;
        let request: GatewayFetchRequest = GatewayFetchRequest::new("orchestral_sections")
            .raw_select("id,name,instruments(id,name)")
            .where_condition(Condition::new(
                "instruments.name",
                ConditionOperator::Eq,
                vec![Value::String("flute".to_string())],
            ))
            .limit(10)
            .offset(5);

        let translated: TranslatedQuery = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Postgrest);
        assert_eq!(translated.postgrest_method, Some(PostgrestMethod::Get));
        assert!(
            translated
                .sql
                .contains("select=id,name,instruments(id,name)")
        );
        assert!(translated.sql.contains("instruments.name=eq.flute"));
        assert!(translated.sql.contains("limit=10"));
        assert!(translated.sql.contains("offset=5"));
    }

    /// ## `postgrest_translator_builds_update_with_filters_and_payload`
    /// PostgREST translator builds update with filters and payload.
    ///
    /// # Arguments
    ///
    /// * `translator` - The translator.
    /// * `request` - The request.
    ///
    /// # Returns
    ///
    /// A `BackendResult` containing the translated query.
    ///
    /// # Examples
    ///
    /// ```
    /// let translator = PostgrestTranslator;
    /// let request = GatewayUpdateRequest::new("users", json!({ "status": "active" })).where_condition(Condition::new("email", ConditionOperator::Eq, vec![Value::String("alice@example.com".to_string())]));
    /// let translated = translate(&translator, request).expect("translate ok");
    /// assert_eq!(translated.language, QueryLanguage::Postgrest);
    /// assert_eq!(translated.postgrest_method, Some(PostgrestMethod::Patch));
    /// assert_eq!(translated.table.as_deref(), Some("users"));
    /// assert!(translated.sql.contains("email=eq.alice@example.com"));
    /// assert_eq!(translated.payload, Some(json!({ "status": "active" })));
    /// ```
    #[test]
    fn postgrest_translator_builds_update_with_filters_and_payload() {
        let translator: PostgrestTranslator = PostgrestTranslator;
        let request: crate::client::gateway_api::GatewayMutationRequest<
            crate::client::GatewayUpdateScope,
        > = GatewayUpdateRequest::new(
            "users",
            json!({
                "status": "active"
            }),
        )
        .where_condition(Condition::new(
            "email",
            ConditionOperator::Eq,
            vec![Value::String("alice@example.com".to_string())],
        ));

        let translated = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Postgrest);
        assert_eq!(translated.postgrest_method, Some(PostgrestMethod::Patch));
        assert_eq!(translated.table.as_deref(), Some("users"));
        assert!(translated.sql.contains("email=eq.alice@example.com"));
        assert_eq!(translated.payload, Some(json!({ "status": "active" })));
    }

    #[test]
    fn postgrest_translator_rejects_unfiltered_delete() {
        let translator = PostgrestTranslator;
        let request = GatewayDeleteRequest::new("users");

        let err = translate(&translator, request).expect_err("delete without filters should fail");
        assert!(
            err.to_string()
                .contains("requires at least one filter condition")
        );
    }

    #[test]
    fn sql_translator_builds_lateral_join() {
        let translator = SqlTranslator;
        let request = GatewayFetchRequest::new("orchestral_sections")
            .raw_select("id,name,instruments(id,name)");

        let translated = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Sql);
        assert!(translated.sql.contains("jsonb_agg"));
        assert!(translated.sql.contains("FROM instruments instruments_0"));
        assert!(translated.sql.contains("instruments_0_agg as instruments"));
    }

    #[test]
    fn sql_translator_child_side_fk_uses_jsonb_agg() {
        // authors -> posts: author_id is on posts (child). One author has many posts.
        // posts!author_id should produce posts.author_id = authors.id and jsonb_agg (not LIMIT 1).
        let translator = SqlTranslator;
        let request =
            GatewayFetchRequest::new("authors").raw_select("id,name,posts!author_id(id,title)");

        let translated = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Sql);
        // Child-side FK: join must be posts_0.author_id = t0.id
        assert!(
            translated.sql.contains("posts_0.author_id = t0.id"),
            "expected child-side join condition, got: {}",
            translated.sql
        );
        // One-to-many: must use jsonb_agg, not LIMIT 1
        assert!(
            translated.sql.contains("jsonb_agg"),
            "expected jsonb_agg for one-to-many, got: {}",
            translated.sql
        );
        assert!(
            !translated.sql.contains("LIMIT 1"),
            "one-to-many should not use LIMIT 1, got: {}",
            translated.sql
        );
    }

    #[test]
    fn sql_update_translator_combines_builder_conditions_and_row_id() {
        let translator = SqlTranslator;
        let request = GatewayUpdateRequest::new(
            "users",
            json!({
                "name": "Alice"
            }),
        )
        .row_id("user_42")
        .where_condition(Condition::new(
            "status",
            ConditionOperator::Eq,
            vec![Value::String("active".to_string())],
        ));

        let translated = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Sql);
        assert_eq!(
            translated.sql,
            "UPDATE users SET name = 'Alice' WHERE status = 'active' AND id = 'user_42'"
        );
    }

    #[test]
    fn sql_delete_translator_supports_condition_filters_without_row_id() {
        let translator = SqlTranslator;
        let request = GatewayDeleteRequest::new("users")
            .where_condition(Condition::new(
                "age",
                ConditionOperator::Gt,
                vec![Value::from(18)],
            ))
            .where_condition(Condition::new(
                "status",
                ConditionOperator::Neq,
                vec![Value::String("archived".to_string())],
            ));

        let translated = translate(&translator, request).expect("translate ok");
        assert_eq!(translated.language, QueryLanguage::Sql);
        assert_eq!(
            translated.sql,
            "DELETE FROM users WHERE age > 18 AND status <> 'archived'"
        );
    }
}