diff --git a/app/api/events/activity/route.test.ts b/app/api/events/activity/route.test.ts new file mode 100644 index 0000000..1f734a2 --- /dev/null +++ b/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 & 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 { + 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; + 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'); + }); +}); diff --git a/app/api/events/activity/route.ts b/app/api/events/activity/route.ts index b3fefca..48739ba 100644 --- a/app/api/events/activity/route.ts +++ b/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')); diff --git a/src/server/activity/state.ts b/src/server/activity/state.ts new file mode 100644 index 0000000..aedf637 --- /dev/null +++ b/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; +type FixtureShape = Record; + +type DealsState = { store: Map; 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): 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(); diff --git a/src/server/routers/deals.ts b/src/server/routers/deals.ts index fad140a..1ab1b18 100644 --- a/src/server/routers/deals.ts +++ b/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; -type FixtureShape = Record; - -function cloneFixture(): FixtureShape { - return JSON.parse(JSON.stringify(scenariosFixture)) as FixtureShape; -} - -type DealsState = { store: Map; 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): 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[] {