noema-actix-webapi 0.1.0

use std::cell::RefCell;
use std::future::Future;
use std::sync::{Arc, OnceLock, RwLock};

use actix_web::http::StatusCode;
use actix_web::{HttpRequest, HttpResponse, web};
use actix_ws::{Message, MessageStream, Session as WsIo};
use dashmap::{DashMap, DashSet};
use noema::core::{Container, Injectable, Resolver};
use noema::events::{Event, EventDispatch};
use noema::resolve;
use serde::{Deserialize, Serialize};

use crate::error::{MappedError, error_ws_envelope};
use tokio::sync::mpsc;
use tracing::Instrument;
use uuid::Uuid;

pub type SessionId = Uuid;
pub type RoomId = String;

/// Reject a WebSocket handshake before the socket is upgraded.
#[derive(Debug, Clone)]
pub struct WsError {
    status: StatusCode,
    message: String,
}

impl WsError {
    pub fn new(status: StatusCode, message: impl Into<String>) -> Self {
        Self {
            status,
            message: message.into(),
        }
    }

    pub fn unauthorized(message: impl Into<String>) -> Self {
        Self::new(StatusCode::UNAUTHORIZED, message)
    }

    pub fn forbidden(message: impl Into<String>) -> Self {
        Self::new(StatusCode::FORBIDDEN, message)
    }

    pub fn bad_request(message: impl Into<String>) -> Self {
        Self::new(StatusCode::BAD_REQUEST, message)
    }

    pub fn status(&self) -> StatusCode {
        self.status
    }

    pub fn message(&self) -> &str {
        &self.message
    }

    pub fn into_response(self) -> HttpResponse {
        HttpResponse::build(self.status).body(self.message)
    }
}

impl std::fmt::Display for WsError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.status, self.message)
    }
}

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

/// Transport that can be mounted with [`connect`]. Auth belongs in [`on_connect`](WsConnection::on_connect).
///
/// `?Send`: Actix handlers are worker-local (`HttpRequest` is `Rc`). The type itself stays `Send + Sync`.
#[async_trait::async_trait(?Send)]
pub trait WsConnection: EventDispatch + Send + Sync + 'static {
    /// App-defined identity for this socket (player session, tenant, `()`).
    type Ctx: Clone + Send + Sync + 'static;

    /// Called with the handshake request **before** the upgrade. `Err` → HTTP error, no socket.
    async fn on_connect(&self, req: &HttpRequest) -> Result<Self::Ctx, WsError>;

    /// Called after the socket loop ends, while the hub still has this session.
    async fn on_disconnect(&self, _ctx: &Self::Ctx) {}

    /// Inbound JSON failed, unknown `name`, or `dispatch` returned `Err` (`InvokeMode::Await`).
    ///
    /// Default: log and send [`error_ws_envelope`] to this socket
    /// (`{ "name": "error", "data": ErrorBody }`). Override to silence, remap, or close.
    async fn on_dispatch_error(
        &self,
        err: &(dyn std::error::Error + Send + Sync + 'static),
        event_name: &str,
    ) {
        tracing::error!(event_name, error = %err, "ws dispatch error");
        let Some(id) = SessionHub::current_id() else {
            return;
        };
        let _ = resolve::<SessionHub>().send_raw(id, error_ws_envelope(err));
    }
}

/// The socket that received the message currently being dispatched (`InvokeMode::Await`).
#[derive(Clone)]
pub struct Session<T: WsConnection> {
    session_id: SessionId,
    ctx: T::Ctx,
}

impl<T: WsConnection> Session<T> {
    pub fn new(session_id: SessionId, ctx: T::Ctx) -> Self {
        Self { session_id, ctx }
    }

    /// Bound only inside `EventListener::handle` for events from `connect::<T>()`.
    /// `T` must be the same type as that `connect::<T>()` — one session per task, no type tag.
    pub fn get() -> Option<Self> {
        CURRENT
            .try_with(|slot| {
                // SAFETY: `with_session::<T>` stored `T::Ctx`. This task binds one transport.
                let ctx = unsafe { &*(slot.ptr as *const T::Ctx) };
                Self {
                    session_id: slot.session_id,
                    ctx: ctx.clone(),
                }
            })
            .ok()
    }

    pub fn id(&self) -> SessionId {
        self.session_id
    }

    pub fn ctx(&self) -> &T::Ctx {
        &self.ctx
    }
}

const SESSION_NOT_BOUND: &str =
    "SessionContext not bound; dispatch via connect::<T>() or inject a test double";

/// Injectable WS session port. Production: `resolve::<dyn SessionContext<T> + Send + Sync>()`
/// (reads the task-local bound during `connect::<T>()` dispatch). Tests: pass your own impl.
pub trait SessionContext<T: WsConnection>: Send + Sync {
    fn id(&self) -> SessionId;
    fn ctx(&self) -> T::Ctx;
}

impl<T: WsConnection> SessionContext<T> for Session<T> {
    fn id(&self) -> SessionId {
        self.session_id
    }

    fn ctx(&self) -> T::Ctx {
        self.ctx.clone()
    }
}

struct AmbientSessionContext<T>(std::marker::PhantomData<fn() -> T>);

impl<T: WsConnection> SessionContext<T> for AmbientSessionContext<T> {
    fn id(&self) -> SessionId {
        Session::<T>::get().expect(SESSION_NOT_BOUND).id()
    }

    fn ctx(&self) -> T::Ctx {
        Session::<T>::get().expect(SESSION_NOT_BOUND).ctx
    }
}

impl<T: WsConnection> Resolver<dyn SessionContext<T> + Send + Sync> for Container {
    fn resolve() -> Arc<dyn SessionContext<T> + Send + Sync> {
        Arc::new(AmbientSessionContext(std::marker::PhantomData))
    }
}

#[derive(Clone, Copy)]
struct Slot {
    session_id: SessionId,
    ptr: usize,
}

tokio::task_local! {
    static CURRENT: Slot;
}

/// Bind `T::Ctx` for `fut` (same slot `connect::<T>()` uses). Downstream tests; not a type tag.
#[doc(hidden)]
pub async fn with_session<T, F, R>(session_id: SessionId, ctx: &T::Ctx, fut: F) -> R
where
    T: WsConnection,
    F: Future<Output = R>,
{
    let slot = Slot {
        session_id,
        ptr: ctx as *const T::Ctx as usize,
    };
    CURRENT.scope(slot, fut).await
}

#[derive(Deserialize)]
struct Incoming {
    name: String,
    #[serde(default)]
    data: serde_json::Value,
}

#[derive(Serialize)]
struct Outgoing<T: Serialize> {
    name: &'static str,
    data: T,
}

/// Envelope a bus subscriber deserializes. Local last-mile uses [`SessionHub::broadcast_raw`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WsFanoutMessage {
    pub origin: String,
    pub room: String,
    pub body: String,
}

impl WsFanoutMessage {
    pub fn is_local(&self) -> bool {
        self.origin == process_origin()
    }
}

pub type WsPublishFn = Arc<dyn Fn(String, String) + Send + Sync>;

static ORIGIN: OnceLock<String> = OnceLock::new();
static PUBLISH: RwLock<Option<WsPublishFn>> = RwLock::new(None);

thread_local! {
    static TEST_PUBLISH: RefCell<Option<WsPublishFn>> = const { RefCell::new(None) };
}

pub(crate) fn install_origin() {
    let _ = ORIGIN.set(Uuid::now_v7().to_string());
}

pub fn process_origin() -> &'static str {
    ORIGIN.get_or_init(|| Uuid::now_v7().to_string()).as_str()
}

/// Opt-in bus publish. `broadcast_event` sends locally then calls this with `(room, envelope_json)`.
/// Subscribers must use [`SessionHub::broadcast_raw`] (no second publish).
pub fn on_ws_publish(publish: impl Fn(String, String) + Send + Sync + 'static) {
    *PUBLISH.write().expect("ws publish lock") = Some(Arc::new(publish));
}

fn active_publish() -> Option<WsPublishFn> {
    let local = TEST_PUBLISH.with(|c| c.borrow().clone());
    if local.is_some() {
        return local;
    }
    PUBLISH.read().ok().and_then(|g| g.clone())
}

fn fanout(room: &str, body: &str) {
    let Some(publish) = active_publish() else {
        return;
    };
    let msg = WsFanoutMessage {
        origin: process_origin().to_string(),
        room: room.to_string(),
        body: body.to_string(),
    };
    let Ok(envelope) = serde_json::to_string(&msg) else {
        return;
    };
    publish(room.to_string(), envelope);
}

pub struct SessionHub {
    sessions: DashMap<SessionId, mpsc::UnboundedSender<String>>,
    rooms: DashMap<RoomId, DashSet<SessionId>>,
    session_rooms: DashMap<SessionId, DashSet<RoomId>>,
}

impl SessionHub {
    pub fn new() -> Self {
        Self {
            sessions: DashMap::new(),
            rooms: DashMap::new(),
            session_rooms: DashMap::new(),
        }
    }

    pub fn register(&self, id: SessionId, tx: mpsc::UnboundedSender<String>) {
        self.sessions.insert(id, tx);
    }

    pub fn unregister(&self, id: SessionId) {
        self.drop_session(id);
    }

    fn drop_session(&self, id: SessionId) {
        self.sessions.remove(&id);
        if let Some((_, rooms)) = self.session_rooms.remove(&id) {
            for room in rooms.iter() {
                if let Some(members) = self.rooms.get(room.as_str()) {
                    members.remove(&id);
                }
            }
        }
    }

    pub fn join_room(&self, id: SessionId, room: impl Into<RoomId>) {
        let room = room.into();
        self.rooms.entry(room.clone()).or_default().insert(id);
        self.session_rooms.entry(id).or_default().insert(room);
    }

    pub fn leave_room(&self, id: SessionId, room: &str) {
        if let Some(members) = self.rooms.get(room) {
            members.remove(&id);
        }
        if let Some(rooms) = self.session_rooms.get(&id) {
            rooms.remove(room);
        }
    }

    /// Session that dispatched the current event (`connect` loop, Await mode).
    pub fn current_id() -> Option<SessionId> {
        CURRENT.try_with(|slot| slot.session_id).ok()
    }

    /// Send `event` to the socket that dispatched the current handler.
    pub fn reply<E: Event + Serialize>(&self, event: &E) -> bool {
        match Self::current_id() {
            Some(id) => self.send_event(id, event),
            None => false,
        }
    }

    pub fn join_current(&self, room: impl Into<RoomId>) -> bool {
        match Self::current_id() {
            Some(id) => {
                self.join_room(id, room);
                true
            }
            None => false,
        }
    }

    pub fn leave_current(&self, room: &str) -> bool {
        match Self::current_id() {
            Some(id) => {
                self.leave_room(id, room);
                true
            }
            None => false,
        }
    }

    pub fn send_raw(&self, id: SessionId, json: String) -> bool {
        self.sessions
            .get(&id)
            .map(|tx| tx.send(json).is_ok())
            .unwrap_or(false)
    }

    pub fn send_event<E: Event + Serialize>(&self, id: SessionId, event: &E) -> bool {
        let Ok(json) = serde_json::to_string(&Outgoing {
            name: E::WIRE_NAME,
            data: event,
        }) else {
            return false;
        };
        self.send_raw(id, json)
    }

    /// Local sockets in `room` only. Bus subscribers use this so the origin is not published again.
    pub fn broadcast_raw(&self, room: &str, json: impl Into<String>) {
        let json = json.into();
        let Some(members) = self.rooms.get(room) else {
            return;
        };
        for id in members.iter() {
            self.send_raw(*id, json.clone());
        }
    }

    pub fn broadcast_event<E: Event + Serialize>(&self, room: &str, event: &E) {
        let Ok(json) = serde_json::to_string(&Outgoing {
            name: E::WIRE_NAME,
            data: event,
        }) else {
            return;
        };
        self.broadcast_raw(room, json.clone());
        fanout(room, &json);
    }
}

impl Default for SessionHub {
    fn default() -> Self {
        Self::new()
    }
}

impl Injectable<Container> for SessionHub {
    fn inject(_: &Container) -> Self {
        Self::new()
    }
}

/// Actix handler: `on_connect`, then upgrade and dispatch loop for `T`.
///
/// Mount as many as you need from `Presentation::configure_scope`:
/// `scope.route("/ws", web::get().to(connect::<ChatWs>))`.
pub async fn connect<T>(
    req: HttpRequest,
    stream: web::Payload,
) -> Result<HttpResponse, actix_web::Error>
where
    T: WsConnection,
    Container: noema::core::Resolver<T>,
{
    let transport = resolve::<T>();
    let ctx = match transport.on_connect(&req).await {
        Ok(ctx) => ctx,
        Err(err) => return Ok(crate::cors::apply_cors(&req, err.into_response())),
    };
    let (res, session, msg_stream) = actix_ws::handle(&req, stream)?;
    let hub = resolve::<SessionHub>();
    actix_web::rt::spawn(run_loop::<T>(transport, hub, session, msg_stream, ctx));
    Ok(crate::cors::apply_cors(&req, res))
}

async fn run_loop<T: WsConnection>(
    transport: Arc<T>,
    hub: Arc<SessionHub>,
    mut session: WsIo,
    mut msg_stream: MessageStream,
    ctx: T::Ctx,
) {
    let ctx = Arc::new(ctx);
    let session_id = Uuid::now_v7();
    let (tx, mut rx) = mpsc::unbounded_channel::<String>();
    hub.register(session_id, tx);
    let mut heartbeat = actix_web::rt::time::interval(std::time::Duration::from_secs(30));

    loop {
        tokio::select! {
            _ = heartbeat.tick() => {
                if session.ping(b"").await.is_err() {
                    break;
                }
            }
            out = rx.recv() => {
                let Some(out) = out else { break };
                if session.text(out).await.is_err() {
                    break;
                }
            }
            incoming = msg_stream.recv() => {
                match incoming {
                    Some(Ok(Message::Ping(bytes))) => {
                        if session.pong(&bytes).await.is_err() {
                            break;
                        }
                    }
                    Some(Ok(Message::Text(text))) => {
                        let incoming = match serde_json::from_str::<Incoming>(&text) {
                            Ok(msg) => msg,
                            Err(err) => {
                                tracing::debug!(error = %err, "ws text is not a name/data envelope");
                                let transport = Arc::clone(&transport);
                                let mapped = MappedError::bad_request("invalid event envelope");
                                with_session::<T, _, _>(session_id, ctx.as_ref(), async move {
                                    transport.on_dispatch_error(&mapped, "inbound").await;
                                })
                                .await;
                                continue;
                            }
                        };
                        tracing::debug!(name = %incoming.name, %session_id, "ws dispatch");
                        let payload = match serde_json::to_vec(&incoming.data) {
                            Ok(bytes) => bytes,
                            Err(_) => {
                                let transport = Arc::clone(&transport);
                                let mapped = MappedError::bad_request("invalid event payload");
                                with_session::<T, _, _>(session_id, ctx.as_ref(), async move {
                                    transport.on_dispatch_error(&mapped, "inbound").await;
                                })
                                .await;
                                continue;
                            }
                        };
                        let transport = Arc::clone(&transport);
                        with_session::<T, _, _>(session_id, ctx.as_ref(), async move {
                            if transport.entries().iter().all(|e| e.name != incoming.name) {
                                let mapped = MappedError::bad_request(format!(
                                    "unknown event: {}",
                                    incoming.name
                                ));
                                transport
                                    .on_dispatch_error(&mapped, &incoming.name)
                                    .await;
                                return;
                            }
                            if let Err(err) =
                                transport.dispatch(&incoming.name, &payload).await
                            {
                                transport
                                    .on_dispatch_error(&*err, &incoming.name)
                                    .await;
                            }
                        }
                        .instrument(tracing::info_span!(
                            "ws.session",
                            session_id = %session_id
                        )))
                        .await;
                    }
                    Some(Ok(Message::Close(_))) | None => break,
                    Some(Ok(_)) => {}
                    Some(Err(_)) => break,
                }
            }
        }
    }

    transport.on_disconnect(ctx.as_ref()).await;
    hub.unregister(session_id);
}

#[cfg(test)]
mod tests {
    use super::*;
    use noema::core::{Container, Injectable};
    use noema::events::{
        DispatchContext, EventDispatcherContext, SubscriberEntry, SubscriberRegistry,
    };
    use serde::{Deserialize, Serialize};
    use std::sync::Arc;

    #[derive(Serialize)]
    struct Ping {
        n: u8,
    }

    impl Event for Ping {
        const WIRE_NAME: &'static str = "ping";
    }

    #[derive(Clone)]
    struct DummyCtx(u8);

    struct ChatWs;

    fn empty_entries() -> &'static [SubscriberEntry] {
        &[]
    }

    impl EventDispatcherContext for ChatWs {
        fn dispatch_context(&self) -> DispatchContext {
            crate::actix::dispatch_context()
        }
    }

    impl SubscriberRegistry for ChatWs {
        fn entries(&self) -> &'static [SubscriberEntry] {
            empty_entries()
        }
    }

    #[async_trait::async_trait(?Send)]
    impl WsConnection for ChatWs {
        type Ctx = DummyCtx;

        async fn on_connect(&self, _req: &HttpRequest) -> Result<Self::Ctx, WsError> {
            Ok(DummyCtx(7))
        }
    }

    #[tokio::test]
    async fn send_event_reaches_socket() {
        let hub = SessionHub::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        assert!(hub.send_event(id, &Ping { n: 1 }));
        let got = rx.recv().await.expect("msg");
        assert!(got.contains("ping"));
        assert!(got.contains("\"n\":1"));
    }

    #[tokio::test]
    async fn broadcast_room() {
        let hub = SessionHub::new();
        let (tx_a, mut rx_a) = mpsc::unbounded_channel();
        let (tx_b, mut rx_b) = mpsc::unbounded_channel();
        let a = Uuid::now_v7();
        let b = Uuid::now_v7();
        hub.register(a, tx_a);
        hub.register(b, tx_b);
        hub.join_room(a, "r1");
        hub.join_room(b, "r1");
        hub.broadcast_event("r1", &Ping { n: 2 });
        assert!(rx_a.recv().await.unwrap().contains("ping"));
        assert!(rx_b.recv().await.unwrap().contains("ping"));
    }

    fn with_publish<R>(
        f: impl Fn(String, String) + Send + Sync + 'static,
        body: impl FnOnce() -> R,
    ) -> R {
        TEST_PUBLISH.with(|c| {
            let prev = c.replace(Some(Arc::new(f) as WsPublishFn));
            let out = body();
            c.replace(prev);
            out
        })
    }

    #[tokio::test]
    async fn broadcast_raw_is_local_only() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::<(String, String)>::new()));
        let seen2 = Arc::clone(&seen);
        let hub = SessionHub::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        hub.join_room(id, "r1");
        with_publish(
            move |room, env| seen2.lock().unwrap().push((room, env)),
            || hub.broadcast_raw("r1", r#"{"name":"x","data":{}}"#),
        );
        assert!(rx.recv().await.unwrap().contains("\"name\":\"x\""));
        assert!(seen.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn broadcast_event_fans_out_envelope() {
        let seen = Arc::new(std::sync::Mutex::new(Vec::<(String, String)>::new()));
        let seen2 = Arc::clone(&seen);
        let hub = SessionHub::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        hub.join_room(id, "lobby");
        with_publish(
            move |room, env| seen2.lock().unwrap().push((room, env)),
            || hub.broadcast_event("lobby", &Ping { n: 9 }),
        );
        assert!(rx.recv().await.unwrap().contains("\"n\":9"));
        let got = seen.lock().unwrap();
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].0, "lobby");
        let msg: WsFanoutMessage = serde_json::from_str(&got[0].1).unwrap();
        assert_eq!(msg.room, "lobby");
        assert!(msg.is_local());
        assert!(msg.body.contains("ping"));
        assert!(msg.body.contains("\"n\":9"));
    }

    #[test]
    fn fanout_subscriber_skips_local_origin() {
        let msg = WsFanoutMessage {
            origin: process_origin().to_string(),
            room: "lobby".into(),
            body: "{}".into(),
        };
        assert!(msg.is_local());
        let other = WsFanoutMessage {
            origin: "other-pod".into(),
            room: "lobby".into(),
            body: "{}".into(),
        };
        assert!(!other.is_local());
    }

    #[tokio::test]
    async fn unregister_drops_room_membership() {
        let hub = SessionHub::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        hub.join_room(id, "r1");
        hub.unregister(id);
        hub.broadcast_event("r1", &Ping { n: 3 });
        assert!(rx.recv().await.is_none());
    }

    #[tokio::test]
    async fn reply_sends_to_bound_session() {
        let hub = SessionHub::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        let ctx = DummyCtx(1);
        with_session::<ChatWs, _, _>(id, &ctx, async {
            assert!(hub.reply(&Ping { n: 4 }));
            assert!(hub.join_current("r1"));
        })
        .await;
        let got = rx.recv().await.expect("reply");
        assert!(got.contains("\"n\":4"));
    }

    #[tokio::test]
    async fn reply_without_bound_session_is_false() {
        let hub = SessionHub::new();
        assert!(!hub.reply(&Ping { n: 1 }));
        assert!(!hub.join_current("r1"));
        assert!(SessionHub::current_id().is_none());
    }

    #[tokio::test]
    async fn session_get_returns_connect_ctx() {
        let ctx = DummyCtx(9);
        let id = Uuid::now_v7();
        with_session::<ChatWs, _, _>(id, &ctx, async {
            let session = Session::<ChatWs>::get().expect("bound");
            assert_eq!(session.id(), id);
            assert_eq!(session.ctx().0, 9);
        })
        .await;
        assert!(Session::<ChatWs>::get().is_none());
    }

    struct UsesSession {
        session: Arc<dyn SessionContext<ChatWs> + Send + Sync>,
    }

    impl UsesSession {
        fn player(&self) -> u8 {
            self.session.ctx().0
        }
    }

    #[test]
    fn handler_accepts_injected_session_context() {
        let id = Uuid::now_v7();
        let h = UsesSession {
            session: Arc::new(Session::<ChatWs>::new(id, DummyCtx(9))),
        };
        assert_eq!(h.player(), 9);
        assert_eq!(h.session.id(), id);
    }

    #[tokio::test]
    async fn resolve_delegates_to_bound_session() {
        let ctx = DummyCtx(4);
        let id = Uuid::now_v7();
        with_session::<ChatWs, _, _>(id, &ctx, async {
            let session = noema::resolve::<dyn SessionContext<ChatWs> + Send + Sync>();
            assert_eq!(session.id(), id);
            assert_eq!(session.ctx().0, 4);
        })
        .await;
    }

    #[tokio::test]
    async fn on_connect_can_reject() {
        struct Denied;
        impl EventDispatcherContext for Denied {
            fn dispatch_context(&self) -> DispatchContext {
                crate::actix::dispatch_context()
            }
        }
        impl SubscriberRegistry for Denied {
            fn entries(&self) -> &'static [SubscriberEntry] {
                empty_entries()
            }
        }
        #[async_trait::async_trait(?Send)]
        impl WsConnection for Denied {
            type Ctx = ();
            async fn on_connect(&self, _req: &HttpRequest) -> Result<Self::Ctx, WsError> {
                Err(WsError::unauthorized("missing token"))
            }
        }

        let req = actix_web::test::TestRequest::default().to_http_request();
        let err = Denied.on_connect(&req).await.expect_err("denied");
        assert_eq!(err.status(), StatusCode::UNAUTHORIZED);
        let res = err.into_response();
        assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn default_on_dispatch_error_sends_envelope() {
        let hub = noema::resolve::<SessionHub>();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        let ctx = DummyCtx(1);
        let err = MappedError::bad_request("room is required");
        with_session::<ChatWs, _, _>(id, &ctx, async {
            ChatWs.on_dispatch_error(&err, "chat.join").await;
        })
        .await;
        let got = rx.recv().await.expect("envelope");
        assert!(got.contains("\"name\":\"error\""), "{got}");
        assert!(got.contains("bad_request"), "{got}");
        assert!(got.contains("room is required"), "{got}");
        hub.unregister(id);
    }

    #[tokio::test]
    async fn on_dispatch_error_override_is_silent() {
        struct Quiet;
        impl EventDispatcherContext for Quiet {
            fn dispatch_context(&self) -> DispatchContext {
                crate::actix::dispatch_context()
            }
        }
        impl SubscriberRegistry for Quiet {
            fn entries(&self) -> &'static [SubscriberEntry] {
                empty_entries()
            }
        }
        #[async_trait::async_trait(?Send)]
        impl WsConnection for Quiet {
            type Ctx = DummyCtx;
            async fn on_connect(&self, _req: &HttpRequest) -> Result<Self::Ctx, WsError> {
                Ok(DummyCtx(0))
            }
            async fn on_dispatch_error(
                &self,
                _err: &(dyn std::error::Error + Send + Sync + 'static),
                _event_name: &str,
            ) {
            }
        }

        let hub = noema::resolve::<SessionHub>();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        let ctx = DummyCtx(0);
        let err = MappedError::bad_request("nope");
        with_session::<Quiet, _, _>(id, &ctx, async {
            Quiet.on_dispatch_error(&err, "chat.join").await;
        })
        .await;
        assert!(rx.try_recv().is_err());
        hub.unregister(id);
    }

    #[derive(Serialize, Deserialize, Clone)]
    #[noema::event(name = "test.boom")]
    struct Boom {
        n: u8,
    }

    struct BoomHandler;

    impl Injectable<Container> for BoomHandler {
        fn inject(_: &Container) -> Self {
            Self
        }
    }

    #[async_trait::async_trait]
    impl noema::events::EventListener<Boom> for BoomHandler {
        async fn handle(
            &self,
            _: Arc<Boom>,
        ) -> noema::events::NoemaResult<()> {
            Err(MappedError::bad_request("room is required").into())
        }
    }

    struct BoomWs;

    impl EventDispatcherContext for BoomWs {
        fn dispatch_context(&self) -> DispatchContext {
            crate::actix::dispatch_context()
        }
    }

    noema::subscribe!(BoomWs, Boom: [BoomHandler]);

    #[async_trait::async_trait(?Send)]
    impl WsConnection for BoomWs {
        type Ctx = DummyCtx;
        async fn on_connect(&self, _req: &HttpRequest) -> Result<Self::Ctx, WsError> {
            Ok(DummyCtx(0))
        }
    }

    #[tokio::test]
    async fn await_handler_error_keeps_mapped_error_on_the_socket() {
        let hub = noema::resolve::<SessionHub>();
        let (tx, mut rx) = mpsc::unbounded_channel();
        let id = Uuid::now_v7();
        hub.register(id, tx);
        let ctx = DummyCtx(0);
        with_session::<BoomWs, _, _>(id, &ctx, async {
            let err = BoomWs
                .dispatch("test.boom", br#"{"n":1}"#)
                .await
                .expect_err("handler err");
            BoomWs.on_dispatch_error(&*err, "test.boom").await;
        })
        .await;
        let got = rx.recv().await.expect("envelope");
        assert!(got.contains("\"name\":\"error\""), "{got}");
        assert!(got.contains("bad_request"), "{got}");
        assert!(got.contains("room is required"), "{got}");
        hub.unregister(id);
    }
}