fix(node:http) allow CONNECT in node http/https servers (#22756)

### What does this PR do?
Fixes https://github.com/oven-sh/bun/issues/22755
Fixes https://github.com/oven-sh/bun/issues/19790
Fixes https://github.com/oven-sh/bun/issues/16372
### How did you verify your code works?

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/js/node/_http_server.ts

@@ -352,8 +352,9 @@ Server.prototype[EventEmitter.captureRejectionSymbol] = function (err, event, ..
 Server.prototype[Symbol.asyncDispose] = function () {
   const { resolve, reject, promise } = Promise.withResolvers();
   this.close(function (err, ...args) {
-    if (err) reject(err);
-    else resolve(...args);
+    if (err) {
+      reject(err);
+    } else resolve(...args);
   });
   return promise;
 };
@@ -474,7 +475,6 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
     if (tls) {
       this.serverName = tls.serverName || host || "localhost";
     }
-
     this[serverSymbol] = Bun.serve<any>({
       idleTimeout: 0, // nodejs dont have a idleTimeout by default
       tls,
@@ -528,10 +528,30 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
         if (isAncientHTTP) {
           http_req.httpVersion = "1.0";
         }
+        if (method === "CONNECT") {
+          // Handle CONNECT method for HTTP tunneling/proxy
+          if (server.listenerCount("connect") > 0) {
+            // For CONNECT, emit the event and let the handler respond
+            // Don't assign the socket to a response for CONNECT
+            // The handler should write the raw response
+            socket[kEnableStreaming](true);
+            const { promise, resolve } = $newPromiseCapability(Promise);
+            socket.once("close", resolve);
+            server.emit("connect", http_req, socket, kEmptyBuffer);
+            return promise;
+          } else {
+            // Node.js will close the socket and will NOT respond with 400 Bad Request
+            socketHandle.close();
+          }
+          return;
+        }
+        socket[kEnableStreaming](false);
+
         const http_res = new ResponseClass(http_req, {
           [kHandle]: handle,
           [kRejectNonStandardBodyWrites]: server.rejectNonStandardBodyWrites,
         });
+
         setIsNextIncomingMessageHTTPS(prevIsNextIncomingMessageHTTPS);
         handle.onabort = onServerRequestEvent.bind(socket);
         // start buffering data if any, the user will need to resume() or .on("data") to read it
@@ -677,6 +697,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort
       //   return promise;
       // },
     });
+
     getBunServerAllClosedPromise(this[serverSymbol]).$then(emitCloseNTServer.bind(this));
     isHTTPS = this[serverSymbol].protocol === "https";
     // always set strict method validation to true for node.js compatibility
@@ -784,14 +805,18 @@ function onServerClientError(ssl: boolean, socket: unknown, errorCode: number, r
   }
 }
 
+const kBytesWritten = Symbol("kBytesWritten");
+const kEnableStreaming = Symbol("kEnableStreaming");
 const NodeHTTPServerSocket = class Socket extends Duplex {
   bytesRead = 0;
   connecting = false;
   timeout = 0;
+  [kBytesWritten] = 0;
   [kHandle];
   server: Server;
   _httpMessage;
   _secureEstablished = false;
+  #pendingCallback = null;
   constructor(server: Server, handle, encrypted) {
     super();
     this.server = server;
@@ -799,15 +824,56 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     this._secureEstablished = !!handle?.secureEstablished;
     handle.onclose = this.#onClose.bind(this);
     handle.duplex = this;
+
     this.encrypted = encrypted;
     this.on("timeout", onNodeHTTPServerSocketTimeout);
   }
 
   get bytesWritten() {
-    return this[kHandle]?.response?.getBytesWritten?.() ?? 0;
+    const handle = this[kHandle];
+    return handle
+      ? (handle.response?.getBytesWritten?.() ?? handle.bytesWritten ?? this[kBytesWritten] ?? 0)
+      : (this[kBytesWritten] ?? 0);
+  }
+  set bytesWritten(value) {
+    this[kBytesWritten] = value;
   }
-  set bytesWritten(value) {}
 
+  [kEnableStreaming](enable: boolean) {
+    const handle = this[kHandle];
+    if (handle) {
+      if (enable) {
+        handle.ondata = this.#onData.bind(this);
+        handle.ondrain = this.#onDrain.bind(this);
+      } else {
+        handle.ondata = undefined;
+        handle.ondrain = undefined;
+      }
+    }
+  }
+  #onDrain() {
+    const handle = this[kHandle];
+    this[kBytesWritten] = handle ? (handle.response?.getBytesWritten?.() ?? handle.bytesWritten ?? 0) : 0;
+    const callback = this.#pendingCallback;
+    if (callback) {
+      this.#pendingCallback = null;
+      (callback as Function)();
+    }
+    this.emit("drain");
+  }
+  #onData(chunk, last) {
+    if (chunk) {
+      this.push(chunk);
+    }
+    if (last) {
+      const handle = this[kHandle];
+      if (handle) {
+        handle.ondata = undefined;
+      }
+
+      this.push(null);
+    }
+  }
   #closeHandle(handle, callback) {
     this[kHandle] = undefined;
     handle.onclose = this.#onCloseForDestroy.bind(this, callback);
@@ -822,8 +888,10 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
   }
   #onClose() {
     this[kHandle] = null;
+
     const message = this._httpMessage;
     const req = message?.req;
+
     if (req && !req.complete && !req[kHandle]?.upgraded) {
       // At this point the socket is already destroyed; let's avoid UAF
       req[kHandle] = undefined;
@@ -833,6 +901,7 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
         req.destroy();
       }
     }
+    this.emit("close");
   }
   #onCloseForDestroy(closeCallback) {
     this.#onClose();
@@ -871,9 +940,10 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
       $isCallable(callback) && callback(err);
       return;
     }
+    handle.ondata = undefined;
     if (handle.closed) {
       const onclose = handle.onclose;
-      handle.onclose = null;
+      handle.onclose = undefined;
       if ($isCallable(onclose)) {
         onclose.$call(handle);
       }
@@ -890,7 +960,8 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
       callback();
       return;
     }
-    this.#closeHandle(handle, callback);
+    handle.end();
+    callback();
   }
 
   get localAddress() {
@@ -998,7 +1069,20 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     return this;
   }
 
-  _write(_chunk, _encoding, _callback) {}
+  _write(_chunk, _encoding, _callback) {
+    const handle = this[kHandle];
+    // only enable writting if we can drain
+    let err;
+    try {
+      if (handle && handle.ondrain && !handle.write(_chunk, _encoding)) {
+        this.#pendingCallback = _callback;
+        return false;
+      }
+    } catch (e) {
+      err = e;
+    }
+    err ? _callback(err) : _callback();
+  }
 
   pause() {
     const handle = this[kHandle];
@@ -1006,6 +1090,7 @@ const NodeHTTPServerSocket = class Socket extends Duplex {
     if (response) {
       response.pause();
     }
+
     return super.pause();
   }
 
@@ -1138,8 +1223,12 @@ function ServerResponse(req, options): void {
 
     if (handle) {
       this[kHandle] = handle;
+    } else {
+      this[kHandle] = req[kHandle];
     }
     this[kRejectNonStandardBodyWrites] = options[kRejectNonStandardBodyWrites] ?? false;
+  } else {
+    this[kHandle] = req[kHandle];
   }
 
   this.statusCode = 200;
@@ -1622,6 +1711,7 @@ function ServerResponse_finalDeprecated(chunk, encoding, callback) {
     chunk = Buffer.from(chunk, encoding);
   }
   const req = this.req;
+
   const shouldEmitClose = req && req.emit && !this.finished;
   if (!this.headersSent) {
     let data = this[firstWriteSymbol];