telar 0.2.1

//! Filesystem watching that arrives on the UI thread.
//!
//! The hard half of "reload when this file changes" is not noticing the change — `notify` does that — it is getting the notification onto the thread that owns the signals, since the watcher calls back from its own. Every app that wanted it wrote that bridge again, and one of them settled for polling mtime on a timer rather than build it.
//!
//! [`reactive_core::spawn_stream`] is the bridge, already here for exactly this shape: many values from a worker, each callback run on the UI thread during a later frame's `drain_tasks`.

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

use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use reactive_core::{Emitter, Task, spawn_stream};

/// One editor save is several filesystem events — a truncate, a write, a rename into place — and a directory copy is thousands. Events are collected until this long has passed with none, so a caller reloads once.
const COALESCE: Duration = Duration::from_millis(50);

/// How long the worker waits before checking whether it has been cancelled. It is parked the rest of the time; this only bounds how long a retired watcher's thread outlives the [`Task`] that owned it.
const CANCEL_POLL: Duration = Duration::from_millis(500);

/// Calls `on_change` on **this** thread whenever `path` changes — a file, or a directory and everything under it. Coalesced, so one save is one call however many events the platform reported.
///
/// The returned [`Task`] owns the watch: keep it for as long as the reload should happen, and drop or [`cancel`](Task::cancel) it to stop. Dropping it detaches instead — the watcher keeps running and the callback keeps firing — which is what a watch that should outlive its setup function wants.
///
/// `on_change` takes no argument on purpose. What changed is a question with a different answer on every platform (and no answer at all for a coalesced batch), while *something under here changed, re-read it* is the same everywhere and is what a reloading caller acts on.
///
/// ```ignore
/// let _watch = telar::watch_path(config_dir, move || settings.set(load_settings()));
/// ```
pub fn watch_path(path: impl Into<PathBuf>, mut on_change: impl FnMut() + 'static) -> Task {
    let path = path.into();
    spawn_stream(move |out| run(&path, out), move |()| on_change(), || {})
}

/// Whether an event is worth looking at the tree for. `Access` is the class inotify opens and reads under, so dropping it keeps a caller that re-reads on every change from waking itself forever.
///
/// A filter, not the answer: only inotify classifies that finely. FSEvents coalesces and re-labels, so what a read looks like there is decided by [`fingerprint`] instead.
fn worth_checking(event: &Event) -> bool {
    !matches!(event.kind, EventKind::Access(_))
}

/// What a change moves and a look does not: every path under `path`, with its length and its modification time. Reading a file leaves all three alone, so two equal fingerprints mean nothing happened however the platform labelled its event.
///
/// Cheap because it stats rather than reads, and only reached after a burst of events — which is already the moment the caller re-reads the tree anyway.
fn fingerprint(path: &Path) -> Vec<(PathBuf, u64, Option<SystemTime>)> {
    fn walk(path: &Path, out: &mut Vec<(PathBuf, u64, Option<SystemTime>)>) {
        let Ok(meta) = std::fs::symlink_metadata(path) else {
            return;
        };
        if meta.is_dir() {
            let Ok(entries) = std::fs::read_dir(path) else {
                return;
            };
            for entry in entries.flatten() {
                walk(&entry.path(), out);
            }
            return;
        }
        out.push((path.to_path_buf(), meta.len(), meta.modified().ok()));
    }
    let mut out = Vec::new();
    walk(path, &mut out);
    // `read_dir` answers in whatever order the filesystem holds, which is not stable across calls.
    out.sort();
    out
}

fn run(path: &Path, out: Emitter<()>) {
    let (tx, rx) = mpsc::channel();
    let mut watcher: RecommendedWatcher =
        match notify::recommended_watcher(move |result: notify::Result<Event>| {
            if result.is_ok_and(|event| worth_checking(&event)) {
                let _ = tx.send(());
            }
        }) {
            Ok(watcher) => watcher,
            Err(e) => {
                tracing::warn!("cannot watch {}: {e}", path.display());
                return;
            }
        };
    // Recursive whether `path` is a directory or a file: watching a file non-recursively is the same thing, and it saves the caller a `is_dir()` that races with the change it is asking to be told about.
    if let Err(e) = watcher.watch(path, RecursiveMode::Recursive) {
        tracing::warn!("cannot watch {}: {e}", path.display());
        return;
    }

    let mut seen = fingerprint(path);
    loop {
        match rx.recv_timeout(CANCEL_POLL) {
            Ok(()) => {}
            Err(mpsc::RecvTimeoutError::Timeout) => {
                if out.is_cancelled() {
                    return;
                }
                continue;
            }
            // The watcher was dropped, which only happens on the way out of this function.
            Err(mpsc::RecvTimeoutError::Disconnected) => return,
        }
        // Drain the rest of the burst before reporting, so a save that lands as four events is one reload.
        while rx.recv_timeout(COALESCE).is_ok() {}
        if out.is_cancelled() {
            return;
        }
        let now = fingerprint(path);
        if now == seen {
            continue;
        }
        seen = now;
        out.emit(());
    }
}

#[cfg(test)]
#[path = "watch_test.rs"]
mod tests;