1// @effect-diagnostics nodeBuiltinImport:off - Bridges Playwright's WebSocket transport to the desktop's fds.
2/** Playwright needs a CDP WebSocket; the secret loopback endpoint relays desktop debugger frames over bootstrap fds. */
3import * as NodeStream from "@effect/platform-node/NodeStream";
4import * as NodeSocketServer from "@effect/platform-node/NodeSocketServer";
5import {
6 DesktopBrowserCommand,
7 DesktopBrowserEvent,
8 type DesktopBrowserCommand as DesktopBrowserCommandType,
9 type PreviewAutomationLoadError,
10} from "@t3tools/contracts";
11import * as Context from "effect/Context";
12import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as PubSub from "effect/PubSub";
17import * as Queue from "effect/Queue";
18import * as Schema from "effect/Schema";
19import * as Scope from "effect/Scope";
20import * as Semaphore from "effect/Semaphore";
21import * as Stream from "effect/Stream";
22import * as Ndjson from "effect/encoding/Ndjson";
23import * as NodeCrypto from "node:crypto";
24import * as NodeNet from "node:net";
25
26import * as ServerConfig from "../config.ts";
27import { writeAllToFileDescriptor } from "../resourceTelemetry/DesktopTelemetryReceiver.ts";
28
29const decodeEvent = Schema.decodeUnknownEffect(DesktopBrowserEvent);
30const encodeCommand = Schema.encodeEffect(Schema.fromJsonString(DesktopBrowserCommand));
31
32export interface DesktopTabKey {
33 readonly threadId: string;
34 readonly tabId: string;
35}
36
37const keyOf = ({ threadId, tabId }: DesktopTabKey) => `${threadId}\u0000${tabId}`;
38
39export class DesktopBrowserChannel extends Context.Service<
40 DesktopBrowserChannel,
41 {
42 readonly available: boolean;
43 /** Subscribe before checking so concurrent attaches cannot be missed. */
44 readonly awaitAttached: (key: DesktopTabKey, timeout: Duration.Input) => Effect.Effect<boolean>;
45 readonly detached: Stream.Stream<DesktopTabKey>;
46 readonly attached: Stream.Stream<DesktopTabKey>;
47 readonly isAttached: (key: DesktopTabKey) => Effect.Effect<boolean>;
48 /** Whether a desktop window draws the tab on screen, which no streamed viewer counts. */
49 readonly isShown: (key: DesktopTabKey) => boolean;
50 /** A load failure the desktop saw, possibly before the server connected to the tab. */
51 readonly loadFailure: (key: DesktopTabKey) => PreviewAutomationLoadError | undefined;
52 readonly endpoint: (key: DesktopTabKey) => Effect.Effect<string, never, Scope.Scope>;
53 readonly pointer: (
54 key: DesktopTabKey,
55 pointer: { readonly phase: "move" | "click"; readonly x: number; readonly y: number },
56 ) => Effect.Effect<void>;
57 /** Never fails. A failed write means a broken channel; the next backend starts with no turns. */
58 readonly turn: (threadId: string, active: boolean) => Effect.Effect<void>;
59 }
60>()("t3/preview/DesktopBrowserChannel") {}
61
62const make = Effect.gen(function* () {
63 const config = yield* ServerConfig.ServerConfig;
64 const inputFd = config.desktopBrowserFd;
65 const controlFd = config.desktopBrowserControlFd;
66 const changes = yield* PubSub.unbounded<{ key: DesktopTabKey; attached: boolean }>();
67 const attachedTabs = new Set<string>();
68 // Missing reports protect desktop tabs until the renderer explicitly marks them hidden.
69 const shownTabs = new Map<string, boolean>();
70 const loadFailures = new Map<string, PreviewAutomationLoadError>();
71 const inbound = new Map<string, Queue.Queue<string>>();
72 const writeLock = yield* Semaphore.make(1);
73
74 if (inputFd === undefined || controlFd === undefined) {
75 return DesktopBrowserChannel.of({
76 available: false,
77 awaitAttached: Effect.fn("DesktopBrowserChannel.awaitAttached")(function () {
78 return Effect.succeed(false);
79 }),
80 detached: Stream.empty,
81 attached: Stream.empty,
82 isAttached: Effect.fn("DesktopBrowserChannel.isAttached")(function () {
83 return Effect.succeed(false);
84 }),
85 endpoint: Effect.fn("DesktopBrowserChannel.endpoint")(function () {
86 return Effect.die("No desktop app is attached to this server.");
87 }),
88 turn: Effect.fn("DesktopBrowserChannel.turn")(function () {
89 return Effect.void;
90 }),
91 pointer: Effect.fn("DesktopBrowserChannel.pointer")(function () {
92 return Effect.void;
93 }),
94 isShown: () => false,
95 loadFailure: () => undefined,
96 });
97 }
98
99 const command = (message: DesktopBrowserCommandType) =>
100 writeLock.withPermits(1)(
101 encodeCommand(message).pipe(
102 Effect.flatMap((line) => writeAllToFileDescriptor(controlFd, Buffer.from(`${line}\n`))),
103 Effect.catchCause((cause) =>
104 Effect.logWarning("desktop browser command failed", { cause }),
105 ),
106 ),
107 );
108
109 // Use cancellable socket reads; filesystem pipe reads keep Node alive after scope shutdown.
110 const readable = yield* Effect.acquireRelease(
111 Effect.sync(() => new NodeNet.Socket({ fd: inputFd, readable: true, writable: false })),
112 (stream) => Effect.sync(() => stream.destroy()),
113 );
114 yield* NodeStream.fromReadable<Uint8Array, Error>({
115 evaluate: () => readable,
116 closeOnDone: true,
117 onError: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
118 }).pipe(
119 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
120 Stream.mapEffect((value) => decodeEvent(value).pipe(Effect.option)),
121 Stream.runForEach((decoded) => {
122 if (Option.isNone(decoded)) return Effect.void;
123 const event = decoded.value;
124 const key = { threadId: event.threadId, tabId: event.tabId };
125 const id = keyOf(key);
126 switch (event.type) {
127 case "cdp": {
128 const queue = inbound.get(id);
129 return queue ? Queue.offer(queue, event.message).pipe(Effect.asVoid) : Effect.void;
130 }
131 case "attached":
132 attachedTabs.add(id);
133 return PubSub.publish(changes, { key, attached: true });
134 case "loadFailed":
135 loadFailures.set(id, {
136 url: event.url.slice(0, 2048),
137 code: event.code,
138 description: event.description,
139 });
140 return Effect.void;
141 case "detached": {
142 attachedTabs.delete(id);
143 shownTabs.delete(id);
144 loadFailures.delete(id);
145 const queue = inbound.get(id);
146 return (queue ? Queue.shutdown(queue) : Effect.void).pipe(
147 Effect.andThen(PubSub.publish(changes, { key, attached: false })),
148 );
149 }
150 case "shown":
151 shownTabs.set(id, event.shown);
152 return Effect.void;
153 }
154 }),
155 Effect.catchCause((cause) => Effect.logWarning("desktop browser channel stopped", { cause })),
156 Effect.ensuring(
157 Effect.sync(() => {
158 shownTabs.clear();
159 loadFailures.clear();
160 }),
161 ),
162 Effect.forkScoped,
163 );
164
165 const endpoint = Effect.fn("DesktopBrowserChannel.endpoint")(function* (key: DesktopTabKey) {
166 const id = keyOf(key);
167 const secret = NodeCrypto.randomBytes(24).toString("base64url");
168 const server = yield* NodeSocketServer.makeWebSocket({
169 host: "127.0.0.1",
170 port: 0,
171 path: `/${secret}`,
172 }).pipe(Effect.orDie);
173 const queue = yield* Queue.unbounded<string>();
174 inbound.set(id, queue);
175 // A detach before this registration shut down no queue, so check again.
176 if (!attachedTabs.has(id)) {
177 inbound.delete(id);
178 return yield* Effect.die("The desktop tab detached before the server connected.");
179 }
180 yield* Effect.addFinalizer(() =>
181 Effect.gen(function* () {
182 if (inbound.get(id) === queue) inbound.delete(id);
183 yield* Queue.shutdown(queue);
184 yield* command({ type: "release", ...key });
185 }),
186 );
187 // The relay serves one Playwright connection; a second would see the first's sessions.
188 let connected = false;
189 yield* server
190 .run((socket) =>
191 Effect.gen(function* () {
192 if (connected) return;
193 connected = true;
194 const writer = yield* socket.writer;
195 const reader = yield* socket.reader;
196 // A detach shuts the queue down. Ending here closes the socket, so Playwright's
197 // pending commands fail instead of waiting forever for the tab to answer.
198 const toPlaywright = Stream.fromQueue(queue).pipe(
199 Stream.runForEach((message) => writer.write(message)),
200 Effect.exit,
201 );
202 const decoder = new TextDecoder();
203 const toDesktop = reader.pull.pipe(
204 Effect.flatMap((frames) =>
205 Effect.forEach(
206 frames,
207 (frame) =>
208 command({
209 type: "cdp",
210 ...key,
211 message: typeof frame === "string" ? frame : decoder.decode(frame),
212 }),
213 { discard: true },
214 ),
215 ),
216 Effect.forever,
217 );
218 return yield* Effect.raceFirst(toPlaywright, toDesktop);
219 }).pipe(Effect.scoped, Effect.ignore),
220 )
221 .pipe(Effect.forkScoped);
222 const address = server.address;
223 if (address._tag !== "InetAddressV4") return yield* Effect.die("Unexpected relay address.");
224 return `ws://127.0.0.1:${address.port}/${secret}`;
225 });
226
227 return DesktopBrowserChannel.of({
228 available: true,
229 awaitAttached: Effect.fn("DesktopBrowserChannel.awaitAttached")(function (key, timeout) {
230 return Effect.scoped(
231 Effect.gen(function* () {
232 const subscription = yield* PubSub.subscribe(changes);
233 if (attachedTabs.has(keyOf(key))) return true;
234 return yield* Stream.fromSubscription(subscription).pipe(
235 Stream.filter((change) => change.attached && keyOf(change.key) === keyOf(key)),
236 Stream.runHead,
237 Effect.map(Option.isSome),
238 Effect.timeoutOption(timeout),
239 Effect.map((result) => Option.getOrElse(result, () => false)),
240 );
241 }),
242 );
243 }),
244 detached: Stream.fromPubSub(changes).pipe(
245 Stream.filter((change) => !change.attached),
246 Stream.map((change) => change.key),
247 ),
248 attached: Stream.fromPubSub(changes).pipe(
249 Stream.filter((change) => change.attached),
250 Stream.map((change) => change.key),
251 ),
252 isAttached: Effect.fn("DesktopBrowserChannel.isAttached")(function (key) {
253 return Effect.sync(() => attachedTabs.has(keyOf(key)));
254 }),
255 isShown: (key) => shownTabs.get(keyOf(key)) ?? true,
256 loadFailure: (key) => loadFailures.get(keyOf(key)),
257 endpoint,
258 turn: Effect.fn("DesktopBrowserChannel.turn")(function (threadId, active) {
259 return command({ type: "turn", threadId, active });
260 }),
261 pointer: Effect.fn("DesktopBrowserChannel.pointer")(function (key, pointer) {
262 return command({ type: "pointer", ...key, ...pointer });
263 }),
264 });
265});
266
267export const layer = Layer.effect(DesktopBrowserChannel, make);