1import * as Duration from "effect/Duration";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import type * as Scope from "effect/Scope";
5import * as Stream from "effect/Stream";
6import * as ChildProcess from "effect/process/ChildProcess";
7import type { ChildProcessHandle } from "effect/process/ChildProcessSpawner";
8
9/**
10 * Wait for the child to exit (with timeout). Uses `handle.isRunning`
11 * rather than `handle.exitCode` because the latter raises a
12 * `PlatformError` for processes killed by signal (no exit code), which
13 * is a perfectly normal outcome for the SIGKILL test cases.
14 */
15export const waitForExit = (
16 handle: ChildProcessHandle,
17 timeout: Duration.Input,
18): Effect.Effect<void, Error> =>
19 handle.isRunning.pipe(
20 Effect.orElseSucceed(() => false),
21 Effect.repeat({
22 schedule: Schedule.spaced(Duration.millis(50)),
23 until: (running) => !running,
24 }),
25 Effect.timeout(timeout),
26 Effect.catchTag("TimeoutError", () =>
27 Effect.fail(new Error("child did not exit in time")),
28 ),
29 );
30
31export const assertPidExited = (pid: number): Effect.Effect<void, Error> =>
32 isAlive(pid).pipe(
33 Effect.orElseSucceed(() => false),
34 Effect.repeat({
35 schedule: Schedule.spaced(Duration.millis(50)),
36 until: (alive) => !alive,
37 }),
38 Effect.timeout("5 seconds"),
39 Effect.catchTag("TimeoutError", () =>
40 Effect.fail(new Error("child did not exit in time")),
41 ),
42 );
43
44/**
45 * `process.kill(pid, 0)` is a sync syscall that probes a pid we don't
46 * own a handle to (e.g. a grandchild spawned by the parent fixture).
47 * Wrapped in `Effect.sync` so it participates in the runtime.
48 */
49export const isAlive = (pid: number): Effect.Effect<boolean> =>
50 Effect.sync(() => {
51 try {
52 process.kill(pid, 0);
53 return true;
54 } catch {
55 return false;
56 }
57 });
58
59/**
60 * Resolves the pid currently LISTENing on the port of `wsUrl`. Uses an
61 * `lsof` invocation (`netstat -ano` on Windows, which has no `lsof`); we
62 * don't own a handle to whatever process is listening so there's no
63 * ChildProcessHandle equivalent.
64 */
65export const pidListeningOn = (wsUrl: string) => {
66 const port = new URL(wsUrl).port;
67 if (process.platform === "win32") {
68 return ChildProcess.make("netstat", ["-ano", "-p", "TCP"], {
69 stdout: "pipe",
70 }).pipe(
71 Effect.flatMap((handle) =>
72 handle.stdout.pipe(Stream.decodeText, Stream.mkString),
73 ),
74 Effect.map((stdout) => {
75 // Columns: Proto | Local Address | Foreign Address | State | PID
76 const line = stdout
77 .split("\n")
78 .find((l) => l.includes("LISTENING") && l.includes(`:${port} `));
79 return Number.parseInt(line?.trim().split(/\s+/).at(-1) ?? "", 10);
80 }),
81 );
82 }
83 return ChildProcess.make("lsof", [`-iTCP:${port}`, "-sTCP:LISTEN", "-t"], {
84 stdout: "pipe",
85 }).pipe(
86 Effect.flatMap((handle) =>
87 handle.stdout.pipe(Stream.decodeText, Stream.mkString),
88 ),
89 Effect.map((stdout) => Number.parseInt(stdout.trim().split("\n")[0]!, 10)),
90 );
91};
92
93/**
94 * POSIX process-group id of a pid via `ps` (no Node API exposes another
95 * process's pgid). Returns NaN when the pid is gone.
96 */
97export const pgidOf = (pid: number) =>
98 ChildProcess.make("ps", ["-o", "pgid=", "-p", String(pid)], {
99 stdout: "pipe",
100 }).pipe(
101 Effect.flatMap((handle) =>
102 handle.stdout.pipe(Stream.decodeText, Stream.mkString),
103 ),
104 Effect.map((stdout) => Number.parseInt(stdout.trim(), 10)),
105 );
106
107/** Send a signal to a pid we don't own a handle to. */
108export const killPid = (
109 pid: number,
110 signal: NodeJS.Signals,
111): Effect.Effect<void> =>
112 Effect.sync(() => {
113 try {
114 process.kill(pid, signal);
115 } catch {}
116 });
117
118/**
119 * Open a WebSocket inside a scope so it's reliably closed at scope
120 * end. Resolves once `open` fires or fails on error / close.
121 */
122export const openWebSocket = (
123 url: string | URL,
124): Effect.Effect<WebSocket, Error, Scope.Scope> =>
125 Effect.acquireRelease(
126 Effect.callback<WebSocket, Error>((resume) => {
127 const ws = new WebSocket(url);
128 const cleanup = () => {
129 ws.removeEventListener("open", onOpen);
130 ws.removeEventListener("error", onError);
131 };
132 const onOpen = () => {
133 cleanup();
134 resume(Effect.succeed(ws));
135 };
136 const onError = () => {
137 cleanup();
138 try {
139 ws.close();
140 } catch {}
141 resume(Effect.fail(new Error(`websocket connect failed: ${url}`)));
142 };
143 ws.addEventListener("open", onOpen, { once: true });
144 ws.addEventListener("error", onError, { once: true });
145 }),
146 (ws) =>
147 Effect.sync(() => {
148 try {
149 ws.close();
150 } catch {}
151 }),
152 );
153
154/** Probe whether a websocket can be opened. Never fails. */
155export const canOpenWebSocket = (
156 url: string | URL,
157 timeout: Duration.Input = Duration.millis(1_500),
158): Effect.Effect<boolean> =>
159 Effect.callback<boolean>((resume) => {
160 const ws = new WebSocket(url);
161 let settled = false;
162 const settle = (v: boolean) => {
163 if (settled) return;
164 settled = true;
165 try {
166 ws.close();
167 } catch {}
168 resume(Effect.succeed(v));
169 };
170 ws.addEventListener("open", () => settle(true), { once: true });
171 ws.addEventListener("error", () => settle(false), { once: true });
172 ws.addEventListener("close", () => settle(false), { once: true });
173 }).pipe(
174 Effect.timeoutOrElse({
175 duration: timeout,
176 orElse: () => Effect.succeed(false),
177 }),
178 );