1import { unwrapRpcHandlers } from "@/Local/RpcSerialization.ts";
2import type { RpcProxyApi } from "@/Local/RpcServer.ts";
3import { PlatformServices } from "@/Util/PlatformServices.ts";
4import { assert, describe, expect, it } from "alchemy-test";
5import { newWebSocketRpcSession, type RpcStub } from "capnweb";
6import * as Clock from "effect/Clock";
7import * as Effect from "effect/Effect";
8import * as Sink from "effect/Sink";
9import * as Stream from "effect/Stream";
10import * as ChildProcess from "effect/process/ChildProcess";
11import { fileURLToPath } from "node:url";
12import { openWebSocket, waitForExit } from "./fixtures/process-effect.ts";
13import { runtimes } from "./fixtures/runtimes.ts";
14
15const FIXTURE_TS = fileURLToPath(
16 new URL("./fixtures/rpc-server-entry.ts", import.meta.url),
17);
18
19const ADDRESS_RE = /<ALCHEMY_RPC_ADDRESS>(.+?)<\/ALCHEMY_RPC_ADDRESS>/;
20
21const sampleEnv = () =>
22 JSON.stringify({
23 profile: null,
24 envFile: null,
25 alchemyContext: {
26 dotAlchemy: "/tmp/.alchemy",
27 updateStateStore: false,
28 dev: true,
29 adopt: false,
30 },
31 stack: { name: "test", stage: "dev" },
32 });
33
34// Concurrent at both levels: every test spawns its own isolated child
35// process, and two of them deliberately wait out the sidecar's ~10s
36// parent-connect self-termination. Run serially (the file default) those
37// two waits alone would stack to ~20s of wall clock.
38describe.concurrent("Local.RpcServer", { tags: ["local"] }, () => {
39 for (const runtime of runtimes()) {
40 describe.concurrent.skipIf(!runtime.available)(runtime.name, () => {
41 const [bin, ...args] = runtime.argv(FIXTURE_TS);
42 const launch = ChildProcess.make(bin, args, {
43 env: {
44 ALCHEMY_RPC_SERVER_ENVIRONMENT: sampleEnv(),
45 },
46 extendEnv: true,
47 // We never write to the child's stdin, so close it. stdout/stderr
48 // default to "pipe" which is what we want for the buffering forks
49 // below.
50 stdin: "ignore",
51 // SIGTERM first, escalate to SIGKILL after 1s if the child hasn't
52 // exited. Matches the behavior of the old hand-rolled finalizer.
53 killSignal: "SIGTERM",
54 forceKillAfter: "1 second",
55 });
56
57 it.live(
58 "prints the RPC address marker on stdout and accepts /parent + session connections",
59 () =>
60 Effect.gen(function* () {
61 const proc = yield* launch;
62 const url = yield* proc.stdout.pipe(
63 Stream.decodeText,
64 Stream.run(
65 Sink.fold(
66 () => "",
67 (acc) => !acc.includes("</ALCHEMY_RPC_ADDRESS>"),
68 (acc, chunk) => Effect.succeed(acc + chunk),
69 ),
70 ),
71 Effect.timeout("5 seconds"),
72 Effect.map((output) => output.match(ADDRESS_RE)?.[1]),
73 );
74 assert(url, `url not found in output: "${url}"`);
75 expect(url).toMatch(/^ws:\/\//);
76
77 // Open the parent websocket inside the scope so it stays alive
78 // for the duration of the RPC exchange below; closing it later
79 // is exactly what triggers the child to exit.
80 const parent = yield* openWebSocket(new URL("/parent", url));
81
82 // Drive a real RPC call through a session websocket. capnweb's
83 // surface is Promise-based, so we wrap exactly at the boundary
84 // and let everything above and below stay in Effect.
85
86 // TODO(sam): tsc (typescript 7) vomits here, so we cast to any.
87 const stub = (newWebSocketRpcSession as any)(
88 url,
89 ) as RpcStub<RpcProxyApi>;
90 const result = yield* Effect.promise(async () => {
91 const provider = await stub.getProvider(
92 "Test.Echo",
93 new URL("./fixtures/rpc-server-entry.ts", import.meta.url).href,
94 );
95 const handlers = unwrapRpcHandlers(provider as any) as {
96 echo: (msg: string) => Effect.Effect<string>;
97 };
98 return await Effect.runPromise(handlers.echo("hello"));
99 });
100 expect(result).toBe("echo:hello");
101
102 // Closing the parent ws should cause the child to exit promptly.
103 yield* Effect.sync(() => parent.close());
104 // waitForExit fails if the child is still running after the
105 // timeout, so reaching this point means the child exited.
106 yield* waitForExit(proc, "5 seconds");
107 }).pipe(Effect.scoped, Effect.provide(PlatformServices)),
108 { timeout: 30_000 },
109 );
110
111 it.live(
112 "exits if the parent never connects within ~10s",
113 () =>
114 Effect.gen(function* () {
115 const start = yield* Clock.currentTimeMillis;
116 const proc = yield* launch;
117 // Never open /parent — the server should self-terminate via the
118 // connect timeout.
119 yield* waitForExit(proc, "20 seconds");
120 const elapsed = (yield* Clock.currentTimeMillis) - start;
121 expect(elapsed).toBeLessThan(18_000);
122 }).pipe(Effect.scoped, Effect.provide(PlatformServices)),
123 { timeout: 30_000 },
124 );
125 });
126 }
127});