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.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-28 16:13:57 +00:00
commit 23643904da
20 changed files with 661 additions and 0 deletions
+22
View File
@@ -0,0 +1,22 @@
PORT=3000
# Postgres, routed through pgCat/PgBouncer
DATABASE_URL=postgres://continuum:continuum@localhost:6432/continuum
DATABASE_POOL_MAX=10
# Redis (state + streams)
REDIS_URL=redis://localhost:6379
# Cloudflare R2 (S3-compatible)
R2_ACCOUNT_ID=
R2_BUCKET=continuum
R2_ENDPOINT=
R2_ACCESS_KEY_ID=
R2_SECRET_ACCESS_KEY=
# Supabase Auth
SUPABASE_JWT_SECRET=
SUPABASE_URL=
# Minted for continuum-proxy edge gateways via POST /auth/edge-token
EDGE_JWT_SECRET=
+7
View File
@@ -0,0 +1,7 @@
node_modules/
dist/
.env
.env.local
drizzle/
*.log
.DS_Store
+27
View File
@@ -0,0 +1,27 @@
# continuum-backend
Cloud control plane for Continuum. Bun + ElysiaJS, PostgreSQL (via pgCat/PgBouncer,
with pgvector for frame embeddings), Redis Streams, and Cloudflare R2 storage.
## Run
```sh
bun install
cp .env.example .env # fill in DATABASE_URL, REDIS_URL, R2_*, SUPABASE_JWT_SECRET
bun run db:generate && bun run db:migrate
bun run dev
```
## Layout
- `src/db/` — Drizzle schema (`printers`, `farms`, `print_jobs`, `frame_embeddings`)
and client, pointed at pgCat.
- `src/s3/` — R2 presigned URL helpers (`PUT` for client uploads, `GET` for the
edge proxy).
- `src/queue/` — Redis client + `XADD` helper pushing onto `stream:camera_frames`
for `continuum-ai-worker`.
- `src/routes/``/auth`, `/farms`, `/printers`, `/jobs`, and the edge-gateway
WebSocket at `/ws/edge/v1`.
`frame_embeddings.embedding` requires the `vector` extension:
`CREATE EXTENSION IF NOT EXISTS vector;`
+12
View File
@@ -0,0 +1,12 @@
import { defineConfig } from "drizzle-kit";
export default defineConfig({
schema: "./src/db/schema/index.ts",
out: "./drizzle",
dialect: "postgresql",
dbCredentials: {
url: process.env.DATABASE_URL ?? "postgres://continuum:continuum@localhost:6432/continuum",
},
verbose: true,
strict: true,
});
+30
View File
@@ -0,0 +1,30 @@
{
"name": "@continuum/backend",
"version": "0.1.0",
"private": true,
"type": "module",
"scripts": {
"dev": "bun run --watch src/index.ts",
"start": "bun run src/index.ts",
"db:generate": "drizzle-kit generate",
"db:migrate": "drizzle-kit migrate",
"db:studio": "drizzle-kit studio",
"typecheck": "tsc --noEmit"
},
"dependencies": {
"elysia": "^1.1.26",
"@elysiajs/cors": "^1.1.1",
"@elysiajs/jwt": "^1.1.1",
"@elysiajs/swagger": "^1.1.6",
"@aws-sdk/client-s3": "^3.658.1",
"@aws-sdk/s3-request-presigner": "^3.658.1",
"drizzle-orm": "^0.33.0",
"postgres": "^3.4.4",
"ioredis": "^5.4.1"
},
"devDependencies": {
"drizzle-kit": "^0.24.2",
"@types/bun": "^1.1.10",
"typescript": "^5.6.2"
}
}
+18
View File
@@ -0,0 +1,18 @@
import { drizzle } from "drizzle-orm/postgres-js";
import postgres from "postgres";
import * as schema from "./schema";
// Points at pgCat/PgBouncer, not Postgres directly, so keep prepared
// statements disabled and cap the pool per-instance connections low.
const connectionString = process.env.DATABASE_URL;
if (!connectionString) {
throw new Error("DATABASE_URL is not set");
}
const client = postgres(connectionString, {
prepare: false,
max: Number(process.env.DATABASE_POOL_MAX ?? 10),
});
export const db = drizzle(client, { schema });
export type Database = typeof db;
+48
View File
@@ -0,0 +1,48 @@
import { pgTable, uuid, text, timestamp, customType, index } from "drizzle-orm/pg-core";
import { printers } from "./printers";
import { printJobs } from "./print_jobs";
const vector = customType<{ data: number[]; driverData: string }>({
dataType(config) {
const dims = (config as { dimensions?: number } | undefined)?.dimensions ?? 512;
return `vector(${dims})`;
},
toDriver(value) {
return `[${value.join(",")}]`;
},
fromDriver(value) {
return value
.slice(1, -1)
.split(",")
.filter(Boolean)
.map(Number);
},
});
/**
* CLIP-style embeddings of camera frames captured during a print job, used for
* nearest-neighbour failure-pattern search (pgvector, cosine distance).
*/
export const frameEmbeddings = pgTable(
"frame_embeddings",
{
id: uuid("id").primaryKey().defaultRandom(),
printJobId: uuid("print_job_id")
.notNull()
.references(() => printJobs.id, { onDelete: "cascade" }),
printerId: uuid("printer_id")
.notNull()
.references(() => printers.id, { onDelete: "cascade" }),
r2ObjectKey: text("r2_object_key").notNull(),
embedding: vector("embedding", { dimensions: 512 }).notNull(),
capturedAt: timestamp("captured_at", { withTimezone: true }).defaultNow().notNull(),
},
(table) => ({
embeddingIvfflatIdx: index("frame_embeddings_embedding_ivfflat_idx")
.using("ivfflat", table.embedding.op("vector_cosine_ops"))
.with({ lists: 100 }),
}),
);
export type FrameEmbedding = typeof frameEmbeddings.$inferSelect;
export type NewFrameEmbedding = typeof frameEmbeddings.$inferInsert;
+15
View File
@@ -0,0 +1,15 @@
import { pgTable, uuid, text, timestamp, jsonb } from "drizzle-orm/pg-core";
export const farms = pgTable("farms", {
id: uuid("id").primaryKey().defaultRandom(),
name: text("name").notNull(),
ownerId: uuid("owner_id").notNull(),
edgeProxyToken: text("edge_proxy_token").notNull(),
location: text("location"),
settings: jsonb("settings").$type<Record<string, unknown>>().default({}),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
});
export type Farm = typeof farms.$inferSelect;
export type NewFarm = typeof farms.$inferInsert;
+4
View File
@@ -0,0 +1,4 @@
export * from "./farms";
export * from "./printers";
export * from "./print_jobs";
export * from "./embeddings";
+36
View File
@@ -0,0 +1,36 @@
import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real } from "drizzle-orm/pg-core";
import { printers } from "./printers";
import { farms } from "./farms";
export const printJobStatusEnum = pgEnum("print_job_status", [
"queued",
"uploading",
"printing",
"paused",
"completed",
"failed",
"cancelled",
]);
export const printJobs = pgTable("print_jobs", {
id: uuid("id").primaryKey().defaultRandom(),
farmId: uuid("farm_id")
.notNull()
.references(() => farms.id, { onDelete: "cascade" }),
printerId: uuid("printer_id").references(() => printers.id, { onDelete: "set null" }),
fileName: text("file_name").notNull(),
r2ObjectKey: text("r2_object_key").notNull(),
status: printJobStatusEnum("status").notNull().default("queued"),
progressPct: real("progress_pct").notNull().default(0),
estimatedTimeS: real("estimated_time_s"),
elapsedTimeS: real("elapsed_time_s"),
submittedBy: uuid("submitted_by").notNull(),
metadata: jsonb("metadata").$type<Record<string, unknown>>().default({}),
startedAt: timestamp("started_at", { withTimezone: true }),
completedAt: timestamp("completed_at", { withTimezone: true }),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
});
export type PrintJob = typeof printJobs.$inferSelect;
export type NewPrintJob = typeof printJobs.$inferInsert;
+34
View File
@@ -0,0 +1,34 @@
import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real } from "drizzle-orm/pg-core";
import { farms } from "./farms";
export const printerVendorEnum = pgEnum("printer_vendor", ["bambu", "prusa", "klipper"]);
export const printerStatusEnum = pgEnum("printer_status", [
"offline",
"idle",
"printing",
"paused",
"error",
"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<Record<string, unknown>>().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 type Printer = typeof printers.$inferSelect;
export type NewPrinter = typeof printers.$inferInsert;
+30
View File
@@ -0,0 +1,30 @@
import { Elysia } from "elysia";
import { cors } from "@elysiajs/cors";
import { swagger } from "@elysiajs/swagger";
import { sql } from "drizzle-orm";
import { db } from "./db";
import { redis } from "./queue";
import { authRoutes } from "./routes/auth";
import { farmRoutes } from "./routes/farms";
import { printerRoutes } from "./routes/printers";
import { jobRoutes } from "./routes/jobs";
import { edgeWsRoute } from "./routes/ws/edge";
const app = new Elysia()
.use(cors())
.use(swagger({ path: "/docs" }))
.get("/health", async () => {
const [{ ok }] = await db.execute<{ ok: number }>(sql`select 1 as ok`);
const redisPing = await redis.ping();
return { status: "ok", db: ok === 1, redis: redisPing === "PONG" };
})
.use(authRoutes)
.use(farmRoutes)
.use(printerRoutes)
.use(jobRoutes)
.use(edgeWsRoute)
.listen(Number(process.env.PORT ?? 3000));
console.log(`continuum-backend listening on :${app.server?.port}`);
export type App = typeof app;
+33
View File
@@ -0,0 +1,33 @@
import Redis from "ioredis";
export const redis = new Redis(process.env.REDIS_URL ?? "redis://localhost:6379", {
maxRetriesPerRequest: 3,
});
export const CAMERA_FRAMES_STREAM = "stream:camera_frames";
export interface FrameIngestionTask {
farmId: string;
printerId: string;
printJobId: string | null;
r2ObjectKey: string;
capturedAt: string;
}
/** Pushes a frame-ingestion task for continuum-ai-worker to pick up via XREADGROUP. */
export async function enqueueFrameIngestion(task: FrameIngestionTask): Promise<string> {
return redis.xadd(
CAMERA_FRAMES_STREAM,
"*",
"farmId",
task.farmId,
"printerId",
task.printerId,
"printJobId",
task.printJobId ?? "",
"r2ObjectKey",
task.r2ObjectKey,
"capturedAt",
task.capturedAt,
) as Promise<string>;
}
+61
View File
@@ -0,0 +1,61 @@
import { Elysia, t } from "elysia";
import jwt from "@elysiajs/jwt";
import { db } from "../db";
import { farms } from "../db/schema";
import { eq } from "drizzle-orm";
/**
* Verifies Supabase-issued JWTs (HS256, shared secret) and issues short-lived
* edge-gateway tokens scoped to a single farm for the /ws/edge/v1 handshake.
*/
export const authRoutes = new Elysia({ prefix: "/auth" })
.use(
jwt({
name: "supabaseJwt",
secret: process.env.SUPABASE_JWT_SECRET ?? "dev-secret-change-me",
}),
)
.use(
jwt({
name: "edgeJwt",
secret: process.env.EDGE_JWT_SECRET ?? "dev-edge-secret-change-me",
}),
)
.post(
"/session",
async ({ supabaseJwt, headers, set }) => {
const authHeader = headers.authorization;
const token = authHeader?.startsWith("Bearer ") ? authHeader.slice(7) : undefined;
if (!token) {
set.status = 401;
return { error: { code: "UNAUTHORIZED", message: "Missing bearer token" } };
}
const payload = await supabaseJwt.verify(token);
if (!payload) {
set.status = 401;
return { error: { code: "UNAUTHORIZED", message: "Invalid or expired token" } };
}
return { userId: payload.sub, email: payload.email, role: payload.role };
},
)
.post(
"/edge-token",
async ({ edgeJwt, body, set }) => {
const farm = await db.query.farms.findFirst({ where: eq(farms.id, body.farmId) });
if (!farm || farm.edgeProxyToken !== body.edgeProxyToken) {
set.status = 401;
return { error: { code: "UNAUTHORIZED", message: "Unknown farm or bad proxy token" } };
}
const token = await edgeJwt.sign({ farmId: farm.id, scope: "edge" });
return { token, expiresInSeconds: 3600 };
},
{
body: t.Object({
farmId: t.String({ format: "uuid" }),
edgeProxyToken: t.String(),
}),
},
);
+38
View File
@@ -0,0 +1,38 @@
import { Elysia, t } from "elysia";
import { eq } from "drizzle-orm";
import { db } from "../db";
import { farms } from "../db/schema";
export const farmRoutes = new Elysia({ prefix: "/farms" })
.get("/", async () => db.select().from(farms))
.get("/:id", async ({ params, set }) => {
const farm = await db.query.farms.findFirst({ where: eq(farms.id, params.id) });
if (!farm) {
set.status = 404;
return { error: { code: "NOT_FOUND", message: `Farm ${params.id} not found` } };
}
return farm;
})
.post(
"/",
async ({ body, set }) => {
const [farm] = await db
.insert(farms)
.values({
name: body.name,
ownerId: body.ownerId,
edgeProxyToken: crypto.randomUUID(),
location: body.location,
})
.returning();
set.status = 201;
return farm;
},
{
body: t.Object({
name: t.String({ minLength: 1 }),
ownerId: t.String({ format: "uuid" }),
location: t.Optional(t.String()),
}),
},
);
+51
View File
@@ -0,0 +1,51 @@
import { Elysia, t } from "elysia";
import { eq } from "drizzle-orm";
import { db } from "../db";
import { printJobs } from "../db/schema";
import { getUploadUrl, jobFileKey } from "../s3";
export const jobRoutes = new Elysia({ prefix: "/jobs" })
.get("/", async ({ query }) => {
if (query.farmId) {
return db.select().from(printJobs).where(eq(printJobs.farmId, query.farmId));
}
return db.select().from(printJobs);
}, { query: t.Object({ farmId: t.Optional(t.String({ format: "uuid" })) }) })
.get("/:id", async ({ params, set }) => {
const job = await db.query.printJobs.findFirst({ where: eq(printJobs.id, params.id) });
if (!job) {
set.status = 404;
return { error: { code: "NOT_FOUND", message: `Print job ${params.id} not found` } };
}
return job;
})
.post(
"/",
async ({ body, set }) => {
const jobId = crypto.randomUUID();
const r2ObjectKey = jobFileKey(body.farmId, jobId, body.fileName);
const [job] = await db
.insert(printJobs)
.values({
id: jobId,
farmId: body.farmId,
fileName: body.fileName,
r2ObjectKey,
submittedBy: body.submittedBy,
})
.returning();
const uploadUrl = await getUploadUrl(r2ObjectKey, body.contentType);
set.status = 201;
return { job, uploadUrl };
},
{
body: t.Object({
farmId: t.String({ format: "uuid" }),
fileName: t.String({ minLength: 1 }),
contentType: t.Optional(t.String()),
submittedBy: t.String({ format: "uuid" }),
}),
},
);
+48
View File
@@ -0,0 +1,48 @@
import { Elysia, t } from "elysia";
import { eq } from "drizzle-orm";
import { db } from "../db";
import { printers } from "../db/schema";
export const printerRoutes = new Elysia({ prefix: "/printers" })
.get("/", async ({ query }) => {
if (query.farmId) {
return db.select().from(printers).where(eq(printers.farmId, query.farmId));
}
return db.select().from(printers);
}, { query: t.Object({ farmId: t.Optional(t.String({ format: "uuid" })) }) })
.get("/:id", async ({ params, set }) => {
const printer = await db.query.printers.findFirst({ where: eq(printers.id, params.id) });
if (!printer) {
set.status = 404;
return { error: { code: "NOT_FOUND", message: `Printer ${params.id} not found` } };
}
return printer;
})
.post(
"/",
async ({ body, set }) => {
const [printer] = await db
.insert(printers)
.values({
farmId: body.farmId,
name: body.name,
vendor: body.vendor,
lanAddress: body.lanAddress,
serialNumber: body.serialNumber,
accessCode: body.accessCode,
})
.returning();
set.status = 201;
return printer;
},
{
body: t.Object({
farmId: t.String({ format: "uuid" }),
name: t.String({ minLength: 1 }),
vendor: t.Union([t.Literal("bambu"), t.Literal("prusa"), t.Literal("klipper")]),
lanAddress: t.Optional(t.String()),
serialNumber: t.Optional(t.String()),
accessCode: t.Optional(t.String()),
}),
},
);
+89
View File
@@ -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);
},
});
+34
View File
@@ -0,0 +1,34 @@
import { S3Client, PutObjectCommand, GetObjectCommand } from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
const accountId = process.env.R2_ACCOUNT_ID ?? "";
const bucket = process.env.R2_BUCKET ?? "continuum";
export const s3 = new S3Client({
region: "auto",
endpoint: process.env.R2_ENDPOINT ?? `https://${accountId}.r2.cloudflarestorage.com`,
credentials: {
accessKeyId: process.env.R2_ACCESS_KEY_ID ?? "",
secretAccessKey: process.env.R2_SECRET_ACCESS_KEY ?? "",
},
});
/** Presigned PUT for clients uploading a G-code/3MF file directly to R2. */
export async function getUploadUrl(key: string, contentType = "application/octet-stream", expiresInSeconds = 900) {
const command = new PutObjectCommand({ Bucket: bucket, Key: key, ContentType: contentType });
return getSignedUrl(s3, command, { expiresIn: expiresInSeconds });
}
/** Presigned GET for the edge proxy to download a job file over LAN via FTPS relay or direct fetch. */
export async function getDownloadUrl(key: string, expiresInSeconds = 900) {
const command = new GetObjectCommand({ Bucket: bucket, Key: key });
return getSignedUrl(s3, command, { expiresIn: expiresInSeconds });
}
export function jobFileKey(farmId: string, jobId: string, fileName: string) {
return `farms/${farmId}/jobs/${jobId}/${fileName}`;
}
export function frameKey(farmId: string, printerId: string, frameId: string) {
return `farms/${farmId}/printers/${printerId}/frames/${frameId}.jpg`;
}
+24
View File
@@ -0,0 +1,24 @@
{
"compilerOptions": {
"target": "ESNext",
"module": "ESNext",
"moduleResolution": "Bundler",
"lib": ["ESNext"],
"types": ["bun-types"],
"strict": true,
"noUncheckedIndexedAccess": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"allowJs": true,
"isolatedModules": true,
"verbatimModuleSyntax": true,
"noEmit": true,
"baseUrl": ".",
"paths": {
"@/*": ["src/*"]
}
},
"include": ["src/**/*.ts"]
}