icydb-core 0.200.2

//! Module: data::structural_row
//! Responsibility: canonical structural persisted-row decode helpers.
//! Does not own: typed entity reconstruction, slot layout planning, or query semantics.
//! Boundary: runtime paths use this module when they need persisted-row structure without `E`.

use crate::{
    db::{
        codec::decode_row_payload_bytes,
        data::{RawRow, decode_runtime_value_from_row_contract},
        schema::{
            AcceptedCatalogSnapshotSelection, AcceptedFieldAbsencePolicy,
            AcceptedFieldDecodeContract, AcceptedRowDecodeContract,
            AcceptedRowLayoutRuntimeContract, AcceptedSchemaSnapshot,
            OwnedAcceptedRelationEdgeContract,
        },
    },
    error::InternalError,
    model::{entity::EntityModel, field::LeafCodec},
    value::Value,
};
use std::{borrow::Cow, rc::Rc};

type SlotSpan = Option<(usize, usize)>;
type SlotSpans = Vec<SlotSpan>;
type RowFieldSpans<'a> = (Cow<'a, [u8]>, SlotSpans);
type RowSlotTableSections<'a> = (usize, usize, &'a [u8], &'a [u8]);

/// Accepted snapshot and structural row contract selected from one catalog root.
#[derive(Clone, Debug)]
pub(in crate::db) struct AcceptedStructuralRowAuthority {
    accepted_schema: AcceptedSchemaSnapshot,
    row_contract: StructuralRowContract,
}

impl AcceptedStructuralRowAuthority {
    /// Build accepted-only row authority without separating snapshot and catalog.
    pub(in crate::db) fn from_catalog_selection(
        entity_path: &'static str,
        selection: &AcceptedCatalogSnapshotSelection,
    ) -> Result<Self, InternalError> {
        let accepted_schema = selection.decode_verified()?;
        let descriptor = AcceptedRowLayoutRuntimeContract::from_accepted_schema(&accepted_schema)?;
        let row_contract = Self::catalog_backed_row_contract(entity_path, &descriptor, selection);

        Ok(Self {
            accepted_schema,
            row_contract,
        })
    }

    /// Build generated-compatible row authority from one catalog selection.
    pub(in crate::db) fn from_generated_compatible_catalog_selection(
        entity_path: &'static str,
        model: &'static EntityModel,
        selection: &AcceptedCatalogSnapshotSelection,
    ) -> Result<Self, InternalError> {
        let accepted_schema = selection.decode_verified()?;
        let (descriptor, _row_proof) =
            AcceptedRowLayoutRuntimeContract::from_generated_compatible_schema(
                &accepted_schema,
                model,
            )?;
        let row_contract = Self::catalog_backed_row_contract(entity_path, &descriptor, selection);

        Ok(Self {
            accepted_schema,
            row_contract,
        })
    }

    fn catalog_backed_row_contract(
        entity_path: &'static str,
        descriptor: &AcceptedRowLayoutRuntimeContract<'_>,
        selection: &AcceptedCatalogSnapshotSelection,
    ) -> StructuralRowContract {
        let identity = selection.identity();
        let row_decode_contract =
            descriptor.row_decode_contract_with_catalog(selection.enum_catalog().clone());
        debug_assert_eq!(
            row_decode_contract.accepted_schema_revision(),
            Some(identity.accepted_schema_revision())
        );
        debug_assert!(
            row_decode_contract.enum_catalog().is_some_and(|catalog| {
                std::ptr::eq(catalog, selection.enum_catalog().catalog())
            })
        );
        StructuralRowContract::from_accepted_decode_contract(entity_path, row_decode_contract)
    }

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

    /// Consume this authority into its still-paired accepted artifacts.
    #[must_use]
    pub(in crate::db) fn into_parts(self) -> (AcceptedSchemaSnapshot, StructuralRowContract) {
        (self.accepted_schema, self.row_contract)
    }

    /// Consume this authority when the caller only needs structural row decode.
    #[must_use]
    pub(in crate::db) fn into_row_contract(self) -> StructuralRowContract {
        self.row_contract
    }
}

///
/// StructuralRowContract
///
/// StructuralRowContract is the compact static row-shape authority used by
/// structural row readers that do not need the full semantic `EntityModel`.
/// It keeps the entity path and accepted row-decode contract required to open
/// canonical persisted rows through the data-layer decode boundary.
///

#[derive(Clone, Debug)]
pub(in crate::db) struct StructuralRowContract {
    entity_path: &'static str,
    field_count: usize,
    max_physical_slot_count: usize,
    primary_key_slot: usize,
    accepted_decode_contract: Rc<AcceptedRowDecodeContract>,
}

impl StructuralRowContract {
    /// Build an accepted structural row contract from one model proposal for tests.
    #[cfg(test)]
    pub(in crate::db) fn from_model_proposal_for_test(
        model: &'static crate::model::entity::EntityModel,
    ) -> Self {
        Self::from_accepted_decode_contract(
            model.path(),
            AcceptedRowDecodeContract::from_model_proposal_for_test(model),
        )
    }

    /// Build one structural row contract from accepted persisted schema only.
    #[must_use]
    pub(in crate::db) fn from_accepted_decode_contract(
        entity_path: &'static str,
        accepted_decode_contract: AcceptedRowDecodeContract,
    ) -> Self {
        Self {
            entity_path,
            field_count: accepted_decode_contract.required_slot_count(),
            max_physical_slot_count: accepted_decode_contract.max_physical_slot_count(),
            primary_key_slot: accepted_decode_contract.first_primary_key_slot_index(),
            accepted_decode_contract: Rc::new(accepted_decode_contract),
        }
    }

    /// Build one structural row contract from an accepted persisted schema snapshot.
    ///
    /// This is the data-layer owner for the repeated accepted-schema projection
    /// sequence after callers have already crossed schema reconciliation.
    /// Runtime row readers receive an accepted-only structural contract so they
    /// cannot fall back to generated schema metadata after acceptance.
    #[cfg(test)]
    pub(in crate::db) fn from_accepted_schema_snapshot(
        entity_path: &'static str,
        accepted_schema: &AcceptedSchemaSnapshot,
    ) -> Result<Self, InternalError> {
        let descriptor = AcceptedRowLayoutRuntimeContract::from_accepted_schema(accepted_schema)?;

        Ok(Self::from_accepted_decode_contract(
            entity_path,
            descriptor.row_decode_contract(),
        ))
    }

    /// Borrow the owning entity path for diagnostics.
    #[must_use]
    pub(in crate::db) const fn entity_path(&self) -> &'static str {
        self.entity_path
    }

    /// Return the declared structural field count.
    #[must_use]
    pub(in crate::db) const fn field_count(&self) -> usize {
        self.field_count
    }

    /// Return the maximum physical slot count this row contract can accept.
    ///
    /// Accepted contracts may allow current-format rows to carry trailing
    /// retired slots that are no longer visible through the active schema.
    #[must_use]
    pub(in crate::db) const fn max_physical_slot_count(&self) -> usize {
        self.max_physical_slot_count
    }

    /// Return the authoritative primary-key slot.
    #[must_use]
    pub(in crate::db) const fn primary_key_slot(&self) -> usize {
        self.primary_key_slot
    }

    /// Borrow ordered primary-key physical slots in key order.
    #[must_use]
    pub(in crate::db) fn primary_key_slot_indices(&self) -> &[usize] {
        self.accepted_decode_contract.primary_key_slot_indices()
    }

    pub(in crate::db) fn required_accepted_field_decode_contract(
        &self,
        slot: usize,
    ) -> Result<AcceptedFieldDecodeContract<'_>, InternalError> {
        Ok(self
            .accepted_decode_contract
            .required_field_for_slot(self.entity_path(), slot)?
            .decode_contract())
    }

    /// Borrow the catalog authority carried by this accepted row contract.
    #[must_use]
    pub(in crate::db) fn accepted_enum_catalog_handle(
        &self,
    ) -> Option<&crate::db::schema::AcceptedEnumCatalogHandle> {
        self.accepted_decode_contract.enum_catalog_handle()
    }

    /// Borrow accepted relation-edge metadata declared on this source row.
    #[must_use]
    pub(in crate::db) fn accepted_relation_edges(&self) -> &[OwnedAcceptedRelationEdgeContract] {
        self.accepted_decode_contract.relation_edges()
    }

    /// Return whether a physical slot is active in this row contract.
    ///
    /// Accepted row layouts may retain retired physical slots as allocation
    /// history. Those slots can still be present in old rows, but they are no
    /// longer active fields and must be skipped by dense validation/emission.
    #[must_use]
    pub(in crate::db) fn has_active_field_slot(&self, slot: usize) -> bool {
        self.accepted_decode_contract.field_for_slot(slot).is_some()
    }

    /// Return the leaf codec for one structural slot.
    ///
    pub(in crate::db) fn field_leaf_codec(&self, slot: usize) -> Result<LeafCodec, InternalError> {
        self.required_accepted_field_decode_contract(slot)
            .map(|field| field.leaf_codec())
    }

    /// Return the persisted field name for diagnostics at one row slot.
    pub(in crate::db) fn field_name(&self, slot: usize) -> Result<&str, InternalError> {
        self.required_accepted_field_decode_contract(slot)
            .map(|field| field.field_name())
    }

    /// Return one field's physical row slot by persisted field name.
    ///
    pub(in crate::db) fn field_slot_index_by_name(
        &self,
        field_name: &str,
    ) -> Result<usize, InternalError> {
        for slot in 0..self.field_count() {
            let Some(field) = self.accepted_decode_contract.field_for_slot(slot) else {
                continue;
            };
            if field.field_name() == field_name {
                return Ok(slot);
            }
        }

        Err(InternalError::persisted_row_declared_field_missing(
            field_name,
        ))
    }

    /// Return the missing-slot policy for one accepted physical slot.
    #[must_use]
    pub(in crate::db) fn accepted_field_absence_policy(
        &self,
        slot: usize,
    ) -> Option<AcceptedFieldAbsencePolicy> {
        let field = self.accepted_decode_contract.field_for_slot(slot)?;

        Some(field.absence_policy())
    }

    /// Materialize the logical runtime value for an absent accepted slot.
    ///
    /// Only accepted-schema row contracts may synthesize missing fields.
    /// Nullable slots materialize as `NULL`, slots with explicit database
    /// defaults materialize through the accepted default payload, and required
    /// accepted slots remain fail-closed.
    pub(in crate::db) fn missing_slot_value(&self, slot: usize) -> Result<Value, InternalError> {
        let field_name = self.field_name(slot)?;
        match self.accepted_field_absence_policy(slot) {
            Some(AcceptedFieldAbsencePolicy::NullIfMissing) => Ok(Value::Null),
            Some(AcceptedFieldAbsencePolicy::DefaultIfMissing) => {
                let field = self
                    .accepted_decode_contract
                    .required_field_for_slot(self.entity_path(), slot)?;
                let Some(default_payload) = field.default().slot_payload() else {
                    return Err(InternalError::persisted_row_declared_field_missing(
                        field_name,
                    ));
                };

                decode_runtime_value_from_row_contract(self, slot, default_payload)
            }
            Some(AcceptedFieldAbsencePolicy::Required) | None => Err(
                InternalError::persisted_row_declared_field_missing(field_name),
            ),
        }
    }

    // Validate that one physical row slot count can be read through this
    // structural contract. Rows may be short when trailing additions can
    // materialize from schema metadata, or long when trailing retired slots
    // still exist in older physical rows from the current format.
    fn validate_physical_slot_count(&self, physical_count: usize) -> Result<(), InternalError> {
        if physical_count > self.max_physical_slot_count() {
            return Err(InternalError::persisted_row_decode_corruption());
        }

        for slot in physical_count..self.field_count() {
            if !self.has_active_field_slot(slot) {
                continue;
            }
            let _ = self.missing_slot_value(slot)?;
        }

        Ok(())
    }
}

///
/// StructuralRowFieldBytes
///
/// StructuralRowFieldBytes is the top-level persisted-row field scanner for
/// slot-driven proof paths.
/// It keeps the original encoded field payload bytes and records one byte span
/// per model slot so callers can decode only the fields they actually need.
///

#[derive(Clone, Debug)]
pub(in crate::db::data) struct StructuralRowFieldBytes<'a> {
    payload: Cow<'a, [u8]>,
    spans: SlotSpans,
}

impl<'a> StructuralRowFieldBytes<'a> {
    /// Decode one raw row payload into contract slot-aligned encoded field spans.
    fn from_row_bytes_with_contract(
        row_bytes: &'a [u8],
        contract: &StructuralRowContract,
    ) -> Result<Self, StructuralRowDecodeError> {
        let payload = decode_structural_row_payload_bytes(row_bytes)?;
        let (payload, spans) = decode_row_field_spans(payload, contract)?;

        Ok(Self { payload, spans })
    }

    /// Decode one raw row into contract slot-aligned encoded field payload spans.
    pub(in crate::db::data) fn from_raw_row_with_contract(
        raw_row: &'a RawRow,
        contract: &StructuralRowContract,
    ) -> Result<Self, StructuralRowDecodeError> {
        Self::from_row_bytes_with_contract(raw_row.as_bytes(), contract)
    }

    /// Borrow one encoded persisted field payload by stable slot index.
    #[must_use]
    pub(in crate::db::data) fn field(&self, slot: usize) -> Option<&[u8]> {
        let (start, end) = self.spans.get(slot).copied().flatten()?;

        Some(&self.payload[start..end])
    }
}

///
/// SparseRequiredRowFieldBytes
///
/// SparseRequiredRowFieldBytes carries the shared payload plus just the two
/// slot spans needed by the narrow sparse required-slot decode path.
/// Executor one-slot reads use this to preserve full row-table validation
/// without allocating one field-count-sized span vector on every row.
///

#[derive(Clone, Debug)]
pub(in crate::db::data) struct SparseRequiredRowFieldBytes<'a> {
    payload: Cow<'a, [u8]>,
    required_span: Option<(usize, usize)>,
    primary_key_span: (usize, usize),
}

impl<'a> SparseRequiredRowFieldBytes<'a> {
    /// Decode one raw row into the selected and primary-key field spans needed
    /// by sparse direct slot reads.
    pub(in crate::db::data) fn from_raw_row_with_contract(
        raw_row: &'a RawRow,
        contract: &StructuralRowContract,
        required_slot: usize,
    ) -> Result<Self, StructuralRowDecodeError> {
        let payload = decode_structural_row_payload_bytes(raw_row.as_bytes())?;
        let (payload, required_span, primary_key_span) =
            decode_sparse_required_row_field_spans(payload, contract, required_slot)?;

        Ok(Self {
            payload,
            required_span,
            primary_key_span,
        })
    }

    /// Borrow the selected required field payload bytes.
    #[must_use]
    pub(in crate::db::data) fn required_field(&self) -> Option<&[u8]> {
        let (start, end) = self.required_span?;

        Some(&self.payload[start..end])
    }

    /// Borrow the primary-key field payload bytes.
    #[must_use]
    pub(in crate::db::data) fn primary_key_field(&self) -> &[u8] {
        &self.payload[self.primary_key_span.0..self.primary_key_span.1]
    }
}

///
/// StructuralRowDecodeError
///
/// StructuralRowDecodeError captures shape failures after persisted-row bytes
/// have already decoded successfully through the shared structural path.
///

#[derive(Debug)]
pub(in crate::db::data) enum StructuralRowDecodeError {
    Deserialize(InternalError),
}

impl From<InternalError> for StructuralRowDecodeError {
    fn from(err: InternalError) -> Self {
        Self::Deserialize(err)
    }
}

impl StructuralRowDecodeError {
    // Collapse the local structural decode wrapper back into the internal taxonomy.
    pub(in crate::db::data) const fn into_internal_error(self) -> InternalError {
        match self {
            Self::Deserialize(err) => err,
        }
    }

    // Build one structural row corruption error at the manual decode boundary.
    fn corruption() -> Self {
        Self::Deserialize(InternalError::persisted_row_decode_corruption())
    }
}

/// Decode one persisted row through the structural row-envelope validation path.
///
/// The only supported persisted row shape is the slot-framed payload envelope,
/// so this helper returns the validated enclosed payload bytes directly.
pub(in crate::db) fn decode_structural_row_payload(
    raw_row: &RawRow,
) -> Result<Cow<'_, [u8]>, InternalError> {
    decode_structural_row_payload_bytes(raw_row.as_bytes())
        .map_err(StructuralRowDecodeError::into_internal_error)
}

// Decode one persisted row envelope into the enclosed slot payload bytes.
fn decode_structural_row_payload_bytes(
    bytes: &[u8],
) -> Result<Cow<'_, [u8]>, StructuralRowDecodeError> {
    decode_row_payload_bytes(bytes).map_err(StructuralRowDecodeError::from)
}

// Decode the canonical slot-container header into slot-aligned payload spans.
fn decode_row_field_spans<'payload>(
    payload: Cow<'payload, [u8]>,
    contract: &StructuralRowContract,
) -> Result<RowFieldSpans<'payload>, StructuralRowDecodeError> {
    let bytes = payload.as_ref();
    let (data_start, physical_count, table, data_section) =
        decode_slot_table_sections(bytes, contract)?;
    let mut spans: SlotSpans = vec![None; contract.field_count()];

    for (slot, span) in spans.iter_mut().take(physical_count).enumerate() {
        let entry_start = slot
            .checked_mul(8)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        let entry = table
            .get(entry_start..entry_start + 8)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        let start = usize::try_from(u32::from_be_bytes([entry[0], entry[1], entry[2], entry[3]]))
            .map_err(|_| StructuralRowDecodeError::corruption())?;
        let len = usize::try_from(u32::from_be_bytes([entry[4], entry[5], entry[6], entry[7]]))
            .map_err(|_| StructuralRowDecodeError::corruption())?;
        if len == 0 {
            return Err(StructuralRowDecodeError::corruption());
        }
        let end = start
            .checked_add(len)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        if end > data_section.len() {
            return Err(StructuralRowDecodeError::corruption());
        }
        *span = Some((start, end));
    }

    let payload = match payload {
        Cow::Borrowed(bytes) => Cow::Borrowed(&bytes[data_start..]),
        Cow::Owned(bytes) => Cow::Owned(bytes[data_start..].to_vec()),
    };

    Ok((payload, spans))
}

type SparseRequiredRowFieldSpans<'a> =
    Result<(Cow<'a, [u8]>, Option<(usize, usize)>, (usize, usize)), StructuralRowDecodeError>;

// Decode the canonical slot-container header while retaining only one required
// slot span plus the primary-key span for sparse direct slot reads.
fn decode_sparse_required_row_field_spans<'payload>(
    payload: Cow<'payload, [u8]>,
    contract: &StructuralRowContract,
    required_slot: usize,
) -> SparseRequiredRowFieldSpans<'payload> {
    let bytes = payload.as_ref();
    let (data_start, physical_count, table, data_section) =
        decode_slot_table_sections(bytes, contract)?;
    let primary_key_slot = contract.primary_key_slot();
    let mut required_span = None;
    let mut primary_key_span = None;

    for slot in 0..physical_count {
        let entry_start = slot
            .checked_mul(8)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        let entry = table
            .get(entry_start..entry_start + 8)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        let start = usize::try_from(u32::from_be_bytes([entry[0], entry[1], entry[2], entry[3]]))
            .map_err(|_| StructuralRowDecodeError::corruption())?;
        let len = usize::try_from(u32::from_be_bytes([entry[4], entry[5], entry[6], entry[7]]))
            .map_err(|_| StructuralRowDecodeError::corruption())?;
        if len == 0 {
            return Err(StructuralRowDecodeError::corruption());
        }
        let end = start
            .checked_add(len)
            .ok_or_else(StructuralRowDecodeError::corruption)?;
        if end > data_section.len() {
            return Err(StructuralRowDecodeError::corruption());
        }
        if slot == required_slot {
            required_span = Some((start, end));
        }
        if slot == primary_key_slot {
            primary_key_span = Some((start, end));
        }
    }

    if required_span.is_none() {
        let _ = contract
            .missing_slot_value(required_slot)
            .map_err(StructuralRowDecodeError::from)?;
    }
    let primary_key_span = primary_key_span.ok_or_else(StructuralRowDecodeError::corruption)?;
    let payload = match payload {
        Cow::Borrowed(bytes) => Cow::Borrowed(&bytes[data_start..]),
        Cow::Owned(bytes) => Cow::Owned(bytes[data_start..].to_vec()),
    };

    Ok((payload, required_span, primary_key_span))
}

// Decode the shared slot-table header and validate that the physical row slot
// count matches the structural contract before any full or sparse slot scanner
// walks the table. This keeps accepted raw-row shape authority in one place.
fn decode_slot_table_sections<'bytes>(
    bytes: &'bytes [u8],
    contract: &StructuralRowContract,
) -> Result<RowSlotTableSections<'bytes>, StructuralRowDecodeError> {
    let field_count_bytes = bytes
        .get(..2)
        .ok_or_else(StructuralRowDecodeError::corruption)?;
    let field_count = usize::from(u16::from_be_bytes([
        field_count_bytes[0],
        field_count_bytes[1],
    ]));
    contract
        .validate_physical_slot_count(field_count)
        .map_err(StructuralRowDecodeError::from)?;
    let table_len = field_count
        .checked_mul(8)
        .ok_or_else(StructuralRowDecodeError::corruption)?;
    let data_start = 2usize
        .checked_add(table_len)
        .ok_or_else(StructuralRowDecodeError::corruption)?;
    let table = bytes
        .get(2..data_start)
        .ok_or_else(StructuralRowDecodeError::corruption)?;
    let data_section = bytes
        .get(data_start..)
        .ok_or_else(StructuralRowDecodeError::corruption)?;

    Ok((data_start, field_count, table, data_section))
}