fix(vue): keep live queries subscribed across refreshes

.changeset/steady-live-subscriptions.md

@@ -0,0 +1,5 @@
+---
+"@effect-app/vue": patch
+---
+
+Keep live queries subscribed across successive repository writes: dispatch each write/reset to a snapshot of current registrations, skipping ones removed or replaced mid-dispatch. Live queries now also honor `invalidation: "soft"`.

packages/vue/src/atomQuery.ts

@@ -394,14 +394,15 @@ export const withQueryOptions = <A, E>(
   }
   if (opts.live && liveKey !== undefined) {
     const liveOptions = opts.live === true ? {} : opts.live
-    atom = Atom.transform(atom, (get) => {
+    const inner = atom
+    atom = Atom.transform(inner, (get) => {
       const unregister = registerLiveQuery(
         liveKey,
         () => getQueryReadDependencies(liveKey),
         liveOptions
       )
       get.addFinalizer(unregister)
-      return get(self)
+      return get(inner)
     }, { initialValueTarget: self })
   }
   const revalidateOnFocus = opts.revalidateOnFocus ?? true

packages/vue/src/liveQueryInvalidation.ts

@@ -17,6 +17,7 @@ export interface LiveQueryInvalidationSource {
 }
 
 interface LiveQueryEntry {
+  readonly hash: number
   readonly key: ReadonlyArray<unknown>
   readonly reads: () => DataDependencies.DataDependencies
   observers: number
@@ -42,10 +43,13 @@ const schedule = (entry: LiveQueryEntry) => {
   entry.timer = setTimeout(() => invalidateEntry(entry), maxDelayMs)
 }
 
+// a synchronous refresh can remove or replace registrations while dispatching
+const isCurrent = (entry: LiveQueryEntry) => entries.get(entry.hash) === entry
+
 const dispatch = (writes: DataDependencies.DataDependencies) => {
   if (!DataDependencies.isNonEmpty(writes)) return
-  for (const entry of entries.values()) {
-    if (DataDependencies.intersects(entry.reads(), writes)) schedule(entry)
+  for (const entry of Array.from(entries.values())) {
+    if (isCurrent(entry) && DataDependencies.intersects(entry.reads(), writes)) schedule(entry)
   }
 }
 
@@ -59,8 +63,8 @@ const receive = (writes: DataDependencies.DataDependencies) => {
 
 const reset = () => {
   bufferedWrites = DataDependencies.empty()
-  for (const entry of entries.values()) {
-    if (DataDependencies.isNonEmpty(entry.reads())) schedule(entry)
+  for (const entry of Array.from(entries.values())) {
+    if (isCurrent(entry) && DataDependencies.isNonEmpty(entry.reads())) schedule(entry)
   }
 }
 
@@ -90,6 +94,7 @@ export const registerLiveQuery = (
     current.delays.set(delay, (current.delays.get(delay) ?? 0) + 1)
   } else {
     entries.set(hash, {
+      hash,
       key,
       reads,
       observers: 1,

packages/vue/test/liveQueryInvalidation.test.ts

@@ -1,5 +1,11 @@
-import { DataDependencies } from "effect-app/client"
+import { defaultRegistry } from "@effect/atom-vue"
+import { DataDependencies, makeQueryKey } from "effect-app/client"
+import * as Effect from "effect-app/Effect"
+import * as Layer from "effect-app/Layer"
+import * as ManagedRuntime from "effect/ManagedRuntime"
 import { afterEach, describe, expect, it, vi } from "vitest"
+import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, invalidateSoft, makeAtomClientRuntime, withQueryOptions } from "../src/atomQuery.js"
+import { getQueryInvalidationMode } from "../src/dependencyMetadata.js"
 import { beginLiveQueryFetch, configureLiveQueryInvalidation, endLiveQueryFetch, registerLiveQuery } from "../src/liveQueryInvalidation.js"
 
 const repo = DataDependencies.repo("Inventory", ["item-1"])
@@ -13,6 +19,114 @@ describe("live query invalidation", () => {
     vi.useRealTimers()
   })
 
+  it.each(["await", "soft"] as const)(
+    "keeps an observed %s query subscribed after successive repository-write refreshes",
+    async (invalidation) => {
+      const dependency = DataDependencies.repo("LiveOrders")
+      const mrt = ManagedRuntime.make(Layer.empty)
+      const rt = makeAtomClientRuntime(() => Layer.empty, mrt.memoMap)
+      let calls = 0
+      let receive = (_writes: DataDependencies.DataDependencies) => {}
+      let reset = () => {}
+      let settled = Promise.resolve()
+      let refreshes = 0
+      configureLiveQueryInvalidation({
+        ready: () => Promise.resolve(),
+        subscribe: (onWrites, onReset) => {
+          receive = onWrites
+          reset = onReset
+          return () => {
+            receive = () => {}
+            reset = () => {}
+          }
+        },
+        invalidate: (keys) => {
+          refreshes++
+          settled = Effect.runPromise(
+            keys.every((key) => getQueryInvalidationMode(key) === "soft")
+              ? invalidateSoft(keys).pipe(Effect.andThen(awaitAtomResult(defaultRegistry, atom)))
+              : invalidateAndAwait(keys)
+          )
+        }
+      })
+      const query = {
+        id: "LiveOrders.List",
+        handler: () => DataDependencies.read(dependency).pipe(Effect.andThen(Effect.sync(() => ++calls)))
+      }
+      const base = buildQueryFamily(rt, query)(undefined)
+      const atom = withQueryOptions(base, { live: true, invalidation, revalidateOnFocus: false }, [
+        ...makeQueryKey(query),
+        undefined
+      ])
+      const unmount = defaultRegistry.mount(atom)
+      try {
+        await Effect.runPromise(awaitAtomResult(defaultRegistry, atom))
+        expect(calls).toBe(1)
+        expect(getQueryInvalidationMode([...makeQueryKey(query), undefined])).toBe(invalidation)
+        receive(new Set([dependency]))
+        await settled
+        expect(calls).toBe(2)
+        expect(refreshes).toBe(1)
+        receive(new Set([dependency]))
+        await settled
+        expect(calls).toBe(3)
+        expect(refreshes).toBe(2)
+        reset()
+        await settled
+        expect(calls).toBe(4)
+        expect(refreshes).toBe(3)
+        receive(new Set([dependency]))
+        await settled
+        expect(calls).toBe(5)
+        expect(refreshes).toBe(4)
+      } finally {
+        unmount()
+        defaultRegistry.reset()
+        await mrt.dispose()
+      }
+    }
+  )
+
+  it.each(["write", "reset"] as const)(
+    "skips registrations removed or replaced by a synchronous refresh during %s dispatch",
+    (event) => {
+      vi.useFakeTimers()
+      const invalidations: Array<ReadonlyArray<ReadonlyArray<unknown>>> = []
+      let receive = (_writes: DataDependencies.DataDependencies) => {}
+      let reset = () => {}
+      const first = ["$Inventory", "List", undefined]
+      const removed = ["$Inventory", "Get", { id: "item-1" }]
+      const replaced = ["$Inventory", "Get", { id: "item-2" }]
+      let unregisterRemoved = () => {}
+      let unregisterReplaced = () => {}
+      configureLiveQueryInvalidation({
+        ready: () => Promise.resolve(),
+        subscribe: (onWrites, onReset) => {
+          receive = onWrites
+          reset = onReset
+          return () => {}
+        },
+        invalidate: (keys) => {
+          invalidations.push(keys)
+          if (keys[0] !== first) return
+          unregisterRemoved()
+          unregisterReplaced()
+          unregisterReplaced = registerLiveQuery(replaced, () => new Set([repo]), { maxDelayMs: 1_000 })
+        }
+      })
+
+      cleanups.push(registerLiveQuery(first, () => new Set([repo]), {}))
+      unregisterRemoved = registerLiveQuery(removed, () => new Set([repo]), {})
+      unregisterReplaced = registerLiveQuery(replaced, () => new Set([repo]), {})
+      cleanups.push(() => unregisterReplaced())
+
+      if (event === "write") receive(new Set([repo]))
+      else reset()
+      vi.runAllTimers()
+      expect(invalidations).toEqual([[first]])
+    }
+  )
+
   it("invalidates only live queries whose recorded reads intersect", () => {
     const invalidations: Array<ReadonlyArray<ReadonlyArray<unknown>>> = []
     let receive = (_writes: DataDependencies.DataDependencies) => {}