refactor: consolidate SDK Watchable with generic map/eq and rename call to fetch

sdk/base/lib/util/Watchable.ts

@@ -1,55 +1,124 @@
 import { Effects } from '../Effects'
 import { AbortedError } from './AbortedError'
+import { deepEqual } from './deepEqual'
 import { DropGenerator, DropPromise } from './Drop'
 
-export abstract class Watchable<T> {
-  constructor(readonly effects: Effects) {}
+export abstract class Watchable<Raw, Mapped = Raw> {
+  protected readonly mapFn: (value: Raw) => Mapped
+  protected readonly eqFn: (a: Mapped, b: Mapped) => boolean
 
-  protected abstract call(callback?: () => void): Promise<T>
+  constructor(
+    readonly effects: Effects,
+    options?: {
+      map?: (value: Raw) => Mapped
+      eq?: (a: Mapped, b: Mapped) => boolean
+    },
+  ) {
+    this.mapFn = options?.map ?? ((a) => a as unknown as Mapped)
+    this.eqFn = options?.eq ?? ((a, b) => deepEqual(a, b))
+  }
+
+  /**
+   * Fetch the current value, optionally registering a callback for change notification.
+   * The callback should be invoked when the underlying data changes.
+   */
+  protected abstract fetch(callback?: () => void): Promise<Raw>
   protected abstract readonly label: string
 
   /**
-   * Returns the value. Reruns the context from which it has been called if the underlying value changes
+   * Produce a stream of raw values. Default implementation uses fetch() with
+   * effects callback in a loop. Override for custom subscription mechanisms
+   * (e.g. fs.watch).
    */
-  const(): Promise<T> {
-    return this.call(
-      this.effects.constRetry &&
-        (() => this.effects.constRetry && this.effects.constRetry()),
-    )
-  }
-
-  /**
-   * Returns the value. Does nothing if the value changes
-   */
-  once(): Promise<T> {
-    return this.call()
-  }
-
-  private async *watchGen(abort?: AbortSignal) {
+  protected async *produce(abort: AbortSignal): AsyncGenerator<Raw, void> {
     const resolveCell = { resolve: () => {} }
     this.effects.onLeaveContext(() => {
       resolveCell.resolve()
     })
-    abort?.addEventListener('abort', () => resolveCell.resolve())
-    while (this.effects.isInContext && !abort?.aborted) {
+    abort.addEventListener('abort', () => resolveCell.resolve())
+    while (this.effects.isInContext && !abort.aborted) {
       let callback: () => void = () => {}
       const waitForNext = new Promise<void>((resolve) => {
         callback = resolve
         resolveCell.resolve = resolve
       })
-      yield await this.call(() => callback())
+      yield await this.fetch(() => callback())
       await waitForNext
     }
-    return new Promise<never>((_, rej) => rej(new AbortedError()))
+  }
+
+  /**
+   * Lifecycle hook called when const() registers a subscription.
+   * Return a cleanup function to be called when the subscription ends.
+   * Override for side effects like FileHelper's consts tracking.
+   */
+  protected onConstRegistered(_value: Mapped): (() => void) | void {}
+
+  /**
+   * Internal generator that maps raw values and deduplicates using eq.
+   */
+  private async *watchGen(
+    abort: AbortSignal,
+  ): AsyncGenerator<Mapped, void, unknown> {
+    let prev: { value: Mapped } | null = null
+    for await (const raw of this.produce(abort)) {
+      if (abort.aborted) return
+      const mapped = this.mapFn(raw)
+      if (!prev || !this.eqFn(prev.value, mapped)) {
+        prev = { value: mapped }
+        yield mapped
+      }
+    }
+  }
+
+  /**
+   * Returns the value. Reruns the context from which it has been called if the underlying value changes
+   */
+  async const(): Promise<Mapped> {
+    const abort = new AbortController()
+    const gen = this.watchGen(abort.signal)
+    const res = await gen.next()
+    const value = res.value as Mapped
+    if (this.effects.constRetry) {
+      const constRetry = this.effects.constRetry
+      const cleanup = this.onConstRegistered(value)
+      gen.next().then(
+        () => {
+          abort.abort()
+          cleanup?.()
+          constRetry()
+        },
+        () => {
+          abort.abort()
+          cleanup?.()
+        },
+      )
+    } else {
+      abort.abort()
+    }
+    return value
+  }
+
+  /**
+   * Returns the value. Does nothing if the value changes
+   */
+  async once(): Promise<Mapped> {
+    return this.mapFn(await this.fetch())
   }
 
   /**
    * Watches the value. Returns an async iterator that yields whenever the value changes
    */
-  watch(abort?: AbortSignal): AsyncGenerator<T, never, unknown> {
+  watch(abort?: AbortSignal): AsyncGenerator<Mapped, never, unknown> {
     const ctrl = new AbortController()
     abort?.addEventListener('abort', () => ctrl.abort())
-    return DropGenerator.of(this.watchGen(ctrl.signal), () => ctrl.abort())
+    return DropGenerator.of(
+      (async function* (gen): AsyncGenerator<Mapped, never, unknown> {
+        yield* gen
+        throw new AbortedError()
+      })(this.watchGen(ctrl.signal)),
+      () => ctrl.abort(),
+    )
   }
 
   /**
@@ -57,13 +126,13 @@ export abstract class Watchable<T> {
    */
   onChange(
     callback: (
-      value: T | undefined,
+      value: Mapped | undefined,
       error?: Error,
     ) => { cancel: boolean } | Promise<{ cancel: boolean }>,
   ) {
     ;(async () => {
       const ctrl = new AbortController()
-      for await (const value of this.watch(ctrl.signal)) {
+      for await (const value of this.watchGen(ctrl.signal)) {
         try {
           const res = await callback(value)
           if (res.cancel) {
@@ -90,7 +159,7 @@ export abstract class Watchable<T> {
   /**
    * Watches the value. Returns when the predicate is true
    */
-  waitFor(pred: (value: T) => boolean): Promise<T> {
+  waitFor(pred: (value: Mapped) => boolean): Promise<Mapped> {
     const ctrl = new AbortController()
     return DropPromise.of(
       Promise.resolve().then(async () => {