diff --git a/drizzle/0004_printer_alerts.sql b/drizzle/0004_printer_alerts.sql new file mode 100644 index 0000000..73f2379 --- /dev/null +++ b/drizzle/0004_printer_alerts.sql @@ -0,0 +1,32 @@ +DO $$ BEGIN + CREATE TYPE "public"."alert_kind" AS ENUM('print_failed', 'print_paused', 'printer_error', 'ai_detection'); +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +CREATE TABLE IF NOT EXISTS "printer_alerts" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "farm_id" uuid NOT NULL, + "printer_id" uuid, + "kind" "alert_kind" NOT NULL, + "message" text NOT NULL, + "confidence" real, + "object_key" text, + "acknowledged_by" uuid, + "acknowledged_at" timestamp with time zone, + "created_at" timestamp with time zone DEFAULT now() NOT NULL +); +--> statement-breakpoint +DO $$ BEGIN + ALTER TABLE "printer_alerts" ADD CONSTRAINT "printer_alerts_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 "printer_alerts" ADD CONSTRAINT "printer_alerts_printer_id_printers_id_fk" FOREIGN KEY ("printer_id") REFERENCES "public"."printers"("id") ON DELETE set null ON UPDATE no action; +EXCEPTION + WHEN duplicate_object THEN null; +END $$; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "printer_alerts_farm_open_idx" ON "printer_alerts" USING btree ("farm_id","acknowledged_at"); \ No newline at end of file diff --git a/drizzle/meta/0004_snapshot.json b/drizzle/meta/0004_snapshot.json new file mode 100644 index 0000000..9c31a48 --- /dev/null +++ b/drizzle/meta/0004_snapshot.json @@ -0,0 +1,841 @@ +{ + "id": "ee6e6f93-7aec-431b-955d-75c485352ea7", + "prevId": "e4751490-d3cf-44ca-8807-f7dba03415ea", + "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 + }, + "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 10968dc..119f479 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -29,6 +29,13 @@ "when": 1792399397819, "tag": "0003_colorful_electro", "breakpoints": true + }, + { + "idx": 4, + "version": "7", + "when": 1793285676878, + "tag": "0004_printer_alerts", + "breakpoints": true } ] } \ No newline at end of file diff --git a/scripts/e2e-alerts.sh b/scripts/e2e-alerts.sh new file mode 100755 index 0000000..bd6f708 --- /dev/null +++ b/scripts/e2e-alerts.sh @@ -0,0 +1,216 @@ +#!/usr/bin/env bash +# End-to-end test of alerts: a stand-in gateway reports a printer going +# from printing to failed over the real /ws/edge/v1, and an AI detection is +# published on the real Valkey channel; both must become rows an operator can +# list and acknowledge. +# +# This is the test that the unit tests cannot be. tests/authz.test.ts proves +# tokens are verified and that the role ladder orders correctly; only this +# proves that each route actually asks, that scoping is applied in the query +# rather than after it, and that credentials stay out of responses - the three +# things that were wrong when every route was unauthenticated. +# +# Four users, two organisations, no printers or gateways needed: +# owner_a, operator_a, viewer_a in org A, which owns farm A and printer A +# outsider_b in org B, which owns nothing +# +# Usage: +# scripts/dev-env.sh up && scripts/dev-env.sh schema +# scripts/e2e-authz.sh +# +# KEEP_DATA=1 leave the seeded organisations behind for inspection +set -euo pipefail + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$REPO_ROOT" + +PORT="${PORT:-3030}" +AUTH_PORT="${AUTH_PORT:-3031}" +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}" +# Presigning a PUT is local HMAC with no network call, so POST /jobs works +# offline - but the SDK refuses to sign with empty credentials. +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 +export R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY R2_ENDPOINT + +WORK="$(mktemp -d)" +BACKEND_PID="" +AUTH_PID="" +FAILURES=0 + +# `uuidgen` is not installed everywhere; the kernel's generator always is. +new_uuid() { + if [ -r /proc/sys/kernel/random/uuid ]; then + cat /proc/sys/kernel/random/uuid + else + python3 -c 'import uuid; print(uuid.uuid4())' + fi +} + +# Uses DATABASE_URL rather than a hardcoded localhost. The backend under test +# reads that variable, so a psql pointed anywhere else seeds fixtures into a +# different database than the one being exercised - and the failure message +# below names DATABASE_URL, which would then be naming a URL nothing tried. +psql_q() { psql "$DATABASE_URL" -tAq -c "$1"; } + +cleanup() { + local rc=$? + # Never `kill $PID` with an unset PID: `kill 0` signals the whole process + # group, which includes this script and whatever launched it. + local pids=() + [ -n "$BACKEND_PID" ] && pids+=("$BACKEND_PID") + [ -n "$AUTH_PID" ] && pids+=("$AUTH_PID") + if [ ${#pids[@]} -gt 0 ]; then + kill "${pids[@]}" 2>/dev/null || true + sleep 0.3 + kill -9 "${pids[@]}" 2>/dev/null || true + fi + if [ -z "${KEEP_DATA:-}" ]; then + for org in "${ORG_A:-}" "${ORG_B:-}"; do + [ -n "$org" ] && psql_q "DELETE FROM organisations WHERE id = '$org'" >/dev/null 2>&1 || true + done + fi + if [ $rc -ne 0 ]; then + echo + echo "--- backend log (tail) ---"; tail -30 "$WORK/backend.log" 2>/dev/null || true + fi + rm -rf "$WORK" + exit $rc +} +trap cleanup EXIT + +fail() { echo "FAIL: $*" >&2; exit 1; } +ok() { echo " ok: $*"; } +bad() { echo " FAILED: $*" >&2; FAILURES=$((FAILURES + 1)); } + +# `req [json body]` -> prints the status code, +# leaves the response body in $WORK/last.json. +req() { + local method="$1" path="$2" token="$3" body="${4:-}" + local args=(-s -o "$WORK/last.json" -w '%{http_code}' -X "$method" "$BASE$path") + [ "$token" != "-" ] && args+=(-H "authorization: Bearer $token") + [ -n "$body" ] && args+=(-H 'content-type: application/json' -d "$body") + curl "${args[@]}" +} + +# `expect [body] -- ` +expect() { + local want="$1"; shift + local method="$1" path="$2" token="$3"; shift 3 + local body="" + if [ "${1:-}" != "--" ]; then body="$1"; shift; fi + shift # the -- + local what="$*" + local got; got="$(req "$method" "$path" "$token" "$body")" + if [ "$got" = "$want" ]; then + ok "$what ($got)" + else + bad "$what: expected $want, got $got - $(head -c 300 "$WORK/last.json")" + fi +} + +body_has() { grep -q "$1" "$WORK/last.json" || bad "expected '$1' in: $(head -c 300 "$WORK/last.json")"; } +body_lacks() { grep -q "$1" "$WORK/last.json" && bad "'$1' leaked into: $(head -c 300 "$WORK/last.json")" || true; } +json() { python3 -c "import json,sys; d=json.load(sys.stdin); print($1)" < "$WORK/last.json"; } + +# --- 0. preconditions ------------------------------------------------------ + +echo "==> checking the database is up and migrated" +psql_q "SELECT 1" >/dev/null 2>&1 || fail "no database at $DATABASE_URL - run scripts/dev-env.sh up" +psql_q "SELECT org_id FROM farms LIMIT 0" >/dev/null 2>&1 \ + || fail "farms.org_id is missing - run scripts/dev-env.sh schema" +psql_q "SELECT role FROM org_members LIMIT 0" >/dev/null 2>&1 \ + || fail "org_members is missing - run scripts/dev-env.sh schema" +ok "schema has the org layer" + +# --- 1. the issuer and the backend ---------------------------------------- + +echo "==> starting the dev-auth JWKS issuer on :$AUTH_PORT" +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 +curl -sf "$AUTH_URL/auth/v1/.well-known/jwks.json" >/dev/null 2>&1 \ + || fail "the dev-auth issuer did not come up: $(cat "$WORK/auth.log")" +export SUPABASE_URL="$AUTH_URL" +ok "issuer up, SUPABASE_URL=$SUPABASE_URL" + +mint() { curl -sf "$AUTH_URL/token?sub=$1" | python3 -c 'import json,sys; print(json.load(sys.stdin)["access_token"])'; } + + +echo "==> starting continuum-backend on :$PORT" +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 2>&1 || fail "backend did not become healthy: $(cat "$WORK/backend.log")" +ok "backend healthy" + +# --- 2. a farm with one claimed printer --------------------------------- + +OWNER_ID="$(new_uuid)"; OWNER="$(mint "$OWNER_ID")" +expect 201 POST /orgs "$OWNER" '{"name":"e2e alerts org"}' -- "create an org" +ORG_A="$(json 'd["id"]')" +expect 201 POST /farms "$OWNER" "{\"name\":\"e2e alerts farm\",\"orgId\":\"$ORG_A\"}" -- "create a farm" +FARM_A="$(json 'd["id"]')" +expect 201 POST /printers "$OWNER" \ + "{\"farmId\":\"$FARM_A\",\"name\":\"Fake P1\",\"vendor\":\"bambu\",\"gatewayPrinterId\":\"fake1\"}" -- "claim a printer" +PRINTER_A="$(json 'd["id"]')" +expect 200 POST "/farms/$FARM_A/edge-proxy-token" "$OWNER" -- "mint the gateway token" +PROXY_TOKEN="$(json 'd["edgeProxyToken"]')" +expect 200 POST /auth/edge-token - "{\"farmId\":\"$FARM_A\",\"edgeProxyToken\":\"$PROXY_TOKEN\"}" -- "trade it for an edge JWT" +EDGE_JWT="$(json 'd["token"]')" +[ -n "$EDGE_JWT" ] || fail "no edge token in the response" + +# --- 3. telemetry -> alert ------------------------------------------------ + +echo "==> a print that fails raises exactly one alert" +# The first frame is only an observation (nothing is known before it), so +# IDLE,RUNNING,RUNNING,FAILED,FAILED has one transition worth raising. +bun scripts/fake-gateway.ts "ws://localhost:$PORT/ws/edge/v1" "$EDGE_JWT" fake1 IDLE,RUNNING,RUNNING,FAILED,FAILED \ + || fail "the fake gateway could not talk to the backend" +N="$(psql_q "SELECT count(*) FROM printer_alerts WHERE farm_id='$FARM_A' AND kind='print_failed'")" +[ "$N" = "1" ] && ok "one print_failed alert from a printing -> failed transition" || bad "expected 1 print_failed alert, found $N" +TOTAL="$(psql_q "SELECT count(*) FROM printer_alerts WHERE farm_id='$FARM_A'")" +[ "$TOTAL" = "1" ] && ok "and nothing else: staying failed and idle -> running are quiet" || bad "expected 1 alert in total, found $TOTAL" + +echo "==> a reconnect does not re-announce what was already known" +bun scripts/fake-gateway.ts "ws://localhost:$PORT/ws/edge/v1" "$EDGE_JWT" fake1 FAILED,FAILED || fail "reconnect failed" +TOTAL="$(psql_q "SELECT count(*) FROM printer_alerts WHERE farm_id='$FARM_A'")" +[ "$TOTAL" = "1" ] && ok "still one alert after reconnecting into the same state" || bad "reconnect raised alerts: $TOTAL" + +expect 200 GET "/alerts?open=true&farmId=$FARM_A" "$OWNER" -- "the owner lists open alerts" +body_has "the print failed" +body_has "Fake P1" + +# --- 4. AI detection -> alert -------------------------------------------- + +echo "==> an AI detection on the Valkey channel becomes a row (above the floor only)" +publish() { valkey-cli -u "$VALKEY_URL" PUBLISH channel:print_alerts "$1" >/dev/null; } +mk() { printf '{"printer_id":"%s","label":"spaghetti","confidence":%s,"captured_at":"2026-10-29T10:00:00Z","r2_object_key":"frames/e2e.jpg"}' "$PRINTER_A" "$1"; } +publish "$(mk 0.95)" +publish "$(mk 0.10)" +publish 'this is not json' +sleep 1 +AI="$(psql_q "SELECT count(*) FROM printer_alerts WHERE farm_id='$FARM_A' AND kind='ai_detection'")" +[ "$AI" = "1" ] && ok "the 0.95 detection was recorded; the 0.10 one and the junk were not" || bad "expected 1 ai_detection alert, found $AI" +CONF="$(psql_q "SELECT confidence FROM printer_alerts WHERE farm_id='$FARM_A' AND kind='ai_detection'")" +[ "$CONF" = "0.95" ] && ok "confidence stored as published" || bad "confidence is '$CONF'" + +echo +if [ "$FAILURES" -eq 0 ]; then + echo "PASS - telemetry transitions and AI detections both become alerts." +else + echo "FAIL - $FAILURES assertion(s) failed." >&2 + exit 1 +fi diff --git a/scripts/e2e-authz.sh b/scripts/e2e-authz.sh index 8a63db3..e5932af 100755 --- a/scripts/e2e-authz.sh +++ b/scripts/e2e-authz.sh @@ -305,6 +305,28 @@ CODE="$(req POST /auth/edge-token - "{\"farmId\":\"$FARM_A\",\"edgeProxyToken\": expect 401 POST /auth/edge-token - "{\"farmId\":\"$FARM_A\",\"edgeProxyToken\":\"not-the-token\"}" \ -- "a wrong proxy token is refused" +# --- 7b. alerts --------------------------------------------------------- + +echo "==> alerts are scoped, and acknowledging is an operator's decision" +ALERT_A="$(new_uuid)" +psql_q "INSERT INTO printer_alerts (id, farm_id, printer_id, kind, message) VALUES ('$ALERT_A', '$FARM_A', '$PRINTER_A', 'print_failed', 'e2e: the print failed')" >/dev/null +expect 200 GET "/alerts?open=true" "$VIEWER_A" -- "a viewer lists open alerts" +body_has "$ALERT_A" +expect 200 GET "/alerts?open=true" "$OUTSIDER" -- "the outsider lists open alerts" +body_lacks "$ALERT_A" +ok "the outsider's alert list does not contain farm A's alert" +expect 404 POST "/alerts/$ALERT_A/acknowledge" "$OUTSIDER" -- "the outsider cannot acknowledge another org's alert" +expect 403 POST "/alerts/$ALERT_A/acknowledge" "$VIEWER_A" -- "a viewer cannot acknowledge" +expect 200 POST "/alerts/$ALERT_A/acknowledge" "$OPERATOR_A" -- "an operator acknowledges" +WHO="$(json 'd["acknowledgedBy"]')" +[ "$WHO" = "$OPERATOR_A_ID" ] && ok "acknowledgedBy came from the token" || bad "acknowledgedBy is '$WHO'" +expect 200 POST "/alerts/$ALERT_A/acknowledge" "$OWNER_A" -- "acknowledging again is harmless" +WHO2="$(json 'd["acknowledgedBy"]')" +[ "$WHO2" = "$OPERATOR_A_ID" ] && ok "the first acknowledger stays on the record" || bad "second ack overwrote the record: '$WHO2'" +expect 200 GET "/alerts?open=true" "$VIEWER_A" -- "an acknowledged alert leaves the open list" +body_lacks "$ALERT_A" +expect 401 GET /alerts - -- "GET /alerts rejects anonymous" + # --- 8. the guards that stop an org becoming unadministrable -------------- echo "==> an organisation cannot be left without an owner" diff --git a/scripts/fake-gateway.ts b/scripts/fake-gateway.ts new file mode 100644 index 0000000..f182798 --- /dev/null +++ b/scripts/fake-gateway.ts @@ -0,0 +1,36 @@ +/** + * A stand-in gateway for scripts/e2e-alerts.sh: connects to /ws/edge/v1 with + * an edge JWT, says hello, then reports one printer through a sequence of + * states, one telemetry frame each, and waits for the backend to take them. + * + * Usage: bun scripts/fake-gateway.ts STATE[,STATE...] + * A state is a Bambu gcode_state word (IDLE, RUNNING, PAUSE, FAILED). + */ +const [url, jwt, printerId, states] = process.argv.slice(2); +if (!url || !jwt || !printerId || !states) { + console.error("usage: fake-gateway.ts STATE[,STATE...]"); + process.exit(2); +} + +const ws = new WebSocket(url, { headers: { authorization: `Bearer ${jwt}` } } as never); +ws.onerror = () => { + console.error("socket error"); + process.exit(1); +}; +ws.onopen = async () => { + ws.send(JSON.stringify({ type: "hello", gateway_id: "fake-gw" })); + for (const state of states.split(",")) { + ws.send( + JSON.stringify({ + type: "telemetry", + printers: [{ printer_id: printerId, name: "Fake", state, percent: 10, print_error: 0 }], + }), + ); + // Telemetry is persisted in order per socket; a pause keeps the frames + // from being coalesced into one observation. + await Bun.sleep(400); + } + await Bun.sleep(500); + ws.close(); + process.exit(0); +}; diff --git a/src/alerts/derive.ts b/src/alerts/derive.ts new file mode 100644 index 0000000..fc0901e --- /dev/null +++ b/src/alerts/derive.ts @@ -0,0 +1,55 @@ +/** + * Which telemetry changes deserve an operator's attention, on their own with + * no imports - same reason as `dispatch/job_status.ts`: `../db` opens a + * connection at import time, and this is the decision worth testing without + * one. + */ + +export type AlertKind = "print_failed" | "print_paused" | "printer_error"; +export type Status = "offline" | "idle" | "printing" | "paused" | "error" | "maintenance"; + +/** + * What changed, if anything, that is worth raising. + * + * Transitions and not levels: a printer that sits in `error` for an hour is + * one alert, not 720. `previous` undefined means this connection has not seen + * the printer before, which raises nothing - a gateway reconnecting would + * otherwise re-announce every fault that was already known. + * + * A fault while printing is a failed print; a fault while idle is the printer + * needing attention, which is a different thing to tell someone. + */ +export function deriveAlert( + previous: Status | undefined, + next: Status | undefined, + printerName: string, + vendorState?: string | null, +): { kind: AlertKind; message: string } | undefined { + if (previous === undefined || next === undefined || previous === next) return undefined; + const detail = vendorState ? ` (printer reports ${vendorState})` : ""; + + if (next === "error") { + return previous === "printing" || previous === "paused" + ? { kind: "print_failed", message: `${printerName}: the print failed${detail}` } + : { kind: "printer_error", message: `${printerName} reported an error${detail}` }; + } + // Paused by the printer itself is how an AMS fault shows up - the print is + // still on the bed and nobody is going to notice without being told. A + // pause the operator asked for is also announced; telling them what they + // just did is cheap, and the console cannot tell the two apart. + if (next === "paused" && previous === "printing") { + return { kind: "print_paused", message: `${printerName}: the print is paused${detail}` }; + } + return undefined; +} + +/** Whether an AI detection is worth a row; the worker deliberately does not threshold. */ +export function worthRaising(confidence: number, minimum: number): boolean { + return confidence >= minimum; +} + +/** The floor for AI detections, overridable because it is a taste, not a fact. */ +export function minAiConfidence(env: Record): number { + const parsed = Number(env.ALERT_MIN_CONFIDENCE); + return env.ALERT_MIN_CONFIDENCE !== undefined && parsed >= 0 && parsed <= 1 ? parsed : 0.6; +} diff --git a/src/alerts/listener.ts b/src/alerts/listener.ts new file mode 100644 index 0000000..dc7a8f7 --- /dev/null +++ b/src/alerts/listener.ts @@ -0,0 +1,49 @@ +import { eq } from "drizzle-orm"; +import { db } from "../db"; +import { printerAlerts, printers } from "../db/schema"; +import { valkey } from "../queue"; +import { minAiConfidence, worthRaising } from "./derive"; +import { parsePrintAlert } from "./parse"; + +const CHANNEL = "channel:print_alerts"; + +/** + * Turns the AI worker's fire-and-forget publications into rows. + * + * PUBLISH keeps nothing, so a detection published while this process is down + * is gone - the same limit the schema documents. Everything that arrives + * while it is up is kept, which is the difference between an alert and a + * flicker. + * + * A subscribed connection can do nothing else in RESP, hence `duplicate()` + * rather than reusing the shared client. + */ +export function startAiAlertListener(): void { + const floor = minAiConfidence(process.env); + const subscriber = valkey.duplicate(); + subscriber.on("error", (err) => console.warn("[alerts] listener connection", err.message)); + + subscriber.subscribe(CHANNEL).catch((err: unknown) => console.warn("[alerts] cannot subscribe", err)); + subscriber.on("message", async (_channel, raw) => { + const alert = parsePrintAlert(raw); + if (!alert || !worthRaising(alert.confidence, floor)) return; + try { + const printer = await db.query.printers.findFirst({ + columns: { id: true, farmId: true, name: true }, + where: eq(printers.id, alert.printer_id), + }); + // An alert for a printer we do not have has no farm to show it to. + if (!printer) return; + await db.insert(printerAlerts).values({ + farmId: printer.farmId, + printerId: printer.id, + kind: "ai_detection", + message: `${printer.name}: looks like ${alert.label} (${Math.round(alert.confidence * 100)}%)`, + confidence: alert.confidence, + objectKey: alert.r2_object_key, + }); + } catch (err) { + console.warn("[alerts] could not record an AI alert", err); + } + }); +} diff --git a/src/alerts/parse.ts b/src/alerts/parse.ts new file mode 100644 index 0000000..c86fed3 --- /dev/null +++ b/src/alerts/parse.ts @@ -0,0 +1,22 @@ +import { TypeCompiler } from "@sinclair/typebox/compiler"; +import { PrintAlertSchema, type PrintAlert } from "@continuum/ts-types"; +// Must come first - see queue/frame_task.ts: an unregistered format fails. +import "../validation/formats"; + +const check = TypeCompiler.Compile(PrintAlertSchema); + +/** + * Parses a `channel:print_alerts` message, or returns undefined for anything + * that is not a valid `PrintAlert`. A subscriber must never throw on what + * arrives: the channel is shared and a bad publisher should cost one message, + * not the listener. + */ +export function parsePrintAlert(raw: string): PrintAlert | undefined { + let value: unknown; + try { + value = JSON.parse(raw); + } catch { + return undefined; + } + return check.Check(value) ? (value as PrintAlert) : undefined; +} diff --git a/src/auth/index.ts b/src/auth/index.ts index 71000d6..129a283 100644 --- a/src/auth/index.ts +++ b/src/auth/index.ts @@ -2,10 +2,12 @@ export { requireUser, type AuthenticatedUser } from "./plugin"; export { roleSatisfies, ORG_ROLES, type OrgRole } from "./roles"; export { listMemberships, + requireAlert, requireFarm, requireJob, requireOrg, requirePrinter, + visibleAlerts, visibleFarms, visibleJobs, visiblePrinters, diff --git a/src/auth/scope.ts b/src/auth/scope.ts index b497bf2..84421b4 100644 --- a/src/auth/scope.ts +++ b/src/auth/scope.ts @@ -1,7 +1,7 @@ import { and, eq, inArray } from "drizzle-orm"; import { ForbiddenError, NotFoundError } from "@continuum/ts-core"; import { db } from "../db"; -import { farms, orgMembers, printJobs, printers } from "../db/schema"; +import { farms, orgMembers, printJobs, printerAlerts, printers } from "../db/schema"; import { roleSatisfies, type OrgRole } from "./roles"; /** @@ -64,6 +64,14 @@ export function visibleJobs(userId: string) { ); } +/** `where` condition limiting `printer_alerts` to those the user can see. */ +export function visibleAlerts(userId: string) { + return inArray( + printerAlerts.farmId, + db.select({ id: farms.id }).from(farms).where(visibleFarms(userId)), + ); +} + function check(role: OrgRole | null, minimum: OrgRole, missing: string, action: string): OrgRole { if (role === null) throw new NotFoundError(missing); if (!roleSatisfies(role, minimum)) { @@ -156,3 +164,25 @@ export async function requireJob( const role = check(row?.role ?? null, minimum, `Print job ${jobId} not found`, action); return { farmId: row!.farmId, orgId: row!.orgId, role }; } + +/** As `requireFarm`, resolved through the alert's farm. */ +export async function requireAlert( + userId: string, + alertId: string, + minimum: OrgRole, + action = "this operation", +): Promise<{ farmId: string; orgId: string; role: OrgRole }> { + const [row] = await db + .select({ farmId: printerAlerts.farmId, orgId: farms.orgId, role: orgMembers.role }) + .from(printerAlerts) + .innerJoin(farms, eq(farms.id, printerAlerts.farmId)) + .leftJoin( + orgMembers, + and(eq(orgMembers.orgId, farms.orgId), eq(orgMembers.userId, userId)), + ) + .where(eq(printerAlerts.id, alertId)) + .limit(1); + + const role = check(row?.role ?? null, minimum, `Alert ${alertId} not found`, action); + return { farmId: row!.farmId, orgId: row!.orgId, role }; +} diff --git a/src/db/schema/alerts.ts b/src/db/schema/alerts.ts new file mode 100644 index 0000000..9c3cd52 --- /dev/null +++ b/src/db/schema/alerts.ts @@ -0,0 +1,45 @@ +import { pgTable, uuid, text, timestamp, real, pgEnum, index } from "drizzle-orm/pg-core"; +import { farms } from "./farms"; +import { printers } from "./printers"; + +export const alertKindEnum = pgEnum("alert_kind", [ + "print_failed", + "print_paused", + "printer_error", + "ai_detection", +]); + +/** + * The permanent record of something an operator should look at. + * + * A table and not a channel: Valkey PUBLISH keeps nothing, so an alert raised + * while no console was open would otherwise be gone for good. The console + * lists open (unacknowledged) rows, which is also what makes "I was away, + * what happened" answerable. + */ +export const printerAlerts = pgTable( + "printer_alerts", + { + id: uuid("id").primaryKey().defaultRandom(), + farmId: uuid("farm_id") + .notNull() + .references(() => farms.id, { onDelete: "cascade" }), + // Set null rather than cascade: the history of why a deleted printer was + // flagged is still worth having, and the farm cascade bounds it anyway. + printerId: uuid("printer_id").references(() => printers.id, { onDelete: "set null" }), + kind: alertKindEnum("kind").notNull(), + message: text("message").notNull(), + // Only for ai_detection; null is "not applicable", not zero confidence. + confidence: real("confidence"), + // Where the frame the detection came from lives, for ai_detection. + objectKey: text("object_key"), + acknowledgedBy: uuid("acknowledged_by"), + acknowledgedAt: timestamp("acknowledged_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), + }, + (table) => ({ + farmOpen: index("printer_alerts_farm_open_idx").on(table.farmId, table.acknowledgedAt), + }), +); + +export type PrinterAlert = typeof printerAlerts.$inferSelect; diff --git a/src/db/schema/index.ts b/src/db/schema/index.ts index 2d1b9a6..0c22411 100644 --- a/src/db/schema/index.ts +++ b/src/db/schema/index.ts @@ -3,3 +3,4 @@ export * from "./farms"; export * from "./printers"; export * from "./print_jobs"; export * from "./embeddings"; +export * from "./alerts"; diff --git a/src/index.ts b/src/index.ts index 17f509f..192d375 100644 --- a/src/index.ts +++ b/src/index.ts @@ -10,6 +10,8 @@ import { orgRoutes } from "./routes/orgs"; import { farmRoutes } from "./routes/farms"; import { printerRoutes } from "./routes/printers"; import { jobRoutes } from "./routes/jobs"; +import { alertRoutes } from "./routes/alerts"; +import { startAiAlertListener } from "./alerts/listener"; import { edgeWsRoute } from "./routes/ws/edge"; import { telemetryWsRoute } from "./routes/ws/telemetry"; import { startDispatcher } from "./dispatch"; @@ -27,6 +29,7 @@ const app = new Elysia() { name: "farms", description: "Sites, each belonging to one organisation" }, { name: "printers", description: "Machines, each belonging to one farm" }, { name: "jobs", description: "Print jobs" }, + { name: "alerts", description: "Things an operator should look at" }, ], components: { securitySchemes: { @@ -62,6 +65,7 @@ const app = new Elysia() .use(farmRoutes) .use(printerRoutes) .use(jobRoutes) + .use(alertRoutes) .use(edgeWsRoute) .use(telemetryWsRoute) .listen(Number(process.env.PORT ?? 3000)); @@ -72,5 +76,6 @@ console.log(`continuum-backend listening on :${app.server?.port}`); // only ever sends work to farms with a connected gateway, so on a fresh boot // it does nothing until one connects. startDispatcher(); +startAiAlertListener(); export type App = typeof app; diff --git a/src/routes/alerts.ts b/src/routes/alerts.ts new file mode 100644 index 0000000..5cc1a8a --- /dev/null +++ b/src/routes/alerts.ts @@ -0,0 +1,59 @@ +import { Elysia, t } from "elysia"; +import { and, desc, eq, isNull } from "drizzle-orm"; +import { NotFoundError } from "@continuum/ts-core"; +import { db } from "../db"; +import { printerAlerts } from "../db/schema"; +import { requireAlert, requireUser, visibleAlerts } from "../auth"; +import { PaginationQuery, paginate } from "./pagination"; + +const IdParams = t.Object({ id: t.String({ format: "uuid" }) }); + +export const alertRoutes = new Elysia({ prefix: "/alerts" }) + .use(requireUser) + .get( + "/", + async ({ user, query }) => { + const filters = [visibleAlerts(user.id)]; + if (query.farmId) filters.push(eq(printerAlerts.farmId, query.farmId)); + // Open means not acknowledged: what is still waiting on a person. + if (query.open) filters.push(isNull(printerAlerts.acknowledgedAt)); + return db.query.printerAlerts.findMany({ + where: and(...filters), + orderBy: [desc(printerAlerts.createdAt)], + ...paginate(query), + }); + }, + { + query: t.Object({ + farmId: t.Optional(t.String({ format: "uuid" })), + open: t.Optional(t.BooleanString()), + ...PaginationQuery, + }), + detail: { tags: ["alerts"], summary: "Alerts in the caller's organisations" }, + }, + ) + /** + * `operator`, not `viewer`: acknowledging clears it from everyone's list, so + * it is a decision about the farm and not a read. + * + * Idempotent on purpose - the second person to click gets the same answer, + * and the first acknowledger stays on the record. + */ + .post( + "/:id/acknowledge", + async ({ user, params }) => { + await requireAlert(user.id, params.id, "operator", "acknowledging an alert"); + const [row] = await db + .update(printerAlerts) + .set({ acknowledgedBy: user.id, acknowledgedAt: new Date() }) + .where(and(eq(printerAlerts.id, params.id), isNull(printerAlerts.acknowledgedAt))) + .returning(); + if (row) return row; + const existing = await db.query.printerAlerts.findFirst({ + where: eq(printerAlerts.id, params.id), + }); + if (!existing) throw new NotFoundError(`Alert ${params.id} not found`); + return existing; + }, + { params: IdParams, detail: { tags: ["alerts"], summary: "Mark an alert as dealt with" } }, + ); diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index eaad3f4..db7753e 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -9,7 +9,8 @@ import { type PrinterTelemetry, } from "@continuum/ts-types"; import { db } from "../../db"; -import { printers } from "../../db/schema"; +import { printerAlerts, printers } from "../../db/schema"; +import { deriveAlert, type Status } from "../../alerts/derive"; import { advanceJobsFrom } from "../../dispatch"; import { enqueueFrameIngestionBatch } from "../../queue"; import { edgeJwtSecret } from "../../auth/edge_secret"; @@ -108,6 +109,12 @@ interface EdgeConn { * re-query the same mapping every few seconds. */ printerIds: Map; + /** + * The last status seen per printer on THIS connection, to tell a change + * from a level when deciding whether to raise an alert. Per connection on + * purpose: after a reconnect nothing is known, so nothing is raised. + */ + lastStatus: Map; /** Commands sent but not yet answered, keyed by command_id. */ pending: Map; /** @@ -432,6 +439,25 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { }); } + // After the write, and never allowed to break the batch: an alert that + // fails to save is a lost notification, not a reason to stop recording + // the rest of the farm's telemetry. + if (updated) { + const alert = deriveAlert( + conn.lastStatus.get(printerId), + status, + updated.name, + snapshot.state, + ); + if (status) conn.lastStatus.set(printerId, status); + if (alert) { + await db + .insert(printerAlerts) + .values({ farmId: conn.farmId, printerId, kind: alert.kind, message: alert.message }) + .catch((err: unknown) => console.warn("[alerts] could not record an alert", err)); + } + } + // The printer's row is the latest snapshot; the job's row is the history // that outlives it. This is the only thing that moves a job out of // `uploading` - see `advanceJobsFrom` for why the gateway's `ok` on a @@ -616,6 +642,7 @@ export const edgeWsRoute = new Elysia() lastSeenAt: Date.now(), send: (message) => ws.send(message), printerIds: new Map(), + lastStatus: new Map(), pending: new Map(), webrtc: new Map(), }; diff --git a/tests/alerts.test.ts b/tests/alerts.test.ts new file mode 100644 index 0000000..aec10d7 --- /dev/null +++ b/tests/alerts.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, test } from "bun:test"; +import { deriveAlert, minAiConfidence, worthRaising } from "../src/alerts/derive"; + +describe("deriveAlert", () => { + test("a fault during a print is a failed print", () => { + expect(deriveAlert("printing", "error", "H2C")?.kind).toBe("print_failed"); + expect(deriveAlert("paused", "error", "H2C")?.kind).toBe("print_failed"); + }); + + test("a fault on an idle printer is a printer error, not a failed print", () => { + expect(deriveAlert("idle", "error", "P1P")?.kind).toBe("printer_error"); + expect(deriveAlert("offline", "error", "P1P")?.kind).toBe("printer_error"); + }); + + test("the printer pausing itself is announced, with its own words", () => { + const alert = deriveAlert("printing", "paused", "H2C", "PAUSE"); + expect(alert?.kind).toBe("print_paused"); + expect(alert?.message).toBe("H2C: the print is paused (printer reports PAUSE)"); + }); + + test("a level is not an alert: staying in error raises nothing", () => { + expect(deriveAlert("error", "error", "H2C")).toBeUndefined(); + }); + + test("a printer first seen on this connection raises nothing", () => { + expect(deriveAlert(undefined, "error", "H2C")).toBeUndefined(); + }); + + test("an unmapped status never raises", () => { + expect(deriveAlert("printing", undefined, "H2C")).toBeUndefined(); + }); + + test("ordinary transitions are quiet", () => { + expect(deriveAlert("idle", "printing", "H2C")).toBeUndefined(); + expect(deriveAlert("printing", "idle", "H2C")).toBeUndefined(); + expect(deriveAlert("paused", "printing", "H2C")).toBeUndefined(); + }); +}); + +describe("AI detection floor", () => { + test("defaults to 0.6 and respects the environment", () => { + expect(minAiConfidence({})).toBe(0.6); + expect(minAiConfidence({ ALERT_MIN_CONFIDENCE: "0.9" })).toBe(0.9); + }); + test("an out-of-range or non-numeric floor falls back rather than disabling alerts", () => { + expect(minAiConfidence({ ALERT_MIN_CONFIDENCE: "7" })).toBe(0.6); + expect(minAiConfidence({ ALERT_MIN_CONFIDENCE: "lots" })).toBe(0.6); + }); + test("worthRaising is inclusive at the floor", () => { + expect(worthRaising(0.6, 0.6)).toBe(true); + expect(worthRaising(0.59, 0.6)).toBe(false); + }); +}); + +import { parsePrintAlert } from "../src/alerts/parse"; + +describe("parsePrintAlert", () => { + const good = { + printer_id: "0b6d3b4e-8f58-4a3b-9c69-9f3a2f6a0a11", + label: "spaghetti", + confidence: 0.91, + captured_at: "2026-10-29T10:00:00Z", + r2_object_key: "frames/x.jpg", + }; + test("accepts exactly the published shape", () => { + expect(parsePrintAlert(JSON.stringify(good))?.label).toBe("spaghetti"); + }); + test("rejects garbage, missing fields and out-of-range confidence instead of throwing", () => { + expect(parsePrintAlert("not json")).toBeUndefined(); + expect(parsePrintAlert(JSON.stringify({ ...good, printer_id: undefined }))).toBeUndefined(); + expect(parsePrintAlert(JSON.stringify({ ...good, confidence: 3 }))).toBeUndefined(); + // Tolerant of extra fields: a newer worker adding one must not blind this consumer. + expect(parsePrintAlert(JSON.stringify({ ...good, extra: 1 }))?.label).toBe("spaghetti"); + }); +});