nornir 0.5.0

//! Server-monitored self-sync (server-user `nornir`): the poll loop and the
//! per-workspace fetch step.
//!
//! The loop runs inside `nornir-server`: every tick it fetches each *monitored*
//! workspace's git members (pure-Rust HTTPS via [`crate::gitio`]), records sync
//! state in the [`crate::registry`], and — when any member's HEAD moved —
//! invokes the republish hook. [`fetch_workspace`] is shared with the CLI's
//! `nornir workspace fetch`.

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

use anyhow::{anyhow, bail, Context, Result};

use crate::registry::{Mode, Registry};

/// One member's clone/fetch outcome — the durable, remotely-readable twin of the
/// per-member `eprintln!("…fetch error…")` that used to be the only signal. Every
/// fetched member produces exactly one of these (ok OR error), so a thin viz/CLI
/// client can see *why* a member didn't populate (auth, vanished repo, network),
/// not just that its data is missing. Persisted to the `clone_events` warehouse
/// table via [`crate::warehouse::clone_events::record_fetch_report`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberOutcome {
    /// Workspace member name.
    pub member: String,
    /// The git remote that was cloned/fetched (empty for local members — but those
    /// are skipped, so an outcome always carries a real remote).
    pub remote: String,
    /// `clone` | `fetch` (today both go through `clone_or_fetch`; recorded as
    /// `clone-fetch` so the op string matches the error chain users already saw).
    pub op: String,
    /// `ok` | `error`.
    pub status: String,
    /// The error chain (`{e:#}`) on failure; the resolved SHA on success.
    pub detail: String,
    /// Wall-clock duration of this member's clone/fetch, milliseconds.
    pub elapsed_ms: i64,
}

/// Outcome of fetching one workspace's members.
#[derive(Debug, Default)]
pub struct FetchReport {
    pub workspace: String,
    /// Members with a remote that were clone/fetched.
    pub fetched: usize,
    /// Members whose HEAD moved since last sync.
    pub changed: Vec<String>,
    /// `(member, error)` for members that failed.
    pub errors: Vec<(String, String)>,
    /// One outcome per fetched member (ok or error) — the readable populate-status
    /// stream. Same set as `fetched`, but with op/status/detail/elapsed so the
    /// failure (or success) is first-class and persistable, not log-only.
    pub outcomes: Vec<MemberOutcome>,
}

impl FetchReport {
    pub fn changed_any(&self) -> bool {
        !self.changed.is_empty()
    }
}

/// Clone/fetch every git member of `name` into `<root>/<name>/git/<member>`,
/// write back each member's sync state, and return what changed. Shared by the
/// CLI verb and the poll loop. HTTPS only (SSH refused; see [`crate::gitio`]).
pub fn fetch_workspace(reg: &Registry, root: &Path, name: &str) -> Result<FetchReport> {
    fetch_workspace_with_progress(reg, root, name, |_| {})
}

/// Like [`fetch_workspace`], but invokes `on_member` with each member's
/// [`MemberOutcome`] **as that member's clone/fetch completes** — the streaming
/// seam. The server uses it to emit + finish a per-member *child* job (under the
/// populate parent) the instant each clone lands, so the Jobs panel shows members
/// appear one at a time instead of all-at-once at the end. `on_member` is
/// best-effort progress reporting; it must not fail the fetch (it returns `()`).
pub fn fetch_workspace_with_progress(
    reg: &Registry,
    root: &Path,
    name: &str,
    mut on_member: impl FnMut(&MemberOutcome),
) -> Result<FetchReport> {
    let mut ws = reg
        .get(name)?
        .ok_or_else(|| anyhow!("no workspace `{name}` in the registry"))?;
    let ws_root = root.join(&ws.name);
    let git_dir = ws_root.join("git");
    std::fs::create_dir_all(&git_dir)?;
    std::fs::create_dir_all(ws_root.join("builds"))?;

    let now = chrono::Utc::now().to_rfc3339();
    let mut report = FetchReport {
        workspace: ws.name.clone(),
        ..Default::default()
    };
    for m in &mut ws.members {
        if m.remote.is_empty() {
            continue; // local/external member — source lives outside
        }
        report.fetched += 1;
        let dest = git_dir.join(&m.name);
        // SSH remotes authenticate with the server's deploy key (resolved inside
        // clone_or_fetch via NORNIR_SSH_DIR / the `nornir` user's home).
        let t0 = std::time::Instant::now();
        let outcome = crate::gitio::clone_or_fetch(&m.remote, &dest, None);
        let elapsed_ms = t0.elapsed().as_millis() as i64;
        match outcome {
            Ok(sha) => {
                let mo = MemberOutcome {
                    member: m.name.clone(),
                    remote: m.remote.clone(),
                    op: "clone-fetch".into(),
                    status: "ok".into(),
                    detail: sha.clone(),
                    elapsed_ms,
                };
                on_member(&mo); // STREAM: this member's clone landed (ok).
                report.outcomes.push(mo);
                if sha != m.last_seen_sha {
                    report.changed.push(m.name.clone());
                }
                m.last_seen_sha = sha;
                m.last_synced = now.clone();
                m.sync_state = "ok".into();
                // AUT6 freshness: record the clone's working-tree digest so a
                // thin client can detect staleness the SHA can't see. A freshly
                // fetched server clone is normally clean; best-effort (a digest
                // error must not fail the fetch).
                match crate::gitio::worktree_freshness(&dest) {
                    Ok(f) => {
                        m.worktree_digest = f.digest;
                        m.worktree_dirty = f.dirty;
                    }
                    Err(e) => eprintln!(
                        "nornir-monitor: {}::{} worktree freshness skipped: {e:#}",
                        report.workspace, m.name
                    ),
                }
            }
            Err(e) => {
                let msg = format!("{e:#}");
                m.sync_state = format!("error: {msg}");
                let mo = MemberOutcome {
                    member: m.name.clone(),
                    remote: m.remote.clone(),
                    op: "clone-fetch".into(),
                    status: "error".into(),
                    detail: msg.clone(),
                    elapsed_ms,
                };
                on_member(&mo); // STREAM: this member's clone landed (error).
                report.outcomes.push(mo);
                report.errors.push((m.name.clone(), msg));
            }
        }
    }
    ws.updated_at = now;
    reg.upsert(&ws)?;
    Ok(report)
}

/// One sweep: fetch every monitored workspace; republish those that changed.
/// Per-member errors are isolated (logged, not fatal) so one bad remote never
/// stalls the others or aborts the sweep. Takes the server's shared registry
/// handle (redb is single-writer — the poll loop and the Workspaces RPC must
/// share one `Registry`, never each open their own).
pub fn sync_once(reg: &Registry, root: &Path) -> Result<()> {
    for ws in reg.list()? {
        if ws.mode != Mode::Monitored {
            continue;
        }
        match fetch_workspace(&reg, root, &ws.name) {
            Ok(rep) => {
                for (m, e) in &rep.errors {
                    eprintln!("nornir-monitor: {}::{m} fetch error: {e}", rep.workspace);
                }
                // Republish on a git change OR when never built (empty
                // `current_snapshot` — e.g. the warehouse was wiped) so a clean
                // warehouse rebuilds from git instead of staying empty.
                let needs_build = reg
                    .get(&ws.name)
                    .ok()
                    .flatten()
                    .map(|r| r.current_snapshot.trim().is_empty())
                    .unwrap_or(false);
                if rep.changed_any() || needs_build {
                    let why = if rep.changed_any() {
                        format!("changed [{}]", rep.changed.join(", "))
                    } else {
                        "empty warehouse".to_string()
                    };
                    eprintln!("nornir-monitor: {} {why} → republish", rep.workspace);
                    // Incremental scoping (F1): pass the changed members so the
                    // rebuild covers only `changed ∪ dependents`. On the
                    // "empty warehouse" path there is no change-set → `None`
                    // forces a full rebuild (first-publish fallback).
                    let changed: Option<&[String]> =
                        if rep.changed_any() { Some(&rep.changed) } else { None };
                    match republish(reg, root, &ws.name, changed) {
                        Ok(snap) => eprintln!(
                            "nornir-monitor: {} republished → snapshot {snap}",
                            ws.name
                        ),
                        Err(e) => {
                            eprintln!("nornir-monitor: {} republish failed: {e:#}", ws.name)
                        }
                    }
                }
            }
            Err(e) => eprintln!("nornir-monitor: {} fetch failed: {e:#}", ws.name),
        }
    }
    Ok(())
}

/// Rebuild + republish a workspace's warehouse from its freshly-fetched
/// sources: knowledge scan + member-scoped code index + a per-member snapshot,
/// all under `<root>/<ws>/builds/` (the workspace's own Iceberg warehouse).
/// Returns the latest snapshot id; records it on the registry row.
///
/// Operates on the materialized clones under `<root>/<ws>/git/<member>/` — only
/// members with a real checkout (HTTPS today; SSH pack-fetch pending).
///
/// **Incremental (F1):** `changed` carries the members whose HEAD moved this
/// sweep. When present, the index rebuild is scoped to the *affected set* —
/// `changed ∪ transitive dependents` from the cross-repo dep graph — so an
/// unaffected member is not re-walked. `changed = None` (first publish / empty
/// warehouse) falls back to a full rebuild over all members, as does a missing
/// dep graph.
///
/// `pub` so the eager `nornir workspace add` (fat mode) can build inline right
/// after fetch — the same build the poll loop runs on a change. Server and fat
/// thus do the *same* thing; the poll loop is just the automated re-run.
pub fn republish(
    reg: &Registry,
    root: &Path,
    ws_name: &str,
    changed: Option<&[String]>,
) -> Result<String> {
    use crate::warehouse::iceberg::IcebergWarehouse;

    let ws_root = root.join(ws_name);
    let git_dir = ws_root.join("git");
    let builds = ws_root.join("builds");
    std::fs::create_dir_all(&builds)?;

    let mut record = reg
        .get(ws_name)?
        .ok_or_else(|| anyhow!("workspace `{ws_name}` vanished from the registry"))?;
    let members: Vec<String> = record.members.iter().map(|m| m.name.clone()).collect();

    let index_dir = builds.join("cache").join("index");
    let wh = IcebergWarehouse::open(&builds)
        .with_context(|| format!("open warehouse {}", builds.display()))?;
    let idx = crate::index::Index::open_at(&git_dir, &index_dir)?
        .with_repo_scope(members.clone());

    // Deep-scan is opted into per-workspace in the descriptor ([workspace]
    // deep_scan = true). Read it from the descriptor content stored IN the registry
    // row (the source of truth — no sibling file); fall back to the on-disk
    // descriptor only for legacy rows written before content was persisted.
    let deep_scan = if !record.descriptor_content.trim().is_empty() {
        crate::workspace::WorkspaceDescriptor::from_content(&record.descriptor_content)
            .map(|d| d.workspace.deep_scan)
            .unwrap_or(false)
    } else {
        crate::workspace::WorkspaceDescriptor::load(Path::new(&record.descriptor))
            .map(|d| d.workspace.deep_scan)
            .unwrap_or(false)
    };

    let snap = republish_with(
        &wh, &idx, &git_dir, &index_dir, &members, ws_name, deep_scan, changed,
    )?;

    record.current_snapshot = snap.clone();
    record.updated_at = chrono::Utc::now().to_rfc3339();
    reg.upsert(&record)?;
    Ok(snap)
}

/// Republish a monitored workspace **through caller-owned handles** — the warehouse
/// + Tantivy index the `nornir-server` already holds open (redb is single-writer,
/// so the server must be the sole opener; the poll loop writes through these same
/// handles rather than opening its own). Runs the producer pipeline over the
/// materialized member checkouts under `git_dir` and returns the latest snapshot
/// id. Members without a `.git` checkout (e.g. SSH-only, not yet clonable) are
/// skipped. Errors when none are present.
///
/// **Incremental (F1):** when `changed` names the members that moved, the dep
/// graph is built *up front* and the index rebuild is scoped to
/// [`affected_set`] (`changed ∪ transitive dependents`); members outside that
/// set are not re-walked. `changed = None` or an unbuildable graph fall back to
/// a full rebuild over every present member.
pub fn republish_with(
    wh: &crate::warehouse::iceberg::IcebergWarehouse,
    idx: &crate::index::Index,
    git_dir: &Path,
    index_dir: &Path,
    members: &[String],
    ws_name: &str,
    deep_scan: bool,
    changed: Option<&[String]>,
) -> Result<String> {
    let present: Vec<&String> = members
        .iter()
        .filter(|name| git_dir.join(name).join(".git").exists())
        .collect();
    if present.is_empty() {
        bail!(
            "no materialized member checkouts under {} (SSH-only members can't be \
             cloned yet)",
            git_dir.display()
        );
    }

    // Cross-repo dep graph (cheap, offline DIRECT-manifest parse — no `cargo
    // metadata` subprocess). Built up front so the affected-set scoping below can
    // consult it; reused for the step-4 persist. Best-effort — a graph hiccup
    // just means a full rebuild, and a single bad manifest is skipped not fatal.
    let graph = match build_member_graph(git_dir, &present) {
        Ok(g) => Some(g),
        Err(e) => {
            eprintln!("nornir-monitor: {ws_name} dep-graph build skipped: {e:#}");
            None
        }
    };

    // Affected set = changed ∪ transitive dependents. `None` (first publish /
    // empty warehouse) or a missing graph → empty scope = full rebuild.
    let scope: Vec<String> = match (changed, graph.as_ref()) {
        (Some(ch), Some(g)) if !ch.is_empty() => affected_set(g, ch, &present),
        _ => Vec::new(),
    };
    if scope.is_empty() {
        eprintln!("nornir-monitor: {ws_name} → full rebuild over all members");
    } else {
        eprintln!(
            "nornir-monitor: {ws_name} → incremental rebuild, affected [{}]",
            scope.join(", ")
        );
    }

    // 1. Knowledge scan (symbols + git-heat) per member → persist.
    //    **Incremental by git-SHA idempotency**: a member whose HEAD SHA was
    //    already scanned is skipped. The members that DO need scanning have their
    //    source files parsed in ONE flattened gatling pool over the union of all
    //    their files (`symbols::scan_repos_flattened`) — 1 huge member or 10 small
    //    ones both saturate every core, size-skew auto-balanced — instead of a
    //    per-member loop each opening its own pool (which would nest/oversubscribe).
    //    Only the warehouse appends below stay serial (single-writer, zero compute).
    use std::time::Instant;
    let t_scan = Instant::now();
    let now = chrono::Utc::now();

    // 1a. Decide which members need a (re)scan — the incremental SHA-ledger skip.
    let mut to_scan: Vec<(&String, std::path::PathBuf, String)> = Vec::new();
    for name in &present {
        let repo_root = git_dir.join(name.as_str());
        let (sha, _branch) = crate::gitio::head_sha_and_branch(&repo_root)
            .unwrap_or_else(|_| ("unknown".into(), "unknown".into()));
        // Skip-on-unchanged is keyed only on the git SHA ledger, so a member that
        // was scanned by an OLDER binary (before a new knowledge table existed) has
        // a ledger row but NO rows in that new table — and the skip would prevent
        // ever backfilling it. Guard the skip on the `test_inventory` (build-free
        // test surface) being present too: when the knowledge scan exists BUT the
        // inventory is empty, fall through and re-scan to backfill it. This is the
        // fix for the 🧪 Test pane showing "no tests discovered" for members last
        // scanned before `test_inventory` shipped. Read is cheap + best-effort.
        let scan_exists = wh.knowledge_scan_exists(name, &sha).unwrap_or(false);
        let inventory_present = scan_exists
            && crate::warehouse::test_inventory::query_test_inventory(wh, name.as_str())
                .map(|rows| !rows.is_empty())
                .unwrap_or(false);
        if should_skip_member_scan(scan_exists, inventory_present) {
            eprintln!("nornir-monitor: {ws_name}/{name} @ {} — knowledge unchanged, skip scan", &sha[..sha.len().min(8)]);
            continue;
        }
        if scan_exists {
            eprintln!(
                "nornir-monitor: {ws_name}/{name} @ {} — knowledge unchanged but test_inventory empty → re-scan to backfill",
                &sha[..sha.len().min(8)]
            );
        }
        to_scan.push(((*name), repo_root, sha));
    }
    let n_scan = to_scan.len();

    // 1b. ONE flattened gatling pool over the UNION of every to-scan member's files
    //     (heaviest-first across the whole set), grouped back per member. Per-member
    //     symbol snapshot ids key each member's warehouse rows + SHA ledger.
    let repos: Vec<(String, std::path::PathBuf)> =
        to_scan.iter().map(|(n, r, _)| ((*n).clone(), r.clone())).collect();
    let sym_snaps: Vec<uuid::Uuid> = (0..repos.len()).map(|_| uuid::Uuid::new_v4()).collect();
    let symbol_scans = if repos.is_empty() {
        Vec::new()
    } else {
        crate::knowledge::symbols::scan_repos_flattened(&repos, &sym_snaps, now)
    };

    // 1c. Per-member git-heat scan + the SERIAL single-writer warehouse appends.
    //     git-heat is a per-repo gix walk (no inner pool); the symbol rows already
    //     came from the flattened pool above. Appends are the only serial step.
    for (i, (name, repo_root, sha)) in to_scan.iter().enumerate() {
        let symbols = &symbol_scans[i];
        wh.append_symbol_scan(symbols)?;
        match crate::knowledge::git_heat::scan_repo(repo_root, name, uuid::Uuid::new_v4(), now) {
            Ok(git) => wh.append_git_heat_scan(&git)?,
            Err(e) => eprintln!("nornir-monitor: {ws_name}/{name} git-heat scan skipped: {e:#}"),
        }
        if let Err(e) = wh.record_knowledge_scan(name, sha, symbols.snapshot_id) {
            eprintln!("nornir-monitor: {ws_name}/{name} knowledge-scan ledger skipped: {e:#}");
        }
    }
    eprintln!(
        "nornir-monitor: {ws_name} ⏱ knowledge-scan {n_scan} member(s) in {:.1}s",
        t_scan.elapsed().as_secs_f64()
    );

    // 1d. Security SBOM warm for the scanned members. Three phases so the only
    //     serial work is the single-writer warehouse append:
    //       prep   — SERIAL: clone-missing + symlink path-dep siblings under the
    //                 shared `git_dir` (concurrent git clones / symlinks would race);
    //       resolve — PARALLEL: per-member `cargo metadata` SBOM closure, all-core
    //                 (read-only, touches no warehouse / shared fs);
    //       persist — SERIAL: append each member's components (single-writer, no compute).
    if !to_scan.is_empty() {
        for (_, repo_root, _) in &to_scan {
            crate::security::warm_prepare(repo_root, git_dir);
        }
        let roots: Vec<std::path::PathBuf> = to_scan.iter().map(|(_, r, _)| r.clone()).collect();
        let resolved = crate::warehouse::dep_graph::parallel_map(&roots, |root| {
            crate::security::warm_resolve(root)
        });
        for (i, res) in resolved.into_iter().enumerate() {
            let name = to_scan[i].0;
            match res {
                Ok((repo_name, comps)) => {
                    if let Err(e) = crate::security::persist_sbom(wh, &repo_name, &comps) {
                        eprintln!("nornir-monitor: {ws_name}/{name} security persist skipped: {e:#}");
                    }
                }
                Err(e) => eprintln!("nornir-monitor: {ws_name}/{name} security warm skipped: {e:#}"),
            }
        }
    }

    // 1b. RESOLVED (SCIP) call-edge ingest — the CHEAP alternative to deep_scan.
    //     Runs the workspace-scoped resolved ingest (the same thing `nornir scip
    //     ingest` does) per present member and persists `scip_occurrences`, so the
    //     metro/architecture feeds' `query::load_preferred` returns IDE-precise
    //     RESOLVED call-edges instead of syn's last-segment heuristic — WITHOUT the
    //     1400-crate transitive-closure deep_scan. Gated on `#[cfg(feature="scip")]`
    //     (default builds compile this away). Incremental + idempotent: a member
    //     whose latest `scip_occurrences` snapshot already sits at its HEAD SHA is
    //     skipped (SHA-keyed, like the step-1 knowledge ledger). Best-effort —
    //     never fails republish. Opt out with `NORNIR_SKIP_RESOLVE=1`.
    //
    //     INTENTIONALLY SERIAL ACROSS MEMBERS (not fanned like steps 1b/1d above).
    //     `resolve_member_scip` resolves a member's WHOLE crate graph at once and
    //     is already INTERNALLY all-core: the `ra-ingest` path
    //     (`ra_ingest::ingest_in_process`) is gatling-work-stolen over the member's
    //     files, and the fallback `rust-analyzer scip` subprocess uses its own
    //     thread pool — so one member already saturates every core, leaving no idle
    //     cores for a second member to reclaim. Running members concurrently would
    //     therefore NEST that internal pool (oversubscription), and the subprocess
    //     path would hold N rust-analyzer workspace analyses in memory at once
    //     (multi-GB each) — unsafe on the server. Per-member work being all-core
    //     already satisfies the saturate-all-cores principle; the only serial cost
    //     is the cheap SHA skip-check and the single-writer `append_scip_scan`.
    #[cfg(feature = "scip")]
    if std::env::var_os("NORNIR_SKIP_RESOLVE").is_none() {
        let t_res = Instant::now();
        let mut n_res = 0usize;
        let mut n_skip = 0usize;
        for name in &present {
            let repo_root = git_dir.join(name.as_str());
            let (sha, _branch) = crate::gitio::head_sha_and_branch(&repo_root)
                .unwrap_or_else(|_| ("unknown".into(), "unknown".into()));
            // Idempotent skip: latest persisted resolved snapshot already at HEAD.
            let resolved_at_head = wh
                .load_latest_scip(name.as_str())
                .map(|s| scip_resolved_at_head(&s, &sha))
                .unwrap_or(false);
            if resolved_at_head {
                eprintln!(
                    "nornir-monitor: {ws_name}/{name} @ {} — resolved edges unchanged, skip ingest",
                    &sha[..sha.len().min(8)]
                );
                n_skip += 1;
                continue;
            }
            match resolve_member_scip(&repo_root, name.as_str(), &sha) {
                Ok(scan) => {
                    let defs = scan.rows.iter().filter(|r| r.is_definition).count();
                    if let Err(e) = wh.append_scip_scan(&scan) {
                        eprintln!("nornir-monitor: {ws_name}/{name} resolved-ingest persist skipped: {e:#}");
                    } else {
                        eprintln!(
                            "nornir-monitor: {ws_name}/{name} @ {} — resolved {} occurrence(s) ({defs} def)",
                            &sha[..sha.len().min(8)], scan.rows.len()
                        );
                        n_res += 1;
                    }
                }
                Err(e) => {
                    eprintln!("nornir-monitor: {ws_name}/{name} resolved-ingest skipped: {e:#}");
                }
            }
        }
        eprintln!(
            "nornir-monitor: {ws_name} ⏱ resolved-ingest {n_res} member(s) ({n_skip} unchanged) in {:.1}s",
            t_res.elapsed().as_secs_f64()
        );
    }

    // 2. (Re)build the workspace code index. Incrementally scoped to the
    //    affected members when `scope` is non-empty; full-workspace otherwise.
    let t_index = Instant::now();
    idx.build_scoped(&scope).context("build workspace index")?;
    eprintln!("nornir-monitor: {ws_name} ⏱ index-build in {:.1}s", t_index.elapsed().as_secs_f64());

    // 3. Snapshot the index into Iceberg, keyed to each member's HEAD.
    let t_snap = Instant::now();
    let mut last_snapshot = String::new();
    for name in &present {
        let repo_root = git_dir.join(name);
        let (sha, branch) = crate::gitio::head_sha_and_branch(&repo_root)
            .unwrap_or_else(|_| ("unknown".into(), "unknown".into()));
        let snap = crate::index::snapshot::snapshot_to_iceberg(
            wh, ws_name, name, &sha, &branch, index_dir,
        )
        .with_context(|| format!("snapshot {name}"))?;
        last_snapshot = snap.snapshot_id.to_string();
    }
    eprintln!(
        "nornir-monitor: {ws_name} ⏱ snapshot {} member(s) in {:.1}s",
        present.len(),
        t_snap.elapsed().as_secs_f64()
    );

    // 4. Persist the cross-repo dependency graph built up front (step 0) so the
    //    viz Dep Graph tab reflects the monitored sources. Note: this graph is
    //    *cross-repo*, so a single-member workspace yields no edges; it lights up
    //    for multi-repo monitored workspaces.
    if let Some(graph) = graph.as_ref() {
        if let Err(e) = wh.record_dep_graph(ws_name, graph) {
            eprintln!("nornir-monitor: {ws_name} dep-graph persist skipped: {e:#}");
        }
    }

    // 5. Opt-in deep-scan: knowledge-scan the members' entire transitive dep
    //    closure (gatling-fanned) into this same warehouse under repo `deps`.
    //    Best-effort: needs `cargo` on PATH + network for the first fetch.
    if deep_scan {
        let t_deep = Instant::now();
        let dirs: Vec<std::path::PathBuf> =
            present.iter().map(|n| git_dir.join(n.as_str())).collect();
        match crate::deepscan::deep_scan(&dirs, wh) {
            Ok(rep) => {
                eprintln!(
                    "nornir-monitor: {ws_name} ⏱ deep-scan: {} dep crate(s) scanned, {} skipped, {} error(s) in {:.1}s",
                    rep.crates,
                    rep.skipped,
                    rep.errors.len(),
                    t_deep.elapsed().as_secs_f64(),
                );
                for e in rep.errors.iter().take(5) {
                    eprintln!("nornir-monitor:   deep-scan error: {e}");
                }
            }
            Err(e) => eprintln!("nornir-monitor: {ws_name} deep-scan skipped: {e:#}"),
        }
    }

    // 6. Test surface (populate): discovery is now BUILD-FREE — step 1's syn scan
    //    already wrote one `test_inventory` row per `#[test]` (no `cargo build`,
    //    no `cargo test --list`). The 🧪 Test pane shows EVERY discovered test as
    //    `X` (not-run) the instant the workspace is added; `ok`/`fail` fills in only
    //    once `nornir release` / `nornir test run` EXECUTES them (never here, never
    //    in the poll loop). This step just logs the per-member inventory count to
    //    `release_events`. Opt out of the LOGGING with NORNIR_POPULATE_SKIP_TEST=1
    //    (discovery itself still happens — it's free, part of the knowledge scan).
    //    Best-effort: never fails republish.
    if std::env::var_os("NORNIR_POPULATE_SKIP_TEST").is_none() {
        let to_scan = members_to_populate(&present, &scope);
        if let Err(e) = discover_test_surface(wh, git_dir, ws_name, &to_scan) {
            eprintln!("nornir-monitor: {ws_name} test discovery skipped: {e:#}");
        }
    }

    Ok(last_snapshot)
}

/// Incremental skip predicate for the step-1b resolved (SCIP) ingest: `true`
/// when the member's latest persisted `scip_occurrences` snapshot has rows AND
/// sits at the member's current HEAD `sha` — i.e. the resolved map is already
/// up to date and the (heavy) re-ingest can be skipped. SHA-keyed + idempotent,
/// mirroring the step-1 knowledge-scan ledger.
#[cfg(feature = "scip")]
fn scip_resolved_at_head(latest: &crate::knowledge::scip::ScipScan, sha: &str) -> bool {
    !latest.rows.is_empty() && latest.git_sha == sha
}

/// Resolve one member's RESOLVED (SCIP) knowledge for the step-1b ingest.
///
/// Prefers the FAST in-process resolver (`ra_ingest::ingest_in_process`, ~63.5s
/// cold then incremental) when the binary is built `--features ra-ingest`; under
/// plain `--features scip` it falls back to spawning `rust-analyzer scip` (the
/// subprocess path — compiles the workspace, so slower cold). Either way the
/// result is a SHA-keyed [`ScipScan`](crate::knowledge::scip::ScipScan) over the
/// workspace's OWN sources (not the dep closure), ready for `append_scip_scan`.
#[cfg(feature = "scip")]
fn resolve_member_scip(
    repo_root: &Path,
    repo: &str,
    sha: &str,
) -> Result<crate::knowledge::scip::ScipScan> {
    let snapshot_id = uuid::Uuid::new_v4();
    let ts = chrono::Utc::now();
    #[cfg(feature = "ra-ingest")]
    {
        crate::knowledge::ra_ingest::ingest_in_process(repo_root, repo, sha, snapshot_id, ts)
    }
    #[cfg(not(feature = "ra-ingest"))]
    {
        crate::knowledge::scip::ingest_via_rust_analyzer(repo_root, repo, sha, snapshot_id, ts)
    }
}

/// Skip-on-unchanged decision for a member's knowledge scan. The git-SHA ledger
/// (`scan_exists`) alone is NOT enough: a member scanned by an OLDER binary — one
/// that predates a knowledge table such as `test_inventory` — has a ledger row but
/// NO rows in that table, and skipping on the ledger alone would prevent ever
/// backfilling it (the 🧪 Test pane shows "no tests discovered"). So we only skip
/// when the SHA was scanned AND its `test_inventory` is present; an empty inventory
/// at a recorded SHA falls through to a re-scan that backfills it.
fn should_skip_member_scan(scan_exists: bool, inventory_present: bool) -> bool {
    scan_exists && inventory_present
}

/// The members a populate step should cover: the full `present` set on a full
/// rebuild (empty `scope`), else just the affected members in `scope`. Pure +
/// testable — the bound that keeps poll-loop populate work small (a poll tick
/// usually moves 0–1 members, so only those get re-discovered/re-tested).
fn members_to_populate<'a>(present: &[&'a String], scope: &[String]) -> Vec<&'a String> {
    if scope.is_empty() {
        present.to_vec()
    } else {
        present
            .iter()
            .copied()
            .filter(|n| scope.iter().any(|s| s == *n))
            .collect()
    }
}

/// Populate **step 6** — log the per-member test-discovery summary. Discovery
/// itself is now **build-free**: it rode the step-1 syn knowledge scan, which
/// wrote one [`test_inventory`](crate::warehouse::test_inventory) row per
/// `#[test]` (no `cargo build`, no `cargo test --list`). This step just reads
/// that inventory back per member and records the count to `release_events`
/// (`op = test_discover`), so a freshly populated workspace's 🧪 Test pane shows
/// every test as `X` (not-run) the instant it's added — the matrix join supplies
/// the `X` for any inventory test with no `test_results` row, so we no longer
/// seed `listed` rows here. EXECUTION (which fills ok/fail) stays gated to
/// `nornir release` / `nornir test run` and NEVER happens on this path.
/// Best-effort: a per-member read/log error is logged and skipped, never fatal.
fn discover_test_surface(
    wh: &crate::warehouse::iceberg::IcebergWarehouse,
    _git_dir: &Path,
    ws_name: &str,
    members: &[&String],
) -> Result<()> {
    use crate::warehouse::release_events::{append_release_events, phase, status, ReleaseEventRow};
    use crate::warehouse::test_inventory::query_test_inventory;
    use crate::warehouse::test_results::new_run_id;

    if members.is_empty() {
        return Ok(());
    }
    let run_id = new_run_id();
    let ts = chrono::Utc::now().timestamp_micros();
    let mut seq: i64 = 0;
    let mut log = |repo: &str, st: &str, detail: String| {
        let row = ReleaseEventRow {
            run_id: run_id.clone(),
            seq,
            ts_micros: ts,
            component: ws_name.to_string(),
            repo: repo.to_string(),
            op: "test_discover".into(),
            phase: phase::END.into(),
            status: st.into(),
            detail,
            depends_on: None,
            elapsed_ms: None,
        };
        seq += 1;
        // Best-effort: a logging-row failure must not abort populate.
        let _ = wh.block_on(append_release_events(wh, std::slice::from_ref(&row)));
    };

    for name in members {
        // The inventory was written by step 1's syn scan — read it back (NO build).
        match query_test_inventory(wh, name.as_str()) {
            Ok(inv) => {
                let heavy = inv.iter().filter(|t| t.is_heavy).count();
                eprintln!(
                    "nornir-monitor: {ws_name}/{name} — {} test(s) inventoried via syn scan ({heavy} heavy) (NOT RUN, X)",
                    inv.len()
                );
                log(name, status::OK, format!("{} tests discovered ({heavy} heavy), build-free", inv.len()));
            }
            Err(e) => {
                eprintln!("nornir-monitor: {ws_name}/{name} test inventory read skipped: {e:#}");
                log(name, status::WARN, format!("inventory read skipped: {e}"));
            }
        }
    }
    Ok(())
}

/// The **affected set** for an incremental republish: the members whose HEAD
/// moved (`changed`) ∪ their transitive dependents, per the cross-repo dep
/// [`graph`](crate::warehouse::dep_graph::WorkspaceGraph). The result is
/// intersected with `present` (the materialized member checkouts) — a changed
/// member that has no checkout, or a graph node that isn't a workspace member,
/// is dropped so the index scope only ever names real member directories.
///
/// Returns an empty vec only when nothing in `changed` resolves to a present
/// member; callers treat an empty scope as "full rebuild", so this stays a
/// safe over-approximation rather than silently indexing nothing.
fn affected_set(
    graph: &crate::warehouse::dep_graph::WorkspaceGraph,
    changed: &[String],
    present: &[&String],
) -> Vec<String> {
    let present_set: std::collections::BTreeSet<&str> =
        present.iter().map(|s| s.as_str()).collect();
    graph
        .affected_by_change(changed)
        .into_iter()
        .filter(|m| present_set.contains(m.as_str()))
        .collect()
}

/// Build a cross-repo [`WorkspaceGraph`] over the materialized member checkouts
/// at `git_dir/<member>`. Uses [`WorkspaceGraph::build_from_members`] — direct
/// `Cargo.toml` parsing, all-core, NO `cargo metadata` subprocess — so the graph
/// builds offline, saturates cores even for a single huge member, and a single
/// broken manifest (e.g. facett) is skipped rather than sinking the whole graph.
fn build_member_graph(
    git_dir: &Path,
    members: &[&String],
) -> Result<crate::warehouse::dep_graph::WorkspaceGraph> {
    let resolved: std::collections::BTreeMap<String, std::path::PathBuf> = members
        .iter()
        .map(|name| ((*name).clone(), git_dir.join(name)))
        .collect();
    crate::warehouse::dep_graph::WorkspaceGraph::build_from_members(&resolved)
}

/// Parse `"60s"`, `"5m"`, `"2h"`, or a bare seconds count; fall back to `default`.
pub fn parse_interval(s: &str, default: Duration) -> Duration {
    let s = s.trim();
    if s.is_empty() {
        return default;
    }
    let (num, mult) = match s.chars().last() {
        Some('s') => (&s[..s.len() - 1], 1),
        Some('m') => (&s[..s.len() - 1], 60),
        Some('h') => (&s[..s.len() - 1], 3600),
        _ => (s, 1),
    };
    num.trim()
        .parse::<u64>()
        .ok()
        .map(|n| Duration::from_secs(n.saturating_mul(mult)))
        .unwrap_or(default)
}

/// Spawn the background poll loop over the server's shared registry. Fetches run
/// on a blocking thread so gix I/O never stalls the async runtime. Returns the
/// join handle immediately.
pub fn spawn_poll_loop(
    reg: Arc<Registry>,
    root: PathBuf,
    tick: Duration,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(tick);
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        loop {
            interval.tick().await;
            let reg = reg.clone();
            let root = root.clone();
            match tokio::task::spawn_blocking(move || sync_once(&reg, &root)).await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => eprintln!("nornir-monitor: sweep error: {e:#}"),
                Err(e) => eprintln!("nornir-monitor: sweep task panicked: {e}"),
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::warehouse::dep_graph::{CrossRepoEdge, RepoFacts, WorkspaceGraph};
    use std::collections::{BTreeMap, BTreeSet};

    fn facts(name: &str, produces: &[&str], consumes: &[&str]) -> RepoFacts {
        RepoFacts {
            name: name.to_string(),
            root: std::path::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 dep graph: app → liba → util, app → libb → util.
    /// Edges point consumer → producer. A change to `util` should blast up to
    /// liba, libb, app; a change to `liba` should reach only app (not libb/util).
    fn diamond() -> WorkspaceGraph {
        let mut fmap = BTreeMap::new();
        for f in [
            facts("app", &["app_c"], &["a_c", "b_c"]),
            facts("liba", &["a_c"], &["util_c"]),
            facts("libb", &["b_c"], &["util_c"]),
            facts("util", &["util_c"], &[]),
        ] {
            fmap.insert(f.name.clone(), f);
        }
        WorkspaceGraph::from_query_parts(
            fmap,
            vec![
                edge("app", "liba", &["a_c"]),
                edge("app", "libb", &["b_c"]),
                edge("liba", "util", &["util_c"]),
                edge("libb", "util", &["util_c"]),
            ],
        )
    }

    /// F1 inject-assert: feed a fake change-set + a real dep graph, assert the
    /// affected set is exactly `changed ∪ transitive dependents`, and that an
    /// unaffected member (one that nothing changed depends on) is *skipped*.
    #[test]
    fn affected_set_is_changed_union_dependents_and_skips_unaffected() {
        let g = diamond();
        // All four members have a materialized checkout.
        let all = ["app".to_string(), "liba".to_string(), "libb".to_string(), "util".to_string()];
        let present: Vec<&String> = all.iter().collect();

        // A change to `liba` only. Affected = liba ∪ its dependents = {liba, app}.
        let affected = affected_set(&g, &["liba".to_string()], &present);
        let got: BTreeSet<&str> = affected.iter().map(|s| s.as_str()).collect();
        assert_eq!(
            got,
            ["liba", "app"].into_iter().collect::<BTreeSet<_>>(),
            "affected set must be changed ∪ transitive dependents"
        );
        // `libb` and `util` are unaffected by a change to `liba` → skipped.
        assert!(!got.contains("libb"), "sibling libb must be skipped — it does not depend on liba");
        assert!(!got.contains("util"), "dependency util must be skipped — liba depends on it, not vice-versa");
        // Build order: a dependency must precede its dependent (liba before app).
        let pos = |n: &str| affected.iter().position(|x| x == n).unwrap();
        assert!(pos("liba") < pos("app"), "affected set must come back in build order");

        // A change to the root `util` blasts up the whole diamond.
        let blast = affected_set(&g, &["util".to_string()], &present);
        let blast_set: BTreeSet<&str> = blast.iter().map(|s| s.as_str()).collect();
        assert_eq!(
            blast_set,
            ["app", "liba", "libb", "util"].into_iter().collect::<BTreeSet<_>>(),
            "a change to the shared leaf must rebuild every member"
        );

        // A changed member with no materialized checkout is dropped from the scope
        // (intersection with `present`), so the index never names a missing dir.
        let only_app = vec![&all[0]]; // present = {app}
        let scoped = affected_set(&g, &["liba".to_string()], &only_app);
        assert_eq!(
            scoped,
            vec!["app".to_string()],
            "affected set is intersected with present checkouts — absent `liba` is dropped"
        );
    }

    #[test]
    fn populate_scope_is_all_on_full_rebuild_else_just_affected() {
        let a = "a".to_string();
        let b = "b".to_string();
        let c = "c".to_string();
        let present = vec![&a, &b, &c];
        // Full rebuild (empty scope) → every present member gets discovered/tested.
        assert_eq!(members_to_populate(&present, &[]), vec![&a, &b, &c]);
        // Incremental (a poll tick moved `b`) → only `b`; an absent name is dropped.
        let scope = vec!["b".to_string(), "ghost".to_string()];
        assert_eq!(members_to_populate(&present, &scope), vec![&b]);
        // No present members → nothing, never a panic.
        assert!(members_to_populate(&[], &scope).is_empty());
    }

    /// Inject-assert: the skip-on-unchanged decision is `scan_exists ∧ inventory_present`.
    /// The load-bearing case is the third one — a member scanned by an OLD binary has the
    /// SHA ledger row but an EMPTY `test_inventory`, and MUST NOT be skipped (else it never
    /// backfills and `nornir test list` shows "no tests discovered").
    #[test]
    fn skip_only_when_scanned_and_inventory_present() {
        // Fresh SHA (never scanned) → must scan.
        assert!(!should_skip_member_scan(false, false), "unseen SHA must be scanned");
        // SHA scanned by the CURRENT binary, inventory written → skip (the steady state).
        assert!(should_skip_member_scan(true, true), "scanned SHA with inventory → skip");
        // SHA scanned by an OLD binary: ledger row present but inventory EMPTY → re-scan
        // to backfill. This is the bug fix — the skip must NOT fire here.
        assert!(
            !should_skip_member_scan(true, false),
            "scanned SHA with EMPTY inventory must re-scan to backfill the new table"
        );
    }

    /// End-to-end inject-assert against a real warehouse: simulate the OLD-binary state
    /// (a `knowledge_scans` ledger row at a SHA, but NO `test_inventory` rows), assert the
    /// monitor decision falls through to re-scan, then write inventory (as the re-scan
    /// would) and assert the decision flips to skip. Proves the backfill actually populates.
    #[test]
    fn empty_inventory_at_recorded_sha_backfills_then_skips() {
        use crate::knowledge::symbols::{SymbolScan, TestDefRow};
        use crate::warehouse::iceberg::IcebergWarehouse;
        use crate::warehouse::test_inventory::query_test_inventory;

        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();
        let repo = "holger";
        let sha = "deadbeefcafef00d";

        // OLD-binary state: the SHA is in the knowledge-scan ledger, but the
        // test_inventory table has no rows for this repo (the table didn't exist
        // when this SHA was scanned).
        wh.record_knowledge_scan(repo, sha, uuid::Uuid::new_v4()).unwrap();
        let scan_exists = wh.knowledge_scan_exists(repo, sha).unwrap();
        let inv0 = query_test_inventory(&wh, repo).unwrap();
        assert!(scan_exists, "ledger row must be present (simulating an old scan)");
        assert!(inv0.is_empty(), "inventory must start empty (old binary wrote none)");
        assert!(
            !should_skip_member_scan(scan_exists, !inv0.is_empty()),
            "empty inventory at a recorded SHA must NOT skip — it must backfill"
        );

        // The re-scan (step 1) would append a SymbolScan carrying test rows. Emulate
        // that write, then assert the inventory is now non-empty and the decision flips.
        let mut scan = SymbolScan {
            snapshot_id: uuid::Uuid::new_v4(),
            ts: chrono::Utc::now(),
            repo: repo.to_string(),
            ..Default::default()
        };
        scan.tests.push(TestDefRow {
            crate_name: repo.to_string(),
            module_path: format!("{repo}::tests"),
            test_name: "it_works".into(),
            file: "src/lib.rs".into(),
            line: 42,
            is_heavy: false,
            is_async: false,
        });
        wh.append_symbol_scan(&scan).unwrap();

        let inv1 = query_test_inventory(&wh, repo).unwrap();
        assert_eq!(inv1.len(), 1, "backfill must write the discovered test row");
        assert_eq!(inv1[0].test_name, "it_works");
        assert!(
            should_skip_member_scan(scan_exists, !inv1.is_empty()),
            "once inventory is backfilled, the SHA is skipped again"
        );
    }

    /// Step-1b resolved (SCIP) ingest is incremental + idempotent: once a member's
    /// resolved rows are persisted at its HEAD SHA, the skip predicate fires (no
    /// re-ingest); a SHA change reopens it; and the persisted rows are exactly what
    /// `load_preferred` then PREFERS (source tag `resolved/scip`). Uses a temp
    /// warehouse — NEVER a live one — and a hand-built index (no rust-analyzer).
    #[cfg(feature = "scip")]
    #[test]
    fn resolved_ingest_is_incremental_and_preferred() {
        use crate::knowledge::scip::{ingest_index, ScipScan};
        use crate::warehouse::iceberg::IcebergWarehouse;
        use scip::types::{symbol_information, Document, Index, Occurrence, SymbolInformation, SymbolRole};

        // A resolved index: `outer` (a fn) CALLS `inner` — a containment edge the
        // syn last-segment heuristic cannot attribute precisely.
        let mut idx = Index::new();
        let mut doc = Document::new();
        doc.relative_path = "src/lib.rs".into();
        let mut si = SymbolInformation::new();
        si.symbol = "rust-analyzer cargo demo 0.1.0 inner().".into();
        si.display_name = "inner".into();
        si.kind = symbol_information::Kind::Function.into();
        doc.symbols.push(si.clone());
        let mut outer = SymbolInformation::new();
        outer.symbol = "rust-analyzer cargo demo 0.1.0 outer().".into();
        outer.display_name = "outer".into();
        outer.kind = symbol_information::Kind::Function.into();
        doc.symbols.push(outer.clone());
        // def of `inner`
        let mut def = Occurrence::new();
        def.range = vec![1, 3, 1, 8];
        def.symbol = si.symbol.clone();
        def.symbol_roles = SymbolRole::Definition as i32;
        doc.occurrences.push(def);
        // def of `outer`, enclosing lines 5..9 (the caller body)
        let mut odef = Occurrence::new();
        odef.range = vec![5, 3, 5, 8];
        odef.symbol = outer.symbol.clone();
        odef.symbol_roles = SymbolRole::Definition as i32;
        odef.enclosing_range = vec![5, 0, 9, 1];
        doc.occurrences.push(odef);
        // ref to `inner` inside outer's body
        let mut r = Occurrence::new();
        r.range = vec![6, 4, 6, 9];
        r.symbol = si.symbol.clone();
        doc.occurrences.push(r);
        idx.documents.push(doc);

        let member = "demo";
        let sha = "deadbeefcafef00d";
        let scan: ScipScan =
            ingest_index(idx, member, sha, uuid::Uuid::new_v4(), chrono::Utc::now());

        let dir = tempfile::tempdir().unwrap();
        let wh = IcebergWarehouse::open(dir.path()).unwrap();

        // Before any ingest: nothing persisted → must NOT skip (would ingest).
        let empty = wh.load_latest_scip(member).unwrap();
        assert!(!scip_resolved_at_head(&empty, sha), "no rows yet → re-ingest");

        // Persist (what step-1b's append_scip_scan does) and assert skip fires.
        wh.append_scip_scan(&scan).unwrap();
        let latest = wh.load_latest_scip(member).unwrap();
        assert!(scip_resolved_at_head(&latest, sha), "rows at HEAD SHA → skip");
        // A moved HEAD reopens the ingest.
        assert!(!scip_resolved_at_head(&latest, "0ddba11"), "new SHA → re-ingest");

        // The consumer seam: load_preferred now PREFERS the resolved map and
        // attributes the call to its true caller via containment.
        let (view, source) =
            crate::knowledge::query::load_preferred(&wh, member).unwrap();
        assert_eq!(source, "resolved/scip", "resolved rows are preferred over syn");
        let callers: Vec<&str> =
            view.callers_of("inner").iter().map(|c| c.caller_path.as_str()).collect();
        assert_eq!(callers, vec!["outer"], "resolved caller via containment");
    }

    #[test]
    fn interval_parsing() {
        let d = Duration::from_secs(60);
        assert_eq!(parse_interval("", d), Duration::from_secs(60));
        assert_eq!(parse_interval("30s", d), Duration::from_secs(30));
        assert_eq!(parse_interval("5m", d), Duration::from_secs(300));
        assert_eq!(parse_interval("2h", d), Duration::from_secs(7200));
        assert_eq!(parse_interval("90", d), Duration::from_secs(90));
        assert_eq!(parse_interval("garbage", d), Duration::from_secs(60));
    }
}