1/**
2 * Node.js implementations of the Effect `HttpClient`.
3 *
4 * This module supplies Node runtime backends for the platform-independent
5 * Effect HTTP client API. It re-exports the fetch-based `Fetch`, `RequestInit`,
6 * and `layerFetch` APIs, defines an Undici-backed client with dispatcher
7 * services and request options, and defines a lower-level `node:http` /
8 * `node:https` client with scoped HTTP agent layers.
9 *
10 * @since 4.0.0
11 */
12import * as Context from "effect/Context"
13import * as Effect from "effect/Effect"
14import { flow } from "effect/Function"
15import * as Cookies from "effect/http/Cookies"
16import * as Headers from "effect/http/Headers"
17import type * as Body from "effect/http/HttpBody"
18import * as Client from "effect/http/HttpClient"
19import * as Error from "effect/http/HttpClientError"
20import type { HttpClientRequest } from "effect/http/HttpClientRequest"
21import * as Response from "effect/http/HttpClientResponse"
22import type { HttpClientResponse } from "effect/http/HttpClientResponse"
23import * as IncomingMessage from "effect/http/HttpIncomingMessage"
24import * as UrlParams from "effect/http/UrlParams"
25import * as Inspectable from "effect/Inspectable"
26import * as Layer from "effect/Layer"
27import * as Option from "effect/Option"
28import { type Pipeable, pipeArguments } from "effect/Pipeable"
29import type * as Schema from "effect/Schema"
30import type * as Scope from "effect/Scope"
31import * as Stream from "effect/Stream"
32import * as Http from "node:http"
33import * as Https from "node:https"
34import { Readable } from "node:stream"
35import { pipeline } from "node:stream/promises"
36import { NodeHttpIncomingMessage } from "./NodeHttpIncomingMessage.ts"
37import * as NodeSink from "./NodeSink.ts"
38import * as NodeStream from "./NodeStream.ts"
39import type * as Undici from "./Undici.ts"
40
41// -----------------------------------------------------------------------------
42// Fetch
43// -----------------------------------------------------------------------------
44
45export {
46 /**
47 * Provides a fetch-based HTTP client implementation for Node.js.
48 *
49 * **When to use**
50 *
51 * Use to access or override the fetch implementation used by the Node
52 * fetch-based HTTP client.
53 *
54 * @category services
55 * @since 4.0.0
56 */
57 Fetch,
58 /**
59 * Layer that provides the fetch-based HTTP client implementation.
60 *
61 * @category layers
62 * @since 4.0.0
63 */
64 layer as layerFetch,
65 /**
66 * Provides request initialization options accepted by the fetch-based HTTP client.
67 *
68 * **When to use**
69 *
70 * Use to provide default fetch request options for Node HTTP requests.
71 *
72 * @category services
73 * @since 4.0.0
74 */
75 RequestInit
76} from "effect/http/FetchHttpClient"
77
78// -----------------------------------------------------------------------------
79// Undici
80// -----------------------------------------------------------------------------
81
82/**
83 * Service tag for the Undici `Dispatcher` used by the Undici-backed HTTP
84 * client.
85 *
86 * @stability unstable
87 * @category services
88 * @since 4.0.0
89 */
90export class Dispatcher extends Context.Service<Dispatcher, Undici.Dispatcher>()(
91 "@effect/platform-node/NodeHttpClient/Dispatcher"
92) {}
93
94const loadUndici = Effect.promise(() => import("./Undici.ts"))
95
96/**
97 * Acquires a new Undici `Agent` dispatcher and destroys it when the enclosing
98 * scope is finalized.
99 *
100 * @stability unstable
101 * @category resource management
102 * @since 4.0.0
103 */
104export const makeDispatcher: Effect.Effect<Undici.Dispatcher, never, Scope.Scope> = Effect.acquireRelease(
105 Effect.map(loadUndici, (_) => new _.Agent()),
106 (dispatcher) => Effect.promise(() => dispatcher.destroy())
107)
108
109/**
110 * Provides the `Dispatcher` service using a scoped Undici `Agent`.
111 *
112 * @stability unstable
113 * @category layers
114 * @since 4.0.0
115 */
116export const layerDispatcher: Layer.Layer<Dispatcher> = Layer.effect(Dispatcher)(makeDispatcher)
117
118/**
119 * Provides the `Dispatcher` service from Undici's process-global dispatcher,
120 * without creating or owning a new agent.
121 *
122 * @stability unstable
123 * @category layers
124 * @since 4.0.0
125 */
126export const dispatcherLayerGlobal: Layer.Layer<Dispatcher> = Layer.effect(Dispatcher)(
127 Effect.map(loadUndici, (_) => _.getGlobalDispatcher())
128)
129
130/**
131 * Fiber reference containing default Undici request options applied to requests
132 * sent by `makeUndici`.
133 *
134 * @stability unstable
135 * @category services
136 * @since 4.0.0
137 */
138export const UndiciOptions = Context.Reference<Partial<Undici.Dispatcher.RequestOptions>>(
139 "@effect/platform-node/NodeHttpClient/UndiciOptions",
140 { defaultValue: () => ({}) }
141)
142
143/**
144 * Creates an `HttpClient` that sends requests through the current Undici
145 * `Dispatcher`, converts Effect HTTP bodies to Undici bodies, and maps
146 * transport and decode failures to `HttpClientError`.
147 *
148 * @stability unstable
149 * @category constructors
150 * @since 4.0.0
151 */
152export const makeUndici = Effect.gen(function*() {
153 const dispatcher = yield* Dispatcher
154 return Client.make((request, url, signal, fiber) =>
155 convertBody(request.body).pipe(
156 Effect.flatMap((body) =>
157 Effect.tryPromise({
158 try: () =>
159 dispatcher.request({
160 ...fiber.getRef(UndiciOptions),
161 signal,
162 method: request.method,
163 headers: request.headers,
164 origin: url.origin,
165 path: url.pathname + url.search,
166 body,
167 // leave timeouts to Effect.timeout etc
168 headersTimeout: 60 * 60 * 1000,
169 bodyTimeout: 0
170 }),
171 catch: (cause) =>
172 new Error.HttpClientError({
173 reason: new Error.TransportError({
174 request,
175 cause
176 })
177 })
178 })
179 ),
180 Effect.map((response) => new UndiciResponse(request, response, url.href.split("#")[0]))
181 )
182 )
183})
184
185function convertBody(
186 body: Body.HttpBody
187): Effect.Effect<Exclude<Undici.Dispatcher.DispatchOptions["body"], undefined>> {
188 switch (body._tag) {
189 case "Empty": {
190 return Effect.succeed(null)
191 }
192 case "Uint8Array":
193 case "Raw": {
194 return Effect.succeed(body.body as Uint8Array)
195 }
196 case "FormData": {
197 return Effect.succeed(body.formData as any)
198 }
199 case "Stream": {
200 return NodeStream.toReadable(body.stream)
201 }
202 }
203}
204
205function noopErrorHandler(_: any) {}
206
207class UndiciResponse extends Inspectable.Class implements HttpClientResponse, Pipeable {
208 readonly [IncomingMessage.TypeId]: typeof IncomingMessage.TypeId
209 readonly [Response.TypeId]: typeof Response.TypeId
210 readonly request: HttpClientRequest
211 readonly source: Undici.Dispatcher.ResponseData
212 readonly url: string
213
214 constructor(
215 request: HttpClientRequest,
216 source: Undici.Dispatcher.ResponseData,
217 url: string
218 ) {
219 super()
220 this[IncomingMessage.TypeId] = IncomingMessage.TypeId
221 this[Response.TypeId] = Response.TypeId
222 this.request = request
223 this.source = source
224 this.url = url
225 source.body.on("error", noopErrorHandler)
226 }
227
228 get status() {
229 return this.source.statusCode!
230 }
231
232 get statusText() {
233 return undefined
234 }
235
236 get headers(): Headers.Headers {
237 return Headers.fromInput(this.source.headers)
238 }
239
240 cachedCookies?: Cookies.Cookies
241 get cookies(): Cookies.Cookies {
242 if (this.cachedCookies !== undefined) {
243 return this.cachedCookies
244 }
245 const header = this.source.headers["set-cookie"]
246 return this.cachedCookies = header ? Cookies.fromSetCookie(header) : Cookies.empty
247 }
248
249 get remoteAddress(): Option.Option<string> {
250 return Option.none()
251 }
252
253 get stream(): Stream.Stream<Uint8Array, Error.HttpClientError> {
254 return NodeStream.fromReadable({
255 evaluate: () => this.source.body,
256 onError: (cause) =>
257 new Error.HttpClientError({
258 reason: new Error.DecodeError({
259 request: this.request,
260 response: this,
261 cause
262 })
263 })
264 })
265 }
266
267 get json(): Effect.Effect<Schema.Json, Error.HttpClientError> {
268 return Effect.flatMap(this.text, (text) =>
269 Effect.try({
270 try: () => text === "" ? null : JSON.parse(text),
271 catch: (cause) =>
272 new Error.HttpClientError({
273 reason: new Error.DecodeError({
274 request: this.request,
275 response: this,
276 cause
277 })
278 })
279 }))
280 }
281
282 private textBody?: Effect.Effect<string, Error.HttpClientError>
283 get text(): Effect.Effect<string, Error.HttpClientError> {
284 if (this.textBody) {
285 return this.textBody
286 }
287 this.textBody = Effect.map(this.arrayBuffer, (_) => new TextDecoder().decode(_))
288 return this.textBody
289 }
290
291 get urlParamsBody(): Effect.Effect<UrlParams.UrlParams, Error.HttpClientError> {
292 return Effect.flatMap(this.text, (_) =>
293 Effect.try({
294 try: () => UrlParams.fromInput(new URLSearchParams(_)),
295 catch: (cause) =>
296 new Error.HttpClientError({
297 reason: new Error.DecodeError({
298 request: this.request,
299 response: this,
300 cause
301 })
302 })
303 }))
304 }
305
306 private formDataBody?: Effect.Effect<FormData, Error.HttpClientError>
307 get formData(): Effect.Effect<FormData, Error.HttpClientError> {
308 return this.formDataBody ??= Effect.flatMap(this.arrayBuffer, (body) =>
309 Effect.tryPromise({
310 try: () => new globalThis.Response(body, { headers: this.headers }).formData(),
311 catch: (cause) =>
312 new Error.HttpClientError({
313 reason: new Error.DecodeError({
314 request: this.request,
315 response: this,
316 cause
317 })
318 })
319 })).pipe(Effect.cached, Effect.runSync)
320 }
321
322 private arrayBufferBody?: Effect.Effect<ArrayBuffer, Error.HttpClientError>
323 get arrayBuffer(): Effect.Effect<ArrayBuffer, Error.HttpClientError> {
324 if (this.arrayBufferBody) {
325 return this.arrayBufferBody
326 }
327 this.arrayBufferBody = Effect.tryPromise({
328 try: () => this.source.body.arrayBuffer(),
329 catch: (cause) =>
330 new Error.HttpClientError({
331 reason: new Error.DecodeError({
332 request: this.request,
333 response: this,
334 cause
335 })
336 })
337 }).pipe(Effect.cached, Effect.runSync)
338 this.textBody = Effect.map(this.arrayBufferBody, (_) => new TextDecoder().decode(_))
339 return this.arrayBufferBody
340 }
341
342 toJSON(): unknown {
343 return IncomingMessage.inspect(this, {
344 _id: "effect/http/HttpClientResponse",
345 request: this.request.toJSON(),
346 status: this.status
347 })
348 }
349
350 pipe() {
351 return pipeArguments(this, arguments)
352 }
353}
354
355/**
356 * Provides an Undici-backed `HttpClient` using the current `Dispatcher`
357 * service.
358 *
359 * @stability unstable
360 * @category layers
361 * @since 4.0.0
362 */
363export const layerUndiciNoDispatcher: Layer.Layer<
364 Client.HttpClient,
365 never,
366 Dispatcher
367> = Client.layerMergedContext(makeUndici)
368
369/**
370 * Provides an Undici-backed `HttpClient` together with a scoped default
371 * Undici `Agent` dispatcher.
372 *
373 * @stability unstable
374 * @category layers
375 * @since 4.0.0
376 */
377export const layerUndici: Layer.Layer<Client.HttpClient> = Layer.provide(layerUndiciNoDispatcher, layerDispatcher)
378
379// -----------------------------------------------------------------------------
380// node:http
381// -----------------------------------------------------------------------------
382
383/**
384 * Service tag for the paired Node `http` and `https` agents used by the
385 * node:http-backed HTTP client.
386 *
387 * @category services
388 * @since 4.0.0
389 */
390export class HttpAgent extends Context.Service<HttpAgent, {
391 readonly http: Http.Agent
392 readonly https: Https.Agent
393}>()("@effect/platform-node/NodeHttpClient/HttpAgent") {}
394
395/**
396 * Acquires Node `http` and `https` agents with the supplied options and
397 * destroys both agents when the enclosing scope is finalized.
398 *
399 * @category resource management
400 * @since 4.0.0
401 */
402export const makeAgent = (options?: Https.AgentOptions): Effect.Effect<HttpAgent["Service"], never, Scope.Scope> =>
403 Effect.zipWith(
404 Effect.acquireRelease(
405 Effect.sync(() => new Http.Agent(options)),
406 (agent) => Effect.sync(() => agent.destroy())
407 ),
408 Effect.acquireRelease(
409 Effect.sync(() => new Https.Agent(options)),
410 (agent) => Effect.sync(() => agent.destroy())
411 ),
412 (http, https) => ({ http, https })
413 )
414
415/**
416 * Provides the `HttpAgent` service using scoped Node `http` and `https`
417 * agents configured with the supplied options.
418 *
419 * @category layers
420 * @since 4.0.0
421 */
422export const layerAgentOptions: (options?: Https.AgentOptions | undefined) => Layer.Layer<
423 HttpAgent
424> = flow(makeAgent, Layer.effect(HttpAgent))
425
426/**
427 * Provides the `HttpAgent` service using default scoped Node `http` and
428 * `https` agents.
429 *
430 * @category layers
431 * @since 4.0.0
432 */
433export const layerAgent: Layer.Layer<HttpAgent> = layerAgentOptions()
434
435/**
436 * Creates an `HttpClient` backed by Node `http` and `https`, using the
437 * current `HttpAgent`, streaming request bodies, and wrapping Node responses
438 * as `HttpClientResponse` values.
439 *
440 * @category constructors
441 * @since 4.0.0
442 */
443export const makeNodeHttp = Effect.gen(function*() {
444 const agent = yield* HttpAgent
445 return Client.make((request, url, signal) => {
446 const nodeRequest = url.protocol === "https:" ?
447 Https.request(url, {
448 agent: agent.https,
449 method: request.method,
450 headers: request.headers,
451 signal
452 }) :
453 Http.request(url, {
454 agent: agent.http,
455 method: request.method,
456 headers: request.headers,
457 signal
458 })
459 return Effect.raceFirst(
460 waitForResponse(nodeRequest, request),
461 sendBody(nodeRequest, request, request.body).pipe(Effect.andThen(Effect.never))
462 ).pipe(
463 Effect.onError(() => Effect.sync(() => nodeRequest.destroy())),
464 Effect.map((_) => new NodeHttpResponse(request, _, url.href.split("#")[0]))
465 )
466 })
467})
468
469const sendBody = (
470 nodeRequest: Http.ClientRequest,
471 request: HttpClientRequest,
472 body: Body.HttpBody
473): Effect.Effect<void, Error.HttpClientError> =>
474 Effect.suspend((): Effect.Effect<void, Error.HttpClientError> => {
475 switch (body._tag) {
476 case "Empty": {
477 nodeRequest.end()
478 return waitForFinish(nodeRequest, request)
479 }
480 case "Uint8Array":
481 case "Raw": {
482 nodeRequest.end(body.body)
483 return waitForFinish(nodeRequest, request)
484 }
485 case "FormData": {
486 const response = new globalThis.Response(body.formData)
487
488 response.headers.forEach((value, key) => {
489 nodeRequest.setHeader(key, value)
490 })
491
492 return Effect.tryPromise({
493 try: () => pipeline(Readable.fromWeb(response.body! as any), nodeRequest),
494 catch: (cause) =>
495 new Error.HttpClientError({
496 reason: new Error.TransportError({
497 request,
498 cause
499 })
500 })
501 })
502 }
503 case "Stream": {
504 return Stream.run(
505 Stream.mapError(body.stream, (cause) =>
506 new Error.HttpClientError({
507 reason: new Error.EncodeError({
508 request,
509 cause
510 })
511 })),
512 NodeSink.fromWritable({
513 evaluate: () => nodeRequest,
514 onError: (cause) =>
515 new Error.HttpClientError({
516 reason: new Error.TransportError({
517 request,
518 cause
519 })
520 })
521 })
522 )
523 }
524 }
525 })
526
527const waitForResponse = (nodeRequest: Http.ClientRequest, request: HttpClientRequest) =>
528 Effect.callback<Http.IncomingMessage, Error.HttpClientError>((resume) => {
529 function onError(cause: Error) {
530 resume(Effect.fail(
531 new Error.HttpClientError({
532 reason: new Error.TransportError({
533 request,
534 cause
535 })
536 })
537 ))
538 }
539 nodeRequest.on("error", onError)
540
541 function onResponse(response: Http.IncomingMessage) {
542 nodeRequest.off("error", onError)
543 resume(Effect.succeed(response))
544 }
545 nodeRequest.on("upgrade", onResponse)
546 nodeRequest.on("response", onResponse)
547
548 return Effect.sync(() => {
549 nodeRequest.off("error", onError)
550 nodeRequest.off("upgrade", onResponse)
551 nodeRequest.off("response", onResponse)
552 })
553 })
554
555const waitForFinish = (nodeRequest: Http.ClientRequest, request: HttpClientRequest) =>
556 Effect.callback<void, Error.HttpClientError>((resume) => {
557 function onError(cause: Error) {
558 resume(Effect.fail(
559 new Error.HttpClientError({
560 reason: new Error.TransportError({
561 request,
562 cause
563 })
564 })
565 ))
566 }
567 nodeRequest.once("error", onError)
568
569 function onFinish() {
570 nodeRequest.off("error", onError)
571 resume(Effect.void)
572 }
573 nodeRequest.once("finish", onFinish)
574
575 return Effect.sync(() => {
576 nodeRequest.off("error", onError)
577 nodeRequest.off("finish", onFinish)
578 })
579 })
580
581class NodeHttpResponse extends NodeHttpIncomingMessage<Error.HttpClientError> implements HttpClientResponse, Pipeable {
582 readonly [Response.TypeId]: typeof Response.TypeId
583 readonly request: HttpClientRequest
584 readonly url: string
585
586 constructor(
587 request: HttpClientRequest,
588 source: Http.IncomingMessage,
589 url: string
590 ) {
591 super(source, (cause) =>
592 new Error.HttpClientError({
593 reason: new Error.DecodeError({
594 request,
595 response: this,
596 cause
597 })
598 }))
599 this[Response.TypeId] = Response.TypeId
600 this.request = request
601 this.url = url
602 }
603
604 get status() {
605 return this.source.statusCode!
606 }
607
608 cachedCookies?: Cookies.Cookies
609 get cookies(): Cookies.Cookies {
610 if (this.cachedCookies !== undefined) {
611 return this.cachedCookies
612 }
613 const header = this.source.headers["set-cookie"]
614 return this.cachedCookies = header ? Cookies.fromSetCookie(header) : Cookies.empty
615 }
616
617 get formData(): Effect.Effect<FormData, Error.HttpClientError> {
618 return Effect.tryPromise({
619 try: () => {
620 const init: {
621 headers: HeadersInit
622 status?: number
623 statusText?: string
624 } = {
625 headers: new globalThis.Headers(this.source.headers as any)
626 }
627
628 if (this.source.statusCode) {
629 init.status = this.source.statusCode
630 }
631
632 if (this.source.statusMessage) {
633 init.statusText = this.source.statusMessage
634 }
635
636 return new globalThis.Response(Readable.toWeb(this.source) as any, init).formData()
637 },
638 catch: this.onError
639 })
640 }
641
642 toJSON(): unknown {
643 return IncomingMessage.inspect(this, {
644 _id: "effect/http/HttpClientResponse",
645 request: this.request.toJSON(),
646 status: this.status
647 })
648 }
649
650 pipe() {
651 return pipeArguments(this, arguments)
652 }
653}
654
655/**
656 * Provides a node:http-backed `HttpClient` using the current `HttpAgent`
657 * service.
658 *
659 * @category layers
660 * @since 4.0.0
661 */
662export const layerNodeHttpNoAgent: Layer.Layer<
663 Client.HttpClient,
664 never,
665 HttpAgent
666> = Client.layerMergedContext(makeNodeHttp)
667
668/**
669 * Provides a node:http-backed `HttpClient` together with default scoped Node
670 * `http` and `https` agents.
671 *
672 * @category layers
673 * @since 4.0.0
674 */
675export const layerNodeHttp: Layer.Layer<Client.HttpClient> = Layer.provide(layerNodeHttpNoAgent, layerAgent)