Open History to see the last commit
1import { EnvironmentId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Equal from "effect/Equal";
5import * as Exit from "effect/Exit";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Ref from "effect/Ref";
9import * as Schema from "effect/Schema";
10import * as Scope from "effect/Scope";
11import * as Semaphore from "effect/Semaphore";
12import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";
14
15import * as ClientCapabilities from "../platform/capabilities.ts";
16import {
17 type ConnectionCatalogEntry,
18 type ConnectionCredential,
19 type ConnectionProfile,
20 type ConnectionRegistration,
21 type ConnectionRoute,
22 type PlatformConnectionRegistration,
23 SshConnectionProfile,
24 connectionRegistrationCatalogEntry,
25} from "./catalog.ts";
26import * as ConnectionCredentialStore from "./credentialStore.ts";
27import * as ConnectionProfileStore from "./profileStore.ts";
28import * as Connectivity from "./connectivity.ts";
29import type {
30 ConnectionAttemptError,
31 ConnectionTarget,
32 NetworkStatus,
33 PersistedConnectionTarget,
34 SupervisorConnectionState,
35} from "./model.ts";
36import { ConnectionBlockedError } from "./model.ts";
37import * as Persistence from "../platform/persistence.ts";
38import * as EnvironmentSupervisor from "./supervisor.ts";
39import * as ConnectionDriver from "./driver.ts";
40import * as ConnectionWakeups from "./wakeups.ts";
41import {
42 PullRequestRoutingPermissions,
43 pullRequestRoutingConnectionKey,
44} from "./pullRequestRoutingPermissions.ts";
45import {
46 RELAY_ROUTE_ID,
47 connectionRouteId,
48 connectionRoutes,
49 credentialConnectionId,
50 entryWithRoutes,
51 findRouteToSameAddress,
52 hasSavedBearerRoute,
53 mergeLearnedRoutes,
54 routesAfterRemoving,
55 upsertRoute,
56 type ReportedEndpoint,
57} from "./routes.ts";
58
59const isSshConnectionProfile = Schema.is(SshConnectionProfile);
60
61function unsupportedState(
62 entry: ConnectionCatalogEntry,
63): Pick<ConnectionCatalogEntry, "unsupportedReason" | "serverUpdateRequired"> {
64 return {
65 ...(entry.unsupportedReason === undefined
66 ? {}
67 : { unsupportedReason: entry.unsupportedReason }),
68 ...(entry.serverUpdateRequired === true ? { serverUpdateRequired: true } : {}),
69 };
70}
71
72export class EnvironmentNotRegisteredError extends Schema.TaggedError<EnvironmentNotRegisteredError>()(
73 "EnvironmentNotRegisteredError",
74 {
75 environmentId: EnvironmentId,
76 },
77) {
78 override get message(): string {
79 return `Environment ${this.environmentId} is not registered.`;
80 }
81}
82
83export class PlatformEnvironmentRemovalError extends Schema.TaggedError<PlatformEnvironmentRemovalError>()(
84 "PlatformEnvironmentRemovalError",
85 {
86 environmentId: EnvironmentId,
87 },
88) {
89 override get message(): string {
90 return `Platform-managed environment ${this.environmentId} cannot be removed.`;
91 }
92}
93
94export class EnvironmentRegistry extends Context.Service<
95 EnvironmentRegistry,
96 {
97 readonly entries: SubscriptionRef.SubscriptionRef<
98 ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>
99 >;
100 readonly networkStatus: SubscriptionRef.SubscriptionRef<NetworkStatus>;
101 readonly start: Effect.Effect<void>;
102 /**
103 * Succeeds with `false` and changes nothing when the environment is
104 * platform-managed, since the host owns that entry. With `knownAddressOnly`, it fails like
105 * `assertKnownAddress`, checked under the environment's lease.
106 */
107 readonly register: (
108 registration: ConnectionRegistration,
109 options?: { readonly knownAddressOnly?: boolean },
110 ) => Effect.Effect<boolean, Persistence.ConnectionPersistenceError | ConnectionBlockedError>;
111 /**
112 * Fails when the route would add a new address to a saved user environment. Pairing that does
113 * not target an environment checks this, because the server reports its own environment id.
114 */
115 readonly assertKnownAddress: (
116 route: ConnectionRoute,
117 ) => Effect.Effect<void, ConnectionBlockedError>;
118 readonly reconcilePlatform: (
119 registrations: ReadonlyArray<PlatformConnectionRegistration>,
120 ) => Effect.Effect<void>;
121 readonly remove: (
122 environmentId: EnvironmentId,
123 ) => Effect.Effect<
124 void,
125 | Persistence.ConnectionPersistenceError
126 | ConnectionAttemptError
127 | EnvironmentNotRegisteredError
128 | PlatformEnvironmentRemovalError
129 >;
130 /**
131 * Removing an environment's last route removes the environment, the same as
132 * `remove`.
133 */
134 readonly removeRoute: (
135 environmentId: EnvironmentId,
136 routeId: string,
137 ) => Effect.Effect<
138 void,
139 | Persistence.ConnectionPersistenceError
140 | ConnectionAttemptError
141 | EnvironmentNotRegisteredError
142 | PlatformEnvironmentRemovalError
143 >;
144 readonly reorderRoutes: (
145 environmentId: EnvironmentId,
146 routeIds: ReadonlyArray<string>,
147 ) => Effect.Effect<
148 void,
149 | Persistence.ConnectionPersistenceError
150 | EnvironmentNotRegisteredError
151 | PlatformEnvironmentRemovalError
152 | ConnectionBlockedError
153 >;
154 readonly removeRelayEnvironments: () => Effect.Effect<
155 void,
156 | Persistence.ConnectionPersistenceError
157 | ConnectionAttemptError
158 | PlatformEnvironmentRemovalError
159 >;
160 readonly retryNow: (environmentId: EnvironmentId) => Effect.Effect<void>;
161 readonly setEnabled: (
162 environmentId: EnvironmentId,
163 enabled: boolean,
164 ) => Effect.Effect<
165 void,
166 | EnvironmentNotRegisteredError
167 | Persistence.ConnectionPersistenceError
168 | ConnectionBlockedError
169 >;
170 readonly setCompatibility: (
171 environmentId: EnvironmentId,
172 error: ConnectionBlockedError | null,
173 ) => Effect.Effect<void, Persistence.ConnectionPersistenceError>;
174 readonly state: (
175 environmentId: EnvironmentId,
176 ) => Effect.Effect<SupervisorConnectionState, EnvironmentNotRegisteredError>;
177 readonly stateChanges: (
178 environmentId: EnvironmentId,
179 ) => Stream.Stream<SupervisorConnectionState, EnvironmentNotRegisteredError>;
180 readonly run: <A, E, R>(
181 environmentId: EnvironmentId,
182 effect: Effect.Effect<A, E, R>,
183 ) => Effect.Effect<
184 A,
185 E | EnvironmentNotRegisteredError,
186 Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
187 >;
188 readonly runStream: <A, E, R>(
189 environmentId: EnvironmentId,
190 stream: Stream.Stream<A, E, R>,
191 ) => Stream.Stream<
192 A,
193 E | EnvironmentNotRegisteredError,
194 Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>
195 >;
196 readonly followStream: <A, E, R>(
197 environmentId: EnvironmentId,
198 stream: Stream.Stream<A, E, R>,
199 ) => Stream.Stream<A, E, Exclude<R, EnvironmentSupervisor.EnvironmentSupervisor>>;
200 }
201>()("@t3tools/client-runtime/connection/registry/EnvironmentRegistry") {}
202
203interface EnvironmentServiceScope {
204 readonly entry: ConnectionCatalogEntry;
205 readonly supervisor: EnvironmentSupervisor.EnvironmentSupervisor["Service"];
206 readonly scope: Scope.Closeable;
207}
208
209/** @public */
210export const make = Effect.gen(function* () {
211 const registryScope = yield* Scope.Scope;
212 const storage = yield* Persistence.ConnectionTargetStore;
213 const registrations = yield* Persistence.ConnectionRegistrationStore;
214 const cache = yield* Persistence.EnvironmentCacheStore;
215 const ownedDataCleanup = yield* Persistence.EnvironmentOwnedDataCleanup;
216 const profiles = yield* ConnectionProfileStore.ConnectionProfileStore;
217 const credentials = yield* ConnectionCredentialStore.ConnectionCredentialStore;
218 const pullRequestRoutingPermissions = yield* PullRequestRoutingPermissions;
219 const connectivity = yield* Connectivity.Connectivity;
220 const driver = yield* ConnectionDriver.ConnectionDriver;
221 const wakeups = yield* ConnectionWakeups.ConnectionWakeups;
222 const ssh = yield* ClientCapabilities.SshEnvironmentGateway;
223 const persistedTargets = yield* storage.list;
224 const disabledEnvironmentIds = new Set(yield* storage.listDisabled);
225 const loadRoute = Effect.fn("EnvironmentRegistry.loadRoute")(function* (
226 target: ConnectionTarget,
227 ) {
228 const profile: Option.Option<ConnectionProfile> =
229 target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget"
230 ? yield* profiles.get(target.connectionId)
231 : Option.none();
232 return { target, profile } satisfies ConnectionRoute;
233 });
234 const persistedRoutesByEnvironment = new Map<EnvironmentId, Array<ConnectionTarget>>();
235 for (const target of persistedTargets) {
236 const routes = persistedRoutesByEnvironment.get(target.environmentId) ?? [];
237 routes.push(target);
238 persistedRoutesByEnvironment.set(target.environmentId, routes);
239 }
240 const initialEntries = new Map(
241 yield* Effect.forEach(
242 persistedRoutesByEnvironment,
243 Effect.fn("EnvironmentRegistry.loadCatalogEntry")(function* ([environmentId, targets]) {
244 const loaded = yield* Effect.forEach(targets, loadRoute, { concurrency: "unbounded" });
245 const seen = new Set<string>();
246 const usable = loaded.filter((route) => {
247 const id = connectionRouteId(route.target);
248 if (seen.has(id)) return false;
249 seen.add(id);
250 return !(id.startsWith("learned:") && Option.isNone(route.profile));
251 });
252 const routes = usable.length > 0 ? usable : loaded.slice(0, 1);
253 const first = routes[0]!;
254 return [
255 environmentId,
256 entryWithRoutes(
257 {
258 target: first.target,
259 profile: first.profile,
260 enabled: !disabledEnvironmentIds.has(environmentId),
261 },
262 routes,
263 ),
264 ] as const;
265 }),
266 { concurrency: "unbounded" },
267 ),
268 );
269 const entries =
270 yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>>(initialEntries);
271 const networkStatus = yield* SubscriptionRef.make(yield* connectivity.status);
272 const serviceScopes = yield* SubscriptionRef.make<
273 ReadonlyMap<EnvironmentId, EnvironmentServiceScope>
274 >(new Map());
275 const platformEnvironmentIds = yield* Ref.make<ReadonlySet<EnvironmentId>>(new Set());
276 const persistedEnvironmentIds = yield* Ref.make<ReadonlySet<EnvironmentId>>(
277 new Set(persistedRoutesByEnvironment.keys()),
278 );
279 interface LeaseLock {
280 readonly semaphore: Semaphore.Semaphore;
281 readonly users: number;
282 }
283
284 const leaseLocks = yield* Ref.make<ReadonlyMap<EnvironmentId, LeaseLock>>(new Map());
285 const leaseLocksGuard = yield* Semaphore.make(1);
286 const started = yield* Ref.make(false);
287
288 const withLeaseLock = <A, E, R>(
289 environmentId: EnvironmentId,
290 effect: Effect.Effect<A, E, R>,
291 ): Effect.Effect<A, E, R> =>
292 Effect.acquireUseRelease(
293 leaseLocksGuard.withPermits(1)(
294 Effect.gen(function* () {
295 const current = yield* Ref.get(leaseLocks);
296 const existing = current.get(environmentId);
297 if (existing !== undefined) {
298 yield* Ref.set(
299 leaseLocks,
300 new Map(current).set(environmentId, {
301 semaphore: existing.semaphore,
302 users: existing.users + 1,
303 }),
304 );
305 return existing.semaphore;
306 }
307 const semaphore = yield* Semaphore.make(1);
308 yield* Ref.set(leaseLocks, new Map(current).set(environmentId, { semaphore, users: 1 }));
309 return semaphore;
310 }),
311 ),
312 (semaphore) => semaphore.withPermits(1)(effect),
313 (semaphore) =>
314 leaseLocksGuard.withPermits(1)(
315 Ref.update(leaseLocks, (current) => {
316 const existing = current.get(environmentId);
317 if (existing === undefined || existing.semaphore !== semaphore) {
318 return current;
319 }
320 const next = new Map(current);
321 if (existing.users === 1) {
322 next.delete(environmentId);
323 } else {
324 next.set(environmentId, {
325 semaphore,
326 users: existing.users - 1,
327 });
328 }
329 return next;
330 }),
331 ),
332 ).pipe(Effect.withSpan("EnvironmentRegistry.withLeaseLock"));
333
334 const getEntry = Effect.fn("EnvironmentRegistry.getEntry")(function* (
335 environmentId: EnvironmentId,
336 ) {
337 const entry = (yield* SubscriptionRef.get(entries)).get(environmentId);
338 if (entry === undefined) {
339 return yield* new EnvironmentNotRegisteredError({
340 environmentId,
341 });
342 }
343 return entry;
344 });
345
346 const closeServiceScope = Effect.fn("EnvironmentRegistry.closeServiceScope")(function* (
347 environmentId: EnvironmentId,
348 ) {
349 const current = yield* SubscriptionRef.get(serviceScopes);
350 const lease = current.get(environmentId);
351 if (lease === undefined) {
352 return;
353 }
354 const next = new Map(current);
355 next.delete(environmentId);
356 yield* SubscriptionRef.set(serviceScopes, next);
357 yield* Scope.close(lease.scope, Exit.void);
358 });
359
360 const createServiceScope = Effect.fn("EnvironmentRegistry.createServiceScope")(
361 (entry: ConnectionCatalogEntry) =>
362 Effect.uninterruptible(
363 Effect.gen(function* () {
364 const environmentId = entry.target.environmentId;
365 const scope = yield* Scope.fork(registryScope);
366 const supervisor = yield* EnvironmentSupervisor.make(entry, {
367 // Seed the intent to avoid publishing a false switched-off state
368 // before an enabled, compatible environment starts connecting.
369 initiallyDesired: entry.enabled && entry.unsupportedReason === undefined,
370 learnRoutes: (input) => learnRoutes({ environmentId, ...input }),
371 }).pipe(
372 Effect.provideService(Connectivity.Connectivity, connectivity),
373 Effect.provideService(ConnectionDriver.ConnectionDriver, driver),
374 Effect.provideService(ConnectionWakeups.ConnectionWakeups, wakeups),
375 Scope.provide(scope),
376 Effect.onError(() => Scope.close(scope, Exit.void)),
377 );
378 yield* SubscriptionRef.update(serviceScopes, (current) => {
379 const next = new Map(current);
380 next.set(environmentId, { entry, supervisor, scope });
381 return next;
382 });
383 yield* SubscriptionRef.changes(supervisor.state).pipe(
384 Stream.runForEach((state) =>
385 state.phase === "blocked" && state.lastFailure?.reason === "unsupported"
386 ? setCompatibility(environmentId, state.lastFailure)
387 : Effect.void,
388 ),
389 Effect.forkIn(scope),
390 );
391 return supervisor;
392 }),
393 ),
394 );
395
396 const acquireSupervisor = Effect.fn("EnvironmentRegistry.acquireSupervisor")(function* (
397 environmentId: EnvironmentId,
398 ) {
399 return yield* withLeaseLock(
400 environmentId,
401 Effect.gen(function* () {
402 const entry = yield* getEntry(environmentId);
403 const existing = (yield* SubscriptionRef.get(serviceScopes)).get(environmentId);
404 if (existing !== undefined) {
405 if (Equal.equals(existing.entry, entry)) {
406 return existing.supervisor;
407 }
408 yield* closeServiceScope(environmentId);
409 }
410 return yield* createServiceScope(entry);
411 }),
412 );
413 });
414
415 const run: EnvironmentRegistry["Service"]["run"] = Effect.fn("EnvironmentRegistry.run")(
416 function* <A, E, R>(environmentId: EnvironmentId, effect: Effect.Effect<A, E, R>) {
417 const supervisor = yield* acquireSupervisor(environmentId);
418 return yield* Effect.provideService(
419 effect,
420 EnvironmentSupervisor.EnvironmentSupervisor,
421 supervisor,
422 );
423 },
424 );
425
426 const runStream: EnvironmentRegistry["Service"]["runStream"] = <A, E, R>(
427 environmentId: EnvironmentId,
428 stream: Stream.Stream<A, E, R>,
429 ) =>
430 Stream.unwrap(
431 acquireSupervisor(environmentId).pipe(
432 Effect.map((supervisor) =>
433 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
434 ),
435 ),
436 );
437
438 const followStream: EnvironmentRegistry["Service"]["followStream"] = <A, E, R>(
439 environmentId: EnvironmentId,
440 stream: Stream.Stream<A, E, R>,
441 ) =>
442 Stream.concat(
443 Stream.fromEffect(SubscriptionRef.get(entries)),
444 SubscriptionRef.changes(entries),
445 ).pipe(
446 Stream.map((current) => Option.fromUndefinedOr(current.get(environmentId))),
447 Stream.changesWith(
448 (previous, current) => Option.getOrNull(previous) === Option.getOrNull(current),
449 ),
450 Stream.switchMap(
451 Option.match({
452 onNone: () => Stream.empty,
453 onSome: () =>
454 Stream.unwrap(
455 acquireSupervisor(environmentId).pipe(
456 Effect.match({
457 onFailure: () => Stream.empty,
458 onSuccess: (supervisor) =>
459 Stream.provideService(
460 stream,
461 EnvironmentSupervisor.EnvironmentSupervisor,
462 supervisor,
463 ),
464 }),
465 ),
466 ),
467 }),
468 ),
469 );
470
471 const start = Effect.gen(function* () {
472 if (yield* Ref.getAndSet(started, true)) {
473 return;
474 }
475 yield* Effect.forEach(
476 persistedRoutesByEnvironment.keys(),
477 (environmentId) =>
478 acquireSupervisor(environmentId).pipe(
479 Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }),
480 ),
481 {
482 concurrency: "unbounded",
483 discard: true,
484 },
485 );
486 }).pipe(Effect.withSpan("EnvironmentRegistry.start"));
487
488 const installEntryLocked = Effect.fn("EnvironmentRegistry.installEntryLocked")(function* (
489 entry: ConnectionCatalogEntry,
490 options?: { readonly retainEquivalentRuntime?: boolean },
491 ) {
492 const target = entry.target;
493 const previous = (yield* SubscriptionRef.get(entries)).get(target.environmentId);
494 const existingScope = (yield* SubscriptionRef.get(serviceScopes)).get(target.environmentId);
495 if (
496 options?.retainEquivalentRuntime === true &&
497 previous !== undefined &&
498 Equal.equals(previous, entry) &&
499 existingScope !== undefined &&
500 Equal.equals(existingScope.entry, entry)
501 ) {
502 return;
503 }
504
505 yield* closeServiceScope(target.environmentId);
506 yield* SubscriptionRef.update(entries, (current) => {
507 const next = new Map(current);
508 next.set(target.environmentId, entry);
509 return next;
510 });
511 yield* createServiceScope(entry);
512 });
513
514 const forgetRoutingTrust = (
515 environmentId: EnvironmentId,
516 operation: "register-connection" | "set-connection-routes",
517 ) =>
518 pullRequestRoutingPermissions.forget(environmentId).pipe(
519 Effect.mapError(
520 (error) =>
521 new Persistence.ConnectionPersistenceError({
522 operation,
523 message: error.message,
524 }),
525 ),
526 );
527
528 const persistedRoutes = (entry: ConnectionCatalogEntry) =>
529 connectionRoutes(entry).map((route) => route.target as PersistedConnectionTarget);
530
531 /**
532 * Pull request routing trust and an unsupported verdict both belong to the saved addresses: a changed set may
533 * reach a different server, so both are dropped and the next connection decides.
534 */
535 const withRoutes = (previous: ConnectionCatalogEntry, routes: ReadonlyArray<ConnectionRoute>) => {
536 const next = entryWithRoutes(previous, routes);
537 if (pullRequestRoutingConnectionKey(previous) === pullRequestRoutingConnectionKey(next)) {
538 return { entry: next, addressesChanged: false };
539 }
540 const { unsupportedReason: _reason, serverUpdateRequired: _update, ...rest } = next;
541 return { entry: rest, addressesChanged: true };
542 };
543
544 const assertKnownAddress = Effect.fn("EnvironmentRegistry.assertKnownAddress")(function* (
545 route: ConnectionRoute,
546 ) {
547 const environmentId = route.target.environmentId;
548 if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) return;
549 const saved = (yield* SubscriptionRef.get(entries)).get(environmentId);
550 if (
551 saved !== undefined &&
552 findRouteToSameAddress(connectionRoutes(saved), route) === undefined
553 ) {
554 return yield* new ConnectionBlockedError({
555 reason: "configuration",
556 detail: `That address claims to be ${saved.target.label}, which is already saved. Add it as a route from that environment's settings instead.`,
557 });
558 }
559 });
560
561 const register = Effect.fn("EnvironmentRegistry.register")(function* (
562 registration: ConnectionRegistration,
563 options?: { readonly knownAddressOnly?: boolean },
564 ) {
565 const registered = connectionRegistrationCatalogEntry(registration);
566 const environmentId = registered.target.environmentId;
567 const route: ConnectionRoute = { target: registered.target, profile: registered.profile };
568 return yield* withLeaseLock(
569 environmentId,
570 Effect.gen(function* () {
571 if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) {
572 return false;
573 }
574 if (options?.knownAddressOnly === true) {
575 yield* assertKnownAddress(route);
576 }
577 const previous = (yield* SubscriptionRef.get(entries)).get(environmentId);
578 let entry = registered;
579 if (previous !== undefined) {
580 const existing = connectionRoutes(previous);
581 const sameAddress = findRouteToSameAddress(existing, route);
582 const routes = existing.filter(
583 (existing) =>
584 existing !== sameAddress ||
585 connectionRouteId(existing.target) === connectionRouteId(route.target),
586 );
587 const next = withRoutes(previous, upsertRoute(routes, route));
588 if (next.addressesChanged) {
589 yield* forgetRoutingTrust(environmentId, "register-connection");
590 }
591 entry = next.entry;
592 }
593 yield* registrations.register(registration, persistedRoutes(entry));
594 yield* Ref.update(persistedEnvironmentIds, (current) =>
595 new Set(current).add(environmentId),
596 );
597 yield* installEntryLocked(entry);
598 return true;
599 }),
600 );
601 });
602
603 const replaceRoutesLocked = Effect.fn("EnvironmentRegistry.replaceRoutesLocked")(function* (
604 previous: ConnectionCatalogEntry,
605 routes: ReadonlyArray<ConnectionRoute>,
606 ) {
607 const environmentId = previous.target.environmentId;
608 const next = withRoutes(previous, routes);
609 if (next.addressesChanged) {
610 yield* forgetRoutingTrust(environmentId, "set-connection-routes");
611 }
612 yield* registrations.setRoutes(environmentId, persistedRoutes(next.entry));
613 yield* installEntryLocked(next.entry);
614 });
615
616 const userEntry = Effect.fn("EnvironmentRegistry.userEntry")(function* (
617 environmentId: EnvironmentId,
618 ) {
619 if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) {
620 return yield* new PlatformEnvironmentRemovalError({ environmentId });
621 }
622 return yield* getEntry(environmentId);
623 });
624
625 const installPlatformRegistration = Effect.fn("EnvironmentRegistry.installPlatformRegistration")(
626 function* (registration: PlatformConnectionRegistration) {
627 const registered = connectionRegistrationCatalogEntry(registration);
628 const target = registered.target;
629 return yield* withLeaseLock(
630 target.environmentId,
631 Effect.gen(function* () {
632 let applied = true;
633 const previous = (yield* SubscriptionRef.get(entries)).get(target.environmentId);
634 const entry: ConnectionCatalogEntry =
635 previous?.unsupportedReason !== undefined &&
636 pullRequestRoutingConnectionKey(previous) ===
637 pullRequestRoutingConnectionKey(registered)
638 ? { ...registered, ...unsupportedState(previous) }
639 : registered;
640 const persisted = (yield* Ref.get(persistedEnvironmentIds)).has(target.environmentId);
641 if (
642 persisted ||
643 (previous !== undefined &&
644 pullRequestRoutingConnectionKey(previous) !== pullRequestRoutingConnectionKey(entry))
645 ) {
646 const revoked = yield* pullRequestRoutingPermissions.forget(target.environmentId).pipe(
647 Effect.tapError((error) =>
648 Effect.logWarning(
649 "Could not clear pull request routing permission for a platform environment.",
650 {
651 environmentId: target.environmentId,
652 error,
653 },
654 ),
655 ),
656 Effect.exit,
657 );
658 if (Exit.isFailure(revoked)) return false;
659 }
660 yield* Ref.update(platformEnvironmentIds, (current) => {
661 const next = new Set(current);
662 next.add(target.environmentId);
663 return next;
664 });
665
666 // Secondary desktop-local backends (e.g. a parallel WSL backend) live
667 // on their own loopback origin, so they authenticate with a bearer
668 // token instead of the primary's same-origin cookie. Stash it where
669 // the resolver's bearer broker looks it up.
670 let bearerChanged = false;
671 if (registration._tag === "BearerConnectionRegistration") {
672 const stored = yield* credentials
673 .get(registration.target.connectionId)
674 .pipe(Effect.orElseSucceed(() => Option.none<ConnectionCredential>()));
675 bearerChanged =
676 Option.isSome(stored) && !Equal.equals(stored.value, registration.credential);
677 }
678
679 if (persisted) {
680 const removed = yield* registrations.remove(target.environmentId).pipe(
681 Effect.tap(() =>
682 Ref.update(persistedEnvironmentIds, (current) => {
683 const next = new Set(current);
684 next.delete(target.environmentId);
685 return next;
686 }),
687 ),
688 Effect.as(true),
689 Effect.catch((error) =>
690 Effect.logWarning(
691 "Could not remove a persisted registration shadowed by a platform environment.",
692 {
693 environmentId: target.environmentId,
694 error,
695 },
696 ).pipe(Effect.as(false)),
697 ),
698 );
699 applied &&= removed;
700 }
701
702 let bearerReplaced = false;
703 if (registration._tag === "BearerConnectionRegistration") {
704 const written = yield* credentials
705 .put(registration.target.connectionId, registration.credential)
706 .pipe(
707 Effect.as(true),
708 Effect.catch((error) =>
709 Effect.logWarning("Could not store the platform bearer credential.", {
710 environmentId: target.environmentId,
711 error,
712 }).pipe(Effect.as(false)),
713 ),
714 );
715 // Only a bearer that was actually stored can be retried with; a
716 // failed write leaves the rejected one in place.
717 bearerReplaced = written && bearerChanged;
718 applied &&= written;
719 }
720
721 yield* installEntryLocked(entry, { retainEquivalentRuntime: true });
722
723 if (bearerReplaced && registration._tag === "BearerConnectionRegistration") {
724 const scope = (yield* SubscriptionRef.get(serviceScopes)).get(target.environmentId);
725 if (scope !== undefined) {
726 const state = yield* SubscriptionRef.get(scope.supervisor.state);
727 if (
728 state.phase === "connecting" ||
729 state.phase === "backoff" ||
730 (state.phase === "blocked" &&
731 state.lastFailure?._tag === "ConnectionBlockedError" &&
732 state.lastFailure.reason === "authentication")
733 ) {
734 yield* scope.supervisor.retryNow;
735 } else if (state.phase === "connected") {
736 const token = registration.credential.token;
737 yield* SubscriptionRef.updateSome(scope.supervisor.prepared, (current) =>
738 Option.isSome(current) &&
739 current.value.target._tag === "BearerConnectionTarget" &&
740 credentialConnectionId(current.value.target.connectionId) ===
741 registration.target.connectionId &&
742 current.value.httpAuthorization?._tag === "Bearer" &&
743 current.value.httpAuthorization.token !== token
744 ? Option.some(
745 Option.some({
746 ...current.value,
747 httpAuthorization: { _tag: "Bearer", token },
748 }),
749 )
750 : Option.none(),
751 );
752 }
753 }
754 }
755 return applied;
756 }),
757 );
758 },
759 );
760
761 const removePlatformEnvironment = Effect.fn("EnvironmentRegistry.removePlatformEnvironment")(
762 function* (environmentId: EnvironmentId) {
763 return yield* withLeaseLock(
764 environmentId,
765 Effect.gen(function* () {
766 const entry = (yield* SubscriptionRef.get(entries)).get(environmentId);
767 const revoked = yield* pullRequestRoutingPermissions.forget(environmentId).pipe(
768 Effect.tapError((error) =>
769 Effect.logWarning(
770 "Could not clear pull request routing permission after platform removal.",
771 {
772 environmentId,
773 error,
774 },
775 ),
776 ),
777 Effect.exit,
778 );
779 if (Exit.isFailure(revoked)) return false;
780 yield* Ref.update(platformEnvironmentIds, (current) => {
781 const next = new Set(current);
782 next.delete(environmentId);
783 return next;
784 });
785 yield* closeServiceScope(environmentId);
786 yield* SubscriptionRef.update(entries, (current) => {
787 const next = new Map(current);
788 next.delete(environmentId);
789 return next;
790 });
791 if (entry !== undefined && entry.target._tag === "BearerConnectionTarget") {
792 yield* credentials.remove(entry.target.connectionId).pipe(
793 Effect.catch((error) =>
794 Effect.logWarning("Could not clear the platform bearer credential.", {
795 environmentId,
796 error,
797 }),
798 ),
799 );
800 }
801 yield* Effect.all(
802 [
803 cache.clear(environmentId).pipe(
804 Effect.catch((error) =>
805 Effect.logWarning("Could not clear cached environment data after removal.", {
806 environmentId,
807 error,
808 }),
809 ),
810 ),
811 ownedDataCleanup.clear(environmentId),
812 ],
813 { concurrency: "unbounded", discard: true },
814 );
815 return true;
816 }),
817 );
818 },
819 );
820
821 /** Succeeds with `false` when a best-effort write failed, so an identical repeat retries it. */
822 const applyPlatformRegistrations = Effect.fn("EnvironmentRegistry.reconcilePlatform")(function* (
823 platformRegistrations: ReadonlyArray<PlatformConnectionRegistration>,
824 ) {
825 const desiredIds = new Set(
826 platformRegistrations.map((registration) => registration.target.environmentId),
827 );
828 const currentPlatformIds = yield* Ref.get(platformEnvironmentIds);
829 const removed = yield* Effect.forEach(currentPlatformIds, (environmentId) =>
830 desiredIds.has(environmentId)
831 ? Effect.succeed(true)
832 : removePlatformEnvironment(environmentId),
833 );
834 const installed = yield* Effect.forEach(platformRegistrations, installPlatformRegistration);
835 return [...removed, ...installed].every(Boolean);
836 });
837
838 const reconcileLock = yield* Semaphore.make(1);
839 let appliedPlatformRegistrations: ReadonlyArray<PlatformConnectionRegistration> | undefined;
840 // Untraced: platforms repeat unchanged registrations every few seconds, and a
841 // span per poll would make idle clients export traces continuously.
842 const reconcilePlatform = Effect.fnUntraced(function* (
843 platformRegistrations: ReadonlyArray<PlatformConnectionRegistration>,
844 ) {
845 if (Equal.equals(appliedPlatformRegistrations, platformRegistrations)) return;
846 appliedPlatformRegistrations = undefined;
847 if (yield* applyPlatformRegistrations(platformRegistrations)) {
848 appliedPlatformRegistrations = platformRegistrations;
849 }
850 }, reconcileLock.withPermits(1));
851
852 /** Callers hold its lease lock. */
853 const removeLocked = Effect.fn("EnvironmentRegistry.removeLocked")(function* (
854 environmentId: EnvironmentId,
855 ) {
856 if ((yield* Ref.get(platformEnvironmentIds)).has(environmentId)) {
857 return yield* new PlatformEnvironmentRemovalError({
858 environmentId,
859 });
860 }
861 const entry = yield* getEntry(environmentId);
862
863 yield* pullRequestRoutingPermissions.forget(environmentId);
864 yield* registrations.remove(environmentId);
865 yield* Ref.update(persistedEnvironmentIds, (current) => {
866 const next = new Set(current);
867 next.delete(environmentId);
868 return next;
869 });
870 yield* closeServiceScope(environmentId);
871 yield* SubscriptionRef.update(entries, (current) => {
872 const next = new Map(current);
873 next.delete(environmentId);
874 return next;
875 });
876 yield* Effect.all(
877 [
878 cache.clear(environmentId).pipe(
879 Effect.catch((error) =>
880 Effect.logWarning("Could not clear cached environment data after removal.", {
881 environmentId,
882 error,
883 }),
884 ),
885 ),
886 ownedDataCleanup.clear(environmentId),
887 ],
888 { concurrency: "unbounded", discard: true },
889 );
890
891 for (const route of connectionRoutes(entry)) {
892 const profile = Option.getOrNull(route.profile);
893 if (profile !== null && isSshConnectionProfile(profile)) {
894 yield* disconnectSsh(environmentId, profile);
895 }
896 }
897 });
898
899 const remove = Effect.fn("EnvironmentRegistry.remove")(function* (environmentId: EnvironmentId) {
900 return yield* withLeaseLock(environmentId, removeLocked(environmentId));
901 });
902
903 const disconnectSsh = (environmentId: EnvironmentId, profile: SshConnectionProfile) =>
904 ssh.disconnect(profile.target).pipe(
905 Effect.tapError((error) =>
906 Effect.logWarning("Could not disconnect the managed SSH environment.", {
907 environmentId,
908 error,
909 }),
910 ),
911 Effect.ignore,
912 );
913
914 // Plain HTTP routes are unusable from an HTTPS page, which blocks mixed content.
915 const allowInsecureRoutes =
916 typeof globalThis.location === "undefined" || globalThis.location.protocol !== "https:";
917
918 /**
919 * Learned routes reuse the active route's credential, so they do not revoke pull request routing
920 * trust the way a newly paired address does.
921 */
922 const learnRoutes = Effect.fn("EnvironmentRegistry.learnRoutes")(function* (input: {
923 readonly environmentId: EnvironmentId;
924 readonly activeRoute: ConnectionRoute;
925 readonly reported: ReadonlyArray<ReportedEndpoint>;
926 }) {
927 return yield* withLeaseLock(
928 input.environmentId,
929 Effect.gen(function* () {
930 if ((yield* Ref.get(platformEnvironmentIds)).has(input.environmentId)) {
931 return Option.none<ConnectionCatalogEntry>();
932 }
933 const entry = (yield* SubscriptionRef.get(entries)).get(input.environmentId);
934 if (entry === undefined) return Option.none<ConnectionCatalogEntry>();
935 const routes = mergeLearnedRoutes({
936 entry,
937 activeRoute: input.activeRoute,
938 reported: input.reported,
939 allowInsecure: allowInsecureRoutes,
940 });
941 if (routes === null) return Option.none<ConnectionCatalogEntry>();
942 const next = entryWithRoutes(entry, routes);
943 const previousProfiles = new Map(
944 connectionRoutes(entry).map((route) => [
945 connectionRouteId(route.target),
946 Option.getOrNull(route.profile),
947 ]),
948 );
949 for (const route of routes) {
950 const profile = Option.getOrNull(route.profile);
951 if (
952 profile === null ||
953 previousProfiles.get(connectionRouteId(route.target)) === profile
954 ) {
955 continue;
956 }
957 yield* profiles.put(profile);
958 }
959 yield* registrations.setRoutes(input.environmentId, persistedRoutes(next));
960 const lease = (yield* SubscriptionRef.get(serviceScopes)).get(input.environmentId);
961 if (lease !== undefined) {
962 yield* SubscriptionRef.update(serviceScopes, (current) =>
963 new Map(current).set(input.environmentId, { ...lease, entry: next }),
964 );
965 }
966 yield* SubscriptionRef.update(entries, (current) =>
967 new Map(current).set(input.environmentId, next),
968 );
969 return Option.some(next);
970 }),
971 ).pipe(
972 Effect.catch((error) =>
973 Effect.logWarning("Could not save routes learned from the environment.", {
974 environmentId: input.environmentId,
975 error,
976 }).pipe(Effect.as(Option.none<ConnectionCatalogEntry>())),
977 ),
978 );
979 });
980
981 const removeRoute = Effect.fn("EnvironmentRegistry.removeRoute")(function* (
982 environmentId: EnvironmentId,
983 routeId: string,
984 ) {
985 // One lock for the whole decision: a route registered between "this was
986 // the last route" and the removal must not be deleted with it.
987 yield* withLeaseLock(
988 environmentId,
989 Effect.gen(function* () {
990 const entry = yield* userEntry(environmentId);
991 const routes = connectionRoutes(entry);
992 const route = routes.find((candidate) => connectionRouteId(candidate.target) === routeId);
993 if (route === undefined) return;
994 const remaining = routesAfterRemoving(routes, routeId);
995 if (remaining.length === 0) return yield* removeLocked(environmentId);
996 yield* replaceRoutesLocked(entry, remaining);
997 const profile = Option.getOrNull(route.profile);
998 if (profile !== null && isSshConnectionProfile(profile)) {
999 yield* disconnectSsh(environmentId, profile);
1000 }
1001 }),
1002 );
1003 });
1004
1005 const reorderRoutes = Effect.fn("EnvironmentRegistry.reorderRoutes")(function* (
1006 environmentId: EnvironmentId,
1007 routeIds: ReadonlyArray<string>,
1008 ) {
1009 yield* withLeaseLock(
1010 environmentId,
1011 Effect.gen(function* () {
1012 const entry = yield* userEntry(environmentId);
1013 const routes = connectionRoutes(entry);
1014 const byId = new Map(routes.map((route) => [connectionRouteId(route.target), route]));
1015 const reordered = routeIds.flatMap((id) => byId.get(id) ?? []);
1016 if (reordered.length !== routes.length || new Set(routeIds).size !== routes.length) {
1017 return yield* new ConnectionBlockedError({
1018 reason: "configuration",
1019 detail: "The route order must list every saved route once.",
1020 });
1021 }
1022 if (reordered.every((route, index) => route === routes[index])) return;
1023 yield* replaceRoutesLocked(entry, reordered);
1024 }),
1025 );
1026 });
1027
1028 const removeRelayEnvironments = Effect.fn("EnvironmentRegistry.removeRelayEnvironments")(
1029 function* () {
1030 const relayEnvironmentIds = [...(yield* SubscriptionRef.get(entries)).values()]
1031 .filter((entry) =>
1032 connectionRoutes(entry).some((route) => route.target._tag === "RelayConnectionTarget"),
1033 )
1034 .map((entry) => entry.target.environmentId);
1035
1036 yield* Effect.forEach(
1037 relayEnvironmentIds,
1038 (environmentId) =>
1039 removeRoute(environmentId, RELAY_ROUTE_ID).pipe(
1040 Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }),
1041 ),
1042 {
1043 concurrency: "unbounded",
1044 discard: true,
1045 },
1046 );
1047 },
1048 );
1049
1050 const retryNow = (environmentId: EnvironmentId) =>
1051 acquireSupervisor(environmentId).pipe(
1052 Effect.flatMap((supervisor) => supervisor.retryNow),
1053 Effect.catchTags({ EnvironmentNotRegisteredError: () => Effect.void }),
1054 Effect.withSpan("EnvironmentRegistry.retryNow"),
1055 );
1056 const setEnabled = Effect.fn("EnvironmentRegistry.setEnabled")(function* (
1057 environmentId: EnvironmentId,
1058 enabled: boolean,
1059 ) {
1060 yield* withLeaseLock(
1061 environmentId,
1062 Effect.gen(function* () {
1063 const entry = yield* getEntry(environmentId);
1064 const retryUnsupported = enabled && entry.unsupportedReason !== undefined;
1065 if (retryUnsupported && !hasSavedBearerRoute(entry)) {
1066 return yield* new ConnectionBlockedError({
1067 reason: "unsupported",
1068 detail: entry.unsupportedReason,
1069 });
1070 }
1071 if (entry.enabled === enabled && !retryUnsupported) {
1072 return;
1073 }
1074 if (!(yield* Ref.get(platformEnvironmentIds)).has(environmentId)) {
1075 yield* registrations.setEnabled(environmentId, enabled);
1076 }
1077 // A cached rejection can outlive a server update.
1078 const { unsupportedReason: _reason, serverUpdateRequired: _update, ...rest } = entry;
1079 const next: ConnectionCatalogEntry = { ...(retryUnsupported ? rest : entry), enabled };
1080 // Update the lease in place so the supervisor keeps its generation and
1081 // durable streams; `installEntryLocked` would tear it down instead.
1082 const lease = (yield* SubscriptionRef.get(serviceScopes)).get(environmentId);
1083 if (lease !== undefined) {
1084 yield* SubscriptionRef.update(serviceScopes, (current) => {
1085 const nextScopes = new Map(current);
1086 nextScopes.set(environmentId, { ...lease, entry: next });
1087 return nextScopes;
1088 });
1089 }
1090 yield* SubscriptionRef.update(entries, (current) => {
1091 const nextEntries = new Map(current);
1092 nextEntries.set(environmentId, next);
1093 return nextEntries;
1094 });
1095 if (lease !== undefined) {
1096 yield* enabled ? lease.supervisor.connect : lease.supervisor.disconnect;
1097 } else if (enabled) {
1098 yield* createServiceScope(next);
1099 }
1100 // The supervisor only owns the RPC session. A managed SSH backend and
1101 // its tunnel outlive it, so switching off tears those down as well.
1102 if (!enabled) {
1103 for (const route of connectionRoutes(entry)) {
1104 const profile = Option.getOrNull(route.profile);
1105 if (profile !== null && isSshConnectionProfile(profile)) {
1106 yield* disconnectSsh(environmentId, profile);
1107 }
1108 }
1109 }
1110 }),
1111 );
1112 });
1113
1114 const state = Effect.fn("EnvironmentRegistry.state")(function* (environmentId: EnvironmentId) {
1115 const supervisor = yield* acquireSupervisor(environmentId);
1116 return yield* SubscriptionRef.get(supervisor.state);
1117 });
1118 const stateChanges = (environmentId: EnvironmentId) =>
1119 followStream(
1120 environmentId,
1121 Stream.unwrap(
1122 EnvironmentSupervisor.EnvironmentSupervisor.pipe(
1123 Effect.map((supervisor) => SubscriptionRef.changes(supervisor.state)),
1124 ),
1125 ),
1126 );
1127
1128 yield* Effect.addFinalizer(() =>
1129 SubscriptionRef.get(serviceScopes).pipe(
1130 Effect.flatMap((current) =>
1131 Effect.forEach(current.values(), (lease) => Scope.close(lease.scope, Exit.void), {
1132 concurrency: "unbounded",
1133 discard: true,
1134 }),
1135 ),
1136 ),
1137 );
1138 yield* connectivity.changes.pipe(
1139 Stream.runForEach((status) => SubscriptionRef.set(networkStatus, status)),
1140 Effect.forkScoped,
1141 );
1142
1143 const setCompatibility = Effect.fn("EnvironmentRegistry.setCompatibility")(function* (
1144 environmentId: EnvironmentId,
1145 error: ConnectionBlockedError | null,
1146 ) {
1147 yield* withLeaseLock(
1148 environmentId,
1149 Effect.gen(function* () {
1150 const entry = (yield* SubscriptionRef.get(entries)).get(environmentId);
1151 if (
1152 entry === undefined ||
1153 (entry.unsupportedReason === (error?.message ?? undefined) &&
1154 entry.serverUpdateRequired === (error?.serverUpdateRequired ?? undefined))
1155 )
1156 return;
1157 const {
1158 unsupportedReason: _previousReason,
1159 serverUpdateRequired: _previousUpdateRequired,
1160 ...rest
1161 } = entry;
1162 // Compatibility pauses connections without changing the user's saved on/off choice.
1163 const next: ConnectionCatalogEntry =
1164 error === null
1165 ? rest
1166 : {
1167 ...rest,
1168 unsupportedReason: error.message,
1169 ...(error.serverUpdateRequired === true ? { serverUpdateRequired: true } : {}),
1170 };
1171 const lease = (yield* SubscriptionRef.get(serviceScopes)).get(environmentId);
1172 if (lease !== undefined) {
1173 yield* SubscriptionRef.update(serviceScopes, (current) =>
1174 new Map(current).set(environmentId, { ...lease, entry: next }),
1175 );
1176 if (error !== null) yield* lease.supervisor.disconnect;
1177 else if (next.enabled) yield* lease.supervisor.connect;
1178 }
1179 yield* SubscriptionRef.update(entries, (current) =>
1180 new Map(current).set(environmentId, next),
1181 );
1182 }),
1183 );
1184 });
1185
1186 return EnvironmentRegistry.of({
1187 entries,
1188 networkStatus,
1189 start,
1190 register,
1191 assertKnownAddress,
1192 reconcilePlatform,
1193 remove,
1194 removeRoute,
1195 reorderRoutes,
1196 removeRelayEnvironments,
1197 retryNow,
1198 setEnabled,
1199 setCompatibility,
1200 state,
1201 stateChanges,
1202 run,
1203 runStream,
1204 followStream,
1205 });
1206});
1207
1208export const layer = Layer.effect(EnvironmentRegistry, make);