parking_solution/apps/server/src/modules/carwash/review-outbox.ts

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;
  /** Queue one in N ENTRY vehicle reads (no order attached) for the reviewer — the gate
   *  view is exactly what the classifier is trained on, and the entry stream is many times
   *  the wash stream. 0 = off. */
  readonly entrySample: 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);
  const sample = Number(env.CARWASH_REVIEW_ENTRY_SAMPLE ?? 0);
  return {
    url, token, boothId,
    intervalSec: Number.isFinite(raw) && raw >= 10 ? raw : 60,
    entrySample: Number.isInteger(sample) && sample > 0 ? sample : 0,
  };
}

/** 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;
  /** 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;
}

/**
 * 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;
  /** 0 = entry sampling off; N = one in N entry reads is queued. */
  readonly entrySample: number;
}

export class ReviewOutbox {
  readonly #db: Db;
  readonly #logger: FastifyBaseLogger;
  readonly #cfg: ReviewUploadConfig | null;
  readonly #fetch: FetchLike;
  #timer: NodeJS.Timeout | null = null;
  #draining = false;
  #entrySeen = 0;

  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;
    return this.#queue(item.orderId, read, (id, crop) => ({
      v: 1,
      kind: "wash",
      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, classes: [...item.categoryClasses] },
      service: item.serviceName,
      vision: { class: read.bodyType, confidence: read.confidence, categoryId: item.visionCategoryId },
      downgraded: item.downgraded,
      image: crop,
    }));
  }

  /** Every Nth entry read is a sample (N = entrySample); the caller queues it. Counted
   *  in-process, so "1 in 5" is exactly that across a booth's day. */
  sampleEntry(): boolean {
    const n = this.#cfg?.entrySample ?? 0;
    if (n <= 0) return false;
    this.#entrySeen += 1;
    return this.#entrySeen % n === 0;
  }

  /** Queue an ENTRY sample: the crop and the camera's class only — no order, no operator,
   *  no category. Pure training material in the gate view; the reviewer labels it. */
  async enqueueEntry(read: VehicleRead): Promise<boolean> {
    if (!this.#cfg) return false;
    return this.#queue(`entry:${read.snapshotId ?? "?"}`, read, (id, crop) => ({
      v: 1,
      kind: "entry",
      booth: this.#cfg!.boothId,
      item: id,
      at: new Date().toISOString(),
      vision: { class: read.bodyType, confidence: read.confidence },
      image: crop,
    }));
  }

  async #queue(
    ref: string,
    read: VehicleRead,
    build: (id: string, image: { width: number; height: number; plateBlurred: boolean }) => Record<string, unknown>,
  ): Promise<boolean> {
    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) — ${ref} not queued`);
        return false;
      }
      const crop = await makeReviewCrop(snap.bytes, read.box, read.plateBox);
      const id = randomUUID();
      const payload = build(id, { width: crop.width, height: crop.height, plateBlurred: crop.plateBlurred });
      this.#db
        .insert(carwashReviewOutbox)
        .values({ id, orderId: ref, 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 ${ref}: ${(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,
      entrySample: this.#cfg?.entrySample ?? 0,
      queued: count("queued"),
      sent: count("sent"),
      failed: count("failed"),
      lastSentAt: lastSent,
      lastError: lastErr,
    };
  }
}