feat(carwash): review outbox, booth side — plate-blurred vehicle crop + the operator's choice, queued for a trusted remote reviewer

The operator's category choice is a hypothesis, not truth (user, 2026-09-06): each wash order with a vehicle read queues a package for a trusted reviewer over the private overlay (Netbird); the verdict becomes the phase-B training label and the per-operator error rate. wiki/concepts/vision-review-outbox.md. - Boxes: the vision service returns the vehicle bbox; snapshot.ts stores the vehicle and plate boxes on the read as FRACTIONS of the analysed frame (the stored snapshot is a downscaled copy); vehicleForIdentity() returns them. - carwash_review_outbox (migration 0031) + review-outbox.ts: crop = detector box + 8 % margin, ≤ 640 px, plate blurred in place from the plate box; payload carries a pseudonymous booth id and a keyed operator hash — no site name, no plate, no OSD, no bystanders; multipart POST with a per-booth bearer; 2xx → sent (image dropped); 400/404/413/415/422 → abandoned; anything else → backoff 1 min·2^n capped 6 h; voided orders and items older than 14 days abandoned unsent. Nothing queued while unconfigured. - Enqueue is fire-and-forget off the intake path in createOrder; the loop runs every CARWASH_REVIEW_INTERVAL_SEC (60) and stops on close. - GET /api/carwash/review/status (site:read) + a "Remote review" line in Setup → Car wash. - Env CARWASH_REVIEW_URL / _TOKEN / _BOOTH_ID (all three or off) documented in .env.example and forwarded by compose. - Tests: review-outbox.test.ts (crop + blur on a synthetic frame, config/pseudonyms, queue/drain/backoff/abandon, through the app). Wiki: new concept page, index, venue-modules As built, log. The collector is not built. Claude-Session: https://claude.ai/code/session_01FWncR69HgGPuei1dLrW3cU
2026-09-06 22:33:43 +02:00
parent 78ca58d264
commit e67f0ccef0
25 changed files with 834 additions and 13 deletions
@@ -0,0 +1,327 @@
+import { createHash, randomUUID } from "node:crypto";
+import sharp from "sharp";
+import { and, asc, carwashOrders, carwashReviewOutbox, eq, isNull, lte, or, snapshots, sql, type Db } from "@parking/db";
+import type { NormBox, VehicleRead } from "@parking/shared";
+import type { FastifyBaseLogger } from "fastify";
+
+// The Car Wash REVIEW OUTBOX — booth side (wiki/concepts/vision-review-outbox.md).
+//
+// The operator's category choice at intake is a HYPOTHESIS, not truth (the threat model:
+// the operator may err or cheat). So every wash order that has a vehicle read queues a
+// small package for a trusted remote reviewer: the vehicle CROP cut out of the entry
+// snapshot with the plate BLURRED, the operator's choice, and what the camera thought.
+// The reviewer's verdict becomes the training label for the body-type classifier (phase
+// B) and, per operator, the honest-mistake / fraud rate.
+//
+// Rules that shape this file:
+//   - OFFLINE-FIRST: the wash never waits. Enqueue is fire-and-forget off the intake path;
+//     a background loop drains the queue when the private overlay (Netbird) is up, with
+//     backoff, and gives up loudly after EXPIRE_DAYS.
+//   - ONE-WAY: the booth POSTs; nothing ever comes back into the booth's decisions. The
+//     signed ledger stays the only record of what happened at the wash.
+//   - NOTHING THAT NAMES THE SITE LEAVES: only the crop (no walls, no camera OSD, no
+//     bystanders), the plate blurred in place, a per-booth pseudonymous id set at deploy,
+//     the operator as a keyed hash. The mapping back to people and places stays with the
+//     reviewer, off the collector.
+//   - THE NETWORK IS NOT THE AUTH: a per-booth bearer token on top of the overlay; the
+//     booth can do nothing at the collector but this one POST.
+
+export interface ReviewUploadConfig {
+  /** The collector's ingest URL (reachable only over the overlay). */
+  readonly url: string;
+  /** Per-booth bearer token. */
+  readonly token: string;
+  /** Pseudonymous booth id — a label the reviewer maps to a site; never the site name. */
+  readonly boothId: string;
+  readonly intervalSec: number;
+}
+
+/** From the server env (Komodo stack env). All three of URL, token and booth id, or off. */
+export function reviewUploadConfigFromEnv(env: NodeJS.ProcessEnv = process.env): ReviewUploadConfig | null {
+  const url = (env.CARWASH_REVIEW_URL ?? "").trim();
+  const token = (env.CARWASH_REVIEW_TOKEN ?? "").trim();
+  const boothId = (env.CARWASH_REVIEW_BOOTH_ID ?? "").trim();
+  if (!url || !token || !boothId) return null;
+  const raw = Number(env.CARWASH_REVIEW_INTERVAL_SEC ?? 60);
+  return { url, token, boothId, intervalSec: Number.isFinite(raw) && raw >= 10 ? raw : 60 };
+}
+
+/** The crop's longest edge, in pixels — enough for a reviewer and a classifier, small
+ *  enough that a day of washes is a few megabytes. */
+export const CROP_MAX_EDGE = 640;
+/** Margin around the detector's box, as a fraction of the box (context for the reviewer). */
+const CROP_MARGIN = 0.08;
+/** Items older than this are abandoned (failed "expired") — a booth cut off for two weeks
+ *  should not resurface a fortnight of crops in one burst. */
+export const EXPIRE_DAYS = 14;
+/** Backoff: 1 min · 2^attempts, capped. */
+const BACKOFF_BASE_MS = 60_000;
+const BACKOFF_CAP_MS = 6 * 60 * 60 * 1000;
+const UPLOAD_TIMEOUT_MS = 20_000;
+
+/** What one order contributes to the package (the service hands this over at intake). */
+export interface ReviewItemInput {
+  readonly orderId: string;
+  readonly createdAt: string;
+  readonly createdBy: string;
+  readonly categoryId: string;
+  readonly categoryName: string;
+  readonly serviceName: string;
+  readonly visionCategoryId: string | null;
+  readonly downgraded: boolean;
+}
+
+/**
+ * Cut the vehicle out of the snapshot and blur the plate inside it. Boxes are fractions
+ * of the frame, so this works on the stored (downscaled) copy. Returns a JPEG.
+ */
+export async function makeReviewCrop(
+  snapshotBytes: Buffer,
+  box: NormBox,
+  plateBox: NormBox | null | undefined,
+): Promise<{ bytes: Buffer; width: number; height: number; plateBlurred: boolean }> {
+  const img = sharp(snapshotBytes, { failOn: "none" }).rotate();
+  const meta = await img.metadata();
+  const W = meta.width ?? 0;
+  const H = meta.height ?? 0;
+  if (!W || !H) throw new Error("snapshot has no dimensions");
+  const px = (b: NormBox) => ({
+    left: Math.round(b.x1 * W), top: Math.round(b.y1 * H),
+    right: Math.round(b.x2 * W), bottom: Math.round(b.y2 * H),
+  });
+  const v = px(box);
+  const mw = Math.round((v.right - v.left) * CROP_MARGIN);
+  const mh = Math.round((v.bottom - v.top) * CROP_MARGIN);
+  const left = Math.max(0, v.left - mw);
+  const top = Math.max(0, v.top - mh);
+  const right = Math.min(W, v.right + mw);
+  const bottom = Math.min(H, v.bottom + mh);
+  const width = right - left;
+  const height = bottom - top;
+  if (width < 8 || height < 8) throw new Error("vehicle box too small to crop");
+
+  let crop = img.clone().extract({ left, top, width, height });
+  let plateBlurred = false;
+  if (plateBox) {
+    // The plate region, in CROP coordinates, padded a little so the blur eats the edges.
+    const p = px(plateBox);
+    const pad = Math.round(Math.max(p.right - p.left, p.bottom - p.top) * 0.25);
+    const pl = Math.max(0, p.left - pad - left);
+    const pt = Math.max(0, p.top - pad - top);
+    const pr = Math.min(width, p.right + pad - left);
+    const pb = Math.min(height, p.bottom + pad - top);
+    if (pr - pl >= 2 && pb - pt >= 2) {
+      const region = await sharp(await crop.clone().toBuffer())
+        .extract({ left: pl, top: pt, width: pr - pl, height: pb - pt })
+        .blur(Math.max(6, Math.round((pr - pl) / 6)))
+        .toBuffer();
+      crop = sharp(await crop.toBuffer()).composite([{ input: region, left: pl, top: pt }]);
+      plateBlurred = true;
+    }
+  }
+  const out = await crop
+    .resize({ width: CROP_MAX_EDGE, height: CROP_MAX_EDGE, fit: "inside", withoutEnlargement: true })
+    .jpeg({ quality: 85, mozjpeg: true })
+    .toBuffer({ resolveWithObject: true });
+  return { bytes: out.data, width: out.info.width, height: out.info.height, plateBlurred };
+}
+
+/** The operator as a keyed hash — stable per booth so the reviewer can count per person,
+ *  meaningless anywhere else. */
+export function operatorRef(boothId: string, username: string): string {
+  return createHash("sha256").update(`${boothId}:${username}`).digest("hex").slice(0, 16);
+}
+
+type FetchLike = (input: string, init: RequestInit) => Promise<Response>;
+
+export interface OutboxStatus {
+  readonly enabled: boolean;
+  readonly boothId: string | null;
+  readonly queued: number;
+  readonly sent: number;
+  readonly failed: number;
+  readonly lastSentAt: string | null;
+  readonly lastError: string | null;
+}
+
+export class ReviewOutbox {
+  readonly #db: Db;
+  readonly #logger: FastifyBaseLogger;
+  readonly #cfg: ReviewUploadConfig | null;
+  readonly #fetch: FetchLike;
+  #timer: NodeJS.Timeout | null = null;
+  #draining = false;
+
+  constructor(db: Db, logger: FastifyBaseLogger, cfg: ReviewUploadConfig | null, fetchFn?: FetchLike) {
+    this.#db = db;
+    this.#logger = logger;
+    this.#cfg = cfg;
+    this.#fetch = fetchFn ?? ((input, init) => fetch(input, init));
+  }
+
+  get enabled(): boolean {
+    return this.#cfg != null;
+  }
+
+  /** Queue one order's package. Fire-and-forget: the caller does NOT await this on the
+   *  intake path; every failure is logged, none is thrown. Skipped when there is no
+   *  vehicle box (nothing to crop — a frame without a detected vehicle is no training
+   *  sample) or when upload is not configured (an unbounded queue nobody drains). */
+  async enqueue(item: ReviewItemInput, read: VehicleRead): Promise<boolean> {
+    if (!this.#cfg) return false;
+    if (!read.box || !read.snapshotId) return false;
+    try {
+      const snap = this.#db.select().from(snapshots).where(eq(snapshots.id, read.snapshotId)).get();
+      if (!snap) {
+        this.#logger.info(`carwash review: snapshot ${read.snapshotId} gone (pruned) — order ${item.orderId} not queued`);
+        return false;
+      }
+      const crop = await makeReviewCrop(snap.bytes, read.box, read.plateBox);
+      const id = randomUUID();
+      const payload = {
+        v: 1,
+        booth: this.#cfg.boothId,
+        item: id,
+        order: item.orderId,
+        at: item.createdAt,
+        operator: operatorRef(this.#cfg.boothId, item.createdBy),
+        operatorCategory: { id: item.categoryId, name: item.categoryName },
+        service: item.serviceName,
+        vision: { class: read.bodyType, confidence: read.confidence, categoryId: item.visionCategoryId },
+        downgraded: item.downgraded,
+        image: { width: crop.width, height: crop.height, plateBlurred: crop.plateBlurred },
+      };
+      this.#db
+        .insert(carwashReviewOutbox)
+        .values({ id, orderId: item.orderId, createdAt: new Date().toISOString(), status: "queued", attempts: 0, nextAttemptAt: null, image: crop.bytes, payload })
+        .run();
+      return true;
+    } catch (err) {
+      this.#logger.warn(`carwash review: could not queue order ${item.orderId}: ${(err as Error).message}`);
+      return false;
+    }
+  }
+
+  start(): void {
+    if (!this.#cfg || this.#timer) return;
+    const tick = () => {
+      void this.drain().catch((err) => this.#logger.warn(`carwash review: drain failed: ${(err as Error).message}`));
+    };
+    this.#timer = setInterval(tick, this.#cfg.intervalSec * 1000);
+    this.#timer.unref?.();
+    setTimeout(tick, 5_000).unref?.();
+  }
+
+  stop(): void {
+    if (this.#timer) clearInterval(this.#timer);
+    this.#timer = null;
+  }
+
+  /** Send what is due, oldest first. Returns the tally; never throws for a single item. */
+  async drain(limit = 20): Promise<{ sent: number; failed: number; deferred: number }> {
+    const tally = { sent: 0, failed: 0, deferred: 0 };
+    if (!this.#cfg || this.#draining) return tally;
+    this.#draining = true;
+    try {
+      const now = new Date().toISOString();
+      const due = this.#db
+        .select()
+        .from(carwashReviewOutbox)
+        .where(and(eq(carwashReviewOutbox.status, "queued"), or(isNull(carwashReviewOutbox.nextAttemptAt), lte(carwashReviewOutbox.nextAttemptAt, now))))
+        .orderBy(asc(carwashReviewOutbox.createdAt))
+        .limit(limit)
+        .all();
+      for (const row of due) {
+        const outcome = await this.#send(row);
+        tally[outcome] += 1;
+      }
+      if (tally.sent || tally.failed) this.#logger.info(`carwash review: sent ${tally.sent}, failed ${tally.failed}, deferred ${tally.deferred}`);
+    } finally {
+      this.#draining = false;
+    }
+    return tally;
+  }
+
+  async #send(row: typeof carwashReviewOutbox.$inferSelect): Promise<"sent" | "failed" | "deferred"> {
+    const cfg = this.#cfg!;
+    const ageMs = Date.now() - Date.parse(row.createdAt);
+    if (ageMs > EXPIRE_DAYS * 24 * 60 * 60 * 1000) return this.#fail(row, `expired after ${EXPIRE_DAYS} days`);
+    // A wash voided before delivery is not a sample (and not a decision to review).
+    const order = this.#db.select({ status: carwashOrders.status }).from(carwashOrders).where(eq(carwashOrders.id, row.orderId)).get();
+    if (order?.status === "void") return this.#fail(row, "order voided");
+    if (!row.image) return this.#fail(row, "image missing");
+
+    const form = new FormData();
+    form.set("meta", JSON.stringify(row.payload));
+    form.set("image", new Blob([new Uint8Array(row.image)], { type: "image/jpeg" }), `${row.id}.jpg`);
+    const ac = new AbortController();
+    const t = setTimeout(() => ac.abort(), UPLOAD_TIMEOUT_MS);
+    try {
+      const res = await this.#fetch(cfg.url, {
+        method: "POST",
+        headers: { authorization: `Bearer ${cfg.token}`, "x-booth-id": cfg.boothId },
+        body: form,
+        signal: ac.signal,
+      });
+      if (res.ok) {
+        this.#db
+          .update(carwashReviewOutbox)
+          .set({ status: "sent", sentAt: new Date().toISOString(), image: null, lastError: null, attempts: row.attempts + 1 })
+          .where(eq(carwashReviewOutbox.id, row.id))
+          .run();
+        return "sent";
+      }
+      // The collector refused the package itself → no retry will help.
+      if ([400, 404, 413, 415, 422].includes(res.status)) return this.#fail(row, `rejected: HTTP ${res.status}`);
+      // Everything else (auth not yet fixed, throttled, collector down) → try again later.
+      return this.#defer(row, `HTTP ${res.status}`);
+    } catch (err) {
+      return this.#defer(row, (err as Error).name === "AbortError" ? "timeout" : (err as Error).message);
+    } finally {
+      clearTimeout(t);
+    }
+  }
+
+  #fail(row: typeof carwashReviewOutbox.$inferSelect, why: string): "failed" {
+    this.#db
+      .update(carwashReviewOutbox)
+      .set({ status: "failed", lastError: why, image: null, attempts: row.attempts + 1 })
+      .where(eq(carwashReviewOutbox.id, row.id))
+      .run();
+    this.#logger.warn(`carwash review: item ${row.id} (order ${row.orderId}) abandoned — ${why}`);
+    return "failed";
+  }
+
+  #defer(row: typeof carwashReviewOutbox.$inferSelect, why: string): "deferred" {
+    const attempts = row.attempts + 1;
+    const wait = Math.min(BACKOFF_BASE_MS * 2 ** Math.min(attempts, 20), BACKOFF_CAP_MS);
+    this.#db
+      .update(carwashReviewOutbox)
+      .set({ attempts, lastError: why, nextAttemptAt: new Date(Date.now() + wait).toISOString() })
+      .where(eq(carwashReviewOutbox.id, row.id))
+      .run();
+    return "deferred";
+  }
+
+  status(): OutboxStatus {
+    const count = (s: "queued" | "sent" | "failed") =>
+      this.#db.select({ n: sql<number>`count(*)` }).from(carwashReviewOutbox).where(eq(carwashReviewOutbox.status, s)).get()?.n ?? 0;
+    const lastSent = this.#db.select({ at: sql<string | null>`max(${carwashReviewOutbox.sentAt})` }).from(carwashReviewOutbox).get()?.at ?? null;
+    const lastErr = this.#db
+      .select({ e: carwashReviewOutbox.lastError })
+      .from(carwashReviewOutbox)
+      .where(sql`${carwashReviewOutbox.lastError} is not null`)
+      .orderBy(sql`coalesce(${carwashReviewOutbox.sentAt}, ${carwashReviewOutbox.nextAttemptAt}, ${carwashReviewOutbox.createdAt}) desc`)
+      .limit(1)
+      .get()?.e ?? null;
+    return {
+      enabled: this.enabled,
+      boothId: this.#cfg?.boothId ?? null,
+      queued: count("queued"),
+      sent: count("sent"),
+      failed: count("failed"),
+      lastSentAt: lastSent,
+      lastError: lastErr,
+    };
+  }
+}