openlatch-client 0.1.18

//! Filesystem watcher for the configuration plane.
//!
//! Wraps `notify-debouncer-full` with a 500 ms debounce window so VS Code /
//! Vim / JetBrains atomic-save patterns collapse into one logical event.
//! When inotify exhausts its watch budget (`MaxFilesWatch` ≡ ENOSPC) the
//! affected paths fall back to `notify::PollWatcher` with a 5 s interval.
//!
//! User-scope paths are registered eagerly at daemon startup. Project-scope
//! paths land via `ConfigChangeRequest::ProjectScopeRegister` (Phase 2 wires
//! the SessionStart trigger; Phase 1 supports the channel but does not
//! actively watch project files on the FS).

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

use notify::{ErrorKind as NotifyErrorKind, EventKind, RecursiveMode};
use notify_debouncer_full::{
    new_debouncer, DebounceEventResult, DebouncedEvent, Debouncer, RecommendedCache,
};
use tokio::sync::mpsc;

use super::manifest::{expand_path, AgentPath, Manifest};
use super::monitor::ConfigChangeRequest;

/// Production debounce window for FS events. Aligned with the tamper-plane
/// watcher so the two surfaces share a single recency budget.
const PRODUCTION_DEBOUNCE_MS: u64 = 500;
/// PollWatcher fallback cadence used when inotify is exhausted (ENOSPC).
const POLL_FALLBACK_INTERVAL_SECS: u64 = 5;

/// Owns a live `notify` watcher. Dropping this guard unwatches all paths.
pub enum WatcherGuard {
    /// Standard debounced watcher (inotify / FSEvents / ReadDirectoryChangesW).
    Notify(Box<Debouncer<notify::RecommendedWatcher, RecommendedCache>>),
    /// Polling fallback used per-path when inotify ENOSPCs.
    Poll(notify::PollWatcher),
}

/// Spawn watchers for every user-scope path in the manifest.
///
/// Returns one or more `WatcherGuard`s — typically a single debouncer plus,
/// on Linux when inotify is exhausted, an additional `PollWatcher` covering
/// the ENOSPC paths. Every guard MUST stay alive for the daemon's lifetime;
/// dropping it terminates watching.
pub fn spawn_watchers(
    manifest: &Manifest,
    request_tx: mpsc::Sender<ConfigChangeRequest>,
    debounce_ms: u64,
) -> anyhow::Result<Vec<WatcherGuard>> {
    let user_scope_paths = collect_user_scope_paths(manifest);
    if user_scope_paths.is_empty() {
        tracing::debug!("config_monitor: no user-scope paths to watch");
        return Ok(Vec::new());
    }

    let mut guards: Vec<WatcherGuard> = Vec::new();

    // The debouncer callback runs on a background thread; it calls
    // `try_send` on the tokio channel and warns on backpressure.
    let cb_tx = request_tx.clone();
    let debounce_clamped = debounce_ms.clamp(50, 5000).max(PRODUCTION_DEBOUNCE_MS / 5);
    let debouncer_result = new_debouncer(
        Duration::from_millis(debounce_clamped),
        None,
        move |res: DebounceEventResult| match res {
            Ok(events) => {
                for ev in events {
                    if let Some(req) = event_to_request(&ev) {
                        if cb_tx.try_send(req).is_err() {
                            tracing::warn!(
                                "config_monitor: request channel full — skipping fs event"
                            );
                        }
                    }
                }
            }
            Err(errors) => {
                for e in errors {
                    tracing::debug!(error = %e, "config_monitor: debouncer reported error");
                }
            }
        },
    );

    let mut debouncer = match debouncer_result {
        Ok(d) => Box::new(d),
        Err(e) => {
            tracing::error!(
                code = crate::error::ERR_INVENTORY_WATCHER_FAILED,
                error = %e,
                "failed to create config_monitor debouncer; falling back to poll-only"
            );
            return spawn_poll_only_fallback(&user_scope_paths, request_tx);
        }
    };

    let mut enospc_paths: Vec<PathBuf> = Vec::new();
    for path in &user_scope_paths {
        // Watch the parent directory non-recursively so we observe atomic
        // saves (write tmp + rename) and creates of glob-matched files that
        // do not yet exist. The monitor loop filters events by manifest
        // membership before emitting CloudEvents.
        let watch_target = if path.is_dir() {
            path.clone()
        } else {
            match path.parent() {
                Some(p) => p.to_path_buf(),
                None => path.clone(),
            }
        };
        if !watch_target.exists() {
            // Don't error — many users won't have every declared path.
            tracing::debug!(
                path = %watch_target.display(),
                "config_monitor: watch target missing; skipping"
            );
            continue;
        }
        match debouncer.watch(&watch_target, RecursiveMode::NonRecursive) {
            Ok(_) => {}
            Err(e) if matches!(e.kind, NotifyErrorKind::MaxFilesWatch) => {
                tracing::warn!(
                    code = crate::error::ERR_INVENTORY_WATCHER_FAILED,
                    path = %watch_target.display(),
                    "inotify max_user_watches exhausted; falling back to PollWatcher"
                );
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_watcher_init_failed("enospc", None),
                );
                enospc_paths.push(watch_target);
            }
            Err(e) => {
                tracing::warn!(
                    code = crate::error::ERR_INVENTORY_WATCHER_FAILED,
                    path = %watch_target.display(),
                    error = %e,
                    "config_monitor: watcher failed for path; skipping"
                );
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_watcher_init_failed(os_error_class(&e), None),
                );
            }
        }
    }

    guards.push(WatcherGuard::Notify(debouncer));

    if !enospc_paths.is_empty() {
        if let Some(g) = spawn_poll_for_paths(&enospc_paths, request_tx)? {
            guards.push(g);
        }
    }

    Ok(guards)
}

/// Total fallback path: notify-debouncer-full could not be constructed at all
/// (rare — only seen on broken kernel ABIs). Spin up a `PollWatcher` over
/// every requested path with a 5 s interval.
fn spawn_poll_only_fallback(
    paths: &[PathBuf],
    request_tx: mpsc::Sender<ConfigChangeRequest>,
) -> anyhow::Result<Vec<WatcherGuard>> {
    crate::telemetry::capture_global(crate::telemetry::Event::config_watcher_init_failed(
        "other", None,
    ));
    if let Some(g) = spawn_poll_for_paths(paths, request_tx)? {
        Ok(vec![g])
    } else {
        Ok(Vec::new())
    }
}

fn spawn_poll_for_paths(
    paths: &[PathBuf],
    request_tx: mpsc::Sender<ConfigChangeRequest>,
) -> anyhow::Result<Option<WatcherGuard>> {
    if paths.is_empty() {
        return Ok(None);
    }
    let cfg = notify::Config::default()
        .with_poll_interval(Duration::from_secs(POLL_FALLBACK_INTERVAL_SECS))
        .with_compare_contents(true);

    let cb_tx = request_tx.clone();
    let mut watcher = notify::PollWatcher::new(
        move |res: notify::Result<notify::Event>| {
            if let Ok(ev) = res {
                if let Some(req) = native_event_to_request(&ev) {
                    if cb_tx.try_send(req).is_err() {
                        tracing::warn!(
                            "config_monitor: request channel full — skipping poll event"
                        );
                    }
                }
            }
        },
        cfg,
    )?;

    use notify::Watcher;
    for p in paths {
        let target = if p.is_dir() {
            p.clone()
        } else {
            match p.parent() {
                Some(parent) if parent.exists() => parent.to_path_buf(),
                _ => p.clone(),
            }
        };
        if !target.exists() {
            continue;
        }
        if let Err(e) = watcher.watch(&target, RecursiveMode::NonRecursive) {
            tracing::warn!(
                path = %target.display(),
                error = %e,
                "config_monitor: PollWatcher failed for fallback path"
            );
        }
    }

    Ok(Some(WatcherGuard::Poll(watcher)))
}

/// Walk the manifest and produce the deduplicated set of user-scope paths
/// to watch. `paths_glob` patterns are expanded against the filesystem; if
/// the pattern matches no entries we still register the parent directory
/// (so future creates fire events).
pub(crate) fn collect_user_scope_paths(manifest: &Manifest) -> Vec<PathBuf> {
    let mut out: Vec<PathBuf> = Vec::new();
    for agent in &manifest.agents {
        for ap in &agent.paths {
            if ap.is_project_scoped() {
                continue;
            }
            for p in &ap.paths {
                if let Ok(expanded) = expand_path(p, None) {
                    out.push(expanded);
                }
            }
            if let Some(g) = &ap.paths_glob {
                if let Ok(expanded) = expand_path(g, None) {
                    // Watch the directory holding the glob so creates fire.
                    if let Some(parent) = expanded.parent() {
                        out.push(parent.to_path_buf());
                    }
                    out.extend(glob_expand(&expanded));
                }
            }
            for slice in &ap.json_slice_paths {
                if let Ok(expanded) = expand_path(&slice.path, None) {
                    out.push(expanded);
                }
            }
        }
    }
    out.sort();
    out.dedup();
    out
}

/// Expand a glob pattern against the filesystem. Failures (invalid pattern,
/// I/O) return an empty Vec; the watcher loop logs nothing because missing
/// glob matches are normal (user has no skills directory yet).
pub(crate) fn glob_expand(pattern: &Path) -> Vec<PathBuf> {
    let pattern_str = match pattern.to_str() {
        Some(s) => s,
        None => return Vec::new(),
    };
    match glob::glob(pattern_str) {
        Ok(iter) => iter.flatten().collect(),
        Err(e) => {
            tracing::debug!(
                pattern = %pattern.display(),
                error = %e,
                "config_monitor: glob expansion failed"
            );
            Vec::new()
        }
    }
}

fn event_to_request(ev: &DebouncedEvent) -> Option<ConfigChangeRequest> {
    match ev.event.kind {
        EventKind::Create(_) => Some(ConfigChangeRequest::FsAdded(ev.event.paths.clone())),
        EventKind::Modify(_) => Some(ConfigChangeRequest::FsModified(ev.event.paths.clone())),
        EventKind::Remove(_) => Some(ConfigChangeRequest::FsRemoved(ev.event.paths.clone())),
        _ => None,
    }
}

fn native_event_to_request(ev: &notify::Event) -> Option<ConfigChangeRequest> {
    match ev.kind {
        EventKind::Create(_) => Some(ConfigChangeRequest::FsAdded(ev.paths.clone())),
        EventKind::Modify(_) => Some(ConfigChangeRequest::FsModified(ev.paths.clone())),
        EventKind::Remove(_) => Some(ConfigChangeRequest::FsRemoved(ev.paths.clone())),
        _ => None,
    }
}

fn os_error_class(e: &notify::Error) -> &'static str {
    match e.kind {
        NotifyErrorKind::MaxFilesWatch => "enospc",
        NotifyErrorKind::PathNotFound => "not_found",
        NotifyErrorKind::Io(_) => "io",
        _ => "other",
    }
}

/// Whether a path matches one of the manifest's exclude patterns. Used by
/// the monitor loop to suppress events for excluded paths (cache files,
/// session locks, OAuth blobs).
pub(crate) fn is_excluded(path: &Path, agent_path: &AgentPath, manifest: &Manifest) -> bool {
    // Find the parent agent block for `agent_path` so we can check its
    // `[agent.exclude]` list.
    for agent in &manifest.agents {
        let owns = agent.paths.iter().any(|ap| std::ptr::eq(ap, agent_path));
        if !owns {
            continue;
        }
        if let Some(exclude) = &agent.exclude {
            for pat in &exclude.patterns {
                if let Ok(expanded) = expand_path(pat, None) {
                    if path_matches_pattern(path, &expanded) {
                        return true;
                    }
                }
            }
        }
        break;
    }
    false
}

fn path_matches_pattern(path: &Path, pattern: &Path) -> bool {
    let Some(pattern_str) = pattern.to_str() else {
        return false;
    };
    let Some(path_str) = path.to_str() else {
        return false;
    };
    glob::Pattern::new(pattern_str)
        .map(|p| p.matches(path_str))
        .unwrap_or(false)
}

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

    #[test]
    fn collect_user_scope_paths_skips_project_scope() {
        let raw = r#"
            [[agent]]
            name = "claude-code"
            [[agent.path]]
            kind = "rules"
            scope = "project"
            paths_relative = ["CLAUDE.md"]
            watch_strategy = "exact_file"
            [[agent.path]]
            kind = "rules"
            paths = ["${OPENLATCH_DIR}/note.md"]
            watch_strategy = "exact_file"
        "#;
        let m: super::super::manifest::Manifest = toml::from_str(raw).unwrap();
        let paths = collect_user_scope_paths(&m);
        assert!(
            paths.iter().any(|p| p.ends_with("note.md")),
            "user-scope path must be present"
        );
        assert!(
            paths.iter().all(|p| !p.ends_with("CLAUDE.md")),
            "project-scope path must be skipped"
        );
    }

    #[test]
    fn glob_expand_returns_empty_on_no_matches() {
        let temp = tempfile::tempdir().unwrap();
        let pattern = temp.path().join("*.no_such_extension");
        let v = glob_expand(&pattern);
        assert!(v.is_empty());
    }

    #[test]
    fn os_error_class_maps_known_kinds() {
        let io_err = notify::Error::new(NotifyErrorKind::Io(std::io::Error::other("x")));
        assert_eq!(os_error_class(&io_err), "io");
        let not_found = notify::Error::new(NotifyErrorKind::PathNotFound);
        assert_eq!(os_error_class(&not_found), "not_found");
        let max = notify::Error::new(NotifyErrorKind::MaxFilesWatch);
        assert_eq!(os_error_class(&max), "enospc");
    }
}