verit 0.2.0

//! Packed structs: presence bitmap + slots for present fields only, offsets
//! recovered by popcount rank queries. These tests prove the offset math is
//! correct across every presence pattern, that the wire is smaller on sparse
//! data, and that mode is invisible to the ID-based evolution model.

use verit::{
    dump_json, encode, Dt, Error, Message, Ref, Resolver, Schema, SchemaBuilder, SchemaMode, Value,
};

/// A packed struct mixing every size class (8/4/2/1) and a heap field, so the
/// popcount layout has to interleave classes correctly.
fn wide_schema() -> Schema {
    SchemaBuilder::new()
        .add_packed_struct(
            "Wide",
            vec![
                (1, "a", Dt::U8),
                (2, "b", Dt::U64),
                (3, "c", Dt::U16),
                (4, "d", Dt::Str),
                (5, "e", Dt::U32),
                (6, "f", Dt::Bool),
                (7, "g", Dt::F64),
                (8, "h", Dt::U8),
            ],
        )
        .build("Wide")
        .unwrap()
}

fn field_value(id: u16) -> Value {
    match id {
        1 => Value::U8(0xAB),
        2 => Value::U64(0x1122_3344_5566_7788),
        3 => Value::U16(0xBEEF),
        4 => Value::str("popcount"),
        5 => Value::U32(0xDEAD_BEEF),
        6 => Value::Bool(true),
        7 => Value::F64(std::f64::consts::PI),
        8 => Value::U8(0xCD),
        _ => unreachable!(),
    }
}

fn check_field(root: &verit::StructReader, id: u16, present: bool) {
    let got = root.get(id).unwrap();
    if !present {
        assert!(got.is_none(), "field {id} should be absent");
        return;
    }
    match (id, got.unwrap()) {
        (1, Ref::U8(v)) => assert_eq!(v, 0xAB),
        (2, Ref::U64(v)) => assert_eq!(v, 0x1122_3344_5566_7788),
        (3, Ref::U16(v)) => assert_eq!(v, 0xBEEF),
        (4, Ref::Str(v)) => assert_eq!(v, "popcount"),
        (5, Ref::U32(v)) => assert_eq!(v, 0xDEAD_BEEF),
        (6, Ref::Bool(v)) => assert!(v),
        (7, Ref::F64(v)) => assert_eq!(v, std::f64::consts::PI),
        (8, Ref::U8(v)) => assert_eq!(v, 0xCD),
        (id, other) => panic!("field {id} wrong type {}", other.kind()),
    }
}

/// Exhaustive: for all 2^8 subsets of the 8 fields, encode exactly that subset
/// and verify present fields read their value and absent fields read `None`.
/// This exercises `field_offset` across every possible presence bitmap.
#[test]
fn popcount_offsets_correct_for_every_subset() {
    let schema = wide_schema();
    let resolver = Resolver::identity(&schema).unwrap();
    for mask in 0u32..256 {
        let mut fields = Vec::new();
        for bit in 0..8u16 {
            if mask & (1 << bit) != 0 {
                let id = bit + 1;
                fields.push((id, field_value(id)));
            }
        }
        let bytes = encode(&schema, &Value::Struct(fields), SchemaMode::HashOnly).unwrap();
        let msg = Message::parse(&bytes).unwrap();
        let root = msg.root(&resolver).unwrap();
        for bit in 0..8u16 {
            check_field(&root, bit + 1, mask & (1 << bit) != 0);
        }
    }
}

#[test]
fn empty_packed_struct_is_tiny() {
    let schema = wide_schema();
    let bytes = encode(&schema, &Value::Struct(vec![]), SchemaMode::HashOnly).unwrap();
    let msg = Message::parse(&bytes).unwrap();
    let resolver = Resolver::identity(&schema).unwrap();
    let root = msg.root(&resolver).unwrap();
    for id in 1..=8u16 {
        assert!(root.get(id).unwrap().is_none());
    }
}

/// The headline win: a wide record that is mostly empty. Packed stores the
/// bitmap + only the present slots; sparse stores a slot for every field.
#[test]
fn packed_beats_sparse_on_sparse_data() {
    let fields = || {
        vec![
            (1, "f1", Dt::U64),
            (2, "f2", Dt::U64),
            (3, "f3", Dt::U64),
            (4, "f4", Dt::U64),
            (5, "f5", Dt::U64),
            (6, "f6", Dt::U64),
            (7, "f7", Dt::U64),
            (8, "f8", Dt::U64),
        ]
    };
    let packed = SchemaBuilder::new()
        .add_packed_struct("R", fields())
        .build("R")
        .unwrap();
    let sparse = SchemaBuilder::new()
        .add_struct("R", fields())
        .build("R")
        .unwrap();

    // Only 1 of 8 fields set.
    let value = Value::Struct(vec![(3, Value::U64(42))]);
    let p = encode(&packed, &value, SchemaMode::HashOnly).unwrap();
    let s = encode(&sparse, &value, SchemaMode::HashOnly).unwrap();
    // Sparse pays 8 u64 slots (64 bytes); packed pays 1 (8 bytes) + bitmap.
    assert!(
        s.len() >= p.len() + 40,
        "packed {} should be far smaller than sparse {}",
        p.len(),
        s.len()
    );

    // And it still reads back correctly.
    let resolver = Resolver::identity(&packed).unwrap();
    let msg = Message::parse(&p).unwrap();
    let root = msg.root(&resolver).unwrap();
    assert_eq!(root.get_u64(3).unwrap(), Some(42));
    assert_eq!(root.get_u64(1).unwrap(), None);
}

#[test]
fn packed_as_nested_field_and_list_element() {
    // Packed struct used both as a nested field and as a list element (where
    // it must be stored by-offset because its size is variable per element).
    let schema = SchemaBuilder::new()
        .add_packed_struct(
            "Attr",
            vec![
                (1, "key", Dt::Str),
                (2, "int_val", Dt::I64),
                (3, "flag", Dt::Bool),
            ],
        )
        .add_struct(
            "Event",
            vec![
                (1, "primary", Dt::named("Attr")),
                (2, "extras", Dt::list(Dt::named("Attr"))),
            ],
        )
        .build("Event")
        .unwrap();
    let bytes = encode(
        &schema,
        &Value::Struct(vec![
            (
                1,
                Value::Struct(vec![(1, Value::str("root")), (3, Value::Bool(true))]),
            ),
            (
                2,
                Value::List(vec![
                    Value::Struct(vec![(2, Value::I64(-7))]),
                    Value::Struct(vec![
                        (1, Value::str("k2")),
                        (2, Value::I64(99)),
                        (3, Value::Bool(false)),
                    ]),
                    Value::Struct(vec![]), // empty packed element
                ]),
            ),
        ]),
        SchemaMode::HashOnly,
    )
    .unwrap();

    let resolver = Resolver::identity(&schema).unwrap();
    let msg = Message::parse(&bytes).unwrap();
    let root = msg.root(&resolver).unwrap();

    let primary = root.get_struct(1).unwrap().unwrap();
    assert_eq!(primary.get_str(1).unwrap(), Some("root"));
    assert_eq!(primary.get_i64(2).unwrap(), None);
    assert_eq!(primary.get_bool(3).unwrap(), Some(true));

    let extras = root.get_list(2).unwrap().unwrap();
    assert_eq!(extras.len(), 3);
    let e0 = match extras.get(0).unwrap() {
        Ref::Struct(s) => s,
        o => panic!("{}", o.kind()),
    };
    assert_eq!(e0.get_i64(2).unwrap(), Some(-7));
    assert_eq!(e0.get_str(1).unwrap(), None);
    let e1 = match extras.get(1).unwrap() {
        Ref::Struct(s) => s,
        o => panic!("{}", o.kind()),
    };
    assert_eq!(e1.get_str(1).unwrap(), Some("k2"));
    assert_eq!(e1.get_i64(2).unwrap(), Some(99));
    assert_eq!(e1.get_bool(3).unwrap(), Some(false));
    let e2 = match extras.get(2).unwrap() {
        Ref::Struct(s) => s,
        o => panic!("{}", o.kind()),
    };
    assert_eq!(e2.get_str(1).unwrap(), None);
    assert_eq!(e2.get_i64(2).unwrap(), None);
}

#[test]
fn packed_missing_field_is_just_absent_not_an_error() {
    // Unlike dense, packed fields are all optional.
    let schema = wide_schema();
    let bytes = encode(
        &schema,
        &Value::Struct(vec![(2, Value::U64(1))]),
        SchemaMode::HashOnly,
    )
    .unwrap();
    let resolver = Resolver::identity(&schema).unwrap();
    let msg = Message::parse(&bytes).unwrap();
    let root = msg.root(&resolver).unwrap();
    assert_eq!(root.get_u64(2).unwrap(), Some(1));
    assert_eq!(root.get_u8(1).unwrap(), None);
}

#[test]
fn packed_duplicate_and_unknown_fields_rejected() {
    let schema = wide_schema();
    assert!(matches!(
        encode(
            &schema,
            &Value::Struct(vec![(2, Value::U64(1)), (2, Value::U64(2))]),
            SchemaMode::HashOnly
        ),
        Err(Error::DuplicateField(2))
    ));
    assert!(matches!(
        encode(
            &schema,
            &Value::Struct(vec![(99, Value::U64(1))]),
            SchemaMode::HashOnly
        ),
        Err(Error::UnknownFieldId(99))
    ));
}

/// Mode is a storage detail: evolution is by field ID and works across any
/// mix of writer/reader modes.
#[test]
fn mode_is_invisible_to_evolution() {
    let packed = SchemaBuilder::new()
        .add_packed_struct("P", vec![(1, "x", Dt::U16), (2, "s", Dt::Str)])
        .build("P")
        .unwrap();
    let sparse = SchemaBuilder::new()
        .add_struct(
            "P",
            vec![(1, "x", Dt::U32), (2, "s", Dt::Str), (3, "extra", Dt::Bool)],
        )
        .build("P")
        .unwrap();

    // packed writer -> sparse reader (with u16->u32 widening and an added field)
    let bytes = encode(
        &packed,
        &Value::Struct(vec![(1, Value::U16(1000)), (2, Value::str("hi"))]),
        SchemaMode::HashOnly,
    )
    .unwrap();
    let up = Resolver::new(&packed, &sparse).unwrap();
    let msg = Message::parse(&bytes).unwrap();
    let root = msg.root(&up).unwrap();
    assert_eq!(root.get_u32(1).unwrap(), Some(1000));
    assert_eq!(root.get_str(2).unwrap(), Some("hi"));
    assert_eq!(root.get_bool(3).unwrap(), None);

    // sparse writer -> packed reader (narrowing must still be refused loudly)
    let down = Resolver::new(&sparse, &packed);
    assert!(
        matches!(down, Err(Error::Incompatible(_))),
        "u32->u16 must fail"
    );

    // sparse writer (u16) -> packed reader (u16): fine
    let sparse16 = SchemaBuilder::new()
        .add_struct("P", vec![(1, "x", Dt::U16), (2, "s", Dt::Str)])
        .build("P")
        .unwrap();
    let bytes = encode(
        &sparse16,
        &Value::Struct(vec![(1, Value::U16(7)), (2, Value::str("yo"))]),
        SchemaMode::HashOnly,
    )
    .unwrap();
    let r = Resolver::new(&sparse16, &packed).unwrap();
    let msg = Message::parse(&bytes).unwrap();
    let root = msg.root(&r).unwrap();
    assert_eq!(root.get_u16(1).unwrap(), Some(7));
    assert_eq!(root.get_str(2).unwrap(), Some("yo"));
}

#[test]
fn packed_is_self_describing() {
    let schema = wide_schema();
    let bytes = encode(
        &schema,
        &Value::Struct(vec![
            (1, Value::U8(5)),
            (4, Value::str("hi")),
            (7, Value::F64(1.5)),
        ]),
        SchemaMode::Inline,
    )
    .unwrap();
    // Recovered from bytes alone, present fields in schema (id) order.
    assert_eq!(dump_json(&bytes).unwrap(), r#"{"a":5,"d":"hi","g":1.5}"#);
}

#[test]
fn packed_survives_canonical_roundtrip_and_has_distinct_id() {
    let packed = wide_schema();
    let recovered = Schema::from_canonical(packed.canonical_bytes()).unwrap();
    assert_eq!(recovered.id(), packed.id());

    // Same fields, sparse: different layout => different identity.
    let sparse = SchemaBuilder::new()
        .add_struct(
            "Wide",
            vec![
                (1, "a", Dt::U8),
                (2, "b", Dt::U64),
                (3, "c", Dt::U16),
                (4, "d", Dt::Str),
                (5, "e", Dt::U32),
                (6, "f", Dt::Bool),
                (7, "g", Dt::F64),
                (8, "h", Dt::U8),
            ],
        )
        .build("Wide")
        .unwrap();
    assert_ne!(packed.id(), sparse.id());
}

#[test]
fn packed_field_limit_enforced() {
    let mut fields = Vec::new();
    for i in 0..65u16 {
        // leak-free: names live for the builder call only via format storage
        fields.push((
            i + 1,
            Box::leak(format!("f{i}").into_boxed_str()) as &str,
            Dt::U8,
        ));
    }
    let result = SchemaBuilder::new()
        .add_packed_struct("Big", fields)
        .build("Big");
    assert!(matches!(result, Err(Error::BadSchema(_))));
}

#[test]
fn codegen_rejects_packed() {
    let schema = wide_schema();
    assert!(verit::codegen::generate_rust(&schema).is_err());
}