From b485e9870bf03a66fd79251f6c66221ef9cfc9a1 Mon Sep 17 00:00:00 2001 From: Julian Cuni Date: Mon, 7 Sep 2026 08:26:22 +0200 Subject: [PATCH] =?UTF-8?q?feat(collector):=20review=20collector=20skeleto?= =?UTF-8?q?n=20=E2=80=94=20apps/collector,=20its=20own=20Komodo=20stack=20?= =?UTF-8?q?on=20the=20reviewer's=20host?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The far end of the Car Wash review outbox (wiki/concepts/vision-review-outbox.md): a small Fastify + SQLite service in the monorepo (shares the payload contract and the class vocabulary via @parking/shared), delivered to art-docker-station by its own stack so nothing booth-side lands there and nothing of it on a booth. - POST /ingest: bearer token per booth (constant-time), X-Booth-Id must match, multipart meta + JPEG (magic checked, 2 MB cap), meta validated against the contract, idempotent on the item id; crop stored at crops//.jpg on the volume + one items row. - /review + /api/*: the reviewer's screen served by the process (Basic auth, one login): one pending crop at a time, operator's pick and camera's pick beside it, one button/key per vocabulary class + unusable + skip; stats per booth and per hashed operator (agree / disagree / unusable — disagree = the reviewer's class is outside the operator's category). - GET /export/labels.csv: reviewed usable rows for training; formula-leading cells are neutralised (booth-supplied names). Crops stay on the volume for the trainer on the host. - Booth payload now carries operatorCategory.classes so the comparison needs no site setup. - Delivery: apps/collector/Dockerfile (monorepo context), docker-compose.collector.yml (bind to the overlay IP; commented `trainer` profile seam for the GPU), a third build step in build-images.yml, a `wash-collector` stack in komodo/resources.toml with one secret per booth referenced from both the collector's token list and the booth's own stack (park-2 lines templated, commented, DNS name for the URL). - Tests: app.test.ts (ingest ok/dup/refusals, review + stats + export, config). Image built and smoke-tested locally (health, ingest, duplicate, auth, verdict, export). Claude-Session: https://claude.ai/code/session_01FWncR69HgGPuei1dLrW3cU --- .gitea/workflows/build-images.yml | 15 +- apps/collector/.env.example | 16 ++ apps/collector/Dockerfile | 48 ++++ apps/collector/package.json | 28 +++ apps/collector/src/app.test.ts | 141 +++++++++++ apps/collector/src/app.ts | 231 ++++++++++++++++++ apps/collector/src/config.ts | 36 +++ apps/collector/src/db.ts | 179 ++++++++++++++ apps/collector/src/index.ts | 16 ++ apps/collector/src/review-page.ts | 117 +++++++++ apps/collector/tsconfig.json | 10 + apps/collector/vitest.config.ts | 5 + apps/server/Dockerfile | 1 + .../src/modules/carwash/review-outbox.test.ts | 4 +- .../src/modules/carwash/review-outbox.ts | 5 +- apps/server/src/modules/carwash/service.ts | 1 + docker-compose.collector.yml | 47 ++++ komodo/resources.toml | 43 ++++ pnpm-lock.yaml | 52 ++++ wiki/concepts/vision-review-outbox.md | 41 +++- wiki/decisions/fleet-deployment-komodo.md | 8 + wiki/log.md | 11 + 22 files changed, 1045 insertions(+), 10 deletions(-) create mode 100644 apps/collector/.env.example create mode 100644 apps/collector/Dockerfile create mode 100644 apps/collector/package.json create mode 100644 apps/collector/src/app.test.ts create mode 100644 apps/collector/src/app.ts create mode 100644 apps/collector/src/config.ts create mode 100644 apps/collector/src/db.ts create mode 100644 apps/collector/src/index.ts create mode 100644 apps/collector/src/review-page.ts create mode 100644 apps/collector/tsconfig.json create mode 100644 apps/collector/vitest.config.ts create mode 100644 docker-compose.collector.yml diff --git a/.gitea/workflows/build-images.yml b/.gitea/workflows/build-images.yml index 424dcff..ab1e6f7 100644 --- a/.gitea/workflows/build-images.yml +++ b/.gitea/workflows/build-images.yml @@ -1,6 +1,6 @@ name: Build & push images -# Build the SERVER (API + SPA) and VISION (ANPR) container images and push them to the +# Build the SERVER (API + SPA), COLLECTOR (wash review) and VISION (ANPR) container images and push them to the # house Gitea registry, tagged by BRANCH + short SHA (branch-aware: dev→:dev, stage→:stage, # main→:main). Separate from ci.yml (checks-only) and release.yml (tag-only desktop bundle). # Mirrors the house pattern (cf. trm/processor build.yml). See @@ -13,6 +13,7 @@ on: - 'apps/server/**' - 'apps/web/**' - 'apps/vision/**' + - 'apps/collector/**' - 'packages/**' - 'package.json' - 'pnpm-lock.yaml' @@ -100,6 +101,18 @@ jobs: cache-from: type=registry,ref=${{ env.REGISTRY }}/parking-server:buildcache cache-to: type=registry,ref=${{ env.REGISTRY }}/parking-server:buildcache,mode=max + - name: Build & push COLLECTOR (wash review) + uses: docker/build-push-action@v5 + with: + context: . + file: apps/collector/Dockerfile + push: true + tags: | + ${{ env.REGISTRY }}/parking-collector:${{ steps.meta.outputs.branch }} + ${{ env.REGISTRY }}/parking-collector:${{ steps.meta.outputs.branch }}-${{ steps.meta.outputs.sha }} + cache-from: type=registry,ref=${{ env.REGISTRY }}/parking-collector:buildcache + cache-to: type=registry,ref=${{ env.REGISTRY }}/parking-collector:buildcache,mode=max + - name: Build & push VISION (ANPR) uses: docker/build-push-action@v5 with: diff --git a/apps/collector/.env.example b/apps/collector/.env.example new file mode 100644 index 0000000..8ab3f82 --- /dev/null +++ b/apps/collector/.env.example @@ -0,0 +1,16 @@ +# Car Wash review collector (wiki/concepts/vision-review-outbox.md). Runs on the +# reviewer's host (art-docker-station), reachable by the booths ONLY over the Netbird +# overlay. Deployed by its own Komodo stack (komodo/resources.toml, "wash-collector"). + +# COLLECTOR_HOST=0.0.0.0 # in Docker the compose file binds the published port to the overlay IP +# COLLECTOR_PORT=8090 +# COLLECTOR_DATA_DIR=/data # collector.sqlite + crops//.jpg + +# One bearer token per booth: ":" pairs, comma- or newline-separated. The +# booth id is the pseudonymous CARWASH_REVIEW_BOOTH_ID that booth was deployed with — never +# a site name. Generate tokens with: openssl rand -hex 32 +COLLECTOR_BOOTH_TOKENS=booth-7:REPLACE,booth-9:REPLACE + +# The reviewer's login for the review screen and the export (HTTP Basic over the overlay). +COLLECTOR_REVIEWER_USER=reviewer +COLLECTOR_REVIEWER_PASS=REPLACE diff --git a/apps/collector/Dockerfile b/apps/collector/Dockerfile new file mode 100644 index 0000000..2a2ac0e --- /dev/null +++ b/apps/collector/Dockerfile @@ -0,0 +1,48 @@ +# parking-collector — the Car Wash review collector (wiki/concepts/vision-review-outbox.md). +# Built from the monorepo root (context: .) like the server image, so it shares the +# lockfile and @parking/shared. Runs on the REVIEWER's host (not a booth), delivered by +# its own Komodo stack (docker-compose.collector.yml). Data on /data: collector.sqlite + +# crops//.jpg — the trainer on the same host reads the crops off that volume. + +FROM node:22-alpine AS deps +WORKDIR /app +RUN apk add --no-cache python3 make g++ # node-gyp for better-sqlite3 +RUN corepack enable && corepack prepare pnpm@10.24.0 --activate +COPY package.json pnpm-lock.yaml pnpm-workspace.yaml turbo.json ./ +COPY apps/server/package.json apps/server/ +COPY apps/web/package.json apps/web/ +COPY apps/vision/package.json apps/vision/ +COPY apps/collector/package.json apps/collector/ +COPY packages/db/package.json packages/db/ +COPY packages/devices/package.json packages/devices/ +COPY packages/shared/package.json packages/shared/ +RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \ + pnpm fetch + +FROM deps AS build +ENV CI=true +COPY . . +RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \ + pnpm install --frozen-lockfile --offline +RUN pnpm turbo run build --filter=@parking/collector +RUN --mount=type=cache,id=pnpm-store,target=/root/.local/share/pnpm/store \ + pnpm --filter=@parking/collector --legacy deploy --prod /deploy + +FROM node:22-alpine AS runtime +WORKDIR /app +ARG BUILD_VERSION="" +ENV BUILD_VERSION=$BUILD_VERSION +ENV NODE_ENV=production +RUN apk add --no-cache libstdc++ wget # better-sqlite3 native runtime; wget for the healthcheck +RUN addgroup -S app && adduser -S -G app app +COPY --from=build --chown=app:app /deploy ./ +ENV COLLECTOR_DATA_DIR=/data +ENV COLLECTOR_HOST=0.0.0.0 +ENV COLLECTOR_PORT=8090 +RUN mkdir -p /data && chown app:app /data +VOLUME ["/data"] +USER app +EXPOSE 8090 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ + CMD wget -qO- "http://localhost:${COLLECTOR_PORT:-8090}/health" >/dev/null 2>&1 || exit 1 +CMD ["node", "dist/index.js"] diff --git a/apps/collector/package.json b/apps/collector/package.json new file mode 100644 index 0000000..e244da6 --- /dev/null +++ b/apps/collector/package.json @@ -0,0 +1,28 @@ +{ + "name": "@parking/collector", + "version": "0.0.0", + "private": true, + "type": "module", + "description": "Car Wash review collector: receives plate-blurred vehicle crops + the operator's category choice from booths over the private overlay, serves the reviewer's screen, exports labels for training. See wiki/concepts/vision-review-outbox.md.", + "scripts": { + "build": "tsc -b", + "dev": "tsx watch --env-file-if-exists=.env src/index.ts", + "start": "node --env-file-if-exists=.env dist/index.js", + "typecheck": "tsc --noEmit", + "lint": "tsc --noEmit", + "test": "vitest run" + }, + "dependencies": { + "@fastify/multipart": "^9.2.1", + "@parking/shared": "workspace:*", + "better-sqlite3": "12.10.1", + "fastify": "5.8.5" + }, + "devDependencies": { + "@types/better-sqlite3": "7.6.13", + "@types/node": "25.9.3", + "tsx": "4.22.4", + "typescript": "6.0.3", + "vitest": "^4.1.9" + } +} diff --git a/apps/collector/src/app.test.ts b/apps/collector/src/app.test.ts new file mode 100644 index 0000000..c8d6755 --- /dev/null +++ b/apps/collector/src/app.test.ts @@ -0,0 +1,141 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { buildCollector, type CollectorApp } from "./app.js"; +import { parseBoothTokens } from "./config.js"; + +// The collector: one ingest surface (bearer per booth, idempotent), one review surface +// (Basic), one export. Exercised over app.inject with a hand-built multipart body. + +let app: CollectorApp; +let dir: string; +const TOKENS = new Map([["booth-7", "0123456789abcdef0123456789abcdef"], ["booth-9", "fedcba9876543210fedcba9876543210"]]); +const REVIEWER = { user: "julian", pass: "review-pass-123" }; +const basic = "Basic " + Buffer.from(`${REVIEWER.user}:${REVIEWER.pass}`).toString("base64"); + +beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), "collector-")); + app = await buildCollector({ host: "127.0.0.1", port: 0, dataDir: dir, boothTokens: TOKENS, reviewer: REVIEWER }, { dbFile: ":memory:" }); + await app.ready(); +}); +afterEach(async () => { + await app.close(); + await rm(dir, { recursive: true, force: true }); +}); + +/** A minimal JPEG-looking blob (SOI marker + padding) — the collector checks the magic only. */ +const JPEG = Buffer.concat([Buffer.from([0xff, 0xd8, 0xff, 0xe0]), Buffer.alloc(200, 1)]); + +function meta(over: Record = {}) { + return { + v: 1, booth: "booth-7", item: "item-1", order: "o-1", at: "2026-09-06T10:00:00.000Z", operator: "ab12cd34ef56ab12", + operatorCategory: { id: "car", name: "Vetura", classes: ["car", "sedan", "hatchback"] }, service: "Standard", + vision: { class: "suv", confidence: 0.91, categoryId: "suv" }, downgraded: true, + image: { width: 320, height: 200, plateBlurred: true }, + ...over, + }; +} + +function multipart(fields: Record, file: Buffer | null): { body: Buffer; type: string } { + const b = "----collector-test"; + const parts: Buffer[] = []; + for (const [k, v] of Object.entries(fields)) parts.push(Buffer.from(`--${b}\r\nContent-Disposition: form-data; name="${k}"\r\n\r\n${v}\r\n`)); + if (file) parts.push(Buffer.from(`--${b}\r\nContent-Disposition: form-data; name="image"; filename="x.jpg"\r\nContent-Type: image/jpeg\r\n\r\n`), file, Buffer.from("\r\n")); + parts.push(Buffer.from(`--${b}--\r\n`)); + return { body: Buffer.concat(parts), type: `multipart/form-data; boundary=${b}` }; +} + +async function ingest(m: Record, token = TOKENS.get("booth-7")!, file: Buffer | null = JPEG, extra: Record = {}) { + const { body, type } = multipart({ meta: JSON.stringify(m) }, file); + return app.inject({ method: "POST", url: "/ingest", headers: { authorization: `Bearer ${token}`, "content-type": type, ...extra }, payload: body }); +} + +describe("ingest", () => { + it("stores the crop and the decision under the token's booth; retries are idempotent", async () => { + const r = await ingest(meta()); + expect(r.statusCode).toBe(201); + const row = app.collectorDb.get("item-1")!; + expect(row).toMatchObject({ booth: "booth-7", operatorCategoryName: "Vetura", visionClass: "suv", downgraded: 1, plateBlurred: 1, imagePath: "crops/booth-7/item-1.jpg" }); + expect(JSON.parse(row.operatorClasses)).toEqual(["car", "sedan", "hatchback"]); + const again = await ingest(meta()); + expect(again.statusCode).toBe(200); + expect(again.json()).toEqual({ ok: true, duplicate: true }); + expect((await app.inject({ method: "GET", url: "/health" })).json()).toMatchObject({ ok: true, booths: 1, pending: 1 }); + }); + + it("refuses a bad token, a booth mismatch, a non-JPEG, and malformed meta", async () => { + expect((await ingest(meta(), "nope-nope-nope-nope-nope")).statusCode).toBe(401); + expect((await ingest(meta({ booth: "booth-9" }))).statusCode).toBe(422); // token is booth-7's + expect((await ingest(meta(), TOKENS.get("booth-7")!, JPEG, { "x-booth-id": "booth-9" })).statusCode).toBe(403); + expect((await ingest(meta(), TOKENS.get("booth-7")!, Buffer.alloc(300, 7))).statusCode).toBe(415); + expect((await ingest(meta(), TOKENS.get("booth-7")!, null)).statusCode).toBe(400); + expect((await ingest(meta({ vision: { class: "spaceship", confidence: 0.5, categoryId: null } }))).statusCode).toBe(422); + expect((await ingest(meta({ item: "../../etc/passwd" }))).statusCode).toBe(422); + expect((await ingest(meta({ v: 2 }))).statusCode).toBe(422); + expect(app.collectorDb.stats().booths).toEqual([]); + }); +}); + +describe("review + export", () => { + it("the reviewer lists pending items, sees the crop, labels it; stats compare the label with the operator's category; the export lists usable labels only", async () => { + await ingest(meta()); + await ingest(meta({ item: "item-2", operator: "ab12cd34ef56ab12", vision: { class: "car", confidence: 0.8, categoryId: "car" }, downgraded: false })); + await ingest(meta({ item: "item-3", booth: "booth-9", operator: "9999999999999999" }), TOKENS.get("booth-9")!); + + // No login → 401 with a challenge; nothing without a configured reviewer is tested in config. + const anon = await app.inject({ method: "GET", url: "/api/items" }); + expect(anon.statusCode).toBe(401); + expect(anon.headers["www-authenticate"]).toContain("Basic"); + expect((await app.inject({ method: "GET", url: "/review", headers: { authorization: basic } })).headers["content-type"]).toContain("text/html"); + + const list = (await app.inject({ method: "GET", url: "/api/items?status=pending", headers: { authorization: basic } })).json(); + expect(list.items.map((i: { id: string }) => i.id)).toEqual(["item-1", "item-2", "item-3"]); + expect(list.items[0].imagePath).toBeUndefined(); + const img = await app.inject({ method: "GET", url: "/api/items/item-1/image", headers: { authorization: basic } }); + expect(img.statusCode).toBe(200); + expect(img.headers["content-type"]).toBe("image/jpeg"); + expect(img.rawPayload.subarray(0, 3)).toEqual(Buffer.from([0xff, 0xd8, 0xff])); + + // item-1: operator said Vetura (car/sedan/hatchback), reviewer says suv → disagree. + // item-2: reviewer says sedan → inside Vetura → agree. item-3: unusable. + const post = (id: string, label: string) => + app.inject({ method: "POST", url: `/api/items/${id}/review`, headers: { authorization: basic, "content-type": "application/json" }, payload: { label } }); + expect((await post("item-1", "suv")).json()).toMatchObject({ reviewLabel: "suv", reviewer: "julian" }); + expect((await post("item-2", "sedan")).statusCode).toBe(200); + expect((await post("item-3", "unusable")).statusCode).toBe(200); + expect((await post("item-3", "spaceship")).statusCode).toBe(400); + expect((await post("nope", "suv")).statusCode).toBe(404); + + const stats = (await app.inject({ method: "GET", url: "/api/stats", headers: { authorization: basic } })).json(); + expect(stats.booths).toEqual([ + { booth: "booth-7", received: 2, pending: 0, reviewed: 2 }, + { booth: "booth-9", received: 1, pending: 0, reviewed: 1 }, + ]); + expect(stats.operators).toEqual([ + { booth: "booth-7", operatorRef: "ab12cd34ef56ab12", reviewed: 2, agree: 1, disagree: 1, unusable: 0 }, + { booth: "booth-9", operatorRef: "9999999999999999", reviewed: 1, agree: 0, disagree: 0, unusable: 1 }, + ]); + + const csv = await app.inject({ method: "GET", url: "/export/labels.csv", headers: { authorization: basic } }); + expect(csv.statusCode).toBe(200); + const lines = csv.body.trim().split("\n"); + expect(lines[0]).toBe("item,booth,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"); + expect(lines).toHaveLength(3); // header + 2 usable labels; the unusable one is left out + expect(lines[1]).toContain('"item-1","booth-7","crops/booth-7/item-1.jpg","suv","Vetura","car|sedan|hatchback","suv"'); + + // A booth-supplied name that looks like a spreadsheet formula is neutralised in the export. + await ingest(meta({ item: "item-4", operatorCategory: { id: "x", name: "=HYPERLINK(\"http://evil\")", classes: ["car"] } })); + await post("item-4", "car"); + const csv2 = (await app.inject({ method: "GET", url: "/export/labels.csv", headers: { authorization: basic } })).body; + expect(csv2).toContain(`"'=HYPERLINK(""http://evil"")"`); + }); +}); + +describe("config", () => { + it("parses booth:token pairs and refuses short tokens", () => { + expect([...parseBoothTokens("a:0123456789abcdef, b:fedcba9876543210\nc:0000000000000000").keys()]).toEqual(["a", "b", "c"]); + expect(() => parseBoothTokens("a:short")).toThrow(/too short/); + expect(() => parseBoothTokens("nocolon")).toThrow(/bad pair/); + }); +}); diff --git a/apps/collector/src/app.ts b/apps/collector/src/app.ts new file mode 100644 index 0000000..cbaa298 --- /dev/null +++ b/apps/collector/src/app.ts @@ -0,0 +1,231 @@ +import { timingSafeEqual } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; +import Fastify, { type FastifyInstance, type FastifyReply, type FastifyRequest } from "fastify"; +import multipart from "@fastify/multipart"; +import { isVehicleClass } from "@parking/shared"; +import type { CollectorConfig } from "./config.js"; +import { CollectorDb, type ItemRow, type ReviewVerdict } from "./db.js"; +import { reviewPage } from "./review-page.js"; + +// The collector — the far end of the booth's review outbox +// (wiki/concepts/vision-review-outbox.md). Three surfaces and nothing else: +// POST /ingest one package from one booth (bearer token per booth; idempotent) +// /review + /api/* the reviewer's screen (HTTP Basic, one login) +// GET /export/labels.csv the training set: reviewed, usable rows (crops sit beside it on +// the volume, so the trainer on this host reads them directly) +// It deliberately has no fleet features and no path back into a booth. + +/** The package's `meta` part, as the booth sends it (review-outbox.ts). */ +interface IngestMeta { + v: number; + booth: string; + item: string; + order: string; + at: string; + operator: string; + operatorCategory: { id: string; name: string; classes?: string[] }; + service: string; + vision: { class: string; confidence: number; categoryId: string | null }; + downgraded: boolean; + image: { width: number; height: number; plateBlurred: boolean }; +} + +const ID_RE = /^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$/; +const MAX_IMAGE_BYTES = 2 * 1024 * 1024; + +function str(v: unknown, max = 200): string | null { + return typeof v === "string" && v.length > 0 && v.length <= max ? v : null; +} + +/** Validate the meta part; returns a message on the first problem. */ +function checkMeta(m: unknown, booth: string): { ok: true; meta: IngestMeta } | { ok: false; why: string } { + if (!m || typeof m !== "object") return { ok: false, why: "meta must be an object" }; + const x = m as Record; + if (x.v !== 1) return { ok: false, why: "unsupported meta version" }; + if (x.booth !== booth) return { ok: false, why: "meta.booth does not match the token's booth" }; + if (!str(x.item, 64) || !ID_RE.test(x.item as string)) return { ok: false, why: "bad item id" }; + if (!str(x.order, 64)) return { ok: false, why: "bad order ref" }; + if (!str(x.at, 40) || Number.isNaN(Date.parse(x.at as string))) return { ok: false, why: "bad timestamp" }; + if (!str(x.operator, 64)) return { ok: false, why: "bad operator ref" }; + const oc = x.operatorCategory as Record | undefined; + if (!oc || !str(oc.id, 64) || !str(oc.name, 120)) return { ok: false, why: "bad operatorCategory" }; + if (oc.classes !== undefined && (!Array.isArray(oc.classes) || !oc.classes.every(isVehicleClass))) return { ok: false, why: "bad operatorCategory.classes" }; + if (!str(x.service, 120)) return { ok: false, why: "bad service" }; + const v = x.vision as Record | undefined; + if (!v || !isVehicleClass(v.class) || typeof v.confidence !== "number" || v.confidence < 0 || v.confidence > 1) return { ok: false, why: "bad vision read" }; + if (v.categoryId != null && !str(v.categoryId, 64)) return { ok: false, why: "bad vision.categoryId" }; + if (typeof x.downgraded !== "boolean") return { ok: false, why: "bad downgraded" }; + const im = x.image as Record | undefined; + if (!im || typeof im.width !== "number" || typeof im.height !== "number" || typeof im.plateBlurred !== "boolean") return { ok: false, why: "bad image meta" }; + return { ok: true, meta: x as unknown as IngestMeta }; +} + +function safeEqual(a: string, b: string): boolean { + const ba = Buffer.from(a); + const bb = Buffer.from(b); + return ba.length === bb.length && timingSafeEqual(ba, bb); +} + +export interface CollectorApp extends FastifyInstance { + collectorDb: CollectorDb; +} + +export async function buildCollector(cfg: CollectorConfig, opts: { dbFile?: string } = {}): Promise { + await mkdir(path.join(cfg.dataDir, "crops"), { recursive: true }); + const db = new CollectorDb(opts.dbFile ?? path.join(cfg.dataDir, "collector.sqlite")); + const app = Fastify({ logger: { level: process.env.LOG_LEVEL ?? "info" }, bodyLimit: 64 * 1024 }) as unknown as CollectorApp; + app.collectorDb = db; + await app.register(multipart, { limits: { fileSize: MAX_IMAGE_BYTES, files: 1, fields: 4, parts: 6 } }); + app.addHook("onClose", async () => db.close()); + + /** Which booth this bearer token belongs to, or null. Constant-time per candidate. */ + function boothForToken(req: FastifyRequest): string | null { + const h = req.headers.authorization ?? ""; + if (!h.startsWith("Bearer ")) return null; + const token = h.slice(7).trim(); + let found: string | null = null; + for (const [booth, t] of cfg.boothTokens) if (safeEqual(token, t)) found = booth; + return found; + } + + /** HTTP Basic for the reviewer. */ + async function requireReviewer(req: FastifyRequest, reply: FastifyReply): Promise { + if (!cfg.reviewer) return reply.code(503).send({ error: "reviewer login not configured" }); + const h = req.headers.authorization ?? ""; + if (h.startsWith("Basic ")) { + const [user, ...rest] = Buffer.from(h.slice(6), "base64").toString("utf8").split(":"); + const pass = rest.join(":"); + if (user && safeEqual(user, cfg.reviewer.user) && safeEqual(pass, cfg.reviewer.pass)) return; + } + return reply.code(401).header("www-authenticate", 'Basic realm="wash review", charset="UTF-8"').send({ error: "unauthorized" }); + } + + app.get("/health", async () => { + const s = db.stats(); + return { ok: true, booths: s.booths.length, pending: s.booths.reduce((n, b) => n + b.pending, 0) }; + }); + + // --- Ingest (booths) ----------------------------------------------------------------- + app.post("/ingest", async (req, reply) => { + const booth = boothForToken(req); + if (!booth) return reply.code(401).send({ error: "unauthorized" }); + const claimed = req.headers["x-booth-id"]; + if (typeof claimed === "string" && claimed !== booth) return reply.code(403).send({ error: "booth id does not match the token" }); + if (!req.isMultipart()) return reply.code(415).send({ error: "multipart/form-data expected" }); + + let metaRaw: string | null = null; + let image: Buffer | null = null; + try { + for await (const part of req.parts()) { + if (part.type === "file" && part.fieldname === "image") { + image = await part.toBuffer(); + } else if (part.type === "field" && part.fieldname === "meta") { + metaRaw = String(part.value); + } + } + } catch (err) { + const code = (err as { code?: string }).code; + return reply.code(code === "FST_REQ_FILE_TOO_LARGE" ? 413 : 400).send({ error: (err as Error).message }); + } + if (!metaRaw) return reply.code(400).send({ error: "meta part missing" }); + if (!image || image.length < 100) return reply.code(400).send({ error: "image part missing" }); + if (!(image[0] === 0xff && image[1] === 0xd8 && image[2] === 0xff)) return reply.code(415).send({ error: "image must be a JPEG" }); + let parsed: unknown; + try { + parsed = JSON.parse(metaRaw); + } catch { + return reply.code(400).send({ error: "meta is not JSON" }); + } + const checked = checkMeta(parsed, booth); + if (!checked.ok) return reply.code(422).send({ error: checked.why }); + const meta = checked.meta; + + // Idempotent on the item id: a booth retrying after a lost 2xx must not duplicate. + if (db.get(meta.item)) return reply.code(200).send({ ok: true, duplicate: true }); + + const rel = path.posix.join("crops", booth, `${meta.item}.jpg`); + await mkdir(path.join(cfg.dataDir, "crops", booth), { recursive: true }); + await writeFile(path.join(cfg.dataDir, rel), image); + db.insert({ + id: meta.item, + booth, + orderRef: meta.order, + at: meta.at, + operatorRef: meta.operator, + operatorCategoryId: meta.operatorCategory.id, + operatorCategoryName: meta.operatorCategory.name, + operatorClasses: JSON.stringify(meta.operatorCategory.classes ?? []), + service: meta.service, + visionClass: meta.vision.class, + visionConfidence: meta.vision.confidence, + visionCategoryId: meta.vision.categoryId ?? null, + downgraded: meta.downgraded ? 1 : 0, + imageWidth: meta.image.width, + imageHeight: meta.image.height, + plateBlurred: meta.image.plateBlurred ? 1 : 0, + imagePath: rel, + receivedAt: new Date().toISOString(), + }); + req.log.info(`ingest: ${booth} item ${meta.item} (${meta.vision.class} → ${meta.operatorCategory.name})`); + return reply.code(201).send({ ok: true }); + }); + + // --- Review (the trusted person) ----------------------------------------------------- + const page = reviewPage(); + app.get("/", { preHandler: requireReviewer }, async (_req, reply) => reply.redirect("/review")); + app.get("/review", { preHandler: requireReviewer }, async (_req, reply) => reply.type("text/html; charset=utf-8").send(page)); + + app.get<{ Querystring: { status?: string; limit?: string; booth?: string } }>( + "/api/items", + { preHandler: requireReviewer }, + async (req) => { + const status = req.query.status === "reviewed" ? "reviewed" : "pending"; + const limit = Math.min(Math.max(Number(req.query.limit) || 25, 1), 200); + return { items: db.list(status, limit, req.query.booth || undefined).map(publicItem) }; + }, + ); + + app.get<{ Params: { id: string } }>("/api/items/:id/image", { preHandler: requireReviewer }, async (req, reply) => { + const row = db.get(req.params.id); + if (!row) return reply.code(404).send({ error: "not found" }); + return reply.type("image/jpeg").header("cache-control", "private, max-age=3600").send(createReadStream(path.join(cfg.dataDir, row.imagePath))); + }); + + app.post<{ Params: { id: string }; Body: { label?: unknown } }>("/api/items/:id/review", { preHandler: requireReviewer }, async (req, reply) => { + const label = req.body?.label; + if (label !== "unusable" && !isVehicleClass(label)) return reply.code(400).send({ error: "label must be a vehicle class or 'unusable'" }); + if (!db.get(req.params.id)) return reply.code(404).send({ error: "not found" }); + const row = db.review(req.params.id, label as ReviewVerdict, cfg.reviewer!.user); + return publicItem(row!); + }); + + app.get("/api/stats", { preHandler: requireReviewer }, async () => db.stats()); + + // --- Export (the training set) -------------------------------------------------------- + app.get("/export/labels.csv", { preHandler: requireReviewer }, async (_req, reply) => { + const rows = db.labelled(); + // Quote every cell; a cell starting like a spreadsheet formula (=, +, -, @, tab, CR) + // gets a leading apostrophe — the category/service names are booth-supplied text and + // the reviewer will open this in a spreadsheet (CSV formula injection). + const q = (s: string | number | null) => { + let v = String(s ?? ""); + if (/^[=+\-@\t\r]/.test(v)) v = `'${v}`; + return `"${v.replace(/"/g, '""')}"`; + }; + const head = "item,booth,path,label,operator_category,operator_classes,vision_class,vision_confidence,downgraded,at,reviewed_at"; + const lines = rows.map((r) => + [r.id, r.booth, r.imagePath, r.reviewLabel, r.operatorCategoryName, JSON.parse(r.operatorClasses).join("|"), r.visionClass, r.visionConfidence, r.downgraded, r.at, r.reviewedAt].map(q).join(","), + ); + return reply.type("text/csv; charset=utf-8").header("content-disposition", 'attachment; filename="labels.csv"').send([head, ...lines].join("\n") + "\n"); + }); + + return app; +} + +/** The row as the review screen sees it (no server paths). */ +function publicItem(r: ItemRow): Omit { + const { imagePath: _p, ...rest } = r; + return rest; +} diff --git a/apps/collector/src/config.ts b/apps/collector/src/config.ts new file mode 100644 index 0000000..77e579b --- /dev/null +++ b/apps/collector/src/config.ts @@ -0,0 +1,36 @@ +export interface CollectorConfig { + readonly host: string; + readonly port: number; + readonly dataDir: string; + /** boothId → bearer token. */ + readonly boothTokens: ReadonlyMap; + /** The single reviewer login; null = review screen and export refuse (503). */ + readonly reviewer: { readonly user: string; readonly pass: string } | null; +} + +/** "booth-7:abc,booth-9:def" (commas, whitespace or newlines between pairs). */ +export function parseBoothTokens(raw: string): Map { + const out = new Map(); + for (const pair of raw.split(/[,\s]+/)) { + if (!pair) continue; + const i = pair.indexOf(":"); + if (i <= 0) throw new Error(`COLLECTOR_BOOTH_TOKENS: bad pair "${pair}" (want boothId:token)`); + const booth = pair.slice(0, i).trim(); + const token = pair.slice(i + 1).trim(); + if (!booth || token.length < 16) throw new Error(`COLLECTOR_BOOTH_TOKENS: token for "${booth}" too short (>=16 chars)`); + out.set(booth, token); + } + return out; +} + +export function configFromEnv(env: NodeJS.ProcessEnv = process.env): CollectorConfig { + const user = (env.COLLECTOR_REVIEWER_USER ?? "").trim(); + const pass = env.COLLECTOR_REVIEWER_PASS ?? ""; + return { + host: env.COLLECTOR_HOST ?? "0.0.0.0", + port: Number(env.COLLECTOR_PORT ?? 8090), + dataDir: env.COLLECTOR_DATA_DIR ?? "/data", + boothTokens: parseBoothTokens(env.COLLECTOR_BOOTH_TOKENS ?? ""), + reviewer: user && pass.length >= 8 ? { user, pass } : null, + }; +} diff --git a/apps/collector/src/db.ts b/apps/collector/src/db.ts new file mode 100644 index 0000000..5273482 --- /dev/null +++ b/apps/collector/src/db.ts @@ -0,0 +1,179 @@ +import Database from "better-sqlite3"; +import type { VehicleClass } from "@parking/shared"; + +// One table. Each row is one booth decision: what the camera saw, what the operator +// chose, and (once reviewed) what a trusted person says the vehicle is. The crop itself +// lives on disk beside the DB (crops//.jpg) so the trainer on the same host +// reads it straight off the volume. + +export interface ItemRow { + id: string; + booth: string; + orderRef: string; + at: string; + operatorRef: string; + operatorCategoryId: string; + operatorCategoryName: string; + /** The vision classes the operator's category covers at that site (its mapping) — what + * lets a reviewer's CLASS be compared with an operator's CATEGORY. JSON array. */ + operatorClasses: string; + service: string; + visionClass: string; + visionConfidence: number; + visionCategoryId: string | null; + downgraded: number; + imageWidth: number; + imageHeight: number; + plateBlurred: number; + imagePath: string; + receivedAt: string; + reviewLabel: string | null; // a VehicleClass, or "unusable" + reviewedAt: string | null; + reviewer: string | null; +} + +export type ReviewVerdict = VehicleClass | "unusable"; + +export class CollectorDb { + readonly #db: Database.Database; + + constructor(file: string) { + this.#db = new Database(file); + this.#db.pragma("journal_mode = WAL"); + this.#db.exec(` + CREATE TABLE IF NOT EXISTS items ( + id TEXT PRIMARY KEY, + booth TEXT NOT NULL, + order_ref TEXT NOT NULL, + at TEXT NOT NULL, + operator_ref TEXT NOT NULL, + operator_category_id TEXT NOT NULL, + operator_category_name TEXT NOT NULL, + operator_classes TEXT NOT NULL DEFAULT '[]', + service TEXT NOT NULL, + vision_class TEXT NOT NULL, + vision_confidence REAL NOT NULL, + vision_category_id TEXT, + downgraded INTEGER NOT NULL DEFAULT 0, + image_width INTEGER NOT NULL, + image_height INTEGER NOT NULL, + plate_blurred INTEGER NOT NULL, + image_path TEXT NOT NULL, + received_at TEXT NOT NULL, + review_label TEXT, + reviewed_at TEXT, + reviewer TEXT + ); + CREATE INDEX IF NOT EXISTS items_pending ON items (reviewed_at, received_at); + CREATE INDEX IF NOT EXISTS items_booth ON items (booth, received_at); + `); + } + + close(): void { + this.#db.close(); + } + + static #map(r: Record): ItemRow { + return { + id: r.id as string, + booth: r.booth as string, + orderRef: r.order_ref as string, + at: r.at as string, + operatorRef: r.operator_ref as string, + operatorCategoryId: r.operator_category_id as string, + operatorCategoryName: r.operator_category_name as string, + operatorClasses: r.operator_classes as string, + service: r.service as string, + visionClass: r.vision_class as string, + visionConfidence: r.vision_confidence as number, + visionCategoryId: (r.vision_category_id as string | null) ?? null, + downgraded: r.downgraded as number, + imageWidth: r.image_width as number, + imageHeight: r.image_height as number, + plateBlurred: r.plate_blurred as number, + imagePath: r.image_path as string, + receivedAt: r.received_at as string, + reviewLabel: (r.review_label as string | null) ?? null, + reviewedAt: (r.reviewed_at as string | null) ?? null, + reviewer: (r.reviewer as string | null) ?? null, + }; + } + + get(id: string): ItemRow | null { + const r = this.#db.prepare("SELECT * FROM items WHERE id = ?").get(id) as Record | undefined; + return r ? CollectorDb.#map(r) : null; + } + + insert(row: Omit): void { + this.#db + .prepare( + `INSERT INTO items (id, booth, order_ref, at, operator_ref, operator_category_id, operator_category_name, + operator_classes, service, vision_class, vision_confidence, vision_category_id, downgraded, + image_width, image_height, plate_blurred, image_path, received_at) + VALUES (@id, @booth, @orderRef, @at, @operatorRef, @operatorCategoryId, @operatorCategoryName, + @operatorClasses, @service, @visionClass, @visionConfidence, @visionCategoryId, @downgraded, + @imageWidth, @imageHeight, @plateBlurred, @imagePath, @receivedAt)`, + ) + .run(row); + } + + list(status: "pending" | "reviewed", limit: number, booth?: string): ItemRow[] { + const where = [status === "pending" ? "reviewed_at IS NULL" : "reviewed_at IS NOT NULL"]; + const params: unknown[] = []; + if (booth) { + where.push("booth = ?"); + params.push(booth); + } + const order = status === "pending" ? "received_at ASC" : "reviewed_at DESC"; + const rows = this.#db + .prepare(`SELECT * FROM items WHERE ${where.join(" AND ")} ORDER BY ${order} LIMIT ?`) + .all(...params, limit) as Record[]; + return rows.map((r) => CollectorDb.#map(r)); + } + + review(id: string, label: ReviewVerdict, reviewer: string): ItemRow | null { + this.#db + .prepare("UPDATE items SET review_label = ?, reviewed_at = ?, reviewer = ? WHERE id = ?") + .run(label, new Date().toISOString(), reviewer, id); + return this.get(id); + } + + /** Per booth: received / pending / reviewed. Per operator (booth + hash): how often the + * reviewer's class fell inside the operator's chosen category (agree) or outside + * (disagree) — the honest-mistake / fraud rate the outbox exists for. */ + stats(): { + booths: { booth: string; received: number; pending: number; reviewed: number }[]; + operators: { booth: string; operatorRef: string; reviewed: number; agree: number; disagree: number; unusable: number }[]; + } { + const booths = this.#db + .prepare( + `SELECT booth, COUNT(*) AS received, + SUM(CASE WHEN reviewed_at IS NULL THEN 1 ELSE 0 END) AS pending, + SUM(CASE WHEN reviewed_at IS NOT NULL THEN 1 ELSE 0 END) AS reviewed + FROM items GROUP BY booth ORDER BY booth`, + ) + .all() as { booth: string; received: number; pending: number; reviewed: number }[]; + const reviewed = this.#db + .prepare("SELECT booth, operator_ref, operator_classes, review_label FROM items WHERE reviewed_at IS NOT NULL") + .all() as { booth: string; operator_ref: string; operator_classes: string; review_label: string }[]; + const ops = new Map(); + for (const r of reviewed) { + const key = `${r.booth} ${r.operator_ref}`; + let o = ops.get(key); + if (!o) ops.set(key, (o = { booth: r.booth, operatorRef: r.operator_ref, reviewed: 0, agree: 0, disagree: 0, unusable: 0 })); + o.reviewed += 1; + if (r.review_label === "unusable") o.unusable += 1; + else if ((JSON.parse(r.operator_classes) as string[]).includes(r.review_label)) o.agree += 1; + else o.disagree += 1; + } + return { booths, operators: [...ops.values()].sort((a, b) => b.disagree - a.disagree) }; + } + + /** Reviewed, usable rows — the training set. */ + labelled(): ItemRow[] { + const rows = this.#db + .prepare("SELECT * FROM items WHERE reviewed_at IS NOT NULL AND review_label != 'unusable' ORDER BY reviewed_at") + .all() as Record[]; + return rows.map((r) => CollectorDb.#map(r)); + } +} diff --git a/apps/collector/src/index.ts b/apps/collector/src/index.ts new file mode 100644 index 0000000..512a57d --- /dev/null +++ b/apps/collector/src/index.ts @@ -0,0 +1,16 @@ +import { buildCollector } from "./app.js"; +import { configFromEnv } from "./config.js"; + +const cfg = configFromEnv(); +const app = await buildCollector(cfg); +if (cfg.boothTokens.size === 0) app.log.warn("COLLECTOR_BOOTH_TOKENS is empty — no booth can ingest"); +if (!cfg.reviewer) app.log.warn("COLLECTOR_REVIEWER_USER/PASS not set — the review screen and export refuse"); +app.log.info(`collector: ${cfg.boothTokens.size} booth token(s), data in ${cfg.dataDir}`); +await app.listen({ host: cfg.host, port: cfg.port }); + +const stop = async () => { + await app.close(); + process.exit(0); +}; +process.on("SIGTERM", () => void stop()); +process.on("SIGINT", () => void stop()); diff --git a/apps/collector/src/review-page.ts b/apps/collector/src/review-page.ts new file mode 100644 index 0000000..0e42479 --- /dev/null +++ b/apps/collector/src/review-page.ts @@ -0,0 +1,117 @@ +import { VEHICLE_CLASSES } from "@parking/shared"; + +// The reviewer's screen: one pending crop at a time, the operator's pick and the camera's +// pick beside it, one button per vocabulary class + "unusable". Served by the collector +// itself (no build step, no framework) — this is deliberately the whole UI. + +export function reviewPage(): string { + const classes = JSON.stringify(VEHICLE_CLASSES); + return ` + + + + +Wash review + + + +
Wash review
+
+
+

Loading…

+
+
+
boothoperatorreviewedagreedisagreeunusable
+
+

Keys: 1–9, 0 pick a class in order · u unusable · s skip. Skipped items come back after a reload. Your verdict is the training label; the operator's pick is only compared against it.

+
+ + +`; +} diff --git a/apps/collector/tsconfig.json b/apps/collector/tsconfig.json new file mode 100644 index 0000000..7962103 --- /dev/null +++ b/apps/collector/tsconfig.json @@ -0,0 +1,10 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "./src", + "outDir": "./dist" + }, + "references": [{ "path": "../../packages/shared" }], + "include": ["src/**/*"], + "exclude": ["src/**/*.test.ts"] +} diff --git a/apps/collector/vitest.config.ts b/apps/collector/vitest.config.ts new file mode 100644 index 0000000..9578e1a --- /dev/null +++ b/apps/collector/vitest.config.ts @@ -0,0 +1,5 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { include: ["src/**/*.test.ts"], env: { LOG_LEVEL: "silent" } }, +}); diff --git a/apps/server/Dockerfile b/apps/server/Dockerfile index c4c9869..c9d85e2 100644 --- a/apps/server/Dockerfile +++ b/apps/server/Dockerfile @@ -15,6 +15,7 @@ COPY package.json pnpm-lock.yaml pnpm-workspace.yaml turbo.json ./ COPY apps/server/package.json apps/server/ COPY apps/web/package.json apps/web/ COPY apps/vision/package.json apps/vision/ +COPY apps/collector/package.json apps/collector/ COPY packages/db/package.json packages/db/ COPY packages/devices/package.json packages/devices/ COPY packages/shared/package.json packages/shared/ diff --git a/apps/server/src/modules/carwash/review-outbox.test.ts b/apps/server/src/modules/carwash/review-outbox.test.ts index ff4d649..cf349a5 100644 --- a/apps/server/src/modules/carwash/review-outbox.test.ts +++ b/apps/server/src/modules/carwash/review-outbox.test.ts @@ -83,7 +83,7 @@ describe("queue + drain", () => { const cfg = { url: "https://collector.overlay/ingest", token: "secret-1", boothId: "booth-7", intervalSec: 60 }; const read = { bodyType: "car" as const, confidence: 0.86, snapshotId: "snap-1", box: CAR, plateBox: PLATE }; - const item = { orderId: "o-1", createdAt: "2026-09-06T10:00:00.000Z", createdBy: "lavazhier", categoryId: "car", categoryName: "Vetura", serviceName: "Standard", visionCategoryId: "car", downgraded: false }; + const item = { orderId: "o-1", createdAt: "2026-09-06T10:00:00.000Z", createdBy: "lavazhier", categoryId: "car", categoryName: "Vetura", categoryClasses: ["car", "sedan"], serviceName: "Standard", visionCategoryId: "car", downgraded: false }; async function seed(): Promise { db.insert(snapshots).values({ id: "snap-1", direction: "entry", identity: "T-1", contentType: "image/jpeg", bytes: await frame(), capturedAt: new Date().toISOString() }).run(); @@ -105,7 +105,7 @@ describe("queue + drain", () => { const row = db.select().from(carwashReviewOutbox).all()[0]!; expect(row.status).toBe("queued"); expect(row.image!.length).toBeGreaterThan(500); - expect(row.payload).toMatchObject({ v: 1, booth: "booth-7", order: "o-1", operatorCategory: { id: "car", name: "Vetura" }, vision: { class: "car", confidence: 0.86 }, downgraded: false, image: { plateBlurred: true } }); + expect(row.payload).toMatchObject({ v: 1, booth: "booth-7", order: "o-1", operatorCategory: { id: "car", name: "Vetura", classes: ["car", "sedan"] }, vision: { class: "car", confidence: 0.86 }, downgraded: false, image: { plateBlurred: true } }); expect(JSON.stringify(row.payload)).not.toContain("lavazhier"); expect(await ob.drain()).toEqual({ sent: 1, failed: 0, deferred: 0 }); diff --git a/apps/server/src/modules/carwash/review-outbox.ts b/apps/server/src/modules/carwash/review-outbox.ts index 510031a..e4ce895 100644 --- a/apps/server/src/modules/carwash/review-outbox.ts +++ b/apps/server/src/modules/carwash/review-outbox.ts @@ -66,6 +66,9 @@ export interface ReviewItemInput { readonly createdBy: string; readonly categoryId: string; readonly categoryName: string; + /** The vision classes the chosen category covers at this site (its mapping) — lets the + * reviewer's class be judged against the operator's category without the site's setup. */ + readonly categoryClasses: readonly string[]; readonly serviceName: string; readonly visionCategoryId: string | null; readonly downgraded: boolean; @@ -185,7 +188,7 @@ export class ReviewOutbox { order: item.orderId, at: item.createdAt, operator: operatorRef(this.#cfg.boothId, item.createdBy), - operatorCategory: { id: item.categoryId, name: item.categoryName }, + operatorCategory: { id: item.categoryId, name: item.categoryName, classes: [...item.categoryClasses] }, service: item.serviceName, vision: { class: read.bodyType, confidence: read.confidence, categoryId: item.visionCategoryId }, downgraded: item.downgraded, diff --git a/apps/server/src/modules/carwash/service.ts b/apps/server/src/modules/carwash/service.ts index e8b4b03..49ed412 100644 --- a/apps/server/src/modules/carwash/service.ts +++ b/apps/server/src/modules/carwash/service.ts @@ -535,6 +535,7 @@ export class CarwashService { createdBy: input.actor, categoryId: category.id, categoryName: category.name, + categoryClasses: category.visionClasses, serviceName: service.name, visionCategoryId: visionCategory?.id ?? null, downgraded: downgradeEventId != null, diff --git a/docker-compose.collector.yml b/docker-compose.collector.yml new file mode 100644 index 0000000..1d8f1cf --- /dev/null +++ b/docker-compose.collector.yml @@ -0,0 +1,47 @@ +# The Car Wash REVIEW COLLECTOR — deployed on the REVIEWER's host (art-docker-station), NOT +# on a booth. Its own Komodo stack ("wash-collector" in komodo/resources.toml) points at this +# file alone, so nothing here reaches a booth and nothing of the booth stack reaches this +# host. See wiki/concepts/vision-review-outbox.md. +# +# Reachability: booths POST to /ingest over the Netbird overlay only. Bind the published +# port to the host's OVERLAY address (COLLECTOR_BIND), never 0.0.0.0 on a host that also +# has a public interface. The Netbird policy should allow booths → this host:8090 and +# nothing else on it. + +services: + collector: + image: ${REGISTRY:-git.infra.msai.al/mca/parking_solution}/parking-collector:${TAG:-dev} + restart: unless-stopped + ports: + - "${COLLECTOR_BIND:-127.0.0.1}:8090:8090" + environment: + # ":" pairs — one per booth, the booth's CARWASH_REVIEW_TOKEN under its + # pseudonymous CARWASH_REVIEW_BOOTH_ID. A Komodo secret reference in the stack env. + COLLECTOR_BOOTH_TOKENS: ${COLLECTOR_BOOTH_TOKENS:?set COLLECTOR_BOOTH_TOKENS in the stack env} + # The single reviewer login (HTTP Basic over the overlay). + COLLECTOR_REVIEWER_USER: ${COLLECTOR_REVIEWER_USER:-reviewer} + COLLECTOR_REVIEWER_PASS: ${COLLECTOR_REVIEWER_PASS:?set COLLECTOR_REVIEWER_PASS in the stack env} + LOG_LEVEL: ${LOG_LEVEL:-info} + volumes: + - collector-data:/data + + # Phase B trainer — a one-off job on this host's GPU, NOT a service (profile "train": it + # only runs when asked: `docker compose --profile train run --rm trainer`). Reads the + # collector's export + crops straight off the same volume; writes the ONNX classifier the + # vision image then bakes in. The image/script are the next increment; this is the seam. + # trainer: + # image: ${REGISTRY:-git.infra.msai.al/mca/parking_solution}/parking-trainer:${TAG:-dev} + # profiles: ["train"] + # deploy: + # resources: + # reservations: + # devices: + # - driver: nvidia + # count: all + # capabilities: [gpu] + # volumes: + # - collector-data:/data:ro + # - ./models:/out + +volumes: + collector-data: diff --git a/komodo/resources.toml b/komodo/resources.toml index 3e857db..1199605 100644 --- a/komodo/resources.toml +++ b/komodo/resources.toml @@ -91,6 +91,12 @@ COOKIE_SECURE=0 # this set in Setup → Site). Unset = every registered module. See wiki/decisions/venue-modules.md. # park-2 pilots the Car Wash module (2026-09-05). MODULES_ENTITLED=parking,carwash +# Car Wash review outbox (wiki/concepts/vision-review-outbox.md): the collector's ingest URL +# on the Netbird overlay, this booth's pseudonymous id, and its token — the SAME secret the +# wash-collector stack lists under that id. Leave all three unset to keep the outbox off. +#CARWASH_REVIEW_URL=http://docker-station.nb.infra:8090/ingest +#CARWASH_REVIEW_BOOTH_ID=booth-2 +#CARWASH_REVIEW_TOKEN=[[wash_review_token_booth_2]] VISION_ENABLED=1 # Desktop app WS handshake: Origin is tauri://localhost (set explicitly by # platform-ws.ts, since the native WS plugin has no page context to auto-attach @@ -100,3 +106,40 @@ JWT_SECRET=[[park_2_jwt_secret]] EVENT_SIGNING_KEY=[[park_2_event_signing_key]] BACKUP_KEY=[[park_2_backup_key]] """ + +############################################################################## +# Stack — the Car Wash REVIEW COLLECTOR on the reviewer's host (art-docker-station), +# NOT a booth. Same repo/branch/TAG promotion as the booths, but its file_paths name +# ONLY docker-compose.collector.yml, so nothing booth-side lands here and nothing here +# lands on a booth. Booths reach it over the Netbird overlay only (bind to the overlay +# address). Its own secrets. See wiki/concepts/vision-review-outbox.md. +############################################################################## + +[[stack]] +name = "wash-collector" +[stack.config] +server = "art-docker-station" +git_provider = "git.infra.msai.al" +git_account = "komodo" +repo = "mca/parking_solution" +branch = "stage" +file_paths = [ + "docker-compose.collector.yml" +] +registry_provider = "git.infra.msai.al" +registry_account = "komodo" +environment = """ +REGISTRY=git.infra.msai.al/mca/parking_solution +# Pinned like the booths: bump to the stage- that carries the collector. +TAG=stage-REPLACE +# The host's NETBIRD address (an IP: Docker port bindings take no hostname) — the ingest port +# is published on the overlay only. Booths reach it by its Netbird DNS name. +COLLECTOR_BIND=100.75.184.156 +# ":" pairs, one per booth. ONE secret per booth, referenced here AND in +# that booth's own stack as its CARWASH_REVIEW_TOKEN — one value, two consumers, nothing +# to keep in sync, and rotating a booth touches one secret. The booth id is the booth's +# pseudonymous CARWASH_REVIEW_BOOTH_ID, never a site name. Add a pair per booth. +COLLECTOR_BOOTH_TOKENS=booth-2:[[wash_review_token_booth_2]] +COLLECTOR_REVIEWER_USER=reviewer +COLLECTOR_REVIEWER_PASS=[[wash_collector_reviewer_pass]] +""" diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 745e270..f1d1e0e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -18,6 +18,37 @@ importers: specifier: 6.0.3 version: 6.0.3 + apps/collector: + dependencies: + '@fastify/multipart': + specifier: ^9.2.1 + version: 9.4.0 + '@parking/shared': + specifier: workspace:* + version: link:../../packages/shared + better-sqlite3: + specifier: 12.10.1 + version: 12.10.1 + fastify: + specifier: 5.8.5 + version: 5.8.5 + devDependencies: + '@types/better-sqlite3': + specifier: 7.6.13 + version: 7.6.13 + '@types/node': + specifier: 25.9.3 + version: 25.9.3 + tsx: + specifier: 4.22.4 + version: 4.22.4 + typescript: + specifier: 6.0.3 + version: 6.0.3 + vitest: + specifier: ^4.1.9 + version: 4.1.9(@types/node@25.9.3)(jsdom@25.0.1)(vite@8.0.16(@types/node@25.9.3)(esbuild@0.28.1)(jiti@2.7.0)(tsx@4.22.4)) + apps/desktop: dependencies: '@tauri-apps/plugin-process': @@ -744,12 +775,18 @@ packages: '@fastify/ajv-compiler@4.0.5': resolution: {integrity: sha512-KoWKW+MhvfTRWL4qrhUwAAZoaChluo0m0vbiJlGMt2GXvL4LVPQEjt8kSpHI3IBq5Rez8fg+XeH3cneztq+C7A==} + '@fastify/busboy@3.2.2': + resolution: {integrity: sha512-yXSS27qPExaXeuLvMRMXOLtpipzfQYNjG3FkunDWKGfMYjKuhFXko9CVzqxm8jcF+lmtS9Fd89QNdh9XDjnbNg==} + '@fastify/cookie@11.0.2': resolution: {integrity: sha512-GWdwdGlgJxyvNv+QcKiGNevSspMQXncjMZ1J8IvuDQk0jvkzgWWZFNC2En3s+nHndZBGV8IbLwOI/sxCZw/mzA==} '@fastify/cors@11.2.0': resolution: {integrity: sha512-LbLHBuSAdGdSFZYTLVA3+Ch2t+sA6nq3Ejc6XLAKiQ6ViS2qFnvicpj0htsx03FyYeLs04HfRNBsz/a8SvbcUw==} + '@fastify/deepmerge@3.2.1': + resolution: {integrity: sha512-N5Oqvltoa2r9z1tbx4xjky0oRR60v+T47Ic4J1ukoVQcptLOrIdRnCSdTGmOmajZuHVKlTnfcmrjyqsGEW1ztA==} + '@fastify/error@4.2.0': resolution: {integrity: sha512-RSo3sVDXfHskiBZKBPRgnQTtIqpi/7zhJOEmAxCiBcM7d0uwdGdxLlsCaLzGs8v8NnxIRlfG0N51p5yFaOentQ==} @@ -765,6 +802,9 @@ packages: '@fastify/merge-json-schemas@0.2.1': resolution: {integrity: sha512-OA3KGBCy6KtIvLf8DINC5880o5iBlDX4SxzLQS8HorJAbqluzLRn80UXU0bxZn7UOFhFgpRJDasfwn9nG4FG4A==} + '@fastify/multipart@9.4.0': + resolution: {integrity: sha512-Z404bzZeLSXTBmp/trCBuoVFX28pM7rhv849Q5TsbTFZHuk1lc4QjQITTPK92DKVpXmNtJXeHSSc7GYvqFpxAQ==} + '@fastify/proxy-addr@5.1.0': resolution: {integrity: sha512-INS+6gh91cLUjB+PVHfu1UqcB76Sqtpyp7bnL+FYojhjygvOPA9ctiD/JDKsyD9Xgu4hUhCSJBPig/w7duNajw==} @@ -3380,6 +3420,8 @@ snapshots: ajv-formats: 3.0.1(ajv@8.20.0) fast-uri: 3.1.2 + '@fastify/busboy@3.2.2': {} + '@fastify/cookie@11.0.2': dependencies: cookie: 1.1.1 @@ -3390,6 +3432,8 @@ snapshots: fastify-plugin: 5.1.0 toad-cache: 3.7.1 + '@fastify/deepmerge@3.2.1': {} + '@fastify/error@4.2.0': {} '@fastify/fast-json-stringify-compiler@5.0.3': @@ -3410,6 +3454,14 @@ snapshots: dependencies: dequal: 2.0.3 + '@fastify/multipart@9.4.0': + dependencies: + '@fastify/busboy': 3.2.2 + '@fastify/deepmerge': 3.2.1 + '@fastify/error': 4.2.0 + fastify-plugin: 5.1.0 + secure-json-parse: 4.1.0 + '@fastify/proxy-addr@5.1.0': dependencies: '@fastify/forwarded': 3.0.1 diff --git a/wiki/concepts/vision-review-outbox.md b/wiki/concepts/vision-review-outbox.md index 77159d4..6a02e61 100644 --- a/wiki/concepts/vision-review-outbox.md +++ b/wiki/concepts/vision-review-outbox.md @@ -60,10 +60,39 @@ Setup → Car wash show queued / delivered / abandoned + the last error. is off and **nothing is queued** (an unbounded queue nobody drains is worse than none). Set per booth in the Komodo stack env; compose forwards them. -## Not built yet: the collector +## The collector — skeleton built 2026-09-06 (`apps/collector`) -A deliberately small service on the overlay: one ingest endpoint (token per booth, size cap), -one review screen (crop, the operator's pick, the camera's pick → the reviewer picks the truth), -one export (crops + reviewer labels, nothing else) for phase B training. Keep it that small — -it must not grow into a fleet console. The Netbird policy: booths may reach the collector's -ingest port and nothing else on it. +A deliberately small Fastify + SQLite service **in this monorepo** (so it imports the payload +contract and the class vocabulary from `@parking/shared` — the two ends cannot drift), delivered +to the reviewer's host by **its own Komodo stack** (`wash-collector` in `komodo/resources.toml` +→ `docker-compose.collector.yml` only; the booth stacks never see it and it never sees booth +services). Image `parking-collector:-` from the same workflow as the others. +Three surfaces, nothing else — it must not grow into a fleet console: + +- **`POST /ingest`** — bearer token **per booth** (`COLLECTOR_BOOTH_TOKENS`, `boothId:token` + pairs; constant-time compare), `X-Booth-Id` must match the token's booth, multipart `meta` + + `image` (JPEG magic checked, 2 MB cap), `meta` validated field by field against the contract + above (unknown vision class, non-id item, wrong booth → 422), **idempotent on the item id** + (a retry after a lost 2xx → 200 `duplicate`). Stored: `crops//.jpg` on the + volume + one `items` row. The booth now also sends `operatorCategory.classes` (the classes + the chosen category covers at that site) so a reviewer's CLASS can be judged against the + operator's CATEGORY without the site's setup. +- **`/review`** (+ `/api/items`, `/api/items/:id/image`, `/api/items/:id/review`, `/api/stats`) + — the reviewer's screen, served by the process itself (no build, no framework): one pending + crop at a time, the operator's pick and the camera's pick beside it, one button (and one + key) per vocabulary class + *unusable* + *skip*. HTTP Basic, one login + (`COLLECTOR_REVIEWER_USER/PASS`), over the overlay. Stats: per booth received / pending / + reviewed; per operator (booth + hash) **agree / disagree / unusable** — disagree = the + reviewer's class is outside the operator's chosen category. That column is the honest-mistake + / fraud rate. +- **`GET /export/labels.csv`** — reviewed, usable rows: item, booth, crop path, the reviewer's + label, the operator's category + classes, the camera's class + confidence, downgraded, at. + Crops are not packaged: the phase-B trainer runs **on the same host** (its GPU) and reads them + off the volume — `docker-compose.collector.yml` carries the `trainer` seam as a commented + `profiles: [train]` one-off job (next increment). + +**Deploy notes.** Bind the published port to the host's **Netbird address** (`COLLECTOR_BIND`), +never `0.0.0.0` on a host with a public interface; Netbird policy: booths → this host:8090 and +nothing else. The host must be onboarded as a Komodo server like the booths. `TAG` is pinned +and promoted with the booths (one sha for all stacks) — fine while the collector stays small; +its own repo the day it needs its own cadence. diff --git a/wiki/decisions/fleet-deployment-komodo.md b/wiki/decisions/fleet-deployment-komodo.md index 8192eb5..a61f24e 100644 --- a/wiki/decisions/fleet-deployment-komodo.md +++ b/wiki/decisions/fleet-deployment-komodo.md @@ -151,6 +151,14 @@ here so it isn't re-litigated. `komodo/resources.toml` + README) so the control plane is itself reviewable + version-controlled. +## A non-booth stack (2026-09-06) + +The Stack model turned out to fit a service that is *not* a booth: the Car Wash review +collector ([[vision-review-outbox]]) runs on the reviewer's GPU host as its own `[[stack]]` +(`wash-collector`, `server = "art-docker-station"`, `file_paths = ["docker-compose.collector.yml"]`). +Same repo, branch and pinned `TAG` promotion, its own secret references, and — because a stack +names its compose files — nothing booth-side lands on that host and nothing of it on a booth. + ## Open / not yet done - **Per-booth secret generation + rotation flow** — how a new site's unique `EVENT_SIGNING_KEY` diff --git a/wiki/log.md b/wiki/log.md index a71e646..422fed8 100644 --- a/wiki/log.md +++ b/wiki/log.md @@ -3110,3 +3110,14 @@ plate blurred in place, pseudonymous booth id + keyed operator hash, multipart P per-booth bearer, backoff, permanent rejections, void/expiry abandon, image dropped once sent); enqueue off the intake path in `createOrder`; `/api/carwash/review/status` + a Setup line; env + compose. New concept page [[vision-review-outbox]]; [[venue-modules]] As built; index. + +## [2026-09-06] ingest | Review collector skeleton — apps/collector + its own Komodo stack +Built `apps/collector` (Fastify + SQLite, shares the contract via @parking/shared): `POST /ingest` +(bearer per booth, X-Booth-Id must match, multipart meta+JPEG validated, idempotent on item id, +crop on the volume), the reviewer's screen served by the process (Basic auth; one button/key per +class + unusable + skip; per-operator agree/disagree/unusable stats), `GET /export/labels.csv`. +Delivery: `apps/collector/Dockerfile` (monorepo context), `docker-compose.collector.yml` (bind to +the overlay address; commented `trainer` profile seam for the GPU), a third build step in +build-images.yml, and a `wash-collector` stack on `art-docker-station` in komodo/resources.toml +(secret refs to fill). Booth payload now carries `operatorCategory.classes`. Tests: app.test.ts. +Updated [[vision-review-outbox]], [[fleet-deployment-komodo]].