push updated state to clients over SSE (ISS-129)

lib/db/client.ts

@@ -211,12 +211,14 @@ export async function persist(instance: GameDb): Promise<void> {
   await fs.rename(tmp, instance.file);
 }
 
-// Singleton for the app process; openGameDb() is the uncached variant (tests).
-let ready: Promise<GameDb> | null = null;
+// Singleton for the app process; openGameDb() is the uncached variant (tests). ISS-129 (v0.57.0):
+// lives on globalThis because Next.js can load this module more than once per process —
+// instrumentation.ts and route handlers get separate module instances, so a plain `let` would give
+// each its own in-memory sql.js DB (both persisting to the same file). One world per process.
+const g = globalThis as unknown as { __flightGameDb?: Promise<GameDb> };
 
 export function getGameDb(): Promise<GameDb> {
-  if (!ready) ready = open(defaultDbPath());
-  return ready;
+  return (g.__flightGameDb ??= open(defaultDbPath()));
 }
 
 export async function openGameDb(file: string): Promise<GameDb> {

lib/openapi.test.ts

@@ -3,9 +3,10 @@ import { describe, expect, it } from 'vitest';
 import { openApiSpec } from './openapi';
 
 describe('OpenAPI spec', () => {
-  it('documents exactly the twelve public endpoints (3 read + 9 player actions)', () => {
+  it('documents exactly the thirteen public endpoints (4 read + 9 player actions)', () => {
     expect(Object.keys(openApiSpec.paths).sort()).toEqual([
       '/api/airports',
+      '/api/events', // v0.57.0 (ISS-129) — SSE state stream
       '/api/market/buy-plane',
       '/api/market/sell-plane',
       '/api/packages/{id}/abandon',

lib/openapi.ts

@@ -51,6 +51,22 @@ export const openApiSpec = {
         },
       },
     },
+    '/api/events': {
+      get: {
+        summary: 'Live state stream (SSE)',
+        description:
+          'Server-Sent Events feed of full game-state snapshots (v0.57.0, ISS-129) — the current snapshot is sent immediately on connect and a fresh one after every sim tick (~1 s), so clients can subscribe instead of polling /api/state. Each event\'s data payload is JSON with exactly the StateSnapshot shape; comment lines starting with ":" are heartbeats.',
+        operationId: 'subscribeToEvents',
+        responses: {
+          200: {
+            description: 'SSE stream of state snapshots (text/event-stream)',
+            content: {
+              'text/event-stream': { schema: { type: 'string' } },
+            },
+          },
+        },
+      },
+    },
     '/api/version': {
       get: {
         summary: 'App version',

lib/sim/engine.ts

@@ -7,7 +7,8 @@ import { haversineKm, initialBearingDeg } from '@/lib/geo';
 import { weatherFor } from '@/lib/weather';
 import { FUEL_DISCOUNT_PER_REP, FUEL_PRICE_PER_L, MAX_FUEL_DISCOUNT, PLANE_TYPES, REPAIR_DURATION_MS, repRate, WEAR_PER_KM } from '@/lib/game/catalog';
 import { SPECIALS } from '@/lib/game/specials';
-import { parseIcaoQueue } from '@/lib/game/state';
+import { getStateSnapshot, parseIcaoQueue } from '@/lib/game/state';
+import { broadcast } from '@/lib/sse';
 import { IDLE_WORLD_TIME_SCALE } from '@/lib/worldClock';
 import {
   generatePackage,
@@ -46,13 +47,15 @@ export function reputationGainFor(km: number): number {
   return Math.round((REP_BASE + REP_PER_KM * km) * 100) / 100;
 }
 
-// Serialize ticks across concurrent requests (Next.js route handlers run in the
-// same process): each tick re-reads sim_state after the previous one wrote it, so
-// no advance is ever applied twice.
-let chain: Promise<unknown> = Promise.resolve();
+// Serialize ticks across concurrent requests (Next.js route handlers run in the same process): each
+// tick re-reads sim_state after the previous one wrote it, so no advance is ever applied twice.
+// ISS-129 (v0.57.0): the chain lives on globalThis — instrumentation.ts and route handlers get
+// separate module instances from Next.js, so a plain `let` would give each layer its own lock and let
+// their ticks interleave against the shared DB below.
+const g = globalThis as unknown as { __flightTickChain?: Promise<unknown> };
 function withLock<T>(fn: () => Promise<T>): Promise<T> {
-  const run = chain.then(fn);
-  chain = run.then(
+  const run = (g.__flightTickChain ??= Promise.resolve()).then(fn);
+  g.__flightTickChain = run.then(
     () => undefined,
     () => undefined,
   );
@@ -416,8 +419,24 @@ export async function spawnCoveragePackages(
 
 // --- Public API (singleton DB + lock) — call from every read/action endpoint ---
 
+// ISS-129 (v0.57.0): after each tick on the singleton path, push a fresh snapshot to all SSE
+// subscribers (/api/events) — the exact same body GET /api/state returns (getStateSnapshot), so
+// pushed and polled states can never diverge in shape. Failures are logged, not fatal: polling
+// still works either way. Background ticks (~1 s) cover action mutations for other clients too.
+async function broadcastState(): Promise<void> {
+  try {
+    broadcast(JSON.stringify(await getStateSnapshot()));
+  } catch (err) {
+    console.error('flight: state broadcast failed', err);
+  }
+}
+
 export function tickNow(): Promise<schema.SimStateRow> {
-  return withLock(async () => tickOnce(await getGameDb()));
+  return withLock(async () => {
+    const row = await tickOnce(await getGameDb());
+    await broadcastState();
+    return row;
+  });
 }
 
 // ISS-117: the world clock runs even with nobody polling — a ~1 s interval drives the same locked

lib/sse.test.ts

@@ -0,0 +1,45 @@
+import { describe, expect, it } from 'vitest';
+import { broadcast, subscribe } from './sse';
+
+// ISS-129 (v0.57.0): the fan-out registry behind /api/events — plain Set semantics, so no DB or
+// server needed. Subscribers leak across tests in this file (module-level state), which is fine:
+// each test only asserts on its own arrays.
+
+describe('lib/sse', () => {
+  it('delivers broadcasts to subscribers and stops after unsubscribe', () => {
+    const a: string[] = [];
+    const b: string[] = [];
+    const unsubA = subscribe((t) => a.push(t));
+    const unsubB = subscribe((t) => b.push(t));
+
+    broadcast('x');
+    expect(a).toEqual(['x']);
+    expect(b).toEqual(['x']);
+
+    unsubA();
+    broadcast('y');
+    expect(a).toEqual(['x']); // unsubscribed — no more events
+    expect(b).toEqual(['x', 'y']);
+    unsubB();
+  });
+
+  it('a throwing subscriber does not break the others', () => {
+    const got: string[] = [];
+    subscribe(() => {
+      throw new Error('boom');
+    });
+    const unsub = subscribe((t) => got.push(t));
+    broadcast('z');
+    expect(got).toEqual(['z']);
+    unsub();
+  });
+
+  it('unsubscribe is idempotent', () => {
+    const got: string[] = [];
+    const unsub = subscribe((t) => got.push(t));
+    unsub();
+    unsub(); // second call must not throw
+    broadcast('w');
+    expect(got).toEqual([]);
+  });
+});

lib/sse.ts

@@ -0,0 +1,29 @@
+// ISS-129 (v0.57.0): in-process fan-out registry for the SSE state stream (/api/events). The app
+// runs as one long-lived server process (`next start`; instrumentation.ts already keeps background
+// sim state alive) — no new dependencies, no WS upgrade needed. The Set lives on globalThis because
+// Next.js gives instrumentation.ts (where broadcast() fires from the background tick) and route
+// handlers (where subscribe() happens in /api/events) separate module instances; a plain `let` would
+// give each its own empty set and no push would ever be delivered.
+
+type Push = (text: string) => void;
+
+const g = globalThis as unknown as { __flightSseSubscribers?: Set<Push> };
+const subscribers: Set<Push> = (g.__flightSseSubscribers ??= new Set());
+
+export function subscribe(fn: Push): () => void {
+  subscribers.add(fn);
+  return () => {
+    subscribers.delete(fn);
+  };
+}
+
+// Call each subscriber safely — one dead stream must not take the others down.
+export function broadcast(text: string): void {
+  for (const fn of [...subscribers]) {
+    try {
+      fn(text);
+    } catch {
+      // subscriber's stream is gone; its own cleanup unsubscribes it
+    }
+  }
+}

lib/useGameState.ts

@@ -35,8 +35,14 @@ export async function fetchState(): Promise<StateSnapshot> {
   return snapshot;
 }
 
-// The shell's single polling source (v0.9.0): one GET /api/state loop for the whole page,
-// so every pane renders from the same snapshot instead of running its own poll loops.
+// The shell's single state source (v0.9.0): one snapshot feed for the whole page, so every pane
+// renders from the same snapshot instead of running its own poll loops. ISS-129 (v0.57.0): primary
+// transport is SSE — /api/events pushes a full snapshot after every sim tick; polling stays as an
+// automatic fallback (no event within 5 s of connect, or on stream error) and stops again when fresh
+// events flow. EventSource reconnects natively, so there is no custom reconnect logic here.
+const SSE_URL = '/api/events';
+const SSE_SILENCE_MS = 5_000;
+
 export function useGameState(intervalMs = 1000): GameStateResult {
   const [state, setState] = useState<StateSnapshot | null>(null);
   const [loading, setLoading] = useState(true);
@@ -70,10 +76,53 @@ export function useGameState(intervalMs = 1000): GameStateResult {
   }, []);
 
   useEffect(() => {
-    refresh(); // fetch immediately on mount, don't wait for the first interval tick
-    const id = setInterval(refresh, intervalMs);
-    return () => clearInterval(id);
-  }, [intervalMs, refresh]);
+    let pollId: ReturnType<typeof setInterval> | null = null;
+    let silenceTimer: ReturnType<typeof setTimeout> | null = null;
+    let closed = false;
+
+    const stopPolling = () => {
+      if (pollId !== null) clearInterval(pollId);
+      pollId = null;
+    };
+    // Fallback loop: immediate catch-up fetch, then the old interval.
+    const startPolling = () => {
+      if (closed || pollId !== null) return;
+      refresh();
+      pollId = setInterval(refresh, intervalMs);
+    };
+
+    let source: EventSource | null = null;
+    try {
+      source = new EventSource(SSE_URL);
+    } catch {
+      source = null; // no EventSource support — pure polling below
+    }
+
+    if (source !== null) {
+      source.onmessage = (event) => {
+        try {
+          applySnapshot(JSON.parse(event.data));
+        } catch {
+          // malformed event — the next one overwrites it anyway
+        }
+        stopPolling(); // fresh push flow → polling no longer needed
+        if (silenceTimer !== null) clearTimeout(silenceTimer); // connect check served its purpose
+      };
+      source.onerror = () => startPolling(); // stream trouble → poll until pushes resume
+      silenceTimer = setTimeout(startPolling, SSE_SILENCE_MS); // connected but silent for 5 s → poll
+    } else {
+      startPolling();
+    }
+
+    refresh(); // mount: same immediate fetch as before (fast first paint + __flightLastSnapshot)
+
+    return () => {
+      closed = true;
+      source?.close();
+      stopPolling();
+      if (silenceTimer !== null) clearTimeout(silenceTimer);
+    };
+  }, [intervalMs, refresh, applySnapshot]);
 
   return { state, loading, error, refresh, applySnapshot };
 }