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 PlanTimeframes, type ReasonCode } from "@parking/shared";
import type { FastifyBaseLogger } from "fastify";
import { printWindowChargeNotice } from "./booth-print.js";
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";
import { planVersionById, windowCharge, windowOwedBetween } from "./subscription-window.js";
import type { VisionClient } from "./vision-client.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>();
  /** Optional vision client — passed to snapshotAsync so ANPR runs on the subscriber image. */
  readonly #vision: VisionClient | null;

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

  /** 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) {
      // CHANNEL AGREEMENT (clone defense, 2026-07-04). When the reader CONFIRMED the
      // physical channel (DT-008 output prefixes), the credential kind must agree: an
      // OPTICAL decode may not claim an RF credential — otherwise printing a card's
      // UID (often written on the card face) as a barcode clones the card. Symmetric
      // for an RF read claiming a QR credential (a mis-encoded clone tag). A legacy
      // untagged read (channel undefined) matches as before — enforcement only bites
      // where prefixes are deployed. The attempt itself is a fraud signal → signed
      // anomaly, then treated as no-match (the flows refuse it as unknown).
      const mismatch =
        (e.channel === "optical" && cred.kind === "rf") ||
        (e.channel === "rf" && cred.kind === "qr");
      if (mismatch) {
        this.#logger.warn(
          `credential channel mismatch: ${cred.kind} credential '${e.value}' presented via ${e.channel} (sub ${cred.subscriptionId}) — possible clone`,
        );
        void this.#log
          .append({
            type: "anomaly",
            identity: cred.subscriptionId,
            payload: {
              ...reasonPayload("sub.refused.channelMismatch", {
                credentialKind: cred.kind,
                channel: e.channel === "optical" ? "optical" : "rf",
              }),
              channelMismatch: true,
              credentialKind: cred.kind,
              channel: e.channel,
              value: e.value,
              deviceId: e.deviceId,
            },
          })
          .catch((err) => this.#logger.error(`channel-mismatch anomaly append failed: ${(err as Error).message}`));
        return null;
      }
      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();
    // A soft-deleted (recycle-bin) subscription must NOT open the barrier — treat it as
    // gone. (Its credential rows are kept for restore, so the dispatcher can still match
    // it; the gate is here.)
    if (!sub || sub.deletedAt) 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;

      // TARIFF BRIDGE — exit gate. Total owed = carried early-entry charge (signed on the
      // entry payload) + a late-exit charge (window-close→now) computed fresh. If the
      // subscriber owes money and hasn't paid it, REFUSE the exit (like the transient
      // unpaid/overstay gate) — they settle at the booth (a signed `payment` keyed to the
      // occurrence), then re-scan. This is a host-ONLINE business gate; the offline path
      // still fails open. See wiki/entities/subscription.md ("tariff bridge").
      const owed = this.#windowOwed(occurrenceId, m.subscriptionId, sub.planVersionId);
      const paid = this.#windowPaidMinor(occurrenceId);
      if (owed.totalMinor - paid > 0) {
        const reason = await this.#reject(m, "exit", "sub.refused.unpaidWindow", {
          amount: ((owed.totalMinor - paid) / 100).toFixed(2),
          currency: owed.currency ?? "",
        });
        return { accepted: false, direction: "exit", reason };
      }

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

    // TARIFF BRIDGE — out-of-window entry. If the plan has time windows and this scan is
    // OUTSIDE the allowed window, the subscriber will owe the transient tariff for the time
    // they actually park out-of-window. The AMOUNT is NOT knowable now — it depends on when
    // they leave (a subscriber who enters early and leaves before the window opens owes only
    // their parked minutes, NOT the whole gap-to-window-open). So we stamp only a MARKER
    // (`outOfWindow`) + the tariff version, and price it live at settlement from
    // minutesOutsideWindow(entry → pay-time), which caps at the window edges. Open now
    // (never trap); the charge is gated at exit. Plans without timeframes → null → no
    // marker. See wiki/entities/subscription.md ("tariff bridge").
    const outOfWindow = windowCharge(this.#db, sub.planVersionId, now, "entry");

    // 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,
      // The subscription IS the authorization (no fee for in-window use). `permitId`/`permit`
      // are the on-chain field names (immutable). An out-of-window entry is MARKED here
      // (`outOfWindow` + the tariff version for reproducible pricing) so the booth/exit gate
      // know to charge the parked-out-of-window minutes — priced live, not a fixed amount.
      payload: {
        sessionRef: occurrenceId,
        permitId: m.subscriptionId,
        permit: true,
        via: m.via,
        ...(outOfWindow
          ? {
              outOfWindow: true,
              windowTariffVersionId: outOfWindow.tariffVersionId,
            }
          : {}),
      },
      occurredAt: now,
    });
    if (outOfWindow) {
      this.#logger.info(
        `subscription out-of-window entry marked on ${occurrenceId} (charge priced from parked minutes at exit)`,
      );
    }
    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}`);
    }
    // BEST-EFFORT: print the out-of-window TICKET so the subscriber has the paper the
    // operator scans to settle at the booth. It carries the occurrence id as a scannable
    // code; the amount is computed at settlement from the minutes actually parked
    // out-of-window (capped at the window edges). AFTER the open + cache, and fully
    // swallowed — a missing/failed printer must NEVER block or delay the barrier.
    if (outOfWindow) {
      const tf = (planVersionById(this.#db, sub.planVersionId)?.timeframes ?? null) as PlanTimeframes | null;
      void printWindowChargeNotice(
        this.#db,
        { occurrenceId, holderName: sub.holderName, at: now, windowOpensMin: tf?.fromMin, edge: "entry" },
        this.#logger,
      ).catch((err) => this.#logger.warn(`out-of-window slip print failed for ${occurrenceId}: ${(err as Error).message}`));
    }
    return { accepted: true, direction: "entry" };
  }

  /**
   * Total out-of-window charge owed for an occurrence right now: the transient cost of the
   * minutes parked OUTSIDE the plan's window over the WHOLE stay `[entry, now]` — ONE
   * computation covering early entry AND late exit (not entry-gap + exit-gap, which
   * double-counts and lets the exit gap reach a previous day's close). A plan without
   * timeframes yields 0. Single source of truth shared with the booth quote.
   */
  #windowOwed(
    occurrenceId: string,
    _subscriptionId: string,
    planVersionId: string | null,
  ): { totalMinor: number; currency: string | null } {
    const entryRow = this.#db
      .select()
      .from(ledgerEvents)
      .where(eq(ledgerEvents.identity, occurrenceId))
      .all()
      .find((r) => r.type === "vehicle_entry");
    if (!entryRow) return { totalMinor: 0, currency: null };
    const owed = windowOwedBetween(this.#db, planVersionId, entryRow.occurredAt, new Date().toISOString());
    return { totalMinor: owed?.amountMinor ?? 0, currency: owed?.currency ?? null };
  }

  /** Sum of signed `payment` events keyed to this occurrence (what the subscriber has
   *  already paid toward their window charge). Folds the append-only ledger. */
  #windowPaidMinor(occurrenceId: string): number {
    const rows = this.#db.select().from(ledgerEvents).where(eq(ledgerEvents.identity, occurrenceId)).all();
    let paid = 0;
    for (const r of rows) {
      if (r.type !== "payment") continue;
      const pl = (r.payload ?? {}) as { amountMinor?: number };
      if (typeof pl.amountMinor === "number") paid += pl.amountMinor;
    }
    return paid;
  }

  /**
   * 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.
   */
  /** How many occurrences this subscription currently has OPEN (entries not yet exited).
   *  Public so the ANPR bridge can detect a credential (card/QR) exit landing mid-poll — if
   *  the count drops while it's polling, the subscriber already transacted and the bridge must
   *  NOT also emit (which would exit the NEXT open occurrence — a phantom double-exit, esp. for
   *  a fleet sub). See anpr-entry.ts. */
  openOccurrenceCount(subscriptionId: string): number {
    return this.#openOccurrences(subscriptionId).length;
  }

  #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, vision: this.#vision }).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;
    }
  }
}