holger-agent-lib 0.6.9

//! Agent transfer engine: the concrete Source→Target moves behind the CLI, one
//! step above the `ConnectorTrait` seam. Pull side downloads a connector's assets
//! and lands them as a `.znippy` archive (with ecosystem `pkg_type` tags so the
//! compressor runs the right plugin and the archive is natively browsable), a plain
//! `.tar.zst`, or a loose directory. Push side reverses it: a directory, `.znippy`,
//! or `.tar.zst` becomes a stream of `connector.upload_asset` calls.
//!
//! `pkg_type` is derived, never literal — [`pkg_type_from_format`] delegates to
//! [`traits::ArtifactFormat`] (znippy's own registry), and [`pkg_type_for_file`] is a
//! deliberately CURATED extension table that tags only unambiguous packaging (`.crate`,
//! `.jar`, `.whl`, …) and leaves ambiguous ones (`.tar.gz` sdist vs npm tgz) untyped for
//! an explicit `--format` override; a drift-guard test pins it as a subset of znippy.
//!
//! Uploads fan out through a bounded in-flight pool (`znippy_zoomies::gatling`,
//! `UPLOAD_CONCURRENCY`): a per-asset upload failure is counted, not fatal, while a
//! source-read/decode error short-circuits the whole push. The one gotcha: archive
//! members are UNTRUSTED airgap input — `read_capped`/`MAX_TAR_ENTRY_BYTES` bound
//! per-entry expansion (decompression bombs) and `is_safe_member_path`/`safe_join`
//! reject absolute or `..` paths, since a connector only trims a leading `/` and a
//! stray `..` could otherwise write outside the repo (or drop an RCE file on the host).

use anyhow::{Result, Context};
use std::path::PathBuf;
use std::sync::Arc;
use traits::{ConnectorTrait, RemoteAsset};
use znippy_compress::stream_packer::{ArchiveEntry, StreamCompressor};

/// Per-entry expansion cap when reading an untrusted `.tar.zst` (M7). Bounds a
/// single decompressed tar member so a crafted archive cannot OOM the host.
const MAX_TAR_ENTRY_BYTES: u64 = 8 * 1024 * 1024 * 1024;

/// Concurrency bound for the SAVE stage: how many uploads to a network sink may
/// be in flight at once. The push paths are latency-bound (one HTTP round-trip
/// per asset), so overlapping ~8 PUTs hides per-request latency while keeping
/// peak memory (≈ cap buffered payloads) and sink load bounded. Drives the
/// no-barrier in-flight pool [`znippy_zoomies::gatling::io::run`].
const UPLOAD_CONCURRENCY: usize = 8;

/// Outcome of one upload job in a parallel push pool. An upload *failure* is data
/// (counted, not fatal) so one bad asset does not abort the whole push — matching
/// the old sequential loops; only a source-read/decode error short-circuits.
enum UploadOutcome {
    Ok(String),
    Failed(String, String),
}

/// One unit pulled off the sequential tar/zstd decoder, ready to feed the upload
/// pool. The decoder is single-threaded, so members are read in order at job
/// admission; `Fatal` carries a decode/read error that must short-circuit the
/// whole push (matching the old `?`), while `Upload` carries a ready payload.
enum TarJob {
    Upload { path: String, data: Vec<u8> },
    Fatal(anyhow::Error),
}

/// Tally one finished upload, mirroring the per-item `✓ / ✗` line the sequential
/// loops printed (order is completion order now that uploads overlap).
fn tally_upload(outcome: UploadOutcome, pushed: &mut usize, failed: &mut usize) {
    match outcome {
        UploadOutcome::Ok(p) => {
            *pushed += 1;
            println!("   ✓ {p}");
        }
        UploadOutcome::Failed(p, e) => {
            *failed += 1;
            println!("   ✗ {p}: {e}");
        }
    }
}

/// Read at most `max` bytes from `reader`, erroring if the source holds more.
/// Used to bound per-entry decompression of an untrusted archive (M7).
fn read_capped<R: std::io::Read>(reader: &mut R, max: u64, what: &str) -> Result<Vec<u8>> {
    let mut data = Vec::new();
    let mut limited = std::io::Read::take(reader, max.saturating_add(1));
    std::io::Read::read_to_end(&mut limited, &mut data)
        .with_context(|| format!("Failed to read entry: {what}"))?;
    if data.len() as u64 > max {
        anyhow::bail!(
            "tar entry {what} exceeds per-entry cap of {max} bytes (decompression bomb?)"
        );
    }
    Ok(data)
}

/// Map a Nexus/Artifactory/CLI format string → znippy `pkg_type` discriminant.
///
/// znippy is the single source of truth for these numbers: this delegates to
/// [`ArtifactFormat`], which derives them from znippy's own handler registry.
/// Unknown formats map to `0` (raw / unknown).
pub fn pkg_type_from_format(format: &str) -> i8 {
    use traits::ArtifactFormat;
    ArtifactFormat::from_format_str(format)
        .map(|f| f.znippy_type_id())
        .unwrap_or(0)
}

/// Join an upstream-supplied relative `asset.path` onto `base`, guaranteeing the
/// result stays inside `base`.
///
/// `asset.path` comes verbatim from a remote registry listing (Nexus/Artifactory/
/// PyPI), so a malicious or compromised upstream can return an absolute path or
/// one containing `..` to write outside the download directory (drop a cron/shell
/// file → RCE on the internet-side agent host). Unlike a single artifact-id field,
/// a legitimate asset path *may* contain `/` separators, so this validates each
/// component instead of rejecting separators: reject absolute/root/prefix paths
/// and any `..` (`ParentDir`) component; `.`/`CurDir` is harmless and skipped.
fn safe_join(base: &std::path::Path, rel: &str) -> Result<PathBuf> {
    use std::path::Component;
    let rel_path = std::path::Path::new(rel);
    let mut out = base.to_path_buf();
    for comp in rel_path.components() {
        match comp {
            Component::Normal(c) => out.push(c),
            Component::CurDir => {}
            Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
                anyhow::bail!("refusing unsafe asset path {rel:?} (escapes {})", base.display());
            }
        }
    }
    Ok(out)
}

/// `true` iff `rel` is a safe relative member path to PUSH to a connector: no
/// absolute/root/prefix and no `..` component. Archive member paths come from an
/// untrusted airgap-carried .znippy/.tar.zst, and `connector.upload_asset` only
/// trims a leading `/`, so a `..` member could place an artifact outside the
/// intended repo path on the target. Mirrors the `safe_join` component policy;
/// shared by both push paths so an offending member is skipped+counted, never
/// uploaded.
fn is_safe_member_path(rel: &str) -> bool {
    use std::path::Component;
    std::path::Path::new(rel).components().all(|c| {
        matches!(c, Component::Normal(_) | Component::CurDir)
    })
}

/// Feed assets from one repo into an already-open compressor sender.
/// Returns (downloaded, failed) counts.
pub async fn feed_repo_to_sender(
    connector: Arc<dyn ConnectorTrait>,
    assets: Vec<RemoteAsset>,
    sender: crossbeam_channel::Sender<ArchiveEntry>,
    pkg_type: i8,
    repo_name: String,
) -> Result<(usize, usize)> {
    let total = assets.len();
    let mut downloaded = 0usize;
    let mut failed = 0usize;

    for (i, asset) in assets.iter().enumerate() {
        print!("   [{}/{}] {} ... ", i + 1, total, asset.path);
        match connector.download_asset(asset).await {
            Ok(bytes) => {
                let n = bytes.len();
                sender.send(ArchiveEntry {
                    relative_path: asset.path.clone(),
                    data: bytes,
                    pkg_type: Some(pkg_type),
                    repo: Some(repo_name.clone()),
                }).context("compressor channel closed")?;
                downloaded += 1;
                println!("✓ ({} bytes)", n);
            }
            Err(e) => {
                failed += 1;
                println!("✗ {}", e);
            }
        }
    }
    Ok((downloaded, failed))
}

/// Dump all repos from a connector into one znippy archive.
///
/// Parallel mode (default): all repos downloaded concurrently — tokio tasks share cloned
/// senders. The compressor's bounded channel provides backpressure so memory stays bounded.
/// Sequential mode: one repo at a time — lower peak memory and network usage.
pub async fn dump_all_repos_to_znippy(
    connector: Arc<dyn ConnectorTrait>,
    repos: Vec<traits::RemoteRepository>,
    output: PathBuf,
    sequential: bool,
) -> Result<()> {
    use znippy_compress::stream_packer::compress_stream;

    println!("📦 Creating znippy archive: {}", output.display());
    let compressor: StreamCompressor = compress_stream(&output, false)?;

    let total_downloaded;
    let total_failed;

    if sequential {
        let sender = compressor.sender().clone();
        let mut dl = 0usize;
        let mut fail = 0usize;
        for repo in &repos {
            let pkg_type = pkg_type_from_format(&repo.format);
            println!("\n→ {} (format: {}, pkg_type: {})", repo.name, repo.format, pkg_type);
            let assets = connector.list_assets(&repo.name).await?;
            println!("  {} assets", assets.len());
            let (d, f) = feed_repo_to_sender(
                connector.clone(), assets, sender.clone(), pkg_type, repo.name.clone(),
            ).await?;
            dl += d;
            fail += f;
        }
        drop(sender);
        total_downloaded = dl;
        total_failed = fail;
    } else {
        // Parallel: spawn one task per repo, all share cloned senders.
        let mut tasks = Vec::new();
        for repo in &repos {
            let pkg_type = pkg_type_from_format(&repo.format);
            println!("\n→ {} (format: {}, pkg_type: {})", repo.name, repo.format, pkg_type);
            let assets = connector.list_assets(&repo.name).await?;
            println!("  {} assets queued", assets.len());

            let conn = connector.clone();
            let sender = compressor.sender().clone();
            let repo_name = repo.name.clone();
            tasks.push(tokio::spawn(async move {
                feed_repo_to_sender(conn, assets, sender, pkg_type, repo_name).await
            }));
        }

        let mut dl = 0usize;
        let mut fail = 0usize;
        for task in tasks {
            let (d, f) = task.await
                .context("repo task panicked")?
                .context("repo download failed")?;
            dl += d;
            fail += f;
        }
        total_downloaded = dl;
        total_failed = fail;
    }

    let report = compressor.finish()?;
    println!("\n✅ Archive complete: {} downloaded, {} failed ({} bytes compressed)",
             total_downloaded, total_failed, report.compressed_bytes);
    Ok(())
}

/// Transfer assets from a connector to a znippy archive (single repo, no pkg_type tagging).
pub async fn transfer_to_znippy(
    connector: &dyn ConnectorTrait,
    assets: Vec<RemoteAsset>,
    output: PathBuf,
) -> Result<()> {
    use znippy_compress::stream_packer::compress_stream;

    println!("📦 Creating znippy archive: {}", output.display());
    let compressor = compress_stream(&output, false)?;
    let sender = compressor.sender().clone();

    let total = assets.len();
    let mut downloaded = 0usize;

    for (i, asset) in assets.iter().enumerate() {
        print!("   [{}/{}] {} ... ", i + 1, total, asset.path);
        match connector.download_asset(asset).await {
            Ok(bytes) => {
                let n = bytes.len();
                sender.send(ArchiveEntry {
                    relative_path: asset.path.clone(),
                    data: bytes,
                    pkg_type: None,
                    repo: None,
                }).context("Failed to send to compressor")?;
                downloaded += 1;
                println!("✓ ({} bytes)", n);
            }
            Err(e) => {
                println!("✗ {}", e);
            }
        }
    }

    drop(sender);
    let report = compressor.finish()?;
    println!("\n✅ Znippy archive complete: {} assets packed ({} bytes compressed)",
             downloaded, report.compressed_bytes);
    Ok(())
}

/// Map an artifact filename → znippy `pkg_type`, so the matching ecosystem plugin
/// runs at compress time and writes its index columns (maven GAV, python name/ver,
/// rust crate coords). Returns `None` for files with no recognized packaging
/// extension — they are still packed, just untyped (served by raw path). The
/// `.tar.gz`/`.tgz` cases are ambiguous (python sdist vs npm tarball); they are left
/// untyped here and resolved by an explicit `format` override when one is supplied.
///
/// znippy is the source of truth for the *numbers*: the resolved discriminant comes
/// from [`pkg_type_from_format`] → [`traits::ArtifactFormat::znippy_type_id`] →
/// znippy's handler register, never a literal here. The extension → ecosystem
/// **table** below is a deliberately CURATED adapter (not a copy of znippy's plugin
/// `meta().extensions`): the agent intentionally tags only the unambiguous packaging
/// extensions and leaves znippy's ambiguous/secondary claims (`.tar.gz`, `.zip`,
/// `.tgz`, `.snupkg`, …) untyped for an explicit `--format` override. The
/// `pkg_type_for_file_agrees_with_znippy_plugin_extensions` test pins the curated
/// rows that znippy-common can confirm, so the two cannot silently drift.
fn pkg_type_for_file(path: &std::path::Path) -> Option<i8> {
    let name = path.file_name()?.to_str()?;
    let fmt = match name.rsplit_once('.').map(|(_, e)| e)? {
        "crate" => "rust",
        "jar" | "pom" | "war" | "ear" => "maven",
        "whl" => "python",
        "nupkg" => "nuget",
        "gem" => "gem",
        _ => return None,
    };
    match pkg_type_from_format(fmt) {
        0 => None,
        t => Some(t),
    }
}

/// Build a `.znippy` archive from every file under `directory` (recursively),
/// preserving each file's path relative to `directory` as its archive member path.
///
/// This is the offline inverse of [`push_znippy_to_connector`]: stage a tree of
/// real artifacts (`.crate`, `.jar`+`.pom`, `.whl`, …) and pack it into one
/// immutable, content-addressed archive carriable across an airgap. Each member is
/// tagged with its ecosystem `pkg_type` (by extension, or forced by `format` when
/// given) so the znippy compressor runs that ecosystem's plugin and writes the
/// typed index columns — the archive is then natively *browsable* by holger
/// (`as_maven()`/`as_python()`/…), not just a raw file bag. A `format` override
/// (e.g. `"python"`) tags every file with that ecosystem, which is how ambiguous
/// extensions (`.tar.gz` sdists) get indexed.
pub async fn pack_directory_to_znippy(
    directory: PathBuf,
    output: PathBuf,
    format: Option<&str>,
) -> Result<()> {
    use znippy_compress::stream_packer::compress_stream;

    if !directory.is_dir() {
        anyhow::bail!("not a directory: {}", directory.display());
    }
    let forced = format.map(pkg_type_from_format).filter(|t| *t != 0);

    let mut files = Vec::new();
    walk_all_files(&directory, &mut files)?;
    files.sort();
    if files.is_empty() {
        anyhow::bail!("no files found under {}", directory.display());
    }

    println!("📦 Packing {} files from {} → {}", files.len(), directory.display(), output.display());
    let compressor = compress_stream(&output, false)?;
    let sender = compressor.sender().clone();

    let total = files.len();
    let mut packed = 0usize;
    for (i, path) in files.iter().enumerate() {
        let rel = path
            .strip_prefix(&directory)
            .unwrap_or(path)
            .to_string_lossy()
            .replace('\\', "/");
        let pkg_type = forced.or_else(|| pkg_type_for_file(path));
        let data = std::fs::read(path)
            .with_context(|| format!("Failed to read {}", path.display()))?;
        let n = data.len();
        sender
            .send(ArchiveEntry { relative_path: rel.clone(), data, pkg_type, repo: None })
            .context("Failed to send to compressor")?;
        packed += 1;
        println!("   [{}/{}] {} ({} bytes, pkg_type={:?})", i + 1, total, rel, n, pkg_type);
    }

    drop(sender);
    let report = compressor.finish()?;
    println!(
        "\n✅ Znippy archive complete: {} files packed ({} bytes compressed)",
        packed, report.compressed_bytes
    );
    Ok(())
}

/// Recursively collect every regular file under `dir` (any extension).
fn walk_all_files(dir: &std::path::Path, files: &mut Vec<PathBuf>) -> Result<()> {
    if !dir.is_dir() {
        return Ok(());
    }
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let path = entry.path();
        if path.is_dir() {
            walk_all_files(&path, files)?;
        } else if path.is_file() {
            files.push(path);
        }
    }
    Ok(())
}

/// Transfer assets from a connector to a zstd-compressed tar archive (.tar.zst)
pub async fn transfer_to_tar_zstd(
    connector: &dyn ConnectorTrait,
    assets: Vec<RemoteAsset>,
    output: PathBuf,
) -> Result<()> {
    use std::io::BufWriter;

    println!("📦 Creating tar.zst archive: {}", output.display());
    let file = std::fs::File::create(&output)
        .with_context(|| format!("Failed to create output file: {}", output.display()))?;
    let zstd_encoder = zstd::Encoder::new(BufWriter::new(file), 3)
        .context("Failed to create zstd encoder")?;
    let mut tar_builder = tar::Builder::new(zstd_encoder);

    let total = assets.len();
    let mut downloaded = 0usize;
    let mut failed = 0usize;

    for (i, asset) in assets.iter().enumerate() {
        print!("   [{}/{}] {} ... ", i + 1, total, asset.path);
        match connector.download_asset(asset).await {
            Ok(bytes) => {
                let mut header = tar::Header::new_gnu();
                header.set_size(bytes.len() as u64);
                header.set_mode(0o644);
                header.set_cksum();
                tar_builder.append_data(&mut header, &asset.path, bytes.as_slice())
                    .with_context(|| format!("Failed to append {} to tar", asset.path))?;
                downloaded += 1;
                println!("✓ ({} bytes)", bytes.len());
            }
            Err(e) => {
                failed += 1;
                println!("✗ {}", e);
            }
        }
    }

    let zstd_encoder = tar_builder.into_inner()
        .context("Failed to finish tar archive")?;
    zstd_encoder.finish()
        .context("Failed to finish zstd compression")?;

    println!("\n✅ tar.zst archive complete: {} assets packed, {} failed", downloaded, failed);
    Ok(())
}

/// Transfer assets from a connector to a directory
pub async fn transfer_to_directory(
    connector: &dyn ConnectorTrait,
    assets: Vec<RemoteAsset>,
    output: PathBuf,
) -> Result<()> {
    println!("📁 Downloading to directory: {}", output.display());
    std::fs::create_dir_all(&output)
        .with_context(|| format!("Failed to create output directory: {}", output.display()))?;

    let total = assets.len();
    let mut downloaded = 0usize;
    let mut failed = 0usize;

    for (i, asset) in assets.iter().enumerate() {
        let dest = match safe_join(&output, &asset.path) {
            Ok(d) => d,
            Err(e) => {
                failed += 1;
                println!("   [{}/{}] {} ... ✗ {}", i + 1, total, asset.path, e);
                continue;
            }
        };
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)?;
        }

        print!("   [{}/{}] {} ... ", i + 1, total, asset.path);

        match connector.download_asset(asset).await {
            Ok(bytes) => {
                std::fs::write(&dest, &bytes)
                    .with_context(|| format!("Failed to write {}", dest.display()))?;
                downloaded += 1;
                println!("✓ ({} bytes)", bytes.len());
            }
            Err(e) => {
                failed += 1;
                println!("✗ {}", e);
            }
        }
    }

    println!("\n✅ Done! {} downloaded, {} failed", downloaded, failed);
    Ok(())
}

/// Push assets from a directory to a connector
pub async fn push_directory_to_connector(
    connector: &dyn ConnectorTrait,
    directory: PathBuf,
    repository: &str,
) -> Result<()> {
    println!("📦 Pushing from directory: {}", directory.display());

    let crate_files = find_crate_files(&directory)?;
    println!("   Found {} .crate files to push", crate_files.len());

    if crate_files.is_empty() {
        println!("   Nothing to push.");
        return Ok(());
    }

    let mut pushed = 0usize;
    let mut failed = 0usize;

    // SAVE stage → network sink: a bounded parallel pool of in-flight uploads
    // (≤ UPLOAD_CONCURRENCY) instead of one PUT at a time. The engine pulls jobs
    // lazily, so at most ~cap files are read into memory at once; a source-read
    // error short-circuits (Err), an upload error is counted (Ok(Failed)).
    let jobs = crate_files.iter().map(move |crate_path| async move {
        let upload_path = extract_crate_relative_path(crate_path)
            .to_string_lossy()
            .into_owned();
        let data = std::fs::read(crate_path)
            .with_context(|| format!("Failed to read {}", crate_path.display()))?;
        anyhow::Ok(match connector.upload_asset(repository, &upload_path, &data).await {
            Ok(()) => UploadOutcome::Ok(upload_path),
            Err(e) => UploadOutcome::Failed(upload_path, e.to_string()),
        })
    });

    znippy_zoomies::gatling::io::run(jobs, UPLOAD_CONCURRENCY, |outcome| {
        tally_upload(outcome, &mut pushed, &mut failed);
        Ok(())
    })
    .await?;

    println!("\n✅ Push complete: {} succeeded, {} failed", pushed, failed);
    Ok(())
}

/// Push assets from a znippy archive to a connector
pub async fn push_znippy_to_connector(
    connector: &dyn ConnectorTrait,
    archive: PathBuf,
    repository: &str,
) -> Result<()> {
    use znippy_common::{ZnippyArchive, ZnippyReader};

    println!("📦 Reading znippy archive: {}", archive.display());
    let znippy = ZnippyArchive::open(&archive)?;
    let files = znippy.list_files()?;
    println!("   Found {} files in archive", files.len());

    let mut pushed = 0usize;
    let mut failed = 0usize;

    // SAVE stage → network sink: extract each member (sync, at admission) and fan
    // the uploads out as a bounded parallel pool, ≤ UPLOAD_CONCURRENCY PUTs in
    // flight. Extract/upload failures are counted, never fatal — same policy as
    // the old sequential loop.
    let znippy = &znippy;
    let jobs = files.iter().map(move |path| async move {
        if !is_safe_member_path(path) {
            return anyhow::Ok(UploadOutcome::Failed(
                path.clone(),
                "unsafe member path (absolute or '..') — skipped".to_string(),
            ));
        }
        anyhow::Ok(match znippy.extract_file(path) {
            Ok(data) => match connector.upload_asset(repository, path, &data).await {
                Ok(()) => UploadOutcome::Ok(path.clone()),
                Err(e) => UploadOutcome::Failed(path.clone(), e.to_string()),
            },
            Err(e) => UploadOutcome::Failed(path.clone(), format!("extract: {e}")),
        })
    });

    znippy_zoomies::gatling::io::run(jobs, UPLOAD_CONCURRENCY, |outcome| {
        tally_upload(outcome, &mut pushed, &mut failed);
        Ok(())
    })
    .await?;

    println!("\n✅ Push complete: {} succeeded, {} failed", pushed, failed);
    Ok(())
}

/// Push assets from a tar.zst archive to a connector
pub async fn push_tar_zstd_to_connector(
    connector: &dyn ConnectorTrait,
    archive: PathBuf,
    repository: &str,
) -> Result<()> {
    use std::io::BufReader;

    println!("📦 Reading tar.zst archive: {}", archive.display());
    let file = std::fs::File::open(&archive)
        .with_context(|| format!("Failed to open archive: {}", archive.display()))?;
    let zstd_decoder = zstd::Decoder::new(BufReader::new(file))
        .context("Failed to create zstd decoder")?;
    let mut tar_archive = tar::Archive::new(zstd_decoder);

    let mut entries = tar_archive.entries().context("Failed to read tar entries")?;
    let mut pushed = 0usize;
    let mut failed = 0usize;

    // The tar/zstd stream is a single sequential decoder, so members are read in
    // order at admission (M7 per-entry cap still applies). The SAVE stage — the
    // uploads — then runs as a bounded parallel pool: ≤ UPLOAD_CONCURRENCY PUTs
    // in flight, so member k+1 is read and dispatched while k's PUT is on the
    // wire. A decode/read error short-circuits the pool (TarJob::Fatal → Err).
    let jobs = std::iter::from_fn(move || loop {
        let mut entry = match entries.next()? {
            Ok(e) => e,
            Err(e) => {
                return Some(TarJob::Fatal(
                    anyhow::Error::new(e).context("Failed to read tar entry"),
                ));
            }
        };
        if entry.header().entry_type() != tar::EntryType::Regular {
            continue;
        }
        let path = match entry.path() {
            Ok(p) => p.to_string_lossy().into_owned(),
            Err(e) => {
                return Some(TarJob::Fatal(
                    anyhow::Error::new(e).context("Failed to read entry path"),
                ));
            }
        };
        // M7: bound per-entry expansion. A zstd-compressed tar carried across
        // an airgap is untrusted; an unbounded `read_to_end` lets one entry
        // expand without limit (decompression bomb).
        return Some(match read_capped(&mut entry, MAX_TAR_ENTRY_BYTES, &path) {
            Ok(data) => TarJob::Upload { path, data },
            Err(e) => TarJob::Fatal(e),
        });
    })
    .map(move |job| async move {
        match job {
            TarJob::Fatal(e) => Err(e),
            TarJob::Upload { path, data } => {
                if !is_safe_member_path(&path) {
                    return anyhow::Ok(UploadOutcome::Failed(
                        path,
                        "unsafe member path (absolute or '..') — skipped".to_string(),
                    ));
                }
                anyhow::Ok(
                    match connector.upload_asset(repository, &path, &data).await {
                        Ok(()) => UploadOutcome::Ok(path),
                        Err(e) => UploadOutcome::Failed(path, e.to_string()),
                    },
                )
            }
        }
    });

    znippy_zoomies::gatling::io::run(jobs, UPLOAD_CONCURRENCY, |outcome| {
        tally_upload(outcome, &mut pushed, &mut failed);
        Ok(())
    })
    .await?;

    println!("\n✅ Push complete: {} succeeded, {} failed", pushed, failed);
    Ok(())
}

/// Unpack a znippy archive to a directory — the inverse of
/// [`pack_directory_to_znippy`]. Each member is extracted and written under
/// `output` at its (path-safe) relative path. Members with hostile paths
/// (absolute or containing `..`) are skipped, never written outside `output`.
pub async fn unpack_znippy_to_directory(archive: PathBuf, output: PathBuf) -> Result<()> {
    use znippy_common::{ZnippyArchive, ZnippyReader};

    println!("📦 Reading znippy archive: {}", archive.display());
    let znippy = ZnippyArchive::open(&archive)?;
    let files = znippy.list_files()?;
    println!("   Found {} files in archive → {}", files.len(), output.display());
    std::fs::create_dir_all(&output)
        .with_context(|| format!("Failed to create output directory: {}", output.display()))?;

    let total = files.len();
    let mut written = 0usize;
    let mut failed = 0usize;
    for (i, path) in files.iter().enumerate() {
        let dest = match safe_join(&output, path) {
            Ok(d) => d,
            Err(e) => {
                failed += 1;
                println!("   [{}/{}] {} ... ✗ {}", i + 1, total, path, e);
                continue;
            }
        };
        match znippy.extract_file(path) {
            Ok(data) => {
                if let Some(parent) = dest.parent() {
                    std::fs::create_dir_all(parent)?;
                }
                std::fs::write(&dest, &data)
                    .with_context(|| format!("Failed to write {}", dest.display()))?;
                written += 1;
                println!("   [{}/{}] {} ✓ ({} bytes)", i + 1, total, path, data.len());
            }
            Err(e) => {
                failed += 1;
                println!("   [{}/{}] {} ... ✗ extract: {}", i + 1, total, path, e);
            }
        }
    }
    println!("\n✅ Unpacked znippy: {} files written, {} failed", written, failed);
    if failed > 0 {
        anyhow::bail!("{failed} member(s) failed to extract from {}", archive.display());
    }
    Ok(())
}

/// Unpack a `.tar.zst` archive to a directory. Each regular entry is extracted
/// (bounded by [`MAX_TAR_ENTRY_BYTES`] — the stream is untrusted airgap input) and
/// written under `output` at its (path-safe) relative path.
pub async fn unpack_tar_zstd_to_directory(archive: PathBuf, output: PathBuf) -> Result<()> {
    use std::io::BufReader;

    println!("📦 Reading tar.zst archive: {} → {}", archive.display(), output.display());
    std::fs::create_dir_all(&output)
        .with_context(|| format!("Failed to create output directory: {}", output.display()))?;
    let file = std::fs::File::open(&archive)
        .with_context(|| format!("Failed to open archive: {}", archive.display()))?;
    let zstd_decoder = zstd::Decoder::new(BufReader::new(file))
        .context("Failed to create zstd decoder")?;
    let mut tar_archive = tar::Archive::new(zstd_decoder);

    let mut written = 0usize;
    let mut failed = 0usize;
    for entry in tar_archive.entries().context("Failed to read tar entries")? {
        let mut entry = entry.context("Failed to read tar entry")?;
        if entry.header().entry_type() != tar::EntryType::Regular {
            continue;
        }
        let path = entry
            .path()
            .context("Failed to read entry path")?
            .to_string_lossy()
            .into_owned();
        let dest = match safe_join(&output, &path) {
            Ok(d) => d,
            Err(e) => {
                failed += 1;
                println!("   {} ... ✗ {}", path, e);
                continue;
            }
        };
        let data = read_capped(&mut entry, MAX_TAR_ENTRY_BYTES, &path)?;
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)?;
        }
        std::fs::write(&dest, &data)
            .with_context(|| format!("Failed to write {}", dest.display()))?;
        written += 1;
        println!("   {} ✓ ({} bytes)", path, data.len());
    }
    println!("\n✅ Unpacked tar.zst: {} files written, {} skipped", written, failed);
    Ok(())
}

/// Pack a directory into a `.tar.zst` archive — the inverse of
/// [`unpack_tar_zstd_to_directory`], and the on-disk twin of
/// [`transfer_to_tar_zstd`] (which packs from a connector). Every regular file
/// under `directory` is appended at its path relative to `directory`.
pub async fn pack_directory_to_tar_zstd(directory: PathBuf, output: PathBuf) -> Result<()> {
    use std::io::BufWriter;

    if !directory.is_dir() {
        anyhow::bail!("not a directory: {}", directory.display());
    }
    let mut files = Vec::new();
    walk_all_files(&directory, &mut files)?;
    files.sort();
    if files.is_empty() {
        anyhow::bail!("no files found under {}", directory.display());
    }

    println!("📦 Packing {} files from {} → {}", files.len(), directory.display(), output.display());
    let file = std::fs::File::create(&output)
        .with_context(|| format!("Failed to create output file: {}", output.display()))?;
    let zstd_encoder = zstd::Encoder::new(BufWriter::new(file), 3)
        .context("Failed to create zstd encoder")?;
    let mut tar_builder = tar::Builder::new(zstd_encoder);

    let total = files.len();
    for (i, path) in files.iter().enumerate() {
        let rel = path
            .strip_prefix(&directory)
            .unwrap_or(path)
            .to_string_lossy()
            .replace('\\', "/");
        let data = std::fs::read(path)
            .with_context(|| format!("Failed to read {}", path.display()))?;
        let mut header = tar::Header::new_gnu();
        header.set_size(data.len() as u64);
        header.set_mode(0o644);
        header.set_cksum();
        tar_builder
            .append_data(&mut header, &rel, data.as_slice())
            .with_context(|| format!("Failed to append {} to tar", rel))?;
        println!("   [{}/{}] {} ({} bytes)", i + 1, total, rel, data.len());
    }

    let zstd_encoder = tar_builder.into_inner().context("Failed to finish tar archive")?;
    zstd_encoder.finish().context("Failed to finish zstd compression")?;
    println!("\n✅ tar.zst archive complete: {} files packed", total);
    Ok(())
}

/// Convert a znippy archive to a `.tar.zst` archive by staging through a temp
/// directory (unpack → repack). The direct archive↔archive conversions compose
/// the directory primitives rather than duplicating format logic.
pub async fn convert_znippy_to_tar_zstd(input: PathBuf, output: PathBuf) -> Result<()> {
    let staging = tempfile::tempdir().context("create staging dir for znippy→tar.zst")?;
    unpack_znippy_to_directory(input, staging.path().to_path_buf()).await?;
    pack_directory_to_tar_zstd(staging.path().to_path_buf(), output).await
}

/// Convert a `.tar.zst` archive to a znippy archive by staging through a temp
/// directory (unpack → repack). `format` optionally forces the znippy pkg_type.
pub async fn convert_tar_zstd_to_znippy(
    input: PathBuf,
    output: PathBuf,
    format: Option<&str>,
) -> Result<()> {
    let staging = tempfile::tempdir().context("create staging dir for tar.zst→znippy")?;
    unpack_tar_zstd_to_directory(input, staging.path().to_path_buf()).await?;
    pack_directory_to_znippy(staging.path().to_path_buf(), output, format).await
}

fn extract_crate_relative_path(path: &std::path::Path) -> PathBuf {
    let components: Vec<_> = path.components().collect();
    for (i, comp) in components.iter().enumerate() {
        if comp.as_os_str() == "crates" {
            return components[i..].iter().collect();
        }
    }
    PathBuf::from(path.file_name().unwrap_or_default())
}

fn find_crate_files(dir: &std::path::Path) -> Result<Vec<PathBuf>> {
    let mut files = Vec::new();
    walk_dir_recursive(dir, &mut files)?;
    files.sort();
    Ok(files)
}

fn walk_dir_recursive(dir: &std::path::Path, files: &mut Vec<PathBuf>) -> Result<()> {
    if !dir.is_dir() {
        return Ok(());
    }
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let path = entry.path();
        if path.is_dir() {
            walk_dir_recursive(&path, files)?;
        } else if path.extension().and_then(|e| e.to_str()) == Some("crate") {
            files.push(path);
        }
    }
    Ok(())
}

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

    #[test]
    fn safe_join_allows_normal_relative_paths() {
        let base = std::path::Path::new("/srv/out");
        let j = safe_join(base, "crates/se/serde/serde-1.0.0.crate").unwrap();
        assert_eq!(j, base.join("crates/se/serde/serde-1.0.0.crate"));
        // A leading `./` is harmless and normalized away.
        assert_eq!(safe_join(base, "./a/b").unwrap(), base.join("a/b"));
    }

    #[test]
    fn safe_join_rejects_hostile_upstream_paths() {
        let base = std::path::Path::new("/srv/out");
        for hostile in [
            "../../../../etc/cron.d/holger",
            "ok/../../../../etc/passwd",
            "/etc/passwd",
            "/etc/cron.d/x",
        ] {
            assert!(
                safe_join(base, hostile).is_err(),
                "must reject hostile asset path {hostile:?}"
            );
        }
        // Confirm none of them could ever escape: the join stays under base.
        assert!(safe_join(base, "a/../b").is_err()); // any ParentDir is refused
    }

    /// Drift guard (validate as DATA): holger's curated extension → pkg_type
    /// table must agree with znippy's plugin handlers wherever znippy-common can
    /// confirm it. The expected discriminants are pulled from the znippy handler's
    /// own `type_id()` — never a literal copied into this test — and each chosen
    /// extension must be one the handler's `meta().extensions` actually claims, so
    /// holger's table provably stays a SUBSET of znippy's source of truth.
    #[test]
    fn pkg_type_for_file_agrees_with_znippy_plugin_extensions() {
        use znippy_common::plugin::ArchiveTypePlugin;
        use znippy_common::plugins::cargo_native::CargoPlugin;
        use znippy_common::plugins::gem_native::GemPlugin;
        use znippy_common::plugins::skeletons;

        let cargo = CargoPlugin::new();
        let gem = GemPlugin;
        let nuget = skeletons::NugetPlugin;

        let cases: Vec<(&str, i8, Vec<String>)> = vec![
            ("serde-1.0.0.crate", cargo.type_id(), cargo.meta().extensions),
            ("rails-7.1.0.gem", gem.type_id(), gem.meta().extensions),
            ("Newtonsoft.Json.13.0.3.nupkg", nuget.type_id(), nuget.meta().extensions),
        ];

        for (filename, znippy_id, exts) in &cases {
            // The chosen sample matches an extension znippy's handler claims.
            assert!(
                exts.iter().any(|e| filename.ends_with(e.as_str())),
                "{filename}: not covered by znippy handler claims {exts:?}"
            );
            // holger resolves the SAME discriminant znippy's handler writes.
            assert_eq!(
                pkg_type_for_file(std::path::Path::new(filename)),
                Some(*znippy_id),
                "{filename}: holger pkg_type drifted from znippy handler type_id {znippy_id}"
            );
        }

        // jar/whl resolve through ArtifactFormat → znippy's native discriminants
        // (maven/python handlers live in sibling plugin crates this crate doesn't
        // depend on, so their numbers come via the znippy-sourced ArtifactFormat).
        assert_eq!(
            pkg_type_for_file(std::path::Path::new("guava-33.0.jar")),
            Some(traits::ArtifactFormat::Maven3.znippy_type_id()),
        );
        assert_eq!(
            pkg_type_for_file(std::path::Path::new("numpy-1.26-cp311.whl")),
            Some(traits::ArtifactFormat::Pip.znippy_type_id()),
        );

        // Curated-adapter policy: ambiguous / secondary extensions znippy ALSO
        // claims (.tar.gz/.zip via python, .tgz via npm/helm, .snupkg via nuget)
        // stay UNtyped here so the agent never mis-tags an sdist or symbol package;
        // an explicit `--format` override resolves them instead.
        for ambiguous in ["pkg-1.0.tar.gz", "pkg-1.0.tgz", "pkg-1.0.zip", "sym-1.0.snupkg"] {
            assert_eq!(
                pkg_type_for_file(std::path::Path::new(ambiguous)),
                None,
                "{ambiguous} must stay untyped (curated adapter policy)"
            );
        }
    }

    #[test]
    fn read_capped_accepts_within_limit_and_rejects_bomb() {
        // Under the cap: returned verbatim.
        let mut ok = std::io::Cursor::new(vec![7u8; 100]);
        let out = read_capped(&mut ok, 128, "small").unwrap();
        assert_eq!(out.len(), 100);

        // Exactly at the cap is allowed.
        let mut at = std::io::Cursor::new(vec![1u8; 128]);
        assert_eq!(read_capped(&mut at, 128, "edge").unwrap().len(), 128);

        // Over the cap (decompression-bomb stand-in): rejected, not buffered.
        let mut bomb = std::io::Cursor::new(vec![0u8; 1000]);
        let err = read_capped(&mut bomb, 128, "bomb").expect_err("must reject over-cap entry");
        assert!(err.to_string().contains("exceeds per-entry cap"), "got: {err}");
    }

    /// Build a small tree of fixture files under `root`, returning the
    /// (relative-path, bytes) pairs so a round-trip can assert byte-for-byte
    /// recovery. Uses real crate-like names so pkg_type detection has something to
    /// chew on.
    fn seed_fixture_tree(root: &std::path::Path) -> Vec<(String, Vec<u8>)> {
        let fixtures = vec![
            ("crates/se/serde/serde-1.0.0.crate".to_string(), b"serde crate bytes \x00\x01\x02".to_vec()),
            ("crates/to/tokio/tokio-1.2.3.crate".to_string(), vec![0xABu8; 4096]),
            ("readme.txt".to_string(), b"plain top-level file".to_vec()),
        ];
        for (rel, data) in &fixtures {
            let dest = root.join(rel);
            std::fs::create_dir_all(dest.parent().unwrap()).unwrap();
            std::fs::write(&dest, data).unwrap();
        }
        fixtures
    }

    /// Read every file the fixtures declared back out of `dir` and assert each
    /// recovered byte-for-byte.
    fn assert_tree_recovered(dir: &std::path::Path, fixtures: &[(String, Vec<u8>)]) {
        for (rel, data) in fixtures {
            let got = std::fs::read(dir.join(rel))
                .unwrap_or_else(|e| panic!("recovered file {rel} missing: {e}"));
            assert_eq!(&got, data, "file {rel} must round-trip byte-for-byte");
        }
    }

    /// Directory → znippy → directory recovers every file byte-for-byte. Exercises
    /// [`pack_directory_to_znippy`] + the new [`unpack_znippy_to_directory`].
    #[tokio::test]
    async fn znippy_directory_round_trip_is_byte_exact() {
        let work = tempfile::tempdir().unwrap();
        let src = work.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        let fixtures = seed_fixture_tree(&src);

        let archive = work.path().join("out.znippy");
        pack_directory_to_znippy(src, archive.clone(), None)
            .await
            .expect("pack directory → znippy");
        assert!(archive.exists(), "znippy archive was written");

        let out = work.path().join("out");
        unpack_znippy_to_directory(archive, out.clone())
            .await
            .expect("unpack znippy → directory");
        assert_tree_recovered(&out, &fixtures);
    }

    /// Directory → tar.zst → directory recovers every file byte-for-byte. Exercises
    /// the new [`pack_directory_to_tar_zstd`] + [`unpack_tar_zstd_to_directory`].
    #[tokio::test]
    async fn tar_zstd_directory_round_trip_is_byte_exact() {
        let work = tempfile::tempdir().unwrap();
        let src = work.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        let fixtures = seed_fixture_tree(&src);

        let archive = work.path().join("out.tar.zst");
        pack_directory_to_tar_zstd(src, archive.clone())
            .await
            .expect("pack directory → tar.zst");
        assert!(archive.exists(), "tar.zst archive was written");

        let out = work.path().join("out");
        unpack_tar_zstd_to_directory(archive, out.clone())
            .await
            .expect("unpack tar.zst → directory");
        assert_tree_recovered(&out, &fixtures);
    }

    /// The archive↔archive conversions (staged through a temp directory) recover
    /// every file byte-for-byte in both directions: znippy → tar.zst → directory,
    /// and tar.zst → znippy → directory.
    #[tokio::test]
    async fn cross_archive_conversions_are_byte_exact() {
        let work = tempfile::tempdir().unwrap();
        let src = work.path().join("src");
        std::fs::create_dir_all(&src).unwrap();
        let fixtures = seed_fixture_tree(&src);

        // Seed one znippy + one tar.zst from the same tree.
        let znippy = work.path().join("a.znippy");
        let tarzst = work.path().join("a.tar.zst");
        pack_directory_to_znippy(src.clone(), znippy.clone(), None).await.unwrap();
        pack_directory_to_tar_zstd(src, tarzst.clone()).await.unwrap();

        // znippy → tar.zst → directory.
        let z2t = work.path().join("z2t.tar.zst");
        convert_znippy_to_tar_zstd(znippy, z2t.clone()).await.expect("znippy → tar.zst");
        let out1 = work.path().join("out1");
        unpack_tar_zstd_to_directory(z2t, out1.clone()).await.unwrap();
        assert_tree_recovered(&out1, &fixtures);

        // tar.zst → znippy → directory.
        let t2z = work.path().join("t2z.znippy");
        convert_tar_zstd_to_znippy(tarzst, t2z.clone(), None).await.expect("tar.zst → znippy");
        let out2 = work.path().join("out2");
        unpack_znippy_to_directory(t2z, out2.clone()).await.unwrap();
        assert_tree_recovered(&out2, &fixtures);
    }
}