Mutation refactoring

packages/example/src/QueryErrorHandler.tsx

@@ -0,0 +1,57 @@
+import { AlertDialog, Button, Flex, Text } from "@radix-ui/themes"
+import { ErrorHandler } from "@reffuse/extension-query"
+import { Cause, Console, Context, Effect, Either, flow, Match, Option, Stream } from "effect"
+import { useState } from "react"
+import { AppQueryErrorHandler } from "./query"
+import { R } from "./reffuse"
+
+
+export function VQueryErrorHandler() {
+    const [failure, setFailure] = useState(Option.none<Cause.Cause<
+        ErrorHandler.Error<Context.Tag.Service<AppQueryErrorHandler>>
+    >>())
+
+    R.useFork(() => AppQueryErrorHandler.pipe(Effect.flatMap(handler =>
+        Stream.runForEach(handler.errors, v => Console.error(v).pipe(
+            Effect.andThen(Effect.sync(() => { setFailure(Option.some(v)) }))
+        ))
+    )), [])
+
+    return Option.match(failure, {
+        onSome: v => (
+            <AlertDialog.Root open>
+                <AlertDialog.Content maxWidth="450px">
+                    <AlertDialog.Title>Error</AlertDialog.Title>
+                    <AlertDialog.Description size="2">
+                        {Either.match(Cause.failureOrCause(v), {
+                            onLeft: flow(
+                                Match.value,
+                                Match.tag("RequestError", () => <Text>HTTP request error</Text>),
+                                Match.tag("ResponseError", () => <Text>HTTP response error</Text>),
+                                Match.exhaustive,
+                            ),
+
+                            onRight: flow(
+                                Cause.dieOption,
+                                Option.match({
+                                    onSome: () => <Text>Unrecoverable defect</Text>,
+                                    onNone: () => <Text>Unknown error</Text>,
+                                }),
+                            ),
+                        })}
+                    </AlertDialog.Description>
+
+                    <Flex gap="3" mt="4" justify="end">
+                        <AlertDialog.Action>
+                            <Button variant="solid" color="red" onClick={() => setFailure(Option.none())}>
+                                Ok
+                            </Button>
+                        </AlertDialog.Action>
+                    </Flex>
+                </AlertDialog.Content>
+            </AlertDialog.Root>
+        ),
+
+        onNone: () => <></>,
+    })
+}

packages/example/src/main.tsx

@@ -5,6 +5,7 @@ import { Layer } from "effect"
 import { StrictMode } from "react"
 import { createRoot } from "react-dom/client"
 import { ReffuseRuntime } from "reffuse"
+import { AppQueryClient, AppQueryErrorHandler } from "./query"
 import { GlobalContext } from "./reffuse"
 import { routeTree } from "./routeTree.gen"
 
@@ -14,6 +15,8 @@ const layer = Layer.empty.pipe(
     Layer.provideMerge(Geolocation.layer),
     Layer.provideMerge(Permissions.layer),
     Layer.provideMerge(FetchHttpClient.layer),
+    Layer.provideMerge(AppQueryClient.Live),
+    Layer.provideMerge(AppQueryErrorHandler.Live),
 )
 
 const router = createRouter({ routeTree })

packages/example/src/query.ts

@@ -0,0 +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,
+    HttpClientError.HttpClientError
+>()(
+    "AppQueryErrorHandler",
+
+    (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/services/Uuid4Query.ts

@@ -1,4 +1,3 @@
-import { HttpClientError } from "@effect/platform"
 import { QueryService } from "@reffuse/extension-query"
 import { ParseResult, Schema } from "effect"
 
@@ -8,5 +7,5 @@ export const Result = Schema.Array(Schema.String)
 export class Uuid4Query extends QueryService.Tag("Uuid4Query")<Uuid4Query,
     readonly ["uuid4", number],
     typeof Result.Type,
-    HttpClientError.HttpClientError | ParseResult.ParseError
+    ParseResult.ParseError
 >() {}

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/reffuse.ts

@@ -3,6 +3,7 @@ import { Clipboard, Geolocation, Permissions } from "@effect/platform-browser"
 import { LazyRefExtension } from "@reffuse/extension-lazyref"
 import { QueryExtension } from "@reffuse/extension-query"
 import { Reffuse, ReffuseContext } from "reffuse"
+import { AppQueryClient, AppQueryErrorHandler } from "./query"
 
 
 export const GlobalContext = ReffuseContext.make<
@@ -10,6 +11,8 @@ export const GlobalContext = ReffuseContext.make<
     | Geolocation.Geolocation
     | Permissions.Permissions
     | HttpClient.HttpClient
+    | AppQueryClient
+    | AppQueryErrorHandler
 >()
 
 export class GlobalReffuse extends Reffuse.Reffuse.pipe(

packages/example/src/routeTree.gen.ts

@@ -19,6 +19,7 @@ import { Route as CountImport } from './routes/count'
 import { Route as BlankImport } from './routes/blank'
 import { Route as IndexImport } from './routes/index'
 import { Route as QueryUsequeryImport } from './routes/query/usequery'
+import { Route as QueryUsemutationImport } from './routes/query/usemutation'
 import { Route as QueryServiceImport } from './routes/query/service'
 
 // Create/Update Routes
@@ -71,6 +72,12 @@ const QueryUsequeryRoute = QueryUsequeryImport.update({
   getParentRoute: () => rootRoute,
 } as any)
 
+const QueryUsemutationRoute = QueryUsemutationImport.update({
+  id: '/query/usemutation',
+  path: '/query/usemutation',
+  getParentRoute: () => rootRoute,
+} as any)
+
 const QueryServiceRoute = QueryServiceImport.update({
   id: '/query/service',
   path: '/query/service',
@@ -137,6 +144,13 @@ declare module '@tanstack/react-router' {
       preLoaderRoute: typeof QueryServiceImport
       parentRoute: typeof rootRoute
     }
+    '/query/usemutation': {
+      id: '/query/usemutation'
+      path: '/query/usemutation'
+      fullPath: '/query/usemutation'
+      preLoaderRoute: typeof QueryUsemutationImport
+      parentRoute: typeof rootRoute
+    }
     '/query/usequery': {
       id: '/query/usequery'
       path: '/query/usequery'
@@ -158,6 +172,7 @@ export interface FileRoutesByFullPath {
   '/tests': typeof TestsRoute
   '/time': typeof TimeRoute
   '/query/service': typeof QueryServiceRoute
+  '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
 }
 
@@ -170,6 +185,7 @@ export interface FileRoutesByTo {
   '/tests': typeof TestsRoute
   '/time': typeof TimeRoute
   '/query/service': typeof QueryServiceRoute
+  '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
 }
 
@@ -183,6 +199,7 @@ export interface FileRoutesById {
   '/tests': typeof TestsRoute
   '/time': typeof TimeRoute
   '/query/service': typeof QueryServiceRoute
+  '/query/usemutation': typeof QueryUsemutationRoute
   '/query/usequery': typeof QueryUsequeryRoute
 }
 
@@ -197,6 +214,7 @@ export interface FileRouteTypes {
     | '/tests'
     | '/time'
     | '/query/service'
+    | '/query/usemutation'
     | '/query/usequery'
   fileRoutesByTo: FileRoutesByTo
   to:
@@ -208,6 +226,7 @@ export interface FileRouteTypes {
     | '/tests'
     | '/time'
     | '/query/service'
+    | '/query/usemutation'
     | '/query/usequery'
   id:
     | '__root__'
@@ -219,6 +238,7 @@ export interface FileRouteTypes {
     | '/tests'
     | '/time'
     | '/query/service'
+    | '/query/usemutation'
     | '/query/usequery'
   fileRoutesById: FileRoutesById
 }
@@ -232,6 +252,7 @@ export interface RootRouteChildren {
   TestsRoute: typeof TestsRoute
   TimeRoute: typeof TimeRoute
   QueryServiceRoute: typeof QueryServiceRoute
+  QueryUsemutationRoute: typeof QueryUsemutationRoute
   QueryUsequeryRoute: typeof QueryUsequeryRoute
 }
 
@@ -244,6 +265,7 @@ const rootRouteChildren: RootRouteChildren = {
   TestsRoute: TestsRoute,
   TimeRoute: TimeRoute,
   QueryServiceRoute: QueryServiceRoute,
+  QueryUsemutationRoute: QueryUsemutationRoute,
   QueryUsequeryRoute: QueryUsequeryRoute,
 }
 
@@ -265,6 +287,7 @@ export const routeTree = rootRoute
         "/tests",
         "/time",
         "/query/service",
+        "/query/usemutation",
         "/query/usequery"
       ]
     },
@@ -292,6 +315,9 @@ export const routeTree = rootRoute
     "/query/service": {
       "filePath": "query/service.tsx"
     },
+    "/query/usemutation": {
+      "filePath": "query/usemutation.tsx"
+    },
     "/query/usequery": {
       "filePath": "query/usequery.tsx"
     }

packages/example/src/routes/__root.tsx

@@ -1,3 +1,4 @@
+import { VQueryErrorHandler } from "@/QueryErrorHandler"
 import { Container, Flex, Theme } from "@radix-ui/themes"
 import { createRootRoute, Link, Outlet } from "@tanstack/react-router"
 import { TanStackRouterDevtools } from "@tanstack/router-devtools"
@@ -26,6 +27,8 @@ function Root() {
             </Container>
 
             <Outlet />
+
+            <VQueryErrorHandler />
             <TanStackRouterDevtools />
         </Theme>
     )

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

@@ -0,0 +1,81 @@
+import { R } from "@/reffuse"
+import { HttpClient } from "@effect/platform"
+import { Button, Container, Flex, Slider, Text } from "@radix-ui/themes"
+import { QueryProgress } from "@reffuse/extension-query"
+import { createFileRoute } from "@tanstack/react-router"
+import * as AsyncData from "@typed/async-data"
+import { Array, Console, Effect, flow, Option, Schema, Stream } from "effect"
+import { useState } from "react"
+
+
+export const Route = createFileRoute("/query/usemutation")({
+    component: RouteComponent
+})
+
+
+const Result = Schema.Array(Schema.String)
+
+function RouteComponent() {
+    const runFork = R.useRunFork()
+
+    const [count, setCount] = useState(1)
+
+    const mutation = R.useMutation({
+        mutation: ([count]: readonly [count: number]) => Console.log(`Querying ${ count } IDs...`).pipe(
+            Effect.andThen(QueryProgress.QueryProgress.update(() =>
+                AsyncData.Progress.make({ loaded: 0, total: Option.some(100) })
+            )),
+            Effect.andThen(Effect.sleep("500 millis")),
+            Effect.tap(() => QueryProgress.QueryProgress.update(() =>
+                AsyncData.Progress.make({ loaded: 50, total: Option.some(100) })
+            )),
+            Effect.andThen(HttpClient.get(`https://www.uuidtools.com/api/generate/v4/count/${ count }`)),
+            HttpClient.withTracerPropagation(false),
+            Effect.flatMap(res => res.json),
+            Effect.flatMap(Schema.decodeUnknown(Result)),
+            Effect.scoped,
+        )
+    })
+
+    const [state] = R.useRefState(mutation.state)
+
+
+    return (
+        <Container>
+            <Flex direction="column" align="center" gap="2">
+                <Slider
+                    min={1}
+                    max={100}
+                    value={[count]}
+                    onValueChange={flow(
+                        Array.head,
+                        Option.getOrThrow,
+                        setCount,
+                    )}
+                />
+
+                <Text>
+                    {AsyncData.match(state, {
+                        NoData: () => "No data yet",
+                        Loading: progress =>
+                            `Loading...
+                            ${ Option.match(progress, {
+                                onSome: ({ loaded, total }) => ` (${ loaded }/${ Option.getOrElse(total, () => "unknown") })`,
+                                onNone: () => "",
+                            }) }`,
+                        Success: value => `Value: ${ value }`,
+                        Failure: cause => `Error: ${ cause }`,
+                    })}
+                </Text>
+
+                <Button onClick={() => mutation.forkMutate(count).pipe(
+                    Effect.flatMap(([, state]) => Stream.runForEach(state, Console.log)),
+                    Effect.andThen(Console.log("Mutation done.")),
+                    runFork,
+                )}>
+                    Get
+                </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

@@ -12,9 +12,10 @@ export const LazyRefExtension = ReffuseExtension.make(() => ({
         const initialState = this.useMemo(() => ref, [], { doNotReExecuteOnRuntimeOrContextChange: true })
         const [reactStateValue, setReactStateValue] = React.useState(initialState)
 
-        this.useFork(() => Stream.runForEach(ref.changes, v => Effect.sync(() =>
+        this.useFork(() => Stream.runForEach(
-            setReactStateValue(v)
+            Stream.changesWith(ref.changes, (x, y) => x === y),
-        )), [ref])
+            v => Effect.sync(() => setReactStateValue(v)),
+        ), [ref])
 
         const setValue = this.useCallbackSync((setStateAction: React.SetStateAction<A>) =>
             LazyRef.update(ref, prevState =>

packages/extension-query/src/ErrorHandler.ts

@@ -0,0 +1,54 @@
+import { Cause, Context, Effect, identity, Layer, Queue, Stream } from "effect"
+import type { Mutable } from "effect/Types"
+
+
+export interface ErrorHandler<E> {
+    readonly errors: Stream.Stream<Cause.Cause<E>>
+    readonly handle: <A, SelfE, R>(self: Effect.Effect<A, SelfE, R>) => Effect.Effect<A, Exclude<SelfE, E>, R>
+}
+
+export type Error<T> = T extends ErrorHandler<infer E> ? E : never
+
+
+export interface ServiceResult<Self, Id extends string, E> extends Context.TagClass<Self, Id, ErrorHandler<E>> {
+    readonly Live: Layer.Layer<Self>
+}
+
+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.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<DefaultErrorHandler>()(
+    "@reffuse/extension-query/DefaultErrorHandler",
+    identity,
+) {}

packages/extension-query/src/MutationService.ts

@@ -0,0 +1,16 @@
+import type * as AsyncData from "@typed/async-data"
+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<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) => <
+    Self, K extends readonly unknown[], A, E = never,
+>() => Effect.Tag(id)<Self, MutationService<K, A, E>>()

packages/extension-query/src/QueryClient.ts

@@ -0,0 +1,39 @@
+import { Context, Layer } from "effect"
+import type { Mutable } from "effect/Types"
+import * as ErrorHandler from "./ErrorHandler.js"
+
+
+export interface QueryClient<EH, HandledE> {
+    readonly ErrorHandler: Context.Tag<EH, ErrorHandler.ErrorHandler<HandledE>>
+}
+
+
+const id = "@reffuse/extension-query/QueryClient"
+
+export type TagClassShape<EH, HandledE> = Context.TagClassShape<typeof id, QueryClient<EH, HandledE>>
+export type GenericTagClass<EH, HandledE> = Context.TagClass<TagClassShape<EH, HandledE>, typeof id, QueryClient<EH, HandledE>>
+export const makeGenericTagClass = <EH = never, HandledE = never>(): GenericTagClass<EH, HandledE> => Context.Tag(id)()
+
+
+export interface ServiceProps<EH, HandledE> {
+    readonly ErrorHandler?: Context.Tag<EH, ErrorHandler.ErrorHandler<HandledE>>
+}
+
+export interface ServiceResult<Self, EH, HandledE> extends Context.TagClass<Self, typeof id, QueryClient<EH, HandledE>> {
+    readonly Live: Layer.Layer<Self>
+}
+
+export const Service = <Self>() => (
+    <
+        EH = ErrorHandler.DefaultErrorHandler,
+        HandledE = ErrorHandler.Error<Context.Tag.Service<ErrorHandler.DefaultErrorHandler>>,
+    >(
+        props?: ServiceProps<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>>
+        })
+        return TagClass
+    }
+)

packages/extension-query/src/QueryExtension.ts

@@ -2,20 +2,27 @@ import type * as AsyncData from "@typed/async-data"
 import { type Cause, type Context, Effect, type Fiber, Layer, type Option, type Stream, type SubscriptionRef } from "effect"
 import * as React from "react"
 import { ReffuseExtension, type ReffuseHelpers } from "reffuse"
-import * as QueryRunner from "./QueryRunner.js"
+import type * as MutationService from "./MutationService.js"
+import * as QueryClient from "./QueryClient.js"
+import type * as QueryProgress from "./QueryProgress.js"
 import type * as QueryService from "./QueryService.js"
+import { MutationRunner, QueryRunner } from "./internal/index.js"
 
 
 export interface UseQueryProps<K extends readonly unknown[], A, E, R> {
     readonly key: Stream.Stream<K>
-    readonly query: (key: K) => Effect.Effect<A, E, R>
+    readonly query: (key: K) => Effect.Effect<A, E, R | QueryProgress.QueryProgress>
     readonly refreshOnWindowFocus?: boolean
 }
 
 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>>
@@ -23,12 +30,40 @@ export interface UseQueryResult<K extends readonly unknown[], A, E> {
 }
 
 
+export interface UseMutationProps<K extends readonly unknown[], A, E, R> {
+    readonly mutation: (key: K) => Effect.Effect<A, E, R | QueryProgress.QueryProgress>
+}
+
+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>>,
+        state: Stream.Stream<AsyncData.AsyncData<A, E>>,
+    ]>
+
+    readonly layer: <Self, Id extends string>(
+        tag: Context.TagClass<Self, Id, MutationService.MutationService<K, A, E>>
+    ) => Layer.Layer<Self>
+}
+
+
 export const QueryExtension = ReffuseExtension.make(() => ({
-    useQuery<K extends readonly unknown[], A, E, R>(
+    useQuery<
-        this: ReffuseHelpers.ReffuseHelpers<R>,
+        EH,
-        props: UseQueryProps<K, A, E, R>,
+        QK extends readonly unknown[],
-    ): UseQueryResult<K, A, E> {
+        QA,
+        QE,
+        HandledE,
+        QR extends R,
+        R,
+    >(
+        this: ReffuseHelpers.ReffuseHelpers<R | QueryClient.TagClassShape<EH, HandledE> | EH>,
+        props: UseQueryProps<QK, QA, QE, QR>,
+    ): UseQueryResult<QK, QA, Exclude<QE, HandledE>> {
         const runner = this.useMemo(() => QueryRunner.make({
+            QueryClient: QueryClient.makeGenericTagClass<EH, HandledE>(),
             key: props.key,
             query: props.query,
         }), [props.key])
@@ -43,13 +78,45 @@ 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])
-    }
+    },
+
+    useMutation<
+        EH,
+        QK extends readonly unknown[],
+        QA,
+        QE,
+        HandledE,
+        QR extends R,
+        R,
+    >(
+        this: ReffuseHelpers.ReffuseHelpers<R | QueryClient.TagClassShape<EH, HandledE> | EH>,
+        props: UseMutationProps<QK, QA, QE, QR>,
+    ): UseMutationResult<QK, QA, Exclude<QE, HandledE>> {
+        const runner = this.useMemo(() => MutationRunner.make({
+            QueryClient: QueryClient.makeGenericTagClass<EH, HandledE>(),
+            mutation: props.mutation,
+        }), [])
+
+        return React.useMemo(() => ({
+            state: runner.stateRef,
+
+            mutate: runner.mutate,
+            forkMutate: runner.forkMutate,
+
+            layer: tag => Layer.succeed(tag, {
+                state: runner.stateRef,
+                mutate: runner.mutate,
+                forkMutate: runner.forkMutate,
+            }),
+        }), [runner])
+    },
 }))

packages/extension-query/src/QueryProgress.ts

@@ -0,0 +1,37 @@
+import * as AsyncData from "@typed/async-data"
+import { Effect, flow, Layer, Match, Option } from "effect"
+import { QueryState } from "./internal/index.js"
+
+
+export class QueryProgress extends Effect.Tag("@reffuse/extension-query/QueryProgress")<QueryProgress, {
+    readonly get: Effect.Effect<Option.Option<AsyncData.Progress>>
+
+    readonly update: (
+        f: (previous: Option.Option<AsyncData.Progress>) => AsyncData.Progress
+    ) => Effect.Effect<void>
+}>() {
+    static readonly Live: Layer.Layer<
+        QueryProgress,
+        never,
+        QueryState.QueryState<any, any>
+    > = Layer.effect(this, Effect.gen(function*() {
+        const state = yield* QueryState.makeTag()
+
+        const get = state.get.pipe(
+            Effect.map(flow(Match.value,
+                Match.tag("Loading", v => v.progress),
+                Match.tag("Refreshing", v => v.progress),
+                Match.orElse(() => Option.none()),
+            ))
+        )
+
+        const update = (f: (previous: Option.Option<AsyncData.Progress>) => AsyncData.Progress) => get.pipe(
+            Effect.map(f),
+            Effect.flatMap(progress => state.update(previous =>
+                AsyncData.updateProgress(previous, progress)
+            )),
+        )
+
+        return { get, update }
+    }))
+}

packages/extension-query/src/QueryRunner.ts

@@ -1,144 +0,0 @@
-import { BrowserStream } from "@effect/platform-browser"
-import * as AsyncData from "@typed/async-data"
-import { type Cause, Effect, Fiber, identity, Option, Ref, type Scope, Stream, SubscriptionRef } from "effect"
-
-
-export interface QueryRunner<K extends readonly unknown[], A, E, R> {
-    readonly query: (key: K) => Effect.Effect<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 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 fetchOnKeyChange: Effect.Effect<void, Cause.NoSuchElementException, Scope.Scope>
-    readonly refreshOnWindowFocus: Effect.Effect<void>
-}
-
-
-export interface MakeProps<K extends readonly unknown[], A, E, R> {
-    readonly key: Stream.Stream<K>
-    readonly query: (key: K) => Effect.Effect<A, E, R>
-}
-
-export const make = <K extends readonly unknown[], A, E, R>(
-    { key, query }: MakeProps<K, A, E, R>
-): Effect.Effect<QueryRunner<K, A, E, R>, never, R> => Effect.gen(function*() {
-    const context = yield* Effect.context<R>()
-
-    const latestKeyRef = yield* SubscriptionRef.make(Option.none<K>())
-    const stateRef = yield* SubscriptionRef.make(AsyncData.noData<A, E>())
-    const fiberRef = yield* SubscriptionRef.make(Option.none<Fiber.RuntimeFiber<void, Cause.NoSuchElementException>>())
-
-    const interrupt = fiberRef.pipe(
-        Effect.flatMap(Option.match({
-            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
-                Effect.andThen(Fiber.interrupt(fiber))
-            ),
-            onNone: () => Effect.void,
-        }))
-    )
-
-    const forkInterrupt = fiberRef.pipe(
-        Effect.flatMap(Option.match({
-            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
-                Effect.andThen(Fiber.interrupt(fiber).pipe(
-                    Effect.asVoid,
-                    Effect.forkDaemon,
-                ))
-            ),
-            onNone: () => Effect.forkDaemon(Effect.void),
-        }))
-    )
-
-    const forkFetch = interrupt.pipe(
-        Effect.andThen(
-            Ref.set(stateRef, AsyncData.loading()).pipe(
-                Effect.andThen(latestKeyRef),
-                Effect.flatMap(identity),
-                Effect.flatMap(key => query(key).pipe(
-                    Effect.matchCauseEffect({
-                        onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
-                        onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-                    })
-                )),
-
-                Effect.provide(context),
-                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.forkDaemon,
-    )
-
-    const forkRefresh = interrupt.pipe(
-        Effect.andThen(
-            Ref.update(stateRef, 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(
-                Effect.andThen(latestKeyRef),
-                Effect.flatMap(identity),
-                Effect.flatMap(key => query(key).pipe(
-                    Effect.matchCauseEffect({
-                        onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
-                        onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-                    })
-                )),
-
-                Effect.provide(context),
-                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.forkDaemon,
-    )
-
-    const fetchOnKeyChange = Effect.addFinalizer(() => interrupt).pipe(
-        Effect.andThen(Stream.runForEach(key, latestKey =>
-            Ref.set(latestKeyRef, Option.some(latestKey)).pipe(
-                Effect.andThen(forkFetch)
-            )
-        ))
-    )
-
-    const refreshOnWindowFocus = Stream.runForEach(
-        BrowserStream.fromEventListenerWindow("focus"),
-        () => forkRefresh,
-    )
-
-    return {
-        query,
-
-        latestKeyRef,
-        stateRef,
-        fiberRef,
-
-        forkInterrupt,
-        forkFetch,
-        forkRefresh,
-
-        fetchOnKeyChange,
-        refreshOnWindowFocus,
-    }
-})

packages/extension-query/src/QueryService.ts

@@ -1,32 +1,16 @@
 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) => <
     Self, K extends readonly unknown[], A, E = never,
 >() => Effect.Tag(id)<Self, QueryService<K, A, E>>()
-
-
-// export interface LayerProps<A, E, R> {
-//     readonly query: Effect.Effect<A, E, R>
-// }
-
-// export const layer = <Self, Id extends string, A, E, R>(
-//     tag: Context.TagClass<Self, Id, QueryService<A, E>>,
-//     props: LayerProps<A, E, R>,
-// ): Layer.Layer<Self, never, R> => Layer.effect(tag, Effect.gen(function*() {
-//     const runner = yield* QueryRunner.make({
-//         query: props.query
-//     })
-
-//     return {
-//         state: runner.stateRef,
-//         refresh: runner.forkRefresh,
-//     }
-// }))

packages/extension-query/src/index.ts

@@ -1,3 +1,6 @@
+export * as ErrorHandler from "./ErrorHandler.js"
+export * as MutationService from "./MutationService.js"
+export * as QueryClient from "./QueryClient.js"
 export * from "./QueryExtension.js"
-export * as QueryRunner from "./QueryRunner.js"
+export * as QueryProgress from "./QueryProgress.js"
 export * as QueryService from "./QueryService.js"

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

@@ -0,0 +1,98 @@
+import * as AsyncData from "@typed/async-data"
+import { type Context, Effect, type Fiber, Queue, Ref, Stream, SubscriptionRef } from "effect"
+import type * as QueryClient from "../QueryClient.js"
+import * as QueryProgress from "../QueryProgress.js"
+import * as QueryState from "./QueryState.js"
+
+
+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<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 interface MakeProps<EH, K extends readonly unknown[], A, E, HandledE, R> {
+    readonly QueryClient: QueryClient.GenericTagClass<EH, HandledE>
+    readonly mutation: (key: K) => Effect.Effect<A, E, R | QueryProgress.QueryProgress>
+}
+
+export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
+    {
+        QueryClient,
+        mutation,
+    }: MakeProps<EH, K, A, E, HandledE, R>
+): Effect.Effect<
+    MutationRunner<K, A, Exclude<E, HandledE>, R>,
+    never,
+    R | QueryClient.TagClassShape<EH, HandledE> | EH
+> => Effect.gen(function*() {
+    const context = yield* Effect.context<R | QueryClient.TagClassShape<EH, HandledE> | EH>()
+    const globalStateRef = yield* SubscriptionRef.make(AsyncData.noData<A, Exclude<E, HandledE>>())
+
+    const queryStateTag = QueryState.makeTag<A, Exclude<E, HandledE>>()
+
+    const run = (key: K) => Effect.all([
+        queryStateTag,
+        QueryClient.pipe(Effect.flatMap(client => client.ErrorHandler)),
+    ]).pipe(
+        Effect.flatMap(([state, errorHandler]) => state.set(AsyncData.loading()).pipe(
+            Effect.andThen(mutation(key)),
+            errorHandler.handle,
+            Effect.matchCauseEffect({
+                onSuccess: v => Effect.succeed(AsyncData.success(v)).pipe(
+                    Effect.tap(state.set)
+                ),
+                onFailure: c => Effect.succeed(AsyncData.failure(c)).pipe(
+                    Effect.tap(state.set)
+                ),
+            }),
+        )),
+
+        Effect.provide(context),
+        Effect.provide(QueryProgress.QueryProgress.Live),
+    )
+
+    const mutate = (...key: K) => Effect.provide(run(key), QueryState.layer(
+        queryStateTag,
+        globalStateRef,
+        value => Ref.set(globalStateRef, value),
+    ))
+
+    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,
+
+                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)),
+                        Effect.andThen(Ref.set(globalStateRef, value)),
+                    ),
+                )),
+            )
+        )
+    )
+
+    return {
+        context,
+        stateRef: globalStateRef,
+
+        mutate,
+        forkMutate,
+    }
+})

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

@@ -0,0 +1,193 @@
+import { BrowserStream } from "@effect/platform-browser"
+import * as AsyncData from "@typed/async-data"
+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"
+
+
+export interface QueryRunner<K extends readonly unknown[], A, E, R> {
+    readonly context: Context.Context<R>
+
+    readonly latestKeyRef: SubscriptionRef.SubscriptionRef<Option.Option<K>>
+    readonly stateRef: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
+    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<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>
+}
+
+
+export interface MakeProps<EH, K extends readonly unknown[], A, E, HandledE, R> {
+    readonly QueryClient: QueryClient.GenericTagClass<EH, HandledE>
+    readonly key: Stream.Stream<K>
+    readonly query: (key: K) => Effect.Effect<A, E, R | QueryProgress.QueryProgress>
+}
+
+export const make = <EH, K extends readonly unknown[], A, E, HandledE, R>(
+    {
+        QueryClient,
+        key,
+        query,
+    }: MakeProps<EH, K, A, E, HandledE, R>
+): Effect.Effect<
+    QueryRunner<K, A, Exclude<E, HandledE>, R>,
+    never,
+    R | QueryClient.TagClassShape<EH, HandledE> | EH
+> => Effect.gen(function*() {
+    const context = yield* Effect.context<R | QueryClient.TagClassShape<EH, HandledE> | EH>()
+
+    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<
+        AsyncData.Success<A> | AsyncData.Failure<Exclude<E, HandledE>>,
+        Cause.NoSuchElementException
+    >>())
+
+    const queryStateTag = QueryState.makeTag<A, Exclude<E, HandledE>>()
+
+    const interrupt = fiberRef.pipe(
+        Effect.flatMap(Option.match({
+            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
+                Effect.andThen(Fiber.interrupt(fiber))
+            ),
+            onNone: () => Effect.void,
+        }))
+    )
+
+    const forkInterrupt = fiberRef.pipe(
+        Effect.flatMap(Option.match({
+            onSome: fiber => Ref.set(fiberRef, Option.none()).pipe(
+                Effect.andThen(Fiber.interrupt(fiber).pipe(
+                    Effect.asVoid,
+                    Effect.forkDaemon,
+                ))
+            ),
+            onNone: () => Effect.forkDaemon(Effect.void),
+        }))
+    )
+
+    const run = Effect.all([
+        queryStateTag,
+        QueryClient.pipe(Effect.flatMap(client => client.ErrorHandler)),
+    ]).pipe(
+        Effect.flatMap(([state, errorHandler]) => latestKeyRef.pipe(
+            Effect.flatMap(identity),
+            Effect.flatMap(key => query(key).pipe(
+                errorHandler.handle,
+                Effect.matchCauseEffect({
+                    onSuccess: v => Effect.succeed(AsyncData.success(v)).pipe(
+                        Effect.tap(state.set)
+                    ),
+                    onFailure: c => Effect.succeed(AsyncData.failure(c)).pipe(
+                        Effect.tap(state.set)
+                    ),
+                }),
+            )),
+        )),
+
+        Effect.provide(context),
+        Effect.provide(QueryProgress.QueryProgress.Live),
+    )
+
+    const forkFetch = Queue.unbounded<AsyncData.AsyncData<A, Exclude<E, HandledE>>>().pipe(
+        Effect.flatMap(stateQueue => queryStateTag.pipe(
+            Effect.flatMap(state => interrupt.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.scoped,
+                    Effect.forkDaemon,
+                )),
+
+                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 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()
+        }))
+    )
+
+    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.scoped,
+                Effect.forkDaemon,
+            )),
+
+            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 fetchOnKeyChange = Effect.addFinalizer(() => interrupt).pipe(
+        Effect.andThen(Stream.runForEach(Stream.changes(key), latestKey =>
+            Ref.set(latestKeyRef, Option.some(latestKey)).pipe(
+                Effect.andThen(forkFetch)
+            )
+        ))
+    )
+
+    const refreshOnWindowFocus = Stream.runForEach(
+        BrowserStream.fromEventListenerWindow("focus"),
+        () => forkRefresh,
+    )
+
+    return {
+        context,
+
+        latestKeyRef,
+        stateRef,
+        fiberRef,
+
+        forkInterrupt,
+        forkFetch,
+        forkRefresh,
+
+        fetchOnKeyChange,
+        refreshOnWindowFocus,
+    }
+})

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

@@ -0,0 +1,24 @@
+import type * as AsyncData from "@typed/async-data"
+import { Context, Effect, Layer } from "effect"
+
+
+export interface QueryState<A, E> {
+    readonly get: Effect.Effect<AsyncData.AsyncData<A, E>>
+    readonly set: (value: AsyncData.AsyncData<A, E>) => Effect.Effect<void>
+    readonly update: (f: (previous: AsyncData.AsyncData<A, E>) => AsyncData.AsyncData<A, E>) => Effect.Effect<void>
+}
+
+export const makeTag = <A, E>(): Context.Tag<QueryState<A, E>, QueryState<A, E>> => Context.GenericTag("@reffuse/query-extension/QueryState")
+
+export const layer = <A, E>(
+    tag: Context.Tag<QueryState<A, E>, QueryState<A, E>>,
+    get: Effect.Effect<AsyncData.AsyncData<A, E>>,
+    set: (value: AsyncData.AsyncData<A, E>) => Effect.Effect<void>,
+): Layer.Layer<QueryState<A, E>> => Layer.succeed(tag, {
+    get,
+    set,
+    update: f => get.pipe(
+        Effect.map(f),
+        Effect.flatMap(set),
+    ),
+})

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

@@ -0,0 +1,3 @@
+export * as MutationRunner from "./MutationRunner.js"
+export * as QueryRunner from "./QueryRunner.js"
+export * as QueryState from "./QueryState.js"

packages/reffuse/src/ReffuseHelpers.ts

@@ -395,9 +395,10 @@ export abstract class ReffuseHelpers<R> {
         const initialState = this.useMemo(() => ref, [], { doNotReExecuteOnRuntimeOrContextChange: true })
         const [reactStateValue, setReactStateValue] = React.useState(initialState)
 
-        this.useFork(() => Stream.runForEach(ref.changes, v => Effect.sync(() =>
+        this.useFork(() => Stream.runForEach(
-            setReactStateValue(v)
+            Stream.changesWith(ref.changes, (x, y) => x === y),
-        )), [ref])
+            v => Effect.sync(() => setReactStateValue(v)),
+        ), [ref])
 
         const setValue = this.useCallbackSync((setStateAction: React.SetStateAction<A>) =>
             Ref.update(ref, prevState =>