fxcp_core/

cas_store.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wiramu Pauling <aenertia@aenertia.net>
//
// fxcp-core/src/cas_store.rs  --  Content-addressable chunk store for filesystem-agnostic targets

//! Content-addressable storage (CAS) directory for foxing replication targets.
//!
//! Stores files as BLAKE3-addressed chunks in a sharded directory layout.
//! Works on any filesystem (exFAT, FAT32, NTFS, NFS)  --  no xattrs, permissions,
//! or symlinks required. All metadata is in the append-only manifest.
//!
//! Layout:
//! ```text
//! .foxing_cas/
//!   config.json        --  store parameters (chunk sizes, compression)
//!   manifest.jsonl     --  append-only operation log (one JSON line per op)
//!   chunks/
//!     a1/b2c3d4...    --  chunk files named by BLAKE3 hash
//!     tmp/            --  staging area for crash-safe writes
//! ```

use std::collections::{HashMap, HashSet};
use std::io::{self, BufRead, Write};
use std::path::{Path, PathBuf};
use std::sync::RwLock;
use serde::{Serialize, Deserialize};
use tracing::{info, warn, debug};

use crate::chunk_delta::{ChunkRef, compute_chunk_delta};
use crate::chunker::GearChunker;
use crate::error::FxcpError;

const CAS_DIR: &str = ".foxing_cas";
const CHUNKS_DIR: &str = "chunks";
const TMP_DIR: &str = "chunks/tmp";
const MANIFEST_FILE: &str = "manifest.jsonl";
const CONFIG_FILE: &str = "config.json";

// -----------------------------------------------------------------------
// Types
// -----------------------------------------------------------------------

/// Store configuration (persisted as config.json).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CasConfig {
    pub version: u32,
    pub chunk_min: usize,
    pub chunk_avg: usize,
    pub chunk_max: usize,
    pub compression: String,
    /// Normalization level for gear-hash chunk-size distribution (0..=2).
    #[serde(default)]
    pub normalization_level: u8,
}

impl Default for CasConfig {
    fn default() -> Self {
        Self {
            version: 1,
            chunk_min: crate::chunker::DEFAULT_CHUNK_MIN,
            chunk_avg: crate::chunker::DEFAULT_CHUNK_AVG,
            chunk_max: crate::chunker::DEFAULT_CHUNK_MAX,
            compression: "zstd".into(),
            normalization_level: 0,
        }
    }
}

/// File metadata stored in manifest entries.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileMeta {
    pub size: u64,
    pub mode: u32,
    pub mtime: i64,
    pub uid: u32,
    pub gid: u32,
}

/// A single operation in the manifest log.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "op")]
pub enum ManifestOp {
    #[serde(rename = "put")]
    Put {
        path: String,
        #[serde(flatten)]
        meta: FileMeta,
        blake3: String,
        chunks: Vec<String>,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        chunk_sizes: Vec<u64>,
    },
    #[serde(rename = "delete")]
    Delete { path: String },
    #[serde(rename = "rename")]
    Rename { from: String, to: String },
    #[serde(rename = "snapshot")]
    Snapshot { name: String },
}

/// Current state of a file in the store (after replaying manifest).
#[derive(Debug, Clone)]
pub struct FileEntry {
    pub path: String,
    pub meta: FileMeta,
    pub blake3: String,
    pub chunks: Vec<String>,
    pub chunk_sizes: Vec<u64>,
}

/// Statistics from a store_file operation.
#[derive(Debug, Default)]
pub struct CasStoreStats {
    pub chunks_written: u64,
    pub chunks_deduped: u64,
    pub bytes_written: u64,
    pub bytes_deduped: u64,
}

/// Statistics from a compact operation.
#[derive(Debug, Default)]
pub struct CompactStats {
    pub files_kept: u64,
    pub ops_removed: u64,
    pub chunks_removed: u64,
    pub bytes_freed: u64,
}

/// Statistics from a batched store operation (amortized fsync).
#[derive(Debug, Clone)]
pub struct BatchStats {
    /// Number of files stored in the batch.
    pub files: usize,
    /// Total fsync calls issued (chunk dirs + manifest). Much less than 2*N
    /// for N files because chunk directory fsyncs are deduplicated across the
    /// batch and the manifest receives a single fsync for all entries.
    pub fsyncs: usize,
}

/// Statistics from a delta store operation.
#[derive(Debug)]
pub struct DeltaStoreStats {
    pub chunks_added: u64,
    pub chunks_reused: u64,
    pub bytes_uploaded: u64,
    pub bytes_skipped: u64,
    pub efficiency: f64,
}

// -----------------------------------------------------------------------
// CasStore
// -----------------------------------------------------------------------

/// Content-addressable chunk store.
pub struct CasStore {
    root: PathBuf,
    config: CasConfig,
    /// Serializes `compact()` (write) against `store_file()` (read) to prevent
    /// chunk loss when a concurrent store writes chunks between the referenced-set
    /// computation and the orphan-deletion pass in compact.
    compact_lock: RwLock<()>,
}

impl std::fmt::Debug for CasStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CasStore")
            .field("root", &self.root)
            .field("config", &self.config)
            .finish()
    }
}

impl CasStore {
    /// Open or create a CAS store at `{path}/.foxing_cas/`.
    pub fn open(path: &Path) -> crate::Result<Self> {
        let root = path.join(CAS_DIR);
        std::fs::create_dir_all(root.join(TMP_DIR))?;

        // Load or create config
        let config_path = root.join(CONFIG_FILE);
        let config = if config_path.exists() {
            let data = std::fs::read_to_string(&config_path)?;
            serde_json::from_str(&data).unwrap_or_default()
        } else {
            let config = CasConfig::default();
            let json = serde_json::to_string_pretty(&config)
                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
            std::fs::write(&config_path, json)?;
            config
        };

        if config.normalization_level > 2 {
            return Err(FxcpError::Config(
                "normalization_level must be 0..=2".into(),
            ));
        }

        Ok(Self { root, config, compact_lock: RwLock::new(()) })
    }

    /// Store a file by chunking its data and writing unique chunks.
    pub fn store_file(&self, rel_path: &str, data: &[u8], meta: &FileMeta) -> crate::Result<CasStoreStats> {
        let _guard = self.compact_lock.read()
            .unwrap_or_else(|e| e.into_inner());

        let chunker = GearChunker::new(self.config.chunk_min, self.config.chunk_avg, self.config.chunk_max)?
            .with_normalization(self.config.normalization_level)?;
        let file_hash = blake3::hash(data);
        let mut stats = CasStoreStats::default();
        let mut chunk_hashes = Vec::new();
        let mut chunk_sizes = Vec::new();

        let boundaries = chunker.find_boundaries(data);
        let mut start = 0;
        for end in &boundaries {
            let slice = &data[start..*end];
            let hash = blake3::hash(slice);
            let hash_cid = crate::cid::blake3_to_cid_string(hash.as_bytes());

            chunk_sizes.push(slice.len() as u64);
            if self.chunk_exists(&hash_cid) {
                stats.chunks_deduped += 1;
                stats.bytes_deduped += slice.len() as u64;
            } else {
                let compressed = crate::fxar::compress_chunk(slice, &self.config.compression);
                self.write_chunk_atomic(&hash_cid, &compressed)?;
                stats.chunks_written += 1;
                stats.bytes_written += compressed.len() as u64;
            }
            chunk_hashes.push(hash_cid);
            start = *end;
        }

        let op = ManifestOp::Put {
            path: rel_path.to_string(),
            meta: meta.clone(),
            blake3: crate::cid::blake3_to_cid_string(file_hash.as_bytes()),
            chunks: chunk_hashes,
            chunk_sizes,
        };
        self.append_manifest(&op)?;

        Ok(stats)
    }

    /// Store a file using chunk-level delta, writing only changed chunks.
    pub fn delta_store_file(
        &self,
        rel_path: &str,
        data: &[u8],
        meta: &FileMeta,
    ) -> crate::Result<DeltaStoreStats> {
        let files = self.list_files()?;
        let old_entry = files.iter().find(|f| f.path == rel_path);

        let old_entry = match old_entry {
            Some(e) if e.chunks.len() >= 3 && !e.chunk_sizes.is_empty() => e,
            _ => {
                let stats = self.store_file(rel_path, data, meta)?;
                return Ok(DeltaStoreStats {
                    chunks_added: stats.chunks_written + stats.chunks_deduped,
                    chunks_reused: 0,
                    bytes_uploaded: stats.bytes_written,
                    bytes_skipped: 0,
                    efficiency: 0.0,
                });
            }
        };

        let old_chunks: Vec<ChunkRef> = old_entry
            .chunks
            .iter()
            .zip(old_entry.chunk_sizes.iter().copied())
            .map(|(h, s)| ChunkRef { hash: h.clone(), size: s })
            .collect();

        let _guard = self.compact_lock.read().unwrap_or_else(|e| e.into_inner());

        let chunker = GearChunker::new(
            self.config.chunk_min,
            self.config.chunk_avg,
            self.config.chunk_max,
        )?
        .with_normalization(self.config.normalization_level)?;

        let file_hash = blake3::hash(data);
        let boundaries = chunker.find_boundaries(data);

        let mut new_chunk_data: Vec<(&[u8], String, u64)> = Vec::new();
        let mut start = 0;
        for end in &boundaries {
            let slice = &data[start..*end];
            let hash_cid = crate::cid::blake3_to_cid_string(blake3::hash(slice).as_bytes());
            let size = slice.len() as u64;
            new_chunk_data.push((slice, hash_cid, size));
            start = *end;
        }

        let new_chunks: Vec<ChunkRef> = new_chunk_data
            .iter()
            .map(|(_, h, s)| ChunkRef { hash: h.clone(), size: *s })
            .collect();

        let delta = compute_chunk_delta(&old_chunks, &new_chunks);

        if delta.full_upload_recommended {
            drop(_guard);
            let stats = self.store_file(rel_path, data, meta)?;
            return Ok(DeltaStoreStats {
                chunks_added: stats.chunks_written + stats.chunks_deduped,
                chunks_reused: 0,
                bytes_uploaded: stats.bytes_written,
                bytes_skipped: 0,
                efficiency: 0.0,
            });
        }

        let added_set: HashSet<&str> = delta.added.iter().map(|c| c.hash.as_str()).collect();
        let mut bytes_uploaded: u64 = 0;
        let mut bytes_skipped: u64 = 0;
        let mut chunks_written: Vec<String> = Vec::new();

        for (slice, hash_cid, size) in &new_chunk_data {
            if added_set.contains(hash_cid.as_str()) {
                if !self.chunk_exists(hash_cid) {
                    let compressed =
                        crate::fxar::compress_chunk(slice, &self.config.compression);
                    self.write_chunk_atomic(hash_cid, &compressed)?;
                    bytes_uploaded += compressed.len() as u64;
                }
                chunks_written.push(hash_cid.clone());
            } else {
                bytes_skipped += size;
            }
        }

        // All writes succeeded -> update manifest
        let chunk_hashes: Vec<String> = new_chunks.iter().map(|c| c.hash.clone()).collect();
        let chunk_sizes: Vec<u64> = new_chunks.iter().map(|c| c.size).collect();

        let op = ManifestOp::Put {
            path: rel_path.to_string(),
            meta: meta.clone(),
            blake3: crate::cid::blake3_to_cid_string(file_hash.as_bytes()),
            chunks: chunk_hashes,
            chunk_sizes,
        };
        self.append_manifest(&op)?;

        Ok(DeltaStoreStats {
            chunks_added: delta.added.len() as u64,
            chunks_reused: delta.unchanged as u64,
            bytes_uploaded,
            bytes_skipped,
            efficiency: delta.efficiency,
        })
    }

    /// Record a file deletion in the manifest.
    pub fn delete_file(&self, rel_path: &str) -> crate::Result<()> {
        self.append_manifest(&ManifestOp::Delete { path: rel_path.to_string() })
    }

    /// Record a file rename in the manifest.
    pub fn rename_file(&self, from: &str, to: &str) -> crate::Result<()> {
        self.append_manifest(&ManifestOp::Rename {
            from: from.to_string(),
            to: to.to_string(),
        })
    }

    /// Record a snapshot marker in the manifest.
    pub fn snapshot(&self, name: &str) -> crate::Result<()> {
        self.append_manifest(&ManifestOp::Snapshot { name: name.to_string() })
    }

    /// Reconstruct a file from its chunks.
    pub fn read_file(&self, rel_path: &str) -> crate::Result<Vec<u8>> {
        let files = self.list_files()?;
        let entry = files.iter().find(|f| f.path == rel_path)
            .ok_or_else(|| FxcpError::CasStoreError(
                format!("file not found in CAS store: {}", rel_path)))?;

        let mut data = Vec::with_capacity(entry.meta.size as usize);
        for chunk_hash in &entry.chunks {
            let chunk_data = self.read_chunk(chunk_hash)?;
            data.extend_from_slice(&chunk_data);
        }

        // Verify whole-file BLAKE3 (accepts both hex and CID manifest fields)
        let actual = blake3::hash(&data);
        let expected: [u8; 32] = if crate::cid::is_blake3_cid_string(&entry.blake3) {
            crate::cid::cid_string_to_blake3(&entry.blake3).unwrap_or_default()
        } else {
            hex::decode(&entry.blake3).ok().and_then(|v| v.try_into().ok()).unwrap_or_default()
        };
        if *actual.as_bytes() != expected {
            return Err(FxcpError::ManifestCorrupt(
                format!("BLAKE3 mismatch for {}", rel_path)));
        }

        Ok(data)
    }

    /// Replay manifest to get current file state.
    pub fn list_files(&self) -> crate::Result<Vec<FileEntry>> {
        let ops = self.read_manifest()?;
        let mut state: HashMap<String, FileEntry> = HashMap::new();

        for op in ops {
            match op {
                ManifestOp::Put { path, meta, blake3, chunks, chunk_sizes } => {
                    state.insert(path.clone(), FileEntry { path, meta, blake3, chunks, chunk_sizes });
                }
                ManifestOp::Delete { path } => {
                    state.remove(&path);
                }
                ManifestOp::Rename { from, to } => {
                    if let Some(mut entry) = state.remove(&from) {
                        entry.path = to.clone();
                        state.insert(to, entry);
                    }
                }
                ManifestOp::Snapshot { .. } => {} // metadata only
            }
        }

        let mut files: Vec<FileEntry> = state.into_values().collect();
        files.sort_by(|a, b| a.path.cmp(&b.path));
        Ok(files)
    }

    /// Compact the manifest and remove orphaned chunks.
    pub fn compact(&self) -> crate::Result<CompactStats> {
        let _guard = self.compact_lock.write()
            .unwrap_or_else(|e| e.into_inner());

        let ops = self.read_manifest()?;
        let files = self.list_files()?;
        let mut stats = CompactStats::default();

        // Collect all referenced chunk hashes
        let referenced: HashSet<&str> = files.iter()
            .flat_map(|f| f.chunks.iter().map(|c| c.as_str()))
            .collect();

        stats.files_kept = files.len() as u64;
        stats.ops_removed = ops.len().saturating_sub(files.len()) as u64;

        // Rewrite manifest with only current state
        let manifest_path = self.root.join(MANIFEST_FILE);
        let tmp_path = self.root.join(format!("{}.tmp", MANIFEST_FILE));
        {
            let mut f = std::fs::File::create(&tmp_path)?;
            for file in &files {
                let op = ManifestOp::Put {
                    path: file.path.clone(),
                    meta: file.meta.clone(),
                    blake3: file.blake3.clone(),
                    chunks: file.chunks.clone(),
                    chunk_sizes: file.chunk_sizes.clone(),
                };
                let line = serde_json::to_string(&op)
                    .map_err(|e| FxcpError::ManifestCorrupt(e.to_string()))?;
                writeln!(f, "{}", line)?;
            }
            f.sync_data()?;
        }
        std::fs::rename(&tmp_path, &manifest_path)?;

        // Remove orphaned chunks
        let chunks_dir = self.root.join(CHUNKS_DIR);
        for prefix_entry in std::fs::read_dir(&chunks_dir)? {
            let prefix_entry = match prefix_entry { Ok(e) => e, Err(_) => continue };
            let prefix_name = prefix_entry.file_name().to_string_lossy().into_owned();
            if prefix_name == "tmp" { continue; }
            if !prefix_entry.file_type()?.is_dir() { continue; }

            for chunk_entry in std::fs::read_dir(prefix_entry.path())? {
                let chunk_entry = match chunk_entry { Ok(e) => e, Err(_) => continue };
                let chunk_name = chunk_entry.file_name().to_string_lossy().into_owned();
                if !referenced.contains(chunk_name.as_str()) {
                    let size = chunk_entry.metadata().map(|m| m.len()).unwrap_or(0);
                    if std::fs::remove_file(chunk_entry.path()).is_ok() {
                        stats.chunks_removed += 1;
                        stats.bytes_freed += size;
                    }
                }
            }
        }

        // Clean empty prefix directories
        for prefix_entry in std::fs::read_dir(&chunks_dir)? {
            let prefix_entry = match prefix_entry { Ok(e) => e, Err(_) => continue };
            let prefix_name = prefix_entry.file_name().to_string_lossy().into_owned();
            if prefix_name == "tmp" { continue; }
            let _ = std::fs::remove_dir(prefix_entry.path()); // only succeeds if empty
        }

        info!("CAS compact: {} files kept, {} ops removed, {} orphan chunks freed ({} bytes)",
              stats.files_kept, stats.ops_removed, stats.chunks_removed, stats.bytes_freed);

        Ok(stats)
    }

    /// Get store statistics.
    pub fn stats(&self) -> crate::Result<(u64, u64, u64)> {
        let files = self.list_files()?;
        let file_count = files.len() as u64;
        let apparent_bytes: u64 = files.iter().map(|f| f.meta.size).sum();
        let unique_chunks: HashSet<&str> = files.iter()
            .flat_map(|f| f.chunks.iter().map(|c| c.as_str()))
            .collect();
        Ok((file_count, apparent_bytes, unique_chunks.len() as u64))
    }

    // ---- FXAR CAS storage ----

    /// Store an FXAR archive as a CAS-chunked entity under a synthetic path.
    pub fn store_fxar_as_cas(&self, archive_name: &str, data: &[u8]) -> crate::Result<()> {
        let synthetic = format!("__fxar__/{archive_name}");
        let meta = FileMeta { size: data.len() as u64, mode: 0o644, mtime: 0, uid: 0, gid: 0 };
        self.store_file(&synthetic, data, &meta)?;
        Ok(())
    }

    /// Read an FXAR archive back from CAS, reconstructing it byte-identically.
    pub fn read_fxar_from_cas(&self, archive_name: &str) -> crate::Result<Vec<u8>> {
        let synthetic = format!("__fxar__/{archive_name}");
        let files = self.list_files()?;
        if !files.iter().any(|f| f.path == synthetic) {
            return Err(FxcpError::ChunkMissing(archive_name.to_string()));
        }
        self.read_file(&synthetic)
    }

    /// Check whether an FXAR archive exists in this CAS store.
    pub fn has_fxar_in_cas(&self, archive_name: &str) -> crate::Result<bool> {
        let synthetic = format!("__fxar__/{archive_name}");
        let files = self.list_files()?;
        Ok(files.iter().any(|f| f.path == synthetic))
    }

    // ---- FXAR interchange ----

    /// Import an FXAR v2 archive into this CAS store.
    /// Reads the archive, stores unique chunks, and creates manifest entries.
    pub fn import_fxar<R: io::Read + io::Seek>(&self, reader: R) -> crate::Result<CasStoreStats> {
        let mut fxar = crate::fxar::FxarReader::open(reader)?;
        let manifest = fxar.read_manifest()?;
        let mut stats = CasStoreStats::default();

        // Read each file via restore, then store in CAS (chunks are re-computed)
        for file_entry in &manifest.files {
            let data = match fxar.restore_file(&file_entry.path) {
                Ok(d) => d,
                Err(e) => {
                    warn!("CAS import: skipping {}: {}", file_entry.path, e);
                    continue;
                }
            };
            let meta = FileMeta {
                size: file_entry.size,
                mode: file_entry.mode,
                mtime: file_entry.mtime,
                uid: file_entry.uid,
                gid: file_entry.gid,
            };
            // Strip snapshot/tree prefix from FXAR path (e.g. "snap1/tree/data/file.txt" -> "data/file.txt")
            let cas_path = file_entry.path.find("/tree/")
                .map(|idx| &file_entry.path[idx + 6..])
                .unwrap_or(&file_entry.path);
            let file_stats = self.store_file(cas_path, &data, &meta)?;
            stats.chunks_written += file_stats.chunks_written;
            stats.chunks_deduped += file_stats.chunks_deduped;
            stats.bytes_written += file_stats.bytes_written;
            stats.bytes_deduped += file_stats.bytes_deduped;
        }

        info!("CAS import from FXAR: {} files, {} chunks written, {} deduped",
              manifest.files.len(), stats.chunks_written, stats.chunks_deduped);
        Ok(stats)
    }

    /// Export this CAS store as an FXAR v2 archive.
    pub fn export_fxar<W: io::Write + io::Seek>(&self, writer: W, compress: &str) -> crate::Result<crate::fxar::FxarExportStats> {
        let files = self.list_files()?;
        if files.is_empty() {
            return Ok(crate::fxar::FxarExportStats::default());
        }

        // Reconstruct all files and write via the FXAR writer infrastructure
        // Create a temporary version store layout that the FXAR writer can consume
        let tmp_dir = std::env::temp_dir().join(format!("foxing-cas-export-{}", std::process::id()));
        std::fs::create_dir_all(&tmp_dir)?;
        // Ensure cleanup on exit
        struct CleanupGuard(PathBuf);
        impl Drop for CleanupGuard {
            fn drop(&mut self) { let _ = std::fs::remove_dir_all(&self.0); }
        }
        let _guard = CleanupGuard(tmp_dir.clone());
        let tmp = &tmp_dir;
        let vs_root = tmp.join(".foxing_versions");
        let snap_ts = chrono::Utc::now().format("%Y-%m-%dT%H%M%S").to_string();
        let snap_dir = vs_root.join(&snap_ts);
        let tree_dir = snap_dir.join("tree");
        std::fs::create_dir_all(&tree_dir)?;

        // Write summary.json
        let summary = serde_json::json!({
            "timestamp": chrono::Utc::now().to_rfc3339(),
            "status": "success", "type": "full",
            "source": "cas-store", "trigger": "export",
            "files": files.len(), "size_bytes": 0,
            "disk_usage_bytes": 0, "savings_pct": 0.0, "elapsed_ms": 0
        });
        std::fs::write(snap_dir.join("summary.json"),
            serde_json::to_string_pretty(&summary).unwrap_or_default())?;

        // Reconstruct files from CAS store into tmp tree
        for file in &files {
            let data = self.read_file(&file.path)?;
            let dest = tree_dir.join(&file.path);
            if let Some(parent) = dest.parent() {
                std::fs::create_dir_all(parent)?;
            }
            std::fs::write(&dest, &data)?;
        }

        // Use FXAR writer
        let store = crate::version_store::VersionStore::open(tmp);
        crate::fxar::write_archive_seekable(&store, writer, compress, None)
    }

    // ---- Batch store (amortized fsync) ----

    /// Store multiple files in a single batch with amortized fsync.
    ///
    /// Reduces NFS COMMIT RPCs from 2N (per-chunk sync + per-chunk dir sync)
    /// to ~3 for an N-file batch by deferring all fsyncs to two bulk phases:
    ///   Phase 1: write all chunks (tmp -> rename, no fsync)
    ///   Phase 2: fsync each unique chunk shard directory (typically 1-3)
    ///   Phase 3: append all manifest entries + single fsync
    pub fn store_files_batch(
        &self,
        files: &[(String, &[u8], &FileMeta)],
    ) -> crate::Result<BatchStats> {
        let _guard = self.compact_lock.read()
            .unwrap_or_else(|e| e.into_inner());

        let mut all_chunk_dirs: HashSet<PathBuf> = HashSet::new();
        let mut manifest_ops: Vec<ManifestOp> = Vec::with_capacity(files.len());
        let mut fsync_count: usize = 0;

        // Phase 1: write all chunks without per-chunk fsync
        for (rel_path, data, meta) in files {
            let (chunk_hashes, chunk_sizes, dirs) = self.write_chunks_no_fsync(data)?;
            all_chunk_dirs.extend(dirs);

            let file_hash = blake3::hash(data);
            manifest_ops.push(ManifestOp::Put {
                path: rel_path.clone(),
                meta: (*meta).clone(),
                blake3: crate::cid::blake3_to_cid_string(file_hash.as_bytes()),
                chunks: chunk_hashes,
                chunk_sizes,
            });
        }

        // Phase 2: fsync each unique chunk shard directory
        for dir in &all_chunk_dirs {
            if let Ok(dir_file) = std::fs::File::open(dir)
                && let Err(e) = dir_file.sync_data() {
                    warn!("CAS batch dir fsync failed for {}: {e}", dir.display());
                }
            fsync_count += 1;
        }

        // Phase 3: append all manifest entries with single fsync
        self.append_manifest_batch(&manifest_ops)?;
        fsync_count += 1;

        Ok(BatchStats { files: files.len(), fsyncs: fsync_count })
    }

    /// Write all chunks for a single file without calling sync_data.
    ///
    /// Returns (chunk_cid_hashes, chunk_sizes, touched_shard_dirs).
    /// Caller is responsible for fsyncing the shard directories afterward.
    fn write_chunks_no_fsync(
        &self,
        data: &[u8],
    ) -> crate::Result<(Vec<String>, Vec<u64>, Vec<PathBuf>)> {
        let chunker = GearChunker::new(
            self.config.chunk_min,
            self.config.chunk_avg,
            self.config.chunk_max,
        )?.with_normalization(self.config.normalization_level)?;

        let boundaries = chunker.find_boundaries(data);
        let mut chunk_hashes = Vec::with_capacity(boundaries.len());
        let mut chunk_sizes = Vec::with_capacity(boundaries.len());
        let mut touched_dirs: Vec<PathBuf> = Vec::new();
        let mut seen_dirs: HashSet<PathBuf> = HashSet::new();

        let mut start = 0;
        for end in &boundaries {
            let slice = &data[start..*end];
            let hash = blake3::hash(slice);
            let hash_cid = crate::cid::blake3_to_cid_string(hash.as_bytes());

            chunk_sizes.push(slice.len() as u64);

            if !self.chunk_exists(&hash_cid) {
                let compressed = crate::fxar::compress_chunk(slice, &self.config.compression);
                self.write_chunk_no_sync(&hash_cid, &compressed)?;

                if let Some(parent) = self.chunk_path(&hash_cid).parent() {
                    let parent_buf = parent.to_path_buf();
                    if seen_dirs.insert(parent_buf.clone()) {
                        touched_dirs.push(parent_buf);
                    }
                }
            }

            chunk_hashes.push(hash_cid);
            start = *end;
        }

        Ok((chunk_hashes, chunk_sizes, touched_dirs))
    }

    /// Write a chunk to disk without any fsync (tmp -> rename only).
    fn write_chunk_no_sync(&self, hash_cid: &str, data: &[u8]) -> crate::Result<()> {
        let final_path = self.chunk_path(hash_cid);
        if final_path.exists() { return Ok(()); }

        if let Some(parent) = final_path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let tmp_path = self.root.join(TMP_DIR).join(hash_cid);
        {
            let mut f = std::fs::File::create(&tmp_path)?;
            f.write_all(data)?;
            // No sync_data  --  deferred to batch phase 2
        }

        std::fs::rename(&tmp_path, &final_path)?;
        Ok(())
    }

    /// Append multiple manifest operations with a single fsync.
    fn append_manifest_batch(&self, ops: &[ManifestOp]) -> crate::Result<()> {
        let manifest_path = self.root.join(MANIFEST_FILE);
        let mut f = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&manifest_path)?;
        for op in ops {
            let line = serde_json::to_string(op)
                .map_err(|e| FxcpError::ManifestCorrupt(e.to_string()))?;
            writeln!(f, "{}", line)?;
        }
        f.sync_data()?;
        Ok(())
    }

    // ---- Internal helpers ----

    fn chunk_path(&self, hash_cid: &str) -> PathBuf {
        let shard = if hash_cid.len() > 9 { &hash_cid[7..9] } else { &hash_cid[..2.min(hash_cid.len())] };
        self.root.join(CHUNKS_DIR).join(shard).join(hash_cid)
    }

    fn chunk_exists(&self, hash_cid: &str) -> bool {
        self.chunk_path(hash_cid).exists()
    }

    /// Crash-safe chunk write: tmp -> sync_data -> rename -> dir sync_data
    fn write_chunk_atomic(&self, hash_cid: &str, data: &[u8]) -> crate::Result<()> {
        let final_path = self.chunk_path(hash_cid);
        if final_path.exists() { return Ok(()); } // already stored

        // Ensure prefix directory exists
        if let Some(parent) = final_path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        // Write to tmp
        let tmp_path = self.root.join(TMP_DIR).join(hash_cid);
        {
            let mut f = std::fs::File::create(&tmp_path)?;
            f.write_all(data)?;
            f.sync_data()?;
        }

        // Atomic rename
        std::fs::rename(&tmp_path, &final_path)?;

        // Sync target directory to persist the rename
        if let Some(parent) = final_path.parent()
            && let Ok(dir) = std::fs::File::open(parent)
                && let Err(e) = dir.sync_data() {
                    warn!("CAS store dir fsync failed: {e}");
                }

        Ok(())
    }

    fn read_chunk(&self, hash_cid: &str) -> crate::Result<Vec<u8>> {
        let path = self.chunk_path(hash_cid);
        let compressed = std::fs::read(&path)
            .map_err(|e| FxcpError::ChunkMissing(format!("{}: {}", hash_cid, e)))?;
        // Decompress (zstd flag = 1)
        let flag = match self.config.compression.as_str() {
            "none" => 0,
            "lz4" => 2,
            "gzip" => 3,
            "xz" => 4,
            _ => 1, // zstd default
        };
        Ok(crate::fxar::decompress_chunk(&compressed, flag)?)
    }

    fn append_manifest(&self, op: &ManifestOp) -> crate::Result<()> {
        let manifest_path = self.root.join(MANIFEST_FILE);
        let line = serde_json::to_string(op)
            .map_err(|e| FxcpError::ManifestCorrupt(e.to_string()))?;
        let mut f = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&manifest_path)?;
        writeln!(f, "{}", line)?;
        f.sync_data()?;
        Ok(())
    }

    fn read_manifest(&self) -> crate::Result<Vec<ManifestOp>> {
        let manifest_path = self.root.join(MANIFEST_FILE);
        if !manifest_path.exists() {
            return Ok(Vec::new());
        }

        let file = std::fs::File::open(&manifest_path)?;
        let reader = io::BufReader::new(file);
        let mut ops = Vec::new();

        for line in reader.lines() {
            let line = match line { Ok(l) => l, Err(_) => break };
            let trimmed = line.trim();
            if trimmed.is_empty() { continue; }
            // Partial last line (crash recovery)  --  skip silently
            match serde_json::from_str::<ManifestOp>(trimmed) {
                Ok(op) => ops.push(op),
                Err(e) => {
                    debug!("CAS manifest: skipping malformed line: {}", e);
                }
            }
        }

        Ok(ops)
    }
}

// -----------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    use super::*;
    use tempfile::TempDir;

    fn test_store() -> (TempDir, CasStore) {
        let dir = TempDir::new().unwrap();
        let store = CasStore::open(dir.path()).unwrap();
        (dir, store)
    }

    #[test]
    fn test_store_read_roundtrip() {
        let (_dir, store) = test_store();
        let data = b"hello world, this is CAS store test data";
        let meta = FileMeta { size: data.len() as u64, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };

        store.store_file("test/file.txt", data, &meta).unwrap();
        let read_back = store.read_file("test/file.txt").unwrap();
        assert_eq!(read_back, data);
    }

    #[test]
    fn test_dedup_identical_files() {
        let (dir, store) = test_store();
        let data = vec![0x42u8; 100_000];
        let meta = FileMeta { size: data.len() as u64, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };

        let stats1 = store.store_file("file1.dat", &data, &meta).unwrap();
        let stats2 = store.store_file("file2.dat", &data, &meta).unwrap();

        assert!(stats1.chunks_written > 0);
        assert_eq!(stats2.chunks_written, 0, "second store should dedup all chunks");
        assert_eq!(stats2.chunks_deduped, stats1.chunks_written);

        // Verify both files read back correctly
        assert_eq!(store.read_file("file1.dat").unwrap(), data);
        assert_eq!(store.read_file("file2.dat").unwrap(), data);

        // Count actual chunk files on disk
        let chunks_dir = dir.path().join(CAS_DIR).join(CHUNKS_DIR);
        let chunk_count: usize = walkdir::WalkDir::new(&chunks_dir)
            .into_iter()
            .filter_map(|e| e.ok())
            .filter(|e| e.file_type().is_file())
            .count();
        assert_eq!(chunk_count as u64, stats1.chunks_written, "only one set of chunks on disk");
    }

    #[test]
    fn test_dedup_partial_change() {
        // Use a store with smaller chunk params for testability
        let dir = TempDir::new().unwrap();
        let root = dir.path().join(CAS_DIR);
        std::fs::create_dir_all(root.join(TMP_DIR)).unwrap();
        let config = CasConfig {
            version: 1,
            chunk_min: 512,
            chunk_avg: 4096,
            chunk_max: 16384,
            compression: "none".into(),
            normalization_level: 0,
        };
        let json = serde_json::to_string_pretty(&config).unwrap();
        std::fs::write(root.join(CONFIG_FILE), json).unwrap();
        let store = CasStore::open(dir.path()).unwrap();

        // Generate pseudo-random data that creates many chunks at 4KB avg
        let mut data = vec![0u8; 100_000];
        for (i, b) in data.iter_mut().enumerate() {
            *b = (i.wrapping_mul(0x9E3779B9) >> 24) as u8;
        }
        let meta = FileMeta { size: data.len() as u64, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };

        let stats1 = store.store_file("file.dat", &data, &meta).unwrap();
        assert!(stats1.chunks_written >= 4, "need multiple chunks for dedup test (got {})", stats1.chunks_written);

        // Modify ~5% of data near the middle
        let mut modified = data.clone();
        let mid = data.len() / 2;
        for i in mid..mid + 5000 {
            modified[i] = 0xFF;
        }
        let meta2 = FileMeta { size: modified.len() as u64, mode: 0o644, mtime: 2000, uid: 1000, gid: 1000 };
        let stats2 = store.store_file("file.dat", &modified, &meta2).unwrap();

        // Most chunks should be deduped (gear hash resynchronizes after the modified region)
        assert!(stats2.chunks_deduped > 0, "partial change should dedup some chunks (wrote={}, deduped={})",
            stats2.chunks_written, stats2.chunks_deduped);

        // Read back should return modified version
        assert_eq!(store.read_file("file.dat").unwrap(), modified);
    }

    #[test]
    fn test_delete_and_compact() {
        let (_dir, store) = test_store();
        let meta = FileMeta { size: 5, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };

        store.store_file("keep.txt", b"keepx", &meta).unwrap();
        store.store_file("delete.txt", b"delxx", &meta).unwrap();

        assert_eq!(store.list_files().unwrap().len(), 2);

        store.delete_file("delete.txt").unwrap();
        assert_eq!(store.list_files().unwrap().len(), 1);

        let stats = store.compact().unwrap();
        assert_eq!(stats.files_kept, 1);
        assert!(stats.chunks_removed > 0 || stats.ops_removed > 0);

        // Verify keep.txt still works
        assert_eq!(store.read_file("keep.txt").unwrap(), b"keepx");
        // Verify delete.txt is gone
        assert!(store.read_file("delete.txt").is_err());
    }

    #[test]
    fn test_crash_safe_partial_manifest() {
        let (dir, store) = test_store();
        let meta = FileMeta { size: 5, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };
        store.store_file("good.txt", b"good!", &meta).unwrap();

        // Append a partial/corrupt line to manifest
        let manifest_path = dir.path().join(CAS_DIR).join(MANIFEST_FILE);
        let mut f = std::fs::OpenOptions::new().append(true).open(&manifest_path).unwrap();
        write!(f, "{{\"op\":\"put\",\"path\":\"corrupt").unwrap(); // incomplete JSON

        // list_files should skip the corrupt line and return the good file
        let files = store.list_files().unwrap();
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].path, "good.txt");
    }

    #[test]
    fn test_snapshot_marker() {
        let (_dir, store) = test_store();
        store.snapshot("2026-03-14T080000").unwrap();

        let ops = store.read_manifest().unwrap();
        assert_eq!(ops.len(), 1);
        match &ops[0] {
            ManifestOp::Snapshot { name } => assert_eq!(name, "2026-03-14T080000"),
            _ => panic!("expected snapshot op"),
        }
    }

    #[test]
    fn test_rename_file() {
        let (_dir, store) = test_store();
        let meta = FileMeta { size: 4, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };
        store.store_file("old.txt", b"data", &meta).unwrap();
        store.rename_file("old.txt", "new.txt").unwrap();

        let files = store.list_files().unwrap();
        assert_eq!(files.len(), 1);
        assert_eq!(files[0].path, "new.txt");
        assert_eq!(store.read_file("new.txt").unwrap(), b"data");
        assert!(store.read_file("old.txt").is_err());
    }

    #[test]
    fn test_fxar_roundtrip() {
        // Create a CAS store with some files
        let (_dir, store) = test_store();
        let meta = FileMeta { size: 100, mode: 0o755, mtime: 1773484800, uid: 1000, gid: 1000 };
        store.store_file("data/test.txt", &[0x42u8; 100], &meta).unwrap();
        store.store_file("data/other.txt", &[0x43u8; 200], &FileMeta { size: 200, ..meta.clone() }).unwrap();

        // Export CAS -> FXAR
        let mut fxar_buf = std::io::Cursor::new(Vec::new());
        store.export_fxar(&mut fxar_buf, "none").unwrap();
        let fxar_data = fxar_buf.into_inner();
        assert!(fxar_data.len() > 64, "FXAR archive should have content");
        assert_eq!(&fxar_data[..4], b"FXAR");

        // Import FXAR -> new CAS store
        let dir2 = TempDir::new().unwrap();
        let store2 = CasStore::open(dir2.path()).unwrap();
        let cursor = std::io::Cursor::new(fxar_data);
        store2.import_fxar(cursor).unwrap();

        // Verify files match
        let files1 = store.list_files().unwrap();
        let files2 = store2.list_files().unwrap();
        assert_eq!(files1.len(), files2.len());
        for (a, b) in files1.iter().zip(files2.iter()) {
            assert_eq!(a.blake3, b.blake3, "BLAKE3 mismatch for {}", a.path);
        }

        // Verify content matches
        let d1 = store.read_file("data/test.txt").unwrap();
        let d2 = store2.read_file("data/test.txt").unwrap();
        assert_eq!(d1, d2);
    }

    #[test]
    fn test_error_type_alignment() {
        let result = CasStore::open(Path::new("/nonexistent/path/that/does/not/exist"));
        assert!(result.is_err());
        match result.err().unwrap() {
            crate::error::FxcpError::Io(_) => {}
            other => panic!("expected FxcpError::Io, got: {:?}", other),
        }

        let dir = TempDir::new().unwrap();
        let store = CasStore::open(dir.path()).unwrap();
        let result = store.read_file("nonexistent.txt");
        assert!(result.is_err());
        match result.err().unwrap() {
            crate::error::FxcpError::CasStoreError(msg) => {
                assert!(msg.contains("not found"), "msg: {}", msg);
            }
            other => panic!("expected FxcpError::CasStoreError, got: {:?}", other),
        }
    }

    #[test]
    fn test_stats() {
        let (_dir, store) = test_store();
        let meta = FileMeta { size: 100, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };
        store.store_file("a.dat", &[0x42u8; 100], &meta).unwrap();
        store.store_file("b.dat", &[0x43u8; 200], &FileMeta { size: 200, ..meta }).unwrap();

        let (files, apparent, chunks) = store.stats().unwrap();
        assert_eq!(files, 2);
        assert_eq!(apparent, 300);
        assert!(chunks > 0);
    }

    #[test]
    fn test_casconfig_default_normalization() {
        let json = r#"{"version":1,"chunk_min":2048,"chunk_avg":65536,"chunk_max":2097152,"compression":"zstd"}"#;
        let cfg: CasConfig = serde_json::from_str(json).unwrap();
        assert_eq!(cfg.normalization_level, 0);
    }

    #[test]
    fn test_casconfig_with_normalization() {
        let json = r#"{"version":1,"chunk_min":2048,"chunk_avg":65536,"chunk_max":2097152,"compression":"zstd","normalization_level":1}"#;
        let cfg: CasConfig = serde_json::from_str(json).unwrap();
        assert_eq!(cfg.normalization_level, 1);
    }

    #[test]
    fn test_cas_compact_concurrent_store() {
        use std::sync::{Arc, Barrier};
        use std::thread;

        let dir = TempDir::new().unwrap();
        let store = Arc::new(CasStore::open(dir.path()).unwrap());
        let barrier = Arc::new(Barrier::new(2));

        let store_w = Arc::clone(&store);
        let barrier_w = Arc::clone(&barrier);
        let writer = thread::spawn(move || {
            barrier_w.wait();
            for i in 0..10 {
                let data = format!("file-{}-payload-{}", i, "x".repeat(1024));
                let meta = FileMeta {
                    size: data.len() as u64,
                    mode: 0o644,
                    mtime: 1000 + i as i64,
                    uid: 1000,
                    gid: 1000,
                };
                store_w
                    .store_file(&format!("concurrent/{}.txt", i), data.as_bytes(), &meta)
                    .unwrap();
            }
        });

        let store_c = Arc::clone(&store);
        let barrier_c = Arc::clone(&barrier);
        let compactor = thread::spawn(move || {
            barrier_c.wait();
            for _ in 0..5 {
                let _ = store_c.compact();
                thread::yield_now();
            }
        });

        writer.join().unwrap();
        compactor.join().unwrap();

        for i in 0..10 {
            let path = format!("concurrent/{}.txt", i);
            let data = store
                .read_file(&path)
                .unwrap_or_else(|e| panic!("file {} lost after concurrent compact: {}", path, e));
            let expected = format!("file-{}-payload-{}", i, "x".repeat(1024));
            assert_eq!(data, expected.as_bytes(), "content mismatch for {}", path);
        }
    }

    #[test]
    fn test_casconfig_invalid_normalization() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join(CAS_DIR);
        std::fs::create_dir_all(root.join(TMP_DIR)).unwrap();
        let config = CasConfig {
            version: 1,
            chunk_min: 2048,
            chunk_avg: 65536,
            chunk_max: 2097152,
            compression: "zstd".into(),
            normalization_level: 3,
        };
        let json = serde_json::to_string_pretty(&config).unwrap();
        std::fs::write(root.join(CONFIG_FILE), json).unwrap();

        let result = CasStore::open(dir.path());
        assert!(result.is_err());
        match result.err().unwrap() {
            FxcpError::Config(msg) => assert!(msg.contains("normalization_level"), "msg: {}", msg),
            other => panic!("expected FxcpError::Config, got: {:?}", other),
        }
    }

    #[test]
    fn test_local_cas_chunk_sizes() {
        let (_dir, store) = test_store();
        let data = vec![0xABu8; 50_000];
        let meta = FileMeta { size: data.len() as u64, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 };

        store.store_file("chunked.bin", &data, &meta).unwrap();
        let files = store.list_files().unwrap();
        let entry = files.iter().find(|f| f.path == "chunked.bin").unwrap();

        assert_eq!(
            entry.chunk_sizes.len(),
            entry.chunks.len(),
            "chunk_sizes.len() ({}) must equal chunks.len() ({})",
            entry.chunk_sizes.len(),
            entry.chunks.len()
        );
        assert_eq!(
            entry.chunk_sizes.iter().sum::<u64>(),
            data.len() as u64,
            "sum(chunk_sizes) must equal file size"
        );
        for (i, &sz) in entry.chunk_sizes.iter().enumerate() {
            assert!(sz > 0, "chunk_sizes[{i}] must be > 0");
        }
    }

    #[test]
    fn test_manifest_op_put_chunk_sizes_backward_compat() {
        // Old JSON without chunk_sizes must deserialize to vec![]
        let json = r#"{"op":"put","path":"a.txt","size":0,"mode":0,"mtime":0,"uid":0,"gid":0,"blake3":"abc","chunks":["x"]}"#;
        let op: ManifestOp = serde_json::from_str(json).unwrap();
        match op {
            ManifestOp::Put { chunk_sizes, .. } => {
                assert_eq!(
                    chunk_sizes,
                    vec![] as Vec<u64>,
                    "chunk_sizes must default to empty for old manifests"
                );
            }
            _ => panic!("expected Put"),
        }
    }

    #[test]
    fn test_local_cas_fxar_roundtrip() {
        let (_dir, store) = test_store();
        let fxar_data: Vec<u8> = (0..4096u32).map(|i| (i.wrapping_mul(0x9E37) >> 8) as u8).collect();
        store.store_fxar_as_cas("test.fxar", &fxar_data).unwrap();
        let restored = store.read_fxar_from_cas("test.fxar").unwrap();
        assert_eq!(fxar_data, restored, "local CAS FXAR round-trip must be byte-identical");
    }

    #[test]
    fn test_local_cas_fxar_has_before_and_after() {
        let (_dir, store) = test_store();
        assert!(!store.has_fxar_in_cas("missing.fxar").unwrap());
        store.store_fxar_as_cas("present.fxar", &[0xAB; 512]).unwrap();
        assert!(store.has_fxar_in_cas("present.fxar").unwrap());
    }

    #[test]
    fn test_local_cas_fxar_read_missing_returns_error() {
        let (_dir, store) = test_store();
        let err = store.read_fxar_from_cas("nonexistent.fxar");
        assert!(err.is_err(), "reading missing FXAR must return Err");
    }

    #[test]
    fn test_local_cas_fxar_synthetic_path() {
        let (_dir, store) = test_store();
        store.store_fxar_as_cas("archive.fxar", &[0x42; 256]).unwrap();
        let files = store.list_files().unwrap();
        let fxar_entry = files.iter().find(|f| f.path == "__fxar__/archive.fxar");
        assert!(fxar_entry.is_some(), "FXAR must be stored under __fxar__/ synthetic path");
        let bare = files.iter().find(|f| f.path == "archive.fxar");
        assert!(bare.is_none(), "FXAR must NOT be stored under bare name");
    }

    // ---- Delta store tests ----

    fn test_meta() -> FileMeta {
        FileMeta { size: 0, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000 }
    }

    fn delta_test_data(size: usize) -> Vec<u8> {
        let mut data = Vec::with_capacity(size);
        let mut state: u32 = 0xDEAD_BEEF;
        for _ in 0..size {
            state = state.wrapping_mul(1103515245).wrapping_add(12345);
            data.push((state >> 16) as u8);
        }
        data
    }

    fn delta_test_store() -> (TempDir, CasStore) {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join(CAS_DIR);
        std::fs::create_dir_all(root.join(TMP_DIR)).unwrap();
        let config = CasConfig {
            version: 1,
            chunk_min: 512,
            chunk_avg: 4096,
            chunk_max: 16384,
            compression: "none".into(),
            normalization_level: 0,
        };
        let json = serde_json::to_string_pretty(&config).unwrap();
        std::fs::write(root.join(CONFIG_FILE), json).unwrap();
        let store = CasStore::open(dir.path()).unwrap();
        (dir, store)
    }

    #[test]
    fn test_local_delta_new_path_falls_through() {
        let (_dir, store) = delta_test_store();
        let data = delta_test_data(50_000);
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        let stats = store.delta_store_file("new.bin", &data, &meta).unwrap();
        assert!(stats.chunks_added > 0);
        assert_eq!(stats.chunks_reused, 0);
        assert_eq!(stats.efficiency, 0.0);
        assert_eq!(store.read_file("new.bin").unwrap(), data);
    }

    #[test]
    fn test_local_delta_unchanged_uploads_zero() {
        let (_dir, store) = delta_test_store();
        let data = delta_test_data(50_000);
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        store.store_file("same.bin", &data, &meta).unwrap();
        let stats = store.delta_store_file("same.bin", &data, &meta).unwrap();
        assert_eq!(stats.chunks_added, 0);
        assert!(stats.chunks_reused > 0);
        assert_eq!(stats.bytes_uploaded, 0);
        assert!((stats.efficiency - 1.0).abs() < f64::EPSILON);
    }

    #[test]
    fn test_local_delta_ten_percent_change() {
        let (_dir, store) = delta_test_store();
        let data = delta_test_data(50_000);
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        store.store_file("partial.bin", &data, &meta).unwrap();

        let mut modified = data.clone();
        let change_len = modified.len() / 10;
        for b in &mut modified[..change_len] {
            *b ^= 0xFF;
        }
        let meta2 = FileMeta { size: modified.len() as u64, ..test_meta() };
        let stats = store.delta_store_file("partial.bin", &modified, &meta2).unwrap();
        assert!(stats.efficiency > 0.5, "efficiency {} should be > 0.5", stats.efficiency);
        assert!(stats.bytes_skipped > 0);
    }

    #[test]
    fn test_local_delta_produces_correct_content() {
        let (_dir, store) = delta_test_store();
        let data = delta_test_data(50_000);
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        store.store_file("content.bin", &data, &meta).unwrap();

        let mut modified = data.clone();
        modified[100..200].fill(0xAA);
        let meta2 = FileMeta { size: modified.len() as u64, ..test_meta() };
        store.delta_store_file("content.bin", &modified, &meta2).unwrap();

        assert_eq!(store.read_file("content.bin").unwrap(), modified);
    }

    #[test]
    fn test_local_delta_small_file_skips() {
        let (_dir, store) = delta_test_store();
        let data = b"tiny file";
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        store.store_file("tiny.bin", data, &meta).unwrap();

        let files = store.list_files().unwrap();
        let entry = files.iter().find(|f| f.path == "tiny.bin").unwrap();
        assert!(entry.chunks.len() < 3, "test setup: need <3 chunks");

        let stats = store.delta_store_file("tiny.bin", data, &meta).unwrap();
        assert_eq!(stats.efficiency, 0.0);
    }

    #[test]
    fn test_local_delta_idempotent() {
        let (_dir, store) = delta_test_store();
        let data = delta_test_data(50_000);
        let meta = FileMeta { size: data.len() as u64, ..test_meta() };
        store.store_file("idem.bin", &data, &meta).unwrap();

        let stats1 = store.delta_store_file("idem.bin", &data, &meta).unwrap();
        assert_eq!(stats1.chunks_added, 0);

        let stats2 = store.delta_store_file("idem.bin", &data, &meta).unwrap();
        assert_eq!(stats2.chunks_added, 0);
        assert!((stats2.efficiency - 1.0).abs() < f64::EPSILON);
    }

    #[test]
    fn test_cid_chunk_path_sharding() {
        let (_dir, store) = test_store();
        let hash = blake3::hash(b"test-shard-input");
        let cid = crate::cid::blake3_to_cid_string(hash.as_bytes());
        assert!(cid.len() > 9, "CID string must be longer than 9 chars");

        let path = store.chunk_path(&cid);
        // chunk_path returns: root / CHUNKS_DIR / shard / cid
        // shard must equal cid[7..9]
        let expected_shard = &cid[7..9];
        let parent = path.parent().unwrap();
        let actual_shard = parent.file_name().unwrap().to_str().unwrap();
        assert_eq!(actual_shard, expected_shard,
            "shard directory should be chars 7-9 of CID string: cid={cid}");
    }

    // ---- Batch store tests ----

    #[test]
    fn test_store_files_batch_content_integrity() {
        let (_dir, store) = test_store();
        let payloads: Vec<(String, Vec<u8>)> = (0..5).map(|i| {
            let content = format!("batch-file-{}-payload-{}", i, "x".repeat(4096));
            (format!("batch/{}.txt", i), content.into_bytes())
        }).collect();

        let files: Vec<(String, &[u8], &FileMeta)> = payloads.iter().map(|(p, d)| {
            let meta = Box::leak(Box::new(FileMeta {
                size: d.len() as u64, mode: 0o644, mtime: 1000, uid: 1000, gid: 1000,
            }));
            (p.clone(), d.as_slice(), meta as &FileMeta)
        }).collect();

        let stats = store.store_files_batch(&files).unwrap();
        assert_eq!(stats.files, 5);

        for (path, data) in &payloads {
            let read_back = store.read_file(path).unwrap();
            assert_eq!(&read_back, data, "content mismatch for {}", path);
        }
    }

    #[test]
    fn test_store_files_batch_fsync_reduction() {
        let (_dir, store) = test_store();
        let payloads: Vec<(String, Vec<u8>)> = (0..10).map(|i| {
            let mut data = vec![0u8; 4096];
            for (j, b) in data.iter_mut().enumerate() {
                *b = ((i * 1000 + j).wrapping_mul(0x9E3779B9) >> 24) as u8;
            }
            (format!("fsync_test/{}.bin", i), data)
        }).collect();

        let files: Vec<(String, &[u8], &FileMeta)> = payloads.iter().map(|(p, d)| {
            let meta = Box::leak(Box::new(FileMeta {
                size: d.len() as u64, mode: 0o644, mtime: 2000, uid: 1000, gid: 1000,
            }));
            (p.clone(), d.as_slice(), meta as &FileMeta)
        }).collect();

        let stats = store.store_files_batch(&files).unwrap();
        assert_eq!(stats.files, 10);
        // Per-file store_file: 3 fsyncs each (chunk sync + dir sync + manifest sync) = 30.
        // Batch: 0 chunk syncs + unique_dirs + 1 manifest sync. With 10 random CIDs,
        // up to 10 shard dirs possible, so max = 11. Always < 30.
        assert!(stats.fsyncs < 3 * 10,
            "batch fsyncs ({}) must be less than per-file total (30)", stats.fsyncs);

        for (path, data) in &payloads {
            let read_back = store.read_file(path).unwrap();
            assert_eq!(&read_back, data, "content mismatch for {}", path);
        }

        let listed = store.list_files().unwrap();
        let batch_files: Vec<_> = listed.iter()
            .filter(|f| f.path.starts_with("fsync_test/"))
            .collect();
        assert_eq!(batch_files.len(), 10);
    }
}