5b07bb4e14
Wires the gateway's signed SubscribeEvents stream end-to-end:
- backend: emit game.turn.ready from lobby.OnRuntimeSnapshot on every
current_turn advance, addressed to every active membership, push-only
channel, idempotency key turn-ready:<game_id>:<turn>;
- ui: single EventStream singleton replaces revocation-watcher.ts and
carries both per-event dispatch and revocation detection; toast
primitive (store + host) lives in lib/; GameStateStore gains
pendingTurn/markPendingTurn/advanceToPending and a persisted
lastViewedTurn so a return after multiple turns surfaces the same
"view now" affordance as a live push event;
- mandatory event-signature verification through ui/core
(verifyPayloadHash + verifyEvent), full-jitter exponential backoff
1s -> 30s on transient failure, signOut("revoked") on
Unauthenticated or clean end-of-stream;
- catalog and migration accept the new kind; tests cover producer
(testcontainers + capturing publisher), consumer (Vitest event
stream, toast, game-state extensions), and a Playwright e2e
delivering a signed frame to the live UI.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
325 lines
8.4 KiB
TypeScript
325 lines
8.4 KiB
TypeScript
// Vitest coverage for the SubscribeEvents stream consumer in
|
|
// `src/api/events.svelte.ts`. The tests drive the singleton through
|
|
// its lifecycle with a `createRouterTransport` fake — the same
|
|
// pattern `galaxy-client.test.ts` uses for unary calls, extended to
|
|
// async-generator handlers for server-streaming RPCs.
|
|
//
|
|
// The session store is mocked so `signOut("revoked")` is observable
|
|
// without instantiating the real keystore/IndexedDB chain.
|
|
|
|
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
|
|
import { create } from "@bufbuild/protobuf";
|
|
import {
|
|
Code,
|
|
ConnectError,
|
|
createClient,
|
|
createRouterTransport,
|
|
} from "@connectrpc/connect";
|
|
import {
|
|
EdgeGateway,
|
|
GatewayEventSchema,
|
|
type GatewayEvent,
|
|
} from "../src/proto/galaxy/gateway/v1/edge_gateway_pb";
|
|
|
|
let sessionStatus: "anonymous" | "authenticated" = "anonymous";
|
|
const signOutSpy = vi.fn();
|
|
vi.mock("../src/lib/session-store.svelte", () => ({
|
|
session: {
|
|
get status(): string {
|
|
return sessionStatus;
|
|
},
|
|
signOut: (...args: unknown[]) => signOutSpy(...args),
|
|
},
|
|
}));
|
|
|
|
// The import must come after vi.mock so the module reads the mocked
|
|
// session reference.
|
|
const {
|
|
eventStream,
|
|
} = await import("../src/api/events.svelte");
|
|
|
|
import type { Core } from "../src/platform/core/index";
|
|
import type { DeviceKeypair } from "../src/platform/store/index";
|
|
|
|
beforeEach(() => {
|
|
eventStream.resetForTests();
|
|
signOutSpy.mockReset();
|
|
sessionStatus = "anonymous";
|
|
});
|
|
|
|
afterEach(() => {
|
|
eventStream.resetForTests();
|
|
});
|
|
|
|
function mockCore(overrides?: Partial<Core>): Core {
|
|
return {
|
|
signRequest: () => new Uint8Array([1, 2, 3]),
|
|
verifyResponse: () => true,
|
|
verifyEvent: () => true,
|
|
verifyPayloadHash: () => true,
|
|
driveEffective: () => 0,
|
|
emptyMass: () => 0,
|
|
weaponsBlockMass: () => 0,
|
|
fullMass: () => 0,
|
|
speed: () => 0,
|
|
cargoCapacity: () => 0,
|
|
carryingMass: () => 0,
|
|
blockUpgradeCost: () => 0,
|
|
...overrides,
|
|
} as Core;
|
|
}
|
|
|
|
function mockKeypair(): DeviceKeypair {
|
|
return {
|
|
publicKey: new Uint8Array(32),
|
|
sign: async () => new Uint8Array(64),
|
|
};
|
|
}
|
|
|
|
function buildEvent(eventType: string, payload: Uint8Array): GatewayEvent {
|
|
return create(GatewayEventSchema, {
|
|
eventType,
|
|
eventId: `event-${eventType}-${Math.random().toString(16).slice(2, 8)}`,
|
|
timestampMs: 1n,
|
|
payloadBytes: payload,
|
|
payloadHash: new Uint8Array(32).fill(0xaa),
|
|
signature: new Uint8Array(64).fill(0xbb),
|
|
requestId: "req-1",
|
|
traceId: "trace-1",
|
|
});
|
|
}
|
|
|
|
function makeRouter(
|
|
streamFactory: () => AsyncIterable<GatewayEvent>,
|
|
): ReturnType<typeof createClient<typeof EdgeGateway>> {
|
|
const transport = createRouterTransport(({ service }) => {
|
|
service(EdgeGateway, {
|
|
executeCommand() {
|
|
throw new Error("not used in this test");
|
|
},
|
|
async *subscribeEvents() {
|
|
for await (const e of streamFactory()) {
|
|
yield e;
|
|
}
|
|
},
|
|
});
|
|
});
|
|
return createClient(EdgeGateway, transport);
|
|
}
|
|
|
|
describe("EventStream", () => {
|
|
test("verified events reach the registered handler", async () => {
|
|
const handler = vi.fn();
|
|
eventStream.on("game.turn.ready", handler);
|
|
|
|
const event = buildEvent(
|
|
"game.turn.ready",
|
|
new TextEncoder().encode(JSON.stringify({ game_id: "g", turn: 2 })),
|
|
);
|
|
const client = makeRouter(async function* () {
|
|
yield event;
|
|
});
|
|
|
|
const sleep = vi.fn(async () => {});
|
|
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep,
|
|
random: () => 0,
|
|
});
|
|
|
|
await vi.waitFor(() => {
|
|
expect(handler).toHaveBeenCalled();
|
|
});
|
|
expect(handler).toHaveBeenCalledTimes(1);
|
|
expect(handler.mock.calls[0]?.[0].eventType).toBe("game.turn.ready");
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("handlers for other event types are not invoked", async () => {
|
|
const turnHandler = vi.fn();
|
|
const mailHandler = vi.fn();
|
|
eventStream.on("game.turn.ready", turnHandler);
|
|
eventStream.on("mail.received", mailHandler);
|
|
|
|
const event = buildEvent(
|
|
"game.turn.ready",
|
|
new TextEncoder().encode("{}"),
|
|
);
|
|
const client = makeRouter(async function* () {
|
|
yield event;
|
|
});
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: async () => {},
|
|
random: () => 0,
|
|
});
|
|
|
|
await vi.waitFor(() => {
|
|
expect(turnHandler).toHaveBeenCalled();
|
|
});
|
|
expect(mailHandler).not.toHaveBeenCalled();
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("unsubscribe removes the handler", async () => {
|
|
const handler = vi.fn();
|
|
const off = eventStream.on("game.turn.ready", handler);
|
|
off();
|
|
|
|
const event = buildEvent(
|
|
"game.turn.ready",
|
|
new TextEncoder().encode("{}"),
|
|
);
|
|
const client = makeRouter(async function* () {
|
|
yield event;
|
|
});
|
|
const sleepSpy = vi.fn(async () => {});
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: sleepSpy,
|
|
random: () => 0,
|
|
});
|
|
|
|
await vi.waitFor(() => {
|
|
// Stream finished — either status became idle, or the loop
|
|
// is at backoff after a clean close on an anonymous
|
|
// session (which goes straight to idle as well).
|
|
expect(eventStream.connectionStatus).toBe("idle");
|
|
});
|
|
expect(handler).not.toHaveBeenCalled();
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("bad signature tears down the stream and reconnects", async () => {
|
|
const handler = vi.fn();
|
|
eventStream.on("game.turn.ready", handler);
|
|
|
|
let verifyCalls = 0;
|
|
const core = mockCore({
|
|
verifyEvent: () => {
|
|
verifyCalls += 1;
|
|
return verifyCalls > 1; // first event fails, then passes
|
|
},
|
|
});
|
|
|
|
let streamCalls = 0;
|
|
const client = makeRouter(async function* () {
|
|
streamCalls += 1;
|
|
yield buildEvent(
|
|
"game.turn.ready",
|
|
new TextEncoder().encode("{}"),
|
|
);
|
|
});
|
|
|
|
const sleepCalls: number[] = [];
|
|
eventStream.start({
|
|
core,
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: async (ms) => {
|
|
sleepCalls.push(ms);
|
|
},
|
|
random: () => 0, // full-jitter = 0 → instant retry
|
|
});
|
|
|
|
await vi.waitFor(() => {
|
|
expect(handler).toHaveBeenCalled();
|
|
});
|
|
// Two stream openings: first one rejected on bad signature,
|
|
// second one delivered the good event.
|
|
expect(streamCalls).toBeGreaterThanOrEqual(2);
|
|
// Backoff was scheduled between attempts.
|
|
expect(sleepCalls.length).toBeGreaterThanOrEqual(1);
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("unauthenticated error signs the session out", async () => {
|
|
sessionStatus = "authenticated";
|
|
const client = makeRouter(async function* () {
|
|
yield* [];
|
|
throw new ConnectError("revoked", Code.Unauthenticated);
|
|
});
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: async () => {},
|
|
random: () => 0,
|
|
});
|
|
|
|
await vi.waitFor(() => {
|
|
expect(signOutSpy).toHaveBeenCalled();
|
|
});
|
|
expect(signOutSpy).toHaveBeenCalledWith("revoked");
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("clean end-of-stream on an authenticated session is the revocation signal", async () => {
|
|
sessionStatus = "authenticated";
|
|
const client = makeRouter(async function* () {
|
|
yield* [];
|
|
});
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: async () => {},
|
|
random: () => 0,
|
|
});
|
|
await vi.waitFor(() => {
|
|
expect(signOutSpy).toHaveBeenCalledWith("revoked");
|
|
});
|
|
eventStream.stop();
|
|
});
|
|
|
|
test("connectionStatus transitions through connecting → connected → idle", async () => {
|
|
expect(eventStream.connectionStatus).toBe("idle");
|
|
const event = buildEvent(
|
|
"game.turn.ready",
|
|
new TextEncoder().encode("{}"),
|
|
);
|
|
const observed: string[] = [];
|
|
const client = makeRouter(async function* () {
|
|
yield event;
|
|
});
|
|
const handler = vi.fn(() => {
|
|
observed.push(eventStream.connectionStatus);
|
|
});
|
|
eventStream.on("game.turn.ready", handler);
|
|
eventStream.start({
|
|
core: mockCore(),
|
|
keypair: mockKeypair(),
|
|
deviceSessionId: "device-1",
|
|
gatewayResponsePublicKey: new Uint8Array(32),
|
|
client,
|
|
sleep: async () => {},
|
|
random: () => 0,
|
|
});
|
|
await vi.waitFor(() => {
|
|
expect(handler).toHaveBeenCalled();
|
|
});
|
|
// Inside the handler, status had already flipped to connected.
|
|
expect(observed).toContain("connected");
|
|
eventStream.stop();
|
|
});
|
|
});
|