nornir 0.4.18

//! Release gates. Each gate returns `Ok(())` on pass, `Err` on fail.
//! Generated binaries propagate the first error and abort the release.

use std::path::Path;

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

use crate::bench::{history, BenchRun};

/// Gate 1: no `[patch.crates-io]` znippy entries in the repo's
/// `Cargo.toml`. Implementation: textual scan; refine later with a
/// proper TOML parse if needed.
pub fn no_path_patches(repo_root: &Path) -> Result<()> {
    let cargo = repo_root.join("Cargo.toml");
    let text = std::fs::read_to_string(&cargo)
        .with_context(|| format!("read {}", cargo.display()))?;
    let mut in_patch = false;
    for (i, line) in text.lines().enumerate() {
        let l = line.trim();
        if l.starts_with('[') {
            in_patch = l.starts_with("[patch.crates-io")
                || l.starts_with("[patch.\"crates-io\"");
            continue;
        }
        if in_patch && l.contains("znippy") && !l.starts_with('#') {
            return Err(anyhow!(
                "[patch.crates-io] znippy entry at {}:{} — strip before release",
                cargo.display(),
                i + 1
            ));
        }
    }
    Ok(())
}

/// Gate 3: holger ops/sec must be ≥ nexus ops/sec for every result.
/// Looks for `holger_ops_sec` and `nexus_ops_sec` in each result's
/// metrics map. Results lacking both keys are skipped.
pub fn nexus_floor(run: &BenchRun) -> Result<()> {
    for r in &run.results {
        let h = r.metrics.get("holger_ops_sec").and_then(|v| v.as_f64());
        let n = r.metrics.get("nexus_ops_sec").and_then(|v| v.as_f64());
        if let (Some(h), Some(n)) = (h, n) {
            if h < n {
                return Err(anyhow!(
                    "nexus floor: {} holger={:.0} < nexus={:.0}",
                    r.name,
                    h,
                    n
                ));
            }
        }
    }
    Ok(())
}

/// Gate 4: no result drops more than `max_drop_pct` versus the last
/// same-machine entry in the history. Compares the first numeric
/// metric present in each result (so works for both ops/sec and MB/s
/// shaped runs).
pub fn no_regression(run: &BenchRun, history_path: &Path, max_drop_pct: f64) -> Result<()> {
    let history = history::read_all(history_path)?;
    no_regression_against(run, &history, max_drop_pct)
}

/// Core of [`no_regression`] operating on an in-memory history slice
/// (e.g. read back from the Iceberg warehouse via
/// `query_bench_runs_async`) instead of a JSONL file. The baseline is the
/// **newest** prior run for the same `machine`, chosen deterministically
/// by `timestamp` rather than slice/scan order (warehouse scans don't
/// guarantee ordering). No prior run for the machine ⇒ `Ok` (nothing to
/// regress against — first run bootstraps the baseline).
///
/// Metric direction: treats **higher = better** (throughput-shaped, e.g.
/// `*_mbs`), matching nornir's bench convention. Latency-style
/// (lower=better) metrics are NOT yet handled — see issue note in plan.
pub fn no_regression_against(run: &BenchRun, history: &[BenchRun], max_drop_pct: f64) -> Result<()> {
    let same: Vec<&BenchRun> = history.iter().filter(|h| h.machine == run.machine).collect();
    let Some(last) = pick_baseline(&same) else {
        return Ok(());
    };
    for r in &run.results {
        let Some(prev) = last.find(&r.name) else { continue };
        for (key, new_val) in &r.metrics {
            let Some(new_f) = new_val.as_f64() else { continue };
            let Some(prev_f) = prev.metrics.get(key).and_then(|v| v.as_f64()) else {
                continue;
            };
            if prev_f <= 0.0 {
                continue;
            }
            let drop_pct = (prev_f - new_f) / prev_f * 100.0;
            if drop_pct > max_drop_pct {
                return Err(anyhow!(
                    "regression: {} {} dropped {:.1}% ({:.2} → {:.2})",
                    r.name,
                    key,
                    drop_pct,
                    prev_f,
                    new_f
                ));
            }
        }
    }
    Ok(())
}

/// Pick the newest baseline run deterministically: by `timestamp` when
/// all candidates carry one (lexicographic compare of RFC3339 strings is
/// chronological), else fall back to last-in-slice (legacy JSONL append
/// order). Returns `None` for an empty candidate set.
fn pick_baseline<'a>(runs: &[&'a BenchRun]) -> Option<&'a BenchRun> {
    if runs.is_empty() {
        return None;
    }
    if runs.iter().all(|r| r.timestamp.is_some()) {
        runs.iter().copied().max_by(|a, b| a.timestamp.cmp(&b.timestamp))
    } else {
        runs.last().copied()
    }
}

/// Gate 5: integration round-trip. Caller supplies a closure that
/// performs `agent push → server store → agent pull` for one artifact
/// kind; this gate runs them in order.
pub fn integration_roundtrip<F>(kinds: &[&str], mut run_one: F) -> Result<()>
where
    F: FnMut(&str) -> Result<()>,
{
    for k in kinds {
        run_one(k).with_context(|| format!("roundtrip failed for {k}"))?;
    }
    Ok(())
}

/// Gate 5 driver: invoke gate 5 by shelling out to
/// `cargo test --test roundtrip_<kind> --release` per kind. Consumer
/// repos (holger, znippy) implement the actual roundtrip logic as
/// Rust `#[test]` functions under `tests/roundtrip_<kind>.rs`. This
/// is the one allowed cargo subprocess pattern (matches the
/// `run_cargo_publish` decision), keeping nornir free of bash
/// shellouts.
pub fn integration_roundtrip_via_cargo_test(repo_root: &Path, kinds: &[&str]) -> Result<()> {
    integration_roundtrip(kinds, |k| {
        let test_name = format!("roundtrip_{k}");
        let status = std::process::Command::new("cargo")
            .args(["test", "--test", &test_name, "--release"])
            .current_dir(repo_root)
            .status()
            .with_context(|| format!("spawn cargo test --test {test_name}"))?;
        if !status.success() {
            return Err(anyhow!("cargo test --test {test_name} exited {status}"));
        }
        Ok(())
    })
}

// ─── Cargo-pipeline gates (features 1, 4, 5, 8, 16) ─────────────────

/// One audit finding from [`path_dep_audit`].
#[derive(Debug, Clone)]
pub struct PathDepFinding {
    pub manifest: std::path::PathBuf,
    pub dep_name: String,
    pub dep_path: String,
    pub has_version: bool,
    pub version_req: Option<String>,
}

impl PathDepFinding {
    pub fn ok(&self) -> bool { self.has_version }
}

/// Gate (feature 1): every `path =` dep in every `Cargo.toml` in the
/// workspace must also carry a `version =` field. Otherwise
/// `cargo publish` rejects the manifest because the patched-out
/// path-dep loses its version constraint in the uploaded `.crate`.
///
/// Walks every `Cargo.toml` under `repo_root` (skipping `target/`,
/// `.git/`). Parses with the `toml` crate and inspects
/// `[dependencies.*]`, `[dev-dependencies.*]`, `[build-dependencies.*]`
/// and the `[target.*]`-scoped equivalents.
pub fn path_dep_audit(repo_root: &Path) -> Result<Vec<PathDepFinding>> {
    let mut findings = Vec::new();
    for manifest in walk_cargo_tomls(repo_root)? {
        let text = std::fs::read_to_string(&manifest)
            .with_context(|| format!("read {}", manifest.display()))?;
        let doc: toml::Value = toml::from_str(&text)
            .with_context(|| format!("parse {}", manifest.display()))?;
        for section in [
            "dependencies",
            "dev-dependencies",
            "build-dependencies",
        ] {
            collect_path_deps(&doc, section, &manifest, &mut findings);
        }
        if let Some(targets) = doc.get("target").and_then(|t| t.as_table()) {
            for (_cfg, t) in targets {
                for section in [
                    "dependencies",
                    "dev-dependencies",
                    "build-dependencies",
                ] {
                    collect_path_deps(t, section, &manifest, &mut findings);
                }
            }
        }
    }
    Ok(findings)
}

fn collect_path_deps(
    parent: &toml::Value,
    section: &str,
    manifest: &Path,
    out: &mut Vec<PathDepFinding>,
) {
    let Some(deps) = parent.get(section).and_then(|d| d.as_table()) else { return };
    for (name, v) in deps {
        let Some(t) = v.as_table() else { continue };
        let path = t.get("path").and_then(|p| p.as_str());
        let Some(p) = path else { continue };
        let version = t.get("version").and_then(|v| v.as_str()).map(|s| s.to_string());
        out.push(PathDepFinding {
            manifest: manifest.to_path_buf(),
            dep_name: name.clone(),
            dep_path: p.to_string(),
            has_version: version.is_some(),
            version_req: version,
        });
    }
}

fn walk_cargo_tomls(root: &Path) -> Result<Vec<std::path::PathBuf>> {
    let mut out = Vec::new();
    let mut stack = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let name = dir.file_name().and_then(|n| n.to_str()).unwrap_or("");
        if matches!(name, "target" | ".git" | "node_modules") { continue }
        for entry in std::fs::read_dir(&dir).with_context(|| format!("read_dir {}", dir.display()))? {
            let entry = entry?;
            let p = entry.path();
            let ft = entry.file_type()?;
            if ft.is_dir() {
                stack.push(p);
            } else if ft.is_file() && p.file_name().and_then(|n| n.to_str()) == Some("Cargo.toml") {
                out.push(p);
            }
        }
    }
    Ok(out)
}

/// One metadata-check row from [`crate_metadata_check`].
#[derive(Debug, Clone)]
pub struct CrateMetaCheck {
    pub manifest: std::path::PathBuf,
    pub crate_name: String,
    pub version: String,
    pub has_readme: bool,
    pub has_license: bool,
    pub license_expr: Option<String>,
    pub has_repository: bool,
    pub repository_url: Option<String>,
    pub has_description: bool,
    pub description_len: Option<usize>,
}

impl CrateMetaCheck {
    pub fn ok(&self) -> bool {
        self.has_readme && self.has_license && self.has_repository && self.has_description
    }
}

/// Gate (feature 4): every publishable crate's `[package]` table must
/// have `readme`, `license`, `repository`, `description` populated —
/// crates.io rejects missing/short metadata and we discovered this by
/// trial-and-error mid-publish (nornir 0.1.0 needed `readme=README.md`
/// added). Skips workspace virtual roots (no `[package]`) and any
/// manifest with `publish = false`.
pub fn crate_metadata_check(repo_root: &Path) -> Result<Vec<CrateMetaCheck>> {
    let mut out = Vec::new();
    for manifest in walk_cargo_tomls(repo_root)? {
        let text = std::fs::read_to_string(&manifest)
            .with_context(|| format!("read {}", manifest.display()))?;
        let doc: toml::Value = toml::from_str(&text)
            .with_context(|| format!("parse {}", manifest.display()))?;
        let Some(pkg) = doc.get("package").and_then(|p| p.as_table()) else { continue };
        if pkg.get("publish").and_then(|p| p.as_bool()) == Some(false) { continue }
        let crate_name = pkg.get("name").and_then(|n| n.as_str()).unwrap_or("?").to_string();
        let version = pkg.get("version").and_then(|v| v.as_str()).unwrap_or("0.0.0").to_string();
        let readme = pkg.get("readme");
        let license_expr = pkg.get("license").and_then(|v| v.as_str()).map(|s| s.to_string());
        let repo_url = pkg.get("repository").and_then(|v| v.as_str()).map(|s| s.to_string());
        let desc = pkg.get("description").and_then(|v| v.as_str());
        let readme_ok = match readme {
            Some(toml::Value::String(s)) => !s.is_empty(),
            Some(toml::Value::Boolean(b)) => *b,
            _ => false,
        };
        out.push(CrateMetaCheck {
            manifest: manifest.clone(),
            crate_name,
            version,
            has_readme: readme_ok,
            has_license: license_expr.as_ref().map(|s| !s.is_empty()).unwrap_or(false),
            license_expr,
            has_repository: repo_url.as_ref().map(|s| !s.is_empty()).unwrap_or(false),
            repository_url: repo_url,
            has_description: desc.map(|s| !s.is_empty()).unwrap_or(false),
            description_len: desc.map(|s| s.len()),
        });
    }
    Ok(out)
}

/// One link-declaration record from [`links_declarations_scan`].
#[derive(Debug, Clone)]
pub struct LinkDecl {
    pub crate_name: String,
    pub version: String,
    pub links_value: String,
    pub manifest: std::path::PathBuf,
}

/// One conflict surfaced by [`detect_links_conflicts`].
#[derive(Debug, Clone)]
pub struct LinksConflict {
    pub links_value: String,
    pub crates: Vec<(String, String)>, // (crate_name, version)
}

/// Gate (feature 8): collect every `links =` declaration in the
/// resolved dep tree via `cargo metadata`. Multiple crates with the
/// same `links` key cause cryptic linker failures at build time —
/// detect them up front.
pub fn links_declarations_scan(repo_root: &Path) -> Result<Vec<LinkDecl>> {
    let meta = cargo_metadata::MetadataCommand::new()
        .current_dir(repo_root)
        .exec()
        .with_context(|| format!("cargo metadata in {}", repo_root.display()))?;
    let mut out = Vec::new();
    for pkg in &meta.packages {
        if let Some(links) = &pkg.links {
            out.push(LinkDecl {
                crate_name: pkg.name.to_string(),
                version: pkg.version.to_string(),
                links_value: links.clone(),
                manifest: pkg.manifest_path.clone().into_std_path_buf(),
            });
        }
    }
    Ok(out)
}

pub fn detect_links_conflicts(decls: &[LinkDecl]) -> Vec<LinksConflict> {
    use std::collections::BTreeMap;
    let mut buckets: BTreeMap<String, Vec<(String, String)>> = BTreeMap::new();
    for d in decls {
        buckets.entry(d.links_value.clone())
            .or_default()
            .push((d.crate_name.clone(), d.version.clone()));
    }
    buckets.into_iter()
        .filter(|(_, c)| c.iter().map(|(n, _)| n).collect::<std::collections::HashSet<_>>().len() > 1)
        .map(|(links_value, crates)| LinksConflict { links_value, crates })
        .collect()
}

/// Gate (feature 16): the `v<version>` git tag for the most-recently
/// published crate version must exist locally and point at HEAD.
/// Run after a successful `cargo publish` to catch the "shipped to
/// crates.io but forgot to tag" failure mode.
///
/// Pure-Rust via [`crate::gitio`] (gix); annotated tags are peeled to
/// their target commit before the HEAD comparison.
pub fn git_tag_matches_published(repo_root: &Path, version: &str) -> Result<bool> {
    crate::gitio::tag_points_at_head(repo_root, &format!("v{version}"))
}

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

    #[test]
    fn path_dep_audit_flags_missing_version() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), r#"
[package]
name = "x"
version = "0.1.0"
edition = "2021"

[dependencies]
sibling-a = { path = "../a" }
sibling-b = { path = "../b", version = "0.2" }
"#).unwrap();
        let findings = path_dep_audit(dir.path()).unwrap();
        assert_eq!(findings.len(), 2);
        let bad = findings.iter().find(|f| f.dep_name == "sibling-a").unwrap();
        assert!(!bad.ok());
        let good = findings.iter().find(|f| f.dep_name == "sibling-b").unwrap();
        assert!(good.ok());
        assert_eq!(good.version_req.as_deref(), Some("0.2"));
    }

    #[test]
    fn crate_metadata_check_flags_missing_fields() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), r#"
[package]
name = "y"
version = "0.1.0"
edition = "2021"
license = "MIT"
"#).unwrap();
        let checks = crate_metadata_check(dir.path()).unwrap();
        assert_eq!(checks.len(), 1);
        let c = &checks[0];
        assert!(c.has_license);
        assert!(!c.has_repository);
        assert!(!c.has_description);
        assert!(!c.ok());
    }
}

#[cfg(test)]
mod regression_tests {
    use super::*;
    use crate::bench::{BenchResult, BenchRun};

    fn run(machine: &str, ts: &str, mbs: f64) -> BenchRun {
        let mut r = BenchResult { name: "codec".into(), metrics: Default::default() };
        r.metrics.insert("throughput_mbs".into(), serde_json::Value::from(mbs));
        BenchRun {
            date: ts[..10].to_string(),
            timestamp: Some(ts.to_string()),
            version: "0.1.0".into(),
            machine: machine.into(),
            cores: 1,
            results: vec![r],
            tests: Vec::new(),
        }
    }

    #[test]
    fn no_baseline_for_machine_passes() {
        // First-ever run for a machine: nothing to regress against.
        let candidate = run("ci", "2026-01-02T00:00:00+00:00", 100.0);
        let history = vec![run("other-box", "2026-01-01T00:00:00+00:00", 999.0)];
        assert!(no_regression_against(&candidate, &history, 10.0).is_ok());
    }

    #[test]
    fn improvement_passes() {
        let candidate = run("ci", "2026-01-02T00:00:00+00:00", 120.0);
        let history = vec![run("ci", "2026-01-01T00:00:00+00:00", 100.0)];
        assert!(no_regression_against(&candidate, &history, 10.0).is_ok());
    }

    #[test]
    fn drop_within_threshold_passes() {
        // 5% drop, threshold 10% → ok.
        let candidate = run("ci", "2026-01-02T00:00:00+00:00", 95.0);
        let history = vec![run("ci", "2026-01-01T00:00:00+00:00", 100.0)];
        assert!(no_regression_against(&candidate, &history, 10.0).is_ok());
    }

    #[test]
    fn drop_beyond_threshold_rejected() {
        // 20% drop, threshold 10% → regression.
        let candidate = run("ci", "2026-01-02T00:00:00+00:00", 80.0);
        let history = vec![run("ci", "2026-01-01T00:00:00+00:00", 100.0)];
        let err = no_regression_against(&candidate, &history, 10.0).unwrap_err();
        assert!(err.to_string().contains("regression"), "got: {err}");
    }

    #[test]
    fn baseline_is_newest_by_timestamp_not_slice_order() {
        // History deliberately NOT in chronological order. The newest
        // (by timestamp) baseline is 100 mbs; candidate 90 is a 10% drop,
        // which is NOT > 10% threshold → ok. If the picker wrongly used
        // the LAST slice element (the 50-mbs older run), 90 would look
        // like an 80% *improvement* and also pass — so to truly exercise
        // ordering we set the OLDER run much higher.
        let candidate = run("ci", "2026-03-01T00:00:00+00:00", 90.0);
        let history = vec![
            run("ci", "2026-02-01T00:00:00+00:00", 100.0), // newest → baseline
            run("ci", "2026-01-01T00:00:00+00:00", 1000.0), // oldest, last in slice
        ];
        // Against newest(100): 10% drop, threshold 10% → ok.
        assert!(no_regression_against(&candidate, &history, 10.0).is_ok());
        // Against the (wrong) last-in-slice 1000: 91% drop → would error.
        // So passing here proves we picked the newest by timestamp.
    }

    #[test]
    fn baseline_picks_newest_even_when_older_run_is_last_in_slice_and_regressed() {
        // Newest baseline = 100; candidate 50 = 50% drop → must reject,
        // regardless of an older, slower run appearing later in the slice.
        let candidate = run("ci", "2026-03-01T00:00:00+00:00", 50.0);
        let history = vec![
            run("ci", "2026-02-01T00:00:00+00:00", 100.0),
            run("ci", "2026-01-01T00:00:00+00:00", 40.0),
        ];
        assert!(no_regression_against(&candidate, &history, 10.0).is_err());
    }
}

#[cfg(test)]
mod nexus_floor_tests {
    use super::*;
    use crate::bench::{BenchResult, BenchRun};

    /// One bench result carrying an arbitrary set of `key=value` f64 metrics.
    fn result(name: &str, kv: &[(&str, f64)]) -> BenchResult {
        let mut r = BenchResult { name: name.into(), metrics: Default::default() };
        for (k, v) in kv {
            r.metrics.insert((*k).to_string(), serde_json::Value::from(*v));
        }
        r
    }

    fn run_with(results: Vec<BenchResult>) -> BenchRun {
        BenchRun {
            date: "2026-01-01".into(),
            timestamp: Some("2026-01-01T00:00:00+00:00".into()),
            version: "0.1.0".into(),
            machine: "ci".into(),
            cores: 1,
            results,
            tests: Vec::new(),
        }
    }

    #[test]
    fn holger_above_nexus_passes() {
        // holger_ops_sec (5000) ≥ nexus_ops_sec (1000) → floor held.
        let run = run_with(vec![result(
            "decode",
            &[("holger_ops_sec", 5000.0), ("nexus_ops_sec", 1000.0)],
        )]);
        assert!(nexus_floor(&run).is_ok());
    }

    #[test]
    fn holger_equal_to_nexus_passes() {
        // Boundary: equal throughput is NOT below the floor (`h < n` is strict).
        let run = run_with(vec![result(
            "decode",
            &[("holger_ops_sec", 1000.0), ("nexus_ops_sec", 1000.0)],
        )]);
        assert!(nexus_floor(&run).is_ok());
    }

    #[test]
    fn holger_below_nexus_rejected() {
        // holger (900) < nexus (1000) → release-blocking floor breach.
        let run = run_with(vec![result(
            "decode",
            &[("holger_ops_sec", 900.0), ("nexus_ops_sec", 1000.0)],
        )]);
        let err = nexus_floor(&run).unwrap_err().to_string();
        assert!(err.contains("nexus floor"), "got: {err}");
        assert!(err.contains("decode"), "error should name the result: {err}");
    }

    #[test]
    fn one_result_below_floor_fails_the_whole_run() {
        // First result is fine; second breaches → the gate must reject the run.
        let run = run_with(vec![
            result("warm", &[("holger_ops_sec", 5000.0), ("nexus_ops_sec", 1000.0)]),
            result("cold", &[("holger_ops_sec", 50.0), ("nexus_ops_sec", 1000.0)]),
        ]);
        let err = nexus_floor(&run).unwrap_err().to_string();
        assert!(err.contains("cold"), "should blame the breaching result: {err}");
    }

    #[test]
    fn results_missing_either_key_are_skipped() {
        // A znippy-shaped run (compress_mbs / decompress_mbs, no holger/nexus
        // keys) carries neither floor metric → the gate has nothing to check
        // and passes rather than erroring on the absent keys.
        let run = run_with(vec![
            result("compress", &[("compress_mbs", 800.0)]),
            result("only_holger", &[("holger_ops_sec", 10.0)]),
            result("only_nexus", &[("nexus_ops_sec", 9999.0)]),
        ]);
        assert!(nexus_floor(&run).is_ok());
    }
}

#[cfg(test)]
mod roundtrip_tests {
    use super::*;
    use std::cell::RefCell;

    #[test]
    fn all_kinds_succeed_passes() {
        // Every kind's closure returns Ok → the gate passes and runs each once.
        let seen = RefCell::new(Vec::new());
        let res = integration_roundtrip(&["blob", "symbol"], |k| {
            seen.borrow_mut().push(k.to_string());
            Ok(())
        });
        assert!(res.is_ok());
        assert_eq!(seen.into_inner(), vec!["blob", "symbol"]);
    }

    #[test]
    fn empty_kind_list_passes_vacuously() {
        // No configured kinds ⇒ nothing to round-trip ⇒ Ok.
        let res = integration_roundtrip(&[], |_| -> Result<()> {
            panic!("closure must not run for an empty kind list")
        });
        assert!(res.is_ok());
    }

    #[test]
    fn first_failure_aborts_and_is_contextualised() {
        // The closure fails on the second kind; the gate must propagate the
        // error, tag it with the failing kind, and NOT invoke later kinds.
        let seen = RefCell::new(Vec::new());
        let res = integration_roundtrip(&["blob", "symbol", "edge"], |k| {
            seen.borrow_mut().push(k.to_string());
            if k == "symbol" {
                anyhow::bail!("store rejected the artifact")
            }
            Ok(())
        });
        let err = res.unwrap_err();
        let chain = format!("{err:#}");
        assert!(chain.contains("roundtrip failed for symbol"), "got: {chain}");
        assert!(chain.contains("store rejected the artifact"), "got: {chain}");
        // "edge" must never have been attempted after the failure.
        assert_eq!(seen.into_inner(), vec!["blob", "symbol"]);
    }
}