icydb-core 0.216.2

//! Module: db::schema::inspection_plan
//! Responsibility: canonical accepted-schema projection for integrity inspection.
//! Does not own: physical traversal, inspection progress, or diagnostic rendering.
//! Boundary: binds one verified accepted entity snapshot and value catalog to
//! the row-program authority and fingerprint consumed by Quick and Deep inspection.

use crate::{
    db::{
        Db,
        codec::{
            finalize_hash_sha256, new_hash_sha256_prefixed, write_hash_str_u32, write_hash_u32,
            write_hash_u64,
        },
        data::StructuralRowContract,
        index::AcceptedIndexInspectionPlan,
        relation::{RelationConstraintProjection, ReverseRelationSourceInfo},
        schema::{
            AcceptedCatalogIdentity, AcceptedFieldKind, AcceptedRowLayoutRuntimeContract,
            AcceptedSchemaFingerprint, AcceptedSchemaSnapshot, AcceptedValueCatalogHandle,
            CompiledAcceptedRowConstraints, FieldId, FieldInsertGeneration,
        },
    },
    error::InternalError,
    traits::CanisterKind,
};
use sha2::Digest;

const ACCEPTED_INSPECTION_PLAN_FINGERPRINT_DOMAIN: &[u8] = b"icydb.accepted-inspection-plan.v1";

/// Semantic fingerprint of one accepted inspection plan.
///
/// The fingerprint binds the selected entity schema, its accepted store-local
/// value catalog, and the inspection semantics version. It is not a second
/// schema authority: the selected accepted snapshot and root fingerprints are
/// its inputs.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub(in crate::db) struct AcceptedInspectionPlanFingerprint([u8; 32]);

impl AcceptedInspectionPlanFingerprint {
    /// Return the canonical fingerprint bytes.
    #[must_use]
    pub(in crate::db) const fn to_bytes(self) -> [u8; 32] {
        self.0
    }
}

/// Bounded accepted-native input to integrity inspection.
///
/// This artifact carries the exact accepted entity snapshot, catalog authority,
/// and precompiled row program already used by write admission. Later
/// inspection phases add physical traversal contracts to this owner rather
/// than rebuilding schema meaning from generated models.
#[derive(Clone, Debug)]
pub(in crate::db) struct AcceptedInspectionPlan {
    identity: AcceptedCatalogIdentity,
    snapshot: AcceptedSchemaSnapshot,
    value_catalog: AcceptedValueCatalogHandle,
    row_contract: StructuralRowContract,
    write_constraints: CompiledAcceptedRowConstraints,
    index_inspection: AcceptedIndexInspectionPlan,
    relation_inspection: Vec<RelationConstraintProjection>,
    identity_inspection: Option<AcceptedIdentityInspection>,
    fingerprint: AcceptedInspectionPlanFingerprint,
}

/// Exact accepted Identity declaration needed by bounded physical inspection.
#[derive(Clone, Debug)]
pub(in crate::db) struct AcceptedIdentityInspection {
    field_id: FieldId,
    field_name: String,
    accepted_kind: AcceptedFieldKind,
}

impl AcceptedIdentityInspection {
    #[must_use]
    pub(in crate::db) const fn field_id(&self) -> FieldId {
        self.field_id
    }

    #[must_use]
    pub(in crate::db) const fn field_name(&self) -> &str {
        self.field_name.as_str()
    }

    #[must_use]
    pub(in crate::db) const fn accepted_kind(&self) -> &AcceptedFieldKind {
        &self.accepted_kind
    }
}

impl AcceptedInspectionPlan {
    /// Build one plan from a verified accepted selection.
    pub(in crate::db) fn compile<C: CanisterKind>(
        db: &Db<C>,
        identity: AcceptedCatalogIdentity,
        snapshot: AcceptedSchemaSnapshot,
        value_catalog: AcceptedValueCatalogHandle,
    ) -> Result<Self, InternalError> {
        let relation_identity = identity.clone();
        Self::compile_with_relation_builder(
            identity,
            snapshot,
            value_catalog,
            move |snapshot, row| {
                let source = ReverseRelationSourceInfo::new(
                    relation_identity.entity_path_handle(),
                    relation_identity.entity_tag(),
                );
                snapshot
                    .persisted_snapshot()
                    .relations()
                    .iter()
                    .map(|edge| {
                        RelationConstraintProjection::new_active(
                            db,
                            source.clone(),
                            snapshot.persisted_snapshot(),
                            row,
                            edge,
                        )
                    })
                    .collect()
            },
        )
    }

    #[cfg(test)]
    /// Compile a relation-free plan without constructing a runtime database.
    ///
    /// Relation-bearing fixtures must use [`Self::compile`] so tests cannot
    /// create a plan that omits accepted relation authority.
    pub(in crate::db) fn compile_relation_free_for_tests(
        identity: AcceptedCatalogIdentity,
        snapshot: AcceptedSchemaSnapshot,
        value_catalog: AcceptedValueCatalogHandle,
    ) -> Result<Self, InternalError> {
        Self::compile_with_relation_builder(identity, snapshot, value_catalog, |snapshot, _row| {
            if !snapshot.persisted_snapshot().relations().is_empty() {
                return Err(InternalError::store_invariant());
            }
            Ok(Vec::new())
        })
    }

    fn compile_with_relation_builder(
        identity: AcceptedCatalogIdentity,
        snapshot: AcceptedSchemaSnapshot,
        value_catalog: AcceptedValueCatalogHandle,
        build_relations: impl FnOnce(
            &AcceptedSchemaSnapshot,
            &StructuralRowContract,
        )
            -> Result<Vec<RelationConstraintProjection>, InternalError>,
    ) -> Result<Self, InternalError> {
        if value_catalog.revision() != identity.accepted_schema_revision() {
            return Err(InternalError::store_invariant());
        }

        let accepted_schema_fingerprint = identity.accepted_schema_fingerprint();
        let row_layout = AcceptedRowLayoutRuntimeContract::from_accepted_schema(&snapshot)?;
        let row_contract = StructuralRowContract::from_accepted_decode_contract(
            identity.entity_path_handle(),
            row_layout.row_decode_contract(value_catalog.clone()),
        );
        let write_constraints = CompiledAcceptedRowConstraints::compile(
            &snapshot,
            &value_catalog,
            accepted_schema_fingerprint,
        )
        .map_err(|_| InternalError::accepted_row_constraint_program_corrupt())?;
        let index_inspection =
            AcceptedIndexInspectionPlan::compile(&snapshot, value_catalog.clone(), &row_contract)?;
        let relation_inspection = build_relations(&snapshot, &row_contract)?;
        let identity_inspection = accepted_identity_inspection(&snapshot)?;
        let fingerprint = accepted_inspection_plan_fingerprint(
            &identity,
            value_catalog.authority().fingerprint(),
        );

        Ok(Self {
            identity,
            snapshot,
            value_catalog,
            row_contract,
            write_constraints,
            index_inspection,
            relation_inspection,
            identity_inspection,
            fingerprint,
        })
    }

    /// Return the selected accepted catalog identity.
    #[must_use]
    pub(in crate::db) fn identity(&self) -> AcceptedCatalogIdentity {
        self.identity.clone()
    }

    /// Borrow the selected accepted catalog identity without cloning its path.
    #[must_use]
    pub(in crate::db) const fn identity_ref(&self) -> &AcceptedCatalogIdentity {
        &self.identity
    }

    /// Borrow the selected accepted entity snapshot.
    #[must_use]
    pub(in crate::db) const fn snapshot(&self) -> &AcceptedSchemaSnapshot {
        &self.snapshot
    }

    /// Borrow the selected store-local value catalog.
    #[must_use]
    pub(in crate::db) const fn value_catalog(&self) -> &AcceptedValueCatalogHandle {
        &self.value_catalog
    }

    /// Borrow the accepted current/historical structural row contract.
    #[must_use]
    pub(in crate::db) const fn row_contract(&self) -> &StructuralRowContract {
        &self.row_contract
    }

    /// Borrow the write-admission row program for this accepted identity.
    #[must_use]
    pub(in crate::db) const fn write_constraints(&self) -> &CompiledAcceptedRowConstraints {
        &self.write_constraints
    }

    /// Borrow precompiled active forward-index witness authority.
    #[must_use]
    pub(in crate::db) const fn index_inspection(&self) -> &AcceptedIndexInspectionPlan {
        &self.index_inspection
    }

    /// Borrow precompiled active source-owned relation witness authority.
    #[must_use]
    pub(in crate::db) const fn relation_inspection(&self) -> &[RelationConstraintProjection] {
        self.relation_inspection.as_slice()
    }

    /// Borrow the exact accepted Identity declaration, when present.
    #[must_use]
    pub(in crate::db) const fn identity_inspection(&self) -> Option<&AcceptedIdentityInspection> {
        self.identity_inspection.as_ref()
    }

    /// Return the fingerprint of the complete accepted inspection projection.
    #[must_use]
    pub(in crate::db) const fn fingerprint(&self) -> AcceptedInspectionPlanFingerprint {
        self.fingerprint
    }
}

fn accepted_identity_inspection(
    snapshot: &AcceptedSchemaSnapshot,
) -> Result<Option<AcceptedIdentityInspection>, InternalError> {
    let mut identity = None;
    for field in snapshot
        .persisted_snapshot()
        .fields()
        .iter()
        .filter(|field| {
            field.write_policy().insert_generation() == Some(FieldInsertGeneration::Identity)
        })
    {
        if identity.is_some() {
            return Err(InternalError::identity_state_corruption());
        }
        identity = Some(AcceptedIdentityInspection {
            field_id: field.id(),
            field_name: field.name().to_string(),
            accepted_kind: field.kind().clone(),
        });
    }
    Ok(identity)
}

fn accepted_inspection_plan_fingerprint(
    identity: &AcceptedCatalogIdentity,
    accepted_root_fingerprint: AcceptedSchemaFingerprint,
) -> AcceptedInspectionPlanFingerprint {
    let mut hasher = new_hash_sha256_prefixed(ACCEPTED_INSPECTION_PLAN_FINGERPRINT_DOMAIN);
    write_hash_u64(&mut hasher, identity.entity_tag().value());
    write_hash_str_u32(&mut hasher, identity.entity_path());
    write_hash_str_u32(&mut hasher, identity.store_path());
    write_hash_u64(&mut hasher, identity.accepted_schema_revision().get());
    write_hash_u32(&mut hasher, identity.accepted_schema_version().get());
    hasher.update([identity.fingerprint_method_version()]);
    hasher.update(identity.accepted_schema_fingerprint());
    hasher.update(accepted_root_fingerprint.as_bytes());

    AcceptedInspectionPlanFingerprint(finalize_hash_sha256(hasher))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        db::schema::{FieldStorageDecode, LeafCodec, ScalarCodec},
        db::{
            commit::CommitSchemaFingerprint,
            schema::{
                AcceptedCompositeCatalog, AcceptedFieldKind, AcceptedSchemaRevision, FieldId,
                PersistedFieldSnapshot, PersistedSchemaSnapshot, SchemaFieldSlot,
                SchemaInsertDefault, SchemaRowLayout, SchemaVersion,
                empty_accepted_enum_catalog_for_tests,
            },
        },
        types::EntityTag,
    };

    fn identity(
        revision: AcceptedSchemaRevision,
        fingerprint: CommitSchemaFingerprint,
    ) -> AcceptedCatalogIdentity {
        AcceptedCatalogIdentity::new(
            EntityTag::new(17),
            "tests::InspectionEntity",
            "tests::InspectionStore",
            revision,
            SchemaVersion::initial(),
            fingerprint,
        )
    }

    fn value_catalog(revision: AcceptedSchemaRevision) -> AcceptedValueCatalogHandle {
        AcceptedValueCatalogHandle::new_for_tests(
            empty_accepted_enum_catalog_for_tests(),
            AcceptedCompositeCatalog::empty(),
            revision,
        )
    }

    fn snapshot() -> AcceptedSchemaSnapshot {
        AcceptedSchemaSnapshot::new(PersistedSchemaSnapshot::new(
            SchemaVersion::initial(),
            "tests::InspectionEntity".to_string(),
            "InspectionEntity".to_string(),
            FieldId::new(1),
            SchemaRowLayout::initial(vec![(FieldId::new(1), SchemaFieldSlot::new(0))]),
            vec![PersistedFieldSnapshot::new_initial(
                FieldId::new(1),
                "id".to_string(),
                SchemaFieldSlot::new(0),
                AcceptedFieldKind::Nat64,
                Vec::new(),
                false,
                SchemaInsertDefault::None,
                FieldStorageDecode::ByKind,
                LeafCodec::Scalar(ScalarCodec::Nat64),
            )],
        ))
    }

    #[test]
    fn accepted_inspection_plan_compiles_the_write_admission_program_once() {
        let revision = AcceptedSchemaRevision::INITIAL;
        let identity = identity(revision, [0x11; 16]);

        let plan = AcceptedInspectionPlan::compile_relation_free_for_tests(
            identity.clone(),
            snapshot(),
            value_catalog(revision),
        )
        .expect("verified accepted inputs should compile one inspection plan");

        assert_eq!(plan.identity(), identity);
        assert!(!plan.write_constraints().is_empty());
        assert_eq!(plan.write_constraints().required_slots(), &[0]);
        assert_eq!(plan.write_constraints().integrity_constraint_count(), 0);
        assert_ne!(plan.fingerprint().to_bytes(), [0; 32]);
    }

    #[test]
    fn accepted_inspection_plan_fingerprint_binds_schema_and_root_identity() {
        let revision = AcceptedSchemaRevision::INITIAL;
        let baseline = accepted_inspection_plan_fingerprint(&identity(revision, [0x11; 16]), {
            AcceptedSchemaFingerprint::new([0x22; 32])
        });

        assert_eq!(
            baseline,
            accepted_inspection_plan_fingerprint(&identity(revision, [0x11; 16]), {
                AcceptedSchemaFingerprint::new([0x22; 32])
            }),
        );
        assert_ne!(
            baseline,
            accepted_inspection_plan_fingerprint(&identity(revision, [0x33; 16]), {
                AcceptedSchemaFingerprint::new([0x22; 32])
            }),
        );
        assert_ne!(
            baseline,
            accepted_inspection_plan_fingerprint(&identity(revision, [0x11; 16]), {
                AcceptedSchemaFingerprint::new([0x44; 32])
            }),
        );
    }

    #[test]
    fn accepted_inspection_plan_rejects_mismatched_catalog_revision() {
        let error = AcceptedInspectionPlan::compile_relation_free_for_tests(
            identity(AcceptedSchemaRevision::INITIAL, [0x11; 16]),
            snapshot(),
            value_catalog(AcceptedSchemaRevision::new(2)),
        )
        .expect_err("a plan must not combine different accepted revisions");

        assert_eq!(
            error.diagnostic_code(),
            icydb_diagnostic_code::DiagnosticCode::StoreInvariantViolation,
        );
    }
}