verit-core 0.2.0

//! Schema resolution: the piece that buys evolution without giving up
//! zero-copy. A [`Resolver`] is built **once** per (writer schema, reader
//! schema) pair and compiles, for every corresponding struct type, an access
//! plan mapping each reader field ID to either a concrete byte slot in the
//! writer's layout (with an optional lossless widening) or `Absent`. After
//! that, reading any number of messages costs no per-message resolution work.

use std::collections::HashMap;

use crate::error::{Error, Result};
use crate::layout::slot_size_align;
use crate::schema::{Schema, Type};

/// Ceiling on resolver recursion depth. `pair` ↔ `compat` descend once per
/// nested type-expression node *and* once per struct-type reference; the memo
/// in `pair` stops cyclic references but not a long acyclic *chain* of distinct
/// structs (A0→A1→…→An). Without this bound a hostile inline schema (tens of
/// thousands of chained structs, still well-formed and small on the wire) would
/// overflow the stack when the message is resolved — e.g. via `dump_json`.
/// Every recursive step increments `depth`, so total live frames stay ≤ this
/// limit. It is comfortably above `schema::MAX_TYPE_DEPTH` (64) so any schema
/// that validates resolves, and matches the 128 depth cap the dump/verify
/// walkers already use, so a schema that resolves also renders.
const MAX_RESOLVE_DEPTH: u32 = 128;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum NumKind {
    U8,
    U16,
    U32,
    U64,
    I8,
    I16,
    I32,
    I64,
    F32,
    F64,
}

pub fn num_kind(ty: &Type) -> Option<NumKind> {
    Some(match ty {
        Type::U8 => NumKind::U8,
        Type::U16 => NumKind::U16,
        Type::U32 => NumKind::U32,
        Type::U64 => NumKind::U64,
        Type::I8 => NumKind::I8,
        Type::I16 => NumKind::I16,
        Type::I32 => NumKind::I32,
        Type::I64 => NumKind::I64,
        Type::F32 => NumKind::F32,
        Type::F64 => NumKind::F64,
        _ => return None,
    })
}

/// Lossless conversions permitted at read time: identity, unsigned widening,
/// signed widening, f32→f64. Anything else fails loudly at plan-build time.
fn widenable(from: NumKind, to: NumKind) -> bool {
    use NumKind::*;
    if from == to {
        return true;
    }
    matches!(
        (from, to),
        (U8, U16)
            | (U8, U32)
            | (U8, U64)
            | (U16, U32)
            | (U16, U64)
            | (U32, U64)
            | (I8, I16)
            | (I8, I32)
            | (I8, I64)
            | (I16, I32)
            | (I16, I64)
            | (I32, I64)
            | (F32, F64)
    )
}

/// How to materialize a reader-visible value from writer bytes.
#[derive(Clone, Debug)]
pub enum Load {
    Bool,
    Num {
        from: NumKind,
        to: NumKind,
    },
    Enum,
    Str,
    Bytes,
    /// Plan index. In a field slot this reads a u32 offset to the block; as a
    /// list element the block is inline at the element position.
    Struct(usize),
    List(Box<ElemPlan>),
    Map(Box<MapPlan>),
    Union(Box<UnionPlan>),
}

/// Read plan for a `union<…>`: how to load each variant and where its payload
/// sits within the union block (in the writer's layout). The u32 tag selects
/// the variant.
#[derive(Clone, Debug)]
pub struct UnionPlan {
    pub variants: Vec<VariantPlan>,
}

#[derive(Clone, Debug)]
pub struct VariantPlan {
    pub load: Load,
    pub payload_off: u32,
}

/// Read plan for a `map<K, V>`: how to load a key and a value out of each entry
/// block, plus that block's stride and the key/value offsets within it — all in
/// the **writer's** entry layout.
#[derive(Clone, Debug)]
pub struct MapPlan {
    pub key: Load,
    pub value: Load,
    pub stride: u32,
    pub align: u32,
    pub key_off: u32,
    pub value_off: u32,
}

#[derive(Clone, Debug)]
pub struct ElemPlan {
    pub load: Load,
    /// Element stride and alignment in the *writer's* layout.
    pub stride: u32,
    pub align: u32,
    /// True when elements are fixed-size struct blocks stored inline (the
    /// element position IS the block). False when the element position holds
    /// a u32 offset to the value (strings, bytes, nested lists, and
    /// variable-size *packed* struct elements).
    pub struct_inline: bool,
}

#[derive(Clone, Debug)]
pub enum FieldSource {
    /// Field does not exist in the writer schema: reader sees `None`.
    Absent,
    /// Field lives at a constant offset (fixed/dense writer struct).
    Slot {
        /// Slot offset relative to the struct block base (writer layout).
        offset: u32,
        /// Presence bit position, relative to block base. A zero mask means
        /// the field is always present (dense writer struct: no bitmap).
        presence_byte: u32,
        presence_mask: u8,
        load: Load,
    },
    /// Field lives in a *packed* writer struct: its presence bit and byte
    /// offset are recovered from the per-message bitmap via the writer's
    /// [`crate::layout::PackedLayout`] (looked up by the plan's `writer_type`).
    Packed {
        /// Field position (ID-sorted) in the writer struct.
        writer_pos: u32,
        load: Load,
    },
}

#[derive(Clone, Debug)]
pub struct FieldPlan {
    pub id: u16,
    pub source: FieldSource,
}

#[derive(Clone, Debug)]
pub struct StructPlan {
    /// Index of the corresponding struct in the *reader* schema.
    pub reader_type: u16,
    /// Index of the corresponding struct in the *writer* schema.
    pub writer_type: u16,
    /// Whether the writer struct uses the packed layout (offsets are dynamic).
    pub writer_packed: bool,
    /// Parallel to the reader struct's ID-sorted fields.
    pub fields: Vec<FieldPlan>,
}

#[derive(Clone, Debug)]
pub struct Resolver {
    writer: Schema,
    reader: Schema,
    plans: Vec<StructPlan>,
    root_plan: usize,
}

impl Resolver {
    /// Resolve a writer schema against a reader schema. Struct types
    /// correspond structurally from the root downward; fields match by ID.
    /// Incompatibilities (e.g. integer narrowing) error here — at plan time,
    /// once — never as silent corruption at read time.
    pub fn new(writer: &Schema, reader: &Schema) -> Result<Resolver> {
        let mut b = PlanBuilder {
            writer,
            reader,
            map: HashMap::new(),
            plans: Vec::new(),
        };
        let root_plan = b.pair(writer.root_index(), reader.root_index(), 0)?;
        Ok(Resolver {
            writer: writer.clone(),
            reader: reader.clone(),
            plans: b.plans,
            root_plan,
        })
    }

    /// The fast path for reading data written with your own schema.
    pub fn identity(schema: &Schema) -> Result<Resolver> {
        Resolver::new(schema, schema)
    }

    pub fn writer_id(&self) -> u128 {
        self.writer.id()
    }

    pub fn writer_schema(&self) -> &Schema {
        &self.writer
    }

    pub fn reader_schema(&self) -> &Schema {
        &self.reader
    }

    pub(crate) fn plan(&self, index: usize) -> &StructPlan {
        &self.plans[index]
    }

    pub(crate) fn root_plan_index(&self) -> usize {
        self.root_plan
    }
}

struct PlanBuilder<'a> {
    writer: &'a Schema,
    reader: &'a Schema,
    map: HashMap<(u16, u16), usize>,
    plans: Vec<StructPlan>,
}

impl<'a> PlanBuilder<'a> {
    fn pair(&mut self, writer_idx: u16, reader_idx: u16, depth: u32) -> Result<usize> {
        if depth > MAX_RESOLVE_DEPTH {
            return Err(Error::DepthLimitExceeded);
        }
        if let Some(&i) = self.map.get(&(writer_idx, reader_idx)) {
            return Ok(i);
        }
        // Insert a placeholder first so recursive type references terminate.
        let plan_idx = self.plans.len();
        let writer_packed = self.writer.struct_def_unchecked(writer_idx).is_packed();
        self.plans.push(StructPlan {
            reader_type: reader_idx,
            writer_type: writer_idx,
            writer_packed,
            fields: Vec::new(),
        });
        self.map.insert((writer_idx, reader_idx), plan_idx);

        let ws = self.writer.struct_def_unchecked(writer_idx);
        let rs = self.reader.struct_def_unchecked(reader_idx);
        let mut fields = Vec::with_capacity(rs.fields.len());
        for rf in &rs.fields {
            let source = match ws.fields.binary_search_by_key(&rf.id, |f| f.id) {
                Err(_) => FieldSource::Absent,
                Ok(wpos) => {
                    let wf = &ws.fields[wpos];
                    let load = self
                        .compat(&wf.ty, &rf.ty, depth + 1)
                        .map_err(|e| match e {
                            Error::Incompatible(msg) => Error::Incompatible(format!(
                                "field {} (id {}): {msg}",
                                rf.name, rf.id
                            )),
                            other => other,
                        })?;
                    if writer_packed {
                        FieldSource::Packed {
                            writer_pos: wpos as u32,
                            load,
                        }
                    } else {
                        let wlay = self.writer.layout_unchecked(writer_idx).as_fixed();
                        FieldSource::Slot {
                            offset: wlay.slots[wpos],
                            presence_byte: if ws.is_dense() { 0 } else { wpos as u32 / 8 },
                            presence_mask: if ws.is_dense() { 0 } else { 1 << (wpos % 8) },
                            load,
                        }
                    }
                }
            };
            fields.push(FieldPlan { id: rf.id, source });
        }
        self.plans[plan_idx].fields = fields;
        Ok(plan_idx)
    }

    fn compat(&mut self, w: &Type, r: &Type, depth: u32) -> Result<Load> {
        if depth > MAX_RESOLVE_DEPTH {
            return Err(Error::DepthLimitExceeded);
        }
        if let (Some(from), Some(to)) = (num_kind(w), num_kind(r)) {
            return if widenable(from, to) {
                Ok(Load::Num { from, to })
            } else {
                Err(Error::Incompatible(format!(
                    "cannot read writer {} as reader {} (only lossless widening is allowed)",
                    w.describe(self.writer),
                    r.describe(self.reader)
                )))
            };
        }
        match (w, r) {
            (Type::Bool, Type::Bool) => Ok(Load::Bool),
            (Type::String, Type::String) => Ok(Load::Str),
            (Type::Bytes, Type::Bytes) => Ok(Load::Bytes),
            // Enums are open u32s; variant names are documentation.
            (Type::Enum(_), Type::Enum(_)) => Ok(Load::Enum),
            (Type::Struct(wi), Type::Struct(ri)) => {
                Ok(Load::Struct(self.pair(*wi, *ri, depth + 1)?))
            }
            (Type::List(we), Type::List(re)) => {
                let load = self.compat(we, re, depth + 1)?;
                let (stride, align, struct_inline) = writer_elem_stride_align(self.writer, we);
                Ok(Load::List(Box::new(ElemPlan {
                    load,
                    stride,
                    align,
                    struct_inline,
                })))
            }
            (Type::Map(wk, wv), Type::Map(rk, rv)) => {
                // Keys must match exactly: widening a key could collapse two
                // distinct keys or reorder entries. Values evolve like any field.
                if wk != rk {
                    return Err(Error::Incompatible(format!(
                        "map key type changed: writer {} vs reader {}",
                        wk.describe(self.writer),
                        rk.describe(self.reader)
                    )));
                }
                let key = self.compat(wk, rk, depth + 1)?;
                let value = self.compat(wv, rv, depth + 1)?;
                let lay = crate::layout::map_entry_layout(wk, wv);
                Ok(Load::Map(Box::new(MapPlan {
                    key,
                    value,
                    stride: lay.size,
                    align: lay.align,
                    key_off: lay.slots[0],
                    value_off: lay.slots[1],
                })))
            }
            (Type::Union(wv), Type::Union(rv)) => {
                // Same number of variants; each variant value evolves like a
                // field (widening ok). The tag is positional, so variant order
                // is part of the type — a different count/order is incompatible.
                if wv.len() != rv.len() {
                    return Err(Error::Incompatible(format!(
                        "union variant count changed: writer {} vs reader {}",
                        wv.len(),
                        rv.len()
                    )));
                }
                let mut variants = Vec::with_capacity(wv.len());
                for (w, r) in wv.iter().zip(rv) {
                    variants.push(VariantPlan {
                        load: self.compat(w, r, depth + 1)?,
                        payload_off: crate::layout::union_payload_offset(w),
                    });
                }
                Ok(Load::Union(Box::new(UnionPlan { variants })))
            }
            _ => Err(Error::Incompatible(format!(
                "writer {} vs reader {}",
                w.describe(self.writer),
                r.describe(self.reader)
            ))),
        }
    }
}

/// Element stride/alignment as laid down by the writer, plus whether struct
/// elements are stored inline (fixed layout) or by u32 offset (packed).
fn writer_elem_stride_align(writer: &Schema, elem: &Type) -> (u32, u32, bool) {
    match elem {
        Type::Struct(i) => match writer.layout_unchecked(*i) {
            crate::layout::StructLayout::Fixed(f) => (f.size, f.align, true),
            // Variable-size packed elements ride in the list by u32 offset.
            crate::layout::StructLayout::Packed(_) => (4, 4, false),
        },
        Type::String | Type::Bytes | Type::List(_) => (4, 4, false),
        other => {
            let (s, a) = slot_size_align(other);
            (s, a, false)
        }
    }
}