1import * as Effect from "effect/Effect";
2import * as RpcServer from "./RpcServer.ts";
3import { SESSION_ENV_PARAM } from "./RpcServerEnvironment.ts";
4import type { ServerRpcSession } from "./RpcServerSession.ts";
5
6export const RpcServerBun = RpcServer.layerServer(
7 Effect.fn(function* ({
8 parentConnected,
9 parentDisconnected,
10 createRpcSession,
11 }) {
12 const server = yield* Effect.sync(() =>
13 Bun.serve<
14 | { type: "session"; session: ServerRpcSession<any> }
15 | { type: "pending"; sessionEnv: string | undefined }
16 | { type: "parent" }
17 >({
18 port: 0,
19 fetch: (request, server) => {
20 const url = new URL(request.url);
21 if (
22 server.upgrade(request, {
23 data:
24 url.pathname === "/parent"
25 ? { type: "parent" }
26 : {
27 type: "pending",
28 sessionEnv:
29 url.searchParams.get(SESSION_ENV_PARAM) ?? undefined,
30 },
31 })
32 ) {
33 return;
34 }
35 return new Response("Upgrade failed", { status: 400 });
36 },
37 websocket: {
38 open: (ws) => {
39 if (ws.data && ws.data.type === "parent") {
40 parentConnected();
41 } else {
42 const sessionEnv =
43 ws.data && ws.data.type === "pending"
44 ? ws.data.sessionEnv
45 : undefined;
46 ws.data = {
47 type: "session",
48 session: createRpcSession(ws, sessionEnv),
49 };
50 }
51 },
52 message: (ws, message) => {
53 if (ws.data.type === "session") {
54 ws.data.session.dispatch.message(message);
55 }
56 },
57 close: (ws, code, reason) => {
58 if (ws.data.type === "session") {
59 ws.data.session.dispatch.close(code, reason);
60 } else if (ws.data.type === "parent") {
61 parentDisconnected();
62 }
63 },
64 },
65 }),
66 );
67 yield* Effect.addFinalizer(() => Effect.sync(() => server.stop(true)));
68 return {
69 url: `ws://${server.hostname}:${server.port}`,
70 };
71 }),
72);