845 lines
27 KiB
TypeScript
845 lines
27 KiB
TypeScript
import {
|
|
useCallback,
|
|
useEffect,
|
|
useMemo,
|
|
useState,
|
|
useSyncExternalStore,
|
|
} from "react";
|
|
import {
|
|
hashKey,
|
|
type QueryClient,
|
|
useMutation,
|
|
useQuery,
|
|
useQueryClient,
|
|
} from "@tanstack/react-query";
|
|
|
|
import {
|
|
deriveAsyncState,
|
|
type AsyncState,
|
|
} from "../../../application/view-models/async-state.ts";
|
|
import type { Result } from "../../../contracts/result.ts";
|
|
import {
|
|
createFailure,
|
|
normalizeUnknownFailure,
|
|
withFailureEffect,
|
|
type AppFailure,
|
|
} from "../../../contracts/errors.ts";
|
|
import type {
|
|
QueryInvalidationCoordinator,
|
|
QueryInvalidationTopic,
|
|
} from "../../../contracts/query-invalidation.ts";
|
|
import type { MutationIntent } from "../../../contracts/mutation-intent.ts";
|
|
import type { CacheScopeSnapshot } from "../../../contracts/server-state-scope.ts";
|
|
import {
|
|
admitQueryResult,
|
|
type BoundMutation,
|
|
type BoundQuery,
|
|
MUTATION_COORDINATOR_BOUNDS,
|
|
type MutationDuplicatePolicy,
|
|
} from "../../../contracts/server-state.ts";
|
|
import { runtimeIdentityToken } from "../../../contracts/query-keys.ts";
|
|
import { useQueryInvalidationCoordinator } from "./query-invalidation-provider.tsx";
|
|
import { useMutationIntentFactory } from "./mutation-intent-provider.tsx";
|
|
import {
|
|
createOptimisticLayerRuntime,
|
|
type OptimisticLayerLease,
|
|
} from "./optimistic-layer-runtime.ts";
|
|
|
|
export type ApplicationResult<Value> = Result<Value>;
|
|
|
|
class ApplicationQueryError extends Error {
|
|
readonly failure: AppFailure;
|
|
|
|
constructor(failure: AppFailure) {
|
|
super(failure.kind);
|
|
this.name = "ApplicationQueryError";
|
|
this.failure = failure;
|
|
}
|
|
}
|
|
|
|
export function useApplicationQuery<Value>(
|
|
options:
|
|
| BoundQuery<Value>
|
|
| Readonly<{
|
|
queryKey: readonly unknown[];
|
|
execute(context: Readonly<{ signal: AbortSignal }>): Promise<
|
|
ApplicationResult<Value>
|
|
>;
|
|
enabled?: boolean;
|
|
}>,
|
|
): Readonly<{
|
|
data: Value | undefined;
|
|
state: AsyncState;
|
|
retry(): Promise<void>;
|
|
}> {
|
|
const { queryKey, execute } = options;
|
|
const enabled = "enabled" in options ? options.enabled ?? true : true;
|
|
const profile = "profile" in options ? options.profile : undefined;
|
|
const scope = "scope" in options ? options.scope : undefined;
|
|
const identity = "identity" in options ? options.identity : undefined;
|
|
const measureResult =
|
|
"measureResult" in options ? options.measureResult : undefined;
|
|
const queryDefinitionId =
|
|
"definitionId" in options ? options.definitionId : "APPLICATION_QUERY";
|
|
const [staleFailure, setStaleFailure] = useState(false);
|
|
useEffect(() => {
|
|
identity?.acquire();
|
|
return () => identity?.release();
|
|
}, [identity]);
|
|
const query = useQuery<Value, ApplicationQueryError>({
|
|
queryKey,
|
|
enabled,
|
|
retry: false,
|
|
staleTime: profile?.staleTimeMs,
|
|
gcTime: profile?.gcTimeMs,
|
|
refetchOnMount: profile?.refetchOnMount,
|
|
refetchOnWindowFocus: profile?.refetchOnFocus,
|
|
refetchOnReconnect: profile?.refetchOnReconnect,
|
|
queryFn: async ({ signal }) => {
|
|
identity?.acquire();
|
|
try {
|
|
if (scope && !scope.isCurrent()) {
|
|
throw new ApplicationQueryError(
|
|
createFailure(
|
|
"SCOPE_GENERATION_CHANGED",
|
|
queryDefinitionId,
|
|
0,
|
|
{ code: "QUERY_SCOPE_STALE" },
|
|
),
|
|
);
|
|
}
|
|
const result = await execute({ signal });
|
|
if (scope && !scope.isCurrent()) {
|
|
throw new ApplicationQueryError(
|
|
createFailure(
|
|
"SCOPE_GENERATION_CHANGED",
|
|
queryDefinitionId,
|
|
0,
|
|
{ code: "QUERY_SCOPE_CHANGED" },
|
|
),
|
|
);
|
|
}
|
|
if (result.ok) {
|
|
if (profile && measureResult) {
|
|
const admission = admitQueryResult(
|
|
measureResult,
|
|
result.value,
|
|
profile,
|
|
);
|
|
if (!admission.ok) {
|
|
throw new ApplicationQueryError(
|
|
createFailure(
|
|
"RESULT_LIMIT_EXCEEDED",
|
|
queryDefinitionId,
|
|
0,
|
|
{ code: admission.code },
|
|
),
|
|
);
|
|
}
|
|
}
|
|
return result.value;
|
|
}
|
|
if (signal.aborted) {
|
|
throw new DOMException("Query cancelled", "AbortError");
|
|
}
|
|
throw new ApplicationQueryError(result.error);
|
|
} catch (error) {
|
|
if (signal.aborted) {
|
|
throw new DOMException("Query cancelled", "AbortError");
|
|
}
|
|
if (error instanceof ApplicationQueryError) throw error;
|
|
throw new ApplicationQueryError(
|
|
normalizeUnknownFailure(error, {
|
|
operationId: "APPLICATION_QUERY",
|
|
}),
|
|
);
|
|
} finally {
|
|
identity?.release();
|
|
}
|
|
},
|
|
});
|
|
const hasData = query.data !== undefined && query.data !== null;
|
|
|
|
useEffect(() => {
|
|
if (query.isError && hasData) {
|
|
setStaleFailure(true);
|
|
} else if (query.isSuccess && !query.isFetching) {
|
|
setStaleFailure(false);
|
|
}
|
|
}, [hasData, query.isError, query.isFetching, query.isSuccess]);
|
|
|
|
const retry = useCallback(async () => {
|
|
setStaleFailure(false);
|
|
await query.refetch();
|
|
}, [query]);
|
|
|
|
return Object.freeze({
|
|
data: query.data,
|
|
state: deriveAsyncState({
|
|
data: query.data,
|
|
isInitialLoading: query.isPending,
|
|
failure:
|
|
!hasData && query.error instanceof ApplicationQueryError
|
|
? query.error.failure
|
|
: undefined,
|
|
isFetching: query.isFetching && !query.isPending,
|
|
isStale: staleFailure,
|
|
isDegraded: staleFailure,
|
|
}),
|
|
retry,
|
|
});
|
|
}
|
|
|
|
type LegacyMutationBase<Input, Value> = Readonly<{
|
|
execute(input: Input): Promise<ApplicationResult<Value>>;
|
|
duplicatePolicy?: MutationDuplicatePolicy;
|
|
invalidate?: readonly QueryInvalidationTopic[];
|
|
currentData?: unknown;
|
|
}>;
|
|
|
|
type LegacyMutationOptions<Input, Value> = LegacyMutationBase<Input, Value> &
|
|
(
|
|
| Readonly<{
|
|
/** Stable logical identity required to retain an unknown effect across remounts. */
|
|
definitionId: string;
|
|
optimistic?: never;
|
|
}>
|
|
| Readonly<{
|
|
definitionId?: string;
|
|
optimistic: Readonly<{
|
|
queryKey: readonly unknown[];
|
|
update(previous: unknown, input: Input): unknown;
|
|
}>;
|
|
}>
|
|
);
|
|
|
|
type ApplicationMutationController<Input, Value> = Readonly<{
|
|
state: AsyncState;
|
|
submit(input: Input): Promise<ApplicationResult<Value>>;
|
|
resolveConflict(): Promise<void>;
|
|
reconcileUnknownEffect(
|
|
resolution: "APPLIED" | "NOT_APPLIED",
|
|
): Promise<void>;
|
|
}>;
|
|
|
|
type MutationExecution<Input> =
|
|
| Readonly<{ kind: "BOUND"; input: Input; intent: MutationIntent }>
|
|
| Readonly<{ kind: "LEGACY"; input: Input }>;
|
|
|
|
type MutationAdmission = {
|
|
sequence: number;
|
|
state: "ACTIVE" | "UNKNOWN" | "RECONCILING" | "SETTLED";
|
|
intent: MutationIntent | null;
|
|
scope: CacheScopeSnapshot | null;
|
|
optimisticLayer: OptimisticLayerLease | null;
|
|
optimisticQueryKey: readonly unknown[] | null;
|
|
readonly invalidate: readonly QueryInvalidationTopic[];
|
|
invalidationCoordinator: QueryInvalidationCoordinator | null;
|
|
};
|
|
|
|
type UnknownEffectChannel = {
|
|
readonly key: string;
|
|
readonly registry: UnknownEffectRegistry;
|
|
readonly owner: QueryClient;
|
|
readonly scope: CacheScopeSnapshot | null;
|
|
nextSequence: number;
|
|
admissions: MutationAdmission[];
|
|
version: number;
|
|
reconciliationInFlight: boolean;
|
|
readonly listeners: Set<() => void>;
|
|
disposeScopeListener: (() => void) | null;
|
|
};
|
|
|
|
type UnknownEffectRegistry = {
|
|
readonly channels: Map<string, UnknownEffectChannel>;
|
|
activeAdmissions: number;
|
|
};
|
|
|
|
const LEGACY_OPTIMISTIC_SCOPE = Object.freeze({
|
|
isCurrent: () => true,
|
|
});
|
|
|
|
export function useApplicationMutation<Input, Value>(
|
|
options: BoundMutation<Input, Value>,
|
|
): ApplicationMutationController<Input, Value>;
|
|
export function useApplicationMutation<Input, Value>(
|
|
options: LegacyMutationOptions<Input, Value>,
|
|
): ApplicationMutationController<Input, Value>;
|
|
export function useApplicationMutation<Input, Value>(
|
|
options: BoundMutation<Input, Value> | LegacyMutationOptions<Input, Value>,
|
|
): ApplicationMutationController<Input, Value> {
|
|
const queryClient = useQueryClient();
|
|
const invalidationCoordinator = useQueryInvalidationCoordinator();
|
|
const mutationIntentFactory = useMutationIntentFactory();
|
|
const invalidate = useMemo(
|
|
() => options.invalidate ?? [],
|
|
[options.invalidate],
|
|
);
|
|
const optimistic = "optimistic" in options ? options.optimistic : undefined;
|
|
const currentData = "currentData" in options ? options.currentData : undefined;
|
|
const definitionId =
|
|
"definitionId" in options
|
|
? options.definitionId ?? "LEGACY_MUTATION"
|
|
: "LEGACY_MUTATION";
|
|
const definitionVersion =
|
|
"definitionVersion" in options ? options.definitionVersion : 0;
|
|
const duplicatePolicy =
|
|
"duplicatePolicy" in options ? options.duplicatePolicy : "REJECT_WHILE_ACTIVE";
|
|
const [conflict, setConflict] = useState<AppFailure | null>(null);
|
|
const scope = "scope" in options ? options.scope : undefined;
|
|
const mutationOperationId =
|
|
"operationId" in options ? options.operationId : definitionId;
|
|
const requiresIdempotencyKey =
|
|
"requiresIdempotencyKey" in options
|
|
? options.requiresIdempotencyKey
|
|
: false;
|
|
const unknownEffectChannelKey = scope
|
|
? `bound:${scope.generation}:${scope.fingerprint}:${definitionId}:${definitionVersion}`
|
|
: `legacy:${definitionId}:${
|
|
optimistic
|
|
? hashKey(optimistic.queryKey)
|
|
: "non-optimistic"
|
|
}`;
|
|
const [acquiredUnknownEffectChannel, setAcquiredUnknownEffectChannel] =
|
|
useState<UnknownEffectChannel | null>(null);
|
|
const unknownEffectChannel =
|
|
acquiredUnknownEffectChannel?.owner === queryClient &&
|
|
acquiredUnknownEffectChannel.key === unknownEffectChannelKey &&
|
|
acquiredUnknownEffectChannel.scope === (scope ?? null)
|
|
? acquiredUnknownEffectChannel
|
|
: null;
|
|
useEffect(() => {
|
|
const acquired = acquireUnknownEffectChannel(
|
|
queryClient,
|
|
unknownEffectChannelKey,
|
|
scope,
|
|
);
|
|
setAcquiredUnknownEffectChannel(acquired);
|
|
return () => {
|
|
if (acquired) {
|
|
releaseUnknownEffectChannelIfUnused(queryClient, acquired);
|
|
}
|
|
};
|
|
}, [queryClient, scope, unknownEffectChannelKey]);
|
|
const subscribeToUnknownEffects = useCallback(
|
|
(listener: () => void) => {
|
|
if (!unknownEffectChannel) return () => {};
|
|
unknownEffectChannel.listeners.add(listener);
|
|
return () => {
|
|
unknownEffectChannel.listeners.delete(listener);
|
|
releaseUnknownEffectChannelIfUnused(
|
|
queryClient,
|
|
unknownEffectChannel,
|
|
);
|
|
};
|
|
},
|
|
[queryClient, unknownEffectChannel],
|
|
);
|
|
useSyncExternalStore(
|
|
subscribeToUnknownEffects,
|
|
() => unknownEffectChannel?.version ?? 0,
|
|
() => unknownEffectChannel?.version ?? 0,
|
|
);
|
|
const mutation = useMutation<
|
|
Value,
|
|
ApplicationQueryError,
|
|
MutationExecution<Input>
|
|
>({
|
|
retry: false,
|
|
mutationFn: async (execution) => {
|
|
const input = execution.input;
|
|
if (scope && !scope.isCurrent()) {
|
|
throw new ApplicationQueryError(
|
|
createFailure(
|
|
"SCOPE_GENERATION_CHANGED",
|
|
definitionId,
|
|
0,
|
|
{ code: "MUTATION_SCOPE_STALE", effect: "NOT_STARTED" },
|
|
),
|
|
);
|
|
}
|
|
const result =
|
|
"scope" in options
|
|
? await options.execute(input, {
|
|
signal: options.scope.signal,
|
|
intent:
|
|
execution.kind === "BOUND"
|
|
? execution.intent
|
|
: (() => {
|
|
throw new TypeError(
|
|
"Bound mutation execution requires an intent.",
|
|
);
|
|
})(),
|
|
})
|
|
: await options.execute(input);
|
|
if (scope && !scope.isCurrent()) {
|
|
const effect =
|
|
!result.ok && result.error.effect !== undefined
|
|
? result.error.effect
|
|
: "MAYBE_APPLIED";
|
|
throw new ApplicationQueryError(
|
|
createFailure(
|
|
"SCOPE_GENERATION_CHANGED",
|
|
definitionId,
|
|
0,
|
|
{ code: "MUTATION_SCOPE_CHANGED", effect },
|
|
),
|
|
);
|
|
}
|
|
if (result.ok) return result.value;
|
|
throw new ApplicationQueryError(result.error);
|
|
},
|
|
});
|
|
|
|
const submit = useCallback(
|
|
(input: Input): Promise<ApplicationResult<Value>> => {
|
|
let identity: string;
|
|
let identityLease: ReturnType<
|
|
NonNullable<typeof scope>["identities"]["intern"]
|
|
> | null = null;
|
|
try {
|
|
if (scope) {
|
|
if (!scope.isCurrent()) {
|
|
return Promise.resolve({
|
|
ok: false,
|
|
error: createFailure(
|
|
"SCOPE_GENERATION_CHANGED",
|
|
definitionId,
|
|
0,
|
|
{ code: "MUTATION_SCOPE_STALE", effect: "NOT_STARTED" },
|
|
),
|
|
});
|
|
}
|
|
identityLease = scope.identities.intern(input);
|
|
identityLease.acquire();
|
|
identity = `${scope.fingerprint}:${definitionId}:${identityLease.token}`;
|
|
} else {
|
|
identity = `${definitionId}:${runtimeIdentityToken(input)}`;
|
|
}
|
|
} catch (error) {
|
|
return Promise.resolve({
|
|
ok: false,
|
|
error: withFailureEffect(
|
|
normalizeUnknownFailure(error, {
|
|
operationId: definitionId,
|
|
}),
|
|
"NOT_STARTED",
|
|
),
|
|
});
|
|
}
|
|
const active = mutationExecutions(queryClient).get(identity) as
|
|
| Promise<ApplicationResult<Value>>
|
|
| undefined;
|
|
if (active && duplicatePolicy === "JOIN_IDENTICAL") {
|
|
identityLease?.release();
|
|
return active;
|
|
}
|
|
if (active && duplicatePolicy === "REJECT_WHILE_ACTIVE") {
|
|
identityLease?.release();
|
|
return Promise.resolve({
|
|
ok: false,
|
|
error: createFailure(
|
|
"DUPLICATE_IN_FLIGHT",
|
|
definitionId,
|
|
0,
|
|
{ code: "DUPLICATE_IN_FLIGHT", effect: "NOT_STARTED" },
|
|
),
|
|
});
|
|
}
|
|
if (
|
|
!unknownEffectChannel ||
|
|
unknownEffectChannel.registry.activeAdmissions >=
|
|
MUTATION_COORDINATOR_BOUNDS.activeIntentsTotal
|
|
) {
|
|
identityLease?.release();
|
|
return Promise.resolve({
|
|
ok: false,
|
|
error: createFailure(
|
|
"IDENTITY_INTERN_LIMIT_EXCEEDED",
|
|
definitionId,
|
|
0,
|
|
{
|
|
code: "UNKNOWN_EFFECT_CHANNEL_LIMIT_EXCEEDED",
|
|
effect: "NOT_STARTED",
|
|
},
|
|
),
|
|
});
|
|
}
|
|
const admission: MutationAdmission = {
|
|
sequence: unknownEffectChannel.nextSequence++,
|
|
state: "ACTIVE",
|
|
intent: null,
|
|
scope: scope ?? null,
|
|
optimisticLayer: null,
|
|
optimisticQueryKey: optimistic?.queryKey ?? null,
|
|
invalidate,
|
|
invalidationCoordinator: invalidationCoordinator ?? null,
|
|
};
|
|
unknownEffectChannel.admissions.push(admission);
|
|
unknownEffectChannel.registry.activeAdmissions += 1;
|
|
notifyUnknownEffectChannel(unknownEffectChannel);
|
|
setConflict(null);
|
|
mutation.reset();
|
|
|
|
const pending = (async (): Promise<ApplicationResult<Value>> => {
|
|
let execution: MutationExecution<Input>;
|
|
if (scope) {
|
|
const intent = mutationIntentFactory.create({
|
|
operationId: mutationOperationId,
|
|
canonicalInputIdentity: identity,
|
|
requiresIdempotencyKey,
|
|
});
|
|
admission.intent = intent;
|
|
execution = Object.freeze({ kind: "BOUND", input, intent });
|
|
} else {
|
|
execution = Object.freeze({ kind: "LEGACY", input });
|
|
}
|
|
const mutationLease =
|
|
invalidate.length === 0
|
|
? null
|
|
: invalidationCoordinator?.beginMutation(invalidate);
|
|
if (invalidate.length > 0 && !mutationLease) {
|
|
throw new Error("Query invalidation coordinator is not installed.");
|
|
}
|
|
try {
|
|
let optimisticLayer: OptimisticLayerLease | null = null;
|
|
if (optimistic) {
|
|
await queryClient.cancelQueries({
|
|
queryKey: optimistic.queryKey,
|
|
exact: true,
|
|
});
|
|
optimisticLayer = optimisticLayers(queryClient).begin(
|
|
optimistic.queryKey,
|
|
input,
|
|
optimistic.update,
|
|
scope ?? LEGACY_OPTIMISTIC_SCOPE,
|
|
);
|
|
admission.optimisticLayer = optimisticLayer;
|
|
}
|
|
|
|
let value: Value;
|
|
try {
|
|
value = await mutation.mutateAsync(execution);
|
|
} catch (error: unknown) {
|
|
const reportedFailure =
|
|
error instanceof ApplicationQueryError
|
|
? error.failure
|
|
: normalizeUnknownFailure(error, {
|
|
operationId: "APPLICATION_MUTATION",
|
|
});
|
|
const failure = mutationFailureWithEffect(reportedFailure);
|
|
switch (failure.effect) {
|
|
case "NOT_STARTED":
|
|
case "NOT_APPLIED":
|
|
optimisticLayer?.rollback();
|
|
if (failure.kind === "CONFLICT") setConflict(failure);
|
|
settleMutationAdmission(unknownEffectChannel, admission);
|
|
break;
|
|
case "APPLIED_CONFIRMED":
|
|
optimisticLayer?.commit();
|
|
if (!scope || scope.isCurrent()) {
|
|
try {
|
|
await invalidationCoordinator?.invalidate(invalidate);
|
|
} catch {
|
|
// A confirmed command remains committed if refresh fails.
|
|
}
|
|
}
|
|
settleMutationAdmission(unknownEffectChannel, admission);
|
|
break;
|
|
case "MAYBE_APPLIED":
|
|
optimisticLayer?.markUncertain();
|
|
if (!scope || scope.isCurrent()) {
|
|
admission.state = "UNKNOWN";
|
|
notifyUnknownEffectChannel(unknownEffectChannel);
|
|
} else {
|
|
settleMutationAdmission(unknownEffectChannel, admission);
|
|
}
|
|
break;
|
|
}
|
|
return { ok: false, error: failure };
|
|
}
|
|
|
|
optimisticLayer?.commit();
|
|
try {
|
|
await invalidationCoordinator?.invalidate(invalidate);
|
|
} catch {
|
|
// Cache refresh remains best effort after the server has committed.
|
|
}
|
|
settleMutationAdmission(unknownEffectChannel, admission);
|
|
return { ok: true, value };
|
|
} finally {
|
|
try {
|
|
await mutationLease?.release();
|
|
} catch {
|
|
// A cache coordination defect cannot change the committed command.
|
|
}
|
|
}
|
|
})()
|
|
.catch((error: unknown) => {
|
|
const failure = withFailureEffect(
|
|
normalizeUnknownFailure(error, {
|
|
operationId: "APPLICATION_MUTATION",
|
|
}),
|
|
"NOT_STARTED",
|
|
);
|
|
settleMutationAdmission(unknownEffectChannel, admission);
|
|
if (failure.kind === "CONFLICT") setConflict(failure);
|
|
return { ok: false as const, error: failure };
|
|
})
|
|
.finally(() => {
|
|
identityLease?.release();
|
|
if (mutationExecutions(queryClient).get(identity) === pending) {
|
|
mutationExecutions(queryClient).delete(identity);
|
|
}
|
|
});
|
|
if (duplicatePolicy !== "ALLOW_PARALLEL") {
|
|
mutationExecutions(queryClient).set(identity, pending);
|
|
}
|
|
return pending;
|
|
},
|
|
[
|
|
invalidate,
|
|
invalidationCoordinator,
|
|
mutation,
|
|
optimistic,
|
|
queryClient,
|
|
definitionId,
|
|
duplicatePolicy,
|
|
mutationIntentFactory,
|
|
mutationOperationId,
|
|
requiresIdempotencyKey,
|
|
scope,
|
|
unknownEffectChannel,
|
|
],
|
|
);
|
|
|
|
const resolveConflict = useCallback(async () => {
|
|
setConflict(null);
|
|
mutation.reset();
|
|
if (invalidate.length > 0 && !invalidationCoordinator) {
|
|
throw new Error("Query invalidation coordinator is not installed.");
|
|
}
|
|
await invalidationCoordinator?.invalidate(invalidate);
|
|
}, [invalidate, invalidationCoordinator, mutation]);
|
|
|
|
const reconcileUnknownEffect = useCallback(
|
|
async (resolution: "APPLIED" | "NOT_APPLIED") => {
|
|
if (
|
|
!unknownEffectChannel ||
|
|
unknownEffectChannel.reconciliationInFlight
|
|
) {
|
|
return;
|
|
}
|
|
const pending = unknownEffectChannel.admissions[0];
|
|
if (!pending || pending.state !== "UNKNOWN") return;
|
|
if (pending.scope && !pending.intent) return;
|
|
unknownEffectChannel.reconciliationInFlight = true;
|
|
pending.state = "RECONCILING";
|
|
notifyUnknownEffectChannel(unknownEffectChannel);
|
|
try {
|
|
pending.optimisticLayer?.reconcile(resolution);
|
|
if (
|
|
resolution === "APPLIED" &&
|
|
(!pending.scope || pending.scope.isCurrent())
|
|
) {
|
|
try {
|
|
await pending.invalidationCoordinator?.invalidate(
|
|
pending.invalidate,
|
|
);
|
|
} catch {
|
|
// Explicit reconciliation remains settled if refresh fails.
|
|
}
|
|
}
|
|
} finally {
|
|
if (pending.scope && !pending.scope.isCurrent()) {
|
|
removeAdmissionOptimisticQuery(queryClient, pending);
|
|
}
|
|
settleMutationAdmission(unknownEffectChannel, pending);
|
|
if (
|
|
!unknownEffectChannel.admissions.some(
|
|
(admission) => admission.state === "ACTIVE",
|
|
)
|
|
) {
|
|
mutation.reset();
|
|
}
|
|
// Keep the head intent locked through the current event turn so two
|
|
// clicks cannot consume two FIFO records when reconciliation itself
|
|
// has no asynchronous invalidation work.
|
|
await Promise.resolve();
|
|
unknownEffectChannel.reconciliationInFlight = false;
|
|
notifyUnknownEffectChannel(unknownEffectChannel);
|
|
}
|
|
},
|
|
[mutation, queryClient, unknownEffectChannel],
|
|
);
|
|
|
|
const unknownEffectHead = unknownEffectChannel?.admissions[0];
|
|
|
|
return Object.freeze({
|
|
state: deriveAsyncState({
|
|
data: currentData ?? true,
|
|
isMutationPending:
|
|
mutation.isPending ||
|
|
unknownEffectHead?.state === "ACTIVE" ||
|
|
unknownEffectHead?.state === "RECONCILING",
|
|
hasMutationEffectUnknown: unknownEffectHead?.state === "UNKNOWN",
|
|
hasMutationConflict: conflict !== null,
|
|
}),
|
|
submit,
|
|
resolveConflict,
|
|
reconcileUnknownEffect,
|
|
});
|
|
}
|
|
|
|
function mutationFailureWithEffect(failure: AppFailure): AppFailure {
|
|
switch (failure.effect) {
|
|
case "NOT_STARTED":
|
|
case "NOT_APPLIED":
|
|
case "APPLIED_CONFIRMED":
|
|
case "MAYBE_APPLIED":
|
|
return withFailureEffect(failure, failure.effect);
|
|
case "NOT_APPLICABLE":
|
|
case undefined:
|
|
return withFailureEffect(failure, "MAYBE_APPLIED");
|
|
}
|
|
}
|
|
|
|
const UNKNOWN_EFFECT_CHANNELS = new WeakMap<
|
|
object,
|
|
UnknownEffectRegistry
|
|
>();
|
|
|
|
function acquireUnknownEffectChannel(
|
|
queryClient: QueryClient,
|
|
key: string,
|
|
scope: CacheScopeSnapshot | undefined,
|
|
): UnknownEffectChannel | null {
|
|
let registry = UNKNOWN_EFFECT_CHANNELS.get(queryClient);
|
|
if (!registry) {
|
|
registry = { channels: new Map(), activeAdmissions: 0 };
|
|
UNKNOWN_EFFECT_CHANNELS.set(queryClient, registry);
|
|
}
|
|
const { channels } = registry;
|
|
const existing = channels.get(key);
|
|
if (existing) return existing.scope === (scope ?? null) ? existing : null;
|
|
if (
|
|
channels.size >= MUTATION_COORDINATOR_BOUNDS.activeDefinitionsPerRuntime
|
|
) {
|
|
return null;
|
|
}
|
|
const created: UnknownEffectChannel = {
|
|
key,
|
|
registry,
|
|
owner: queryClient,
|
|
scope: scope ?? null,
|
|
nextSequence: 1,
|
|
admissions: [],
|
|
version: 0,
|
|
reconciliationInFlight: false,
|
|
listeners: new Set(),
|
|
disposeScopeListener: null,
|
|
};
|
|
channels.set(key, created);
|
|
if (scope) {
|
|
const discardScope = () => {
|
|
for (const admission of created.admissions) {
|
|
removeAdmissionOptimisticQuery(queryClient, admission);
|
|
markMutationAdmissionSettled(created, admission);
|
|
}
|
|
created.admissions = [];
|
|
notifyUnknownEffectChannel(created);
|
|
releaseUnknownEffectChannelIfUnused(queryClient, created);
|
|
};
|
|
scope.signal.addEventListener("abort", discardScope, { once: true });
|
|
created.disposeScopeListener = () =>
|
|
scope.signal.removeEventListener("abort", discardScope);
|
|
if (scope.signal.aborted || !scope.isCurrent()) discardScope();
|
|
}
|
|
return created;
|
|
}
|
|
|
|
function releaseUnknownEffectChannelIfUnused(
|
|
queryClient: QueryClient,
|
|
channel: UnknownEffectChannel,
|
|
): void {
|
|
if (channel.admissions.length > 0 || channel.listeners.size > 0) return;
|
|
const registry = UNKNOWN_EFFECT_CHANNELS.get(queryClient);
|
|
const channels = registry?.channels;
|
|
if (channels?.get(channel.key) !== channel) return;
|
|
channel.disposeScopeListener?.();
|
|
channels.delete(channel.key);
|
|
if (channels.size === 0 && registry?.activeAdmissions === 0) {
|
|
UNKNOWN_EFFECT_CHANNELS.delete(queryClient);
|
|
}
|
|
}
|
|
|
|
function notifyUnknownEffectChannel(channel: UnknownEffectChannel): void {
|
|
channel.version += 1;
|
|
for (const listener of [...channel.listeners]) listener();
|
|
}
|
|
|
|
function settleMutationAdmission(
|
|
channel: UnknownEffectChannel,
|
|
admission: MutationAdmission,
|
|
): void {
|
|
markMutationAdmissionSettled(channel, admission);
|
|
while (channel.admissions[0]?.state === "SETTLED") {
|
|
channel.admissions.shift();
|
|
}
|
|
notifyUnknownEffectChannel(channel);
|
|
releaseUnknownEffectChannelIfUnused(channel.owner, channel);
|
|
}
|
|
|
|
function markMutationAdmissionSettled(
|
|
channel: UnknownEffectChannel,
|
|
admission: MutationAdmission,
|
|
): void {
|
|
if (admission.state === "SETTLED") return;
|
|
admission.state = "SETTLED";
|
|
channel.registry.activeAdmissions = Math.max(
|
|
0,
|
|
channel.registry.activeAdmissions - 1,
|
|
);
|
|
}
|
|
|
|
function removeAdmissionOptimisticQuery(
|
|
queryClient: QueryClient,
|
|
admission: MutationAdmission,
|
|
): void {
|
|
if (!admission.optimisticQueryKey) return;
|
|
queryClient.removeQueries({
|
|
queryKey: admission.optimisticQueryKey,
|
|
exact: true,
|
|
});
|
|
}
|
|
|
|
const RUNTIME_MUTATION_EXECUTIONS = new WeakMap<
|
|
object,
|
|
Map<string, Promise<ApplicationResult<unknown>>>
|
|
>();
|
|
|
|
function mutationExecutions(
|
|
owner: object,
|
|
): Map<string, Promise<ApplicationResult<unknown>>> {
|
|
const existing = RUNTIME_MUTATION_EXECUTIONS.get(owner);
|
|
if (existing) return existing;
|
|
const created = new Map<string, Promise<ApplicationResult<unknown>>>();
|
|
RUNTIME_MUTATION_EXECUTIONS.set(owner, created);
|
|
return created;
|
|
}
|
|
|
|
const OPTIMISTIC_LAYER_RUNTIMES = new WeakMap<
|
|
object,
|
|
ReturnType<typeof createOptimisticLayerRuntime>
|
|
>();
|
|
|
|
function optimisticLayers(
|
|
queryClient: Parameters<typeof createOptimisticLayerRuntime>[0],
|
|
): ReturnType<typeof createOptimisticLayerRuntime> {
|
|
const existing = OPTIMISTIC_LAYER_RUNTIMES.get(queryClient);
|
|
if (existing) return existing;
|
|
const created = createOptimisticLayerRuntime(queryClient);
|
|
OPTIMISTIC_LAYER_RUNTIMES.set(queryClient, created);
|
|
return created;
|
|
}
|