v1.4.0: per-player snapshot + shared market view with owner info (ISS-112)

app/api/events/route.test.ts

@@ -2,8 +2,11 @@ import fs from 'node:fs/promises';
 import os from 'node:os';
 import path from 'node:path';
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';
-import { createSession } from '@/lib/auth';
+import { createSession, hashPassword } from '@/lib/auth';
+import * as schema from '@/lib/db/schema';
 import { getGameDb } from '@/lib/db/client';
+import { getPlayer } from '@/lib/db/localPlayer';
+import { tickNow } from '@/lib/sim/engine';
 import { GET } from './route';
 
 let dir: string;
@@ -49,8 +52,63 @@ describe('GET /api/events', () => {
       expect(text.startsWith('data: ')).toBe(true);
       const snap = JSON.parse(text.slice('data: '.length)) as Record<string, unknown>;
       expect(typeof (snap.simState as { clockMs: number }).clockMs).toBe('number');
+
+      // v1.4.0 (ISS-112): the connect frame comes from the SAME shared projection as GET /api/state —
+      // planes carry owner info and packages carry carrierName, so pushed frames can never drift.
+      const planes = snap.planes as Array<Record<string, unknown>>;
+      expect(planes.length).toBeGreaterThanOrEqual(1);
+      for (const key of ['ownerId', 'ownerName']) {
+        expect(planes[0], `plane.${key}`).toHaveProperty(key);
+      }
+      const pkgs = snap.packages as Array<Record<string, unknown>>;
+      for (const p of pkgs) expect(p, 'package.carrierName').toHaveProperty('carrierName');
     } finally {
       await reader.cancel(); // → stream cancel → unsubscribe from the registry
     }
   });
+
+  it('pushes PER-PLAYER frames on tick — each stream gets its own account row (v1.4.0, ISS-112)', async () => {
+    const inst = await getGameDb();
+    // A second player with a distinct economy + their own session (rows minted like /api/register).
+    const idB = 'test-player-events-b';
+    await inst.db.insert(schema.users).values({
+      id: idB, username: 'e2e-events-b', passwordHash: hashPassword('unused'), admin: false, createdAt: Date.now(),
+    });
+    await inst.db.insert(schema.players).values({
+      id: idB, provider: 'local', username: 'e2e-events-b', cash: 77_777.77, reputation: 3,
+    });
+    const sidB = await createSession(inst, idB);
+    const localCash = (await getPlayer(inst, 'local')).cash;
+
+    // Two live streams: the seeded local player and player B — both subscribed under their own ids.
+    const resA = await GET(new Request(`http://localhost/api/events?session=${sessionId}`));
+    const readerA = (resA.body as ReadableStream<Uint8Array>).getReader();
+    const resB = await GET(new Request(`http://localhost/api/events?session=${sidB}`));
+    const readerB = (resB.body as ReadableStream<Uint8Array>).getReader();
+
+    // Reads the next `data:` event off a stream and parses its snapshot.
+    const readFrame = async (reader: ReadableStreamDefaultReader<Uint8Array>) => {
+      const { value } = await reader.read();
+      return JSON.parse(new TextDecoder().decode(value).slice('data: '.length)) as Record<string, unknown>;
+    };
+
+    try {
+      // Drain both connect frames (each already carries its own account row)…
+      const connectA = await readFrame(readerA);
+      const connectB = await readFrame(readerB);
+      expect((connectA.account as { cash: number }).cash).toBeCloseTo(localCash, 9);
+      expect((connectB.account as { cash: number }).cash).toBeCloseTo(77_777.77, 9);
+
+      // …then one tick — the broadcast projects per connected player id and fans each frame out to its group only.
+      await tickNow();
+      const pushA = await readFrame(readerA);
+      const pushB = await readFrame(readerB);
+
+      expect((pushA.account as { cash: number }).cash).toBeCloseTo(localCash, 9); // A gets LOCAL's row…
+      expect((pushB.account as { cash: number }).cash).toBeCloseTo(77_777.77, 9); // …B gets ITS OWN — never crossed over
+    } finally {
+      await readerA.cancel(); // → stream cancel → unsubscribe from the registry
+      await readerB.cancel();
+    }
+  });
 });

app/api/events/route.ts

@@ -10,7 +10,10 @@ import { subscribe } from '@/lib/sse';
 // EventSource cannot set headers, so the session comes from X-Flight-Session OR ?session=
 // (requireAuthSse). ponytail: validated at CONNECT only — a session deleted/expired mid-stream keeps
 // receiving pushes until it reconnects (private infra, 7-day sessions; per-push re-validation would
-// be one indexed lookup per tick if that ever matters).
+// be one indexed lookup per tick if that ever matters). v1.4.0 (ISS-112): frames are PER PLAYER — the
+// route subscribes under THIS session's player id and the engine projects one snapshot per connected
+// player, pushing each frame only to its group; connect + pushed frames share one projection
+// (getStateSnapshot + sseFrame), so they can never drift.
 
 const encoder = new TextEncoder();
 const HEARTBEAT_MS = 15_000;
@@ -30,12 +33,12 @@ export async function GET(req: Request) {
     start(controller) {
       // Snapshot FIRST, then subscribe — a broadcast can never land before the initial event, so
       // events are strictly ordered by sim clock. A tick in the gap just means the client waits ≤1 s
-      // for the next push instead of seeing an out-of-order frame. v1.3.0 (ISS-110): the connect
-      // frame carries THIS session's account row; later pushes come from the shared broadcast
-      // (local player) until per-player frames land (#112).
+      // for the next push instead of seeing an out-of-order frame. v1.4.0 (ISS-112): connect frame and
+      // every later push carry THIS session's player projection — subscribe keys on auth.player.id so
+      // the tick broadcast only sends this client frames projected for its own account row.
       getStateSnapshot(auth.player.id)
         .then((snapshot) => {
-          unsubscribe = subscribe((text) => {
+          unsubscribe = subscribe(auth.player.id, (text) => {
             try {
               // broadcast carries the bare snapshot JSON — wrap it in an SSE frame here (the route owns
               // wire framing, matching the initial snapshot + heartbeat above; lib/sse stays generic).