1import type { QueryPersistence } from "./queryPersistence.ts";
2import { createCommandPermissions } from "./commandPermissions.ts";
3import { followStreamInEnvironment } from "./environmentStreams.ts";
4export { runStreamInEnvironment, followStreamInEnvironment } from "./environmentStreams.ts";
5import {
6 type ClientGuardedRpcTag,
7 EnvironmentId,
8 type EnvironmentId as EnvironmentIdType,
9} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";
11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";
13import * as Option from "effect/Option";
14import * as Stream from "effect/Stream";
15import * as SubscriptionRef from "effect/SubscriptionRef";
16import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
17
18import type { ConnectionAttemptError } from "../connection/model.ts";
19import * as EnvironmentRegistry from "../connection/registry.ts";
20import {
21 type EnvironmentRpcInput,
22 type EnvironmentRpcSuccess,
23 type EnvironmentRpcFailure,
24 type EnvironmentRpcStreamFailure,
25 type EnvironmentRpcStreamValue,
26 type EnvironmentSubscriptionRpcTag,
27 type EnvironmentUnaryRpcTag,
28 EnvironmentRpcUnavailableError,
29 request,
30 requestGuarded,
31 RpcPermissionGuard,
32 subscribe,
33} from "../rpc/client.ts";
34import * as EnvironmentSupervisor from "../connection/supervisor.ts";
35
36interface EnvironmentAtomOptions<Input, A, E, R> {
37 readonly label: string;
38 readonly execute: (input: Input) => Effect.Effect<A, E, R>;
39 readonly scheduler?: AtomCommandScheduler;
40 readonly concurrency?: AtomCommandConcurrency<{
41 readonly environmentId: EnvironmentIdType;
42 readonly input: Input;
43 }>;
44}
45
46interface EnvironmentCommandAtomOptions<Input, A, E, R> extends Omit<
47 EnvironmentAtomOptions<Input, A, E, R>,
48 "execute"
49> {
50 readonly execute: (
51 input: Input,
52 registry: AtomRegistry.AtomRegistry,
53 environmentId: EnvironmentIdType,
54 ) => Effect.Effect<A, E, R>;
55}
56
57interface EnvironmentQueryAtomOptions<Input, A, E, R> extends Omit<
58 EnvironmentAtomOptions<Input, A, E, R>,
59 "execute"
60> {
61 /**
62 * `emit` shows a provisional value while the query keeps waiting. The settled
63 * result replaces it, so catch failures that should keep it on screen.
64 */
65 readonly execute: (
66 input: Input,
67 emit: (value: A) => Effect.Effect<void>,
68 ) => Effect.Effect<A, E, R>;
69 readonly executeDisconnected?: (
70 input: Input,
71 emit: (value: A) => Effect.Effect<void>,
72 ) => Effect.Effect<A, E, R>;
73 readonly persistence?: QueryPersistence<A>;
74 readonly staleTimeMs?: number;
75 readonly idleTtlMs?: number;
76 readonly refreshIntervalMs?: number;
77 readonly refreshTrigger?: (target: {
78 readonly environmentId: EnvironmentIdType;
79 readonly input: Input;
80 }) => Atom.Atom<unknown> | undefined;
81 /** A value that can no longer change. It stops the interval and `refreshTrigger` re-reads. */
82 readonly isFinal?: (value: A) => boolean;
83}
84
85interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> {
86 readonly sensitiveInput?: boolean;
87 readonly label: string;
88 readonly subscribe: (input: Input) => Stream.Stream<A, E, R>;
89 readonly completeWhen?: (value: A) => boolean;
90 readonly idleTtlMs?: number;
91}
92
93export type SettledAsyncResult<A, E> = AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>;
94
95export type AtomCommandResult<A, E> = SettledAsyncResult<A, E>;
96
97export type AtomCommandSuccess<R> = R extends AtomCommandResult<infer A, infer _E> ? A : never;
98
99export type AtomCommandFailure<R> = R extends AtomCommandResult<infer _A, infer E> ? E : never;
100
101export interface AtomCommandOptions {
102 readonly label?: string;
103 readonly reportFailure?: boolean;
104 readonly reportDefect?: boolean;
105}
106
107export interface AtomCommandReporter {
108 readonly warn: (message: string, cause: Cause.Cause<unknown>) => void;
109 readonly error: (message: string, cause: Cause.Cause<unknown>) => void;
110}
111
112export interface AtomCommand<W, A, E> {
113 readonly label: string;
114 readonly run: (registry: AtomRegistry.AtomRegistry, input: W) => Promise<AtomCommandResult<A, E>>;
115}
116
117export type AtomCommandConcurrency<W> =
118 /** Every invocation runs independently. */
119 | { readonly mode: "parallel" }
120 | {
121 /**
122 * `serial` preserves every invocation in FIFO order, `singleFlight` shares an active
123 * invocation, and `latest` coalesces queued invocations to the newest input.
124 */
125 readonly mode: "serial" | "singleFlight" | "latest";
126 readonly key: (input: W) => string;
127 };
128
129interface AtomCommandSchedulerState {
130 readonly serial: Map<string, Promise<unknown>>;
131 readonly singleFlight: Map<string, Promise<unknown>>;
132 readonly latest: Map<string, AtomCommandLatestLane>;
133}
134
135interface AtomCommandLatestBatch {
136 execute: () => Promise<AtomCommandResult<unknown, unknown>>;
137 readonly resolve: Array<(result: AtomCommandResult<unknown, unknown>) => void>;
138}
139
140interface AtomCommandLatestLane {
141 running: boolean;
142 pending: AtomCommandLatestBatch | undefined;
143}
144
145export interface AtomCommandScheduler {
146 readonly schedule: <W, A, E>(
147 registry: AtomRegistry.AtomRegistry,
148 concurrency: AtomCommandConcurrency<W>,
149 input: W,
150 execute: () => Promise<AtomCommandResult<A, E>>,
151 ) => Promise<AtomCommandResult<A, E>>;
152}
153
154async function settleAtomCommandResult<A, E>(
155 execute: () => Promise<AtomCommandResult<A, E>>,
156): Promise<AtomCommandResult<A, E>> {
157 try {
158 return await execute();
159 } catch (defect) {
160 return AsyncResult.failure(Cause.die(defect));
161 }
162}
163
164export function createAtomCommandScheduler(): AtomCommandScheduler {
165 const registryStates = new WeakMap<AtomRegistry.AtomRegistry, AtomCommandSchedulerState>();
166
167 const stateFor = (registry: AtomRegistry.AtomRegistry): AtomCommandSchedulerState => {
168 const existing = registryStates.get(registry);
169 if (existing !== undefined) {
170 return existing;
171 }
172 const state: AtomCommandSchedulerState = {
173 serial: new Map(),
174 singleFlight: new Map(),
175 latest: new Map(),
176 };
177 registryStates.set(registry, state);
178 return state;
179 };
180
181 return {
182 schedule: <W, A, E>(
183 registry: AtomRegistry.AtomRegistry,
184 concurrency: AtomCommandConcurrency<W>,
185 input: W,
186 execute: () => Promise<AtomCommandResult<A, E>>,
187 ): Promise<AtomCommandResult<A, E>> => {
188 if (concurrency.mode === "parallel") {
189 return execute();
190 }
191
192 const key = concurrency.key(input);
193 const state = stateFor(registry);
194 if (concurrency.mode === "singleFlight") {
195 const existing = state.singleFlight.get(key) as
196 | Promise<AtomCommandResult<A, E>>
197 | undefined;
198 if (existing !== undefined) {
199 return existing;
200 }
201 const current = execute();
202 state.singleFlight.set(key, current);
203 void current.then(
204 () => {
205 if (state.singleFlight.get(key) === current) {
206 state.singleFlight.delete(key);
207 }
208 },
209 () => {
210 if (state.singleFlight.get(key) === current) {
211 state.singleFlight.delete(key);
212 }
213 },
214 );
215 return current;
216 }
217
218 if (concurrency.mode === "serial") {
219 const previous = state.serial.get(key);
220 const current = previous === undefined ? execute() : previous.then(execute, execute);
221 state.serial.set(key, current);
222 void current.then(
223 () => {
224 if (state.serial.get(key) === current) {
225 state.serial.delete(key);
226 }
227 },
228 () => {
229 if (state.serial.get(key) === current) {
230 state.serial.delete(key);
231 }
232 },
233 );
234 return current;
235 }
236
237 let lane = state.latest.get(key);
238 if (lane === undefined) {
239 lane = { running: false, pending: undefined };
240 state.latest.set(key, lane);
241 }
242 const activeLane = lane;
243
244 const result = new Promise<AtomCommandResult<A, E>>((resolve) => {
245 if (activeLane.pending === undefined) {
246 activeLane.pending = {
247 execute: execute as () => Promise<AtomCommandResult<unknown, unknown>>,
248 resolve: [resolve as (result: AtomCommandResult<unknown, unknown>) => void],
249 };
250 return;
251 }
252 activeLane.pending.execute = execute as () => Promise<AtomCommandResult<unknown, unknown>>;
253 activeLane.pending.resolve.push(
254 resolve as (result: AtomCommandResult<unknown, unknown>) => void,
255 );
256 });
257
258 if (!activeLane.running) {
259 activeLane.running = true;
260 void (async () => {
261 while (activeLane.pending !== undefined) {
262 const batch = activeLane.pending;
263 activeLane.pending = undefined;
264 let batchResult: AtomCommandResult<unknown, unknown>;
265 try {
266 batchResult = await batch.execute();
267 } catch (defect) {
268 batchResult = AsyncResult.failure(Cause.die(defect));
269 }
270 for (const resolve of batch.resolve) {
271 resolve(batchResult);
272 }
273 }
274 activeLane.running = false;
275 if (state.latest.get(key) === activeLane) {
276 state.latest.delete(key);
277 }
278 })();
279 }
280
281 return result;
282 },
283 };
284}
285
286/** Runs one effect inside an existing command scheduler lane. */
287export function scheduleAtomCommandEffect<W, A, E, R>(
288 registry: AtomRegistry.AtomRegistry,
289 scheduler: AtomCommandScheduler,
290 concurrency: AtomCommandConcurrency<W>,
291 input: W,
292 effect: Effect.Effect<A, E, R>,
293): Effect.Effect<A, E, R> {
294 return Effect.gen(function* () {
295 const context = yield* Effect.context<R>();
296 const result = yield* Effect.promise((signal) =>
297 scheduler.schedule<W, A, E>(registry, concurrency, input, async () => {
298 const exit = await Effect.runPromiseExitWith(context)(effect, { signal });
299 return Exit.isSuccess(exit)
300 ? AsyncResult.success(exit.value)
301 : AsyncResult.failure(exit.cause);
302 }),
303 );
304 return result._tag === "Success" ? result.value : yield* Effect.failCause(result.cause);
305 });
306}
307
308export async function runAtomCommand<W, A, E>(
309 registry: AtomRegistry.AtomRegistry,
310 command: AtomCommand<W, A, E>,
311 input: W,
312 options: AtomCommandOptions = {},
313 reporter: AtomCommandReporter = console,
314): Promise<AtomCommandResult<A, E>> {
315 const result = await settleAtomCommandResult(() => command.run(registry, input));
316 reportAtomCommandResult(result, { ...options, label: options.label ?? command.label }, reporter);
317 return result;
318}
319
320export function mapAtomCommandResult<A, E, B>(
321 result: AtomCommandResult<A, E>,
322 map: (value: A) => B,
323): AtomCommandResult<B, E> {
324 return result._tag === "Success"
325 ? AsyncResult.success(map(result.value))
326 : AsyncResult.failure(result.cause);
327}
328
329export function isAtomCommandInterrupted(result: AtomCommandResult<unknown, unknown>): boolean {
330 return result._tag === "Failure" && Cause.hasInterruptsOnly(result.cause);
331}
332
333export function squashAtomCommandFailure(result: {
334 readonly cause: Cause.Cause<unknown>;
335}): unknown {
336 return Cause.squash(result.cause);
337}
338
339export async function settleAsyncResult<A, E>(
340 execute: () => Promise<Exit.Exit<A, E>>,
341): Promise<SettledAsyncResult<A, E>> {
342 try {
343 return AsyncResult.fromExit(await execute());
344 } catch (defect) {
345 return AsyncResult.failure(Cause.die(defect));
346 }
347}
348
349export async function executeAtomCommand<A, E>(
350 execute: () => Promise<Exit.Exit<A, E>>,
351 options: AtomCommandOptions = {},
352 reporter: AtomCommandReporter = console,
353): Promise<AtomCommandResult<A, E>> {
354 const result = await settleAsyncResult(execute);
355 reportAtomCommandResult(result, options, reporter);
356 return result;
357}
358
359export interface AtomQueryOptions extends AtomCommandOptions {
360 /**
361 * Force a fresh execution instead of accepting a value the atom already
362 * holds (e.g. an SWR-cached result within its stale window). Used by
363 * verification flows where a cached failure must not satisfy a retry.
364 */
365 readonly refresh?: boolean;
366 /** Interrupt the query wait when its caller no longer wants the result. */
367 readonly signal?: AbortSignal;
368}
369
370export async function executeAtomQuery<A, E>(
371 registry: AtomRegistry.AtomRegistry,
372 atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
373 options: AtomQueryOptions = {},
374 reporter: AtomCommandReporter = console,
375): Promise<AtomCommandResult<A, E>> {
376 const query = Effect.scoped(
377 Effect.gen(function* () {
378 yield* AtomRegistry.mount(registry, atom);
379 if (options.refresh) {
380 yield* Effect.sync(() => {
381 // Only a settled value can be a leftover from an earlier read; a
382 // computation that mounting just started is already fresh.
383 const current = registry.get(atom);
384 if (current._tag !== "Initial" && !current.waiting) {
385 registry.refresh(atom);
386 }
387 });
388 }
389 return yield* AtomRegistry.getResult(registry, atom, {
390 suspendOnWaiting: true,
391 });
392 }),
393 );
394 return executeAtomCommand(
395 () => Effect.runPromiseExit(query, { signal: options.signal }),
396 options,
397 reporter,
398 );
399}
400
401export function createRuntimeCommand<R, ER, W, A, E>(
402 runtime: Atom.AtomRuntime<R, ER>,
403 options: {
404 readonly label: string;
405 readonly execute: (input: W, registry: AtomRegistry.AtomRegistry) => Effect.Effect<A, E, R>;
406 readonly scheduler?: AtomCommandScheduler;
407 readonly concurrency?: AtomCommandConcurrency<W>;
408 },
409): AtomCommand<W, A, E | ER> {
410 const scheduler = options.scheduler ?? createAtomCommandScheduler();
411 const concurrency = options.concurrency ?? { mode: "parallel" as const };
412 return {
413 label: options.label,
414 run: (registry, input) =>
415 settleAtomCommandResult(() =>
416 scheduler.schedule(registry, concurrency, input, () => {
417 const atom = runtime
418 .atom(options.execute(input, registry))
419 .pipe(Atom.withLabel(options.label));
420 return executeAtomQuery(registry, atom, { reportDefect: false, reportFailure: false });
421 }),
422 ),
423 };
424}
425
426export function reportAtomCommandResult(
427 result: AtomCommandResult<unknown, unknown>,
428 options: AtomCommandOptions = {},
429 reporter: AtomCommandReporter = console,
430): void {
431 if (AsyncResult.isSuccess(result) || Cause.hasInterruptsOnly(result.cause)) {
432 return;
433 }
434
435 const label = options.label ?? "atom command";
436 if (Cause.hasDies(result.cause)) {
437 if (options.reportDefect ?? true) {
438 reporter.error(`[atom-command] ${label} defected`, result.cause);
439 }
440 } else if (options.reportFailure ?? true) {
441 reporter.warn(`[atom-command] ${label} failed`, result.cause);
442 }
443}
444
445export async function settlePromise<A>(
446 execute: () => Promise<A>,
447): Promise<AtomCommandResult<A, never>> {
448 try {
449 return AsyncResult.success(await execute());
450 } catch (defect) {
451 return AsyncResult.failure(Cause.die(defect));
452 }
453}
454
455export function environmentRpcKey<Input>(target: {
456 readonly environmentId: EnvironmentIdType;
457 readonly input: Input;
458}): string {
459 return JSON.stringify([target.environmentId, target.input]);
460}
461
462function parseEnvironmentRpcKey<Input>(key: string): {
463 readonly environmentId: EnvironmentIdType;
464 readonly input: Input;
465} {
466 const decoded = JSON.parse(key) as [EnvironmentIdType, Input];
467 return {
468 environmentId: EnvironmentId.make(decoded[0]),
469 input: decoded[1],
470 };
471}
472
473function runInEnvironment<A, E, R>(
474 environmentId: EnvironmentIdType,
475 effect: Effect.Effect<A, E, R>,
476): Effect.Effect<
477 A,
478 E | EnvironmentRegistry.EnvironmentNotRegisteredError,
479 EnvironmentRegistry.EnvironmentRegistry | Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
480> {
481 return EnvironmentRegistry.EnvironmentRegistry.pipe(
482 Effect.flatMap((registry) => registry.run(environmentId, effect)),
483 );
484}
485
486/**
487 * Refreshes a query when `signal` changes, but only while something reads it.
488 * `Atom.makeRefreshOnSignal` subscribes to the signal, so a query that outlives its view on an
489 * idle TTL still re-runs on every change. Reading the signal as a dependency instead lets the
490 * registry re-run a mounted query right away and only mark an idle one stale, which then
491 * refreshes on its next read.
492 */
493const refreshOnSignalWhileRead =
494 <V>(signal: Atom.Atom<unknown>, isFinal: (value: V) => boolean) =>
495 <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> => {
496 // The signal value each registry's data was read under. Held outside the node because the
497 // registry can sweep this node while `self` keeps the data, and the rebuilt node must still
498 // see a change it missed.
499 const readUnder = new WeakMap<AtomRegistry.AtomRegistry, unknown>();
500 return Atom.transform(
501 self,
502 (get) => {
503 const current = get(signal);
504 get.subscribe(self, (value) => get.setSelf(value));
505 const changed =
506 readUnder.has(get.registry) && !Object.is(readUnder.get(get.registry), current);
507 readUnder.set(get.registry, current);
508 if (changed && !isFinal(get.once(self))) get.refresh(self);
509 return get.once(self);
510 },
511 { initialValueTarget: self },
512 );
513 };
514
515/** `Atom.withRefresh`, except the timer stops once the value is final. */
516const refreshEveryUntilFinal =
517 <V>(intervalMs: number, isFinal: (value: V) => boolean) =>
518 <A extends V>(self: Atom.Atom<A>): Atom.Atom<A> =>
519 Atom.transform(
520 self,
521 (get) => {
522 const value = get(self);
523 if (!isFinal(value)) {
524 // @effect-diagnostics-next-line globalTimers:off - Mirrors `Atom.withRefresh`, outside an Effect runtime.
525 const handle = setTimeout(() => get.refresh(self), intervalMs);
526 get.addFinalizer(() => clearTimeout(handle));
527 }
528 return value;
529 },
530 { initialValueTarget: self },
531 );
532
533export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>(
534 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
535 options: EnvironmentQueryAtomOptions<
536 Input,
537 A,
538 E,
539 | EnvironmentSupervisor.EnvironmentSupervisor
540 | EnvironmentRegistry.EnvironmentRegistry
541 | AtomRegistry.AtomRegistry
542 | R
543 >,
544): (target: {
545 readonly environmentId: EnvironmentIdType;
546 readonly input: Input;
547}) => Atom.Atom<AsyncResult.AsyncResult<A, E | ER | Error>> {
548 const connectionAtom = Atom.family((environmentId: EnvironmentIdType) =>
549 runtime.atom(
550 followStreamInEnvironment(
551 environmentId,
552 Stream.unwrap(
553 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
554 Effect.map((supervisor) =>
555 SubscriptionRef.changes(supervisor.state).pipe(
556 Stream.zipLatest(SubscriptionRef.changes(supervisor.session)),
557 ),
558 ),
559 ),
560 ),
561 ),
562 { initialValue: null },
563 ),
564 );
565 const family = Atom.family((key: string) => {
566 const target = parseEnvironmentRpcKey<Input>(key);
567 const idleTtlMs = options.idleTtlMs ?? 5 * 60_000;
568 const queryAtom = runtime
569 .atom<
570 A,
571 | E
572 | ConnectionAttemptError
573 | EnvironmentRegistry.EnvironmentNotRegisteredError
574 | EnvironmentRpcUnavailableError
575 >((get) => {
576 const connection = Option.getOrNull(
577 AsyncResult.value(get(connectionAtom(target.environmentId))),
578 );
579 const emit = (value: A) =>
580 Effect.sync(() => get.setSelf(AsyncResult.success(value, { waiting: true }))).pipe(
581 // A value set while the atom builds wins over the build's own
582 // result, so a query that then settled synchronously would stay
583 // waiting forever. Suspend on a promise (`yieldNow` is drained
584 // inside the build) so the settled result arrives asynchronously.
585 Effect.andThen(Effect.promise(() => Promise.resolve())),
586 );
587 const execute = (): Effect.Effect<
588 A,
589 | E
590 | ConnectionAttemptError
591 | EnvironmentRegistry.EnvironmentNotRegisteredError
592 | EnvironmentRpcUnavailableError,
593 | EnvironmentRegistry.EnvironmentRegistry
594 | AtomRegistry.AtomRegistry
595 | Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
596 > => {
597 if (connection === null) return Effect.never;
598 const [connectionState, session] = connection;
599 if (
600 (connectionState.phase !== "connected" || Option.isNone(session)) &&
601 options.executeDisconnected
602 )
603 return runInEnvironment(
604 target.environmentId,
605 options.executeDisconnected(target.input, emit),
606 );
607 switch (connectionState.phase) {
608 case "connected":
609 return Option.isSome(session)
610 ? runInEnvironment(target.environmentId, options.execute(target.input, emit))
611 : Effect.never;
612 case "connecting":
613 case "backoff":
614 return Effect.never;
615 case "available":
616 case "offline":
617 case "blocked":
618 if (connectionState.lastFailure !== null) {
619 return Effect.fail(connectionState.lastFailure);
620 }
621 return Effect.fail(
622 new EnvironmentRpcUnavailableError({
623 environmentId: target.environmentId,
624 message: `Environment ${target.environmentId} is ${
625 connectionState.phase === "available" ? "not connected" : connectionState.phase
626 }.`,
627 }),
628 );
629 }
630 };
631 if (!options.persistence) return execute();
632 const persistence = options.persistence;
633 const cacheKey = JSON.stringify([options.label, target.input]);
634 const previous = Option.flatMap(
635 get.self<AsyncResult.AsyncResult<A, unknown>>(),
636 AsyncResult.value,
637 );
638 return Effect.gen(function* () {
639 const cached = Option.isSome(previous)
640 ? previous
641 : yield* persistence.load(target.environmentId, cacheKey);
642 if (
643 Option.isSome(cached) &&
644 (connection === null || connection[0].phase !== "connected")
645 ) {
646 return cached.value;
647 }
648 if (Option.isNone(previous) && Option.isSome(cached)) yield* emit(cached.value);
649 return yield* execute().pipe(
650 Effect.tap((value) => persistence.save(target.environmentId, cacheKey, value)),
651 );
652 });
653 })
654 .pipe(
655 Atom.swr({
656 staleTime: options.staleTimeMs ?? 30_000,
657 revalidateOnMount: true,
658 }),
659 Atom.setIdleTTL(idleTtlMs),
660 );
661 const isFinal = (result: AsyncResult.AsyncResult<A, unknown>) =>
662 options.isFinal !== undefined && Option.exists(AsyncResult.value(result), options.isFinal);
663 const intervalQuery =
664 options.refreshIntervalMs === undefined
665 ? queryAtom
666 : queryAtom.pipe(refreshEveryUntilFinal(options.refreshIntervalMs, isFinal));
667 const refreshTrigger = options.refreshTrigger?.(target);
668 return (
669 refreshTrigger === undefined
670 ? intervalQuery
671 : intervalQuery.pipe(refreshOnSignalWhileRead(refreshTrigger, isFinal))
672 ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`));
673 });
674 return (target) => family(environmentRpcKey(target));
675}
676
677export function createEnvironmentSubscriptionAtomFamily<R, ER, Input, A, E>(
678 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
679 options: EnvironmentSubscriptionAtomOptions<
680 Input,
681 A,
682 E,
683 EnvironmentSupervisor.EnvironmentSupervisor | R
684 >,
685) {
686 const family = Atom.family((key: string) => {
687 const target = parseEnvironmentRpcKey<Input>(key);
688 return runtime
689 .atom(
690 followStreamInEnvironment(target.environmentId, options.subscribe(target.input)).pipe(
691 options.completeWhen ? Stream.takeUntil(options.completeWhen) : (stream) => stream,
692 ),
693 )
694 .pipe(
695 Atom.setIdleTTL(options.idleTtlMs ?? 5 * 60_000),
696 Atom.withLabel(
697 options.sensitiveInput
698 ? `${options.label}:${target.environmentId}`
699 : `${options.label}:${key}`,
700 ),
701 );
702 });
703 return (target: { readonly environmentId: EnvironmentIdType; readonly input: Input }) =>
704 family(environmentRpcKey(target));
705}
706
707export function createEnvironmentCommand<R, ER, Input, A, E>(
708 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
709 options: EnvironmentCommandAtomOptions<
710 Input,
711 A,
712 E,
713 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry | R
714 >,
715) {
716 return createRuntimeCommand(runtime, {
717 label: options.label,
718 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
719 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
720 execute: (target, registry) =>
721 runInEnvironment(
722 target.environmentId,
723 options.execute(target.input, registry, target.environmentId),
724 ),
725 });
726}
727
728export function createEnvironmentRpcQueryAtomFamily<
729 R,
730 ER,
731 TTag extends Exclude<EnvironmentUnaryRpcTag, ClientGuardedRpcTag>,
732>(
733 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
734 options: {
735 readonly label: string;
736 readonly tag: TTag;
737 readonly execute?: (
738 input: EnvironmentRpcInput<TTag>,
739 emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>,
740 ) => Effect.Effect<
741 EnvironmentRpcSuccess<TTag>,
742 EnvironmentRpcFailure<TTag> | EnvironmentRpcUnavailableError,
743 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry
744 >;
745 readonly persistence?: QueryPersistence<EnvironmentRpcSuccess<TTag>>;
746 readonly staleTimeMs?: number;
747 readonly idleTtlMs?: number;
748 readonly refreshIntervalMs?: number;
749 readonly refreshTrigger?: (target: {
750 readonly environmentId: EnvironmentIdType;
751 readonly input: EnvironmentRpcInput<TTag>;
752 }) => Atom.Atom<unknown> | undefined;
753 readonly isFinal?: (value: EnvironmentRpcSuccess<TTag>) => boolean;
754 },
755) {
756 return createEnvironmentQueryAtomFamily(runtime, {
757 label: options.label,
758 ...(options.persistence === undefined ? {} : { persistence: options.persistence }),
759 ...(options.staleTimeMs === undefined ? {} : { staleTimeMs: options.staleTimeMs }),
760 ...(options.idleTtlMs === undefined ? {} : { idleTtlMs: options.idleTtlMs }),
761 ...(options.refreshIntervalMs === undefined
762 ? {}
763 : { refreshIntervalMs: options.refreshIntervalMs }),
764 ...(options.refreshTrigger === undefined ? {} : { refreshTrigger: options.refreshTrigger }),
765 ...(options.isFinal === undefined ? {} : { isFinal: options.isFinal }),
766 execute: (
767 input: EnvironmentRpcInput<TTag>,
768 emit: (value: EnvironmentRpcSuccess<TTag>) => Effect.Effect<void>,
769 ) => options.execute?.(input, emit) ?? request(options.tag, input),
770 });
771}
772
773export function createEnvironmentRpcSubscriptionAtomFamily<
774 R,
775 ER,
776 TTag extends EnvironmentSubscriptionRpcTag,
777 B = EnvironmentRpcStreamValue<TTag>,
778>(
779 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
780 options: {
781 readonly label: string;
782 readonly tag: TTag;
783 readonly idleTtlMs?: number;
784 readonly completeWhen?: (value: B) => boolean;
785 readonly transform?: (
786 stream: Stream.Stream<
787 EnvironmentRpcStreamValue<TTag>,
788 EnvironmentRpcStreamFailure<TTag>,
789 EnvironmentSupervisor.EnvironmentSupervisor | R
790 >,
791 ) => Stream.Stream<
792 B,
793 EnvironmentRpcStreamFailure<TTag>,
794 EnvironmentSupervisor.EnvironmentSupervisor | R
795 >;
796 },
797) {
798 return createEnvironmentSubscriptionAtomFamily(runtime, {
799 label: options.label,
800 ...(options.completeWhen === undefined ? {} : { completeWhen: options.completeWhen }),
801 ...(options.idleTtlMs === undefined ? {} : { idleTtlMs: options.idleTtlMs }),
802 subscribe: (input: EnvironmentRpcInput<TTag>) => {
803 const stream = subscribe(options.tag, input);
804 return options.transform === undefined
805 ? (stream as Stream.Stream<
806 B,
807 EnvironmentRpcStreamFailure<TTag>,
808 EnvironmentSupervisor.EnvironmentSupervisor | R
809 >)
810 : options.transform(stream);
811 },
812 });
813}
814
815export function createEnvironmentRpcCommand<
816 R,
817 ER,
818 TTag extends EnvironmentUnaryRpcTag,
819 Input extends EnvironmentRpcInput<TTag> = EnvironmentRpcInput<TTag>,
820>(
821 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
822 options: {
823 readonly label: string;
824 readonly tag: TTag;
825 readonly execute?: (
826 input: Input,
827 ) => Effect.Effect<
828 EnvironmentRpcSuccess<TTag>,
829 EnvironmentRpcFailure<TTag> | EnvironmentRpcUnavailableError,
830 EnvironmentSupervisor.EnvironmentSupervisor | EnvironmentRegistry.EnvironmentRegistry
831 >;
832 readonly scheduler?: AtomCommandScheduler;
833 readonly concurrency?: AtomCommandConcurrency<{
834 readonly environmentId: EnvironmentIdType;
835 readonly input: NoInfer<Input>;
836 }>;
837 readonly onSuccess?: (
838 target: {
839 readonly environmentId: EnvironmentIdType;
840 readonly input: NoInfer<Input>;
841 },
842 registry: AtomRegistry.AtomRegistry,
843 result: EnvironmentRpcSuccess<TTag>,
844 ) => Effect.Effect<void, never, R>;
845 readonly onSettled?: (
846 target: {
847 readonly environmentId: EnvironmentIdType;
848 readonly input: NoInfer<Input>;
849 },
850 registry: AtomRegistry.AtomRegistry,
851 ) => Effect.Effect<void, never, R>;
852 },
853) {
854 const permissions = createCommandPermissions(runtime, options.tag);
855 const command = createEnvironmentCommand(runtime, {
856 label: options.label,
857 ...(options.scheduler === undefined ? {} : { scheduler: options.scheduler }),
858 ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }),
859 execute: (input: Input, registry, environmentId) => {
860 const target = {
861 environmentId,
862 input,
863 };
864 // Routing requires consent on the origin as well as the actual destination.
865 // The transport check below deliberately checks the destination again.
866 return permissions.authorize(registry, environmentId, input).pipe(
867 Effect.andThen(() => options.execute?.(input) ?? requestGuarded(options.tag, input)),
868 Effect.provideService(RpcPermissionGuard, {
869 authorize: (id, method, payload) =>
870 createCommandPermissions(runtime, method).authorize(registry, id, payload),
871 }),
872 Effect.tap((result) => options.onSuccess?.(target, registry, result) ?? Effect.void),
873 Effect.ensuring(options.onSettled?.(target, registry) ?? Effect.void),
874 );
875 },
876 });
877 return {
878 ...command,
879 requiredScopes: permissions.requiredScopes,
880 permissionAtom: permissions.permissionAtom,
881 };
882}