From 7a0fa06807897d93845fff0cbd34c517770d2392 Mon Sep 17 00:00:00 2001 From: deepfates Date: Mon, 13 Jul 2026 17:10:18 -0700 Subject: [PATCH] feat(presence): typed awareness on the presence frame (lync 0.4.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build the LIVE-SHARED awareness half on lync's existing ephemeral presence frame, per the pinned presence contract. The relay stays a stateless fanout — untouched — and nothing presence-related is ever written to disk. - sync-protocol: replace opaque `data?: unknown` with the typed LyncPresence schema (clock + state{actor,via?,focus?,typing?}) and a per-connection `client` id on the frame — the LWW key, distinct from the durable `actor`. decodeFrame validates + canonicalizes it (drops unknown extras, rejects poisoned clocks) so a bad payload never reaches the awareness clock. - presence-awareness: a client-side per-participant state machine — apply-iff-clock-strictly-greater (LWW per participant, no CRDT), heard-from lastSeen refresh, ~15s self-rebroadcast heartbeat, ~30s TTL local removal, state=null immediate leave, and an {added,updated,removed} callback keyed by clientId. - synced-store: onPresence / presence() now carry (root, client, LyncPresence). Tests (retire the untested-primitive gap — only a codec roundtrip existed before): - presence-schema-lock: pins the exact wire shape (the seam textile builds to). - presence-awareness unit: LWW, immediate leave, TTL sweep, heartbeat liveness + recovery, and start()/stop() over real fake-clocked timers. - presence-awareness integration: real relay + two real synced stores — typed presence A->B, silent-A TTL removal, null-leave immediate removal, and an assertion the relay wrote NOTHING to disk. Bump 0.3.0 -> 0.4.0 (additive minor). Do not publish. --- package.json | 7 +- src/presence-awareness.ts | 248 ++++++++++++++++++ src/sync-protocol.ts | 79 +++++- src/synced-store.ts | 15 +- .../presence-awareness.integration.test.ts | 130 +++++++++ test/presence-awareness.test.ts | 134 ++++++++++ test/presence-schema-lock.test.ts | 99 +++++++ test/sync-protocol.test.ts | 8 +- 8 files changed, 708 insertions(+), 12 deletions(-) create mode 100644 src/presence-awareness.ts create mode 100644 test/cli/presence-awareness.integration.test.ts create mode 100644 test/presence-awareness.test.ts create mode 100644 test/presence-schema-lock.test.ts diff --git a/package.json b/package.json index 67b1c3d..23b9e0e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@deepfates/lync", - "version": "0.3.0", + "version": "0.4.0", "description": "The lync format: append-only JSONL event logs merged by set union. Parsing, stores, views, looms, live sync, loom client, indexes, the sync relay, and the lync command. Zero dependencies.", "type": "module", "license": "MIT", @@ -96,6 +96,11 @@ "import": "./dist/synced-store.js", "default": "./dist/synced-store.js" }, + "./presence-awareness": { + "types": "./dist/presence-awareness.d.ts", + "import": "./dist/presence-awareness.js", + "default": "./dist/presence-awareness.js" + }, "./uuid": { "types": "./dist/uuid.d.ts", "import": "./dist/uuid.js", diff --git a/src/presence-awareness.ts b/src/presence-awareness.ts new file mode 100644 index 0000000..e89c0d6 --- /dev/null +++ b/src/presence-awareness.ts @@ -0,0 +1,248 @@ +import type { LyncPresence } from "./sync-protocol.js"; + +/** + * Client-side awareness on top of lync's ephemeral presence frame. + * + * The relay is a stateless fanout: it holds no roster and persists nothing. + * Every participant instead maintains its OWN view of who is present by + * listening to the presence frames the relay echoes. This module is that view + * — a per-root, per-participant state machine implementing the pinned presence + * contract: + * + * - Identity is a per-client id (a participant). One `actor` (human) may run + * several clients; the map is keyed by client, never by actor. + * - `clock` is a monotonic uint minted by the sender. A remote entry's STATE + * is applied IFF its clock is strictly greater than the last one seen from + * that client — last-writer-wins per participant, no CRDT merge. + * - Any inbound frame (even a stale-clock heartbeat) counts as "heard from" + * and refreshes lastSeen; a participant not heard from within the TTL is + * removed locally. Liveness (heard-from) and versioning (clock) are + * separate on purpose, so a heartbeat need not burn a new clock. + * - `state === null` is a graceful leave: remove that participant at once. + * - Self re-broadcasts its current state on a heartbeat so late joiners and + * peers who TTL'd it out recover it. + * + * The machine is pure and injectable: `receive`, `sweep`, and `heartbeat` are + * driven explicitly (deterministic in tests with an injected `now`), while + * `start()` wires real intervals over them for production use. + */ + +/** A present participant, as this client currently sees them. */ +export interface PresenceParticipant { + /** The sender's per-client participant id (the map key). */ + client: string; + /** Their last applied non-null state (they are present, so state is set). */ + state: NonNullable; + /** The clock of that applied state. */ + clock: number; + /** `now` at which we last heard ANY frame from this client. */ + lastSeen: number; +} + +/** What changed in one roster transition, keyed by client via each entry. */ +export interface PresenceDelta { + added: PresenceParticipant[]; + updated: PresenceParticipant[]; + /** Their last known state before they left / timed out. */ + removed: PresenceParticipant[]; +} + +export interface PresenceAwarenessOptions { + /** This client's stable participant id. Defaults to a random id. */ + client?: string; + /** Outbound: send a presence frame. Wire to `SyncedStore.presence`. */ + send: (root: string, client: string, data: LyncPresence) => void; + /** Fires whenever a root's remote roster changes. */ + onDelta?: (root: string, delta: PresenceDelta) => void; + /** Self re-broadcast period, ms. Default 15_000. */ + heartbeatMs?: number; + /** Remove a participant not heard from within this many ms. Default 30_000. */ + ttlMs?: number; + /** How often `start()` runs the TTL sweep, ms. Default = heartbeatMs. */ + sweepMs?: number; + /** Injectable clock (ms). Default `Date.now`. */ + now?: () => number; +} + +type Timer = ReturnType; + +interface LocalRoot { + clock: number; + /** Our last SENT state for this root; null once we have left. */ + state: LyncPresence["state"]; +} + +export interface PresenceAwareness { + /** This client's participant id (the `client` on every frame it sends). */ + readonly client: string; + /** + * Publish this client's presence on a root. Mints a strictly-greater clock, + * remembers the state for heartbeats, and sends the frame. Pass `null` to + * leave gracefully (peers remove this client immediately). Returns the clock. + */ + setLocal(root: string, state: LyncPresence["state"], now?: number): number; + /** Ingest one inbound presence frame (from `SyncedStore.onPresence`). */ + receive(root: string, client: string, presence: LyncPresence, now?: number): void; + /** Re-broadcast current local state on every joined root (heartbeat tick). */ + heartbeat(now?: number): void; + /** Remove participants past the TTL on every root (sweep tick). */ + sweep(now?: number): void; + /** Current remote roster for a root (excludes self). */ + roster(root: string): PresenceParticipant[]; + /** Begin real-timer heartbeats + TTL sweeps. Idempotent. */ + start(): void; + /** Stop timers and leave every joined root gracefully. */ + stop(): void; +} + +let idCounter = 0; +function defaultClientId(): string { + idCounter += 1; + const rand = Math.random().toString(36).slice(2, 10); + return `client-${Date.now().toString(36)}-${idCounter}-${rand}`; +} + +export function createPresenceAwareness(options: PresenceAwarenessOptions): PresenceAwareness { + const client = options.client ?? defaultClientId(); + const heartbeatMs = options.heartbeatMs ?? 15_000; + const ttlMs = options.ttlMs ?? 30_000; + const sweepMs = options.sweepMs ?? heartbeatMs; + const now = options.now ?? (() => Date.now()); + + // Remote participants: root -> client -> entry. Never contains `client`. + const rosters = new Map>(); + // Our own last-sent state per root, for heartbeats. + const locals = new Map(); + let heartbeatTimer: Timer | undefined; + let sweepTimer: Timer | undefined; + + const view = (p: PresenceParticipant): PresenceParticipant => ({ ...p, state: { ...p.state } }); + + const emit = (root: string, delta: PresenceDelta) => { + if (delta.added.length === 0 && delta.updated.length === 0 && delta.removed.length === 0) return; + options.onDelta?.(root, delta); + }; + + const rosterFor = (root: string): Map => { + let m = rosters.get(root); + if (!m) { + m = new Map(); + rosters.set(root, m); + } + return m; + }; + + return { + client, + + setLocal(root, state, at = now()) { + const prev = locals.get(root); + const clock = (prev?.clock ?? 0) + 1; + locals.set(root, { clock, state }); + // A graceful leave need not keep re-broadcasting; drop the local record + // AFTER sending the null so the frame still carries the bumped clock. + const data: LyncPresence = { clock, state }; + options.send(root, client, data); + if (state === null) locals.delete(root); + void at; // `at` reserved for callers that pin send time; clock is the ordering key + return clock; + }, + + receive(root, from, presence, at = now()) { + if (from === client) return; // never track ourselves + const map = rosterFor(root); + const entry = map.get(from); + + // Stale or equal clock: still "heard from" (refresh liveness), but LWW + // rejects the state — no transition. A heartbeat re-sending the same + // clock lands here for peers who already have us. + if (entry && presence.clock <= entry.clock) { + entry.lastSeen = at; + return; + } + + // Strictly newer clock (or a client we do not know yet). + if (presence.state === null) { + // Graceful leave. Unknown client: nothing to remove. + if (entry) { + map.delete(from); + emit(root, { added: [], updated: [], removed: [view(entry)] }); + } + return; + } + + if (entry) { + entry.clock = presence.clock; + entry.state = presence.state; + entry.lastSeen = at; + emit(root, { added: [], updated: [view(entry)], removed: [] }); + } else { + const fresh: PresenceParticipant = { + client: from, + state: presence.state, + clock: presence.clock, + lastSeen: at, + }; + map.set(from, fresh); + emit(root, { added: [view(fresh)], updated: [], removed: [] }); + } + }, + + heartbeat(at = now()) { + for (const [root, local] of locals) { + if (local.state === null) continue; + // Re-send WITHOUT bumping the clock: this is liveness, not a new state. + options.send(root, client, { clock: local.clock, state: local.state }); + } + void at; + }, + + sweep(at = now()) { + for (const [root, map] of rosters) { + const removed: PresenceParticipant[] = []; + for (const [from, entry] of map) { + if (at - entry.lastSeen > ttlMs) { + map.delete(from); + removed.push(view(entry)); + } + } + if (removed.length > 0) emit(root, { added: [], updated: [], removed }); + } + }, + + roster(root) { + const map = rosters.get(root); + return map ? [...map.values()].map(view) : []; + }, + + start() { + if (heartbeatTimer === undefined) { + heartbeatTimer = setInterval(() => this.heartbeat(), heartbeatMs); + if (typeof (heartbeatTimer as { unref?: () => void }).unref === "function") { + (heartbeatTimer as { unref: () => void }).unref(); + } + } + if (sweepTimer === undefined) { + sweepTimer = setInterval(() => this.sweep(), sweepMs); + if (typeof (sweepTimer as { unref?: () => void }).unref === "function") { + (sweepTimer as { unref: () => void }).unref(); + } + } + }, + + stop() { + if (heartbeatTimer !== undefined) { + clearInterval(heartbeatTimer); + heartbeatTimer = undefined; + } + if (sweepTimer !== undefined) { + clearInterval(sweepTimer); + sweepTimer = undefined; + } + // Leave every root we are still present on. + for (const [root, local] of [...locals]) { + if (local.state !== null) this.setLocal(root, null); + } + }, + }; +} diff --git a/src/sync-protocol.ts b/src/sync-protocol.ts index 21d816e..5780a06 100644 --- a/src/sync-protocol.ts +++ b/src/sync-protocol.ts @@ -9,7 +9,7 @@ * server → client {"t":"ev", "root": string, "seq": number, "line": string, "gen"?: string} * server → client {"t":"live", "root": string, "seq": number, "gen"?: string} * client → server {"t":"ev", "root": string, "line": string} - * either direction {"t":"presence", "root": string, "data"?: unknown} + * either direction {"t":"presence", "root": string, "client": string, "data": LyncPresence} * either direction {"t":"err", "root"?: string, "reason": string, "detail"?: string} * * `seq` is the server's own per-root arrival counter — a resume cursor, not @@ -52,10 +52,43 @@ export interface LiveFrame { gen?: string; } +/** + * Ephemeral awareness payload — who is on a loom right now and where their + * attention sits. Carried ONLY on a {t:"presence"} frame and NEVER stored as a + * durable event: the relay fans presence out and forgets it. + * + * `clock` is a monotonic uint minted per client (a participant). A receiver + * applies an incoming entry for a client IFF its clock is strictly greater than + * the last one seen from that same client — last-writer-wins PER PARTICIPANT, + * no CRDT merge. `state === null` is a graceful leave: remove that participant + * immediately. + */ +export interface LyncPresence { + /** Monotonic uint per client. Apply iff strictly greater than the last seen. */ + clock: number; + /** null == graceful leave (remove immediately). */ + state: null | { + /** Author identity — the SAME string used for durable turn authorship. */ + actor: string; + /** Controller, e.g. "textile-browser". */ + via?: string; + /** Id of the node the participant's attention is on (their tree cursor). */ + focus?: string | null; + /** Is the participant composing right now. */ + typing?: boolean; + }; +} + export interface PresenceFrame { t: "presence"; root: string; - data?: unknown; + /** + * Per-connection participant id — the key the awareness layer applies LWW + * over and reports in its {added,updated,removed} callback. Distinct from + * `data.state.actor`: one actor (human) may drive several clients. + */ + client: string; + data: LyncPresence; } export interface ErrFrame { @@ -77,6 +110,37 @@ export function isCursor(value: unknown): value is number { return typeof value === "number" && Number.isInteger(value) && value >= 0; } +/** + * Validate and canonicalize a LyncPresence payload. Returns a fresh object + * carrying ONLY the known fields (unknown extras from a newer peer are dropped, + * never fatal), or undefined if the shape is not a LyncPresence. A malformed + * awareness payload must never poison the per-participant clock, so this is + * strict about the fields it does read: `clock` a nonnegative integer, `state` + * either null or an object with a string `actor` and optional well-typed + * via/focus/typing. + */ +function normalizePresence(value: unknown): LyncPresence | undefined { + if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined; + const raw = value as Record; + if (!isCursor(raw.clock)) return undefined; + if (raw.state === null) return { clock: raw.clock, state: null }; + if (typeof raw.state !== "object" || Array.isArray(raw.state)) return undefined; + const s = raw.state as Record; + if (typeof s.actor !== "string") return undefined; + if (s.via !== undefined && typeof s.via !== "string") return undefined; + if (s.focus !== undefined && s.focus !== null && typeof s.focus !== "string") return undefined; + if (s.typing !== undefined && typeof s.typing !== "boolean") return undefined; + return { + clock: raw.clock, + state: { + actor: s.actor, + ...(s.via !== undefined ? { via: s.via as string } : {}), + ...(s.focus !== undefined ? { focus: s.focus as string | null } : {}), + ...(s.typing !== undefined ? { typing: s.typing as boolean } : {}), + }, + }; +} + export function encodeFrame(frame: SyncFrame): string { return JSON.stringify(frame); } @@ -142,11 +206,16 @@ export function decodeFrame(raw: string | Uint8Array): SyncFrame { seq: frame.seq as number, ...(frame.gen !== undefined ? { gen: frame.gen as string } : {}), }; - case "presence": - if (typeof frame.root !== "string") { + case "presence": { + if (typeof frame.root !== "string" || typeof frame.client !== "string") { return { t: "err", reason: "malformed-presence" }; } - return { t: "presence", root: frame.root, data: frame.data }; + const presence = normalizePresence(frame.data); + if (presence === undefined) { + return { t: "err", reason: "malformed-presence", detail: "data is not a LyncPresence" }; + } + return { t: "presence", root: frame.root, client: frame.client, data: presence }; + } default: return { t: "err", diff --git a/src/synced-store.ts b/src/synced-store.ts index c5fd74c..0b256b5 100644 --- a/src/synced-store.ts +++ b/src/synced-store.ts @@ -1,6 +1,6 @@ import type { EventStore, StoredEvent, AppendResult } from "./store.js"; import type { LyncEventBody } from "./events.js"; -import { decodeFrame, encodeFrame, type SyncFrame } from "./sync-protocol.js"; +import { decodeFrame, encodeFrame, type LyncPresence, type SyncFrame } from "./sync-protocol.js"; /** * Live sync for an EventStore, built on the dumb line-union protocol. @@ -47,14 +47,19 @@ export interface SyncTransport { export interface SyncedStoreOptions { onStatus?: (status: SyncStatus) => void; - onPresence?: (root: string, data: unknown) => void; + /** + * Fires for every inbound presence frame: the root, the sender's per-client + * participant id, and the typed awareness payload. Ephemeral — never a stored + * event. Feed this straight into a PresenceAwareness (see presence-awareness). + */ + onPresence?: (root: string, client: string, presence: LyncPresence) => void; } export interface SyncedStore extends EventStore { /** Begin syncing a root: push local backlog, then subscribe from the cursor. */ syncRoot(rootId: string): void; /** Relay an ephemeral presence frame for a root; never stored. */ - presence(root: string, data: unknown): void; + presence(root: string, client: string, data: LyncPresence): void; status(): SyncStatus; close(): void; } @@ -221,7 +226,7 @@ export function createSyncedStore( return; } case "presence": { - options.onPresence?.(frame.root, frame.data); + options.onPresence?.(frame.root, frame.client, frame.data); return; } case "err": { @@ -281,7 +286,7 @@ export function createSyncedStore( ...(inner.exportRootBytes ? { exportRootBytes: (rootId: string) => inner.exportRootBytes!(rootId) } : {}), ...(inner.diagnostics ? { diagnostics: () => inner.diagnostics!() } : {}), syncRoot: ensureSynced, - presence: (root, data) => transport.send({ t: "presence", root, data }), + presence: (root, client, data) => transport.send({ t: "presence", root, client, data }), status: () => ({ connection, liveRoots: [...liveRoots], conflicts: [...conflicts], failures: [...failures] }), close: () => transport.close(), }; diff --git a/test/cli/presence-awareness.integration.test.ts b/test/cli/presence-awareness.integration.test.ts new file mode 100644 index 0000000..7d6aa86 --- /dev/null +++ b/test/cli/presence-awareness.integration.test.ts @@ -0,0 +1,130 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { mkdtemp, readdir } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createMemoryEventStore } from "@deepfates/lync/memory-log"; +import { createSyncedStore, createWebSocketTransport, type SyncedStore } from "@deepfates/lync/synced-store"; +import { startLyncServe, type LyncSyncServer } from "@deepfates/lync/relay"; +import { createPresenceAwareness, type PresenceAwareness, type PresenceDelta } from "@deepfates/lync/presence-awareness"; + +/** + * Awareness end to end: a REAL relay + two REAL synced stores over the global + * WebSocket. Peer A's typed presence reaches peer B; A going silent times out + * of B's roster; a null leave removes A immediately; and the relay writes + * NOTHING to disk for presence. This retires the untested-primitive gap — the + * only presence test before this was a codec roundtrip. + */ + +async function waitFor(check: () => boolean | Promise, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await check()) return; + await new Promise((r) => setTimeout(r, 20)); + } + throw new Error("waitFor: condition not met within timeout"); +} + +interface Peer { + store: SyncedStore; + aware: PresenceAwareness; + deltas: Array<{ root: string; delta: PresenceDelta }>; + /** Live roster keyed by client id, folded from the deltas (what B "sees"). */ + seen: Map; +} + +// A shrunk TTL so "A goes silent for ~30s" runs in milliseconds. The awareness +// default is 30_000ms (see presence-awareness); we drive the same machine with +// ttlMs=120 and hand-tick `sweep`, so the test is deterministic, not timed. +const TTL_MS = 120; + +function makePeer(url: string, actor: string): Peer { + const deltas: Array<{ root: string; delta: PresenceDelta }> = []; + const seen = new Map(); + const inner = createMemoryEventStore(); + const transport = createWebSocketTransport(url, { reconnectMs: 0 }); + let aware: PresenceAwareness; + const store = createSyncedStore(inner, transport, { + onPresence: (root, client, presence) => aware.receive(root, client, presence), + }); + aware = createPresenceAwareness({ + client: `${actor}-conn`, + ttlMs: TTL_MS, + send: (root, client, data) => store.presence(root, client, data), + onDelta: (root, delta) => { + deltas.push({ root, delta }); + for (const p of [...delta.added, ...delta.updated]) seen.set(p.client, p); + for (const p of delta.removed) seen.delete(p.client); + }, + }); + return { store, aware, deltas, seen }; +} + +describe("presence awareness over a real relay", () => { + let server: LyncSyncServer | undefined; + let dir: string; + const closers: Array<() => void> = []; + + afterEach(async () => { + for (const close of closers.splice(0)) close(); + await server?.close(); + server = undefined; + }); + + it("delivers typed presence A->B, TTL-removes a silent A, removes A on leave, and never touches disk", async () => { + dir = await mkdtemp(path.join(os.tmpdir(), "lync-presence-")); + server = await startLyncServe({ dir, log: () => {} }); + const url = `ws://localhost:${server.port}`; + const root = "presence-room"; // never carries a durable event + + const a = makePeer(url, "alice"); + const b = makePeer(url, "bob"); + closers.push(a.store.close, b.store.close); + + // Both must subscribe to the room to be in the relay's fanout set. + a.store.syncRoot(root); + b.store.syncRoot(root); + await waitFor(() => { + const room = server!.status().find((r) => r.root === root); + return (room?.subscribers ?? 0) >= 2; + }); + + const dirBefore = (await readdir(dir)).sort(); + + // ── ACCEPTANCE 1: A's typed presence (cursor/focus/typing + actor) reaches B. + a.aware.setLocal(root, { actor: "alice", via: "textile-browser", focus: "node-7", typing: true }); + await waitFor(() => b.seen.has("alice-conn")); + const seenA = b.seen.get("alice-conn")!; + expect(seenA.state).toEqual({ actor: "alice", via: "textile-browser", focus: "node-7", typing: true }); + expect(seenA.client).toBe("alice-conn"); // keyed by client id, not actor + + // An update (A moves its cursor + stops typing) is applied by clock order. + a.aware.setLocal(root, { actor: "alice", focus: "node-9", typing: false }); + await waitFor(() => b.seen.get("alice-conn")?.state.focus === "node-9"); + expect(b.seen.get("alice-conn")!.state.typing).toBe(false); + + // ── ACCEPTANCE 2: A goes silent -> removed from B's roster after the TTL. + // No more frames from A. B's TTL sweep (shrunk to 120ms) drops it. + await new Promise((r) => setTimeout(r, TTL_MS + 40)); + b.aware.sweep(); + expect(b.aware.roster(root)).toEqual([]); + expect(b.seen.has("alice-conn")).toBe(false); + const removedByTtl = b.deltas.flatMap((d) => d.delta.removed.map((p) => p.client)); + expect(removedByTtl).toContain("alice-conn"); + + // ── ACCEPTANCE 3: a state=null leave removes A immediately. + // A rejoins, B sees it, then A leaves gracefully. + a.aware.setLocal(root, { actor: "alice", typing: false }); + await waitFor(() => b.seen.has("alice-conn")); + const beforeLeave = b.deltas.length; + a.aware.setLocal(root, null); // graceful leave + await waitFor(() => !b.seen.has("alice-conn")); + const leaveDelta = b.deltas.slice(beforeLeave).flatMap((d) => d.delta.removed.map((p) => p.client)); + expect(leaveDelta).toContain("alice-conn"); + expect(b.aware.roster(root)).toEqual([]); + + // ── ACCEPTANCE 4: the relay wrote NOTHING to disk for presence. + const dirAfter = (await readdir(dir)).sort(); + expect(dirAfter).toEqual(dirBefore); // no new files from any presence traffic + expect(dirAfter).not.toContain(`${root}.lync`); // presence-only root has no log + }); +}); diff --git a/test/presence-awareness.test.ts b/test/presence-awareness.test.ts new file mode 100644 index 0000000..0012d73 --- /dev/null +++ b/test/presence-awareness.test.ts @@ -0,0 +1,134 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createPresenceAwareness, type PresenceDelta } from "@deepfates/lync/presence-awareness"; +import type { LyncPresence } from "@deepfates/lync/sync-protocol"; + +/** + * The awareness state machine in isolation — driven deterministically, and + * once over REAL setInterval timers (fake-clocked) to prove start()'s + * heartbeat + TTL wiring, not just the hand-driven methods. + */ + +function collector() { + const deltas: Array<{ root: string; delta: PresenceDelta }> = []; + return { + deltas, + onDelta: (root: string, delta: PresenceDelta) => deltas.push({ root, delta }), + added: () => deltas.flatMap((d) => d.delta.added.map((p) => p.client)), + updated: () => deltas.flatMap((d) => d.delta.updated.map((p) => p.client)), + removed: () => deltas.flatMap((d) => d.delta.removed.map((p) => p.client)), + }; +} + +describe("presence awareness state machine", () => { + it("adds, updates, and LWW-rejects a stale clock per participant", () => { + const c = collector(); + const aware = createPresenceAwareness({ client: "me", send: () => {}, onDelta: c.onDelta }); + + aware.receive("story", "peer", { clock: 1, state: { actor: "bob", typing: false } }, 1000); + aware.receive("story", "peer", { clock: 3, state: { actor: "bob", typing: true } }, 1100); + // Stale clock (< last applied): no state transition, but still heard-from. + aware.receive("story", "peer", { clock: 2, state: { actor: "bob", typing: false } }, 1200); + + expect(c.added()).toEqual(["peer"]); + expect(c.updated()).toEqual(["peer"]); + const roster = aware.roster("story"); + expect(roster).toHaveLength(1); + expect(roster[0].state.typing).toBe(true); // clock-3 state, not clock-2 + expect(roster[0].lastSeen).toBe(1200); // stale frame still refreshed liveness + }); + + it("never tracks itself", () => { + const c = collector(); + const aware = createPresenceAwareness({ client: "me", send: () => {}, onDelta: c.onDelta }); + aware.receive("story", "me", { clock: 1, state: { actor: "alice" } }, 0); + expect(aware.roster("story")).toEqual([]); + expect(c.deltas).toEqual([]); + }); + + it("removes a participant immediately on a state=null leave", () => { + const c = collector(); + const aware = createPresenceAwareness({ client: "me", send: () => {}, onDelta: c.onDelta }); + aware.receive("story", "peer", { clock: 1, state: { actor: "bob" } }, 0); + aware.receive("story", "peer", { clock: 2, state: null }, 10); + expect(c.removed()).toEqual(["peer"]); + expect(aware.roster("story")).toEqual([]); + }); + + it("removes a participant past the TTL on sweep, and a heartbeat clock refreshes liveness", () => { + const c = collector(); + const aware = createPresenceAwareness({ + client: "me", + send: () => {}, + onDelta: c.onDelta, + ttlMs: 100, + }); + aware.receive("story", "peer", { clock: 1, state: { actor: "bob" } }, 0); + // Heartbeat at t=80 re-sends the SAME clock; must still refresh lastSeen. + aware.receive("story", "peer", { clock: 1, state: { actor: "bob" } }, 80); + aware.sweep(150); // 150 - 80 = 70 <= 100 ttl -> survives + expect(aware.roster("story")).toHaveLength(1); + aware.sweep(200); // 200 - 80 = 120 > 100 -> removed + expect(c.removed()).toEqual(["peer"]); + expect(aware.roster("story")).toEqual([]); + }); + + it("recovers a TTL-dropped participant from a later same-clock heartbeat", () => { + const c = collector(); + const aware = createPresenceAwareness({ client: "me", send: () => {}, onDelta: c.onDelta, ttlMs: 100 }); + aware.receive("story", "peer", { clock: 5, state: { actor: "bob" } }, 0); + aware.sweep(200); // dropped + expect(aware.roster("story")).toEqual([]); + // Same clock arrives again -> unknown client now -> re-added. + aware.receive("story", "peer", { clock: 5, state: { actor: "bob" } }, 210); + expect(aware.roster("story").map((p) => p.client)).toEqual(["peer"]); + expect(c.added()).toEqual(["peer", "peer"]); + }); + + it("setLocal mints strictly-increasing clocks and sends typed frames", () => { + const sent: Array<{ root: string; client: string; data: LyncPresence }> = []; + const aware = createPresenceAwareness({ + client: "me", + send: (root, client, data) => sent.push({ root, client, data }), + }); + const c1 = aware.setLocal("story", { actor: "me", typing: true }); + const c2 = aware.setLocal("story", { actor: "me", typing: false }); + const c3 = aware.setLocal("story", null); // leave + expect([c1, c2, c3]).toEqual([1, 2, 3]); + expect(sent.map((s) => s.client)).toEqual(["me", "me", "me"]); + expect(sent[2].data.state).toBeNull(); + }); + + describe("real timers (start/stop)", () => { + afterEach(() => vi.useRealTimers()); + + it("emits heartbeats every heartbeatMs and sweeps stale peers at ttl", () => { + vi.useFakeTimers(); + const sent: LyncPresence[] = []; + const c = collector(); + const aware = createPresenceAwareness({ + client: "me", + send: (_root, _client, data) => sent.push(data), + onDelta: c.onDelta, + heartbeatMs: 1000, + ttlMs: 2500, + sweepMs: 1000, + now: () => Date.now(), + }); + aware.setLocal("story", { actor: "me", typing: false }); // clock 1, sent once + aware.receive("story", "peer", { clock: 1, state: { actor: "bob" } }); + aware.start(); + + vi.advanceTimersByTime(3000); // 3 heartbeats + 3 sweeps + // Heartbeats re-sent our state (same clock 1) 3 times, plus the initial send. + expect(sent.length).toBe(4); + expect(sent.every((d) => d.clock === 1)).toBe(true); + // Peer last heard at t=0; by t=3000 that is > 2500 ttl -> swept. + expect(c.removed()).toEqual(["peer"]); + + aware.stop(); + const after = sent.length; + vi.advanceTimersByTime(5000); + expect(sent.length).toBe(after); // timers cleared -> no more heartbeats + }); + }); +}); diff --git a/test/presence-schema-lock.test.ts b/test/presence-schema-lock.test.ts new file mode 100644 index 0000000..85f79af --- /dev/null +++ b/test/presence-schema-lock.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it } from "vitest"; +import { + decodeFrame, + encodeFrame, + type LyncPresence, + type PresenceFrame, +} from "@deepfates/lync/sync-protocol"; + +/** + * SCHEMA LOCK — the pinned presence contract, pinned in code. + * + * This is the seam a co-author (textile) builds to. It asserts the EXACT wire + * shape of the awareness payload so neither side can drift it silently: a + * change to the presence frame or LyncPresence that breaks this test is a + * deliberate, reviewed contract change, never an accident. + */ +describe("presence schema lock (pinned contract)", () => { + it("LyncPresence carries clock + state{actor,via?,focus?,typing?}", () => { + // A fully-populated present state. + const present: LyncPresence = { + clock: 1, + state: { actor: "alice", via: "textile-browser", focus: "node-42", typing: true }, + }; + // A graceful leave. + const leave: LyncPresence = { clock: 2, state: null }; + // Optional fields really are optional. + const minimal: LyncPresence = { clock: 3, state: { actor: "bob" } }; + // focus may be explicitly null (attention on nothing). + const unfocused: LyncPresence = { clock: 4, state: { actor: "cara", focus: null } }; + + for (const data of [present, leave, minimal, unfocused]) { + const frame: PresenceFrame = { t: "presence", root: "story", client: "c1", data }; + const round = decodeFrame(encodeFrame(frame)); + expect(round).toEqual(frame); + } + }); + + it("keys awareness by a per-connection `client`, distinct from the durable `actor`", () => { + // Two clients, SAME actor (one human, two tabs) — the frame's `client` + // distinguishes them; `actor` agrees with durable turn authorship. + const tabA: PresenceFrame = { + t: "presence", + root: "story", + client: "conn-a", + data: { clock: 1, state: { actor: "alice" } }, + }; + const tabB: PresenceFrame = { + t: "presence", + root: "story", + client: "conn-b", + data: { clock: 1, state: { actor: "alice" } }, + }; + expect(tabA.client).not.toBe(tabB.client); + expect(tabA.data.state?.actor).toBe(tabB.data.state?.actor); + }); + + it("rejects a presence frame missing `client` or carrying a non-LyncPresence data", () => { + // Missing client. + expect( + decodeFrame('{"t":"presence","root":"r","data":{"clock":1,"state":null}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + // Missing data entirely (was optional/opaque before — now required & typed). + expect(decodeFrame('{"t":"presence","root":"r","client":"c"}')).toMatchObject({ + t: "err", + reason: "malformed-presence", + }); + // clock not a nonnegative integer — a poisoned clock must never reach LWW. + expect( + decodeFrame('{"t":"presence","root":"r","client":"c","data":{"clock":1.5,"state":null}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + expect( + decodeFrame('{"t":"presence","root":"r","client":"c","data":{"clock":-1,"state":null}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + // state present but actor missing. + expect( + decodeFrame('{"t":"presence","root":"r","client":"c","data":{"clock":1,"state":{"typing":true}}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + // Wrong field types. + expect( + decodeFrame('{"t":"presence","root":"r","client":"c","data":{"clock":1,"state":{"actor":"a","typing":"yes"}}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + expect( + decodeFrame('{"t":"presence","root":"r","client":"c","data":{"clock":1,"state":{"actor":"a","focus":7}}}'), + ).toMatchObject({ t: "err", reason: "malformed-presence" }); + }); + + it("drops unknown extra fields from a newer peer inside state (forward-compatible, not fatal)", () => { + const decoded = decodeFrame( + '{"t":"presence","root":"r","client":"c","data":{"clock":1,"state":{"actor":"a","future":true},"extra":9}}', + ); + // Canonicalized down to exactly the known shape — the extras are gone, not fatal. + expect(decoded).toEqual({ + t: "presence", + root: "r", + client: "c", + data: { clock: 1, state: { actor: "a" } }, + }); + }); +}); diff --git a/test/sync-protocol.test.ts b/test/sync-protocol.test.ts index bd30fbe..45a7105 100644 --- a/test/sync-protocol.test.ts +++ b/test/sync-protocol.test.ts @@ -8,7 +8,13 @@ describe("lync sync protocol frames", () => { { t: "ev", root: "story", line: '{"id":"a"}', seq: 3 }, { t: "ev", root: "story", line: '{"id":"b"}' }, { t: "live", root: "story", seq: 7 }, - { t: "presence", root: "story", data: { cursor: 4 } }, + { + t: "presence", + root: "story", + client: "client-1", + data: { clock: 4, state: { actor: "alice", via: "textile-browser", focus: "node-7", typing: true } }, + }, + { t: "presence", root: "story", client: "client-1", data: { clock: 5, state: null } }, { t: "err", root: "story", reason: "same-id-conflict", detail: "a" }, ]; for (const frame of frames) {