zeph_core/
config_watcher.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use std::path::Path;
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 ConfigWatcherError {
15    #[error("config path has no parent directory")]
16    NoParentDir,
17
18    #[error("config path has no filename")]
19    NoFilename,
20
21    #[error("filesystem watcher error: {0}")]
22    Notify(#[from] notify::Error),
23}
24
25#[non_exhaustive]
26#[derive(Clone)]
27pub enum ConfigEvent {
28    Changed,
29}
30
31pub struct ConfigWatcher {
32    _handle: BlockingHandle<()>,
33}
34
35impl ConfigWatcher {
36    /// Start watching a config file for changes.
37    ///
38    /// Watches the parent directory and filters for the target filename.
39    /// Sends `ConfigEvent::Changed` on any modification (debounced 500ms).
40    ///
41    /// # Errors
42    ///
43    /// Returns an error if the filesystem watcher cannot be initialized
44    /// or the config file path has no parent directory.
45    pub fn start(
46        path: &Path,
47        tx: mpsc::Sender<ConfigEvent>,
48        supervisor: &Arc<TaskSupervisor>,
49    ) -> Result<Self, ConfigWatcherError> {
50        let dir = path
51            .parent()
52            .ok_or(ConfigWatcherError::NoParentDir)?
53            .to_path_buf();
54        let filename = path
55            .file_name()
56            .ok_or(ConfigWatcherError::NoFilename)?
57            .to_os_string();
58
59        let (notify_tx, mut notify_rx) = mpsc::channel(16);
60
61        let mut debouncer = new_debouncer(
62            Duration::from_millis(500),
63            move |events: Result<Vec<notify_debouncer_mini::DebouncedEvent>, notify::Error>| {
64                let events = match events {
65                    Ok(events) => events,
66                    Err(e) => {
67                        tracing::warn!("config watcher error: {e}");
68                        return;
69                    }
70                };
71
72                let has_change = events.iter().any(|e| {
73                    e.kind == DebouncedEventKind::Any
74                        && e.path.file_name().is_some_and(|n| n == filename)
75                });
76
77                if has_change {
78                    let _ = notify_tx.blocking_send(());
79                }
80            },
81        )?;
82
83        debouncer
84            .watcher()
85            .watch(&dir, notify::RecursiveMode::NonRecursive)?;
86
87        let handle = supervisor.spawn_oneshot(
88            std::sync::Arc::from("core.config_watcher"),
89            move || async move {
90                let _debouncer = debouncer;
91                while notify_rx.recv().await.is_some() {
92                    if tx.send(ConfigEvent::Changed).await.is_err() {
93                        break;
94                    }
95                }
96            },
97        );
98
99        Ok(Self { _handle: handle })
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use tokio_util::sync::CancellationToken;
106    use zeph_common::TaskSupervisor;
107
108    use super::*;
109
110    fn make_supervisor() -> Arc<TaskSupervisor> {
111        Arc::new(TaskSupervisor::new(CancellationToken::new()))
112    }
113
114    #[tokio::test]
115    async fn start_with_valid_config_file() {
116        let dir = tempfile::tempdir().unwrap();
117        let config_path = dir.path().join("config.toml");
118        std::fs::write(&config_path, "key = 1").unwrap();
119        let (tx, _rx) = mpsc::channel(16);
120        let sup = make_supervisor();
121        let watcher = ConfigWatcher::start(&config_path, tx, &sup);
122        assert!(watcher.is_ok());
123    }
124
125    #[tokio::test]
126    async fn start_with_nonexistent_parent_fails() {
127        let (tx, _rx) = mpsc::channel(16);
128        let sup = make_supervisor();
129        let result = ConfigWatcher::start(Path::new("/nonexistent/dir/config.toml"), tx, &sup);
130        assert!(result.is_err());
131    }
132
133    #[tokio::test]
134    async fn detects_config_file_change() {
135        let dir = tempfile::tempdir().unwrap();
136        let config_path = dir.path().join("config.toml");
137        std::fs::write(&config_path, "initial = true").unwrap();
138
139        let (tx, mut rx) = mpsc::channel(16);
140        let sup = make_supervisor();
141        let _watcher = ConfigWatcher::start(&config_path, tx, &sup).unwrap();
142
143        tokio::time::sleep(Duration::from_millis(100)).await;
144        std::fs::write(&config_path, "updated = true").unwrap();
145
146        let result = tokio::time::timeout(Duration::from_secs(3), rx.recv()).await;
147        assert!(
148            result.is_ok(),
149            "expected ConfigEvent::Changed within timeout"
150        );
151    }
152
153    #[tokio::test]
154    async fn ignores_other_files_in_directory() {
155        let dir = tempfile::tempdir().unwrap();
156        let config_path = dir.path().join("config.toml");
157        std::fs::write(&config_path, "key = 1").unwrap();
158
159        let sup = make_supervisor();
160        let (tx, mut rx) = mpsc::channel(16);
161        let _watcher = ConfigWatcher::start(&config_path, tx, &sup).unwrap();
162
163        // Drain any late FSEvents from the initial config write before creating other file
164        tokio::time::sleep(Duration::from_millis(800)).await;
165        while rx.try_recv().is_ok() {}
166
167        let other_path = dir.path().join("other.txt");
168        std::fs::write(&other_path, "content").unwrap();
169
170        let result = tokio::time::timeout(Duration::from_millis(1500), rx.recv()).await;
171        assert!(
172            result.is_err(),
173            "should not receive event for non-config file"
174        );
175    }
176}
zeph_core/config_watcher.rs

zeph_core/
config_watcher.rs
