travelagent-core 1.10.3

//! Live file-watcher primitives for "live review mode" (Phase L1).
//!
//! This module owns a [`notify::RecommendedWatcher`] rooted at the repo
//! root and emits debounced [`LiveEvent`]s on a bounded
//! [`tokio::sync::mpsc::Sender`] for the TUI to drain.
//!
//! Design goals:
//! - Debounce: batch events for [`DEBOUNCE_WINDOW`] (200 ms) after the last
//!   one arrives, with a [`DEBOUNCE_HARD_CAP`] (1 s) ceiling so large saves
//!   don't starve the UI.
//! - Ignore rules: drop events under `.git/`, `.jj/`, or `.hg/` (VCS
//!   metadata dirs that never meaningfully change the diff) AND under any
//!   path matched by the repo's `.gitignore` + `.trvignore` rules (so
//!   `node_modules/`, `target/`, `dist/` churn during an `npm install` /
//!   `cargo build` doesn't flood the debounce worker). User-level
//!   customisation belongs in `.trvignore`, not baked into this file.
//! - Bounded channels: the raw-event channel between the notify callback
//!   and the debounce pump uses [`try_send`] and drops on overflow. A
//!   burst of `npm install`-style events can't exhaust memory; the pump
//!   coalesces to one [`LiveEvent::Rescan`] anyway, so drops on the raw
//!   side just mean a slightly-earlier-than-strict rescan — no correctness
//!   loss.
//! - Errors surface as [`LiveEvent::WatcherError`] rather than panics — the
//!   TUI disables live mode and shows a status-bar error.

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use ignore::gitignore::Gitignore;
use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::mpsc::{Receiver, Sender, channel};
use tokio::task::JoinHandle;
use tokio::time::{Instant, sleep_until};

use crate::trvignore;

/// Debounce window: batch events for this long after the last arrival.
pub const DEBOUNCE_WINDOW: Duration = Duration::from_millis(200);

/// Hard cap on total debounce latency. Prevents a chatty long-running save
/// from starving the UI forever.
pub const DEBOUNCE_HARD_CAP: Duration = Duration::from_secs(1);

/// Raw-event channel capacity. The pump coalesces bursts into a single
/// `Rescan`, so a single pending token is enough to remember "something
/// changed, please rescan." A small slack buffer avoids dropping the
/// very first event of a batch under momentary scheduler pressure.
const RAW_CHANNEL_CAP: usize = 64;

/// Outgoing `LiveEvent` channel capacity. Events are coalesced to one
/// per debounce batch, so a shallow queue is fine.
const EVENT_CHANNEL_CAP: usize = 16;

/// An event delivered from the live watcher to the TUI.
#[derive(Debug)]
pub enum LiveEvent {
    /// A debounced batch of filesystem changes has settled — the TUI should
    /// rebuild the diff. Phase L1 treats every batch as a full rescan; L2
    /// will refine this to a per-file mapping for comment survival.
    Rescan,
    /// The watcher hit an unrecoverable error (permission denied, too many
    /// files, platform-specific failure, ...). The TUI should auto-disable
    /// live mode and surface `message` via `App::set_error`.
    WatcherError(String),
}

/// Return `true` if `path` is under a VCS metadata directory we never
/// want to rescan on. Component-based match, so a file literally named
/// `.git` or `.jj` under a repo (unusual but possible) is not watched —
/// but a source file that merely *contains* those strings is fine.
///
/// This is intentionally a small fixed list: it covers VCS dirs whose
/// churn never corresponds to a review-relevant change. User-level
/// ignores (`target/`, `node_modules/`, `dist/`, …) now belong in
/// `.gitignore` / `.trvignore` and are applied separately via
/// [`is_ignored_by_trv`], which is how [`should_forward_event`] composes
/// the two checks.
#[must_use]
pub fn is_vcs_metadata_path(path: &Path) -> bool {
    const VCS_DIRS: &[&str] = &[".git", ".jj", ".hg"];
    path.components().any(|component| {
        if let Some(name) = component.as_os_str().to_str() {
            VCS_DIRS.contains(&name)
        } else {
            false
        }
    })
}

/// Return `true` if `path` is ignored by the repo's `.gitignore` +
/// `.trvignore` rules (or either alone). Accepts an already-built
/// matcher so callers can cache it across many event checks.
#[must_use]
pub fn is_ignored_by_trv(matcher: &Gitignore, path: &Path) -> bool {
    trvignore::is_ignored(matcher, path)
}

/// Compatibility alias retained for any out-of-tree callers; returns
/// `true` for VCS metadata only. New code should call
/// [`is_vcs_metadata_path`] or compose with [`is_ignored_by_trv`].
#[must_use]
#[deprecated(note = "use is_vcs_metadata_path (+ is_ignored_by_trv for user ignores)")]
pub fn is_ignored_path(path: &Path) -> bool {
    is_vcs_metadata_path(path)
}

/// Handle to a running live watcher. Drop the handle to stop watching:
/// the `_watcher` field holds the `notify` handle whose Drop unregisters
/// from the platform backend, and the debounce task is aborted.
pub struct LiveWatcherHandle {
    _watcher: RecommendedWatcher,
    task: JoinHandle<()>,
}

impl LiveWatcherHandle {
    /// Stop the watcher. Equivalent to dropping the handle but more explicit
    /// at the call site.
    pub fn stop(self) {
        drop(self);
    }
}

impl Drop for LiveWatcherHandle {
    fn drop(&mut self) {
        // Abort the debounce pump so the spawned task can be reaped even if
        // the watcher channel still has queued events. The notify handle
        // drops naturally alongside.
        self.task.abort();
    }
}

/// Spawn a live watcher rooted at `root`, returning a handle and the
/// receiving end of the debounced event channel.
///
/// Must be called from within a Tokio runtime — the debounce pump uses
/// `tokio::spawn` and `tokio::time::sleep_until`.
///
/// Loads `.gitignore` + `.trvignore` once at spawn time. The watcher
/// continues to use the initial matcher for the life of the session;
/// editing an ignore file mid-session currently requires toggling
/// `:live!` off and on. Keeping the matcher immutable avoids locking
/// on the hot event path.
pub fn spawn_live_watcher(
    root: PathBuf,
) -> Result<(LiveWatcherHandle, Receiver<LiveEvent>), String> {
    let (event_tx, event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
    // Tokio channel between the notify callback and the debounce pump: the
    // callback converts raw filesystem events into a simple "something
    // changed" signal so the pump can focus purely on timing.
    let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);

    let matcher: Option<Arc<Gitignore>> = trvignore::load_matcher(&root).map(Arc::new);
    let matcher_cb = matcher.clone();

    let err_tx = event_tx.clone();
    let mut watcher = notify::recommended_watcher(move |res: notify::Result<Event>| match res {
        Ok(event) => {
            if should_forward_event(&event, matcher_cb.as_deref()) {
                // Bounded channel: drop on overflow. The debounce pump
                // coalesces multiple raw events into one `Rescan`, so a
                // dropped raw event just means the next accepted one
                // triggers the rescan that was going to happen anyway.
                let _ = raw_tx.try_send(());
            }
        }
        Err(e) => {
            // `try_send` + drop-on-overflow, matching the raw_tx policy
            // above. `blocking_send` here would park the notify worker
            // thread whenever the TUI stalled on draining `event_tx`
            // (16-slot bounded), halting ALL subsequent filesystem
            // notifications — the opposite of the "errors surface rather
            // than block" promise at the top of the module. Losing the
            // occasional error notification on a full channel is
            // preferable to freezing the event pipeline.
            let _ = err_tx.try_send(LiveEvent::WatcherError(format!(
                "filesystem watcher error: {e}"
            )));
        }
    })
    .map_err(|e| format!("failed to create filesystem watcher: {e}"))?;

    watcher
        .watch(&root, RecursiveMode::Recursive)
        .map_err(|e| format!("failed to watch {}: {e}", root.display()))?;

    let task = tokio::spawn(debounce_pump(raw_rx, event_tx));

    Ok((
        LiveWatcherHandle {
            _watcher: watcher,
            task,
        },
        event_rx,
    ))
}

/// Decide whether a raw `notify::Event` should feed the debounce pump.
/// Filters out metadata-only events, VCS metadata paths, and paths
/// ignored by the repo's `.gitignore` / `.trvignore` (when a matcher
/// was loaded).
fn should_forward_event(event: &Event, matcher: Option<&Gitignore>) -> bool {
    let relevant = matches!(
        event.kind,
        EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
    );
    if !relevant {
        return false;
    }
    // Forward if at least one path is neither VCS metadata nor matched
    // by .gitignore/.trvignore. Most events carry one path; the "any"
    // disjunction preserves the prior behavior for rare multi-path
    // events (atomic rename sometimes carries both from and to).
    event.paths.iter().any(|p| {
        if is_vcs_metadata_path(p) {
            return false;
        }
        if let Some(m) = matcher
            && is_ignored_by_trv(m, p)
        {
            return false;
        }
        true
    })
}

/// Debounce loop: collapse bursts of raw events into a single
/// [`LiveEvent::Rescan`].
///
/// Semantics:
/// 1. Block until the first raw event arrives (or the channel closes).
/// 2. Record the batch-start time and set a soft deadline at
///    `batch_start + DEBOUNCE_WINDOW`. Every subsequent event within that
///    window pushes the deadline out by `DEBOUNCE_WINDOW`, but never past
///    `batch_start + DEBOUNCE_HARD_CAP`.
/// 3. When the deadline elapses with no fresh event, emit a single
///    `LiveEvent::Rescan` and return to step 1.
///
/// Exposed at `pub(crate)` visibility so the unit tests can drive it
/// directly against a tokio channel without spinning up the real `notify`
/// stack.
pub(crate) async fn debounce_pump(mut raw_rx: Receiver<()>, event_tx: Sender<LiveEvent>) {
    loop {
        // Wait for the first event of a batch.
        if raw_rx.recv().await.is_none() {
            return;
        }
        let batch_start = Instant::now();
        let hard_deadline = batch_start + DEBOUNCE_HARD_CAP;
        let mut soft_deadline = batch_start + DEBOUNCE_WINDOW;

        loop {
            let deadline = soft_deadline.min(hard_deadline);
            tokio::select! {
                maybe = raw_rx.recv() => {
                    match maybe {
                        Some(()) => {
                            soft_deadline = (Instant::now() + DEBOUNCE_WINDOW).min(hard_deadline);
                        }
                        None => {
                            // Channel closed mid-batch — emit what we have and exit.
                            // `try_send` to avoid blocking on a full event queue at shutdown.
                            let _ = event_tx.try_send(LiveEvent::Rescan);
                            return;
                        }
                    }
                }
                () = sleep_until(deadline) => {
                    break;
                }
            }
        }

        // Bounded event channel: on overflow, drop. One queued rescan is
        // semantically equivalent to two — both mean "the UI needs to
        // rebuild the diff." The TUI drains greedily on each tick.
        let _ = event_tx.try_send(LiveEvent::Rescan);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ignore::gitignore::GitignoreBuilder;
    use std::path::PathBuf;

    // ── is_vcs_metadata_path ──

    #[test]
    fn flags_paths_under_dot_git_as_vcs_metadata() {
        assert!(is_vcs_metadata_path(&PathBuf::from(".git/HEAD")));
        assert!(is_vcs_metadata_path(&PathBuf::from(
            "/repo/.git/refs/heads/main"
        )));
    }

    #[test]
    fn flags_paths_under_jj_and_hg_as_vcs_metadata() {
        assert!(is_vcs_metadata_path(&PathBuf::from(".jj/repo/store")));
        assert!(is_vcs_metadata_path(&PathBuf::from(
            ".hg/store/00changelog.i"
        )));
    }

    #[test]
    fn does_not_flag_source_paths_as_vcs_metadata() {
        assert!(!is_vcs_metadata_path(&PathBuf::from("src/main.rs")));
        assert!(!is_vcs_metadata_path(&PathBuf::from(
            "crates/travelagent-core/src/lib.rs"
        )));
    }

    #[test]
    fn target_is_no_longer_hardcoded_as_vcs_metadata() {
        // H10: `target/` is no longer in the built-in VCS list — that
        // concern belongs in `.gitignore`. A repo without a `.gitignore`
        // rule for `target/` will now watch the directory (and the
        // diff-level `.trvignore` filter catches anything that slips
        // through before the user sees it).
        //
        // This test both flips the pre-H10 pinned over-match regression
        // AND regression-pins the new policy: the hardcoded list is
        // VCS-only, user ignores live in `.gitignore` / `.trvignore`.
        assert!(!is_vcs_metadata_path(&PathBuf::from("target")));
        assert!(!is_vcs_metadata_path(&PathBuf::from("src/target")));
        assert!(!is_vcs_metadata_path(&PathBuf::from("target/debug/app")));
    }

    // ── is_ignored_by_trv ──

    #[test]
    fn honors_gitignore_rule_for_target_directory() {
        // Mirrors a real repo: `target/` is ignored via `.gitignore`.
        // The live watcher now filters events under it via the
        // `.gitignore`/`.trvignore` matcher rather than a hardcoded list.
        let dir = tempfile::tempdir().expect("tmpdir");
        std::fs::write(dir.path().join(".gitignore"), "target/\n").expect("write");
        let matcher = trvignore::load_matcher(dir.path()).expect("matcher");

        assert!(is_ignored_by_trv(
            &matcher,
            &dir.path().join("target/debug/foo"),
        ));
        assert!(is_ignored_by_trv(&matcher, &dir.path().join("target")));
        assert!(!is_ignored_by_trv(
            &matcher,
            &dir.path().join("src/main.rs"),
        ));
    }

    #[test]
    fn honors_trvignore_unignore_over_gitignore() {
        let dir = tempfile::tempdir().expect("tmpdir");
        std::fs::write(dir.path().join(".gitignore"), "*.lock\n").expect("write");
        std::fs::write(dir.path().join(".trvignore"), "!Cargo.lock\n").expect("write");
        let matcher = trvignore::load_matcher(dir.path()).expect("matcher");

        assert!(!is_ignored_by_trv(&matcher, &dir.path().join("Cargo.lock"),));
        assert!(is_ignored_by_trv(&matcher, &dir.path().join("yarn.lock")));
    }

    // ── should_forward_event ──

    fn mk_event(kind: EventKind, paths: Vec<PathBuf>) -> Event {
        Event {
            kind,
            paths,
            attrs: notify::event::EventAttributes::new(),
        }
    }

    #[test]
    fn should_forward_event_drops_vcs_metadata_even_without_matcher() {
        let e = mk_event(
            EventKind::Modify(notify::event::ModifyKind::Any),
            vec![PathBuf::from(".git/index")],
        );
        assert!(!should_forward_event(&e, None));
    }

    #[test]
    fn should_forward_event_forwards_when_no_matcher_and_not_vcs() {
        let e = mk_event(
            EventKind::Modify(notify::event::ModifyKind::Any),
            vec![PathBuf::from("src/main.rs")],
        );
        assert!(should_forward_event(&e, None));
    }

    #[test]
    fn should_forward_event_drops_matches_against_gitignore() {
        // Build an absolute-rooted matcher so the paths we pass in have
        // a real prefix relationship with `repo_root`. The `ignore` crate
        // supports absolute paths natively when the matcher is rooted at
        // a real directory.
        let dir = tempfile::tempdir().expect("tmpdir");
        let root = dir.path();
        let mut builder = GitignoreBuilder::new(root);
        builder.add_line(None, "target/").expect("add_line");
        let matcher = builder.build().expect("build");

        let ignored = mk_event(
            EventKind::Modify(notify::event::ModifyKind::Any),
            vec![root.join("target/debug/foo")],
        );
        assert!(!should_forward_event(&ignored, Some(&matcher)));

        let kept = mk_event(
            EventKind::Modify(notify::event::ModifyKind::Any),
            vec![root.join("src/main.rs")],
        );
        assert!(should_forward_event(&kept, Some(&matcher)));
    }

    // ── debounce pump ──

    #[tokio::test(start_paused = true)]
    async fn emits_single_rescan_after_debounce_window() {
        let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);
        let (event_tx, mut event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
        let _task = tokio::spawn(debounce_pump(raw_rx, event_tx));

        raw_tx.send(()).await.unwrap();

        tokio::time::advance(DEBOUNCE_WINDOW + Duration::from_millis(10)).await;
        let evt = event_rx.recv().await.expect("rescan delivered");
        assert!(matches!(evt, LiveEvent::Rescan));

        // No further event without more input.
        tokio::time::advance(DEBOUNCE_HARD_CAP * 3).await;
        assert!(event_rx.try_recv().is_err(), "no extra rescan");
    }

    #[tokio::test(start_paused = true)]
    async fn coalesces_burst_into_one_rescan() {
        let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);
        let (event_tx, mut event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
        let _task = tokio::spawn(debounce_pump(raw_rx, event_tx));

        // Fire ten events, each 20 ms apart (well under the 200 ms window).
        for _ in 0..10 {
            raw_tx.send(()).await.unwrap();
            tokio::time::advance(Duration::from_millis(20)).await;
        }

        // Now let the debounce window drain.
        tokio::time::advance(DEBOUNCE_WINDOW + Duration::from_millis(50)).await;

        // Exactly one Rescan must have been emitted for the whole burst.
        let evt = event_rx.recv().await.expect("first rescan");
        assert!(matches!(evt, LiveEvent::Rescan));
        assert!(event_rx.try_recv().is_err(), "only one rescan");
    }

    #[tokio::test(start_paused = true)]
    async fn hard_cap_forces_emission_during_long_write() {
        let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);
        let (event_tx, mut event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
        let _task = tokio::spawn(debounce_pump(raw_rx, event_tx));

        // Drive events every 50 ms past the 1 s hard cap. The soft deadline
        // keeps getting pushed out, but the hard cap must still force a
        // Rescan.
        let total = DEBOUNCE_HARD_CAP + DEBOUNCE_HARD_CAP / 2;
        let step = Duration::from_millis(50);
        let steps = total.as_millis() / step.as_millis();
        for _ in 0..steps {
            raw_tx.send(()).await.unwrap();
            tokio::time::advance(step).await;
        }

        let evt = event_rx.recv().await.expect("hard-cap rescan");
        assert!(matches!(evt, LiveEvent::Rescan));
    }

    #[tokio::test(start_paused = true)]
    async fn bounded_raw_channel_drops_overflow_without_blocking() {
        // Regression (H10): the notify callback uses `try_send` on a
        // bounded raw channel. A burst that exceeds the channel capacity
        // must drop overflow events (not block, not panic), and the
        // pump must still coalesce whatever made it through into a
        // single Rescan.
        let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);
        let (event_tx, mut event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
        let _task = tokio::spawn(debounce_pump(raw_rx, event_tx));

        // Fire 10x the capacity via try_send. Overflow is dropped
        // silently — that's the whole point.
        let mut accepted = 0;
        for _ in 0..(RAW_CHANNEL_CAP * 10) {
            if raw_tx.try_send(()).is_ok() {
                accepted += 1;
            }
        }
        assert!(
            accepted >= 1,
            "at least the first send landed ({accepted} accepted)"
        );
        assert!(
            accepted <= RAW_CHANNEL_CAP + 1,
            "sender respects the bound ({accepted} <= {} + slack)",
            RAW_CHANNEL_CAP
        );

        // Drain the debounce window — one coalesced Rescan emerges.
        tokio::time::advance(DEBOUNCE_WINDOW + Duration::from_millis(50)).await;
        let evt = event_rx.recv().await.expect("rescan delivered");
        assert!(matches!(evt, LiveEvent::Rescan));
    }

    #[tokio::test(start_paused = true)]
    async fn two_spaced_batches_produce_two_rescans() {
        let (raw_tx, raw_rx) = channel::<()>(RAW_CHANNEL_CAP);
        let (event_tx, mut event_rx) = channel::<LiveEvent>(EVENT_CHANNEL_CAP);
        let _task = tokio::spawn(debounce_pump(raw_rx, event_tx));

        raw_tx.send(()).await.unwrap();
        tokio::time::advance(DEBOUNCE_WINDOW + Duration::from_millis(50)).await;
        let _ = event_rx.recv().await.expect("first rescan");

        raw_tx.send(()).await.unwrap();
        tokio::time::advance(DEBOUNCE_WINDOW + Duration::from_millis(50)).await;
        let _ = event_rx.recv().await.expect("second rescan");
    }
}