diff --git a/components/grid-wallet-prod/src/app/api/webhooks/stream/route.ts b/components/grid-wallet-prod/src/app/api/webhooks/stream/route.ts index 3b3ad4e1..0fe1c726 100644 --- a/components/grid-wallet-prod/src/app/api/webhooks/stream/route.ts +++ b/components/grid-wallet-prod/src/app/api/webhooks/stream/route.ts @@ -1,4 +1,4 @@ -import { subscribe, type WebhookEvent } from '@/lib/webhookEvents'; +import { eventsSince, listenerCount, subscribe, type WebhookEvent } from '@/lib/webhookEvents'; export const runtime = 'nodejs'; export const dynamic = 'force-dynamic'; @@ -8,11 +8,38 @@ const KEEPALIVE_MS = 15_000; /** * Server-sent events: every webhook that passes signature verification is pushed - * to connected panels as it lands. One-way and text-only, so it survives the - * ngrok tunnel a real Grid webhook arrives through — no polling, no client - * timers, and the panel shows the event at the moment Grid delivers it. + * to EVERY connected panel as it lands. The demo drives one hardcoded customer, so + * all clients are watching the same account and all of them get every delivery. + * One-way and text-only, so it survives the ngrok tunnel a real Grid webhook + * arrives through — no polling, no client timers, and the panel shows the event at + * the moment Grid delivers it. + * + * Each frame carries an `id:` (a per-process sequence). EventSource echoes the last + * one back as `Last-Event-ID` when it reconnects, and anything that landed during + * the gap is replayed — so a blip doesn't silently drop deliveries. */ +/** + * Refuse a subscription initiated by ANOTHER site. Browsers set Sec-Fetch-Site on + * EventSource requests, so a page on evil.example can't quietly attach to a panel + * running on your machine. It is deliberately narrow: only an explicit + * `cross-site` is rejected, so `curl -N` and the verification scripts (which send + * no such header) still work. + * + * This is NOT access control. The stream carries whatever Grid delivered — + * transaction ids, amounts, and `counterpartyInformation`, which per Grid's schema + * can include a counterparty's name, birth date and nationality — and anyone who + * can reach the host can read it. Behind a tunnel, that's anyone with the URL. + * Gating it properly (a session cookie, or not exposing it beyond localhost) is a + * pre-prod requirement, tracked alongside proxy customer-scoping. + */ +function crossSite(req: Request): boolean { + return req.headers.get('sec-fetch-site') === 'cross-site'; +} + export async function GET(req: Request) { + if (crossSite(req)) { + return new Response('cross-site subscription refused', { status: 403 }); + } const encoder = new TextEncoder(); let unsubscribe: (() => void) | undefined; let keepalive: ReturnType | undefined; @@ -21,14 +48,41 @@ export async function GET(req: Request) { start(controller) { const send = (event: WebhookEvent) => { try { - controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); + controller.enqueue( + encoder.encode(`id: ${event.seq}\ndata: ${JSON.stringify(event)}\n\n`), + ); } catch { // Client vanished between the push and the enqueue — drop it. } }; - // An opening comment flushes headers so EventSource fires `open`. - controller.enqueue(encoder.encode(': connected\n\n')); - unsubscribe = subscribe(send); + // An opening comment flushes headers so EventSource fires `open`. The count + // is a comment (ignored by EventSource) and shows in `curl -N` when checking + // that every panel is attached. + controller.enqueue(encoder.encode(`: connected clients=${listenerCount() + 1}\n\n`)); + + // Catch-up must neither GAP nor DUPLICATE, which rules out both naive + // orderings: replay-then-subscribe loses whatever lands in between, and + // subscribe-then-replay sends that same event twice (live, then again from + // the backlog). So: subscribe immediately but HOLD live events, replay the + // backlog, then flush the held ones that the replay didn't already cover. + let holding = true; + const held: WebhookEvent[] = []; + unsubscribe = subscribe((event) => { + if (holding) held.push(event); + else send(event); + }); + const lastSeen = Number.parseInt(req.headers.get('last-event-id') ?? '', 10); + let replayedThrough = Number.isFinite(lastSeen) ? lastSeen : 0; + if (Number.isFinite(lastSeen)) { + for (const missed of eventsSince(lastSeen)) { + send(missed); + replayedThrough = missed.seq; + } + } + holding = false; + for (const event of held) { + if (event.seq > replayedThrough) send(event); + } keepalive = setInterval(() => { try { controller.enqueue(encoder.encode(': keepalive\n\n')); diff --git a/components/grid-wallet-prod/src/lib/webhookEvents.test.ts b/components/grid-wallet-prod/src/lib/webhookEvents.test.ts new file mode 100644 index 00000000..6147d0f6 --- /dev/null +++ b/components/grid-wallet-prod/src/lib/webhookEvents.test.ts @@ -0,0 +1,112 @@ +import { describe, it, expect, beforeEach, vi } from 'vitest'; +import { + pushEvent, + subscribe, + eventsSince, + listEvents, + listenerCount, + clearEvents, +} from './webhookEvents'; + +const delivery = (n: number) => ({ + id: `Webhook:${n}`, + type: 'INCOMING_PAYMENT.COMPLETED', + data: { n }, +}); +const latestSeq = () => listEvents().at(-1)?.seq ?? 0; + +describe('webhook bus', () => { + beforeEach(() => clearEvents()); + + // The demo drives ONE hardcoded customer, so every connected panel is watching + // the same account and must see every delivery — there is no filtering. + it('delivers each event to every subscriber', () => { + const a = vi.fn(); + const b = vi.fn(); + const c = vi.fn(); + const stop = [subscribe(a), subscribe(b), subscribe(c)]; + + pushEvent(delivery(1)); + + expect(a).toHaveBeenCalledTimes(1); + expect(b).toHaveBeenCalledTimes(1); + expect(c).toHaveBeenCalledTimes(1); + expect(a.mock.calls[0][0]).toMatchObject({ id: 'Webhook:1', data: { n: 1 } }); + stop.forEach((s) => s()); + }); + + it('keeps delivering to the rest after one subscriber leaves', () => { + const stays = vi.fn(); + const leaves = vi.fn(); + const stopStays = subscribe(stays); + subscribe(leaves)(); + + pushEvent(delivery(1)); + + expect(stays).toHaveBeenCalledTimes(1); + expect(leaves).not.toHaveBeenCalled(); + stopStays(); + }); + + // A subscriber that throws must not stop the others, or the 200 back to Grid. + it('survives a broken subscriber', () => { + const after = vi.fn(); + const stop = [ + subscribe(() => { + throw new Error('boom'); + }), + subscribe(after), + ]; + + expect(() => pushEvent(delivery(1))).not.toThrow(); + expect(after).toHaveBeenCalledTimes(1); + stop.forEach((s) => s()); + }); + + it('counts live subscribers', () => { + const before = listenerCount(); + const stop = subscribe(() => {}); + expect(listenerCount()).toBe(before + 1); + stop(); + expect(listenerCount()).toBe(before); + }); + + describe('replay window', () => { + it('stamps each event with an increasing seq', () => { + pushEvent(delivery(1)); + pushEvent(delivery(2)); + const [first, second] = listEvents().slice(-2); + expect(second.seq).toBeGreaterThan(first.seq); + }); + + // What a reconnecting client asks for with Last-Event-ID. + it('returns only the events after the one a client last saw', () => { + pushEvent(delivery(1)); + const afterFirst = latestSeq(); + pushEvent(delivery(2)); + pushEvent(delivery(3)); + + expect(eventsSince(afterFirst).map((e) => (e.data as { n: number }).n)).toEqual([2, 3]); + }); + + it('returns nothing when the client is already current', () => { + pushEvent(delivery(1)); + expect(eventsSince(latestSeq())).toEqual([]); + }); + + // seq 0 is "I have seen nothing" — a client that presents it gets the buffer. + it('replays everything buffered for seq 0', () => { + pushEvent(delivery(1)); + pushEvent(delivery(2)); + expect(eventsSince(0)).toHaveLength(2); + }); + + it('drops the oldest beyond the ring, so a long-gone client loses the overflow', () => { + for (let n = 1; n <= 55; n++) pushEvent(delivery(n)); + const kept = listEvents(); + expect(kept).toHaveLength(50); + expect((kept[0].data as { n: number }).n).toBe(6); + expect(eventsSince(0)).toHaveLength(50); + }); + }); +}); diff --git a/components/grid-wallet-prod/src/lib/webhookEvents.ts b/components/grid-wallet-prod/src/lib/webhookEvents.ts index 1c91222c..3bbc1d5b 100644 --- a/components/grid-wallet-prod/src/lib/webhookEvents.ts +++ b/components/grid-wallet-prod/src/lib/webhookEvents.ts @@ -1,4 +1,7 @@ export interface WebhookEvent { + /** Monotonic per-process sequence — the SSE `id:` field, so a reconnecting + * client can ask for everything after the last one it saw. */ + seq: number; id?: string; type?: string; timestamp?: string; @@ -10,9 +13,15 @@ const RING = 50; /** * Live subscribers (the SSE stream at /api/webhooks/stream). Verified events are - * PUSHED to the panel as they arrive — nothing polls. In-process only, which is - * all the demo needs: the receiver and the stream run in the same Next server. A - * multi-instance deployment would need a shared channel (Redis pub/sub) instead. + * PUSHED to every connected client as they arrive — nothing polls. The demo drives + * ONE hardcoded customer, so every connected panel is looking at the same account + * and every one of them should see every delivery; there is deliberately no + * per-client filtering. + * + * In-process only, which is all the demo needs: the receiver and the stream run in + * the same Next server. A multi-instance deployment would need a shared channel + * (Redis pub/sub) — with replicas, only the instance that received a delivery can + * fan it out to its own clients. */ type Listener = (event: WebhookEvent) => void; @@ -25,11 +34,13 @@ type Listener = (event: WebhookEvent) => void; interface WebhookBus { events: WebhookEvent[]; listeners: Set; + seq: number; } const globalBus = globalThis as typeof globalThis & { __gridWebhookBus?: WebhookBus }; const bus: WebhookBus = (globalBus.__gridWebhookBus ??= { events: [], listeners: new Set(), + seq: 0, }); export function subscribe(listener: Listener): () => void { @@ -39,9 +50,26 @@ export function subscribe(listener: Listener): () => void { }; } +/** How many clients are currently listening (surfaced for diagnostics). */ +export function listenerCount(): number { + return bus.listeners.size; +} + +/** + * Deliveries this client missed. EventSource reconnects on its own after a network + * blip and sends `Last-Event-ID`; without a replay, anything that landed during the + * gap would be lost silently. Empty when the client has seen everything (and for a + * fresh connection, which sends no id — a new panel starts from now rather than + * replaying history as if it just arrived). + */ +export function eventsSince(lastSeq: number): WebhookEvent[] { + return bus.events.filter((e) => e.seq > lastSeq); +} + export function pushEvent(raw: unknown): void { const e = raw as { id?: string; type?: string; timestamp?: string; data?: unknown }; const event: WebhookEvent = { + seq: ++bus.seq, id: e.id, type: e.type, timestamp: e.timestamp, @@ -50,7 +78,8 @@ export function pushEvent(raw: unknown): void { }; bus.events.push(event); if (bus.events.length > RING) bus.events = bus.events.slice(-RING); - // One broken subscriber must not stop the others, or the 200 back to Grid. + // Every connected client gets it. One broken subscriber must not stop the + // others, or the 200 back to Grid. bus.listeners.forEach((listener) => { try { listener(event);