maestro/crates/maestro-daemon/src/local_events.rs

//! Daemon-local fan-out of `RegistryEvent`s for the `WS /api/events`
//! endpoint (NAP-248).
//!
//! Two sources feed this bus:
//!
//! 1. **Own-session events** — the daemon's `SessionManager::events()` stream,
//!    translated from `SessionEvent` into the matching `RegistryEvent` variant
//!    by [`spawn_session_publisher`].
//! 2. **Peer-catalog events** — applied to the daemon's [`CatalogMirror`] when
//!    the registry fans them out (NAP-246). `CatalogMirror::apply` publishes
//!    each event here after mutating its local view.
//!
//! Subscribers get the merged stream. The `WS /api/events` handler builds a
//! one-shot `Snapshot` of own ∪ peer state on connect, then forwards live
//! deltas until the client disconnects. `Lagged` follows the registry's
//! existing convention — clients resync via REST.

use std::sync::Arc;

use maestro_protocol::RegistryEvent;
use tokio::sync::broadcast;

use crate::session::{SessionEvent, SessionManager};

/// Capacity of the per-bus broadcast channel. Sourced from the shared
/// [`maestro_config::BroadcastTuning`], which the registry's
/// `EVENT_CHANNEL_CAPACITY` also tracks, so a slow client behaves the same on
/// either side (NAP-273).
const BUS_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.events_capacity;

/// Cheap-to-clone handle around a `broadcast::Sender<RegistryEvent>`.
#[derive(Debug, Clone)]
pub struct LocalEventBus {
    sender: Arc<broadcast::Sender<RegistryEvent>>,
}

impl LocalEventBus {
    pub fn new() -> Self {
        let (sender, _) = broadcast::channel(BUS_CAPACITY);
        Self {
            sender: Arc::new(sender),
        }
    }

    /// Push an event onto the bus. No-op if there are no subscribers.
    pub fn publish(&self, event: RegistryEvent) {
        let _ = self.sender.send(event);
    }

    pub fn subscribe(&self) -> broadcast::Receiver<RegistryEvent> {
        self.sender.subscribe()
    }
}

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

/// Spawn a task that converts each `SessionEvent` from `manager` into the
/// matching `RegistryEvent` and publishes it on `bus`. Exits when the
/// `SessionManager` is dropped (the events channel closes).
pub fn spawn_session_publisher(manager: SessionManager, bus: LocalEventBus) {
    let daemon_id = manager.daemon_id().to_string();
    tokio::spawn(async move {
        let mut rx = manager.events();
        loop {
            match rx.recv().await {
                Ok(SessionEvent::Added(session)) => {
                    bus.publish(RegistryEvent::SessionAdded { session });
                }
                Ok(SessionEvent::Updated(session)) => {
                    bus.publish(RegistryEvent::SessionUpdated { session });
                }
                Ok(SessionEvent::Removed(id)) => {
                    bus.publish(RegistryEvent::SessionRemoved {
                        daemon_id: daemon_id.clone(),
                        id,
                    });
                }
                Err(broadcast::error::RecvError::Lagged(skipped)) => {
                    bus.publish(RegistryEvent::Lagged { skipped });
                }
                Err(broadcast::error::RecvError::Closed) => return,
            }
        }
    });
}

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

    #[tokio::test]
    async fn publish_visible_to_subscriber() {
        let bus = LocalEventBus::new();
        let mut rx = bus.subscribe();
        bus.publish(RegistryEvent::Lagged { skipped: 7 });
        match rx.recv().await {
            Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 7),
            other => panic!("unexpected: {other:?}"),
        }
    }

    #[tokio::test]
    async fn no_subscribers_swallows_event() {
        let bus = LocalEventBus::new();
        // No one's listening — `publish` must not panic and must not block.
        bus.publish(RegistryEvent::DaemonRemoved {
            daemon_id: "ghost".into(),
        });
    }

    #[tokio::test]
    async fn clones_share_state() {
        let a = LocalEventBus::new();
        let b = a.clone();
        let mut rx = a.subscribe();
        b.publish(RegistryEvent::Lagged { skipped: 3 });
        match rx.recv().await {
            Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 3),
            other => panic!("unexpected: {other:?}"),
        }
    }
}