Merge pull request #1267 from streamich/node-to-fsa-watcher-bridge

Node-to-FSA watcher bridge

Author: streamich

packages/fs-fsa-to-node/README.md

@@ -32,6 +32,34 @@ console.log(content); // 'Hello, World!'
 - Supports file operations: read, write, append, truncate
 - Supports directory operations: mkdir, readdir, rmdir
 - Includes read and write streams
+- Supports `fs.watch` and `fs.watchFile`
+
+## Watching
+
+`fs.watch` is powered by a [`FileSystemObserver`][observer]: the constructor
+passed through the `FileSystemObserver` option is used when provided, otherwise
+the global one — shipped natively in Chrome 133+, which makes `fs.watch` work
+over real OPFS in the browser.
+
+```ts
+const fs = new FsaNodeFs(dirHandle, undefined, { FileSystemObserver });
+
+const watcher = fs.watch('/', { recursive: true }, (eventType, filename) => {
+  console.log(eventType, filename);
+});
+```
+
+Divergences from Node.js:
+
+- The FSA backend is asynchronous, so startup errors (e.g. a missing path) are
+  emitted as an `'error'` event on the returned watcher instead of being
+  thrown synchronously. The `ignore`, `signal`, and `throwIfNoEntry` options
+  are not supported, and `persistent` is a no-op.
+- `fs.watchFile` polls the file at `interval` — comparing the handle's
+  `File#lastModified` and size — since the FSA API exposes no stat change
+  notifications.
+
+[observer]: https://developer.mozilla.org/en-US/docs/Web/API/FileSystemObserver
 
 ## License
 

packages/fs-fsa-to-node/src/__tests__/watch-chain.test.ts

@@ -0,0 +1,93 @@
+import { memfs } from 'memfs';
+import { nodeToFsa, NodeFileSystemObserver } from '@jsonjoy.com/fs-node-to-fsa';
+import { FsaNodeFs } from '../FsaNodeFs';
+import { onlyOnNode20 } from './util';
+
+const tick = (ms: number = 1) => new Promise(r => setTimeout(r, ms));
+
+const until = async (check: () => boolean | Promise<boolean>, pollInterval: number = 1) => {
+  do {
+    if (await check()) return;
+    await tick(pollInterval);
+  } while (true);
+};
+
+/**
+ * Chains both bridges: a memfs volume is exposed as FSA through
+ * `fs-node-to-fsa` (including its `NodeFileSystemObserver`), and that FSA is
+ * exposed back as a Node.js `fs` API through `fs-fsa-to-node`, so `fs.watch`
+ * events flow across the whole stack.
+ */
+const setup = () => {
+  const { fs: mfs } = memfs({ mountpoint: null });
+  const dir = nodeToFsa(mfs, '/mountpoint', { mode: 'readwrite' });
+  class Observer extends NodeFileSystemObserver {
+    constructor(callback: ConstructorParameters<typeof NodeFileSystemObserver>[1]) {
+      super(mfs as any, callback);
+    }
+  }
+  const fs = new FsaNodeFs(dir, undefined, { FileSystemObserver: Observer });
+  return { fs, mfs };
+};
+
+onlyOnNode20('fs.watch() across both bridges', () => {
+  test('file creation arrives as a "rename" event', async () => {
+    const { fs, mfs } = setup();
+    const events: [string, unknown][] = [];
+    const watcher = fs.watch('/', (eventType, filename) => events.push([eventType, filename]));
+    await tick();
+    mfs.writeFileSync('/mountpoint/file.txt', 'hello');
+    await until(() => events.length >= 1);
+    expect(events[0]).toEqual(['rename', 'file.txt']);
+    watcher.close();
+  });
+
+  test('file modification arrives as a "change" event', async () => {
+    const { fs, mfs } = setup();
+    mfs.writeFileSync('/mountpoint/file.txt', '');
+    const events: [string, unknown][] = [];
+    const watcher = fs.watch('/', (eventType, filename) => events.push([eventType, filename]));
+    await tick();
+    mfs.appendFileSync('/mountpoint/file.txt', 'more');
+    await until(() => events.length >= 1);
+    expect(events[0]).toEqual(['change', 'file.txt']);
+    watcher.close();
+  });
+
+  test('file deletion arrives as a "rename" event', async () => {
+    const { fs, mfs } = setup();
+    mfs.writeFileSync('/mountpoint/file.txt', 'x');
+    const events: [string, unknown][] = [];
+    const watcher = fs.watch('/', (eventType, filename) => events.push([eventType, filename]));
+    await tick();
+    mfs.unlinkSync('/mountpoint/file.txt');
+    await until(() => events.length >= 1);
+    expect(events[0]).toEqual(['rename', 'file.txt']);
+    watcher.close();
+  });
+
+  test('a rename arrives as two "rename" events, one per path', async () => {
+    const { fs, mfs } = setup();
+    mfs.writeFileSync('/mountpoint/a.txt', 'x');
+    const events: [string, unknown][] = [];
+    const watcher = fs.watch('/', (eventType, filename) => events.push([eventType, filename]));
+    await tick();
+    mfs.renameSync('/mountpoint/a.txt', '/mountpoint/b.txt');
+    await until(() => events.length >= 2);
+    expect(events.every(([eventType]) => eventType === 'rename')).toBe(true);
+    expect(events.map(([, filename]) => filename).sort()).toEqual(['a.txt', 'b.txt']);
+    watcher.close();
+  });
+
+  test('recursive watch reports nested paths', async () => {
+    const { fs, mfs } = setup();
+    mfs.mkdirSync('/mountpoint/sub');
+    const events: [string, unknown][] = [];
+    const watcher = fs.watch('/', { recursive: true }, (eventType, filename) => events.push([eventType, filename]));
+    await tick();
+    mfs.writeFileSync('/mountpoint/sub/deep.txt', 'x');
+    await until(() => events.length >= 1);
+    expect(events[0]).toEqual(['rename', 'sub/deep.txt']);
+    watcher.close();
+  });
+});

packages/fs-node-to-fsa/README.md

@@ -22,9 +22,28 @@ for await (const [name, handle] of dir.entries()) {
 }
 ```
 
+## FileSystemObserver
+
+`NodeFileSystemObserver` implements the [`FileSystemObserver` proposal][observer]
+on top of the Node.js `fs.watch` API. It is a best-effort implementation, per
+the proposal's allowance for local file systems: `rename` events are classified
+into `"appeared"`/`"disappeared"` records by stat-ing the path, and no
+`"moved"` records are ever produced.
+
+```ts
+import { nodeToFsa, NodeFileSystemObserver } from '@jsonjoy.com/fs-node-to-fsa';
+import * as fs from 'fs';
+
+const dir = nodeToFsa(fs, '/path/to/directory', { mode: 'readwrite' });
+const observer = new NodeFileSystemObserver(fs, records => console.log(records));
+await observer.observe(dir, { recursive: true });
+```
+
 ## Reference
 
 - [`nodeToFsa(fs, path, ctx)`](src/index.ts) - Converts a Node.js `fs` module to a `FileSystemDirectoryHandle`.
+- [`NodeFileSystemObserver`](src/NodeFileSystemObserver.ts) - A `FileSystemObserver` implementation backed by `fs.watch`.
 
 [node-fs]: https://nodejs.org/api/fs.html
 [fsa]: https://developer.mozilla.org/en-US/docs/Web/API/File_System_Access_API
+[observer]: https://developer.mozilla.org/en-US/docs/Web/API/FileSystemObserver

packages/fs-node-to-fsa/src/NodeFileSystemObserver.ts

@@ -0,0 +1,145 @@
+import { FileSystemChangeRecord } from '@jsonjoy.com/fs-fsa';
+import { NodeFileSystemDirectoryHandle } from './NodeFileSystemDirectoryHandle';
+import { NodeFileSystemFileHandle } from './NodeFileSystemFileHandle';
+import { newNotAllowedError, newNotFoundError } from './util';
+import type {
+  IFileSystemChangeRecord,
+  IFileSystemDirectoryHandle,
+  IFileSystemFileHandle,
+  IFileSystemHandle,
+  IFileSystemObserver,
+  IFileSystemObserverObserveOptions,
+  IFileSystemSyncAccessHandle,
+} from '@jsonjoy.com/fs-fsa';
+import type { FsCallbackApi } from '@jsonjoy.com/fs-node-utils';
+import type * as misc from '@jsonjoy.com/fs-node-utils/lib/types/misc';
+import type { NodeFsaContext, NodeFsaFs } from './types';
+
+export type NodeFsaWatchFs = NodeFsaFs & Pick<FsCallbackApi, 'watch'>;
+
+/**
+ * A `FileSystemObserver` implementation backed by the underlying Node.js-like
+ * `fs.watch`. This is a best-effort profile, per the File System Observer
+ * proposal's allowance for local file systems: `rename` events are classified
+ * into `"appeared"`/`"disappeared"` records by stat-ing the path, no
+ * `"moved"` records are ever produced (pairing renames is unreliable — same
+ * as Chrome on Windows), and a backend watcher error surfaces as a terminal
+ * `"errored"` record for that observation.
+ *
+ * @see https://developer.mozilla.org/en-US/docs/Web/API/FileSystemObserver
+ */
+export class NodeFileSystemObserver implements IFileSystemObserver {
+  protected readonly _observations = new Map<
+    IFileSystemFileHandle | IFileSystemDirectoryHandle | IFileSystemSyncAccessHandle,
+    misc.IFSWatcher
+  >();
+  protected _records: IFileSystemChangeRecord[] = [];
+  protected _flushScheduled: boolean = false;
+
+  constructor(
+    protected readonly fs: NodeFsaWatchFs,
+    protected readonly callback: (records: IFileSystemChangeRecord[], observer: IFileSystemObserver) => void,
+  ) {}
+
+  public async observe(
+    handle: IFileSystemFileHandle | IFileSystemDirectoryHandle | IFileSystemSyncAccessHandle,
+    options?: IFileSystemObserverObserveOptions,
+  ): Promise<void> {
+    const path = (handle as unknown as { __path?: unknown }).__path;
+    const ctx = ((handle as any).ctx ?? (handle as any)._ctx) as NodeFsaContext | undefined;
+    if (typeof path !== 'string' || !ctx || (ctx.separator !== '/' && ctx.separator !== '\\'))
+      throw new TypeError("Failed to execute 'observe' on 'FileSystemObserver': Invalid handle.");
+    const isDirectory = (handle as IFileSystemHandle).kind === 'directory';
+    const last = path[path.length - 1];
+    const isTrimmableSeparator = path.length > 1 && (last === '/' || last === '\\') && path[path.length - 2] !== ':';
+    const target = isTrimmableSeparator ? path.slice(0, -1) : path;
+    try {
+      await this.fs.promises.stat(target);
+    } catch (error) {
+      if (error && typeof error === 'object') {
+        switch (error.code) {
+          case 'ENOENT':
+            throw newNotFoundError();
+          case 'EACCES':
+          case 'EPERM':
+            throw newNotAllowedError();
+        }
+      }
+      throw error;
+    }
+    const recursive = isDirectory && !!options?.recursive;
+    const watcher = this.fs.watch(target, { recursive }, (eventType, filename) => {
+      void this.onEvent(handle, watcher, target, isDirectory, ctx, eventType, filename ? String(filename) : '');
+    });
+    watcher.on('error', () => {
+      if (this._observations.get(handle) !== watcher) return;
+      this._observations.delete(handle);
+      watcher.close();
+      this._enqueue(new FileSystemChangeRecord(handle, 'errored', null, []));
+    });
+    this._observations.get(handle)?.close();
+    this._observations.set(handle, watcher);
+  }
+
+  public unobserve(handle: IFileSystemFileHandle | IFileSystemDirectoryHandle | IFileSystemSyncAccessHandle): void {
+    const watcher = this._observations.get(handle);
+    if (!watcher) return;
+    watcher.close();
+    this._observations.delete(handle);
+  }
+
+  /** Disconnect and stop all observations. */
+  public disconnect(): void {
+    for (const watcher of this._observations.values()) watcher.close();
+    this._observations.clear();
+    this._records = [];
+  }
+
+  protected async onEvent(
+    root: IFileSystemFileHandle | IFileSystemDirectoryHandle | IFileSystemSyncAccessHandle,
+    watcher: misc.IFSWatcher,
+    rootPath: string,
+    isDirectory: boolean,
+    ctx: NodeFsaContext,
+    eventType: string,
+    filename: string,
+  ): Promise<void> {
+    const sep = ctx.separator;
+    const steps = isDirectory && filename ? filename.split(sep) : [];
+    const absolute = isDirectory ? (rootPath === sep ? rootPath + filename : rootPath + sep + filename) : rootPath;
+    let stats: misc.IStats | null = null;
+    try {
+      stats = (await this.fs.promises.stat(absolute)) as misc.IStats;
+    } catch {
+      stats = null;
+    }
+    if (this._observations.get(root) !== watcher) return;
+    if (eventType === 'rename') {
+      if (stats) {
+        this._enqueue(new FileSystemChangeRecord(root, 'appeared', this._handle(absolute, stats, ctx), steps));
+      } else {
+        this._enqueue(new FileSystemChangeRecord(root, 'disappeared', null, steps));
+      }
+    } else if (stats) {
+      this._enqueue(new FileSystemChangeRecord(root, 'modified', this._handle(absolute, stats, ctx), steps));
+    }
+  }
+
+  protected _handle(absolute: string, stats: misc.IStats, ctx: NodeFsaContext): IFileSystemHandle {
+    return stats.isDirectory()
+      ? new NodeFileSystemDirectoryHandle(this.fs, absolute, ctx)
+      : new NodeFileSystemFileHandle(this.fs, absolute, ctx);
+  }
+
+  protected _enqueue(record: IFileSystemChangeRecord): void {
+    this._records.push(record);
+    if (this._flushScheduled) return;
+    this._flushScheduled = true;
+    queueMicrotask(() => {
+      this._flushScheduled = false;
+      const records = this._records;
+      this._records = [];
+      if (records.length) this.callback(records, this);
+    });
+  }
+}

packages/fs-node-to-fsa/src/NodeFileSystemSyncAccessHandle.ts

@@ -11,11 +11,11 @@ export class NodeFileSystemSyncAccessHandle implements IFileSystemSyncAccessHand
 
   constructor(
     protected readonly fs: NodeFsaFs,
-    protected readonly path: string,
+    public readonly __path: string,
     protected readonly ctx: NodeFsaContext,
   ) {
-    this.fd = fs.openSync(path, 'r+');
-    this.ctx.locks.acquireLock(this.path);
+    this.fd = fs.openSync(__path, 'r+');
+    this.ctx.locks.acquireLock(this.__path);
   }
 
   /**
@@ -24,7 +24,7 @@ export class NodeFileSystemSyncAccessHandle implements IFileSystemSyncAccessHand
   public async close(): Promise<void> {
     assertCanWrite(this.ctx.mode);
     this.fs.closeSync(this.fd);
-    this.ctx.locks.releaseLock(this.path);
+    this.ctx.locks.releaseLock(this.__path);
   }
 
   /**
@@ -39,7 +39,7 @@ export class NodeFileSystemSyncAccessHandle implements IFileSystemSyncAccessHand
    * @see https://developer.mozilla.org/en-US/docs/Web/API/FileSystemSyncAccessHandle/getSize
    */
   public async getSize(): Promise<number> {
-    return this.fs.statSync(this.path).size;
+    return this.fs.statSync(this.__path).size;
   }
 
   /**