icydb-core 0.180.2

//! Module: db::schema::proposal
//! Responsibility: compiled schema proposal projection from generated entity metadata.
//! Does not own: live schema persistence or compatibility reconciliation.
//! Boundary: turns trusted `EntityModel` data into a typed schema proposal for 0.146.

use crate::{
    db::schema::{
        FieldId, PersistedFieldKind, PersistedFieldSnapshot, PersistedIndexExpressionOp,
        PersistedIndexExpressionSnapshot, PersistedIndexFieldPathSnapshot,
        PersistedIndexKeyItemSnapshot, PersistedIndexKeySnapshot, PersistedIndexSnapshot,
        PersistedNestedLeafSnapshot, PersistedRelationEdgeSnapshot, PersistedSchemaSnapshot,
        SchemaFieldDefault, SchemaFieldSlot, SchemaFieldWritePolicy, SchemaRowLayout,
        SchemaVersion, sql_capabilities,
    },
    model::{
        entity::{EntityModel, RelationEdgeModel},
        field::{FieldDatabaseDefault, FieldKind, FieldModel, FieldStorageDecode, LeafCodec},
        index::{IndexExpression, IndexKeyItem, IndexKeyItemsRef, IndexModel},
    },
};

///
/// CompiledSchemaProposal
///
/// Runtime projection of generated entity metadata into schema-identity terms.
/// This is not the live schema authority; it is the compiled proposal that
/// startup reconciliation will compare with the persisted schema snapshot.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledSchemaProposal {
    entity_path: &'static str,
    entity_name: &'static str,
    declared_schema_version: SchemaVersion,
    primary_key_field_id: FieldId,
    primary_key_field_ids: Vec<FieldId>,
    fields: Vec<CompiledFieldProposal>,
    indexes: Vec<CompiledIndexProposal>,
    relations: Vec<CompiledRelationEdgeProposal>,
}

impl CompiledSchemaProposal {
    /// Return the generated entity path for diagnostics and reconciliation keys.
    #[must_use]
    pub(in crate::db) const fn entity_path(&self) -> &'static str {
        self.entity_path
    }

    /// Return the generated external entity name.
    #[must_use]
    pub(in crate::db) const fn entity_name(&self) -> &'static str {
        self.entity_name
    }

    /// Return the generated source-declared schema version carried by this proposal.
    #[must_use]
    pub(in crate::db) const fn declared_schema_version(&self) -> SchemaVersion {
        self.declared_schema_version
    }

    /// Return the schema field ID assigned to the generated primary key.
    #[must_use]
    pub(in crate::db) const fn first_primary_key_field_id(&self) -> FieldId {
        self.primary_key_field_id
    }

    /// Return the ordered generated primary-key field IDs.
    #[must_use]
    pub(in crate::db) const fn primary_key_field_ids(&self) -> &[FieldId] {
        self.primary_key_field_ids.as_slice()
    }

    /// Return generated field proposals in generated slot order.
    #[must_use]
    pub(in crate::db) const fn fields(&self) -> &[CompiledFieldProposal] {
        self.fields.as_slice()
    }

    /// Return generated field-path index proposals that can already be
    /// represented as accepted schema contracts.
    #[must_use]
    pub(in crate::db) const fn indexes(&self) -> &[CompiledIndexProposal] {
        self.indexes.as_slice()
    }

    /// Return generated relation-edge proposals that can be represented as
    /// accepted source-schema contracts.
    #[must_use]
    pub(in crate::db) const fn relations(&self) -> &[CompiledRelationEdgeProposal] {
        self.relations.as_slice()
    }

    /// Build the initial row layout implied by this compiled proposal.
    ///
    /// This uses proposal-assigned IDs only for first initialization. Once a
    /// persisted schema exists, reconciliation must build the row layout from
    /// stored field IDs and slots instead.
    #[must_use]
    pub(in crate::db) fn initial_row_layout(&self) -> SchemaRowLayout {
        let field_to_slot = self
            .fields()
            .iter()
            .map(|field| (field.id(), field.slot()))
            .collect::<Vec<_>>();

        SchemaRowLayout::new(self.declared_schema_version(), field_to_slot)
    }

    /// Build the initial persisted-schema snapshot implied by this proposal.
    ///
    /// This is only valid for first initialization when no stored schema exists.
    /// Reconciliation must preserve stored field IDs, retired slots, and defaults
    /// once a live persisted schema has been written.
    #[must_use]
    pub(in crate::db) fn initial_persisted_schema_snapshot(&self) -> PersistedSchemaSnapshot {
        let fields = self
            .fields()
            .iter()
            .map(CompiledFieldProposal::initial_persisted_field_snapshot)
            .collect::<Vec<_>>();

        let indexes = self
            .indexes()
            .iter()
            .map(CompiledIndexProposal::initial_persisted_index_snapshot)
            .collect::<Vec<_>>();
        let relations = self
            .relations()
            .iter()
            .map(CompiledRelationEdgeProposal::initial_persisted_relation_snapshot)
            .collect::<Vec<_>>();

        PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
            self.declared_schema_version(),
            self.entity_path().to_string(),
            self.entity_name().to_string(),
            self.primary_key_field_ids().to_vec(),
            self.initial_row_layout(),
            fields,
            indexes,
        )
        .with_relations(relations)
    }
}

///
/// CompiledFieldProposal
///
/// One generated field projected into the schema-identity proposal surface.
/// It carries both durable identity and current generated slot metadata so
/// reconciliation can separate logical field identity from row layout.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledFieldProposal {
    id: FieldId,
    name: &'static str,
    slot: SchemaFieldSlot,
    kind: FieldKind,
    nested_leaves: Vec<PersistedNestedLeafSnapshot>,
    nullable: bool,
    database_default: FieldDatabaseDefault,
    write_policy: SchemaFieldWritePolicy,
    storage_decode: FieldStorageDecode,
    leaf_codec: LeafCodec,
}

impl CompiledFieldProposal {
    /// Return the proposed durable identity for this field.
    #[must_use]
    pub(in crate::db) const fn id(&self) -> FieldId {
        self.id
    }

    /// Return the generated field name.
    #[must_use]
    pub(in crate::db) const fn name(&self) -> &'static str {
        self.name
    }

    /// Return the generated row slot for this field.
    #[must_use]
    pub(in crate::db) const fn slot(&self) -> SchemaFieldSlot {
        self.slot
    }

    /// Return the generated runtime field kind.
    #[must_use]
    pub(in crate::db) const fn kind(&self) -> FieldKind {
        self.kind
    }

    /// Borrow the nested leaf snapshots generated under this top-level field.
    #[must_use]
    pub(in crate::db) const fn nested_leaves(&self) -> &[PersistedNestedLeafSnapshot] {
        self.nested_leaves.as_slice()
    }

    /// Return whether the generated contract permits explicit `NULL`.
    #[must_use]
    pub(in crate::db) const fn nullable(&self) -> bool {
        self.nullable
    }

    /// Return the generated database-level default contract.
    #[must_use]
    pub(in crate::db) const fn database_default(&self) -> FieldDatabaseDefault {
        self.database_default
    }

    /// Return the generated database-level write policy.
    #[must_use]
    pub(in crate::db) const fn write_policy(&self) -> SchemaFieldWritePolicy {
        self.write_policy
    }

    /// Return the generated persisted decode contract.
    #[must_use]
    pub(in crate::db) const fn storage_decode(&self) -> FieldStorageDecode {
        self.storage_decode
    }

    /// Return the generated leaf codec contract.
    #[must_use]
    pub(in crate::db) const fn leaf_codec(&self) -> LeafCodec {
        self.leaf_codec
    }

    /// Build the initial persisted field snapshot implied by this proposal.
    ///
    /// Database defaults intentionally start as `None`; generated Rust defaults
    /// remain construction behavior and are not imported into live schema
    /// authority by this projection.
    #[must_use]
    pub(in crate::db) fn initial_persisted_field_snapshot(&self) -> PersistedFieldSnapshot {
        PersistedFieldSnapshot::new_with_write_policy(
            self.id(),
            self.name().to_string(),
            self.slot(),
            PersistedFieldKind::from_model_kind(self.kind()),
            self.nested_leaves().to_vec(),
            self.nullable(),
            SchemaFieldDefault::from_model_default(self.database_default()),
            self.write_policy(),
            self.storage_decode(),
            self.leaf_codec(),
        )
    }
}

///
/// CompiledIndexProposal
///
/// One generated index projected into accepted schema terms.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledIndexProposal {
    ordinal: u16,
    name: &'static str,
    store: &'static str,
    unique: bool,
    key: CompiledIndexKeyProposal,
    predicate_sql: Option<&'static str>,
}

impl CompiledIndexProposal {
    /// Return the generated stable index ordinal.
    #[must_use]
    pub(in crate::db) const fn ordinal(&self) -> u16 {
        self.ordinal
    }

    /// Return the generated stable index name.
    #[must_use]
    pub(in crate::db) const fn name(&self) -> &'static str {
        self.name
    }

    /// Return the generated backing index store path.
    #[must_use]
    pub(in crate::db) const fn store(&self) -> &'static str {
        self.store
    }

    /// Return whether this index enforces value uniqueness.
    #[must_use]
    pub(in crate::db) const fn unique(&self) -> bool {
        self.unique
    }

    /// Borrow the accepted key proposal.
    #[must_use]
    pub(in crate::db) const fn key(&self) -> &CompiledIndexKeyProposal {
        &self.key
    }

    /// Borrow optional schema-declared predicate SQL display metadata.
    #[must_use]
    pub(in crate::db) const fn predicate_sql(&self) -> Option<&'static str> {
        self.predicate_sql
    }

    /// Build the initial persisted index snapshot implied by this proposal.
    #[must_use]
    pub(in crate::db) fn initial_persisted_index_snapshot(&self) -> PersistedIndexSnapshot {
        PersistedIndexSnapshot::new(
            self.ordinal(),
            self.name().to_string(),
            self.store().to_string(),
            self.unique(),
            self.key().initial_persisted_key_snapshot(),
            self.predicate_sql().map(str::to_string),
        )
    }
}

///
/// CompiledRelationEdgeProposal
///
/// One generated relation edge projected into accepted source-schema terms.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledRelationEdgeProposal {
    name: &'static str,
    target_path: &'static str,
    local_field_ids: Vec<FieldId>,
}

impl CompiledRelationEdgeProposal {
    /// Return the generated relation-edge name.
    #[must_use]
    pub(in crate::db) const fn name(&self) -> &'static str {
        self.name
    }

    /// Return the generated target entity path.
    #[must_use]
    pub(in crate::db) const fn target_path(&self) -> &'static str {
        self.target_path
    }

    /// Borrow ordered accepted local field IDs.
    #[must_use]
    pub(in crate::db) const fn local_field_ids(&self) -> &[FieldId] {
        self.local_field_ids.as_slice()
    }

    /// Build the initial persisted relation-edge snapshot implied by this
    /// proposal.
    #[must_use]
    pub(in crate::db) fn initial_persisted_relation_snapshot(
        &self,
    ) -> PersistedRelationEdgeSnapshot {
        PersistedRelationEdgeSnapshot::new(
            self.name().to_string(),
            self.target_path().to_string(),
            self.local_field_ids().to_vec(),
        )
    }
}

///
/// CompiledIndexKeyProposal
///
/// Accepted-schema projection of one generated index key. Field-path-only keys
/// keep their compact shape; mixed or expression keys preserve explicit item
/// order through `Items`.
///

#[derive(Clone, Debug)]
pub(in crate::db) enum CompiledIndexKeyProposal {
    FieldPath(Vec<CompiledIndexFieldPathProposal>),
    Items(Vec<CompiledIndexKeyItemProposal>),
}

impl CompiledIndexKeyProposal {
    fn initial_persisted_key_snapshot(&self) -> PersistedIndexKeySnapshot {
        match self {
            Self::FieldPath(fields) => PersistedIndexKeySnapshot::FieldPath(
                fields
                    .iter()
                    .map(CompiledIndexFieldPathProposal::initial_persisted_field_path_snapshot)
                    .collect(),
            ),
            Self::Items(items) => PersistedIndexKeySnapshot::Items(
                items
                    .iter()
                    .map(CompiledIndexKeyItemProposal::initial_persisted_key_item_snapshot)
                    .collect(),
            ),
        }
    }
}

///
/// CompiledIndexKeyItemProposal
///
/// One accepted-schema key-item proposal for explicit generated key metadata.
///

#[derive(Clone, Debug)]
pub(in crate::db) enum CompiledIndexKeyItemProposal {
    FieldPath(CompiledIndexFieldPathProposal),
    Expression(CompiledIndexExpressionProposal),
}

impl CompiledIndexKeyItemProposal {
    fn initial_persisted_key_item_snapshot(&self) -> PersistedIndexKeyItemSnapshot {
        match self {
            Self::FieldPath(field_path) => PersistedIndexKeyItemSnapshot::FieldPath(
                field_path.initial_persisted_field_path_snapshot(),
            ),
            Self::Expression(expression) => PersistedIndexKeyItemSnapshot::Expression(Box::new(
                expression.initial_persisted_expression_snapshot(),
            )),
        }
    }
}

///
/// CompiledIndexFieldPathProposal
///
/// One generated index field path resolved to accepted field identity and row
/// slot metadata.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledIndexFieldPathProposal {
    field_id: FieldId,
    slot: SchemaFieldSlot,
    path: Vec<String>,
    kind: PersistedFieldKind,
    nullable: bool,
}

impl CompiledIndexFieldPathProposal {
    /// Return the accepted top-level field identity.
    #[must_use]
    pub(in crate::db) const fn field_id(&self) -> FieldId {
        self.field_id
    }

    /// Return the accepted top-level row slot.
    #[must_use]
    pub(in crate::db) const fn slot(&self) -> SchemaFieldSlot {
        self.slot
    }

    /// Borrow the full accepted field path.
    #[must_use]
    pub(in crate::db) const fn path(&self) -> &[String] {
        self.path.as_slice()
    }

    /// Borrow the accepted persisted field kind at this path.
    #[must_use]
    pub(in crate::db) const fn kind(&self) -> &PersistedFieldKind {
        &self.kind
    }

    /// Return whether this field path permits explicit `NULL`.
    #[must_use]
    pub(in crate::db) const fn nullable(&self) -> bool {
        self.nullable
    }

    fn initial_persisted_field_path_snapshot(&self) -> PersistedIndexFieldPathSnapshot {
        PersistedIndexFieldPathSnapshot::new(
            self.field_id(),
            self.slot(),
            self.path().to_vec(),
            self.kind().clone(),
            self.nullable(),
        )
    }
}

///
/// CompiledIndexExpressionProposal
///
/// One generated expression key item resolved into accepted field identity,
/// accepted row slot, and canonical expression output metadata.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct CompiledIndexExpressionProposal {
    op: PersistedIndexExpressionOp,
    source: CompiledIndexFieldPathProposal,
    output_kind: PersistedFieldKind,
    canonical_text: String,
}

impl CompiledIndexExpressionProposal {
    fn initial_persisted_expression_snapshot(&self) -> PersistedIndexExpressionSnapshot {
        PersistedIndexExpressionSnapshot::new(
            self.op,
            self.source.initial_persisted_field_path_snapshot(),
            self.source.kind().clone(),
            self.output_kind.clone(),
            self.canonical_text.clone(),
        )
    }
}

/// Build the compiled schema proposal for one trusted generated entity model.
#[must_use]
pub(in crate::db) fn compiled_schema_proposal_for_model(
    model: &EntityModel,
) -> CompiledSchemaProposal {
    let fields = model
        .fields()
        .iter()
        .enumerate()
        .map(compiled_field_proposal_from_model_field)
        .collect::<Vec<_>>();
    let indexes = model
        .indexes()
        .iter()
        .filter_map(|index| compiled_index_proposal_from_model_index(index, &fields))
        .collect::<Vec<_>>();
    let relations = model
        .relations()
        .iter()
        .filter_map(|relation| compiled_relation_proposal_from_model_relation(relation, model))
        .collect::<Vec<_>>();

    let proposal = CompiledSchemaProposal {
        entity_path: model.path(),
        entity_name: model.name(),
        declared_schema_version: SchemaVersion::new(model.declared_schema_version()),
        primary_key_field_id: FieldId::from_initial_slot(model.primary_key_slot()),
        primary_key_field_ids: compiled_primary_key_field_ids(model),
        fields,
        indexes,
        relations,
    };

    debug_assert_compiled_schema_proposal_invariants(model, &proposal);

    proposal
}

fn compiled_primary_key_field_ids(model: &EntityModel) -> Vec<FieldId> {
    model
        .primary_key_model()
        .fields()
        .iter()
        .map(|primary_key_field| {
            let slot = model
                .fields()
                .iter()
                .position(|field| std::ptr::eq(field, primary_key_field))
                .unwrap_or_else(|| {
                    panic!(
                        "primary-key field '{}' must be present in generated field table",
                        primary_key_field.name()
                    )
                });
            FieldId::from_initial_slot(slot)
        })
        .collect()
}

// Check the initial proposal projection remains a pure slot-order projection.
// Startup reconciliation will replace this deterministic first-snapshot ID
// assignment with stored IDs once a live persisted schema exists.
fn debug_assert_compiled_schema_proposal_invariants(
    model: &EntityModel,
    proposal: &CompiledSchemaProposal,
) {
    debug_assert_eq!(
        proposal.first_primary_key_field_id(),
        FieldId::from_initial_slot(model.primary_key_slot())
    );
    debug_assert_eq!(
        proposal.primary_key_field_ids().first().copied(),
        Some(proposal.first_primary_key_field_id())
    );

    let layout = proposal.initial_row_layout();
    let snapshot = proposal.initial_persisted_schema_snapshot();
    debug_assert_eq!(layout.version(), proposal.declared_schema_version());
    debug_assert_eq!(
        layout.version().get(),
        proposal.declared_schema_version().get()
    );
    debug_assert_eq!(layout.field_to_slot().len(), proposal.fields().len());
    debug_assert_eq!(snapshot.version(), proposal.declared_schema_version());
    debug_assert_eq!(snapshot.entity_path(), proposal.entity_path());
    debug_assert_eq!(snapshot.entity_name(), proposal.entity_name());
    debug_assert_eq!(
        snapshot.first_primary_key_field_id(),
        proposal.first_primary_key_field_id()
    );
    debug_assert_eq!(snapshot.row_layout(), &layout);
    debug_assert_eq!(snapshot.fields().len(), proposal.fields().len());
    debug_assert_eq!(snapshot.indexes().len(), proposal.indexes().len());
    debug_assert_eq!(snapshot.relations().len(), proposal.relations().len());

    for field in snapshot.fields() {
        let _ = (
            field.id(),
            field.name(),
            field.slot(),
            field.kind(),
            field.nested_leaves(),
            field.nullable(),
            field.default(),
            field.storage_decode(),
            field.leaf_codec(),
        );

        let capabilities = sql_capabilities(field.kind());
        let aggregate = capabilities.aggregate_input();
        let _ = (
            capabilities.selectable(),
            capabilities.comparable(),
            capabilities.orderable(),
            capabilities.groupable(),
            aggregate.count(),
            aggregate.numeric(),
            aggregate.extrema(),
        );
    }

    for (expected_slot, field) in proposal.fields().iter().enumerate() {
        debug_assert_eq!(field.id(), FieldId::from_initial_slot(expected_slot));
        debug_assert_eq!(
            field.slot(),
            SchemaFieldSlot::from_generated_index(expected_slot)
        );

        let _ = (
            field.name(),
            field.kind(),
            field.nullable(),
            field.database_default(),
            field.write_policy(),
            field.storage_decode(),
            field.leaf_codec(),
            field.nested_leaves(),
            field.initial_persisted_field_snapshot(),
        );
    }

    for index in proposal.indexes() {
        let _ = (
            index.ordinal(),
            index.name(),
            index.store(),
            index.unique(),
            index.key(),
            index.predicate_sql(),
            index.initial_persisted_index_snapshot(),
        );
    }

    for relation in proposal.relations() {
        let _ = (
            relation.name(),
            relation.target_path(),
            relation.local_field_ids(),
            relation.initial_persisted_relation_snapshot(),
        );
    }
}

// Project one generated field and its generated slot into the compiled schema
// proposal. This remains a pure projection until live-schema reconciliation
// starts substituting stored field IDs.
fn compiled_field_proposal_from_model_field(
    (slot, field): (usize, &FieldModel),
) -> CompiledFieldProposal {
    let slot = SchemaFieldSlot::from_generated_index(slot);

    CompiledFieldProposal {
        id: FieldId::from_initial_slot(usize::from(slot.get())),
        name: field.name(),
        slot,
        kind: field.kind(),
        nested_leaves: persisted_nested_leaf_snapshots_from_model_fields(field.nested_fields()),
        nullable: field.nullable(),
        database_default: field.database_default(),
        write_policy: SchemaFieldWritePolicy::from_model_policies(
            field.insert_generation(),
            field.write_management(),
        ),
        storage_decode: field.storage_decode(),
        leaf_codec: field.leaf_codec(),
    }
}

// Project one generated index into accepted schema terms.
fn compiled_index_proposal_from_model_index(
    index: &IndexModel,
    fields: &[CompiledFieldProposal],
) -> Option<CompiledIndexProposal> {
    let key = match index.key_items() {
        IndexKeyItemsRef::Fields(field_names) => CompiledIndexKeyProposal::FieldPath(
            field_names
                .iter()
                .map(|field_name| compiled_index_field_path_proposal_from_name(field_name, fields))
                .collect::<Option<Vec<_>>>()?,
        ),
        IndexKeyItemsRef::Items(items) => items
            .iter()
            .map(|item| match item {
                IndexKeyItem::Field(field_name) => {
                    compiled_index_field_path_proposal_from_name(field_name, fields)
                        .map(CompiledIndexKeyItemProposal::FieldPath)
                }
                IndexKeyItem::Expression(expression) => {
                    compiled_index_expression_proposal_from_expression(*expression, fields)
                        .map(CompiledIndexKeyItemProposal::Expression)
                }
            })
            .collect::<Option<Vec<_>>>()
            .map(CompiledIndexKeyProposal::Items)?,
    };

    Some(CompiledIndexProposal {
        ordinal: index.ordinal(),
        name: index.name(),
        store: index.store(),
        unique: index.is_unique(),
        key,
        predicate_sql: index.predicate(),
    })
}

fn compiled_relation_proposal_from_model_relation(
    relation: &RelationEdgeModel,
    model: &EntityModel,
) -> Option<CompiledRelationEdgeProposal> {
    let local_field_ids = relation
        .local_fields()
        .iter()
        .map(|relation_field| {
            model
                .fields()
                .iter()
                .position(|field| std::ptr::eq(field, *relation_field))
                .map(FieldId::from_initial_slot)
        })
        .collect::<Option<Vec<_>>>()?;

    Some(CompiledRelationEdgeProposal {
        name: relation.name(),
        target_path: relation.target_path(),
        local_field_ids,
    })
}

fn compiled_index_expression_proposal_from_expression(
    expression: IndexExpression,
    fields: &[CompiledFieldProposal],
) -> Option<CompiledIndexExpressionProposal> {
    let source = compiled_index_field_path_proposal_from_name(expression.field(), fields)?;
    let op = persisted_expression_op_from_index_expression(expression);
    let output_kind = persisted_expression_output_kind(op, source.kind())?;
    let canonical_text = canonical_expression_text(op, source.path());

    Some(CompiledIndexExpressionProposal {
        op,
        source,
        output_kind,
        canonical_text,
    })
}

const fn persisted_expression_op_from_index_expression(
    expression: IndexExpression,
) -> PersistedIndexExpressionOp {
    match expression {
        IndexExpression::Lower(_) => PersistedIndexExpressionOp::Lower,
        IndexExpression::Upper(_) => PersistedIndexExpressionOp::Upper,
        IndexExpression::Trim(_) => PersistedIndexExpressionOp::Trim,
        IndexExpression::LowerTrim(_) => PersistedIndexExpressionOp::LowerTrim,
        IndexExpression::Date(_) => PersistedIndexExpressionOp::Date,
        IndexExpression::Year(_) => PersistedIndexExpressionOp::Year,
        IndexExpression::Month(_) => PersistedIndexExpressionOp::Month,
        IndexExpression::Day(_) => PersistedIndexExpressionOp::Day,
    }
}

fn persisted_expression_output_kind(
    op: PersistedIndexExpressionOp,
    source_kind: &PersistedFieldKind,
) -> Option<PersistedFieldKind> {
    match op {
        PersistedIndexExpressionOp::Lower
        | PersistedIndexExpressionOp::Upper
        | PersistedIndexExpressionOp::Trim
        | PersistedIndexExpressionOp::LowerTrim => {
            if matches!(source_kind, PersistedFieldKind::Text { .. }) {
                Some(source_kind.clone())
            } else {
                None
            }
        }
        PersistedIndexExpressionOp::Date => {
            if matches!(
                source_kind,
                PersistedFieldKind::Date | PersistedFieldKind::Timestamp
            ) {
                Some(PersistedFieldKind::Date)
            } else {
                None
            }
        }
        PersistedIndexExpressionOp::Year
        | PersistedIndexExpressionOp::Month
        | PersistedIndexExpressionOp::Day => {
            if matches!(
                source_kind,
                PersistedFieldKind::Date | PersistedFieldKind::Timestamp
            ) {
                Some(PersistedFieldKind::Int64)
            } else {
                None
            }
        }
    }
}

fn canonical_expression_text(op: PersistedIndexExpressionOp, path: &[String]) -> String {
    let path = path.join(".");
    match op {
        PersistedIndexExpressionOp::Lower => format!("expr:v1:LOWER({path})"),
        PersistedIndexExpressionOp::Upper => format!("expr:v1:UPPER({path})"),
        PersistedIndexExpressionOp::Trim => format!("expr:v1:TRIM({path})"),
        PersistedIndexExpressionOp::LowerTrim => format!("expr:v1:LOWER(TRIM({path}))"),
        PersistedIndexExpressionOp::Date => format!("expr:v1:DATE({path})"),
        PersistedIndexExpressionOp::Year => format!("expr:v1:YEAR({path})"),
        PersistedIndexExpressionOp::Month => format!("expr:v1:MONTH({path})"),
        PersistedIndexExpressionOp::Day => format!("expr:v1:DAY({path})"),
    }
}

fn compiled_index_field_path_proposal_from_name(
    field_name: &str,
    fields: &[CompiledFieldProposal],
) -> Option<CompiledIndexFieldPathProposal> {
    let path = field_name
        .split('.')
        .map(str::to_string)
        .collect::<Vec<_>>();
    let (top_level, relative_path) = path.split_first()?;
    let field = fields.iter().find(|field| field.name() == top_level)?;

    if relative_path.is_empty() {
        return Some(CompiledIndexFieldPathProposal {
            field_id: field.id(),
            slot: field.slot(),
            path,
            kind: PersistedFieldKind::from_model_kind(field.kind()),
            nullable: field.nullable(),
        });
    }

    let nested = field
        .nested_leaves()
        .iter()
        .find(|leaf| leaf.path() == relative_path)?;

    Some(CompiledIndexFieldPathProposal {
        field_id: field.id(),
        slot: field.slot(),
        path,
        kind: nested.kind().clone(),
        nullable: nested.nullable(),
    })
}

// Flatten generated nested field metadata into path-addressed persisted leaf
// descriptors rooted at one top-level field. The top-level field owns the
// physical slot; nested entries only carry planning metadata for field paths.
fn persisted_nested_leaf_snapshots_from_model_fields(
    fields: &[FieldModel],
) -> Vec<PersistedNestedLeafSnapshot> {
    let mut leaves = Vec::new();

    for field in fields {
        push_persisted_nested_leaf_snapshots(field, Vec::new(), &mut leaves);
    }

    leaves
}

// Record one nested field itself, then recurse through its children so every
// queryable path segment chain has an accepted-schema descriptor.
fn push_persisted_nested_leaf_snapshots(
    field: &FieldModel,
    mut path: Vec<String>,
    leaves: &mut Vec<PersistedNestedLeafSnapshot>,
) {
    path.push(field.name().to_string());
    leaves.push(PersistedNestedLeafSnapshot::new(
        path.clone(),
        PersistedFieldKind::from_model_kind(field.kind()),
        field.nullable(),
        field.storage_decode(),
        field.leaf_codec(),
    ));

    for nested in field.nested_fields() {
        push_persisted_nested_leaf_snapshots(nested, path.clone(), leaves);
    }
}

///
/// TESTS
///

#[cfg(test)]
mod tests;