icydb-core 0.200.1

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

#[cfg(test)]
use crate::model::entity::EntityModel;
use crate::{
    db::{
        codec::serialize_row_payload,
        data::{
            CanonicalRow, RawRow, StructuralRowContract,
            accepted_kind_supports_primary_key_component_binary,
            decode_structural_field_by_accepted_kind_bytes, decode_structural_value_storage_bytes,
            encode_structural_field_by_accepted_kind_bytes, encode_structural_value_storage_bytes,
            encode_structural_value_storage_null_bytes,
            validate_structural_field_by_accepted_kind_bytes,
            validate_structural_value_storage_bytes, value_storage_bytes_are_null,
        },
        schema::{AcceptedFieldDecodeContract, AcceptedFieldKind},
    },
    error::InternalError,
    model::field::{FieldStorageDecode, LeafCodec, ScalarCodec},
    types::Decimal,
    value::Value,
};
use std::{borrow::Cow, cmp::Ordering};

use crate::db::data::persisted_row::codec::{
    ScalarSlotValueRef, ScalarValueRef, decode_scalar_slot_value, encode_scalar_slot_value,
};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum AcceptedStorageValidationError {
    DecimalScaleMismatch,
}

pub(in crate::db::data::persisted_row) const RETIRED_SLOT_PLACEHOLDER_PAYLOAD: &[u8] = &[0];

/// Decode one structural slot payload into a runtime boundary `Value`.
///
/// Test models are first projected into accepted row authority, matching the
/// production decode path.
#[cfg(test)]
pub(in crate::db) fn decode_slot_into_runtime_value(
    model: &'static EntityModel,
    slot: usize,
    raw_value: &[u8],
) -> Result<Value, InternalError> {
    let contract = StructuralRowContract::from_model_proposal_for_test(model);

    decode_runtime_value_from_row_contract(&contract, slot, raw_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(codec) => {
            match decode_scalar_slot_value(raw_value, codec, field.field_name())? {
                ScalarSlotValueRef::Null => Ok(Value::Null),
                ScalarSlotValueRef::Value(value) => Ok(value.into_value()),
            }
        }
        LeafCodec::StructuralFallback => decode_non_scalar_accepted_slot_value(raw_value, field),
    }
}

/// Decode one slot payload through the accepted row contract.
///
/// This is the row-contract authority boundary for decode sites that know the
/// physical slot.
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 catalog = contract
            .accepted_enum_catalog_handle()
            .ok_or_else(InternalError::persisted_row_decode_corruption)?;
        let admitted = super::canonical::decode_admitted_value_from_accepted_field_contract(
            catalog,
            accepted_field,
            raw_value,
        )?;
        return Ok(admitted.value().clone());
    }

    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)?;

    let LeafCodec::Scalar(codec) = accepted_field.leaf_codec() else {
        return Err(InternalError::persisted_row_decode_corruption());
    };

    decode_scalar_slot_value(raw_value, codec, accepted_field.field_name())
}

/// Encode one runtime boundary `Value` into a persisted slot payload.
///
/// Test models are first projected into accepted row authority. This remains a
/// fixture helper, not permission to persist arbitrary runtime `Value`.
#[cfg(test)]
pub(in crate::db) fn encode_runtime_value_into_slot(
    model: &'static EntityModel,
    slot: usize,
    value: &Value,
) -> Result<Vec<u8>, InternalError> {
    let contract = StructuralRowContract::from_model_proposal_for_test(model);
    let field = contract.required_accepted_field_decode_contract(slot)?;

    encode_runtime_value_for_accepted_field_contract(field, value)
}

/// Encode one runtime boundary `Value` through an accepted field contract.
///
/// This is the accepted-schema counterpart to
/// `encode_runtime_value_for_field_model(...)`. `Value` remains a runtime
/// boundary object; this helper only selects the persisted field-codec lane
/// from accepted schema metadata before emitting slot bytes.
pub(in crate::db) fn encode_runtime_value_for_accepted_field_contract(
    field: AcceptedFieldDecodeContract<'_>,
    value: &Value,
) -> Result<Vec<u8>, InternalError> {
    let value = normalize_decimal_scale_for_accepted_storage(field.kind(), value)
        .map_err(|err| InternalError::persisted_row_field_encode_failed(field.field_name(), err))?;
    let value = value.as_ref();

    if matches!(value, Value::Null) {
        return encode_null_slot_value_for_accepted_field(field);
    }

    match field.storage_decode() {
        FieldStorageDecode::Value => encode_structural_value_storage_bytes(value).map_err(|err| {
            InternalError::persisted_row_field_encode_failed(field.field_name(), err)
        }),
        FieldStorageDecode::ByKind => match field.leaf_codec() {
            LeafCodec::Scalar(codec) => {
                let scalar =
                    scalar_slot_value_ref_from_runtime_value(value, codec).ok_or_else(|| {
                        InternalError::persisted_row_field_encode_internal(field.field_name())
                    })?;

                Ok(encode_scalar_slot_value(scalar))
            }
            LeafCodec::StructuralFallback => encode_structural_field_by_accepted_kind_bytes(
                field.kind(),
                value,
                field.field_name(),
            ),
        },
    }
}

// Encode an explicit nullable `NULL` through the accepted field's storage
// lane. Required fields fail before slot bytes are emitted.
fn encode_null_slot_value_for_accepted_field(
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<Vec<u8>, InternalError> {
    if !field.nullable() {
        return Err(InternalError::persisted_row_field_encode_internal(
            field.field_name(),
        ));
    }

    match field.storage_decode() {
        FieldStorageDecode::Value => Ok(encode_structural_value_storage_null_bytes()),
        FieldStorageDecode::ByKind => match field.leaf_codec() {
            LeafCodec::Scalar(_) => Ok(encode_scalar_slot_value(ScalarSlotValueRef::Null)),
            LeafCodec::StructuralFallback
                if accepted_kind_supports_primary_key_component_binary(field.kind()) =>
            {
                encode_structural_field_by_accepted_kind_bytes(
                    field.kind(),
                    &Value::Null,
                    field.field_name(),
                )
            }
            LeafCodec::StructuralFallback => Ok(encode_structural_value_storage_null_bytes()),
        },
    }
}

// Convert one runtime scalar value into the borrowed scalar-slot view expected
// by the persisted-row scalar codec. Field compatibility has already been
// checked by the model field contract before this storage encoder runs.
const fn scalar_slot_value_ref_from_runtime_value(
    value: &Value,
    codec: ScalarCodec,
) -> Option<ScalarSlotValueRef<'_>> {
    let scalar = match (codec, value) {
        (ScalarCodec::Blob, Value::Blob(value)) => ScalarValueRef::Blob(value.as_slice()),
        (ScalarCodec::Bool, Value::Bool(value)) => ScalarValueRef::Bool(*value),
        (ScalarCodec::Date, Value::Date(value)) => ScalarValueRef::Date(*value),
        (ScalarCodec::Duration, Value::Duration(value)) => ScalarValueRef::Duration(*value),
        (ScalarCodec::Float32, Value::Float32(value)) => ScalarValueRef::Float32(*value),
        (ScalarCodec::Float64, Value::Float64(value)) => ScalarValueRef::Float64(*value),
        (ScalarCodec::Int64, Value::Int64(value)) => ScalarValueRef::Int(*value),
        (ScalarCodec::Principal, Value::Principal(value)) => ScalarValueRef::Principal(*value),
        (ScalarCodec::Subaccount, Value::Subaccount(value)) => ScalarValueRef::Subaccount(*value),
        (ScalarCodec::Text, Value::Text(value)) => ScalarValueRef::Text(value.as_str()),
        (ScalarCodec::Timestamp, Value::Timestamp(value)) => ScalarValueRef::Timestamp(*value),
        (ScalarCodec::Nat64, Value::Nat64(value)) => ScalarValueRef::Nat(*value),
        (ScalarCodec::Ulid, Value::Ulid(value)) => ScalarValueRef::Ulid(*value),
        (ScalarCodec::Unit, Value::Unit) => ScalarValueRef::Unit,
        _ => return None,
    };

    Some(ScalarSlotValueRef::Value(scalar))
}

// Normalize decimal values to accepted persisted storage scale before encoding.
// `AcceptedFieldKind` remains the sole storage contract owner.
fn normalize_decimal_scale_for_accepted_storage<'a>(
    kind: &AcceptedFieldKind,
    value: &'a Value,
) -> Result<Cow<'a, Value>, AcceptedStorageValidationError> {
    if matches!(value, Value::Null) {
        return Ok(Cow::Borrowed(value));
    }

    match (kind, value) {
        (AcceptedFieldKind::Decimal { scale }, Value::Decimal(decimal)) => {
            let normalized = decimal_with_accepted_storage_scale(*decimal, *scale)
                .ok_or(AcceptedStorageValidationError::DecimalScaleMismatch)?;

            if normalized.scale() == decimal.scale() {
                Ok(Cow::Borrowed(value))
            } else {
                Ok(Cow::Owned(Value::Decimal(normalized)))
            }
        }
        (AcceptedFieldKind::Relation { key_kind, .. }, value) => {
            normalize_decimal_scale_for_accepted_storage(key_kind, value)
        }
        (AcceptedFieldKind::List(inner) | AcceptedFieldKind::Set(inner), Value::List(items)) => {
            normalize_accepted_decimal_list_items(inner, items.as_slice()).map(|items| {
                items.map_or_else(
                    || Cow::Borrowed(value),
                    |items| Cow::Owned(Value::List(items)),
                )
            })
        }
        (
            AcceptedFieldKind::Map {
                key,
                value: map_value,
            },
            Value::Map(entries),
        ) => normalize_accepted_decimal_map_entries(key, map_value, entries.as_slice()).map(
            |entries| {
                entries.map_or_else(
                    || Cow::Borrowed(value),
                    |items| Cow::Owned(Value::Map(items)),
                )
            },
        ),
        _ => Ok(Cow::Borrowed(value)),
    }
}

// Convert one accepted decimal into exact persisted storage scale.
fn decimal_with_accepted_storage_scale(decimal: Decimal, scale: u32) -> Option<Decimal> {
    match decimal.scale().cmp(&scale) {
        Ordering::Equal => Some(decimal),
        Ordering::Less => decimal
            .scale_to_integer(scale)
            .and_then(|mantissa| Decimal::try_from_i128_with_scale(mantissa, scale)),
        Ordering::Greater => Some(decimal.round_dp(scale)),
    }
}

// Normalize decimal values inside accepted list/set fields while preserving
// borrowed output when no element changes.
fn normalize_accepted_decimal_list_items(
    kind: &AcceptedFieldKind,
    items: &[Value],
) -> Result<Option<Vec<Value>>, AcceptedStorageValidationError> {
    let mut normalized = None;
    for (index, item) in items.iter().enumerate() {
        let value = normalize_decimal_scale_for_accepted_storage(kind, item)?;
        if let Cow::Owned(value) = value {
            let values = normalized.get_or_insert_with(|| items.to_vec());
            values[index] = value;
        }
    }

    Ok(normalized)
}

// Normalize decimal values inside accepted map fields while preserving
// borrowed output when no key or value changes.
fn normalize_accepted_decimal_map_entries(
    key_kind: &AcceptedFieldKind,
    value_kind: &AcceptedFieldKind,
    entries: &[(Value, Value)],
) -> Result<Option<Vec<(Value, Value)>>, AcceptedStorageValidationError> {
    let mut normalized = None;
    for (index, (entry_key, entry_value)) in entries.iter().enumerate() {
        let key = normalize_decimal_scale_for_accepted_storage(key_kind, entry_key)?;
        let value = normalize_decimal_scale_for_accepted_storage(value_kind, entry_value)?;
        if matches!(key, Cow::Owned(_)) || matches!(value, Cow::Owned(_)) {
            let values = normalized.get_or_insert_with(|| entries.to_vec());
            if let Cow::Owned(key) = key {
                values[index].0 = key;
            }
            if let Cow::Owned(value) = value {
                values[index].1 = value;
            }
        }
    }

    Ok(normalized)
}

// 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| {
        if !contract.has_active_field_slot(slot) {
            return Ok(RETIRED_SLOT_PLACEHOLDER_PAYLOAD.to_vec());
        }

        let value = value_for_slot(slot)?;
        let field = contract.required_accepted_field_decode_contract(slot)?;

        encode_runtime_value_for_accepted_field_contract(field, value.as_ref())
    })
}

// Encode one fixed-width slot table plus concatenated slot payload bytes into
// the canonical row payload container.
pub(in crate::db::data::persisted_row) fn encode_slot_payload_from_table_and_bytes(
    slot_count: usize,
    slot_table: &[(u32, u32)],
    payload_bytes: &[u8],
) -> Result<Vec<u8>, InternalError> {
    let field_count =
        u16::try_from(slot_count).map_err(|_| InternalError::persisted_row_encode_internal())?;
    let mut encoded = Vec::with_capacity(
        usize::from(field_count) * (u32::BITS as usize / 4) + 2 + payload_bytes.len(),
    );
    encoded.extend_from_slice(&field_count.to_be_bytes());
    for (start, len) in slot_table {
        encoded.extend_from_slice(&start.to_be_bytes());
        encoded.extend_from_slice(&len.to_be_bytes());
    }
    encoded.extend_from_slice(payload_bytes);

    Ok(encoded)
}

// Flatten one dense slot payload image into the canonical slot container while
// letting the caller keep ownership of slot-local overflow error wording.
fn encode_slot_payload_from_dense_slot_image<FS, FL>(
    slot_payloads: &[Vec<u8>],
    mut start_error: FS,
    mut len_error: FL,
) -> Result<Vec<u8>, InternalError>
where
    FS: FnMut(usize) -> InternalError,
    FL: FnMut(usize) -> InternalError,
{
    let payload_capacity = slot_payloads
        .iter()
        .try_fold(0usize, |len, payload| len.checked_add(payload.len()))
        .ok_or_else(InternalError::persisted_row_encode_internal)?;
    let mut payload_bytes = Vec::with_capacity(payload_capacity);
    let mut slot_table = Vec::with_capacity(slot_payloads.len());

    for (slot, payload) in slot_payloads.iter().enumerate() {
        let start = u32::try_from(payload_bytes.len()).map_err(|_| start_error(slot))?;
        let len = u32::try_from(payload.len()).map_err(|_| len_error(slot))?;
        payload_bytes.extend_from_slice(payload.as_slice());
        slot_table.push((start, len));
    }

    encode_slot_payload_from_table_and_bytes(
        slot_payloads.len(),
        slot_table.as_slice(),
        &payload_bytes,
    )
}

// Build and emit one canonical row from runtime values through accepted field
// contracts.
pub(in crate::db::data::persisted_row) 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.field_count(), slot_payloads.as_slice())
}

// Wrap one already-encoded canonical slot payload container in the shared row
// envelope so callers that already own a dense slot payload image do not have
// to rebuild the row wrapper choreography themselves.
fn canonical_row_from_slot_payload_bytes(
    row_payload: Vec<u8>,
) -> Result<CanonicalRow, InternalError> {
    let encoded = serialize_row_payload(row_payload)?;
    let raw_row = RawRow::from_untrusted_bytes(encoded).map_err(InternalError::from)?;

    Ok(CanonicalRow::from_canonical_raw_row(raw_row))
}

// Emit one raw row from a dense canonical slot image.
pub(in crate::db::data::persisted_row) fn emit_raw_row_from_slot_payloads(
    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());
    }

    // Phase 1: flatten the already canonicalized dense slot image directly so
    // row re-emission does not clone each slot payload back through the
    // mutable slot-writer staging buffer first.
    let row_payload = encode_slot_payload_from_dense_slot_image(
        slot_payloads,
        |_| InternalError::persisted_row_encode_internal(),
        |_| InternalError::persisted_row_encode_internal(),
    )?;

    // Phase 2: wrap the canonical slot container in the shared row envelope.
    canonical_row_from_slot_payload_bytes(row_payload)
}

// 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_non_primary_key_component_accepted_slot_payload_is_structural_null(
        raw_value, field,
    )? {
        return Ok(Value::Null);
    }

    match field.storage_decode() {
        FieldStorageDecode::ByKind => {
            decode_structural_field_by_accepted_kind_bytes(raw_value, field.kind()).map_err(|err| {
                InternalError::persisted_row_field_kind_decode_failed(
                    field.field_name(),
                    field.kind(),
                    err,
                )
            })
        }
        FieldStorageDecode::Value => {
            decode_structural_value_storage_bytes(raw_value).map_err(|err| {
                InternalError::persisted_row_field_kind_decode_failed(
                    field.field_name(),
                    field.kind(),
                    err,
                )
            })
        }
    }
}

/// Validate one non-scalar slot through an accepted-schema field contract.
/// Recursive payload validation stays on accepted `AcceptedFieldKind` metadata.
pub(in crate::db) fn validate_non_scalar_accepted_slot_value(
    raw_value: &[u8],
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<(), InternalError> {
    if nullable_non_primary_key_component_accepted_slot_payload_is_structural_null(
        raw_value, field,
    )? {
        return Ok(());
    }

    match field.storage_decode() {
        FieldStorageDecode::ByKind => {
            validate_structural_field_by_accepted_kind_bytes(raw_value, field.kind()).map_err(
                |err| {
                    InternalError::persisted_row_field_kind_decode_failed(
                        field.field_name(),
                        field.kind(),
                        err,
                    )
                },
            )
        }
        FieldStorageDecode::Value => {
            validate_structural_value_storage_bytes(raw_value).map_err(|err| {
                InternalError::persisted_row_field_kind_decode_failed(
                    field.field_name(),
                    field.kind(),
                    err,
                )
            })
        }
    }
}

/// 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 catalog = contract
            .accepted_enum_catalog_handle()
            .ok_or_else(InternalError::persisted_row_decode_corruption)?;
        super::canonical::decode_admitted_value_from_accepted_field_contract(
            catalog,
            accepted_field,
            raw_value,
        )?;
        return Ok(());
    }

    validate_non_scalar_accepted_slot_value(raw_value, accepted_field)
}

// Accepted-schema equivalent of the generated-field nullable structural-null
// check. Storage-key-compatible accepted kinds keep their own null encoding
// lane, so only non-storage-key by-kind payloads use the structural null
// sentinel here.
fn nullable_non_primary_key_component_accepted_slot_payload_is_structural_null(
    raw_value: &[u8],
    field: AcceptedFieldDecodeContract<'_>,
) -> Result<bool, InternalError> {
    if !field.nullable()
        || !matches!(field.storage_decode(), FieldStorageDecode::ByKind)
        || accepted_kind_supports_primary_key_component_binary(field.kind())
    {
        return Ok(false);
    }

    value_storage_bytes_are_null(raw_value).map_err(|err| {
        InternalError::persisted_row_field_kind_decode_failed(field.field_name(), field.kind(), err)
    })
}