1import type { RelayEnvironmentStatusResponse } from "@t3tools/contracts/relay";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import * as Option from "effect/Option";
6import * as SubscriptionRef from "effect/SubscriptionRef";
7import { orchestrationProtocolCompatibilityError } from "./compatibility.ts";
8
9import * as ConnectionResolver from "./resolver.ts";
10import * as ConnectionDriver from "./driver.ts";
11import * as EnvironmentRegistry from "./registry.ts";
12import * as ConnectionOnboarding from "./onboarding.ts";
13import { connectionRoutes, hasRelayRoute } from "./routes.ts";
14import * as PlatformConnectionSource from "../platform/source.ts";
15import * as RelayEnvironmentDiscovery from "../relay/discovery.ts";
16import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
17import * as RpcSession from "../rpc/session.ts";
18
19export const watchDiscoveredCompatibility = Effect.fn("connection.watchDiscoveredCompatibility")(
20 function* () {
21 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
22 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
23 const seenChecks = new Map<string, RelayEnvironmentStatusResponse>();
24 yield* Stream.merge(
25 SubscriptionRef.changes(discovery.state),
26 SubscriptionRef.changes(registry.entries),
27 ).pipe(
28 Stream.runForEach(() =>
29 Effect.gen(function* () {
30 const current = yield* SubscriptionRef.get(discovery.state);
31 if (!current.refreshing) {
32 for (const environmentId of seenChecks.keys()) {
33 if (!current.environments.has(environmentId)) seenChecks.delete(environmentId);
34 }
35 }
36 const registered = yield* SubscriptionRef.get(registry.entries);
37 for (const entry of current.environments.values()) {
38 const status = Option.getOrNull(entry.status);
39 const descriptor = status?.descriptor;
40 if (status === null || descriptor === undefined) continue;
41 const environmentId = entry.environment.environmentId;
42 // Discovery describes the server behind the relay route. A direct
43 // connection (the desktop's own server, a saved URL, SSH) can reach
44 // a different server with the same environment id, such as a
45 // preview app that shares the home directory.
46 const saved = registered.get(environmentId);
47 if (
48 saved === undefined ||
49 connectionRoutes(saved).length !== 1 ||
50 !hasRelayRoute(saved)
51 )
52 continue;
53 const previous = seenChecks.get(environmentId);
54 const fresh =
55 previous?.checkedAt !== status.checkedAt ||
56 (previous.descriptor?.orchestrationProtocolVersion ?? 1) !==
57 (descriptor.orchestrationProtocolVersion ?? 1) ||
58 previous.descriptor?.serverVersion !== descriptor.serverVersion;
59 const error = orchestrationProtocolCompatibilityError(descriptor);
60 // A replayed health result must not clear a newer socket rejection.
61 if (error !== null || fresh) yield* registry.setCompatibility(environmentId, error);
62 seenChecks.set(environmentId, status);
63 }
64 }).pipe(
65 Effect.catch((error) =>
66 Effect.logWarning("Could not apply discovered environment compatibility.", { error }),
67 ),
68 ),
69 ),
70 );
71 },
72);
73
74export function layerWithOptions(options: RpcSession.RpcSessionOptions) {
75 const layerDriver = ConnectionDriver.layer.pipe(
76 Layer.provide(Layer.mergeAll(ConnectionResolver.layer, RpcSession.layer(options))),
77 );
78 const layerRegistry = EnvironmentRegistry.layer.pipe(Layer.provide(layerDriver));
79 const layerOnboarding = ConnectionOnboarding.layer.pipe(Layer.provide(layerRegistry));
80 const layerConnectionServices = Layer.mergeAll(
81 layerRegistry,
82 RelayEnvironmentDiscovery.layer,
83 layerOnboarding,
84 ConnectionResolver.layer,
85 );
86 const layerConnectionStartup = Layer.effectDiscard(
87 Effect.gen(function* () {
88 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
89 const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource;
90 yield* watchDiscoveredCompatibility().pipe(Effect.forkScoped);
91 yield* registry.start;
92 yield* platformSource.registrations.pipe(
93 Stream.runForEach(registry.reconcilePlatform),
94 Effect.forkScoped,
95 );
96 }).pipe(Effect.withSpan("clientRuntime.connection.application.start")),
97 );
98 return layerConnectionStartup.pipe(
99 Layer.provideMerge(layerConnectionServices),
100 Layer.provideMerge(RemoteEnvironmentAuthorization.layer),
101 );
102}
103
104export const layer = layerWithOptions({});