1import {
2 ClientCapabilities,
3 PlatformConnectionSource,
4 Persistence,
5} from "@t3tools/client-runtime/platform";
6import {
7 BearerConnectionCredential,
8 BearerConnectionProfile,
9 BearerConnectionRegistration,
10 BearerConnectionTarget,
11 ConnectionBlockedError,
12 type ConnectionAttemptError,
13 ConnectionTransientError,
14 Connectivity,
15 mapRemoteEnvironmentError,
16 type PlatformConnectionRegistration,
17 PrimaryConnectionRegistration,
18 PrimaryConnectionTarget,
19 sshTargetKey,
20 Wakeups,
21} from "@t3tools/client-runtime/connection";
22import { bootstrapRemoteBearerSession } from "@t3tools/client-runtime/authorization";
23import { fetchRemoteEnvironmentDescriptor } from "@t3tools/client-runtime/environment";
24import { managedRelayAccountChanges, managedRelaySessionAtom } from "@t3tools/client-runtime/relay";
25import { EnvironmentRpcRequestObserver } from "@t3tools/client-runtime/rpc";
26import {
27 AuthStandardClientScopes,
28 type DesktopBridge,
29 type DesktopEnvironmentBootstrap,
30 type DesktopSshEnvironmentBootstrap,
31 type DesktopSshEnvironmentTarget,
32 type EnvironmentId,
33 PRIMARY_LOCAL_ENVIRONMENT_ID,
34} from "@t3tools/contracts";
35import * as Clock from "effect/Clock";
36import * as Context from "effect/Context";
37import * as Effect from "effect/Effect";
38import * as Layer from "effect/Layer";
39import * as Option from "effect/Option";
40import * as Queue from "effect/Queue";
41import * as Ref from "effect/Ref";
42import * as Stream from "effect/Stream";
43import { FetchHttpClient } from "effect/http";
44
45import { APP_VERSION } from "../branding";
46import { readDesktopPrimaryBearerToken } from "../environments/primary/desktopAuth";
47import * as PrimaryEnvironmentHttpLayer from "../environments/primary/httpLayer";
48import {
49 readPrimaryEnvironmentTarget,
50 type PrimaryEnvironmentTarget,
51} from "../environments/primary/target";
52import { clearComposerDraftsEnvironment } from "../composerDraftStore";
53import { isHostedStaticApp } from "../hostedPairing";
54import { isLocalEnvironmentDisabled } from "../localEnvironment";
55import { randomUUID } from "../lib/utils";
56import { appAtomRegistry } from "../rpc/atomRegistry";
57import { acknowledgeRpcRequest, trackRpcRequestSent } from "../rpc/requestLatencyState";
58import {
59 desktopLocalConnectionId,
60 readDesktopSecondaryBootstrapsResult,
61 type DesktopSecondaryBootstrapsRead,
62} from "./desktopLocal";
63import * as ConnectionStorage from "./storage";
64import { clientPresentationMetadata } from "./clientMetadata";
65
66let nextObservedRpcRequestId = 0;
67
68function currentNetworkStatus(): "unknown" | "offline" | "online" {
69 if (typeof navigator === "undefined") {
70 return "unknown";
71 }
72 return navigator.onLine ? "online" : "offline";
73}
74
75const layerConnectivity = Connectivity.layer({
76 status: Effect.sync(currentNetworkStatus),
77 changes: Stream.callback((queue) =>
78 Effect.acquireRelease(
79 Effect.sync(() => {
80 const online = () => Queue.offerUnsafe(queue, "online");
81 const offline = () => Queue.offerUnsafe(queue, "offline");
82 window.addEventListener("online", online);
83 window.addEventListener("offline", offline);
84 return { online, offline };
85 }),
86 ({ online, offline }) =>
87 Effect.sync(() => {
88 window.removeEventListener("online", online);
89 window.removeEventListener("offline", offline);
90 }),
91 ).pipe(Effect.asVoid),
92 ),
93});
94
95interface NetworkInformationLike extends EventTarget {
96 readonly type?: string;
97}
98
99// change also fires for same-network bandwidth and latency estimates; count only type changes.
100const networkPathChanges = Stream.callback<"network-changed">((queue) =>
101 Effect.acquireRelease(
102 Effect.sync(() => {
103 const connection =
104 typeof navigator === "undefined"
105 ? undefined
106 : (navigator as Navigator & { readonly connection?: NetworkInformationLike }).connection;
107 if (connection?.type === undefined) return undefined;
108 let previous = connection.type;
109 const listener = () => {
110 const type = connection.type;
111 if (type === undefined || type === previous) return;
112 previous = type;
113 Queue.offerUnsafe(queue, "network-changed");
114 };
115 connection.addEventListener("change", listener);
116 return { connection, listener };
117 }),
118 (subscription) =>
119 Effect.sync(() =>
120 subscription?.connection.removeEventListener("change", subscription.listener),
121 ),
122 ).pipe(Effect.asVoid),
123);
124
125const layerWakeups = Wakeups.layer({
126 changes: Stream.mergeAll(
127 [
128 Stream.callback<"application-active">((queue) =>
129 Effect.acquireRelease(
130 Effect.sync(() => {
131 const listener = () => {
132 if (document.visibilityState === "visible") {
133 Queue.offerUnsafe(queue, "application-active");
134 }
135 };
136 document.addEventListener("visibilitychange", listener);
137 return listener;
138 }),
139 (listener) =>
140 Effect.sync(() => {
141 document.removeEventListener("visibilitychange", listener);
142 }),
143 ).pipe(Effect.asVoid),
144 ),
145 managedRelayAccountChanges(appAtomRegistry).pipe(
146 Stream.map(() => "credentials-changed" as const),
147 ),
148 networkPathChanges,
149 ],
150 { concurrency: "unbounded" },
151 ),
152});
153
154export function currentClientMetadata() {
155 return clientPresentationMetadata({
156 appVersion: APP_VERSION,
157 hosted: isHostedStaticApp(),
158 identity: {
159 userAgent: navigator.userAgent,
160 platform: navigator.platform,
161 maxTouchPoints: navigator.maxTouchPoints,
162 },
163 desktopBridge: window.desktopBridge,
164 });
165}
166
167function sshPreparationError(cause: unknown) {
168 const message = cause instanceof Error ? cause.message : String(cause);
169 if (message.toLowerCase().includes("cancel")) {
170 return new ConnectionBlockedError({
171 reason: "authentication",
172 detail: message,
173 });
174 }
175 return new ConnectionTransientError({
176 reason: "remote-unavailable",
177 detail: `Could not prepare the SSH environment: ${message}`,
178 });
179}
180
181// IPC keeps running after its waiter is interrupted. A retry can adopt only
182// that interrupted waiter's unfinished setup.
183const pendingSshSetups = new Map<
184 string,
185 { readonly promise: Promise<DesktopSshEnvironmentBootstrap>; signal: AbortSignal }
186>();
187
188const ensureDesktopSshEnvironment = (bridge: DesktopBridge, target: DesktopSshEnvironmentTarget) =>
189 Effect.tryPromise({
190 try: (signal) => {
191 const key = sshTargetKey(target);
192 const pending = pendingSshSetups.get(key);
193 if (pending?.signal.aborted) {
194 pending.signal = signal;
195 return pending.promise;
196 }
197 const setup = {
198 promise: bridge.ensureSshEnvironment(target, { issuePairingToken: true }),
199 signal,
200 };
201 pendingSshSetups.set(key, setup);
202 const forget = () => {
203 if (pendingSshSetups.get(key) === setup) pendingSshSetups.delete(key);
204 };
205 setup.promise.then(forget, forget);
206 return setup.promise;
207 },
208 catch: sshPreparationError,
209 });
210
211export const provisionDesktopSshEnvironment = Effect.fn(
212 "web.connectionPlatform.ssh.provisionDesktop",
213)(function* (
214 bridge: DesktopBridge,
215 target: DesktopSshEnvironmentTarget,
216 expectedEnvironmentId?: EnvironmentId,
217) {
218 const bootstrap = yield* ensureDesktopSshEnvironment(bridge, target);
219 const pairingToken = bootstrap.pairingToken;
220 if (pairingToken === null) {
221 return yield* new ConnectionBlockedError({
222 reason: "authentication",
223 detail: "The SSH environment did not issue a pairing credential.",
224 });
225 }
226 const descriptor = yield* Effect.tryPromise({
227 try: () => bridge.fetchSshEnvironmentDescriptor(bootstrap.httpBaseUrl),
228 catch: sshPreparationError,
229 });
230 if (expectedEnvironmentId !== undefined && descriptor.environmentId !== expectedEnvironmentId) {
231 return yield* new ConnectionBlockedError({
232 reason: "configuration",
233 detail: `That host reaches ${descriptor.label}, a different machine. Add it as its own environment instead.`,
234 });
235 }
236 const access = yield* Effect.tryPromise({
237 try: () => bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, pairingToken),
238 catch: (cause) => {
239 const error = sshPreparationError(cause);
240 // A reused remote server on another build can fail to decode the grant
241 // the desktop's runner just minted, and answers 500 on every retry.
242 if (
243 error._tag === "ConnectionBlockedError" ||
244 descriptor.serverVersion === APP_VERSION ||
245 !error.detail.includes("[ssh_http:500]")
246 ) {
247 return error;
248 }
249 return new ConnectionBlockedError({
250 reason: "configuration",
251 detail: `${descriptor.label} runs T3 Code ${descriptor.serverVersion}, but this desktop is ${APP_VERSION}, and that server could not accept its pairing. Restart the T3 server on ${descriptor.label} so it runs ${APP_VERSION}, then retry.`,
252 });
253 },
254 });
255 return {
256 environmentId: descriptor.environmentId,
257 label: descriptor.label,
258 bootstrap,
259 bearerToken: access.access_token,
260 };
261});
262
263/** Reuses SSH pairing tokens with all standard scopes to avoid adding a session on each connection. */
264export const prepareDesktopSshEnvironment = Effect.fn("web.connectionPlatform.ssh.prepareDesktop")(
265 function* (
266 bridge: DesktopBridge,
267 input: {
268 readonly target: DesktopSshEnvironmentTarget;
269 readonly expectedEnvironmentId: EnvironmentId;
270 readonly bearerToken?: string;
271 },
272 ) {
273 const bearerToken = input.bearerToken;
274 if (bearerToken !== undefined) {
275 const bootstrap = yield* Effect.tryPromise({
276 try: () => bridge.ensureSshEnvironment(input.target, { issuePairingToken: false }),
277 catch: sshPreparationError,
278 });
279 // Check identity before sending saved credentials or pairing.
280 const descriptor = yield* Effect.tryPromise({
281 try: () => bridge.fetchSshEnvironmentDescriptor(bootstrap.httpBaseUrl),
282 catch: sshPreparationError,
283 });
284 if (descriptor.environmentId !== input.expectedEnvironmentId) {
285 return yield* new ConnectionBlockedError({
286 reason: "configuration",
287 detail: `That host reaches ${descriptor.label}, a different machine. Add it as its own environment instead.`,
288 });
289 }
290 const session = yield* Effect.tryPromise({
291 try: () => bridge.fetchSshSessionState(bootstrap.httpBaseUrl, bearerToken),
292 catch: sshPreparationError,
293 });
294 // Stored grants never widen; re-pair tokens missing newer standard scopes.
295 const granted = session.permissions ?? session.scopes ?? [];
296 if (
297 session.authenticated &&
298 AuthStandardClientScopes.every((scope) => granted.includes(scope))
299 ) {
300 return { bootstrap, bearerToken };
301 }
302 }
303 const provisioned = yield* provisionDesktopSshEnvironment(
304 bridge,
305 input.target,
306 input.expectedEnvironmentId,
307 );
308 return { bootstrap: provisioned.bootstrap, bearerToken: provisioned.bearerToken };
309 },
310);
311
312const ANALYTICS_DEVICE_ID_KEY = "t3code:analytics-device-id";
313
314// Lives as long as this browser profile's site data. Storage can be blocked, which only drops the id.
315const analyticsDevice = {
316 id: Effect.sync(
317 Option.liftThrowable(() => {
318 const existing = window.localStorage.getItem(ANALYTICS_DEVICE_ID_KEY);
319 if (existing) return existing;
320 const created = randomUUID();
321 window.localStorage.setItem(ANALYTICS_DEVICE_ID_KEY, created);
322 return created;
323 }),
324 ),
325};
326
327const layerCapabilities = Layer.effectContext(
328 Effect.sync(() => {
329 const presentation = ClientCapabilities.ClientPresentation.of({
330 metadata: currentClientMetadata(),
331 });
332 const cloudSession = ClientCapabilities.CloudSession.of({
333 identity: Effect.sync(() =>
334 Option.fromNullishOr(appAtomRegistry.get(managedRelaySessionAtom)),
335 ),
336 clerkToken: Effect.gen(function* () {
337 const session = appAtomRegistry.get(managedRelaySessionAtom);
338 if (session === null) {
339 return yield* new ConnectionBlockedError({
340 reason: "authentication",
341 detail: "Sign in to T3 Connect to connect this environment.",
342 });
343 }
344 const token = yield* session.readClerkToken().pipe(
345 Effect.mapError(
346 (error) =>
347 new ConnectionTransientError({
348 reason: "network",
349 detail: error.message,
350 }),
351 ),
352 );
353 if (token === null) {
354 return yield* new ConnectionBlockedError({
355 reason: "authentication",
356 detail: "The T3 Connect session is unavailable.",
357 });
358 }
359 return token;
360 }),
361 });
362 const identity = ClientCapabilities.RelayDeviceIdentity.of({
363 deviceId: Effect.succeedNone,
364 });
365 const primaryAuth = ClientCapabilities.PrimaryEnvironmentAuth.of({
366 bearerToken: Effect.tryPromise({
367 try: readDesktopPrimaryBearerToken,
368 catch: (cause) =>
369 new ConnectionTransientError({
370 reason: "remote-unavailable",
371 detail: `Could not load the desktop primary credential: ${String(cause)}`,
372 }),
373 }).pipe(Effect.map(Option.fromNullishOr)),
374 });
375 const ssh = ClientCapabilities.SshEnvironmentGateway.of({
376 provision: Effect.fn("web.connectionPlatform.ssh.provision")(
377 function* (target, expectedEnvironmentId) {
378 const bridge = window.desktopBridge;
379 if (bridge === undefined) {
380 return yield* new ConnectionBlockedError({
381 reason: "unsupported",
382 detail: "SSH environments are only available in the desktop app.",
383 });
384 }
385 return yield* provisionDesktopSshEnvironment(bridge, target, expectedEnvironmentId);
386 },
387 ),
388 prepare: Effect.fn("web.connectionPlatform.ssh.prepare")(function* (input) {
389 const bridge = window.desktopBridge;
390 if (bridge === undefined) {
391 return yield* new ConnectionBlockedError({
392 reason: "unsupported",
393 detail: "SSH environments are only available in the desktop app.",
394 });
395 }
396 return yield* prepareDesktopSshEnvironment(bridge, input);
397 }),
398 disconnect: Effect.fn("web.connectionPlatform.ssh.disconnect")(function* (target) {
399 const bridge = window.desktopBridge;
400 if (bridge === undefined) {
401 return;
402 }
403 pendingSshSetups.delete(sshTargetKey(target));
404 yield* Effect.tryPromise({
405 try: () => bridge.disconnectSshEnvironment(target),
406 catch: (cause) =>
407 new ConnectionTransientError({
408 reason: "remote-unavailable",
409 detail: `Could not disconnect the SSH environment: ${String(cause)}`,
410 }),
411 });
412 }),
413 });
414
415 return Context.make(ClientCapabilities.CloudSession, cloudSession).pipe(
416 Context.add(ClientCapabilities.AnalyticsDevice, analyticsDevice),
417 Context.add(ClientCapabilities.PrimaryEnvironmentAuth, primaryAuth),
418 Context.add(ClientCapabilities.RelayDeviceIdentity, identity),
419 Context.add(ClientCapabilities.ClientPresentation, presentation),
420 Context.add(ClientCapabilities.SshEnvironmentGateway, ssh),
421 );
422 }),
423);
424
425const loadPrimaryConnectionRegistration = Effect.fn(
426 "web.connectionPlatform.loadPrimaryConnectionRegistration",
427)(function* (resolved: PrimaryEnvironmentTarget) {
428 const descriptor = yield* fetchRemoteEnvironmentDescriptor({
429 httpBaseUrl: resolved.target.httpBaseUrl,
430 }).pipe(
431 Effect.provide(PrimaryEnvironmentHttpLayer.layer),
432 Effect.mapError(mapRemoteEnvironmentError),
433 );
434 return new PrimaryConnectionRegistration({
435 target: new PrimaryConnectionTarget({
436 environmentId: descriptor.environmentId,
437 label: descriptor.label,
438 httpBaseUrl: resolved.target.httpBaseUrl,
439 wsBaseUrl: resolved.target.wsBaseUrl,
440 }),
441 });
442});
443
444const loadSecondaryConnectionRegistration = Effect.fn(
445 "web.connectionPlatform.loadSecondaryConnectionRegistration",
446)(function* (entry: DesktopEnvironmentBootstrap) {
447 if (
448 entry.httpBaseUrl === null ||
449 entry.wsBaseUrl === null ||
450 entry.bootstrapToken === undefined
451 ) {
452 return yield* new ConnectionTransientError({
453 reason: "endpoint-unavailable",
454 detail: `Desktop-local backend ${entry.id} is not ready yet.`,
455 });
456 }
457 const httpBaseUrl = entry.httpBaseUrl;
458 const wsBaseUrl = entry.wsBaseUrl;
459 const descriptor = yield* fetchRemoteEnvironmentDescriptor({ httpBaseUrl }).pipe(
460 Effect.mapError(mapRemoteEnvironmentError),
461 );
462 const issuedAtEpochMs = yield* Clock.currentTimeMillis;
463 const access = yield* bootstrapRemoteBearerSession({
464 httpBaseUrl,
465 credential: entry.bootstrapToken,
466 scopes: AuthStandardClientScopes,
467 clientMetadata: currentClientMetadata(),
468 }).pipe(Effect.mapError(mapRemoteEnvironmentError));
469 // Desktop operations need the pool backend id to reach the owning instance; RPC uses the environment id.
470 const connectionId = desktopLocalConnectionId(entry.id);
471 const label = entry.label || descriptor.label;
472 return {
473 registration: new BearerConnectionRegistration({
474 target: new BearerConnectionTarget({
475 environmentId: descriptor.environmentId,
476 label,
477 connectionId,
478 }),
479 profile: new BearerConnectionProfile({
480 connectionId,
481 environmentId: descriptor.environmentId,
482 label,
483 httpBaseUrl,
484 wsBaseUrl,
485 }),
486 credential: new BearerConnectionCredential({ token: access.access_token }),
487 }),
488 expiresAtEpochMs: secondaryBearerExpiresAtEpochMs(issuedAtEpochMs, access.expires_in),
489 refreshAtEpochMs: secondaryBearerRefreshAtEpochMs(issuedAtEpochMs, access.expires_in),
490 };
491});
492
493// The bridge has no topology event. Cache registrations until bearer credentials near expiry.
494const PLATFORM_POLL_INTERVAL = "3 seconds";
495const SECONDARY_BEARER_REFRESH_SKEW_MS = 5_000;
496
497function secondaryBearerExpiresAtEpochMs(
498 issuedAtEpochMs: number,
499 expiresInSeconds: number,
500): number {
501 return issuedAtEpochMs + Math.max(0, expiresInSeconds * 1_000);
502}
503
504function secondaryBearerRefreshAtEpochMs(
505 issuedAtEpochMs: number,
506 expiresInSeconds: number,
507): number {
508 return Math.max(
509 issuedAtEpochMs,
510 secondaryBearerExpiresAtEpochMs(issuedAtEpochMs, expiresInSeconds) -
511 SECONDARY_BEARER_REFRESH_SKEW_MS,
512 );
513}
514
515interface CachedPlatformRegistration {
516 readonly signature: string;
517 readonly registration: PlatformConnectionRegistration;
518 readonly expiresAtEpochMs?: number;
519 readonly refreshAtEpochMs?: number;
520 readonly bootstrapToken?: string;
521}
522
523export type PrimaryEnvironmentTargetRead =
524 | {
525 readonly _tag: "Success";
526 readonly target: PrimaryEnvironmentTarget | null;
527 }
528 | {
529 readonly _tag: "Failure";
530 readonly cause: unknown;
531 };
532
533function readPrimaryEnvironmentTargetResult(
534 readTarget: () => PrimaryEnvironmentTarget | null = readPrimaryEnvironmentTarget,
535): PrimaryEnvironmentTargetRead {
536 try {
537 return { _tag: "Success", target: readTarget() };
538 } catch (cause) {
539 return { _tag: "Failure", cause };
540 }
541}
542
543export function primaryRegistrationToRetainAfterTopologyRead(
544 previous: ReadonlyMap<string, CachedPlatformRegistration>,
545 topologyRead: PrimaryEnvironmentTargetRead,
546): CachedPlatformRegistration | undefined {
547 return topologyRead._tag === "Failure" ? previous.get(PRIMARY_LOCAL_ENVIRONMENT_ID) : undefined;
548}
549
550export function canReuseCachedPlatformRegistration(
551 cached: CachedPlatformRegistration,
552 signature: string,
553 nowEpochMs: number,
554): boolean {
555 return (
556 cached.signature === signature &&
557 (cached.refreshAtEpochMs === undefined || nowEpochMs < cached.refreshAtEpochMs)
558 );
559}
560
561export function canRetainCachedPlatformRegistrationAfterRefreshFailure(
562 cached: CachedPlatformRegistration,
563 signature: string,
564 nowEpochMs: number,
565): boolean {
566 return (
567 cached.signature === signature &&
568 cached.expiresAtEpochMs !== undefined &&
569 nowEpochMs < cached.expiresAtEpochMs
570 );
571}
572
573const REJECTED_BOOTSTRAP_RETRY_INITIAL_MS = 60_000;
574const REJECTED_BOOTSTRAP_RETRY_MAX_MS = 30 * 60_000;
575
576export interface RejectedSecondaryBootstrap {
577 readonly signature: string;
578 readonly retryAtEpochMs: number;
579 readonly delayMs: number;
580}
581
582/** Retry with backoff: a backend can restart on the same port with a fresh grant and no topology change. */
583export function isRejectedSecondaryBootstrap(
584 rejected: RejectedSecondaryBootstrap | undefined,
585 signature: string,
586 nowEpochMs: number,
587): rejected is RejectedSecondaryBootstrap {
588 return (
589 rejected !== undefined &&
590 rejected.signature === signature &&
591 nowEpochMs < rejected.retryAtEpochMs
592 );
593}
594
595export function nextRejectedSecondaryBootstrap(
596 previous: RejectedSecondaryBootstrap | undefined,
597 signature: string,
598 nowEpochMs: number,
599): RejectedSecondaryBootstrap {
600 const delayMs =
601 previous?.signature === signature
602 ? Math.min(previous.delayMs * 2, REJECTED_BOOTSTRAP_RETRY_MAX_MS)
603 : REJECTED_BOOTSTRAP_RETRY_INITIAL_MS;
604 return { signature, retryAtEpochMs: nowEpochMs + delayMs, delayMs };
605}
606
607export function isRejectedBootstrapCredentialError(error: ConnectionAttemptError): boolean {
608 return error._tag === "ConnectionBlockedError" && error.reason === "authentication";
609}
610
611export function secondaryRegistrationsToRetainAfterTopologyRead(
612 previous: ReadonlyMap<string, CachedPlatformRegistration>,
613 topologyRead: DesktopSecondaryBootstrapsRead,
614 nowEpochMs: number,
615): ReadonlyMap<string, CachedPlatformRegistration> {
616 if (topologyRead._tag === "Success") {
617 return new Map();
618 }
619 return new Map(
620 [...previous].filter(
621 ([, cached]) => cached.expiresAtEpochMs !== undefined && nowEpochMs < cached.expiresAtEpochMs,
622 ),
623 );
624}
625
626const layerPlatformConnectionSource = Layer.effect(
627 PlatformConnectionSource.PlatformConnectionSource,
628 Effect.gen(function* () {
629 if (isHostedStaticApp() || isLocalEnvironmentDisabled()) {
630 return PlatformConnectionSource.PlatformConnectionSource.of({
631 registrations: Stream.empty,
632 });
633 }
634 const cacheRef = yield* Ref.make(new Map<string, CachedPlatformRegistration>());
635 const rejectedRef = yield* Ref.make(new Map<string, RejectedSecondaryBootstrap>());
636
637 const buildPlatformRegistrations = Effect.gen(function* () {
638 const previous = yield* Ref.get(cacheRef);
639 const nowEpochMs = yield* Clock.currentTimeMillis;
640 const next = new Map<string, CachedPlatformRegistration>();
641 const registrations: Array<PlatformConnectionRegistration> = [];
642
643 const primaryTopologyRead = readPrimaryEnvironmentTargetResult();
644 const retainedPrimary = primaryRegistrationToRetainAfterTopologyRead(
645 previous,
646 primaryTopologyRead,
647 );
648 if (retainedPrimary !== undefined) {
649 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, retainedPrimary);
650 registrations.push(retainedPrimary.registration);
651 }
652
653 if (primaryTopologyRead._tag === "Failure") {
654 yield* Effect.logWarning("Could not read the primary environment topology.", {
655 cause: primaryTopologyRead.cause,
656 });
657 } else if (primaryTopologyRead.target !== null) {
658 const primaryTarget = primaryTopologyRead.target;
659 const signature = `primary|${primaryTarget.target.httpBaseUrl}|${primaryTarget.target.wsBaseUrl}`;
660 const cached = previous.get(PRIMARY_LOCAL_ENVIRONMENT_ID);
661 if (
662 cached !== undefined &&
663 canReuseCachedPlatformRegistration(cached, signature, nowEpochMs)
664 ) {
665 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, cached);
666 registrations.push(cached.registration);
667 } else {
668 const built = yield* loadPrimaryConnectionRegistration(primaryTarget).pipe(
669 Effect.tapError((error) =>
670 Effect.logWarning("Could not discover the primary environment.", { error }),
671 ),
672 Effect.option,
673 );
674 if (Option.isSome(built)) {
675 const cacheEntry = { signature, registration: built.value };
676 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, cacheEntry);
677 registrations.push(built.value);
678 }
679 }
680 }
681
682 const topologyRead = readDesktopSecondaryBootstrapsResult();
683 for (const [id, cached] of secondaryRegistrationsToRetainAfterTopologyRead(
684 previous,
685 topologyRead,
686 nowEpochMs,
687 )) {
688 next.set(id, cached);
689 registrations.push(cached.registration);
690 }
691
692 if (topologyRead._tag === "Failure") {
693 yield* Effect.logWarning("Could not read the desktop-local backend topology.", {
694 cause: topologyRead.cause,
695 });
696 } else {
697 const rejected = yield* Ref.get(rejectedRef);
698 const nextRejected = new Map<string, RejectedSecondaryBootstrap>();
699 for (const bootstrap of topologyRead.bootstraps) {
700 // Token rotation only controls exchange retries; dropping endpoint bearers also clears drafts.
701 const endpointSignature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}`;
702 const signature = `${endpointSignature}|${bootstrap.bootstrapToken ?? ""}`;
703 const cached = previous.get(bootstrap.id);
704 // Rotation may mean a restart; retain the old bearer if the new exchange fails.
705 if (
706 cached !== undefined &&
707 cached.bootstrapToken === bootstrap.bootstrapToken &&
708 canReuseCachedPlatformRegistration(cached, endpointSignature, nowEpochMs)
709 ) {
710 next.set(bootstrap.id, cached);
711 registrations.push(cached.registration);
712 continue;
713 }
714 const previouslyRejected = rejected.get(bootstrap.id);
715 if (isRejectedSecondaryBootstrap(previouslyRejected, signature, nowEpochMs)) {
716 nextRejected.set(bootstrap.id, previouslyRejected);
717 if (
718 cached !== undefined &&
719 canRetainCachedPlatformRegistrationAfterRefreshFailure(
720 cached,
721 endpointSignature,
722 nowEpochMs,
723 )
724 ) {
725 next.set(bootstrap.id, cached);
726 registrations.push(cached.registration);
727 }
728 continue;
729 }
730 const built = yield* loadSecondaryConnectionRegistration(bootstrap).pipe(
731 Effect.tapError((error) =>
732 Effect.logWarning("Could not connect a desktop-local backend.", {
733 id: bootstrap.id,
734 error,
735 }),
736 ),
737 Effect.tapError((error) =>
738 isRejectedBootstrapCredentialError(error)
739 ? // Start backoff at rejection; this poll can outlast the first delay.
740 Clock.currentTimeMillis.pipe(
741 Effect.map((rejectedAtEpochMs) =>
742 nextRejected.set(
743 bootstrap.id,
744 nextRejectedSecondaryBootstrap(
745 previouslyRejected,
746 signature,
747 rejectedAtEpochMs,
748 ),
749 ),
750 ),
751 )
752 : Effect.void,
753 ),
754 Effect.option,
755 );
756 if (Option.isSome(built)) {
757 const cacheEntry = {
758 signature: endpointSignature,
759 ...(bootstrap.bootstrapToken === undefined
760 ? {}
761 : { bootstrapToken: bootstrap.bootstrapToken }),
762 ...built.value,
763 };
764 next.set(bootstrap.id, cacheEntry);
765 registrations.push(built.value.registration);
766 } else if (
767 cached !== undefined &&
768 canRetainCachedPlatformRegistrationAfterRefreshFailure(
769 cached,
770 endpointSignature,
771 nowEpochMs,
772 )
773 ) {
774 next.set(bootstrap.id, cached);
775 registrations.push(cached.registration);
776 }
777 }
778 yield* Ref.set(rejectedRef, nextRejected);
779 }
780
781 yield* Ref.set(cacheRef, next);
782 return registrations as ReadonlyArray<PlatformConnectionRegistration>;
783 }).pipe(Effect.provide(FetchHttpClient.layer));
784
785 return PlatformConnectionSource.PlatformConnectionSource.of({
786 registrations: Stream.tick(PLATFORM_POLL_INTERVAL).pipe(
787 Stream.mapEffect(() => buildPlatformRegistrations),
788 ),
789 });
790 }),
791);
792
793const layerEnvironmentOwnedDataCleanup = Layer.succeed(
794 Persistence.EnvironmentOwnedDataCleanup,
795 Persistence.EnvironmentOwnedDataCleanup.of({
796 clear: Effect.fn("Persistence.EnvironmentOwnedDataCleanup.clear")(function (environmentId) {
797 return Effect.sync(() => {
798 clearComposerDraftsEnvironment(environmentId);
799 });
800 }),
801 }),
802);
803
804const layerRpcRequestObserver = Layer.succeed(
805 EnvironmentRpcRequestObserver,
806 EnvironmentRpcRequestObserver.of({
807 // Every RPC passes through this observer; the request owns its span.
808 observe: Effect.fnUntraced(function* ({ environmentId, environmentLabel, method }) {
809 return yield* Effect.sync(() => {
810 nextObservedRpcRequestId += 1;
811 const requestId = `${environmentId}:${nextObservedRpcRequestId}`;
812 trackRpcRequestSent(requestId, method, `${method} · ${environmentLabel}`);
813 return Effect.sync(() => {
814 acknowledgeRpcRequest(requestId);
815 });
816 });
817 }),
818 }),
819);
820
821type ConnectionPlatformLayerSource =
822 | typeof ConnectionStorage.layer
823 | typeof layerConnectivity
824 | typeof layerWakeups
825 | typeof layerCapabilities
826 | typeof layerPlatformConnectionSource
827 | typeof layerEnvironmentOwnedDataCleanup
828 | typeof layerRpcRequestObserver;
829
830export const layer: Layer.Layer<
831 Layer.Success<ConnectionPlatformLayerSource>,
832 Layer.Error<ConnectionPlatformLayerSource>,
833 Layer.Services<ConnectionPlatformLayerSource>
834> = Layer.mergeAll(
835 ConnectionStorage.layer,
836 layerConnectivity,
837 layerWakeups,
838 layerCapabilities,
839 layerPlatformConnectionSource,
840 layerEnvironmentOwnedDataCleanup,
841 layerRpcRequestObserver,
842);