parking_solution/apps/server/src/subscription-flow.ts

import { randomUUID } from "node:crypto";
import {
  eq,
  ledgerEvents,
  sessions,
  subscriptionCredentials,
  subscriptionPlates,
  subscriptions,
  type Db,
  type DeviceRow,
} from "@parking/db";
import { registry, type AccessControlDevice } from "@parking/devices";
import { reasonPayload, type ReasonCode } from "@parking/shared";
import type { FastifyBaseLogger } from "fastify";
import type { DeviceReadEvent, ReadOutcome } from "./device-events.js";
import type { EventLog } from "./event-log.js";
import { type FlowDirection, type ResolvedRelay } from "./device-resolve.js";
import { snapshotAsync } from "./snapshot.js";

// SUBSCRIPTION flow: a subscriber identified by card/QR/plate enters/exits without
// paying per stay (they're on a recurring plan). Reached from the read dispatcher
// when a read matches a subscription (not an open ticket). See
// wiki/entities/subscription.md.
//
// Two optional, independent bindings:
//   - car-count: `maxConcurrent` (default 1, null = unbound) — how many of the
//     subscription's cars may be inside at once; enforced over the session projection.
//   - plate: optional `plates[]` — when set, a matching plate is an accepted identity
//     too (card/QR OR plate). When unset, any car may use the subscription's card/QR.
//
// Direction is inferred from the SUBSCRIPTION's open-session state, NOT the specific
// credential read — so ANY of a subscription's credentials (QR / RFID / NFC / plate)
// may open or close a session. Entry mints a fresh per-occurrence session id (the
// ledger `identity`); a read with no open occurrence → ENTRY; with ≥1 open → EXIT the
// OLDEST open occurrence (FIFO). A fleet (maxConcurrent > 1) thus has several open
// occurrences at once; each read closes one. This decouples exit from the entry
// credential (you can enter with QR and leave with the card).
//
// NB: the SIGNED ledger payload still carries `permitId` (immutable history — see the
// schema note); the per-occurrence `identity` is the session key. The mutable master
// data / code is "subscription"; the on-chain field name is left as-is so historical
// events keep verifying.

export interface SubscriptionMatch {
  readonly subscriptionId: string;
  /** The specific credential/plate value read (for logging/anomalies). NOT the
   *  session key — sessions are keyed by subscription occurrence, so a different
   *  credential of the same subscription can close the session it opened. */
  readonly carKey: string;
  readonly via: "card" | "qr" | "plate";
}

export class SubscriptionFlow {
  readonly #db: Db;
  readonly #log: EventLog;
  readonly #logger: FastifyBaseLogger;
  readonly #inFlight = new Set<string>();

  constructor(db: Db, log: EventLog, logger: FastifyBaseLogger) {
    this.#db = db;
    this.#log = log;
    this.#logger = logger;
  }

  /** Resolve a read to a subscription (by card/QR credential, or a bound plate), or null. */
  match(e: DeviceReadEvent): SubscriptionMatch | null {
    // Card / QR / generic credential value.
    const cred = this.#db
      .select()
      .from(subscriptionCredentials)
      .where(eq(subscriptionCredentials.value, e.value))
      .get();
    if (cred) {
      return { subscriptionId: cred.subscriptionId, carKey: e.value, via: cred.kind === "qr" ? "qr" : "card" };
    }
    // Plate binding: a read plate that matches a subscription's bound plate is an identity.
    if (e.kind === "plate") {
      const plate = this.#db.select().from(subscriptionPlates).where(eq(subscriptionPlates.plate, e.value)).get();
      if (plate) return { subscriptionId: plate.subscriptionId, carKey: e.value, via: "plate" };
    }
    return null;
  }

  /** Run the subscription entry/exit for a matched read at a barrier. `resolved` is the
   *  reader's bound relay; its direction constrains, "both" defers to session state. */
  async run(resolved: ResolvedRelay, e: DeviceReadEvent, m: SubscriptionMatch): Promise<ReadOutcome> {
    const key = `${m.subscriptionId}:${m.carKey}`;
    if (this.#inFlight.has(key)) return { accepted: false, reason: "duplicate read in flight" };
    this.#inFlight.add(key);
    try {
      return await this.#run(resolved, e, m);
    } catch (err) {
      this.#logger.error(`subscription-flow failed: ${(err as Error).message}`);
      return { accepted: false, reason: (err as Error).message };
    } finally {
      this.#inFlight.delete(key);
    }
  }

  async #run(resolved: ResolvedRelay, e: DeviceReadEvent, m: SubscriptionMatch): Promise<ReadOutcome> {
    // The physical side the reader sits at — used to fire the right camera on a refusal
    // that happens BEFORE we infer the entry/exit verb ("both" defers to entry).
    const lane: FlowDirection = resolved.direction === "exit" ? "exit" : "entry";
    const sub = this.#db.select().from(subscriptions).where(eq(subscriptions.id, m.subscriptionId)).get();
    if (!sub) return { accepted: false, reason: await this.#reject(m, lane, "sub.refused.notFound") };

    // Validity: active + within the coverage window.
    const now = new Date().toISOString();
    const invalid =
      sub.status !== "active" ||
      (sub.validFrom != null && now < sub.validFrom) ||
      (sub.validTo != null && now > sub.validTo);
    if (invalid) {
      const reason = await this.#reject(m, lane, "sub.refused.outOfWindow", { status: sub.status });
      return { accepted: false, reason };
    }

    // Direction: the BARRIER the reader sits at decides the verb — an entry-lane read
    // is an ENTRY, an exit-lane read is an EXIT. (The credential is decoupled from the
    // session, so we can't and needn't infer from "which credential".) A "both" barrier
    // has no physical side, so there we infer from state: open occurrence → exit, else
    // entry. This is what lets a FLEET admit several cars (each entry-lane read is an
    // entry) yet exit any of them with ANY credential (FIFO).
    const open = this.#openOccurrences(m.subscriptionId);
    const verb: FlowDirection =
      resolved.direction === "entry"
        ? "entry"
        : resolved.direction === "exit"
          ? "exit"
          : open.length > 0
            ? "exit"
            : "entry";

    const source = m.via === "plate" ? "lpr" : m.via === "qr" ? "qr" : "wiegand";

    if (verb === "exit") {
      // EXIT: close the OLDEST open occurrence (FIFO). Its occurrence id is the session
      // key; the credential just read may differ from the one that opened it. If the
      // subscription has NOTHING open, an exit read is a no-op anti-passback signal.
      const oldest = open[0];
      if (!oldest) {
        const reason = await this.#reject(m, "exit", "sub.refused.noSession");
        return { accepted: false, direction: "exit", reason };
      }
      const occurrenceId = oldest.identity;
      await this.#log.append({
        type: "vehicle_exit",
        direction: "exit",
        source,
        identity: occurrenceId,
        // `permitId` carries the subscription id; `via` records which credential left.
        payload: { sessionRef: occurrenceId, permitId: m.subscriptionId, via: m.via },
      });
      await this.#open(resolved, "exit", occurrenceId, "subscription exit");
      this.#closeCache(occurrenceId);
      return { accepted: true, direction: "exit" };
    }

    // ENTRY: enforce the car-count binding (maxConcurrent), then sign + open. Mint a
    // fresh per-occurrence id so a fleet can have several open at once.
    if (sub.maxConcurrent != null && open.length >= sub.maxConcurrent) {
      const reason = await this.#reject(m, "entry", "sub.refused.atCapacity", {
        inUse: open.length,
        max: sub.maxConcurrent,
      });
      return { accepted: false, direction: "entry", reason };
    }

    // A short, unique occurrence id. The subscription id is NOT embedded — it rides in
    // the payload's `permitId` (which every fold matches on), so the key stays compact.
    const occurrenceId = `SUBSESS-${randomUUID().replace(/-/g, "").slice(0, 12)}`;
    await this.#log.append({
      type: "vehicle_entry",
      direction: "entry",
      source,
      identity: occurrenceId,
      // No ticket, no fee — the subscription IS the authorization. Recorded for audit.
      // `permitId`/`permit` are the on-chain field names (immutable).
      payload: { sessionRef: occurrenceId, permitId: m.subscriptionId, permit: true, via: m.via },
      occurredAt: now,
    });
    await this.#open(resolved, "entry", occurrenceId, "subscription entry");
    try {
      this.#db
        .insert(sessions)
        .values({
          id: occurrenceId,
          identity: occurrenceId,
          source: m.via === "plate" ? "lpr" : "wiegand",
          subscriptionId: m.subscriptionId,
          enteredAt: now,
          state: "open",
        })
        .run();
    } catch (err) {
      this.#logger.error(`session-cache insert failed for ${occurrenceId}: ${(err as Error).message}`);
    }
    return { accepted: true, direction: "entry" };
  }

  /**
   * The OPEN occurrences of a subscription right now, **oldest first** (FIFO) — a
   * fold over the signed ledger. An occurrence is a `vehicle_entry` (whose
   * `payload.permitId` is this subscription) with no later `vehicle_exit` on the same
   * `identity`. Used to (a) infer entry vs. exit for ANY credential of the
   * subscription, (b) pick which occurrence a read closes, and (c) enforce
   * `maxConcurrent`. The on-chain field is `permitId`, so we match against that.
   */
  #openOccurrences(subscriptionId: string): { identity: string; index: number }[] {
    const rows = this.#db.select().from(ledgerEvents).orderBy(ledgerEvents.index).all();
    // Net entries−exits per occurrence identity, keeping the entry order (oldest first).
    const net = new Map<string, number>();
    const firstIndex = new Map<string, number>();
    for (const r of rows) {
      const id = r.identity;
      if (!id) continue;
      const pl = (r.payload ?? {}) as { permitId?: string };
      if (r.type === "vehicle_entry") {
        if (pl.permitId !== subscriptionId) continue;
        net.set(id, (net.get(id) ?? 0) + 1);
        if (!firstIndex.has(id)) firstIndex.set(id, r.index);
      } else if (r.type === "vehicle_exit") {
        if (!net.has(id)) continue; // not one of this subscription's occurrences
        net.set(id, (net.get(id) ?? 0) - 1);
      }
    }
    const open: { identity: string; index: number }[] = [];
    for (const [id, n] of net) if (n > 0) open.push({ identity: id, index: firstIndex.get(id) ?? 0 });
    open.sort((a, b) => a.index - b.index); // oldest first → FIFO
    return open;
  }

  /** Sign a refused-subscription anomaly with a localizable reason code, fire the
   *  directional evidence camera, and return the rendered English reason for the
   *  caller's ReadOutcome. `dir` is the lane the refusal happened at (entry/exit) so
   *  the right camera captures the turned-away subscriber. `via` records which
   *  credential was presented. */
  async #reject(
    m: SubscriptionMatch,
    dir: FlowDirection,
    code: ReasonCode,
    params?: Record<string, string | number>,
  ): Promise<string> {
    const rp = reasonPayload(code, params);
    await this.#log.append({
      type: "anomaly",
      identity: m.carKey,
      // `permitId`/`permitRefused` are the on-chain field names (immutable).
      payload: { ...rp, permitId: m.subscriptionId, permitRefused: true, via: m.via },
    });
    this.#fireSnapshot(dir, m.carKey);
    this.#logger.warn(`subscription refused (${m.carKey}): ${rp.reason}`);
    return rp.reason;
  }

  /** Fire the directional camera(s) for a refused-subscription event; never awaited
   *  (evidence, not a gate). The accepted entry/exit paths snapshot inside #open. */
  #fireSnapshot(dir: FlowDirection, identity: string): void {
    void snapshotAsync({ db: this.#db, direction: dir, identity, logger: this.#logger }).catch((err) =>
      this.#logger.error(`subscription snapshot error: ${(err as Error).message}`),
    );
  }

  async #open(resolved: ResolvedRelay, dir: FlowDirection, carKey: string, what: string): Promise<void> {
    const access = this.#buildAccess(resolved.controller);
    if (access) await access.pulseOpen(resolved.relay);
    else this.#logger.warn(`${what} signed for ${carKey} but the ${dir} relay won't build`);

    // SNAPSHOT — fire the directional camera(s), never awaited (evidence, not a gate).
    this.#fireSnapshot(dir, carKey);
  }

  #closeCache(carKey: string): void {
    try {
      this.#db.update(sessions).set({ exitedAt: new Date().toISOString(), state: "closed" }).where(eq(sessions.id, carKey)).run();
    } catch (err) {
      this.#logger.error(`session-cache close failed for ${carKey}: ${(err as Error).message}`);
    }
  }

  /** Build a live access adapter from a resolved controller row, or null. */
  #buildAccess(row: DeviceRow): AccessControlDevice | null {
    const driver = registry.get(row.driverId);
    if (!driver) return null;
    try {
      return driver.create(row.config as never) as AccessControlDevice;
    } catch {
      return null;
    }
  }
}