fix(ext/node): support JS underlying stream in TLS (#30465)

Fixes https://github.com/denoland/deno/issues/20594

This implements `JSStreamSocket` which drives the TLS underlying stream
in `rustls_tokio_stream` using 2 sets of channels. One for piping the
encrypted protocol transport and the other for plaintext application
data.

This fixes connecting to `npm:mssql`:
```js
import sql from "npm:mssql";

const sqlConfig = {
  server: "localhost",
  user: "divy",
  password: "123",
  database: "master",
  options: {
    trustServerCertificate: true,
  },
};

const pool = await sql.connect(sqlConfig);
const result = await pool.request().query(`SELECT * FROM sys.databases`);
```

ext/node/polyfills/_tls_wrap.js

@@ -35,8 +35,10 @@ import {
   isArrayBufferView,
 } from "ext:deno_node/internal/util/types.ts";
 import { startTlsInternal } from "ext:deno_net/02_tls.js";
-import { internals } from "ext:core/mod.js";
+import { core, internals } from "ext:core/mod.js";
 import {
+  op_node_tls_handshake,
+  op_node_tls_start,
   op_tls_canonicalize_ipv4_address,
   op_tls_key_null,
   op_tls_key_static,
@@ -47,6 +49,8 @@ const kIsVerified = Symbol("verified");
 const kPendingSession = Symbol("pendingSession");
 const kRes = Symbol("res");
 
+const tlsStreamRids = new Uint32Array(2);
+
 let debug = debuglog("tls", (fn) => {
   debug = fn;
 });
@@ -160,10 +164,17 @@ export class TLSSocket extends net.Socket {
 
     /** Wraps the given socket and adds the tls capability to the underlying
      * handle */
-    function _wrapHandle(tlsOptions, wrap) {
+    function _wrapHandle(tlsOptions, socket) {
       let handle;
+      let wrap;
+
+      if (socket) {
+        if (socket instanceof net.Socket && socket._handle) {
+          wrap = socket;
+        } else {
+          wrap = new JSStreamSocket(socket);
+        }
 
-      if (wrap) {
         handle = wrap._handle;
       }
 
@@ -188,13 +199,14 @@ export class TLSSocket extends net.Socket {
         }
 
         try {
-          const conn = await startTlsInternal(
-            handle[kStreamBaseField],
+          const conn = await startTls(
+            wrap,
+            handle,
             options,
           );
           try {
             const hs = await conn.handshake();
-            if (hs.alpnProtocol) {
+            if (hs?.alpnProtocol) {
               tlssock.alpnProtocol = hs.alpnProtocol;
             } else {
               tlssock.alpnProtocol = false;
@@ -228,7 +240,7 @@ export class TLSSocket extends net.Socket {
       // An example usage of `_parentWrap` in npm module:
       // https://github.com/szmarczak/http2-wrapper/blob/51eeaf59ff9344fb192b092241bfda8506983620/source/utils/js-stream-socket.js#L6
       handle._parent = handle;
-      handle._parentWrap = wrap;
+      handle._parentWrap = socket;
 
       return handle;
     }
@@ -267,10 +279,75 @@ export class TLSSocket extends net.Socket {
     // TODO(kt3k): implement this
   }
 
+  setMaxSendFragment(_maxSendFragment) {
+    // TODO(littledivy): implement this
+  }
+
   getPeerCertificate(detailed = false) {
     const conn = this[kHandle]?.[kStreamBaseField];
     if (conn) return conn[internals.getPeerCertificate](detailed);
   }
+
+  getCipher() {
+    return "";
+  }
+}
+
+class JSStreamSocket {
+  #rid;
+
+  constructor(stream) {
+    this.stream = stream;
+  }
+
+  init(options) {
+    op_node_tls_start(options, tlsStreamRids);
+    this.#rid = tlsStreamRids[0];
+    const channelRid = tlsStreamRids[1];
+
+    this.stream.on("data", (data) => {
+      core.write(channelRid, data);
+    });
+
+    const buf = new Uint8Array(1024 * 16);
+    (async () => {
+      while (true) {
+        try {
+          const nread = await core.read(channelRid, buf);
+          this.stream.write(buf.slice(0, nread));
+        } catch {
+          break;
+        }
+      }
+    })();
+
+    this.stream.on("close", () => {
+      core.close(this.#rid);
+      core.close(channelRid);
+    });
+  }
+
+  handshake() {
+    return op_node_tls_handshake(this.#rid);
+  }
+
+  read(buf) {
+    return core.read(this.#rid, buf);
+  }
+
+  write(data) {
+    return core.write(this.#rid, data);
+  }
+}
+
+function startTls(wrap, handle, options) {
+  if (wrap instanceof JSStreamSocket) {
+    options.caCerts ??= [];
+    wrap.init(options);
+    return wrap;
+  } else {
+    return startTlsInternal(handle[kStreamBaseField], options);
+  }
 }
 
 function normalizeConnectArgs(listArgs) {