fix(ext/node): work correctly with wrapper Response objects, use correct rawHeaders structure

ext/http/00_serve.ts

@@ -58,6 +58,7 @@ import {
   ResponsePrototype,
   toInnerResponse,
 } from "ext:deno_fetch/23_response.js";
+import { headerListFromHeaders } from "ext:deno_fetch/20_headers.js";
 import {
   abortRequest,
   fromInnerRequest,
@@ -500,8 +501,15 @@ function fastSyncResponseOrStream(
     return;
   }
 
-  const stream = respBody.streamOrStatic;
-  const body = stream.body;
+  let stream;
+  let body;
+  if (respBody.streamOrStatic) {
+    stream = respBody.streamOrStatic;
+    body = stream.body;
+  } else {
+    stream = respBody;
+    body = respBody;
+  }
   if (body !== undefined) {
     // We ensure the response has not been consumed yet in the caller of this
     // function.
@@ -632,8 +640,19 @@ function mapToCallback(context, callback, onError) {
       return;
     }
 
-    const status = inner.status;
-    const headers = inner.headerList;
+    let status;
+    let headers;
+    let body;
+    if (inner) {
+      status = inner.status;
+      headers = inner.headerList;
+      body = inner.body;
+    } else {
+      status = response.status;
+      headers = headerListFromHeaders(response.headers);
+      body = response.body;
+    }
+
     if (headers && headers.length > 0) {
       if (headers.length == 1) {
         op_http_set_response_header(req, headers[0][0], headers[0][1]);
@@ -642,7 +661,7 @@ function mapToCallback(context, callback, onError) {
       }
     }
 
-    fastSyncResponseOrStream(req, inner.body, status, innerRequest);
+    fastSyncResponseOrStream(req, body, status, innerRequest);
   };
 
   if (TRACING_ENABLED) {

ext/node/polyfills/http.ts

@@ -1763,6 +1763,8 @@ Object.defineProperty(ServerResponse.prototype, "connection", {
   ),
 });
 
+const kRawHeaders = Symbol("rawHeaders");
+
 // TODO(@AaronO): optimize
 export class IncomingMessageForServer extends NodeReadable {
   #headers: Record<string, string>;
@@ -1802,7 +1804,7 @@ export class IncomingMessageForServer extends NodeReadable {
     this.method = "";
     this.socket = socket;
     this.upgrade = null;
-    this.rawHeaders = [];
+    this[kRawHeaders] = [];
     socket?.on("error", (e) => {
       if (this.listenerCount("error") > 0) {
         this.emit("error", e);
@@ -1825,7 +1827,7 @@ export class IncomingMessageForServer extends NodeReadable {
   get headers() {
     if (!this.#headers) {
       this.#headers = {};
-      const entries = headersEntries(this.rawHeaders);
+      const entries = headersEntries(this[kRawHeaders]);
       for (let i = 0; i < entries.length; i++) {
         const entry = entries[i];
         this.#headers[entry[0]] = entry[1];
@@ -1838,6 +1840,16 @@ export class IncomingMessageForServer extends NodeReadable {
     this.#headers = val;
   }
 
+  get rawHeaders() {
+    const entries = headersEntries(this[kRawHeaders]);
+    const out = new Array(entries.length * 2);
+    for (let i = 0; i < entries.length; i++) {
+      out[i * 2] = entries[i][0];
+      out[i * 2 + 1] = entries[i][1];
+    }
+    return out;
+  }
+
   // connection is deprecated, but still tested in unit test.
   get connection() {
     return this.socket;
@@ -1942,7 +1954,7 @@ export class ServerImpl extends EventEmitter {
       req.upgrade =
         request.headers.get("connection")?.toLowerCase().includes("upgrade") &&
         request.headers.get("upgrade");
-      req.rawHeaders = request.headers;
+      req[kRawHeaders] = request.headers;
 
       if (req.upgrade && this.listenerCount("upgrade") > 0) {
         const { conn, response } = upgradeHttpRaw(request);

tests/specs/node/wrapped_http_response/__test__.jsonc

@@ -0,0 +1,8 @@
+{
+  "tempDir": true,
+
+  "steps": [{
+    "args": "run -A main.ts",
+    "output": "done\n"
+  }]
+}

tests/specs/node/wrapped_http_response/main.ts

@@ -0,0 +1,54 @@
+// Adapted from https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/listener.ts
+// and https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/server.ts
+import {
+  buildOutgoingHttpHeaders,
+  Response as WrappedResponse,
+} from "./response.ts";
+import { createServer, OutgoingHttpHeaders, ServerResponse } from "node:http";
+
+Object.defineProperty(globalThis, "Response", {
+  value: WrappedResponse,
+});
+
+const { promise, resolve } = Promise.withResolvers<void>();
+
+const responseViaResponseObject = async (
+  res: Response,
+  outgoing: ServerResponse,
+) => {
+  const resHeaderRecord: OutgoingHttpHeaders = buildOutgoingHttpHeaders(
+    res.headers,
+  );
+
+  if (res.body) {
+    const buffer = await res.arrayBuffer();
+    resHeaderRecord["content-length"] = buffer.byteLength;
+
+    outgoing.writeHead(res.status, resHeaderRecord);
+    outgoing.end(new Uint8Array(buffer));
+  } else {
+    outgoing.writeHead(res.status, resHeaderRecord);
+    outgoing.end();
+  }
+};
+
+const server = createServer((_req, res) => {
+  const response = new Response("Hello, world!");
+  return responseViaResponseObject(response, res);
+});
+
+using _server = {
+  [Symbol.dispose]() {
+    server.close();
+  },
+};
+
+server.listen(0, async () => {
+  const { port } = server.address() as { port: number };
+  const response = await fetch(`http://localhost:${port}`);
+  await response.text();
+  resolve();
+});
+
+await promise;
+console.log("done");

tests/specs/node/wrapped_http_response/response.ts

@@ -0,0 +1,113 @@
+// Adapted from https://github.com/honojs/node-server/blob/1eb73c6d985665e75458ddd08c23bbc1dbdc7bcd/src/response.ts
+// deno-lint-ignore-file no-explicit-any
+//
+import type { OutgoingHttpHeaders } from "node:http";
+
+interface InternalBody {
+  source: string | Uint8Array | FormData | Blob | null;
+  stream: ReadableStream;
+  length: number | null;
+}
+
+const GlobalResponse = globalThis.Response;
+
+const responseCache = Symbol("responseCache");
+const getResponseCache = Symbol("getResponseCache");
+export const cacheKey = Symbol("cache");
+
+export const buildOutgoingHttpHeaders = (
+  headers: Headers | HeadersInit | null | undefined,
+): OutgoingHttpHeaders => {
+  const res: OutgoingHttpHeaders = {};
+  if (!(headers instanceof Headers)) {
+    headers = new Headers(headers ?? undefined);
+  }
+
+  const cookies = [];
+  for (const [k, v] of headers) {
+    if (k === "set-cookie") {
+      cookies.push(v);
+    } else {
+      res[k] = v;
+    }
+  }
+  if (cookies.length > 0) {
+    res["set-cookie"] = cookies;
+  }
+  res["content-type"] ??= "text/plain; charset=UTF-8";
+
+  return res;
+};
+
+export class Response {
+  #body?: BodyInit | null;
+  #init?: ResponseInit;
+
+  [getResponseCache](): typeof GlobalResponse {
+    delete (this as any)[cacheKey];
+    return ((this as any)[responseCache] ||= new GlobalResponse(
+      this.#body,
+      this.#init,
+    ));
+  }
+
+  constructor(body?: BodyInit | null, init?: ResponseInit) {
+    this.#body = body;
+    if (init instanceof Response) {
+      const cachedGlobalResponse = (init as any)[responseCache];
+      if (cachedGlobalResponse) {
+        this.#init = cachedGlobalResponse;
+        // instantiate GlobalResponse cache and this object always returns value from global.Response
+        this[getResponseCache]();
+        return;
+      } else {
+        this.#init = init.#init;
+      }
+    } else {
+      this.#init = init;
+    }
+
+    if (
+      typeof body === "string" ||
+      typeof (body as ReadableStream)?.getReader !== "undefined"
+    ) {
+      let headers =
+        (init?.headers || { "content-type": "text/plain; charset=UTF-8" }) as
+          | Record<string, string>
+          | Headers
+          | OutgoingHttpHeaders;
+      if (headers instanceof Headers) {
+        headers = buildOutgoingHttpHeaders(headers);
+      }
+
+      (this as any)[cacheKey] = [init?.status || 200, body, headers];
+    }
+  }
+}
+[
+  "body",
+  "bodyUsed",
+  "headers",
+  "ok",
+  "redirected",
+  "status",
+  "statusText",
+  "trailers",
+  "type",
+  "url",
+].forEach((k) => {
+  Object.defineProperty(Response.prototype, k, {
+    get() {
+      return this[getResponseCache]()[k];
+    },
+  });
+});
+["arrayBuffer", "blob", "clone", "formData", "json", "text"].forEach((k) => {
+  Object.defineProperty(Response.prototype, k, {
+    value: function () {
+      return this[getResponseCache]()[k]();
+    },
+  });
+});
+Object.setPrototypeOf(Response, GlobalResponse);
+Object.setPrototypeOf(Response.prototype, GlobalResponse.prototype);

tests/unit_node/http_test.ts

@@ -2041,3 +2041,42 @@ Deno.test("[node/http] 'close' event is emitted on ServerResponse object when th
   await new Promise((resolve) => server.close(resolve));
   assert(responseCloseEmitted);
 });
+
+Deno.test("[node/http] rawHeaders are in flattened format", async () => {
+  const getHeader = (req: IncomingMessage, name: string) => {
+    const idx = req.rawHeaders.indexOf(name);
+    if (idx < 0) {
+      throw new Error(`Header ${name} not found`);
+    }
+    return [name, req.rawHeaders[idx + 1]];
+  };
+  const { promise, resolve } = Promise.withResolvers<void>();
+  const server = http.createServer((req, res) => {
+    resolve();
+    // TODO(nathanwhit): the raw headers should not be lowercased, they should be
+    // exactly as they appeared in the request
+    assertEquals(getHeader(req, "content-type"), [
+      "content-type",
+      "text/plain",
+    ]);
+    assertEquals(getHeader(req, "set-cookie"), [
+      "set-cookie",
+      "foo=bar",
+    ]);
+    res.end();
+  });
+
+  server.listen(0, async () => {
+    const { port } = server.address() as { port: number };
+    const response = await fetch(`http://localhost:${port}`, {
+      headers: {
+        "Set-Cookie": "foo=bar",
+        "Content-Type": "text/plain",
+      },
+    });
+    await response.body?.cancel();
+  });
+
+  await promise;
+  await new Promise((resolve) => server.close(resolve));
+});