feat(desktop,host-service): forward remote workspace ports to the local machine

apps/desktop/src/lib/trpc/routers/index.ts

@@ -20,6 +20,7 @@ import { createNotificationsRouter } from "./notifications";
 import { createPageContentRouter } from "./page-content";
 import { createPermissionsRouter } from "./permissions";
 import { createPluginsRouter } from "./plugins";
+import { createPortForwardsRouter } from "./port-forwards";
 import { createPortsRouter } from "./ports";
 import { createProjectsRouter } from "./projects";
 import { createResourceMetricsRouter } from "./resource-metrics";
@@ -50,6 +51,7 @@ export const createAppRouter = (getWindow: () => BrowserWindow | null) => {
 		permissions: createPermissionsRouter(),
 		plugins: createPluginsRouter(),
 		ports: createPortsRouter(),
+		portForwards: createPortForwardsRouter(),
 		resourceMetrics: createResourceMetricsRouter(),
 		menu: createMenuRouter(),
 		external: createExternalRouter(),

apps/desktop/src/lib/trpc/routers/port-forwards/index.ts

@@ -0,0 +1 @@
+export { createPortForwardsRouter } from "./port-forwards";

apps/desktop/src/lib/trpc/routers/port-forwards/port-forwards.ts

@@ -0,0 +1,48 @@
+import { observable } from "@trpc/server/observable";
+import { portForwardManager, setRelayToken } from "main/lib/port-forward";
+import type { PortForward } from "shared/types";
+import { z } from "zod";
+import { publicProcedure, router } from "../..";
+
+export const createPortForwardsRouter = () => {
+	return router({
+		setRelayToken: publicProcedure
+			.input(z.object({ token: z.string().nullable() }))
+			.mutation(({ input }) => {
+				setRelayToken(input.token);
+			}),
+
+		list: publicProcedure.query((): PortForward[] => portForwardManager.list()),
+
+		subscribe: publicProcedure.subscription(() => {
+			return observable<PortForward[]>((emit) => {
+				const onChange = (forwards: PortForward[]) => emit.next(forwards);
+				portForwardManager.on("change", onChange);
+				emit.next(portForwardManager.list());
+				return () => {
+					portForwardManager.off("change", onChange);
+				};
+			});
+		}),
+
+		sync: publicProcedure
+			.input(
+				z.object({
+					hostUrl: z.string(),
+					workspaceId: z.string(),
+					ports: z.array(z.number().int().positive()),
+				}),
+			)
+			.mutation(
+				({ input }): Promise<PortForward[]> => portForwardManager.sync(input),
+			),
+
+		retryEphemeral: publicProcedure
+			.input(z.object({ id: z.string() }))
+			.mutation(({ input }) => portForwardManager.retryEphemeral(input.id)),
+
+		killLocalOwner: publicProcedure
+			.input(z.object({ id: z.string() }))
+			.mutation(({ input }) => portForwardManager.killLocalOwner(input.id)),
+	});
+};

apps/desktop/src/main/index.ts

@@ -38,6 +38,7 @@ import {
 	shutdownTanstackDbPersistence,
 } from "./lib/persistence/persistence";
 import { syncInstalledPluginMcpServers } from "./lib/plugin-installs";
+import { portForwardManager } from "./lib/port-forward";
 import { ensureProjectIconsDir, getProjectIconPath } from "./lib/project-icons";
 import { runQuitCleanup } from "./lib/quit-sequence";
 import { initSentry } from "./lib/sentry";
@@ -242,6 +243,9 @@ app.on("before-quit", async (event) => {
 	}
 
 	isQuitting = true;
+	// Local port-forward listeners hold no state worth draining; drop them so
+	// nothing keeps 127.0.0.1:<port> bound after the app is gone.
+	portForwardManager.stopAll();
 	// Snapshot all open windows (bounds + org) before they close, so relaunch
 	// restores them. markAppQuitting() stops per-window close handlers from
 	// shrinking the set as windows close one-by-one.

apps/desktop/src/main/lib/host-service-utils.ts

@@ -47,7 +47,7 @@ function normalizePort(port: number): number | null {
 	return port;
 }
 
-function canBindPort(port: number): Promise<boolean> {
+export function canBindPort(port: number): Promise<boolean> {
 	return new Promise((resolve) => {
 		const server = createServer();
 		const finish = (available: boolean) => {

apps/desktop/src/main/lib/port-forward/index.ts

@@ -0,0 +1,22 @@
+import { canBindPort } from "../host-service-utils";
+import { portManager } from "../terminal/port-manager";
+import { PortForwardManager } from "./port-forward-manager";
+import { RelayForwardTransport } from "./relay-forward-transport";
+
+export { PortForwardManager } from "./port-forward-manager";
+export { RelayForwardTransport } from "./relay-forward-transport";
+export type { ForwardTransport } from "./types";
+
+let relayToken: string | null = null;
+
+export function setRelayToken(token: string | null): void {
+	relayToken = token;
+}
+
+export const portForwardManager = new PortForwardManager({
+	transport: new RelayForwardTransport({ getToken: () => relayToken }),
+	getLocalPorts: () => portManager.getAllPorts(),
+	killLocalPort: ({ terminalId, workspaceId, port }) =>
+		portManager.killPort({ terminalId, workspaceId, port }),
+	canBindPort,
+});

apps/desktop/src/main/lib/port-forward/port-forward-manager.test.ts

@@ -0,0 +1,233 @@
+import { describe, expect, test } from "bun:test";
+import net from "node:net";
+import { type Duplex, PassThrough } from "node:stream";
+import type { DetectedPort } from "@superset/port-scanner";
+import type { ForwardTarget } from "shared/types";
+import { PortForwardManager } from "./port-forward-manager";
+import type { ForwardTransport } from "./types";
+
+const HOST = "https://relay.test/hosts/org:machine";
+
+async function startEcho(): Promise<{ port: number; close: () => void }> {
+	const server = net.createServer((s) => s.pipe(s));
+	const port = await new Promise<number>((resolve) =>
+		server.listen(0, "127.0.0.1", () =>
+			resolve((server.address() as net.AddressInfo).port),
+		),
+	);
+	return { port, close: () => server.close() };
+}
+
+async function freePort(): Promise<number> {
+	const server = net.createServer();
+	return new Promise((resolve) =>
+		server.listen(0, "127.0.0.1", () => {
+			const { port } = server.address() as net.AddressInfo;
+			server.close(() => resolve(port));
+		}),
+	);
+}
+
+/** A transport that opens a TCP connection to an echo server on this machine. */
+function echoTransport(echoPort: number): ForwardTransport & {
+	opened: ForwardTarget[];
+} {
+	const opened: ForwardTarget[] = [];
+	return {
+		kind: "relay",
+		opened,
+		probe: async () => {},
+		openStream: async (target) => {
+			opened.push(target);
+			const socket = net.connect({ host: "127.0.0.1", port: echoPort });
+			await new Promise<void>((r) => socket.once("connect", () => r()));
+			return socket as Duplex;
+		},
+	};
+}
+
+function manager(
+	transport: ForwardTransport,
+	overrides: Partial<ConstructorParameters<typeof PortForwardManager>[0]> = {},
+) {
+	return new PortForwardManager({
+		transport,
+		getLocalPorts: () => [],
+		killLocalPort: async () => ({ success: true }),
+		canBindPort: async () => true,
+		...overrides,
+	});
+}
+
+async function roundTrip(port: number, payload: string): Promise<string> {
+	const socket = net.connect({ host: "127.0.0.1", port });
+	await new Promise<void>((r) => socket.once("connect", () => r()));
+	socket.write(payload);
+	const data = await new Promise<Buffer>((r) => socket.once("data", r));
+	socket.destroy();
+	return data.toString();
+}
+
+describe("PortForwardManager", () => {
+	test("sync listens on the requested port and bridges bytes", async () => {
+		const echo = await startEcho();
+		const transport = echoTransport(echo.port);
+		const m = manager(transport);
+		const local = await freePort();
+		const [fwd] = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			ports: [local],
+		});
+		expect(fwd?.status).toEqual({ state: "active", localPort: local });
+		expect(await roundTrip(local, "ping")).toBe("ping");
+		expect(transport.opened[0]).toEqual({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			remotePort: local,
+		});
+		m.stopAll();
+		echo.close();
+	});
+
+	test("sync for another workspace stops the previous forwards", async () => {
+		const echo = await startEcho();
+		const m = manager(echoTransport(echo.port));
+		const a = await freePort();
+		await m.sync({ hostUrl: HOST, workspaceId: "ws1", ports: [a] });
+		const b = await freePort();
+		const list = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws2",
+			ports: [b],
+		});
+		expect(list.map((f) => f.target.workspaceId)).toEqual(["ws2"]);
+		expect(await freeToBind(a)).toBe(true);
+		m.stopAll();
+		echo.close();
+	});
+
+	test("a busy local port reports busy with the owner when known", async () => {
+		const echo = await startEcho();
+		const owner: DetectedPort = {
+			port: echo.port,
+			pid: 42,
+			processName: "node",
+			terminalId: "t1",
+			workspaceId: "local-ws",
+			detectedAt: 0,
+			address: "127.0.0.1",
+		};
+		const m = manager(echoTransport(echo.port), {
+			getLocalPorts: () => [owner],
+		});
+		const [fwd] = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			ports: [echo.port],
+		});
+		expect(fwd?.status).toEqual({
+			state: "busy",
+			localPort: echo.port,
+			localOwner: {
+				pid: 42,
+				processName: "node",
+				terminalId: "t1",
+				workspaceId: "local-ws",
+			},
+		});
+		m.stopAll();
+		echo.close();
+	});
+
+	test("retryEphemeral moves a busy forward to another local port", async () => {
+		const echo = await startEcho();
+		const m = manager(echoTransport(echo.port));
+		const [fwd] = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			ports: [echo.port],
+		});
+		expect(fwd?.status.state).toBe("busy");
+		const retried = await m.retryEphemeral(fwd?.id ?? "");
+		expect(retried?.status.state).toBe("active");
+		if (retried?.status.state !== "active") throw new Error("not active");
+		expect(retried.status.localPort).not.toBe(echo.port);
+		expect(await roundTrip(retried.status.localPort, "x")).toBe("x");
+		m.stopAll();
+		echo.close();
+	});
+
+	test("killLocalOwner restarts the forward once the port frees", async () => {
+		const echo = await startEcho();
+		let killed = false;
+		const m = manager(echoTransport(echo.port), {
+			getLocalPorts: () => [
+				{
+					port: echo.port,
+					pid: 1,
+					processName: "node",
+					terminalId: "t1",
+					workspaceId: "local-ws",
+					detectedAt: 0,
+					address: "127.0.0.1",
+				},
+			],
+			killLocalPort: async () => {
+				killed = true;
+				echo.close();
+				return { success: true };
+			},
+			canBindPort: freeToBind,
+		});
+		const [fwd] = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			ports: [echo.port],
+		});
+		const result = await m.killLocalOwner(fwd?.id ?? "");
+		expect(killed).toBe(true);
+		expect(result).toEqual({ success: true });
+		expect(m.list()[0]?.status).toEqual({
+			state: "active",
+			localPort: echo.port,
+		});
+		m.stopAll();
+	});
+
+	test("a failed probe yields an error status", async () => {
+		const m = manager({
+			kind: "relay",
+			probe: async () => {
+				throw new Error("protocol v1");
+			},
+			openStream: async () => new PassThrough(),
+		});
+		const [fwd] = await m.sync({
+			hostUrl: HOST,
+			workspaceId: "ws1",
+			ports: [await freePort()],
+		});
+		expect(fwd?.status).toEqual({ state: "error", message: "protocol v1" });
+		m.stopAll();
+	});
+
+	test("stopAll releases every listener", async () => {
+		const echo = await startEcho();
+		const m = manager(echoTransport(echo.port));
+		const local = await freePort();
+		await m.sync({ hostUrl: HOST, workspaceId: "ws1", ports: [local] });
+		m.stopAll();
+		expect(m.list()).toEqual([]);
+		expect(await freeToBind(local)).toBe(true);
+		echo.close();
+	});
+});
+
+function freeToBind(port: number): Promise<boolean> {
+	return new Promise((resolve) => {
+		const server = net.createServer();
+		server.once("error", () => resolve(false));
+		server.listen(port, "127.0.0.1", () => server.close(() => resolve(true)));
+	});
+}