1import * as Cache from "effect/Cache";
2import type * as Cause from "effect/Cause";
3import * as Config from "effect/Config";
4import * as Console from "effect/Console";
5import * as Context from "effect/Context";
6import * as Deferred from "effect/Deferred";
7import * as Duration from "effect/Duration";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
11import type { PlatformError } from "effect/PlatformError";
12import * as Queue from "effect/Queue";
13import * as Schedule from "effect/Schedule";
14import * as Schema from "effect/Schema";
15import * as Scope from "effect/Scope";
16import * as Stream from "effect/Stream";
17import * as HttpClient from "effect/http/HttpClient";
18import * as HttpServer from "effect/http/HttpServer";
19import { HttpServerRequest } from "effect/http/HttpServerRequest";
20import * as HttpServerResponse from "effect/http/HttpServerResponse";
21import * as ChildProcess from "effect/process/ChildProcess";
22import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
23import { fileURLToPath } from "node:url";
24import { pipedColorEnv } from "../Util/Terminal.ts";
25import { nodeLoaderArgs } from "../Util/Node.ts";
26import { httpServer } from "../Util/PlatformServices.ts";
27import { SPAWNER_URL_ENV_KEY } from "./RpcProviderProxy.ts";
28import {
29 RPC_SERVER_ENVIRONMENT_KEY,
30 type RpcServerEnvironment,
31} from "./RpcServerEnvironment.ts";
32
33export class RpcSpawner extends Context.Service<
34 RpcSpawner,
35 {
36 readonly url: string;
37 }
38>()("alchemy/Local/RpcSpawner") {}
39
40/**
41 * The spawner forks one child per distinct `serverEntryUrl`. Stack-specific
42 * context (stack name/stage, AlchemyContext) is NOT part of the spawn key —
43 * it travels per RPC session (see `SESSION_ENV_PARAM`), so many stacks
44 * (e.g. every test file in a run) share a single sidecar process.
45 */
46export const RpcSpawnPayload = Schema.Struct({ serverEntryUrl: Schema.String });
47export type RpcSpawnPayload = typeof RpcSpawnPayload.Type;
48
49/**
50 * One line of sidecar child output, tagged with the channel it arrived on.
51 * Streamed as NDJSON over the spawner's {@link LOGS_PATH} endpoint.
52 */
53export interface SidecarLogLine {
54 readonly channel: "stdout" | "stderr";
55 readonly line: string;
56}
57
58export const HeartbeatFrame = Schema.Struct({
59 channel: Schema.Literal("heartbeat"),
60});
61export type HeartbeatFrame = typeof HeartbeatFrame.Type;
62
63const SidecarLogLineSchema = Schema.Struct({
64 channel: Schema.Literals(["stdout", "stderr"]),
65 line: Schema.String,
66});
67
68/**
69 * Path on the spawner's HTTP server that streams sidecar output as NDJSON
70 * ({@link SidecarLogLine} per line). Consumed by {@link forwardSidecarLogs}
71 * from the exec child, which owns the terminal renderer.
72 */
73export const LOGS_PATH = "/logs";
74
75export const make = Effect.fn(function* ({
76 profile,
77 envFile,
78}: Pick<RpcServerEnvironment, "profile" | "envFile">) {
79 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
80 const scope = yield* Effect.scope;
81 const cache = yield* Cache.make({
82 lookup: (serverEntryUrl: string) =>
83 spawn(serverEntryUrl).pipe(Scope.provide(scope)),
84 capacity: Infinity,
85 });
86
87 // Sidecar output hub. During `alchemy dev` this process (the outer dev
88 // command) shares the tty with the exec child, and the exec child owns the
89 // repainting progress renderer. Printing sidecar lines RAW from here
90 // interleaves with the renderer's repaints and corrupts the region
91 // (stacked/duplicated frames). So: when an exec child is subscribed via
92 // the /logs endpoint, hand lines to it and let its logger route them into
93 // the renderer; only log from this process as a fallback when no
94 // subscriber is connected (e.g. during a --watch restart gap, when no
95 // renderer is alive either).
96 const subscribers = new Set<(line: SidecarLogLine) => void>();
97 const publish = (line: SidecarLogLine): Effect.Effect<void> =>
98 Effect.suspend(() => {
99 if (subscribers.size > 0) {
100 for (const notify of subscribers) notify(line);
101 return Effect.void;
102 }
103 // Fallback path (no exec child subscribed): sidecar lines already carry
104 // the sidecar's logger prefix, so print verbatim rather than stamping
105 // this process's logger prefix on top.
106 return line.channel === "stderr"
107 ? Console.error(line.line)
108 : Console.log(line.line);
109 });
110
111 const spawn = Effect.fn(function* (serverEntryUrl: string) {
112 const bin = typeof globalThis.Bun !== "undefined" ? "bun" : "node";
113 const main = fileURLToPath(serverEntryUrl);
114 // The sidecar runs on THIS process's runtime binary, not whatever PATH
115 // resolves: a `.ts` entry (checkout) needs the dev-mode loader hooks,
116 // which are gated on this node's version, and a published `.js` entry
117 // must not silently switch Node versions between supervisor and sidecar.
118 const program = bin === "bun" ? "bun" : process.execPath;
119 // Stack-specific context is deliberately absent: sessions carry their
120 // own SessionEnvironment, so one child serves every stack.
121 const environment: RpcServerEnvironment = {
122 profile,
123 envFile,
124 };
125 // Sidecar stdio is piped, so toolchains down the chain (vite, workerd,
126 // pretty loggers) detect a non-TTY and drop ANSI colors — but their
127 // output ultimately renders on THIS process's terminal. Force color
128 // through the pipe when that terminal supports it, unless the user
129 // already decided (NO_COLOR / FORCE_COLOR). `extendEnv` propagates it
130 // from the sidecar to its own children (dev servers, workerd).
131 const command = ChildProcess.make(
132 program,
133 {
134 bun: ["run", main],
135 // Under Node the sidecar starts with alchemy's Oxc loader hooks, the
136 // same way `dev.ts` starts the exec child: a `.ts` entry (src/ in a
137 // checkout) also gets src-condition resolution, a `.js` entry (lib/
138 // in a published install) gets the loader alone.
139 node: [...nodeLoaderArgs(main), main],
140 }[bin],
141 {
142 stdout: "pipe",
143 // Piped (NOT inherited) so the child's output routes through the
144 // Effect Console service: raw writes to the parent's fd corrupt the
145 // test runner's reporter/TUI. The drain below is mandatory — an
146 // unread pipe eventually fills and blocks the child.
147 stderr: "pipe",
148 // `detached` deliberately unset: the spawner defaults to a new process
149 // group on POSIX (and none on Windows, where a detached console child
150 // has its own quirks). A terminal Ctrl+C must NOT reach the sidecar
151 // directly — its lifecycle belongs to this process's finalizer (the
152 // bounded kill below), and a sidecar tearing itself down concurrently
153 // yanks local providers out from under the exec child mid-shutdown.
154 // As a group leader the kill also reaps the sidecar's own children
155 // (workerd, emulators) instead of orphaning them.
156 env: {
157 [RPC_SERVER_ENVIRONMENT_KEY]: JSON.stringify(environment),
158 ...pipedColorEnv(),
159 },
160 extendEnv: true,
161 },
162 );
163 const handle = yield* spawner.spawn(command);
164 yield* handle.stderr.pipe(
165 Stream.decodeText,
166 Stream.splitLines,
167 Stream.runForEach((line) => publish({ channel: "stderr", line })),
168 Effect.ignore,
169 Effect.forkScoped,
170 );
171 // This scope is the child handle's sole owner. Graceful shutdown runs this
172 // finalizer; abrupt parent loss closes the RPC parent connection and the
173 // child self-terminates (both paths are covered by RpcSpawnerCleanup).
174 // Command cleanup allows 1s for TERM and 1s for KILL; reserve 1s for sidecar teardown.
175 const kill = handle.kill({ forceKillAfter: "3 seconds" });
176 yield* Effect.addFinalizer(() => kill.pipe(Effect.ignore));
177 const url = yield* getRpcAddress(handle.stdout, (line) =>
178 publish({ channel: "stdout", line }),
179 );
180 const ws = yield* Effect.acquireRelease(
181 Effect.sync(() => new WebSocket(new URL("/parent", url))),
182 (ws) => Effect.sync(() => ws.close()),
183 );
184 return {
185 url,
186 isRunning: Effect.zipWith(
187 handle.isRunning,
188 Effect.sync(
189 () =>
190 ws.readyState === WebSocket.CONNECTING ||
191 ws.readyState === WebSocket.OPEN,
192 ),
193 (a, b) => a && b,
194 { concurrent: true },
195 ),
196 kill,
197 };
198 });
199
200 const register = Effect.fn(function* (
201 serverEntryUrl: string,
202 attempt = 0,
203 ): Effect.fn.Return<string, PlatformError> {
204 const child = yield* Cache.get(cache, serverEntryUrl);
205 if (yield* child.isRunning) {
206 return child.url;
207 }
208 if (attempt > 3) {
209 return yield* Effect.die(
210 new Error(
211 `Failed to spawn RPC server for "${serverEntryUrl}" after ${attempt} attempts.`,
212 ),
213 );
214 }
215 yield* child.kill;
216 yield* Cache.invalidate(cache, serverEntryUrl);
217 return yield* register(serverEntryUrl, attempt + 1);
218 });
219
220 const server = yield* HttpServer.HttpServer;
221
222 const encoder = new TextEncoder();
223 // The first heartbeat flushes the response headers immediately; the
224 // periodic ones defeat idle timeouts (Bun kills sockets that stay silent
225 // for ~10s). Entries without a `line` are skipped by the client.
226 const heartbeat: HeartbeatFrame = { channel: "heartbeat" };
227 const HEARTBEAT = encoder.encode(`${JSON.stringify(heartbeat)}\n`);
228 const heartbeats = Stream.make(HEARTBEAT).pipe(
229 Stream.concat(
230 Stream.fromSchedule(Schedule.spaced(Duration.seconds(5))).pipe(
231 Stream.map(() => HEARTBEAT),
232 ),
233 ),
234 );
235
236 yield* server.serve(
237 Effect.gen(function* () {
238 const request = yield* HttpServerRequest;
239 // `request.url` is relative under Node and absolute under Bun —
240 // normalize to the pathname before matching.
241 const pathname = new URL(request.url, "http://localhost").pathname;
242 if (pathname === LOGS_PATH) {
243 // Long-lived NDJSON stream of sidecar output. The subscriber (exec
244 // child) routes these lines through its logger, which the renderer
245 // inserts above the progress region instead of tearing it. Client
246 // disconnect interrupts the stream and unregisters the subscriber.
247 const queue = yield* Queue.make<Uint8Array, Cause.Done>();
248 const notify = (line: SidecarLogLine) => {
249 Queue.offerUnsafe(queue, encoder.encode(`${JSON.stringify(line)}\n`));
250 };
251 subscribers.add(notify);
252 return HttpServerResponse.stream(
253 Stream.merge(
254 Stream.fromQueue(queue).pipe(
255 Stream.ensuring(Effect.sync(() => subscribers.delete(notify))),
256 ),
257 heartbeats,
258 ),
259 { contentType: "application/x-ndjson" },
260 );
261 }
262 const decoded = yield* Effect.result(
263 request.json.pipe(
264 Effect.flatMap(Schema.decodeUnknownEffect(RpcSpawnPayload)),
265 ),
266 );
267 if (decoded._tag === "Failure") {
268 return HttpServerResponse.text("Invalid RPC spawn payload.", {
269 status: 400,
270 });
271 }
272 const entry = yield* Effect.result(
273 Effect.try({
274 try: () => new URL(decoded.success.serverEntryUrl),
275 catch: (cause) => cause,
276 }),
277 );
278 if (entry._tag === "Failure" || entry.success.protocol !== "file:") {
279 return HttpServerResponse.text(
280 "serverEntryUrl must be a valid file URL.",
281 { status: 400 },
282 );
283 }
284 const url = yield* register(entry.success.href);
285 return HttpServerResponse.text(url);
286 }),
287 );
288
289 return RpcSpawner.of({
290 url: HttpServer.formatAddress(server.address),
291 });
292});
293
294export const layerServer = (
295 environment: Pick<RpcServerEnvironment, "profile" | "envFile">,
296) =>
297 Layer.effect(RpcSpawner, make(environment)).pipe(
298 // `/logs` is deliberately long-lived. On shutdown the exec child still
299 // owns that response while this scope owns the exec child, so waiting for
300 // Node's default 20-second graceful HTTP drain creates a finalizer cycle.
301 // Stop accepting/serving immediately so the watcher finalizer can close
302 // the child; all actual sidecars still use their own bounded finalizers.
303 Layer.provide(httpServer(0, "127.0.0.1", { gracefulShutdownTimeout: 0 })),
304 );
305
306const RPC_ADDRESS_REGEX =
307 /(<ALCHEMY_RPC_ADDRESS>)(.+)(<\/ALCHEMY_RPC_ADDRESS>)/;
308
309const getRpcAddress = (
310 stdout: Stream.Stream<Uint8Array, PlatformError>,
311 publish: (line: string) => Effect.Effect<void>,
312) =>
313 Effect.gen(function* () {
314 const address = yield* Deferred.make<string>();
315 // Set once the address line is seen; lines before it are handshake noise.
316 let addressSeen = false;
317 yield* stdout.pipe(
318 Stream.decodeText,
319 Stream.splitLines,
320 Stream.runForEach((line) => {
321 const match = line.match(RPC_ADDRESS_REGEX);
322 if (match) {
323 addressSeen = true;
324 return Deferred.succeed(address, match[2]);
325 }
326 return addressSeen ? publish(line) : Effect.void;
327 }),
328 Effect.forkScoped,
329 );
330 return yield* Deferred.await(address);
331 });
332
333export const parseSidecarLogLine = (
334 raw: string,
335): SidecarLogLine | undefined => {
336 try {
337 return Schema.decodeUnknownOption(SidecarLogLineSchema)(
338 JSON.parse(raw),
339 ).pipe(Option.getOrUndefined);
340 } catch {
341 return undefined;
342 }
343};
344
345/**
346 * Pull sidecar output from the spawner (the outer `alchemy dev` process)
347 * into THIS process's Console. The exec child owns the terminal renderer —
348 * Sigil patches its `console`, so lines printed here are inserted cleanly
349 * above the repainting progress region instead of racing it on the shared
350 * tty. Forks in the ambient scope and never fails: when no spawner is
351 * configured (`ALCHEMY_RPC_SPAWNER_URL` absent — plain deploy/destroy) it is
352 * a no-op, and if the connection drops the spawner's own fallback printing
353 * takes over.
354 */
355export const forwardSidecarLogs = (
356 /** Mirrors every forwarded line (e.g. into a dev log file). */
357 tee?: (line: SidecarLogLine) => void,
358): Effect.Effect<void, never, HttpClient.HttpClient | Scope.Scope> =>
359 Config.String(SPAWNER_URL_ENV_KEY).pipe(
360 Effect.flatMap((spawnerUrl) => {
361 const streamOnce = Effect.gen(function* () {
362 const client = yield* HttpClient.HttpClient;
363 const response = yield* client.get(
364 new URL(LOGS_PATH, spawnerUrl).toString(),
365 );
366 yield* response.stream.pipe(
367 Stream.decodeText,
368 Stream.splitLines,
369 Stream.runForEach((raw) =>
370 Effect.suspend(() => {
371 const parsed = parseSidecarLogLine(raw);
372 if (parsed === undefined) return Effect.void;
373 tee?.(parsed);
374 // Sidecar lines already carry the sidecar's own logger prefix
375 // (timestamp/level/fiber) — print them verbatim; routing through
376 // this process's logger would stamp a second prefix on top.
377 return parsed.channel === "stderr"
378 ? Console.error(parsed.line)
379 : Console.log(parsed.line);
380 }),
381 ),
382 );
383 });
384 // Keep the subscription alive for the whole dev session: reconnect
385 // (paced) if the stream ends or errors. While disconnected the spawner's
386 // fallback printing covers the gap; the loop dies with the ambient scope.
387 return streamOnce.pipe(
388 Effect.ignore,
389 Effect.andThen(Effect.sleep("1 second")),
390 Effect.forever,
391 );
392 }),
393 Effect.ignore,
394 Effect.forkScoped,
395 Effect.asVoid,
396 );