nornir 0.5.3

//! **inotify watch mode** (Linux) — event-driven delta re-scans (design doc
//! `P0.2`). The steady-state introspection trigger today is a timer-poll
//! (`monitor::sync_once`); this turns kernel file-change events into *targeted*
//! [`crate::knowledge::delta`] re-scans, so a save fires a re-index in
//! milliseconds instead of on the next poll tick.
//!
//! ## Charter & degradation
//! - **No new dependency.** The watch rides `libc`'s `inotify` syscalls directly
//!   (`libc` is already a hard dep), honouring the pure-Rust / reuse-existing-
//!   crates charter — no `notify`/`inotify` crate pulled in.
//! - **Feature- and platform-gated.** The real watcher is
//!   `#[cfg(all(feature = "inotify", target_os = "linux"))]`. Everywhere else
//!   (feature off, or non-Linux) the same [`watch`] symbol exists but returns
//!   [`WatchError::Unsupported`] — callers degrade to the timer-poll cleanly,
//!   never a compile break, never a panic.
//!
//! ## Debounce
//! An editor "save" is several inotify events (a temp write + rename, a
//! `MODIFY`, a `CLOSE_WRITE`, …). [`Debouncer`] coalesces a burst into one
//! delta: paths accumulate until the stream is quiet for `debounce`, then the
//! batch is handed to the callback exactly once. The debounce policy is a pure,
//! unit-tested state machine independent of any syscall.

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

/// Why a watch could not run (or stopped). Non-fatal to the caller: fall back to
/// the timer-poll on [`Unsupported`](WatchError::Unsupported).
#[derive(Debug)]
pub enum WatchError {
    /// inotify is unavailable here — the `inotify` feature is off, or the target
    /// is not Linux. The message names which. Callers should degrade to polling.
    Unsupported(String),
    /// A syscall failed after the watch started.
    Io(std::io::Error),
}

impl std::fmt::Display for WatchError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            WatchError::Unsupported(s) => write!(f, "inotify watch unavailable: {s}"),
            WatchError::Io(e) => write!(f, "inotify watch io error: {e}"),
        }
    }
}
impl std::error::Error for WatchError {}

/// True iff a real inotify watch can run in this build on this platform.
pub const fn is_supported() -> bool {
    cfg!(all(feature = "inotify", target_os = "linux"))
}

/// A batch of changed `.rs` paths coalesced from one inotify burst — what the
/// [`watch`] callback receives, and what a caller turns into a targeted
/// [`crate::knowledge::delta::SymbolDelta::rescan`].
pub type ChangedBatch = Vec<PathBuf>;

/// Pure debounce state machine (no syscalls) — coalesce a burst of change events
/// into one batch that fires only after `quiet` of silence. Unit-tested directly;
/// the inotify loop feeds it real events, tests feed it synthetic ones.
#[derive(Debug)]
pub struct Debouncer {
    quiet: Duration,
    pending: std::collections::BTreeSet<PathBuf>,
    last_event: Option<Instant>,
}

impl Debouncer {
    pub fn new(quiet: Duration) -> Self {
        Self { quiet, pending: Default::default(), last_event: None }
    }

    /// Record a changed path (only `.rs` files are retained). Resets the quiet
    /// timer to `now`.
    pub fn record(&mut self, path: PathBuf, now: Instant) {
        if path.extension().and_then(|e| e.to_str()) == Some("rs") {
            self.pending.insert(path);
            self.last_event = Some(now);
        }
    }

    /// If a batch is pending AND the stream has been quiet for `quiet`, drain and
    /// return it; otherwise `None`. Idempotent — after draining, `pending` is
    /// empty until the next `record`.
    pub fn take_ready(&mut self, now: Instant) -> Option<ChangedBatch> {
        let last = self.last_event?;
        if self.pending.is_empty() || now.duration_since(last) < self.quiet {
            return None;
        }
        self.last_event = None;
        Some(std::mem::take(&mut self.pending).into_iter().collect())
    }

    /// How long until the pending batch is ready (for the poll timeout), or
    /// `None` when nothing is pending.
    pub fn time_to_ready(&self, now: Instant) -> Option<Duration> {
        let last = self.last_event?;
        if self.pending.is_empty() {
            return None;
        }
        Some(self.quiet.saturating_sub(now.duration_since(last)))
    }

    pub fn has_pending(&self) -> bool {
        !self.pending.is_empty()
    }
}

/// Directory names never watched (mirrors the scan's `is_skipped_dir`): build
/// output, VCS, tooling. Used only by the real (feature+Linux) watcher.
#[cfg(all(feature = "inotify", target_os = "linux"))]
pub(crate) fn is_skipped_dir(name: &str) -> bool {
    matches!(name, "target" | ".git" | "node_modules" | ".nornir" | ".claude")
}

// ─────────────────────────────────────────────────────────────────────────────
// Real implementation — Linux + `inotify` feature.
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(all(feature = "inotify", target_os = "linux"))]
mod imp {
    use super::*;
    use std::collections::HashMap;
    use std::os::unix::ffi::OsStrExt;
    use std::path::Path;
    use std::sync::atomic::{AtomicBool, Ordering};

    const WATCH_MASK: u32 = libc::IN_CLOSE_WRITE
        | libc::IN_CREATE
        | libc::IN_DELETE
        | libc::IN_MOVED_FROM
        | libc::IN_MOVED_TO;

    /// Recursively add an inotify watch to `dir` and every scannable sub-dir,
    /// recording `wd → dir` so events can be resolved back to a full path.
    fn add_watches_recursive(fd: i32, dir: &Path, wds: &mut HashMap<i32, PathBuf>) {
        add_one(fd, dir, wds);
        for entry in walkdir::WalkDir::new(dir)
            .into_iter()
            .filter_entry(|e| !super::is_skipped_dir(&e.file_name().to_string_lossy()))
        {
            let Ok(entry) = entry else { continue };
            if entry.file_type().is_dir() && entry.path() != dir {
                add_one(fd, entry.path(), wds);
            }
        }
    }

    fn add_one(fd: i32, dir: &Path, wds: &mut HashMap<i32, PathBuf>) {
        let Ok(c) = std::ffi::CString::new(dir.as_os_str().as_bytes()) else { return };
        // SAFETY: `fd` is a live inotify fd, `c` a valid NUL-terminated path.
        let wd = unsafe { libc::inotify_add_watch(fd, c.as_ptr(), WATCH_MASK) };
        if wd >= 0 {
            wds.insert(wd, dir.to_path_buf());
        }
    }

    /// Run the inotify loop until `stop` is set, coalescing bursts and invoking
    /// `on_delta` once per quiet batch. Blocking — run it on its own thread.
    pub fn watch(
        repo_root: &Path,
        debounce: Duration,
        stop: &AtomicBool,
        mut on_delta: impl FnMut(ChangedBatch),
    ) -> Result<(), WatchError> {
        // SAFETY: no args; returns a new inotify fd or -1.
        let fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) };
        if fd < 0 {
            return Err(WatchError::Io(std::io::Error::last_os_error()));
        }
        let _guard = FdGuard(fd);

        let mut wds: HashMap<i32, PathBuf> = HashMap::new();
        add_watches_recursive(fd, repo_root, &mut wds);

        let mut deb = Debouncer::new(debounce);
        // inotify_event is 16 bytes + a NUL-terminated name; size the read buffer
        // for a healthy burst.
        let mut buf = [0u8; 64 * 1024];

        while !stop.load(Ordering::Relaxed) {
            // Poll with a timeout so we can (a) honour `stop` and (b) fire a
            // ready batch even with no further events.
            let now = Instant::now();
            let timeout_ms = deb
                .time_to_ready(now)
                .map(|d| d.as_millis().min(1_000) as i32)
                .unwrap_or(250);
            let mut pfd = libc::pollfd { fd, events: libc::POLLIN, revents: 0 };
            // SAFETY: single valid pollfd, count 1.
            let pr = unsafe { libc::poll(&mut pfd, 1, timeout_ms.max(1)) };
            if pr < 0 {
                let e = std::io::Error::last_os_error();
                if e.kind() == std::io::ErrorKind::Interrupted {
                    continue;
                }
                return Err(WatchError::Io(e));
            }

            if pr > 0 && (pfd.revents & libc::POLLIN) != 0 {
                let saw_dir_create = drain_events(fd, &mut buf, &wds, &mut deb);
                // Newly created sub-dirs: re-add recursively so their files are
                // covered on the NEXT event (cheap; only when dirs appear).
                if saw_dir_create {
                    add_watches_recursive(fd, repo_root, &mut wds);
                }
            }

            if let Some(batch) = deb.take_ready(Instant::now()) {
                if !batch.is_empty() {
                    on_delta(batch);
                }
            }
        }
        Ok(())
    }

    /// Read + parse all currently-queued inotify events into the debouncer.
    /// Returns `true` if any event created a directory (⇒ re-add watches).
    fn drain_events(
        fd: i32,
        buf: &mut [u8],
        wds: &HashMap<i32, PathBuf>,
        deb: &mut Debouncer,
    ) -> bool {
        let mut saw_dir_create = false;
        loop {
            // SAFETY: read into an owned buffer of known length.
            let n = unsafe {
                libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len())
            };
            if n <= 0 {
                break; // EAGAIN (nothing left) or EOF
            }
            let n = n as usize;
            let hdr = std::mem::size_of::<libc::inotify_event>();
            let mut off = 0usize;
            while off + hdr <= n {
                // SAFETY: `off + hdr <= n` and buf is initialised for [0,n).
                let ev = unsafe { &*(buf.as_ptr().add(off) as *const libc::inotify_event) };
                let name_len = ev.len as usize;
                let name_start = off + hdr;
                let name_end = name_start + name_len;
                if name_end > n {
                    break;
                }
                let is_dir = ev.mask & libc::IN_ISDIR != 0;
                if is_dir && ev.mask & libc::IN_CREATE != 0 {
                    saw_dir_create = true;
                }
                if !is_dir {
                    if let Some(dir) = wds.get(&ev.wd) {
                        // Name is NUL-padded; cut at the first NUL.
                        let raw = &buf[name_start..name_end];
                        let name_bytes = raw.split(|&b| b == 0).next().unwrap_or(&[]);
                        if !name_bytes.is_empty() {
                            let name = std::ffi::OsStr::from_bytes(name_bytes);
                            deb.record(dir.join(name), Instant::now());
                        }
                    }
                }
                off = name_end;
            }
        }
        saw_dir_create
    }

    /// Close the inotify fd on drop.
    struct FdGuard(i32);
    impl Drop for FdGuard {
        fn drop(&mut self) {
            // SAFETY: `self.0` is the live fd we own.
            unsafe { libc::close(self.0) };
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Fallback — feature off or non-Linux. Clean degradation, no compile break.
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(not(all(feature = "inotify", target_os = "linux")))]
mod imp {
    use super::*;
    use std::path::Path;
    use std::sync::atomic::AtomicBool;

    pub fn watch(
        _repo_root: &Path,
        _debounce: Duration,
        _stop: &AtomicBool,
        _on_delta: impl FnMut(ChangedBatch),
    ) -> Result<(), WatchError> {
        Err(WatchError::Unsupported(if cfg!(target_os = "linux") {
            "build without the `inotify` feature".to_string()
        } else {
            "inotify is Linux-only".to_string()
        }))
    }
}

/// Watch `repo_root` for `.rs` changes, invoking `on_delta` once per debounced
/// burst with the changed paths. Blocks until `stop` is set (run on its own
/// thread). Returns [`WatchError::Unsupported`] where inotify is unavailable —
/// the caller then keeps the timer-poll.
///
/// See [`Debouncer`] for the coalescing policy and [`is_supported`] to branch
/// before spawning.
pub fn watch(
    repo_root: &std::path::Path,
    debounce: Duration,
    stop: &std::sync::atomic::AtomicBool,
    on_delta: impl FnMut(ChangedBatch),
) -> Result<(), WatchError> {
    imp::watch(repo_root, debounce, stop, on_delta)
}

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

    #[test]
    fn debouncer_coalesces_a_burst_into_one_batch() {
        let quiet = Duration::from_millis(50);
        let mut d = Debouncer::new(quiet);
        let t0 = Instant::now();
        // A "save" burst: three events on two files within the quiet window.
        d.record(PathBuf::from("/w/a.rs"), t0);
        d.record(PathBuf::from("/w/a.rs"), t0 + Duration::from_millis(5));
        d.record(PathBuf::from("/w/b.rs"), t0 + Duration::from_millis(10));
        // Still inside the quiet window ⇒ nothing ready yet.
        assert!(d.take_ready(t0 + Duration::from_millis(30)).is_none());
        // After quiet elapses ⇒ exactly one coalesced batch of the two files.
        let batch = d.take_ready(t0 + Duration::from_millis(200)).expect("ready");
        assert_eq!(batch.len(), 2);
        assert!(batch.contains(&PathBuf::from("/w/a.rs")));
        assert!(batch.contains(&PathBuf::from("/w/b.rs")));
        // Drained ⇒ nothing more.
        assert!(d.take_ready(t0 + Duration::from_millis(300)).is_none());
    }

    #[test]
    fn debouncer_ignores_non_rs_files() {
        let mut d = Debouncer::new(Duration::from_millis(10));
        let t0 = Instant::now();
        d.record(PathBuf::from("/w/Cargo.toml"), t0);
        d.record(PathBuf::from("/w/notes.md"), t0);
        assert!(!d.has_pending(), "non-.rs events must not arm the debouncer");
        assert!(d.take_ready(t0 + Duration::from_secs(1)).is_none());
    }

    #[test]
    fn debouncer_resets_timer_on_each_event() {
        let quiet = Duration::from_millis(50);
        let mut d = Debouncer::new(quiet);
        let t0 = Instant::now();
        d.record(PathBuf::from("/w/a.rs"), t0);
        // A second event at t0+40 pushes readiness out to t0+90, not t0+50.
        d.record(PathBuf::from("/w/a.rs"), t0 + Duration::from_millis(40));
        assert!(d.take_ready(t0 + Duration::from_millis(60)).is_none(), "timer reset by 2nd event");
        assert!(d.take_ready(t0 + Duration::from_millis(100)).is_some());
    }

    #[test]
    fn unsupported_is_reported_not_panicked_when_feature_off() {
        // When the feature is off this build returns Unsupported; when on (Linux)
        // is_supported() is true. Either way the call must not panic.
        assert_eq!(is_supported(), cfg!(all(feature = "inotify", target_os = "linux")));
    }
}