//! Ingest and query benchmarks: the vault layer rather than the DSP layer.
//!
//! The analysis bench in `main.rs` measures per-file DSP cost. This measures
//! what happens to a vault as it fills up: import throughput, dedup, and the
//! query latency that backs the browser UI.
//!
//! Two properties of the store make scale worth measuring rather than
//! assuming. Blobs are sharded one level deep on the hash prefix
//! (`{root}/{ab}/{hash}.{ext}`, 256 leaves), which is what a run on this bench
//! bought: the flat layout it replaced lost about 90% of its import throughput
//! between an empty vault and a 40,000-entry one. Directory cost is filesystem
//! dependent either way, so it stays worth measuring. And the DB runs in WAL mode
//! with several worker connections, so insert cost moves with index depth.
//!
//! Reported per batch rather than as one average, because the number that
//! matters is whether throughput is flat or degrading as the vault grows.
//!
//! Take baselines with the corpus drive otherwise idle. Import is I/O bound, so
//! anything else touching the same spindle lands directly in the number: a run
//! taken while a dataset was downloading to the same drive read 40.7 files/s
//! against 204 for the same corpus on an idle one. The JSON output makes that
//! kind of contamination easy to mistake for a regression.

use std::path::{Path, PathBuf};
use std::time::Instant;

use audiofiles_core::analysis::{self, config::AnalysisConfig};
use audiofiles_core::db::Database;
use audiofiles_core::id_types::SampleHash;
use audiofiles_core::search::{self, SearchFilter, SearchScope};
use audiofiles_core::store::SampleStore;
use audiofiles_core::vfs;
use rayon::prelude::*;

use crate::report::{Report, peak_rss_mb};
use crate::storage;

/// How many samples to analyse after import.
///
/// Bounded because analysis is orders of magnitude dearer than import and the
/// point here is to populate `audio_analysis` so the filter queries below
/// measure something, not to benchmark the DSP (the default bench mode does
/// that per file). At NSynth scale an unbounded pass would dominate the run.
fn analyze_budget() -> usize {
    std::env::var("AF_BENCH_ANALYZE")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(2000)
}

/// How many files to run the per-stage probe over at each checkpoint, or 0 for
/// no probe.
///
/// Off by default, and that is deliberate rather than timid. The probe analyses
/// files in the middle of an import, so it competes for the drive and the page
/// cache with the thing being measured. Every ingest baseline saved before this
/// existed was recorded without it, and turning it on by default would make new
/// runs quietly incomparable with `benchmarks/ingest-2026-07-29-*.json`. Opt in
/// when the per-stage curve is the question; leave it off when the ingest curve
/// is.
fn stage_probe_size() -> usize {
    std::env::var("AF_BENCH_STAGES")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(0)
}

/// One per-stage probe: the DSP breakdown plus persistence, at a known vault size.
///
/// `persist_ms` is the reason this exists. The DSP stages are pure CPU over one
/// decoded buffer and have no way to know how large the vault is, so a flat line
/// from them is the expected result and the useful one: it says the per-file
/// numbers in section 1 keep their meaning at scale. Persistence is the stage
/// that can degrade, because it writes into a growing database, and section 1
/// never touches the DB at all.
struct StageProbe {
    samples: i64,
    decode_ms: f64,
    loudness_ms: f64,
    spectral_ms: f64,
    mfcc_ms: f64,
    vector_ms: f64,
    bpm_key_ms: f64,
    loop_ms: f64,
    fingerprint_ms: f64,
    total_ms: f64,
    persist_ms: f64,
    files: usize,
}

/// Run the per-stage probe over `files` against a vault currently holding
/// `samples` rows.
///
/// Medians, not means: one file that happens to be long drags a mean far enough
/// to invent a trend across checkpoints that is not there.
fn stage_probe(
    db: &Database,
    samples: i64,
    files: &[(String, PathBuf)],
    config: &AnalysisConfig,
) -> Option<StageProbe> {
    if files.is_empty() {
        return None;
    }
    let mut decode = Vec::new();
    let mut loud = Vec::new();
    let mut spec = Vec::new();
    let mut mfcc = Vec::new();
    let mut vector = Vec::new();
    let mut bpm_key = Vec::new();
    let mut loops = Vec::new();
    let mut fp = Vec::new();
    let mut total = Vec::new();

    for (_, path) in files {
        let Some((t, _, _)) = crate::time_stages(path) else {
            continue;
        };
        decode.push(t.decode_ms);
        loud.push(t.loudness_ms);
        spec.push(t.spectral_ms);
        mfcc.push(t.mfcc_ms);
        vector.push(t.vector_ms);
        bpm_key.push(t.bpm_key_ms);
        loops.push(t.loop_ms);
        fp.push(t.fingerprint_ms);
        total.push(t.total_ms);
    }
    if total.is_empty() {
        return None;
    }

    // Persistence is measured on a real analysis of the same files, written into
    // the live vault. Timing a fabricated row would measure the wrong statement.
    let results: Vec<_> = files
        .iter()
        .filter_map(|(hash, path)| analysis::analyze_sample(hash, path, config).ok())
        .collect();
    let persist_start = Instant::now();
    let persisted = analysis::save_analysis_batch(db, &results).is_ok();
    let persist_total_ms = persist_start.elapsed().as_secs_f64() * 1000.0;
    let persist_ms = if persisted && !results.is_empty() {
        persist_total_ms / results.len() as f64
    } else {
        f64::NAN
    };

    let files_probed = total.len();
    Some(StageProbe {
        samples,
        decode_ms: crate::percentile(&mut decode, 50.0),
        loudness_ms: crate::percentile(&mut loud, 50.0),
        spectral_ms: crate::percentile(&mut spec, 50.0),
        mfcc_ms: crate::percentile(&mut mfcc, 50.0),
        vector_ms: crate::percentile(&mut vector, 50.0),
        bpm_key_ms: crate::percentile(&mut bpm_key, 50.0),
        loop_ms: crate::percentile(&mut loops, 50.0),
        fingerprint_ms: crate::percentile(&mut fp, 50.0),
        total_ms: crate::percentile(&mut total, 50.0),
        persist_ms,
        files: files_probed,
    })
}

/// The per-stage curve as a JSON array, one object per checkpoint.
fn stage_series(probes: &[StageProbe]) -> serde_json::Value {
    serde_json::Value::Array(
        probes
            .iter()
            .map(|p| {
                let round = |v: f64| (v * 1000.0).round() / 1000.0;
                serde_json::json!({
                    "samples": p.samples,
                    "files_probed": p.files,
                    "decode_ms": round(p.decode_ms),
                    "loudness_ms": round(p.loudness_ms),
                    "spectral_ms": round(p.spectral_ms),
                    "mfcc_ms": round(p.mfcc_ms),
                    "vector_ms": round(p.vector_ms),
                    "bpm_key_ms": round(p.bpm_key_ms),
                    "loop_ms": round(p.loop_ms),
                    "fingerprint_ms": round(p.fingerprint_ms),
                    "total_ms": round(p.total_ms),
                    // NaN is not representable in JSON, so a failed persist
                    // reads as null rather than as a plausible number.
                    "persist_ms": if p.persist_ms.is_finite() {
                        serde_json::json!(round(p.persist_ms))
                    } else {
                        serde_json::Value::Null
                    },
                })
            })
            .collect(),
    )
}

/// Print the per-stage curve, and say whether anything actually drifted.
fn report_stage_series(probes: &[StageProbe]) {
    if probes.is_empty() {
        return;
    }
    println!();
    println!("━━━ PER-STAGE TIMING AT VAULT SCALE ━━━");
    println!();
    println!("  Medians in ms per file, probed mid-import at each vault size.");
    println!("  DSP stages should be flat: they cannot see the vault. persist");
    println!("  writes into the growing DB and is the one that can drift.");
    println!();
    println!(
        "  {:>8}  {:>7} {:>7} {:>7} {:>7} {:>7} {:>7} {:>8}",
        "samples", "decode", "spectral", "mfcc", "bpm/key", "loop", "total", "persist"
    );
    println!("  {}", "-".repeat(70));
    // Anything under the display resolution prints as `<0.01`, never as `0.00`.
    // The corpus is what makes this matter: `detect_bpm_key` and `is_loop` bail
    // out early on a file too short to hold a beat, so a one-shot corpus drives
    // both to microseconds. Those are real early returns, not free work, and
    // `0.00` in a column somebody is scanning for a bottleneck reads as "this
    // stage costs nothing" rather than "this stage declined to run here".
    let cell = |v: f64| {
        if v.abs() < 0.005 {
            "<0.01".to_string()
        } else {
            format!("{v:.2}")
        }
    };
    for p in probes {
        let persist = if p.persist_ms.is_finite() {
            format!("{:.3}", p.persist_ms)
        } else {
            "FAILED".to_string()
        };
        println!(
            "  {:>8}  {:>7} {:>7} {:>7} {:>7} {:>7} {:>7} {:>8}",
            p.samples,
            cell(p.decode_ms),
            cell(p.spectral_ms),
            cell(p.mfcc_ms),
            cell(p.bpm_key_ms),
            cell(p.loop_ms),
            cell(p.total_ms),
            persist
        );
    }

    println!();
    if probes.len() < 3 {
        // Deliberately no trend line here. The first probe carries the warmup
        // this whole section has to exclude, so with fewer than three
        // checkpoints there is nothing left to compare after dropping it.
        println!("  fewer than 3 checkpoints, so no trend is reported: the first");
        println!("  probe carries cold-cache warmup and there is nothing left to");
        println!("  compare against once it is dropped. Raise the file count or");
        println!("  lower AF_BENCH_BATCH.");
        return;
    }

    // THE FIRST PROBE IS DISCARDED, and this is the correction that makes the
    // section honest. An early version compared first to last and reported
    // "DSP total -84.7%" on a run where nothing had degraded at all: at 15 files
    // per probe, the first checkpoint is dominated by cold page cache and first
    // decode, so it is 3-5x the steady-state cost. The import curve above can
    // compare first to last because its batches are 500 files and amortise that
    // away; a probe this small cannot. Reporting a warmup artifact as a scaling
    // trend is precisely the failure mode the measurement-traps list exists for.
    let steady = &probes[1..];
    let spread = |mut v: Vec<f64>| {
        let hi = crate::percentile(&mut v.clone(), 100.0);
        let lo = crate::percentile(&mut v, 0.0);
        (lo, hi)
    };
    let (dsp_lo, dsp_hi) = spread(steady.iter().map(|p| p.total_ms).collect());
    println!(
        "  Steady state ({} checkpoints, first dropped as warmup):",
        steady.len()
    );
    println!(
        "    DSP total    {dsp_lo:.2} to {dsp_hi:.2} ms{}",
        if dsp_lo > 1e-9 && dsp_hi / dsp_lo > 2.0 {
            "   <- spread over 2x, treat as noise not trend"
        } else {
            ""
        }
    );
    let persists: Vec<f64> = steady
        .iter()
        .map(|p| p.persist_ms)
        .filter(|v| v.is_finite())
        .collect();
    if persists.is_empty() {
        println!("    persist      not measured");
    } else {
        let (p_lo, p_hi) = spread(persists);
        println!("    persist      {p_lo:.3} to {p_hi:.3} ms");
        if p_lo > 1e-9 && p_hi / p_lo > 2.0 {
            println!("    ^ persistence moved more than 2x across the fill: the stage");
            println!("      that writes into the growing DB is the one to look at.");
        }
    }
    println!();
    println!("  A range, not a delta: two endpoints cannot tell a trend from");
    println!("  noise. The machine has to be idle for any of this to mean");
    println!("  anything (wiki af-benchmarks, \"Measurement traps\").");
}

/// Deterministic reorder of the file list.
///
/// Without this the list arrives sorted by path, which groups files by class,
/// and classes have very different mean file sizes. Batch N would then differ
/// from batch 1 in content as well as in vault size, so a files/s trend across
/// batches would measure file-size composition rather than scaling behaviour.
/// FNV-1a over the path keeps it deterministic without pulling in `rand`.
fn shuffle_deterministic(files: &mut [PathBuf]) {
    fn fnv1a(s: &str) -> u64 {
        let mut h: u64 = 0xcbf2_9ce4_8422_2325;
        for b in s.as_bytes() {
            h ^= u64::from(*b);
            h = h.wrapping_mul(0x100_0000_01b3);
        }
        h
    }
    files.sort_by_key(|p| fnv1a(&p.to_string_lossy()));
}

/// One batch of imports.
struct BatchStat {
    /// Cumulative sample count in the vault after this batch.
    cumulative: usize,
    files: usize,
    bytes: u64,
    elapsed_s: f64,
}

impl BatchStat {
    fn files_per_sec(&self) -> f64 {
        if self.elapsed_s <= 0.0 {
            return 0.0;
        }
        self.files as f64 / self.elapsed_s
    }

    fn mb_per_sec(&self) -> f64 {
        if self.elapsed_s <= 0.0 {
            return 0.0;
        }
        (self.bytes as f64 / 1e6) / self.elapsed_s
    }
}

/// The per-batch curve as JSON, one object per batch in order.
///
/// The shape of the curve is what the import benchmark is actually asking
/// about (does throughput degrade as the vault grows), and it used to live only
/// in the stdout table. Serialising it means a killed run still leaves the
/// measurement behind.
fn batch_series(stats: &[BatchStat]) -> serde_json::Value {
    serde_json::Value::Array(
        stats
            .iter()
            .enumerate()
            .map(|(i, s)| {
                serde_json::json!({
                    "batch": i + 1,
                    "cumulative": s.cumulative,
                    "files": s.files,
                    "bytes": s.bytes,
                    "elapsed_s": (s.elapsed_s * 100.0).round() / 100.0,
                    "files_per_sec": (s.files_per_sec() * 10.0).round() / 10.0,
                    "mb_per_sec": (s.mb_per_sec() * 10.0).round() / 10.0,
                })
            })
            .collect(),
    )
}

/// Collect audio files under `dir`, recursively.
fn collect(dir: &Path, out: &mut Vec<PathBuf>) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            collect(&path, out);
        } else if path.extension().and_then(|e| e.to_str()).is_some_and(|e| {
            matches!(
                e.to_lowercase().as_str(),
                "wav" | "aif" | "aiff" | "flac" | "mp3" | "ogg"
            )
        }) {
            out.push(path);
        }
    }
}

fn count_samples(db: &Database) -> i64 {
    db.conn()
        .query_row("SELECT count(*) FROM samples", [], |r| r.get(0))
        .unwrap_or(-1)
}

/// Count blobs anywhere under the store root, descending into shard directories.
///
/// Must recurse. Blobs live at `{root}/{ab}/{hash}.{ext}`, so counting the root's
/// own entries returns the number of shard directories (at most 256) rather than
/// the number of blobs, which would make the dedup check below report "unchanged"
/// no matter what the store did.
fn count_blobs(root: &Path) -> usize {
    let Ok(entries) = std::fs::read_dir(root) else {
        return 0;
    };
    entries
        .filter_map(std::result::Result::ok)
        .map(|e| {
            let path = e.path();
            if path.is_dir() { count_blobs(&path) } else { 1 }
        })
        .sum()
}

/// Time a query, returning milliseconds. Runs it `reps` times and takes the
/// median, since a single cold query mostly measures page-cache state.
fn time_query(reps: usize, mut f: impl FnMut()) -> f64 {
    let mut times: Vec<f64> = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        f();
        times.push(t.elapsed().as_secs_f64() * 1000.0);
    }
    times.sort_by(f64::total_cmp);
    times[times.len() / 2]
}

/// Measure the query paths the browser list and filter panel depend on.
fn report_query_latency(db: &Database, report: &mut Report) {
    let n = count_samples(db);
    let nodes: i64 = db
        .conn()
        .query_row("SELECT count(*) FROM vfs_nodes", [], |r| r.get(0))
        .unwrap_or(-1);
    // Printed together on purpose: a large sample count with zero nodes means
    // the timings below are measuring empty result sets.
    println!("  {n} samples / {nodes} vfs nodes (median of 5):");
    let rows = search::search_global(
        db,
        &SearchFilter {
            scope: SearchScope::Global,
            ..Default::default()
        },
    )
    .map_or(0, |r| r.len());
    // search.rs caps every result set at SEARCH_RESULT_LIMIT (500), so list
    // latency is bounded by design no matter how large the vault gets. The
    // scan and sort underneath it are not bounded, which is what these numbers
    // actually track.
    println!("  unfiltered search returns {rows} rows (capped at 500 by SEARCH_RESULT_LIMIT)");

    let analyzed: i64 = db
        .conn()
        .query_row("SELECT count(*) FROM audio_analysis", [], |r| r.get(0))
        .unwrap_or(0);
    if analyzed == 0 {
        println!("  NOTE: audio_analysis is empty, so the class and bpm filters below");
        println!("        match nothing and their timings are not meaningful. Run");
        println!("        the analysis pipeline over this vault to benchmark them.");
    }

    let ms = time_query(5, || {
        let _ = count_samples(db);
    });
    println!("    count(*)                {ms:>8.2} ms");
    report.set("count_star_ms", (ms * 100.0).round() / 100.0);

    let mut filter = SearchFilter {
        scope: SearchScope::Global,
        ..Default::default()
    };
    let ms = time_query(5, || {
        let _ = search::search_global(db, &filter);
    });
    println!("    search_global (no filter) {ms:>6.2} ms   <- worst-case list load");
    report.set("search_unfiltered_ms", (ms * 100.0).round() / 100.0);

    filter.text_query = "kick".to_string();
    let ms = time_query(5, || {
        let _ = search::search_global(db, &filter);
    });
    println!("    search_global (text)      {ms:>6.2} ms   <- search box keystroke");
    report.set("search_text_ms", (ms * 100.0).round() / 100.0);

    filter.text_query.clear();
    filter.bpm_min = Some(120.0);
    filter.bpm_max = Some(130.0);
    let ms = time_query(5, || {
        let _ = search::search_global(db, &filter);
    });
    println!("    search_global (bpm range) {ms:>6.2} ms");
    report.set("search_bpm_ms", (ms * 100.0).round() / 100.0);
}

/// Run the ingest benchmark against `corpus`, building a scratch vault at
/// `vault`. Any existing scratch vault is removed first so runs are comparable.
pub(crate) fn run(corpus: &Path, vault: &Path, batch: usize, limit: Option<usize>) {
    println!("━━━ INGEST AT SCALE ━━━");
    println!();
    println!("  corpus: {}", corpus.display());
    println!("  vault:  {}", vault.display());

    let mut report = Report::new("ingest");

    let mut files = Vec::new();
    collect(corpus, &mut files);
    files.sort();
    shuffle_deterministic(&mut files);
    if let Some(lim) = limit {
        files.truncate(lim);
    }
    if files.is_empty() {
        eprintln!("no audio files under {}", corpus.display());
        return;
    }
    println!("  files:  {}", files.len());
    println!();

    if vault.exists()
        && let Err(e) = std::fs::remove_dir_all(vault)
    {
        eprintln!("could not clear scratch vault: {e}");
        return;
    }
    let samples_dir = vault.join("samples");
    if let Err(e) = std::fs::create_dir_all(&samples_dir) {
        eprintln!("could not create scratch vault: {e}");
        return;
    }

    // Described after the vault exists so its path canonicalizes, and reported
    // as two roles because corpus and vault are routinely on different drives:
    // reading from the external disk while writing to the internal one is a
    // third measurement, distinct from either drive on its own.
    let corpus_storage = storage::describe(corpus);
    let vault_storage = storage::describe(vault);
    report.set_storage("corpus", &corpus_storage);
    report.set_storage("vault", &vault_storage);
    let corpus_bytes: u64 = files
        .iter()
        .filter_map(|p| std::fs::metadata(p).ok())
        .map(|m| m.len())
        .sum();
    storage::print_conditions(
        &[("corpus", &corpus_storage), ("vault", &vault_storage)],
        Some(corpus_bytes),
    );

    let db = match Database::open(vault.join("audiofiles.db")) {
        Ok(db) => db,
        Err(e) => {
            // Worth surfacing loudly: on a filesystem that cannot support WAL
            // this is exactly where a vault fails, and the app surfaces it as a
            // generic init error.
            eprintln!("Database::open failed (WAL unsupported on this fs?): {e}");
            return;
        }
    };
    let store = match SampleStore::new(&samples_dir) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("SampleStore::new failed: {e}");
            return;
        }
    };

    // `store.import` writes the blob and the `samples` row but no VFS node.
    // The browser's import workflow creates those separately, and every query
    // the UI runs goes through `vfs_nodes`. Without them `search_global`
    // returns an empty set instantly and the query numbers below would be
    // measuring nothing.
    let vfs_id = match vfs::create_vfs(&db, "bench") {
        Ok(id) => id,
        Err(e) => {
            eprintln!("could not create bench vfs: {e}");
            return;
        }
    };

    println!("  batch      cumulative      files/s       MB/s     elapsed");
    println!("  ---------------------------------------------------------");

    let mut stats: Vec<BatchStat> = Vec::new();
    let mut cumulative = 0usize;
    let mut failures = 0usize;
    let mut link_failures = 0usize;
    // (hash, source path) for the analysis pass. The source file and the stored
    // blob are byte-identical by construction, so analysing either is the same
    // measurement and this avoids an extension lookup per sample.
    let mut imported: Vec<(String, PathBuf)> = Vec::with_capacity(files.len());

    let probe_size = stage_probe_size();
    let mut probes: Vec<StageProbe> = Vec::new();
    let probe_config = AnalysisConfig::default();
    if probe_size > 0 {
        println!("  (per-stage probe on, {probe_size} file(s) per batch: import numbers below");
        println!("   are NOT comparable with baselines recorded without it)");
    }

    for chunk in files.chunks(batch) {
        let mut bytes = 0u64;
        let start = Instant::now();
        for path in chunk {
            match store.import(path, &db) {
                Ok(hash) => {
                    // Name links by index: sample names must be unique among
                    // siblings, and the corpus has repeated basenames across
                    // packs.
                    let name = format!(
                        "{cumulative:06}_{}",
                        path.file_name().unwrap_or_default().to_string_lossy()
                    );
                    if vfs::create_sample_link(
                        &db,
                        vfs_id,
                        None,
                        &name,
                        &SampleHash::from_trusted(hash.clone()),
                    )
                    .is_err()
                    {
                        link_failures += 1;
                    }
                    imported.push((hash, path.clone()));
                    bytes += std::fs::metadata(path).map_or(0, |m| m.len());
                    cumulative += 1;
                }
                Err(_) => failures += 1,
            }
        }
        let stat = BatchStat {
            cumulative,
            files: chunk.len(),
            bytes,
            elapsed_s: start.elapsed().as_secs_f64(),
        };
        println!(
            "  {:>5}      {:>10}   {:>10.1} {:>10.1}   {:>7.2}s",
            stats.len() + 1,
            stat.cumulative,
            stat.files_per_sec(),
            stat.mb_per_sec(),
            stat.elapsed_s,
        );
        stats.push(stat);

        // Checkpoint the curve after every batch. A long import is the run most
        // worth recording and the one most likely to be killed part way, and
        // the summary metrics below only exist once the loop finishes.
        report.set("import_batches", batch_series(&stats));
        report.set("import_files", stats.iter().map(|s| s.files).sum::<usize>());
        report.set("import_bytes", stats.iter().map(|s| s.bytes).sum::<u64>());
        report.set("import_complete", false);

        // Probe AFTER the batch timer has stopped, so the probe's own decode and
        // DB work never lands inside a files/s figure. It still perturbs the
        // drive and the page cache for the batches that follow, which is why the
        // whole thing is opt-in.
        if probe_size > 0 {
            let sample_rows = count_samples(&db);
            let recent: Vec<(String, PathBuf)> =
                imported.iter().rev().take(probe_size).cloned().collect();
            if let Some(p) = stage_probe(&db, sample_rows, &recent, &probe_config) {
                probes.push(p);
                report.set("stage_series", stage_series(&probes));
            }
        }

        report.checkpoint();
    }

    println!();
    if failures > 0 {
        println!("  {failures} file(s) failed to import");
    }
    if link_failures > 0 {
        println!("  {link_failures} vfs link(s) failed -- query numbers below undercount");
    }

    // Degradation is the actual question. Comparing first batch to last is the
    // cheapest signal that the flat blob directory or an index has started to
    // bite; a flat profile means it has not.
    if stats.len() >= 2 {
        let first = stats[0].files_per_sec();
        let last = stats[stats.len() - 1].files_per_sec();
        let delta = if first > 0.0 {
            (last - first) / first * 100.0
        } else {
            0.0
        };
        println!("  first batch: {first:.1} files/s");
        println!("  last  batch: {last:.1} files/s   ({delta:+.1}%)");
        if delta < -25.0 {
            println!("  ^ throughput degraded as the vault grew");
        }
    }

    let total_files: usize = stats.iter().map(|s| s.files).sum();
    let total_bytes: u64 = stats.iter().map(|s| s.bytes).sum();
    let total_s: f64 = stats.iter().map(|s| s.elapsed_s).sum();
    println!();
    println!(
        "  total: {total_files} files, {:.2} GB in {total_s:.1}s ({:.1} files/s, {:.1} MB/s)",
        total_bytes as f64 / 1e9,
        total_files as f64 / total_s,
        (total_bytes as f64 / 1e6) / total_s,
    );
    // `import_complete` separates a finished import from a checkpoint of one
    // that was killed, so a comparison tool does not read a partial curve as a
    // regression.
    report.set("import_complete", true);
    report.set("import_files", total_files);
    report.set("import_bytes", total_bytes);
    report.set(
        "import_files_per_sec",
        ((total_files as f64 / total_s) * 10.0).round() / 10.0,
    );
    report.set(
        "import_mb_per_sec",
        (((total_bytes as f64 / 1e6) / total_s) * 10.0).round() / 10.0,
    );
    if let (Some(first), Some(last)) = (stats.first(), stats.last()) {
        // The scaling signal, not a throughput figure: negative means the vault
        // got slower as it filled.
        let delta = (last.files_per_sec() - first.files_per_sec()) / first.files_per_sec() * 100.0;
        report.set("import_throughput_drift_pct", (delta * 10.0).round() / 10.0);
    }

    report_stage_series(&probes);

    // Dedup: re-importing the same files must hit the content-addressed store
    // and skip the copy. If this is not dramatically faster, dedup is not
    // working and every duplicate costs a full hash-and-copy.
    println!();
    println!("━━━ DEDUP (re-import of identical content) ━━━");
    println!();
    let blobs_before = count_blobs(&samples_dir);
    let rows_before = count_samples(&db);

    let redo: Vec<&PathBuf> = files.iter().take(batch.min(files.len())).collect();
    let start = Instant::now();
    for path in &redo {
        let _ = store.import(path, &db);
    }
    let redo_s = start.elapsed().as_secs_f64();

    let blobs_after = count_blobs(&samples_dir);
    let rows_after = count_samples(&db);

    println!(
        "  re-imported {} files in {redo_s:.2}s ({:.1} files/s)",
        redo.len(),
        redo.len() as f64 / redo_s.max(1e-9)
    );
    println!("  blobs on disk: {blobs_before} -> {blobs_after}  (want: unchanged)");
    println!("  sample rows:   {rows_before} -> {rows_after}");

    // Distinguishing these two matters: equal blob counts with equal row counts
    // means full dedup; equal blobs with more rows means the blob was reused but
    // a duplicate row was still written.
    if blobs_after == blobs_before {
        println!("  blob dedup: OK (no new blobs written)");
    } else {
        println!(
            "  blob dedup: {} new blob(s) written",
            blobs_after - blobs_before
        );
    }

    // Analysis pass. Two purposes: it is a throughput number in its own right at
    // vault scale, and without it `audio_analysis` stays empty and the class and
    // bpm filter timings below measure nothing.
    println!();
    println!("━━━ ANALYSIS AT SCALE ━━━");
    println!();
    let budget = analyze_budget();
    let to_analyze: Vec<(String, PathBuf)> = imported.into_iter().take(budget).collect();
    if to_analyze.is_empty() {
        println!("  skipped (AF_BENCH_ANALYZE=0)");
    } else {
        println!(
            "  analysing {} samples (AF_BENCH_ANALYZE={budget})",
            to_analyze.len()
        );
        let config = AnalysisConfig::default();
        let start = Instant::now();
        let results: Vec<_> = to_analyze
            .par_iter()
            .filter_map(|(hash, path)| analysis::analyze_sample(hash, path, &config).ok())
            .collect();
        let analyze_s = start.elapsed().as_secs_f64();

        let save_start = Instant::now();
        let saved = analysis::save_analysis_batch(&db, &results).is_ok();
        let save_s = save_start.elapsed().as_secs_f64();

        let rate = results.len() as f64 / analyze_s.max(1e-9);
        println!(
            "  analysed {} in {analyze_s:.1}s ({rate:.1} files/s)",
            results.len()
        );
        println!(
            "  persisted {} rows in {save_s:.2}s{}",
            results.len(),
            if saved { "" } else { "  (SAVE FAILED)" }
        );
        if results.len() < to_analyze.len() {
            println!(
                "  {} file(s) failed analysis",
                to_analyze.len() - results.len()
            );
        }
        report.set("analysis_files", results.len());
        report.set("analysis_files_per_sec", (rate * 10.0).round() / 10.0);
        report.set("analysis_persist_s", (save_s * 100.0).round() / 100.0);
    }

    println!();
    println!("━━━ QUERY LATENCY (backs the browser UI) ━━━");
    println!();
    report_query_latency(&db, &mut report);

    // Left in place deliberately: the vault is the artifact to point the app at
    // for eyeballing UI responsiveness at this size.
    println!();
    if let Some(rss) = peak_rss_mb() {
        // High-water across the whole run, so it covers the import and analysis
        // peaks rather than whatever happens to be resident at the end.
        println!("  peak RSS: {rss:.1} MB");
    }
    println!(
        "  scratch vault left at {} for UI inspection",
        vault.display()
    );
    report.write();
}

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

    fn probe(samples: i64, total_ms: f64, persist_ms: f64) -> StageProbe {
        StageProbe {
            samples,
            decode_ms: 1.0,
            loudness_ms: 1.0,
            spectral_ms: 1.0,
            mfcc_ms: 1.0,
            vector_ms: 1.0,
            bpm_key_ms: 1.0,
            loop_ms: 1.0,
            fingerprint_ms: 1.0,
            total_ms,
            persist_ms,
            files: 25,
        }
    }

    #[test]
    fn stage_probe_is_off_unless_asked_for() {
        // Guards the default. Turning the probe on silently would make every
        // new ingest run incomparable with the saved baselines.
        // SAFETY: single-threaded test, no other thread reads the environment.
        unsafe { std::env::remove_var("AF_BENCH_STAGES") };
        assert_eq!(stage_probe_size(), 0);
    }

    #[test]
    fn stage_series_carries_one_object_per_checkpoint() {
        let series = stage_series(&[probe(500, 12.0, 0.4), probe(1000, 12.5, 0.6)]);
        let rows = series.as_array().unwrap();
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0]["samples"], 500);
        assert_eq!(rows[1]["samples"], 1000);
        assert_eq!(rows[0]["files_probed"], 25);
        assert!((rows[1]["persist_ms"].as_f64().unwrap() - 0.6).abs() < 1e-9);
    }

    #[test]
    fn a_failed_persist_serialises_as_null_not_as_a_number() {
        // NaN has no JSON spelling. Emitting it as 0.0 would read downstream as
        // "persistence was free", which is the opposite of what happened.
        let series = stage_series(&[probe(500, 12.0, f64::NAN)]);
        assert!(series[0]["persist_ms"].is_null());
    }

    #[test]
    fn a_trend_needs_more_than_two_checkpoints() {
        // Regression guard on a real mistake: an early version compared the
        // first probe to the last and reported "DSP total -84.7%" on a run
        // where nothing degraded. The first probe is warmup, so it is dropped,
        // and with fewer than three checkpoints nothing survives to compare.
        // These call the printer for absence of panic; the contract they pin is
        // that `steady` is `probes[1..]` and is only read when len >= 3.
        report_stage_series(&[]);
        report_stage_series(&[probe(100, 40.0, 0.05)]);
        report_stage_series(&[probe(100, 40.0, 0.05), probe(200, 12.0, 0.05)]);
        report_stage_series(&[
            probe(100, 40.0, 0.05),
            probe(200, 12.0, 0.05),
            probe(300, 12.5, 0.06),
        ]);
    }

    #[test]
    fn stage_probe_returns_nothing_for_an_empty_file_set() {
        let db = Database::open_in_memory().unwrap();
        let config = AnalysisConfig::default();
        assert!(stage_probe(&db, 0, &[], &config).is_none());
    }
}