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 Wakeups,
20} from "@t3tools/client-runtime/connection";
21import { bootstrapRemoteBearerSession } from "@t3tools/client-runtime/authorization";
22import { fetchRemoteEnvironmentDescriptor } from "@t3tools/client-runtime/environment";
23import { managedRelayAccountChanges, managedRelaySessionAtom } from "@t3tools/client-runtime/relay";
24import { EnvironmentRpcRequestObserver } from "@t3tools/client-runtime/rpc";
25import {
26 AuthStandardClientScopes,
27 type DesktopBridge,
28 type DesktopEnvironmentBootstrap,
29 type DesktopSshEnvironmentTarget,
30 type EnvironmentId,
31 PRIMARY_LOCAL_ENVIRONMENT_ID,
32} from "@t3tools/contracts";
33import * as Clock from "effect/Clock";
34import * as Context from "effect/Context";
35import * as Effect from "effect/Effect";
36import * as Layer from "effect/Layer";
37import * as Option from "effect/Option";
38import * as Queue from "effect/Queue";
39import * as Ref from "effect/Ref";
40import * as Stream from "effect/Stream";
41import { FetchHttpClient } from "effect/http";
42
43import { hostedAuthHeaders } from "./hostedAuth";
44import { APP_VERSION } from "../branding";
45import { readDesktopPrimaryBearerToken } from "../environments/primary/desktopAuth";
46import * as PrimaryEnvironmentHttpLayer from "../environments/primary/httpLayer";
47import {
48 readPrimaryEnvironmentTarget,
49 type PrimaryEnvironmentTarget,
50} from "../environments/primary/target";
51import { clearComposerDraftsEnvironment } from "../composerDraftStore";
52import { isHostedStaticApp } from "../hostedPairing";
53import { isLocalEnvironmentDisabled } from "../localEnvironment";
54import { appAtomRegistry } from "../rpc/atomRegistry";
55import { acknowledgeRpcRequest, trackRpcRequestSent } from "../rpc/requestLatencyState";
56import {
57 desktopLocalConnectionId,
58 readDesktopSecondaryBootstrapsResult,
59 type DesktopSecondaryBootstrapsRead,
60} from "./desktopLocal";
61import * as ConnectionStorage from "./storage";
62import { clientPresentationMetadata } from "./clientMetadata";
63
64let nextObservedRpcRequestId = 0;
65
66function currentNetworkStatus(): "unknown" | "offline" | "online" {
67 if (typeof navigator === "undefined") {
68 return "unknown";
69 }
70 return navigator.onLine ? "online" : "offline";
71}
72
73const layerHostedAuthAccess = Layer.succeed(ClientCapabilities.HostedAuthAccess, {
74 headers: (serverUrl) =>
75 Effect.tryPromise({
76 try: () => hostedAuthHeaders(serverUrl),
77 catch: () =>
78 new ConnectionBlockedError({
79 reason: "authentication",
80 detail: "Sign in to your server again to reconnect.",
81 }),
82 }),
83});
84
85const layerConnectivity = Connectivity.layer({
86 status: Effect.sync(currentNetworkStatus),
87 changes: Stream.callback((queue) =>
88 Effect.acquireRelease(
89 Effect.sync(() => {
90 const online = () => Queue.offerUnsafe(queue, "online");
91 const offline = () => Queue.offerUnsafe(queue, "offline");
92 window.addEventListener("online", online);
93 window.addEventListener("offline", offline);
94 return { online, offline };
95 }),
96 ({ online, offline }) =>
97 Effect.sync(() => {
98 window.removeEventListener("online", online);
99 window.removeEventListener("offline", offline);
100 }),
101 ).pipe(Effect.asVoid),
102 ),
103});
104
105interface NetworkInformationLike extends EventTarget {
106 readonly type?: string;
107}
108
109/**
110 * Wakes connections when the browser reports a different network type, such
111 * as a laptop moving from Wi-Fi to a phone hotspot. `change` also fires for
112 * bandwidth and latency estimates on the same network, so only a type change
113 * counts. Browsers without `navigator.connection.type` rely on the periodic
114 * route check instead.
115 */
116const networkPathChanges = Stream.callback<"network-changed">((queue) =>
117 Effect.acquireRelease(
118 Effect.sync(() => {
119 const connection =
120 typeof navigator === "undefined"
121 ? undefined
122 : (navigator as Navigator & { readonly connection?: NetworkInformationLike }).connection;
123 if (connection?.type === undefined) return undefined;
124 let previous = connection.type;
125 const listener = () => {
126 const type = connection.type;
127 if (type === undefined || type === previous) return;
128 previous = type;
129 Queue.offerUnsafe(queue, "network-changed");
130 };
131 connection.addEventListener("change", listener);
132 return { connection, listener };
133 }),
134 (subscription) =>
135 Effect.sync(() =>
136 subscription?.connection.removeEventListener("change", subscription.listener),
137 ),
138 ).pipe(Effect.asVoid),
139);
140
141const layerWakeups = Wakeups.layer({
142 changes: Stream.mergeAll(
143 [
144 Stream.callback<"application-active">((queue) =>
145 Effect.acquireRelease(
146 Effect.sync(() => {
147 const listener = () => {
148 if (document.visibilityState === "visible") {
149 Queue.offerUnsafe(queue, "application-active");
150 }
151 };
152 document.addEventListener("visibilitychange", listener);
153 return listener;
154 }),
155 (listener) =>
156 Effect.sync(() => {
157 document.removeEventListener("visibilitychange", listener);
158 }),
159 ).pipe(Effect.asVoid),
160 ),
161 managedRelayAccountChanges(appAtomRegistry).pipe(
162 Stream.map(() => "credentials-changed" as const),
163 ),
164 networkPathChanges,
165 ],
166 { concurrency: "unbounded" },
167 ),
168});
169
170function clientMetadata() {
171 return clientPresentationMetadata({
172 appVersion: APP_VERSION,
173 identity: {
174 userAgent: navigator.userAgent,
175 platform: navigator.platform,
176 maxTouchPoints: navigator.maxTouchPoints,
177 },
178 desktopBridge: window.desktopBridge,
179 });
180}
181
182function sshPreparationError(cause: unknown) {
183 const message = cause instanceof Error ? cause.message : String(cause);
184 if (message.toLowerCase().includes("cancel")) {
185 return new ConnectionBlockedError({
186 reason: "authentication",
187 detail: message,
188 });
189 }
190 return new ConnectionTransientError({
191 reason: "remote-unavailable",
192 detail: `Could not prepare the SSH environment: ${message}`,
193 });
194}
195
196export const provisionDesktopSshEnvironment = Effect.fn(
197 "web.connectionPlatform.ssh.provisionDesktop",
198)(function* (
199 bridge: DesktopBridge,
200 target: DesktopSshEnvironmentTarget,
201 expectedEnvironmentId?: EnvironmentId,
202) {
203 const bootstrap = yield* Effect.tryPromise({
204 try: () =>
205 bridge.ensureSshEnvironment(target, {
206 issuePairingToken: true,
207 }),
208 catch: sshPreparationError,
209 });
210 const pairingToken = bootstrap.pairingToken;
211 if (pairingToken === null) {
212 return yield* new ConnectionBlockedError({
213 reason: "authentication",
214 detail: "The SSH environment did not issue a pairing credential.",
215 });
216 }
217 const descriptor = yield* Effect.tryPromise({
218 try: () => bridge.fetchSshEnvironmentDescriptor(bootstrap.httpBaseUrl),
219 catch: sshPreparationError,
220 });
221 if (expectedEnvironmentId !== undefined && descriptor.environmentId !== expectedEnvironmentId) {
222 return yield* new ConnectionBlockedError({
223 reason: "configuration",
224 detail: `That host reaches ${descriptor.label}, a different machine. Add it as its own environment instead.`,
225 });
226 }
227 const access = yield* Effect.tryPromise({
228 try: () => bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, pairingToken),
229 catch: sshPreparationError,
230 });
231 return {
232 environmentId: descriptor.environmentId,
233 label: descriptor.label,
234 bootstrap,
235 bearerToken: access.access_token,
236 };
237});
238
239const layerCapabilities = Layer.effectContext(
240 Effect.sync(() => {
241 const presentation = ClientCapabilities.ClientPresentation.of({
242 metadata: clientMetadata(),
243 });
244 const cloudSession = ClientCapabilities.CloudSession.of({
245 identity: Effect.sync(() =>
246 Option.fromNullishOr(appAtomRegistry.get(managedRelaySessionAtom)),
247 ),
248 clerkToken: Effect.gen(function* () {
249 const session = appAtomRegistry.get(managedRelaySessionAtom);
250 if (session === null) {
251 return yield* new ConnectionBlockedError({
252 reason: "authentication",
253 detail: "Sign in to T3 Connect to connect this environment.",
254 });
255 }
256 const token = yield* session.readClerkToken().pipe(
257 Effect.mapError(
258 (error) =>
259 new ConnectionTransientError({
260 reason: "network",
261 detail: error.message,
262 }),
263 ),
264 );
265 if (token === null) {
266 return yield* new ConnectionBlockedError({
267 reason: "authentication",
268 detail: "The T3 Connect session is unavailable.",
269 });
270 }
271 return token;
272 }),
273 });
274 const identity = ClientCapabilities.RelayDeviceIdentity.of({
275 deviceId: Effect.succeedNone,
276 });
277 const primaryAuth = ClientCapabilities.PrimaryEnvironmentAuth.of({
278 bearerToken: Effect.tryPromise({
279 try: readDesktopPrimaryBearerToken,
280 catch: (cause) =>
281 new ConnectionTransientError({
282 reason: "remote-unavailable",
283 detail: `Could not load the desktop primary credential: ${String(cause)}`,
284 }),
285 }).pipe(Effect.map(Option.fromNullishOr)),
286 });
287 const ssh = ClientCapabilities.SshEnvironmentGateway.of({
288 provision: Effect.fn("web.connectionPlatform.ssh.provision")(
289 function* (target, expectedEnvironmentId) {
290 const bridge = window.desktopBridge;
291 if (bridge === undefined) {
292 return yield* new ConnectionBlockedError({
293 reason: "unsupported",
294 detail: "SSH environments are only available in the desktop app.",
295 });
296 }
297 return yield* provisionDesktopSshEnvironment(bridge, target, expectedEnvironmentId);
298 },
299 ),
300 prepare: Effect.fn("web.connectionPlatform.ssh.prepare")(function* (input) {
301 const bridge = window.desktopBridge;
302 if (bridge === undefined) {
303 return yield* new ConnectionBlockedError({
304 reason: "unsupported",
305 detail: "SSH environments are only available in the desktop app.",
306 });
307 }
308 const bootstrap = yield* Effect.tryPromise({
309 try: () =>
310 bridge.ensureSshEnvironment(input.target, {
311 issuePairingToken: true,
312 }),
313 catch: sshPreparationError,
314 });
315 if (bootstrap.pairingToken === null) {
316 return yield* new ConnectionBlockedError({
317 reason: "authentication",
318 detail: "The SSH environment did not issue a pairing credential.",
319 });
320 }
321 const access = yield* Effect.tryPromise({
322 try: () =>
323 bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, bootstrap.pairingToken!),
324 catch: sshPreparationError,
325 });
326 return {
327 bootstrap,
328 bearerToken: access.access_token,
329 };
330 }),
331 disconnect: Effect.fn("web.connectionPlatform.ssh.disconnect")(function* (target) {
332 const bridge = window.desktopBridge;
333 if (bridge === undefined) {
334 return;
335 }
336 yield* Effect.tryPromise({
337 try: () => bridge.disconnectSshEnvironment(target),
338 catch: (cause) =>
339 new ConnectionTransientError({
340 reason: "remote-unavailable",
341 detail: `Could not disconnect the SSH environment: ${String(cause)}`,
342 }),
343 });
344 }),
345 });
346
347 return Context.make(ClientCapabilities.CloudSession, cloudSession).pipe(
348 Context.add(ClientCapabilities.PrimaryEnvironmentAuth, primaryAuth),
349 Context.add(ClientCapabilities.RelayDeviceIdentity, identity),
350 Context.add(ClientCapabilities.ClientPresentation, presentation),
351 Context.add(ClientCapabilities.SshEnvironmentGateway, ssh),
352 );
353 }),
354);
355
356const loadPrimaryConnectionRegistration = Effect.fn(
357 "web.connectionPlatform.loadPrimaryConnectionRegistration",
358)(function* (resolved: PrimaryEnvironmentTarget) {
359 const descriptor = yield* fetchRemoteEnvironmentDescriptor({
360 httpBaseUrl: resolved.target.httpBaseUrl,
361 }).pipe(
362 Effect.provide(PrimaryEnvironmentHttpLayer.layer),
363 Effect.mapError(mapRemoteEnvironmentError),
364 );
365 return new PrimaryConnectionRegistration({
366 target: new PrimaryConnectionTarget({
367 environmentId: descriptor.environmentId,
368 label: descriptor.label,
369 httpBaseUrl: resolved.target.httpBaseUrl,
370 wsBaseUrl: resolved.target.wsBaseUrl,
371 }),
372 });
373});
374
375// A desktop-local secondary backend (e.g. a parallel WSL backend) lives on its
376// own loopback origin, so — unlike the same-origin primary — it authenticates
377// with a bearer token minted from the bootstrap credential the desktop issues.
378const loadSecondaryConnectionRegistration = Effect.fn(
379 "web.connectionPlatform.loadSecondaryConnectionRegistration",
380)(function* (entry: DesktopEnvironmentBootstrap) {
381 if (
382 entry.httpBaseUrl === null ||
383 entry.wsBaseUrl === null ||
384 entry.bootstrapToken === undefined
385 ) {
386 return yield* new ConnectionTransientError({
387 reason: "endpoint-unavailable",
388 detail: `Desktop-local backend ${entry.id} is not ready yet.`,
389 });
390 }
391 const httpBaseUrl = entry.httpBaseUrl;
392 const wsBaseUrl = entry.wsBaseUrl;
393 const descriptor = yield* fetchRemoteEnvironmentDescriptor({ httpBaseUrl }).pipe(
394 Effect.mapError(mapRemoteEnvironmentError),
395 );
396 const issuedAtEpochMs = yield* Clock.currentTimeMillis;
397 // The desktop seed grant is administrative so the primary window can manage
398 // access; a secondary backend session only needs to operate its environment.
399 const access = yield* bootstrapRemoteBearerSession({
400 httpBaseUrl,
401 credential: entry.bootstrapToken,
402 scopes: AuthStandardClientScopes,
403 clientMetadata: clientMetadata(),
404 }).pipe(Effect.mapError(mapRemoteEnvironmentError));
405 // Keep the desktop pool's stable backend id in the connection id. The
406 // descriptor environment id still scopes projects and RPC state, while the
407 // backend id lets desktop-only operations (notably the WSL folder picker)
408 // route back to the instance that owns the environment.
409 const connectionId = desktopLocalConnectionId(entry.id);
410 // Prefer the desktop's bootstrap label (it identifies the backend and distro,
411 // e.g. "WSL: Ubuntu") over the generic descriptor label, so consumers can show
412 // a meaningful name without recovering it from the bootstrap list later.
413 const label = entry.label || descriptor.label;
414 return {
415 registration: new BearerConnectionRegistration({
416 target: new BearerConnectionTarget({
417 environmentId: descriptor.environmentId,
418 label,
419 connectionId,
420 }),
421 profile: new BearerConnectionProfile({
422 connectionId,
423 environmentId: descriptor.environmentId,
424 label,
425 httpBaseUrl,
426 wsBaseUrl,
427 }),
428 credential: new BearerConnectionCredential({ token: access.access_token }),
429 }),
430 expiresAtEpochMs: secondaryBearerExpiresAtEpochMs(issuedAtEpochMs, access.expires_in),
431 refreshAtEpochMs: secondaryBearerRefreshAtEpochMs(issuedAtEpochMs, access.expires_in),
432 };
433});
434
435// Poll cadence for the desktop bootstrap topology. There is no change event on
436// the bridge, so the renderer polls; successful registrations are cached by a
437// signature of their endpoint until bearer credentials approach expiry.
438const PLATFORM_POLL_INTERVAL = "3 seconds";
439const SECONDARY_BEARER_REFRESH_SKEW_MS = 5_000;
440
441export function secondaryBearerExpiresAtEpochMs(
442 issuedAtEpochMs: number,
443 expiresInSeconds: number,
444): number {
445 return issuedAtEpochMs + Math.max(0, expiresInSeconds * 1_000);
446}
447
448export function secondaryBearerRefreshAtEpochMs(
449 issuedAtEpochMs: number,
450 expiresInSeconds: number,
451): number {
452 return Math.max(
453 issuedAtEpochMs,
454 secondaryBearerExpiresAtEpochMs(issuedAtEpochMs, expiresInSeconds) -
455 SECONDARY_BEARER_REFRESH_SKEW_MS,
456 );
457}
458
459interface CachedPlatformRegistration {
460 readonly signature: string;
461 readonly registration: PlatformConnectionRegistration;
462 readonly expiresAtEpochMs?: number;
463 readonly refreshAtEpochMs?: number;
464 /** The bootstrap token a desktop-local bearer was exchanged for. */
465 readonly bootstrapToken?: string;
466}
467
468export type PrimaryEnvironmentTargetRead =
469 | {
470 readonly _tag: "Success";
471 readonly target: PrimaryEnvironmentTarget | null;
472 }
473 | {
474 readonly _tag: "Failure";
475 readonly cause: unknown;
476 };
477
478export function readPrimaryEnvironmentTargetResult(
479 readTarget: () => PrimaryEnvironmentTarget | null = readPrimaryEnvironmentTarget,
480): PrimaryEnvironmentTargetRead {
481 try {
482 return { _tag: "Success", target: readTarget() };
483 } catch (cause) {
484 return { _tag: "Failure", cause };
485 }
486}
487
488export function primaryRegistrationToRetainAfterTopologyRead(
489 previous: ReadonlyMap<string, CachedPlatformRegistration>,
490 topologyRead: PrimaryEnvironmentTargetRead,
491): CachedPlatformRegistration | undefined {
492 return topologyRead._tag === "Failure" ? previous.get(PRIMARY_LOCAL_ENVIRONMENT_ID) : undefined;
493}
494
495export function canReuseCachedPlatformRegistration(
496 cached: CachedPlatformRegistration,
497 signature: string,
498 nowEpochMs: number,
499): boolean {
500 return (
501 cached.signature === signature &&
502 (cached.refreshAtEpochMs === undefined || nowEpochMs < cached.refreshAtEpochMs)
503 );
504}
505
506export function canRetainCachedPlatformRegistrationAfterRefreshFailure(
507 cached: CachedPlatformRegistration,
508 signature: string,
509 nowEpochMs: number,
510): boolean {
511 return (
512 cached.signature === signature &&
513 cached.expiresAtEpochMs !== undefined &&
514 nowEpochMs < cached.expiresAtEpochMs
515 );
516}
517
518const REJECTED_BOOTSTRAP_RETRY_INITIAL_MS = 60_000;
519const REJECTED_BOOTSTRAP_RETRY_MAX_MS = 30 * 60_000;
520
521/** A bootstrap token a backend rejected, and when the poll may try it again. */
522export interface RejectedSecondaryBootstrap {
523 readonly signature: string;
524 readonly retryAtEpochMs: number;
525 readonly delayMs: number;
526}
527
528/**
529 * A backend that rejected a bootstrap token will usually keep rejecting it, so
530 * the poll backs off on that exact signature instead of re-presenting a dead
531 * credential every few seconds. It still retries on a capped backoff: a
532 * backend that restarts on the same port seeds a fresh grant for the same
533 * token, and nothing in the topology says it restarted. A new token or
534 * endpoint retries at once.
535 */
536export function isRejectedSecondaryBootstrap(
537 rejected: RejectedSecondaryBootstrap | undefined,
538 signature: string,
539 nowEpochMs: number,
540): rejected is RejectedSecondaryBootstrap {
541 return (
542 rejected !== undefined &&
543 rejected.signature === signature &&
544 nowEpochMs < rejected.retryAtEpochMs
545 );
546}
547
548export function nextRejectedSecondaryBootstrap(
549 previous: RejectedSecondaryBootstrap | undefined,
550 signature: string,
551 nowEpochMs: number,
552): RejectedSecondaryBootstrap {
553 const delayMs =
554 previous?.signature === signature
555 ? Math.min(previous.delayMs * 2, REJECTED_BOOTSTRAP_RETRY_MAX_MS)
556 : REJECTED_BOOTSTRAP_RETRY_INITIAL_MS;
557 return { signature, retryAtEpochMs: nowEpochMs + delayMs, delayMs };
558}
559
560export function isRejectedBootstrapCredentialError(error: ConnectionAttemptError): boolean {
561 return error._tag === "ConnectionBlockedError" && error.reason === "authentication";
562}
563
564export function secondaryRegistrationsToRetainAfterTopologyRead(
565 previous: ReadonlyMap<string, CachedPlatformRegistration>,
566 topologyRead: DesktopSecondaryBootstrapsRead,
567 nowEpochMs: number,
568): ReadonlyMap<string, CachedPlatformRegistration> {
569 if (topologyRead._tag === "Success") {
570 return new Map();
571 }
572 return new Map(
573 [...previous].filter(
574 ([, cached]) => cached.expiresAtEpochMs !== undefined && nowEpochMs < cached.expiresAtEpochMs,
575 ),
576 );
577}
578
579const layerPlatformConnectionSource = Layer.effect(
580 PlatformConnectionSource.PlatformConnectionSource,
581 Effect.gen(function* () {
582 if (isHostedStaticApp() || isLocalEnvironmentDisabled()) {
583 return PlatformConnectionSource.PlatformConnectionSource.of({
584 registrations: Stream.empty,
585 });
586 }
587 const cacheRef = yield* Ref.make(new Map<string, CachedPlatformRegistration>());
588 const rejectedRef = yield* Ref.make(new Map<string, RejectedSecondaryBootstrap>());
589
590 // Resolve the full set of platform-managed environments the host currently
591 // reports: the primary (same-origin cookie auth) plus any desktop-local
592 // backends running alongside it (bearer auth). Reused registrations come
593 // from the cache; a failed entry is skipped and retried on the next poll.
594 const buildPlatformRegistrations = Effect.gen(function* () {
595 const previous = yield* Ref.get(cacheRef);
596 const nowEpochMs = yield* Clock.currentTimeMillis;
597 const next = new Map<string, CachedPlatformRegistration>();
598 const registrations: Array<PlatformConnectionRegistration> = [];
599
600 const primaryTopologyRead = readPrimaryEnvironmentTargetResult();
601 const retainedPrimary = primaryRegistrationToRetainAfterTopologyRead(
602 previous,
603 primaryTopologyRead,
604 );
605 if (retainedPrimary !== undefined) {
606 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, retainedPrimary);
607 registrations.push(retainedPrimary.registration);
608 }
609
610 if (primaryTopologyRead._tag === "Failure") {
611 yield* Effect.logWarning("Could not read the primary environment topology.", {
612 cause: primaryTopologyRead.cause,
613 });
614 } else if (primaryTopologyRead.target !== null) {
615 const primaryTarget = primaryTopologyRead.target;
616 const signature = `primary|${primaryTarget.target.httpBaseUrl}|${primaryTarget.target.wsBaseUrl}`;
617 const cached = previous.get(PRIMARY_LOCAL_ENVIRONMENT_ID);
618 if (
619 cached !== undefined &&
620 canReuseCachedPlatformRegistration(cached, signature, nowEpochMs)
621 ) {
622 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, cached);
623 registrations.push(cached.registration);
624 } else {
625 const built = yield* loadPrimaryConnectionRegistration(primaryTarget).pipe(
626 Effect.tapError((error) =>
627 Effect.logWarning("Could not discover the primary environment.", { error }),
628 ),
629 Effect.option,
630 );
631 if (Option.isSome(built)) {
632 const cacheEntry = { signature, registration: built.value };
633 next.set(PRIMARY_LOCAL_ENVIRONMENT_ID, cacheEntry);
634 registrations.push(built.value);
635 }
636 }
637 }
638
639 const topologyRead = readDesktopSecondaryBootstrapsResult();
640 for (const [id, cached] of secondaryRegistrationsToRetainAfterTopologyRead(
641 previous,
642 topologyRead,
643 nowEpochMs,
644 )) {
645 next.set(id, cached);
646 registrations.push(cached.registration);
647 }
648
649 if (topologyRead._tag === "Failure") {
650 yield* Effect.logWarning("Could not read the desktop-local backend topology.", {
651 cause: topologyRead.cause,
652 });
653 } else {
654 const rejected = yield* Ref.get(rejectedRef);
655 const nextRejected = new Map<string, RejectedSecondaryBootstrap>();
656 for (const bootstrap of topologyRead.bootstraps) {
657 // The cached bearer belongs to the endpoint, not to the bootstrap
658 // token it was exchanged for: a new token must not drop a live
659 // session (its removal also clears the environment's drafts). The
660 // token only decides whether a rejected exchange is retried.
661 const endpointSignature = `${bootstrap.httpBaseUrl}|${bootstrap.wsBaseUrl}`;
662 const signature = `${endpointSignature}|${bootstrap.bootstrapToken ?? ""}`;
663 const cached = previous.get(bootstrap.id);
664 // A new token from the desktop means something changed (rotation, a
665 // restarted backend): exchange it rather than reusing a bearer that
666 // may be dead. The old bearer stays registered if that exchange fails.
667 if (
668 cached !== undefined &&
669 cached.bootstrapToken === bootstrap.bootstrapToken &&
670 canReuseCachedPlatformRegistration(cached, endpointSignature, nowEpochMs)
671 ) {
672 next.set(bootstrap.id, cached);
673 registrations.push(cached.registration);
674 continue;
675 }
676 const previouslyRejected = rejected.get(bootstrap.id);
677 if (isRejectedSecondaryBootstrap(previouslyRejected, signature, nowEpochMs)) {
678 nextRejected.set(bootstrap.id, previouslyRejected);
679 // The bearer minted before the token died is still good until it expires.
680 if (
681 cached !== undefined &&
682 canRetainCachedPlatformRegistrationAfterRefreshFailure(
683 cached,
684 endpointSignature,
685 nowEpochMs,
686 )
687 ) {
688 next.set(bootstrap.id, cached);
689 registrations.push(cached.registration);
690 }
691 continue;
692 }
693 const built = yield* loadSecondaryConnectionRegistration(bootstrap).pipe(
694 Effect.tapError((error) =>
695 Effect.logWarning("Could not connect a desktop-local backend.", {
696 id: bootstrap.id,
697 error,
698 }),
699 ),
700 Effect.tapError((error) =>
701 isRejectedBootstrapCredentialError(error)
702 ? // Back off from when the rejection arrived; the exchanges in
703 // this poll can take longer than the first backoff step.
704 Clock.currentTimeMillis.pipe(
705 Effect.map((rejectedAtEpochMs) =>
706 nextRejected.set(
707 bootstrap.id,
708 nextRejectedSecondaryBootstrap(
709 previouslyRejected,
710 signature,
711 rejectedAtEpochMs,
712 ),
713 ),
714 ),
715 )
716 : Effect.void,
717 ),
718 Effect.option,
719 );
720 if (Option.isSome(built)) {
721 const cacheEntry = {
722 signature: endpointSignature,
723 ...(bootstrap.bootstrapToken === undefined
724 ? {}
725 : { bootstrapToken: bootstrap.bootstrapToken }),
726 ...built.value,
727 };
728 next.set(bootstrap.id, cacheEntry);
729 registrations.push(built.value.registration);
730 } else if (
731 cached !== undefined &&
732 canRetainCachedPlatformRegistrationAfterRefreshFailure(
733 cached,
734 endpointSignature,
735 nowEpochMs,
736 )
737 ) {
738 next.set(bootstrap.id, cached);
739 registrations.push(cached.registration);
740 }
741 }
742 yield* Ref.set(rejectedRef, nextRejected);
743 }
744
745 yield* Ref.set(cacheRef, next);
746 return registrations as ReadonlyArray<PlatformConnectionRegistration>;
747 }).pipe(Effect.provide(FetchHttpClient.layer));
748
749 return PlatformConnectionSource.PlatformConnectionSource.of({
750 registrations: Stream.tick(PLATFORM_POLL_INTERVAL).pipe(
751 Stream.mapEffect(() => buildPlatformRegistrations),
752 ),
753 });
754 }),
755);
756
757const layerEnvironmentOwnedDataCleanup = Layer.succeed(
758 Persistence.EnvironmentOwnedDataCleanup,
759 Persistence.EnvironmentOwnedDataCleanup.of({
760 clear: (environmentId, options) =>
761 Effect.sync(() => {
762 if (!options?.retainDrafts) clearComposerDraftsEnvironment(environmentId);
763 }),
764 }),
765);
766
767const layerRpcRequestObserver = Layer.succeed(
768 EnvironmentRpcRequestObserver,
769 EnvironmentRpcRequestObserver.of({
770 observe: ({ environmentId, method }) =>
771 Effect.sync(() => {
772 nextObservedRpcRequestId += 1;
773 const requestId = `${environmentId}:${nextObservedRpcRequestId}`;
774 trackRpcRequestSent(requestId, method, `${method} · ${environmentId}`);
775 return Effect.sync(() => {
776 acknowledgeRpcRequest(requestId);
777 });
778 }),
779 }),
780);
781
782type ConnectionPlatformLayerSource =
783 | typeof ConnectionStorage.layer
784 | typeof layerConnectivity
785 | typeof layerHostedAuthAccess
786 | typeof layerWakeups
787 | typeof layerCapabilities
788 | typeof layerPlatformConnectionSource
789 | typeof layerEnvironmentOwnedDataCleanup
790 | typeof layerRpcRequestObserver;
791
792export const layer: Layer.Layer<
793 Layer.Success<ConnectionPlatformLayerSource>,
794 Layer.Error<ConnectionPlatformLayerSource>,
795 Layer.Services<ConnectionPlatformLayerSource>
796> = Layer.mergeAll(
797 ConnectionStorage.layer,
798 layerHostedAuthAccess,
799 layerConnectivity,
800 layerWakeups,
801 layerCapabilities,
802 layerPlatformConnectionSource,
803 layerEnvironmentOwnedDataCleanup,
804 layerRpcRequestObserver,
805);