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" 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 extends Pipeable.Pipeable { readonly [QueryTypeId]: QueryTypeId readonly context: Context.Context readonly key: Subscribable.Subscribable readonly keyEquivalence: Equivalence.Equivalence readonly f: (key: K) => Effect.Effect readonly staleTime: Duration.Duration readonly refreshOnWindowFocus: boolean readonly fiber: Subscribable.Subscribable>> readonly state: Subscribable.Subscribable> readonly latestFinalState: Subscribable.Subscribable>> readonly run: Effect.Effect fetch(key: K): Effect.Effect, Cause.NoSuchElementError> fetchSubscribable(key: K): Effect.Effect>, Cause.NoSuchElementError> readonly refresh: Effect.Effect, Cause.NoSuchElementError> readonly refreshSubscribable: Effect.Effect>, Cause.NoSuchElementError> readonly invalidateCache: Effect.Effect invalidateCacheEntry(key: K): Effect.Effect } export interface QueryState { readonly key: K readonly result: AsyncResult.AsyncResult } export interface FinalQueryState { readonly key: K readonly result: AsyncResult.Success | AsyncResult.Failure } export const isQuery = (u: unknown): u is Query => Predicate.hasProperty(u, QueryTypeId) export class QueryImpl extends Pipeable.Class implements Query { readonly [QueryTypeId]: QueryTypeId = QueryTypeId constructor( readonly context: Context.Context, readonly key: Lens.Lens, readonly keyEquivalence: Equivalence.Equivalence, readonly f: (key: K) => Effect.Effect, readonly staleTime: Duration.Duration, readonly refreshOnWindowFocus: boolean, readonly fiber: Lens.Lens>>, readonly state: Lens.Lens>, readonly latestFinalState: Lens.Lens>>, readonly runSemaphore: Semaphore.Semaphore, ) { super() } get run(): Effect.Effect { return Effect.all([ Stream.runFoldEffect( this.key.changes, () => Option.none(), (previous, key) => Effect.as( Option.isSome(previous) && this.keyEquivalence(key, previous.value) ? this.refreshSubscribable : this.fetchSubscribable(key), Option.some(key), ), ), 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), ) } get interrupt(): Effect.Effect { return Effect.flatMap(Lens.get(this.fiber), Option.match({ onSome: Fiber.interrupt, onNone: () => Effect.void, })) } fetch(key: K): Effect.Effect, Cause.NoSuchElementError> { return Effect.gen({ self: this }, function*() { yield* this.interrupt yield* Lens.set(this.key, key) const state = yield* this.startCached({ key, result: AsyncResult.initial(false), }) return yield* this.watch(state) }).pipe( Effect.provide(this.context), ) } fetchSubscribable(key: K): Effect.Effect< Subscribable.Subscribable>, Cause.NoSuchElementError > { return Effect.gen({ self: this }, function*() { yield* this.interrupt yield* Lens.set(this.key, key) const state = yield* this.startCached({ key, result: AsyncResult.initial(false), }) yield* Effect.forkScoped(this.watch(state)) return state }).pipe( Effect.provide(this.context), ) } get refresh(): Effect.Effect, Cause.NoSuchElementError> { return Effect.gen({ self: this }, function*() { yield* this.interrupt const latestKey = yield* Lens.get(this.key) const latestFinalState = yield* Lens.get(this.latestFinalState) const state = yield* this.startCached( Option.isSome(latestFinalState) && this.keyEquivalence(latestKey, latestFinalState.value.key) ? latestFinalState.value : { key: latestKey, result: AsyncResult.initial(false), } ) return yield* this.watch(state) }).pipe( Effect.provide(this.context), ) } get refreshSubscribable(): Effect.Effect< Subscribable.Subscribable>, Cause.NoSuchElementError > { return Effect.gen({ self: this }, function*() { yield* this.interrupt const latestKey = yield* Lens.get(this.key) const latestFinalState = yield* Lens.get(this.latestFinalState) const state = yield* this.startCached( Option.isSome(latestFinalState) && this.keyEquivalence(latestKey, latestFinalState.value.key) ? latestFinalState.value : { key: latestKey, result: AsyncResult.initial(false), } ) yield* Effect.forkScoped(this.watch(state)) return state }).pipe( Effect.provide(this.context), ) } startCached( previous: QueryState, ): Effect.Effect< Subscribable.Subscribable>, Cause.NoSuchElementError, Scope.Scope | QueryClient.QueryClient | R > { return Effect.flatMap(this.getCacheEntry(previous.key), Option.match({ onSome: entry => Effect.flatMap( QueryClient.isQueryClientCacheEntryStale(entry), isStale => isStale ? this.start({ key: previous.key, result: entry.result as AsyncResult.AsyncResult, }) : Effect.succeed(Subscribable.make({ get: Effect.succeed({ key: previous.key, result: entry.result as AsyncResult.AsyncResult, }), get changes() { return Stream.make({ key: previous.key, result: entry.result as AsyncResult.AsyncResult, }) }, })), ), onNone: () => this.start(previous), })) } start( previous: QueryState, ): Effect.Effect< Subscribable.Subscribable>, never, Scope.Scope | R > { return Effect.gen({ self: this }, function*() { const subscribable = Lens.fromSubscriptionRef(yield* SubscriptionRef.make(previous)) const fiber = yield* Effect.forkScoped(Effect.andThen( 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: result => ({ key: previous.key, result: AsyncResult.failure(result.cause, { waiting: true, previousSuccess: result.previousSuccess, }), }), })), Effect.onExit(this.f(previous.key), exit => Lens.update( subscribable, previous => Exit.match(exit, { 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 => ({ key: previous.key, result: AsyncResult.failure(c, { previousSuccess: v.previousSuccess, }), }), }), }), ).pipe( Effect.andThen(Effect.all([ Effect.fiberId, Lens.get(this.fiber), ])), Effect.flatMap(([fiberId, fiber]) => Option.match(fiber, { onSome: v => fiberId === v.id ? Lens.set(this.fiber, Option.none()) : Effect.void, onNone: () => Effect.void, })), )), )) yield* Lens.set(this.fiber, Option.some(fiber)) return subscribable }) } watch( subscribable: Subscribable.Subscribable> ): Effect.Effect, never, QueryClient.QueryClient> { return Effect.gen({ self: this }, function*() { 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> yield* Lens.set(this.latestFinalState, Option.some(final)) if (AsyncResult.isSuccess(final.result)) yield* this.setCacheEntry(final.key, final.result) return final }) } makeCacheKey(key: K): QueryClient.QueryClientCacheKey { return new QueryClient.QueryClientCacheKey(key, this.f as (key: unknown) => Effect.Effect) } getCacheEntry( key: K ): Effect.Effect, never, QueryClient.QueryClient> { return Effect.andThen( Effect.all([ Effect.succeed(this.makeCacheKey(key)), QueryClient.QueryClient, ]), ([key, client]) => client.getCacheEntry(key), ) } setCacheEntry( key: K, result: AsyncResult.Success, ): Effect.Effect { return Effect.flatMap( Effect.all([ Effect.succeed(this.makeCacheKey(key)), QueryClient.QueryClient, ]), ([key, client]) => client.setCacheEntry(key, result, this.staleTime), ) } get invalidateCache(): Effect.Effect { return QueryClient.QueryClient.pipe( Effect.andThen(client => client.invalidateCacheEntries(this.f as (key: unknown) => Effect.Effect)), Effect.provide(this.context), ) } invalidateCacheEntry(key: K): Effect.Effect { return Effect.all([ Effect.succeed(this.makeCacheKey(key)), QueryClient.QueryClient, ]).pipe( Effect.andThen(([key, client]) => client.invalidateCacheEntry(key)), Effect.provide(this.context), ) } } export declare namespace make { export interface Options { readonly key: Lens.Lens, readonly keyEquivalence?: Equivalence.Equivalence, readonly f: (key: K) => Effect.Effect readonly staleTime?: Duration.Input readonly refreshOnWindowFocus?: boolean } } export const make = Effect.fnUntraced(function* ( options: make.Options ): Effect.fn.Return< Query, Cause.NoSuchElementError, Scope.Scope | QueryClient.QueryClient | R > { const client = yield* QueryClient.QueryClient return new QueryImpl( yield* Effect.context(), 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>())), Lens.fromSubscriptionRef(yield* SubscriptionRef.make>({ key: yield* Lens.get(options.key), result: AsyncResult.initial(false), })), Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none>())), yield* Semaphore.make(1), ) }) export const service = ( options: make.Options ): Effect.Effect< Query, Cause.NoSuchElementError, Scope.Scope | QueryClient.QueryClient | R > => Effect.tap( make(options), query => Effect.forkScoped(query.run), )