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";
27import { requestUrl } from "../requestUrl.ts";
28
29const PREVIEW_STREAM_ROUTE_PREFIX = "/api/preview-stream";
30const TAB_GONE_CODE = 4404;
31
32const DEFAULT_QUALITY = 70;
33const MAX_SOCKET_BUFFER_BYTES = 8 * 1024 * 1024;
34const textDecoder = new TextDecoder();
35
36const intParam = (params: URLSearchParams, name: string, fallback: number, max: number) => {
37 const value = Number(params.get(name));
38 return Number.isFinite(value) && value > 0 ? Math.min(Math.round(value), max) : fallback;
39};
40
41type Frame = Extract<ServerBrowser.ServerBrowserViewerOutput, { readonly _tag: "frame" }>;
42
43const isAck = (message: unknown) =>
44 typeof message === "object" && message !== null && "type" in message && message.type === "ack";
45
46const parseMessage = (chunk: Uint8Array | string): unknown => {
47 try {
48 return JSON.parse(typeof chunk === "string" ? chunk : textDecoder.decode(chunk));
49 } catch {
50 return null;
51 }
52};
53
54// Request middleware owns tracing, including long-lived WebSocket sessions.
55const makeHandler = Effect.fnUntraced(function* (browser: ServerBrowser.ServerBrowser["Service"]) {
56 const request = yield* HttpServerRequest.HttpServerRequest;
57 const url = requestUrl(request);
58 if (Option.isNone(url)) return HttpServerResponse.text("Bad Request", { status: 400 });
59 if (url.value.pathname === `${PREVIEW_STREAM_ROUTE_PREFIX}/download`) {
60 return yield* serveDownload(browser, url.value.searchParams);
61 }
62 if (url.value.pathname === `${PREVIEW_STREAM_ROUTE_PREFIX}/upload`) {
63 return request.method === "POST"
64 ? yield* receiveUpload(browser, url.value.searchParams)
65 : HttpServerResponse.text("Method Not Allowed", { status: 405 });
66 }
67 if (
68 url.value.pathname !== `${PREVIEW_STREAM_ROUTE_PREFIX}/ws` ||
69 request.headers.upgrade?.toLowerCase() !== "websocket"
70 ) {
71 return HttpServerResponse.text("Not Found", { status: 404 });
72 }
73 const session = yield* authenticateMediaRequest(AuthOrchestrationReadScope);
74 const params = url.value.searchParams;
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 const reader = yield* socket.reader;
112 const writer = yield* socket.writer;
113 // A refused upgrade reads as an auth failure to ticket clients, so a
114 // missing tab is a close code they stop on.
115 const gone = writer.write(new Socket.CloseEvent(TAB_GONE_CODE, "tab closed"));
116 if (attached._tag === "gone") {
117 yield* gone;
118 return HttpServerResponse.empty();
119 }
120 if (attached._tag === "hostSetup") {
121 yield* writer.write(
122 new Socket.CloseEvent(PREVIEW_STREAM_HOST_SETUP_CLOSE_CODE, attached.reason),
123 );
124 return HttpServerResponse.empty();
125 }
126 const viewer = attached.viewer;
127 // `write` returns once the frame is queued, so Chromium's ack for each
128 // frame waits for the viewer's `ack` message instead. Only one frame
129 // is on the wire and only the newest waits behind it, so replies never
130 // queue behind stale images on a slow link.
131 let inFlight: Effect.Effect<void> | null = null;
132 let pending: Frame | null = null;
133 const disconnectSlowViewer = Effect.sync(() => transport.destroy()).pipe(
134 Effect.andThen(Effect.interrupt),
135 );
136 const write = (data: Uint8Array | string) =>
137 Effect.suspend(() => {
138 const bytes = typeof data === "string" ? Buffer.byteLength(data) : data.byteLength;
139 if (transport.writableLength + bytes + 14 > MAX_SOCKET_BUFFER_BYTES) {
140 return disconnectSlowViewer;
141 }
142 return writer.write(data);
143 });
144 const sendFrame = (frame: Frame) => {
145 inFlight = frame.ack;
146 return write(frame.data);
147 };
148 const sendOutput = Queue.take(viewer.output).pipe(
149 Effect.flatMap((output) => {
150 switch (output._tag) {
151 case "frame": {
152 if (inFlight === null) return sendFrame(output);
153 const superseded = pending;
154 pending = output;
155 // Never sent, but Chromium still waits for its ack.
156 return superseded ? Effect.forkScoped(superseded.ack) : Effect.void;
157 }
158 case "viewport":
159 case "control":
160 case "clipboard":
161 case "download":
162 case "fileChooser":
163 case "fileChooserClosed":
164 case "popup":
165 case "pointer":
166 case "probe": {
167 const { _tag: type, ...data } = output;
168 return write(JSON.stringify({ type, ...data }));
169 }
170 case "gone":
171 return gone.pipe(Effect.andThen(Effect.interrupt));
172 case "reconnect":
173 return writer
174 .write(new Socket.CloseEvent(1012, "reconnect"))
175 .pipe(Effect.andThen(Effect.interrupt));
176 }
177 }),
178 );
179 // Incoming messages share the session span; frame acknowledgements are frequent.
180 const receive = Effect.fnUntraced(function* (chunk: Uint8Array | string) {
181 const message = parseMessage(chunk);
182 // Ownership serializes actions in the service. Keep reading so dialog
183 // replies and takeover can unblock an action already waiting on the page.
184 if (!isAck(message))
185 return yield* canOperate
186 ? viewer.input(message).pipe(Effect.forkScoped, Effect.asVoid)
187 : Effect.void;
188 const ack = inFlight;
189 const next = pending;
190 inFlight = null;
191 pending = null;
192 // Forked: Chromium acks are paced and must not hold up input.
193 return yield* Effect.andThen(
194 ack ? Effect.forkScoped(ack) : Effect.void,
195 next ? sendFrame(next) : Effect.void,
196 );
197 });
198 const receiveInput = reader.pull.pipe(
199 Effect.flatMap((chunks) => Effect.forEach(chunks, receive, { discard: true })),
200 );
201 return yield* Effect.raceFirst(Effect.forever(sendOutput), Effect.forever(receiveInput));
202 }),
203 ).pipe(
204 Effect.catchTags({
205 ServerBrowserLaunchError: (error) =>
206 Effect.logWarning("server preview browser failed to start", { cause: error.cause }).pipe(
207 Effect.as(HttpServerResponse.text("Service Unavailable", { status: 503 })),
208 ),
209 }),
210 Effect.orElseSucceed(() => HttpServerResponse.empty()),
211 );
212});
213
214const hostSetup = (cause: unknown): PreviewStreamHostSetup | undefined =>
215 isSandboxError(cause)
216 ? { need: "sandbox", command: cause.setupCommand }
217 : isLibrariesError(cause)
218 ? { need: "libraries", command: cause.setupCommand }
219 : undefined;
220const isSandboxError = Schema.is(PreviewBrowserHost.PreviewBrowserSandboxError);
221const isLibrariesError = Schema.is(PreviewBrowserHost.PreviewBrowserLibrariesError);
222const encodeHostSetup = Schema.encodeSync(Schema.fromJsonString(PreviewStreamHostSetup));
223
224// The request span already covers this download route.
225const serveDownload = Effect.fnUntraced(function* (
226 browser: ServerBrowser.ServerBrowser["Service"],
227 params: URLSearchParams,
228) {
229 yield* authenticateMediaRequest(AuthOrchestrationReadScope);
230 const download = yield* browser.openDownload({
231 threadId: params.get("threadId") ?? "",
232 tabId: params.get("tabId") ?? "",
233 downloadId: params.get("id") ?? "",
234 });
235 if (Option.isNone(download)) return HttpServerResponse.text("Not Found", { status: 404 });
236 return yield* HttpServerResponse.file(download.value.path, {
237 headers: assetResponseHeaders(download.value.path, {
238 download: true,
239 fileName: download.value.fileName,
240 }),
241 });
242});
243
244const UPLOAD_MAX_FILE_BYTES = 100 * 1024 * 1024;
245const UPLOAD_MAX_FILES = 20;
246
247// The request span already covers this upload route.
248const receiveUpload = Effect.fnUntraced(
249 function* (browser: ServerBrowser.ServerBrowser["Service"], params: URLSearchParams) {
250 yield* authenticateMediaRequest(AuthPreviewOperateScope);
251 const request = yield* HttpServerRequest.HttpServerRequest;
252 const fs = yield* FileSystem.FileSystem;
253 const parts = yield* request.multipart.pipe(
254 Effect.provideContext(
255 Multipart.limitsServices({
256 maxParts: UPLOAD_MAX_FILES,
257 maxFileSize: UPLOAD_MAX_FILE_BYTES,
258 maxTotalSize: UPLOAD_MAX_FILE_BYTES,
259 }),
260 ),
261 );
262 const files = yield* Effect.forEach(
263 Object.values(parts).flatMap((value) =>
264 Array.isArray(value) ? value.filter(Multipart.isPersistedFile) : [],
265 ),
266 (file) =>
267 fs.readFile(file.path).pipe(
268 Effect.map((bytes) => ({
269 name: file.name,
270 mimeType: file.contentType,
271 buffer: Buffer.from(bytes),
272 })),
273 ),
274 );
275 const answered = yield* browser.answerFileChooser({
276 threadId: params.get("threadId") ?? "",
277 tabId: params.get("tabId") ?? "",
278 chooserId: params.get("chooser") ?? "",
279 files,
280 });
281 return answered
282 ? HttpServerResponse.empty({ status: 204 })
283 : HttpServerResponse.text("The page's file picker is no longer open.", { status: 409 });
284 },
285 (effect, _browser, _params) => effect.pipe(Effect.scoped),
286);
287
288export const routeLayer = HttpRouter.use((router) =>
289 Effect.gen(function* () {
290 const browser = yield* ServerBrowser.ServerBrowser;
291 const platform = yield* Effect.context<
292 HttpPlatform.HttpPlatform | FileSystem.FileSystem | Path.Path
293 >();
294 const handler = makeHandler(browser).pipe(Effect.provideContext(platform));
295 yield* router.add("GET", `${PREVIEW_STREAM_ROUTE_PREFIX}/*`, handler);
296 yield* router.add("POST", `${PREVIEW_STREAM_ROUTE_PREFIX}/upload`, handler);
297 }),
298);