//! Sando's concrete event payload, carried on the generic
//! [`ops_core::eventbus`] bus.
//!
//! Sites that log via `tracing::info!` also emit a typed event onto the bus. The WS handler at `/events` subscribes to both of the bus's
//! channels and forwards each envelope to the connected TUI as a JSON text
//! frame. `GateLogChunk` is the high-rate variant and rides the log channel;
//! see [`ops_core::eventbus`] for why that split exists.

use crate::domain::{GateKind, GateRunId, GitSha, NodeId, TierId, Version};
use crate::outcome::{DeployFailureKind, GateOutcome};
use ops_core::eventbus::{self, BusEvent};
use serde::{Deserialize, Serialize};

pub use eventbus::{LOG_CAPACITY, STATUS_CAPACITY};

/// The handle build/gate/deploy sites hold and emit on. Kept named `EventTx` so
/// the many `events: EventTx` fields and `s.events.clone()` sites read the same.
pub type EventTx = eventbus::EventTx<Event>;
pub type EventBus = eventbus::EventBus<Event>;
pub type EventEnvelope = eventbus::EventEnvelope<Event>;

#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Event {
    /// A /rebuild was accepted (post-receive hook or operator).
    RebuildRequested {
        sha: GitSha,
    },
    /// A previous in-flight build was aborted because a newer /rebuild arrived.
    BuildAborted {
        sha_aborted: GitSha,
    },
    BuildStart {
        sha: GitSha,
        version: Version,
    },
    BuildOk {
        sha: GitSha,
        version: Version,
        elapsed_s: u64,
    },
    BuildFailed {
        sha: GitSha,
        version: Version,
        elapsed_s: u64,
    },
    GateStart {
        run_id: GateRunId,
        tier: TierId,
        version: Version,
        gate: GateKind,
    },
    /// Chunk of combined stdout+stderr from a gate that's currently running.
    /// `run_id` correlates back to the `GateStart` for the same gate run; the
    /// TUI uses it to group chunks if it wants a per-run buffer. `seq` is a
    /// monotonic counter scoped to one run (resets across runs). `text` is a
    /// UTF-8-lossy slice of bytes — chunks reflect tokio read boundaries, not
    /// line boundaries; the on-disk log at `outcome.log_ref` is the full,
    /// byte-exact stream.
    GateLogChunk {
        run_id: GateRunId,
        seq: u32,
        text: String,
    },
    /// `outcome` carries classification, blocker variants, and the log_ref —
    /// consumers should read `outcome.status` to decide pass/fail/blocked.
    GateDone {
        run_id: GateRunId,
        tier: TierId,
        version: Version,
        gate: GateKind,
        outcome: GateOutcome,
    },
    DeployStart {
        tier: TierId,
        node: NodeId,
        version: Version,
    },
    DeployOk {
        tier: TierId,
        node: NodeId,
        version: Version,
    },
    DeployFailed {
        tier: TierId,
        node: NodeId,
        version: Version,
        failure: DeployFailureKind,
    },
    PromoteComplete {
        tier: TierId,
        version: Version,
    },
    Rollback {
        tier: TierId,
        from: Version,
        to: Version,
    },
    /// `source` is an ssh URL, kept freeform on purpose — it's a transport
    /// detail, not a domain identifier.
    BackupFetched {
        source: String,
        byte_size: i64,
    },
    ManualConfirm {
        tier: TierId,
        version: Version,
    },
}

impl BusEvent for Event {
    /// A busy gate emits `GateLogChunk` hundreds of times a second; every other
    /// variant is lifecycle status the operator must not miss.
    fn is_high_rate(&self) -> bool {
        matches!(self, Event::GateLogChunk { .. })
    }
}

pub fn channel() -> EventTx {
    eventbus::channel()
}

/// Send an event without caring whether anyone is listening. `GateLogChunk` goes
/// to the high-rate log channel, everything else to the status channel.
pub fn emit(bus: &EventTx, event: Event) {
    eventbus::emit(bus, event);
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;
    use tokio::sync::broadcast;

    #[test]
    fn emit_with_zero_subscribers_does_not_panic() {
        // The whole point of `let _ = tx.send(...)` is that emitting into an
        // unsubscribed bus is fine. Verify the contract — if this regresses
        // to `.unwrap()` someday, every build/deploy site will start
        // crashing.
        let tx = channel();
        emit(
            &tx,
            Event::RebuildRequested {
                sha: GitSha::parse("abc1234").unwrap(),
            },
        );
        emit(
            &tx,
            Event::BackupFetched {
                source: "x".into(),
                byte_size: 1,
            },
        );
    }

    #[tokio::test]
    async fn emit_reaches_a_subscriber() {
        let tx = channel();
        let mut rx = tx.subscribe_status();
        emit(
            &tx,
            Event::PromoteComplete {
                tier: TierId::new("a"),
                version: "0.8.12".parse().unwrap(),
            },
        );
        let env = rx.recv().await.expect("envelope");
        match env.event {
            Event::PromoteComplete { tier, version } => {
                assert_eq!(tier.as_str(), "a");
                assert_eq!(version.to_string(), "0.8.12");
            }
            _ => panic!("wrong event kind"),
        }
    }

    #[tokio::test]
    async fn status_survives_a_log_firehose() {
        // The point of the split: flooding the log channel past its capacity
        // must NOT evict a status event. Subscribe to both, overflow logs, then
        // confirm the status event still arrives.
        let tx = channel();
        let mut status_rx = tx.subscribe_status();
        let mut logs_rx = tx.subscribe_logs();
        emit(
            &tx,
            Event::PromoteComplete {
                tier: TierId::new("b"),
                version: "0.8.12".parse().unwrap(),
            },
        );
        for i in 0..(LOG_CAPACITY + 50) {
            emit(
                &tx,
                Event::GateLogChunk {
                    run_id: GateRunId(1),
                    seq: i as u32,
                    text: "x".into(),
                },
            );
        }
        // The status event is intact despite the log overflow.
        let env = status_rx.recv().await.expect("status event survived");
        assert!(matches!(env.event, Event::PromoteComplete { .. }));
        // The log channel is the one that lagged.
        assert!(matches!(
            logs_rx.recv().await,
            Err(broadcast::error::RecvError::Lagged(_))
        ));
    }

    #[tokio::test]
    async fn envelope_serializes_with_flat_kind() {
        // Contract for the WS handler + TUI's `format_event`: the JSON has a
        // top-level `kind` field, not nested under `event`. Locking this in.
        let env = EventEnvelope {
            at: Utc::now(),
            event: Event::GateStart {
                run_id: GateRunId(42),
                tier: TierId::new("host"),
                version: "0.8.12".parse().unwrap(),
                gate: GateKind::CargoTest,
            },
        };
        let s = serde_json::to_string(&env).unwrap();
        let v: serde_json::Value = serde_json::from_str(&s).unwrap();
        assert_eq!(v["kind"], "gate_start");
        assert_eq!(v["tier"], "host");
        assert_eq!(v["gate"], "cargo_test");
        // No nested `event` object.
        assert!(v.get("event").is_none());
    }

    #[tokio::test]
    async fn lagged_subscriber_observes_recv_error_lagged() {
        // If a subscriber falls behind by more than CAPACITY, the next
        // recv() returns RecvError::Lagged(n) — not Closed, not a panic.
        // The WS handler turns this into a `lagged` envelope.
        let tx = channel();
        let mut rx = tx.subscribe_status();
        for i in 0..(STATUS_CAPACITY + 10) {
            // 7+ hex chars satisfy GitSha::parse; pad i into that shape.
            let sha = GitSha::parse(&format!("{i:0>7x}")).unwrap();
            emit(&tx, Event::RebuildRequested { sha });
        }
        let err = rx.recv().await.expect_err("expected Lagged");
        match err {
            tokio::sync::broadcast::error::RecvError::Lagged(n) => assert!(n >= 10),
            other @ tokio::sync::broadcast::error::RecvError::Closed => {
                panic!("unexpected error: {other:?}")
            }
        }
    }
}