arcature 2026.2.1

//! Typed errors for the realtime surface (AGENTS.md §18).
//!
//! Each variant corresponds to a failure that can actually happen on the
//! realtime path. No raw `String` errors, no "future-proof" variants, no
//! giant framework-wide enum. The variants are deliberately coarse where the
//! upstream `Display` may carry attacker-controlled bytes (a malformed
//! frame, an oversize message): the error records the *category* and a
//! bounded, safe hint, never the raw payload (observe spec §53/§59).
//!
//! The realtime path is attacker-facing (PROGRAM.md §AP2.1-8): a hostile
//! client may send oversize frames, malformed text, or a cross-origin
//! upgrade request. None of these may panic (AGENTS.md §17); they map to a
//! typed variant and a clean connection close.
//!
//! The errors are hand-rolled `Display`/`Debug` impls (matching the crate's
//! existing `EngineError` convention) so the `realtime` feature pulls no
//! error-derive dependency — only `tokio`.

use crate::axum::http::StatusCode;

/// The realtime failure surface.
///
/// Grouped by where on the path the failure originates:
/// - [`RealtimeError::Origin`] / [`RealtimeError::Unauthorized`] /
///   [`RealtimeError::ConnectionLimit`] — admission (before the upgrade
///   completes or the connection is tracked).
/// - [`RealtimeError::Protocol`] — a malformed or oversize frame after
///   upgrade.
/// - [`RealtimeError::Channel`] — broadcast/subscribe transport failures
///   (lagged slow consumer, closed channel, full buffer).
/// - [`RealtimeError::Shutdown`] — drain did not complete within the bound.
#[derive(Debug)]
pub enum RealtimeError {
    /// The request's `Origin` (or lack thereof) is not authorized by the
    /// configured [`crate::realtime::origin::OriginPolicy`].
    Origin,
    /// The connection or channel subscription was denied by the
    /// application's explicit authorization hook. Channel names never
    /// implicitly authorize (PROGRAM.md §AP2.1-8); this is the explicit
    /// denial outcome.
    Unauthorized,
    /// The connection registry is at its configured maximum. New
    /// connections are rejected so an attacker cannot open unbounded
    /// connections (AGENTS.md §29).
    ConnectionLimit,
    /// A malformed or oversize frame after upgrade. The `hint` is a
    /// fixed, low-cardinality category (e.g. `"oversize"`,
    /// `"malformed"`, `"utf8"`) — never the raw frame bytes.
    Protocol {
        /// A safe, fixed category for the protocol failure.
        hint: ProtocolHint,
    },
    /// A broadcast/subscribe transport failure. See [`ChannelError`].
    Channel(ChannelError),
    /// Graceful drain did not complete within the bound; `remaining` is the
    /// number of connections still open when the timeout elapsed.
    Shutdown { remaining: usize },
}

impl std::fmt::Display for RealtimeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Origin => f.write_str("realtime origin not authorized"),
            Self::Unauthorized => f.write_str("realtime authorization denied"),
            Self::ConnectionLimit => f.write_str("realtime connection limit reached"),
            Self::Protocol { hint } => {
                f.write_str("realtime protocol error: ")?;
                std::fmt::Display::fmt(hint, f)
            }
            Self::Channel(source) => {
                f.write_str("realtime channel error: ")?;
                std::fmt::Display::fmt(source, f)
            }
            Self::Shutdown { remaining } => {
                f.write_str("realtime drain timed out: ")?;
                std::fmt::Display::fmt(remaining, f)?;
                f.write_str(" connections remaining")
            }
        }
    }
}

impl std::error::Error for RealtimeError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Channel(source) => Some(source),
            Self::Origin
            | Self::Unauthorized
            | Self::ConnectionLimit
            | Self::Protocol { .. }
            | Self::Shutdown { .. } => None,
        }
    }
}

impl From<ChannelError> for RealtimeError {
    fn from(source: ChannelError) -> Self {
        Self::Channel(source)
    }
}

/// A fixed, low-cardinality category for a protocol-level failure.
///
/// The string is emitted via `Display` so it is safe to record in a tracing
/// span field (observe spec §53). It never carries attacker-controlled
/// bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProtocolHint {
    /// A frame or message exceeded the configured size limit.
    Oversize,
    /// A frame could not be decoded as a valid WebSocket message.
    Malformed,
    /// A text frame was not valid UTF-8.
    Utf8,
    /// The underlying WebSocket stream returned an error.
    Stream,
}

impl ProtocolHint {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Oversize => "oversize",
            Self::Malformed => "malformed",
            Self::Utf8 => "utf8",
            Self::Stream => "stream",
        }
    }
}

impl std::fmt::Display for ProtocolHint {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// Broadcast/subscribe transport failures (the typed wrapper over
/// `tokio::sync::broadcast`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChannelError {
    /// A subscriber fell behind the bounded buffer and missed messages.
    /// The application decides whether to resync or close the slow
    /// consumer; the wrapper does not silently drop application messages.
    Lagged,
    /// The channel has no remaining senders (the owning half was dropped).
    Closed,
    /// The channel buffer is full and the message was not delivered. With
    /// the bounded wrapper this is surfaced to the publisher so it can
    /// apply backpressure rather than blocking indefinitely.
    Full,
}

impl std::fmt::Display for ChannelError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Lagged => f.write_str("channel subscriber lagged"),
            Self::Closed => f.write_str("channel closed"),
            Self::Full => f.write_str("channel full"),
        }
    }
}

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

/// Map a realtime admission error to an HTTP status for the upgrade
/// response. Used by the WebSocket and SSE entry points before the long
/// connection is established.
///
/// This is a small, explicit mapping — not a generic `IntoResponse`, which
/// would invite payload leakage. Only admission errors map here; protocol
/// errors after upgrade close the connection with a WebSocket close code,
/// not an HTTP status.
#[must_use]
pub fn admission_status(error: &RealtimeError) -> StatusCode {
    match error {
        RealtimeError::Origin | RealtimeError::Unauthorized => StatusCode::FORBIDDEN,
        RealtimeError::ConnectionLimit => StatusCode::SERVICE_UNAVAILABLE,
        // Protocol/Channel/Shutdown surface after the connection is
        // established. Defensively map to 500 so a misuse never produces a
        // misleading 2xx/3xx.
        RealtimeError::Protocol { .. }
        | RealtimeError::Channel(_)
        | RealtimeError::Shutdown { .. } => StatusCode::INTERNAL_SERVER_ERROR,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::error::Error as _;

    #[test]
    fn protocol_hint_display_is_stable_and_safe() {
        assert_eq!(ProtocolHint::Oversize.to_string(), "oversize");
        assert_eq!(ProtocolHint::Malformed.to_string(), "malformed");
        assert_eq!(ProtocolHint::Utf8.to_string(), "utf8");
        assert_eq!(ProtocolHint::Stream.to_string(), "stream");
    }

    #[test]
    fn admission_status_maps_origin_and_authz_to_403() {
        assert_eq!(
            admission_status(&RealtimeError::Origin),
            StatusCode::FORBIDDEN
        );
        assert_eq!(
            admission_status(&RealtimeError::Unauthorized),
            StatusCode::FORBIDDEN
        );
    }

    #[test]
    fn admission_status_maps_connection_limit_to_503() {
        assert_eq!(
            admission_status(&RealtimeError::ConnectionLimit),
            StatusCode::SERVICE_UNAVAILABLE
        );
    }

    #[test]
    fn admission_status_maps_post_upgrade_errors_to_500() {
        assert_eq!(
            admission_status(&RealtimeError::Protocol {
                hint: ProtocolHint::Oversize
            }),
            StatusCode::INTERNAL_SERVER_ERROR
        );
        assert_eq!(
            admission_status(&RealtimeError::Channel(ChannelError::Lagged)),
            StatusCode::INTERNAL_SERVER_ERROR
        );
        assert_eq!(
            admission_status(&RealtimeError::Shutdown { remaining: 1 }),
            StatusCode::INTERNAL_SERVER_ERROR
        );
    }

    #[test]
    fn display_does_not_carry_attacker_bytes() {
        // The hint is a fixed enum string, not frame content.
        let err = RealtimeError::Protocol {
            hint: ProtocolHint::Malformed,
        };
        assert_eq!(err.to_string(), "realtime protocol error: malformed");
    }

    #[test]
    fn channel_error_source_is_itself() {
        let err = RealtimeError::from(ChannelError::Closed);
        assert!(err.source().is_some());
        assert_eq!(err.source().unwrap().to_string(), "channel closed");
    }
}