holger-ui 0.1.4

//! Live artifact-lineage view-model — the **proxy-cache-filling** facet.
//!
//! holger is a pull-through proxy/cache. As it caches and serves artifacts, this
//! view-model maintains a live **lineage graph** of what the cache holds:
//!
//! * **nodes** — every cached artifact (grouped/clustered by repository +
//!   [`Ecosystem`]), plus the repository that requested it and the upstream it was
//!   proxied from,
//! * **edges** — the lineage: artifact → the repo that requested it
//!   ([`EdgeKind::RequestedBy`]) and artifact → its upstream/proxied source
//!   ([`EdgeKind::ProxiedFrom`]),
//! * **events** — a time-ordered list of arrivals (the "when did the cache fill"
//!   temporal view the helix winds along a coil).
//!
//! The feed is **transport-agnostic**: the model consumes a plain
//! [`LineageEvent`] and knows nothing about how it was produced. Two producers are
//! supported:
//!
//! * **live** — [`LineageView::connect`] attaches an `mpsc::Receiver<LineageEvent>`;
//!   each [`LineageView::pump`] drains newly-arrived events and grows the graph
//!   incrementally (nodes pop in as the cache fills),
//! * **replay** — [`LineageView::replay`] folds an iterator of events (e.g. the
//!   whole history) into the same state.
//!
//! The seam onto holger's real data feed — the append-only Arrow-IPC **audit log**
//! (`server_lib::audit`) — lives in the gated [`audit_tap`] submodule: it maps an
//! `AuditEvent` → [`LineageEvent`], provides [`replay_audit_dir`](LineageView::replay_audit_dir)
//! over a segment directory, and a [`ChannelAuditLog`] tee that a running holger can
//! install as its `Arc<dyn AuditLog>` so every fetch/put streams straight into a
//! connected `LineageView`.
//!
//! There is **no egui here**: like every view in [`crate::data`], the whole state
//! is observable as [`state_json`](LineageView::state_json) for headless robot tests.

use std::collections::{HashMap, HashSet};
use std::sync::mpsc::{Receiver, Sender};

use serde::Serialize;
use serde_json::Value;

/// The package ecosystem an artifact belongs to — the cluster key the 3D fill
/// groups nodes by and the lane the helix winds each arrival onto. Derived from
/// the repository name (and, weakly, the artifact shape) since holger's audit
/// record carries no explicit format column.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Ecosystem {
    Maven,
    Pypi,
    Rust,
    Npm,
    Nuget,
    Go,
    Gem,
    Deb,
    Rpm,
    Helm,
    Docker,
    Conda,
    Composer,
    /// Unclassified / not one of the known package ecosystems.
    Other,
}

impl Ecosystem {
    /// Classify a repository (and artifact) into an ecosystem by substring match on
    /// the repo name — the ecosystem is not an audit column, so it is recovered
    /// from the repo the request targeted (the same repo→backend mapping holger
    /// wires per `znippy-{maven,python,rust,…}` crate), falling back to the artifact
    /// shape and finally [`Ecosystem::Other`].
    pub fn classify(repo: &str, artifact: &str) -> Ecosystem {
        let r = repo.to_ascii_lowercase();
        let has = |needles: &[&str]| needles.iter().any(|n| r.contains(n));
        if has(&["maven", "mvn", "jar", "gradle"]) {
            Ecosystem::Maven
        } else if has(&["pypi", "python", "wheel", "pip"]) {
            Ecosystem::Pypi
        } else if has(&["rust", "crate", "cargo"]) {
            Ecosystem::Rust
        } else if has(&["npm", "node"]) {
            Ecosystem::Npm
        } else if has(&["nuget", "dotnet"]) {
            Ecosystem::Nuget
        } else if has(&["golang", "goproxy", "go-"]) || r == "go" {
            Ecosystem::Go
        } else if has(&["gem", "ruby"]) {
            Ecosystem::Gem
        } else if has(&["deb", "apt", "ubuntu", "debian"]) {
            Ecosystem::Deb
        } else if has(&["rpm", "yum", "dnf", "rocky", "fedora"]) {
            Ecosystem::Rpm
        } else if has(&["helm", "chart"]) {
            Ecosystem::Helm
        } else if has(&["docker", "oci", "image", "registry"]) {
            Ecosystem::Docker
        } else if has(&["conda", "anaconda"]) {
            Ecosystem::Conda
        } else if has(&["composer", "php", "packagist"]) {
            Ecosystem::Composer
        } else if artifact.contains(".crate") {
            Ecosystem::Rust
        } else if artifact.ends_with(".whl") || artifact.ends_with(".tar.gz") {
            Ecosystem::Pypi
        } else if artifact.ends_with(".jar") {
            Ecosystem::Maven
        } else {
            Ecosystem::Other
        }
    }

    /// Stable lowercase label — the helix lane name and the cluster key surfaced in
    /// `state_json`.
    pub fn label(self) -> &'static str {
        match self {
            Ecosystem::Maven => "maven",
            Ecosystem::Pypi => "pypi",
            Ecosystem::Rust => "rust",
            Ecosystem::Npm => "npm",
            Ecosystem::Nuget => "nuget",
            Ecosystem::Go => "go",
            Ecosystem::Gem => "gem",
            Ecosystem::Deb => "deb",
            Ecosystem::Rpm => "rpm",
            Ecosystem::Helm => "helm",
            Ecosystem::Docker => "docker",
            Ecosystem::Conda => "conda",
            Ecosystem::Composer => "composer",
            Ecosystem::Other => "other",
        }
    }

    /// The public upstream host a pull-through proxy of this ecosystem fills from.
    /// Used to synthesize the `proxied-from` source when the audit record does not
    /// carry an explicit upstream (holger's audit log records the local
    /// fetch/put, not the origin — see the module docs). `None` for ecosystems with
    /// no single canonical public registry.
    pub fn upstream_host(self) -> Option<&'static str> {
        Some(match self {
            Ecosystem::Maven => "repo1.maven.org",
            Ecosystem::Pypi => "pypi.org",
            Ecosystem::Rust => "crates.io",
            Ecosystem::Npm => "registry.npmjs.org",
            Ecosystem::Nuget => "nuget.org",
            Ecosystem::Go => "proxy.golang.org",
            Ecosystem::Gem => "rubygems.org",
            Ecosystem::Conda => "conda.anaconda.org",
            Ecosystem::Composer => "packagist.org",
            Ecosystem::Deb | Ecosystem::Rpm | Ecosystem::Helm | Ecosystem::Docker | Ecosystem::Other => {
                return None
            }
        })
    }

    /// A neon RGB base colour for this ecosystem cluster (the egui-free palette the
    /// `gui` layer lifts into a `Color32`).
    pub fn color(self) -> [u8; 3] {
        match self {
            Ecosystem::Maven => [224, 122, 95],
            Ecosystem::Pypi => [75, 150, 220],
            Ecosystem::Rust => [222, 165, 132],
            Ecosystem::Npm => [203, 86, 85],
            Ecosystem::Nuget => [120, 100, 210],
            Ecosystem::Go => [80, 200, 220],
            Ecosystem::Gem => [220, 70, 70],
            Ecosystem::Deb => [200, 70, 110],
            Ecosystem::Rpm => [90, 130, 200],
            Ecosystem::Helm => [70, 170, 200],
            Ecosystem::Docker => [70, 150, 225],
            Ecosystem::Conda => [90, 180, 165],
            Ecosystem::Composer => [150, 165, 205],
            Ecosystem::Other => [160, 160, 172],
        }
    }
}

/// The operation an arrival describes — a transport-agnostic mirror of holger's
/// `audit::AuditAction`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum LineageAction {
    Download,
    Upload,
    Delete,
    List,
    Promote,
    Other,
}

impl LineageAction {
    /// Stable lowercase wire string.
    pub fn as_str(self) -> &'static str {
        match self {
            LineageAction::Download => "download",
            LineageAction::Upload => "upload",
            LineageAction::Delete => "delete",
            LineageAction::List => "list",
            LineageAction::Promote => "promote",
            LineageAction::Other => "other",
        }
    }
}

/// One cache-fill event the [`LineageView`] consumes — the transport-agnostic input
/// the model maps into a node + edges + a time event. It carries everything an
/// `audit::AuditEvent` does, plus an optional explicit `upstream` (a real feed can
/// supply the proxied origin; when `None` the model derives one from the
/// ecosystem).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageEvent {
    /// Event time, nanoseconds since the Unix epoch (UTC).
    pub ts_nanos: i64,
    /// Authenticated principal (or `"anonymous"`).
    pub ident: String,
    /// The operation performed.
    pub action: LineageAction,
    /// Repository the request targeted (the cluster + `requested-by` node).
    pub repo: String,
    /// Artifact identifier (`namespace/name@version` or an archive path).
    pub artifact: String,
    /// Client source address (`ip:port`), if the transport exposed one.
    pub source_ip: String,
    /// Result status (HTTP-like: 200/403/404/…).
    pub status: u16,
    /// Bytes transferred.
    pub bytes: u64,
    /// Explicit proxied-from origin, when a real feed knows it; else the model
    /// derives one from the ecosystem's canonical upstream.
    pub upstream: Option<String>,
}

impl LineageEvent {
    /// Build an arrival with no explicit upstream (the common case — the audit log
    /// records the local fetch/put, and the origin is derived from the ecosystem).
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        ts_nanos: i64,
        ident: impl Into<String>,
        action: LineageAction,
        repo: impl Into<String>,
        artifact: impl Into<String>,
        source_ip: impl Into<String>,
        status: u16,
        bytes: u64,
    ) -> Self {
        Self {
            ts_nanos,
            ident: ident.into(),
            action,
            repo: repo.into(),
            artifact: artifact.into(),
            source_ip: source_ip.into(),
            status,
            bytes,
            upstream: None,
        }
    }

    /// Attach an explicit proxied-from origin (builder style).
    pub fn with_upstream(mut self, upstream: impl Into<String>) -> Self {
        self.upstream = Some(upstream.into());
        self
    }
}

/// What a lineage-graph node represents.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum NodeKind {
    /// A cached artifact.
    Artifact,
    /// A repository that requested artifacts (the `requested-by` anchor).
    Repo,
    /// An upstream / proxied-from source.
    Upstream,
}

/// The lineage an edge encodes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EdgeKind {
    /// artifact → the repository that requested it.
    RequestedBy,
    /// artifact → its upstream / proxied source.
    ProxiedFrom,
}

/// One node in the live lineage graph.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageNode {
    /// Stable id (`art:{repo}/{artifact}`, `repo:{repo}`, or `up:{host}`).
    pub id: String,
    /// Display label.
    pub label: String,
    /// What the node is.
    pub kind: NodeKind,
    /// The ecosystem cluster this node belongs to.
    pub ecosystem: Ecosystem,
    /// The repository the node is associated with (its own name for a repo node).
    pub repo: String,
    /// When the artifact first landed in the cache (nanoseconds UTC); the arrival
    /// time for repo/upstream nodes.
    pub cached_at_nanos: i64,
    /// Artifact size in bytes (max seen); 0 for repo/upstream nodes.
    pub size: u64,
    /// The proxied-from upstream, if known.
    pub upstream: Option<String>,
    /// How many times this node has been touched (cache hits + the initial fill).
    pub hits: u32,
}

/// One directed lineage edge (by node id, so it survives incremental growth).
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct LineageEdge {
    pub from: String,
    pub to: String,
    pub kind: EdgeKind,
}

/// One time-ordered arrival — the temporal event the helix winds onto a per-
/// ecosystem coil.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct TimeEvent {
    pub ts_nanos: i64,
    pub artifact: String,
    pub ecosystem: Ecosystem,
    pub repo: String,
    pub action: LineageAction,
}

/// The live lineage graph state — nodes (deduplicated), edges (deduplicated), and a
/// time-ordered event list, grown by [`ingest`](Self::ingest) from either a live
/// channel ([`connect`](Self::connect) + [`pump`](Self::pump)) or a replay
/// ([`replay`](Self::replay)). Every field is observable via
/// [`state_json`](Self::state_json).
#[derive(Default, Serialize)]
pub struct LineageView {
    /// Cached-artifact / repository / upstream nodes, in first-seen order (so a
    /// renderer sees them "pop in" as the cache fills).
    pub nodes: Vec<LineageNode>,
    /// Lineage edges by node id.
    pub edges: Vec<LineageEdge>,
    /// Time-ordered arrivals.
    pub events: Vec<TimeEvent>,
    /// Total arrivals ingested (including cache hits on already-known artifacts).
    pub total_events: u64,
    /// Total bytes transferred across all arrivals.
    pub total_bytes: u64,

    /// id → index into `nodes`, for O(1) dedup/upsert. Not observable.
    #[serde(skip)]
    node_index: HashMap<String, usize>,
    /// Seen `(from,to)` pairs, for edge dedup. Not observable.
    #[serde(skip)]
    edge_seen: HashSet<(String, String)>,
    /// The live feed, when connected. Drained by [`pump`](Self::pump).
    #[serde(skip)]
    rx: Option<Receiver<LineageEvent>>,
}

/// Convenience: a fresh unbounded lineage feed channel. The `Sender` half is
/// handed to a producer (e.g. a [`ChannelAuditLog`] tee installed on a running
/// holger, or a demo driver); the `Receiver` is handed to
/// [`LineageView::connect`].
pub fn channel() -> (Sender<LineageEvent>, Receiver<LineageEvent>) {
    std::sync::mpsc::channel()
}

impl LineageView {
    /// A fresh, empty lineage view with no feed attached.
    pub fn new() -> Self {
        Self::default()
    }

    /// Attach a live feed. Subsequent [`pump`](Self::pump) calls drain it.
    pub fn connect(&mut self, rx: Receiver<LineageEvent>) {
        self.rx = Some(rx);
    }

    /// `true` once a live feed is attached.
    pub fn is_live(&self) -> bool {
        self.rx.is_some()
    }

    /// Drain all currently-available events from the live feed and fold each into
    /// the graph. Returns how many were ingested. Non-blocking; safe to call every
    /// frame. A disconnected sender is tolerated (the feed simply goes quiet).
    pub fn pump(&mut self) -> usize {
        let drained: Vec<LineageEvent> = match &self.rx {
            Some(rx) => rx.try_iter().collect(),
            None => return 0,
        };
        let n = drained.len();
        for ev in drained {
            self.ingest(ev);
        }
        n
    }

    /// Fold a batch (e.g. the whole audit history) into the graph — the
    /// replay-from-log mode.
    pub fn replay(&mut self, events: impl IntoIterator<Item = LineageEvent>) {
        for ev in events {
            self.ingest(ev);
        }
    }

    /// Number of distinct cached **artifact** nodes (excludes repo/upstream nodes).
    pub fn artifact_count(&self) -> usize {
        self.nodes.iter().filter(|n| n.kind == NodeKind::Artifact).count()
    }

    /// Fold one arrival into the live graph: upsert the artifact node (clustered by
    /// ecosystem), the requesting repo node and the proxied-from upstream node, wire
    /// the two lineage edges, and append the time event. An arrival with no specific
    /// artifact (a bare listing) is counted but adds no node.
    pub fn ingest(&mut self, ev: LineageEvent) {
        self.total_events += 1;
        self.total_bytes = self.total_bytes.saturating_add(ev.bytes);

        if ev.artifact.is_empty() {
            return; // a bare list/other with no artifact — nothing to cache.
        }

        let eco = Ecosystem::classify(&ev.repo, &ev.artifact);
        let upstream = ev
            .upstream
            .clone()
            .or_else(|| eco.upstream_host().map(|h| h.to_string()));

        // The repository node (requested-by anchor).
        let repo_id = format!("repo:{}", ev.repo);
        self.upsert(&repo_id, || LineageNode {
            id: repo_id.clone(),
            label: ev.repo.clone(),
            kind: NodeKind::Repo,
            ecosystem: eco,
            repo: ev.repo.clone(),
            cached_at_nanos: ev.ts_nanos,
            size: 0,
            upstream: None,
            hits: 0,
        });

        // The cached artifact node. `upsert` owns the hit count (1 on the first
        // fill, +1 per later cache hit); here we only refresh size/upstream from the
        // latest arrival.
        let art_id = format!("art:{}/{}", ev.repo, ev.artifact);
        self.upsert(&art_id, || LineageNode {
            id: art_id.clone(),
            label: ev.artifact.clone(),
            kind: NodeKind::Artifact,
            ecosystem: eco,
            repo: ev.repo.clone(),
            cached_at_nanos: ev.ts_nanos,
            size: ev.bytes,
            upstream: upstream.clone(),
            hits: 0,
        });
        if let Some(&i) = self.node_index.get(&art_id) {
            let n = &mut self.nodes[i];
            n.size = n.size.max(ev.bytes);
            if n.upstream.is_none() {
                n.upstream = upstream.clone();
            }
        }

        // artifact → repo (requested-by).
        self.add_edge(&art_id, &repo_id, EdgeKind::RequestedBy);

        // The upstream / proxied-from source node + edge.
        if let Some(host) = &upstream {
            let up_id = format!("up:{host}");
            self.upsert(&up_id, || LineageNode {
                id: up_id.clone(),
                label: host.clone(),
                kind: NodeKind::Upstream,
                ecosystem: eco,
                repo: ev.repo.clone(),
                cached_at_nanos: ev.ts_nanos,
                size: 0,
                upstream: None,
                hits: 0,
            });
            self.add_edge(&art_id, &up_id, EdgeKind::ProxiedFrom);
        }

        self.events.push(TimeEvent {
            ts_nanos: ev.ts_nanos,
            artifact: ev.artifact,
            ecosystem: eco,
            repo: ev.repo,
            action: ev.action,
        });
    }

    /// Insert `id`'s node (from `make`) if unseen; always bump its `hits`. Returns
    /// `true` if a new node was created.
    fn upsert(&mut self, id: &str, make: impl FnOnce() -> LineageNode) -> bool {
        if let Some(&i) = self.node_index.get(id) {
            self.nodes[i].hits = self.nodes[i].hits.saturating_add(1);
            false
        } else {
            let idx = self.nodes.len();
            let mut node = make();
            node.hits = 1;
            self.nodes.push(node);
            self.node_index.insert(id.to_string(), idx);
            true
        }
    }

    /// Add a deduplicated directed edge.
    fn add_edge(&mut self, from: &str, to: &str, kind: EdgeKind) {
        let key = (from.to_string(), to.to_string());
        if self.edge_seen.insert(key) {
            self.edges.push(LineageEdge {
                from: from.to_string(),
                to: to.to_string(),
                kind,
            });
        }
    }

    /// The whole lineage state as observable JSON (the headless robot contract).
    pub fn state_json(&self) -> Value {
        serde_json::to_value(self).unwrap_or(Value::Null)
    }
}

/// A self-contained demo backlog: a realistic multi-ecosystem cache-fill sequence
/// (rust/pypi/maven/npm/go arrivals with a couple of repeat hits) the `gui` layer
/// drips one-per-frame through the LIVE channel so the graph visibly fills and
/// rotates when no real audit feed is wired. Also handy as a test fixture.
pub fn demo_events() -> Vec<LineageEvent> {
    const S: i64 = 1_000_000_000; // one second in nanos
    let base = 1_700_000_000 * S;
    let mk = |i: i64, action, repo: &str, artifact: &str, bytes| {
        LineageEvent::new(base + i * S, "anonymous", action, repo, artifact, "10.0.0.7:5555", 200, bytes)
    };
    use LineageAction::{Download, Upload};
    vec![
        mk(0, Download, "rust-proxy", "serde@1.0.203", 92_160),
        mk(1, Download, "pypi-proxy", "numpy-1.26.4.whl", 5_242_880),
        mk(2, Download, "rust-proxy", "tokio@1.37.0", 720_896),
        mk(3, Download, "maven-central", "org.slf4j/slf4j-api@2.0.12", 61_440),
        mk(4, Download, "npm-proxy", "react@18.2.0", 311_296),
        mk(5, Download, "rust-proxy", "serde@1.0.203", 92_160), // cache hit
        mk(6, Download, "pypi-proxy", "requests-2.31.0.whl", 143_360),
        mk(7, Download, "go-proxy", "golang.org/x/text@v0.14.0", 1_048_576),
        mk(8, Download, "maven-central", "com.google.guava/guava@33.1.0", 2_883_584),
        mk(9, Upload, "rust-dev", "holger-ui@0.1.2", 204_800),
        mk(10, Download, "npm-proxy", "lodash@4.17.21", 552_960),
        mk(11, Download, "rust-proxy", "tokio@1.37.0", 720_896), // cache hit
        mk(12, Download, "pypi-proxy", "pandas-2.2.1.whl", 12_582_912),
        mk(13, Download, "maven-central", "org.junit/junit-jupiter@5.10.2", 122_880),
        mk(14, Download, "go-proxy", "github.com/gin-gonic/gin@v1.9.1", 819_200),
        mk(15, Download, "rust-proxy", "anyhow@1.0.81", 45_056),
    ]
}

// ── audit-log seam ──────────────────────────────────────────────────────────
//
// The adapter onto holger's real feed needs `server_lib::audit`, which is only a
// dependency under `gui` (real build) or as a dev-dependency (`test`). Gated with
// the same `any(feature = "gui", test)` idiom the runner methods in `crate::data`
// use, so the transport-agnostic model above compiles with no server-lib in the
// default build.
#[cfg(any(feature = "gui", test))]
mod audit_tap {
    use super::*;
    use server_lib::audit::{AuditAction, AuditEvent, AuditError, AuditLog, ArrowIpcAuditLog};
    use std::path::Path;

    impl From<AuditAction> for LineageAction {
        fn from(a: AuditAction) -> Self {
            match a {
                AuditAction::Download => LineageAction::Download,
                AuditAction::Upload => LineageAction::Upload,
                AuditAction::Delete => LineageAction::Delete,
                AuditAction::List => LineageAction::List,
                AuditAction::Promote => LineageAction::Promote,
                AuditAction::Other => LineageAction::Other,
            }
        }
    }

    impl From<&AuditEvent> for LineageEvent {
        /// Map one audit record onto a lineage event. Ecosystem + upstream are not
        /// audit columns, so they are derived downstream in [`LineageView::ingest`];
        /// `upstream` is left `None` here (the audit log records the local
        /// fetch/put, not the proxied origin).
        fn from(e: &AuditEvent) -> Self {
            LineageEvent {
                ts_nanos: e.ts_nanos,
                ident: e.ident.clone(),
                action: e.action.into(),
                repo: e.repo.clone(),
                artifact: e.artifact.clone(),
                source_ip: e.source_ip.clone(),
                status: e.status,
                bytes: e.bytes,
                upstream: None,
            }
        }
    }

    impl LineageView {
        /// Replay-from-log mode over holger's audit stream: read every
        /// append-only Arrow-IPC segment under `dir` (oldest-first) and fold each
        /// recorded fetch/put into the lineage graph. Returns the number of events
        /// replayed. A missing directory replays as empty.
        pub fn replay_audit_dir(&mut self, dir: impl AsRef<Path>) -> anyhow::Result<usize> {
            let events = ArrowIpcAuditLog::read_dir(dir)
                .map_err(|e: AuditError| anyhow::anyhow!("audit replay failed: {e}"))?;
            let n = events.len();
            for ae in &events {
                self.ingest(LineageEvent::from(ae));
            }
            Ok(n)
        }
    }

    /// An [`AuditLog`] that **tees** every recorded event into a lineage feed
    /// channel — the live tap. Install it (or a fan-out wrapper) as a running
    /// holger's `Arc<dyn AuditLog>` and hand the paired `Receiver` to
    /// [`LineageView::connect`]; every fetch/put then streams live into the graph.
    /// `record` never fails (a disconnected receiver is dropped silently), matching
    /// holger's fire-and-forget audit contract.
    pub struct ChannelAuditLog {
        tx: Sender<LineageEvent>,
    }

    impl ChannelAuditLog {
        /// Wrap a lineage-feed sender (the `Sender` half of [`super::channel`]).
        pub fn new(tx: Sender<LineageEvent>) -> Self {
            Self { tx }
        }
    }

    impl AuditLog for ChannelAuditLog {
        fn record(&self, event: AuditEvent) -> Result<(), AuditError> {
            let _ = self.tx.send(LineageEvent::from(&event));
            Ok(())
        }
    }
}

#[cfg(any(feature = "gui", test))]
pub use audit_tap::ChannelAuditLog;

#[cfg(test)]
mod tests {
    use super::*;
    use server_lib::audit::{AuditAction, AuditEvent, AuditLog};

    /// Emit one functional-status row into the `nornir test` matrix. Gated behind
    /// `--features testmatrix` so a release build strips it entirely.
    #[cfg(feature = "testmatrix")]
    fn fstatus(component: &str, check: &str, ok: bool, detail: &str) {
        nornir_testmatrix::functional_status(component, check, ok, detail);
    }

    /// An audit `AuditEvent` maps → `LineageEvent`, and `ingest` derives the
    /// ecosystem cluster (from the repo) and the proxied-from upstream (from the
    /// ecosystem) even though neither is an audit column.
    #[test]
    fn audit_event_maps_and_derives_ecosystem_and_upstream() {
        let ae = AuditEvent::new(
            "anonymous",
            AuditAction::Download,
            "rust-proxy",
            "serde@1.0.203",
            "10.0.0.9:4444",
            200,
            92_160,
        );
        let mut view = LineageView::new();
        view.ingest(LineageEvent::from(&ae));

        let art = view
            .nodes
            .iter()
            .find(|n| n.kind == NodeKind::Artifact)
            .expect("artifact node");
        let eco_ok = art.ecosystem == Ecosystem::Rust;
        let up_ok = art.upstream.as_deref() == Some("crates.io");
        assert!(eco_ok, "repo 'rust-proxy' derives the rust ecosystem: {:?}", art.ecosystem);
        assert!(up_ok, "rust upstream derived as crates.io: {:?}", art.upstream);
        // artifact → repo (requested-by) + artifact → upstream (proxied-from).
        assert_eq!(view.edges.len(), 2, "two lineage edges wired");
        assert!(view
            .edges
            .iter()
            .any(|e| e.kind == EdgeKind::RequestedBy && e.to == "repo:rust-proxy"));
        assert!(view
            .edges
            .iter()
            .any(|e| e.kind == EdgeKind::ProxiedFrom && e.to == "up:crates.io"));

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_audit_map_derive",
            eco_ok && up_ok && view.edges.len() == 2,
            &format!(
                "AuditEvent→LineageEvent: eco={} upstream={:?} edges={}",
                art.ecosystem.label(),
                art.upstream,
                view.edges.len()
            ),
        );
    }

    /// The `ChannelAuditLog` tap: install it as a holger `Arc<dyn AuditLog>`, record
    /// events through the SAME `record()` the gRPC fetch/put path calls, and prove
    /// they stream live into a connected `LineageView` via `pump()`.
    #[test]
    fn channel_audit_tap_streams_live_into_view() {
        let (tx, rx) = channel();
        let mut view = LineageView::new();
        view.connect(rx);
        assert!(view.is_live());

        let sink: std::sync::Arc<dyn AuditLog> = std::sync::Arc::new(ChannelAuditLog::new(tx));
        // Two ecosystems + one cache hit (serde downloaded twice).
        sink.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:1", 200, 92_160))
            .unwrap();
        sink.record(AuditEvent::new("anon", AuditAction::Download, "pypi-proxy", "numpy-1.26.4.whl", "ip:2", 200, 5_242_880))
            .unwrap();
        sink.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:3", 200, 92_160))
            .unwrap();

        let drained = view.pump();
        assert_eq!(drained, 3, "pump drained all three recorded events");

        let s = view.state_json();
        assert_eq!(s["total_events"].as_u64(), Some(3), "state_json event tally: {s}");
        // Two distinct artifacts (serde deduped across its two hits) → two artifact nodes.
        assert_eq!(view.artifact_count(), 2, "serde deduped across two hits");
        // The serde artifact recorded a cache hit (>1 touch).
        let serde = view
            .nodes
            .iter()
            .find(|n| n.label == "serde@1.0.203")
            .expect("serde artifact node");
        assert!(serde.hits >= 2, "serde touched twice (fill + cache hit): {}", serde.hits);
        // Grouped by ecosystem: rust + pypi clusters present.
        let ecos: Vec<&str> = view
            .nodes
            .iter()
            .filter(|n| n.kind == NodeKind::Artifact)
            .map(|n| n.ecosystem.label())
            .collect();
        assert!(ecos.contains(&"rust") && ecos.contains(&"pypi"), "both clusters: {ecos:?}");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_channel_tap_live",
            drained == 3 && view.artifact_count() == 2 && serde.hits >= 2,
            &format!(
                "ChannelAuditLog tap → pump: {drained} events, {} artifacts, serde hits {}",
                view.artifact_count(),
                serde.hits
            ),
        );
    }

    /// Replay-from-log: write real audit events through `ArrowIpcAuditLog` into a
    /// segment dir, then `replay_audit_dir` folds them back into a fresh view.
    #[test]
    fn replay_from_audit_dir_round_trips() {
        use server_lib::audit::ArrowIpcAuditLog;
        let dir = std::env::temp_dir().join(format!("holger-lineage-replay-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        {
            let log = ArrowIpcAuditLog::new(&dir).expect("open audit segment");
            log.record(AuditEvent::new("anon", AuditAction::Download, "rust-proxy", "serde@1.0.203", "ip:1", 200, 92_160))
                .unwrap();
            log.record(AuditEvent::new("anon", AuditAction::Upload, "maven-central", "org.slf4j/slf4j-api@2.0.12", "ip:2", 200, 61_440))
                .unwrap();
            // dropped → IPC stream finished.
        }

        let mut view = LineageView::new();
        let n = view.replay_audit_dir(&dir).expect("replay");
        let _ = std::fs::remove_dir_all(&dir);

        assert_eq!(n, 2, "replayed both recorded events");
        assert_eq!(view.artifact_count(), 2, "two cached artifacts after replay");
        assert_eq!(view.events.len(), 2, "two time events after replay");
        let ecos: HashSet<&str> = view
            .nodes
            .iter()
            .filter(|n| n.kind == NodeKind::Artifact)
            .map(|n| n.ecosystem.label())
            .collect();
        assert!(ecos.contains(&"rust") && ecos.contains(&"maven"), "replayed clusters: {ecos:?}");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_replay_audit_dir",
            n == 2 && view.artifact_count() == 2 && view.events.len() == 2,
            &format!("replay_audit_dir: {n} events → {} artifacts", view.artifact_count()),
        );
    }

    /// The demo backlog folds cleanly, dedups repeat hits, and clusters by ecosystem
    /// — the fixture the gui layer drips through the live channel.
    #[test]
    fn demo_events_fill_and_dedup() {
        let mut view = LineageView::new();
        view.replay(demo_events());

        assert_eq!(view.total_events as usize, demo_events().len(), "every demo event counted");
        // serde and tokio each appear twice (cache hits) → fewer artifact nodes than events.
        assert!(view.artifact_count() < demo_events().len(), "repeat hits deduped");
        let clusters: HashSet<&str> = view
            .nodes
            .iter()
            .filter(|n| n.kind == NodeKind::Artifact)
            .map(|n| n.ecosystem.label())
            .collect();
        assert!(clusters.len() >= 4, "multi-ecosystem fill: {clusters:?}");
        // state_json exposes the full picture for the robot.
        let s = view.state_json();
        assert!(s["nodes"].as_array().map(|a| !a.is_empty()).unwrap_or(false));
        assert!(s["edges"].as_array().map(|a| !a.is_empty()).unwrap_or(false));

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_demo_fill",
            view.artifact_count() < demo_events().len() && clusters.len() >= 4,
            &format!("demo fill: {} events → {} artifacts, {} clusters", view.total_events, view.artifact_count(), clusters.len()),
        );
    }

    /// `Ecosystem::classify` recovers the cluster from the repo name across every
    /// known ecosystem, falls back to the artifact shape when the repo name is
    /// unhelpful, and lands on `Other` for the truly unknown — the grouping key the
    /// 3D fill and the helix lanes rely on.
    #[test]
    fn classify_covers_ecosystems_and_falls_back() {
        // Repo-name driven (the primary path — the repo→backend mapping holger wires).
        let by_repo = [
            ("maven-central", Ecosystem::Maven),
            ("pypi-proxy", Ecosystem::Pypi),
            ("rust-crates", Ecosystem::Rust),
            ("npm-proxy", Ecosystem::Npm),
            ("nuget-feed", Ecosystem::Nuget),
            ("go-proxy", Ecosystem::Go),
            ("ruby-gems", Ecosystem::Gem),
            ("debian-apt", Ecosystem::Deb),
            ("rpm-yum", Ecosystem::Rpm),
            ("helm-charts", Ecosystem::Helm),
            ("docker-registry", Ecosystem::Docker),
            ("conda-forge", Ecosystem::Conda),
            ("composer-packagist", Ecosystem::Composer),
        ];
        for (repo, want) in by_repo {
            assert_eq!(Ecosystem::classify(repo, "whatever"), want, "repo {repo:?}");
        }

        // Artifact-shape fallback when the repo name carries no ecosystem hint.
        assert_eq!(Ecosystem::classify("mirror", "foo-0.1.0.crate"), Ecosystem::Rust);
        assert_eq!(Ecosystem::classify("mirror", "numpy-1.26.4.whl"), Ecosystem::Pypi);
        assert_eq!(Ecosystem::classify("mirror", "slf4j-api-2.0.12.jar"), Ecosystem::Maven);
        // Truly unknown → Other. Its label round-trips too.
        let other = Ecosystem::classify("mirror", "readme.txt");
        assert_eq!(other, Ecosystem::Other);
        assert_eq!(other.label(), "other");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_classify_matrix",
            by_repo.iter().all(|(r, w)| Ecosystem::classify(r, "x") == *w)
                && Ecosystem::classify("mirror", "readme.txt") == Ecosystem::Other,
            "classify: 13 ecosystems by repo + artifact-shape fallback + Other",
        );
    }

    /// An explicit `upstream` on the event (a real feed that knows the proxied
    /// origin) overrides the ecosystem-derived default — the proxied-from node and
    /// edge point at the supplied host, not the canonical public registry.
    #[test]
    fn explicit_upstream_overrides_derived() {
        let mut view = LineageView::new();
        // rust would derive crates.io; the feed instead names an internal mirror.
        view.ingest(
            LineageEvent::new(1, "anon", LineageAction::Download, "rust-proxy", "serde@1.0", "ip:1", 200, 10)
                .with_upstream("mirror.corp.internal"),
        );
        let art = view.nodes.iter().find(|n| n.kind == NodeKind::Artifact).expect("artifact");
        assert_eq!(art.upstream.as_deref(), Some("mirror.corp.internal"), "explicit upstream wins");
        assert!(
            view.edges.iter().any(|e| e.kind == EdgeKind::ProxiedFrom && e.to == "up:mirror.corp.internal"),
            "proxied-from edge targets the explicit mirror, not crates.io: {:?}",
            view.edges
        );
        assert!(
            !view.edges.iter().any(|e| e.to == "up:crates.io"),
            "the derived crates.io upstream is not used when one is supplied",
        );

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_explicit_upstream",
            art.upstream.as_deref() == Some("mirror.corp.internal"),
            "explicit LineageEvent.upstream overrides the ecosystem-derived registry",
        );
    }

    /// An ecosystem with no single canonical public registry (helm/docker/…) derives
    /// no upstream: only the requested-by edge is wired, no proxied-from node/edge —
    /// the graph never invents a phantom source.
    #[test]
    fn no_upstream_ecosystem_wires_only_requested_by() {
        let mut view = LineageView::new();
        view.ingest(LineageEvent::new(1, "anon", LineageAction::Download, "helm-charts", "njord-0.1.0.tgz", "ip:1", 200, 4096));

        let art = view.nodes.iter().find(|n| n.kind == NodeKind::Artifact).expect("artifact");
        assert_eq!(art.ecosystem, Ecosystem::Helm);
        assert!(art.upstream.is_none(), "helm has no canonical upstream host");
        assert!(Ecosystem::Helm.upstream_host().is_none());
        // Exactly one edge: artifact → repo. No proxied-from, no upstream node.
        assert_eq!(view.edges.len(), 1, "only the requested-by edge: {:?}", view.edges);
        assert_eq!(view.edges[0].kind, EdgeKind::RequestedBy);
        assert!(!view.nodes.iter().any(|n| n.kind == NodeKind::Upstream), "no phantom upstream node");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_no_upstream",
            art.upstream.is_none() && view.edges.len() == 1 && !view.nodes.iter().any(|n| n.kind == NodeKind::Upstream),
            "helm arrival: requested-by only, no derived proxied-from",
        );
    }

    /// A bare listing (no specific artifact) is tallied into `total_events` /
    /// `total_bytes` but adds no node/edge/time-event — a directory `list` doesn't
    /// fill the cache.
    #[test]
    fn bare_list_event_counts_but_adds_no_node() {
        let mut view = LineageView::new();
        view.ingest(LineageEvent::new(1, "anon", LineageAction::List, "rust-proxy", "", "ip:1", 200, 512));

        assert_eq!(view.total_events, 1, "the list is tallied");
        assert_eq!(view.total_bytes, 512, "its bytes are tallied");
        assert!(view.nodes.is_empty(), "no node for a bare listing");
        assert!(view.edges.is_empty(), "no edge for a bare listing");
        assert!(view.events.is_empty(), "no time-event for a bare listing");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_bare_list",
            view.total_events == 1 && view.nodes.is_empty() && view.events.is_empty(),
            "bare list arrival counted but adds no cache node",
        );
    }

    /// The live feed grows the graph incrementally across successive `pump`s (nodes
    /// "pop in" frame by frame), and the first-seen node order is preserved — the
    /// contract the renderer relies on to animate arrivals landing.
    #[test]
    fn incremental_pump_pops_nodes_in_across_frames() {
        let (tx, rx) = channel();
        let mut view = LineageView::new();
        view.connect(rx);

        // Frame 1: one arrival.
        tx.send(LineageEvent::new(1, "a", LineageAction::Download, "rust-proxy", "serde@1.0", "ip:1", 200, 10)).unwrap();
        assert_eq!(view.pump(), 1, "first pump drains one");
        assert_eq!(view.artifact_count(), 1, "one artifact popped in");

        // Frame 2: nothing new yet → pump is a no-op, state unchanged.
        assert_eq!(view.pump(), 0, "empty pump");
        assert_eq!(view.artifact_count(), 1);

        // Frame 3: two more arrivals in a different cluster.
        tx.send(LineageEvent::new(2, "a", LineageAction::Download, "pypi-proxy", "numpy.whl", "ip:1", 200, 20)).unwrap();
        tx.send(LineageEvent::new(3, "a", LineageAction::Download, "npm-proxy", "react", "ip:1", 200, 30)).unwrap();
        assert_eq!(view.pump(), 2, "third pump drains the two");
        assert_eq!(view.artifact_count(), 3, "graph grew to three artifacts");

        // A dropped sender is tolerated: the feed just goes quiet.
        drop(tx);
        assert_eq!(view.pump(), 0, "disconnected sender pumps empty, no panic");

        // First-seen order preserved: serde's artifact node predates numpy's.
        let art_ids: Vec<&str> = view.nodes.iter().filter(|n| n.kind == NodeKind::Artifact).map(|n| n.label.as_str()).collect();
        assert_eq!(art_ids, ["serde@1.0", "numpy.whl", "react"], "first-seen pop-in order");

        #[cfg(feature = "testmatrix")]
        fstatus(
            "holger-ui",
            "lineage_incremental_pump",
            view.artifact_count() == 3 && art_ids == ["serde@1.0", "numpy.whl", "react"],
            "live feed grows the graph incrementally in first-seen order",
        );
    }
}