Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252import { CLOSE, PING_INTERVAL_MS, PROTOCOL_VERSION, parseMessage, type EventMessage, type Hello, type Response } from "../shared/protocol";import { Dispatcher, type Settled } from "./dispatcher";import { tokenMatches } from "./guard";import { newRunId, newToken } from "./ids";
export const OPEN = 1;
/** One accepted websocket, seen from the bridge's logic. */export interface BridgeSocket { readyState: number; send(data: string): void; close(code: number, reason: string): void; terminate(): void; onMessage(fn: (bytes: Uint8Array) => void): void; onClose(fn: () => void): void;}
/** Where connections come from. The real one is a loopback http + websocket server; tests use an in-memory one. */export interface Transport { /** Starts accepting. `presentedToken` is whatever the client put in the url (null if nothing). Resolves with the port. */ listen(port: number, onConnection: (socket: BridgeSocket, presentedToken: string | null) => void): Promise<number>; close(): Promise<void>;}
export interface BridgeOptions { port?: number; token?: string; runId?: string; pingIntervalMs?: number; helloTimeoutMs?: number;}
export interface Peer { extVersion: string; chrome: string; profile: string; resumed: boolean;}
export interface CoreDeps { setTimer(fn: () => void, ms: number): unknown; clearTimer(handle: unknown): void; setRepeating(fn: () => void, ms: number): unknown; clearRepeating(handle: unknown): void; now(): number;}
export const realCoreDeps: CoreDeps = { setTimer: (fn, ms) => setTimeout(fn, ms), clearTimer: (h) => clearTimeout(h as NodeJS.Timeout), setRepeating: (fn, ms) => { const t = setInterval(fn, ms); t.unref(); return t; }, clearRepeating: (h) => clearInterval(h as NodeJS.Timeout), now: () => Date.now(),};
interface Connection { socket: BridgeSocket; helloed: boolean; lastSeen: number; lastPingAt: number; missed: number; helloTimer: unknown; /** Set once this connection's end has been counted. */ ended: boolean;}
/** * The local end of the protocol. One extension connects; this class does the token check, the handshake, * the keep-alive and hands requests to the Dispatcher. It knows nothing about sockets beyond BridgeSocket. */export class BridgeCore { readonly token: string; readonly runId: string; port = 0; /** How many times an extension has been welcomed, and how many of those connections have since ended. */ readonly stats = { welcomes: 0, closes: 0 }; private readonly dispatcher: Dispatcher; private readonly sockets = new Set<BridgeSocket>(); private readonly eventHandlers: Array<(e: EventMessage) => void> = []; private readonly waiters: Array<{ wake: () => void; fail: (e: Error) => void }> = []; private readonly pingIntervalMs: number; private readonly helloTimeoutMs: number; private current: Connection | undefined; private currentPeer: Peer | undefined; private pingTimer: unknown; private closed = false;
constructor(private readonly transport: Transport, options: BridgeOptions = {}, private readonly deps: CoreDeps = realCoreDeps) { this.token = options.token ?? newToken(); this.runId = options.runId ?? newRunId(); this.pingIntervalMs = options.pingIntervalMs ?? PING_INTERVAL_MS; this.helloTimeoutMs = options.helloTimeoutMs ?? 5000; this.dispatcher = new Dispatcher({ send: (m) => { const c = this.current; if (!c || c.socket.readyState !== OPEN) throw new Error("no open connection"); c.socket.send(JSON.stringify(m)); }, setTimer: (fn, ms) => deps.setTimer(fn, ms), clearTimer: (h) => deps.clearTimer(h), }); }
async listen(port = 0): Promise<void> { this.port = await this.transport.listen(port, (socket, presented) => this.accept(socket, presented)); this.pingTimer = this.deps.setRepeating(() => this.tick(), this.pingIntervalMs); }
get endpoint(): string { return `ws://127.0.0.1:${this.port}/ext`; }
get connected(): boolean { return this.current?.helloed === true; }
get peer(): Peer | undefined { return this.connected ? this.currentPeer : undefined; }
request(body: { cmd: string; [key: string]: unknown }, options: { timeoutMs?: number } = {}): Promise<Settled> { return this.dispatcher.request(body, options); }
onEvent(handler: (e: EventMessage) => void): void { this.eventHandlers.push(handler); }
waitForConnection(timeoutMs: number): Promise<void> { if (this.connected) return Promise.resolve(); return new Promise((resolve, reject) => { const timer = this.deps.setTimer(() => { this.waiters.splice(this.waiters.indexOf(waiter), 1); reject(new Error(`no extension connected within ${timeoutMs} ms`)); }, timeoutMs); const waiter = { wake: () => { this.deps.clearTimer(timer); resolve(); }, fail: (e: Error) => { this.deps.clearTimer(timer); reject(e); }, }; this.waiters.push(waiter); }); }
async close(): Promise<void> { if (this.closed) return; this.closed = true; this.deps.clearRepeating(this.pingTimer); this.dispatcher.failAll("connection-lost", "the bridge was stopped"); for (const w of this.waiters.splice(0)) w.fail(new Error("the bridge was stopped")); for (const s of this.sockets) s.close(1001, "bridge stopped"); const force = this.deps.setTimer(() => { for (const s of this.sockets) s.terminate(); }, 300); await this.transport.close(); this.deps.clearTimer(force); }
private accept(socket: BridgeSocket, presented: string | null): void { this.sockets.add(socket); socket.onClose(() => this.sockets.delete(socket)); if (this.closed) { socket.close(1001, "bridge stopped"); return; } if (!tokenMatches(presented, this.token)) { socket.close(CLOSE.badToken, "bad token"); return; } const now = this.deps.now(); const conn: Connection = { socket, helloed: false, lastSeen: now, lastPingAt: now, missed: 0, helloTimer: undefined, ended: false }; conn.helloTimer = this.deps.setTimer(() => { if (!conn.helloed) socket.close(CLOSE.helloTimeout, "no hello"); }, this.helloTimeoutMs); socket.onMessage((bytes) => this.onMessage(conn, bytes)); socket.onClose(() => { this.deps.clearTimer(conn.helloTimer); this.end(conn); this.drop(conn); }); }
/** Forgets the connection as soon as it is known to be over, so `connected` never lags behind the truth. */ private drop(conn: Connection): void { if (this.current !== conn) return; this.end(conn); this.current = undefined; this.currentPeer = undefined; this.dispatcher.onDisconnected(); }
/** Counts the end of a connection that had said hello, once, however it ended (closed, replaced, or gone silent). */ private end(conn: Connection): void { if (conn.helloed && !conn.ended) { conn.ended = true; this.stats.closes++; } }
private onMessage(conn: Connection, bytes: Uint8Array): void { conn.lastSeen = this.deps.now(); const parsed = parseMessage(bytes); if (!parsed.ok) return; const m = parsed.message; if ("type" in m && m.type === "hello") return this.onHello(conn, m as Hello); if (!conn.helloed) return; // nothing counts before hello if ("type" in m) { if (m.type === "ping") conn.socket.send(JSON.stringify({ type: "pong" })); else if (m.type === "event") for (const h of this.eventHandlers) h(m as EventMessage); return; } if ("ok" in m) this.dispatcher.onResponse(m as Response); }
private onHello(conn: Connection, hello: Hello): void { if (conn.helloed) return; this.deps.clearTimer(conn.helloTimer); const previous = this.current; conn.helloed = true; this.stats.welcomes++; this.current = conn; this.currentPeer = { extVersion: hello.extVersion, chrome: hello.chrome, profile: hello.profile, resumed: hello.resume !== undefined }; if (previous && previous !== conn) previous.socket.close(CLOSE.replaced, "replaced by a newer connection"); conn.socket.send(JSON.stringify({ type: "welcome", runId: this.runId, protocol: PROTOCOL_VERSION })); this.dispatcher.onConnected(); for (const w of this.waiters.splice(0)) w.wake(); }
private tick(): void { const c = this.current; if (!c || c.socket.readyState !== OPEN) return; c.missed = c.lastSeen < c.lastPingAt ? c.missed + 1 : 0; if (c.missed >= 2) { this.drop(c); c.socket.close(CLOSE.silent, "no answer to pings"); this.deps.setTimer(() => c.socket.terminate(), 200); return; } c.lastPingAt = this.deps.now(); c.socket.send(JSON.stringify({ type: "ping" })); }}