verit-core 0.2.0

//! Schema model, builder, canonical binary encoding ("VSC1"), and content
//! hashing. The canonical form sorts types by name, fields by ID, and enum
//! variants by value, so one logical schema has exactly one byte encoding and
//! therefore exactly one id.

use crate::error::{Error, Result};
use crate::layout::{self, StructLayout};
use crate::value::Value;

pub const SCHEMA_MAGIC: &[u8; 4] = b"VSC1";

/// Maximum `list<…>` nesting depth. Bounds recursion in schema decode and in
/// resolver plan-building so a hostile schema cannot overflow the stack. Far
/// above any real schema (a handful of levels).
pub const MAX_TYPE_DEPTH: u32 = 64;

/// A field or element type. `Struct`/`Enum` reference other types in the same
/// schema by index into the canonically (name-)sorted type table.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Type {
    Bool,
    U8,
    U16,
    U32,
    U64,
    I8,
    I16,
    I32,
    I64,
    F32,
    F64,
    String,
    Bytes,
    Struct(u16),
    Enum(u16),
    List(Box<Type>),
    /// `map<K, V>`: an ordered set of (key, value) entries, entries sorted by
    /// key in canonical order so one logical map has exactly one encoding. `K`
    /// is restricted to a comparable key type (see [`Type::is_valid_map_key`]).
    Map(Box<Type>, Box<Type>),
    /// `union<T0, T1, …>`: a tagged sum. A value is a `u32` tag (an index into
    /// the variant list, which is positional and therefore part of the type's
    /// identity) followed by that variant's value.
    Union(Vec<Type>),
}

impl Type {
    pub fn describe(&self, schema: &Schema) -> String {
        match self {
            Type::Struct(i) => format!("struct {}", schema.type_name(*i)),
            Type::Enum(i) => format!("enum {}", schema.type_name(*i)),
            Type::List(e) => format!("list<{}>", e.describe(schema)),
            Type::Map(k, v) => format!("map<{}, {}>", k.describe(schema), v.describe(schema)),
            Type::Union(variants) => {
                let parts: Vec<String> = variants.iter().map(|t| t.describe(schema)).collect();
                format!("union<{}>", parts.join(", "))
            }
            other => format!("{other:?}").to_lowercase(),
        }
    }

    /// Legal `map` key types: booleans, integers, `string`, and enums — types
    /// with a total, well-defined canonical order. Floats (NaN), bytes, and
    /// composite types are not usable as keys.
    pub fn is_valid_map_key(&self) -> bool {
        matches!(
            self,
            Type::Bool
                | Type::U8
                | Type::U16
                | Type::U32
                | Type::U64
                | Type::I8
                | Type::I16
                | Type::I32
                | Type::I64
                | Type::String
                | Type::Enum(_)
        )
    }
}

/// A scalar **custom default** for a field: the value a reader synthesizes when
/// the field is absent (see [`StructReader::get_or_default`]). v1 supports
/// scalar defaults only; floats are stored as their bit patterns so the schema
/// types stay `Eq`.
///
/// [`StructReader::get_or_default`]: crate::StructReader::get_or_default
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Default {
    Bool(bool),
    U8(u8),
    U16(u16),
    U32(u32),
    U64(u64),
    I8(i8),
    I16(i16),
    I32(i32),
    I64(i64),
    /// f32 default, stored as `to_bits()`.
    F32(u32),
    /// f64 default, stored as `to_bits()`.
    F64(u64),
    Enum(u32),
}

impl Default {
    /// The number of little-endian bytes this default occupies in the canonical
    /// schema (matches its scalar type's slot size).
    fn wire_size(&self) -> usize {
        match self {
            Default::Bool(_) | Default::U8(_) | Default::I8(_) => 1,
            Default::U16(_) | Default::I16(_) => 2,
            Default::U32(_) | Default::I32(_) | Default::F32(_) | Default::Enum(_) => 4,
            Default::U64(_) | Default::I64(_) | Default::F64(_) => 8,
        }
    }

    /// The default's value as low-`wire_size` little-endian bytes.
    fn to_bits(self) -> u64 {
        match self {
            Default::Bool(b) => b as u64,
            Default::U8(x) => x as u64,
            Default::U16(x) => x as u64,
            Default::U32(x) => x as u64,
            Default::U64(x) => x,
            Default::I8(x) => x as u8 as u64,
            Default::I16(x) => x as u16 as u64,
            Default::I32(x) => x as u32 as u64,
            Default::I64(x) => x as u64,
            Default::F32(bits) => bits as u64,
            Default::F64(bits) => bits,
            Default::Enum(x) => x as u64,
        }
    }

    /// Reconstruct a default for scalar field type `ty` from `bits` (the low
    /// bytes read from the schema). Returns `None` if `ty` is not a scalar that
    /// can carry a default.
    fn from_bits(ty: &Type, bits: u64) -> Option<Default> {
        Some(match ty {
            Type::Bool => Default::Bool(bits != 0),
            Type::U8 => Default::U8(bits as u8),
            Type::U16 => Default::U16(bits as u16),
            Type::U32 => Default::U32(bits as u32),
            Type::U64 => Default::U64(bits),
            Type::I8 => Default::I8(bits as i8),
            Type::I16 => Default::I16(bits as i16),
            Type::I32 => Default::I32(bits as i32),
            Type::I64 => Default::I64(bits as i64),
            Type::F32 => Default::F32(bits as u32),
            Type::F64 => Default::F64(bits),
            Type::Enum(_) => Default::Enum(bits as u32),
            _ => return None,
        })
    }

    /// The scalar `Type` this default is for (used to validate it against the
    /// field). `None` for `Enum` (an enum default matches any enum field).
    fn scalar_type(&self) -> Option<Type> {
        Some(match self {
            Default::Bool(_) => Type::Bool,
            Default::U8(_) => Type::U8,
            Default::U16(_) => Type::U16,
            Default::U32(_) => Type::U32,
            Default::U64(_) => Type::U64,
            Default::I8(_) => Type::I8,
            Default::I16(_) => Type::I16,
            Default::I32(_) => Type::I32,
            Default::I64(_) => Type::I64,
            Default::F32(_) => Type::F32,
            Default::F64(_) => Type::F64,
            Default::Enum(_) => return None,
        })
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FieldDef {
    pub id: u16,
    pub name: String,
    pub ty: Type,
    /// Optional scalar default (see [`Default`]).
    pub default: Option<Default>,
}

/// How a struct's fields are stored.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StructMode {
    /// Presence bitmap + a fixed slot for every field, present or not
    /// (canonical kind byte 0). Best general default.
    Sparse,
    /// No bitmap; every field always present and mandatory (kind byte 2).
    /// Smallest and fastest for records with no optionality.
    Dense,
    /// Presence bitmap + slots for **present fields only**, packed in
    /// size-class order (kind byte 3). Field offsets are recovered in O(1)
    /// with popcount rank queries over the bitmap — sparse-data sizes without
    /// per-object vtables and without giving up zero-copy. At most 64 fields.
    Packed,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StructDef {
    pub name: String,
    /// Sorted by `id`, strictly ascending.
    pub fields: Vec<FieldDef>,
    pub mode: StructMode,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct EnumDef {
    pub name: String,
    /// (value, name), sorted by value, strictly ascending. Enums are open:
    /// unknown values round-trip as raw numbers.
    pub variants: Vec<(u32, String)>,
}

impl EnumDef {
    pub fn name_of(&self, value: u32) -> Option<&str> {
        self.variants
            .binary_search_by_key(&value, |(v, _)| *v)
            .ok()
            .map(|i| self.variants[i].1.as_str())
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TypeDef {
    Struct(StructDef),
    Enum(EnumDef),
}

impl TypeDef {
    pub fn name(&self) -> &str {
        match self {
            TypeDef::Struct(s) => &s.name,
            TypeDef::Enum(e) => &e.name,
        }
    }
}

impl StructDef {
    pub fn is_dense(&self) -> bool {
        self.mode == StructMode::Dense
    }

    pub fn is_packed(&self) -> bool {
        self.mode == StructMode::Packed
    }
}

/// A validated schema: canonical type table, cached canonical bytes, content
/// id, and precomputed struct layouts.
#[derive(Clone, Debug)]
pub struct Schema {
    types: Vec<TypeDef>,
    root: u16,
    canonical: Vec<u8>,
    id: u128,
    layouts: Vec<Option<StructLayout>>,
}

impl Schema {
    /// 128-bit content hash of the canonical schema bytes (truncated SHA-256,
    /// see [`crate::hash`]). Acts as a content-address / cache key; wide
    /// enough (~2^64 birthday resistance) to resist schema-confusion.
    pub fn id(&self) -> u128 {
        self.id
    }

    pub fn canonical_bytes(&self) -> &[u8] {
        &self.canonical
    }

    pub fn root_index(&self) -> u16 {
        self.root
    }

    pub fn type_count(&self) -> u16 {
        self.types.len() as u16
    }

    pub fn type_def(&self, index: u16) -> Option<&TypeDef> {
        self.types.get(index as usize)
    }

    pub fn type_name(&self, index: u16) -> &str {
        self.types
            .get(index as usize)
            .map(|t| t.name())
            .unwrap_or("<bad type index>")
    }

    /// Panics on a non-struct index: schema validation guarantees every
    /// `Type::Struct(i)` reference points at a struct def.
    pub(crate) fn struct_def_unchecked(&self, index: u16) -> &StructDef {
        match &self.types[index as usize] {
            TypeDef::Struct(s) => s,
            TypeDef::Enum(_) => panic!("schema invariant: type {index} is not a struct"),
        }
    }

    pub(crate) fn enum_def_unchecked(&self, index: u16) -> &EnumDef {
        match &self.types[index as usize] {
            TypeDef::Enum(e) => e,
            TypeDef::Struct(_) => panic!("schema invariant: type {index} is not an enum"),
        }
    }

    pub(crate) fn layout_unchecked(&self, index: u16) -> &StructLayout {
        self.layouts[index as usize]
            .as_ref()
            .expect("schema invariant: struct type has a layout")
    }

    pub(crate) fn packed_layout_unchecked(&self, index: u16) -> &crate::layout::PackedLayout {
        self.layout_unchecked(index).as_packed()
    }

    /// Position (in ID-sorted order) and definition of a field, by ID.
    pub fn find_field(&self, struct_index: u16, id: u16) -> Option<(usize, &FieldDef)> {
        let sd = match self.type_def(struct_index)? {
            TypeDef::Struct(s) => s,
            TypeDef::Enum(_) => return None,
        };
        sd.fields
            .binary_search_by_key(&id, |f| f.id)
            .ok()
            .map(|pos| (pos, &sd.fields[pos]))
    }

    /// Decode a schema from its canonical bytes (e.g. the inline schema
    /// region of a message). Rejects malformed and non-canonical encodings.
    pub fn from_canonical(bytes: &[u8]) -> Result<Schema> {
        let mut cur = Cur { b: bytes, p: 0 };
        let magic = cur.take(4)?;
        if magic != SCHEMA_MAGIC {
            return Err(Error::BadSchema("bad VSC1 magic".into()));
        }
        let type_count = cur.u16()?;
        let mut types = Vec::with_capacity(type_count as usize);
        for _ in 0..type_count {
            let kind = cur.u8()?;
            let name = cur.name()?;
            match kind {
                0 | 2 | 3 => {
                    let field_count = cur.u16()?;
                    let mut fields = Vec::with_capacity(field_count as usize);
                    for _ in 0..field_count {
                        let id = cur.u16()?;
                        let fname = cur.name()?;
                        let ty = cur.type_expr(type_count)?;
                        fields.push(FieldDef {
                            id,
                            name: fname,
                            ty,
                            default: None,
                        });
                    }
                    let mode = match kind {
                        2 => StructMode::Dense,
                        3 => StructMode::Packed,
                        _ => StructMode::Sparse,
                    };
                    types.push(TypeDef::Struct(StructDef { name, fields, mode }));
                }
                1 => {
                    let variant_count = cur.u16()?;
                    let mut variants = Vec::with_capacity(variant_count as usize);
                    for _ in 0..variant_count {
                        let value = cur.u32()?;
                        let vname = cur.name()?;
                        variants.push((value, vname));
                    }
                    types.push(TypeDef::Enum(EnumDef { name, variants }));
                }
                k => return Err(Error::BadSchema(format!("unknown type kind {k}"))),
            }
        }
        let root = cur.u16()?;
        // Optional defaults section (present only if there are trailing bytes).
        if cur.p < bytes.len() {
            let count = cur.u16()?;
            // The encoder omits the section entirely when there are no defaults,
            // so a present-but-empty section (count == 0) is a non-canonical
            // encoding of a default-less schema. Reject it: exactly one byte
            // string per schema.
            if count == 0 {
                return Err(Error::BadSchema(
                    "empty defaults section must be omitted, not encoded as count 0".into(),
                ));
            }
            for _ in 0..count {
                let ti = cur.u16()? as usize;
                let fid = cur.u16()?;
                let ty = match types.get(ti) {
                    Some(TypeDef::Struct(sd)) => {
                        sd.fields.iter().find(|f| f.id == fid).map(|f| f.ty.clone())
                    }
                    _ => None,
                }
                .ok_or_else(|| Error::BadSchema("default references unknown field".into()))?;
                let size = default_wire_size(&ty)
                    .ok_or_else(|| Error::BadSchema("default on a non-scalar field".into()))?;
                let raw = cur.take(size)?;
                let mut word = [0u8; 8];
                word[..size].copy_from_slice(raw);
                let d = Default::from_bits(&ty, u64::from_le_bytes(word))
                    .ok_or_else(|| Error::BadSchema("default on a non-scalar field".into()))?;
                if let Some(TypeDef::Struct(sd)) = types.get_mut(ti) {
                    if let Some(f) = sd.fields.iter_mut().find(|f| f.id == fid) {
                        f.default = Some(d);
                    }
                }
            }
        }
        if cur.p != bytes.len() {
            return Err(Error::BadSchema("trailing bytes after schema".into()));
        }
        let schema = Schema::assemble(types, root)?;
        // Strict canonicity: a decoded schema must re-encode to *exactly* the
        // bytes it came from. The structural checks above (name/id sort order,
        // no empty defaults section) reject most non-canonical forms, but not
        // every one — e.g. a scalar default whose bytes are a non-canonical
        // encoding of its value (a `bool` default byte other than 0/1, high
        // bits set past the type), or defaults-section entries out of canonical
        // order or duplicated (last-wins on decode). Comparing against the
        // re-encoding closes the whole class at once, enforcing the format's
        // one-schema-id ⇔ one-byte-string contract that the content address and
        // the schema fuzz oracle both rely on.
        if schema.canonical_bytes() != bytes {
            return Err(Error::BadSchema(
                "schema is not in canonical form (re-encoding differs)".into(),
            ));
        }
        Ok(schema)
    }

    /// Validate, canonically encode, hash, and compute layouts.
    fn assemble(types: Vec<TypeDef>, root: u16) -> Result<Schema> {
        validate(&types, root)?;
        let canonical = encode_canonical(&types, root);
        let id = crate::hash::schema_id(&canonical);
        let layouts = types
            .iter()
            .map(|t| match t {
                TypeDef::Struct(s) => Some(layout::compute(&s.fields, s.mode)),
                TypeDef::Enum(_) => None,
            })
            .collect();
        Ok(Schema {
            types,
            root,
            canonical,
            id,
            layouts,
        })
    }
}

fn validate(types: &[TypeDef], root: u16) -> Result<()> {
    if types.is_empty() {
        return Err(Error::BadSchema("schema has no types".into()));
    }
    // Types strictly ascending by name (canonical order, implies uniqueness).
    for w in types.windows(2) {
        if w[0].name() >= w[1].name() {
            return Err(Error::BadSchema(format!(
                "types not in canonical (name-sorted) order: {:?} then {:?}",
                w[0].name(),
                w[1].name()
            )));
        }
    }
    fn check_type(types: &[TypeDef], t: &Type, depth: u32) -> Result<()> {
        if depth > MAX_TYPE_DEPTH {
            return Err(Error::BadSchema(format!(
                "type nesting exceeds limit of {MAX_TYPE_DEPTH}"
            )));
        }
        match t {
            Type::Struct(i) => match types.get(*i as usize) {
                Some(TypeDef::Struct(_)) => Ok(()),
                _ => Err(Error::BadSchema(format!("type ref {i} is not a struct"))),
            },
            Type::Enum(i) => match types.get(*i as usize) {
                Some(TypeDef::Enum(_)) => Ok(()),
                _ => Err(Error::BadSchema(format!("type ref {i} is not an enum"))),
            },
            Type::List(e) => check_type(types, e, depth + 1),
            Type::Map(k, v) => {
                if !k.is_valid_map_key() {
                    return Err(Error::BadSchema(format!(
                        "map key type {k:?} is not a valid key (use bool, an integer, string, or an enum)"
                    )));
                }
                check_type(types, k, depth + 1)?;
                check_type(types, v, depth + 1)
            }
            Type::Union(variants) => {
                if variants.is_empty() {
                    return Err(Error::BadSchema("union has no variants".into()));
                }
                for v in variants {
                    check_type(types, v, depth + 1)?;
                }
                Ok(())
            }
            _ => Ok(()),
        }
    }
    let check_ref = |ty: &Type| -> Result<()> { check_type(types, ty, 0) };
    for td in types {
        match td {
            TypeDef::Struct(s) => {
                if s.name.is_empty() {
                    return Err(Error::BadSchema("empty type name".into()));
                }
                if s.mode == StructMode::Packed && s.fields.len() > 64 {
                    return Err(Error::BadSchema(format!(
                        "packed struct {} has {} fields; packed structs are \
                         limited to 64 (the bitmap must fit one u64 rank word)",
                        s.name,
                        s.fields.len()
                    )));
                }
                for w in s.fields.windows(2) {
                    if w[0].id >= w[1].id {
                        return Err(Error::BadSchema(format!(
                            "fields of {} not strictly ascending by id",
                            s.name
                        )));
                    }
                }
                for f in &s.fields {
                    if f.name.is_empty() {
                        return Err(Error::BadSchema(format!("empty field name in {}", s.name)));
                    }
                    check_ref(&f.ty)?;
                    if let Some(d) = f.default {
                        let ok = match d.scalar_type() {
                            Some(t) => f.ty == t,
                            None => matches!(f.ty, Type::Enum(_)), // Default::Enum
                        };
                        if !ok {
                            return Err(Error::BadSchema(format!(
                                "field {} in {} has a default whose type does not match the field",
                                f.name, s.name
                            )));
                        }
                    }
                }
            }
            TypeDef::Enum(e) => {
                if e.name.is_empty() {
                    return Err(Error::BadSchema("empty type name".into()));
                }
                for w in e.variants.windows(2) {
                    if w[0].0 >= w[1].0 {
                        return Err(Error::BadSchema(format!(
                            "variants of {} not strictly ascending by value",
                            e.name
                        )));
                    }
                }
            }
        }
    }
    match types.get(root as usize) {
        Some(TypeDef::Struct(_)) => Ok(()),
        Some(TypeDef::Enum(_)) => Err(Error::BadSchema("root type must be a struct".into())),
        None => Err(Error::BadSchema("root type index out of range".into())),
    }
}

// ---------------------------------------------------------------------------
// Canonical encoding
// ---------------------------------------------------------------------------

fn push_u16(b: &mut Vec<u8>, v: u16) {
    b.extend_from_slice(&v.to_le_bytes());
}

fn push_u32(b: &mut Vec<u8>, v: u32) {
    b.extend_from_slice(&v.to_le_bytes());
}

fn push_name(b: &mut Vec<u8>, s: &str) {
    push_u16(b, s.len() as u16);
    b.extend_from_slice(s.as_bytes());
}

fn push_type(b: &mut Vec<u8>, ty: &Type) {
    match ty {
        Type::Bool => b.push(0x01),
        Type::U8 => b.push(0x02),
        Type::U16 => b.push(0x03),
        Type::U32 => b.push(0x04),
        Type::U64 => b.push(0x05),
        Type::I8 => b.push(0x06),
        Type::I16 => b.push(0x07),
        Type::I32 => b.push(0x08),
        Type::I64 => b.push(0x09),
        Type::F32 => b.push(0x0A),
        Type::F64 => b.push(0x0B),
        Type::String => b.push(0x10),
        Type::Bytes => b.push(0x11),
        Type::Struct(i) => {
            b.push(0x20);
            push_u16(b, *i);
        }
        Type::Enum(i) => {
            b.push(0x21);
            push_u16(b, *i);
        }
        Type::List(e) => {
            b.push(0x22);
            push_type(b, e);
        }
        Type::Map(k, v) => {
            b.push(0x23);
            push_type(b, k);
            push_type(b, v);
        }
        Type::Union(variants) => {
            b.push(0x24);
            push_u16(b, variants.len() as u16);
            for v in variants {
                push_type(b, v);
            }
        }
    }
}

fn encode_canonical(types: &[TypeDef], root: u16) -> Vec<u8> {
    let mut b = Vec::new();
    b.extend_from_slice(SCHEMA_MAGIC);
    push_u16(&mut b, types.len() as u16);
    for td in types {
        match td {
            TypeDef::Struct(s) => {
                b.push(match s.mode {
                    StructMode::Sparse => 0,
                    StructMode::Dense => 2,
                    StructMode::Packed => 3,
                });
                push_name(&mut b, &s.name);
                push_u16(&mut b, s.fields.len() as u16);
                for f in &s.fields {
                    push_u16(&mut b, f.id);
                    push_name(&mut b, &f.name);
                    push_type(&mut b, &f.ty);
                }
            }
            TypeDef::Enum(e) => {
                b.push(1);
                push_name(&mut b, &e.name);
                push_u16(&mut b, e.variants.len() as u16);
                for (v, n) in &e.variants {
                    push_u32(&mut b, *v);
                    push_name(&mut b, n);
                }
            }
        }
    }
    push_u16(&mut b, root);
    // Defaults section — appended ONLY if some field carries a default, so a
    // schema without defaults is byte-identical to the pre-defaults format
    // (existing ids and golden vectors are unchanged). Entries are in canonical
    // order (types are name-sorted, fields id-sorted).
    let mut defaults: Vec<(u16, u16, Default)> = Vec::new();
    for (ti, td) in types.iter().enumerate() {
        if let TypeDef::Struct(sd) = td {
            for f in &sd.fields {
                if let Some(d) = f.default {
                    defaults.push((ti as u16, f.id, d));
                }
            }
        }
    }
    if !defaults.is_empty() {
        push_u16(&mut b, defaults.len() as u16);
        for (ti, fid, d) in defaults {
            push_u16(&mut b, ti);
            push_u16(&mut b, fid);
            let bytes = d.to_bits().to_le_bytes();
            b.extend_from_slice(&bytes[..d.wire_size()]);
        }
    }
    b
}

/// Byte width of a default for scalar type `ty`, or `None` if `ty` cannot carry
/// a default.
fn default_wire_size(ty: &Type) -> Option<usize> {
    Some(match ty {
        Type::Bool | Type::U8 | Type::I8 => 1,
        Type::U16 | Type::I16 => 2,
        Type::U32 | Type::I32 | Type::F32 | Type::Enum(_) => 4,
        Type::U64 | Type::I64 | Type::F64 => 8,
        _ => return None,
    })
}

// ---------------------------------------------------------------------------
// Canonical decoding cursor
// ---------------------------------------------------------------------------

struct Cur<'a> {
    b: &'a [u8],
    p: usize,
}

impl<'a> Cur<'a> {
    fn take(&mut self, n: usize) -> Result<&'a [u8]> {
        let end = self
            .p
            .checked_add(n)
            .ok_or_else(|| Error::BadSchema("length overflow".into()))?;
        let s = self
            .b
            .get(self.p..end)
            .ok_or_else(|| Error::BadSchema("schema truncated".into()))?;
        self.p = end;
        Ok(s)
    }

    fn u8(&mut self) -> Result<u8> {
        Ok(self.take(1)?[0])
    }

    fn u16(&mut self) -> Result<u16> {
        Ok(u16::from_le_bytes(self.take(2)?.try_into().unwrap()))
    }

    fn u32(&mut self) -> Result<u32> {
        Ok(u32::from_le_bytes(self.take(4)?.try_into().unwrap()))
    }

    fn name(&mut self) -> Result<String> {
        let len = self.u16()? as usize;
        let bytes = self.take(len)?;
        String::from_utf8(bytes.to_vec())
            .map_err(|_| Error::BadSchema("name is not valid UTF-8".into()))
    }

    fn type_expr(&mut self, type_count: u16) -> Result<Type> {
        self.type_expr_depth(type_count, 0)
    }

    /// Decode a type expression, bounding list nesting so a hostile inline
    /// schema (e.g. thousands of `list<` bytes) cannot overflow the stack.
    fn type_expr_depth(&mut self, type_count: u16, depth: u32) -> Result<Type> {
        if depth > MAX_TYPE_DEPTH {
            return Err(Error::BadSchema(format!(
                "type nesting exceeds limit of {MAX_TYPE_DEPTH}"
            )));
        }
        let code = self.u8()?;
        Ok(match code {
            0x01 => Type::Bool,
            0x02 => Type::U8,
            0x03 => Type::U16,
            0x04 => Type::U32,
            0x05 => Type::U64,
            0x06 => Type::I8,
            0x07 => Type::I16,
            0x08 => Type::I32,
            0x09 => Type::I64,
            0x0A => Type::F32,
            0x0B => Type::F64,
            0x10 => Type::String,
            0x11 => Type::Bytes,
            0x20 => {
                let i = self.u16()?;
                if i >= type_count {
                    return Err(Error::BadSchema("struct type index out of range".into()));
                }
                Type::Struct(i)
            }
            0x21 => {
                let i = self.u16()?;
                if i >= type_count {
                    return Err(Error::BadSchema("enum type index out of range".into()));
                }
                Type::Enum(i)
            }
            0x22 => Type::List(Box::new(self.type_expr_depth(type_count, depth + 1)?)),
            0x23 => {
                let key = self.type_expr_depth(type_count, depth + 1)?;
                let value = self.type_expr_depth(type_count, depth + 1)?;
                Type::Map(Box::new(key), Box::new(value))
            }
            0x24 => {
                let count = self.u16()?;
                if count == 0 {
                    return Err(Error::BadSchema("union has no variants".into()));
                }
                let mut variants = Vec::with_capacity(count as usize);
                for _ in 0..count {
                    variants.push(self.type_expr_depth(type_count, depth + 1)?);
                }
                Type::Union(variants)
            }
            c => return Err(Error::BadSchema(format!("unknown type code {c:#04x}"))),
        })
    }
}

// ---------------------------------------------------------------------------
// Builder
// ---------------------------------------------------------------------------

/// Draft type used in [`SchemaBuilder`]: like [`Type`] but names other types
/// by string instead of index (indices only exist after canonical sorting).
#[derive(Clone, Debug)]
pub enum Dt {
    Bool,
    U8,
    U16,
    U32,
    U64,
    I8,
    I16,
    I32,
    I64,
    F32,
    F64,
    Str,
    Bytes,
    Named(String),
    List(Box<Dt>),
    Map(Box<Dt>, Box<Dt>),
    Union(Vec<Dt>),
}

impl Dt {
    pub fn named(name: &str) -> Dt {
        Dt::Named(name.to_string())
    }

    pub fn list(elem: Dt) -> Dt {
        Dt::List(Box::new(elem))
    }

    pub fn map(key: Dt, value: Dt) -> Dt {
        Dt::Map(Box::new(key), Box::new(value))
    }

    pub fn union(variants: Vec<Dt>) -> Dt {
        Dt::Union(variants)
    }
}

enum DraftDef {
    Struct(Vec<(u16, String, Dt)>, StructMode),
    Enum(Vec<(u32, String)>),
}

/// Declaration order never matters: `build` sorts types by name and fields by
/// ID before encoding, so equivalent declarations produce identical schema ids.
pub struct SchemaBuilder {
    types: Vec<(String, DraftDef)>,
    defaults: Vec<(String, u16, Value)>,
}

impl SchemaBuilder {
    #[allow(clippy::new_without_default)]
    pub fn new() -> SchemaBuilder {
        SchemaBuilder {
            types: Vec::new(),
            defaults: Vec::new(),
        }
    }

    /// Give a scalar field a **custom default**: the value a reader synthesizes
    /// via [`StructReader::get_or_default`] when the field is absent. `default`
    /// must be a scalar [`Value`] matching the field's type (`Value::Enum` for
    /// an enum field). Applied at [`build`](SchemaBuilder::build).
    ///
    /// [`StructReader::get_or_default`]: crate::StructReader::get_or_default
    pub fn set_default(
        mut self,
        struct_name: &str,
        field_id: u16,
        default: Value,
    ) -> SchemaBuilder {
        self.defaults
            .push((struct_name.to_string(), field_id, default));
        self
    }

    pub fn add_struct(mut self, name: &str, fields: Vec<(u16, &str, Dt)>) -> SchemaBuilder {
        self.push_struct(name, fields, StructMode::Sparse);
        self
    }

    /// A dense struct stores no presence bitmap: every field is always
    /// present (encoding with one missing is an error). Smaller and slightly
    /// faster to read; use for records whose fields are all mandatory, e.g.
    /// geometric points, samples, matrix rows.
    pub fn add_dense_struct(mut self, name: &str, fields: Vec<(u16, &str, Dt)>) -> SchemaBuilder {
        self.push_struct(name, fields, StructMode::Dense);
        self
    }

    /// A packed struct stores a presence bitmap followed by slots for the
    /// *present* fields only. Field offsets are recovered in O(1) with
    /// popcount rank queries over the bitmap — sparse-data wire sizes without
    /// giving up zero-copy random access or paying for per-object vtables.
    /// Limited to 64 fields. Best for wide records that are usually sparse
    /// (events with many optional attributes, config with many defaults).
    pub fn add_packed_struct(mut self, name: &str, fields: Vec<(u16, &str, Dt)>) -> SchemaBuilder {
        self.push_struct(name, fields, StructMode::Packed);
        self
    }

    fn push_struct(&mut self, name: &str, fields: Vec<(u16, &str, Dt)>, mode: StructMode) {
        self.types.push((
            name.to_string(),
            DraftDef::Struct(
                fields
                    .into_iter()
                    .map(|(id, n, t)| (id, n.to_string(), t))
                    .collect(),
                mode,
            ),
        ));
    }

    pub fn add_enum(mut self, name: &str, variants: Vec<(u32, &str)>) -> SchemaBuilder {
        self.types.push((
            name.to_string(),
            DraftDef::Enum(
                variants
                    .into_iter()
                    .map(|(v, n)| (v, n.to_string()))
                    .collect(),
            ),
        ));
        self
    }

    pub fn build(mut self, root: &str) -> Result<Schema> {
        if self.types.len() > u16::MAX as usize {
            return Err(Error::BadSchema("too many types".into()));
        }
        self.types.sort_by(|a, b| a.0.cmp(&b.0));
        for w in self.types.windows(2) {
            if w[0].0 == w[1].0 {
                return Err(Error::BadSchema(format!(
                    "duplicate type name {:?}",
                    w[0].0
                )));
            }
        }
        let index_of = |name: &str| -> Result<u16> {
            self.types
                .binary_search_by(|(n, _)| n.as_str().cmp(name))
                .map(|i| i as u16)
                .map_err(|_| Error::BadSchema(format!("unknown type name {name:?}")))
        };
        let resolve = |dt: &Dt| -> Result<Type> {
            fn go(
                types: &[(String, DraftDef)],
                index_of: &dyn Fn(&str) -> Result<u16>,
                dt: &Dt,
            ) -> Result<Type> {
                Ok(match dt {
                    Dt::Bool => Type::Bool,
                    Dt::U8 => Type::U8,
                    Dt::U16 => Type::U16,
                    Dt::U32 => Type::U32,
                    Dt::U64 => Type::U64,
                    Dt::I8 => Type::I8,
                    Dt::I16 => Type::I16,
                    Dt::I32 => Type::I32,
                    Dt::I64 => Type::I64,
                    Dt::F32 => Type::F32,
                    Dt::F64 => Type::F64,
                    Dt::Str => Type::String,
                    Dt::Bytes => Type::Bytes,
                    Dt::Named(n) => {
                        let i = index_of(n)?;
                        match &types[i as usize].1 {
                            DraftDef::Struct(..) => Type::Struct(i),
                            DraftDef::Enum(_) => Type::Enum(i),
                        }
                    }
                    Dt::List(e) => Type::List(Box::new(go(types, index_of, e)?)),
                    Dt::Map(k, v) => Type::Map(
                        Box::new(go(types, index_of, k)?),
                        Box::new(go(types, index_of, v)?),
                    ),
                    Dt::Union(variants) => {
                        let mut out = Vec::with_capacity(variants.len());
                        for v in variants {
                            out.push(go(types, index_of, v)?);
                        }
                        Type::Union(out)
                    }
                })
            }
            go(&self.types, &index_of, dt)
        };

        let mut types = Vec::with_capacity(self.types.len());
        for (name, draft) in &self.types {
            match draft {
                DraftDef::Struct(fields, mode) => {
                    if fields.len() > u16::MAX as usize {
                        return Err(Error::BadSchema(format!("too many fields in {name}")));
                    }
                    let mut fds = Vec::with_capacity(fields.len());
                    for (id, fname, dt) in fields {
                        if fname.len() > u16::MAX as usize || name.len() > u16::MAX as usize {
                            return Err(Error::BadSchema("name too long".into()));
                        }
                        fds.push(FieldDef {
                            id: *id,
                            name: fname.clone(),
                            ty: resolve(dt)?,
                            default: None,
                        });
                    }
                    fds.sort_by_key(|f| f.id);
                    for w in fds.windows(2) {
                        if w[0].id == w[1].id {
                            return Err(Error::BadSchema(format!(
                                "duplicate field id {} in {name}",
                                w[0].id
                            )));
                        }
                    }
                    types.push(TypeDef::Struct(StructDef {
                        name: name.clone(),
                        fields: fds,
                        mode: *mode,
                    }));
                }
                DraftDef::Enum(variants) => {
                    let mut vs = variants.clone();
                    vs.sort_by_key(|(v, _)| *v);
                    for w in vs.windows(2) {
                        if w[0].0 == w[1].0 {
                            return Err(Error::BadSchema(format!(
                                "duplicate variant value {} in {name}",
                                w[0].0
                            )));
                        }
                    }
                    types.push(TypeDef::Enum(EnumDef {
                        name: name.clone(),
                        variants: vs,
                    }));
                }
            }
        }
        // Apply custom defaults (types are name-sorted; validate() will reject
        // a default whose type does not match its field).
        for (sname, fid, value) in &self.defaults {
            let d = value_to_default(value).ok_or_else(|| {
                Error::BadSchema(format!("default for {sname} field {fid} is not a scalar"))
            })?;
            let td = types
                .iter_mut()
                .find(|t| t.name() == sname)
                .ok_or_else(|| {
                    Error::BadSchema(format!("default references unknown type {sname:?}"))
                })?;
            match td {
                TypeDef::Struct(sd) => {
                    let f = sd.fields.iter_mut().find(|f| f.id == *fid).ok_or_else(|| {
                        Error::BadSchema(format!(
                            "default references unknown field {fid} in {sname}"
                        ))
                    })?;
                    f.default = Some(d);
                }
                TypeDef::Enum(_) => {
                    return Err(Error::BadSchema(format!(
                        "cannot set a default on enum type {sname}"
                    )))
                }
            }
        }
        let root_idx = index_of(root)?;
        Schema::assemble(types, root_idx)
    }
}

/// Convert a scalar [`Value`] to a [`Default`], or `None` for a non-scalar.
fn value_to_default(v: &Value) -> Option<Default> {
    Some(match v {
        Value::Bool(b) => Default::Bool(*b),
        Value::U8(x) => Default::U8(*x),
        Value::U16(x) => Default::U16(*x),
        Value::U32(x) => Default::U32(*x),
        Value::U64(x) => Default::U64(*x),
        Value::I8(x) => Default::I8(*x),
        Value::I16(x) => Default::I16(*x),
        Value::I32(x) => Default::I32(*x),
        Value::I64(x) => Default::I64(*x),
        Value::F32(x) => Default::F32(x.to_bits()),
        Value::F64(x) => Default::F64(x.to_bits()),
        Value::Enum(x) => Default::Enum(*x),
        _ => return None,
    })
}