Refactor Query

packages/effect-fc-next/src/Query.ts

@@ -1,4 +1,4 @@
-import { type Cause, type Context, Duration, Effect, Equal, Exit, Fiber, Option, Pipeable, Predicate, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
+import { type Cause, type Context, Duration, Effect, Equal, type Equivalence, Exit, Fiber, Option, Pipeable, Predicate, type Scope, Semaphore, Stream, SubscriptionRef } from "effect"
 import { AsyncResult } from "effect/unstable/reactivity"
 import * as Lens from "./Lens.js"
 import * as QueryClient from "./QueryClient.js"
@@ -8,51 +8,61 @@ import * as Subscribable from "./Subscribable.js"
 export const QueryTypeId: unique symbol = Symbol.for("@effect-fc/Query/Query")
 export type QueryTypeId = typeof QueryTypeId
 
-export interface Query<in out K, in out A, in out KE = never, in out KR = never, in out E = never, in out R = never>
+export interface Query<in out K, in out A, in out E = never, in out R = never>
 extends Pipeable.Pipeable {
     readonly [QueryTypeId]: QueryTypeId
 
-    readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | KR | R>
-    readonly key: Stream.Stream<K, KE, KR>
+    readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>
     readonly f: (key: K) => Effect.Effect<A, E, R>
+    readonly keyEquivalence: Equivalence.Equivalence<K>
 
     readonly staleTime: Duration.Duration
     readonly refreshOnWindowFocus: boolean
 
     readonly latestKey: Subscribable.Subscribable<Option.Option<K>>
     readonly fiber: Subscribable.Subscribable<Option.Option<Fiber.Fiber<A, E>>>
-    readonly result: Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>
-    readonly latestFinalResult: Subscribable.Subscribable<Option.Option<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>>>
+    readonly state: Subscribable.Subscribable<QueryState<K, A, E>>
+    readonly latestFinalState: Subscribable.Subscribable<Option.Option<FinalQueryState<K, A, E>>>
 
     readonly run: Effect.Effect<void>
-    fetch(key: K): Effect.Effect<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>, Cause.NoSuchElementError>
-    fetchSubscribable(key: K): Effect.Effect<Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>, Cause.NoSuchElementError>
-    readonly refresh: Effect.Effect<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>, Cause.NoSuchElementError>
-    readonly refreshSubscribable: Effect.Effect<Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>, Cause.NoSuchElementError>
+    fetch(key: K): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError>
+    fetchSubscribable(key: K): Effect.Effect<Subscribable.Subscribable<QueryState<K, A, E>>, Cause.NoSuchElementError>
+    readonly refresh: Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError>
+    readonly refreshSubscribable: Effect.Effect<Subscribable.Subscribable<QueryState<K, A, E>>, Cause.NoSuchElementError>
 
     readonly invalidateCache: Effect.Effect<void>
     invalidateCacheEntry(key: K): Effect.Effect<void>
 }
 
-export const isQuery = (u: unknown): u is Query<unknown, unknown, unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)
+export interface QueryState<in out K, in out A, in out E = never> {
+    readonly key: K
+    readonly result: AsyncResult.AsyncResult<A, E>
+}
+
+export interface FinalQueryState<in out K, in out A, in out E = never> {
+    readonly key: K
+    readonly result: AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>
+}
+
+export const isQuery = (u: unknown): u is Query<unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)
 
 
-export class QueryImpl<in out K, in out A, in out KE = never, in out KR = never, in out E = never, in out R = never>
-extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
+export class QueryImpl<in out K, in out A, in out E = never, in out R = never>
+extends Pipeable.Class implements Query<K, A, E, R> {
     readonly [QueryTypeId]: QueryTypeId = QueryTypeId
 
     constructor(
-        readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | KR | R>,
-        readonly key: Stream.Stream<K, KE, KR>,
+        readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>,
         readonly f: (key: K) => Effect.Effect<A, E, R>,
+        readonly keyEquivalence: Equivalence.Equivalence<K>,
 
         readonly staleTime: Duration.Duration,
         readonly refreshOnWindowFocus: boolean,
 
         readonly latestKey: Lens.Lens<Option.Option<K>>,
         readonly fiber: Lens.Lens<Option.Option<Fiber.Fiber<A, E>>>,
-        readonly result: Lens.Lens<AsyncResult.AsyncResult<A, E>>,
-        readonly latestFinalResult: Lens.Lens<Option.Option<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>>>,
+        readonly state: Lens.Lens<QueryState<K, A, E>>,
+        readonly latestFinalState: Lens.Lens<Option.Option<FinalQueryState<K, A, E>>>,
 
         readonly runSemaphore: Semaphore.Semaphore,
     ) {
@@ -60,10 +70,7 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
     }
 
     get run(): Effect.Effect<void> {
-        return Effect.all([
-            Stream.runForEach(this.key, key => this.fetchSubscribable(key)),
-
-            Effect.promise(() => import("@effect/platform-browser")).pipe(
+        return Effect.promise(() => import("@effect/platform-browser")).pipe(
             Effect.flatMap(({ BrowserStream }) => this.refreshOnWindowFocus
                 ? Stream.runForEach(
                     BrowserStream.fromEventListenerWindow("focus"),
@@ -72,8 +79,6 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
                 : Effect.void
             ),
             Effect.catchDefect(() => Effect.void),
-            ),
-        ], { concurrency: "unbounded" }).pipe(
             Effect.ignore,
             this.runSemaphore.withPermits(1),
             Effect.provide(this.context),
@@ -87,117 +92,156 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
         }))
     }
 
-    fetch(key: K): Effect.Effect<
-        AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>,
-        Cause.NoSuchElementError
-    > {
+    fetch(key: K): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
         return this.interrupt.pipe(
             Effect.andThen(Lens.set(this.latestKey, Option.some(key))),
-            Effect.andThen(this.startCached(key, AsyncResult.initial())),
-            Effect.flatMap(state => this.watch(key, state)),
+            Effect.andThen(this.startCached({
+                key,
+                result: AsyncResult.initial(false)
+            })),
+            Effect.flatMap(state => this.watch(state)),
             Effect.provide(this.context),
         )
     }
 
     fetchSubscribable(key: K): Effect.Effect<
-        Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>,
+        Subscribable.Subscribable<QueryState<K, A, E>>,
         Cause.NoSuchElementError
     > {
         return this.interrupt.pipe(
             Effect.andThen(Lens.set(this.latestKey, Option.some(key))),
-            Effect.andThen(this.startCached(key, AsyncResult.initial(false))),
-            Effect.tap(state => Effect.forkScoped(this.watch(key, state))),
+            Effect.andThen(this.startCached({
+                key,
+                result: AsyncResult.initial(false)
+            })),
+            Effect.tap(state => Effect.forkScoped(this.watch(state))),
             Effect.provide(this.context),
         )
     }
 
-    get refresh(): Effect.Effect<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>, Cause.NoSuchElementError> {
+    get refresh(): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
         return this.interrupt.pipe(
             Effect.andThen(Effect.Do),
             Effect.bind("latestKey", () => Effect.flatMap(Lens.get(this.latestKey), Effect.fromOption)),
-            Effect.bind("latestFinalResult", () => Lens.get(this.latestFinalResult)),
-            Effect.bind("subscribable", ({ latestKey, latestFinalResult }) =>
-                this.startCached(latestKey, Option.getOrElse(latestFinalResult, () => AsyncResult.initial(true)))
+            Effect.bind("latestFinalState", () => Lens.get(this.latestFinalState)),
+            Effect.bind("subscribable", ({ latestKey, latestFinalState }) =>
+                this.startCached(Option.getOrElse(latestFinalState, () => ({
+                    key: latestKey,
+                    result: AsyncResult.initial(true),
+                })))
             ),
-            Effect.flatMap(({ latestKey, subscribable }) => this.watch(latestKey, subscribable)),
+            Effect.flatMap(({ subscribable }) => this.watch(subscribable)),
             Effect.provide(this.context),
         )
     }
 
     get refreshSubscribable(): Effect.Effect<
-        Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>,
+        Subscribable.Subscribable<QueryState<K, A, E>>,
         Cause.NoSuchElementError
     > {
         return this.interrupt.pipe(
             Effect.andThen(Effect.Do),
             Effect.bind("latestKey", () => Effect.flatMap(Lens.get(this.latestKey), Effect.fromOption)),
-            Effect.bind("latestFinalResult", () => Lens.get(this.latestFinalResult)),
-            Effect.bind("subscribable", ({ latestKey, latestFinalResult }) =>
-                this.startCached(latestKey, Option.getOrElse(latestFinalResult, () => AsyncResult.initial(true)))
+            Effect.bind("latestFinalState", () => Lens.get(this.latestFinalState)),
+            Effect.bind("subscribable", ({ latestKey, latestFinalState }) =>
+                this.startCached(Option.getOrElse(latestFinalState, () => ({
+                    key: latestKey,
+                    result: AsyncResult.initial(true),
+                })))
             ),
-            Effect.tap(({ latestKey, subscribable }) => Effect.forkScoped(this.watch(latestKey, subscribable))),
+            Effect.tap(({ subscribable }) => Effect.forkScoped(this.watch(subscribable))),
             Effect.map(({ subscribable }) => subscribable),
             Effect.provide(this.context),
         )
     }
 
     startCached(
-        key: K,
-        previous: AsyncResult.AsyncResult<A, E>,
+        previous: QueryState<K, A, E>,
     ): Effect.Effect<
-        Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>,
+        Subscribable.Subscribable<QueryState<K, A, E>>,
         Cause.NoSuchElementError,
         Scope.Scope | QueryClient.QueryClient | R
     > {
-        return Effect.flatMap(this.getCacheEntry(key), Option.match({
+        return Effect.flatMap(this.getCacheEntry(previous.key), Option.match({
             onSome: entry => Effect.flatMap(
                 QueryClient.isQueryClientCacheEntryStale(entry),
                 isStale => isStale
-                    ? this.start(key, entry.result as AsyncResult.AsyncResult<A, E>)
+                    ? this.start({
+                        key: previous.key,
+                        result: entry.result as AsyncResult.AsyncResult<A, E>,
+                    })
                     : Effect.succeed(Subscribable.make({
-                        get: Effect.succeed(entry.result as AsyncResult.AsyncResult<A, E>),
-                        get changes() { return Stream.make(entry.result as AsyncResult.AsyncResult<A, E>) },
+                        get: Effect.succeed({
+                            key: previous.key,
+                            result: entry.result as AsyncResult.AsyncResult<A, E>,
+                        }),
+                        get changes() {
+                            return Stream.make({
+                                key: previous.key,
+                                result: entry.result as AsyncResult.AsyncResult<A, E>,
+                            })
+                        },
                     })),
             ),
-            onNone: () => this.start(key, previous),
+            onNone: () => this.start(previous),
         }))
     }
 
     start(
-        key: K,
-        previous: AsyncResult.AsyncResult<A, E>,
+        previous: QueryState<K, A, E>,
     ): Effect.Effect<
-        Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>,
+        Subscribable.Subscribable<QueryState<K, A, E>>,
         never,
         Scope.Scope | R
     > {
         return Effect.gen({ self: this }, function*() {
-            const state = Lens.fromSubscriptionRef(yield* SubscriptionRef.make(previous))
+            const subscribable = Lens.fromSubscriptionRef(yield* SubscriptionRef.make(previous))
 
             const fiber = yield* Effect.forkScoped(Effect.andThen(
-                Lens.update(state, AsyncResult.match({
-                    onInitial: () => AsyncResult.initial(true),
-                    onSuccess: v => AsyncResult.success(v.value, {
+                Lens.update(subscribable, previous => AsyncResult.match(previous.result, {
+                    onInitial: () => ({
+                        key: previous.key,
+                        result: AsyncResult.initial(true),
+                    }),
+                    onSuccess: result => ({
+                        key: previous.key,
+                        result: AsyncResult.success(result.value, {
                             waiting: true,
                         }),
-                    onFailure: v => AsyncResult.failure(v.cause, {
+                    }),
+                    onFailure: result => ({
+                        key: previous.key,
+                        result: AsyncResult.failure(result.cause, {
                             waiting: true,
-                        previousSuccess: v.previousSuccess,
-                    })
+                            previousSuccess: result.previousSuccess,
+                        }),
+                    }),
                 })),
 
-                Effect.onExit(this.f(key), exit => Lens.update(
-                    state,
+                Effect.onExit(this.f(previous.key), exit => Lens.update(
+                    subscribable,
                     previous => Exit.match(exit, {
-                        onSuccess: v => AsyncResult.success(v),
-                        onFailure: c => AsyncResult.match(previous, {
-                            onInitial: () => AsyncResult.failure(c),
-                            onSuccess: v => AsyncResult.failure(c, {
+                        onSuccess: v => ({
+                            key: previous.key,
+                            result: AsyncResult.success(v),
+                        }),
+                        onFailure: c => AsyncResult.match(previous.result, {
+                            onInitial: () => ({
+                                key: previous.key,
+                                result: AsyncResult.failure(c),
+                            }),
+                            onSuccess: v => ({
+                                key: previous.key,
+                                result: AsyncResult.failure(c, {
                                     previousSuccess: Option.some(v),
                                 }),
-                            onFailure: v => AsyncResult.failure(c, {
+                            }),
+                            onFailure: v => ({
+                                key: previous.key,
+                                result: AsyncResult.failure(c, {
                                     previousSuccess: v.previousSuccess,
-                            })
+                                }),
+                            }),
                         }),
                     }),
                 ).pipe(
@@ -206,7 +250,7 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
                         Lens.get(this.fiber),
                     ])),
                     Effect.flatMap(([fiberId, fiber]) => Option.match(fiber, {
-                        onSome: v => Equal.equals(fiberId, v.id)
+                        onSome: v => fiberId === v.id
                             ? Lens.set(this.fiber, Option.none())
                             : Effect.void,
                         onNone: () => Effect.void,
@@ -215,23 +259,22 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
             ))
 
             yield* Lens.set(this.fiber, Option.some(fiber))
-            return state
+            return subscribable
         })
     }
 
     watch(
-        key: K,
-        state: Subscribable.Subscribable<AsyncResult.AsyncResult<A, E>>
-    ): Effect.Effect<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>, never, QueryClient.QueryClient> {
-        return state.get.pipe(
+        subscribable: Subscribable.Subscribable<QueryState<K, A, E>>
+    ): Effect.Effect<FinalQueryState<K, A, E>, never, QueryClient.QueryClient> {
+        return subscribable.get.pipe(
             Effect.flatMap(initial => Stream.runFoldEffect(
-                state.changes,
+                subscribable.changes,
                 () => initial,
-                (_, result) => Effect.as(Lens.set(this.result, result), result),
-            ) as Effect.Effect<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>>),
-            Effect.tap(result => Lens.set(this.latestFinalResult, Option.some(result))),
-            Effect.tap(result => AsyncResult.isSuccess(result)
-                ? this.setCacheEntry(key, result)
+                (_, state) => Effect.as(Lens.set(this.state, state), state),
+            ) as Effect.Effect<FinalQueryState<K, A, E>>),
+            Effect.tap(state => Lens.set(this.latestFinalState, Option.some(state))),
+            Effect.tap(state => AsyncResult.isSuccess(state.result)
+                ? this.setCacheEntry(state.key, state.result)
                 : Effect.void
             ),
         )
@@ -285,27 +328,27 @@ extends Pipeable.Class implements Query<K, A, KE, KR, E, R> {
 }
 
 export declare namespace make {
-    export interface Options<K, A, KE = never, KR = never, E = never, R = never> {
-        readonly key: Stream.Stream<K, KE, KR>
-        readonly f: (key: NoInfer<K>) => Effect.Effect<A, E, R>
+    export interface Options<K, A, E = never, R = never> {
+        readonly f: (key: K) => Effect.Effect<A, E, R>
+        readonly keyEquivalence?: Equivalence.Equivalence<K>,
         readonly staleTime?: Duration.Input
         readonly refreshOnWindowFocus?: boolean
     }
 }
 
-export const make = Effect.fnUntraced(function* <K, A, KE = never, KR = never, E = never, R = never>(
-    options: make.Options<K, A, KE, KR, E, R>
+export const make = Effect.fnUntraced(function* <K, A, E = never, R = never>(
+    options: make.Options<K, A, E, R>
 ): Effect.fn.Return<
-    Query<K, A, KE, KR, E, R>,
+    Query<K, A, E, R>,
     Cause.NoSuchElementError,
-    Scope.Scope | QueryClient.QueryClient | KR | R
+    Scope.Scope | QueryClient.QueryClient | R
 > {
     const client = yield* QueryClient.QueryClient
 
     return new QueryImpl(
-        yield* Effect.context<Scope.Scope | QueryClient.QueryClient | KR | R>(),
-        options.key,
+        yield* Effect.context<Scope.Scope | QueryClient.QueryClient | R>(),
         options.f,
+        options.keyEquivalence ?? Equal.asEquivalence(),
 
         options.staleTime ? yield* Effect.fromOption(Duration.fromInput(options.staleTime)) : client.defaultStaleTime,
         options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus,
@@ -319,12 +362,12 @@ export const make = Effect.fnUntraced(function* <K, A, KE = never, KR = never, E
     )
 })
 
-export const service = <K, A, KE = never, KR = never, E = never, R = never>(
-    options: make.Options<K, A, KE, KR, E, R>
+export const service = <K, A, E = never, R = never>(
+    options: make.Options<K, A, E, R>
 ): Effect.Effect<
-    Query<K, A, KE, KR, E, R>,
+    Query<K, A, E, R>,
     Cause.NoSuchElementError,
-    Scope.Scope | QueryClient.QueryClient | KR | R
+    Scope.Scope | QueryClient.QueryClient | R
 > => Effect.tap(
     make(options),
     query => Effect.forkScoped(query.run),