rapira 0.13.6

use armour_typ::scalar::ScalarTyp;
use armour_typ::typ::{Fields, Typ};
use armour_typ::{SchemaFields, SchemaTyp};

use crate::{Rapira, RapiraError, Result, byte_rapira};

/// Result of deep comparison between a local schema (`Typ`) and a peer schema (`SchemaTyp`).
#[derive(PartialEq, Eq, Clone, Copy, Debug)]
pub enum ShapeRel {
    /// Identical across the whole subtree → read with ordinary `from_slice`.
    Exact,
    /// Struct/enum evolution somewhere inside → reconciliation required.
    Compatible,
    /// Leaf/form/identity change → `SchemaMismatch`.
    Mismatch,
}

/// Deep data-only schema comparison (no `Rapira` / `GetType` on elements).
pub fn shape_compat(local: &Typ, peer: &SchemaTyp) -> ShapeRel {
    use ShapeRel::*;
    match (local, peer) {
        (Typ::Scalar(a), SchemaTyp::Scalar(b)) => {
            if a == b {
                Exact
            } else {
                Mismatch
            }
        }
        (Typ::Vec(li), SchemaTyp::Vec(pi)) => shape_compat(li, pi),
        (Typ::Optional(li), SchemaTyp::Optional(pi)) => shape_compat(li, pi),
        (Typ::Array(ln, li), SchemaTyp::Array(pn, pi)) => {
            if ln != pn {
                Mismatch
            } else {
                shape_compat(li, pi)
            }
        }
        (Typ::Struct(ls), SchemaTyp::Struct { name, fields }) => {
            if ls.name != name.as_str() {
                return Mismatch;
            }
            compat_fields(&ls.fields, fields)
        }
        (Typ::Enum(le), SchemaTyp::Enum { name, variants }) => {
            if le.name != name.as_str() {
                return Mismatch;
            }
            compat_enum(le.variants, variants)
        }
        (Typ::SimpleEnum(le), SchemaTyp::SimpleEnum { name, variants }) => {
            if le.name != name.as_str() {
                return Mismatch;
            }
            compat_simple_enum(le.variants, variants)
        }
        (Typ::Custom(ln, _), SchemaTyp::Custom(pn, _)) => {
            if ln == &pn.as_str() {
                Exact
            } else {
                Mismatch
            }
        }
        _ => Mismatch,
    }
}

/// Compare struct fields (named — by name; unnamed — by position).
fn compat_fields(local: &Fields, peer: &SchemaFields) -> ShapeRel {
    use ShapeRel::*;
    match (local, peer) {
        (Fields::Named(lf), SchemaFields::Named(pf)) => {
            let mut worst = Exact;
            let mut matched = 0usize;
            for (pname, pty) in pf.iter() {
                match lf.iter().find(|(ln, _)| ln == pname) {
                    Some((_, lty)) => {
                        matched += 1;
                        worst = worse(worst, shape_compat(lty, pty));
                    }
                    None => worst = worse(worst, Compatible),
                }
            }
            if matched != lf.len() {
                worst = worse(worst, Compatible);
            }
            if worst == Exact && !same_named_order(lf, pf) {
                worst = Compatible;
            }
            worst
        }
        (Fields::Unnamed(lf), SchemaFields::Unnamed(pf)) => {
            let mut worst = Exact;
            let n = lf.len().min(pf.len());
            for i in 0..n {
                worst = worse(worst, shape_compat(&lf[i], &pf[i]));
            }
            if lf.len() != pf.len() {
                worst = worse(worst, Compatible);
            }
            worst
        }
        _ => Mismatch,
    }
}

fn same_named_order(lf: &[(&'static str, Typ)], pf: &[(String, SchemaTyp)]) -> bool {
    lf.len() == pf.len() && lf.iter().zip(pf).all(|((ln, _), (pn, _))| ln == pn)
}

fn compat_simple_enum(local: &[(u8, &str)], peer: &[(u8, String)]) -> ShapeRel {
    use ShapeRel::*;
    let mut worst = Exact;
    for (ptag, pname) in peer.iter() {
        match local.iter().find(|(lt, _)| lt == ptag) {
            Some((_, lname)) => {
                if lname != pname {
                    return Mismatch;
                }
            }
            None => worst = worse(worst, Compatible),
        }
    }
    if local.len() != peer.len() && worst != Mismatch {
        worst = Compatible;
    }
    worst
}

fn compat_enum(local: &[(u8, (&'static str, Typ))], peer: &[(u8, String, SchemaTyp)]) -> ShapeRel {
    use ShapeRel::*;
    let mut worst = Exact;
    for (ptag, pname, pty) in peer.iter() {
        match local.iter().find(|(lt, _)| lt == ptag) {
            Some((_, (lname, lty))) => {
                if lname != pname {
                    return Mismatch;
                }
                worst = worse(worst, shape_compat(lty, pty));
            }
            None => worst = worse(worst, Compatible),
        }
    }
    if local.len() != peer.len() && worst != Mismatch {
        worst = Compatible;
    }
    worst
}

/// Skips a value described by `ty` without knowing the Rust type.
/// Reads rapira wire format: u32 length prefixes, 1-byte discriminants.
#[inline]
pub fn skip_value(slice: &mut &[u8], ty: &SchemaTyp) -> Result<()> {
    match ty {
        SchemaTyp::Scalar(s) => skip_scalar(slice, *s),
        SchemaTyp::Optional(inner) => {
            let b = byte_rapira::from_slice(slice)?;
            if b != 0 {
                skip_value(slice, inner)?;
            }
            Ok(())
        }
        SchemaTyp::Vec(inner) => {
            let len = u32::from_slice(slice)? as usize;
            for _ in 0..len {
                skip_value(slice, inner)?;
            }
            Ok(())
        }
        SchemaTyp::Array(n, inner) => {
            for _ in 0..*n {
                skip_value(slice, inner)?;
            }
            Ok(())
        }
        SchemaTyp::Struct { fields, .. } => match fields {
            SchemaFields::Named(fs) => {
                for (_, fty) in fs {
                    skip_value(slice, fty)?;
                }
                Ok(())
            }
            SchemaFields::Unnamed(fs) => {
                for fty in fs {
                    skip_value(slice, fty)?;
                }
                Ok(())
            }
        },
        SchemaTyp::Enum { variants, .. } => {
            let tag = byte_rapira::from_slice(slice)?;
            let (_, _, payload) = variants
                .iter()
                .find(|(t, _, _)| *t == tag)
                .ok_or(RapiraError::UnknownVariant)?;
            skip_value(slice, payload)
        }
        SchemaTyp::SimpleEnum { .. } => {
            let _ = byte_rapira::from_slice(slice)?;
            Ok(())
        }
        SchemaTyp::Custom(..) => Err(RapiraError::UnsupportedSchema),
    }
}

#[inline]
fn skip_scalar(slice: &mut &[u8], s: ScalarTyp) -> Result<()> {
    use ScalarTyp::*;
    match s {
        Bool | U8 => advance(slice, 1),
        U16 => advance(slice, 2),
        U32 | I32 | F32 | Id32 => advance(slice, 4),
        U64 | I64 | F64 | Datetime | Timestamp | Id64 | Fuid | LowId => advance(slice, 8),
        Decimal => advance(slice, 16),
        Void => Ok(()),
        ArrayBytes(n) => advance(slice, n as usize),
        // Length-prefixed (u32 LEN_SIZE + bytes). `JsonBytes` is a raw byte vector
        // holding a JSON string, so it skips like `Bytes`.
        Str | Bytes | JsonBytes => {
            let len = u32::from_slice(slice)? as usize;
            advance(slice, len)
        }
        RustJson => skip_rust_json(slice),
    }
}

#[inline]
fn skip_rust_json(slice: &mut &[u8]) -> Result<()> {
    let tag = byte_rapira::from_slice(slice)?;
    match tag {
        0 => Ok(()),
        1 => advance(slice, 1),
        2 => {
            let number_tag = byte_rapira::from_slice(slice)?;
            match number_tag {
                0..=2 => advance(slice, 8),
                _ => Err(RapiraError::EnumVariant),
            }
        }
        3 => {
            let len = u32::from_slice(slice)? as usize;
            advance(slice, len)
        }
        4 => {
            let len = u32::from_slice(slice)? as usize;
            for _ in 0..len {
                skip_rust_json(slice)?;
            }
            Ok(())
        }
        5 => {
            let len = u32::from_slice(slice)? as usize;
            for _ in 0..len {
                let key_len = u32::from_slice(slice)? as usize;
                advance(slice, key_len)?;
                skip_rust_json(slice)?;
            }
            Ok(())
        }
        _ => Err(RapiraError::EnumVariant),
    }
}

#[inline]
fn advance(slice: &mut &[u8], n: usize) -> Result<()> {
    if slice.len() < n {
        return Err(RapiraError::SliceLen);
    }
    *slice = &slice[n..];
    Ok(())
}

#[inline]
fn worse(a: ShapeRel, b: ShapeRel) -> ShapeRel {
    use ShapeRel::*;
    match (a, b) {
        (Mismatch, _) | (_, Mismatch) => Mismatch,
        (Compatible, _) | (_, Compatible) => Compatible,
        _ => Exact,
    }
}

#[cfg(test)]
mod shape_tests {
    use super::*;
    use armour_typ::scalar::ScalarTyp;
    use armour_typ::typ::{Fields, StructType, Typ};
    use armour_typ::{SchemaFields, SchemaTyp};

    #[test]
    fn scalar_exact_and_mismatch() {
        assert_eq!(
            shape_compat(
                &Typ::Scalar(ScalarTyp::U32),
                &SchemaTyp::Scalar(ScalarTyp::U32)
            ),
            ShapeRel::Exact
        );
        assert_eq!(
            shape_compat(
                &Typ::Scalar(ScalarTyp::U32),
                &SchemaTyp::Scalar(ScalarTyp::U64)
            ),
            ShapeRel::Mismatch
        );
    }

    #[test]
    fn vec_recurses_to_inner_scalar() {
        let local = Typ::Vec(&Typ::Scalar(ScalarTyp::U32));
        let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U64)));
        assert_eq!(shape_compat(&local, &peer), ShapeRel::Mismatch);
    }

    #[test]
    fn struct_added_field_is_compatible() {
        static LOCAL: Typ = Typ::Struct(StructType {
            name: "S",
            fields: Fields::Named(&[("a", Typ::Scalar(ScalarTyp::U64))]),
        });
        let peer = SchemaTyp::Struct {
            name: "S".into(),
            fields: SchemaFields::Named(vec![
                ("a".into(), SchemaTyp::Scalar(ScalarTyp::U64)),
                ("b".into(), SchemaTyp::Scalar(ScalarTyp::Str)),
            ]),
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Compatible);
    }

    #[test]
    fn struct_identical_is_exact_and_renamed_is_mismatch() {
        static LOCAL: Typ = Typ::Struct(StructType {
            name: "S",
            fields: Fields::Named(&[("a", Typ::Scalar(ScalarTyp::U64))]),
        });
        let same = SchemaTyp::Struct {
            name: "S".into(),
            fields: SchemaFields::Named(vec![("a".into(), SchemaTyp::Scalar(ScalarTyp::U64))]),
        };
        assert_eq!(shape_compat(&LOCAL, &same), ShapeRel::Exact);
        let renamed = SchemaTyp::Struct {
            name: "T".into(),
            fields: SchemaFields::Named(vec![("a".into(), SchemaTyp::Scalar(ScalarTyp::U64))]),
        };
        assert_eq!(shape_compat(&LOCAL, &renamed), ShapeRel::Mismatch);
    }

    #[test]
    fn simple_enum_same_variants_is_exact() {
        use armour_typ::typ::SimpleEnumType;
        static LOCAL: Typ = Typ::SimpleEnum(SimpleEnumType {
            name: "E",
            variants: &[(0, "A"), (1, "B")],
        });
        let peer = SchemaTyp::SimpleEnum {
            name: "E".into(),
            variants: vec![(0, "A".into()), (1, "B".into())],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Exact);
    }

    #[test]
    fn simple_enum_renamed_variant_is_mismatch() {
        use armour_typ::typ::SimpleEnumType;
        static LOCAL: Typ = Typ::SimpleEnum(SimpleEnumType {
            name: "E",
            variants: &[(0, "A"), (1, "B")],
        });
        let peer = SchemaTyp::SimpleEnum {
            name: "E".into(),
            variants: vec![(0, "A".into()), (1, "C".into())],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Mismatch);
    }

    #[test]
    fn simple_enum_append_only_is_compatible() {
        use armour_typ::typ::SimpleEnumType;
        static LOCAL: Typ = Typ::SimpleEnum(SimpleEnumType {
            name: "E",
            variants: &[(0, "A")],
        });
        let peer = SchemaTyp::SimpleEnum {
            name: "E".into(),
            variants: vec![(0, "A".into()), (1, "B".into())],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Compatible);
    }

    use armour_typ::typ::EnumType;

    #[test]
    fn enum_identical_is_exact() {
        static LOCAL: Typ = Typ::Enum(EnumType {
            name: "E",
            variants: &[(0, ("A", Typ::Scalar(ScalarTyp::U32)))],
        });
        let peer = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![(0, "A".into(), SchemaTyp::Scalar(ScalarTyp::U32))],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Exact);
    }

    #[test]
    fn enum_appended_variant_is_compatible() {
        // append-only: peer has an extra variant at a new tag (R: forward-compat)
        static LOCAL: Typ = Typ::Enum(EnumType {
            name: "E",
            variants: &[(0, ("A", Typ::Scalar(ScalarTyp::U32)))],
        });
        let peer = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![
                (0, "A".into(), SchemaTyp::Scalar(ScalarTyp::U32)),
                (1, "B".into(), SchemaTyp::Scalar(ScalarTyp::Bool)),
            ],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Compatible);
    }

    #[test]
    fn enum_renamed_variant_same_tag_is_mismatch() {
        static LOCAL: Typ = Typ::Enum(EnumType {
            name: "E",
            variants: &[(0, ("A", Typ::Scalar(ScalarTyp::U32)))],
        });
        let peer = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![(0, "Z".into(), SchemaTyp::Scalar(ScalarTyp::U32))],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Mismatch);
    }

    #[test]
    fn enum_changed_payload_is_mismatch() {
        static LOCAL: Typ = Typ::Enum(EnumType {
            name: "E",
            variants: &[(0, ("A", Typ::Scalar(ScalarTyp::U32)))],
        });
        let peer = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![(0, "A".into(), SchemaTyp::Scalar(ScalarTyp::U64))],
        };
        assert_eq!(shape_compat(&LOCAL, &peer), ShapeRel::Mismatch);
    }
}

#[cfg(test)]
mod skip_tests {
    use super::*;
    use crate::serialize;
    use armour_typ::SchemaTyp;
    use armour_typ::scalar::ScalarTyp;

    #[test]
    fn skip_scalar_u32_advances_4() {
        let bytes = serialize(&123u32);
        let mut s = bytes.as_slice();
        skip_value(&mut s, &SchemaTyp::Scalar(ScalarTyp::U32)).unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_str_advances_len_prefixed() {
        let bytes = serialize(&String::from("hello"));
        let mut s = bytes.as_slice();
        skip_value(&mut s, &SchemaTyp::Scalar(ScalarTyp::Str)).unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_vec_of_u32() {
        let bytes = serialize(&vec![1u32, 2, 3]);
        let mut s = bytes.as_slice();
        skip_value(
            &mut s,
            &SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U32))),
        )
        .unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_option_some_and_none() {
        let some = serialize(&Some(7u32));
        let mut s = some.as_slice();
        skip_value(
            &mut s,
            &SchemaTyp::Optional(Box::new(SchemaTyp::Scalar(ScalarTyp::U32))),
        )
        .unwrap();
        assert!(s.is_empty());

        let none = serialize(&Option::<u32>::None);
        let mut s = none.as_slice();
        skip_value(
            &mut s,
            &SchemaTyp::Optional(Box::new(SchemaTyp::Scalar(ScalarTyp::U32))),
        )
        .unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_custom_is_unsupported() {
        let mut s: &[u8] = &[];
        let err = skip_value(&mut s, &SchemaTyp::Custom("x".into(), vec![]));
        assert!(matches!(err, Err(crate::RapiraError::UnsupportedSchema)));
    }

    #[test]
    fn skip_rustjson_nested_value_advances_without_materializing() {
        use serde_json::json;

        let value = json!({
            "items": [
                null,
                true,
                42_u64,
                -7_i64,
                1.5_f64,
                "hello",
                {"inner": ["x", "y"]}
            ]
        });
        let bytes = crate::serialize(&value);
        let mut slice = bytes.as_slice();

        skip_value(&mut slice, &SchemaTyp::Scalar(ScalarTyp::RustJson)).unwrap();
        assert!(slice.is_empty());
    }

    #[test]
    fn skip_rustjson_invalid_string_payload_does_not_validate_utf8() {
        let mut bytes = Vec::new();
        bytes.push(3); // Value::String
        bytes.extend_from_slice(&2u32.to_le_bytes());
        bytes.extend_from_slice(&[0xff, 0xff]);
        let mut slice = bytes.as_slice();

        skip_value(&mut slice, &SchemaTyp::Scalar(ScalarTyp::RustJson)).unwrap();
        assert!(slice.is_empty());
    }

    #[test]
    fn skip_rustjson_nan_number_does_not_validate_number() {
        let mut bytes = Vec::new();
        bytes.push(2); // Value::Number
        bytes.push(2); // f64 branch
        bytes.extend_from_slice(&f64::NAN.to_le_bytes());
        let mut slice = bytes.as_slice();

        skip_value(&mut slice, &SchemaTyp::Scalar(ScalarTyp::RustJson)).unwrap();
        assert!(slice.is_empty());
    }

    #[test]
    fn skip_array_of_u32() {
        // Array(3, U32) skips 3 contiguous u32 (no length prefix).
        let mut bytes = serialize(&1u32);
        bytes.extend_from_slice(&serialize(&2u32));
        bytes.extend_from_slice(&serialize(&3u32));
        let mut s = bytes.as_slice();
        skip_value(
            &mut s,
            &SchemaTyp::Array(3, Box::new(SchemaTyp::Scalar(ScalarTyp::U32))),
        )
        .unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_struct_named_then_reads_next() {
        // Wire: u32 (a) + String (b), followed by a raw trailing sentinel byte.
        let mut bytes = serialize(&7u32);
        bytes.extend_from_slice(&serialize(&String::from("hi")));
        bytes.push(0xAB); // raw sentinel (u8 uses byte_rapira, not Rapira)
        let mut s = bytes.as_slice();
        let st = SchemaTyp::Struct {
            name: "S".into(),
            fields: SchemaFields::Named(vec![
                ("a".into(), SchemaTyp::Scalar(ScalarTyp::U32)),
                ("b".into(), SchemaTyp::Scalar(ScalarTyp::Str)),
            ]),
        };
        skip_value(&mut s, &st).unwrap();
        // sentinel must remain, proving the struct skip landed exactly.
        assert_eq!(s, &[0xABu8][..]);
    }

    #[test]
    fn skip_struct_unnamed() {
        let mut bytes = serialize(&7u32);
        bytes.extend_from_slice(&serialize(&true));
        let mut s = bytes.as_slice();
        let st = SchemaTyp::Struct {
            name: "T".into(),
            fields: SchemaFields::Unnamed(vec![
                SchemaTyp::Scalar(ScalarTyp::U32),
                SchemaTyp::Scalar(ScalarTyp::Bool),
            ]),
        };
        skip_value(&mut s, &st).unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_simple_enum_advances_one_tag_byte() {
        let bytes = vec![3u8]; // a bare discriminant byte (raw)
        let mut s = bytes.as_slice();
        skip_value(
            &mut s,
            &SchemaTyp::SimpleEnum {
                name: "E".into(),
                variants: vec![(3, "D".into())],
            },
        )
        .unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_enum_reads_tag_then_payload() {
        // tag=1 (raw byte), payload=u32(99)
        let mut bytes = vec![1u8];
        bytes.extend_from_slice(&serialize(&99u32));
        let mut s = bytes.as_slice();
        let e = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![
                (0, "A".into(), SchemaTyp::Scalar(ScalarTyp::Bool)),
                (1, "B".into(), SchemaTyp::Scalar(ScalarTyp::U32)),
            ],
        };
        skip_value(&mut s, &e).unwrap();
        assert!(s.is_empty());
    }

    #[test]
    fn skip_enum_unknown_tag_errors() {
        let bytes = vec![9u8]; // tag 9 not in schema (raw byte)
        let mut s = bytes.as_slice();
        let e = SchemaTyp::Enum {
            name: "E".into(),
            variants: vec![(0, "A".into(), SchemaTyp::Scalar(ScalarTyp::Bool))],
        };
        let err = skip_value(&mut s, &e);
        assert!(matches!(err, Err(crate::RapiraError::UnknownVariant)));
    }
}

#[cfg(test)]
mod default_method_tests {
    use crate::{PeerRead, RapiraError, serialize};
    use armour_typ::SchemaTyp;
    use armour_typ::scalar::ScalarTyp;

    #[test]
    fn scalar_exact_reads() {
        let bytes = serialize(&42u32);
        let mut s = bytes.as_slice();
        let v = <u32 as PeerRead>::from_slice_with_peer(&mut s, &SchemaTyp::Scalar(ScalarTyp::U32))
            .unwrap();
        assert_eq!(v, 42);
    }

    #[test]
    fn scalar_mismatch_errors() {
        // R1: mismatch must not consume input bytes
        let bytes = serialize(&42u64);
        let mut s = bytes.as_slice();
        let r = <u32 as PeerRead>::from_slice_with_peer(&mut s, &SchemaTyp::Scalar(ScalarTyp::U64));
        assert!(matches!(r, Err(RapiraError::SchemaMismatch)));
        assert_eq!(s.len(), bytes.len());
    }

    #[test]
    fn vec_mismatch_errors_without_reading() {
        let bytes = serialize(&vec![1u32, 2, 3]);
        let peer = SchemaTyp::Vec(Box::new(SchemaTyp::Scalar(ScalarTyp::U64)));
        let mut s = bytes.as_slice();
        let r = <Vec<u32> as PeerRead>::from_slice_with_peer(&mut s, &peer);
        assert!(matches!(r, Err(RapiraError::SchemaMismatch)));
        assert_eq!(s.len(), bytes.len());
    }
}