Implement `fs.watchFile()`fs-watch-file

Closes #3812

2 files changed, 264 insertions, 0 deletions

diff --git a/src/js/node/fs.js b/src/js/node/fs.js
index 5e72d6e27..37f109a33 100644
--- a/src/js/node/fs.js
+++ b/src/js/node/fs.js
@@ -8,6 +8,7 @@ var { direct, isPromise, isCallable } = $lazy("primordials");
 import promises from "node:fs/promises";
 export { default as promises } from "node:fs/promises";
 import * as Stream from "node:stream";
+import { resolve } from "node:path";
 
 var fs = Bun.fs();
 var debug = process.env.DEBUG ? console.log : () => {};
@@ -68,6 +69,124 @@ class FSWatcher extends EventEmitter {
     this.#watcher?.unref();
   }
 }
+
+/** @type {Map<string, Array<[Function, StatWatcher]>>} */
+const statWatchers = new Map();
+
+/** @link https://nodejs.org/api/fs.html#class-fsstatwatcher */
+class StatWatcher extends EventEmitter {
+  #filename;
+  #options;
+  #listener;
+  #watcher;
+  #timer;
+  #stat;
+
+  constructor(filename, options, listener) {
+    super();
+    this.#filename = filename;
+    if (typeof options === "function") {
+      listener = options;
+      options = undefined;
+    } else if (typeof listener !== "function") {
+      listener = () => {};
+    }
+    this.#listener = listener;
+    this.#options = options;
+    const watchKey = resolve(filename);
+    const watchers = statWatchers.get(watchKey);
+    if (watchers === undefined) {
+      statWatchers.set(watchKey, [[this.#listener, this]]);
+    } else {
+      watchers.push([this.#listener, this]);
+    }
+    this.#watch();
+  }
+
+  #watch() {
+    let previous = this.#stat;
+    let current;
+    try {
+      current = this.#stat = fs.statSync(this.#filename);
+      debug("fs.watchFile mtime", current.mtime);
+
+      if (this.#watcher === undefined) {
+        this.#watcher = fs.watch(this.#filename, this.#options, this.#onEvent.bind(this));
+      }
+    } catch (error) {
+      debug("fs.watchFile error", error);
+      if (error.code !== "ENOENT") {
+        throw error;
+      }
+
+      // When an `fs.watchFile` operation results in an ENOENT error,
+      // it will invoke the listener once, with all the fields zeroed (or, for dates, the Unix Epoch).
+      // If the file is created later on, the listener will be called again, with the latest stat objects.
+      if (previous === undefined) {
+        current = this.#stat = new fs.Stats(this.#options?.bigint === true);
+        this.#listener?.(current, current);
+      }
+
+      if (this.#timer === undefined) {
+        this.#timer = setInterval(
+          this.#watch.bind(this),
+          this.#options?.interval ?? 5007, // libuv default
+        );
+      }
+      return;
+    }
+    if (previous !== undefined && previous.mtimeMs !== current.mtimeMs) {
+      this.#listener?.(current, previous);
+    }
+    this.#clear();
+  }
+
+  #onEvent(eventType, filename) {
+    debug("fs.watchFile event", eventType, filename);
+    switch (eventType) {
+      case "close":
+        this.close();
+        break;
+      case "error":
+        this.close();
+      // fallthrough
+      case "rename":
+      case "change":
+        this.#watch();
+        break;
+    }
+  }
+
+  #clear() {
+    if (this.#timer !== undefined) {
+      debug("fs.watchFile clear timer");
+      clearInterval(this.#timer);
+      this.#timer = undefined;
+    }
+  }
+
+  close() {
+    debug("fs.watchFile close");
+    this.#watcher?.close();
+    this.#watcher = undefined;
+    this.#clear();
+  }
+
+  ref() {
+    debug("fs.watchFile ref");
+    this.#watcher?.ref();
+    this.#timer?.ref();
+    return this;
+  }
+
+  unref() {
+    debug("fs.watchFile unref");
+    this.#watcher?.unref();
+    this.#timer?.unref();
+    return this;
+  }
+}
+
 export var access = function access(...args) {
     callbackify(fs.accessSync, args);
   },
@@ -250,6 +369,43 @@ export var access = function access(...args) {
   Stats = fs.Stats,
   watch = function watch(path, options, listener) {
     return new FSWatcher(path, options, listener);
+  },
+  watchFile = function watchFile(path, options, listener) {
+    return new StatWatcher(path, options, listener);
+  },
+  unwatchFile = function unwatchFile(path, listener) {
+    const watchKey = resolve(path);
+    const watchers = statWatchers.get(watchKey);
+    if (watchers === undefined) {
+      return;
+    }
+    if (typeof listener === "function") {
+      const deleted = new Set();
+      for (const [func, watcher] of watchers) {
+        if (listener !== func) {
+          continue;
+        }
+        try {
+          watcher.close();
+        } finally {
+          deleted.add(watcher);
+        }
+      }
+      const remaining = watchers.filter(([_, watcher]) => !deleted.has(watcher));
+      if (remaining.length) {
+        statWatchers.set(watchKey, remaining);
+      } else {
+        statWatchers.delete(watchKey);
+      }
+      return;
+    }
+    try {
+      for (const [_, watcher] of watchers) {
+        watcher.close();
+      }
+    } finally {
+      statWatchers.delete(watchKey);
+    }
   };
 
 function callbackify(fsFunction, args) {
@@ -1102,6 +1258,9 @@ export default {
   ReadStream,
   watch,
   FSWatcher,
+  watchFile,
+  unwatchFile,
+  StatWatcher,
   writev,
   writevSync,
   readv,
diff --git a/src/js/out/modules/node/fs.js b/src/js/out/modules/node/fs.js
index b7457f104..5c67f3e0c 100644
--- a/src/js/out/modules/node/fs.js
+++ b/src/js/out/modules/node/fs.js
@@ -2,6 +2,7 @@ import {EventEmitter} from "node:events";
 import promises2 from "node:fs/promises";
 import {default as default2} from "node:fs/promises";
 import * as Stream from "node:stream";
+import {resolve} from "node:path";
 var callbackify = function(fsFunction, args) {
   try {
     const result = fsFunction.apply(fs, args.slice(0, args.length - 1)), callback = args[args.length - 1];
@@ -61,6 +62,75 @@ class FSWatcher extends EventEmitter {
     this.#watcher?.unref();
   }
 }
+var statWatchers = new Map;
+
+class StatWatcher extends EventEmitter {
+  #filename;
+  #options;
+  #listener;
+  #watcher;
+  #timer;
+  #stat;
+  constructor(filename, options, listener) {
+    super();
+    if (this.#filename = filename, typeof options === "function")
+      listener = options, options = void 0;
+    else if (typeof listener !== "function")
+      listener = () => {
+      };
+    this.#listener = listener, this.#options = options;
+    const watchKey = resolve(filename), watchers = statWatchers.get(watchKey);
+    if (watchers === void 0)
+      statWatchers.set(watchKey, [[this.#listener, this]]);
+    else
+      watchers.push([this.#listener, this]);
+    this.#watch();
+  }
+  #watch() {
+    let previous = this.#stat, current;
+    try {
+      if (current = this.#stat = fs.statSync(this.#filename), debug("fs.watchFile mtime", current.mtime), this.#watcher === void 0)
+        this.#watcher = fs.watch(this.#filename, this.#options, this.#onEvent.bind(this));
+    } catch (error) {
+      if (debug("fs.watchFile error", error), error.code !== "ENOENT")
+        throw error;
+      if (previous === void 0)
+        current = this.#stat = new fs.Stats(this.#options?.bigint === !0), this.#listener?.(current, current);
+      if (this.#timer === void 0)
+        this.#timer = setInterval(this.#watch.bind(this), this.#options?.interval ?? 5007);
+      return;
+    }
+    if (previous !== void 0 && previous.mtimeMs !== current.mtimeMs)
+      this.#listener?.(current, previous);
+    this.#clear();
+  }
+  #onEvent(eventType, filename) {
+    switch (debug("fs.watchFile event", eventType, filename), eventType) {
+      case "close":
+        this.close();
+        break;
+      case "error":
+        this.close();
+      case "rename":
+      case "change":
+        this.#watch();
+        break;
+    }
+  }
+  #clear() {
+    if (this.#timer !== void 0)
+      debug("fs.watchFile clear timer"), clearInterval(this.#timer), this.#timer = void 0;
+  }
+  close() {
+    debug("fs.watchFile close"), this.#watcher?.close(), this.#watcher = void 0, this.#clear();
+  }
+  ref() {
+    return debug("fs.watchFile ref"), this.#watcher?.ref(), this.#timer?.ref(), this;
+  }
+  unref() {
+    return debug("fs.watchFile unref"), this.#watcher?.unref(), this.#timer?.unref(), this;
+  }
+}
 var access = function access2(...args) {
   callbackify(fs.accessSync, args);
 }, appendFile = function appendFile2(...args) {
@@ -157,6 +227,36 @@ var access = function access2(...args) {
   });
 }, readvSync = fs.readvSync.bind(fs), Dirent = fs.Dirent, Stats = fs.Stats, watch = function watch2(path, options, listener) {
   return new FSWatcher(path, options, listener);
+}, watchFile = function watchFile2(path, options, listener) {
+  return new StatWatcher(path, options, listener);
+}, unwatchFile = function unwatchFile2(path, listener) {
+  const watchKey = resolve(path), watchers = statWatchers.get(watchKey);
+  if (watchers === void 0)
+    return;
+  if (typeof listener === "function") {
+    const deleted = new Set;
+    for (let [func, watcher] of watchers) {
+      if (listener !== func)
+        continue;
+      try {
+        watcher.close();
+      } finally {
+        deleted.add(watcher);
+      }
+    }
+    const remaining = watchers.filter(([_, watcher]) => !deleted.has(watcher));
+    if (remaining.length)
+      statWatchers.set(watchKey, remaining);
+    else
+      statWatchers.delete(watchKey);
+    return;
+  }
+  try {
+    for (let [_, watcher] of watchers)
+      watcher.close();
+  } finally {
+    statWatchers.delete(watchKey);
+  }
 }, readStreamPathFastPathSymbol = Symbol.for("Bun.Node.readStreamPathFastPath"), readStreamSymbol = Symbol.for("Bun.NodeReadStream"), readStreamPathOrFdSymbol = Symbol.for("Bun.NodeReadStreamPathOrFd"), writeStreamSymbol = Symbol.for("Bun.NodeWriteStream"), writeStreamPathFastPathSymbol = Symbol.for("Bun.NodeWriteStreamFastPath"), writeStreamPathFastPathCallSymbol = Symbol.for("Bun.NodeWriteStreamFastPathCall"), kIoDone = Symbol.for("kIoDone"), defaultReadStreamOptions = {
   file: void 0,
   fd: void 0,
@@ -664,6 +764,9 @@ var fs_default = {
   ReadStream,
   watch,
   FSWatcher,
+  watchFile,
+  unwatchFile,
+  StatWatcher,
   writev,
   writevSync,
   readv,
@@ -680,9 +783,11 @@ export {
   writeFileSync,
   writeFile,
   write,
+  watchFile,
   watch,
   utimesSync,
   utimes,
+  unwatchFile,
   unlinkSync,
   unlink,
   truncateSync,