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

@@ -13,13 +13,13 @@ extends Pipeable.Pipeable {
     readonly [QueryTypeId]: QueryTypeId
 
     readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>
-    readonly f: (key: K) => Effect.Effect<A, E, R>
+    readonly key: Subscribable.Subscribable<K>
     readonly keyEquivalence: Equivalence.Equivalence<K>
+    readonly f: (key: K) => Effect.Effect<A, E, R>
 
     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 state: Subscribable.Subscribable<QueryState<K, A, E>>
     readonly latestFinalState: Subscribable.Subscribable<Option.Option<FinalQueryState<K, A, E>>>
@@ -53,13 +53,13 @@ extends Pipeable.Class implements Query<K, A, E, R> {
 
     constructor(
         readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>,
-        readonly f: (key: K) => Effect.Effect<A, E, R>,
+        readonly key: Lens.Lens<K>,
         readonly keyEquivalence: Equivalence.Equivalence<K>,
+        readonly f: (key: K) => Effect.Effect<A, E, R>,
 
         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 state: Lens.Lens<QueryState<K, A, E>>,
         readonly latestFinalState: Lens.Lens<Option.Option<FinalQueryState<K, A, E>>>,
@@ -70,15 +70,29 @@ extends Pipeable.Class implements Query<K, A, E, R> {
     }
 
     get run(): Effect.Effect<void> {
-        return Effect.promise(() => import("@effect/platform-browser")).pipe(
+        return Effect.all([
-            Effect.flatMap(({ BrowserStream }) => this.refreshOnWindowFocus
+            Stream.runFoldEffect(
-                ? Stream.runForEach(
+                this.key.changes,
-                    BrowserStream.fromEventListenerWindow("focus"),
+                () => Option.none<K>(),
-                    () => this.refreshSubscribable,
+                (previous, key) => Effect.as(
-                )
+                    Option.isSome(previous) && this.keyEquivalence(key, previous.value)
-                : Effect.void
+                        ? this.refreshSubscribable
+                        : this.fetchSubscribable(key),
+                    Option.some(key),
+                ),
             ),
-            Effect.catchDefect(() => Effect.void),
+
+            Effect.promise(() => import("@effect/platform-browser")).pipe(
+                Effect.flatMap(({ BrowserStream }) => this.refreshOnWindowFocus
+                    ? Stream.runForEach(
+                        BrowserStream.fromEventListenerWindow("focus"),
+                        () => this.refreshSubscribable,
+                    )
+                    : Effect.void
+                ),
+                Effect.catchDefect(() => Effect.void),
+            ),
+        ], { concurrency: "unbounded" }).pipe(
             Effect.ignore,
             this.runSemaphore.withPermits(1),
             Effect.provide(this.context),
@@ -93,13 +107,16 @@ extends Pipeable.Class implements Query<K, A, E, R> {
     }
 
     fetch(key: K): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
-        return this.interrupt.pipe(
+        return Effect.gen({ self: this }, function*() {
-            Effect.andThen(Lens.set(this.latestKey, Option.some(key))),
+            yield* this.interrupt
-            Effect.andThen(this.startCached({
+            yield* Lens.set(this.key, key)
+
+            const state = yield* this.startCached({
                 key,
-                result: AsyncResult.initial(false)
+                result: AsyncResult.initial(false),
-            })),
+            })
-            Effect.flatMap(state => this.watch(state)),
+            return yield* this.watch(state)
+        }).pipe(
             Effect.provide(this.context),
         )
     }
@@ -108,29 +125,39 @@ extends Pipeable.Class implements Query<K, A, E, R> {
         Subscribable.Subscribable<QueryState<K, A, E>>,
         Cause.NoSuchElementError
     > {
-        return this.interrupt.pipe(
+        return Effect.gen({ self: this }, function*() {
-            Effect.andThen(Lens.set(this.latestKey, Option.some(key))),
+            yield* this.interrupt
-            Effect.andThen(this.startCached({
+            yield* Lens.set(this.key, key)
+
+            const state = yield* this.startCached({
                 key,
-                result: AsyncResult.initial(false)
+                result: AsyncResult.initial(false),
-            })),
+            })
-            Effect.tap(state => Effect.forkScoped(this.watch(state))),
+
+            yield* Effect.forkScoped(this.watch(state))
+            return state
+        }).pipe(
             Effect.provide(this.context),
         )
     }
 
     get refresh(): Effect.Effect<FinalQueryState<K, A, E>, Cause.NoSuchElementError> {
-        return this.interrupt.pipe(
+        return Effect.gen({ self: this }, function*() {
-            Effect.andThen(Effect.Do),
+            yield* this.interrupt
-            Effect.bind("latestKey", () => Effect.flatMap(Lens.get(this.latestKey), Effect.fromOption)),
+            const latestKey = yield* Lens.get(this.key)
-            Effect.bind("latestFinalState", () => Lens.get(this.latestFinalState)),
+            const latestFinalState = yield* Lens.get(this.latestFinalState)
-            Effect.bind("subscribable", ({ latestKey, latestFinalState }) =>
+
-                this.startCached(Option.getOrElse(latestFinalState, () => ({
+            const state = yield* this.startCached(
-                    key: latestKey,
+                Option.isSome(latestFinalState) && this.keyEquivalence(latestKey, latestFinalState.value.key)
-                    result: AsyncResult.initial(true),
+                    ? latestFinalState.value
-                })))
+                    : {
-            ),
+                        key: latestKey,
-            Effect.flatMap(({ subscribable }) => this.watch(subscribable)),
+                        result: AsyncResult.initial(false),
+                    }
+            )
+
+            return yield* this.watch(state)
+        }).pipe(
             Effect.provide(this.context),
         )
     }
@@ -139,18 +166,23 @@ extends Pipeable.Class implements Query<K, A, E, R> {
         Subscribable.Subscribable<QueryState<K, A, E>>,
         Cause.NoSuchElementError
     > {
-        return this.interrupt.pipe(
+        return Effect.gen({ self: this }, function*() {
-            Effect.andThen(Effect.Do),
+            yield* this.interrupt
-            Effect.bind("latestKey", () => Effect.flatMap(Lens.get(this.latestKey), Effect.fromOption)),
+            const latestKey = yield* Lens.get(this.key)
-            Effect.bind("latestFinalState", () => Lens.get(this.latestFinalState)),
+            const latestFinalState = yield* Lens.get(this.latestFinalState)
-            Effect.bind("subscribable", ({ latestKey, latestFinalState }) =>
+
-                this.startCached(Option.getOrElse(latestFinalState, () => ({
+            const state = yield* this.startCached(
-                    key: latestKey,
+                Option.isSome(latestFinalState) && this.keyEquivalence(latestKey, latestFinalState.value.key)
-                    result: AsyncResult.initial(true),
+                    ? latestFinalState.value
-                })))
+                    : {
-            ),
+                        key: latestKey,
-            Effect.tap(({ subscribable }) => Effect.forkScoped(this.watch(subscribable))),
+                        result: AsyncResult.initial(false),
-            Effect.map(({ subscribable }) => subscribable),
+                    }
+            )
+
+            yield* Effect.forkScoped(this.watch(state))
+            return state
+        }).pipe(
             Effect.provide(this.context),
         )
     }
@@ -266,18 +298,20 @@ extends Pipeable.Class implements Query<K, A, E, R> {
     watch(
         subscribable: Subscribable.Subscribable<QueryState<K, A, E>>
     ): Effect.Effect<FinalQueryState<K, A, E>, never, QueryClient.QueryClient> {
-        return subscribable.get.pipe(
+        return Effect.gen({ self: this }, function*() {
-            Effect.flatMap(initial => Stream.runFoldEffect(
+            const initial = yield* subscribable.get
+            const final = yield* Stream.runFoldEffect(
                 subscribable.changes,
                 () => initial,
                 (_, state) => Effect.as(Lens.set(this.state, state), state),
-            ) as Effect.Effect<FinalQueryState<K, A, E>>),
+            ) as Effect.Effect<FinalQueryState<K, A, E>>
-            Effect.tap(state => Lens.set(this.latestFinalState, Option.some(state))),
+
-            Effect.tap(state => AsyncResult.isSuccess(state.result)
+            yield* Lens.set(this.latestFinalState, Option.some(final))
-                ? this.setCacheEntry(state.key, state.result)
+            if (AsyncResult.isSuccess(final.result))
-                : Effect.void
+                yield* this.setCacheEntry(final.key, final.result)
-            ),
+
-        )
+            return final
+        })
     }
 
     makeCacheKey(key: K): QueryClient.QueryClientCacheKey {
@@ -329,8 +363,10 @@ extends Pipeable.Class implements Query<K, A, E, R> {
 
 export declare namespace make {
     export interface Options<K, A, E = never, R = never> {
-        readonly f: (key: K) => Effect.Effect<A, E, R>
+        readonly key: Lens.Lens<K>,
         readonly keyEquivalence?: Equivalence.Equivalence<K>,
+        readonly f: (key: K) => Effect.Effect<A, E, R>
+
         readonly staleTime?: Duration.Input
         readonly refreshOnWindowFocus?: boolean
     }
@@ -347,16 +383,19 @@ export const make = Effect.fnUntraced(function* <K, A, E = never, R = never>(
 
     return new QueryImpl(
         yield* Effect.context<Scope.Scope | QueryClient.QueryClient | R>(),
-        options.f,
+        options.key,
         options.keyEquivalence ?? Equal.asEquivalence(),
+        options.f,
 
         options.staleTime ? yield* Effect.fromOption(Duration.fromInput(options.staleTime)) : client.defaultStaleTime,
         options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus,
 
-        Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<K>())),
         Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>())),
-        Lens.fromSubscriptionRef(yield* SubscriptionRef.make<AsyncResult.AsyncResult<A, E>>(AsyncResult.initial())),
+        Lens.fromSubscriptionRef(yield* SubscriptionRef.make<QueryState<K, A, E>>({
-        Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>>())),
+            key: yield* Lens.get(options.key),
+            result: AsyncResult.initial(false),
+        })),
+        Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none<FinalQueryState<K, A, E>>())),
 
         yield* Semaphore.make(1),
     )