1import {
2 parseDeploymentLogRecord,
3 tailDeploymentLogs,
4} from "@/Prisma/PrismaLogs";
5import { Credentials } from "@/Prisma/Credentials";
6import { createServer, type Server, type Socket } from "node:net";
7import { describe, expect, it } from "alchemy-test";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Redacted from "effect/Redacted";
12import * as Stream from "effect/Stream";
13import { TestClock } from "effect/testing";
14import { WebSocketServer } from "ws";
15
16describe(
17 "Prisma deployment logs",
18 { tags: ["unit", "provider:prisma", "local"] },
19 () => {
20 it.effect("decodes compute log records into Alchemy log lines", () =>
21 Effect.gen(function* () {
22 const timestamp = new Date("2026-01-01T00:00:00.000Z");
23 const record = yield* parseDeploymentLogRecord(
24 JSON.stringify({
25 type: "log",
26 text: "server started",
27 byteStart: 0,
28 byteEnd: 15,
29 }),
30 timestamp,
31 );
32
33 expect(record).toMatchObject({
34 _tag: "log",
35 line: { timestamp, message: "server started" },
36 raw: { byteStart: 0, byteEnd: 15 },
37 });
38 }),
39 );
40
41 it.effect("decodes terminal records", () =>
42 Effect.gen(function* () {
43 const record = yield* parseDeploymentLogRecord(
44 JSON.stringify({
45 type: "terminal",
46 kind: "end",
47 code: "segment_time_limit",
48 message: "segment ended",
49 retryable: true,
50 cursor: "42",
51 }),
52 new Date("2026-01-01T00:00:00.000Z"),
53 );
54
55 expect(record).toEqual({
56 _tag: "terminal",
57 raw: {
58 type: "terminal",
59 kind: "end",
60 code: "segment_time_limit",
61 message: "segment ended",
62 retryable: true,
63 cursor: "42",
64 },
65 });
66 }),
67 );
68
69 it.effect("fails malformed records", () =>
70 Effect.gen(function* () {
71 const secret = "postgres://admin:do-not-retain@db.example.test/main";
72 const error = yield* parseDeploymentLogRecord(
73 JSON.stringify({ type: "wat", payload: secret }),
74 new Date("2026-01-01T00:00:00.000Z"),
75 ).pipe(Effect.flip);
76
77 expect(error).toBeInstanceOf(Error);
78 expect(error.message).toContain("type: unknown");
79 expect(String(error)).not.toContain(secret);
80 expect(String(error.cause)).not.toContain(secret);
81 expect(JSON.stringify(error)).not.toContain(secret);
82 }),
83 );
84
85 it.effect("rejects non-integer deployment log byte offsets", () =>
86 parseDeploymentLogRecord(
87 JSON.stringify({
88 type: "log",
89 text: "partial",
90 byteStart: 0.5,
91 byteEnd: 7,
92 }),
93 new Date("2026-01-01T00:00:00.000Z"),
94 ).pipe(
95 Effect.flip,
96 Effect.map((error) => {
97 expect(error.message).toContain("type: log");
98 }),
99 ),
100 );
101
102 it.effect("streams log lines from Prisma's WebSocket endpoint", () =>
103 withWebSocketServer((server) =>
104 Effect.gen(function* () {
105 const url = yield* listenUrl(server);
106 let authorization: string | undefined;
107
108 server.on("connection", (socket, request) => {
109 authorization = request.headers.authorization;
110 socket.send(
111 JSON.stringify({
112 type: "log",
113 text: "first",
114 byteStart: 0,
115 byteEnd: 5,
116 }),
117 );
118 socket.send(
119 JSON.stringify({
120 type: "log",
121 text: "second",
122 byteStart: 6,
123 byteEnd: 12,
124 }),
125 );
126 socket.send(
127 JSON.stringify({
128 type: "terminal",
129 kind: "end",
130 code: "vm_stopped",
131 message: "done",
132 retryable: false,
133 cursor: null,
134 }),
135 );
136 });
137
138 const lines = yield* tailWith(url, "deployment-1", {
139 tail: 2,
140 }).pipe(Stream.runCollect);
141
142 expect(lines.map((line) => line.message)).toEqual([
143 "first",
144 "second",
145 ]);
146 expect(authorization).toBe("Bearer test-token");
147 }),
148 ),
149 );
150
151 it.effect("rejects deployment log frames larger than 1 MiB", () =>
152 withWebSocketServer((server) =>
153 Effect.gen(function* () {
154 const url = yield* listenUrl(server);
155 const marker = "oversized-secret-marker";
156 server.on("connection", (socket) => {
157 socket.send(marker.repeat(60_000));
158 });
159
160 const error = yield* tailWith(url, "deployment-1").pipe(
161 Stream.runCollect,
162 Effect.flip,
163 );
164
165 expect(String(error)).toContain(
166 "exceeds the 1048576-byte frame limit",
167 );
168 expect(String(error)).not.toContain(marker);
169 expect(JSON.stringify(error)).not.toContain(marker);
170 }),
171 ),
172 );
173
174 it.effect(
175 "fails instead of buffering unbounded logs for a slow consumer",
176 () =>
177 withWebSocketServer((server) =>
178 Effect.gen(function* () {
179 const url = yield* listenUrl(server);
180 server.on("connection", (socket) => {
181 let byteOffset = 0;
182 for (let index = 0; index < 1_000; index++) {
183 const text = `line-${index}`;
184 socket.send(logRecord(text, byteOffset));
185 byteOffset += text.length;
186 }
187 socket.send(
188 terminalRecord({
189 kind: "end",
190 code: "vm_stopped",
191 retryable: false,
192 cursor: null,
193 }),
194 );
195 });
196
197 const error = yield* tailWith(url, "deployment-1").pipe(
198 Stream.runForEach(() => Effect.yieldNow),
199 Effect.flip,
200 );
201
202 expect(String(error)).toContain("consumer fell behind");
203 expect(String(error)).toContain("64-record safety buffer");
204 }),
205 ),
206 );
207
208 it.effect("reconnects from retryable terminal cursors exactly once", () =>
209 withWebSocketServer((server) =>
210 Effect.gen(function* () {
211 const url = yield* listenUrl(server);
212 const queries: Array<{ cursor?: string }> = [];
213 let connections = 0;
214
215 server.on("connection", (socket, request) => {
216 connections += 1;
217 const cursor = new URL(request.url ?? "/", url).searchParams.get(
218 "cursor",
219 );
220 queries.push(cursor === null ? {} : { cursor });
221 if (connections === 1) {
222 socket.send(logRecord("first"));
223 const terminal = terminalRecord({
224 kind: "end",
225 code: "segment_time_limit",
226 retryable: true,
227 cursor: "cursor-1",
228 });
229 socket.send(terminal);
230 // A duplicate frame from the old segment must not open another
231 // reconnect in parallel.
232 socket.send(terminal);
233 return;
234 }
235 socket.send(logRecord("second", 5));
236 socket.send(
237 terminalRecord({
238 kind: "end",
239 code: "vm_stopped",
240 retryable: false,
241 cursor: null,
242 }),
243 );
244 });
245
246 const lines = yield* tailWith(url, "deployment-1").pipe(
247 Stream.runCollect,
248 );
249
250 expect(lines.map((line) => line.message)).toEqual([
251 "first",
252 "second",
253 ]);
254 expect(connections).toBe(2);
255 expect(queries).toEqual([{}, { cursor: "cursor-1" }]);
256 }),
257 ),
258 );
259
260 it.effect("fails non-retryable terminal errors", () =>
261 withWebSocketServer((server) =>
262 Effect.gen(function* () {
263 const url = yield* listenUrl(server);
264 server.on("connection", (socket) => {
265 socket.send(
266 terminalRecord({
267 kind: "error",
268 code: "permission_denied",
269 message: "logs are unavailable",
270 retryable: false,
271 cursor: null,
272 }),
273 );
274 });
275
276 const error = yield* tailWith(url, "deployment-1").pipe(
277 Stream.runCollect,
278 Effect.flip,
279 );
280
281 expect(error).toBeInstanceOf(Error);
282 expect(String(error)).toContain("permission_denied");
283 expect(String(error)).toContain("logs are unavailable");
284 }),
285 ),
286 );
287
288 it.effect(
289 "reconnects a transient close from the latest confirmed byte",
290 () =>
291 withWebSocketServer((server) =>
292 Effect.gen(function* () {
293 const url = yield* listenUrl(server);
294 let connections = 0;
295 server.on("connection", (socket) => {
296 connections += 1;
297 if (connections === 1) {
298 socket.send(logRecord("partial"));
299 socket.close(1000, "unexpected end");
300 return;
301 }
302 socket.send(
303 terminalRecord({
304 kind: "end",
305 code: "vm_stopped",
306 retryable: false,
307 cursor: null,
308 }),
309 );
310 });
311
312 const lines = yield* tailWith(url, "deployment-1").pipe(
313 Stream.runCollect,
314 );
315
316 expect(lines.map((line) => line.message)).toEqual(["partial"]);
317 expect(connections).toBe(2);
318 }),
319 ),
320 );
321
322 it.effect(
323 "reconnects a retryable null-cursor terminal from the last byte",
324 () =>
325 withWebSocketServer((server) =>
326 Effect.gen(function* () {
327 const url = yield* listenUrl(server);
328 const cursors: Array<string | null> = [];
329 server.on("connection", (socket, request) => {
330 cursors.push(
331 new URL(request.url ?? "/", url).searchParams.get("cursor"),
332 );
333 if (cursors.length === 1) {
334 socket.send(logRecord("first"));
335 socket.send(
336 terminalRecord({
337 kind: "error",
338 code: "upstream_error",
339 retryable: true,
340 cursor: null,
341 }),
342 );
343 return;
344 }
345 socket.send(
346 terminalRecord({
347 kind: "end",
348 code: "vm_stopped",
349 retryable: false,
350 cursor: null,
351 }),
352 );
353 });
354
355 const lines = yield* tailWith(url, "deployment-1").pipe(
356 Stream.runCollect,
357 );
358
359 expect(lines.map((line) => line.message)).toEqual(["first"]);
360 expect(cursors).toEqual([null, "5"]);
361 }),
362 ),
363 );
364
365 it.effect(
366 "fails retry loops that alternate cursors without byte progress",
367 () =>
368 withWebSocketServer((server) =>
369 Effect.gen(function* () {
370 const url = yield* listenUrl(server);
371 let connections = 0;
372 server.on("connection", (socket) => {
373 connections += 1;
374 socket.send(
375 terminalRecord({
376 kind: "end",
377 code: "segment_time_limit",
378 retryable: true,
379 cursor: connections % 2 === 0 ? "cursor-2" : "cursor-1",
380 }),
381 );
382 });
383
384 const error = yield* tailWith(url, "deployment-1").pipe(
385 Stream.runCollect,
386 Effect.flip,
387 );
388
389 expect(connections).toBe(4);
390 expect(String(error)).toContain(
391 "made no progress after 3 reconnect",
392 );
393 }),
394 ),
395 );
396
397 it.effect("times out a WebSocket that never completes its handshake", () =>
398 withSilentTcpServer((url, connected) =>
399 Effect.gen(function* () {
400 const fiber = yield* tailWith(url, "deployment-1").pipe(
401 Stream.runCollect,
402 Effect.flip,
403 Effect.forkChild({ startImmediately: true }),
404 );
405
406 yield* connected;
407 yield* Effect.yieldNow;
408 yield* TestClock.adjust("11 seconds");
409 const error = yield* Fiber.join(fiber);
410
411 expect(error).toBeInstanceOf(Error);
412 expect(String(error)).toContain(
413 "WebSocket handshake timed out after 10 seconds",
414 );
415 }).pipe(Effect.provide(TestClock.layer())),
416 ),
417 );
418
419 it.effect("interrupts cleanly while a WebSocket is still connecting", () =>
420 withSilentTcpServer((url, connected) =>
421 Effect.gen(function* () {
422 const fiber = yield* tailWith(url, "deployment-1").pipe(
423 Stream.runDrain,
424 Effect.forkChild({ startImmediately: true }),
425 );
426
427 yield* connected;
428 yield* Fiber.interrupt(fiber);
429 const exit = fiber.pollUnsafe();
430
431 expect(exit?._tag).toBe("Failure");
432 }),
433 ),
434 );
435 },
436);
437
438/**
439 * The credentials the carved-out logs client resolves its request from: the
440 * test WebSocket server as the API base, and a fixed service token.
441 */
442const tailWith = (
443 baseUrl: string,
444 deploymentId: string,
445 query?: Parameters<typeof tailDeploymentLogs>[1],
446) =>
447 tailDeploymentLogs(deploymentId, query).pipe(
448 Stream.provideService(
449 Credentials,
450 Effect.succeed({
451 apiToken: Redacted.make("test-token"),
452 apiBaseUrl: baseUrl,
453 }),
454 ),
455 );
456
457const logRecord = (text: string, byteStart = 0) =>
458 JSON.stringify({
459 type: "log",
460 text,
461 byteStart,
462 byteEnd: byteStart + text.length,
463 });
464
465const terminalRecord = (input: {
466 kind: "end" | "error";
467 code: string;
468 message?: string;
469 retryable: boolean;
470 cursor: string | null;
471}) =>
472 JSON.stringify({
473 type: "terminal",
474 message: "segment ended",
475 ...input,
476 });
477
478const withWebSocketServer = <A, E, R>(
479 f: (server: WebSocketServer) => Effect.Effect<A, E, R>,
480) =>
481 Effect.acquireUseRelease(
482 Effect.sync(() => new WebSocketServer({ host: "127.0.0.1", port: 0 })),
483 f,
484 (server) =>
485 Effect.callback<void>((resume) => {
486 // This release runs as an uninterruptible finalizer, and bun's ws
487 // shim does not reliably fire the `server.close` callback once a
488 // connection was closed server-side — terminate stragglers and
489 // bound the wait so a shim quirk can't hang the test process.
490 let timer: ReturnType<typeof setTimeout> | undefined;
491 const done = () => {
492 if (timer !== undefined) clearTimeout(timer);
493 resume(Effect.void);
494 };
495 for (const client of server.clients) client.terminate();
496 timer = setTimeout(done, 1_000);
497 server.close(done);
498 }).pipe(Effect.ignore),
499 );
500
501const listenUrl = (server: WebSocketServer) =>
502 Effect.callback<string, Error>((resume) => {
503 const complete = () => {
504 cleanup();
505 const address = server.address();
506 if (address && typeof address === "object") {
507 resume(Effect.succeed(`http://127.0.0.1:${address.port}`));
508 } else {
509 resume(Effect.fail(new Error("WebSocket server has no TCP address")));
510 }
511 };
512 const fail = (cause: unknown) => {
513 cleanup();
514 resume(
515 Effect.fail(cause instanceof Error ? cause : new Error(String(cause))),
516 );
517 };
518 const cleanup = () => {
519 server.off("listening", complete);
520 server.off("error", fail);
521 };
522
523 if (server.address()) {
524 complete();
525 return;
526 }
527
528 server.once("listening", complete);
529 server.once("error", fail);
530 return Effect.sync(cleanup);
531 });
532
533interface SilentTcpServer {
534 server: Server;
535 sockets: Set<Socket>;
536 url: string;
537}
538
539const withSilentTcpServer = <A, E, R>(
540 f: (url: string, connected: Effect.Effect<void>) => Effect.Effect<A, E, R>,
541) =>
542 Effect.gen(function* () {
543 const connected = yield* Deferred.make<void>();
544 return yield* Effect.acquireUseRelease(
545 Effect.callback<SilentTcpServer, Error>((resume) => {
546 const sockets = new Set<Socket>();
547 const server = createServer((socket) => {
548 sockets.add(socket);
549 Deferred.doneUnsafe(connected, Effect.void);
550 socket.on("data", () => {
551 // Intentionally never complete the HTTP upgrade handshake.
552 });
553 socket.on("error", () => {});
554 socket.on("close", () => sockets.delete(socket));
555 });
556 const onError = (cause: Error) => {
557 cleanup();
558 resume(Effect.fail(cause));
559 };
560 const onListening = () => {
561 cleanup();
562 const address = server.address();
563 if (!address || typeof address === "string") {
564 resume(Effect.fail(new Error("TCP server has no address")));
565 return;
566 }
567 resume(
568 Effect.succeed({
569 server,
570 sockets,
571 url: `http://127.0.0.1:${address.port}`,
572 }),
573 );
574 };
575 const cleanup = () => {
576 server.off("error", onError);
577 server.off("listening", onListening);
578 };
579 server.once("error", onError);
580 server.once("listening", onListening);
581 server.listen(0, "127.0.0.1");
582 return Effect.sync(() => {
583 cleanup();
584 for (const socket of sockets) socket.destroy();
585 server.close();
586 });
587 }),
588 ({ url }) => f(url, Deferred.await(connected)),
589 ({ server, sockets }) =>
590 Effect.callback<void>((resume) => {
591 for (const socket of sockets) socket.destroy();
592 server.close(() => resume(Effect.void));
593 }).pipe(Effect.ignore),
594 );
595 });