toasty-core 0.8.0

use super::BuildSchema;
use crate::{
    Error, Result, driver,
    schema::{
        app::{self, Model, ModelId, ModelRoot},
        db::{self, ColumnId, IndexId, Table, TableId},
        mapping::{self, Mapping, TableToModel},
    },
    stmt,
};

struct BuildTableFromModels<'a> {
    /// Application schema (for looking up model definitions)
    app: &'a app::Schema,

    /// Database-specific capabilities
    db: &'a driver::Capability,

    /// The table being built from the set of models
    table: &'a mut Table,

    /// Schema mapping
    mapping: &'a mut Mapping,

    /// When true, column names should be prefixed with their associated model
    /// names
    prefix_table_names: bool,

    /// Optional prefix for database object names (tables, enum types).
    name_prefix: Option<String>,
}

/// Computes a model's mapping, creating table columns and mapping expressions
/// in a single recursive pass over the model's fields.
///
/// Holds state that persists across the entire mapping process: the shared
/// mutable accumulators (columns, lowering expressions, bit counter) plus
/// references to the table and schema. The recursive field-mapping logic lives
/// on [`MapField`], which borrows `BuildMapping` and carries per-level context.
struct BuildMapping<'a> {
    app: &'a app::Schema,
    db: &'a driver::Capability,
    table: &'a mut Table,
    mapping: &'a mut mapping::Model,
    /// Model-name prefix used when multiple models share one table, separated
    /// from the rest of the column name with `__`. None for single-model tables.
    schema_prefix: Option<String>,
    /// Optional prefix for database object names (enum types).
    name_prefix: Option<String>,
    next_bit: usize,
    lowering_columns: Vec<ColumnId>,
    model_to_table: Vec<stmt::Expr>,
    table_to_model: Vec<stmt::Expr>,
    /// Column-sharing registry, active only while mapping the variant fields of
    /// an embedded enum. Maps a flattened column name to the column an earlier
    /// variant already created for it, so a later variant field with the same
    /// column name shares that column instead of producing a duplicate. `None`
    /// outside an enum's variant mapping. Saved and restored around each enum so
    /// sharing is scoped to a single enum's variants (and never leaks across
    /// nested enums).
    shared_columns: Option<std::collections::HashMap<String, SharedColumn>>,
}

/// A column created by an earlier enum variant that later variants may reuse.
#[derive(Clone)]
struct SharedColumn {
    column: ColumnId,
    /// Index into `model_to_table` of the column's encode expression. Later
    /// variants merge their discriminant-guarded arm into this `Match`.
    lowering: usize,
    /// App type of the field that first created the column. Reused fields must
    /// match it exactly so the column needs no per-variant casting.
    ty: stmt::Type,
    /// Discriminant of the variant that created the column. Reuse is only sound
    /// across *different* variants; a field in the same variant reaching the same
    /// column is an invalid mapping (both would be active for one discriminant).
    variant_discriminant: Option<stmt::Value>,
}

/// Per-level state for the recursive `map_field*` methods.
///
/// Analogous to `LowerStatement` in `lower.rs`: `MapField` holds context that
/// may change between recursive calls, while [`BuildMapping`] holds the shared
/// mutable accumulators (columns, lowering expressions, bit counter) that
/// persist across the entire mapping process.
struct MapField<'a, 'b> {
    /// State shared across the entire mapping process.
    build: &'a mut BuildMapping<'b>,

    /// Accumulated embed-prefix components (without schema_prefix), pushed on
    /// entry to each nested field and popped on exit.
    ///
    /// Final column names join these with `_`, append the field name, then
    /// prepend the schema prefix (if any) with `__`. Keeping components
    /// separate ensures schema_prefix is applied exactly once.
    prefix: Vec<String>,

    /// When true, columns are created nullable regardless of the field's own
    /// nullability. Set while processing fields whose columns are only populated
    /// conditionally: enum variant fields (only the active variant's columns are
    /// populated) and nullable embedded-struct fields (all columns are NULL when
    /// the embed is `None`).
    force_nullable: bool,

    /// Base expression for the current nesting level.
    ///
    /// `None` at the top level. `Expr::Project(source_ref, proj)` at any nested
    /// level, where `source_ref` is the top-level source field reference and
    /// `proj` is the projection from that source down to the current container
    /// (not including the final `field_index` step). `field_expr` and
    /// `sub_projection` extend it by `field_index` to reach a specific field.
    field_base: Option<stmt::Expr>,

    /// A template expression with `Expr::arg(0)` as a placeholder for the raw
    /// field expression. `field_expr` substitutes the raw field expression into
    /// this template before returning. The identity value is `Expr::arg(0)`
    /// itself, which substitutes to the raw expression unchanged.
    ///
    /// Used by variant-specific `MapField` instances to automatically wrap
    /// field expressions in the discriminant match guard.
    field_expr_base: stmt::Expr,

    /// True when an outer single-field embed flagged its column as
    /// auto-increment. Single-field newtype embeds flatten to one column, so
    /// the outer field's `#[auto]` (Increment) must apply to that flattened
    /// column. ORed into the column's flag at creation time so an outer or
    /// inner declaration both take effect.
    inherited_auto_increment: bool,

    /// Discriminant of the enum variant currently being mapped, set by
    /// [`Self::for_variant`] and inherited by nested embeds within the variant.
    /// `None` outside an enum variant. Used to distinguish a sound cross-variant
    /// column share from an invalid same-variant collision.
    variant_discriminant: Option<stmt::Value>,
}

impl BuildSchema<'_> {
    pub(super) fn build_table_stub_for_model(&mut self, model: &ModelRoot) -> TableId {
        let table_name = self.prefix_table_name(&model.table_name);

        if !self.table_lookup.contains_key(&table_name) {
            let id = self.register_table(&table_name);
            self.tables.push(Table::new(id, table_name.clone()));
        }

        *self.table_lookup.get(&table_name).unwrap()
    }

    pub(super) fn build_tables_from_models(
        &mut self,
        app: &app::Schema,
        db: &driver::Capability,
    ) -> Result<()> {
        for table in &mut self.tables {
            let models = app
                .models()
                .filter(|model| model.is_root())
                .filter(|model| self.mapping.model(model.id()).table == table.id)
                .collect::<Vec<_>>();

            assert!(
                models.len() == 1,
                "TODO: handle mapping many models to one table"
            );

            BuildTableFromModels {
                app,
                db,
                table,
                mapping: &mut self.mapping,
                prefix_table_names: models.len() > 1,
                name_prefix: self.builder.table_name_prefix.clone(),
            }
            .build(models[0])?;
        }

        Ok(())
    }

    pub(super) fn register_table(&mut self, name: impl AsRef<str>) -> TableId {
        assert!(!self.table_lookup.contains_key(name.as_ref()));
        let id = TableId(self.table_lookup.len());
        self.table_lookup.insert(name.as_ref().to_string(), id);
        id
    }

    fn prefix_table_name(&self, name: &str) -> String {
        if let Some(prefix) = &self.builder.table_name_prefix {
            format!("{prefix}{name}")
        } else {
            name.to_string()
        }
    }
}

impl BuildTableFromModels<'_> {
    fn build(&mut self, model: &Model) -> Result<()> {
        self.map_model_fields(model)?;
        self.update_index_names();
        Ok(())
    }

    fn map_model_fields(&mut self, model: &Model) -> Result<()> {
        let root = model.as_root_unwrap();
        let schema_prefix = if self.prefix_table_names {
            Some(model.name().snake_case())
        } else {
            None
        };

        BuildMapping {
            app: self.app,
            db: self.db,
            table: self.table,
            mapping: self.mapping.model_mut(model),
            schema_prefix,
            name_prefix: self.name_prefix.clone(),
            next_bit: 0,
            lowering_columns: vec![],
            model_to_table: vec![],
            table_to_model: vec![],
            shared_columns: None,
        }
        .build_mapping(root)?;

        let model_fields = &self.mapping.model(model.id()).fields;
        let mut indices = Vec::new();
        self.collect_indices(&root.fields, model_fields, &root.indices, &mut indices)?;

        for index in indices {
            if index.primary_key {
                self.table.primary_key.columns = index.columns.iter().map(|c| c.column).collect();
            }

            self.table.indices.push(index);
        }

        Ok(())
    }

    /// Collects DB-level indices from app-level index definitions, then recurses
    /// into embedded struct fields to collect their indices as well.
    fn collect_indices(
        &self,
        fields: &[app::Field],
        field_mappings: &[mapping::Field],
        indices: &[app::Index],
        out: &mut Vec<db::Index>,
    ) -> Result<()> {
        for app_index in indices {
            let mut index = db::Index {
                id: IndexId {
                    table: self.table.id,
                    index: out.len(),
                },
                name: app_index.name.clone().unwrap_or_default(),
                on: self.table.id,
                columns: vec![],
                unique: app_index.unique,
                primary_key: app_index.primary_key,
            };

            for index_field in &app_index.fields {
                let mapping = &field_mappings[index_field.field.index];

                // Resolve the mapped column for this indexed field. Primitive
                // fields map directly. Newtype embedded structs (a single
                // unnamed field) are transparent wrappers around a primitive, so
                // we unwrap one level.
                //
                // Multi-field or named-field embedded structs are not yet
                // supported in indices because the column ordering within the
                // index matters and there is no syntax to specify it. That will
                // likely require an explicit field-order annotation on the
                // index.
                let column = self.resolve_indexed_column(mapping);

                index.columns.push(db::IndexColumn {
                    column,
                    op: index_field.op,
                    scope: index_field.scope,
                });
            }

            out.push(index);
        }

        for (field_index, field) in fields.iter().enumerate() {
            let app::FieldTy::Embedded(embedded) = &field.ty else {
                continue;
            };

            let target = lookup_embedded_model(self.app, embedded.target, field)?;

            match target {
                app::Model::EmbeddedStruct(embedded_struct) => {
                    let field_mapping = field_mappings[field_index]
                        .as_struct()
                        .expect("embedded struct field should have struct mapping");

                    self.collect_indices(
                        &embedded_struct.fields,
                        &field_mapping.fields,
                        &embedded_struct.indices,
                        out,
                    )?;
                }
                app::Model::EmbeddedEnum(embedded_enum) => {
                    if embedded_enum.indices.is_empty() {
                        continue;
                    }

                    let field_mapping = field_mappings[field_index]
                        .as_enum()
                        .expect("embedded enum field should have enum mapping");

                    // Build a flat mapping from global field index to the
                    // field's mapping within its variant. Enum fields use
                    // global indices; each field belongs to a specific variant
                    // and has a local offset within that variant.
                    let mut flat_mappings: Vec<mapping::Field> =
                        vec![
                            mapping::Field::Relation(mapping::FieldRelation {
                                field_mask: crate::stmt::PathFieldSet::new(),
                            });
                            embedded_enum.fields.len()
                        ];

                    for (variant_idx, variant_mapping) in field_mapping.variants.iter().enumerate()
                    {
                        let mut local_idx = 0;
                        for (global_idx, f) in embedded_enum.fields.iter().enumerate() {
                            let app::VariantId { index: vi, .. } =
                                f.variant.expect("enum field must have variant");
                            if vi != variant_idx {
                                continue;
                            }
                            flat_mappings[global_idx] = variant_mapping.fields[local_idx].clone();
                            local_idx += 1;
                        }
                    }

                    self.collect_indices(
                        &embedded_enum.fields,
                        &flat_mappings,
                        &embedded_enum.indices,
                        out,
                    )?;
                }
                _ => continue,
            }
        }

        Ok(())
    }

    /// Walks newtype embed layers down to the underlying primitive column.
    ///
    /// Indexed fields may be either a primitive directly or a chain of
    /// single-unnamed-field embeds wrapping a primitive. Multi-field or
    /// named-field embeds are not supported as index targets — the index
    /// would have to choose a column ordering and there is no syntax for
    /// that yet.
    fn resolve_indexed_column(&self, mut mapping: &mapping::Field) -> ColumnId {
        loop {
            match mapping {
                mapping::Field::Primitive(p) => return p.column,
                mapping::Field::Enum(p) => {
                    // Only unit (data-less) enums can be indexed: the
                    // discriminant column alone represents the value.
                    // Data-carrying enums span multiple columns and have no
                    // single index column. The derive macro rejects this at
                    // compile time via `IndexableField`; this assert is the
                    // backstop for schemas built without the macro.
                    assert!(
                        p.variants.iter().all(|v| v.fields.is_empty()),
                        "only unit (data-less) embedded enums can be indexed; \
                         data-carrying enum variants span multiple columns"
                    );
                    return p.discriminant.column;
                }
                mapping::Field::Struct(s) => {
                    let embedded_struct = self.app.model(s.id).as_embedded_struct_unwrap();
                    assert!(
                        embedded_struct.fields.len() == 1
                            && embedded_struct.fields[0].name.app.is_none(),
                        "only newtype embedded structs (single unnamed \
                         field) can be indexed; multi-field or named-field \
                         embedded structs require explicit index field \
                         ordering"
                    );
                    mapping = &s.fields[0];
                }
                _ => panic!(
                    "only primitive and newtype embedded structs can be \
                     indexed"
                ),
            }
        }
    }

    fn update_index_names(&mut self) {
        for index in &mut self.table.indices {
            // Preserve user-provided names from `#[index(name = "...", ...)]`.
            if !index.name.is_empty() {
                continue;
            }

            let mut name = format!("index_{}_by", self.table.name);

            for (i, index_column) in index.columns.iter().enumerate() {
                let column = &self.table.columns[index_column.column.index];

                if i > 0 {
                    name.push_str("_and");
                }

                name.push('_');
                name.push_str(&column.name);
            }

            index.name = if let Some(limit) = self.db.max_identifier_length {
                truncate_identifier(name, limit)
            } else {
                name
            };
        }
    }
}

impl BuildMapping<'_> {
    fn build_mapping(mut self, model: &ModelRoot) -> Result<()> {
        let mut fields = MapField::new(&mut self).map_fields(&model.fields)?;

        assert!(!self.model_to_table.is_empty());
        assert_eq!(self.model_to_table.len(), self.lowering_columns.len());

        self.build_table_to_model(model, &fields)?;

        // Compute the default `RETURNING` expression for the model and for
        // each nested embedded type. Mutates `fields` to populate the
        // per-embed `default_returning` along the way.
        let default_returning = self.build_default_returning_root(model, &mut fields)?;

        self.mapping.fields = fields;
        self.mapping.columns = self.lowering_columns;
        self.mapping.model_to_table = stmt::ExprRecord::from_vec(self.model_to_table);
        self.mapping.table_to_model =
            TableToModel::new(stmt::ExprRecord::from_vec(self.table_to_model));
        self.mapping.default_returning = default_returning;

        Ok(())
    }

    /// Builds the model's default `RETURNING` expression — the same shape as
    /// `table_to_model` but with every deferred field, at this level or
    /// inside a nested embedded type, pre-masked to `Null`. Also writes each
    /// embed's own default expression into the corresponding mapping node so
    /// lowering can splice it in when an `.include()` activates a deferred
    /// embed.
    fn build_default_returning_root(
        &self,
        model: &ModelRoot,
        fields: &mut [mapping::Field],
    ) -> Result<stmt::Expr> {
        let exprs: Vec<stmt::Expr> = model
            .fields
            .iter()
            .zip(fields.iter_mut())
            .map(|(field, mapping)| self.build_default_returning_field(field, mapping))
            .collect::<Result<_>>()?;
        Ok(stmt::Expr::record(exprs))
    }

    /// Builds the default returning expression for a single field and, if
    /// the field is an embedded type, populates the embed's own
    /// `default_returning` cache.
    fn build_default_returning_field(
        &self,
        field: &app::Field,
        mapping: &mut mapping::Field,
    ) -> Result<stmt::Expr> {
        // Deferred fields are `Null` in the default expression. Still
        // recurse through deferred embeds so the nested `default_returning`
        // is populated — `process_includes` reads it during a `.include()`
        // splice.
        if field.deferred {
            if matches!(&field.ty, app::FieldTy::Embedded(_)) {
                self.populate_embed_default_returning(field, mapping)?;
            }
            return Ok(stmt::Expr::null());
        }

        match &field.ty {
            app::FieldTy::Primitive(primitive) => {
                let column_id = mapping.as_primitive().unwrap().column;
                Ok(self.map_table_column_to_model(column_id, primitive))
            }
            app::FieldTy::Embedded(_) => self.populate_embed_default_returning(field, mapping),
            app::FieldTy::BelongsTo(_) | app::FieldTy::Has(_) | app::FieldTy::Via(_) => {
                Ok(stmt::Value::Null.into())
            }
        }
    }

    /// Resolves the embedded target, recurses to compute its default
    /// expression, stores it on the mapping node, and returns a clone for
    /// the caller to splice in for the parent field.
    fn populate_embed_default_returning(
        &self,
        field: &app::Field,
        mapping: &mut mapping::Field,
    ) -> Result<stmt::Expr> {
        let app::FieldTy::Embedded(embedded) = &field.ty else {
            unreachable!("populate_embed_default_returning called on non-embed");
        };
        let target = lookup_embedded_model(self.app, embedded.target, field)?;

        match (target, mapping) {
            (app::Model::EmbeddedStruct(embed_model), mapping::Field::Struct(s)) => {
                let record = {
                    let exprs: Vec<stmt::Expr> = embed_model
                        .fields
                        .iter()
                        .zip(s.fields.iter_mut())
                        .map(|(f, m)| self.build_default_returning_field(f, m))
                        .collect::<Result<_>>()?;
                    stmt::Expr::record(exprs)
                };
                // A nullable embedded struct guards its record on the head
                // column, matching `build_table_to_model_field_struct`.
                let expr = match s.presence {
                    None => record,
                    Some(presence) => self.wrap_presence_match(presence.index, record),
                };
                s.default_returning = expr.clone();
                Ok(expr)
            }
            (app::Model::EmbeddedEnum(embed_model), mapping::Field::Enum(e)) => {
                let expr = self.build_default_returning_enum(embed_model, e, field.nullable)?;
                e.default_returning = expr.clone();
                Ok(expr)
            }
            _ => unreachable!("invalid schema: embedded field maps to root model"),
        }
    }

    /// Builds the enum's default `Match` expression, recursing into each
    /// variant's fields so a deferred sub-field nested inside a variant's
    /// embed-struct is pre-masked. The shape mirrors
    /// [`Self::build_table_to_model_field_enum`] but routes each variant
    /// field through [`Self::build_default_returning_field`] instead of
    /// the raw column emitter.
    ///
    /// Deferred variant fields route through the same default-returning
    /// masking as fields inside embedded structs.
    fn build_default_returning_enum(
        &self,
        model: &app::EmbeddedEnum,
        mapping: &mut mapping::FieldEnum,
        nullable: bool,
    ) -> Result<stmt::Expr> {
        let disc_col_ref = stmt::Expr::column(stmt::ExprColumn {
            nesting: 0,
            table: 0,
            column: mapping.discriminant.column.index,
        });

        if !model.has_data_variants() {
            return Ok(disc_col_ref);
        }

        let mut arms = Vec::new();
        for (variant_index, (variant, variant_mapping)) in model
            .variants
            .iter()
            .zip(mapping.variants.iter_mut())
            .enumerate()
        {
            let variant_fields: Vec<&app::Field> = model.variant_fields(variant_index).collect();
            let arm_expr = if variant_fields.is_empty() {
                disc_col_ref.clone()
            } else {
                let mut record_elems = vec![disc_col_ref.clone()];
                for (local_idx, field) in variant_fields.iter().enumerate() {
                    let mapping_field = &mut variant_mapping.fields[local_idx];
                    record_elems.push(self.build_default_returning_field(field, mapping_field)?);
                }
                stmt::Expr::record(record_elems)
            };
            arms.push(stmt::MatchArm {
                pattern: variant.discriminant.clone(),
                expr: arm_expr,
            });
        }

        let else_expr = self.enum_decode_else(model, nullable, &disc_col_ref);
        Ok(stmt::Expr::match_expr(disc_col_ref, arms, else_expr))
    }

    fn next_bit(&mut self) -> usize {
        let bit = self.next_bit;
        self.next_bit += 1;
        bit
    }

    fn build_table_to_model(
        &mut self,
        model: &ModelRoot,
        mapping: &[mapping::Field],
    ) -> Result<()> {
        for (index, field) in model.fields.iter().enumerate() {
            let expr = self.build_table_to_model_field(field, &mapping[index])?;
            self.table_to_model.push(expr);
        }
        Ok(())
    }

    /// Builds the `table_to_model` expression for an embedded enum field.
    ///
    /// For unit-only enums the discriminant column reference suffices.
    /// For mixed/data-carrying enums a `Match` expression dispatches on the
    /// discriminant: unit arms return the discriminant directly, data arms
    /// return `Record([disc, field1, ...])` matching the shape expected by
    /// `Primitive::load`.
    ///
    /// When `nullable` (an `Option<EmbeddedEnum>`), the discriminant column is
    /// `NULL` for `None`: a unit-only enum's column reference already reads back
    /// as `None`, and a data enum's `Match` else branch yields `null` so a
    /// `NULL` (unmatched) discriminant decodes to `None`.
    fn build_table_to_model_field_enum(
        &self,
        model: &app::EmbeddedEnum,
        mapping: &mapping::FieldEnum,
        nullable: bool,
    ) -> Result<stmt::Expr> {
        let disc_col_ref = stmt::Expr::column(stmt::ExprColumn {
            nesting: 0,
            table: 0,
            column: mapping.discriminant.column.index,
        });

        if !model.has_data_variants() {
            return Ok(disc_col_ref);
        }

        let mut arms = Vec::new();

        for (variant_index, (variant, mapping)) in
            model.variants.iter().zip(&mapping.variants).enumerate()
        {
            let variant_fields: Vec<_> = model.variant_fields(variant_index).collect();
            let arm_expr = if variant_fields.is_empty() {
                disc_col_ref.clone()
            } else {
                let mut record_elems = vec![disc_col_ref.clone()];

                for (local_idx, field) in variant_fields.iter().enumerate() {
                    let expr =
                        self.build_table_to_model_field(field, &mapping.fields[local_idx])?;
                    record_elems.push(expr);
                }
                stmt::Expr::record(record_elems)
            };
            arms.push(stmt::MatchArm {
                pattern: variant.discriminant.clone(),
                expr: arm_expr,
            });
        }

        let else_expr = self.enum_decode_else(model, nullable, &disc_col_ref);
        Ok(stmt::Expr::match_expr(disc_col_ref, arms, else_expr))
    }

    /// The `else` branch shared by the enum decode and default-returning
    /// `Match` expressions.
    ///
    /// When `nullable`, a `NULL` (unmatched) discriminant is the `None` case, so
    /// the else is `null`. Otherwise it mirrors the data-arm `Record` shape but
    /// fills field positions with `Expr::Error`, making projections distribute
    /// uniformly: projecting `[0]` extracts the discriminant (pruning the
    /// errors) while a deeper projection yields `Expr::Error` (unreachable at
    /// runtime).
    fn enum_decode_else(
        &self,
        model: &app::EmbeddedEnum,
        nullable: bool,
        disc_col_ref: &stmt::Expr,
    ) -> stmt::Expr {
        if nullable {
            return stmt::Expr::null();
        }
        let max_fields = (0..model.variants.len())
            .map(|i| model.variant_fields(i).count())
            .max()
            .unwrap_or(0);
        if max_fields == 0 {
            stmt::Expr::error("unexpected enum discriminant")
        } else {
            let mut elems = vec![disc_col_ref.clone()];
            for _ in 0..max_fields {
                elems.push(stmt::Expr::error("unexpected enum discriminant"));
            }
            stmt::Expr::record(elems)
        }
    }

    /// Encodes `expr` for `column_id`, appends the result to `model_to_table`,
    /// records the column in `lowering_columns`, and returns the lowering index.
    fn push_lowering(
        &mut self,
        column_id: ColumnId,
        ty: &stmt::Type,
        expr: impl Into<stmt::Expr>,
    ) -> usize {
        let lowering_expr = self.encode_column(column_id, ty, expr);
        let lowering_index = self.model_to_table.len();
        self.lowering_columns.push(column_id);
        self.model_to_table.push(lowering_expr);
        lowering_index
    }

    /// Merges a later variant field's encode into the shared column's existing
    /// encode at `lowering`, so one column encodes every contributing variant.
    ///
    /// Both encodes are produced by [`MapField::for_variant`], so each is a
    /// discriminant-guarded `Match` (optionally wrapped in a `Cast`) whose arm
    /// fires only for that variant. Appending `new_encode`'s arm to the existing
    /// `Match` yields `match disc { d1 => v1, d2 => v2, .. } else null` — exactly
    /// the per-variant dispatch a shared column needs.
    fn merge_shared_encode(&mut self, lowering: usize, new_encode: stmt::Expr) -> Result<()> {
        let new_arms = into_match_arms(new_encode).ok_or_else(|| {
            Error::invalid_schema(
                "shared enum column encode is not a discriminant match expression",
            )
        })?;
        let existing = match_arms_mut(&mut self.model_to_table[lowering]).ok_or_else(|| {
            Error::invalid_schema(
                "shared enum column encode is not a discriminant match expression",
            )
        })?;
        existing.extend(new_arms);
        Ok(())
    }

    fn encode_column(
        &self,
        column_id: ColumnId,
        ty: &stmt::Type,
        expr: impl Into<stmt::Expr>,
    ) -> stmt::Expr {
        let expr = expr.into();
        let column = self.table.column(column_id);

        assert_ne!(stmt::Type::Null, *ty);

        match &column.ty {
            column_ty if column_ty == ty => expr,
            // If the types do not match, attempt casting as a fallback.
            _ => stmt::Expr::cast(expr, &column.ty),
        }
    }

    /// Maps table columns to model field expressions during query lowering.
    ///
    /// Called during query planning to replace model field references with the
    /// appropriate table column expressions. Handles type conversions between
    /// table storage and model types.
    fn map_table_column_to_model(
        &self,
        column_id: ColumnId,
        primitive: &app::FieldPrimitive,
    ) -> stmt::Expr {
        let column = self.table.column(column_id);

        // NOTE: nesting and table are stubs here (though often the actual values).
        // The engine must substitute these with the actual TableRef index in the query's TableSource.
        let expr_column = stmt::Expr::column(stmt::ExprColumn {
            nesting: 0,
            table: 0,
            column: column_id.index,
        });

        match &column.ty {
            c_ty if *c_ty == primitive.ty => expr_column,
            // If the types do not match, attempt casting as a fallback.
            _ => stmt::Expr::cast(expr_column, &primitive.ty),
        }
    }

    fn build_table_to_model_field_struct(
        &self,
        model: &app::EmbeddedStruct,
        mapping: &mapping::FieldStruct,
    ) -> Result<stmt::Expr> {
        let exprs: Vec<stmt::Expr> = model
            .fields
            .iter()
            .enumerate()
            .map(|(index, field)| self.build_table_to_model_field(field, &mapping.fields[index]))
            .collect::<Result<_>>()?;
        let record = stmt::Expr::record(exprs);

        // A nullable embedded struct decodes through a `Match` on its head
        // column (see `wrap_presence_match`).
        match mapping.presence {
            None => Ok(record),
            Some(presence) => Ok(self.wrap_presence_match(presence.index, record)),
        }
    }

    /// Wraps `record` in the presence `Match` that decodes a nullable embedded
    /// struct: when the head column is non-`NULL` the record is reconstructed,
    /// otherwise the value is `null`, which `impl Load for Option<T>` reads
    /// back as `None`. The engine evaluates this `Match` against the returned
    /// row before `Load` runs.
    ///
    /// The head column is either a dedicated `bool` presence column (only ever
    /// `true`/`NULL`) or, for a single-column embed, the embed's own flattened
    /// leaf column — in both cases its null-ness is the option's none-ness, so
    /// guarding on `is_null(head) == false` (present) is the uniform check. We
    /// match `is_null(head)` against `false` rather than wrapping in `not(..)`
    /// because the table→model evaluator handles `IsNull`/`Match`/`Project`
    /// but not `Not`.
    fn wrap_presence_match(&self, presence_column: usize, record: stmt::Expr) -> stmt::Expr {
        stmt::Expr::match_expr(
            stmt::Expr::is_null(stmt::Expr::column(stmt::ExprColumn {
                nesting: 0,
                table: 0,
                column: presence_column,
            })),
            vec![stmt::MatchArm {
                pattern: stmt::Value::Bool(false),
                expr: record,
            }],
            stmt::Expr::null(),
        )
    }

    fn build_table_to_model_field(
        &self,
        field: &app::Field,
        mapping: &mapping::Field,
    ) -> Result<stmt::Expr> {
        match &field.ty {
            app::FieldTy::Primitive(primitive) => {
                let column_id = mapping.as_primitive().unwrap().column;
                Ok(self.map_table_column_to_model(column_id, primitive))
            }
            app::FieldTy::Embedded(embedded) => {
                let target = lookup_embedded_model(self.app, embedded.target, field)?;

                match target {
                    app::Model::EmbeddedEnum(embedded) => {
                        let mapping = mapping
                            .as_enum()
                            .expect("embedded enum field should have enum mapping");
                        self.build_table_to_model_field_enum(embedded, mapping, field.nullable)
                    }
                    app::Model::EmbeddedStruct(embedded) => {
                        let mapping = mapping
                            .as_struct()
                            .expect("embedded struct field should have struct mapping");
                        self.build_table_to_model_field_struct(embedded, mapping)
                    }
                    _ => unreachable!("invalid schema"),
                }
            }
            app::FieldTy::BelongsTo(_) | app::FieldTy::Has(_) | app::FieldTy::Via(_) => {
                Ok(stmt::Value::Null.into())
            }
        }
    }
}

impl<'a, 'b> MapField<'a, 'b> {
    fn new(build: &'a mut BuildMapping<'b>) -> Self {
        MapField {
            build,
            prefix: vec![],
            force_nullable: false,
            field_base: None,
            field_expr_base: stmt::Expr::arg(0),
            inherited_auto_increment: false,
            variant_discriminant: None,
        }
    }

    fn map_fields(&mut self, fields: &[app::Field]) -> Result<Vec<mapping::Field>> {
        fields
            .iter()
            .enumerate()
            .map(|(index, field)| self.map_field(index, field))
            .collect()
    }

    fn map_field(&mut self, index: usize, field: &app::Field) -> Result<mapping::Field> {
        match &field.ty {
            app::FieldTy::Primitive(primitive) => self.map_field_primitive(index, field, primitive),
            app::FieldTy::Embedded(embedded) => {
                let target = lookup_embedded_model(self.build.app, embedded.target, field)?;

                match target {
                    app::Model::EmbeddedEnum(embedded_enum) => {
                        self.map_field_enum(index, field, embedded_enum)
                    }
                    app::Model::EmbeddedStruct(embedded_struct) => {
                        self.map_field_struct(index, field, embedded.target, embedded_struct)
                    }
                    _ => unreachable!(),
                }
            }
            app::FieldTy::BelongsTo(_) | app::FieldTy::Has(_) | app::FieldTy::Via(_) => {
                assert!(!self.force_nullable);
                let bit = self.build.next_bit();
                Ok(mapping::Field::Relation(mapping::FieldRelation {
                    field_mask: stmt::PathFieldSet::from_iter([bit]),
                }))
            }
        }
    }

    /// Creates the column and builds the mapping for a primitive field in one step.
    ///
    /// When column sharing is active (i.e. mapping enum variant fields) and an
    /// earlier variant already created a column with the same flattened name,
    /// this reuses that column — merging the field's discriminant-guarded encode
    /// into the shared column's `Match` — instead of creating a duplicate.
    fn map_field_primitive(
        &mut self,
        field_index: usize,
        field: &app::Field,
        primitive: &app::FieldPrimitive,
    ) -> Result<mapping::Field> {
        let expr = self.field_expr(field, field_index);
        let column_name = self.column_name(field);

        let shared = self
            .build
            .shared_columns
            .as_ref()
            .and_then(|columns| columns.get(&column_name).cloned());

        let (column_id, lowering_index) = if let Some(shared) = shared {
            // Reuse is only sound across *different* variants — the merged encode
            // dispatches on the discriminant, and at most one variant is active
            // per row. Two fields in the *same* variant reaching one column would
            // append a second arm under the same discriminant, so the encode
            // would silently keep only the first and both fields would decode from
            // the one column. Reject that invalid mapping. The `Embed` derive
            // catches it at compile time; this is the backstop for schemas built
            // without the macro.
            if shared.variant_discriminant.is_some()
                && shared.variant_discriminant == self.variant_discriminant
            {
                return Err(Error::invalid_schema(format!(
                    "two fields in the same enum variant map to the column \
                     `{column_name}`; a column can be shared only across different \
                     variants",
                )));
            }

            // A shared column needs no per-variant casting, so the contributing
            // fields must have identical types. The `Embed` derive rejects a
            // mismatch at compile time via `SameColumnType`; this is the
            // backstop for schemas built without the macro. Width or kind
            // mismatches (e.g. `i32` vs `i64`, or `String` vs `i64`) are
            // rejected here rather than silently coerced.
            if shared.ty != primitive.ty {
                return Err(Error::invalid_schema(format!(
                    "enum variant fields mapped to the shared column `{column_name}` \
                     have incompatible types ({:?} and {:?}); fields sharing a column \
                     must have the same type",
                    shared.ty, primitive.ty,
                )));
            }

            self.build.merge_shared_encode(shared.lowering, expr)?;
            (shared.column, shared.lowering)
        } else {
            let column_id = self.create_column(field, primitive);
            let lowering_index = self.build.push_lowering(column_id, &primitive.ty, expr);

            if let Some(columns) = self.build.shared_columns.as_mut() {
                columns.insert(
                    column_name,
                    SharedColumn {
                        column: column_id,
                        lowering: lowering_index,
                        ty: primitive.ty.clone(),
                        variant_discriminant: self.variant_discriminant.clone(),
                    },
                );
            }

            (column_id, lowering_index)
        };

        let bit = self.build.next_bit();
        let sub_projection = self.sub_projection(field_index);
        let column_expr = self.build.map_table_column_to_model(column_id, primitive);

        Ok(mapping::Field::Primitive(mapping::FieldPrimitive {
            column: column_id,
            lowering: lowering_index,
            field_mask: stmt::PathFieldSet::from_iter([bit]),
            sub_projection,
            column_expr,
        }))
    }

    /// Creates the discriminant and variant-field columns, then builds the
    /// enum mapping — all in a single pass.
    fn map_field_enum(
        &mut self,
        field_index: usize,
        field: &app::Field,
        embedded_enum: &app::EmbeddedEnum,
    ) -> Result<mapping::Field> {
        // Create the discriminant column. It inherits nullability from the enum field.
        let column_id = self.create_column(field, &embedded_enum.discriminant);
        let field_expr = self.field_expr(field, field_index);

        // For data-carrying enums the model value is Record([I64(disc), ...]),
        // so project [0] to extract the discriminant; for unit-only enums the
        // value IS the I64 discriminant directly.
        let disc_expr = if embedded_enum.has_data_variants() {
            stmt::Expr::project(field_expr.clone(), stmt::Projection::single(0))
        } else {
            field_expr.clone()
        };

        let lowering_index =
            self.build
                .push_lowering(column_id, &embedded_enum.discriminant.ty, disc_expr);

        let bit = self.build.next_bit();
        let sub_projection = self.sub_projection(field_index);

        let disc_proj = stmt::Expr::project(field_expr.clone(), stmt::Projection::single(0));

        // Activate a fresh column-sharing registry for this enum's variant
        // fields. A nested enum installs (and restores) its own, so sharing
        // never crosses enum boundaries. Restored after the variants are mapped.
        let saved_shared = self
            .build
            .shared_columns
            .replace(std::collections::HashMap::new());

        let variants: Result<Vec<mapping::EnumVariant>> = embedded_enum
            .variants
            .iter()
            .enumerate()
            .map(|(variant_index, variant)| {
                let mut mapper =
                    self.for_variant(field, field_index, disc_proj.clone(), &variant.discriminant);

                let fields: Vec<mapping::Field> = embedded_enum
                    .variant_fields(variant_index)
                    .enumerate()
                    .map(|(index, field)| {
                        // Variant fields are stored at positions 1.. in the Record
                        // (position 0 is the discriminant), so adjust the index.
                        mapper.map_field(index + 1, field)
                    })
                    .collect::<Result<_>>()?;

                Ok(mapping::EnumVariant {
                    discriminant: variant.discriminant.clone(),
                    fields,
                })
            })
            .collect();

        self.build.shared_columns = saved_shared;
        let variants = variants?;

        let field_mask = stmt::PathFieldSet::from_iter([bit]);

        let disc_column_expr = self
            .build
            .map_table_column_to_model(column_id, &embedded_enum.discriminant);

        Ok(mapping::Field::Enum(mapping::FieldEnum {
            discriminant: mapping::FieldPrimitive {
                column: column_id,
                lowering: lowering_index,
                field_mask: field_mask.clone(),
                sub_projection: stmt::Projection::identity(),
                column_expr: disc_column_expr,
            },
            variants,
            field_mask,
            sub_projection,
            default_returning: stmt::Expr::null(),
        }))
    }

    fn map_field_struct(
        &mut self,
        field_index: usize,
        field: &app::Field,
        model_id: ModelId,
        embedded_struct: &app::EmbeddedStruct,
    ) -> Result<mapping::Field> {
        let sub_projection = self.sub_projection(field_index);

        // A nullable embedded struct (`Option<Embed>`) is decoded/encoded
        // through a *head* column whose null-ness is the option's none-ness
        // (`NULL` = `None`), keeping `None` represented as `NULL` like
        // `Option<scalar>` and an embedded enum's (nullable) discriminant. The
        // flattened leaf columns are forced nullable and presence-guarded, so a
        // `None` value writes all-`NULL` rather than panicking when projecting
        // `Null`. There are two ways to obtain the head column:
        //
        //  * Multi-column embed (`Option<Address>` over `Address { street, city
        //    }`): a *dedicated* nullable `bool` presence column (`NULL`/`true`)
        //    named after the field. `None` → `(NULL, NULL, NULL)`; `Some(..)` →
        //    `(true, "..", "..")`. The presence column distinguishes `None`
        //    from a `Some` whose own fields are all `NULL`.
        //
        //  * Newtype embed (`Option<Email(String)>` over `struct Email(String)`):
        //    the one flattened leaf is unnamed and collapses to the field name,
        //    so a dedicated presence column would *collide* with it. Reuse that
        //    leaf as the head — a non-`Option` inner is non-`NULL` exactly when
        //    the embed is present, so its null-ness already encodes none-ness,
        //    like `Option<scalar>`. (A *named* single field flattens to
        //    `{field}_{name}`, doesn't collide, and keeps the disambiguating
        //    presence column above.)
        //
        //    A newtype wrapping an *optional* (`Option<MaybeBody>` over
        //    `MaybeBody(Option<String>)`) is rejected: the single column can't
        //    tell `None` from `Some(MaybeBody(None))`, and a dedicated presence
        //    column would collide with the leaf, so there is no sound mapping.
        let reuse_leaf = if field.nullable {
            match self.newtype_leaf_nullable(embedded_struct) {
                // Newtype over a non-optional scalar → reuse the leaf as head.
                Some(false) => true,
                // Newtype over an optional → no unambiguous single-column mapping.
                Some(true) => {
                    let parent = self
                        .build
                        .app
                        .get_model(field.id.model)
                        .map(|m| m.name().upper_camel_case())
                        .unwrap_or_else(|| "?".to_string());
                    let embed = embedded_struct.name.upper_camel_case();
                    return Err(Error::invalid_schema(format!(
                        "field `{parent}::{}` is `Option<{embed}>`, but `{embed}` is \
                         a newtype wrapping an optional value; it can't be flattened \
                         to a single column because that column can't distinguish \
                         `None` from `Some({embed}(None))`. Use a named struct field \
                         instead of a tuple newtype, or remove a layer of `Option`.",
                        field.name,
                    )));
                }
                // Not a newtype → dedicated presence column below.
                None => false,
            }
        } else {
            false
        };

        // The dedicated presence column is created *before* the leaves so it
        // sorts first in the table. Only the `(column, lowering)` pair is
        // needed: `lowering` records the column's encode expression in
        // `columns`, and the column id becomes the struct's presence head.
        let dedicated_presence = if field.nullable && !reuse_leaf {
            let presence_primitive = app::FieldPrimitive {
                ty: stmt::Type::Bool,
                storage_ty: None,
                serialize: None,
            };
            // Nullable because `field.nullable` is set.
            let column = self.create_column(field, &presence_primitive);
            // Encode `true` for `Some`, `NULL` for `None` — the same presence
            // guard the leaf columns use, applied to the sentinel `true`.
            let presence_expr = self.presence_guard(field, field_index, stmt::Expr::from(true));
            let lowering = self
                .build
                .push_lowering(column, &stmt::Type::Bool, presence_expr);
            Some((column, lowering))
        } else {
            None
        };

        // For a single-field newtype the outer field's `#[auto]` flattens
        // down to the one inner column. Multi-field embeds have no clear
        // target column, so the inherited flag stops at the boundary —
        // `Foo(Bar(u64))` with `#[auto]` on the outer flows through both
        // newtypes, but `Foo { a, b: Bar(u64) }` would not propagate it
        // from any outer past the `Foo` layer.
        let single_field = embedded_struct.fields.len() == 1;
        let mut child = if field.nullable {
            self.for_nullable_struct(field, field_index)
        } else {
            self.for_struct(field, field_index)
        };
        if single_field {
            child.inherited_auto_increment |= field.is_auto_increment();
        } else {
            child.inherited_auto_increment = false;
        }
        let nested_fields = child.map_fields(&embedded_struct.fields)?;

        let mut columns: indexmap::IndexMap<ColumnId, usize> =
            nested_fields.iter().flat_map(|f| f.columns()).collect();

        // The head column: the dedicated presence column (added to `columns`
        // here so its encode lowering is tracked), or the reused single leaf
        // for a single-column embed (already in `columns` as a real leaf).
        let presence = if let Some((column, lowering)) = dedicated_presence {
            columns.insert(column, lowering);
            Some(column)
        } else if reuse_leaf {
            Some(
                single_leaf_primitive(&nested_fields[0])
                    .expect("single-column embed flattens to one primitive leaf")
                    .column,
            )
        } else {
            None
        };

        let field_mask = nested_fields
            .iter()
            .fold(stmt::PathFieldSet::new(), |acc, f| acc | f.field_mask());

        Ok(mapping::Field::Struct(mapping::FieldStruct {
            id: model_id,
            fields: nested_fields,
            columns,
            field_mask,
            sub_projection,
            default_returning: stmt::Expr::null(),
            presence,
        }))
    }

    /// If `embed` is a newtype chain — a single *unnamed* field (recursively,
    /// through nested single-field newtype wrappers) bottoming out in a
    /// primitive — returns `Some(leaf_nullable)`, the bottom field's own
    /// nullability. Returns `None` when `embed` is not a newtype.
    ///
    /// A newtype flattens to one column that takes the parent field's own name,
    /// so when the parent is nullable it reuses that single leaf as the head
    /// column (a dedicated presence column would collide). That reuse is sound
    /// only when the leaf is non-`NULL` whenever the embed is present —
    /// `Some(false)`. A `Some(true)` leaf (the newtype wraps an `Option`) has no
    /// unambiguous single-column mapping and is rejected by the caller. A
    /// *named* single field (`Foo { value }`) flattens to `{field}_value`,
    /// doesn't collide, so it returns `None` here and keeps a dedicated presence
    /// column (preserving the `None`-vs-`Some(all-NULL)` disambiguation).
    fn newtype_leaf_nullable(&self, embed: &app::EmbeddedStruct) -> Option<bool> {
        let [field] = &embed.fields[..] else {
            return None;
        };
        if field.name.storage_name().is_some() {
            return None;
        }
        match &field.ty {
            app::FieldTy::Primitive(_) => Some(field.nullable),
            app::FieldTy::Embedded(embedded) => {
                match lookup_embedded_model(self.build.app, embedded.target, field) {
                    Ok(app::Model::EmbeddedStruct(inner)) => self.newtype_leaf_nullable(inner),
                    _ => None,
                }
            }
            _ => None,
        }
    }

    /// Builds the final database column name for `field` at the current nesting level.
    ///
    /// Joins `self.prefix` components with `_`, appends the field name, then
    /// prepends `schema_prefix` (if any) with `__`. Because `schema_prefix` is
    /// applied here — never stored in `self.prefix` — it is always applied
    /// exactly once regardless of nesting depth.
    fn column_name(&self, field: &app::Field) -> String {
        let embed = match field.name.storage_name() {
            Some(field_name) => {
                if self.prefix.is_empty() {
                    field_name.to_owned()
                } else {
                    format!("{}_{field_name}", self.prefix.join("_"))
                }
            }
            None => {
                // Unnamed field (newtype wrapper) — use just the prefix
                // components so the column name is identical to the parent
                // field's name.
                assert!(
                    !self.prefix.is_empty(),
                    "unnamed field with empty prefix; a newtype field must be \
                     nested inside a parent field"
                );
                self.prefix.join("_")
            }
        };
        match self.build.schema_prefix.as_deref() {
            None => embed,
            Some(sp) => format!("{sp}__{embed}"),
        }
    }

    /// Creates a column for `field` using `primitive` for the storage type.
    ///
    /// Derives the column name from `self.column_name(field)`, nullability from
    /// `field.nullable || self.force_nullable`, and auto-increment from
    /// `field.is_auto_increment()`.
    fn create_column(&mut self, field: &app::Field, primitive: &app::FieldPrimitive) -> ColumnId {
        let nullable = field.nullable || self.force_nullable;
        let is_auto_increment = (field.is_auto_increment() || self.inherited_auto_increment)
            && self.build.db.auto_increment;

        let mut storage_ty = db::Type::from_app_column(
            &primitive.ty,
            primitive.storage_ty.as_ref(),
            self.build.db,
            is_auto_increment,
        )
        .expect("unsupported storage type");

        // Prefix native enum type names so they don't collide across test runs
        // or multi-tenant deployments.
        if let db::Type::Enum(ref mut type_enum) = storage_ty
            && let (Some(prefix), Some(name)) = (&self.build.name_prefix, &mut type_enum.name)
        {
            *name = format!("{prefix}{name}");
        }

        let id = ColumnId {
            table: self.build.table.id,
            index: self.build.table.columns.len(),
        };

        self.build.table.columns.push(db::Column {
            id,
            name: self.column_name(field),
            ty: storage_ty.bridge_type(&primitive.ty),
            storage_ty,
            nullable,
            primary_key: false,
            auto_increment: is_auto_increment,
            versionable: field.is_versionable(),
        });

        id
    }

    /// Extends `parent_base` by one projection step to produce the `field_base`
    /// for a child `MapField` entering `field` at `field_index` within the parent.
    ///
    /// If `parent_base` is the top-level sentinel (`Expr::arg(0)`), the child
    /// starts a fresh projection rooted at `field.id`. Otherwise the existing
    /// `ExprProject` is extended by `field_index`.
    fn extend_field_base(&self, field: &app::Field, field_index: usize) -> stmt::Expr {
        match &self.field_base {
            None => stmt::Expr::ref_self_field(field.id),
            Some(stmt::Expr::Project(ep)) => {
                let mut proj = ep.projection.clone();
                proj.push(field_index);
                stmt::Expr::project(*ep.base.clone(), proj)
            }
            Some(expr) => stmt::Expr::project(expr.clone(), [field_index]),
        }
    }

    /// Returns the sub-projection from the root source field to a field at
    /// `field_index` within the current nesting level.
    ///
    /// If `field_base` is an `ExprProject`, the sub-projection is its
    /// projection extended by `field_index`. At the top level (`field_base`
    /// is `Expr::arg(0)`) the field is its own root, so identity is returned.
    fn sub_projection(&self, field_index: usize) -> stmt::Projection {
        match &self.field_base {
            None => stmt::Projection::identity(),
            Some(stmt::Expr::Project(ep)) => {
                let mut proj = ep.projection.clone();
                proj.push(field_index);
                proj
            }
            Some(_) => [field_index].into(),
        }
    }

    /// Builds the lowering expression for a field at the current nesting level.
    ///
    /// At the top level (`field_base` is `Expr::arg(0)`) each field references
    /// itself directly. Inside an embedded struct/variant the expression extends
    /// `field_base` by `field_index`. The raw expression is then substituted
    /// into `field_expr_base` (which may wrap it in a match guard).
    fn field_expr(&self, field: &app::Field, field_index: usize) -> stmt::Expr {
        let raw = match self.field_base.clone() {
            None => stmt::Expr::ref_self_field(field.id),
            Some(stmt::Expr::Project(mut expr_project)) => {
                expr_project.projection.push(field_index);
                expr_project.into()
            }
            Some(expr) => stmt::Expr::project(expr, [field_index]),
        };

        let mut result = self.field_expr_base.clone();
        result.substitute(&[raw]);
        result
    }

    /// Creates a child `MapField` for recursing into an embedded field.
    ///
    /// The child inherits the current prefix extended by `name` and inherits
    /// `force_nullable`, `field_base`, and `field_expr_base` unchanged. Used
    /// when entering struct/variant fields so that sub-field columns are named
    /// `{..prefix..}_{name}_{sub_field}`.
    fn with_prefix(&mut self, name: Option<&str>) -> MapField<'_, 'b> {
        let mut prefix = self.prefix.clone();
        if let Some(name) = name {
            prefix.push(name.to_owned());
        }
        MapField {
            build: self.build,
            prefix,
            force_nullable: self.force_nullable,
            field_base: self.field_base.clone(),
            field_expr_base: self.field_expr_base.clone(),
            inherited_auto_increment: self.inherited_auto_increment,
            // A nested embed stays within its enclosing variant, so it inherits
            // the variant discriminant; `for_variant` overrides it per variant.
            variant_discriminant: self.variant_discriminant.clone(),
        }
    }

    /// Creates a child `MapField` for an embedded type whose columns are only
    /// conditionally populated: an enum variant (only the active variant's
    /// columns are written) or a nullable struct embed (every column is `NULL`
    /// when the embed is `None`).
    ///
    /// Sets `field_base` so that `field_expr` on the child projects from
    /// `field`, forces the flattened columns nullable, and installs `guard` as
    /// the `field_expr_base` so that every `field_expr` call is automatically
    /// wrapped in it. `guard` is a `Match` template with `Expr::arg(0)` as the
    /// placeholder for the raw field expression — a discriminant check (see
    /// [`Self::for_variant`]) or a presence guard (see
    /// [`Self::for_nullable_struct`]) — whose `else` branch encodes the
    /// inactive/`None` case to `null` instead of projecting into `Null`.
    fn for_guarded_embed(
        &mut self,
        field: &app::Field,
        field_index: usize,
        guard: stmt::Expr,
    ) -> MapField<'_, 'b> {
        let field_base = self.extend_field_base(field, field_index);
        let mut child = self.with_prefix(field.name.storage_name());
        child.force_nullable = true;
        child.field_base = Some(field_base);
        child.field_expr_base.substitute(&[guard]);
        child
    }

    /// Creates a variant-specific child `MapField`, guarding each `field_expr`
    /// in the discriminant check `match_expr(disc_proj, [arm(discriminant,
    /// Expr::arg(0))], null())`. See [`Self::for_guarded_embed`].
    fn for_variant(
        &mut self,
        field: &app::Field,
        field_index: usize,
        disc_proj: stmt::Expr,
        discriminant: &stmt::Value,
    ) -> MapField<'_, 'b> {
        let guard = stmt::Expr::match_expr(
            disc_proj,
            vec![stmt::MatchArm {
                pattern: discriminant.clone(),
                expr: stmt::Expr::arg(0),
            }],
            stmt::Expr::null(),
        );
        let mut child = self.for_guarded_embed(field, field_index, guard);
        child.variant_discriminant = Some(discriminant.clone());
        child
    }

    /// Creates a child `MapField` for recursing into an embedded struct field.
    ///
    /// Updates `field_base` to reflect the new nesting level: if entering the
    /// first embedded level, sets the source to this field with an identity
    /// projection; at deeper levels, extends the existing projection by
    /// `field_index`.
    fn for_struct(&mut self, field: &app::Field, field_index: usize) -> MapField<'_, 'b> {
        let field_base = self.extend_field_base(field, field_index);
        // Unnamed (newtype) inner fields keep the parent's prefix so the
        // flattened column name comes from the outermost named field.
        let mut child = self.with_prefix(field.name.storage_name());
        child.field_base = Some(field_base);
        child
    }

    /// Builds the presence guard that encodes a nullable struct embed
    /// (`Option<Embed>`):
    ///
    /// `Match(is_not_null(field), [Bool(true) => present], else => null())`.
    ///
    /// `present` is what's written when the embed is `Some`: the sentinel
    /// `Bool(true)` for the presence column itself, or `Expr::arg(0)` (the raw
    /// leaf expression) for the guard wrapped around each flattened leaf column.
    /// A `None` value (the field is `Null`, so `is_not_null` is `false`) takes
    /// the `else` branch and encodes `null`.
    fn presence_guard(
        &self,
        field: &app::Field,
        field_index: usize,
        present: stmt::Expr,
    ) -> stmt::Expr {
        stmt::Expr::match_expr(
            stmt::Expr::is_not_null(self.field_expr(field, field_index)),
            vec![stmt::MatchArm {
                pattern: stmt::Value::Bool(true),
                expr: present,
            }],
            stmt::Expr::null(),
        )
    }

    /// Creates a child `MapField` for recursing into a nullable embedded struct
    /// (`Option<Embed>`), guarding each `field_expr` in a [`Self::presence_guard`]
    /// over `Expr::arg(0)` so a `None` value encodes each leaf to `null` (the
    /// guard's `else` branch). See [`Self::for_guarded_embed`].
    fn for_nullable_struct(&mut self, field: &app::Field, field_index: usize) -> MapField<'_, 'b> {
        let guard = self.presence_guard(field, field_index, stmt::Expr::arg(0));
        self.for_guarded_embed(field, field_index, guard)
    }
}

/// Returns a mutable reference to the arms of a discriminant-guarded encode,
/// peeling a `Cast` wrapper if present. Used to merge a later variant field's
/// arm into a shared column's existing `Match`.
fn match_arms_mut(expr: &mut stmt::Expr) -> Option<&mut Vec<stmt::MatchArm>> {
    match expr {
        stmt::Expr::Match(m) => Some(&mut m.arms),
        stmt::Expr::Cast(c) => match_arms_mut(&mut c.expr),
        _ => None,
    }
}

/// Consumes a discriminant-guarded encode, returning its arms (peeling a `Cast`
/// wrapper if present). The subject and `else` branch are dropped: they are
/// identical across the variants sharing a column, so the existing `Match`
/// retains them.
fn into_match_arms(expr: stmt::Expr) -> Option<Vec<stmt::MatchArm>> {
    match expr {
        stmt::Expr::Match(m) => Some(m.arms),
        stmt::Expr::Cast(c) => into_match_arms(*c.expr),
        _ => None,
    }
}

/// Returns the single primitive leaf of a newtype-chain embed, recursing
/// through nested single-field newtype wrappers. Paired with
/// [`MapField::newtype_leaf_nullable`]: when that returns `Some(_)`, the mapped
/// field has exactly one nested field reachable here as a primitive.
fn single_leaf_primitive(field: &mapping::Field) -> Option<&mapping::FieldPrimitive> {
    match field {
        mapping::Field::Primitive(primitive) => Some(primitive),
        mapping::Field::Struct(s) if s.fields.len() == 1 => single_leaf_primitive(&s.fields[0]),
        _ => None,
    }
}

/// Truncate `name` to `limit` bytes, appending a 5-character stable hash
/// suffix (`_XXXX`) so the result is unique and deterministic across builds.
///
/// The hash is FNV-1a over the *full* pre-truncation name so two names that
/// share a long prefix produce different suffixes. `DefaultHasher` is
/// intentionally avoided — it is unstable across Rust versions and would
/// churn migration snapshots.
fn truncate_identifier(name: String, limit: usize) -> String {
    if name.len() <= limit {
        return name;
    }

    // FNV-1a 32-bit hash of the full name for a stable 4-hex-digit suffix.
    const FNV_OFFSET: u32 = 2_166_136_261;
    const FNV_PRIME: u32 = 16_777_619;
    let hash = name
        .bytes()
        .fold(FNV_OFFSET, |acc, b| acc.wrapping_mul(FNV_PRIME) ^ b as u32);
    let suffix = format!("_{:04x}", hash & 0xFFFF);

    // Truncate at a char boundary so we never split a multi-byte character.
    // The condition uses the char's END position so we never exceed `budget`.
    let budget = limit.saturating_sub(suffix.len());
    let truncated = &name[..name
        .char_indices()
        .take_while(|(i, c)| i + c.len_utf8() <= budget)
        .last()
        .map(|(i, c)| i + c.len_utf8())
        .unwrap_or(0)];

    format!("{truncated}{suffix}")
}

/// Look up an embedded model by ID, returning a descriptive error if not found.
fn lookup_embedded_model<'a>(
    app: &'a app::Schema,
    target: ModelId,
    field: &app::Field,
) -> Result<&'a Model> {
    app.get_model(target).ok_or_else(|| {
        let parent_name = app
            .get_model(field.id.model)
            .map(|m| m.name().upper_camel_case())
            .unwrap_or_else(|| "?".to_string());

        Error::invalid_schema(format!(
            "field `{parent_name}::{}` references an embedded type that is not registered \
             in the schema; did you forget to include the embedded type in \
             `toasty::models!(..)`?",
            field.name,
        ))
    })
}