wsb 0.153.17

//! Sync engine — discovers Claude JSONL session files, parses them,
//! and upserts to the sessions database. Supports incremental sync
//! (skip unchanged files) and full resync.
//!
//! The engine runs as an actor on a dedicated task, processing one sync
//! at a time. Callers send a `SyncRequest` and receive a `SyncResult`
//! back on a one-shot channel. This means the engine never holds a
//! shared lock across async I/O — any number of API requests can be
//! served concurrently, and a long sync operation only blocks the
//! dedicated actor task.

use anyhow::Result;
use notify_debouncer_mini::DebouncedEventKind;
use std::collections::HashMap;
use std::sync::{Arc, RwLock as StdRwLock};
use tokio::sync::{mpsc, oneshot};

use super::db::SessionDb;
use super::home_dir;
use super::ops::{OpKind, OpsRegistry};
use super::parser::{self, DiscoveredFile, ParseResult};
use super::state::WebState;

/// The result of a sync operation.
#[derive(Debug, Clone, serde::Serialize)]
pub struct SyncStats {
    pub synced: usize,
    pub failed: usize,
    pub skipped: usize,
    pub total: usize,
    pub aborted: bool,
}

/// Requests sent to the sync actor.
enum SyncRequest {
    Incremental {
        reply: oneshot::Sender<Result<SyncStats>>,
        ops: Option<OpsRegistry>,
    },
    Full {
        reply: oneshot::Sender<Result<SyncStats>>,
        ops: Option<OpsRegistry>,
    },
    /// Sync only a specific set of files — no discovery, no full scan.
    Targeted {
        paths: Vec<String>,
        reply: oneshot::Sender<Result<SyncStats>>,
        ops: Option<OpsRegistry>,
    },
}

/// A cheap, clone-able handle to the sync actor.
///
/// `last_stats` is stored in a shared `RwLock` updated by the actor after
/// each sync — reads are always instant and never queue behind a running sync.
#[derive(Clone)]
pub struct SyncHandle {
    tx: mpsc::Sender<SyncRequest>,
    last_stats: Arc<StdRwLock<Option<SyncStats>>>,
}

impl SyncHandle {
    /// Trigger an incremental sync and wait for the result.
    /// Pass an `OpsRegistry` to have the sync registered as a tracked background op.
    pub async fn sync_incremental(&self) -> Result<SyncStats> {
        self.sync_incremental_tracked(None).await
    }

    pub async fn sync_incremental_tracked(&self, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(SyncRequest::Incremental { reply: reply_tx, ops })
            .await
            .map_err(|_| anyhow::anyhow!("sync actor is shut down"))?;
        reply_rx
            .await
            .map_err(|_| anyhow::anyhow!("sync actor dropped reply"))?
    }

    /// Trigger a full resync and wait for the result.
    pub async fn sync_full(&self) -> Result<SyncStats> {
        self.sync_full_tracked(None).await
    }

    pub async fn sync_full_tracked(&self, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(SyncRequest::Full { reply: reply_tx, ops })
            .await
            .map_err(|_| anyhow::anyhow!("sync actor is shut down"))?;
        reply_rx
            .await
            .map_err(|_| anyhow::anyhow!("sync actor dropped reply"))?
    }

    /// Sync only the given paths — skips discovery entirely.
    pub async fn sync_targeted(&self, paths: Vec<String>, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(SyncRequest::Targeted { paths, reply: reply_tx, ops })
            .await
            .map_err(|_| anyhow::anyhow!("sync actor is shut down"))?;
        reply_rx
            .await
            .map_err(|_| anyhow::anyhow!("sync actor dropped reply"))?
    }

    /// Return the stats from the last completed sync without touching the actor.
    /// This is a non-blocking read — it never waits for an in-progress sync.
    pub fn last_stats(&self) -> Option<SyncStats> {
        self.last_stats.read().unwrap().clone()
    }
}

/// Spawn the sync actor and return a handle to it.
/// The actor runs until the last `SyncHandle` is dropped.
pub fn spawn_sync_actor(db: SessionDb) -> SyncHandle {
    let (tx, rx) = mpsc::channel(8);
    let last_stats = Arc::new(StdRwLock::new(None));
    tokio::spawn(run_sync_actor(db, rx, Arc::clone(&last_stats)));
    SyncHandle { tx, last_stats }
}

async fn run_sync_actor(
    db: SessionDb,
    mut rx: mpsc::Receiver<SyncRequest>,
    last_stats: Arc<StdRwLock<Option<SyncStats>>>,
) {
    let mut engine = SyncEngine::new(db);
    if let Err(e) = engine.load_skip_cache().await {
        eprintln!("[wsb] sync actor: failed to load skip cache: {}", e);
    }

    while let Some(req) = rx.recv().await {
        match req {
            SyncRequest::Incremental { reply, ops } => {
                let result = engine.sync_incremental(ops).await;
                if let Ok(ref stats) = result {
                    *last_stats.write().unwrap() = Some(stats.clone());
                }
                let _ = reply.send(result);
            }
            SyncRequest::Full { reply, ops } => {
                let result = engine.sync_full(ops).await;
                if let Ok(ref stats) = result {
                    *last_stats.write().unwrap() = Some(stats.clone());
                }
                let _ = reply.send(result);
            }
            SyncRequest::Targeted { paths, reply, ops } => {
                let result = engine.sync_targeted(paths, ops).await;
                if let Ok(ref stats) = result {
                    // Merge targeted stats into last_stats so the total stays meaningful.
                    let mut guard = last_stats.write().unwrap();
                    if let Some(ref mut prev) = *guard {
                        prev.synced += stats.synced;
                        prev.failed += stats.failed;
                        prev.skipped += stats.skipped;
                    } else {
                        *guard = Some(stats.clone());
                    }
                }
                let _ = reply.send(result);
            }
        }
    }
}

// ---------------------------------------------------------------------------
// SyncEngine — private implementation, only ever touched by the actor task
// ---------------------------------------------------------------------------

struct SyncEngine {
    db: SessionDb,
    skip_cache: HashMap<String, i64>, // path → mtime (nanoseconds)
}

impl SyncEngine {
    fn new(db: SessionDb) -> Self {
        Self {
            db,
            skip_cache: HashMap::new(),
        }
    }

    /// Load the skip cache from the database on startup.
    async fn load_skip_cache(&mut self) -> Result<()> {
        self.skip_cache = self.db.load_skip_cache().await?;
        Ok(())
    }

    /// Incremental sync: discover all Claude session files, skip unchanged,
    /// parse changed ones, upsert to DB.
    async fn sync_incremental(&mut self, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        let op_handle = if let Some(ref registry) = ops {
            Some(registry.start_op(OpKind::Sync, "Sync Claude history").await)
        } else {
            None
        };

        let files = match tokio::task::spawn_blocking(parser::discover_claude_sessions).await {
            Ok(Ok(f)) => f,
            Ok(Err(e)) => {
                if let Some(h) = op_handle {
                    h.fail(format!("discovery failed: {}", e)).await;
                }
                return Err(e);
            }
            Err(e) => {
                let err = anyhow::anyhow!("discovery panicked: {}", e);
                if let Some(h) = op_handle {
                    h.fail(format!("discovery panicked: {}", e)).await;
                }
                return Err(err);
            }
        };
        let total = files.len();
        let mut synced = 0;
        let mut failed = 0;
        let mut skipped = 0;

        for (file_idx, file) in files.iter().enumerate() {
            // Report progress as fraction of files processed.
            if let Some(ref h) = op_handle {
                if total > 0 {
                    let progress = file_idx as f32 / total as f32;
                    h.progress_labeled(
                        progress,
                        format!("Claude history ({}/{})", file_idx + 1, total),
                    )
                    .await;
                }
            }

            let path_str = file.path.to_string_lossy().to_string();

            // Check file metadata — stat is blocking I/O, must not run on the async executor.
            let stat_path = file.path.clone();
            let metadata = match tokio::task::spawn_blocking(move || std::fs::metadata(&stat_path))
                .await
            {
                Ok(Ok(m)) => m,
                Ok(Err(e)) => {
                    eprintln!("[wsb] sync: stat error {}: {}", file.path.display(), e);
                    failed += 1;
                    continue;
                }
                Err(e) => {
                    eprintln!("[wsb] sync: stat panicked {}: {}", file.path.display(), e);
                    failed += 1;
                    continue;
                }
            };

            let file_size = metadata.len() as i64;
            let file_mtime = metadata
                .modified()
                .ok()
                .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                .map(|d| d.as_nanos() as i64)
                .unwrap_or(0);

            // Check in-memory skip cache
            if let Some(&cached_mtime) = self.skip_cache.get(&path_str) {
                if cached_mtime == file_mtime {
                    skipped += 1;
                    continue;
                }
            }

            // Check DB skip (by session ID for Claude)
            if let Ok(Some((stored_size, stored_mtime))) =
                self.db.get_session_file_info(&file.session_id).await
            {
                if stored_size == file_size && stored_mtime == file_mtime {
                    skipped += 1;
                    continue;
                }
            }

            // Check if excluded
            if let Ok(true) = self.db.is_excluded(&file.session_id).await {
                self.skip_cache.insert(path_str.clone(), file_mtime);
                skipped += 1;
                continue;
            }

            // Try incremental parse first
            let incremental_result = self.try_incremental(&file, file_size, file_mtime).await;
            if let Some(Ok(())) = incremental_result {
                synced += 1;
                self.skip_cache.remove(&path_str);
                continue;
            }

            // Full parse — file I/O must not run on the async executor.
            let parse_path = file.path.clone();
            let parse_project = file.project.clone();
            let parse_result = tokio::task::spawn_blocking(move || {
                parser::parse_claude_session(&parse_path, &parse_project, "local")
            })
            .await;

            match parse_result {
                Ok(Ok(results)) => {
                    if results.is_empty() {
                        self.skip_cache.insert(path_str, file_mtime);
                        continue;
                    }

                    for result in &results {
                        if let Err(e) = self.write_result(result).await {
                            eprintln!("[wsb] sync: write error for {}: {}", result.session.id, e);
                            failed += 1;
                            continue;
                        }
                    }
                    synced += results.len();
                    self.skip_cache.remove(&path_str);
                }
                Ok(Err(e)) => {
                    eprintln!("[wsb] sync: parse error {}: {}", file.path.display(), e);
                    self.skip_cache.insert(path_str, file_mtime);
                    failed += 1;
                }
                Err(e) => {
                    eprintln!("[wsb] sync: parse panicked {}: {}", file.path.display(), e);
                    self.skip_cache.insert(path_str, file_mtime);
                    failed += 1;
                }
            }
        }

        // Link subagent sessions after all writes
        if let Err(e) = self.db.link_subagent_sessions().await {
            eprintln!("[wsb] sync: link subagent sessions: {}", e);
        }

        // Persist skip cache
        if let Err(e) = self.db.persist_skip_cache(&self.skip_cache).await {
            eprintln!("[wsb] sync: persist skip cache: {}", e);
        }

        let stats = SyncStats {
            synced,
            failed,
            skipped,
            total,
            aborted: false,
        };
        if let Some(h) = op_handle {
            if failed > 0 && synced == 0 {
                h.fail(format!("{} file(s) failed to sync", failed)).await;
            } else {
                h.complete().await;
            }
        }

        Ok(stats)
    }

    /// Full resync: re-parse everything, preserving user data.
    /// Targeted sync: process only the given file paths, no full discovery scan.
    /// Resolves each path into a DiscoveredFile by running a filtered discovery,
    /// then delegates to the same per-file logic as incremental sync.
    async fn sync_targeted(&mut self, paths: Vec<String>, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        let op_handle = if let Some(ref registry) = ops {
            Some(registry.start_op(OpKind::Sync, format!("Sync {} Claude session(s)", paths.len())).await)
        } else {
            None
        };

        // Resolve paths → DiscoveredFile by running discovery and filtering.
        let path_set: std::collections::HashSet<String> = paths.into_iter().collect();
        let path_set_clone = path_set.clone();
        let all_files = match tokio::task::spawn_blocking(parser::discover_claude_sessions).await {
            Ok(Ok(f)) => f,
            Ok(Err(e)) => {
                if let Some(h) = op_handle { h.fail(format!("discovery failed: {}", e)).await; }
                return Err(e);
            }
            Err(e) => {
                let err = anyhow::anyhow!("discovery panicked: {}", e);
                if let Some(h) = op_handle { h.fail(format!("discovery panicked: {}", e)).await; }
                return Err(err);
            }
        };
        let files: Vec<DiscoveredFile> = all_files
            .into_iter()
            .filter(|f| path_set_clone.contains(&f.path.to_string_lossy().to_string()))
            .collect();

        let total = files.len();
        let mut synced = 0;
        let mut failed = 0;
        let mut skipped = 0;

        for (file_idx, file) in files.iter().enumerate() {
            if let Some(ref h) = op_handle {
                if total > 0 {
                    let progress = file_idx as f32 / total as f32;
                    h.progress_labeled(progress, format!("Claude history ({}/{})", file_idx + 1, total)).await;
                }
            }

            let path_str = file.path.to_string_lossy().to_string();
            let stat_path = file.path.clone();
            let metadata = match tokio::task::spawn_blocking(move || std::fs::metadata(&stat_path)).await {
                Ok(Ok(m)) => m,
                Ok(Err(e)) => { eprintln!("[wsb] sync: stat error {}: {}", file.path.display(), e); failed += 1; continue; }
                Err(e) => { eprintln!("[wsb] sync: stat panicked {}: {}", file.path.display(), e); failed += 1; continue; }
            };

            let file_size = metadata.len() as i64;
            let file_mtime = metadata.modified().ok()
                .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                .map(|d| d.as_nanos() as i64)
                .unwrap_or(0);

            if let Some(&cached_mtime) = self.skip_cache.get(&path_str) {
                if cached_mtime == file_mtime { skipped += 1; continue; }
            }

            if let Ok(Some((stored_size, stored_mtime))) = self.db.get_session_file_info(&file.session_id).await {
                if stored_size == file_size && stored_mtime == file_mtime { skipped += 1; continue; }
            }

            if let Ok(true) = self.db.is_excluded(&file.session_id).await {
                self.skip_cache.insert(path_str.clone(), file_mtime);
                skipped += 1;
                continue;
            }

            let incremental_result = self.try_incremental(file, file_size, file_mtime).await;
            if let Some(Ok(())) = incremental_result {
                synced += 1;
                self.skip_cache.remove(&path_str);
                continue;
            }

            let parse_path = file.path.clone();
            let parse_project = file.project.clone();
            let parse_result = tokio::task::spawn_blocking(move || {
                parser::parse_claude_session(&parse_path, &parse_project, "local")
            }).await;

            match parse_result {
                Ok(Ok(results)) => {
                    if results.is_empty() {
                        self.skip_cache.insert(path_str, file_mtime);
                        continue;
                    }
                    for result in &results {
                        if let Err(e) = self.write_result(result).await {
                            eprintln!("[wsb] sync: write error for {}: {}", result.session.id, e);
                            failed += 1;
                            continue;
                        }
                    }
                    synced += results.len();
                    self.skip_cache.remove(&path_str);
                }
                Ok(Err(e)) => { eprintln!("[wsb] sync: parse error {}: {}", path_str, e); failed += 1; }
                Err(e) => { eprintln!("[wsb] sync: spawn_blocking panicked {}: {}", path_str, e); failed += 1; }
            }
        }

        if let Err(e) = self.db.persist_skip_cache(&self.skip_cache).await {
            eprintln!("[wsb] sync: failed to persist skip cache: {}", e);
        }

        let stats = SyncStats { synced, failed, skipped, total, aborted: false };
        if let Some(h) = op_handle { h.complete().await; }
        Ok(stats)
    }

    async fn sync_full(&mut self, ops: Option<OpsRegistry>) -> Result<SyncStats> {
        // Replace the ops label so the UI shows the full-resync variant.
        let op_handle = if let Some(ref registry) = ops {
            Some(registry.start_op(OpKind::Sync, "Resync Claude history (full)").await)
        } else {
            None
        };

        self.skip_cache.clear();
        if let Err(e) = self.db.persist_skip_cache(&HashMap::new()).await {
            eprintln!("[wsb] resync: clear skip cache: {}", e);
        }

        // Run incremental (which now drives file-level progress) with no extra op handle —
        // we own the handle ourselves.
        let result = self.sync_incremental(None).await;

        match &result {
            Ok(_) => {
                if let Some(h) = op_handle {
                    h.complete().await;
                }
            }
            Err(e) => {
                if let Some(h) = op_handle {
                    h.fail(format!("{:#}", e)).await;
                }
            }
        }

        result
    }

    /// Try incremental parse for a file that has grown since last sync.
    async fn try_incremental(
        &self,
        file: &DiscoveredFile,
        current_size: i64,
        current_mtime: i64,
    ) -> Option<Result<()>> {
        // Check if we have a stored file size for this session
        let (stored_size, _stored_mtime) =
            match self.db.get_session_file_info(&file.session_id).await {
                Ok(Some(info)) => info,
                _ => return None,
            };

        // Only try incremental if file has grown
        if current_size <= stored_size {
            return None;
        }

        // Must have existing messages
        let max_ordinal = match self.db.max_ordinal(&file.session_id).await {
            Ok(Some(ord)) => ord,
            _ => return None,
        };

        // Check this file maps to exactly one session (no forks)
        let session_count = match self
            .db
            .count_sessions_for_path(&file.path.to_string_lossy())
            .await
        {
            Ok(count) => count,
            Err(_) => return None,
        };
        if session_count != 1 {
            return None; // Fork detected, needs full parse
        }

        // Parse only new bytes — file I/O must not run on the async executor.
        let inc_path = file.path.clone();
        let inc_session_id = file.session_id.clone();
        let parse_result = tokio::task::spawn_blocking(move || {
            parser::parse_claude_session_from(&inc_path, stored_size, max_ordinal + 1, &inc_session_id)
        })
        .await;

        let parse_result = match parse_result {
            Ok(r) => r,
            Err(e) => {
                eprintln!(
                    "[wsb] incremental parse panicked for {}: {}",
                    file.session_id, e
                );
                return None;
            }
        };

        match parse_result {
            Ok((new_msgs, ended_at, bytes_consumed)) => {
                if new_msgs.is_empty() && bytes_consumed > 0 {
                    if let Err(e) = self
                        .db
                        .update_session_incremental(
                            &file.session_id,
                            ended_at.as_deref(),
                            -1,
                            -1,
                            stored_size + bytes_consumed,
                            current_mtime,
                            0,
                            0,
                        )
                        .await
                    {
                        return Some(Err(e));
                    }
                    return Some(Ok(()));
                }

                if new_msgs.is_empty() {
                    return None;
                }

                // Insert new messages
                if let Err(e) = self.db.insert_messages(&new_msgs).await {
                    return Some(Err(e));
                }

                // Compute new totals
                let mut total_output: i64 = 0;
                let mut peak_context: i64 = 0;
                let mut new_user_count: i64 = 0;
                for msg in &new_msgs {
                    total_output += msg.output_tokens;
                    if msg.context_tokens > peak_context {
                        peak_context = msg.context_tokens;
                    }
                    if msg.role == "user" {
                        new_user_count += 1;
                    }
                }

                let existing_user_count = self
                    .db
                    .get_session_user_count(&file.session_id)
                    .await
                    .unwrap_or(0);

                let new_offset = stored_size + bytes_consumed;
                if let Err(e) = self
                    .db
                    .update_session_incremental(
                        &file.session_id,
                        ended_at.as_deref(),
                        max_ordinal + 1 + new_msgs.len() as i64,
                        existing_user_count + new_user_count,
                        new_offset,
                        current_mtime,
                        total_output,
                        peak_context,
                    )
                    .await
                {
                    return Some(Err(e));
                }

                Some(Ok(()))
            }
            Err(_) => {
                // DAG fork or other incremental parse issue — fall back to full parse silently.
                None
            }
        }
    }

    /// Write a single parse result to the database.
    async fn write_result(&self, result: &ParseResult) -> Result<()> {
        self.db.upsert_session(&result.session).await?;
        if !result.messages.is_empty() {
            self.db.insert_messages(&result.messages).await?;
        }
        Ok(())
    }
}

/// Start the file watcher that triggers incremental sync on changes.
/// Uses notify crate with 500ms debounce.
pub async fn start_watcher(state: WebState) -> Result<()> {
    use notify_debouncer_mini::{new_debouncer, notify::RecursiveMode, DebouncedEvent};
    use std::time::Duration;

    let home = home_dir()?;
    let watch_dir = home.join(".claude").join("projects");

    if !watch_dir.exists() {
        eprintln!(
            "[wsb] watch dir {} does not exist, file watcher disabled",
            watch_dir.display()
        );
        return Ok(());
    }

    let (tx, mut rx) = tokio::sync::mpsc::channel(256);
    let runtime_handle = tokio::runtime::Handle::current();

    let watch_dir_clone = watch_dir.clone();
    std::thread::spawn(move || {
        let rt = runtime_handle;
        let mut debouncer = match new_debouncer(
            Duration::from_millis(500),
            move |events: Result<Vec<DebouncedEvent>, notify::Error>| match events {
                Ok(events) => {
                    let paths = extract_changed_jsonl_paths(events);
                    if !paths.is_empty() {
                        let tx = tx.clone();
                        rt.spawn(async move {
                            let _ = tx.send(paths).await;
                        });
                    }
                }
                Err(err) => {
                    eprintln!("[wsb] file watcher event error: {}", err);
                }
            },
        ) {
            Ok(debouncer) => debouncer,
            Err(err) => {
                eprintln!("[wsb] failed to create file watcher: {}", err);
                return;
            }
        };

        if let Err(err) = debouncer
            .watcher()
            .watch(&watch_dir_clone, RecursiveMode::Recursive)
        {
            eprintln!(
                "[wsb] failed to watch directory {}: {}",
                watch_dir_clone.display(),
                err
            );
            return;
        }

        loop {
            std::thread::sleep(Duration::from_secs(3600));
        }
    });

    // Each file-change event triggers a targeted sync of only the changed files.
    // The actor serializes requests, so concurrent events queue rather than pile up.
    while let Some(changed_paths) = rx.recv().await {
        if let Err(e) = state.sync.sync_targeted(changed_paths, Some(state.ops.clone())).await {
            eprintln!("[wsb] watcher sync failed: {}", e);
        }
    }

    Ok(())
}

fn extract_changed_jsonl_paths(
    events: Vec<notify_debouncer_mini::DebouncedEvent>,
) -> Vec<String> {
    let mut paths = Vec::new();
    for event in events {
        if event.kind != DebouncedEventKind::Any
            && event.kind != DebouncedEventKind::AnyContinuous
        {
            continue;
        }
        if event.path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") {
            continue;
        }
        paths.push(event.path.to_string_lossy().to_string());
    }
    paths.sort();
    paths.dedup();
    paths
}

#[cfg(test)]
mod tests {
    use super::*;
    use notify_debouncer_mini::{DebouncedEvent, DebouncedEventKind};
    use std::path::PathBuf;

    #[test]
    fn extract_changed_jsonl_paths_keeps_only_jsonl_and_deduplicates() {
        let paths = extract_changed_jsonl_paths(vec![
            DebouncedEvent {
                path: PathBuf::from("/tmp/a.jsonl"),
                kind: DebouncedEventKind::Any,
            },
            DebouncedEvent {
                path: PathBuf::from("/tmp/a.jsonl"),
                kind: DebouncedEventKind::AnyContinuous,
            },
            DebouncedEvent {
                path: PathBuf::from("/tmp/b.txt"),
                kind: DebouncedEventKind::Any,
            },
            DebouncedEvent {
                path: PathBuf::from("/tmp/c.jsonl"),
                kind: DebouncedEventKind::AnyContinuous,
            },
        ]);

        assert_eq!(
            paths,
            vec!["/tmp/a.jsonl".to_string(), "/tmp/c.jsonl".to_string(),]
        );
    }

    /// Verifies that multiple concurrent callers all read the same stats from the
    /// shared RwLock without mixing results or blocking each other.
    #[tokio::test]
    async fn sync_handle_last_stats_reads_shared_value() {
        let (tx, _rx) = tokio::sync::mpsc::channel::<SyncRequest>(8);
        let last_stats = std::sync::Arc::new(std::sync::RwLock::new(Some(SyncStats {
            synced: 42,
            failed: 0,
            skipped: 0,
            total: 42,
            aborted: false,
        })));
        let handle = SyncHandle {
            tx,
            last_stats: std::sync::Arc::clone(&last_stats),
        };

        // 8 concurrent readers — all should see synced=42.
        let mut futures = Vec::new();
        for _ in 0..8 {
            let h = handle.clone();
            futures.push(tokio::spawn(async move { h.last_stats() }));
        }
        for fut in futures {
            let stats = fut.await.expect("task did not panic");
            let stats = stats.expect("should return Some");
            assert_eq!(stats.synced, 42);
        }
    }

    /// Verifies that last_stats() is non-blocking: it must return immediately
    /// even when a long-running spawn_blocking task is in flight on the thread pool.
    #[tokio::test]
    async fn last_stats_does_not_block_on_spawn_blocking() {
        let (tx, _rx) = tokio::sync::mpsc::channel::<SyncRequest>(8);
        let last_stats = std::sync::Arc::new(std::sync::RwLock::new(Some(SyncStats {
            synced: 1,
            failed: 0,
            skipped: 0,
            total: 1,
            aborted: false,
        })));
        let handle = SyncHandle { tx, last_stats };

        // Simulate a slow spawn_blocking task (e.g., file I/O).
        tokio::spawn(async move {
            tokio::task::spawn_blocking(|| {
                std::thread::sleep(std::time::Duration::from_millis(200));
            })
            .await
            .expect("spawn_blocking must not panic");
        });

        tokio::task::yield_now().await;

        let start = std::time::Instant::now();
        let stats = handle.last_stats();
        let elapsed = start.elapsed();

        assert!(stats.is_some(), "must return the cached stats");
        assert!(
            elapsed < std::time::Duration::from_millis(10),
            "last_stats took {:?} — RwLock read should be near-instant",
            elapsed
        );
    }
}