Add the operator console's live telemetry socket

The only socket until now was the gateway-facing /ws/edge/v1, which
authenticates as a *farm* with a credential any of its gateways holds,
and which accepts telemetry. Pointing a browser at it would mean handing
the browser something that can drive printers. /ws/telemetry
authenticates as a *user* with a Supabase access token, and accepts
nothing at all: it is push-only, and a frame arriving from the browser is
ignored. Everything that changes a printer keeps going over HTTP, where
the role checks and the audit trail already are.

Auth is checked at the upgrade, as on the edge socket, so an
unauthenticated browser never gets a socket. The token rides in
`Sec-WebSocket-Protocol` - `new WebSocket(url, ["bearer", token])` -
because that field is the whole of what JavaScript controls about a
browser handshake, and the usual workaround of `?token=` would put a
bearer credential into every access and proxy log. "bearer" is first
because the runtime echoes the *first* offered value back in the
response header, so the token must not be in that position; asserted,
not assumed.

The payload is the printer resource itself, returned straight out of
`persistTelemetry`'s UPDATE through `printerProjection` - the same
projection the REST routes use. So a push is byte-for-byte what
`GET /printers/:id` would have answered, and a client applies it to the
row it already holds. Rebuilding it from the gateway's snapshot instead
would be a second translation of the same data, free to drift, and
missing the columns a gateway never reports at all like `name` and
`vendor`.

`maySee` is what keeps one organisation's telemetry out of another's
console, and it lives in `telemetry/broadcast.ts` with no imports but a
type so it can be tested without a socket or a database - the same reason
as `dispatch/job_status.ts`, `stream.ts` and `auth/edge_upgrade.ts`. It
refuses a printer with no farm rather than matching it against anything,
which no row can be today, because this is the one function where being
wrong leaks another org's machines.

**A latent bug this turned up:** the edge route's resolve was
`{ as: "scoped" }`, copied from `requireUser`. For `requireUser` that is
right - it is used inside each route file and must propagate `user` to
that file's routes. On a ws plugin it propagates the check to the parent
app, where it gates every route mounted *after* that plugin. Measured
with a two-plugin probe: the second plugin's own socket is refused
because the first one's check runs first and finds no edge token. Nothing
shipped broken only because /ws/telemetry is the first route mounted
after it - and it was refused outright until this was found. Both ws
resolves now use the default `local` scope.

Known limits, both written down rather than discovered later:

- **In-process fan-out.** A `Set` of callbacks in one process, so two
  backend instances would each push only the telemetry their own
  gateways are connected to and an operator would see half a fleet. The
  fix is Valkey pub/sub, already a dependency; doing it now would mean
  testing a message broker to deliver events between two objects in the
  same process.
- **Visibility is resolved once, when the socket opens.** Re-deriving it
  per message would put a database round trip on every push to every
  viewer. The direction that matters: someone removed from an
  organisation keeps receiving its telemetry until their socket drops.
  Closing that needs a signal on membership change that this backend does
  not have; a periodic re-resolve would narrow the window without closing
  it. Telemetry is also the least sensitive thing here, and every HTTP
  route re-checks on each request.

Verified against the real stack, not just mirrored in unit tests:
`scripts/e2e-edge.sh` now drives the actual route with a real
JWKS-verified token and asserts the handshake echoes "bearer", that a
`ready` frame arrives, and that a real push carries the printer resource
with live readings - "2 push(es); last was H2C state=FINISH nozzle=30
camera=h2c". A bogus token is refused before any socket exists.
`scripts/watch-telemetry.ts` is the client it uses, and is the quickest
way to see by hand whether a farm's telemetry reaches a console.

110 tests (was 95), `bunx tsc --noEmit` clean.

scripts/e2e-edge.sh

@@ -335,6 +335,50 @@ if [ -n "$NO_CAMERA" ]; then
   ok "a printer with no camera is refused with 404, blaming the right half"
 fi
 
+# --- 5c. the operator console's live feed ----------------------------------
+
+# The user-facing socket, driven for real. The unit tests can only mirror its
+# composition, because the route imports ../../db; this is the only thing that
+# proves it is wired into the app at all, that the Supabase token verifies on
+# the upgrade, and that a gateway's telemetry actually reaches a console.
+echo "==> /ws/telemetry: connecting as the operator and waiting for a push"
+WS_OUT="$WORK/telemetry.json"
+if bun scripts/watch-telemetry.ts "ws://localhost:$PORT" "$USER_TOKEN" 12 > "$WS_OUT" 2>"$WORK/ws.err"; then
+  python3 - "$WS_OUT" <<'PYCHK' || fail "the telemetry socket did not behave: $(cat "$WS_OUT" "$WORK/ws.err")"
+import json, sys
+d = json.load(open(sys.argv[1]))
+# "bearer" and not the token: the runtime echoes the first offered subprotocol
+# back in the response header, so the credential must not be in that position.
+assert d["negotiated"] == "bearer", f'echoed subprotocol was {d["negotiated"]!r}, expected "bearer"'
+kinds = [f.get("type") for f in d["frames"]]
+assert "ready" in kinds, f"no ready frame; got {kinds}"
+ready = next(f for f in d["frames"] if f.get("type") == "ready")
+assert ready["visiblePrinters"] >= 1, f"ready says {ready['visiblePrinters']} visible printers"
+pushes = [f for f in d["frames"] if f.get("type") == "printer.telemetry"]
+assert pushes, f"no printer.telemetry push arrived; got {kinds}"
+# A push must be the printer resource, not a telemetry-shaped subset, and must
+# carry real readings rather than a touched row of nulls.
+p = pushes[-1]["printer"]
+for key in ("id", "farmId", "name", "vendor"):
+    assert key in p, f"push is missing {key}: {p}"
+assert p.get("vendorState"), f"push carries no vendor state: {p}"
+print(f'  ok: {len(pushes)} push(es); last was {p["name"]} '
+      f'state={p["vendorState"]} nozzle={p.get("nozzleTempC")} camera={p.get("cameraStream")}')
+PYCHK
+else
+  fail "could not connect to /ws/telemetry: $(cat "$WORK/ws.err")"
+fi
+ok "the console socket authenticated and received live telemetry"
+
+# A socket with no valid credential must not open at all - the whole point of
+# checking at the upgrade rather than in a first message.
+echo "==> an unauthenticated console socket must be refused"
+if bun scripts/watch-telemetry.ts "ws://localhost:$PORT" "not-a-real-token" 5 \
+     > "$WORK/ws-bad.json" 2>&1; then
+  fail "a socket opened with a bogus token: $(cat "$WORK/ws-bad.json")"
+fi
+ok "a bogus token is refused before any socket exists"
+
 # --- 6. the negative case --------------------------------------------------
 
 echo "==> an unclaimed printer must be refused, not guessed at"

scripts/watch-telemetry.ts

@@ -0,0 +1,38 @@
+/**
+ * Connects to /ws/telemetry as a user and prints what arrives, then exits.
+ *
+ * Exists for scripts/e2e-edge.sh, which can then assert on real frames from
+ * the real route rather than on a mirrored composition in a unit test. Also
+ * useful by hand: it is the quickest way to see whether a farm's telemetry is
+ * actually reaching a console.
+ *
+ * Usage: bun scripts/watch-telemetry.ts <ws-base> <access-token> [seconds]
+ */
+const [base, token, secondsArg] = process.argv.slice(2);
+if (!base || !token) {
+  console.error("usage: watch-telemetry.ts <ws-base> <access-token> [seconds]");
+  process.exit(2);
+}
+const seconds = Number(secondsArg ?? 15);
+
+// The token goes in the subprotocol, not the URL - a browser cannot set
+// headers, and a query parameter would put a bearer credential in every
+// access log. "bearer" first: the server echoes the first offered value back,
+// and the token must not be what comes back.
+const ws = new WebSocket(`${base}/ws/telemetry`, ["bearer", token]);
+
+const frames: unknown[] = [];
+ws.onmessage = (event) => {
+  frames.push(typeof event.data === "string" ? JSON.parse(event.data) : event.data);
+};
+ws.onerror = () => {
+  console.error("REFUSED: the socket would not open");
+  process.exit(1);
+};
+
+setTimeout(() => {
+  // Printed as one JSON document so the caller can parse it rather than
+  // scraping log lines.
+  console.log(JSON.stringify({ negotiated: ws.protocol, frames }, null, 2));
+  process.exit(0);
+}, seconds * 1000);

src/auth/ws_subprotocol.ts

@@ -0,0 +1,50 @@
+import { UnauthorizedError } from "@continuum/ts-core";
+
+/**
+ * Pulls a bearer token out of a WebSocket's `Sec-WebSocket-Protocol` header.
+ *
+ * A leaf module with no imports but the error type, like
+ * `auth/edge_upgrade.ts` next to it, and for the same reason: this is an
+ * authentication boundary and should not need a database to test.
+ *
+ * ## Why the subprotocol and not a query parameter
+ *
+ * A browser cannot set headers on a WebSocket - `new WebSocket(url, protos)`
+ * is the whole of what JavaScript controls about the handshake. So the usual
+ * `Authorization: Bearer` is unavailable, and the common workaround is
+ * `?token=...`, which puts a bearer credential in a URL and therefore into
+ * access logs, proxy logs and error breadcrumbs. This codebase already goes
+ * out of its way to avoid that: continuum-proxy hand-writes a `Debug` so a
+ * presigned URL prints as `<redacted>`.
+ *
+ * The subprotocol field is a header, is not logged by default, and - unlike
+ * a message-based handshake - is available *before* the upgrade, so an
+ * unauthenticated peer never gets a socket at all. The gateway socket gets
+ * the same property through a real `Authorization` header, which it can use
+ * because it is not a browser.
+ *
+ * The client sends `["bearer", "<token>"]`. "bearer" goes first because the
+ * server must echo exactly one offered value back or the browser fails the
+ * connection, and the runtime echoes the first - so the token itself must
+ * not be in that position, or it would come back in the response header.
+ */
+const SCHEME = "bearer";
+
+export function tokenFromSubprotocol(header: string | null | undefined): string {
+  const offered = (header ?? "")
+    .split(",")
+    .map((part) => part.trim())
+    .filter((part) => part.length > 0);
+
+  // Exactly two: the scheme and the token. Being strict here rather than
+  // scanning for "the entry that looks like a JWT" means a client that gets
+  // the order or the count wrong is told so, instead of working by accident
+  // until the day something else is in the list.
+  if (offered.length !== 2 || offered[0] !== SCHEME) {
+    throw new UnauthorizedError(
+      `WebSocket subprotocol must be ["${SCHEME}", "<access token>"]`,
+    );
+  }
+
+  return offered[1]!;
+}

src/index.ts

@@ -11,6 +11,7 @@ import { farmRoutes } from "./routes/farms";
 import { printerRoutes } from "./routes/printers";
 import { jobRoutes } from "./routes/jobs";
 import { edgeWsRoute } from "./routes/ws/edge";
+import { telemetryWsRoute } from "./routes/ws/telemetry";
 import { startDispatcher } from "./dispatch";
 
 const app = new Elysia()
@@ -62,6 +63,7 @@ const app = new Elysia()
   .use(printerRoutes)
   .use(jobRoutes)
   .use(edgeWsRoute)
+  .use(telemetryWsRoute)
   .listen(Number(process.env.PORT ?? 3000));
 
 console.log(`continuum-backend listening on :${app.server?.port}`);

src/routes/ws/edge.ts

@@ -13,6 +13,8 @@ import { advanceJobsFrom } from "../../dispatch";
 import { enqueueFrameIngestionBatch } from "../../queue";
 import { edgeJwtSecret } from "../../auth/edge_secret";
 import { farmFromEdgeUpgrade } from "../../auth/edge_upgrade";
+import { publish } from "../../telemetry/broadcast";
+import { printerProjection } from "../../db/projections";
 
 /**
  * Edge-gateway uplink. A continuum-proxy gateway connects here with a
@@ -259,7 +261,7 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) {
 
     const status = mapStatus(snapshot.state, snapshot.print_error);
 
-    await db
+    const [updated] = await db
       .update(printers)
       .set({
         // `?? null` rather than leaving the key out: a field the gateway
@@ -285,7 +287,27 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) {
         lastSeenAt: new Date(),
         updatedAt: new Date(),
       })
-      .where(eq(printers.id, printerId));
+      .where(eq(printers.id, printerId))
+      // Returned through the same projection the REST routes use, so what a
+      // watching console is pushed is byte-for-byte what it would have got
+      // from `GET /printers/:id`. Deriving it from `snapshot` instead would
+      // be a second translation of the same data, free to drift - and it
+      // would be missing the columns the gateway does not report at all,
+      // like `name` and `vendor`.
+      .returning(printerProjection.returning);
+
+    // Push to anyone watching. After the write, so a console cannot be told
+    // about a state that failed to persist - and `publish` swallows a
+    // subscriber's error, so one broken browser tab cannot abort a whole
+    // farm's telemetry batch here.
+    if (updated) {
+      publish({
+        ...updated,
+        // Serialised, because this goes to JSON and a Date would otherwise
+        // be stringified inconsistently with the REST response.
+        lastSeenAt: updated.lastSeenAt?.toISOString() ?? null,
+      });
+    }
 
     // The printer's row is the latest snapshot; the job's row is the history
     // that outlives it. This is the only thing that moves a job out of
@@ -370,11 +392,20 @@ export const edgeWsRoute = new Elysia()
    *
    * Throwing here fails the upgrade with an HTTP status, so a peer without a
    * valid token never reaches `open` - which is the whole point of
-   * authenticating at the handshake rather than in the first frame. Modelled
+   * authenticating at the handshake rather than in the first frame.
-   * on `requireUser`'s scoped resolve, for the same reason: one verification
+   *
-   * per connection, in one place, ahead of every handler.
+   * The default `local` scope, NOT `scoped` like `requireUser`, and the
+   * difference is load-bearing. `requireUser` is `.use()`d inside each route
+   * file and has to propagate `user` to that file's routes. This resolve
+   * belongs to the one socket defined just below it, and `scoped` here
+   * propagates it to the parent app, where it then gates every route
+   * registered *after* this plugin. Measured: with `scoped`, a second ws
+   * plugin's own route is refused, because this check runs first and finds
+   * no edge token. Nothing shipped broken only because /ws/telemetry was the
+   * first route mounted after this one - and it was refused outright until
+   * this was changed.
    */
-  .resolve({ as: "scoped" }, async ({ headers, edgeJwt }): Promise<{ edgeFarmId: string }> => ({
+  .resolve(async ({ headers, edgeJwt }): Promise<{ edgeFarmId: string }> => ({
     edgeFarmId: await farmFromEdgeUpgrade(headers.authorization, (token) => edgeJwt.verify(token)),
   }))
   .ws("/ws/edge/v1", {

src/routes/ws/telemetry.ts

@@ -0,0 +1,149 @@
+import { Elysia } from "elysia";
+import { JwtVerificationError, UnauthorizedError } from "@continuum/ts-core";
+import type { AppPrinterState, AppTelemetryMessage } from "@continuum/ts-types";
+import { db } from "../../db";
+import { farms, printers } from "../../db/schema";
+import { visibleFarms, visiblePrinters } from "../../auth";
+import { jwtVerifier } from "../../auth/verifier";
+import { tokenFromSubprotocol } from "../../auth/ws_subprotocol";
+import { maySee, subscribe } from "../../telemetry/broadcast";
+
+/**
+ * The operator console's live fleet feed.
+ *
+ * Read-only and push-only: the browser sends nothing after the handshake,
+ * and a frame arriving from it is ignored. Everything that changes a printer
+ * goes over HTTP, where the role checks and the audit trail already live -
+ * `POST /printers/:id/commands`. A socket that can also issue commands is a
+ * second authorization surface to get right for no gain.
+ *
+ * ## Not /ws/edge/v1
+ *
+ * That socket authenticates as a *farm*, with an edge JWT that any gateway
+ * for that farm holds, and it accepts telemetry. This one authenticates as a
+ * *user*, with a Supabase access token, and accepts nothing. Pointing a
+ * browser at the gateway socket would mean handing the browser a credential
+ * that can drive printers.
+ *
+ * ## The wire format
+ *
+ * `continuum/v1/app/telemetry_message.schema.json`, mirrored in
+ * `@continuum/ts-types` as `AppTelemetryMessageSchema`. camelCase, matching
+ * this backend's HTTP routes rather than the gateway protocol - see that
+ * file. The payload is the printer resource itself, so a client applies a
+ * push onto the row it already fetched from `GET /printers`.
+ */
+
+/**
+ * The farms a user can see, resolved once when their socket opens.
+ *
+ * Once, not per message: a farm's telemetry arrives every few seconds per
+ * printer, and re-deriving visibility each time would put a database round
+ * trip on every push to every viewer. The cost is staleness in one
+ * direction - someone added to an organisation mid-connection does not see
+ * its printers until they reconnect, which a page reload does.
+ *
+ * Staleness in the *other* direction is the one that would matter, and it is
+ * bounded by the same thing: someone removed from an organisation keeps
+ * receiving its telemetry until their socket drops. That is a real gap worth
+ * knowing about. Closing it properly means invalidating open sockets on a
+ * membership change, which needs a signal this backend does not have yet; a
+ * periodic re-resolve would narrow the window but not close it, so it is
+ * written down rather than half-done. Telemetry is also the least sensitive
+ * thing in the system - nozzle temperatures, not credentials - and a removed
+ * member can be cut off immediately by anything that matters, because every
+ * HTTP route re-checks on each request.
+ */
+async function visibleFarmIds(userId: string): Promise<Set<string>> {
+  const rows = await db.select({ id: farms.id }).from(farms).where(visibleFarms(userId));
+  return new Set(rows.map((row) => row.id));
+}
+
+/**
+ * How many printers this user can see, for the `ready` message.
+ *
+ * Counting farms would be the cheaper query and the wrong answer: a farm
+ * with no printers claimed yet is exactly the first-run state this message
+ * exists to make legible.
+ */
+async function countVisiblePrinters(userId: string): Promise<number> {
+  const rows = await db.select({ id: printers.id }).from(printers).where(visiblePrinters(userId));
+  return rows.length;
+}
+
+/** One unsubscribe per open socket, removed when it closes. */
+const unsubscribers = new Map<string | number, () => void>();
+
+export const telemetryWsRoute = new Elysia()
+  /**
+   * Verifies the user's access token on the upgrade, before a socket exists.
+   *
+   * Throwing fails the upgrade with an HTTP status, so an unauthenticated
+   * browser never reaches `open`. Same shape as the edge socket's resolve
+   * next door; the difference is only where the token comes from and which
+   * verifier checks it.
+   *
+   * The default `local` scope deliberately - see the note on the edge
+   * route's resolve. `scoped` would push this check onto every route
+   * mounted after this plugin, which would then demand a WebSocket
+   * subprotocol from plain HTTP requests.
+   */
+  .resolve(async ({ headers }): Promise<{ viewer: { id: string } }> => {
+    const token = tokenFromSubprotocol(headers["sec-websocket-protocol"]);
+
+    try {
+      const claims = await jwtVerifier.verify(token);
+      if (!claims.sub) throw new UnauthorizedError("Access token has no subject claim");
+      return { viewer: { id: claims.sub } };
+    } catch (err) {
+      // Never pass the reason through - "expired" versus "bad signature"
+      // tells a forger which part to fix next. Logged instead, exactly as
+      // `requireUser` does it.
+      if (err instanceof JwtVerificationError) {
+        console.warn(
+          `[telemetry] token rejected: ${err.cause instanceof Error ? err.cause.message : err.message}`,
+        );
+        throw new UnauthorizedError("Invalid or expired access token");
+      }
+      throw err;
+    }
+  })
+  .ws("/ws/telemetry", {
+    async open(ws) {
+      const { viewer } = ws.data as { viewer: { id: string } };
+      const allowed = await visibleFarmIds(viewer.id);
+
+      const send = (message: AppTelemetryMessage) => ws.send(message);
+
+      const unsubscribe = subscribe((printer: AppPrinterState) => {
+        // The line that keeps one organisation's telemetry out of another's
+        // console. `maySee` is in `telemetry/broadcast.ts` so it can be
+        // tested without standing up a socket.
+        if (!maySee(allowed, printer)) return;
+        send({ type: "printer.telemetry", printer });
+      });
+
+      // Cleanup is attached to the socket rather than kept in a module-level
+      // map: there is exactly one unsubscribe per socket and the close
+      // handler is the only thing that needs it. A leaked subscriber would
+      // push into a dead socket forever.
+      unsubscribers.set(ws.id, unsubscribe);
+
+      // Sent before any telemetry, so a client knows the difference between
+      // "nothing to show you" and "not connected yet". Counting farms the
+      // user can see is not the same as counting printers, so this asks for
+      // the printers.
+      send({ type: "ready", visiblePrinters: await countVisiblePrinters(viewer.id) });
+    },
+
+    message(_ws, _message) {
+      // Deliberately nothing. This socket is push-only; see the module
+      // comment. Ignoring rather than closing, because a well-meaning client
+      // sending a keepalive should not be punished for it.
+    },
+
+    close(ws) {
+      unsubscribers.get(ws.id)?.();
+      unsubscribers.delete(ws.id);
+    },
+  });

src/telemetry/broadcast.ts

@@ -0,0 +1,81 @@
+import type { AppPrinterState } from "@continuum/ts-types";
+
+/**
+ * Fan-out from the gateway socket to whoever is watching, with no imports but
+ * a type - the same reason as `dispatch/job_status.ts`, `stream.ts` and
+ * `auth/edge_upgrade.ts`: `../db` opens a connection at import time, and the
+ * part worth testing here is which subscriber is allowed to see what.
+ *
+ * ## In-process only, and that is a real limit
+ *
+ * A `Set` of callbacks in one Node/Bun process. Two backend instances behind
+ * a load balancer would each only push the telemetry their own gateways
+ * happen to be connected to, so an operator would see half a fleet depending
+ * on which instance their socket landed on.
+ *
+ * Not fixed here because the fix is Valkey pub/sub - already a dependency,
+ * already used by the AI frame queue - and swapping this `Set` for a channel
+ * subscription is a contained change once there is a second instance to
+ * justify it. Doing it now would mean testing a message broker to deliver
+ * events between two objects in the same process. Written down rather than
+ * discovered: see PLAN.md.
+ */
+
+/** What a subscriber is handed. `farmId` is what authorization turns on. */
+export type Listener = (printer: AppPrinterState) => void;
+
+const listeners = new Set<Listener>();
+
+/**
+ * Registers a subscriber and returns its own removal.
+ *
+ * Returning the unsubscribe rather than exposing a `remove(listener)` means
+ * a caller cannot get the identity wrong, and a socket's close handler has
+ * exactly one thing to call. Leaking a subscriber here would keep pushing
+ * into a dead socket forever.
+ */
+export function subscribe(listener: Listener): () => void {
+  listeners.add(listener);
+  return () => listeners.delete(listener);
+}
+
+/**
+ * Publishes one printer's new state to every subscriber.
+ *
+ * Each listener is called inside its own try/catch. One viewer's socket
+ * throwing on send - which a socket closing mid-iteration does - must not
+ * stop the other viewers from getting the same update, and must not
+ * propagate into `persistTelemetry`, where it would abort a whole farm's
+ * telemetry batch over one broken browser tab.
+ */
+export function publish(printer: AppPrinterState): void {
+  for (const listener of listeners) {
+    try {
+      listener(printer);
+    } catch (err) {
+      console.warn("[telemetry] a subscriber threw while being notified", err);
+    }
+  }
+}
+
+/** How many subscribers are attached. For tests and `/health`. */
+export function subscriberCount(): number {
+  return listeners.size;
+}
+
+/**
+ * Whether a user watching `visibleFarms` should be sent this printer.
+ *
+ * Separate from the socket so it can be tested without one, because this is
+ * the function that decides whether one organisation's operator sees
+ * another's machines. The set is resolved once when a socket opens - see
+ * `routes/ws/telemetry.ts` for why, and for what that costs.
+ */
+export function maySee(visibleFarms: ReadonlySet<string>, printer: AppPrinterState): boolean {
+  // A printer with no farm cannot be matched against anything, so it is not
+  // sent. There is no such row - `farm_id` is not nullable - but this is the
+  // one place where being wrong means leaking another org's machine, so it
+  // does not rely on that staying true.
+  if (!printer.farmId) return false;
+  return visibleFarms.has(printer.farmId);
+}

tests/telemetry.test.ts

@@ -0,0 +1,283 @@
+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> = {}): 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");
+  });
+});