1 files changed, 91 insertions, 30 deletions

diff --git a/runtime/ops/fs_events.rs b/runtime/ops/fs_events.rs
index d88a32d91..c8e0228bc 100644
--- a/runtime/ops/fs_events.rs
+++ b/runtime/ops/fs_events.rs
@@ -1,6 +1,5 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
-use deno_core::error::AnyError;
 use deno_core::parking_lot::Mutex;
 use deno_core::AsyncRefCell;
 use deno_core::CancelFuture;
@@ -20,13 +19,14 @@ use notify::EventKind;
 use notify::RecommendedWatcher;
 use notify::RecursiveMode;
 use notify::Watcher;
-use serde::Deserialize;
 use serde::Serialize;
 use std::borrow::Cow;
 use std::cell::RefCell;
 use std::convert::From;
+use std::path::Path;
 use std::path::PathBuf;
 use std::rc::Rc;
+use std::sync::Arc;
 use tokio::sync::mpsc;
 
 deno_core::extension!(
@@ -35,9 +35,7 @@ deno_core::extension!(
 );
 
 struct FsEventsResource {
-  #[allow(unused)]
-  watcher: RecommendedWatcher,
-  receiver: AsyncRefCell<mpsc::Receiver<Result<FsEvent, AnyError>>>,
+  receiver: AsyncRefCell<mpsc::Receiver<Result<FsEvent, NotifyError>>>,
   cancel: CancelHandle,
 }
 
@@ -59,7 +57,7 @@ impl Resource for FsEventsResource {
 ///
 /// Feel free to expand this struct as long as you can add tests to demonstrate
 /// the complexity.
-#[derive(Serialize, Debug)]
+#[derive(Serialize, Debug, Clone)]
 struct FsEvent {
   kind: &'static str,
   paths: Vec<PathBuf>,
@@ -93,43 +91,102 @@ impl From<NotifyEvent> for FsEvent {
   }
 }
 
-#[derive(Deserialize)]
-pub struct OpenArgs {
-  recursive: bool,
-  paths: Vec<String>,
+type WatchSender = (Vec<String>, mpsc::Sender<Result<FsEvent, NotifyError>>);
+
+struct WatcherState {
+  senders: Arc<Mutex<Vec<WatchSender>>>,
+  watcher: RecommendedWatcher,
 }
 
-#[op2]
-#[smi]
-fn op_fs_events_open(
+fn starts_with_canonicalized(path: &Path, prefix: &str) -> bool {
+  #[allow(clippy::disallowed_methods)]
+  let path = path.canonicalize().ok();
+  #[allow(clippy::disallowed_methods)]
+  let prefix = std::fs::canonicalize(prefix).ok();
+  match (path, prefix) {
+    (Some(path), Some(prefix)) => path.starts_with(prefix),
+    _ => false,
+  }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum FsEventsError {
+  #[error(transparent)]
+  Resource(deno_core::error::AnyError),
+  #[error(transparent)]
+  Permission(#[from] deno_permissions::PermissionCheckError),
+  #[error(transparent)]
+  Notify(#[from] NotifyError),
+  #[error(transparent)]
+  Canceled(#[from] deno_core::Canceled),
+}
+
+fn start_watcher(
   state: &mut OpState,
-  #[serde] args: OpenArgs,
-) -> Result<ResourceId, AnyError> {
-  let (sender, receiver) = mpsc::channel::<Result<FsEvent, AnyError>>(16);
-  let sender = Mutex::new(sender);
-  let mut watcher: RecommendedWatcher = Watcher::new(
+  paths: Vec<String>,
+  sender: mpsc::Sender<Result<FsEvent, NotifyError>>,
+) -> Result<(), FsEventsError> {
+  if let Some(watcher) = state.try_borrow_mut::<WatcherState>() {
+    watcher.senders.lock().push((paths, sender));
+    return Ok(());
+  }
+
+  let senders = Arc::new(Mutex::new(vec![(paths, sender)]));
+
+  let sender_clone = senders.clone();
+  let watcher: RecommendedWatcher = Watcher::new(
     move |res: Result<NotifyEvent, NotifyError>| {
-      let res2 = res.map(FsEvent::from).map_err(AnyError::from);
-      let sender = sender.lock();
-      // Ignore result, if send failed it means that watcher was already closed,
-      // but not all messages have been flushed.
-      let _ = sender.try_send(res2);
+      let res2 = res.map(FsEvent::from).map_err(FsEventsError::Notify);
+      for (paths, sender) in sender_clone.lock().iter() {
+        // Ignore result, if send failed it means that watcher was already closed,
+        // but not all messages have been flushed.
+
+        // Only send the event if the path matches one of the paths that the user is watching
+        if let Ok(event) = &res2 {
+          if paths.iter().any(|path| {
+            event.paths.iter().any(|event_path| {
+              same_file::is_same_file(event_path, path).unwrap_or(false)
+                || starts_with_canonicalized(event_path, path)
+            })
+          }) {
+            let _ = sender.try_send(Ok(event.clone()));
+          }
+        }
+      }
     },
     Default::default(),
   )?;
-  let recursive_mode = if args.recursive {
+
+  state.put::<WatcherState>(WatcherState { watcher, senders });
+
+  Ok(())
+}
+
+#[op2]
+#[smi]
+fn op_fs_events_open(
+  state: &mut OpState,
+  recursive: bool,
+  #[serde] paths: Vec<String>,
+) -> Result<ResourceId, FsEventsError> {
+  let (sender, receiver) = mpsc::channel::<Result<FsEvent, NotifyError>>(16);
+
+  start_watcher(state, paths.clone(), sender)?;
+
+  let recursive_mode = if recursive {
     RecursiveMode::Recursive
   } else {
     RecursiveMode::NonRecursive
   };
-  for path in &args.paths {
+  for path in &paths {
     let path = state
       .borrow_mut::<PermissionsContainer>()
       .check_read(path, "Deno.watchFs()")?;
-    watcher.watch(&path, recursive_mode)?;
+
+    let watcher = state.borrow_mut::<WatcherState>();
+    watcher.watcher.watch(&path, recursive_mode)?;
   }
   let resource = FsEventsResource {
-    watcher,
     receiver: AsyncRefCell::new(receiver),
     cancel: Default::default(),
   };
@@ -142,14 +199,18 @@ fn op_fs_events_open(
 async fn op_fs_events_poll(
   state: Rc<RefCell<OpState>>,
   #[smi] rid: ResourceId,
-) -> Result<Option<FsEvent>, AnyError> {
-  let resource = state.borrow().resource_table.get::<FsEventsResource>(rid)?;
+) -> Result<Option<FsEvent>, FsEventsError> {
+  let resource = state
+    .borrow()
+    .resource_table
+    .get::<FsEventsResource>(rid)
+    .map_err(FsEventsError::Resource)?;
   let mut receiver = RcRef::map(&resource, |r| &r.receiver).borrow_mut().await;
   let cancel = RcRef::map(resource, |r| &r.cancel);
   let maybe_result = receiver.recv().or_cancel(cancel).await?;
   match maybe_result {
     Some(Ok(value)) => Ok(Some(value)),
-    Some(Err(err)) => Err(err),
+    Some(Err(err)) => Err(FsEventsError::Notify(err)),
     None => Ok(None),
   }
 }