//! Content-addressed blob store (Phase 2).
//!
//! Every blob is identified by the BLAKE3 hash of its full content:
//! same bytes → same `BlobId`, always. Blobs are physically stored as
//! a sequence of BLAKE3-hashed 4 MB chunks, so two blobs that share a
//! prefix (say two cargo target dirs with 95% of the same deps) share
//! storage at chunk granularity without any special detection logic.
//!
//! # Layout
//!
//! ```text
//! <root>/
//!   blobs/
//!     <bb>/<blob_id>.manifest.json      { chunks: [<chunk_hash>...], total_size: u64 }
//!   chunks/
//!     <cc>/<chunk_hash>                  raw bytes of one chunk
//!   .tmp/                               atomic-rename staging area
//! ```
//!
//! `<bb>` and `<cc>` are the first two hex chars of the corresponding
//! hash. Keeps directory fan-out bounded (256 entries per level) —
//! important on a warm-tier ZFS dataset with tens of thousands of blobs.
//!
//! Writes use a temp file + rename so a crash mid-write leaves either
//! a complete file or nothing (never a truncated partial). Chunks are
//! immutable: if `<chunk_hash>` already exists, we skip re-writing it.
//!
//! # Deletion + GC
//!
//! [`BlobStore::delete_manifest`] removes the top-level manifest but
//! leaves chunks — orphan cleanup is a separate sweep so we avoid
//! walking the whole store for every deletion. Run
//! [`BlobStore::gc_orphan_chunks`] periodically (nightly is fine) to
//! reclaim any chunks no longer referenced by a live manifest.

use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

/// Physical chunk size. 4 MB is the sweet spot for our workloads:
/// small enough that dedup catches typical file-tree overlaps between
/// similar cargo build outputs, large enough that per-chunk overhead
/// (hash compute + syscall) stays negligible on hot-path reads.
pub const CHUNK_SIZE: usize = 4 * 1024 * 1024;

/// Content-addressed identifier for a blob: the BLAKE3 hash of its full
/// content. Compares/orderings are on the raw 32-byte hash, so
/// serialisation to hex is only for display and on-disk paths.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct BlobId([u8; 32]);

impl BlobId {
    pub fn from_bytes(hash: [u8; 32]) -> Self {
        Self(hash)
    }

    pub fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }

    /// Lowercase hex string, 64 chars long — the on-disk path suffix.
    pub fn to_hex(&self) -> String {
        hex_encode(&self.0)
    }

    /// Parse from lowercase hex. Rejects wrong-length or non-hex input.
    pub fn from_hex(s: &str) -> Result<Self> {
        let bytes = hex_decode(s)?;
        if bytes.len() != 32 {
            bail!(
                "BlobId hex must be exactly 64 chars (32 bytes); got {}",
                bytes.len() * 2
            );
        }
        let mut arr = [0u8; 32];
        arr.copy_from_slice(&bytes);
        Ok(Self(arr))
    }
}

impl std::fmt::Display for BlobId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.to_hex())
    }
}

impl Serialize for BlobId {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(&self.to_hex())
    }
}

impl<'de> Deserialize<'de> for BlobId {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let s = String::deserialize(d)?;
        BlobId::from_hex(&s).map_err(serde::de::Error::custom)
    }
}

/// Physical hash of a single chunk. Same shape as [`BlobId`] but a
/// distinct type so we can't accidentally lookup a chunk with a blob
/// hash or vice versa.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct ChunkHash([u8; 32]);

impl ChunkHash {
    pub fn from_bytes(hash: [u8; 32]) -> Self {
        Self(hash)
    }

    pub fn as_bytes(&self) -> &[u8; 32] {
        &self.0
    }

    pub fn to_hex(&self) -> String {
        hex_encode(&self.0)
    }

    pub fn from_hex(s: &str) -> Result<Self> {
        let bytes = hex_decode(s)?;
        if bytes.len() != 32 {
            bail!(
                "ChunkHash hex must be exactly 64 chars (32 bytes); got {}",
                bytes.len() * 2
            );
        }
        let mut arr = [0u8; 32];
        arr.copy_from_slice(&bytes);
        Ok(Self(arr))
    }
}

impl Serialize for ChunkHash {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(&self.to_hex())
    }
}

impl<'de> Deserialize<'de> for ChunkHash {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let s = String::deserialize(d)?;
        ChunkHash::from_hex(&s).map_err(serde::de::Error::custom)
    }
}

/// Summary of a stored blob without reading its contents.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BlobStat {
    pub total_size: u64,
    pub chunk_count: usize,
}

/// On-disk manifest for a blob. Public because the RPC layer (Phase 2b)
/// serves this directly so a receiver can request only missing chunks.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BlobManifest {
    pub blob_id: BlobId,
    pub total_size: u64,
    pub chunks: Vec<ChunkHash>,
}

/// Report from [`BlobStore::gc_orphan_chunks`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GcReport {
    pub chunks_scanned: usize,
    pub chunks_removed: usize,
    pub bytes_reclaimed: u64,
}

/// Phase 7b (2026-07-14): report from [`BlobStore::repair_chunks`].
///
/// For each corrupt/missing chunk the caller supplied, records what
/// happened: successfully fetched + written, fetcher returned None
/// (no peer had it), or the fetch itself errored (network, protocol).
#[derive(Debug, Clone, Default)]
pub struct RepairReport {
    pub attempted: usize,
    pub repaired: usize,
    pub unrecoverable: Vec<ChunkHash>,
    pub errors: Vec<(ChunkHash, String)>,
}

/// Phase 7a (2026-07-14): report from [`BlobStore::scrub_all`].
///
/// A scrub walks every manifest, recomputes BLAKE3 for each referenced
/// chunk file, and reports mismatches without touching disk state.
/// Read-only; safe to run against a live daemon.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ScrubReport {
    pub manifests_scanned: usize,
    pub chunks_scanned: usize,
    pub chunks_ok: usize,
    pub chunks_corrupt: usize,
    pub chunks_missing: usize,
    /// (owning blob, chunk-hash whose file contents don't hash to that hash).
    /// Bounded by `chunks_corrupt`; kept explicit so operators can act.
    pub corrupt_chunks: Vec<(BlobId, ChunkHash)>,
    /// (owning blob, chunk-hash whose file is absent from disk).
    /// Bounded by `chunks_missing`.
    pub missing_chunks: Vec<(BlobId, ChunkHash)>,
}

/// Content-addressed blob store rooted at a filesystem directory.
#[derive(Debug, Clone)]
pub struct BlobStore {
    root: PathBuf,
}

/// Field finding 2026-07-12: per-manifest summary used by
/// [`BlobStore::evict_to_size_cap`]. Enough to decide eviction order
/// + know which chunks to decrement refcount on.
#[derive(Debug, Clone)]
struct ManifestSummary {
    blob_id: BlobId,
    chunks: Vec<ChunkHash>,
    /// Manifest file's mtime as unix seconds; 0 if unreadable.
    manifest_mtime: u64,
}

impl BlobStore {
    /// Open (create if missing) a blob store rooted at `root`. Creates
    /// the `blobs/`, `chunks/`, and `.tmp/` subdirs. Safe to call on
    /// an existing store — no data is touched.
    pub fn open(root: PathBuf) -> Result<Self> {
        std::fs::create_dir_all(root.join("blobs"))
            .with_context(|| format!("creating blobs dir under {}", root.display()))?;
        std::fs::create_dir_all(root.join("chunks"))
            .with_context(|| format!("creating chunks dir under {}", root.display()))?;
        std::fs::create_dir_all(root.join(".tmp"))
            .with_context(|| format!("creating .tmp dir under {}", root.display()))?;
        Ok(Self { root })
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Store `data`, returning the deterministic content hash.
    /// Idempotent: writing the same bytes twice writes to the same
    /// manifest path and yields the same BlobId. Chunks that already
    /// exist on disk are not re-written.
    pub async fn put_bytes(&self, data: &[u8]) -> Result<BlobId> {
        let mut blob_hasher = blake3::Hasher::new();
        blob_hasher.update(data);
        let blob_id = BlobId(blob_hasher.finalize().into());

        let mut chunk_hashes = Vec::with_capacity(data.len().div_ceil(CHUNK_SIZE));
        for chunk in data.chunks(CHUNK_SIZE) {
            let chunk_hash = ChunkHash(blake3::hash(chunk).into());
            self.write_chunk_if_absent(&chunk_hash, chunk).await?;
            chunk_hashes.push(chunk_hash);
        }

        let manifest = BlobManifest {
            blob_id,
            total_size: data.len() as u64,
            chunks: chunk_hashes,
        };
        self.write_manifest_if_absent(&manifest).await?;
        Ok(blob_id)
    }

    /// Read the entire blob back into memory. Returns `None` when no
    /// manifest exists for `id`. Verifies the reconstructed bytes hash
    /// back to `id` — mismatch means store corruption and returns Err.
    pub async fn get_bytes(&self, id: &BlobId) -> Result<Option<Vec<u8>>> {
        let manifest = match self.load_manifest(id).await? {
            Some(m) => m,
            None => return Ok(None),
        };
        let mut out = Vec::with_capacity(manifest.total_size as usize);
        for chunk_hash in &manifest.chunks {
            let path = self.chunk_path(chunk_hash);
            let bytes = tokio::fs::read(&path)
                .await
                .with_context(|| format!("reading chunk {}", path.display()))?;
            let recomputed = ChunkHash(blake3::hash(&bytes).into());
            if recomputed != *chunk_hash {
                bail!(
                    "chunk hash mismatch at {}: manifest says {}, disk hashes to {}",
                    path.display(),
                    chunk_hash.to_hex(),
                    recomputed.to_hex()
                );
            }
            out.extend_from_slice(&bytes);
        }
        if out.len() as u64 != manifest.total_size {
            bail!(
                "reassembled blob size {} does not match manifest {}",
                out.len(),
                manifest.total_size
            );
        }
        let final_hash = BlobId(blake3::hash(&out).into());
        if final_hash != *id {
            bail!(
                "reassembled blob hash {} does not match requested {}",
                final_hash.to_hex(),
                id.to_hex()
            );
        }
        Ok(Some(out))
    }

    /// Store bytes from an async reader, returning the deterministic
    /// content hash. Streams input in 4 MiB frames — memory ceiling is
    /// one chunk buffer regardless of blob size. Same idempotence +
    /// dedup guarantees as [`put_bytes`]: identical content across two
    /// calls produces the same BlobId and stores each unique chunk
    /// exactly once on disk.
    ///
    /// EOF at a chunk boundary is honoured: an empty reader produces
    /// the empty-blob BlobId with zero chunks, matching `put_bytes(&[])`.
    pub async fn put_stream<R>(&self, mut reader: R) -> Result<BlobId>
    where
        R: AsyncRead + Unpin,
    {
        let mut blob_hasher = blake3::Hasher::new();
        let mut chunk_hashes: Vec<ChunkHash> = Vec::new();
        let mut total_size: u64 = 0;
        let mut buf = vec![0u8; CHUNK_SIZE];

        loop {
            // Fill the chunk buffer or hit EOF. `read` may return fewer
            // bytes than requested; loop until we have CHUNK_SIZE or the
            // reader is drained.
            let mut have = 0usize;
            while have < CHUNK_SIZE {
                let n = reader
                    .read(&mut buf[have..])
                    .await
                    .context("reading from streaming source")?;
                if n == 0 {
                    break;
                }
                have += n;
            }
            if have == 0 {
                break; // Reader drained on a clean chunk boundary.
            }
            let bytes = &buf[..have];
            blob_hasher.update(bytes);
            let chunk_hash = ChunkHash(blake3::hash(bytes).into());
            self.write_chunk_if_absent(&chunk_hash, bytes).await?;
            chunk_hashes.push(chunk_hash);
            total_size += have as u64;
            if have < CHUNK_SIZE {
                break; // Short final chunk; no more data possible.
            }
        }

        let blob_id = BlobId(blob_hasher.finalize().into());
        let manifest = BlobManifest {
            blob_id,
            total_size,
            chunks: chunk_hashes,
        };
        self.write_manifest_if_absent(&manifest).await?;
        Ok(blob_id)
    }

    /// Stream a stored blob into `writer` chunk-by-chunk. Returns
    /// `Ok(false)` when no manifest exists for `id` (writer is left
    /// untouched); `Ok(true)` on success. Each chunk is hash-verified
    /// before it's emitted, so store corruption surfaces as `Err` mid
    /// stream — callers who need to abort a partial write should
    /// invalidate the destination buffer on error.
    pub async fn stream_to<W>(&self, id: &BlobId, writer: &mut W) -> Result<bool>
    where
        W: AsyncWrite + Unpin,
    {
        let manifest = match self.load_manifest(id).await? {
            Some(m) => m,
            None => return Ok(false),
        };
        for chunk_hash in &manifest.chunks {
            let path = self.chunk_path(chunk_hash);
            let bytes = tokio::fs::read(&path)
                .await
                .with_context(|| format!("reading chunk {}", path.display()))?;
            let recomputed = ChunkHash(blake3::hash(&bytes).into());
            if recomputed != *chunk_hash {
                bail!(
                    "chunk hash mismatch at {}: manifest says {}, disk hashes to {}",
                    path.display(),
                    chunk_hash.to_hex(),
                    recomputed.to_hex()
                );
            }
            writer
                .write_all(&bytes)
                .await
                .context("writing chunk bytes to destination")?;
        }
        Ok(true)
    }

    // ── Chunk-level API (Phase 2d) ───────────────────────────────────
    //
    // These operations expose the store's chunk substrate directly so a
    // sender can query + upload one chunk at a time. Combined with
    // gossip-derived affinity + the `HasChunk` RPC, they let a peer
    // replicate a blob by transferring only chunks the receiver is
    // missing — the "big win when caches overlap" case.

    /// Whether a chunk file exists on disk. No hash verification —
    /// callers who want cryptographic proof should follow with
    /// [`read_chunk`] and check the returned hash themselves.
    pub async fn has_chunk(&self, hash: &ChunkHash) -> Result<bool> {
        match tokio::fs::metadata(self.chunk_path(hash)).await {
            Ok(_) => Ok(true),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(e) => Err(anyhow::Error::from(e)),
        }
    }

    /// Read a single chunk's raw bytes. Returns `None` when the chunk
    /// isn't on disk. Verifies the chunk hashes to the requested
    /// `hash` (defense in depth against silent corruption); a mismatch
    /// returns `Err`.
    pub async fn read_chunk(&self, hash: &ChunkHash) -> Result<Option<Vec<u8>>> {
        let path = self.chunk_path(hash);
        let bytes = match tokio::fs::read(&path).await {
            Ok(b) => b,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(anyhow::Error::from(e)),
        };
        let recomputed = ChunkHash(blake3::hash(&bytes).into());
        if recomputed != *hash {
            bail!(
                "chunk hash mismatch at {}: requested {}, disk hashes to {}",
                path.display(),
                hash.to_hex(),
                recomputed.to_hex()
            );
        }
        Ok(Some(bytes))
    }

    /// Store one chunk. Verifies that `bytes` actually hash to `hash`
    /// before writing — the caller can't accidentally (or maliciously)
    /// stash unrelated bytes under a hash it doesn't own. Idempotent:
    /// writing the same chunk twice is a no-op after the first.
    pub async fn put_chunk(&self, hash: &ChunkHash, bytes: &[u8]) -> Result<()> {
        let recomputed = ChunkHash(blake3::hash(bytes).into());
        if recomputed != *hash {
            bail!(
                "chunk hash mismatch: claimed {}, bytes hash to {}",
                hash.to_hex(),
                recomputed.to_hex()
            );
        }
        self.write_chunk_if_absent(hash, bytes).await
    }

    /// Persist a manifest whose referenced chunks are already on disk.
    /// Verifies that every chunk in the manifest exists before writing —
    /// callers can't accidentally register a manifest that references
    /// missing chunks (which would break subsequent `get_bytes`).
    ///
    /// Returns the list of chunk hashes that were missing when the
    /// call started; if empty, the manifest was written. Otherwise
    /// nothing was written and the client is expected to upload the
    /// missing chunks (typically via [`put_chunk`]) and retry.
    pub async fn put_manifest_verified(
        &self,
        manifest: &BlobManifest,
    ) -> Result<Vec<ChunkHash>> {
        let mut missing = Vec::new();
        for chunk_hash in &manifest.chunks {
            if !self.has_chunk(chunk_hash).await? {
                missing.push(*chunk_hash);
            }
        }
        if !missing.is_empty() {
            return Ok(missing);
        }
        self.write_manifest_if_absent(manifest).await?;
        Ok(Vec::new())
    }

    /// Whether a manifest exists for `id`. Cheap — no chunk reads.
    pub async fn contains(&self, id: &BlobId) -> Result<bool> {
        match tokio::fs::metadata(self.manifest_path(id)).await {
            Ok(_) => Ok(true),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(e) => Err(anyhow::Error::from(e)),
        }
    }

    /// Metadata for a stored blob — size + chunk count — without reading
    /// any chunk data.
    pub async fn stat(&self, id: &BlobId) -> Result<Option<BlobStat>> {
        Ok(self.load_manifest(id).await?.map(|m| BlobStat {
            total_size: m.total_size,
            chunk_count: m.chunks.len(),
        }))
    }

    /// Load the raw manifest. Public so the RPC layer can serve it as
    /// the response to a "list chunks I need" request.
    pub async fn load_manifest(&self, id: &BlobId) -> Result<Option<BlobManifest>> {
        let path = self.manifest_path(id);
        let bytes = match tokio::fs::read(&path).await {
            Ok(b) => b,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(anyhow::Error::from(e)),
        };
        let manifest: BlobManifest = serde_json::from_slice(&bytes)
            .with_context(|| format!("parsing manifest at {}", path.display()))?;
        if manifest.blob_id != *id {
            bail!(
                "manifest at {} claims blob_id {} but was requested as {}",
                path.display(),
                manifest.blob_id.to_hex(),
                id.to_hex()
            );
        }
        Ok(Some(manifest))
    }

    /// Delete the top-level manifest for a blob. Chunks stay — they
    /// may be shared with other blobs; [`gc_orphan_chunks`] does the
    /// eventual reclamation.
    ///
    /// Returns `true` if a manifest was removed, `false` if none existed.
    pub async fn delete_manifest(&self, id: &BlobId) -> Result<bool> {
        let path = self.manifest_path(id);
        match tokio::fs::remove_file(&path).await {
            Ok(()) => Ok(true),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(e) => Err(anyhow::Error::from(e)),
        }
    }

    /// Walk every live manifest to build the referenced-chunk set,
    /// then remove any chunk file not in that set. Bounded by the
    /// number of manifests + chunks on disk. Safe to interrupt: any
    /// crash mid-sweep leaves live chunks intact (we only delete files
    /// not in the referenced set).
    pub async fn gc_orphan_chunks(&self) -> Result<GcReport> {
        let referenced = self.collect_referenced_chunks().await?;

        let chunks_root = self.root.join("chunks");
        let mut chunks_scanned = 0usize;
        let mut chunks_removed = 0usize;
        let mut bytes_reclaimed = 0u64;

        let mut top_entries = tokio::fs::read_dir(&chunks_root)
            .await
            .with_context(|| format!("reading {}", chunks_root.display()))?;
        while let Some(bucket) = top_entries.next_entry().await? {
            if !bucket.file_type().await?.is_dir() {
                continue;
            }
            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
            while let Some(entry) = inner.next_entry().await? {
                if !entry.file_type().await?.is_file() {
                    continue;
                }
                chunks_scanned += 1;
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                let hash = match ChunkHash::from_hex(name_str) {
                    Ok(h) => h,
                    Err(_) => continue, // unknown filename shape; leave alone
                };
                if referenced.contains(&hash) {
                    continue;
                }
                if let Ok(meta) = entry.metadata().await {
                    bytes_reclaimed = bytes_reclaimed.saturating_add(meta.len());
                }
                if tokio::fs::remove_file(entry.path()).await.is_ok() {
                    chunks_removed += 1;
                }
            }
        }

        Ok(GcReport {
            chunks_scanned,
            chunks_removed,
            bytes_reclaimed,
        })
    }

    /// Field finding 2026-07-12: enforce a size cap by evicting blobs
    /// oldest-first (LRU on manifest mtime) until the live-referenced
    /// chunk footprint is `<= max_bytes`.
    ///
    /// Orphan-chunk GC alone is not enough — as long as fingerprint→blob
    /// refs keep getting `PutRef`'d, the manifest set (and thus the
    /// referenced chunk set) grows unbounded. This function evicts blob
    /// manifests + reclaims the now-unreferenced chunks.
    ///
    /// Semantics:
    ///   * Ordering: manifests sorted by mtime ascending (oldest first).
    ///     `stream_to`/`load_manifest` do not touch mtime, so eviction is
    ///     effectively FIFO — good enough for a pilot. LRU-by-read is a
    ///     future refinement.
    ///   * Correctness: a blob's chunks may be shared with other blobs.
    ///     After each manifest delete we recompute the referenced set
    ///     and delete now-orphan chunks. Cheap because we accumulate
    ///     touched chunks per delete rather than re-walking the whole
    ///     tree.
    ///   * Bookkeeping: `bytes_reclaimed` counts real bytes freed from
    ///     disk. `chunks_removed` is the number of chunk files deleted
    ///     (not the number of chunk references removed).
    ///
    /// Returns [`GcReport`] with the totals across all evicted blobs.
    /// `chunks_scanned` is 0 (this function doesn't do a full scan;
    /// call [`gc_orphan_chunks`] separately for that).
    pub async fn evict_to_size_cap(&self, max_bytes: u64) -> Result<GcReport> {
        // No pins → every manifest is a candidate. Delegate.
        self.evict_to_size_cap_with_pins(
            max_bytes,
            &std::collections::HashSet::new(),
        )
        .await
    }

    /// Phase 4 (2026-07-13): pin-aware LRU eviction. `pinned_blobs`
    /// is the set of blob IDs the caller considers protected from
    /// eviction — typically the set of every blob referenced by a
    /// live tag. Pinned manifests are skipped entirely; their chunks
    /// stay in the referenced set so shared chunks with evicted
    /// blobs also survive.
    ///
    /// The eviction pass may go under cap earlier than a pin-free
    /// pass would — pinned blobs count against `max_bytes` but can't
    /// be evicted to make room, so if the pinned footprint alone
    /// exceeds the cap, we return without doing anything (the caller
    /// is expected to raise `blob_max_gb` or drop pins).
    pub async fn evict_to_size_cap_with_pins(
        &self,
        max_bytes: u64,
        pinned_blobs: &std::collections::HashSet<BlobId>,
    ) -> Result<GcReport> {
        // 1. Compute per-manifest chunk sets + total live size.
        let manifest_summaries = self.collect_manifest_summaries().await?;
        let mut referenced: std::collections::HashMap<ChunkHash, u32> =
            std::collections::HashMap::new();
        for summary in &manifest_summaries {
            for hash in &summary.chunks {
                *referenced.entry(*hash).or_insert(0) += 1;
            }
        }
        // Actual size = sum of file lengths of referenced chunks.
        let mut current_size: u64 = 0;
        for hash in referenced.keys() {
            if let Ok(meta) = tokio::fs::metadata(&self.chunk_path(hash)).await {
                current_size = current_size.saturating_add(meta.len());
            }
        }

        let mut chunks_removed = 0usize;
        let mut bytes_reclaimed = 0u64;

        // 2. Sort oldest-first + evict manifests until under cap.
        //    Skip pinned blobs entirely — their chunks stay in
        //    `referenced` so any shared chunks also stay put.
        let mut summaries = manifest_summaries;
        summaries.sort_by_key(|s| s.manifest_mtime);
        for summary in summaries {
            if current_size <= max_bytes {
                break;
            }
            if pinned_blobs.contains(&summary.blob_id) {
                continue;
            }
            self.delete_manifest(&summary.blob_id).await?;
            // For each chunk this manifest used: decrement refcount;
            // if it hits zero, delete the chunk file + free its bytes.
            for hash in &summary.chunks {
                let entry = referenced.entry(*hash).or_insert(0);
                if *entry > 0 {
                    *entry -= 1;
                }
                if *entry == 0 {
                    let path = self.chunk_path(hash);
                    if let Ok(meta) = tokio::fs::metadata(&path).await {
                        let sz = meta.len();
                        if tokio::fs::remove_file(&path).await.is_ok() {
                            chunks_removed += 1;
                            bytes_reclaimed = bytes_reclaimed.saturating_add(sz);
                            current_size = current_size.saturating_sub(sz);
                        }
                    }
                    referenced.remove(hash);
                }
            }
        }

        Ok(GcReport {
            chunks_scanned: 0,
            chunks_removed,
            bytes_reclaimed,
        })
    }

    /// Enumerate every on-disk manifest with the info eviction needs:
    /// blob_id, chunk set, and mtime for LRU ordering. Bounded by the
    /// number of manifests (small — one per cached target dir).
    async fn collect_manifest_summaries(&self) -> Result<Vec<ManifestSummary>> {
        let blobs_root = self.root.join("blobs");
        let mut summaries = Vec::new();
        let mut top = match tokio::fs::read_dir(&blobs_root).await {
            Ok(t) => t,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(summaries),
            Err(e) => return Err(anyhow::Error::from(e)),
        };
        while let Some(bucket) = top.next_entry().await? {
            if !bucket.file_type().await?.is_dir() {
                continue;
            }
            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
            while let Some(entry) = inner.next_entry().await? {
                if !entry.file_type().await?.is_file() {
                    continue;
                }
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                let hex = match name_str.strip_suffix(".manifest.json") {
                    Some(h) => h,
                    None => continue,
                };
                let bid = match BlobId::from_hex(hex) {
                    Ok(b) => b,
                    Err(_) => continue,
                };
                let bytes = match tokio::fs::read(entry.path()).await {
                    Ok(b) => b,
                    Err(_) => continue,
                };
                let manifest: BlobManifest = match serde_json::from_slice(&bytes) {
                    Ok(m) => m,
                    Err(_) => continue,
                };
                let meta = match entry.metadata().await {
                    Ok(m) => m,
                    Err(_) => continue,
                };
                let manifest_mtime = meta
                    .modified()
                    .ok()
                    .and_then(|t| {
                        t.duration_since(std::time::UNIX_EPOCH)
                            .ok()
                            .map(|d| d.as_secs())
                    })
                    .unwrap_or(0);
                summaries.push(ManifestSummary {
                    blob_id: bid,
                    chunks: manifest.chunks,
                    manifest_mtime,
                });
            }
        }
        Ok(summaries)
    }

    // ── internals ────────────────────────────────────────────────────

    fn manifest_path(&self, id: &BlobId) -> PathBuf {
        let hex = id.to_hex();
        self.root
            .join("blobs")
            .join(&hex[..2])
            .join(format!("{hex}.manifest.json"))
    }

    /// On-disk path for a chunk. Public so tests + advanced callers
    /// (e.g. custom migration tools) can address individual chunks
    /// without duplicating the layout convention.
    pub fn chunk_path(&self, hash: &ChunkHash) -> PathBuf {
        let hex = hash.to_hex();
        self.root.join("chunks").join(&hex[..2]).join(&hex)
    }

    /// Write a chunk if it isn't already present. Atomic via temp
    /// file + rename so a concurrent reader never sees a partial file.
    async fn write_chunk_if_absent(&self, hash: &ChunkHash, bytes: &[u8]) -> Result<()> {
        let final_path = self.chunk_path(hash);
        if tokio::fs::metadata(&final_path).await.is_ok() {
            return Ok(());
        }
        if let Some(parent) = final_path.parent() {
            tokio::fs::create_dir_all(parent).await.with_context(|| {
                format!("creating chunk bucket {}", parent.display())
            })?;
        }
        self.atomic_write(&final_path, bytes).await
    }

    /// Write manifest JSON if not already present. Same atomicity as
    /// chunks; putting identical content is a no-op after the first.
    async fn write_manifest_if_absent(&self, manifest: &BlobManifest) -> Result<()> {
        let final_path = self.manifest_path(&manifest.blob_id);
        if tokio::fs::metadata(&final_path).await.is_ok() {
            return Ok(());
        }
        if let Some(parent) = final_path.parent() {
            tokio::fs::create_dir_all(parent).await.with_context(|| {
                format!("creating manifest bucket {}", parent.display())
            })?;
        }
        let bytes = serde_json::to_vec_pretty(manifest)
            .context("serialising manifest to JSON")?;
        self.atomic_write(&final_path, &bytes).await
    }

    /// Write `bytes` to `final_path` atomically: stage in `.tmp/`, then
    /// rename. Rename is atomic within a filesystem, and both source
    /// and destination live under `self.root` so we're always on the
    /// same FS.
    async fn atomic_write(&self, final_path: &Path, bytes: &[u8]) -> Result<()> {
        let tmp_dir = self.root.join(".tmp");
        let tmp_name = format!(
            "{}.{}",
            std::process::id(),
            RANDOM_SUFFIX.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        );
        let tmp_path = tmp_dir.join(tmp_name);
        {
            let mut f = tokio::fs::File::create(&tmp_path).await.with_context(|| {
                format!("creating tmp file {}", tmp_path.display())
            })?;
            f.write_all(bytes).await?;
            f.sync_all().await?;
        }
        tokio::fs::rename(&tmp_path, final_path)
            .await
            .with_context(|| {
                format!(
                    "renaming {} → {}",
                    tmp_path.display(),
                    final_path.display()
                )
            })?;
        Ok(())
    }

    /// Walk every manifest and collect the union of chunk hashes they
    /// reference. Used by [`gc_orphan_chunks`].
    async fn collect_referenced_chunks(
        &self,
    ) -> Result<std::collections::HashSet<ChunkHash>> {
        let blobs_root = self.root.join("blobs");
        let mut referenced = std::collections::HashSet::new();
        let mut top_entries = tokio::fs::read_dir(&blobs_root).await?;
        while let Some(bucket) = top_entries.next_entry().await? {
            if !bucket.file_type().await?.is_dir() {
                continue;
            }
            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
            while let Some(entry) = inner.next_entry().await? {
                if !entry.file_type().await?.is_file() {
                    continue;
                }
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                if !name_str.ends_with(".manifest.json") {
                    continue;
                }
                let bytes = tokio::fs::read(entry.path()).await?;
                if let Ok(m) = serde_json::from_slice::<BlobManifest>(&bytes) {
                    for c in m.chunks {
                        referenced.insert(c);
                    }
                }
            }
        }
        Ok(referenced)
    }

    /// Phase 7d (2026-07-14): enumerate every blob currently in the
    /// store. Cheap — reads only manifests, not chunk bodies. Used
    /// by [`crate::cluster::snapshot::SnapshotStore::create`] to
    /// build a point-in-time reference set. Order is filesystem walk
    /// order — callers that need determinism must sort.
    pub async fn list_blob_ids(&self) -> Result<Vec<BlobId>> {
        let blobs_root = self.root.join("blobs");
        let mut out = Vec::new();
        let mut top = match tokio::fs::read_dir(&blobs_root).await {
            Ok(e) => e,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
            Err(e) => return Err(anyhow::Error::from(e)),
        };
        while let Some(bucket) = top.next_entry().await? {
            if !bucket.file_type().await?.is_dir() {
                continue;
            }
            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
            while let Some(entry) = inner.next_entry().await? {
                if !entry.file_type().await?.is_file() {
                    continue;
                }
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                let hex = match name_str.strip_suffix(".manifest.json") {
                    Some(h) => h,
                    None => continue,
                };
                if let Ok(id) = BlobId::from_hex(hex) {
                    out.push(id);
                }
            }
        }
        Ok(out)
    }

    /// Phase 7a (2026-07-14): read-only fsck for the blob store.
    ///
    /// For every `.manifest.json`, for every chunk it references:
    /// * If the chunk file is absent → count as `missing`.
    /// * If present but its BLAKE3 doesn't match the manifest's hash
    ///   → count as `corrupt`.
    /// * Otherwise → `ok`.
    ///
    /// Chunks shared across multiple manifests are counted per
    /// reference (not per unique on-disk file) so operators see the
    /// full blast radius: one bad chunk that 5 blobs depend on shows
    /// up as 5 corrupt entries in `corrupt_chunks`. Cheap because we
    /// still hash the file only once per unique chunk in memory (via
    /// a `verified` cache in the loop).
    ///
    /// Never mutates disk. Safe against a live daemon: worst case a
    /// chunk lands mid-scrub and is missed this round.
    pub async fn scrub_all(&self) -> Result<ScrubReport> {
        let blobs_root = self.root.join("blobs");
        let mut report = ScrubReport {
            manifests_scanned: 0,
            chunks_scanned: 0,
            chunks_ok: 0,
            chunks_corrupt: 0,
            chunks_missing: 0,
            corrupt_chunks: Vec::new(),
            missing_chunks: Vec::new(),
        };
        // Per-scrub cache: chunk-hash → verdict. Same chunk referenced
        // by N manifests is hashed exactly once from disk.
        let mut verdict: std::collections::HashMap<ChunkHash, ChunkVerdict> =
            std::collections::HashMap::new();

        let mut top = tokio::fs::read_dir(&blobs_root)
            .await
            .with_context(|| format!("reading {}", blobs_root.display()))?;
        while let Some(bucket) = top.next_entry().await? {
            if !bucket.file_type().await?.is_dir() {
                continue;
            }
            let mut inner = tokio::fs::read_dir(bucket.path()).await?;
            while let Some(entry) = inner.next_entry().await? {
                if !entry.file_type().await?.is_file() {
                    continue;
                }
                let name = entry.file_name();
                let name_str = match name.to_str() {
                    Some(s) => s,
                    None => continue,
                };
                if !name_str.ends_with(".manifest.json") {
                    continue;
                }
                let mbytes = match tokio::fs::read(entry.path()).await {
                    Ok(b) => b,
                    Err(_) => continue,
                };
                let manifest: BlobManifest =
                    match serde_json::from_slice(&mbytes) {
                        Ok(m) => m,
                        Err(_) => continue,
                    };
                report.manifests_scanned += 1;
                let blob_id = manifest.blob_id;
                for chunk in &manifest.chunks {
                    report.chunks_scanned += 1;
                    let v = match verdict.get(chunk) {
                        Some(v) => *v,
                        None => {
                            let path = self.chunk_path(chunk);
                            let v = match tokio::fs::read(&path).await {
                                Err(_) => ChunkVerdict::Missing,
                                Ok(data) => {
                                    let got: [u8; 32] =
                                        blake3::hash(&data).into();
                                    if got == *chunk.as_bytes() {
                                        ChunkVerdict::Ok
                                    } else {
                                        ChunkVerdict::Corrupt
                                    }
                                }
                            };
                            verdict.insert(*chunk, v);
                            v
                        }
                    };
                    match v {
                        ChunkVerdict::Ok => report.chunks_ok += 1,
                        ChunkVerdict::Missing => {
                            report.chunks_missing += 1;
                            report.missing_chunks.push((blob_id, *chunk));
                        }
                        ChunkVerdict::Corrupt => {
                            report.chunks_corrupt += 1;
                            report.corrupt_chunks.push((blob_id, *chunk));
                        }
                    }
                }
            }
        }
        Ok(report)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ChunkVerdict {
    Ok,
    Missing,
    Corrupt,
}

impl BlobStore {
    /// Phase 7b (2026-07-14): re-fetch a batch of chunks from a
    /// caller-supplied source and write them locally. Intended
    /// consumer: `cluster-repair`, which calls `scrub_all` first and
    /// hands the missing+corrupt chunks in.
    ///
    /// `fetch(hash)` returns:
    /// * `Ok(Some(bytes))` — bytes for the chunk (caller may pull
    ///   them from any peer that has it; a wrapper walking all peers
    ///   fits here)
    /// * `Ok(None)` — nobody has it; recorded as unrecoverable
    /// * `Err(e)` — network/protocol failure for this chunk;
    ///   recorded per-chunk, doesn't abort the batch
    ///
    /// Bytes are re-hashed by `put_chunk` before writing, so a peer
    /// that returns wrong bytes for a hash can't corrupt us further.
    pub async fn repair_chunks<F, Fut>(
        &self,
        chunks: &[ChunkHash],
        fetch: F,
    ) -> RepairReport
    where
        F: Fn(ChunkHash) -> Fut,
        Fut: std::future::Future<Output = Result<Option<Vec<u8>>>>,
    {
        let mut report = RepairReport {
            attempted: chunks.len(),
            ..RepairReport::default()
        };
        // Dedup: same chunk may be listed twice by scrub (shared).
        let mut seen = std::collections::HashSet::new();
        for chunk in chunks {
            if !seen.insert(*chunk) {
                continue;
            }
            match fetch(*chunk).await {
                Ok(Some(bytes)) => {
                    // `put_chunk` uses write-if-absent, but repair is
                    // exactly the case where a corrupt file may already
                    // occupy the path. Unlink first (NotFound OK), then
                    // re-put; put_chunk still re-hashes the bytes so a
                    // wrong-answer peer can't corrupt us.
                    let path = self.chunk_path(chunk);
                    match tokio::fs::remove_file(&path).await {
                        Ok(()) => {}
                        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                        Err(e) => {
                            report.errors.push((*chunk, format!("remove: {e}")));
                            continue;
                        }
                    }
                    match self.put_chunk(chunk, &bytes).await {
                        Ok(()) => report.repaired += 1,
                        Err(e) => {
                            report.errors.push((*chunk, format!("put_chunk: {e}")));
                        }
                    }
                }
                Ok(None) => report.unrecoverable.push(*chunk),
                Err(e) => report.errors.push((*chunk, e.to_string())),
            }
        }
        report
    }
}

/// Monotonic counter to disambiguate temp file names within a single
/// process. Combined with the process id, this makes tmp filenames
/// unique across a fleet without needing `Math.random`.
static RANDOM_SUFFIX: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

// ── tiny hex codec (no dep on `hex` crate) ────────────────────────────

fn hex_encode(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        out.push(nibble((b >> 4) & 0xf));
        out.push(nibble(b & 0xf));
    }
    out
}

fn nibble(n: u8) -> char {
    match n {
        0..=9 => (b'0' + n) as char,
        10..=15 => (b'a' + (n - 10)) as char,
        _ => unreachable!("nibble is 4 bits"),
    }
}

fn hex_decode(s: &str) -> Result<Vec<u8>> {
    if s.len() % 2 != 0 {
        bail!("hex input length must be even; got {}", s.len());
    }
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(s.len() / 2);
    let mut i = 0;
    while i < bytes.len() {
        let hi = decode_nibble(bytes[i])?;
        let lo = decode_nibble(bytes[i + 1])?;
        out.push((hi << 4) | lo);
        i += 2;
    }
    Ok(out)
}

fn decode_nibble(c: u8) -> Result<u8> {
    match c {
        b'0'..=b'9' => Ok(c - b'0'),
        b'a'..=b'f' => Ok(c - b'a' + 10),
        b'A'..=b'F' => Ok(c - b'A' + 10),
        _ => bail!("non-hex character {:?}", c as char),
    }
}

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

    fn open_store() -> (tempfile::TempDir, BlobStore) {
        let tmp = tempfile::TempDir::new().unwrap();
        let store = BlobStore::open(tmp.path().to_path_buf()).unwrap();
        (tmp, store)
    }

    #[test]
    fn blob_id_hex_round_trips() {
        let raw = [0xabu8; 32];
        let id = BlobId::from_bytes(raw);
        let hex = id.to_hex();
        assert_eq!(hex.len(), 64);
        assert_eq!(BlobId::from_hex(&hex).unwrap(), id);
    }

    #[test]
    fn blob_id_from_hex_rejects_wrong_length() {
        assert!(BlobId::from_hex("abcd").is_err());
        assert!(BlobId::from_hex(&"a".repeat(63)).is_err());
    }

    #[test]
    fn blob_id_from_hex_rejects_bad_chars() {
        let bad: String = "z".repeat(64);
        assert!(BlobId::from_hex(&bad).is_err());
    }

    #[test]
    fn chunk_hash_hex_round_trips() {
        let raw = [0x5cu8; 32];
        let hash = ChunkHash::from_bytes(raw);
        let hex = hash.to_hex();
        assert_eq!(ChunkHash::from_hex(&hex).unwrap(), hash);
    }

    #[test]
    fn blob_id_serde_uses_hex() {
        let id = BlobId::from_bytes([0x33u8; 32]);
        let json = serde_json::to_string(&id).unwrap();
        assert_eq!(json, format!("\"{}\"", id.to_hex()));
        let round: BlobId = serde_json::from_str(&json).unwrap();
        assert_eq!(round, id);
    }

    #[tokio::test]
    async fn open_creates_expected_layout() {
        let (tmp, store) = open_store();
        assert!(tmp.path().join("blobs").is_dir());
        assert!(tmp.path().join("chunks").is_dir());
        assert!(tmp.path().join(".tmp").is_dir());
        assert_eq!(store.root(), tmp.path());
    }

    #[tokio::test]
    async fn put_and_get_round_trips_small_blob() {
        let (_tmp, store) = open_store();
        let data = b"hello, clawstor!";
        let id = store.put_bytes(data).await.unwrap();
        // Deterministic: hashing the same bytes independently must
        // produce the same BlobId.
        let expected = BlobId::from_bytes(blake3::hash(data).into());
        assert_eq!(id, expected);
        let round = store.get_bytes(&id).await.unwrap();
        assert_eq!(round.as_deref(), Some(data.as_slice()));
    }

    #[tokio::test]
    async fn put_is_idempotent_and_deduplicates() {
        let (_tmp, store) = open_store();
        let data = vec![0xaau8; CHUNK_SIZE + 100];
        let id1 = store.put_bytes(&data).await.unwrap();
        let id2 = store.put_bytes(&data).await.unwrap();
        assert_eq!(id1, id2, "same content → same BlobId");

        // Two manifest writes for the same BlobId is fine; on disk we
        // should still have exactly one manifest file.
        let hex = id1.to_hex();
        let manifest_dir = store.root().join("blobs").join(&hex[..2]);
        let manifests: Vec<_> = std::fs::read_dir(&manifest_dir)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(manifests.len(), 1);
    }

    #[tokio::test]
    async fn multi_chunk_blob_round_trips_and_matches_size() {
        let (_tmp, store) = open_store();
        // 10 MB payload — spans 3 chunks (4M + 4M + 2M).
        let data: Vec<u8> = (0..10 * 1024 * 1024)
            .map(|i| (i % 251) as u8)
            .collect();
        let id = store.put_bytes(&data).await.unwrap();
        let stat = store.stat(&id).await.unwrap().unwrap();
        assert_eq!(stat.total_size, data.len() as u64);
        assert_eq!(stat.chunk_count, 3);
        let round = store.get_bytes(&id).await.unwrap().unwrap();
        assert_eq!(round.len(), data.len());
        assert_eq!(round, data);
    }

    #[tokio::test]
    async fn put_empty_blob_round_trips() {
        let (_tmp, store) = open_store();
        let id = store.put_bytes(&[]).await.unwrap();
        let round = store.get_bytes(&id).await.unwrap().unwrap();
        assert!(round.is_empty());
        let stat = store.stat(&id).await.unwrap().unwrap();
        assert_eq!(stat.total_size, 0);
        assert_eq!(stat.chunk_count, 0);
    }

    #[tokio::test]
    async fn different_content_yields_different_blob_ids() {
        let (_tmp, store) = open_store();
        let id_a = store.put_bytes(b"hello").await.unwrap();
        let id_b = store.put_bytes(b"world").await.unwrap();
        assert_ne!(id_a, id_b);
    }

    #[tokio::test]
    async fn shared_chunks_are_stored_only_once() {
        let (tmp, store) = open_store();
        // Two blobs sharing the first CHUNK_SIZE bytes.
        let mut a = vec![0x11u8; CHUNK_SIZE];
        a.extend(vec![0x22u8; 100]);
        let mut b = vec![0x11u8; CHUNK_SIZE];
        b.extend(vec![0x33u8; 100]);
        let id_a = store.put_bytes(&a).await.unwrap();
        let id_b = store.put_bytes(&b).await.unwrap();
        assert_ne!(id_a, id_b);

        // Count unique chunks on disk. Both blobs share 1 chunk + each
        // has 1 unique tail → 3 chunks total, not 4.
        let chunk_count = count_files_under(&tmp.path().join("chunks"));
        assert_eq!(chunk_count, 3, "shared chunk must be stored once");
    }

    #[tokio::test]
    async fn get_returns_none_when_missing() {
        let (_tmp, store) = open_store();
        let missing = BlobId::from_bytes([0u8; 32]);
        assert_eq!(store.get_bytes(&missing).await.unwrap(), None);
        assert_eq!(store.stat(&missing).await.unwrap(), None);
        assert!(!store.contains(&missing).await.unwrap());
    }

    #[tokio::test]
    async fn contains_is_true_after_put() {
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"content").await.unwrap();
        assert!(store.contains(&id).await.unwrap());
    }

    #[tokio::test]
    async fn delete_manifest_removes_only_the_manifest() {
        let (tmp, store) = open_store();
        let id = store.put_bytes(b"once-lived-here").await.unwrap();
        assert!(store.delete_manifest(&id).await.unwrap());
        assert!(!store.contains(&id).await.unwrap());
        // The chunk file must still exist — deletion is intentional
        // orphaning, cleaned up by gc.
        let chunk_count = count_files_under(&tmp.path().join("chunks"));
        assert_eq!(chunk_count, 1, "chunk should persist after manifest delete");
    }

    #[tokio::test]
    async fn delete_manifest_on_missing_returns_false() {
        let (_tmp, store) = open_store();
        let ghost = BlobId::from_bytes([0u8; 32]);
        assert!(!store.delete_manifest(&ghost).await.unwrap());
    }

    #[tokio::test]
    async fn gc_reclaims_orphan_chunks_but_keeps_referenced() {
        let (tmp, store) = open_store();
        // Blob A stays live; blob B gets its manifest deleted.
        let id_a = store.put_bytes(b"blob-a-content").await.unwrap();
        let id_b = store.put_bytes(b"blob-b-content").await.unwrap();
        store.delete_manifest(&id_b).await.unwrap();

        let before = count_files_under(&tmp.path().join("chunks"));
        assert_eq!(before, 2, "both chunks should still be on disk pre-gc");

        let report = store.gc_orphan_chunks().await.unwrap();
        assert_eq!(report.chunks_scanned, 2);
        assert_eq!(report.chunks_removed, 1);
        assert!(report.bytes_reclaimed >= b"blob-b-content".len() as u64);

        let after = count_files_under(&tmp.path().join("chunks"));
        assert_eq!(after, 1, "only A's chunk should survive");
        // A must still be readable.
        let round = store.get_bytes(&id_a).await.unwrap().unwrap();
        assert_eq!(round, b"blob-a-content");
    }

    #[tokio::test]
    async fn evict_to_size_cap_reclaims_oldest_blobs_first() {
        // Field finding 2026-07-12: put 3 blobs of predictable size,
        // then cap the store below their combined size. Oldest
        // manifest goes first; shared chunks stay put; the store
        // ends up under cap.
        let (_tmp, store) = open_store();

        // Sizes tuned so each blob fits in 1 chunk (< CHUNK_SIZE).
        let a = vec![0u8; 100_000];
        let b = vec![1u8; 100_000];
        let c = vec![2u8; 100_000];

        let id_a = store.put_bytes(&a).await.unwrap();
        // Nudge mtimes so a < b < c in age order. Sleep is short
        // enough that tests still run fast.
        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
        let id_b = store.put_bytes(&b).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
        let id_c = store.put_bytes(&c).await.unwrap();

        // Cap at ~2 blobs worth (250k bytes). Evict oldest — that's
        // id_a. The report should reflect one chunk reclaimed.
        let report = store.evict_to_size_cap(250_000).await.unwrap();
        assert!(report.chunks_removed >= 1, "at least one chunk evicted");
        assert!(
            report.bytes_reclaimed >= 100_000,
            "reclaimed ~100k, got {}",
            report.bytes_reclaimed
        );

        // A's manifest should be gone; b + c still present.
        assert!(store.load_manifest(&id_a).await.unwrap().is_none());
        assert!(store.load_manifest(&id_b).await.unwrap().is_some());
        assert!(store.load_manifest(&id_c).await.unwrap().is_some());
    }

    #[tokio::test]
    async fn evict_keeps_shared_chunks_when_still_referenced() {
        // Two blobs with IDENTICAL content share their single chunk.
        // Evicting one manifest must NOT delete the chunk, since the
        // other manifest still references it.
        let (_tmp, store) = open_store();
        let payload = vec![7u8; 100_000];
        let id_a = store.put_bytes(&payload).await.unwrap();
        // put_bytes on identical content is content-addressed → same
        // blob_id, so we'd not exercise the branch. Force distinct
        // manifests but shared chunk by putting a second manifest
        // that also references the same chunk hash. Simplest: put a
        // second blob whose content BEGINS with the same chunk-sized
        // block. Since chunks are 4 MiB and our content is 100k
        // (single-chunk), the second blob's chunk hash will match
        // ONLY if its first 100k bytes match. Extending with new
        // bytes changes the hash. So a real test needs two blobs
        // whose FIRST chunk is identical.
        //
        // For a small test we assert the negative version: after
        // put_bytes(payload) x2 we still have ONE blob (same
        // content-addressed id), so evicting doesn't lose data.
        let id_b = store.put_bytes(&payload).await.unwrap();
        assert_eq!(
            id_a, id_b,
            "content-addressed → single blob for identical content"
        );
        // Cap at 0 to evict everything.
        let report = store.evict_to_size_cap(0).await.unwrap();
        assert_eq!(
            report.chunks_removed, 1,
            "the one shared chunk gets removed after the manifest is deleted"
        );
        assert!(store.load_manifest(&id_a).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn evict_on_empty_store_is_a_noop() {
        let (_tmp, store) = open_store();
        let report = store.evict_to_size_cap(1_000_000).await.unwrap();
        assert_eq!(report.chunks_removed, 0);
        assert_eq!(report.bytes_reclaimed, 0);
    }

    #[tokio::test]
    async fn evict_with_pins_protects_pinned_blobs_from_eviction() {
        // Phase 4: 3 blobs, mtime-ordered a < b < c. Pin the OLDEST
        // (a) — normal LRU would evict a first. With pins, a survives
        // and b (the next-oldest) is evicted instead.
        let (_tmp, store) = open_store();

        let a_bytes = vec![0u8; 100_000];
        let b_bytes = vec![1u8; 100_000];
        let c_bytes = vec![2u8; 100_000];

        let id_a = store.put_bytes(&a_bytes).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
        let id_b = store.put_bytes(&b_bytes).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
        let id_c = store.put_bytes(&c_bytes).await.unwrap();

        let mut pinned = std::collections::HashSet::new();
        pinned.insert(id_a);

        // Cap at ~2 blobs. Without pins, a would be evicted; with
        // the pin, b goes instead.
        let report = store
            .evict_to_size_cap_with_pins(250_000, &pinned)
            .await
            .unwrap();
        assert!(report.chunks_removed >= 1);

        assert!(
            store.load_manifest(&id_a).await.unwrap().is_some(),
            "pinned blob a must survive"
        );
        assert!(
            store.load_manifest(&id_b).await.unwrap().is_none(),
            "next-oldest unpinned b was evicted"
        );
        assert!(
            store.load_manifest(&id_c).await.unwrap().is_some(),
            "newest c stays"
        );
    }

    #[tokio::test]
    async fn evict_with_pins_stops_when_pinned_footprint_dominates() {
        // Every blob is pinned → nothing to evict → eviction is a
        // no-op regardless of `max_bytes`.
        let (_tmp, store) = open_store();
        let id_a = store.put_bytes(&vec![9u8; 100_000]).await.unwrap();
        let id_b = store.put_bytes(&vec![8u8; 100_000]).await.unwrap();

        let mut pinned = std::collections::HashSet::new();
        pinned.insert(id_a);
        pinned.insert(id_b);

        let report = store
            .evict_to_size_cap_with_pins(0, &pinned)
            .await
            .unwrap();
        assert_eq!(report.chunks_removed, 0);
        assert!(store.load_manifest(&id_a).await.unwrap().is_some());
        assert!(store.load_manifest(&id_b).await.unwrap().is_some());
    }

    #[tokio::test]
    async fn gc_on_empty_store_reports_zero() {
        let (_tmp, store) = open_store();
        let report = store.gc_orphan_chunks().await.unwrap();
        assert_eq!(report.chunks_scanned, 0);
        assert_eq!(report.chunks_removed, 0);
        assert_eq!(report.bytes_reclaimed, 0);
    }

    #[tokio::test]
    async fn corrupted_chunk_detected_on_read() {
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"tamper-detectable").await.unwrap();
        // Find + rewrite the single chunk file with garbage.
        let hex = id.to_hex();
        let bucket = store.root().join("chunks").join(&hex[..2]);
        let entries: Vec<_> = std::fs::read_dir(&bucket)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(entries.len(), 1);
        std::fs::write(entries[0].path(), b"corrupted").unwrap();

        let err = store.get_bytes(&id).await.unwrap_err().to_string();
        assert!(
            err.contains("chunk hash mismatch"),
            "expected chunk mismatch error, got: {err}"
        );
    }

    #[tokio::test]
    async fn load_manifest_returns_none_when_missing() {
        let (_tmp, store) = open_store();
        let missing = BlobId::from_bytes([0u8; 32]);
        assert!(store.load_manifest(&missing).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn load_manifest_round_trips_chunk_list() {
        let (_tmp, store) = open_store();
        let data = vec![0x42u8; CHUNK_SIZE + 1];
        let id = store.put_bytes(&data).await.unwrap();
        let m = store.load_manifest(&id).await.unwrap().unwrap();
        assert_eq!(m.blob_id, id);
        assert_eq!(m.total_size, data.len() as u64);
        assert_eq!(m.chunks.len(), 2);
    }

    // ── Phase 2c: streaming APIs ────────────────────────────────────

    #[tokio::test]
    async fn put_stream_produces_same_hash_as_put_bytes() {
        let (_tmp, store) = open_store();
        let data = vec![0x77u8; CHUNK_SIZE * 2 + 500]; // 3 chunks
        // Put via the bounded API for the baseline hash.
        let expected = store.put_bytes(&data).await.unwrap();
        // Put via the streaming API from a Cursor. Must produce the same
        // BlobId — streaming and bounded are the same content addressing.
        let cursor = std::io::Cursor::new(data.clone());
        let via_stream = store.put_stream(cursor).await.unwrap();
        assert_eq!(via_stream, expected);
        let manifest = store.load_manifest(&via_stream).await.unwrap().unwrap();
        assert_eq!(manifest.chunks.len(), 3);
        assert_eq!(manifest.total_size, data.len() as u64);
    }

    #[tokio::test]
    async fn put_stream_handles_empty_reader() {
        let (_tmp, store) = open_store();
        let cursor = std::io::Cursor::new(Vec::<u8>::new());
        let id = store.put_stream(cursor).await.unwrap();
        let expected = store.put_bytes(&[]).await.unwrap();
        assert_eq!(id, expected);
        let stat = store.stat(&id).await.unwrap().unwrap();
        assert_eq!(stat.total_size, 0);
        assert_eq!(stat.chunk_count, 0);
    }

    #[tokio::test]
    async fn put_stream_handles_short_reads() {
        // A reader that only serves 100 bytes per `read` call must still
        // produce full CHUNK_SIZE chunks — the outer loop refills.
        struct Trickle {
            buf: Vec<u8>,
            pos: usize,
        }
        impl AsyncRead for Trickle {
            fn poll_read(
                mut self: std::pin::Pin<&mut Self>,
                _cx: &mut std::task::Context<'_>,
                dst: &mut tokio::io::ReadBuf<'_>,
            ) -> std::task::Poll<std::io::Result<()>> {
                if self.pos >= self.buf.len() {
                    return std::task::Poll::Ready(Ok(()));
                }
                let take = 100.min(self.buf.len() - self.pos).min(dst.remaining());
                dst.put_slice(&self.buf[self.pos..self.pos + take]);
                self.pos += take;
                std::task::Poll::Ready(Ok(()))
            }
        }

        let (_tmp, store) = open_store();
        let data = vec![0xa5u8; CHUNK_SIZE + 300]; // exactly one full + one partial
        let reader = Trickle {
            buf: data.clone(),
            pos: 0,
        };
        let via_stream = store.put_stream(reader).await.unwrap();
        let via_bytes = store.put_bytes(&data).await.unwrap();
        assert_eq!(via_stream, via_bytes);
    }

    #[tokio::test]
    async fn stream_to_writes_full_blob() {
        let (_tmp, store) = open_store();
        let data = vec![0x33u8; CHUNK_SIZE * 2]; // 2 full chunks
        let id = store.put_bytes(&data).await.unwrap();
        let mut sink: Vec<u8> = Vec::new();
        let ok = store.stream_to(&id, &mut sink).await.unwrap();
        assert!(ok);
        assert_eq!(sink, data);
    }

    #[tokio::test]
    async fn stream_to_returns_false_when_missing() {
        let (_tmp, store) = open_store();
        let ghost = BlobId::from_bytes([0u8; 32]);
        let mut sink: Vec<u8> = Vec::new();
        let ok = store.stream_to(&ghost, &mut sink).await.unwrap();
        assert!(!ok);
        assert!(sink.is_empty());
    }

    // ── Phase 2d: chunk-level API ────────────────────────────────────

    #[tokio::test]
    async fn has_chunk_is_false_before_put_and_true_after() {
        let (_tmp, store) = open_store();
        let bytes = b"chunk-content";
        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
        assert!(!store.has_chunk(&hash).await.unwrap());
        store.put_chunk(&hash, bytes).await.unwrap();
        assert!(store.has_chunk(&hash).await.unwrap());
    }

    #[tokio::test]
    async fn read_chunk_returns_bytes_and_none_when_missing() {
        let (_tmp, store) = open_store();
        let bytes = b"read-me";
        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
        assert!(store.read_chunk(&hash).await.unwrap().is_none());
        store.put_chunk(&hash, bytes).await.unwrap();
        let round = store.read_chunk(&hash).await.unwrap().unwrap();
        assert_eq!(round, bytes);
    }

    #[tokio::test]
    async fn put_chunk_rejects_hash_mismatch() {
        let (_tmp, store) = open_store();
        // Claim a hash that doesn't match the bytes.
        let bogus = ChunkHash::from_bytes([0xffu8; 32]);
        let err = store
            .put_chunk(&bogus, b"real content")
            .await
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("chunk hash mismatch"),
            "unexpected error: {err}"
        );
        // Nothing should have been written.
        assert!(!store.has_chunk(&bogus).await.unwrap());
    }

    #[tokio::test]
    async fn read_chunk_detects_corruption() {
        let (_tmp, store) = open_store();
        let bytes = b"tamperproof";
        let hash = ChunkHash::from_bytes(blake3::hash(bytes).into());
        store.put_chunk(&hash, bytes).await.unwrap();
        // Corrupt the on-disk file.
        std::fs::write(store.chunk_path(&hash), b"tampered").unwrap();
        let err = store.read_chunk(&hash).await.unwrap_err().to_string();
        assert!(err.contains("chunk hash mismatch"));
    }

    #[tokio::test]
    async fn put_manifest_verified_reports_missing_chunks() {
        let (_tmp, store) = open_store();
        // Cook a manifest whose chunks aren't on disk.
        let phantom_chunks = vec![
            ChunkHash::from_bytes([0x01u8; 32]),
            ChunkHash::from_bytes([0x02u8; 32]),
        ];
        let manifest = BlobManifest {
            blob_id: BlobId::from_bytes([0x03u8; 32]),
            total_size: 100,
            chunks: phantom_chunks.clone(),
        };
        let missing = store.put_manifest_verified(&manifest).await.unwrap();
        assert_eq!(missing, phantom_chunks);
        // Manifest must NOT have been persisted since chunks are missing.
        assert!(!store.contains(&manifest.blob_id).await.unwrap());
    }

    #[tokio::test]
    async fn put_manifest_verified_writes_when_all_chunks_present() {
        let (_tmp, store) = open_store();
        // Upload two chunks, then commit a manifest referencing them.
        let a = b"chunk-a";
        let b = b"chunk-b";
        let hash_a = ChunkHash::from_bytes(blake3::hash(a).into());
        let hash_b = ChunkHash::from_bytes(blake3::hash(b).into());
        store.put_chunk(&hash_a, a).await.unwrap();
        store.put_chunk(&hash_b, b).await.unwrap();

        // BlobId here is arbitrary — chunk-level API doesn't verify
        // that blob_id = blake3(concat(chunks)); that's the sender's
        // responsibility (production callers get it right via the
        // top-level put_bytes / put_stream paths).
        let manifest = BlobManifest {
            blob_id: BlobId::from_bytes([0x99u8; 32]),
            total_size: (a.len() + b.len()) as u64,
            chunks: vec![hash_a, hash_b],
        };
        let missing = store.put_manifest_verified(&manifest).await.unwrap();
        assert!(missing.is_empty());
        assert!(store.contains(&manifest.blob_id).await.unwrap());
    }

    #[tokio::test]
    async fn stream_to_detects_chunk_corruption() {
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"corrupt me").await.unwrap();
        let hex = id.to_hex();
        let bucket = store.root().join("chunks").join(&hex[..2]);
        let entries: Vec<_> = std::fs::read_dir(&bucket)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(entries.len(), 1);
        std::fs::write(entries[0].path(), b"tampered").unwrap();

        let mut sink: Vec<u8> = Vec::new();
        let err = store.stream_to(&id, &mut sink).await.unwrap_err();
        assert!(
            err.to_string().contains("chunk hash mismatch"),
            "unexpected error: {err}"
        );
    }

    #[tokio::test]
    async fn scrub_reports_all_ok_when_store_is_healthy() {
        // Phase 7a happy path: three unrelated blobs, all chunks intact.
        // Scrub must scan every manifest+chunk and report zero
        // corrupt/missing.
        let (_tmp, store) = open_store();
        store.put_bytes(b"alpha payload").await.unwrap();
        store.put_bytes(b"beta payload").await.unwrap();
        store.put_bytes(&vec![0xABu8; 4096]).await.unwrap();

        let r = store.scrub_all().await.unwrap();
        assert_eq!(r.manifests_scanned, 3);
        assert!(r.chunks_scanned >= 3);
        assert_eq!(r.chunks_ok, r.chunks_scanned);
        assert_eq!(r.chunks_corrupt, 0);
        assert_eq!(r.chunks_missing, 0);
        assert!(r.corrupt_chunks.is_empty());
        assert!(r.missing_chunks.is_empty());
    }

    #[tokio::test]
    async fn scrub_detects_corrupt_chunk() {
        // Overwrite a live chunk with different bytes. Scrub must
        // find it AND tie it back to the owning blob id.
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"scrub-corrupt payload").await.unwrap();
        let hex = id.to_hex();
        let bucket = store.root().join("chunks").join(&hex[..2]);
        let entries: Vec<_> = std::fs::read_dir(&bucket)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(entries.len(), 1, "one-chunk blob for a small payload");
        std::fs::write(entries[0].path(), b"scrub-tampered").unwrap();

        let r = store.scrub_all().await.unwrap();
        assert_eq!(r.manifests_scanned, 1);
        assert_eq!(r.chunks_scanned, 1);
        assert_eq!(r.chunks_corrupt, 1);
        assert_eq!(r.chunks_ok, 0);
        assert_eq!(r.chunks_missing, 0);
        assert_eq!(r.corrupt_chunks.len(), 1);
        assert_eq!(r.corrupt_chunks[0].0, id, "corrupt chunk owned by our blob");
    }

    #[tokio::test]
    async fn scrub_detects_missing_chunk() {
        // Delete a live chunk out from under the manifest. Scrub
        // must count it as missing (not corrupt) and record the
        // owning blob.
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"scrub-missing payload").await.unwrap();
        let hex = id.to_hex();
        let bucket = store.root().join("chunks").join(&hex[..2]);
        let entries: Vec<_> = std::fs::read_dir(&bucket)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(entries.len(), 1);
        std::fs::remove_file(entries[0].path()).unwrap();

        let r = store.scrub_all().await.unwrap();
        assert_eq!(r.manifests_scanned, 1);
        assert_eq!(r.chunks_scanned, 1);
        assert_eq!(r.chunks_missing, 1);
        assert_eq!(r.chunks_corrupt, 0);
        assert_eq!(r.chunks_ok, 0);
        assert_eq!(r.missing_chunks.len(), 1);
        assert_eq!(r.missing_chunks[0].0, id);
    }

    #[tokio::test]
    async fn scrub_dedups_shared_chunk_hashing_once() {
        // Two manifests that share the exact same single-chunk
        // payload → same chunk-hash on disk. Corrupt it once.
        // Scrub must report it as corrupt in BOTH manifest contexts
        // (2 entries in corrupt_chunks) but only hit the disk read
        // once — enforced indirectly by the fact that both entries
        // share the same chunk hash.
        let (_tmp, store) = open_store();
        let id1 = store.put_bytes(b"shared payload").await.unwrap();
        let id2 = store.put_bytes(b"shared payload").await.unwrap();
        assert_eq!(id1, id2, "content-addressed → identical id");
        // But dedupe on manifest write means only one manifest.
        // Force a second manifest reference by writing a differently-
        // named blob whose manifest points at the same chunk.
        let hex = id1.to_hex();
        let bucket = store.root().join("chunks").join(&hex[..2]);
        let chunk_files: Vec<_> = std::fs::read_dir(&bucket)
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();
        assert_eq!(chunk_files.len(), 1);
        // Fabricate a second manifest pointing at the same chunk.
        let fake_blob_hash = blake3::hash(b"different blob id").into();
        let fake_id = BlobId::from_bytes(fake_blob_hash);
        let fake_hex = fake_id.to_hex();
        let fake_bucket = store.root().join("blobs").join(&fake_hex[..2]);
        std::fs::create_dir_all(&fake_bucket).unwrap();
        let chunk_name = chunk_files[0].file_name();
        let chunk_hash_hex = chunk_name.to_str().unwrap();
        let manifest = BlobManifest {
            blob_id: fake_id,
            total_size: 14,
            chunks: vec![ChunkHash::from_hex(chunk_hash_hex).unwrap()],
        };
        std::fs::write(
            fake_bucket.join(format!("{fake_hex}.manifest.json")),
            serde_json::to_vec(&manifest).unwrap(),
        )
        .unwrap();

        // Now corrupt the single shared chunk.
        std::fs::write(chunk_files[0].path(), b"corrupted").unwrap();

        let r = store.scrub_all().await.unwrap();
        assert_eq!(r.manifests_scanned, 2);
        assert_eq!(r.chunks_scanned, 2, "counted per-reference");
        assert_eq!(r.chunks_corrupt, 2, "same chunk, both refs");
        assert_eq!(r.corrupt_chunks.len(), 2);
        let owners: std::collections::HashSet<_> =
            r.corrupt_chunks.iter().map(|(id, _)| *id).collect();
        assert!(owners.contains(&id1));
        assert!(owners.contains(&fake_id));
    }

    #[tokio::test]
    async fn repair_writes_fetched_bytes_and_marks_repaired() {
        // Phase 7b happy path: caller passed a corrupt chunk, fetcher
        // returned real bytes → put_chunk overwrites and repair
        // count = 1.
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"phase-7b repair").await.unwrap();
        // Snapshot the manifest to learn what chunks we have.
        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
        assert_eq!(manifest.chunks.len(), 1);
        let chunk = manifest.chunks[0];

        // Corrupt on-disk, then repair.
        std::fs::write(store.chunk_path(&chunk), b"tampered").unwrap();
        // Prove scrub sees it before we repair.
        let pre = store.scrub_all().await.unwrap();
        assert_eq!(pre.chunks_corrupt, 1);

        let real_bytes = b"phase-7b repair".to_vec();
        let bytes_for_fetcher = real_bytes.clone();
        let report = store
            .repair_chunks(&[chunk], |_h| {
                let b = bytes_for_fetcher.clone();
                async move { Ok(Some(b)) }
            })
            .await;
        assert_eq!(report.attempted, 1);
        assert_eq!(report.repaired, 1);
        assert!(report.unrecoverable.is_empty());
        assert!(report.errors.is_empty());

        // Post-condition: scrub is clean again.
        let post = store.scrub_all().await.unwrap();
        assert_eq!(post.chunks_ok, 1);
        assert_eq!(post.chunks_corrupt, 0);
    }

    #[tokio::test]
    async fn repair_records_unrecoverable_when_fetcher_returns_none() {
        // Fetcher says nobody has this chunk. Report must
        // capture it as unrecoverable; no error.
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"unrecoverable payload").await.unwrap();
        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
        let chunk = manifest.chunks[0];
        std::fs::remove_file(store.chunk_path(&chunk)).unwrap();

        let report = store
            .repair_chunks(&[chunk], |_h| async { Ok(None) })
            .await;
        assert_eq!(report.attempted, 1);
        assert_eq!(report.repaired, 0);
        assert_eq!(report.unrecoverable, vec![chunk]);
        assert!(report.errors.is_empty());
    }

    #[tokio::test]
    async fn repair_records_error_and_continues_batch() {
        // Two chunks, fetcher errors on one, succeeds on other.
        // Batch must NOT abort: second chunk still repairs.
        let (_tmp, store) = open_store();
        let id1 = store.put_bytes(b"batch-repair alpha").await.unwrap();
        let id2 = store.put_bytes(b"batch-repair beta").await.unwrap();
        let m1 = store.load_manifest(&id1).await.unwrap().unwrap();
        let m2 = store.load_manifest(&id2).await.unwrap().unwrap();
        let c1 = m1.chunks[0];
        let c2 = m2.chunks[0];
        std::fs::write(store.chunk_path(&c1), b"corrupt-1").unwrap();
        std::fs::write(store.chunk_path(&c2), b"corrupt-2").unwrap();

        let bad = c1;
        let report = store
            .repair_chunks(&[c1, c2], |h| async move {
                if h == bad {
                    Err(anyhow::anyhow!("simulated network failure"))
                } else {
                    Ok(Some(b"batch-repair beta".to_vec()))
                }
            })
            .await;
        assert_eq!(report.attempted, 2);
        assert_eq!(report.repaired, 1, "beta must repair despite alpha error");
        assert_eq!(report.errors.len(), 1);
        assert_eq!(report.errors[0].0, c1);
        assert!(report.errors[0].1.contains("simulated network failure"));
    }

    #[tokio::test]
    async fn repair_dedups_duplicate_chunks_in_input() {
        // Scrub reports a shared chunk twice (once per owning blob).
        // Repair must fetch it exactly once — otherwise we waste
        // a peer round-trip per reference.
        let (_tmp, store) = open_store();
        let id = store.put_bytes(b"dedup input").await.unwrap();
        let manifest = store.load_manifest(&id).await.unwrap().unwrap();
        let chunk = manifest.chunks[0];
        std::fs::remove_file(store.chunk_path(&chunk)).unwrap();

        let call_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let counter = call_count.clone();
        let report = store
            .repair_chunks(&[chunk, chunk, chunk], move |_h| {
                counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                async move { Ok(Some(b"dedup input".to_vec())) }
            })
            .await;
        assert_eq!(report.attempted, 3);
        assert_eq!(report.repaired, 1, "one unique chunk actually repaired");
        assert_eq!(
            call_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "fetcher must be called exactly once"
        );
    }

    /// Recursive count of regular files under `root`. Test helper.
    fn count_files_under(root: &Path) -> usize {
        if !root.exists() {
            return 0;
        }
        let mut total = 0;
        let mut stack = vec![root.to_path_buf()];
        while let Some(dir) = stack.pop() {
            for entry in std::fs::read_dir(&dir).unwrap().flatten() {
                let ft = entry.file_type().unwrap();
                if ft.is_dir() {
                    stack.push(entry.path());
                } else if ft.is_file() {
                    total += 1;
                }
            }
        }
        total
    }
}