#![cfg(feature = "schema")]

use armour_typ::scalar::ScalarTyp;
use armour_typ::typ::{Fields, StructType, Typ};
use armour_typ::{GetType, SchemaFields, SchemaTyp};
use rapira::schema::{ShapeRel, shape_compat, skip_value};
use rapira::{PeerRead, Rapira, RapiraError, serialize};
use std::collections::{BTreeMap, HashMap};

// Local (old) struct: single field `a: u32`.
#[derive(Debug, PartialEq, Default)]
struct OldMsg {
    a: u32,
}

impl GetType for OldMsg {
    const TYPE: Typ = Typ::Struct(StructType {
        name: "Msg",
        fields: Fields::Named(&[("a", Typ::Scalar(ScalarTyp::U32))]),
    });
}

// Minimal Rapira impl for the test.
impl Rapira for OldMsg {
    const MIN_SIZE: usize = 4;
    fn size(&self) -> usize {
        4
    }
    fn check_bytes(slice: &mut &[u8]) -> rapira::Result<()> {
        u32::check_bytes(slice)
    }
    fn from_slice(slice: &mut &[u8]) -> rapira::Result<Self> {
        Ok(OldMsg {
            a: u32::from_slice(slice)?,
        })
    }
    fn convert_to_bytes(&self, slice: &mut [u8], cursor: &mut usize) {
        self.a.convert_to_bytes(slice, cursor);
    }
}

impl PeerRead for OldMsg {
    fn from_slice_with_peer_ctx(
        slice: &mut &[u8],
        peer: &SchemaTyp,
        flags: rapira::RapiraFlags,
    ) -> rapira::Result<Self>
    where
        Self: Sized,
    {
        let SchemaTyp::Struct {
            name,
            fields: SchemaFields::Named(pf),
        } = peer
        else {
            return Err(RapiraError::SchemaMismatch);
        };
        if name != "Msg" {
            return Err(RapiraError::SchemaMismatch);
        }
        let mut temp_slice = *slice;
        let mut out = OldMsg::default();
        for (pname, pty) in pf {
            match pname.as_str() {
                "a" => match shape_compat(&Typ::Scalar(ScalarTyp::U32), pty) {
                    ShapeRel::Exact => out.a = u32::from_slice_ctx(&mut temp_slice, flags)?,
                    _ => return Err(RapiraError::SchemaMismatch),
                },
                _ => skip_value(&mut temp_slice, pty)?,
            }
        }
        *slice = temp_slice;
        Ok(out)
    }
}

#[test]
fn new_sender_old_receiver_skips_extra_field() {
    // New sender: fields a:u32, b:String. Old receiver (OldMsg) only knows `a`.
    let peer = SchemaTyp::Struct {
        name: "Msg".into(),
        fields: SchemaFields::Named(vec![
            ("a".into(), SchemaTyp::Scalar(ScalarTyp::U32)),
            ("b".into(), SchemaTyp::Scalar(ScalarTyp::Str)),
        ]),
    };
    // New sender bytes: a=7, b="hi".
    let mut bytes = serialize(&7u32);
    bytes.extend_from_slice(&serialize(&String::from("hi")));

    let mut s = bytes.as_slice();
    let msg = <OldMsg as PeerRead>::from_slice_with_peer(&mut s, &peer).unwrap();
    assert_eq!(msg, OldMsg { a: 7 });
    assert!(s.is_empty(), "extra field b must be skipped via skip_value");
}

#[test]
fn mismatch_on_field_a_does_not_consume_slice() {
    let peer = SchemaTyp::Struct {
        name: "Msg".into(),
        fields: SchemaFields::Named(vec![("a".into(), SchemaTyp::Scalar(ScalarTyp::U64))]),
    };
    let bytes = serialize(&42u64);
    let mut s = bytes.as_slice();
    let len_before = s.len();
    let r = <OldMsg as PeerRead>::from_slice_with_peer(&mut s, &peer);
    assert!(matches!(r, Err(RapiraError::SchemaMismatch)));
    assert_eq!(s.len(), len_before);
}

#[test]
fn old_sender_new_receiver_defaults_missing() {
    // Peer (old sender) sends only `a`; receiver OldMsg also knows only `a`.
    let peer = SchemaTyp::Struct {
        name: "Msg".into(),
        fields: SchemaFields::Named(vec![("a".into(), SchemaTyp::Scalar(ScalarTyp::U32))]),
    };
    let bytes = serialize(&9u32);
    let mut s = bytes.as_slice();
    let msg = <OldMsg as PeerRead>::from_slice_with_peer(&mut s, &peer).unwrap();
    assert_eq!(msg, OldMsg { a: 9 });
}

#[derive(Debug, Default, PartialEq)]
struct Inner {
    x: u32,
}

impl GetType for Inner {
    const TYPE: Typ = Typ::Struct(StructType {
        name: "Inner",
        fields: Fields::Named(&[("x", Typ::Scalar(ScalarTyp::U32))]),
    });
}

impl Rapira for Inner {
    const MIN_SIZE: usize = 4;
    fn size(&self) -> usize {
        self.x.size()
    }
    fn check_bytes(slice: &mut &[u8]) -> rapira::Result<()> {
        u32::check_bytes(slice)
    }
    fn from_slice(slice: &mut &[u8]) -> rapira::Result<Self> {
        Ok(Inner {
            x: u32::from_slice(slice)?,
        })
    }
    fn convert_to_bytes(&self, slice: &mut [u8], cursor: &mut usize) {
        self.x.convert_to_bytes(slice, cursor);
    }
}

impl PeerRead for Inner {
    fn from_slice_with_peer_ctx(
        slice: &mut &[u8],
        peer: &SchemaTyp,
        flags: rapira::RapiraFlags,
    ) -> rapira::Result<Self> {
        let SchemaTyp::Struct {
            name,
            fields: SchemaFields::Named(fields),
        } = peer
        else {
            return Err(RapiraError::SchemaMismatch);
        };

        if name != "Inner" {
            return Err(RapiraError::SchemaMismatch);
        }

        let mut out = Inner::default();
        for (field, ty) in fields {
            match field.as_str() {
                "x" => {
                    if matches!(
                        shape_compat(&Typ::Scalar(ScalarTyp::U32), ty),
                        ShapeRel::Exact
                    ) {
                        out.x = u32::from_slice_ctx(slice, flags)?;
                    } else {
                        return Err(RapiraError::SchemaMismatch);
                    }
                }
                _ => skip_value(slice, ty)?,
            }
        }

        Ok(out)
    }
}

#[test]
fn vec_peer_read_reconciles_each_compatible_element() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Struct {
        name: "Inner".to_string(),
        fields: SchemaFields::Named(vec![
            ("x".to_string(), SchemaTyp::Scalar(ScalarTyp::U32)),
            ("extra".to_string(), SchemaTyp::Scalar(ScalarTyp::Str)),
        ]),
    }));
    let mut bytes = serialize(&2u32);
    bytes.extend_from_slice(&serialize(&1u32));
    bytes.extend_from_slice(&serialize(&"a".to_string()));
    bytes.extend_from_slice(&serialize(&2u32));
    bytes.extend_from_slice(&serialize(&"b".to_string()));
    let mut slice = bytes.as_slice();

    let got = <Vec<Inner> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap();
    assert_eq!(got, vec![Inner { x: 1 }, Inner { x: 2 }]);
    assert!(slice.is_empty());
}

#[test]
fn vec_peer_read_leaf_mismatch_is_schema_mismatch() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U64)));
    let bytes = serialize(&vec![1u64]);
    let mut slice = bytes.as_slice();

    let err = <Vec<u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();
    assert!(matches!(err, RapiraError::SchemaMismatch));
}

#[test]
fn option_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Optional(Box::new(SchemaTyp::Scalar(ScalarTyp::U64)));
    let mut bytes = vec![1u8];
    bytes.extend_from_slice(&serialize(&42u64));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <Option<u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn option_peer_read_none_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Optional(Box::new(SchemaTyp::Scalar(ScalarTyp::U64)));
    let bytes = vec![0u8];
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <Option<u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple2_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple2".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U64),
        ]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u32));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <(u32, u32) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple2_peer_read_missing_fields_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple2".into(),
        fields: SchemaFields::Unnamed(vec![SchemaTyp::Scalar(ScalarTyp::U32)]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u16));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <(u32, u16) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple3_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple3".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U64),
        ]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u32));
    bytes.extend_from_slice(&serialize(&3u32));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <(u32, u32, u32) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple3_peer_read_missing_fields_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple3".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U16),
        ]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u16));
    bytes.extend_from_slice(&serialize(&true));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <(u32, u16, bool) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple4_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple4".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U64),
        ]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u32));
    bytes.extend_from_slice(&serialize(&3u32));
    bytes.extend_from_slice(&serialize(&4u32));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err =
        <(u32, u32, u32, u32) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn tuple4_peer_read_missing_fields_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Struct {
        name: "Tuple4".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U16),
            SchemaTyp::Scalar(ScalarTyp::Bool),
        ]),
    };
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&2u16));
    bytes.extend_from_slice(&serialize(&true));
    bytes.extend_from_slice(&serialize(&4u64));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err =
        <(u32, u16, bool, u64) as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn btree_map_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Struct {
        name: "Pair".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U64),
        ]),
    }));
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&10u32));
    bytes.extend_from_slice(&serialize(&20u32));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err =
        <BTreeMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn btree_map_peer_read_invalid_pair_shape_rejects_before_reading_len() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U32)));
    let bytes = Vec::new();
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err =
        <BTreeMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn btree_map_peer_read_pair_with_missing_value_rejects_empty_payload() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Struct {
        name: "Pair".into(),
        fields: SchemaFields::Unnamed(vec![SchemaTyp::Scalar(ScalarTyp::U32)]),
    }));
    let bytes = serialize(&0u32);
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err =
        <BTreeMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn hash_map_peer_read_mismatch_is_schema_mismatch_without_consuming_slice() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Struct {
        name: "Pair".into(),
        fields: SchemaFields::Unnamed(vec![
            SchemaTyp::Scalar(ScalarTyp::U32),
            SchemaTyp::Scalar(ScalarTyp::U64),
        ]),
    }));
    let mut bytes = serialize(&1u32);
    bytes.extend_from_slice(&serialize(&10u32));
    bytes.extend_from_slice(&serialize(&20u32));
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <HashMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn hash_map_peer_read_invalid_pair_shape_rejects_before_reading_len() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U32)));
    let bytes = Vec::new();
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <HashMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn hash_map_peer_read_pair_with_missing_value_rejects_empty_payload() {
    let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Struct {
        name: "Pair".into(),
        fields: SchemaFields::Unnamed(vec![SchemaTyp::Scalar(ScalarTyp::U32)]),
    }));
    let bytes = serialize(&0u32);
    let len_before = bytes.len();
    let mut slice = bytes.as_slice();

    let err = <HashMap<u32, u32> as PeerRead>::from_slice_with_peer(&mut slice, &peer).unwrap_err();

    assert!(matches!(err, RapiraError::SchemaMismatch));
    assert_eq!(slice.len(), len_before);
}

#[test]
fn deserialize_with_peer_uses_peerread_entrypoint() {
    let peer = SchemaTyp::Struct {
        name: "Msg".into(),
        fields: SchemaFields::Named(vec![
            ("a".into(), SchemaTyp::Scalar(ScalarTyp::U32)),
            ("extra".into(), SchemaTyp::Scalar(ScalarTyp::Str)),
        ]),
    };
    let mut bytes = serialize(&11u32);
    bytes.extend_from_slice(&serialize(&"ignored".to_string()));

    let msg: OldMsg = rapira::deserialize_with_peer(&bytes, &peer).unwrap();
    assert_eq!(msg, OldMsg { a: 11 });
}
