1import * as Effect from "effect/Effect";
2import { WebSocketServer, type Server } from "ws";
3import * as RpcServer from "./RpcServer.ts";
4import { SESSION_ENV_PARAM } from "./RpcServerEnvironment.ts";
5
6export const RpcServerNode = RpcServer.layerServer(
7 Effect.fn(function* ({
8 parentConnected,
9 parentDisconnected,
10 createRpcSession,
11 }) {
12 const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
13 const url = yield* Effect.callback<string>((resume) => {
14 server.on("connection", (ws, req) => {
15 if (req.url?.startsWith("/parent")) {
16 parentConnected();
17 ws.on("close", () => {
18 parentDisconnected();
19 });
20 return;
21 }
22 const sessionEnv =
23 new URL(req.url ?? "/", "http://localhost").searchParams.get(
24 SESSION_ENV_PARAM,
25 ) ?? undefined;
26 const session = createRpcSession(ws, sessionEnv);
27 ws.on("message", (data) => {
28 session.dispatch.message(data.toString());
29 });
30 ws.on("close", (code, reason) => {
31 session.dispatch.close(code, reason.toString());
32 });
33 });
34 server.on("error", (error) => {
35 resume(Effect.die(error));
36 });
37 server.on("listening", () => {
38 resume(getServerAddress(server));
39 });
40 });
41 yield* Effect.addFinalizer(() => Effect.sync(() => server.close()));
42 return { url };
43 }),
44);
45
46function getServerAddress(server: Server): Effect.Effect<string> {
47 const address = server.address();
48 if (
49 typeof address === "object" &&
50 address !== null &&
51 "address" in address &&
52 "port" in address
53 ) {
54 return Effect.succeed(
55 `ws://${address.address === "::" ? "localhost" : address.address}:${address.port}`,
56 );
57 }
58 return Effect.die(
59 new Error(
60 `Server address is not an object with address and port properties: ${JSON.stringify(address)}`,
61 ),
62 );
63}