arcature 2026.2.1

//! Typed in-process broadcast channel — the fan-out core shared by the
//! WebSocket and SSE wrappers (PROGRAM.md §AP2.1-8).
//!
//! A thin typed wrapper over `tokio::sync::broadcast` that adds:
//! - a bounded buffer with explicit, documented capacity (backpressure),
//! - typed [`ChannelError`] surfacing of lag/full/closed,
//! - live-subscriber tracking so the application can observe/cap
//!   per-channel fan-out,
//! - payload opacity: the application's serialized bytes cross the channel;
//!   the wrapper owns no wire format and logs no payload (observe spec §53).
//!
//! There is intentionally **no distributed pub/sub platform** here. The
//! server owns an explicit, app-owned in-process broadcast per channel;
//! distributed fanout (Redis/Valkey `PUBLISH`, NATS) is deferred to a later
//! wave (PROGRAM.md §AP2.1-8: "start concrete, traits only at natural
//! seams"). The natural seam is [`Broadcast::subscribe`] /
//! [`Broadcast::publish`]; a future `DistributedBroadcast` trait would
//! mirror them but is not implemented this wave.
//!
//! # Ownership (AGENTS.md §20)
//!
//! A [`Broadcast`] is owned by the application — typically built once and
//! stored in `AppState` (or resolved via a typed service). It is **not** a
//! process-global singleton, and the connection count lives inside the
//! `Broadcast` value the application owns. The realtime module keeps no
//! hidden global channel registry. The [`crate::realtime::registry`] tracks
//! *live connections* (for drain), not channels.
//!
//! # Closed-channel semantics
//!
//! A channel "closes" (subsequent `recv` returns [`ChannelError::Closed`])
//! when all **senders** are dropped. The `Subscription` deliberately does
//! **not** keep the sender alive: it holds only a separate counter `Arc`
//! (for the live-subscriber count), not the broadcast sender. So dropping
//! the last `Broadcast` clone closes the channel even if a subscription is
//! still live — matching the `tokio::sync::broadcast` contract the wrapper
//! surfaces.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use tokio::sync::broadcast;

use crate::realtime::error::ChannelError;

/// The serialized payload that crosses a realtime channel. The application
/// serializes (its `Serialize` impl) before publishing; the wrapper
/// transports opaque bytes so it owns no wire format and adds no proprietary
/// protocol (PROGRAM.md §AP2.1-8).
///
/// The bytes are the application's chosen encoding (typically JSON). The
/// wrapper does not inspect, log, or transform them.
#[derive(Debug, Clone)]
pub struct ChannelPayload(Arc<[u8]>);

impl ChannelPayload {
    /// Construct a payload from owned bytes. Cloned cheaply across
    /// subscribers via `Arc`.
    #[must_use]
    pub fn from_bytes(bytes: Vec<u8>) -> Self {
        Self(bytes.into())
    }

    /// Construct a payload from a static byte slice (no allocation).
    #[must_use]
    pub fn from_static(bytes: &'static [u8]) -> Self {
        Self(Arc::from(bytes))
    }

    /// Borrow the payload bytes.
    #[must_use]
    pub fn as_bytes(&self) -> &[u8] {
        &self.0
    }
}

struct Shared {
    tx: broadcast::Sender<ChannelPayload>,
    capacity: usize,
    // Live subscriber count, shared across all `Broadcast` clones and all
    // `Subscription` guards (so dropping a guard anywhere decrements the one
    // shared counter). The `Subscription` holds its own `Arc` to this
    // counter — NOT to `Shared` — so it does not keep the broadcast sender
    // alive (see "Closed-channel semantics" above).
    subscribers: Arc<AtomicUsize>,
}

/// A bounded in-process broadcast channel.
///
/// Construct with [`Broadcast::new`]; cloning is cheap (an `Arc` to the
/// shared sender and counter). Each subscriber gets its own
/// [`broadcast::Receiver`] over the shared bounded buffer — a slow consumer
/// that falls behind `capacity` messages receives
/// [`ChannelError::Lagged`], which the WS/SSE wrapper surfaces to the
/// application (no silent drop).
pub struct Broadcast {
    shared: Arc<Shared>,
}

impl Broadcast {
    /// Create a new broadcast channel with the given bounded capacity.
    ///
    /// `capacity` is the per-subscriber lag bound: a subscriber that falls
    /// more than `capacity` messages behind receives
    /// [`ChannelError::Lagged`]. A bounded buffer gives backpressure, not
    /// unbounded queueing (AGENTS.md §29).
    ///
    /// Returns `None` if `capacity` is zero — `tokio::sync::broadcast::channel`
    /// panics on zero capacity, and production source must not panic on
    /// hostile/misconfigured input (AGENTS.md §17). Callers should fall
    /// back to a sensible default capacity when this returns `None`, or
    /// surface a typed configuration error upstream.
    ///
    /// # Examples
    ///
    /// ```
    /// use arcature::realtime::channel::Broadcast;
    /// // A positive capacity constructs a channel.
    /// assert!(Broadcast::new(8).is_some());
    /// // A zero capacity does not panic; it returns `None`.
    /// assert!(Broadcast::new(0).is_none());
    /// ```
    #[must_use]
    pub fn new(capacity: usize) -> Option<Self> {
        if capacity == 0 {
            return None;
        }
        let (tx, _rx) = broadcast::channel(capacity);
        Some(Self {
            shared: Arc::new(Shared {
                tx,
                capacity,
                subscribers: Arc::new(AtomicUsize::new(0)),
            }),
        })
    }

    /// The configured bounded-buffer capacity.
    #[must_use]
    pub fn capacity(&self) -> usize {
        self.shared.capacity
    }

    /// The number of currently-tracked live subscribers on this channel.
    #[must_use]
    pub fn subscriber_count(&self) -> usize {
        self.shared.subscribers.load(Ordering::Relaxed)
    }

    /// Publish a payload to all current subscribers. Returns the number of
    /// subscribers the message was delivered to (`0` means no live
    /// subscribers; this is not an error for fire-and-forget publishes — the
    /// wrapper checks `receiver_count` first so a publish to no one is
    /// `Ok(0)`, matching the fire-and-forget contract).
    ///
    /// # Errors
    ///
    /// Returns [`ChannelError::Closed`] only if the channel has no remaining
    /// senders — which, because `Broadcast` owns its sender, can only
    /// happen after the last `Broadcast` clone was dropped. In practice the
    /// no-receivers case returns `Ok(0)` rather than an error, so this
    /// variant is rare from `publish`; it is preserved for callers building
    /// on the primitive directly.
    pub fn publish(&self, payload: ChannelPayload) -> Result<usize, ChannelError> {
        // tokio's `broadcast::Sender::send` returns `Err(SendError)` when
        // there are no receivers. For fire-and-forget publishes this is not
        // an error — the application does not care that no one received it.
        // Check the receiver count first so a publish to no one returns
        // `Ok(0)` rather than a misleading `Closed`.
        if self.shared.tx.receiver_count() == 0 {
            return Ok(0);
        }
        match self.shared.tx.send(payload) {
            Ok(n) => Ok(n),
            Err(broadcast::error::SendError(_)) => Err(ChannelError::Closed),
        }
    }

    /// Subscribe to the channel. The returned guard tracks its own
    /// lifetime: dropping it decrements the shared subscriber count, so the
    /// count reflects only live receivers (no orphaned counters).
    ///
    /// Broadcast semantics: a subscriber does **not** receive messages
    /// published before it subscribed.
    ///
    /// The guard holds a separate counter `Arc` (not the broadcast sender),
    /// so dropping the last `Broadcast` clone closes the channel even while
    /// a subscription is still live.
    #[must_use]
    pub fn subscribe(&self) -> Subscription {
        let rx = self.shared.tx.subscribe();
        self.shared.subscribers.fetch_add(1, Ordering::Relaxed);
        Subscription {
            rx,
            subscribers: Arc::clone(&self.shared.subscribers),
        }
    }
}

impl Clone for Broadcast {
    fn clone(&self) -> Self {
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}

impl std::fmt::Debug for Broadcast {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Broadcast")
            .field("capacity", &self.shared.capacity)
            .field("subscribers", &self.subscriber_count())
            .finish()
    }
}

/// A live subscription to a [`Broadcast`]. Dropping it decrements the
/// channel's subscriber count.
///
/// The receiver is exposed (rather than a custom `recv`) so the application
/// can compose it with `tokio::select!` against a heartbeat timer or a
/// shutdown signal — the WS/SSE wrappers do exactly that. A lagged receiver
/// is surfaced via [`Subscription::recv`] as [`ChannelError::Lagged`].
///
/// The guard holds a separate counter `Arc` (not the broadcast sender),
/// so dropping the last `Broadcast` clone closes the channel even while a
/// subscription is still live — matching the `tokio::sync::broadcast`
/// closed-channel contract the wrapper surfaces.
pub struct Subscription {
    /// The underlying broadcast receiver (composable with `select!`).
    pub rx: broadcast::Receiver<ChannelPayload>,
    subscribers: Arc<AtomicUsize>,
}

impl Subscription {
    /// Receive the next payload, mapping lagged/closed to typed errors.
    ///
    /// # Errors
    ///
    /// - [`ChannelError::Lagged`] — the subscriber fell behind; the
    ///   application decides whether to resync or close.
    /// - [`ChannelError::Closed`] — the channel has no remaining senders.
    pub async fn recv(&mut self) -> Result<ChannelPayload, ChannelError> {
        match self.rx.recv().await {
            Ok(payload) => Ok(payload),
            Err(broadcast::error::RecvError::Lagged(_)) => Err(ChannelError::Lagged),
            Err(broadcast::error::RecvError::Closed) => Err(ChannelError::Closed),
        }
    }
}

impl Drop for Subscription {
    fn drop(&mut self) {
        // Saturating subtract: never underflow even if a misbehaving caller
        // fabricated a guard. AtomicUsize wraps on overflow, so guard with a
        // compare-exchange floor of 0.
        loop {
            let current = self.subscribers.load(Ordering::Relaxed);
            if current == 0 {
                break;
            }
            // Relaxed is fine: this is a best-effort live count for
            // observation/caps, not a synchronization primitive.
            if self
                .subscribers
                .compare_exchange(current, current - 1, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn publish_delivers_to_live_subscriber() {
        let bc = Broadcast::new(8).expect("positive capacity");
        let mut sub = bc.subscribe();
        assert_eq!(bc.subscriber_count(), 1);
        let delivered = bc
            .publish(ChannelPayload::from_static(b"hello"))
            .expect("publish to live subscriber");
        assert_eq!(delivered, 1);
        let payload = sub.recv().await.expect("receive payload");
        assert_eq!(payload.as_bytes(), b"hello");
    }

    #[tokio::test]
    async fn lagged_subscriber_is_surfaced_not_silently_dropped() {
        // Capacity 1: a subscriber that does not drain falls behind after 2
        // publishes.
        let bc = Broadcast::new(1).expect("positive capacity");
        let mut slow = bc.subscribe();
        let _ = bc.publish(ChannelPayload::from_static(b"first"));
        let _ = bc.publish(ChannelPayload::from_static(b"second"));
        // The next recv reports the lag (no silent drop of application msgs).
        let outcome = slow.recv().await;
        assert!(matches!(outcome, Err(ChannelError::Lagged)), "{outcome:?}");
    }

    #[tokio::test]
    async fn closed_channel_surfaces_after_all_senders_drop() {
        let bc = Broadcast::new(4).expect("positive capacity");
        let mut sub = bc.subscribe();
        // The subscription holds only the counter Arc, not the sender, so
        // dropping the last Broadcast clone closes the channel.
        drop(bc);
        let outcome = sub.recv().await;
        assert!(matches!(outcome, Err(ChannelError::Closed)), "{outcome:?}");
    }

    #[tokio::test]
    async fn dropping_subscription_decrements_shared_count() {
        let bc = Broadcast::new(4).expect("positive capacity");
        let bc2 = bc.clone();
        {
            let _sub = bc.subscribe();
            assert_eq!(bc.subscriber_count(), 1);
            assert_eq!(bc2.subscriber_count(), 1, "count is shared across clones");
        }
        assert_eq!(bc.subscriber_count(), 0);
        assert_eq!(bc2.subscriber_count(), 0);
    }

    #[tokio::test]
    async fn publish_with_no_subscribers_reports_zero_not_error() {
        let bc = Broadcast::new(4).expect("positive capacity");
        let delivered = bc
            .publish(ChannelPayload::from_static(b"orphan"))
            .expect("send with no receivers is not an error");
        assert_eq!(delivered, 0);
    }

    #[test]
    fn zero_capacity_does_not_panic() {
        // tokio::sync::broadcast::channel panics on zero capacity; Broadcast::new
        // must return None instead (AGENTS.md §17: no panic on misconfigured
        // input).
        assert!(Broadcast::new(0).is_none());
        assert!(Broadcast::new(1).is_some());
    }

    #[test]
    fn payload_from_bytes_and_static_share() {
        let a = ChannelPayload::from_bytes(vec![1, 2, 3]);
        let b = ChannelPayload::from_static(&[1, 2, 3]);
        assert_eq!(a.as_bytes(), b.as_bytes());
        let a2 = a.clone();
        assert_eq!(a2.as_bytes(), a.as_bytes());
    }

    #[test]
    fn broadcast_is_send_sync_clone_for_appstate() {
        fn assert_send_sync_clone<T: Send + Sync + Clone>() {}
        assert_send_sync_clone::<Broadcast>();
    }
}