1/**
2 * Native Deno implementation of the Effect `HttpServer`.
3 *
4 * @since 4.0.0
5 */
6import * as Config from "effect/Config"
7import type { ConfigError } from "effect/Config"
8import * as Context from "effect/Context"
9import * as Duration from "effect/Duration"
10import * as Effect from "effect/Effect"
11import * as Fiber from "effect/Fiber"
12import type * as FileSystem from "effect/FileSystem"
13import { flow } from "effect/Function"
14import * as Cookies from "effect/http/Cookies"
15import * as Etag from "effect/http/Etag"
16import * as FetchHttpClient from "effect/http/FetchHttpClient"
17import * as Headers from "effect/http/Headers"
18import type * as HttpBody from "effect/http/HttpBody"
19import type { HttpClient } from "effect/http/HttpClient"
20import * as HttpEffect from "effect/http/HttpEffect"
21import * as IncomingMessage from "effect/http/HttpIncomingMessage"
22import type { HttpMethod } from "effect/http/HttpMethod"
23import type { HttpPlatform } from "effect/http/HttpPlatform"
24import * as Server from "effect/http/HttpServer"
25import * as Error from "effect/http/HttpServerError"
26import * as ServerRequest from "effect/http/HttpServerRequest"
27import * as ServerResponse from "effect/http/HttpServerResponse"
28import type * as Multipart from "effect/http/Multipart"
29import * as UrlParams from "effect/http/UrlParams"
30import * as Inspectable from "effect/Inspectable"
31import * as Layer from "effect/Layer"
32import * as NetAddress from "effect/net/NetAddress"
33import * as Option from "effect/Option"
34import type * as Path from "effect/Path"
35import type * as Record from "effect/Record"
36import type * as Schema from "effect/Schema"
37import * as Scope from "effect/Scope"
38import * as Socket from "effect/socket/Socket"
39import * as Stream from "effect/Stream"
40import * as Platform from "./DenoHttpPlatform.ts"
41import * as DenoMultipart from "./DenoMultipart.ts"
42import * as DenoServices from "./DenoServices.ts"
43
44/**
45 * Native Deno TCP, TLS, or Unix serve options managed by the scoped server.
46 *
47 * **Details**
48 *
49 * `signal` and `onError` are omitted because the scope and Effect HTTP handler
50 * own server and failure lifecycles. Server constructors additionally accept
51 * WebSocket settings that apply to every upgrade; `protocol` therefore cannot
52 * vary per request.
53 *
54 * @category options
55 * @since 4.0.0
56 */
57export type ServeOptions =
58 | (Omit<Deno.ServeTcpOptions, "signal" | "onError"> & Partial<Deno.TlsCertifiedKeyPem>)
59 | Omit<Deno.ServeUnixOptions, "signal" | "onError">
60
61/**
62 * Creates a scoped native Deno HTTP server.
63 *
64 * @category constructors
65 * @since 4.0.0
66 */
67export const make = Effect.fnUntraced(function*(
68 options: ServeOptions & {
69 readonly disablePreemptiveShutdown?: boolean | undefined
70 readonly gracefulShutdownTimeout?: Duration.Input | undefined
71 readonly websocket?: Deno.UpgradeWebSocketOptions | undefined
72 }
73) {
74 const scope = yield* Effect.scope
75 type Handler = (request: Request, info: Deno.ServeHandlerInfo<Deno.NetAddr | Deno.UnixAddr>) => Promise<Response>
76 const notFound: Handler = (_request, _info) => Promise.resolve(new Response("not found", { status: 404 }))
77 const handlerStack: Array<Handler> = [notFound]
78 let currentHandler = notFound
79 const {
80 disablePreemptiveShutdown,
81 gracefulShutdownTimeout,
82 websocket,
83 ...serveOptions
84 } = options
85 const server = Deno.serve(
86 serveOptions as Deno.ServeTcpOptions,
87 (request, info) => currentHandler(request, info)
88 ) as Deno.HttpServer<Deno.NetAddr | Deno.UnixAddr>
89
90 const shutdown = yield* Effect.promise(() => server.shutdown()).pipe(Effect.cached)
91 const preemptiveShutdown = disablePreemptiveShutdown ? Effect.void : Effect.timeoutOrElse(shutdown, {
92 duration: gracefulShutdownTimeout ?? Duration.seconds(20),
93 orElse: () => Effect.void
94 })
95
96 yield* Scope.addFinalizer(scope, shutdown)
97
98 const serverAddress = server.addr
99 const address = yield* (serverAddress.transport === "unix"
100 ? Effect.succeed(NetAddress.unixPathAddress(serverAddress.path))
101 : Effect.fromResult(NetAddress.inetAddressFromIpString(
102 (serverAddress as Deno.NetAddr).hostname,
103 (serverAddress as Deno.NetAddr).port
104 )).pipe(Effect.mapError((cause) => new Error.ServeError({ cause }))))
105
106 return Server.make({
107 address,
108 serve: Effect.fnUntraced(function*(httpApp, middleware) {
109 const parent = yield* Effect.fiber
110 const services = parent.context
111 const serveScope = Context.getUnsafe(services, Scope.Scope)
112 const scope = Scope.forkUnsafe(serveScope, "parallel")
113
114 const httpEffect = HttpEffect.toHandled(httpApp, (request, response) => {
115 const denoRequest = request as DenoServerRequest
116 if (denoRequest.upgraded) return cancelResponseBody(response.body)
117 return Effect.flatMap(
118 makeResponse(request, response, services, scope),
119 (response) => Effect.sync(() => denoRequest.resolve(response))
120 )
121 }, middleware)
122
123 function handler(
124 request: Request,
125 info: Deno.ServeHandlerInfo<Deno.NetAddr | Deno.UnixAddr>
126 ): Promise<Response> {
127 return new Promise<Response>((resolve) => {
128 const context = Context.add(
129 services,
130 ServerRequest.HttpServerRequest,
131 new DenoServerRequest(request, info.remoteAddr, resolve, removeHost(request.url), websocket)
132 )
133 const fiber = Fiber.runIn(Effect.runForkWith(context)(httpEffect), scope)
134 request.signal.addEventListener("abort", () => {
135 fiber.interruptUnsafe(parent.id, Error.ClientAbort.annotation)
136 }, { once: true })
137 })
138 }
139
140 yield* Scope.addFinalizerExit(serveScope, () => {
141 const index = handlerStack.lastIndexOf(handler)
142 if (index !== -1) handlerStack.splice(index, 1)
143 currentHandler = handlerStack[handlerStack.length - 1] ?? notFound
144 return preemptiveShutdown
145 })
146 handlerStack.push(handler)
147 currentHandler = handler
148 })
149 })
150})
151
152const makeResponse = Effect.fnUntraced(function*(
153 request: ServerRequest.HttpServerRequest,
154 response: ServerResponse.HttpServerResponse,
155 context: Context.Context<never>,
156 scope: Scope.Scope
157) {
158 const fields: {
159 headers: globalThis.Headers
160 status?: number
161 statusText?: string
162 } = {
163 headers: new globalThis.Headers(response.headers),
164 status: response.status
165 }
166
167 if (!Cookies.isEmpty(response.cookies)) {
168 for (const header of Cookies.toSetCookieHeaders(response.cookies)) {
169 fields.headers.append("set-cookie", header)
170 }
171 }
172 if (response.statusText !== undefined) fields.statusText = response.statusText
173
174 if (ServerResponse.omitsBody(response, request.method === "HEAD")) {
175 yield* cancelResponseBody(response.body)
176 return new Response(undefined, fields)
177 }
178 response = HttpEffect.scopeTransferToStream(response)
179 const body = response.body
180 switch (body._tag) {
181 case "Empty":
182 return new Response(undefined, fields)
183 case "Uint8Array":
184 case "Raw": {
185 if (body.body instanceof Response) {
186 for (const [key, value] of fields.headers.entries()) body.body.headers.set(key, value)
187 return body.body
188 }
189 return new Response(body.body as any, fields)
190 }
191 case "FormData":
192 return new Response(body.formData as any, fields)
193 case "Stream":
194 return new Response(
195 Stream.toReadableStreamWith(
196 Stream.unwrap(Effect.withFiber((fiber) => {
197 Fiber.runIn(fiber, scope)
198 return Effect.succeed(body.stream)
199 })),
200 context
201 ),
202 fields
203 )
204 }
205})
206
207/**
208 * Provides only the native Deno HTTP server.
209 *
210 * @category layers
211 * @since 4.0.0
212 */
213export const layerServer: (
214 options: ServeOptions & {
215 readonly disablePreemptiveShutdown?: boolean | undefined
216 readonly gracefulShutdownTimeout?: Duration.Input | undefined
217 readonly websocket?: Deno.UpgradeWebSocketOptions | undefined
218 }
219) => Layer.Layer<Server.HttpServer, Error.ServeError> = flow(
220 make,
221 Layer.effect(Server.HttpServer)
222)
223
224/**
225 * Provides Deno HTTP platform services and the standard Deno service set.
226 *
227 * @category layers
228 * @since 4.0.0
229 */
230export const layerHttpServices: Layer.Layer<HttpPlatform | Etag.Generator | DenoServices.DenoServices> = Layer.mergeAll(
231 Platform.layer,
232 Etag.layerWeak,
233 DenoServices.layer
234)
235
236/**
237 * Provides a native Deno HTTP server together with Deno HTTP services.
238 *
239 * @category layers
240 * @since 4.0.0
241 */
242export const layer = (
243 options: ServeOptions & {
244 readonly disablePreemptiveShutdown?: boolean | undefined
245 readonly gracefulShutdownTimeout?: Duration.Input | undefined
246 readonly websocket?: Deno.UpgradeWebSocketOptions | undefined
247 }
248): Layer.Layer<Server.HttpServer | HttpPlatform | Etag.Generator | DenoServices.DenoServices, Error.ServeError> =>
249 Layer.mergeAll(layerServer(options), layerHttpServices)
250
251/**
252 * Starts a Deno HTTP server on an ephemeral loopback port for tests.
253 *
254 * @category testing
255 * @since 4.0.0
256 */
257export const layerTest: Layer.Layer<
258 Server.HttpServer | HttpPlatform | FileSystem.FileSystem | Etag.Generator | Path.Path | HttpClient
259> = Server.layerTestClient.pipe(
260 Layer.provide(FetchHttpClient.layer.pipe(
261 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false }))
262 )),
263 Layer.provideMerge(Layer.orDie(layer({ hostname: "127.0.0.1", port: 0, onListen: () => {} })))
264)
265
266/**
267 * Creates the Deno HTTP server and support-services layer from configurable options.
268 *
269 * @category layers
270 * @since 4.0.0
271 */
272export const layerConfig = (
273 options: Config.Wrap<
274 ServeOptions & {
275 readonly disablePreemptiveShutdown?: boolean | undefined
276 readonly gracefulShutdownTimeout?: Duration.Input | undefined
277 readonly websocket?: Deno.UpgradeWebSocketOptions | undefined
278 }
279 >
280): Layer.Layer<
281 Server.HttpServer | HttpPlatform | FileSystem.FileSystem | Etag.Generator | Path.Path,
282 ConfigError | Error.ServeError
283> =>
284 Layer.mergeAll(
285 Layer.effect(Server.HttpServer)(Effect.flatMap(Config.unwrap(options), make)),
286 layerHttpServices
287 )
288
289class DenoServerRequest extends Inspectable.Class implements ServerRequest.HttpServerRequest {
290 readonly [ServerRequest.TypeId]: typeof ServerRequest.TypeId
291 readonly [IncomingMessage.TypeId]: typeof IncomingMessage.TypeId
292 readonly source: Request
293 readonly remoteAddr: Deno.NetAddr | Deno.UnixAddr
294 readonly url: string
295 readonly websocketOptions: Deno.UpgradeWebSocketOptions | undefined
296 public resolve: (response: Response) => void
297 public upgraded = false
298 public headersOverride?: Headers.Headers | undefined
299 private remoteAddressOverride?: Option.Option<string> | undefined
300
301 constructor(
302 source: Request,
303 remoteAddr: Deno.NetAddr | Deno.UnixAddr,
304 resolve: (response: Response) => void,
305 url: string,
306 websocketOptions: Deno.UpgradeWebSocketOptions | undefined,
307 headersOverride?: Headers.Headers,
308 remoteAddressOverride?: Option.Option<string>
309 ) {
310 super()
311 this[ServerRequest.TypeId] = ServerRequest.TypeId
312 this[IncomingMessage.TypeId] = IncomingMessage.TypeId
313 this.source = source
314 this.remoteAddr = remoteAddr
315 this.resolve = resolve
316 this.url = url
317 this.websocketOptions = websocketOptions
318 this.headersOverride = headersOverride
319 this.remoteAddressOverride = remoteAddressOverride
320 }
321 toJSON(): unknown {
322 return IncomingMessage.inspect(this, {
323 _id: "HttpServerRequest",
324 method: this.method,
325 url: this.originalUrl
326 })
327 }
328 modify(options: {
329 readonly url?: string | undefined
330 readonly headers?: Headers.Headers | undefined
331 readonly remoteAddress?: Option.Option<string> | undefined
332 }) {
333 return new DenoServerRequest(
334 this.source,
335 this.remoteAddr,
336 this.resolve,
337 options.url ?? this.url,
338 this.websocketOptions,
339 options.headers ?? this.headersOverride,
340 "remoteAddress" in options ? options.remoteAddress : this.remoteAddressOverride
341 )
342 }
343 get method(): HttpMethod {
344 return this.source.method.toUpperCase() as HttpMethod
345 }
346 get originalUrl() {
347 return this.source.url
348 }
349 get remoteAddress(): Option.Option<string> {
350 return this.remoteAddressOverride ?? (this.remoteAddr.transport === "tcp"
351 ? Option.some(this.remoteAddr.hostname)
352 : Option.none())
353 }
354 get headers(): Headers.Headers {
355 this.headersOverride ??= Headers.fromInput(this.source.headers)
356 return this.headersOverride
357 }
358
359 private cachedCookies: Record.ReadonlyRecord<string, string> | undefined
360 get cookies() {
361 return this.cachedCookies ??= Cookies.parseHeader(this.headers.cookie ?? "")
362 }
363
364 get stream(): Stream.Stream<Uint8Array, Error.HttpServerError> {
365 return this.source.body
366 ? Stream.fromReadableStream({
367 evaluate: () => this.source.body ?? emptyReadableStream,
368 onError: (cause) =>
369 new Error.HttpServerError({
370 reason: new Error.RequestParseError({ request: this, cause })
371 })
372 })
373 : Stream.fail(
374 new Error.HttpServerError({
375 reason: new Error.RequestParseError({
376 request: this,
377 description: "can not create stream from empty body"
378 })
379 })
380 )
381 }
382
383 private textEffect: Effect.Effect<string, Error.HttpServerError> | undefined
384 get text(): Effect.Effect<string, Error.HttpServerError> {
385 return this.textEffect ??= Effect.runSync(Effect.cached(
386 Effect.tryPromise({
387 try: () => this.source.text(),
388 catch: (cause) =>
389 new Error.HttpServerError({
390 reason: new Error.RequestParseError({ request: this, cause })
391 })
392 })
393 ))
394 }
395 get json(): Effect.Effect<Schema.Json, Error.HttpServerError> {
396 return Effect.flatMap(this.text, (_) =>
397 Effect.try({
398 try: () => JSON.parse(_) as Schema.Json,
399 catch: (cause) =>
400 new Error.HttpServerError({
401 reason: new Error.RequestParseError({ request: this, cause })
402 })
403 }))
404 }
405 get urlParamsBody(): Effect.Effect<UrlParams.UrlParams, Error.HttpServerError> {
406 return Effect.flatMap(this.text, (_) =>
407 Effect.try({
408 try: () => UrlParams.fromInput(new URLSearchParams(_)),
409 catch: (cause) =>
410 new Error.HttpServerError({
411 reason: new Error.RequestParseError({ request: this, cause })
412 })
413 }))
414 }
415
416 private multipartEffect:
417 | Effect.Effect<Multipart.Persisted, Multipart.MultipartError, Scope.Scope | FileSystem.FileSystem | Path.Path>
418 | undefined
419 get multipart(): Effect.Effect<
420 Multipart.Persisted,
421 Multipart.MultipartError,
422 Scope.Scope | FileSystem.FileSystem | Path.Path
423 > {
424 return this.multipartEffect ??= Effect.runSync(Effect.cached(DenoMultipart.persisted(this.source)))
425 }
426 get multipartStream(): Stream.Stream<Multipart.Part, Multipart.MultipartError> {
427 return DenoMultipart.stream(this.source)
428 }
429
430 private arrayBufferEffect: Effect.Effect<ArrayBuffer, Error.HttpServerError> | undefined
431 get arrayBuffer(): Effect.Effect<ArrayBuffer, Error.HttpServerError> {
432 if (this.arrayBufferEffect) return this.arrayBufferEffect
433 this.arrayBufferEffect = Effect.runSync(Effect.cached(
434 Effect.tryPromise({
435 try: () => this.source.arrayBuffer(),
436 catch: (cause) =>
437 new Error.HttpServerError({
438 reason: new Error.RequestParseError({ request: this, cause })
439 })
440 })
441 ))
442 this.textEffect = Effect.map(this.arrayBufferEffect, (_) => new TextDecoder().decode(_))
443 return this.arrayBufferEffect
444 }
445
446 get upgrade(): Effect.Effect<Socket.Socket, Error.HttpServerError> {
447 return Effect.flatMap(
448 Effect.try({
449 try: () => Deno.upgradeWebSocket(this.source, this.websocketOptions),
450 catch: (cause) =>
451 new Error.HttpServerError({
452 reason: new Error.RequestParseError({
453 request: this,
454 cause,
455 description: "Not an upgradeable ServerRequest"
456 })
457 })
458 }),
459 (upgrade) => {
460 const ws = bufferedWebSocket(upgrade.socket)
461 this.upgraded = true
462 this.resolve(upgrade.response)
463 return Socket.fromWebSocket(
464 Effect.acquireRelease(
465 Effect.succeed(ws),
466 () => Effect.sync(() => upgrade.socket.close(1000))
467 )
468 )
469 }
470 )
471 }
472}
473
474// Deno's WebSocket cannot be paused, so events that arrive between the
475// upgrade and the first reader acquisition would be lost. This facade buffers
476// message, error, and close events until listeners attach, then replays them
477// so they come out of the first pull.
478const bufferedWebSocket = (ws: WebSocket): Socket.WebSocketLike => {
479 interface Entry {
480 readonly listener: (event: Socket.WebSocketEvent) => void
481 readonly once: boolean
482 }
483 const listeners = new Map<string, Array<Entry>>()
484 const buffered: Array<[string, Socket.WebSocketEvent]> = []
485 let buffering = true
486
487 ws.binaryType = "arraybuffer"
488
489 function removeListener(type: string, listener: (event: Socket.WebSocketEvent) => void) {
490 const entries = listeners.get(type)
491 if (!entries) return
492 const index = entries.findIndex((entry) => entry.listener === listener)
493 if (index >= 0) entries.splice(index, 1)
494 }
495 function dispatch(type: string, event: Socket.WebSocketEvent) {
496 const entries = listeners.get(type)
497 if (!entries || entries.length === 0) return
498 for (const entry of entries.slice()) {
499 if (entry.once) removeListener(type, entry.listener)
500 entry.listener(event)
501 }
502 }
503 function onEvent(type: string) {
504 return (event: Event) => {
505 const wsEvent = event as unknown as Socket.WebSocketEvent
506 if (buffering) {
507 buffered.push([type, wsEvent])
508 } else {
509 dispatch(type, wsEvent)
510 }
511 }
512 }
513 ws.addEventListener("message", onEvent("message"))
514 ws.addEventListener("error", onEvent("error"))
515 ws.addEventListener("close", onEvent("close"))
516
517 function replay() {
518 if (!buffering) return
519 buffering = false
520 for (const [type, event] of buffered.splice(0)) {
521 dispatch(type, event)
522 }
523 }
524
525 return {
526 get readyState() {
527 return ws.readyState
528 },
529 addEventListener(type, listener, options) {
530 if (type === "open") {
531 ws.addEventListener(type, listener as EventListener, options as AddEventListenerOptions)
532 return
533 }
534 let entries = listeners.get(type)
535 if (!entries) {
536 entries = []
537 listeners.set(type, entries)
538 }
539 entries.push({ listener, once: options?.once === true })
540 if (buffering) queueMicrotask(replay)
541 },
542 removeEventListener(type, listener) {
543 if (type === "open") {
544 ws.removeEventListener(type, listener as EventListener)
545 return
546 }
547 removeListener(type, listener)
548 },
549 close: (code, reason) => ws.close(code, reason),
550 send: (data) => ws.send(data)
551 }
552}
553
554const cancelResponseBody = (body: HttpBody.HttpBody): Effect.Effect<void> => {
555 if (body._tag === "Raw" && typeof ReadableStream !== "undefined" && body.body instanceof ReadableStream) {
556 const stream = body.body
557 return Effect.ignoreCause(Effect.promise(() => stream.cancel()))
558 }
559 return Effect.void
560}
561
562const emptyReadableStream = new ReadableStream<Uint8Array>({
563 start(controller) {
564 controller.enqueue(new Uint8Array())
565 controller.close()
566 }
567})
568
569const removeHost = (url: string) => {
570 if (url[0] === "/") return url
571 const index = url.indexOf("/", url.indexOf("//") + 2)
572 return index === -1 ? "/" : url.slice(index)
573}