icydb-core 0.138.1

use crate::{
    db::data::collection::{decode as collection_decode, encode as collection_encode},
    error::InternalError,
    model::field::FieldKind,
};
use std::collections::BTreeMap;

// Resolve list/set by-kind metadata once so encode and decode traversal helpers
// do not each own parallel FieldKind validation.
fn resolve_collection_item_kind(
    kind: FieldKind,
    field_name: &'static str,
    encode: bool,
) -> Result<FieldKind, InternalError> {
    collection_item_kind(kind).ok_or_else(|| {
        let message = format!("field kind {kind:?} does not accept collection payloads");
        if encode {
            InternalError::persisted_row_field_encode_failed(field_name, message)
        } else {
            InternalError::persisted_row_field_decode_failed(field_name, message)
        }
    })
}

// Resolve map by-kind metadata once so encode and decode traversal helpers do
// not each own parallel FieldKind validation.
fn resolve_map_entry_kinds(
    kind: FieldKind,
    field_name: &'static str,
    encode: bool,
) -> Result<(FieldKind, FieldKind), InternalError> {
    map_entry_kinds(kind).ok_or_else(|| {
        let message = format!("field kind {kind:?} does not accept map payloads");
        if encode {
            InternalError::persisted_row_field_encode_failed(field_name, message)
        } else {
            InternalError::persisted_row_field_decode_failed(field_name, message)
        }
    })
}

// Encode by-kind list-like containers. The caller provides the concrete field
// kind and field name, so item callbacks no longer need fake strategy/optional
// field metadata.
pub(in crate::db::data::persisted_row::codec) fn encode_collection<'a, I, T>(
    kind: FieldKind,
    items: I,
    field_name: &'static str,
    mut encode_item: impl FnMut(FieldKind, &T, &'static str) -> Result<Vec<u8>, InternalError>,
) -> Result<Vec<u8>, InternalError>
where
    I: IntoIterator<Item = &'a T>,
    T: 'a,
{
    let item_kind = resolve_collection_item_kind(kind, field_name, true)?;

    let iter = items.into_iter();
    let mut item_payloads = Vec::with_capacity(iter.size_hint().0);
    for item in iter {
        item_payloads.push(encode_item(item_kind, item, field_name)?);
    }

    let item_slices = item_payloads.iter().map(Vec::as_slice).collect::<Vec<_>>();

    collection_encode::field_items(item_slices.as_slice(), kind, field_name)
}

// Encode structured list-like containers without forcing structured callers
// through the by-kind callback shape.
pub(in crate::db::data::persisted_row::codec) fn encode_structured_collection<'a, I, T>(
    items: I,
    mut encode_item: impl FnMut(&T) -> Result<Vec<u8>, InternalError>,
) -> Result<Vec<u8>, InternalError>
where
    I: IntoIterator<Item = &'a T>,
    T: 'a,
{
    let iter = items.into_iter();
    let mut item_payloads = Vec::with_capacity(iter.size_hint().0);
    for item in iter {
        item_payloads.push(encode_item(item)?);
    }

    let item_slices = item_payloads.iter().map(Vec::as_slice).collect::<Vec<_>>();

    Ok(collection_encode::item(item_slices.as_slice()))
}

// Decode by-kind list-like containers. Callers choose whether the result
// remains ordered or is later validated as a set.
pub(in crate::db::data::persisted_row::codec) fn decode_collection<T>(
    kind: FieldKind,
    bytes: &[u8],
    field_name: &'static str,
    mut decode_item: impl FnMut(
        FieldKind,
        &[u8],
        &'static str,
        &'static str,
    ) -> Result<T, InternalError>,
) -> Result<Vec<T>, InternalError> {
    let item_kind = resolve_collection_item_kind(kind, field_name, false)?;
    let item_bytes = collection_decode::field_items(bytes, kind)
        .map_err(|err| InternalError::persisted_row_field_decode_failed(field_name, err))?;

    item_bytes
        .iter()
        .map(|item| decode_item(item_kind, item.as_slice(), field_name, "item"))
        .collect()
}

// Decode structured list-like containers without exposing by-kind metadata to
// the structured leaf decoder.
pub(in crate::db::data::persisted_row::codec) fn decode_structured_collection<T>(
    bytes: &[u8],
    decode_item: impl FnMut(&[u8]) -> Result<T, InternalError>,
) -> Result<Vec<T>, InternalError> {
    let item_bytes =
        collection_decode::item(bytes).map_err(InternalError::persisted_row_decode_failed)?;

    item_bytes.into_iter().map(decode_item).collect()
}

// Encode by-kind maps. The field envelope stays by-kind-specific, while
// callbacks receive concrete key/value kinds directly.
pub(in crate::db::data::persisted_row::codec) fn encode_map<K, V>(
    kind: FieldKind,
    entries: &BTreeMap<K, V>,
    field_name: &'static str,
    mut encode_key: impl FnMut(FieldKind, &K, &'static str) -> Result<Vec<u8>, InternalError>,
    mut encode_value: impl FnMut(FieldKind, &V, &'static str) -> Result<Vec<u8>, InternalError>,
) -> Result<Vec<u8>, InternalError>
where
    K: Ord,
{
    let (key_kind, value_kind) = resolve_map_entry_kinds(kind, field_name, true)?;

    let mut entry_payloads = Vec::with_capacity(entries.len());
    for (entry_key, entry_value) in entries {
        entry_payloads.push((
            encode_key(key_kind, entry_key, field_name)?,
            encode_value(value_kind, entry_value, field_name)?,
        ));
    }

    let entry_slices = entry_payloads
        .iter()
        .map(|(key_bytes, value_bytes)| (key_bytes.as_slice(), value_bytes.as_slice()))
        .collect::<Vec<_>>();

    collection_encode::field_entries(entry_slices.as_slice(), kind, field_name)
}

// Encode structured maps without carrying by-kind field metadata through every
// key/value callback.
pub(in crate::db::data::persisted_row::codec) fn encode_structured_map<K, V>(
    entries: &BTreeMap<K, V>,
    mut encode_key: impl FnMut(&K) -> Result<Vec<u8>, InternalError>,
    mut encode_value: impl FnMut(&V) -> Result<Vec<u8>, InternalError>,
) -> Result<Vec<u8>, InternalError>
where
    K: Ord,
{
    let mut entry_payloads = Vec::with_capacity(entries.len());
    for (entry_key, entry_value) in entries {
        entry_payloads.push((encode_key(entry_key)?, encode_value(entry_value)?));
    }

    let entry_slices = entry_payloads
        .iter()
        .map(|(key_bytes, value_bytes)| (key_bytes.as_slice(), value_bytes.as_slice()))
        .collect::<Vec<_>>();

    Ok(collection_encode::map(entry_slices.as_slice()))
}

// Decode by-kind maps. By-kind maps retain field-entry semantics and existing
// last-write-wins insertion behavior.
pub(in crate::db::data::persisted_row::codec) fn decode_map<K, V>(
    kind: FieldKind,
    bytes: &[u8],
    field_name: &'static str,
    decode_key: impl FnMut(FieldKind, &[u8], &'static str, &'static str) -> Result<K, InternalError>,
    decode_value: impl FnMut(FieldKind, &[u8], &'static str, &'static str) -> Result<V, InternalError>,
) -> Result<BTreeMap<K, V>, InternalError>
where
    K: Ord,
{
    let (key_kind, value_kind) = resolve_map_entry_kinds(kind, field_name, false)?;
    let entry_bytes = collection_decode::field_entries(bytes, kind)
        .map_err(|err| InternalError::persisted_row_field_decode_failed(field_name, err))?;

    decode_map_entries(
        entry_bytes
            .iter()
            .map(|(key_bytes, value_bytes)| (key_bytes.as_slice(), value_bytes.as_slice())),
        key_kind,
        value_kind,
        field_name,
        false,
        decode_key,
        decode_value,
    )
}

// Decode structured maps while preserving canonical ordering and duplicate
// checks without requiring structured callers to accept by-kind callback
// arguments.
pub(in crate::db::data::persisted_row::codec) fn decode_structured_map<K, V>(
    bytes: &[u8],
    mut decode_key: impl FnMut(&[u8]) -> Result<K, InternalError>,
    mut decode_value: impl FnMut(&[u8]) -> Result<V, InternalError>,
) -> Result<BTreeMap<K, V>, InternalError>
where
    K: Ord,
{
    let entry_bytes =
        collection_decode::map(bytes).map_err(InternalError::persisted_row_decode_failed)?;

    decode_entries(entry_bytes, true, |key_bytes, value_bytes| {
        Ok((decode_key(key_bytes)?, decode_value(value_bytes)?))
    })
}

// Decode already-framed by-kind map entries into a map.
fn decode_map_entries<'a, K, V>(
    entries: impl IntoIterator<Item = (&'a [u8], &'a [u8])>,
    key_kind: FieldKind,
    value_kind: FieldKind,
    field_name: &'static str,
    enforce_order: bool,
    mut decode_key: impl FnMut(FieldKind, &[u8], &'static str, &'static str) -> Result<K, InternalError>,
    mut decode_value: impl FnMut(
        FieldKind,
        &[u8],
        &'static str,
        &'static str,
    ) -> Result<V, InternalError>,
) -> Result<BTreeMap<K, V>, InternalError>
where
    K: 'a + Ord,
    V: 'a,
{
    decode_entries(entries, enforce_order, |key_bytes, value_bytes| {
        let key = decode_key(key_kind, key_bytes, field_name, "map key")?;
        let value = decode_value(value_kind, value_bytes, field_name, "map value")?;

        Ok((key, value))
    })
}

// Decode entry payload pairs into a map and optionally enforce structured-map
// ordering. This keeps by-kind and structured map decoding on one insertion
// path while allowing each caller to own its callback shape.
fn decode_entries<'a, K, V>(
    entries: impl IntoIterator<Item = (&'a [u8], &'a [u8])>,
    enforce_order: bool,
    mut decode_entry: impl FnMut(&'a [u8], &'a [u8]) -> Result<(K, V), InternalError>,
) -> Result<BTreeMap<K, V>, InternalError>
where
    K: 'a + Ord,
    V: 'a,
{
    let mut out = BTreeMap::new();
    for (key_bytes, value_bytes) in entries {
        let (key, value) = decode_entry(key_bytes, value_bytes)?;
        if enforce_order
            && let Some((previous_key, _)) = out.last_key_value()
            && key <= *previous_key
        {
            return Err(structured_map_decode_failed::<K, V>());
        }
        if out.insert(key, value).is_some() && enforce_order {
            return Err(structured_map_decode_failed::<K, V>());
        }
    }

    Ok(out)
}

// Build the canonical structured-map shape error from one place so both
// ordering and duplicate checks preserve exactly the same message.
fn structured_map_decode_failed<K, V>() -> InternalError {
    structured_container_decode_failed(&format!(
        "BTreeMap<{}, {}>",
        std::any::type_name::<K>(),
        std::any::type_name::<V>()
    ))
}

// Build the shared structured container mismatch error while keeping each
// caller responsible for its precise rendered type name.
pub(in crate::db::data::persisted_row::codec) fn structured_container_decode_failed(
    type_name: &str,
) -> InternalError {
    InternalError::persisted_row_decode_failed(format!("value payload does not match {type_name}"))
}

// Select the item kind shared by list and set wrappers. Keeping this decision
// here prevents encode and decode from carrying parallel FieldKind checks.
const fn collection_item_kind(kind: FieldKind) -> Option<FieldKind> {
    let (FieldKind::List(inner) | FieldKind::Set(inner)) = kind else {
        return None;
    };

    Some(*inner)
}

// Select the key/value kinds for a map wrapper. This is the single shape gate
// used by both map encode and map decode.
const fn map_entry_kinds(kind: FieldKind) -> Option<(FieldKind, FieldKind)> {
    let FieldKind::Map { key, value } = kind else {
        return None;
    };

    Some((*key, *value))
}