zeph_core/
file_watcher.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use std::path::PathBuf;
5use std::sync::Arc;
6use std::time::Duration;
7
8use notify_debouncer_mini::{DebouncedEventKind, new_debouncer};
9use tokio::sync::mpsc;
10use zeph_common::{TaskSupervisor, task_supervisor::BlockingHandle};
11
12#[non_exhaustive]
13#[derive(Debug, thiserror::Error)]
14pub enum FileWatcherError {
15    #[error("no watch paths configured")]
16    NoWatchPaths,
17
18    #[error("filesystem watcher error: {0}")]
19    Notify(#[from] notify::Error),
20}
21
22/// Filesystem change event for a watched path.
23#[derive(Debug, Clone)]
24pub struct FileChangedEvent {
25    pub path: PathBuf,
26}
27
28/// Watches a set of paths and sends `FileChangedEvent` on any change.
29///
30/// Uses `notify-debouncer-mini` to debounce rapid filesystem events.
31/// Paths are resolved once at construction time from the project root.
32///
33/// Call `stop()` on the watcher to shut it down cleanly. The watcher
34/// is also stopped automatically when all senders are dropped.
35pub struct FileChangeWatcher {
36    handle: BlockingHandle<()>,
37}
38
39impl std::fmt::Debug for FileChangeWatcher {
40    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        f.debug_struct("FileChangeWatcher").finish_non_exhaustive()
42    }
43}
44
45impl Drop for FileChangeWatcher {
46    fn drop(&mut self) {
47        self.handle.abort();
48    }
49}
50
51impl FileChangeWatcher {
52    /// Start watching the given paths.
53    ///
54    /// `watch_paths` are watched recursively if they are directories.
55    /// Each path in `watch_paths` is watched with `RecursiveMode::Recursive`.
56    ///
57    /// # Errors
58    ///
59    /// Returns an error if no paths are provided or if the watcher cannot be initialized.
60    pub fn start(
61        watch_paths: &[PathBuf],
62        debounce_ms: u64,
63        tx: mpsc::Sender<FileChangedEvent>,
64        supervisor: &Arc<TaskSupervisor>,
65    ) -> Result<Self, FileWatcherError> {
66        if watch_paths.is_empty() {
67            return Err(FileWatcherError::NoWatchPaths);
68        }
69
70        let (notify_tx, mut notify_rx) = mpsc::channel::<PathBuf>(64);
71
72        let mut debouncer = new_debouncer(
73            Duration::from_millis(debounce_ms),
74            move |events: Result<Vec<notify_debouncer_mini::DebouncedEvent>, notify::Error>| {
75                let events = match events {
76                    Ok(e) => e,
77                    Err(e) => {
78                        tracing::warn!("file watcher error: {e}");
79                        return;
80                    }
81                };
82                for event in events {
83                    if event.kind == DebouncedEventKind::Any {
84                        let _ = notify_tx.blocking_send(event.path);
85                    }
86                }
87            },
88        )?;
89
90        for path in watch_paths {
91            if let Err(e) = debouncer
92                .watcher()
93                .watch(path, notify::RecursiveMode::Recursive)
94            {
95                tracing::warn!(path = %path.display(), error = %e, "file watcher: failed to watch path");
96            }
97        }
98
99        let handle = supervisor.spawn_oneshot(
100            std::sync::Arc::from("core.file_watcher"),
101            move || async move {
102                let _debouncer = debouncer;
103                while let Some(path) = notify_rx.recv().await {
104                    if tx.send(FileChangedEvent { path }).await.is_err() {
105                        break;
106                    }
107                }
108            },
109        );
110
111        Ok(Self { handle })
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use std::assert_matches;
118    use tokio_util::sync::CancellationToken;
119    use zeph_common::TaskSupervisor;
120
121    use super::*;
122
123    fn make_supervisor() -> Arc<TaskSupervisor> {
124        Arc::new(TaskSupervisor::new(CancellationToken::new()))
125    }
126
127    #[tokio::test]
128    async fn start_with_empty_paths_fails() {
129        let sup = make_supervisor();
130        let (tx, _rx) = mpsc::channel(16);
131        let result = FileChangeWatcher::start(&[], 500, tx, &sup);
132        assert!(result.is_err());
133        assert_matches!(result.unwrap_err(), FileWatcherError::NoWatchPaths);
134    }
135
136    #[tokio::test]
137    async fn start_with_valid_dir() {
138        let dir = tempfile::tempdir().unwrap();
139        let sup = make_supervisor();
140        let (tx, _rx) = mpsc::channel(16);
141        let result = FileChangeWatcher::start(&[dir.path().to_path_buf()], 500, tx, &sup);
142        assert!(result.is_ok());
143    }
144
145    #[tokio::test]
146    async fn detects_file_change() {
147        let dir = tempfile::tempdir().unwrap();
148        let file_path = dir.path().join("test.txt");
149        std::fs::write(&file_path, "initial").unwrap();
150
151        let sup = make_supervisor();
152        let (tx, mut rx) = mpsc::channel(16);
153        let _watcher =
154            FileChangeWatcher::start(&[dir.path().to_path_buf()], 500, tx, &sup).unwrap();
155
156        // Wait for watcher to settle before modifying.
157        tokio::time::sleep(Duration::from_millis(100)).await;
158        std::fs::write(&file_path, "updated").unwrap();
159
160        let result = tokio::time::timeout(Duration::from_secs(3), rx.recv()).await;
161        assert!(result.is_ok(), "expected FileChangedEvent within timeout");
162        // Event received — path granularity varies by OS/watcher backend (e.g. macOS FSEvents
163        // may return intermediate temp paths or symlink-resolved paths), so we only verify
164        // that an event arrived from within the watched directory tree.
165        assert!(result.unwrap().is_some());
166    }
167}
zeph_core/file_watcher.rs

zeph_core/
file_watcher.rs
