diff --git a/drizzle/0005_bouncy_cannonball.sql b/drizzle/0005_bouncy_cannonball.sql new file mode 100644 index 0000000..7b94ab4 --- /dev/null +++ b/drizzle/0005_bouncy_cannonball.sql @@ -0,0 +1 @@ +ALTER TABLE "printers" ADD COLUMN "camera_snapshot" boolean DEFAULT false NOT NULL; \ No newline at end of file diff --git a/drizzle/meta/0005_snapshot.json b/drizzle/meta/0005_snapshot.json new file mode 100644 index 0000000..71aeca9 --- /dev/null +++ b/drizzle/meta/0005_snapshot.json @@ -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": {} + } +} \ No newline at end of file diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 119f479..6a5dbdd 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -36,6 +36,13 @@ "when": 1793285676878, "tag": "0004_printer_alerts", "breakpoints": true + }, + { + "idx": 5, + "version": "7", + "when": 1793289367708, + "tag": "0005_bouncy_cannonball", + "breakpoints": true } ] } \ No newline at end of file diff --git a/scripts/e2e-camera-snapshot.sh b/scripts/e2e-camera-snapshot.sh new file mode 100755 index 0000000..50ae7c5 --- /dev/null +++ b/scripts/e2e-camera-snapshot.sh @@ -0,0 +1,153 @@ +#!/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:-}" + +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 + 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 "${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/null || fail "$CAMERA_ID ($CAM_HOST) refuses TCP 6000: no snapshot camera, or it is off" + ok "$CAMERA_ID accepts connections on 6000" +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 dev-auth issuer and the backend" +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"])')" +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" + +[ -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." diff --git a/src/db/projections.ts b/src/db/projections.ts index c75b30d..0a6918c 100644 --- a/src/db/projections.ts +++ b/src/db/projections.ts @@ -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", diff --git a/src/db/schema/printers.ts b/src/db/schema/printers.ts index 57f5e0a..3e6bb3c 100644 --- a/src/db/schema/printers.ts +++ b/src/db/schema/printers.ts @@ -1,4 +1,4 @@ -import { pgTable, uuid, text, timestamp, jsonb, pgEnum, real, 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"]); @@ -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 diff --git a/src/routes/printers.ts b/src/routes/printers.ts index bce4fdf..0e5a342 100644 --- a/src/routes/printers.ts +++ b/src/routes/printers.ts @@ -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. diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index 1fd1a94..c74c2cd 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -75,6 +75,11 @@ interface PendingWebrtc { timer: ReturnType; } +interface PendingSnapshot { + resolve: (result: { jpeg?: Buffer; error?: string }) => void; + timer: ReturnType; +} + interface EdgeConn { farmId: string; gatewayId: string; @@ -124,6 +129,8 @@ interface EdgeConn { * retried. */ webrtc: Map; + /** In-flight camera frame reads, keyed by request id. */ + snapshots: Map; } const connections = new Map(); @@ -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>(); +const snapshotCache = new Map(); + +/** + * 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) { @@ -885,6 +963,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); } diff --git a/src/stream.ts b/src/stream.ts index 43276e3..8b2ad0f 100644 --- a/src/stream.ts +++ b/src/stream.ts @@ -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 }; +} diff --git a/tests/stream.test.ts b/tests/stream.test.ts index bf46f73..a93e1f6 100644 --- a/tests/stream.test.ts +++ b/tests/stream.test.ts @@ -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 }); + }); +});