Compare commits

..
9 Commits
Author SHA1 Message Date
octoturge fc8cf1d509 Add organisation invitations and one-time-code gateway enrolment
Invitations: token shown once, only its hash stored, 7-day expiry, the
accepting account's email must match. Enrolment codes: 15 minutes,
single use, hashed; POST /auth/enrol returns the farm's existing gateway
token without rotating it and is rate-limited per IP. No email is sent;
the inviter shares the link. Migration 0007, unit tests and
scripts/e2e-onboarding.sh.
2026-10-30 08:13:00 +00:00
octoturge b431ee4030 Apply job outcomes a gateway kept while the cloud was away
A job_event from a gateway moves a job that is still in flight (uploading,
printing, paused) to printing, completed or failed, and is acknowledged
whether or not it changed anything so the gateway stops sending it. A job
the cloud has cancelled, completed or put back in the queue is left alone,
and a gateway can only touch jobs of its own farm. If the database is
unreachable it is not acknowledged, so it is sent again.

scripts/e2e-vendors.sh gains a phase that takes the backend away for the
whole of a fake print and checks the job completes on reconnect from the
proxy's kept record.
2026-10-30 08:05:05 +00:00
octoturge 64a43593b2 Fail a job when the gateway reports it could not start it
A start_print is acknowledged as soon as the gateway accepts it, and the
download and transfer that follow can still fail. That late result was only
logged, so the job sat in uploading until the stall reaper gave up twenty
minutes later.

The dispatcher now remembers which job each start command belonged to. A
late failure for it fails the job with the gateway's own reason in
metadata.lastDispatchError - but only while the job is still uploading, so a
job the printer has since started is left to its telemetry.

scripts/e2e-vendors.sh gains a phase where the proxy refuses the download
and the job must fail at once with that reason.
2026-10-30 07:52:44 +00:00
octoturge d53c22b641 Add Klipper, OctoPrint and PrusaLink printers from the API
POST /farms/:id/printers now takes any of the four vendors the gateway can
drive. Credentials depend on the vendor: Bambu needs an access code and
serial, OctoPrint and PrusaLink an API key, Klipper nothing. A missing one
is a 422 naming what is needed. The API key is relayed to the gateway and
kept out of logs and errors like the access code.

The printer_vendor enum gains octoprint (migration 0006); PrusaLink is
stored as prusa, the name the enum already had.

scripts/e2e-vendors.sh adds each kind through the API against a fake REST
printer (scripts/fake-http-printer.ts) and runs a job through the real
scheduler to completion. The fake shares its author's reading of the docs
with the driver, so this proves the plumbing, not the protocols.
2026-10-30 07:49:31 +00:00
octoturge 69b6efea9b Check the P1/A1 camera in real Chrome
BROWSER=1 on e2e-camera-snapshot.sh signs in with the real Supabase test
user, opens the fleet page and checks the snapshot view decodes, refreshes
and has picture content.
2026-10-30 07:37:41 +00:00
octoturge fcde6f5155 Let a job choose its AMS slot; soak test for telemetry
POST /jobs takes an optional amsSlot (stored in the job metadata) which
dispatch passes to the gateway. The camera e2e script gains SOAK_S.
2026-10-30 06:56:24 +00:00
octoturge 58d5b013bc Let the real-print script use a LAN object store 2026-10-29 22:13:47 +00:00
octoturge 1f9fd08b40 Add a gated script for the first real print
scripts/real-print-p1p.sh queues a job in a throwaway one-printer farm and
watches the machine print it. Refuses without I_AUTHORISE_A_REAL_PRINT=yes.
2026-10-29 16:20:04 +00:00
octoturge a190c896da Relay P1/A1 camera frames to viewers
GET /printers/:id/snapshot asks the farm's gateway for one JPEG frame
(camera_snapshot_request) and returns it no-store. Concurrent viewers
share one read and a frame under 800 ms old is reused, because the
printer's camera serves one client. /stream answers kind "snapshot" for
these printers; printers.camera_snapshot is set from telemetry.

scripts/e2e-camera-snapshot.sh reads real frames from the P1P through
the real proxy and backend.
2026-10-29 16:11:10 +00:00
35 changed files with 4618 additions and 40 deletions
+1
View File
@@ -0,0 +1 @@
ALTER TABLE "printers" ADD COLUMN "camera_snapshot" boolean DEFAULT false NOT NULL;
+1
View File
@@ -0,0 +1 @@
ALTER TYPE "printer_vendor" ADD VALUE 'octoprint';
+40
View File
@@ -0,0 +1,40 @@
CREATE TABLE IF NOT EXISTS "enrolment_codes" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"farm_id" uuid NOT NULL,
"code_hash" text NOT NULL,
"created_by" uuid NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"expires_at" timestamp with time zone NOT NULL,
"used_at" timestamp with time zone,
CONSTRAINT "enrolment_codes_code_hash_unique" UNIQUE("code_hash")
);
--> statement-breakpoint
CREATE TABLE IF NOT EXISTS "org_invitations" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"org_id" uuid NOT NULL,
"email" text NOT NULL,
"role" "org_role" NOT NULL,
"token_hash" text NOT NULL,
"invited_by" uuid NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
"expires_at" timestamp with time zone NOT NULL,
"accepted_at" timestamp with time zone,
"accepted_by" uuid,
CONSTRAINT "org_invitations_token_hash_unique" UNIQUE("token_hash")
);
--> statement-breakpoint
ALTER TABLE "org_members" ADD COLUMN "email" text;--> statement-breakpoint
DO $$ BEGIN
ALTER TABLE "enrolment_codes" ADD CONSTRAINT "enrolment_codes_farm_id_farms_id_fk" FOREIGN KEY ("farm_id") REFERENCES "public"."farms"("id") ON DELETE cascade ON UPDATE no action;
EXCEPTION
WHEN duplicate_object THEN null;
END $$;
--> statement-breakpoint
DO $$ BEGIN
ALTER TABLE "org_invitations" ADD CONSTRAINT "org_invitations_org_id_organisations_id_fk" FOREIGN KEY ("org_id") REFERENCES "public"."organisations"("id") ON DELETE cascade ON UPDATE no action;
EXCEPTION
WHEN duplicate_object THEN null;
END $$;
--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "enrolment_codes_farm_idx" ON "enrolment_codes" USING btree ("farm_id");--> statement-breakpoint
CREATE INDEX IF NOT EXISTS "org_invitations_org_idx" ON "org_invitations" USING btree ("org_id");
+848
View File
@@ -0,0 +1,848 @@
{
"id": "aef1c8fd-6a0a-48b5-9957-26b3630990f3",
"prevId": "ee6e6f93-7aec-431b-955d-75c485352ea7",
"version": "7",
"dialect": "postgresql",
"tables": {
"public.org_members": {
"name": "org_members",
"schema": "",
"columns": {
"org_id": {
"name": "org_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"user_id": {
"name": "user_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"role": {
"name": "role",
"type": "org_role",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'viewer'"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"org_members_user_id_idx": {
"name": "org_members_user_id_idx",
"columns": [
{
"expression": "user_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"org_members_org_id_organisations_id_fk": {
"name": "org_members_org_id_organisations_id_fk",
"tableFrom": "org_members",
"tableTo": "organisations",
"columnsFrom": [
"org_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {
"org_members_org_id_user_id_pk": {
"name": "org_members_org_id_user_id_pk",
"columns": [
"org_id",
"user_id"
]
}
},
"uniqueConstraints": {}
},
"public.organisations": {
"name": "organisations",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"settings": {
"name": "settings",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {},
"foreignKeys": {},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.farms": {
"name": "farms",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"org_id": {
"name": "org_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"edge_proxy_token": {
"name": "edge_proxy_token",
"type": "text",
"primaryKey": false,
"notNull": true
},
"location": {
"name": "location",
"type": "text",
"primaryKey": false,
"notNull": false
},
"settings": {
"name": "settings",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"farms_org_id_idx": {
"name": "farms_org_id_idx",
"columns": [
{
"expression": "org_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"farms_org_id_organisations_id_fk": {
"name": "farms_org_id_organisations_id_fk",
"tableFrom": "farms",
"tableTo": "organisations",
"columnsFrom": [
"org_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.printers": {
"name": "printers",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"vendor": {
"name": "vendor",
"type": "printer_vendor",
"typeSchema": "public",
"primaryKey": false,
"notNull": true
},
"status": {
"name": "status",
"type": "printer_status",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'offline'"
},
"lan_address": {
"name": "lan_address",
"type": "text",
"primaryKey": false,
"notNull": false
},
"serial_number": {
"name": "serial_number",
"type": "text",
"primaryKey": false,
"notNull": false
},
"gateway_printer_id": {
"name": "gateway_printer_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"nozzle_temp_c": {
"name": "nozzle_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"nozzle_target_temp_c": {
"name": "nozzle_target_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"bed_temp_c": {
"name": "bed_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"bed_target_temp_c": {
"name": "bed_target_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"progress_percent": {
"name": "progress_percent",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"current_layer": {
"name": "current_layer",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"total_layers": {
"name": "total_layers",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"remaining_minutes": {
"name": "remaining_minutes",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"current_job_name": {
"name": "current_job_name",
"type": "text",
"primaryKey": false,
"notNull": false
},
"active_alerts": {
"name": "active_alerts",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"camera_stream": {
"name": "camera_stream",
"type": "text",
"primaryKey": false,
"notNull": false
},
"camera_snapshot": {
"name": "camera_snapshot",
"type": "boolean",
"primaryKey": false,
"notNull": true,
"default": false
},
"vendor_state": {
"name": "vendor_state",
"type": "text",
"primaryKey": false,
"notNull": false
},
"vendor_error_code": {
"name": "vendor_error_code",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"metadata": {
"name": "metadata",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"last_seen_at": {
"name": "last_seen_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"printers_farm_gateway_printer_id_idx": {
"name": "printers_farm_gateway_printer_id_idx",
"columns": [
{
"expression": "farm_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "gateway_printer_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": true,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"printers_farm_id_farms_id_fk": {
"name": "printers_farm_id_farms_id_fk",
"tableFrom": "printers",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.print_jobs": {
"name": "print_jobs",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"file_name": {
"name": "file_name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"r2_object_key": {
"name": "r2_object_key",
"type": "text",
"primaryKey": false,
"notNull": true
},
"status": {
"name": "status",
"type": "print_job_status",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'queued'"
},
"progress_pct": {
"name": "progress_pct",
"type": "real",
"primaryKey": false,
"notNull": true,
"default": 0
},
"estimated_time_s": {
"name": "estimated_time_s",
"type": "real",
"primaryKey": false,
"notNull": false
},
"elapsed_time_s": {
"name": "elapsed_time_s",
"type": "real",
"primaryKey": false,
"notNull": false
},
"submitted_by": {
"name": "submitted_by",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"metadata": {
"name": "metadata",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"started_at": {
"name": "started_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"completed_at": {
"name": "completed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {},
"foreignKeys": {
"print_jobs_farm_id_farms_id_fk": {
"name": "print_jobs_farm_id_farms_id_fk",
"tableFrom": "print_jobs",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"print_jobs_printer_id_printers_id_fk": {
"name": "print_jobs_printer_id_printers_id_fk",
"tableFrom": "print_jobs",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "set null",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.frame_embeddings": {
"name": "frame_embeddings",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"print_job_id": {
"name": "print_job_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"r2_object_key": {
"name": "r2_object_key",
"type": "text",
"primaryKey": false,
"notNull": true
},
"embedding": {
"name": "embedding",
"type": "vector(512)",
"primaryKey": false,
"notNull": true
},
"captured_at": {
"name": "captured_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"frame_embeddings_embedding_ivfflat_idx": {
"name": "frame_embeddings_embedding_ivfflat_idx",
"columns": [
{
"expression": "embedding",
"isExpression": false,
"asc": true,
"nulls": "last",
"opclass": "vector_cosine_ops"
}
],
"isUnique": false,
"concurrently": false,
"method": "ivfflat",
"with": {
"lists": 100
}
}
},
"foreignKeys": {
"frame_embeddings_print_job_id_print_jobs_id_fk": {
"name": "frame_embeddings_print_job_id_print_jobs_id_fk",
"tableFrom": "frame_embeddings",
"tableTo": "print_jobs",
"columnsFrom": [
"print_job_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"frame_embeddings_printer_id_printers_id_fk": {
"name": "frame_embeddings_printer_id_printers_id_fk",
"tableFrom": "frame_embeddings",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.printer_alerts": {
"name": "printer_alerts",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"kind": {
"name": "kind",
"type": "alert_kind",
"typeSchema": "public",
"primaryKey": false,
"notNull": true
},
"message": {
"name": "message",
"type": "text",
"primaryKey": false,
"notNull": true
},
"confidence": {
"name": "confidence",
"type": "real",
"primaryKey": false,
"notNull": false
},
"object_key": {
"name": "object_key",
"type": "text",
"primaryKey": false,
"notNull": false
},
"acknowledged_by": {
"name": "acknowledged_by",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"acknowledged_at": {
"name": "acknowledged_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"printer_alerts_farm_open_idx": {
"name": "printer_alerts_farm_open_idx",
"columns": [
{
"expression": "farm_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "acknowledged_at",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"printer_alerts_farm_id_farms_id_fk": {
"name": "printer_alerts_farm_id_farms_id_fk",
"tableFrom": "printer_alerts",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"printer_alerts_printer_id_printers_id_fk": {
"name": "printer_alerts_printer_id_printers_id_fk",
"tableFrom": "printer_alerts",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "set null",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
}
},
"enums": {
"public.org_role": {
"name": "org_role",
"schema": "public",
"values": [
"viewer",
"operator",
"owner"
]
},
"public.printer_status": {
"name": "printer_status",
"schema": "public",
"values": [
"offline",
"idle",
"printing",
"paused",
"error",
"maintenance"
]
},
"public.printer_vendor": {
"name": "printer_vendor",
"schema": "public",
"values": [
"bambu",
"prusa",
"klipper"
]
},
"public.print_job_status": {
"name": "print_job_status",
"schema": "public",
"values": [
"queued",
"uploading",
"printing",
"paused",
"completed",
"failed",
"cancelled"
]
},
"public.alert_kind": {
"name": "alert_kind",
"schema": "public",
"values": [
"print_failed",
"print_paused",
"printer_error",
"ai_detection"
]
}
},
"schemas": {},
"sequences": {},
"_meta": {
"columns": {},
"schemas": {},
"tables": {}
}
}
+849
View File
@@ -0,0 +1,849 @@
{
"id": "97bbd615-092b-4859-9b89-de9c6295a4f9",
"prevId": "aef1c8fd-6a0a-48b5-9957-26b3630990f3",
"version": "7",
"dialect": "postgresql",
"tables": {
"public.org_members": {
"name": "org_members",
"schema": "",
"columns": {
"org_id": {
"name": "org_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"user_id": {
"name": "user_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"role": {
"name": "role",
"type": "org_role",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'viewer'"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"org_members_user_id_idx": {
"name": "org_members_user_id_idx",
"columns": [
{
"expression": "user_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"org_members_org_id_organisations_id_fk": {
"name": "org_members_org_id_organisations_id_fk",
"tableFrom": "org_members",
"tableTo": "organisations",
"columnsFrom": [
"org_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {
"org_members_org_id_user_id_pk": {
"name": "org_members_org_id_user_id_pk",
"columns": [
"org_id",
"user_id"
]
}
},
"uniqueConstraints": {}
},
"public.organisations": {
"name": "organisations",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"settings": {
"name": "settings",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {},
"foreignKeys": {},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.farms": {
"name": "farms",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"org_id": {
"name": "org_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"edge_proxy_token": {
"name": "edge_proxy_token",
"type": "text",
"primaryKey": false,
"notNull": true
},
"location": {
"name": "location",
"type": "text",
"primaryKey": false,
"notNull": false
},
"settings": {
"name": "settings",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"farms_org_id_idx": {
"name": "farms_org_id_idx",
"columns": [
{
"expression": "org_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"farms_org_id_organisations_id_fk": {
"name": "farms_org_id_organisations_id_fk",
"tableFrom": "farms",
"tableTo": "organisations",
"columnsFrom": [
"org_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.printers": {
"name": "printers",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"name": {
"name": "name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"vendor": {
"name": "vendor",
"type": "printer_vendor",
"typeSchema": "public",
"primaryKey": false,
"notNull": true
},
"status": {
"name": "status",
"type": "printer_status",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'offline'"
},
"lan_address": {
"name": "lan_address",
"type": "text",
"primaryKey": false,
"notNull": false
},
"serial_number": {
"name": "serial_number",
"type": "text",
"primaryKey": false,
"notNull": false
},
"gateway_printer_id": {
"name": "gateway_printer_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"nozzle_temp_c": {
"name": "nozzle_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"nozzle_target_temp_c": {
"name": "nozzle_target_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"bed_temp_c": {
"name": "bed_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"bed_target_temp_c": {
"name": "bed_target_temp_c",
"type": "real",
"primaryKey": false,
"notNull": false
},
"progress_percent": {
"name": "progress_percent",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"current_layer": {
"name": "current_layer",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"total_layers": {
"name": "total_layers",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"remaining_minutes": {
"name": "remaining_minutes",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"current_job_name": {
"name": "current_job_name",
"type": "text",
"primaryKey": false,
"notNull": false
},
"active_alerts": {
"name": "active_alerts",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"camera_stream": {
"name": "camera_stream",
"type": "text",
"primaryKey": false,
"notNull": false
},
"camera_snapshot": {
"name": "camera_snapshot",
"type": "boolean",
"primaryKey": false,
"notNull": true,
"default": false
},
"vendor_state": {
"name": "vendor_state",
"type": "text",
"primaryKey": false,
"notNull": false
},
"vendor_error_code": {
"name": "vendor_error_code",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"metadata": {
"name": "metadata",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"last_seen_at": {
"name": "last_seen_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"printers_farm_gateway_printer_id_idx": {
"name": "printers_farm_gateway_printer_id_idx",
"columns": [
{
"expression": "farm_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "gateway_printer_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": true,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"printers_farm_id_farms_id_fk": {
"name": "printers_farm_id_farms_id_fk",
"tableFrom": "printers",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.print_jobs": {
"name": "print_jobs",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"file_name": {
"name": "file_name",
"type": "text",
"primaryKey": false,
"notNull": true
},
"r2_object_key": {
"name": "r2_object_key",
"type": "text",
"primaryKey": false,
"notNull": true
},
"status": {
"name": "status",
"type": "print_job_status",
"typeSchema": "public",
"primaryKey": false,
"notNull": true,
"default": "'queued'"
},
"progress_pct": {
"name": "progress_pct",
"type": "real",
"primaryKey": false,
"notNull": true,
"default": 0
},
"estimated_time_s": {
"name": "estimated_time_s",
"type": "real",
"primaryKey": false,
"notNull": false
},
"elapsed_time_s": {
"name": "elapsed_time_s",
"type": "real",
"primaryKey": false,
"notNull": false
},
"submitted_by": {
"name": "submitted_by",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"metadata": {
"name": "metadata",
"type": "jsonb",
"primaryKey": false,
"notNull": false,
"default": "'{}'::jsonb"
},
"started_at": {
"name": "started_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"completed_at": {
"name": "completed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"updated_at": {
"name": "updated_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {},
"foreignKeys": {
"print_jobs_farm_id_farms_id_fk": {
"name": "print_jobs_farm_id_farms_id_fk",
"tableFrom": "print_jobs",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"print_jobs_printer_id_printers_id_fk": {
"name": "print_jobs_printer_id_printers_id_fk",
"tableFrom": "print_jobs",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "set null",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.frame_embeddings": {
"name": "frame_embeddings",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"print_job_id": {
"name": "print_job_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"r2_object_key": {
"name": "r2_object_key",
"type": "text",
"primaryKey": false,
"notNull": true
},
"embedding": {
"name": "embedding",
"type": "vector(512)",
"primaryKey": false,
"notNull": true
},
"captured_at": {
"name": "captured_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"frame_embeddings_embedding_ivfflat_idx": {
"name": "frame_embeddings_embedding_ivfflat_idx",
"columns": [
{
"expression": "embedding",
"isExpression": false,
"asc": true,
"nulls": "last",
"opclass": "vector_cosine_ops"
}
],
"isUnique": false,
"concurrently": false,
"method": "ivfflat",
"with": {
"lists": 100
}
}
},
"foreignKeys": {
"frame_embeddings_print_job_id_print_jobs_id_fk": {
"name": "frame_embeddings_print_job_id_print_jobs_id_fk",
"tableFrom": "frame_embeddings",
"tableTo": "print_jobs",
"columnsFrom": [
"print_job_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"frame_embeddings_printer_id_printers_id_fk": {
"name": "frame_embeddings_printer_id_printers_id_fk",
"tableFrom": "frame_embeddings",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
},
"public.printer_alerts": {
"name": "printer_alerts",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"farm_id": {
"name": "farm_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"printer_id": {
"name": "printer_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"kind": {
"name": "kind",
"type": "alert_kind",
"typeSchema": "public",
"primaryKey": false,
"notNull": true
},
"message": {
"name": "message",
"type": "text",
"primaryKey": false,
"notNull": true
},
"confidence": {
"name": "confidence",
"type": "real",
"primaryKey": false,
"notNull": false
},
"object_key": {
"name": "object_key",
"type": "text",
"primaryKey": false,
"notNull": false
},
"acknowledged_by": {
"name": "acknowledged_by",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"acknowledged_at": {
"name": "acknowledged_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"printer_alerts_farm_open_idx": {
"name": "printer_alerts_farm_open_idx",
"columns": [
{
"expression": "farm_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "acknowledged_at",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"printer_alerts_farm_id_farms_id_fk": {
"name": "printer_alerts_farm_id_farms_id_fk",
"tableFrom": "printer_alerts",
"tableTo": "farms",
"columnsFrom": [
"farm_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
},
"printer_alerts_printer_id_printers_id_fk": {
"name": "printer_alerts_printer_id_printers_id_fk",
"tableFrom": "printer_alerts",
"tableTo": "printers",
"columnsFrom": [
"printer_id"
],
"columnsTo": [
"id"
],
"onDelete": "set null",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {}
}
},
"enums": {
"public.org_role": {
"name": "org_role",
"schema": "public",
"values": [
"viewer",
"operator",
"owner"
]
},
"public.printer_status": {
"name": "printer_status",
"schema": "public",
"values": [
"offline",
"idle",
"printing",
"paused",
"error",
"maintenance"
]
},
"public.printer_vendor": {
"name": "printer_vendor",
"schema": "public",
"values": [
"bambu",
"prusa",
"klipper",
"octoprint"
]
},
"public.print_job_status": {
"name": "print_job_status",
"schema": "public",
"values": [
"queued",
"uploading",
"printing",
"paused",
"completed",
"failed",
"cancelled"
]
},
"public.alert_kind": {
"name": "alert_kind",
"schema": "public",
"values": [
"print_failed",
"print_paused",
"printer_error",
"ai_detection"
]
}
},
"schemas": {},
"sequences": {},
"_meta": {
"columns": {},
"schemas": {},
"tables": {}
}
}
File diff suppressed because it is too large Load Diff
+21
View File
@@ -36,6 +36,27 @@
"when": 1793285676878,
"tag": "0004_printer_alerts",
"breakpoints": true
},
{
"idx": 5,
"version": "7",
"when": 1793289367708,
"tag": "0005_bouncy_cannonball",
"breakpoints": true
},
{
"idx": 6,
"version": "7",
"when": 1793346305671,
"tag": "0006_octoprint_vendor",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1793347631289,
"tag": "0007_onboarding",
"breakpoints": true
}
]
}
+116
View File
@@ -0,0 +1,116 @@
/**
* The P1/A1 camera in the real web console, in a real Chrome, for
* scripts/e2e-camera-snapshot.sh (BROWSER=1). Signs in the same way as
* e2e-camera-browser.ts. Reads pixel statistics and keeps no screenshot: the
* frame is the inside of someone's workshop.
*
* Results are `ok<TAB>label` / `FAILED<TAB>label`.
* Environment: APP_URL, SUPABASE_E2E_EMAIL, SUPABASE_E2E_PASSWORD, CHROME_BIN.
*/
import puppeteer from "puppeteer-core";
const appUrl = process.env.APP_URL!;
const email = process.env.SUPABASE_E2E_EMAIL!;
const password = process.env.SUPABASE_E2E_PASSWORD!;
const chromeBin = process.env.CHROME_BIN ?? "/usr/bin/google-chrome";
let failed = false;
const say = (good: boolean, label: string, detail = "") => {
if (!good) failed = true;
console.log(`${good ? "ok" : "FAILED"}\t${label}${detail ? ` (${detail})` : ""}`);
};
const browser = await puppeteer.launch({
executablePath: chromeBin,
headless: true,
args: ["--no-sandbox", "--disable-dev-shm-usage"],
});
try {
const page = await browser.newPage();
const pageErrors: string[] = [];
page.on("pageerror", (err) => pageErrors.push(String((err as Error).message ?? err).slice(0, 160)));
await page.goto(`${appUrl}/login`, { waitUntil: "domcontentloaded", timeout: 120_000 });
await page.waitForFunction(
() => typeof (window as any).useNuxtApp === "function" && Boolean((window as any).useNuxtApp().$supabase),
{ timeout: 120_000 },
);
const signInError = await page.evaluate(
async (e: string, p: string) => {
const { client } = (window as any).useNuxtApp().$supabase;
const { error } = await client.auth.signInWithPassword({ email: e, password: p });
return error ? String(error.message) : null;
},
email,
password,
);
say(signInError === null, "the app's own Supabase client signs the user in", signInError ?? "");
if (signInError) process.exit(1);
await page.goto(`${appUrl}/`, { waitUntil: "domcontentloaded", timeout: 60_000 });
const stayed = await page.waitForSelector(".fleet", { timeout: 60_000 }).then(() => true).catch(() => false);
say(stayed && !page.url().includes("/login"), "the fleet page loads and does not bounce back to /login");
const live = await page
.waitForFunction(() => document.querySelector(".app-shell__status.is-online") !== null, { timeout: 30_000 })
.then(() => true)
.catch(() => false);
say(live, "the header reports Live");
const shown = await page
.waitForFunction(() => document.querySelector(".snapshot-view__status.is-connected") !== null, { timeout: 60_000 })
.then(() => true)
.catch(() => false);
const status = await page.evaluate(
() => document.querySelector(".snapshot-view__status")?.textContent?.trim() ?? "no snapshot view rendered",
);
say(shown, "the snapshot view reports connected", `status: ${status}`);
say(
(await page.$(".webrtc-view")) === null,
"no WebRTC player is rendered for a snapshot camera",
);
if (shown) {
// Two different frames, a few seconds apart: the picture is refreshing.
const result = await page.evaluate(async () => {
const img = document.querySelector<HTMLImageElement>(".snapshot-view img")!;
const first = img.src;
const t0 = Date.now();
while (img.src === first && Date.now() - t0 < 15_000) await new Promise((r) => setTimeout(r, 250));
const refreshed = img.src !== first;
const canvas = document.createElement("canvas");
canvas.width = 64;
canvas.height = 36;
const ctx = canvas.getContext("2d")!;
ctx.drawImage(img, 0, 0, canvas.width, canvas.height);
const px = ctx.getImageData(0, 0, canvas.width, canvas.height).data;
let sum = 0;
let sumSq = 0;
const n = px.length / 4;
for (let i = 0; i < px.length; i += 4) {
const y = 0.299 * px[i] + 0.587 * px[i + 1] + 0.114 * px[i + 2];
sum += y;
sumSq += y * y;
}
const mean = sum / n;
return {
refreshed,
width: img.naturalWidth,
height: img.naturalHeight,
mean,
stdev: Math.sqrt(Math.max(0, sumSq / n - mean * mean)),
};
});
say(result.width > 0, "Chrome decoded the JPEG", `${result.width}x${result.height}`);
say(result.refreshed, "the picture refreshes by itself");
say(result.stdev > 2, "the frame has picture content, not black or one flat colour",
`mean luma ${result.mean.toFixed(0)}, stdev ${result.stdev.toFixed(1)}`);
}
say(pageErrors.length === 0, "the page raised no uncaught errors", pageErrors.slice(0, 2).join(" | "));
} finally {
await browser.close();
}
process.exit(failed ? 1 : 0);
+215
View File
@@ -0,0 +1,215 @@
#!/usr/bin/env bash
# A P1/A1 camera, end to end: the real proxy reads a JPEG frame from the real
# printer's port 6000 (read-only - it only watches), the real backend relays
# it, and an authenticated viewer fetches it.
#
# Needs a printer in ../continuum-proxy/printers.toml whose camera is the
# P1/A1 kind (CAMERA_ID, default p1p) and that is switched on. Touches the
# printer's camera and MQTT status, nothing else; no command is sent.
set -euo pipefail
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
CAMERA_ID="${CAMERA_ID:-p1p}"
GO2RTC_BIN="${GO2RTC_BIN:-$HOME/.local/bin/go2rtc}"
PROXY_DIR="${PROXY_DIR:-$REPO_ROOT/../continuum-proxy}"
PORT="${PORT:-3050}"
AUTH_PORT="${AUTH_PORT:-3051}"
BASE="http://localhost:$PORT"
DATABASE_URL="${DATABASE_URL:-postgres://continuum:continuum@localhost:5432/continuum}"
VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
EDGE_JWT_SECRET="${EDGE_JWT_SECRET:-e2e-edge-secret}"
R2_ACCESS_KEY_ID="${R2_ACCESS_KEY_ID:-e2e}"
R2_SECRET_ACCESS_KEY="${R2_SECRET_ACCESS_KEY:-e2e}"
R2_ENDPOINT="${R2_ENDPOINT:-http://localhost:9000}"
export DATABASE_URL VALKEY_URL EDGE_JWT_SECRET PORT R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY R2_ENDPOINT
# Where to keep the last frame for a human to look at. Unset: not kept.
SAVE_FRAME="${SAVE_FRAME:-}"
# BROWSER=1: also open the real web console in real Chrome (needs the real
# Supabase test user, so source ../dev-secrets.sh).
BROWSER="${BROWSER:-0}"
APP_HOST="${APP_HOST:-127.0.0.1}"
APP_DIR="${APP_DIR:-$REPO_ROOT/../continuum-app}"
APP_PORT="${APP_PORT:-3052}"
CHROME_BIN="${CHROME_BIN:-/usr/bin/google-chrome}"
APP_PID=""
WORK="$(mktemp -d)"
BACKEND_PID=""; PROXY_PID=""; AUTH_PID=""
fail() { echo "FAIL: $*" >&2; exit 1; }
ok() { echo " ok: $*"; }
psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; }
USER_ID="$(cat /proc/sys/kernel/random/uuid)"
cleanup() {
local rc=$?
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill "$pid" 2>/dev/null || true; done
[ -n "$APP_PID" ] && kill -- "-$APP_PID" 2>/dev/null || true
for _ in $(seq 1 20); do [ -n "$PROXY_PID" ] && kill -0 "$PROXY_PID" 2>/dev/null || break; sleep 0.3; done
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true; done
[ -n "$APP_PID" ] && kill -9 -- "-$APP_PID" 2>/dev/null || true
[ -n "${ORG_ID:-}" ] && psql_q "DELETE FROM organisations WHERE id = '$ORG_ID'" >/dev/null 2>&1 || true
if [ $rc -ne 0 ]; then
echo; echo "--- backend log (tail) ---"; tail -20 "$WORK/backend.log" 2>/dev/null || true
echo "--- proxy log (tail) ---"; tail -30 "$WORK/proxy.log" 2>/dev/null || true
fi
rm -rf "$WORK"
sleep 0.5
if ps -eo comm | grep -qx go2rtc; then
echo "WARNING: a go2rtc process is still running after cleanup" >&2; [ $rc -eq 0 ] && rc=1
fi
exit $rc
}
trap cleanup EXIT
echo "==> checking preconditions"
psql_q "SELECT camera_snapshot FROM printers LIMIT 0" >/dev/null 2>&1 \
|| fail "database lacks printers.camera_snapshot - run scripts/dev-env.sh schema"
[ -x "$GO2RTC_BIN" ] || fail "no go2rtc at $GO2RTC_BIN"
[ -f "$PROXY_DIR/printers.toml" ] || fail "no $PROXY_DIR/printers.toml"
grep -q "^id *= *\"$CAMERA_ID\"" "$PROXY_DIR/printers.toml" || fail "printers.toml has no printer '$CAMERA_ID'"
ps -eo comm | grep -qx go2rtc && fail "a go2rtc is already running"
CAM_HOST="$(awk -v id="$CAMERA_ID" -F'"' '/^id *=/ {cur=$2} cur==id && /^(host|address|ip) *=/ {print $2; exit}' "$PROXY_DIR/printers.toml")"
if [ -n "$CAM_HOST" ]; then
timeout 4 bash -c "</dev/tcp/$CAM_HOST/6000" 2>/dev/null || fail "$CAMERA_ID ($CAM_HOST) refuses TCP 6000: no snapshot camera, or it is off"
ok "$CAMERA_ID accepts connections on 6000"
fi
if [ "$BROWSER" = "1" ]; then
for name in SUPABASE_URL SUPABASE_SECRET_KEY SUPABASE_PUBLISHABLE_KEY; do
[ -n "${!name:-}" ] || fail "$name is not set - source ../dev-secrets.sh (BROWSER=1)"
done
[ -x "$CHROME_BIN" ] || fail "no Chrome at $CHROME_BIN"
[ -d "$APP_DIR/apps/web/node_modules" ] || fail "$APP_DIR has no installed web app"
SB_USER="$(bun scripts/e2e-supabase-user.ts)" || fail "could not provision the Supabase test user"
sb_field() { printf '%s' "$SB_USER" | python3 -c 'import json,sys; print(json.load(sys.stdin)[sys.argv[1]])' "$1"; }
USER_ID="$(sb_field id)"; USER_TOKEN="$(sb_field accessToken)"; SB_EMAIL="$(sb_field email)"; SB_PASSWORD="$(sb_field password)"
unset SB_USER
ok "signed in as the real Supabase test user"
fi
echo "==> seeding an organisation and a farm claiming '$CAMERA_ID'"
EDGE_PROXY_TOKEN="e2e-$(date +%s)"
ORG_ID="$(psql_q "INSERT INTO organisations (name) VALUES ('e2e snapshot org') RETURNING id")"
psql_q "INSERT INTO org_members (org_id, user_id, role) VALUES ('$ORG_ID', '$USER_ID', 'viewer')" >/dev/null
FARM_ID="$(psql_q "INSERT INTO farms (name, org_id, edge_proxy_token) VALUES ('e2e snapshot farm', '$ORG_ID', '$EDGE_PROXY_TOKEN') RETURNING id")"
psql_q "INSERT INTO printers (farm_id, name, vendor, gateway_printer_id) VALUES ('$FARM_ID', 'snapshot under test', 'bambu', '$CAMERA_ID')" >/dev/null
CAM_UUID="$(psql_q "SELECT id FROM printers WHERE farm_id='$FARM_ID'")"
echo "==> starting the auth issuer and the backend"
if [ "$BROWSER" != "1" ]; then
bun scripts/dev-auth.ts --port "$AUTH_PORT" > "$WORK/auth.log" 2>&1 &
AUTH_PID=$!
AUTH_URL="http://localhost:$AUTH_PORT"
for _ in $(seq 1 40); do curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 && break; sleep 0.25; done
export SUPABASE_URL="$AUTH_URL"
USER_TOKEN="$(curl -sf "$AUTH_URL/token?sub=$USER_ID" | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])')"
fi
DISPATCH_ENABLED=false bun run src/index.ts > "$WORK/backend.log" 2>&1 &
BACKEND_PID=$!
for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && break; sleep 0.5; done
curl -sf "$BASE/health" >/dev/null || fail "backend did not become healthy"
echo "==> starting the real continuum-proxy"
( cd "$PROXY_DIR" && cargo build --quiet 2>&1 | tail -3 ) || fail "proxy did not build"
env -C "$PROXY_DIR" \
CONTINUUM_GATEWAY_ID=e2e-snap-gw CONTINUUM_FARM_ID="$FARM_ID" CONTINUUM_GATEWAY_TOKEN="$EDGE_PROXY_TOKEN" \
CONTINUUM_UPLINK_URL="ws://localhost:$PORT/ws/edge/v1" \
CONTINUUM_GO2RTC_BIN="$GO2RTC_BIN" CONTINUUM_GO2RTC_CONFIG="$WORK/go2rtc.yaml" \
CONTINUUM_PRINTERS_PATH=printers.toml CONTINUUM_STORE_PATH="$WORK/continuum.db" RUST_LOG=info \
"$PROXY_DIR/target/debug/continuum-proxy" > "$WORK/proxy.log" 2>&1 &
PROXY_PID=$!
echo "==> waiting for the camera to be advertised as a snapshot camera"
STREAM=""
for _ in $(seq 1 45); do
STREAM="$(curl -sf -H "authorization: Bearer $USER_TOKEN" "$BASE/printers/$CAM_UUID/stream" 2>/dev/null || true)"
[ -n "$STREAM" ] && break
sleep 1
done
[ -n "$STREAM" ] || fail "GET /printers/$CAM_UUID/stream never answered 200"
printf '%s' "$STREAM" | python3 -c 'import json,sys; d=json.load(sys.stdin); assert d["kind"]=="snapshot" and "iceServers" not in d, d' \
|| fail "the stream answer is not a snapshot answer"
ok "the stream endpoint says kind=snapshot (no WebRTC session offered)"
echo "==> fetching frames"
check_frame() {
local file="$1" label="$2" size
size="$(stat -c %s "$file")"
[ "$size" -gt 1000 ] || fail "$label: only $size bytes"
[ "$(head -c2 "$file" | od -An -tx1 | tr -d ' \n')" = "ffd8" ] || fail "$label: does not start with a JPEG marker"
[ "$(tail -c2 "$file" | od -An -tx1 | tr -d ' \n')" = "ffd9" ] || fail "$label: does not end with a JPEG marker"
ok "$label: a $size-byte JPEG"
}
fetch() { curl -s -o "$1" -D "$1.h" -w '%{http_code}' -H "authorization: Bearer $USER_TOKEN" "$BASE/printers/$CAM_UUID/snapshot"; }
code="$(fetch "$WORK/f1.jpg")"
[ "$code" = 200 ] || fail "first frame: HTTP $code: $(head -c 300 "$WORK/f1.jpg")"
grep -qi '^content-type: image/jpeg' "$WORK/f1.jpg.h" || fail "wrong content-type"
grep -qi '^cache-control: no-store' "$WORK/f1.jpg.h" || fail "frames must be no-store"
check_frame "$WORK/f1.jpg" "first frame"
sleep 1.2
code="$(fetch "$WORK/f2.jpg")"; [ "$code" = 200 ] || fail "second frame: HTTP $code"
check_frame "$WORK/f2.jpg" "second frame (a fresh read after the cache window)"
cmp -s "$WORK/f1.jpg" "$WORK/f2.jpg" && fail "two frames a second apart are byte-identical - is the cache serving stale pictures?"
ok "consecutive frames differ"
echo "==> five viewers at once share the camera"
viewers=()
for i in 1 2 3 4 5; do fetch "$WORK/c$i.jpg" > "$WORK/c$i.code" & viewers+=($!); done
# Not a bare `wait`: that would also wait for the backend and the proxy.
for pid in "${viewers[@]}"; do wait "$pid"; done
for i in 1 2 3 4 5; do [ "$(cat "$WORK/c$i.code")" = 200 ] || fail "viewer $i: HTTP $(cat "$WORK/c$i.code")"; check_frame "$WORK/c$i.jpg" "viewer $i"; done
ok "all five got a picture"
echo "==> access control"
[ "$(curl -s -o /dev/null -w '%{http_code}' "$BASE/printers/$CAM_UUID/snapshot")" = 401 ] || fail "no token must be 401"
ok "no token: 401"
# The code is never read out of printers.toml here; the camera login happens
# on the gateway, and the only thing that should cross is the picture.
if grep -qi 'access_code\|accessCode' "$WORK/backend.log"; then
fail "the backend log mentions an access code"
fi
ok "the backend log never mentions an access code"
# Optional soak: the first real dispatch froze the proxy ~4.5 min in (an
# MQTT publish that never completed while the fleet lock was held). Staying
# connected well past that and still receiving telemetry is the regression test.
if [ -n "${SOAK_S:-}" ]; then
echo "==> soaking for ${SOAK_S}s: telemetry must keep arriving"
sleep "$SOAK_S"
AGE="$(psql_q "SELECT extract(epoch FROM now() - last_seen_at)::int FROM printers WHERE id='$CAM_UUID'")"
[ -n "$AGE" ] && [ "$AGE" -lt 20 ] || fail "telemetry is ${AGE:-never}s old after the soak - the proxy froze"
ok "telemetry still fresh (${AGE}s old) after ${SOAK_S}s"
code="$(fetch "$WORK/after.jpg")"; [ "$code" = 200 ] || fail "no camera frame after the soak: HTTP $code"
ok "camera still answers after the soak"
fi
if [ "$BROWSER" = "1" ]; then
echo "==> the real console in real Chrome"
setsid env -C "$APP_DIR/apps/web" NUXT_PUBLIC_API_BASE="$BASE" NUXT_PUBLIC_WS_BASE="ws://localhost:$PORT" \
SUPABASE_URL="$SUPABASE_URL" SUPABASE_KEY="$SUPABASE_PUBLISHABLE_KEY" NUXT_PUBLIC_PASSWORD_LOGIN=true \
pnpm exec nuxt dev --host "$APP_HOST" --port "$APP_PORT" > "$WORK/app.log" 2>&1 &
APP_PID=$!
for _ in $(seq 1 120); do curl -sf -o /dev/null "http://127.0.0.1:$APP_PORT/login" 2>/dev/null && break; sleep 1; done
curl -sf -o /dev/null "http://127.0.0.1:$APP_PORT/login" || { tail -20 "$WORK/app.log" >&2; fail "the web console did not start"; }
OUT="$(env APP_URL="http://127.0.0.1:$APP_PORT" SUPABASE_E2E_EMAIL="$SB_EMAIL" SUPABASE_E2E_PASSWORD="$SB_PASSWORD" \
CHROME_BIN="$CHROME_BIN" timeout 240 bun scripts/e2e-camera-snapshot-browser.ts 2>&1 || true)"
seen=0; bad=0
while IFS=$'\t' read -r verdict label; do
case "$verdict" in
ok) ok "$label"; seen=$((seen + 1)) ;;
FAILED) echo " FAILED: $label" >&2; seen=$((seen + 1)); bad=$((bad + 1)) ;;
esac
done < <(printf '%s\n' "$OUT" | grep -E "^(ok|FAILED)$(printf '\t')")
if [ "$seen" -lt 8 ] && [ "$bad" -eq 0 ]; then bad=1; echo " FAILED: only $seen of 8 browser results" >&2; fi
if [ "$bad" -ne 0 ]; then
printf '%s\n' "$OUT" | sed 's/^/ | /' >&2; tail -15 "$WORK/app.log" >&2
fail "the real browser did not show the snapshot camera"
fi
fi
[ -n "$SAVE_FRAME" ] && cp "$WORK/f2.jpg" "$SAVE_FRAME" && ok "kept a frame at $SAVE_FRAME"
echo
echo "PASS - a P1/A1 camera frame travelled printer -> proxy -> backend -> an authenticated viewer."
+154
View File
@@ -0,0 +1,154 @@
#!/usr/bin/env bash
# Onboarding end to end, against the real routes and database: inviting
# someone by email, and connecting a gateway with a one-time code instead of
# copying a token. With the proxy built, the last step runs the real
# `continuum-proxy enrol` against the live backend and starts the proxy on
# what it wrote.
#
# Nothing here sends an email: the invitation link is returned to the inviter
# (see the invitations route).
#
# scripts/dev-env.sh up && scripts/dev-env.sh schema
# scripts/e2e-onboarding.sh
set -euo pipefail
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
PORT="${PORT:-3064}"; AUTH_PORT="${AUTH_PORT:-3065}"
BASE="http://localhost:$PORT"
PROXY_DIR="${PROXY_DIR:-$REPO_ROOT/../continuum-proxy}"
DATABASE_URL="${DATABASE_URL:-postgres://continuum:continuum@localhost:5432/continuum}"
VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
EDGE_JWT_SECRET="${EDGE_JWT_SECRET:-e2e-edge-secret}"
export DATABASE_URL VALKEY_URL EDGE_JWT_SECRET PORT
export R2_ACCESS_KEY_ID="${R2_ACCESS_KEY_ID:-e2e}" R2_SECRET_ACCESS_KEY="${R2_SECRET_ACCESS_KEY:-e2e}" R2_ENDPOINT="${R2_ENDPOINT:-http://localhost:9000}"
WORK="$(mktemp -d)"
BACKEND_PID=""; AUTH_PID=""; PROXY_PID=""; ORG_ID=""; FAILURES=0
psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; }
new_uuid() { cat /proc/sys/kernel/random/uuid; }
fail() { echo "FAIL: $*" >&2; exit 1; }
ok() { echo " ok: $*"; }
bad() { echo " FAILED: $*" >&2; FAILURES=$((FAILURES + 1)); }
cleanup() {
local rc=$?
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill "$pid" 2>/dev/null || true; done
sleep 0.5
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true; done
[ -z "${KEEP_DATA:-}" ] && [ -n "$ORG_ID" ] && psql_q "DELETE FROM organisations WHERE id = '$ORG_ID'" >/dev/null 2>&1 || true
if [ $rc -ne 0 ]; then echo "--- backend log (tail) ---"; tail -20 "$WORK/backend.log" 2>/dev/null || true; fi
rm -rf "$WORK"; exit $rc
}
trap cleanup EXIT
req() { # method path token [json] -> status; body in $WORK/last.json
local args=(-s -o "$WORK/last.json" -w '%{http_code}' -X "$1" "$BASE$2")
[ "$3" != "-" ] && args+=(-H "authorization: Bearer $3")
[ -n "${4:-}" ] && args+=(-H 'content-type: application/json' -d "$4")
curl "${args[@]}"
}
expect() { # code method path token [body] -- what
local want="$1" method="$2" path="$3" token="$4"; shift 4
local body=""; [ "${1:-}" != "--" ] && { body="$1"; shift; }; shift
local got; got="$(req "$method" "$path" "$token" "$body")"
[ "$got" = "$want" ] && ok "$* ($got)" || bad "$*: expected $want, got $got - $(head -c 250 "$WORK/last.json")"
}
json() { python3 -c "import json,sys; d=json.load(sys.stdin); print($1)" < "$WORK/last.json"; }
bun scripts/dev-auth.ts --port "$AUTH_PORT" > "$WORK/auth.log" 2>&1 &
AUTH_PID=$!
AUTH_URL="http://localhost:$AUTH_PORT"
for _ in $(seq 1 40); do curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 && break; sleep 0.25; done
export SUPABASE_URL="$AUTH_URL"
mint() { curl -sf "$AUTH_URL/token?sub=$1&email=$2" | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])'; }
DISPATCH_ENABLED=false bun run src/index.ts > "$WORK/backend.log" 2>&1 &
BACKEND_PID=$!
for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && break; sleep 0.5; done
curl -sf "$BASE/health" >/dev/null || fail "backend did not become healthy"
OWNER_ID="$(new_uuid)"; GUEST_ID="$(new_uuid)"; STRANGER_ID="$(new_uuid)"
TAG="$(new_uuid | cut -c1-8)"
OWNER="$(mint "$OWNER_ID" "owner-$TAG@example.test")"
GUEST="$(mint "$GUEST_ID" "Guest-$TAG@Example.Test")"
STRANGER="$(mint "$STRANGER_ID" "stranger-$TAG@example.test")"
echo "==> an organisation and a farm"
expect 201 POST /orgs "$OWNER" '{"name":"e2e onboarding org"}' -- "owner creates an org"
ORG_ID="$(json 'd["id"]')"
expect 201 POST /farms "$OWNER" "{\"name\":\"e2e onboarding farm\",\"orgId\":\"$ORG_ID\"}" -- "owner creates a farm"
FARM_ID="$(json 'd["id"]')"
echo "==> inviting by email"
expect 201 POST "/orgs/$ORG_ID/invitations" "$OWNER" "{\"email\":\"guest-$TAG@example.test\",\"role\":\"operator\"}" -- "owner invites an address"
TOKEN="$(json 'd["token"]')"; INV_ID="$(json 'd["id"]')"
grep -q "$TOKEN" <<<"$(psql_q "SELECT token_hash FROM org_invitations WHERE id='$INV_ID'")" && bad "the token is stored in the clear" || ok "only a hash of the token is stored"
expect 404 POST "/orgs/$ORG_ID/invitations" "$GUEST" "{\"email\":\"x-$TAG@example.test\",\"role\":\"viewer\"}" -- "a non-member cannot invite (the org does not exist for them)"
expect 200 GET "/orgs/$ORG_ID/invitations" "$OWNER" -- "owner lists pending invitations"
[ "$(json 'len(d)')" = 1 ] && ok "one pending" || bad "pending list: $(cat "$WORK/last.json")"
grep -q "$TOKEN" "$WORK/last.json" && bad "the list leaks the token" || ok "the list does not show the token"
echo "==> accepting"
expect 403 POST /invitations/accept "$STRANGER" "{\"token\":\"$TOKEN\"}" -- "someone signed in with another address cannot use the link"
expect 404 POST /invitations/accept "$GUEST" '{"token":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}' -- "a made-up token is refused"
expect 200 POST /invitations/accept "$GUEST" "{\"token\":\"$TOKEN\"}" -- "the invited address accepts (case-insensitively)"
[ "$(json 'd["role"]')" = operator ] && ok "joined as operator" || bad "role: $(cat "$WORK/last.json")"
expect 404 POST /invitations/accept "$GUEST" "{\"token\":\"$TOKEN\"}" -- "the link works once"
expect 200 GET "/orgs/$ORG_ID/members" "$OWNER" -- "owner lists members"
grep -q "guest-$TAG@example.test" "$WORK/last.json" && ok "the member list shows the email" || bad "no email in $(cat "$WORK/last.json")"
expect 200 GET "/orgs/$ORG_ID" "$GUEST" -- "the new member can see the org"
echo "==> expiry and revocation"
expect 201 POST "/orgs/$ORG_ID/invitations" "$OWNER" "{\"email\":\"stranger-$TAG@example.test\",\"role\":\"viewer\"}" -- "a second invitation"
T2="$(json 'd["token"]')"; I2="$(json 'd["id"]')"
psql_q "UPDATE org_invitations SET expires_at = now() - interval '1 minute' WHERE id='$I2'" >/dev/null
expect 404 POST /invitations/accept "$STRANGER" "{\"token\":\"$T2\"}" -- "an expired invitation is refused"
expect 201 POST "/orgs/$ORG_ID/invitations" "$OWNER" "{\"email\":\"stranger-$TAG@example.test\",\"role\":\"viewer\"}" -- "re-inviting issues a fresh one"
T3="$(json 'd["token"]')"; I3="$(json 'd["id"]')"
expect 200 DELETE "/orgs/$ORG_ID/invitations/$I3" "$OWNER" -- "owner revokes it"
expect 404 POST /invitations/accept "$STRANGER" "{\"token\":\"$T3\"}" -- "a revoked invitation is refused"
echo "==> enrolling a gateway with a one-time code"
expect 403 POST "/farms/$FARM_ID/enrolment-codes" "$GUEST" -- "an operator cannot make an enrolment code"
expect 201 POST "/farms/$FARM_ID/enrolment-codes" "$OWNER" -- "owner makes a code"
CODE="$(json 'd["code"]')"
[ "$(psql_q "SELECT count(*) FROM enrolment_codes WHERE code_hash = '$CODE'")" = 0 ] && ok "the code is not stored in the clear" || bad "code stored in the clear"
TYPED="$(echo "$CODE" | tr 'A-Z' 'a-z' | tr -d '-')"
expect 200 POST /auth/enrol - "{\"code\":\"$TYPED\"}" -- "the gateway trades the code (typed sloppily)"
GW_TOKEN="$(json 'd["gatewayToken"]')"
[ "$(json 'd["farmId"]')" = "$FARM_ID" ] && ok "it is told which farm" || bad "farm: $(cat "$WORK/last.json")"
expect 200 POST /auth/edge-token - "{\"farmId\":\"$FARM_ID\",\"edgeProxyToken\":\"$GW_TOKEN\"}" -- "the credential it got works"
expect 401 POST /auth/enrol - "{\"code\":\"$CODE\"}" -- "a code works once"
expect 201 POST "/farms/$FARM_ID/enrolment-codes" "$OWNER" -- "a second code"
C2="$(json 'd["code"]')"
expect 200 POST /auth/enrol - "{\"code\":\"$C2\"}" -- "enrolling a second machine"
[ "$(json 'd["gatewayToken"]')" = "$GW_TOKEN" ] && ok "the farm's token was not rotated, so the first gateway still works" || bad "the token changed"
expect 201 POST "/farms/$FARM_ID/enrolment-codes" "$OWNER" -- "a third code, to expire"
C3="$(json 'd["code"]')"
psql_q "UPDATE enrolment_codes SET expires_at = now() - interval '1 minute' WHERE farm_id='$FARM_ID' AND used_at IS NULL" >/dev/null
expect 401 POST /auth/enrol - "{\"code\":\"$C3\"}" -- "an expired code is refused"
if [ -x "$PROXY_DIR/target/debug/continuum-proxy" ] && "$PROXY_DIR/target/debug/continuum-proxy" enrol --help >/dev/null 2>&1; then
echo "==> the real proxy enrols itself"
expect 201 POST "/farms/$FARM_ID/enrolment-codes" "$OWNER" -- "a code for the proxy"
PCODE="$(json 'd["code"]')"
CONF="$WORK/proxy.toml"
( cd "$WORK" && CONTINUUM_CONFIG_PATH="$CONF" "$PROXY_DIR/target/debug/continuum-proxy" enrol "$BASE" "$PCODE" > "$WORK/enrol.log" 2>&1 ) || { cat "$WORK/enrol.log"; fail "proxy enrol failed"; }
[ "$(stat -c %a "$CONF")" = 600 ] && ok "the config it wrote is owner-only" || bad "config mode $(stat -c %a "$CONF")"
grep -q "$PCODE" "$WORK/enrol.log" && bad "the code is in the proxy output" || ok "the code is not echoed"
grep -q "gateway_token" "$CONF" && grep -q "farm_id = \"$FARM_ID\"" "$CONF" && ok "farm and token are in the config" || bad "config: $(sed 's/gateway_token.*/gateway_token = <hidden>/' "$CONF")"
env -C "$PROXY_DIR" CONTINUUM_CONFIG_PATH="$CONF" CONTINUUM_GO2RTC_BIN=/nonexistent CONTINUUM_GO2RTC_CONFIG="$WORK/go2rtc.yaml" \
CONTINUUM_PRINTERS_PATH="$WORK/none.toml" CONTINUUM_STORE_PATH="$WORK/store/continuum.db" RUST_LOG=info \
"$PROXY_DIR/target/debug/continuum-proxy" > "$WORK/proxy.log" 2>&1 &
PROXY_PID=$!
for _ in $(seq 1 40); do grep -q "connected for farm $FARM_ID" "$WORK/backend.log" && break; sleep 0.5; done
grep -q "connected for farm $FARM_ID" "$WORK/backend.log" && ok "it started on that config alone and the cloud accepted it" || bad "the enrolled proxy never connected: $(tail -5 "$WORK/proxy.log")"
else
echo " (skipping the proxy step: build continuum-proxy with the enrol command first)"
fi
# Last, because it blocks this address from enrolling for ten minutes.
echo "==> guessing"
for i in $(seq 1 12); do req POST /auth/enrol - '{"code":"ZZZZZ-ZZZZZ"}' >/dev/null; done
expect 429 POST /auth/enrol - '{"code":"ZZZZZ-ZZZZZ"}' -- "guessing is rate-limited"
echo
[ "$FAILURES" -eq 0 ] && echo "PASS - invitations and one-time-code enrolment." || { echo "FAIL - $FAILURES assertion(s)." >&2; exit 1; }
+171
View File
@@ -0,0 +1,171 @@
#!/usr/bin/env bash
# The non-Bambu vendors end to end, against FAKE printers: add each one through
# the API (so the provision frame, the real proxy's driver and its store are
# all exercised), then run a job through the real scheduler and watch it go
# queued -> uploading -> printing -> completed from the fake's own telemetry.
#
# What this proves is the plumbing around the drivers. It does NOT prove the
# drivers speak the real Klipper / OctoPrint / PrusaLink protocols: the fake
# (scripts/fake-http-printer.ts) is written from the same reading of the docs
# as the driver, so a shared misreading passes here and fails on hardware.
#
# Needs the dev stack (scripts/dev-env.sh up && schema) and the object store
# credentials in ../dev-secrets.sh (the job file goes through the real
# presigned URLs). No real printer is touched.
set -euo pipefail
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
[ -f "$REPO_ROOT/../dev-secrets.sh" ] && . "$REPO_ROOT/../dev-secrets.sh"
KINDS="${KINDS:-klipper octoprint prusa_link}"
PROXY_DIR="${PROXY_DIR:-$REPO_ROOT/../continuum-proxy}"
PORT="${PORT:-3062}"; AUTH_PORT="${AUTH_PORT:-3063}"; FAKE_PORT="${FAKE_PORT:-7141}"
BASE="http://localhost:$PORT"
DATABASE_URL="${DATABASE_URL:-postgres://continuum:continuum@localhost:5432/continuum}"
VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
EDGE_JWT_SECRET="${EDGE_JWT_SECRET:-e2e-edge-secret}"
export DATABASE_URL VALKEY_URL EDGE_JWT_SECRET PORT R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY R2_ENDPOINT
WORK="$(mktemp -d)"
BACKEND_PID=""; PROXY_PID=""; AUTH_PID=""; FAKE_PID=""; ORGS=""
fail() { echo "FAIL: $*" >&2; exit 1; }
ok() { echo " ok: $*"; }
psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; }
new_uuid() { cat /proc/sys/kernel/random/uuid; }
json() { python3 -c "import json,sys; d=json.load(sys.stdin); print($1)"; }
stop() { for pid in "$@"; do [ -n "$pid" ] && kill "$pid" 2>/dev/null || true; done; sleep 0.5; for pid in "$@"; do [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true; done; }
cleanup() {
local rc=$?
stop "$PROXY_PID" "$FAKE_PID" "$BACKEND_PID" "$AUTH_PID"
for org in $ORGS; do psql_q "DELETE FROM organisations WHERE id = '$org'" >/dev/null 2>&1 || true; done
if [ $rc -ne 0 ]; then
echo "--- backend log (tail) ---"; tail -25 "$WORK/backend.log" 2>/dev/null || true
echo "--- proxy log (tail) ---"; tail -30 "$WORK/proxy.log" 2>/dev/null || true
echo "--- fake printer log (tail) ---"; tail -15 "$WORK/fake.log" 2>/dev/null || true
fi
[ -n "${KEEP_LOGS:-}" ] && cp "$WORK"/*.log "$KEEP_LOGS"/ 2>/dev/null || true
rm -rf "$WORK"; exit $rc
}
trap cleanup EXIT
[ -n "${R2_ENDPOINT:-}" ] || fail "no object store credentials - ../dev-secrets.sh"
( cd "$PROXY_DIR" && cargo build --quiet 2>&1 | tail -3 ) || fail "proxy did not build"
bun scripts/dev-auth.ts --port "$AUTH_PORT" > "$WORK/auth.log" 2>&1 &
AUTH_PID=$!
AUTH_URL="http://localhost:$AUTH_PORT"
for _ in $(seq 1 40); do curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 && break; sleep 0.25; done
export SUPABASE_URL="$AUTH_URL"
USER_ID="$(new_uuid)"
TOKEN="$(curl -sf "$AUTH_URL/token?sub=$USER_ID" | json 'd["access_token"]')"
auth=(-H "authorization: Bearer $TOKEN")
start_backend() {
DISPATCH_ENABLED=true bun run src/index.ts >> "$WORK/backend.log" 2>&1 &
BACKEND_PID=$!
for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && return 0; sleep 0.5; done
fail "backend did not become healthy"
}
start_backend
# run_kind <kind> <allow insecure downloads: true|false> [outage]
# With downloads refused the proxy accepts the start and then fails it, which
# is the late-failure path: the job must fail with the proxy's reason.
run_kind() {
local KIND="$1" ALLOW="$2" OUTAGE="${3:-}"
echo "==> $KIND (insecure local downloads: $ALLOW${OUTAGE:+, cloud outage mid-job})"
KEY="key-$(new_uuid | cut -c1-8)"
env ${OUTAGE:+RUN_S=14} bun scripts/fake-http-printer.ts "$KIND" "$FAKE_PORT" "$KEY" > "$WORK/fake.log" 2>&1 &
FAKE_PID=$!
for _ in $(seq 1 40); do grep -q listening "$WORK/fake.log" && break; sleep 0.25; done
ORG_ID="$(curl -sf "${auth[@]}" -H 'content-type: application/json' -d "{\"name\":\"e2e vendors $KIND\"}" "$BASE/orgs" | json 'd["id"]')"
ORGS="$ORGS $ORG_ID"
FARM_ID="$(curl -sf "${auth[@]}" -H 'content-type: application/json' -d "{\"name\":\"farm $KIND\",\"orgId\":\"$ORG_ID\"}" "$BASE/farms" | json 'd["id"]')"
GW_TOKEN="$(curl -sf "${auth[@]}" -X POST "$BASE/farms/$FARM_ID/edge-proxy-token" | json 'd["edgeProxyToken"]')"
env -C "$PROXY_DIR" \
CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS="$ALLOW" CONTINUUM_GATEWAY_ID=e2e-vendors-gw CONTINUUM_FARM_ID="$FARM_ID" CONTINUUM_GATEWAY_TOKEN="$GW_TOKEN" \
CONTINUUM_UPLINK_URL="ws://localhost:$PORT/ws/edge/v1" CONTINUUM_GO2RTC_BIN=/nonexistent-go2rtc CONTINUUM_GO2RTC_CONFIG="$WORK/go2rtc.yaml" \
CONTINUUM_PRINTERS_PATH="$WORK/no-such-printers.toml" CONTINUUM_STORE_PATH="$WORK/store-$KIND/continuum.db" RUST_LOG=info \
"$PROXY_DIR/target/debug/continuum-proxy" > "$WORK/proxy.log" 2>&1 &
PROXY_PID=$!
# This farm's own connect line: the log is shared across the loop.
for _ in $(seq 1 60); do grep -q "connected for farm $FARM_ID" "$WORK/backend.log" && break; sleep 0.5; done
grep -q "connected for farm $FARM_ID" "$WORK/backend.log" || fail "the proxy never connected"
# Klipper needs no credential; the others take an API key.
CRED=""; [ "$KIND" != klipper ] && CRED=",\"apiKey\":\"$KEY\""
CODE="$(curl -s -o "$WORK/add.json" -w '%{http_code}' "${auth[@]}" -H 'content-type: application/json' \
-d "{\"name\":\"Fake $KIND\",\"vendor\":\"$KIND\",\"host\":\"127.0.0.1:$FAKE_PORT\"$CRED}" "$BASE/farms/$FARM_ID/printers")"
[ "$CODE" = 201 ] || fail "adding the $KIND printer returned $CODE: $(head -c 300 "$WORK/add.json")"
PRN_UUID="$(json 'd["id"]' < "$WORK/add.json")"
ok "added through the API and connected by the proxy"
grep -q "$KEY" "$WORK/proxy.log" "$WORK/backend.log" && fail "the API key is in a log"
ok "the API key is in neither log"
if [ "$KIND" != klipper ]; then
CODE="$(curl -s -o /dev/null -w '%{http_code}' "${auth[@]}" -H 'content-type: application/json' \
-d "{\"name\":\"No key\",\"vendor\":\"$KIND\",\"host\":\"127.0.0.1:$FAKE_PORT\"}" "$BASE/farms/$FARM_ID/printers")"
[ "$CODE" = 422 ] && ok "a missing API key is refused (422)" || fail "a keyless add returned $CODE"
fi
for _ in $(seq 1 30); do [ "$(psql_q "SELECT status FROM printers WHERE id='$PRN_UUID'")" = idle ] && break; sleep 1; done
[ "$(psql_q "SELECT status FROM printers WHERE id='$PRN_UUID'")" = idle ] || fail "the cloud never saw the printer idle"
ok "the cloud sees it idle (telemetry from the driver's poller)"
J="$(curl -sf "${auth[@]}" -H 'content-type: application/json' \
-d "{\"farmId\":\"$FARM_ID\",\"fileName\":\"e2e-part.gcode\",\"contentType\":\"application/octet-stream\"}" "$BASE/jobs")"
JOB_ID="$(printf '%s' "$J" | json 'd["job"]["id"]')"
OBJECT_KEY="$(printf '%s' "$J" | json 'd["job"]["r2ObjectKey"]')"
UPLOAD_URL="$(printf '%s' "$J" | json 'd["uploadUrl"]')"
printf 'G28\nG1 X10 Y10 F3000\nM104 S0\n' > "$WORK/part.gcode"
CODE="$(curl -s -o /dev/null -w '%{http_code}' -X PUT -H 'Content-Type: application/octet-stream' --data-binary @"$WORK/part.gcode" "$UPLOAD_URL")"
[ "$CODE" = 200 ] || fail "job file upload returned $CODE"
if [ -n "$OUTAGE" ]; then
# Take the cloud away the moment the job is dispatched - before any
# telemetry can have told it the print began - and bring it back only
# after the fake print is over. Telemetry alone would leave the job
# `uploading` (idle printer, never seen printing); only the gateway's
# own kept record can complete it.
for _ in $(seq 1 300); do [ "$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")" = uploading ] && break; sleep 0.1; done
[ "$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")" = uploading ] || fail "the job was not dispatched"
stop "$BACKEND_PID"; BACKEND_PID=""
echo " (cloud down; the fake print runs 14s without it)"
sleep 20
start_backend
fi
SEEN=""; START=$SECONDS; STATUS=""
while [ $((SECONDS - START)) -lt 60 ]; do
STATUS="$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")"
case " $SEEN " in *" $STATUS "*) ;; *) SEEN="$SEEN $STATUS" ;; esac
[ "$STATUS" = completed ] || [ "$STATUS" = failed ] && break
sleep 1
done
echo " job status history:$SEEN"
if [ -n "$OUTAGE" ]; then
[ "$STATUS" = completed ] || fail "after the outage the $KIND job is '$STATUS', not completed"
grep -Eq "gateway reported (started|finished)" "$WORK/backend.log" || fail "it completed, but not from the gateway's kept record"
ok "the cloud was away for the whole print; the proxy's kept record completed the job on reconnect"
elif [ "$ALLOW" = true ]; then
[ "$STATUS" = completed ] || fail "the $KIND job ended as '$STATUS'"
ok "$KIND: queued, scheduled, uploaded, printed and completed (fake printer)"
else
[ "$STATUS" = failed ] || fail "the refused download left the job '$STATUS', not failed"
REASON="$(psql_q "SELECT metadata->>'lastDispatchError' FROM print_jobs WHERE id='$JOB_ID'")"
case "$REASON" in *"non-HTTPS"*) ok "the job failed at once with the proxy's reason: $REASON" ;; *) fail "the job failed without the proxy's reason: '$REASON'" ;; esac
fi
bun -e "
import { s3 } from './src/s3';
import { DeleteObjectCommand } from '@aws-sdk/client-s3';
await s3.send(new DeleteObjectCommand({ Bucket: process.env.R2_BUCKET ?? 'continuum', Key: process.argv[1] }));
" "$OBJECT_KEY" >/dev/null 2>&1 || true
stop "$PROXY_PID" "$FAKE_PID"; PROXY_PID=""; FAKE_PID=""
}
for KIND in $KINDS; do run_kind "$KIND" true; done
run_kind klipper false
run_kind klipper true outage
echo "PASS - $KINDS: added from the API, scheduled and completed against fake printers; a refused download fails the job with its reason; a job finished while the cloud was away completes on reconnect."
+103
View File
@@ -0,0 +1,103 @@
// A stand-in for the REST printers (Klipper's Moonraker, OctoPrint, PrusaLink)
// so the real proxy can be driven end to end without owning one. It is only as
// good as my reading of those APIs - it proves the plumbing, not the hardware.
//
// bun scripts/fake-http-printer.ts <klipper|octoprint|prusa_link> <port> [apiKey]
//
// A file uploaded with "print" set starts a fake print that runs for RUN_S
// seconds (default 8) and then finishes. Every request is logged to stdout.
const [kind, portArg, apiKey = "e2e-key"] = process.argv.slice(2);
if (!["klipper", "octoprint", "prusa_link"].includes(kind ?? "")) {
console.error("usage: fake-http-printer.ts <klipper|octoprint|prusa_link> <port> [apiKey]");
process.exit(2);
}
const RUN_S = Number(process.env.RUN_S ?? 8);
let file: string | undefined;
let startedAt: number | undefined;
let cancelled = false;
const progress = () => (startedAt === undefined ? 0 : cancelled ? 0 : Math.min(1, (Date.now() - startedAt) / (RUN_S * 1000)));
const phase = () => (startedAt === undefined ? "idle" : cancelled ? "cancelled" : progress() >= 1 ? "done" : "printing");
const json = (body: unknown, status = 200) =>
new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
function startFrom(body: string, name?: string) {
file = name ?? /filename="([^"]+)"/.exec(body)?.[1] ?? "unknown";
startedAt = Date.now();
cancelled = false;
}
Bun.serve({
port: Number(portArg),
async fetch(req) {
const url = new URL(req.url);
const body = req.method === "GET" ? "" : new TextDecoder().decode(await req.arrayBuffer());
console.log(`${req.method} ${url.pathname} ${body.length}b`);
const route = `${req.method} ${url.pathname}`;
if (kind === "klipper") {
if (route === "GET /printer/info") return json({ result: { state: "ready" } });
if (route === "GET /printer/objects/query") {
const p = progress();
const state = { idle: "standby", printing: "printing", done: "complete", cancelled: "cancelled" }[phase()];
return json({
result: {
status: {
print_stats: { state, filename: file ?? "", print_duration: p * RUN_S, info: { current_layer: Math.ceil(p * 10), total_layer: 10 } },
virtual_sdcard: { progress: p },
extruder: { temperature: phase() === "printing" ? 210 : 24, target: phase() === "printing" ? 210 : 0 },
heater_bed: { temperature: phase() === "printing" ? 60 : 23, target: phase() === "printing" ? 60 : 0 },
},
},
});
}
if (route === "POST /server/files/upload") {
if (!/name="print"\r\n\r\ntrue/.test(body)) return json({ error: "print flag missing" }, 400);
startFrom(body);
return json({ print_started: true }, 201);
}
if (route === "POST /printer/print/cancel") { cancelled = true; return json({ result: "ok" }); }
if (route === "POST /printer/gcode/script") return json({ result: "ok" });
} else {
if (req.headers.get("x-api-key") !== apiKey) return json({ error: "bad key" }, 403);
const ph = phase();
if (kind === "octoprint") {
if (route === "GET /api/job") {
return json({
state: ph === "printing" ? "Printing" : "Operational",
job: { file: { name: file ?? null, display: file ?? null } },
progress: { completion: ph === "idle" || ph === "cancelled" ? null : progress() * 100, printTimeLeft: Math.round((1 - progress()) * RUN_S) },
});
}
if (route === "GET /api/printer") {
return json({ temperature: { tool0: { actual: ph === "printing" ? 210 : 24, target: ph === "printing" ? 210 : 0 }, bed: { actual: 60, target: ph === "printing" ? 60 : 0 } } });
}
if (route === "POST /api/files/local") {
if (!/name="print"\r\n\r\ntrue/.test(body)) return json({ error: "print flag missing" }, 400);
startFrom(body);
return json({ done: true }, 201);
}
if (route === "POST /api/job") { if (body.includes("cancel")) cancelled = true; return new Response(null, { status: 204 }); }
if (route === "POST /api/printer/command") return new Response(null, { status: 204 });
} else {
const state = { idle: "IDLE", printing: "PRINTING", done: "FINISHED", cancelled: "STOPPED" }[ph];
if (route === "GET /api/v1/status") {
return json({
printer: { state, temp_nozzle: ph === "printing" ? 215 : 25, target_nozzle: ph === "printing" ? 215 : 0, temp_bed: 60, target_bed: 0 },
...(ph === "printing" ? { job: { id: 1, progress: progress() * 100, time_remaining: Math.round((1 - progress()) * RUN_S) } } : {}),
});
}
if (route === "GET /api/v1/job") return json({ id: 1, file: { display_name: file } });
if (req.method === "PUT" && url.pathname.startsWith("/api/v1/files/local/")) {
if (req.headers.get("print-after-upload") !== "?1") return json({ error: "print flag missing" }, 400);
startFrom(body, decodeURIComponent(url.pathname.split("/").pop() ?? ""));
return new Response(null, { status: 201 });
}
if (route === "DELETE /api/v1/job/1") { cancelled = true; return new Response(null, { status: 204 }); }
}
}
return json({ error: `no route ${route}` }, 404);
},
});
console.log(`fake ${kind} listening on :${portArg}`);
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env bash
# THE FIRST REAL DISPATCH: queue a job in a throwaway farm that holds only the
# P1P, let the real scheduler send it through the real proxy, and watch the
# machine print it.
#
# **This starts a physical print.** It refuses to run without
# I_AUTHORISE_A_REAL_PRINT=yes, and only ever targets PRINTER_ID's printer
# (default p1p) because the throwaway farm contains nothing else. The file is
# read from the printer's own SD card root (FILE, default
# auto-test-001.gcode.3mf), uploaded to the object store, and dispatched like
# any customer job; the proxy then downloads it, FTPS-uploads it to the card
# and starts it, mapping filament from the archive itself.
#
# Nothing here cancels the print. To stop it: the printer's own screen, or
# the console's Stop control.
set -euo pipefail
[ "${I_AUTHORISE_A_REAL_PRINT:-}" = "yes" ] || { echo "refusing: set I_AUTHORISE_A_REAL_PRINT=yes to start a real print" >&2; exit 2; }
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$REPO_ROOT"
PRINTER_ID="${PRINTER_ID:-p1p}"
FILE="${FILE:-auto-test-001.gcode.3mf}"
# 0-based across all AMS units (AMS1 slot 4 is 3). Unset: map by the project's filament.
AMS_SLOT="${AMS_SLOT:-}"
PROXY_DIR="${PROXY_DIR:-$REPO_ROOT/../continuum-proxy}"
GO2RTC_BIN="${GO2RTC_BIN:-$HOME/.local/bin/go2rtc}"
PORT="${PORT:-3060}"; AUTH_PORT="${AUTH_PORT:-3061}"
BASE="http://localhost:$PORT"
DATABASE_URL="${DATABASE_URL:-postgres://continuum:continuum@localhost:5432/continuum}"
VALKEY_URL="${VALKEY_URL:-redis://localhost:6379}"
EDGE_JWT_SECRET="${EDGE_JWT_SECRET:-e2e-edge-secret}"
export DATABASE_URL VALKEY_URL EDGE_JWT_SECRET PORT R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY R2_ENDPOINT
MAX_WAIT_S="${MAX_WAIT_S:-5400}"
WORK="$(mktemp -d)"
BACKEND_PID=""; PROXY_PID=""; AUTH_PID=""; OBJECT_KEY=""
fail() { echo "FAIL: $*" >&2; exit 1; }
ok() { echo " ok: $*"; }
psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; }
USER_ID="$(cat /proc/sys/kernel/random/uuid)"
cleanup() {
local rc=$?
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill "$pid" 2>/dev/null || true; done
for _ in $(seq 1 20); do [ -n "$PROXY_PID" ] && kill -0 "$PROXY_PID" 2>/dev/null || break; sleep 0.3; done
for pid in "$PROXY_PID" "$BACKEND_PID" "$AUTH_PID"; do [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true; done
[ -n "$OBJECT_KEY" ] && bun -e "
import { s3 } from './src/s3';
import { DeleteObjectCommand } from '@aws-sdk/client-s3';
await s3.send(new DeleteObjectCommand({ Bucket: process.env.R2_BUCKET ?? 'continuum', Key: process.argv[1] }));
" "$OBJECT_KEY" >/dev/null 2>&1 || true
[ -z "${KEEP_DATA:-}" ] && [ -n "${ORG_ID:-}" ] && psql_q "DELETE FROM organisations WHERE id = '$ORG_ID'" >/dev/null 2>&1 || true
if [ $rc -ne 0 ]; then
echo "--- backend log (tail) ---"; tail -25 "$WORK/backend.log" 2>/dev/null || true
echo "--- proxy log (tail) ---"; tail -40 "$WORK/proxy.log" 2>/dev/null || true
fi
rm -rf "$WORK"
exit $rc
}
trap cleanup EXIT
[ -f "$PROXY_DIR/printers.toml" ] || fail "no printers.toml"
grep -q "^id *= *\"$PRINTER_ID\"" "$PROXY_DIR/printers.toml" || fail "printers.toml has no printer '$PRINTER_ID'"
ps -eo comm | grep -qx go2rtc && fail "a go2rtc is already running"
echo "==> reading $FILE from $PRINTER_ID's SD card (read-only)"
( cd "$PROXY_DIR" && cargo run -q --example bambu_mapping_check -- "$PRINTER_ID" "$FILE" "$WORK/job.3mf" ) > "$WORK/mapping.txt" 2>&1 || { cat "$WORK/mapping.txt"; fail "could not read the file"; }
grep -E 'gcode_state|project filament|mapping' "$WORK/mapping.txt" | sed 's/^/ /'
grep -q 'gcode_state: Some("IDLE")' "$WORK/mapping.txt" || grep -q 'gcode_state: Some("FINISH")' "$WORK/mapping.txt" || fail "the printer is not idle"
grep -q 'mapping the production path would use' "$WORK/mapping.txt" || fail "the filament would not map; not starting"
WANT="$(sha256sum "$WORK/job.3mf" | cut -d' ' -f1)"
echo "==> seeding a throwaway farm holding only $PRINTER_ID"
EDGE_PROXY_TOKEN="real-$(date +%s)"
ORG_ID="$(psql_q "INSERT INTO organisations (name) VALUES ('real print org') RETURNING id")"
psql_q "INSERT INTO org_members (org_id, user_id, role) VALUES ('$ORG_ID', '$USER_ID', 'owner')" >/dev/null
FARM_ID="$(psql_q "INSERT INTO farms (name, org_id, edge_proxy_token) VALUES ('real print farm', '$ORG_ID', '$EDGE_PROXY_TOKEN') RETURNING id")"
PRN_NAME="$(awk -F'"' -v id="$PRINTER_ID" '/^id *=/ {cur=$2} cur==id && /^name *=/ {print $2; exit}' "$PROXY_DIR/printers.toml")"
psql_q "INSERT INTO printers (farm_id, name, vendor, gateway_printer_id) VALUES ('$FARM_ID', '${PRN_NAME:-P1P}', 'bambu', '$PRINTER_ID')" >/dev/null
PRN_UUID="$(psql_q "SELECT id FROM printers WHERE farm_id='$FARM_ID'")"
[ "$(psql_q "SELECT count(*) FROM printers WHERE farm_id='$FARM_ID'")" = 1 ] || fail "the farm must hold exactly one printer"
bun scripts/dev-auth.ts --port "$AUTH_PORT" > "$WORK/auth.log" 2>&1 &
AUTH_PID=$!
AUTH_URL="http://localhost:$AUTH_PORT"
for _ in $(seq 1 40); do curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 && break; sleep 0.25; done
export SUPABASE_URL="$AUTH_URL"
TOKEN="$(curl -sf "$AUTH_URL/token?sub=$USER_ID" | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])')"
echo "==> starting the backend WITH dispatch on"
DISPATCH_ENABLED=true bun run src/index.ts > "$WORK/backend.log" 2>&1 &
BACKEND_PID=$!
for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && break; sleep 0.5; done
curl -sf "$BASE/health" >/dev/null || fail "backend did not become healthy"
echo "==> starting the real proxy"
env -C "$PROXY_DIR" \
CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS=true CONTINUUM_GATEWAY_ID=real-print-gw CONTINUUM_FARM_ID="$FARM_ID" CONTINUUM_GATEWAY_TOKEN="$EDGE_PROXY_TOKEN" \
CONTINUUM_UPLINK_URL="ws://localhost:$PORT/ws/edge/v1" \
CONTINUUM_GO2RTC_BIN="$GO2RTC_BIN" CONTINUUM_GO2RTC_CONFIG="$WORK/go2rtc.yaml" \
CONTINUUM_PRINTERS_PATH=printers.toml CONTINUUM_STORE_PATH="$WORK/continuum.db" RUST_LOG=info \
"$PROXY_DIR/target/debug/continuum-proxy" > "$WORK/proxy.log" 2>&1 &
PROXY_PID=$!
for _ in $(seq 1 60); do
[ "$(psql_q "SELECT status FROM printers WHERE id='$PRN_UUID'")" = "idle" ] && break
sleep 1
done
[ "$(psql_q "SELECT status FROM printers WHERE id='$PRN_UUID'")" = "idle" ] || fail "the cloud never saw $PRINTER_ID idle"
ok "the cloud sees $PRINTER_ID idle"
echo "==> queueing the job"
J="$(curl -sf -X POST "$BASE/jobs" -H "authorization: Bearer $TOKEN" -H 'content-type: application/json' \
-d "{\"farmId\":\"$FARM_ID\",\"fileName\":\"$FILE\",\"contentType\":\"application/octet-stream\"${AMS_SLOT:+,\"amsSlot\":$AMS_SLOT}}")"
JOB_ID="$(printf '%s' "$J" | python3 -c 'import json,sys; print(json.load(sys.stdin)["job"]["id"])')"
OBJECT_KEY="$(printf '%s' "$J" | python3 -c 'import json,sys; print(json.load(sys.stdin)["job"]["r2ObjectKey"])')"
UPLOAD_URL="$(printf '%s' "$J" | python3 -c 'import json,sys; print(json.load(sys.stdin)["uploadUrl"])')"
# The job is queued the moment it exists; dispatch needs the bytes, so the
# window between the two is a race the scheduler can only lose (the proxy's
# download then fails and the job fails), which is why this is done at once.
CODE="$(curl -s -o /dev/null -w '%{http_code}' -X PUT -H 'Content-Type: application/octet-stream' --data-binary @"$WORK/job.3mf" "$UPLOAD_URL")"
[ "$CODE" = 200 ] || fail "upload returned HTTP $CODE"
ok "job $JOB_ID queued, $(stat -c %s "$WORK/job.3mf") bytes uploaded"
echo "==> watching (up to ${MAX_WAIT_S}s)"
SEEN=""; LAST=""; START=$SECONDS
while [ $((SECONDS - START)) -lt "$MAX_WAIT_S" ]; do
STATUS="$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")"
PRN="$(psql_q "SELECT coalesce(vendor_state,'-') || ' ' || coalesce(progress_percent::text,'-') || '% layer ' || coalesce(current_layer::text,'-') || '/' || coalesce(total_layers::text,'-') || ' nozzle ' || coalesce(nozzle_temp_c::text,'-') FROM printers WHERE id='$PRN_UUID'")"
NOW="job=$STATUS printer=$PRN"
[ "$NOW" != "$LAST" ] && { echo " [$((SECONDS - START))s] $NOW"; LAST="$NOW"; }
case " $SEEN " in *" $STATUS "*) ;; *) SEEN="$SEEN $STATUS" ;; esac
[ "$STATUS" = completed ] && break
[ "$STATUS" = failed ] && break
sleep 5
done
echo
echo "job status history:$SEEN"
if [ "$STATUS" = completed ]; then
grep -E 'uploaded a project file|mapped the project|start' "$WORK/proxy.log" | sed 's/^/ proxy: /' | head -8
echo "PASS - the first real dispatch: queued in the cloud, scheduled, uploaded to the P1P, printed and completed."
else
fail "the job ended as '$STATUS'"
fi
+1
View File
@@ -81,6 +81,7 @@ export const printerProjection = projection(printers, [
// `GET /printers/:id/stream` per printer, and in a mixed fleet most of
// those 404 by design.
"cameraStream",
"cameraSnapshot",
"vendorState",
"vendorErrorCode",
"lastSeenAt",
+1
View File
@@ -4,3 +4,4 @@ export * from "./printers";
export * from "./print_jobs";
export * from "./embeddings";
export * from "./alerts";
export * from "./onboarding";
+51
View File
@@ -0,0 +1,51 @@
import { pgTable, uuid, text, timestamp, index } from "drizzle-orm/pg-core";
import { orgRoleEnum, organisations } from "./orgs";
import { farms } from "./farms";
/**
* An invitation to join an organisation, addressed to an email.
*
* Only a hash of the token is stored: the token is the whole credential, and
* the table is the kind of thing that ends up in a backup or a screenshot.
* It is shown to the inviter once, when the invitation is made.
*/
export const orgInvitations = pgTable(
"org_invitations",
{
id: uuid("id").primaryKey().defaultRandom(),
orgId: uuid("org_id")
.notNull()
.references(() => organisations.id, { onDelete: "cascade" }),
/** Lower-cased: addresses are compared case-insensitively. */
email: text("email").notNull(),
role: orgRoleEnum("role").notNull(),
tokenHash: text("token_hash").notNull().unique(),
invitedBy: uuid("invited_by").notNull(),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
acceptedAt: timestamp("accepted_at", { withTimezone: true }),
acceptedBy: uuid("accepted_by"),
},
(table) => ({ orgIdx: index("org_invitations_org_idx").on(table.orgId) }),
);
/**
* A short-lived, single-use code that lets a gateway fetch its farm's
* credentials without anyone copying a token around. Hashed like an
* invitation token, and for the same reason.
*/
export const enrolmentCodes = pgTable(
"enrolment_codes",
{
id: uuid("id").primaryKey().defaultRandom(),
farmId: uuid("farm_id")
.notNull()
.references(() => farms.id, { onDelete: "cascade" }),
codeHash: text("code_hash").notNull().unique(),
createdBy: uuid("created_by").notNull(),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
expiresAt: timestamp("expires_at", { withTimezone: true }).notNull(),
usedAt: timestamp("used_at", { withTimezone: true }),
},
(table) => ({ farmIdx: index("enrolment_codes_farm_idx").on(table.farmId) }),
);
+6
View File
@@ -47,6 +47,12 @@ export const orgMembers = pgTable(
.references(() => organisations.id, { onDelete: "cascade" }),
userId: uuid("user_id").notNull(),
role: orgRoleEnum("role").notNull().default("viewer"),
/**
* Shown in member lists, which would otherwise be bare ids. Taken from
* the token's `email` claim when someone creates an organisation or
* accepts an invitation; null for a member added by id.
*/
email: text("email"),
createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
},
+7 -2
View File
@@ -1,7 +1,7 @@
import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real, integer, uniqueIndex } from "drizzle-orm/pg-core";
import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real, integer, boolean, uniqueIndex } from "drizzle-orm/pg-core";
import { farms } from "./farms";
export const printerVendorEnum = pgEnum("printer_vendor", ["bambu", "prusa", "klipper"]);
export const printerVendorEnum = pgEnum("printer_vendor", ["bambu", "prusa", "klipper", "octoprint"]);
export const printerStatusEnum = pgEnum("printer_status", [
"offline",
"idle",
@@ -70,6 +70,11 @@ export const printers = pgTable(
* restream it.
*/
cameraStream: text("camera_stream"),
/**
* The camera can only be read as single JPEG frames (a P1/A1), which the
* gateway serves on request. Mutually exclusive with `cameraStream`.
*/
cameraSnapshot: boolean("camera_snapshot").notNull().default(false),
/**
* The printer's own state string, kept alongside the narrowed
* `status` enum. Worth storing both: the enum is what the UI filters
+87 -3
View File
@@ -1,9 +1,10 @@
import { and, asc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm";
import { and, asc, eq, inArray, isNotNull, isNull, lt, sql } from "drizzle-orm";
import { db } from "../db";
import { printers, printJobs } from "../db/schema";
import { getDownloadUrl, headObject } from "../s3";
import { isGatewayOnline, sendPrinterCommand } from "../routes/ws/edge";
import { nextJobStatus } from "./job_status";
import { jobStatusFromGatewayEvent, nextJobStatus } from "./job_status";
import { rememberStart, takeStart } from "./late_failure";
import { dispatchEnabled, dispatchSwitchWarning } from "./enabled";
/**
@@ -213,8 +214,13 @@ export async function dispatchTick(): Promise<number> {
job_id: job.id,
file_name: job.fileName,
url,
// Chosen by whoever queued the job; absent means "by the project's filament".
...(typeof job.metadata?.amsSlot === "number" ? { ams_slot: job.metadata.amsSlot } : {}),
});
// Accepted or not, a late result for this command can name the job.
rememberStart(result.commandId, job.id);
if (!result.ok) {
// Requeued, not failed. The gateway refusing a start is usually
// transient from the cloud's point of view - the printer was busier
@@ -247,6 +253,84 @@ export async function dispatchTick(): Promise<number> {
return dispatched;
}
/**
* A gateway that had accepted a `start_print` reports that the download or
* transfer then failed. Fails the job with that reason - but only while it is
* still `uploading`: one that has since reached `printing` is being driven by
* the printer's own telemetry, and a late message must not overrule it.
*
* Failed rather than requeued, for the reason `reapStalledUploads` gives.
*/
export async function failJobFromGateway(commandId: string, error: string): Promise<void> {
const jobId = takeStart(commandId);
if (!jobId) return;
const failed = await db
.update(printJobs)
.set({
status: "failed",
completedAt: new Date(),
updatedAt: new Date(),
metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({
lastDispatchError: `the gateway could not start it: ${error}`,
lastDispatchErrorAt: new Date().toISOString(),
})}::jsonb`,
})
.where(and(eq(printJobs.id, jobId), eq(printJobs.status, "uploading")))
.returning({ id: printJobs.id });
if (failed.length > 0) console.warn(`[dispatch] job ${jobId} failed: the gateway could not start it: ${error}`);
}
/**
* Applies an outcome a gateway kept for us. Throws only when the database
* could not be reached - the one case where the gateway must NOT be told it
* was handled, so that it sends it again. A job that is unknown, belongs to
* another farm, or is no longer in flight is not an error: there is nothing
* to change, and the gateway should stop sending it.
*/
export async function applyGatewayJobEvent(
farmId: string,
event: { job_id: string; event: "started" | "finished" | "failed"; detail?: string | null; at_ms: number },
): Promise<"applied" | "ignored"> {
// The farm in the `where` is the authorisation: a gateway can only move
// jobs of the farm it is connected for.
const job = await db.query.printJobs.findFirst({
columns: { id: true, status: true },
where: and(eq(printJobs.id, event.job_id), eq(printJobs.farmId, farmId)),
});
if (!job) return "ignored";
const next = jobStatusFromGatewayEvent(job.status, event.event);
if (!next) return "ignored";
const at = new Date(event.at_ms);
const terminal = next === "completed" || next === "failed";
const updated = await db
.update(printJobs)
.set({
status: next,
updatedAt: new Date(),
...(next === "completed" ? { progressPct: 100 } : {}),
...(terminal ? { completedAt: at } : {}),
...(next === "failed"
? {
metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({
lastDispatchError: event.detail ?? "the gateway reported the job failed",
lastDispatchErrorAt: at.toISOString(),
})}::jsonb`,
}
: {}),
})
// Re-checked in the write: the status may have moved since the read.
.where(and(eq(printJobs.id, job.id), eq(printJobs.status, job.status)))
.returning({ id: printJobs.id });
if (updated.length === 0) return "ignored";
console.log(`[dispatch] job ${job.id} ${job.status} -> ${next} (gateway reported ${event.event})`);
return "applied";
}
/**
* Fails jobs that have been `uploading` too long.
*
@@ -272,7 +356,7 @@ async function reapStalledUploads(): Promise<void> {
lastDispatchError:
`the gateway accepted the start but the printer never reported printing within ` +
`${Math.round(UPLOADING_TIMEOUT_MS / 60_000)} minutes - check the gateway's log for ` +
`the download or transfer error, which has no path back to the cloud yet`,
`the download or transfer error (a failure it reports promptly fails the job sooner)`,
lastDispatchErrorAt: new Date().toISOString(),
},
})
+27
View File
@@ -36,3 +36,30 @@ export function nextJobStatus(
return current === "printing" || current === "paused" ? "completed" : undefined;
}
}
/**
* What a job's status becomes when its gateway reports an outcome it kept
* for us (`job_event`) - possibly long after it happened, because the
* gateway holds each one until the cloud acknowledges it.
*
* Only jobs still in flight move. A job the cloud has cancelled, failed or
* completed meanwhile is left exactly as it is, which is what makes a stale
* report from a gateway harmless; so is `queued`, a job the cloud took back
* and may already have given to another printer. `undefined` means leave it.
*/
export function jobStatusFromGatewayEvent(
current: string,
event: "started" | "finished" | "failed",
): "printing" | "completed" | "failed" | undefined {
const inFlight = current === "uploading" || current === "printing" || current === "paused";
if (!inFlight) return undefined;
switch (event) {
case "started":
// Only the first step; a job already printing or paused is further on.
return current === "uploading" ? "printing" : undefined;
case "finished":
return "completed";
case "failed":
return "failed";
}
}
+33
View File
@@ -0,0 +1,33 @@
/**
* A `start_print` is answered the moment the gateway accepts it, because the
* download and transfer take minutes and this side stops waiting after ten
* seconds. If that work then fails, the gateway sends a second `command_result`
* for the same command id - by which time nothing here is waiting for it.
*
* This remembers which job each start belongs to, so that second result can
* fail the job with the gateway's own reason instead of the job sitting in
* `uploading` until the stall reaper gives up on it twenty minutes later.
*
* In memory on purpose: after a restart the reaper is still there to catch
* what this forgot.
*/
const MAX_REMEMBERED = 1000;
const REMEMBER_MS = 60 * 60 * 1000;
const starts = new Map<string, { jobId: string; at: number }>();
export function rememberStart(commandId: string, jobId: string, now = Date.now()): void {
starts.set(commandId, { jobId, at: now });
// Maps iterate in insertion order, so the oldest entries come first.
for (const [id, entry] of starts) {
if (starts.size <= MAX_REMEMBERED && now - entry.at < REMEMBER_MS) break;
starts.delete(id);
}
}
/** The job a start command was for, forgetting it: a command is only answered late once. */
export function takeStart(commandId: string): string | undefined {
const entry = starts.get(commandId);
starts.delete(commandId);
return entry?.jobId;
}
+2 -1
View File
@@ -6,7 +6,7 @@ import { toErrorResponse } from "@continuum/ts-core";
import { db } from "./db";
import { valkey } from "./queue";
import { authRoutes } from "./routes/auth";
import { orgRoutes } from "./routes/orgs";
import { invitationRoutes, orgRoutes } from "./routes/orgs";
import { farmRoutes } from "./routes/farms";
import { printerRoutes } from "./routes/printers";
import { jobRoutes } from "./routes/jobs";
@@ -62,6 +62,7 @@ const app = new Elysia()
})
.use(authRoutes)
.use(orgRoutes)
.use(invitationRoutes)
.use(farmRoutes)
.use(printerRoutes)
.use(jobRoutes)
+62
View File
@@ -0,0 +1,62 @@
/**
* Invitation tokens and gateway enrolment codes. Pure, so the parts worth
* getting exactly right are testable without a database.
*/
import { createHash, randomBytes, randomInt } from "node:crypto";
export const INVITE_TTL_MS = 7 * 24 * 60 * 60 * 1000;
export const ENROL_TTL_MS = 15 * 60 * 1000;
/** What is stored: a hash, never the secret. SHA-256 is enough for 256-bit random input. */
export const hashSecret = (secret: string) => createHash("sha256").update(secret).digest("hex");
/** 256 bits, URL-safe: goes in a link. */
export const newInviteToken = () => randomBytes(32).toString("base64url");
// Crockford base32: no I, L, O or U, so a code read off a screen and typed
// into a terminal survives the usual mix-ups.
const ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";
const CODE_LENGTH = 10;
/** `ABCDE-FGHJK`: 50 bits, single-use and valid for fifteen minutes. */
export function newEnrolmentCode(): string {
let code = "";
for (let i = 0; i < CODE_LENGTH; i++) code += ALPHABET[randomInt(ALPHABET.length)];
return `${code.slice(0, 5)}-${code.slice(5)}`;
}
/** What was typed, reduced to what was issued: case, dashes and spaces ignored, look-alikes folded. */
export function normaliseCode(input: string): string {
return input
.toUpperCase()
.replace(/[^0-9A-Z]/g, "")
.replace(/O/g, "0")
.replace(/[IL]/g, "1");
}
/** The form that is hashed for storage and lookup. */
export const codeKey = (input: string) => hashSecret(normaliseCode(input));
/**
* Counts failures per key inside a sliding window, so guessing an enrolment
* code from one address stops being worth trying. In memory: it resets on
* restart, which only ever helps an attacker by a window's worth of guesses.
*/
export class FailureLimiter {
private failures = new Map<string, number[]>();
constructor(
private readonly max: number,
private readonly windowMs: number,
) {}
blocked(key: string, now = Date.now()): boolean {
const recent = (this.failures.get(key) ?? []).filter((t) => now - t < this.windowMs);
if (recent.length === 0) this.failures.delete(key);
else this.failures.set(key, recent);
return recent.length >= this.max;
}
fail(key: string, now = Date.now()): void {
this.failures.set(key, [...(this.failures.get(key) ?? []), now]);
}
}
+31 -2
View File
@@ -3,8 +3,37 @@
* they test without a database - same reasoning as `dispatch/job_status.ts`.
*/
/** Vendors a gateway can actually drive today. Everything else is found by discovery but not yet controllable. */
export const PROVISIONABLE_VENDORS = ["bambu"] as const;
/**
* Vendors a gateway can drive, by the name the gateway knows them as, and
* what each needs to be told. `prusa_link` is stored as `prusa` and
* `octoprint` as itself - the database enum predates the gateway's names.
*
* Bambu is verified against real printers. The other three are written from
* their public APIs and tested against mock servers only.
*/
export const PROVISIONABLE_VENDORS = ["bambu", "klipper", "octoprint", "prusa_link"] as const;
export type ProvisionableVendor = (typeof PROVISIONABLE_VENDORS)[number];
const NEEDS: Record<ProvisionableVendor, { accessCode?: true; serialNumber?: true; apiKey?: true; dbVendor: "bambu" | "klipper" | "octoprint" | "prusa" }> = {
bambu: { accessCode: true, serialNumber: true, dbVendor: "bambu" },
klipper: { dbVendor: "klipper" },
octoprint: { apiKey: true, dbVendor: "octoprint" },
prusa_link: { apiKey: true, dbVendor: "prusa" },
};
export const dbVendorFor = (vendor: ProvisionableVendor) => NEEDS[vendor].dbVendor;
/** The first credential this vendor needs that was not supplied, named for the operator. */
export function missingCredential(
vendor: ProvisionableVendor,
given: { accessCode?: string; serialNumber?: string; apiKey?: string },
): string | undefined {
const needs = NEEDS[vendor];
if (needs.accessCode && !given.accessCode?.trim()) return "an access code";
if (needs.serialNumber && !given.serialNumber?.trim()) return "a serial number";
if (needs.apiKey && !given.apiKey?.trim()) return "an API key";
return undefined;
}
/**
* The id the gateway will know this printer by, chosen here so telemetry
+59 -2
View File
@@ -1,11 +1,15 @@
import { Elysia, t } from "elysia";
import jwt from "@elysiajs/jwt";
import { eq } from "drizzle-orm";
import { and, eq, gt, isNull } from "drizzle-orm";
import { UnauthorizedError } from "@continuum/ts-core";
import { db } from "../db";
import { farms } from "../db/schema";
import { enrolmentCodes, farms } from "../db/schema";
import { listMemberships, requireUser } from "../auth";
import { edgeJwtSecret } from "../auth/edge_secret";
import { FailureLimiter, codeKey } from "../onboarding/tokens";
/** Ten wrong codes from one address in ten minutes and it is told to wait. */
const enrolFailures = new FailureLimiter(10, 10 * 60 * 1000);
const EDGE_TOKEN_TTL_SECONDS = 3600;
@@ -64,6 +68,59 @@ export const authRoutes = new Elysia({ prefix: "/auth" })
detail: { tags: ["auth"], summary: "Trade a farm's proxy token for an edge JWT" },
},
)
/**
* A gateway trades a one-time code (`POST /farms/:id/enrolment-codes`) for
* its farm's credentials. Not behind `requireUser` for the same reason as
* `/edge-token`: the gateway has no account, and the code is the credential.
*
* The farm's existing token is returned, not rotated, so enrolling a second
* machine cannot lock out the first. A farm created before tokens were
* minted at creation gets one here.
*/
.post(
"/enrol",
async ({ body, server, request, set }) => {
const who = server?.requestIP(request)?.address ?? "unknown";
if (enrolFailures.blocked(who)) {
set.status = 429;
return { error: { code: "TOO_MANY_ATTEMPTS", message: "Too many wrong codes; wait a few minutes and try again" } };
}
// Claimed in one statement: of two gateways presenting the same code at
// once, exactly one gets a row back.
const [claimed] = await db
.update(enrolmentCodes)
.set({ usedAt: new Date() })
.where(
and(
eq(enrolmentCodes.codeHash, codeKey(body.code)),
isNull(enrolmentCodes.usedAt),
gt(enrolmentCodes.expiresAt, new Date()),
),
)
.returning({ farmId: enrolmentCodes.farmId });
if (!claimed) {
enrolFailures.fail(who);
throw new UnauthorizedError("That code is wrong, expired or already used");
}
const farm = await db.query.farms.findFirst({
columns: { id: true, edgeProxyToken: true },
where: eq(farms.id, claimed.farmId),
});
if (!farm) throw new UnauthorizedError("That code is wrong, expired or already used");
let token = farm.edgeProxyToken;
if (!token) {
token = crypto.randomUUID();
await db.update(farms).set({ edgeProxyToken: token, updatedAt: new Date() }).where(eq(farms.id, farm.id));
}
return { farmId: farm.id, gatewayToken: token };
},
{
body: t.Object({ code: t.String({ minLength: 4, maxLength: 64 }) }),
detail: { tags: ["auth"], summary: "Trade a one-time enrolment code for a gateway's credentials" },
},
)
.use(requireUser)
/**
* Who the caller is and what they can reach. A client calls this once after
+54 -9
View File
@@ -2,10 +2,18 @@ import { Elysia, t } from "elysia";
import { and, desc, eq } from "drizzle-orm";
import { ConflictError, NotFoundError } from "@continuum/ts-core";
import { db } from "../db";
import { farms, printers } from "../db/schema";
import { enrolmentCodes, farms, printers } from "../db/schema";
import { farmProjection, printerProjection } from "../db/projections";
import { PROVISIONABLE_VENDORS, gatewayIdFor, safeGatewayError } from "../provision/plan";
import {
PROVISIONABLE_VENDORS,
dbVendorFor,
gatewayIdFor,
missingCredential,
safeGatewayError,
type ProvisionableVendor,
} from "../provision/plan";
import { requireFarm, requireOrg, requireUser, visibleFarms } from "../auth";
import { ENROL_TTL_MS, codeKey, newEnrolmentCode } from "../onboarding/tokens";
import { PaginationQuery, paginate } from "./pagination";
import { getDiscoveredPrinters, isGatewayOnline, listUnclaimedPrinters, requestGateway } from "./ws/edge";
@@ -129,6 +137,35 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
},
{ params: IdParams, detail: { tags: ["farms"] } },
)
/**
* A one-time code for connecting a gateway to this farm. The gateway
* trades it at `POST /auth/enrol` for the farm's credentials, so nobody has
* to copy a token between machines.
*
* Unlike rotating the token, this leaves the farm's existing credential
* alone: enrolling a gateway cannot lock out the one already running.
* Fifteen minutes, one use, and only its hash is kept.
*/
.post(
"/:id/enrolment-codes",
async ({ user, params, set }) => {
await requireFarm(user.id, params.id, "owner", "enrolling a gateway");
const code = newEnrolmentCode();
const expiresAt = new Date(Date.now() + ENROL_TTL_MS);
await db.insert(enrolmentCodes).values({
farmId: params.id,
codeHash: codeKey(code),
createdBy: user.id,
expiresAt,
});
set.status = 201;
return { code, expiresAt };
},
{
params: IdParams,
detail: { tags: ["farms"], summary: "A one-time code to enrol a gateway (owner only)" },
},
)
/**
* Issues a *new* `edgeProxyToken` for this farm and returns it once.
*
@@ -198,7 +235,7 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
* Adds a printer to a farm: tells the farm's gateway to connect to it, and
* creates the printer row once it has.
*
* **The access code passes through this process and is kept nowhere.** It
* **The access code (or API key) passes through this process and is kept nowhere.** It
* is read from the body, placed in one frame to the gateway, and dropped:
* not stored, not logged, not echoed in a response or an error. The
* gateway keeps it in its own database. That is the platform's rule for
@@ -228,14 +265,20 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
},
};
}
const vendor = body.vendor as ProvisionableVendor;
const missing = missingCredential(vendor, body);
if (missing) {
set.status = 422;
return { error: { code: "CREDENTIAL_REQUIRED", message: `a ${vendor} printer needs ${missing}` } };
}
if (!isGatewayOnline(params.id)) {
throw new ConflictError(
"this farm's gateway is not connected, so it cannot be given the printer's access code - start it and try again",
"this farm's gateway is not connected, so it cannot be given the printer's credentials - start it and try again",
);
}
const gatewayPrinterId = gatewayIdFor(body.name);
const secrets = [body.accessCode];
const secrets = [body.accessCode ?? "", body.apiKey ?? ""];
const answer = await requestGateway(params.id, (requestId) => ({
type: "provision_printer",
request_id: requestId,
@@ -246,6 +289,7 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
host: body.host,
access_code: body.accessCode,
sn: body.serialNumber,
api_key: body.apiKey,
},
})).catch((err: unknown) => ({
ok: false,
@@ -268,8 +312,7 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
.values({
farmId: params.id,
name: body.name,
// Narrowed by the PROVISIONABLE_VENDORS check above.
vendor: body.vendor as (typeof PROVISIONABLE_VENDORS)[number],
vendor: dbVendorFor(vendor),
lanAddress: body.host,
serialNumber: body.serialNumber,
gatewayPrinterId,
@@ -295,9 +338,11 @@ export const farmRoutes = new Elysia({ prefix: "/farms" })
vendor: t.String({ minLength: 1, maxLength: 32 }),
// An address or hostname: letters, digits, dots, dashes, colons (IPv6).
host: t.String({ minLength: 1, maxLength: 255, pattern: "^[A-Za-z0-9.:_-]+$" }),
accessCode: t.String({ minLength: 1, maxLength: 128 }),
// Which of these a vendor needs is checked above (`missingCredential`).
accessCode: t.Optional(t.String({ minLength: 1, maxLength: 128 })),
// Bambu's MQTT topics are addressed by serial number.
serialNumber: t.String({ minLength: 5, maxLength: 64, pattern: "^[A-Za-z0-9]+$" }),
serialNumber: t.Optional(t.String({ minLength: 5, maxLength: 64, pattern: "^[A-Za-z0-9]+$" })),
apiKey: t.Optional(t.String({ minLength: 1, maxLength: 256 })),
}),
detail: {
tags: ["farms"],
+4
View File
@@ -73,6 +73,7 @@ export const jobRoutes = new Elysia({ prefix: "/jobs" })
farmId: body.farmId,
fileName: body.fileName,
r2ObjectKey,
metadata: body.amsSlot === undefined ? {} : { amsSlot: body.amsSlot },
// From the verified token, never from the body. It used to be a
// `submittedBy` field the caller filled in, which made the audit
// trail a suggestion rather than a record.
@@ -89,6 +90,9 @@ export const jobRoutes = new Elysia({ prefix: "/jobs" })
farmId: t.String({ format: "uuid" }),
fileName: t.String({ minLength: 1, maxLength: 500 }),
contentType: t.Optional(t.String()),
// 0-based across all AMS units (AMS1 slot 4 is 3). Absent: the
// gateway maps by the project's own filament.
amsSlot: t.Optional(t.Integer({ minimum: 0, maximum: 15 })),
}),
detail: { tags: ["jobs"], summary: "Queue a print job and get an upload URL" },
},
+169 -8
View File
@@ -1,9 +1,10 @@
import { Elysia, t } from "elysia";
import { and, count, eq, ne } from "drizzle-orm";
import { ConflictError, NotFoundError } from "@continuum/ts-core";
import { and, count, eq, gt, isNull, ne } from "drizzle-orm";
import { ConflictError, ForbiddenError, NotFoundError } from "@continuum/ts-core";
import { db } from "../db";
import { farms, orgMembers, organisations, ORG_ROLES } from "../db/schema";
import { farms, orgInvitations, orgMembers, organisations, ORG_ROLES } from "../db/schema";
import { listMemberships, requireOrg, requireUser } from "../auth";
import { INVITE_TTL_MS, hashSecret, newInviteToken } from "../onboarding/tokens";
const RoleSchema = t.Union(ORG_ROLES.map((role) => t.Literal(role)));
const IdParams = t.Object({ id: t.String({ format: "uuid" }) });
@@ -48,7 +49,7 @@ export const orgRoutes = new Elysia({ prefix: "/orgs" })
.returning();
await tx
.insert(orgMembers)
.values({ orgId: created!.id, userId: user.id, role: "owner" });
.values({ orgId: created!.id, userId: user.id, role: "owner", email: user.email?.toLowerCase() });
return created!;
});
set.status = 201;
@@ -125,6 +126,7 @@ export const orgRoutes = new Elysia({ prefix: "/orgs" })
return db
.select({
userId: orgMembers.userId,
email: orgMembers.email,
role: orgMembers.role,
joinedAt: orgMembers.createdAt,
})
@@ -138,10 +140,8 @@ export const orgRoutes = new Elysia({ prefix: "/orgs" })
* states the role this user should have, and sending it twice is the same
* as sending it once.
*
* There is no invitation flow yet - this takes a Supabase user id directly,
* so adding someone means knowing their id. That is the right shape for an
* API and the wrong shape for a UI, which will need an invite-by-email
* endpoint on top of it.
* This takes a Supabase user id directly, which is the right shape for an
* API and the wrong one for a person: they use the invitations below.
*/
.put(
"/:id/members/:userId",
@@ -180,6 +180,106 @@ export const orgRoutes = new Elysia({ prefix: "/orgs" })
detail: { tags: ["orgs"], summary: "Add a member or change their role" },
},
)
/**
* Invites someone by email. They join when they sign in with that address
* and open the link; nobody has to know their user id.
*
* **No email is sent from here.** The token is returned once, in the
* response, for the inviter to pass on (the console builds the link). A
* mail provider is a deployment decision this service does not make; until
* one is wired up, the inviter is the courier. The address still matters:
* only an account with that address can use the link, so a link that ends
* up somewhere else is useless to whoever finds it.
*
* A second invitation to the same address replaces the first, so there is
* only ever one live link per person per organisation.
*/
.post(
"/:id/invitations",
async ({ user, params, body, set }) => {
await requireOrg(user.id, params.id, "owner", "inviting someone");
const email = body.email.trim().toLowerCase();
const token = newInviteToken();
const expiresAt = new Date(Date.now() + INVITE_TTL_MS);
const invitation = await db.transaction(async (tx) => {
await tx
.delete(orgInvitations)
.where(
and(eq(orgInvitations.orgId, params.id), eq(orgInvitations.email, email), isNull(orgInvitations.acceptedAt)),
);
const [created] = await tx
.insert(orgInvitations)
.values({
orgId: params.id,
email,
role: body.role,
tokenHash: hashSecret(token),
invitedBy: user.id,
expiresAt,
})
.returning({
id: orgInvitations.id,
email: orgInvitations.email,
role: orgInvitations.role,
expiresAt: orgInvitations.expiresAt,
});
return created!;
});
set.status = 201;
return { ...invitation, token, acceptPath: `/invite?token=${token}` };
},
{
params: IdParams,
body: t.Object({ email: t.String({ format: "email", maxLength: 320 }), role: RoleSchema }),
detail: { tags: ["orgs"], summary: "Invite someone by email (the link is returned once)" },
},
)
.get(
"/:id/invitations",
async ({ user, params }) => {
await requireOrg(user.id, params.id, "owner", "viewing invitations");
return db
.select({
id: orgInvitations.id,
email: orgInvitations.email,
role: orgInvitations.role,
createdAt: orgInvitations.createdAt,
expiresAt: orgInvitations.expiresAt,
})
.from(orgInvitations)
.where(
and(
eq(orgInvitations.orgId, params.id),
isNull(orgInvitations.acceptedAt),
gt(orgInvitations.expiresAt, new Date()),
),
);
},
{ params: IdParams, detail: { tags: ["orgs"], summary: "Invitations not yet accepted" } },
)
.delete(
"/:id/invitations/:invitationId",
async ({ user, params }) => {
await requireOrg(user.id, params.id, "owner", "revoking an invitation");
const [removed] = await db
.delete(orgInvitations)
.where(
and(
eq(orgInvitations.id, params.invitationId),
eq(orgInvitations.orgId, params.id),
isNull(orgInvitations.acceptedAt),
),
)
.returning({ id: orgInvitations.id });
if (!removed) throw new NotFoundError("No such pending invitation");
return { revoked: removed.id };
},
{
params: t.Object({ id: t.String({ format: "uuid" }), invitationId: t.String({ format: "uuid" }) }),
detail: { tags: ["orgs"], summary: "Revoke an invitation" },
},
)
.delete(
"/:id/members/:userId",
async ({ user, params }) => {
@@ -200,3 +300,64 @@ export const orgRoutes = new Elysia({ prefix: "/orgs" })
},
{ params: MemberParams, detail: { tags: ["orgs"], summary: "Remove a member" } },
);
/**
* Accepting an invitation. Separate from `/orgs` because the caller is not
* yet a member of anything it could be scoped to.
*/
export const invitationRoutes = new Elysia({ prefix: "/invitations" })
.use(requireUser)
.post(
"/accept",
async ({ user, body }) => {
const invitation = await db.query.orgInvitations.findFirst({
where: and(
eq(orgInvitations.tokenHash, hashSecret(body.token)),
isNull(orgInvitations.acceptedAt),
gt(orgInvitations.expiresAt, new Date()),
),
});
// One answer for unknown, used, revoked and expired: which of those it
// was is not for someone holding a guessed or stale link to learn.
if (!invitation) throw new NotFoundError("This invitation is invalid, expired or already used");
if (!user.email) {
throw new ForbiddenError("Your sign-in carries no email address, so it cannot be matched to an invitation");
}
if (user.email.toLowerCase() !== invitation.email) {
throw new ForbiddenError(`This invitation was sent to a different address than the one you are signed in with`);
}
const rank = (role: string) => ORG_ROLES.indexOf(role as (typeof ORG_ROLES)[number]);
await db.transaction(async (tx) => {
// Claimed first and conditionally: two tabs opening the link at once
// cannot both succeed.
const [claimed] = await tx
.update(orgInvitations)
.set({ acceptedAt: new Date(), acceptedBy: user.id })
.where(and(eq(orgInvitations.id, invitation.id), isNull(orgInvitations.acceptedAt)))
.returning({ id: orgInvitations.id });
if (!claimed) throw new ConflictError("This invitation was just used");
const existing = await tx.query.orgMembers.findFirst({
where: and(eq(orgMembers.orgId, invitation.orgId), eq(orgMembers.userId, user.id)),
});
// An invitation never lowers a role someone already has.
const role = existing && rank(existing.role) > rank(invitation.role) ? existing.role : invitation.role;
await tx
.insert(orgMembers)
.values({ orgId: invitation.orgId, userId: user.id, role, email: invitation.email })
.onConflictDoUpdate({
target: [orgMembers.orgId, orgMembers.userId],
set: { role, email: invitation.email, updatedAt: new Date() },
});
});
const org = await db.query.organisations.findFirst({ where: eq(organisations.id, invitation.orgId) });
return { orgId: invitation.orgId, orgName: org?.name ?? "", role: invitation.role };
},
{
body: t.Object({ token: t.String({ minLength: 20, maxLength: 200 }) }),
detail: { tags: ["orgs"], summary: "Accept an invitation with the token from its link" },
},
);
+41 -4
View File
@@ -8,9 +8,9 @@ import { PrinterCreateSchema, PrinterUpdateSchema } from "../db/validators";
import { printerProjection } from "../db/projections";
import { requireFarm, requirePrinter, requireUser, visiblePrinters } from "../auth";
import { PaginationQuery, paginate } from "./pagination";
import { resolveStream } from "../stream";
import { resolveSnapshot, resolveStream } from "../stream";
import { mintIceServers, turnConfigured } from "../camera/ice";
import { isGatewayOnline, negotiateCameraStream, requestGateway, sendPrinterCommand } from "./ws/edge";
import { isGatewayOnline, negotiateCameraStream, requestCameraSnapshot, requestGateway, sendPrinterCommand } from "./ws/edge";
const IdParams = t.Object({ id: t.String({ format: "uuid" }) });
@@ -227,11 +227,22 @@ export const printerRoutes = new Elysia({ prefix: "/printers" })
async ({ user, params }) => {
await requirePrinter(user.id, params.id, "viewer");
const printer = await db.query.printers.findFirst({
columns: { farmId: true, cameraStream: true },
columns: { farmId: true, cameraStream: true, cameraSnapshot: true },
where: eq(printers.id, params.id),
});
if (!printer) throw new NotFoundError(`Printer ${params.id} not found`);
// A P1/A1: no stream to negotiate, a picture to fetch again and again.
if (!printer.cameraStream && printer.cameraSnapshot) {
const snap = resolveSnapshot(true, isGatewayOnline(printer.farmId));
if (!snap.ok) throw new NotFoundError(snap.message);
return {
kind: "snapshot" as const,
snapshotPath: `/printers/${params.id}/snapshot`,
intervalMs: 1000,
};
}
// The gateway has to be connected, because it is what answers the
// offer - a camera name alone would start a negotiation with nobody.
const result = resolveStream(printer.cameraStream, isGatewayOnline(printer.farmId));
@@ -241,7 +252,7 @@ export const printerRoutes = new Elysia({ prefix: "/printers" })
// credential is useful is small. Never cached in the client beyond the
// session it opens.
const iceServers = await mintIceServers();
return { ...result.stream, iceServers: iceServers ?? [], turnConfigured: turnConfigured() };
return { kind: "webrtc" as const, ...result.stream, iceServers: iceServers ?? [], turnConfigured: turnConfigured() };
},
{
params: IdParams,
@@ -251,6 +262,32 @@ export const printerRoutes = new Elysia({ prefix: "/printers" })
},
},
)
/**
* One JPEG frame from a printer whose camera is not RTSP (Bambu P1/A1).
*
* Viewer-level like the stream. The picture crosses the cloud, which the
* WebRTC path avoids; see `requestCameraSnapshot` for why that is bounded.
* `no-store` because each fetch must be the next frame, not a cached one.
*/
.get(
"/:id/snapshot",
async ({ user, params, set, status }) => {
await requirePrinter(user.id, params.id, "viewer");
let result: { jpeg?: Buffer; error?: string };
try {
result = await requestCameraSnapshot(params.id);
} catch (err) {
return status(404, { error: "NO_SNAPSHOT_CAMERA", message: err instanceof Error ? err.message : String(err) });
}
if (!result.jpeg) {
return status(503, { error: "CAMERA_UNAVAILABLE", message: result.error ?? "no picture" });
}
set.headers["content-type"] = "image/jpeg";
set.headers["cache-control"] = "no-store";
return new Response(result.jpeg, { headers: { "content-type": "image/jpeg", "cache-control": "no-store" } });
},
{ params: IdParams, detail: { tags: ["printers"], summary: "One camera frame (P1/A1)" } },
)
/**
* Relays a browser's WebRTC offer to the printer's gateway and returns
* go2rtc's answer.
+106 -7
View File
@@ -11,7 +11,7 @@ import {
import { db } from "../../db";
import { printerAlerts, printers } from "../../db/schema";
import { deriveAlert, type Status } from "../../alerts/derive";
import { advanceJobsFrom } from "../../dispatch";
import { advanceJobsFrom, applyGatewayJobEvent, failJobFromGateway } from "../../dispatch";
import { enqueueFrameIngestionBatch } from "../../queue";
import { edgeJwtSecret } from "../../auth/edge_secret";
import { farmFromEdgeUpgrade } from "../../auth/edge_upgrade";
@@ -75,6 +75,11 @@ interface PendingWebrtc {
timer: ReturnType<typeof setTimeout>;
}
interface PendingSnapshot {
resolve: (result: { jpeg?: Buffer; error?: string }) => void;
timer: ReturnType<typeof setTimeout>;
}
interface EdgeConn {
farmId: string;
gatewayId: string;
@@ -124,6 +129,8 @@ interface EdgeConn {
* retried.
*/
webrtc: Map<string, PendingWebrtc>;
/** In-flight camera frame reads, keyed by request id. */
snapshots: Map<string, PendingSnapshot>;
}
const connections = new Map<string, EdgeConn>();
@@ -414,6 +421,7 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) {
// Cleared like the rest when absent: a printer that lost its stream
// must stop being offered one.
cameraStream: snapshot.camera_stream ?? null,
cameraSnapshot: snapshot.camera_snapshot ?? false,
lastSeenAt: new Date(),
updatedAt: new Date(),
})
@@ -615,6 +623,62 @@ export async function negotiateCameraStream(
});
}
/** A frame this fresh is served again instead of asking the printer. */
const SNAPSHOT_MAX_AGE_MS = 800;
/** Connect, TLS and a frame from the printer take a few seconds. */
const SNAPSHOT_TIMEOUT_MS = 15_000;
const snapshotInFlight = new Map<string, Promise<{ jpeg?: Buffer; error?: string }>>();
const snapshotCache = new Map<string, { at: number; jpeg: Buffer }>();
/**
* One JPEG frame from a P1/A1-style camera, read by the printer's gateway.
*
* Unlike the WebRTC path the picture itself crosses the cloud - there is no
* codec a browser could be given instead - so this is deliberately small:
* one frame per request, concurrent viewers of a printer share one read, and
* a frame under a second old is served again. The printer's camera is a
* single-client port, and ten browsers must not become ten logins.
*
* Throws for an unknown printer, one with no snapshot camera, or a farm with
* no gateway connected; resolves `{ error }` for a read that failed.
*/
export async function requestCameraSnapshot(printerId: string): Promise<{ jpeg?: Buffer; error?: string }> {
const printer = await db.query.printers.findFirst({
columns: { farmId: true, gatewayPrinterId: true, cameraSnapshot: true },
where: eq(printers.id, printerId),
});
if (!printer) throw new Error(`printer ${printerId} not found`);
if (!printer.cameraSnapshot) throw new Error(`printer ${printerId} has no snapshot camera`);
if (!printer.gatewayPrinterId) throw new Error(`printer ${printerId} is not claimed by a gateway`);
const cached = snapshotCache.get(printerId);
if (cached && Date.now() - cached.at < SNAPSHOT_MAX_AGE_MS) return { jpeg: cached.jpeg };
const running = snapshotInFlight.get(printerId);
if (running) return running;
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 requestId = crypto.randomUUID();
const gatewayPrinterId = printer.gatewayPrinterId;
const promise = new Promise<{ jpeg?: Buffer; error?: string }>((resolve) => {
const timer = setTimeout(() => {
conn.snapshots.delete(requestId);
resolve({ error: `gateway did not answer within ${SNAPSHOT_TIMEOUT_MS}ms` });
}, SNAPSHOT_TIMEOUT_MS);
conn.snapshots.set(requestId, { resolve, timer });
conn.send({ type: "camera_snapshot_request", request_id: requestId, printer_id: gatewayPrinterId });
}).then((result) => {
if (result.jpeg) snapshotCache.set(printerId, { at: Date.now(), jpeg: result.jpeg });
snapshotInFlight.delete(printerId);
return result;
});
snapshotInFlight.set(printerId, promise);
return promise;
}
/** Whether a farm currently has a gateway connected. */
export function isGatewayOnline(farmId: string): boolean {
const wsId = gatewayByFarm.get(farmId);
@@ -681,6 +745,7 @@ export const edgeWsRoute = new Elysia()
lastStatus: new Map(),
pending: new Map(),
webrtc: new Map(),
snapshots: new Map(),
};
conn.handshakeTimer = setTimeout(() => {
@@ -776,6 +841,19 @@ export const edgeWsRoute = new Elysia()
break;
}
case "camera_snapshot": {
const waiting = conn.snapshots.get(message.request_id);
if (!waiting) return; // the viewer's request already timed out
clearTimeout(waiting.timer);
conn.snapshots.delete(message.request_id);
waiting.resolve(
message.jpeg_base64
? { jpeg: Buffer.from(message.jpeg_base64, "base64") }
: { error: message.error ?? "the gateway returned no picture" },
);
break;
}
case "command_result": {
const pending = conn.pending.get(message.command_id);
if (!pending) {
@@ -789,17 +867,19 @@ export const edgeWsRoute = new Elysia()
// gives up after ten seconds - so its real outcome can only ever
// arrive late.
//
// The reason is logged rather than dropped. It is the difference
// between an operator seeing "job failed" and seeing which half
// of a download-then-upload failed and what the printer said.
// It does not change the job's fate: the job never reaching
// `printing` is what fails it, and that already works. Recording
// it against the job needs a column print_jobs does not have.
// The reason is logged, and when the command was a start we
// remember (`rememberStart`), it also fails that job with the
// gateway's own words rather than leaving it in `uploading`.
const reason = message.ok ? "reported success" : `failed: ${message.error ?? "no reason given"}`;
console.warn(
`[edge] gateway ${conn.gatewayId} sent a result for unknown, ` +
`expired or already-answered command ${message.command_id} - ${reason}`,
);
if (!message.ok) {
failJobFromGateway(message.command_id, message.error ?? "no reason given").catch((err) =>
console.error("[edge] could not record a late start failure", err),
);
}
return;
}
clearTimeout(pending.timer);
@@ -811,6 +891,20 @@ export const edgeWsRoute = new Elysia()
break;
}
case "job_event": {
if (!conn.farmId) return;
try {
await applyGatewayJobEvent(conn.farmId, message);
// Acknowledged whether or not it changed anything: "nothing to
// change" is an answer, and the gateway stops sending it.
conn.send({ type: "job_event_ack", event_id: message.event_id });
} catch (err) {
// No ack, so the gateway keeps it and sends it again.
console.error(`[edge] could not record job event ${message.event_id} from gateway ${conn.gatewayId}`, err);
}
break;
}
case "discovered_printers": {
if (!conn.farmId) return;
// No gateway id until `hello`, and the handshake timer closes a
@@ -885,6 +979,11 @@ export const edgeWsRoute = new Elysia()
waiting.resolve({ error: "gateway disconnected before answering" });
conn.webrtc.delete(sessionId);
}
for (const [requestId, waiting] of conn.snapshots) {
clearTimeout(waiting.timer);
waiting.resolve({ error: "gateway disconnected before answering" });
conn.snapshots.delete(requestId);
}
if (conn.farmId && gatewayByFarm.get(conn.farmId) === ws.id) {
gatewayByFarm.delete(conn.farmId);
}
+15
View File
@@ -35,3 +35,18 @@ export function resolveStream(
}
return { ok: true, stream: { cameraId: cameraStream } };
}
export const NO_SNAPSHOT_MESSAGE = "this printer's camera cannot be read as pictures";
/**
* For a P1/A1-style camera: no stream to negotiate, a picture to fetch
* repeatedly. Same ordering rule as `resolveStream` - the printer first.
*/
export function resolveSnapshot(
cameraSnapshot: boolean | null | undefined,
gatewayOnline: boolean,
): { ok: true } | { ok: false; reason: "no_camera" | "gateway_offline"; message: string } {
if (!cameraSnapshot) return { ok: false, reason: "no_camera", message: NO_SNAPSHOT_MESSAGE };
if (!gatewayOnline) return { ok: false, reason: "gateway_offline", message: NO_GATEWAY_MESSAGE };
return { ok: true };
}
+51
View File
@@ -0,0 +1,51 @@
import { describe, expect, test } from "bun:test";
import { rememberStart, takeStart } from "../src/dispatch/late_failure";
describe("late start failures", () => {
test("a remembered start is returned once, then forgotten", () => {
rememberStart("c1", "job-1");
expect(takeStart("c1")).toBe("job-1");
expect(takeStart("c1")).toBeUndefined();
});
test("a command nobody started is unknown", () => {
expect(takeStart("never-sent")).toBeUndefined();
});
test("an hour-old start is dropped rather than failing a job long since resolved", () => {
const longAgo = Date.now() - 2 * 60 * 60 * 1000;
rememberStart("old", "job-old", longAgo);
rememberStart("new", "job-new");
expect(takeStart("old")).toBeUndefined();
expect(takeStart("new")).toBe("job-new");
});
test("the table is bounded", () => {
for (let i = 0; i < 1500; i++) rememberStart(`bulk-${i}`, `job-${i}`);
expect(takeStart("bulk-0")).toBeUndefined();
expect(takeStart("bulk-1499")).toBe("job-1499");
});
});
import { jobStatusFromGatewayEvent } from "../src/dispatch/job_status";
describe("jobStatusFromGatewayEvent", () => {
test("a finish is believed for any job still in flight, even one never seen printing", () => {
for (const current of ["uploading", "printing", "paused"]) {
expect(jobStatusFromGatewayEvent(current, "finished")).toBe("completed");
expect(jobStatusFromGatewayEvent(current, "failed")).toBe("failed");
}
});
test("started only promotes a job that was still uploading", () => {
expect(jobStatusFromGatewayEvent("uploading", "started")).toBe("printing");
expect(jobStatusFromGatewayEvent("printing", "started")).toBeUndefined();
expect(jobStatusFromGatewayEvent("paused", "started")).toBeUndefined();
});
test("a job the cloud has already settled or taken back is left alone", () => {
for (const current of ["queued", "completed", "failed", "cancelled"]) {
for (const event of ["started", "finished", "failed"] as const) {
expect(jobStatusFromGatewayEvent(current, event)).toBeUndefined();
}
}
});
});
+52
View File
@@ -0,0 +1,52 @@
import { describe, expect, test } from "bun:test";
import {
FailureLimiter,
codeKey,
hashSecret,
newEnrolmentCode,
newInviteToken,
normaliseCode,
} from "../src/onboarding/tokens";
describe("enrolment codes", () => {
test("have the shape that was promised and do not repeat", () => {
const codes = new Set(Array.from({ length: 200 }, newEnrolmentCode));
expect(codes.size).toBe(200);
for (const code of codes) expect(code).toMatch(/^[0-9A-HJKMNP-TV-Z]{5}-[0-9A-HJKMNP-TV-Z]{5}$/);
});
test("a code typed carelessly still finds the one that was issued", () => {
const issued = "AB0C1-DEF23";
for (const typed of ["ab0c1-def23", "AB0C1DEF23", " ab0c1 def23 ", "ABOC1-DEF23", "AB0Cl-DEF23", "ABOCI-DEF23"]) {
expect(codeKey(typed)).toBe(codeKey(issued));
}
expect(codeKey("AB0C1-DEF24")).not.toBe(codeKey(issued));
});
test("normalising is idempotent", () => {
expect(normaliseCode(normaliseCode("ab-oc1"))).toBe(normaliseCode("ab-oc1"));
});
});
describe("invitation tokens", () => {
test("are long, URL-safe and unique, and only their hash is kept", () => {
const a = newInviteToken();
expect(a).toMatch(/^[A-Za-z0-9_-]{43}$/);
expect(newInviteToken()).not.toBe(a);
expect(hashSecret(a)).toMatch(/^[0-9a-f]{64}$/);
expect(hashSecret(a)).not.toContain(a);
});
});
describe("FailureLimiter", () => {
test("blocks after too many failures and forgets them after the window", () => {
const limiter = new FailureLimiter(3, 1000);
for (let i = 0; i < 3; i++) {
expect(limiter.blocked("ip", i)).toBe(false);
limiter.fail("ip", i);
}
expect(limiter.blocked("ip", 10)).toBe(true);
expect(limiter.blocked("other", 10)).toBe(false);
expect(limiter.blocked("ip", 2000)).toBe(false);
});
});
+19 -1
View File
@@ -1,5 +1,5 @@
import { describe, expect, test } from "bun:test";
import { gatewayIdFor, safeGatewayError } from "../src/provision/plan";
import { dbVendorFor, gatewayIdFor, missingCredential, safeGatewayError } from "../src/provision/plan";
describe("gatewayIdFor", () => {
const accepted = /^[A-Za-z0-9_-]{1,64}$/;
@@ -38,3 +38,21 @@ describe("safeGatewayError", () => {
expect(safeGatewayError(undefined, [])).toBe("the gateway did not say why");
});
});
describe("missingCredential", () => {
test("a Bambu needs its access code and serial number", () => {
expect(missingCredential("bambu", {})).toBe("an access code");
expect(missingCredential("bambu", { accessCode: "12345678" })).toBe("a serial number");
expect(missingCredential("bambu", { accessCode: "12345678", serialNumber: "01P00A123" })).toBeUndefined();
});
test("OctoPrint and PrusaLink need an API key; Klipper needs nothing", () => {
expect(missingCredential("octoprint", { accessCode: "x" })).toBe("an API key");
expect(missingCredential("prusa_link", { apiKey: " " })).toBe("an API key");
expect(missingCredential("prusa_link", { apiKey: "k" })).toBeUndefined();
expect(missingCredential("klipper", {})).toBeUndefined();
});
test("the gateway's PrusaLink is stored under the database's older name", () => {
expect(dbVendorFor("prusa_link")).toBe("prusa");
expect(dbVendorFor("octoprint")).toBe("octoprint");
});
});
+21 -1
View File
@@ -1,7 +1,7 @@
import { describe, expect, test } from "bun:test";
import { Value } from "@sinclair/typebox/value";
import { EdgeClientMessageSchema } from "@continuum/ts-types";
import { NO_CAMERA_MESSAGE, NO_GATEWAY_MESSAGE, resolveStream } from "../src/stream";
import { NO_CAMERA_MESSAGE, NO_GATEWAY_MESSAGE, NO_SNAPSHOT_MESSAGE, resolveSnapshot, resolveStream } from "../src/stream";
const hello = { type: "hello", gateway_id: "gw-1", farm_id: "f1", token: "t" };
@@ -63,3 +63,23 @@ describe("resolveStream", () => {
expect(!both.ok && both.reason).toBe("no_camera");
});
});
describe("snapshot cameras (P1/A1)", () => {
test("telemetry and the gateway's answer survive the validators", () => {
expect(
Value.Check(EdgeClientMessageSchema, {
type: "telemetry",
printers: [{ printer_id: "p1", name: "P1P", camera_snapshot: true }],
}),
).toBe(true);
expect(Value.Check(EdgeClientMessageSchema, { type: "camera_snapshot", request_id: "r", jpeg_base64: "/9j/" })).toBe(true);
expect(Value.Check(EdgeClientMessageSchema, { type: "camera_snapshot", request_id: "r", error: "no frame" })).toBe(true);
expect(Value.Check(EdgeClientMessageSchema, { type: "camera_snapshot", jpeg_base64: "/9j/" })).toBe(false);
});
test("the printer is checked before the gateway", () => {
expect(resolveSnapshot(false, false)).toMatchObject({ ok: false, reason: "no_camera", message: NO_SNAPSHOT_MESSAGE });
expect(resolveSnapshot(true, false)).toMatchObject({ ok: false, reason: "gateway_offline", message: NO_GATEWAY_MESSAGE });
expect(resolveSnapshot(true, true)).toEqual({ ok: true });
});
});