wsb 0.93.391653

//! Sync engine — discovers Claude JSONL session files, parses them,
//! and upserts to the sessions database. Supports incremental sync
//! (skip unchanged files) and full resync. File watching via notify crate.

use anyhow::{Context, Result};
use std::collections::HashMap;

use super::db::SessionDb;
use super::parser::{self, DiscoveredFile, ParseResult};
use super::state::WebState;

/// Sync engine wrapping the session database.
pub struct SyncEngine {
    db: SessionDb,
    skip_cache: HashMap<String, i64>, // path → mtime (nanoseconds)
    last_sync_stats: Option<SyncStats>,
}

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

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

    pub fn last_stats(&self) -> Option<&SyncStats> {
        self.last_sync_stats.as_ref()
    }

    /// Load the skip cache from the database on startup.
    pub 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.
    pub async fn sync_incremental(&mut self) -> Result<SyncStats> {
        let files = parser::discover_claude_sessions()?;
        let total = files.len();
        let mut synced = 0;
        let mut failed = 0;
        let mut skipped = 0;

        for file in &files {
            let path_str = file.path.to_string_lossy().to_string();

            // Check file metadata
            let metadata = match std::fs::metadata(&file.path) {
                Ok(m) => m,
                Err(e) => {
                    eprintln!(
                        "[wsb] sync: stat error {}: {}",
                        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
            match parser::parse_claude_session(
                &file.path,
                &file.project,
                "local",
            ) {
                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);
                }
                Err(e) => {
                    eprintln!(
                        "[wsb] sync: parse error {}: {}",
                        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,
        };
        self.last_sync_stats = Some(stats.clone());
        Ok(stats)
    }

    /// Full resync: re-parse everything, preserving user data.
    /// For now this is implemented as a sync_incremental with the skip cache cleared.
    /// A production-grade implementation would use the atomic DB swap pattern
    /// from agentsview's ResyncAll.
    pub async fn sync_full(&mut self) -> Result<SyncStats> {
        self.skip_cache.clear();
        // Clear DB skip cache too
        if let Err(e) = self
            .db
            .persist_skip_cache(&HashMap::new())
            .await
        {
            eprintln!("[wsb] resync: clear skip cache: {}", e);
        }
        self.sync_incremental().await
    }

    /// 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
        match parser::parse_claude_session_from(
            &file.path,
            stored_size,
            max_ordinal + 1,
            &file.session_id,
        ) {
            Ok((new_msgs, ended_at, bytes_consumed)) => {
                if new_msgs.is_empty() && bytes_consumed > 0 {
                    // Non-message events, just advance file size
                    if let Err(e) = self
                        .db
                        .update_session_incremental(
                            &file.session_id,
                            ended_at.as_deref(),
                            // Keep existing counts
                            -1, // sentinel: handler should use existing
                            -1,
                            stored_size + bytes_consumed,
                            current_mtime,
                            0,
                            0,
                        )
                        .await
                    {
                        return Some(Err(e));
                    }
                    return Some(Ok(()));
                }

                if new_msgs.is_empty() {
                    return None; // Nothing new, skip
                }

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

                // Get existing user count from DB to compute accurate total
                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(e) => {
                // DAG fork detected or other error — fall through to full parse
                eprintln!(
                    "[wsb] incremental parse failed for {}: {} (falling back to full)",
                    file.session_id, e
                );
                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, DebouncedEventKind};
    use std::time::Duration;

    let home = dirs::home_dir().context("cannot determine home directory")?;
    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();

    // Spawn the blocking notify watcher in a separate thread
    let watch_dir_clone = watch_dir.clone();
    std::thread::spawn(move || {
        let rt = runtime_handle;
        let mut debouncer = new_debouncer(
            Duration::from_millis(500),
            move |events: Result<Vec<notify_debouncer_mini::DebouncedEvent>, notify::Error>| {
                if let Ok(events) = events {
                    let paths: Vec<String> = events
                        .into_iter()
                        .filter(|e| e.kind == DebouncedEventKind::Any)
                        .filter_map(|e| {
                            let path = e.path;
                            if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
                                Some(path.to_string_lossy().to_string())
                            } else {
                                None
                            }
                        })
                        .collect();

                    if !paths.is_empty() {
                        let tx = tx.clone();
                        rt.spawn(async move {
                            let _ = tx.send(paths).await;
                        });
                    }
                }
            },
        )
        .expect("failed to create file watcher");

        debouncer
            .watcher()
            .watch(
                &watch_dir_clone,
                notify::RecursiveMode::Recursive,
            )
            .expect("failed to watch directory");

        // Keep the watcher alive
        loop {
            std::thread::sleep(Duration::from_secs(3600));
        }
    });

    // Process file change events
    while let Some(_changed_paths) = rx.recv().await {
        let engine = state.sync_engine.clone();
        let mut eng = engine.lock().await;
        if let Err(e) = eng.sync_incremental().await {
            eprintln!("[wsb] watcher sync failed: {}", e);
        }
    }

    Ok(())
}