feat(collector): review collector skeleton — apps/collector, its own Komodo stack on the reviewer's host

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/<booth>/<item>.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

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<string, unknown>;
+  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<string, unknown> | 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<string, unknown> | 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<string, unknown> | 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<CollectorApp> {
+  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<void> {
+    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<ItemRow, "imagePath"> {
+  const { imagePath: _p, ...rest } = r;
+  return rest;
+}