lanspread/crates/lanspread-proto/src/lib.rs

use std::{
    fmt::{self, Write as _},
    marker::PhantomData,
    net::SocketAddr,
    str::FromStr,
};

use bytes::Bytes;
use lanspread_db::content_manifest::validate_portable_component;
pub use lanspread_db::content_manifest::{CanonicalCatalogPath, ContentId};
use serde::{
    Deserialize,
    Deserializer,
    Serialize,
    Serializer,
    de::{DeserializeOwned, IgnoredAny, SeqAccess, Visitor},
};

pub const PROTOCOL_VERSION: u32 = 8;
pub const ALPN_PROTOCOL: &[u8] = b"lanspread/8";
/// Upper bound for one control-plane frame in either direction. Responses
/// (peer-state snapshots) can legitimately approach this size.
pub const MAX_CONTROL_FRAME_BYTES: usize = 8 * 1024 * 1024;
/// Upper bound for one inbound `Request` frame. Every request variant is a
/// few kilobytes at most, so public responders decode requests with this much
/// smaller allowance instead of reserving a snapshot-sized buffer per stream.
pub const MAX_REQUEST_FRAME_BYTES: usize = 64 * 1024;
pub const MAX_STREAM_INSTALL_FRAME_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_LIBRARY_GAMES: usize = 4_096;
pub const MAX_GAME_ID_BYTES: usize = 255;
pub const MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR: usize = 4_096;
pub const MAX_CALL_TO_PLAY_DISPLAY_NAME_CHARS: usize = 24;
pub const MAX_CALL_TO_PLAY_MESSAGE_CHARS: usize = 500;
pub const MAX_CALL_TO_PLAY_AUTHOR_SNAPSHOT_BYTES: usize = 4 * 1024 * 1024;

const PEER_ID_BYTE_LENGTH: usize = 32;
const PEER_ID_ENCODED_LENGTH: usize = 52;
const BASE32_ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567";

/// Stable responder identity encoded as canonical lowercase unpadded RFC 4648 base32.
#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PeerId([u8; PEER_ID_BYTE_LENGTH]);

impl PeerId {
    /// Constructs an identity from its 32-byte value.
    #[must_use]
    pub const fn from_bytes(bytes: [u8; PEER_ID_BYTE_LENGTH]) -> Self {
        Self(bytes)
    }

    /// Returns the identity's 32-byte value.
    #[must_use]
    pub const fn as_bytes(&self) -> &[u8; PEER_ID_BYTE_LENGTH] {
        &self.0
    }
}

impl fmt::Debug for PeerId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("PeerId(\"")?;
        fmt::Display::fmt(self, formatter)?;
        formatter.write_str("\")")
    }
}

impl fmt::Display for PeerId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        for byte in encode_base32(&self.0) {
            formatter.write_char(char::from(byte))?;
        }
        Ok(())
    }
}

impl FromStr for PeerId {
    type Err = ParsePeerIdError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        decode_base32(value).map(Self).ok_or(ParsePeerIdError)
    }
}

impl Serialize for PeerId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(self)
    }
}

impl<'de> Deserialize<'de> for PeerId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?
            .parse()
            .map_err(serde::de::Error::custom)
    }
}

/// Error returned when a peer ID is not in canonical string form.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ParsePeerIdError;

impl fmt::Display for ParsePeerIdError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(
            "peer ID must be 52 characters of canonical lowercase unpadded RFC 4648 base32",
        )
    }
}

impl std::error::Error for ParsePeerIdError {}

/// One authenticated peer identity and its current transport address.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PeerEndpoint {
    pub peer_id: PeerId,
    pub addr: SocketAddr,
}

impl PeerEndpoint {
    #[must_use]
    pub const fn new(peer_id: PeerId, addr: SocketAddr) -> Self {
        Self { peer_id, addr }
    }
}

fn encode_base32(input: &[u8; PEER_ID_BYTE_LENGTH]) -> [u8; PEER_ID_ENCODED_LENGTH] {
    let mut output = [0_u8; PEER_ID_ENCODED_LENGTH];
    let mut written = 0_usize;
    let mut accumulator = 0_u16;
    let mut bits = 0_u8;

    for byte in input {
        accumulator = (accumulator << 8) | u16::from(*byte);
        bits += 8;

        while bits >= 5 {
            bits -= 5;
            let index = usize::from((accumulator >> bits) & 0x1f);
            output[written] = BASE32_ALPHABET[index];
            written += 1;
            accumulator &= (1_u16 << bits) - 1;
        }
    }

    if bits != 0 {
        let index = usize::from((accumulator << (5 - bits)) & 0x1f);
        output[written] = BASE32_ALPHABET[index];
        written += 1;
    }

    debug_assert_eq!(written, output.len());
    output
}

fn decode_base32(value: &str) -> Option<[u8; PEER_ID_BYTE_LENGTH]> {
    if value.len() != PEER_ID_ENCODED_LENGTH {
        return None;
    }

    let mut output = [0_u8; PEER_ID_BYTE_LENGTH];
    let mut written = 0_usize;
    let mut accumulator = 0_u16;
    let mut bits = 0_u8;

    for byte in value.bytes() {
        let digit = match byte {
            b'a'..=b'z' => byte - b'a',
            b'2'..=b'7' => byte - b'2' + 26,
            _ => return None,
        };

        accumulator = (accumulator << 5) | u16::from(digit);
        bits += 5;

        if bits >= 8 {
            bits -= 8;
            if written == output.len() {
                return None;
            }
            output[written] = u8::try_from(accumulator >> bits).ok()?;
            written += 1;
            accumulator &= (1_u16 << bits) - 1;
        }
    }

    (written == output.len() && bits == 4 && accumulator == 0).then_some(output)
}

const NONCE_BYTE_LENGTH: usize = 16;
const NONCE_ENCODED_LENGTH: usize = NONCE_BYTE_LENGTH * 2;

macro_rules! fixed_hex_id {
    ($name:ident, $error:ident, $label:literal) => {
        #[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
        pub struct $name([u8; NONCE_BYTE_LENGTH]);

        impl $name {
            #[must_use]
            pub const fn from_bytes(bytes: [u8; NONCE_BYTE_LENGTH]) -> Self {
                Self(bytes)
            }

            #[must_use]
            pub const fn as_bytes(&self) -> &[u8; NONCE_BYTE_LENGTH] {
                &self.0
            }
        }

        impl fmt::Debug for $name {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                formatter.write_str(concat!(stringify!($name), "(\""))?;
                fmt::Display::fmt(self, formatter)?;
                formatter.write_str("\")")
            }
        }

        impl fmt::Display for $name {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                for byte in self.0 {
                    write!(formatter, "{byte:02x}")?;
                }
                Ok(())
            }
        }

        impl FromStr for $name {
            type Err = $error;

            fn from_str(value: &str) -> Result<Self, Self::Err> {
                decode_fixed_lower_hex(value).map(Self).ok_or($error)
            }
        }

        impl Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
            where
                S: Serializer,
            {
                serializer.collect_str(self)
            }
        }

        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: Deserializer<'de>,
            {
                String::deserialize(deserializer)?
                    .parse()
                    .map_err(serde::de::Error::custom)
            }
        }

        #[derive(Clone, Copy, Debug, Eq, PartialEq)]
        pub struct $error;

        impl fmt::Display for $error {
            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
                formatter.write_str(concat!(
                    $label,
                    " must contain exactly 32 lowercase hexadecimal characters"
                ))
            }
        }

        impl std::error::Error for $error {}
    };
}

fixed_hex_id!(
    RuntimeSessionId,
    ParseRuntimeSessionIdError,
    "runtime session ID"
);
fixed_hex_id!(CallNonce, ParseCallNonceError, "call nonce");
fixed_hex_id!(EventNonce, ParseEventNonceError, "event nonce");

fn decode_fixed_lower_hex(value: &str) -> Option<[u8; NONCE_BYTE_LENGTH]> {
    if value.len() != NONCE_ENCODED_LENGTH {
        return None;
    }

    let mut output = [0_u8; NONCE_BYTE_LENGTH];
    for (decoded, encoded) in output.iter_mut().zip(value.as_bytes().as_chunks::<2>().0) {
        *decoded =
            (decode_lower_hex_nibble(encoded[0])? << 4) | decode_lower_hex_nibble(encoded[1])?;
    }
    Some(output)
}

const fn decode_lower_hex_nibble(value: u8) -> Option<u8> {
    match value {
        b'0'..=b'9' => Some(value - b'0'),
        b'a'..=b'f' => Some(value - b'a' + 10),
        _ => None,
    }
}

#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct CallId {
    pub creator: PeerId,
    pub random_nonce: CallNonce,
}

impl CallId {
    #[must_use]
    pub const fn new(creator: PeerId, random_nonce: CallNonce) -> Self {
        Self {
            creator,
            random_nonce,
        }
    }
}

impl fmt::Debug for CallId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("CallId(\"")?;
        fmt::Display::fmt(self, formatter)?;
        formatter.write_str("\")")
    }
}

impl fmt::Display for CallId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "{}.{}", self.creator, self.random_nonce)
    }
}

impl FromStr for CallId {
    type Err = ParseCallIdError;

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        let (creator, random_nonce) = value.split_once('.').ok_or(ParseCallIdError)?;
        Ok(Self {
            creator: creator.parse().map_err(|_| ParseCallIdError)?,
            random_nonce: random_nonce.parse().map_err(|_| ParseCallIdError)?,
        })
    }
}

impl Serialize for CallId {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.collect_str(self)
    }
}

impl<'de> Deserialize<'de> for CallId {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        String::deserialize(deserializer)?
            .parse()
            .map_err(serde::de::Error::custom)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ParseCallIdError;

impl fmt::Display for ParseCallIdError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(
            "call ID must be a canonical PeerId, '.', and 32 lowercase hexadecimal characters",
        )
    }
}

impl std::error::Error for ParseCallIdError {}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PeerRevisions {
    pub runtime_session_id: RuntimeSessionId,
    pub library_revision: u64,
    pub call_to_play_revision: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct GameAvailability {
    pub game_id: String,
    pub content_id: ContentId,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct LibrarySnapshot {
    pub revision: u64,
    #[serde(deserialize_with = "deserialize_library_games")]
    pub games: Vec<GameAvailability>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ChangeHint {
    pub claimed_peer_id: PeerId,
    pub runtime_session_id: RuntimeSessionId,
    pub revision: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CallToPlayAuthorEvent {
    pub id: EventNonce,
    pub call_id: CallId,
    pub at: i64,
    pub action: CallToPlayAction,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub enum CallToPlayAction {
    Create {
        game_id: String,
        max_players: u16,
        scheduled_for: Option<i64>,
        deadline: i64,
    },
    Respond {
        ready_at: Option<i64>,
    },
    Rsvp,
    SendMessage {
        text: String,
    },
    Leave,
    Cancel,
    Start,
    AddTime {
        deadline: i64,
    },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CallToPlayAuthorSnapshot {
    pub revision: u64,
    pub display_name: String,
    #[serde(deserialize_with = "deserialize_call_to_play_events")]
    pub events: Vec<CallToPlayAuthorEvent>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct PeerStateSnapshot {
    pub runtime_session_id: RuntimeSessionId,
    pub library: LibrarySnapshot,
    pub call_to_play: CallToPlayAuthorSnapshot,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub enum Request {
    Ping,
    Hello,
    LibraryChanged(ChangeHint),
    CallToPlayChanged(ChangeHint),
    GetGameFileChunk {
        game_id: String,
        content_id: ContentId,
        relative_path: CanonicalCatalogPath,
        offset: u64,
        length: u64,
    },
    StreamInstall {
        game_id: String,
        content_id: ContentId,
    },
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum ControlErrorCode {
    InvalidRequest,
    Internal,
    Unavailable,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub enum Response {
    Pong(PeerRevisions),
    HelloSnapshot(PeerStateSnapshot),
    Error(ControlErrorCode),
}

const STREAM_INSTALL_CONTROL_FRAME_TAG: u8 = 0;
const STREAM_INSTALL_FILE_CHUNK_FRAME_TAG: u8 = 1;
const STREAM_INSTALL_ENCODE_ERROR_FRAME: &[u8] =
    b"\0{\"Error\":{\"message\":\"stream install frame encoding error\"}}";

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub enum StreamInstallFrame {
    ArchiveBegin {
        archive_name: CanonicalCatalogPath,
        solid: bool,
        unpacked_size: u64,
    },
    Directory {
        relative_path: CanonicalCatalogPath,
    },
    FileBegin {
        relative_path: CanonicalCatalogPath,
        size: u64,
        crc32: u32,
    },
    FileChunk {
        bytes: Bytes,
    },
    FileEnd {
        relative_path: CanonicalCatalogPath,
    },
    ArchiveEnd {
        archive_name: CanonicalCatalogPath,
    },
    Complete,
    Error {
        message: String,
    },
}

#[derive(Debug)]
pub enum StreamInstallFrameDecodeError {
    Empty,
    FrameTooLarge { actual: usize, maximum: usize },
    UnknownTag(u8),
    InvalidControl(serde_json::Error),
    FileChunkInControl,
}

impl fmt::Display for StreamInstallFrameDecodeError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => formatter.write_str("stream install frame is empty"),
            Self::FrameTooLarge { actual, maximum } => write!(
                formatter,
                "stream install frame is {actual} bytes; maximum is {maximum}"
            ),
            Self::UnknownTag(tag) => write!(formatter, "unknown stream install frame tag {tag}"),
            Self::InvalidControl(error) => {
                write!(formatter, "invalid stream install control frame: {error}")
            }
            Self::FileChunkInControl => {
                formatter.write_str("stream install control frame cannot contain file bytes")
            }
        }
    }
}

impl std::error::Error for StreamInstallFrameDecodeError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::InvalidControl(error) => Some(error),
            Self::Empty
            | Self::FrameTooLarge { .. }
            | Self::UnknownTag(_)
            | Self::FileChunkInControl => None,
        }
    }
}

impl StreamInstallFrame {
    /// Strictly decodes one bounded streamed-install frame.
    ///
    /// An explicit sender [`StreamInstallFrame::Error`] remains a successfully
    /// decoded frame. Malformed framing, JSON, or canonical paths return an
    /// error so a receiver can classify them as integrity failures.
    ///
    /// # Errors
    ///
    /// Returns an error for an empty or oversized frame, an unknown tag,
    /// malformed typed control JSON, or file bytes encoded under the control
    /// tag.
    pub fn decode_checked(mut bytes: Bytes) -> Result<Self, StreamInstallFrameDecodeError> {
        if bytes.len() > MAX_STREAM_INSTALL_FRAME_BYTES {
            return Err(StreamInstallFrameDecodeError::FrameTooLarge {
                actual: bytes.len(),
                maximum: MAX_STREAM_INSTALL_FRAME_BYTES,
            });
        }
        if bytes.is_empty() {
            return Err(StreamInstallFrameDecodeError::Empty);
        }
        let tag = bytes.split_to(1)[0];
        let payload = bytes;
        match tag {
            STREAM_INSTALL_CONTROL_FRAME_TAG => {
                let frame = serde_json::from_slice(&payload)
                    .map_err(StreamInstallFrameDecodeError::InvalidControl)?;
                if matches!(frame, Self::FileChunk { .. }) {
                    return Err(StreamInstallFrameDecodeError::FileChunkInControl);
                }
                Ok(frame)
            }
            STREAM_INSTALL_FILE_CHUNK_FRAME_TAG => Ok(Self::FileChunk { bytes: payload }),
            _ => Err(StreamInstallFrameDecodeError::UnknownTag(tag)),
        }
    }
}

/// Non-fallible framed codec retained for streamed-install data frames.
pub trait Message {
    fn decode(bytes: Bytes) -> Self;
    fn encode(&self) -> Bytes;
}

/// Strict, bounded JSON codec for one protocol control frame.
pub trait ControlMessage: Sized {
    /// Decodes one bounded, strict control frame.
    ///
    /// # Errors
    ///
    /// Returns an error when the frame exceeds its bound, is not the exact
    /// current wire shape, or fails request-level semantic validation.
    fn decode(bytes: Bytes) -> Result<Self, ControlCodecError>;

    /// Encodes one bounded, semantically valid control frame.
    ///
    /// # Errors
    ///
    /// Returns an error when the value fails semantic validation, cannot be
    /// serialized, or its encoded frame exceeds the control-frame bound.
    fn encode(&self) -> Result<Bytes, ControlCodecError>;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ControlValidationError {
    EmptyField {
        field: &'static str,
    },
    FieldTooLong {
        field: &'static str,
        maximum: usize,
    },
    EncodedTooLarge {
        field: &'static str,
        actual: usize,
        maximum: usize,
    },
    TooManyItems {
        field: &'static str,
        maximum: usize,
    },
    InvalidPathComponent {
        field: &'static str,
    },
    DuplicateGameId,
    UnsortedGameIds,
}

impl fmt::Display for ControlValidationError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyField { field } => write!(formatter, "{field} cannot be blank"),
            Self::FieldTooLong { field, maximum } => {
                write!(formatter, "{field} exceeds its {maximum}-unit limit")
            }
            Self::EncodedTooLarge {
                field,
                actual,
                maximum,
            } => write!(
                formatter,
                "encoded {field} is {actual} bytes; maximum is {maximum}"
            ),
            Self::TooManyItems { field, maximum } => {
                write!(formatter, "{field} exceeds its {maximum}-item limit")
            }
            Self::InvalidPathComponent { field } => {
                write!(formatter, "{field} is not a single plain path component")
            }
            Self::DuplicateGameId => formatter.write_str("library contains a duplicate game ID"),
            Self::UnsortedGameIds => {
                formatter.write_str("library game IDs are not strictly sorted")
            }
        }
    }
}

impl std::error::Error for ControlValidationError {}

#[derive(Debug)]
pub enum ControlCodecError {
    FrameTooLarge { actual: usize, maximum: usize },
    Invalid(ControlValidationError),
    Encode(serde_json::Error),
    Decode(serde_json::Error),
}

impl fmt::Display for ControlCodecError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::FrameTooLarge { actual, maximum } => {
                write!(
                    formatter,
                    "control frame is {actual} bytes; maximum is {maximum}"
                )
            }
            Self::Invalid(error) => write!(formatter, "invalid control message: {error}"),
            Self::Encode(error) => write!(formatter, "failed to encode control message: {error}"),
            Self::Decode(error) => write!(formatter, "failed to decode control message: {error}"),
        }
    }
}

impl std::error::Error for ControlCodecError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Invalid(error) => Some(error),
            Self::Encode(error) | Self::Decode(error) => Some(error),
            Self::FrameTooLarge { .. } => None,
        }
    }
}

trait ValidateControlMessage {
    fn validate_control(&self) -> Result<(), ControlValidationError>;
}

impl ControlMessage for Request {
    fn decode(bytes: Bytes) -> Result<Self, ControlCodecError> {
        check_request_frame_length(bytes.len())?;
        decode_control_message(&bytes)
    }

    fn encode(&self) -> Result<Bytes, ControlCodecError> {
        let bytes = encode_control_message(self)?;
        check_request_frame_length(bytes.len())?;
        Ok(bytes)
    }
}

impl ControlMessage for Response {
    fn decode(bytes: Bytes) -> Result<Self, ControlCodecError> {
        decode_control_message_unvalidated(&bytes)
    }

    fn encode(&self) -> Result<Bytes, ControlCodecError> {
        encode_control_message(self)
    }
}

fn decode_control_message<T>(bytes: &[u8]) -> Result<T, ControlCodecError>
where
    T: DeserializeOwned + ValidateControlMessage,
{
    let message = decode_control_message_unvalidated::<T>(bytes)?;
    message
        .validate_control()
        .map_err(ControlCodecError::Invalid)?;
    Ok(message)
}

fn decode_control_message_unvalidated<T>(bytes: &[u8]) -> Result<T, ControlCodecError>
where
    T: DeserializeOwned,
{
    check_control_frame_length(bytes.len())?;
    serde_json::from_slice::<T>(bytes).map_err(ControlCodecError::Decode)
}

fn encode_control_message<T>(message: &T) -> Result<Bytes, ControlCodecError>
where
    T: Serialize + ValidateControlMessage,
{
    message
        .validate_control()
        .map_err(ControlCodecError::Invalid)?;
    let bytes = serde_json::to_vec(message).map_err(ControlCodecError::Encode)?;
    check_control_frame_length(bytes.len())?;
    Ok(Bytes::from(bytes))
}

const fn check_control_frame_length(length: usize) -> Result<(), ControlCodecError> {
    if length > MAX_CONTROL_FRAME_BYTES {
        return Err(ControlCodecError::FrameTooLarge {
            actual: length,
            maximum: MAX_CONTROL_FRAME_BYTES,
        });
    }
    Ok(())
}

const fn check_request_frame_length(length: usize) -> Result<(), ControlCodecError> {
    if length > MAX_REQUEST_FRAME_BYTES {
        return Err(ControlCodecError::FrameTooLarge {
            actual: length,
            maximum: MAX_REQUEST_FRAME_BYTES,
        });
    }
    Ok(())
}

impl ValidateControlMessage for Request {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        match self {
            Self::Ping | Self::Hello | Self::LibraryChanged(_) | Self::CallToPlayChanged(_) => {
                Ok(())
            }
            Self::GetGameFileChunk { game_id, .. } | Self::StreamInstall { game_id, .. } => {
                validate_game_id(game_id)
            }
        }
    }
}

impl ValidateControlMessage for Response {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        match self {
            Self::Pong(_) | Self::Error(_) => Ok(()),
            Self::HelloSnapshot(snapshot) => snapshot.validate_control(),
        }
    }
}

impl ValidateControlMessage for PeerStateSnapshot {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        self.library.validate_control()?;
        self.call_to_play.validate_control()
    }
}

impl ValidateControlMessage for LibrarySnapshot {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        self.validate()
    }
}

impl LibrarySnapshot {
    /// Validates the bounded, canonical library domain independently of the
    /// other domains in a peer-state snapshot.
    ///
    /// # Errors
    ///
    /// Returns the first resource-bound or canonical-ordering violation.
    pub fn validate(&self) -> Result<(), ControlValidationError> {
        if self.games.len() > MAX_LIBRARY_GAMES {
            return Err(ControlValidationError::TooManyItems {
                field: "library games",
                maximum: MAX_LIBRARY_GAMES,
            });
        }

        for game in &self.games {
            validate_game_id(&game.game_id)?;
        }
        for games in self.games.windows(2) {
            match games[0].game_id.cmp(&games[1].game_id) {
                std::cmp::Ordering::Less => {}
                std::cmp::Ordering::Equal => {
                    return Err(ControlValidationError::DuplicateGameId);
                }
                std::cmp::Ordering::Greater => {
                    return Err(ControlValidationError::UnsortedGameIds);
                }
            }
        }
        Ok(())
    }
}

impl ValidateControlMessage for CallToPlayAuthorSnapshot {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        self.validate()
    }
}

impl CallToPlayAuthorSnapshot {
    /// Validates the bounded Call-to-Play domain independently of the other
    /// domains in a peer-state snapshot.
    ///
    /// # Errors
    ///
    /// Returns the first display-name, event-count, or event resource-bound
    /// violation.
    pub fn validate(&self) -> Result<(), ControlValidationError> {
        validate_bounded_chars(
            &self.display_name,
            MAX_CALL_TO_PLAY_DISPLAY_NAME_CHARS,
            "Call to Play display name",
        )?;
        if self.events.len() > MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR {
            return Err(ControlValidationError::TooManyItems {
                field: "Call to Play author events",
                maximum: MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR,
            });
        }
        for event in &self.events {
            event.validate()?;
        }
        let actual = serde_json::to_vec(self)
            .map_err(|_| ControlValidationError::EncodedTooLarge {
                field: "Call to Play author snapshot",
                actual: usize::MAX,
                maximum: MAX_CALL_TO_PLAY_AUTHOR_SNAPSHOT_BYTES,
            })?
            .len();
        if actual > MAX_CALL_TO_PLAY_AUTHOR_SNAPSHOT_BYTES {
            return Err(ControlValidationError::EncodedTooLarge {
                field: "Call to Play author snapshot",
                actual,
                maximum: MAX_CALL_TO_PLAY_AUTHOR_SNAPSHOT_BYTES,
            });
        }
        Ok(())
    }
}

impl ValidateControlMessage for CallToPlayAuthorEvent {
    fn validate_control(&self) -> Result<(), ControlValidationError> {
        self.validate()
    }
}

impl CallToPlayAuthorEvent {
    /// Validates the wire-level resource bounds of one author event.
    ///
    /// # Errors
    ///
    /// Returns an error when a game ID or message violates its wire bound.
    pub fn validate(&self) -> Result<(), ControlValidationError> {
        match &self.action {
            CallToPlayAction::Create { game_id, .. } => validate_game_id(game_id),
            CallToPlayAction::SendMessage { text } => {
                validate_bounded_chars(text, MAX_CALL_TO_PLAY_MESSAGE_CHARS, "Call to Play message")
            }
            CallToPlayAction::Respond { .. }
            | CallToPlayAction::Rsvp
            | CallToPlayAction::Leave
            | CallToPlayAction::Cancel
            | CallToPlayAction::Start
            | CallToPlayAction::AddTime { .. } => Ok(()),
        }
    }
}

/// Game IDs name one catalog directory. The wire boundary therefore admits
/// only a single plain path component: no separators, no control characters,
/// and neither of the `.`/`..` pseudo-components. On top of that it applies
/// the catalog's own portable component rules (no trailing dot or space, no
/// Windows-reserved characters, no Windows device name stem) by calling the
/// `lanspread-db` validator, so anything the catalog can publish is accepted
/// and anything it would refuse never reaches a filesystem-backed handler.
/// Every consumer still resolves the ID against the local catalog before
/// touching the filesystem; this is defence in depth, not the only gate.
fn validate_game_id(game_id: &str) -> Result<(), ControlValidationError> {
    if game_id.trim().is_empty() {
        return Err(ControlValidationError::EmptyField { field: "game ID" });
    }
    if game_id.len() > MAX_GAME_ID_BYTES {
        return Err(ControlValidationError::FieldTooLong {
            field: "game ID",
            maximum: MAX_GAME_ID_BYTES,
        });
    }
    if matches!(game_id, "." | "..")
        || game_id
            .chars()
            .any(|character| character.is_control() || matches!(character, '/' | '\\'))
        || validate_portable_component(game_id).is_err()
    {
        return Err(ControlValidationError::InvalidPathComponent { field: "game ID" });
    }
    Ok(())
}

fn deserialize_library_games<'de, D>(deserializer: D) -> Result<Vec<GameAvailability>, D::Error>
where
    D: Deserializer<'de>,
{
    deserialize_bounded_vec(deserializer, MAX_LIBRARY_GAMES, "library games")
}

fn deserialize_call_to_play_events<'de, D>(
    deserializer: D,
) -> Result<Vec<CallToPlayAuthorEvent>, D::Error>
where
    D: Deserializer<'de>,
{
    deserialize_bounded_vec(
        deserializer,
        MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR,
        "Call to Play author events",
    )
}

/// Deserializes a wire sequence without ever materialising more than
/// `maximum + 1` elements.
///
/// The semantic validators (`LibrarySnapshot::validate`,
/// `CallToPlayAuthorSnapshot::validate`) report `TooManyItems` when a
/// collection exceeds its limit, and `Response::decode` deliberately leaves
/// that judgement to the caller so one invalid domain does not discard the
/// other. Truncating to exactly one element past the limit keeps both
/// behaviours intact while bounding the memory an oversize frame can claim
/// during parsing: the remaining elements are consumed as `IgnoredAny` and
/// never allocated.
fn deserialize_bounded_vec<'de, D, T>(
    deserializer: D,
    maximum: usize,
    field: &'static str,
) -> Result<Vec<T>, D::Error>
where
    D: Deserializer<'de>,
    T: Deserialize<'de>,
{
    struct BoundedVecVisitor<T> {
        maximum: usize,
        field: &'static str,
        marker: PhantomData<T>,
    }

    impl<'de, T> Visitor<'de> for BoundedVecVisitor<T>
    where
        T: Deserialize<'de>,
    {
        type Value = Vec<T>;

        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
            write!(formatter, "a bounded sequence of {}", self.field)
        }

        fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
        where
            A: SeqAccess<'de>,
        {
            let mut values = Vec::new();
            while let Some(value) = sequence.next_element()? {
                values.push(value);
                if values.len() > self.maximum {
                    while sequence.next_element::<IgnoredAny>()?.is_some() {}
                    break;
                }
            }
            Ok(values)
        }
    }

    deserializer.deserialize_seq(BoundedVecVisitor {
        maximum,
        field,
        marker: PhantomData,
    })
}

fn validate_bounded_chars(
    value: &str,
    maximum: usize,
    field: &'static str,
) -> Result<(), ControlValidationError> {
    if value.trim().is_empty() {
        return Err(ControlValidationError::EmptyField { field });
    }
    if value.chars().count() > maximum {
        return Err(ControlValidationError::FieldTooLong { field, maximum });
    }
    Ok(())
}

impl Message for StreamInstallFrame {
    fn decode(bytes: Bytes) -> Self {
        match Self::decode_checked(bytes) {
            Ok(frame) => frame,
            Err(error) => {
                tracing::error!(?error, "StreamInstallFrame decoding error");
                stream_install_decode_error(error.to_string())
            }
        }
    }

    fn encode(&self) -> Bytes {
        match self {
            StreamInstallFrame::FileChunk { bytes } => {
                tagged_stream_install_frame(STREAM_INSTALL_FILE_CHUNK_FRAME_TAG, bytes)
            }
            _ => match serde_json::to_vec(self) {
                Ok(payload) => {
                    tagged_stream_install_frame(STREAM_INSTALL_CONTROL_FRAME_TAG, &payload)
                }
                Err(e) => {
                    tracing::error!(?e, "StreamInstallFrame encoding error");
                    Bytes::from_static(STREAM_INSTALL_ENCODE_ERROR_FRAME)
                }
            },
        }
    }
}

fn tagged_stream_install_frame(tag: u8, payload: &[u8]) -> Bytes {
    let mut frame = Vec::with_capacity(1 + payload.len());
    frame.push(tag);
    frame.extend_from_slice(payload);
    Bytes::from(frame)
}

fn stream_install_decode_error(message: impl Into<String>) -> StreamInstallFrame {
    StreamInstallFrame::Error {
        message: message.into(),
    }
}

#[cfg(test)]
mod tests {
    use std::net::SocketAddr;

    use serde_json::{Value, json};

    use super::*;

    fn path(value: &str) -> CanonicalCatalogPath {
        CanonicalCatalogPath::new(value).expect("test path should be canonical")
    }

    fn peer(seed: u8) -> PeerId {
        PeerId::from_bytes([seed; 32])
    }

    fn session(seed: u8) -> RuntimeSessionId {
        RuntimeSessionId::from_bytes([seed; 16])
    }

    fn call_nonce(seed: u8) -> CallNonce {
        CallNonce::from_bytes([seed; 16])
    }

    fn event_nonce(seed: u8) -> EventNonce {
        EventNonce::from_bytes([seed; 16])
    }

    fn content(seed: u8) -> ContentId {
        ContentId::from_bytes([seed; 32])
    }

    fn event(action: CallToPlayAction) -> CallToPlayAuthorEvent {
        CallToPlayAuthorEvent {
            id: event_nonce(4),
            call_id: CallId::new(peer(5), call_nonce(6)),
            at: 1_700_000_000_000,
            action,
        }
    }

    fn state_snapshot(events: Vec<CallToPlayAuthorEvent>) -> PeerStateSnapshot {
        PeerStateSnapshot {
            runtime_session_id: session(1),
            library: LibrarySnapshot {
                revision: 2,
                games: vec![
                    GameAvailability {
                        game_id: "alpha".to_owned(),
                        content_id: content(2),
                    },
                    GameAvailability {
                        game_id: "bravo".to_owned(),
                        content_id: content(3),
                    },
                ],
            },
            call_to_play: CallToPlayAuthorSnapshot {
                revision: 3,
                display_name: "Alice".to_owned(),
                events,
            },
        }
    }

    fn encode_text(message: &impl ControlMessage) -> String {
        String::from_utf8(
            message
                .encode()
                .expect("test control message should encode")
                .to_vec(),
        )
        .expect("control JSON should be UTF-8")
    }

    #[test]
    fn peer_id_uses_canonical_lowercase_unpadded_base32() {
        let zero = PeerId::from_bytes([0_u8; 32]);
        assert_eq!(zero.to_string(), "a".repeat(52));

        let ones = PeerId::from_bytes([u8::MAX; 32]);
        assert_eq!(ones.to_string(), format!("{}q", "7".repeat(51)));
        assert_eq!(
            ones.to_string()
                .parse::<PeerId>()
                .expect("canonical peer ID should parse"),
            ones
        );
        assert_eq!(ones.as_bytes(), &[u8::MAX; 32]);
    }

    #[test]
    fn peer_id_round_trips_every_byte() {
        let mut bytes = [0_u8; 32];
        for (value, byte) in (0_u8..32).zip(&mut bytes) {
            *byte = value;
        }

        let peer_id = PeerId::from_bytes(bytes);
        let encoded = peer_id.to_string();

        assert_eq!(encoded.len(), 52);
        assert_eq!(
            encoded
                .parse::<PeerId>()
                .expect("encoded peer ID should parse"),
            peer_id
        );
    }

    #[test]
    fn peer_id_rejects_every_noncanonical_shape() {
        let canonical = PeerId::from_bytes([0_u8; 32]).to_string();
        let mut nonzero_trailing_bits = canonical.clone();
        nonzero_trailing_bits.replace_range(51..52, "b");

        for invalid in [
            canonical[..51].to_owned(),
            format!("{canonical}a"),
            canonical.to_ascii_uppercase(),
            format!("{canonical}="),
            canonical.replacen('a', "0", 1),
            nonzero_trailing_bits,
            format!(" {canonical}"),
        ] {
            assert!(
                invalid.parse::<PeerId>().is_err(),
                "accepted noncanonical peer ID {invalid:?}"
            );
        }
    }

    #[test]
    fn peer_id_serde_is_exactly_one_canonical_string() {
        let peer_id = PeerId::from_bytes([0x5a; 32]);
        let encoded = peer_id.to_string();
        let json = serde_json::to_string(&peer_id).expect("peer ID should serialize");

        assert_eq!(json, format!(r#""{encoded}""#));
        assert_eq!(
            serde_json::from_str::<PeerId>(&json).expect("peer ID should deserialize"),
            peer_id
        );
        assert!(serde_json::from_str::<PeerId>("42").is_err());
        assert!(
            serde_json::from_str::<PeerId>(
                r#""AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA""#,
            )
            .is_err()
        );
    }

    #[test]
    fn peer_endpoint_carries_strict_identity_and_address() {
        let peer_id = PeerId::from_bytes([0x3c; 32]);
        let addr = SocketAddr::from(([127, 0, 0, 1], 42_424));
        let endpoint = PeerEndpoint::new(peer_id, addr);
        let endpoint_json = serde_json::to_value(endpoint).expect("peer endpoint should serialize");
        let encoded_peer_id = peer_id.to_string();

        assert_eq!(endpoint.peer_id, peer_id);
        assert_eq!(endpoint.addr, addr);
        assert_eq!(
            endpoint_json.get("peer_id").and_then(Value::as_str),
            Some(encoded_peer_id.as_str())
        );
        assert_eq!(
            endpoint_json.get("addr").and_then(Value::as_str),
            Some("127.0.0.1:42424")
        );
        assert!(
            serde_json::from_value::<PeerEndpoint>(json!({
                "peer_id": encoded_peer_id,
                "addr": "127.0.0.1:42424",
                "extra": true,
            }))
            .is_err()
        );
    }

    #[test]
    fn protocol_version_and_alpn_are_one_v8_cutover() {
        assert_eq!(PROTOCOL_VERSION, 8);
        assert_eq!(ALPN_PROTOCOL, b"lanspread/8");
        assert_eq!(MAX_CONTROL_FRAME_BYTES, 8 * 1024 * 1024);
        assert_eq!(MAX_REQUEST_FRAME_BYTES, 64 * 1024);
        assert_eq!(MAX_CALL_TO_PLAY_AUTHOR_SNAPSHOT_BYTES, 4 * 1024 * 1024);
    }

    #[test]
    fn maximal_requests_fit_the_request_frame_bound() {
        use lanspread_db::content_manifest::{MAX_CATALOG_COMPONENT_BYTES, MAX_CATALOG_PATH_BYTES};

        let directory = "d".repeat(MAX_CATALOG_COMPONENT_BYTES);
        let relative_path = format!(
            "{directory}/{directory}/{directory}/{}",
            "f".repeat(MAX_CATALOG_PATH_BYTES - 3 * (MAX_CATALOG_COMPONENT_BYTES + 1))
        );
        assert_eq!(relative_path.len(), MAX_CATALOG_PATH_BYTES);
        let request = Request::GetGameFileChunk {
            game_id: "g".repeat(MAX_GAME_ID_BYTES),
            content_id: content(0xff),
            relative_path: CanonicalCatalogPath::new(relative_path)
                .expect("maximal catalog path should be canonical"),
            offset: u64::MAX,
            length: u64::MAX,
        };
        let encoded = request.encode().expect("maximal request must encode");
        assert!(encoded.len() < MAX_REQUEST_FRAME_BYTES / 8);
        Request::decode(encoded).expect("maximal request must decode");
    }

    #[test]
    fn fixed_hex_ids_are_canonical_and_distinct() {
        let runtime = session(0xab);
        let call = call_nonce(0xcd);
        let event = event_nonce(0xef);
        assert_eq!(runtime.to_string(), "ab".repeat(16));
        assert_eq!(call.to_string(), "cd".repeat(16));
        assert_eq!(event.to_string(), "ef".repeat(16));
        assert_eq!(runtime.as_bytes(), &[0xab; 16]);
        assert_eq!(call.as_bytes(), &[0xcd; 16]);
        assert_eq!(event.as_bytes(), &[0xef; 16]);
        assert_eq!(runtime.to_string().parse::<RuntimeSessionId>(), Ok(runtime));
        assert_eq!(call.to_string().parse::<CallNonce>(), Ok(call));
        assert_eq!(event.to_string().parse::<EventNonce>(), Ok(event));

        for invalid in [
            "a".repeat(31),
            "a".repeat(33),
            "AB".repeat(16),
            format!("{}g", "a".repeat(31)),
        ] {
            assert!(invalid.parse::<RuntimeSessionId>().is_err());
            assert!(invalid.parse::<CallNonce>().is_err());
            assert!(invalid.parse::<EventNonce>().is_err());
        }
    }

    #[test]
    fn call_id_is_one_canonical_string() {
        let call_id = CallId::new(peer(0), call_nonce(0xab));
        let expected = format!("{}.{}", "a".repeat(52), "ab".repeat(16));
        assert_eq!(call_id.to_string(), expected);
        assert_eq!(
            serde_json::to_string(&call_id).expect("serialize"),
            format!(r#""{expected}""#)
        );
        assert_eq!(expected.parse::<CallId>(), Ok(call_id));
        for invalid in [
            "a".repeat(52),
            format!("{}.{}", "A".repeat(52), "ab".repeat(16)),
            format!("{}.{}", "a".repeat(52), "AB".repeat(16)),
            format!("{}.{}.x", "a".repeat(52), "ab".repeat(16)),
        ] {
            assert!(invalid.parse::<CallId>().is_err(), "accepted {invalid:?}");
        }
        assert!(
            serde_json::from_value::<CallId>(json!({
                "creator": peer(0).to_string(),
                "random_nonce": call_nonce(0xab).to_string(),
            }))
            .is_err()
        );
    }

    #[test]
    fn owned_json_values_deserialize_typed_ids() {
        let runtime_session_id = session(0xab);
        let call_nonce = call_nonce(0xcd);
        let event_nonce = event_nonce(0xef);
        let call_id = CallId::new(peer(0), call_nonce);

        assert_eq!(
            serde_json::from_value::<RuntimeSessionId>(json!(runtime_session_id.to_string()))
                .expect("owned runtime-session JSON value should deserialize"),
            runtime_session_id
        );
        assert_eq!(
            serde_json::from_value::<CallNonce>(json!(call_nonce.to_string()))
                .expect("owned call-nonce JSON value should deserialize"),
            call_nonce
        );
        assert_eq!(
            serde_json::from_value::<EventNonce>(json!(event_nonce.to_string()))
                .expect("owned event-nonce JSON value should deserialize"),
            event_nonce
        );
        assert_eq!(
            serde_json::from_value::<CallId>(json!(call_id.to_string()))
                .expect("owned call-ID JSON value should deserialize"),
            call_id
        );
    }

    #[test]
    fn request_and_response_json_are_golden() {
        let peer_id = peer(0).to_string();
        let runtime = session(1).to_string();
        let content_id = content(2).to_string();
        assert_eq!(encode_text(&Request::Ping), r#""Ping""#);
        assert_eq!(encode_text(&Request::Hello), r#""Hello""#);
        assert_eq!(
            encode_text(&Request::LibraryChanged(ChangeHint {
                claimed_peer_id: peer(0),
                runtime_session_id: session(1),
                revision: 7,
            })),
            format!(
                r#"{{"LibraryChanged":{{"claimed_peer_id":"{peer_id}","runtime_session_id":"{runtime}","revision":7}}}}"#
            ),
        );
        assert_eq!(
            encode_text(&Request::CallToPlayChanged(ChangeHint {
                claimed_peer_id: peer(0),
                runtime_session_id: session(1),
                revision: 8,
            })),
            format!(
                r#"{{"CallToPlayChanged":{{"claimed_peer_id":"{peer_id}","runtime_session_id":"{runtime}","revision":8}}}}"#
            ),
        );
        assert_eq!(
            encode_text(&Request::GetGameFileChunk {
                game_id: "game".to_owned(),
                content_id: content(2),
                relative_path: path("bin/game.exe"),
                offset: 7,
                length: 9,
            }),
            format!(
                r#"{{"GetGameFileChunk":{{"game_id":"game","content_id":"{content_id}","relative_path":"bin/game.exe","offset":7,"length":9}}}}"#
            ),
        );
        assert_eq!(
            encode_text(&Request::StreamInstall {
                game_id: "game".to_owned(),
                content_id: content(2),
            }),
            format!(r#"{{"StreamInstall":{{"game_id":"game","content_id":"{content_id}"}}}}"#),
        );
        assert_eq!(
            encode_text(&Response::Pong(PeerRevisions {
                runtime_session_id: session(1),
                library_revision: 2,
                call_to_play_revision: 3,
            })),
            format!(
                r#"{{"Pong":{{"runtime_session_id":"{runtime}","library_revision":2,"call_to_play_revision":3}}}}"#
            ),
        );
        assert_eq!(
            encode_text(&Response::Error(ControlErrorCode::Unavailable)),
            r#"{"Error":"Unavailable"}"#,
        );

        let empty = PeerStateSnapshot {
            runtime_session_id: session(1),
            library: LibrarySnapshot {
                revision: 2,
                games: Vec::new(),
            },
            call_to_play: CallToPlayAuthorSnapshot {
                revision: 3,
                display_name: "Alice".to_owned(),
                events: Vec::new(),
            },
        };
        assert_eq!(
            encode_text(&Response::HelloSnapshot(empty)),
            format!(
                r#"{{"HelloSnapshot":{{"runtime_session_id":"{runtime}","library":{{"revision":2,"games":[]}},"call_to_play":{{"revision":3,"display_name":"Alice","events":[]}}}}}}"#
            ),
        );
    }

    #[test]
    fn call_to_play_actions_have_golden_json() {
        let cases = [
            (
                CallToPlayAction::Create {
                    game_id: "game".to_owned(),
                    max_players: 4,
                    scheduled_for: None,
                    deadline: 42,
                },
                r#"{"Create":{"game_id":"game","max_players":4,"scheduled_for":null,"deadline":42}}"#,
            ),
            (
                CallToPlayAction::Respond { ready_at: Some(12) },
                r#"{"Respond":{"ready_at":12}}"#,
            ),
            (CallToPlayAction::Rsvp, r#""Rsvp""#),
            (
                CallToPlayAction::SendMessage {
                    text: "hello".to_owned(),
                },
                r#"{"SendMessage":{"text":"hello"}}"#,
            ),
            (CallToPlayAction::Leave, r#""Leave""#),
            (CallToPlayAction::Cancel, r#""Cancel""#),
            (CallToPlayAction::Start, r#""Start""#),
            (
                CallToPlayAction::AddTime { deadline: 99 },
                r#"{"AddTime":{"deadline":99}}"#,
            ),
        ];
        for (action, expected) in cases {
            assert_eq!(
                serde_json::to_string(&action).expect("serialize action"),
                expected
            );
        }
    }

    #[test]
    fn all_control_variants_round_trip() {
        let hint = ChangeHint {
            claimed_peer_id: peer(7),
            runtime_session_id: session(8),
            revision: 9,
        };
        let requests = [
            Request::Ping,
            Request::Hello,
            Request::LibraryChanged(hint),
            Request::CallToPlayChanged(hint),
            Request::GetGameFileChunk {
                game_id: "game".to_owned(),
                content_id: content(9),
                relative_path: path("data/file.bin"),
                offset: 10,
                length: 11,
            },
            Request::StreamInstall {
                game_id: "game".to_owned(),
                content_id: content(9),
            },
        ];
        for request in requests {
            let encoded = request.encode().expect("encode request");
            assert_eq!(Request::decode(encoded).expect("decode request"), request);
        }

        let responses = [
            Response::Pong(PeerRevisions {
                runtime_session_id: session(1),
                library_revision: 2,
                call_to_play_revision: 3,
            }),
            Response::HelloSnapshot(state_snapshot(vec![event(CallToPlayAction::Rsvp)])),
            Response::Error(ControlErrorCode::InvalidRequest),
            Response::Error(ControlErrorCode::Internal),
            Response::Error(ControlErrorCode::Unavailable),
        ];
        for response in responses {
            let encoded = response.encode().expect("encode response");
            assert_eq!(
                Response::decode(encoded).expect("decode response"),
                response
            );
        }
    }

    #[test]
    fn strict_serde_rejects_unknown_missing_duplicate_and_identity_fields() {
        let content_id = content(1);
        for invalid in [
            format!(
                r#"{{"StreamInstall":{{"game_id":"game","content_id":"{content_id}","extra":true}}}}"#
            ),
            r#"{"StreamInstall":{"game_id":"game"}}"#.to_owned(),
            format!(
                r#"{{"StreamInstall":{{"game_id":"game","game_id":"other","content_id":"{content_id}"}}}}"#
            ),
            r#"{"Hello":{"peer_id":"legacy","proto_ver":7}}"#.to_owned(),
        ] {
            assert!(Request::decode(Bytes::from(invalid)).is_err());
        }

        let mut value = serde_json::to_value(Response::HelloSnapshot(state_snapshot(vec![event(
            CallToPlayAction::Rsvp,
        )])))
        .expect("serialize snapshot");
        value["HelloSnapshot"]["call_to_play"]["events"][0]["actor_id"] = json!(peer(8));
        value["HelloSnapshot"]["call_to_play"]["events"][0]["actor_name"] = json!("Mallory");
        assert!(Response::decode(Bytes::from(serde_json::to_vec(&value).expect("json"))).is_err());
    }

    #[test]
    fn oversize_wire_collections_are_truncated_to_one_past_their_limit() {
        // Build the JSON by hand: the encoder refuses to produce these frames.
        let event_json = serde_json::to_string(&event(CallToPlayAction::Rsvp)).expect("event json");
        let events =
            std::iter::repeat_n(event_json.as_str(), MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR * 4)
                .collect::<Vec<_>>()
                .join(",");
        let snapshot = format!(r#"{{"revision":1,"display_name":"Alice","events":[{events}]}}"#);
        let decoded: CallToPlayAuthorSnapshot =
            serde_json::from_str(&snapshot).expect("oversize sequence still parses");
        assert_eq!(decoded.events.len(), MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR + 1);
        assert!(matches!(
            decoded.validate(),
            Err(ControlValidationError::TooManyItems {
                field: "Call to Play author events",
                ..
            })
        ));

        let content_id = content(1);
        let games = (0..MAX_LIBRARY_GAMES * 4)
            .map(|index| format!(r#"{{"game_id":"game-{index:05}","content_id":"{content_id}"}}"#))
            .collect::<Vec<_>>()
            .join(",");
        let library = format!(r#"{{"revision":1,"games":[{games}]}}"#);
        let decoded: LibrarySnapshot =
            serde_json::from_str(&library).expect("oversize sequence still parses");
        assert_eq!(decoded.games.len(), MAX_LIBRARY_GAMES + 1);
        assert!(matches!(
            decoded.validate(),
            Err(ControlValidationError::TooManyItems {
                field: "library games",
                ..
            })
        ));

        // A sequence exactly at the limit is untouched.
        let games = (0..MAX_LIBRARY_GAMES)
            .map(|index| format!(r#"{{"game_id":"game-{index:05}","content_id":"{content_id}"}}"#))
            .collect::<Vec<_>>()
            .join(",");
        let decoded: LibrarySnapshot =
            serde_json::from_str(&format!(r#"{{"revision":1,"games":[{games}]}}"#))
                .expect("sequence at the limit parses");
        assert_eq!(decoded.games.len(), MAX_LIBRARY_GAMES);
        decoded.validate().expect("sequence at the limit is valid");
    }

    #[test]
    fn response_decode_isolates_domain_validation() {
        let invalid_call_to_play = state_snapshot(vec![
            event(CallToPlayAction::Rsvp);
            MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR + 1
        ]);
        let decoded = Response::decode(Bytes::from(
            serde_json::to_vec(&Response::HelloSnapshot(invalid_call_to_play.clone()))
                .expect("structural response should serialize"),
        ))
        .expect("semantic domain validation must not reject the response frame");
        let Response::HelloSnapshot(decoded) = decoded else {
            panic!("decoded the wrong response variant");
        };
        decoded.library.validate().expect("library is valid");
        assert_eq!(
            decoded.call_to_play.events.len(),
            MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR + 1
        );
        assert!(decoded.call_to_play.validate().is_err());
        assert!(
            Response::HelloSnapshot(invalid_call_to_play)
                .encode()
                .is_err()
        );

        let mut invalid_library = state_snapshot(Vec::new());
        invalid_library.library.games.swap(0, 1);
        let decoded = Response::decode(Bytes::from(
            serde_json::to_vec(&Response::HelloSnapshot(invalid_library.clone()))
                .expect("structural response should serialize"),
        ))
        .expect("semantic domain validation must not reject the response frame");
        let Response::HelloSnapshot(decoded) = decoded else {
            panic!("decoded the wrong response variant");
        };
        assert!(decoded.library.validate().is_err());
        decoded
            .call_to_play
            .validate()
            .expect("Call to Play slice is valid");
        assert!(Response::HelloSnapshot(invalid_library).encode().is_err());
    }

    #[test]
    fn canonical_wire_scalars_are_enforced_during_decode() {
        let peer_id = peer(1).to_string();
        let runtime = session(0xab).to_string();
        for invalid in [
            format!(
                r#"{{"LibraryChanged":{{"claimed_peer_id":"{}","runtime_session_id":"{runtime}","revision":1}}}}"#,
                peer_id.to_ascii_uppercase()
            ),
            format!(
                r#"{{"LibraryChanged":{{"claimed_peer_id":"{peer_id}","runtime_session_id":"{}","revision":1}}}}"#,
                runtime.to_ascii_uppercase()
            ),
            format!(
                r#"{{"GetGameFileChunk":{{"game_id":"game","content_id":"{}","relative_path":"file","offset":0,"length":1}}}}"#,
                content(0xab).to_string().to_ascii_uppercase()
            ),
            format!(
                r#"{{"GetGameFileChunk":{{"game_id":"game","content_id":"{}","relative_path":"../file","offset":0,"length":1}}}}"#,
                content(3)
            ),
        ] {
            assert!(Request::decode(Bytes::from(invalid)).is_err());
        }
    }

    #[test]
    fn established_resource_boundaries_are_exact() {
        for game_id in [
            "g".repeat(MAX_GAME_ID_BYTES),
            "é".repeat(MAX_GAME_ID_BYTES / 2),
        ] {
            Request::StreamInstall {
                game_id,
                content_id: content(1),
            }
            .encode()
            .expect("bounded game ID should encode");
        }
        assert!(
            Request::StreamInstall {
                game_id: "g".repeat(MAX_GAME_ID_BYTES + 1),
                content_id: content(1),
            }
            .encode()
            .is_err()
        );
        for game_id in [
            ".",
            "..",
            "../game",
            "game/../other",
            "dir/game",
            "dir\\game",
            "game\0",
            "game\n",
            "game.",
            "game ",
            "CON",
            "nul.txt",
            "com1",
            "C:game",
            "game?",
            "game*",
        ] {
            assert!(
                matches!(
                    Request::StreamInstall {
                        game_id: game_id.to_owned(),
                        content_id: content(1),
                    }
                    .encode(),
                    Err(ControlCodecError::Invalid(
                        ControlValidationError::InvalidPathComponent { field: "game ID" }
                    ))
                ),
                "accepted game ID {game_id:?}"
            );
        }
        for game_id in ["game..v1 (final)", "console.txt", "com10", "Jörg"] {
            Request::StreamInstall {
                game_id: game_id.to_owned(),
                content_id: content(1),
            }
            .encode()
            .expect("catalog-valid game ID should encode");
        }

        let mut snapshot = state_snapshot(Vec::new());
        snapshot.call_to_play.display_name = "é".repeat(MAX_CALL_TO_PLAY_DISPLAY_NAME_CHARS);
        Response::HelloSnapshot(snapshot.clone())
            .encode()
            .expect("bounded name");
        snapshot.call_to_play.display_name = "x".repeat(MAX_CALL_TO_PLAY_DISPLAY_NAME_CHARS + 1);
        assert!(Response::HelloSnapshot(snapshot).encode().is_err());

        let bounded_message = "x".repeat(MAX_CALL_TO_PLAY_MESSAGE_CHARS);
        Response::HelloSnapshot(state_snapshot(vec![event(CallToPlayAction::SendMessage {
            text: bounded_message,
        })]))
        .encode()
        .expect("bounded message");
        assert!(
            Response::HelloSnapshot(state_snapshot(vec![event(CallToPlayAction::SendMessage {
                text: "x".repeat(MAX_CALL_TO_PLAY_MESSAGE_CHARS + 1),
            },)]))
            .encode()
            .is_err()
        );

        let bounded_events =
            vec![event(CallToPlayAction::Rsvp); MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR];
        Response::HelloSnapshot(state_snapshot(bounded_events.clone()))
            .encode()
            .expect("bounded events");
        let mut oversized_events = bounded_events;
        oversized_events.push(event(CallToPlayAction::Rsvp));
        assert!(
            Response::HelloSnapshot(state_snapshot(oversized_events))
                .encode()
                .is_err()
        );
    }

    #[test]
    fn library_is_bounded_unique_and_strictly_sorted() {
        let games = (0..MAX_LIBRARY_GAMES)
            .map(|index| GameAvailability {
                game_id: format!("game-{index:04}"),
                content_id: content(1),
            })
            .collect::<Vec<_>>();
        let mut snapshot = state_snapshot(Vec::new());
        snapshot.library.games = games.clone();
        Response::HelloSnapshot(snapshot)
            .encode()
            .expect("bounded sorted library");

        let mut oversized = games.clone();
        oversized.push(GameAvailability {
            game_id: "zzzz".to_owned(),
            content_id: content(1),
        });
        let mut snapshot = state_snapshot(Vec::new());
        snapshot.library.games = oversized;
        assert!(matches!(
            Response::HelloSnapshot(snapshot).encode(),
            Err(ControlCodecError::Invalid(
                ControlValidationError::TooManyItems { .. }
            ))
        ));

        for games in [
            vec![
                GameAvailability {
                    game_id: "same".to_owned(),
                    content_id: content(1),
                },
                GameAvailability {
                    game_id: "same".to_owned(),
                    content_id: content(2),
                },
            ],
            vec![
                GameAvailability {
                    game_id: "z".to_owned(),
                    content_id: content(1),
                },
                GameAvailability {
                    game_id: "a".to_owned(),
                    content_id: content(2),
                },
            ],
        ] {
            let mut snapshot = state_snapshot(Vec::new());
            snapshot.library.games = games;
            assert!(Response::HelloSnapshot(snapshot).encode().is_err());
        }
    }

    #[test]
    fn control_frame_limit_and_single_document_are_enforced() {
        assert!(matches!(
            Request::decode(Bytes::from(vec![b' '; MAX_REQUEST_FRAME_BYTES])),
            Err(ControlCodecError::Decode(_))
        ));
        assert!(matches!(
            Request::decode(Bytes::from(vec![b' '; MAX_REQUEST_FRAME_BYTES + 1])),
            Err(ControlCodecError::FrameTooLarge { actual, maximum })
                if actual == MAX_REQUEST_FRAME_BYTES + 1 && maximum == MAX_REQUEST_FRAME_BYTES
        ));
        assert!(matches!(
            Response::decode(Bytes::from(vec![b' '; MAX_CONTROL_FRAME_BYTES])),
            Err(ControlCodecError::Decode(_))
        ));
        assert!(matches!(
            Response::decode(Bytes::from(vec![b' '; MAX_CONTROL_FRAME_BYTES + 1])),
            Err(ControlCodecError::FrameTooLarge { actual, maximum })
                if actual == MAX_CONTROL_FRAME_BYTES + 1 && maximum == MAX_CONTROL_FRAME_BYTES
        ));
        assert!(matches!(
            Request::decode(Bytes::from_static(br#""Ping""Hello""#)),
            Err(ControlCodecError::Decode(_))
        ));

        let escaped = format!("x{}", "\0".repeat(MAX_CALL_TO_PLAY_MESSAGE_CHARS - 1));
        let events = vec![
            event(CallToPlayAction::SendMessage { text: escaped });
            MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR
        ];
        assert!(matches!(
            Response::HelloSnapshot(state_snapshot(events)).encode(),
            Err(ControlCodecError::Invalid(
                ControlValidationError::EncodedTooLarge { .. }
            ))
        ));
    }

    #[test]
    fn maximal_author_budget_still_fits_with_a_bounded_library() {
        fn author_with_message_chars(char_count: usize) -> CallToPlayAuthorSnapshot {
            let text = "😀".repeat(char_count);
            let events = (0..MAX_CALL_TO_PLAY_EVENTS_PER_AUTHOR)
                .map(|index| {
                    let mut event = event(CallToPlayAction::SendMessage { text: text.clone() });
                    event.id = EventNonce::from_bytes((index as u128).to_be_bytes());
                    event
                })
                .collect();
            state_snapshot(events).call_to_play
        }

        let mut accepted_chars = 0;
        let mut rejected_chars = MAX_CALL_TO_PLAY_MESSAGE_CHARS + 1;
        while accepted_chars + 1 < rejected_chars {
            let candidate = usize::midpoint(accepted_chars, rejected_chars);
            if author_with_message_chars(candidate).validate().is_ok() {
                accepted_chars = candidate;
            } else {
                rejected_chars = candidate;
            }
        }
        assert!(accepted_chars > 0);
        assert!(rejected_chars <= MAX_CALL_TO_PLAY_MESSAGE_CHARS);

        let accepted_author = author_with_message_chars(accepted_chars);
        accepted_author.validate().expect("author at byte budget");
        let rejected_author = author_with_message_chars(rejected_chars);
        assert!(matches!(
            rejected_author.validate(),
            Err(ControlValidationError::EncodedTooLarge { .. })
        ));

        let games = (0..MAX_LIBRARY_GAMES)
            .map(|index| GameAvailability {
                game_id: format!("{index:04}-{}", "g".repeat(MAX_GAME_ID_BYTES - 5)),
                content_id: content(1),
            })
            .collect();
        let response = Response::HelloSnapshot(PeerStateSnapshot {
            runtime_session_id: session(1),
            library: LibrarySnapshot { revision: 1, games },
            call_to_play: accepted_author,
        });
        let encoded = response
            .encode()
            .expect("a maximally accepted author plus bounded library must fit");
        assert!(encoded.len() <= MAX_CONTROL_FRAME_BYTES);

        let isolated = PeerStateSnapshot {
            runtime_session_id: session(1),
            library: LibrarySnapshot {
                revision: 1,
                games: Vec::new(),
            },
            call_to_play: rejected_author,
        };
        let decoded = Response::decode(Bytes::from(
            serde_json::to_vec(&Response::HelloSnapshot(isolated))
                .expect("structural response should serialize"),
        ))
        .expect("invalid author remains domain-isolated during response decode");
        let Response::HelloSnapshot(decoded) = decoded else {
            panic!("wrong response variant");
        };
        decoded.library.validate().expect("library remains valid");
        assert!(matches!(
            decoded.call_to_play.validate(),
            Err(ControlValidationError::EncodedTooLarge { .. })
        ));
    }

    #[test]
    fn removed_v7_control_surface_has_no_decode_fallback() {
        for variant in [
            "ListGames",
            "GetGame",
            "GetGameFileData",
            "LibraryDelta",
            "CallToPlayEvents",
            "Goodbye",
            "Invalid",
        ] {
            let old = format!(r#"{{"{variant}":{{}}}}"#);
            assert!(
                Request::decode(Bytes::from(old)).is_err(),
                "accepted v7 {variant}"
            );
        }
        for variant in [
            "ListGames",
            "GetGame",
            "HelloAck",
            "CallToPlayAck",
            "GameNotFound",
            "InvalidRequest",
            "EncodingError",
            "DecodingError",
            "InternalPeerError",
        ] {
            let old = format!(r#"{{"{variant}":{{}}}}"#);
            assert!(
                Response::decode(Bytes::from(old)).is_err(),
                "accepted v7 {variant}"
            );
        }
        assert!(Response::decode(Bytes::from_static(br#""Pong""#)).is_err());
        assert!(Request::decode(Bytes::from_static(
            br#"{"GetGameFileChunk":{"game_id":"game","relative_path":"file","offset":0,"length":1}}"#,
        )).is_err());
        assert!(
            Request::decode(Bytes::from_static(
                br#"{"StreamInstall":{"game_id":"game"}}"#,
            ))
            .is_err()
        );

        // Ping is intentionally a unit value in both schemas; version-bound ALPN
        // rejects a protocol-7 connection before any control JSON is decoded.
        assert_eq!(
            Request::decode(Bytes::from_static(br#""Ping""#)).expect("v8 Ping should decode"),
            Request::Ping
        );
    }

    #[test]
    fn stream_install_paths_are_canonical_but_codec_semantics_stay_in_band() {
        let frame = StreamInstallFrame::FileBegin {
            relative_path: path("bin/game.exe"),
            size: 42,
            crc32: 7,
        };
        assert_eq!(StreamInstallFrame::decode(frame.encode()), frame);
        match StreamInstallFrame::decode(Bytes::new()) {
            StreamInstallFrame::Error { message } => assert!(message.contains("empty")),
            other => panic!("expected error frame, got {other:?}"),
        }
        let invalid = Bytes::from_static(b"\0{\"Directory\":{\"relative_path\":\"../escape\"}}");
        assert!(matches!(
            StreamInstallFrame::decode(invalid),
            StreamInstallFrame::Error { .. }
        ));
    }

    #[test]
    fn checked_stream_decode_distinguishes_wire_errors_from_sender_errors() {
        let explicit = StreamInstallFrame::Error {
            message: "provider unavailable".to_owned(),
        };
        assert_eq!(
            StreamInstallFrame::decode_checked(explicit.encode()).expect("explicit Error is data"),
            explicit
        );
        assert!(matches!(
            StreamInstallFrame::decode_checked(Bytes::new()),
            Err(StreamInstallFrameDecodeError::Empty)
        ));
        assert!(matches!(
            StreamInstallFrame::decode_checked(Bytes::from_static(b"\x7f{}")),
            Err(StreamInstallFrameDecodeError::UnknownTag(0x7f))
        ));
        assert!(matches!(
            StreamInstallFrame::decode_checked(Bytes::from_static(
                b"\0{\"Directory\":{\"relative_path\":\"../escape\"}}"
            )),
            Err(StreamInstallFrameDecodeError::InvalidControl(_))
        ));

        let file_chunk_under_control = tagged_stream_install_frame(
            STREAM_INSTALL_CONTROL_FRAME_TAG,
            &serde_json::to_vec(&StreamInstallFrame::FileChunk {
                bytes: Bytes::from_static(b"bytes"),
            })
            .expect("serialize test frame"),
        );
        assert!(matches!(
            StreamInstallFrame::decode_checked(file_chunk_under_control),
            Err(StreamInstallFrameDecodeError::FileChunkInControl)
        ));
        assert!(matches!(
            StreamInstallFrame::decode_checked(Bytes::from(vec![
                STREAM_INSTALL_FILE_CHUNK_FRAME_TAG;
                MAX_STREAM_INSTALL_FRAME_BYTES + 1
            ])),
            Err(StreamInstallFrameDecodeError::FrameTooLarge { actual, maximum })
                if actual == MAX_STREAM_INSTALL_FRAME_BYTES + 1
                    && maximum == MAX_STREAM_INSTALL_FRAME_BYTES
        ));
    }
}