icydb-core 0.221.0

//! Module: db::session::accepted_schema
//! Responsibility: accepted-schema runtime-root publication and session lookup.
//! Does not own: schema reconciliation policy, query planning, or mutation staging.
//! Boundary: captures every registered accepted store root and publishes one
//! immutable database-wide runtime authority for query, SQL, and write adapters.

use crate::{
    db::{
        DbSession,
        commit::{CommitSchemaFingerprint, database_incarnation_id},
        executor::EntityAuthority,
        registry::StoreHandle,
        runtime_entity_catalog::AcceptedRuntimeEntity,
        schema::{
            AcceptedCatalogIdentity, AcceptedEnumCatalog, AcceptedInspectionPlan,
            AcceptedSchemaAuthority, AcceptedSchemaRevision, AcceptedSchemaRuntimeRootIdentity,
            AcceptedSchemaRuntimeStoreRoot, AcceptedSchemaSnapshot, AcceptedValueCatalogHandle,
            CompiledAcceptedRowConstraints, SchemaInfo, SchemaStore, SchemaVersion,
            enum_catalog::AcceptedSchemaRootSelection,
        },
    },
    error::InternalError,
    traits::CanisterKind,
};
#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
use std::cell::Cell;
use std::{cell::RefCell, collections::HashMap, rc::Rc, sync::Arc};

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub(in crate::db) struct AcceptedSchemaRuntimeBuildCounts {
    pub root_identity_builds: u64,
    pub root_publications: u64,
    pub entity_compilations: u64,
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
thread_local! {
    static ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS: Cell<AcceptedSchemaRuntimeBuildCounts> =
        const { Cell::new(AcceptedSchemaRuntimeBuildCounts {
            root_identity_builds: 0,
            root_publications: 0,
            entity_compilations: 0,
        }) };
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
fn record_accepted_schema_entity_runtime_compilation() {
    ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS.with(|cell| {
        let mut counts = cell.get();
        counts.entity_compilations = counts.entity_compilations.saturating_add(1);
        cell.set(counts);
    });
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
fn record_accepted_schema_runtime_root_identity_build() {
    ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS.with(|cell| {
        let mut counts = cell.get();
        counts.root_identity_builds = counts.root_identity_builds.saturating_add(1);
        cell.set(counts);
    });
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
fn record_accepted_schema_runtime_root_publication() {
    ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS.with(|cell| {
        let mut counts = cell.get();
        counts.root_publications = counts.root_publications.saturating_add(1);
        cell.set(counts);
    });
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
pub(in crate::db) fn reset_accepted_schema_runtime_build_counts_for_tests() {
    ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS
        .with(|counts| counts.set(AcceptedSchemaRuntimeBuildCounts::default()));
}

#[cfg(all(test, feature = "sql", feature = "diagnostics"))]
#[must_use]
pub(in crate::db) fn accepted_schema_runtime_build_counts_for_tests()
-> AcceptedSchemaRuntimeBuildCounts {
    ACCEPTED_SCHEMA_RUNTIME_BUILD_COUNTS.with(Cell::get)
}

///
/// AcceptedSchemaEntityRuntime
///
/// Immutable per-entity runtime state compiled exactly once for one accepted
/// database-wide root. It owns all derived query and row-decode authority used
/// by session execution.
///

#[derive(Debug)]
struct AcceptedSchemaEntityRuntime {
    inspection_plan: AcceptedInspectionPlan,
    schema_info: Arc<SchemaInfo>,
    authority: EntityAuthority,
}

impl AcceptedSchemaEntityRuntime {
    fn compile<C: CanisterKind>(
        db: &crate::db::Db<C>,
        root_identity: AcceptedSchemaRuntimeRootIdentity,
        runtime_entity: AcceptedRuntimeEntity,
        store: StoreHandle,
    ) -> Result<Self, AcceptedInspectionPlanLoadError> {
        let selection = store
            .with_schema(|schema_store| {
                schema_store.current_accepted_catalog_selection(
                    runtime_entity.entity_tag(),
                    runtime_entity.entity_path(),
                    runtime_entity.store_path(),
                )
            })
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?
            .ok_or_else(|| {
                AcceptedInspectionPlanLoadError::Unselected(InternalError::store_corruption())
            })?;
        let identity = selection.identity();
        let snapshot = selection.decode_verified().map_err(|error| {
            AcceptedInspectionPlanLoadError::Selected {
                identity: identity.clone(),
                error,
            }
        })?;
        let inspection_plan = AcceptedInspectionPlan::compile(
            db,
            identity.clone(),
            snapshot,
            selection.value_catalog_handle().clone(),
        )
        .map_err(|error| AcceptedInspectionPlanLoadError::Selected {
            identity: identity.clone(),
            error,
        })?;
        let schema_info = Arc::new(SchemaInfo::from_accepted_snapshot_and_catalog(
            inspection_plan.snapshot(),
            inspection_plan.value_catalog().clone(),
            true,
        ));
        debug_assert!(std::ptr::eq(
            schema_info.enum_catalog(),
            inspection_plan.value_catalog().enum_catalog(),
        ));
        let authority = EntityAuthority::from_accepted_runtime_contracts(
            identity.entity_path_handle(),
            identity.entity_tag(),
            identity.store_path(),
            inspection_plan.row_contract().clone(),
            schema_info.clone(),
            identity.accepted_schema_fingerprint(),
            root_identity,
        );

        let runtime = Self {
            inspection_plan,
            schema_info,
            authority,
        };
        #[cfg(all(test, feature = "sql", feature = "diagnostics"))]
        record_accepted_schema_entity_runtime_compilation();

        Ok(runtime)
    }
}

///
/// AcceptedSchemaRuntimeRoot
///
/// One atomically published runtime view of every accepted entity authority in
/// a database incarnation. Store-root facts are retained to revalidate a warm
/// root without serializing or hashing accepted entity snapshots.
///

#[derive(Debug)]
struct AcceptedSchemaRuntimeRoot {
    identity: AcceptedSchemaRuntimeRootIdentity,
    store_roots: Vec<AcceptedSchemaRuntimeStoreRoot>,
    entities: Vec<Rc<AcceptedSchemaEntityRuntime>>,
    entities_by_path: HashMap<Rc<str>, Rc<AcceptedSchemaEntityRuntime>>,
    entities_by_name: HashMap<Rc<str>, Rc<AcceptedSchemaEntityRuntime>>,
}

impl AcceptedSchemaRuntimeRoot {
    fn compile<C: CanisterKind>(
        db: &crate::db::Db<C>,
        identity: AcceptedSchemaRuntimeRootIdentity,
        store_roots: Vec<AcceptedSchemaRuntimeStoreRoot>,
    ) -> Result<Self, AcceptedInspectionPlanLoadError> {
        let runtime_entities = db
            .accepted_runtime_entities()
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        let mut entities = Vec::with_capacity(runtime_entities.len());
        let mut entities_by_path = HashMap::with_capacity(runtime_entities.len());
        let mut entities_by_name = HashMap::with_capacity(runtime_entities.len());

        for runtime_entity in runtime_entities {
            let store = runtime_entity
                .store(db)
                .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
            let entity = Rc::new(AcceptedSchemaEntityRuntime::compile(
                db,
                identity,
                runtime_entity,
                store,
            )?);
            let entity_path = entity.inspection_plan.identity().entity_path_handle();
            let entity_name: Rc<str> = Rc::from(entity.inspection_plan.snapshot().entity_name());
            if entities_by_path
                .insert(entity_path, entity.clone())
                .is_some()
                || entities_by_name
                    .insert(entity_name, entity.clone())
                    .is_some()
            {
                return Err(AcceptedInspectionPlanLoadError::Unselected(
                    InternalError::store_corruption(),
                ));
            }
            entities.push(entity);
        }

        let root = Self {
            identity,
            store_roots,
            entities,
            entities_by_path,
            entities_by_name,
        };
        #[cfg(all(test, feature = "sql", feature = "diagnostics"))]
        record_accepted_schema_runtime_root_publication();

        Ok(root)
    }

    #[must_use]
    const fn identity(&self) -> AcceptedSchemaRuntimeRootIdentity {
        self.identity
    }

    #[must_use]
    fn matches(
        &self,
        database_incarnation: crate::db::DatabaseIncarnationId,
        store_roots: &[AcceptedSchemaRuntimeStoreRoot],
    ) -> bool {
        self.identity.database_incarnation() == database_incarnation
            && self.store_roots == store_roots
    }

    fn entity_for_runtime_entity(
        &self,
        runtime_entity: &AcceptedRuntimeEntity,
    ) -> Result<Rc<AcceptedSchemaEntityRuntime>, InternalError> {
        let entity = self
            .entities_by_path
            .get(runtime_entity.entity_path())
            .cloned()
            .ok_or_else(InternalError::store_corruption)?;
        let identity = entity.inspection_plan.identity_ref();
        if identity.entity_tag() != runtime_entity.entity_tag()
            || identity.store_path() != runtime_entity.store_path()
        {
            return Err(InternalError::store_corruption());
        }

        Ok(entity)
    }

    #[must_use]
    fn entity_for_path(&self, entity_path: &str) -> Option<Rc<AcceptedSchemaEntityRuntime>> {
        self.entities_by_path.get(entity_path).cloned()
    }

    #[must_use]
    fn entity_for_name(&self, entity_name: &str) -> Option<Rc<AcceptedSchemaEntityRuntime>> {
        self.entities_by_name.get(entity_name).cloned()
    }

    #[must_use]
    fn first_entity(&self) -> Option<Rc<AcceptedSchemaEntityRuntime>> {
        self.entities.first().cloned()
    }
}

///
/// AcceptedSchemaCatalogContext
///
/// One entity projection borrowed from a captured database-wide accepted
/// runtime root. Cloning the context retains that exact root publication.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct AcceptedSchemaCatalogContext {
    root: Rc<AcceptedSchemaRuntimeRoot>,
    entity: Rc<AcceptedSchemaEntityRuntime>,
}

impl AcceptedSchemaCatalogContext {
    const fn new(
        root: Rc<AcceptedSchemaRuntimeRoot>,
        entity: Rc<AcceptedSchemaEntityRuntime>,
    ) -> Self {
        Self { root, entity }
    }

    #[must_use]
    pub(in crate::db) fn snapshot(&self) -> &AcceptedSchemaSnapshot {
        self.entity.inspection_plan.snapshot()
    }

    #[must_use]
    pub(in crate::db) fn enum_catalog(&self) -> &AcceptedEnumCatalog {
        self.entity.inspection_plan.value_catalog().enum_catalog()
    }

    #[must_use]
    pub(in crate::db) fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
        self.entity.inspection_plan.value_catalog()
    }

    #[must_use]
    pub(in crate::db) fn schema_version(&self) -> SchemaVersion {
        self.entity
            .inspection_plan
            .identity_ref()
            .accepted_schema_version()
    }

    #[must_use]
    pub(in crate::db) fn revision(&self) -> AcceptedSchemaRevision {
        self.entity
            .inspection_plan
            .identity_ref()
            .accepted_schema_revision()
    }

    #[must_use]
    pub(in crate::db) fn fingerprint(&self) -> CommitSchemaFingerprint {
        self.entity
            .inspection_plan
            .identity_ref()
            .accepted_schema_fingerprint()
    }

    /// Return the database-wide root identity captured by this context.
    #[must_use]
    pub(in crate::db) fn runtime_root_identity(&self) -> AcceptedSchemaRuntimeRootIdentity {
        self.root.identity()
    }

    /// Borrow the accepted row-constraint program compiled for this fingerprint.
    #[must_use]
    pub(in crate::db) fn accepted_row_constraints(&self) -> &CompiledAcceptedRowConstraints {
        self.entity.inspection_plan.write_constraints()
    }

    /// Borrow the canonical accepted inspection projection.
    #[must_use]
    pub(in crate::db) fn inspection_plan(&self) -> &AcceptedInspectionPlan {
        &self.entity.inspection_plan
    }

    #[must_use]
    pub(in crate::db) fn fingerprint_method_version(&self) -> u8 {
        self.entity
            .inspection_plan
            .identity_ref()
            .fingerprint_method_version()
    }

    #[must_use]
    pub(in crate::db) fn identity(&self) -> AcceptedCatalogIdentity {
        self.entity.inspection_plan.identity()
    }

    /// Clone executor authority from this immutable accepted entity runtime.
    #[must_use]
    pub(in crate::db) fn accepted_entity_authority(&self) -> EntityAuthority {
        self.entity.authority.clone()
    }

    #[must_use]
    pub(in crate::db) fn accepted_or_provided_entity_authority(
        &self,
        accepted_authority: Option<&EntityAuthority>,
    ) -> EntityAuthority {
        match accepted_authority {
            Some(authority) => authority.clone(),
            None => self.accepted_entity_authority(),
        }
    }

    /// Borrow schema metadata compiled once with the accepted runtime root.
    #[must_use]
    pub(in crate::db) fn accepted_schema_info(&self) -> &SchemaInfo {
        self.entity.schema_info.as_ref()
    }
}

///
/// AcceptedInspectionPlanLoadError
///
/// Distinguishes failure before entity selection from failure compiling one
/// selected accepted entity so integrity callers can retain entity identity.
///

pub(in crate::db::session) enum AcceptedInspectionPlanLoadError {
    Unselected(InternalError),
    Selected {
        identity: AcceptedCatalogIdentity,
        error: InternalError,
    },
}

impl AcceptedInspectionPlanLoadError {
    pub(in crate::db::session) fn into_internal(self) -> InternalError {
        match self {
            Self::Unselected(error) | Self::Selected { error, .. } => error,
        }
    }
}

thread_local! {
    // Each registry owns one database-wide accepted runtime root. A cache hit
    // revalidates only the compact store-root records and never serializes or
    // hashes an accepted entity snapshot.
    static ACCEPTED_SCHEMA_RUNTIME_ROOTS: RefCell<HashMap<usize, Rc<AcceptedSchemaRuntimeRoot>>> =
        RefCell::new(HashMap::default());
}

impl<C: CanisterKind> DbSession<C> {
    fn capture_accepted_runtime_store_roots(
        &self,
    ) -> Result<Vec<AcceptedSchemaRuntimeStoreRoot>, InternalError> {
        let mut stores = self
            .db
            .with_store_registry(|registry| registry.iter().collect::<Vec<_>>());
        stores.sort_unstable_by_key(|(store_path, _)| *store_path);
        stores
            .into_iter()
            .map(|(store_path, store)| {
                let root = store
                    .with_schema(SchemaStore::current_accepted_schema_root)?
                    .map(AcceptedSchemaRootSelection::root);
                Ok(AcceptedSchemaRuntimeStoreRoot::new(store_path, root))
            })
            .collect()
    }

    fn accepted_schema_runtime_root(
        &self,
    ) -> Result<Rc<AcceptedSchemaRuntimeRoot>, AcceptedInspectionPlanLoadError> {
        self.db
            .ensure_recovered_state()
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        let database_incarnation =
            database_incarnation_id().map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        let store_roots = self
            .capture_accepted_runtime_store_roots()
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        let scope_id = self.db.cache_scope_id();
        let cached = ACCEPTED_SCHEMA_RUNTIME_ROOTS.with(|roots| {
            roots
                .borrow()
                .get(&scope_id)
                .filter(|root| root.matches(database_incarnation, store_roots.as_slice()))
                .cloned()
        });
        if let Some(root) = cached {
            return Ok(root);
        }

        #[cfg(all(test, feature = "sql", feature = "diagnostics"))]
        record_accepted_schema_runtime_root_identity_build();
        let identity = AcceptedSchemaRuntimeRootIdentity::from_store_roots(
            database_incarnation,
            store_roots.as_slice(),
        )
        .map_err(AcceptedInspectionPlanLoadError::Unselected)?;

        let root = Rc::new(AcceptedSchemaRuntimeRoot::compile(
            &self.db,
            identity,
            store_roots.clone(),
        )?);
        let current_incarnation =
            database_incarnation_id().map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        let current_store_roots = self
            .capture_accepted_runtime_store_roots()
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        if current_incarnation != database_incarnation || current_store_roots != store_roots {
            return Err(AcceptedInspectionPlanLoadError::Unselected(
                InternalError::store_invariant(),
            ));
        }

        ACCEPTED_SCHEMA_RUNTIME_ROOTS.with(|roots| {
            roots.borrow_mut().insert(scope_id, root.clone());
        });

        Ok(root)
    }

    pub(in crate::db::session) fn accepted_schema_catalog_context_for_runtime_entity(
        &self,
        runtime_entity: AcceptedRuntimeEntity,
        store: StoreHandle,
    ) -> Result<AcceptedSchemaCatalogContext, InternalError> {
        let expected_store = runtime_entity.store(&self.db)?;
        if !std::ptr::eq(store.schema_store(), expected_store.schema_store()) {
            return Err(InternalError::store_invariant());
        }
        let root = self
            .accepted_schema_runtime_root()
            .map_err(AcceptedInspectionPlanLoadError::into_internal)?;
        let entity = root.entity_for_runtime_entity(&runtime_entity)?;

        Ok(AcceptedSchemaCatalogContext::new(root, entity))
    }

    /// Resolve one accepted catalog through its immutable authored source key.
    pub(in crate::db::session) fn accepted_schema_catalog_context_for_entity_source_key(
        &self,
        entity_source: &str,
    ) -> Result<AcceptedSchemaCatalogContext, InternalError> {
        self.find_accepted_schema_catalog_context_for_entity_source_key(entity_source)?
            .ok_or_else(|| InternalError::unsupported_entity_path(entity_source))
    }

    /// Find one accepted catalog through immutable source identity.
    pub(in crate::db::session) fn find_accepted_schema_catalog_context_for_entity_source_key(
        &self,
        entity_source: &str,
    ) -> Result<Option<AcceptedSchemaCatalogContext>, InternalError> {
        let root = self
            .accepted_schema_runtime_root()
            .map_err(AcceptedInspectionPlanLoadError::into_internal)?;

        Ok(root
            .entity_for_path(entity_source)
            .map(|entity| AcceptedSchemaCatalogContext::new(root, entity)))
    }

    /// Resolve one accepted catalog by its editable SQL/display entity name.
    pub(in crate::db::session) fn accepted_schema_catalog_context_for_entity_name(
        &self,
        entity_name: Option<&str>,
    ) -> Result<AcceptedSchemaCatalogContext, InternalError> {
        let root = self
            .accepted_schema_runtime_root()
            .map_err(AcceptedInspectionPlanLoadError::into_internal)?;
        let entity = match entity_name {
            Some(entity_name) => root.entity_for_name(entity_name),
            None => root.first_entity(),
        }
        .ok_or_else(|| InternalError::unsupported_entity_path(entity_name))?;

        Ok(AcceptedSchemaCatalogContext::new(root, entity))
    }

    /// Resolve an exact accepted SQL/display entity name from one root.
    pub(in crate::db::session) fn find_accepted_schema_catalog_context_for_entity_name(
        &self,
        entity_name: &str,
    ) -> Result<Option<AcceptedSchemaCatalogContext>, InternalError> {
        let root = self
            .accepted_schema_runtime_root()
            .map_err(AcceptedInspectionPlanLoadError::into_internal)?;

        Ok(root
            .entity_for_name(entity_name)
            .map(|entity| AcceptedSchemaCatalogContext::new(root, entity)))
    }

    pub(in crate::db::session) fn accepted_inspection_plan_for_runtime_entity(
        &self,
        runtime_entity: AcceptedRuntimeEntity,
        store: StoreHandle,
    ) -> Result<AcceptedInspectionPlan, AcceptedInspectionPlanLoadError> {
        let expected_store = runtime_entity
            .store(&self.db)
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;
        if !std::ptr::eq(store.schema_store(), expected_store.schema_store()) {
            return Err(AcceptedInspectionPlanLoadError::Unselected(
                InternalError::store_invariant(),
            ));
        }
        let root = self.accepted_schema_runtime_root()?;
        let entity = root
            .entity_for_runtime_entity(&runtime_entity)
            .map_err(AcceptedInspectionPlanLoadError::Unselected)?;

        Ok(entity.inspection_plan.clone())
    }

    /// Verify accepted authority for a schema-resolved structural operation.
    pub(in crate::db::session) fn ensure_accepted_schema_authority_is_current_for_store_path(
        &self,
        store_path: &'static str,
        expected: &AcceptedSchemaAuthority,
    ) -> Result<(), InternalError> {
        let store = self.db.recovered_store(store_path)?;
        if store.with_schema(|schema_store| {
            schema_store.current_accepted_schema_authority_matches(expected)
        })? {
            return Ok(());
        }

        let current_revision = store.with_schema(SchemaStore::current_accepted_schema_revision)?;

        Err(InternalError::query_stale_accepted_schema_revision(
            expected.revision().get(),
            current_revision.map(AcceptedSchemaRevision::get),
        ))
    }

    /// Drop the complete cached root after schema publication by this session.
    pub(in crate::db::session) fn invalidate_accepted_schema_runtime_root(&self) {
        let scope_id = self.db.cache_scope_id();
        ACCEPTED_SCHEMA_RUNTIME_ROOTS.with(|roots| {
            roots.borrow_mut().remove(&scope_id);
        });
    }
}