1import * as Cause from "effect/Cause";
2import * as Data from "effect/Data";
3import * as Deferred from "effect/Deferred";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Queue from "effect/Queue";
8import * as Redacted from "effect/Redacted";
9import * as Stream from "effect/Stream";
10import type WebSocket from "ws";
11import type { RawData } from "ws";
12import type { LogLine } from "../Provider.ts";
13import type { Config as CredentialsConfig } from "@distilled.cloud/prisma";
14import { Credentials } from "./Credentials.ts";
15import { getDeploymentLogsRequest } from "./Internal/LogsClient.ts";
16import type { DeploymentLogsQuery } from "./Types.ts";
17
18export class PrismaLogStreamError extends Data.TaggedError(
19 "PrismaLogStreamError",
20)<{
21 message: string;
22 cause?: unknown;
23}> {}
24
25interface PrismaDeploymentLogLine {
26 type: "log";
27 text: string;
28 byteStart: number;
29 byteEnd: number;
30}
31
32interface PrismaDeploymentTerminalLine {
33 type: "terminal";
34 kind: "end" | "error";
35 code: string;
36 message: string;
37 retryable: boolean;
38 cursor: string | null;
39 details?: Record<string, unknown>;
40}
41
42export type PrismaDeploymentLogRecord =
43 | PrismaDeploymentLogLine
44 | PrismaDeploymentTerminalLine;
45
46export type ParsedDeploymentLogRecord =
47 | { _tag: "log"; line: LogLine; raw: PrismaDeploymentLogLine }
48 | { _tag: "terminal"; raw: PrismaDeploymentTerminalLine };
49
50const WEBSOCKET_HANDSHAKE_TIMEOUT = "10 seconds" as const;
51const RECONNECT_BACKOFF_MS = 100;
52const MAX_NO_PROGRESS_RECONNECTS = 3;
53const LOG_QUEUE_CAPACITY = 64;
54const MAX_WEBSOCKET_PAYLOAD_BYTES = 1024 * 1024;
55
56export const parseDeploymentLogRecord = (
57 message: string,
58 timestamp: Date,
59): Effect.Effect<ParsedDeploymentLogRecord, PrismaLogStreamError> => {
60 const byteLength = new TextEncoder().encode(message).byteLength;
61 // The socket is opened with `maxPayload`, but not every runtime's `ws`
62 // implementation enforces it (bun's ws shim delivers oversized frames), so
63 // bound the frame here too — without retaining any of its contents.
64 if (byteLength > MAX_WEBSOCKET_PAYLOAD_BYTES) {
65 return Effect.fail(
66 new PrismaLogStreamError({
67 message: `Prisma deployment log record exceeds the ${MAX_WEBSOCKET_PAYLOAD_BYTES}-byte frame limit (${byteLength} bytes)`,
68 }),
69 );
70 }
71 let recordType: "log" | "terminal" | "unknown" | "invalid-json" =
72 "invalid-json";
73 return Effect.try({
74 try: () => {
75 const parsed = JSON.parse(message) as unknown;
76 if (parsed === null || typeof parsed !== "object") {
77 recordType = "unknown";
78 throw new Error("Invalid Prisma deployment log record shape");
79 }
80 const raw = parsed as Partial<PrismaDeploymentLogRecord>;
81 recordType =
82 raw.type === "log" || raw.type === "terminal" ? raw.type : "unknown";
83 if (
84 raw.type === "log" &&
85 typeof raw.text === "string" &&
86 typeof raw.byteStart === "number" &&
87 Number.isSafeInteger(raw.byteStart) &&
88 raw.byteStart >= 0 &&
89 typeof raw.byteEnd === "number" &&
90 Number.isSafeInteger(raw.byteEnd) &&
91 raw.byteEnd >= raw.byteStart
92 ) {
93 return {
94 _tag: "log" as const,
95 line: { timestamp, message: raw.text },
96 raw: raw as PrismaDeploymentLogLine,
97 };
98 }
99 if (
100 raw.type === "terminal" &&
101 (raw.kind === "end" || raw.kind === "error") &&
102 typeof raw.code === "string" &&
103 typeof raw.message === "string" &&
104 typeof raw.retryable === "boolean" &&
105 (raw.cursor === null || typeof raw.cursor === "string") &&
106 (raw.details === undefined ||
107 (raw.details !== null &&
108 typeof raw.details === "object" &&
109 !Array.isArray(raw.details)))
110 ) {
111 return {
112 _tag: "terminal" as const,
113 raw: raw as PrismaDeploymentTerminalLine,
114 };
115 }
116 throw new Error("Invalid Prisma deployment log record shape");
117 },
118 catch: () =>
119 new PrismaLogStreamError({
120 message: `Failed to decode Prisma deployment log record (${byteLength} bytes; type: ${recordType})`,
121 }),
122 });
123};
124
125export const tailDeploymentLogs = (
126 deploymentId: string,
127 query?: DeploymentLogsQuery,
128) =>
129 Stream.unwrap(
130 Effect.gen(function* () {
131 // Resolve the credentials on the caller's fiber once; the reconnect
132 // loop below forks with `Effect.runFork`, which cannot carry context.
133 const credentials = yield* Credentials;
134 return tailDeploymentLogsWith(yield* credentials, deploymentId, query);
135 }).pipe(
136 Effect.mapError(
137 (cause) =>
138 new PrismaLogStreamError({
139 message: "Failed to resolve Prisma credentials for log tailing",
140 cause,
141 }),
142 ),
143 ),
144 );
145
146const tailDeploymentLogsWith = (
147 credentials: CredentialsConfig,
148 deploymentId: string,
149 query?: DeploymentLogsQuery,
150) =>
151 Stream.callback<LogLine, PrismaLogStreamError>(
152 (queue) =>
153 Effect.gen(function* () {
154 const WebSocket = yield* loadWebSocketConstructor;
155 const sockets = new Set<WebSocket>();
156 let latestCursor = query?.cursor;
157 let latestByteEnd =
158 query?.cursor !== undefined &&
159 Number.isSafeInteger(Number(query.cursor)) &&
160 Number(query.cursor) >= 0
161 ? Number(query.cursor)
162 : undefined;
163 let noProgressReconnects = 0;
164 let connectionCount = 0;
165 let stopped = false;
166
167 const end = () => {
168 if (stopped) return;
169 stopped = true;
170 Queue.endUnsafe(queue);
171 };
172 const fail = (error: PrismaLogStreamError) => {
173 if (stopped) return;
174 stopped = true;
175 Queue.failCauseUnsafe(queue, Cause.fail(error));
176 };
177
178 const connect = (
179 cursor: string | undefined,
180 ): Effect.Effect<void, PrismaLogStreamError> =>
181 Effect.gen(function* () {
182 if (stopped) return;
183 const request = yield* getDeploymentLogsRequest(
184 deploymentId,
185 cursor === undefined ? query : { ...query, cursor },
186 ).pipe(
187 Effect.provideService(Credentials, Effect.succeed(credentials)),
188 Effect.mapError(
189 () =>
190 new PrismaLogStreamError({
191 message:
192 "Failed to prepare the Prisma deployment log stream request",
193 }),
194 ),
195 );
196 const auth = Redacted.value(request.headers.Authorization);
197 const socket = yield* Effect.try({
198 try: () =>
199 new WebSocket(request.url, {
200 headers: { Authorization: auth },
201 maxPayload: MAX_WEBSOCKET_PAYLOAD_BYTES,
202 }),
203 catch: () =>
204 new PrismaLogStreamError({
205 message: "Failed to open Prisma deployment log WebSocket",
206 }),
207 });
208 if (stopped) {
209 socket.close(1000, "tail stopped");
210 return;
211 }
212 sockets.add(socket);
213 const isReconnect = connectionCount > 0;
214 connectionCount += 1;
215 let terminalHandled = false;
216 let madeProgress = false;
217 let socketOpened = false;
218 const connectionStartByteEnd = latestByteEnd;
219 const opened = yield* Deferred.make<void, PrismaLogStreamError>();
220
221 const scheduleReconnect = (
222 reconnectCursor: string | undefined,
223 code: string,
224 closeMode: "close" | "terminate" | "already-closed",
225 ) => {
226 if (terminalHandled || stopped) return;
227 terminalHandled = true;
228 if (madeProgress) {
229 noProgressReconnects = 0;
230 } else if (isReconnect) {
231 noProgressReconnects += 1;
232 }
233 if (noProgressReconnects >= MAX_NO_PROGRESS_RECONNECTS) {
234 fail(
235 new PrismaLogStreamError({
236 message: `${code}: Prisma deployment log stream made no progress after ${MAX_NO_PROGRESS_RECONNECTS} reconnect attempts`,
237 }),
238 );
239 if (closeMode === "close") {
240 socket.close(1000, "reconnect progress limit reached");
241 } else if (closeMode === "terminate") {
242 socket.terminate();
243 }
244 return;
245 }
246 if (closeMode === "close") {
247 socket.close(1000, "reconnecting from latest log cursor");
248 } else if (closeMode === "terminate") {
249 socket.terminate();
250 }
251 Effect.runFork(
252 Effect.sleep(
253 Duration.millis(
254 Math.min(
255 RECONNECT_BACKOFF_MS * 2 ** noProgressReconnects,
256 1_000,
257 ),
258 ),
259 ).pipe(
260 Effect.andThen(connect(reconnectCursor)),
261 Effect.catch((error) =>
262 Effect.sync(() => {
263 fail(error);
264 }),
265 ),
266 ),
267 );
268 };
269
270 socket.once("open", () => {
271 if (terminalHandled || stopped) return;
272 socketOpened = true;
273 Deferred.doneUnsafe(opened, Effect.void);
274 });
275
276 socket.on("message", (raw) => {
277 if (terminalHandled || stopped) return;
278 const decoded = Effect.runSyncExit(
279 parseDeploymentLogRecord(rawDataToString(raw), new Date()),
280 );
281 if (!Exit.isSuccess(decoded)) {
282 terminalHandled = true;
283 Queue.failCauseUnsafe(queue, decoded.cause);
284 stopped = true;
285 socket.close(1000, "decode failed");
286 return;
287 }
288 const record = decoded.value;
289 if (record._tag === "log") {
290 if (
291 latestByteEnd !== undefined &&
292 record.raw.byteEnd <= latestByteEnd
293 ) {
294 // Exact/older replay after reconnect: drop it and keep the
295 // no-progress budget intact.
296 return;
297 }
298 if (
299 latestByteEnd !== undefined &&
300 record.raw.byteStart < latestByteEnd
301 ) {
302 terminalHandled = true;
303 fail(
304 new PrismaLogStreamError({
305 message:
306 "Prisma deployment log stream returned an overlapping byte range",
307 }),
308 );
309 socket.close(1000, "overlapping log range");
310 return;
311 }
312 madeProgress =
313 madeProgress ||
314 (connectionStartByteEnd === undefined
315 ? record.raw.byteEnd > record.raw.byteStart
316 : record.raw.byteEnd > connectionStartByteEnd);
317 latestByteEnd = record.raw.byteEnd;
318 latestCursor = String(record.raw.byteEnd);
319 if (madeProgress) noProgressReconnects = 0;
320 if (!Queue.offerUnsafe(queue, record.line)) {
321 terminalHandled = true;
322 fail(
323 new PrismaLogStreamError({
324 message: `Prisma deployment log consumer fell behind the ${LOG_QUEUE_CAPACITY}-record safety buffer`,
325 }),
326 );
327 socket.terminate();
328 }
329 return;
330 }
331
332 if (record.raw.retryable) {
333 const reconnectCursor =
334 record.raw.cursor && record.raw.cursor.length > 0
335 ? record.raw.cursor
336 : latestCursor;
337 scheduleReconnect(reconnectCursor, record.raw.code, "close");
338 return;
339 }
340
341 terminalHandled = true;
342 if (record.raw.kind === "error") {
343 fail(
344 new PrismaLogStreamError({
345 message: `${record.raw.code}: ${record.raw.message}`,
346 cause: record.raw.details,
347 }),
348 );
349 socket.close(1000, "non-retryable terminal error");
350 return;
351 }
352
353 end();
354 socket.close(1000, "terminal record received");
355 });
356 socket.on("error", (cause) => {
357 if (terminalHandled || stopped) return;
358 const oversized =
359 cause instanceof Error &&
360 cause.message.includes("Max payload size exceeded");
361 const error = new PrismaLogStreamError({
362 message: "Prisma deployment log WebSocket failed",
363 });
364 if (!socketOpened || oversized) {
365 terminalHandled = true;
366 Deferred.doneUnsafe(opened, Effect.fail(error));
367 fail(error);
368 return;
369 }
370 scheduleReconnect(latestCursor, "transport_error", "terminate");
371 });
372 socket.on("close", () => {
373 sockets.delete(socket);
374 if (!terminalHandled && !stopped) {
375 if (!socketOpened) {
376 terminalHandled = true;
377 const error = new PrismaLogStreamError({
378 message:
379 "Prisma deployment log WebSocket closed before opening",
380 });
381 Deferred.doneUnsafe(opened, Effect.fail(error));
382 fail(error);
383 return;
384 }
385 scheduleReconnect(
386 latestCursor,
387 "transport_closed",
388 "already-closed",
389 );
390 }
391 });
392
393 yield* Deferred.await(opened).pipe(
394 Effect.timeoutOrElse({
395 duration: WEBSOCKET_HANDSHAKE_TIMEOUT,
396 orElse: () => {
397 const error = new PrismaLogStreamError({
398 message: `Prisma deployment log WebSocket handshake timed out after ${WEBSOCKET_HANDSHAKE_TIMEOUT}`,
399 });
400 return Effect.sync(() => {
401 terminalHandled = true;
402 sockets.delete(socket);
403 try {
404 socket.terminate();
405 } catch {
406 // The socket may have closed concurrently with the timeout.
407 }
408 fail(error);
409 }).pipe(Effect.andThen(Effect.fail(error)));
410 },
411 }),
412 );
413 });
414
415 yield* Effect.addFinalizer(() =>
416 Effect.sync(() => {
417 stopped = true;
418 for (const socket of sockets) {
419 try {
420 if (
421 socket.readyState === WebSocket.CONNECTING ||
422 socket.readyState === WebSocket.CLOSING
423 ) {
424 socket.terminate();
425 } else if (socket.readyState === WebSocket.OPEN) {
426 socket.close(1000, "tail stopped");
427 }
428 } catch {
429 // Cleanup must never turn cancellation into a finalizer defect.
430 }
431 }
432 sockets.clear();
433 }),
434 );
435
436 yield* connect(query?.cursor);
437 }),
438 { bufferSize: LOG_QUEUE_CAPACITY, strategy: "dropping" },
439 );
440
441const loadWebSocketConstructor = Effect.tryPromise({
442 try: () => import("ws").then((module) => module.default),
443 catch: (cause) =>
444 new PrismaLogStreamError({
445 message:
446 "Prisma deployment log tailing requires the optional `ws` package.",
447 cause,
448 }),
449});
450
451const rawDataToString = (raw: RawData): string => {
452 if (typeof raw === "string") return raw;
453 if (raw instanceof ArrayBuffer) return new TextDecoder().decode(raw);
454 if (Array.isArray(raw)) return raw.map(rawDataToString).join("");
455 return raw.toString();
456};