magellan 4.12.3

//! Filesystem watcher with debounced batch events.
//!
//! Provides deterministic event coalescing: all events within a debounce window
//! are collected, de-duplicated, sorted lexicographically, and emitted as a single
//! batch. This ensures the same final DB state regardless of event arrival order.
//!
//! # Threading Design
//!
//! This watcher uses thread-safe synchronization for concurrent access.
//! The legacy pending state fields use `Arc<parking_lot::Mutex<T>>` to allow safe access
//! from multiple threads during concurrent operations and shutdown.
//!
//! **Thread safety:** `Arc<parking_lot::Mutex<T>>` provides runtime mutual exclusion
//! and is safe to share across threads. parking_lot Mutex does not poison on panic.
//!
//! # Global Lock Ordering
//!
//! This module participates in the global lock ordering hierarchy:
//!
//! 1. **watcher state locks** (legacy_pending_batch, legacy_pending_index)—acquired first
//! 2. **indexer shared state locks** (dirty_paths)—acquired second
//! 3. **wakeup channel send** (highest priority)—acquired last
//!
//! **Rule:** Never send to wakeup channel while holding other locks.
//!
//! See `src/indexer.rs::PipelineSharedState` for full lock ordering documentation.
//!
//! See MANUAL.md for architecture details.

pub mod async_watcher;

use anyhow::Result;
use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeSet, HashMap};
use std::mem::ManuallyDrop;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use crate::diagnostics::SkipReason;
use crate::graph::filter::FileFilter;

/// Deterministic batch of dirty file paths.
///
/// Contains ONLY paths (no timestamps, no event types) to ensure deterministic
/// behavior. Paths are sorted lexicographically before emission.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WatcherBatch {
    /// Dirty file paths to reconcile, in lexicographic order
    pub paths: Vec<PathBuf>,
}

impl WatcherBatch {
    /// Create a new batch from a set of paths, sorting them deterministically.
    pub fn from_set(paths: BTreeSet<PathBuf>) -> Self {
        Self {
            paths: paths.into_iter().collect(),
        }
    }

    /// Empty batch for when no dirty paths exist after filtering.
    pub fn empty() -> Self {
        Self { paths: Vec::new() }
    }

    /// Whether this batch contains any paths.
    pub fn is_empty(&self) -> bool {
        self.paths.is_empty()
    }
}

/// Filesystem watcher configuration
#[derive(Debug, Clone)]
pub struct WatcherConfig {
    /// Root directory for path validation
    pub root_path: PathBuf,
    /// Debounce delay in milliseconds
    pub debounce_ms: u64,
    /// Enable .gitignore filtering (default: true)
    pub gitignore_aware: bool,
}

impl Default for WatcherConfig {
    fn default() -> Self {
        Self {
            root_path: PathBuf::from("."),
            debounce_ms: 500,
            gitignore_aware: true,
        }
    }
}

/// Filesystem watcher that emits debounced batches of dirty paths.
///
/// Uses `notify::RecommendedWatcher` directly with custom debouncing that
/// filters out read-only events (ACCESS/OPEN/CLOSE_NOWRITE) at the source.
/// All paths within the debounce window are collected, de-duplicated, sorted,
/// and emitted as a single `WatcherBatch`.
pub struct FileSystemWatcher {
    /// Watcher thread handle (wrapped in ManuallyDrop for custom Drop/shutdown logic)
    _watcher_thread: ManuallyDrop<thread::JoinHandle<()>>,
    batch_receiver: Receiver<WatcherBatch>,
    /// Legacy compatibility: pending batch to emit one path at a time
    /// Thread-safe: wrapped in `Arc<parking_lot::Mutex<T>>` for concurrent access
    legacy_pending_batch: Arc<Mutex<Option<WatcherBatch>>>,
    /// Legacy compatibility: current index into pending batch
    /// Thread-safe: wrapped in `Arc<parking_lot::Mutex<T>>` for concurrent access
    legacy_pending_index: Arc<Mutex<usize>>,
}

impl FileSystemWatcher {
    /// Create a new watcher for the given directory.
    ///
    /// # Arguments
    /// * `path` - Directory to watch recursively (also used as root_path for validation)
    /// * `config` - Watcher configuration
    /// * `shutdown` - AtomicBool for graceful shutdown
    ///
    /// # Returns
    /// A watcher that can be polled for batch events
    pub fn new(path: PathBuf, config: WatcherConfig, shutdown: Arc<AtomicBool>) -> Result<Self> {
        let (batch_tx, batch_rx) = mpsc::channel();

        let config = WatcherConfig {
            root_path: path.clone(),
            ..config
        };

        let thread = thread::spawn(move || {
            if let Err(e) = run_watcher(path, batch_tx, config, shutdown) {
                eprintln!("Watcher error: {:?}", e);
            }
        });

        Ok(Self {
            _watcher_thread: ManuallyDrop::new(thread),
            batch_receiver: batch_rx,
            legacy_pending_batch: Arc::new(Mutex::new(None)),
            legacy_pending_index: Arc::new(Mutex::new(0)),
        })
    }

    /// Receive the next batch, blocking until available.
    ///
    /// # Returns
    /// `None` if the watcher thread has terminated
    pub fn recv_batch(&self) -> Option<WatcherBatch> {
        self.batch_receiver.recv().ok()
    }

    /// Try to receive a batch without blocking.
    ///
    /// # Returns
    /// - `Some(batch)` if a batch is available
    /// - `None` if no batch is available or watcher terminated
    pub fn try_recv_batch(&self) -> Option<WatcherBatch> {
        self.batch_receiver.try_recv().ok()
    }

    /// Receive the next batch with a timeout.
    ///
    /// # Returns
    /// - `Ok(Some(batch))` if a batch is available
    /// - `Ok(None)` if the watcher thread has terminated
    /// - `Err` if timeout elapsed
    #[allow(
        clippy::result_unit_err,
        reason = "simple timeout/empty signaling; no error details needed"
    )]
    pub fn recv_batch_timeout(&self, timeout: Duration) -> Result<Option<WatcherBatch>, ()> {
        match self.batch_receiver.recv_timeout(timeout) {
            Ok(batch) => Ok(Some(batch)),
            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Err(()),
            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => Ok(None),
        }
    }

    // ========================================================================
    // LEGACY: Old single-event API for backward compatibility during migration
    // ========================================================================

    /// Legacy: Try to receive a single event without blocking (DEPRECATED).
    ///
    /// This method converts batch events to single events for backward
    /// compatibility. Paths from each batch are returned one at a time
    /// in sorted order.
    ///
    /// # Deprecated
    /// Use `try_recv_batch()` instead for deterministic batch processing.
    ///
    /// # Errors
    pub fn try_recv_event(&self) -> Result<Option<FileEvent>> {
        {
            let mut pending_batch = self.legacy_pending_batch.lock();
            let mut pending_index = self.legacy_pending_index.lock();

            if let Some(ref batch) = *pending_batch {
                if *pending_index < batch.paths.len() {
                    let path = batch.paths[*pending_index].clone();
                    *pending_index += 1;

                    if *pending_index >= batch.paths.len() {
                        *pending_batch = None;
                        *pending_index = 0;
                    }

                    return Ok(Some(FileEvent {
                        path,
                        event_type: EventType::Modify,
                    }));
                }
            }
        }

        if let Ok(batch) = self.batch_receiver.try_recv() {
            if batch.paths.is_empty() {
                return Ok(None);
            }

            if batch.paths.len() > 1 {
                let path = batch.paths[0].clone();
                let mut pending_batch = self.legacy_pending_batch.lock();
                let mut pending_index = self.legacy_pending_index.lock();
                *pending_batch = Some(batch);
                *pending_index = 1;
                drop(pending_batch);
                drop(pending_index);
                return Ok(Some(FileEvent {
                    path,
                    event_type: EventType::Modify,
                }));
            }

            Ok(Some(FileEvent {
                path: batch.paths[0].clone(),
                event_type: EventType::Modify,
            }))
        } else {
            Ok(None)
        }
    }

    /// Legacy: Receive the next event, blocking until available (DEPRECATED).
    ///
    /// This method converts batch events to single events for backward
    /// compatibility. Paths from each batch are returned one at a time
    /// in sorted order.
    ///
    /// # Deprecated
    /// Use `recv_batch()` instead for deterministic batch processing.
    ///
    /// # Errors
    pub fn recv_event(&self) -> Result<Option<FileEvent>> {
        {
            let mut pending_batch = self.legacy_pending_batch.lock();
            let mut pending_index = self.legacy_pending_index.lock();

            if let Some(ref batch) = *pending_batch {
                if *pending_index < batch.paths.len() {
                    let path = batch.paths[*pending_index].clone();
                    *pending_index += 1;

                    if *pending_index >= batch.paths.len() {
                        *pending_batch = None;
                        *pending_index = 0;
                    }

                    return Ok(Some(FileEvent {
                        path,
                        event_type: EventType::Modify,
                    }));
                }
            }
        }

        if let Ok(batch) = self.batch_receiver.recv() {
            if batch.paths.is_empty() {
                return Ok(None);
            }

            if batch.paths.len() > 1 {
                let path = batch.paths[0].clone();
                let mut pending_batch = self.legacy_pending_batch.lock();
                let mut pending_index = self.legacy_pending_index.lock();
                *pending_batch = Some(batch);
                *pending_index = 1;
                drop(pending_batch);
                drop(pending_index);
                return Ok(Some(FileEvent {
                    path,
                    event_type: EventType::Modify,
                }));
            }

            Ok(Some(FileEvent {
                path: batch.paths[0].clone(),
                event_type: EventType::Modify,
            }))
        } else {
            Ok(None)
        }
    }

    /// Explicitly shut down the watcher and join all background threads.
    ///
    /// This method consumes the watcher, ensuring that:
    /// 1. The pub/sub receiver is shut down cleanly (if present)
    /// 2. The watcher thread is joined (waits for clean termination)
    ///
    /// # Note
    ///
    /// This method should be called during graceful shutdown to ensure
    /// all threads have terminated before the program exits.
    pub fn shutdown(mut self) {
        let thread = unsafe { ManuallyDrop::take(&mut self._watcher_thread) };
        let _ = thread.join();
    }
}

impl Drop for FileSystemWatcher {
    fn drop(&mut self) {
        let _thread = unsafe { ManuallyDrop::take(&mut self._watcher_thread) };
        drop(_thread);
    }
}

/// Whether a notify event kind represents a write-side mutation worth indexing.
///
/// Read-only events (ACCESS, OPEN, CLOSE_NOWRITE) are excluded because
/// `reconcile_file_path` calls `fs::read`, which triggers those inotify
/// events and would otherwise create an infinite feedback loop.
fn is_mutation_event(kind: &EventKind) -> bool {
    matches!(
        kind,
        EventKind::Create(_)
            | EventKind::Modify(_)
            | EventKind::Remove(_)
            | EventKind::Any
            | EventKind::Other
    )
}

/// Run the debounced watcher in a dedicated thread.
///
/// Uses `notify::RecommendedWatcher` directly with custom debouncing that
/// filters out read-only events (ACCESS/OPEN/CLOSE_NOWRITE) at the source.
/// This prevents the infinite feedback loop where `reconcile_file_path`
/// reads a file → triggers ACCESS inotify event → re-indexes → reads again.
fn run_watcher(
    path: PathBuf,
    tx: Sender<WatcherBatch>,
    config: WatcherConfig,
    shutdown: Arc<AtomicBool>,
) -> Result<()> {
    let debounce_duration = Duration::from_millis(config.debounce_ms);
    let root_path = config.root_path.clone();

    let filter = if config.gitignore_aware {
        match FileFilter::new(&root_path, &[], &[]) {
            Ok(f) => Some(f),
            Err(e) => {
                eprintln!("Warning: Failed to create gitignore filter: {}", e);
                None
            }
        }
    } else {
        None
    };

    let (raw_tx, raw_rx): (Sender<Vec<PathBuf>>, Receiver<Vec<PathBuf>>) = mpsc::channel();

    let mut watcher = RecommendedWatcher::new(
        move |result: notify::Result<notify::Event>| match result {
            Ok(event) => {
                if !is_mutation_event(&event.kind) {
                    return;
                }
                let paths: Vec<PathBuf> = event.paths;
                if !paths.is_empty() {
                    let _ = raw_tx.send(paths);
                }
            }
            Err(error) => {
                eprintln!("Watcher error: {:?}", error);
            }
        },
        notify::Config::default(),
    )?;

    watcher.watch(&path, RecursiveMode::Recursive)?;

    let mut pending: HashMap<PathBuf, Instant> = HashMap::new();
    let idle_sleep = Duration::from_millis(50);

    while !shutdown.load(Ordering::SeqCst) {
        let mut new_paths: Vec<PathBuf> = Vec::new();

        match raw_rx.recv_timeout(debounce_duration) {
            Ok(paths) => {
                new_paths.extend(paths);
                let drain_until = Instant::now() + Duration::from_millis(10);
                while Instant::now() < drain_until {
                    match raw_rx.try_recv() {
                        Ok(paths) => new_paths.extend(paths),
                        Err(_) => break,
                    }
                }
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {}
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
        }

        let now = Instant::now();
        for p in new_paths {
            pending.insert(p, now);
        }

        let mut expired: BTreeSet<PathBuf> = BTreeSet::new();
        pending.retain(|p, ts| {
            if now.duration_since(*ts) >= debounce_duration {
                expired.insert(p.clone());
                false
            } else {
                true
            }
        });

        if !expired.is_empty() {
            let dirty_paths = filter_dirty_paths(expired, &root_path, filter.as_ref());
            if !dirty_paths.is_empty() {
                let batch = WatcherBatch::from_set(dirty_paths);
                let _ = tx.send(batch);
            }
        }

        if pending.is_empty() {
            thread::sleep(idle_sleep);
        }
    }

    Ok(())
}

/// Filter a set of expired paths through gitignore, database, and validation checks.
fn filter_dirty_paths(
    candidates: BTreeSet<PathBuf>,
    root: &Path,
    filter: Option<&FileFilter>,
) -> BTreeSet<PathBuf> {
    let mut dirty_paths = BTreeSet::new();

    for path in candidates {
        if path.is_dir() {
            continue;
        }

        let path_str = path.to_string_lossy();
        if is_database_file(&path_str) {
            continue;
        }

        if let Some(f) = filter {
            match f.should_skip(&path) {
                None => {}
                Some(SkipReason::NotAFile) => {}
                Some(_) => continue,
            }
        }

        match crate::validation::validate_path_within_root(&path, root) {
            Ok(_) => {
                let normalized = crate::validation::normalize_path(&path)
                    .unwrap_or_else(|_| path.to_string_lossy().to_string());
                dirty_paths.insert(PathBuf::from(normalized));
            }
            Err(crate::validation::PathValidationError::OutsideRoot(p, _)) => {
                eprintln!("WARNING: Watcher rejected path outside project root: {}", p);
            }
            Err(crate::validation::PathValidationError::SuspiciousTraversal(p)) => {
                eprintln!(
                    "WARNING: Watcher rejected suspicious traversal pattern: {}",
                    p
                );
            }
            Err(crate::validation::PathValidationError::SymlinkEscape(from, to)) => {
                eprintln!(
                    "WARNING: Watcher rejected symlink escaping root: {} -> {}",
                    from, to
                );
            }
            Err(crate::validation::PathValidationError::CannotCanonicalize(_)) => {
                let normalized = crate::validation::normalize_path(&path)
                    .unwrap_or_else(|_| path.to_string_lossy().to_string());
                dirty_paths.insert(PathBuf::from(normalized));
            }
        }
    }

    dirty_paths
}

/// Check if a path is a database file that should be excluded from watching.
///
/// Database files are excluded because the indexer writes to them, which
/// would create a feedback loop (write event -> indexer writes again -> ...).
fn is_database_file(path: &str) -> bool {
    let path_lower = path.to_lowercase();
    path_lower.ends_with(".db")
        || path_lower.ends_with(".db-journal")
        || path_lower.ends_with(".db-wal")
        || path_lower.ends_with(".db-shm")
        || path_lower.ends_with(".sqlite")
        || path_lower.ends_with(".sqlite3")
}

// ============================================================================
// LEGACY: Old single-event types for backward compatibility during migration
// ============================================================================

/// Legacy: File event emitted by the watcher (DEPRECATED).
///
/// This type is kept for backward compatibility during the migration to
/// batch-based processing. New code should use `WatcherBatch` instead.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct FileEvent {
    /// Path of the affected file
    pub path: PathBuf,
    /// Type of event (DEPRECATED - not used in batch processing)
    pub event_type: EventType,
}

/// Type of file event (DEPRECATED - not used in batch processing).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum EventType {
    /// File was created
    Create,
    /// File was modified
    Modify,
    /// File was deleted
    Delete,
}

impl std::fmt::Display for EventType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EventType::Create => write!(f, "CREATE"),
            EventType::Modify => write!(f, "MODIFY"),
            EventType::Delete => write!(f, "DELETE"),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_batch_is_empty() {
        let batch = WatcherBatch::empty();
        assert!(batch.is_empty());
    }

    #[test]
    fn test_batch_from_set_sorts_deterministically() {
        let mut set = BTreeSet::new();
        set.insert(PathBuf::from("/zebra.rs"));
        set.insert(PathBuf::from("/alpha.rs"));
        set.insert(PathBuf::from("/beta.rs"));

        let batch = WatcherBatch::from_set(set);

        assert_eq!(batch.paths[0], PathBuf::from("/alpha.rs"));
        assert_eq!(batch.paths[1], PathBuf::from("/beta.rs"));
        assert_eq!(batch.paths[2], PathBuf::from("/zebra.rs"));
    }

    #[test]
    fn test_database_file_detection() {
        assert!(is_database_file("test.db"));
        assert!(is_database_file("test.sqlite"));
        assert!(is_database_file("test.db-journal"));
        assert!(is_database_file("test.DB"));
        assert!(is_database_file("test.SQLITE"));

        assert!(!is_database_file("test.rs"));
        assert!(!is_database_file("test.py"));
        assert!(!is_database_file("database.rs"));
    }

    #[test]
    fn test_batch_serialization() {
        let batch = WatcherBatch {
            paths: vec![PathBuf::from("/alpha.rs"), PathBuf::from("/beta.rs")],
        };

        let json = serde_json::to_string(&batch).unwrap();
        let deserialized: WatcherBatch = serde_json::from_str(&json).unwrap();

        assert_eq!(batch.paths, deserialized.paths);
    }

    #[test]
    fn test_watcher_config_has_root() {
        let config = WatcherConfig {
            root_path: PathBuf::from("/test/root"),
            debounce_ms: 100,
            gitignore_aware: true,
        };

        assert_eq!(config.root_path, PathBuf::from("/test/root"));
        assert_eq!(config.debounce_ms, 100);
        assert!(config.gitignore_aware);
    }

    #[test]
    fn test_watcher_config_default() {
        let config = WatcherConfig::default();

        assert_eq!(config.root_path, PathBuf::from("."));
        assert_eq!(config.debounce_ms, 500);
        assert!(config.gitignore_aware);
    }

    #[test]
    fn test_extract_dirty_paths_filters_traversal() {
        use std::fs;
        use tempfile::TempDir;

        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();

        let valid_file = root.join("valid.rs");
        fs::write(&valid_file, b"fn valid() {}").unwrap();

        let result = crate::validation::validate_path_within_root(&valid_file, root);
        assert!(result.is_ok());

        let outside = root.join("../../../etc/passwd");
        let result_outside = crate::validation::validate_path_within_root(&outside, root);
        assert!(result_outside.is_err());
    }

    #[test]
    fn test_is_mutation_event_accepts_create_modify_remove() {
        assert!(is_mutation_event(&EventKind::Create(
            notify::event::CreateKind::File
        )));
        assert!(is_mutation_event(&EventKind::Modify(
            notify::event::ModifyKind::Data(notify::event::DataChange::Content)
        )));
        assert!(is_mutation_event(&EventKind::Remove(
            notify::event::RemoveKind::File
        )));
        assert!(is_mutation_event(&EventKind::Any));
    }

    #[test]
    fn test_is_mutation_event_rejects_access() {
        assert!(!is_mutation_event(&EventKind::Access(
            notify::event::AccessKind::Read
        )));
        assert!(!is_mutation_event(&EventKind::Access(
            notify::event::AccessKind::Close(notify::event::AccessMode::Read)
        )));
    }

    #[test]
    fn test_filter_dirty_paths_excludes_db_files() {
        let temp_dir = tempfile::TempDir::new().unwrap();
        let root = temp_dir.path();
        let db_file = root.join("index.db");
        std::fs::write(&db_file, b"fake db").unwrap();

        let mut candidates = BTreeSet::new();
        candidates.insert(db_file);

        let result = filter_dirty_paths(candidates, root, None);
        assert!(result.is_empty());
    }
}