//! Blob-directory layout: the flat-to-sharded migration and its bookkeeping.
//!
//! Vaults created before 2026-07-29 keep every blob in one flat directory. That
//! collapses at scale: measured on a 289k-file library, import throughput fell
//! about 90% between an empty vault and a 40,000-entry one, and a *dedup* pass
//! doing strictly less work per file (no blob write, no fsync) still ran five
//! times slower than a full write into an empty directory. The cost is kernel
//! filesystem metadata work, so it cannot be optimised away on the read side; the
//! directory has to stop being flat. Numbers and method in wiki `af-benchmarks`.
//!
//! [`migrate_to_sharded`] relocates a vault forward. It is resumable and
//! idempotent, because on the filesystems that need it most a full sweep of a
//! large library takes long enough to be interrupted: every step is a rename to a
//! path derived from the blob's own content, so re-running continues rather than
//! repeating, and an interrupted sweep leaves a vault that still resolves (reads
//! check both layouts, see [`super::existing_blob_path`]).
//!
//! <!-- wiki: af-benchmarks -->

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

use tracing::{instrument, warn};

use super::{SampleStore, blob_shard, store_blob_path};
use crate::config_key::ConfigKey;
use crate::db::Database;
use crate::error::{Result, io_err};

/// On-disk layout of a vault's blob directory.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BlobLayout {
    /// Every blob directly in the store root: `{root}/{hash}.{ext}`. Pre-2026-07-29.
    Flat,
    /// Blobs under a hash-prefix shard: `{root}/{ab}/{hash}.{ext}`.
    Sharded,
}

impl BlobLayout {
    /// The `blob_layout` config value.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            BlobLayout::Flat => "flat",
            BlobLayout::Sharded => "sharded",
        }
    }
}

/// The layout recorded for this vault.
///
/// Absent or unrecognised reads as [`BlobLayout::Flat`], which is the fail-safe
/// direction: it schedules a sweep that finds nothing on an already-sharded vault
/// (one `read_dir`), whereas defaulting to `Sharded` would leave a genuinely flat
/// vault permanently unmigrated and paying the cost this module exists to remove.
///
/// Deliberately not a schema migration. The key's only job is to hold a default,
/// and a migration whose entire body inserts one default row is more moving parts
/// than reading `None` as `Flat`.
pub fn recorded_layout(db: &Database) -> Result<BlobLayout> {
    Ok(match db.get_config(ConfigKey::BlobLayout)?.as_deref() {
        Some(v) if v == BlobLayout::Sharded.as_str() => BlobLayout::Sharded,
        _ => BlobLayout::Flat,
    })
}

/// Outcome of a [`migrate_to_sharded`] pass.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct LayoutMigration {
    /// Blobs relocated into a shard directory.
    pub moved: usize,
    /// Flat blobs discarded because the shard already held the same content.
    pub deduped: usize,
    /// Blobs that could not be relocated. Each is logged.
    pub errors: usize,
    /// The sweep stopped early because cancellation was requested.
    pub cancelled: bool,
    /// The root is clean and the vault is now recorded as [`BlobLayout::Sharded`].
    pub completed: bool,
}

/// Split a store-root filename into `(hash, ext)`, or `None` if it is not a blob.
///
/// Strict on purpose. The store root also holds shard directories, and can hold
/// `{hash}.{ext}.{pid}.tmp` leftovers from an import that died between create and
/// rename. Requiring exactly one dot and a 64-char lowercase-hex stem rejects both
/// a temp file (two dots) and anything a user dropped in by hand, so the sweep
/// only ever renames files it is certain are blobs.
fn parse_blob_name(name: &str) -> Option<(&str, &str)> {
    let (hash, ext) = match name.split_once('.') {
        Some((hash, ext)) => (hash, ext),
        None => (name, ""),
    };
    let is_hash = hash.len() == 64
        && hash
            .bytes()
            .all(|b| b.is_ascii_digit() || b.is_ascii_lowercase() && b.is_ascii_hexdigit());
    if !is_hash || ext.contains('.') {
        return None;
    }
    Some((hash, ext))
}

/// Every flat blob presently in the store root, as `(hash, ext)`.
///
/// Non-recursive: shard directories are entries of the root and are skipped, so
/// this counts exactly the work a sweep still has left.
fn flat_blobs(store_root: &Path) -> Result<Vec<(String, String)>> {
    let mut out = Vec::new();
    let entries = std::fs::read_dir(store_root).map_err(|e| io_err(store_root, e))?;
    for entry in entries.flatten() {
        // A directory here is a shard (or something a user made); never a blob.
        if !entry.file_type().is_ok_and(|t| t.is_file()) {
            continue;
        }
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        if let Some((hash, ext)) = parse_blob_name(name) {
            out.push((hash.to_string(), ext.to_string()));
        }
    }
    Ok(out)
}

/// How many flat blobs are still in the store root.
///
/// Cheap enough to call before deciding whether to start a sweep: one `read_dir`.
pub fn count_flat_blobs(store_root: &Path) -> Result<usize> {
    Ok(flat_blobs(store_root)?.len())
}

/// Relocate every flat blob in the store root into its hash-prefix shard.
///
/// Resumable, idempotent and cancellable. `on_progress` is called with
/// `(done, total)` as each blob is handled. On a clean pass (nothing cancelled, no
/// errors, root verified empty of blobs afterwards) the vault is recorded as
/// [`BlobLayout::Sharded`] and [`LayoutMigration::completed`] is set; otherwise the
/// recorded layout is left alone so the next run picks up the remainder.
///
/// No per-blob fsync. The relocation is a pure rename of content-addressed data, so
/// an untimely crash can only lose the *dirent update*, never bytes: the blob is
/// still at one of the two paths a read checks, and re-running the sweep finishes
/// the job. Paying a metadata flush per file would reproduce the per-file fsync cost
/// that the same measurement found was worth about 27% of import time.
#[instrument(skip_all)]
pub fn migrate_to_sharded(
    store: &SampleStore,
    db: &Database,
    cancel: &AtomicBool,
    on_progress: &mut dyn FnMut(usize, usize),
) -> Result<LayoutMigration> {
    let root = store.root();
    let pending = flat_blobs(root)?;
    let total = pending.len();
    let mut report = LayoutMigration::default();

    for (done, (hash, ext)) in pending.iter().enumerate() {
        if cancel.load(Ordering::Acquire) {
            report.cancelled = true;
            break;
        }
        let src = super::legacy_flat_blob_path(root, hash, ext);
        let dest = store_blob_path(root, hash, ext);
        let shard_dir = root.join(blob_shard(hash));

        if let Err(e) = std::fs::create_dir_all(&shard_dir) {
            warn!(shard = %shard_dir.display(), "layout: shard create failed: {e}");
            report.errors += 1;
            continue;
        }

        // A blob already at the destination is the same content by construction,
        // so the flat copy is redundant and should go. Size is still checked
        // first: if they differ, one of them is a truncated blob from a
        // pre-atomic-rename crash, and silently deleting either could destroy the
        // intact one. That case is left alone for a human and counted as an error.
        match (std::fs::metadata(&dest), std::fs::metadata(&src)) {
            (Ok(d), Ok(s)) if d.len() == s.len() => match std::fs::remove_file(&src) {
                Ok(()) => report.deduped += 1,
                Err(e) => {
                    warn!(path = %src.display(), "layout: redundant flat blob unlink failed: {e}");
                    report.errors += 1;
                }
            },
            (Ok(d), Ok(s)) => {
                warn!(
                    path = %src.display(),
                    flat_len = s.len(),
                    sharded_len = d.len(),
                    "layout: size mismatch between flat and sharded blob, leaving both for inspection"
                );
                report.errors += 1;
            }
            _ => match std::fs::rename(&src, &dest) {
                Ok(()) => report.moved += 1,
                Err(e) => {
                    warn!(path = %src.display(), "layout: rename into shard failed: {e}");
                    report.errors += 1;
                }
            },
        }
        on_progress(done + 1, total);
    }

    // One directory fsync for the whole sweep, so the renames are durable without
    // paying a metadata flush per blob. Best-effort: not every filesystem supports
    // it, and failure here only weakens durability, never correctness.
    if let Ok(d) = std::fs::File::open(root) {
        let _ = d.sync_all();
    }

    // Only claim completion against a re-scan. A blob could have been written
    // flat by another process between the enumeration and here, and recording
    // `Sharded` over one would strand it: reads would still find it via the
    // fallback, but nothing would ever move it.
    if !report.cancelled && report.errors == 0 && count_flat_blobs(root)? == 0 {
        db.set_config(ConfigKey::BlobLayout, BlobLayout::Sharded.as_str())?;
        report.completed = true;
    }

    Ok(report)
}

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

    const HASH_A: &str = "aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899";
    const HASH_B: &str = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff";

    fn store_with_flat_blob(root: &Path, hash: &str, ext: &str, bytes: &[u8]) {
        std::fs::write(super::super::legacy_flat_blob_path(root, hash, ext), bytes).unwrap();
    }

    #[test]
    fn parse_blob_name_accepts_a_blob_with_and_without_extension() {
        assert_eq!(parse_blob_name(HASH_A), Some((HASH_A, "")));
        assert_eq!(
            parse_blob_name(&format!("{HASH_A}.wav")),
            Some((HASH_A, "wav"))
        );
    }

    #[test]
    fn parse_blob_name_rejects_temp_files_and_non_blobs() {
        // The exact shape import leaves behind when it dies before the rename.
        assert_eq!(parse_blob_name(&format!("{HASH_A}.wav.12345.tmp")), None);
        assert_eq!(parse_blob_name("audiofiles.db"), None);
        assert_eq!(parse_blob_name("notes.txt"), None);
        // Uppercase hex is not the spelling the store writes.
        assert_eq!(parse_blob_name(&HASH_A.to_uppercase()), None);
        // Right charset, wrong length.
        assert_eq!(parse_blob_name("aabbcc.wav"), None);
    }

    #[test]
    fn sweep_relocates_flat_blobs_and_records_the_layout() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();

        store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
        store_with_flat_blob(&root, HASH_B, "flac", b"bbbbbb");

        assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
        assert_eq!(count_flat_blobs(&root).unwrap(), 2);

        let cancel = AtomicBool::new(false);
        let mut seen = Vec::new();
        let report =
            migrate_to_sharded(&store, &db, &cancel, &mut |d, t| seen.push((d, t))).unwrap();

        assert_eq!(report.moved, 2);
        assert_eq!(report.deduped, 0);
        assert_eq!(report.errors, 0);
        assert!(report.completed);
        assert_eq!(seen, vec![(1, 2), (2, 2)]);

        // Bytes are where the sharded layout says, and gone from the flat one.
        assert!(store_blob_path(&root, HASH_A, "wav").exists());
        assert!(root.join("aa").is_dir());
        assert!(!super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
        assert_eq!(count_flat_blobs(&root).unwrap(), 0);
        assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Sharded);

        // And the store resolves them.
        let resolved = store
            .sample_path(&SampleHash::from_trusted(HASH_A.to_string()), "wav")
            .unwrap();
        assert_eq!(resolved, store_blob_path(&root, HASH_A, "wav"));
    }

    #[test]
    fn sweep_is_idempotent() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();
        store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");

        let cancel = AtomicBool::new(false);
        let first = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
        let second = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();

        assert_eq!(first.moved, 1);
        assert_eq!(second.moved, 0);
        assert!(
            second.completed,
            "a clean already-sharded vault stays sharded"
        );
        assert!(store_blob_path(&root, HASH_A, "wav").exists());
    }

    #[test]
    fn sweep_discards_a_redundant_flat_blob_matching_its_shard() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();

        // Same content in both layouts: an interrupted sweep plus a repair write.
        store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
        std::fs::create_dir_all(root.join("aa")).unwrap();
        std::fs::write(store_blob_path(&root, HASH_A, "wav"), b"aaaa").unwrap();

        let cancel = AtomicBool::new(false);
        let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();

        assert_eq!(report.deduped, 1);
        assert_eq!(report.moved, 0);
        assert_eq!(report.errors, 0);
        assert!(report.completed);
        assert!(!super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
    }

    #[test]
    fn sweep_leaves_a_size_mismatch_alone_and_does_not_complete() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();

        // One of these is a truncated blob from a pre-atomic-rename crash. Deleting
        // either could destroy the intact copy, so both must survive the sweep.
        store_with_flat_blob(&root, HASH_A, "wav", b"aaaaaaaa");
        std::fs::create_dir_all(root.join("aa")).unwrap();
        std::fs::write(store_blob_path(&root, HASH_A, "wav"), b"aa").unwrap();

        let cancel = AtomicBool::new(false);
        let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();

        assert_eq!(report.errors, 1);
        assert_eq!(report.moved, 0);
        assert_eq!(report.deduped, 0);
        assert!(
            !report.completed,
            "an unresolved blob must not record Sharded"
        );
        assert!(super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
        assert!(store_blob_path(&root, HASH_A, "wav").exists());
        assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
    }

    #[test]
    fn cancelled_sweep_keeps_the_flat_layout_recorded() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();
        store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");

        let cancel = AtomicBool::new(true);
        let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();

        assert!(report.cancelled);
        assert_eq!(report.moved, 0);
        assert!(!report.completed);
        assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
        // The blob is untouched and still resolvable, so a cancel is not data loss.
        assert!(super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
    }

    #[test]
    fn sweep_ignores_temp_leftovers_and_shard_directories() {
        let dir = tempfile::TempDir::new().unwrap();
        let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
        let root = dir.path().join("samples");
        let store = SampleStore::new(&root).unwrap();

        let tmp = root.join(format!("{HASH_A}.wav.999.tmp"));
        std::fs::write(&tmp, b"partial").unwrap();
        store_with_flat_blob(&root, HASH_B, "wav", b"bbbb");

        let cancel = AtomicBool::new(false);
        let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();

        assert_eq!(report.moved, 1, "only the real blob moves");
        assert_eq!(report.errors, 0);
        assert!(tmp.exists(), "a temp leftover is not the sweep's business");
        assert!(
            report.completed,
            "a temp file is not a flat blob, so it must not block completion"
        );
    }
}