nornir 0.4.44

//! Publish-order walker. Per-repo TOML defines `publish_order` as
//! `Vec<Vec<String>>`: outer entries are sequential phases, inner
//! entries are independent and may be published in parallel.
//!
//! Tag creation is done in-process via the pure-Rust `gix` crate (no
//! `git` shellout). Tag *pushing* over the network is deferred to a
//! follow-up that wires `gix` network features — for now we print the
//! exact `git push` invocation so a human or CI can complete the step.
//!
//! `cargo publish` itself remains a `cargo` subprocess because cargo
//! owns crate packaging (sources → `.crate` tarball → upload). That is
//! cargo's domain, not a script we can replace with a few lines of
//! HTTP. Isolated to [`run_cargo_publish`] and clearly annotated.

use std::path::Path;
use std::process::Command;

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

/// Outcome reported per crate so the orchestrator (and progress
/// writer) can distinguish "really uploaded" from "already on the
/// registry" — re-running a release of a tree that's partially-shipped
/// is a routine case, not an error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublishOutcome {
    Published,
    AlreadyPublished,
    DryRun,
}

/// Walk the publish order, invoking cargo to publish each crate.
/// `dry_run` adds `--dry-run`. Returns the per-crate outcomes in the
/// same order they were published (outer phase order, inner declared
/// order). Errors only on real cargo failures — "crate already on the
/// registry" downgrades to [`PublishOutcome::AlreadyPublished`].
pub fn publish_all(
    repo_root: &Path,
    order: &[Vec<String>],
    dry_run: bool,
) -> Result<Vec<(String, PublishOutcome)>> {
    // No `publish_order` configured → DERIVE it from the workspace dependency
    // graph (topological, deps-first) instead of silently publishing nothing
    // (the bug that made `nornir release publish` exit 0 having uploaded nothing).
    // The manual `publish_order` is now an OVERRIDE, not a requirement.
    let derived;
    let order: &[Vec<String>] = if order.iter().all(|phase| phase.is_empty()) {
        derived = derive_publish_order(repo_root).with_context(|| {
            "no `publish_order` configured and could not derive one from `cargo metadata`"
        })?;
        if derived.iter().all(|p| p.is_empty()) {
            anyhow::bail!(
                "nothing to publish: no publishable crates in the workspace at {} \
                 (all `publish = false`?). Set `publish_order` in nornir.toml to override.",
                repo_root.display()
            );
        }
        &derived
    } else {
        order
    };
    let mut out: Vec<(String, PublishOutcome)> = Vec::new();
    for phase in order {
        for krate in phase {
            let outcome = run_cargo_publish(repo_root, krate, dry_run)
                .with_context(|| format!("cargo publish -p {krate}"))?;
            out.push((krate.clone(), outcome));
        }
    }
    Ok(out)
}

/// Deterministically derive the publish order from the workspace dependency
/// graph via `cargo metadata`: a topological sort (deps first) of the
/// **publishable** workspace members, grouped into phases (crates with no
/// remaining intra-workspace dependency publish together). Always possible for a
/// valid workspace — a publishable dep graph is a DAG (cargo forbids cycles); the
/// only filter is dropping `publish = false` crates. Fully deterministic: members
/// and phase contents are alphabetically ordered.
pub fn derive_publish_order(repo_root: &Path) -> Result<Vec<Vec<String>>> {
    let meta = cargo_metadata::MetadataCommand::new()
        .current_dir(repo_root)
        .exec()
        .with_context(|| format!("cargo metadata in {}", repo_root.display()))?;
    let ws: std::collections::BTreeSet<String> =
        meta.workspace_packages().iter().map(|p| p.name.to_string()).collect();
    let mut members: Vec<(String, Vec<String>)> = Vec::new();
    for p in meta.workspace_packages() {
        // `publish = false` serializes as `Some([])` → unpublishable, skip it.
        if matches!(&p.publish, Some(v) if v.is_empty()) {
            continue;
        }
        let deps: Vec<String> = p
            .dependencies
            .iter()
            .map(|d| d.name.clone())
            .filter(|n| ws.contains(n))
            .collect();
        members.push((p.name.to_string(), deps));
    }
    Ok(topo_phases(members))
}

/// Pure, deterministic Kahn topological sort into phases (deps-first). Each
/// `(name, deps)` lists the crate's intra-workspace dependencies. A crate is
/// "ready" when none of its workspace deps remain; ready crates form one phase
/// (alphabetical). Robust to a cycle (shouldn't occur for a publishable
/// workspace): the unresolved remainder is appended as a final phase so nothing
/// is silently dropped.
pub(crate) fn topo_phases(members: Vec<(String, Vec<String>)>) -> Vec<Vec<String>> {
    use std::collections::{BTreeMap, BTreeSet};
    let names: BTreeSet<String> = members.iter().map(|(n, _)| n.clone()).collect();
    let mut indeg: BTreeMap<String, usize> = BTreeMap::new();
    let mut dependents: BTreeMap<String, Vec<String>> = BTreeMap::new(); // dep → its dependents
    for (n, deps) in &members {
        let real: Vec<&String> =
            deps.iter().filter(|d| names.contains(*d) && d.as_str() != n).collect();
        indeg.insert(n.clone(), real.len());
        for d in real {
            dependents.entry(d.clone()).or_default().push(n.clone());
        }
    }
    let mut phases: Vec<Vec<String>> = Vec::new();
    let mut remaining: BTreeSet<String> = names;
    while !remaining.is_empty() {
        let ready: Vec<String> = remaining
            .iter()
            .filter(|n| indeg.get(*n).copied().unwrap_or(0) == 0)
            .cloned()
            .collect();
        if ready.is_empty() {
            phases.push(remaining.iter().cloned().collect()); // cycle guard — drop nothing
            break;
        }
        for n in &ready {
            remaining.remove(n);
            if let Some(deps) = dependents.get(n) {
                for dep in deps {
                    if let Some(e) = indeg.get_mut(dep) {
                        *e = e.saturating_sub(1);
                    }
                }
            }
        }
        phases.push(ready); // already alphabetical (from the BTreeSet iteration)
    }
    phases
}

/// The single intentional `cargo` subprocess. Cargo owns crate
/// packaging+upload; replacing it would require reimplementing
/// `cargo package` (thousands of lines covering manifest validation,
/// `.cargo_vcs_info.json`, exclude/include globs, tarball
/// determinism, license-file handling, etc.). Out of scope; not
/// every wheel needs reinventing.
fn run_cargo_publish(repo_root: &Path, krate: &str, dry_run: bool) -> Result<PublishOutcome> {
    let mut cmd = Command::new("cargo");
    cmd.arg("publish").arg("-p").arg(krate);
    if dry_run {
        cmd.arg("--dry-run");
    }
    cmd.current_dir(repo_root);
    let output = cmd.output().context("spawn cargo publish")?;
    if output.status.success() {
        return Ok(if dry_run {
            PublishOutcome::DryRun
        } else {
            PublishOutcome::Published
        });
    }
    // cargo's "already uploaded" diagnostic. Treat as success so a
    // re-run of a partially-shipped tree is idempotent.
    let stderr = String::from_utf8_lossy(&output.stderr);
    if stderr.contains("already uploaded")
        || stderr.contains("already exists on crates.io")
        || stderr.contains("crate version") && stderr.contains("is already uploaded")
    {
        eprintln!("    (skip: {krate} already on registry)");
        return Ok(PublishOutcome::AlreadyPublished);
    }
    // Re-emit cargo's stderr so the operator sees the real reason.
    eprintln!("{stderr}");
    Err(anyhow!(
        "cargo publish -p {krate} exited {}",
        output.status
    ))
}

/// Stage every worktree change (modifications, new files, deletions)
/// and create a commit on `HEAD`, returning the new commit SHA — or
/// `Ok(None)` when the working tree had nothing to commit (the common
/// case after a release where the README/CHANGELOG rendered
/// byte-identical to what's on disk).
///
/// 100% pure-Rust via `gix` — no `git` subprocess. The flow mirrors
/// `git add -A && git commit`:
///   1. enumerate `HEAD`→index and index→worktree status in one pass;
///   2. refuse to run if the index carries *staged* changes that differ
///      from `HEAD`. The release pipeline always regenerates docs on a
///      clean checkout (index == HEAD), so a dirty index signals an
///      unexpected state; committing anyway could silently diverge from
///      `git add -A` semantics, so we bail loudly instead;
///   3. apply every worktree delta onto a tree editor seeded from the
///      `HEAD` tree, write the tree, and commit it with `HEAD` as parent
///      (updates the `HEAD` ref + reflog);
///   4. refresh the on-disk index from the new tree so `git status`
///      reads clean afterwards.
///
/// Note: blobs are written from raw worktree bytes. nornir only commits
/// its own generated docs (UTF-8 markdown/TOML) in repos without
/// CRLF/clean `.gitattributes` filters, so no smudge/clean normalisation
/// is performed; repos relying on content filters are out of scope.
pub fn commit_release(repo_root: &Path, message: &str) -> Result<Option<String>> {
    use gix::bstr::BString;
    use gix::dir::entry::{Kind as DiskKind, Status as DiskStatus};
    use gix::status::index_worktree::Item as IwItem;
    use gix::status::plumbing::index_as_worktree::{Change, EntryStatus};

    let repo =
        gix::open(repo_root).with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo
        .head_commit()
        .context("resolve HEAD commit (unborn branch is unsupported here)")?;
    let head_id = head_commit.id;
    let head_tree_id = head_commit.tree().context("resolve HEAD tree")?.id;

    let iter = repo
        .status(gix::progress::Discard)
        .context("init status")?
        .head_tree(head_tree_id)
        .untracked_files(gix::status::UntrackedFiles::Files)
        .into_iter(Vec::<BString>::new())
        .context("start status walk")?;

    let mut editor = repo
        .edit_tree(head_tree_id)
        .context("seed tree editor from HEAD")?;
    let mut changed = false;

    for item in iter {
        match item.context("status item")? {
            // A tree→index difference means the index is dirty (staged
            // content differs from HEAD). Out of contract for releases.
            gix::status::Item::TreeIndex(change) => {
                return Err(anyhow!(
                    "refusing to commit: index has a staged change at `{}`; \
                     commit or reset it before releasing",
                    change.location()
                ));
            }
            gix::status::Item::IndexWorktree(iw) => match iw {
                IwItem::Modification {
                    rela_path, status, ..
                } => match status {
                    EntryStatus::Change(Change::Removed) => {
                        let rp: &gix::bstr::BStr = rela_path.as_ref();
                        editor
                            .remove(rp)
                            .with_context(|| format!("tree remove {rela_path}"))?;
                        changed = true;
                    }
                    EntryStatus::Conflict { .. } => {
                        return Err(anyhow!(
                            "refusing to commit: unresolved merge conflict at `{rela_path}`"
                        ));
                    }
                    // Modification / Type change / NeedsUpdate / IntentToAdd /
                    // SubmoduleModification: re-stage current worktree content.
                    _ => {
                        upsert_from_worktree(&repo, &mut editor, repo_root, rela_path.as_ref())?;
                        changed = true;
                    }
                },
                IwItem::DirectoryContents { entry, .. } => {
                    if matches!(entry.status, DiskStatus::Untracked)
                        && matches!(
                            entry.disk_kind,
                            Some(DiskKind::File) | Some(DiskKind::Symlink)
                        )
                    {
                        upsert_from_worktree(
                            &repo,
                            &mut editor,
                            repo_root,
                            entry.rela_path.as_ref(),
                        )?;
                        changed = true;
                    }
                }
                IwItem::Rewrite { .. } => {}
            },
        }
    }

    if !changed {
        return Ok(None);
    }
    let new_tree = editor.write().context("write release tree")?.detach();
    if new_tree == head_tree_id {
        return Ok(None);
    }
    let commit = repo
        .commit("HEAD", message, new_tree, Some(head_id))
        .context("create release commit")?;

    // Refresh the on-disk index from the new tree so the working tree
    // reads clean after the commit (as `git commit` would leave it).
    let mut index = repo
        .index_from_tree(&new_tree)
        .context("rebuild index from release tree")?;
    index
        .write(gix::index::write::Options::default())
        .context("persist refreshed index")?;

    Ok(Some(commit.to_string()))
}

/// Read `rela_path` from the working tree and upsert it into `editor`
/// with the correct blob kind (regular / executable / symlink).
fn upsert_from_worktree(
    repo: &gix::Repository,
    editor: &mut gix::object::tree::Editor<'_>,
    repo_root: &Path,
    rela_path: &gix::bstr::BStr,
) -> Result<()> {
    use gix::object::tree::EntryKind;

    let disk_path = repo_root.join(gix::path::from_bstr(rela_path).as_ref());
    let meta = std::fs::symlink_metadata(&disk_path)
        .with_context(|| format!("stat {}", disk_path.display()))?;

    let (bytes, kind): (Vec<u8>, EntryKind) = if meta.file_type().is_symlink() {
        let target = std::fs::read_link(&disk_path)
            .with_context(|| format!("readlink {}", disk_path.display()))?;
        let target = gix::path::into_bstr(target).into_owned();
        (target.into(), EntryKind::Link)
    } else {
        let bytes = std::fs::read(&disk_path)
            .with_context(|| format!("read {}", disk_path.display()))?;
        #[cfg(unix)]
        let kind = {
            use std::os::unix::fs::PermissionsExt;
            if meta.permissions().mode() & 0o111 != 0 {
                EntryKind::BlobExecutable
            } else {
                EntryKind::Blob
            }
        };
        #[cfg(not(unix))]
        let kind = EntryKind::Blob;
        (bytes, kind)
    };

    let blob = repo.write_blob(&bytes).context("write blob")?;
    editor
        .upsert(rela_path, kind, blob.detach())
        .with_context(|| format!("tree upsert {rela_path}"))?;
    Ok(())
}

/// Release commits are created locally; pushing is intentionally left to
/// the operator/CI. gix (0.84) implements fetch but not push/send-pack,
/// and nornir forbids both `git` subprocesses and C dependencies
/// (libgit2), so there is no in-process push path. This mirrors
/// [`tag()`], which likewise creates the ref locally and prints the push
/// command. Returns `false` to signal "not pushed" so callers don't
/// report a push that didn't happen.
pub fn push(repo_root: &Path, push_tags: bool) -> Result<bool> {
    let branch = crate::gitio::head_branch(repo_root).unwrap_or_else(|_| "HEAD".to_string());
    eprintln!("  ⏭  push skipped (pure-Rust build has no in-process push).");
    eprintln!("     run: git -C {} push origin {branch}", repo_root.display());
    if push_tags {
        eprintln!("     run: git -C {} push origin --tags", repo_root.display());
    }
    Ok(false)
}

/// Create an annotated `vX.Y.Z` tag pointing at HEAD using the
/// pure-Rust `gix` crate (no `git` shellout). Does *not* push — prints
/// the exact push command the operator/CI should run.
pub fn tag(repo_root: &Path, version: &str) -> Result<()> {
    let tag = format!("v{version}");
    let repo = gix::open(repo_root)
        .with_context(|| format!("gix::open {}", repo_root.display()))?;
    let head_commit = repo.head_commit().context("resolve HEAD commit")?;

    let message = format!("Release {tag}\n");
    let signature = repo
        .author()
        .ok_or_else(|| anyhow!("git author not configured (user.name / user.email)"))?
        .map_err(|e| anyhow!("read git author: {e}"))?;

    repo.tag(
        &tag,
        head_commit.id,
        gix::objs::Kind::Commit,
        Some(signature),
        &message,
        gix::refs::transaction::PreviousValue::MustNotExist,
    )
    .with_context(|| format!("create tag {tag}"))?;

    eprintln!(
        "created local tag {tag} at {}. Push with: git push origin {tag}",
        head_commit.id
    );
    Ok(())
}

// ─── Feature 3: crates.io index propagation poll ────────────────────

/// Block until `crates.io/api/v1/crates/<name>` advertises `version`
/// as `max_version`, or `timeout` elapses. Returns the wait time in
/// milliseconds (0 if the version was already there) so callers can
/// log it / persist it into `publish_attempts.wait_ms`.
///
/// Replaces "just `sleep 20` between phases" — crates.io's index
/// propagates in 5-90 s depending on load, so a fixed sleep is either
/// too long (slowing the release) or too short (next phase tries to
/// resolve a dep that's not yet visible and fails).
///
/// Uses `curl` rather than a Rust HTTP client to avoid pulling
/// reqwest/tokio into the library footprint just for one poll loop —
/// matches the existing crates.io-probe pattern in the publish flow.
/// crates.io returns HTTP 403 without a non-default User-Agent.
pub fn wait_for_index(
    crate_name: &str,
    version: &str,
    timeout: std::time::Duration,
) -> Result<u64> {
    use std::time::{Duration, Instant};
    let url = format!("https://crates.io/api/v1/crates/{crate_name}");
    let agent = "nornir-release-bot/0.1 (cargo-pipeline; +https://crates.io)";
    let started = Instant::now();
    loop {
        let out = Command::new("curl")
            .args(["-sSL", "-A", agent, "--max-time", "10", &url])
            .output()
            .context("spawn curl for crates.io poll")?;
        if out.status.success() {
            if let Ok(json) = serde_json::from_slice::<serde_json::Value>(&out.stdout) {
                let max = json.get("crate")
                    .and_then(|c| c.get("max_version"))
                    .and_then(|v| v.as_str());
                if max == Some(version) {
                    return Ok(started.elapsed().as_millis() as u64);
                }
            }
        }
        if started.elapsed() >= timeout {
            return Err(anyhow!(
                "timed out waiting for crates.io index: {crate_name}@{version} after {:?}",
                timeout
            ));
        }
        std::thread::sleep(Duration::from_secs(3));
    }
}

// ─── Feature 5: .crate tarball stats ────────────────────────────────

/// Stats from a `cargo package --list` run. Used by the bloat gate
/// and persisted into the `crate_tarball_stats` iceberg table.
#[derive(Debug, Clone)]
pub struct TarballStats {
    pub crate_name: String,
    pub version: String,
    pub tarball_bytes: u64,
    pub file_count: usize,
    pub largest_file: Option<String>,
    pub largest_file_bytes: Option<u64>,
}

/// Run `cargo package -p <krate> --list --allow-dirty` and inspect
/// the resulting `.crate` tarball in `target/package/`. Does *not*
/// upload anything. Reports file count + size of largest entry so the
/// caller can warn on bloat (typical libraries are <100 kB; >5 MB is
/// almost always test fixtures or planet-scale data leaking in via a
/// missing `exclude = [...]`).
pub fn tarball_stats(repo_root: &Path, krate: &str) -> Result<TarballStats> {
    use std::process::Command;
    let pkg = Command::new("cargo")
        .args(["package", "-p", krate, "--allow-dirty", "--no-verify"])
        .current_dir(repo_root)
        .output()
        .context("spawn cargo package")?;
    if !pkg.status.success() {
        return Err(anyhow!(
            "cargo package -p {krate} exited {}: {}",
            pkg.status,
            String::from_utf8_lossy(&pkg.stderr)
        ));
    }
    let target = repo_root.join("target").join("package");
    // Find the .crate file we just produced; pick the newest matching.
    let mut newest: Option<(std::path::PathBuf, std::time::SystemTime)> = None;
    if let Ok(rd) = std::fs::read_dir(&target) {
        for e in rd.flatten() {
            let p = e.path();
            if p.extension().and_then(|s| s.to_str()) != Some("crate") { continue }
            let fname = p.file_name().and_then(|s| s.to_str()).unwrap_or("");
            if !fname.starts_with(&format!("{krate}-")) { continue }
            let mtime = e.metadata().and_then(|m| m.modified()).unwrap_or(std::time::UNIX_EPOCH);
            if newest.as_ref().map(|(_, t)| mtime > *t).unwrap_or(true) {
                newest = Some((p, mtime));
            }
        }
    }
    let (path, _) = newest.ok_or_else(|| anyhow!("no .crate tarball found in {}", target.display()))?;
    let tarball_bytes = std::fs::metadata(&path)?.len();
    let fname = path.file_name().and_then(|s| s.to_str()).unwrap_or("");
    let version = fname
        .strip_prefix(&format!("{krate}-"))
        .and_then(|s| s.strip_suffix(".crate"))
        .unwrap_or("?")
        .to_string();
    // Walk the tarball to count files + find the largest.
    use flate2::read::GzDecoder;
    use tar::Archive;
    let f = std::fs::File::open(&path)?;
    let dec = GzDecoder::new(f);
    let mut ar = Archive::new(dec);
    let mut file_count = 0usize;
    let mut largest: Option<(String, u64)> = None;
    for entry in ar.entries()? {
        let entry = entry?;
        let size = entry.header().size().unwrap_or(0);
        let p = entry.path()?.into_owned();
        file_count += 1;
        if largest.as_ref().map(|(_, s)| size > *s).unwrap_or(true) {
            largest = Some((p.display().to_string(), size));
        }
    }
    Ok(TarballStats {
        crate_name: krate.to_string(),
        version,
        tarball_bytes,
        file_count,
        largest_file: largest.as_ref().map(|(n, _)| n.clone()),
        largest_file_bytes: largest.map(|(_, s)| s),
    })
}

/// Default bloat threshold: 5 MB. Tweakable per-call.
pub const DEFAULT_TARBALL_BYTES_THRESHOLD: u64 = 5 * 1024 * 1024;

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

    fn m(n: &str, deps: &[&str]) -> (String, Vec<String>) {
        (n.to_string(), deps.iter().map(|s| s.to_string()).collect())
    }

    #[test]
    fn topo_phases_orders_deps_before_dependents_deterministically() {
        // testmatrix ← robotui ← nornir ; airgap ← nornir ; testmatrix ← nornir.
        let members = vec![
            m("nornir", &["nornir-testmatrix", "nornir-robotui", "nornir-airgap"]),
            m("nornir-robotui", &["nornir-testmatrix"]),
            m("nornir-testmatrix", &[]),
            m("nornir-airgap", &[]),
        ];
        let phases = topo_phases(members);
        let flat: Vec<&str> = phases.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c).unwrap();
        // deps strictly before dependents
        assert!(pos("nornir-testmatrix") < pos("nornir-robotui"));
        assert!(pos("nornir-robotui") < pos("nornir"));
        assert!(pos("nornir-airgap") < pos("nornir"));
        assert!(pos("nornir-testmatrix") < pos("nornir"));
        // deterministic: leaves (testmatrix, airgap) share phase 0, alphabetical
        assert_eq!(phases[0], vec!["nornir-airgap", "nornir-testmatrix"]);
        // nornir is last (its own phase)
        assert_eq!(phases.last().unwrap(), &vec!["nornir".to_string()]);
    }

    #[test]
    fn topo_phases_cycle_drops_nothing() {
        let members = vec![m("a", &["b"]), m("b", &["a"])]; // pathological cycle
        let phases = topo_phases(members);
        let flat: std::collections::BTreeSet<&str> =
            phases.iter().flatten().map(String::as_str).collect();
        assert!(flat.contains("a") && flat.contains("b"), "cycle members not dropped");
    }

    #[test]
    fn derive_publish_order_from_this_real_workspace() {
        // Deterministic derivation against nornir's OWN workspace: testmatrix must
        // come before nornir (nornir depends on it), and nornir is the last crate.
        let order = derive_publish_order(std::path::Path::new(".")).expect("derive");
        let flat: Vec<&str> = order.iter().flatten().map(String::as_str).collect();
        let pos = |c: &str| flat.iter().position(|x| *x == c);
        assert!(flat.contains(&"nornir"), "nornir is publishable: {flat:?}");
        assert!(flat.contains(&"nornir-testmatrix"), "testmatrix is publishable: {flat:?}");
        assert!(
            pos("nornir-testmatrix") < pos("nornir"),
            "testmatrix publishes before nornir: {flat:?}"
        );
    }
}