toasty-core 0.5.0

use super::BuildSchema;
use crate::{
    Error, Result, driver,
    schema::{
        Name,
        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>,
}

/// 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 that belong to an enum variant,
    /// since only the active variant's columns are populated.
    in_enum_variant: 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,
}

impl BuildSchema<'_> {
    pub(super) fn build_table_stub_for_model(&mut self, model: &ModelRoot) -> TableId {
        if let Some(table_name) = model.table_name.as_ref() {
            let table_name = self.prefix_table_name(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()
        } else {
            let name = self.table_name_from_model(&model.name);
            let id = self.register_table(&name);

            self.tables.push(Table::new(id, name));
            id
        }
    }

    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 table_name_from_model(&self, model_name: &Name) -> String {
        let base = pluralizer::pluralize(&model_name.snake_case(), 2, false);
        self.prefix_table_name(&base)
    }

    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![],
        }
        .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: String::new(),
                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 = match mapping {
                    mapping::Field::Primitive(p) => p.column,
                    mapping::Field::Struct(s) => {
                        // Look up the app-level embedded struct to verify this
                        // is a true newtype: exactly one unnamed field.
                        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"
                        );

                        s.fields[0]
                            .as_primitive()
                            .expect(
                                "newtype embedded struct should contain a \
                                 primitive for indexing",
                            )
                            .column
                    }
                    _ => panic!(
                        "only primitive and newtype embedded structs can be \
                         indexed"
                    ),
                };

                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(())
    }

    fn update_index_names(&mut self) {
        for index in &mut self.table.indices {
            index.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 {
                    index.name.push_str("_and");
                }

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

impl BuildMapping<'_> {
    fn build_mapping(mut self, model: &ModelRoot) -> Result<()> {
        let 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)?;

        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));

        Ok(())
    }

    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`.
    fn build_table_to_model_field_enum(
        &self,
        model: &app::EmbeddedEnum,
        mapping: &mapping::FieldEnum,
    ) -> 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,
            });
        }
        // The else branch uses the same Record shape as data arms but with
        // Expr::Error for each field slot. This makes projections work
        // uniformly: projecting [0] extracts disc_col (pruning the errors),
        // while projecting [1] yields Expr::Error (unreachable at runtime).
        let max_fields = model
            .variants
            .iter()
            .enumerate()
            .map(|(i, _)| model.variant_fields(i).count())
            .max()
            .unwrap_or(0);
        let else_expr = 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)
        };

        Ok(stmt::Expr::match_expr(disc_col_ref, arms, else_expr))
    }

    /// 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
    }

    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<_>>()?;
        Ok(stmt::Expr::record(exprs))
    }

    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)
                    }
                    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::HasMany(_) | app::FieldTy::HasOne(_) => {
                Ok(stmt::Value::Null.into())
            }
        }
    }
}

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

    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) => {
                Ok(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::HasMany(_) | app::FieldTy::HasOne(_) => {
                assert!(!self.in_enum_variant);
                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.
    fn map_field_primitive(
        &mut self,
        field_index: usize,
        field: &app::Field,
        primitive: &app::FieldPrimitive,
    ) -> mapping::Field {
        let column_id = self.create_column(field, primitive);
        let expr = self.field_expr(field, field_index);
        let lowering_index = self.build.push_lowering(column_id, &primitive.ty, expr);
        let bit = self.build.next_bit();
        let sub_projection = self.sub_projection(field_index);

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

    /// 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));

        let variants = 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::<Result<_>>()?;

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

        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(),
            },
            variants,
            field_mask,
            sub_projection,
        }))
    }

    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);

        let nested_fields = self
            .for_struct(field, field_index)
            .map_fields(&embedded_struct.fields)?;

        let columns: indexmap::IndexMap<ColumnId, usize> =
            nested_fields.iter().flat_map(|f| f.columns()).collect();
        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,
        }))
    }

    /// 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.in_enum_variant`, and auto-increment from
    /// `field.is_auto_increment()`.
    fn create_column(&mut self, field: &app::Field, primitive: &app::FieldPrimitive) -> ColumnId {
        let mut storage_ty = db::Type::from_app(
            &primitive.ty,
            primitive.storage_ty.as_ref(),
            &self.build.db.storage_types,
        )
        .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: field.nullable || self.in_enum_variant,
            primary_key: false,
            auto_increment: field.is_auto_increment() && self.build.db.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
    /// `in_enum_variant`, `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: &str) -> MapField<'_, 'b> {
        let mut prefix = self.prefix.clone();
        prefix.push(name.to_owned());
        MapField {
            build: self.build,
            prefix,
            in_enum_variant: self.in_enum_variant,
            field_base: self.field_base.clone(),
            field_expr_base: self.field_expr_base.clone(),
        }
    }

    /// Creates a variant-specific child `MapField`.
    ///
    /// Sets `field_base` so that `field_expr` on the child projects from the
    /// enum field, sets `in_enum_variant = true`, and installs a
    /// `field_expr_base` of `match_expr(disc_proj, [arm(discriminant,
    /// Expr::arg(0))], null())` so that every `field_expr` call is
    /// automatically wrapped in the discriminant check.
    fn for_variant(
        &mut self,
        field: &app::Field,
        field_index: usize,
        disc_proj: stmt::Expr,
        discriminant: &stmt::Value,
    ) -> MapField<'_, 'b> {
        let field_base = self.extend_field_base(field, field_index);

        let field_expr_base = stmt::Expr::match_expr(
            disc_proj,
            vec![stmt::MatchArm {
                pattern: discriminant.clone(),
                expr: stmt::Expr::arg(0),
            }],
            stmt::Expr::null(),
        );
        let mut child = self.with_prefix(field.name.storage_name_unwrap());
        child.in_enum_variant = true;
        child.field_base = Some(field_base);
        child.field_expr_base.substitute(&[field_expr_base]);
        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);
        let mut child = self.with_prefix(field.name.storage_name_unwrap());
        child.field_base = Some(field_base);
        child
    }
}

/// 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,
        ))
    })
}