icydb-core 0.67.6

//! Module: executor::terminal::row_decode
//! Responsibility: structural scalar row decode from persisted bytes into kernel rows.
//! Does not own: typed response reconstruction or access-path iteration policy.
//! Boundary: scalar runtime row production consumes this structural decode contract.

#[cfg(test)]
use crate::model::field::EnumVariantModel;
#[cfg(test)]
use crate::types::Ulid;
use crate::{
    db::{
        data::{DataRow, RawRow, StorageKey, StructuralSlotReader},
        executor::terminal::page::KernelRow,
    },
    error::InternalError,
    model::entity::{EntityModel, resolve_primary_key_slot},
    value::Value,
};

///
/// RowLayout
///
/// RowLayout is the structural scalar row-decode plan built once at the typed
/// boundary.
/// It captures stable field ordering, persisted field-name lookup keys, and
/// primary-key slot metadata so row production no longer needs typed entity
/// materialization.
///

#[derive(Clone, Debug)]
pub(in crate::db::executor) struct RowLayout {
    model: &'static EntityModel,
    field_count: usize,
    primary_key_slot: Option<usize>,
}

impl RowLayout {
    /// Build one structural row layout from model metadata.
    #[must_use]
    pub(in crate::db::executor) fn from_model(model: &'static EntityModel) -> Self {
        Self {
            model,
            field_count: model.fields().len(),
            primary_key_slot: resolve_primary_key_slot(model),
        }
    }
}

///
/// RowDecoder
///
/// RowDecoder is the named structural decode contract for scalar row
/// production.
/// The scalar runtime owns this decoder and feeds it raw persisted rows plus a
/// precomputed `RowLayout`, keeping typed entity reconstruction out of the hot
/// execution loop.
///

#[derive(Clone, Copy, Debug)]
pub(in crate::db::executor) struct RowDecoder {
    decode: fn(&RowLayout, DataRow) -> Result<KernelRow, InternalError>,
    decode_slots: RowDecodeSlotsFn,
}

type RowDecodeSlotsFn =
    fn(&RowLayout, StorageKey, &RawRow) -> Result<Vec<Option<Value>>, InternalError>;

impl RowDecoder {
    /// Build the canonical structural row decoder used by scalar execution.
    #[must_use]
    pub(in crate::db::executor) const fn structural() -> Self {
        Self {
            decode: decode_kernel_row_structural,
            decode_slots: decode_structural_slots,
        }
    }

    /// Decode one persisted row into one structural kernel row.
    pub(in crate::db::executor) fn decode(
        self,
        layout: &RowLayout,
        data_row: DataRow,
    ) -> Result<KernelRow, InternalError> {
        (self.decode)(layout, data_row)
    }

    /// Decode one persisted row into slot-indexed structural values without
    /// constructing one full kernel-row envelope.
    pub(in crate::db::executor) fn decode_slots(
        self,
        layout: &RowLayout,
        expected_key: StorageKey,
        row: &RawRow,
    ) -> Result<Vec<Option<Value>>, InternalError> {
        (self.decode_slots)(layout, expected_key, row)
    }
}

// Decode one persisted data row into one structural kernel row using the
// precomputed slot layout and structural field decoders only.
fn decode_kernel_row_structural(
    layout: &RowLayout,
    data_row: DataRow,
) -> Result<KernelRow, InternalError> {
    let slots = decode_structural_slots(layout, data_row.0.storage_key(), &data_row.1)?;

    Ok(KernelRow::new(data_row, slots))
}

// Decode one persisted row directly into slot-indexed structural values while
// still validating the primary-key slot against storage identity.
fn decode_structural_slots(
    layout: &RowLayout,
    expected_key: StorageKey,
    row: &RawRow,
) -> Result<Vec<Option<Value>>, InternalError> {
    // Phase 1: reuse the canonical structural slot cache that `from_raw_row`
    // already forced through full declared-slot decode.
    let slots = decode_row_fields(row, layout.model)?.into_decoded_values()?;
    debug_assert_eq!(slots.len(), layout.field_count);

    // Phase 2: verify the decoded primary-key value still matches storage identity.
    validate_primary_key_slot(layout, expected_key, slots.as_slice())?;

    Ok(slots)
}

// Decode the persisted row envelope into slot-aligned encoded field payload spans.
fn decode_row_fields<'a>(
    row: &'a RawRow,
    model: &'static EntityModel,
) -> Result<StructuralSlotReader<'a>, InternalError> {
    StructuralSlotReader::from_raw_row(row, model)
}

// Validate the decoded primary-key slot against the authoritative data-key suffix.
fn validate_primary_key_slot(
    layout: &RowLayout,
    expected_key: StorageKey,
    slots: &[Option<Value>],
) -> Result<(), InternalError> {
    let Some(primary_key_slot) = layout.primary_key_slot else {
        return Err(InternalError::row_layout_primary_key_slot_required());
    };
    let Some(Some(primary_key_value)) = slots.get(primary_key_slot) else {
        return Err(InternalError::row_decode_primary_key_slot_missing());
    };
    let decoded_key = StorageKey::try_from_value(primary_key_value)
        .map_err(InternalError::row_decode_primary_key_not_storage_encodable)?;

    if decoded_key != expected_key {
        return Err(InternalError::row_decode_key_mismatch(
            expected_key,
            decoded_key,
        ));
    }

    Ok(())
}

///
/// TESTS
///

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        db::data::decode_structural_field_by_kind_bytes,
        error::{ErrorClass, ErrorOrigin},
        model::field::{FieldKind, FieldStorageDecode},
        serialize::serialize,
        traits::EntitySchema,
        types::{Blob, Text},
        value::{Value, ValueEnum},
    };
    use icydb_derive::{FieldProjection, PersistedRow};
    use serde::{Deserialize, Serialize};
    use std::collections::BTreeMap;

    crate::test_canister! {
        ident = RowDecodeCanister,
        commit_memory_id = crate::testing::test_commit_memory_id(),
    }

    crate::test_store! {
        ident = RowDecodeStore,
        canister = RowDecodeCanister,
    }

    #[derive(
        Clone, Debug, Default, Deserialize, FieldProjection, PartialEq, PersistedRow, Serialize,
    )]
    struct RowDecodeEntity {
        id: Ulid,
        title: Text,
        tags: Vec<Text>,
        portrait: Blob,
    }

    crate::test_entity_schema! {
        ident = RowDecodeEntity,
        id = Ulid,
        id_field = id,
        entity_name = "RowDecodeEntity",
        entity_tag = crate::testing::PROBE_ENTITY_TAG,
        pk_index = 0,
        fields = [
            ("id", FieldKind::Ulid),
            ("title", FieldKind::Text),
            ("tags", FieldKind::List(&FieldKind::Text)),
            ("portrait", FieldKind::Blob),
        ],
        indexes = [],
        store = RowDecodeStore,
        canister = RowDecodeCanister,
    }

    fn decode_test_row(entity: &RowDecodeEntity) -> KernelRow {
        let key = crate::db::data::DataKey::try_new::<RowDecodeEntity>(entity.id)
            .expect("test key construction should succeed");
        let row = RawRow::from_entity(entity).expect("test row serialization should succeed");

        RowDecoder::structural()
            .decode(&RowLayout::from_model(RowDecodeEntity::MODEL), (key, row))
            .expect("structural row decode should succeed")
    }

    fn to_cbor_bytes<T: Serialize>(value: &T) -> Vec<u8> {
        serde_cbor::to_vec(value).expect("test fixture should serialize into CBOR bytes")
    }

    #[test]
    fn structural_row_decoder_materializes_slot_values_without_entity_decode() {
        let entity = RowDecodeEntity {
            id: Ulid::from_u128(7),
            title: "alpha".to_string(),
            tags: vec!["one".to_string(), "two".to_string()],
            portrait: Blob::from(vec![0x10, 0x20, 0x30]),
        };
        let row = decode_test_row(&entity);

        assert_eq!(row.slot(0), Some(Value::Ulid(entity.id)));
        assert_eq!(row.slot(1), Some(Value::Text(entity.title)));
        assert_eq!(
            row.slot(2),
            Some(Value::List(vec![
                Value::Text("one".to_string()),
                Value::Text("two".to_string()),
            ])),
        );
        assert_eq!(row.slot(3), Some(Value::Blob(vec![0x10, 0x20, 0x30])));
    }

    #[test]
    fn structural_row_decoder_rejects_raw_cbor_scalar_slot_payloads() {
        let entity = RowDecodeEntity {
            id: Ulid::from_u128(8),
            title: "alpha".to_string(),
            tags: vec!["one".to_string(), "two".to_string()],
            portrait: Blob::from(vec![0x10, 0x20, 0x30]),
        };
        let key = crate::db::data::DataKey::try_new::<RowDecodeEntity>(entity.id)
            .expect("test key construction should succeed");
        let id_bytes = crate::db::data::encode_persisted_scalar_slot_payload(&entity.id, "id")
            .expect("id payload should encode");
        let raw_title = serialize(&entity.title).expect("raw scalar title should encode");
        let tags_bytes = crate::db::data::encode_persisted_slot_payload(&entity.tags, "tags")
            .expect("tags payload should encode");
        let portrait_bytes =
            crate::db::data::encode_persisted_scalar_slot_payload(&entity.portrait, "portrait")
                .expect("portrait payload should encode");
        let slot_payloads = [
            id_bytes.as_slice(),
            raw_title.as_slice(),
            tags_bytes.as_slice(),
            portrait_bytes.as_slice(),
        ];
        let mut payload = Vec::new();
        let mut offset = 0_u32;

        payload.extend_from_slice(&4_u16.to_be_bytes());
        for bytes in slot_payloads {
            let len = u32::try_from(bytes.len()).expect("slot length should fit u32");
            payload.extend_from_slice(&offset.to_be_bytes());
            payload.extend_from_slice(&len.to_be_bytes());
            offset = offset.saturating_add(len);
        }
        for bytes in slot_payloads {
            payload.extend_from_slice(bytes);
        }
        let row = RawRow::try_new(
            crate::db::codec::serialize_row_payload(payload).expect("serialize row payload"),
        )
        .expect("build raw row");

        let Err(err) = RowDecoder::structural()
            .decode(&RowLayout::from_model(RowDecodeEntity::MODEL), (key, row))
        else {
            panic!("raw CBOR scalar slot payloads must fail closed");
        };

        assert_eq!(err.class, ErrorClass::Corruption);
        assert_eq!(err.origin, ErrorOrigin::Serialize);
        assert!(
            err.message.contains("field 'title'"),
            "unexpected error: {err:?}"
        );
        assert!(
            err.message
                .contains("expected slot envelope prefix byte 0xFF"),
            "unexpected error: {err:?}"
        );
    }

    #[test]
    fn structural_row_decoder_rejects_primary_key_mismatch() {
        let entity = RowDecodeEntity {
            id: Ulid::from_u128(9),
            title: "alpha".to_string(),
            tags: vec![],
            portrait: Blob::default(),
        };
        let wrong_key = crate::db::data::DataKey::try_new::<RowDecodeEntity>(Ulid::from_u128(10))
            .expect("wrong test key construction should succeed");
        let row = RawRow::from_entity(&entity).expect("test row serialization should succeed");
        let Err(err) = RowDecoder::structural().decode(
            &RowLayout::from_model(RowDecodeEntity::MODEL),
            (wrong_key, row),
        ) else {
            panic!("key mismatch must fail closed")
        };

        assert_eq!(err.class, ErrorClass::Corruption);
        assert_eq!(err.origin, ErrorOrigin::Store);
    }

    #[test]
    fn structural_row_decoder_returns_null_for_structured_field_kind() {
        let decoded = decode_structural_field_by_kind_bytes(
            &to_cbor_bytes(&vec!["x".to_string(), "y".to_string()]),
            FieldKind::Structured { queryable: false },
        )
        .expect("structured field decode should succeed");

        assert_eq!(decoded, Value::Null);
    }

    #[test]
    fn structural_row_decoder_preserves_enum_payload_shape_best_effort() {
        static ENUM_VARIANTS: &[EnumVariantModel] = &[EnumVariantModel::new(
            "Loaded",
            Some(&FieldKind::Uint),
            FieldStorageDecode::ByKind,
        )];

        let decoded = decode_structural_field_by_kind_bytes(
            &to_cbor_bytes(&serde_cbor::Value::Map(BTreeMap::from([(
                serde_cbor::Value::Text("Loaded".to_string()),
                serde_cbor::Value::Integer(7),
            )]))),
            FieldKind::Enum {
                path: "tests::State",
                variants: ENUM_VARIANTS,
            },
        )
        .expect("enum payload decode should succeed");

        assert_eq!(
            decoded,
            Value::Enum(
                ValueEnum::new("Loaded", Some("tests::State")).with_payload(Value::Uint(7)),
            ),
        );
    }

    #[derive(Clone, Debug, Deserialize, FieldProjection, PartialEq, PersistedRow, Serialize)]
    struct RowDecodeValueEntity {
        id: Ulid,
        status: Value,
    }

    impl Default for RowDecodeValueEntity {
        fn default() -> Self {
            Self {
                id: Ulid::from_u128(0),
                status: Value::Null,
            }
        }
    }

    crate::test_entity_schema! {
        ident = RowDecodeValueEntity,
        id = Ulid,
        id_field = id,
        entity_name = "RowDecodeValueEntity",
        entity_tag = crate::testing::PROBE_ENTITY_TAG,
        pk_index = 0,
        fields = [
            ("id", FieldKind::Ulid),
            (
                "status",
                FieldKind::Enum {
                    path: "tests::Status",
                    variants: &[],
                },
                crate::model::field::FieldStorageDecode::Value
            ),
        ],
        indexes = [],
        store = RowDecodeStore,
        canister = RowDecodeCanister,
    }

    #[test]
    fn structural_row_decoder_respects_value_storage_decode_contract() {
        let entity = RowDecodeValueEntity {
            id: Ulid::from_u128(77),
            status: Value::Enum(
                ValueEnum::new("Paid", Some("tests::Status")).with_payload(Value::Uint(7)),
            ),
        };
        let key = crate::db::data::DataKey::try_new::<RowDecodeValueEntity>(entity.id)
            .expect("test key construction should succeed");
        let row = RawRow::from_entity(&entity).expect("test row serialization should succeed");
        let decoded = RowDecoder::structural()
            .decode(
                &RowLayout::from_model(RowDecodeValueEntity::MODEL),
                (key, row),
            )
            .expect("structural row decode should succeed");

        assert_eq!(decoded.slot(1), Some(entity.status));
    }
}