pulses 0.2.0

//! Compile-time list of registered handlers.
//!
//! Handlers are accumulated into a left-nested tuple (`((), H1, H2, ...)` shape)
//! by [`App::register`](crate::App::register). The [`HandlerSet`] trait walks
//! that list at compile time to (a) build the stream→handler routing table and
//! (b) spawn one [`HandlerPool`] per handler. The trait is sealed: only the
//! tuple shapes this crate defines can implement it.

use std::collections::HashMap;
use std::sync::Arc;

use tokio::sync::mpsc;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

use crate::config::AppConfig;
use crate::core::Broker;
use crate::core::Context;
use crate::core::Handler;
use crate::handler_pool::HandlerPool;
use crate::handler_pool::HandlerTask;
use crate::runtime::backoff::Backoff;

/// Bitmask identifying which handlers subscribe to a stream (one bit per
/// handler, by registration order).
pub(crate) type RouteMask = u128;

/// Maximum number of handlers a single [`App`](crate::App) can register, bounded
/// by the width of [`RouteMask`].
pub(crate) const MAX_HANDLERS: usize = RouteMask::BITS as usize;

/// Stream name → set of subscribing handlers.
pub(crate) type RoutingTable = HashMap<Arc<str>, RouteMask>;

mod sealed {
    pub trait Sealed {}
    impl Sealed for () {}
    impl<Tail, H> Sealed for (Tail, H) {}
}

/// Bundle of everything needed to spawn the handler pools, threaded through the
/// compile-time handler list.
///
/// Public only because it appears in the (sealed) [`HandlerSet`] trait's
/// internal methods; it cannot be constructed outside this crate.
#[doc(hidden)]
pub struct PoolWiring<'a, B: Broker> {
    pub(crate) context: Context<B>,
    pub(crate) config: AppConfig,
    pub(crate) backoff: Arc<Backoff>,
    pub(crate) ack_retry_tx: mpsc::Sender<B::AckToken>,
    pub(crate) cancellation: &'a CancellationToken,
    pub(crate) handles: &'a mut JoinSet<()>,
    pub(crate) senders: &'a mut Vec<mpsc::Sender<HandlerTask<B::AckToken>>>,
}

/// A compile-time list of [`Handler`]s. Sealed; implemented only for the tuple
/// shapes produced by [`App::register`](crate::App::register).
pub trait HandlerSet<B: Broker>: sealed::Sealed {
    /// Number of handlers in the list.
    const LEN: usize;
    /// Compile-time assertion that the list fits within [`MAX_HANDLERS`].
    const ASSERT_WITHIN_LIMIT: ();

    #[doc(hidden)]
    fn append_routes(routes: &mut RoutingTable);

    #[doc(hidden)]
    fn spawn_pools(self, wiring: &mut PoolWiring<'_, B>);
}

/// Build the routing table for a handler list.
pub(crate) fn build_routing_table<B, HS>() -> RoutingTable
where
    B: Broker,
    HS: HandlerSet<B>,
{
    let () = HS::ASSERT_WITHIN_LIMIT;
    let mut routes = RoutingTable::new();
    HS::append_routes(&mut routes);
    routes
}

impl<B: Broker> HandlerSet<B> for () {
    const LEN: usize = 0;
    const ASSERT_WITHIN_LIMIT: () = ();

    fn append_routes(_routes: &mut RoutingTable) {}

    fn spawn_pools(self, _wiring: &mut PoolWiring<'_, B>) {}
}

impl<B, Tail, H> HandlerSet<B> for (Tail, H)
where
    B: Broker,
    Tail: HandlerSet<B>,
    H: Handler<B>,
{
    const LEN: usize = Tail::LEN + 1;
    const ASSERT_WITHIN_LIMIT: () = {
        let () = Tail::ASSERT_WITHIN_LIMIT;
        assert!(Self::LEN <= MAX_HANDLERS, "handler count exceeds RouteMask capacity");
    };

    fn append_routes(routes: &mut RoutingTable) {
        let () = Self::ASSERT_WITHIN_LIMIT;
        Tail::append_routes(routes);

        let handler_bit = 1_u128 << Tail::LEN;
        for stream in H::STREAMS {
            routes.entry(Arc::<str>::from(*stream)).and_modify(|mask| *mask |= handler_bit).or_insert(handler_bit);
        }
    }

    fn spawn_pools(self, wiring: &mut PoolWiring<'_, B>) {
        let (tail, handler) = self;
        // Recurse tail-first so senders are indexed by registration order
        // (handler bit `i` == senders[i]).
        tail.spawn_pools(wiring);

        let (sender, pool) = HandlerPool::new(
            handler,
            wiring.context.clone(),
            &wiring.config,
            Arc::clone(&wiring.backoff),
            wiring.ack_retry_tx.clone(),
        );
        let cancellation = wiring.cancellation.clone();
        wiring.handles.spawn(pool.run(cancellation));
        wiring.senders.push(sender);
    }
}

#[cfg(test)]
mod tests {
    use super::build_routing_table;
    use crate::core::Context;
    use crate::core::Envelope;
    use crate::core::Handler;
    use crate::core::HandlerError;
    use crate::core::Outcome;
    use crate::test_support::MockBroker;

    struct OrdersHandler;
    struct SharedHandler;

    impl Handler<MockBroker> for OrdersHandler {
        const STREAMS: &'static [&'static str] = &["orders", "shared"];
        async fn handle(&self, _ctx: &Context<MockBroker>, _msg: &Envelope) -> Result<Outcome, HandlerError> {
            Ok(Outcome::Ack)
        }
    }

    impl Handler<MockBroker> for SharedHandler {
        const STREAMS: &'static [&'static str] = &["shared", "payments"];
        async fn handle(&self, _ctx: &Context<MockBroker>, _msg: &Envelope) -> Result<Outcome, HandlerError> {
            Ok(Outcome::Ack)
        }
    }

    #[test]
    fn routing_table_builds_expected_masks() {
        type Handlers = (((), OrdersHandler), SharedHandler);
        let routing = build_routing_table::<MockBroker, Handlers>();

        assert_eq!(routing.get("orders"), Some(&(1_u128 << 0)));
        assert_eq!(routing.get("shared"), Some(&((1_u128 << 0) | (1_u128 << 1))));
        assert_eq!(routing.get("payments"), Some(&(1_u128 << 1)));
        assert_eq!(routing.len(), 3);
    }
}