Watch selected folders and reconcile bounded changes with media freshness guards

apps/desktop/src-tauri/src/watching.rs

@@ -0,0 +1,644 @@
+//! Native notifications are bounded hints; periodic reconciliation is authoritative.
+use notify::{Event, EventKind, RecursiveMode, Watcher};
+use serde::Serialize;
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    path::{Path, PathBuf},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Condvar, Mutex,
+    },
+    thread::{self, JoinHandle},
+    time::{Duration, Instant},
+};
+
+const MAX_ROOTS: usize = 64;
+const MAX_PATHS: usize = 128;
+
+pub struct ScanRequest {
+    pub root: PathBuf,
+    pub paths: Option<Vec<PathBuf>>,
+    pub cancelled: Arc<AtomicBool>,
+    pub media_hint: bool,
+}
+pub enum ScanOutcome {
+    Completed,
+    Busy,
+    Failed(String),
+}
+#[derive(Clone, Serialize)]
+pub struct WatchStatus {
+    pub root: PathBuf,
+    pub attached: bool,
+    pub pending: bool,
+    pub error: Option<String>,
+    pub watch_error: Option<String>,
+}
+#[derive(Clone, Copy)]
+struct Timing {
+    debounce: Duration,
+    max_delay: Duration,
+    periodic: Duration,
+    retry: Duration,
+}
+impl Default for Timing {
+    fn default() -> Self {
+        Self {
+            debounce: Duration::from_millis(750),
+            max_delay: Duration::from_secs(5),
+            periodic: Duration::from_secs(300),
+            retry: Duration::from_secs(5),
+        }
+    }
+}
+struct Root {
+    paths: BTreeSet<PathBuf>,
+    full: bool,
+    media_hint: bool,
+    ever_attached: bool,
+    first: Option<Instant>,
+    last: Instant,
+    next_scan: Instant,
+    retry_at: Instant,
+    attach_retry: Instant,
+    failures: u32,
+    attached: bool,
+    error: Option<String>,
+    watch_error: Option<String>,
+}
+impl Root {
+    fn new(now: Instant) -> Self {
+        Self {
+            paths: BTreeSet::new(),
+            full: true,
+            media_hint: false,
+            ever_attached: false,
+            first: Some(now),
+            last: now,
+            next_scan: now,
+            retry_at: now,
+            attach_retry: now,
+            failures: 0,
+            attached: false,
+            error: None,
+            watch_error: None,
+        }
+    }
+    fn dirty(&mut self, paths: Option<&[PathBuf]>, now: Instant, media_hint: bool) {
+        self.media_hint |= media_hint;
+        self.first.get_or_insert(now);
+        self.last = now;
+        match paths {
+            Some(paths) if !self.full => {
+                self.paths.extend(paths.iter().cloned());
+                if self.paths.len() > MAX_PATHS {
+                    self.paths.clear();
+                    self.full = true;
+                }
+            }
+            None => {
+                self.paths.clear();
+                self.full = true;
+            }
+            _ => {}
+        }
+    }
+}
+struct State {
+    roots: BTreeMap<PathBuf, Root>,
+    stop: bool,
+}
+struct Shared {
+    state: Mutex<State>,
+    changed: Condvar,
+    ignored: Vec<PathBuf>,
+    timing: Timing,
+    cancelled: Arc<AtomicBool>,
+}
+/// Own this service for the application lifetime. Dropping joins its worker after
+/// the currently executing reconciliation callback finishes.
+pub struct WatchService {
+    shared: Arc<Shared>,
+    worker: Mutex<Option<JoinHandle<()>>>,
+}
+impl WatchService {
+    pub fn start(
+        roots: Vec<PathBuf>,
+        ignored: Vec<PathBuf>,
+        callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
+    ) -> Result<Self, String> {
+        Self::start_timed(roots, ignored, callback, Timing::default())
+    }
+    fn start_timed(
+        roots: Vec<PathBuf>,
+        ignored: Vec<PathBuf>,
+        callback: impl Fn(ScanRequest) -> ScanOutcome + Send + 'static,
+        timing: Timing,
+    ) -> Result<Self, String> {
+        validate(&roots)?;
+        if ignored.len() > 128 || ignored.iter().any(|p| !p.is_absolute()) {
+            return Err("Watcher exclusions must be bounded absolute paths.".into());
+        }
+        let shared = Arc::new(Shared {
+            state: Mutex::new(State {
+                roots: BTreeMap::new(),
+                stop: false,
+            }),
+            changed: Condvar::new(),
+            ignored,
+            timing,
+            cancelled: Arc::new(AtomicBool::new(false)),
+        });
+        let target = shared.clone();
+        let worker = thread::Builder::new()
+            .name("searchi-reconcile".into())
+            .spawn(move || run(target, callback))
+            .map_err(|e| e.to_string())?;
+        let service = Self {
+            shared,
+            worker: Mutex::new(Some(worker)),
+        };
+        service.replace_roots(roots)?;
+        Ok(service)
+    }
+    /// Stop and join only this owned worker. Call on application exit to break
+    /// callback ownership of the desktop emitter. Cancellation waits for the
+    /// current bounded parser operation to reach its next cooperative check.
+    pub fn stop(&self) {
+        self.shared.cancelled.store(true, Ordering::Release);
+        if let Ok(mut state) = self.shared.state.lock() {
+            state.stop = true;
+            self.shared.changed.notify_all();
+        }
+        let worker = self.worker.lock().unwrap_or_else(|e| e.into_inner()).take();
+        if let Some(worker) = worker {
+            if worker.thread().id() != thread::current().id() {
+                let _ = worker.join();
+            }
+        }
+    }
+
+    pub fn add_root(&self, root: PathBuf) -> Result<(), String> {
+        let mut s = self
+            .shared
+            .state
+            .lock()
+            .map_err(|_| "Watcher is unavailable.")?;
+        if s.roots.contains_key(&root) {
+            return Ok(());
+        }
+        if s.roots.len() >= MAX_ROOTS || !root.is_absolute() {
+            return Err("Watcher folders must be absolute and limited to 64.".into());
+        }
+        s.roots.insert(root, Root::new(Instant::now()));
+        self.shared.changed.notify_all();
+        Ok(())
+    }
+    pub fn replace_roots(&self, roots: Vec<PathBuf>) -> Result<(), String> {
+        validate(&roots)?;
+        let mut s = self
+            .shared
+            .state
+            .lock()
+            .map_err(|_| "Watcher is unavailable.")?;
+        s.roots.retain(|p, _| roots.contains(p));
+        for p in roots {
+            s.roots
+                .entry(p)
+                .or_insert_with(|| Root::new(Instant::now()));
+        }
+        self.shared.changed.notify_all();
+        Ok(())
+    }
+    pub fn status(&self) -> Vec<WatchStatus> {
+        self.shared
+            .state
+            .lock()
+            .map(|s| {
+                s.roots
+                    .iter()
+                    .map(|(p, r)| WatchStatus {
+                        root: p.clone(),
+                        attached: r.attached,
+                        pending: r.first.is_some(),
+                        error: r.error.clone().or_else(|| r.watch_error.clone()),
+                        watch_error: r.watch_error.clone(),
+                    })
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+}
+impl Drop for WatchService {
+    fn drop(&mut self) {
+        self.stop();
+    }
+}
+fn validate(roots: &[PathBuf]) -> Result<(), String> {
+    if roots.len() > MAX_ROOTS || roots.iter().any(|p| !p.is_absolute()) {
+        return Err("Watcher folders must be absolute and limited to 64.".into());
+    }
+    Ok(())
+}
+fn excluded(path: &Path, shared: &Shared) -> bool {
+    shared.ignored.iter().any(|p| path.starts_with(p))
+        || path.components().any(|c| {
+            matches!(
+                c.as_os_str().to_str(),
+                Some(".git" | "node_modules" | "target")
+            )
+        })
+}
+fn hint(shared: &Shared, event: notify::Result<Event>) {
+    let mut s = match shared.state.lock() {
+        Ok(s) => s,
+        Err(_) => return,
+    };
+    let now = Instant::now();
+    match event {
+        Ok(e)
+            if !e.need_rescan()
+                && matches!(
+                    e.kind,
+                    EventKind::Access(_)
+                        | EventKind::Modify(notify::event::ModifyKind::Metadata(
+                            notify::event::MetadataKind::AccessTime
+                        ))
+                ) => {}
+        Ok(e) if !e.need_rescan() && !e.paths.is_empty() => {
+            // Events can contain unbounded path vectors; consume a bounded prefix
+            // and invalidate all roots if its representation overflowed.
+            if e.paths.len() > MAX_PATHS {
+                for r in s.roots.values_mut() {
+                    r.dirty(None, now, true);
+                }
+            } else {
+                for (root, r) in &mut s.roots {
+                    let paths: Vec<_> = e
+                        .paths
+                        .iter()
+                        .filter(|p| p.starts_with(root.as_path()) && !excluded(p, shared))
+                        .cloned()
+                        .collect();
+                    if !paths.is_empty() {
+                        r.dirty(Some(&paths), now, true);
+                    }
+                }
+            }
+        }
+        _ => {
+            for r in s.roots.values_mut() {
+                r.dirty(None, now, true);
+            }
+        }
+    }
+    shared.changed.notify_all();
+}
+fn run(shared: Arc<Shared>, callback: impl Fn(ScanRequest) -> ScanOutcome) {
+    let events = shared.clone();
+    let mut watcher = notify::recommended_watcher(move |event| hint(&events, event)).ok();
+    let mut watched = BTreeSet::new();
+    loop {
+        let roots = {
+            let s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
+            if s.stop {
+                break;
+            }
+            s.roots.keys().cloned().collect::<Vec<_>>()
+        };
+        for removed in watched
+            .iter()
+            .filter(|p| !roots.contains(p))
+            .cloned()
+            .collect::<Vec<_>>()
+        {
+            if let Some(w) = &mut watcher {
+                let _ = w.unwatch(&removed);
+            }
+            watched.remove(&removed);
+        }
+        let now = Instant::now();
+        for root in &roots {
+            if watched.contains(root) && root.is_dir() {
+                continue;
+            }
+            if watched.remove(root) {
+                if let Some(r) = shared
+                    .state
+                    .lock()
+                    .unwrap_or_else(|e| e.into_inner())
+                    .roots
+                    .get_mut(root)
+                {
+                    r.attached = false;
+                }
+                if let Some(w) = &mut watcher {
+                    let _ = w.unwatch(root);
+                }
+            }
+            let due = shared
+                .state
+                .lock()
+                .unwrap_or_else(|e| e.into_inner())
+                .roots
+                .get(root)
+                .is_some_and(|r| now >= r.attach_retry);
+            if !due {
+                continue;
+            }
+            if watcher.is_none() {
+                let events = shared.clone();
+                watcher = notify::recommended_watcher(move |e| hint(&events, e)).ok();
+            }
+            let result = match &mut watcher {
+                Some(w) => w
+                    .watch(root, RecursiveMode::Recursive)
+                    .map_err(|e| e.to_string()),
+                None => Err(
+                    "Native watcher is unavailable; periodic reconciliation remains enabled."
+                        .into(),
+                ),
+            };
+            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
+            if let Some(r) = s.roots.get_mut(root) {
+                r.attached = result.is_ok();
+                r.attach_retry = if result.is_ok() {
+                    now
+                } else {
+                    now + shared.timing.retry
+                };
+                if result.is_ok() {
+                    r.watch_error = None;
+                    r.dirty(None, now, r.ever_attached);
+                    r.ever_attached = true;
+                    watched.insert(root.clone());
+                } else {
+                    r.watch_error = result.err().map(|e| e.chars().take(512).collect());
+                }
+            }
+        }
+        let request = {
+            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
+            if s.stop {
+                break;
+            }
+            let now = Instant::now();
+            for r in s.roots.values_mut() {
+                if now >= r.next_scan {
+                    r.dirty(None, now, false);
+                    r.next_scan = now + shared.timing.periodic;
+                }
+            }
+            s.roots.iter_mut().find_map(|(root, r)| {
+                let due = r.first.is_some_and(|first| {
+                    now >= r.retry_at
+                        && (now.duration_since(r.last) >= shared.timing.debounce
+                            || now.duration_since(first) >= shared.timing.max_delay)
+                });
+                if !due {
+                    return None;
+                }
+                let paths = if r.full {
+                    None
+                } else {
+                    Some(std::mem::take(&mut r.paths).into_iter().collect())
+                };
+                r.full = false;
+                let media_hint = std::mem::take(&mut r.media_hint);
+                r.first = None;
+                Some(ScanRequest {
+                    root: root.clone(),
+                    paths,
+                    cancelled: shared.cancelled.clone(),
+                    media_hint,
+                })
+            })
+        };
+        if let Some(request) = request {
+            let root = request.root.clone();
+            let paths = request.paths.clone();
+            let media_hint = request.media_hint;
+            let result =
+                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| callback(request)))
+                    .unwrap_or_else(|_| {
+                        ScanOutcome::Failed(
+                            "Folder reconciliation failed unexpectedly; retrying.".into(),
+                        )
+                    });
+            let mut s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
+            if let Some(r) = s.roots.get_mut(&root) {
+                let now = Instant::now();
+                match result {
+                    ScanOutcome::Completed => {
+                        r.failures = 0;
+                        r.error = None;
+                        r.retry_at = now;
+                    }
+                    ScanOutcome::Busy => {
+                        r.dirty(paths.as_deref(), now, media_hint);
+                        r.retry_at = now + Duration::from_millis(250);
+                    }
+                    ScanOutcome::Failed(e) => {
+                        r.dirty(paths.as_deref(), now, media_hint);
+                        r.failures = r.failures.saturating_add(1);
+                        r.error = Some(e.chars().take(512).collect());
+                        r.retry_at = now
+                            + shared
+                                .timing
+                                .retry
+                                .saturating_mul(1u32 << r.failures.saturating_sub(1).min(4))
+                                .min(Duration::from_secs(60));
+                    }
+                }
+            }
+        } else {
+            let s = shared.state.lock().unwrap_or_else(|e| e.into_inner());
+            if s.stop {
+                break;
+            }
+            let _ = shared.changed.wait_timeout(s, Duration::from_millis(100));
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::mpsc;
+    fn fast() -> Timing {
+        Timing {
+            debounce: Duration::from_millis(40),
+            max_delay: Duration::from_millis(120),
+            periodic: Duration::from_secs(30),
+            retry: Duration::from_millis(20),
+        }
+    }
+    fn receive(rx: &mpsc::Receiver<ScanRequest>) -> ScanRequest {
+        rx.recv_timeout(Duration::from_secs(10))
+            .expect("watch reconciliation")
+    }
+    #[test]
+    fn native_windows_or_linux_create_update_rename_delete() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                tx.send(r).unwrap();
+                ScanOutcome::Completed
+            },
+            fast(),
+        )
+        .unwrap();
+        assert!(receive(&rx).paths.is_none());
+        let first = root.path().join("first.txt");
+        let second = root.path().join("second.txt");
+        std::fs::write(&first, "one").unwrap();
+        assert!(receive(&rx).paths.unwrap().contains(&first));
+        std::fs::write(&first, "two").unwrap();
+        assert!(receive(&rx).paths.unwrap().contains(&first));
+        std::fs::rename(&first, &second).unwrap();
+        let rename = receive(&rx).paths.unwrap();
+        assert!(rename.contains(&first) || rename.contains(&second));
+        std::fs::remove_file(&second).unwrap();
+        assert!(receive(&rx).paths.unwrap().contains(&second));
+        assert!(service.status()[0].attached);
+    }
+    #[test]
+    fn overflow_rescan_and_events_during_scan_are_retained() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                tx.send(r).unwrap();
+                ScanOutcome::Completed
+            },
+            fast(),
+        )
+        .unwrap();
+        let _ = receive(&rx);
+        let paths = (0..129)
+            .map(|i| root.path().join(format!("{i}.txt")))
+            .collect();
+        hint(
+            &service.shared,
+            Ok(Event::new(EventKind::Any).add_path(root.path().join("a"))),
+        );
+        let mut e = Event::new(EventKind::Any);
+        e.paths = paths;
+        hint(&service.shared, Ok(e));
+        let overflow = receive(&rx);
+        assert!(overflow.paths.is_none());
+        assert!(overflow.media_hint);
+    }
+    #[test]
+    fn events_arriving_during_reconciliation_survive_completion() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let release_rx = Mutex::new(release_rx);
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                tx.send(r).unwrap();
+                release_rx.lock().unwrap().recv().unwrap();
+                ScanOutcome::Completed
+            },
+            fast(),
+        )
+        .unwrap();
+        let _ = receive(&rx);
+        let path = root.path().join("later.txt");
+        hint(
+            &service.shared,
+            Ok(Event::new(EventKind::Any).add_path(path.clone())),
+        );
+        release_tx.send(()).unwrap();
+        assert!(receive(&rx).paths.unwrap().contains(&path));
+        release_tx.send(()).unwrap();
+        drop(service);
+    }
+
+    #[test]
+    fn callback_panic_retries_and_recovers() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let attempts = AtomicBool::new(false);
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                if !attempts.swap(true, Ordering::AcqRel) {
+                    panic!("owned callback failure");
+                }
+                tx.send(r).unwrap();
+                ScanOutcome::Completed
+            },
+            fast(),
+        )
+        .unwrap();
+        assert!(receive(&rx).paths.is_none());
+        service.stop();
+        assert!(!service.status()[0].pending);
+    }
+
+    #[test]
+    fn busy_retry_preserves_native_media_hint() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                if r.media_hint {
+                    tx.send(r).unwrap();
+                    ScanOutcome::Busy
+                } else {
+                    tx.send(r).unwrap();
+                    ScanOutcome::Completed
+                }
+            },
+            fast(),
+        )
+        .unwrap();
+        assert!(!receive(&rx).media_hint);
+        let mut event = Event::new(EventKind::Any);
+        event.attrs.set_flag(notify::event::Flag::Rescan);
+        hint(&service.shared, Ok(event));
+        let first = receive(&rx);
+        assert!(first.paths.is_none());
+        assert!(first.media_hint);
+        assert!(receive(&rx).media_hint);
+        service.stop();
+    }
+
+    #[test]
+    fn periodic_reconciliation_and_shutdown_signal() {
+        let root = tempfile::tempdir().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let mut timing = fast();
+        timing.periodic = Duration::from_millis(150);
+        let service = WatchService::start_timed(
+            vec![root.path().to_owned()],
+            vec![],
+            move |r| {
+                tx.send(r).unwrap();
+                ScanOutcome::Completed
+            },
+            timing,
+        )
+        .unwrap();
+        let initial = receive(&rx);
+        assert!(initial.paths.is_none());
+        assert!(!initial.media_hint);
+        let periodic = receive(&rx);
+        assert!(periodic.paths.is_none());
+        assert!(!periodic.media_hint);
+        drop(service);
+        assert!(initial.cancelled.load(Ordering::Acquire));
+    }
+}