From a1a6f5defb2135a86c87335c9417aedd2c9fcbae Mon Sep 17 00:00:00 2001 From: peng Date: Fri, 24 Jul 2026 20:51:09 -0700 Subject: [PATCH 1/2] fix(grid-wallet-prod): replay missed webhooks to reconnecting SSE clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every connected panel already received every delivery — the demo drives one hardcoded customer, so there is no per-client filtering and pushEvent walks all subscribers. Verified with three concurrent clients: all three got the event, and one disconnecting left the others streaming. What was missing is replay. A client only saw events that arrived while it was connected, so an EventSource auto-reconnect after a network blip silently dropped anything that landed in the gap. Each frame now carries an `id:` (a per-process sequence). EventSource echoes it back as `Last-Event-ID` on reconnect, and the stream replays events after that point. A FRESH client replays nothing: it has no id, and rendering old deliveries would date them to the moment of receipt. 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 the stream subscribes immediately but HOLDS live events, replays the backlog, then flushes the held ones the replay didn't already cover. Verified by reconnecting while deliveries stream in: no gaps, no duplicates. The connect handshake comment now reports `clients=N` — invisible to EventSource, visible in `curl -N`, useful when checking that every panel is attached. Tests cover the bus contract: fan-out to every subscriber, survival of a broken one, the seq/eventsSince replay windows, and the ring dropping the overflow a long-gone client can no longer catch up on. Limits unchanged and documented in the module: the bus is per server process (with replicas, only the instance that received a delivery fans it out), and the ring buffer holds the last 50 events. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Pjqydq34z8DBK1kBrioXK2 --- .../src/app/api/webhooks/stream/route.ts | 49 ++++++-- .../src/lib/webhookEvents.test.ts | 112 ++++++++++++++++++ .../grid-wallet-prod/src/lib/webhookEvents.ts | 37 +++++- 3 files changed, 186 insertions(+), 12 deletions(-) create mode 100644 components/grid-wallet-prod/src/lib/webhookEvents.test.ts 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..47327dc3 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,9 +8,15 @@ 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. */ export async function GET(req: Request) { const encoder = new TextEncoder(); @@ -21,14 +27,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); From 63fb4ad64b3dbd75670256c6bd9fc96bf8f1765f Mon Sep 17 00:00:00 2001 From: peng Date: Fri, 24 Jul 2026 20:57:14 -0700 Subject: [PATCH 2/2] fix(grid-wallet-prod): refuse cross-site subscriptions to the webhook stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review follow-up. /api/webhooks/stream is unauthenticated and carries whatever Grid delivered — transaction ids, amounts, and counterpartyInformation, which per Grid's schema can include a counterparty's name, birth date and nationality. That is the same exposure the polling route was deleted for earlier in this stack; the SSE replacement inherited it. Reject an EventSource opened by another site (Sec-Fetch-Site: cross-site), so a page on someone else's origin can't attach to a panel running on your machine. Deliberately narrow: only an explicit cross-site is refused, so `curl -N` and the verification scripts keep working. This is NOT access control, and the comment says so: anyone who can reach the host can still read the stream, which behind a tunnel means anyone with the URL. Gating it properly (session cookie, or not exposing it past localhost) stays a pre-prod requirement alongside proxy customer-scoping. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Pjqydq34z8DBK1kBrioXK2 --- .../src/app/api/webhooks/stream/route.ts | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) 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 47327dc3..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 @@ -18,7 +18,28 @@ const KEEPALIVE_MS = 15_000; * 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;