Refactoring

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

@@ -0,0 +1,97 @@
+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 run = (key: K) => Effect.all([
+        QueryClient,
+        QueryState.makeTag<A, Exclude<E, HandledE>>(),
+    ]).pipe(
+        Effect.flatMap(([client, state]) => client.ErrorHandler.pipe(
+            Effect.flatMap(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) => run(key).pipe(
+        Effect.provide(QueryState.layer(
+            QueryState.makeTag<A, Exclude<E, HandledE>>(),
+            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 fiber = yield* Effect.forkDaemon(run(key).pipe(
+            Effect.tap(() => Queue.shutdown(stateQueue)),
+
+            Effect.provide(QueryState.layer(
+                QueryState.makeTag<A, Exclude<E, HandledE>>(),
+                stateRef,
+                value => Queue.offer(stateQueue, value).pipe(
+                    Effect.andThen(Ref.set(stateRef, value)),
+                    Effect.andThen(Ref.set(globalStateRef, value)),
+                ),
+            )),
+        ))
+
+        return [fiber, Stream.fromQueue(stateQueue)] as const
+    })
+
+
+    return {
+        context,
+        stateRef: globalStateRef,
+
+        mutate,
+        forkMutate,
+    }
+})

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

@@ -0,0 +1,148 @@
+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 * as QueryClient from "../QueryClient.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<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<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>
+}
+
+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<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 run = QueryClient.pipe(
+        Effect.flatMap(client => client.ErrorHandler),
+        Effect.flatMap(errorHandler => latestKeyRef.pipe(
+            Effect.flatMap(identity),
+            Effect.flatMap(key => query(key).pipe(
+                errorHandler.handle,
+                Effect.matchCauseEffect({
+                    onSuccess: v => Ref.set(stateRef, AsyncData.success(v)),
+                    onFailure: c => Ref.set(stateRef, AsyncData.failure(c)),
+                }),
+            )),
+        )),
+
+        Effect.provide(context),
+    )
+
+    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(
+                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(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.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 {
+        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,6 @@
+/** @internal */
+export * as MutationRunner from "./MutationRunner.js"
+/** @internal */
+export * as QueryRunner from "./QueryRunner.js"
+/** @internal */
+export * as QueryState from "./QueryState.js"