fix(grid-wallet-prod): replay missed webhooks to reconnecting SSE clients

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<typeof setInterval> | 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'));

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

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<Listener>;
+  seq: number;
 }
 const globalBus = globalThis as typeof globalThis & { __gridWebhookBus?: WebhookBus };
 const bus: WebhookBus = (globalBus.__gridWebhookBus ??= {
   events: [],
   listeners: new Set<Listener>(),
+  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);