feat(server): compression plugin filter option should override default content type filter (#880)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- New Features
- Added support to override default compression decisions via a custom
filter, enabling compression for custom content types.
- Bug Fixes
- Ensured event-stream (Server-Sent Events) and Event Iterator responses
are never compressed, even with a custom filter.
  - Confirmed no compression occurs when responses lack a content-type.
- Documentation
- Clarified default compression behavior and the filter option,
including the event-stream exception.
- Tests
- Expanded coverage for custom filter behavior, event-stream handling,
and missing content-type scenarios.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: unnoq

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

@@ -313,6 +313,36 @@ describe('compressionPlugin', () => {
     expect(response?.status).toBe(204)
   })
 
+  it('should not compress when response has no content-type', async () => {
+    const handler = new RPCHandler(os.handler(() => 'output'), {
+      strictGetMethodPluginEnabled: false,
+      plugins: [
+        new CompressionPlugin(),
+      ],
+      adapterInterceptors: [
+        async (options) => {
+          const result = await options.next()
+          result.response?.headers.delete('content-type') // Simulate no content-type
+          return result
+        },
+      ],
+    })
+
+    const { response } = await handler.handle(new Request('https://example.com/', {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json',
+        'accept-encoding': 'gzip',
+      },
+      body: JSON.stringify({}),
+    }))
+
+    expect(response?.headers.has('content-encoding')).toBe(false)
+    expect(response?.headers.has('content-type')).toBe(false)
+    expect(response?.status).toBe(200)
+    await expect(response?.text()).resolves.toContain('output')
+  })
+
   it('should not compress non-compressible content types', async () => {
     const handler = new RPCHandler(os.handler(() => 'output'), {
       plugins: [
@@ -492,7 +522,50 @@ describe('compressionPlugin', () => {
     expect(response).toBeUndefined()
   })
 
-  it('should not compress when custom filter returns false', async () => {
+  it('should override filter and compress if filter return true', async () => {
+    const filter = vi.fn(() => true)
+
+    const handler = new RPCHandler(os.handler(() => largeText), {
+      plugins: [
+        new CompressionPlugin({
+          filter,
+        }),
+      ],
+      interceptors: [
+        async (options) => {
+          const result = await options.next()
+
+          if (!result.matched) {
+            return result
+          }
+
+          return {
+            ...result,
+            response: {
+              ...result.response,
+              // image/jpeg is not compressible by default
+              body: new Blob([largeText], { type: 'image/jpeg' }),
+            },
+          }
+        },
+      ],
+    })
+
+    const { response } = await handler.handle(new Request('https://example.com/', {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json',
+        'accept-encoding': 'gzip',
+      },
+      body: JSON.stringify({}),
+    }))
+
+    expect(response?.headers.get('content-encoding')).toBe('gzip')
+
+    expect(filter).toHaveBeenCalledWith(expect.any(Request), expect.any(Response))
+  })
+
+  it('should not compress when filter returns false', async () => {
     const filter = vi.fn(() => false)
 
     const handler = new RPCHandler(os.handler(() => largeText), {
@@ -524,7 +597,7 @@ describe('compressionPlugin', () => {
       yield 'event2'
     }), {
       plugins: [
-        new CompressionPlugin(),
+        new CompressionPlugin({ filter: () => true }),
       ],
     })
 

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

@@ -28,9 +28,10 @@ export interface CompressionPluginOptions {
   threshold?: number
 
   /**
-   * 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.
+   * Override the default content-type filter used to determine which responses should be compressed.
+   *
+   * @warning [Event Iterator](https://orpc.unnoq.com/docs/event-iterator) responses are never compressed, regardless of this filter's return value.
+   * @default only responses with compressible content types are compressed.
    */
   filter?: (request: Request, response: Response) => boolean
 }
@@ -45,12 +46,26 @@ export interface CompressionPluginOptions {
 export class CompressionPlugin<T extends Context> implements FetchHandlerPlugin<T> {
   private readonly encodings: Exclude<CompressionPluginOptions['encodings'], undefined>
   private readonly threshold: Exclude<CompressionPluginOptions['threshold'], undefined>
-  private readonly filter: CompressionPluginOptions['filter']
+  private readonly filter: Exclude<CompressionPluginOptions['filter'], undefined>
 
   constructor(options: CompressionPluginOptions = {}) {
     this.encodings = options.encodings ?? ORDERED_SUPPORTED_ENCODINGS
     this.threshold = options.threshold ?? 1024
-    this.filter = options.filter
+    this.filter = (request, response) => {
+      const hasContentDisposition = response.headers.has('content-disposition')
+      const contentType = response.headers.get('content-type')
+
+      /**
+       * Never compress Event Iterator responses.
+       */
+      if (!hasContentDisposition && contentType?.startsWith('text/event-stream')) {
+        return false
+      }
+
+      return options.filter
+        ? options.filter(request, response)
+        : isCompressibleContentType(contentType)
+    }
   }
 
   initRuntimeAdapter(options: FetchHandlerOptions<T>): void {
@@ -71,7 +86,6 @@ export class CompressionPlugin<T extends Context> implements FetchHandlerPlugin<
       if (
         response.headers.has('content-encoding') // already encoded
         || response.headers.has('transfer-encoding') // already encoded or chunked
-        || !isCompressibleContentType(response.headers.get('content-type')) // not compressible
         || isNoTransformCacheControl(response.headers.get('cache-control')) // no-transform directive
       ) {
         return result
@@ -93,7 +107,7 @@ export class CompressionPlugin<T extends Context> implements FetchHandlerPlugin<
         return result
       }
 
-      if (this.filter && !this.filter(options.request, response)) {
+      if (!this.filter(options.request, response)) {
         return result
       }
 

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

@@ -5,6 +5,8 @@ import { CompressionPlugin } from './compression-plugin'
 import { RPCHandler } from './rpc-handler'
 
 describe('compressionPlugin', () => {
+  const largeText = 'x'.repeat(2000) // 2KB of text, above default threshold
+
   it('should compress response when accept-encoding includes gzip', async () => {
     const res = await request(async (req: IncomingMessage, res: ServerResponse) => {
       const handler = new RPCHandler(os.handler(() => 'output'), {
@@ -100,6 +102,68 @@ describe('compressionPlugin', () => {
     expect(res.headers['content-encoding']).toBe('deflate')
   })
 
+  it('should override default filter and compress if filter returns true', async () => {
+    const filter = vi.fn(() => true)
+
+    const res = await request(async (req: IncomingMessage, res: ServerResponse) => {
+      const handler = new RPCHandler(os.handler(() => 'output'), {
+        plugins: [
+          new CompressionPlugin({ filter }),
+        ],
+        interceptors: [
+          async (options) => {
+            const result = await options.next()
+
+            if (!result.matched) {
+              return result
+            }
+
+            return {
+              ...result,
+              response: {
+                ...result.response,
+                // image/jpeg is not compressible by default
+                body: new Blob([largeText], { type: 'image/jpeg' }),
+              },
+            }
+          },
+        ],
+      })
+
+      await handler.handle(req, res)
+    })
+      .post('/')
+      .set('accept-encoding', 'gzip, deflate')
+      .send({ input: 'test' })
+
+    expect(res.status).toBe(200)
+    expect(res.headers['content-encoding']).toBe('gzip')
+
+    expect(filter).toHaveBeenCalledWith(expect.any(Object), expect.any(Object))
+  })
+
+  it('should not compress if filter return false', async () => {
+    const filter = vi.fn(() => false)
+
+    const res = await request(async (req: IncomingMessage, res: ServerResponse) => {
+      const handler = new RPCHandler(os.handler(() => 'output'), {
+        plugins: [
+          new CompressionPlugin({ filter }),
+        ],
+      })
+
+      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(filter).toHaveBeenCalledWith(expect.any(Object), expect.any(Object))
+  })
+
   it('should throw if rootInterceptor throws', async () => {
     const res = await request(async (req: IncomingMessage, res: ServerResponse) => {
       const handler = new RPCHandler(os.handler(() => 'output'), {
@@ -158,7 +222,7 @@ describe('compressionPlugin', () => {
         yield 'yield2'
       }), {
         plugins: [
-          new CompressionPlugin(),
+          new CompressionPlugin({ filter: () => true }),
         ],
       })
 

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

@@ -6,9 +6,10 @@ 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.
+   * Override the default content-type filter used to determine which responses should be compressed.
+   *
+   * @warning [Event Iterator](https://orpc.unnoq.com/docs/event-iterator) responses are never compressed, regardless of this filter's return value.
+   * @default only responses with compressible content types are compressed.
    */
   filter?: (req: NodeHttpRequest, res: NodeHttpResponse) => boolean
 }
@@ -24,17 +25,19 @@ export class CompressionPlugin<T extends Context> implements NodeHttpHandlerPlug
     this.compressionHandler = compression({
       ...options,
       filter: (req, res) => {
-        if (res.getHeader('content-type')?.toString().startsWith('text/event-stream')) {
-          return false
-        }
+        const hasContentDisposition = res.hasHeader('content-disposition')
+        const contentType = res.getHeader('content-type')?.toString()
 
-        if (!compression.filter(req, res)) {
+        /**
+         * Never compress Event Iterator responses.
+         */
+        if (!hasContentDisposition && contentType?.startsWith('text/event-stream')) {
           return false
         }
 
         return options.filter
           ? options.filter(req, res)
-          : true
+          : compression.filter(req, res)
       },
     })
   }