icydb-core 0.180.10

//! Module: db::schema::info
//! Responsibility: schema model/index integrity checks used during schema info construction.
//! Does not own: query planning policy or runtime predicate evaluation.
//! Boundary: validates entity/index model consistency for predicate schema metadata.

#[cfg(feature = "sql")]
use crate::db::schema::{SqlCapabilities, sql_capabilities};
#[cfg(any(test, feature = "sql"))]
use crate::{
    db::schema::canonicalize_strict_sql_literal_for_persisted_kind,
    model::canonicalize_strict_sql_literal_for_kind, value::Value,
};
use crate::{
    db::schema::{
        AcceptedSchemaSnapshot, FieldId, FieldType, PersistedFieldKind, PersistedFieldSnapshot,
        PersistedIndexExpressionOp, PersistedIndexFieldPathSnapshot, PersistedIndexKeyItemSnapshot,
        PersistedIndexKeySnapshot, PersistedIndexSnapshot, PersistedNestedLeafSnapshot,
        PersistedRelationStrength, PersistedSchemaSnapshot, SchemaFieldSlot,
        field_type_from_model_kind, field_type_from_persisted_kind,
    },
    model::{
        entity::EntityModel,
        field::{FieldKind, FieldModel, LeafCodec},
        index::{IndexKeyItem, IndexKeyItemsRef, IndexModel},
    },
};
#[cfg(test)]
use std::cell::Cell;
use std::sync::{Mutex, OnceLock};

type SchemaFieldEntry = (String, SchemaFieldInfo);
type CachedSchemaEntries = Vec<(&'static str, &'static SchemaInfo)>;
const EMPTY_GENERATED_NESTED_FIELDS: &[FieldModel] = &[];

#[cfg(test)]
thread_local! {
    static ACCEPTED_SCHEMA_INFO_PROJECTIONS: Cell<u64> = const { Cell::new(0) };
}

#[cfg(test)]
pub(in crate::db) fn reset_accepted_schema_info_projection_count_for_tests() {
    ACCEPTED_SCHEMA_INFO_PROJECTIONS.with(|projections| projections.set(0));
}

#[cfg(test)]
pub(in crate::db) fn accepted_schema_info_projection_count_for_tests() -> u64 {
    ACCEPTED_SCHEMA_INFO_PROJECTIONS.with(Cell::get)
}

fn schema_field_info<'a>(
    fields: &'a [SchemaFieldEntry],
    name: &str,
) -> Option<&'a SchemaFieldInfo> {
    fields
        .binary_search_by(|(field_name, _)| field_name.as_str().cmp(name))
        .ok()
        .map(|index| &fields[index].1)
}

fn generated_field_by_name<'a>(
    model: &'a EntityModel,
    field_name: &str,
) -> Option<(usize, &'a FieldModel)> {
    model
        .fields()
        .iter()
        .enumerate()
        .find(|(_, field)| field.name() == field_name)
}

// Attach generated index-membership facts to generated `SchemaInfo` views.
fn generated_field_is_indexed(model: &EntityModel, field_name: &str) -> bool {
    model
        .indexes()
        .iter()
        .any(|index| index.fields().contains(&field_name))
}

// Resolve top-level index membership from accepted persisted index contracts
// once per schema view. Runtime accepted schema views must not reopen generated
// `EntityModel` indexes after schema acceptance.
fn accepted_indexed_field_ids(snapshot: &PersistedSchemaSnapshot) -> Vec<FieldId> {
    let mut field_ids = Vec::new();

    for index in snapshot.indexes() {
        for field in snapshot.fields() {
            if index.key().references_field(field.id()) && !field_ids.contains(&field.id()) {
                field_ids.push(field.id());
            }
        }
    }

    field_ids
}

fn accepted_field_name(snapshot: &PersistedSchemaSnapshot, field_id: FieldId) -> Option<&str> {
    snapshot
        .fields()
        .iter()
        .find(|field| field.id() == field_id)
        .map(PersistedFieldSnapshot::name)
}

// Convert a schema-owned row-layout slot into the usize slot surface consumed
// by planner and executor DTOs.
fn accepted_slot_index(slot: SchemaFieldSlot) -> usize {
    usize::from(slot.get())
}

fn persisted_kind_has_strong_relation(kind: &PersistedFieldKind) -> bool {
    match kind {
        PersistedFieldKind::Relation { strength, .. } => {
            *strength == PersistedRelationStrength::Strong
        }
        PersistedFieldKind::List(inner) | PersistedFieldKind::Set(inner) => {
            persisted_kind_has_strong_relation(inner)
        }
        _ => false,
    }
}

///
/// SchemaInfo
///
/// Lightweight, runtime-usable field-type map for one entity.
/// This is the *only* schema surface the predicate validator depends on.
///

///
/// SchemaFieldInfo
///
/// Compact per-field schema entry used by `SchemaInfo`.
/// Keeps reduced predicate type metadata and the temporary generated field-kind
/// bridge in one table while accepted persisted facts become the field-list
/// authority for SQL/session paths.
///

#[derive(Clone, Debug)]
struct SchemaFieldInfo {
    slot: usize,
    ty: FieldType,
    kind: Option<FieldKind>,
    nullable: bool,
    leaf_codec: LeafCodec,
    #[cfg(feature = "sql")]
    sql_capabilities: SqlCapabilities,
    #[cfg(feature = "sql")]
    persisted_kind: Option<PersistedFieldKind>,
    indexed: bool,
    nested_leaves: Option<Vec<PersistedNestedLeafSnapshot>>,
    nested_fields: &'static [FieldModel],
}

///
/// SchemaIndexInfo
///
/// Compact field-path index contract exposed by `SchemaInfo`.
/// Accepted schema views source this from persisted index snapshots; generated
/// schema views source it from generated field-only index metadata for
/// proposal/model-only compatibility callers.
///
#[derive(Clone, Debug)]
#[allow(
    dead_code,
    reason = "0.150 staged accepted-index authority surface; planner/explain routing consumes this DTO in the next runtime slice"
)]
pub(in crate::db) struct SchemaIndexInfo {
    ordinal: u16,
    name: String,
    store: String,
    unique: bool,
    generated: bool,
    fields: Vec<SchemaIndexFieldPathInfo>,
    predicate_sql: Option<String>,
}

#[allow(
    dead_code,
    reason = "0.150 staged accepted-index authority surface; planner/explain routing consumes this DTO in the next runtime slice"
)]
impl SchemaIndexInfo {
    /// Return the accepted or generated stable per-entity index ordinal.
    #[must_use]
    pub(in crate::db) const fn ordinal(&self) -> u16 {
        self.ordinal
    }

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

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

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

    /// Return whether this index is declared by the generated entity model.
    #[must_use]
    pub(in crate::db) const fn generated(&self) -> bool {
        self.generated
    }

    /// Borrow accepted field-path key item metadata for this index.
    #[must_use]
    pub(in crate::db) const fn fields(&self) -> &[SchemaIndexFieldPathInfo] {
        self.fields.as_slice()
    }

    /// Borrow optional predicate SQL display metadata.
    #[must_use]
    pub(in crate::db) const fn predicate_sql(&self) -> Option<&str> {
        match &self.predicate_sql {
            Some(sql) => Some(sql.as_str()),
            None => None,
        }
    }
}

///
/// SchemaExpressionIndexInfo
///
/// Compact accepted expression-index contract exposed by `SchemaInfo`.
/// Accepted schema views source this from persisted index snapshots so
/// expression-index runtime routing can stop reopening generated `IndexModel`.
///
#[derive(Clone, Debug)]
#[allow(
    dead_code,
    reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
)]
pub(in crate::db) struct SchemaExpressionIndexInfo {
    ordinal: u16,
    name: String,
    store: String,
    unique: bool,
    generated: bool,
    key_items: Vec<SchemaExpressionIndexKeyItemInfo>,
    predicate_sql: Option<String>,
}

#[allow(
    dead_code,
    reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
)]
impl SchemaExpressionIndexInfo {
    /// Return the accepted stable per-entity index ordinal.
    #[must_use]
    pub(in crate::db) const fn ordinal(&self) -> u16 {
        self.ordinal
    }

    /// Borrow the accepted stable index name.
    #[must_use]
    pub(in crate::db) const fn name(&self) -> &str {
        self.name.as_str()
    }

    /// Borrow the accepted backing index store path.
    #[must_use]
    pub(in crate::db) const fn store(&self) -> &str {
        self.store.as_str()
    }

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

    /// Return whether this expression index came from generated entity metadata.
    #[must_use]
    pub(in crate::db) const fn generated(&self) -> bool {
        self.generated
    }

    /// Borrow accepted key-item contracts in index key order.
    #[must_use]
    pub(in crate::db) const fn key_items(&self) -> &[SchemaExpressionIndexKeyItemInfo] {
        self.key_items.as_slice()
    }

    /// Borrow optional accepted index-membership predicate SQL metadata.
    #[must_use]
    pub(in crate::db) const fn predicate_sql(&self) -> Option<&str> {
        match &self.predicate_sql {
            Some(sql) => Some(sql.as_str()),
            None => None,
        }
    }
}

///
/// SchemaExpressionIndexKeyItemInfo
///
/// Accepted expression-index key item surfaced through `SchemaInfo`.
///
#[derive(Clone, Debug)]
#[allow(
    dead_code,
    reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
)]
pub(in crate::db) enum SchemaExpressionIndexKeyItemInfo {
    FieldPath(SchemaIndexFieldPathInfo),
    Expression(Box<SchemaIndexExpressionInfo>),
}

#[expect(
    dead_code,
    reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
)]
impl SchemaExpressionIndexKeyItemInfo {
    /// Borrow this key item as a field-path component, when applicable.
    #[must_use]
    pub(in crate::db) const fn field_path(&self) -> Option<&SchemaIndexFieldPathInfo> {
        match self {
            Self::FieldPath(field_path) => Some(field_path),
            Self::Expression(_) => None,
        }
    }

    /// Borrow this key item as an expression component, when applicable.
    #[must_use]
    pub(in crate::db) fn expression(&self) -> Option<&SchemaIndexExpressionInfo> {
        match self {
            Self::FieldPath(_) => None,
            Self::Expression(expression) => Some(expression.as_ref()),
        }
    }
}

///
/// SchemaIndexExpressionInfo
///
/// Compact accepted expression key contract for one expression-index key item.
///
#[derive(Clone, Debug)]
#[allow(
    dead_code,
    reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
)]
pub(in crate::db) struct SchemaIndexExpressionInfo {
    op: PersistedIndexExpressionOp,
    source: SchemaIndexFieldPathInfo,
    input_kind: PersistedFieldKind,
    output_kind: PersistedFieldKind,
    canonical_text: String,
}

#[cfg_attr(
    not(test),
    expect(
        dead_code,
        reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
    )
)]
impl SchemaIndexExpressionInfo {
    /// Return the accepted expression operation.
    #[must_use]
    pub(in crate::db) const fn op(&self) -> PersistedIndexExpressionOp {
        self.op
    }

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

    /// Borrow the accepted expression input kind.
    #[must_use]
    pub(in crate::db) const fn input_kind(&self) -> &PersistedFieldKind {
        &self.input_kind
    }

    /// Borrow the accepted expression output kind.
    #[must_use]
    pub(in crate::db) const fn output_kind(&self) -> &PersistedFieldKind {
        &self.output_kind
    }

    /// Borrow the accepted canonical expression text.
    #[must_use]
    pub(in crate::db) const fn canonical_text(&self) -> &str {
        self.canonical_text.as_str()
    }
}

///
/// SchemaIndexFieldPathInfo
///
/// Compact key-item contract for one field-path index component.
/// Accepted schema views carry durable field IDs and persisted kinds; generated
/// compatibility views omit field IDs until generated metadata is reconciled.
///
#[derive(Clone, Debug)]
#[allow(
    dead_code,
    reason = "0.150 staged accepted-index authority surface; planner/explain routing consumes this DTO in the next runtime slice"
)]
pub(in crate::db) struct SchemaIndexFieldPathInfo {
    field_id: Option<FieldId>,
    field_name: String,
    slot: usize,
    path: Vec<String>,
    ty: FieldType,
    persisted_kind: Option<PersistedFieldKind>,
    nullable: bool,
}

#[cfg_attr(
    not(test),
    expect(
        dead_code,
        reason = "0.150 staged accepted-index authority surface; planner/explain routing consumes this DTO in the next runtime slice"
    )
)]
impl SchemaIndexFieldPathInfo {
    /// Return the accepted durable top-level field ID, when this came from a
    /// persisted schema snapshot.
    #[must_use]
    pub(in crate::db) const fn field_id(&self) -> Option<FieldId> {
        self.field_id
    }

    /// Borrow the top-level field name for this key item.
    #[must_use]
    pub(in crate::db) const fn field_name(&self) -> &str {
        self.field_name.as_str()
    }

    /// Return the schema-owned top-level row slot for this key item.
    #[must_use]
    pub(in crate::db) const fn slot(&self) -> usize {
        self.slot
    }

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

    /// Borrow reduced predicate/query type metadata for this key item.
    #[must_use]
    pub(in crate::db) const fn ty(&self) -> &FieldType {
        &self.ty
    }

    /// Borrow the persisted field kind, when this key item came from accepted
    /// schema authority.
    #[must_use]
    pub(in crate::db) const fn persisted_kind(&self) -> Option<&PersistedFieldKind> {
        match &self.persisted_kind {
            Some(kind) => Some(kind),
            None => None,
        }
    }

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

#[derive(Clone, Debug)]
pub(crate) struct SchemaInfo {
    fields: Vec<SchemaFieldEntry>,
    indexes: Vec<SchemaIndexInfo>,
    expression_indexes: Vec<SchemaExpressionIndexInfo>,
    entity_path: Option<String>,
    entity_name: Option<String>,
    primary_key_names: Vec<String>,
    has_any_strong_relations: bool,
}

impl SchemaInfo {
    // Build one compact field table from trusted generated field metadata.
    fn from_trusted_field_models(fields: &[FieldModel]) -> Self {
        let mut fields = fields
            .iter()
            .enumerate()
            .map(|(slot, field)| {
                (
                    field.name().to_string(),
                    SchemaFieldInfo {
                        slot,
                        ty: field_type_from_model_kind(&field.kind()),
                        kind: Some(field.kind()),
                        nullable: field.nullable(),
                        leaf_codec: field.leaf_codec(),
                        #[cfg(feature = "sql")]
                        sql_capabilities: sql_capabilities(&PersistedFieldKind::from_model_kind(
                            field.kind(),
                        )),
                        #[cfg(feature = "sql")]
                        persisted_kind: None,
                        indexed: false,
                        nested_leaves: None,
                        nested_fields: field.nested_fields(),
                    },
                )
            })
            .collect::<Vec<_>>();

        fields.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));

        Self {
            fields,
            indexes: Vec::new(),
            expression_indexes: Vec::new(),
            entity_path: None,
            entity_name: None,
            primary_key_names: Vec::new(),
            has_any_strong_relations: false,
        }
    }

    // Build one compact field table from trusted generated entity metadata.
    fn from_trusted_entity_model(model: &EntityModel) -> Self {
        let mut schema = Self::from_trusted_field_models(model.fields());
        schema.entity_path = Some(model.path().to_string());
        schema.entity_name = Some(model.name().to_string());
        schema.primary_key_names = model
            .primary_key_model()
            .fields()
            .iter()
            .map(|field| field.name().to_string())
            .collect();
        schema.has_any_strong_relations = model.has_any_strong_relations();

        for (field_name, field) in &mut schema.fields {
            field.indexed = generated_field_is_indexed(model, field_name.as_str());
        }
        schema.indexes = model
            .indexes()
            .iter()
            .filter_map(|index| schema_index_info_from_generated_index(index, &schema.fields))
            .collect();
        schema.expression_indexes = Vec::new();

        schema
    }

    #[must_use]
    pub(crate) fn field(&self, name: &str) -> Option<&FieldType> {
        schema_field_info(self.fields.as_slice(), name).map(|field| &field.ty)
    }

    #[must_use]
    pub(crate) fn field_kind(&self, name: &str) -> Option<&FieldKind> {
        schema_field_info(self.fields.as_slice(), name).and_then(|field| field.kind.as_ref())
    }

    /// Return the top-level physical row slot for one field.
    ///
    /// Accepted schema views source this from `SchemaRowLayout`; generated
    /// schema views keep using generated field-table position. The method gives
    /// planning validation one schema-owned slot surface instead of requiring
    /// direct `EntityModel` field-table checks.
    #[must_use]
    pub(in crate::db) fn field_slot_index(&self, name: &str) -> Option<usize> {
        schema_field_info(self.fields.as_slice(), name).map(|field| field.slot)
    }

    /// Return whether one top-level field permits explicit persisted `NULL`.
    ///
    /// Accepted schema views source this from persisted field snapshots, while
    /// generated schema views retain generated field metadata for test-only
    /// compatibility callers.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn field_nullable(&self, name: &str) -> Option<bool> {
        schema_field_info(self.fields.as_slice(), name).map(|field| field.nullable)
    }

    /// Return whether one top-level row slot is backed by a scalar leaf codec.
    ///
    /// Accepted schema views source this from persisted field snapshots, giving
    /// predicate fast-path classification schema authority instead of generated
    /// model field tables.
    #[must_use]
    pub(in crate::db) fn field_slot_has_scalar_leaf(&self, slot: usize) -> bool {
        self.fields
            .iter()
            .find(|(_, field)| field.slot == slot)
            .is_some_and(|(_, field)| matches!(field.leaf_codec, LeafCodec::Scalar(_)))
    }

    /// Borrow the schema-owned entity name when this schema view was built
    /// from an entity model or accepted persisted snapshot.
    #[must_use]
    #[cfg(any(test, feature = "sql"))]
    pub(in crate::db) fn entity_name(&self) -> Option<&str> {
        self.entity_name.as_deref()
    }

    /// Borrow the schema-owned entity path when this schema view was built
    /// from an entity model or accepted persisted snapshot.
    #[must_use]
    #[expect(
        dead_code,
        reason = "schema views retain entity-path authority for diagnostics and DDL/reporting slices"
    )]
    pub(in crate::db) fn entity_path(&self) -> Option<&str> {
        self.entity_path.as_deref()
    }

    /// Borrow the schema-owned primary-key field name for scalar primary-key
    /// entities. Composite entities return `None` so scalar access-planning
    /// helpers cannot silently treat the first component as a complete key.
    #[must_use]
    pub(in crate::db) fn scalar_primary_key_name(&self) -> Option<&str> {
        (self.primary_key_names.len() == 1).then(|| self.primary_key_names[0].as_str())
    }

    /// Borrow schema-owned primary-key field names in accepted key order.
    ///
    /// Callers that need deterministic ordering or composite identity must use
    /// the full ordered slice.
    #[must_use]
    pub(in crate::db) const fn primary_key_names(&self) -> &[String] {
        self.primary_key_names.as_slice()
    }

    /// Return whether this entity has any strong relation checks.
    ///
    /// Accepted schema views source this from persisted relation field
    /// contracts. Generated schema views source it from generated model
    /// metadata only for proposal/model-only callers.
    #[must_use]
    pub(in crate::db) const fn has_any_strong_relations(&self) -> bool {
        self.has_any_strong_relations
    }

    /// Return whether one top-level field participates in any index.
    ///
    /// Accepted schema views source this from persisted index contracts.
    /// Generated schema views source it from generated index metadata for
    /// proposal/model-only callers.
    #[must_use]
    pub(in crate::db) fn field_is_indexed(&self, name: &str) -> bool {
        schema_field_info(self.fields.as_slice(), name).is_some_and(|field| field.indexed)
    }

    /// Borrow field-path index contracts visible through this schema view.
    ///
    /// Accepted schema views source this from persisted index contracts.
    /// Generated schema views source it from generated field-only indexes for
    /// proposal/model-only compatibility.
    #[must_use]
    #[allow(
        dead_code,
        reason = "0.150 staged accepted-index authority surface; planner/explain routing consumes this accessor in the next runtime slice"
    )]
    pub(in crate::db) const fn field_path_indexes(&self) -> &[SchemaIndexInfo] {
        self.indexes.as_slice()
    }

    /// Borrow accepted expression-index contracts visible through this schema view.
    ///
    /// Accepted schema views source this from persisted expression index
    /// contracts. Generated schema views leave this empty until generated
    /// expression indexes have been reconciled into accepted metadata.
    #[must_use]
    #[allow(
        dead_code,
        reason = "0.151 stages accepted expression-index authority for the next planner/write routing slice"
    )]
    pub(in crate::db) const fn expression_indexes(&self) -> &[SchemaExpressionIndexInfo] {
        self.expression_indexes.as_slice()
    }

    /// Return SQL operation capabilities for one top-level field.
    ///
    /// Accepted live schema views derive this from persisted field kinds so SQL
    /// admission follows reconciled schema authority. Generated schema views
    /// use generated model metadata for compile-time-only callers.
    ///
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn sql_capabilities(&self, name: &str) -> Option<SqlCapabilities> {
        schema_field_info(self.fields.as_slice(), name).map(|field| field.sql_capabilities)
    }

    /// Return whether one top-level field stores a structured record value.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn field_is_structured_value(&self, name: &str) -> bool {
        schema_field_info(self.fields.as_slice(), name)
            .is_some_and(|field| matches!(field.ty, FieldType::Structured { .. }))
    }

    /// Return SQL operation capabilities for one nested field path.
    ///
    /// Accepted schema views resolve nested paths from persisted nested leaf
    /// metadata. Generated schema views derive the same facts from generated
    /// nested `FieldModel` metadata until live row-layout authority exists.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn nested_sql_capabilities(
        &self,
        name: &str,
        segments: &[String],
    ) -> Option<SqlCapabilities> {
        let field = schema_field_info(self.fields.as_slice(), name)?;

        if let Some(nested_leaves) = field.nested_leaves.as_ref() {
            return nested_leaves
                .iter()
                .find(|leaf| leaf.path() == segments)
                .map(|leaf| sql_capabilities(leaf.kind()));
        }

        resolve_nested_field_path_kind(field.nested_fields, segments)
            .map(|kind| sql_capabilities(&PersistedFieldKind::from_model_kind(kind)))
    }

    /// Return the first top-level field that SQL cannot project directly.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn first_non_sql_selectable_field(&self) -> Option<&str> {
        self.fields
            .iter()
            .find(|(_, field)| !field.sql_capabilities.selectable())
            .map(|(field_name, _)| field_name.as_str())
    }

    /// Return the type for one nested field path rooted at a top-level field.
    ///
    /// Accepted schema views resolve nested paths from persisted nested leaf
    /// metadata. Generated schema views retain generated nested `FieldModel`
    /// traversal for compile-time-only callers.
    #[must_use]
    pub(crate) fn nested_field_type(&self, name: &str, segments: &[String]) -> Option<FieldType> {
        let field = schema_field_info(self.fields.as_slice(), name)?;

        if let Some(nested_leaves) = field.nested_leaves.as_ref() {
            return nested_leaves
                .iter()
                .find(|leaf| leaf.path() == segments)
                .map(|leaf| field_type_from_persisted_kind(leaf.kind()));
        }

        resolve_nested_field_path_kind(field.nested_fields, segments)
            .map(|kind| field_type_from_model_kind(&kind))
    }

    /// Return whether one top-level field exposes any nested path metadata.
    #[must_use]
    pub(crate) fn field_has_nested_paths(&self, name: &str) -> bool {
        schema_field_info(self.fields.as_slice(), name).is_some_and(|field| {
            field.nested_leaves.as_ref().map_or_else(
                || !field.nested_fields.is_empty(),
                |leaves| !leaves.is_empty(),
            )
        })
    }

    /// Canonicalize one strict SQL literal against this schema's field authority.
    ///
    /// Accepted live schemas use persisted field kinds so SQL read predicates
    /// follow the same top-level type boundary as SQL writes and planning.
    /// Generated schema views retain the old generated-kind fallback for
    /// direct lowering tests and compile-time-only callers.
    ///
    #[cfg(any(test, feature = "sql"))]
    #[must_use]
    pub(in crate::db) fn canonicalize_strict_sql_literal(
        &self,
        field_name: &str,
        value: &Value,
    ) -> Option<Value> {
        let field = schema_field_info(self.fields.as_slice(), field_name)?;

        if let Some(kind) = field.persisted_kind.as_ref() {
            return canonicalize_strict_sql_literal_for_persisted_kind(kind, value);
        }

        field
            .kind
            .as_ref()
            .and_then(|kind| canonicalize_strict_sql_literal_for_kind(kind, value))
    }

    /// Build one owned schema view from trusted generated field metadata.
    #[must_use]
    pub(crate) fn from_field_models(fields: &[FieldModel]) -> Self {
        Self::from_trusted_field_models(fields)
    }

    /// Build one owned schema view from an accepted persisted snapshot.
    ///
    /// This is the live-schema counterpart to the generated metadata cache.
    /// Accepted nested-path authority comes from persisted nested leaf
    /// descriptors; generated nested metadata is retained only for model-only
    /// views and compile-time/test bridge parity.
    #[must_use]
    pub(in crate::db) fn from_accepted_snapshot_for_model(
        model: &EntityModel,
        schema: &AcceptedSchemaSnapshot,
    ) -> Self {
        Self::from_accepted_snapshot_for_model_with_expression_indexes(model, schema, false)
    }

    /// Build one owned schema view from an accepted persisted snapshot and
    /// optionally project accepted expression-index metadata.
    ///
    /// Expression-index metadata is intentionally opt-in until planner/write
    /// routing consumes it. The ordinary accepted `SchemaInfo` path is hot
    /// during query loops and should not pay for staged DTO projection.
    #[must_use]
    pub(in crate::db) fn from_accepted_snapshot_for_model_with_expression_indexes(
        model: &EntityModel,
        schema: &AcceptedSchemaSnapshot,
        include_expression_indexes: bool,
    ) -> Self {
        #[cfg(test)]
        ACCEPTED_SCHEMA_INFO_PROJECTIONS
            .with(|projections| projections.set(projections.get().saturating_add(1)));

        let snapshot = schema.persisted_snapshot();
        let indexed_field_ids = accepted_indexed_field_ids(snapshot);
        let mut fields = snapshot
            .fields()
            .iter()
            .map(|field| {
                let generated_field = generated_field_by_name(model, field.name());
                let slot = snapshot
                    .row_layout()
                    .slot_for_field(field.id())
                    .map_or_else(|| usize::from(field.slot().get()), accepted_slot_index);
                let generated_kind = generated_field.map(|(_, field)| field.kind());
                let generated_nested_fields = generated_field
                    .map_or(EMPTY_GENERATED_NESTED_FIELDS, |(_, field)| {
                        field.nested_fields()
                    });

                (
                    field.name().to_string(),
                    SchemaFieldInfo {
                        slot,
                        ty: field_type_from_persisted_kind(field.kind()),
                        kind: generated_kind,
                        nullable: field.nullable(),
                        leaf_codec: field.leaf_codec(),
                        #[cfg(feature = "sql")]
                        sql_capabilities: sql_capabilities(field.kind()),
                        #[cfg(feature = "sql")]
                        persisted_kind: Some(field.kind().clone()),
                        indexed: indexed_field_ids.contains(&field.id()),
                        nested_leaves: Some(field.nested_leaves().to_vec()),
                        nested_fields: generated_nested_fields,
                    },
                )
            })
            .collect::<Vec<_>>();

        fields.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));

        let primary_key_names = snapshot
            .primary_key_field_ids()
            .iter()
            .filter_map(|field_id| {
                snapshot
                    .fields()
                    .iter()
                    .find(|field| field.id() == *field_id)
                    .map(|field| field.name().to_string())
            })
            .collect();

        Self {
            fields,
            indexes: snapshot
                .indexes()
                .iter()
                .filter_map(|index| schema_index_info_from_accepted_index(index, snapshot))
                .collect(),
            expression_indexes: snapshot
                .indexes()
                .iter()
                .filter_map(|index| {
                    include_expression_indexes
                        .then(|| schema_expression_index_info_from_accepted_index(index, snapshot))
                        .flatten()
                })
                .collect(),
            entity_path: Some(schema.entity_path().to_string()),
            entity_name: Some(schema.entity_name().to_string()),
            primary_key_names,
            has_any_strong_relations: !snapshot.relations().is_empty()
                || snapshot
                    .fields()
                    .iter()
                    .any(|field| persisted_kind_has_strong_relation(field.kind())),
        }
    }

    /// Build one accepted schema view with expression-index metadata projected.
    ///
    /// This constructor exists for the 0.151 expression-index routing slice and
    /// tests that need to inspect accepted expression contracts without adding
    /// allocation work to every existing accepted schema view.
    #[must_use]
    #[cfg(test)]
    fn from_accepted_snapshot_for_model_including_expression_indexes(
        model: &EntityModel,
        schema: &AcceptedSchemaSnapshot,
    ) -> Self {
        Self::from_accepted_snapshot_for_model_with_expression_indexes(model, schema, true)
    }

    /// Return one cached schema view for a trusted generated entity model.
    pub(crate) fn cached_for_generated_entity_model(model: &EntityModel) -> &'static Self {
        static CACHE: OnceLock<Mutex<CachedSchemaEntries>> = OnceLock::new();

        let cache = CACHE.get_or_init(|| Mutex::new(CachedSchemaEntries::new()));
        let mut guard = cache.lock().expect("schema info cache mutex poisoned");
        if let Some(cached) = guard
            .iter()
            .find(|(entity_path, _)| *entity_path == model.path())
            .map(|(_, schema)| *schema)
        {
            return cached;
        }

        let schema = Box::leak(Box::new(Self::from_trusted_entity_model(model)));
        guard.push((model.path(), schema));
        schema
    }
}

fn schema_index_info_from_generated_index(
    index: &IndexModel,
    fields: &[SchemaFieldEntry],
) -> Option<SchemaIndexInfo> {
    let key_fields = generated_index_field_names(index)?
        .into_iter()
        .map(|field_name| {
            let field = schema_field_info(fields, field_name)?;
            Some(SchemaIndexFieldPathInfo {
                field_id: None,
                field_name: field_name.to_string(),
                slot: field.slot,
                path: vec![field_name.to_string()],
                ty: field.ty.clone(),
                persisted_kind: None,
                nullable: field.nullable,
            })
        })
        .collect::<Option<Vec<_>>>()?;

    Some(SchemaIndexInfo {
        ordinal: index.ordinal(),
        name: index.name().to_string(),
        store: index.store().to_string(),
        unique: index.is_unique(),
        generated: true,
        fields: key_fields,
        predicate_sql: index.predicate().map(str::to_string),
    })
}

fn generated_index_field_names(index: &IndexModel) -> Option<Vec<&'static str>> {
    match index.key_items() {
        IndexKeyItemsRef::Fields(fields) => Some(fields.to_vec()),
        IndexKeyItemsRef::Items(items) => items
            .iter()
            .map(|item| match item {
                IndexKeyItem::Field(field) => Some(*field),
                IndexKeyItem::Expression(_) => None,
            })
            .collect(),
    }
}

fn schema_index_info_from_accepted_index(
    index: &PersistedIndexSnapshot,
    snapshot: &PersistedSchemaSnapshot,
) -> Option<SchemaIndexInfo> {
    if !index.key().is_field_path_only() {
        return None;
    }

    Some(SchemaIndexInfo {
        ordinal: index.ordinal(),
        name: index.name().to_string(),
        store: index.store().to_string(),
        unique: index.unique(),
        generated: index.generated(),
        fields: index
            .key()
            .field_paths()
            .iter()
            .map(|path| schema_index_field_path_info_from_accepted(path, snapshot))
            .collect(),
        predicate_sql: index.predicate_sql().map(str::to_string),
    })
}

fn schema_expression_index_info_from_accepted_index(
    index: &PersistedIndexSnapshot,
    snapshot: &PersistedSchemaSnapshot,
) -> Option<SchemaExpressionIndexInfo> {
    let PersistedIndexKeySnapshot::Items(items) = index.key() else {
        return None;
    };

    if !items
        .iter()
        .any(|item| matches!(item, PersistedIndexKeyItemSnapshot::Expression(_)))
    {
        return None;
    }

    Some(SchemaExpressionIndexInfo {
        ordinal: index.ordinal(),
        name: index.name().to_string(),
        store: index.store().to_string(),
        unique: index.unique(),
        generated: index.generated(),
        key_items: items
            .iter()
            .map(|item| schema_expression_index_key_item_info(item, snapshot))
            .collect(),
        predicate_sql: index.predicate_sql().map(str::to_string),
    })
}

fn schema_expression_index_key_item_info(
    item: &PersistedIndexKeyItemSnapshot,
    snapshot: &PersistedSchemaSnapshot,
) -> SchemaExpressionIndexKeyItemInfo {
    match item {
        PersistedIndexKeyItemSnapshot::FieldPath(path) => {
            SchemaExpressionIndexKeyItemInfo::FieldPath(schema_index_field_path_info_from_accepted(
                path, snapshot,
            ))
        }
        PersistedIndexKeyItemSnapshot::Expression(expression) => {
            SchemaExpressionIndexKeyItemInfo::Expression(Box::new(SchemaIndexExpressionInfo {
                op: expression.op(),
                source: schema_index_field_path_info_from_accepted(expression.source(), snapshot),
                input_kind: expression.input_kind().clone(),
                output_kind: expression.output_kind().clone(),
                canonical_text: expression.canonical_text().to_string(),
            }))
        }
    }
}

fn schema_index_field_path_info_from_accepted(
    path: &PersistedIndexFieldPathSnapshot,
    snapshot: &PersistedSchemaSnapshot,
) -> SchemaIndexFieldPathInfo {
    let field_name = accepted_field_name(snapshot, path.field_id())
        .or_else(|| path.path().first().map(String::as_str))
        .unwrap_or_default()
        .to_string();

    SchemaIndexFieldPathInfo {
        field_id: Some(path.field_id()),
        field_name,
        slot: accepted_slot_index(path.slot()),
        path: path.path().to_vec(),
        ty: field_type_from_persisted_kind(path.kind()),
        persisted_kind: Some(path.kind().clone()),
        nullable: path.nullable(),
    }
}

// Resolve generated nested metadata for compile-time-only schema views. Accepted
// schema views use persisted nested leaf descriptors before this fallback is
// considered.
fn resolve_nested_field_path_kind(fields: &[FieldModel], segments: &[String]) -> Option<FieldKind> {
    let (segment, rest) = segments.split_first()?;
    let field = fields
        .iter()
        .find(|field| field.name() == segment.as_str())?;

    if rest.is_empty() {
        return Some(field.kind());
    }

    resolve_nested_field_path_kind(field.nested_fields(), rest)
}

///
/// TESTS
///

#[cfg(test)]
mod tests;