fxcp_core/sync/

mod.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
//
// fxcp-core/src/sync/mod.rs  --  cli_main() dispatch + public entry points

//! Shared copy/sync engine used by both fxcp CLI and foxingd sync command.
//!
//! Provides recursive directory copy with auto-adaptive strategy selection
//! (reflink -> copy_file_range -> sendfile -> io_uring), stdin pipe mode with
//! compression auto-detection, delta copy via BLAKE3 Merkle trees, and
//! foxingd-compatible signature generation.

#![allow(clippy::expect_used)]
mod engine;
mod tree;
mod stdin;
#[cfg(feature = "cloud")]
mod s3;

// Re-exports  --  preserve public API surface for external crates
pub use tree::{store_foxing_signatures, preserve_metadata};
#[allow(unused_imports)]
pub(crate) use tree::create_copier;
#[cfg(feature = "cloud")]
pub use s3::{sync_from_s3, SyncFromS3Stats, parse_s3_url, build_s3_store};

use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use tracing::{info, error};

use crate::hashing;

// Auto-adaptive thresholds
const FICLONE: u64 = crate::constants::FICLONE_IOCTL;

// -----------------------------------------------------------------------
// Public types
// -----------------------------------------------------------------------

/// Options for a sync/copy operation (replaces CLI flags).
#[derive(Debug, Clone)]
pub struct SyncOptions {
    pub source: PathBuf,
    pub destination: PathBuf,
    pub archive: bool,
    pub recursive: bool,
    pub delete: bool,
    pub dry_run: bool,
    pub exclude: Vec<String>,
    pub include: Vec<String>,
    pub generate_sigs: bool,
    pub snapshot: bool,
    pub throttle: bool,
    pub cleanup: bool,
    // stdin-specific
    pub size: Option<u64>,
    pub checkpoint_interval: Option<u64>,
    pub checkpoint_keep: usize,
    pub zero_copy: bool,
    pub verify: bool,
    pub checksums: Vec<hashing::ChecksumType>,
    pub checksums_overwrite: bool,
    pub cid_output: bool,
    pub owner: bool,
    pub group: bool,
    pub perms: bool,
    pub xattrs: bool,
    pub acls: bool,
}

impl Default for SyncOptions {
    fn default() -> Self {
        Self {
            source: PathBuf::new(),
            destination: PathBuf::new(),
            archive: false,
            recursive: false,
            delete: false,
            dry_run: false,
            exclude: vec![],
            include: vec![],
            generate_sigs: false,
            snapshot: false,
            throttle: false,
            cleanup: false,
            size: None,
            checkpoint_interval: None,
            checkpoint_keep: 5,
            zero_copy: false,
            verify: false,
            checksums: Vec::new(),
            checksums_overwrite: false,
            cid_output: false,
            owner: true,
            group: true,
            perms: true,
            xattrs: true,
            acls: false,
        }
    }
}

/// Progress tracking for live reporting during copy operations.
pub struct ProgressTracker {
    pub files_done: std::sync::atomic::AtomicU64,
    pub files_total: std::sync::atomic::AtomicU64,
    pub bytes_done: std::sync::atomic::AtomicU64,
    pub bytes_total: std::sync::atomic::AtomicU64,
    pub active: std::sync::atomic::AtomicBool,
    /// 0 = counting, 1 = syncing, 2 = finalizing (dir hashes)
    pub phase: std::sync::atomic::AtomicU8,
    pub status_detail: parking_lot::Mutex<String>,
}

impl ProgressTracker {
    pub fn new() -> std::sync::Arc<Self> {
        std::sync::Arc::new(Self {
            files_done: std::sync::atomic::AtomicU64::new(0),
            files_total: std::sync::atomic::AtomicU64::new(0),
            bytes_done: std::sync::atomic::AtomicU64::new(0),
            bytes_total: std::sync::atomic::AtomicU64::new(0),
            active: std::sync::atomic::AtomicBool::new(true),
            phase: std::sync::atomic::AtomicU8::new(0),
            status_detail: parking_lot::Mutex::new(String::new()),
        })
    }
}

// Global progress tracker  --  set by cli_main when --progress is used.
lazy_static::lazy_static! {
    static ref ACTIVE_PROGRESS: parking_lot::Mutex<Option<std::sync::Arc<ProgressTracker>>> = parking_lot::Mutex::new(None);
}

/// Record a file completion for progress reporting.
fn record_file_progress(bytes: u64) {
    if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
        progress.files_done.fetch_add(1, Ordering::Relaxed);
        progress.bytes_done.fetch_add(bytes, Ordering::Relaxed);
    }
}

fn set_progress_phase(phase: u8) {
    if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
        progress.phase.store(phase, Ordering::Relaxed);
    }
}

fn set_progress_detail(detail: &str) {
    if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
        *progress.status_detail.lock() = detail.to_string();
    }
}

/// Spawn a progress reporter thread that prints updates to stderr.
/// Returns a join handle. Set active=false to stop.
fn spawn_progress_reporter(progress: std::sync::Arc<ProgressTracker>, json_mode: bool) -> std::thread::JoinHandle<()> {
    std::thread::spawn(move || {
        let start = std::time::Instant::now();
        while progress.active.load(Ordering::Relaxed) {
            std::thread::sleep(std::time::Duration::from_millis(if json_mode { crate::constants::PROGRESS_INTERVAL_JSON_MS } else { crate::constants::PROGRESS_INTERVAL_HUMAN_MS }));

            let done = progress.files_done.load(Ordering::Relaxed);
            let total = progress.files_total.load(Ordering::Relaxed);
            let bytes_done = progress.bytes_done.load(Ordering::Relaxed);
            let bytes_total = progress.bytes_total.load(Ordering::Relaxed);

            if total == 0 { continue; }

            let pct = if bytes_total > 0 { (bytes_done as f64 / bytes_total as f64) * 100.0 } else { 0.0 };
            let elapsed = start.elapsed().as_secs_f64();
            let throughput = if elapsed > 0.0 { bytes_done as f64 / elapsed / 1_048_576.0 } else { 0.0 };
            let eta = if throughput > 0.0 && bytes_total > bytes_done {
                ((bytes_total - bytes_done) as f64 / (throughput * 1_048_576.0)) as u64
            } else { 0 };

            let phase = progress.phase.load(Ordering::Relaxed);

            if json_mode {
                let json = serde_json::json!({
                    "progress_pct": (pct * 10.0).round() / 10.0,
                    "bytes_done": bytes_done,
                    "bytes_total": bytes_total,
                    "files_done": done,
                    "files_total": total,
                    "throughput_mbps": (throughput * 10.0).round() / 10.0,
                    "eta_secs": eta,
                    "phase": match phase { 2 => "finalizing", _ => "syncing" },
                });
                eprintln!("{}", serde_json::to_string(&json).unwrap_or_default());
            } else if phase >= 2 {
                let detail = progress.status_detail.lock().clone();
                let label = if detail.is_empty() {
                    "storing dir hashes...".to_string()
                } else {
                    let max_path = 40;
                    if detail.len() > max_path {
                        format!("...{}", &detail[detail.len() - max_path..])
                    } else {
                        detail
                    }
                };
                eprint!("\r[done] {} files | {:<60}",
                        done, label);
            } else {
                let done_str = format_size_compact(bytes_done);
                let total_str = format_size_compact(bytes_total);
                eprint!("\r[{:>5.1}%] {} / {} | {} / {} files | {:.0} MB/s | ETA {}s   ",
                        pct, done_str, total_str, done, total, throughput, eta);
            }
        }
        if !json_mode { eprintln!(); } // Final newline
    })
}

/// Check available disk space on the filesystem containing `path`.
/// Returns None if statvfs fails (e.g., FUSE without statvfs support).
fn check_available_space(path: &Path) -> Option<u64> {
    use std::ffi::CString;
    let c_path = CString::new(path.to_string_lossy().as_bytes()).ok()?;
    // SAFETY: statvfs is a C struct where all-zeros is a valid initial state.
    let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
    // SAFETY: c_path is a valid null-terminated C string, stat is a valid pointer.
    let rc = unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) };
    if rc == 0 {
        Some(stat.f_bavail * stat.f_bsize)
    } else {
        None
    }
}

fn format_size_compact(bytes: u64) -> String {
    crate::fmt::format_size(bytes)
}

/// Copy statistics returned from sync operations.
#[derive(Debug, Default, serde::Serialize)]
pub struct SyncStats {
    pub files_copied: u64,
    pub files_reflinked: u64,
    pub files_cfr: u64,
    pub files_small: u64,
    pub files_skipped: u64,
    pub files_delta: u64,
    pub files_deleted: u64,
    pub dirs_created: u64,
    pub dirs_pruned: u64,
    pub bytes_copied: u64,
    pub bytes_reflinked: u64,
    pub bytes_cfr: u64,
    pub bytes_small: u64,
    pub bytes_delta: u64,
    pub errors: u64,
    pub sigs_stored: u64,
    pub symlinks_replicated: u64,
    pub dirs_hashed: u64,
    pub files_verified: u64,
    pub verify_failures: u64,
    #[cfg(feature = "nfs-bypass")]
    pub files_nfs_bypass: u64,
    #[cfg(feature = "nfs-bypass")]
    pub bytes_nfs_bypass: u64,
}

impl SyncStats {
    /// Accumulate stats from another run (for multi-source operations).
    pub fn merge(&mut self, other: &SyncStats) {
        self.files_copied += other.files_copied;
        self.files_reflinked += other.files_reflinked;
        self.files_cfr += other.files_cfr;
        self.files_small += other.files_small;
        self.files_skipped += other.files_skipped;
        self.files_delta += other.files_delta;
        self.files_deleted += other.files_deleted;
        self.dirs_created += other.dirs_created;
        self.dirs_pruned += other.dirs_pruned;
        self.bytes_copied += other.bytes_copied;
        self.bytes_reflinked += other.bytes_reflinked;
        self.bytes_cfr += other.bytes_cfr;
        self.bytes_small += other.bytes_small;
        self.bytes_delta += other.bytes_delta;
        self.errors += other.errors;
        self.sigs_stored += other.sigs_stored;
        self.symlinks_replicated += other.symlinks_replicated;
        self.dirs_hashed += other.dirs_hashed;
        self.files_verified += other.files_verified;
        self.verify_failures += other.verify_failures;
        #[cfg(feature = "nfs-bypass")]
        {
            self.files_nfs_bypass += other.files_nfs_bypass;
            self.bytes_nfs_bypass += other.bytes_nfs_bypass;
        }
    }
}

// -----------------------------------------------------------------------
// Main entry point
// -----------------------------------------------------------------------

/// Run a sync operation with the given options.
pub async fn run(opts: SyncOptions) -> crate::Result<SyncStats> {
    if opts.cleanup {
        tree::run_cleanup(&opts.source);
        return Ok(SyncStats::default());
    }

    if opts.source.as_os_str() == "-" {
        return stdin::run_stdin_to_file(&opts).await;
    }

    engine::run_sync(&opts).await
}

// -----------------------------------------------------------------------
// CLI definition (public for man page / completion generation)
// -----------------------------------------------------------------------

use clap::{Parser, Subcommand};

#[derive(Parser)]
#[command(name = "fxcp", version,
    about = "Smart filesystem copy with CoW/reflink/io_uring support",
    long_about = "Smart filesystem copy with CoW/reflink/io_uring support.\n\
\n\
Automatically selects the optimal copy strategy: FICLONE (instant CoW clone), \
copy_file_range (server-side NFS copy), sendfile (kernel-optimized for small files), \
or io_uring (async pipelined for large/cross-device files). Generates BLAKE3 Merkle \
signatures for incremental resync and is compatible with foxingd daemon signatures.\n\
\n\
Sparse files are detected via SEEK_HOLE/SEEK_DATA and replicated without materializing holes.\n\
Multiple sources can be specified with the last argument as destination.\n\
\n\
DIRECTORY SEMANTICS (rsync-compatible):\n\
  fxcp -a /src  /dest    no trailing slash, dest exists   -> /dest/src/\n\
  fxcp -a /src/ /dest/   trailing slash on source         -> /dest/{contents}\n\
  fxcp -a /src  /newdir  dest does not yet exist          -> /newdir/\n\
\n\
This matches rsync behavior. The trailing slash on the source controls whether the\n\
source directory itself appears in the destination."
)]
#[command(after_help = r#"SUBCOMMANDS:
  snap             Snapshot management (MARS versioning -- list, export, restore, prune)
  index            Index files for semantic search (requires --features ai)
  search           Semantic search across indexed files (requires --features ai)
  enrich           Enrich files with LLM-generated metadata (requires --features enrichment)
  restore          Restore files from S3 CAS store
  export-vectors   Export embedding vectors to pgvector (requires --features pgvector)

Run 'fxcp <SUBCOMMAND> --help' for more information on a specific subcommand.

DIRECTORY SEMANTICS:
  fxcp -a /src  /dest     # dest exists, no trailing slash -> /dest/src/
  fxcp -a /src/ /dest/    # trailing slash -> /dest/{files}
  fxcp -a /src  /newdir   # dest absent -> /newdir/ (rename semantics)

  Matches rsync behavior. Use trailing slash on source to copy contents only."#)]
pub struct FxcpCli {
    /// Source path(s) and destination  --  last argument is destination (use '-' for stdin).
    /// Trailing slash on source controls directory behavior (rsync-compatible):
    /// 'fxcp src dest' creates dest/src/; 'fxcp src/ dest' copies contents into dest/
    #[arg(required = true, num_args = 2..)]
    pub paths: Vec<PathBuf>,
    #[arg(short = 'a', long, help = "Archive mode (recursive, preserve attributes)")]
    pub archive: bool,
    #[arg(short = 'r', long, help = "Recursive copy (implied by -a)")]
    pub recursive: bool,
    #[arg(short = 'v', long, help = "BLAKE3 verification after copy")]
    pub verify: bool,
    #[arg(long, help = "Delete files in target not present in source")]
    pub delete: bool,
    #[arg(short = 'n', long, help = "Dry run -- show what would be copied")]
    pub dry_run: bool,
    #[arg(short = 'e', long, help = "Exclude pattern (glob)")]
    pub exclude: Vec<String>,
    #[arg(long, help = "Include pattern -- override excludes (glob)")]
    pub include: Vec<String>,
    #[arg(long, help = "Read exclude patterns from FILE (one per line)")]
    pub exclude_from: Option<PathBuf>,
    #[arg(long, help = "Read include patterns from FILE (one per line)")]
    pub include_from: Option<PathBuf>,
    #[arg(long, help = "Create reflink snapshots of target files before overwriting (versioning)")]
    pub snapshot: bool,
    #[arg(long, help = "Enable PSI-based system stress throttling")]
    pub throttle: bool,
    #[arg(long, help = "Clean orphaned .tmp files and stale dirty flags")]
    pub cleanup: bool,
    #[arg(long, help = "Expected size in bytes (for stdin pre-allocation)")]
    pub size: Option<u64>,
    #[arg(long, help = "Interval in seconds to create CoW checkpoints of stdin stream")]
    pub checkpoint_interval: Option<u64>,
    #[arg(long, default_value = "5", help = "Number of stream checkpoints to keep")]
    pub checkpoint_keep: usize,
    #[arg(long, help = "Use zero-copy splice (mutually exclusive with sparse detection)")]
    pub zero_copy: bool,
    #[arg(long, default_value_t = false, help = "Increase verbosity")]
    pub debug: bool,
    #[arg(long, help = "Output copy results as JSON (machine-readable)")]
    pub json: bool,
    #[arg(long, help = "Show live progress during copy")]
    pub progress: bool,
    #[arg(long, help = "Generate foxingd-compatible sync signatures (xattr/sidecar) for fast resync")]
    pub generate_sigs: bool,
    #[arg(long, value_name = "TYPES", default_missing_value = "blake3", num_args = 0..=1,
          help = "Write checksum sidecar files alongside copies. Comma-separated: blake3 (FREE), sha256, sha1, md5. Default: blake3")]
    pub checksums: Option<String>,
    #[arg(long, help = "Overwrite existing checksum sidecar files (default: skip if present)")]
    pub checksums_overwrite: bool,
    #[arg(long, help = "Output BLAKE3 checksums as CID base32lower strings instead of hex")]
    pub cid: bool,
    #[arg(long, default_value_t = 0, value_parser = clap::value_parser!(u8).range(0..=2),
          help = "Chunk normalization level (0=legacy, 1=normalized, 2=aggressive). WARNING: changing this breaks CAS dedup compatibility with existing data")]
    pub normalization_level: u8,
    #[arg(long, default_value_t = true, action = clap::ArgAction::Set,
          help = "Preserve file owner (use --no-owner to disable)")]
    pub owner: bool,
    #[arg(long, default_value_t = true, action = clap::ArgAction::Set,
          help = "Preserve file group (use --no-group to disable)")]
    pub group: bool,
    #[arg(long, default_value_t = true, action = clap::ArgAction::Set,
          help = "Preserve permissions (use --no-perms to disable)")]
    pub perms: bool,
    #[arg(long, default_value_t = true, action = clap::ArgAction::Set,
          help = "Preserve extended attributes (use --no-xattrs to disable)")]
    pub xattrs: bool,
    #[arg(long, default_value_t = false, action = clap::ArgAction::Set,
          help = "Preserve ACLs (subset of xattrs, explicit opt-in)")]
    pub acls: bool,
}

/// Snapshot management subcommands (MARS versioning)
#[derive(Parser)]
#[command(name = "fxcp", version, about = "Smart filesystem copy with CoW/reflink/io_uring support")]
pub struct FxcpSnapCli {
    #[command(subcommand)]
    pub command: SnapCommand,
    #[arg(long, default_value_t = false, help = "Increase verbosity")]
    pub debug: bool,
}

#[derive(Subcommand)]
pub enum SnapCommand {
    /// List snapshots or versions of a specific file
    List {
        /// Target directory or specific file path
        path: Option<String>,
        /// Output as JSON (machine-readable)
        #[arg(long)]
        json: bool,
    },
    /// Show details of a specific snapshot
    Show { timestamp: String },
    /// Revert a file to a previous version (atomic reflink swap)
    Revert { path: String, epoch: u64 },
    /// Copy a specific version to a new file
    Copy { path: String, epoch: u64, destination: String },
    /// Remove old snapshots by age, count, or size
    Prune {
        /// Delete snapshots older than this duration (e.g. 30d, 7d, 24h)
        #[arg(long)]
        older_than: Option<String>,
        /// Keep only the last N snapshots
        #[arg(long)]
        keep_last: Option<usize>,
        /// Delete oldest until total size is under this limit (e.g. 50G, 10G)
        #[arg(long)]
        max_size: Option<String>,
        /// Target directory containing .foxing_versions
        #[arg(default_value = ".")]
        path: String,
    },
    /// Tag a snapshot (tagged snapshots are exempt from auto-pruning)
    Tag { timestamp: String, tag: String },
    /// Rebuild index.json from on-disk snapshot state
    RebuildIndex {
        /// Target directory containing .foxing_versions
        #[arg(default_value = ".")]
        path: String,
    },
    /// Show aggregate storage statistics (apparent vs on-disk, CoW savings)
    Stats {
        /// Target directory containing .foxing_versions
        #[arg(default_value = ".")]
        path: String,
        /// Output as JSON
        #[arg(long)]
        json: bool,
    },
    /// Export snapshots as a .fxar archive (content-addressable, deduped)
    Export {
        /// Source directory containing .foxing_versions or .foxing_cas
        path: String,
        /// Output file (default: stdout)
        #[arg(short, long)]
        output: Option<String>,
        /// Compression: zstd (default), zstd:N, lz4, gzip, xz, xz:N, none
        #[arg(long, default_value = "zstd")]
        compress: String,
        /// Export only a specific snapshot timestamp
        #[arg(long)]
        timestamp: Option<String>,
        /// Archive format: fxar2 (default, chunk-dedup) or tar (legacy whole-file)
        #[arg(long, default_value = "fxar2")]
        format: String,
        /// Bundle .foxing_index/ AI sections (embeddings, HNSW) into the archive
        #[arg(long)]
        include_vectors: bool,
        /// Export from .foxing_cas/ CAS store instead of .foxing_versions/
        #[arg(long)]
        from_cas: bool,
    },
    /// Import snapshots from a .fxar archive
    Import {
        /// Target directory to restore into
        path: String,
        /// Input file (default: stdin)
        #[arg(short, long)]
        input: Option<String>,
        /// Decompression: auto (default), zstd, lz4, gzip, xz, none
        #[arg(long, default_value = "auto")]
        compress: String,
        /// Generate foxingd-compatible signatures for fast incremental sync
        #[arg(long)]
        generate_sigs: bool,
        /// Preserve raw snapshot/tree/ path structure (default: extract latest version with flat paths)
        #[arg(long)]
        raw: bool,
        /// Show per-file version info during extraction
        #[arg(short, long)]
        verbose: bool,
        /// Import into .foxing_cas/ CAS store (filesystem-agnostic, no xattr needed)
        #[arg(long)]
        to_cas: bool,
    },
    /// Inspect a .fxar archive without extracting
    Inspect {
        /// Archive file
        archive: String,
        /// Show full file listing
        #[arg(long)]
        list: bool,
        /// Output as JSON
        #[arg(long)]
        json: bool,
        /// Show versions of a specific file
        #[arg(long)]
        file: Option<String>,
    },
    /// Restore specific files/snapshots from a .fxar archive
    Restore {
        /// Archive file
        archive: String,
        /// File path pattern to restore (glob)
        #[arg(long)]
        file: Option<String>,
        /// Snapshot date to restore from (prefix match)
        #[arg(long)]
        date: Option<String>,
        /// Restore most recent version
        #[arg(long)]
        latest: bool,
        /// Restore all versions of matched files
        #[arg(long)]
        all_versions: bool,
        /// Output directory
        #[arg(short, long, default_value = ".")]
        output: String,
    },
    /// Interactive MC-style dual-pane browser for snapshots and archives
    #[cfg(feature = "tui")]
    Browse {
        /// Directory with .foxing_versions or .fxar archive file
        #[arg(default_value = ".")]
        path: String,
        /// Load AI index from archive for vector search (requires --features ai)
        #[arg(long)]
        ai: bool,
    },
}

// -----------------------------------------------------------------------
// CLI entry point (for symlink dispatch from foxingd)
// -----------------------------------------------------------------------

/// Parse CLI args and run  --  used when foxingd is called as `fxcp` via symlink.
pub fn cli_main() -> anyhow::Result<()> {
    // Pre-parse: check if first arg is "snap" for subcommand dispatch
    let args: Vec<String> = std::env::args().collect();
    if args.len() > 1 && args[1] == "snap" {
        return cli_snap_main();
    }

    let cli = FxcpCli::parse();


    let log_filter = if cli.debug { "debug" } else { "info" };
    // Use try_init()  --  subscriber may already be set by the calling binary (e.g. fxcp/main.rs)
    let _ = tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::new(log_filter))
        .with_target(false)
        .try_init();

    // Log detected crypto acceleration (once at startup)
    if cli.checksums.is_some() || cli.generate_sigs || cli.verify {
        crate::hashing::log_crypto_capabilities();
    }

    // Split positional args: all-but-last = sources, last = destination
    let (sources, destination) = crate::filter::split_paths(cli.paths)?;

    // Merge file-based patterns into CLI patterns
    let mut excludes = cli.exclude;
    if let Some(ref file) = cli.exclude_from {
        excludes.extend(crate::filter::read_patterns(file)?);
    }
    let mut includes = cli.include;
    if let Some(ref file) = cli.include_from {
        includes.extend(crate::filter::read_patterns(file)?);
    }

    // Stdin mode: only valid with a single source
    if sources.len() > 1 && sources.iter().any(|s| s.as_os_str() == "-") {
        anyhow::bail!("stdin (-) cannot be used with multiple sources");
    }

    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("Failed to create tokio runtime");

    // Set up progress reporting if requested
    let progress_handle = if cli.progress {
        let tracker = ProgressTracker::new();
        *ACTIVE_PROGRESS.lock() = Some(tracker.clone());
        Some(spawn_progress_reporter(tracker, cli.json))
    } else {
        None
    };

    let mut total_stats = SyncStats::default();
    let sync_start = std::time::Instant::now();

    // S3 source detection  --  sync from S3 to local
    #[cfg(feature = "cloud")]
    {
        if !sources.is_empty() {
            let src_str = sources[0].to_string_lossy();
            if src_str.starts_with("s3://") {
                let (endpoint, bucket, prefix) = s3::parse_s3_url(&src_str)?;
                let dest_path = &destination;
                info!(
                    "S3 source detected: endpoint={}, bucket={}, prefix={} -> {}",
                    endpoint, bucket, prefix, dest_path.display()
                );

                let s3_stats = rt.block_on(s3::sync_from_s3_inner(
                    &endpoint, &bucket, &prefix, dest_path,
                    None, cli.delete,
                ))?;

                if let Some(handle) = progress_handle {
                    if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
                        progress.active.store(false, Ordering::Relaxed);
                    }
                    let _ = handle.join();
                    *ACTIVE_PROGRESS.lock() = None;
                }

                if cli.json {
                    let json_output = serde_json::json!({
                        "status": if s3_stats.errors == 0 { "success" } else { "partial" },
                        "source": "s3",
                        "endpoint": endpoint,
                        "bucket": bucket,
                        "files_synced": s3_stats.files_synced,
                        "bytes_synced": s3_stats.bytes_synced,
                        "dirs_skipped": s3_stats.dirs_skipped,
                        "dirs_synced": s3_stats.dirs_synced,
                        "errors": s3_stats.errors,
                    });
                    println!("{}", serde_json::to_string_pretty(&json_output).unwrap_or_default());
                } else {
                    println!(
                        "S3 sync from: {} files ({} bytes), {} dirs synced, {} dirs skipped, {} errors",
                        s3_stats.files_synced,
                        format_bytes(s3_stats.bytes_synced),
                        s3_stats.dirs_synced,
                        s3_stats.dirs_skipped,
                        s3_stats.errors,
                    );
                }

                if s3_stats.errors > 0 {
                    std::process::exit(1);
                }
                return Ok(());
            }
        }
    }

    // S3 target detection  --  route to S3CasStore instead of POSIX sync
    #[cfg(feature = "cloud")]
    {
        let dest_str = destination.to_string_lossy();
        if dest_str.starts_with("s3://") {
            let (endpoint, bucket, prefix) = s3::parse_s3_url(&dest_str)?;
            info!("S3 target detected: endpoint={}, bucket={}, prefix={}", endpoint, bucket, prefix);

            let s3_stats = rt.block_on(s3::sync_to_s3(
                &sources, &endpoint, &bucket, &prefix, cli.normalization_level,
            ))?;

            total_stats.files_copied = s3_stats.files_copied;
            total_stats.bytes_copied = s3_stats.bytes_copied;
            total_stats.errors = s3_stats.errors;

            if let Some(handle) = progress_handle {
                if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
                    progress.active.store(false, Ordering::Relaxed);
                }
                let _ = handle.join();
                *ACTIVE_PROGRESS.lock() = None;
            }

            if cli.json {
                let json_output = serde_json::json!({
                    "status": if total_stats.errors == 0 { "success" } else { "partial" },
                    "target": "s3",
                    "endpoint": endpoint,
                    "bucket": bucket,
                    "files_copied": total_stats.files_copied,
                    "bytes_copied": total_stats.bytes_copied,
                    "errors": total_stats.errors,
                });
                println!("{}", serde_json::to_string_pretty(&json_output).unwrap_or_default());
            } else {
                println!(
                    "S3 sync: {} files, {} bytes, {} errors",
                    total_stats.files_copied, total_stats.bytes_copied, total_stats.errors,
                );
            }

            if total_stats.errors > 0 {
                std::process::exit(1);
            }
            return Ok(());
        }
    }

    for source in &sources {
        let opts = SyncOptions {
            source: source.clone(),
            destination: destination.clone(),
            archive: cli.archive,
            recursive: cli.recursive || cli.archive,
            delete: cli.delete,
            dry_run: cli.dry_run,
            exclude: excludes.clone(),
            include: includes.clone(),
            generate_sigs: cli.generate_sigs,
            checksums: cli.checksums.as_deref().map(hashing::ChecksumType::parse_list).unwrap_or_default(),
            checksums_overwrite: cli.checksums_overwrite,
            cid_output: cli.cid,
            snapshot: cli.snapshot,
            throttle: cli.throttle,
            cleanup: cli.cleanup,
            size: cli.size,
            checkpoint_interval: cli.checkpoint_interval,
            checkpoint_keep: cli.checkpoint_keep,
            zero_copy: cli.zero_copy,
            verify: cli.verify,
            owner: cli.archive || cli.owner,
            group: cli.archive || cli.group,
            perms: cli.archive || cli.perms,
            xattrs: cli.archive || cli.xattrs,
            acls: cli.acls,
        };

        match rt.block_on(run(opts)) {
            Ok(stats) => total_stats.merge(&stats),
            Err(e) => {
                error!("fxcp failed for {}: {}", source.display(), e);
                total_stats.errors += 1;
            }
        }
    }

    // Stop progress reporter
    if let Some(handle) = progress_handle {
        if let Some(ref progress) = *ACTIVE_PROGRESS.lock() {
            progress.active.store(false, Ordering::Relaxed);
        }
        let _ = handle.join();
        *ACTIVE_PROGRESS.lock() = None;
    }

    if cli.json {
        // Machine-readable JSON output
        let _total_files = total_stats.files_copied + total_stats.files_skipped;
        let status = if total_stats.errors == 0 { "success" }
                     else if total_stats.files_copied > 0 { "partial" }
                     else { "failed" };
        let json_output = serde_json::json!({
            "status": status,
            "files_copied": total_stats.files_copied,
            "files_reflinked": total_stats.files_reflinked,
            "files_skipped": total_stats.files_skipped,
            "files_deleted": total_stats.files_deleted,
            "symlinks_replicated": total_stats.symlinks_replicated,
            "dirs_created": total_stats.dirs_created,
            "dirs_pruned": total_stats.dirs_pruned,
            "bytes_copied": total_stats.bytes_copied,
            "bytes_reflinked": total_stats.bytes_reflinked,
            "errors": total_stats.errors,
            "method_breakdown": {
                "reflink": total_stats.files_reflinked,
                "copy_file_range": total_stats.files_cfr,
                "sendfile": total_stats.files_small,
                "io_uring": total_stats.files_copied.saturating_sub(
                    total_stats.files_reflinked + total_stats.files_cfr + total_stats.files_small
                ),
            },
        });
        println!("{}", serde_json::to_string_pretty(&json_output).unwrap_or_default());
    } else {
        print_summary(&total_stats, sync_start.elapsed());
    }

    // Granular exit codes
    if total_stats.errors > 0 {
        if total_stats.files_copied > 0 {
            std::process::exit(1);  // Partial success
        } else {
            std::process::exit(2);  // Complete failure
        }
    }
    Ok(())
}

/// Snapshot management subcommand handler.
fn cli_snap_main() -> anyhow::Result<()> {
    // Re-parse with snap-aware CLI (skip argv[0], "snap" is the subcommand)
    let snap_cli = FxcpSnapCli::parse_from(
        std::iter::once("fxcp-snap".to_string())
            .chain(std::env::args().skip(2))
    );

    let log_filter = if snap_cli.debug { "debug" } else { "info" };
    let _ = tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::new(log_filter))
        .with_target(false)
        .try_init();

    let _rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("Failed to create tokio runtime");

    match snap_cli.command {
        SnapCommand::List { path, json } => handle_snap_list(path, json)?,
        SnapCommand::Show { timestamp } => {
            let store = crate::version_store::VersionStore::open(&std::env::current_dir().unwrap_or_default());
            let snapshots = store.list_snapshots();
            if let Some(snap) = snapshots.iter().find(|s| s.timestamp.contains(&timestamp)) {
                println!("{}", serde_json::to_string_pretty(snap).unwrap_or_default());
            } else {
                anyhow::bail!("Snapshot not found: {}", timestamp);
            }
        }
        SnapCommand::Revert { path, epoch } => {
            let p = std::path::PathBuf::from(&path);
            crate::versioning::revert_file(&p, epoch)?;
            info!("Reverted {:?} to epoch {}", path, epoch);
        }
        SnapCommand::Copy { path, epoch, destination } => {
            let p = std::path::PathBuf::from(&path);
            let d = std::path::PathBuf::from(&destination);
            crate::versioning::copy_version_to_path(&p, epoch, &d)?;
            info!("Copied version {} of {:?} to {:?}", epoch, path, destination);
        }
        SnapCommand::Prune { older_than, keep_last, max_size, path } => {
            let p = std::path::PathBuf::from(&path);
            let store = crate::version_store::VersionStore::open(&p);
            let mut total = crate::version_store::PruneStats::default();

            if let Some(ref age_str) = older_than {
                let duration = parse_duration(age_str)?;
                let stats = store.prune_by_age(duration)?;
                total.snapshots_removed += stats.snapshots_removed;
                total.bytes_freed += stats.bytes_freed;
            }
            if let Some(count) = keep_last {
                let stats = store.prune_by_count(count)?;
                total.snapshots_removed += stats.snapshots_removed;
                total.bytes_freed += stats.bytes_freed;
            }
            if let Some(ref size_str) = max_size {
                let bytes = parse_size(size_str)?;
                let stats = store.prune_by_size(bytes)?;
                total.snapshots_removed += stats.snapshots_removed;
                total.bytes_freed += stats.bytes_freed;
            }

            if total.snapshots_removed > 0 {
                info!("Pruned {} snapshots, freed {} bytes", total.snapshots_removed, total.bytes_freed);
            } else {
                info!("Nothing to prune.");
            }
        }
        SnapCommand::Tag { timestamp, tag } => {
            let cwd = std::env::current_dir().unwrap_or_default();
            let store = crate::version_store::VersionStore::open(&cwd);
            store.tag_snapshot(&timestamp, &tag)?;
            info!("Tagged snapshot {} as '{}'", timestamp, tag);
        }
        SnapCommand::RebuildIndex { path } => {
            let p = std::path::PathBuf::from(&path);
            let store = crate::version_store::VersionStore::open(&p);
            let index = store.rebuild_index()?;
            info!("Rebuilt index: {} snapshots", index.snapshots.len());
        }
        SnapCommand::Stats { path, json } => {
            let p = std::path::PathBuf::from(&path);
            let store = crate::version_store::VersionStore::open(&p);
            let snapshots = store.list_snapshots();
            let versions_root = p.join(".foxing_versions");
            let stats = crate::version_store::compute_store_stats(
                &snapshots,
                if versions_root.exists() { Some(&versions_root) } else { None }
            );
            if json {
                println!("{}", serde_json::to_string_pretty(&stats).unwrap_or_default());
            } else {
                crate::version_store::print_store_stats(&stats, &p);
            }
        }
        SnapCommand::Export { path, output, compress, timestamp, format, include_vectors, from_cas } => {
            handle_snap_export(&path, output.as_deref(), &compress, timestamp.as_deref(), &format, include_vectors, from_cas)?;
        }
        SnapCommand::Import { path, input, compress, generate_sigs, raw, verbose, to_cas } => {
            handle_snap_import(&path, input.as_deref(), &compress, generate_sigs, raw, verbose, to_cas)?;
        }
        SnapCommand::Inspect { archive, list, json, file } => {
            handle_snap_inspect(&archive, list, json, file.as_deref())?;
        }
        SnapCommand::Restore { archive, file, date, latest, all_versions: _, output } => {
            handle_snap_restore(&archive, file.as_deref(), date.as_deref(), latest, &output)?;
        }
        #[cfg(feature = "tui")]
        SnapCommand::Browse { path, ai: _ } => {
            let p = std::path::PathBuf::from(&path);
            let is_fxar = path.ends_with(".fxar") || {
                std::fs::File::open(&p).ok()
                    .and_then(|mut f| {
                        let mut magic = [0u8; 4];
                        use std::io::Read;
                        f.read_exact(&mut magic).ok()?;
                        Some(magic == *b"FXAR")
                    })
                    .unwrap_or(false)
            };
            if is_fxar {
                let mut app = crate::browser::BrowserApp::new_archive(&p, &p)
                    .map_err(|e| anyhow::anyhow!("Failed to open archive: {}", e))?;
                app.run().map_err(|e| anyhow::anyhow!("TUI error: {}", e))?;
            } else {
                let mut app = crate::browser::BrowserApp::new(&p);
                app.run().map_err(|e| anyhow::anyhow!("TUI error: {}", e))?;
            }
        }
    }
    Ok(())
}

/// Handle `fxcp snap list`  --  list snapshots or file versions.
fn handle_snap_list(path: Option<String>, json: bool) -> anyhow::Result<()> {
    if let Some(ref p) = path {
        let p = std::path::PathBuf::from(p);
        if p.is_file() {
            let versions = crate::versioning::list_versions(&p)?;
            if json {
                let json_versions: Vec<serde_json::Value> = versions.iter().map(|v| {
                    serde_json::json!({
                        "epoch": v.epoch_seq,
                        "timestamp": v.timestamp,
                        "size": v.size,
                        "path": v.path.to_string_lossy(),
                    })
                }).collect();
                println!("{}", serde_json::to_string_pretty(&json_versions).unwrap_or_default());
            } else if versions.is_empty() {
                println!("No versions found for {:?}", p);
            } else {
                crate::versioning::print_versions_table(versions, 50);
            }
        } else {
            let store = crate::version_store::VersionStore::open(&p);
            let snapshots = store.list_snapshots();
            if json {
                println!("{}", serde_json::to_string_pretty(&snapshots).unwrap_or_default());
            } else {
                crate::version_store::print_snapshot_table(&snapshots);
            }
        }
    } else {
        let store = crate::version_store::VersionStore::open(&std::env::current_dir().unwrap_or_default());
        let snapshots = store.list_snapshots();
        if json {
            println!("{}", serde_json::to_string_pretty(&snapshots).unwrap_or_default());
        } else {
            crate::version_store::print_snapshot_table(&snapshots);
        }
    }
    Ok(())
}

/// Handle `fxcp snap export`  --  export as FXAR v2 or legacy tar.
///
/// Auto-detects source type:
/// - If `{path}/.foxing_versions/` exists -> export from VersionStore (snapshot-aware)
/// - Otherwise -> export directory directly (bare paths, no snapshot prefix)
fn handle_snap_export(
    path: &str,
    output: Option<&str>,
    compress: &str,
    timestamp: Option<&str>,
    format: &str,
    include_vectors: bool,
    from_cas: bool,
) -> anyhow::Result<()> {
    let p = std::path::PathBuf::from(path);

    // --from-cas: export from .foxing_cas/ CAS store
    if from_cas {
        let cas = crate::cas_store::CasStore::open(&p)?;
        let out_path = output.ok_or_else(|| anyhow::anyhow!("--from-cas requires -o <output.fxar>"))?;
        let tmp_path = format!("{}.tmp", out_path);
        let file = std::fs::File::create(&tmp_path)
            .map_err(|e| anyhow::anyhow!("Cannot create {}: {}", tmp_path, e))?;
        let stats = cas.export_fxar(file, compress)?;
        std::fs::rename(&tmp_path, out_path)
            .map_err(|e| anyhow::anyhow!("Atomic rename {} -> {}: {}", tmp_path, out_path, e))?;
        info!("Exported from CAS to {} (FXAR v2): {} files, {} unique chunks, {:.1}% dedup",
              out_path, stats.total_files, stats.unique_chunks, stats.dedup_ratio() * 100.0);
        return Ok(());
    }

    let has_version_store = p.join(".foxing_versions").is_dir();

    if format == "tar" {
        if !has_version_store {
            anyhow::bail!("tar format requires .foxing_versions/ (use fxar2 format for direct directory export)");
        }
        let store = crate::version_store::VersionStore::open(&p);
        if let Some(out_path) = output {
            let file = std::fs::File::create(out_path)
                .map_err(|e| anyhow::anyhow!("Cannot create {}: {}", out_path, e))?;
            let stats = store.export(file, compress, timestamp)?;
            info!("Exported to {} (tar): {} snapshots, {} unique files",
                  out_path, stats.snapshots_exported, stats.unique_chunks);
        } else {
            let stdout = std::io::stdout().lock();
            let stats = store.export(stdout, compress, timestamp)?;
            eprintln!("Exported {} snapshots, {} unique files ({} deduped)",
                      stats.snapshots_exported, stats.unique_chunks, stats.dedup_chunks);
        }
    } else if has_version_store {
        let store = crate::version_store::VersionStore::open(&p);
        let index_dir = p.join(".foxing_index");
        let ai_index_dir = if include_vectors && index_dir.is_dir() {
            Some(index_dir.as_path())
        } else {
            None
        };

        if let Some(out_path) = output {
            let tmp_path = format!("{}.tmp", out_path);
            let file = std::fs::File::create(&tmp_path)
                .map_err(|e| anyhow::anyhow!("Cannot create {}: {}", tmp_path, e))?;
            let stats = if ai_index_dir.is_some() {
                crate::fxar::write_archive_with_ai_index(&store, file, compress, timestamp, ai_index_dir, include_vectors)?
            } else {
                crate::fxar::write_archive_seekable(&store, file, compress, timestamp)?
            };
            std::fs::rename(&tmp_path, out_path)
                .map_err(|e| anyhow::anyhow!("Atomic rename {} -> {}: {}", tmp_path, out_path, e))?;
            info!("Exported to {} (FXAR v2): {} snapshots, {} files, {} unique chunks, {:.1}% dedup",
                  out_path, stats.snapshots_exported, stats.total_files,
                  stats.unique_chunks, stats.dedup_ratio() * 100.0);
        } else {
            let stdout = std::io::stdout().lock();
            let stats = crate::fxar::write_archive(&store, stdout, compress, timestamp)?;
            eprintln!("Exported {} snapshots, {} files ({} unique chunks, {:.1}% dedup)",
                      stats.snapshots_exported, stats.total_files,
                      stats.unique_chunks, stats.dedup_ratio() * 100.0);
        }
    } else {
        // Direct directory export  --  no VersionStore required
        if let Some(out_path) = output {
            let tmp_path = format!("{}.tmp", out_path);
            let file = std::fs::File::create(&tmp_path)
                .map_err(|e| anyhow::anyhow!("Cannot create {}: {}", tmp_path, e))?;
            let stats = crate::fxar::write_archive_from_directory(&p, file, compress)?;
            std::fs::rename(&tmp_path, out_path)
                .map_err(|e| anyhow::anyhow!("Atomic rename {} -> {}: {}", tmp_path, out_path, e))?;
            info!("Exported to {} (FXAR v2, directory): {} files, {} unique chunks, {:.1}% dedup",
                  out_path, stats.total_files, stats.unique_chunks, stats.dedup_ratio() * 100.0);
        } else {
            let stdout = std::io::stdout().lock();
            let stats = crate::fxar::write_archive_from_directory_stream(&p, stdout, compress)?;
            eprintln!("Exported {} files ({} unique chunks, {:.1}% dedup)",
                      stats.total_files, stats.unique_chunks, stats.dedup_ratio() * 100.0);
        }
    }
    Ok(())
}

/// Handle `fxcp snap import`  --  import from FXAR v2 or legacy tar (file or stdin).
///
/// Default: `restore_latest` (flat paths, latest version of each file, like tar -xvf).
/// With `--raw`: `restore_all` (full snapshot/tree/ path structure preserved).
fn handle_snap_import(
    path: &str,
    input: Option<&str>,
    compress: &str,
    generate_sigs: bool,
    raw: bool,
    verbose: bool,
    to_cas: bool,
) -> anyhow::Result<()> {
    let p = std::path::PathBuf::from(path);

    // --to-cas: import FXAR into .foxing_cas/ CAS store (filesystem-agnostic)
    if to_cas {
        let in_path = input.ok_or_else(|| anyhow::anyhow!("--to-cas requires -i <archive.fxar>"))?;
        let file = std::fs::File::open(in_path)
            .map_err(|e| anyhow::anyhow!("Cannot open {}: {}", in_path, e))?;
        let cas = crate::cas_store::CasStore::open(&p)?;
        let stats = cas.import_fxar(file)?;
        info!("Imported to CAS from {} (FXAR): {} chunks written, {} deduped",
              in_path, stats.chunks_written, stats.chunks_deduped);
        return Ok(());
    }

    if let Some(in_path) = input {
        let mut file = std::fs::File::open(in_path)
            .map_err(|e| anyhow::anyhow!("Cannot open {}: {}", in_path, e))?;
        let mut magic = [0u8; 4];
        use std::io::Read;
        file.read_exact(&mut magic)
            .map_err(|e| anyhow::anyhow!("Cannot read {}: {}", in_path, e))?;

        if &magic == b"FXAR" {
            drop(file);
            let file = std::fs::File::open(in_path)
                .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", in_path, e))?;
            let mut reader = crate::fxar::FxarReader::open(file)
                .map_err(|e| anyhow::anyhow!("FXAR open: {}", e))?;
            let stats = if raw {
                reader.restore_all(&p, generate_sigs)
                    .map_err(|e| anyhow::anyhow!("FXAR import (raw): {}", e))?
            } else {
                reader.restore_latest(&p, generate_sigs, verbose)
                    .map_err(|e| anyhow::anyhow!("FXAR import: {}", e))?
            };
            info!("Imported from {} (FXAR v2): {} files, {} bytes",
                  in_path, stats.files_restored, stats.bytes_restored);
        } else {
            drop(file);
            let file = std::fs::File::open(in_path)
                .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", in_path, e))?;
            let store = crate::version_store::VersionStore::open(&p);
            let comp = if compress == "auto" {
                if in_path.ends_with(".zst") || in_path.ends_with(".zstd") { "zstd" }
                else if in_path.ends_with(".lz4") { "lz4" }
                else if in_path.ends_with(".gz") { "gzip" }
                else if in_path.ends_with(".xz") { "xz" }
                else { "none" }
            } else { compress };
            let stats = store.import(file, comp)?;
            info!("Imported from {} (tar): {} files restored", in_path, stats.files_restored);
        }
    } else {
        let stdin = std::io::stdin().lock();
        let mut buf_reader = std::io::BufReader::new(stdin);
        let mut magic = [0u8; 4];
        use std::io::Read;
        buf_reader.read_exact(&mut magic)
            .map_err(|e| anyhow::anyhow!("Cannot read stdin: {}", e))?;

        if &magic == b"FXAR" {
            if raw {
                let chain = std::io::Cursor::new(magic.to_vec()).chain(buf_reader);
                let stats = crate::fxar::read_archive_stream(chain, &p, generate_sigs)
                    .map_err(|e| anyhow::anyhow!("FXAR stream import (raw): {}", e))?;
                info!("Imported from stdin (FXAR v2, raw): {} files", stats.files_restored);
            } else {
                // restore_latest needs Seek  --  spool stdin to a temp file
                let tmp_path = std::env::temp_dir().join(format!("fxcp-import-{}.fxar", std::process::id()));
                {
                    let mut tmp = std::fs::File::create(&tmp_path)
                        .map_err(|e| anyhow::anyhow!("Cannot create temp file: {}", e))?;
                    use std::io::Write;
                    tmp.write_all(&magic)?;
                    std::io::copy(&mut buf_reader, &mut tmp)?;
                }
                let tmp_file = std::fs::File::open(&tmp_path)
                    .map_err(|e| anyhow::anyhow!("Cannot reopen temp file: {}", e))?;
                let mut reader = crate::fxar::FxarReader::open(tmp_file)
                    .map_err(|e| anyhow::anyhow!("FXAR open: {}", e))?;
                let stats = reader.restore_latest(&p, generate_sigs, verbose)
                    .map_err(|e| anyhow::anyhow!("FXAR stream import: {}", e))?;
                let _ = std::fs::remove_file(&tmp_path);
                info!("Imported from stdin (FXAR v2): {} files", stats.files_restored);
            }
        } else {
            let store = crate::version_store::VersionStore::open(&p);
            let comp = if compress == "auto" { "none" } else { compress };
            let chain = std::io::Cursor::new(magic.to_vec()).chain(buf_reader);
            let stats = store.import(chain, comp)?;
            info!("Imported from stdin (tar): {} files", stats.files_restored);
        }
    }
    Ok(())
}

/// Handle `fxcp snap inspect`  --  inspect archive contents without extracting.
fn handle_snap_inspect(
    archive: &str,
    list: bool,
    json: bool,
    file: Option<&str>,
) -> anyhow::Result<()> {
    let mut f = std::fs::File::open(archive)
        .map_err(|e| anyhow::anyhow!("Cannot open {}: {}", archive, e))?;
    let mut magic = [0u8; 4];
    use std::io::Read;
    f.read_exact(&mut magic)
        .map_err(|e| anyhow::anyhow!("Cannot read {}: {}", archive, e))?;
    drop(f);

    if &magic == b"FXAR" {
        let f = std::fs::File::open(archive)
            .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", archive, e))?;
        let result = crate::fxar::inspect_archive(f)
            .map_err(|e| anyhow::anyhow!("FXAR inspect: {}", e))?;

        let filtered: Vec<_> = if let Some(pattern) = file {
            result.files.iter().filter(|f| f.path.contains(pattern)).collect()
        } else {
            result.files.iter().collect()
        };

        if json {
            println!("{}", serde_json::to_string_pretty(&result).unwrap_or_default());
        } else if list {
            for f in &filtered {
                println!("{:>10}  {}  [{}]", f.size, f.path, &f.blake3[..16]);
            }
            println!("\n{} files, {} chunks ({:.1}% dedup ratio)",
                     filtered.len(), result.chunk_count, result.dedup_ratio * 100.0);
        } else {
            println!("Archive: {} (FXAR v2)", archive);
            println!("  Format:      FXAR v{}", result.version);
            println!("  Snapshots:   {}", result.snapshots.len());
            println!("  Files:       {}", result.total_files);
            println!("  Unique files:{}", result.unique_files);
            println!("  Chunks:      {}", result.chunk_count);
            println!("  Apparent:    {} bytes", result.total_apparent_bytes);
            println!("  Chunk data:  {} bytes", result.total_chunk_bytes);
            println!("  Compressed:  {} bytes", result.total_compressed_bytes);
            println!("  Dedup ratio: {:.1}%", result.dedup_ratio * 100.0);
            for snap in &result.snapshots {
                println!("  Snapshot:    {}", snap);
            }
        }
    } else {
        let f = std::fs::File::open(archive)
            .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", archive, e))?;
        let comp = if archive.ends_with(".zst") { "zstd" }
                   else if archive.ends_with(".lz4") { "lz4" }
                   else if archive.ends_with(".gz") { "gzip" }
                   else if archive.ends_with(".xz") { "xz" }
                   else { "none" };
        let entries = crate::version_store::VersionStore::inspect_archive(f, comp)?;

        let filtered: Vec<_> = if let Some(pattern) = file {
            entries.into_iter().filter(|e| e.path.contains(pattern)).collect()
        } else {
            entries
        };

        if json {
            println!("{}", serde_json::to_string_pretty(&filtered).unwrap_or_default());
        } else if list {
            for e in &filtered {
                let marker = if e.is_dedup_ref { " [dedup]" }
                             else if e.is_metadata { " [meta]" }
                             else { "" };
                println!("{:>10}  {}{}", e.size, e.path, marker);
            }
            let total_files = filtered.iter().filter(|e| !e.is_dedup_ref && !e.is_metadata).count();
            let dedup_refs = filtered.iter().filter(|e| e.is_dedup_ref).count();
            println!("\n{} files, {} dedup references", total_files, dedup_refs);
        } else {
            let total_files = filtered.iter().filter(|e| !e.is_dedup_ref && !e.is_metadata).count();
            let dedup_refs = filtered.iter().filter(|e| e.is_dedup_ref).count();
            let total_size: u64 = filtered.iter().map(|e| e.size).sum();
            println!("Archive: {} (tar)", archive);
            println!("  Files:       {}", total_files);
            println!("  Dedup refs:  {}", dedup_refs);
            println!("  Total size:  {} bytes", total_size);
        }
    }
    Ok(())
}

/// Handle `fxcp snap restore`  --  restore specific files from an archive.
fn handle_snap_restore(
    archive: &str,
    file: Option<&str>,
    date: Option<&str>,
    latest: bool,
    output: &str,
) -> anyhow::Result<()> {
    let mut f = std::fs::File::open(archive)
        .map_err(|e| anyhow::anyhow!("Cannot open {}: {}", archive, e))?;
    let mut magic = [0u8; 4];
    use std::io::Read;
    f.read_exact(&mut magic)
        .map_err(|e| anyhow::anyhow!("Cannot read {}: {}", archive, e))?;
    drop(f);

    let out_dir = std::path::PathBuf::from(output);

    if &magic == b"FXAR" {
        let f = std::fs::File::open(archive)
            .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", archive, e))?;
        let mut reader = crate::fxar::FxarReader::open(f)
            .map_err(|e| anyhow::anyhow!("FXAR open: {}", e))?;
        let manifest = reader.read_manifest()
            .map_err(|e| anyhow::anyhow!("FXAR manifest: {}", e))?;

        let mut matched: Vec<_> = manifest.files.iter()
            .filter(|e| {
                if let Some(pattern) = file {
                    glob::Pattern::new(pattern).map(|p| p.matches(&e.path)).unwrap_or(false)
                        || e.path.contains(pattern)
                } else { true }
            })
            .filter(|e| {
                if let Some(d) = date { e.path.contains(d) }
                else { true }
            })
            .cloned()
            .collect();

        if latest {
            matched.sort_by(|a, b| b.path.cmp(&a.path));
            matched.truncate(1);
        }

        if matched.is_empty() {
            anyhow::bail!("No matching files found in archive");
        }

        info!("Restoring {} files to {}", matched.len(), output);

        for entry in &matched {
            let data = reader.restore_file(&entry.path)
                .map_err(|e| anyhow::anyhow!("FXAR restore {}: {}", entry.path, e))?;
            let dest = out_dir.join(std::path::Path::new(&entry.path).file_name().unwrap_or_default());
            if let Some(parent) = dest.parent() {
                let _ = std::fs::create_dir_all(parent);
            }
            std::fs::write(&dest, &data)?;

            // Restore permissions, ownership, mtime, and xattrs from manifest
            use std::os::unix::fs::PermissionsExt;
            let _ = std::fs::set_permissions(&dest, std::fs::Permissions::from_mode(entry.mode));
            if unsafe { libc::geteuid() } == 0 {
                use nix::unistd::{chown, Uid, Gid};
                let _ = chown(&dest, Some(Uid::from_raw(entry.uid)), Some(Gid::from_raw(entry.gid)));
            }
            let mtime = filetime::FileTime::from_unix_time(entry.mtime, 0);
            let _ = filetime::set_file_mtime(&dest, mtime);

            for (key, hex_val) in &entry.xattr {
                if let Ok(val) = hex::decode(hex_val) {
                    if key.starts_with("user.foxing") {
                        let _ = crate::sidecar::set_metadata(&dest, key, &val);
                    } else {
                        let _ = xattr::set(&dest, key, &val);
                    }
                }
            }

            info!("Restored: {} ({} bytes, BLAKE3 verified)", dest.display(), data.len());
        }
    } else {
        let f = std::fs::File::open(archive)
            .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", archive, e))?;
        let comp = if archive.ends_with(".zst") { "zstd" }
                   else if archive.ends_with(".lz4") { "lz4" }
                   else if archive.ends_with(".gz") { "gzip" }
                   else if archive.ends_with(".xz") { "xz" }
                   else { "none" };
        let entries = crate::version_store::VersionStore::inspect_archive(f, comp)?;

        let mut matched: Vec<_> = entries.into_iter()
            .filter(|e| !e.is_metadata && !e.is_dedup_ref)
            .filter(|e| {
                if let Some(pattern) = file {
                    glob::Pattern::new(pattern).map(|p| p.matches(&e.path)).unwrap_or(false)
                        || e.path.contains(pattern)
                } else { true }
            })
            .filter(|e| {
                if let Some(d) = date { e.path.contains(d) }
                else { true }
            })
            .collect();

        if latest {
            matched.sort_by(|a, b| b.path.cmp(&a.path));
            matched.truncate(1);
        }

        if matched.is_empty() {
            anyhow::bail!("No matching files found in archive");
        }

        info!("Restoring {} files to {}", matched.len(), output);

        let f2 = std::fs::File::open(archive)
            .map_err(|e| anyhow::anyhow!("Cannot reopen {}: {}", archive, e))?;
        let decompressed = crate::version_store::wrap_import_decompressor(f2, comp);
        let mut tar_archive = tar::Archive::new(decompressed);
        let match_paths: std::collections::HashSet<String> = matched.iter().map(|e| e.path.clone()).collect();

        for entry in tar_archive.entries().map_err(|e| anyhow::anyhow!("tar error: {}", e))? {
            let mut entry = match entry { Ok(e) => e, Err(_) => continue };
            let path = entry.path().map(|p| p.to_string_lossy().into_owned()).unwrap_or_default();
            if match_paths.contains(&path) {
                let dest = out_dir.join(std::path::Path::new(&path).file_name().unwrap_or_default());
                if let Some(parent) = dest.parent() {
                    let _ = std::fs::create_dir_all(parent);
                }
                entry.unpack(&dest).map_err(|e| anyhow::anyhow!("unpack error: {}", e))?;
                info!("Restored: {}", dest.display());
            }
        }
    }
    Ok(())
}

/// Parse a human-readable duration string (e.g., "30d", "7d", "24h", "2w").
fn parse_duration(s: &str) -> anyhow::Result<std::time::Duration> {
    let s = s.trim();
    let (num_str, unit) = s.split_at(s.len().saturating_sub(1));
    let num: u64 = num_str.parse().map_err(|_| anyhow::anyhow!("Invalid duration: {}", s))?;
    let secs = match unit {
        "s" => num,
        "m" => num * 60,
        "h" => num * 3600,
        "d" => num * 86400,
        "w" => num * 604800,
        _ => anyhow::bail!("Unknown duration unit '{}' (use s/m/h/d/w)", unit),
    };
    Ok(std::time::Duration::from_secs(secs))
}

/// Parse a human-readable size string (e.g., "50G", "10G", "500M").
fn parse_size(s: &str) -> anyhow::Result<u64> {
    let s = s.trim();
    let (num_str, unit) = s.split_at(s.len().saturating_sub(1));
    let num: u64 = num_str.parse().map_err(|_| anyhow::anyhow!("Invalid size: {}", s))?;
    let bytes = match unit.to_uppercase().as_str() {
        "K" => num * 1024,
        "M" => num * 1024 * 1024,
        "G" => num * 1024 * 1024 * 1024,
        "T" => num * 1024 * 1024 * 1024 * 1024,
        _ => anyhow::bail!("Unknown size unit '{}' (use K/M/G/T)", unit),
    };
    Ok(bytes)
}

// -----------------------------------------------------------------------
// Formatting & display helpers
// -----------------------------------------------------------------------

fn format_bytes(b: u64) -> String {
    crate::fmt::format_size(b)
}

/// Print a human-readable summary of sync results.
pub fn print_summary(stats: &SyncStats, elapsed: std::time::Duration) {
    #[allow(unused_mut)]
    let mut total_bytes = stats.bytes_copied + stats.bytes_reflinked + stats.bytes_cfr + stats.bytes_small + stats.bytes_delta;
    #[allow(unused_mut)]
    let mut total_files = stats.files_copied + stats.files_reflinked + stats.files_cfr + stats.files_small + stats.files_delta;
    #[cfg(feature = "nfs-bypass")]
    {
        total_bytes += stats.bytes_nfs_bypass;
        total_files += stats.files_nfs_bypass;
    }

    let secs = elapsed.as_secs_f64();
    let throughput_mb = if secs > 0.0 { total_bytes as f64 / 1_048_576.0 / secs } else { 0.0 };

    println!("fxcp sync complete:");
    println!("  Elapsed:       {}", format_duration(elapsed));
    println!("  Files synced:  {}", total_files);
    #[cfg(feature = "nfs-bypass")]
    if stats.files_nfs_bypass > 0 {
        println!("  - NFS bypass:  {} ({}, compound RPC)", stats.files_nfs_bypass, format_bytes(stats.bytes_nfs_bypass));
    }
    if stats.files_reflinked > 0 {
        println!("  - reflink:     {} ({}, instant CoW)", stats.files_reflinked, format_bytes(stats.bytes_reflinked));
    }
    if stats.files_cfr > 0 {
        println!("  - server copy: {} ({}, copy_file_range)", stats.files_cfr, format_bytes(stats.bytes_cfr));
    }
    if stats.files_small > 0 {
        println!("  - sendfile:    {} ({})", stats.files_small, format_bytes(stats.bytes_small));
    }
    if stats.files_copied > 0 {
        println!("  - io_uring:    {} ({})", stats.files_copied, format_bytes(stats.bytes_copied));
    }
    if stats.files_delta > 0 {
        println!("  - delta:       {} ({}, chunk-level)", stats.files_delta, format_bytes(stats.bytes_delta));
    }
    if stats.files_skipped > 0 { println!("  Skipped:       {} (unchanged)", stats.files_skipped); }
    if stats.dirs_pruned > 0 { println!("  Dirs pruned:   {} (hash match)", stats.dirs_pruned); }
    if stats.symlinks_replicated > 0 { println!("  Symlinks:      {}", stats.symlinks_replicated); }
    if stats.files_deleted > 0 { println!("  Deleted:       {}", stats.files_deleted); }
    if stats.dirs_created > 0 { println!("  Dirs created:  {}", stats.dirs_created); }
    if stats.dirs_hashed > 0 { println!("  Dirs hashed:   {}", stats.dirs_hashed); }
    if stats.sigs_stored > 0 { println!("  Sigs stored:   {} (foxingd-compatible)", stats.sigs_stored); }
    println!("  Total:         {} in {} ({:.0} MB/s)",
             format_bytes(total_bytes), format_duration(elapsed), throughput_mb);
    if stats.files_verified > 0 { println!("  Verified:      {} (BLAKE3)", stats.files_verified); }
    if stats.verify_failures > 0 { println!("  Verify FAIL:   {}", stats.verify_failures); }
    if stats.errors > 0 { println!("  Errors:        {}", stats.errors); }
}

fn format_duration(d: std::time::Duration) -> String {
    let secs = d.as_secs();
    if secs >= 3600 {
        format!("{}h{:02}m{:02}s", secs / 3600, (secs % 3600) / 60, secs % 60)
    } else if secs >= 60 {
        format!("{}m{:02}.{:01}s", secs / 60, secs % 60, d.subsec_millis() / 100)
    } else if secs > 0 {
        format!("{}.{:02}s", secs, d.subsec_millis() / 10)
    } else {
        format!("{}ms", d.as_millis())
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    /// Helper: create a minimal snapshot directory with summary.json (no tag).
    fn create_test_snapshot(root: &std::path::Path, timestamp: &str) {
        let versions_dir = root.join(".foxing_versions");
        let snap_dir = versions_dir.join(timestamp);
        let tree_dir = snap_dir.join("tree");
        std::fs::create_dir_all(&tree_dir).unwrap();
        let summary = crate::version_store::SnapshotSummary {
            timestamp: timestamp.to_string(),
            status: "success".into(),
            snap_type: "full".into(),
            tag: None,
            source: "/tmp/src".into(),
            trigger: "test".into(),
            files: 1,
            size_bytes: 100,
            disk_usage_bytes: 100,
            savings_pct: 0.0,
            reference: None,
            elapsed_ms: 10,
        };
        let json = serde_json::to_string_pretty(&summary).unwrap();
        std::fs::write(snap_dir.join("summary.json"), json).unwrap();
    }

    #[test]
    fn test_snap_tag_persists() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();
        let ts = "2026-04-25T120000";
        create_test_snapshot(root, ts);

        // Tag the snapshot
        let store = crate::version_store::VersionStore::open(root);
        store.tag_snapshot(ts, "pre-migration").unwrap();

        // Read summary.json back and verify tag is present
        let summary_path = root.join(".foxing_versions").join(ts).join("summary.json");
        let data = std::fs::read_to_string(&summary_path).unwrap();
        let summary: crate::version_store::SnapshotSummary = serde_json::from_str(&data).unwrap();
        assert_eq!(summary.tag, Some("pre-migration".to_string()));
    }

    #[test]
    fn test_tagged_snapshot_survives_prune() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path();

        // Create two snapshots with old timestamps
        let ts_tagged = "2024-01-01T000000";
        let ts_untagged = "2024-01-02T000000";
        create_test_snapshot(root, ts_tagged);
        create_test_snapshot(root, ts_untagged);

        // Tag one snapshot
        let store = crate::version_store::VersionStore::open(root);
        store.tag_snapshot(ts_tagged, "keep-me").unwrap();

        // Prune by count (keep only 0  --  should delete untagged, keep tagged)
        let stats = store.prune_by_count(0).unwrap();
        assert_eq!(stats.snapshots_removed, 1);

        // Tagged snapshot directory must still exist
        let tagged_dir = root.join(".foxing_versions").join(ts_tagged);
        assert!(tagged_dir.exists(), "Tagged snapshot should survive prune");

        // Untagged snapshot directory must be gone
        let untagged_dir = root.join(".foxing_versions").join(ts_untagged);
        assert!(!untagged_dir.exists(), "Untagged snapshot should be pruned");
    }
}