1/**
2 * Node.js implementation of the Effect HTTP platform service.
3 *
4 * This module connects the portable `HttpPlatform` file response helpers to
5 * Node runtime primitives. It serves local files through Node readable streams,
6 * supports byte ranges, converts Web `File` values to readable streams, and
7 * fills in content type and content length headers when needed.
8 *
9 * @since 4.0.0
10 */
11import * as NodeHttpCompression from "@effect/platform-node-shared/NodeHttpCompression"
12import * as Effect from "effect/Effect"
13import { pipe } from "effect/Function"
14import * as EtagImpl from "effect/http/Etag"
15import * as Headers from "effect/http/Headers"
16import * as HttpBody from "effect/http/HttpBody"
17import * as Platform from "effect/http/HttpPlatform"
18import * as ServerResponse from "effect/http/HttpServerResponse"
19import * as Mime from "effect/http/Mime"
20import * as Layer from "effect/Layer"
21import * as Option from "effect/Option"
22import * as Fs from "node:fs"
23import { pipeline, Readable } from "node:stream"
24import * as NodeFileSystem from "./NodeFileSystem.ts"
25import * as NodeStream from "./NodeStream.ts"
26
27// Keep the native Raw/pipeline path without opening discarded response bodies.
28const lazyReadable = (evaluate: () => Readable, onDestroy?: () => void): Readable => {
29 let source: Readable | undefined
30 return new Readable({
31 read() {
32 if (source === undefined) {
33 try {
34 source = evaluate()
35 } catch (cause) {
36 this.destroy(cause instanceof Error ? cause : new Error(String(cause)))
37 return
38 }
39 source.on("data", (chunk) => {
40 if (!this.push(chunk)) source!.pause()
41 })
42 source.once("end", () => this.push(null))
43 source.once("error", (cause) => this.destroy(cause))
44 source.once("close", () => {
45 if (!source!.readableEnded) this.destroy(new Error("Response stream closed prematurely"))
46 })
47 }
48 source.resume()
49 },
50 destroy(error, callback) {
51 source?.destroy(error ?? undefined)
52 onDestroy?.()
53 callback(error)
54 }
55 })
56}
57
58// replaces the response body while keeping every other field, dropping the
59// now-stale Content-Length header
60const compressedBody = (
61 response: ServerResponse.HttpServerResponse,
62 body: HttpBody.HttpBody
63): ServerResponse.HttpServerResponse =>
64 ServerResponse.removeHeader(ServerResponse.setBody(response, body), "content-length")
65
66const compression = NodeHttpCompression.make({
67 algorithms: NodeHttpCompression.algorithms,
68 compressResponse(response, algorithm, options) {
69 const body = response.body
70 switch (body._tag) {
71 case "Stream": {
72 return Effect.succeed(compressedBody(
73 response,
74 HttpBody.stream(
75 NodeStream.pipeThroughDuplex(body.stream, {
76 evaluate: () => NodeHttpCompression.compressTransform(algorithm, options)
77 }),
78 response.headers["content-type"] ?? body.contentType
79 )
80 ))
81 }
82 case "Raw": {
83 let readable = body.body instanceof Readable ? body.body : undefined
84 const compressed = lazyReadable(() => {
85 const transform = NodeHttpCompression.compressTransform(algorithm, options)
86 readable ??= Readable.fromWeb(new Response(body.body as BodyInit).body as any)
87 return pipeline(readable, transform, (cause) => {
88 if (cause) compressed.destroy(cause)
89 })
90 }, () => readable?.destroy())
91 return Effect.succeed(
92 compressedBody(
93 response,
94 HttpBody.raw(compressed, {
95 contentType: response.headers["content-type"] ?? body.contentType
96 })
97 )
98 )
99 }
100 default: {
101 return Effect.succeed(response)
102 }
103 }
104 }
105})
106
107/**
108 * Creates the Node `HttpPlatform`, serving file responses from Node readable
109 * streams and adding MIME type and content-length headers when needed.
110 *
111 * @category constructors
112 * @since 4.0.0
113 */
114export const make = Platform.make({
115 platform: "node",
116 compression,
117 fileResponse(path, status, statusText, headers, start, end, contentLength) {
118 const stream = lazyReadable(() =>
119 contentLength === BigInt(0)
120 ? Readable.from([])
121 : Fs.createReadStream(path, { start, end: end === undefined ? undefined : end - 1 })
122 )
123 return ServerResponse.raw(stream, {
124 headers: {
125 ...headers,
126 "content-type": headers["content-type"] ??
127 Option.getOrElse(Mime.getType(path), () => "application/octet-stream"),
128 "content-length": contentLength.toString()
129 },
130 status,
131 statusText
132 })
133 },
134 fileWebResponse(file, status, statusText, headers, _options) {
135 return ServerResponse.raw(
136 lazyReadable(() => Readable.fromWeb(file.stream() as any)),
137 {
138 headers: Headers.merge(
139 headers,
140 Headers.fromRecordUnsafe({
141 "content-type": headers["content-type"] ??
142 (file.type === ""
143 ? Option.getOrElse(Mime.getType(file.name), () => "application/octet-stream")
144 : file.type),
145 "content-length": file.size.toString()
146 })
147 ),
148 status,
149 statusText
150 }
151 )
152 }
153})
154
155/**
156 * Provides the Node `HttpPlatform` together with the filesystem and ETag
157 * services it needs for file responses.
158 *
159 * @category layers
160 * @since 4.0.0
161 */
162export const layer: Layer.Layer<Platform.HttpPlatform> = pipe(
163 Layer.effect(Platform.HttpPlatform)(make),
164 Layer.provide(NodeFileSystem.layer),
165 Layer.provide(EtagImpl.layer)
166)