nornir 0.5.4

//! nornir's warehouse-side of the cross-repo dependency graph.
//!
//! The compute + query + layout **core now lives in the shared `nornir-depgraph`
//! edda leaf** (so nornir AND dwarves render the same suite dependency view);
//! this module RE-EXPORTS it and keeps the two nornir-only halves that need
//! nornir's own subsystems:
//!   * [`build`] — the `cargo_metadata` builder that resolves a
//!     [`WorkspaceDescriptor`] through nornir's `workspace::resolve` (the leaf's
//!     descriptor-free [`WorkspaceGraph::build_from_members`] is the other path).
//!   * [`record_dep_graph`] / [`query_dep_graph_snapshots`] — persistence of a
//!     [`DepGraphSnapshot`] into the suite warehouse's `dep_graph_edges` Iceberg
//!     table.

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

use anyhow::{anyhow, Result};
use arrow::array::{Array, RecordBatch, StringArray, TimestampMicrosecondArray};
use chrono::Utc;
use iceberg::arrow::schema_to_arrow_schema;
use iceberg::Catalog;
use uuid::Uuid;

pub use nornir_depgraph::{
    inspect_repo, parallel_map, topo_order_from_edges, CrossRepoEdge, DepGraphSnapshot, DepKind,
    RepoFacts, WorkspaceGraph,
};

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

/// Build the cross-repo graph by probing every repo the descriptor resolves with
/// `cargo metadata --no-deps` ([`inspect_repo`]), then assembling the graph via
/// [`WorkspaceGraph::from_facts`].
///
/// Each repo's probe is independent, so they fan across cores through the shared
/// [`parallel_map`] pool (the dependency-Mímir build hot path,
/// `nornir.dep_graph_build`). The pool sizes itself to `available_parallelism()`
/// and honours a positive `RAYON_NUM_THREADS` cap, so the bench's single-core
/// (`_st`) pass collapses to a serial loop and stays a genuine single-core figure
/// while the multi-core arm saturates. `from_facts` gets the same
/// `(name → RepoFacts)` map the old serial loop produced, so the graph is
/// identical. The `cargo metadata`-free twin is [`WorkspaceGraph::build_from_members`].
pub fn build(desc: &WorkspaceDescriptor) -> Result<WorkspaceGraph> {
    let resolved = crate::workspace::resolve::resolve_sources(desc)?;
    let entries: Vec<(String, PathBuf)> = resolved.into_iter().collect();
    let probed = parallel_map(&entries, |(name, root)| {
        inspect_repo(name, root).map(|f| (name.clone(), f))
    });
    let mut facts: BTreeMap<String, RepoFacts> = BTreeMap::new();
    for probe in probed {
        let (name, f) = probe?;
        facts.insert(name, f);
    }
    WorkspaceGraph::from_facts(facts)
}

// ─── persistence (suite warehouse) ───────────────────────────────────────

/// 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();
    let mut kinds: Vec<String> = Vec::new();
    for e in &graph.edges {
        if e.via.is_empty() {
            // Unresolvable/placeholder edge: keep the `from→to` topology and its
            // `kind` label as a SINGLE placeholder row (empty `via_crate`) so the
            // reader reconstructs it as a via-less edge rather than dropping it.
            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(String::new());
            kinds.push(e.kind.as_str().to_string());
            continue;
        }
        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());
            kinds.push(e.kind.as_str().to_string());
        }
    }
    if snapshot_ids.is_empty() {
        // Empty snapshot — still record a no-edges marker.
        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());
        kinds.push(DepKind::Normal.as_str().to_string());
    }

    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 mut 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)),
    ];
    // Only append the `kind` column when the (possibly pre-existing) table schema
    // actually carries it — a warehouse created before kinds were tracked has a
    // 6-column `dep_graph_edges`, and `arrow_schema` mirrors the table.
    if table.metadata().current_schema().field_by_name("kind").is_some() {
        cols.push(Arc::new(StringArray::from(kinds)));
    }
    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. `kind` was added later: try WITH kind, fall back WITHOUT — an old
    // (6-col) snapshot then reads every edge as `Normal` (its prior behavior).
    let filter = skade::ScanFilter::eq("workspace_name", workspace_name);
    let base_cols = [
        "snapshot_id",
        "workspace_name",
        "ts_micros",
        "from_repo",
        "to_repo",
        "via_crate",
    ];
    let mut with_kind: Vec<&str> = base_cols.to_vec();
    with_kind.push("kind");
    let (batches, has_kind): (Vec<RecordBatch>, bool) = match super::iceberg::load_and_read_filtered(
        wh,
        super::iceberg::TABLE_DEP_GRAPH_EDGES,
        &filter,
        &with_kind,
    )
    .await
    {
        Ok(b) => (b, true),
        Err(_) => (
            super::iceberg::load_and_read_filtered(
                wh,
                super::iceberg::TABLE_DEP_GRAPH_EDGES,
                &filter,
                &base_cols,
            )
            .await?,
            false,
        ),
    };

    // (snapshot_id, ts) → (workspace_name, BTreeMap<(from,to), (BTreeSet<via>, kind)>)
    let mut by_snapshot: BTreeMap<
        (Uuid, i64),
        (String, BTreeMap<(String, String), (BTreeSet<String>, DepKind)>),
    > = 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")?;
        let kinds = if has_kind { col::<StringArray>(batch, "kind").ok() } else { None };
        for i in 0..batch.num_rows() {
            // Residual guard: pushdown prunes at file/row-group granularity.
            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();
            // Kind column is optional (null on a row written before kinds) → Normal.
            let kind = kinds
                .filter(|k| !k.is_null(i))
                .map(|k| DepKind::from_str_lenient(k.value(i)))
                .unwrap_or(DepKind::Normal);
            if !f.is_empty() || !t.is_empty() {
                // Init the edge as `Dev` (most restrictive); ANY order-gating row
                // promotes it to `Normal` — the normal edge wins for ordering (#8).
                let slot = entry.1.entry((f, t)).or_insert_with(|| (BTreeSet::new(), DepKind::Dev));
                let via = vias.value(i);
                if !via.is_empty() {
                    slot.0.insert(via.to_string());
                }
                if kind.is_order_gating() {
                    slot.1 = DepKind::Normal;
                }
            }
        }
    }

    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, kind))| CrossRepoEdge { from, to, via, kind })
                .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)
}

/// 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 parity_tests {
    use super::*;
    use crate::workspace::descriptor::{RepoSpec, WorkspaceDescriptor, WorkspaceMeta};

    /// Write a minimal member crate `<root>/<dir>/` with the given package name,
    /// `publish` flag, and path-dep crate names.
    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()
    }

    /// Parity gate: on the SAME on-disk fixture, the `cargo metadata` graph
    /// ([`build`]) and the direct-manifest graph ([`WorkspaceGraph::build_from_members`])
    /// agree on produces + cross-repo edges. Skipped gracefully when `cargo
    /// metadata` can't run (no cargo / offline).
    #[test]
    fn manifest_graph_matches_cargo_metadata_graph() {
        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"];

        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 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");
    }
}

#[cfg(test)]
mod warehouse_roundtrip_tests {
    use super::*;
    use crate::warehouse::iceberg::IcebergWarehouse;

    /// A dep-graph snapshot carrying an UNRESOLVABLE edge — one with an empty
    /// `via` set — round-trips through the warehouse: the placeholder row keeps
    /// the edge's `from→to` topology AND its `kind` label, and the resolved
    /// edges alongside it are unaffected.
    #[test]
    fn placeholder_edge_survives_round_trip_and_keeps_kind_label() {
        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        let edges = vec![
            CrossRepoEdge::normal("app", "core", ["core_c".to_string()].into_iter().collect()),
            CrossRepoEdge {
                from: "app".to_string(),
                to: "ghost".to_string(),
                via: BTreeSet::new(),
                kind: DepKind::Dev,
            },
        ];
        let graph = WorkspaceGraph::from_query_parts(BTreeMap::new(), edges);

        wh.block_on(record_dep_graph(&wh, "ws-placeholder", &graph)).unwrap();
        let snaps = wh.block_on(query_dep_graph_snapshots(&wh, "ws-placeholder", None)).unwrap();
        assert_eq!(snaps.len(), 1, "one snapshot recorded");

        let read: BTreeMap<(String, String), &CrossRepoEdge> = snaps[0]
            .edges
            .iter()
            .map(|e| ((e.from.clone(), e.to.clone()), e))
            .collect();

        let ghost = read
            .get(&("app".to_string(), "ghost".to_string()))
            .expect("placeholder edge app→ghost survived the round-trip");
        assert!(ghost.via.is_empty(), "placeholder edge has no via crate");
        assert_eq!(ghost.kind, DepKind::Dev, "placeholder edge keeps its kind label");

        let resolved = read
            .get(&("app".to_string(), "core".to_string()))
            .expect("resolved edge app→core round-trips");
        assert_eq!(resolved.via.iter().cloned().collect::<Vec<_>>(), vec!["core_c".to_string()]);
        assert_eq!(resolved.kind, DepKind::Normal);
        assert!(resolved.kind.is_order_gating());
    }
}