feat: add authenticated realtime presence

2026-08-11 20:22:12 -07:00
parent 16dc85a6f8
commit 92ebf8abb5
20 changed files with 3746 additions and 1 deletions
@@ -0,0 +1,412 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import {
+  createRealtimeClient,
+  type ActorPoseSnapshot,
+  type RealtimeClientTimers,
+} from "../realtime/index.ts";
+
+const OFFICE = { kind: "office" as const, officeId: "hq" };
+
+function actor(sequence = 0, xM = 0): ActorPoseSnapshot {
+  return {
+    entity: "actor",
+    actorId: "actor-1",
+    kind: "humanoid",
+    sequence,
+    timestampMs: 1_765_000_000_000 + sequence,
+    pose: {
+      space: "local",
+      cell: OFFICE,
+      xM,
+      yM: 0,
+      zM: 0,
+      headingDeg: 0,
+      pitchDeg: 0,
+    },
+    velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+  };
+}
+
+function joinGrant(requestId: string) {
+  return {
+    type: "join-grant",
+    protocolVersion: 1,
+    requestId,
+    sessionId: "session-1",
+    actorId: "actor-1",
+    role: "member",
+    serverEpoch: "epoch-1",
+    serverTimeMs: 1_765_000_000_000,
+    nextSequence: 4,
+    resumeToken: "resume-secret",
+    interests: [OFFICE],
+    initial: [],
+  };
+}
+
+function resumeGrant(requestId: string, interests = [OFFICE]) {
+  return {
+    type: "resume-grant",
+    protocolVersion: 1,
+    requestId,
+    sessionId: "session-1",
+    serverEpoch: "epoch-1",
+    serverTimeMs: 1_765_000_000_001,
+    nextSequence: 5,
+    resumeToken: "rotated-resume-secret",
+    continuous: true,
+    snapshot: [],
+    interests,
+  };
+}
+
+function jsonResponse(value: unknown, status = 200): Response {
+  return new Response(JSON.stringify(value), {
+    status,
+    headers: { "Content-Type": "application/json" },
+  });
+}
+
+class FakeTimers implements RealtimeClientTimers {
+  readonly pending = new Map<number, { callback: () => void; delay: number }>();
+  private next = 1;
+
+  setTimeout(callback: () => void, delay: number): ReturnType<typeof setTimeout> {
+    const id = this.next++;
+    this.pending.set(id, { callback, delay });
+    return id as unknown as ReturnType<typeof setTimeout>;
+  }
+
+  clearTimeout(handle: ReturnType<typeof setTimeout>): void {
+    this.pending.delete(handle as unknown as number);
+  }
+
+  runFirst(): number {
+    const entry = this.pending.entries().next().value as [number, { callback: () => void; delay: number }] | undefined;
+    if (!entry) throw new Error("no timer pending");
+    this.pending.delete(entry[0]);
+    entry[1].callback();
+    return entry[1].delay;
+  }
+}
+
+async function settle(): Promise<void> {
+  await Promise.resolve();
+  await Promise.resolve();
+  await new Promise<void>((resolve) => globalThis.setTimeout(resolve, 0));
+}
+
+describe("realtime browser client", () => {
+  it("uses memory-only bearer auth, same-origin credentials, and validates join identity", async () => {
+    const calls: Array<{ url: string; init: RequestInit; body: Record<string, unknown> }> = [];
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      calls.push({ url: String(input), init, body });
+      return jsonResponse(joinGrant(String(body.requestId)));
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", accessToken: "access-secret", fetch: fetcher });
+    await client.join(OFFICE, actor());
+
+    assert.equal(calls[0]?.url, "/api/v1/realtime/join");
+    assert.equal(calls[0]?.init.credentials, "same-origin");
+    assert.equal(calls[0]?.init.cache, "no-store");
+    assert.equal(new Headers(calls[0]?.init.headers).get("Authorization"), "Bearer access-secret");
+    assert.equal(calls[0]?.url.includes("access-secret"), false);
+    assert.equal(JSON.stringify(client.state()).includes("secret"), false);
+    assert.throws(() => client.publishPose({
+      ...actor(), velocity: { ...actor().velocity, xMps: Number.NaN },
+    }), /invalid/);
+  });
+
+  it("uses an HttpOnly same-origin session when no browser-readable bearer exists", async () => {
+    const calls: RequestInit[] = [];
+    const fetcher: typeof fetch = async (_input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      calls.push(init);
+      return jsonResponse(joinGrant(String(body.requestId)), 201);
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+    await client.join(OFFICE, actor());
+
+    assert.equal(new Headers(calls[0]?.headers).has("authorization"), false);
+    assert.equal(calls[0]?.credentials, "same-origin");
+  });
+
+  it("accepts one authenticated request adapter without retaining its credential", async () => {
+    let used = false;
+    const authenticatedFetch: typeof fetch = async (_input, init = {}) => {
+      used = true;
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      return jsonResponse(joinGrant(String(body.requestId)), 201);
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", authenticatedFetch });
+    await client.join(OFFICE, actor());
+    assert.equal(used, true);
+    assert.throws(
+      () => createRealtimeClient({ actorId: "actor-1", authenticatedFetch, fetch }),
+      /either fetch or authenticatedFetch/,
+    );
+  });
+
+  it("coalesces poses at 10-15Hz and assigns monotonic client sequences", async () => {
+    const timers = new FakeTimers();
+    const poses: ActorPoseSnapshot[] = [];
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+      if (String(input).endsWith("/pose")) {
+        poses.push(body.snapshot as ActorPoseSnapshot);
+        return new Response(null, { status: 204 });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({
+      actorId: "actor-1", accessToken: "token", fetch: fetcher, timers, sendIntervalMs: 80,
+    });
+    await client.join(OFFICE, actor());
+    client.publishPose(actor(0, 1));
+    client.publishPose(actor(0, 2));
+    assert.equal(timers.pending.size, 1);
+    assert.equal(timers.runFirst(), 80);
+    await settle();
+    assert.equal(poses.length, 1);
+    assert.equal(poses[0]?.pose.space === "local" ? poses[0].pose.xM : -1, 2);
+    assert.equal(poses[0]?.sequence, 5, "the superseding pose keeps the later monotonic sequence");
+
+    client.publishPose(actor(0, 3));
+    timers.runFirst();
+    await settle();
+    assert.equal(poses[1]?.sequence, 6);
+    assert.throws(
+      () => createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher, sendIntervalMs: 50 }),
+      /66\.67ms/,
+    );
+  });
+
+  it("holds a pose until a stream resume delivers the rotated credential", async () => {
+    const timers = new FakeTimers();
+    const poseTokens: unknown[] = [];
+    const stream: { current: ReadableStreamDefaultController<Uint8Array> | null } = { current: null };
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+      if (String(input).endsWith("/events")) {
+        return new Response(new ReadableStream<Uint8Array>({ start(controller) { stream.current = controller; } }), {
+          headers: { "Content-Type": "text/event-stream" },
+        });
+      }
+      if (String(input).endsWith("/pose")) {
+        poseTokens.push(body.token);
+        return new Response(null, { status: 202 });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({
+      actorId: "actor-1", accessToken: "token", fetch: fetcher, timers,
+    });
+    client.subscribe({});
+    await client.join(OFFICE, actor());
+    await settle();
+    client.publishPose(actor(0, 1));
+    timers.runFirst();
+    await settle();
+    assert.equal(poseTokens.length, 0);
+
+    stream.current?.enqueue(new TextEncoder().encode(
+      `data: ${JSON.stringify(resumeGrant("stream-resume"))}\n\n`,
+    ));
+    await settle();
+    timers.runFirst();
+    await settle();
+    assert.deepEqual(poseTokens, ["rotated-resume-secret"]);
+    await client.dispose();
+  });
+
+  it("parses validated POST SSE events, rejects malformed input, and reconnects with backoff", async () => {
+    const timers = new FakeTimers();
+    const malformed: string[] = [];
+    const messages: string[] = [];
+    let streamCalls = 0;
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+      if (String(input).endsWith("/events")) {
+        streamCalls += 1;
+        const delta = {
+          type: "pose-delta", protocolVersion: 1, serverEpoch: "epoch-1", sequence: 4,
+          timestampMs: 1_765_000_000_004, updates: [], removedEntityIds: [],
+        };
+        return new Response(`data: not-json\n\ndata: ${JSON.stringify(delta)}\n\n`, {
+          headers: { "Content-Type": "text/event-stream" },
+        });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({
+      actorId: "actor-1",
+      accessToken: "token",
+      fetch: fetcher,
+      timers,
+      reconnectBaseMs: 200,
+      reconnectMaximumMs: 2_000,
+      reconnectJitter: 0,
+      random: () => 0.5,
+    });
+    client.subscribe({
+      onMalformedEvent: (error) => malformed.push(error.message),
+      onMessage: (message) => messages.push(message.type),
+    });
+    await client.join(OFFICE, actor());
+    await settle();
+    assert.deepEqual(messages, ["pose-delta"]);
+    assert.equal(malformed.length, 1);
+    assert.equal(client.state().status, "reconnecting");
+    assert.equal(timers.runFirst(), 200);
+    await settle();
+    assert.equal(streamCalls, 2);
+  });
+
+  it("moves interest through resume and aborts streaming before leave/dispose", async () => {
+    const requests: Array<{ url: string; body: Record<string, unknown>; signal: AbortSignal | null }> = [];
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      requests.push({ url: String(input), body, signal: init.signal ?? null });
+      if (String(input).endsWith("/join")) {
+        if (body.type === "resume-request") return jsonResponse(resumeGrant(String(body.requestId)));
+        return jsonResponse(joinGrant(String(body.requestId)));
+      }
+      if (String(input).endsWith("/events")) {
+        return new Response(new ReadableStream({ start() { /* held open until aborted */ } }), {
+          headers: { "Content-Type": "text/event-stream" },
+        });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher });
+    client.subscribe({});
+    await client.join(OFFICE, actor());
+    await settle();
+    const firstStream = requests.find((request) => request.url.endsWith("/events"));
+    assert.equal(firstStream?.signal?.aborted, false);
+
+    const moved = { kind: "floor" as const, officeId: "hq", floorId: "two" };
+    await client.moveInterest(moved);
+    assert.deepEqual(client.state().interests, [moved]);
+    assert.equal(firstStream?.signal?.aborted, true);
+    await client.dispose();
+    assert.equal(client.state().status, "disposed");
+    assert.equal(requests.filter((request) => request.url.endsWith("/leave")).length, 1);
+    assert.equal(JSON.stringify(client.state()).includes("resume-secret"), false);
+  });
+
+  it("closes the old-cell stream when an interest handoff fails", async () => {
+    const requests: Array<{ url: string; signal: AbortSignal | null }> = [];
+    let resumes = 0;
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const url = String(input);
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      requests.push({ url, signal: init.signal ?? null });
+      if (url.endsWith("/join") && body.type === "resume-request") {
+        resumes += 1;
+        return jsonResponse({ error: "interest" }, 409);
+      }
+      if (url.endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+      if (url.endsWith("/events")) {
+        return new Response(new ReadableStream({ start() { /* held until abort */ } }), {
+          headers: { "Content-Type": "text/event-stream" },
+        });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher });
+    client.subscribe({});
+    await client.join(OFFICE, actor());
+    await settle();
+    const oldStream = requests.find((request) => request.url.endsWith("/events"));
+    await assert.rejects(
+      client.moveInterest({ kind: "city", cityId: "bay-area" }),
+      /request failed/,
+    );
+    assert.equal(resumes, 1);
+    assert.equal(oldStream?.signal?.aborted, true);
+    assert.equal(client.state().status, "joined");
+    assert.deepEqual(client.state().interests, [OFFICE]);
+    await client.dispose();
+  });
+
+  it("closes the old-cell stream before a slow interest handoff settles", async () => {
+    const requests: Array<{ url: string; signal: AbortSignal | null }> = [];
+    let finishResume: ((response: Response) => void) | null = null;
+    let resumeRequestId = "";
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const url = String(input);
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      requests.push({ url, signal: init.signal ?? null });
+      if (url.endsWith("/join") && body.type === "resume-request") {
+        resumeRequestId = String(body.requestId);
+        return new Promise<Response>((resolve) => { finishResume = resolve; });
+      }
+      if (url.endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+      if (url.endsWith("/events")) {
+        return new Response(new ReadableStream({ start() { /* held until abort */ } }), {
+          headers: { "Content-Type": "text/event-stream" },
+        });
+      }
+      return new Response(null, { status: 204 });
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+    client.subscribe({});
+    await client.join(OFFICE, actor());
+    await settle();
+    const oldStream = requests.find((request) => request.url.endsWith("/events"));
+    const moved = { kind: "city" as const, cityId: "bay-area" as const };
+    const handoff = client.moveInterest(moved);
+    assert.equal(oldStream?.signal?.aborted, true, "privacy-cell stream closes before HTTP completes");
+    assert.ok(finishResume);
+    (finishResume as (response: Response) => void)(jsonResponse(resumeGrant(resumeRequestId)));
+    await handoff;
+    assert.deepEqual(client.state().interests, [moved]);
+    await client.dispose();
+  });
+
+  it("lets only the newest concurrent join commit client state", async () => {
+    const pending: Array<{
+      body: Record<string, unknown>;
+      resolve: (response: Response) => void;
+    }> = [];
+    const leaves: Record<string, unknown>[] = [];
+    const fetcher: typeof fetch = async (input, init = {}) => {
+      const body = JSON.parse(String(init.body)) as Record<string, unknown>;
+      if (String(input).endsWith("/leave")) {
+        leaves.push(body);
+        return new Response(null, { status: 204 });
+      }
+      return new Promise<Response>((resolve) => pending.push({ body, resolve }));
+    };
+    const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+    const first = client.join(OFFICE, actor());
+    const latestInterest = { kind: "city" as const, cityId: "socal" as const };
+    const latest = client.join(latestInterest, actor());
+    const secondRequest = pending[1];
+    assert.ok(secondRequest);
+    secondRequest.resolve(jsonResponse({
+      ...joinGrant(String(secondRequest.body.requestId)),
+      sessionId: "session-latest",
+      interests: [latestInterest],
+    }));
+    await latest;
+    const firstRequest = pending[0];
+    assert.ok(firstRequest);
+    firstRequest.resolve(jsonResponse({
+      ...joinGrant(String(firstRequest.body.requestId)),
+      sessionId: "session-stale",
+    }));
+    await assert.rejects(first, /superseded/);
+    await settle();
+    assert.equal(client.state().sessionId, "session-latest");
+    assert.deepEqual(client.state().interests, [latestInterest]);
+    assert.deepEqual(leaves, [{ sessionId: "session-stale", token: "resume-secret" }]);
+    await client.dispose();
+  });
+});