openlatch-client 0.5.4

use std::collections::{BTreeMap, BTreeSet};
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use notify::{EventKind, RecommendedWatcher, RecursiveMode};
use notify_debouncer_full::{
    new_debouncer, new_debouncer_opt, DebounceEventResult, Debouncer, NoCache, RecommendedCache,
};
use tokio::sync::{mpsc, Notify};

use super::reconciler::request::ReconcileRequest;

/// Tamper-plane debounce window. Short enough to feel reactive yet long
/// enough that debouncer-full's file-id correlation collapses Vim /
/// JetBrains / VS Code atomic-save patterns into a single logical event.
const TAMPER_DEBOUNCE_MS: u64 = 500;

pub fn spawn_watcher(
    settings_path: &Path,
    tx: mpsc::Sender<ReconcileRequest>,
) -> Result<Debouncer<RecommendedWatcher, RecommendedCache>, notify::Error> {
    let (Some(dir), Some(target_name)) = (settings_path.parent(), settings_path.file_name()) else {
        return Err(notify::Error::generic(&format!(
            "'{}' has no parent directory or file name to watch",
            settings_path.display()
        )));
    };
    let dir = dir.to_path_buf();
    let target_name = target_name.to_os_string();

    let mut debouncer = new_debouncer(
        Duration::from_millis(TAMPER_DEBOUNCE_MS),
        None,
        move |res: DebounceEventResult| {
            if let Ok(events) = res {
                let hit = events
                    .iter()
                    .any(|ev| event_targets_settings(ev, &target_name));
                if hit && tx.try_send(ReconcileRequest::Fs).is_err() {
                    // Reconcile channel is full or closed. Fail-open: the
                    // 30s poll fallback will catch up. Per architecture.md.
                    tracing::warn!("reconcile channel full — skipping fs-triggered reconcile");
                }
            }
        },
    )?;

    debouncer.watch(&dir, RecursiveMode::NonRecursive)?;

    Ok(debouncer)
}

fn event_targets_settings(
    ev: &notify_debouncer_full::DebouncedEvent,
    target_name: &std::ffi::OsStr,
) -> bool {
    let is_relevant_kind = matches!(
        ev.event.kind,
        EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
    );
    if !is_relevant_kind {
        return false;
    }
    ev.event
        .paths
        .iter()
        .any(|p| p.file_name() == Some(target_name))
}

/// How long a file must be quiet before a change to it triggers a wiring pass.
///
/// An editor saves by writing a temp file and renaming it, and some save twice
/// in quick succession; waiting for the dust to settle turns that burst into one
/// pass reading the final bytes.
pub const FILE_TRIGGER_DEBOUNCE: Duration = Duration::from_millis(500);

/// A live watch on a set of files, notifying one [`tokio::sync::Notify`] when
/// any of them changes. Dropping it stops the watch.
pub struct FileTrigger {
    _debouncer: Debouncer<RecommendedWatcher, NoCache>,
    watched: Vec<PathBuf>,
}

impl FileTrigger {
    /// The files this trigger was armed for.
    pub fn watched(&self) -> &[PathBuf] {
        &self.watched
    }
}

/// Watch `files` and wake `trigger` whenever one is created, modified, renamed
/// over or removed.
///
/// Each file's PARENT directory is watched, non-recursively, and events are
/// filtered to the watched file names: an editor's atomic save replaces the
/// file, which a watch on the file itself would lose.
///
/// **No file-id cache, on every platform.** The debouncer's recommended cache
/// resolves a file id for every file in a watched directory, and on Windows that
/// OPENS each one. Cline's data directory holds `secrets.json` beside
/// `globalState.json`, and PRD C-6 forbids opening it — so the cache is
/// `NoCache`, and rename correlation, which only affects how events are merged,
/// is given up.
///
/// A file whose parent directory does not exist yet is skipped; the caller
/// re-arms on its next full pass, which is how a directory the agent creates
/// later gets watched. Returns `Ok(None)` when no file could be watched.
///
/// **Blocks** for as long as the OS takes to start the watch (FSEvents: about a
/// third of a second per directory), and dropping the trigger blocks too: call
/// both from a blocking context, never on a runtime worker.
pub fn spawn_file_trigger(
    files: &[PathBuf],
    trigger: Arc<Notify>,
) -> Result<Option<FileTrigger>, notify::Error> {
    let mut names: BTreeMap<PathBuf, BTreeSet<OsString>> = BTreeMap::new();
    for file in files {
        let (Some(dir), Some(name)) = (file.parent(), file.file_name()) else {
            continue;
        };
        if !dir.is_dir() {
            continue;
        }
        names
            .entry(dir.to_path_buf())
            .or_default()
            .insert(name.to_os_string());
    }
    if names.is_empty() {
        return Ok(None);
    }
    // The backend may report a directory by its resolved path: macOS FSEvents
    // names `/private/var/...` for a watch set on `/var/...`. Both spellings
    // are keys, so a symlinked data directory still triggers.
    let mut filter = names.clone();
    for (dir, files) in &names {
        if let Ok(resolved) = std::fs::canonicalize(dir) {
            if resolved != *dir {
                filter
                    .entry(resolved)
                    .or_default()
                    .extend(files.iter().cloned());
            }
        }
    }
    let mut debouncer = new_debouncer_opt::<_, RecommendedWatcher, NoCache>(
        FILE_TRIGGER_DEBOUNCE,
        None,
        move |res: DebounceEventResult| {
            let Ok(events) = res else {
                return;
            };
            let hit = events.iter().any(|ev| {
                matches!(
                    ev.event.kind,
                    EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_)
                ) && ev
                    .event
                    .paths
                    .iter()
                    .any(|p| match (p.parent(), p.file_name()) {
                        (Some(dir), Some(name)) => {
                            filter.get(dir).is_some_and(|wanted| wanted.contains(name))
                        }
                        _ => false,
                    })
            });
            if hit {
                trigger.notify_one();
            }
        },
        NoCache,
        notify::Config::default(),
    )?;
    let mut watched = Vec::new();
    for (dir, files) in &names {
        debouncer.watch(dir, RecursiveMode::NonRecursive)?;
        watched.extend(files.iter().map(|f| dir.join(f)));
    }
    Ok(Some(FileTrigger {
        _debouncer: debouncer,
        watched,
    }))
}

pub fn spawn_poll_fallback(tx: mpsc::Sender<ReconcileRequest>) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(30));
        interval.tick().await;

        loop {
            interval.tick().await;
            if tx.send(ReconcileRequest::Poll).await.is_err() {
                break;
            }
        }
    })
}

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

    async fn fired(trigger: &Notify, within: Duration) -> bool {
        tokio::time::timeout(within, trigger.notified())
            .await
            .is_ok()
    }

    /// An editor saves by writing a temp file and renaming it over the target;
    /// the watch is on the directory, so that save is seen.
    #[tokio::test(flavor = "multi_thread")]
    async fn an_atomic_save_of_a_watched_file_fires_the_trigger() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        std::fs::write(&state, "{}").expect("seed");
        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        // FSEvents on macOS starts delivering a beat after the watch is set up.
        tokio::time::sleep(Duration::from_millis(300)).await;

        let tmp = dir.path().join("globalState.json.tmp-editor");
        std::fs::write(&tmp, r#"{"actModeApiProvider":"ollama"}"#).expect("write temp");
        std::fs::rename(&tmp, &state).expect("rename over");

        assert!(fired(&notify, Duration::from_secs(5)).await);
    }

    /// A change to another file in the same directory is not a change to a
    /// watched file.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_sibling_change_does_not_fire_the_trigger() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        std::fs::write(&state, "{}").expect("seed");
        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        tokio::time::sleep(Duration::from_millis(300)).await;

        std::fs::write(dir.path().join("taskHistory.json"), "[]").expect("sibling");
        assert!(!fired(&notify, Duration::from_millis(1500)).await);
    }

    /// A file whose directory does not exist is skipped rather than failing the
    /// watch; the caller re-arms once the agent creates it.
    #[test]
    fn a_missing_directory_is_skipped() {
        let dir = tempfile::tempdir().expect("tempdir");
        let notify = Arc::new(Notify::new());
        let trigger = spawn_file_trigger(
            &[dir.path().join("not-yet").join("globalState.json")],
            notify,
        )
        .expect("no error");
        assert!(trigger.is_none());
    }

    /// PRD C-6: watching Cline's data directory must not open the secrets file
    /// beside the state file. The debouncer's file-id cache would, on Windows;
    /// this asserts the read-avoidance where atime can show it.
    #[cfg(unix)]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_watch_opens_no_sibling_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = dir.path().join("globalState.json");
        let secrets = dir.path().join("secrets.json");
        std::fs::write(&state, "{}").expect("seed");
        std::fs::write(&secrets, r#"{"apiKey":"SENTINEL"}"#).expect("seed secrets");
        let status = std::process::Command::new("touch")
            .args(["-a", "-t", "202001010000"])
            .arg(&secrets)
            .status()
            .expect("touch");
        assert!(status.success());
        let before = std::fs::metadata(&secrets)
            .expect("stat")
            .accessed()
            .expect("atime");

        let notify = Arc::new(Notify::new());
        let _trigger = spawn_file_trigger(std::slice::from_ref(&state), notify.clone())
            .expect("watch")
            .expect("armed");
        tokio::time::sleep(Duration::from_millis(300)).await;
        std::fs::write(&state, r#"{"a":1}"#).expect("save");
        assert!(
            fired(&notify, Duration::from_secs(5)).await,
            "positive control"
        );

        let after = std::fs::metadata(&secrets)
            .expect("stat")
            .accessed()
            .expect("atime");
        assert_eq!(before, after, "the watch opened secrets.json");
    }
}