test(relay): observe control events before socket open

.github/workflows/deploy-relay.yml

@@ -27,6 +27,6 @@ jobs:
         run: npm run typecheck --workspace=@getpaseo/relay
 
       - name: Deploy worker
-        run: cd packages/relay && npx wrangler deploy
+        run: cd packages/relay && npx wrangler deploy -c wrangler.staging.toml
         env:
           CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }}

packages/relay/src/cutover-proxy.test.ts

@@ -0,0 +1,86 @@
+import { createServer, type IncomingMessage, type Server } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterEach, describe, expect, it } from "vitest";
+
+import { createCutoverProxy } from "./cutover-proxy.js";
+
+interface ReceivedRequest {
+  method: string;
+  path: string;
+  probe: string;
+}
+
+class RecordingOrigin {
+  private readonly requests: ReceivedRequest[] = [];
+  private server: Server | null = null;
+
+  async start(): Promise<string> {
+    this.server = createServer((request, response) => {
+      this.requests.push(recordRequest(request));
+      response.writeHead(202, { "content-type": "application/json", "x-relay-origin": "fly" });
+      response.end(JSON.stringify({ status: "forwarded" }));
+    });
+    await new Promise<void>((resolve) => this.server!.listen(0, "127.0.0.1", resolve));
+    const address = this.server.address() as AddressInfo;
+    return `http://127.0.0.1:${address.port}`;
+  }
+
+  received(): ReceivedRequest[] {
+    return this.requests;
+  }
+
+  async close(): Promise<void> {
+    const server = this.server;
+    this.server = null;
+    if (!server) return;
+    await new Promise<void>((resolve, reject) => {
+      server.close((error) => {
+        if (error) {
+          reject(error);
+          return;
+        }
+        resolve();
+      });
+    });
+  }
+}
+
+function recordRequest(request: IncomingMessage): ReceivedRequest {
+  return {
+    method: request.method ?? "",
+    path: request.url ?? "",
+    probe: String(request.headers["x-relay-probe"] ?? ""),
+  };
+}
+
+describe("cutover proxy", () => {
+  const origins: RecordingOrigin[] = [];
+
+  afterEach(async () => {
+    await Promise.all(origins.splice(0).map((origin) => origin.close()));
+  });
+
+  it("forwards the request to the configured origin without changing its route", async () => {
+    const origin = new RecordingOrigin();
+    origins.push(origin);
+    const originUrl = await origin.start();
+    const proxy = createCutoverProxy(originUrl);
+
+    const response = await proxy.fetch(
+      new Request("https://relay-staging.example/ws?serverId=srv_stage&role=server&v=2", {
+        headers: { "x-relay-probe": "staging" },
+      }),
+    );
+
+    expect(origin.received()).toEqual([
+      {
+        method: "GET",
+        path: "/ws?serverId=srv_stage&role=server&v=2",
+        probe: "staging",
+      },
+    ]);
+    expect(response.status).toBe(202);
+    await expect(response.json()).resolves.toEqual({ status: "forwarded" });
+    expect(response.headers.get("x-relay-origin")).toBe("fly");
+  });
+});

packages/relay/src/cutover-proxy.ts

@@ -0,0 +1,19 @@
+interface CutoverProxy {
+  fetch(request: Request): Promise<Response>;
+}
+
+export function createCutoverProxy(origin: string): CutoverProxy {
+  const originUrl = new URL(origin);
+
+  return {
+    async fetch(request: Request): Promise<Response> {
+      const upstreamUrl = new URL(request.url);
+      upstreamUrl.protocol = originUrl.protocol;
+      upstreamUrl.host = originUrl.host;
+
+      return fetch(new Request(upstreamUrl, request));
+    },
+  };
+}
+
+export default createCutoverProxy("https://paseo-relay-next.fly.dev");

packages/relay/src/live-relay.e2e.test.ts

@@ -12,7 +12,7 @@ import {
 // This live test uses the hosted relay's real TLS endpoint. Self-hosted relay TLS
 // opt-in is covered at URL-building/integration level so the local E2E does not
 // need to provision trusted certificates.
-const RELAY_BASE_URL = "wss://relay.paseo.sh";
+const RELAY_BASE_URL = process.env.PASEO_LIVE_RELAY_URL ?? "wss://relay.paseo.sh";
 
 async function withRetry<T>(
   fn: () => Promise<T>,
@@ -119,6 +119,7 @@ describe("Live relay (relay.paseo.sh) E2E", () => {
         // === Connect ===
         const daemonControlWs = new WebSocket(serverControlUrl);
         const clientWs = new WebSocket(clientUrl);
+        const connected = waitForConnected(daemonControlWs, connectionId);
         let daemonWs: WebSocket | null = null;
 
         try {
@@ -127,7 +128,7 @@ describe("Live relay (relay.paseo.sh) E2E", () => {
             waitOpen(clientWs, "client"),
           ]);
 
-          await waitForConnected(daemonControlWs, connectionId);
+          await connected;
 
           daemonWs = new WebSocket(serverDataUrl);
           await waitOpen(daemonWs, "server-data");

packages/relay/wrangler.staging.toml

@@ -0,0 +1,8 @@
+name = "paseo-relay-cutover-staging"
+main = "src/cutover-proxy.ts"
+compatibility_date = "2024-12-01"
+workers_dev = true
+preview_urls = false
+
+[observability]
+enabled = true