signal_fish_client/

event.rs

//! High-level events emitted by the Signal Fish client.
//!
//! [`SignalFishEvent`] provides a 1:1 mapping from every [`ServerMessage`] variant
//! plus three synthetic events (`Connected`, `Disconnected`, and `DecodeFailed`)
//! that originate from the transport layer rather than the server.
//!
//! Boxed payload types ([`RoomJoinedPayload`], [`ReconnectedPayload`],
//! [`SpectatorJoinedPayload`]) are flattened into inline fields so callers can
//! pattern-match directly without an extra dereference.
//!
//! [`ServerMessage`]: crate::protocol::ServerMessage
//! [`RoomJoinedPayload`]: crate::protocol::RoomJoinedPayload
//! [`ReconnectedPayload`]: crate::protocol::ReconnectedPayload
//! [`SpectatorJoinedPayload`]: crate::protocol::SpectatorJoinedPayload

use crate::error_codes::ErrorCode;
use crate::protocol::{
    DeliveryClass, DeliveryReportPayload, GameDataEncoding, IceServer, LobbyState,
    PeerConnectionInfo, PlayerId, PlayerInfo, ProtocolInfoPayload, RateLimitInfo, ReplayStatus,
    RoomId, SenderWatermark, ServerMessage, SessionPeer, SpectatorInfo, SpectatorStateChangeReason,
    Topology, TransportKind,
};

/// Events emitted by the Signal Fish client.
///
/// Each variant corresponds to either a [`ServerMessage`] received from the
/// signaling server or a synthetic event generated by the transport layer.
///
/// # Synthetic events
///
/// | Variant | Origin |
/// |---|---|
/// | [`Connected`](Self::Connected) | Transport layer opened successfully |
/// | [`Disconnected`](Self::Disconnected) | Transport layer closed or errored |
/// | [`DecodeFailed`](Self::DecodeFailed) | An inbound frame could not be decoded |
///
/// # Example
///
/// ```text
/// // Assuming `events` is an async receiver of SignalFishEvent:
/// match event {
///     SignalFishEvent::RoomJoined { room_code, current_players, .. } => { /* … */ }
///     SignalFishEvent::PlayerJoined { player } => { /* … */ }
///     SignalFishEvent::Disconnected { reason, .. } => { /* … */ }
///     _ => {}
/// }
/// ```
#[derive(Clone)]
pub enum SignalFishEvent {
    // ── Synthetic events ────────────────────────────────────────────
    /// The client has started and will begin communicating with the server.
    ///
    /// This is a **synthetic event** — it is not triggered by a server message.
    ///
    /// - **`SignalFishClient`** (async): emitted at the start of the transport
    ///   loop, after the transport has already been connected via
    ///   `.connect().await`.
    /// - **`SignalFishPollingClient`**: emitted once
    ///   [`Transport::is_ready()`](crate::Transport::is_ready) returns `true`
    ///   during a [`poll()`](crate::SignalFishPollingClient::poll) cycle. For
    ///   transports that are already connected at construction time, this is
    ///   the first `poll()` call. For transports with asynchronous handshakes
    ///   (e.g., `EmscriptenWebSocketTransport`), `Connected` is deferred
    ///   until the handshake completes.
    Connected,

    /// The transport connection was closed.
    Disconnected {
        /// Human-readable reason for the disconnection, if available.
        ///
        /// When the transport captured a WebSocket Close frame with a reason
        /// (see [`Transport::close_info`](crate::Transport::close_info)),
        /// it is included here as `"closed by server: …"`.
        reason: Option<String>,
        /// The most recent `Error`/`AuthenticationError` received on this
        /// connection, if any.
        ///
        /// This is a correlation aid, not a server-attributed close reason:
        /// a server that evicts a slow consumer writes a best-effort
        /// `Error { error_code: SlowConsumer }` farewell before closing, and
        /// when that frame arrives it is surfaced here so the disconnect can
        /// be attributed. The farewell may never arrive (the socket is
        /// congested by definition), in which case this is `None`.
        last_server_error: Option<ServerErrorInfo>,
    },

    /// An inbound server frame could not be decoded into a
    /// [`ServerMessage`].
    ///
    /// This is a **synthetic event**. The connection stays open and later
    /// frames are unaffected. Typical causes: a server newer than this SDK
    /// (an unknown message `type`, or an unknown `error_code` string inside
    /// an otherwise-known message), a proxy injecting non-protocol frames,
    /// or payload corruption.
    ///
    /// Every undecodable frame processed during normal operation produces
    /// exactly one `DecodeFailed` event (no coalescing) and increments the
    /// `messages_undecodable` counter in
    /// [`ClientStats`](crate::ClientStats). Steady growth of that counter
    /// means protocol drift (upgrade this SDK) or a corrupting middlebox. The
    /// async driver delivers the event with channel backpressure; the polling
    /// driver returns it from the current bounded polling cycle. Explicit
    /// shutdown/close boundaries retain their documented delivery caveats.
    DecodeFailed {
        /// The wire `type` tag, when the frame was valid JSON with one.
        ///
        /// `Some("Error")` plus a decode failure strongly implies an
        /// `error_code` string this SDK does not know; any other
        /// `Some(...)` implies an unknown or malformed message type;
        /// `None` implies the frame was not valid JSON at all.
        message_type: Option<String>,
        /// The deserialization error text.
        error: String,
        /// The raw frame text, truncated to at most
        /// [`DECODE_FAILED_RAW_PREFIX_MAX`] bytes on a UTF-8 boundary.
        raw_prefix: String,
    },

    /// The server violated protocol-v3 delivery-accountability invariants.
    ProtocolViolation {
        /// Stable category suitable for metrics and policy handling.
        kind: ProtocolViolationKind,
        /// Detailed diagnostic retaining sender/epoch/sequence context.
        diagnostic: String,
    },

    // ── Authentication ──────────────────────────────────────────────
    /// Authentication succeeded.
    Authenticated {
        /// Application name confirmed by the server.
        app_name: String,
        /// Organization the app belongs to, if any.
        organization: Option<String>,
        /// Rate limits enforced for this application.
        rate_limits: RateLimitInfo,
    },

    /// SDK/protocol compatibility details advertised after authentication.
    ///
    /// Wrapped as a single payload rather than flattened because most fields
    /// are optional configuration details that callers typically access as a group.
    ProtocolInfo(ProtocolInfoPayload),

    /// Authentication failed.
    AuthenticationError {
        /// Human-readable error description.
        error: String,
        /// Structured error code for programmatic handling.
        error_code: ErrorCode,
    },

    // ── Room lifecycle ──────────────────────────────────────────────
    /// Successfully joined a room. Fields are flattened from [`RoomJoinedPayload`].
    ///
    /// [`RoomJoinedPayload`]: crate::protocol::RoomJoinedPayload
    RoomJoined {
        /// Unique room identifier.
        room_id: RoomId,
        /// Human-readable room code.
        room_code: String,
        /// The local player's identifier.
        player_id: PlayerId,
        /// Name of the game this room is for.
        game_name: String,
        /// Maximum number of players allowed.
        max_players: u8,
        /// Whether the room supports authority delegation.
        supports_authority: bool,
        /// Players already present in the room.
        current_players: Vec<PlayerInfo>,
        /// Whether the local player is the authority.
        is_authority: bool,
        /// Current lobby readiness state.
        lobby_state: LobbyState,
        /// Players that have signaled readiness.
        ready_players: Vec<PlayerId>,
        /// Relay transport type label (e.g. `"auto"`, `"tcp"`).
        relay_type: String,
        /// Spectators currently watching.
        current_spectators: Vec<SpectatorInfo>,
        /// ICE (STUN/TURN) servers for early WebRTC candidate gathering
        /// (protocol v3 "pre-gather"). Empty unless the server opted this
        /// connection into ICE pre-gather.
        ice_servers: Vec<IceServer>,
        /// Server-issued token to retain for an unexpected disconnect.
        reconnection_token: Option<String>,
    },

    /// Failed to join a room.
    RoomJoinFailed {
        /// Human-readable failure reason.
        reason: String,
        /// Structured error code, if provided.
        error_code: Option<ErrorCode>,
    },

    /// Successfully left the current room.
    RoomLeft,

    // ── Player presence ─────────────────────────────────────────────
    /// Another player joined the room.
    PlayerJoined {
        /// Information about the new player.
        player: PlayerInfo,
    },

    /// Another player left the room.
    PlayerLeft {
        /// Identifier of the player who left.
        player_id: PlayerId,
        /// Departed incarnation epoch (protocol v3 only).
        epoch: Option<u32>,
        /// Terminal relay watermark (protocol v3 only).
        final_seq: Option<u64>,
    },

    // ── Game data ───────────────────────────────────────────────────
    /// JSON game data received from another player.
    GameData {
        /// Identifier of the sending player.
        from_player: PlayerId,
        /// Arbitrary JSON payload.
        data: serde_json::Value,
        /// Server-stamped sequence number (protocol v3 only).
        seq: Option<u64>,
        /// Server-tracked sender incarnation (protocol v3 only).
        epoch: Option<u32>,
        /// Echoed delivery class (protocol v3 only).
        class: Option<DeliveryClass>,
        /// Coalescing key for latest delivery (protocol v3 only).
        key: Option<u32>,
    },

    /// Binary game data received from another player.
    GameDataBinary {
        /// Identifier of the sending player.
        from_player: PlayerId,
        /// Encoding format of the binary payload.
        encoding: GameDataEncoding,
        /// Raw binary payload.
        payload: Vec<u8>,
        /// Mandatory non-zero server sequence for a v3 binary envelope.
        seq: Option<u64>,
        /// Mandatory non-zero sender incarnation for a v3 binary envelope.
        epoch: Option<u32>,
    },

    // ── Authority ───────────────────────────────────────────────────
    /// The room's authority assignment changed.
    AuthorityChanged {
        /// The player who now holds authority, if any.
        authority_player: Option<PlayerId>,
        /// Whether the local player is now the authority.
        you_are_authority: bool,
    },

    /// Response to an authority request.
    AuthorityResponse {
        /// Whether the request was granted.
        granted: bool,
        /// Human-readable reason if the request was denied.
        reason: Option<String>,
        /// Structured error code, if provided.
        error_code: Option<ErrorCode>,
    },

    // ── Lobby ───────────────────────────────────────────────────────
    /// The lobby readiness state changed.
    LobbyStateChanged {
        /// New lobby state.
        lobby_state: LobbyState,
        /// Players that have signaled readiness.
        ready_players: Vec<PlayerId>,
        /// Whether all players are ready.
        all_ready: bool,
    },

    /// The game is starting with peer connection information.
    GameStarting {
        /// Connection details for every peer.
        peer_connections: Vec<PeerConnectionInfo>,
    },

    // ── Mesh / WebRTC signaling (protocol v3) ───────────────────────
    /// The server delivered this client's per-recipient session plan.
    ///
    /// **Protocol v3 only.** Arrives only on a v3-negotiated connection.
    ///
    /// May arrive multiple times (host re-election, late-join re-plan); each
    /// one **fully replaces** the previous plan. Fields are flattened from
    /// [`SessionPlanPayload`].
    ///
    /// [`SessionPlanPayload`]: crate::protocol::SessionPlanPayload
    SessionPlan {
        /// Chosen session topology.
        topology: Topology,
        /// Chosen data-path transport.
        transport: TransportKind,
        /// The elected host (present for `host` topology).
        host: Option<PlayerId>,
        /// Peers this client should connect to, each with its `initiate` flag.
        peers: Vec<SessionPeer>,
        /// ICE (STUN/TURN) servers for WebRTC.
        ice_servers: Vec<IceServer>,
        /// The universal fallback transport (always relay).
        fallback: TransportKind,
    },

    /// A late-joining peer to connect to after the session was finalized.
    ///
    /// **Protocol v3 only.** Arrives only on a v3-negotiated connection.
    NewPeer {
        /// The new peer's identifier.
        peer_id: PlayerId,
        /// Whether this client sends the WebRTC offer (server-assigned; obey it).
        you_initiate: bool,
    },

    /// An opaque WebRTC signal relayed from a peer.
    ///
    /// **Protocol v3 only.** Arrives only on a v3-negotiated connection.
    ///
    /// Convert with [`PeerSignal::try_from(&signal)`](crate::PeerSignal) for the
    /// common offer/answer/ICE-candidate shapes; the raw `Value` is preserved
    /// for any other shape.
    SignalReceived {
        /// The peer the signal came from.
        from: PlayerId,
        /// The opaque signal payload.
        signal: serde_json::Value,
    },

    /// A peer's data-path transport state changed (informational).
    ///
    /// **Protocol v3 only.** Arrives only on a v3-negotiated connection.
    PeerTransportStatus {
        /// The peer whose transport state changed.
        peer_id: PlayerId,
        /// The transport being reported on.
        transport: TransportKind,
        /// Whether that transport is connected for the peer.
        connected: bool,
    },

    /// Cumulative relay-delivery diagnostics (protocol v3 only).
    RelayStats {
        interval_ms: u64,
        sent_to_you: u64,
        dropped_for_you: u64,
        backpressure_events: u64,
    },

    /// Graceful server-shutdown advisory (protocol v3 only).
    GoingAway {
        deadline_ms: u64,
        retry_after_secs: Option<u64>,
    },

    /// Exact delivery-accountability report (protocol v3 only).
    DeliveryReport(DeliveryReportPayload),

    // ── Heartbeat ───────────────────────────────────────────────────
    /// Pong response to a ping.
    Pong,

    // ── Reconnection ────────────────────────────────────────────────
    /// Reconnection succeeded. Fields are flattened from [`ReconnectedPayload`].
    ///
    /// [`ReconnectedPayload`]: crate::protocol::ReconnectedPayload
    Reconnected {
        /// Unique room identifier.
        room_id: RoomId,
        /// Human-readable room code.
        room_code: String,
        /// The local player's identifier.
        player_id: PlayerId,
        /// Name of the game this room is for.
        game_name: String,
        /// Maximum number of players allowed.
        max_players: u8,
        /// Whether the room supports authority delegation.
        supports_authority: bool,
        /// Players currently in the room.
        current_players: Vec<PlayerInfo>,
        /// Whether the local player is the authority.
        is_authority: bool,
        /// Current lobby readiness state.
        lobby_state: LobbyState,
        /// Players that have signaled readiness.
        ready_players: Vec<PlayerId>,
        /// Relay transport type label.
        relay_type: String,
        /// Spectators currently watching.
        current_spectators: Vec<SpectatorInfo>,
        /// ICE (STUN/TURN) servers for early WebRTC candidate gathering
        /// (protocol v3 "pre-gather"). Empty unless the server opted this
        /// connection into ICE pre-gather.
        ice_servers: Vec<IceServer>,
        /// Events that occurred while the client was disconnected.
        missed_events: Vec<SignalFishEvent>,
        /// Completeness of the replayed control-event suffix.
        replay: Option<ReplayStatus>,
        /// Authoritative relay baselines for current senders.
        sender_watermarks: Vec<SenderWatermark>,
        /// Fresh token replacing the consumed reconnect token.
        reconnection_token: Option<String>,
    },

    /// Reconnection failed.
    ReconnectionFailed {
        /// Human-readable failure reason.
        reason: String,
        /// Structured error code.
        error_code: ErrorCode,
    },

    /// Another player reconnected to the room.
    PlayerReconnected {
        /// Identifier of the player who reconnected.
        player_id: PlayerId,
        /// New incarnation epoch (protocol v3 only).
        epoch: Option<u32>,
    },

    // ── Spectator ───────────────────────────────────────────────────
    /// Successfully joined a room as a spectator.
    /// Fields are flattened from [`SpectatorJoinedPayload`].
    ///
    /// [`SpectatorJoinedPayload`]: crate::protocol::SpectatorJoinedPayload
    SpectatorJoined {
        /// Unique room identifier.
        room_id: RoomId,
        /// Human-readable room code.
        room_code: String,
        /// The local spectator's identifier.
        spectator_id: PlayerId,
        /// Name of the game this room is for.
        game_name: String,
        /// Players currently in the room.
        current_players: Vec<PlayerInfo>,
        /// Spectators currently watching.
        current_spectators: Vec<SpectatorInfo>,
        /// Current lobby readiness state.
        lobby_state: LobbyState,
        /// Reason the spectator state changed, if applicable.
        reason: Option<SpectatorStateChangeReason>,
    },

    /// Failed to join as a spectator.
    SpectatorJoinFailed {
        /// Human-readable failure reason.
        reason: String,
        /// Structured error code, if provided.
        error_code: Option<ErrorCode>,
    },

    /// Successfully left spectator mode.
    SpectatorLeft {
        /// Room identifier, if available.
        room_id: Option<RoomId>,
        /// Room code, if available.
        room_code: Option<String>,
        /// Reason for leaving, if available.
        reason: Option<SpectatorStateChangeReason>,
        /// Remaining spectators in the room.
        current_spectators: Vec<SpectatorInfo>,
    },

    /// Another spectator joined the room.
    NewSpectatorJoined {
        /// Information about the new spectator.
        spectator: SpectatorInfo,
        /// All spectators currently watching.
        current_spectators: Vec<SpectatorInfo>,
        /// Reason for the state change, if available.
        reason: Option<SpectatorStateChangeReason>,
    },

    /// Another spectator disconnected from the room.
    SpectatorDisconnected {
        /// Identifier of the spectator who disconnected.
        spectator_id: PlayerId,
        /// Reason for disconnection, if available.
        reason: Option<SpectatorStateChangeReason>,
        /// Remaining spectators in the room.
        current_spectators: Vec<SpectatorInfo>,
    },

    // ── Errors ──────────────────────────────────────────────────────
    /// A generic server error.
    Error {
        /// Human-readable error message.
        message: String,
        /// Structured error code, if provided.
        error_code: Option<ErrorCode>,
    },
}

impl std::fmt::Debug for SignalFishEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Events can carry reconnect credentials and arbitrary application
        // payloads. Keep the ubiquitous `Debug` path safe for production logs;
        // callers can explicitly inspect fields after pattern matching.
        f.write_str(match self {
            Self::Connected => "Connected",
            Self::Disconnected { .. } => "Disconnected",
            Self::DecodeFailed { .. } => "DecodeFailed",
            Self::ProtocolViolation { .. } => "ProtocolViolation",
            Self::Authenticated { .. } => "Authenticated",
            Self::ProtocolInfo(_) => "ProtocolInfo",
            Self::AuthenticationError { .. } => "AuthenticationError",
            Self::RoomJoined { .. } => "RoomJoined",
            Self::RoomJoinFailed { .. } => "RoomJoinFailed",
            Self::RoomLeft => "RoomLeft",
            Self::PlayerJoined { .. } => "PlayerJoined",
            Self::PlayerLeft { .. } => "PlayerLeft",
            Self::GameData { .. } => "GameData",
            Self::GameDataBinary { .. } => "GameDataBinary",
            Self::AuthorityChanged { .. } => "AuthorityChanged",
            Self::AuthorityResponse { .. } => "AuthorityResponse",
            Self::LobbyStateChanged { .. } => "LobbyStateChanged",
            Self::GameStarting { .. } => "GameStarting",
            Self::SessionPlan { .. } => "SessionPlan",
            Self::NewPeer { .. } => "NewPeer",
            Self::SignalReceived { .. } => "SignalReceived",
            Self::PeerTransportStatus { .. } => "PeerTransportStatus",
            Self::RelayStats { .. } => "RelayStats",
            Self::GoingAway { .. } => "GoingAway",
            Self::DeliveryReport(_) => "DeliveryReport",
            Self::Pong => "Pong",
            Self::Reconnected { .. } => "Reconnected",
            Self::ReconnectionFailed { .. } => "ReconnectionFailed",
            Self::PlayerReconnected { .. } => "PlayerReconnected",
            Self::SpectatorJoined { .. } => "SpectatorJoined",
            Self::SpectatorJoinFailed { .. } => "SpectatorJoinFailed",
            Self::SpectatorLeft { .. } => "SpectatorLeft",
            Self::NewSpectatorJoined { .. } => "NewSpectatorJoined",
            Self::SpectatorDisconnected { .. } => "SpectatorDisconnected",
            Self::Error { .. } => "Error",
        })
    }
}

/// Category of a delivery-accountability protocol violation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProtocolViolationKind {
    Snapshot,
    Lifecycle,
    DeliveryGap,
    Counters,
    Causality,
    Stamp,
    UnexpectedMetadata,
}

impl ProtocolViolationKind {
    #[cfg(any(test, feature = "tokio-runtime", feature = "polling-client"))]
    pub(crate) fn from_diagnostic(diagnostic: &str) -> Self {
        if diagnostic.contains("immediately followed") || diagnostic.contains("preceding causal") {
            Self::Causality
        } else if diagnostic.contains("PlayerLeft")
            || diagnostic.contains("PlayerReconnected")
            || diagnostic.contains("announced epoch")
        {
            Self::Lifecycle
        } else if diagnostic.contains("snapshot") || diagnostic.contains("watermark") {
            Self::Snapshot
        } else if diagnostic.contains("counter") || diagnostic.contains("RelayStats") {
            Self::Counters
        } else if diagnostic.contains("gap") || diagnostic.contains("DeliveryReport") {
            Self::DeliveryGap
        } else if diagnostic.contains("v2")
            || diagnostic.contains("class/key")
            || diagnostic.contains("frame representation")
        {
            Self::UnexpectedMetadata
        } else {
            Self::Stamp
        }
    }
}

/// Maximum number of bytes of raw frame text preserved in
/// [`SignalFishEvent::DecodeFailed::raw_prefix`].
///
/// Truncation always lands on a UTF-8 character boundary, so the prefix may
/// be a few bytes shorter than this cap.
pub const DECODE_FAILED_RAW_PREFIX_MAX: usize = 512;

/// A server-sent error remembered for disconnect attribution.
///
/// Carried by [`SignalFishEvent::Disconnected::last_server_error`]: the most
/// recent `Error` or `AuthenticationError` frame received on the connection.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServerErrorInfo {
    /// Human-readable error message from the server.
    pub message: String,
    /// Structured error code, if the server provided one.
    pub error_code: Option<ErrorCode>,
}

impl SignalFishEvent {
    /// Builds the [`DecodeFailed`](Self::DecodeFailed) event for a frame that
    /// failed to deserialize.
    ///
    /// Shared by the async and polling clients so both surface identical
    /// diagnostics: the wire `type` tag when the frame was valid JSON, the
    /// serde error text, and a bounded raw prefix.
    #[cfg(any(feature = "tokio-runtime", feature = "polling-client"))]
    pub(crate) fn decode_failed(raw: &str, error: &serde_json::Error) -> Self {
        // Compute the bounded prefix once and reuse it for both the type-tag
        // recovery and the stored `raw_prefix`, so work stays bounded on large
        // or hostile input.
        let prefix = truncate_on_char_boundary(raw, DECODE_FAILED_RAW_PREFIX_MAX);
        // Only recover the wire `type` tag when the frame was well-formed JSON
        // that simply didn't match our types (a `Data` error — e.g. an unknown
        // message type or error-code token). Malformed JSON (`Syntax`/`Eof`)
        // has no recoverable tag, so skip the re-parse entirely rather than
        // re-scanning untrusted garbage. Parsing only the bounded prefix caps
        // the secondary parse; a frame larger than the cap yields `None`
        // (its `type` is still visible in `raw_prefix`).
        let message_type = if error.classify() == serde_json::error::Category::Data {
            serde_json::from_str::<serde_json::Value>(prefix)
                .ok()
                .and_then(|v| v.get("type")?.as_str().map(str::to_string))
        } else {
            None
        };
        Self::DecodeFailed {
            message_type,
            error: error.to_string(),
            raw_prefix: prefix.to_string(),
        }
    }
}

/// Returns the longest prefix of `s` that is at most `max_bytes` long and
/// ends on a UTF-8 character boundary.
#[cfg(any(feature = "tokio-runtime", feature = "polling-client"))]
fn truncate_on_char_boundary(s: &str, max_bytes: usize) -> &str {
    if s.len() <= max_bytes {
        return s;
    }
    let mut end = max_bytes;
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    s.get(..end).unwrap_or_default()
}

// ── Conversion ──────────────────────────────────────────────────────

impl From<ServerMessage> for SignalFishEvent {
    fn from(msg: ServerMessage) -> Self {
        match msg {
            ServerMessage::Authenticated {
                app_name,
                organization,
                rate_limits,
            } => Self::Authenticated {
                app_name,
                organization,
                rate_limits,
            },
            ServerMessage::ProtocolInfo(payload) => Self::ProtocolInfo(payload),
            ServerMessage::AuthenticationError { error, error_code } => {
                Self::AuthenticationError { error, error_code }
            }
            ServerMessage::RoomJoined(payload) => {
                let p = *payload;
                Self::RoomJoined {
                    room_id: p.room_id,
                    room_code: p.room_code,
                    player_id: p.player_id,
                    game_name: p.game_name,
                    max_players: p.max_players,
                    supports_authority: p.supports_authority,
                    current_players: p.current_players,
                    is_authority: p.is_authority,
                    lobby_state: p.lobby_state,
                    ready_players: p.ready_players,
                    relay_type: p.relay_type,
                    current_spectators: p.current_spectators,
                    ice_servers: p.ice_servers,
                    reconnection_token: p.reconnection_token,
                }
            }
            ServerMessage::RoomJoinFailed { reason, error_code } => {
                Self::RoomJoinFailed { reason, error_code }
            }
            ServerMessage::RoomLeft => Self::RoomLeft,
            ServerMessage::PlayerJoined { player } => Self::PlayerJoined { player },
            ServerMessage::PlayerLeft {
                player_id,
                epoch,
                final_seq,
            } => Self::PlayerLeft {
                player_id,
                epoch,
                final_seq,
            },
            ServerMessage::GameData {
                from_player,
                data,
                seq,
                epoch,
                class,
                key,
            } => Self::GameData {
                from_player,
                data,
                seq,
                epoch,
                class,
                key,
            },
            ServerMessage::GameDataBinary {
                from_player,
                encoding,
                payload,
                seq,
                epoch,
            } => Self::GameDataBinary {
                from_player,
                encoding,
                payload,
                seq,
                epoch,
            },
            ServerMessage::AuthorityChanged {
                authority_player,
                you_are_authority,
            } => Self::AuthorityChanged {
                authority_player,
                you_are_authority,
            },
            ServerMessage::AuthorityResponse {
                granted,
                reason,
                error_code,
            } => Self::AuthorityResponse {
                granted,
                reason,
                error_code,
            },
            ServerMessage::LobbyStateChanged {
                lobby_state,
                ready_players,
                all_ready,
            } => Self::LobbyStateChanged {
                lobby_state,
                ready_players,
                all_ready,
            },
            ServerMessage::GameStarting { peer_connections } => {
                Self::GameStarting { peer_connections }
            }
            ServerMessage::Pong => Self::Pong,
            ServerMessage::Reconnected(payload) => {
                let p = *payload;
                Self::Reconnected {
                    room_id: p.room_id,
                    room_code: p.room_code,
                    player_id: p.player_id,
                    game_name: p.game_name,
                    max_players: p.max_players,
                    supports_authority: p.supports_authority,
                    current_players: p.current_players,
                    is_authority: p.is_authority,
                    lobby_state: p.lobby_state,
                    ready_players: p.ready_players,
                    relay_type: p.relay_type,
                    current_spectators: p.current_spectators,
                    ice_servers: p.ice_servers,
                    missed_events: p
                        .missed_events
                        .into_iter()
                        .map(SignalFishEvent::from)
                        .collect(),
                    replay: p.replay,
                    sender_watermarks: p.sender_watermarks,
                    reconnection_token: p.reconnection_token,
                }
            }
            ServerMessage::ReconnectionFailed { reason, error_code } => {
                Self::ReconnectionFailed { reason, error_code }
            }
            ServerMessage::PlayerReconnected { player_id, epoch } => {
                Self::PlayerReconnected { player_id, epoch }
            }
            ServerMessage::SpectatorJoined(payload) => {
                let p = *payload;
                Self::SpectatorJoined {
                    room_id: p.room_id,
                    room_code: p.room_code,
                    spectator_id: p.spectator_id,
                    game_name: p.game_name,
                    current_players: p.current_players,
                    current_spectators: p.current_spectators,
                    lobby_state: p.lobby_state,
                    reason: p.reason,
                }
            }
            ServerMessage::SpectatorJoinFailed { reason, error_code } => {
                Self::SpectatorJoinFailed { reason, error_code }
            }
            ServerMessage::SpectatorLeft {
                room_id,
                room_code,
                reason,
                current_spectators,
            } => Self::SpectatorLeft {
                room_id,
                room_code,
                reason,
                current_spectators,
            },
            ServerMessage::NewSpectatorJoined {
                spectator,
                current_spectators,
                reason,
            } => Self::NewSpectatorJoined {
                spectator,
                current_spectators,
                reason,
            },
            ServerMessage::SpectatorDisconnected {
                spectator_id,
                reason,
                current_spectators,
            } => Self::SpectatorDisconnected {
                spectator_id,
                reason,
                current_spectators,
            },
            ServerMessage::Error {
                message,
                error_code,
            } => Self::Error {
                message,
                error_code,
            },
            ServerMessage::Signal { from, signal } => Self::SignalReceived { from, signal },
            ServerMessage::NewPeer {
                peer_id,
                you_initiate,
            } => Self::NewPeer {
                peer_id,
                you_initiate,
            },
            ServerMessage::SessionPlan(payload) => {
                let p = *payload;
                Self::SessionPlan {
                    topology: p.topology,
                    transport: p.transport,
                    host: p.host,
                    peers: p.peers,
                    ice_servers: p.ice_servers,
                    fallback: p.fallback,
                }
            }
            ServerMessage::PeerTransportStatus {
                peer_id,
                transport,
                connected,
            } => Self::PeerTransportStatus {
                peer_id,
                transport,
                connected,
            },
            ServerMessage::RelayStats {
                interval_ms,
                sent_to_you,
                dropped_for_you,
                backpressure_events,
            } => Self::RelayStats {
                interval_ms,
                sent_to_you,
                dropped_for_you,
                backpressure_events,
            },
            ServerMessage::GoingAway {
                deadline_ms,
                retry_after_secs,
            } => Self::GoingAway {
                deadline_ms,
                retry_after_secs,
            },
            ServerMessage::DeliveryReport(payload) => Self::DeliveryReport(*payload),
        }
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic,
    clippy::todo,
    clippy::unimplemented,
    clippy::indexing_slicing
)]
mod tests {
    use super::*;
    use crate::protocol::{
        LobbyState, ReconnectedPayload, RoomJoinedPayload, SpectatorJoinedPayload,
    };

    #[test]
    fn connected_event_is_constructible() {
        let event = SignalFishEvent::Connected;
        let debug = format!("{event:?}");
        assert!(debug.contains("Connected"));
    }

    #[test]
    fn debug_never_exposes_reconnection_tokens_or_application_payloads() {
        let event = SignalFishEvent::RoomJoined {
            room_id: uuid::Uuid::nil(),
            room_code: "ROOM".into(),
            player_id: uuid::Uuid::nil(),
            game_name: "game".into(),
            max_players: 2,
            supports_authority: false,
            current_players: vec![],
            is_authority: false,
            lobby_state: LobbyState::Waiting,
            ready_players: vec![],
            relay_type: "websocket".into(),
            current_spectators: vec![],
            ice_servers: vec![],
            reconnection_token: Some("top-secret-token".into()),
        };
        let debug = format!("{event:?}");
        assert_eq!(debug, "RoomJoined");
        assert!(!debug.contains("top-secret-token"));
    }

    #[test]
    fn violation_diagnostics_are_classified_by_semantic_precedence() {
        let cases = [
            (
                "PlayerLeft final_seq disagreed with terminal watermark",
                ProtocolViolationKind::Lifecycle,
            ),
            (
                "DeliveryReport was not immediately followed by Error",
                ProtocolViolationKind::Causality,
            ),
            (
                "reconnect snapshot watermark mismatch",
                ProtocolViolationKind::Snapshot,
            ),
            (
                "cumulative counter moved backward",
                ProtocolViolationKind::Counters,
            ),
            ("uncovered delivery gap", ProtocolViolationKind::DeliveryGap),
            (
                "v2 exposed class/key",
                ProtocolViolationKind::UnexpectedMetadata,
            ),
            ("sequence stamp was zero", ProtocolViolationKind::Stamp),
        ];
        for (diagnostic, expected) in cases {
            assert_eq!(ProtocolViolationKind::from_diagnostic(diagnostic), expected);
        }
    }

    #[test]
    fn disconnected_event_contains_reason() {
        let event = SignalFishEvent::Disconnected {
            reason: Some("server shutdown".into()),
            last_server_error: None,
        };
        if let SignalFishEvent::Disconnected { reason, .. } = event {
            assert_eq!(reason.as_deref(), Some("server shutdown"));
        } else {
            panic!("expected Disconnected variant");
        }
    }

    #[test]
    fn from_server_message_pong() {
        let event = SignalFishEvent::from(ServerMessage::Pong);
        assert!(matches!(event, SignalFishEvent::Pong));
    }

    #[test]
    fn from_server_message_room_left() {
        let event = SignalFishEvent::from(ServerMessage::RoomLeft);
        assert!(matches!(event, SignalFishEvent::RoomLeft));
    }

    #[test]
    fn from_server_message_room_joined_flattens_payload() {
        let payload = RoomJoinedPayload {
            room_id: uuid::Uuid::nil(),
            room_code: "ABC123".into(),
            player_id: uuid::Uuid::nil(),
            game_name: "test-game".into(),
            max_players: 4,
            supports_authority: true,
            current_players: vec![],
            is_authority: false,
            lobby_state: LobbyState::Waiting,
            ready_players: vec![],
            relay_type: "auto".into(),
            current_spectators: vec![],
            ice_servers: vec![],
            reconnection_token: None,
        };
        let msg = ServerMessage::RoomJoined(Box::new(payload));
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::RoomJoined {
            room_code,
            max_players,
            game_name,
            ..
        } = event
        {
            assert_eq!(room_code, "ABC123");
            assert_eq!(max_players, 4);
            assert_eq!(game_name, "test-game");
        } else {
            panic!("expected RoomJoined variant");
        }
    }

    #[test]
    fn from_server_message_error() {
        let msg = ServerMessage::Error {
            message: "oops".into(),
            error_code: Some(ErrorCode::InternalError),
        };
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::Error {
            message,
            error_code,
        } = event
        {
            assert_eq!(message, "oops");
            assert_eq!(error_code, Some(ErrorCode::InternalError));
        } else {
            panic!("expected Error variant");
        }
    }

    #[test]
    fn event_is_clone() {
        let event = SignalFishEvent::Pong;
        let cloned = event.clone();
        assert!(matches!(cloned, SignalFishEvent::Pong));
    }

    #[test]
    fn from_server_message_reconnected_flattens_payload() {
        let payload = ReconnectedPayload {
            room_id: uuid::Uuid::nil(),
            room_code: "RECON1".into(),
            player_id: uuid::Uuid::nil(),
            game_name: "recon-game".into(),
            max_players: 6,
            supports_authority: false,
            current_players: vec![],
            is_authority: true,
            lobby_state: LobbyState::Waiting,
            ready_players: vec![],
            relay_type: "tcp".into(),
            current_spectators: vec![],
            ice_servers: vec![],
            missed_events: vec![ServerMessage::Pong],
            replay: None,
            sender_watermarks: vec![],
            reconnection_token: None,
        };
        let msg = ServerMessage::Reconnected(Box::new(payload));
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::Reconnected {
            room_code,
            max_players,
            is_authority,
            missed_events,
            ..
        } = event
        {
            assert_eq!(room_code, "RECON1");
            assert_eq!(max_players, 6);
            assert!(is_authority);
            assert_eq!(missed_events.len(), 1);
            assert!(matches!(missed_events[0], SignalFishEvent::Pong));
        } else {
            panic!("expected Reconnected variant");
        }
    }

    #[test]
    fn from_server_message_spectator_joined_flattens_payload() {
        let payload = SpectatorJoinedPayload {
            room_id: uuid::Uuid::nil(),
            room_code: "SPEC1".into(),
            spectator_id: uuid::Uuid::nil(),
            game_name: "spec-game".into(),
            current_players: vec![],
            current_spectators: vec![],
            lobby_state: LobbyState::Waiting,
            reason: None,
        };
        let msg = ServerMessage::SpectatorJoined(Box::new(payload));
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::SpectatorJoined {
            room_code,
            game_name,
            reason,
            ..
        } = event
        {
            assert_eq!(room_code, "SPEC1");
            assert_eq!(game_name, "spec-game");
            assert!(reason.is_none());
        } else {
            panic!("expected SpectatorJoined variant");
        }
    }

    #[test]
    fn from_server_message_game_data_binary() {
        let msg = ServerMessage::GameDataBinary {
            from_player: uuid::Uuid::nil(),
            encoding: GameDataEncoding::MessagePack,
            payload: vec![0xDE, 0xAD],
            seq: None,
            epoch: None,
        };
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::GameDataBinary {
            from_player,
            encoding,
            payload,
            seq,
            epoch,
        } = event
        {
            assert_eq!(from_player, uuid::Uuid::nil());
            assert!(matches!(encoding, GameDataEncoding::MessagePack));
            assert_eq!(payload, vec![0xDE, 0xAD]);
            assert!(seq.is_none());
            assert!(epoch.is_none());
        } else {
            panic!("expected GameDataBinary variant");
        }
    }

    #[test]
    fn from_server_message_authentication_error() {
        let msg = ServerMessage::AuthenticationError {
            error: "bad token".into(),
            error_code: ErrorCode::InvalidAppId,
        };
        let event = SignalFishEvent::from(msg);
        if let SignalFishEvent::AuthenticationError { error, error_code } = event {
            assert_eq!(error, "bad token");
            assert_eq!(error_code, ErrorCode::InvalidAppId);
        } else {
            panic!("expected AuthenticationError variant");
        }
    }

    /// `decode_failed` bounds its work: the wire `type` tag is recovered only
    /// from well-formed JSON within the prefix cap, never by re-parsing a full
    /// oversized or malformed frame.
    #[cfg(any(feature = "tokio-runtime", feature = "polling-client"))]
    #[test]
    fn decode_failed_recovers_type_only_for_bounded_well_formed_frames() {
        // 1. Small, well-formed frame, unknown type (a serde `Data` error) →
        //    type recovered; the whole frame fits in the prefix.
        let small = r#"{"type":"SomeFutureMessage","data":{}}"#;
        let err = serde_json::from_str::<ServerMessage>(small).unwrap_err();
        assert_eq!(err.classify(), serde_json::error::Category::Data);
        match SignalFishEvent::decode_failed(small, &err) {
            SignalFishEvent::DecodeFailed {
                message_type,
                raw_prefix,
                ..
            } => {
                assert_eq!(message_type.as_deref(), Some("SomeFutureMessage"));
                assert_eq!(raw_prefix, small);
            }
            other => panic!("expected DecodeFailed, got {other:?}"),
        }

        // 2. Malformed JSON (a `Syntax` error) → the re-parse is skipped
        //    entirely, so no `type` is recovered.
        let garbage = "not valid json {{{";
        let err = serde_json::from_str::<ServerMessage>(garbage).unwrap_err();
        assert_ne!(err.classify(), serde_json::error::Category::Data);
        match SignalFishEvent::decode_failed(garbage, &err) {
            SignalFishEvent::DecodeFailed { message_type, .. } => {
                assert_eq!(message_type, None);
            }
            other => panic!("expected DecodeFailed, got {other:?}"),
        }

        // 3. Well-formed unknown-type frame larger than the prefix cap → work
        //    stays bounded: `message_type` is None (the full frame is never
        //    re-parsed), while the type is still visible in the capped prefix.
        let big = format!(
            r#"{{"type":"SomeFutureMessage","data":{{"pad":"{}"}}}}"#,
            "x".repeat(DECODE_FAILED_RAW_PREFIX_MAX)
        );
        let err = serde_json::from_str::<ServerMessage>(&big).unwrap_err();
        assert_eq!(err.classify(), serde_json::error::Category::Data);
        match SignalFishEvent::decode_failed(&big, &err) {
            SignalFishEvent::DecodeFailed {
                message_type,
                raw_prefix,
                ..
            } => {
                assert_eq!(
                    message_type, None,
                    "an oversized frame must not be re-parsed to recover its type"
                );
                assert!(raw_prefix.len() <= DECODE_FAILED_RAW_PREFIX_MAX);
                assert!(
                    raw_prefix.contains("SomeFutureMessage"),
                    "the type remains visible in the bounded prefix"
                );
            }
            other => panic!("expected DecodeFailed, got {other:?}"),
        }
    }
}