//! `/api/state` cold-load snapshot plus the dashboard's live read side:
//! the `StateSnapshot` shape and its view builders, the
//! `/api/dashboard/stream` SSE channel. SPA shape + SSE channels:
//! docs/web-ui/shape.md.

use std::convert::Infallible;

use axum::{
    extract::State,
    http::HeaderMap,
    response::sse::{Event, KeepAlive, Sse},
};
use serde::{Deserialize, Serialize};
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::{Stream, StreamExt};
use utoipa::IntoParams;

use crate::container_view::ContainerView;

use super::AppState;
use super::meta_inputs::{MetaInputView, read_meta_inputs};
use super::tombstones::{TombstoneView, build_tombstone_views};

#[allow(clippy::struct_excessive_bools)]
#[derive(Serialize)]
pub(super) struct StateSnapshot {
    /// Broker seq at the moment this snapshot was assembled. Clients
    /// dedupe their buffered SSE traffic against this value: any
    /// `MessageEvent` with `seq <= snapshot.seq` is already reflected in
    /// the snapshot (or pre-dates it); anything with `seq > snapshot.seq`
    /// is post-snapshot and should be applied. Set to 0 in the
    /// pre-emit case (no events ever fired) — clients treat that as
    /// "apply everything you've buffered".
    seq: u64,
    hostname: String,
    /// Current rev of the pinned `hyperhive` flake input, resolved the
    /// same way `get_agent_meta`'s per-agent `hyperhive_rev` is (see
    /// `auto_update::current_flake_rev`). `None` when the flake ref
    /// isn't a local path pin (e.g. a bare `github:` url) or the rev
    /// can't be resolved. Feeds the dashboard start page so an operator
    /// can tell what build a hive is running without shelling in.
    hyperhive_rev: Option<String>,
    any_stale: bool,
    containers: Vec<ContainerView>,
    transients: Vec<TransientView>,
    /// State dirs (config history + claude creds + /state/ notes) that
    /// survive after a destroy-without-purge. The operator can re-spawn
    /// with the same name to resume, or PURG3 to wipe them.
    tombstones: Vec<TombstoneView>,
    /// Agents whose FNV-1a hashed web UI port collides with at least
    /// one other. Operator resolves by renaming. The dashboard renders
    /// a banner at the top listing each cluster.
    port_conflicts: Vec<PortConflict>,
    /// Inputs in `meta/flake.lock` the operator can selectively
    /// `nix flake update`. Hyperhive first, then `agent-<n>` rows.
    meta_inputs: Vec<MetaInputView>,
    /// True while a dashboard-triggered `meta-update` (flake lock bump +
    /// agent rebuild ripple) is running in the background. Lets a
    /// client that cold-loads mid-update render the META INPUTS panel's
    /// disabled "updating…" state; live transitions arrive via the
    /// `MetaUpdateRunning` event.
    meta_update_running: bool,
    /// Whether the hive-forge container is up. When true the dashboard
    /// links each container's config into the forge's `agent-configs`
    /// repos.
    forge_present: bool,
    /// Whether `hive-gateway` is in front of this dashboard. Sourced
    /// from the `HIVE_GATEWAY_ENABLED` env var, which the c0re NixOS
    /// module now always sets (the gateway runs unconditionally
    /// alongside hyperhive), so this is effectively always true: the
    /// dashboard frontend builds same-origin `/agent/<name>/` links to
    /// the per-agent web UI (the gateway routes them via the
    /// runtime-generated `agents.conf` include file — see
    /// `gateway_nginx.rs`). The `false` branch (direct
    /// `http://<hostname>:<port>/` TCP links) is retained as a defensive
    /// fallback for the env being unset. See `docs/networking/gateway.md::Vhost map`.
    gateway_enabled: bool,
    /// Public URL of the forge (e.g.
    /// `"https://forge.pr1ma.darkest.space"`). Sourced from the
    /// `HIVE_FORGE_PUBLIC_URL` env var, which the c0re NixOS module
    /// sets from `services.hyperhive.swarm.forge.publicUrl` (defaults to
    /// the gateway vhost URL). `None` when absent — the frontend **hides**
    /// forge links rather than guessing `http://<hostname>:3000`,
    /// which is only right by accident on deployments that aren't
    /// plain localhost.
    forge_public_url: Option<String>,
    /// Browser URL of the swarm UI (the swarm apex domain). Sourced from
    /// `HIVE_SWARM_PUBLIC_URL`, which the c0re NixOS module sets from
    /// `services.hyperhive.swarm.domain`. `None` when the domain is unset —
    /// the dashboard's pages menu then omits the swarm entry.
    swarm_public_url: Option<String>,
    /// Human name of this single-host hive instance (e.g. `"pr1ma"`).
    /// Sourced from `HYPERHIVE_HIVE_NAME` env var, set by the c0re
    /// NixOS module from `services.hyperhive.hiveName`. `None` when
    /// the option is unset — chrome falls back to `hostname`.
    hive_name: Option<String>,
    /// Human name of the wider swarm this hive belongs to (e.g.
    /// `"constellat1on"`). Sourced from `HYPERHIVE_SWARM_NAME` env
    /// var, set from `services.hyperhive.swarm.name`. `None` when
    /// unset — chrome omits the swarm segment of the breadcrumb.
    swarm_name: Option<String>,
}

#[derive(Serialize)]
struct PortConflict {
    port: u16,
    /// All agent names sharing this port (sorted, ≥2 entries).
    agents: Vec<String>,
}

#[derive(Serialize)]
struct TransientView {
    name: String,
    /// Owned: the label is the running node's wire tag, not one of a fixed set.
    kind: String,
    secs: u64,
}

/// Cold-load snapshot of the whole dashboard.
///
/// Includes the roster, tombstones, job queue, meta inputs, and more. Live
/// clients then follow `/api/dashboard/stream` (SSE) for incremental updates
/// keyed off `seq`.
// `StateSnapshot` is a large tree of nested view types (`ContainerView`,
// `TransientView`, ...) with no `ToSchema` anywhere in that
// graph; wiring it up is a schema-modelling project of its own, well past
// "annotate what's reachable". `serde_json::Value` placeholder for now —
// see the batch report.
#[utoipa::path(
    get,
    path = "/api/state",
    responses((status = 200, description = "full dashboard snapshot", body = serde_json::Value)),
    tag = "state_snapshot"
)]
pub(super) async fn api_state(
    headers: HeaderMap,
    State(state): State<AppState>,
) -> axum::Json<StateSnapshot> {
    let host = headers
        .get("host")
        .and_then(|h| h.to_str().ok())
        .unwrap_or("localhost");
    let hostname = host.split(':').next().unwrap_or(host).to_owned();

    // Capture the unified dashboard-channel seq *before* any read so the
    // dedupe contract is "events with seq > snapshot.seq are
    // post-snapshot, never missed." An event landing during snapshot
    // construction may be doubly applied (snapshot caught the write +
    // client also applies the SSE frame) — that's a renderer's problem
    // to make idempotent, not ours to avoid here.
    let seq = state.coord.current_seq();

    // Refresh the coordinator's cached container snapshot before
    // reading. Cold-load clients then see whatever the latest rescan
    // produced; live clients converge via the matching
    // `ContainerStateChanged` / `ContainerRemoved` events the rescan
    // emits.
    //
    // Bound the rescan: it shells out (`nixos-container list` etc.), so a
    // saturated/wedged build backend — e.g. hive-c0re mid-startup-sweep
    // hammering slow `nixos-container update` subprocesses — can stall it
    // long enough that `/api/state` hangs for the whole request (the
    // ~minute-long /state reported in the field). On timeout we skip the
    // fresh rescan and serve the last cached snapshot instead; live
    // clients still converge via the SSE events a later successful rescan
    // emits, and the next /state call retries the refresh. Introspection
    // stays responsive regardless of the build backend's health.
    if tokio::time::timeout(
        std::time::Duration::from_secs(3),
        state.coord.rescan_containers_and_emit(),
    )
    .await
    .is_err()
    {
        tracing::warn!(
            "api_state: container rescan exceeded 3s (build backend likely saturated); \
             serving last cached snapshot"
        );
    }
    let containers = state.coord.containers_snapshot().await;
    let any_stale = containers.iter().any(|c| c.needs_update);
    let transient_snapshot = state.coord.transient_snapshot();
    let transients = build_transient_views(&containers, &transient_snapshot);
    let tombstones = build_tombstone_views(&containers);
    let port_conflicts = build_port_conflicts(&containers);

    axum::Json(StateSnapshot {
        seq,
        hostname,
        hyperhive_rev: crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake),
        any_stale,
        containers,
        transients,
        meta_inputs: read_meta_inputs(),
        meta_update_running: state.coord.meta_update_in_progress(),
        tombstones,
        port_conflicts,
        forge_present: crate::forge::is_present().await,
        gateway_enabled: std::env::var_os("HIVE_GATEWAY_ENABLED").is_some_and(|v| {
            // Accept any truthy string ("1", "true", "yes"): the c0re
            // NixOS module sets the literal "1" (the gateway always runs
            // alongside hyperhive), and manual overrides also parse.
            let s = v.to_string_lossy().to_ascii_lowercase();
            matches!(s.as_str(), "1" | "true" | "yes")
        }),
        forge_public_url: std::env::var("HIVE_FORGE_PUBLIC_URL")
            .ok()
            .filter(|s| !s.is_empty()),
        swarm_public_url: std::env::var("HIVE_SWARM_PUBLIC_URL")
            .ok()
            .filter(|s| !s.is_empty()),
        hive_name: std::env::var("HYPERHIVE_HIVE_NAME")
            .ok()
            .filter(|s| !s.is_empty()),
        swarm_name: std::env::var("HYPERHIVE_SWARM_NAME")
            .ok()
            .filter(|s| !s.is_empty()),
    })
}

/// Group containers by their assigned web UI port; clusters with more
/// than one member are port-hash collisions the operator resolves by
/// renaming an agent.
///
/// Every container's port comes from [`crate::lifecycle::agent_web_port`],
/// which hashes the name into one range for all of them — so the manager
/// is a candidate for a collision exactly like any sub-agent.
fn build_port_conflicts(containers: &[ContainerView]) -> Vec<PortConflict> {
    let mut by_port: std::collections::BTreeMap<u16, Vec<String>> =
        std::collections::BTreeMap::new();
    for c in containers {
        by_port.entry(c.port).or_default().push(c.name.clone());
    }
    by_port
        .into_iter()
        .filter(|(_, agents)| agents.len() > 1)
        .map(|(port, mut agents)| {
            agents.sort();
            PortConflict { port, agents }
        })
        .collect()
}

/// Transient state for agents whose container does NOT yet exist
/// (`Spawning`). Lifecycle ops on existing containers surface as
/// `ContainerView.pending` inline; this list only catches pre-creation.
fn build_transient_views(
    containers: &[ContainerView],
    transient_snapshot: &std::collections::HashMap<String, Vec<crate::job_queue::RunningTransient>>,
) -> Vec<TransientView> {
    let mut out: Vec<TransientView> = transient_snapshot
        .iter()
        .filter(|(name, _)| !containers.iter().any(|c| &c.name == *name))
        // One row per running node, so an agent with several shows several
        // rather than one of them arbitrarily.
        .flat_map(|(name, sts)| sts.iter().map(move |st| (name, st)))
        .map(|(name, st)| TransientView {
            name: name.clone(),
            kind: st.label.clone(),
            // Clamped at 0: `since` is wall-clock now (the node's own
            // `started_at`), so a backwards clock adjustment could otherwise
            // render a negative age.
            secs: (chrono::Utc::now() - st.since)
                .num_seconds()
                .max(0)
                .cast_unsigned(),
        })
        .collect();
    // The caller rebuilds this map on every poll, so its iteration order
    // differs between renders and the rows would shuffle under an operator
    // watching an operation run. Sorted for the same reason
    // `build_port_conflicts` sorts its clusters.
    out.sort_by(|a, b| {
        a.name
            .cmp(&b.name)
            .then_with(|| a.kind.cmp(&b.kind))
            .then_with(|| a.secs.cmp(&b.secs))
    });
    out
}

/// `/api/jobq/graph` query string — the generic jobq/wire query shape (any
/// host serving a `GraphWire` projection over HTTP takes the same one; see
/// `hive_jobq_wire::parse_states`, which does the actual parsing this side
/// of the query string). Today's only field is `states`: a comma-separated
/// allow-list of `hive_jobq::State` names (`"Pending"`, `"Running"`, ...).
/// Empty / absent ⇒ no filter (current behaviour, every visible root). Set
/// ⇒ only **root** groups whose own state is named are served — a root's
/// state is already its subtree's rolled-up answer (see `hive_jobq_wire`'s
/// doc), so filtering the root filters the whole group. Unknown tokens are
/// silently ignored (an unrecognised name matches nothing rather than
/// erroring the whole request), mirroring `DashboardStreamQuery::kinds`
/// above.
#[derive(Deserialize, Default, IntoParams)]
pub(super) struct JobqGraphQuery {
    states: Option<String>,
}

#[utoipa::path(
    get,
    path = "/api/jobq/graph",
    params(JobqGraphQuery),
    responses(
        (status = 200, description = "every node of every retained job group, \
            as generic `hive_jobq` graph nodes: identity, the parent tree, \
            dependency edges with their accepted-outcome sets, lifecycle, and \
            one opaque per-node payload. This is the generic jobq/wire shape \
            (`hive_jobq_wire::GraphWire::wire_snapshot`), not a hive-c0re-only \
            projection — a `swarm-controller` serving its own graph responds \
            with the same shape at the same path. Group roots ride as \
            ordinary nodes (`parent: null`) and `Done` nodes are not \
            filtered by default — a consumer renders the graph without \
            knowing what any node means. `?states=` narrows to root groups \
            in the named states.",
            body = Vec<hive_jobq_wire::GraphNode>),
    ),
    tag = "state_snapshot"
)]
pub(super) async fn jobq_graph(
    State(state): State<AppState>,
    axum::extract::Query(q): axum::extract::Query<JobqGraphQuery>,
) -> axum::Json<Vec<hive_jobq_wire::GraphNode>> {
    let states = hive_jobq_wire::parse_states(q.states.as_deref());
    axum::Json(state.coord.job_queue.graph_snapshot(states.as_deref()))
}

#[utoipa::path(
    get,
    path = "/api/jobq/rollup",
    responses(
        (status = 200, description = "counts by lifecycle state over the same \
            groups `/api/jobq/graph` serves, as `(state, nodes, roots)` \
            triples — the generic jobq/wire roll-up shape \
            (`hive_jobq_wire::state_rollup`), same as `/api/jobq/graph` \
            above. Every state is present, zero counts included, in a fixed \
            order — a consumer renders a summary (\"3 running · 2 queued\") \
            without fetching the graph and without re-deriving the tally. \
            `roots` counts groups, `nodes` counts every step at any depth: one \
            rebuild is 1 root and ~7 nodes, so a summary meaning *operations* \
            reads `roots` and one meaning *steps* reads `nodes`.",
            body = Vec<hive_jobq_wire::StateCount>),
    ),
    tag = "state_snapshot"
)]
pub(super) async fn jobq_rollup(
    State(state): State<AppState>,
) -> axum::Json<Vec<hive_jobq_wire::StateCount>> {
    axum::Json(state.coord.job_queue.state_rollup())
}

/// `/dashboard/stream` query string. Today's only field is `kinds`:
/// a comma-separated allow-list of event-`kind` strings.
/// Empty / absent ⇒ no filter (current behaviour, all variants
/// forwarded). Set ⇒ only the named kinds reach the subscriber,
/// non-matches are skipped before the JSON serialise cost.
///
/// Useful for narrow pages that only care about a few kinds and want to
/// drop the dispatch overhead on every unrelated mutation.
#[derive(Deserialize, Default, IntoParams)]
pub(super) struct DashboardStreamQuery {
    /// Comma-separated event kinds to forward. Each token is
    /// trimmed; unknown kinds are silently ignored on lookup
    /// (subscriber sees nothing instead of an error).
    kinds: Option<String>,
}

#[utoipa::path(
    get,
    path = "/api/dashboard/stream",
    params(DashboardStreamQuery),
    responses(
        (status = 200, description = "server-sent event stream; each event's \
            `data` is a JSON-serialised `DashboardEvent` (seq-tagged)",
            body = String, content_type = "text/event-stream"),
    ),
    tag = "state_snapshot"
)]
pub(super) async fn dashboard_stream(
    State(state): State<AppState>,
    axum::extract::Query(q): axum::extract::Query<DashboardStreamQuery>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let rx = state.coord.dashboard_subscribe();
    // Pre-parse the allow-list once at subscription time, so the
    // per-event hot path is just a `HashSet::contains` on a
    // `&'static str` — no string churn per frame.
    let kind_filter: Option<std::collections::HashSet<String>> = q.kinds.and_then(|raw| {
        let set: std::collections::HashSet<String> = raw
            .split(',')
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(str::to_owned)
            .collect();
        if set.is_empty() { None } else { Some(set) }
    });
    let stream = BroadcastStream::new(rx).filter_map(move |res| {
        // Drop lagged frames. Browsers reconnect; the seq dedupe on
        // reconnect skips any frame already reflected in the snapshot.
        let event = res.ok()?;
        if let Some(filter) = kind_filter.as_ref()
            && !filter.contains(event.kind_tag())
        {
            return None;
        }
        let json = serde_json::to_string(&event).ok()?;
        Some(Ok(Event::default().data(json)))
    });
    Sse::new(stream).keep_alive(KeepAlive::default())
}

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

    use chrono::{Duration, Utc};

    use super::{ContainerView, build_port_conflicts, build_transient_views};
    use crate::job_queue::RunningTransient;

    fn rt(agent: &str, label: &str, age_secs: i64) -> RunningTransient {
        RunningTransient {
            agent: agent.to_owned(),
            label: label.to_owned(),
            takes_container_down: false,
            since: Utc::now() - Duration::seconds(age_secs),
        }
    }

    fn snapshot(rows: Vec<RunningTransient>) -> HashMap<String, Vec<RunningTransient>> {
        let mut m: HashMap<String, Vec<RunningTransient>> = HashMap::new();
        for r in rows {
            m.entry(r.agent.clone()).or_default().push(r);
        }
        m
    }

    #[test]
    fn an_agent_whose_container_exists_is_not_transient() {
        let snap = snapshot(vec![rt("alice", "spawn", 5)]);
        let live = [cv("alice", 8101, true)];
        assert!(build_transient_views(&live, &snap).is_empty());
        // Control: with no container by that name it *is* reported, so the
        // emptiness above is the filter and not an inert builder.
        assert_eq!(build_transient_views(&[], &snap).len(), 1);
    }

    #[test]
    fn an_agent_with_several_running_nodes_gets_a_row_each() {
        let snap = snapshot(vec![rt("alice", "spawn", 5), rt("alice", "update", 3)]);
        assert_eq!(build_transient_views(&[], &snap).len(), 2);
    }

    /// The snapshot is a fresh `HashMap` per poll, so its iteration order is
    /// not stable across renders. Sorting is what keeps a row still.
    #[test]
    fn rows_are_ordered_by_name_then_kind() {
        let snap = snapshot(vec![
            rt("zoe", "spawn", 1),
            rt("adam", "update", 1),
            rt("adam", "spawn", 1),
        ]);
        let got: Vec<(String, String)> = build_transient_views(&[], &snap)
            .into_iter()
            .map(|v| (v.name, v.kind))
            .collect();
        assert_eq!(
            got,
            [
                ("adam".to_owned(), "spawn".to_owned()),
                ("adam".to_owned(), "update".to_owned()),
                ("zoe".to_owned(), "spawn".to_owned()),
            ]
        );
    }

    /// A `since` in the future yields a negative duration, and casting that
    /// unsigned would render an age near `u64::MAX` rather than a small one.
    #[test]
    fn a_future_since_clamps_to_zero_instead_of_wrapping() {
        let snap = snapshot(vec![rt("alice", "spawn", -600)]);
        let got = build_transient_views(&[], &snap);
        assert_eq!(got[0].secs, 0, "clamped, not wrapped");
    }

    #[test]
    fn an_empty_snapshot_produces_no_rows() {
        assert!(build_transient_views(&[], &snapshot(vec![])).is_empty());
    }

    fn cv(name: &str, port: u16, running: bool) -> ContainerView {
        ContainerView {
            name: name.to_owned(),
            container: format!("h-{name}"),
            port,
            running,
            failed: false,
            needs_update: false,
            needs_login: false,
            deployed_sha: None,
            active_model: None,
            status_text: None,
            status_set_at: None,
            paused: false,
        }
    }

    #[test]
    fn distinct_ports_produce_no_conflicts() {
        let cs = [cv("alice", 8101, true), cv("bob", 8102, true)];
        assert!(build_port_conflicts(&cs).is_empty());
    }

    #[test]
    fn a_lone_container_does_not_conflict_with_itself() {
        assert!(build_port_conflicts(&[cv("alice", 8101, true)]).is_empty());
    }

    #[test]
    fn two_agents_on_one_port_are_reported_together() {
        let cs = [cv("alice", 8101, true), cv("bob", 8101, true)];
        let got = build_port_conflicts(&cs);
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].port, 8101);
        assert_eq!(got[0].agents, ["alice", "bob"]);
    }

    /// The manager takes its port from the same hash as everyone else, so
    /// it collides like any other name. A doc comment here used to claim it
    /// sat at a fixed 8000 and could not — no such special case exists.
    #[test]
    fn the_manager_can_collide_with_a_sub_agent() {
        let cs = [cv("manager", 8101, true), cv("alice", 8101, true)];
        let got = build_port_conflicts(&cs);
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].agents, ["alice", "manager"]);
    }

    /// A stopped agent keeps its hashed port, so the clash is real the
    /// moment it starts. Reporting it while it is down is the point.
    #[test]
    fn a_stopped_container_still_claims_its_port() {
        let cs = [cv("alice", 8101, true), cv("bob", 8101, false)];
        assert_eq!(build_port_conflicts(&cs).len(), 1);
    }

    #[test]
    fn names_are_sorted_within_a_conflict() {
        let cs = [cv("zoe", 8101, true), cv("adam", 8101, true)];
        assert_eq!(build_port_conflicts(&cs)[0].agents, ["adam", "zoe"]);
    }

    #[test]
    fn more_than_two_agents_land_in_one_cluster() {
        let cs = [
            cv("c", 8101, true),
            cv("a", 8101, true),
            cv("b", 8101, true),
        ];
        let got = build_port_conflicts(&cs);
        assert_eq!(got.len(), 1, "one cluster, not one row per pair");
        assert_eq!(got[0].agents, ["a", "b", "c"]);
    }

    #[test]
    fn separate_clusters_are_ordered_by_port() {
        let cs = [
            cv("d", 8300, true),
            cv("c", 8300, true),
            cv("b", 8200, true),
            cv("a", 8200, true),
        ];
        let ports: Vec<u16> = build_port_conflicts(&cs).iter().map(|p| p.port).collect();
        assert_eq!(ports, [8200, 8300]);
    }

    #[test]
    fn no_containers_is_not_a_conflict() {
        assert!(build_port_conflicts(&[]).is_empty());
    }
}