wss dial: connectToRemote over ws/wss URLs (mesh.exadev.io)

src/core/connection-approval.ts

@@ -98,7 +98,7 @@ export class ConnectionApproval {
     }));
   }
 
-  /** Initiate an outbound connection to a remote coordinator requiring approval. Fires the connect_request and returns immediately. The connection completes asynchronously when the coordinator accepts or rejects. */
+  /** Initiate an outbound connection to a remote coordinator requiring approval. Fires the connect_request and returns immediately. The connection completes asynchronously when the coordinator accepts or rejects. `host` is either a hostname (TLS-dialled with `port`) or a ws:// / wss:// URL (WebSocket-dialled to a hub such as mesh.exadev.io; the port is ignored). */
   async connectToRemote(host: string, port: number): Promise<void> {
     const peerId = this.deps.getPeerId();
     const agent = this.deps.agents.get(peerId);

src/core/tool.ts

@@ -603,7 +603,10 @@ export class CommsTool {
     const connectToRemote = this.store.connectToRemote.bind(this.store);
     return tryMeshAction("connect", async () => {
       await connectToRemote(action.host, action.port);
-      return `Connection request sent to ${action.host}:${String(action.port)}.`;
+      const target = /^wss?:\/\//.test(action.host)
+        ? action.host
+        : `${action.host}:${String(action.port)}`;
+      return `Connection request sent to ${target}.`;
     });
   }
 

src/core/wire-mesh-transport.ts

@@ -9,6 +9,7 @@
  */
 
 import { createTlsTransport } from "wire-mesh-core/adapters/tls-transport";
+import { connectWsUrl } from "./ws-dial.js";
 import {
   acceptMeshSession,
   type AcceptedMeshSession,
@@ -797,9 +798,20 @@ export class WireMeshTransport implements MeshTransport {
     name: string,
     fingerprint: string,
   ): Promise<void> {
-    const connection = await this.wireTransport.connect(
-      `${host}:${String(port)}`,
-    );
+    // A ws:// or wss:// URL in the host position dials a WebSocket-served
+    // hub (e.g. the mesh.exadev.io cloudflare-hub) instead of raw TLS --
+    // the port is meaningless in URL form, so callers pass 0. Any other URL
+    // scheme is refused here rather than surfacing as an opaque DNS error
+    // from the TLS dial treating the whole URL as a hostname.
+    const isWsUrl = /^wss?:\/\//.test(host);
+    if (!isWsUrl && host.includes("://")) {
+      throw new Error(
+        `expected a hostname or a ws:// / wss:// URL, got "${host}"`,
+      );
+    }
+    const connection = isWsUrl
+      ? await connectWsUrl(host)
+      : await this.wireTransport.connect(`${host}:${String(port)}`);
     const identity = await this.identityReady;
     const session = await acceptMeshSession(connection, identity, [DOMAIN], {
       onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),

src/core/ws-dial.ts

@@ -0,0 +1,139 @@
+// A ws-library dial adapter for outbound remote connections to a WebSocket-served hub (e.g. the mesh.exadev.io cloudflare-hub, which answers wss:// only): one CBOR frame per binary WebSocket message, the identical framing wire-mesh-node's own WebSocket transport uses, so a hub sees the same bytes from this client as from any other. Agent-comms' own edge adapter rather than a wire-mesh-node dependency because wire-mesh-node is not published to npm as a consumable package (a placeholder 0.0.0), while the dial is genuinely this app's concern at its own edge -- the same layering its TLS transport adapter already occupies.
+
+import { WebSocket as WsSocket } from "ws";
+import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2";
+import { frameSchema, type Frame } from "wire-mesh-core/generated/protocol";
+import type { Connection } from "wire-mesh-core/ports/transport";
+
+const CONNECT_TIMEOUT_MS = 10_000;
+// RFC 6455 close codes, named rather than bare: 1000 normal closure, 1002 protocol error.
+const CLOSE_NORMAL = 1000;
+const CLOSE_PROTOCOL_ERROR = 1002;
+
+/** Rejects addresses that are neither ws:// nor wss:// URLs -- a wrong-scheme dial would otherwise surface as an opaque socket error. */
+function assertWsUrl(url: string): void {
+  if (!/^wss?:\/\/./.test(url)) {
+    throw new Error(`expected a ws:// or wss:// URL, got "${url}"`);
+  }
+}
+
+export async function connectWsUrl(url: string): Promise<Connection> {
+  assertWsUrl(url);
+  return new Promise<Connection>((resolve, reject) => {
+    const socket = new WsSocket(url);
+    socket.binaryType = "arraybuffer";
+    const timer = setTimeout(() => {
+      socket.terminate();
+      reject(
+        new Error(
+          `ws dial timed out after ${String(CONNECT_TIMEOUT_MS)}ms: ${url}`,
+        ),
+      );
+    }, CONNECT_TIMEOUT_MS);
+    socket.once("open", () => {
+      clearTimeout(timer);
+      resolve(wrapSocket(socket));
+    });
+    socket.once("error", (error) => {
+      clearTimeout(timer);
+      reject(error instanceof Error ? error : new Error(String(error)));
+    });
+  });
+}
+
+function wrapSocket(socket: WsSocket): Connection {
+  const pending: Frame[] = [];
+  const waiters: {
+    resolve: (result: IteratorResult<Frame>) => void;
+    reject: (error: unknown) => void;
+  }[] = [];
+  let ended = false;
+  let failure: Error | null = null;
+
+  function endAll(): void {
+    ended = true;
+    for (const waiter of waiters.splice(0)) {
+      waiter.resolve({ value: undefined, done: true });
+    }
+  }
+
+  function failAll(error: Error): void {
+    failure = error;
+    ended = true;
+    for (const waiter of waiters.splice(0)) {
+      waiter.reject(error);
+    }
+  }
+
+  socket.on("message", (data) => {
+    if (!(data instanceof ArrayBuffer)) {
+      failAll(new Error("expected a binary WebSocket message"));
+      socket.close(CLOSE_PROTOCOL_ERROR, "protocol error");
+      return;
+    }
+    let frame: Frame;
+    try {
+      const decoded: unknown = decode(new Uint8Array(data), cdeDecodeOptions);
+      const parsed = frameSchema.safeParse(decoded);
+      if (!parsed.success) {
+        // A decodable-but-unrecognised frame is dropped, keeping the connection -- version negotiation exists to tolerate it.
+        return;
+      }
+      frame = parsed.data;
+    } catch (error) {
+      failAll(
+        error instanceof Error
+          ? error
+          : new Error(`frame body failed to decode: ${String(error)}`),
+      );
+      socket.close(CLOSE_PROTOCOL_ERROR, "protocol error");
+      return;
+    }
+    const waiter = waiters.shift();
+    if (waiter) {
+      waiter.resolve({ value: frame, done: false });
+    } else {
+      pending.push(frame);
+    }
+  });
+  socket.on("close", endAll);
+  socket.on("error", () => {
+    // The receive stream's failure signal; the raw error is not otherwise actionable here.
+  });
+
+  const receiveStream: AsyncIterable<Frame> = {
+    [Symbol.asyncIterator]() {
+      return {
+        async next(): Promise<IteratorResult<Frame>> {
+          const next = pending.shift();
+          if (next !== undefined) {
+            return { value: next, done: false };
+          }
+          if (failure !== null) {
+            throw failure;
+          }
+          if (ended) {
+            return { value: undefined, done: true };
+          }
+          return new Promise<IteratorResult<Frame>>((resolve, reject) => {
+            waiters.push({ resolve, reject });
+          });
+        },
+      };
+    },
+  };
+
+  return {
+    async send(frame: Frame): Promise<void> {
+      if (ended) {
+        throw new Error("connection is closed");
+      }
+      socket.send(new Uint8Array(encode(frame, cdeEncodeOptions)));
+    },
+    receive: () => receiveStream,
+    close: async () => {
+      socket.close(CLOSE_NORMAL);
+      return Promise.resolve();
+    },
+  };
+}

src/test/ws-dial.test.ts

@@ -0,0 +1,107 @@
+import { createServer } from "node:http";
+import { WebSocketServer, type WebSocket as WsSocket } from "ws";
+import { afterEach, describe, expect, it } from "vitest";
+import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2";
+import type { Frame } from "wire-mesh-core/generated/protocol";
+import { connectWsUrl } from "../core/ws-dial.js";
+
+const SHUTDOWN_GRACE_MS = 250;
+
+interface TestServer {
+  url: string;
+  close: () => Promise<void>;
+  received: Frame[];
+}
+
+/** A real local ws server echoing every received frame back, so the dial round trip is exercised against a genuine socket. */
+async function echoServer(): Promise<TestServer> {
+  const received: Frame[] = [];
+  const http = createServer();
+  const wss = new WebSocketServer({ server: http });
+  wss.on("connection", (socket: WsSocket) => {
+    socket.on("message", (data) => {
+      const decodedFrame: unknown = decode(
+        new Uint8Array(data as ArrayBuffer),
+        cdeDecodeOptions,
+      );
+      const frame = decodedFrame as Frame;
+      received.push(frame);
+      socket.send(new Uint8Array(encode(frame, cdeEncodeOptions)));
+    });
+  });
+  await new Promise<void>((resolve) => {
+    http.listen(0, "127.0.0.1", () => {
+      resolve();
+    });
+  });
+  const address = http.address();
+  if (address === null || typeof address === "string") {
+    throw new Error("expected a TCP listen address");
+  }
+  return {
+    url: `ws://127.0.0.1:${String(address.port)}/`,
+    received,
+    close: async () =>
+      new Promise<void>((resolve) => {
+        // wss.close() alone waits for connected clients, which would
+        // deadlock a test asserting behaviour AFTER the server side hangs
+        // up -- terminate them first, then close the listener.
+        for (const client of wss.clients) {
+          client.terminate();
+        }
+        wss.close();
+        http.close(() => {
+          resolve();
+        });
+        setTimeout(resolve, SHUTDOWN_GRACE_MS);
+      }),
+  };
+}
+
+const servers: TestServer[] = [];
+
+afterEach(async () => {
+  for (const server of servers.splice(0)) {
+    await server.close();
+  }
+});
+
+describe("connectWsUrl", () => {
+  it("round-trips a frame against a real ws server", async () => {
+    const server = await echoServer();
+    servers.push(server);
+    const connection = await connectWsUrl(server.url);
+    const ping: Frame = { type: "ping" };
+    await connection.send(ping);
+
+    const iterator = connection.receive()[Symbol.asyncIterator]();
+    const first = await iterator.next();
+    expect(first.done).toBe(false);
+    expect(first.value).toEqual(ping);
+    expect(server.received).toEqual([ping]);
+
+    await connection.close();
+  });
+
+  it("rejects a non-ws/wss URL outright, without a socket attempt", async () => {
+    await expect(connectWsUrl("ftp://example.com/")).rejects.toThrow(
+      /ws:\/\/ or wss:\/\//,
+    );
+    await expect(connectWsUrl("example.com:1234")).rejects.toThrow(
+      /ws:\/\/ or wss:\/\//,
+    );
+  });
+
+  it("ends its receive stream when the server closes the socket", async () => {
+    const server = await echoServer();
+    const connection = await connectWsUrl(server.url);
+    await connection.send({ type: "ping" });
+    const iterator = connection.receive()[Symbol.asyncIterator]();
+    await iterator.next(); // the echo
+    // Close the server first: wss.close() itself waits for client sockets,
+    // so closing the connection first would deadlock the shutdown.
+    await server.close();
+    const afterClose = await iterator.next();
+    expect(afterClose.done).toBe(true);
+  });
+});