nornir 0.4.21

//! 🚀 **Release** tab — the release-op DAG (`release_events`) as a viz surface.
//!
//! Two panes over the same data, both driven by the warehouse
//! [`release_events`](crate::warehouse::release_events) table:
//!
//!   1. **Op-log** (B3) — a streaming worklog, one line per
//!      `component × op × phase` boundary, prefixed by the component it acts on:
//!      `[znippy] test start`, `[znippy] test end ok (1.2s)`, `[holger] gate end ok`.
//!      Grouped by `run_id`, **newest run on top**, `seq`-ordered within a run.
//!   2. **Lit-up dep-graph** (B4) — the rectangular dep-graph (reusing
//!      [`super::graph`]'s node/edge geometry) where each node is **coloured by
//!      its latest `release_events` status for the selected run**: running = amber
//!      pulse, ok = green, fail = red, idle = dim. As a run progresses (the table
//!      gains rows), the graph lights up.
//!
//! Read path: **local** mode opens the warehouse read-only and calls
//! [`query_release_events`]. **Remote** mode reads the SAME rows over the
//! `Viz.ReleaseEvents` gRPC (the warehouse lives on the server, which holds the
//! redb lock), decoding them into the identical [`ReleaseEventRow`] — so the
//! op-log + lit-up dep-graph render byte-identically whether the viz is embedded
//! or thin. This is the same local-or-remote split the other tabs use.

use std::collections::BTreeMap;
use std::path::PathBuf;
use std::time::Instant;

use eframe::egui::{
    self, Align2, Color32, CornerRadius, FontId, Pos2, Rect, RichText, ScrollArea, Sense, Stroke,
    Vec2,
};

use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::release_events::{
    query_release_events, status, EventSelector, ReleaseEventRow,
};

use super::facett_theme::{Theme, AMBER, GREEN, RED};

/// Where the release events come from (mirrors the other tabs' local/remote split).
enum Src {
    Local(PathBuf),
    /// Thin mode — read `release_events` over the `Viz.ReleaseEvents` RPC. The
    /// `workspace` selects which served workspace's events (the gRPC header).
    Remote { endpoint: String, token: String, workspace: String },
}

/// One "run" — the rows of a single `run_id`, newest run on top.
#[derive(Clone)]
struct RunGroup {
    run_id: String,
    /// `seq`-ordered rows for this run.
    rows: Vec<ReleaseEventRow>,
    /// Latest `ts_micros` in the run — used to sort runs newest-first.
    latest_ts: i64,
}

pub struct ReleaseTabState {
    src: Src,
    loaded: bool,
    error: Option<String>,
    /// All runs found, newest first.
    runs: Vec<RunGroup>,
    /// The selected run_id (the graph + op-log scope). Defaults to newest.
    selected_run: Option<String>,
    /// Pulse phase for the amber "running" glow (advances every frame).
    started_at: Instant,
    /// Active palette — every colour below derives from it so a palette switch
    /// re-skins the whole pane.
    theme: Theme,
}

impl ReleaseTabState {
    pub fn local(root: PathBuf) -> Self {
        Self::with(Src::Local(root))
    }

    pub fn remote(endpoint: String, token: String, workspace: String) -> Self {
        Self::with(Src::Remote { endpoint, token, workspace })
    }

    /// Re-point the thin-mode RPC at a new workspace (the picker switched). No-op
    /// when local; drops the cache so the next draw refetches.
    pub fn set_workspace(&mut self, workspace: String) {
        if let Src::Remote { workspace: w, .. } = &mut self.src {
            *w = workspace;
        }
        self.reload();
    }

    fn with(src: Src) -> Self {
        Self {
            src,
            loaded: false,
            error: None,
            runs: Vec::new(),
            selected_run: None,
            started_at: Instant::now(),
            theme: Theme::default(),
        }
    }

    /// Swap the active palette so the next draw re-skins every colour.
    pub fn set_palette(&mut self, t: Theme) {
        self.theme = t;
    }

    /// Test-only: inject release-op rows directly (no warehouse on disk), marking
    /// the tab loaded so `draw`/`state_json` render the injected DAG. Mirrors
    /// the app's `inject_timeline_for_test` for this surface.
    #[doc(hidden)]
    pub fn inject_for_test(&mut self, rows: Vec<ReleaseEventRow>) {
        self.runs = group_runs(rows);
        self.selected_run = self.runs.first().map(|r| r.run_id.clone());
        self.loaded = true;
        self.error = None;
    }

    /// Re-scope (workspace switch / reload): drop the cache so the next draw reloads.
    pub fn reload(&mut self) {
        self.loaded = false;
        self.error = None;
        self.runs.clear();
        self.selected_run = None;
    }

    /// Read every release event from the warehouse and group by run (local only).
    fn load(&mut self) {
        if self.loaded {
            return;
        }
        self.loaded = true;
        let rows = match &self.src {
            Src::Local(root) => {
                // Read-only + lock-tolerant: coexist with a live server holding
                // the catalog lock (the snapshot-copy fallback) exactly as the
                // `nornir release events` CLI read path does.
                match IcebergWarehouse::open_read_only(root)
                    .and_then(|wh| wh.block_on(query_release_events(&wh, &EventSelector::All)))
                {
                    Ok(rows) => rows,
                    Err(e) => {
                        self.error = Some(format!("{e:#}"));
                        return;
                    }
                }
            }
            Src::Remote { endpoint, token, workspace } => {
                // Thin mode: read the SAME rows over Viz.ReleaseEvents (the server
                // holds the warehouse lock). Decoded into ReleaseEventRow, so the
                // grouping/graph below is source-agnostic.
                match super::remote::fetch_release_events(endpoint, token, workspace) {
                    Ok(rows) => rows,
                    Err(e) => {
                        self.error = Some(format!("{e:#}"));
                        return;
                    }
                }
            }
        };
        self.runs = group_runs(rows);
        self.selected_run = self.runs.first().map(|r| r.run_id.clone());
    }

    /// The currently-selected run (or the newest if none chosen yet).
    fn current(&self) -> Option<&RunGroup> {
        let sel = self.selected_run.as_deref();
        sel.and_then(|id| self.runs.iter().find(|r| r.run_id == id))
            .or_else(|| self.runs.first())
    }

    pub fn draw(&mut self, ui: &mut egui::Ui) {
        let theme = self.theme;
        self.load();

        if let Some(err) = self.error.clone() {
            ui.colored_label(RED, format!("release_events read failed:\n{err}"));
            return;
        }
        if self.runs.is_empty() {
            ui.vertical_centered(|ui| {
                ui.add_space(40.0);
                ui.heading("🚀 Release — no release runs recorded yet");
                ui.label("Every `nornir release run` records its op DAG here:");
                ui.monospace("nornir release run");
                ui.monospace("nornir release events <run-id|repo>   # CLI twin");
            });
            return;
        }

        // ── run picker (newest first) ────────────────────────────────────────
        let runs: Vec<(String, usize, i64)> = self
            .runs
            .iter()
            .map(|r| (r.run_id.clone(), r.rows.len(), r.latest_ts))
            .collect();
        egui::TopBottomPanel::top("release_controls").show_inside(ui, |ui| {
            ui.horizontal_wrapped(|ui| {
                ui.label("run:");
                let sel = self
                    .selected_run
                    .clone()
                    .unwrap_or_else(|| runs.first().map(|r| r.0.clone()).unwrap_or_default());
                egui::ComboBox::from_id_salt("release_run")
                    .selected_text(short_run(&sel))
                    .show_ui(ui, |ui| {
                        for (id, n, _) in &runs {
                            ui.selectable_value(
                                &mut self.selected_run,
                                Some(id.clone()),
                                format!("{} · {n} events", short_run(id)),
                            );
                        }
                    });
                if ui.button("↻ reload").on_hover_text("re-read release_events").clicked() {
                    self.reload();
                }
                ui.separator();
                legend(ui, &theme);
            });
        });

        // Resolve the selected run AFTER the picker may have changed it.
        let Some(run) = self.current().cloned() else { return };

        // Latest status per component for THIS run (drives the graph colours).
        let comp_status = component_status(&run.rows);

        // ── B4: the lit-up dep-graph (top half) ──────────────────────────────
        let graph_h = (ui.available_height() * 0.5).max(180.0);
        egui::TopBottomPanel::top("release_graph")
            .resizable(true)
            .default_height(graph_h)
            .show_inside(ui, |ui| {
                ScrollArea::both().auto_shrink([false, false]).show(ui, |ui| {
                    self.draw_lit_graph(ui, &run, &comp_status);
                });
            });

        // ── B3: the op-log (bottom half) ─────────────────────────────────────
        egui::CentralPanel::default().show_inside(ui, |ui| {
            ui.horizontal(|ui| {
                ui.strong(format!("run {}", short_run(&run.run_id)));
                ui.separator();
                let overall = run_outcome(&run.rows);
                let (col, txt) = match overall {
                    RunOutcome::Running => (AMBER, "● running"),
                    RunOutcome::Ok => (GREEN, "✓ ok"),
                    RunOutcome::Fail => (RED, "✗ failed"),
                };
                ui.colored_label(col, txt);
                ui.separator();
                ui.weak(format!("{} events", run.rows.len()));
            });
            ui.separator();
            ScrollArea::vertical()
                .auto_shrink([false, false])
                .stick_to_bottom(true)
                .show(ui, |ui| {
                    for r in &run.rows {
                        op_log_row(ui, &theme, r);
                    }
                });
        });

        // Keep the amber pulse animating while any op is still running.
        if comp_status.values().any(|s| *s == status::RUNNING) {
            ui.ctx().request_repaint_after(std::time::Duration::from_millis(80));
        }
    }

    /// B4 — the rectangular dep-graph, nodes coloured by latest release status.
    fn draw_lit_graph(
        &self,
        ui: &mut egui::Ui,
        run: &RunGroup,
        comp_status: &BTreeMap<String, String>,
    ) {
        let theme = self.theme;
        // Components (graph nodes) = every component named in this run, ordered
        // deps-first by depends_on (a stable toposort; falls back to first-seen).
        let order = topo_order(run);
        if order.is_empty() {
            ui.label("(no components in this run)");
            return;
        }

        const NODE_W: f32 = 150.0;
        const NODE_H: f32 = 52.0;
        const COL_GAP: f32 = 210.0;
        const LEFT_PAD: f32 = 24.0;
        const TOP_PAD: f32 = 40.0;

        let col_of: BTreeMap<&str, usize> =
            order.iter().enumerate().map(|(i, c)| (c.as_str(), i)).collect();
        let pos_for = |rect: Rect, comp: &str| -> Pos2 {
            let col = *col_of.get(comp).unwrap_or(&0);
            Pos2::new(rect.left() + LEFT_PAD + col as f32 * COL_GAP, rect.top() + TOP_PAD)
        };

        let needed_w = LEFT_PAD + order.len() as f32 * COL_GAP + NODE_W;
        let canvas = Vec2::new(ui.available_width().max(needed_w), (NODE_H + TOP_PAD * 2.0).max(160.0));
        let (rect, _resp) = ui.allocate_exact_size(canvas, Sense::hover());
        let painter = ui.painter_at(rect);
        painter.rect_filled(rect, CornerRadius::ZERO, theme.bg);

        // amber pulse 0..1 for the running glow.
        let pulse = {
            let t = self.started_at.elapsed().as_secs_f32();
            0.5 + 0.5 * (t * 3.0).sin()
        };

        // edges (consumer → producer), drawn under nodes.
        for comp in &order {
            if let Some(deps) = depends_on_of(run, comp) {
                let from = pos_for(rect, comp) + Vec2::new(0.0, NODE_H / 2.0);
                for dep in deps {
                    if !col_of.contains_key(dep.as_str()) {
                        continue;
                    }
                    let to = pos_for(rect, dep) + Vec2::new(NODE_W, NODE_H / 2.0);
                    painter.line_segment(
                        [from, to],
                        Stroke::new(2.0, theme.edge),
                    );
                }
            }
        }

        // nodes.
        for comp in &order {
            let node_pos = pos_for(rect, comp);
            let node_rect = Rect::from_min_size(node_pos, Vec2::new(NODE_W, NODE_H));
            let st = comp_status.get(comp.as_str()).map(String::as_str).unwrap_or("idle");
            let base = lit_color(&theme, st);
            // Running nodes pulse between dim and full amber.
            let fill = if st == status::RUNNING {
                let a = (110.0 + 110.0 * pulse) as u8;
                AMBER.linear_multiply(a as f32 / 255.0)
            } else {
                theme.node_fill
            };
            painter.rect_filled(node_rect, CornerRadius::same(6), fill);
            painter.rect_stroke(
                node_rect,
                CornerRadius::same(6),
                Stroke::new(2.5, base),
                egui::StrokeKind::Inside,
            );
            painter.text(
                node_pos + Vec2::new(NODE_W / 2.0, 12.0),
                Align2::CENTER_TOP,
                comp,
                FontId::proportional(14.0),
                theme.text,
            );
            painter.text(
                node_pos + Vec2::new(NODE_W / 2.0, 30.0),
                Align2::CENTER_TOP,
                st,
                FontId::monospace(11.0),
                base,
            );
        }
    }

    /// Test/introspection hook: the release-op rows + run + per-component status
    /// the tab is rendering, as a JSON value folded into the app's `state_json`.
    pub fn state_json(&self) -> serde_json::Value {
        let current = self.current();
        let rows: Vec<serde_json::Value> = current
            .map(|run| {
                run.rows
                    .iter()
                    .map(|r| {
                        serde_json::json!({
                            "component": r.component,
                            "op": r.op,
                            "phase": r.phase,
                            "status": r.status,
                            "line": op_log_line(r),
                        })
                    })
                    .collect()
            })
            .unwrap_or_default();
        let comp_status = current
            .map(|run| component_status(&run.rows))
            .unwrap_or_default();
        serde_json::json!({
            "source": match &self.src {
                Src::Local(p) => format!("local {}", p.display()),
                Src::Remote { endpoint, workspace, .. } => {
                    format!("remote {endpoint} ws={workspace} (Viz.ReleaseEvents)")
                }
            },
            "error": self.error,
            "runs": self.runs.len(),
            "selected_run": current.map(|r| r.run_id.clone()),
            "rows": rows,
            "component_status": comp_status,
            "palette": self.theme.name,
        })
    }
}

// ─── grouping + status reduction ─────────────────────────────────────────────

/// Group rows into runs (newest first; rows within a run kept in `seq` order —
/// `query_release_events` already sorts by `(run_id, seq)`).
fn group_runs(rows: Vec<ReleaseEventRow>) -> Vec<RunGroup> {
    let mut by_run: BTreeMap<String, Vec<ReleaseEventRow>> = BTreeMap::new();
    for r in rows {
        by_run.entry(r.run_id.clone()).or_default().push(r);
    }
    let mut runs: Vec<RunGroup> = by_run
        .into_iter()
        .map(|(run_id, mut rows)| {
            rows.sort_by_key(|r| r.seq);
            let latest_ts = rows.iter().map(|r| r.ts_micros).max().unwrap_or(0);
            RunGroup { run_id, rows, latest_ts }
        })
        .collect();
    // Newest run on top.
    runs.sort_by(|a, b| b.latest_ts.cmp(&a.latest_ts));
    runs
}

/// Latest status per component within a run, walking rows in `seq` order so the
/// final state wins (a `running` start followed by an `ok` end → `ok`).
fn component_status(rows: &[ReleaseEventRow]) -> BTreeMap<String, String> {
    let mut out: BTreeMap<String, String> = BTreeMap::new();
    for r in rows {
        // Skip the `run/*` envelope rows (no real component lane).
        if r.op == "run" {
            continue;
        }
        out.insert(r.component.clone(), r.status.clone());
    }
    out
}

/// Components in deps-first order: a simple Kahn toposort over the run's
/// `depends_on` edges; falls back to first-seen order on a cycle / missing data.
fn topo_order(run: &RunGroup) -> Vec<String> {
    use std::collections::{BTreeSet, VecDeque};
    // Components that are real ops (not the run envelope).
    let mut comps: Vec<String> = Vec::new();
    let mut seen: BTreeSet<String> = BTreeSet::new();
    for r in &run.rows {
        if r.op == "run" {
            continue;
        }
        if seen.insert(r.component.clone()) {
            comps.push(r.component.clone());
        }
    }
    let set: BTreeSet<&str> = comps.iter().map(String::as_str).collect();
    let mut indeg: BTreeMap<&str, usize> = comps.iter().map(|c| (c.as_str(), 0)).collect();
    let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
    for c in &comps {
        if let Some(deps) = depends_on_of(run, c) {
            for d in deps {
                if set.contains(d.as_str()) {
                    // edge c → d means c consumes d; we want d first.
                    adj.entry(d.as_str()).or_default().push(c.as_str());
                    *indeg.entry(c.as_str()).or_insert(0) += 1;
                }
            }
        }
    }
    let mut q: VecDeque<&str> =
        comps.iter().map(String::as_str).filter(|c| indeg[c] == 0).collect();
    let mut out: Vec<String> = Vec::new();
    while let Some(c) = q.pop_front() {
        out.push(c.to_string());
        if let Some(children) = adj.get(c) {
            for &ch in children {
                let Some(d) = indeg.get_mut(ch) else { continue };
                *d = d.saturating_sub(1);
                if *d == 0 {
                    q.push_back(ch);
                }
            }
        }
    }
    if out.len() == comps.len() {
        out
    } else {
        comps
    }
}

/// The `depends_on` list recorded for a component in this run (the first non-null
/// occurrence — every per-repo row carries the same list).
fn depends_on_of<'a>(run: &'a RunGroup, comp: &str) -> Option<&'a Vec<String>> {
    run.rows
        .iter()
        .find(|r| r.component == comp && r.depends_on.is_some())
        .and_then(|r| r.depends_on.as_ref())
}

enum RunOutcome {
    Running,
    Ok,
    Fail,
}

/// The run's overall outcome: any `fail` → Fail; else any non-`end` op still open
/// → Running; else Ok.
fn run_outcome(rows: &[ReleaseEventRow]) -> RunOutcome {
    if rows.iter().any(|r| r.status == status::FAIL) {
        return RunOutcome::Fail;
    }
    // A run that emitted a `run/end ok` is done.
    let ended_ok = rows
        .iter()
        .any(|r| r.op == "run" && r.phase == "end" && r.status == status::OK);
    if ended_ok {
        RunOutcome::Ok
    } else if rows.iter().any(|r| r.status == status::RUNNING) {
        RunOutcome::Running
    } else {
        RunOutcome::Ok
    }
}

// ─── rendering helpers ───────────────────────────────────────────────────────

/// The op-log line for a row: `[component] op phase status (elapsed) — detail`.
fn op_log_line(r: &ReleaseEventRow) -> String {
    let elapsed = r
        .elapsed_ms
        .map(|ms| format!(" ({:.1}s)", ms as f32 / 1000.0))
        .unwrap_or_default();
    let detail = if r.detail.is_empty() {
        String::new()
    } else {
        format!(" — {}", r.detail)
    };
    format!(
        "[{}] {} {} {}{}{}",
        r.component, r.op, r.phase, r.status, elapsed, detail
    )
}

/// One coloured op-log row (the component-prefixed worklog line).
fn op_log_row(ui: &mut egui::Ui, theme: &Theme, r: &ReleaseEventRow) {
    let (glyph, col) = match r.status.as_str() {
        status::OK => ("✓", GREEN),
        status::FAIL => ("✗", RED),
        status::WARN => ("⚠", AMBER),
        status::RUNNING => ("…", AMBER),
        _ => ("·", theme.text_dim),
    };
    ui.horizontal(|ui| {
        ui.colored_label(col, glyph);
        // `[component]` prefix in a distinct tint so lanes are scannable.
        ui.label(
            RichText::new(format!("[{}]", r.component))
                .monospace()
                .size(12.0)
                .color(theme.text_dim),
        );
        ui.colored_label(col, op_log_line_suffix(r));
    });
}

/// The part of the op-log line after the `[component]` prefix (so the row can
/// tint the prefix separately).
fn op_log_line_suffix(r: &ReleaseEventRow) -> String {
    let elapsed = r
        .elapsed_ms
        .map(|ms| format!(" ({:.1}s)", ms as f32 / 1000.0))
        .unwrap_or_default();
    let detail = if r.detail.is_empty() {
        String::new()
    } else {
        format!(" — {}", r.detail)
    };
    format!("{} {} {}{}{}", r.op, r.phase, r.status, elapsed, detail)
}

/// Node outline / status-text colour for a component's latest status.
fn lit_color(theme: &Theme, st: &str) -> Color32 {
    match st {
        status::OK => GREEN,
        status::FAIL => RED,
        status::WARN => AMBER,
        status::RUNNING => AMBER,
        _ => theme.node_stroke, // idle / dim
    }
}

fn short_run(id: &str) -> String {
    if id.chars().count() > 12 {
        let head: String = id.chars().take(12).collect();
        format!("{head}…")
    } else {
        id.to_string()
    }
}

/// Status legend for the graph + op-log.
fn legend(ui: &mut egui::Ui, theme: &Theme) {
    for (st, label) in [
        (status::RUNNING, "running"),
        (status::OK, "ok"),
        (status::FAIL, "fail"),
        ("idle", "idle"),
    ] {
        ui.colored_label(lit_color(theme, st), "●");
        ui.weak(label);
        ui.add_space(4.0);
    }
}

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

    fn row(
        run: &str,
        seq: i64,
        comp: &str,
        op: &str,
        phase: &str,
        st: &str,
        deps: Option<Vec<&str>>,
        elapsed: Option<i64>,
    ) -> ReleaseEventRow {
        ReleaseEventRow {
            run_id: run.into(),
            seq,
            ts_micros: seq * 10,
            component: comp.into(),
            repo: comp.into(),
            op: op.into(),
            phase: phase.into(),
            status: st.into(),
            detail: String::new(),
            depends_on: deps.map(|v| v.into_iter().map(String::from).collect()),
            elapsed_ms: elapsed,
        }
    }

    #[test]
    fn group_runs_newest_first_and_seq_ordered() {
        let rows = vec![
            row("runA", 1, "znippy", "test", "end", status::OK, None, Some(1200)),
            row("runA", 0, "znippy", "test", "start", status::RUNNING, None, None),
            row("runB", 0, "korp", "test", "start", status::RUNNING, Some(vec![]), None),
        ];
        // runB has the highest ts (seq 0 * 10 = 0) vs runA (max seq 1 → ts 10),
        // so runA is newest. Verify ordering + per-run seq order.
        let runs = group_runs(rows);
        assert_eq!(runs.len(), 2);
        assert_eq!(runs[0].run_id, "runA", "newest (highest ts) run on top");
        assert_eq!(runs[0].rows.iter().map(|r| r.seq).collect::<Vec<_>>(), vec![0, 1]);
    }

    #[test]
    fn component_status_takes_latest() {
        let rows = vec![
            row("r", 0, "znippy", "test", "start", status::RUNNING, None, None),
            row("r", 1, "znippy", "test", "end", status::OK, None, Some(5)),
            row("r", 2, "holger", "gate", "start", status::RUNNING, Some(vec!["znippy"]), None),
            row("r", 3, "r", "run", "end", status::OK, None, None), // envelope skipped
        ];
        let st = component_status(&rows);
        assert_eq!(st.get("znippy").map(String::as_str), Some(status::OK));
        assert_eq!(st.get("holger").map(String::as_str), Some(status::RUNNING));
        assert!(!st.contains_key("r"), "run-envelope component is not a node");
    }

    #[test]
    fn topo_order_is_deps_first() {
        let run = RunGroup {
            run_id: "r".into(),
            latest_ts: 0,
            rows: vec![
                row("r", 0, "holger", "gate", "start", status::RUNNING, Some(vec!["znippy"]), None),
                row("r", 1, "znippy", "test", "start", status::RUNNING, Some(vec![]), None),
            ],
        };
        // holger depends on znippy → znippy must come first.
        let order = topo_order(&run);
        assert_eq!(order, vec!["znippy".to_string(), "holger".to_string()]);
    }

    #[test]
    fn op_log_line_is_component_prefixed() {
        let r = row("r", 0, "znippy", "test", "end", status::OK, None, Some(1200));
        assert_eq!(op_log_line(&r), "[znippy] test end ok (1.2s)");
    }

    #[test]
    fn run_outcome_fail_wins() {
        let rows = vec![
            row("r", 0, "znippy", "test", "end", status::OK, None, Some(5)),
            row("r", 1, "holger", "test", "end", status::FAIL, None, Some(3)),
        ];
        assert!(matches!(run_outcome(&rows), RunOutcome::Fail));
    }
}