//! Bus de eventos in-process. Cualquier subsistema (watcher, init, supervisor futuro) puede
//! `publish(event)` para que **todas** las conexiones del bus de agente abiertas reciban una
//! copia. No hay tópicos: el filtrado lo hace el cliente.
//!
//! Implementación: lista de canales `mpsc::Sender<Event>`, uno por conexión. Cuando un
//! `send` falla (el peer cerró), el slot se marca caído y se purga en la siguiente publicación.
//!
//! Concurrencia: un `Mutex` sobre el vector basta. La cadencia esperada es baja (mutaciones
//! humanas y builds, no logs de tracing); cuando se vuelva crítico, sustituiremos por
//! `tokio::broadcast` o un sharded broadcaster.

use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{Arc, Mutex};

use hammer_core::proto::Event;

#[derive(Clone)]
pub struct EventBus {
    subs: Arc<Mutex<Vec<Sender<Event>>>>,
}

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

impl EventBus {
    pub fn new() -> Self {
        Self { subs: Arc::new(Mutex::new(Vec::new())) }
    }

    /// Registra una nueva suscripción. El `Receiver` lo consume el thread de la conexión;
    /// cuando ese thread termina (peer cerró), el `Sender` se desreferencia y queda muerto
    /// implícitamente. La purga real ocurre en `publish` (perezosa).
    pub fn subscribe(&self) -> Receiver<Event> {
        let (tx, rx) = mpsc::channel();
        if let Ok(mut g) = self.subs.lock() {
            g.push(tx);
        }
        rx
    }

    /// Difunde un evento a todos los suscriptores vivos. Los muertos (send falla) se
    /// purgan en sitio.
    pub fn publish(&self, ev: &Event) {
        let Ok(mut g) = self.subs.lock() else { return };
        g.retain(|tx| tx.send(ev.clone()).is_ok());
    }

    /// Sólo para tests / introspección.
    #[cfg(test)]
    pub fn live_subs(&self) -> usize {
        self.subs.lock().map(|g| g.len()).unwrap_or(0)
    }
}

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

    #[test]
    fn publish_reaches_all_subs() {
        let bus = EventBus::new();
        let a = bus.subscribe();
        let b = bus.subscribe();
        bus.publish(&Event::Crashed { service: "web".into(), code: 1 });
        assert!(matches!(a.try_recv().unwrap(), Event::Crashed { code: 1, .. }));
        assert!(matches!(b.try_recv().unwrap(), Event::Crashed { code: 1, .. }));
    }

    #[test]
    fn dead_subs_get_purged() {
        let bus = EventBus::new();
        let a = bus.subscribe();
        let b = bus.subscribe();
        assert_eq!(bus.live_subs(), 2);
        drop(b); // simula desconexión del peer
        bus.publish(&Event::Crashed { service: "x".into(), code: 0 });
        // a sigue vivo; b se purgó al fallar el send.
        assert!(a.try_recv().is_ok());
        assert_eq!(bus.live_subs(), 1);
    }

    #[test]
    fn no_subs_is_noop() {
        let bus = EventBus::new();
        bus.publish(&Event::Crashed { service: "x".into(), code: 0 });
        // No panic, no nada.
    }
}