diff --git a/package.json b/package.json index a0eeba2..17ccd3f 100644 --- a/package.json +++ b/package.json @@ -10,9 +10,11 @@ "db:generate": "drizzle-kit generate", "db:migrate": "drizzle-kit migrate", "db:studio": "drizzle-kit studio", + "test": "bun test", "typecheck": "tsc --noEmit" }, "dependencies": { + "@continuum/ts-types": "file:../continuum-schemas/packages/ts-types", "elysia": "^1.1.26", "@elysiajs/cors": "^1.1.1", "@elysiajs/jwt": "^1.1.1", diff --git a/src/db/schema/printers.ts b/src/db/schema/printers.ts index 9ec3b78..828c207 100644 --- a/src/db/schema/printers.ts +++ b/src/db/schema/printers.ts @@ -1,4 +1,4 @@ -import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real } from "drizzle-orm/pg-core"; +import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real, integer, uniqueIndex } from "drizzle-orm/pg-core"; import { farms } from "./farms"; export const printerVendorEnum = pgEnum("printer_vendor", ["bambu", "prusa", "klipper"]); @@ -11,24 +11,81 @@ export const printerStatusEnum = pgEnum("printer_status", [ "maintenance", ]); -export const printers = pgTable("printers", { - id: uuid("id").primaryKey().defaultRandom(), - farmId: uuid("farm_id") - .notNull() - .references(() => farms.id, { onDelete: "cascade" }), - name: text("name").notNull(), - vendor: printerVendorEnum("vendor").notNull(), - status: printerStatusEnum("status").notNull().default("offline"), - lanAddress: text("lan_address"), - serialNumber: text("serial_number"), - accessCode: text("access_code"), - nozzleTempC: real("nozzle_temp_c"), - bedTempC: real("bed_temp_c"), - metadata: jsonb("metadata").$type>().default({}), - lastSeenAt: timestamp("last_seen_at", { withTimezone: true }), - createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), - updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), -}); +export const printers = pgTable( + "printers", + { + id: uuid("id").primaryKey().defaultRandom(), + farmId: uuid("farm_id") + .notNull() + .references(() => farms.id, { onDelete: "cascade" }), + name: text("name").notNull(), + vendor: printerVendorEnum("vendor").notNull(), + status: printerStatusEnum("status").notNull().default("offline"), + lanAddress: text("lan_address"), + serialNumber: text("serial_number"), + accessCode: text("access_code"), + + /** + * The id this printer has in its gateway's own `printers.toml`, e.g. + * "p1p". The missing link between the two halves of the system: a + * gateway identifies its printers by this and knows nothing about + * `id` above, so every telemetry row and every outbound command is + * translated through it. Nullable because a printer can be registered + * in the cloud before any gateway claims it. + * + * Not globally unique - two farms may each have a printer called + * "p1p" - hence the per-farm unique index below rather than + * `.unique()`. + */ + gatewayPrinterId: text("gateway_printer_id"), + + /** + * Latest-snapshot telemetry, written by the /ws/edge/v1 route. This + * table deliberately keeps only the *current* value of each of these + * rather than a history: a time-series of them is a separate concern + * with very different storage characteristics, and nothing needs it + * yet. `status` and the two measured temperatures were already part of + * this pattern; the rest extend it. + */ + nozzleTempC: real("nozzle_temp_c"), + nozzleTargetTempC: real("nozzle_target_temp_c"), + bedTempC: real("bed_temp_c"), + bedTargetTempC: real("bed_target_temp_c"), + progressPercent: integer("progress_percent"), + currentLayer: integer("current_layer"), + totalLayers: integer("total_layers"), + remainingMinutes: integer("remaining_minutes"), + currentJobName: text("current_job_name"), + /** + * How many Health-Management-System entries the printer reports as + * active. A count, not decoded messages - that needs Bambu's HMS + * tables, which Continuum does not have. + */ + activeAlerts: integer("active_alerts"), + /** + * The printer's own state string, kept alongside the narrowed + * `status` enum. Worth storing both: the enum is what the UI filters + * and sorts on, but a vendor state Continuum does not recognise yet + * would otherwise be flattened into "idle" and lost. If this reads + * something unexpected, that is the signal to widen the mapping. + */ + vendorState: text("vendor_state"), + /** Vendor fault code; non-zero means the printer is reporting a fault. */ + vendorErrorCode: integer("vendor_error_code"), + + metadata: jsonb("metadata").$type>().default({}), + lastSeenAt: timestamp("last_seen_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), + }, + (table) => ({ + // What the edge route looks printers up by, on every telemetry batch. + gatewayPrinterIdx: uniqueIndex("printers_farm_gateway_printer_id_idx").on( + table.farmId, + table.gatewayPrinterId, + ), + }), +); export type Printer = typeof printers.$inferSelect; export type NewPrinter = typeof printers.$inferInsert; diff --git a/src/routes/printers.ts b/src/routes/printers.ts index aaec463..db289ef 100644 --- a/src/routes/printers.ts +++ b/src/routes/printers.ts @@ -3,6 +3,8 @@ import { eq } from "drizzle-orm"; import { db } from "../db"; import { printers } from "../db/schema"; import { PrinterInsertSchema } from "../db/validators"; +import { PrinterCommandSchema } from "@continuum/ts-types"; +import { isGatewayOnline, sendPrinterCommand } from "./ws/edge"; export const printerRoutes = new Elysia({ prefix: "/printers" }) .get("/", async ({ query }) => { @@ -32,4 +34,55 @@ export const printerRoutes = new Elysia({ prefix: "/printers" }) // Derived from `printers` via drizzle-typebox - see src/db/validators.ts. // Add or rename a column in the schema and this validator updates itself. { body: PrinterInsertSchema }, - ); + ) + .post( + "/:id/commands", + async ({ params, body, set }) => { + try { + const result = await sendPrinterCommand(params.id, body); + if (!result.ok) { + // 502, not 400: the request was fine and was delivered: it was the + // printer or its gateway that would not carry it out. + set.status = 502; + return { + error: { + code: "COMMAND_FAILED", + message: result.error ?? "the gateway reported the command failed", + }, + commandId: result.commandId, + }; + } + return { ok: true, commandId: result.commandId }; + } catch (err) { + // `sendPrinterCommand` throws only for things that are true before + // anything is sent: no such printer, no gateway has claimed it, or + // its farm has no gateway connected right now. + set.status = 409; + return { + error: { code: "PRINTER_UNREACHABLE", message: err instanceof Error ? err.message : String(err) }, + }; + } + }, + // The same command set the gateway accepts, straight from the shared + // schema - so this route cannot drift into offering something no gateway + // can execute. Note what is absent: there is no "start a print" command + // yet, deliberately. + { body: PrinterCommandSchema }, + ) + .get("/:id/gateway", async ({ params, set }) => { + const printer = await db.query.printers.findFirst({ + columns: { farmId: true, gatewayPrinterId: true }, + where: eq(printers.id, params.id), + }); + if (!printer) { + set.status = 404; + return { error: { code: "NOT_FOUND", message: `Printer ${params.id} not found` } }; + } + // Whether a command sent right now could even be delivered - worth being + // able to ask before trying, since the answer changes on its own. + return { + claimed: printer.gatewayPrinterId !== null, + gatewayPrinterId: printer.gatewayPrinterId, + gatewayOnline: isGatewayOnline(printer.farmId), + }; + }); diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index 1412116..96f4cae 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -1,89 +1,360 @@ -import { Elysia, t } from "elysia"; +import { Elysia } from "elysia"; import jwt from "@elysiajs/jwt"; +import { and, eq } from "drizzle-orm"; +import { + EdgeClientMessageSchema, + type EdgeServerMessage, + type PrinterCommand, + type PrinterTelemetry, +} from "@continuum/ts-types"; +import { db } from "../../db"; +import { printers } from "../../db/schema"; import { enqueueFrameIngestion } from "../../queue"; +/** + * Edge-gateway uplink. A continuum-proxy gateway connects here with a + * short-lived edge JWT (minted via POST /auth/edge-token), streams telemetry + * for its whole printer fleet, and receives commands for individual printers + * on the same socket. + * + * The wire format is `schemas/continuum/v1/edge/` in continuum-schemas, + * validated here through its TypeBox mirror in `@continuum/ts-types`. It is + * snake_case, unlike this backend's HTTP routes: the other end is + * serde-derived Rust, and this route previously drifted into a protocol the + * gateway did not speak at all (an `auth` handshake it never sends, one + * printer per telemetry message instead of a batch, UUID printer ids a + * gateway has no way to know, and no command path whatsoever) precisely + * because there was nothing written down for both sides to mirror. + */ + +interface PendingCommand { + resolve: (result: { ok: boolean; error?: string }) => void; + timer: ReturnType; +} + interface EdgeConn { farmId: string; - lastPing: number; + gatewayId: string; + sessionId: string; + lastSeenAt: number; + send: (message: EdgeServerMessage) => void; + /** + * Gateway printer id -> our `printers.id`. Filled on the first telemetry + * batch and reused after that, so a steady-state connection does not + * re-query the same mapping every few seconds. + */ + printerIds: Map; + /** Commands sent but not yet answered, keyed by command_id. */ + pending: Map; } const connections = new Map(); -const edgeMessage = t.Union([ - t.Object({ type: t.Literal("auth"), token: t.String() }), - t.Object({ type: t.Literal("heartbeat") }), - t.Object({ - type: t.Literal("frame_batch"), - printerId: t.String({ format: "uuid" }), - printJobId: t.Optional(t.String({ format: "uuid" })), - frames: t.Array(t.Object({ r2ObjectKey: t.String(), capturedAt: t.String() })), - }), - t.Object({ - type: t.Literal("telemetry"), - printerId: t.String({ format: "uuid" }), - status: t.String(), - nozzleTempC: t.Optional(t.Number()), - bedTempC: t.Optional(t.Number()), - }), -]); +/** + * farmId -> ws.id. One gateway per farm, so a second connection for the same + * farm replaces the first: that is what a gateway reconnecting after a + * network drop looks like, and the stale entry would otherwise keep receiving + * commands nothing is listening for. + */ +const gatewayByFarm = new Map(); + +/** How long to wait for a gateway to answer a command before giving up. */ +const COMMAND_TIMEOUT_MS = 10_000; /** - * Edge-gateway uplink: continuum-proxy connects here with a short-lived - * edge JWT (minted via POST /auth/edge-token) and streams telemetry + - * camera-frame batches for the lifetime of the connection. + * Maps a printer's own state string onto the narrowed `printer_status` enum. + * + * Unrecognised states deliberately fall through to the current value rather + * than to "idle": claiming a printer is idle when we do not understand what + * it said is the one wrong answer here, since idle is what the scheduler + * treats as "safe to send work to". The raw string is kept in + * `vendor_state` either way. */ +function mapStatus(state: string | null | undefined, printError: number | null | undefined) { + // `!= null` deliberately, not `!== undefined`: these fields arrive as either + // absent or an explicit null (see the Maybe() helper in @continuum/ts-types), + // and treating a null error code as a non-zero one would report every + // printer that has nothing to say as faulted. + if (printError != null && printError !== 0) return "error" as const; + + switch (state?.toUpperCase()) { + case "IDLE": + case "FINISH": + return "idle" as const; + case "RUNNING": + case "PREPARE": + case "SLICING": + return "printing" as const; + case "PAUSE": + return "paused" as const; + case "FAILED": + return "error" as const; + default: + return undefined; + } +} + +/** + * Writes one telemetry batch to the `printers` table. + * + * Only rows whose `gateway_printer_id` is already claimed by this farm are + * updated - an unknown id is logged and skipped rather than inserted. A + * gateway should not be able to create printer records just by mentioning + * them: that is what POST /printers is for, and auto-inserting here would + * turn a typo in someone's printers.toml into a permanent phantom printer. + */ +async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { + for (const snapshot of batch) { + let printerId = conn.printerIds.get(snapshot.printer_id); + + if (!printerId) { + const row = await db.query.printers.findFirst({ + columns: { id: true }, + where: and( + eq(printers.farmId, conn.farmId), + eq(printers.gatewayPrinterId, snapshot.printer_id), + ), + }); + + if (!row) { + console.warn( + `[edge] gateway ${conn.gatewayId} reported unknown printer ` + + `'${snapshot.printer_id}'; no printer in farm ${conn.farmId} claims that ` + + `gateway_printer_id, skipping`, + ); + continue; + } + + printerId = row.id; + conn.printerIds.set(snapshot.printer_id, printerId); + } + + const status = mapStatus(snapshot.state, snapshot.print_error); + + await db + .update(printers) + .set({ + // `?? null` rather than leaving the key out: a field the gateway + // stopped reporting has genuinely gone away (a finished job has no + // remaining minutes), and keeping the last value would leave stale + // numbers on screen forever. + ...(status ? { status } : {}), + vendorState: snapshot.state ?? null, + vendorErrorCode: snapshot.print_error ?? null, + nozzleTempC: snapshot.nozzle_temp ?? null, + nozzleTargetTempC: snapshot.nozzle_target ?? null, + bedTempC: snapshot.bed_temp ?? null, + bedTargetTempC: snapshot.bed_target ?? null, + progressPercent: snapshot.percent ?? null, + currentLayer: snapshot.layer ?? null, + totalLayers: snapshot.total_layers ?? null, + remainingMinutes: snapshot.remaining_minutes ?? null, + currentJobName: snapshot.job_name ?? null, + activeAlerts: snapshot.active_alerts ?? null, + lastSeenAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(printers.id, printerId)); + } +} + +/** + * Sends a command to the printer's gateway and waits for its result. + * + * Takes our own `printers.id` and translates to the gateway's id on the way + * out, so no caller elsewhere in the backend has to know that gateways use + * different identifiers. + * + * Throws if the printer is unknown, unclaimed, or its farm has no gateway + * connected right now. Resolves `{ ok: false }` if the gateway answered with + * a failure or did not answer within `COMMAND_TIMEOUT_MS` - a command that + * was delivered but never acknowledged is not the same as one that was never + * sent, and callers need to be able to tell those apart. + */ +export async function sendPrinterCommand( + printerId: string, + command: PrinterCommand, +): Promise<{ ok: boolean; error?: string; commandId: string }> { + const printer = await db.query.printers.findFirst({ + columns: { id: true, farmId: true, gatewayPrinterId: true }, + where: eq(printers.id, printerId), + }); + + if (!printer) throw new Error(`printer ${printerId} not found`); + if (!printer.gatewayPrinterId) { + throw new Error(`printer ${printerId} has no gateway_printer_id - no gateway has claimed it`); + } + + const wsId = gatewayByFarm.get(printer.farmId); + const conn = wsId ? connections.get(wsId) : undefined; + if (!conn) throw new Error(`no gateway connected for farm ${printer.farmId}`); + + const commandId = crypto.randomUUID(); + + const result = await new Promise<{ ok: boolean; error?: string }>((resolve) => { + const timer = setTimeout(() => { + conn.pending.delete(commandId); + resolve({ ok: false, error: `gateway did not answer within ${COMMAND_TIMEOUT_MS}ms` }); + }, COMMAND_TIMEOUT_MS); + + conn.pending.set(commandId, { resolve, timer }); + + conn.send({ + type: "command", + command_id: commandId, + printer_id: printer.gatewayPrinterId!, + command, + }); + }); + + return { ...result, commandId }; +} + +/** Whether a farm currently has a gateway connected. */ +export function isGatewayOnline(farmId: string): boolean { + const wsId = gatewayByFarm.get(farmId); + return wsId !== undefined && connections.has(wsId); +} + export const edgeWsRoute = new Elysia() .use(jwt({ name: "edgeJwt", secret: process.env.EDGE_JWT_SECRET ?? "dev-edge-secret-change-me" })) .ws("/ws/edge/v1", { - body: edgeMessage, + body: EdgeClientMessageSchema, + open(ws) { - connections.set(ws.id, { farmId: "", lastPing: Date.now() }); + // No farmId yet - set by `hello`, and every other message type refuses + // to do anything until it is. + connections.set(ws.id, { + farmId: "", + gatewayId: "", + sessionId: crypto.randomUUID(), + lastSeenAt: Date.now(), + send: (message) => ws.send(message), + printerIds: new Map(), + pending: new Map(), + }); }, + async message(ws, message) { const conn = connections.get(ws.id); if (!conn) return; + conn.lastSeenAt = Date.now(); + switch (message.type) { - case "auth": { + case "hello": { const payload = await ws.data.edgeJwt.verify(message.token); if (!payload || payload.scope !== "edge") { - ws.send({ type: "auth_error", message: "invalid edge token" }); + ws.send({ type: "hello_error", message: "invalid edge token" }); ws.close(); return; } - conn.farmId = payload.farmId as string; - ws.send({ type: "auth_ok", farmId: conn.farmId }); + + // The token decides the farm, never the message body: a gateway + // claiming a farm_id its token was not minted for is exactly what + // this check is here to stop. + const farmId = payload.farmId as string; + if (message.farm_id && message.farm_id !== farmId) { + ws.send({ type: "hello_error", message: "farm_id does not match the token's farm" }); + ws.close(); + return; + } + + conn.farmId = farmId; + conn.gatewayId = message.gateway_id; + + // One gateway per farm: drop whatever was registered before. On a + // reconnect that stale socket is already dead, and leaving it + // registered would send commands into it. + const previous = gatewayByFarm.get(farmId); + if (previous && previous !== ws.id) { + console.warn( + `[edge] farm ${farmId} already had gateway connection ${previous}; ` + + `replacing it with ${ws.id} (gateway ${message.gateway_id})`, + ); + } + gatewayByFarm.set(farmId, ws.id); + + ws.send({ type: "hello_ack", session_id: conn.sessionId }); + console.info(`[edge] gateway ${message.gateway_id} connected for farm ${farmId}`); break; } + case "heartbeat": { - conn.lastPing = Date.now(); - ws.send({ type: "heartbeat_ack" }); + // Echo the sequence number back. The gateway uses it to tell a + // stalled connection from a slow one, which it cannot do against a + // bare ack. + ws.send({ type: "heartbeat_ack", seq: message.seq }); break; } + case "telemetry": { if (!conn.farmId) return; - // Fanned out to subscribers of the dashboard telemetry store; boilerplate - // just acknowledges receipt here. - ws.send({ type: "telemetry_ack", printerId: message.printerId }); + await persistTelemetry(conn, message.printers); break; } + + case "command_result": { + const pending = conn.pending.get(message.command_id); + if (!pending) { + // Either the command already timed out, or this gateway answered + // something we never asked. Both are worth saying out loud. + console.warn( + `[edge] gateway ${conn.gatewayId} sent a result for unknown or ` + + `expired command ${message.command_id}`, + ); + return; + } + clearTimeout(pending.timer); + conn.pending.delete(message.command_id); + pending.resolve({ ok: message.ok, error: message.error }); + break; + } + case "frame_batch": { if (!conn.farmId) return; + + const printerId = conn.printerIds.get(message.printer_id); + if (!printerId) { + console.warn( + `[edge] frame_batch for unknown printer '${message.printer_id}' from ` + + `gateway ${conn.gatewayId}, dropping ${message.frames.length} frame(s)`, + ); + return; + } + for (const frame of message.frames) { await enqueueFrameIngestion({ farmId: conn.farmId, - printerId: message.printerId, - printJobId: message.printJobId ?? null, - r2ObjectKey: frame.r2ObjectKey, - capturedAt: frame.capturedAt, + printerId, + printJobId: message.print_job_id ?? null, + r2ObjectKey: frame.object_key, + capturedAt: frame.captured_at, }); } - ws.send({ type: "frame_batch_ack", count: message.frames.length }); break; } } }, + close(ws) { + const conn = connections.get(ws.id); + if (conn) { + // Anything still waiting on this gateway will never be answered now. + // Failing them immediately beats making every caller wait out the + // full timeout for a socket that is already gone. + for (const [commandId, pending] of conn.pending) { + clearTimeout(pending.timer); + pending.resolve({ ok: false, error: "gateway disconnected before answering" }); + conn.pending.delete(commandId); + } + if (conn.farmId && gatewayByFarm.get(conn.farmId) === ws.id) { + gatewayByFarm.delete(conn.farmId); + } + if (conn.gatewayId) { + console.info(`[edge] gateway ${conn.gatewayId} disconnected (farm ${conn.farmId})`); + } + } connections.delete(ws.id); }, }); diff --git a/tests/edge_protocol.test.ts b/tests/edge_protocol.test.ts new file mode 100644 index 0000000..f2f2c8d --- /dev/null +++ b/tests/edge_protocol.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it } from "bun:test"; +import { Value } from "@sinclair/typebox/value"; +import { EdgeClientMessageSchema, EdgeServerMessageSchema } from "@continuum/ts-types"; + +import currentCapture from "./fixtures/gateway_frames.json"; +import legacyCapture from "./fixtures/gateway_frames_legacy_nulls.json"; + +/** + * A contract test, not a unit test: it checks that what continuum-proxy + * actually puts on the wire is what this backend's validator accepts. + * + * The fixtures are real captures from a live gateway talking to a real P1P and + * H2C, not hand-written examples - which is the whole point. The two ends of + * /ws/edge/v1 had silently drifted into completely different protocols + * (different handshake, different telemetry shape, printer ids the gateway + * could not know), and every hand-written example agreed with whichever side + * wrote it. Re-capture with the script in the proxy's examples/ when the + * protocol changes, rather than editing these by hand. + */ + +function check(schema: Parameters[0], value: unknown) { + const ok = Value.Check(schema, value); + if (!ok) { + // Surface *why*, or a failure here is just "false !== true". + const errors = [...Value.Errors(schema, value)].map( + (e) => `${e.path || "/"}: ${e.message} (got ${JSON.stringify(e.value)})`, + ); + throw new Error(`validation failed:\n ${errors.join("\n ")}`); + } +} + +describe("gateway -> backend frames", () => { + for (const frame of currentCapture.frames) { + it(`accepts a real '${frame.type}' frame`, () => { + check(EdgeClientMessageSchema, frame); + }); + } + + it("accepts the pre-fix capture, where empty fields were explicit nulls", () => { + // Liberal reader: the gateway now omits these, but a serde-derived + // producer emits null by default and one such field must not cost us a + // whole farm's telemetry batch. + for (const frame of legacyCapture.frames) { + check(EdgeClientMessageSchema, frame); + } + }); + + it("carries the live telemetry this was captured from", () => { + const telemetry = currentCapture.frames.find((f) => f.type === "telemetry"); + expect(telemetry).toBeDefined(); + const printers = (telemetry as { printers: Array<{ printer_id: string }> }).printers; + expect(printers.length).toBeGreaterThan(0); + // Guards against a fixture that validates because it is empty. + expect(printers.every((p) => typeof p.printer_id === "string" && p.printer_id.length > 0)).toBe(true); + }); + + it("rejects a telemetry batch whose printer has no id", () => { + // printer_id and name are the only required fields; everything else is + // genuinely unknowable for a printer that has only just connected. + check(EdgeClientMessageSchema, { type: "telemetry", printers: [{ printer_id: "p1p", name: "P1P" }] }); + expect(Value.Check(EdgeClientMessageSchema, { type: "telemetry", printers: [{ name: "P1P" }] })).toBe(false); + }); + + it("rejects an unknown message type", () => { + // The old protocol's handshake, which no gateway ever sent. + expect(Value.Check(EdgeClientMessageSchema, { type: "auth", token: "x" })).toBe(false); + }); +}); + +describe("backend -> gateway frames", () => { + // Exactly what src/routes/ws/edge.ts sends. If one of these stops + // validating, the gateway's serde enum will have stopped accepting it too - + // see the mirror of this test in continuum-proxy/tests/uplink_protocol.rs. + const outbound = [ + { type: "hello_ack", session_id: "3f2b1c5e-0000-4000-8000-000000000000" }, + { type: "hello_error", message: "invalid edge token" }, + { type: "heartbeat_ack", seq: 7 }, + { + type: "command", + command_id: "9a7d2f10-0000-4000-8000-000000000000", + printer_id: "p1p", + command: { action: "set_fan_speed", percent: 40 }, + }, + { + type: "command", + command_id: "9a7d2f11-0000-4000-8000-000000000000", + printer_id: "p1p", + command: { action: "send_gcode_line", line: "M106 S0" }, + }, + ]; + + for (const frame of outbound) { + it(`emits a valid '${frame.type}' (${(frame as { command?: { action: string } }).command?.action ?? "no command"})`, () => { + check(EdgeServerMessageSchema, frame); + }); + } + + it("has no start-print command to send", () => { + // Deliberate: the gateway can start an uploaded file, but that path has + // never been run against real hardware. Delete this test when it has been, + // and add the variant to the schema. + expect( + Value.Check(EdgeServerMessageSchema, { + type: "command", + command_id: "x", + printer_id: "p1p", + command: { action: "start_print", file_name: "plate_1.3mf" }, + }), + ).toBe(false); + }); +}); diff --git a/tests/fixtures/gateway_frames.json b/tests/fixtures/gateway_frames.json new file mode 100644 index 0000000..e876c60 --- /dev/null +++ b/tests/fixtures/gateway_frames.json @@ -0,0 +1,56 @@ +{ + "_note": "Real frames captured from a continuum-proxy gateway on 2026-10-03, talking to a live Bambu P1P (idle) and H2C (mid-print). One of each message type. This is the current wire format: fields with no value are omitted, not null.", + "frames": [ + { + "type": "hello", + "gateway_id": "dev-gw", + "farm_id": "dev-farm", + "token": "dev-token" + }, + { + "type": "heartbeat", + "seq": 2 + }, + { + "type": "telemetry", + "printers": [ + { + "printer_id": "p1p", + "name": "P1P", + "state": "IDLE", + "percent": 0, + "layer": 0, + "total_layers": 0, + "remaining_minutes": 0, + "nozzle_temp": 23.375, + "nozzle_target": 0.0, + "bed_temp": 22.40625, + "bed_target": 0.0, + "print_error": 0, + "active_alerts": 0 + }, + { + "printer_id": "h2c", + "name": "H2C", + "state": "RUNNING", + "percent": 99, + "layer": 50, + "total_layers": 50, + "remaining_minutes": 1, + "job_name": "świeczka-24.stl", + "nozzle_temp": 230.0, + "nozzle_target": 230.0, + "bed_temp": 35.0, + "bed_target": 35.0, + "print_error": 0, + "active_alerts": 3 + } + ] + }, + { + "type": "command_result", + "command_id": "cap-1", + "ok": true + } + ] +} diff --git a/tests/fixtures/gateway_frames_legacy_nulls.json b/tests/fixtures/gateway_frames_legacy_nulls.json new file mode 100644 index 0000000..df67f73 --- /dev/null +++ b/tests/fixtures/gateway_frames_legacy_nulls.json @@ -0,0 +1,58 @@ +{ + "_note": "The same capture from before the gateway was taught to omit empty fields, kept deliberately: it is what a serde-derived producer emits by default, and the validator must keep accepting it. Note p1p's \"job_name\": null - an idle printer with no job - which used to fail validation and take the whole telemetry batch down with it.", + "frames": [ + { + "type": "hello", + "gateway_id": "dev-gw", + "farm_id": "dev-farm", + "token": "dev-token" + }, + { + "type": "heartbeat", + "seq": 2 + }, + { + "type": "telemetry", + "printers": [ + { + "printer_id": "p1p", + "name": "P1P", + "state": "IDLE", + "percent": 0, + "layer": 0, + "total_layers": 0, + "remaining_minutes": 0, + "job_name": null, + "nozzle_temp": 23.40625, + "nozzle_target": 0.0, + "bed_temp": 22.4375, + "bed_target": 0.0, + "print_error": 0, + "active_alerts": 0 + }, + { + "printer_id": "h2c", + "name": "H2C", + "state": "RUNNING", + "percent": 97, + "layer": 50, + "total_layers": 50, + "remaining_minutes": 3, + "job_name": "świeczka-24.stl", + "nozzle_temp": 230.0, + "nozzle_target": 230.0, + "bed_temp": 35.0, + "bed_target": 35.0, + "print_error": 0, + "active_alerts": 3 + } + ] + }, + { + "type": "command_result", + "command_id": "cap-1", + "ok": true, + "error": null + } + ] +}