fix(vue): flush stream write-deps on first arrival

.changeset/stream-first-write-flush.md

@@ -0,0 +1,5 @@
+---
+"@effect-app/vue": patch
+---
+
+Flush stream mutation write-deps once when the first write arrives, then again on settlement. Long-running streams can refresh queries like GetActiveJob without invalidating list queries on every subsequent item write.

packages/vue/src/mutate.ts

@@ -478,18 +478,28 @@ export const makeStreamMutation2 = <RInvalidator>(queryInvalidator: QueryInvalid
     const makeInvocationEffect = (input: unknown, source: Stream.Stream<any, any, any>) =>
       Effect.gen(function*() {
         const keysRef = yield* Ref.make<ReadonlyArray<InvalidationKey>>([])
-        // Stream metadata can arrive after every emitted RPC value. Accumulate its invalidation
-        // keys and flush them once from `ensuring`; invalidating from `add` would refetch live
-        // queries once per message and repeatedly cancel the preceding request.
+        // Server invalidation keys stay settlement-only: flushing them from `add` refetched
+        // live queries once per chunk (One-Pick List storms). Write-deps flush once when the
+        // first write arrives (job create → GetActiveJob) and again from `ensuring`.
         const invKeys = makeInvalidationKeysService(keysRef)
         const readsRef = yield* Ref.make(DataDependencies.empty())
         const writesRef = yield* Ref.make(DataDependencies.empty())
         const dependencyRecorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef)
         const lastRef = yield* Ref.make<any>(undefined)
+        let flushedFirstWrites = false
+        const flushFirstWrites = Effect.gen(function*() {
+          if (flushedFirstWrites) return
+          const writeDependencies = yield* Ref.get(writesRef)
+          if (!DataDependencies.isNonEmpty(writeDependencies)) return
+          flushedFirstWrites = true
+          const lastValue = yield* Ref.get(lastRef)
+          yield* invCache(input, Exit.succeed(lastValue), [], writeDependencies)
+        })
         return source.pipe(
           Stream.provideService(InvalidationKeysFromServer, invKeys),
           Stream.provideService(DataDependencies.DataDependencyRecorder, dependencyRecorder),
           Stream.tap((v) => Ref.set(lastRef, v)),
+          Stream.tap(() => flushFirstWrites),
           Stream.ensuring(
             Effect.gen(function*() {
               const lastValue = yield* Ref.get(lastRef)

packages/vue/test/dependencyInvalidation.test.ts

@@ -5,6 +5,7 @@ import { QueryClient, VueQueryPlugin } from "@tanstack/vue-query"
 import { DataDependencies, type InvalidationKey, InvalidationKeysFromServer, makeQueryKey } from "effect-app/client"
 import * as Context from "effect-app/Context"
 import * as Effect from "effect-app/Effect"
+import * as Deferred from "effect/Deferred"
 import * as Fiber from "effect/Fiber"
 import * as Layer from "effect/Layer"
 import * as ManagedRuntime from "effect/ManagedRuntime"
@@ -40,6 +41,52 @@ it.live("stream mutations accumulate repeated server keys and invalidate once wh
     expect(calls).toEqual([[key]])
   }))
 
+it.live("stream mutations flush write-deps once when they first arrive, then again on settle", () =>
+  Effect.gen(function*() {
+    const jobKey: InvalidationKey = ["$PickList", "GetActiveJob"]
+    const listKey: InvalidationKey = ["$PickList", "List"]
+    const jobRepo = DataDependencies.repo("PickJob")
+    const itemRepo = DataDependencies.repo("PickItem")
+    setQueryReadDependencies(jobKey, new Set([jobRepo]))
+    setQueryReadDependencies(listKey, new Set([itemRepo]))
+
+    const calls: Array<ReadonlyArray<ReadonlyArray<unknown>>> = []
+    const firstFlush = yield* Deferred.make<void>()
+    const gate = yield* Deferred.make<void>()
+    const queryInvalidator = {
+      invalidateAndAwait: (keys: ReadonlyArray<ReadonlyArray<unknown>>) =>
+        Effect
+          .sync(() => {
+            calls.push(keys)
+          })
+          .pipe(
+            Effect.flatMap(() => calls.length === 1 ? Deferred.succeed(firstFlush, undefined) : Effect.void)
+          )
+    }
+    const mutation = makeStreamMutation2(queryInvalidator)({
+      id: "PickList.StartBatchPrint",
+      handler: () =>
+        Stream.make(1).pipe(
+          Stream.tap(() => DataDependencies.write(jobRepo)),
+          Stream.concat(Stream.fromEffect(Deferred.await(gate).pipe(Effect.as(2)))),
+          Stream.tap((n) => n === 2 ? DataDependencies.write(itemRepo) : Effect.void)
+        )
+    })
+
+    try {
+      const fiber = yield* Effect.forkChild(Stream.runDrain(mutation(undefined)))
+      yield* Deferred.await(firstFlush)
+      expect(calls).toEqual([[jobKey]])
+
+      yield* Deferred.succeed(gate, undefined)
+      yield* Fiber.join(fiber)
+      expect(calls).toEqual([[jobKey], [jobKey, listKey]])
+    } finally {
+      clearQueryReadDependencies(jobKey)
+      clearQueryReadDependencies(listKey)
+    }
+  }))
+
 // --- shared registry + derivation logic --------------------------------------------------------
 
 it("getDerivedInvalidationKeys returns keys of queries whose reads intersect the writes", () => {

wiki/repository-derived-query-invalidation.md

@@ -37,8 +37,8 @@ type DataDependency =
 Each request has a `DataDependencyRecorder` in context. Code can record:
 
 ```ts
-yield* DataDependencies.read(DataDependencies.repo("PickList"))
-yield* DataDependencies.write(DataDependencies.repo("PickList"))
+yield * DataDependencies.read(DataDependencies.repo("PickList"))
+yield * DataDependencies.write(DataDependencies.repo("PickList"))
 ```
 
 Repository operations do this automatically, so most resource handlers do not
@@ -136,8 +136,11 @@ the local `DataDependencyRecorder`. This makes dependency propagation work for
 both direct RPC clients and the Vue query/mutation helpers.
 
 Stream commands emit dependency metadata in the same metadata chunks already
-used for server-driven invalidation keys. Writes can therefore invalidate
-queries mid-stream or when the stream completes.
+used for server-driven invalidation keys. `@effect-app/vue` `makeStreamMutation2`
+flushes **write-deps once** when the first write arrives (so a created job
+refreshes `GetActiveJob` while the stream is still open) and again when the
+stream settles. Server invalidation keys stay settlement-only — flushing them
+per chunk refetched live list queries on every item.
 
 ## Vue cache integration