import { isIP } from "node:net"; import { TextDecoder } from "node:util"; import WebSocket, { RawData } from "ws"; import { ActorSnapshot, decodeNotification, decodeResponse, isActorRef, MAX_MESSAGE_BYTES, PROTOCOL_VERSION, ProtocolError, ReadCommand, ReadReply, WEBSOCKET_SUBPROTOCOL, } from "./protocol"; export function parseEndpoint(endpoint: string): { host: string; port: number } { const match = /^(?:\[([^\]]+)\]|([^:]+)):(\d+)$/.exec(endpoint); if (!match) throw new Error("Use a numeric loopback address, e.g. 127.0.0.1:3737 or [::1]:3737"); const host = match[1] ?? match[2]; if ( host === undefined || !((isIP(host) === 4 && host.startsWith("127.")) || (isIP(host) === 6 && host === "::1")) ) { throw new Error("Only numeric IPv4 loopback or [::1] endpoints are supported"); } const port = Number(match[3]); if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error("Invalid WebSocket port"); return { host, port }; } interface Pending { requestId: string; command: ReadCommand; resolve: (reply: ReadReply) => void; reject: (error: Error) => void; timer: NodeJS.Timeout; } export class ImageClient { private pending?: Pending; private queue: Promise = Promise.resolve(); private sequence = 0; private revision = 0n; private closed = false; private ready = false; private constructor( private readonly socket: WebSocket, private readonly onDisconnect: (error?: Error) => void, private readonly onImageChanged: (revision: bigint) => void, private readonly requestTimeout: number, ) { socket.on("message", (data, binary) => this.receive(data, binary)); socket.on("error", (error) => this.close(error)); socket.on("close", () => this.close(new Error("Image WebSocket closed"))); } static async connect( endpoint: string, onDisconnect: (error?: Error) => void, onImageChanged: (revision: bigint) => void = () => {}, timeouts = { connect: 5000, request: 30000 }, ): Promise { parseEndpoint(endpoint); const socket = new WebSocket(`ws://${endpoint}/`, WEBSOCKET_SUBPROTOCOL, { handshakeTimeout: timeouts.connect, maxPayload: MAX_MESSAGE_BYTES, perMessageDeflate: false, followRedirects: false, }); const client = new ImageClient(socket, onDisconnect, onImageChanged, timeouts.request); try { await new Promise((resolve, reject) => { const fail = (error: Error) => { socket.off("open", opened); reject(error); }; const opened = () => { socket.off("error", fail); if (socket.protocol !== WEBSOCKET_SUBPROTOCOL) { reject(new Error("Server did not negotiate runa-debug")); } else { client.ready = true; resolve(); } }; socket.once("error", fail); socket.once("open", opened); }); } catch (error) { client.close(); throw error; } return client; } async listActors(): Promise { const reply = await this.request({ operation: "list_actors" }); if (reply.kind !== "actors") throw new Error("Expected actor listing"); return reply.data; } async inspectActor(actorRef: string): Promise { if (!isActorRef(actorRef)) throw new Error("Invalid actor reference"); const reply = await this.request({ operation: "inspect_actor", parameters: { actor_ref: actorRef }, }); if (reply.kind !== "actor") throw new Error("Expected actor snapshot"); return reply.data; } private request(command: ReadCommand): Promise { const request = this.queue.then( () => new Promise((resolve, reject) => { if (this.closed) { reject(new Error("Not connected to an image")); return; } const requestId = `runa-vscode-${++this.sequence}`; const message = JSON.stringify({ version: PROTOCOL_VERSION, request_id: requestId, command, }); const timer = setTimeout( () => this.close(new Error("Image request timed out; connection closed without retry")), this.requestTimeout, ); this.pending = { requestId, command, resolve, reject, timer }; this.socket.send(message, (error) => { if (error) this.close(error); }); }), ); // Each caller retains its own rejection, without poisoning subsequent requests. this.queue = request.then( () => undefined, () => undefined, ); return request; } private receive(data: RawData, binary: boolean): void { if (this.closed) return; try { if (binary) throw new Error("Only WebSocket text protocol messages are accepted"); const bytes = Array.isArray(data) ? Buffer.concat(data) : data instanceof ArrayBuffer ? Buffer.from(data) : data; if (bytes.length > MAX_MESSAGE_BYTES) throw new Error("Image response exceeds 1 MiB"); const text = new TextDecoder("utf-8", { fatal: true }).decode(bytes); const revision = decodeNotification(text); if (revision !== undefined) { if (revision <= this.revision) throw new Error("Non-increasing image notification revision"); this.revision = revision; this.onImageChanged(revision); return; } const pending = this.pending; if (!pending) throw new Error("Unexpected unsolicited image response"); try { pending.resolve(decodeResponse(text, pending.requestId, pending.command)); } catch (error) { if (!(error instanceof ProtocolError)) throw error; pending.reject(error); } this.pending = undefined; clearTimeout(pending.timer); } catch (error) { this.close(error instanceof Error ? error : new Error(String(error))); } } close(error?: Error): void { if (this.closed) return; this.closed = true; this.socket.terminate(); if (this.pending) { clearTimeout(this.pending.timer); this.pending.reject(error ?? new Error("Disconnected from image")); this.pending = undefined; } if (this.ready) this.onDisconnect(error); } }