udb 0.4.25

//! Generic SQL compiler core shared by the four relational backends
//! (Postgres, MySQL, SQLite, MSSQL) — the #12 dedup.
//!
//! The four `postgres`/`mysql`/`sqlite`/`mssql` compilers were ~90%
//! identical: the same `resolve_table` / `column_for` / `render_where` /
//! `render_filter` / `render_aggregate` / `resolve_sort_field` /
//! `render_having` / `resolve_having_field` / `sql_op_for` logic, differing
//! only along a handful of dialect axes. Those axes are captured by the
//! [`SqlDialect`] trait; everything structurally identical lives here on
//! [`SqlCompiler`] so each backend keeps only a small `impl SqlDialect`
//! plus its genuinely-divergent statement assembly (upsert idiom,
//! pagination idiom, search surface, DDL).
//!
//! The emitted SQL is byte-identical to the hand-written per-backend code:
//! every quote, placeholder, false-literal, and LIKE-escape idiom routes
//! through a `SqlDialect` hook whose output matches the original.

use crate::generation::ManifestTable;
use crate::ir::filter::{ComparisonOp, LogicalFilter};
use crate::ir::operations::{AggregateExpr, AggregateFunc, LogicalAggregate};
use crate::ir::value::LogicalValue;

use super::CompileError;

/// The per-dialect hooks that distinguish the four relational compilers.
///
/// Only the axes that actually differ live here; everything else is shared
/// on [`SqlCompiler`]. Implementors are zero-sized marker types
/// (`Postgres`, `Mysql`, `Sqlite`, `Mssql`).
pub(super) trait SqlDialect {
    /// Quote an identifier: `"x"` (Postgres/SQLite), `` `x` `` (MySQL),
    /// `[x]` (MSSQL). Returns the full quoted token including delimiters.
    fn quote(ident: &str) -> String;

    /// Render the placeholder for the value that has just been pushed,
    /// where `index` is the 1-based position in the params vec
    /// (`params.len()` after the push). `$N` (Postgres), `?` (MySQL,
    /// SQLite), `@PN` (MSSQL).
    fn placeholder(index: usize) -> String;

    /// The literal a guaranteed-false WHERE body lowers to: `FALSE`
    /// (Postgres/MySQL), `0` (SQLite), `1=0` (MSSQL).
    fn false_literal() -> &'static str;

    /// HAVING-position true / false literals (an empty AND / empty OR).
    /// Postgres/MySQL use `TRUE`/`FALSE`; SQLite `1`/`0`; MSSQL `1=1`/`1=0`.
    fn having_true_literal() -> &'static str;
    fn having_false_literal() -> &'static str;

    /// Transform the RHS of a comparison for substring operators
    /// (`Contains`/`StartsWith`/`EndsWith`) — the LIKE-concat + escape
    /// idiom (`||` vs `CONCAT` vs `+`, plus the per-backend metacharacter
    /// escaping). For all other operators this returns `placeholder`
    /// unchanged. `placeholder` is the already-rendered placeholder token.
    fn wrap_value_for_op(op: ComparisonOp, placeholder: &str) -> String;

    /// Map a comparison operator to its SQL token. The default matches
    /// Postgres (which is the only dialect with a distinct `ILIKE`); the
    /// other three fold `ILike` into `LIKE` by overriding.
    fn sql_op_for(op: ComparisonOp) -> &'static str {
        match op {
            ComparisonOp::Eq => "=",
            ComparisonOp::Ne => "<>",
            ComparisonOp::Lt => "<",
            ComparisonOp::Le => "<=",
            ComparisonOp::Gt => ">",
            ComparisonOp::Ge => ">=",
            ComparisonOp::Like
            | ComparisonOp::Contains
            | ComparisonOp::StartsWith
            | ComparisonOp::EndsWith => "LIKE",
            ComparisonOp::ILike => "ILIKE",
        }
    }

    /// Cast a comparison placeholder to a column's SQL type when the dialect's
    /// operator resolution is type-exact. Postgres has no implicit `uuid = text`
    /// operator, so a `LogicalValue::String` bound for a `UUID` column must be cast
    /// (`$1::UUID`) in WHERE / IN comparisons, or the query fails with
    /// `operator does not exist: uuid = text`. INSERT `VALUES` need NO cast — there
    /// Postgres coerces the input to the known target column type. MySQL / SQLite /
    /// MSSQL coerce in comparisons too, so the default is a no-op and their emitted
    /// SQL stays byte-identical.
    fn cast_compare_placeholder(_column_sql_type: &str, placeholder: &str) -> String {
        placeholder.to_string()
    }
}

/// Generic SQL compiler parameterised over a [`SqlDialect`]. Holds no
/// state — it's a namespace for the shared lowering logic, called by each
/// backend's thin `Compiler` impl.
pub(super) struct SqlCompiler<D: SqlDialect> {
    _dialect: std::marker::PhantomData<D>,
}

impl<D: SqlDialect> SqlCompiler<D> {
    /// Resolve a logical message type to its physical manifest table.
    pub(super) fn resolve_table<'a>(
        message_type: &str,
        manifest: &'a crate::generation::CatalogManifest,
    ) -> Result<&'a ManifestTable, CompileError> {
        match crate::broker::table_lookup(manifest, message_type) {
            crate::broker::TableLookup::Found(table) => Ok(table),
            // fix_plan §4.1: an ambiguous short name names its candidates so the
            // caller can FQN-qualify — never a silent first-wins misroute.
            crate::broker::TableLookup::Ambiguous { .. } => Err(CompileError::Malformed {
                reason: crate::broker::describe_table_lookup_miss(manifest, message_type),
            }),
            crate::broker::TableLookup::Missing => Err(CompileError::UnknownMessageType {
                message_type: message_type.to_string(),
            }),
        }
    }

    /// Map an IR field name to the manifest column name (field-name or
    /// column-name match, case-insensitive on the field name).
    pub(super) fn column_for<'a>(
        table: &'a ManifestTable,
        field: &str,
        message_type: &str,
    ) -> Result<&'a str, CompileError> {
        table
            .columns
            .iter()
            .find(|c| c.field_name.eq_ignore_ascii_case(field) || c.column_name == field)
            .map(|c| c.column_name.as_str())
            .ok_or_else(|| CompileError::UnknownField {
                message_type: message_type.to_string(),
                field: field.to_string(),
            })
    }

    /// Like [`Self::column_for`] but returns the full manifest column so the caller
    /// can consult its `sql_type` (e.g. to cast a comparison placeholder for a
    /// type-exact dialect — see [`SqlDialect::cast_compare_placeholder`]).
    pub(super) fn column_meta_for<'a>(
        table: &'a ManifestTable,
        field: &str,
        message_type: &str,
    ) -> Result<&'a crate::generation::ManifestColumn, CompileError> {
        table
            .columns
            .iter()
            .find(|c| c.field_name.eq_ignore_ascii_case(field) || c.column_name == field)
            .ok_or_else(|| CompileError::UnknownField {
                message_type: message_type.to_string(),
                field: field.to_string(),
            })
    }

    /// Push a value and return its dialect placeholder token.
    pub(super) fn push_param(params: &mut Vec<LogicalValue>, v: LogicalValue) -> String {
        params.push(v);
        D::placeholder(params.len())
    }

    /// Tenant/project context predicates AND'd into a statement WHERE for
    /// compiler-layer scoping (the A2 posture already used by ClickHouse and
    /// Cassandra via `util::append_context_predicates`): aggregate reads have
    /// no runtime RLS seam on file/columnar engines, so the compiler is the
    /// scoping layer. Renders through the dialect quote/placeholder hooks so
    /// `$N` / `?` / `@PN` numbering stays consistent with earlier params.
    /// Returns `None` when the context carries nothing to inject or the table
    /// declares no tenant/project column.
    pub(super) fn context_predicates(
        table: &ManifestTable,
        ctx: &super::CompileContext<'_>,
        params: &mut Vec<LogicalValue>,
    ) -> Option<String> {
        let mut parts: Vec<String> = Vec::new();
        if let Some(tid) = ctx.tenant_id
            && !tid.is_empty()
            && let Some(col_name) = super::util::resolve_tenant_column(table)
        {
            let col = table
                .columns
                .iter()
                .find(|c| c.column_name == col_name || c.field_name.eq_ignore_ascii_case(col_name));
            let placeholder = Self::push_param(params, LogicalValue::String(tid.to_string()));
            let placeholder = col
                .map(|c| D::cast_compare_placeholder(&c.sql_type, &placeholder))
                .unwrap_or(placeholder);
            parts.push(format!("{} = {placeholder}", D::quote(col_name)));
        }
        if let Some(pid) = ctx.project_id
            && !pid.is_empty()
            && let Some(col_name) = super::util::resolve_project_column(table)
        {
            let col = table
                .columns
                .iter()
                .find(|c| c.column_name == col_name || c.field_name.eq_ignore_ascii_case(col_name));
            let placeholder = Self::push_param(params, LogicalValue::String(pid.to_string()));
            let placeholder = col
                .map(|c| D::cast_compare_placeholder(&c.sql_type, &placeholder))
                .unwrap_or(placeholder);
            parts.push(format!("{} = {placeholder}", D::quote(col_name)));
        }
        if parts.is_empty() {
            None
        } else {
            Some(parts.join(" AND "))
        }
    }

    /// Render the `WHERE ...` body. Returns `None` for an empty filter so
    /// the caller can decide whether to emit `WHERE` at all; an empty OR
    /// lowers to the dialect false-literal.
    pub(super) fn render_where(
        filter: &LogicalFilter,
        table: &ManifestTable,
        message_type: &str,
        params: &mut Vec<LogicalValue>,
    ) -> Result<Option<String>, CompileError> {
        match filter {
            LogicalFilter::And(clauses) if clauses.is_empty() => Ok(None),
            LogicalFilter::Or(clauses) if clauses.is_empty() => {
                Ok(Some(D::false_literal().to_string()))
            }
            _ => Ok(Some(Self::render_filter(
                filter,
                table,
                message_type,
                params,
            )?)),
        }
    }

    /// Render a filter expression. Always non-empty; callers dispatch to
    /// `render_where` for the empty-filter case.
    pub(super) fn render_filter(
        filter: &LogicalFilter,
        table: &ManifestTable,
        message_type: &str,
        params: &mut Vec<LogicalValue>,
    ) -> Result<String, CompileError> {
        match filter {
            LogicalFilter::And(clauses) => {
                let parts = clauses
                    .iter()
                    .map(|c| Self::render_filter(c, table, message_type, params))
                    .collect::<Result<Vec<_>, _>>()?;
                Ok(format!("({})", parts.join(" AND ")))
            }
            LogicalFilter::Or(clauses) => {
                let parts = clauses
                    .iter()
                    .map(|c| Self::render_filter(c, table, message_type, params))
                    .collect::<Result<Vec<_>, _>>()?;
                Ok(format!("({})", parts.join(" OR ")))
            }
            LogicalFilter::Not(inner) => {
                let rendered = Self::render_filter(inner, table, message_type, params)?;
                Ok(format!("(NOT {rendered})"))
            }
            LogicalFilter::Comparison { field, op, value } => {
                let col = Self::column_meta_for(table, field, message_type)?;
                // SQL surprise: `col = NULL` is always UNKNOWN. Reject it
                // here so callers explicitly use `IsNull`.
                if value.is_null() {
                    return Err(CompileError::Malformed {
                        reason: format!(
                            "comparison with NULL on field '{field}' must use IsNull, not {}",
                            op.token()
                        ),
                    });
                }
                let placeholder = Self::push_param(params, value.clone());
                // Cast the placeholder to the column's SQL type when the dialect needs
                // it (Postgres `uuid = $1::UUID`); a no-op for the coercing dialects.
                let placeholder = D::cast_compare_placeholder(&col.sql_type, &placeholder);
                let sql_op = D::sql_op_for(*op);
                let rhs = D::wrap_value_for_op(*op, &placeholder);
                Ok(format!("{} {sql_op} {rhs}", D::quote(&col.column_name)))
            }
            LogicalFilter::IsNull(field) => {
                let column = Self::column_for(table, field, message_type)?;
                Ok(format!("{} IS NULL", D::quote(column)))
            }
            LogicalFilter::InList { field, values } => {
                if values.is_empty() {
                    return Ok(D::false_literal().to_string());
                }
                for value in values {
                    if matches!(value, LogicalValue::Array(_) | LogicalValue::Json(_)) {
                        return Err(CompileError::Malformed {
                            reason: format!(
                                "IN list for field '{field}' accepts scalar values only; got {}",
                                value.type_token()
                            ),
                        });
                    }
                    if value.is_null() {
                        return Err(CompileError::Malformed {
                            reason: format!(
                                "IN list for field '{field}' cannot contain NULL; use IsNull explicitly"
                            ),
                        });
                    }
                }
                let col = Self::column_meta_for(table, field, message_type)?;
                let placeholders = values
                    .iter()
                    .map(|v| {
                        let p = Self::push_param(params, v.clone());
                        D::cast_compare_placeholder(&col.sql_type, &p)
                    })
                    .collect::<Vec<_>>()
                    .join(", ");
                Ok(format!(
                    "{} IN ({placeholders})",
                    D::quote(&col.column_name)
                ))
            }
        }
    }

    /// Render one aggregate expression as `FUNC(arg) AS "alias"`.
    pub(super) fn render_aggregate(
        agg: &AggregateExpr,
        table: &ManifestTable,
        message_type: &str,
    ) -> Result<String, CompileError> {
        let token = agg.func.sql_token();
        let body = match agg.func {
            AggregateFunc::Count if agg.field == "*" => "*".to_string(),
            AggregateFunc::Count => {
                let col = Self::column_for(table, &agg.field, message_type)?;
                D::quote(col)
            }
            AggregateFunc::CountDistinct => {
                if agg.field == "*" {
                    return Err(CompileError::Malformed {
                        reason: "COUNT(DISTINCT *) is not allowed; specify a field".into(),
                    });
                }
                let col = Self::column_for(table, &agg.field, message_type)?;
                format!("DISTINCT {}", D::quote(col))
            }
            AggregateFunc::Sum | AggregateFunc::Avg | AggregateFunc::Min | AggregateFunc::Max => {
                if agg.field == "*" {
                    return Err(CompileError::Malformed {
                        reason: format!("{} requires a field name, not '*'", agg.func.sql_token()),
                    });
                }
                let col = Self::column_for(table, &agg.field, message_type)?;
                D::quote(col)
            }
        };
        Ok(format!("{token}({body}) AS {}", D::quote(&agg.alias)))
    }

    /// Resolve a field-or-alias for ORDER BY in an aggregate query:
    /// aggregate aliases win, then group-by manifest columns. Anything
    /// else is rejected (can't sort by an ungrouped raw column).
    pub(super) fn resolve_sort_field(
        field: &str,
        op: &LogicalAggregate,
        table: &ManifestTable,
    ) -> Result<String, CompileError> {
        if op.aggregates.iter().any(|a| a.alias == field) {
            return Ok(D::quote(field));
        }
        if op.group_by.iter().any(|f| f == field) {
            let col = Self::column_for(table, field, &op.message_type)?;
            return Ok(D::quote(col));
        }
        Err(CompileError::Malformed {
            reason: format!(
                "ORDER BY field '{field}' is neither an aggregate alias nor a GROUP BY column; \
                 aggregated queries can only sort by grouped or computed values"
            ),
        })
    }

    /// Render a HAVING filter — like `render_filter` but field references
    /// resolve to aggregate aliases / group-by columns rather than raw
    /// manifest columns.
    pub(super) fn render_having(
        filter: &LogicalFilter,
        op: &LogicalAggregate,
        table: &ManifestTable,
        params: &mut Vec<LogicalValue>,
    ) -> Result<String, CompileError> {
        match filter {
            LogicalFilter::And(clauses) if clauses.is_empty() => {
                Ok(D::having_true_literal().to_string())
            }
            LogicalFilter::Or(clauses) if clauses.is_empty() => {
                Ok(D::having_false_literal().to_string())
            }
            LogicalFilter::And(clauses) => {
                let parts = clauses
                    .iter()
                    .map(|c| Self::render_having(c, op, table, params))
                    .collect::<Result<Vec<_>, _>>()?;
                Ok(format!("({})", parts.join(" AND ")))
            }
            LogicalFilter::Or(clauses) => {
                let parts = clauses
                    .iter()
                    .map(|c| Self::render_having(c, op, table, params))
                    .collect::<Result<Vec<_>, _>>()?;
                Ok(format!("({})", parts.join(" OR ")))
            }
            LogicalFilter::Not(inner) => {
                let rendered = Self::render_having(inner, op, table, params)?;
                Ok(format!("(NOT {rendered})"))
            }
            LogicalFilter::Comparison {
                field,
                op: cmp,
                value,
            } => {
                if value.is_null() {
                    return Err(CompileError::Malformed {
                        reason: format!(
                            "HAVING comparison with NULL on '{field}' must use IsNull, not {}",
                            cmp.token()
                        ),
                    });
                }
                let token = Self::resolve_having_field(field, op, table, &op.message_type)?;
                let placeholder = Self::push_param(params, value.clone());
                let sql_op = D::sql_op_for(*cmp);
                let rhs = D::wrap_value_for_op(*cmp, &placeholder);
                Ok(format!("{token} {sql_op} {rhs}"))
            }
            LogicalFilter::IsNull(field) => {
                let token = Self::resolve_having_field(field, op, table, &op.message_type)?;
                Ok(format!("{token} IS NULL"))
            }
            LogicalFilter::InList { field, values } => {
                if values.is_empty() {
                    return Ok(D::having_false_literal().to_string());
                }
                // Same scalar/NULL contract as `render_filter`'s InList: a
                // nested Array/Json would bind a structure to one placeholder
                // (malformed SQL / bind error), and SQL `IN (NULL)` never
                // matches — use IsNull explicitly.
                for value in values {
                    if matches!(value, LogicalValue::Array(_) | LogicalValue::Json(_)) {
                        return Err(CompileError::Malformed {
                            reason: format!(
                                "HAVING IN list for field '{field}' accepts scalar values only; got {}",
                                value.type_token()
                            ),
                        });
                    }
                    if value.is_null() {
                        return Err(CompileError::Malformed {
                            reason: format!(
                                "HAVING IN list for field '{field}' cannot contain NULL; use IsNull explicitly"
                            ),
                        });
                    }
                }
                let token = Self::resolve_having_field(field, op, table, &op.message_type)?;
                let placeholders = values
                    .iter()
                    .map(|v| Self::push_param(params, v.clone()))
                    .collect::<Vec<_>>()
                    .join(", ");
                Ok(format!("{token} IN ({placeholders})"))
            }
        }
    }

    /// Resolve a HAVING field reference: aggregate alias or group-by
    /// column, both re-emitted as the quoted identifier (every dialect
    /// references the column/alias expression, not a positional index).
    pub(super) fn resolve_having_field(
        field: &str,
        op: &LogicalAggregate,
        table: &ManifestTable,
        message_type: &str,
    ) -> Result<String, CompileError> {
        if op.aggregates.iter().any(|a| a.alias == field) {
            return Ok(D::quote(field));
        }
        if op.group_by.iter().any(|f| f == field) {
            let column = Self::column_for(table, field, message_type)?;
            return Ok(D::quote(column));
        }
        Err(CompileError::Malformed {
            reason: format!(
                "HAVING field '{field}' is neither an aggregate alias nor a GROUP BY column"
            ),
        })
    }
}