1// WS-over-iroh bridge. One iroh connection carries exactly one WebSocket2// connection (v1: the Jazz upstream link — no multiplexing needed).3//4// Dialer side: a loopback WS listener; JazzServer's upstreamUrl points at it.5// Each accepted WS dials the peer over iroh, sends a HELLO frame naming the6// request path/subprotocol (the tunnel terminates WS on both ends, so upgrade7// parameters don't carry themselves), then bridges frames.8// Acceptor side: an accept loop; each inbound iroh conn reads HELLO, opens a9// WebSocket to the LOCAL JazzServer, and bridges the same way.1011import type { IrohConn, IrohNode } from "./iroh/node.ts";1213const DEBUG = Deno.env.get("LOFI_NODE_DEBUG") === "1";14function debug(...args: unknown[]): void {15 if (DEBUG) console.error("[tunnel]", ...args);16}1718export const FRAME_HELLO = 0;19export const FRAME_TEXT = 1;20export const FRAME_BIN = 2;21export const FRAME_CLOSE = 3;2223const encoder = new TextEncoder();24const decoder = new TextDecoder();2526export function encodeFrame(type: number, payload: Uint8Array): Uint8Array {27 const out = new Uint8Array(1 + payload.length);28 out[0] = type;29 out.set(payload, 1);30 return out;31}3233export function decodeFrame(bytes: Uint8Array): { type: number; payload: Uint8Array } {34 if (bytes.length === 0) return { type: -1, payload: new Uint8Array(0) };35 return { type: bytes[0], payload: bytes.subarray(1) };36}3738export function encodeClose(code: number, reason: string): Uint8Array {39 const reasonBytes = encoder.encode(reason);40 const out = new Uint8Array(2 + reasonBytes.length);41 new DataView(out.buffer).setUint16(0, code, false);42 out.set(reasonBytes, 2);43 return out;44}4546export function decodeClose(payload: Uint8Array): { code: number; reason: string } {47 if (payload.length < 2) return { code: 1000, reason: "" };48 const code = new DataView(payload.buffer, payload.byteOffset).getUint16(0, false);49 return { code, reason: decoder.decode(payload.subarray(2)) };50}5152/** WS close codes 1005/1006/1015 (and out-of-range values) may not be passed53 * to close() — normalize what we relay. */54export function sanitizeCloseCode(code: number): number {55 if (code < 1000 || code > 4999 || code === 1005 || code === 1006 || code === 1015) return 1000;56 return code;57}5859/** kind "ws" (or absent, legacy): one WebSocket per conn. kind "http": one60 * request/response per conn — HELLO{method,path,headers} + BIN body up,61 * HELLO{status,headers} + BIN body back. Jazz leaves proxy their catalogue62 * (schema/permissions) HTTP reads to their upstream, so the tunnel must carry63 * plain HTTP as well as the sync WebSocket. */64interface Hello {65 kind?: "ws" | "http";66 path: string;67 protocol?: string | null;68 method?: string;69 headers?: Record<string, string>;70}7172interface HttpResponseHead {73 status: number;74 headers: Record<string, string>;75}7677const HOP_BY_HOP = new Set([78 "connection",79 "keep-alive",80 "transfer-encoding",81 "upgrade",82 "host",83 "content-length",84]);8586function toArrayBuffer(bytes: Uint8Array): ArrayBuffer {87 return bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) as ArrayBuffer;88}8990export function forwardableHeaders(headers: Headers): Record<string, string> {91 const out: Record<string, string> = {};92 for (const [name, value] of headers) {93 if (!HOP_BY_HOP.has(name.toLowerCase())) out[name] = value;94 }95 return out;96}9798/** Pump both directions between an OPEN WebSocket and an iroh conn until99 * either side closes. `backlog` holds WS data that arrived before the conn was100 * ready (the WS side is push-based; the conn side is pull-based and needs no101 * buffering) — it is flushed before the message listener attaches, and both102 * happen synchronously, so no event can interleave. Resolves on teardown;103 * teardown is idempotent. */104export function bridge(105 conn: IrohConn,106 ws: WebSocket,107 backlog: (string | ArrayBuffer)[] = [],108): Promise<void> {109 ws.binaryType = "arraybuffer";110 let done: () => void;111 const finished = new Promise<void>((res) => (done = res));112 let torndown = false;113114 const teardown = (notifyPeer: boolean, code = 1000, reason = "") => {115 if (torndown) return;116 torndown = true;117 if (notifyPeer) {118 try {119 conn.sendMsg(encodeFrame(FRAME_CLOSE, encodeClose(code, reason)));120 } catch {121 // conn already gone122 }123 }124 conn.close().catch(() => {});125 if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {126 try {127 ws.close(sanitizeCloseCode(code), reason.slice(0, 120));128 } catch {129 // already closing130 }131 }132 done();133 };134135 const forward = (data: string | ArrayBuffer) => {136 try {137 if (typeof data === "string") {138 conn.sendMsg(encodeFrame(FRAME_TEXT, encoder.encode(data)));139 } else {140 conn.sendMsg(encodeFrame(FRAME_BIN, new Uint8Array(data)));141 }142 } catch {143 teardown(false, 1011, "tunnel send failed");144 }145 };146 for (const data of backlog) forward(data);147 backlog.length = 0;148 ws.addEventListener("message", (ev) => forward(ev.data));149 ws.addEventListener("close", (ev) => teardown(true, ev.code || 1000, ev.reason ?? ""));150 ws.addEventListener("error", () => teardown(true, 1011, "ws error"));151152 (async () => {153 while (!torndown) {154 const msg = await conn.recvMsg();155 if (msg === null) {156 teardown(false, 1006, "tunnel closed");157 break;158 }159 const { type, payload } = decodeFrame(msg);160 if (type === FRAME_TEXT) ws.send(decoder.decode(payload));161 else if (type === FRAME_BIN) ws.send(payload);162 else if (type === FRAME_CLOSE) {163 const { code, reason } = decodeClose(payload);164 teardown(false, code, reason);165 break;166 }167 // Unknown frame types are ignored (forward compat).168 }169 })();170171 return finished;172}173174/** Live-connection observability (from upstream ConnectionStats/paths). */175export interface TunnelConnStat {176 direction: "out" | "in";177 rtt: number | null;178 paths: number;179}180181export interface TunnelListener {182 /** Loopback port; use ws://127.0.0.1:<port> as JazzServer upstreamUrl. */183 port: number;184 /** Stats for the currently live outbound tunnel connections. */185 stats(): TunnelConnStat[];186 close(): Promise<void>;187}188189/** Dialer side: serve a loopback WS endpoint that carries each connection to190 * the peer named by `peerTicket` over iroh. */191export function startTunnelListener(192 node: IrohNode,193 peerTicket: string,194 options: { port?: number } = {},195): TunnelListener {196 const live = new Set<IrohConn>();197 const track = (conn: IrohConn) => {198 live.add(conn);199 };200 const collectStats = (direction: "out" | "in") => {201 const out: TunnelConnStat[] = [];202 for (const conn of live) {203 if (conn.closed) {204 live.delete(conn);205 continue;206 }207 out.push({ direction, ...conn.stats() });208 }209 return out;210 };211212 async function proxyHttp(req: Request): Promise<Response> {213 const url = new URL(req.url);214 let conn: IrohConn;215 try {216 conn = await node.connect(peerTicket);217 track(conn);218 } catch {219 return new Response("tunnel dial failed", { status: 502 });220 }221 try {222 const hello: Hello = {223 kind: "http",224 method: req.method,225 path: url.pathname + url.search,226 headers: forwardableHeaders(req.headers),227 };228 conn.sendMsg(encodeFrame(FRAME_HELLO, encoder.encode(JSON.stringify(hello))));229 conn.sendMsg(encodeFrame(FRAME_BIN, new Uint8Array(await req.arrayBuffer())));230 const headFrame = await conn.recvMsg();231 const bodyFrame = headFrame === null ? null : await conn.recvMsg();232 if (headFrame === null || bodyFrame === null) {233 return new Response("tunnel closed mid-response", { status: 502 });234 }235 const head = JSON.parse(decoder.decode(decodeFrame(headFrame).payload)) as HttpResponseHead;236 debug("dialer: http", hello.method, hello.path, "→", head.status);237 return new Response(toArrayBuffer(decodeFrame(bodyFrame).payload), {238 status: head.status,239 headers: head.headers,240 });241 } catch (e) {242 debug("dialer: http proxy error", (e as Error).message);243 return new Response("tunnel proxy error", { status: 502 });244 } finally {245 conn.close().catch(() => {});246 }247 }248249 const server = Deno.serve(250 { hostname: "127.0.0.1", port: options.port ?? 0, onListen: () => {} },251 (req) => {252 if (req.headers.get("upgrade")?.toLowerCase() !== "websocket") {253 return proxyHttp(req);254 }255 const requestedProtocol = req.headers.get("sec-websocket-protocol");256 const { socket, response } = Deno.upgradeWebSocket(257 req,258 requestedProtocol ? { protocol: requestedProtocol.split(",")[0].trim() } : {},259 );260 const url = new URL(req.url);261 // Buffer WS data that lands while the iroh dial is in flight; bridge()262 // flushes it before attaching its own listener (same task, no gap).263 const backlog: (string | ArrayBuffer)[] = [];264 const buffer = (ev: MessageEvent) => backlog.push(ev.data);265 socket.addEventListener("message", buffer);266 socket.addEventListener("open", async () => {267 try {268 const conn = await node.connect(peerTicket);269 track(conn);270 const hello: Hello = {271 kind: "ws",272 path: url.pathname + url.search,273 protocol: requestedProtocol,274 };275 debug("dialer: connected, HELLO", hello);276 conn.sendMsg(encodeFrame(FRAME_HELLO, encoder.encode(JSON.stringify(hello))));277 socket.removeEventListener("message", buffer);278 bridge(conn, socket, backlog);279 } catch {280 try {281 socket.close(1011, "tunnel dial failed");282 } catch {283 // already closed284 }285 }286 });287 return response;288 },289 );290 return {291 port: (server.addr as Deno.NetAddr).port,292 stats: () => collectStats("out"),293 close: () => server.shutdown(),294 };295}296297export interface TunnelAcceptor {298 /** Stats for the currently live inbound tunnel connections. */299 stats(): TunnelConnStat[];300 /** Stops handing new conns to the bridge; the loop also exits cleanly when301 * the endpoint closes (accept resolves null). */302 close(): void;303}304305/** Acceptor side: bridge every inbound iroh conn to the local Jazz server. */306export function runTunnelAcceptor(node: IrohNode, localWsUrl: string): TunnelAcceptor {307 let stopped = false;308 const localHttpUrl = localWsUrl.replace(/^ws/, "http");309 const live = new Set<IrohConn>();310311 async function handleHttp(conn: IrohConn, hello: Hello): Promise<void> {312 const bodyFrame = await conn.recvMsg();313 if (bodyFrame === null) {314 await conn.close();315 return;316 }317 const body = decodeFrame(bodyFrame).payload;318 const method = hello.method ?? "GET";319 let head: HttpResponseHead;320 let responseBody: Uint8Array;321 try {322 const res = await fetch(new URL(hello.path || "/", localHttpUrl), {323 method,324 headers: hello.headers,325 body: method === "GET" || method === "HEAD" ? undefined : toArrayBuffer(body),326 });327 head = { status: res.status, headers: forwardableHeaders(res.headers) };328 responseBody = new Uint8Array(await res.arrayBuffer());329 } catch (e) {330 debug("acceptor: local http error", (e as Error).message);331 head = { status: 502, headers: {} };332 responseBody = encoder.encode("local fetch failed");333 }334 debug("acceptor: http", method, hello.path, "→", head.status);335 try {336 conn.sendMsg(encodeFrame(FRAME_HELLO, encoder.encode(JSON.stringify(head))));337 conn.sendMsg(encodeFrame(FRAME_BIN, responseBody));338 } catch {339 // peer gone340 }341 await conn.close();342 }343344 async function handleInbound(conn: IrohConn): Promise<void> {345 const first = await conn.recvMsg();346 if (first === null) return;347 const { type, payload } = decodeFrame(first);348 if (type !== FRAME_HELLO) {349 await conn.close();350 return;351 }352 let hello: Hello;353 try {354 hello = JSON.parse(decoder.decode(payload)) as Hello;355 } catch {356 await conn.close();357 return;358 }359 if (hello.kind === "http") {360 await handleHttp(conn, hello);361 return;362 }363 const target = new URL(hello.path || "/", localWsUrl);364 debug("acceptor: HELLO", hello, "→ dialing local", target.href);365 const ws = hello.protocol366 ? new WebSocket(target, hello.protocol.split(",").map((s) => s.trim()))367 : new WebSocket(target);368 ws.addEventListener("open", () => {369 debug("acceptor: local ws open, bridging");370 bridge(conn, ws);371 });372 ws.addEventListener("error", (ev) => {373 debug("acceptor: local ws error", (ev as ErrorEvent).message ?? "");374 conn.close().catch(() => {});375 });376 }377378 (async () => {379 while (!stopped) {380 let conn: IrohConn | null;381 try {382 conn = await node.accept();383 } catch {384 if (stopped) break;385 await new Promise((r) => setTimeout(r, 250));386 continue;387 }388 if (conn === null) break; // endpoint closed — clean exit389 if (stopped) {390 conn.close().catch(() => {});391 break;392 }393 live.add(conn);394 handleInbound(conn).catch(() => {});395 }396 })();397398 return {399 stats: () => {400 const out: TunnelConnStat[] = [];401 for (const conn of live) {402 if (conn.closed) {403 live.delete(conn);404 continue;405 }406 out.push({ direction: "in", ...conn.stats() });407 }408 return out;409 },410 close: () => {411 stopped = true;412 },413 };414}