import { describe, expect, it } from "bun:test"; import { Elysia } from "elysia"; import { FormatRegistry } from "@sinclair/typebox"; import { Value } from "@sinclair/typebox/value"; import { AppTelemetryMessageSchema, type AppPrinterState } from "@continuum/ts-types"; import { UnauthorizedError } from "@continuum/ts-core"; import { maySee, publish, subscribe, subscriberCount } from "../src/telemetry/broadcast"; import { tokenFromSubprotocol } from "../src/auth/ws_subprotocol"; /** * TypeBox knows no string formats until something registers them. Elysia * does that for itself at runtime, which is why `format: "uuid"` works on * the HTTP routes - but a bare `Value.Check` here does not, and an * unregistered format fails every value rather than being ignored. So the * schema keeps its format and the test learns the format, instead of the * schema being weakened to suit the test. */ FormatRegistry.Set("uuid", (value) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value), ); const printer = (over: Partial = {}): AppPrinterState => ({ id: "3f2b1c5e-0000-4000-8000-000000000001", farmId: "3f2b1c5e-0000-4000-8000-0000000000aa", name: "H2C", vendor: "bambu", status: "idle", vendorState: "FINISH", vendorErrorCode: 0, nozzleTempC: 30, nozzleTargetTempC: 0, bedTempC: 23, bedTargetTempC: 0, progressPercent: 100, currentLayer: 134, totalLayers: 134, remainingMinutes: 0, currentJobName: "a-real-job", activeAlerts: 6, cameraStream: "h2c", lastSeenAt: "2026-10-18T13:40:00.000Z", ...over, }); /** * The function that decides whether one organisation's operator sees * another's machines. It lives in its own module precisely so this test does * not need a socket or a database. */ describe("maySee", () => { const mine = "3f2b1c5e-0000-4000-8000-0000000000aa"; const theirs = "3f2b1c5e-0000-4000-8000-0000000000bb"; it("sends a printer in a farm the viewer can see", () => { expect(maySee(new Set([mine]), printer({ farmId: mine }))).toBe(true); }); it("does not send a printer from another organisation's farm", () => { expect(maySee(new Set([mine]), printer({ farmId: theirs }))).toBe(false); }); it("sends nothing to a viewer who can see no farms", () => { expect(maySee(new Set(), printer())).toBe(false); }); /** * `farm_id` is not nullable, so there is no such row today. Asserted * anyway: this is the one function where being wrong leaks another * organisation's machines, so it should not depend on a schema constraint * staying as it is. */ it("refuses a printer with no farm rather than matching it against anything", () => { expect(maySee(new Set([mine]), printer({ farmId: undefined as never }))).toBe(false); expect(maySee(new Set([""]), printer({ farmId: "" as never }))).toBe(false); }); }); describe("the broadcast fan-out", () => { it("delivers to every subscriber and stops on unsubscribe", () => { const seen: string[] = []; const off1 = subscribe((p) => seen.push(`a:${p.name}`)); const off2 = subscribe((p) => seen.push(`b:${p.name}`)); publish(printer({ name: "first" })); expect(seen).toEqual(["a:first", "b:first"]); off1(); publish(printer({ name: "second" })); expect(seen).toEqual(["a:first", "b:first", "b:second"]); off2(); publish(printer({ name: "third" })); expect(seen).toEqual(["a:first", "b:first", "b:second"]); expect(subscriberCount()).toBe(0); }); /** * A socket closing mid-iteration makes its `send` throw. If that * propagated, it would come out of `persistTelemetry` and abort a whole * farm's telemetry batch - one dead browser tab taking out the fleet. */ it("keeps delivering after a subscriber throws, and does not propagate", () => { const seen: string[] = []; const off1 = subscribe(() => { throw new Error("socket already closed"); }); const off2 = subscribe((p) => seen.push(p.name)); expect(() => publish(printer({ name: "survivor" }))).not.toThrow(); expect(seen).toEqual(["survivor"]); off1(); off2(); }); }); /** * A browser cannot set headers on a WebSocket, so the token rides in the * subprotocol field rather than a query parameter - which would put a bearer * credential into access and proxy logs. */ describe("tokenFromSubprotocol", () => { it("takes the token from a well-formed offer", () => { expect(tokenFromSubprotocol("bearer, tok-abc")).toBe("tok-abc"); // Whatever whitespace the runtime joins the list with. expect(tokenFromSubprotocol("bearer,tok-abc")).toBe("tok-abc"); }); it("refuses anything else", () => { for (const header of [ undefined, null, "", "bearer", "tok-abc", // Reversed: the scheme must be first, because the runtime echoes the // first offered value back and the token must not be what comes back. "tok-abc, bearer", "basic, tok-abc", "bearer, tok-abc, extra", ]) { expect(() => tokenFromSubprotocol(header)).toThrow(UnauthorizedError); } }); }); /** * That what the backend actually pushes matches the published contract. * * The payload is built from `printerProjection`, and the schema lives in * continuum-schemas - two places that can drift. This is the test that * notices. It is also the reason the frontend can be written against the * schema rather than against a guess, which is how the last version of that * store ended up consuming fields no backend ever sent. */ describe("the pushed messages match continuum/v1/app", () => { function check(value: unknown) { const ok = Value.Check(AppTelemetryMessageSchema, value); if (!ok) { const errors = [...Value.Errors(AppTelemetryMessageSchema, value)].map( (e) => `${e.path || "/"}: ${e.message} (got ${JSON.stringify(e.value)})`, ); throw new Error(`validation failed:\n ${errors.join("\n ")}`); } } it("accepts a full telemetry push", () => { check({ type: "printer.telemetry", printer: printer() }); }); /** * The state the schema's nullability exists for: a claimed printer whose * gateway has never reported. Every reading is null, and a client that * renders `?? 0` would draw this as a cold printer rather than an unknown * one. */ it("accepts a printer that has never reported anything", () => { check({ type: "printer.telemetry", printer: { id: "3f2b1c5e-0000-4000-8000-000000000002", farmId: "3f2b1c5e-0000-4000-8000-0000000000aa", name: "never seen", vendor: "bambu", status: null, vendorState: null, nozzleTempC: null, bedTempC: null, progressPercent: null, cameraStream: null, lastSeenAt: null, }, }); }); it("accepts the ready message", () => { check({ type: "ready", visiblePrinters: 0 }); check({ type: "ready", visiblePrinters: 12 }); }); it("rejects the shape the old frontend invented", () => { // Every field of this was made up against nothing: a `data` envelope, // `printerId`, `nozzleTempC` at the top level, `progressPct`. Asserted // so nobody reintroduces it by writing the client first. expect( Value.Check(AppTelemetryMessageSchema, { type: "printer.telemetry", data: { printerId: "p1p", nozzleTempC: 30, progressPct: 50 }, }), ).toBe(false); }); it("rejects a push with no printer and an unknown message type", () => { expect(Value.Check(AppTelemetryMessageSchema, { type: "printer.telemetry" })).toBe(false); expect(Value.Check(AppTelemetryMessageSchema, { type: "printer.update", printer: printer() })).toBe(false); }); }); /** * That the subprotocol check actually gates the upgrade, so an * unauthenticated browser never gets a socket. * * Mirrors the real route's composition rather than importing it, because * `routes/ws/telemetry.ts` pulls in `../../db` and so a live Postgres - the * same constraint as `tests/edge_upgrade.test.ts`. What this pins is the * framework behaviour the design rests on: that a throwing `resolve` fails * the HTTP upgrade, and that the runtime echoes back the *first* offered * subprotocol, which is why "bearer" is first and the token second. If it * echoed the token instead, the credential would come back in a response * header. `scripts/e2e-edge.sh` drives the real route. */ describe("the subprotocol gates the websocket upgrade", () => { const app = new Elysia() .resolve({ as: "scoped" }, ({ headers }): { viewer: string } => { const token = tokenFromSubprotocol(headers["sec-websocket-protocol"]); if (token !== "good-token") throw new UnauthorizedError("Invalid or expired access token"); return { viewer: "user-1" }; }) .ws("/ws/telemetry", { open(ws) { ws.send(JSON.stringify({ viewer: (ws.data as { viewer: string }).viewer })); }, }) .listen(0); const base = `localhost:${app.server!.port}`; function attempt(protocols?: string[]): Promise<{ result: string; negotiated?: string }> { return new Promise((resolve) => { const ws = protocols ? new WebSocket(`ws://${base}/ws/telemetry`, protocols) : new WebSocket(`ws://${base}/ws/telemetry`); const done = (result: string) => { const negotiated = ws.protocol; try { ws.close(); } catch { // Already closed by the server. } resolve({ result, negotiated }); }; ws.onmessage = (event) => done(`opened:${event.data}`); ws.onerror = () => done("refused"); ws.onclose = () => done("refused"); setTimeout(() => done("timed out"), 5_000); }); } it("opens for a valid token and echoes back the scheme, not the token", async () => { const { result, negotiated } = await attempt(["bearer", "good-token"]); expect(result).toBe('opened:{"viewer":"user-1"}'); // The credential must not be what comes back in the response header. expect(negotiated).toBe("bearer"); }); it("never opens a socket without a valid token", async () => { expect((await attempt()).result).toBe("refused"); expect((await attempt(["bearer", "forged"])).result).toBe("refused"); expect((await attempt(["bearer"])).result).toBe("refused"); // Scheme second: rejected, because the first value is what gets echoed. expect((await attempt(["good-token", "bearer"])).result).toBe("refused"); }); });