robotrt-data-model 0.1.0-beta.2

use core_types::{Timestamp, TransportDomain};
use data_model::{
    compat::CompatibilityPolicy,
    envelope::{ControlEnvelope, DataEnvelope, DataPayload, Packet, PacketHeader},
    registry::{
        CodecDescriptor, CodecRegistry, SchemaRegistry, SimpleCodecRegistry, SimpleSchemaRegistry,
    },
    schema::{SchemaDescriptor, SchemaId, SchemaVersion},
};

fn schema_v(ver: u16) -> SchemaDescriptor {
    SchemaDescriptor::new("test.msg", ver, "TestMessage")
}

fn schema_id() -> SchemaId {
    SchemaId::new("test.msg")
}

fn base_header(seq: u64) -> PacketHeader {
    PacketHeader {
        version: 1,
        domain: TransportDomain::Local,
        session_id: None,
        stream_id: None,
        sequence: seq,
        ack: None,
        timestamp: Timestamp::now(),
        schema_id: schema_id(),
        schema_version: SchemaVersion(1),
    }
}

// ─── SchemaRegistry tests ─────────────────────────────────────────────────────

#[test]
fn schema_register_and_get() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(schema_v(1))
        .expect("first registration must succeed");
    let found = reg.get(&schema_id());
    assert!(found.is_some());
    assert_eq!(found.unwrap().version, SchemaVersion(1));
}

#[test]
fn schema_duplicate_register_fails() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(schema_v(1)).unwrap();
    let result = reg.register(schema_v(2)); // same schema_id
    assert!(result.is_err(), "duplicate registration must fail");
}

#[test]
fn schema_get_unknown_returns_none() {
    let reg = SimpleSchemaRegistry::default();
    assert!(reg.get(&SchemaId::new("not.registered")).is_none());
}

#[test]
fn schema_compatibility_exact_same_version() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(schema_v(2)).unwrap();
    assert!(
        reg.is_compatible(&schema_id(), SchemaVersion(2), CompatibilityPolicy::Exact),
        "same version must be exact-compatible"
    );
}

#[test]
fn schema_compatibility_exact_rejects_different_version() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(schema_v(2)).unwrap();
    assert!(
        !reg.is_compatible(&schema_id(), SchemaVersion(3), CompatibilityPolicy::Exact),
        "different version must fail exact check"
    );
}

#[test]
fn schema_compatibility_backward_allows_newer_writer() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(SchemaDescriptor::new("msg.b", 3, "B"))
        .unwrap();
    let id = SchemaId::new("msg.b");
    // writer=3, reader=2 → backward: writer >= reader → ok
    assert!(reg.is_compatible(
        &id,
        SchemaVersion(2),
        CompatibilityPolicy::BackwardCompatible
    ));
    // writer=3, reader=4 → backward: writer < reader → fail
    assert!(!reg.is_compatible(
        &id,
        SchemaVersion(4),
        CompatibilityPolicy::BackwardCompatible
    ));
}

#[test]
fn schema_compatibility_forward_allows_older_writer() {
    let mut reg = SimpleSchemaRegistry::default();
    reg.register(SchemaDescriptor::new("msg.f", 2, "F"))
        .unwrap();
    let id = SchemaId::new("msg.f");
    // writer=2, reader=3 → forward: writer <= reader → ok
    assert!(reg.is_compatible(
        &id,
        SchemaVersion(3),
        CompatibilityPolicy::ForwardCompatible
    ));
    // writer=2, reader=1 → forward: writer > reader → fail
    assert!(!reg.is_compatible(
        &id,
        SchemaVersion(1),
        CompatibilityPolicy::ForwardCompatible
    ));
}

#[test]
fn schema_compatibility_unknown_id_returns_false() {
    let reg = SimpleSchemaRegistry::default();
    assert!(!reg.is_compatible(
        &SchemaId::new("nope"),
        SchemaVersion(1),
        CompatibilityPolicy::Exact
    ));
}

// ─── CodecRegistry tests ──────────────────────────────────────────────────────

fn make_codec(schema: &str, zero_copy: bool) -> CodecDescriptor {
    CodecDescriptor {
        schema_id: SchemaId::new(schema),
        name: format!("{schema}-codec"),
        zero_copy,
    }
}

#[test]
fn codec_register_and_get() {
    let mut reg = SimpleCodecRegistry::default();
    reg.register(make_codec("img.raw", true))
        .expect("first registration must succeed");
    let found = reg.get(&SchemaId::new("img.raw"));
    assert!(found.is_some());
    assert!(found.unwrap().zero_copy);
}

#[test]
fn codec_duplicate_register_fails() {
    let mut reg = SimpleCodecRegistry::default();
    reg.register(make_codec("img.raw", true)).unwrap();
    let result = reg.register(make_codec("img.raw", false));
    assert!(result.is_err());
}

#[test]
fn codec_get_unknown_returns_none() {
    let reg = SimpleCodecRegistry::default();
    assert!(reg.get(&SchemaId::new("unknown")).is_none());
}

// ─── Packet / Envelope tests ──────────────────────────────────────────────────

#[test]
fn packet_header_fields_round_trip() {
    let h = base_header(42);
    assert_eq!(h.version, 1);
    assert_eq!(h.sequence, 42);
    assert_eq!(h.domain, TransportDomain::Local);
    assert_eq!(h.schema_id, schema_id());
    assert_eq!(h.schema_version, SchemaVersion(1));
}

#[test]
fn control_envelope_label_preserved() {
    let env = ControlEnvelope {
        header: base_header(1),
        label: "mission.start".into(),
        payload: b"{}".to_vec(),
    };
    assert_eq!(env.label, "mission.start");
    assert_eq!(env.payload, b"{}");
}

#[test]
fn data_envelope_inline_payload() {
    let env = DataEnvelope {
        header: base_header(2),
        payload: DataPayload::Inline(b"hello".to_vec()),
    };
    let DataPayload::Inline(bytes) = &env.payload else {
        panic!("expected inline")
    };
    assert_eq!(bytes, b"hello");
}

#[test]
fn packet_variant_matching() {
    let ctrl = Packet::Control(ControlEnvelope {
        header: base_header(0),
        label: "ping".into(),
        payload: Vec::new(),
    });
    let data = Packet::Data(DataEnvelope {
        header: base_header(1),
        payload: DataPayload::Inline(vec![0xFFu8]),
    });

    assert!(matches!(ctrl, Packet::Control(_)));
    assert!(matches!(data, Packet::Data(_)));
}