Fix

packages/extension-query/src/FailureHandler.ts

@@ -0,0 +1,20 @@
+import { type Context, Effect, Layer, Queue, Stream } from "effect"
+
+
+export interface FailureHandler<E> {
+    readonly failures: Stream.Stream<E>
+    readonly queue: Queue.Queue<E>
+}
+
+export const Tag = <const Id extends string>(id: Id) => <
+    Self, E = never,
+>() => Effect.Tag(id)<Self, FailureHandler<E>>()
+
+export const layer = <Self, Id extends string, E>(
+    tag: Context.TagClass<Self, Id, FailureHandler<E>>
+): Layer.Layer<Self> => Layer.effect(tag, Queue.unbounded<E>().pipe(
+    Effect.map(queue => ({
+        failures: Stream.fromQueue(queue),
+        queue,
+    }))
+))

packages/extension-query/src/MutationRunner.ts

@@ -0,0 +1,134 @@
+// 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 MutationRunner<K extends readonly unknown[], A, E, R> {
+//     readonly mutation: (...args: K) => Effect.Effect<A, E, R>
+
+//     readonly stateRef: SubscriptionRef.SubscriptionRef<AsyncData.AsyncData<A, E>>
+
+//     readonly forkMutate: Effect.Effect<Fiber.RuntimeFiber<void>>
+// }
+
+
+// export interface MakeProps<K extends readonly unknown[], A, E, R> {
+//     readonly mutation: (...args: 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<MutationRunner<K, A, E, R>, never, R> => Effect.gen(function*() {
+//     const context = yield* Effect.context<R>()
+
+//     const stateRef = yield* SubscriptionRef.make(AsyncData.noData<A, E>())
+
+//     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/QueryExtension.ts

@@ -6,6 +6,13 @@ import * as QueryRunner from "./QueryRunner.js"
 import type * as QueryService from "./QueryService.js"
 
 
+export interface QueryExtension<HandlerE> {
+    useQuery<K extends readonly unknown[], A, E, R>(
+        this: ReffuseHelpers.ReffuseHelpers<R>,
+        props: UseQueryProps<K, A, E, R>,
+    ): UseQueryResult<K, A, Exclude<E, HandlerE>>
+}
+
 export interface UseQueryProps<K extends readonly unknown[], A, E, R> {
     readonly key: Stream.Stream<K>
     readonly query: (key: K) => Effect.Effect<A, E, R>

packages/extension-query/src/QueryRunner.ts

@@ -54,22 +54,23 @@ export const make = <K extends readonly unknown[], A, E, R>(
         }))
     )
 
-    const forkFetch = interrupt.pipe(
+    const run = latestKeyRef.pipe(
-        Effect.andThen(
+        Effect.flatMap(identity),
-            Ref.set(stateRef, AsyncData.loading()).pipe(
+        Effect.flatMap(key => query(key).pipe(
-                Effect.andThen(latestKeyRef),
+            Effect.matchCauseEffect({
-                Effect.flatMap(identity),
+                onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
-                Effect.flatMap(key => query(key).pipe(
+                onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-                    Effect.matchCauseEffect({
+            })
-                        onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
+        )),
-                        onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
-                    })
-                )),
 
-                Effect.provide(context),
+        Effect.provide(context),
-                Effect.fork,
+    )
-            )
+
-        ),
+    const forkFetch = interrupt.pipe(
+        Effect.andThen(Ref.set(stateRef, AsyncData.loading()).pipe(
+            Effect.andThen(run),
+            Effect.fork,
+        )),
 
         Effect.flatMap(fiber =>
             Ref.set(fiberRef, Option.some(fiber)).pipe(
@@ -82,27 +83,16 @@ export const make = <K extends readonly unknown[], A, E, R>(
     )
 
     const forkRefresh = interrupt.pipe(
-        Effect.andThen(
+        Effect.andThen(Ref.update(stateRef, previous => {
-            Ref.update(stateRef, previous => {
+            if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous))
-                if (AsyncData.isSuccess(previous) || AsyncData.isFailure(previous))
+                return AsyncData.refreshing(previous)
-                    return AsyncData.refreshing(previous)
+            if (AsyncData.isRefreshing(previous))
-                if (AsyncData.isRefreshing(previous))
+                return AsyncData.refreshing(previous.previous)
-                    return AsyncData.refreshing(previous.previous)
+            return AsyncData.loading()
-                return AsyncData.loading()
+        }).pipe(
-            }).pipe(
+            Effect.andThen(run),
-                Effect.andThen(latestKeyRef),
+            Effect.fork,
-                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(