1import * as BunHttpServer from "@effect/platform-bun/BunHttpServer"
2import { assert, describe, it } from "@effect/vitest"
3import * as Deferred from "effect/Deferred"
4import * as Effect from "effect/Effect"
5import * as Exit from "effect/Exit"
6import * as Fiber from "effect/Fiber"
7import * as FetchHttpClient from "effect/http/FetchHttpClient"
8import * as HttpClient from "effect/http/HttpClient"
9import * as HttpServer from "effect/http/HttpServer"
10import * as HttpServerRequest from "effect/http/HttpServerRequest"
11import * as HttpServerResponse from "effect/http/HttpServerResponse"
12import * as NetAddress from "effect/net/NetAddress"
13import * as Scope from "effect/Scope"
14import * as Socket from "effect/socket/Socket"
15import * as Stream from "effect/Stream"
16import { mkdtemp, rm } from "node:fs/promises"
17import * as Net from "node:net"
18import { tmpdir } from "node:os"
19import { join } from "node:path"
20
21const fetchText = (url: string) =>
22 Effect.promise(() => fetch(url, { headers: { connection: "close" } }).then((response) => response.text()))
23
24const readWebSocketClose = (port: number, opened: Deferred.Deferred<void>) =>
25 Effect.callback<number, Error>((resume) => {
26 const socket = Net.createConnection({ host: "127.0.0.1", port })
27 let received = Buffer.alloc(0)
28 let upgraded = false
29 let closeCode: number | undefined
30 socket.on("close", () => {
31 if (closeCode !== undefined) resume(Effect.succeed(closeCode))
32 })
33 socket.on("connect", () =>
34 socket.write([
35 "GET / HTTP/1.1",
36 "Host: 127.0.0.1:" + port,
37 "Connection: Upgrade",
38 "Upgrade: websocket",
39 "Sec-WebSocket-Version: 13",
40 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
41 "",
42 ""
43 ].join("\r\n")))
44 socket.on("error", (error) => resume(Effect.fail(error)))
45 socket.on("data", (chunk) => {
46 received = Buffer.concat([received, typeof chunk === "string" ? Buffer.from(chunk) : chunk])
47 if (!upgraded) {
48 const headerEnd = received.indexOf("\r\n\r\n")
49 if (headerEnd === -1) return
50 upgraded = true
51 received = received.subarray(headerEnd + 4)
52 Effect.runSync(Deferred.succeed(opened, undefined))
53 }
54 if (received.length < 4 || (received[0] & 0x0f) !== 8) return
55 if (closeCode !== undefined) return
56 closeCode = received.readUInt16BE(2)
57 socket.write(Buffer.from([0x88, 0x82, 0, 0, 0, 0, closeCode >> 8, closeCode & 0xff]))
58 })
59 return Effect.sync(() => socket.destroy())
60 })
61
62interface WebSocketFrame {
63 readonly opcode: number
64 readonly payload: Uint8Array
65 readonly payloadLength: number
66 readonly rsv1: boolean
67}
68
69interface WebSocketFrames {
70 readonly frames: ReadonlyArray<WebSocketFrame>
71 readonly headers: string
72}
73
74const readWebSocketFrames = (port: number, perMessageDeflate: boolean) =>
75 Effect.callback<WebSocketFrames, Error>((resume) => {
76 const socket = Net.createConnection({ host: "127.0.0.1", port })
77 let received = Buffer.alloc(0)
78 let result: WebSocketFrames | undefined
79
80 socket.on("connect", () => {
81 socket.write([
82 "GET / HTTP/1.1",
83 `Host: 127.0.0.1:${port}`,
84 "Connection: Upgrade",
85 "Upgrade: websocket",
86 "Sec-WebSocket-Version: 13",
87 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
88 ...(perMessageDeflate ? ["Sec-WebSocket-Extensions: permessage-deflate"] : []),
89 "",
90 ""
91 ].join("\r\n"))
92 })
93 socket.on("error", (error) => resume(Effect.fail(error)))
94 socket.on("close", () => {
95 if (result) resume(Effect.succeed(result))
96 })
97 socket.on("data", (chunk) => {
98 if (result) return
99 received = Buffer.concat([received, typeof chunk === "string" ? Buffer.from(chunk) : chunk])
100 const headerEnd = received.indexOf("\r\n\r\n")
101 if (headerEnd === -1) return
102
103 const headers = received.subarray(0, headerEnd).toString()
104 const frames: Array<WebSocketFrame> = []
105 let offset = headerEnd + 4
106 while (frames.length < 2) {
107 if (received.length < offset + 2) return
108 const first = received[offset]
109 const length = received[offset + 1] & 0x7f
110 const headerLength = length === 126 ? 4 : 2
111 if (received.length < offset + headerLength) return
112 const payloadLength = length === 126 ? received.readUInt16BE(offset + 2) : length
113 if (received.length < offset + headerLength + payloadLength) return
114 frames.push({
115 opcode: first & 0x0f,
116 payload: Uint8Array.from(received.subarray(offset + headerLength, offset + headerLength + payloadLength)),
117 payloadLength,
118 rsv1: (first & 0x40) !== 0
119 })
120 offset += headerLength + payloadLength
121 }
122 result = { frames, headers }
123 socket.write(Buffer.from([0x88, 0x82, 0, 0, 0, 0, 0x03, 0xe8]))
124 })
125
126 return Effect.sync(() => socket.destroy())
127 })
128
129const makeWebSocketServer = Effect.fnUntraced(function*(payload: string, compressionThreshold?: number) {
130 const server = yield* BunHttpServer.make({
131 hostname: "127.0.0.1",
132 port: 0,
133 websocket: { perMessageDeflate: true, compressionThreshold }
134 })
135 yield* server.serve(Effect.gen(function*() {
136 const request = yield* HttpServerRequest.HttpServerRequest
137 const socket = yield* request.upgrade
138 yield* Effect.gen(function*() {
139 const { pull } = yield* socket.reader
140 const writer = yield* socket.writer
141 yield* Effect.orDie(writer.write(payload))
142 yield* Effect.orDie(writer.write(new TextEncoder().encode(payload)))
143 while (true) {
144 yield* pull
145 }
146 }).pipe(
147 Effect.scoped,
148 Effect.catchTag("SocketError", () => Effect.void),
149 Effect.orDie
150 )
151 return HttpServerResponse.empty()
152 }))
153 return server
154})
155
156describe("BunHttpServer", () => {
157 describe("body omission", () => {
158 for (const status of [204, 205, 304]) {
159 for (const bodyKind of ["text", "stream"]) {
160 it.live(`omits ${bodyKind} bodies for status ${status}`, () =>
161 Effect.gen(function*() {
162 let finalized = false
163 let streamStarted = false
164 yield* HttpServer.serveEffect(Effect.gen(function*() {
165 yield* Effect.addFinalizer(() =>
166 Effect.sync(() => {
167 finalized = true
168 })
169 )
170 const options = { status, contentType: "text/plain", contentLength: 4 }
171 return bodyKind === "text"
172 ? HttpServerResponse.text("body", options)
173 : HttpServerResponse.stream(
174 Stream.fromEffect(Effect.sync(() => {
175 streamStarted = true
176 return new TextEncoder().encode("body")
177 })),
178 options
179 )
180 }))
181 const response = yield* HttpClient.get("/")
182 assert.strictEqual(response.status, status)
183 assert.strictEqual(yield* response.text, "")
184 assert.strictEqual(streamStarted, false)
185 assert.strictEqual(finalized, true)
186 if (status === 304) {
187 assert.strictEqual(response.headers["content-type"], "text/plain")
188 }
189 }).pipe(
190 Effect.timeout("2 seconds"),
191 Effect.provide(BunHttpServer.layerTest)
192 ), 5000)
193 }
194 }
195
196 for (const [method, status] of [["HEAD", 200], ["GET", 204], ["GET", 205], ["GET", 304]] as const) {
197 it.live(`cancels raw streams for ${method} status ${status}`, () =>
198 Effect.gen(function*() {
199 let cancelled = false
200 const body = new ReadableStream<Uint8Array>({
201 pull(controller) {
202 controller.enqueue(new TextEncoder().encode("body"))
203 controller.close()
204 },
205 cancel() {
206 cancelled = true
207 }
208 }, { highWaterMark: 0 })
209 yield* Effect.addFinalizer(() => Effect.ignoreCause(Effect.promise(() => body.cancel())))
210 yield* HttpServer.serveEffect(Effect.succeed(HttpServerResponse.raw(body, {
211 status,
212 contentType: "text/plain",
213 contentLength: 4
214 })))
215 const response = yield* (method === "HEAD" ? HttpClient.head("/") : HttpClient.get("/"))
216 assert.strictEqual(response.status, status)
217 assert.strictEqual(yield* response.text, "")
218 assert.strictEqual(cancelled, true)
219 if (method === "HEAD") {
220 assert.strictEqual(response.headers["content-length"], "4")
221 }
222 if (method === "HEAD" || status === 304) {
223 assert.strictEqual(response.headers["content-type"], "text/plain")
224 }
225 }).pipe(
226 Effect.timeout("2 seconds"),
227 Effect.provide(BunHttpServer.layerTest)
228 ), 5000)
229 }
230 })
231
232 for (
233 const [name, options, expectedTag, expectedIp] of [
234 ["omitted hostname", { port: 0 }, "InetAddressV6", "::"],
235 ["undefined Unix path and omitted hostname", { port: 0, unix: undefined }, "InetAddressV6", "::"],
236 ["explicit wildcard hostname", { port: 0, hostname: "0.0.0.0" }, "InetAddressV4", "0.0.0.0"]
237 ] as const
238 ) {
239 it.effect(`starts a layer with ${name}`, () =>
240 Effect.gen(function*() {
241 const server = yield* HttpServer.HttpServer
242 if (server.address._tag !== expectedTag) {
243 return assert.fail(`expected ${expectedTag}, got ${server.address._tag}`)
244 }
245 assert.strictEqual(NetAddress.formatIp(server.address.address), expectedIp)
246 assert.isAbove(server.address.port, 0)
247
248 yield* server.serve(Effect.succeed(HttpServerResponse.text("default hostname")))
249 const client = yield* HttpServer.makeTestClient.pipe(Effect.provide(FetchHttpClient.layer))
250 const response = yield* client.get("/")
251 assert.strictEqual(response.status, 200)
252 assert.strictEqual(yield* response.text, "default hostname")
253 }).pipe(Effect.provide(BunHttpServer.layer(options))))
254 }
255
256 it.effect("treats an undefined Unix path as a TCP listener", () =>
257 Effect.gen(function*() {
258 for (const hostname of ["localhost", "127.0.0.1"]) {
259 const server = yield* BunHttpServer.make({ unix: undefined, hostname, port: 0 })
260 assert.isTrue(server.address._tag === "InetAddressV4" || server.address._tag === "InetAddressV6")
261 yield* server.serve(Effect.succeed(HttpServerResponse.text("tcp")))
262 const client = yield* HttpServer.makeTestClient.pipe(
263 Effect.provideService(HttpServer.HttpServer, server),
264 Effect.provide(FetchHttpClient.layer)
265 )
266 const response = yield* client.get("/")
267 assert.strictEqual(yield* response.text, "tcp")
268 }
269 }))
270
271 it.effect("resolves hostnames and formats Unix socket addresses", () =>
272 Effect.gen(function*() {
273 const server = yield* BunHttpServer.make({ hostname: "localhost", port: 0 })
274 assert.isTrue(server.address._tag === "InetAddressV4" || server.address._tag === "InetAddressV6")
275
276 const directory = yield* Effect.acquireRelease(
277 Effect.promise(() => mkdtemp(join(tmpdir(), "bun-http-"))),
278 (directory) => Effect.promise(() => rm(directory, { recursive: true, force: true }))
279 )
280 const path = join(directory, "server.sock")
281 const unixServer = yield* BunHttpServer.make({ unix: path })
282
283 assert.strictEqual(unixServer.address._tag, "UnixPathAddress")
284 assert.strictEqual(unixServer.address._tag === "UnixPathAddress" ? unixServer.address.path : undefined, path)
285 assert.strictEqual(NetAddress.formatUrlUnsafe(unixServer.address), `unix://${path}`)
286 }))
287
288 it.effect("closing an older serve scope keeps the newer handler active", () =>
289 Effect.gen(function*() {
290 const ownerScope = yield* Effect.scope
291 const server = yield* BunHttpServer.make({
292 hostname: "127.0.0.1",
293 port: 0
294 })
295 const firstScope = yield* Scope.fork(ownerScope)
296 const secondScope = yield* Scope.fork(ownerScope)
297
298 yield* server.serve(Effect.succeed(HttpServerResponse.text("first"))).pipe(Scope.provide(firstScope))
299 yield* server.serve(Effect.succeed(HttpServerResponse.text("second"))).pipe(Scope.provide(secondScope))
300 const url = NetAddress.formatUrlUnsafe(server.address)
301
302 assert.strictEqual(yield* fetchText(url), "second")
303 yield* Scope.close(firstScope, Exit.void)
304 assert.strictEqual(yield* fetchText(url), "second")
305 }))
306
307 it.effect("closing the newer serve scope restores the older handler", () =>
308 Effect.gen(function*() {
309 const ownerScope = yield* Effect.scope
310 const server = yield* BunHttpServer.make({
311 hostname: "127.0.0.1",
312 port: 0
313 })
314 const firstScope = yield* Scope.fork(ownerScope)
315 const secondScope = yield* Scope.fork(ownerScope)
316
317 yield* server.serve(Effect.succeed(HttpServerResponse.text("first"))).pipe(Scope.provide(firstScope))
318 yield* server.serve(Effect.succeed(HttpServerResponse.text("second"))).pipe(Scope.provide(secondScope))
319 const url = NetAddress.formatUrlUnsafe(server.address)
320
321 assert.strictEqual(yield* fetchText(url), "second")
322 yield* Scope.close(secondScope, Exit.void)
323 assert.strictEqual(yield* fetchText(url), "first")
324 }))
325
326 it.effect("preserves configured routes while changing handlers", () =>
327 Effect.gen(function*() {
328 const ownerScope = yield* Effect.scope
329 const server = yield* BunHttpServer.make({
330 hostname: "127.0.0.1",
331 port: 0,
332 routes: { "/static": new Response("static") }
333 })
334 const firstScope = yield* Scope.fork(ownerScope)
335 const secondScope = yield* Scope.fork(ownerScope)
336 const url = NetAddress.formatUrlUnsafe(server.address)
337
338 yield* server.serve(Effect.succeed(HttpServerResponse.text("first"))).pipe(Scope.provide(firstScope))
339 assert.strictEqual(yield* fetchText(`${url}/static`), "static")
340 assert.strictEqual(yield* fetchText(`${url}/fallback`), "first")
341
342 yield* server.serve(Effect.succeed(HttpServerResponse.text("second"))).pipe(Scope.provide(secondScope))
343 assert.strictEqual(yield* fetchText(`${url}/static`), "static")
344 assert.strictEqual(yield* fetchText(`${url}/fallback`), "second")
345
346 yield* Scope.close(secondScope, Exit.void)
347 assert.strictEqual(yield* fetchText(`${url}/static`), "static")
348 assert.strictEqual(yield* fetchText(`${url}/fallback`), "first")
349 }))
350
351 it.effect("compresses outgoing WebSocket messages when per-message deflate is negotiated", () =>
352 Effect.gen(function*() {
353 const payload = "a".repeat(4_096)
354 const server = yield* makeWebSocketServer(payload)
355 const port = (server.address as NetAddress.InetAddress).port
356 const { frames, headers } = yield* readWebSocketFrames(port, true)
357
358 assert.match(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
359 assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
360 assert.isTrue(frames.every((frame) => frame.rsv1))
361 assert.isTrue(frames.every((frame) => frame.payloadLength < payload.length))
362 }))
363
364 it.effect("leaves small WebSocket messages uncompressed even when per-message deflate is negotiated", () =>
365 Effect.gen(function*() {
366 const payload = "a".repeat(64)
367 const server = yield* makeWebSocketServer(payload)
368 const port = (server.address as NetAddress.InetAddress).port
369 const { frames, headers } = yield* readWebSocketFrames(port, true)
370
371 assert.match(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
372 assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
373 assert.isFalse(frames.some((frame) => frame.rsv1))
374 assert.isTrue(frames.every((frame) => frame.payloadLength === payload.length))
375 }))
376
377 it.effect("compresses small WebSocket messages when below a custom compressionThreshold", () =>
378 Effect.gen(function*() {
379 const payload = "a".repeat(64)
380 const server = yield* makeWebSocketServer(payload, 32)
381 const port = (server.address as NetAddress.InetAddress).port
382 const { frames } = yield* readWebSocketFrames(port, true)
383
384 assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
385 assert.isTrue(frames.every((frame) => frame.rsv1))
386 assert.isTrue(frames.every((frame) => frame.payloadLength < payload.length))
387 }))
388
389 it.effect("supports WebSocket clients without per-message deflate", () =>
390 Effect.gen(function*() {
391 const payload = "a".repeat(4_096)
392 const server = yield* makeWebSocketServer(payload)
393 const port = (server.address as NetAddress.InetAddress).port
394 const { frames, headers } = yield* readWebSocketFrames(port, false)
395
396 assert.notMatch(headers, /^sec-websocket-extensions:.*permessage-deflate/im)
397 assert.deepStrictEqual(frames.map((frame) => frame.opcode), [1, 2])
398 assert.isFalse(frames.some((frame) => frame.rsv1))
399 assert.isTrue(frames.every((frame) => frame.payloadLength === payload.length))
400 assert.deepStrictEqual(frames.map((frame) => new TextDecoder().decode(frame.payload)), [payload, payload])
401 }))
402
403 for (
404 const [name, exit, code] of [
405 ["success", "success", 1000],
406 ["interrupt", "interrupt", 1001],
407 ["failure", "failure", 1011],
408 ["defect", "defect", 1011],
409 ["explicit close before failure", "explicit", 4400]
410 ] as const
411 ) {
412 it.effect(`closes a WebSocket with the handler's ${name} code`, () =>
413 Effect.gen(function*() {
414 const opened = yield* Deferred.make<void>()
415 // Force-stop Bun after the close frame; graceful stop can hang here.
416 const serve = Bun.serve
417 let forceStop: (() => void) | undefined
418 Bun.serve = ((options: Parameters<typeof Bun.serve>[0]) => {
419 const bunServer = serve(options)
420 forceStop = () => {
421 bunServer.stop(true)
422 }
423 return bunServer
424 }) as typeof Bun.serve
425 const server = yield* BunHttpServer.make({
426 hostname: "127.0.0.1",
427 port: 0,
428 gracefulShutdownTimeout: "100 millis"
429 }).pipe(Effect.ensuring(Effect.sync(() => {
430 Bun.serve = serve
431 })))
432 yield* server.serve(Effect.gen(function*() {
433 const request = yield* HttpServerRequest.HttpServerRequest
434 const socket = yield* request.upgrade
435 const readerScope = yield* Scope.fork(yield* Effect.scope)
436 yield* socket.reader.pipe(Scope.provide(readerScope))
437 yield* Deferred.await(opened)
438 if (exit === "explicit") {
439 const writer = yield* socket.writer
440 yield* writer.write(new Socket.CloseEvent(4400, "handler closed"))
441 }
442 if (exit === "interrupt") return yield* Effect.interrupt
443 if (exit === "failure" || exit === "explicit") return yield* Effect.fail(new Error("handler failed"))
444 if (exit === "defect") return yield* Effect.die(new Error("handler defect"))
445 return HttpServerResponse.empty()
446 }))
447 yield* Effect.addFinalizer(() => Effect.sync(() => forceStop?.()))
448 const port = (server.address as NetAddress.InetAddress).port
449 const actual = yield* readWebSocketClose(port, opened)
450 forceStop?.()
451 assert.strictEqual(actual, code)
452 }).pipe(Effect.timeout("5 seconds")), 10000)
453 }
454
455 it.effect("fails a concurrent reader waiting behind a closed reader", () =>
456 Effect.gen(function*() {
457 const secondReaderFailed = yield* Deferred.make<boolean>()
458 const server = yield* BunHttpServer.make({
459 hostname: "127.0.0.1",
460 port: 0
461 })
462 yield* server.serve(Effect.gen(function*() {
463 const request = yield* HttpServerRequest.HttpServerRequest
464 const socket = yield* request.upgrade
465 const ownerScope = yield* Effect.scope
466 const firstScope = yield* Scope.fork(ownerScope)
467 const secondScope = yield* Scope.fork(ownerScope)
468
469 const { pull } = yield* socket.reader.pipe(Scope.provide(firstScope))
470 const writer = yield* socket.writer
471 const secondReader = yield* socket.reader.pipe(
472 Scope.provide(secondScope),
473 Effect.exit,
474 Effect.forkChild({ startImmediately: true })
475 )
476 yield* writer.write("first")
477 yield* writer.write("second")
478 yield* Effect.exit(pull)
479 yield* Scope.close(firstScope, Exit.void)
480 const secondExit = yield* Fiber.join(secondReader)
481 yield* Scope.close(secondScope, Exit.void)
482 yield* Deferred.succeed(secondReaderFailed, Exit.isFailure(secondExit))
483 return HttpServerResponse.empty()
484 }))
485
486 const port = (server.address as NetAddress.InetAddress).port
487 yield* readWebSocketFrames(port, false)
488 const failed = yield* Deferred.await(secondReaderFailed)
489 assert.isTrue(failed)
490 }))
491})