From 015aa0c8167a99662c947d1549b89bdc6c0a4cfe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Julien=20Valverd=C3=A9?= Date: Thu, 25 Jun 2026 04:37:14 +0200 Subject: [PATCH] Refactor Query --- bun.lock | 3 + packages/effect-fc-next/package.json | 1 + packages/effect-fc-next/src/Query.ts | 387 +++++++++++++++------------ 3 files changed, 224 insertions(+), 167 deletions(-) diff --git a/bun.lock b/bun.lock index cf8ffc5..69dcdf1 100644 --- a/bun.lock +++ b/bun.lock @@ -61,6 +61,7 @@ "effect-lens": "2.0.0-beta.0", }, "devDependencies": { + "@effect/platform-browser": "^4.0.0-beta.85", "@testing-library/react": "^16.3.0", "jsdom": "^26.1.0", "vitest": "^3.2.4", @@ -3062,6 +3063,8 @@ "dot-prop/is-obj": ["is-obj@2.0.0", "", {}, "sha512-drqDG3cbczxxEJRoOXcOjtdp1J/lyp1mNn0xaznRs8+muBhgQcrnbspox5X5fOw0HnMnbfDzvnEMEtqDEJEo8w=="], + "effect-fc-next/@effect/platform-browser": ["@effect/platform-browser@4.0.0-beta.87", "", { "dependencies": { "multipasta": "^0.2.7" }, "peerDependencies": { "effect": "^4.0.0-beta.87" } }, "sha512-sBUUdRt5PjFPLI7+ayxEX/jgnjCbOuc3I//98CmqiwhfgWbCMZ/rofYYb8Zk5tLpNsmdHnWgHWTsBLvH2SFrCw=="], + "effect-fc-next/effect": ["effect@4.0.0-beta.85", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.8.0", "find-my-way-ts": "^0.1.6", "ini": "^7.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^2.0.1", "multipasta": "^0.2.7", "toml": "^4.1.1", "uuid": "^14.0.0", "yaml": "^2.9.0" } }, "sha512-Cjv9YQyv4CiIccmRAQIWAoeESCpCpiuHYY8zb5vqiYs3Ac2yE5RQAnKq7z4Ir/3VJYZb8kxh6rS8czDDnpTdkQ=="], "effect-fc-next/effect-lens": ["effect-lens@2.0.0-beta.0", "", { "peerDependencies": { "effect": "4.0.0-beta.85" } }, "sha512-ntsRQSrzoX+GsNpkNVEGqebl/g1YOEDPIFRAHigf+4P6AtGnnOFcklqDHLrbm4zqnaxuWEgQvvn1J6S3eyC4yw=="], diff --git a/packages/effect-fc-next/package.json b/packages/effect-fc-next/package.json index f29a70b..fa29dbf 100644 --- a/packages/effect-fc-next/package.json +++ b/packages/effect-fc-next/package.json @@ -39,6 +39,7 @@ "clean:modules": "rm -rf node_modules" }, "devDependencies": { + "@effect/platform-browser": "^4.0.0-beta.85", "@testing-library/react": "^16.3.0", "jsdom": "^26.1.0", "vitest": "^3.2.4" diff --git a/packages/effect-fc-next/src/Query.ts b/packages/effect-fc-next/src/Query.ts index e9e317b..670f15d 100644 --- a/packages/effect-fc-next/src/Query.ts +++ b/packages/effect-fc-next/src/Query.ts @@ -1,44 +1,35 @@ -import { - type Cause, - type Context, - type Duration, - Effect, - Equal, - Fiber, - Option, - Pipeable, - Predicate, - type Scope, - Semaphore, - Stream, - SubscriptionRef, -} from "effect" -import { Subscribable } from "effect-lens" +import { type Cause, type Context, Duration, Effect, Equal, 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 Result from "./Result.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 +export interface Query extends Pipeable.Pipeable { readonly [QueryTypeId]: QueryTypeId + readonly context: Context.Context readonly key: Stream.Stream readonly f: (key: K) => Effect.Effect - readonly initialProgress: P - readonly staleTime: Duration.Input + + readonly staleTime: Duration.Duration readonly refreshOnWindowFocus: boolean + readonly latestKey: Subscribable.Subscribable> readonly fiber: Subscribable.Subscribable>> - readonly result: Subscribable.Subscribable> - readonly latestFinalResult: Subscribable.Subscribable>> + readonly result: Subscribable.Subscribable> + readonly latestFinalResult: Subscribable.Subscribable | AsyncResult.Failure>> + readonly run: Effect.Effect - fetch(key: K): Effect.Effect> - fetchSubscribable(key: K): Effect.Effect>> - readonly refresh: Effect.Effect, Cause.NoSuchElementError> - readonly refreshSubscribable: Effect.Effect>, Cause.NoSuchElementError> + fetch(key: K): Effect.Effect | AsyncResult.Failure, Cause.NoSuchElementError> + fetchSubscribable(key: K): Effect.Effect>, Cause.NoSuchElementError> + readonly refresh: Effect.Effect | AsyncResult.Failure, Cause.NoSuchElementError> + readonly refreshSubscribable: Effect.Effect>, Cause.NoSuchElementError> + readonly invalidateCache: Effect.Effect invalidateCacheEntry(key: K): Effect.Effect } @@ -47,42 +38,43 @@ export declare namespace Query { export type AnyKey = readonly any[] } -export class QueryImpl -extends Pipeable.Class implements Query { +export class QueryImpl +extends Pipeable.Class implements Query { readonly [QueryTypeId]: QueryTypeId = QueryTypeId - readonly latestKey: Subscribable.Subscribable> - readonly fiber: Subscribable.Subscribable>> - readonly result: Subscribable.Subscribable> - readonly latestFinalResult: Subscribable.Subscribable>> constructor( readonly context: Context.Context, readonly key: Stream.Stream, readonly f: (key: K) => Effect.Effect, - readonly initialProgress: P, - readonly staleTime: Duration.Input, + + readonly staleTime: Duration.Duration, readonly refreshOnWindowFocus: boolean, - readonly latestKeyRef: SubscriptionRef.SubscriptionRef>, - readonly fiberRef: SubscriptionRef.SubscriptionRef>>, - readonly resultRef: SubscriptionRef.SubscriptionRef>, - readonly latestFinalResultRef: SubscriptionRef.SubscriptionRef>>, + + readonly latestKey: Lens.Lens>, + readonly fiber: Lens.Lens>>, + readonly result: Lens.Lens>, + readonly latestFinalResult: Lens.Lens | AsyncResult.Failure>>, + readonly runSemaphore: Semaphore.Semaphore, ) { super() - this.latestKey = fromSubscriptionRef(latestKeyRef) - this.fiber = fromSubscriptionRef(fiberRef) - this.result = fromSubscriptionRef(resultRef) - this.latestFinalResult = fromSubscriptionRef(latestFinalResultRef) } get run(): Effect.Effect { - const focus = this.refreshOnWindowFocus && typeof window !== "undefined" - ? Stream.runForEach(Stream.fromEventListener(window, "focus"), () => this.refreshSubscribable) - : Effect.succeed(undefined) return Effect.all([ Stream.runForEach(this.key, key => this.fetchSubscribable(key)), - focus, - ], { concurrency: "unbounded", discard: true }).pipe( + + 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), @@ -90,120 +82,159 @@ extends Pipeable.Class implements Query { } get interrupt(): Effect.Effect { - return Effect.flatMap(SubscriptionRef.get(this.fiberRef), Option.match({ + return Effect.flatMap(Lens.get(this.fiber), Option.match({ onSome: Fiber.interrupt, - onNone: () => Effect.succeed(undefined), + onNone: () => Effect.void, })) } - fetch(key: K): Effect.Effect> { - const self = this - return Effect.gen(function*() { - yield* self.interrupt - yield* SubscriptionRef.set(self.latestKeyRef, Option.some(key)) - const previous = yield* SubscriptionRef.get(self.latestFinalResultRef) - const sub = yield* self.startCached(key, Option.isSome(previous) - ? Result.willFetch(previous.value) as Result.Final - : Result.initial()) - return yield* self.watch(key, sub) - }).pipe(Effect.provide(this.context)) + fetch(key: K): Effect.Effect< + AsyncResult.Success | AsyncResult.Failure, + 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.provide(this.context), + ) } - fetchSubscribable(key: K): Effect.Effect>> { - const self = this - return Effect.gen(function*() { - yield* self.interrupt - yield* SubscriptionRef.set(self.latestKeyRef, Option.some(key)) - const previous = yield* SubscriptionRef.get(self.latestFinalResultRef) - const sub = yield* self.startCached(key, Option.isSome(previous) - ? Result.willFetch(previous.value) as Result.Final - : Result.initial()) - yield* Effect.forkScoped(self.watch(key, sub)) - return sub - }).pipe(Effect.provide(this.context)) + fetchSubscribable(key: K): Effect.Effect< + Subscribable.Subscribable>, + 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.provide(this.context), + ) } - get refresh(): Effect.Effect, Cause.NoSuchElementError> { - const self = this - return Effect.gen(function*() { - yield* self.interrupt - const key = yield* Effect.fromOption(yield* SubscriptionRef.get(self.latestKeyRef)) - const previous = yield* SubscriptionRef.get(self.latestFinalResultRef) - const sub = yield* self.startCached(key, Option.isSome(previous) - ? Result.willRefresh(previous.value) as Result.Final - : Result.initial()) - return yield* self.watch(key, sub) - }).pipe(Effect.provide(this.context)) + get refresh(): Effect.Effect | AsyncResult.Failure, 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.flatMap(({ latestKey, subscribable }) => this.watch(latestKey, subscribable)), + Effect.provide(this.context), + ) } - get refreshSubscribable(): Effect.Effect>, Cause.NoSuchElementError> { - const self = this - return Effect.gen(function*() { - yield* self.interrupt - const key = yield* Effect.fromOption(yield* SubscriptionRef.get(self.latestKeyRef)) - const previous = yield* SubscriptionRef.get(self.latestFinalResultRef) - const sub = yield* self.startCached(key, Option.isSome(previous) - ? Result.willRefresh(previous.value) as Result.Final - : Result.initial()) - yield* Effect.forkScoped(self.watch(key, sub)) - return sub - }).pipe(Effect.provide(this.context)) + get refreshSubscribable(): Effect.Effect< + Subscribable.Subscribable>, + 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.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, - ): Effect.Effect>, never, Scope.Scope | QueryClient.QueryClient | R> { + previous: AsyncResult.AsyncResult, + ): Effect.Effect< + Subscribable.Subscribable>, + Cause.NoSuchElementError, + Scope.Scope | QueryClient.QueryClient | R + > { return Effect.flatMap(this.getCacheEntry(key), Option.match({ onSome: entry => Effect.flatMap( QueryClient.isQueryClientCacheEntryStale(entry), isStale => isStale - ? this.start(key, Result.willRefresh(entry.result) as Result.Final) + ? this.start(key, entry.result as AsyncResult.AsyncResult) : Effect.succeed(Subscribable.make({ - get: Effect.succeed(entry.result as Result.Result), - changes: Stream.make(entry.result as Result.Result), + get: Effect.succeed(entry.result as AsyncResult.AsyncResult), + get changes() { return Stream.make(entry.result as AsyncResult.AsyncResult) }, })), ), - onNone: () => this.start(key, initial), + onNone: () => this.start(key, previous), })) } start( key: K, - initial: Result.Initial | Result.Final, - ): Effect.Effect>, never, Scope.Scope | R> { - const self = this - return Effect.gen(function*() { - const [sub, fiber] = yield* Result.unsafeForkEffect( - Effect.onExit(self.f(key), () => Effect.flatMap( - Effect.all([Effect.fiberId, SubscriptionRef.get(self.fiberRef)]), - ([currentFiberId, current]) => Option.match(current, { - onSome: value => Equal.equals(currentFiberId, value.id) - ? SubscriptionRef.set(self.fiberRef, Option.none()) - : Effect.succeed(undefined), - onNone: () => Effect.succeed(undefined), + previous: AsyncResult.AsyncResult, + ): Effect.Effect< + Subscribable.Subscribable>, + never, + Scope.Scope | R + > { + return Effect.gen({ self: this }, function*() { + const state = 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, { + waiting: true, }), + onFailure: v => AsyncResult.failure(v.cause, { + waiting: true, + previousSuccess: v.previousSuccess, + }) + })), + + Effect.onExit(this.f(key), exit => Lens.update( + state, + previous => Exit.match(exit, { + onSuccess: v => AsyncResult.success(v), + onFailure: c => AsyncResult.match(previous, { + onInitial: () => AsyncResult.failure(c), + onSuccess: v => AsyncResult.failure(c, { + previousSuccess: Option.some(v), + }), + onFailure: v => 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 => Equal.equals(fiberId, v.id) + ? Lens.set(this.fiber, Option.none()) + : Effect.void, + onNone: () => Effect.void, + })), )), - { initial, initialProgress: self.initialProgress }, - ) - yield* SubscriptionRef.set(self.fiberRef, Option.some(fiber)) - return sub + )) + + yield* Lens.set(this.fiber, Option.some(fiber)) + return state }) } watch( key: K, - sub: Subscribable.Subscribable>, - ): Effect.Effect, never, QueryClient.QueryClient> { - return Effect.flatMap(sub.get, initial => Stream.runFoldEffect( - Stream.takeUntil(sub.changes, result => Result.isFinal(result) && !Result.hasFlag(result)), - () => initial, - (_, result) => Effect.as(SubscriptionRef.set(this.resultRef, result), result), - ) as Effect.Effect>).pipe( - Effect.tap(result => SubscriptionRef.set(this.latestFinalResultRef, Option.some(result))), - Effect.tap(result => Result.isSuccess(result) - ? Effect.asVoid(this.setCacheEntry(key, result)) - : Effect.succeed(undefined)), + state: Subscribable.Subscribable> + ): Effect.Effect | AsyncResult.Failure, never, QueryClient.QueryClient> { + return state.get.pipe( + Effect.flatMap(initial => Stream.runFoldEffect( + state.changes, + () => initial, + (_, result) => Effect.as(Lens.set(this.result, result), result), + ) as Effect.Effect | AsyncResult.Failure>), + Effect.tap(result => Lens.set(this.latestFinalResult, Option.some(result))), + Effect.tap(result => AsyncResult.isSuccess(result) + ? this.setCacheEntry(key, result) + : Effect.void + ), ) } @@ -211,73 +242,95 @@ extends Pipeable.Class implements Query { return new QueryClient.QueryClientCacheKey(key, this.f as (key: Query.AnyKey) => Effect.Effect) } - getCacheEntry(key: K): Effect.Effect, never, QueryClient.QueryClient> { - return Effect.flatMap(QueryClient.QueryClient, client => client.getCacheEntry(this.makeCacheKey(key))) + 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: Result.Success): Effect.Effect { - return Effect.flatMap(QueryClient.QueryClient, client => client.setCacheEntry(this.makeCacheKey(key), result, this.staleTime)) + 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 Effect.flatMap( - QueryClient.QueryClient, - client => client.invalidateCacheEntries(this.f as (key: Query.AnyKey) => Effect.Effect), - ).pipe(Effect.provide(this.context)) + return QueryClient.QueryClient.pipe( + Effect.andThen(client => client.invalidateCacheEntries(this.f as (key: Query.AnyKey) => Effect.Effect)), + Effect.provide(this.context), + ) } invalidateCacheEntry(key: K): Effect.Effect { - return Effect.flatMap( + return Effect.all([ + Effect.succeed(this.makeCacheKey(key)), QueryClient.QueryClient, - client => client.invalidateCacheEntry(this.makeCacheKey(key)), - ).pipe(Effect.provide(this.context)) + ]).pipe( + Effect.andThen(([key, client]) => client.invalidateCacheEntry(key)), + Effect.provide(this.context), + ) } } + export const isQuery = (u: unknown): u is Query => Predicate.hasProperty(u, QueryTypeId) + export declare namespace make { - export interface Options { + export interface Options { readonly key: Stream.Stream - readonly f: (key: NoInfer) => Effect.Effect>> - readonly initialProgress?: P + readonly f: (key: NoInfer) => Effect.Effect readonly staleTime?: Duration.Input readonly refreshOnWindowFocus?: boolean } } -export const make = Effect.fnUntraced(function* ( - options: make.Options, +export const make = Effect.fnUntraced(function* ( + options: make.Options ): Effect.fn.Return< - Query, P>, - never, - Scope.Scope | QueryClient.QueryClient | KR | Result.forkEffect.OutputContext + Query, + Cause.NoSuchElementError, + Scope.Scope | QueryClient.QueryClient | KR | R > { const client = yield* QueryClient.QueryClient - return new QueryImpl, P>( - yield* Effect.context>(), + + return new QueryImpl( + yield* Effect.context(), options.key, - options.f as any, - options.initialProgress as P, - options.staleTime ?? client.defaultStaleTime, + options.f, + + options.staleTime ? yield* Effect.fromOption(Duration.fromInput(options.staleTime)) : client.defaultStaleTime, options.refreshOnWindowFocus ?? client.defaultRefreshOnWindowFocus, - yield* SubscriptionRef.make(Option.none()), - yield* SubscriptionRef.make(Option.none>()), - yield* SubscriptionRef.make(Result.initial()), - yield* SubscriptionRef.make(Option.none>()), + + Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none())), + Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none>())), + Lens.fromSubscriptionRef(yield* SubscriptionRef.make>(AsyncResult.initial())), + Lens.fromSubscriptionRef(yield* SubscriptionRef.make(Option.none | AsyncResult.Failure>())), + yield* Semaphore.make(1), ) }) -export const service = ( - options: make.Options, +export const service = ( + options: make.Options ): Effect.Effect< - Query, P>, - never, - Scope.Scope | QueryClient.QueryClient | KR | Result.forkEffect.OutputContext -> => Effect.tap(make(options), query => Effect.asVoid(Effect.forkScoped(query.run))) - -const fromSubscriptionRef = (ref: SubscriptionRef.SubscriptionRef): Subscribable.Subscribable => Subscribable.make({ - get: SubscriptionRef.get(ref), - changes: SubscriptionRef.changes(ref), -}) + Query, + Cause.NoSuchElementError, + Scope.Scope | QueryClient.QueryClient | KR | R +> => Effect.tap( + make(options), + query => Effect.forkScoped(query.run), +)