0.2.2 (#31)

Co-authored-by: Julien Valverdé <julien.valverde@mailo.com> Co-authored-by: Renovate Bot <renovate-bot@valverde.cloud> Reviewed-on: Thilawyn/effect-fc#31
2026-01-16 17:05:30 +01:00
parent 89f966d93e
commit 0e8adf8506
14 changed files with 656 additions and 461 deletions
@@ -1,50 +1,85 @@
-import { type Cause, type Context, Effect, Fiber, identity, Option, Pipeable, Predicate, type Scope, Stream, type Subscribable, SubscriptionRef } from "effect"
+import { type Cause, type Context, DateTime, type Duration, Effect, Equal, Equivalence, Fiber, HashMap, identity, Option, Pipeable, Predicate, type Scope, Stream, Subscribable, SubscriptionRef } from "effect"
+import * as QueryClient from "./QueryClient.js"
 import * as Result from "./Result.js"


 export const QueryTypeId: unique symbol = Symbol.for("@effect-fc/Query/Query")
 export type QueryTypeId = typeof QueryTypeId

-export interface Query<in out K extends readonly any[], in out A, in out E = never, in out R = never, in out P = never>
+export interface Query<in out K extends Query.AnyKey, in out A, in out E = never, in out R = never, in out P = never>
 extends Pipeable.Pipeable {
    readonly [QueryTypeId]: QueryTypeId

-    readonly context: Context.Context<Scope.Scope | R>
+    readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>
    readonly key: Stream.Stream<K>
    readonly f: (key: K) => Effect.Effect<A, E, R>
    readonly initialProgress: P

+    readonly staleTime: Duration.DurationInput
+    readonly refreshOnWindowFocus: boolean
+
    readonly latestKey: Subscribable.Subscribable<Option.Option<K>>
    readonly fiber: Subscribable.Subscribable<Option.Option<Fiber.Fiber<A, E>>>
    readonly result: Subscribable.Subscribable<Result.Result<A, E, P>>
+    readonly latestFinalResult: Subscribable.Subscribable<Option.Option<Result.Final<A, E, P>>>

+    readonly run: Effect.Effect<void>
    fetch(key: K): Effect.Effect<Result.Final<A, E, P>>
    fetchSubscribable(key: K): Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>>
-    readonly refetch: Effect.Effect<Result.Final<A, E, P>, Cause.NoSuchElementException>
-    readonly refetchSubscribable: Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>, Cause.NoSuchElementException>
    readonly refresh: Effect.Effect<Result.Final<A, E, P>, Cause.NoSuchElementException>
    readonly refreshSubscribable: Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>, Cause.NoSuchElementException>
+
+    readonly invalidateCache: Effect.Effect<void>
+    invalidateCacheEntry(key: K): Effect.Effect<void>
 }

-export class QueryImpl<in out K extends readonly any[], in out A, in out E = never, in out R = never, in out P = never>
+export declare namespace Query {
+    export type AnyKey = readonly any[]
+}
+
+export class QueryImpl<in out K extends Query.AnyKey, in out A, in out E = never, in out R = never, in out P = never>
 extends Pipeable.Class() implements Query<K, A, E, R, P> {
    readonly [QueryTypeId]: QueryTypeId = QueryTypeId

    constructor(
-        readonly context: Context.Context<Scope.Scope | NoInfer<R>>,
+        readonly context: Context.Context<Scope.Scope | QueryClient.QueryClient | R>,
        readonly key: Stream.Stream<K>,
        readonly f: (key: K) => Effect.Effect<A, E, R>,
        readonly initialProgress: P,

+        readonly staleTime: Duration.DurationInput,
+        readonly refreshOnWindowFocus: boolean,
+
        readonly latestKey: SubscriptionRef.SubscriptionRef<Option.Option<K>>,
        readonly fiber: SubscriptionRef.SubscriptionRef<Option.Option<Fiber.Fiber<A, E>>>,
        readonly result: SubscriptionRef.SubscriptionRef<Result.Result<A, E, P>>,
+        readonly latestFinalResult: SubscriptionRef.SubscriptionRef<Option.Option<Result.Final<A, E, P>>>,

        readonly runSemaphore: Effect.Semaphore,
    ) {
        super()
    }

+    get run(): Effect.Effect<void> {
+        return Effect.all([
+            Stream.runForEach(this.key, key => this.fetchSubscribable(key)),
+
+            Effect.promise(() => import("@effect/platform-browser")).pipe(
+                Effect.andThen(({ BrowserStream }) => this.refreshOnWindowFocus
+                    ? Stream.runForEach(
+                        BrowserStream.fromEventListenerWindow("focus"),
+                        () => this.refreshSubscribable,
+                    )
+                    : Effect.void
+                ),
+                Effect.catchAllDefect(() => Effect.void),
+            ),
+        ], { concurrency: "unbounded" }).pipe(
+            Effect.ignore,
+            this.runSemaphore.withPermits(1),
+        )
+    }
+
    get interrupt(): Effect.Effect<void, never, never> {
        return Effect.andThen(this.fiber, Option.match({
            onSome: Fiber.interrupt,
@@ -55,138 +90,230 @@ extends Pipeable.Class() implements Query<K, A, E, R, P> {
    fetch(key: K): Effect.Effect<Result.Final<A, E, P>> {
        return this.interrupt.pipe(
            Effect.andThen(SubscriptionRef.set(this.latestKey, Option.some(key))),
-            Effect.andThen(Effect.provide(this.start(key), this.context)),
-            Effect.andThen(sub => this.watch(sub)),
+            Effect.andThen(this.latestFinalResult),
+            Effect.andThen(previous => this.startCached(key, Option.isSome(previous)
+                ? Result.willFetch(previous.value) as Result.Final<A, E, P>
+                : Result.initial()
+            )),
+            Effect.andThen(sub => this.watch(key, sub)),
+            Effect.provide(this.context),
        )
    }
+
    fetchSubscribable(key: K): Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>> {
        return this.interrupt.pipe(
            Effect.andThen(SubscriptionRef.set(this.latestKey, Option.some(key))),
-            Effect.andThen(Effect.provide(this.start(key), this.context)),
-        )
-    }
-    get refetch(): Effect.Effect<Result.Final<A, E, P>, Cause.NoSuchElementException> {
-        return this.interrupt.pipe(
-            Effect.andThen(this.latestKey),
-            Effect.andThen(identity),
-            Effect.andThen(key => Effect.provide(this.start(key), this.context)),
-            Effect.andThen(sub => this.watch(sub)),
-        )
-    }
-    get refetchSubscribable(): Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>, Cause.NoSuchElementException> {
-        return this.interrupt.pipe(
-            Effect.andThen(this.latestKey),
-            Effect.andThen(identity),
-            Effect.andThen(key => Effect.provide(this.start(key), this.context)),
+            Effect.andThen(this.latestFinalResult),
+            Effect.andThen(previous => this.startCached(key, Option.isSome(previous)
+                ? Result.willFetch(previous.value) as Result.Final<A, E, P>
+                : Result.initial()
+            )),
+            Effect.tap(sub => Effect.forkScoped(this.watch(key, sub))),
+            Effect.provide(this.context),
        )
    }
+
    get refresh(): Effect.Effect<Result.Final<A, E, P>, Cause.NoSuchElementException> {
        return this.interrupt.pipe(
-            Effect.andThen(this.latestKey),
-            Effect.andThen(identity),
-            Effect.andThen(key => Effect.provide(this.start(key, true), this.context)),
-            Effect.andThen(sub => this.watch(sub)),
+            Effect.andThen(Effect.Do),
+            Effect.bind("latestKey", () => Effect.andThen(this.latestKey, identity)),
+            Effect.bind("latestFinalResult", () => this.latestFinalResult),
+            Effect.bind("subscribable", ({ latestKey, latestFinalResult }) =>
+                this.startCached(latestKey, Option.isSome(latestFinalResult)
+                    ? Result.willRefresh(latestFinalResult.value) as Result.Final<A, E, P>
+                    : Result.initial()
+                )
+            ),
+            Effect.andThen(({ latestKey, subscribable }) => this.watch(latestKey, subscribable)),
+            Effect.provide(this.context),
        )
    }
-    get refreshSubscribable(): Effect.Effect<Subscribable.Subscribable<Result.Result<A, E, P>>, Cause.NoSuchElementException> {
+
+    get refreshSubscribable(): Effect.Effect<
+        Subscribable.Subscribable<Result.Result<A, E, P>>,
+        Cause.NoSuchElementException
+    > {
        return this.interrupt.pipe(
-            Effect.andThen(this.latestKey),
-            Effect.andThen(identity),
-            Effect.andThen(key => Effect.provide(this.start(key, true), this.context)),
+            Effect.andThen(Effect.Do),
+            Effect.bind("latestKey", () => Effect.andThen(this.latestKey, identity)),
+            Effect.bind("latestFinalResult", () => this.latestFinalResult),
+            Effect.bind("subscribable", ({ latestKey, latestFinalResult }) =>
+                this.startCached(latestKey, Option.isSome(latestFinalResult)
+                    ? Result.willRefresh(latestFinalResult.value) as Result.Final<A, E, P>
+                    : Result.initial()
+                )
+            ),
+            Effect.tap(({ latestKey, subscribable }) => Effect.forkScoped(this.watch(latestKey, subscribable))),
+            Effect.map(({ subscribable }) => subscribable),
+            Effect.provide(this.context),
        )
    }

+    startCached(
+        key: K,
+        initial: Result.Initial | Result.Final<A, E, P>,
+    ): Effect.Effect<
+        Subscribable.Subscribable<Result.Result<A, E, P>>,
+        never,
+        Scope.Scope | QueryClient.QueryClient | R
+    > {
+        return Effect.andThen(this.getCacheEntry(key), Option.match({
+            onSome: entry => Effect.andThen(
+                QueryClient.isQueryClientCacheEntryStale(entry, this.staleTime),
+                isStale => isStale
+                    ? this.start(key, Result.willRefresh(entry.result) as Result.Final<A, E, P>)
+                    : Effect.succeed(Subscribable.make({
+                        get: Effect.succeed(entry.result as Result.Result<A, E, P>),
+                        get changes() { return Stream.make(entry.result as Result.Result<A, E, P>) },
+                    })),
+            ),
+            onNone: () => this.start(key, initial),
+        }))
+    }
+
    start(
        key: K,
-        refresh?: boolean,
+        initial: Result.Initial | Result.Final<A, E, P>,
    ): Effect.Effect<
        Subscribable.Subscribable<Result.Result<A, E, P>>,
        never,
        Scope.Scope | R
    > {
-        return this.result.pipe(
-            Effect.map(previous => Result.isFinal(previous)
-                ? previous
-                : undefined
-            ),
-            Effect.andThen(previous => Result.unsafeForkEffect(
-                Effect.onExit(this.f(key), () => SubscriptionRef.set(this.fiber, Option.none())),
-                {
-                    initialProgress: this.initialProgress,
-                    refresh: refresh && previous,
-                    previous,
-                } as Result.unsafeForkEffect.Options<A, E, P>,
+        return Result.unsafeForkEffect(
+            Effect.onExit(this.f(key), () => Effect.andThen(
+                Effect.all([Effect.fiberId, this.fiber]),
+                ([currentFiberId, fiber]) => Option.match(fiber, {
+                    onSome: v => Equal.equals(currentFiberId, v.id())
+                        ? SubscriptionRef.set(this.fiber, Option.none())
+                        : Effect.void,
+                    onNone: () => Effect.void,
+                }),
            )),
+
+            {
+                initial,
+                initialProgress: this.initialProgress,
+            } as Result.unsafeForkEffect.Options<A, E, P>,
+        ).pipe(
            Effect.tap(([, fiber]) => SubscriptionRef.set(this.fiber, Option.some(fiber))),
            Effect.map(([sub]) => sub),
        )
    }

    watch(
+        key: K,
        sub: Subscribable.Subscribable<Result.Result<A, E, P>>
-    ): Effect.Effect<Result.Final<A, E, P>> {
-        return Effect.andThen(
-            sub.get,
-            initial => Stream.runFoldEffect(
-                Stream.filter(sub.changes, Predicate.not(Result.isInitial)),
+    ): Effect.Effect<Result.Final<A, E, P>, never, QueryClient.QueryClient> {
+        return sub.get.pipe(
+            Effect.andThen(initial => Stream.runFoldEffect(
+                sub.changes,
                initial,
                (_, result) => Effect.as(SubscriptionRef.set(this.result, result), result),
+            ) as Effect.Effect<Result.Final<A, E, P>>),
+            Effect.tap(result => SubscriptionRef.set(this.latestFinalResult, Option.some(result))),
+            Effect.tap(result => Result.isSuccess(result)
+                ? this.updateCacheEntry(key, result)
+                : Effect.void
            ),
-        ) as Effect.Effect<Result.Final<A, E, P>>
+        )
+    }
+
+    makeCacheKey(key: K): QueryClient.QueryClientCacheKey {
+        return new QueryClient.QueryClientCacheKey(key, this.f as (key: Query.AnyKey) => Effect.Effect<unknown, unknown, unknown>)
+    }
+
+    getCacheEntry(
+        key: K
+    ): Effect.Effect<Option.Option<QueryClient.QueryClientCacheEntry>, never, QueryClient.QueryClient> {
+        return QueryClient.QueryClient.pipe(
+            Effect.andThen(client => client.cache),
+            Effect.map(HashMap.get(this.makeCacheKey(key))),
+        )
+    }
+
+    updateCacheEntry(
+        key: K,
+        result: Result.Success<A>,
+    ): Effect.Effect<QueryClient.QueryClientCacheEntry, never, QueryClient.QueryClient> {
+        return Effect.Do.pipe(
+            Effect.bind("client", () => QueryClient.QueryClient),
+            Effect.bind("now", () => DateTime.now),
+            Effect.let("entry", ({ now }) => new QueryClient.QueryClientCacheEntry(result, now)),
+            Effect.tap(({ client, entry }) => SubscriptionRef.update(
+                client.cache,
+                HashMap.set(this.makeCacheKey(key), entry),
+            )),
+            Effect.map(({ entry }) => entry),
+        )
+    }
+
+    get invalidateCache(): Effect.Effect<void> {
+        return QueryClient.QueryClient.pipe(
+            Effect.andThen(client => SubscriptionRef.update(
+                client.cache,
+                HashMap.filter((_, key) => !Equivalence.strict()(key.f, this.f)),
+            )),
+            Effect.provide(this.context),
+        )
+    }
+
+    invalidateCacheEntry(key: K): Effect.Effect<void> {
+        return QueryClient.QueryClient.pipe(
+            Effect.andThen(client => SubscriptionRef.update(
+                client.cache,
+                HashMap.remove(this.makeCacheKey(key)),
+            )),
+            Effect.provide(this.context),
+        )
    }
 }

-export const isQuery = (u: unknown): u is Query<unknown[], unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)
+export const isQuery = (u: unknown): u is Query<readonly unknown[], unknown, unknown, unknown, unknown> => Predicate.hasProperty(u, QueryTypeId)

 export declare namespace make {
-    export interface Options<K extends readonly any[], A, E = never, R = never, P = never> {
+    export interface Options<K extends Query.AnyKey, A, E = never, R = never, P = never> {
        readonly key: Stream.Stream<K>
        readonly f: (key: NoInfer<K>) => Effect.Effect<A, E, Result.forkEffect.InputContext<R, NoInfer<P>>>
        readonly initialProgress?: P
+        readonly staleTime?: Duration.DurationInput
+        readonly refreshOnWindowFocus?: boolean
    }
 }

-export const make = Effect.fnUntraced(function* <K extends readonly any[], A, E = never, R = never, P = never>(
+export const make = Effect.fnUntraced(function* <K extends Query.AnyKey, A, E = never, R = never, P = never>(
    options: make.Options<K, A, E, R, P>
 ): Effect.fn.Return<
    Query<K, A, E, Result.forkEffect.OutputContext<A, E, R, P>, P>,
    never,
-    Scope.Scope | Result.forkEffect.OutputContext<A, E, R, P>
+    Scope.Scope | QueryClient.QueryClient | Result.forkEffect.OutputContext<A, E, R, P>
 > {
+    const client = yield* QueryClient.QueryClient
+
    return new QueryImpl(
-        yield* Effect.context<Scope.Scope | Result.forkEffect.OutputContext<A, E, R, P>>(),
+        yield* Effect.context<Scope.Scope | QueryClient.QueryClient | Result.forkEffect.OutputContext<A, E, R, P>>(),
        options.key,
        options.f as any,
        options.initialProgress as P,

+        options.staleTime ?? client.defaultStaleTime,
+        options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus,
+
        yield* SubscriptionRef.make(Option.none<K>()),
        yield* SubscriptionRef.make(Option.none<Fiber.Fiber<A, E>>()),
        yield* SubscriptionRef.make(Result.initial<A, E, P>()),
+        yield* SubscriptionRef.make(Option.none<Result.Final<A, E, P>>()),

        yield* Effect.makeSemaphore(1),
    )
 })

-export const service = <K extends readonly any[], A, E = never, R = never, P = never>(
+export const service = <K extends Query.AnyKey, A, E = never, R = never, P = never>(
    options: make.Options<K, A, E, R, P>
 ): Effect.Effect<
    Query<K, A, E, Result.forkEffect.OutputContext<A, E, R, P>, P>,
    never,
-    Scope.Scope | Result.forkEffect.OutputContext<A, E, R, P>
+    Scope.Scope | QueryClient.QueryClient | Result.forkEffect.OutputContext<A, E, R, P>
 > => Effect.tap(
    make(options),
-    query => Effect.forkScoped(run(query)),
+    query => Effect.forkScoped(query.run),
 )
-
-export const run = <K extends readonly any[], A, E, R, P>(
-    self: Query<K, A, E, R, P>
-): Effect.Effect<void> => {
-    const _self = self as QueryImpl<K, A, E, R, P>
-    return Stream.runForEach(_self.key, key => _self.interrupt.pipe(
-        Effect.andThen(SubscriptionRef.set(_self.latestKey, Option.some(key))),
-        Effect.andThen(_self.start(key)),
-        Effect.andThen(sub => Effect.forkScoped(_self.watch(sub))),
-        Effect.provide(_self.context),
-        _self.runSemaphore.withPermits(1),
-    ))
-}