udb 0.4.21

#![allow(clippy::result_large_err)]

//! PostgreSQL query building helpers: plan execution, row binding, join fusion,
//! and record serialisation. `rows_to_record_set` and its encryption-aware
//! siblings remain in `core.rs` because they depend on the private
//! `EncryptionMetrics` type.

use serde_json::Value as JsonValue;
use sqlx::Postgres;
use sqlx::postgres::PgArguments;
use sqlx::query::Query;
use uuid::Uuid;

use crate::broker::{RequestContext, resolve_table_for_message};
use crate::generation::sql::{resolve_tenant_column_ref, table_requires_tenant_column};
use crate::generation::{CatalogManifest, ManifestColumn, ManifestTable};
use crate::proto::{Mutation, SelectRequest, UpsertRequest};

use super::executor_utils::{
    invalid_argument_fields, json_scalar_to_string, qi_runtime, reject_plan, struct_to_json,
};

fn postgres_invalid_field(
    field: impl Into<String>,
    description: impl Into<String>,
    message: impl Into<String>,
) -> tonic::Status {
    invalid_argument_fields(message, [(field.into(), description.into())])
}

fn join_fusion_missing_tenant_column_status(table: &ManifestTable) -> tonic::Status {
    crate::runtime::executor_utils::schema_status(
        tonic::Code::FailedPrecondition,
        "postgres",
        "join_fusion",
        "tenant_column_required",
        format!(
            "join fusion cannot safely select scoped table {}.{} without a tenant column",
            table.schema, table.table
        ),
    )
}

fn postgres_internal_status(
    operation: impl Into<String>,
    message: impl Into<String>,
) -> tonic::Status {
    crate::runtime::executor_utils::internal_status("postgres", operation, message)
}

fn postgres_json_i64(field: &str, value: &JsonValue) -> Result<i64, tonic::Status> {
    value
        .as_i64()
        .or_else(|| value.as_str()?.parse().ok())
        .ok_or_else(|| {
            postgres_invalid_field(
                field,
                "must be an integer or integer string",
                format!("expected integer, got {value}"),
            )
        })
}

fn postgres_json_f64(field: &str, value: &JsonValue) -> Result<f64, tonic::Status> {
    value
        .as_f64()
        .or_else(|| value.as_str()?.parse().ok())
        .ok_or_else(|| {
            postgres_invalid_field(
                field,
                "must be a number or numeric string",
                format!("expected number, got {value}"),
            )
        })
}

// ── Transaction plan execution ────────────────────────────────────────────────

pub(crate) async fn execute_tx_plan(
    tx: &mut sqlx::Transaction<'_, Postgres>,
    manifest: &CatalogManifest,
    message_type: &str,
    sql: &str,
    columns: &[String],
    values: &[JsonValue],
    errors: &[String],
) -> Result<u64, tonic::Status> {
    reject_plan(errors)?;
    let table = resolve_table_for_message(manifest, message_type).map_err(|error| {
        postgres_invalid_field(
            "message_type",
            "must match exactly one manifest table message type",
            error.to_string(),
        )
    })?;
    let query = bind_values(sqlx::query(sql), table, columns, values)?;
    let result = query.execute(&mut **tx).await.map_err(|err| {
        postgres_internal_status(
            "execute_tx_plan",
            format!("transaction mutation failed: {err}"),
        )
    })?;
    Ok(result.rows_affected())
}

// ── Join fusion ───────────────────────────────────────────────────────────────

pub(crate) struct JoinFusionPlan {
    pub(crate) sql: String,
    pub(crate) bindings: Vec<(ManifestColumn, JsonValue)>,
}

pub(crate) fn build_join_fusion_sql(
    manifest: &CatalogManifest,
    request: &SelectRequest,
    context: &RequestContext,
    filter: &JsonValue,
) -> Result<JoinFusionPlan, tonic::Status> {
    if context.tenant_id.trim().is_empty() {
        return Err(postgres_invalid_field(
            "tenant_id",
            "must be non-empty for join fusion",
            "tenant_id is required for join fusion",
        ));
    }
    let message_types = split_join_message_types(&request.message_type);
    if message_types.len() < 2 {
        return Err(postgres_invalid_field(
            "message_type",
            "must contain at least two comma- or plus-separated message types",
            "join fusion requires at least two message types",
        ));
    }
    let tables = message_types
        .iter()
        .map(|message_type| {
            resolve_table_for_message(manifest, message_type).map_err(|error| {
                postgres_invalid_field(
                    "message_type",
                    "must match exactly one manifest table message type",
                    error.to_string(),
                )
            })
        })
        .collect::<Result<Vec<_>, _>>()?;
    let aliases = (0..tables.len())
        .map(|idx| format!("t{idx}"))
        .collect::<Vec<_>>();
    let select_list = join_select_list(&tables, &aliases, &request.fields)?;
    let mut sql = format!(
        "SELECT {} FROM {}.{} {}",
        select_list.join(", "),
        qi_runtime(&tables[0].schema),
        qi_runtime(&tables[0].table),
        qi_runtime(&aliases[0])
    );
    for idx in 1..tables.len() {
        let join = find_join_edge(
            &tables[0..idx],
            &aliases[0..idx],
            tables[idx],
            &aliases[idx],
        )
        .ok_or_else(|| {
            postgres_invalid_field(
                "message_type",
                "joined message types must have a foreign key path",
                format!(
                    "no foreign key path found for join fusion target {}",
                    message_types[idx]
                ),
            )
        })?;
        sql.push_str(" JOIN ");
        sql.push_str(&format!(
            "{}.{} {} ON {}",
            qi_runtime(&tables[idx].schema),
            qi_runtime(&tables[idx].table),
            qi_runtime(&aliases[idx]),
            join
        ));
    }

    let mut bindings = Vec::new();
    let mut predicates = Vec::new();
    for (table_idx, table) in tables.iter().enumerate() {
        let Some(column) = tenant_column_ref(table) else {
            if table_requires_tenant_column(table) {
                return Err(join_fusion_missing_tenant_column_status(table));
            }
            continue;
        };
        bindings.push((column.clone(), JsonValue::String(context.tenant_id.clone())));
        predicates.push(format!(
            "{}.{} = ${}",
            qi_runtime(&aliases[table_idx]),
            qi_runtime(&column.column_name),
            bindings.len()
        ));
    }
    if let JsonValue::Object(map) = filter {
        for (field, value) in map {
            if field.starts_with('$') || value.is_object() || value.is_array() {
                return Err(postgres_invalid_field(
                    "filter",
                    "must contain only simple equality fields for join fusion",
                    "join fusion supports only simple equality filters",
                ));
            }
            let (table_idx, column_name) = parse_join_field(field, &message_types)?;
            let column = tables[table_idx]
                .columns
                .iter()
                .find(|column| column.column_name == column_name)
                .ok_or_else(|| {
                    postgres_invalid_field(
                        "filter",
                        "must reference a column from one of the joined tables",
                        format!("unknown join filter field {field}"),
                    )
                })?;
            bindings.push((column.clone(), value.clone()));
            predicates.push(format!(
                "{}.{} = ${}",
                qi_runtime(&aliases[table_idx]),
                qi_runtime(&column.column_name),
                bindings.len()
            ));
        }
    }
    if !predicates.is_empty() {
        sql.push_str(" WHERE ");
        sql.push_str(&predicates.join(" AND "));
    }
    if request.limit > 0 {
        sql.push_str(&format!(" LIMIT {}", request.limit));
    }
    Ok(JoinFusionPlan { sql, bindings })
}

pub(crate) fn split_join_message_types(message_type: &str) -> Vec<String> {
    message_type
        .split([',', '+'])
        .map(str::trim)
        .filter(|part| !part.is_empty())
        .map(ToString::to_string)
        .collect()
}

pub(crate) fn is_join_fusion_message_type(message_type: &str) -> bool {
    split_join_message_types(message_type).len() > 1
}

fn join_select_list(
    tables: &[&ManifestTable],
    aliases: &[String],
    fields: &[String],
) -> Result<Vec<String>, tonic::Status> {
    if fields.is_empty() {
        return Ok(tables
            .iter()
            .zip(aliases)
            .flat_map(|(table, alias)| {
                table.columns.iter().map(move |column| {
                    format!(
                        "{}.{} AS {}",
                        qi_runtime(alias),
                        qi_runtime(&column.column_name),
                        qi_runtime(&format!("{}__{}", table.message_name, column.column_name))
                    )
                })
            })
            .collect());
    }
    fields
        .iter()
        .map(|field| {
            let message_types = tables
                .iter()
                .map(|table| table.message_name.clone())
                .collect::<Vec<_>>();
            let (table_idx, column_name) = parse_join_field(field, &message_types)?;
            if !tables[table_idx]
                .columns
                .iter()
                .any(|column| column.column_name == column_name)
            {
                return Err(postgres_invalid_field(
                    "fields",
                    "must reference columns from the joined tables",
                    format!("unknown join selected field {field}"),
                ));
            }
            Ok(format!(
                "{}.{} AS {}",
                qi_runtime(&aliases[table_idx]),
                qi_runtime(&column_name),
                qi_runtime(&field.replace('.', "__"))
            ))
        })
        .collect()
}

fn parse_join_field(
    field: &str,
    message_types: &[String],
) -> Result<(usize, String), tonic::Status> {
    if let Some((message_type, column)) = field.split_once('.') {
        let idx = message_types
            .iter()
            .position(|candidate| candidate.eq_ignore_ascii_case(message_type))
            .ok_or_else(|| {
                postgres_invalid_field(
                    "field",
                    "prefix must match one of the joined message types",
                    format!("unknown join field prefix {message_type}"),
                )
            })?;
        return Ok((idx, column.to_ascii_lowercase()));
    }
    Ok((0, field.to_ascii_lowercase()))
}

fn find_join_edge(
    prior_tables: &[&ManifestTable],
    prior_aliases: &[String],
    next_table: &ManifestTable,
    next_alias: &str,
) -> Option<String> {
    for (prior, prior_alias) in prior_tables.iter().zip(prior_aliases) {
        for fk in &prior.foreign_keys {
            if fk.ref_schema == next_table.schema && fk.ref_table == next_table.table {
                return Some(join_predicate(
                    prior_alias,
                    &fk.columns,
                    next_alias,
                    &fk.ref_columns,
                ));
            }
        }
        for fk in &next_table.foreign_keys {
            if fk.ref_schema == prior.schema && fk.ref_table == prior.table {
                return Some(join_predicate(
                    next_alias,
                    &fk.columns,
                    prior_alias,
                    &fk.ref_columns,
                ));
            }
        }
    }
    None
}

fn join_predicate(
    left_alias: &str,
    left_columns: &[String],
    right_alias: &str,
    right_columns: &[String],
) -> String {
    left_columns
        .iter()
        .zip(right_columns)
        .map(|(left, right)| {
            format!(
                "{}.{} = {}.{}",
                qi_runtime(left_alias),
                qi_runtime(left),
                qi_runtime(right_alias),
                qi_runtime(right)
            )
        })
        .collect::<Vec<_>>()
        .join(" AND ")
}

pub(crate) fn tenant_column_ref(table: &ManifestTable) -> Option<&ManifestColumn> {
    resolve_tenant_column_ref(table)
}

// ── Parameter binding ─────────────────────────────────────────────────────────

pub(crate) fn bind_values<'q>(
    mut query: Query<'q, Postgres, PgArguments>,
    table: &ManifestTable,
    columns: &[String],
    values: &[JsonValue],
) -> Result<Query<'q, Postgres, PgArguments>, tonic::Status> {
    if columns.len() != values.len() {
        return Err(postgres_invalid_field(
            "values",
            "number of values must match number of columns",
            format!(
                "parameter mismatch: {} columns, {} values",
                columns.len(),
                values.len()
            ),
        ));
    }
    for (column_name, value) in columns.iter().zip(values.iter()) {
        let column = table
            .columns
            .iter()
            .find(|column| column.column_name == *column_name);
        query = bind_one(query, column, value)?;
    }
    Ok(query)
}

pub(crate) fn bind_one<'q>(
    query: Query<'q, Postgres, PgArguments>,
    column: Option<&ManifestColumn>,
    value: &JsonValue,
) -> Result<Query<'q, Postgres, PgArguments>, tonic::Status> {
    let sql_type = column
        .map(|column| column.sql_type.to_ascii_uppercase())
        .unwrap_or_default();
    // DAT-001: a JSON/JSONB target column stores the value VERBATIM — including a
    // top-level JSON array or object. This decision MUST precede the filter-array
    // path below: otherwise a genuine JSON array destined for a JSONB entity
    // column is misrouted into the typed `text[]` bind and PostgreSQL rejects the
    // mutation with SQLSTATE 42846. Typed SQL-array binding (the branch below)
    // stays for a scalar column in a `$in` / `= ANY` filter context only.
    if sql_type.contains("JSON") {
        return Ok(query.bind(sqlx::types::Json(strip_nul_json(value))));
    }
    // Array value — used for `$in` / `col = ANY($N)` filters. (#121)
    // Bind a *typed* array matching the column type: PostgreSQL does NOT
    // implicitly cast a `text[]` element to the column type inside `= ANY`, so a
    // `uuid`/`int`/`numeric` column compared against a text array fails at
    // execution. Typed arrays keep the predicate index-usable (no `::type` cast
    // on the column needed). Checked before the scalar branches below so a
    // `uuid`/temporal array isn't misrouted into the scalar (single-value) path.
    if let JsonValue::Array(items) = value {
        if sql_type == "UUID" {
            let mut arr: Vec<Uuid> = Vec::with_capacity(items.len());
            for item in items {
                let parsed = item
                    .as_str()
                    .ok_or_else(|| {
                        postgres_invalid_field(
                            "value",
                            "UUID $in array values must be strings",
                            "UUID $in value must be a string",
                        )
                    })?
                    .parse::<Uuid>()
                    .map_err(|err| {
                        postgres_invalid_field(
                            "value",
                            "UUID $in array values must be valid UUID strings",
                            format!("invalid UUID in $in: {err}"),
                        )
                    })?;
                arr.push(parsed);
            }
            return Ok(query.bind(arr));
        }
        if sql_type.contains("INT") || sql_type.contains("SERIAL") {
            let mut arr: Vec<i64> = Vec::with_capacity(items.len());
            for item in items {
                arr.push(postgres_json_i64("value", item)?);
            }
            return Ok(query.bind(arr));
        }
        if sql_type.contains("REAL")
            || sql_type.contains("DOUBLE")
            || sql_type.contains("FLOAT")
            || sql_type.contains("NUMERIC")
            || sql_type.contains("DECIMAL")
        {
            let mut arr: Vec<f64> = Vec::with_capacity(items.len());
            for item in items {
                arr.push(postgres_json_f64("value", item)?);
            }
            return Ok(query.bind(arr));
        }
        if sql_type.contains("BOOL") {
            let arr: Vec<bool> = items.iter().map(|i| i.as_bool().unwrap_or(false)).collect();
            return Ok(query.bind(arr));
        }
        // text / varchar / enum / timestamp and friends: a text[] compares
        // correctly for text-typed columns (`text = ANY(text[])`).
        let arr: Vec<String> = items
            .iter()
            .map(json_scalar_to_string)
            .map(|s| strip_nul(&s))
            .collect();
        return Ok(query.bind(arr));
    }

    // Typed scalar binds for the column types that have NO implicit/assignment
    // cast from `text` in PostgreSQL: uuid, timestamptz, timestamp, date.
    // Binding one of these as text — or as a *text-typed* NULL, which the old
    // fallback did — makes an INSERT/upsert that supplies the column fail to
    // PLAN with SQLSTATE 42804 (datatype_mismatch), which surfaced only as an
    // opaque INTERNAL. A fresh INSERT omits the server-defaulted audit columns
    // (created_at/updated_at fill from CURRENT_TIMESTAMP) so never hit this; a
    // read-modify-write (Select → edit one field → Upsert) re-supplies those
    // columns as the RFC-3339 strings the SELECT serializer emits, and did.
    // Bind the real Rust type, mirroring the compiler bind path in
    // `core/helpers.rs`.
    if sql_type == "UUID" {
        return match value {
            JsonValue::Null => Ok(query.bind(Option::<Uuid>::None)),
            JsonValue::String(raw) if raw.trim().is_empty() => Ok(query.bind(Option::<Uuid>::None)),
            JsonValue::String(raw) => {
                raw.parse::<Uuid>()
                    .map(|uuid| query.bind(uuid))
                    .map_err(|err| {
                        postgres_invalid_field(
                            "value",
                            "UUID value must be a valid UUID string",
                            format!("invalid UUID: {err}"),
                        )
                    })
            }
            _ => Err(postgres_invalid_field(
                "value",
                "UUID value must be a string",
                "UUID value must be a string",
            )),
        };
    }
    if sql_type.contains("TIMESTAMPTZ") || sql_type.contains("TIMESTAMP WITH TIME ZONE") {
        return match value {
            JsonValue::Null => Ok(query.bind(Option::<chrono::DateTime<chrono::Utc>>::None)),
            JsonValue::String(raw) if raw.trim().is_empty() => {
                Ok(query.bind(Option::<chrono::DateTime<chrono::Utc>>::None))
            }
            JsonValue::String(raw) => chrono::DateTime::parse_from_rfc3339(raw)
                .map(|dt| query.bind(dt.with_timezone(&chrono::Utc)))
                .map_err(|err| {
                    postgres_invalid_field(
                        "value",
                        "timestamptz value must be an RFC3339 string",
                        format!("timestamptz value must be an RFC3339 string: {err}"),
                    )
                }),
            _ => Err(postgres_invalid_field(
                "value",
                "timestamptz value must be a string or null",
                "timestamptz value must be a string or null",
            )),
        };
    }
    if sql_type.contains("TIMESTAMP") {
        return match value {
            JsonValue::Null => Ok(query.bind(Option::<chrono::NaiveDateTime>::None)),
            JsonValue::String(raw) if raw.trim().is_empty() => {
                Ok(query.bind(Option::<chrono::NaiveDateTime>::None))
            }
            JsonValue::String(raw) => parse_naive_datetime(raw)
                .map(|dt| query.bind(dt))
                .ok_or_else(|| {
                    postgres_invalid_field(
                        "value",
                        "timestamp value must be an ISO-8601 string or null",
                        "timestamp value must be an ISO-8601 string or null",
                    )
                }),
            _ => Err(postgres_invalid_field(
                "value",
                "timestamp value must be a string or null",
                "timestamp value must be a string or null",
            )),
        };
    }
    if sql_type.contains("DATE") {
        return match value {
            JsonValue::Null => Ok(query.bind(Option::<chrono::NaiveDate>::None)),
            JsonValue::String(raw) if raw.trim().is_empty() => {
                Ok(query.bind(Option::<chrono::NaiveDate>::None))
            }
            JsonValue::String(raw) => raw
                .parse::<chrono::NaiveDate>()
                .map(|date| query.bind(date))
                .map_err(|err| {
                    postgres_invalid_field(
                        "value",
                        "date value must be a valid date string",
                        format!("invalid date: {err}"),
                    )
                }),
            _ => Err(postgres_invalid_field(
                "value",
                "date value must be a string or null",
                "date value must be a string or null",
            )),
        };
    }

    // Remaining scalar types. A NULL binds as the matching nullable Rust type so
    // the parameter's Postgres type is unambiguous (bool/int/float each get a
    // typed NULL; text/varchar/enum accept a plain text NULL).
    if value.is_null() {
        if sql_type.contains("BOOL") {
            return Ok(query.bind(Option::<bool>::None));
        }
        if sql_type.contains("INT") || sql_type.contains("BIGSERIAL") || sql_type.contains("SERIAL")
        {
            return Ok(query.bind(Option::<i64>::None));
        }
        if sql_type.contains("REAL")
            || sql_type.contains("DOUBLE")
            || sql_type.contains("FLOAT")
            || sql_type.contains("NUMERIC")
            || sql_type.contains("DECIMAL")
        {
            return Ok(query.bind(Option::<f64>::None));
        }
        return Ok(query.bind(Option::<String>::None));
    }
    if sql_type.contains("BOOL") {
        return Ok(query.bind(value.as_bool().unwrap_or(false)));
    }
    if sql_type.contains("INT") || sql_type.contains("BIGSERIAL") || sql_type.contains("SERIAL") {
        return Ok(query.bind(postgres_json_i64("value", value)?));
    }
    if sql_type.contains("REAL")
        || sql_type.contains("DOUBLE")
        || sql_type.contains("FLOAT")
        || sql_type.contains("NUMERIC")
        || sql_type.contains("DECIMAL")
    {
        return Ok(query.bind(postgres_json_f64("value", value)?));
    }
    Ok(query.bind(strip_nul(&json_scalar_to_string(value))))
}

/// Parse the datetime string forms the SELECT serializer can emit for a
/// `TIMESTAMP` (no time zone) column back into a `NaiveDateTime`. `row_value_to_json`
/// renders these via `NaiveDateTime::to_string()` (space-separated), but accept the
/// ISO `T` form too so a client-built payload round-trips as well.
fn parse_naive_datetime(raw: &str) -> Option<chrono::NaiveDateTime> {
    const FORMATS: &[&str] = &[
        "%Y-%m-%d %H:%M:%S%.f",
        "%Y-%m-%dT%H:%M:%S%.f",
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%dT%H:%M:%S",
    ];
    FORMATS
        .iter()
        .find_map(|fmt| chrono::NaiveDateTime::parse_from_str(raw, fmt).ok())
}

/// A NUL (`0x00`) byte cannot be stored in a Postgres `text`/`varchar`/`json(b)`
/// value. Strip it at the typed-record bind edge so a hostile/garbage byte cannot
/// fault the whole upsert with Postgres' UTF-8 NUL rejection (B14). This path does
/// NOT go through `bind_generic_pg_param` (which already strips), so it needs its
/// own guard.
fn strip_nul(s: &str) -> String {
    if s.contains('\u{0}') {
        s.replace('\u{0}', "")
    } else {
        s.to_string()
    }
}

fn strip_nul_json(value: &JsonValue) -> JsonValue {
    match value {
        JsonValue::String(s) if s.contains('\u{0}') => JsonValue::String(s.replace('\u{0}', "")),
        JsonValue::Array(items) => JsonValue::Array(items.iter().map(strip_nul_json).collect()),
        JsonValue::Object(map) => JsonValue::Object(
            map.iter()
                .map(|(k, v)| (k.clone(), strip_nul_json(v)))
                .collect(),
        ),
        other => other.clone(),
    }
}

// ── Record serialisation ──────────────────────────────────────────────────────

pub(crate) fn upsert_record_json(request: &UpsertRequest) -> Result<JsonValue, tonic::Status> {
    if let Some(payload) = &request.payload {
        return Ok(struct_to_json(payload));
    }
    if !request.record_json.is_empty() {
        return serde_json::from_slice(&request.record_json).map_err(|err| {
            postgres_invalid_field(
                "record_json",
                "must contain valid JSON",
                format!("record_json must be valid JSON: {err}"),
            )
        });
    }
    Err(postgres_invalid_field(
        "payload",
        "payload or record_json must be provided",
        "payload or record_json is required",
    ))
}

pub(crate) fn mutation_record_json(mutation: &Mutation) -> Result<JsonValue, tonic::Status> {
    if let Some(payload) = &mutation.payload {
        return Ok(struct_to_json(payload));
    }
    if !mutation.record_json.is_empty() {
        return serde_json::from_slice(&mutation.record_json).map_err(|err| {
            postgres_invalid_field(
                "record_json",
                "must contain valid JSON",
                format!("record_json must be valid JSON: {err}"),
            )
        });
    }
    Err(postgres_invalid_field(
        "payload",
        "payload or record_json must be provided",
        "payload or record_json is required",
    ))
}

pub(crate) fn record_values(
    record: &JsonValue,
    columns: &[String],
) -> Result<Vec<JsonValue>, tonic::Status> {
    let object = record.as_object().ok_or_else(|| {
        postgres_invalid_field(
            "record",
            "must be a JSON object",
            "record must be a JSON object",
        )
    })?;
    Ok(columns
        .iter()
        .map(|column| object.get(column).cloned().unwrap_or(JsonValue::Null))
        .collect())
}

pub(crate) fn filter_bind_values(filter: &JsonValue) -> Vec<JsonValue> {
    let mut out = Vec::new();
    collect_filter_values(filter, &mut out);
    out
}

fn collect_filter_values(value: &JsonValue, out: &mut Vec<JsonValue>) {
    match value {
        JsonValue::Object(map) => {
            for (key, nested) in map {
                let normalized = key.to_ascii_lowercase();
                if matches!(normalized.as_str(), "$and" | "and" | "$or" | "or") {
                    collect_filter_values(nested, out);
                } else if normalized.starts_with('$') {
                    // Skip the null predicates — `$is_null`/`$not_null` compile to
                    // `IS NULL` / `IS NOT NULL`, which bind no SQL parameter. Pushing
                    // a value for them would desync the placeholder↔value count and
                    // make `bind_values` fail (or bind onto the wrong $N).
                    if !matches!(normalized.as_str(), "$is_null" | "$not_null") {
                        out.push(nested.clone());
                    }
                } else if let JsonValue::Object(_) = nested {
                    collect_filter_values(nested, out);
                } else {
                    out.push(nested.clone());
                }
            }
        }
        JsonValue::Array(items) => {
            for item in items {
                collect_filter_values(item, out);
            }
        }
        _ => {}
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::generation::{CatalogManifest, ManifestForeignKey, ManifestTableSecurity};
    use crate::proto::{ErrorDetail, ErrorKind};
    use crate::runtime::executor_utils::ERROR_DETAIL_METADATA_KEY;

    fn decode_detail(status: &tonic::Status) -> ErrorDetail {
        let raw = status
            .metadata()
            .get_bin(ERROR_DETAIL_METADATA_KEY)
            .expect("typed error detail trailer");
        crate::runtime::executor_utils::decode_error_detail_from_raw(&raw)
    }

    fn assert_single_field_violation(status: &tonic::Status, field: &str, description: &str) {
        assert_eq!(status.code(), tonic::Code::InvalidArgument);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Validation as i32);
        assert!(!detail.retryable);
        assert_eq!(detail.field_violations.len(), 1);
        assert_eq!(detail.field_violations[0].field, field);
        assert_eq!(detail.field_violations[0].description, description);
    }

    fn assert_schema_detail(
        status: &tonic::Status,
        backend: &str,
        operation: &str,
        schema_code: &str,
        message: &str,
    ) {
        assert_eq!(status.code(), tonic::Code::FailedPrecondition);
        assert_eq!(status.message(), message);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Schema as i32);
        assert_eq!(detail.backend, backend);
        assert_eq!(detail.operation, operation);
        assert_eq!(detail.capability_required, schema_code);
        assert!(!detail.retryable);
        assert_eq!(detail.retry_after_ms, 0);
        assert!(detail.field_violations.is_empty());
    }

    fn assert_internal_detail(
        status: &tonic::Status,
        backend: &str,
        operation: &str,
        message: &str,
    ) {
        assert_eq!(status.code(), tonic::Code::Internal);
        assert_eq!(status.message(), message);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Internal as i32);
        assert_eq!(detail.backend, backend);
        assert_eq!(detail.operation, operation);
        assert!(!detail.retryable);
        assert_eq!(detail.retry_after_ms, 0);
        assert!(detail.field_violations.is_empty());
    }

    fn expect_status<T>(result: Result<T, tonic::Status>) -> tonic::Status {
        match result {
            Ok(_) => panic!("expected validation status"),
            Err(status) => status,
        }
    }

    fn ctx() -> RequestContext {
        RequestContext {
            tenant_id: "acme".to_string(),
            ..RequestContext::default()
        }
    }

    fn col(name: &str) -> ManifestColumn {
        ManifestColumn {
            field_name: name.to_string(),
            column_name: name.to_string(),
            proto_type: "string".to_string(),
            sql_type: "text".to_string(),
            ..ManifestColumn::default()
        }
    }

    fn tenant_col(field_name: &str, column_name: &str, flagged: bool) -> ManifestColumn {
        ManifestColumn {
            field_name: field_name.to_string(),
            column_name: column_name.to_string(),
            proto_type: "string".to_string(),
            sql_type: "text".to_string(),
            is_tenant_column: flagged,
            ..ManifestColumn::default()
        }
    }

    fn table(message: &str, physical: &str, columns: Vec<ManifestColumn>) -> ManifestTable {
        ManifestTable {
            message_name: format!("acme.test.v1.{message}"),
            schema: "public".to_string(),
            table: physical.to_string(),
            columns,
            primary_key: vec!["id".to_string()],
            ..ManifestTable::default()
        }
    }

    fn join_manifest(mut left: ManifestTable, right: ManifestTable) -> CatalogManifest {
        left.foreign_keys.push(ManifestForeignKey {
            name: "fk_right".to_string(),
            columns: vec!["right_id".to_string()],
            ref_schema: right.schema.clone(),
            ref_table: right.table.clone(),
            ref_columns: vec!["id".to_string()],
            ..ManifestForeignKey::default()
        });
        CatalogManifest {
            tables: vec![left, right],
            ..CatalogManifest::default()
        }
    }

    fn join_request() -> SelectRequest {
        SelectRequest {
            message_type: "Left,Right".to_string(),
            limit: 25,
            ..SelectRequest::default()
        }
    }

    #[test]
    fn tenant_column_ref_prefers_declared_table_security_column() {
        let mut table = table(
            "Left",
            "lefts",
            vec![
                col("id"),
                tenant_col("tenant_id", "tenant_id", true),
                tenant_col("account", "account_id", false),
            ],
        );
        table.table_security = ManifestTableSecurity {
            tenant_column: "account".to_string(),
            ..ManifestTableSecurity::default()
        };

        let resolved = tenant_column_ref(&table).expect("tenant column");

        assert_eq!(resolved.column_name, "account_id");
    }

    #[test]
    fn tenant_column_ref_uses_system_and_legacy_names() {
        let system = table(
            "Left",
            "lefts",
            vec![col("id"), tenant_col("_tenant_id", "_tenant_id", false)],
        );
        assert_eq!(
            tenant_column_ref(&system).map(|column| column.column_name.as_str()),
            Some("_tenant_id")
        );

        let legacy = table(
            "Left",
            "lefts",
            vec![col("id"), tenant_col("org_id", "organization_id", false)],
        );
        assert_eq!(
            tenant_column_ref(&legacy).map(|column| column.column_name.as_str()),
            Some("organization_id")
        );
    }

    #[test]
    fn join_fusion_adds_tenant_predicate_for_every_joined_tenant_table() {
        let mut left = table(
            "Left",
            "lefts",
            vec![
                col("id"),
                col("right_id"),
                tenant_col("tenant_id", "tenant_id", true),
            ],
        );
        left.enable_rls = true;
        let mut right = table(
            "Right",
            "rights",
            vec![col("id"), tenant_col("tenant_id", "tenant_id", true)],
        );
        right.enable_rls = true;
        let manifest = join_manifest(left, right);

        let plan =
            build_join_fusion_sql(&manifest, &join_request(), &ctx(), &JsonValue::Null).unwrap();

        assert!(
            plan.sql.contains(r#""t0"."tenant_id" = $1"#),
            "{}",
            plan.sql
        );
        assert!(
            plan.sql.contains(r#""t1"."tenant_id" = $2"#),
            "{}",
            plan.sql
        );
        assert_eq!(plan.bindings.len(), 2);
        assert_eq!(plan.bindings[0].1, JsonValue::String("acme".to_string()));
        assert_eq!(plan.bindings[1].1, JsonValue::String("acme".to_string()));
    }

    #[test]
    fn join_fusion_fails_closed_for_scoped_table_without_tenant_column() {
        let mut left = table("Left", "lefts", vec![col("id"), col("right_id")]);
        left.enable_rls = true;
        let right = table("Right", "rights", vec![col("id")]);
        let manifest = join_manifest(left, right);

        let err = build_join_fusion_sql(&manifest, &join_request(), &ctx(), &JsonValue::Null)
            .err()
            .expect("scoped table without tenant column must fail closed");

        assert_schema_detail(
            &err,
            "postgres",
            "join_fusion",
            "tenant_column_required",
            "join fusion cannot safely select scoped table public.lefts without a tenant column",
        );
    }

    #[test]
    fn postgres_internal_status_carries_typed_detail() {
        let status = postgres_internal_status("execute_tx_plan", "transaction mutation failed");

        assert_internal_detail(
            &status,
            "postgres",
            "execute_tx_plan",
            "transaction mutation failed",
        );
    }

    #[test]
    fn join_fusion_validation_carries_field_violations() {
        let mut request = join_request();
        request.message_type = "Left".to_string();
        let manifest = CatalogManifest::default();

        let err = build_join_fusion_sql(&manifest, &request, &ctx(), &JsonValue::Null)
            .err()
            .expect("single message join must fail");

        assert_single_field_violation(
            &err,
            "message_type",
            "must contain at least two comma- or plus-separated message types",
        );

        request.message_type = "Left,Missing".to_string();
        let manifest = CatalogManifest {
            tables: vec![table("Left", "lefts", vec![col("id")])],
            ..CatalogManifest::default()
        };
        let err = build_join_fusion_sql(&manifest, &request, &ctx(), &JsonValue::Null)
            .err()
            .expect("unknown message type must fail");

        assert_single_field_violation(
            &err,
            "message_type",
            "must match exactly one manifest table message type",
        );
    }

    /// DAT-001: a top-level JSON array destined for a JSON/JSONB column must bind
    /// as JSON (verbatim), NOT be misrouted into the typed `text[]` filter-array
    /// path (which fails at PostgreSQL with 42846). The JSON decision must precede
    /// the array branch, so a JSONB column accepts an array of strings, an array
    /// of objects, and an empty array without string-wrapping — while a scalar
    /// column still gets its typed `$in` array.
    #[test]
    fn postgres_binds_json_arrays_into_jsonb_columns() {
        let mut jsonb = col("permissions_json");
        jsonb.sql_type = "JSONB".to_string();
        // Array of strings, array of objects, and an empty array all bind OK for a
        // JSONB target (before the fix these hit the typed text[] path).
        for value in [
            serde_json::json!(["partner.fleet.read", "partner.fleet.write"]),
            serde_json::json!([{"role": "reader"}, {"role": "writer"}]),
            serde_json::json!([]),
            serde_json::json!({"nested": ["a", "b"]}),
        ] {
            assert!(
                bind_one(sqlx::query("SELECT $1"), Some(&jsonb), &value).is_ok(),
                "JSONB column must accept JSON value {value} without a text[] misbind"
            );
        }
        // A scalar UUID column still routes an array into the typed $in path and
        // rejects non-string elements there (proves the array branch is intact).
        let mut uuid_col = col("id");
        uuid_col.sql_type = "UUID".to_string();
        assert!(
            bind_one(
                sqlx::query("SELECT $1"),
                Some(&uuid_col),
                &serde_json::json!([1])
            )
            .is_err(),
            "a scalar UUID $in array must still validate its elements"
        );
    }

    #[test]
    fn postgres_bind_validation_carries_field_violations() {
        let table = table("Left", "lefts", vec![col("id")]);
        let err = expect_status(bind_values(
            sqlx::query("SELECT 1"),
            &table,
            &["id".to_string()],
            &[],
        ));

        assert_single_field_violation(
            &err,
            "values",
            "number of values must match number of columns",
        );

        let mut uuid_col = col("id");
        uuid_col.sql_type = "UUID".to_string();
        let err = expect_status(bind_one(
            sqlx::query("SELECT $1"),
            Some(&uuid_col),
            &serde_json::json!([1]),
        ));

        assert_single_field_violation(&err, "value", "UUID $in array values must be strings");

        let mut int_col = col("age");
        int_col.sql_type = "BIGINT".to_string();
        let err = expect_status(bind_one(
            sqlx::query("SELECT $1"),
            Some(&int_col),
            &serde_json::json!("not-an-int"),
        ));

        assert_single_field_violation(&err, "value", "must be an integer or integer string");
    }

    #[test]
    fn record_json_validation_carries_field_violations() {
        let upsert = UpsertRequest {
            record_json: b"{".to_vec(),
            ..UpsertRequest::default()
        };
        let err = upsert_record_json(&upsert)
            .err()
            .expect("invalid record_json must fail");

        assert_single_field_violation(&err, "record_json", "must contain valid JSON");

        let mutation = Mutation::default();
        let err = mutation_record_json(&mutation)
            .err()
            .expect("missing mutation payload must fail");

        assert_single_field_violation(&err, "payload", "payload or record_json must be provided");

        let err = record_values(&JsonValue::Array(Vec::new()), &["id".to_string()])
            .err()
            .expect("non-object record must fail");

        assert_single_field_violation(&err, "record", "must be a JSON object");
    }
}