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