origin_workspace_watch/
lib.rs

1//! The workspace watcher contract (`WorkspaceWatcher`) over `notify` (B4).
2//!
3//! Event-driven local sync: a product like Gitbit does not have to poll a working
4//! tree. `notify` runs the platform mechanism (FSEvents, inotify, ReadDirectoryChanges)
5//! on its own thread and calls back; the callback converts each event to a
6//! [`WorkspaceChange`] and pushes it onto a broadcast channel, which the returned
7//! [`WatchHandle`] reads. Nothing here touches Tauri.
8
9use async_trait::async_trait;
10use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
11use origin_domain::{AppError, Result};
12use origin_platform::{WatchHandle, WorkspaceChange, WorkspaceRoot, WorkspaceWatcher};
13use std::collections::HashMap;
14use std::path::Path;
15use std::sync::Mutex;
16use tokio::sync::broadcast;
17
18/// How many change events to buffer while a subscriber catches up.
19const BUFFER: usize = 64;
20
21/// Watches workspace roots with `notify`.
22///
23/// One `notify` watcher is kept per root for the lifetime of the adapter — dropping it
24/// stops the platform watch. Calling `watch` twice for the same root reuses the
25/// existing watcher and returns a fresh handle to the same channel.
26#[derive(Debug, Default)]
27pub struct NotifyWorkspaceWatcher {
28    watchers: Mutex<HashMap<WorkspaceRoot, RecommendedWatcher>>,
29}
30
31impl NotifyWorkspaceWatcher {
32    pub fn new() -> Self {
33        Self::default()
34    }
35}
36
37#[async_trait]
38impl WorkspaceWatcher for NotifyWorkspaceWatcher {
39    async fn watch(&self, root: &WorkspaceRoot) -> Result<WatchHandle> {
40        let (sender, _) = broadcast::channel(BUFFER);
41
42        {
43            let watchers = self.watchers.lock().expect("watcher map poisoned");
44            if watchers.contains_key(root) {
45                return Ok(WatchHandle::new(sender.subscribe()));
46            }
47        }
48
49        let root_path = root.as_path().to_path_buf();
50        let forward = sender.clone();
51        let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
52            match event {
53                Ok(event) => {
54                    if let Some(change) = change_for(&event, &root_path) {
55                        // `send` fails only when nobody is subscribed, which is fine.
56                        let _ = forward.send(change);
57                    }
58                }
59                Err(error) => tracing::warn!(%error, "workspace watcher reported an error"),
60            }
61        })
62        .map_err(|error| {
63            AppError::internal(format!("cannot create a filesystem watcher: {error}"))
64        })?;
65
66        watcher
67            .watch(root.as_path(), RecursiveMode::Recursive)
68            .map_err(|error| {
69                AppError::storage(format!(
70                    "cannot watch {}: {error}",
71                    root.as_path().display()
72                ))
73            })?;
74
75        self.watchers
76            .lock()
77            .expect("watcher map poisoned")
78            .insert(root.clone(), watcher);
79
80        tracing::debug!(root = %root.as_path().display(), "workspace watch started");
81        Ok(WatchHandle::new(sender.subscribe()))
82    }
83}
84
85/// Translate a `notify` event into a workspace change, relative to the root.
86fn change_for(event: &Event, root: &Path) -> Option<WorkspaceChange> {
87    let path = event.paths.first()?;
88    let relative = path.strip_prefix(root).unwrap_or(path);
89    // A POSIX-style relative path, so a change reads the same on every platform.
90    let relative = relative.to_string_lossy().replace('\\', "/");
91
92    match event.kind {
93        EventKind::Create(_) => Some(WorkspaceChange::created(relative)),
94        EventKind::Modify(_) => Some(WorkspaceChange::modified(relative)),
95        EventKind::Remove(_) => Some(WorkspaceChange::removed(relative)),
96        // Access events and metadata-only kinds are noise for a working-tree sync.
97        _ => None,
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use std::time::Duration;
105
106    #[test]
107    fn only_write_kinds_map_to_a_change() {
108        let root = Path::new("/repo");
109
110        let create = Event {
111            kind: EventKind::Create(notify::event::CreateKind::File),
112            paths: vec![std::path::PathBuf::from("/repo/src/new.rs")],
113            attrs: Default::default(),
114        };
115        assert_eq!(
116            change_for(&create, root),
117            Some(WorkspaceChange::created("src/new.rs"))
118        );
119
120        let access = Event {
121            kind: EventKind::Access(notify::event::AccessKind::Read),
122            paths: vec![std::path::PathBuf::from("/repo/src/main.rs")],
123            attrs: Default::default(),
124        };
125        assert_eq!(change_for(&access, root), None);
126    }
127
128    #[tokio::test]
129    async fn a_file_created_inside_the_root_is_reported() {
130        let dir =
131            std::env::temp_dir().join(format!("origin-workspace-watch-{}", std::process::id()));
132        let _ = std::fs::remove_dir_all(&dir);
133        std::fs::create_dir_all(&dir).unwrap();
134
135        let root = WorkspaceRoot::new(dir.clone()).expect("absolute root");
136        let watcher = NotifyWorkspaceWatcher::new();
137        let mut handle = watcher.watch(&root).await.expect("watch");
138
139        // Let the platform watcher register before the change is made.
140        tokio::time::sleep(Duration::from_millis(300)).await;
141        std::fs::write(dir.join("new.txt"), b"hello").unwrap();
142
143        let change = tokio::time::timeout(Duration::from_secs(10), handle.recv())
144            .await
145            .expect("a change must arrive within the timeout")
146            .expect("the channel is open");
147
148        assert!(
149            change.path().contains("new.txt"),
150            "unexpected change: {change:?}"
151        );
152
153        std::fs::remove_dir_all(&dir).ok();
154    }
155}
origin_workspace_watch/lib.rs

origin_workspace_watch/
lib.rs
