import net from 'node:net'
import {
  METHODS,
  STATUS_CODES,
  ServerResponse,
  IncomingMessage,
} from 'node:http'
import { invariant } from 'outvariant'
import { HttpResponseEvent, type HttpRequestEventMap } from '../../events/http'
import { RequestController } from '../../request-controller'
import {
  getRawFetchHeaders,
  recordRawFetchHeaders,
} from '../ClientRequest/utils/record-raw-headers'
import { SocketInterceptor } from '../net'
import { connectionOptionsToUrl } from '../net/utils/connection-options-to-url'
import { toBuffer } from '../../utils/buffer-utils'
import { createRequestId } from '../../create-request-id'
import { HttpRequestParser, HttpResponseParser } from './http-parser'
import { handleRequest, HandleRequestOptions } from '../../utils/handle-request'
import { isResponseError, kErrorResponse } from '../../utils/response-utils'
import { createLogger } from '../../utils/logger'
import {
  kRawSocket,
  SocketController,
  type FlushPendingDataFunction,
} from '../net/socket-controller'
import { unwrapPendingData } from '../net/utils/flush-writes'
import { FetchResponse } from '../../utils/fetch-utils'
import { cloneResponse } from '../../utils/clone-response'
import { requestContext } from '../../request-context'
import { Interceptor } from '#/src/interceptor'

const httpLogger = createLogger('http-request')

/**
 * Interceptor for HTTP requests in Node.js.
 * Routes socket connections through an HTTP parser.
 */
export class NodeHttpRequestSource extends Interceptor<HttpRequestEventMap> {
  static symbol = Symbol.for('node-http-request-source')

  protected predicate(): boolean {
    return true
  }

  protected setup(): void {
    const socketInterceptor = Interceptor.singleton(SocketInterceptor)
    socketInterceptor.apply(this)
    this.subscriptions.push(() => {
      socketInterceptor.dispose(this)
    })

    /**
     * @note Record the raw values provided to Headers set/append
     * in order to support "IncomingMessage.prototype.rawHeaders".
     * This is meant for the headers in mocked responses.
     */
    this.subscriptions.push(recordRawFetchHeaders())

    const controller = new AbortController()
    this.subscriptions.push(() => controller.abort())

    /**
     * @note The client keeps idle sockets in its keep-alive pool
     * (e.g. Undici) and writes its next requests to them. Once this
     * source is disposed, nothing handles those requests: destroy the
     * idle sockets and let the client connect anew. A request in
     * flight finishes first. Passed-through sockets are exchanging
     * with the real server and keep doing so: leave them intact.
     */
    const idleSocketDisposals = new Set<() => void>()
    this.subscriptions.push(() => {
      for (const destroyIdleSocket of idleSocketDisposals) {
        destroyIdleSocket()
      }
      idleSocketDisposals.clear()
    })

    socketInterceptor.on(
      'connection',
      ({ connectionOptions, socket, controller: socketController }) => {
        let isHttpConnection: boolean | undefined
        let requestParser: HttpRequestParser | undefined
        let tunnelUrl: URL | undefined
        let abortPendingRequest: (() => void) | undefined
        let pendingRequestController: RequestController | undefined

        /**
         * @note An exchange is in flight from the first byte the client
         * writes (e.g. request headers spanning multiple packets, parsed
         * only partially so far) until its mocked response is delivered.
         * The request handling alone settles once the listener returns
         * its verdict, while the response may still be written to the
         * socket (e.g. a streaming body).
         */
        let hasActiveExchange = false

        const destroyIdleSocket = () => {
          if (
            !hasActiveExchange &&
            socketController.readyState !== SocketController.PASSTHROUGH
          ) {
            socket.destroy()
          }
        }
        idleSocketDisposals.add(destroyIdleSocket)
        socket.once('close', () => {
          idleSocketDisposals.delete(destroyIdleSocket)
        })

        const shouldPassthrough = () => {
          return (
            socketController.readyState === SocketController.PENDING &&
            !socket.destroyed
          )
        }

        // Protocol detection runs inside a client write. Let the write finish,
        // other data observers run, and the remaining "connection" listeners
        // get their chance to claim before flushing it to the real socket.
        const passthroughNonHttp = () => {
          setImmediate(() => {
            if (shouldPassthrough()) {
              socketController.passthrough()
            }
          })
        }

        // A malformed request loses the boundary for subsequent requests.
        // Preserve the original bytes for the client and server to handle.
        const stopParsingRequests = (error: Error) => {
          httpLogger.verbose('stopping HTTP request parsing: %o', error)
          isHttpConnection = false

          if (
            pendingRequestController?.readyState === RequestController.PENDING
          ) {
            void pendingRequestController.passthrough()
          } else {
            passthroughNonHttp()
          }

          requestParser?.free(error)
        }

        /**
         * @note Capture the request context of the connection itself.
         * The socket is created synchronously within the request async
         * context (e.g. inside the patched `http.request()`), but the
         * first data may reach the socket from a foreign context (e.g.
         * a form-data stream piped into the request), where sampling
         * the request context yields nothing.
         */
        const connectionRequestContext = requestContext.getStore()

        /**
         * @note The client destroys the socket synchronously (e.g. Undici
         * on request abort) but the socket teardown events ("error",
         * "close") are emitted asynchronously, after the consumer has
         * already observed the rejected request promise. Hook into the
         * destroy itself so the pending request is aborted before any
         * of its listeners can resume.
         */
        const rawSocket = socketController[kRawSocket]
        const realSocketDestroy = rawSocket._destroy.bind(rawSocket)
        rawSocket._destroy = (error, callback) => {
          abortPendingRequest?.()
          realSocketDestroy(error, callback)
        }

        const executeRequestParser = (
          parser: HttpRequestParser,
          chunk: Buffer
        ) => {
          // llhttp pauses permanently at an upgrade boundary. Release the
          // parser after execute returns, outside its native callbacks.
          if (parser.execute(chunk) !== null) {
            socket.removeListener('data', onRequestData)
            parser.free()
            requestParser = undefined
          }
        }

        /**
         * @note Inspect the first sent packet to determine the protocol,
         * including when entering a mocked "CONNECT" tunnel.
         */
        const onRequestData = (chunk: Buffer) => {
          hasActiveExchange = true

          if (isHttpConnection === false) {
            passthroughNonHttp()
            return
          }

          /**
           * @note A mocked "CONNECT" request has established a tunnel.
           * The data that follows belongs to a new exchange addressed to
           * the tunnel target. The previous parser was freed at the upgrade
           * boundary, so detect the tunneled protocol anew.
           */
          if (tunnelUrl && !requestParser) {
            isHttpConnection = undefined

            /**
             * @note Retarget the connection to the tunnel authority.
             * The exchanges that follow belong to the tunnel target,
             * so an unclaimed exchange (HTTP or not) must pass through
             * to that target — not to the proxy, which never actually
             * established this tunnel — like a real established tunnel
             * relays its traffic.
             */
            socketController.reset({
              host: tunnelUrl.hostname,
              port: Number(tunnelUrl.port) || 80,
              path: null,
            })
          }

          if (requestParser) {
            executeRequestParser(requestParser, toBuffer(chunk))
            return
          }

          const httpMessage = chunk.toString()
          const httpMethod = httpMessage.split(' ')[0] || ''

          // Stop parsing connections that do not carry HTTP traffic.
          if (!METHODS.includes(httpMethod.toUpperCase())) {
            isHttpConnection = false
            passthroughNonHttp()
            return
          }

          isHttpConnection = true

          const baseUrl =
            tunnelUrl ?? connectionOptionsToUrl(connectionOptions, socket)

          httpLogger.verbose('handling http message %o', {
            httpMessage,
            httpMethod,
            baseUrl,
          })

          // Get the request initiator from the async context, falling
          // back to the context captured at the connection time, then
          // to the underlying socket.
          const requestContextValue =
            requestContext.getStore() ?? connectionRequestContext
          const initiator = requestContextValue?.initiator || socket

          requestParser = new HttpRequestParser({
            onError: stopParsingRequests,
            connectionOptions: {
              method: httpMethod,
              url: baseUrl,
            },
            /**
             * @note The message boundary ends the current exchange.
             * Schedule the controller reset so the next write on this
             * (kept-alive) socket opens a new exchange and buffers for
             * its own verdict instead of following the settled one
             * (e.g. leaking a mocked request to the server of a
             * previously passed-through exchange).
             */
            onMessageComplete: () => {
              socketController.scheduleReset()
            },
            onRequest: async (parsedRequest, requestAbortController) => {
              const request =
                requestContextValue?.transformRequest?.(parsedRequest) ??
                parsedRequest

              /**
               * @note A subsequent request arriving on a kept-alive socket
               * that has already been handled (passed through or mocked).
               * Clients like Undici reuse sockets without emitting the
               * "free" event, so reset the controller here, at the HTTP
               * message boundary, to handle the new request from the
               * pending state again.
               */
              if (socketController['readyState'] !== SocketController.PENDING) {
                socketController.reset()
              }

              const requestId = createRequestId()
              const requestLogger = requestContextValue?.logger ?? httpLogger

              httpLogger.verbose('received a parsed HTTP request %o', {
                method: request.method,
                url: request.url,
              })

              const requestController = new RequestController(
                request,
                {
                  respondWith: async (rawResponse) => {
                    httpLogger.verbose('respondWith() %o', {
                      status: rawResponse.status,
                      statusText: rawResponse.statusText,
                      hasBody: rawResponse.body != null,
                    })

                    /**
                     * @note The client may destroy the socket (e.g. on request
                     * abort) moments before a response arrives. A destroyed
                     * socket cannot be claimed and has no one reading it.
                     */
                    if (socket.destroyed) {
                      return
                    }

                    socketController.claim()

                    const originalResponse = FetchResponse.from(rawResponse, {
                      url: request.url,
                    })
                    const [response, responseClone] =
                      !isResponseError(originalResponse) &&
                      this.emitter.listenerCount('response') > 0
                        ? cloneResponse(originalResponse)
                        : [originalResponse, null]

                    /**
                     * @note A successful mocked response to a "CONNECT"
                     * request establishes a tunnel to the requested authority
                     * (e.g. "127.0.0.1:80"). The exchange that follows on this
                     * socket is addressed to that authority, not to the proxy.
                     */
                    if (request.method === 'CONNECT' && response.ok) {
                      tunnelUrl = new URL(`http://${request.url}`)
                      socket.on('data', onRequestData)
                    }

                    const respond = async () => {
                      try {
                        await this.respondWith({
                          socket: socketController[kRawSocket],
                          request: context.request,
                          response,
                          connectionRequestContext,
                        })
                      } finally {
                        hasActiveExchange = false
                      }

                      /**
                       * @note This source got disposed while the request
                       * was in flight. The response is delivered: end the
                       * connection, like a server responding with
                       * "Connection: close", so the client does not reuse
                       * it for requests nobody handles.
                       */
                      if (controller.signal.aborted) {
                        socket.end()
                      }
                    }

                    if (responseClone) {
                      await this.emitter.emitAsPromise(
                        new HttpResponseEvent({
                          initiator,
                          requestId,
                          request: context.request,
                          response: responseClone,
                          responseType: 'mock',
                        })
                      )
                    }

                    if (socket.connecting) {
                      // Send a mocked response once the socket connects, just like the real server would.
                      // This preserves the correct order of events (e.g. connect, then data).
                      socket.once('connect', respond)
                    } else {
                      /**
                       * @note Reused sockets stay connected between requests and will not
                       * emit "connect" anymore. If that's the case, respond immediately.
                       */
                      await respond()
                    }
                  },
                  errorWith: (reason) => {
                    if (reason instanceof Error) {
                      socket.destroy(reason)
                    }
                  },
                  passthrough: () => {
                    const realSocket = socketController.passthrough(
                      isHttpConnection === false
                        ? undefined
                        : this.#modifyHttpHeaders(context.request)
                    )

                    if (isHttpConnection === false) {
                      return
                    }

                    if (this.emitter.listenerCount('response') > 0) {
                      httpLogger.verbose(
                        'found "response" listener, corking socket reads'
                      )

                      /**
                       * Suspend the delivery of the original response to the client
                       * until the "response" event listeners settle. This guarantees
                       * that the request promise (e.g. `await fetch()`) does not
                       * resolve before the listeners are done. The real socket keeps
                       * emitting data for the response parser meanwhile.
                       */
                      socketController.corkReads()

                      let responseParserDisposed = false
                      let responseComplete = false
                      let hasFinalResponse = false
                      const responseParser = new HttpResponseParser({
                        onError: (error) => {
                          disposeResponseParser(error)
                          socketController.uncorkReads()
                        },
                        onMessageComplete: (status) => {
                          responseComplete = status >= 200 || status === 101
                        },
                        onResponse: async (response) => {
                          hasFinalResponse =
                            response.status >= 200 || response.status === 101
                          httpLogger.verbose(
                            'HTTP response parser parsed: %d %s',
                            response.status,
                            response.statusText
                          )

                          if (isResponseError(response)) {
                            httpLogger.verbose(
                              'response is an error response, uncorking socket reads...'
                            )

                            socketController.uncorkReads()
                            return
                          }

                          FetchResponse.setUrl(request.url, response)

                          try {
                            httpLogger.verbose('emitting "response" event')
                            await this.emitter.emitAsPromise(
                              new HttpResponseEvent({
                                initiator,
                                requestId,
                                request: context.request,
                                response,
                                responseType: 'original',
                              })
                            )
                          } finally {
                            httpLogger.verbose('uncorking socket reads')
                            socketController.uncorkReads()

                            /**
                             * @note Informational responses other than
                             * "101 Switching Protocols" are followed by a final
                             * response on the same connection. Keep gating that
                             * final response on the "response" event listeners.
                             */
                            if (
                              !responseParserDisposed &&
                              response.status < 200 &&
                              response.status !== 101
                            ) {
                              socketController.corkReads()
                            }
                          }
                        },
                      })

                      const onResponseData = (chunk: Buffer) => {
                        responseParser.execute(chunk)

                        // Free only after llhttp returns from its callbacks.
                        if (responseComplete) {
                          disposeResponseParser()
                        }
                      }

                      const onResponseEnd = () => {
                        disposeResponseParser()

                        // Without a final response, release EOF now. A half-open
                        // socket cannot close until the client consumes it.
                        if (!hasFinalResponse) {
                          socketController.uncorkReads()
                        }
                      }

                      const disposeResponseParser = (error?: Error) => {
                        responseParserDisposed = true
                        realSocket.removeListener('data', onResponseData)
                        realSocket.removeListener('end', onResponseEnd)
                        realSocket.removeListener('close', onResponseEnd)
                        responseParser.free(error)
                      }

                      realSocket
                        .on('data', onResponseData)
                        .once('end', onResponseEnd)
                        .once('close', onResponseEnd)
                    }
                  },
                },
                {
                  logger: requestLogger,
                  requestId,
                }
              )

              invariant(
                socketController['readyState'] === SocketController.PENDING,
                'CANNOT HANDLE ALREADY HANDLED REQUEST',
                request.method,
                request.url,
                socketController['readyState']
              )

              /**
               * @note Create a request resolution context.
               * This is so modifications to the "request" in upstream interceptors
               * are correctly picked up by the underlying HTTP interceptor.
               */
              const context: HandleRequestOptions = {
                initiator,
                requestId,
                request,
                controller: requestController,
                emitter: this.emitter,
                logger: requestLogger,
              }

              /**
               * @note The client destroying the socket while the request
               * is still pending means the request was aborted (e.g. via
               * `AbortController`). Abort the parsed request so its
               * handling settles and late interactions with the request
               * controller become controlled errors.
               */
              abortPendingRequest = () => {
                if (
                  requestController.readyState === RequestController.PENDING
                ) {
                  requestAbortController.abort()
                }
              }

              pendingRequestController = requestController

              try {
                await handleRequest(context)
              } finally {
                pendingRequestController = undefined
                abortPendingRequest = undefined
              }
            },
          })

          // Forward the first frame to the parser.
          executeRequestParser(requestParser, toBuffer(chunk))
        }

        socket.on('data', onRequestData)
        socket.on('close', () => requestParser?.free())

        /**
         * @note A client that waits for the server to speak first (e.g. a
         * SQL handshake) starts reading without writing anything, so
         * protocol detection never runs. Pass such connections through
         * once the client reads and still no bytes have arrived. Clients
         * that write after an async setup (e.g. Undici loading its parser)
         * attach their readers right before writing; "setImmediate" lets
         * those writes, including ones from nested ticks of the "connect"
         * listeners, land first.
         */
        const onClientRead = (event: string | symbol) => {
          if (event !== 'data' && event !== 'readable') {
            return
          }

          rawSocket.removeListener('newListener', onClientRead)

          setImmediate(() => {
            if (isHttpConnection === undefined && shouldPassthrough()) {
              socketController.passthrough()
            }
          })
        }

        rawSocket.on('newListener', onClientRead)

        if (
          rawSocket.listenerCount('data') +
            rawSocket.listenerCount('readable') >
          0
        ) {
          onClientRead('data')
        }
      },
      {
        signal: controller.signal,
      }
    )
  }

  private async respondWith(args: {
    socket: net.Socket
    request: Request
    response: Response
    connectionRequestContext: ReturnType<typeof requestContext.getStore>
  }): Promise<void> {
    const { socket, request, response, connectionRequestContext } = args

    if (socket.destroyed) {
      return
    }

    if (isResponseError(response)) {
      /**
       * @note Reference the error response on the socket error so the
       * client-side interceptors (e.g. fetch) can surface it to the
       * consumer as the reason behind the failed request. Keep the
       * reference non-enumerable so the error remains observably
       * identical for the clients that expose it as-is.
       */
      socket.destroy(
        Object.defineProperty(new TypeError('Network error'), kErrorResponse, {
          value: response,
          enumerable: false,
        })
      )
      return
    }

    invariant(
      !socket.connecting,
      'Failed to mock a response for "%s %s": socket has not connected',
      request.method,
      request.url
    )

    /**
     * Use native server response handling in Node.js.
     * @see https://github.com/nodejs/node/blob/13eb80f3b718452213e0fc449702aefbbfe4110f/lib/_http_server.js#L202
     */
    const incomingMessage = new IncomingMessage(socket)

    /**
     * @note Describe the request method so the response body is
     * handled appropriately (e.g. "HEAD" responses must not write
     * a body). The HTTP version is deliberately left unset: with it,
     * `ServerResponse` frames bodies of unknown length as chunked,
     * polluting the mocked response headers with "Transfer-Encoding"
     * the mock never specified.
     */
    incomingMessage.method = request.method

    const serverResponse = new ServerResponse(incomingMessage)

    const responseSocket = new net.Socket()

    responseSocket._writeGeneric = (writev, data, encoding, callback) => {
      unwrapPendingData(data, (chunk, encoding) => {
        socket.push(toBuffer(chunk), encoding)
      })
      callback?.()
    }

    responseSocket._destroy = (
      error: Error | null,
      callback: (error: Error | null) => void
    ) => {
      /**
       * Only destroy the socket on stream errors.
       * On a clean end, the socket is already signaled via `socket.push(null)`
       * in the main response flow. Destroying it here prematurely would prevent
       * the client from processing the response (e.g. calling `response.destroy()`).
       * @see https://github.com/mswjs/interceptors/issues/738
       */
      if (error) {
        socket.destroy()
      }

      callback(null)
    }

    responseSocket.on('drain', () => serverResponse.emit('drain'))
    serverResponse.assignSocket(responseSocket)

    serverResponse.removeHeader('connection')
    serverResponse.removeHeader('date')

    const rawResponseHeaders = getRawFetchHeaders(response.headers)
    serverResponse.writeHead(
      response.status,
      response.statusText || STATUS_CODES[response.status],
      rawResponseHeaders
    )

    /**
     * @note Override the socket's `_destroy` before writing the response body.
     * The underlying TCP handle (from `socket.connect()`) makes `_destroy` async
     * (`_handle.close()` callback), which delays the 'error' event. Since the real
     * TCP connection is irrelevant for mocked responses, take the synchronous path
     * so that user-initiated `response.destroy(error)` emits the error promptly.
     * This must happen before `serverResponse.end()` because the HTTP parser may
     * fire the 'response' event synchronously during `socket.push()`.
     */
    socket._destroy = function (
      error: Error | null,
      callback: (error: Error | null) => void
    ) {
      if (error) {
        /**
         * Emit the error event as a microtask instead of relying on the default
         * `process.nextTick(emitErrorNT)` from `callback(error)`. This is necessary
         * because `respondWith` runs inside a microtask (from `await reader.read()`).
         * A resolved promise continuation (from toWebResponse) is queued as
         * another microtask during the same phase. Since microtasks are drained before
         * nextTick, the test's `await` would resolve before the error event fires.
         * Using `queueMicrotask` ensures the error event is emitted within the current
         * microtask phase, before other queued microtasks.
         */
        queueMicrotask(() => this.emit('error', error))
      }

      callback(null)

      /**
       * @note `net.Socket` is constructed with `emitClose: false`, so Node's
       * stream destroy machinery does not emit `'close'` automatically; the
       * stock `net.Socket._destroy` only emits it via `_handle.close()`.
       * Since this override replaces `_destroy`, emit `'close'` here so the
       * mocked socket completes its lifecycle (otherwise consumers waiting
       * on `'close'`, like `http.ClientRequest`, hang).
       */
      const emitClose = () => this.emit('close', error != null)

      /**
       * @note Emit "close" within the request context the socket was
       * created in, as the "close" of a real socket is. A client keeping
       * this socket in its pool (e.g. Undici) may reconnect from its
       * "close" listener for a request dispatched in the meantime; that
       * connection then captures this context and the request stays
       * attributed to its initiator (e.g. `fetch`) instead of being
       * left to nobody. Node.js 24+ scopes callbacks to the context
       * current when they were scheduled, which is not that context.
       */
      process.nextTick(() => {
        if (connectionRequestContext) {
          requestContext.run(connectionRequestContext, emitClose)
        } else {
          emitClose()
        }
      })
    }

    if (response.body) {
      const reader = response.body.getReader()

      try {
        while (true) {
          const { done, value } = await reader.read()

          if (done) {
            serverResponse.end()
            break
          }

          if (!serverResponse.write(value)) {
            await new Promise<void>((resolve) => {
              serverResponse.once('drain', resolve)
            })
          }
        }
      } catch {
        /**
         * @note Delay the socket destruction to allow the event loop
         * to flush already-pushed response data (headers + body chunks)
         * through the HTTP parser. Without this, the socket is destroyed
         * on the same tick as `socket.push(data)` and the client never
         * reads the response.
         */
        await new Promise<void>((resolve) => process.nextTick(resolve))
        socket.destroy()
        return
      }
    } else {
      serverResponse.end()
    }

    /**
     * @note Self-delimiting responses (chunked, explicit "Content-Length",
     * or bodiless by definition) must NOT signal the end-of-stream.
     * The client parser completes them from their framing alone, and
     * ending the socket would kill the kept-alive connection that
     * agents pool and reuse for subsequent requests.
     */
    const isSelfDelimitingResponse =
      request.method === 'HEAD' ||
      response.headers.has('content-length') ||
      response.headers.has('transfer-encoding') ||
      !FetchResponse.isResponseWithBody(response.status)

    if (
      /**
       * @note A non-2xx response to a "CONNECT" request means the proxy
       * refused the tunnel. Real proxies terminate the connection after
       * sending such a response, regardless of its framing.
       */
      (request.method === 'CONNECT' && !response.ok) ||
      (request.method !== 'CONNECT' && !isSelfDelimitingResponse)
    ) {
      /**
       * @note Defer the end-of-stream signal so the HTTP parser has a chance
       * to process already-pushed response data and fire the 'response' event
       * before the socket is ended. Without this, the parser marks the response
       * as "complete" before the client can interact with it (e.g. `response.destroy()`).
       */
      await new Promise<void>((resolve) => process.nextTick(resolve))
      socket.push(null)
    }
  }

  #modifyHttpHeaders(request: Request): FlushPendingDataFunction {
    const transformRequestMessage = (
      httpMessage: string | Buffer,
      encoding?: BufferEncoding | 'buffer'
    ): string | Buffer => {
      /**
       * @note Socket can write a buffer (e.g. uploaded file) even before
       * it writes the HTTP message. Bypass those cases.
       */
      if (encoding === 'buffer') {
        return httpMessage
      }

      const parts = httpMessage.toString(encoding).split('\r\n')
      const headersEndIndex = parts.findIndex((field) => field === '')
      const httpMessageHeaderPairs = parts.slice(1, headersEndIndex)

      // Extract raw [name, value] tuples from the wire format so they
      // can be compared against the request's raw fetch headers.
      const httpMessageRawHeaders = httpMessageHeaderPairs.map(
        (line): [string, string] => {
          const separatorIndex = line.indexOf(': ')
          return [line.slice(0, separatorIndex), line.slice(separatorIndex + 2)]
        }
      )

      const requestRawHeaders = getRawFetchHeaders(request.headers)

      // If the raw headers from the outgoing HTTP message and the request
      // headers are identical, send the message as-is to avoid the cost
      // (and side effects) of reserializing the headers block.
      const headersUnchanged =
        httpMessageRawHeaders.length === requestRawHeaders.length &&
        httpMessageRawHeaders.every((tuple, index) => {
          const requestTuple = requestRawHeaders[index]
          return tuple[0] === requestTuple[0] && tuple[1] === requestTuple[1]
        })

      if (headersUnchanged) {
        return httpMessage
      }

      const httpMessageHeaders = FetchResponse.parseRawHeaders(
        httpMessageHeaderPairs.flatMap((header) => header.split(': '))
      )

      const visitedHeaders = new Set<string>()

      for (const [headerName] of requestRawHeaders) {
        const normalizedHeaderName = headerName.toLowerCase()

        if (visitedHeaders.has(normalizedHeaderName)) {
          continue
        }

        visitedHeaders.add(normalizedHeaderName)

        /**
         * @note Forbidden Fetch headers (e.g. Host, Origin, Connection)
         * are stripped from `request.headers` but remain in the raw
         * headers list. Skip them so the original values from the
         * outgoing HTTP message are preserved.
         */
        const headerValue = request.headers.get(headerName)
        if (headerValue === null) {
          continue
        }

        // Use the merged value from Headers to correctly handle
        // appended headers (e.g. "1, 2" instead of just "2").
        httpMessageHeaders.set(headerName, headerValue)
      }

      visitedHeaders.clear()

      /**
       * @note Remove the headers deleted from the request (e.g. via
       * `request.headers.delete()`). The raw headers start as a copy of
       * the outgoing HTTP message headers and track every mutation, so a
       * header present in the message but absent from the raw headers has
       * been deleted. Forbidden headers stripped from `request.headers`
       * remain in the raw headers and are preserved as-is.
       */
      const rawHeaderNames = new Set(
        requestRawHeaders.map(([headerName]) => headerName.toLowerCase())
      )

      for (const [headerName] of httpMessageRawHeaders) {
        if (!rawHeaderNames.has(headerName.toLowerCase())) {
          httpMessageHeaders.delete(headerName)
        }
      }

      const httpMessageHeadersString = Array.from(httpMessageHeaders)
        .map(([name, value]) => `${name}: ${value}`)
        .join('\r\n')
      parts.splice(1, headersEndIndex - 1, httpMessageHeadersString)

      return parts.join('\r\n')
    }

    return (pendingData, encoding, callback) => {
      if (Array.isArray(pendingData)) {
        pendingData[0].chunk = transformRequestMessage(
          pendingData[0].chunk,
          pendingData[0].encoding
        )
      } else {
        pendingData = transformRequestMessage(pendingData, encoding)
      }

      callback(pendingData)
    }
  }
}