1// @effect-diagnostics nodeBuiltinImport:off -- Reads the private pipes inherited from Electron.
2import {
3 DesktopTerminalRequest,
4 DesktopTerminalResponse,
5 TerminalRpcGroup,
6 ORCHESTRATION_PROTOCOL_VERSION,
7 ORCHESTRATION_PROTOCOL_QUERY_PARAM,
8} from "@t3tools/contracts";
9import * as NodeStream from "@effect/platform-node/NodeStream";
10import * as Exit from "effect/Exit";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Schema from "effect/Schema";
16import * as Scope from "effect/Scope";
17import * as Semaphore from "effect/Semaphore";
18import * as Stream from "effect/Stream";
19import * as Ndjson from "effect/encoding/Ndjson";
20import { HttpServerRequest } from "effect/http";
21import { RpcServer, RpcSerialization } from "effect/rpc";
22import type { FromClientEncoded } from "effect/rpc/RpcMessage";
23import * as NodeNet from "node:net";
24import * as ServerConfig from "../config.ts";
25import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
26import * as SessionStore from "../auth/SessionStore.ts";
27import * as RpcAuthorization from "../auth/RpcAuthorization.ts";
28import {
29 RpcInstrumentation,
30 rpcInstrumentationLayer,
31} from "../observability/RpcInstrumentation.ts";
32import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
33import * as TerminalManager from "./Manager.ts";
34import { makeTerminalRpcHandlers } from "./Rpc.ts";
35import { withTerminalOutputWindow } from "./OutputProtocol.ts";
36
37const group = TerminalRpcGroup.middleware(RpcInstrumentation);
38const decode = Schema.decodeUnknownOption(DesktopTerminalRequest);
39const encode = Schema.encodeSync(Schema.fromJsonString(DesktopTerminalResponse));
40
41export const run = (
42 requests: Stream.Stream<DesktopTerminalRequest>,
43 send: (response: DesktopTerminalResponse) => Effect.Effect<void>,
44) =>
45 Effect.gen(function* () {
46 const auth = yield* EnvironmentAuth.EnvironmentAuth;
47 const sessions = yield* SessionStore.SessionStore;
48 const manager = yield* TerminalManager.TerminalManager;
49 const identity = yield* ServerEnvironment.ServerEnvironmentIdentity;
50 const environmentId = yield* identity.getEnvironmentId;
51 const parentScope = yield* Scope.Scope;
52 const serialization = yield* RpcSerialization.RpcSerialization;
53 const connections = new Map<
54 string,
55 { scope: Scope.Closeable; messages: Queue.Queue<string> }
56 >();
57 const close = (id: string) =>
58 Effect.gen(function* () {
59 const connection = connections.get(id);
60 if (!connection) return;
61 connections.delete(id);
62 yield* Scope.close(connection.scope, Exit.void);
63 yield* send({ type: "close", id });
64 });
65 const open = (request: Extract<DesktopTerminalRequest, { type: "open" }>) =>
66 Effect.gen(function* () {
67 if (connections.has(request.id)) return;
68 const scope = yield* Scope.fork(parentScope);
69 const messages = yield* Queue.unbounded<string>();
70 connections.set(request.id, { scope, messages });
71 yield* Effect.gen(function* () {
72 yield* Effect.addFinalizer(() => Queue.shutdown(messages));
73 const url = new URL(request.url);
74 if (
75 request.environmentId !== environmentId ||
76 url.pathname !== "/ws" ||
77 url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) !==
78 String(ORCHESTRATION_PROTOCOL_VERSION)
79 )
80 return;
81 url.protocol = url.protocol === "wss:" ? "https:" : "http:";
82 const httpRequest = HttpServerRequest.fromWeb(new Request(url));
83 const session = yield* auth.authenticateWebSocketUpgrade(httpRequest);
84 const protocol = yield* RpcServer.Protocol.make((receive) =>
85 Effect.gen(function* () {
86 const parser = serialization.makeUnsafe();
87 yield* Stream.fromQueue(messages).pipe(
88 Stream.runForEach((data) =>
89 // The RPC parser owns encoded envelopes; RpcServer validates payloads against the group.
90 Effect.forEach(
91 parser.decode(data) as ReadonlyArray<FromClientEncoded>,
92 (message) => receive(0, message),
93 { discard: true },
94 ),
95 ),
96 Effect.forkScoped,
97 );
98 return {
99 disconnects: yield* Queue.unbounded<number>(),
100 send: (_clientId, message) =>
101 Effect.suspend(() => {
102 const data = parser.encode(message);
103 return data === undefined
104 ? Effect.void
105 : send({
106 type: "message",
107 id: request.id,
108 data: typeof data === "string" ? data : new TextDecoder().decode(data),
109 });
110 }),
111 end: () => Queue.shutdown(messages),
112 clientIds: Effect.succeed(new Set([0])),
113 initialMessage: Effect.succeedNone,
114 supportsAck: true,
115 supportsTransferables: false,
116 supportsSpanPropagation: true,
117 supportsNotifications: true,
118 codecFor: serialization.codecFor,
119 };
120 }),
121 );
122 yield* Effect.acquireUseRelease(
123 sessions.markConnected(session.sessionId),
124 () =>
125 Effect.raceFirst(
126 RpcServer.make(group, { disableTracing: true, disableFatalDefects: true }).pipe(
127 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
128 Effect.provide(
129 Layer.mergeAll(
130 group.toLayer(makeTerminalRpcHandlers(manager)),
131 RpcAuthorization.layer(session.scopes),
132 rpcInstrumentationLayer,
133 ),
134 ),
135 ),
136 sessions.awaitInvalidation(session.sessionId),
137 ),
138 () => sessions.markDisconnected(session.sessionId),
139 );
140 }).pipe(
141 Scope.provide(scope),
142 Effect.catchCause(() => Effect.void),
143 Effect.ensuring(Effect.forkIn(close(request.id), parentScope).pipe(Effect.asVoid)),
144 Effect.forkIn(scope),
145 );
146 });
147 yield* send({ type: "ready", environmentId });
148 yield* requests.pipe(
149 Stream.runForEach((value) => {
150 const request = value;
151 if (request.type === "open") return open(request);
152 if (request.type === "close") return close(request.id);
153 const connection = connections.get(request.id);
154 return connection ? Queue.offer(connection.messages, request.data) : Effect.void;
155 }),
156 Effect.ensuring(
157 Effect.suspend(() => Effect.forEach([...connections.keys()], close, { discard: true })),
158 ),
159 );
160 }).pipe(Effect.provide(RpcSerialization.layerJson));
161
162export const layer = Layer.effectDiscard(
163 Effect.gen(function* () {
164 const config = yield* ServerConfig.ServerConfig;
165 if (config.desktopTerminalFd === undefined || config.desktopTerminalControlFd === undefined)
166 return;
167 const outputFd = config.desktopTerminalControlFd;
168 const lock = yield* Semaphore.make(1);
169 const writable = yield* Effect.acquireRelease(
170 Effect.sync(() => {
171 const socket = new NodeNet.Socket({ fd: outputFd, readable: false, writable: true });
172 socket.on("error", () => socket.destroy());
173 return socket;
174 }),
175 (socket) => Effect.sync(() => socket.destroy()),
176 );
177 const readable = yield* Effect.acquireRelease(
178 Effect.sync(
179 () => new NodeNet.Socket({ fd: config.desktopTerminalFd, readable: true, writable: false }),
180 ),
181 (stream) => Effect.sync(() => stream.destroy()),
182 );
183 const requests = NodeStream.fromReadable({
184 evaluate: () => readable,
185 onError: (cause) => cause,
186 }).pipe(
187 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
188 Stream.map((value) => decode(value)),
189 Stream.filter(Option.isSome),
190 Stream.map((value) => value.value),
191 Stream.orDie,
192 );
193 yield* run(requests, (response) =>
194 lock.withPermits(1)(
195 Effect.callback<void>((resume) => {
196 writable.write(Buffer.from(`${encode(response)}\n`), (error) =>
197 resume(error ? Effect.die(error) : Effect.void),
198 );
199 }),
200 ),
201 ).pipe(Effect.forkScoped);
202 }),
203);