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";
18import * as Hub from "./hub.ts";
19import * as Wakeups from "./wakeups.ts";
20
21export const watchDiscoveredCompatibility = Effect.fn("connection.watchDiscoveredCompatibility")(
22 function* () {
23 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
24 const discovery = yield* RelayEnvironmentDiscovery.RelayEnvironmentDiscovery;
25 const seenChecks = new Map<string, RelayEnvironmentStatusResponse>();
26 yield* Stream.merge(
27 SubscriptionRef.changes(discovery.state),
28 SubscriptionRef.changes(registry.entries),
29 ).pipe(
30 Stream.runForEach(() =>
31 Effect.gen(function* () {
32 const current = yield* SubscriptionRef.get(discovery.state);
33 if (!current.refreshing) {
34 for (const environmentId of seenChecks.keys()) {
35 if (!current.environments.has(environmentId)) seenChecks.delete(environmentId);
36 }
37 }
38 const registered = yield* SubscriptionRef.get(registry.entries);
39 for (const entry of current.environments.values()) {
40 const status = Option.getOrNull(entry.status);
41 const descriptor = status?.descriptor;
42 if (status === null || descriptor === undefined) continue;
43 const environmentId = entry.environment.environmentId;
44 // Discovery describes the server behind the relay route. A direct
45 // connection (the desktop's own server, a saved URL, SSH) can reach
46 // a different server with the same environment id, such as a
47 // preview app that shares the home directory. Its socket handshake
48 // already checks the protocol.
49 const saved = registered.get(environmentId);
50 if (
51 saved === undefined ||
52 connectionRoutes(saved).length !== 1 ||
53 !hasRelayRoute(saved)
54 )
55 continue;
56 const previous = seenChecks.get(environmentId);
57 const fresh =
58 previous?.checkedAt !== status.checkedAt ||
59 (previous.descriptor?.orchestrationProtocolVersion ?? 1) !==
60 (descriptor.orchestrationProtocolVersion ?? 1) ||
61 previous.descriptor?.serverVersion !== descriptor.serverVersion;
62 const error = orchestrationProtocolCompatibilityError(descriptor);
63 // A replayed health result must not clear a newer socket rejection.
64 if (error !== null || fresh) yield* registry.setCompatibility(environmentId, error);
65 seenChecks.set(environmentId, status);
66 }
67 }).pipe(
68 Effect.catch((error) =>
69 Effect.logWarning("Could not apply discovered environment compatibility.", { error }),
70 ),
71 ),
72 ),
73 );
74 },
75);
76
77export function layerWithOptions(options: RpcSession.RpcSessionOptions) {
78 const layerDriver = ConnectionDriver.layer.pipe(
79 Layer.provide(Layer.mergeAll(ConnectionResolver.layer, RpcSession.layer(options))),
80 );
81 const layerRegistry = EnvironmentRegistry.layer.pipe(Layer.provide(layerDriver));
82 const layerOnboarding = ConnectionOnboarding.layer.pipe(Layer.provide(layerRegistry));
83 const layerHub = Hub.layer.pipe(Layer.provide(layerRegistry));
84 const layerConnectionServices = Layer.mergeAll(
85 layerRegistry,
86 RelayEnvironmentDiscovery.layer,
87 layerOnboarding,
88 layerHub,
89 // Exposed for updating hosts too old to connect through the driver.
90 ConnectionResolver.layer,
91 );
92 const layerConnectionStartup = Layer.effectDiscard(
93 Effect.gen(function* () {
94 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
95 const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource;
96 const hub = yield* Hub.HubConnection;
97 const wakeups = yield* Wakeups.ConnectionWakeups;
98 yield* watchDiscoveredCompatibility().pipe(Effect.forkScoped);
99 yield* registry.start;
100 yield* Stream.merge(Stream.tick("1 minute"), wakeups.changes).pipe(
101 Stream.runForEach(() => hub.refresh.pipe(Effect.ignore)),
102 Effect.forkScoped,
103 );
104 yield* hub.refresh.pipe(Effect.ignore, Effect.forkScoped);
105 yield* Stream.tick("15 seconds").pipe(
106 Stream.runForEach(() => hub.refreshInbox.pipe(Effect.ignore)),
107 Effect.forkScoped,
108 );
109 yield* platformSource.registrations.pipe(
110 Stream.runForEach(registry.reconcilePlatform),
111 Effect.forkScoped,
112 );
113 }).pipe(Effect.withSpan("clientRuntime.connection.application.start")),
114 );
115 return layerConnectionStartup.pipe(
116 Layer.provideMerge(layerConnectionServices),
117 Layer.provideMerge(RemoteEnvironmentAuthorization.layer),
118 );
119}
120
121export const layer = layerWithOptions({});