1/**
2 * Node.js implementation of the Effect `HttpServer`.
3 *
4 * This module adapts a supplied Node `http.Server` into Effect's
5 * platform-independent HTTP server service. It starts the server with Node
6 * `listen` options, converts `request` events into `HttpServerRequest` values,
7 * writes `HttpServerResponse` bodies through Node's `ServerResponse`, and
8 * handles `upgrade` events by exposing the upgraded socket through
9 * `HttpServerRequest.upgrade`. It also exports request and upgrade handler
10 * constructors plus layers for the server alone, HTTP support services, the
11 * combined server, configurable options, and tests.
12 *
13 * @since 4.0.0
14 */
15import * as Cause from "effect/Cause"
16import * as Config from "effect/Config"
17import * as Context from "effect/Context"
18import * as Duration from "effect/Duration"
19import * as Effect from "effect/Effect"
20import * as Exit from "effect/Exit"
21import * as Fiber from "effect/Fiber"
22import type * as FileSystem from "effect/FileSystem"
23import { flow, type LazyArg } from "effect/Function"
24import * as Cookies from "effect/http/Cookies"
25import * as Etag from "effect/http/Etag"
26import * as FetchHttpClient from "effect/http/FetchHttpClient"
27import type * as Headers from "effect/http/Headers"
28import type { HttpClient } from "effect/http/HttpClient"
29import * as HttpEffect from "effect/http/HttpEffect"
30import * as HttpIncomingMessage from "effect/http/HttpIncomingMessage"
31import type { HttpMethod } from "effect/http/HttpMethod"
32import type * as Middleware from "effect/http/HttpMiddleware"
33import type * as HttpPlatform from "effect/http/HttpPlatform"
34import * as HttpServer from "effect/http/HttpServer"
35import {
36 causeResponse,
37 ClientAbort,
38 HttpServerError,
39 RequestParseError,
40 ResponseError,
41 ServeError
42} from "effect/http/HttpServerError"
43import * as Request from "effect/http/HttpServerRequest"
44import { HttpServerRequest } from "effect/http/HttpServerRequest"
45import type { HttpServerResponse } from "effect/http/HttpServerResponse"
46import type * as Multipart from "effect/http/Multipart"
47import * as Latch from "effect/Latch"
48import * as Layer from "effect/Layer"
49import * as NetAddress from "effect/net/NetAddress"
50import type * as Option from "effect/Option"
51import type * as Path from "effect/Path"
52import type * as Record from "effect/Record"
53import * as Scope from "effect/Scope"
54import * as Socket from "effect/socket/Socket"
55import * as Stream from "effect/Stream"
56import * as Http from "node:http"
57import type * as Net from "node:net"
58import type { Duplex } from "node:stream"
59import { Readable } from "node:stream"
60import { pipeline } from "node:stream/promises"
61import { NodeHttpIncomingMessage } from "./NodeHttpIncomingMessage.ts"
62import * as NodeHttpPlatform from "./NodeHttpPlatform.ts"
63import * as NodeMultipart from "./NodeMultipart.ts"
64import * as NodeServices from "./NodeServices.ts"
65import { NodeWS } from "./NodeSocket.ts"
66
67/**
68 * Options accepted by the Node `HttpServer` constructors and layers.
69 *
70 * @category options
71 * @since 4.0.0
72 */
73export interface Options extends Net.ListenOptions {
74 readonly disablePreemptiveShutdown?: boolean | undefined
75 readonly gracefulShutdownTimeout?: Duration.Input | undefined
76 /**
77 * Options forwarded to the underlying `ws` `WebSocketServer`, minus the
78 * wiring options the server manages itself. Use this to enable
79 * `permessage-deflate` compression or tune payload limits, e.g.
80 * `websocket: { perMessageDeflate: true }`.
81 *
82 * @stability unstable
83 */
84 readonly websocket?:
85 | Omit<NodeWS.ServerOptions, "noServer" | "server" | "host" | "port" | "path">
86 | undefined
87}
88
89/**
90 * Creates a scoped `HttpServer` from a Node `http.Server`, starts listening
91 * with the supplied options, registers request and upgrade handling, and closes
92 * the server during scope finalization with optional graceful-shutdown control.
93 *
94 * @category constructors
95 * @since 4.0.0
96 */
97export const make = Effect.fnUntraced(function*(
98 evaluate: LazyArg<Http.Server>,
99 options: Options
100) {
101 const scope = yield* Effect.scope
102 const server = evaluate()
103
104 const shutdown = yield* Effect.callback<void>((resume) => {
105 if (!server.listening) {
106 return resume(Effect.void)
107 }
108 server.close((error) => {
109 if (error) {
110 resume(Effect.die(error))
111 } else {
112 resume(Effect.void)
113 }
114 })
115 }).pipe(Effect.cached)
116
117 const preemptiveShutdown = options.disablePreemptiveShutdown ?
118 Effect.void :
119 Effect.timeoutOrElse(shutdown, {
120 duration: options.gracefulShutdownTimeout ?? Duration.seconds(20),
121 orElse: () => Effect.void
122 })
123
124 yield* Scope.addFinalizer(scope, shutdown)
125
126 yield* Effect.callback<void, ServeError>((resume) => {
127 function onError(cause: Error) {
128 resume(Effect.fail(new ServeError({ cause })))
129 }
130 server.on("error", onError)
131 server.listen(options, () => {
132 server.off("error", onError)
133 resume(Effect.void)
134 })
135 })
136
137 const address = server.address()!
138 const effectAddress = typeof address === "string"
139 ? Effect.succeed(NetAddress.unixPathAddress(address))
140 : Effect.fromResult(NetAddress.inetAddressFromIpString(address.address, address.port)).pipe(
141 Effect.mapError((cause) => new ServeError({ cause }))
142 )
143 const boundAddress = yield* effectAddress
144
145 const wss = yield* Effect.acquireRelease(
146 Effect.sync(() => new NodeWS.WebSocketServer({ ...options.websocket, noServer: true })),
147 (wss) =>
148 Effect.callback<void>((resume) => {
149 wss.close(() => resume(Effect.void))
150 })
151 ).pipe(
152 Scope.provide(scope),
153 Effect.cached
154 )
155
156 return HttpServer.make({
157 address: boundAddress,
158 serve: Effect.fnUntraced(function*(httpApp, middleware) {
159 const serveScope = yield* Effect.scope
160 const scope = Scope.forkUnsafe(serveScope, "parallel")
161 const handler = yield* (makeHandler(httpApp, {
162 middleware: middleware as any,
163 scope
164 }) as Effect.Effect<(nodeRequest: Http.IncomingMessage, nodeResponse: Http.ServerResponse) => void>)
165 const upgradeHandler = yield* makeUpgradeHandler(wss, httpApp, {
166 middleware: middleware as any,
167 scope
168 })
169 yield* Scope.addFinalizerExit(serveScope, () => {
170 server.off("request", handler)
171 server.off("upgrade", upgradeHandler)
172 return preemptiveShutdown
173 })
174 server.on("request", handler)
175 server.on("upgrade", upgradeHandler)
176 })
177 })
178})
179
180/**
181 * Creates a Node `request` event handler for an Effect HTTP application,
182 * injecting a `HttpServerRequest` and interrupting the request fiber if the
183 * client closes the response before it finishes.
184 *
185 * @category handlers
186 * @since 4.0.0
187 */
188export const makeHandler = <
189 R,
190 E,
191 App extends Effect.Effect<HttpServerResponse, any, any> = Effect.Effect<HttpServerResponse, E, R>
192>(
193 httpEffect: Effect.Effect<HttpServerResponse, E, R>,
194 options: {
195 readonly scope: Scope.Scope
196 readonly middleware?: Middleware.HttpMiddleware.Applied<App, E, R> | undefined
197 }
198): Effect.Effect<
199 (nodeRequest: Http.IncomingMessage, nodeResponse: Http.ServerResponse) => void,
200 never,
201 Exclude<Effect.Services<App>, HttpServerRequest | Scope.Scope>
202> => {
203 const handled = HttpEffect.toHandled(httpEffect, handleResponse, options.middleware as any)
204 return Effect.withFiber((parent) => {
205 const services = parent.context
206 return Effect.succeed(function handler(
207 nodeRequest: Http.IncomingMessage,
208 nodeResponse: Http.ServerResponse
209 ) {
210 const context = Context.add(services, HttpServerRequest, new ServerRequestImpl(nodeRequest, nodeResponse))
211 const fiber = Fiber.runIn(Effect.runForkWith(context as Context.Context<any>)(handled), options.scope)
212 if (fiber.pollUnsafe() === undefined) {
213 nodeResponse.on("close", () => {
214 if (!nodeResponse.writableEnded) {
215 fiber.interruptUnsafe(parent.id, ClientAbort.annotation)
216 }
217 })
218 }
219 })
220 })
221}
222
223/**
224 * Creates a Node `upgrade` event handler for an Effect HTTP application,
225 * exposing the upgraded WebSocket as the request's `upgrade` effect and
226 * interrupting the request fiber when the socket closes early.
227 *
228 * @stability unstable
229 * @category handlers
230 * @since 4.0.0
231 */
232export const makeUpgradeHandler = <
233 R,
234 E,
235 App extends Effect.Effect<HttpServerResponse, any, any> = Effect.Effect<HttpServerResponse, E, R>
236>(
237 lazyWss: Effect.Effect<NodeWS.WebSocketServer>,
238 httpEffect: Effect.Effect<HttpServerResponse, E, R>,
239 options: {
240 readonly scope: Scope.Scope
241 readonly middleware?: Middleware.HttpMiddleware.Applied<App, E, R> | undefined
242 }
243): Effect.Effect<
244 (nodeRequest: Http.IncomingMessage, socket: Duplex, head: Buffer) => void,
245 never,
246 Exclude<Effect.Services<App>, HttpServerRequest | Scope.Scope>
247> => {
248 const handledApp = HttpEffect.toHandled(httpEffect, handleResponse, options.middleware as any)
249 return Effect.withFiber((parent) => {
250 const services = parent.context
251 return Effect.succeed(function handler(
252 nodeRequest: Http.IncomingMessage,
253 socket: Duplex,
254 head: Buffer
255 ) {
256 let upgraded = false
257 let nodeResponse_: Http.ServerResponse | undefined = undefined
258 const nodeResponse = () => {
259 if (nodeResponse_ === undefined) {
260 nodeResponse_ = new Http.ServerResponse(nodeRequest)
261 if (upgraded || socket.destroyed) {
262 // End without assigning the socket so handleResponse skips HTTP writes.
263 nodeResponse_.end()
264 } else {
265 nodeResponse_.assignSocket(socket as any)
266 nodeResponse_.on("finish", () => {
267 socket.end()
268 })
269 }
270 }
271 return nodeResponse_
272 }
273 const upgradeEffect = Socket.fromWebSocket(Effect.flatMap(
274 lazyWss,
275 (wss) =>
276 Effect.acquireRelease(
277 Effect.callback<NodeWS.WebSocket, Socket.SocketError>((resume) => {
278 // A refused handshake never invokes the callback, so fail on close instead.
279 const onClose = () =>
280 resume(Effect.fail(
281 new Socket.SocketError({
282 reason: new Socket.SocketOpenError({
283 kind: "Unknown",
284 cause: new Error("The socket closed before the upgrade")
285 })
286 })
287 ))
288 if (socket.destroyed) return onClose()
289 socket.once("close", onClose)
290 wss.handleUpgrade(nodeRequest, socket, head, (ws) => {
291 socket.off("close", onClose)
292 upgraded = true
293 resume(Effect.succeed(ws))
294 })
295 }),
296 (ws, exit) => Effect.sync(() => ws.close(closeCode(exit)))
297 )
298 ))
299 const context = Context.add(
300 services,
301 HttpServerRequest,
302 new ServerRequestImpl(nodeRequest, nodeResponse, upgradeEffect)
303 )
304 const fiber = Fiber.runIn(Effect.runForkWith(context as Context.Context<any>)(handledApp), options.scope)
305 socket.on("error", () => {})
306 socket.on("close", () => {
307 if (!socket.writableEnded) {
308 fiber.interruptUnsafe(parent.id, ClientAbort.annotation)
309 }
310 })
311 })
312 })
313}
314
315class ServerRequestImpl extends NodeHttpIncomingMessage<HttpServerError> implements HttpServerRequest {
316 readonly [Request.TypeId]: typeof Request.TypeId
317 readonly response: Http.ServerResponse | LazyArg<Http.ServerResponse>
318 private upgradeEffect?: Effect.Effect<Socket.Socket, HttpServerError> | undefined
319 readonly url: string
320 private headersOverride?: Headers.Headers | undefined
321
322 constructor(
323 source: Http.IncomingMessage,
324 response: Http.ServerResponse | LazyArg<Http.ServerResponse>,
325 upgradeEffect?: Effect.Effect<Socket.Socket, HttpServerError>,
326 url = source.url!,
327 headersOverride?: Headers.Headers,
328 remoteAddressOverride?: Option.Option<string>
329 ) {
330 super(source, (cause) =>
331 new HttpServerError({
332 reason: new RequestParseError({
333 request: this,
334 cause
335 })
336 }), remoteAddressOverride)
337 this[Request.TypeId] = Request.TypeId
338 this.response = response
339 this.upgradeEffect = upgradeEffect
340 this.url = url
341 this.headersOverride = headersOverride
342 }
343
344 private cachedCookies: Record.ReadonlyRecord<string, string> | undefined
345 get cookies() {
346 if (this.cachedCookies) {
347 return this.cachedCookies
348 }
349 return this.cachedCookies = Cookies.parseHeader(this.headers.cookie ?? "")
350 }
351
352 get resolvedResponse(): Http.ServerResponse {
353 return typeof this.response === "function" ? this.response() : this.response
354 }
355
356 modify(
357 options: {
358 readonly url?: string | undefined
359 readonly headers?: Headers.Headers | undefined
360 readonly remoteAddress?: Option.Option<string> | undefined
361 }
362 ) {
363 return new ServerRequestImpl(
364 this.source,
365 this.response,
366 this.upgradeEffect,
367 options.url ?? this.url,
368 options.headers ?? this.headersOverride,
369 "remoteAddress" in options ? options.remoteAddress : this.remoteAddressOverride
370 )
371 }
372
373 get originalUrl(): string {
374 return this.source.url!
375 }
376
377 private cachedMethod: HttpMethod | undefined
378 get method(): HttpMethod {
379 return this.cachedMethod ??= this.source.method!.toUpperCase() as HttpMethod
380 }
381
382 override get headers(): Headers.Headers {
383 this.headersOverride ??= this.source.headers as Headers.Headers
384 return this.headersOverride
385 }
386
387 private multipartEffect:
388 | Effect.Effect<
389 Multipart.Persisted,
390 Multipart.MultipartError,
391 Scope.Scope | FileSystem.FileSystem | Path.Path
392 >
393 | undefined
394 get multipart(): Effect.Effect<
395 Multipart.Persisted,
396 Multipart.MultipartError,
397 Scope.Scope | FileSystem.FileSystem | Path.Path
398 > {
399 if (this.multipartEffect) {
400 return this.multipartEffect
401 }
402 this.multipartEffect = Effect.runSync(Effect.cached(
403 NodeMultipart.persisted(this.source, this.source.headers)
404 ))
405 return this.multipartEffect
406 }
407
408 get multipartStream(): Stream.Stream<Multipart.Part, Multipart.MultipartError> {
409 return NodeMultipart.stream(this.source, this.source.headers)
410 }
411
412 get upgrade(): Effect.Effect<Socket.Socket, HttpServerError> {
413 return this.upgradeEffect ?? Effect.fail(
414 new HttpServerError({
415 reason: new RequestParseError({
416 request: this,
417 description: "not an upgradeable ServerRequest"
418 })
419 })
420 )
421 }
422
423 override toString(): string {
424 return `ServerRequest(${this.method} ${this.url})`
425 }
426
427 toJSON(): unknown {
428 return HttpIncomingMessage.inspect(this, {
429 _id: "HttpServerRequest",
430 method: this.method,
431 url: this.originalUrl
432 })
433 }
434}
435
436/**
437 * Provides an `HttpServer` by creating and managing a scoped Node
438 * `http.Server` with the supplied listen and shutdown options.
439 *
440 * @category layers
441 * @since 4.0.0
442 */
443export const layerServer: (
444 evaluate: LazyArg<Http.Server<typeof Http.IncomingMessage, typeof Http.ServerResponse>>,
445 options: Options
446) => Layer.Layer<HttpServer.HttpServer, ServeError> = flow(make, Layer.effect(HttpServer.HttpServer))
447
448/**
449 * Provides the Node HTTP support services used by `NodeHttpServer`, including
450 * the HTTP platform, ETag generator, and core Node platform services.
451 *
452 * @category layers
453 * @since 4.0.0
454 */
455export const layerHttpServices: Layer.Layer<
456 NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator
457> = Layer.mergeAll(
458 NodeHttpPlatform.layer,
459 Etag.layerWeak,
460 NodeServices.layer
461)
462
463/**
464 * Provides a Node `HttpServer` together with the Node HTTP platform, ETag, and
465 * core platform services required to serve requests.
466 *
467 * @category layers
468 * @since 4.0.0
469 */
470export const layer = (
471 evaluate: LazyArg<Http.Server>,
472 options: Options
473): Layer.Layer<
474 HttpServer.HttpServer | NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator,
475 ServeError
476> =>
477 Layer.mergeAll(
478 layerServer(evaluate, options),
479 layerHttpServices
480 )
481
482/**
483 * Provides a Node `HttpServer` together with the Node HTTP platform, ETag,
484 * and core Node platform services, reading the listen and shutdown options from
485 * a `Config` value.
486 *
487 * @category layers
488 * @since 4.0.0
489 */
490export const layerConfig = (
491 evaluate: LazyArg<Http.Server>,
492 options: Config.Wrap<Options>
493): Layer.Layer<
494 HttpServer.HttpServer | NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator,
495 ServeError | Config.ConfigError
496> =>
497 Layer.mergeAll(
498 Layer.effect(HttpServer.HttpServer)(
499 Effect.flatMap(Config.unwrap(options), (options) => make(evaluate, options))
500 ),
501 layerHttpServices
502 )
503
504/**
505 * Provides a test HTTP server listening on an ephemeral port together with a
506 * Fetch-backed `HttpClient` configured for server integration tests.
507 *
508 * @category testing
509 * @since 4.0.0
510 */
511export const layerTest: Layer.Layer<
512 | HttpServer.HttpServer
513 | FileSystem.FileSystem
514 | Path.Path
515 | HttpPlatform.HttpPlatform
516 | Etag.Generator
517 | HttpClient,
518 ServeError,
519 never
520> = HttpServer.layerTestClient.pipe(
521 Layer.provide(
522 Layer.fresh(FetchHttpClient.layer).pipe(
523 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false }))
524 )
525 ),
526 Layer.provideMerge(layer(Http.createServer, { port: 0 }))
527)
528
529// -----------------------------------------------------------------------------
530// Internal
531// -----------------------------------------------------------------------------
532
533const closeCode = (exit: Exit.Exit<unknown, unknown>): number =>
534 Exit.isSuccess(exit) ? 1000 : Cause.hasInterruptsOnly(exit.cause) ? 1001 : 1011
535
536const handleResponse = (
537 request: HttpServerRequest,
538 response: HttpServerResponse
539): Effect.Effect<void, HttpServerError> => {
540 const nodeResponse = (request as ServerRequestImpl).resolvedResponse
541 if (nodeResponse.writableEnded) {
542 return Effect.void
543 }
544
545 let headers: Record<string, string | Array<string>> = response.headers
546 if (!Cookies.isEmpty(response.cookies)) {
547 headers = { ...headers }
548 const toSet = Cookies.toSetCookieHeaders(response.cookies)
549 if (headers["set-cookie"] !== undefined) {
550 toSet.push(headers["set-cookie"] as string)
551 }
552 headers["set-cookie"] = toSet
553 }
554
555 if (request.method === "HEAD") {
556 nodeResponse.writeHead(response.status, response.statusText, headers)
557 return Effect.andThen(
558 cancelResponseBody(response.body),
559 Effect.callback<void>((resume) => {
560 let completed = false
561 const done = () => {
562 if (completed) return
563 completed = true
564 nodeResponse.off("close", done)
565 resume(Effect.void)
566 }
567 nodeResponse.once("close", done)
568 nodeResponse.end(done)
569 })
570 )
571 }
572 const body = response.body
573 switch (body._tag) {
574 case "Empty": {
575 nodeResponse.writeHead(response.status, response.statusText, headers)
576 nodeResponse.end()
577 return Effect.void
578 }
579 case "Raw": {
580 nodeResponse.writeHead(response.status, response.statusText, headers)
581 if (
582 typeof body.body === "object" && body.body !== null && "pipe" in body.body &&
583 typeof body.body.pipe === "function"
584 ) {
585 return Effect.tryPromise({
586 try: (signal) => pipeline(body.body as any, nodeResponse, { signal, end: true }),
587 catch: (cause) =>
588 new HttpServerError({
589 reason: new ResponseError({
590 request,
591 response,
592 description: "Error writing raw response",
593 cause
594 })
595 })
596 }).pipe(
597 Effect.interruptible,
598 Effect.tapCause(handleCause(nodeResponse, response))
599 )
600 }
601 return Effect.callback<void>((resume) => {
602 nodeResponse.end(body.body, () => resume(Effect.void))
603 })
604 }
605 case "Uint8Array": {
606 nodeResponse.writeHead(response.status, response.statusText, headers)
607 // If the body is less than 1MB, we skip the callback
608 if (body.contentLength < 1024 * 1024) {
609 // Writing text directly lets Node flush headers and body together.
610 nodeResponse.end(body.text ?? body.body)
611 return Effect.void
612 }
613 return Effect.callback<void>((resume) => {
614 nodeResponse.end(body.text ?? body.body, () => resume(Effect.void))
615 })
616 }
617 case "FormData": {
618 return Effect.suspend(() => {
619 const r = new globalThis.Response(body.formData)
620 nodeResponse.writeHead(response.status, response.statusText, {
621 ...headers,
622 ...Object.fromEntries(r.headers)
623 })
624 return Effect.callback<void, HttpServerError>((resume, signal) => {
625 Readable.fromWeb(r.body as any, { signal })
626 .pipe(nodeResponse)
627 .on("error", (cause) => {
628 resume(Effect.fail(
629 new HttpServerError({
630 reason: new ResponseError({
631 request,
632 response,
633 description: "Error writing FormData response",
634 cause
635 })
636 })
637 ))
638 })
639 .once("finish", () => {
640 resume(Effect.void)
641 })
642 }).pipe(
643 Effect.interruptible,
644 Effect.tapCause(handleCause(nodeResponse, response))
645 )
646 })
647 }
648 case "Stream": {
649 nodeResponse.writeHead(response.status, response.statusText, headers)
650 const drainLatch = Latch.makeUnsafe()
651 nodeResponse.on("drain", () => drainLatch.openUnsafe())
652 return body.stream.pipe(
653 Stream.orDie,
654 Stream.runForEachArray((array) => {
655 const chunk = array.length > 1 ? Buffer.concat(array) : array[0]
656 if (nodeResponse.write(chunk)) return Effect.void
657 drainLatch.closeUnsafe()
658 return drainLatch.await
659 }),
660 Effect.interruptible,
661 Effect.matchCauseEffect({
662 onSuccess: () => Effect.sync(() => nodeResponse.end()),
663 onFailure: handleCause(nodeResponse, response)
664 })
665 )
666 }
667 }
668}
669
670const cancelResponseBody = (body: HttpServerResponse["body"]): Effect.Effect<void> => {
671 const stream = body._tag === "Raw" ? body.body : undefined
672 if (stream instanceof Readable) {
673 return Effect.sync(() => stream.destroy())
674 }
675 return Effect.void
676}
677
678const handleCause = (
679 nodeResponse: Http.ServerResponse,
680 originalResponse: HttpServerResponse
681) =>
682<E>(originalCause: Cause.Cause<E>) =>
683 Effect.flatMap(causeResponse(originalCause), ([response, cause]) => {
684 const headersSent = nodeResponse.headersSent
685 if (!headersSent) {
686 nodeResponse.writeHead(response.status, response.statusText)
687 }
688 if (!nodeResponse.writableEnded) {
689 nodeResponse.end()
690 }
691 return Effect.failCause(
692 headersSent
693 ? Cause.combine(originalCause, Cause.die(originalResponse))
694 : cause
695 )
696 })