Fix local dev server streaming resources (to unblock streaming search…

… and cody streaming)

client/web/dev/utils/get-api-proxy-settings.ts

@@ -1,6 +1,10 @@
+import type * as http from 'http'
+import * as zlib from 'zlib'
+
 import { Options, responseInterceptor } from 'http-proxy-middleware'
 
 import { ENVIRONMENT_CONFIG, HTTPS_WEB_SERVER_URL } from './environment-config'
+import { STREAMING_ENDPOINTS } from './should-compress-response'
 
 // One of the API routes: "/-/sign-in".
 const PROXY_ROUTES = ['/.api', '/search/stream', '/-', '/.auth']
@@ -34,7 +38,7 @@ export function getAPIProxySettings(options: GetAPIProxySettingsOptions): ProxyS
         // Prevent automatic call of res.end() in `onProxyRes`. It is handled by `responseInterceptor`.
         selfHandleResponse: true,
         // eslint-disable-next-line @typescript-eslint/no-misused-promises, @typescript-eslint/require-await
-        onProxyRes: responseInterceptor(async (responseBuffer, proxyRes) => {
+        onProxyRes: conditionalResponseInterceptor(STREAMING_ENDPOINTS, async (responseBuffer, proxyRes) => {
             // Propagate cookies to enable authentication on the remote server.
             if (proxyRes.headers['set-cookie']) {
                 // Remove `Secure` and `SameSite` from `set-cookie` headers.
@@ -108,3 +112,78 @@ function getRemoteJsContextScript(remoteIndexHTML: string): string {
 
     return remoteIndexHTML.slice(remoteJsContextStart, remoteJsContextEnd) + jsContextChanges
 }
+
+type Interceptor = (
+    buffer: Buffer,
+    proxyRes: http.IncomingMessage,
+    req: http.IncomingMessage,
+    res: http.ServerResponse
+) => Promise<Buffer | string>
+
+function conditionalResponseInterceptor(
+    ignoredRoutes: string[],
+    interceptor: Interceptor
+): (proxyRes: http.IncomingMessage, req: http.IncomingMessage, res: http.ServerResponse) => Promise<void> {
+    const unconditionalResponseInterceptor = responseInterceptor(interceptor)
+
+    return async function proxyResResponseInterceptor(
+        proxyRes: http.IncomingMessage,
+        req: http.IncomingMessage,
+        res: http.ServerResponse
+    ): Promise<void> {
+        let shouldStream = false
+        for (const route of ignoredRoutes) {
+            if (req.url?.startsWith(route)) {
+                shouldStream = true
+            }
+        }
+
+        if (shouldStream) {
+            return new Promise(resolve => {
+                res.setHeader('content-type', 'text/event-stream')
+                const _proxyRes = decompress(proxyRes, proxyRes.headers['content-encoding'])
+
+                _proxyRes.on('data', (chunk: any) => res.write(chunk))
+                _proxyRes.on('end', () => {
+                    res.end()
+                    resolve()
+                })
+                _proxyRes.on('error', () => {
+                    res.end()
+                    resolve()
+                })
+            })
+        }
+
+        return unconditionalResponseInterceptor(proxyRes, req, res)
+    }
+}
+
+function decompress<TReq extends http.IncomingMessage = http.IncomingMessage>(
+    proxyRes: TReq,
+    contentEncoding?: string
+): TReq | zlib.Gunzip | zlib.Inflate | zlib.BrotliDecompress {
+    let _proxyRes: TReq | zlib.Gunzip | zlib.Inflate | zlib.BrotliDecompress = proxyRes
+    let decompress
+
+    switch (contentEncoding) {
+        case 'gzip':
+            decompress = zlib.createGunzip()
+            break
+        case 'br':
+            decompress = zlib.createBrotliDecompress()
+            break
+        case 'deflate':
+            decompress = zlib.createInflate()
+            break
+        default:
+            break
+    }
+
+    if (decompress) {
+        _proxyRes.pipe(decompress)
+        _proxyRes = decompress
+    }
+
+    return _proxyRes
+}

client/web/dev/utils/should-compress-response.ts

@@ -1,18 +1,15 @@
 import compression, { CompressionFilter } from 'compression'
 
+export const STREAMING_ENDPOINTS = ['/search/stream', '/.api/compute/stream', '/.api/completions/stream']
+
 export const shouldCompressResponse: CompressionFilter = (request, response) => {
     // Disable compression because gzip buffers the full response
     // before sending it, blocking streaming on some endpoints.
-    if (request.path.startsWith('/search/stream')) {
-        return false
-    }
-
-    if (request.path.startsWith('/.api/compute/stream')) {
-        return false
-    }
 
-    if (request.path.startsWith('/.api/.completions/stream')) {
-        return false
+    for (const endpoint of STREAMING_ENDPOINTS) {
+        if (request.path.startsWith(endpoint)) {
+            return false
+        }
     }
 
     // fallback to standard filter function

package.json

@@ -268,7 +268,7 @@
     "graphql": "^15.4.0",
     "graphql-schema-linter": "^2.0.1",
     "gulp": "^4.0.2",
-    "http-proxy-middleware": "^1.1.2",
+    "http-proxy-middleware": "^2.0.6",
     "identity-obj-proxy": "^3.0.0",
     "jest": "^28.1.0",
     "jest-canvas-mock": "^2.3.0",