nornir 0.5.3

//! `nornir release promote` — the irreversible tier: bump → strip the arrow-58
//! `[patch]` forks → publish the unblocked crates to crates.io, deps-first, waiting
//! for each to index before its dependents, and recording every mutating step to the
//! reversible `release_changes` ledger so `nornir release undo` can un-bump + yank.
//!
//! Symmetric with `release stage` (the `/sparring` rehearsal) but for the real
//! registry. The promote gate ([`super::doctor::compute_promote_block`]) holds every
//! crate that transitively rides a forked dep (arrow-58 `iceberg`), so only genuinely
//! crates.io-publishable crates ship.
//!
//! Safety model:
//!   * BUMP is applied + committed on the working branch (it persists; `undo` reverses
//!     it), so the published versions are real and in git history.
//!   * the `[patch.crates-io]` STRIP lives only on an ephemeral release branch that is
//!     deleted on return ([`super::stage::BranchGuard`]) — the dev tree keeps its fork.
//!   * a `--dry-run` mutates NOTHING: it prints the bump + publish plan and runs
//!     `cargo publish --dry-run`.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Command;

use anyhow::{Context, Result};

use crate::release::cargo::{self, BumpLevel};
use crate::release::doctor;
use crate::release::publish::{self, PublishOutcome};
use crate::release::registry::{CratesIoBackend, RegistryBackend};
use crate::release::source_drift::{self, CratesIoSources, DriftVerdict};
use crate::release::stage::BranchGuard;
use crate::warehouse::iceberg::IcebergWarehouse;
use crate::warehouse::release_changes::ChangeRecorder;

/// Knobs for a promote run.
#[derive(Debug, Clone)]
pub struct PromoteOpts {
    /// Bump every publishable crate by this level before publishing. `None` publishes
    /// the current versions (which crates.io rejects if already uploaded — so a real
    /// promote almost always wants `Some`).
    pub bump: Option<BumpLevel>,
    /// After publishing a crate, poll crates.io until it's the indexed `max_version`
    /// before publishing its dependents (avoids resolve failures on propagation lag).
    pub wait_index: bool,
    /// Plan only — print the bump + publish plan, run `cargo publish --dry-run`, mutate
    /// nothing (no bump, no branch, no upload).
    pub dry_run: bool,
    /// Ephemeral branch the `[patch]` strip lives on (deleted on return).
    pub branch: String,
    /// Seconds to wait for each crate to index (when `wait_index`).
    pub wait_secs: u64,
    /// SKIP re-publishing crates whose local source is byte-identical to the copy
    /// already published (content-diff against the crates.io tarball). Ends the
    /// rate-limit grind of re-bumping large UNCHANGED crate sets (the facett
    /// "timeout"). Costs one CDN fetch + `cargo package --list` per publishable
    /// crate, so it's OFF by default and enabled by the doctor cascade.
    pub skip_unchanged: bool,
    /// SCOPE the publish set to EXACTLY these crate names (intersected with the
    /// repo's publishable set). `None` = every publishable crate in the repo (the
    /// default). Set by the doctor cascade to honour `--only`/the smart selection:
    /// the cascade has already decided the exact per-repo crate set, so a repo that
    /// produces several crates publishes only the ones actually selected.
    pub only_crates: Option<std::collections::BTreeSet<String>>,
}

impl Default for PromoteOpts {
    fn default() -> Self {
        PromoteOpts {
            bump: Some(BumpLevel::Patch),
            wait_index: true,
            dry_run: true,
            branch: "release/promote".to_string(),
            wait_secs: 300,
            skip_unchanged: false,
            only_crates: None,
        }
    }
}

/// One planned crate version change.
#[derive(Debug, Clone)]
pub struct PlannedBump {
    pub crate_name: String,
    pub old_version: String,
    pub new_version: String,
}

/// The result of a promote run (or its dry-run plan).
#[derive(Debug, Default)]
pub struct PromoteReport {
    /// Crates that will be / were version-bumped.
    pub bumps: Vec<PlannedBump>,
    /// Crates NOT bumped because their source matches the published copy (skip-
    /// unchanged) — they resolve `AlreadyPublished` in the publish phase.
    pub skipped_unchanged: Vec<String>,
    /// Crates held from crates.io by the promote gate (fork riders + dependents).
    pub held: Vec<String>,
    /// The publish phases (deps-first) actually attempted, crate → outcome.
    pub published: Vec<(String, PublishOutcome)>,
    /// `(crate, waited_ms)` for crates we waited to index.
    pub waited: Vec<(String, u64)>,
    /// Non-fatal per-step problems.
    pub errors: Vec<String>,
    /// Whether this was a dry-run (no mutation).
    pub dry_run: bool,
}

fn git(path: &Path, args: &[&str]) -> Result<()> {
    let st = Command::new("git")
        .arg("-C")
        .arg(path)
        .args(args)
        .status()
        .with_context(|| format!("git {}", args.join(" ")))?;
    if !st.success() {
        anyhow::bail!("git {} failed in {}", args.join(" "), path.display());
    }
    Ok(())
}

fn git_head(path: &Path) -> String {
    Command::new("git")
        .arg("-C")
        .arg(path)
        .args(["rev-parse", "HEAD"])
        .output()
        .ok()
        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
        .unwrap_or_default()
}

/// Compute the set of publishable crates that can SKIP re-publishing because their
/// source is byte-identical to the copy already on crates.io. Empty when `enabled`
/// is false (no network at all). Correctness rules:
///   * only skip a crate we can PROVE unchanged vs its published-max (see
///     [`source_drift::can_skip_republish`]); a first publish, a drift, or an
///     inconclusive verdict ⇒ NOT skipped (it bumps + publishes);
///   * UNIFIED-versioning workspaces (every crate shares `[workspace.package].version`,
///     e.g. facett) skip ALL-OR-NOTHING — a shared version cannot ship some members
///     old and some new — so the skip is honored only when EVERY publishable crate
///     is unchanged (otherwise the shared version bumps and all members republish).
fn plan_skip_unchanged(
    repo_root: &Path,
    all_pub: &[(String, String)],
    publishable: &std::collections::BTreeSet<String>,
    enabled: bool,
) -> std::collections::BTreeSet<String> {
    let mut skip = std::collections::BTreeSet::new();
    if !enabled {
        return skip;
    }
    let backend = CratesIoBackend;
    for (name, _cur) in all_pub.iter().filter(|(n, _)| publishable.contains(n)) {
        let pmax = backend.published_max_version(name);
        let verdict = match (pmax.as_deref(), cargo::crate_dir(repo_root, name)) {
            (Some(v), Some(dir)) => source_drift::assess(&CratesIoSources, &dir, name, v),
            (None, _) => DriftVerdict::Unknown("first publish".to_string()),
            (_, None) => DriftVerdict::Unknown("crate dir not found".to_string()),
        };
        if source_drift::can_skip_republish(pmax.as_deref(), &verdict) {
            skip.insert(name.clone());
        }
    }
    let unified = cargo::workspace_package_version(repo_root)
        .ok()
        .flatten()
        .is_some_and(|(_, wsv)| {
            all_pub.iter().filter(|(n, _)| publishable.contains(n)).all(|(_, cur)| *cur == wsv)
        });
    let publishable_names: Vec<String> =
        all_pub.iter().filter(|(n, _)| publishable.contains(n)).map(|(n, _)| n.clone()).collect();
    resolve_unified_skip(skip, &publishable_names, unified)
}

/// Apply the UNIFIED-workspace all-or-nothing rule to a per-crate skippable set.
/// Unified ⇒ the skip is honored only if EVERY publishable crate is skippable
/// (a shared version can't ship some members old and some new); otherwise the
/// per-crate set is returned unchanged. Pure — the unit-tested core of caveat #2.
fn resolve_unified_skip(
    per_crate_skip: std::collections::BTreeSet<String>,
    publishable: &[String],
    unified: bool,
) -> std::collections::BTreeSet<String> {
    if unified && !publishable.iter().all(|n| per_crate_skip.contains(n)) {
        std::collections::BTreeSet::new()
    } else {
        per_crate_skip
    }
}

/// Run (or dry-run) a promote. `recorder` carries the warehouse + ledger recorder so
/// the bump/publish steps are reversible; pass `None` to skip recording (e.g. tests).
pub fn run(
    repo_root: &Path,
    gate_repos: &[(String, PathBuf)],
    repo_name: &str,
    opts: &PromoteOpts,
    recorder: Option<(&IcebergWarehouse, &ChangeRecorder)>,
) -> Result<PromoteReport> {
    let mut report = PromoteReport { dry_run: opts.dry_run, ..Default::default() };

    // 1 — gate: which crates are crates.io-publishable vs held by a fork. `gate_repos`
    // is the CONFIGURED constellation (the same set `release doctor` scans) — NOT a
    // filesystem sweep, which would slurp the fork dirs (`iceberg-arrow58` produces a
    // crate literally named `iceberg`) and make the gate think the fork is "ours".
    let block = doctor::compute_promote_block(
        gate_repos.iter().map(|(n, p)| (n.clone(), p.as_path())),
    );
    report.held = block.blocked.iter().cloned().collect();

    // 2 — the publishable crate set: every publishable [package] (robust toml-walk,
    //     no cargo metadata) MINUS the gate-held crates.
    let all_pub = cargo::publishable_crate_versions(repo_root)?;
    let publishable: std::collections::BTreeSet<String> = all_pub
        .iter()
        .map(|(n, _)| n.clone())
        .filter(|n| !block.blocked.contains(n))
        // `--only`/smart-selection scope: when the cascade named an explicit crate
        // set, publish EXACTLY that intersection (a multi-crate repo ships only its
        // selected crates). `None` keeps every publishable crate (the default).
        .filter(|n| opts.only_crates.as_ref().is_none_or(|only| only.contains(n)))
        .collect();

    // 3 — publish order (deps-first), held crates dropped. `derive_publish_order` needs
    //     `cargo metadata`; if that's unavailable, fall back to one unordered phase so
    //     the plan still works (sequencing only matters for the real upload).
    let order = publish::derive_publish_order(repo_root).unwrap_or_default();
    let mut filtered: Vec<Vec<String>> = order
        .iter()
        .map(|phase| {
            phase.iter().filter(|k| publishable.contains(*k)).cloned().collect::<Vec<_>>()
        })
        .filter(|p: &Vec<String>| !p.is_empty())
        .collect();
    if filtered.is_empty() && !publishable.is_empty() {
        filtered = vec![publishable.iter().cloned().collect()];
    }

    // 4 — plan the bumps for every publishable crate (independent of cargo metadata).
    let mut next_ver: BTreeMap<String, String> = BTreeMap::new();
    if let Some(level) = opts.bump {
        // SKIP-UNCHANGED (the facett-grind fix): a crate whose local source is
        // byte-identical to its published copy needs no new version — leave it live
        // as-is so the publish phase resolves it `AlreadyPublished` (step 6) instead
        // of grinding a gratuitous re-bump + re-publish through the rate-limiter.
        let skip = plan_skip_unchanged(repo_root, &all_pub, &publishable, opts.skip_unchanged);
        for name in &skip {
            report.skipped_unchanged.push(name.clone());
            println!("  · skip {name}: local source matches the published copy — not bumping");
        }
        for (name, cur) in &all_pub {
            if !publishable.contains(name) || skip.contains(name) {
                continue;
            }
            let nv = cargo::next_version(cur, level)?;
            next_ver.insert(name.clone(), nv.clone());
            report.bumps.push(PlannedBump {
                crate_name: name.clone(),
                old_version: cur.clone(),
                new_version: nv,
            });
        }
    }

    // DRY-RUN: just the plan (bumps + gate + deps-first order), mutate nothing and
    // DON'T compile — a fast, useful preview. (`cargo publish` validates packaging on
    // the real run.) Each crate is reported as a would-publish `DryRun`.
    if opts.dry_run {
        for phase in &filtered {
            for krate in phase {
                report.published.push((krate.clone(), PublishOutcome::DryRun));
            }
        }
        return Ok(report);
    }

    // ── REAL promote (mutating, irreversible upload) ───────────────────────────
    // 4 — apply + commit the bumps on the working branch (persist; `undo` reverses).
    if !report.bumps.is_empty() {
        let level = opts.bump.expect("bumps non-empty ⇒ a level was set");
        // Unified-versioning workspace (facett): bump [workspace.package].version ONCE
        // — `version.workspace = true` members all inherit it. Detected when every
        // publishable crate shares the workspace version. Otherwise (znippy: each crate
        // its own number) bump per-crate.
        let ws = cargo::workspace_package_version(repo_root)?;
        let unified = ws
            .as_ref()
            .filter(|(_, old)| report.bumps.iter().all(|b| &b.old_version == old));
        if let Some((_, old_ws)) = unified {
            let new_ws = cargo::next_version(old_ws, level)?;
            // Applies the edit AND returns each manifest's pre-bump text.
            let edited = cargo::bump_workspace_version(repo_root, old_ws, &new_ws)
                .with_context(|| format!("bump workspace version {old_ws} → {new_ws}"))?;
            if let Some((wh, rec)) = recorder {
                crate::release::undo::record_applied_bump(
                    wh,
                    rec,
                    repo_name,
                    "[workspace.package]",
                    old_ws,
                    &new_ws,
                    &edited,
                )
                .context("record workspace bump for undo")?;
            }
        } else {
            for b in &report.bumps {
                // bump_consumers=FALSE: only bump each crate's OWN version, never rewrite
                // a consumer's dependency requirement. Pinning a consumer to the exact new
                // version (e.g. `ljar = "0.2.3"`) breaks workspace resolution — that version
                // isn't on crates.io yet, and a version-only (no-path) dep has no local
                // fallback, so EVERY publish in the workspace fails to resolve. The existing
                // `^0.2.2`-style reqs already admit the new patch, so leaving them is correct
                // SemVer and lets the deps-first cascade publish cleanly.
                let plan =
                    cargo::plan_version_bump(repo_root, &b.crate_name, &b.new_version, false)
                        .with_context(|| format!("plan bump {} → {}", b.crate_name, b.new_version))?;
                if plan.edits.is_empty() {
                    continue;
                }
                // record_bump APPLIES + records (don't also apply, or it double-bumps).
                if let Some((wh, rec)) = recorder {
                    crate::release::undo::record_bump(wh, rec, repo_name, &plan)
                        .context("record bump for undo")?;
                } else {
                    cargo::apply_bump_plan(&plan)
                        .with_context(|| format!("apply bump {} → {}", b.crate_name, b.new_version))?;
                }
            }
        }
        git(repo_root, &["add", "-A"])?;
        git(
            repo_root,
            &["commit", "-q", "-m", "chore(release): bump versions for crates.io promote"],
        )?;
    }

    // 5 — cut the ephemeral release branch (auto-restored on return) and strip the
    //     [patch.crates-io] forks ON it, committed so the strip never leaks back.
    let mut guard = BranchGuard::new(&opts.branch);
    guard.capture(repo_root);
    let prev_head = git_head(repo_root);
    git(repo_root, &["checkout", "-B", &opts.branch])?;
    if let Some((wh, rec)) = recorder {
        let _ = crate::release::undo::record_branch(wh, rec, repo_name, &opts.branch, &prev_head);
    }
    let (files, blocks) = crate::release::cargo::ensure_no_patch_crates_io(repo_root)?;
    if blocks > 0 {
        git(repo_root, &["add", "-A"])?;
        git(
            repo_root,
            &["commit", "-q", "-m", "chore(release): strip [patch.crates-io] for publish (ephemeral)"],
        )?;
        eprintln!("promote: stripped {blocks} [patch.crates-io] block(s) across {files} Cargo.toml(s) (ephemeral branch)");
    }

    // 6 — publish deps-first; record each, then wait for it to index.
    let registry = "crates.io";
    for phase in &filtered {
        for krate in phase {
            match publish::run_cargo_publish(repo_root, krate, false) {
                Ok(outcome) => {
                    report.published.push((krate.clone(), outcome.clone()));
                    let ver = next_ver.get(krate).cloned();
                    if matches!(outcome, PublishOutcome::Published) {
                        if let (Some((wh, rec)), Some(v)) = (recorder, ver.as_ref()) {
                            crate::release::undo::record_publish(
                                wh, rec, repo_name, krate, v, registry, true,
                            )
                            .context("record publish for undo")?;
                        }
                        if opts.wait_index {
                            if let Some(v) = ver.as_ref() {
                                match publish::wait_for_index(
                                    krate,
                                    v,
                                    std::time::Duration::from_secs(opts.wait_secs),
                                ) {
                                    Ok(ms) => report.waited.push((krate.clone(), ms)),
                                    Err(e) => report
                                        .errors
                                        .push(format!("{krate}: wait-for-index: {e}")),
                                }
                            }
                        }
                    }
                }
                Err(e) => report.errors.push(format!("{krate}: {e}")),
            }
        }
    }

    // guard drops here → working branch restored (bump kept, strip dropped with branch).
    Ok(report)
}

/// Human-readable promote report.
pub fn format_report(r: &PromoteReport) -> String {
    let mut s = String::new();
    let head = if r.dry_run { "nornir release promote — PLAN (dry-run)" } else { "nornir release promote — executed" };
    s.push_str(head);
    s.push('\n');
    if !r.bumps.is_empty() {
        s.push_str("\n  Version bumps:\n");
        for b in &r.bumps {
            s.push_str(&format!("    {} {} → {}\n", b.crate_name, b.old_version, b.new_version));
        }
    }
    if !r.skipped_unchanged.is_empty() {
        s.push_str(&format!(
            "\n  ⏭ skipped {} unchanged crate(s) (source matches published, not re-bumped): {}\n",
            r.skipped_unchanged.len(),
            r.skipped_unchanged.join(", ")
        ));
    }
    if !r.held.is_empty() {
        s.push_str(&format!("\n  ⏸ held from crates.io ({}): {}\n", r.held.len(), r.held.join(", ")));
    }
    s.push_str("\n  Publish (deps-first):\n");
    for (k, o) in &r.published {
        s.push_str(&format!("    {k:40} {o:?}\n"));
    }
    if !r.waited.is_empty() {
        s.push_str("\n  Indexed:\n");
        for (k, ms) in &r.waited {
            s.push_str(&format!("    {k} visible after {ms} ms\n"));
        }
    }
    if r.errors.is_empty() {
        if r.dry_run {
            s.push_str("\n  → plan only; re-run without --dry-run to bump + publish to crates.io.\n");
        } else {
            s.push_str("\n  ✅ promote complete (working tree restored; `release undo` reverses bump + yanks).\n");
        }
    } else {
        for e in &r.errors {
            s.push_str(&format!("  ⚠ {e}\n"));
        }
    }
    s
}

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

    fn set(items: &[&str]) -> std::collections::BTreeSet<String> {
        items.iter().map(|s| s.to_string()).collect()
    }

    #[test]
    fn unified_skip_is_all_or_nothing() {
        let pubs = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        // Unified + EVERY crate skippable → keep the whole set.
        assert_eq!(
            resolve_unified_skip(set(&["a", "b", "c"]), &pubs, true),
            set(&["a", "b", "c"])
        );
        // Unified + only SOME skippable → clear (shared version must bump together).
        assert!(resolve_unified_skip(set(&["a", "b"]), &pubs, true).is_empty());
    }

    #[test]
    fn per_crate_skip_is_kept_as_is() {
        let pubs = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        // Not unified → per-crate partial skip is honored verbatim.
        assert_eq!(resolve_unified_skip(set(&["a", "c"]), &pubs, false), set(&["a", "c"]));
    }

    #[test]
    fn skip_disabled_yields_empty() {
        use std::path::Path;
        // enabled=false ⇒ no network, empty skip set regardless of inputs.
        let got = plan_skip_unchanged(
            Path::new("/nonexistent"),
            &[("a".to_string(), "1.0.0".to_string())],
            &set(&["a"]),
            false,
        );
        assert!(got.is_empty());
    }

    /// Real-world E2E: `plan_skip_unchanged` over a repo whose crate(s) are live and
    /// unchanged should skip them. Opt-in via NORNIR_DRIFT_SMOKE_DIR/_KRATE (a
    /// single-crate repo checkout matching its published version); skips if unset.
    #[test]
    #[ignore = "network + local checkout; run with NORNIR_DRIFT_SMOKE_DIR/_KRATE set"]
    fn e2e_skip_unchanged_live_crate() {
        let (Ok(dir), Ok(krate)) = (
            std::env::var("NORNIR_DRIFT_SMOKE_DIR"),
            std::env::var("NORNIR_DRIFT_SMOKE_KRATE"),
        ) else {
            eprintln!("skipped: set NORNIR_DRIFT_SMOKE_DIR/_KRATE");
            return;
        };
        let all_pub = crate::release::cargo::publishable_crate_versions(std::path::Path::new(&dir))
            .expect("read publishable crates");
        let publishable: std::collections::BTreeSet<String> =
            all_pub.iter().map(|(n, _)| n.clone()).collect();
        let skip = plan_skip_unchanged(std::path::Path::new(&dir), &all_pub, &publishable, true);
        assert!(skip.contains(&krate), "expected {krate} to be skipped (unchanged), got {skip:?}");
    }

    #[test]
    fn next_version_levels() {
        use crate::release::cargo::{next_version, BumpLevel};
        assert_eq!(next_version("1.2.3", BumpLevel::Patch).unwrap(), "1.2.4");
        assert_eq!(next_version("1.2.3", BumpLevel::Minor).unwrap(), "1.3.0");
        assert_eq!(next_version("1.2.3", BumpLevel::Major).unwrap(), "2.0.0");
        assert_eq!(next_version("0.4.52", BumpLevel::Patch).unwrap(), "0.4.53");
        // pre/build suffixes are dropped before incrementing.
        assert_eq!(next_version("1.0.0-rc.1", BumpLevel::Patch).unwrap(), "1.0.1");
    }

    #[test]
    fn dry_run_plans_bumps_and_holds_without_mutating() {
        // A tiny one-crate workspace; no forks → nothing held, one patch bump planned.
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join("Cargo.toml"),
            "[package]\nname = \"solo\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
        )
        .unwrap();
        let opts = PromoteOpts { dry_run: true, wait_index: false, ..Default::default() };
        let gate = vec![("solo".to_string(), root.to_path_buf())];
        let report = run(root, &gate, "solo", &opts, None).unwrap();
        assert!(report.dry_run);
        assert!(report.held.is_empty(), "no forks → nothing held");
        let bump = report.bumps.iter().find(|b| b.crate_name == "solo").unwrap();
        assert_eq!(bump.new_version, "0.1.1");
        // dry-run must not have touched the manifest.
        let txt = std::fs::read_to_string(root.join("Cargo.toml")).unwrap();
        assert!(txt.contains("version = \"0.1.0\""), "dry-run left version unchanged");
    }
}