feat(server): live booth WebSocket feed (/api/ws)

Add @fastify/websocket. EventLog fires an onAppended callback after each durable
append; device-events gains a ledger channel (emitLedger). /api/ws fans out
ledger + occupancy + printer-status to authenticated booth clients. Origin
allowlist (WS_ALLOWED_ORIGINS) replaces CSRF for the handshake (anti-CSWSH).

Note: server.ts also reflects later booth route wiring; the final HEAD builds.
This commit is contained in:
2026-06-18 11:00:22 +02:00
parent 58d8f06ba0
commit c2f06a5d2a
6 changed files with 162 additions and 4 deletions
+4
View File
@@ -25,3 +25,7 @@ EVENT_SIGNING_KEY=
# First admin (seed once): pnpm --filter @parking/server seed-admin # First admin (seed once): pnpm --filter @parking/server seed-admin
# ADMIN_USER=admin # ADMIN_USER=admin
# ADMIN_PASS= # ADMIN_PASS=
# Comma-separated extra origins allowed to open the booth WebSocket (/api/ws).
# In dev, set the Vite SPA origin. Same-origin is always allowed without this.
WS_ALLOWED_ORIGINS=http://localhost:5173
+1
View File
@@ -16,6 +16,7 @@
"@fastify/cors": "11.2.0", "@fastify/cors": "11.2.0",
"@fastify/jwt": "10.1.0", "@fastify/jwt": "10.1.0",
"@fastify/static": "9.1.3", "@fastify/static": "9.1.3",
"@fastify/websocket": "^11.2.0",
"@parking/db": "workspace:*", "@parking/db": "workspace:*",
"@parking/devices": "workspace:*", "@parking/devices": "workspace:*",
"@parking/shared": "workspace:*", "@parking/shared": "workspace:*",
+16
View File
@@ -1,5 +1,6 @@
import { EventEmitter } from "node:events"; import { EventEmitter } from "node:events";
import type { PrinterStatus } from "@parking/devices"; import type { PrinterStatus } from "@parking/devices";
import type { LedgerEventRow } from "@parking/db";
// Internal event bus for device-originated events (button presses, etc.). // Internal event bus for device-originated events (button presses, etc.).
// Hardware drivers / inbound device pushes emit here; business logic (entry // Hardware drivers / inbound device pushes emit here; business logic (entry
@@ -74,6 +75,21 @@ class DeviceEventBus extends EventEmitter {
this.on("printer-status", cb); this.on("printer-status", cb);
return () => this.off("printer-status", cb); return () => this.off("printer-status", cb);
} }
/**
* Emitted AFTER a signed business event is appended to the ledger (entry, exit,
* payment, void, …). The payload is the persisted row — business facts only, no
* secrets — so it is safe to fan out to authenticated booth clients over the WS.
* This is a read-side notification ONLY: it never feeds back into append/sign/
* chain logic. See event-log.ts (emitted from EventLog.append) and routes/ws.ts.
*/
emitLedger(event: LedgerEventRow): void {
this.emit("ledger", event);
}
onLedger(cb: (event: LedgerEventRow) => void): () => void {
this.on("ledger", cb);
return () => this.off("ledger", cb);
}
} }
/** Process-wide device event bus. */ /** Process-wide device event bus. */
+21 -2
View File
@@ -92,13 +92,23 @@ export class EventLog {
* (JWT-fallback → dedicated key → ATECC608). Defaults to the append signer for * (JWT-fallback → dedicated key → ATECC608). Defaults to the append signer for
* callers that don't pass one (single-key chains, tests). */ * callers that don't pass one (single-key chains, tests). */
readonly #resolveVerifier: SignerResolver; readonly #resolveVerifier: SignerResolver;
/** Optional read-side notification, fired AFTER a row is durably inserted. Used
* to fan the event out to live booth clients (WS). It is best-effort and must
* NOT influence the append/sign/chain path — a throwing/absent sink is ignored. */
readonly #onAppended?: (row: LedgerEventRow) => void;
/** Serialize appends: each waits for the previous to finish. */ /** Serialize appends: each waits for the previous to finish. */
#tail: Promise<unknown> = Promise.resolve(); #tail: Promise<unknown> = Promise.resolve();
constructor(db: Db, signer: Signer, resolveVerifier?: SignerResolver) { constructor(
db: Db,
signer: Signer,
resolveVerifier?: SignerResolver,
onAppended?: (row: LedgerEventRow) => void,
) {
this.#db = db; this.#db = db;
this.#signer = signer; this.#signer = signer;
this.#resolveVerifier = resolveVerifier ?? (() => signer); this.#resolveVerifier = resolveVerifier ?? (() => signer);
this.#onAppended = onAppended;
} }
/** Append one event to the chain. Returns the persisted row. Serialized. */ /** Append one event to the chain. Returns the persisted row. Serialized. */
@@ -106,7 +116,16 @@ export class EventLog {
const run = this.#tail.then(() => this.#appendNow(input)); const run = this.#tail.then(() => this.#appendNow(input));
// Keep the chain going even if one append rejects (don't wedge the lock). // Keep the chain going even if one append rejects (don't wedge the lock).
this.#tail = run.catch(() => undefined); this.#tail = run.catch(() => undefined);
return run; // Read-side notification, AFTER the row is durably written. Wrapped so a
// failing sink can never reject the append or break the chain lock above.
return run.then((row) => {
try {
this.#onAppended?.(row);
} catch {
// best-effort fan-out only — swallow.
}
return row;
});
} }
#appendNow(input: AppendInput): LedgerEventRow { #appendNow(input: AppendInput): LedgerEventRow {
+104
View File
@@ -0,0 +1,104 @@
import type { FastifyInstance } from "fastify";
import type { Db } from "@parking/db";
import type { Role } from "@parking/shared";
import { deviceEvents } from "../device-events.js";
import { getOccupancy } from "../occupancy.js";
// Live booth feed over a WebSocket. The booth UI opens ONE socket and receives
// server-pushed updates instead of polling: each signed ledger append (entry,
// exit, payment, void) is fanned out, and the recomputed occupancy rides along
// so the screen's count stays exact (occupancy is a fold over the same ledger,
// never a counter). Printer-status changes are forwarded too.
//
// Auth: the handshake is a normal GET through Fastify's lifecycle, so the same
// HttpOnly JWT cookie that guards the REST API guards this. We verify the JWT and
// role here. A browser's WebSocket constructor cannot set custom headers, so the
// CSRF double-submit header the REST mutations use is unavailable — which would
// leave the socket open to Cross-Site WebSocket Hijacking: a malicious page in the
// operator's browser could open ws://<booth>/api/ws, the browser would auto-attach
// the HttpOnly cookie, and the attacker would receive the live entry/exit/payment
// stream. The cookie alone is NOT a control here. So we replace the CSRF check with
// an Origin allowlist: the handshake's Origin must be same-origin (or an explicitly
// allowed booth UI origin). Non-browser clients (no Origin) are rejected too.
// See auth.ts, event-log.ts (emitLedger), capacity-occupancy.md.
/** Roles allowed to watch the live feed (everyone signed in; readonly included —
* it's a read-only stream). */
const WATCH_ROLES: Role[] = ["admin", "operator", "cashier", "readonly"];
/**
* Is the handshake's Origin trusted? Same-origin (Origin host === Host header) is
* always allowed; additional origins can be allowlisted via WS_ALLOWED_ORIGINS
* (comma-separated) for a booth UI served from a different origin. A missing or
* mismatched Origin is rejected — that is the anti-CSWSH control.
*/
function isAllowedOrigin(origin: string | undefined, host: string | undefined): boolean {
if (!origin) return false; // no Origin → not a same-origin browser request
let originHost: string;
try {
originHost = new URL(origin).host;
} catch {
return false; // malformed Origin
}
if (host && originHost === host) return true; // same-origin (any scheme/port match via host)
const allow = (process.env.WS_ALLOWED_ORIGINS ?? "")
.split(",")
.map((s) => s.trim())
.filter(Boolean);
return allow.includes(origin);
}
type OutMsg =
| { kind: "hello"; occupancy: ReturnType<typeof getOccupancy> }
| { kind: "ledger"; event: unknown; occupancy: ReturnType<typeof getOccupancy> }
| { kind: "printer-status"; event: unknown };
export async function wsRoutes(app: FastifyInstance, db: Db): Promise<void> {
app.get(
"/api/ws",
{
websocket: true,
// Origin allowlist (anti-CSWSH, replaces CSRF — see file header) THEN JWT +
// role. Reject a cross/absent origin before touching the token, so a hijack
// attempt never reaches an authenticated socket. jwtVerify reads the cookie.
preHandler: async (req) => {
if (!isAllowedOrigin(req.headers.origin, req.headers.host)) {
throw Object.assign(new Error("forbidden origin"), { statusCode: 403 });
}
await req.jwtVerify();
if (!req.user || !WATCH_ROLES.includes(req.user.role)) {
throw Object.assign(new Error("forbidden"), { statusCode: 403 });
}
},
},
(socket) => {
const send = (msg: OutMsg) => {
// readyState 1 = OPEN; never throw out of an event-bus callback.
if (socket.readyState === 1) {
try {
socket.send(JSON.stringify(msg));
} catch {
/* drop on a broken socket */
}
}
};
// Initial snapshot so the client renders immediately, before any event.
send({ kind: "hello", occupancy: getOccupancy(db) });
// Subscribe to the live buses. Each handler recomputes occupancy from the
// ledger (cheap fold) so the pushed count is always authoritative.
const offLedger = deviceEvents.onLedger((event) => {
send({ kind: "ledger", event, occupancy: getOccupancy(db) });
});
const offPrinter = deviceEvents.onPrinterStatus((event) => {
send({ kind: "printer-status", event });
});
socket.on("close", () => {
offLedger();
offPrinter();
});
},
);
}
+16 -2
View File
@@ -1,5 +1,6 @@
import cookie from "@fastify/cookie"; import cookie from "@fastify/cookie";
import jwt from "@fastify/jwt"; import jwt from "@fastify/jwt";
import websocket from "@fastify/websocket";
import Fastify, { type FastifyInstance } from "fastify"; import Fastify, { type FastifyInstance } from "fastify";
import { randomUUID } from "node:crypto"; import { randomUUID } from "node:crypto";
import { createDb, deviceEvents as deviceEventsTable, type Db } from "@parking/db"; import { createDb, deviceEvents as deviceEventsTable, type Db } from "@parking/db";
@@ -26,6 +27,7 @@ import { snapshotRoutes } from "./routes/snapshots.js";
import { tariffRoutes } from "./routes/tariffs.js"; import { tariffRoutes } from "./routes/tariffs.js";
import { printerRoutes } from "./routes/printers.js"; import { printerRoutes } from "./routes/printers.js";
import { setupRoutes } from "./routes/setup.js"; import { setupRoutes } from "./routes/setup.js";
import { wsRoutes } from "./routes/ws.js";
// The backend is Fastify (Node). Hardware drivers live as isolated Fastify // The backend is Fastify (Node). Hardware drivers live as isolated Fastify
// plugins emitting onto a shared internal event bus; auth is fully local // plugins emitting onto a shared internal event bus; auth is fully local
@@ -44,6 +46,10 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
await app.register(cookie); await app.register(cookie);
// WebSocket support for the live booth feed (/api/ws). Registered before the
// routes so the `{ websocket: true }` route option is available.
await app.register(websocket);
// Local JWT signing with a local secret — no external identity provider. // Local JWT signing with a local secret — no external identity provider.
// Fail fast rather than fall back to a known default: a booth machine started // Fail fast rather than fall back to a known default: a booth machine started
// without a real secret would sign tokens anyone could forge (incl. an admin // without a real secret would sign tokens anyone could forge (incl. an admin
@@ -86,9 +92,17 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
// in device_events. The entry flow (TODO) turns an input into a signed // in device_events. The entry flow (TODO) turns an input into a signed
// vehicle_entry once a ticket prints + the barrier is commanded. // vehicle_entry once a ticket prints + the barrier is commanded.
// See wiki/decisions/event-streams-split.md. // See wiki/decisions/event-streams-split.md.
const eventLog = new EventLog(db, buildSigner(app.log), buildVerifier); // The 4th arg is a read-side fan-out fired AFTER each durable append — used to
// push the event to live booth clients (WS). It cannot affect the sign/chain path.
const eventLog = new EventLog(db, buildSigner(app.log), buildVerifier, (row) =>
deviceEvents.emitLedger(row),
);
await eventRoutes(app, db, eventLog); await eventRoutes(app, db, eventLog);
// Live booth feed: server-pushed ledger + occupancy + printer-status over a
// single authenticated WebSocket (/api/ws). See routes/ws.ts.
await wsRoutes(app, db);
// Entry/exit camera snapshots (BLOB-in-DB), read-only. See snapshot.ts. // Entry/exit camera snapshots (BLOB-in-DB), read-only. See snapshot.ts.
await snapshotRoutes(app, db); await snapshotRoutes(app, db);
@@ -121,7 +135,7 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
// Pay station (pay-on-foot): quote an open session against the active tariff + // Pay station (pay-on-foot): quote an open session against the active tariff +
// take payment → signed `payment` event. See wiki/concepts/tariff.md. // take payment → signed `payment` event. See wiki/concepts/tariff.md.
const payStation = new PayStation(db, eventLog, app.log); const payStation = new PayStation(db, eventLog, app.log);
await payRoutes(app, payStation); await payRoutes(app, db, payStation, exitFlow);
// Tariff composer: admin publishes effective-dated, immutable rate-card versions // Tariff composer: admin publishes effective-dated, immutable rate-card versions
// the pay station prices against. See wiki/concepts/tariff.md. // the pay station prices against. See wiki/concepts/tariff.md.