feat: Support WebSocket API routes, Upgrade requests

packages/next/src/export/routes/app-route.ts

@@ -85,6 +85,12 @@ export async function exportAppRoute(
     const module = await RouteModuleLoader.load<AppRouteRouteModule>(filename)
     const response = await module.handle(request, context)
 
+    if (response === 'Upgraded') {
+      return {
+        revalidate: 0,
+      }
+    }
+
     const isValidStatus = response.status < 400 || response.status === 404
     if (!isValidStatus) {
       return { revalidate: 0 }

packages/next/src/server/base-server.ts

@@ -2237,6 +2237,9 @@ export default abstract class Server<ServerOptions extends Options = Options> {
             )
 
             const response = await routeModule.handle(request, context)
+            if (response === 'Upgraded') {
+              return null
+            }
 
             ;(req as any).fetchMetrics = (
               context.renderOpts as any

packages/next/src/server/future/route-modules/app-route/module.ts

@@ -1,8 +1,11 @@
 import type { NextConfig } from '../../../config-shared'
 import type { AppRouteRouteDefinition } from '../../route-definitions/app-route-route-definition'
 import type { AppConfig } from '../../../../build/utils'
-import type { NextRequest } from '../../../web/spec-extension/request'
 import type { PrerenderManifest } from '../../../../build'
+import {
+  isUpgradeError,
+  type NextRequest,
+} from '../../../web/spec-extension/request'
 
 import {
   RouteModule,
@@ -453,14 +456,18 @@ export class AppRouteRouteModule extends RouteModule<
   public async handle(
     request: NextRequest,
     context: AppRouteRouteHandlerContext
-  ): Promise<Response> {
+  ): Promise<Response | 'Upgraded'> {
     try {
       // Execute the route to get the response.
       const response = await this.execute(request, context)
 
       // The response was handled, return it.
       return response
     } catch (err) {
+      if (isUpgradeError(err)) {
+        return 'Upgraded'
+      }
+
       // Try to resolve the error to a response, else throw it again.
       const response = resolveHandlerError(err)
       if (!response) throw err

packages/next/src/server/lib/router-server.ts

@@ -1,5 +1,5 @@
 // this must come first as it includes require hooks
-import type { WorkerRequestHandler, WorkerUpgradeHandler } from './types'
+import type { WorkerRequestHandler } from './types'
 import type { DevBundler } from './router-utils/setup-dev-bundler'
 import type { NextUrlWithParsedQuery } from '../request-meta'
 // This is required before other imports to ensure the require hook is setup.
@@ -24,6 +24,10 @@ import { removePathPrefix } from '../../shared/lib/router/utils/remove-path-pref
 import setupCompression from 'next/dist/compiled/compression'
 import { NoFallbackError } from '../base-server'
 import { signalFromNodeResponse } from '../web/spec-extension/adapters/next-request'
+import {
+  handleClientError,
+  markSocketUpgraded,
+} from '../web/spec-extension/request-upgrade'
 import { isPostpone } from './router-utils/is-postpone'
 
 import {
@@ -64,7 +68,7 @@ export async function initialize(opts: {
   experimentalTestProxy?: boolean
   experimentalHttpsServer?: boolean
   startServerSpan?: Span
-}): Promise<[WorkerRequestHandler, WorkerUpgradeHandler]> {
+}): Promise<WorkerRequestHandler> {
   if (!process.env.NODE_ENV) {
     // @ts-ignore not readonly
     process.env.NODE_ENV = opts.dev ? 'development' : 'production'
@@ -178,8 +182,28 @@ export async function initialize(opts: {
     developmentBundler?.ensureMiddleware
   )
 
+  opts.server?.on('clientError', handleClientError)
   const requestHandlerImpl: WorkerRequestHandler = async (req, res) => {
-    if (compress) {
+    const isUpgradeReq =
+      req?.method === 'GET' &&
+      req?.headers?.connection?.toLowerCase() === 'upgrade'
+
+    if (
+      isUpgradeReq &&
+      opts.dev &&
+      developmentBundler &&
+      req.url?.includes(`/_next/webpack-hmr`)
+    ) {
+      markSocketUpgraded(req.socket)
+
+      return developmentBundler.hotReloader.onHMR(
+        req,
+        req.socket,
+        Buffer.alloc(0)
+      )
+    }
+
+    if (compress && !isUpgradeReq) {
       // @ts-expect-error not express req/res
       compress(req, res, () => {})
     }
@@ -300,7 +324,7 @@ export async function initialize(opts: {
       } = await resolveRoutes({
         req,
         res,
-        isUpgradeReq: false,
+        isUpgradeReq,
         signal: signalFromNodeResponse(res),
         invokedOutputs,
       })
@@ -571,47 +595,5 @@ export async function initialize(opts: {
   }
   requestHandlers[opts.dir] = requestHandler
 
-  const upgradeHandler: WorkerUpgradeHandler = async (req, socket, head) => {
-    try {
-      req.on('error', (_err) => {
-        // TODO: log socket errors?
-        // console.error(_err);
-      })
-      socket.on('error', (_err) => {
-        // TODO: log socket errors?
-        // console.error(_err);
-      })
-
-      if (opts.dev && developmentBundler) {
-        if (req.url?.includes(`/_next/webpack-hmr`)) {
-          return developmentBundler.hotReloader.onHMR(req, socket, head)
-        }
-      }
-
-      const { matchedOutput, parsedUrl } = await resolveRoutes({
-        req,
-        res: socket as any,
-        isUpgradeReq: true,
-        signal: signalFromNodeResponse(socket),
-      })
-
-      // TODO: allow upgrade requests to pages/app paths?
-      // this was not previously supported
-      if (matchedOutput) {
-        return socket.end()
-      }
-
-      if (parsedUrl.protocol) {
-        return await proxyRequest(req, socket as any, parsedUrl, head)
-      }
-
-      // If there's no matched output, we don't handle the request as user's
-      // custom WS server may be listening on the same path.
-    } catch (err) {
-      console.error('Error handling upgrade request', err)
-      socket.end()
-    }
-  }
-
-  return [requestHandler, upgradeHandler]
+  return requestHandler
 }

packages/next/src/server/lib/start-server.ts

@@ -6,7 +6,7 @@ import '../require-hook'
 
 import type { IncomingMessage, ServerResponse } from 'http'
 import type { SelfSignedCertificate } from '../../lib/mkcert'
-import type { WorkerRequestHandler, WorkerUpgradeHandler } from './types'
+import type { WorkerRequestHandler } from './types'
 
 import fs from 'fs'
 import v8 from 'v8'
@@ -116,17 +116,6 @@ export async function startServer(
     }
     throw new Error('Invariant request handler was not setup')
   }
-  let upgradeHandler: WorkerUpgradeHandler = async (
-    req,
-    socket,
-    head
-  ): Promise<void> => {
-    if (handlersPromise) {
-      await handlersPromise
-      return upgradeHandler(req, socket, head)
-    }
-    throw new Error('Invariant upgrade handler was not setup')
-  }
 
   // setup server listener as fast as possible
   if (selfSignedCertificate && !isDev) {
@@ -182,15 +171,6 @@ export async function startServer(
   if (keepAliveTimeout) {
     server.keepAliveTimeout = keepAliveTimeout
   }
-  server.on('upgrade', async (req, socket, head) => {
-    try {
-      await upgradeHandler(req, socket, head)
-    } catch (err) {
-      socket.destroy()
-      Log.error(`Failed to handle request for ${req.url}`)
-      console.error(err)
-    }
-  })
 
   let portRetryCount = 0
 
@@ -291,7 +271,7 @@ export async function startServer(
         process.on('uncaughtException', exception)
         process.on('unhandledRejection', exception)
 
-        const initResult = await getRequestHandlers({
+        requestHandler = await getRequestHandlers({
           dir,
           port,
           isDev,
@@ -303,8 +283,6 @@ export async function startServer(
           experimentalTestProxy: !!isExperimentalTestProxy,
           experimentalHttpsServer: !!selfSignedCertificate,
         })
-        requestHandler = initResult[0]
-        upgradeHandler = initResult[1]
 
         const startServerProcessDuration =
           performance.mark('next-start-end') &&

packages/next/src/server/lib/types.ts

@@ -1,14 +1,6 @@
 import type { IncomingMessage, ServerResponse } from 'http'
 
-import type { Duplex } from 'stream'
-
 export type WorkerRequestHandler = (
   req: IncomingMessage,
   res: ServerResponse
 ) => Promise<any>
-
-export type WorkerUpgradeHandler = (
-  req: IncomingMessage,
-  socket: Duplex,
-  head: Buffer
-) => any

packages/next/src/server/next.ts

@@ -7,7 +7,7 @@ import type { UrlWithParsedQuery } from 'url'
 import type { NextConfigComplete } from './config-shared'
 import type { IncomingMessage, ServerResponse } from 'http'
 import type { NextUrlWithParsedQuery } from './request-meta'
-import type { WorkerRequestHandler, WorkerUpgradeHandler } from './lib/types'
+import type { WorkerRequestHandler } from './lib/types'
 
 import './require-hook'
 import './node-polyfill-crypto'
@@ -268,41 +268,21 @@ class NextCustomServer extends NextServer {
 
   // @ts-expect-error These are initialized in prepare()
   protected requestHandler: WorkerRequestHandler
-  // @ts-expect-error These are initialized in prepare()
-  protected upgradeHandler: WorkerUpgradeHandler
 
   async prepare() {
     const { getRequestHandlers } =
       require('./lib/start-server') as typeof import('./lib/start-server')
 
     const isNodeDebugging = !!checkNodeDebugType()
 
-    const initResult = await getRequestHandlers({
+    this.requestHandler = await getRequestHandlers({
       dir: this.options.dir!,
       port: this.options.port || 3000,
       isDev: !!this.options.dev,
       hostname: this.options.hostname || 'localhost',
       minimalMode: this.options.minimalMode,
       isNodeDebugging: !!isNodeDebugging,
     })
-    this.requestHandler = initResult[0]
-    this.upgradeHandler = initResult[1]
-  }
-
-  private setupWebSocketHandler(
-    customServer?: import('http').Server,
-    _req?: IncomingMessage
-  ) {
-    if (!this.didWebSocketSetup) {
-      this.didWebSocketSetup = true
-      customServer = customServer || (_req?.socket as any)?.server
-
-      if (customServer) {
-        customServer.on('upgrade', async (req, socket, head) => {
-          this.upgradeHandler(req, socket, head)
-        })
-      }
-    }
   }
 
   getRequestHandler() {
@@ -311,8 +291,6 @@ class NextCustomServer extends NextServer {
       res: ServerResponse,
       parsedUrl?: UrlWithParsedQuery
     ) => {
-      this.setupWebSocketHandler(this.options.httpServer, req)
-
       if (parsedUrl) {
         req.url = formatUrl(parsedUrl)
       }
@@ -323,7 +301,6 @@ class NextCustomServer extends NextServer {
 
   async render(...args: Parameters<Server['render']>) {
     let [req, res, pathname, query, parsedUrl] = args
-    this.setupWebSocketHandler(this.options.httpServer, req as any)
 
     if (!pathname.startsWith('/')) {
       console.error(`Cannot render page with path "${pathname}"`)

packages/next/src/server/web/edge-route-module-wrapper.ts

@@ -110,6 +110,9 @@ export class EdgeRouteModuleWrapper {
 
     // Get the response from the handler.
     const res = await this.routeModule.handle(request, context)
+    if (res === 'Upgraded') {
+      throw new Error('Unreachable - Edge routes cannot be upgraded')
+    }
 
     const waitUntilPromises = [internal_getCurrentFunctionWaitUntil()]
     if (context.renderOpts.waitUntil) {

packages/next/src/server/web/spec-extension/adapters/next-request.ts

@@ -102,6 +102,7 @@ export class NextRequestAdapter {
       // @ts-expect-error - see https://github.com/whatwg/fetch/pull/1457
       duplex: 'half',
       signal,
+      rawRequest: request.originalRequest,
       // geo
       // ip
       // nextConfig