1import * as Context from "effect/Context";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as MutableHashMap from "effect/MutableHashMap";
5import * as HttpServerError from "effect/http/HttpServerError";
6import type * as HttpServerRequest from "effect/http/HttpServerRequest";
7import type * as HttpServerResponse from "effect/http/HttpServerResponse";
8import * as NodeCrypto from "node:crypto";
9import * as NodeHttp from "node:http";
10import type { RuntimeError } from "../RuntimeError.shared.ts";
11import { isRuntimeError, SystemError } from "../RuntimeError.shared.ts";
12import { getAddress } from "../internal/get-address.ts";
13import { makeErrorEnvelope } from "../internal/response.shared.ts";
14
15export class LoopbackServer extends Context.Service<
16 LoopbackServer,
17 {
18 readonly address: string;
19 readonly secret: string;
20 readonly route: (
21 name: string,
22 handler: RouteHandler,
23 ) => Effect.Effect<void>;
24 }
25>()("cloudflare-runtime/LoopbackServer") {}
26
27export type EffectHandler = Effect.Effect<
28 HttpServerResponse.HttpServerResponse,
29 any,
30 HttpServerRequest.HttpServerRequest
31>;
32export type RawHandler = (
33 req: NodeHttp.IncomingMessage,
34 res: NodeHttp.ServerResponse,
35) => void | Promise<void>;
36export type RouteHandler = EffectHandler | RawHandler;
37
38export const LoopbackServerHeaders = {
39 TARGET: "loopback-target",
40 SECRET: "loopback-secret",
41} as const;
42
43export const LoopbackServerLive = Layer.effect(
44 LoopbackServer,
45 Effect.gen(function* () {
46 const routes = MutableHashMap.empty<string, RawHandler>();
47 const secret = crypto.randomUUID();
48 const server = yield* Effect.sync(() =>
49 NodeHttp.createServer(async (req, res) => {
50 const secretHeader = req.headers[LoopbackServerHeaders.SECRET] as
51 | string
52 | undefined;
53 const targetHeader = req.headers[LoopbackServerHeaders.TARGET] as
54 | string
55 | undefined;
56
57 if (!secretHeader) {
58 return writeErrorResponse(
59 res,
60 new SystemError({
61 subtag: "BadRequest",
62 message: `The "${LoopbackServerHeaders.SECRET}" header is required.`,
63 }),
64 400,
65 );
66 } else if (!targetHeader) {
67 return writeErrorResponse(
68 res,
69 new SystemError({
70 subtag: "BadRequest",
71 message: `The "${LoopbackServerHeaders.TARGET}" header is required.`,
72 }),
73 400,
74 );
75 } else if (!timingSafeEqual(secretHeader, secret)) {
76 return writeErrorResponse(
77 res,
78 new SystemError({
79 subtag: "Unauthorized",
80 message: "Unauthorized",
81 }),
82 401,
83 );
84 }
85
86 const route = MutableHashMap.get(routes, targetHeader);
87
88 if (route._tag === "None") {
89 return writeErrorResponse(
90 res,
91 new SystemError({
92 subtag: "NotFound",
93 message: `The route "${targetHeader}" is not found.`,
94 }),
95 404,
96 );
97 }
98
99 try {
100 await route.value(req, res);
101 } catch (error) {
102 return writeErrorResponse(
103 res,
104 isRuntimeError(error)
105 ? error
106 : new SystemError({
107 subtag: "InternalServerError",
108 message: "Internal Server Error",
109 cause: error,
110 }),
111 HttpServerError.isHttpServerError(error)
112 ? (error.response?.status ?? 500)
113 : 500,
114 );
115 }
116 }),
117 );
118 // workerd re-connects for every loopback subrequest once its pooled
119 // connection is closed. Node's default `keepAliveTimeout` of 5s closes
120 // idle sockets between test/request bursts, producing per-burst TCP churn
121 // that piles up TIME_WAIT sockets — on Windows CI this exhausts ephemeral
122 // ports/AFD buffers (WSAENOBUFS #10055, ConnectEx ERROR_DUP_NAME #52).
123 // Keep idle connections around for a minute so they actually get reused.
124 server.keepAliveTimeout = 60_000;
125 // Must stay above `keepAliveTimeout` so Node doesn't tear down a reused
126 // socket while request headers are still arriving.
127 server.headersTimeout = 65_000;
128 yield* Effect.callback<void>((resume) => {
129 server.listen(0, "127.0.0.1", () => resume(Effect.void));
130 });
131 const address = yield* getAddress(server);
132 yield* Effect.addFinalizer(() =>
133 Effect.sync(() => {
134 // `close()` only stops listening; long-lived keep-alive connections
135 // would otherwise linger until the peer closes them.
136 server.close();
137 server.closeAllConnections();
138 }),
139 );
140 const scope = yield* Effect.scope;
141 const makeHandler = yield* Effect.promise(
142 async () =>
143 await import("@effect/platform-node/NodeHttpServer").then(
144 (m) => m.makeHandler,
145 ),
146 );
147 return LoopbackServer.of({
148 address,
149 secret,
150 route: (name, handler) =>
151 Effect.isEffect(handler)
152 ? makeHandler(handler, { scope }).pipe(
153 Effect.map((handler) => {
154 MutableHashMap.set(routes, name, handler);
155 return Effect.void;
156 }),
157 )
158 : Effect.sync(() => {
159 MutableHashMap.set(routes, name, handler);
160 return Effect.void;
161 }),
162 });
163 }),
164);
165
166const timingSafeEqual = (a: string, b: string) => {
167 if (a.length !== b.length) {
168 return false;
169 }
170 return NodeCrypto.timingSafeEqual(Buffer.from(a), Buffer.from(b));
171};
172
173const writeErrorResponse = (
174 res: NodeHttp.ServerResponse,
175 error: RuntimeError,
176 status: number = 500,
177) => {
178 const body = makeErrorEnvelope(error);
179 if (!res.headersSent) {
180 res.writeHead(status, { "content-type": "application/json" });
181 }
182 res.end(JSON.stringify(body));
183};