1import type { QueryPersistence } from "./queryPersistence.ts";
2import { createCommandPermissions } from "./commandPermissions.ts";
3import { followStreamInEnvironment } from "./environmentStreams.ts";
4import { reuseUnchanged } from "./serverConfigProjection.ts";
5export { runStreamInEnvironment, followStreamInEnvironment } from "./environmentStreams.ts";
6import {
7 type ClientGuardedRpcTag,
8 EnvironmentId,
9 type EnvironmentId as EnvironmentIdType,
10} from "@t3tools/contracts";
11import * as Cause from "effect/Cause";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import { identity } from "effect/Function";
15import * as Option from "effect/Option";
16import * as Schema from "effect/Schema";
17import * as Stream from "effect/Stream";
18import * as SubscriptionRef from "effect/SubscriptionRef";
19import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
20
21import type { ConnectionAttemptError, SupervisorConnectionState } from "../connection/model.ts";
22import * as EnvironmentRegistry from "../connection/registry.ts";
23import {
24 type EnvironmentRpcInput,
25 type EnvironmentRpcSuccess,
26 type EnvironmentRpcFailure,
27 type EnvironmentRpcStreamFailure,
28 type EnvironmentRpcStreamValue,
29 type EnvironmentSubscriptionRpcTag,
30 type EnvironmentUnaryRpcTag,
31 EnvironmentRpcUnavailableError,
32 request,
33 requestGuarded,
34 RpcPermissionGuard,
35 subscribe,
36} from "../rpc/client.ts";
37import * as EnvironmentSupervisor from "../connection/supervisor.ts";
38import type * as RpcSession from "../rpc/session.ts";
39
40const isEnvironmentRpcUnavailable = Schema.is(EnvironmentRpcUnavailableError);
41
42/**
43 * Holds back the unavailable failure of a query whose session went away. The
44 * query atom reruns when it sees that session change, so settling first would
45 * show a failure for a query that is about to retry. While `session` is still
46 * current no rerun is coming, and the failure stands.
47 */
48const pendingOnceSessionReplaced =
49 (session: RpcSession.RpcSession) =>
50 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
51 effect.pipe(
52 // The query's error type is generic, so its tags are unknown to `catchTags`.
53 Effect.catchIf(isEnvironmentRpcUnavailable, (error) =>
54 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
55 Effect.flatMap((supervisor) => SubscriptionRef.get(supervisor.session)),
56 Effect.flatMap((current) =>
57 Option.isSome(current) && current.value === session ? Effect.fail(error) : Effect.never,
58 ),
59 ),
60 ),
61 );
62
63interface EnvironmentAtomOptions<Input, A, E, R> {
64 readonly label: string;
65 readonly execute: (input: Input) => Effect.Effect<A, E, R>;
66 readonly scheduler?: AtomCommandScheduler;
67 readonly concurrency?: AtomCommandConcurrency<{
68 readonly environmentId: EnvironmentIdType;
69 readonly input: Input;
70 }>;
71}
72
73interface EnvironmentCommandAtomOptions<Input, A, E, R> extends Omit<
74 EnvironmentAtomOptions<Input, A, E, R>,
75 "execute"
76> {
77 readonly execute: (
78 input: Input,
79 registry: AtomRegistry.AtomRegistry,
80 environmentId: EnvironmentIdType,
81 ) => Effect.Effect<A, E, R>;
82}
83
84interface EnvironmentQueryAtomOptions<Input, A, E, R> extends Omit<
85 EnvironmentAtomOptions<Input, A, E, R>,
86 "execute"
87> {
88 /**
89 * `emit` shows a provisional value while the query keeps waiting. The settled
90 * result replaces it, so catch failures that should keep it on screen.
91 */
92 readonly execute: (
93 input: Input,
94 emit: (value: A) => Effect.Effect<void>,
95 previous: Option.Option<A>,
96 ) => Effect.Effect<A, E, R>;
97 readonly persistence?: QueryPersistence<A>;
98 readonly staleTimeMs?: number;
99 readonly idleTtlMs?: number;
100 /**
101 * Check staleness each time the query is mounted again. By default the idle
102 * TTL keeps the stale-while-revalidate node warm, so a remount within it
103 * reads the cached value without checking. The cached value still shows
104 * while the query revalidates.
105 */
106 readonly revalidateOnRemount?: boolean;
107 readonly refreshIntervalMs?: number;
108 readonly refreshTrigger?: (target: {
109 readonly environmentId: EnvironmentIdType;
110 readonly input: Input;
111 }) => Atom.Atom<unknown> | undefined;
112 readonly isFinal?: (value: A) => boolean;
113 /**
114 * Runs where each server answer lands, before readers see it, so derived records (summaries,
115 * snapshots, watches) come from the producer instead of from components watching the data.
116 * Cached and emitted provisional values are not answers.
117 */
118 readonly onAnswer?: QueryAnswerHandler<Input, A>;
119}
120
121export type QueryAnswerHandler<Input, A> = (
122 target: { readonly environmentId: EnvironmentIdType; readonly input: Input },
123 value: A,
124 registry: AtomRegistry.AtomRegistry,
125) => Effect.Effect<void>;
126
127interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> {
128 readonly sensitiveInput?: boolean;
129 readonly label: string;
130 readonly subscribe: (input: Input) => Stream.Stream<A, E, R>;
131 readonly completeWhen?: (value: A) => boolean;
132 readonly idleTtlMs?: number;
133 /** Keeps a completed value this long after its last reader leaves, so reopening skips a resubscribe. */
134 readonly completedTtlMs?: number;
135}
136
137export type SettledAsyncResult<A, E> = AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>;
138
139export type AtomCommandResult<A, E> = SettledAsyncResult<A, E>;
140
141export type AtomCommandSuccess<R> = R extends AtomCommandResult<infer A, infer _E> ? A : never;
142
143export type AtomCommandFailure<R> = R extends AtomCommandResult<infer _A, infer E> ? E : never;
144
145export interface AtomCommandOptions {
146 readonly label?: string;
147 readonly reportFailure?: boolean;
148 readonly reportDefect?: boolean;
149}
150
151export interface AtomCommandReporter {
152 readonly warn: (message: string, cause: Cause.Cause<unknown>) => void;
153 readonly error: (message: string, cause: Cause.Cause<unknown>) => void;
154}
155
156export interface AtomCommand<W, A, E> {
157 readonly label: string;
158 readonly run: (registry: AtomRegistry.AtomRegistry, input: W) => Promise<AtomCommandResult<A, E>>;
159}
160
161export type AtomCommandConcurrency<W> =
162 | { readonly mode: "parallel" }
163 | {
164 readonly mode: "serial" | "singleFlight" | "latest";
165 readonly key: (input: W) => string;
166 };
167
168interface AtomCommandSchedulerState {
169 readonly serial: Map<string, Promise<unknown>>;
170 readonly singleFlight: Map<string, Promise<unknown>>;
171 readonly latest: Map<string, AtomCommandLatestLane>;
172}
173
174interface AtomCommandLatestBatch {
175 execute: () => Promise<AtomCommandResult<unknown, unknown>>;
176 readonly resolve: Array<(result: AtomCommandResult<unknown, unknown>) => void>;
177}
178
179interface AtomCommandLatestLane {
180 running: boolean;
181 pending: AtomCommandLatestBatch | undefined;
182}
183
184export interface AtomCommandScheduler {
185 readonly schedule: <W, A, E>(
186 registry: AtomRegistry.AtomRegistry,
187 concurrency: AtomCommandConcurrency<W>,
188 input: W,
189 execute: () => Promise<AtomCommandResult<A, E>>,
190 ) => Promise<AtomCommandResult<A, E>>;
191}
192
193async function settleAtomCommandResult<A, E>(
194 execute: () => Promise<AtomCommandResult<A, E>>,
195): Promise<AtomCommandResult<A, E>> {
196 try {
197 return await execute();
198 } catch (defect) {
199 return AsyncResult.failure(Cause.die(defect));
200 }
201}
202
203export function createAtomCommandScheduler(): AtomCommandScheduler {
204 const registryStates = new WeakMap<AtomRegistry.AtomRegistry, AtomCommandSchedulerState>();
205
206 const stateFor = (registry: AtomRegistry.AtomRegistry): AtomCommandSchedulerState => {
207 const existing = registryStates.get(registry);
208 if (existing !== undefined) {
209 return existing;
210 }
211 const state: AtomCommandSchedulerState = {
212 serial: new Map(),
213 singleFlight: new Map(),
214 latest: new Map(),
215 };
216 registryStates.set(registry, state);
217 return state;
218 };
219
220 return {
221 schedule: <W, A, E>(
222 registry: AtomRegistry.AtomRegistry,
223 concurrency: AtomCommandConcurrency<W>,
224 input: W,
225 execute: () => Promise<AtomCommandResult<A, E>>,
226 ): Promise<AtomCommandResult<A, E>> => {
227 if (concurrency.mode === "parallel") {
228 return execute();
229 }
230
231 const key = concurrency.key(input);
232 const state = stateFor(registry);
233 if (concurrency.mode === "singleFlight") {
234 const existing = state.singleFlight.get(key) as
235 | Promise<AtomCommandResult<A, E>>
236 | undefined;
237 if (existing !== undefined) {
238 return existing;
239 }
240 const current = execute();
241 state.singleFlight.set(key, current);
242 void current.then(
243 () => {
244 if (state.singleFlight.get(key) === current) {
245 state.singleFlight.delete(key);
246 }
247 },
248 () => {
249 if (state.singleFlight.get(key) === current) {
250 state.singleFlight.delete(key);
251 }
252 },
253 );
254 return current;
255 }
256
257 if (concurrency.mode === "serial") {
258 const previous = state.serial.get(key);
259 const current = previous === undefined ? execute() : previous.then(execute, execute);
260 state.serial.set(key, current);
261 void current.then(
262 () => {
263 if (state.serial.get(key) === current) {
264 state.serial.delete(key);
265 }
266 },
267 () => {
268 if (state.serial.get(key) === current) {
269 state.serial.delete(key);
270 }
271 },
272 );
273 return current;
274 }
275
276 let lane = state.latest.get(key);
277 if (lane === undefined) {
278 lane = { running: false, pending: undefined };
279 state.latest.set(key, lane);
280 }
281 const activeLane = lane;
282
283 const result = new Promise<AtomCommandResult<A, E>>((resolve) => {
284 if (activeLane.pending === undefined) {
285 activeLane.pending = {
286 execute: execute as () => Promise<AtomCommandResult<unknown, unknown>>,
287 resolve: [resolve as (result: AtomCommandResult<unknown, unknown>) => void],
288 };
289 return;
290 }
291 activeLane.pending.execute = execute as () => Promise<AtomCommandResult<unknown, unknown>>;
292 activeLane.pending.resolve.push(
293 resolve as (result: AtomCommandResult<unknown, unknown>) => void,
294 );
295 });
296
297 if (!activeLane.running) {
298 activeLane.running = true;
299 void (async () => {
300 while (activeLane.pending !== undefined) {
301 const batch = activeLane.pending;
302 activeLane.pending = undefined;
303 let batchResult: AtomCommandResult<unknown, unknown>;
304 try {
305 batchResult = await batch.execute();
306 } catch (defect) {
307 batchResult = AsyncResult.failure(Cause.die(defect));
308 }
309 for (const resolve of batch.resolve) {
310 resolve(batchResult);
311 }
312 }
313 activeLane.running = false;
314 if (state.latest.get(key) === activeLane) {
315 state.latest.delete(key);
316 }
317 })();
318 }
319
320 return result;
321 },
322 };
323}
324
325export const scheduleAtomCommandEffect =
326 /* Every atom command passes through this scheduler; the command owns its tracing boundary. */
327 Effect.fnUntraced(function* <W, A, E, R>(
328 registry: AtomRegistry.AtomRegistry,
329 scheduler: AtomCommandScheduler,
330 concurrency: AtomCommandConcurrency<W>,
331 input: W,
332 effect: Effect.Effect<A, E, R>,
333 ): Effect.fn.Return<A, E, R> {
334 const context = yield* Effect.context<R>();
335 const result = yield* Effect.promise((signal) =>
336 scheduler.schedule<W, A, E>(registry, concurrency, input, async () => {
337 const exit = await Effect.runPromiseExitWith(context)(effect, { signal });
338 return Exit.isSuccess(exit)
339 ? AsyncResult.success(exit.value)
340 : AsyncResult.failure(exit.cause);
341 }),
342 );
343 return result._tag === "Success" ? result.value : yield* Effect.failCause(result.cause);
344 });
345
346export async function runAtomCommand<W, A, E>(
347 registry: AtomRegistry.AtomRegistry,
348 command: AtomCommand<W, A, E>,
349 input: W,
350 options: AtomCommandOptions = {},
351 reporter: AtomCommandReporter = console,
352): Promise<AtomCommandResult<A, E>> {
353 const result = await settleAtomCommandResult(() => command.run(registry, input));
354 reportAtomCommandResult(result, { ...options, label: options.label ?? command.label }, reporter);
355 return result;
356}
357
358export function mapAtomCommandResult<A, E, B>(
359 result: AtomCommandResult<A, E>,
360 map: (value: A) => B,
361): AtomCommandResult<B, E> {
362 return result._tag === "Success"
363 ? AsyncResult.success(map(result.value))
364 : AsyncResult.failure(result.cause);
365}
366
367export function isAtomCommandInterrupted(result: AtomCommandResult<unknown, unknown>): boolean {
368 return result._tag === "Failure" && Cause.hasInterruptsOnly(result.cause);
369}
370
371export function squashAtomCommandFailure(result: {
372 readonly cause: Cause.Cause<unknown>;
373}): unknown {
374 return Cause.squash(result.cause);
375}
376
377export async function settleAsyncResult<A, E>(
378 execute: () => Promise<Exit.Exit<A, E>>,
379): Promise<SettledAsyncResult<A, E>> {
380 try {
381 return AsyncResult.fromExit(await execute());
382 } catch (defect) {
383 return AsyncResult.failure(Cause.die(defect));
384 }
385}
386
387async function executeAtomCommand<A, E>(
388 execute: () => Promise<Exit.Exit<A, E>>,
389 options: AtomCommandOptions = {},
390 reporter: AtomCommandReporter = console,
391): Promise<AtomCommandResult<A, E>> {
392 const result = await settleAsyncResult(execute);
393 reportAtomCommandResult(result, options, reporter);
394 return result;
395}
396
397export interface AtomQueryOptions extends AtomCommandOptions {
398 /**
399 * Force a fresh execution instead of accepting a value the atom already
400 * holds (e.g. an SWR-cached result within its stale window). Used by
401 * verification flows where a cached failure must not satisfy a retry.
402 */
403 readonly refresh?: boolean;
404 readonly signal?: AbortSignal;
405}
406
407export async function executeAtomQuery<A, E>(
408 registry: AtomRegistry.AtomRegistry,
409 atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
410 options: AtomQueryOptions = {},
411 reporter: AtomCommandReporter = console,
412): Promise<AtomCommandResult<A, E>> {
413 const query = Effect.scoped(
414 Effect.gen(function* () {
415 yield* AtomRegistry.mount(registry, atom);
416 if (options.refresh) {
417 yield* Effect.sync(() => {
418 const current = registry.get(atom);
419 if (current._tag !== "Initial" && !current.waiting) {
420 registry.refresh(atom);
421 }
422 });
423 }
424 return yield* AtomRegistry.getResult(registry, atom, {
425 suspendOnWaiting: true,
426 });
427 }),
428 );
429 return executeAtomCommand(
430 () => Effect.runPromiseExit(query, { signal: options.signal }),
431 options,
432 reporter,
433 );
434}
435
436export function createRuntimeCommand<R, ER, W, A, E>(
437 runtime: Atom.AtomRuntime<R, ER>,
438 options: {
439 readonly label: string;
440 readonly execute: (input: W, registry: AtomRegistry.AtomRegistry) => Effect.Effect<A, E, R>;
441 readonly scheduler?: AtomCommandScheduler;
442 readonly concurrency?: AtomCommandConcurrency<W>;
443 },
444): AtomCommand<W, A, E | ER> {
445 const scheduler = options.scheduler ?? createAtomCommandScheduler();
446 const concurrency = options.concurrency ?? { mode: "parallel" as const };
447 return {
448 label: options.label,
449 run: (registry, input) =>
450 settleAtomCommandResult(() =>
451 scheduler.schedule(registry, concurrency, input, () => {
452 const atom = runtime
453 .atom(options.execute(input, registry))
454 .pipe(Atom.withLabel(options.label));
455 return executeAtomQuery(registry, atom, { reportDefect: false, reportFailure: false });
456 }),
457 ),
458 };
459}
460
461export function reportAtomCommandResult(
462 result: AtomCommandResult<unknown, unknown>,
463 options: AtomCommandOptions = {},
464 reporter: AtomCommandReporter = console,
465): void {
466 if (AsyncResult.isSuccess(result) || Cause.hasInterruptsOnly(result.cause)) {
467 return;
468 }
469
470 const label = options.label ?? "atom command";
471 if (Cause.hasDies(result.cause)) {
472 if (options.reportDefect ?? true) {
473 reporter.error(`[atom-command] ${label} defected`, result.cause);
474 }
475 } else if (options.reportFailure ?? true) {
476 reporter.warn(`[atom-command] ${label} failed`, result.cause);
477 }
478}
479
480export async function settlePromise<A>(
481 execute: () => Promise<A>,
482): Promise<AtomCommandResult<A, never>> {
483 try {
484 return AsyncResult.success(await execute());
485 } catch (defect) {
486 return AsyncResult.failure(Cause.die(defect));
487 }
488}
489
490/** Where a query's answer is persisted, so an action's reply can be saved where a read would be. */
491export const queryPersistenceKey = (label: string, input: unknown): string =>
492 JSON.stringify([label, input]);
493
494function environmentRpcKey<Input>(target: {
495 readonly environmentId: EnvironmentIdType;
496 readonly input: Input;
497}): string {
498 return JSON.stringify([target.environmentId, target.input]);
499}
500
501function parseEnvironmentRpcKey<Input>(key: string): {
502 readonly environmentId: EnvironmentIdType;
503 readonly input: Input;
504} {
505 const decoded = JSON.parse(key) as [EnvironmentIdType, Input];
506 return {
507 environmentId: EnvironmentId.make(decoded[0]),
508 input: decoded[1],
509 };
510}
511
512function runInEnvironment<A, E, R>(
513 environmentId: EnvironmentIdType,
514 effect: Effect.Effect<A, E, R>,
515): Effect.Effect<
516 A,
517 E | EnvironmentRegistry.EnvironmentNotRegisteredError,
518 EnvironmentRegistry.EnvironmentRegistry | Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
519> {
520 return EnvironmentRegistry.EnvironmentRegistry.pipe(
521 Effect.flatMap((registry) => registry.run(environmentId, effect)),
522 );
523}
524
525/**
526 * `Atom.makeRefreshOnSignal` subscribes to the signal, so a query that outlives its view on an
527 * idle TTL still re-runs on every change. Reading the signal as a dependency instead lets the
528 * registry re-run a mounted query right away and only mark an idle one stale, which then
529 * refreshes on its next read.
530 */
531const refreshOnSignalWhileRead =
532 <V>(signal: Atom.Atom<unknown>, isFinal: (value: V) => boolean) =>
533 <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> => {
534 // The signal value each registry's data was read under. Held outside the node because the
535 // registry can sweep this node while `self` keeps the data, and the rebuilt node must still
536 // see a change it missed.
537 const readUnder = new WeakMap<AtomRegistry.AtomRegistry, unknown>();
538 return Atom.transform(
539 self,
540 (get) => {
541 const current = get(signal);
542 get.subscribe(self, (value) => get.setSelf(value));
543 const changed =
544 readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current);
545 readUnder.set(get.registry, current);
546 if (changed && !isFinal(get.once(self))) get.refresh(self);
547 return get.once(self);
548 },
549 { initialValueTarget: self },
550 );
551 };
552
553const refreshEveryUntilFinal =
554 <V>(intervalMs: number, isFinal: (value: V) => boolean) =>
555 <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> =>
556 Atom.transform(
557 self,
558 (get) => {
559 const value = get(self);
560 if (!isFinal(value)) {
561 // @effect-diagnostics-next-line globalTimers:off - Mirrors `Atom.withRefresh`, outside an Effect runtime.
562 const handle = setTimeout(() => get.refresh(self), intervalMs);
563 get.addFinalizer(() => clearTimeout(handle));
564 }
565 return value;
566 },
567 { initialValueTarget: self },
568 );
569
570/**
571 * A query result's data alone. A refetch always publishes a new result (a new
572 * timestamp, `waiting` on and off), while equal data keeps its identity, so a
573 * reader of only this skips the refetches that changed nothing.
574 */
575export const selectEnvironmentQueryData = <A, E>(result: AsyncResult.AsyncResult<A, E>): A | null =>
576 AsyncResult.getOrElse(result, () => null);
577
578type EnvironmentConnection = readonly [
579 SupervisorConnectionState,
580 Option.Option<RpcSession.RpcSession>,
581];
582
583const connectionAtomsByRuntime = new WeakMap<
584 object,
585 (
586 environmentId: EnvironmentIdType,
587 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentConnection | null>>
588>();
589
590/**
591 * One connection atom per environment for every query on a runtime, so a reader of the session
592 * generation and a query it re-runs see the same connection change at once.
593 */
594function environmentConnectionAtoms<R, ER>(
595 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
596) {
597 const existing = connectionAtomsByRuntime.get(runtime);
598 if (existing !== undefined) return existing;
599 const family = Atom.family((environmentId: EnvironmentIdType) =>
600 runtime.atom(
601 followStreamInEnvironment(
602 environmentId,
603 Stream.unwrap(
604 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
605 Effect.map((supervisor) =>
606 SubscriptionRef.changes(supervisor.state).pipe(
607 Stream.zipLatest(SubscriptionRef.changes(supervisor.session)),
608 ),
609 ),
610 ),
611 ),
612 ),
613 { initialValue: null },
614 ),
615 ) as (
616 environmentId: EnvironmentIdType,
617 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentConnection | null>>;
618 connectionAtomsByRuntime.set(runtime, family);
619 return family;
620}
621
622const sessionGenerationAtomsByRuntime = new WeakMap<
623 object,
624 (environmentId: EnvironmentIdType) => Atom.Atom<number>
625>();
626
627/**
628 * Counts established sessions; it changes only when a new one comes up, so failed one-shot reads
629 * can watch it to run again. Keyed on the session object because the supervisor's `generation`
630 * restarts when the registry installs a new supervisor (re-pairing, route edits).
631 */
632export function environmentSessionGenerationAtom<R, ER>(
633 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
634 environmentId: EnvironmentIdType,
635): Atom.Atom<number> {
636 let family = sessionGenerationAtomsByRuntime.get(runtime);
637 if (family === undefined) {
638 const connectionAtom = environmentConnectionAtoms(runtime);
639 const sessionsSeen = Atom.family((id: EnvironmentIdType) =>
640 Atom.make(
641 (get): { readonly session: RpcSession.RpcSession | null; readonly count: number } => {
642 const previous = Option.getOrElse(
643 get.self<{ readonly session: RpcSession.RpcSession | null; readonly count: number }>(),
644 () => ({ session: null, count: 0 }),
645 );
646 const connection = AsyncResult.getOrElse(get(connectionAtom(id)), () => null);
647 if (connection === null || connection[0].phase !== "connected") return previous;
648 const session = Option.getOrNull(connection[1]);
649 return session === null || session === previous.session
650 ? previous
651 : { session, count: previous.count + 1 };
652 },
653 ),
654 );
655 family = Atom.family((id: EnvironmentIdType) =>
656 Atom.make((get) => get(sessionsSeen(id)).count).pipe(
657 Atom.withLabel(`environment-session-generation:${id}`),
658 ),
659 );
660 sessionGenerationAtomsByRuntime.set(runtime, family);
661 }
662 return family(environmentId);
663}
664
665export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>(
666 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
667 options: EnvironmentQueryAtomOptions<
668 Input,
669 A,
670 E,
671 | EnvironmentSupervisor.EnvironmentSupervisor
672 | EnvironmentRegistry.EnvironmentRegistry
673 | AtomRegistry.AtomRegistry
674 | R
675 >,
676): (target: {
677 readonly environmentId: EnvironmentIdType;
678 readonly input: Input;
679}) => Atom.Atom<AsyncResult.AsyncResult<A, E | ER | Error>> {
680 const connectionAtom = environmentConnectionAtoms(runtime);
681 const family = Atom.family((key: string) => {
682 const target = parseEnvironmentRpcKey<Input>(key);
683 const idleTtlMs = options.idleTtlMs ?? 5 * 60_000;
684 // `Atom.swr` checks staleness only when its node builds. To revalidate on
685 // remount, the wrappers drop on unmount and the source alone keeps the
686 // cached value warm.
687 const keepWarm = Atom.setIdleTTL(idleTtlMs);
688 const keepSourceWarm = options.revalidateOnRemount ? keepWarm : identity;
689 const keepWrapperWarm = options.revalidateOnRemount ? identity : keepWarm;
690 const queryAtom = runtime
691 .atom<
692 A,
693 | E
694 | ConnectionAttemptError
695 | EnvironmentRegistry.EnvironmentNotRegisteredError
696 | EnvironmentRpcUnavailableError
697 >((get) => {
698 const connection = AsyncResult.getOrElse(
699 get(connectionAtom(target.environmentId)),
700 () => null,
701 );
702 // A refetch that returns equal content keeps the previous value's
703 // references, so readers of the data skip re-rendering on idle polls.
704 const share = (value: A): A =>
705 reuseUnchanged(
706 Option.getOrUndefined(
707 Option.flatMap(get.self<AsyncResult.AsyncResult<A, unknown>>(), AsyncResult.value),
708 ),
709 value,
710 );
711 const emit = (value: A) =>
712 Effect.sync(() => get.setSelf(AsyncResult.success(share(value), { waiting: true }))).pipe(
713 // A value set while the atom builds wins over the build's own
714 // result, so a query that then settled synchronously would stay
715 // waiting forever. Suspend on a promise (`yieldNow` is drained
716 // inside the build) so the settled result arrives asynchronously.
717 Effect.andThen(Effect.promise(() => Promise.resolve())),
718 );
719 const execute = (): Effect.Effect<
720 A,
721 | E
722 | ConnectionAttemptError
723 | EnvironmentRegistry.EnvironmentNotRegisteredError
724 | EnvironmentRpcUnavailableError,
725 | EnvironmentRegistry.EnvironmentRegistry
726 | AtomRegistry.AtomRegistry
727 | Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
728 > => {
729 if (connection === null) return Effect.never;
730 const [connectionState, session] = connection;
731 switch (connectionState.phase) {
732 case "connected":
733 return Option.isSome(session)
734 ? runInEnvironment(
735 target.environmentId,
736 options
737 .execute(
738 target.input,
739 emit,
740 Option.flatMap(
741 get.self<AsyncResult.AsyncResult<A, unknown>>(),
742 AsyncResult.value,
743 ),
744 )
745 .pipe(pendingOnceSessionReplaced(session.value)),
746 )
747 : Effect.never;
748 case "connecting":
749 case "backoff":
750 // A retry can take minutes. A query with an answer keeps showing it while it
751 // waits; one with nothing to show reports the failure the retry is recovering
752 // from, so its reader stops loading, and re-runs when the connection settles.
753 return connectionState.lastFailure !== null &&
754 Option.isNone(
755 Option.flatMap(
756 get.self<AsyncResult.AsyncResult<A, unknown>>(),
757 AsyncResult.value,
758 ),
759 )
760 ? Effect.fail(connectionState.lastFailure)
761 : Effect.never;
762 case "available":
763 case "offline":
764 case "blocked":
765 if (connectionState.lastFailure !== null) {
766 return Effect.fail(connectionState.lastFailure);
767 }
768 return Effect.fail(
769 new EnvironmentRpcUnavailableError({
770 environmentId: target.environmentId,
771 message: `Environment ${target.environmentId} is ${
772 connectionState.phase === "available" ? "not connected" : connectionState.phase
773 }.`,
774 }),
775 );
776 }
777 };
778 const onAnswer = options.onAnswer;
779 const answer = () =>
780 onAnswer === undefined
781 ? execute().pipe(Effect.map(share))
782 : execute().pipe(
783 Effect.map(share),
784 Effect.tap((value) => onAnswer(target, value, get.registry)),
785 );
786 if (!options.persistence) return answer();
787 const persistence = options.persistence;
788 const cacheKey = queryPersistenceKey(options.label, target.input);
789 const previous = Option.flatMap(
790 get.self<AsyncResult.AsyncResult<A, unknown>>(),
791 AsyncResult.value,
792 );
793 return Effect.gen(function* () {
794 const cached = Option.isSome(previous)
795 ? previous
796 : yield* persistence.load(target.environmentId, cacheKey);
797 if (
798 Option.isSome(cached) &&
799 (connection === null || connection[0].phase !== "connected")
800 ) {
801 return cached.value;
802 }
803 if (Option.isNone(previous) && Option.isSome(cached)) yield* emit(cached.value);
804 // Readers must not wait on the disk: web saves one IndexedDB row per listed PR's stats.
805 // The save ends with the query, so a removed environment's cache stays cleared.
806 return yield* answer().pipe(
807 Effect.tap((value) =>
808 persistence.save(target.environmentId, cacheKey, value).pipe(Effect.forkScoped),
809 ),
810 );
811 });
812 })
813 .pipe(
814 keepSourceWarm,
815 Atom.swr({
816 staleTime: options.staleTimeMs ?? 30_000,
817 revalidateOnMount: true,
818 }),
819 keepWrapperWarm,
820 );
821 const isFinal = (result: AsyncResult.AsyncResult<A, unknown>) =>
822 options.isFinal !== undefined && Option.exists(AsyncResult.value(result), options.isFinal);
823 const intervalQuery =
824 options.refreshIntervalMs === undefined
825 ? queryAtom
826 : queryAtom.pipe(refreshEveryUntilFinal(options.refreshIntervalMs, isFinal));
827 const refreshTrigger = options.refreshTrigger?.(target);
828 return (
829 refreshTrigger === undefined
830 ? intervalQuery
831 : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger, isFinal))
832 ).pipe(keepWrapperWarm, Atom.withLabel(`${options.label}:${key}`));
833 });
834 return (target) => family(environmentRpcKey(target));
835}
836
837export function createEnvironmentSubscriptionAtomFamily<R, ER, Input, A, E>(
838 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
839 options: EnvironmentSubscriptionAtomOptions<
840 Input,
841 A,
842 E,
843 EnvironmentSupervisor.EnvironmentSupervisor | R
844 >,
845) {
846 const idleTtlMs = options.idleTtlMs ?? 5 * 60_000;
847 const label = (key: string) =>
848 options.sensitiveInput
849 ? `${options.label}:${parseEnvironmentRpcKey<Input>(key).environmentId}`
850 : `${options.label}:${key}`;
851 const live = Atom.family((key: string) => {
852 const target = parseEnvironmentRpcKey<Input>(key);
853 return runtime
854 .atom(
855 Stream.suspend(() =>
856 followStreamInEnvironment(target.environmentId, options.subscribe(target.input)),
857 ).pipe(options.completeWhen ? Stream.takeUntil(options.completeWhen) : (stream) => stream),
858 )
859 .pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(label(key)));
860 });
861 const { completeWhen, completedTtlMs } = options;
862 if (completeWhen === undefined || completedTtlMs === undefined) {
863 return (target: { readonly environmentId: EnvironmentIdType; readonly input: Input }) =>
864 live(environmentRpcKey(target));
865 }
866 const completed = Atom.family((key: string) =>
867 Atom.make(Option.none<A>()).pipe(
868 Atom.setIdleTTL(completedTtlMs),
869 Atom.withLabel(`${label(key)}:completed`),
870 ),
871 );
872 // Reading `completed` first drops the live subscription once its value is final.
873 const family = Atom.family((key: string) =>
874 Atom.make((get) => {
875 const done = get(completed(key));
876 if (Option.isSome(done)) return AsyncResult.success(done.value);
877 const result = get(live(key));
878 if (AsyncResult.isSuccess(result) && !result.waiting && completeWhen(result.value)) {
879 get.set(completed(key), Option.some(result.value));
880 }
881 return result;
882 }).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(label(key))),
883 );
884 return (target: { readonly environmentId: EnvironmentIdType; readonly input: Input }) =>
885 family(environmentRpcKey(target));
886}
887
888export function createEnvironmentCommand<R, ER, Input, A, E>(
889 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
890 options: EnvironmentCommandAtomOptions<
891 Input,
892 A,
893 E,
894 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry | R
895 >,
896) {
897 return createRuntimeCommand(runtime, {
898 label: options.label,
899 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
900 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
901 execute: (target, registry) =>
902 runInEnvironment(
903 target.environmentId,
904 options.execute(target.input, registry, target.environmentId),
905 ),
906 });
907}
908
909export function createEnvironmentRpcQueryAtomFamily<
910 R,
911 ER,
912 TTag extends Exclude<EnvironmentUnaryRpcTag, ClientGuardedRpcTag>,
913>(
914 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
915 options: {
916 readonly label: string;
917 readonly tag: TTag;
918 readonly execute?: (
919 input: EnvironmentRpcInput<TTag>,
920 emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>,
921 previous: Option.Option<EnvironmentRpcSuccess<TTag>>,
922 ) => Effect.Effect<
923 EnvironmentRpcSuccess<TTag>,
924 EnvironmentRpcFailure<TTag> | EnvironmentRpcUnavailableError,
925 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry
926 >;
927 readonly persistence?: QueryPersistence<EnvironmentRpcSuccess<TTag>>;
928 readonly staleTimeMs?: number;
929 readonly idleTtlMs?: number;
930 readonly revalidateOnRemount?: boolean;
931 readonly refreshIntervalMs?: number;
932 readonly refreshTrigger?: (target: {
933 readonly environmentId: EnvironmentIdType;
934 readonly input: EnvironmentRpcInput<TTag>;
935 }) => Atom.Atom<unknown> | undefined;
936 readonly isFinal?: (value: EnvironmentRpcSuccess<TTag>) => boolean;
937 readonly onAnswer?: QueryAnswerHandler<EnvironmentRpcInput<TTag>, EnvironmentRpcSuccess<TTag>>;
938 },
939) {
940 return createEnvironmentQueryAtomFamily(runtime, {
941 label: options.label,
942 ...(options.persistence === undefined ? {} : { persistence: options.persistence }),
943 ...(options.staleTimeMs === undefined ? {} : { staleTimeMs: options.staleTimeMs }),
944 ...(options.idleTtlMs === undefined ? {} : { idleTtlMs: options.idleTtlMs }),
945 ...(options.revalidateOnRemount === undefined
946 ? {}
947 : { revalidateOnRemount: options.revalidateOnRemount }),
948 ...(options.refreshIntervalMs === undefined
949 ? {}
950 : { refreshIntervalMs: options.refreshIntervalMs }),
951 ...(options.refreshTrigger === undefined ? {} : { refreshTrigger: options.refreshTrigger }),
952 ...(options.isFinal === undefined ? {} : { isFinal: options.isFinal }),
953 ...(options.onAnswer === undefined ? {} : { onAnswer: options.onAnswer }),
954 execute: (
955 input: EnvironmentRpcInput<TTag>,
956 emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>,
957 previous: Option.Option<EnvironmentRpcSuccess<TTag>>,
958 ) => options.execute?.(input, emit, previous) ?? request(options.tag, input),
959 });
960}
961
962export function createEnvironmentRpcSubscriptionAtomFamily<
963 R,
964 ER,
965 TTag extends EnvironmentSubscriptionRpcTag,
966 B = EnvironmentRpcStreamValue<TTag>,
967>(
968 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
969 options: {
970 readonly label: string;
971 readonly tag: TTag;
972 readonly idleTtlMs?: number;
973 readonly completeWhen?: (value: B) => boolean;
974 readonly completedTtlMs?: number;
975 readonly transform?: (
976 stream: Stream.Stream<
977 EnvironmentRpcStreamValue<TTag>,
978 EnvironmentRpcStreamFailure<TTag>,
979 EnvironmentSupervisor.EnvironmentSupervisor | R
980 >,
981 ) => Stream.Stream<
982 B,
983 EnvironmentRpcStreamFailure<TTag>,
984 EnvironmentSupervisor.EnvironmentSupervisor | R
985 >;
986 },
987) {
988 return createEnvironmentSubscriptionAtomFamily(runtime, {
989 label: options.label,
990 ...(options.completeWhen === undefined ? {} : { completeWhen: options.completeWhen }),
991 ...(options.completedTtlMs === undefined ? {} : { completedTtlMs: options.completedTtlMs }),
992 ...(options.idleTtlMs === undefined ? {} : { idleTtlMs: options.idleTtlMs }),
993 subscribe: (input: EnvironmentRpcInput<TTag>) => {
994 const stream = subscribe(options.tag, input);
995 return options.transform === undefined
996 ? (stream as Stream.Stream<
997 B,
998 EnvironmentRpcStreamFailure<TTag>,
999 EnvironmentSupervisor.EnvironmentSupervisor | R
1000 >)
1001 : options.transform(stream);
1002 },
1003 });
1004}
1005
1006export function createEnvironmentRpcCommand<
1007 R,
1008 ER,
1009 TTag extends EnvironmentUnaryRpcTag,
1010 Input extends EnvironmentRpcInput<TTag> = EnvironmentRpcInput<TTag>,
1011 Started = undefined,
1012>(
1013 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
1014 options: {
1015 readonly label: string;
1016 readonly tag: TTag;
1017 /** Runs once authorized, as the request starts; what it returns reaches `onSuccess`. */
1018 readonly onStart?: (
1019 target: {
1020 readonly environmentId: EnvironmentIdType;
1021 readonly input: NoInfer<Input>;
1022 },
1023 registry: AtomRegistry.AtomRegistry,
1024 ) => Started;
1025 readonly execute?: (
1026 input: Input,
1027 registry: AtomRegistry.AtomRegistry,
1028 environmentId: EnvironmentIdType,
1029 ) => Effect.Effect<
1030 EnvironmentRpcSuccess<TTag>,
1031 EnvironmentRpcFailure<TTag> | EnvironmentRpcUnavailableError,
1032 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry
1033 >;
1034 readonly scheduler?: AtomCommandScheduler;
1035 readonly concurrency?: AtomCommandConcurrency<{
1036 readonly environmentId: EnvironmentIdType;
1037 readonly input: NoInfer<Input>;
1038 }>;
1039 readonly onSuccess?: (
1040 target: {
1041 readonly environmentId: EnvironmentIdType;
1042 readonly input: NoInfer<Input>;
1043 },
1044 registry: AtomRegistry.AtomRegistry,
1045 result: EnvironmentRpcSuccess<TTag>,
1046 started: Started | undefined,
1047 ) => Effect.Effect<void, never, R>;
1048 readonly onSettled?: (
1049 target: {
1050 readonly environmentId: EnvironmentIdType;
1051 readonly input: NoInfer<Input>;
1052 },
1053 registry: AtomRegistry.AtomRegistry,
1054 ) => Effect.Effect<void, never, R>;
1055 },
1056) {
1057 const permissions = createCommandPermissions(runtime, options.tag);
1058 const command = createEnvironmentCommand(runtime, {
1059 label: options.label,
1060 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
1061 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
1062 execute: (input: Input, registry, environmentId) => {
1063 const target = {
1064 environmentId,
1065 input,
1066 };
1067 let started: Started | undefined;
1068 // Routing requires consent on the origin as well as the actual destination.
1069 // The transport check below deliberately checks the destination again.
1070 return permissions.authorize(registry, environmentId, input).pipe(
1071 Effect.andThen(() => {
1072 started = options.onStart?.(target, registry);
1073 return (
1074 options.execute?.(input, registry, environmentId) ?? requestGuarded(options.tag, input)
1075 );
1076 }),
1077 Effect.provideService(RpcPermissionGuard, {
1078 authorize: (id, method, payload) =>
1079 createCommandPermissions(runtime, method).authorize(registry, id, payload),
1080 }),
1081 Effect.tap(
1082 (result) => options.onSuccess?.(target, registry, result, started) ?? Effect.void,
1083 ),
1084 Effect.ensuring(options.onSettled?.(target, registry) ?? Effect.void),
1085 );
1086 },
1087 });
1088 return {
1089 ...command,
1090 requiredScopes: permissions.requiredScopes,
1091 permissionAtom: permissions.permissionAtom,
1092 };
1093}