Mutation refactoring

packages/example/src/query.ts

@@ -1,9 +1,21 @@
 import { HttpClientError } from "@effect/platform"
 import { ErrorHandler, QueryClient } from "@reffuse/extension-query"
+import { Effect } from "effect"
 
 
-export class AppQueryErrorHandler extends ErrorHandler.Service("AppQueryErrorHandler")<AppQueryErrorHandler,
+export class AppQueryErrorHandler extends ErrorHandler.Service<AppQueryErrorHandler,
     HttpClientError.HttpClientError
->() {}
+>()(
+    "AppQueryErrorHandler",
 
-export class AppQueryClient extends QueryClient.Service({ ErrorHandler: AppQueryErrorHandler })<AppQueryClient>() {}
+    (self, failure, defect) => self.pipe(
+        Effect.catchTags({
+            RequestError: failure,
+            ResponseError: failure,
+        }),
+
+        Effect.catchAllDefect(defect),
+    ),
+) {}
+
+export class AppQueryClient extends QueryClient.Service<AppQueryClient>()({ ErrorHandler: AppQueryErrorHandler }) {}

packages/example/src/query/views/Uuid4QueryService.tsx

@@ -5,7 +5,7 @@ import { Uuid4Query } from "../services"
 
 
 export function Uuid4QueryService() {
-    const runSync = R.useRunSync()
+    const runFork = R.useRunFork()
 
     const query = R.useMemo(() => Uuid4Query.Uuid4Query, [])
     const [state] = R.useRefState(query.state)
@@ -25,7 +25,7 @@ export function Uuid4QueryService() {
                     })}
                 </Text>
 
-                <Button onClick={() => runSync(query.refresh)}>Refresh</Button>
+                <Button onClick={() => runFork(query.forkRefresh)}>Refresh</Button>
             </Flex>
         </Container>
     )

packages/example/src/routes/query/usequery.tsx

@@ -3,7 +3,7 @@ import { HttpClient } from "@effect/platform"
 import { Button, Container, Flex, Slider, Text } from "@radix-ui/themes"
 import { createFileRoute } from "@tanstack/react-router"
 import * as AsyncData from "@typed/async-data"
-import { Array, Console, Effect, flow, Option, Schema } from "effect"
+import { Array, Console, Effect, flow, Option, Schema, Stream } from "effect"
 import { useState } from "react"
 
 
@@ -15,7 +15,7 @@ export const Route = createFileRoute("/query/usequery")({
 const Result = Schema.Array(Schema.String)
 
 function RouteComponent() {
-    const runSync = R.useRunSync()
+    const runFork = R.useRunFork()
 
     const [count, setCount] = useState(1)
 
@@ -59,7 +59,15 @@ function RouteComponent() {
                     })}
                 </Text>
 
-                <Button onClick={() => runSync(query.refresh)}>Refresh</Button>
+                <Button
+                    onClick={() => query.forkRefresh.pipe(
+                        Effect.flatMap(([, state]) => Stream.runForEach(state, Console.log)),
+                        Effect.andThen(Console.log("Refresh finished or stopped")),
+                        runFork,
+                    )}
+                >
+                    Refresh
+                </Button>
             </Flex>
         </Container>
     )

packages/extension-lazyref/src/index.ts

@@ -13,7 +13,7 @@ export const LazyRefExtension = ReffuseExtension.make(() => ({
         const [reactStateValue, setReactStateValue] = React.useState(initialState)
 
         this.useFork(() => Stream.runForEach(
-            Stream.changes(ref.changes),
+            Stream.changesWith(ref.changes, (x, y) => x === y),
             v => Effect.sync(() => setReactStateValue(v)),
         ), [ref])
 

packages/extension-query/src/ErrorHandler.ts

@@ -1,4 +1,4 @@
-import { type Cause, Context, Effect, Layer, Queue, Stream } from "effect"
+import { Cause, Context, Effect, identity, Layer, Queue, Stream } from "effect"
 import type { Mutable } from "effect/Types"
 
 
@@ -14,24 +14,41 @@ export interface ServiceResult<Self, Id extends string, E> extends Context.TagCl
     readonly Live: Layer.Layer<Self>
 }
 
-export const Service = <const Id extends string>(id: Id) => (
-    <Self, E = never>(): ServiceResult<Self, Id, E> => {
+export const Service = <Self, E = never>() => (
+    <const Id extends string>(
+        id: Id,
+        f: <A, R>(
+            self: Effect.Effect<A, E, R>,
+            failure: (failure: E) => Effect.Effect<never>,
+            defect: (defect: unknown) => Effect.Effect<never>,
+        ) => Effect.Effect<A, never, R>,
+    ): ServiceResult<Self, Id, E> => {
         const TagClass = Context.Tag(id)() as ServiceResult<Self, Id, E>
+
         (TagClass as Mutable<typeof TagClass>).Live = Layer.effect(TagClass, Effect.gen(function*() {
             const queue = yield* Queue.unbounded<Cause.Cause<E>>()
             const errors = Stream.fromQueue(queue)
 
             const handle = <A, SelfE, R>(
                 self: Effect.Effect<A, SelfE, R>
-            ) => Effect.tapErrorCause(self, cause =>
-                Queue.offer(queue, cause as Cause.Cause<E>)
-            ) as Effect.Effect<A, Exclude<SelfE, E>, R>
+            ): Effect.Effect<A, Exclude<SelfE, E>, R> => f(self as unknown as Effect.Effect<A, E, R>,
+                (failure: E) => Queue.offer(queue, Cause.fail(failure)).pipe(
+                    Effect.andThen(Effect.failCause(Cause.empty))
+                ),
+                (defect: unknown) => Queue.offer(queue, Cause.die(defect)).pipe(
+                    Effect.andThen(Effect.failCause(Cause.empty))
+                ),
+            )
 
             return { errors, handle }
         }))
+
         return TagClass
     }
 )
 
 
-export class DefaultErrorHandler extends Service("@reffuse/extension-query/DefaultErrorHandler")<DefaultErrorHandler>() {}
+export class DefaultErrorHandler extends Service<DefaultErrorHandler>()(
+    "@reffuse/extension-query/DefaultErrorHandler",
+    identity,
+) {}

packages/extension-query/src/QueryClient.ts

@@ -23,13 +23,13 @@ export interface ServiceResult<Self, EH, HandledE> extends Context.TagClass<Self
     readonly Live: Layer.Layer<Self>
 }
 
-export const Service = <
+export const Service = <Self>() => (
+    <
         EH = ErrorHandler.DefaultErrorHandler,
         HandledE = ErrorHandler.Error<Context.Tag.Service<ErrorHandler.DefaultErrorHandler>>,
     >(
         props?: ServiceProps<EH, HandledE>
-) => (
-    <Self>(): ServiceResult<Self, EH, HandledE> => {
+    ): ServiceResult<Self, EH, HandledE> => {
         const TagClass = Context.Tag(id)() as ServiceResult<Self, EH, HandledE>
         (TagClass as Mutable<typeof TagClass>).Live = Layer.succeed(TagClass, {
             ErrorHandler: (props?.ErrorHandler ?? ErrorHandler.DefaultErrorHandler) as Context.Tag<EH, ErrorHandler.ErrorHandler<HandledE>>

packages/extension-query/src/QueryExtension.ts

@@ -18,7 +18,11 @@ export interface UseQueryProps<K extends readonly unknown[], A, E, R> {
 export interface UseQueryResult<K extends readonly unknown[], A, E> {
     readonly latestKey: SubscriptionRef.SubscriptionRef<Option.Option<K>>
     readonly state: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly refresh: Effect.Effect<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>
+
+    readonly forkRefresh: Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>, Cause.NoSuchElementException>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
 
     readonly layer: <Self, Id extends string>(
         tag: Context.TagClass<Self, Id, QueryService.QueryService<K, A, E>>
@@ -32,6 +36,7 @@ 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<AsyncData.Success<A> | AsyncData.Failure<E>>
     readonly forkMutate: (...key: K) => Effect.Effect<readonly [
         fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>>,
@@ -73,12 +78,13 @@ export const QueryExtension = ReffuseExtension.make(() => ({
         return React.useMemo(() => ({
             latestKey: runner.latestKeyRef,
             state: runner.stateRef,
-            refresh: runner.forkRefresh,
+
+            forkRefresh: runner.forkRefresh,
 
             layer: tag => Layer.succeed(tag, {
                 latestKey: runner.latestKeyRef,
                 state: runner.stateRef,
-                refresh: runner.forkRefresh,
+                forkRefresh: runner.forkRefresh,
             }),
         }), [runner])
     },
@@ -102,6 +108,7 @@ export const QueryExtension = ReffuseExtension.make(() => ({
 
         return React.useMemo(() => ({
             state: runner.stateRef,
+
             mutate: runner.mutate,
             forkMutate: runner.forkMutate,
 

packages/extension-query/src/QueryService.ts

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

packages/extension-query/src/internal/MutationRunner.ts

@@ -58,20 +58,23 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
         Effect.provide(QueryProgress.QueryProgress.Live),
     )
 
-    const mutate = (...key: K) => run(key).pipe(
-        Effect.provide(QueryState.layer(
+    const mutate = (...key: K) => Effect.provide(run(key), QueryState.layer(
         queryStateTag,
         globalStateRef,
         value => Ref.set(globalStateRef, value),
     ))
-    )
 
-    const forkMutate = (...key: K) => Effect.gen(function*() {
-        const stateRef = yield* Ref.make(AsyncData.noData<A, Exclude<E, HandledE>>())
-        const stateQueue = yield* Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>()
+    const forkMutate = (...key: K) => Effect.all([
+        Ref.make(AsyncData.noData<A, Exclude<E, HandledE>>()),
+        Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>(),
+    ]).pipe(
+        Effect.flatMap(([stateRef, stateQueue]) =>
+            Effect.addFinalizer(() => Queue.shutdown(stateQueue)).pipe(
+                Effect.andThen(run(key)),
+                Effect.scoped,
+                Effect.forkDaemon,
 
-        const fiber = yield* Effect.forkDaemon(run(key).pipe(
-            Effect.tap(() => Queue.shutdown(stateQueue)),
+                Effect.map(fiber => [fiber, Stream.fromQueue(stateQueue)] as const),
 
                 Effect.provide(QueryState.layer(
                     queryStateTag,
@@ -81,11 +84,9 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
                         Effect.andThen(Ref.set(globalStateRef, value)),
                     ),
                 )),
-        ))
-
-        return [fiber, Stream.fromQueue(stateQueue)] as const
-    })
-
+            )
+        )
+    )
 
     return {
         context,

packages/extension-query/src/internal/QueryRunner.ts

@@ -1,6 +1,6 @@
 import { BrowserStream } from "@effect/platform-browser"
 import * as AsyncData from "@typed/async-data"
-import { type Cause, type Context, Effect, Fiber, identity, Option, Ref, type Scope, Stream, SubscriptionRef } from "effect"
+import { type Cause, type Context, Effect, Fiber, identity, Option, Queue, Ref, type Scope, Stream, SubscriptionRef } from "effect"
 import type * as QueryClient from "../QueryClient.js"
 import * as QueryProgress from "../QueryProgress.js"
 import * as QueryState from "./QueryState.js"
@@ -11,11 +11,20 @@ export interface QueryRunner<K extends readonly unknown[], A, E, R> {
 
     readonly latestKeyRef: SubscriptionRef.SubscriptionRef<Option.Option<K>>
     readonly stateRef: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
-    readonly fiberRef: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>>
+    readonly fiberRef: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.RuntimeFiber<
+        AsyncData.Success<A> | AsyncData.Failure<E>,
+        Cause.NoSuchElementException
+    >>>
 
     readonly forkInterrupt: Effect.Effect<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>
-    readonly forkFetch: Effect.Effect<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>
-    readonly forkRefresh: Effect.Effect<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>
+    readonly forkFetch: Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>, Cause.NoSuchElementException>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
+    readonly forkRefresh: Effect.Effect<readonly [
+        fiber: Fiber.RuntimeFiber<AsyncData.Success<A> | AsyncData.Failure<E>, Cause.NoSuchElementException>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
 
     readonly fetchOnKeyChange: Effect.Effect<void, Cause.NoSuchElementException, Scope.Scope>
     readonly refreshOnWindowFocus: Effect.Effect<void>
@@ -43,10 +52,12 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
 
     const latestKeyRef = yield* SubscriptionRef.make(Option.none<K>())
     const stateRef = yield* SubscriptionRef.make(AsyncData.noData<A, Exclude<E, HandledE>>())
-    const fiberRef = yield* SubscriptionRef.make(Option.none<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>())
+    const fiberRef = yield* SubscriptionRef.make(Option.none<Fiber.RuntimeFiber<
+        AsyncData.Success<A> | AsyncData.Failure<Exclude<E, HandledE>>,
+        Cause.NoSuchElementException
+    >>())
 
     const queryStateTag = QueryState.makeTag<A, Exclude<E, HandledE>>()
-    const queryStateLayer = QueryState.layer(queryStateTag, stateRef, value => Ref.set(stateRef, value))
 
     const interrupt = fiberRef.pipe(
         Effect.flatMap(Option.match({
@@ -78,8 +89,12 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
             Effect.flatMap(key => query(key).pipe(
                 errorHandler.handle,
                 Effect.matchCauseEffect({
-                    onSuccess: v => state.set(AsyncData.success(v)),
-                    onFailure: c => state.set(AsyncData.failure(c)),
+                    onSuccess: v => Effect.succeed(AsyncData.success(v)).pipe(
+                        Effect.tap(state.set)
+                    ),
+                    onFailure: c => Effect.succeed(AsyncData.failure(c)).pipe(
+                        Effect.tap(state.set)
+                    ),
                 }),
             )),
         )),
@@ -88,48 +103,64 @@ export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
         Effect.provide(QueryProgress.QueryProgress.Live),
     )
 
-    const forkFetch = queryStateTag.pipe(
+    const forkFetch = Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>().pipe(
+        Effect.flatMap(stateQueue => queryStateTag.pipe(
             Effect.flatMap(state => interrupt.pipe(
-            Effect.andThen(state.set(AsyncData.loading()).pipe(
+                Effect.andThen(Effect.addFinalizer(() => Ref.set(fiberRef, Option.none()).pipe(
+                    Effect.andThen(Queue.shutdown(stateQueue))
+                )).pipe(
+                    Effect.andThen(state.set(AsyncData.loading())),
                     Effect.andThen(run),
-                Effect.fork,
-            )),
-        )),
-
-        Effect.flatMap(fiber =>
-            Ref.set(fiberRef, Option.some(fiber)).pipe(
-                Effect.andThen(Fiber.join(fiber)),
-                Effect.andThen(Ref.set(fiberRef, Option.none())),
-            )
-        ),
-
+                    Effect.scoped,
                     Effect.forkDaemon,
-        Effect.provide(queryStateLayer),
+                )),
+
+                Effect.tap(fiber => Ref.set(fiberRef, Option.some(fiber))),
+                Effect.map(fiber => [fiber, Stream.fromQueue(stateQueue)] as const),
+            )),
+
+            Effect.provide(QueryState.layer(
+                queryStateTag,
+                stateRef,
+                value => Queue.offer(stateQueue, value).pipe(
+                    Effect.andThen(Ref.set(stateRef, value))
+                ),
+            )),
+        ))
     )
 
-    const forkRefresh = queryStateTag.pipe(
-        Effect.flatMap(state => interrupt.pipe(
-            Effect.andThen(state.update(previous => {
+    const setInitialRefreshState = queryStateTag.pipe(
+        Effect.flatMap(state => state.update(previous => {
             if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous))
                 return AsyncData.refreshing(previous)
             if (AsyncData.isRefreshing(previous))
                 return AsyncData.refreshing(previous.previous)
             return AsyncData.loading()
-            }).pipe(
+        }))
+    )
+
+    const forkRefresh = Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>().pipe(
+        Effect.flatMap(stateQueue => interrupt.pipe(
+            Effect.andThen(Effect.addFinalizer(() => Ref.set(fiberRef, Option.none()).pipe(
+                Effect.andThen(Queue.shutdown(stateQueue))
+            )).pipe(
+                Effect.andThen(setInitialRefreshState),
                 Effect.andThen(run),
-                Effect.fork,
-            ))
+                Effect.scoped,
+                Effect.forkDaemon,
             )),
 
-        Effect.flatMap(fiber =>
-            Ref.set(fiberRef, Option.some(fiber)).pipe(
-                Effect.andThen(Fiber.join(fiber)),
-                Effect.andThen(Ref.set(fiberRef, Option.none())),
-            )
-        ),
+            Effect.tap(fiber => Ref.set(fiberRef, Option.some(fiber))),
+            Effect.map(fiber => [fiber, Stream.fromQueue(stateQueue)] as const),
 
-        Effect.forkDaemon,
-        Effect.provide(queryStateLayer),
+            Effect.provide(QueryState.layer(
+                queryStateTag,
+                stateRef,
+                value => Queue.offer(stateQueue, value).pipe(
+                    Effect.andThen(Ref.set(stateRef, value))
+                ),
+            )),
+        ))
     )
 
     const fetchOnKeyChange = Effect.addFinalizer(() => interrupt).pipe(

packages/reffuse/src/ReffuseHelpers.ts

@@ -396,7 +396,7 @@ export abstract class ReffuseHelpers<R> {
         const [reactStateValue, setReactStateValue] = React.useState(initialState)
 
         this.useFork(() => Stream.runForEach(
-            Stream.changes(ref.changes),
+            Stream.changesWith(ref.changes, (x, y) => x === y),
             v => Effect.sync(() => setReactStateValue(v)),
         ), [ref])