//! Flat-to-sharded blob migration, exercised headlessly at scale.
//!
//! The sweep in `audiofiles_core::store::layout` auto-starts at vault open on any
//! vault created before 2026-07-29, so the first real exercise of it lands on a
//! library someone cares about unless it is done deliberately first. Unit tests
//! cover the branches on two or three blobs each; they do not answer whether the
//! thing survives fifty thousand renames, whether an interrupted pass really
//! resumes, or whether the mirror's symlinks are still pointing at anything
//! afterwards.
//!
//! This mode fabricates flat vaults and falsifies those claims one at a time. It
//! is a checker first and a benchmark second: every scenario prints PASS or FAIL
//! and the process exits non-zero if any of them failed, so a run that scrolls
//! past unread still fails loudly in a pipeline.
//!
//! What it cannot cover, and what still needs a human at the app: the progress
//! strip's rendering above the footer, and browsing/preview/search staying usable
//! mid-sweep. The store-layer half of that second claim is covered here (every
//! blob resolves through a partial migration), but the GUI half is an eyeball
//! pass.
//!
//! Vaults are fabricated rather than imported: the sweep reads the filesystem, not
//! the DB, so blob provenance is irrelevant to it and a real import pass would
//! spend its time in hashing and per-file fsync instead. Blobs are chmod'd
//! read-only exactly as `import` leaves them, because two branches of the sweep
//! unlink a flat blob and that is the permission shape they will meet in the
//! field.
//!
//! Usage:
//!   `cargo run --release -p audiofiles-bench -- layout`
//!
//! Env: `AF_BENCH_VAULT` (scratch vault root, default `<tmp>/af-bench-layout`),
//! `AF_BENCH_LAYOUT_N` (blobs in the timed sweep, default 50,000),
//! `AF_BENCH_JSON` (machine-readable output path).
//!
//! <!-- wiki: af-benchmarks -->

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Instant;

use audiofiles_core::config_key::ConfigKey;
use audiofiles_core::db::Database;
use audiofiles_core::id_types::SampleHash;
use audiofiles_core::store::layout::{
    BlobLayout, LayoutMigration, count_flat_blobs, migrate_to_sharded, recorded_layout,
};
use audiofiles_core::store::{
    SampleStore, existing_blob_path, hash_file, legacy_flat_blob_path, store_blob_path,
};
use audiofiles_core::vfs;
use audiofiles_core::vfs_mirror::{MirrorConfig, sync_mirror};

use crate::report::Report;
use crate::storage;

/// Blobs in each correctness scenario.
///
/// Small on purpose: these check branches, not throughput, and every scenario
/// rebuilds its vault from scratch. Large enough that a cancellation lands in the
/// middle of a pass rather than racing the first blob.
const SCENARIO_BLOBS: usize = 2_000;

/// Blobs in the timed sweep unless `AF_BENCH_LAYOUT_N` says otherwise.
///
/// Matches the af-bench-nsynth corpus on the T9 (53,286 blobs) closely enough to
/// stand in for it, so the throughput number here is comparable to what that vault
/// will do.
pub(crate) const DEFAULT_TIMED_BLOBS: usize = 50_000;

/// A fabricated blob: its content hash and extension.
type Blob = (String, String);

/// Deterministic blob content for index `i`.
///
/// Length varies with the index so that a planted size mismatch is a real
/// difference in bytes rather than a difference the sweep could only see by
/// hashing, and so the fabricated vault is not one file repeated.
fn payload(i: usize) -> Vec<u8> {
    let mut bytes = format!("af-bench-layout blob {i}\n").into_bytes();
    bytes.resize(64 + (i % 97) * 8, b'\0');
    bytes
}

/// Make a blob read-only, as `SampleStore::import` leaves every canonical blob.
///
/// Best-effort for the same reason the store's own version is: a filesystem that
/// rejects the chmod must not fail the run. Unlinking still works either way,
/// since that needs write on the directory rather than on the file.
fn set_readonly(path: &Path) {
    if let Ok(meta) = std::fs::metadata(path) {
        let mut perms = meta.permissions();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            perms.set_mode(0o444);
        }
        #[cfg(not(unix))]
        perms.set_readonly(true);
        let _ = std::fs::set_permissions(path, perms);
    }
}

/// A fabricated vault in the legacy flat layout.
struct FlatVault {
    db: Database,
    store: SampleStore,
    blobs: Vec<Blob>,
    root: PathBuf,
}

/// Build a vault of `n` flat blobs at `vault`, clearing anything already there.
///
/// Rows go into `samples` so the mirror and the resolver have something to read;
/// VFS links are created only when asked, because they cost a statement per blob
/// and only the mirror scenario needs them.
fn fabricate(vault: &Path, n: usize, with_vfs: bool) -> Option<FlatVault> {
    if vault.exists()
        && let Err(e) = std::fs::remove_dir_all(vault)
    {
        eprintln!("could not clear scratch vault: {e}");
        return None;
    }
    let root = vault.join("samples");
    let stage = vault.join("stage");
    for dir in [&root, &stage] {
        if let Err(e) = std::fs::create_dir_all(dir) {
            eprintln!("could not create {}: {e}", dir.display());
            return None;
        }
    }

    let db = match Database::open(vault.join("audiofiles.db")) {
        Ok(db) => db,
        Err(e) => {
            eprintln!("Database::open failed (WAL unsupported on this fs?): {e}");
            return None;
        }
    };
    let store = match SampleStore::new(&root) {
        Ok(s) => s,
        Err(e) => {
            eprintln!("SampleStore::new failed: {e}");
            return None;
        }
    };

    let vfs_id = if with_vfs {
        match vfs::create_vfs(&db, "bench") {
            Ok(id) => Some(id),
            Err(e) => {
                eprintln!("could not create bench vfs: {e}");
                return None;
            }
        }
    } else {
        None
    };

    let mut blobs = Vec::with_capacity(n);
    // `.wav` because `hash_file` refuses anything `is_audio_file` does not
    // recognise. It hashes bytes rather than decoding them, so the extension is
    // the only part of "is this audio" that has to be true here.
    let staged = stage.join("blob.wav");
    // One transaction for the whole fabrication. Per-row commits would make
    // building a 50k vault slower than the sweep it exists to measure.
    if db.conn().execute_batch("BEGIN").is_err() {
        eprintln!("could not open the fabrication transaction");
        return None;
    }
    for i in 0..n {
        let content = payload(i);
        if std::fs::write(&staged, &content).is_err() {
            eprintln!("could not stage blob {i}");
            return None;
        }
        let (hash, size) = match hash_file(&staged) {
            Ok(pair) => pair,
            Err(e) => {
                eprintln!("could not hash blob {i}: {e}");
                return None;
            }
        };
        let dest = legacy_flat_blob_path(&root, &hash, "wav");
        if std::fs::rename(&staged, &dest).is_err() {
            eprintln!("could not place flat blob {i}");
            return None;
        }
        set_readonly(&dest);
        let now = i as i64;
        if db
            .conn()
            .execute(
                "INSERT OR IGNORE INTO samples
                 (hash, original_name, file_extension, file_size, import_date, last_modified)
                 VALUES (?1, ?2, 'wav', ?3, ?4, ?4)",
                rusqlite::params![hash, format!("blob-{i:06}.wav"), size, now],
            )
            .is_err()
        {
            eprintln!("could not insert sample row {i}");
            return None;
        }
        if let Some(vfs_id) = vfs_id
            && vfs::create_sample_link(
                &db,
                vfs_id,
                None,
                &format!("blob-{i:06}.wav"),
                &SampleHash::from_trusted(hash.clone()),
            )
            .is_err()
        {
            eprintln!("could not link blob {i} into the vfs");
            return None;
        }
        blobs.push((hash, "wav".to_string()));
    }
    if db.conn().execute_batch("COMMIT").is_err() {
        eprintln!("could not commit the fabrication transaction");
        return None;
    }
    let _ = std::fs::remove_dir_all(&stage);

    Some(FlatVault {
        db,
        store,
        blobs,
        root,
    })
}

/// Run a sweep with no cancellation and no progress interest.
fn sweep(v: &FlatVault) -> Option<LayoutMigration> {
    let cancel = AtomicBool::new(false);
    migrate_to_sharded(&v.store, &v.db, &cancel, &mut |_, _| {})
        .inspect_err(|e| eprintln!("sweep failed: {e}"))
        .ok()
}

/// Every blob resolves to a file that exists, in whichever layout holds it.
///
/// This is the store-layer half of "the library stays usable mid-migration": the
/// resolver is what browsing, preview and export all go through, so a hash that
/// stops resolving is a sample that has disappeared from the app.
fn all_resolve(v: &FlatVault) -> bool {
    v.blobs
        .iter()
        .all(|(hash, ext)| existing_blob_path(&v.root, hash, ext).is_some_and(|p| p.is_file()))
}

/// Every blob sits at its canonical sharded path and nowhere else.
fn all_sharded(v: &FlatVault) -> bool {
    v.blobs.iter().all(|(hash, ext)| {
        store_blob_path(&v.root, hash, ext).is_file()
            && !legacy_flat_blob_path(&v.root, hash, ext).exists()
    })
}

/// Names directly in the store root that are not shard directories.
fn root_strays(root: &Path) -> Vec<String> {
    let Ok(entries) = std::fs::read_dir(root) else {
        return Vec::new();
    };
    entries
        .flatten()
        .filter(|e| !e.file_type().is_ok_and(|t| t.is_dir()))
        .map(|e| e.file_name().to_string_lossy().into_owned())
        .collect()
}

/// Collected scenario results, printed as one table and folded into the exit code.
struct Checks {
    rows: Vec<(String, bool, String)>,
}

impl Checks {
    fn new() -> Self {
        Self { rows: Vec::new() }
    }

    fn add(&mut self, name: &str, ok: bool, detail: impl Into<String>) {
        let detail = detail.into();
        println!(
            "    {:<44} {}   {detail}",
            name,
            if ok { "PASS" } else { "FAIL" }
        );
        self.rows.push((name.to_string(), ok, detail));
    }

    fn failed(&self) -> usize {
        self.rows.iter().filter(|(_, ok, _)| !ok).count()
    }
}

/// A pass over a complete vault relocates everything and records the layout.
fn scenario_full(vault: &Path, checks: &mut Checks) {
    let Some(v) = fabricate(vault, SCENARIO_BLOBS, false) else {
        checks.add("full sweep", false, "could not fabricate the vault");
        return;
    };
    checks.add(
        "pending before the sweep",
        !matches!(recorded_layout(&v.db), Ok(BlobLayout::Sharded))
            && count_flat_blobs(&v.root).unwrap_or(0) == SCENARIO_BLOBS,
        format!("{SCENARIO_BLOBS} flat blobs, layout unrecorded"),
    );

    // Progress is what the strip draws, so its shape is worth checking even
    // though the strip itself is not on screen here: a callback that skips the
    // final call or walks backwards would show a bar that never fills.
    let mut seen: Vec<(usize, usize)> = Vec::new();
    let cancel = AtomicBool::new(false);
    let Ok(report) = migrate_to_sharded(&v.store, &v.db, &cancel, &mut |done, total| {
        seen.push((done, total));
    }) else {
        checks.add("full sweep", false, "sweep returned an error");
        return;
    };

    checks.add(
        "full sweep relocates every blob",
        report.moved == SCENARIO_BLOBS
            && report.deduped == 0
            && report.errors == 0
            && report.completed
            && !report.cancelled,
        format!(
            "moved {} deduped {} errors {}",
            report.moved, report.deduped, report.errors
        ),
    );
    checks.add(
        "progress is monotonic and ends full",
        seen.windows(2).all(|w| w[1].0 == w[0].0 + 1)
            && seen.last() == Some(&(SCENARIO_BLOBS, SCENARIO_BLOBS)),
        format!("{} callbacks", seen.len()),
    );
    checks.add(
        "blobs land under their hash prefix",
        all_sharded(&v),
        "{root}/{ab}/{hash}.wav".to_string(),
    );
    let strays = root_strays(&v.root);
    checks.add(
        "root holds only shard directories",
        strays.is_empty(),
        if strays.is_empty() {
            "clean".to_string()
        } else {
            format!("{} left: {}", strays.len(), strays.join(", "))
        },
    );
    checks.add(
        "layout recorded as sharded",
        matches!(recorded_layout(&v.db), Ok(BlobLayout::Sharded)),
        "blob_layout=sharded".to_string(),
    );

    // The second open the task asks about: a completed vault must not re-sweep.
    let Some(again) = sweep(&v) else {
        checks.add(
            "re-open does not re-sweep",
            false,
            "sweep returned an error",
        );
        return;
    };
    checks.add(
        "re-open does not re-sweep",
        again.moved == 0 && again.deduped == 0 && again.errors == 0 && again.completed,
        format!("moved {} on the second pass", again.moved),
    );
}

/// Cancelling mid-pass leaves a resolvable vault that the next pass finishes.
fn scenario_resume(vault: &Path, checks: &mut Checks) {
    let Some(v) = fabricate(vault, SCENARIO_BLOBS, false) else {
        checks.add("resume", false, "could not fabricate the vault");
        return;
    };

    // Cancel from inside the progress callback, which is where the GUI's cancel
    // button effectively lands: the flag is set while the sweep is running.
    let cancel = AtomicBool::new(false);
    let stop_at = SCENARIO_BLOBS / 3;
    let Ok(first) = migrate_to_sharded(&v.store, &v.db, &cancel, &mut |done, _| {
        if done >= stop_at {
            cancel.store(true, Ordering::Release);
        }
    }) else {
        checks.add("resume", false, "first pass returned an error");
        return;
    };

    checks.add(
        "cancel stops the pass early",
        first.cancelled
            && !first.completed
            && first.moved >= stop_at
            && first.moved < SCENARIO_BLOBS,
        format!("moved {} of {SCENARIO_BLOBS} then stopped", first.moved),
    );
    checks.add(
        "cancelled vault stays recorded flat",
        matches!(recorded_layout(&v.db), Ok(BlobLayout::Flat)),
        "so the next open resumes".to_string(),
    );
    checks.add(
        "every blob resolves mid-migration",
        all_resolve(&v),
        "reads span both layouts".to_string(),
    );

    let remaining = count_flat_blobs(&v.root).unwrap_or(0);
    let Some(second) = sweep(&v) else {
        checks.add("resume", false, "second pass returned an error");
        return;
    };
    checks.add(
        "the next pass resumes rather than restarts",
        second.moved == remaining && first.moved + second.moved == SCENARIO_BLOBS,
        format!("{} + {} = {SCENARIO_BLOBS}", first.moved, second.moved),
    );
    checks.add(
        "resumed vault ends fully sharded",
        second.completed && all_sharded(&v) && root_strays(&v.root).is_empty(),
        "blob_layout=sharded".to_string(),
    );
}

/// A size mismatch is left for a human; a matching duplicate is discarded.
fn scenario_mismatch(vault: &Path, checks: &mut Checks) {
    let Some(v) = fabricate(vault, 32, false) else {
        checks.add("size mismatch", false, "could not fabricate the vault");
        return;
    };

    // Plant a shard-side copy of blob 0 with different bytes (a truncated blob
    // from a pre-atomic-rename crash) and an identical-size copy of blob 1 (the
    // same content already migrated by a repair write).
    let (mismatch, ext) = v.blobs[0].clone();
    let (dupe, _) = v.blobs[1].clone();
    for (hash, content) in [(&mismatch, b"short".to_vec()), (&dupe, payload(1))] {
        let dest = store_blob_path(&v.root, hash, &ext);
        let Some(shard) = dest.parent() else { continue };
        if std::fs::create_dir_all(shard).is_err() || std::fs::write(&dest, &content).is_err() {
            checks.add("size mismatch", false, "could not plant the shard copy");
            return;
        }
    }
    // Leftovers the sweep must ignore rather than rename: an import that died
    // between create and rename, and a file a user dropped in by hand.
    let tmp = v.root.join(format!("{mismatch}.wav.12345.tmp"));
    let note = v.root.join("notes.txt");
    let _ = std::fs::write(&tmp, b"partial");
    let _ = std::fs::write(&note, b"mine");

    let Some(report) = sweep(&v) else {
        checks.add("size mismatch", false, "sweep returned an error");
        return;
    };

    checks.add(
        "size mismatch is counted, not resolved",
        report.errors == 1 && !report.completed,
        format!("errors {} completed {}", report.errors, report.completed),
    );
    checks.add(
        "both copies of a mismatch survive",
        legacy_flat_blob_path(&v.root, &mismatch, &ext).is_file()
            && store_blob_path(&v.root, &mismatch, &ext).is_file(),
        "left for inspection".to_string(),
    );
    checks.add(
        "redundant flat copy is discarded",
        report.deduped == 1 && !legacy_flat_blob_path(&v.root, &dupe, &ext).exists(),
        format!("deduped {}", report.deduped),
    );
    checks.add(
        "an errored pass stays recorded flat",
        matches!(recorded_layout(&v.db), Ok(BlobLayout::Flat)),
        "so the mismatch is swept again after repair".to_string(),
    );
    checks.add(
        "temp leftovers and stray files are untouched",
        tmp.is_file() && note.is_file(),
        "not blobs, never renamed".to_string(),
    );
}

/// The mirror's symlinks are stale after a sweep and whole again after a rebuild.
fn scenario_mirror(vault: &Path, checks: &mut Checks) {
    let Some(v) = fabricate(vault, 256, true) else {
        checks.add("mirror", false, "could not fabricate the vault");
        return;
    };
    let mirror_root = vault.join("mirror");
    let config = MirrorConfig {
        mirror_root: mirror_root.clone(),
        store_root: v.root.clone(),
    };
    if v.db.set_config(ConfigKey::MirrorEnabled, "true").is_err()
        || v.db
            .set_config(ConfigKey::MirrorPath, &mirror_root.to_string_lossy())
            .is_err()
    {
        checks.add("mirror", false, "could not record the mirror config");
        return;
    }

    // Built while the vault is still flat, so every link points at a flat path:
    // the state a real pre-2026-07-29 vault with a mirror is in at open.
    let Ok(before) = sync_mirror(&v.db, &config) else {
        checks.add("mirror", false, "the first sync failed");
        return;
    };
    checks.add(
        "mirror links a flat vault",
        before.links_created == 256 && dangling(&mirror_root) == 0,
        format!("{} links, none dangling", before.links_created),
    );

    let Some(report) = sweep(&v) else {
        checks.add("mirror", false, "sweep returned an error");
        return;
    };
    let stale = dangling(&mirror_root);
    checks.add(
        "sweep leaves the old links dangling",
        report.completed && stale == 256,
        format!("{stale} of 256 dangle, which is why a rebuild fires"),
    );

    let Ok(after) = sync_mirror(&v.db, &config) else {
        checks.add("mirror", false, "the rebuild failed");
        return;
    };
    checks.add(
        "rebuild repoints every link",
        dangling(&mirror_root) == 0,
        format!("{} relinked", after.links_created),
    );
}

/// Symlinks under `root` whose target does not exist.
fn dangling(root: &Path) -> usize {
    let Ok(entries) = std::fs::read_dir(root) else {
        return 0;
    };
    entries
        .flatten()
        .map(|e| {
            let path = e.path();
            if e.file_type().is_ok_and(|t| t.is_dir()) {
                dangling(&path)
            } else {
                // `symlink_metadata` sees the link, `metadata` follows it: present
                // as a link but absent as a file is exactly a dangling symlink.
                usize::from(path.symlink_metadata().is_ok() && !path.exists())
            }
        })
        .sum()
}

/// Time one uninterrupted sweep over `n` blobs.
fn timed_sweep(vault: &Path, n: usize, report: &mut Report, checks: &mut Checks) {
    println!();
    println!("━━━ TIMED SWEEP ({n} blobs) ━━━");
    println!();

    let build = Instant::now();
    let Some(v) = fabricate(vault, n, false) else {
        checks.add("timed sweep", false, "could not fabricate the vault");
        return;
    };
    println!("  fabricated in {:.1}s", build.elapsed().as_secs_f64());

    let mut callbacks = 0usize;
    let cancel = AtomicBool::new(false);
    let start = Instant::now();
    let Ok(pass) = migrate_to_sharded(&v.store, &v.db, &cancel, &mut |_, _| callbacks += 1) else {
        checks.add("timed sweep", false, "sweep returned an error");
        return;
    };
    let elapsed = start.elapsed().as_secs_f64();
    let per_sec = if elapsed > 0.0 {
        pass.moved as f64 / elapsed
    } else {
        0.0
    };

    println!();
    println!("  moved {} blobs in {elapsed:.2}s", pass.moved);
    println!("  {per_sec:.0} blobs/s   {:.3} ms/blob", 1000.0 / per_sec);
    println!();
    report.set("layout_blobs", n);
    report.set("layout_sweep_s", (elapsed * 100.0).round() / 100.0);
    report.set("layout_blobs_per_sec", per_sec.round());

    checks.add(
        "timed sweep completes cleanly",
        pass.moved == n && pass.errors == 0 && pass.completed && callbacks == n,
        format!("{n} moved, {callbacks} progress callbacks"),
    );
    checks.add(
        "timed sweep leaves a clean root",
        root_strays(&v.root).is_empty() && count_flat_blobs(&v.root).unwrap_or(1) == 0,
        "only shard directories".to_string(),
    );
}

/// Run the layout exercise against a scratch vault at `vault`.
pub(crate) fn run(vault: &Path, timed_blobs: usize) {
    println!("━━━ BLOB LAYOUT MIGRATION ━━━");
    println!();
    println!("  vault: {}", vault.display());
    println!();

    let mut report = Report::new("layout");
    let vault_storage = storage::describe(
        vault
            .parent()
            .filter(|p| p.exists())
            .unwrap_or_else(|| Path::new(".")),
    );
    report.set_storage("vault", &vault_storage);
    storage::print_conditions(&[("vault", &vault_storage)], None);

    let mut checks = Checks::new();
    println!("━━━ SCENARIOS ({SCENARIO_BLOBS} blobs unless stated) ━━━");
    println!();
    scenario_full(vault, &mut checks);
    println!();
    scenario_resume(vault, &mut checks);
    println!();
    scenario_mismatch(vault, &mut checks);
    println!();
    scenario_mirror(vault, &mut checks);

    timed_sweep(vault, timed_blobs, &mut report, &mut checks);

    let failed = checks.failed();
    report.set("layout_checks", checks.rows.len());
    report.set("layout_checks_failed", failed);
    report.write();

    println!();
    if failed == 0 {
        println!("  {} checks, all passed", checks.rows.len());
    } else {
        println!("  {failed} of {} checks FAILED", checks.rows.len());
    }
    // The scratch vault is left behind on failure: the whole value of a failed
    // check is the state that produced it.
    if failed == 0 {
        let _ = std::fs::remove_dir_all(vault);
    } else {
        println!("  vault left at {} for inspection", vault.display());
        std::process::exit(1);
    }
}

/// The layout module owns its own naming rules for what is and is not a blob, and
/// this harness fabricates names against those rules rather than through the store.
/// These check the fabrication itself, so a harness bug cannot pass as a clean run.
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn payloads_differ_in_length_across_the_cycle() {
        assert_ne!(payload(0).len(), payload(1).len());
        assert_eq!(payload(0).len(), payload(97).len());
    }

    #[test]
    fn fabricated_vault_is_flat_and_countable() {
        let dir = tempfile::TempDir::new().unwrap();
        let vault = dir.path().join("vault");
        let v = fabricate(&vault, 8, true).expect("fabrication failed");
        assert_eq!(v.blobs.len(), 8);
        assert_eq!(count_flat_blobs(&v.root).unwrap(), 8);
        assert!(v.blobs.iter().all(|(h, e)| {
            legacy_flat_blob_path(&v.root, h, e).is_file()
                && !store_blob_path(&v.root, h, e).exists()
        }));
        assert!(all_resolve(&v));
        // Distinct content per index, so a sweep of 8 is a sweep of 8 blobs.
        let unique: std::collections::HashSet<_> = v.blobs.iter().map(|(h, _)| h).collect();
        assert_eq!(unique.len(), 8);
    }

    #[test]
    fn dangling_counts_only_broken_links() {
        let dir = tempfile::TempDir::new().unwrap();
        let root = dir.path();
        let real = root.join("real");
        std::fs::write(&real, b"x").unwrap();
        #[cfg(unix)]
        {
            std::os::unix::fs::symlink(&real, root.join("good")).unwrap();
            std::os::unix::fs::symlink(root.join("gone"), root.join("bad")).unwrap();
            assert_eq!(dangling(root), 1);
        }
    }
}