413 lines
16 KiB
TypeScript
413 lines
16 KiB
TypeScript
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();
|
|
});
|
|
});
|