parking_solution/apps/server/src/snapshot.ts

import { randomUUID } from "node:crypto";
import sharp from "sharp";
import { and, eq, gte, sessions, deviceEvents as deviceEventsTable, snapshots, type Db } from "@parking/db";
import { registry, type CameraDevice, type Snapshot } from "@parking/devices";
import { reasonPayload } from "@parking/shared";
import type { FastifyBaseLogger } from "fastify";
import { devicesByDirection, type FlowDirection } from "./device-resolve.js";
import { deviceEvents } from "./device-events.js";
import type { EventLog } from "./event-log.js";
import type { VisionClient } from "./vision-client.js";

// Camera snapshot capture, fired AFTER the barrier opens and never awaited on the
// open path (decision 2026-06-16): a snapshot is EVIDENCE, not a gate. A camera
// failure must never delay or prevent an open — the signed ledger is the decision,
// the image is an independent, prunable record stored as a BLOB in `snapshots`.
// See wiki/concepts/entry-exit-points.md and append-only-event-chain.md.
//
// Every camera serving the firing direction (entry/exit, or both) snapshots. Each
// capture is independent — one camera down doesn't stop the others. A captured image
// → a `snapshots` row + a `kind:"snapshot"` telemetry device_event; a failure → a
// telemetry device_event only. The caller passes the session `identity` so the image
// links to the signed vehicle_entry/exit.
//
// ANPR rides this snapshot (2026-06-19). A transient button-press or a subscriber
// QR/RFID read triggers the entry/exit, which fires THIS snapshot — that is exactly the
// moment to recognize the plate, off the SAME image, tied to the SAME session identity.
// So when a `vision` client is passed AND the camera opts in (config.anpr), each stored
// snapshot is sent to the vision service and the extracted plate is RECORDED against the
// session (a `kind:"read"` device_event with plate/confidence/snapshotId). ADVISORY +
// fire-and-forget: it never blocks the open and never changes the entry/exit decision —
// it's a record ("session X entered on plate AA558EE"). No polling; recognition only
// happens on a real entry/exit. See wiki/entities/opencv-anpr-service.md.
//
// STORAGE RE-ENCODE (2026-06-28). Cameras serve full-res JPEGs (a Hikvision main stream is
// 2688×1520 / ~600 KB); stored raw, snapshots dominated the appliance DB (~72%). Each frame
// is now downscaled (long edge ≤ SNAPSHOT_MAX_EDGE) + re-compressed (q SNAPSHOT_JPEG_QUALITY)
// BEFORE storage — ~6–10× smaller, plate still clearly readable. RECOGNITION runs on the
// ORIGINAL full-res bytes (downscaling hurts OCR); the re-encode is storage-only. Fail-soft:
// a re-encode error stores the original, never drops the snapshot or blocks the open.

/** Long-edge cap (px) + JPEG quality for the STORED snapshot. Env-overridable per appliance. */
const SNAP_MAX_EDGE = Number(process.env.SNAPSHOT_MAX_EDGE ?? 1280);
const SNAP_QUALITY = Number(process.env.SNAPSHOT_JPEG_QUALITY ?? 80);

/** Strip a camera's `; charset=...` cruft from a content type (a JPEG is binary). A bare
 *  `image/jpeg` renders; `image/jpeg; charset="UTF-8"` (what some cameras return, e.g.
 *  Hikvision) is malformed for a binary body and browsers refuse to decode it. Applied
 *  both on capture AND when serving, so legacy rows stored before this normalization
 *  existed still serve a clean type. */
export function cleanType(ct: string | null | undefined): string {
  const base = ct?.split(";")[0]?.trim();
  return base || "image/jpeg";
}

/** Downscale + re-encode a captured frame for STORAGE (evidence, not OCR). Caps the long edge
 *  and re-compresses to JPEG. Fail-soft: any error (e.g. a non-image body) returns the original
 *  bytes with a cleaned content type, so a snapshot is never lost. */
export async function encodeForStorage(
  shot: Snapshot,
  logger: FastifyBaseLogger,
): Promise<{ bytes: Buffer; contentType: string }> {
  try {
    const out = await sharp(shot.bytes, { failOn: "none" })
      .rotate() // honor EXIF orientation before we drop the metadata
      .resize({ width: SNAP_MAX_EDGE, height: SNAP_MAX_EDGE, fit: "inside", withoutEnlargement: true })
      .jpeg({ quality: SNAP_QUALITY, mozjpeg: true })
      .toBuffer();
    return { bytes: out, contentType: "image/jpeg" };
  } catch (err) {
    logger.warn(`snapshot re-encode failed, storing original: ${(err as Error).message}`);
    return { bytes: shot.bytes, contentType: cleanType(shot.contentType) };
  }
}

interface SnapshotJob {
  readonly db: Db;
  readonly direction: FlowDirection;
  /** Session/credential ref (ticket id, plate, subscription car key) — links to the ledger. */
  readonly identity: string;
  readonly logger: FastifyBaseLogger;
  /** Optional vision client — when present, ANPR runs on each captured image from an
   *  `anpr`-enabled camera and records the plate against `identity`. Advisory only. */
  readonly vision?: VisionClient | null;
  /** Optional signed ledger — when present (the transient ENTRY path passes it), a
   *  recognized entry plate that is already OPEN under another recent session signs an
   *  `entry.duplicatePlate` anomaly (same car, second ticket). Post-hoc; never a gate. */
  readonly log?: EventLog | null;
}

/** Camera config flag opting it into snapshot-triggered ANPR. */
interface CameraConfig {
  readonly anpr?: boolean;
  readonly [k: string]: unknown;
}

/**
 * Fire snapshots for the directional camera set. Returns immediately with a promise
 * the caller MAY ignore (fire-and-forget) — it resolves to the captured snapshot ids.
 * The caller must NOT block its open path on this.
 */
export function snapshotAsync(job: SnapshotJob): Promise<string[]> {
  const { db, direction, identity, logger, vision, log } = job;
  const rows = devicesByDirection(db, "camera", direction);
  if (rows.length === 0) return Promise.resolve([]);

  return Promise.all(
    rows.map(async (row): Promise<string | null> => {
      const camera = buildCamera(row);
      if (!camera) {
        recordFailure(db, direction, row.id, identity, "camera config won't build", logger);
        return null;
      }
      try {
        // Shared capture: if the ANPR bridge just pulled this camera's frame for the
        // same vehicle, reuse it instead of a 2nd concurrent GET (which 503s).
        const shot = await captureSnapshotShared(row.id, camera, { direction });
        const id: string = randomUUID();
        // Re-encode for STORAGE only (downscale + recompress). Recognition below still
        // uses the original full-res `shot`.
        const stored = await encodeForStorage(shot, logger);
        db.insert(snapshots)
          .values({
            id,
            direction,
            deviceId: row.id,
            identity,
            contentType: stored.contentType,
            bytes: stored.bytes,
            capturedAt: shot.capturedAt,
          })
          .run();
        // Telemetry breadcrumb pointing at the stored image (NOT the bytes).
        recordEvent(db, direction, row.id, identity, { snapshotId: id, ok: true }, logger);

        // ANPR off the SAME image, tied to the SAME session — when vision is enabled
        // and this camera opts in. Fire-and-forget: never delays the open path.
        if (vision?.enabled && (row.config as CameraConfig)?.anpr === true) {
          void recognizePlate(db, vision, row.id, direction, identity, id, shot, logger, log);
        }
        return id;
      } catch (err) {
        recordFailure(db, direction, row.id, identity, (err as Error).message, logger);
        return null;
      }
    }),
  ).then((ids) => ids.filter((id): id is string => id != null));
}

/**
 * Recognize the plate off a captured entry/exit image and RECORD it against the session
 * `identity` — an unsigned `kind:"read"` device_event carrying the plate, confidence,
 * region, and the `snapshotId` it was read from. Advisory: this records the plate
 * observed for the session; it does NOT feed the access decision (the flow already
 * decided). A low-confidence/no-plate result records nothing (a shaky read isn't a fact).
 * Best-effort + fail-soft — a vision error never surfaces on the (already-open) path.
 */
async function recognizePlate(
  db: Db,
  vision: VisionClient,
  deviceId: string,
  direction: FlowDirection,
  identity: string,
  snapshotId: string,
  shot: { bytes: Buffer; contentType: string },
  logger: FastifyBaseLogger,
  log?: EventLog | null,
): Promise<void> {
  try {
    const result = await vision.analyze(shot.bytes, shot.contentType);
    if (!result) return;
    // Boxes are kept as FRACTIONS of the analysed frame (the stored snapshot is a
    // downscaled copy — see reencodeForStorage), so the wash's review crop can cut the
    // vehicle out of whatever copy survives and blur the plate inside it.
    const frame = await frameSize(shot.bytes);
    const norm = (b: { x1: number; y1: number; x2: number; y2: number } | null | undefined) =>
      b && frame
        ? {
            x1: clamp01(b.x1 / frame.w), y1: clamp01(b.y1 / frame.h),
            x2: clamp01(b.x2 / frame.w), y2: clamp01(b.y2 / frame.h),
          }
        : null;
    // The vehicle's body type (advisory; the wash desk's category suggestion — see
    // venue-modules.md §Vehicle category). Rides the plate's read row when there is one,
    // else a row of its own: a car with an unreadable plate is still a car of some class.
    const vehicleBox = norm(result.vehicle?.bbox);
    const vehicle = result.vehicle
      ? { bodyType: result.vehicle.bodyType, bodyConfidence: result.vehicle.confidence, ...(vehicleBox ? { vehicleBox } : {}) }
      : {};
    const plate = !result.plate || result.lowConfidence ? "" : result.plate.text.trim().toUpperCase();
    const plateBox = plate ? norm(result.plate?.bbox) : null;
    if (!plate && !result.vehicle) return; // nothing trustworthy to record
    db.insert(deviceEventsTable)
      .values({
        id: randomUUID(),
        deviceId,
        category: "camera",
        kind: "read",
        // `identity` ties the read to the session; `snapshotId` to the evidence image.
        detail: {
          identity,
          direction,
          ...(plate
            ? { plate, confidence: result.plate!.confidence, region: result.plate!.region ?? null, ...(plateBox ? { plateBox } : {}) }
            : {}),
          ...vehicle,
          modelVersion: result.modelVersion,
          snapshotId,
          source: "entry-exit-snapshot",
        },
        occurredAt: new Date().toISOString(),
      })
      .run();
    if (result.vehicle) logger.info(`vision vehicle '${result.vehicle.bodyType}' (${result.vehicle.confidence.toFixed(3)}) for ${identity}`);
    if (!plate) return;
    logger.info(`anpr plate '${plate}' (${result.plate!.confidence.toFixed(3)}) for ${identity}`);
    // The session's entry/exit event already shipped without this (async) plate — tell the
    // booth so it backfills the plate badge in place (no refresh). Advisory; ledger untouched.
    deviceEvents.emitPlateRecognized({ identity, plate, direction });

    // ENTRY-SIDE duplicate check: this plate already OPEN under another recent session is
    // most likely the SAME car that minted a second ticket (a motion radar drops a
    // stationary car → the button re-arms). Signed anomaly for the operator to void.
    if (direction === "entry" && log) {
      await flagDuplicateEntryPlate({ db, log, identity, plate, snapshotId, logger });
    }
  } catch (err) {
    logger.warn(`anpr recognize failed (${identity}): ${(err as Error).message}`);
  }
}

function clamp01(v: number): number {
  return Math.max(0, Math.min(1, v));
}

/** Pixel size of the analysed frame (JPEG header only — cheap). Null when unreadable. */
async function frameSize(bytes: Buffer): Promise<{ w: number; h: number } | null> {
  try {
    const m = await sharp(bytes, { failOn: "none" }).metadata();
    return m.width && m.height ? { w: m.width, h: m.height } : null;
  } catch {
    return null;
  }
}

/** How far back a recognized entry plate is compared against other OPEN sessions'
 *  entry plates. Short on purpose: the duplicate-ticket scenario is the same car
 *  re-pressing within minutes; a long window would flag legit re-visits. */
function dupPlateWindowMs(): number {
  const raw = Number(process.env.ENTRY_DUP_PLATE_WINDOW_MIN ?? 15);
  return (Number.isFinite(raw) && raw > 0 ? raw : 15) * 60_000;
}

/**
 * Flag a freshly-recognized ENTRY plate that is already open under a DIFFERENT recent
 * session: sign ONE `entry.duplicatePlate` anomaly keyed to the new session, pointing at
 * the prior one. Mirrors the exit-side plateSwapSuspected pattern (advisory, post-hoc —
 * the barrier already opened; the operator voids the duplicate ticket). Exported for tests.
 */
export async function flagDuplicateEntryPlate(opts: {
  db: Db;
  log: EventLog;
  /** The session the plate was just recognized for (the NEW ticket). */
  identity: string;
  plate: string;
  snapshotId: string;
  logger: FastifyBaseLogger;
}): Promise<void> {
  const { db, log, identity, plate, snapshotId, logger } = opts;
  try {
    const cutoff = new Date(Date.now() - dupPlateWindowMs()).toISOString();
    // Recent entry-plate reads (unsigned `kind:"read"` telemetry, written above) for the
    // same plate under a different identity. detail is JSON — filter in JS; read volume
    // inside the window is tiny (one row per entry).
    const reads = db
      .select()
      .from(deviceEventsTable)
      .where(and(eq(deviceEventsTable.kind, "read"), gte(deviceEventsTable.occurredAt, cutoff)))
      .all();
    const prior = reads
      .map((r) => r.detail as { identity?: string; direction?: string; plate?: string })
      .find((d) => d.direction === "entry" && d.plate === plate && d.identity && d.identity !== identity);
    if (!prior?.identity) return;
    // Only a still-OPEN prior session is a duplicate suspect (a closed one drove off).
    const open = db
      .select()
      .from(sessions)
      .where(and(eq(sessions.id, prior.identity), eq(sessions.state, "open")))
      .get();
    if (!open) return;
    await log.append({
      type: "anomaly",
      identity,
      payload: {
        ...reasonPayload("entry.duplicatePlate", { plate, otherIdentity: prior.identity }),
        duplicateEntrySuspected: true,
        plate,
        otherIdentity: prior.identity,
        snapshotId,
      },
    });
    logger.warn(`duplicate entry suspected: plate ${plate} on ${identity} already open under ${prior.identity}`);
  } catch (err) {
    // Best-effort, post-hoc — never let the duplicate check surface on the open path.
    logger.error(`duplicate-plate check failed (${identity}): ${(err as Error).message}`);
  }
}

/** Build a live camera adapter from a resolved devices row, or null. Exported so the
 *  ANPR bridge (anpr-entry.ts) reuses the identical registry-build-or-null logic. */
export function buildCamera(row: { driverId: string; config: unknown }): CameraDevice | null {
  const driver = registry.get(row.driverId);
  if (!driver) return null;
  try {
    return driver.create(row.config as never) as CameraDevice;
  } catch {
    return null;
  }
}

// --- shared snapshot capture (one HTTP pull per camera per vehicle) -----------
// A Hikvision camera serves /ISAPI/.../picture SINGLE-THREADED: two concurrent
// snapshot GETs to the same unit return HTTP 503 "service busy". On a vehicle entry
// TWO paths capture the SAME camera within ~1s — the ANPR bridge (barrier-driving,
// anpr-entry.ts) and the advisory snapshotAsync (evidence + telemetry, below). They
// each `buildCamera()` a SEPARATE adapter instance, so a per-instance cache can't
// dedupe them. This module-level, deviceId-keyed cache does: it coalesces in-flight
// captures (the 2nd caller awaits the 1st's pull) AND serves a result captured within
// SNAPSHOT_TTL_MS, so the bridge + advisory share ONE frame instead of colliding into
// a 503 (which then burned the bridge's 12s debounce → the slow entry observed
// 2026-06-25; see wiki/concepts/lane-presence-and-anpr-entry.md).

/** How long a fresh capture is reused for the same camera. A car is one event for a
 *  couple of seconds; 1.5s comfortably spans the bridge→advisory gap without ever
 *  serving a stale frame for a *different* vehicle (entries are seconds apart). */
const SNAPSHOT_TTL_MS = 1500;

interface CacheEntry {
  /** A capture in flight — concurrent callers await this instead of issuing a 2nd GET. */
  inflight?: Promise<Snapshot>;
  /** The last SUCCESSFUL capture + when it resolved, for the freshness window. */
  last?: { shot: Snapshot; at: number };
}

const snapshotCache = new Map<string, CacheEntry>();

/**
 * Capture a snapshot for a camera, sharing ONE HTTP pull across concurrent/near-
 * simultaneous callers (the ANPR bridge and the advisory snapshot). Same contract as
 * `camera.captureSnapshot` (throws on failure) — a failed pull is NOT cached, so the
 * next caller retries rather than inheriting the error. Key by the stable `deviceId`.
 */
export function captureSnapshotShared(
  deviceId: string,
  camera: CameraDevice,
  ctx: { direction: FlowDirection },
): Promise<Snapshot> {
  const now = Date.now();
  let entry = snapshotCache.get(deviceId);
  if (!entry) {
    entry = {};
    snapshotCache.set(deviceId, entry);
  }
  // Fresh enough → reuse the last frame (same vehicle, no second hardware hit).
  if (entry.last && now - entry.last.at < SNAPSHOT_TTL_MS) {
    return Promise.resolve(entry.last.shot);
  }
  // A capture is already running → join it (this is what prevents the 503 collision).
  if (entry.inflight) return entry.inflight;
  // Otherwise issue the single real pull; record it as the in-flight promise.
  const pull = camera
    .captureSnapshot(ctx)
    .then((shot) => {
      entry.last = { shot, at: Date.now() };
      return shot;
    })
    .finally(() => {
      // Clear the in-flight slot whether it resolved or threw; a failure is never cached.
      if (entry.inflight === pull) entry.inflight = undefined;
    });
  entry.inflight = pull;
  return pull;
}

function recordFailure(
  db: Db,
  direction: FlowDirection,
  deviceId: string,
  identity: string,
  error: string,
  logger: FastifyBaseLogger,
): void {
  logger.warn(`snapshot failed (${direction}, ${identity}): ${error}`);
  recordEvent(db, direction, deviceId, identity, { ok: false, error }, logger);
}

function recordEvent(
  db: Db,
  direction: FlowDirection,
  deviceId: string,
  identity: string,
  detail: Record<string, unknown>,
  logger: FastifyBaseLogger,
): void {
  try {
    db.insert(deviceEventsTable)
      .values({
        id: randomUUID(),
        deviceId,
        category: "camera",
        kind: "snapshot",
        detail: { ...detail, direction, identity },
        occurredAt: new Date().toISOString(),
      })
      .run();
  } catch (err) {
    // Telemetry is best-effort; never let it surface on the (already-open) path.
    logger.error(`snapshot device-event insert failed: ${(err as Error).message}`);
  }
}