Fix: Live Ops events stuck at 0 on cold SSE Lambda

app/api/events/activity/route.test.ts

@@ -0,0 +1,65 @@
+import { NextRequest } from 'next/server';
+import { beforeEach, describe, expect, it } from 'vitest';
+import { GET } from './route';
+
+const DEALS_GLOBAL_KEY = Symbol.for('super.routers.deals');
+const LOG_GLOBAL_KEY = Symbol.for('super.activity.events.log');
+const BROADCASTER_GLOBAL_KEY = Symbol.for('super.activity.events.broadcaster');
+
+type GlobalRecord = Record<symbol, unknown> & typeof globalThis;
+
+// Simulates the Vercel split-Lambda case: the SSE function cold-starts on an
+// instance that has never handled a tRPC call, so `deals.ts`'s top-level
+// `ensureSeeded()` side effect has never run on this instance.
+function simulateColdSseLambda(): void {
+  const g = globalThis as GlobalRecord;
+  delete g[DEALS_GLOBAL_KEY];
+  delete g[LOG_GLOBAL_KEY];
+  delete g[BROADCASTER_GLOBAL_KEY];
+}
+
+async function readFirstFrames(response: Response, ms: number): Promise<string> {
+  const reader = response.body!.getReader();
+  const decoder = new TextDecoder();
+  const chunks: string[] = [];
+  const deadline = Date.now() + ms;
+  try {
+    while (Date.now() < deadline) {
+      const timeout = new Promise<{ done: true; value: undefined }>((resolve) =>
+        setTimeout(() => resolve({ done: true, value: undefined }), deadline - Date.now()),
+      );
+      const { done, value } = (await Promise.race([reader.read(), timeout])) as ReadableStreamReadResult<Uint8Array>;
+      if (done) break;
+      chunks.push(decoder.decode(value, { stream: true }));
+    }
+  } finally {
+    await reader.cancel().catch(() => undefined);
+  }
+  return chunks.join('');
+}
+
+beforeEach(() => {
+  simulateColdSseLambda();
+});
+
+describe('GET /api/events/activity', () => {
+  it('replays seed events even when the Lambda cold-starts without tRPC', async () => {
+    const request = new NextRequest('http://localhost/api/events/activity');
+    const response = await GET(request);
+    const payload = await readFirstFrames(response, 250);
+
+    const snapshotMatch = payload.match(/event: snapshot\nid: (\d+)\ndata: (\{[^\n]+\})/);
+    expect(snapshotMatch, `no snapshot frame in: ${payload.slice(0, 200)}`).not.toBeNull();
+    const snapshot = JSON.parse(snapshotMatch![2]) as {
+      latest_seq: number;
+      replayed_count: number;
+    };
+
+    // Before the fix: replayed_count is 0 because the SSE Lambda never imported
+    // deals.ts, so ensureSeeded() never ran on this instance. The client sees
+    // an empty stream and the Live Ops counter stays stuck at zero.
+    expect(snapshot.replayed_count).toBeGreaterThan(0);
+    expect(snapshot.latest_seq).toBeGreaterThan(0);
+    expect(payload).toContain('event: activity');
+  });
+});

app/api/events/activity/route.ts

@@ -1,4 +1,5 @@
 import { NextRequest } from 'next/server';
+import { ensureSeeded } from '@/server/activity/state';
 import { getEventsSince, getLatestSeq } from '@/server/events/log';
 import { subscribe } from '@/server/events/broadcaster';
 import type { ActivityEvent } from '@/server/events/types';
@@ -27,6 +28,12 @@ function sseFrame(event: string | null, data: string, id?: number): string {
 }
 
 export async function GET(request: NextRequest) {
+  // Vercel routes SSE, tRPC, and webhook traffic to separate Lambdas, each
+  // with their own in-memory state. A cold SSE Lambda would otherwise send
+  // `replayed_count: 0` and strand the client on an empty snapshot. The seed
+  // call is idempotent.
+  ensureSeeded();
+
   const encoder = new TextEncoder();
   const lastEventId = parseLastEventId(request.headers.get('last-event-id'));
 

src/server/activity/state.ts

@@ -0,0 +1,90 @@
+import { __resetActivityEventLog, appendEvent, getEventsForOffer } from '../events/log';
+import { projectOfferFromEvents } from '../events/projection';
+import type { Actor, OfferSnapshot, ScenarioKey } from '../events/types';
+import scenariosFixture from '../fixtures/scenarios.json';
+
+export const SCENARIO_KEYS = [
+  'tracking',
+  'needs_attention',
+  'in_progress',
+  'rewarded',
+] as const satisfies readonly ScenarioKey[];
+
+const SEED_ACTOR: Actor = { kind: 'system' };
+
+type FixtureEntry = Omit<OfferSnapshot, 'scenario_key'>;
+type FixtureShape = Record<ScenarioKey, FixtureEntry[]>;
+
+type DealsState = { store: Map<string, OfferSnapshot>; seeded: boolean };
+
+const DEALS_GLOBAL_KEY = Symbol.for('super.routers.deals');
+
+type GlobalWithDeals = typeof globalThis & { [DEALS_GLOBAL_KEY]?: DealsState };
+
+export function getDealsState(): DealsState {
+  const g = globalThis as GlobalWithDeals;
+  let existing = g[DEALS_GLOBAL_KEY];
+  if (!existing) {
+    existing = { store: new Map(), seeded: false };
+    g[DEALS_GLOBAL_KEY] = existing;
+  }
+  return existing;
+}
+
+function cloneFixture(): FixtureShape {
+  return JSON.parse(JSON.stringify(scenariosFixture)) as FixtureShape;
+}
+
+function seedIntoStore(store: Map<string, OfferSnapshot>): void {
+  const seed = cloneFixture();
+  for (const scenarioKey of SCENARIO_KEYS) {
+    for (const entry of seed[scenarioKey]) {
+      const snapshot: OfferSnapshot = { ...entry, scenario_key: scenarioKey };
+      appendEvent({
+        kind: 'offer_created',
+        actor: SEED_ACTOR,
+        offer_id: snapshot.id,
+        payload: snapshot,
+      });
+      store.set(snapshot.id, snapshot);
+    }
+  }
+}
+
+// Idempotent. On Vercel, the SSE, tRPC, and webhook routes each run as
+// separate Lambda functions with per-process `globalThis`; whichever handler
+// touches state first must seed it. Any module that needs seeded state should
+// import from here (the top-level call below guarantees it), or call this
+// directly.
+export function ensureSeeded(): void {
+  const state = getDealsState();
+  if (state.seeded) return;
+  seedIntoStore(state.store);
+  state.seeded = true;
+}
+
+export function __resetActivityStore(): void {
+  const state = getDealsState();
+  state.store.clear();
+  __resetActivityEventLog();
+  seedIntoStore(state.store);
+  state.seeded = true;
+}
+
+export function reprojectOffer(offerId: string): OfferSnapshot | null {
+  const { store } = getDealsState();
+  const events = getEventsForOffer(offerId);
+  const snapshot = projectOfferFromEvents(events);
+  if (snapshot) {
+    store.set(offerId, snapshot);
+  } else {
+    store.delete(offerId);
+  }
+  return snapshot;
+}
+
+export function getOfferSnapshot(offerId: string): OfferSnapshot | null {
+  return getDealsState().store.get(offerId) ?? null;
+}
+
+ensureSeeded();

src/server/routers/deals.ts

@@ -3,15 +3,15 @@ import { randomUUID } from 'node:crypto';
 import { z } from 'zod';
 import { publicProcedure, router } from '../trpc';
 import type { Scenario } from '../context';
-import scenariosFixture from '../fixtures/scenarios.json';
 import {
-  __resetActivityEventLog,
-  appendEvent,
-  getEventsForOffer,
-  getEventsSince,
-  getLatestSeq,
-} from '../events/log';
-import { projectOfferFromEvents } from '../events/projection';
+  __resetActivityStore,
+  SCENARIO_KEYS,
+  ensureSeeded,
+  getDealsState,
+  getOfferSnapshot,
+  reprojectOffer,
+} from '../activity/state';
+import { appendEvent, getEventsSince, getLatestSeq } from '../events/log';
 import { buildOffer, type WireOffer } from '../events/buildOffer';
 import type {
   ActivityEvent,
@@ -25,6 +25,8 @@ import { PARTNER_EVENT_NAMES } from '../webhooks/translate';
 import { findCachedInvestigation } from '../ai/cache';
 import { isDemoMode } from '../ai/anthropic';
 
+export { __resetActivityStore, getOfferSnapshot, reprojectOffer };
+
 const eventSchema = z.object({
   id: z.string(),
   name: z.string(),
@@ -85,86 +87,9 @@ const responseSchema = z.object({
 });
 
 const BUCKETS = ['pending', 'in_progress', 'completed'] as const satisfies readonly Bucket[];
-const SCENARIO_KEYS = [
-  'tracking',
-  'needs_attention',
-  'in_progress',
-  'rewarded',
-] as const satisfies readonly ScenarioKey[];
-
-type FixtureEntry = Omit<OfferSnapshot, 'scenario_key'>;
-type FixtureShape = Record<ScenarioKey, FixtureEntry[]>;
-
-function cloneFixture(): FixtureShape {
-  return JSON.parse(JSON.stringify(scenariosFixture)) as FixtureShape;
-}
-
-type DealsState = { store: Map<string, OfferSnapshot>; seeded: boolean };
-
-const DEALS_GLOBAL_KEY = Symbol.for('super.routers.deals');
-
-type GlobalWithDeals = typeof globalThis & { [DEALS_GLOBAL_KEY]?: DealsState };
 
-function getDealsState(): DealsState {
-  const g = globalThis as GlobalWithDeals;
-  let existing = g[DEALS_GLOBAL_KEY];
-  if (!existing) {
-    existing = { store: new Map(), seeded: false };
-    g[DEALS_GLOBAL_KEY] = existing;
-  }
-  return existing;
-}
-
-const SEED_ACTOR: Actor = { kind: 'system' };
 const USER_ACTOR: Actor = { kind: 'user', id: 'demo-user' };
 
-function seedIntoStore(store: Map<string, OfferSnapshot>): void {
-  const seed = cloneFixture();
-  for (const scenarioKey of SCENARIO_KEYS) {
-    for (const entry of seed[scenarioKey]) {
-      const snapshot: OfferSnapshot = { ...entry, scenario_key: scenarioKey };
-      appendEvent({
-        kind: 'offer_created',
-        actor: SEED_ACTOR,
-        offer_id: snapshot.id,
-        payload: snapshot,
-      });
-      store.set(snapshot.id, snapshot);
-    }
-  }
-}
-
-export function reprojectOffer(offerId: string): OfferSnapshot | null {
-  const { store } = getDealsState();
-  const events = getEventsForOffer(offerId);
-  const snapshot = projectOfferFromEvents(events);
-  if (snapshot) {
-    store.set(offerId, snapshot);
-  } else {
-    store.delete(offerId);
-  }
-  return snapshot;
-}
-
-export function getOfferSnapshot(offerId: string): OfferSnapshot | null {
-  return getDealsState().store.get(offerId) ?? null;
-}
-
-export function __resetActivityStore(): void {
-  const state = getDealsState();
-  state.store.clear();
-  __resetActivityEventLog();
-  seedIntoStore(state.store);
-  state.seeded = true;
-}
-
-function ensureSeeded(): void {
-  const state = getDealsState();
-  if (state.seeded) return;
-  seedIntoStore(state.store);
-  state.seeded = true;
-}
-
 ensureSeeded();
 
 function keysForScenario(scenario: Scenario): ScenarioKey[] {