Initial boilerplate scaffold for continuum-backend
Bun + ElysiaJS control plane: Drizzle/Postgres schema (printers, farms, print_jobs, pgvector frame_embeddings), R2 presigned URLs, Redis Streams frame ingestion queue, REST routes, and the /ws/edge/v1 edge-gateway WebSocket uplink.
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
import { Elysia, t } from "elysia";
|
||||
import jwt from "@elysiajs/jwt";
|
||||
import { enqueueFrameIngestion } from "../../queue";
|
||||
|
||||
interface EdgeConn {
|
||||
farmId: string;
|
||||
lastPing: number;
|
||||
}
|
||||
|
||||
const connections = new Map<string, EdgeConn>();
|
||||
|
||||
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()),
|
||||
}),
|
||||
]);
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
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,
|
||||
open(ws) {
|
||||
connections.set(ws.id, { farmId: "", lastPing: Date.now() });
|
||||
},
|
||||
async message(ws, message) {
|
||||
const conn = connections.get(ws.id);
|
||||
if (!conn) return;
|
||||
|
||||
switch (message.type) {
|
||||
case "auth": {
|
||||
const payload = await ws.data.edgeJwt.verify(message.token);
|
||||
if (!payload || payload.scope !== "edge") {
|
||||
ws.send({ type: "auth_error", message: "invalid edge token" });
|
||||
ws.close();
|
||||
return;
|
||||
}
|
||||
conn.farmId = payload.farmId as string;
|
||||
ws.send({ type: "auth_ok", farmId: conn.farmId });
|
||||
break;
|
||||
}
|
||||
case "heartbeat": {
|
||||
conn.lastPing = Date.now();
|
||||
ws.send({ type: "heartbeat_ack" });
|
||||
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 });
|
||||
break;
|
||||
}
|
||||
case "frame_batch": {
|
||||
if (!conn.farmId) 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,
|
||||
});
|
||||
}
|
||||
ws.send({ type: "frame_batch_ack", count: message.frames.length });
|
||||
break;
|
||||
}
|
||||
}
|
||||
},
|
||||
close(ws) {
|
||||
connections.delete(ws.id);
|
||||
},
|
||||
});
|
||||
Reference in New Issue
Block a user