launcher/src/web/events.rs

//! SSE event bus + state snapshot for the dashboard.
//!
//! One multiplexed `broadcast` channel; `/events` forwards every event and
//! the UI routes by `type`. Emits `state` (supervisor + job finishes),
//! `job` (registry transitions), and `log` (attached window lines).

use crate::config::Config;
use crate::web::probe::ContainerSnap;
use serde::Serialize;
use std::time::{SystemTime, UNIX_EPOCH};

/// One SSE event (serialized as `{"type":"state"|"job", ...}` on the wire).
#[derive(Serialize, Clone, Debug)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum Event {
    State {
        ts: u64,
        snapshot: Snapshot,
    },
    Job {
        id: String,
        /// "serve" | "stop" | "power"
        kind: &'static str,
        event: String,
        cluster: String,
        recipe: Option<String>,
        message: String,
    },
    Log {
        /// Attach id (matches `POST /api/logs` response).
        attach: String,
        cluster: String,
        node: String,
        recipe: String,
        rank: u32,
        line: String,
    },
}

/// Per-node view for the LED grid.
#[derive(Serialize, Clone, Debug)]
pub struct NodeSnap {
    pub name: String,
    pub ssh_ip: String,
    pub reachable: bool,
    pub occupied: bool,
    pub containers: Vec<ContainerSnap>,
    /// Node rank — log windows attach the rank's `docker logs` stream.
    pub rank: u32,
}

/// Per-cluster view.
#[derive(Serialize, Clone, Debug)]
pub struct ClusterSnap {
    pub name: String,
    pub nodes: Vec<NodeSnap>,
    /// All nodes reachable AND blocking HA switches on (cluster verdict).
    pub up: bool,
    /// A blocking `extra_entity` reports anything but "on".
    pub switch_off: bool,
    /// Non-blocking entities that are off (display only, e.g. a fan).
    pub soft_off: Vec<String>,
    /// Free nodes; 0 when the cluster is down.
    pub free: u32,
}

/// One log window attach (for state snapshots — the UI restores open
/// log windows from `/api/state` after a reload).
#[derive(Serialize, Clone, Debug)]
pub struct AttachSnap {
    pub id: String,
    pub cluster: String,
    pub node: String,
    pub recipe: String,
    pub rank: u32,
    /// Stream ended on its own (container stopped, SSH drop).
    pub done: bool,
}

/// The full dashboard snapshot (also the `GET /api/state` body).
#[derive(Serialize, Clone, Debug)]
pub struct Snapshot {
    pub ts: u64,
    pub clusters: Vec<ClusterSnap>,
    pub jobs: Vec<crate::web::jobs::Job>,
    pub attaches: Vec<AttachSnap>,
}

fn now_ts() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Pure snapshot builder — unit-testable without SSH.
///
/// Cluster-level semantics via [`crate::cluster_state::verdict`]: one
/// offline node or a switch off takes the whole cluster down, and a down
/// cluster reports no containers (even on nodes that are alive).
pub fn build_snapshot(
    config: &Config,
    fleet: &crate::web::probe::Fleet,
    jobs: &[crate::web::jobs::Job],
    attaches: &[AttachSnap],
) -> Snapshot {
    let nodes = &fleet.nodes;
    let switch_off = &fleet.switch_off;
    let soft_off = &fleet.soft_off;
    let mut clusters = Vec::with_capacity(config.clusters.len());
    for c in &config.clusters {
        let mut node_snaps = Vec::with_capacity(c.nodes.len());
        let mut reachables: Vec<(bool, bool)> = Vec::with_capacity(c.nodes.len());
        for n in &c.nodes {
            let s = nodes.get(&n.name);
            let reachable = s.is_some_and(|x| x.reachable);
            let occupied = s.is_some_and(|x| x.occupied);
            reachables.push((reachable, occupied));
        }
        let so = switch_off.get(&c.name).copied().unwrap_or(false);
        let v = crate::cluster_state::verdict(&reachables, so);
        for n in &c.nodes {
            let s = nodes.get(&n.name);
            let reachable = s.is_some_and(|x| x.reachable);
            let occupied = s.is_some_and(|x| x.occupied);
            // A down cluster reports no containers — the cluster is off as
            // a whole, whatever individual nodes still run.
            let containers = if v.up {
                s.map(|x| x.containers.clone()).unwrap_or_default()
            } else {
                Vec::new()
            };
            node_snaps.push(NodeSnap {
                name: n.name.clone(),
                ssh_ip: n.ssh_ip.clone(),
                reachable,
                occupied,
                containers,
                rank: n.rank,
            });
        }
        clusters.push(ClusterSnap {
            name: c.name.clone(),
            nodes: node_snaps,
            up: v.up,
            switch_off: so,
            soft_off: soft_off.get(&c.name).cloned().unwrap_or_default(),
            free: v.free,
        });
    }
    Snapshot {
        ts: now_ts(),
        clusters,
        jobs: jobs.to_vec(),
        attaches: attaches.to_vec(),
    }
}

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

    fn config() -> Config {
        Config {
            ssh_user: "u".into(),
            ssh_key_path: "/k".into(),
            clusters: vec![
                crate::config::Cluster {
                    name: "a".into(),
                    master_ib_ip: None,
                    nodes: vec![
                        crate::config::Node {
                            name: "n0".into(),
                            ssh_ip: "10.0.0.1".into(),
                            ib_ip: None,
                            rank: 0,
                            host_key: None,
                            entity: None,
                        },
                        crate::config::Node {
                            name: "n1".into(),
                            ssh_ip: "10.0.0.2".into(),
                            ib_ip: None,
                            rank: 1,
                            host_key: None,
                            entity: None,
                        },
                    ],
                    extra_entities: vec![],
                },
                crate::config::Cluster {
                    name: "b".into(),
                    master_ib_ip: None,
                    nodes: vec![crate::config::Node {
                        name: "n2".into(),
                        ssh_ip: "10.0.0.3".into(),
                        ib_ip: None,
                        rank: 0,
                        host_key: None,
                        entity: None,
                    }],
                    extra_entities: vec![],
                },
            ],
            nccl: std::collections::HashMap::new(),
            hf_cache_dir: None,
            init_port: 25000,
            before_serve: std::collections::HashMap::new(),
            exec_aliases: std::collections::HashMap::new(),
            home_assistant: None,
            global_mods_dir: None,
            drop_caches: true,
            mount_aliases: std::collections::HashMap::new(),
            recipe_sources: Vec::new(),
            cluster_sources: Vec::new(),
        }
    }

    #[test]
    fn snapshot_reflects_probe_state() {
        let config = config();
        let mut nodes = HashMap::new();
        nodes.insert(
            "n0".into(),
            crate::web::probe::WebNodeState {
                reachable: true,
                occupied: true,
                containers: vec![ContainerSnap {
                    name: "sm121_r_0".into(),
                    status: "Up 3 hours".into(),
                    cluster: "a".into(),
                    recipe: "r".into(),
                }],
            },
        );
        nodes.insert(
            "n1".into(),
            crate::web::probe::WebNodeState {
                reachable: true,
                occupied: false,
                containers: vec![],
            },
        );
        // n2 absent → unreachable.
        let fleet = crate::web::probe::Fleet {
            nodes,
            switch_off: std::collections::HashMap::new(),
            soft_off: std::collections::HashMap::new(),
        };
        let snap = build_snapshot(&config, &fleet, &[], &[]);
        assert_eq!(snap.clusters.len(), 2);
        let a = &snap.clusters[0];
        assert!(a.up);
        assert!(a.nodes[0].reachable && a.nodes[0].occupied);
        assert_eq!(a.nodes[0].containers[0].recipe, "r");
        assert!(a.nodes[1].reachable && !a.nodes[1].occupied);
        assert_eq!(a.free, 1);
        // cluster b has one offline node → down as a whole
        let b = &snap.clusters[1];
        assert!(!b.up);
        assert!(!b.nodes[0].reachable);
        assert_eq!(b.free, 0);
        assert!(snap.jobs.is_empty());
    }

    #[test]
    fn down_cluster_hides_containers_on_live_nodes() {
        // Both nodes of cluster a are live; n1 runs a container but n0 is
        // offline → cluster a is down as a whole and the live node's
        // container must not be reported.
        let config = config();
        let mut nodes = HashMap::new();
        nodes.insert("n0".into(), crate::web::probe::WebNodeState::default());
        nodes.insert(
            "n1".into(),
            crate::web::probe::WebNodeState {
                reachable: true,
                occupied: true,
                containers: vec![ContainerSnap {
                    name: "c2".into(),
                    status: "Up".into(),
                    cluster: "a".into(),
                    recipe: "r".into(),
                }],
            },
        );
        let fleet = crate::web::probe::Fleet {
            nodes,
            switch_off: std::collections::HashMap::new(),
            soft_off: std::collections::HashMap::new(),
        };
        let snap = build_snapshot(&config, &fleet, &[], &[]);
        let a = &snap.clusters[0];
        assert!(!a.up, "one offline node takes the cluster down");
        assert_eq!(a.free, 0);
        assert!(
            a.nodes[1].containers.is_empty(),
            "containers hidden on a down cluster"
        );
        assert!(a.nodes[1].reachable, "reachability is still shown");
    }

    #[test]
    fn switch_off_takes_cluster_down() {
        let config = config();
        let mut nodes = HashMap::new();
        for n in ["n0", "n1", "n2"] {
            nodes.insert(
                n.into(),
                crate::web::probe::WebNodeState {
                    reachable: true,
                    occupied: false,
                    containers: vec![],
                },
            );
        }
        let mut so = HashMap::new();
        so.insert("a".into(), true);
        let fleet = crate::web::probe::Fleet {
            nodes,
            switch_off: so,
            soft_off: std::collections::HashMap::new(),
        };
        let snap = build_snapshot(&config, &fleet, &[], &[]);
        assert!(!snap.clusters[0].up);
        assert!(snap.clusters[0].switch_off);
        assert_eq!(snap.clusters[0].free, 0);
        assert!(snap.clusters[1].up);
        assert_eq!(snap.clusters[1].free, 1);
    }

    #[test]
    fn soft_off_does_not_take_cluster_down() {
        // A non-blocking entity (fan) is off, all nodes live → cluster
        // stays up; the off entity is surfaced for display only.
        let config = config();
        let mut nodes = HashMap::new();
        for n in ["n0", "n1", "n2"] {
            nodes.insert(
                n.into(),
                crate::web::probe::WebNodeState {
                    reachable: true,
                    occupied: false,
                    containers: vec![],
                },
            );
        }
        let mut soft = HashMap::new();
        soft.insert("a".into(), vec!["switch.fan".to_string()]);
        let fleet = crate::web::probe::Fleet {
            nodes,
            switch_off: HashMap::new(),
            soft_off: soft,
        };
        let snap = build_snapshot(&config, &fleet, &[], &[]);
        assert!(
            snap.clusters[0].up,
            "non-blocking off must not down the cluster"
        );
        assert_eq!(snap.clusters[0].free, 2);
        assert_eq!(snap.clusters[0].soft_off, vec!["switch.fan".to_string()]);
        assert!(snap.clusters[1].soft_off.is_empty());
    }

    #[test]
    fn job_event_serializes_with_type_tag() {
        let ev = Event::Job {
            id: "j1".into(),
            kind: "serve",
            event: "started".into(),
            cluster: "c".into(),
            recipe: Some("r".into()),
            message: "ok".into(),
        };
        let json = serde_json::to_value(&ev).unwrap();
        assert_eq!(json["type"], "job");
        assert_eq!(json["kind"], "serve");
        assert_eq!(json["id"], "j1");
        assert_eq!(json["event"], "started");
    }
}