//! Generic event bus for live operator visibility.
//!
//! Sando and Bento both broadcast typed events to a connected TUI over
//! `tokio::sync::broadcast` channels; the WS handler forwards each envelope as
//! a JSON text frame. The only thing that differs between the tools is the
//! concrete event payload, so the bus is generic over `E`.
//!
//! The envelope flattens the payload, so a payload enum tagged
//! `#[serde(tag = "kind", rename_all = "snake_case")]` serializes with a
//! top-level `kind` field plus the variant's fields inline — the wire shape
//! the TUIs parse.
//!
//! ## Why two channels
//!
//! Both tools emit two very different rates of event on one bus: a low-rate
//! *status* stream (a promote completed, a build failed — the things an
//! operator must not miss) and a high-rate *log chunk* stream (hundreds per
//! second from a busy gate or build step). On a single broadcast channel the
//! log firehose evicts the status events, because `broadcast` drops the oldest
//! entries once a subscriber falls `CAPACITY` behind — a subscriber that lags
//! on chunk spam loses the `PromoteComplete` sitting in the same ring.
//!
//! Sando hit this for real (ultra-fuzz Run 2) and split its bus in two. This
//! module is that design, generalized: [`BusEvent::is_high_rate`] classifies
//! each event, [`emit`] routes it to the matching channel, and a WS handler
//! subscribes to both and merges. A lag on one stream can no longer drop the
//! other.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;

/// Capacity of the low-rate *status* channel (build/gate/deploy/promote
/// lifecycle events). Subscribers that fall behind by more than this many
/// events get `RecvError::Lagged`; WS handlers treat that as a recoverable
/// hiccup (emit a `lagged` notice), not a disconnect.
pub const STATUS_CAPACITY: usize = 256;

/// Capacity of the high-rate *log* channel. Larger than [`STATUS_CAPACITY`] so
/// a momentarily-paused TUI lags the log stream rather than the status stream.
pub const LOG_CAPACITY: usize = 1024;

/// Lets the bus tell a high-rate log-chunk event from a low-rate status event,
/// so each rides its own channel. See the module docs for why this matters.
pub trait BusEvent: Clone {
    /// `true` for the high-rate log-chunk variants (Sando's `GateLogChunk`,
    /// Bento's `StepLogChunk`) — the ones a busy run emits hundreds of per
    /// second. Everything an operator must not miss returns `false`.
    fn is_high_rate(&self) -> bool;
}

/// An event payload `E` stamped with the time it was emitted.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct EventEnvelope<E> {
    pub at: DateTime<Utc>,
    #[serde(flatten)]
    pub event: E,
}

/// Two broadcast channels behind one handle: low-rate status and high-rate log
/// chunks. [`emit`] routes each event to the right one; a WS handler subscribes
/// to both and merges them, so a lag on one never drops the other.
///
/// This is the handle every emit site clones, and it stays named `EventTx` via
/// the alias below so call sites read the same as a plain sender.
#[derive(Clone)]
pub struct EventBus<E> {
    status: broadcast::Sender<EventEnvelope<E>>,
    logs: broadcast::Sender<EventEnvelope<E>>,
}

impl<E: Clone> EventBus<E> {
    /// Subscribe to the low-rate status stream (every event that is not
    /// [`BusEvent::is_high_rate`]).
    pub fn subscribe_status(&self) -> broadcast::Receiver<EventEnvelope<E>> {
        self.status.subscribe()
    }

    /// Subscribe to the high-rate log-chunk stream.
    pub fn subscribe_logs(&self) -> broadcast::Receiver<EventEnvelope<E>> {
        self.logs.subscribe()
    }
}

/// The handle every emit site clones. `subscribe_status`/`subscribe_logs` yield
/// receivers for the WS handler.
pub type EventTx<E> = EventBus<E>;

/// Create a fresh bus. The dropped receivers from `broadcast::channel` are fine
/// — emitters don't care whether anyone is listening (see [`emit`]).
pub fn channel<E: Clone>() -> EventTx<E> {
    EventBus {
        status: broadcast::channel(STATUS_CAPACITY).0,
        logs: broadcast::channel(LOG_CAPACITY).0,
    }
}

/// Send an event without caring whether anyone is subscribed. High-rate events
/// go to the log channel, everything else to the status channel. `send` errors
/// only when there are zero receivers, which is the normal idle case for an
/// operator tool, so the error is intentionally dropped.
pub fn emit<E: BusEvent>(tx: &EventTx<E>, event: E) {
    let envelope = EventEnvelope {
        at: Utc::now(),
        event,
    };
    let sender = if envelope.event.is_high_rate() {
        &tx.logs
    } else {
        &tx.status
    };
    let _ = sender.send(envelope);
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde::{Deserialize, Serialize};

    #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
    #[serde(tag = "kind", rename_all = "snake_case")]
    enum TestEvent {
        Started { name: String },
        Done { code: i32 },
        LogChunk { seq: u32 },
    }

    impl BusEvent for TestEvent {
        fn is_high_rate(&self) -> bool {
            matches!(self, TestEvent::LogChunk { .. })
        }
    }

    #[test]
    fn emit_with_zero_subscribers_does_not_panic() {
        let tx = channel::<TestEvent>();
        emit(&tx, TestEvent::Started { name: "x".into() });
    }

    #[tokio::test]
    async fn emit_reaches_a_subscriber() {
        let tx = channel::<TestEvent>();
        let mut rx = tx.subscribe_status();
        emit(&tx, TestEvent::Done { code: 7 });
        let env = rx.recv().await.expect("envelope");
        assert_eq!(env.event, TestEvent::Done { code: 7 });
    }

    #[tokio::test]
    async fn high_rate_events_ride_the_log_channel() {
        let tx = channel::<TestEvent>();
        let mut status_rx = tx.subscribe_status();
        let mut logs_rx = tx.subscribe_logs();
        emit(&tx, TestEvent::LogChunk { seq: 1 });
        // It arrives on logs...
        assert_eq!(
            logs_rx.recv().await.unwrap().event,
            TestEvent::LogChunk { seq: 1 }
        );
        // ...and never on status.
        assert!(
            status_rx.try_recv().is_err(),
            "log chunk must not hit the status channel"
        );
    }

    #[tokio::test]
    async fn status_survives_a_log_firehose() {
        // The whole point of the split: flooding the log channel past its
        // capacity must NOT evict a status event.
        let tx = channel::<TestEvent>();
        let mut status_rx = tx.subscribe_status();
        let mut logs_rx = tx.subscribe_logs();
        emit(&tx, TestEvent::Done { code: 0 });
        for i in 0..(LOG_CAPACITY + 50) {
            emit(&tx, TestEvent::LogChunk { seq: i as u32 });
        }
        // The status event is intact despite the log overflow.
        assert_eq!(
            status_rx.recv().await.expect("status survived").event,
            TestEvent::Done { code: 0 }
        );
        // The log channel is the one that lagged.
        assert!(matches!(
            logs_rx.recv().await,
            Err(broadcast::error::RecvError::Lagged(_))
        ));
    }

    #[test]
    fn envelope_serializes_with_flat_kind() {
        let env = EventEnvelope {
            at: Utc::now(),
            event: TestEvent::Started {
                name: "build".into(),
            },
        };
        let v: serde_json::Value =
            serde_json::from_str(&serde_json::to_string(&env).unwrap()).unwrap();
        assert_eq!(v["kind"], "started");
        assert_eq!(v["name"], "build");
        // Flattened, not nested under `event`.
        assert!(v.get("event").is_none());
        // `at` is preserved.
        assert!(v.get("at").is_some());
    }

    #[tokio::test]
    async fn lagged_subscriber_observes_recv_error_lagged() {
        let tx = channel::<TestEvent>();
        let mut rx = tx.subscribe_status();
        for i in 0..(STATUS_CAPACITY + 10) {
            emit(&tx, TestEvent::Done { code: i as i32 });
        }
        let err = rx.recv().await.expect_err("expected Lagged");
        match err {
            broadcast::error::RecvError::Lagged(n) => assert!(n >= 10),
            broadcast::error::RecvError::Closed => panic!("unexpected error: Closed"),
        }
    }
}