ops(relay): bridge production traffic to Fly

packages/relay/src/cloudflare-adapter.ts

@@ -17,6 +17,7 @@
  * ```
  */
 
+import { createCutoverProxy } from "./cutover-proxy.js";
 import type { ConnectionRole, RelaySessionAttachment } from "./types.js";
 
 type RelayProtocolVersion = "1" | "2";
@@ -97,6 +98,7 @@ function getGlobalWebSocketPair(): (new () => WebSocketPair) | undefined {
 
 interface Env {
   RELAY: DurableObjectNamespace;
+  PASEO_RELAY_UPSTREAM?: string;
 }
 
 interface DurableObjectNamespace {
@@ -572,6 +574,10 @@ export class RelayDurableObject {
  */
 export default {
   async fetch(request: Request, env: Env): Promise<Response> {
+    if (env.PASEO_RELAY_UPSTREAM) {
+      return createCutoverProxy(env.PASEO_RELAY_UPSTREAM).fetch(request);
+    }
+
     const url = new URL(request.url);
 
     // Health check

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

@@ -0,0 +1,88 @@
+import { createServer, type IncomingMessage, type Server } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterEach, describe, expect, it } from "vitest";
+
+import relayWorker from "./cloudflare-adapter.js";
+
+type RelayEnv = Parameters<typeof relayWorker.fetch>[1];
+
+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 response = await relayWorker.fetch(
+      new Request("https://relay.paseo.sh/ws?serverId=srv_prod&role=server&v=2", {
+        headers: { "x-relay-probe": "production" },
+      }),
+      { PASEO_RELAY_UPSTREAM: originUrl } as RelayEnv,
+    );
+
+    expect(origin.received()).toEqual([
+      {
+        method: "GET",
+        path: "/ws?serverId=srv_prod&role=server&v=2",
+        probe: "production",
+      },
+    ]);
+    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,17 @@
+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));
+    },
+  };
+}

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.toml

@@ -8,6 +8,9 @@ routes = [{ pattern = "relay.paseo.sh", custom_domain = true }]
 [observability]
 enabled = true
 
+[vars]
+PASEO_RELAY_UPSTREAM = "https://paseo-relay-next.fly.dev"
+
 [[durable_objects.bindings]]
 name = "RELAY"
 class_name = "RelayDurableObject"