pharos_runtime_database/

relational.rs

//! Parameterized SQL lowering shared by the three relational backend adapters.
//! Connection, isolation, error mapping, and migration execution remain with
//! the SQLite, PostgreSQL, and MySQL adapter crates.

use std::collections::{BTreeMap, BTreeSet};

use crate::operations::{
    Aggregate, AggregateQuery, AppSchema, Claim, DatabaseError, Entity, Field, FieldConstraint,
    FieldDefault, FieldType, Identity, Index, IndexKey, MAX_TEXT_LENGTH, Mutation, OnDelete,
    Predicate, Query, SchemaChange, WriteValue, decimal_cents, decimal_from_cents, entity,
    materialize_generated_predicate, valid_identifier, validate_claim, validate_copy_field,
    validate_predicate, validate_query,
};
use serde_json::Value;

/// The physical placeholder and identifier convention selected by an adapter.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum Dialect {
    Sqlite,
    Postgresql,
    Mysql,
}

#[cfg(test)]
mod schema_literal_tests {
    use super::{Dialect, ddl_literal};
    use serde_json::json;

    /// MySQL literals avoid every quote-scanner and SQL-mode-dependent escape.
    #[test]
    fn mysql_schema_literals_encode_utf8_bytes() {
        assert_eq!(
            ddl_literal(Dialect::Mysql, &json!("\\'")),
            Ok("_utf8mb4 X'5c27'".to_owned())
        );
        assert_eq!(
            ddl_literal(Dialect::Mysql, &json!("é")),
            Ok("_utf8mb4 X'c3a9'".to_owned())
        );
        assert_eq!(
            ddl_literal(Dialect::Mysql, &json!("")),
            Ok("_utf8mb4 X''".to_owned())
        );
    }

    /// Standard SQL literal delimiters double quotes while preserving backslashes.
    #[test]
    fn standard_schema_literals_preserve_values() {
        for dialect in [Dialect::Sqlite, Dialect::Postgresql] {
            assert_eq!(ddl_literal(dialect, &json!("\\'")), Ok("'\\'''".to_owned()));
        }
    }
}

/// One safely parameterized statement; values never enter its SQL text.
#[derive(Debug, Clone, PartialEq)]
pub struct Statement {
    pub sql: String,
    pub parameters: Vec<Value>,
    /// Positions whose declared logical field requires a physical byte buffer.
    pub binary_parameters: BTreeSet<usize>,
}

struct Builder {
    dialect: Dialect,
    parameters: Vec<Value>,
    binary_parameters: BTreeSet<usize>,
}

impl Builder {
    /// Starts one statement with an empty parameter list.
    fn new(dialect: Dialect) -> Self {
        Self {
            dialect,
            parameters: Vec::new(),
            binary_parameters: BTreeSet::new(),
        }
    }

    /// Quotes a validated logical identifier using the target SQL dialect.
    fn name(&self, value: &str) -> Result<String, DatabaseError> {
        if !valid_identifier(value) {
            return Err(DatabaseError::InvalidContract);
        }
        Ok(match self.dialect {
            Dialect::Mysql => format!("`{value}`"),
            Dialect::Sqlite | Dialect::Postgresql => format!("\"{value}\""),
        })
    }

    /// Appends one bind value without interpolating it into SQL text.
    fn bind(&mut self, value: Value) -> String {
        self.parameters.push(value);
        match self.dialect {
            Dialect::Postgresql => format!("${}", self.parameters.len()),
            Dialect::Sqlite | Dialect::Mysql => "?".to_owned(),
        }
    }

    /// Encodes a logical field value before binding it to the physical dialect.
    fn bind_field(&mut self, field: &Field, value: Value) -> Result<String, DatabaseError> {
        let value = if field.kind == FieldType::Json && !value.is_null() {
            crate::operations::canonical_json_value(&value)?
        } else {
            value
        };
        if field.kind == FieldType::Binary && !value.is_null() {
            self.binary_parameters.insert(self.parameters.len());
            return Ok(self.bind(value));
        }
        if self.dialect == Dialect::Mysql && field.kind == FieldType::Json && !value.is_null() {
            return Ok(format!("CAST({} AS JSON)", self.bind(value)));
        }
        if field.kind == FieldType::Decimal {
            if value.is_null() {
                return Ok(match self.dialect {
                    Dialect::Postgresql => {
                        format!("CAST({}::TEXT AS NUMERIC(10,2))", self.bind(Value::Null))
                    }
                    _ => self.bind(Value::Null),
                });
            }
            let cents = decimal_cents(&value)?;
            return match self.dialect {
                Dialect::Sqlite => Ok(self.bind(Value::from(cents as f64 / 100.0))),
                Dialect::Postgresql => Ok(format!(
                    "CAST({}::TEXT AS NUMERIC(10,2))",
                    self.bind(Value::String(decimal_from_cents(cents)))
                )),
                Dialect::Mysql => Ok(self.bind(Value::String(decimal_from_cents(cents)))),
            };
        }
        if field.kind == FieldType::Timestamp && !value.is_null() {
            let value = crate::operations::canonical_stored_timestamp(
                value.as_str().ok_or(DatabaseError::InvalidInput)?,
            )?;
            return Ok(self.bind(Value::String(value)));
        }
        Ok(self.bind(value))
    }

    /// Finds the declared field type for a parameterized comparison.
    fn bind_named_field(
        &mut self,
        declaration: &Entity,
        field: &str,
        value: Value,
    ) -> Result<String, DatabaseError> {
        let field = declaration
            .fields
            .iter()
            .find(|item| item.name == field)
            .ok_or(DatabaseError::InvalidContract)?;
        if field.kind == FieldType::Json {
            return Ok(self.bind(crate::index_values::json_comparison_value(&value)?));
        }
        self.bind_field(field, value)
    }

    /// Resolves a comparison key without changing the projected app field.
    fn comparison_name(&self, declaration: &Entity, name: &str) -> Result<String, DatabaseError> {
        let field = declaration
            .fields
            .iter()
            .find(|field| field.name == name)
            .ok_or(DatabaseError::InvalidContract)?;
        self.name(&if field.kind == FieldType::Json {
            crate::index_values::json_comparison_key(name)
        } else {
            name.to_owned()
        })
    }

    /// Lowers a validated predicate recursively while preserving parameter order.
    fn filter(
        &mut self,
        declaration: &Entity,
        predicate: &Predicate,
    ) -> Result<String, DatabaseError> {
        match predicate {
            Predicate::Equal(field, value) => self.compare(declaration, field, "=", value),
            Predicate::GeneratedEqual(_, _) => Err(DatabaseError::InvalidInput),
            Predicate::NotEqual(field, value) => self.compare(declaration, field, "<>", value),
            Predicate::LessThan(field, value) => self.compare(declaration, field, "<", value),
            Predicate::AtMost(field, value) => self.compare(declaration, field, "<=", value),
            Predicate::GreaterThan(field, value) => self.compare(declaration, field, ">", value),
            Predicate::AtLeast(field, value) => self.compare(declaration, field, ">=", value),
            Predicate::Contains(field, needle) => {
                let name = self.name(field)?;
                let value = self.bind(Value::String(needle.clone()));
                Ok(match self.dialect {
                    Dialect::Mysql => format!("LOCATE(BINARY {value}, BINARY {name}) > 0"),
                    Dialect::Sqlite => format!("INSTR({name}, {value}) > 0"),
                    Dialect::Postgresql => format!("STRPOS({name}, {value}) > 0"),
                })
            }
            Predicate::IsNull(field) => Ok(format!(
                "{} IS NULL",
                self.comparison_name(declaration, field)?
            )),
            Predicate::In(field, values) => {
                if values.len() > 64 {
                    return Err(DatabaseError::InvalidInput);
                }
                if values.is_empty() {
                    return Ok("1 = 0".to_owned());
                }
                let name = self.comparison_name(declaration, field)?;
                let placeholders = values
                    .iter()
                    .map(|value| self.bind_named_field(declaration, field, value.clone()))
                    .collect::<Result<Vec<_>, _>>()?
                    .join(", ");
                Ok(format!("{name} IN ({placeholders})"))
            }
            Predicate::All(parts) => self.combine(declaration, parts, " AND "),
            Predicate::Any(parts) => self.combine(declaration, parts, " OR "),
        }
    }

    /// Handles null equality with SQL null semantics.
    fn compare(
        &mut self,
        declaration: &Entity,
        field: &str,
        operator: &str,
        value: &Value,
    ) -> Result<String, DatabaseError> {
        let name = self.comparison_name(declaration, field)?;
        if value.is_null() {
            return match operator {
                "=" => Ok(format!("{name} IS NULL")),
                "<>" => Ok(format!("{name} IS NOT NULL")),
                _ => Err(DatabaseError::InvalidInput),
            };
        }
        let placeholder = self.bind_named_field(declaration, field, value.clone())?;
        if self.dialect == Dialect::Mysql
            && declaration
                .fields
                .iter()
                .any(|entry| entry.name == field && entry.kind == FieldType::Text)
        {
            return Ok(format!("BINARY {name} {operator} BINARY {placeholder}"));
        }
        Ok(format!("{name} {operator} {placeholder}"))
    }

    /// Joins a bounded group of nested predicates.
    fn combine(
        &mut self,
        declaration: &Entity,
        parts: &[Predicate],
        separator: &str,
    ) -> Result<String, DatabaseError> {
        if parts.is_empty() || parts.len() > 32 {
            return Err(DatabaseError::InvalidInput);
        }
        let fragments = parts
            .iter()
            .map(|part| self.filter(declaration, part))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(format!("({})", fragments.join(separator)))
    }

    /// Finishes a statement without altering its bind order.
    fn finish(self, sql: String) -> Statement {
        Statement {
            sql,
            parameters: self.parameters,
            binary_parameters: self.binary_parameters,
        }
    }
}

/// Compiles one bounded logical query to parameterized SQL.
pub fn compile_query(
    dialect: Dialect,
    schema: &AppSchema,
    query: &Query,
) -> Result<Statement, DatabaseError> {
    validate_query(schema, query)?;
    let mut builder = Builder::new(dialect);
    let declaration = entity(schema, &query.entity)?;
    let columns = query
        .fields
        .iter()
        .map(|field| {
            let name = builder.name(field)?;
            if dialect == Dialect::Postgresql
                && declaration
                    .fields
                    .iter()
                    .any(|item| item.name == *field && item.kind == FieldType::Decimal)
            {
                Ok(format!("CAST({name} AS TEXT) AS {name}"))
            } else {
                Ok(name)
            }
        })
        .collect::<Result<Vec<_>, _>>()?
        .join(", ");
    let mut sql = format!("SELECT {columns} FROM {}", builder.name(&query.entity)?);
    if let Some(filter) = &query.filter {
        sql.push_str(" WHERE ");
        sql.push_str(&builder.filter(declaration, filter)?);
    }
    if !query.order.is_empty() {
        sql.push_str(" ORDER BY ");
        let order = query
            .order
            .iter()
            .map(|key| {
                let name = builder.comparison_name(declaration, &key.field)?;
                let direction = if key.descending { "DESC" } else { "ASC" };
                let nullable = declaration
                    .fields
                    .iter()
                    .any(|field| field.name == key.field && field.nullable);
                if nullable && dialect == Dialect::Postgresql {
                    // PostgreSQL reverses the logical default. Use its native
                    // null-order clause so indexed sorting remains available.
                    let null_order = if key.descending { "LAST" } else { "FIRST" };
                    return Ok(format!("{name} {direction} NULLS {null_order}"));
                }
                Ok(format!("{name} {direction}"))
            })
            .collect::<Result<Vec<_>, DatabaseError>>()?;
        sql.push_str(&order.join(", "));
    }
    sql.push_str(&format!(" LIMIT {} OFFSET {}", query.limit, query.offset));
    Ok(builder.finish(sql))
}

/// Compiles an ordered, one-record claim selection with backend row locking.
pub fn compile_claim_select(
    dialect: Dialect,
    schema: &AppSchema,
    claim: &Claim,
) -> Result<Statement, DatabaseError> {
    validate_claim(schema, claim)?;
    let identity = entity(schema, &claim.entity)?
        .identity
        .generated_field()
        .ok_or(DatabaseError::InvalidInput)?
        .to_owned();
    let mut statement = compile_query(
        dialect,
        schema,
        &Query {
            entity: claim.entity.clone(),
            fields: vec![identity],
            filter: Some(claim.filter.clone()),
            order: claim.order.clone(),
            limit: 1,
            offset: 0,
        },
    )?;
    if dialect != Dialect::Sqlite {
        statement.sql.push_str(" FOR UPDATE SKIP LOCKED");
    }
    Ok(statement)
}

/// Compiles one count, sum, minimum, or maximum over declared fields.
pub fn compile_aggregate(
    dialect: Dialect,
    schema: &AppSchema,
    query: &AggregateQuery,
) -> Result<Statement, DatabaseError> {
    let declaration = entity(schema, &query.entity)?;
    let result_kind = crate::operations::aggregate_result_kind(schema, query)?;
    let mut builder = Builder::new(dialect);
    let expression = match &query.aggregate {
        Aggregate::Count => "COUNT(*)".to_owned(),
        Aggregate::Sum(field) | Aggregate::Minimum(field) | Aggregate::Maximum(field) => {
            if !declaration.fields.iter().any(|item| item.name == *field) {
                return Err(DatabaseError::InvalidInput);
            }
            let function = match &query.aggregate {
                Aggregate::Sum(_) => "SUM",
                Aggregate::Minimum(_) => "MIN",
                Aggregate::Maximum(_) => "MAX",
                Aggregate::Count => unreachable!(),
            };
            let name = builder.comparison_name(declaration, field)?;
            let name = match (dialect, result_kind) {
                (Dialect::Postgresql, FieldType::Boolean) => format!("CAST({name} AS INTEGER)"),
                (Dialect::Postgresql, FieldType::Binary) => format!("encode({name}, 'hex')"),
                (Dialect::Mysql, FieldType::Binary) => format!("HEX({name})"),
                _ => name,
            };
            format!("{function}({name})")
        }
    };
    let expression = match dialect {
        Dialect::Postgresql => format!("CAST({expression} AS TEXT)"),
        Dialect::Mysql => format!("CAST({expression} AS CHAR)"),
        Dialect::Sqlite => expression,
    };
    let mut sql = format!("SELECT {expression} FROM {}", builder.name(&query.entity)?);
    if let Some(filter) = &query.filter {
        validate_predicate(declaration, filter)?;
        sql.push_str(" WHERE ");
        sql.push_str(&builder.filter(declaration, filter)?);
    }
    Ok(builder.finish(sql))
}

/// Resolves a transaction-local value from a previous insert result.
fn resolve_value(
    value: &WriteValue,
    generated_ids: &[Option<i64>],
) -> Result<Value, DatabaseError> {
    crate::operations::resolve_write_value(value, generated_ids)
}

/// Looks up the identity of an inserted or pre-existing unique record inside
/// the same transaction. The caller must hold its transaction until this read.
pub fn compile_ensured_identity_query(
    dialect: Dialect,
    schema: &AppSchema,
    name: &str,
    unique_key: &[String],
    values: &BTreeMap<String, WriteValue>,
    generated_ids: &[Option<i64>],
) -> Result<Statement, DatabaseError> {
    let declaration = entity(schema, name)?;
    let identity = declaration
        .identity
        .generated_field()
        .ok_or(DatabaseError::InvalidContract)?;
    if !declaration.accepts_upsert_key(unique_key) {
        return Err(DatabaseError::InvalidContract);
    }
    let filters = unique_key
        .iter()
        .map(|field| {
            let value = values.get(field).ok_or(DatabaseError::InvalidInput)?;
            Ok(Predicate::Equal(
                field.clone(),
                resolve_value(value, generated_ids)?,
            ))
        })
        .collect::<Result<Vec<_>, DatabaseError>>()?;
    compile_query(
        dialect,
        schema,
        &Query {
            entity: name.to_owned(),
            fields: vec![identity.to_owned()],
            filter: Some(Predicate::All(filters)),
            order: Vec::new(),
            limit: 1,
            offset: 0,
        },
    )
}

/// Prepared statements for one reservation decision under a backend transaction.
pub struct ReservationStatements {
    pub expire: Option<Statement>,
    pub owner: Statement,
    pub occupied: Statement,
    pub insert: Statement,
}

pub fn compile_reservation(
    dialect: Dialect,
    schema: &AppSchema,
    mutation: &Mutation,
    generated_ids: &[Option<i64>],
) -> Result<ReservationStatements, DatabaseError> {
    let Mutation::Reserve {
        entity: name,
        status_field,
        active_statuses,
        owner_filter,
        scope_filter,
        expiry,
        values,
        ..
    } = mutation
    else {
        return Err(DatabaseError::InvalidContract);
    };
    let identity = entity(schema, name)?
        .identity
        .generated_field()
        .ok_or(DatabaseError::InvalidContract)?
        .to_owned();
    let owner_filter = materialize_generated_predicate(owner_filter, generated_ids)?;
    let scope_filter = materialize_generated_predicate(scope_filter, generated_ids)?;
    let active = Predicate::In(status_field.clone(), active_statuses.clone());
    let owner = compile_query(
        dialect,
        schema,
        &Query {
            entity: name.clone(),
            fields: vec![identity],
            filter: Some(Predicate::All(vec![owner_filter.clone(), active.clone()])),
            order: Vec::new(),
            limit: 1,
            offset: 0,
        },
    )?;
    let occupied = compile_aggregate(
        dialect,
        schema,
        &AggregateQuery {
            entity: name.clone(),
            filter: Some(Predicate::All(vec![scope_filter.clone(), active])),
            aggregate: Aggregate::Count,
        },
    )?;
    let expire = expiry
        .as_ref()
        .map(|rule| {
            let mut values = BTreeMap::from([(
                status_field.clone(),
                WriteValue::Literal(rule.to_status.clone()),
            )]);
            if let Some((name, value)) = &rule.released_at {
                values.insert(name.clone(), WriteValue::Literal(value.clone()));
            }
            if let Some((name, value)) = &rule.reason {
                values.insert(name.clone(), WriteValue::Literal(value.clone()));
            }
            compile_mutation(
                dialect,
                schema,
                &Mutation::Update {
                    entity: name.clone(),
                    filter: crate::operations::reservation_expiry_filter(
                        scope_filter.clone(),
                        status_field,
                        rule,
                        generated_ids,
                    )?,
                    values,
                    expected_rows: None,
                },
                generated_ids,
            )
        })
        .transpose()?;
    let insert = compile_mutation(
        dialect,
        schema,
        &Mutation::Insert {
            entity: name.clone(),
            values: values.clone(),
        },
        generated_ids,
    )?;
    Ok(ReservationStatements {
        expire,
        owner,
        occupied,
        insert,
    })
}

/// Compiles one validated transaction step against previous generated IDs.
pub fn compile_mutation(
    dialect: Dialect,
    schema: &AppSchema,
    mutation: &Mutation,
    generated_ids: &[Option<i64>],
) -> Result<Statement, DatabaseError> {
    let name = match mutation {
        Mutation::Insert { entity, .. }
        | Mutation::Upsert { entity, .. }
        | Mutation::InsertIfAbsent { entity, .. }
        | Mutation::Update { entity, .. }
        | Mutation::Delete { entity, .. }
        | Mutation::RequireCount { entity, .. }
        | Mutation::Reserve { entity, .. } => entity,
    };
    let declaration = entity(schema, name)?;
    let mut builder = Builder::new(dialect);
    let table = builder.name(name)?;
    let sql = match mutation {
        Mutation::Insert { values, .. }
        | Mutation::Upsert { values, .. }
        | Mutation::InsertIfAbsent { values, .. } => {
            let mut columns = Vec::with_capacity(declaration.fields.len());
            let mut placeholders = Vec::with_capacity(declaration.fields.len());
            for field in &declaration.fields {
                let value = if let Some(value) = values.get(&field.name) {
                    Some(builder.bind_field(
                        field,
                        crate::operations::resolve_field_write(field, value, generated_ids)?,
                    )?)
                } else {
                    match &field.default {
                        Some(FieldDefault::Literal(value)) => {
                            Some(builder.bind_field(field, value.clone())?)
                        }
                        Some(FieldDefault::CurrentTimestamp) => {
                            Some(current_timestamp_sql(dialect).to_owned())
                        }
                        None => None,
                    }
                };
                if let Some(value) = value {
                    columns.push(builder.name(&field.name)?);
                    placeholders.push(value);
                }
            }
            for name in crate::index_values::derived_fields(declaration) {
                let field = declaration
                    .fields
                    .iter()
                    .find(|field| field.name == name)
                    .ok_or(DatabaseError::InvalidContract)?;
                let original = if let Some(value) = values.get(name) {
                    crate::operations::resolve_field_write(field, value, generated_ids)?
                } else if let Some(FieldDefault::Literal(value)) = &field.default {
                    value.clone()
                } else {
                    continue;
                };
                columns.push(builder.name(&crate::index_values::derived_key(declaration, name)?)?);
                placeholders.push(builder.bind(crate::index_values::derived_value(
                    declaration,
                    name,
                    &original,
                )?));
            }
            let columns = columns.join(", ");
            let placeholders = placeholders.join(", ");
            let mut sql = format!("INSERT INTO {table} ({columns}) VALUES ({placeholders})");
            if let Mutation::Upsert { unique_key, .. } = mutation {
                if !declaration.accepts_upsert_key(unique_key) {
                    return Err(DatabaseError::InvalidContract);
                }
                let mut assignments = values
                    .keys()
                    .filter(|field| !unique_key.contains(field))
                    .map(|field| {
                        let column = builder.name(field)?;
                        Ok(match dialect {
                            Dialect::Mysql => format!("{column} = VALUES({column})"),
                            Dialect::Sqlite | Dialect::Postgresql => {
                                format!("{column} = EXCLUDED.{column}")
                            }
                        })
                    })
                    .collect::<Result<Vec<_>, DatabaseError>>()?;
                for name in crate::index_values::derived_fields(declaration) {
                    if values.contains_key(name) && !unique_key.iter().any(|key| key == name) {
                        let column =
                            builder.name(&crate::index_values::derived_key(declaration, name)?)?;
                        assignments.push(match dialect {
                            Dialect::Mysql => format!("{column} = VALUES({column})"),
                            _ => format!("{column} = EXCLUDED.{column}"),
                        });
                    }
                }
                if assignments.is_empty() {
                    // A keys-only upsert ensures membership without rewriting defaults.
                    match dialect {
                        Dialect::Mysql => {
                            let key = builder.name(&unique_key[0])?;
                            sql.push_str(&format!(" ON DUPLICATE KEY UPDATE {key} = {key}"));
                        }
                        Dialect::Sqlite | Dialect::Postgresql => {
                            let key = unique_key
                                .iter()
                                .map(|field| builder.name(field))
                                .collect::<Result<Vec<_>, _>>()?
                                .join(", ");
                            sql.push_str(&format!(" ON CONFLICT ({key}) DO NOTHING"));
                        }
                    }
                    return Ok(builder.finish(sql));
                }
                match dialect {
                    Dialect::Mysql => sql.push_str(&format!(
                        " ON DUPLICATE KEY UPDATE {}",
                        assignments.join(", ")
                    )),
                    Dialect::Sqlite | Dialect::Postgresql => {
                        let key = unique_key
                            .iter()
                            .map(|field| builder.name(field))
                            .collect::<Result<Vec<_>, _>>()?
                            .join(", ");
                        sql.push_str(&format!(
                            " ON CONFLICT ({key}) DO UPDATE SET {}",
                            assignments.join(", ")
                        ));
                    }
                }
            }
            if let Mutation::InsertIfAbsent { unique_key, .. } = mutation {
                if !declaration.accepts_upsert_key(unique_key) {
                    return Err(DatabaseError::InvalidContract);
                }
                if dialect != Dialect::Mysql {
                    let key = unique_key
                        .iter()
                        .map(|field| builder.name(field))
                        .collect::<Result<Vec<_>, _>>()?
                        .join(", ");
                    sql.push_str(&format!(" ON CONFLICT ({key}) DO NOTHING"));
                }
            }
            sql
        }
        Mutation::Update { values, filter, .. } => {
            validate_predicate(declaration, filter)?;
            let filter = materialize_generated_predicate(filter, generated_ids)?;
            let mut assignments = values
                .iter()
                .map(|(field, value)| {
                    let column = builder.name(field)?;
                    let declared_field = declaration
                        .fields
                        .iter()
                        .find(|item| item.name == *field)
                        .ok_or(DatabaseError::InvalidContract)?;
                    let placeholder = builder.bind_field(
                        declared_field,
                        crate::operations::resolve_field_write(
                            declared_field,
                            value,
                            generated_ids,
                        )?,
                    )?;
                    Ok(format!("{column} = {placeholder}"))
                })
                .collect::<Result<Vec<_>, DatabaseError>>()?;
            for name in crate::index_values::derived_fields(declaration) {
                if let Some(value) = values.get(name) {
                    let field = declaration
                        .fields
                        .iter()
                        .find(|field| field.name == name)
                        .ok_or(DatabaseError::InvalidContract)?;
                    let value =
                        crate::operations::resolve_field_write(field, value, generated_ids)?;
                    let key =
                        builder.name(&crate::index_values::derived_key(declaration, name)?)?;
                    let binding = builder.bind(crate::index_values::derived_value(
                        declaration,
                        name,
                        &value,
                    )?);
                    assignments.push(format!("{key} = {binding}"));
                }
            }
            let assignments = assignments.join(", ");
            let condition = builder.filter(declaration, &filter)?;
            format!("UPDATE {table} SET {assignments} WHERE {condition}")
        }
        Mutation::Delete { filter, .. } | Mutation::RequireCount { filter, .. } => {
            validate_predicate(declaration, filter)?;
            let filter = materialize_generated_predicate(filter, generated_ids)?;
            let condition = builder.filter(declaration, &filter)?;
            match mutation {
                Mutation::Delete { .. } => format!("DELETE FROM {table} WHERE {condition}"),
                Mutation::RequireCount { .. } => {
                    format!("SELECT COUNT(*) FROM {table} WHERE {condition}")
                }
                _ => unreachable!(),
            }
        }
        Mutation::Reserve { .. } => return Err(DatabaseError::InvalidContract),
    };
    Ok(builder.finish(sql))
}

/// Keeps JSON relationship values synchronized when native constraints null their keys.
/// The caller executes these bounded statements in the same transaction as deletion.
pub fn compile_delete_effects(
    dialect: Dialect,
    schema: &AppSchema,
    mutation: &Mutation,
    generated_ids: &[Option<i64>],
) -> Result<Vec<Statement>, DatabaseError> {
    let Mutation::Delete {
        entity: name,
        filter,
        ..
    } = mutation
    else {
        return Ok(Vec::new());
    };
    let declaration = entity(schema, name)?;
    validate_predicate(declaration, filter)?;
    let mut builder = Builder::new(dialect);
    let condition = builder.filter(
        declaration,
        &materialize_generated_predicate(filter, generated_ids)?,
    )?;
    let mut pending = vec![(name.clone(), condition, BTreeSet::new())];
    let mut statements = Vec::new();
    let mut traversed = 0usize;
    while let Some((parent_name, condition, mut path)) = pending.pop() {
        if !path.insert(parent_name.clone()) {
            continue;
        }
        traversed += 1;
        if traversed > 256 {
            return Err(DatabaseError::CapacityExceeded);
        }
        let parent = entity(schema, &parent_name)?;
        for child in &schema.entities {
            for field in &child.fields {
                let Some((target, key)) = &field.references else {
                    continue;
                };
                if target != &parent_name {
                    continue;
                }
                let source = builder.name(&crate::index_values::reference_keys(field)?.0)?;
                let selected = format!(
                    "SELECT {} FROM {} WHERE {condition}",
                    builder.comparison_name(parent, key)?,
                    builder.name(&parent_name)?
                );
                let selected = if dialect == Dialect::Mysql {
                    format!("SELECT * FROM ({selected}) AS _pharos_delete_source")
                } else {
                    selected
                };
                let child_condition = format!("{source} IN ({selected})");
                match field.on_delete.unwrap_or(OnDelete::Restrict) {
                    OnDelete::Cascade => {
                        pending.push((child.name.clone(), child_condition, path.clone()))
                    }
                    OnDelete::SetNull if field.kind == FieldType::Json => {
                        let sql = format!(
                            "UPDATE {} SET {} = NULL, {source} = NULL WHERE {child_condition}",
                            builder.name(&child.name)?,
                            builder.name(&field.name)?
                        );
                        statements.push(Statement {
                            sql,
                            parameters: builder.parameters.clone(),
                            binary_parameters: builder.binary_parameters.clone(),
                        });
                    }
                    _ => {}
                }
            }
        }
    }
    Ok(statements)
}

/// Lowers referential delete actions when an adapter temporarily detaches native constraints.
/// Descendants are removed before their parents; the adapter verifies restrictive references
/// before committing. Every predicate retains the ordinary typed parameter bindings.
pub fn compile_detached_delete_effects(
    dialect: Dialect,
    schema: &AppSchema,
    name: &str,
    filter: &Predicate,
) -> Result<Vec<Statement>, DatabaseError> {
    let declaration = entity(schema, name)?;
    validate_predicate(declaration, filter)?;
    let mut builder = Builder::new(dialect);
    let condition = builder.filter(declaration, filter)?;
    let mut pending = vec![(name.to_owned(), condition, BTreeSet::new(), false)];
    let mut statements = Vec::new();
    let mut traversed = 0usize;
    while let Some((parent_name, condition, mut path, delete)) = pending.pop() {
        if delete {
            statements.push(Statement {
                sql: format!(
                    "DELETE FROM {} WHERE {condition}",
                    builder.name(&parent_name)?
                ),
                parameters: builder.parameters.clone(),
                binary_parameters: builder.binary_parameters.clone(),
            });
            continue;
        }
        if !path.insert(parent_name.clone()) {
            return Err(DatabaseError::Conflict);
        }
        traversed += 1;
        if traversed > 256 {
            return Err(DatabaseError::CapacityExceeded);
        }
        let parent = entity(schema, &parent_name)?;
        if parent_name != name {
            pending.push((parent_name.clone(), condition.clone(), path.clone(), true));
        }
        for child in &schema.entities {
            for field in &child.fields {
                let Some((target, key)) = &field.references else {
                    continue;
                };
                if target != &parent_name {
                    continue;
                }
                let source = builder.name(&crate::index_values::reference_keys(field)?.0)?;
                let selected = format!(
                    "SELECT {} FROM {} WHERE {condition}",
                    builder.comparison_name(parent, key)?,
                    builder.name(&parent_name)?
                );
                let child_condition = format!("{source} IN ({selected})");
                match field.on_delete.unwrap_or(OnDelete::Restrict) {
                    OnDelete::Cascade => {
                        pending.push((child.name.clone(), child_condition, path.clone(), false))
                    }
                    OnDelete::SetNull => {
                        let column = builder.name(&field.name)?;
                        let assignments = if column == source {
                            format!("{column} = NULL")
                        } else {
                            format!("{column} = NULL, {source} = NULL")
                        };
                        statements.push(Statement {
                            sql: format!(
                                "UPDATE {} SET {assignments} WHERE {child_condition}",
                                builder.name(&child.name)?
                            ),
                            parameters: builder.parameters.clone(),
                            binary_parameters: builder.binary_parameters.clone(),
                        });
                    }
                    OnDelete::Restrict => {}
                }
            }
        }
    }
    Ok(statements)
}

/// Resolves logical record fields to safe SQL bind values before import.
pub fn record_values(fields: &serde_json::Map<String, Value>) -> BTreeMap<String, WriteValue> {
    fields
        .iter()
        .map(|(name, value)| (name.clone(), WriteValue::Literal(value.clone())))
        .collect()
}

/// Declares a nullable physical key whose Unicode mapping is owned by Rust.
pub fn derived_column(
    dialect: Dialect,
    declaration: &Entity,
    name: &str,
) -> Result<String, DatabaseError> {
    column(dialect, declaration, &derived_field(declaration, name)?)
}

/// Describes a protected comparison column for creation and interrupted-DDL verification.
pub fn derived_field(declaration: &Entity, name: &str) -> Result<Field, DatabaseError> {
    let source = declaration
        .fields
        .iter()
        .find(|field| field.name == name)
        .ok_or(DatabaseError::InvalidContract)?;
    let mut key = source.clone();
    key.name = crate::index_values::derived_key(declaration, name)?;
    key.kind = FieldType::Text;
    key.max_length = if source.kind == FieldType::Json {
        source.max_length
    } else {
        source
            .max_length
            .and_then(|length| length.checked_mul(crate::index_values::LOWERCASE_EXPANSION))
    };
    key.default = match &source.default {
        Some(FieldDefault::Literal(value)) => Some(FieldDefault::Literal(
            crate::index_values::derived_value(declaration, name, value)?,
        )),
        None => None,
        Some(FieldDefault::CurrentTimestamp) => return Err(DatabaseError::InvalidContract),
    };
    key.references = None;
    key.constraints.clear();
    key.nullable = true;
    Ok(key)
}

/// Builds an ordered, bounded migration page for expression-key adoption.
pub fn derived_adoption_query(
    declaration: &Entity,
    after: Option<Value>,
) -> Result<Query, DatabaseError> {
    let mut fields = declaration
        .identity
        .fields()
        .into_iter()
        .map(str::to_owned)
        .collect::<Vec<_>>();
    for name in crate::index_values::derived_fields(declaration) {
        if !fields.iter().any(|field| field == name) {
            fields.push(name.to_owned());
        }
    }
    let (filter, order) = crate::operations::identity_page(declaration, after)?;
    Ok(Query {
        entity: declaration.name.clone(),
        fields,
        filter,
        order,
        limit: 1000,
        offset: 0,
    })
}

/// Repairs expression keys from a locked migration page without changing original values.
pub fn derived_adoption_statements(
    dialect: Dialect,
    declaration: &Entity,
    rows: &[serde_json::Map<String, Value>],
) -> Result<Vec<Statement>, DatabaseError> {
    rows.iter()
        .map(|row| {
            let mut builder = Builder::new(dialect);
            let mut assignments = Vec::new();
            for name in crate::index_values::derived_fields(declaration) {
                let key = builder.name(&crate::index_values::derived_key(declaration, name)?)?;
                let value = crate::index_values::derived_value(
                    declaration,
                    name,
                    row.get(name).ok_or(DatabaseError::StorageFailure)?,
                )?;
                let binding = builder.bind(value);
                assignments.push(format!("{key} = {binding}"));
            }
            let identity = Predicate::All(
                declaration
                    .identity
                    .fields()
                    .into_iter()
                    .map(|name| {
                        row.get(name)
                            .cloned()
                            .map(|value| Predicate::Equal(name.to_owned(), value))
                            .ok_or(DatabaseError::StorageFailure)
                    })
                    .collect::<Result<Vec<_>, _>>()?,
            );
            let condition = builder.filter(declaration, &identity)?;
            let table = builder.name(&declaration.name)?;
            Ok(builder.finish(format!(
                "UPDATE {table} SET {} WHERE {condition}",
                assignments.join(", ")
            )))
        })
        .collect()
}

/// Extracts the next stable identity cursor from a bounded adoption page.
pub fn adoption_cursor(
    declaration: &Entity,
    row: &serde_json::Map<String, Value>,
) -> Result<Value, DatabaseError> {
    match &declaration.identity {
        Identity::AutoInteger(name) => row.get(name).cloned().ok_or(DatabaseError::StorageFailure),
        Identity::Composite(names) => names
            .iter()
            .map(|name| {
                row.get(name)
                    .cloned()
                    .map(|value| (name.clone(), value))
                    .ok_or(DatabaseError::StorageFailure)
            })
            .collect::<Result<serde_json::Map<String, Value>, _>>()
            .map(Value::Object),
    }
}

/// Selects one bounded page for a typed field-copy migration.
pub fn field_copy_query(
    schema: &AppSchema,
    change: &SchemaChange,
    after: Option<Value>,
) -> Result<Query, DatabaseError> {
    let SchemaChange::CopyField {
        entity: name,
        source,
        target,
        filter,
    } = change
    else {
        return Err(DatabaseError::InvalidContract);
    };
    let declaration = entity(schema, name)?;
    validate_copy_field(declaration, source, target, filter)?;
    let (cursor, order) = crate::operations::identity_page(declaration, after)?;
    let mut fields = declaration
        .identity
        .fields()
        .into_iter()
        .map(str::to_owned)
        .collect::<Vec<_>>();
    if !fields.contains(source) {
        fields.push(source.clone());
    }
    Ok(Query {
        entity: name.clone(),
        fields,
        filter: Some(match cursor {
            Some(cursor) => Predicate::All(vec![filter.clone(), cursor]),
            None => filter.clone(),
        }),
        order,
        limit: 1000,
        offset: 0,
    })
}

/// Uses ordinary typed mutation lowering for historical copy values and derived keys.
pub fn field_copy_statements(
    dialect: Dialect,
    schema: &AppSchema,
    change: &SchemaChange,
    rows: &[serde_json::Map<String, Value>],
) -> Result<Vec<Statement>, DatabaseError> {
    let SchemaChange::CopyField {
        entity: name,
        source,
        target,
        ..
    } = change
    else {
        return Err(DatabaseError::InvalidContract);
    };
    let declaration = entity(schema, name)?;
    rows.iter()
        .map(|row| {
            let filter = Predicate::All(
                declaration
                    .identity
                    .fields()
                    .into_iter()
                    .map(|name| {
                        row.get(name)
                            .cloned()
                            .map(|value| Predicate::Equal(name.to_owned(), value))
                            .ok_or(DatabaseError::StorageFailure)
                    })
                    .collect::<Result<Vec<_>, _>>()?,
            );
            compile_mutation(
                dialect,
                schema,
                &Mutation::Update {
                    entity: name.clone(),
                    filter,
                    values: BTreeMap::from([(
                        target.clone(),
                        WriteValue::Literal(
                            row.get(source)
                                .cloned()
                                .ok_or(DatabaseError::StorageFailure)?,
                        ),
                    )]),
                    expected_rows: Some(1),
                },
                &[],
            )
        })
        .collect()
}

/// Maps a logical field to one backend-specific physical scalar type.
/// Returns the physical field declaration used by schema creation and catalog verification.
pub fn sql_type(dialect: Dialect, field: &Field) -> Result<String, DatabaseError> {
    Ok(match (dialect, field.kind) {
        (Dialect::Mysql, FieldType::Text) => {
            let storage = match field.max_length {
                Some(length) if length > 0 && length <= 4096 => format!("VARCHAR({length})"),
                None | Some(4097..=MAX_TEXT_LENGTH) => "MEDIUMTEXT".to_owned(),
                Some(_) => return Err(DatabaseError::InvalidContract),
            };
            format!("{storage} CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin")
        }
        (dialect, FieldType::Text) => {
            let storage = match field.max_length {
                Some(length) if length > 0 && length <= MAX_TEXT_LENGTH => {
                    format!("VARCHAR({length})")
                }
                Some(_) => return Err(DatabaseError::InvalidContract),
                None => "TEXT".to_owned(),
            };
            match dialect {
                Dialect::Postgresql => format!("{storage} COLLATE \"C\""),
                Dialect::Sqlite => format!("{storage} COLLATE BINARY"),
                Dialect::Mysql => unreachable!("MySQL text storage is handled above"),
            }
        }
        (Dialect::Sqlite, FieldType::Integer | FieldType::Boolean) => "INTEGER".to_owned(),
        (Dialect::Sqlite, FieldType::Decimal) => "NUMERIC".to_owned(),
        (Dialect::Sqlite, FieldType::Timestamp | FieldType::Json) => "TEXT".to_owned(),
        (Dialect::Sqlite, FieldType::Binary) => "BLOB".to_owned(),
        (Dialect::Postgresql, FieldType::Integer) => "BIGINT".to_owned(),
        (Dialect::Postgresql, FieldType::Decimal) => "NUMERIC(10,2)".to_owned(),
        (Dialect::Postgresql, FieldType::Boolean) => "BOOLEAN".to_owned(),
        (Dialect::Postgresql, FieldType::Timestamp) => "VARCHAR(30) COLLATE \"C\"".to_owned(),
        (Dialect::Postgresql, FieldType::Json) => "JSONB".to_owned(),
        (Dialect::Postgresql, FieldType::Binary) => "BYTEA".to_owned(),
        (Dialect::Mysql, FieldType::Integer) => "BIGINT".to_owned(),
        (Dialect::Mysql, FieldType::Decimal) => "DECIMAL(10,2)".to_owned(),
        (Dialect::Mysql, FieldType::Boolean) => "BOOLEAN".to_owned(),
        (Dialect::Mysql, FieldType::Timestamp) => {
            "VARCHAR(30) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin".to_owned()
        }
        (Dialect::Mysql, FieldType::Json) => "JSON".to_owned(),
        (Dialect::Mysql, FieldType::Binary) => match field.max_length {
            Some(length) if length <= 65535 => format!("VARBINARY({length})"),
            _ => "LONGBLOB".to_owned(),
        },
    })
}

/// Encodes a schema-owned literal in DDL; request values never use this path.
fn ddl_literal(dialect: Dialect, value: &Value) -> Result<String, DatabaseError> {
    match value {
        Value::Null => Ok("NULL".to_owned()),
        Value::Bool(value) => Ok(match (dialect, value) {
            (Dialect::Postgresql, true) => "TRUE",
            (Dialect::Postgresql, false) => "FALSE",
            (_, true) => "1",
            (_, false) => "0",
        }
        .to_owned()),
        Value::Number(value) => Ok(value.to_string()),
        Value::String(value) if value.len() <= 4096 && dialect == Dialect::Mysql => {
            // Hexadecimal UTF-8 literals avoid both SQL-mode-dependent escaping
            // and client-side quote scanners during prepared DDL execution.
            let mut encoded = String::with_capacity(value.len() * 2);
            const DIGITS: &[u8; 16] = b"0123456789abcdef";
            for byte in value.bytes() {
                encoded.push(char::from(DIGITS[usize::from(byte >> 4)]));
                encoded.push(char::from(DIGITS[usize::from(byte & 15)]));
            }
            Ok(format!("_utf8mb4 X'{encoded}'"))
        }
        Value::String(value) if value.len() <= 4096 => {
            Ok(format!("'{}'", value.replace('\'', "''")))
        }
        _ => Err(DatabaseError::InvalidContract),
    }
}

/// Keeps generated timestamps consistent in inserts, schema defaults and backfills.
pub fn current_timestamp_sql(dialect: Dialect) -> &'static str {
    match dialect {
        Dialect::Sqlite => "(strftime('%Y-%m-%dT%H:%M:%f', 'now') || '000000Z')",
        Dialect::Mysql => "(CONCAT(DATE_FORMAT(UTC_TIMESTAMP(6), '%Y-%m-%dT%H:%i:%s.%f'), '000Z'))",
        Dialect::Postgresql => {
            "(to_char(CURRENT_TIMESTAMP AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') || '000Z')"
        }
    }
}

/// Lowers one declared default for schema creation and physical storage adoption.
pub fn field_default_sql(dialect: Dialect, field: &Field) -> Result<Option<String>, DatabaseError> {
    field
        .default
        .as_ref()
        .map(|default| match default {
            FieldDefault::Literal(value)
                if dialect == Dialect::Mysql
                    && field.kind == FieldType::Text
                    && field.max_length.is_none_or(|length| length > 4096) =>
            {
                Ok(format!("({})", ddl_literal(dialect, value)?))
            }
            FieldDefault::Literal(value) => ddl_field_literal(dialect, field, value),
            FieldDefault::CurrentTimestamp => Ok(current_timestamp_sql(dialect).to_owned()),
        })
        .transpose()
}

fn ddl_field_literal(
    dialect: Dialect,
    field: &Field,
    value: &Value,
) -> Result<String, DatabaseError> {
    if field.kind == FieldType::Binary && !value.is_null() {
        use std::fmt::Write;
        let bytes = value.as_array().ok_or(DatabaseError::InvalidContract)?;
        if bytes.len() > 1_048_576 {
            return Err(DatabaseError::CapacityExceeded);
        }
        let mut encoded = String::with_capacity(bytes.len() * 2);
        for byte in bytes {
            let byte = byte
                .as_u64()
                .filter(|byte| *byte <= 255)
                .ok_or(DatabaseError::InvalidContract)?;
            write!(&mut encoded, "{byte:02x}").map_err(|_| DatabaseError::InvalidContract)?;
        }
        return Ok(match dialect {
            Dialect::Postgresql => format!("decode('{encoded}', 'hex')"),
            Dialect::Sqlite | Dialect::Mysql => format!("X'{encoded}'"),
        });
    }
    if field.kind == FieldType::Timestamp && !value.is_null() {
        let timestamp = crate::operations::canonical_stored_timestamp(
            value.as_str().ok_or(DatabaseError::InvalidContract)?,
        )?;
        return ddl_literal(dialect, &Value::String(timestamp));
    }
    if field.kind == FieldType::Json && !value.is_null() {
        let canonical = crate::operations::canonical_json_value(value)?;
        let serialized =
            serde_json::to_string(&canonical).map_err(|_| DatabaseError::InvalidContract)?;
        let encoded = ddl_literal(dialect, &Value::String(serialized))?;
        return Ok(match dialect {
            Dialect::Mysql => format!("(CAST({encoded} AS JSON))"),
            Dialect::Sqlite | Dialect::Postgresql => encoded,
        });
    }
    if field.kind == FieldType::Decimal && !value.is_null() {
        let cents = decimal_cents(value)?;
        return Ok(match dialect {
            Dialect::Sqlite | Dialect::Postgresql | Dialect::Mysql => decimal_from_cents(cents),
        });
    }
    ddl_literal(dialect, value)
}

/// Produces one physical column declaration without accepting raw SQL.
pub fn column(dialect: Dialect, entity: &Entity, field: &Field) -> Result<String, DatabaseError> {
    let builder = Builder::new(dialect);
    let name = builder.name(&field.name)?;
    if entity.identity.generated_field() == Some(field.name.as_str()) {
        return Ok(format!(
            "{name} {}",
            match dialect {
                Dialect::Sqlite => "INTEGER PRIMARY KEY AUTOINCREMENT",
                Dialect::Postgresql => "BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY",
                Dialect::Mysql => "BIGINT AUTO_INCREMENT PRIMARY KEY",
            }
        ));
    }
    let mut sql = format!("{name} {}", sql_type(dialect, field)?);
    if !field.nullable {
        sql.push_str(" NOT NULL");
    }
    if let Some(default) = field_default_sql(dialect, field)? {
        sql.push_str(" DEFAULT ");
        sql.push_str(&default);
    }
    if dialect != Dialect::Mysql
        && field.kind != FieldType::Json
        && let Some((parent, parent_field)) = &field.references
    {
        sql.push_str(&format!(
            " REFERENCES {}({})",
            builder.name(parent)?,
            builder.name(parent_field)?
        ));
        sql.push_str(match field.on_delete.unwrap_or(OnDelete::Restrict) {
            OnDelete::Restrict => " ON DELETE RESTRICT",
            OnDelete::Cascade => " ON DELETE CASCADE",
            OnDelete::SetNull => " ON DELETE SET NULL",
        });
    }
    for constraint in &field.constraints {
        let name = builder.name(&field.name)?;
        let check = match constraint {
            FieldConstraint::Allowed(values) => {
                let values = values
                    .iter()
                    .map(|value| ddl_field_literal(dialect, field, value))
                    .collect::<Result<Vec<_>, _>>()?
                    .join(", ");
                format!("{name} IN ({values})")
            }
            FieldConstraint::LengthBetween { minimum, maximum } => {
                let function = if dialect == Dialect::Mysql {
                    "CHAR_LENGTH"
                } else {
                    "LENGTH"
                };
                format!("{function}({name}) BETWEEN {minimum} AND {maximum}")
            }
            FieldConstraint::GreaterThan(value) => {
                format!("{name} > {}", ddl_field_literal(dialect, field, value)?)
            }
            FieldConstraint::AtMost(value) => {
                format!("{name} <= {}", ddl_field_literal(dialect, field, value)?)
            }
            FieldConstraint::Between { minimum, maximum } => format!(
                "{name} BETWEEN {} AND {}",
                ddl_field_literal(dialect, field, minimum)?,
                ddl_field_literal(dialect, field, maximum)?
            ),
        };
        sql.push_str(&format!(" CHECK ({check})"));
    }
    Ok(sql)
}

/// Emits a MySQL table constraint because inline REFERENCES does not enforce it.
fn foreign_key_clause(
    dialect: Dialect,
    table: &str,
    field: &Field,
) -> Result<Option<String>, DatabaseError> {
    let Some(_) = &field.references else {
        return Ok(None);
    };
    let builder = Builder::new(dialect);
    let constraint =
        crate::operations::generated_index_name("fk", table, std::slice::from_ref(&field.name))?;
    let constraint = builder.name(&constraint)?;
    let (physical_source, physical_parent, physical_key) =
        crate::index_values::reference_keys(field)?;
    let source = builder.name(&physical_source)?;
    let target = builder.name(&physical_parent)?;
    let target_field = builder.name(&physical_key)?;
    let on_delete = match field.on_delete.unwrap_or(OnDelete::Restrict) {
        OnDelete::Restrict => "RESTRICT",
        OnDelete::Cascade => "CASCADE",
        OnDelete::SetNull => "SET NULL",
    };
    Ok(Some(format!(
        "CONSTRAINT {constraint} FOREIGN KEY ({source}) REFERENCES {target}({target_field}) ON DELETE {on_delete}"
    )))
}

/// Encodes a schema-owned partial-index condition without request interpolation.
fn ddl_filter(
    builder: &Builder,
    declaration: &Entity,
    predicate: &Predicate,
) -> Result<String, DatabaseError> {
    let literal = |name: &str, value: &Value| {
        let field = declaration
            .fields
            .iter()
            .find(|field| field.name == name)
            .ok_or(DatabaseError::InvalidContract)?;
        if field.kind == FieldType::Json {
            ddl_literal(
                builder.dialect,
                &crate::index_values::json_comparison_value(value)?,
            )
        } else {
            ddl_field_literal(builder.dialect, field, value)
        }
    };
    let compare = |name: &str, operator: &str, value: &Value| {
        Ok::<_, DatabaseError>(format!(
            "{} {operator} {}",
            builder.comparison_name(declaration, name)?,
            literal(name, value)?
        ))
    };
    match predicate {
        Predicate::Equal(name, value) if value.is_null() => Ok(format!(
            "{} IS NULL",
            builder.comparison_name(declaration, name)?
        )),
        Predicate::NotEqual(name, value) if value.is_null() => Ok(format!(
            "{} IS NOT NULL",
            builder.comparison_name(declaration, name)?
        )),
        Predicate::Equal(name, value) => compare(name, "=", value),
        Predicate::GeneratedEqual(_, _) => Err(DatabaseError::InvalidContract),
        Predicate::NotEqual(name, value) => compare(name, "<>", value),
        Predicate::LessThan(name, value) => compare(name, "<", value),
        Predicate::AtMost(name, value) => compare(name, "<=", value),
        Predicate::GreaterThan(name, value) => compare(name, ">", value),
        Predicate::AtLeast(name, value) => compare(name, ">=", value),
        Predicate::Contains(name, needle) => {
            let name = builder.name(name)?;
            let needle = ddl_literal(builder.dialect, &Value::String(needle.clone()))?;
            Ok(match builder.dialect {
                Dialect::Mysql => format!("LOCATE(BINARY {needle}, BINARY {name}) > 0"),
                Dialect::Sqlite => format!("INSTR({name}, {needle}) > 0"),
                Dialect::Postgresql => format!("STRPOS({name}, {needle}) > 0"),
            })
        }
        Predicate::IsNull(name) => Ok(format!(
            "{} IS NULL",
            builder.comparison_name(declaration, name)?
        )),
        Predicate::In(name, values) => {
            if values.is_empty() {
                return Ok("1 = 0".to_owned());
            }
            let values = values
                .iter()
                .map(|value| literal(name, value))
                .collect::<Result<Vec<_>, _>>()?
                .join(", ");
            Ok(format!(
                "{} IN ({values})",
                builder.comparison_name(declaration, name)?
            ))
        }
        Predicate::All(parts) | Predicate::Any(parts) => {
            let separator = if matches!(predicate, Predicate::All(_)) {
                " AND "
            } else {
                " OR "
            };
            let parts = parts
                .iter()
                .map(|part| ddl_filter(builder, declaration, part))
                .collect::<Result<Vec<_>, _>>()?;
            Ok(format!("({})", parts.join(separator)))
        }
    }
}

/// Compiles one declared conditional or case-folded index.
fn conditional_index(
    dialect: Dialect,
    declaration: &Entity,
    index: &Index,
) -> Result<Statement, DatabaseError> {
    expression_index(dialect, declaration, index, false)
}

/// Describes the retired native-lowercase encoding solely for validated storage adoption.
pub fn previous_expression_index(
    dialect: Dialect,
    declaration: &Entity,
    index: &Index,
) -> Result<Statement, DatabaseError> {
    expression_index(dialect, declaration, index, true)
}

/// Lowers an index with an explicit physical encoding; runtime writes use protected keys.
fn expression_index(
    dialect: Dialect,
    declaration: &Entity,
    index: &Index,
    previous: bool,
) -> Result<Statement, DatabaseError> {
    let builder = Builder::new(dialect);
    if !valid_identifier(&index.name) || index.keys.is_empty() {
        return Err(DatabaseError::InvalidContract);
    }
    let condition = index
        .filter
        .as_ref()
        .map(|filter| {
            crate::operations::validate_predicate(declaration, filter)?;
            ddl_filter(&builder, declaration, filter)
        })
        .transpose()?;
    let columns = index
        .keys
        .iter()
        .map(|key| {
            let expression = match key {
                IndexKey::Field(name) => builder.comparison_name(declaration, name)?,
                IndexKey::Lowercase(name) => {
                    if previous {
                        format!("LOWER({})", builder.name(name)?)
                    } else {
                        builder.name(&crate::index_values::lowercase_key(name))?
                    }
                }
            };
            Ok::<_, DatabaseError>(match (dialect, &condition) {
                (Dialect::Mysql, Some(filter)) => {
                    format!("(CASE WHEN {filter} THEN {expression} ELSE NULL END)")
                }
                (Dialect::Mysql, None) if previous && matches!(key, IndexKey::Lowercase(_)) => {
                    format!("({expression})")
                }
                _ => expression,
            })
        })
        .collect::<Result<Vec<_>, DatabaseError>>()?
        .join(", ");
    let mut sql = format!(
        "CREATE {}INDEX {} ON {} ({columns})",
        if index.unique { "UNIQUE " } else { "" },
        builder.name(&index.name)?,
        builder.name(&declaration.name)?
    );
    if let Some(filter) = &condition
        && dialect != Dialect::Mysql
    {
        sql.push_str(" WHERE ");
        sql.push_str(filter);
    }
    Ok(builder.finish(sql))
}

/// Compiles one additive logical migration step into SQL statements.
/// The executing adapter must guard these statements with its migration ledger.
pub fn compile_schema_change(
    dialect: Dialect,
    schema: &AppSchema,
    change: &SchemaChange,
) -> Result<Vec<Statement>, DatabaseError> {
    let builder = Builder::new(dialect);
    match change {
        // Historical copy requires source data and therefore goes through
        // historical::source_query and historical::insert_statements.
        SchemaChange::SeedFrom { .. }
        | SchemaChange::CopyJsonRecords { .. }
        | SchemaChange::RetireFieldConstraint { .. }
        | SchemaChange::ConvertField { .. } => Err(DatabaseError::InvalidInput),
        SchemaChange::WidenIntegerToDecimal {
            entity: name,
            field,
        } => {
            let declaration = entity(schema, name)?;
            let column_field = declaration
                .fields
                .iter()
                .find(|item| item.name == *field)
                .ok_or(DatabaseError::InvalidContract)?;
            if column_field.kind != FieldType::Decimal
                || column_field.references.is_some()
                || declaration.identity.fields().contains(&field.as_str())
            {
                return Err(DatabaseError::InvalidContract);
            }
            let table = builder.name(name)?;
            let name = builder.name(field)?;
            match dialect {
                // SQLite's numeric affinity preserves existing integer values
                // and accepts exact decimal writes without a table rebuild.
                Dialect::Sqlite => Ok(Vec::new()),
                Dialect::Postgresql => Ok(vec![builder.finish(format!("ALTER TABLE {table} ALTER COLUMN {name} TYPE NUMERIC(10,2) USING {name}::NUMERIC(10,2)"))]),
                Dialect::Mysql => Ok(vec![builder.finish(format!("ALTER TABLE {table} MODIFY COLUMN {}", column(dialect, declaration, column_field)?))]),
            }
        }
        SchemaChange::CreateEntity(declaration) => {
            if !schema
                .entities
                .iter()
                .any(|entity| entity.name == declaration.name)
            {
                return Err(DatabaseError::InvalidContract);
            }
            let mut columns = declaration
                .fields
                .iter()
                .map(|field| column(dialect, declaration, field))
                .collect::<Result<Vec<_>, _>>()?;
            for name in crate::index_values::derived_fields(declaration) {
                columns.push(derived_column(dialect, declaration, name)?);
            }
            if dialect == Dialect::Mysql
                || declaration
                    .fields
                    .iter()
                    .any(|field| field.kind == FieldType::Json && field.references.is_some())
            {
                columns.extend(
                    declaration
                        .fields
                        .iter()
                        .filter(|field| dialect == Dialect::Mysql || field.kind == FieldType::Json)
                        .map(|field| foreign_key_clause(dialect, &declaration.name, field))
                        .collect::<Result<Vec<_>, _>>()?
                        .into_iter()
                        .flatten(),
                );
            }
            if let Identity::Composite(names) = &declaration.identity {
                let key = names
                    .iter()
                    .map(|name| builder.comparison_name(declaration, name))
                    .collect::<Result<Vec<_>, _>>()?
                    .join(", ");
                columns.push(format!("PRIMARY KEY ({key})"));
            }
            let columns = columns.join(", ");
            let table = builder.name(&declaration.name)?;
            let mut statements =
                vec![builder.finish(format!("CREATE TABLE IF NOT EXISTS {table} ({columns})"))];
            for fields in &declaration.unique {
                statements.extend(compile_schema_change(
                    dialect,
                    schema,
                    &SchemaChange::AddUnique {
                        entity: declaration.name.clone(),
                        fields: fields.clone(),
                    },
                )?);
            }
            for fields in &declaration.indexes {
                statements.extend(compile_schema_change(
                    dialect,
                    schema,
                    &SchemaChange::AddIndex {
                        entity: declaration.name.clone(),
                        fields: fields.clone(),
                    },
                )?);
            }
            for index in &declaration.conditional_indexes {
                statements.push(conditional_index(dialect, declaration, index)?);
            }
            Ok(statements)
        }
        SchemaChange::AddField {
            entity: name,
            field,
        } => {
            let declaration = entity(schema, name)?;
            if declaration.identity.fields().contains(&field.name.as_str()) {
                return Err(DatabaseError::InvalidContract);
            }
            let table = builder.name(name)?;
            let definition = column(dialect, declaration, field)?;
            let mut statements =
                vec![builder.finish(format!("ALTER TABLE {table} ADD COLUMN {definition}"))];
            if crate::index_values::derived_fields(declaration).contains(field.name.as_str()) {
                let mut definition = derived_column(dialect, declaration, &field.name)?;
                if dialect == Dialect::Sqlite
                    && field.kind == FieldType::Json
                    && field.references.is_some()
                {
                    let (_, parent, key) = crate::index_values::reference_keys(field)?;
                    let names = Builder::new(dialect);
                    let rule = match field.on_delete.unwrap_or(OnDelete::Restrict) {
                        OnDelete::Restrict => "RESTRICT",
                        OnDelete::Cascade => "CASCADE",
                        OnDelete::SetNull => "SET NULL",
                    };
                    definition.push_str(&format!(
                        " REFERENCES {}({}) ON DELETE {rule}",
                        names.name(&parent)?,
                        names.name(&key)?
                    ));
                }
                statements.push(
                    Builder::new(dialect)
                        .finish(format!("ALTER TABLE {table} ADD COLUMN {definition}")),
                );
            }
            if (dialect == Dialect::Mysql
                || (dialect == Dialect::Postgresql && field.kind == FieldType::Json))
                && let Some(constraint) = foreign_key_clause(dialect, name, field)?
            {
                statements.push(
                    Builder::new(dialect).finish(format!("ALTER TABLE {table} ADD {constraint}")),
                );
            }
            Ok(statements)
        }
        SchemaChange::PromoteIdentity {
            entity: name,
            field,
        } => {
            let declaration = entity(schema, name)?;
            if declaration.identity.generated_field() != Some(field.as_str()) {
                return Err(DatabaseError::InvalidContract);
            }
            let table = builder.name(name)?;
            let column = builder.name(field)?;
            match dialect {
                // INTEGER PRIMARY KEY already generates rowids on SQLite.
                Dialect::Sqlite => Ok(Vec::new()),
                Dialect::Postgresql => {
                    let table_literal =
                        ddl_literal(dialect, &serde_json::Value::String(name.clone()))?;
                    let field_literal =
                        ddl_literal(dialect, &serde_json::Value::String(field.clone()))?;
                    Ok(vec![
                        builder.finish(format!("ALTER TABLE {table} ALTER COLUMN {column} ADD GENERATED BY DEFAULT AS IDENTITY")),
                        Builder::new(dialect).finish(format!("SELECT setval(pg_get_serial_sequence({table_literal}, {field_literal}), GREATEST(COALESCE((SELECT MAX({column}) FROM {table}), 0), 1), EXISTS(SELECT 1 FROM {table}))")),
                    ])
                }
                Dialect::Mysql => Ok(vec![builder.finish(format!(
                    "ALTER TABLE {table} MODIFY COLUMN {column} BIGINT NOT NULL AUTO_INCREMENT"
                ))]),
            }
        }
        SchemaChange::AddUnique {
            entity: name,
            fields,
        }
        | SchemaChange::AddIndex {
            entity: name,
            fields,
        } => {
            let declaration = entity(schema, name)?;
            if fields.is_empty()
                || fields.len() > 16
                || fields
                    .iter()
                    .any(|field| !declaration.fields.iter().any(|item| item.name == *field))
            {
                return Err(DatabaseError::InvalidContract);
            }
            let unique = matches!(change, SchemaChange::AddUnique { .. });
            let index_name = crate::operations::generated_index_name(
                if unique { "uq" } else { "ix" },
                name,
                fields,
            )?;
            let columns = fields
                .iter()
                .map(|field| builder.comparison_name(entity(schema, name)?, field))
                .collect::<Result<Vec<_>, _>>()?
                .join(", ");
            let sql = format!(
                "CREATE {}INDEX {} ON {} ({columns})",
                if unique { "UNIQUE " } else { "" },
                builder.name(&index_name)?,
                builder.name(name)?
            );
            Ok(vec![builder.finish(sql)])
        }
        SchemaChange::AddConditionalIndex {
            entity: name,
            index,
        } => Ok(vec![conditional_index(
            dialect,
            entity(schema, name)?,
            index,
        )?]),
        SchemaChange::Seed {
            entity: name,
            values,
        } => {
            let declaration = entity(schema, name)?;
            if values.is_empty()
                || values.len() > 64
                || values
                    .keys()
                    .any(|field| !declaration.fields.iter().any(|item| item.name == *field))
            {
                return Err(DatabaseError::InvalidContract);
            }
            let mut columns = values
                .keys()
                .map(|field| builder.name(field))
                .collect::<Result<Vec<_>, _>>()?
                .join(", ");
            let mut builder = builder;
            let mut placeholders = values
                .iter()
                .map(|(name, value)| {
                    let field = declaration
                        .fields
                        .iter()
                        .find(|item| item.name == *name)
                        .ok_or(DatabaseError::InvalidContract)?;
                    builder.bind_field(field, value.clone())
                })
                .collect::<Result<Vec<_>, _>>()?
                .join(", ");
            for source in crate::index_values::derived_fields(declaration) {
                if let Some(value) = values.get(source) {
                    columns.push_str(&format!(
                        ", {}",
                        builder.name(&crate::index_values::derived_key(declaration, source)?)?
                    ));
                    let binding = builder.bind(crate::index_values::derived_value(
                        declaration,
                        source,
                        value,
                    )?);
                    placeholders.push_str(&format!(", {binding}"));
                }
            }
            let table = builder.name(name)?;
            Ok(vec![builder.finish(format!(
                "INSERT INTO {table} ({columns}) VALUES ({placeholders})"
            ))])
        }
        SchemaChange::Backfill {
            entity: name,
            filter,
            values,
        } => {
            let declaration = entity(schema, name)?;
            if values.is_empty()
                || values.len() > 64
                || values
                    .keys()
                    .any(|field| !declaration.fields.iter().any(|item| item.name == *field))
            {
                return Err(DatabaseError::InvalidContract);
            }
            validate_predicate(declaration, filter)?;
            let values = values
                .iter()
                .map(|(name, value)| (name.clone(), WriteValue::Literal(value.clone())))
                .collect();
            Ok(vec![compile_mutation(
                dialect,
                schema,
                &Mutation::Update {
                    entity: name.clone(),
                    filter: filter.clone(),
                    values,
                    expected_rows: None,
                },
                &[],
            )?])
        }
        SchemaChange::DeleteWhere {
            entity: name,
            filter,
        } => {
            validate_predicate(entity(schema, name)?, filter)?;
            let mutation = Mutation::Delete {
                entity: name.clone(),
                filter: filter.clone(),
                expected_rows: None,
            };
            let mut statements = compile_delete_effects(dialect, schema, &mutation, &[])?;
            statements.push(compile_mutation(dialect, schema, &mutation, &[])?);
            Ok(statements)
        }
        SchemaChange::CopyField {
            entity: name,
            source,
            target,
            filter,
        } => {
            let declaration = entity(schema, name)?;
            validate_copy_field(declaration, source, target, filter)?;
            let mut builder = builder;
            let where_clause = builder.filter(declaration, filter)?;
            let sql = format!(
                "UPDATE {} SET {} = {} WHERE {where_clause}",
                builder.name(name)?,
                builder.name(target)?,
                builder.name(source)?
            );
            Ok(vec![builder.finish(sql)])
        }
    }
}