wsb-daemon 0.258.6441

//! Sync engine — discovers session files from multiple AI agents, parses
//! them, and upserts them to the machine's sessions database, each placed in
//! the project and worktree it ran in. Supports incremental sync (skip
//! unchanged files) and full resync.
//!
//! One engine and one transcript watcher per MACHINE: sessions belong to the
//! machine, and a worktree's copy of them was a copy of the same files — one
//! engine and one watcher per worktree, every one ingesting all of them.
//!
//! Supported agents: Claude, OpenCode, Codex, Gemini, Cursor, Copilot, Pi, OpenHands
//!
//! 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 wsb::prelude::*;

use anyhow::{anyhow, Result};
use notify_debouncer_mini::DebouncedEventKind;
use std::sync::{Arc, RwLock as StdRwLock};
use tokio::sync::{mpsc, oneshot};
use wsb::persist::global::models::SessionPlace;

use super::home_dir;
use super::parser::{self, AgentType, DiscoveredFile, ParseResult};

fn blocking_bool(handle: tokio::task::JoinHandle<Result<bool, anyhow::Error>>) -> bool {
    futures::executor::block_on(handle)
        .map_err(|e| anyhow!("blocking task join error: {}", e))
        .and_then(|r| r)
        .unwrap_or(false)
}

fn blocking_option<T>(
    handle: tokio::task::JoinHandle<Result<Option<T>, anyhow::Error>>,
) -> Option<T> {
    futures::executor::block_on(handle)
        .map_err(|e| anyhow!("blocking task join error: {}", e))
        .and_then(|r| r)
        .ok()
        .flatten()
}

fn blocking_i64(handle: tokio::task::JoinHandle<Result<i64, anyhow::Error>>) -> i64 {
    futures::executor::block_on(handle)
        .map_err(|e| anyhow!("blocking task join error: {}", e))
        .and_then(|r| r)
        .unwrap_or(0)
}

fn blocking_unit(handle: tokio::task::JoinHandle<Result<(), anyhow::Error>>) -> Result<()> {
    futures::executor::block_on(handle)
        .map_err(|e| anyhow!("blocking task join error: {}", e))
        .and_then(|r| r)
}

#[derive(Clone, serde::Serialize, serde::Deserialize)]
pub struct SyncStats {
    pub synced: usize,
    pub failed: usize,
    pub skipped: usize,
    pub total: usize,
    pub aborted: bool,
}

enum SyncRequest {
    Incremental {
        reply: oneshot::Sender<Result<SyncStats>>,
    },
    Full {
        reply: oneshot::Sender<Result<SyncStats>>,
    },
    Targeted {
        paths: Vec<String>,
        reply: oneshot::Sender<Result<SyncStats>>,
    },
}

#[derive(Clone)]
pub struct SyncHandle {
    tx: mpsc::Sender<SyncRequest>,
    last_stats: Arc<StdRwLock<Option<SyncStats>>>,
}

impl SyncHandle {
    async fn ask(&self, request: SyncRequest, reply_rx: oneshot::Receiver<Result<SyncStats>>) -> Result<SyncStats> {
        self.tx
            .send(request)
            .await
            .map_err(|_| anyhow::anyhow!("sync actor is shut down"))?;
        reply_rx
            .await
            .map_err(|_| anyhow::anyhow!("sync actor dropped reply"))?
    }

    pub async fn sync_incremental(&self) -> Result<SyncStats> {
        let (reply, reply_rx) = oneshot::channel();
        self.ask(SyncRequest::Incremental { reply }, reply_rx).await
    }

    pub async fn sync_full(&self) -> Result<SyncStats> {
        let (reply, reply_rx) = oneshot::channel();
        self.ask(SyncRequest::Full { reply }, reply_rx).await
    }

    pub async fn sync_targeted(&self, paths: Vec<String>) -> Result<SyncStats> {
        let (reply, reply_rx) = oneshot::channel();
        self.ask(SyncRequest::Targeted { paths, reply }, reply_rx).await
    }

    pub fn last_stats(&self) -> Option<SyncStats> {
        self.last_stats.read().unwrap().clone()
    }
}

pub fn spawn_sync_actor(db: Arc<SessionRepo>, prj: Option<wsb_prj::Prj>) -> SyncHandle {
    let (tx, rx) = mpsc::channel(8);
    let last_stats = Arc::new(StdRwLock::new(None));
    tokio::spawn(run_sync_actor(db, prj, rx, Arc::clone(&last_stats)));
    SyncHandle { tx, last_stats }
}

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

    while let Some(req) = rx.recv().await {
        match req {
            SyncRequest::Incremental { reply } => {
                let result = engine.sync_files(None).await;
                if let Ok(ref stats) = result {
                    *last_stats.write().unwrap() = Some(stats.clone());
                }
                let _ = reply.send(result);
            }
            SyncRequest::Full { reply } => {
                let result = engine.sync_full().await;
                if let Ok(ref stats) = result {
                    *last_stats.write().unwrap() = Some(stats.clone());
                }
                let _ = reply.send(result);
            }
            SyncRequest::Targeted { paths, reply } => {
                let result = engine.sync_files(Some(paths)).await;
                if let Ok(ref stats) = result {
                    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);
            }
        }
    }
}

/// Where a session ran, from the working directory its transcript recorded.
///
/// Nowhere when it recorded none, or when the machine has no `$PRJ`: a place
/// is a statement about `$PRJ`, and guessing one from an agent's own folder
/// names is the grouping this replaced.
pub fn place_in(prj: Option<&wsb_prj::Prj>, cwd: &str) -> SessionPlace {
    let (Some(prj), false) = (prj, cwd.is_empty()) else {
        return SessionPlace::default();
    };
    match prj.place_of(std::path::Path::new(cwd)) {
        Some((project, worktree)) => SessionPlace {
            project: Some(project),
            worktree,
        },
        None => SessionPlace::default(),
    }
}

struct SyncEngine {
    db: Arc<SessionRepo>,
    skip_cache: HashMap<String, i64>,
    prj: Option<wsb_prj::Prj>,
}

impl SyncEngine {
    fn new(db: Arc<SessionRepo>, prj: Option<wsb_prj::Prj>) -> Self {
        Self {
            db,
            skip_cache: HashMap::new(),
            prj,
        }
    }

    fn load_skip_cache(&mut self) -> Result<()> {
        self.skip_cache = self.db.load_skip_cache()?;
        Ok(())
    }

    /// Sync every discovered session file, or only those at `paths`.
    ///
    /// One loop for both: the incremental and the targeted sync were two
    /// copies of it that had drifted — the targeted one skipped no failed
    /// file on its next pass and never linked subagent sessions.
    async fn sync_files(&mut self, paths: Option<Vec<String>>) -> Result<SyncStats> {
        let discovered = tokio::task::spawn_blocking(parser::discover_all_sessions)
            .await
            .map_err(|e| anyhow::anyhow!("discovery panicked: {}", e))??;
        let files: Vec<DiscoveredFile> = match paths {
            None => discovered,
            Some(paths) => {
                let wanted: std::collections::HashSet<String> = paths.into_iter().collect();
                discovered
                    .into_iter()
                    .filter(|f| wanted.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 in &files {
            let cache_key = file.session_id.clone();

            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)) => {
                        wsb::errln!("[wsb] sync: stat error {}: {}", file.path.display(), e);
                        failed += 1;
                        continue;
                    }
                    Err(e) => {
                        wsb::errln!("[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(&cache_key) {
                if cached_mtime == file_mtime {
                    skipped += 1;
                    continue;
                }
            }

            if let Some((stored_size, stored_mtime)) =
                blocking_option(tokio::task::spawn_blocking({
                    let session_id = file.session_id.clone();
                    let db = self.db.clone();
                    move || db.get_session_file_info(&session_id)
                }))
            {
                if stored_size == file_size && stored_mtime == file_mtime {
                    skipped += 1;
                    continue;
                }
            }

            if blocking_bool(tokio::task::spawn_blocking({
                let session_id = file.session_id.clone();
                let db = self.db.clone();
                move || db.is_excluded(&session_id)
            })) {
                self.skip_cache.insert(cache_key.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(&cache_key);
                continue;
            }

            let parse_path = file.path.clone();
            let parse_cwd = file.cwd.clone();
            let parse_agent_type = file.agent_type.clone();
            let parse_session_id = file.session_id.clone();
            let parse_result = tokio::task::spawn_blocking(move || {
                parser::parse_session_by_agent(
                    &parse_path,
                    parse_cwd.as_deref(),
                    "local",
                    parse_agent_type,
                    &parse_session_id,
                )
            })
            .await;

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

        if let Err(e) = blocking_unit(tokio::task::spawn_blocking({
            let db = self.db.clone();
            move || db.link_subagent_sessions()
        })) {
            wsb::errln!("[wsb] sync: link subagent sessions: {}", e);
        }

        if let Err(e) = blocking_unit(tokio::task::spawn_blocking({
            let db = self.db.clone();
            let cache = self.skip_cache.clone();
            move || db.persist_skip_cache(&cache)
        })) {
            wsb::errln!("[wsb] sync: persist skip cache: {}", e);
        }

        Ok(SyncStats {
            synced,
            failed,
            skipped,
            total,
            aborted: false,
        })
    }

    async fn sync_full(&mut self) -> Result<SyncStats> {
        self.skip_cache.clear();
        blocking_unit(tokio::task::spawn_blocking({
            let db = self.db.clone();
            move || db.persist_skip_cache(&HashMap::new())
        }))
        .map_err(|e| anyhow::anyhow!("cannot clear the skip cache: {e:#}"))?;
        self.sync_files(None).await
    }

    async fn try_incremental(
        &self,
        file: &DiscoveredFile,
        current_size: i64,
        current_mtime: i64,
    ) -> Option<Result<()>> {
        if file.agent_type != AgentType::Claude {
            return None;
        }

        let session_id = file.session_id.clone();
        let db = self.db.clone();

        let (stored_size, _stored_mtime) =
            match blocking_option(tokio::task::spawn_blocking(move || {
                db.get_session_file_info(&session_id)
            })) {
                Some(info) => info,
                _ => return None,
            };

        if current_size <= stored_size {
            return None;
        }

        let max_ordinal = match blocking_option(tokio::task::spawn_blocking({
            let session_id = file.session_id.clone();
            let db = self.db.clone();
            move || db.max_ordinal(&session_id)
        })) {
            Some(ord) => ord,
            _ => return None,
        };

        let path_str = file.path.to_string_lossy().to_string();
        let session_count = blocking_i64(tokio::task::spawn_blocking({
            let db = self.db.clone();
            move || db.count_sessions_for_path(&path_str)
        }));
        if session_count != 1 {
            return None;
        }

        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) => {
                wsb::errln!(
                    "[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) = blocking_unit(tokio::task::spawn_blocking({
                        let session_id = file.session_id.clone();
                        let db = self.db.clone();
                        let ended_at = ended_at.clone();
                        move || {
                            db.update_session_incremental(
                                &session_id,
                                ended_at.as_deref(),
                                -1,
                                -1,
                                stored_size + bytes_consumed,
                                current_mtime,
                                0,
                                0,
                            )
                        }
                    })) {
                        return Some(Err(e));
                    }
                    return Some(Ok(()));
                }

                if new_msgs.is_empty() {
                    return None;
                }

                if let Err(e) = blocking_unit(tokio::task::spawn_blocking({
                    let db = self.db.clone();
                    let msgs = new_msgs.clone();
                    move || db.insert_messages(&msgs)
                })) {
                    return Some(Err(e));
                }

                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 = blocking_i64(tokio::task::spawn_blocking({
                    let session_id = file.session_id.clone();
                    let db = self.db.clone();
                    move || db.get_session_user_count(&session_id)
                }));

                let new_offset = stored_size + bytes_consumed;
                if let Err(e) = blocking_unit(tokio::task::spawn_blocking({
                    let session_id = file.session_id.clone();
                    let db = self.db.clone();
                    let ended_at = ended_at.clone();
                    move || {
                        db.update_session_incremental(
                            &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,
                        )
                    }
                })) {
                    return Some(Err(e));
                }

                Some(Ok(()))
            }
            Err(_) => None,
        }
    }

    async fn write_result(&self, result: &ParseResult) -> Result<()> {
        let db = self.db.clone();
        let session = result.session.clone();
        let messages = result.messages.clone();
        let place = place_in(self.prj.as_ref(), &session.cwd);

        blocking_unit(tokio::task::spawn_blocking(move || {
            db.upsert_session(&session, &place)?;
            if !messages.is_empty() {
                db.insert_messages(&messages)?;
            }
            Ok(())
        }))
    }
}

/// Follow the agents' transcript folder, and sync each transcript that
/// changes.
pub async fn start_watcher(sync: SyncHandle) -> 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() {
        wsb::errln!(
            "[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) => {
                    wsb::errln!("[wsb] file watcher event error: {}", err);
                }
            },
        ) {
            Ok(debouncer) => debouncer,
            Err(err) => {
                wsb::errln!("[wsb] failed to create file watcher: {}", err);
                return;
            }
        };

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

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

    while let Some(changed_paths) = rx.recv().await {
        if let Err(e) = sync.sync_targeted(changed_paths).await {
            wsb::errln!("[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};

    #[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(),]
        );
    }

    #[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),
        };

        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);
        }
    }

    #[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 };

        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
        );
    }
}