zwire_host/
watch.rs

1//! Streaming filesystem observers: `fs_watch` and `fs_tail`.
2//!
3//! `fs_watch` polls a file or directory and streams `{"ev":"fs","kind":…,"path":…}`
4//! frames on create / modify / remove. `fs_tail` streams `{"ev":"line","data":…}`
5//! as lines are appended to a file — `tail -f`, surviving truncation and log
6//! rotation. Each observer runs on a background thread keyed by the request `id`
7//! (so one connection can watch many paths), stops on `watch_stop`, and is torn
8//! down when the connection closes. Poll-based, so no new dependencies.
9use crate::proto::{send_msg, Out};
10use crate::store::expand;
11use serde_json::{json, Value};
12use std::collections::HashMap;
13use std::fs::File;
14use std::io::{Read, Seek, SeekFrom};
15use std::path::{Path, PathBuf};
16use std::sync::atomic::{AtomicBool, Ordering};
17use std::sync::Arc;
18use std::thread::JoinHandle;
19use std::time::{Duration, UNIX_EPOCH};
20
21/// Cap on entries scanned per `fs_watch` tick (guards watching a huge tree).
22const MAX_ENTRIES: usize = 20_000;
23/// Flush a `fs_tail` partial line once it grows past this without a newline.
24const MAX_LINE: usize = 1 << 20;
25
26/// A running observer. Dropping it stops the thread and joins it.
27pub struct Watcher {
28    stop: Arc<AtomicBool>,
29    handle: Option<JoinHandle<()>>,
30}
31
32impl Drop for Watcher {
33    fn drop(&mut self) {
34        self.stop.store(true, Ordering::Relaxed);
35        if let Some(h) = self.handle.take() {
36            let _ = h.join();
37        }
38    }
39}
40
41impl Watcher {
42    /// Watch `path` (file or directory) for changes; `recursive` descends
43    /// subdirectories, `interval_ms` sets the poll cadence (default 1000).
44    pub fn fs_watch(out: &Out, req: &Value, id: String) -> Watcher {
45        let path = expand(req["path"].as_str().unwrap_or("."));
46        let recursive = req["recursive"].as_bool().unwrap_or(false);
47        let interval = Duration::from_millis(req["interval_ms"].as_u64().unwrap_or(1000).max(100));
48        Self::spawn(out, move |out, stop| {
49            watch_loop(out, stop, &path, recursive, interval, &id)
50        })
51    }
52
53    /// Tail `path`, streaming appended lines. `from` = `"start"` replays the
54    /// whole file first; otherwise (default) only new lines are streamed.
55    pub fn fs_tail(out: &Out, req: &Value, id: String) -> Watcher {
56        let path = expand(req["path"].as_str().unwrap_or("."));
57        let from_end = req["from"].as_str() != Some("start");
58        let interval = Duration::from_millis(req["interval_ms"].as_u64().unwrap_or(300).max(50));
59        Self::spawn(out, move |out, stop| {
60            tail_loop(out, stop, &path, from_end, interval, &id)
61        })
62    }
63
64    /// Stream a small rewritten JSON file (the zwire HUD engine meter frames)
65    /// to the client on every change — a PUSH feed so the page never polls (which
66    /// starves during scroll / page build). Pushes `{"ev":"meter","text":…}`.
67    pub fn meter_stream(out: &Out, req: &Value, id: String) -> Watcher {
68        let path = expand(req["path"].as_str().unwrap_or("."));
69        let interval = Duration::from_millis(req["interval_ms"].as_u64().unwrap_or(33).max(16));
70        Self::spawn(out, move |out, stop| {
71            meter_loop(out, stop, &path, interval, &id)
72        })
73    }
74
75    fn spawn(out: &Out, body: impl FnOnce(&Out, &AtomicBool) + Send + 'static) -> Watcher {
76        let stop = Arc::new(AtomicBool::new(false));
77        let (s2, o) = (stop.clone(), out.clone());
78        let handle = std::thread::spawn(move || body(&o, &s2));
79        Watcher {
80            stop,
81            handle: Some(handle),
82        }
83    }
84}
85
86/// Sleep `dur`, waking early (every 100 ms) to notice a stop request.
87fn sleep_sliced(stop: &AtomicBool, dur: Duration) {
88    let mut slept = Duration::ZERO;
89    while slept < dur && !stop.load(Ordering::Relaxed) {
90        let slice = Duration::from_millis(100).min(dur - slept);
91        std::thread::sleep(slice);
92        slept += slice;
93    }
94}
95
96fn with_id(mut ev: Value, id: &str) -> Value {
97    if !id.is_empty() {
98        ev["id"] = json!(id);
99    }
100    ev
101}
102
103// Push the file's content on every change (mtime bump). Host-side thread, so it
104// is immune to the page's main-thread throttling during scroll / initial build.
105fn meter_loop(out: &Out, stop: &AtomicBool, path: &Path, interval: Duration, id: &str) {
106    let mut last = 0u64;
107    while !stop.load(Ordering::Relaxed) {
108        if let Some(mt) = mtime_ms(path) {
109            if mt != last {
110                last = mt;
111                if let Ok(bytes) = std::fs::read(path) {
112                    if let Ok(text) = String::from_utf8(bytes) {
113                        let frame = with_id(json!({ "ev": "meter", "text": text }), id);
114                        if send_msg(out, &frame).is_err() {
115                            break;
116                        }
117                    }
118                }
119            }
120        }
121        sleep_sliced(stop, interval);
122    }
123}
124
125fn mtime_ms(p: &Path) -> Option<u64> {
126    std::fs::metadata(p)
127        .ok()?
128        .modified()
129        .ok()?
130        .duration_since(UNIX_EPOCH)
131        .ok()
132        .map(|d| d.as_millis() as u64)
133}
134
135/// A `path -> mtime(ms)` snapshot of everything under `root`.
136fn snapshot(root: &Path, recursive: bool) -> HashMap<PathBuf, u64> {
137    let mut out = HashMap::new();
138    if root.is_file() {
139        if let Some(m) = mtime_ms(root) {
140            out.insert(root.to_path_buf(), m);
141        }
142        return out;
143    }
144    let mut stack = vec![(root.to_path_buf(), 0usize)];
145    while let Some((dir, depth)) = stack.pop() {
146        let Ok(rd) = std::fs::read_dir(&dir) else {
147            continue;
148        };
149        for e in rd.flatten() {
150            let p = e.path();
151            let is_dir = e.file_type().map(|t| t.is_dir()).unwrap_or(false);
152            if let Some(m) = mtime_ms(&p) {
153                out.insert(p.clone(), m);
154            }
155            if is_dir && recursive && depth < 32 {
156                stack.push((p, depth + 1));
157            }
158            if out.len() >= MAX_ENTRIES {
159                return out;
160            }
161        }
162    }
163    out
164}
165
166fn watch_loop(
167    out: &Out,
168    stop: &AtomicBool,
169    root: &Path,
170    recursive: bool,
171    interval: Duration,
172    id: &str,
173) {
174    let emit = |kind: &str, p: &Path| -> bool {
175        let frame = with_id(
176            json!({"ev": "fs", "kind": kind, "path": p.to_string_lossy()}),
177            id,
178        );
179        send_msg(out, &frame).is_ok()
180    };
181    let mut prev = snapshot(root, recursive);
182    while !stop.load(Ordering::Relaxed) {
183        sleep_sliced(stop, interval);
184        if stop.load(Ordering::Relaxed) {
185            break;
186        }
187        let cur = snapshot(root, recursive);
188        for (p, m) in &cur {
189            let changed = match prev.get(p) {
190                None => emit("created", p),
191                Some(pm) if pm != m => emit("modified", p),
192                _ => true,
193            };
194            if !changed {
195                return; // port closed
196            }
197        }
198        for p in prev.keys() {
199            if !cur.contains_key(p) && !emit("removed", p) {
200                return;
201            }
202        }
203        prev = cur;
204    }
205}
206
207fn tail_loop(
208    out: &Out,
209    stop: &AtomicBool,
210    path: &Path,
211    from_end: bool,
212    interval: Duration,
213    id: &str,
214) {
215    let mut pos: u64 = if from_end {
216        std::fs::metadata(path).map(|m| m.len()).unwrap_or(0)
217    } else {
218        0
219    };
220    let mut carry = String::new();
221    while !stop.load(Ordering::Relaxed) {
222        if let Ok(mut f) = File::open(path) {
223            let len = f.metadata().map(|m| m.len()).unwrap_or(0);
224            if len < pos {
225                // Truncated or rotated: start over from the top.
226                pos = 0;
227                carry.clear();
228            }
229            if len > pos && f.seek(SeekFrom::Start(pos)).is_ok() {
230                let mut buf = Vec::new();
231                if f.take(len - pos).read_to_end(&mut buf).is_ok() {
232                    pos += buf.len() as u64;
233                    carry.push_str(&String::from_utf8_lossy(&buf));
234                    if !flush_lines(out, &mut carry, id) {
235                        return; // port closed
236                    }
237                }
238            }
239        }
240        sleep_sliced(stop, interval);
241    }
242}
243
244/// Emit each complete line in `carry`; returns false if the port closed.
245fn flush_lines(out: &Out, carry: &mut String, id: &str) -> bool {
246    while let Some(idx) = carry.find('\n') {
247        let line: String = carry.drain(..=idx).collect();
248        let line = line.trim_end_matches(['\r', '\n']);
249        if !send_line(out, line, id) {
250            return false;
251        }
252    }
253    if carry.len() > MAX_LINE {
254        let line = std::mem::take(carry);
255        return send_line(out, &line, id);
256    }
257    true
258}
259
260fn send_line(out: &Out, line: &str, id: &str) -> bool {
261    send_msg(out, &with_id(json!({"ev": "line", "data": line}), id)).is_ok()
262}
zwire_host/watch.rs

zwire_host/
watch.rs
