packages/extension-query/src/MutationRunner.ts

@@ -1,5 +1,5 @@
 import * as AsyncData from "@typed/async-data"
-import { type Context, Effect, type Fiber, Ref, SubscriptionRef } from "effect"
+import { type Context, Effect, type Fiber, Queue, Ref, Stream, SubscriptionRef } from "effect"
 import type * as QueryClient from "./QueryClient.js"
 
 
@@ -7,8 +7,11 @@ export interface MutationRunner<K extends readonly unknown[], A, E, R> {
     readonly context: Context.Context<R>
     readonly stateRef: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
 
-    readonly mutate: (...key: K) => Effect.Effect<A, E>
-    readonly forkMutate: (...key: K) => Effect.Effect<Fiber.RuntimeFiber<A, E>>
+    readonly mutate: (...key: K) => Effect.Effect<AsyncData.Success<A> | AsyncData.Failure<E>>
+    readonly forkMutate: (...key: K) => Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
 }
 
 
@@ -30,19 +33,41 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
     const context = yield* Effect.context<R | QueryClient.TagClassShape<EH, HandledE> | EH>()
     const stateRef = yield* SubscriptionRef.make(AsyncData.noData<A, Exclude<E, HandledE>>())
 
-    const mutate = (...key: K) => QueryClient.pipe(
+    const run = (
+        key: K,
+        setState: (value: AsyncData.AsyncData<A, Exclude<E, HandledE>>) => Effect.Effect<void>,
+    ) => QueryClient.pipe(
         Effect.flatMap(client => client.ErrorHandler),
-        Effect.flatMap(errorHandler => Ref.set(stateRef, AsyncData.loading()).pipe(
+        Effect.flatMap(errorHandler => setState(AsyncData.loading()).pipe(
             Effect.andThen(mutation(key)),
             errorHandler.handle,
-            Effect.tapErrorCause(c => Ref.set(stateRef, AsyncData.failure(c))),
-            Effect.tap(v => Ref.set(stateRef, AsyncData.success(v))),
+            Effect.matchCauseEffect({
+                onSuccess: v => Effect.succeed(AsyncData.success(v)).pipe(
+                    Effect.tap(setState)
+                ),
+                onFailure: c => Effect.succeed(AsyncData.failure(c)).pipe(
+                    Effect.tap(setState)
+                ),
+            }),
         )),
 
         Effect.provide(context),
     )
 
-    const forkMutate = (...key: K) => Effect.forkDaemon(mutate(...key))
+    const mutate = (...key: K) => run(key, value => Ref.set(stateRef, value))
+
+    const forkMutate = (...key: K) => Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>().pipe(
+        Effect.flatMap(stateQueue =>
+            run(key, value => Queue.offer(stateQueue, value).pipe(
+                Effect.andThen(Ref.set(stateRef, value))
+            )).pipe(
+                Effect.tap(() => Queue.shutdown(stateQueue)),
+                Effect.forkDaemon,
+                Effect.map(fiber => [fiber, Stream.fromQueue(stateQueue)] as const)
+            )
+        )
+    )
+
 
     return {
         context,

packages/extension-query/src/MutationService.ts

@@ -1,11 +1,14 @@
 import type * as AsyncData from "@typed/async-data"
-import { Effect, type Fiber, type SubscriptionRef } from "effect"
+import { Effect, type Fiber, type Stream, type SubscriptionRef } from "effect"
 
 
 export interface MutationService<K extends readonly unknown[], A, E> {
     readonly state: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly mutate: (...key: K) => Effect.Effect<A, E>
-    readonly forkMutate: (...key: K) => Effect.Effect<Fiber.RuntimeFiber<A, E>>
+    readonly mutate: (...key: K) => Effect.Effect<AsyncData.Success<A> | AsyncData.Failure<E>>
+    readonly forkMutate: (...key: K) => Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
 }
 
 export const Tag = <const Id extends string>(id: Id) => <

packages/extension-query/src/QueryExtension.ts

@@ -32,8 +32,11 @@ export interface UseMutationProps<K extends readonly unknown[], A, E, R> {
 
 export interface UseMutationResult<K extends readonly unknown[], A, E> {
     readonly state: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly mutate: (...key: K) => Effect.Effect<A, E>
-    readonly forkMutate: (...key: K) => Effect.Effect<Fiber.RuntimeFiber<A, E>>
+    readonly mutate: (...key: K) => Effect.Effect<AsyncData.Success<A> | AsyncData.Failure<E>>
+    readonly forkMutate: (...key: K) => Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
 
     readonly layer: <Self, Id extends string>(
         tag: Context.TagClass<Self, Id, MutationService.MutationService<K, A, E>>