fix(node/fs/promises): watch should be async iterable (#24805)

The way `fs.watch` works is different in `node:fs/promises` than
`node:fs`. It has a different function signature and it returns an async
iterable instead, see
https://nodejs.org/api/fs.html#fspromiseswatchfilename-options

Fixes https://github.com/denoland/deno/issues/24661

ext/node/polyfills/_fs/_fs_watch.ts

@@ -155,22 +155,43 @@ export function watch(
   return fsWatcher;
 }
 
-export const watchPromise = promisify(watch) as (
-  & ((
-    filename: string | URL,
-    options: watchOptions,
-    listener: watchListener,
-  ) => Promise<FSWatcher>)
-  & ((
-    filename: string | URL,
-    listener: watchListener,
-  ) => Promise<FSWatcher>)
-  & ((
-    filename: string | URL,
-    options: watchOptions,
-  ) => Promise<FSWatcher>)
-  & ((filename: string | URL) => Promise<FSWatcher>)
-);
+export function watchPromise(
+  filename: string | Buffer | URL,
+  options?: {
+    persistent?: boolean;
+    recursive?: boolean;
+    encoding?: string;
+    signal?: AbortSignal;
+  },
+): AsyncIterable<{ eventType: string; filename: string | Buffer | null }> {
+  const watchPath = getValidatedPath(filename).toString();
+
+  const watcher = Deno.watchFs(watchPath, {
+    recursive: options?.recursive ?? false,
+  });
+
+  if (options?.signal) {
+    options?.signal.addEventListener("abort", () => watcher.close());
+  }
+
+  const fsIterable = watcher[Symbol.asyncIterator]();
+  const iterable = {
+    async next() {
+      const result = await fsIterable.next();
+      if (result.done) return result;
+
+      const eventType = convertDenoFsEventToNodeFsEvent(result.value.kind);
+      return {
+        value: { eventType, filename: basename(result.value.paths[0]) },
+        done: result.done,
+      };
+    },
+  };
+
+  return {
+    [Symbol.asyncIterator]: () => iterable,
+  };
+}
 
 type WatchFileListener = (curr: Stats, prev: Stats) => void;
 type WatchFileOptions = {

tests/unit_node/_fs/_fs_watch_test.ts

@@ -1,6 +1,7 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 import { unwatchFile, watch, watchFile } from "node:fs";
-import { assertEquals } from "@std/assert";
+import { watch as watchPromise } from "node:fs/promises";
+import { assert, assertEquals } from "@std/assert";
 
 function wait(time: number) {
   return new Promise((resolve) => {
@@ -52,3 +53,46 @@ Deno.test({
     watcher.unref();
   },
 });
+
+Deno.test({
+  name: "node [fs/promises] watch should return async iterable",
+  sanitizeOps: false,
+  sanitizeResources: false,
+  async fn() {
+    const file = Deno.makeTempFileSync();
+    Deno.writeTextFileSync(file, "foo");
+
+    const result: { eventType: string; filename: string | null }[] = [];
+
+    const controller = new AbortController();
+    const watcher = watchPromise(file, {
+      // Node types resolved by the LSP clash with ours
+      // deno-lint-ignore no-explicit-any
+      signal: controller.signal as any,
+    });
+
+    const deferred = Promise.withResolvers<void>();
+    let stopLength = 0;
+    setTimeout(async () => {
+      Deno.writeTextFileSync(file, "something");
+      controller.abort();
+      stopLength = result.length;
+      await wait(100);
+      Deno.writeTextFileSync(file, "something else");
+      await wait(100);
+      deferred.resolve();
+    }, 100);
+
+    for await (const event of watcher) {
+      result.push(event);
+    }
+    await deferred.promise;
+
+    assertEquals(result.length, stopLength);
+    assert(
+      result.every((item) =>
+        typeof item.eventType === "string" && typeof item.filename === "string"
+      ),
+    );
+  },
+});