use core_types::{Timestamp, TransportDomain};
use data_model::{
    compat::CompatibilityPolicy,
    contract::{MissionContract, ResourceContract, ResourceKind, ServiceContract, TopicContract},
    envelope::{ControlEnvelope, DataEnvelope, DataPayload, Packet, PacketHeader},
    registry::{
        CodecDescriptor, CodecRegistry, ContractCodecBinding, ContractCodecRegistry,
        SchemaRegistry, SimpleCodecRegistry, SimpleContractCodecRegistry, 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),
        trace_context: None,
    }
}

// ─── 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");
    assert!(reg.is_compatible(
        &id,
        SchemaVersion(2),
        CompatibilityPolicy::BackwardCompatible
    ));
    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");
    assert!(reg.is_compatible(
        &id,
        SchemaVersion(3),
        CompatibilityPolicy::ForwardCompatible
    ));
    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());
}

#[test]
fn contract_compatibility_exact_requires_same_contract() {
    let registered = ResourceContract::Service(ServiceContract {
        request: SchemaDescriptor::new("svc.req", 1, "SvcRequest"),
        response: SchemaDescriptor::new("svc.res", 1, "SvcResponse"),
        encoding_hint: Some(String::from("json")),
    });
    let candidate = ResourceContract::Service(ServiceContract {
        request: SchemaDescriptor::new("svc.req", 1, "SvcRequest"),
        response: SchemaDescriptor::new("svc.res", 2, "SvcResponse"),
        encoding_hint: Some(String::from("json")),
    });

    assert!(!registered.is_compatible_with(&candidate, CompatibilityPolicy::Exact));
    assert!(registered.is_compatible_with(&registered, CompatibilityPolicy::Exact));
}

#[test]
fn contract_compatibility_backward_allows_newer_candidate_version() {
    let registered = ResourceContract::Topic(TopicContract {
        message: SchemaDescriptor::new("topic.msg", 3, "TopicMessage"),
        encoding_hint: Some(String::from("json")),
    });
    let candidate = ResourceContract::Topic(TopicContract {
        message: SchemaDescriptor::new("topic.msg", 2, "TopicMessage"),
        encoding_hint: Some(String::from("json")),
    });

    assert!(registered.is_compatible_with(&candidate, CompatibilityPolicy::BackwardCompatible));
}

#[test]
fn contract_compatibility_rejects_mismatched_optional_feedback_shape() {
    let registered = ResourceContract::Mission(MissionContract {
        open: SchemaDescriptor::new("mission.open", 1, "MissionOpen"),
        command: SchemaDescriptor::new("mission.command", 1, "MissionCommand"),
        ack: SchemaDescriptor::new("mission.ack", 1, "MissionAck"),
        update: SchemaDescriptor::new("mission.update", 1, "MissionUpdate"),
        feedback: None,
        encoding_hint: Some(String::from("json")),
    });
    let candidate = ResourceContract::Mission(MissionContract {
        open: SchemaDescriptor::new("mission.open", 1, "MissionOpen"),
        command: SchemaDescriptor::new("mission.command", 1, "MissionCommand"),
        ack: SchemaDescriptor::new("mission.ack", 1, "MissionAck"),
        update: SchemaDescriptor::new("mission.update", 1, "MissionUpdate"),
        feedback: Some(SchemaDescriptor::new("mission.feedback", 1, "MissionFeedback")),
        encoding_hint: Some(String::from("json")),
    });

    assert!(!registered.is_compatible_with(&candidate, CompatibilityPolicy::Exact));
}

#[test]
fn contract_codec_binding_register_and_get() {
    let mut reg = SimpleContractCodecRegistry::default();
    let binding = ContractCodecBinding {
        resource_kind: ResourceKind::Action,
        contract_hash: String::from("deadbeefcafebabe"),
        codec_family: String::from("serde_json"),
        codec_name: String::from("action.serde_json.v1"),
    };
    reg.register(binding.clone()).expect("binding register must succeed");
    let found = reg.get(ResourceKind::Action, "deadbeefcafebabe");
    assert_eq!(found, Some(&binding));
}

#[test]
fn contract_codec_binding_duplicate_register_fails() {
    let mut reg = SimpleContractCodecRegistry::default();
    reg.register(ContractCodecBinding {
        resource_kind: ResourceKind::Topic,
        contract_hash: String::from("hash-1"),
        codec_family: String::from("json"),
        codec_name: String::from("topic.json.v1"),
    })
    .unwrap();
    let result = reg.register(ContractCodecBinding {
        resource_kind: ResourceKind::Topic,
        contract_hash: String::from("hash-1"),
        codec_family: String::from("msgpack"),
        codec_name: String::from("topic.msgpack.v1"),
    });
    assert!(result.is_err());
}

// ─── 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".as_slice().into()),
    };
    let DataPayload::Inline(bytes) = &env.payload else {
        panic!("expected inline")
    };
    assert_eq!(bytes.as_ref(), 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].into()),
    });

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