Allow enabling websocket compression in NodeHttpServer and BunHttpServer

.changeset/http-server-websocket-options.md

@@ -0,0 +1,22 @@
+---
+"@effect/platform-node": patch
+"@effect/platform-bun": patch
+---
+
+Allow configuring the WebSocket server in `NodeHttpServer` and `BunHttpServer`.
+
+Both servers now accept a `websocket` option that is forwarded to the underlying implementation, with the wiring/lifecycle options the server manages excluded from the type:
+
+```ts
+// Node: forwarded to the `ws` WebSocketServer
+NodeHttpServer.layer(() => createServer(), {
+  port: 3000,
+  websocket: { perMessageDeflate: true }
+})
+
+// Bun: merged into Bun.serve's websocket handler
+BunHttpServer.layer({
+  port: 3000,
+  websocket: { perMessageDeflate: true }
+})
+```

packages/platform-bun/src/BunHttpServer.ts

@@ -66,6 +66,25 @@ export type ServeOptions<R extends string> =
   )
   & { readonly routes?: Bun.Serve.Routes<WebSocketContext, R> }
 
+/**
+ * WebSocket tuning options forwarded to `Bun.serve`'s `websocket` handler.
+ *
+ * **Details**
+ *
+ * The lifecycle handlers (`open`, `message`, `close`, ...) are managed by the
+ * server and cannot be overridden; everything else — such as
+ * `perMessageDeflate` compression, payload limits, and idle timeouts — passes
+ * through, e.g.
+ * `BunHttpServer.layer({ port: 3000, websocket: { perMessageDeflate: true } })`.
+ *
+ * @category options
+ * @since 4.0.0
+ */
+export type WebSocketOptions = Omit<
+  Bun.WebSocketHandler<WebSocketContext>,
+  "open" | "message" | "close" | "drain" | "ping" | "pong" | "data" | "binaryType"
+>
+
 /**
  * Creates a scoped Bun `HttpServer` from `Bun.serve` options, stopping the server on scope finalization with optional graceful shutdown settings.
  *
@@ -77,6 +96,7 @@ export const make = Effect.fnUntraced(
     options: ServeOptions<R> & {
       readonly disablePreemptiveShutdown?: boolean | undefined
       readonly gracefulShutdownTimeout?: Duration.Input | undefined
+      readonly websocket?: WebSocketOptions | undefined
     }
   ) {
     const scope = yield* Effect.scope
@@ -90,6 +110,7 @@ export const make = Effect.fnUntraced(
       ...options as ServeOptions<R>,
       fetch: handlerStack[0],
       websocket: {
+        ...options.websocket,
         open(ws) {
           Deferred.doneUnsafe(ws.data.deferred, Exit.succeed(ws))
         },
@@ -233,6 +254,7 @@ export const layerServer: <R extends string>(
   options: ServeOptions<R> & {
     readonly disablePreemptiveShutdown?: boolean | undefined
     readonly gracefulShutdownTimeout?: Duration.Input | undefined
+    readonly websocket?: WebSocketOptions | undefined
   }
 ) => Layer.Layer<Server.HttpServer> = flow(make, Layer.effect(Server.HttpServer)) as any
 
@@ -262,6 +284,7 @@ export const layer = <R extends string>(
   options: ServeOptions<R> & {
     readonly disablePreemptiveShutdown?: boolean | undefined
     readonly gracefulShutdownTimeout?: Duration.Input | undefined
+    readonly websocket?: WebSocketOptions | undefined
   }
 ): Layer.Layer<
   | Server.HttpServer
@@ -296,6 +319,7 @@ export const layerConfig = <R extends string>(
     ServeOptions<R> & {
       readonly disablePreemptiveShutdown?: boolean | undefined
       readonly gracefulShutdownTimeout?: Duration.Input | undefined
+      readonly websocket?: WebSocketOptions | undefined
     }
   >
 ): Layer.Layer<

packages/platform-node/src/NodeHttpServer.ts

@@ -62,6 +62,26 @@ import * as NodeMultipart from "./NodeMultipart.ts"
 import * as NodeServices from "./NodeServices.ts"
 import { NodeWS } from "./NodeSocket.ts"
 
+/**
+ * Options accepted by the Node `HttpServer` constructors and layers.
+ *
+ * @category options
+ * @since 4.0.0
+ */
+export interface Options extends Net.ListenOptions {
+  readonly disablePreemptiveShutdown?: boolean | undefined
+  readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  /**
+   * Options forwarded to the underlying `ws` `WebSocketServer`, minus the
+   * wiring options the server manages itself. Use this to enable
+   * `permessage-deflate` compression or tune payload limits, e.g.
+   * `websocket: { perMessageDeflate: true }`.
+   */
+  readonly websocket?:
+    | Omit<NodeWS.ServerOptions, "noServer" | "server" | "host" | "port" | "path">
+    | undefined
+}
+
 /**
  * Creates a scoped `HttpServer` from a Node `http.Server`, starts listening
  * with the supplied options, registers request and upgrade handling, and closes
@@ -72,10 +92,7 @@ import { NodeWS } from "./NodeSocket.ts"
  */
 export const make = Effect.fnUntraced(function*(
   evaluate: LazyArg<Http.Server>,
-  options: Net.ListenOptions & {
-    readonly disablePreemptiveShutdown?: boolean | undefined
-    readonly gracefulShutdownTimeout?: Duration.Input | undefined
-  }
+  options: Options
 ) {
   const scope = yield* Effect.scope
   const server = evaluate()
@@ -116,7 +133,7 @@ export const make = Effect.fnUntraced(function*(
   const address = server.address()!
 
   const wss = yield* Effect.acquireRelease(
-    Effect.sync(() => new NodeWS.WebSocketServer({ noServer: true })),
+    Effect.sync(() => new NodeWS.WebSocketServer({ ...options.websocket, noServer: true })),
     (wss) =>
       Effect.callback<void>((resume) => {
         wss.close(() => resume(Effect.void))
@@ -397,10 +414,7 @@ class ServerRequestImpl extends NodeHttpIncomingMessage<HttpServerError> impleme
  */
 export const layerServer: (
   evaluate: LazyArg<Http.Server<typeof Http.IncomingMessage, typeof Http.ServerResponse>>,
-  options: Net.ListenOptions & {
-    readonly disablePreemptiveShutdown?: boolean | undefined
-    readonly gracefulShutdownTimeout?: Duration.Input | undefined
-  }
+  options: Options
 ) => Layer.Layer<HttpServer.HttpServer, ServeError> = flow(make, Layer.effect(HttpServer.HttpServer))
 
 /**
@@ -427,10 +441,7 @@ export const layerHttpServices: Layer.Layer<
  */
 export const layer = (
   evaluate: LazyArg<Http.Server>,
-  options: Net.ListenOptions & {
-    readonly disablePreemptiveShutdown?: boolean | undefined
-    readonly gracefulShutdownTimeout?: Duration.Input | undefined
-  }
+  options: Options
 ): Layer.Layer<
   HttpServer.HttpServer | NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator,
   ServeError
@@ -450,12 +461,7 @@ export const layer = (
  */
 export const layerConfig = (
   evaluate: LazyArg<Http.Server>,
-  options: Config.Wrap<
-    Net.ListenOptions & {
-      readonly disablePreemptiveShutdown?: boolean | undefined
-      readonly gracefulShutdownTimeout?: Duration.Input | undefined
-    }
-  >
+  options: Config.Wrap<Options>
 ): Layer.Layer<
   HttpServer.HttpServer | NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator,
   ServeError | Config.ConfigError

packages/platform-node/test/NodeHttpServer.test.ts

@@ -1,5 +1,6 @@
 /** @effect-diagnostics preferSchemaOverJson:skip-file */
 import { NodeHttpServer } from "@effect/platform-node"
+import { NodeWS } from "@effect/platform-node/NodeSocket"
 import { assert, describe, expect, it } from "@effect/vitest"
 import { Effect } from "effect"
 import * as Duration from "effect/Duration"
@@ -673,8 +674,58 @@ describe("HttpServer", () => {
       )
       expect(root).toEqual("root")
     }).pipe(Effect.provide(NodeHttpServer.layerTest)))
+
+  it.effect("websocket options are forwarded to the WebSocketServer", () =>
+    Effect.gen(function*() {
+      yield* HttpRouter.add(
+        "GET",
+        "/ws",
+        Effect.gen(function*() {
+          const request = yield* HttpServerRequest.HttpServerRequest
+          const socket = yield* Effect.orDie(request.upgrade)
+          yield* Effect.orDie(socket.run(() => Effect.void))
+          return HttpServerResponse.empty()
+        })
+      ).pipe(
+        HttpRouter.serve,
+        Layer.build
+      )
+      const server = yield* HttpServer.HttpServer
+      const port = (server.address as HttpServer.TcpAddress).port
+
+      const connect = (perMessageDeflate: boolean) =>
+        Effect.acquireRelease(
+          Effect.callback<NodeWS.WebSocket, Error>((resume) => {
+            const ws = new NodeWS.WebSocket(`ws://127.0.0.1:${port}/ws`, { perMessageDeflate })
+            ws.on("open", () => resume(Effect.succeed(ws)))
+            ws.on("error", (error) => resume(Effect.fail(error)))
+          }),
+          (ws) => Effect.sync(() => ws.close())
+        )
+
+      // layerTest configures websocket: { perMessageDeflate: true }, so the
+      // server accepts the extension when the client offers it...
+      const compressed = yield* connect(true)
+      expect(compressed.extensions).toContain("permessage-deflate")
+
+      // ...and clients that do not offer it still connect uncompressed.
+      const plain = yield* connect(false)
+      expect(plain.extensions).not.toContain("permessage-deflate")
+    }).pipe(Effect.scoped, Effect.provide(layerTestWebsocket)))
 })
 
+const layerTestWebsocket = HttpServer.layerTestClient.pipe(
+  Layer.provide(
+    Layer.fresh(FetchHttpClient.layer).pipe(
+      Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false }))
+    )
+  ),
+  Layer.provideMerge(NodeHttpServer.layer(Http.createServer, {
+    port: 0,
+    websocket: { perMessageDeflate: true }
+  }))
+)
+
 const tcpPort = (server: Http.Server): number => {
   const address = server.address()
   assert(address !== null && typeof address !== "string")