nordisk/holger

server/repository/znippy-rust/tests/integration_znippy_serve.rs

//! Integration test: download real artifacts from crates.io, Maven Central, and PyPI,
//! pack each into znippy archives, then serve and benchmark retrieval.
//!
//! Downloads ~5GB per type (cached in ~/work/holger_tests).
//! Run with: cargo test --test integration_znippy_serve -- --ignored --nocapture
//!
//! The archives are kept between runs for caching. Delete /tmp/holger_int_test to start fresh.

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

use repository_znippy_maven::MavenRepoZnippy;
use repository_znippy_python::PipRepoZnippy;
use repository_znippy_rust::RustRepoZnippy;
use traits::RepositoryBackendTrait;
use znippy_compress::{compress_stream, ArchiveEntry, StreamCompressor};

#[cfg(feature = "testmatrix")]
use traits::test_support::fstatus;

/// Offline SKIP gate — mirrors the `is_skip` semantics from
/// `bench-scenarios/src/nexus.rs`: a heavy test that needs the network probes
/// for reachability first and, when offline, prints a `skipped: ...` line and
/// returns early instead of grinding through hundreds of timing-out requests
/// (which is what hung `cargo test --workspace`). Returns `true` if the test
/// should skip (no network). Probes the same host the test will then hammer.
fn network_skip(probe_url: &str) -> bool {
    let client = match tls_roots_blocking(reqwest::blocking::Client::builder())
        .user_agent("holger-integration-test/0.1 (+https://git.gunnar.rs/nordisk/holger)")
        .timeout(Duration::from_secs(10))
        .build()
    {
        Ok(c) => c,
        Err(e) => {
            eprintln!("skipped: no network — failed to build probe client: {e}");
            return true;
        }
    };
    match client.head(probe_url).send() {
        Ok(_) => false,
        Err(e) => {
            eprintln!("skipped: no network — {probe_url} unreachable: {e}");
            true
        }
    }
}

/// `$HOME/work/holger_tests` — the pre-cached znippy archives.
fn holger_test_dir() -> String {
    format!(
        "{}/work/holger_tests",
        std::env::var("HOME").unwrap_or_default()
    )
}
fn archive_base() -> String {
    format!("{}/crates_archive", holger_test_dir())
}
fn jar_archive_base() -> String {
    format!("{}/jars_archive", holger_test_dir())
}
fn pip_archive_base() -> String {
    format!("{}/pip_archive", holger_test_dir())
}

/// Target download size in bytes per type
const TARGET_BYTES: u64 = 4 * 1024 * 1024 * 1024;

/// Popular crates — we'll fetch ALL versions of each until we hit the target size.
/// Ordered roughly by download size (largest first to fill up fast).
const CRATE_NAMES: &[&str] = &[
    // Huge crates (10-100MB+ per version)
    "windows",
    "windows-sys",
    "ring",
    "openssl-sys",
    "cargo",
    "rustc-ap-rustc_ast",
    "aws-sdk-s3",
    "aws-sdk-ec2",
    "aws-sdk-dynamodb",
    "aws-sdk-lambda",
    "aws-sdk-iam",
    "aws-sdk-sqs",
    "aws-sdk-sns",
    "aws-sdk-cloudformation",
    // Medium crates (1-10MB per version)
    "image",
    "nix",
    "syn",
    "tokio",
    "clap",
    "hyper",
    "reqwest",
    "axum",
    "rustls",
    "regex",
    "chrono",
    "serde",
    "serde_json",
    "serde_derive",
    "diesel",
    "sqlx",
    "tonic",
    "prost",
    "warp",
    "actix-web",
    "rocket",
    // Smaller but many versions
    "proc-macro2",
    "quote",
    "rand",
    "anyhow",
    "thiserror",
    "futures",
    "bytes",
    "log",
    "tracing",
    "tower",
    "http",
    "url",
    "uuid",
    "base64",
    "sha2",
    "hmac",
    "aes",
    "ed25519-dalek",
    "curve25519-dalek",
    "x25519-dalek",
    "libc",
    "cc",
    "bindgen",
    "num",
    "itertools",
    "rayon",
    "crossbeam",
    "parking_lot",
    "dashmap",
    "flate2",
    "zstd",
    "tar",
    "zip",
    "walkdir",
    "glob",
    "tempfile",
    "once_cell",
    "lazy_static",
    "toml",
    "config",
    "clap_derive",
    "structopt",
    "env_logger",
    "tracing-subscriber",
    "tower-http",
    "tungstenite",
    "rusqlite",
    "redis",
    "lapin",
    "rdkafka",
    "paho-mqtt",
    "arrow",
    "arrow-array",
    "parquet",
    "datafusion",
    "polars",
];

/// Fetch all version strings for a crate from the crates.io API
fn fetch_versions(client: &reqwest::blocking::Client, crate_name: &str) -> Vec<String> {
    let url = format!("https://crates.io/api/v1/crates/{}/versions", crate_name);

    let resp = match client.get(&url).send() {
        Ok(r) if r.status().is_success() => r,
        Ok(r) => {
            eprintln!("  ⚠ API {} for {}", r.status(), crate_name);
            return vec![];
        }
        Err(e) => {
            eprintln!("  ⚠ API error for {}: {}", crate_name, e);
            return vec![];
        }
    };

    let body: serde_json::Value = match resp.json() {
        Ok(v) => v,
        Err(_) => return vec![],
    };

    body["versions"]
        .as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|v| {
                    // Skip yanked versions
                    if v["yanked"].as_bool() == Some(true) {
                        return None;
                    }
                    v["num"].as_str().map(|s| s.to_string())
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Download a .crate file from crates.io. Returns bytes or None on failure.
fn download_crate(
    client: &reqwest::blocking::Client,
    name: &str,
    version: &str,
) -> Option<Vec<u8>> {
    let url = format!(
        "https://static.crates.io/crates/{}/{}-{}.crate",
        name, name, version
    );

    match client.get(&url).send() {
        Ok(resp) if resp.status().is_success() => match resp.bytes() {
            Ok(b) => Some(b.to_vec()),
            Err(_) => None,
        },
        _ => None,
    }
}

/// Phase 1: Download crates until we hit target size (cached on disk)
fn download_all_crates(download_dir: &Path) -> u64 {
    fs::create_dir_all(download_dir).unwrap();

    // Check existing cache size first
    let existing_bytes: u64 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "crate"))
        .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
        .sum();

    if existing_bytes >= TARGET_BYTES {
        eprintln!(
            "✅ Cache already has {:.2} GB — skipping downloads",
            existing_bytes as f64 / 1_073_741_824.0
        );
        return existing_bytes;
    }

    eprintln!(
        "📡 Current cache: {:.2} GB, target: {:.2} GB",
        existing_bytes as f64 / 1_073_741_824.0,
        TARGET_BYTES as f64 / 1_073_741_824.0
    );

    let client = tls_roots_blocking(reqwest::blocking::Client::builder())
        .user_agent("holger-integration-test/0.1 (+https://git.gunnar.rs/nordisk/holger)")
        .timeout(std::time::Duration::from_secs(120))
        .build()
        .unwrap();

    let mut total_bytes = existing_bytes;
    let mut total_files: u32 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "crate"))
        .count() as u32;

    let start = Instant::now();

    // Bound the crawl: if the network drops mid-run (a "missed bind"), don't grind
    // through hundreds of timing-out requests — bail after this many crates yield
    // zero usable versions in a row.
    const MAX_CONSECUTIVE_EMPTY: u32 = 5;
    let mut consecutive_empty: u32 = 0;

    'outer: for crate_name in CRATE_NAMES {
        if total_bytes >= TARGET_BYTES {
            break;
        }

        eprintln!("\n📦 Fetching versions for {}...", crate_name);
        let versions = fetch_versions(&client, crate_name);
        if versions.is_empty() {
            consecutive_empty += 1;
            assert!(
                consecutive_empty < MAX_CONSECUTIVE_EMPTY,
                "crates.io returned no versions for {MAX_CONSECUTIVE_EMPTY} crates in a row \
                 — network likely down; aborting instead of spinning"
            );
            continue;
        }
        consecutive_empty = 0;
        eprintln!("   Found {} versions", versions.len());

        for version in &versions {
            if total_bytes >= TARGET_BYTES {
                break 'outer;
            }

            let filename = format!("{}-{}.crate", crate_name, version);
            let file_path = download_dir.join(&filename);

            // Skip if already cached
            if file_path.exists() {
                continue;
            }

            if let Some(data) = download_crate(&client, crate_name, version) {
                let size = data.len() as u64;
                fs::write(&file_path, &data).unwrap();
                total_bytes += size;
                total_files += 1;

                if total_files.is_multiple_of(50) {
                    let elapsed = start.elapsed().as_secs_f64();
                    let rate = (total_bytes - existing_bytes) as f64 / elapsed / 1_048_576.0;
                    eprintln!(
                        "  ⬇️  {} files | {:.2} GB | {:.1} MB/s | ETA: {:.0}s",
                        total_files,
                        total_bytes as f64 / 1_073_741_824.0,
                        rate,
                        (TARGET_BYTES - total_bytes) as f64 / (rate * 1_048_576.0),
                    );
                }

                // Be nice to crates.io — small delay every 10 downloads
                if total_files.is_multiple_of(10) {
                    std::thread::sleep(std::time::Duration::from_millis(100));
                }
            }
        }
    }

    eprintln!(
        "\n✅ {} crate files ready ({:.2} GB)",
        total_files,
        total_bytes as f64 / 1_073_741_824.0
    );
    total_bytes
}

/// Phase 2: Pack all downloaded .crate files into a znippy archive
fn pack_into_znippy(download_dir: &Path, archive_base: &Path) -> PathBuf {
    let index_path = archive_base.with_extension("znippy");

    // If archive already exists and is substantial, skip repacking
    if index_path.exists() {
        let archive_size = fs::metadata(&index_path).unwrap().len();
        let download_size: u64 = fs::read_dir(download_dir)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.path().extension().is_some_and(|ext| ext == "crate"))
            .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
            .sum();

        // Reuse if archive is at least 80% of source data size
        if archive_size as f64 > download_size as f64 * 0.8 {
            eprintln!(
                "♻️  Reusing existing archive ({:.2} GB)",
                archive_size as f64 / 1_073_741_824.0
            );
            return index_path;
        }
        let _ = fs::remove_file(&index_path);
    }

    eprintln!("🔧 Packing crates into znippy archive...");
    let start = Instant::now();

    let compressor: StreamCompressor = compress_stream(&archive_base.to_path_buf(), true)
        .expect("Failed to create stream compressor");

    let sender = compressor.sender().clone();
    let mut packed_bytes = 0u64;

    // Read all .crate files from download dir and feed to compressor
    let mut entries: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "crate"))
        .collect();
    entries.sort_by_key(|e| e.path());

    let total_entries = entries.len();
    eprintln!("   Feeding {} files to compressor...", total_entries);

    for (i, entry) in entries.iter().enumerate() {
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let data = fs::read(&path).unwrap();
        packed_bytes += data.len() as u64;

        sender
            .send(ArchiveEntry {
                relative_path: filename,
                data,
                pkg_type: None,
                repo: None,
            })
            .unwrap();

        if (i + 1) % 500 == 0 {
            eprintln!(
                "   [{}/{}] {:.2} GB fed...",
                i + 1,
                total_entries,
                packed_bytes as f64 / 1_073_741_824.0
            );
        }
    }

    drop(sender);
    eprintln!("   Waiting for compression to finish...");
    let report = compressor.finish().expect("Compression failed");

    let elapsed = start.elapsed();
    let archive_size = fs::metadata(&index_path).map(|m| m.len()).unwrap_or(0);

    eprintln!(
        "✅ Packed {} files ({:.2} GB input → {:.2} GB archive) in {:.1}s (ratio: {:.1}%)",
        report.total_files,
        packed_bytes as f64 / 1_073_741_824.0,
        archive_size as f64 / 1_073_741_824.0,
        elapsed.as_secs_f64(),
        report.compression_ratio,
    );

    index_path
}

/// Phase 3: Open archive with RustRepoZnippy and verify serving
fn verify_serving(index_path: &Path, download_dir: &Path) {
    eprintln!("\n🔍 Opening archive and verifying extraction...");
    let start = Instant::now();

    let repo = RustRepoZnippy::with_archive("rust-prod".to_string(), index_path.to_path_buf())
        .expect("Failed to open znippy archive");

    let files = repo.list_files();
    eprintln!("   Archive contains {} files", files.len());
    assert!(!files.is_empty(), "Archive should contain files");

    // Verify random sample: extract from archive and compare to original
    let mut verified = 0u32;
    let mut verified_bytes: u64 = 0;

    // Pick every Nth file to get a representative sample
    let mut all_crate_files: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "crate"))
        .collect();
    all_crate_files.sort_by_key(|e| e.path());

    let sample_size = 100.min(all_crate_files.len());
    let step = all_crate_files.len() / sample_size;

    eprintln!(
        "   Verifying {} sample files (step={})...",
        sample_size, step
    );

    for i in (0..all_crate_files.len()).step_by(step).take(sample_size) {
        let entry = &all_crate_files[i];
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();

        let original = fs::read(&path).unwrap();
        let extracted = repo
            .get_file(&filename)
            .unwrap_or_else(|e| panic!("Failed to extract {}: {}", filename, e));

        assert_eq!(
            original.len(),
            extracted.len(),
            "Size mismatch for {}: original={} extracted={}",
            filename,
            original.len(),
            extracted.len()
        );
        assert_eq!(
            original,
            extracted,
            "Content mismatch for {} (first diff at byte {:?})",
            filename,
            original
                .iter()
                .zip(extracted.iter())
                .position(|(a, b)| a != b)
        );

        verified += 1;
        verified_bytes += original.len() as u64;
    }

    // Also test HTTP handler paths
    if let Some(first_file) = all_crate_files.first() {
        let path = first_file.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        // Parse "name-version.crate" → (name, version)
        if let Some(stem) = filename.strip_suffix(".crate") {
            if let Some(dash_pos) = stem.rfind('-') {
                let name = &stem[..dash_pos];
                let version = &stem[dash_pos + 1..];
                let url = format!("/rust-prod/crates/{}/{}/download", name, version);
                let (status, _headers, body) = repo.handle_http2_request("GET", &url, &[]).unwrap();
                assert_eq!(status, 200, "HTTP download should return 200 for {}", url);
                let original = fs::read(&path).unwrap();
                assert_eq!(body, original, "HTTP body mismatch for {}", filename);
            }
        }
    }

    // Test 404 for nonexistent crate
    let (status, _, _) = repo
        .handle_http2_request(
            "GET",
            "/rust-prod/crates/nonexistent-crate-xyz/0.0.0/download",
            &[],
        )
        .unwrap();
    assert_eq!(status, 404);

    let elapsed = start.elapsed();
    eprintln!(
        "✅ Verified {} crates ({:.2} GB) byte-for-byte in {:.1}s",
        verified,
        verified_bytes as f64 / 1_073_741_824.0,
        elapsed.as_secs_f64()
    );
}

#[test]
#[ignore = "heavy: real packages + serve; --ignored pass"]
fn test_znippy_archive_full_roundtrip() {
    let _ = env_logger::try_init();

    if network_skip("https://crates.io/") {
        return;
    }

    let test_dir_buf = holger_test_dir();
    let test_dir = Path::new(&test_dir_buf);
    let download_dir = test_dir.join("crates");
    let archive_base = PathBuf::from(archive_base());

    eprintln!("\n═══════════════════════════════════════════════════════════════");
    eprintln!("  🐱 Holger Znippy Integration Test — ~10GB crate archive");
    eprintln!("  Test dir: {}", holger_test_dir());
    eprintln!("  Strategy: download ALL versions of popular crates");
    eprintln!("═══════════════════════════════════════════════════════════════\n");

    // Phase 1: Download
    let start = Instant::now();
    let _total_bytes = download_all_crates(&download_dir);

    // Phase 2: Pack into znippy
    let index_path = pack_into_znippy(&download_dir, &archive_base);

    // Phase 3: Verify serving
    verify_serving(&index_path, &download_dir);

    let total_elapsed = start.elapsed();
    eprintln!(
        "\n🏁 Full integration test completed in {:.1}s",
        total_elapsed.as_secs_f64()
    );
}

/// Smaller smoke test — just 5 crates, for CI
#[test]
#[ignore = "heavy: real packages + serve; --ignored pass"]
fn test_znippy_archive_smoke() {
    let _ = env_logger::try_init();

    if network_skip("https://static.crates.io/") {
        return;
    }

    let test_dir = tempfile::tempdir().unwrap();
    let download_dir = test_dir.path().join("crates");
    fs::create_dir_all(&download_dir).unwrap();

    let client = tls_roots_blocking(reqwest::blocking::Client::builder())
        .user_agent("holger-integration-test/0.1")
        .timeout(std::time::Duration::from_secs(30))
        .build()
        .unwrap();

    let crates_to_test = [
        ("serde", "1.0.200"),
        ("log", "0.4.22"),
        ("anyhow", "1.0.86"),
        ("bytes", "1.7.1"),
        ("rand", "0.8.5"),
    ];

    // Download
    for (name, version) in &crates_to_test {
        let data = download_crate(&client, name, version)
            .unwrap_or_else(|| panic!("Failed to download {}-{}", name, version));
        let filename = format!("{}-{}.crate", name, version);
        fs::write(download_dir.join(&filename), &data).unwrap();
    }

    // Pack
    let archive_base = test_dir.path().join("smoke_archive");
    let compressor = compress_stream(&archive_base, true).unwrap();
    let sender = compressor.sender().clone();

    for (name, version) in &crates_to_test {
        let filename = format!("{}-{}.crate", name, version);
        let data = fs::read(download_dir.join(&filename)).unwrap();
        sender
            .send(ArchiveEntry {
                relative_path: filename,
                data,
                pkg_type: None,
                repo: None,
            })
            .unwrap();
    }
    drop(sender);
    let _report = compressor.finish().unwrap();

    // Serve and verify
    let index_path = archive_base.with_extension("znippy");
    let repo = RustRepoZnippy::with_archive("test".to_string(), index_path)
        .expect("Failed to open archive");

    assert_eq!(repo.list_files().len(), 5);

    for (name, version) in &crates_to_test {
        let filename = format!("{}-{}.crate", name, version);
        let original = fs::read(download_dir.join(&filename)).unwrap();
        let extracted = repo.get_file(&filename).unwrap();
        assert_eq!(original, extracted, "Mismatch for {}", filename);
    }

    // HTTP path test
    let (status, _, body) = repo
        .handle_http2_request("GET", "/test/crates/serde/1.0.200/download", &[])
        .unwrap();
    assert_eq!(status, 200);
    let original = fs::read(download_dir.join("serde-1.0.200.crate")).unwrap();
    assert_eq!(body, original);

    #[cfg(feature = "testmatrix")]
    fstatus(
        "znippy-rust",
        "archive_smoke_roundtrip",
        repo.list_files().len() == 5 && status == 200 && body == original,
        &format!(
            "5 crates packed+served via znippy; HTTP download status={status}, {} bytes match",
            body.len()
        ),
    );

    eprintln!("✅ Smoke test passed — 5 crates roundtripped through znippy");
}

/// Hermetic sibling of `test_znippy_archive_smoke` — NO network, NO `#[ignore]`.
///
/// Exercises the exact same pack → serve → retrieve path (`compress_stream` →
/// `RustRepoZnippy::with_archive` → `get_file` / `handle_http2_request`) but
/// synthesizes the `.crate` payloads locally instead of downloading ~real
/// artifacts. znippy stores and serves opaque bytes, so the routing, archive
/// open, file listing, byte-roundtrip, and HTTP `/download` mapping are all
/// covered deterministically. This is the cell that actually RUNS in
/// `cargo test --workspace` (and airgap/CI) — the `#[ignore]`d heavy tests above
/// only ever run with `--ignored` + a live network, so without this the
/// serve path had no always-on coverage.
#[test]
fn test_znippy_archive_smoke_hermetic() {
    let _ = env_logger::try_init();

    let test_dir = tempfile::tempdir().unwrap();

    let crates_to_test = [
        ("serde", "1.0.200"),
        ("log", "0.4.22"),
        ("anyhow", "1.0.86"),
        ("bytes", "1.7.1"),
        ("rand", "0.8.5"),
    ];

    // Synthesize deterministic, non-trivial `.crate` payloads (a few KB each,
    // varied per crate) so packing exercises real chunking/compression rather
    // than a degenerate empty file. Content is opaque to znippy.
    let synth = |name: &str, version: &str| -> Vec<u8> {
        let mut data = format!("synthetic-crate {name} {version}\n").into_bytes();
        let seed = name.bytes().fold(0u8, |a, b| a.wrapping_add(b));
        for i in 0..4096u32 {
            data.push(seed.wrapping_add((i % 251) as u8));
        }
        data
    };

    let mut originals: Vec<(String, Vec<u8>)> = Vec::new();
    for (name, version) in &crates_to_test {
        let filename = format!("{}-{}.crate", name, version);
        originals.push((filename, synth(name, version)));
    }

    // Pack
    let archive_base = test_dir.path().join("smoke_archive_hermetic");
    let compressor = compress_stream(&archive_base, true).unwrap();
    let sender = compressor.sender().clone();
    for (filename, data) in &originals {
        sender
            .send(ArchiveEntry {
                relative_path: filename.clone(),
                data: data.clone(),
                pkg_type: None,
                repo: None,
            })
            .unwrap();
    }
    drop(sender);
    let _report = compressor.finish().unwrap();

    // Serve and verify
    let index_path = archive_base.with_extension("znippy");
    let repo = RustRepoZnippy::with_archive("test".to_string(), index_path)
        .expect("Failed to open archive");

    assert_eq!(repo.list_files().len(), originals.len());

    for (filename, original) in &originals {
        let extracted = repo.get_file(filename).unwrap();
        assert_eq!(original, &extracted, "Mismatch for {filename}");
    }

    // HTTP path test — the crate-download URL maps to `{name}-{version}.crate`.
    let (status, _, body) = repo
        .handle_http2_request("GET", "/test/crates/serde/1.0.200/download", &[])
        .unwrap();
    assert_eq!(status, 200);
    let serde_original = &originals[0].1;
    assert_eq!(&body, serde_original);

    #[cfg(feature = "testmatrix")]
    fstatus(
        "znippy-rust",
        "archive_smoke_roundtrip_hermetic",
        repo.list_files().len() == originals.len() && status == 200 && &body == serde_original,
        &format!(
            "5 synthetic crates packed+served via znippy offline; HTTP download status={status}, {} bytes match",
            body.len()
        ),
    );

    eprintln!("✅ Hermetic smoke test passed — 5 synthetic crates roundtripped through znippy (no network)");
}

// ══════════════════════════════════════════════════════════════════════════════
//  Maven Central — download, pack, verify
// ══════════════════════════════════════════════════════════════════════════════

const MAVEN_ARTIFACTS: &[(&str, &str)] = &[
    ("org.springframework.boot", "spring-boot"),
    ("org.springframework.boot", "spring-boot-starter-web"),
    ("org.springframework.boot", "spring-boot-autoconfigure"),
    ("org.springframework", "spring-core"),
    ("org.springframework", "spring-context"),
    ("org.springframework", "spring-web"),
    ("org.springframework", "spring-beans"),
    ("com.google.guava", "guava"),
    ("org.apache.commons", "commons-lang3"),
    ("commons-io", "commons-io"),
    ("org.apache.commons", "commons-compress"),
    ("com.fasterxml.jackson.core", "jackson-databind"),
    ("com.fasterxml.jackson.core", "jackson-core"),
    ("com.fasterxml.jackson.core", "jackson-annotations"),
    ("org.slf4j", "slf4j-api"),
    ("ch.qos.logback", "logback-classic"),
    ("ch.qos.logback", "logback-core"),
    ("org.apache.logging.log4j", "log4j-core"),
    ("org.apache.logging.log4j", "log4j-api"),
    ("io.netty", "netty-all"),
    ("io.netty", "netty-buffer"),
    ("io.netty", "netty-transport"),
    ("org.apache.kafka", "kafka-clients"),
    ("junit", "junit"),
    ("org.junit.jupiter", "junit-jupiter-api"),
    ("org.mockito", "mockito-core"),
    ("org.projectlombok", "lombok"),
    ("com.google.code.gson", "gson"),
    ("org.apache.tomcat.embed", "tomcat-embed-core"),
    ("io.grpc", "grpc-core"),
    ("io.grpc", "grpc-netty"),
    ("com.squareup.okhttp3", "okhttp"),
    ("org.hibernate.orm", "hibernate-core"),
    ("org.postgresql", "postgresql"),
    ("com.zaxxer", "HikariCP"),
    ("org.apache.lucene", "lucene-core"),
    ("software.amazon.awssdk", "s3"),
    ("software.amazon.awssdk", "dynamodb"),
    ("com.amazonaws", "aws-java-sdk-s3"),
    ("com.amazonaws", "aws-java-sdk-core"),
    ("io.micrometer", "micrometer-core"),
    ("org.apache.maven", "maven-core"),
    ("org.apache.maven.plugins", "maven-compiler-plugin"),
    ("mysql", "mysql-connector-java"),
    ("io.vertx", "vertx-core"),
    ("org.eclipse.jetty", "jetty-server"),
    ("com.google.protobuf", "protobuf-java"),
    ("io.quarkus", "quarkus-core"),
    ("org.apache.camel", "camel-core"),
    ("org.apache.spark", "spark-core_2.13"),
];

fn maven_fetch_versions(
    client: &reqwest::blocking::Client,
    group_id: &str,
    artifact_id: &str,
) -> Vec<String> {
    let url = format!(
        "https://search.maven.org/solrsearch/select?q=g:{}+AND+a:{}&core=gav&rows=200&wt=json",
        group_id, artifact_id
    );

    let resp = match client.get(&url).send() {
        Ok(r) if r.status().is_success() => r,
        _ => return vec![],
    };

    let body: serde_json::Value = match resp.json() {
        Ok(v) => v,
        Err(_) => return vec![],
    };

    body["response"]["docs"]
        .as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|doc| doc["v"].as_str().map(|s| s.to_string()))
                .collect()
        })
        .unwrap_or_default()
}

fn download_jar(
    client: &reqwest::blocking::Client,
    group_id: &str,
    artifact_id: &str,
    version: &str,
) -> Option<Vec<u8>> {
    let group_path = group_id.replace('.', "/");
    let url = format!(
        "https://repo1.maven.org/maven2/{}/{}/{}/{}-{}.jar",
        group_path, artifact_id, version, artifact_id, version
    );
    match client.get(&url).send() {
        Ok(resp) if resp.status().is_success() => resp.bytes().ok().map(|b| b.to_vec()),
        _ => None,
    }
}

fn download_all_jars(download_dir: &Path) -> u64 {
    fs::create_dir_all(download_dir).unwrap();

    let existing_bytes: u64 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "jar"))
        .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
        .sum();

    if existing_bytes >= TARGET_BYTES {
        eprintln!(
            "✅ JAR cache already has {:.2} GB — skipping",
            existing_bytes as f64 / 1_073_741_824.0
        );
        return existing_bytes;
    }

    eprintln!(
        "📡 JAR cache: {:.2} GB, target: {:.2} GB",
        existing_bytes as f64 / 1_073_741_824.0,
        TARGET_BYTES as f64 / 1_073_741_824.0
    );

    let client = tls_roots_blocking(reqwest::blocking::Client::builder())
        .user_agent("holger-integration-test/0.1 (+https://git.gunnar.rs/nordisk/holger)")
        .timeout(std::time::Duration::from_secs(120))
        .build()
        .unwrap();

    let mut total_bytes = existing_bytes;
    let mut total_files: u32 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "jar"))
        .count() as u32;

    let start = Instant::now();

    const MAX_CONSECUTIVE_EMPTY: u32 = 5;
    let mut consecutive_empty: u32 = 0;

    'outer: for (group_id, artifact_id) in MAVEN_ARTIFACTS {
        if total_bytes >= TARGET_BYTES {
            break;
        }

        eprintln!("\n📦 Maven versions for {}:{}...", group_id, artifact_id);
        let versions = maven_fetch_versions(&client, group_id, artifact_id);
        if versions.is_empty() {
            consecutive_empty += 1;
            assert!(
                consecutive_empty < MAX_CONSECUTIVE_EMPTY,
                "Maven Central returned no versions for {MAX_CONSECUTIVE_EMPTY} artifacts in a row \
                 — network likely down; aborting instead of spinning"
            );
            continue;
        }
        consecutive_empty = 0;
        eprintln!("   {} versions", versions.len());
        std::thread::sleep(std::time::Duration::from_millis(200));

        for version in &versions {
            if total_bytes >= TARGET_BYTES {
                break 'outer;
            }

            // Store with Maven path embedded in filename for later reconstruction
            let group_path = group_id.replace('.', "/");
            let archive_path = format!(
                "{}/{}/{}/{}-{}.jar",
                group_path, artifact_id, version, artifact_id, version
            );
            let safe_filename = archive_path.replace('/', "__");
            let file_path = download_dir.join(&safe_filename);

            if file_path.exists() {
                continue;
            }

            if let Some(data) = download_jar(&client, group_id, artifact_id, version) {
                let size = data.len() as u64;
                fs::write(&file_path, &data).unwrap();
                total_bytes += size;
                total_files += 1;

                if total_files.is_multiple_of(50) {
                    let elapsed = start.elapsed().as_secs_f64();
                    let rate = (total_bytes - existing_bytes) as f64 / elapsed / 1_048_576.0;
                    eprintln!(
                        "  ⬇️  {} JARs | {:.2} GB | {:.1} MB/s",
                        total_files,
                        total_bytes as f64 / 1_073_741_824.0,
                        rate
                    );
                }
                if total_files.is_multiple_of(5) {
                    std::thread::sleep(std::time::Duration::from_millis(50));
                }
            }
        }
    }

    eprintln!(
        "\n✅ {} JAR files ({:.2} GB)",
        total_files,
        total_bytes as f64 / 1_073_741_824.0
    );
    total_bytes
}

fn pack_jars_into_znippy(download_dir: &Path, archive_base: &Path) -> PathBuf {
    let index_path = archive_base.with_extension("znippy");

    if index_path.exists() {
        let archive_size = fs::metadata(&index_path).unwrap().len();
        let download_size: u64 = fs::read_dir(download_dir)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| e.path().extension().is_some_and(|ext| ext == "jar"))
            .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
            .sum();
        if archive_size as f64 > download_size as f64 * 0.8 {
            eprintln!(
                "♻️  Reusing JAR archive ({:.2} GB)",
                archive_size as f64 / 1_073_741_824.0
            );
            return index_path;
        }
        let _ = fs::remove_file(&index_path);
    }

    eprintln!("🔧 Packing JARs into znippy...");
    let start = Instant::now();

    let compressor: StreamCompressor =
        compress_stream(&archive_base.to_path_buf(), true).expect("Failed to create compressor");
    let sender = compressor.sender().clone();

    let mut entries: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "jar"))
        .collect();
    entries.sort_by_key(|e| e.path());

    let total = entries.len();
    let mut packed_bytes = 0u64;

    for (i, entry) in entries.iter().enumerate() {
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let data = fs::read(&path).unwrap();
        packed_bytes += data.len() as u64;

        // Restore Maven path from encoded filename: a__b__c.jar → a/b/c.jar
        let archive_path = filename.replace("__", "/");

        sender
            .send(ArchiveEntry {
                relative_path: archive_path,
                data,
                pkg_type: None,
                repo: None,
            })
            .unwrap();

        if (i + 1) % 500 == 0 {
            eprintln!(
                "   [{}/{}] {:.2} GB",
                i + 1,
                total,
                packed_bytes as f64 / 1_073_741_824.0
            );
        }
    }

    drop(sender);
    let report = compressor.finish().expect("JAR compression failed");
    let elapsed = start.elapsed();
    let archive_size = fs::metadata(&index_path).map(|m| m.len()).unwrap_or(0);

    eprintln!(
        "✅ {} JARs packed ({:.2} GB → {:.2} GB archive) in {:.1}s",
        report.total_files,
        packed_bytes as f64 / 1_073_741_824.0,
        archive_size as f64 / 1_073_741_824.0,
        elapsed.as_secs_f64()
    );

    index_path
}

fn verify_jar_serving(index_path: &Path, download_dir: &Path) {
    eprintln!("\n🔍 Verifying JAR serving...");
    let repo = MavenRepoZnippy::with_archive("maven".to_string(), index_path.to_path_buf())
        .expect("Failed to open JAR archive");

    let files = repo.list_files();
    eprintln!("   Archive: {} files", files.len());
    assert!(!files.is_empty());

    let mut all_files: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| e.path().extension().is_some_and(|ext| ext == "jar"))
        .collect();
    all_files.sort_by_key(|e| e.path());

    let sample_size = 100.min(all_files.len());
    let step = all_files.len().max(1) / sample_size.max(1);
    let mut verified = 0u32;

    for i in (0..all_files.len()).step_by(step.max(1)).take(sample_size) {
        let entry = &all_files[i];
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let archive_path = filename.replace("__", "/");
        let original = fs::read(&path).unwrap();
        let extracted = repo.get_file(&archive_path).unwrap();
        assert_eq!(original, extracted, "JAR mismatch: {}", archive_path);
        verified += 1;
    }

    // HTTP handler test
    if let Some(first) = all_files.first() {
        let path = first.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let archive_path = filename.replace("__", "/");
        let url = format!("/maven/{}", archive_path);
        let (status, _, body) = repo.handle_http2_request("GET", &url, &[]).unwrap();
        assert_eq!(status, 200, "HTTP 200 expected for {}", url);
        let original = fs::read(&path).unwrap();
        assert_eq!(body, original);
    }

    eprintln!("   ✅ Verified {} JARs + HTTP serving", verified);
}

// ══════════════════════════════════════════════════════════════════════════════
//  PyPI — download, pack, verify
// ══════════════════════════════════════════════════════════════════════════════

const PYPI_PACKAGES: &[&str] = &[
    "numpy",
    "pandas",
    "scipy",
    "matplotlib",
    "scikit-learn",
    "tensorflow",
    "torch",
    "transformers",
    "requests",
    "flask",
    "django",
    "fastapi",
    "uvicorn",
    "pydantic",
    "sqlalchemy",
    "celery",
    "boto3",
    "botocore",
    "awscli",
    "pillow",
    "opencv-python",
    "cryptography",
    "paramiko",
    "grpcio",
    "protobuf",
    "aiohttp",
    "httpx",
    "pytest",
    "black",
    "ruff",
    "mypy",
    "setuptools",
    "wheel",
    "pip",
    "click",
    "typer",
    "rich",
    "pyyaml",
    "jinja2",
    "markupsafe",
    "werkzeug",
    "gunicorn",
    "psycopg2-binary",
    "redis",
    "pymongo",
    "elasticsearch",
    "networkx",
    "sympy",
    "polars",
    "pyarrow",
    "dask",
    "xgboost",
    "lightgbm",
    "nltk",
    "spacy",
    "huggingface-hub",
    "tokenizers",
    "datasets",
    "tqdm",
    "joblib",
    "numba",
    "cython",
    "cffi",
    "six",
    "certifi",
    "urllib3",
    "packaging",
    "typing-extensions",
    "attrs",
    "more-itertools",
    "arrow",
    "pendulum",
    "pytz",
];

fn pypi_fetch_files(client: &reqwest::blocking::Client, package: &str) -> Vec<(String, String)> {
    let url = format!("https://pypi.org/pypi/{}/json", package);
    let resp = match client.get(&url).send() {
        Ok(r) if r.status().is_success() => r,
        _ => return vec![],
    };

    let body: serde_json::Value = match resp.json() {
        Ok(v) => v,
        Err(_) => return vec![],
    };

    let mut files = Vec::new();
    if let Some(releases) = body["releases"].as_object() {
        for (_version, release_files) in releases {
            if let Some(arr) = release_files.as_array() {
                for file_info in arr {
                    if let (Some(filename), Some(url)) =
                        (file_info["filename"].as_str(), file_info["url"].as_str())
                    {
                        if filename.ends_with(".whl") || filename.ends_with(".tar.gz") {
                            files.push((filename.to_string(), url.to_string()));
                        }
                    }
                }
            }
        }
    }
    files
}

fn download_all_wheels(download_dir: &Path) -> u64 {
    fs::create_dir_all(download_dir).unwrap();

    let is_pip_file = |e: &fs::DirEntry| {
        let p = e.path();
        let s = p.to_string_lossy();
        s.ends_with(".whl") || s.ends_with(".tar.gz")
    };

    let existing_bytes: u64 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| is_pip_file(e))
        .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
        .sum();

    if existing_bytes >= TARGET_BYTES {
        eprintln!(
            "✅ PyPI cache already has {:.2} GB — skipping",
            existing_bytes as f64 / 1_073_741_824.0
        );
        return existing_bytes;
    }

    eprintln!(
        "📡 PyPI cache: {:.2} GB, target: {:.2} GB",
        existing_bytes as f64 / 1_073_741_824.0,
        TARGET_BYTES as f64 / 1_073_741_824.0
    );

    let client = tls_roots_blocking(reqwest::blocking::Client::builder())
        .user_agent("holger-integration-test/0.1 (+https://git.gunnar.rs/nordisk/holger)")
        .timeout(std::time::Duration::from_secs(120))
        .build()
        .unwrap();

    let mut total_bytes = existing_bytes;
    let mut total_files: u32 = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| is_pip_file(e))
        .count() as u32;

    let start = Instant::now();

    const MAX_CONSECUTIVE_EMPTY: u32 = 5;
    let mut consecutive_empty: u32 = 0;

    'outer: for package in PYPI_PACKAGES {
        if total_bytes >= TARGET_BYTES {
            break;
        }

        eprintln!("\n🐍 PyPI files for {}...", package);
        let files = pypi_fetch_files(&client, package);
        if files.is_empty() {
            consecutive_empty += 1;
            assert!(
                consecutive_empty < MAX_CONSECUTIVE_EMPTY,
                "PyPI returned no files for {MAX_CONSECUTIVE_EMPTY} packages in a row \
                 — network likely down; aborting instead of spinning"
            );
            continue;
        }
        consecutive_empty = 0;
        eprintln!("   {} files across all versions", files.len());
        std::thread::sleep(std::time::Duration::from_millis(200));

        for (filename, url) in &files {
            if total_bytes >= TARGET_BYTES {
                break 'outer;
            }

            let file_path = download_dir.join(filename);
            if file_path.exists() {
                continue;
            }

            match client.get(url).send() {
                Ok(resp) if resp.status().is_success() => {
                    if let Ok(data) = resp.bytes() {
                        let size = data.len() as u64;
                        fs::write(&file_path, &data).unwrap();
                        total_bytes += size;
                        total_files += 1;

                        if total_files.is_multiple_of(50) {
                            let elapsed = start.elapsed().as_secs_f64();
                            let rate =
                                (total_bytes - existing_bytes) as f64 / elapsed / 1_048_576.0;
                            eprintln!(
                                "  ⬇️  {} files | {:.2} GB | {:.1} MB/s",
                                total_files,
                                total_bytes as f64 / 1_073_741_824.0,
                                rate
                            );
                        }
                        if total_files.is_multiple_of(10) {
                            std::thread::sleep(std::time::Duration::from_millis(50));
                        }
                    }
                }
                _ => {}
            }
        }
    }

    eprintln!(
        "\n✅ {} pip files ({:.2} GB)",
        total_files,
        total_bytes as f64 / 1_073_741_824.0
    );
    total_bytes
}

/// Normalize package name per PEP 503
fn normalize_pip_name(s: &str) -> String {
    s.to_lowercase().replace(['_', '.'], "-")
}

fn pack_wheels_into_znippy(download_dir: &Path, archive_base: &Path) -> PathBuf {
    let index_path = archive_base.with_extension("znippy");

    let is_pip_file = |e: &fs::DirEntry| {
        let s = e.path().to_string_lossy().to_string();
        s.ends_with(".whl") || s.ends_with(".tar.gz")
    };

    if index_path.exists() {
        let archive_size = fs::metadata(&index_path).unwrap().len();
        let download_size: u64 = fs::read_dir(download_dir)
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| is_pip_file(e))
            .map(|e| e.metadata().map(|m| m.len()).unwrap_or(0))
            .sum();
        if archive_size as f64 > download_size as f64 * 0.8 {
            eprintln!(
                "♻️  Reusing pip archive ({:.2} GB)",
                archive_size as f64 / 1_073_741_824.0
            );
            return index_path;
        }
        let _ = fs::remove_file(&index_path);
    }

    eprintln!("🔧 Packing wheels into znippy...");
    let start = Instant::now();

    let compressor: StreamCompressor =
        compress_stream(&archive_base.to_path_buf(), true).expect("Failed to create compressor");
    let sender = compressor.sender().clone();

    let mut entries: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| is_pip_file(e))
        .collect();
    entries.sort_by_key(|e| e.path());

    let total = entries.len();
    let mut packed_bytes = 0u64;

    for (i, entry) in entries.iter().enumerate() {
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let data = fs::read(&path).unwrap();
        packed_bytes += data.len() as u64;

        // PEP 503 layout: packages/{normalized-name}/{filename}
        let pkg_name = filename.split('-').next().unwrap_or("unknown");
        let normalized = normalize_pip_name(pkg_name);
        let archive_path = format!("packages/{}/{}", normalized, filename);

        sender
            .send(ArchiveEntry {
                relative_path: archive_path,
                data,
                pkg_type: None,
                repo: None,
            })
            .unwrap();

        if (i + 1) % 500 == 0 {
            eprintln!(
                "   [{}/{}] {:.2} GB",
                i + 1,
                total,
                packed_bytes as f64 / 1_073_741_824.0
            );
        }
    }

    drop(sender);
    let report = compressor.finish().expect("Pip compression failed");
    let elapsed = start.elapsed();
    let archive_size = fs::metadata(&index_path).map(|m| m.len()).unwrap_or(0);

    eprintln!(
        "✅ {} pip files packed ({:.2} GB → {:.2} GB archive) in {:.1}s",
        report.total_files,
        packed_bytes as f64 / 1_073_741_824.0,
        archive_size as f64 / 1_073_741_824.0,
        elapsed.as_secs_f64()
    );

    index_path
}

fn verify_pip_serving(index_path: &Path, download_dir: &Path) {
    eprintln!("\n🔍 Verifying pip serving...");
    let repo = PipRepoZnippy::with_archive("pip".to_string(), index_path.to_path_buf())
        .expect("Failed to open pip archive");

    let files = repo.list_files();
    eprintln!("   Archive: {} files", files.len());
    assert!(!files.is_empty());

    let is_pip_file = |e: &fs::DirEntry| {
        let s = e.path().to_string_lossy().to_string();
        s.ends_with(".whl") || s.ends_with(".tar.gz")
    };

    let mut all_files: Vec<_> = fs::read_dir(download_dir)
        .unwrap()
        .filter_map(|e| e.ok())
        .filter(|e| is_pip_file(e))
        .collect();
    all_files.sort_by_key(|e| e.path());

    let sample_size = 100.min(all_files.len());
    let step = all_files.len().max(1) / sample_size.max(1);
    let mut verified = 0u32;

    for i in (0..all_files.len()).step_by(step.max(1)).take(sample_size) {
        let entry = &all_files[i];
        let path = entry.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let pkg_name = filename.split('-').next().unwrap_or("unknown");
        let normalized = normalize_pip_name(pkg_name);
        let archive_path = format!("packages/{}/{}", normalized, filename);
        let original = fs::read(&path).unwrap();
        let extracted = repo.get_file(&archive_path).unwrap();
        assert_eq!(original, extracted, "Pip mismatch: {}", filename);
        verified += 1;
    }

    // PEP 503 simple index
    let (status, _, body) = repo
        .handle_http2_request("GET", "/pip/simple/", &[])
        .unwrap();
    assert_eq!(status, 200);
    let html = String::from_utf8_lossy(&body);
    let pkg_count = html.matches("<a href=").count();
    eprintln!("   PEP 503 index: {} packages", pkg_count);
    assert!(pkg_count > 0);

    // Package file download via HTTP
    if let Some(first) = all_files.first() {
        let path = first.path();
        let filename = path.file_name().unwrap().to_string_lossy().to_string();
        let pkg_name = filename.split('-').next().unwrap_or("unknown");
        let normalized = normalize_pip_name(pkg_name);
        let url = format!("/pip/packages/{}/{}", normalized, filename);
        let (status, _, body) = repo.handle_http2_request("GET", &url, &[]).unwrap();
        assert_eq!(status, 200, "HTTP 200 for {}", url);
        let original = fs::read(&path).unwrap();
        assert_eq!(body, original);
    }

    eprintln!("   ✅ Verified {} pip files + PEP 503 + HTTP", verified);
}

// ══════════════════════════════════════════════════════════════════════════════
//  Combined benchmark: download all 3 → znippy → bombard
// ══════════════════════════════════════════════════════════════════════════════

fn bench_retrieval<F>(
    label: &str,
    duration: Duration,
    num_threads: usize,
    get_fn: F,
) -> (u64, Duration)
where
    F: Fn(u64, usize) -> usize + Send + Sync + 'static,
{
    let get_fn = Arc::new(get_fn);
    let start = Instant::now();
    let total_ops = std::sync::atomic::AtomicU64::new(0);
    let total_ops_ref = &total_ops;

    std::thread::scope(|s| {
        for tid in 0..num_threads {
            let get_fn = Arc::clone(&get_fn);
            s.spawn(move || {
                let mut ops = 0u64;
                let deadline = Instant::now() + duration;
                while Instant::now() < deadline {
                    let _ = get_fn(ops, tid);
                    ops += 1;
                }
                total_ops_ref.fetch_add(ops, std::sync::atomic::Ordering::Relaxed);
            });
        }
    });

    let elapsed = start.elapsed();
    let ops = total_ops.load(std::sync::atomic::Ordering::Relaxed);
    let rate = ops as f64 / elapsed.as_secs_f64();
    eprintln!(
        "   {} ({} threads): {:>10.0} ops/sec ({} ops in {:.1}s)",
        label,
        num_threads,
        rate,
        ops,
        elapsed.as_secs_f64()
    );
    (ops, elapsed)
}

#[test]
#[ignore = "heavy: real packages + serve; --ignored pass"]
fn test_all_types_download_pack_benchmark() {
    let _ = env_logger::try_init();

    if network_skip("https://crates.io/")
        || network_skip("https://repo1.maven.org/")
        || network_skip("https://pypi.org/")
    {
        return;
    }

    let test_dir_buf = holger_test_dir();
    let test_dir = Path::new(&test_dir_buf);

    eprintln!("\n═══════════════════════════════════════════════════════════════");
    eprintln!("  🐱 Holger Znippy — Download, Pack & Benchmark ALL Types");
    eprintln!("  Cache dir: {}", holger_test_dir());
    eprintln!("  Target: ~5GB per type (Rust, Maven, pip)");
    eprintln!("═══════════════════════════════════════════════════════════════\n");

    let total_start = Instant::now();

    // ── Phase 1: Download all 3 in parallel ─────────────────────────────────

    let crate_dir = test_dir.join("crates");
    let jar_dir = test_dir.join("jars");
    let pip_dir = test_dir.join("wheels");

    eprintln!("⚡ Downloading all 3 types in parallel...\n");

    let crate_dir_c = crate_dir.clone();
    let jar_dir_c = jar_dir.clone();
    let pip_dir_c = pip_dir.clone();

    std::thread::scope(|s| {
        s.spawn(|| {
            eprintln!("━━━ [Thread 1] Rust Crates ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            download_all_crates(&crate_dir_c);
        });
        s.spawn(|| {
            eprintln!("━━━ [Thread 2] Maven JARs ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            download_all_jars(&jar_dir_c);
        });
        s.spawn(|| {
            eprintln!("━━━ [Thread 3] Python Wheels ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            download_all_wheels(&pip_dir_c);
        });
    });

    eprintln!("\n✅ All downloads complete! Packing into znippy archives...\n");

    // Pack sequentially (CPU-bound, uses all cores internally)
    let crate_archive_base = PathBuf::from(archive_base());
    let crate_index = pack_into_znippy(&crate_dir, &crate_archive_base);

    let jar_archive_base = PathBuf::from(jar_archive_base());
    let jar_index = pack_jars_into_znippy(&jar_dir, &jar_archive_base);

    let pip_archive_base = PathBuf::from(pip_archive_base());
    let pip_index = pack_wheels_into_znippy(&pip_dir, &pip_archive_base);

    // ── Phase 2: Verify correctness ──────────────────────────────────────
    eprintln!("\n━━━ Verification ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
    verify_serving(&crate_index, &crate_dir);
    verify_jar_serving(&jar_index, &jar_dir);
    verify_pip_serving(&pip_index, &pip_dir);

    // ── Phase 3: Benchmark bombardment ───────────────────────────────────
    eprintln!("\n━━━ Benchmark ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");

    let num_cpus = std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(4);
    let bench_duration = Duration::from_secs(10);

    eprintln!(
        "  CPUs: {}, duration: {}s per test\n",
        num_cpus,
        bench_duration.as_secs()
    );

    // Open repos
    let rust_repo =
        Arc::new(RustRepoZnippy::with_archive("rust".to_string(), crate_index).unwrap());
    let maven_repo =
        Arc::new(MavenRepoZnippy::with_archive("maven".to_string(), jar_index).unwrap());
    let pip_repo = Arc::new(PipRepoZnippy::with_archive("pip".to_string(), pip_index).unwrap());

    let rust_files: Vec<String> = rust_repo.list_files();
    let maven_files: Vec<String> = maven_repo.list_files();
    let pip_files: Vec<String> = pip_repo.list_files();

    eprintln!("  Rust archive:  {} files", rust_files.len());
    eprintln!("  Maven archive: {} files", maven_files.len());
    eprintln!("  Pip archive:   {} files\n", pip_files.len());

    // Single-threaded benchmarks
    eprintln!("── Single-threaded ──────────────────────────────────────────────");

    let rust_files_1 = rust_files.clone();
    let r1 = Arc::clone(&rust_repo);
    let (rust_ops_1t, rust_el_1t) =
        bench_retrieval("Rust crates", bench_duration, 1, move |i, _| {
            let f = &rust_files_1[(i as usize) % rust_files_1.len()];
            r1.get_file(f).unwrap().len()
        });

    let maven_files_1 = maven_files.clone();
    let m1 = Arc::clone(&maven_repo);
    let (jar_ops_1t, jar_el_1t) = bench_retrieval("Maven JARs", bench_duration, 1, move |i, _| {
        let f = &maven_files_1[(i as usize) % maven_files_1.len()];
        m1.get_file(f).unwrap().len()
    });

    let pip_files_1 = pip_files.clone();
    let p1 = Arc::clone(&pip_repo);
    let (pip_ops_1t, pip_el_1t) =
        bench_retrieval("Python wheels", bench_duration, 1, move |i, _| {
            let f = &pip_files_1[(i as usize) % pip_files_1.len()];
            p1.get_file(f).unwrap().len()
        });

    // Multi-threaded benchmarks
    eprintln!(
        "\n── Multi-threaded ({} cores) ─────────────────────────────────────",
        num_cpus
    );

    let rust_files_m = rust_files.clone();
    let rm = Arc::clone(&rust_repo);
    let (rust_ops_mt, rust_el_mt) =
        bench_retrieval("Rust crates", bench_duration, num_cpus, move |i, tid| {
            let idx = ((i as usize) * 7 + tid * 1337) % rust_files_m.len();
            rm.get_file(&rust_files_m[idx]).unwrap().len()
        });

    let maven_files_m = maven_files.clone();
    let mm = Arc::clone(&maven_repo);
    let (jar_ops_mt, jar_el_mt) =
        bench_retrieval("Maven JARs", bench_duration, num_cpus, move |i, tid| {
            let idx = ((i as usize) * 11 + tid * 773) % maven_files_m.len();
            mm.get_file(&maven_files_m[idx]).unwrap().len()
        });

    let pip_files_m = pip_files.clone();
    let pm = Arc::clone(&pip_repo);
    let (pip_ops_mt, pip_el_mt) =
        bench_retrieval("Python wheels", bench_duration, num_cpus, move |i, tid| {
            let idx = ((i as usize) * 13 + tid * 997) % pip_files_m.len();
            pm.get_file(&pip_files_m[idx]).unwrap().len()
        });

    // ── Summary ──────────────────────────────────────────────────────────
    let total_elapsed = total_start.elapsed();

    let rust_rate_1t = rust_ops_1t as f64 / rust_el_1t.as_secs_f64();
    let jar_rate_1t = jar_ops_1t as f64 / jar_el_1t.as_secs_f64();
    let pip_rate_1t = pip_ops_1t as f64 / pip_el_1t.as_secs_f64();
    let rust_rate_mt = rust_ops_mt as f64 / rust_el_mt.as_secs_f64();
    let jar_rate_mt = jar_ops_mt as f64 / jar_el_mt.as_secs_f64();
    let pip_rate_mt = pip_ops_mt as f64 / pip_el_mt.as_secs_f64();

    eprintln!("\n╔═══════════════════════════════════════════════════════════════╗");
    eprintln!("║       RESULTS — Real Artifact Benchmark                       ║");
    eprintln!(
        "║  Rust: {} files | Maven: {} files | Pip: {} files    ║",
        rust_files.len(),
        maven_files.len(),
        pip_files.len()
    );
    eprintln!("╠═══════════════════════════════════════════════════════════════╣");
    eprintln!(
        "║  Rust crates    (1 thread):  {:>10.0} ops/sec              ║",
        rust_rate_1t
    );
    eprintln!(
        "║  Maven JARs     (1 thread):  {:>10.0} ops/sec              ║",
        jar_rate_1t
    );
    eprintln!(
        "║  Python wheels   (1 thread):  {:>10.0} ops/sec              ║",
        pip_rate_1t
    );
    eprintln!("╠═══════════════════════════════════════════════════════════════╣");
    eprintln!(
        "║  Rust crates    ({:>2} threads): {:>10.0} ops/sec              ║",
        num_cpus, rust_rate_mt
    );
    eprintln!(
        "║  Maven JARs     ({:>2} threads): {:>10.0} ops/sec              ║",
        num_cpus, jar_rate_mt
    );
    eprintln!(
        "║  Python wheels   ({:>2} threads): {:>10.0} ops/sec              ║",
        num_cpus, pip_rate_mt
    );
    eprintln!("╠═══════════════════════════════════════════════════════════════╣");
    eprintln!(
        "║  Total time: {:.1}s ({:.1} min)                               ║",
        total_elapsed.as_secs_f64(),
        total_elapsed.as_secs_f64() / 60.0
    );
    eprintln!("╚═══════════════════════════════════════════════════════════════╝\n");
}

/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
#[allow(dead_code)]
fn tls_roots_blocking(b: reqwest::blocking::ClientBuilder) -> reqwest::blocking::ClientBuilder {
    let _ = rustls::crypto::ring::default_provider().install_default();
    b.tls_certs_only(
        webpki_root_certs::TLS_SERVER_ROOT_CERTS
            .iter()
            .filter_map(|c| reqwest::Certificate::from_der(c).ok()),
    )
}