julian
2a13b95da6
The poll-until-confident loop (prev commit) opened a race: during its ~8s window a subscriber could scan their card/QR at the reader and exit immediately — but the ANPR loop kept polling and would ALSO emit a confident read a moment later, exiting the NEXT open occurrence (a phantom double-exit, worst for a fleet sub with several open). Guard it with the subscriber's open-occurrence count: the bridge identifies the subscription as soon as a frame reads the bound plate (identity needs no confidence), baselines openOccurrenceCount, then each tick AND before emit checks if it moved. If a credential closed/opened an occurrence mid-poll, the subscriber already transacted → abort, don't emit. New public SubscriptionFlow.openOccurrenceCount(). Bounded loop is unchanged (ANPR_POLL_WINDOW_MS=8000 cap; never infinite). +1 test (credential transacts mid-poll → no double-act); 170 server tests green. Claude-Session: https://claude.ai/code/session_01Xcm6ikLgGoCxxHrxtjkk5V
405 lines
19 KiB
TypeScript
405 lines
19 KiB
TypeScript
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) {
|
||
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;
|
||
}
|
||
}
|
||
}
|