fix(server): prevent compression of server-sent events in CompressionPlugin

Author: unnoq

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

@@ -517,4 +517,31 @@ describe('compressionPlugin', () => {
 
     expect(filter).toHaveBeenCalledWith(expect.any(Request), response)
   })
+
+  it('should not compress event iterator responses', async () => {
+    const handler = new RPCHandler(os.handler(async function* () {
+      yield 'event1'
+      yield 'event2'
+    }), {
+      plugins: [
+        new CompressionPlugin(),
+      ],
+    })
+
+    const { response } = await handler.handle(new Request('https://example.com/', {
+      method: 'POST',
+      headers: {
+        'content-type': 'text/event-stream',
+        'accept-encoding': 'gzip',
+      },
+      body: JSON.stringify({}),
+    }))
+
+    expect(response?.headers.get('content-encoding')).toBeNull()
+    expect(response?.status).toBe(200)
+
+    const text = await response?.text()
+    expect(text).toContain('event1')
+    expect(text).toContain('event2')
+  })
 })

packages/server/src/adapters/node/compression-plugin.test.ts

@@ -150,4 +150,27 @@ describe('compressionPlugin', () => {
     expect(res.status).toBe(500)
     expect(res.headers['content-encoding']).toBeUndefined()
   })
+
+  it('should not compress event iterator responses', async () => {
+    const res = await request(async (req: IncomingMessage, res: ServerResponse) => {
+      const handler = new RPCHandler(os.handler(async function* () {
+        yield 'yield1'
+        yield 'yield2'
+      }), {
+        plugins: [
+          new CompressionPlugin(),
+        ],
+      })
+
+      await handler.handle(req, res)
+    })
+      .post('/')
+      .set('accept-encoding', 'gzip, deflate')
+      .send({ input: 'test' })
+
+    expect(res.status).toBe(200)
+    expect(res.headers['content-encoding']).toBeUndefined()
+    expect(res.text).toContain('yield1')
+    expect(res.text).toContain('yield2')
+  })
 })

packages/server/src/adapters/node/compression-plugin.ts

@@ -1,9 +1,16 @@
+import type { NodeHttpRequest, NodeHttpResponse } from '@orpc/standard-server-node'
 import type { Context } from '../../context'
 import type { NodeHttpHandlerOptions } from './handler'
 import type { NodeHttpHandlerPlugin } from './plugin'
 import compression from '@orpc/interop/compression'
 
 export interface CompressionPluginOptions extends compression.CompressionOptions {
+  /**
+   * A filter function to determine if a response should be compressed.
+   * This function is called in addition to the default compression checks
+   * and allows for custom compression logic based on the request and response.
+   */
+  filter?: (req: NodeHttpRequest, res: NodeHttpResponse) => boolean
 }
 /**
  * The Compression Plugin adds response compression to the Node.js HTTP Server.
@@ -14,7 +21,22 @@ export class CompressionPlugin<T extends Context> implements NodeHttpHandlerPlug
   private readonly compressionHandler: ReturnType<typeof compression>
 
   constructor(options: CompressionPluginOptions = {}) {
-    this.compressionHandler = compression(options)
+    this.compressionHandler = compression({
+      ...options,
+      filter: (req, res) => {
+        if (res.getHeader('content-type')?.toString().startsWith('text/event-stream')) {
+          return false
+        }
+
+        if (!compression.filter(req, res)) {
+          return false
+        }
+
+        return options.filter
+          ? options.filter(req, res)
+          : true
+      },
+    })
   }
 
   initRuntimeAdapter(options: NodeHttpHandlerOptions<T>): void {