fix(start-server): Stop reusing http.ServerResponse for Upgrade

requests. It seems to cause the socket to hang

http/start-server/build-responder.js

@@ -10,7 +10,7 @@ const {
   HTTP2_HEADER_PATH,
 } = http2.constants
 
-export default function (computeResponse, {endAfterBody = true} = {}) {
+export default function (computeResponse) {
   return async (nodeRequest, nodeResponse) => {
     const request = {
       version: nodeRequest.httpVersion,
@@ -30,18 +30,12 @@ export default function (computeResponse, {endAfterBody = true} = {}) {
     nodeResponse.writeHead(response.status, response.headers)
 
     if (!response.body) {
-      if (endAfterBody) {
-        nodeResponse.end()
-      }
+      nodeResponse.end()
     } else if (
       typeof response.body === 'string' ||
       response.body instanceof Buffer
     ) {
-      if (endAfterBody) {
-        nodeResponse.end(response.body)
-      } else {
-        nodeResponse.write(response.body)
-      }
+      nodeResponse.end(response.body)
     } else {
       response.body.pipe(nodeResponse)
     }

http/start-server/build-upgrade-responder.js

@@ -0,0 +1,45 @@
+import {Buffer} from 'node:buffer'
+import {STATUS_CODES} from 'node:http'
+import {parseHttp1Body} from '../parse-body.js'
+
+export default function (computeResponse) {
+  return async (nodeRequest, socket, head) => {
+    const request = {
+      version: nodeRequest.httpVersion,
+      method: nodeRequest.method,
+      url: nodeRequest.url,
+      headers: nodeRequest.headers,
+      body: await parseHttp1Body(nodeRequest),
+    }
+
+    const response = await computeResponse(request)
+
+    socket.write(
+      [
+        `HTTP/1.1 ${response.status} ${STATUS_CODES[response.status]}`,
+        ...Object.keys(response.headers ?? {}).map(
+          (name) => `${name}: ${response.headers[name]}`,
+        ),
+        '\r\n',
+      ].join('\r\n'),
+    )
+
+    if (
+      (typeof response.body === 'string' && response.body) ||
+      response.body instanceof Buffer
+    ) {
+      socket.write(response.body)
+    } else if (response.pipe) {
+      response.body.pipe(socket, {end: false})
+      await new Promise((resolve) => {
+        response.body.once('end', resolve)
+      })
+    }
+
+    if (response.status === 101) {
+      response.upgrade(socket, head)
+    } else {
+      socket.end()
+    }
+  }
+}

http/start-server/index.js

@@ -1,28 +1,12 @@
-import {ServerResponse} from 'node:http'
 import {createServer} from 'httpx-server'
 import buildResponder from './build-responder.js'
+import buildUpgradeResponder from './build-upgrade-responder.js'
 
 export default async function ({port, ...options}, computeResponse) {
   const server = createServer(options)
 
   server.on('request', buildResponder(computeResponse))
-
-  const writeUpgradeResponse = buildResponder(computeResponse, {
-    endAfterBody: false,
-  })
-  server.on('upgrade', async (nodeRequest, socket, head) => {
-    const nodeResponse = new ServerResponse(nodeRequest)
-    nodeResponse.assignSocket(socket)
-    const response = await writeUpgradeResponse(nodeRequest, nodeResponse)
-
-    if (response.status === 101) {
-      nodeResponse.flushHeaders()
-      nodeResponse.detachSocket(socket)
-      response.upgrade(socket, head)
-    } else {
-      nodeResponse.end()
-    }
-  })
+  server.on('upgrade', buildUpgradeResponder(computeResponse))
 
   await new Promise((resolve, reject) => {
     server.listen(port, resolve)

http/start-server/index.test.js

@@ -177,14 +177,22 @@ test('omitting unused fields', async (t) => {
 
 test('allowing upgrading to WebSocket', async (t) => {
   const webSocketHashingConstant = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
+  const keyRegex = /^[+/\dA-Za-z]{22}==$/
 
   const server = await startServer({port: 10_007}, (request) => {
     if (
       request.headers.connection === 'Upgrade' &&
       request.headers.upgrade === 'websocket'
     ) {
       t.log(request)
+
       const key = request.headers['sec-websocket-key']
+      if (!key || !keyRegex.test(key)) {
+        return {
+          status: 400,
+          body: 'Invalid Key',
+        }
+      }
 
       return {
         status: 101,
@@ -238,6 +246,35 @@ test('allowing upgrading to WebSocket', async (t) => {
     },
   )
 
+  t.like(
+    await sendRequest({
+      method: 'GET',
+      url: 'http://localhost:10007',
+      headers: {
+        Connection: 'Upgrade',
+        Upgrade: 'websocket',
+      },
+    }),
+    {
+      status: 400,
+    },
+  )
+
+  t.like(
+    await sendRequest({
+      method: 'GET',
+      url: 'http://localhost:10007',
+      headers: {
+        Connection: 'Upgrade',
+        Upgrade: 'websocket',
+        'Sec-WebSocket-Key': 'invalid',
+      },
+    }),
+    {
+      status: 400,
+    },
+  )
+
   const ws = new WebSocket('ws://localhost:10007')
   await new Promise((resolve) => {
     ws.once('open', resolve)