1import * as NodeHttpServerRequest from "@effect/platform-node/NodeHttpServerRequest";
2import {
3 AuthOrchestrationReadScope,
4 AuthPreviewOperateScope,
5 PREVIEW_STREAM_HOST_SETUP_CLOSE_CODE,
6 PreviewStreamHostSetup,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Option from "effect/Option";
10import * as Queue from "effect/Queue";
11import * as Schema from "effect/Schema";
12import * as FileSystem from "effect/FileSystem";
13import * as Path from "effect/Path";
14import {
15 HttpPlatform,
16 HttpRouter,
17 HttpServerRequest,
18 HttpServerResponse,
19 Multipart,
20} from "effect/http";
21import * as Socket from "effect/socket/Socket";
22
23import { authenticateMediaRequest } from "../auth/http.ts";
24import { assetResponseHeaders } from "../http.ts";
25import * as PreviewBrowserHost from "./PreviewBrowserHost.ts";
26import * as ServerBrowser from "./ServerBrowser.ts";
27
28const PREVIEW_STREAM_ROUTE_PREFIX = "/api/preview-stream";
29/** Matches `PREVIEW_STREAM_TAB_GONE_CODE` in the client. */
30const TAB_GONE_CODE = 4404;
31
32const DEFAULT_QUALITY = 70;
33const MAX_SOCKET_BUFFER_BYTES = 8 * 1024 * 1024;
34const MAX_UNACKNOWLEDGED_FRAMES = 64;
35const textDecoder = new TextDecoder();
36
37const intParam = (params: URLSearchParams, name: string, fallback: number, max: number) => {
38 const value = Number(params.get(name));
39 return Number.isFinite(value) && value > 0 ? Math.min(Math.round(value), max) : fallback;
40};
41
42const isAck = (message: unknown) =>
43 typeof message === "object" && message !== null && "type" in message && message.type === "ack";
44
45const parseMessage = (chunk: Uint8Array | string): unknown => {
46 try {
47 return JSON.parse(typeof chunk === "string" ? chunk : textDecoder.decode(chunk));
48 } catch {
49 return null;
50 }
51};
52
53const makeHandler = (browser: ServerBrowser.ServerBrowser["Service"]) =>
54 Effect.gen(function* () {
55 const request = yield* HttpServerRequest.HttpServerRequest;
56 const url = HttpServerRequest.toURL(request);
57 if (Option.isNone(url)) return HttpServerResponse.text("Bad Request", { status: 400 });
58 if (url.value.pathname === `${PREVIEW_STREAM_ROUTE_PREFIX}/download`) {
59 return yield* serveDownload(browser, url.value.searchParams);
60 }
61 if (url.value.pathname === `${PREVIEW_STREAM_ROUTE_PREFIX}/upload`) {
62 return request.method === "POST"
63 ? yield* receiveUpload(browser, url.value.searchParams)
64 : HttpServerResponse.text("Method Not Allowed", { status: 405 });
65 }
66 if (
67 url.value.pathname !== `${PREVIEW_STREAM_ROUTE_PREFIX}/ws` ||
68 request.headers.upgrade?.toLowerCase() !== "websocket"
69 ) {
70 return HttpServerResponse.text("Not Found", { status: 404 });
71 }
72 const session = yield* authenticateMediaRequest(AuthOrchestrationReadScope);
73 const params = url.value.searchParams;
74 // Watching needs read access; driving the page is a preview mutation.
75 const canOperate =
76 session.scopes.includes(AuthPreviewOperateScope) && params.get("interactive") !== "false";
77 const threadId = params.get("threadId") ?? "";
78 const tabId = params.get("tabId") ?? "";
79 if (threadId.length === 0 || tabId.length === 0) {
80 return HttpServerResponse.text("Not Found", { status: 404 });
81 }
82 return yield* Effect.scoped(
83 Effect.gen(function* () {
84 const attached = yield* browser
85 .attachViewer({
86 threadId,
87 tabId,
88 canOperate,
89 maxWidth: intParam(params, "maxWidth", 1280, 7680),
90 maxHeight: intParam(params, "maxHeight", 800, 4320),
91 quality: intParam(params, "quality", DEFAULT_QUALITY, 100),
92 })
93 .pipe(
94 Effect.map((viewer) => ({ _tag: "attached" as const, viewer })),
95 Effect.catchTags({
96 ServerBrowserTabNotFoundError: () => Effect.succeed({ _tag: "gone" as const }),
97 ServerBrowserLaunchError: (error) => {
98 const setup = hostSetup(error.cause);
99 return setup === undefined
100 ? Effect.fail(error)
101 : Effect.succeed({ _tag: "hostSetup" as const, reason: encodeHostSetup(setup) });
102 },
103 }),
104 );
105 const incoming = NodeHttpServerRequest.toIncomingMessage(request);
106 // JPEGs are already compressed. Disabling deflate also keeps all
107 // pending writes in the socket buffer we bound below, not a zlib queue.
108 delete incoming.headers["sec-websocket-extensions"];
109 const transport = incoming.socket;
110 const socket = yield* request.upgrade;
111 // The reader performs the upgrade; writes wait for it.
112 const reader = yield* socket.reader;
113 const writer = yield* socket.writer;
114 // A refused upgrade reads as an auth failure to ticket clients, so a
115 // missing tab is a close code they stop on.
116 const gone = writer.write(new Socket.CloseEvent(TAB_GONE_CODE, "tab closed"));
117 if (attached._tag === "gone") {
118 yield* gone;
119 return HttpServerResponse.empty();
120 }
121 if (attached._tag === "hostSetup") {
122 yield* writer.write(
123 new Socket.CloseEvent(PREVIEW_STREAM_HOST_SETUP_CLOSE_CODE, attached.reason),
124 );
125 return HttpServerResponse.empty();
126 }
127 const viewer = attached.viewer;
128 // `write` returns once the frame is queued, so Chromium's ack for each
129 // frame waits for the viewer's `ack` message instead.
130 const unacknowledged: Array<Effect.Effect<void>> = [];
131 const disconnectSlowViewer = Effect.sync(() => transport.destroy()).pipe(
132 Effect.andThen(Effect.interrupt),
133 );
134 const write = (data: Uint8Array | string) =>
135 Effect.suspend(() => {
136 const bytes = typeof data === "string" ? Buffer.byteLength(data) : data.byteLength;
137 // Include the WebSocket frame header in the budget.
138 if (transport.writableLength + bytes + 14 > MAX_SOCKET_BUFFER_BYTES) {
139 return disconnectSlowViewer;
140 }
141 return writer.write(data);
142 });
143 const sendOutput = Queue.take(viewer.output).pipe(
144 Effect.flatMap((output) => {
145 switch (output._tag) {
146 case "frame":
147 if (unacknowledged.length >= MAX_UNACKNOWLEDGED_FRAMES) {
148 return disconnectSlowViewer;
149 }
150 return write(output.data).pipe(
151 Effect.andThen(Effect.sync(() => unacknowledged.push(output.ack))),
152 );
153 case "viewport":
154 case "control":
155 case "clipboard":
156 case "download":
157 case "fileChooser":
158 case "fileChooserClosed":
159 case "popup":
160 case "pointer":
161 case "probe": {
162 const { _tag: type, ...data } = output;
163 return write(JSON.stringify({ type, ...data }));
164 }
165 case "gone":
166 return gone.pipe(Effect.andThen(Effect.interrupt));
167 case "reconnect":
168 return writer
169 .write(new Socket.CloseEvent(1012, "reconnect"))
170 .pipe(Effect.andThen(Effect.interrupt));
171 }
172 }),
173 );
174 const receive = (chunk: Uint8Array | string) => {
175 const message = parseMessage(chunk);
176 // Ownership serializes actions in the service. Keep reading so dialog
177 // replies and takeover can unblock an action already waiting on the page.
178 if (!isAck(message))
179 return canOperate
180 ? viewer.input(message).pipe(Effect.forkScoped, Effect.asVoid)
181 : Effect.void;
182 const ack = unacknowledged.shift();
183 // Forked: Chromium acks are paced and must not hold up input.
184 return ack ? Effect.forkScoped(ack).pipe(Effect.asVoid) : Effect.void;
185 };
186 const receiveInput = reader.pull.pipe(
187 Effect.flatMap((chunks) => Effect.forEach(chunks, receive, { discard: true })),
188 );
189 return yield* Effect.raceFirst(Effect.forever(sendOutput), Effect.forever(receiveInput));
190 }),
191 ).pipe(
192 Effect.catchTags({
193 ServerBrowserLaunchError: (error) =>
194 Effect.logWarning("server preview browser failed to start", { cause: error.cause }).pipe(
195 Effect.as(HttpServerResponse.text("Service Unavailable", { status: 503 })),
196 ),
197 }),
198 // A dropped socket is a normal end of viewing.
199 Effect.catch(() => Effect.succeed(HttpServerResponse.empty())),
200 );
201 });
202
203/** Which host setup a launch failure is missing, and the command that fixes it. */
204const hostSetup = (cause: unknown): PreviewStreamHostSetup | undefined =>
205 isSandboxError(cause)
206 ? { need: "sandbox", command: cause.setupCommand }
207 : isLibrariesError(cause)
208 ? { need: "libraries", command: cause.setupCommand }
209 : undefined;
210const isSandboxError = Schema.is(PreviewBrowserHost.PreviewBrowserSandboxError);
211const isLibrariesError = Schema.is(PreviewBrowserHost.PreviewBrowserLibrariesError);
212const encodeHostSetup = Schema.encodeSync(Schema.fromJsonString(PreviewStreamHostSetup));
213
214/** `GET /api/preview-stream/download?threadId&tabId&id`: a file a server tab downloaded. */
215const serveDownload = (browser: ServerBrowser.ServerBrowser["Service"], params: URLSearchParams) =>
216 Effect.gen(function* () {
217 yield* authenticateMediaRequest(AuthOrchestrationReadScope);
218 const download = yield* browser.openDownload({
219 threadId: params.get("threadId") ?? "",
220 tabId: params.get("tabId") ?? "",
221 downloadId: params.get("id") ?? "",
222 });
223 if (Option.isNone(download)) return HttpServerResponse.text("Not Found", { status: 404 });
224 return yield* HttpServerResponse.file(download.value.path, {
225 headers: assetResponseHeaders(download.value.path, {
226 download: true,
227 fileName: download.value.fileName,
228 }),
229 });
230 });
231
232const UPLOAD_MAX_FILE_BYTES = 100 * 1024 * 1024;
233const UPLOAD_MAX_FILES = 20;
234
235/**
236 * `POST /api/preview-stream/upload?threadId&tabId&chooser` with multipart `file`
237 * parts: answers the page's open file picker. No parts cancels it.
238 */
239const receiveUpload = (browser: ServerBrowser.ServerBrowser["Service"], params: URLSearchParams) =>
240 Effect.gen(function* () {
241 yield* authenticateMediaRequest(AuthPreviewOperateScope);
242 const request = yield* HttpServerRequest.HttpServerRequest;
243 const fs = yield* FileSystem.FileSystem;
244 const parts = yield* request.multipart.pipe(
245 Effect.provideContext(
246 Multipart.limitsServices({
247 maxParts: UPLOAD_MAX_FILES,
248 maxFileSize: UPLOAD_MAX_FILE_BYTES,
249 maxTotalSize: UPLOAD_MAX_FILE_BYTES,
250 }),
251 ),
252 );
253 const files = yield* Effect.forEach(
254 Object.values(parts).flatMap((value) =>
255 Array.isArray(value) ? value.filter(Multipart.isPersistedFile) : [],
256 ),
257 (file) =>
258 fs.readFile(file.path).pipe(
259 Effect.map((bytes) => ({
260 name: file.name,
261 mimeType: file.contentType,
262 buffer: Buffer.from(bytes),
263 })),
264 ),
265 );
266 const answered = yield* browser.answerFileChooser({
267 threadId: params.get("threadId") ?? "",
268 tabId: params.get("tabId") ?? "",
269 chooserId: params.get("chooser") ?? "",
270 files,
271 });
272 return answered
273 ? HttpServerResponse.empty({ status: 204 })
274 : HttpServerResponse.text("The page's file picker is no longer open.", { status: 409 });
275 }).pipe(Effect.scoped);
276
277// Capture the browser because handlers only see request-scoped services.
278export const routeLayer = HttpRouter.use((router) =>
279 Effect.gen(function* () {
280 const browser = yield* ServerBrowser.ServerBrowser;
281 const platform = yield* Effect.context<
282 HttpPlatform.HttpPlatform | FileSystem.FileSystem | Path.Path
283 >();
284 const handler = makeHandler(browser).pipe(Effect.provideContext(platform));
285 yield* router.add("GET", `${PREVIEW_STREAM_ROUTE_PREFIX}/*`, handler);
286 yield* router.add("POST", `${PREVIEW_STREAM_ROUTE_PREFIX}/upload`, handler);
287 }),
288);