icydb-core 0.286.3

//! Accepted persisted-row field encoding and decode adapters.
//!
//! Production writes and test fixtures enter through accepted field contracts.

use crate::{
    db::schema::{AcceptedFieldKind, FieldStorageDecode, LeafCodec},
    db::{
        codec::serialize_row_payload,
        data::{
            CanonicalRow, RawRow, StructuralRowContract,
            decode_structural_field_by_accepted_kind_bytes, decode_structural_value_storage_bytes,
            validate_structural_field_by_accepted_kind_bytes,
            validate_structural_value_storage_bytes, value_storage_bytes_are_null,
        },
        schema::AcceptedFieldDecodeContract,
    },
    error::InternalError,
    value::Value,
};
use std::borrow::Cow;

use crate::db::data::persisted_row::codec::{
    ScalarSlotValueRef, ScalarValueRef, decode_scalar_slot_value,
};

/// Decode one slot payload through an accepted-schema field contract.
///
/// It keeps accepted `AcceptedFieldKind` metadata intact for recursive
/// payloads.
pub(in crate::db) fn decode_runtime_value_from_accepted_field_contract(
    field: AcceptedFieldDecodeContract<'_>,
    raw_value: &[u8],
) -> Result<Value, InternalError> {
    match field.leaf_codec() {
        LeafCodec::Scalar(_) => Ok(decode_scalar_slot_value_from_accepted_field_contract(
            field, raw_value,
        )?
        .into_value()),
        LeafCodec::Structural => decode_non_scalar_accepted_slot_value(raw_value, field),
    }
}

/// Decode and validate one slot payload through the accepted row contract.
///
/// This is the row-contract authority boundary for decode sites that know the
/// physical slot. Canonical admission and by-kind codec validation are owned
/// here; callers materializing a value must not prevalidate the payload again.
pub(in crate::db) fn decode_runtime_value_from_row_contract(
    contract: &StructuralRowContract,
    slot: usize,
    raw_value: &[u8],
) -> Result<Value, InternalError> {
    let accepted_field = contract.required_accepted_field_decode_contract(slot)?;

    if accepted_field.uses_canonical_value_wire() {
        let persistence = contract.required_accepted_field_persistence_contract(slot)?;
        let admitted = super::canonical::decode_admitted_value_from_accepted_field_contract(
            persistence,
            raw_value,
        )?;
        return Ok(admitted.into_value());
    }

    decode_runtime_value_from_accepted_field_contract(accepted_field, raw_value)
}

/// Decode one scalar slot payload through accepted row metadata.
pub(in crate::db) fn decode_scalar_slot_value_from_row_contract<'raw>(
    contract: &StructuralRowContract,
    slot: usize,
    raw_value: &'raw [u8],
) -> Result<ScalarSlotValueRef<'raw>, InternalError> {
    let accepted_field = contract.required_accepted_field_decode_contract(slot)?;

    decode_scalar_slot_value_from_accepted_field_contract(accepted_field, raw_value)
}

/// Admit a borrowed scalar against its accepted field without materializing it.
pub(in crate::db::data::persisted_row) fn decode_scalar_slot_value_from_accepted_field_contract<
    'raw,
>(
    field: AcceptedFieldDecodeContract<'_>,
    raw_value: &'raw [u8],
) -> Result<ScalarSlotValueRef<'raw>, InternalError> {
    let LeafCodec::Scalar(codec) = field.leaf_codec() else {
        return Err(InternalError::persisted_row_decode_corruption());
    };

    let value = decode_scalar_slot_value(raw_value, codec)?;
    let admitted = match (field.kind(), value) {
        (_, ScalarSlotValueRef::Null) => field.nullable(),
        (
            AcceptedFieldKind::Blob { max_len },
            ScalarSlotValueRef::Value(ScalarValueRef::Blob(bytes)),
        ) => max_len.is_none_or(|limit| bytes.len() <= limit as usize),
        (
            AcceptedFieldKind::Text { max_len },
            ScalarSlotValueRef::Value(ScalarValueRef::Text(text)),
        ) => max_len.is_none_or(|limit| text.chars().count() <= limit as usize),
        _ => true,
    };
    if !admitted {
        return Err(InternalError::persisted_row_decode_corruption());
    }
    Ok(value)
}

// Build one dense slot image by running one caller-supplied encode step per
// declared slot. This keeps the canonical row-emission loops on one shared
// shape while callers still decide whether they start from raw payload bytes or
// from already decoded runtime values.
fn dense_slot_image_from_source<F>(
    slot_count: usize,
    mut encode_slot: F,
) -> Result<Vec<Vec<u8>>, InternalError>
where
    F: FnMut(usize) -> Result<Vec<u8>, InternalError>,
{
    let mut slot_payloads = Vec::with_capacity(slot_count);

    for slot in 0..slot_count {
        slot_payloads.push(encode_slot(slot)?);
    }

    Ok(slot_payloads)
}

// Build one dense canonical slot image through accepted field metadata.
fn dense_canonical_slot_image_from_runtime_value_source_with_accepted_contract<'a, F>(
    contract: &StructuralRowContract,
    mut value_for_slot: F,
) -> Result<Vec<Vec<u8>>, InternalError>
where
    F: FnMut(usize) -> Result<Cow<'a, Value>, InternalError>,
{
    dense_slot_image_from_source(contract.field_count(), |slot| {
        let value = value_for_slot(slot)?;
        let encoding = contract.required_accepted_field_persistence_contract(slot)?;

        super::canonical::encode_canonical_value_for_accepted_field_contract(
            encoding,
            value.as_ref(),
        )
    })
}

// Build and emit one canonical row from runtime values through accepted field
// contracts.
pub(in crate::db) fn canonical_row_from_runtime_value_source_with_accepted_contract<'a, F>(
    contract: &StructuralRowContract,
    value_for_slot: F,
) -> Result<CanonicalRow, InternalError>
where
    F: FnMut(usize) -> Result<Cow<'a, Value>, InternalError>,
{
    let slot_payloads =
        dense_canonical_slot_image_from_runtime_value_source_with_accepted_contract(
            contract,
            value_for_slot,
        )?;

    emit_raw_row_from_slot_payloads(
        contract.current_layout_version(),
        contract.field_count(),
        slot_payloads.as_slice(),
    )
}

// Emit the directory and admitted field images directly into the bounded row.
pub(in crate::db) fn emit_raw_row_from_slot_payloads(
    layout_version: crate::db::schema::RowLayoutVersion,
    expected_slot_count: usize,
    slot_payloads: &[Vec<u8>],
) -> Result<CanonicalRow, InternalError> {
    if slot_payloads.len() != expected_slot_count {
        return Err(InternalError::persisted_row_encode_internal());
    }
    let field_count = u16::try_from(slot_payloads.len())
        .map_err(|_| InternalError::persisted_row_encode_internal())?;
    let directory_len = 2 + usize::from(field_count) * 8;
    let payload_len = slot_payloads
        .iter()
        .try_fold(directory_len, |len, payload| len.checked_add(payload.len()))
        .ok_or_else(InternalError::persisted_row_encode_internal)?;
    let encoded = serialize_row_payload(layout_version, payload_len, |encoded| {
        encoded.extend_from_slice(&field_count.to_be_bytes());
        let mut start = 0_u32;
        for payload in slot_payloads {
            let len = u32::try_from(payload.len())
                .map_err(|_| InternalError::persisted_row_encode_internal())?;
            encoded.extend_from_slice(&start.to_be_bytes());
            encoded.extend_from_slice(&len.to_be_bytes());
            start = start
                .checked_add(len)
                .ok_or_else(InternalError::persisted_row_encode_internal)?;
        }
        for payload in slot_payloads {
            encoded.extend_from_slice(payload);
        }
        Ok(())
    })?;
    let raw_row = RawRow::from_untrusted_bytes(encoded)?;
    Ok(CanonicalRow::from_canonical_raw_row(raw_row))
}

// Decode one non-scalar slot through the accepted persisted schema contract.
fn decode_non_scalar_accepted_slot_value(
    raw_value: &[u8],
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<Value, InternalError> {
    if nullable_accepted_slot_payload_is_structural_null(raw_value, field)? {
        return Ok(Value::Null);
    }

    // Named values keep canonical wire bytes even under ByKind proposal metadata.
    // Reuse schema's classifier instead of treating that label as a second authority.
    if field.uses_canonical_value_wire() {
        decode_structural_value_storage_bytes(raw_value)
            .map_err(|_| InternalError::persisted_row_decode_corruption())
    } else {
        decode_structural_field_by_accepted_kind_bytes(raw_value, field.kind())
            .map_err(|_| InternalError::persisted_row_decode_corruption())
    }
}

/// Validate one non-scalar slot through an accepted-schema field contract.
/// Schema selects its wire grammar; row contracts retain named-value admission.
pub(in crate::db) fn validate_non_scalar_accepted_slot_value(
    raw_value: &[u8],
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<(), InternalError> {
    if nullable_accepted_slot_payload_is_structural_null(raw_value, field)? {
        return Ok(());
    }

    if field.uses_canonical_value_wire() {
        validate_structural_value_storage_bytes(raw_value)
            .map_err(|_| InternalError::persisted_row_decode_corruption())
    } else {
        validate_structural_field_by_accepted_kind_bytes(raw_value, field.kind())
            .map_err(|_| InternalError::persisted_row_decode_corruption())
    }
}

/// Validate one non-scalar slot through the accepted row contract.
///
pub(in crate::db) fn validate_non_scalar_slot_value_with_row_contract(
    contract: &StructuralRowContract,
    slot: usize,
    raw_value: &[u8],
) -> Result<(), InternalError> {
    let accepted_field = contract.required_accepted_field_decode_contract(slot)?;
    if accepted_field.uses_canonical_value_wire() {
        let persistence = contract.required_accepted_field_persistence_contract(slot)?;
        super::canonical::decode_admitted_value_from_accepted_field_contract(
            persistence,
            raw_value,
        )?;
        return Ok(());
    }

    validate_non_scalar_accepted_slot_value(raw_value, accepted_field)
}

// Every nullable structural field uses the exact structural-null sentinel;
// relation collection membership does not change the field's nullability.
// Scalar codecs retain their separate slot framing. Non-null bytes still use
// the accepted codec rather than a preliminary walk with another grammar.
fn nullable_accepted_slot_payload_is_structural_null(
    raw_value: &[u8],
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<bool, InternalError> {
    let is_null = matches!(field.storage_decode(), FieldStorageDecode::ByKind)
        && value_storage_bytes_are_null(raw_value);
    if is_null && !field.nullable() {
        return Err(InternalError::persisted_row_decode_corruption());
    }
    Ok(is_null)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db::{
        codec::decode_row_payload_bytes,
        data::{
            encode_canonical_value_storage_bytes, encode_structural_field_by_accepted_kind_bytes,
        },
        schema::{AcceptedFieldKind, CompositeTypeId, RowLayoutVersion},
    };
    use crate::value::{
        CanonicalEnumBody, CanonicalEnumValue, EnumTypeId, EnumVariantId, ValueEnum,
    };

    #[test]
    fn named_canonical_fields_decode_and_validate_through_their_wire_owner() {
        let type_id = EnumTypeId::new(7).unwrap();
        let kind = AcceptedFieldKind::Enum { type_id };
        let value = Value::Enum(ValueEnum::from_canonical(CanonicalEnumValue::new(
            type_id,
            EnumVariantId::new(11).unwrap(),
            CanonicalEnumBody::Unit,
        )));
        for (kind, value) in [
            (kind.clone(), value.clone()),
            (
                kind.clone(),
                Value::Enum(ValueEnum::from_canonical(CanonicalEnumValue::new(
                    type_id,
                    EnumVariantId::new(12).unwrap(),
                    CanonicalEnumBody::Payload(Box::new(Value::Nat64(5))),
                ))),
            ),
            (
                AcceptedFieldKind::Composite {
                    type_id: CompositeTypeId::new(3).unwrap(),
                },
                Value::Map(vec![(Value::Text("status".into()), value.clone())]),
            ),
            (
                AcceptedFieldKind::List(Box::new(kind.clone())),
                Value::List(vec![value.clone()]),
            ),
            (
                AcceptedFieldKind::Set(Box::new(kind.clone())),
                Value::List(vec![value.clone()]),
            ),
            (
                AcceptedFieldKind::Map {
                    key: Box::new(AcceptedFieldKind::Nat64),
                    value: Box::new(kind),
                },
                Value::Map(vec![(Value::Nat64(1), value)]),
            ),
        ] {
            let wire = encode_canonical_value_storage_bytes(&value).unwrap();
            for lane in [FieldStorageDecode::ByKind, FieldStorageDecode::CatalogValue] {
                let field = AcceptedFieldDecodeContract::new(
                    "named",
                    &kind,
                    true,
                    lane,
                    LeafCodec::Structural,
                );
                for _ in 0..2 {
                    assert_eq!(
                        decode_runtime_value_from_accepted_field_contract(field, &wire).unwrap(),
                        value
                    );
                    validate_non_scalar_accepted_slot_value(&wire, field).unwrap();
                    assert_eq!(
                        decode_runtime_value_from_accepted_field_contract(field, &[0]).unwrap(),
                        Value::Null
                    );
                    validate_non_scalar_accepted_slot_value(&[0], field).unwrap();
                }
                let mut trailing = wire.clone();
                trailing.push(0);
                for malformed in [&[][..], &wire[..wire.len() - 1], &trailing, &[0xff]] {
                    let decoded =
                        decode_runtime_value_from_accepted_field_contract(field, malformed)
                            .unwrap_err();
                    let validated =
                        validate_non_scalar_accepted_slot_value(malformed, field).unwrap_err();
                    assert_eq!(decoded.class(), crate::error::ErrorClass::Corruption);
                    assert_eq!(decoded.diagnostic_code(), validated.diagnostic_code());
                }
            }
        }
    }

    #[test]
    fn structural_null_requires_nullable_accepted_field() {
        for kind in [
            AcceptedFieldKind::Int128,
            AcceptedFieldKind::Nat128,
            AcceptedFieldKind::Text { max_len: None },
            AcceptedFieldKind::Blob { max_len: None },
            AcceptedFieldKind::List(Box::new(AcceptedFieldKind::Nat64)),
            AcceptedFieldKind::Set(Box::new(AcceptedFieldKind::Nat64)),
            AcceptedFieldKind::Map {
                key: Box::new(AcceptedFieldKind::Nat64),
                value: Box::new(AcceptedFieldKind::Bool),
            },
        ] {
            for nullable in [false, true] {
                let field = AcceptedFieldDecodeContract::new(
                    "value",
                    &kind,
                    nullable,
                    FieldStorageDecode::ByKind,
                    LeafCodec::Structural,
                );
                let decoded = decode_runtime_value_from_accepted_field_contract(field, &[0]);
                let validated = validate_non_scalar_accepted_slot_value(&[0], field);
                if nullable {
                    assert_eq!(decoded.unwrap(), Value::Null);
                    validated.unwrap();
                } else {
                    let decoded = decoded.unwrap_err();
                    let validated = validated.unwrap_err();
                    assert_eq!(decoded.class(), crate::error::ErrorClass::Corruption);
                    assert_eq!(decoded.diagnostic_code(), validated.diagnostic_code());
                }
            }
        }
    }

    #[test]
    fn nullable_by_kind_collections_use_their_accepted_codec() {
        use crate::types::{Float32, Float64};

        for (kind, value) in [
            (AcceptedFieldKind::Nat64, Value::Nat64(7)),
            (AcceptedFieldKind::Float32, Value::Float32(Float32::from(7))),
            (AcceptedFieldKind::Float64, Value::Float64(Float64::from(7))),
        ] {
            let kind = AcceptedFieldKind::List(Box::new(kind));
            let value = Value::List(vec![value]);
            let field = AcceptedFieldDecodeContract::new(
                "items",
                &kind,
                true,
                FieldStorageDecode::ByKind,
                LeafCodec::Structural,
            );
            let wire = encode_structural_field_by_accepted_kind_bytes(&kind, &value).unwrap();
            assert_eq!(
                decode_runtime_value_from_accepted_field_contract(field, &wire).unwrap(),
                value
            );
            validate_non_scalar_accepted_slot_value(&wire, field).unwrap();
            assert_eq!(
                decode_runtime_value_from_accepted_field_contract(field, &[0]).unwrap(),
                Value::Null
            );
            validate_non_scalar_accepted_slot_value(&[0], field).unwrap();

            let mut trailing = wire.clone();
            trailing.push(0);
            for malformed in [&[][..], &[0, 0], &wire[..wire.len() - 1], &trailing] {
                let decoded = decode_runtime_value_from_accepted_field_contract(field, malformed)
                    .unwrap_err();
                let validated =
                    validate_non_scalar_accepted_slot_value(malformed, field).unwrap_err();
                assert_eq!(decoded.class(), crate::error::ErrorClass::Corruption);
                assert_eq!(decoded.diagnostic_code(), validated.diagnostic_code());
            }
        }
    }

    #[test]
    fn nullable_by_kind_collections_keep_accepted_depth_and_kind_boundaries() {
        use crate::db::schema::MAX_ACCEPTED_RECURSIVE_DEPTH;

        let mut kind = AcceptedFieldKind::Nat64;
        let mut value = Value::Nat64(7);
        for _ in 1..MAX_ACCEPTED_RECURSIVE_DEPTH {
            kind = AcceptedFieldKind::List(Box::new(kind));
            value = Value::List(vec![value]);
        }
        let field = AcceptedFieldDecodeContract::new(
            "items",
            &kind,
            true,
            FieldStorageDecode::ByKind,
            LeafCodec::Structural,
        );
        let wire = encode_structural_field_by_accepted_kind_bytes(&kind, &value).unwrap();
        assert_eq!(
            decode_runtime_value_from_accepted_field_contract(field, &wire).unwrap(),
            value
        );
        validate_non_scalar_accepted_slot_value(&wire, field).unwrap();

        // One extra list does not fit this accepted schema, even if a binary
        // walker can determine its framing independently of the field kind.
        let deeper = AcceptedFieldKind::List(Box::new(kind.clone()));
        let excessive =
            encode_structural_field_by_accepted_kind_bytes(&deeper, &Value::List(vec![value]))
                .unwrap();
        for malformed in [&excessive[..], &[0xff], &[0, 0]] {
            assert!(decode_runtime_value_from_accepted_field_contract(field, malformed).is_err());
            assert!(validate_non_scalar_accepted_slot_value(malformed, field).is_err());
        }
    }

    #[test]
    fn row_emission_preserves_collection_bytes_and_slot_offsets() {
        let kind = AcceptedFieldKind::List(Box::new(AcceptedFieldKind::Nat64));
        let value = Value::List((0..1_000).map(Value::Nat64).collect());
        let field = encode_structural_field_by_accepted_kind_bytes(&kind, &value)
            .expect("the accepted list should encode");
        assert_eq!(field.len(), 9_005);
        let mut fields = vec![field];
        let row = emit_raw_row_from_slot_payloads(RowLayoutVersion::INITIAL, 1, &fields)
            .expect("one collection field should emit");
        assert_eq!(row.as_raw_row().as_bytes().len(), 9_026);

        fields.push(vec![0]);
        let row = emit_raw_row_from_slot_payloads(RowLayoutVersion::INITIAL, 2, &fields)
            .expect("the collection and null slots should emit");
        let raw = row.as_raw_row().as_bytes();
        let decoded = decode_row_payload_bytes(raw).expect("the row envelope should decode");
        assert_eq!(decoded.layout_version(), RowLayoutVersion::INITIAL);
        let payload = decoded.into_payload();
        // Frozen count/offset/length directory: list at 0, null at 9,005.
        assert_eq!(
            &payload[..18],
            &[
                0, 2, 0, 0, 0, 0, 0, 0, 0x23, 0x2d, 0, 0, 0x23, 0x2d, 0, 0, 0, 1
            ],
        );
        assert_eq!(&payload[18..18 + 9_005], fields[0]);
        assert_eq!(&payload[18 + 9_005..], fields[1]);
        assert_eq!(
            decode_structural_field_by_accepted_kind_bytes(&payload[18..18 + 9_005], &kind)
                .expect("the emitted collection should decode"),
            value,
        );
        assert!(emit_raw_row_from_slot_payloads(RowLayoutVersion::INITIAL, 1, &fields).is_err());
    }
}