Function handleConnect

packages/computerd/src/cli/computerd.ts:335–404

(
  request: IncomingMessage,
  response: ServerResponse,
  rpc: ReturnType<typeof createWorkspaceServer>,
  upstreamSlot: { ws: WebSocket | undefined },
)

}

async function handleConnect(
  request: IncomingMessage,
  response: ServerResponse,
  rpc: ReturnType<typeof createWorkspaceServer>,
  upstreamSlot: { ws: WebSocket | undefined },
): Promise<void> {
  let body: ConnectBody;
  try {
    body = await readJson<ConnectBody>(request);
  } catch (error) {
    send(response, 400, `invalid JSON body: ${(error as Error).message}\n`, {
      "content-type": "text/plain; charset=utf-8",
    });
    return;
  }

  if (typeof body.url !== "string" || body.url.length === 0) {
    send(response, 400, "missing 'url' in body\n", {
      "content-type": "text/plain; charset=utf-8",
    });
    return;
  }
  const baseUrl = body.url.replace(/\/+$/, "");
  const healthTimeoutMs =
    typeof body.healthTimeoutMs === "number" && body.healthTimeoutMs > 0
      ? body.healthTimeoutMs
      : 30_000;

  try {
    await waitForHealth(baseUrl, healthTimeoutMs);
  } catch (error) {
    send(response, 502, `upstream /health unreachable: ${(error as Error).message}\n`, {
      "content-type": "text/plain; charset=utf-8",
    });
    return;
  }

  // Close any prior outbound session before opening a new one. A DO
  // restart / hibernate hands the new incarnation a fresh /connect;
  // without this, the previous WebSocket leaks for the life of the
  // container.
  const previous = upstreamSlot.ws;
  if (previous !== undefined) {
    upstreamSlot.ws = undefined;
    try {
      previous.close(1000, "replaced by new /connect");
    } catch {
      // already closed; idempotent
    }
  }

  const wsUrl = `${toWebSocketUrl(baseUrl)}/ws`;
  const ws = new WebSocket(wsUrl);
  upstreamSlot.ws = ws;
  ws.once("open", () => {
    console.log(`/connect: attached RPC session to ${wsUrl}`);
    acceptWebSocketSession(ws, rpc);
  });