diff --git a/.changeset/calm-pallets-listen.md b/.changeset/calm-pallets-listen.md new file mode 100644 index 000000000..d015de043 --- /dev/null +++ b/.changeset/calm-pallets-listen.md @@ -0,0 +1,5 @@ +--- +"effect-app": patch +--- + +Derive scoped dependency reads and writes from typed query filters and annotated model relationships, including previous aliases. diff --git a/.changeset/live-query-dependencies.md b/.changeset/live-query-dependencies.md new file mode 100644 index 000000000..d744cdb24 --- /dev/null +++ b/.changeset/live-query-dependencies.md @@ -0,0 +1,5 @@ +--- +"@effect-app/vue": minor +--- + +Add query-owned live invalidation with recorded dependency filtering, connect-before-fetch coordination, race buffering, and configurable client-side coalescing. diff --git a/packages/effect-app/src/Model.ts b/packages/effect-app/src/Model.ts index 5696dee3a..fdfa6b568 100644 --- a/packages/effect-app/src/Model.ts +++ b/packages/effect-app/src/Model.ts @@ -1,4 +1,5 @@ export * from "./Model/dsl.ts" export * as Q from "./Model/query.ts" export { makeRepo } from "./Model/Repository.ts" +export { repositoryDependency } from "./Model/Repository.ts" export { type RegisteredRepository, RepositoryRegistry, RepositoryRegistryLive } from "./Model/Repository.ts" diff --git a/packages/effect-app/src/Model/Repository.ts b/packages/effect-app/src/Model/Repository.ts index 93a7dbf6f..f68b0b3e0 100644 --- a/packages/effect-app/src/Model/Repository.ts +++ b/packages/effect-app/src/Model/Repository.ts @@ -1,3 +1,4 @@ +export { repositoryDependency } from "./Repository/dependency.ts" export * from "./Repository/ext.ts" export * from "./Repository/legacy.ts" export { makeRepo } from "./Repository/makeRepo.ts" diff --git a/packages/effect-app/src/Model/Repository/dependency.ts b/packages/effect-app/src/Model/Repository/dependency.ts new file mode 100644 index 000000000..adc7e64d9 --- /dev/null +++ b/packages/effect-app/src/Model/Repository/dependency.ts @@ -0,0 +1,32 @@ +import * as S from "../../Schema.ts" +import * as SchemaAST from "../../SchemaAST.ts" + +const RepositoryDependencyAnnotation = "effect-app/Repository/dependency" + +/** Marks an encoded model field as an identity alias for repository invalidation. */ +export const repositoryDependency = (schema: Schema): Schema["Rebuild"] => + S.annotateEncoded({ [RepositoryDependencyAnnotation]: true })(schema) + +/** @internal */ +export const repositoryDependencyPaths = (schema: S.Schema): readonly string[] => { + const visit = (ast: SchemaAST.AST, path: readonly string[]): readonly string[] => { + const annotations = ast.checks?.at(-1)?.annotations ?? ast.annotations + if (annotations?.[RepositoryDependencyAnnotation] === true) return [path.join(".")] + if (SchemaAST.isDeclaration(ast)) return ast.typeParameters.flatMap((parameter) => visit(parameter, path)) + if (SchemaAST.isUnion(ast)) return ast.types.flatMap((member) => visit(member, path)) + if (SchemaAST.isObjects(ast)) { + return ast.propertySignatures.flatMap((property) => + typeof property.name === "string" ? visit(property.type, [...path, property.name]) : [] + ) + } + if (SchemaAST.isArrays(ast)) { + return [ + ...ast.elements.flatMap((element, index) => visit(element, [...path, String(index)])), + ...ast.rest.flatMap((element) => visit(element, [...path, "-1"])) + ] + } + return [] + } + + return [...new Set(visit(SchemaAST.toEncoded(schema.ast), []))] +} diff --git a/packages/effect-app/src/Model/Repository/internal/internal.ts b/packages/effect-app/src/Model/Repository/internal/internal.ts index b0c3d4c87..c53f3eadc 100644 --- a/packages/effect-app/src/Model/Repository/internal/internal.ts +++ b/packages/effect-app/src/Model/Repository/internal/internal.ts @@ -25,8 +25,10 @@ import { type Codec, NonNegativeInt } from "../../../Schema.ts" import * as SchemaAST from "../../../SchemaAST.ts" import { setupRequestContextFromCurrent } from "../../../setupRequest.ts" import { type FilterArgs, getContextMap, type PersistenceModelType, type StoreConfig, storeId, StoreMaker } from "../../../Store.ts" +import type { FilterResult } from "../../filter/filterApi.ts" import type { FieldValues } from "../../filter/types.ts" import * as Q from "../../query.ts" +import { repositoryDependencyPaths } from "../dependency.ts" import type { ChangeFeed, ChangeFeedEvent, Repository } from "../service.ts" import { ValidationError, ValidationResult } from "../validation.ts" @@ -147,16 +149,55 @@ export function makeRepoInternal< ) const store = yield* mkStore(args.makeInitial, args.config) + const dependencyPaths = repositoryDependencyPaths(schema) const recordRead = DataDependencies.readRepo(name) const entityDependency = (ids: NonEmptyReadonlyArray) => DataDependencies.repo(name, [String(ids[0]), ...ids.slice(1).map(String)]) + const valuesAtPath = (value: unknown, path: string): readonly unknown[] => { + if (path.length === 0) return [value] + const [head, ...tail] = path.split(".") + if (head === undefined) return [value] + if (head === "-1") { + return globalThis.Array.isArray(value) + ? value.flatMap((item) => valuesAtPath(item, tail.join("."))) + : [] + } + return typeof value === "object" && value !== null + ? valuesAtPath((value as Record)[head], tail.join(".")) + : [] + } + const dependencyIds = (item: T): NonEmptyReadonlyArray => { + const configured = dependencyPaths + .flatMap((path) => valuesAtPath(item, String(path))) + .filter((value): value is string => typeof value === "string") ?? [] + const ids = args.dependencyIds?.(item) ?? [String(item[idKey]), ...configured] + return [ids[0], ...ids.slice(1)] + } const itemDependencies = (item: T) => { - const ids = args.dependencyIds?.(item) ?? [String(item[idKey])] + const ids = dependencyIds(item) return [ DataDependencies.repo(name, ids), ...(args.additionalWriteDependencies?.(item) ?? []) ] } + const queryDependencyIds = (filter: readonly FilterResult[] | undefined): readonly string[] => { + if (!filter) return [] + const paths = new Set([String(idKey), ...dependencyPaths]) + const visit = (items: readonly FilterResult[]): readonly string[] => { + if (items.some((item) => item.t === "or" || item.t === "or-scope")) return [] + return items.flatMap((item) => { + if ("result" in item) return visit(item.result) + if (!paths.has(item.path)) return [] + const value: unknown = item.value + if (item.op === "eq" && typeof value === "string") return [value] + if (item.op === "in" && globalThis.Array.isArray(value)) { + return value.filter((entry): entry is string => typeof entry === "string") + } + return [] + }) + } + return [...new Set(visit(filter))] + } const recordEntityRead = (id: T[IdKey]) => DataDependencies.read(entityDependency([id])) const recordEntityWrite = (ids: NonEmptyReadonlyArray) => DataDependencies.write(entityDependency(ids)) @@ -166,12 +207,6 @@ export function makeRepoInternal< DataDependencies.write, { discard: true } ) - const recordAdditionalItemWrites = (items: ReadonlyArray) => - Effect.forEach( - DataDependencies.merge(new Set(items.flatMap((item) => args.additionalWriteDependencies?.(item) ?? []))), - DataDependencies.write, - { discard: true } - ) const cms = Effect.map(getContextMap.pipe(Effect.orDie), (_) => ({ get: (id: string) => _.get(`${name}.${id}`), set: (id: string, etag: string | undefined) => _.set(`${name}.${id}`, etag) @@ -353,7 +388,7 @@ export function makeRepoInternal< const it = Chunk.fromIterable(items) if (Chunk.isNonEmpty(it)) { const values = Chunk.toReadonlyArray(it) - const previous = args.additionalWriteDependencies + const previous = args.additionalWriteDependencies || args.dependencyIds || dependencyPaths.length > 0 ? yield* loadExistingItems(values.map((item) => item[idKey])) : [] yield* recordItemWrite([values[0], ...values.slice(1), ...previous]) @@ -414,8 +449,9 @@ export function makeRepoInternal< return } yield* recordEntityWrite(ids) - if (args.additionalWriteDependencies) { - yield* loadExistingItems(ids).pipe(Effect.flatMap(recordAdditionalItemWrites)) + if (args.additionalWriteDependencies || args.dependencyIds || dependencyPaths.length > 0) { + const previous = yield* loadExistingItems(ids) + if (Array.isReadonlyArrayNonEmpty(previous)) yield* recordItemWrite(previous) } const { set } = yield* cms const eids = yield* Effect.forEach(ids, (_) => encodeIdOnly(_ as any)).pipe(Effect.orDie) @@ -485,6 +521,10 @@ export function makeRepoInternal< ): Effect.Effect> } = ((q: Q.QAll) => { const a = Q.toFilter(q, schema) + const scopedIds = queryDependencyIds(a.filter) + const recordQueryRead = Array.isReadonlyArrayNonEmpty(scopedIds) + ? DataDependencies.read(DataDependencies.repo(name, scopedIds)) + : recordRead // Mode dispatch — see `Q.project` JSDoc for the contract: // aggregate: GROUP BY + aggregate functions at DB level; decode raw rows with schema; SchemaError surfaces. // project : decode raw encoded rows with schema; no PM reverse-mapping; SchemaError surfaces. @@ -563,7 +603,7 @@ export function makeRepoInternal< "db.response.returned_rows": Array.isArray(r) ? r.length : 1 }) ), - Effect.tap(() => recordRead), + Effect.tap(() => recordQueryRead), Effect.withSpan("Repository.query", { kind: "client", attributes: { "app.entity": name } diff --git a/packages/infra/test/repository-ext.test.ts b/packages/infra/test/repository-ext.test.ts index e5f1201af..ea7958bff 100644 --- a/packages/infra/test/repository-ext.test.ts +++ b/packages/infra/test/repository-ext.test.ts @@ -3,7 +3,7 @@ import * as DataDependencies from "effect-app/DataDependencies" import * as Effect from "effect-app/Effect" import * as Layer from "effect-app/Layer" import { Q } from "effect-app/Model" -import { makeRepo } from "effect-app/Model/Repository" +import { makeRepo, repositoryDependency } from "effect-app/Model/Repository" import { RepositoryRegistryLive } from "effect-app/Model/Repository/Registry" import * as S from "effect-app/Schema" import { setupRequestContextFromCurrent } from "effect-app/setupRequest" @@ -16,6 +16,16 @@ class BatchItem extends S.Class("BatchItem")({ label: S.String }) {} +class DependencyItem extends S.Class("DependencyItem")({ + id: S.String, + label: repositoryDependency(S.StringId) +}) {} + +class NestedDependencyItem extends S.Class("NestedDependencyItem")({ + id: S.String, + parts: S.Array(S.Struct({ id: repositoryDependency(S.String) })) +}) {} + const TestStoreLive = Layer.merge(MemoryStoreLive, RepositoryRegistryLive) const A = S.TaggedStruct("A", { id: S.String }) @@ -218,7 +228,7 @@ describe("repository ext save/remove batching", () => { .toEqual(DataDependencies.repo("DependencyItem")) }) - it.effect("matches an explicit query scope to a write alias", () => + it.effect("derives matching read and write scopes from a schema annotation", () => Effect .gen(function*() { const readsRef = yield* Ref.make(DataDependencies.empty()) @@ -227,17 +237,42 @@ describe("repository ext save/remove batching", () => { yield* Effect .gen(function*() { - const repo = yield* makeRepo("DependencyItem", BatchItem, { - dependencyIds: (item) => [item.id, `alias-${item.id}`] - }) - yield* repo.save(new BatchItem({ id: "1", label: "one" })) - yield* repo.all.pipe(DataDependencies.withRepoReadScope("DependencyItem", ["alias-1"])) + const repo = yield* makeRepo("DependencyItem", DependencyItem, {}) + yield* repo.save(new DependencyItem({ id: "1", label: S.StringId("label-one") })) + yield* repo.query(Q.where("label", "label-one")) }) .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder)) - expect(yield* Ref.get(readsRef)).toEqual(new Set([DataDependencies.repo("DependencyItem", ["alias-1"])])) + expect(yield* Ref.get(readsRef)).toEqual(new Set([DataDependencies.repo("DependencyItem", ["label-one"])])) + expect(yield* Ref.get(writesRef)).toEqual( + new Set([DataDependencies.repo("DependencyItem", ["1", "label-one"])]) + ) + }) + .pipe( + setupRequestContextFromCurrent(), + Effect.provide(TestStoreLive) + )) + + it.effect("derives a nested annotated relationship scope from whereSome", () => + Effect + .gen(function*() { + const readsRef = yield* Ref.make(DataDependencies.empty()) + const writesRef = yield* Ref.make(DataDependencies.empty()) + const recorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef) + + yield* Effect + .gen(function*() { + const repo = yield* makeRepo("NestedDependencyItem", NestedDependencyItem, {}) + yield* repo.save(new NestedDependencyItem({ id: "root", parts: [{ id: "part-1" }] })) + yield* repo.query(Q.whereSome("parts", Q.where("id", "part-1"))) + }) + .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder)) + + expect(yield* Ref.get(readsRef)).toEqual( + new Set([DataDependencies.repo("NestedDependencyItem", ["part-1"])]) + ) expect(yield* Ref.get(writesRef)).toEqual( - new Set([DataDependencies.repo("DependencyItem", ["1", "alias-1"])]) + new Set([DataDependencies.repo("NestedDependencyItem", ["root", "part-1"])]) ) }) .pipe( @@ -245,6 +280,38 @@ describe("repository ext save/remove batching", () => { Effect.provide(TestStoreLive) )) + it.effect("records previous and next relationship aliases", () => + Effect + .gen(function*() { + const readsRef = yield* Ref.make(DataDependencies.empty()) + const writesRef = yield* Ref.make(DataDependencies.empty()) + const recorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef) + + yield* Effect + .gen(function*() { + const repo = yield* makeRepo("DependencyItem", BatchItem, { + dependencyIds: (item) => [item.id, `label-${item.label}`] + }) + yield* repo.save(new BatchItem({ id: "1", label: "old" })) + yield* recorder.drainWrites + + yield* repo.save(new BatchItem({ id: "1", label: "new" })) + expect(yield* recorder.drainWrites).toEqual( + new Set([DataDependencies.repo("DependencyItem", ["1", "label-new", "label-old"])]) + ) + + yield* repo.removeById("1") + expect(yield* recorder.drainWrites).toEqual( + new Set([DataDependencies.repo("DependencyItem", ["1", "label-new"])]) + ) + }) + .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder)) + }) + .pipe( + setupRequestContextFromCurrent(), + Effect.provide(TestStoreLive) + )) + it.effect("records repository and affected-query write dependencies", () => Effect .gen(function*() { diff --git a/packages/vue/src/atomQuery.ts b/packages/vue/src/atomQuery.ts index e48275004..e2a76899f 100644 --- a/packages/vue/src/atomQuery.ts +++ b/packages/vue/src/atomQuery.ts @@ -32,8 +32,9 @@ import { isHttpClientError } from "effect/unstable/http/HttpClientError" import * as AsyncResult from "effect/unstable/reactivity/AsyncResult" import * as Atom from "effect/unstable/reactivity/Atom" import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry" -import { clearQueryReadDependencies, setQueryReadDependencies } from "./dependencyMetadata.ts" +import { clearQueryReadDependencies, getQueryReadDependencies, setQueryReadDependencies } from "./dependencyMetadata.ts" import { reportRuntimeError } from "./lib.ts" +import { beginLiveQueryFetch, endLiveQueryFetch, type LiveQueryOptions, registerLiveQuery } from "./liveQueryInvalidation.ts" /** All non-empty prefixes of a key, longest last. `[a,b,c]` -> `[[a],[a,b],[a,b,c]]`. */ const prefixesOf = (key: ReadonlyArray): ReadonlyArray> => @@ -226,6 +227,8 @@ export interface AtomQueryOptions { readonly structuralSharing?: boolean /** poll: re-fetch every N ms (tanstack refetchInterval). */ readonly refetchInterval?: number + /** Refresh when writes intersect the dependencies recorded by this query. */ + readonly live?: boolean | LiveQueryOptions } const defaults = { staleTime: Duration.seconds(5), gcTime: Duration.minutes(5) } @@ -236,6 +239,11 @@ export interface AtomQueryMetadata { const atomQueryMetadata = new WeakMap>, AtomQueryMetadata>() const atomQueryParentSpans = new WeakMap>, Tracer.AnySpan>() +const atomQueryKeys = new WeakMap>, ReadonlyArray>() + +export const queryKeyForAtom = ( + atom: Atom.Atom> +): ReadonlyArray | undefined => atomQueryKeys.get(atom) const atomSpanTarget = (atom: Atom.Atom>) => { let target = atom @@ -351,11 +359,24 @@ const recoverStuckWaitingOnMount = export const withQueryOptions = ( self: Atom.Atom>, - opts: AtomQueryOptions = {} + opts: AtomQueryOptions = {}, + liveKey?: ReadonlyArray ): Atom.Atom> => { setAtomQueryMetadata(self, opts) const staleTime: Duration.Input = opts.staleTime ?? defaults.staleTime let atom = self + if (opts.live && liveKey !== undefined) { + const liveOptions = opts.live === true ? {} : opts.live + atom = Atom.transform(atom, (get) => { + const unregister = registerLiveQuery( + liveKey, + () => getQueryReadDependencies(liveKey), + liveOptions + ) + get.addFinalizer(unregister) + return get(self) + }, { initialValueTarget: self }) + } const revalidateOnFocus = opts.revalidateOnFocus ?? true atom = Atom.swr({ staleTime, @@ -442,31 +463,38 @@ export const buildQueryFamily = ( // A fetch is now running: the atom is not stuck, and any pending recovery is fulfilled. fetchState.recovering = false fetchState.inFlight++ - const recordReads = Effect.gen(function*() { - const readsRef = yield* Ref.make(DataDependencies.empty()) - const writesRef = yield* Ref.make(DataDependencies.empty()) - const recorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef) - const result = yield* self - .handler(input) - .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder)) - lastReads = yield* Ref.get(readsRef) - setQueryReadDependencies(fullKey, lastReads) - return result - }) - const effect = recordReads.pipe( - Effect.retry({ times: 5, while: isRetryable }), - Effect.tapCauseIf(Cause.hasDies, (cause) => reportRuntimeError(cause)), - // On exit, the compute is no longer in-flight. An interrupt (subscriber lost interest / a - // superseding refresh) may leave the result at `waiting`; with `inFlight` back at 0 that - // reads as "stuck", so the next mount recovers it (`recoverStuckWaitingOnMount`). We do not - // re-fire here — the interrupt was intentional; recovery is driven by a genuine (re)mount. - Effect.onExit(() => - Effect.sync(() => { - fetchState.inFlight = Math.max(0, fetchState.inFlight - 1) - }) - ), - Effect.withSpan(`query ${self.id}`, {}, { captureStackTrace: false }) - ) + const recordReads = Effect + .gen(function*() { + const readsRef = yield* Ref.make(DataDependencies.empty()) + const writesRef = yield* Ref.make(DataDependencies.empty()) + const recorder = DataDependencies.makeDataDependencyRecorder(readsRef, writesRef) + const result = yield* self + .handler(input) + .pipe(Effect.provideService(DataDependencies.DataDependencyRecorder, recorder)) + lastReads = yield* Ref.get(readsRef) + setQueryReadDependencies(fullKey, lastReads) + return result + }) + let liveFetch = false + const effect = Effect + .gen(function*() { + liveFetch = yield* Effect.promise(() => beginLiveQueryFetch(fullKey)) + return yield* recordReads.pipe(Effect.retry({ times: 5, while: isRetryable })) + }) + .pipe( + Effect.ensuring(Effect.sync(() => endLiveQueryFetch(liveFetch))), + Effect.tapCauseIf(Cause.hasDies, (cause) => reportRuntimeError(cause)), + // On exit, the compute is no longer in-flight. An interrupt (subscriber lost interest / a + // superseding refresh) may leave the result at `waiting`; with `inFlight` back at 0 that + // reads as "stuck", so the next mount recovers it (`recoverStuckWaitingOnMount`). We do not + // re-fire here — the interrupt was intentional; recovery is driven by a genuine (re)mount. + Effect.onExit(() => + Effect.sync(() => { + fetchState.inFlight = Math.max(0, fetchState.inFlight - 1) + }) + ), + Effect.withSpan(`query ${self.id}`, {}, { captureStackTrace: false }) + ) const parentSpan = takeAtomQueryParentSpan(atom) return parentSpan === undefined ? effect @@ -500,6 +528,7 @@ export const buildQueryFamily = ( const result = setAtomQueryMetadata(Atom.withLabel(`query-cache:${self.id}`)(writableWithTarget)) // Key the fetch state by the atom `withQueryOptions` receives, so its mount hook can find it. queryFetchStates.set(result, fetchState) + atomQueryKeys.set(result, fullKey) return result }) } diff --git a/packages/vue/src/dependencyMetadata.ts b/packages/vue/src/dependencyMetadata.ts index 68d86ed73..363c8d228 100644 --- a/packages/vue/src/dependencyMetadata.ts +++ b/packages/vue/src/dependencyMetadata.ts @@ -21,6 +21,10 @@ export const clearQueryReadDependencies = (key: ReadonlyArray) => { readDependencies.delete(Hash.hash(key)) } +export const getQueryReadDependencies = ( + key: ReadonlyArray +): DataDependencies.DataDependencies => readDependencies.get(Hash.hash(key))?.reads ?? DataDependencies.empty() + /** * Reactivity keys of every live query whose recorded read-dependencies intersect this * mutation's `writeDependencies`. Returned keys are passed to `invalidateAndAwait`, refreshing diff --git a/packages/vue/src/liveQueryInvalidation.ts b/packages/vue/src/liveQueryInvalidation.ts new file mode 100644 index 000000000..234fcfcc9 --- /dev/null +++ b/packages/vue/src/liveQueryInvalidation.ts @@ -0,0 +1,138 @@ +import { DataDependencies } from "effect-app/client" +import * as Hash from "effect/Hash" + +export interface LiveQueryOptions { + /** Fixed maximum delay from the first matching write. Zero reacts immediately. */ + readonly maxDelayMs?: number +} + +export interface LiveQueryInvalidationSource { + /** Resolve only after events received after this point can no longer be missed. */ + readonly ready: () => Promise + readonly subscribe: ( + onWrites: (writes: DataDependencies.DataDependencies) => void, + onReset: () => void + ) => () => void + readonly invalidate: (keys: ReadonlyArray>) => void +} + +interface LiveQueryEntry { + readonly key: ReadonlyArray + readonly reads: () => DataDependencies.DataDependencies + observers: number + readonly delays: Map + timer: ReturnType | undefined +} + +const entries = new Map() +let source: LiveQueryInvalidationSource | undefined +let unsubscribe: (() => void) | undefined +let discovering = 0 +let bufferedWrites = DataDependencies.empty() + +const invalidateEntry = (entry: LiveQueryEntry) => { + entry.timer = undefined + source?.invalidate([entry.key]) +} + +const schedule = (entry: LiveQueryEntry) => { + if (entry.timer !== undefined) return + const maxDelayMs = Math.min(...entry.delays.keys()) + if (maxDelayMs <= 0) return invalidateEntry(entry) + entry.timer = setTimeout(() => invalidateEntry(entry), maxDelayMs) +} + +const dispatch = (writes: DataDependencies.DataDependencies) => { + if (!DataDependencies.isNonEmpty(writes)) return + for (const entry of entries.values()) { + if (DataDependencies.intersects(entry.reads(), writes)) schedule(entry) + } +} + +const receive = (writes: DataDependencies.DataDependencies) => { + if (discovering > 0) { + bufferedWrites = DataDependencies.merge(bufferedWrites, writes) + return + } + dispatch(writes) +} + +const reset = () => { + bufferedWrites = DataDependencies.empty() + for (const entry of entries.values()) { + if (DataDependencies.isNonEmpty(entry.reads())) schedule(entry) + } +} + +const ensureSubscribed = () => { + if (unsubscribe === undefined && source !== undefined && entries.size > 0) { + unsubscribe = source.subscribe(receive, reset) + } +} + +export const configureLiveQueryInvalidation = (next: LiveQueryInvalidationSource) => { + unsubscribe?.() + unsubscribe = undefined + source = next + ensureSubscribed() +} + +export const registerLiveQuery = ( + key: ReadonlyArray, + reads: () => DataDependencies.DataDependencies, + options: LiveQueryOptions +) => { + const hash = Hash.hash(key) + const delay = options.maxDelayMs ?? 0 + const current = entries.get(hash) + if (current !== undefined) { + current.observers++ + current.delays.set(delay, (current.delays.get(delay) ?? 0) + 1) + } else { + entries.set(hash, { + key, + reads, + observers: 1, + delays: new Map([[delay, 1]]), + timer: undefined + }) + } + ensureSubscribed() + + return () => { + const entry = entries.get(hash) + if (entry === undefined) return + const delayObservers = entry.delays.get(delay) ?? 0 + if (delayObservers <= 1) entry.delays.delete(delay) + else entry.delays.set(delay, delayObservers - 1) + if (--entry.observers > 0) return + if (entry.timer !== undefined) clearTimeout(entry.timer) + entries.delete(hash) + if (entries.size === 0) { + unsubscribe?.() + unsubscribe = undefined + bufferedWrites = DataDependencies.empty() + } + } +} + +export const beginLiveQueryFetch = async (key: ReadonlyArray) => { + if (!entries.has(Hash.hash(key)) || source === undefined) return false + discovering++ + try { + await source.ready() + return true + } catch (error) { + discovering-- + throw error + } +} + +export const endLiveQueryFetch = (wasLive: boolean) => { + if (!wasLive) return + discovering = Math.max(0, discovering - 1) + if (discovering > 0 || !DataDependencies.isNonEmpty(bufferedWrites)) return + const writes = bufferedWrites + bufferedWrites = DataDependencies.empty() + dispatch(writes) +} diff --git a/packages/vue/src/query.ts b/packages/vue/src/query.ts index 302996d1f..6f5211415 100644 --- a/packages/vue/src/query.ts +++ b/packages/vue/src/query.ts @@ -17,7 +17,8 @@ import * as Stream from "effect/Stream" import * as AsyncResult from "effect/unstable/reactivity/AsyncResult" import * as Atom from "effect/unstable/reactivity/Atom" import { computed, type ComputedRef, effectScope, type MaybeRefOrGetter, onBeforeUnmount, onMounted, onScopeDispose, ref, toValue, type WatchSource } from "vue" -import { type AtomClientRuntime, type AtomQueryOptions, awaitAtomResult, buildQueryFamily, buildStreamQueryFamily, disabledQueryAtom, isStaleResult, refreshAtomWithCurrentSpan, staleTimeMsOf, withQueryOptions } from "./atomQuery.ts" +import { type AtomClientRuntime, type AtomQueryOptions, awaitAtomResult, buildQueryFamily, buildStreamQueryFamily, disabledQueryAtom, isStaleResult, queryKeyForAtom, refreshAtomWithCurrentSpan, staleTimeMsOf, withQueryOptions } from "./atomQuery.ts" +import type { LiveQueryOptions } from "./liveQueryInvalidation.ts" import { latestDefined } from "./suspense.ts" // --- minimal local types (replacing the former @tanstack/vue-query type imports) --- @@ -201,6 +202,7 @@ export interface CustomUseQueryOptions< readonly structuralSharing?: boolean /** poll: re-fetch every N ms (tanstack refetchInterval) */ readonly refetchInterval?: number + readonly live?: boolean | LiveQueryOptions readonly select?: (data: TQueryFnData) => TData /** accepted for source compatibility; not used by the atom engine */ readonly retry?: boolean | number @@ -253,6 +255,7 @@ export interface AtomQueryNewOptions TData } @@ -263,6 +266,7 @@ export interface AtomStreamQueryOptions { readonly revalidateOnFocus?: boolean readonly refetchOnWindowFocus?: boolean readonly refreshEvery?: number + readonly live?: boolean | LiveQueryOptions readonly refetchInterval?: number } @@ -322,6 +326,7 @@ const normalizeQueryOptions = (options?: { readonly structuralSharing?: boolean readonly refetchInterval?: number readonly refreshEvery?: number + readonly live?: boolean | LiveQueryOptions }): AtomQueryOptions => { const out: { staleTime?: number @@ -329,6 +334,7 @@ const normalizeQueryOptions = (options?: { revalidateOnFocus?: boolean structuralSharing?: boolean refetchInterval?: number + live?: boolean | LiveQueryOptions } = {} if (options?.staleTime !== undefined) out.staleTime = options.staleTime const gcTime = options?.idleTTL ?? options?.gcTime @@ -338,6 +344,7 @@ const normalizeQueryOptions = (options?: { if (options?.structuralSharing !== undefined) out.structuralSharing = options.structuralSharing const refetchInterval = options?.refreshEvery ?? options?.refetchInterval if (refetchInterval !== undefined) out.refetchInterval = refetchInterval + if (options?.live !== undefined) out.live = options.live return out } @@ -513,8 +520,10 @@ const observedAtom = ( readonly structuralSharing?: boolean readonly refetchInterval?: number readonly refreshEvery?: number + readonly live?: boolean | LiveQueryOptions } -): Atom.Atom> => withQueryOptions(atom, normalizeQueryOptions(options)) +): Atom.Atom> => + withQueryOptions(atom, normalizeQueryOptions(options), queryKeyForAtom(atom)) const observedStreamAtom = ( atom: Atom.Writable, void>, diff --git a/packages/vue/test/liveQueryInvalidation.test.ts b/packages/vue/test/liveQueryInvalidation.test.ts new file mode 100644 index 000000000..9c2df6e9b --- /dev/null +++ b/packages/vue/test/liveQueryInvalidation.test.ts @@ -0,0 +1,108 @@ +import { DataDependencies } from "effect-app/client" +import { afterEach, describe, expect, it, vi } from "vitest" +import { beginLiveQueryFetch, configureLiveQueryInvalidation, endLiveQueryFetch, registerLiveQuery } from "../src/liveQueryInvalidation.js" + +const repo = DataDependencies.repo("Inventory", ["item-1"]) +const otherId = DataDependencies.repo("Inventory", ["item-2"]) + +describe("live query invalidation", () => { + const cleanups: Array<() => void> = [] + + afterEach(() => { + while (cleanups.length > 0) cleanups.pop()?.() + vi.useRealTimers() + }) + + it("invalidates only live queries whose recorded reads intersect", () => { + const invalidations: Array>> = [] + let receive = (_writes: DataDependencies.DataDependencies) => {} + configureLiveQueryInvalidation({ + ready: () => Promise.resolve(), + subscribe: (onWrites) => { + receive = onWrites + return () => {} + }, + invalidate: (keys) => invalidations.push(keys) + }) + + const key = ["$Inventory", "List", undefined] + cleanups.push(registerLiveQuery(key, () => new Set([repo]), {})) + receive(new Set([otherId])) + expect(invalidations).toEqual([]) + + receive(new Set([repo])) + expect(invalidations).toEqual([[key]]) + }) + + it("replays writes received while the initial query discovers its dependencies", async () => { + const invalidations: Array>> = [] + let receive = (_writes: DataDependencies.DataDependencies) => {} + let open = () => {} + const ready = new Promise((resolve) => open = resolve) + configureLiveQueryInvalidation({ + ready: () => ready, + subscribe: (onWrites) => { + receive = onWrites + return () => {} + }, + invalidate: (keys) => invalidations.push(keys) + }) + + const key = ["$Inventory", "Get", { id: "item-1" }] + let reads = DataDependencies.empty() + cleanups.push(registerLiveQuery(key, () => reads, {})) + + const starting = beginLiveQueryFetch(key) + receive(new Set([repo])) + open() + const wasLive = await starting + reads = new Set([repo]) + endLiveQueryFetch(wasLive) + + expect(invalidations).toEqual([[key]]) + }) + + it("coalesces matching writes using the query's maximum delay", () => { + vi.useFakeTimers() + const invalidations: Array>> = [] + let receive = (_writes: DataDependencies.DataDependencies) => {} + configureLiveQueryInvalidation({ + ready: () => Promise.resolve(), + subscribe: (onWrites) => { + receive = onWrites + return () => {} + }, + invalidate: (keys) => invalidations.push(keys) + }) + + const key = ["$Inventory", "List", undefined] + cleanups.push(registerLiveQuery(key, () => new Set([repo]), { maxDelayMs: 2_000 })) + receive(new Set([repo])) + receive(new Set([repo])) + vi.advanceTimersByTime(1_999) + expect(invalidations).toEqual([]) + vi.advanceTimersByTime(1) + expect(invalidations).toEqual([[key]]) + }) + + it("fails safe by invalidating every live query when the source resets", () => { + const invalidations: Array>> = [] + let reset = () => {} + configureLiveQueryInvalidation({ + ready: () => Promise.resolve(), + subscribe: (_onWrites, onReset) => { + reset = onReset + return () => {} + }, + invalidate: (keys) => invalidations.push(keys) + }) + + const first = ["$Inventory", "List", undefined] + const second = ["$Orders", "List", undefined] + cleanups.push(registerLiveQuery(first, () => new Set([repo]), {})) + cleanups.push(registerLiveQuery(second, () => new Set([otherId]), {})) + reset() + + expect(invalidations).toEqual([[first], [second]]) + }) +})