nornir 0.4.54

//! 📦 Holger tab — the **built-in (married) holger registry** pane.
//!
//! nornir LINKS `holger-server-lib` IN-PROCESS (the `embed-holger` feature,
//! default on — see [`crate::holger_embed`]). The dev pair it embeds is a
//! crates.io-mirror (`/cache`) + a writable release-rehearsal registry
//! (`/sparring`), both backed by znippy archives under
//! `<warehouse>/holger-cache` (the `nornir cache up` data path).
//!
//! This pane reads that on-disk registry through holger's **functional
//! (in-process) API**, NOT gRPC/HTTP: it constructs the dev-pair `Holger`,
//! `instantiate_backends`, then enumerates each repo's backend with
//! `list(None, limit)` ([`crate::holger_embed::read_registry`]). NO server is
//! started just to read.
//!
//! The optional **"bring up /cache + /sparring"** button is the in-process
//! [`EmbeddedHolger::start`](crate::holger_embed::EmbeddedHolger) — the
//! `nornir cache up` action — run on a worker thread so the egui paint thread
//! never blocks, with a `wait_ready` readiness probe.
//!
//! Gated behind `embed-holger`; this module is only compiled when that feature
//! is on (see `viz/mod.rs`), and the whole 📦 pane degrades to a short note when
//! the feature is off.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use eframe::egui::{self};

use crate::holger_embed::{self, EmbeddedHolger, RepoSummary};

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

/// How many artifact entries to enumerate per repo when summarising (counts +
/// the recent list). The dev-pair archives are small; this is plenty.
const LIST_LIMIT: usize = 100_000;

/// The async "bring up" state machine. The button spawns a worker that starts
/// the embedded dev pair + probes readiness; the slot carries the outcome back.
type BringUpSlot = Arc<Mutex<Option<Result<(), String>>>>;

/// The async registry read. `read_registry` opens znippy archives, so it runs
/// off the paint thread too.
type ReadSlot = Arc<Mutex<Option<Result<Vec<RepoSummary>, String>>>>;

/// What `read_registry` is run against: the real on-disk dev-pair dir (the
/// `cache up` data path), or — in tests — an injected, pre-built summary so the
/// pane can be asserted WITHOUT a live server or a real archive on disk.
enum Source {
    /// `(data_dir, grpc_addr, http_addr)` — the dev-pair config the read uses.
    Disk { data_dir: PathBuf, grpc: String, http: String },
    /// Thin/remote client: there is no local holger to read, so fetch the
    /// registry summary from a running `nornir-server` over `Viz.HolgerRegistry`
    /// (the server runs `read_registry` on its own data dir). Carries the
    /// endpoint/token/workspace the gRPC call needs.
    Remote { endpoint: String, token: String, workspace: String },
    /// Test seam: a fixed set of repo summaries.
    Inject(Vec<RepoSummary>),
}

pub struct HolgerTab {
    source: Source,
    /// The grpc/http the **bring-up** button binds (defaults match `cache up`).
    grpc: String,
    http: String,
    /// Loaded repo summaries (`None` until the first read completes).
    repos: Option<Result<Vec<RepoSummary>, String>>,
    /// In-flight background read (slot + when it started).
    reading: Option<(ReadSlot, Instant)>,
    loaded: bool,
    /// The live embedded dev pair, once the bring-up button started it. Held so
    /// the servers keep running; dropping it stops them.
    embedded: Option<EmbeddedHolger>,
    /// In-flight bring-up (slot + start). The worker returns the readiness
    /// result; on success the next read re-enumerates the now-warm registry.
    bringing_up: Option<(BringUpSlot, Instant)>,
    /// Where the bring-up worker parks the live [`EmbeddedHolger`] handle it
    /// built (its tokio runtime owns the server threads). `poll_bring_up`
    /// adopts it into `embedded` so the servers outlive the worker.
    bring_up_handle_slot: Option<Arc<Mutex<Option<EmbeddedHolger>>>>,
    /// Last bring-up outcome string for the pane + `state_json`.
    bring_up_status: Option<Result<String, String>>,
    theme: Theme,
}

impl HolgerTab {
    /// The live pane: read the dev-pair registry under `data_dir` (the
    /// `nornir cache up` data path = `<warehouse>/holger-cache`).
    pub fn disk(data_dir: PathBuf, grpc: String, http: String) -> Self {
        Self::with(
            Source::Disk { data_dir, grpc: grpc.clone(), http: http.clone() },
            grpc,
            http,
        )
    }

    /// Thin/remote client: fetch the registry summary from a running
    /// `nornir-server` over `Viz.HolgerRegistry` (the server reads its own
    /// holger data dir). No local holger is touched — the pane lists exactly the
    /// repos the server serves (`crates-io` / `cache` / `sparring`). The bring-up
    /// button is disabled (nothing local to start).
    pub fn remote(endpoint: String, token: String, workspace: String) -> Self {
        Self::with(
            Source::Remote { endpoint, token, workspace },
            "127.0.0.1:18443".into(),
            "127.0.0.1:18444".into(),
        )
    }

    /// Test seam: inject a fixed set of repo summaries so `state_json` / `draw`
    /// can be asserted without a live server or an on-disk archive.
    pub fn inject(repos: Vec<RepoSummary>) -> Self {
        Self::with(
            Source::Inject(repos),
            "127.0.0.1:18443".into(),
            "127.0.0.1:18444".into(),
        )
    }

    fn with(source: Source, grpc: String, http: String) -> Self {
        Self {
            source,
            grpc,
            http,
            repos: None,
            reading: None,
            loaded: false,
            embedded: None,
            bringing_up: None,
            bring_up_handle_slot: None,
            bring_up_status: None,
            theme: Theme::default(),
        }
    }

    pub fn set_palette(&mut self, t: Theme) {
        self.theme = t;
    }

    /// The data dir the pane reads/brings up (the `cache up` path), the server
    /// endpoint it fetches from (remote), or `<injected>` (test seam).
    fn data_dir_label(&self) -> String {
        match &self.source {
            Source::Disk { data_dir, .. } => data_dir.display().to_string(),
            Source::Remote { endpoint, .. } => format!("server {endpoint}"),
            Source::Inject(_) => "<injected>".to_string(),
        }
    }

    /// Whether the embedded dev pair is currently brought up in-process by THIS
    /// pane (the bring-up button succeeded and the handle is held).
    fn is_up(&self) -> bool {
        self.embedded.is_some()
    }

    /// Kick off a background registry read (the functional API, no server). Runs
    /// off the paint thread because opening znippy archives touches disk.
    fn start_read(&mut self) {
        self.loaded = true;
        match &self.source {
            Source::Inject(repos) => {
                // No I/O — resolve immediately so the test seam is synchronous.
                self.repos = Some(Ok(repos.clone()));
            }
            Source::Disk { data_dir, grpc, http } => {
                let slot: ReadSlot = Arc::new(Mutex::new(None));
                let sink = slot.clone();
                let (data_dir, grpc, http) =
                    (data_dir.clone(), grpc.clone(), http.clone());
                super::trace::emit_in(
                    "holger.read_registry",
                    &serde_json::json!({ "data_dir": data_dir.display().to_string() }),
                );
                std::thread::spawn(move || {
                    let res = holger_embed::read_registry(&data_dir, &grpc, &http, LIST_LIMIT)
                        .map_err(|e| format!("{e:#}"));
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) = Some(res);
                });
                self.reading = Some((slot, Instant::now()));
            }
            Source::Remote { endpoint, token, workspace } => {
                // Thin mode: fetch the server's registry over Viz.HolgerRegistry
                // off the paint thread (it's a blocking gRPC round-trip).
                let slot: ReadSlot = Arc::new(Mutex::new(None));
                let sink = slot.clone();
                let (endpoint, token, workspace) =
                    (endpoint.clone(), token.clone(), workspace.clone());
                super::trace::emit_in(
                    "holger.fetch_registry",
                    &serde_json::json!({ "endpoint": endpoint, "workspace": workspace }),
                );
                std::thread::spawn(move || {
                    let res =
                        super::remote::fetch_holger_registry(&endpoint, &token, &workspace)
                            .map_err(|e| format!("{e:#}"));
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) = Some(res);
                });
                self.reading = Some((slot, Instant::now()));
            }
        }
    }

    /// Drain a finished background read into `self.repos`.
    fn poll_read(&mut self) {
        if let Some((slot, _)) = &self.reading {
            let done = slot.lock().unwrap_or_else(|p| p.into_inner()).take();
            if let Some(res) = done {
                self.repos = Some(res);
                self.reading = None;
            }
        }
    }

    /// Start the embedded dev pair (`/cache` + `/sparring`) IN-PROCESS on a
    /// worker thread — the same `EmbeddedHolger::start` the `nornir cache up`
    /// CLI runs — then probe `wait_ready`. The egui thread never blocks; the
    /// outcome lands in `bring_up_status` and triggers a re-read.
    fn start_bring_up(&mut self) {
        if self.bringing_up.is_some() || self.is_up() {
            return;
        }
        let Source::Disk { data_dir, .. } = &self.source else {
            // Nothing to bring up against an injected fixture.
            self.bring_up_status = Some(Err("no on-disk registry (injected fixture)".into()));
            return;
        };
        let data_dir = data_dir.clone();
        let (grpc, http) = (self.grpc.clone(), self.http.clone());
        let slot: BringUpSlot = Arc::new(Mutex::new(None));
        let sink = slot.clone();
        super::trace::emit_in(
            "holger.bring_up",
            &serde_json::json!({
                "data_dir": data_dir.display().to_string(),
                "grpc": grpc, "http": http,
            }),
        );
        // The worker BUILDS the EmbeddedHolger (which owns its own tokio runtime
        // + the server threads) and probes readiness. It parks the live handle
        // in a shared slot so the UI thread can ADOPT it into `self.embedded`
        // (keeping the servers alive), and reports the readiness result via the
        // bring-up slot.
        let handle_slot: Arc<Mutex<Option<EmbeddedHolger>>> = Arc::new(Mutex::new(None));
        let handle_sink = handle_slot.clone();
        std::thread::spawn(move || {
            match EmbeddedHolger::start(&data_dir, &grpc, &http) {
                Ok(h) => {
                    let ready = h.wait_ready(Duration::from_secs(20));
                    *handle_sink.lock().unwrap_or_else(|p| p.into_inner()) = Some(h);
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) =
                        Some(ready.map_err(|e| format!("{e:#}")));
                }
                Err(e) => {
                    *sink.lock().unwrap_or_else(|p| p.into_inner()) =
                        Some(Err(format!("{e:#}")));
                }
            }
        });
        self.bring_up_handle_slot = Some(handle_slot);
        self.bringing_up = Some((slot, Instant::now()));
    }

    /// Drain a finished bring-up: adopt the live handle (so the servers stay up)
    /// and record the readiness outcome; on success, re-read the registry.
    fn poll_bring_up(&mut self) {
        let Some((slot, _)) = &self.bringing_up else { return };
        let done = slot.lock().unwrap_or_else(|p| p.into_inner()).take();
        if let Some(res) = done {
            // Adopt the handle the worker built (keeps the runtime/servers alive).
            if let Some(hs) = self.bring_up_handle_slot.take() {
                if let Some(h) = hs.lock().unwrap_or_else(|p| p.into_inner()).take() {
                    self.embedded = Some(h);
                }
            }
            self.bringing_up = None;
            match res {
                Ok(()) => {
                    self.bring_up_status =
                        Some(Ok(format!("/cache + /sparring up at {}", self.http)));
                    // Re-enumerate now that the registry is warm.
                    self.loaded = false;
                }
                Err(e) => self.bring_up_status = Some(Err(e)),
            }
        }
    }

    /// 📦 Holger tab's slice of `state_json` (LAW #6): is the embedded holger
    /// reachable + which data dir, the repositories with per-repo crate/artifact
    /// counts + a few recent ids, and the bring-up state.
    pub fn state_json(&self) -> serde_json::Value {
        let repos = match &self.repos {
            None => serde_json::Value::Null,
            Some(Err(e)) => serde_json::json!({ "error": e }),
            Some(Ok(rs)) => serde_json::json!(
                rs.iter()
                    .map(|r| serde_json::json!({
                        "name": r.name,
                        "repo_type": r.repo_type,
                        "writable": r.writable,
                        "crate_count": r.crate_count,
                        "artifact_count": r.artifact_count,
                        "recent": r.recent,
                    }))
                    .collect::<Vec<_>>()
            ),
        };
        let repo_names: Vec<String> = match &self.repos {
            Some(Ok(rs)) => rs.iter().map(|r| r.name.clone()).collect(),
            _ => Vec::new(),
        };
        // The crates.io-mirror `/cache` + the `/sparring` rehearsal registry are
        // the two writable repos; surface their crate counts at the top level so
        // the headless matrix reads them without walking the repo list.
        let count_for = |name: &str| -> Option<usize> {
            self.repos.as_ref().and_then(|r| r.as_ref().ok()).and_then(|rs| {
                rs.iter().find(|x| x.name == name).map(|x| x.crate_count)
            })
        };
        let bring_up = match &self.bring_up_status {
            None => serde_json::Value::Null,
            Some(Ok(s)) => serde_json::json!({ "ok": true, "detail": s }),
            Some(Err(e)) => serde_json::json!({ "ok": false, "error": e }),
        };
        serde_json::json!({
            // `embed-holger` is what gates this whole pane; when it's on (this
            // module is compiled) the embedded holger is in-process reachable.
            "embedded": true,
            "data_dir": self.data_dir_label(),
            // Whether THIS pane has the dev pair brought up in-process right now.
            "up": self.is_up(),
            "bringing_up": self.bringing_up.is_some(),
            "loaded": self.loaded,
            "repositories": repo_names,
            "repo_count": match &self.repos { Some(Ok(rs)) => rs.len(), _ => 0 },
            "cache_crates": count_for("cache"),
            "sparring_crates": count_for("sparring"),
            "repos": repos,
            "bring_up": bring_up,
            "grpc": self.grpc,
            "http": self.http,
            "palette": self.theme.name,
        })
    }

    pub fn draw(&mut self, ui: &mut egui::Ui) {
        self.poll_read();
        self.poll_bring_up();
        if !self.loaded && self.reading.is_none() {
            self.start_read();
        }

        let theme = self.theme;
        ui.horizontal(|ui| {
            ui.heading("📦 Holger registry");
            if ui.button("⟳ Reload").clicked() {
                self.loaded = false;
            }
            // The "bring up /cache + /sparring" button = the in-process
            // `nornir cache up` action. Disabled while up or while a bring-up is
            // in flight; the worker keeps the paint thread responsive.
            let busy = self.bringing_up.is_some();
            let up = self.is_up();
            let can_bring_up =
                matches!(self.source, Source::Disk { .. }) && !up && !busy;
            if ui
                .add_enabled(
                    can_bring_up,
                    egui::Button::new("▶ bring up /cache + /sparring"),
                )
                .clicked()
            {
                self.start_bring_up();
            }
            if up {
                ui.colored_label(GREEN, "● up");
            } else if busy {
                ui.spinner();
                ui.colored_label(AMBER, "starting…");
            } else {
                ui.colored_label(theme.text_dim, "○ down");
            }
        });
        ui.label(
            "the married, IN-PROCESS holger dev pair — a crates.io mirror (/cache) + a \
             writable release-rehearsal registry (/sparring). Read here via holger's \
             functional API (no gRPC/HTTP), the same backends `nornir cache up` serves.",
        );
        ui.label(format!("data dir: {}", self.data_dir_label()));
        if let Some(res) = &self.bring_up_status {
            match res {
                Ok(s) => ui.colored_label(GREEN, format!("✔ {s}")),
                Err(e) => ui.colored_label(RED, format!("✘ bring-up: {e}")),
            };
        }
        ui.separator();

        match &self.repos {
            None => {
                ui.spinner();
                ui.label("reading the embedded registry…");
            }
            Some(Err(e)) => {
                ui.colored_label(RED, format!("failed to read registry: {e}"));
            }
            Some(Ok(repos)) => {
                if repos.is_empty() {
                    ui.colored_label(theme.text_dim, "no repositories configured.");
                    return;
                }
                egui::Grid::new("holger_repos_grid")
                    .striped(true)
                    .num_columns(4)
                    .spacing([18.0, 4.0])
                    .show(ui, |ui| {
                        ui.strong("repository");
                        ui.strong("type");
                        ui.strong("crates");
                        ui.strong("versions");
                        ui.end_row();
                        for r in repos {
                            ui.horizontal(|ui| {
                                ui.label(format!("/{}", r.name));
                                if r.writable {
                                    ui.colored_label(theme.accent, "rw");
                                } else {
                                    ui.colored_label(theme.text_dim, "ro");
                                }
                            });
                            ui.label(&r.repo_type);
                            ui.label(r.crate_count.to_string());
                            ui.label(r.artifact_count.to_string());
                            ui.end_row();
                        }
                    });
                ui.add_space(8.0);
                // A few recent /sparring publishes (the rehearsal registry) — the
                // most interesting "what did the last rehearsal push?" view.
                if let Some(sparring) = repos.iter().find(|r| r.name == "sparring") {
                    ui.strong("recent /sparring publishes");
                    if sparring.recent.is_empty() {
                        ui.colored_label(theme.text_dim, "none yet — run a release rehearsal.");
                    } else {
                        for id in &sparring.recent {
                            ui.label(format!("  • {id}"));
                        }
                    }
                }
            }
        }
    }
}