continuum-backend/tests/telemetry.test.ts

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");
  });
});