Restore FileSystem.watch recursive control (#6705)

Author: tylergibbs1

.changeset/wise-files-watch.md

@@ -0,0 +1,7 @@
+---
+"@effect/platform-deno": patch
+"@effect/platform-node-shared": patch
+"effect": patch
+---
+
+Restore the `recursive` option for `FileSystem.watch`, with non-recursive watching as the default.

packages/effect/src/FileSystem.ts

@@ -341,9 +341,15 @@ export interface FileSystem {
     mtime: Date | number
   ) => Effect.Effect<void, PlatformError>
   /**
-   * Watch a directory or file for changes
+   * Watch a directory or file for changes.
+   *
+   * **Details**
+   *
+   * By default, only changes to the direct children of the directory are
+   * reported. Set the `recursive` option to `true` to watch for changes in
+   * subdirectories as well.
    */
-  readonly watch: (path: string) => Stream.Stream<WatchEvent, PlatformError>
+  readonly watch: (path: string, options?: WatchOptions) => Stream.Stream<WatchEvent, PlatformError>
   /**
    * Write data to a file at `path`.
    */
@@ -1247,6 +1253,19 @@ export declare namespace File {
  */
 export type SeekMode = "start" | "current"
 
+/**
+ * Options for watching files or directories.
+ *
+ * @category models
+ * @since 4.0.0
+ */
+export interface WatchOptions {
+  /**
+   * When `true`, changes in subdirectories are also reported.
+   */
+  readonly recursive?: boolean | undefined
+}
+
 /**
  * Represents file system events emitted when watching files or directories.
  *
@@ -1363,5 +1382,9 @@ export declare namespace WatchEvent {
  * @since 4.0.0
  */
 export class WatchBackend extends Context.Service<WatchBackend, {
-  readonly register: (path: string, stat: File.Info) => Option.Option<Stream.Stream<WatchEvent, PlatformError>>
+  readonly register: (
+    path: string,
+    stat: File.Info,
+    options?: WatchOptions
+  ) => Option.Option<Stream.Stream<WatchEvent, PlatformError>>
 }>()("effect/platform/FileSystem/WatchBackend") {}

packages/platform-deno/src/DenoFileSystem.ts

@@ -390,11 +390,14 @@ const truncate: FileSystem.FileSystem["truncate"] = (path, length) =>
 const utimes: FileSystem.FileSystem["utimes"] = (path, atime, mtime) =>
   tryPromise("utimes", path, () => Deno.utime(path, atime, mtime))
 
-const watchNative = (path: string): Stream.Stream<FileSystem.WatchEvent, PlatformError.PlatformError> =>
+const watchNative = (
+  path: string,
+  options?: FileSystem.WatchOptions
+): Stream.Stream<FileSystem.WatchEvent, PlatformError.PlatformError> =>
   Stream.unwrap(
     Effect.map(
       Effect.try({
-        try: () => Deno.watchFs(path, { recursive: true }),
+        try: () => Deno.watchFs(path, { recursive: options?.recursive ?? false }),
         catch: handleError("FileSystem", "watch", path)
       }),
       (watcher) =>
@@ -421,12 +424,16 @@ const watchNative = (path: string): Stream.Stream<FileSystem.WatchEvent, Platfor
     )
   )
 
-const watch = (backend: Option.Option<FileSystem.WatchBackend["Service"]>, path: string) =>
+const watch = (
+  backend: Option.Option<FileSystem.WatchBackend["Service"]>,
+  path: string,
+  options?: FileSystem.WatchOptions
+) =>
   stat(path).pipe(
     Effect.map((info) =>
       backend.pipe(
-        Option.flatMap((backend) => backend.register(path, info)),
-        Option.getOrElse(() => watchNative(path))
+        Option.flatMap((backend) => backend.register(path, info, options)),
+        Option.getOrElse(() => watchNative(path, options))
       )
     ),
     Stream.unwrap
@@ -469,8 +476,8 @@ const makeFileSystem = Effect.map(Effect.serviceOption(FileSystem.WatchBackend),
     symlink,
     truncate,
     utimes,
-    watch(path) {
-      return watch(backend, path)
+    watch(path, options) {
+      return watch(backend, path, options)
     },
     writeFile
   }))

packages/platform-node-shared/src/NodeFileSystem.ts

@@ -550,12 +550,12 @@ const utimes = (() => {
 
 // == watch
 
-const watchNode = (path: string) =>
+const watchNode = (path: string, options?: FileSystem.WatchOptions) =>
   Stream.callback<FileSystem.WatchEvent, Error.PlatformError>((queue) =>
     Effect.acquireRelease(
       Effect.sync(() => {
         const watcher = NFS.watch(path, {
-          recursive: true
+          recursive: options?.recursive ?? false
         }, (event, path) => {
           if (!path) return
           switch (event) {
@@ -595,12 +595,16 @@ const watchNode = (path: string) =>
     )
   )
 
-const watch = (backend: Option.Option<FileSystem.WatchBackend["Service"]>, path: string) =>
+const watch = (
+  backend: Option.Option<FileSystem.WatchBackend["Service"]>,
+  path: string,
+  options?: FileSystem.WatchOptions
+) =>
   stat(path).pipe(
     Effect.map((stat) =>
       backend.pipe(
-        Option.flatMap((_) => _.register(path, stat)),
-        Option.getOrElse(() => watchNode(path))
+        Option.flatMap((_) => _.register(path, stat, options)),
+        Option.getOrElse(() => watchNode(path, options))
       )
     ),
     Stream.unwrap
@@ -652,8 +656,8 @@ const makeFileSystem = Effect.map(Effect.serviceOption(FileSystem.WatchBackend),
     symlink,
     truncate,
     utimes,
-    watch(path) {
-      return watch(backend, path)
+    watch(path, options) {
+      return watch(backend, path, options)
     },
     writeFile
   }))

packages/platform-node-shared/test/NodeFileSystem.test.ts

@@ -1,5 +1,101 @@
 import * as NodeFileSystem from "@effect/platform-node-shared/NodeFileSystem"
-import { describe } from "@effect/vitest"
+import { assert, describe, it } from "@effect/vitest"
+import * as Deferred from "effect/Deferred"
+import * as Effect from "effect/Effect"
+import * as Fiber from "effect/Fiber"
+import * as FileSystem from "effect/FileSystem"
+import * as Stream from "effect/Stream"
+import * as TestClock from "effect/testing/TestClock"
 import { testLayer } from "../../effect/test/FileSystem.test-utils.ts"
 
-describe("FileSystem", () => testLayer(NodeFileSystem.layer))
+const startWatch = <E, R>(
+  fs: FileSystem.FileSystem,
+  root: string,
+  watch: () => Stream.Stream<FileSystem.WatchEvent, E, R>
+) =>
+  Effect.gen(function*() {
+    const ready = yield* Deferred.make<void>()
+    const readyName = ".watch-ready"
+    const fiber = yield* watch().pipe(
+      Stream.tap((event) =>
+        event.path === readyName
+          ? Deferred.succeed(ready, undefined)
+          : Effect.void
+      ),
+      Stream.filter((event) => event.path !== readyName),
+      Stream.runHead,
+      Effect.flatMap(Effect.fromOption),
+      Effect.forkChild
+    )
+    const signalFiber = yield* Effect.sleep("10 millis").pipe(
+      TestClock.withLive,
+      Effect.andThen(fs.writeFileString(`${root}/${readyName}`, "")),
+      Effect.forever,
+      Effect.forkChild
+    )
+    yield* Deferred.await(ready).pipe(
+      Effect.raceFirst(Fiber.join(fiber).pipe(Effect.asVoid)),
+      Effect.ensuring(Fiber.interrupt(signalFiber))
+    )
+    return fiber
+  })
+
+describe("FileSystem", () => {
+  testLayer(NodeFileSystem.layer)
+
+  it.effect("watch does not report nested changes when recursive is false", () =>
+    Effect.gen(function*() {
+      const fs = yield* FileSystem.FileSystem
+      const root = yield* fs.makeTempDirectoryScoped()
+      const nested = `${root}/nested`
+      yield* fs.makeDirectory(nested)
+
+      const fiber = yield* startWatch(fs, root, () => fs.watch(root, { recursive: false }))
+
+      yield* fs.writeFileString(`${nested}/nested.txt`, "")
+      yield* fs.writeFileString(`${root}/direct.txt`, "")
+
+      const event = yield* Fiber.join(fiber)
+      assert.strictEqual(event.path, "direct.txt")
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(NodeFileSystem.layer)
+    ))
+
+  it.effect("watch is non-recursive when options are omitted", () =>
+    Effect.gen(function*() {
+      const fs = yield* FileSystem.FileSystem
+      const root = yield* fs.makeTempDirectoryScoped()
+      const nested = `${root}/nested`
+      yield* fs.makeDirectory(nested)
+
+      const fiber = yield* startWatch(fs, root, () => fs.watch(root))
+
+      yield* fs.writeFileString(`${nested}/nested.txt`, "")
+      yield* fs.writeFileString(`${root}/direct.txt`, "")
+
+      const event = yield* Fiber.join(fiber)
+      assert.strictEqual(event.path, "direct.txt")
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(NodeFileSystem.layer)
+    ))
+
+  it.effect("watch reports nested changes when recursive is true", () =>
+    Effect.gen(function*() {
+      const fs = yield* FileSystem.FileSystem
+      const root = yield* fs.makeTempDirectoryScoped()
+      const nested = `${root}/nested`
+      yield* fs.makeDirectory(nested)
+
+      const fiber = yield* startWatch(fs, root, () => fs.watch(root, { recursive: true }))
+
+      yield* fs.writeFileString(`${nested}/nested.txt`, "")
+
+      const event = yield* Fiber.join(fiber)
+      assert(event.path.endsWith("nested.txt"))
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(NodeFileSystem.layer)
+    ))
+})