1import {
2 type ServerConfig,
3 type ServerConfigStreamEvent,
4 WsSubscribeServerConfigRpc,
5 WS_METHODS,
6} from "@t3tools/contracts";
7import * as Cause from "effect/Cause";
8import * as Context from "effect/Context";
9import * as Deferred from "effect/Deferred";
10import * as Effect from "effect/Effect";
11import * as Equal from "effect/Equal";
12import * as Exit from "effect/Exit";
13import * as Fiber from "effect/Fiber";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as PubSub from "effect/PubSub";
17import * as Ref from "effect/Ref";
18import * as Schedule from "effect/Schedule";
19import * as Schema from "effect/Schema";
20import * as Scope from "effect/Scope";
21import * as Stream from "effect/Stream";
22import type * as Rpc from "effect/rpc/Rpc";
23import * as RpcClient from "effect/rpc/RpcClient";
24import * as RpcClientError from "effect/rpc/RpcClientError";
25import * as RpcSerialization from "effect/rpc/RpcSerialization";
26import * as Socket from "effect/socket/Socket";
27
28import { makeWsRpcProtocolClient, PING_TIMEOUT, type WsRpcProtocolClient } from "./protocol.ts";
29import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
30import type { ConnectionAttemptError, PreparedConnection } from "../connection/model.ts";
31import {
32 ConnectionBlockedError,
33 ConnectionTransientError as ConnectionTransientErrorClass,
34} from "../connection/model.ts";
35import {
36 applyServerConfigProjection,
37 type ServerConfigProjection,
38 withoutEnvironmentThemes,
39} from "../state/serverConfigProjection.ts";
40import { environmentMismatchError } from "../connection/errors.ts";
41
42const SOCKET_OPEN_TIMEOUT = "15 seconds";
43
44export interface RpcSession {
45 readonly client: WsRpcProtocolClient;
46 readonly initialConfig: Effect.Effect<ServerConfig, ConnectionAttemptError>;
47 readonly subscribeServerConfig: (
48 input: ServerConfigSubscriptionInput,
49 ) => ServerConfigSubscription;
50 readonly ready: Effect.Effect<void, ConnectionAttemptError>;
51 readonly probe: Effect.Effect<void, ConnectionAttemptError>;
52 readonly closed: Effect.Effect<never, ConnectionAttemptError>;
53}
54
55export interface RpcSessionOptions {
56 readonly terminalSocket?: (prepared: PreparedConnection) => Socket.Socket | undefined;
57 readonly environmentThemes?: boolean;
58 readonly usageLimitSources?: boolean;
59 /** This client answers /usage-limits itself, so the server may advertise it. */
60 readonly usageLimitsCommand?: boolean;
61}
62
63export class RpcSessionFactory extends Context.Service<
64 RpcSessionFactory,
65 {
66 readonly connect: (
67 connection: PreparedConnection,
68 prepareTerminalConnection: Effect.Effect<PreparedConnection, ConnectionAttemptError>,
69 ) => Effect.Effect<RpcSession, ConnectionAttemptError, Scope.Scope>;
70 }
71>()("@t3tools/client-runtime/rpc/session/RpcSessionFactory") {}
72
73type InitialConfigError = Effect.Error<
74 ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverGetConfig]>
75>;
76type ProbeError = Effect.Error<ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverProbe]>>;
77type ServerConfigSubscriptionError =
78 | Rpc.ErrorExit<typeof WsSubscribeServerConfigRpc>
79 | RpcClientError.RpcClientError;
80type ServerConfigSubscription = Stream.Stream<
81 ServerConfigStreamEvent,
82 ServerConfigSubscriptionError
83>;
84type ServerConfigSubscriptionInput = Parameters<
85 WsRpcProtocolClient[typeof WS_METHODS.subscribeServerConfig]
86>[0];
87type EnvironmentThemesUpdatedEvent = Extract<
88 ServerConfigStreamEvent,
89 { readonly type: "environmentThemesUpdated" }
90>;
91type UsageLimitSourcesUpdatedEvent = Extract<
92 ServerConfigStreamEvent,
93 { readonly type: "usageLimitSourcesUpdated" }
94>;
95
96interface ServerConfigReplayState {
97 readonly projection: ServerConfigProjection;
98 readonly revision: number;
99 readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined;
100 readonly sourcesEvent: UsageLimitSourcesUpdatedEvent | undefined;
101}
102
103interface BufferedServerConfigEvent {
104 readonly event: ServerConfigStreamEvent;
105 readonly replay: ServerConfigReplayState;
106 readonly revision: number;
107}
108
109function serverConfigReplayEvents(
110 state: ServerConfigReplayState,
111): ReadonlyArray<ServerConfigStreamEvent> {
112 const snapshot = {
113 version: 1 as const,
114 type: "snapshot" as const,
115 config: withoutEnvironmentThemes(state.projection.config),
116 };
117 return [
118 snapshot,
119 ...(state.themesEvent === undefined ? [] : [state.themesEvent]),
120 ...(state.sourcesEvent === undefined ? [] : [state.sourcesEvent]),
121 ];
122}
123
124const isSocketErrorReason = Schema.is(Socket.SocketErrorReason);
125
126function mapSessionRpcError(
127 error: InitialConfigError | ProbeError | ServerConfigSubscriptionError,
128 networkHint: string,
129): ConnectionAttemptError {
130 switch (error._tag) {
131 case "EnvironmentAuthorizationError":
132 return new ConnectionBlockedError({
133 reason: "permission",
134 detail: error.message,
135 });
136 case "KeybindingsConfigParseError":
137 case "ServerSettingsError":
138 return new ConnectionTransientErrorClass({
139 reason: "remote-unavailable",
140 detail: error.message,
141 });
142 case "RpcClientError":
143 return new ConnectionTransientErrorClass({
144 reason: "transport",
145 detail: `${error.message}${isSocketErrorReason(error.reason) ? networkHint : ""}`,
146 });
147 }
148}
149
150/** @public Service construction is part of the canonical Effect module API. */
151export const make = Effect.fn("RpcSessionFactory.make")(function* (
152 options: RpcSessionOptions = {},
153) {
154 const webSocketConstructor = yield* Socket.WebSocketConstructor;
155 const serverConfigInput: ServerConfigSubscriptionInput = {
156 ...(options.environmentThemes === true ? { environmentThemes: true } : {}),
157 ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}),
158 ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}),
159 };
160
161 const connect = Effect.fnUntraced(function* (
162 connection: PreparedConnection,
163 prepareTerminalConnection: Effect.Effect<PreparedConnection, ConnectionAttemptError>,
164 ) {
165 const networkHint =
166 connection.target._tag === "RelayConnectionTarget" ? ` ${NETWORK_BLOCKING_HINT}` : "";
167 const mapRpcError = (error: Parameters<typeof mapSessionRpcError>[0]) =>
168 mapSessionRpcError(error, networkHint);
169 yield* Effect.annotateCurrentSpan({
170 "connection.environment.id": connection.environmentId,
171 });
172
173 const scope = yield* Effect.scope;
174 const disconnected = yield* Deferred.make<never, ConnectionAttemptError>();
175 const openTransport = Effect.fnUntraced(function* (
176 prepared: PreparedConnection,
177 terminal = false,
178 ) {
179 const connected = yield* Deferred.make<void>();
180 const pingTimedOut = yield* Ref.make(false);
181 const hooks = RpcClient.ConnectionHooks.of({
182 onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid),
183 onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe(
184 Effect.flatMap(([wasConnected, timedOut]) =>
185 Deferred.fail(
186 disconnected,
187 new ConnectionTransientErrorClass({
188 reason: "transport",
189 detail: `${
190 !wasConnected
191 ? `${connection.label} could not establish a WebSocket connection.`
192 : timedOut
193 ? `${connection.label} stopped responding.`
194 : `${connection.label} disconnected.`
195 }${networkHint}`,
196 }),
197 ),
198 ),
199 Effect.asVoid,
200 ),
201 });
202 const localSocket = terminal ? options.terminalSocket?.(prepared) : undefined;
203 const layerSocket = localSocket
204 ? Layer.succeed(Socket.Socket, localSocket)
205 : Socket.layerWebSocket(prepared.socketUrl, {
206 openTimeout: SOCKET_OPEN_TIMEOUT,
207 }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)));
208 const layerProtocol = Layer.effect(
209 RpcClient.Protocol,
210 RpcClient.makeProtocolSocket({
211 pingTimeout: PING_TIMEOUT,
212 retryTransientErrors: false,
213 retryPolicy: Schedule.recurs(0),
214 onPingTimeout: Ref.set(pingTimedOut, true),
215 }),
216 ).pipe(
217 Layer.provide(
218 Layer.mergeAll(
219 layerSocket,
220 RpcSerialization.layerJson,
221 Layer.succeed(RpcClient.ConnectionHooks, hooks),
222 ),
223 ),
224 );
225 const protocolContext = yield* Layer.build(layerProtocol).pipe(
226 Effect.withSpan("environment.websocket.connect"),
227 );
228 const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
229 return { client, connected: Deferred.await(connected) };
230 });
231 const transport = yield* openTransport(connection);
232 const protocolClient = transport.client;
233 // Canceling one command must not abandon the connection shared by other terminals.
234 const terminalTransport = yield* prepareTerminalConnection.pipe(
235 Effect.flatMap((prepared) => openTransport(prepared, true)),
236 Effect.tapError((error) => Deferred.fail(disconnected, error)),
237 Effect.map(({ client }) => client),
238 Effect.mapError(
239 (cause) =>
240 new RpcClientError.RpcClientError({
241 reason: new RpcClientError.RpcClientDefect({
242 message: `${connection.label} could not establish a WebSocket connection.`,
243 cause,
244 }),
245 }),
246 ),
247 Scope.provide(scope),
248 Effect.forkIn(scope),
249 Effect.cached,
250 );
251 const terminalClient = Effect.flatMap(terminalTransport, Fiber.join);
252 const terminalStream = <
253 Tag extends
254 | typeof WS_METHODS.terminalAttach
255 | typeof WS_METHODS.terminalObserve
256 | typeof WS_METHODS.subscribeTerminalEvents,
257 >(
258 tag: Tag,
259 ): WsRpcProtocolClient[Tag] => {
260 type Method = WsRpcProtocolClient[Tag];
261 type Input = Parameters<Method>[0];
262 type Options = Parameters<Method>[1];
263
264 // The tag correlates its payload and result; `asQueue` selects the RPC overload at runtime.
265 return ((input: Input, options?: Options) => {
266 const stream = Stream.unwrap(
267 Effect.map(terminalClient, (client) => {
268 const method = client[tag] as (
269 input: Input,
270 options: Options & { readonly asQueue: false },
271 ) => Stream.Stream<unknown, unknown>;
272 return method(input, { ...options, asQueue: false });
273 }),
274 );
275 return options?.asQueue
276 ? Stream.toQueue(stream, { capacity: options.streamBufferSize ?? 16 })
277 : stream;
278 }) as Method;
279 };
280 // Metadata stays on the main socket: sidebars observe it without opening a terminal.
281 const client: WsRpcProtocolClient = {
282 ...protocolClient,
283 [WS_METHODS.terminalOpen]: (input, options) =>
284 Effect.flatMap(terminalClient, (client) => client[WS_METHODS.terminalOpen](input, options)),
285 [WS_METHODS.terminalWrite]: (input, options) =>
286 Effect.flatMap(terminalClient, (client) =>
287 client[WS_METHODS.terminalWrite](input, options),
288 ),
289 [WS_METHODS.terminalResize]: (input, options) =>
290 Effect.flatMap(terminalClient, (client) =>
291 client[WS_METHODS.terminalResize](input, options),
292 ),
293 [WS_METHODS.terminalClear]: (input, options) =>
294 Effect.flatMap(terminalClient, (client) =>
295 client[WS_METHODS.terminalClear](input, options),
296 ),
297 [WS_METHODS.terminalRestart]: (input, options) =>
298 Effect.flatMap(terminalClient, (client) =>
299 client[WS_METHODS.terminalRestart](input, options),
300 ),
301 [WS_METHODS.terminalClose]: (input, options) =>
302 Effect.flatMap(terminalClient, (client) =>
303 client[WS_METHODS.terminalClose](input, options),
304 ),
305 [WS_METHODS.terminalAttach]: terminalStream(WS_METHODS.terminalAttach),
306 [WS_METHODS.terminalObserve]: terminalStream(WS_METHODS.terminalObserve),
307 [WS_METHODS.subscribeTerminalEvents]: terminalStream(WS_METHODS.subscribeTerminalEvents),
308 };
309 const initialConfigDeferred = yield* Deferred.make<ServerConfig>();
310 const serverConfigExit = yield* Deferred.make<void, ServerConfigSubscriptionError>();
311 const configSubscriptionClosed = yield* Deferred.make<never, ConnectionAttemptError>();
312 const serverConfigState = yield* Ref.make(Option.none<ServerConfigReplayState>());
313 const serverConfigUpdates = yield* PubSub.sliding<BufferedServerConfigEvent>(64);
314 const configSubscriptionEndedError = new ConnectionTransientErrorClass({
315 reason: "remote-unavailable",
316 detail: `${connection.label} config subscription ended.`,
317 });
318 const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig](
319 serverConfigInput,
320 ).pipe(
321 Stream.runForEach((event) =>
322 Effect.gen(function* () {
323 const buffered = yield* Ref.modify(serverConfigState, (current) => {
324 const projection = applyServerConfigProjection(
325 Option.map(current, (state) => state.projection),
326 event,
327 );
328 if (Option.isNone(projection)) {
329 return [Option.none<BufferedServerConfigEvent>(), current] as const;
330 }
331 const next = {
332 projection: projection.value,
333 revision: Option.match(current, {
334 onNone: () => 1,
335 onSome: (state) => state.revision + 1,
336 }),
337 themesEvent:
338 event.type === "environmentThemesUpdated"
339 ? event
340 : event.type === "snapshot" &&
341 event.config.environment.capabilities.environmentThemes !== true
342 ? undefined
343 : Option.getOrUndefined(current)?.themesEvent,
344 sourcesEvent:
345 event.type === "usageLimitSourcesUpdated"
346 ? event
347 : event.type === "snapshot" &&
348 event.config.environment.capabilities.usageLimitSources !== true
349 ? undefined
350 : Option.getOrUndefined(current)?.sourcesEvent,
351 } satisfies ServerConfigReplayState;
352 return [
353 Option.some({ event, replay: next, revision: next.revision }),
354 Option.some(next),
355 ] as const;
356 });
357 if (Option.isSome(buffered)) {
358 yield* PubSub.publish(serverConfigUpdates, buffered.value);
359 }
360 if (event.type === "snapshot") {
361 yield* Deferred.succeed(initialConfigDeferred, event.config);
362 }
363 }),
364 ),
365 Effect.onExit((exit) => {
366 if (Exit.isSuccess(exit)) {
367 return Effect.all([
368 Deferred.succeed(serverConfigExit, undefined),
369 Deferred.fail(configSubscriptionClosed, configSubscriptionEndedError),
370 ]).pipe(Effect.asVoid);
371 }
372 if (Cause.hasInterruptsOnly(exit.cause)) {
373 return Effect.void;
374 }
375 return Effect.all([
376 Deferred.failCause(serverConfigExit, exit.cause),
377 Deferred.failCause(configSubscriptionClosed, Cause.map(exit.cause, mapRpcError)),
378 ]).pipe(Effect.asVoid);
379 }),
380 );
381 yield* serverConfigSource.pipe(Effect.forkScoped);
382 const initialConfig = Effect.raceFirst(
383 Deferred.await(initialConfigDeferred),
384 Deferred.await(serverConfigExit).pipe(
385 Effect.mapError(mapRpcError),
386 Effect.flatMap(() => Effect.fail(configSubscriptionEndedError)),
387 ),
388 ).pipe(
389 Effect.filterOrElse(
390 (config) => config.environment.environmentId === connection.environmentId,
391 (config) =>
392 environmentMismatchError({
393 expected: connection.environmentId,
394 actual: config.environment.environmentId,
395 }),
396 ),
397 Effect.withSpan("environment.initialSync"),
398 );
399 const serverConfigEvents = Stream.unwrap(
400 Effect.gen(function* () {
401 const subscription = yield* PubSub.subscribe(serverConfigUpdates);
402 const snapshot = yield* Ref.get(serverConfigState);
403 if (Option.isNone(snapshot)) {
404 return Stream.empty;
405 }
406 const updates = Stream.fromSubscription(subscription).pipe(
407 Stream.filter((buffered) => buffered.revision > snapshot.value.revision),
408 Stream.mapAccum(
409 () => snapshot.value.revision,
410 (revision, buffered) => [
411 buffered.revision,
412 buffered.revision === revision + 1
413 ? [buffered.event]
414 : serverConfigReplayEvents(buffered.replay),
415 ],
416 ),
417 );
418 const terminal = Stream.fromEffect(Deferred.await(serverConfigExit)).pipe(Stream.drain);
419 return Stream.concat(
420 Stream.fromIterable(serverConfigReplayEvents(snapshot.value)),
421 Stream.merge(updates, terminal, { haltStrategy: "either" }),
422 );
423 }),
424 ).pipe(
425 Stream.catchCause((cause) => {
426 if (Cause.hasInterruptsOnly(cause)) {
427 return Stream.failCause(cause);
428 }
429 // The supervisor keeps the original cause. Shared durable consumers
430 // need a transport-shaped failure so they wait for its replacement.
431 return Stream.fail(
432 new RpcClientError.RpcClientError({
433 reason: new RpcClientError.RpcClientDefect({
434 message: `${connection.label} config subscription failed.`,
435 cause,
436 }),
437 }),
438 );
439 }),
440 );
441 const validatedInitialConfig = initialConfig.pipe(
442 Effect.mapError(
443 (cause) =>
444 new RpcClientError.RpcClientError({
445 reason: new RpcClientError.RpcClientDefect({
446 message: `${connection.label} config subscription failed.`,
447 cause,
448 }),
449 }),
450 ),
451 );
452 const subscribeServerConfig = (input: ServerConfigSubscriptionInput) =>
453 Stream.unwrap(
454 validatedInitialConfig.pipe(
455 Effect.as(
456 Equal.equals(input, serverConfigInput)
457 ? serverConfigEvents
458 : protocolClient[WS_METHODS.subscribeServerConfig](input),
459 ),
460 ),
461 );
462 const probe = initialConfig.pipe(
463 Effect.flatMap((config) =>
464 (config.environment.capabilities.connectionProbe === true
465 ? protocolClient[WS_METHODS.serverProbe]({})
466 : protocolClient[WS_METHODS.serverGetConfig]({})
467 ).pipe(Effect.mapError(mapRpcError)),
468 ),
469 Effect.asVoid,
470 Effect.withSpan("clientRuntime.connection.rpcSession.probe"),
471 );
472
473 return {
474 client,
475 initialConfig,
476 subscribeServerConfig,
477 ready: transport.connected.pipe(
478 Effect.andThen(initialConfig),
479 Effect.asVoid,
480 Effect.raceFirst(Deferred.await(disconnected)),
481 ),
482 probe,
483 closed: Effect.raceFirst(
484 Deferred.await(disconnected),
485 Deferred.await(configSubscriptionClosed),
486 ),
487 } satisfies RpcSession;
488 });
489
490 return RpcSessionFactory.of({ connect });
491});
492
493export const layer = (options: RpcSessionOptions) => Layer.effect(RpcSessionFactory, make(options));