diff --git a/README.md b/README.md index 4253c43..4ee9960 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,8 @@ ts/ `ts/packages/core` exists: a ports/adapters implementation (Transport, Storage, Identity/crypto, and Clock as first-class ports) consuming Zod schemas generated from `spec/protocol.cddl` by [cddl.js](https://github.com/ExaDev/cddl.js), with real domain logic for handshake negotiation and capability-token verification (including the delegation-chain narrowing rules `tokens.cddl` documents); its `conformance-check` round-trips every vector in `conformance/`'s golden suite through the generated schemas. +`ts/packages/web-console` exists: the browser client, a client of any node rather than just the hub. A browser-side Transport adapter implements core's port over native WebSocket (the same one-CBOR-frame-per-binary-message convention the hub speaks), and a DOM-free session module owns the client side of the handshake (with an explicit unanswered state for relay-only nodes), the gossip-derived peer directory, and the frame log the thin UI renders. Built to static assets with vite and servable from any origin; verified against a locally-running hub with the same adapter code the browser executes. + `ts/packages/cloudflare-hub` exists: a reference Cloudflare Worker deployment of a public, always-on hub node, depending on core as an ordinary workspace consumer — Worker-shaped adapters for core's ports (WebSocket-message Transport, Web Crypto Identity with signature interop proven against core's Node adapter in both directions) rather than any reinvented protocol logic, serving the relay role with gossip-based pairing. CI validates the bundle with `wrangler deploy --dry-run`; a real deploy is an authenticated one-off, not part of CI. `rust/` exists too, mirroring the same architecture: `wire-mesh-wire` carries the CDDL model as hand-written types over a minicbor codec with CDE (canonical) encoding by construction and strict decoding (unknown keys, indefinite lengths, and non-canonical shapes all rejected), and `wire-mesh-core` carries the ports (Transport, KeyValueStorage, Identity, Clock), the same domain logic (handshake negotiation, capability-token verification with narrowing/expiry-clamping and the signed-revocation obligations, including the ancestor-chain sweep), and Tokio TCP / in-memory / Ed25519+ES256 adapters. Its `conformance-check` binary proves every golden vector round-trips byte-exactly in both directions. There is no CDDL-to-Rust generator (cddl.js emits TypeScript/Zod only), so the wire types are hand-written against `spec/*.cddl` — the frozen vectors are the cross-language pin, exactly why they exist. diff --git a/ts/packages/web-console/README.md b/ts/packages/web-console/README.md new file mode 100644 index 0000000..afd2ef0 --- /dev/null +++ b/ts/packages/web-console/README.md @@ -0,0 +1,37 @@ +# @exadev/wire-mesh-web-console + +The browser client for wire-mesh: a client of *any* node, not just the hub. It connects over WebSocket to a node's endpoint, drives the connection through core's Transport port via a browser-side adapter, and renders what the node has to show. Built as static assets (vite), servable from any static origin — including the hub's, for deployment convenience, with no Cloudflare tie whatsoever. + +## What it does + +- **Connect** — enter a node's `ws://`/`wss://` URL (defaulting to `ws://localhost:8787`, the hub's `wrangler dev` port), pick the capability domains to offer, and connect. The console sends its handshake and shows the negotiated result once the node answers. +- **Peer directory** — every `gossip` frame the node sends is folded into a directory table (device, addresses, snapshot time), latest advert per device winning. +- **Frame log** — a live, ordered feed of every frame that crossed the connection in either direction, plus a *Send ping* button. + +## How it maps onto core's ports + +`src/adapters/websocket-transport.ts` implements core's `Transport`/`Connection` port over the browser's native WebSocket, the same message convention as the hub's Worker-side adapter (one CBOR frame per binary message, no length prefix; undecodable bytes reject the connection, a decodable-but-unknown frame drops without disconnecting). `src/mesh-session.ts` is the DOM-free session state machine — handshake exchange with an explicit *unanswered* state (a relay-only node like the hub legitimately never answers a handshake; that's displayed, not treated as an error), directory assembly, and the frame log — unit-tested against a fake Transport. `src/main.ts` is deliberately thin DOM wiring. + +## Connecting to a locally-running node + +```sh +# terminal 1: run the hub (a wire-mesh node) locally on :8787 +cd ts/packages/cloudflare-hub && pnpm dev # wrangler dev, no credentials needed + +# terminal 2: serve the console +cd ts/packages/web-console && pnpm dev # vite on :5173 +``` + +Open the vite URL, keep the default `ws://localhost:8787` address, and connect: the status line reaches *connected*, the handshake moves to *unanswered* after the timeout (the relay-only hub never sends one back — that is the honest state, not a failure), and *Send ping* records the outgoing frame in the log (the hub drops non-relay frames by design). As fuller node implementations start answering handshakes and gossiping, the negotiated-domains line and the directory table light up with no console changes. + +Verified exactly that way during development, including an end-to-end run of the *actual* adapter and session modules (not a reimplementation) against a live `wrangler dev` hub: Node 26 provides the same native WebSocket the browser does, so the same code path a browser executes connected to `ws://localhost:8787`, reached `connected`, observed the handshake go `unanswered` after the timeout (the relay-only hub never sends one -- an honest state, not a failure), and recorded an outgoing ping in the frame log. The unit suite (`pnpm test`) covers the adapter and session against fakes; the live-hub run is a development-time verification, not part of CI, because it would couple the console's CI to a running workerd. + +## What is deliberately deferred + +- **Room browser / join-from-browser** — the plan's phrase for the agent-comms-shaped application layer. wire-mesh's `core/data` domain carries such content as opaque entries, but no node implementation serves room semantics yet; shipping dead UI for it would be dishonest. The connection + directory + frame inspector is the honest first pass; the room UI arrives with the application that defines it. +- **Identity** — the console connects anonymously (no local keypair). When a node requires a client identity, core's Identity port gets a Web Crypto adapter here, exactly like the hub's. +- **TLS in dev** — `ws://` against localhost is fine; production deployments serve the console over HTTPS and dial `wss://`, which the adapter already handles. + +## Type environment + +`src/` typechecks against the DOM lib. The tests run in Node under vitest but import that src, so the test tsconfig loads node types alongside the DOM lib — and since Node 26 ships the same native WebSocket global the browser does, the adapter runs unmodified under Node, which is what makes the automated end-to-end test against a real hub possible. diff --git a/ts/packages/web-console/eslint.config.ts b/ts/packages/web-console/eslint.config.ts new file mode 100644 index 0000000..9786e4d --- /dev/null +++ b/ts/packages/web-console/eslint.config.ts @@ -0,0 +1,29 @@ +import { exadevConfig } from "@exadev/eslint-config"; +import eslintPluginPrettierRecommended from "eslint-plugin-prettier/recommended"; +import globals from "globals"; + +export default exadevConfig( + {}, + { + ignores: ["dist", "coverage", "node_modules", ".turbo"], + }, + { + languageOptions: { + parserOptions: { + project: ["./tsconfig.json", "./tsconfig.node.json"], + tsconfigRootDir: import.meta.dirname, + }, + // Browser globals for the app source, node globals for vite/vitest config and tests -- the tests drive the browser-shaped adapter under Node, which provides the same WebSocket global natively. + globals: { ...globals.browser, ...globals.node }, + }, + }, + { + rules: { + "@typescript-eslint/consistent-type-imports": [ + "error", + { fixStyle: "inline-type-imports" }, + ], + }, + }, + eslintPluginPrettierRecommended, +); diff --git a/ts/packages/web-console/index.html b/ts/packages/web-console/index.html new file mode 100644 index 0000000..254fd63 --- /dev/null +++ b/ts/packages/web-console/index.html @@ -0,0 +1,99 @@ + + + + + + wire-mesh console + + + +

wire-mesh console

+
+
+ + + + + +
+

+ Domains offered: + + + +

+
+

idle

+
+

Peer directory

+

No gossip received yet.

+ + + + + + +
+
+

Frame log

+ + + + + +
directionframe
+
+ + + diff --git a/ts/packages/web-console/package.json b/ts/packages/web-console/package.json new file mode 100644 index 0000000..3ceb722 --- /dev/null +++ b/ts/packages/web-console/package.json @@ -0,0 +1,36 @@ +{ + "name": "@exadev/wire-mesh-web-console", + "version": "0.0.0", + "private": true, + "type": "module", + "packageManager": "pnpm@10.33.0", + "scripts": { + "build": "turbo run _build", + "_build": "vite build", + "dev": "vite", + "test": "turbo run _test", + "_test": "vitest run", + "typecheck": "turbo run _typecheck", + "_typecheck": "tsc -p tsconfig.json --noEmit && tsc -p tsconfig.node.json --noEmit", + "lint": "turbo run _lint", + "_lint": "eslint . --fix --cache --max-warnings 0" + }, + "dependencies": { + "@exadev/wire-mesh-core": "workspace:*", + "cbor2": "2.3.0" + }, + "devDependencies": { + "@exadev/eslint-config": "2.10.6", + "@types/node": "26.4.1", + "eslint": "10.10.0", + "eslint-config-prettier": "10.1.8", + "eslint-plugin-prettier": "5.5.6", + "globals": "17.12.0", + "jiti": "2.7.0", + "prettier": "3.9.6", + "turbo": "2.10.12", + "typescript": "6.0.3", + "vite": "8.2.2", + "vitest": "5.0.0" + } +} diff --git a/ts/packages/web-console/src/adapters/websocket-transport.ts b/ts/packages/web-console/src/adapters/websocket-transport.ts new file mode 100644 index 0000000..cba4052 --- /dev/null +++ b/ts/packages/web-console/src/adapters/websocket-transport.ts @@ -0,0 +1,162 @@ +// A browser Transport implementation over native WebSocket messages, the same convention as the hub's Worker-side adapter: each binary WebSocket message is already self-delimiting, so one message carries exactly one CBOR-encoded frame with no length prefix. Undecodable bytes are a connection-level failure (the receive iteration rejects and the socket closes), matching core's adapters' treatment of hostile wire input; a decodable frame that fails schema validation is dropped rather than disconnecting -- an unrecognised frame from a newer peer is what version negotiation exists to tolerate. + +import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2"; +import { + frameSchema, + type Frame, +} from "@exadev/wire-mesh-core/generated/protocol"; +import type { + Connection, + Listener, + Transport, +} from "@exadev/wire-mesh-core/ports/transport"; + +// RFC 6455 close codes, named rather than bare: 1000 normal closure, 1002 protocol error. +const CLOSE_NORMAL = 1000; +const CLOSE_PROTOCOL_ERROR = 1002; + +/** A console cannot accept inbound connections, so listen() always rejects -- the port is a client here, exactly like core's other edge adapters are servers where the runtime allows it. */ +const LISTEN_UNSUPPORTED = "the web console is a client only: it cannot listen"; + +export function messageFromFrame(frame: Frame): Uint8Array { + // A fresh whole-buffer view over a plain ArrayBuffer: the WebSocket send signatures require it, and it matches the fresh-buffer discipline the other adapters apply to anything crossing a runtime boundary. + return new Uint8Array(encode(frame, cdeEncodeOptions)); +} + +/** A frame that fails schema validation, caught separately from a decode failure so it can be dropped without disconnecting. */ +class SchemaInvalidFrameError extends Error { + constructor(message: string) { + super(message); + this.name = "SchemaInvalidFrameError"; + } +} + +/** Decodes one message, distinguishing a decode failure (connection-level) from a schema failure (drop this frame, keep the connection). */ +function decodeMessage(data: Readonly): Frame { + const decoded: unknown = decode(new Uint8Array(data), cdeDecodeOptions); + const result = frameSchema.safeParse(decoded); + if (!result.success) { + throw new SchemaInvalidFrameError(result.error.message); + } + return result.data; +} + +export function wrapWebSocket(ws: Readonly): Connection { + const pending: Frame[] = []; + const waiters: { + resolve: (result: IteratorResult) => void; + reject: (error: unknown) => void; + }[] = []; + let ended = false; + let failure: Error | null = null; + + function endAll(): void { + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.resolve({ value: undefined, done: true }); + } + } + + function failAll(error: Error): void { + failure = error; + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.reject(error); + } + } + + ws.addEventListener("message", (event: MessageEvent) => { + if (!(event.data instanceof ArrayBuffer)) { + // Only binary messages carry frames; a text message is a protocol violation on this connection, same class as undecodable bytes. + failAll(new Error("expected a binary WebSocket message")); + ws.close(CLOSE_PROTOCOL_ERROR, "protocol error"); + return; + } + let frame: Frame; + try { + frame = decodeMessage(event.data); + } catch (error) { + if (error instanceof SchemaInvalidFrameError) { + return; + } + failAll( + error instanceof Error + ? error + : new Error(`frame body failed to decode: ${String(error)}`), + ); + ws.close(CLOSE_PROTOCOL_ERROR, "protocol error"); + return; + } + const waiter = waiters.shift(); + if (waiter) { + waiter.resolve({ value: frame, done: false }); + } else { + pending.push(frame); + } + }); + ws.addEventListener("close", endAll); + ws.addEventListener("error", endAll); + + const receiveStream: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + next: async (): Promise> => receiveNext(), + }; + }, + }; + + async function receiveNext(): Promise> { + const next = pending.shift(); + if (next !== undefined) { + return Promise.resolve({ value: next, done: false }); + } + if (failure !== null) { + return Promise.reject(failure); + } + if (ended) { + return Promise.resolve({ value: undefined, done: true }); + } + return new Promise((resolve, reject) => { + waiters.push({ resolve, reject }); + }); + } + + return { + async send(frame): Promise { + if (ended) { + return Promise.reject(new Error("connection is closed")); + } + ws.send(messageFromFrame(frame)); + return Promise.resolve(); + }, + receive: () => receiveStream, + async close(): Promise { + ws.close(CLOSE_NORMAL); + return Promise.resolve(); + }, + }; +} + +export function createBrowserTransport(): Transport { + return { + async connect(address): Promise { + const url = new URL(address); + if (url.protocol !== "ws:" && url.protocol !== "wss:") { + return Promise.reject(new Error(`not a WebSocket address: ${address}`)); + } + const ws = new WebSocket(url); + ws.binaryType = "arraybuffer"; + return new Promise((resolve, reject) => { + ws.addEventListener("open", () => { + resolve(); + }); + ws.addEventListener("error", () => { + reject(new Error(`connect to ${url.toString()} failed`)); + }); + }).then(() => wrapWebSocket(ws)); + }, + async listen(): Promise { + return Promise.reject(new Error(LISTEN_UNSUPPORTED)); + }, + }; +} diff --git a/ts/packages/web-console/src/main.ts b/ts/packages/web-console/src/main.ts new file mode 100644 index 0000000..72bfae8 --- /dev/null +++ b/ts/packages/web-console/src/main.ts @@ -0,0 +1,152 @@ +// The console's DOM wiring: one MeshSession per connection attempt, rendering each SessionEvent into the connection status line, the peer directory table, and the frame log. Kept thin on purpose -- everything with behaviour lives in mesh-session.ts so it can be tested without a browser. + +import { createBrowserTransport } from "./adapters/websocket-transport.js"; +import { createMeshSession } from "./mesh-session.js"; +import type { SessionEvent } from "./mesh-session.js"; + +// Passing the constructor rather than asserting: T appears in both the parameter and return, and the instanceof check makes the lookup self-verifying at runtime. +function requireElement( + id: string, + kind: new () => E, +): E { + const element = document.getElementById(id); + if (!(element instanceof kind)) { + throw new Error(`missing element #${id}`); + } + return element; +} + +const form = requireElement("connect-form", HTMLFormElement); +const addressInput = requireElement("node-address", HTMLInputElement); +const connectButton = requireElement("connect-button", HTMLButtonElement); +const pingButton = requireElement("ping-button", HTMLButtonElement); +const closeButton = requireElement("close-button", HTMLButtonElement); +const statusLine = requireElement("connection-status", HTMLParagraphElement); +const directoryEmpty = requireElement("directory-empty", HTMLParagraphElement); +const directoryTable = requireElement("directory-table", HTMLTableElement); +const directoryBody = requireElement("directory-body", HTMLTableSectionElement); +const frameLogBody = requireElement("frame-log", HTMLTableSectionElement); + +const HEX_RADIX = 16; + +function deviceHex(device: Uint8Array): string { + let hex = ""; + for (const byte of device) { + hex += byte.toString(HEX_RADIX).padStart(2, "0"); + } + return hex; +} + +function describeHandshake(event: SessionEvent): string { + if (event.state.status !== "connected") { + return ""; + } + switch (event.state.handshake.status) { + case "pending": + return " · handshake pending"; + case "negotiated": + return ` · v${String(event.state.handshake.version)} · ${event.state.handshake.sharedDomains.join(", ")}`; + case "unanswered": + return " · handshake unanswered (relay-only node?)"; + case "rejected": + return ` · handshake rejected (${event.state.handshake.reason})`; + } + return ""; +} + +function render(event: SessionEvent): void { + const { state } = event; + statusLine.textContent = + state.status === "idle" + ? "idle" + : state.status === "connecting" + ? `connecting to ${state.address}…` + : state.status === "connected" + ? `connected to ${state.address}${describeHandshake(event)}` + : `closed (${state.reason})`; + + const connected = state.status === "connected"; + pingButton.disabled = !connected; + closeButton.disabled = !connected; + connectButton.disabled = connected || state.status === "connecting"; + addressInput.disabled = connected || state.status === "connecting"; + + directoryEmpty.hidden = event.directory.length > 0; + directoryTable.hidden = event.directory.length === 0; + directoryBody.replaceChildren( + ...event.directory.map((entry) => { + const row = document.createElement("tr"); + const deviceCell = document.createElement("td"); + deviceCell.textContent = deviceHex(entry.device); + const addressesCell = document.createElement("td"); + addressesCell.textContent = entry.advert.addresses.join(", "); + const snapshotCell = document.createElement("td"); + snapshotCell.textContent = String(entry.advert["snapshot-seconds"]); + row.append(deviceCell, addressesCell, snapshotCell); + return row; + }), + ); + + frameLogBody.replaceChildren( + ...event.frameLog.map((entry) => { + const row = document.createElement("tr"); + const directionCell = document.createElement("td"); + directionCell.textContent = entry.direction; + const frameCell = document.createElement("td"); + frameCell.textContent = JSON.stringify( + entry.frame, + (_key: string, value: unknown): unknown => + value instanceof Uint8Array + ? `<${String(value.byteLength)} bytes>` + : value, + ); + row.append(directionCell, frameCell); + return row; + }), + ); + const frameLogTable = frameLogBody.parentElement; + if (frameLogTable !== null) { + frameLogTable.scrollTop = frameLogTable.scrollHeight; + } +} + +let session: ReturnType | null = null; + +form.addEventListener("submit", (event) => { + event.preventDefault(); + if (session !== null) { + return; + } + const domains = [ + ...form.querySelectorAll("input[name='domain']:checked"), + ].map((checkbox) => checkbox.value); + session = createMeshSession(createBrowserTransport()); + void (async () => { + for await (const sessionEvent of session.events) { + render(sessionEvent); + } + })(); + void session.connect(addressInput.value, domains).catch((error: unknown) => { + statusLine.textContent = `connect failed: ${error instanceof Error ? error.message : String(error)}`; + session = null; + }); +}); + +pingButton.addEventListener("click", () => { + if (session !== null) { + void session.sendPing().catch((error: unknown) => { + statusLine.textContent = `send failed: ${error instanceof Error ? error.message : String(error)}`; + }); + } +}); + +closeButton.addEventListener("click", () => { + if (session !== null) { + void session.close(); + session = null; + connectButton.disabled = false; + addressInput.disabled = false; + pingButton.disabled = true; + closeButton.disabled = true; + } +}); diff --git a/ts/packages/web-console/src/mesh-session.ts b/ts/packages/web-console/src/mesh-session.ts new file mode 100644 index 0000000..11cfbf0 --- /dev/null +++ b/ts/packages/web-console/src/mesh-session.ts @@ -0,0 +1,237 @@ +// DOM-free connection session: everything the console does once "Connect" is clicked, kept free of browser APIs so it is unit-testable against a fake Transport. Owns the client side of the handshake exchange (send ours, negotiate against theirs, with an explicit unanswered state rather than hanging forever -- a relay-only node like the hub legitimately never answers a handshake), the peer directory assembled from received gossip frames, and the frame feed the UI renders. + +import { + type DeviceId, + type Frame, + type HandshakeFrame, + type PeerAdvert, + type ProtocolVersion, +} from "@exadev/wire-mesh-core/generated/protocol"; +import { + SUPPORTED_PROTOCOL_VERSION, + negotiate, +} from "@exadev/wire-mesh-core/domain/handshake"; +import type { + Connection, + Transport, +} from "@exadev/wire-mesh-core/ports/transport"; + +/** How long to wait for the node's handshake before calling it unanswered. A relay-only node never sends one; that is a state to display, not an error. */ +export const HANDSHAKE_TIMEOUT_MS = 3_000; + +export type ConnectionState = + | { status: "idle" } + | { status: "connecting"; address: string } + | { status: "connected"; address: string; handshake: HandshakeStatus } + | { status: "closed"; address: string; reason: string }; + +export type HandshakeStatus = + | { status: "pending" } + | { status: "negotiated"; version: ProtocolVersion; sharedDomains: string[] } + | { status: "unanswered" } + | { status: "rejected"; reason: string }; + +export interface SessionEvent { + state: ConnectionState; + /** The peer directory as of this event: latest peer-advert per device-id, in first-heard order. */ + directory: readonly DirectoryEntry[]; + /** Every frame that crossed the connection, sent or received, in order. */ + frameLog: readonly FrameLogEntry[]; +} + +export interface DirectoryEntry { + device: DeviceId; + advert: PeerAdvert; +} + +export interface FrameLogEntry { + direction: "sent" | "received"; + frame: Frame; +} + +export interface MeshSession { + readonly events: AsyncIterable; + connect: (address: string, localDomains: readonly string[]) => Promise; + sendPing: () => Promise; + close: () => Promise; +} + +function localHandshake(domains: readonly string[]): HandshakeFrame { + return { + type: "handshake", + version: SUPPORTED_PROTOCOL_VERSION, + domains: [...domains], + }; +} + +export function createMeshSession(transport: Readonly): MeshSession { + let connection: Connection | null = null; + let state: ConnectionState = { status: "idle" }; + let handshake: HandshakeStatus = { status: "pending" }; + const directory = new Map(); + const frameLog: FrameLogEntry[] = []; + let feedCancelled = false; + const eventWaiters: ((event: SessionEvent) => void)[] = []; + const eventBacklog: SessionEvent[] = []; + + function snapshot(): SessionEvent { + return { + state, + directory: [...directory.values()], + frameLog: [...frameLog], + }; + } + + function emit(): void { + const event = snapshot(); + const waiter = eventWaiters.shift(); + if (waiter) { + waiter(event); + } else { + eventBacklog.push(event); + } + } + + function applyFrame(frame: Frame): void { + frameLog.push({ direction: "received", frame }); + if (frame.type === "handshake") { + applyRemoteHandshake(frame); + } else if (frame.type === "gossip") { + for (const advert of frame.peers) { + // Latest advert per device wins, order preserved by first insertion -- a re-advert updates in place. + directory.set(deviceKey(advert.device), { + device: advert.device, + advert, + }); + } + } + } + + const HEX_RADIX = 16; + + function deviceKey(device: DeviceId): string { + // Map key for a device-id: byte-exact hex rather than any coercions that would collide distinct ids. + let key = ""; + for (const byte of device) { + key += byte.toString(HEX_RADIX).padStart(2, "0"); + } + return key; + } + + // The local handshake actually sent on connect, kept for negotiating against the remote's answer. + let localHandshakeSent: HandshakeFrame = localHandshake([]); + + function applyRemoteHandshake(remote: HandshakeFrame): void { + if (handshake.status !== "pending") { + return; + } + const result = negotiate(localHandshakeSent, remote); + handshake = result.ok + ? { + status: "negotiated", + version: result.version, + sharedDomains: result.sharedDomains, + } + : { status: "rejected", reason: "no shared domains or version" }; + if (state.status === "connected") { + state = { ...state, handshake }; + } + } + + async function consume(link: Readonly): Promise { + for await (const frame of link.receive()) { + if (feedCancelled) { + return; + } + applyFrame(frame); + emit(); + } + if (state.status === "connected") { + state = { + status: "closed", + address: state.address, + reason: "node closed the connection", + }; + emit(); + } + } + + return { + events: { + [Symbol.asyncIterator]() { + return { + next: async (): Promise> => + new Promise((resolve) => { + const backlogEvent = eventBacklog.shift(); + if (backlogEvent) { + resolve({ value: backlogEvent, done: false }); + } else { + eventWaiters.push((event) => { + resolve({ value: event, done: false }); + }); + } + }), + }; + }, + }, + async connect(address, localDomains): Promise { + if (connection !== null) { + throw new Error( + "a session connects once; create a new one to reconnect", + ); + } + state = { status: "connecting", address }; + emit(); + connection = await transport.connect(address); + localHandshakeSent = localHandshake(localDomains); + handshake = { status: "pending" }; + state = { status: "connected", address, handshake }; + frameLog.push({ direction: "sent", frame: localHandshakeSent }); + await connection.send(localHandshakeSent); + emit(); + setTimeout(() => { + if (handshake.status === "pending") { + handshake = { status: "unanswered" }; + if (state.status === "connected") { + state = { ...state, handshake }; + } + emit(); + } + }, HANDSHAKE_TIMEOUT_MS); + const consuming = consume(connection); + void consuming.catch((error: unknown) => { + if (state.status === "connected") { + state = { + status: "closed", + address: state.address, + reason: error instanceof Error ? error.message : String(error), + }; + emit(); + } + }); + }, + async sendPing(): Promise { + if (connection === null || state.status !== "connected") { + throw new Error("not connected"); + } + const ping: Frame = { type: "ping" }; + frameLog.push({ direction: "sent", frame: ping }); + await connection.send(ping); + emit(); + }, + async close(): Promise { + feedCancelled = true; + if (connection !== null) { + await connection.close(); + } + if (state.status === "connected" || state.status === "connecting") { + state = { + status: "closed", + address: state.address, + reason: "closed by you", + }; + } + emit(); + }, + }; +} diff --git a/ts/packages/web-console/test/fake-websocket.ts b/ts/packages/web-console/test/fake-websocket.ts new file mode 100644 index 0000000..1c715ec --- /dev/null +++ b/ts/packages/web-console/test/fake-websocket.ts @@ -0,0 +1,48 @@ +// A minimal stand-in for the browser's WebSocket, firing events and recording sends -- it stands in for the platform (the side of the port this adapter does NOT own), exactly what a unit test of an adapter should fake. + +type Listener = (event: { data?: unknown }) => void; + +export class FakeWebSocket { + binaryType = "arraybuffer"; + sent: ArrayBuffer[] = []; + closed = false; + closeCode: number | null = null; + private readonly listeners = new Map(); + + constructor(public readonly url: string) {} + + addEventListener(type: string, listener: Listener): void { + const list = this.listeners.get(type) ?? []; + list.push(listener); + this.listeners.set(type, list); + } + + send(data: Readonly): void { + this.sent.push(data); + } + + close(code?: number): void { + this.closed = true; + this.closeCode = code ?? null; + this.dispatch("close"); + } + + // Test-side drivers + emitMessage(data: Readonly): void { + this.dispatch("message", { data }); + } + + emitText(text: string): void { + this.dispatch("message", { data: text }); + } + + emitError(): void { + this.dispatch("error"); + } + + private dispatch(type: string, event?: { data?: unknown }): void { + for (const listener of this.listeners.get(type) ?? []) { + listener(event ?? {}); + } + } +} diff --git a/ts/packages/web-console/test/hex.ts b/ts/packages/web-console/test/hex.ts new file mode 100644 index 0000000..a415fc5 --- /dev/null +++ b/ts/packages/web-console/test/hex.ts @@ -0,0 +1,22 @@ +// Test-side byte construction from hex strings, avoiding the Buffer global (which resolves as any under mixed browser/node type environments). +const HEX_PAIR_LENGTH = 2; +const HEX_RADIX = 16; +const SHA256_BYTE_LENGTH = 32; + +export function bytesFromHex(hex: string): Uint8Array { + const out = new Uint8Array(hex.length / HEX_PAIR_LENGTH); + for (let i = 0; i < out.length; i++) { + out[i] = Number.parseInt( + hex.slice(i * HEX_PAIR_LENGTH, (i + 1) * HEX_PAIR_LENGTH), + HEX_RADIX, + ); + } + return out; +} + +/** A full synthetic device-id from a one-byte hex fill, the conformance suite's repeated-byte convention. */ +export function deviceIdFromFillHex(fillHex: string): Uint8Array { + const out = new Uint8Array(SHA256_BYTE_LENGTH); + out.fill(Number.parseInt(fillHex, HEX_RADIX)); + return out; +} diff --git a/ts/packages/web-console/test/mesh-session.test.ts b/ts/packages/web-console/test/mesh-session.test.ts new file mode 100644 index 0000000..ee3004f --- /dev/null +++ b/ts/packages/web-console/test/mesh-session.test.ts @@ -0,0 +1,276 @@ +import { describe, expect, it, vi } from "vitest"; +import type { + Frame, + HandshakeFrame, +} from "@exadev/wire-mesh-core/generated/protocol"; +import type { + Connection, + Listener, + Transport, +} from "@exadev/wire-mesh-core/ports/transport"; +import { + HANDSHAKE_TIMEOUT_MS, + createMeshSession, +} from "../src/mesh-session.js"; +import { deviceIdFromFillHex } from "./hex.js"; + +const deviceA = deviceIdFromFillHex("11"); +const deviceB = deviceIdFromFillHex("22"); + +// Event-stream positions: connect() emits connecting + connected, then one event per pushed frame, timeout, or failure. +const EVENTS_THROUGH_REMOTE_HANDSHAKE = 3; +const EVENTS_THROUGH_TIMEOUT = 3; +const EVENTS_THROUGH_THREE_GOSSIPS = 5; +const EVENTS_THROUGH_PING_ROUND_TRIP = 4; +const EVENTS_THROUGH_FAILURE = 3; +const SNAPSHOT_FIRST = 100; +const SNAPSHOT_SECOND = 200; +const SNAPSHOT_UPDATED = 300; + +/** An in-memory Connection the test drives: pushes arrive on the receive iteration, sends are recorded. */ +class FakeConnection { + sent: Frame[] = []; + private readonly inbound: Frame[] = []; + private ended = false; + private failure: Error | null = null; + + get connection(): Readonly { + return { + send: async (frame: Frame): Promise => { + this.sent.push(frame); + return Promise.resolve(); + }, + receive: () => this.stream(), + close: async (): Promise => { + this.ended = true; + this.wake(); + return Promise.resolve(); + }, + }; + } + + push(frame: Frame): void { + this.inbound.push(frame); + this.wake(); + } + + fail(error: Error): void { + this.failure = error; + this.wake(); + } + + private readonly wakeWaiters: (() => void)[] = []; + + private wake(): void { + for (const wake of this.wakeWaiters.splice(0)) { + wake(); + } + } + + private stream(): AsyncIterable { + return { + [Symbol.asyncIterator]: () => ({ + next: async (): Promise> => this.nextFrame(), + }), + }; + } + + private async nextFrame(): Promise> { + for (;;) { + const next = this.inbound.shift(); + if (next !== undefined) { + return { value: next, done: false }; + } + if (this.failure !== null) { + throw this.failure; + } + if (this.ended) { + return { value: undefined, done: true }; + } + await new Promise((resolve) => { + this.wakeWaiters.push(resolve); + }); + } + } +} + +function fakeTransport(): { transport: Transport; connection: FakeConnection } { + const connection = new FakeConnection(); + const transport: Transport = { + connect: async (address: string): Promise => { + if (address !== "ws://node") { + return Promise.reject(new Error(`connect to ${address} failed`)); + } + return Promise.resolve(connection.connection); + }, + listen: async (): Promise => + Promise.reject(new Error("client-only transport")), + }; + return { transport, connection }; +} + +/** Resolves after the session has emitted at least `count` events, returning the latest. */ +async function nthEvent( + session: ReturnType, + count: number, +): Promise { + const iterator = session.events[Symbol.asyncIterator](); + let last: unknown = null; + for (let i = 0; i < count; i++) { + const result = await iterator.next(); + last = result.value; + } + return last; +} + +function gossipFor( + device: Uint8Array, + seconds = 1861833600, +): Frame { + return { + type: "gossip", + peers: [ + { device, addresses: ["203.0.113.5:4433"], "snapshot-seconds": seconds }, + ], + }; +} + +describe("createMeshSession", () => { + it("connects, sends the local handshake, and negotiates against the node's answer", async () => { + const { transport, connection } = fakeTransport(); + const session = createMeshSession(transport); + // Events: connecting, connected(handshake sent/pending), connected/negotiated + const eventsDone = nthEvent(session, EVENTS_THROUGH_REMOTE_HANDSHAKE); + await session.connect("ws://node", ["core/management", "core/data"]); + + expect(connection.sent[0]).toEqual({ + type: "handshake", + version: 1, + domains: ["core/management", "core/data"], + } satisfies HandshakeFrame); + + const answer: HandshakeFrame = { + type: "handshake", + version: 1, + domains: ["core/data", "core/exec"], + }; + connection.push(answer); + const event = (await eventsDone) as { + state: { + status: string; + handshake: { status: string; sharedDomains: string[] }; + }; + }; + expect(event.state.status).toBe("connected"); + expect(event.state.handshake.status).toBe("negotiated"); + expect(event.state.handshake.sharedDomains).toEqual(["core/data"]); + await session.close(); + }); + + it("excludes the retired core/federation domain even when both sides offer it", async () => { + const { transport, connection } = fakeTransport(); + const session = createMeshSession(transport); + // Events: connecting, connected, connected/rejected + const eventsDone = nthEvent(session, EVENTS_THROUGH_REMOTE_HANDSHAKE); + await session.connect("ws://node", ["core/federation"]); + connection.push({ + type: "handshake", + version: 1, + domains: ["core/federation"], + }); + const event = (await eventsDone) as { + state: { handshake: { status: string } }; + }; + expect(event.state.handshake.status).toBe("rejected"); + await session.close(); + }); + + it("marks the handshake unanswered after the timeout instead of hanging", async () => { + vi.useFakeTimers(); + try { + const { transport } = fakeTransport(); + const session = createMeshSession(transport); + await session.connect("ws://node", ["core/data"]); + await vi.advanceTimersByTimeAsync(HANDSHAKE_TIMEOUT_MS); + // Events so far: connecting, connected -- then the timeout emits unanswered + const event = (await nthEvent(session, EVENTS_THROUGH_TIMEOUT)) as { + state: { handshake: { status: string } }; + }; + expect(event.state.handshake.status).toBe("unanswered"); + await session.close(); + } finally { + vi.useRealTimers(); + } + }); + + it("assembles the peer directory from gossip, latest advert per device winning", async () => { + const { transport, connection } = fakeTransport(); + const session = createMeshSession(transport); + await session.connect("ws://node", ["core/data"]); + connection.push(gossipFor(deviceA, SNAPSHOT_FIRST)); + connection.push(gossipFor(deviceB, SNAPSHOT_SECOND)); + connection.push(gossipFor(deviceA, SNAPSHOT_UPDATED)); + // Events: connecting, connected, then one per gossip push -- drain to the last + const event = (await nthEvent(session, EVENTS_THROUGH_THREE_GOSSIPS)) as { + directory: { + device: Uint8Array; + advert: { "snapshot-seconds": number }; + }[]; + }; + expect(event.directory.length).toBe(2); + const entryA = event.directory.find( + (entry) => entry.advert["snapshot-seconds"] === SNAPSHOT_UPDATED, + ); + expect(entryA).toBeDefined(); + expect(event.directory[0]?.device).toEqual(deviceA); + expect(event.directory[1]?.device).toEqual(deviceB); + await session.close(); + }); + + it("records sent and received frames in the frame log in order", async () => { + const { transport, connection } = fakeTransport(); + const session = createMeshSession(transport); + await session.connect("ws://node", ["core/data"]); + await session.sendPing(); + connection.push({ type: "ping" }); + // Events: connecting, connected, ping sent, ping received + const event = (await nthEvent(session, EVENTS_THROUGH_PING_ROUND_TRIP)) as { + frameLog: { direction: string; frame: { type: string } }[]; + }; + expect(event.frameLog.map((entry) => entry.direction)).toEqual([ + "sent", + "sent", + "received", + ]); + await session.close(); + }); + + it("closes with the node's reason when the receive iteration rejects", async () => { + const { transport, connection } = fakeTransport(); + const session = createMeshSession(transport); + await session.connect("ws://node", ["core/data"]); + connection.fail(new Error("node closed abruptly")); + // Events: connecting, connected, closed + const event = (await nthEvent(session, EVENTS_THROUGH_FAILURE)) as { + state: { status: string; reason: string }; + }; + expect(event.state.status).toBe("closed"); + expect(event.state.reason).toBe("node closed abruptly"); + }); + + it("refuses a second connect on the same session", async () => { + const { transport } = fakeTransport(); + const session = createMeshSession(transport); + await session.connect("ws://node", ["core/data"]); + await expect(session.connect("ws://node", ["core/data"])).rejects.toThrow( + "connects once", + ); + await session.close(); + }); + + it("refuses a ping while not connected", async () => { + const { transport } = fakeTransport(); + const session = createMeshSession(transport); + await expect(session.sendPing()).rejects.toThrow("not connected"); + }); +}); diff --git a/ts/packages/web-console/test/websocket-transport.test.ts b/ts/packages/web-console/test/websocket-transport.test.ts new file mode 100644 index 0000000..d98f45f --- /dev/null +++ b/ts/packages/web-console/test/websocket-transport.test.ts @@ -0,0 +1,157 @@ +import { describe, expect, it } from "vitest"; +import { encode } from "cbor2"; +import type { Frame } from "@exadev/wire-mesh-core/generated/protocol"; +import { + createBrowserTransport, + messageFromFrame, + wrapWebSocket, +} from "../src/adapters/websocket-transport.js"; +import { FakeWebSocket } from "./fake-websocket.js"; +import { bytesFromHex } from "./hex.js"; + +const SHA256_BYTE_LENGTH = 32; + +const ping: Frame = { type: "ping" }; +const CBOR_MAP_ONE_ENTRY_FIRST_BYTE = 0xa1; // a one-entry CBOR map head -- the ping frame, no length prefix +const CBOR_BREAK_BYTE = 0xff; // the CBOR break byte on its own: undecodable as a complete value +const CLOSE_PROTOCOL_ERROR = 1002; + +function arrayBuffer(bytes: Uint8Array): ArrayBuffer { + return bytes.buffer.slice( + bytes.byteOffset, + bytes.byteOffset + bytes.byteLength, + ) as ArrayBuffer; +} + +async function collect(iterable: Readonly>): Promise { + const out: T[] = []; + for await (const item of iterable) { + out.push(item); + } + return out; +} + +describe("wrapWebSocket", () => { + it("delivers a valid CBOR frame message and encodes sends as single CBOR messages", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + + const received = connection.receive()[Symbol.asyncIterator](); + const nextFrame = received.next(); + ws.emitMessage(arrayBuffer(messageFromFrame(ping))); + + expect((await nextFrame).value).toEqual(ping); + + await connection.send(ping); + expect(ws.sent.length).toBe(1); + const decoded = new Uint8Array(ws.sent[0] ?? new ArrayBuffer(0)); + expect(decoded[0]).toBe(CBOR_MAP_ONE_ENTRY_FIRST_BYTE); + }); + + it("rejects the receive iteration on bytes that do not decode as CBOR, and closes the socket", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + + const nextFrame = connection.receive()[Symbol.asyncIterator]().next(); + ws.emitMessage(arrayBuffer(Uint8Array.from([CBOR_BREAK_BYTE]))); + + await expect(nextFrame).rejects.toThrow(); + expect(ws.closed).toBe(true); + expect(ws.closeCode).toBe(CLOSE_PROTOCOL_ERROR); + }); + + it("rejects the receive iteration on a text WebSocket message", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + + const nextFrame = connection.receive()[Symbol.asyncIterator]().next(); + ws.emitText("hello"); + + await expect(nextFrame).rejects.toThrow("expected a binary"); + expect(ws.closed).toBe(true); + }); + + it("drops a decodable but schema-invalid frame without ending the connection", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + + const received = connection.receive()[Symbol.asyncIterator](); + const first = received.next(); + // A well-formed CBOR map with the wrong literal: not any known frame. It is dropped, so the next valid frame still arrives and nothing closes. + ws.emitMessage(arrayBuffer(encode({ type: "not-a-real-frame-kind" }))); + ws.emitMessage(arrayBuffer(messageFromFrame(ping))); + + expect((await first).value).toEqual(ping); + expect(ws.closed).toBe(false); + + const after = collect(connection.receive()); + ws.close(); + expect(await after).toEqual([]); + }); + + it("rejects sends after the connection has ended", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + ws.close(); + + await expect(connection.send(ping)).rejects.toThrow("connection is closed"); + }); + + it("ends the iteration cleanly when the socket closes", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + const frames = collect(connection.receive()); + ws.close(); + expect(await frames).toEqual([]); + }); + + it("ends the iteration when the socket errors after delivering pending frames", async () => { + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + const frames = collect(connection.receive()); + ws.emitMessage(arrayBuffer(messageFromFrame(ping))); + ws.emitError(); + expect(await frames).toEqual([ping]); + }); +}); + +describe("browser transport over real CBOR bytes", () => { + it("messageFromFrame produces bytes the adapter itself decodes back", async () => { + const gossip: Frame = { + type: "gossip", + peers: [ + { + device: bytesFromHex("11".repeat(SHA256_BYTE_LENGTH)), + addresses: ["203.0.113.5:4433"], + "snapshot-seconds": 1861833600, + }, + ], + }; + const ws = new FakeWebSocket("ws://node"); + const connection = wrapWebSocket(ws as unknown as WebSocket); + const received = connection.receive()[Symbol.asyncIterator](); + const nextFrame = received.next(); + ws.emitMessage(arrayBuffer(messageFromFrame(gossip))); + expect((await nextFrame).value).toEqual(gossip); + }); +}); + +describe("createBrowserTransport", () => { + it("rejects a non-WebSocket address without constructing anything", async () => { + const transport = createBrowserTransport(); + await expect(transport.connect("http://localhost:8787")).rejects.toThrow( + "not a WebSocket address", + ); + }); + + it("rejects listen outright: the console is a client only", async () => { + const transport = createBrowserTransport(); + let accepted = 0; + await expect( + transport.listen("ws://localhost:8787", () => { + accepted += 1; + }), + ).rejects.toThrow("cannot listen"); + expect(accepted).toBe(0); + }); +}); diff --git a/ts/packages/web-console/tsconfig.json b/ts/packages/web-console/tsconfig.json new file mode 100644 index 0000000..2e6f4c0 --- /dev/null +++ b/ts/packages/web-console/tsconfig.json @@ -0,0 +1,17 @@ +{ + "compilerOptions": { + "target": "ES2024", + "module": "nodenext", + "moduleResolution": "nodenext", + "lib": ["ES2024", "DOM", "DOM.Iterable"], + "strict": true, + "noUncheckedIndexedAccess": true, + "exactOptionalPropertyTypes": true, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": true, + "noEmit": true, + "skipLibCheck": true, + "types": ["vite/client"] + }, + "include": ["src"] +} diff --git a/ts/packages/web-console/tsconfig.node.json b/ts/packages/web-console/tsconfig.node.json new file mode 100644 index 0000000..661e1b4 --- /dev/null +++ b/ts/packages/web-console/tsconfig.node.json @@ -0,0 +1,8 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "allowImportingTsExtensions": true, + "types": ["node", "vite/client"] + }, + "include": ["vite.config.ts", "eslint.config.ts", "test/**/*.ts"] +} diff --git a/ts/packages/web-console/turbo.json b/ts/packages/web-console/turbo.json new file mode 100644 index 0000000..45449dd --- /dev/null +++ b/ts/packages/web-console/turbo.json @@ -0,0 +1,24 @@ +{ + "$schema": "https://turborepo.com/schema.json", + "extends": ["//"], + + "tasks": { + "_build": { + // vite build bundles the browser entry (src/main.ts per index.html), resolving the workspace's core package and every npm dependency into static assets under dist/ -- the package's deployable output, servable from any static origin including the hub's. + "inputs": ["src/**", "index.html", "vite.config.ts", "tsconfig.json"], + "outputs": ["dist/**"] + }, + "_test": { + "dependsOn": ["^_build"] + }, + "_typecheck": { + "dependsOn": ["^_build"], + "inputs": ["**/*.ts", "tsconfig.json", "tsconfig.node.json"] + }, + "_lint": { + "dependsOn": ["^_build"], + "inputs": ["$TURBO_DEFAULT$", "eslint.config.ts"], + "outputs": [".eslintcache"] + } + } +} diff --git a/ts/packages/web-console/vite.config.ts b/ts/packages/web-console/vite.config.ts new file mode 100644 index 0000000..446f551 --- /dev/null +++ b/ts/packages/web-console/vite.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from "vite"; + +// A plain browser app: no framework plugin, no dev proxy -- the console takes full ws:// URLs and WebSocket connections are not same-origin-restricted, so `vite` (dev) serves the page and the page dials the node directly. +export default defineConfig({ + build: { + outDir: "dist", + }, +}); diff --git a/ts/pnpm-lock.yaml b/ts/pnpm-lock.yaml index 6f8e576..7928708 100644 --- a/ts/pnpm-lock.yaml +++ b/ts/pnpm-lock.yaml @@ -113,6 +113,52 @@ importers: specifier: 5.0.0 version: 5.0.0(@types/node@26.4.1)(vite@8.2.2(@types/node@26.4.1)(jiti@2.7.0)) + packages/web-console: + dependencies: + '@exadev/wire-mesh-core': + specifier: workspace:* + version: link:../core + cbor2: + specifier: 2.3.0 + version: 2.3.0 + devDependencies: + '@exadev/eslint-config': + specifier: 2.10.6 + version: 2.10.6(eslint@10.10.0(jiti@2.7.0))(typescript-eslint@8.69.0(eslint@10.10.0(jiti@2.7.0))(typescript@6.0.3))(typescript@6.0.3) + '@types/node': + specifier: 26.4.1 + version: 26.4.1 + eslint: + specifier: 10.10.0 + version: 10.10.0(jiti@2.7.0) + eslint-config-prettier: + specifier: 10.1.8 + version: 10.1.8(eslint@10.10.0(jiti@2.7.0)) + eslint-plugin-prettier: + specifier: 5.5.6 + version: 5.5.6(eslint-config-prettier@10.1.8(eslint@10.10.0(jiti@2.7.0)))(eslint@10.10.0(jiti@2.7.0))(prettier@3.9.6) + globals: + specifier: 17.12.0 + version: 17.12.0 + jiti: + specifier: 2.7.0 + version: 2.7.0 + prettier: + specifier: 3.9.6 + version: 3.9.6 + turbo: + specifier: 2.10.12 + version: 2.10.12 + typescript: + specifier: 6.0.3 + version: 6.0.3 + vite: + specifier: 8.2.2 + version: 8.2.2(@types/node@26.4.1)(jiti@2.7.0) + vitest: + specifier: 5.0.0 + version: 5.0.0(@types/node@26.4.1)(vite@8.2.2(@types/node@26.4.1)(jiti@2.7.0)) + packages: '@andrewbranch/untar.js@1.0.4':