feat(server): compress batch responses with flush-per-message zlib stream

apps/content/docs/plugins/body-compression.md

@@ -25,6 +25,12 @@ const handler = new RPCHandler(router, {
 
 <!--@include: @/shared/common-plugin-handler-compatibility.md -->
 
+## Batch and Event Stream Responses
+
+Event stream responses are never compressed: the web `CompressionStream` API cannot flush between chunks, so early events would sit in the compressor buffer until the stream ends, defeating streaming semantics.
+
+[Batch requests](/docs/plugins/batch-requests) (identified by the `orpc-batch` header) with an `application/octet-stream` body are compressed by the fetch adapter when the runtime exposes the Node.js `zlib` and `stream` builtins via `process.getBuiltinModule` (Node.js ≥ 22.3, Bun, and compatible runtimes): a flush-per-message `zlib` stream is used instead of `CompressionStream`, so each response in a streaming batch is delivered to the client as soon as it is ready — compressed. In runtimes without these builtins, such batch responses stay uncompressed.
+
 ## Learn More
 
 For implementation details, see the [fetch adapter source code](https://github.com/middleapi/orpc/blob/main/packages/server/src/adapters/fetch/body-compression-plugin.ts) and the [node adapter source code](https://github.com/middleapi/orpc/blob/main/packages/server/src/adapters/node/body-compression-plugin.ts).

packages/server/src/adapters/fetch/body-compression-plugin.test.ts

@@ -1,4 +1,6 @@
 import type { FetchHandlerFetchInterceptor } from './handler'
+import { os } from '../../builder'
+import { BatchHandlerPlugin } from '../../plugins/batch'
 import { BodyCompressionHandlerPlugin } from './body-compression-plugin'
 import { RPCHandler } from './rpc-handler'
 
@@ -176,4 +178,240 @@ describe('bodyCompressionHandlerPlugin', () => {
     await expect(response!.text()).resolves.toBe(largeText)
     expect(response!.headers.has('content-encoding')).toBe(false)
   })
+
+  it('does not compress octet-stream responses for non-batch requests', async () => {
+    const body = new TextEncoder().encode(largeText)
+
+    const handler = createHandler(new Response(body, {
+      headers: {
+        'content-length': body.byteLength.toString(),
+        'content-type': 'application/octet-stream',
+      },
+    }))
+
+    const { matched, response } = await handler.handle(new Request('https://example.com/ping', {
+      headers: {
+        'accept-encoding': 'gzip',
+      },
+    }))
+
+    expect(matched).toBe(true)
+    await expect(response!.text()).resolves.toBe(largeText)
+    expect(response!.headers.has('content-encoding')).toBe(false)
+  })
+
+  describe('batch responses', () => {
+    const largeValue = 'a'.repeat(500)
+
+    function makePeerRequestMessage(id: number, url: string) {
+      return {
+        kind: 'request',
+        id,
+        json: { method: 'POST', url, headers: {}, body: undefined },
+        binary: undefined,
+      }
+    }
+
+    function createBatchRequest(mode: 'buffered' | 'streaming', messages: unknown) {
+      return new Request('https://example.com/__batch__', {
+        method: 'POST',
+        headers: {
+          'accept-encoding': 'gzip',
+          'content-type': 'application/json',
+          'orpc-batch': mode,
+        },
+        body: JSON.stringify(messages),
+      })
+    }
+
+    function createBatchHandler(
+      router: Parameters<typeof RPCHandler>[0],
+      options: ConstructorParameters<typeof BodyCompressionHandlerPlugin>[0] = {},
+    ): RPCHandler<any> {
+      return new RPCHandler(router, {
+        plugins: [new BatchHandlerPlugin(), new BodyCompressionHandlerPlugin(options)],
+      })
+    }
+
+    it('compresses streaming batch responses', async () => {
+      const handler = createBatchHandler({
+        ping: os.handler(() => largeValue),
+      })
+
+      const { matched, response } = await handler.handle(createBatchRequest('streaming', [
+        makePeerRequestMessage(0, '/ping'),
+        makePeerRequestMessage(1, '/ping'),
+      ]))
+
+      expect(matched).toBe(true)
+      expect(response!.status).toBe(207)
+      expect(response!.headers.get('content-encoding')).toBe('gzip')
+
+      const decompressed = response!.body!.pipeThrough(new DecompressionStream('gzip'))
+      const text = await new Response(decompressed).text()
+      expect(text).toContain(largeValue)
+    })
+
+    it('flushes each compressed batch message without waiting for the whole batch', async () => {
+      let resolveSlow!: (value: string) => void
+      const slow = new Promise<string>((resolve) => {
+        resolveSlow = resolve
+      })
+
+      const handler = createBatchHandler({
+        fast: os.handler(() => largeValue),
+        slow: os.handler(() => slow),
+      })
+
+      const { response } = await handler.handle(createBatchRequest('streaming', [
+        makePeerRequestMessage(0, '/fast'),
+        makePeerRequestMessage(1, '/slow'),
+      ]))
+
+      expect(response!.headers.get('content-encoding')).toBe('gzip')
+
+      const reader = response!.body!.pipeThrough(new DecompressionStream('gzip')).getReader()
+      const decoder = new TextDecoder()
+
+      /**
+       * A flush-capable compressor delivers the fast message while the slow one
+       * is still pending. A buffering compressor (e.g. `CompressionStream`)
+       * would hold it back until the stream ends and these reads would never
+       * resolve (test times out) because the slow handler is not resolved yet.
+       */
+      let received = ''
+      while (!received.includes(largeValue)) {
+        const { done, value } = await reader.read()
+        if (done) {
+          expect.unreachable('stream ended before the fast message was delivered')
+        }
+        received += decoder.decode(value, { stream: true })
+      }
+
+      resolveSlow('slow-result')
+
+      while (!received.includes('slow-result')) {
+        const { done, value } = await reader.read()
+        if (done) {
+          break
+        }
+        received += decoder.decode(value, { stream: true })
+      }
+
+      expect(received).toContain('slow-result')
+    })
+
+    it('compresses streaming batch responses with deflate when preferred', async () => {
+      const handler = createBatchHandler({
+        ping: os.handler(() => largeValue),
+      })
+
+      const { response } = await handler.handle(new Request('https://example.com/__batch__', {
+        method: 'POST',
+        headers: {
+          'accept-encoding': 'deflate',
+          'content-type': 'application/json',
+          'orpc-batch': 'streaming',
+        },
+        body: JSON.stringify([makePeerRequestMessage(0, '/ping')]),
+      }))
+
+      expect(response!.headers.get('content-encoding')).toBe('deflate')
+
+      const decompressed = response!.body!.pipeThrough(new DecompressionStream('deflate'))
+      const text = await new Response(decompressed).text()
+      expect(text).toContain(largeValue)
+    })
+
+    it('compresses buffered batch responses as json (unchanged behavior)', async () => {
+      const handler = createBatchHandler({
+        ping: os.handler(() => largeValue),
+      })
+
+      const { response } = await handler.handle(createBatchRequest('buffered', [
+        makePeerRequestMessage(0, '/ping'),
+        makePeerRequestMessage(1, '/ping'),
+      ]))
+
+      expect(response!.status).toBe(207)
+      expect(response!.headers.get('content-type')).toContain('application/json')
+      expect(response!.headers.get('content-encoding')).toBe('gzip')
+
+      const decompressed = response!.body!.pipeThrough(new DecompressionStream('gzip'))
+      const text = await new Response(decompressed).text()
+      expect(text).toContain(largeValue)
+    })
+
+    it('respects a custom filter that rejects batch responses', async () => {
+      const filter = vi.fn(() => false)
+
+      const handler = createBatchHandler({
+        ping: os.handler(() => largeValue),
+      }, { filter })
+
+      const { response } = await handler.handle(createBatchRequest('streaming', [
+        makePeerRequestMessage(0, '/ping'),
+      ]))
+
+      expect(filter).toHaveBeenCalled()
+      expect(response!.headers.has('content-encoding')).toBe(false)
+
+      const text = await new Response(response!.body).text()
+      expect(text).toContain(largeValue)
+    })
+
+    it('propagates cancellation through the compressed stream to the source', async () => {
+      let sourceCancelled = false
+      const source = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode(largeValue))
+        },
+        cancel() {
+          sourceCancelled = true
+        },
+      })
+
+      const handler = createHandler(new Response(source, {
+        headers: {
+          'content-type': 'application/octet-stream',
+        },
+      }))
+
+      const { response } = await handler.handle(new Request('https://example.com/__batch__', {
+        headers: {
+          'accept-encoding': 'gzip',
+          'orpc-batch': 'streaming',
+        },
+      }))
+
+      expect(response!.headers.get('content-encoding')).toBe('gzip')
+
+      const reader = response!.body!.getReader()
+      const { value } = await reader.read()
+      expect(value!.byteLength).toBeGreaterThan(0)
+
+      await reader.cancel()
+      await vi.waitFor(() => expect(sourceCancelled).toBe(true))
+    })
+
+    it('supports cancelling a compressed streaming batch response mid-batch', async () => {
+      const handler = createBatchHandler({
+        fast: os.handler(() => largeValue),
+        never: os.handler(() => new Promise(() => {})),
+      })
+
+      const { response } = await handler.handle(createBatchRequest('streaming', [
+        makePeerRequestMessage(0, '/fast'),
+        makePeerRequestMessage(1, '/never'),
+      ]))
+
+      expect(response!.headers.get('content-encoding')).toBe('gzip')
+
+      const reader = response!.body!.getReader()
+      const { value } = await reader.read()
+      expect(value!.byteLength).toBeGreaterThan(0)
+
+      await expect(reader.cancel()).resolves.toBeUndefined()
+    })
+  })
 })

packages/server/src/adapters/fetch/body-compression-plugin.ts

@@ -28,7 +28,10 @@ export interface BodyCompressionHandlerPluginOptions {
    * Override the default content-type filter used to determine which responses should be compressed.
    *
    * @warning Event stream responses are never compressed, regardless of this filter's return value.
-   * @default only responses with compressible content types are compressed.
+   * @default only responses with compressible content types are compressed. Batch responses
+   * (`orpc-batch` request header) with an `application/octet-stream` body are also compressed
+   * when the runtime exposes Node.js `zlib`/`stream` builtins — a flush-per-message compressor
+   * keeps streaming batch semantics intact.
    */
   filter?: (request: Request, response: Response) => boolean
 }
@@ -51,6 +54,10 @@ export class BodyCompressionHandlerPlugin<T extends Context> implements FetchHan
         return false
       }
 
+      if (isFlushCompressibleBatchResponse(request, response)) {
+        return options.filter ? options.filter(request, response) : true
+      }
+
       return options.filter
         ? options.filter(request, response)
         : isCompressibleContentType(contentType)
@@ -98,7 +105,22 @@ export class BodyCompressionHandlerPlugin<T extends Context> implements FetchHan
             return result
           }
 
-          const compressedBody = response.body.pipeThrough(new CompressionStream(encoding))
+          /**
+           * `CompressionStream` cannot flush between chunks, so early messages of a
+           * streaming batch response would sit in the compressor buffer until the
+           * stream ends, defeating streaming semantics. Batch responses therefore
+           * use a flush-per-message zlib stream instead (the filter guarantees it is
+           * available whenever a batch response passes through).
+           */
+          const compression = isFlushCompressibleBatchResponse(interceptorOptions.request, response)
+            ? createFlushableCompressionStream(encoding)
+            : new CompressionStream(encoding)
+
+          if (compression === undefined) {
+            return result
+          }
+
+          const compressedBody = response.body.pipeThrough(compression)
           const compressedHeaders = new Headers(response.headers)
 
           compressedHeaders.delete('content-length')
@@ -136,3 +158,71 @@ function isNoTransformCacheControl(cacheControl: string | null): boolean {
 
   return CACHE_CONTROL_NO_TRANSFORM_REGEX.test(cacheControl)
 }
+
+/**
+ * Batch responses carry a `application/octet-stream` body (length-prefixed peer messages),
+ * which the default content-type filter would never compress even though the payload is
+ * mostly JSON. They are only compressible when a flush-capable compressor is available,
+ * otherwise streaming batch responses would lose their streaming semantics.
+ */
+function isFlushCompressibleBatchResponse(request: Request, response: Response): boolean {
+  return request.headers.has('orpc-batch')
+    && !response.headers.has('content-disposition')
+    && (response.headers.get('content-type')?.startsWith('application/octet-stream') ?? false)
+    && getNodeCompressionCompat() !== undefined
+}
+
+interface NodeCompressionCompat {
+  zlib: typeof import('node:zlib')
+  stream: typeof import('node:stream')
+}
+
+/**
+ * `false` = probed and unavailable. Runtime builtins cannot appear later,
+ * so the probe runs at most once per process.
+ */
+let cachedNodeCompressionCompat: NodeCompressionCompat | false | undefined
+
+function getNodeCompressionCompat(): NodeCompressionCompat | undefined {
+  if (cachedNodeCompressionCompat === undefined) {
+    cachedNodeCompressionCompat = false
+
+    try {
+      /**
+       * Deliberately accessed via `globalThis` (not imported) so this cross-runtime
+       * file never hard-depends on node builtins — runtimes without them fall back
+       * to the uncompressed behavior.
+       */
+      // eslint-disable-next-line node/prefer-global/process
+      const zlib = globalThis.process?.getBuiltinModule?.('node:zlib')
+      // eslint-disable-next-line node/prefer-global/process
+      const stream = globalThis.process?.getBuiltinModule?.('node:stream')
+
+      if (zlib !== undefined && typeof stream?.Duplex?.toWeb === 'function') {
+        cachedNodeCompressionCompat = { zlib, stream }
+      }
+    }
+    catch {
+      // runtimes without node builtins keep the `false` sentinel
+    }
+  }
+
+  return cachedNodeCompressionCompat === false ? undefined : cachedNodeCompressionCompat
+}
+
+function createFlushableCompressionStream(
+  encoding: (typeof ORDERED_SUPPORTED_ENCODINGS)[number],
+): ReadableWritablePair<Uint8Array, Uint8Array> | undefined {
+  const compat = getNodeCompressionCompat()
+
+  if (compat === undefined) {
+    return undefined
+  }
+
+  const zlibOptions = { flush: compat.zlib.constants.Z_SYNC_FLUSH }
+  const compressor = encoding === 'gzip'
+    ? compat.zlib.createGzip(zlibOptions)
+    : compat.zlib.createDeflate(zlibOptions)
+
+  return compat.stream.Duplex.toWeb(compressor) as ReadableWritablePair<Uint8Array, Uint8Array>
+}