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