1// @effect-diagnostics nodeBuiltinImport:off -- Socket callbacks allocate opaque capture identifiers synchronously.
2import * as NodeNet from "node:net";
3import * as NodeCrypto from "node:crypto";
4import {
5 Root,
6 decodeEventStream,
7 encodeEventStream,
8 isVisible,
9 type StreamEvent,
10} from "@clo/lib/progress";
11import {
12 ScriptProgressStreamError,
13 type ScriptProgressEvent,
14 type ScriptProgressStream,
15 type ThreadId,
16} from "@t3tools/contracts";
17import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
18import { progressPreloadSource } from "@t3tools/shared/progressPreload";
19import * as Cause from "effect/Cause";
20import * as Effect from "effect/Effect";
21import * as FileSystem from "effect/FileSystem";
22import * as Layer from "effect/Layer";
23import * as Path from "effect/Path";
24import * as Queue from "effect/Queue";
25import * as Schema from "effect/Schema";
26import * as Stream from "effect/Stream";
27
28import * as ServerConfig from "../config.ts";
29
30export class ScriptProgressSetupError extends Schema.TaggedError<ScriptProgressSetupError>()(
31 "ScriptProgressSetupError",
32 { operation: Schema.Literals(["preload", "listen"]), cause: Schema.Defect() },
33) {
34 override get message(): string {
35 return "Unable to prepare script progress capture.";
36 }
37}
38
39const Hello = Schema.Struct({
40 type: Schema.Literal("hello"),
41 token: Schema.String.check(Schema.isMaxLength(128)),
42 pid: Schema.Int.check(Schema.isGreaterThan(0)),
43 producerId: Schema.String.check(Schema.isMinLength(1)).check(Schema.isMaxLength(128)),
44 label: Schema.optional(Schema.String.check(Schema.isMaxLength(256))),
45});
46const Progress = Schema.Struct({
47 type: Schema.Literal("progress"),
48 streamId: Schema.String.check(Schema.isMaxLength(128)),
49 event: Schema.Array(Schema.Json).check(Schema.isMaxLength(4096)),
50});
51const decodeHello = Schema.decodeUnknownSync(Hello);
52const decodeProgress = Schema.decodeUnknownSync(Progress);
53const decodePacket = Schema.decodeSync(Schema.fromJsonString(Schema.Array(Schema.Json)));
54const encodePreloadPath = Schema.encodeSync(Schema.fromJsonString(Schema.String));
55
56type Capture = {
57 readonly id: string;
58 readonly threadId: ThreadId;
59 readonly pid: number;
60 readonly producerId: string;
61 readonly streamId: string;
62 readonly socket: NodeNet.Socket;
63 readonly label: string;
64 readonly root: Root;
65 readonly input: ReadableStreamDefaultController<StreamEvent>;
66 readonly decoder: ReturnType<typeof decodeEventStream<void>>;
67 status: "connected" | "disconnected";
68};
69
70type Subscriber = {
71 readonly threadId: ThreadId;
72 readonly send: (event: ScriptProgressEvent) => boolean;
73 readonly attach: (capture: Capture) => void;
74 readonly detach: (id: string) => void;
75};
76
77const describe = (capture: Capture): ScriptProgressStream => ({
78 id: capture.id,
79 pid: capture.pid,
80 label: capture.label,
81 status: capture.status,
82 summary: capture.root.active.filter(isVisible).map((node) => ({
83 text: node.text,
84 value: node.value,
85 total: node.total,
86 detail: node.showTotal ? node.valueFormatter(node.value, node.total || null) : "",
87 })),
88});
89
90const make = Effect.gen(function* () {
91 const fileSystem = yield* FileSystem.FileSystem;
92 const path = yield* Path.Path;
93 const config = yield* ServerConfig.ServerConfig;
94 const runtimeDirectory = path.join(config.stateDir, "runtime");
95 const preloadPath = path.join(runtimeDirectory, "script-progress-preload.cjs");
96 yield* fileSystem.makeDirectory(runtimeDirectory, { recursive: true }).pipe(
97 Effect.andThen(fileSystem.writeFileString(preloadPath, progressPreloadSource, { mode: 0o600 })),
98 Effect.mapError((cause) => new ScriptProgressSetupError({ operation: "preload", cause })),
99 );
100
101 const tokens = new Map<ThreadId, string>();
102 const tokenThreads = new Map<string, ThreadId>();
103 const captures = new Map<string, Capture>();
104 const subscribers = new Set<Subscriber>();
105 const sockets = new Set<NodeNet.Socket>();
106 let closed = false;
107
108 const forget = (capture: Capture) => {
109 captures.delete(capture.id);
110 for (const subscriber of subscribers) {
111 if (subscriber.threadId !== capture.threadId) continue;
112 subscriber.detach(capture.id);
113 subscriber.send({ type: "remove", id: capture.id });
114 }
115 capture.decoder.cancel();
116 };
117
118 const server = NodeNet.createServer((socket) => {
119 if (closed || sockets.size >= 256) {
120 socket.destroy();
121 return;
122 }
123 sockets.add(socket);
124 socket.setEncoding("utf8");
125 socket.setTimeout(10_000);
126 let pending = "";
127 let hello: typeof Hello.Type | undefined;
128 let threadId: ThreadId | undefined;
129 let pendingPackets = 0;
130 const roots = new Map<string, Capture>();
131 socket.on("error", () => socket.destroy());
132 socket.on("timeout", () => socket.destroy());
133 socket.on("close", () => {
134 sockets.delete(socket);
135 if (closed) return;
136 for (const capture of roots.values()) {
137 if (!captures.has(capture.id)) continue;
138 capture.status = "disconnected";
139 captures.delete(capture.id);
140 captures.set(capture.id, capture);
141 for (const subscriber of subscribers) {
142 if (subscriber.threadId === capture.threadId) {
143 subscriber.send({ type: "upsert", stream: describe(capture) });
144 }
145 }
146 }
147 const disconnected = [...captures.values()].filter(
148 (capture) => capture.threadId === threadId && capture.status === "disconnected",
149 );
150 for (const capture of disconnected.slice(0, -16)) forget(capture);
151 });
152 socket.on("data", (chunk: string) => {
153 if (socket.destroyed) return;
154 pending += chunk;
155 if (Buffer.byteLength(pending) > 1024 * 1024) {
156 socket.destroy();
157 return;
158 }
159 try {
160 let newline: number;
161 while ((newline = pending.indexOf("\n")) !== -1) {
162 const line = pending.slice(0, newline);
163 pending = pending.slice(newline + 1);
164 const message: unknown = JSON.parse(line);
165 if (!hello) {
166 hello = decodeHello(message);
167 threadId = tokenThreads.get(hello.token);
168 if (!threadId) throw new Error("Unknown progress token");
169 socket.setTimeout(0);
170 continue;
171 }
172 const packet = decodeProgress(message);
173 if (!threadId || ++pendingPackets > 256) throw new Error("Progress input overflow");
174 let capture = roots.get(packet.streamId);
175 if (!capture) {
176 if (roots.size >= 16) throw new Error("Too many progress roots");
177 for (const previous of captures.values()) {
178 if (
179 previous.threadId !== threadId ||
180 previous.producerId !== hello.producerId ||
181 previous.streamId !== packet.streamId
182 )
183 continue;
184 forget(previous);
185 previous.socket.destroy();
186 }
187 const threadCaptures = [...captures.values()].filter(
188 (item) => item.threadId === threadId,
189 );
190 if (threadCaptures.filter((item) => item.status === "connected").length >= 64) {
191 throw new Error("Too many active progress roots");
192 }
193 if (captures.size >= 512) {
194 const oldest = [...captures.values()].find((item) => item.status === "disconnected");
195 if (!oldest) throw new Error("Too many progress captures");
196 forget(oldest);
197 }
198 const root = new Root();
199 let input!: ReadableStreamDefaultController<StreamEvent>;
200 const source = new ReadableStream<StreamEvent>({
201 start(controller) {
202 input = controller;
203 },
204 pull() {
205 pendingPackets = Math.max(0, pendingPackets - 1);
206 },
207 });
208 const decoder = decodeEventStream(source, root);
209 capture = {
210 id: NodeCrypto.randomUUID(),
211 threadId,
212 pid: hello.pid,
213 producerId: hello.producerId,
214 streamId: packet.streamId,
215 socket,
216 label: hello.label ?? "node",
217 root,
218 input,
219 decoder,
220 status: "connected",
221 };
222 const activeCapture = capture;
223 const checkRetainedState = () => {
224 const stack = [...root.active];
225 let count = 0;
226 let bytes = 0;
227 while (stack.length > 0) {
228 const node = stack.pop();
229 if (!node) break;
230 bytes +=
231 Buffer.byteLength(node.text) +
232 Buffer.byteLength(JSON.stringify(node.logs)) +
233 Buffer.byteLength(node.valueFormatter(node.value, node.total || null));
234 if (++count > 4096 || bytes > 1024 * 1024) {
235 forget(activeCapture);
236 roots.delete(packet.streamId);
237 socket.destroy();
238 return;
239 }
240 stack.push(...node.children);
241 }
242 };
243 root.on("change", checkRetainedState);
244 void decoder
245 .then(() => {
246 roots.delete(packet.streamId);
247 forget(activeCapture);
248 })
249 .catch(() => {
250 if (activeCapture.status === "connected") socket.destroy();
251 });
252 roots.set(packet.streamId, capture);
253 captures.set(capture.id, capture);
254 for (const subscriber of subscribers) {
255 if (subscriber.threadId !== threadId) continue;
256 subscriber.send({ type: "upsert", stream: describe(capture) });
257 subscriber.attach(capture);
258 }
259 }
260 // The library decoder owns protocol validation and state transitions.
261 capture.input.enqueue(packet.event as StreamEvent);
262 }
263 } catch {
264 socket.destroy();
265 }
266 });
267 });
268
269 const address = yield* Effect.acquireRelease(
270 Effect.callback<NodeNet.AddressInfo, ScriptProgressSetupError>((resume) => {
271 const onError = (cause: Error) =>
272 resume(Effect.fail(new ScriptProgressSetupError({ operation: "listen", cause })));
273 server.once("error", onError);
274 server.listen(0, "127.0.0.1", () => {
275 server.off("error", onError);
276 const address = server.address();
277 if (address && typeof address !== "string") resume(Effect.succeed(address));
278 });
279 return Effect.sync(() => server.close());
280 }),
281 () =>
282 Effect.sync(() => {
283 closed = true;
284 for (const socket of sockets) socket.destroy();
285 for (const capture of captures.values()) capture.decoder.cancel();
286 tokens.clear();
287 tokenThreads.clear();
288 server.close();
289 }),
290 );
291
292 return ScriptProgress.ScriptProgress.of({
293 environmentForThread: (threadId) =>
294 Effect.sync(() => {
295 let token = tokens.get(threadId);
296 if (!token) {
297 token = NodeCrypto.randomBytes(32).toString("hex");
298 tokens.set(threadId, token);
299 tokenThreads.set(token, threadId);
300 }
301 return {
302 T3_SCRIPT_PROGRESS_ENDPOINT: `127.0.0.1:${address.port}`,
303 T3_SCRIPT_PROGRESS_TOKEN: token,
304 NODE_OPTIONS: `--require=${encodePreloadPath(preloadPath)}`,
305 };
306 }),
307 subscribe: (threadId) =>
308 Stream.callback<ScriptProgressEvent, ScriptProgressStreamError>(
309 (queue) =>
310 Effect.acquireRelease(
311 Effect.sync(() => {
312 const readers = new Map<string, ReadableStreamDefaultReader<StreamEvent>>();
313 let disposed = false;
314 const send = (event: ScriptProgressEvent) => {
315 if (disposed) return false;
316 if (Queue.offerUnsafe(queue, event)) return true;
317 Queue.failCauseUnsafe(queue, Cause.fail(new ScriptProgressStreamError({})));
318 return false;
319 };
320 const subscriber: Subscriber = {
321 threadId,
322 send,
323 attach: (capture) => {
324 const reader = encodeEventStream(capture.root, { throttleMs: 100 }).getReader();
325 readers.set(capture.id, reader);
326 void (async () => {
327 try {
328 while (readers.get(capture.id) === reader) {
329 const result = await reader.read();
330 if (result.done) break;
331 if (
332 !send({
333 type: "progress",
334 id: capture.id,
335 event: decodePacket(JSON.stringify(result.value)),
336 summary: describe(capture).summary,
337 })
338 )
339 break;
340 }
341 } catch {
342 // Disconnect is represented by the descriptor, never success.
343 } finally {
344 await reader.cancel().catch(() => {});
345 reader.releaseLock();
346 }
347 })();
348 },
349 detach: (id) => {
350 const reader = readers.get(id);
351 readers.delete(id);
352 void reader?.cancel().catch(() => {});
353 },
354 };
355 subscribers.add(subscriber);
356 const existing = [...captures.values()].filter(
357 (capture) => capture.threadId === threadId,
358 );
359 send({ type: "snapshot", streams: existing.map(describe) });
360 for (const capture of existing) subscriber.attach(capture);
361 return () => {
362 disposed = true;
363 subscribers.delete(subscriber);
364 for (const id of readers.keys()) subscriber.detach(id);
365 };
366 }),
367 (dispose) => Effect.sync(dispose),
368 ),
369 { bufferSize: 256 },
370 ),
371 });
372});
373
374export const layer = Layer.effect(ScriptProgress.ScriptProgress, make);