nornir 0.5.0

//! Cross-repo dependency graph — computation, build-order topological
//! sort, and warehouse persistence.
//!
//! Two phases:
//!
//! 1. **Compute** ([`WorkspaceGraph::build`]) — for every repo, run
//!    `cargo_metadata --no-deps` to learn what it `produces`
//!    (workspace member crate names) and `consumes` (external dep
//!    names). For each pair `(A, B)` add edge `A → B` justified by
//!    `A.consumes ∩ B.produces`. petgraph then gives the build order.
//!
//! 2. **Persist** ([`record_dep_graph`] /
//!    [`query_dep_graph_snapshots`] on [`IcebergWarehouse`]) — every
//!    `record` writes one row per `(snapshot, edge, crate)` into the
//!    long-format `dep_graph_edges` Iceberg table. Reads roll the long
//!    rows back into [`DepGraphSnapshot`] structs grouped by edge.
//!
//! Graphs are immutable historical artefacts: a snapshot belongs to
//! Urðr the moment the workspace's Cargo.tomls have settled.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{anyhow, Context, Result};
use arrow::array::{Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use cargo_metadata::MetadataCommand;
use chrono::{DateTime, Utc};
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::Catalog;
use petgraph::algo::toposort;
use petgraph::graph::{DiGraph, NodeIndex};
use uuid::Uuid;

use super::iceberg::IcebergWarehouse;
use crate::workspace::descriptor::WorkspaceDescriptor;

#[derive(Debug, Clone)]
pub struct RepoFacts {
    pub name: String,
    pub root: PathBuf,
    pub produces: BTreeSet<String>,
    pub consumes: BTreeSet<String>,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CrossRepoEdge {
    pub from: String,
    pub to: String,
    /// Crate names that justify this edge — `from` consumes these,
    /// `to` produces them.
    pub via: BTreeSet<String>,
}

// (serde derives below enable wire transport of dep-graph snapshots for the
//  viz `Viz.Timeline` RPC; the iceberg row mapping is hand-written, not serde.)

#[derive(Debug)]
pub struct WorkspaceGraph {
    pub facts: BTreeMap<String, RepoFacts>,
    pub edges: Vec<CrossRepoEdge>,
    inner: DiGraph<String, usize>,
}

impl WorkspaceGraph {
    /// Build a **query-only** graph straight from `edges` (and optional `facts`),
    /// bypassing `cargo_metadata`. `inner` is left empty, so build-order/topo is
    /// unavailable; the dependency-query methods (`deps_transitive`, `dep_path`,
    /// `dependents_*`, …), which read `edges`, work. Used by `regression::trace`
    /// to rank suspects off a recorded dep-graph snapshot.
    pub fn from_query_parts(facts: BTreeMap<String, RepoFacts>, edges: Vec<CrossRepoEdge>) -> Self {
        Self { facts, edges, inner: DiGraph::new() }
    }

    pub fn build(desc: &WorkspaceDescriptor) -> Result<Self> {
        let resolved = crate::workspace::resolve::resolve_sources(desc)?;
        let mut facts: BTreeMap<String, RepoFacts> = BTreeMap::new();
        for (name, root) in resolved {
            facts.insert(name.clone(), inspect_repo(&name, &root)?);
        }
        Self::from_facts(facts)
    }

    /// Build the cross-repo graph from a member→checkout-path map **without any
    /// `cargo metadata` subprocess** — parse each member's `Cargo.toml`(s)
    /// directly (the `toml` crate), in PARALLEL (tiny files, one worker per core).
    /// Workspace-aware: a member that is itself a cargo workspace contributes
    /// every member crate it declares. A member whose manifest is missing or
    /// unparseable is **SKIPPED** (logged), never fatal to the whole graph — the
    /// fix for one bad manifest (e.g. facett) sinking the entire republish.
    ///
    /// Produces the SAME [`WorkspaceGraph`] structure as [`build`] (same
    /// `produces`/`consumes` facts → same name-intersection edges), so
    /// `affected_set`, `record_dep_graph`, and the disjoint-name check are
    /// unchanged. This is the republish hot path's graph builder.
    pub fn build_from_members(resolved: &BTreeMap<String, PathBuf>) -> Result<Self> {
        let entries: Vec<(String, PathBuf)> =
            resolved.iter().map(|(n, p)| (n.clone(), p.clone())).collect();
        let parsed = parallel_map(&entries, |(name, root)| {
            match inspect_repo_manifests(name, root) {
                Ok(f) => Some(f),
                Err(e) => {
                    eprintln!(
                        "nornir: dep-graph: skipping member `{name}` — manifest parse failed: {e:#}"
                    );
                    None
                }
            }
        });
        let mut facts: BTreeMap<String, RepoFacts> = BTreeMap::new();
        for f in parsed.into_iter().flatten() {
            facts.insert(f.name.clone(), f);
        }
        Self::from_facts(facts)
    }

    /// Assemble the petgraph + cross-repo edges from already-gathered `facts` —
    /// the half shared by [`build`] (cargo-metadata facts) and
    /// [`build_from_members`] (direct-manifest facts). For each pair `(A, B)` an
    /// edge `A → B` is added justified by `A.consumes ∩ B.produces`. The
    /// disjoint-produced-crate-name invariant is enforced here.
    fn from_facts(facts: BTreeMap<String, RepoFacts>) -> Result<Self> {
        // Inverse index: producing-crate-name → owning repo name.
        let mut producer: BTreeMap<&str, &str> = BTreeMap::new();
        for f in facts.values() {
            for c in &f.produces {
                if let Some(prev) = producer.insert(c.as_str(), f.name.as_str()) {
                    if prev != f.name {
                        return Err(anyhow!(
                            "crate `{c}` is produced by both `{prev}` and `{}` — \
                             workspaces must produce disjoint crate names",
                            f.name
                        ));
                    }
                }
            }
        }

        let mut edges: Vec<CrossRepoEdge> = Vec::new();
        let mut inner: DiGraph<String, usize> = DiGraph::new();
        let mut indices: BTreeMap<String, NodeIndex> = BTreeMap::new();
        for name in facts.keys() {
            indices.insert(name.clone(), inner.add_node(name.clone()));
        }

        for from_facts in facts.values() {
            let mut grouped: BTreeMap<&str, BTreeSet<String>> = BTreeMap::new();
            for consumed in &from_facts.consumes {
                if let Some(&owner) = producer.get(consumed.as_str()) {
                    if owner != from_facts.name {
                        grouped.entry(owner).or_default().insert(consumed.clone());
                    }
                }
            }
            for (to_name, via) in grouped {
                let weight = via.len();
                inner.add_edge(indices[&from_facts.name], indices[to_name], weight);
                edges.push(CrossRepoEdge {
                    from: from_facts.name.clone(),
                    to: to_name.to_string(),
                    via,
                });
            }
        }

        Ok(Self { facts, edges, inner })
    }

    /// Topological build order: dependencies first, consumers last.
    /// Errors if the cross-repo graph has a cycle.
    ///
    /// When `inner` (the petgraph built by [`build`](Self::build)) is empty —
    /// the case for a graph rebuilt from the warehouse's `dep_graph_edges` via
    /// [`from_query_parts`](Self::from_query_parts), which never runs
    /// `cargo metadata` — fall back to an edge-based topological sort over
    /// [`component_names`](Self::component_names). The warehouse snapshot carries
    /// the cross-repo edges, so build order is fully recoverable from them; this
    /// is what makes `build_order` non-empty for monitored multi-repo workspaces
    /// (e.g. `nordisk`) where the descriptor/`cargo metadata` path is unavailable.
    pub fn build_order(&self) -> Result<Vec<String>> {
        if self.inner.node_count() == 0 {
            // Edge-based path: deterministic topo over the known components.
            // `topo_order_from_edges` already uses the deps-first convention
            // (it inverts the consumer→producer edges internally).
            let repos = self.component_names();
            return Ok(topo_order_from_edges(&repos, &self.edges));
        }
        // petgraph's toposort returns sources (in-degree 0) first.
        // Our edges point consumer → producer, so the "source" is a
        // top-level consumer; reverse to get deps-first.
        let order = toposort(&self.inner, None).map_err(|cyc| {
            anyhow!(
                "cross-repo dependency cycle detected at node `{}`",
                self.inner[cyc.node_id()]
            )
        })?;
        Ok(order.into_iter().rev().map(|n| self.inner[n].clone()).collect())
    }

    /// Every component (repo) the graph knows about — the union of the `facts`
    /// keys and the edge endpoints. A graph built via [`from_query_parts`] from
    /// a recorded snapshot has empty `facts` but real `edges`, so callers that
    /// need the node set (the funnel planner's component list / validation) must
    /// use this rather than `facts.keys()`. Sorted, deduped, empty names dropped.
    pub fn component_names(&self) -> Vec<String> {
        let mut set: BTreeSet<String> = self.facts.keys().cloned().collect();
        for e in &self.edges {
            if !e.from.is_empty() {
                set.insert(e.from.clone());
            }
            if !e.to.is_empty() {
                set.insert(e.to.clone());
            }
        }
        set.into_iter().collect()
    }

    /// True when `repo` is a known component (in `facts` or as an edge
    /// endpoint). See [`component_names`](Self::component_names).
    pub fn has_component(&self, repo: &str) -> bool {
        self.facts.contains_key(repo)
            || self.edges.iter().any(|e| e.from == repo || e.to == repo)
    }

    pub fn dependencies_of(&self, repo: &str) -> Vec<&CrossRepoEdge> {
        self.edges.iter().filter(|e| e.from == repo).collect()
    }

    /// Edges where `repo` is the *producer* — i.e. the repos that
    /// directly depend on it. Reverse of [`dependencies_of`].
    pub fn dependents_of(&self, repo: &str) -> Vec<&CrossRepoEdge> {
        self.edges.iter().filter(|e| e.to == repo).collect()
    }

    /// Forward transitive closure: every repo `repo` (transitively)
    /// depends on. Excludes `repo` itself.
    pub fn deps_transitive(&self, repo: &str) -> BTreeSet<String> {
        self.reachable(repo, Direction::Forward)
    }

    /// Reverse transitive closure: every repo that (transitively)
    /// depends on `repo` — the **blast radius** of a change to `repo`.
    /// Excludes `repo` itself.
    pub fn dependents_transitive(&self, repo: &str) -> BTreeSet<String> {
        self.reachable(repo, Direction::Reverse)
    }

    /// BFS over cross-repo edges from `start` in the given direction,
    /// returning every reachable repo (excluding `start`).
    fn reachable(&self, start: &str, dir: Direction) -> BTreeSet<String> {
        use std::collections::VecDeque;
        let mut seen: BTreeSet<String> = BTreeSet::new();
        let mut queue: VecDeque<String> = VecDeque::new();
        queue.push_back(start.to_string());
        while let Some(cur) = queue.pop_front() {
            for e in &self.edges {
                let next = match dir {
                    Direction::Forward if e.from == cur => &e.to,
                    Direction::Reverse if e.to == cur => &e.from,
                    _ => continue,
                };
                if seen.insert(next.clone()) {
                    queue.push_back(next.clone());
                }
            }
        }
        seen.remove(start);
        seen
    }

    /// The invalidation set for a set of changed repos: the changed repos
    /// themselves ∪ everything that (transitively) depends on them,
    /// returned in build order (dependencies first). This is exactly the
    /// set a release/bench pipeline must re-run after `changed` moved.
    pub fn affected_by_change(&self, changed: &[String]) -> Vec<String> {
        let mut set: BTreeSet<String> = BTreeSet::new();
        for c in changed {
            set.insert(c.clone());
            set.extend(self.dependents_transitive(c));
        }
        let repos: Vec<String> = set.into_iter().collect();
        topo_order_from_edges(&repos, &self.edges)
    }

    /// Shortest dependency path `from → … → to` (following dependency
    /// edges), or `None` if `to` is not a transitive dependency of
    /// `from`. The returned vec starts with `from` and ends with `to`.
    pub fn dep_path(&self, from: &str, to: &str) -> Option<Vec<String>> {
        use std::collections::VecDeque;
        if from == to {
            return self.facts.contains_key(from).then(|| vec![from.to_string()]);
        }
        let mut parent: BTreeMap<String, String> = BTreeMap::new();
        let mut seen: BTreeSet<String> = BTreeSet::new();
        let mut queue: VecDeque<String> = VecDeque::new();
        seen.insert(from.to_string());
        queue.push_back(from.to_string());
        while let Some(cur) = queue.pop_front() {
            for e in &self.edges {
                if e.from != cur || !seen.insert(e.to.clone()) {
                    continue;
                }
                parent.insert(e.to.clone(), cur.clone());
                if e.to == to {
                    let mut path = vec![to.to_string()];
                    let mut node = to.to_string();
                    while let Some(p) = parent.get(&node) {
                        path.push(p.clone());
                        node = p.clone();
                    }
                    path.reverse();
                    return Some(path);
                }
                queue.push_back(e.to.clone());
            }
        }
        None
    }

    /// Crates `repo` consumes that no repo in the workspace produces —
    /// i.e. genuinely external dependencies (crates.io etc.).
    pub fn external_deps(&self, repo: &str) -> BTreeSet<String> {
        let produced: BTreeSet<&str> = self
            .facts
            .values()
            .flat_map(|f| f.produces.iter().map(String::as_str))
            .collect();
        match self.facts.get(repo) {
            Some(f) => f
                .consumes
                .iter()
                .filter(|c| !produced.contains(c.as_str()))
                .cloned()
                .collect(),
            None => BTreeSet::new(),
        }
    }

    /// Workspace repos whose `consumes` set contains `krate` (sorted).
    /// Used to answer "who uses external crate X?".
    pub fn external_dep_users(&self, krate: &str) -> Vec<String> {
        self.facts
            .values()
            .filter(|f| f.consumes.contains(krate))
            .map(|f| f.name.clone())
            .collect()
    }
}

/// Direction of traversal over the cross-repo dependency edges.
#[derive(Clone, Copy)]
enum Direction {
    /// Follow `from → to` (a repo's dependencies).
    Forward,
    /// Follow `to → from` (a repo's dependents).
    Reverse,
}

fn inspect_repo(name: &str, root: &Path) -> Result<RepoFacts> {
    let meta = MetadataCommand::new()
        .current_dir(root)
        .no_deps()
        .exec()
        .with_context(|| format!("cargo_metadata for repo `{name}` at {}", root.display()))?;
    let mut produces: BTreeSet<String> = BTreeSet::new();
    let mut all_local: BTreeSet<String> = BTreeSet::new();
    let mut all_deps: BTreeSet<String> = BTreeSet::new();
    for p in &meta.packages {
        all_local.insert(p.name.to_string());
        // `publish == Some([])` means `publish = false` — a strictly
        // local crate (typical for `xtask` helpers). Such crates are
        // not visible to any other workspace, so they don't count as
        // something this repo "produces" for cross-repo wiring.
        let is_private = matches!(&p.publish, Some(v) if v.is_empty());
        if !is_private {
            produces.insert(p.name.to_string());
        }
        for d in &p.dependencies {
            all_deps.insert(d.name.clone());
        }
    }
    // Strip *all* in-workspace dep names (published or not) from the
    // consumed set — even private crates can be intra-workspace deps,
    // they just can't be cross-workspace deps.
    let consumes: BTreeSet<String> = all_deps.difference(&all_local).cloned().collect();
    Ok(RepoFacts {
        name: name.to_string(),
        root: root.to_path_buf(),
        produces,
        consumes,
    })
}

// ─── direct-manifest facts (no `cargo metadata` subprocess) ───────────────

/// The `cargo metadata`-free twin of [`inspect_repo`]: parse the member's
/// `Cargo.toml`(s) directly to derive the SAME `produces`/`consumes` sets.
///
///  - **produces** = every non-private `[package].name` (a `publish = false` or
///    `publish = []` crate is local-only, exactly as `inspect_repo` treats it).
///  - **consumes** = every dependency name across all the repo's manifests
///    (normal/dev/build + target- and workspace-scoped tables, honoring a
///    `package = "…"` rename) MINUS the repo's own local crate names.
///
/// Errors only when the repo has no parseable root `Cargo.toml` — the caller
/// turns that into a SKIP, never a graph-wide failure.
fn inspect_repo_manifests(name: &str, root: &Path) -> Result<RepoFacts> {
    let manifests = collect_repo_manifests(root);
    if manifests.is_empty() {
        return Err(anyhow!(
            "no parseable Cargo.toml for repo `{name}` at {}",
            root.display()
        ));
    }
    let mut produces: BTreeSet<String> = BTreeSet::new();
    let mut all_local: BTreeSet<String> = BTreeSet::new();
    let mut all_deps: BTreeSet<String> = BTreeSet::new();
    for doc in &manifests {
        if let Some(pkg) = doc.get("package") {
            if let Some(pname) = pkg.get("name").and_then(|v| v.as_str()) {
                all_local.insert(pname.to_string());
                if !manifest_is_private(pkg) {
                    produces.insert(pname.to_string());
                }
            }
        }
        collect_dep_names(doc, &mut all_deps);
    }
    let consumes: BTreeSet<String> = all_deps.difference(&all_local).cloned().collect();
    Ok(RepoFacts {
        name: name.to_string(),
        root: root.to_path_buf(),
        produces,
        consumes,
    })
}

/// Every manifest belonging to a member repo: its root `Cargo.toml` plus — when
/// the root is a cargo workspace — every `[workspace].members` manifest. A
/// missing/unparseable root yields an empty vec (the caller skips the member).
fn collect_repo_manifests(root: &Path) -> Vec<toml::Value> {
    let root_manifest = root.join("Cargo.toml");
    let Some(root_doc) = read_manifest_toml(&root_manifest) else {
        return Vec::new();
    };
    let mut out: Vec<toml::Value> = Vec::new();
    let mut seen: BTreeSet<PathBuf> = BTreeSet::new();
    seen.insert(root_manifest.clone());
    // The root may be a `[package]`, a virtual `[workspace]`, or both.
    if let Some(members) = root_doc
        .get("workspace")
        .and_then(|w| w.get("members"))
        .and_then(|m| m.as_array())
    {
        for entry in members {
            let Some(pat) = entry.as_str() else { continue };
            for mpath in resolve_member_glob(root, pat) {
                if seen.insert(mpath.clone()) {
                    if let Some(doc) = read_manifest_toml(&mpath) {
                        out.push(doc);
                    }
                }
            }
        }
    }
    out.push(root_doc);
    out
}

/// Expand a `[workspace].members` entry to the manifest path(s) it names. Handles
/// a literal member dir (`crates/foo`) and a single trailing `*` glob (`crates/*`
/// or `*`) — the forms cargo workspaces use in practice. A `**` or mid-path glob
/// is treated literally (best-effort; such a member just won't resolve and is
/// silently skipped, never fatal).
fn resolve_member_glob(root: &Path, pat: &str) -> Vec<PathBuf> {
    let prefix = if pat == "*" {
        Some("")
    } else {
        pat.strip_suffix("/*")
    };
    if let Some(prefix) = prefix {
        let dir = root.join(prefix);
        let mut out = Vec::new();
        if let Ok(rd) = std::fs::read_dir(&dir) {
            for e in rd.flatten() {
                let p = e.path();
                if p.is_dir() {
                    let mf = p.join("Cargo.toml");
                    if mf.is_file() {
                        out.push(mf);
                    }
                }
            }
        }
        out.sort();
        out
    } else {
        vec![root.join(pat).join("Cargo.toml")]
    }
}

/// Read + parse a `Cargo.toml`. `None` on any read/parse error (missing file,
/// torn bytes, invalid TOML) — the caller treats that as "skip this manifest".
fn read_manifest_toml(path: &Path) -> Option<toml::Value> {
    std::fs::read_to_string(path).ok()?.parse::<toml::Value>().ok()
}

/// `publish = false` (bool) or `publish = []` (empty array) ⇒ a strictly local
/// crate — mirrors `inspect_repo`'s `matches!(&p.publish, Some(v) if v.is_empty())`.
fn manifest_is_private(pkg: &toml::Value) -> bool {
    match pkg.get("publish") {
        Some(toml::Value::Boolean(b)) => !b,
        Some(toml::Value::Array(a)) => a.is_empty(),
        _ => false,
    }
}

/// Collect every dependency name declared anywhere in one manifest: the
/// `[dependencies]` / `[dev-dependencies]` / `[build-dependencies]` tables, their
/// `[target.<cfg>.…]` variants, and `[workspace.dependencies]`. A `package = "…"`
/// rename resolves to the REAL crate name (what `cargo_metadata`'s `Dependency.name`
/// reports), so a renamed path-dep still matches its producer's crate name.
fn collect_dep_names(manifest: &toml::Value, out: &mut BTreeSet<String>) {
    fn from_table(t: &toml::Value, out: &mut BTreeSet<String>) {
        let Some(tbl) = t.as_table() else { return };
        for (key, spec) in tbl {
            let real = spec
                .as_table()
                .and_then(|s| s.get("package"))
                .and_then(|p| p.as_str())
                .unwrap_or(key);
            out.insert(real.to_string());
        }
    }
    const KINDS: [&str; 3] = ["dependencies", "dev-dependencies", "build-dependencies"];
    for k in KINDS {
        if let Some(t) = manifest.get(k) {
            from_table(t, out);
        }
    }
    // [workspace.dependencies] — the inherited-dep declarations.
    if let Some(t) = manifest.get("workspace").and_then(|w| w.get("dependencies")) {
        from_table(t, out);
    }
    // [target.<cfg>.dependencies] (and dev/build variants) — platform-scoped deps.
    if let Some(targets) = manifest.get("target").and_then(|t| t.as_table()) {
        for cfg in targets.values() {
            for k in KINDS {
                if let Some(t) = cfg.get(k) {
                    from_table(t, out);
                }
            }
        }
    }
}

/// Minimal parallel map — one worker per core, atomic-cursor self-dispatch (the
/// same rayon-free idiom as the symbol-scan pool), results returned in input
/// order. Used to parse every member's (tiny) `Cargo.toml` concurrently here, and
/// by the monitor to fan the per-member SBOM `cargo metadata` resolve. Top-level
/// only — callers must pass a closure that opens no inner pool.
pub(crate) fn parallel_map<T, R, F>(items: &[T], f: F) -> Vec<R>
where
    T: Sync,
    R: Send,
    F: Fn(&T) -> R + Sync,
{
    use std::sync::atomic::{AtomicUsize, Ordering};
    let n = items.len();
    if n == 0 {
        return Vec::new();
    }
    let workers = std::thread::available_parallelism()
        .map(|w| w.get())
        .unwrap_or(1)
        .min(n);
    if workers <= 1 {
        return items.iter().map(&f).collect();
    }
    let cursor = AtomicUsize::new(0);
    let cursor_ref = &cursor;
    let f_ref = &f;
    let items_ref = items;
    let collected: Vec<Vec<(usize, R)>> = std::thread::scope(|s| {
        (0..workers)
            .map(|_| {
                s.spawn(move || {
                    let mut local: Vec<(usize, R)> = Vec::new();
                    loop {
                        let i = cursor_ref.fetch_add(1, Ordering::Relaxed);
                        if i >= n {
                            break;
                        }
                        local.push((i, f_ref(&items_ref[i])));
                    }
                    local
                })
            })
            .collect::<Vec<_>>()
            .into_iter()
            .map(|h| h.join().unwrap())
            .collect()
    });
    let mut all: Vec<(usize, R)> = collected.into_iter().flatten().collect();
    all.sort_by_key(|(i, _)| *i);
    all.into_iter().map(|(_, r)| r).collect()
}

// ─── persistence ────────────────────────────────────────────────────────

/// One row materialised from the `dep_graph_edges` table per
/// `(snapshot, edge, crate)`. Use [`group_snapshots`] to fold back into
/// [`DepGraphSnapshot`]s.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct DepGraphSnapshot {
    pub snapshot_id: Uuid,
    pub workspace_name: String,
    pub timestamp: DateTime<Utc>,
    pub edges: Vec<CrossRepoEdge>,
}

/// Append a graph snapshot to the warehouse. Returns the snapshot UUID.
/// Edges with no `via` crates (shouldn't happen — guarded by build)
/// are written as a single placeholder row to preserve the edge.
pub async fn record_dep_graph(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    graph: &WorkspaceGraph,
) -> Result<Uuid> {
    let snapshot_id = Uuid::new_v4();
    let ts = Utc::now();
    let id_str = snapshot_id.to_string();

    let mut snapshot_ids = Vec::new();
    let mut ws_names = Vec::new();
    let mut ts_vals: Vec<i64> = Vec::new();
    let mut from_repos = Vec::new();
    let mut to_repos = Vec::new();
    let mut via_crates = Vec::new();
    for e in &graph.edges {
        for via in &e.via {
            snapshot_ids.push(id_str.clone());
            ws_names.push(workspace_name.to_string());
            ts_vals.push(ts.timestamp_micros());
            from_repos.push(e.from.clone());
            to_repos.push(e.to.clone());
            via_crates.push(via.clone());
        }
    }
    if snapshot_ids.is_empty() {
        // Empty snapshot — still record a no-edges marker by writing a
        // single row with empty from/to/via. (Future: a separate
        // snapshots table makes this cleaner.)
        snapshot_ids.push(id_str);
        ws_names.push(workspace_name.to_string());
        ts_vals.push(ts.timestamp_micros());
        from_repos.push(String::new());
        to_repos.push(String::new());
        via_crates.push(String::new());
    }

    let table = wh.catalog()
        .load_table(&wh.table_ident(super::iceberg::TABLE_DEP_GRAPH_EDGES))
        .await?;
    let arrow_schema = Arc::new(schema_to_arrow_schema(table.metadata().current_schema())?);
    let cols: Vec<Arc<dyn Array>> = vec![
        Arc::new(StringArray::from(snapshot_ids)),
        Arc::new(StringArray::from(ws_names)),
        Arc::new(TimestampMicrosecondArray::from(ts_vals).with_timezone("+00:00")),
        Arc::new(StringArray::from(from_repos)),
        Arc::new(StringArray::from(to_repos)),
        Arc::new(StringArray::from(via_crates)),
    ];
    let batch = RecordBatch::try_new(arrow_schema, cols)?;
    super::iceberg::append_batch(wh.catalog(), table, batch).await?;
    Ok(snapshot_id)
}

/// Read every dep-graph snapshot row for a workspace, optionally
/// limited to the most recent N snapshots (after grouping).
pub async fn query_dep_graph_snapshots(
    wh: &IcebergWarehouse,
    workspace_name: &str,
    limit: Option<usize>,
) -> Result<Vec<DepGraphSnapshot>> {
    // Push the workspace filter into the scan and project only the
    // columns we rebuild, so Iceberg can skip non-matching data files /
    // row-groups instead of materialising the whole edge history and
    // filtering in memory. (Same pattern as index::snapshot's pushdown.)
    // Blank/torn Iceberg metadata JSON → 0 rows, never a JSON-EOF crash (load+read).
    let batches: Vec<RecordBatch> = super::iceberg::load_and_read_filtered(
        wh,
        super::iceberg::TABLE_DEP_GRAPH_EDGES,
        &skade::ScanFilter::eq("workspace_name", workspace_name),
        &["snapshot_id", "workspace_name", "ts_micros", "from_repo", "to_repo", "via_crate"],
    )
    .await?;

    // (snapshot_id, ts) → (workspace_name, BTreeMap<(from,to), BTreeSet<via>>)
    let mut by_snapshot: BTreeMap<
        (Uuid, i64),
        (String, BTreeMap<(String, String), BTreeSet<String>>),
    > = BTreeMap::new();

    for batch in &batches {
        let ids = col::<StringArray>(batch, "snapshot_id")?;
        let wss = col::<StringArray>(batch, "workspace_name")?;
        let tss = col::<TimestampMicrosecondArray>(batch, "ts_micros")?;
        let froms = col::<StringArray>(batch, "from_repo")?;
        let tos = col::<StringArray>(batch, "to_repo")?;
        let vias = col::<StringArray>(batch, "via_crate")?;
        for i in 0..batch.num_rows() {
            // Residual guard: pushdown prunes at file/row-group
            // granularity, not per-row, so a matched file may still carry
            // rows for other workspaces.
            if wss.value(i) != workspace_name {
                continue;
            }
            let uid = Uuid::parse_str(ids.value(i))?;
            let key = (uid, tss.value(i));
            let entry = by_snapshot
                .entry(key)
                .or_insert_with(|| (wss.value(i).to_string(), BTreeMap::new()));
            let f = froms.value(i).to_string();
            let t = tos.value(i).to_string();
            if !f.is_empty() || !t.is_empty() {
                entry.1.entry((f, t)).or_default().insert(vias.value(i).to_string());
            }
        }
    }

    let mut out: Vec<DepGraphSnapshot> = by_snapshot
        .into_iter()
        .map(|((snapshot_id, ts_micros), (ws, edge_map))| {
            let edges = edge_map
                .into_iter()
                .map(|((from, to), via)| CrossRepoEdge { from, to, via })
                .collect();
            let timestamp = chrono::TimeZone::timestamp_micros(&Utc, ts_micros)
                .single()
                .unwrap_or_else(Utc::now);
            DepGraphSnapshot { snapshot_id, workspace_name: ws, timestamp, edges }
        })
        .collect();

    out.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
    if let Some(n) = limit {
        let drop_n = out.len().saturating_sub(n);
        out.drain(..drop_n);
    }
    Ok(out)
}

/// Topological order over `repos` derived from `edges`. Edges with
/// endpoints outside `repos` are ignored. Convention: dependencies
/// first, dependents last — matches [`WorkspaceGraph::build_order`].
/// On cycle / partial graph, returns `repos` in input order so callers
/// always get a deterministic permutation.
pub fn topo_order_from_edges(repos: &[String], edges: &[CrossRepoEdge]) -> Vec<String> {
    use std::collections::{BTreeMap, BTreeSet, VecDeque};
    let set: BTreeSet<&str> = repos.iter().map(|s| s.as_str()).collect();
    let mut indeg: BTreeMap<&str, usize> = repos.iter().map(|r| (r.as_str(), 0)).collect();
    let mut adj: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
    for e in edges {
        let from = e.from.as_str();
        let to = e.to.as_str();
        if !set.contains(from) || !set.contains(to) {
            continue;
        }
        adj.entry(to).or_default().push(from);
        *indeg.entry(from).or_insert(0) += 1;
    }
    let mut q: VecDeque<&str> =
        indeg.iter().filter(|(_, d)| **d == 0).map(|(r, _)| *r).collect();
    let mut out: Vec<String> = Vec::with_capacity(repos.len());
    while let Some(r) = q.pop_front() {
        out.push(r.to_string());
        if let Some(children) = adj.get(r) {
            for &c in children {
                let d = indeg.get_mut(c).unwrap();
                *d -= 1;
                if *d == 0 {
                    q.push_back(c);
                }
            }
        }
    }
    if out.len() == repos.len() {
        out
    } else {
        repos.to_vec()
    }
}

/// Downcast a column **by name** — required once a scan uses `.select`,
/// since projection reorders/drops columns and positional access would
/// read the wrong array. Mirrors `index::snapshot::col`.
fn col<'a, T: 'static>(batch: &'a RecordBatch, name: &str) -> Result<&'a T> {
    batch
        .column_by_name(name)
        .ok_or_else(|| anyhow!("projected batch missing column `{name}`"))?
        .as_any()
        .downcast_ref::<T>()
        .ok_or_else(|| anyhow!("column `{name}` has unexpected arrow type"))
}

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

    /// Write a minimal member crate `<root>/<name>/` with the given package name,
    /// `publish` flag, and path-dep crate names (each `dep` becomes
    /// `dep = { path = "../dep" }`).
    fn member(root: &Path, dir: &str, pkg: &str, publish_false: bool, path_deps: &[&str]) {
        let crate_dir = root.join(dir);
        std::fs::create_dir_all(crate_dir.join("src")).unwrap();
        let mut toml = format!(
            "[package]\nname = \"{pkg}\"\nversion = \"0.1.0\"\nedition = \"2021\"\n"
        );
        if publish_false {
            toml.push_str("publish = false\n");
        }
        if !path_deps.is_empty() {
            toml.push_str("\n[dependencies]\n");
            for d in path_deps {
                toml.push_str(&format!("{d} = {{ path = \"../{d}\" }}\n"));
            }
        }
        std::fs::write(crate_dir.join("Cargo.toml"), toml).unwrap();
        std::fs::write(crate_dir.join("src/lib.rs"), "// fixture\n").unwrap();
    }

    fn resolved(root: &Path, members: &[&str]) -> BTreeMap<String, PathBuf> {
        members
            .iter()
            .map(|m| (m.to_string(), root.join(m)))
            .collect()
    }

    fn edge_set(g: &WorkspaceGraph) -> BTreeSet<(String, String, Vec<String>)> {
        g.edges
            .iter()
            .map(|e| {
                (
                    e.from.clone(),
                    e.to.clone(),
                    e.via.iter().cloned().collect::<Vec<_>>(),
                )
            })
            .collect()
    }

    /// Direct-manifest parse reproduces the cross-repo edges + produces facts of a
    /// path-dep diamond, with ZERO `cargo metadata` subprocess.
    #[test]
    fn manifest_facts_capture_diamond_edges() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        // app → liba → util ; app → libb → util  (consumer → producer edges).
        member(root, "app", "app", false, &["liba", "libb"]);
        member(root, "liba", "liba", false, &["util"]);
        member(root, "libb", "libb", false, &["util"]);
        member(root, "util", "util", false, &[]);

        let g = WorkspaceGraph::build_from_members(&resolved(
            root,
            &["app", "liba", "libb", "util"],
        ))
        .expect("manifest graph builds");

        // produces: each repo produces exactly its own crate.
        for m in ["app", "liba", "libb", "util"] {
            let f = g.facts.get(m).unwrap();
            assert!(f.produces.contains(m), "{m} must produce its own crate");
        }
        // Edges follow the path-deps.
        let edges = edge_set(&g);
        assert!(edges.contains(&("app".into(), "liba".into(), vec!["liba".into()])));
        assert!(edges.contains(&("app".into(), "libb".into(), vec!["libb".into()])));
        assert!(edges.contains(&("liba".into(), "util".into(), vec!["util".into()])));
        assert!(edges.contains(&("libb".into(), "util".into(), vec!["util".into()])));
        // Blast radius of a util change is the whole diamond, in build order.
        let affected = g.affected_by_change(&["util".to_string()]);
        assert_eq!(
            affected.iter().cloned().collect::<BTreeSet<_>>(),
            ["app", "liba", "libb", "util"].iter().map(|s| s.to_string()).collect()
        );
    }

    /// A `publish = false` member is NOT a producer (local-only), so nothing
    /// depends on it cross-repo — mirrors `inspect_repo`'s private-crate rule.
    #[test]
    fn private_member_produces_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        member(root, "app", "app", false, &["xtask"]);
        member(root, "xtask", "xtask", true, &[]); // publish = false → private
        let g = WorkspaceGraph::build_from_members(&resolved(root, &["app", "xtask"])).unwrap();
        assert!(g.facts.get("xtask").unwrap().produces.is_empty(), "private crate produces nothing");
        // `app` consumes `xtask` but no repo PRODUCES it → no cross-repo edge.
        assert!(g.edges.iter().all(|e| e.to != "xtask"), "no edge to a private producer");
    }

    /// A broken/unparseable manifest in ONE member is SKIPPED (logged), and the
    /// rest of the graph still builds — the fix for one bad manifest (facett)
    /// sinking the entire republish dep-graph.
    #[test]
    fn broken_manifest_member_is_skipped_not_fatal() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        member(root, "app", "app", false, &["util"]);
        member(root, "util", "util", false, &[]);
        // A third member with a torn Cargo.toml (invalid TOML).
        let bad = root.join("facett");
        std::fs::create_dir_all(&bad).unwrap();
        std::fs::write(bad.join("Cargo.toml"), "[package\nname = busted = =\n").unwrap();

        let g = WorkspaceGraph::build_from_members(&resolved(root, &["app", "util", "facett"]))
            .expect("a broken member must not fail the whole graph");
        // The good members are present + wired; the broken one is simply absent.
        assert!(g.facts.contains_key("app"));
        assert!(g.facts.contains_key("util"));
        assert!(!g.facts.contains_key("facett"), "broken member skipped from facts");
        assert!(g.edges.iter().any(|e| e.from == "app" && e.to == "util"));
    }

    /// Parity gate: on the SAME on-disk fixture, the direct-manifest graph and the
    /// legacy `cargo metadata` graph ([`WorkspaceGraph::build`]) agree on produces
    /// + cross-repo edges. Skipped gracefully when `cargo metadata` can't run
    /// (no cargo / offline), so it's a real parity check where possible and inert
    /// otherwise.
    #[test]
    fn manifest_graph_matches_cargo_metadata_graph() {
        use crate::workspace::descriptor::{RepoSpec, WorkspaceDescriptor, WorkspaceMeta};

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        member(root, "app", "app", false, &["liba"]);
        member(root, "liba", "liba", false, &["util"]);
        member(root, "util", "util", false, &[]);
        let names = ["app", "liba", "util"];

        // Legacy path: synthesize a path-based descriptor and run `cargo metadata`.
        let mut repos = BTreeMap::new();
        for m in names {
            repos.insert(
                m.to_string(),
                RepoSpec {
                    path: Some(root.join(m).to_string_lossy().into_owned()),
                    git: None,
                    branch: None,
                },
            );
        }
        let desc = WorkspaceDescriptor {
            workspace: WorkspaceMeta { name: "fixture".into(), deep_scan: false },
            repos,
            descriptor_dir: root.to_path_buf(),
        };
        let meta_graph = match WorkspaceGraph::build(&desc) {
            Ok(g) => g,
            Err(e) => {
                eprintln!("skip parity: cargo metadata unavailable: {e:#}");
                return;
            }
        };

        let manifest_graph =
            WorkspaceGraph::build_from_members(&resolved(root, &names)).unwrap();

        assert_eq!(
            manifest_graph.component_names(),
            meta_graph.component_names(),
            "same component set"
        );
        for m in names {
            assert_eq!(
                manifest_graph.facts.get(m).map(|f| &f.produces),
                meta_graph.facts.get(m).map(|f| &f.produces),
                "produces parity for {m}"
            );
        }
        assert_eq!(
            edge_set(&manifest_graph),
            edge_set(&meta_graph),
            "cross-repo edge parity (manifest vs cargo metadata)"
        );
    }
}

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

    /// Build a `WorkspaceGraph` directly from facts + edges, bypassing
    /// `cargo_metadata`. `inner` is left empty: the Mímir methods under
    /// test work purely off `facts`/`edges`.
    fn graph(facts: Vec<RepoFacts>, edges: Vec<CrossRepoEdge>) -> WorkspaceGraph {
        let mut fmap = BTreeMap::new();
        for f in facts {
            fmap.insert(f.name.clone(), f);
        }
        WorkspaceGraph { facts: fmap, edges, inner: DiGraph::new() }
    }

    fn facts(name: &str, produces: &[&str], consumes: &[&str]) -> RepoFacts {
        RepoFacts {
            name: name.to_string(),
            root: PathBuf::from("/dev/null"),
            produces: produces.iter().map(|s| s.to_string()).collect(),
            consumes: consumes.iter().map(|s| s.to_string()).collect(),
        }
    }

    fn edge(from: &str, to: &str, via: &[&str]) -> CrossRepoEdge {
        CrossRepoEdge {
            from: from.to_string(),
            to: to.to_string(),
            via: via.iter().map(|s| s.to_string()).collect(),
        }
    }

    /// Diamond:  app → liba → util,  app → libb → util.
    /// `app` also consumes external `serde`; `util` consumes external `libc`.
    fn diamond() -> WorkspaceGraph {
        graph(
            vec![
                facts("app", &["app_c"], &["a_c", "b_c", "serde"]),
                facts("liba", &["a_c"], &["util_c"]),
                facts("libb", &["b_c"], &["util_c"]),
                facts("util", &["util_c"], &["libc"]),
            ],
            vec![
                edge("app", "liba", &["a_c"]),
                edge("app", "libb", &["b_c"]),
                edge("liba", "util", &["util_c"]),
                edge("libb", "util", &["util_c"]),
            ],
        )
    }

    fn names(edges: Vec<&CrossRepoEdge>, pick_to: bool) -> BTreeSet<String> {
        edges
            .into_iter()
            .map(|e| if pick_to { e.to.clone() } else { e.from.clone() })
            .collect()
    }

    #[test]
    fn dependents_of_is_reverse_of_dependencies() {
        let g = diamond();
        // who depends directly on util? liba, libb.
        assert_eq!(
            names(g.dependents_of("util"), false),
            ["liba", "libb"].iter().map(|s| s.to_string()).collect()
        );
        // app depends directly on liba, libb.
        assert_eq!(
            names(g.dependencies_of("app"), true),
            ["liba", "libb"].iter().map(|s| s.to_string()).collect()
        );
        assert!(g.dependents_of("app").is_empty());
    }

    #[test]
    fn transitive_closures() {
        let g = diamond();
        assert_eq!(
            g.deps_transitive("app"),
            ["liba", "libb", "util"].iter().map(|s| s.to_string()).collect()
        );
        // blast radius of util = everything above it.
        assert_eq!(
            g.dependents_transitive("util"),
            ["app", "liba", "libb"].iter().map(|s| s.to_string()).collect()
        );
        assert!(g.deps_transitive("util").is_empty());
        assert!(g.dependents_transitive("app").is_empty());
    }

    #[test]
    fn affected_by_change_is_blast_radius_in_build_order() {
        let g = diamond();
        let affected = g.affected_by_change(&["util".to_string()]);
        assert_eq!(
            affected.iter().cloned().collect::<BTreeSet<_>>(),
            ["app", "liba", "libb", "util"].iter().map(|s| s.to_string()).collect()
        );
        // build order: dependencies first. util before liba/libb; both before app.
        let pos = |n: &str| affected.iter().position(|x| x == n).unwrap();
        assert!(pos("util") < pos("liba"));
        assert!(pos("util") < pos("libb"));
        assert!(pos("liba") < pos("app"));
        assert!(pos("libb") < pos("app"));
    }

    #[test]
    fn dep_path_finds_shortest_route() {
        let g = diamond();
        let p = g.dep_path("app", "util").expect("path exists");
        assert_eq!(p.len(), 3);
        assert_eq!(p.first().unwrap(), "app");
        assert_eq!(p.last().unwrap(), "util");
        assert_eq!(g.dep_path("app", "app"), Some(vec!["app".to_string()]));
        // util doesn't depend on app — no forward path.
        assert_eq!(g.dep_path("util", "app"), None);
        assert_eq!(g.dep_path("app", "ghost"), None);
    }

    #[test]
    fn external_deps_and_users() {
        let g = diamond();
        assert_eq!(
            g.external_deps("app"),
            ["serde"].iter().map(|s| s.to_string()).collect()
        );
        assert_eq!(
            g.external_deps("util"),
            ["libc"].iter().map(|s| s.to_string()).collect()
        );
        // liba's only consumed crate (util_c) is produced internally.
        assert!(g.external_deps("liba").is_empty());
        assert_eq!(g.external_dep_users("serde"), vec!["app".to_string()]);
        assert_eq!(g.external_dep_users("libc"), vec!["util".to_string()]);
        // util_c is consumed by both liba and libb (sorted by repo name).
        assert_eq!(
            g.external_dep_users("util_c"),
            vec!["liba".to_string(), "libb".to_string()]
        );
    }
}