//! Channels — join/leave/event protocol over the WebSocket transport,
//! PubSub underneath (v0.6, design ratified 2026-06-12).
//!
//! Feature `"channels"`: this module — zero new dependencies. The wire
//! format is abstracted behind the [`Encode`]/[`Decode`] codec traits
//! (one method each); the `"phoenix"` feature supplies a phoenix.js-
//! compatible V2 JSON codec via serde, but nothing here knows about
//! JSON.
//!
//! # Topology
//!
//! - The app builds one [`ChannelHub`] (a [`TopicRouter`] + a
//!   `PubSub<Broadcast<P>>`). **In-runtime only** — `PubSub::new`
//!   spawns the table actor — and the hub must be NON-static (the
//!   `Arc<OnceLock<..>>`-in-the-route-closure pattern from v0.5;
//!   a `static` hub pins the pubsub table and hangs
//!   `serve_with_shutdown`).
//! - [`ChannelHub::upgrade`] turns an HTTP `Conn` into a channel
//!   socket: a [`WsHandler`] running in the connection actor that
//!   decodes frames and routes them by topic.
//! - Each successful join spawns ONE channel actor per `(socket,
//!   topic)`, linked to the connection actor (a channel panic tears
//!   the socket down; the client reconnects — divergence from Phoenix,
//!   where the socket survives, documented and accepted). The channel
//!   actor owns the [`Channel`] impl, its [`ChannelSocket`], and its
//!   pubsub subscription (pinned to the channel actor's pid, so the
//!   monitor scopes cleanup exactly to the channel's life).
//! - The channel actor parks on `select(&[&inbound, &bus_rx])` —
//!   inbound (decoded client frames forwarded by the conn actor) at
//!   index 0, broadcasts at 1. When the socket dies the conn-side
//!   handler drops, the inbound sender drops, the closed arm wakes the
//!   actor, `terminate` runs, the actor exits. Every link in that
//!   chain is in-runtime; no `PubSub`-clone keepalive cycle (the v0.5
//!   relay corollary does not bite here because the channel actor's
//!   exit is conn-driven, never table-driven).
//!
//! # The join ack is first-class
//!
//! The v0.5 discovery (the 101 races `on_open`) is answered
//! structurally: the machinery subscribes the channel actor to its
//! topic **before** sending the ok reply, and `subscribe` is a call.
//! A client that has seen its join reply is guaranteed to be in the
//! broadcast table.
//!
//! # As-landed deviations from the ratified sketch (veto by diff)
//!
//! - `Channel::join` takes `&mut self`: a no-self method is uncallable
//!   on a trait object, so construction moved to [`ChannelFactory`]
//!   (which receives the full raw topic, preserving the
//!   wildcard-extraction guarantee) and `join` became the
//!   first-callback-after-construction. Signature is otherwise the
//!   ratified `(topic, payload, socket) -> Result<P, P>`.
//! - `ChannelSocket::reply(status, payload)` — the ratified text both
//!   listed `ref` as a parameter and said the loop threads it
//!   invisibly; invisible won (the socket carries the in-flight ref,
//!   set by the loop around each `handle_in`).
//! - `ChannelSocket::broadcast` gained an `event` parameter — a push
//!   without an event name cannot be encoded; sketch-level omission.
//! - `TopicRouter::route` returns `Option<Arc<dyn ChannelFactory>>`
//!   (not `Box`): routers hand out shared factories without cloning
//!   the factory itself per join.

use crate::pubsub::PubSub;
use crate::ws::{Message, WsHandler, WsSender};

use smarm::{Pid, Receiver, Sender};

use std::cell::RefCell;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::sync::Arc;

#[cfg(feature = "phoenix")]
pub mod phoenix;

mod session;
pub use session::ChannelSession;

#[cfg(test)]
pub(crate) mod testkit;

// ---------------------------------------------------------------------------
// Wire-neutral frames
// ---------------------------------------------------------------------------

/// Reply status — `ok` / `error` on the wire (spelling is the codec's
/// business).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Status {
    Ok,
    Error,
}

/// A decoded inbound frame. `join_ref` identifies the join generation a
/// frame belongs to; `reference` correlates a reply to the request that
/// asked for it. Both are opaque client-chosen strings.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChannelFrame<P> {
    pub join_ref: Option<String>,
    pub reference: Option<String>,
    pub topic: String,
    pub message: ChannelMessage<P>,
}

/// The message half of a decoded frame. Clients only ever send events;
/// `Reply` exists so a codec is symmetric (and is ignored inbound).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChannelMessage<P> {
    Event { event: String, payload: P },
    Reply { status: Status, payload: Option<P> },
}

/// A borrowed outbound frame — what [`Encode`] consumes. Borrowed so a
/// broadcast fan-out encodes straight from the shared `Arc<Broadcast<P>>`
/// without cloning the payload (the zero-copy claim from the roadmap).
/// `payload: None` on an event encodes as the codec's empty payload
/// (`{}` in the phoenix codec) — server-side control events (`phx_close`)
/// have no payload to conjure for an arbitrary `P`.
#[derive(Debug, Clone, Copy)]
pub struct FrameRef<'a, P> {
    pub join_ref: Option<&'a str>,
    pub reference: Option<&'a str>,
    pub topic: &'a str,
    pub message: MessageRef<'a, P>,
}

/// Borrowed message half of [`FrameRef`].
#[derive(Debug, Clone, Copy)]
pub enum MessageRef<'a, P> {
    Event { event: &'a str, payload: Option<&'a P> },
    Reply { status: Status, payload: Option<&'a P> },
}

/// Decoding failed; the socket is closed `1002` (a peer speaking the
/// wrong dialect is a protocol error, not a recoverable hiccup).
#[derive(Debug)]
pub struct DecodeError(pub String);

impl std::fmt::Display for DecodeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "channel frame decode error: {}", self.0)
    }
}
impl std::error::Error for DecodeError {}

/// Encode one outbound frame into a WebSocket message. One method — the
/// whole wire format, envelope included, lives behind it.
pub trait Encode: Sized {
    fn encode(frame: FrameRef<'_, Self>) -> Message;
}

/// Decode one inbound WebSocket message into a frame. One method.
pub trait Decode: Sized {
    fn decode(msg: &Message) -> Result<ChannelFrame<Self>, DecodeError>;
}

// Protocol event names (phoenix.js vocabulary — these are channel
// PROTOCOL, not JSON; a custom codec still speaks them, just spelled
// however it likes on the wire).
pub const EV_JOIN: &str = "phx_join";
pub const EV_LEAVE: &str = "phx_leave";
pub const EV_REPLY: &str = "phx_reply";
pub const EV_CLOSE: &str = "phx_close";
pub const EV_HEARTBEAT: &str = "heartbeat";
/// The transport-control topic heartbeats arrive on.
pub const TOPIC_PHOENIX: &str = "phoenix";

// ---------------------------------------------------------------------------
// Channel trait + factory + router
// ---------------------------------------------------------------------------

/// One joined topic on one socket. Runs in its own actor; callbacks are
/// sequential. A panic in any callback unwinds the channel actor and —
/// via the link — tears down the whole socket.
pub trait Channel<P: Send + Sync + 'static>: Send + 'static {
    /// The client asked to join `topic`. `Ok(reply)` accepts (the reply
    /// payload goes back with status ok, and the channel is live);
    /// `Err(reply)` rejects (status error, the channel actor exits,
    /// `terminate` is NOT called — the channel never existed).
    ///
    /// Don't call `socket.reply` here; the return value IS the reply.
    /// `push`/`broadcast` are legal but note the subscription to this
    /// topic only becomes live on `Ok` — a broadcast to your own topic
    /// from inside `join` will not reach this socket.
    fn join(&mut self, topic: &str, payload: P, socket: &ChannelSocket<P>) -> Result<P, P>;

    /// A client event on the joined topic. Reply (correlated to this
    /// event's ref) via [`ChannelSocket::reply`]; replying is optional.
    fn handle_in(&mut self, event: &str, payload: P, socket: &ChannelSocket<P>);

    /// The channel is over: explicit leave, socket death, or server
    /// shutdown. Not called after a rejected join or a callback panic.
    fn terminate(&mut self) {}
}

/// Builds a fresh [`Channel`] per accepted-routing join. Receives the
/// full raw topic string, so wildcard-segment extraction is always
/// possible (`"room:lobby"` arrives whole; pull `"lobby"` out yourself).
pub trait ChannelFactory<P: Send + Sync + 'static>: Send + Sync {
    fn create(&self, topic: &str) -> Box<dyn Channel<P>>;

    /// How an accepted-routing join becomes a running channel actor.
    /// Internal seam — the default (ephemeral actor, linked to the
    /// connection, cold start per join) is the contract; only the
    /// session machinery overrides it. Not part of the public API.
    #[doc(hidden)]
    fn deploy(&self, hs: JoinHandshake<P>) -> ChannelInbox<P>
    where
        P: Encode + Decode,
    {
        let ch = self.create(&hs.topic);
        let conn_pid = smarm::self_pid();
        let (tx, rx) = smarm::channel::<In<P>>();
        smarm::spawn(move || {
            run_channel(conn_pid, ch, hs.topic, hs.join_ref, hs.reference, hs.payload, hs.ws, hs.bus, rx)
        });
        ChannelInbox(tx)
    }
}

/// Everything the conn-side handler knows about one `phx_join`, handed
/// to [`ChannelFactory::deploy`]. Opaque: fields are crate-internal.
#[doc(hidden)]
pub struct JoinHandshake<P: Send + Sync + 'static> {
    topic: String,
    join_ref: Option<String>,
    reference: Option<String>,
    payload: P,
    ws: WsSender,
    bus: PubSub<Broadcast<P>>,
}

/// The conn-side handler's sender into a deployed channel actor.
/// Opaque: the field is crate-internal.
#[doc(hidden)]
pub struct ChannelInbox<P: Send + Sync + 'static>(Sender<In<P>>);

impl<P: Send + Sync + 'static, F> ChannelFactory<P> for F
where
    F: Fn(&str) -> Box<dyn Channel<P>> + Send + Sync,
{
    fn create(&self, topic: &str) -> Box<dyn Channel<P>> {
        self(topic)
    }
}

/// [`ChannelFactory`] via `Default` — for channels whose state is built
/// in `join`.
pub struct DefaultFactory<C>(PhantomData<fn() -> C>);

impl<P: Send + Sync + 'static, C: Channel<P> + Default> ChannelFactory<P> for DefaultFactory<C> {
    fn create(&self, _topic: &str) -> Box<dyn Channel<P>> {
        Box::new(C::default())
    }
}

/// Maps a join's topic to the factory that builds its channel. `None`
/// rejects the join (status error).
pub trait TopicRouter<P: Send + Sync + 'static>: Send + Sync + 'static {
    fn route(&self, topic: &str) -> Option<Arc<dyn ChannelFactory<P>>>;
}

/// The shipped router: exact topics and `head:*` prefix patterns.
///
/// Registration scans the pattern to `'*'` and discards the rest;
/// lookup is an exact-map probe, then one prefix-map probe on
/// `topic[..=first_colon]`. No regex, no glob machinery — patterns more
/// exotic than `"room:*"` mean implementing [`TopicRouter`] yourself.
pub struct PrefixRouter<P: Send + Sync + 'static> {
    exact: HashMap<String, Arc<dyn ChannelFactory<P>>>,
    prefix: HashMap<String, Arc<dyn ChannelFactory<P>>>,
}

impl<P: Send + Sync + 'static> Default for PrefixRouter<P> {
    fn default() -> Self {
        Self::new()
    }
}

impl<P: Send + Sync + 'static> PrefixRouter<P> {
    pub fn new() -> Self {
        Self { exact: HashMap::new(), prefix: HashMap::new() }
    }

    /// Register `factory` for `pattern` (`"room:*"` or an exact topic).
    pub fn channel(mut self, pattern: &str, factory: impl ChannelFactory<P> + 'static) -> Self {
        match pattern.find('*') {
            Some(i) => self.prefix.insert(pattern[..i].to_string(), Arc::new(factory)),
            None => self.exact.insert(pattern.to_string(), Arc::new(factory)),
        };
        self
    }

    /// [`channel`](Self::channel) with a [`Default`]-built impl.
    pub fn channel_default<C: Channel<P> + Default>(self, pattern: &str) -> Self
    where
        P: 'static,
    {
        self.channel(pattern, DefaultFactory::<C>(PhantomData))
    }

    /// [`channel`](Self::channel), with opt-in session persistence
    /// keyed and configured by `S`'s [`ChannelSession`] impl (usually
    /// `S` is the channel type itself). **In-runtime only**: this
    /// spawns the pattern's session-registry actor, same law as
    /// [`ChannelHub::new`].
    pub fn channel_session<S>(self, pattern: &str, factory: impl ChannelFactory<P> + 'static) -> Self
    where
        S: ChannelSession<P>,
        S::Key: Clone,
        P: Encode + Decode,
    {
        self.channel(pattern, session::SessionFactory::new::<S>(Arc::new(factory)))
    }

    /// [`channel_session`](Self::channel_session) with a
    /// [`Default`]-built impl that is its own session config.
    pub fn channel_session_default<C>(self, pattern: &str) -> Self
    where
        C: Channel<P> + ChannelSession<P> + Default,
        C::Key: Clone,
        P: Encode + Decode,
    {
        self.channel_session::<C>(pattern, DefaultFactory::<C>(PhantomData))
    }
}

impl<P: Send + Sync + 'static> TopicRouter<P> for PrefixRouter<P> {
    fn route(&self, topic: &str) -> Option<Arc<dyn ChannelFactory<P>>> {
        if let Some(f) = self.exact.get(topic) {
            return Some(f.clone());
        }
        let head = topic.find(':').map(|i| &topic[..=i])?;
        self.prefix.get(head).cloned()
    }
}

// ---------------------------------------------------------------------------
// Broadcast bus payload + ChannelSocket
// ---------------------------------------------------------------------------

/// What rides the hub's pubsub: a named event for everyone joined to a
/// topic. `Arc<Broadcast<P>>` end to end — one allocation per broadcast
/// regardless of subscriber count.
pub struct Broadcast<P> {
    pub event: String,
    pub payload: P,
}

/// The channel's handle to its transport: a [`WsSender`] for this
/// socket, the hub's pubsub for everyone else's, and the in-flight ref
/// the loop threads around each `handle_in` (invisible to the impl —
/// [`reply`](Self::reply) just works).
pub struct ChannelSocket<P: Send + Sync + 'static> {
    ws: WsSender,
    bus: PubSub<Broadcast<P>>,
    topic: String,
    join_ref: Option<String>,
    cur_ref: RefCell<Option<String>>,
}

impl<P: Encode + Decode + Send + Sync + 'static> ChannelSocket<P> {
    /// The full raw topic this channel is joined to.
    pub fn topic(&self) -> &str {
        &self.topic
    }

    /// Reply to the event currently being handled. Correlated to that
    /// event's ref; a refless event gets a refless reply (phoenix.js
    /// drops those — harmless).
    pub fn reply(&self, status: Status, payload: P) {
        self.send_reply(status, Some(&payload));
    }

    /// Push an event to THIS socket only.
    pub fn push(&self, event: &str, payload: P) {
        let _ = self.ws.send(P::encode(FrameRef {
            join_ref: self.join_ref.as_deref(),
            reference: None,
            topic: &self.topic,
            message: MessageRef::Event { event, payload: Some(&payload) },
        }));
    }

    /// Broadcast to every socket joined to this channel's topic,
    /// including this one.
    pub fn broadcast(&self, event: &str, payload: P) {
        let _ = self
            .bus
            .broadcast(self.topic.clone(), Broadcast { event: event.to_string(), payload });
    }

    /// [`broadcast`](Self::broadcast), excluding this socket (the
    /// no-echo shape — phoenix's `broadcast_from`).
    pub fn broadcast_from(&self, event: &str, payload: P) {
        let _ = self.bus.broadcast_from(
            smarm::self_pid(),
            self.topic.clone(),
            Broadcast { event: event.to_string(), payload },
        );
    }

    /// Broadcast to an arbitrary topic (joined sockets there receive
    /// it; this channel's own topic membership is irrelevant).
    pub fn broadcast_to(&self, topic: &str, event: &str, payload: P) {
        let _ = self
            .bus
            .broadcast(topic.to_string(), Broadcast { event: event.to_string(), payload });
    }

    fn send_reply(&self, status: Status, payload: Option<&P>) {
        let r = self.cur_ref.borrow();
        let _ = self.ws.send(P::encode(FrameRef {
            join_ref: self.join_ref.as_deref(),
            reference: r.as_deref(),
            topic: &self.topic,
            message: MessageRef::Reply { status, payload },
        }));
    }

    fn set_ref(&self, r: Option<String>) {
        *self.cur_ref.borrow_mut() = r;
    }
}

// ---------------------------------------------------------------------------
// ChannelHub — the app-facing entry point
// ---------------------------------------------------------------------------

/// One per app (or per channel namespace): the router plus the pubsub
/// bus every channel broadcasts on.
///
/// **In-runtime only** (spawns the pubsub table) and **must not live in
/// a `static`** — use the non-static `Arc<OnceLock<ChannelHub<P>>>`
/// captured by the route closure, exactly the v0.5 pubsub pattern, or
/// graceful shutdown will hang waiting for the table to exit.
pub struct ChannelHub<P: Send + Sync + 'static> {
    bus: PubSub<Broadcast<P>>,
    router: Arc<dyn TopicRouter<P>>,
}

impl<P: Send + Sync + 'static> Clone for ChannelHub<P> {
    fn clone(&self) -> Self {
        Self { bus: self.bus.clone(), router: self.router.clone() }
    }
}

impl<P: Encode + Decode + Send + Sync + 'static> ChannelHub<P> {
    pub fn new(router: impl TopicRouter<P>) -> Self {
        Self { bus: PubSub::new(), router: Arc::new(router) }
    }

    /// Accept the WebSocket upgrade on `conn` and speak channels over
    /// it. The HTTP handshake-validation rules of
    /// [`Conn::upgrade`](crate::Conn::upgrade) apply unchanged.
    pub fn upgrade(&self, conn: crate::Conn) -> crate::Conn {
        conn.upgrade(SocketHandler {
            bus: self.bus.clone(),
            router: self.router.clone(),
            joined: HashMap::new(),
        })
    }

    /// Server-side broadcast (an HTTP handler, a background actor):
    /// every socket joined to `topic` gets the event.
    pub fn broadcast(&self, topic: &str, event: &str, payload: P) {
        let _ = self
            .bus
            .broadcast(topic.to_string(), Broadcast { event: event.to_string(), payload });
    }
}

// ---------------------------------------------------------------------------
// The conn-side socket handler
// ---------------------------------------------------------------------------

enum In<P> {
    Event { reference: Option<String>, event: String, payload: P },
    Leave { reference: Option<String> },
}

struct Joined<P> {
    join_ref: Option<String>,
    tx: Sender<In<P>>,
}

struct SocketHandler<P: Send + Sync + 'static> {
    bus: PubSub<Broadcast<P>>,
    router: Arc<dyn TopicRouter<P>>,
    joined: HashMap<String, Joined<P>>,
}

impl<P: Encode + Decode + Send + Sync + 'static> SocketHandler<P> {
    fn reply_direct(
        sender: &WsSender,
        join_ref: Option<&str>,
        reference: Option<&str>,
        topic: &str,
        status: Status,
    ) {
        let _ = sender.send(P::encode(FrameRef {
            join_ref,
            reference,
            topic,
            message: MessageRef::Reply { status, payload: None },
        }));
    }

    fn close_event_direct(sender: &WsSender, join_ref: Option<&str>, topic: &str) {
        let _ = sender.send(P::encode(FrameRef {
            join_ref,
            reference: None,
            topic,
            message: MessageRef::Event { event: EV_CLOSE, payload: None },
        }));
    }
}

impl<P: Encode + Decode + Send + Sync + 'static> WsHandler for SocketHandler<P> {
    fn on_message(&mut self, msg: Message, sender: &WsSender) {
        let frame = match P::decode(&msg) {
            Ok(f) => f,
            Err(_) => {
                // Wrong dialect: protocol error, not a per-message skip.
                let _ = sender.close(1002, "channel frame decode error");
                return;
            }
        };
        let ChannelMessage::Event { event, payload } = frame.message else {
            return; // clients don't reply; ignore
        };

        // Transport heartbeat — answered here, no channel involved.
        if frame.topic == TOPIC_PHOENIX && event == EV_HEARTBEAT {
            Self::reply_direct(
                sender,
                None,
                frame.reference.as_deref(),
                TOPIC_PHOENIX,
                Status::Ok,
            );
            return;
        }

        match event.as_str() {
            EV_JOIN => {
                // A re-join replaces: the old actor's inbound sender
                // drops (it exits via the closed arm, terminate runs)
                // and the old join generation is told it's over.
                if let Some(old) = self.joined.remove(&frame.topic) {
                    drop(old.tx);
                    Self::close_event_direct(sender, old.join_ref.as_deref(), &frame.topic);
                }
                let Some(factory) = self.router.route(&frame.topic) else {
                    Self::reply_direct(
                        sender,
                        frame.join_ref.as_deref(),
                        frame.reference.as_deref(),
                        &frame.topic,
                        Status::Error,
                    );
                    return;
                };
                let inbox = factory.deploy(JoinHandshake {
                    topic: frame.topic.clone(),
                    join_ref: frame.join_ref.clone(),
                    reference: frame.reference,
                    payload,
                    ws: sender.clone(),
                    bus: self.bus.clone(),
                });
                self.joined
                    .insert(frame.topic, Joined { join_ref: frame.join_ref, tx: inbox.0 });
            }
            EV_LEAVE => match self.joined.remove(&frame.topic) {
                Some(j) => {
                    if j.tx.send(In::Leave { reference: frame.reference.clone() }).is_err() {
                        // Actor already gone (rejected join, earlier
                        // death): leaving a dead channel is a success.
                        Self::reply_direct(
                            sender,
                            frame.join_ref.as_deref(),
                            frame.reference.as_deref(),
                            &frame.topic,
                            Status::Ok,
                        );
                    }
                }
                None => Self::reply_direct(
                    sender,
                    frame.join_ref.as_deref(),
                    frame.reference.as_deref(),
                    &frame.topic,
                    Status::Error,
                ),
            },
            _ => {
                let pruned_or_unjoined = match self.joined.get(&frame.topic) {
                    Some(j) => j
                        .tx
                        .send(In::Event { reference: frame.reference.clone(), event, payload })
                        .is_err(),
                    None => true,
                };
                if pruned_or_unjoined {
                    // Lazy prune (the pubsub discipline): a dead
                    // channel's entry goes on first send failure.
                    self.joined.remove(&frame.topic);
                    Self::reply_direct(
                        sender,
                        frame.join_ref.as_deref(),
                        frame.reference.as_deref(),
                        &frame.topic,
                        Status::Error,
                    );
                }
            }
        }
    }

    // on_close: nothing explicit — this handler drops when the
    // connection actor finishes (or unwinds), every inbound sender
    // drops with it, and each channel actor wakes on its closed arm,
    // runs terminate, and exits. The whole chain is in-runtime.
}

// ---------------------------------------------------------------------------
// The channel actor
// ---------------------------------------------------------------------------

#[allow(clippy::too_many_arguments)]
fn run_channel<P: Encode + Decode + Send + Sync + 'static>(
    conn_pid: Pid,
    mut ch: Box<dyn Channel<P>>,
    topic: String,
    join_ref: Option<String>,
    reference: Option<String>,
    payload: P,
    ws: WsSender,
    bus: PubSub<Broadcast<P>>,
    inbound: Receiver<In<P>>,
) {
    // Linked, per the ratified design: a channel panic is a socket
    // event (conn unwinds, every sibling channel follows through the
    // sender-drop chain), and an abnormal conn death (drain-deadline
    // force-stop) reaps channels immediately instead of waiting on the
    // closed-arm wake.
    smarm::link(conn_pid);

    let socket = ChannelSocket {
        ws,
        bus: bus.clone(),
        topic: topic.clone(),
        join_ref,
        cur_ref: RefCell::new(None),
    };

    socket.set_ref(reference);
    match ch.join(&topic, payload, &socket) {
        Ok(reply) => {
            // Subscribe BEFORE the ok reply: subscribe is a call, so
            // once the client observes its join ack, its membership in
            // the broadcast table is a fact, not a race (the v0.5
            // on_open discovery, answered structurally).
            let bus_rx = match bus.subscribe(&topic) {
                Ok(rx) => rx,
                Err(_) => {
                    socket.send_reply(Status::Error, None);
                    return;
                }
            };
            socket.send_reply(Status::Ok, Some(&reply));
            socket.set_ref(None);
            channel_loop(&mut *ch, &socket, &inbound, &bus_rx);
        }
        Err(reply) => {
            // Rejected: error reply, actor exits, no terminate (the
            // channel never existed). The conn-side entry self-prunes
            // on its first failed send.
            socket.send_reply(Status::Error, Some(&reply));
        }
    }
}

fn channel_loop<P: Encode + Decode + Send + Sync + 'static>(
    ch: &mut dyn Channel<P>,
    socket: &ChannelSocket<P>,
    inbound: &Receiver<In<P>>,
    bus_rx: &Receiver<std::sync::Arc<Broadcast<P>>>,
) {
    // A closed arm stays ready forever (it would starve the other arm
    // under priority order) — drop the bus arm from the set once its
    // disconnect has been observed.
    let mut bus_alive = true;
    loop {
        let idx = if bus_alive {
            smarm::select(&[inbound, bus_rx])
        } else {
            smarm::select(&[inbound])
        };
        if idx == 0 {
            match inbound.try_recv() {
                Ok(Some(In::Event { reference, event, payload })) => {
                    socket.set_ref(reference);
                    ch.handle_in(&event, payload, socket);
                    socket.set_ref(None);
                }
                Ok(Some(In::Leave { reference })) => {
                    socket.set_ref(reference);
                    socket.send_reply(Status::Ok, None);
                    socket.set_ref(None);
                    // Generation over: tell the client, then go. The
                    // subscription dies with the actor (monitor).
                    let _ = socket.ws.send(P::encode(FrameRef {
                        join_ref: socket.join_ref.as_deref(),
                        reference: None,
                        topic: &socket.topic,
                        message: MessageRef::Event { event: EV_CLOSE, payload: None },
                    }));
                    ch.terminate();
                    return;
                }
                Ok(None) => {}
                Err(_) => {
                    // Socket gone (conn actor finished or replaced this
                    // join generation).
                    ch.terminate();
                    return;
                }
            }
        } else {
            match bus_rx.try_recv() {
                Ok(Some(b)) => {
                    let out = P::encode(FrameRef {
                        join_ref: socket.join_ref.as_deref(),
                        reference: None,
                        topic: &socket.topic,
                        message: MessageRef::Event { event: &b.event, payload: Some(&b.payload) },
                    });
                    if socket.ws.send(out).is_err() {
                        ch.terminate();
                        return;
                    }
                }
                Ok(None) => {}
                Err(_) => bus_alive = false, // table down (shutdown tail)
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Tests — runtime-backed, with a toy pipe-delimited codec (no JSON: the
// core must be exercisable without the phoenix feature).
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use super::testkit::{eventually, Harness, TP};
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Mutex;
    use std::time::Duration;


    /// A room channel: rejects topic "room:locked"; echoes events back as
    /// replies; broadcasts on "shout"; flags terminate.
    struct Room {
        terminated: Arc<AtomicBool>,
    }

    impl Channel<TP> for Room {
        fn join(&mut self, topic: &str, payload: TP, _s: &ChannelSocket<TP>) -> Result<TP, TP> {
            if topic == "room:locked" {
                Err(TP("no entry".into()))
            } else {
                Ok(TP(format!("welcome:{}", payload.0)))
            }
        }
        fn handle_in(&mut self, event: &str, payload: TP, s: &ChannelSocket<TP>) {
            match event {
                "shout" => s.broadcast("shouted", payload),
                "shout_from" => s.broadcast_from("shouted", payload),
                _ => s.reply(Status::Ok, TP(format!("echo:{}:{}", event, payload.0))),
            }
        }
        fn terminate(&mut self) {
            self.terminated.store(true, Ordering::SeqCst);
        }
    }

    struct RoomFactory {
        terminated: Arc<AtomicBool>,
    }
    impl ChannelFactory<TP> for RoomFactory {
        fn create(&self, _topic: &str) -> Box<dyn Channel<TP>> {
            Box::new(Room { terminated: self.terminated.clone() })
        }
    }


    fn hub(terminated: &Arc<AtomicBool>) -> ChannelHub<TP> {
        ChannelHub::new(
            PrefixRouter::new().channel("room:*", RoomFactory { terminated: terminated.clone() }),
        )
    }


    #[test]
    fn prefix_router_exact_and_prefix_and_miss() {
        let t = Arc::new(AtomicBool::new(false));
        let r = PrefixRouter::new()
            .channel("room:*", RoomFactory { terminated: t.clone() })
            .channel("lobby", RoomFactory { terminated: t.clone() });
        assert!(r.route("room:a").is_some());
        assert!(r.route("room:").is_some()); // empty wildcard segment still routes
        assert!(r.route("lobby").is_some());
        assert!(r.route("lobbyx").is_none());
        assert!(r.route("hall:a").is_none());
        assert!(r.route("room").is_none()); // no colon, no exact hit
    }

    #[test]
    fn join_ok_replies_then_events_echo() {
        let out = Arc::new(Mutex::new(Vec::<String>::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:a|event|phx_join|hi");
            out2.lock().unwrap().push(h.recv());
            h.send("j1|r2|room:a|event|ping|x");
            out2.lock().unwrap().push(h.recv());
        });
        let v = out.lock().unwrap().clone();
        assert_eq!(v[0], "j1|r1|room:a|reply|ok|welcome:hi");
        assert_eq!(v[1], "j1|r2|room:a|reply|ok|echo:ping:x");
    }

    #[test]
    fn join_rejected_replies_error_no_terminate() {
        let out = Arc::new(Mutex::new(String::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        let t2 = t.clone();
        smarm::run(move || {
            let hub = hub(&t2);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:locked|event|phx_join|hi");
            *out2.lock().unwrap() = h.recv();
            // The rejected channel never existed: terminate must not run.
            smarm::sleep(Duration::from_millis(50));
        });
        assert_eq!(*out.lock().unwrap(), "j1|r1|room:locked|reply|error|no entry");
        assert!(!t.load(Ordering::SeqCst));
    }

    #[test]
    fn unmatched_topic_join_replies_error() {
        let out = Arc::new(Mutex::new(String::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|hall:a|event|phx_join|hi");
            *out2.lock().unwrap() = h.recv();
        });
        assert_eq!(*out.lock().unwrap(), "j1|r1|hall:a|reply|error|");
    }

    #[test]
    fn heartbeat_replies_ok_without_join() {
        let out = Arc::new(Mutex::new(String::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut h = Harness::new(&hub);
            h.send("-|hb1|phoenix|event|heartbeat|");
            *out2.lock().unwrap() = h.recv();
        });
        assert_eq!(*out.lock().unwrap(), "-|hb1|phoenix|reply|ok|");
    }

    #[test]
    fn event_on_unjoined_topic_replies_error() {
        let out = Arc::new(Mutex::new(String::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:a|event|ping|x");
            *out2.lock().unwrap() = h.recv();
        });
        assert_eq!(*out.lock().unwrap(), "j1|r1|room:a|reply|error|");
    }

    #[test]
    fn broadcast_reaches_both_sockets_broadcast_from_skips_sender() {
        let got = Arc::new(Mutex::new(Vec::<(u8, String)>::new()));
        let (got2, t) = (got.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut a = Harness::new(&hub);
            let mut b = Harness::new(&hub);
            a.send("j1|r1|room:a|event|phx_join|A");
            b.send("j1|r1|room:a|event|phx_join|B");
            a.recv();
            b.recv();

            a.send("j1|r2|room:a|event|shout|hello");
            got2.lock().unwrap().push((0, a.recv())); // broadcast includes sender
            got2.lock().unwrap().push((1, b.recv()));

            a.send("j1|r3|room:a|event|shout_from|again");
            got2.lock().unwrap().push((1, b.recv())); // only b sees it
            // a's next outbound must be the echo of a fresh ping, not a
            // stray "again" — proves broadcast_from skipped a.
            a.send("j1|r4|room:a|event|ping|x");
            got2.lock().unwrap().push((0, a.recv()));
        });
        let v = got.lock().unwrap().clone();
        assert_eq!(v[0], (0, "j1|-|room:a|event|shouted|hello".into()));
        assert_eq!(v[1], (1, "j1|-|room:a|event|shouted|hello".into()));
        assert_eq!(v[2], (1, "j1|-|room:a|event|shouted|again".into()));
        assert_eq!(v[3], (0, "j1|r4|room:a|reply|ok|echo:ping:x".into()));
    }

    #[test]
    fn leave_replies_ok_sends_close_and_terminates() {
        let out = Arc::new(Mutex::new(Vec::<String>::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        let t2 = t.clone();
        smarm::run(move || {
            let hub = hub(&t2);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:a|event|phx_join|hi");
            out2.lock().unwrap().push(h.recv());
            h.send("j1|r9|room:a|event|phx_leave|");
            out2.lock().unwrap().push(h.recv());
            out2.lock().unwrap().push(h.recv());
            assert!(eventually(|| t2.load(Ordering::SeqCst)));
            // The conn-side entry is gone: a new event errors.
            h.send("j1|r10|room:a|event|ping|x");
            out2.lock().unwrap().push(h.recv());
        });
        let v = out.lock().unwrap().clone();
        assert_eq!(v[1], "j1|r9|room:a|reply|ok|");
        assert_eq!(v[2], "j1|-|room:a|event|phx_close|");
        assert_eq!(v[3], "j1|r10|room:a|reply|error|");
        assert!(t.load(Ordering::SeqCst));
    }

    #[test]
    fn socket_drop_terminates_channel_and_subscription() {
        let t = Arc::new(AtomicBool::new(false));
        let t2 = t.clone();
        smarm::run(move || {
            let hub = hub(&t2);
            let bus = hub.bus.clone();
            {
                let mut h = Harness::new(&hub);
                h.send("j1|r1|room:a|event|phx_join|hi");
                h.recv();
                assert!(eventually(|| bus.subscriber_count("room:a").unwrap() == 1));
            } // handler drops -> inbound sender drops -> channel actor exits
            assert!(eventually(|| t2.load(Ordering::SeqCst)));
            assert!(eventually(|| bus.subscriber_count("room:a").unwrap() == 0));
        });
        assert!(t.load(Ordering::SeqCst));
    }

    #[test]
    fn rejoin_replaces_old_generation() {
        let out = Arc::new(Mutex::new(Vec::<String>::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        let t2 = t.clone();
        smarm::run(move || {
            let hub = hub(&t2);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:a|event|phx_join|one");
            out2.lock().unwrap().push(h.recv());
            h.send("j2|r2|room:a|event|phx_join|two");
            // Old generation is closed (its actor terminates), then the
            // new join acks.
            out2.lock().unwrap().push(h.recv());
            out2.lock().unwrap().push(h.recv());
            assert!(eventually(|| t2.load(Ordering::SeqCst)));
            h.send("j2|r3|room:a|event|ping|x");
            out2.lock().unwrap().push(h.recv());
        });
        let v = out.lock().unwrap().clone();
        assert_eq!(v[1], "j1|-|room:a|event|phx_close|");
        assert_eq!(v[2], "j2|r2|room:a|reply|ok|welcome:two");
        assert_eq!(v[3], "j2|r3|room:a|reply|ok|echo:ping:x");
    }

    #[test]
    fn hub_broadcast_reaches_joined_socket() {
        let out = Arc::new(Mutex::new(String::new()));
        let (out2, t) = (out.clone(), Arc::new(AtomicBool::new(false)));
        smarm::run(move || {
            let hub = hub(&t);
            let mut h = Harness::new(&hub);
            h.send("j1|r1|room:a|event|phx_join|hi");
            h.recv();
            hub.broadcast("room:a", "news", TP("server says".into()));
            *out2.lock().unwrap() = h.recv();
        });
        assert_eq!(*out.lock().unwrap(), "j1|-|room:a|event|news|server says");
    }
}