1// @effect-diagnostics nodeBuiltinImport:off - Bridges Playwright's WebSocket transport to the desktop's fds.
2/**
3 * The server end of the desktop browser channel (see `DesktopBrowserEvent`).
4 *
5 * Playwright connects over CDP only through a WebSocket URL, so each attached
6 * desktop tab gets a loopback endpoint with an unguessable path. Its frames
7 * cross the bootstrap file descriptors to the desktop's relay, which owns the
8 * tab's `webContents.debugger`. The endpoint only bridges to that one tab.
9 */
10import * as NodeStream from "@effect/platform-node/NodeStream";
11import * as NodeSocketServer from "@effect/platform-node/NodeSocketServer";
12import {
13 DesktopBrowserCommand,
14 DesktopBrowserEvent,
15 type DesktopBrowserCommand as DesktopBrowserCommandType,
16} from "@t3tools/contracts";
17import * as Context from "effect/Context";
18import * as Duration from "effect/Duration";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as PubSub from "effect/PubSub";
23import * as Queue from "effect/Queue";
24import * as Schema from "effect/Schema";
25import * as Scope from "effect/Scope";
26import * as Semaphore from "effect/Semaphore";
27import * as Stream from "effect/Stream";
28import * as Ndjson from "effect/encoding/Ndjson";
29import * as NodeCrypto from "node:crypto";
30import * as NodeNet from "node:net";
31
32import * as ServerConfig from "../config.ts";
33import { writeAllToFileDescriptor } from "../resourceTelemetry/DesktopTelemetryReceiver.ts";
34
35const decodeEvent = Schema.decodeUnknownEffect(DesktopBrowserEvent);
36const encodeCommand = Schema.encodeEffect(Schema.fromJsonString(DesktopBrowserCommand));
37
38export interface DesktopTabKey {
39 readonly threadId: string;
40 readonly tabId: string;
41}
42
43const keyOf = ({ threadId, tabId }: DesktopTabKey) => `${threadId}\u0000${tabId}`;
44
45export class DesktopBrowserChannel extends Context.Service<
46 DesktopBrowserChannel,
47 {
48 /** False when this server was not started by a desktop app. */
49 readonly available: boolean;
50 /**
51 * Waits for a tab to be attached. Subscribes before it checks, so an
52 * attach landing in between is never missed. False after `timeout`.
53 */
54 readonly awaitAttached: (key: DesktopTabKey, timeout: Duration.Input) => Effect.Effect<boolean>;
55 /** Desktop tabs as they detach. */
56 readonly detached: Stream.Stream<DesktopTabKey>;
57 /** Desktop tabs as they attach, including a tab coming back after its DevTools close. */
58 readonly attached: Stream.Stream<DesktopTabKey>;
59 readonly isAttached: (key: DesktopTabKey) => Effect.Effect<boolean>;
60 /**
61 * A one-connection CDP endpoint for an attached tab. Closing the scope
62 * releases the tab on the desktop and stops the endpoint.
63 */
64 readonly endpoint: (key: DesktopTabKey) => Effect.Effect<string, never, Scope.Scope>;
65 /** Draws the agent's cursor over a tab the desktop renders. */
66 readonly pointer: (
67 key: DesktopTabKey,
68 pointer: { readonly phase: "move" | "click"; readonly x: number; readonly y: number },
69 ) => Effect.Effect<void>;
70 }
71>()("t3/preview/DesktopBrowserChannel") {}
72
73const make = Effect.gen(function* () {
74 const config = yield* ServerConfig.ServerConfig;
75 const inputFd = config.desktopBrowserFd;
76 const controlFd = config.desktopBrowserControlFd;
77 const changes = yield* PubSub.unbounded<{ key: DesktopTabKey; attached: boolean }>();
78 const attachedTabs = new Set<string>();
79 /** CDP frames from the desktop, per tab, for the endpoint connected to it. */
80 const inbound = new Map<string, Queue.Queue<string>>();
81 const writeLock = yield* Semaphore.make(1);
82
83 if (inputFd === undefined || controlFd === undefined) {
84 return DesktopBrowserChannel.of({
85 available: false,
86 awaitAttached: () => Effect.succeed(false),
87 detached: Stream.empty,
88 attached: Stream.empty,
89 isAttached: () => Effect.succeed(false),
90 endpoint: () => Effect.die("No desktop app is attached to this server."),
91 pointer: () => Effect.void,
92 });
93 }
94
95 const command = (message: DesktopBrowserCommandType) =>
96 writeLock.withPermits(1)(
97 encodeCommand(message).pipe(
98 Effect.flatMap((line) => writeAllToFileDescriptor(controlFd, Buffer.from(`${line}\n`))),
99 Effect.catchCause((cause) =>
100 Effect.logWarning("desktop browser command failed", { cause }),
101 ),
102 ),
103 );
104
105 // Socket reads can be cancelled while the desktop keeps its write end open.
106 // Filesystem reads on this pipe keep Node alive after scope shutdown.
107 const readable = yield* Effect.acquireRelease(
108 Effect.sync(() => new NodeNet.Socket({ fd: inputFd, readable: true, writable: false })),
109 (stream) => Effect.sync(() => stream.destroy()),
110 );
111 yield* NodeStream.fromReadable<Uint8Array, Error>({
112 evaluate: () => readable,
113 closeOnDone: true,
114 onError: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
115 }).pipe(
116 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
117 Stream.mapEffect((value) => decodeEvent(value).pipe(Effect.option)),
118 Stream.runForEach((decoded) => {
119 if (Option.isNone(decoded)) return Effect.void;
120 const event = decoded.value;
121 const key = { threadId: event.threadId, tabId: event.tabId };
122 const id = keyOf(key);
123 switch (event.type) {
124 case "cdp": {
125 const queue = inbound.get(id);
126 return queue ? Queue.offer(queue, event.message).pipe(Effect.asVoid) : Effect.void;
127 }
128 case "attached":
129 attachedTabs.add(id);
130 return PubSub.publish(changes, { key, attached: true });
131 case "detached": {
132 attachedTabs.delete(id);
133 const queue = inbound.get(id);
134 return (queue ? Queue.shutdown(queue) : Effect.void).pipe(
135 Effect.andThen(PubSub.publish(changes, { key, attached: false })),
136 );
137 }
138 }
139 }),
140 Effect.catchCause((cause) => Effect.logWarning("desktop browser channel stopped", { cause })),
141 Effect.forkScoped,
142 );
143
144 const endpoint = (key: DesktopTabKey) =>
145 Effect.gen(function* () {
146 const id = keyOf(key);
147 const secret = NodeCrypto.randomBytes(24).toString("base64url");
148 const server = yield* NodeSocketServer.makeWebSocket({
149 host: "127.0.0.1",
150 port: 0,
151 path: `/${secret}`,
152 }).pipe(Effect.orDie);
153 const queue = yield* Queue.unbounded<string>();
154 inbound.set(id, queue);
155 // A detach before this registration shut down no queue, so check again.
156 if (!attachedTabs.has(id)) {
157 inbound.delete(id);
158 return yield* Effect.die("The desktop tab detached before the server connected.");
159 }
160 yield* Effect.addFinalizer(() =>
161 Effect.gen(function* () {
162 if (inbound.get(id) === queue) inbound.delete(id);
163 yield* Queue.shutdown(queue);
164 yield* command({ type: "release", ...key });
165 }),
166 );
167 // The relay serves one Playwright connection; a second would see the first's sessions.
168 let connected = false;
169 yield* server
170 .run((socket) =>
171 Effect.gen(function* () {
172 if (connected) return;
173 connected = true;
174 const writer = yield* socket.writer;
175 const reader = yield* socket.reader;
176 yield* Stream.fromQueue(queue).pipe(
177 Stream.runForEach((message) => writer.write(message)),
178 Effect.forkScoped,
179 );
180 const decoder = new TextDecoder();
181 return yield* reader.pull.pipe(
182 Effect.flatMap((frames) =>
183 Effect.forEach(
184 frames,
185 (frame) =>
186 command({
187 type: "cdp",
188 ...key,
189 message: typeof frame === "string" ? frame : decoder.decode(frame),
190 }),
191 { discard: true },
192 ),
193 ),
194 Effect.forever,
195 );
196 }).pipe(Effect.scoped, Effect.ignore),
197 )
198 .pipe(Effect.forkScoped);
199 const address = server.address;
200 if (address._tag !== "InetAddressV4") return yield* Effect.die("Unexpected relay address.");
201 return `ws://127.0.0.1:${address.port}/${secret}`;
202 });
203
204 return DesktopBrowserChannel.of({
205 available: true,
206 awaitAttached: (key, timeout) =>
207 Effect.scoped(
208 Effect.gen(function* () {
209 const subscription = yield* PubSub.subscribe(changes);
210 if (attachedTabs.has(keyOf(key))) return true;
211 return yield* Stream.fromSubscription(subscription).pipe(
212 Stream.filter((change) => change.attached && keyOf(change.key) === keyOf(key)),
213 Stream.runHead,
214 Effect.map(Option.isSome),
215 Effect.timeoutOption(timeout),
216 Effect.map((result) => Option.getOrElse(result, () => false)),
217 );
218 }),
219 ),
220 detached: Stream.fromPubSub(changes).pipe(
221 Stream.filter((change) => !change.attached),
222 Stream.map((change) => change.key),
223 ),
224 attached: Stream.fromPubSub(changes).pipe(
225 Stream.filter((change) => change.attached),
226 Stream.map((change) => change.key),
227 ),
228 isAttached: (key) => Effect.sync(() => attachedTabs.has(keyOf(key))),
229 endpoint,
230 pointer: (key, pointer) => command({ type: "pointer", ...key, ...pointer }),
231 });
232});
233
234export const layer = Layer.effect(DesktopBrowserChannel, make);