add manifest-diff sync coordinator with last-write-wins entries and tombstones

@@ -0,0 +1,176 @@1+import { describe, expect, it } from "vitest";2+import { runChirpSyncInitiator, type ChirpSyncVaultAdapter } from "./chirp-sync-coordinator";3+import type { ChirpSyncMessage } from "../shared/sync-types";4+import type { ChirpSyncSession } from "../shared/chirp-sync-transport";5+6+function inMemoryPair(): [ChirpSyncSession, FakePeer] {7+  const a: Uint8Array[] = [];8+  const b: Uint8Array[] = [];9+  const aR: Array<(v: Uint8Array) => void> = [];10+  const bR: Array<(v: Uint8Array) => void> = [];11+12+  const ourSession: ChirpSyncSession = {13+    send(bytes) {14+      const r = bR.shift();15+      if (r) r(bytes); else b.push(bytes);16+    },17+    receive(): Promise<Uint8Array> {18+      const v = a.shift();19+      if (v) return Promise.resolve(v);20+      return new Promise<Uint8Array>((res) => aR.push(res));21+    },22+    async close() { /* no-op */ },23+    remoteStaticPublicKey: new Uint8Array(32),24+  };25+26+  const peer: FakePeer = {27+    async send(msg: ChirpSyncMessage) {28+      const bytes = new TextEncoder().encode(JSON.stringify(msg));29+      const r = aR.shift();30+      if (r) r(bytes); else a.push(bytes);31+    },32+    async receive(): Promise<ChirpSyncMessage> {33+      const v = b.shift();34+      const bytes = v ?? await new Promise<Uint8Array>((res) => bR.push(res));35+      return JSON.parse(new TextDecoder().decode(bytes)) as ChirpSyncMessage;36+    },37+  };38+  return [ourSession, peer];39+}40+41+interface FakePeer {42+  send(msg: ChirpSyncMessage): Promise<void>;43+  receive(): Promise<ChirpSyncMessage>;44+}45+46+function makeVaultAdapter(opts: {47+  vaultId?: string;48+  deviceId?: string;49+  entries?: Record<string, number>;50+  tombstones?: Record<string, number>;51+  ciphertext?: Record<string, string>;52+}): ChirpSyncVaultAdapter & {53+  appliedEntries: Record<string, string>[];54+  appliedTombstones: Record<string, number>[];55+} {56+  const appliedEntries: Record<string, string>[] = [];57+  const appliedTombstones: Record<string, number>[] = [];58+  return {59+    async computeVaultId() { return opts.vaultId ?? "vault-abc"; },60+    async getOurDeviceId() { return opts.deviceId ?? "device-init"; },61+    async buildEntryManifest() { return { ...(opts.entries ?? {}) }; },62+    async buildTombstoneManifest() { return { ...(opts.tombstones ?? {}) }; },63+    async getEncryptedEntries(ids) {64+      const out: Record<string, string> = {};65+      const cipher = opts.ciphertext ?? {};66+      for (const id of ids) if (cipher[id] !== undefined) out[id] = cipher[id];67+      return out;68+    },69+    async applyEntries(entries) { appliedEntries.push(entries); },70+    async applyTombstones(t) { appliedTombstones.push(t); },71+    appliedEntries,72+    appliedTombstones,73+  };74+}75+76+describe("runChirpSyncInitiator", () => {77+  it("completes a full sync with bi-directional updates", async () => {78+    const [session, peer] = inMemoryPair();79+    const vault = makeVaultAdapter({80+      entries: { a: 100, b: 50 }, // we have 'a' at t=100, 'b' at t=5081+      tombstones: { x: 70 },82+      ciphertext: { a: "ct-a", b: "ct-b" },83+    });84+85+    const initPromise = runChirpSyncInitiator(session, vault);86+87+    const peerProtocol = (async () => {88+      const hello = await peer.receive();89+      expect(hello.type).toBe("SYNC_HELLO");90+      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-abc", deviceId: "device-peer" });91+92+      const ourManifest = await peer.receive();93+      expect(ourManifest.type).toBe("SYNC_MANIFEST");94+      await peer.send({95+        type: "SYNC_MANIFEST",96+        entries: { a: 80, b: 90, c: 200 }, // we have newer 'a'; peer has newer 'b' and 'c'97+        tombstones: { x: 60, y: 99 },        // peer has newer tombstone for 'y'98+      });99+100+      const ourRequest = await peer.receive();101+      expect(ourRequest.type).toBe("SYNC_REQUEST");102+      // We should request the entries the peer reported newer103+      if (ourRequest.type === "SYNC_REQUEST") {104+        expect(ourRequest.entryIds.sort()).toEqual(["b", "c"]);105+      }106+      // Peer asks us for entries the peer claimed it had older versions of107+      await peer.send({ type: "SYNC_REQUEST", entryIds: ["a"] });108+109+      const ourEntries = await peer.receive();110+      expect(ourEntries.type).toBe("SYNC_ENTRIES");111+      if (ourEntries.type === "SYNC_ENTRIES") {112+        expect(ourEntries.entries).toEqual({ a: "ct-a" });113+      }114+      await peer.send({ type: "SYNC_ENTRIES", entries: { b: "ct-b-new", c: "ct-c" } });115+116+      const ourTombs = await peer.receive();117+      expect(ourTombs.type).toBe("SYNC_TOMBSTONES");118+      if (ourTombs.type === "SYNC_TOMBSTONES") {119+        // We had x@70, peer had x@60, so we ship x@70120+        expect(ourTombs.tombstones).toEqual({ x: 70 });121+      }122+      await peer.send({ type: "SYNC_TOMBSTONES", tombstones: { y: 99 } });123+124+      const ourDone = await peer.receive();125+      expect(ourDone.type).toBe("SYNC_DONE");126+      await peer.send({ type: "SYNC_DONE" });127+    })();128+129+    const outcome = await initPromise;130+    await peerProtocol;131+132+    expect(outcome.status).toBe("completed");133+    expect(outcome.entriesSent).toBe(1);134+    expect(outcome.entriesReceived).toBe(2);135+    expect(outcome.tombstonesSent).toBe(1);136+    expect(outcome.tombstonesReceived).toBe(1);137+    expect(vault.appliedEntries).toEqual([{ b: "ct-b-new", c: "ct-c" }]);138+    expect(vault.appliedTombstones).toEqual([{ y: 99 }]);139+  });140+141+  it("aborts with vault_mismatch on a peer with a different vault_id", async () => {142+    const [session, peer] = inMemoryPair();143+    const vault = makeVaultAdapter({ vaultId: "vault-A" });144+145+    const initPromise = runChirpSyncInitiator(session, vault);146+    const peerProtocol = (async () => {147+      const hello = await peer.receive();148+      expect(hello.type).toBe("SYNC_HELLO");149+      await peer.send({ type: "SYNC_HELLO", vaultId: "vault-B", deviceId: "evil" });150+      // Peer should receive a SYNC_ERROR and we should stop.151+      const err = await peer.receive();152+      expect(err.type).toBe("SYNC_ERROR");153+    })();154+155+    const outcome = await initPromise;156+    await peerProtocol;157+    expect(outcome.status).toBe("vault_mismatch");158+  });159+160+  it("returns errored when peer sends a wrong frame", async () => {161+    const [session, peer] = inMemoryPair();162+    const vault = makeVaultAdapter({ vaultId: "v" });163+164+    const initPromise = runChirpSyncInitiator(session, vault);165+    void (async () => {166+      await peer.receive(); // hello167+      await peer.send({ type: "SYNC_HELLO", vaultId: "v", deviceId: "p" });168+      await peer.receive(); // our manifest169+      await peer.send({ type: "SYNC_DONE" }); // wrong frame170+    })();171+172+    const outcome = await initPromise;173+    expect(outcome.status).toBe("errored");174+    expect(outcome.error).toMatch(/SYNC_MANIFEST/);175+  });176+});

@@ -0,0 +1,206 @@1+// Chirp Sync — Phase 2 manifest-diff coordinator.2+//3+// After Noise XX is open and the responder's static key has been verified4+// against the paired-device registry, this coordinator drives the5+// manifest exchange:6+//7+//   ->  SYNC_HELLO { vaultId, deviceId }8+//   <-  SYNC_HELLO { vaultId, deviceId }       (mismatch → SYNC_ERROR)9+//   <-> SYNC_MANIFEST { entries, tombstones }  (each side sends one)10+//   ->  SYNC_REQUEST { entryIds }              (entries newer on the peer)11+//   <-  SYNC_ENTRIES { entries }12+//   <-  SYNC_REQUEST { entryIds }              (entries newer on us)13+//   ->  SYNC_ENTRIES { entries }14+//   <-> SYNC_TOMBSTONES { tombstones }15+//   <-> SYNC_DONE16+//17+// Entries are opaque base64 ciphertext — the coordinator never decrypts.18+// Last-write-wins at the entry level by mtime; tombstones override updates19+// with a later deletion timestamp.20+21+import type { ChirpSyncMessage } from "../shared/sync-types";22+import type { ChirpSyncSession } from "../shared/chirp-sync-transport";23+24+const SESSION_TIMEOUT_MS = 30_000;25+26+/**27+ * Vault-side adapter — the coordinator's dependency on the vault layer.28+ * The vault holds the canonical entry/tombstone tables; the coordinator29+ * pulls a manifest, requests entries by id, applies remote entries, and30+ * applies remote tombstones.31+ */32+export interface ChirpSyncVaultAdapter {33+  computeVaultId(): Promise<string>;34+  getOurDeviceId(): Promise<string>;35+  /** Returns the per-entry mtimes for our active entries. */36+  buildEntryManifest(): Promise<Record<string, number>>;37+  /** Returns the per-entry deletion timestamps for our tombstones. */38+  buildTombstoneManifest(): Promise<Record<string, number>>;39+  /** Returns base64 ciphertext for each requested entry id. */40+  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;41+  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */42+  applyEntries(entries: Record<string, string>): Promise<void>;43+  /** Apply incoming tombstones; vault retains the latest deletion ts. */44+  applyTombstones(tombstones: Record<string, number>): Promise<void>;45+}46+47+export interface SyncOutcome {48+  status: "completed" | "vault_mismatch" | "errored" | "timeout";49+  entriesReceived: number;50+  entriesSent: number;51+  tombstonesReceived: number;52+  tombstonesSent: number;53+  error?: string;54+}55+56+/**57+ * Run one sync round as the initiator. The extension is always the58+ * initiator (it is the WebSocket client). The peer's role is symmetric59+ * after SYNC_HELLO, so a corresponding responder helper exists for the60+ * mobile-side coordinator that calls this same protocol the other way.61+ */62+export async function runChirpSyncInitiator(63+  session: ChirpSyncSession,64+  vault: ChirpSyncVaultAdapter,65+): Promise<SyncOutcome> {66+  const result: SyncOutcome = {67+    status: "errored",68+    entriesReceived: 0,69+    entriesSent: 0,70+    tombstonesReceived: 0,71+    tombstonesSent: 0,72+  };73+74+  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);75+  try {76+    const ourVaultId = await vault.computeVaultId();77+    const ourDeviceId = await vault.getOurDeviceId();78+79+    await sendMessage(session, { type: "SYNC_HELLO", vaultId: ourVaultId, deviceId: ourDeviceId });80+    const helloIn = await receiveMessage(session, timer);81+    if (helloIn.type !== "SYNC_HELLO") {82+      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);83+    }84+    if (helloIn.vaultId !== ourVaultId) {85+      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });86+      result.status = "vault_mismatch";87+      return result;88+    }89+90+    const ourEntries = await vault.buildEntryManifest();91+    const ourTombstones = await vault.buildTombstoneManifest();92+    await sendMessage(session, {93+      type: "SYNC_MANIFEST",94+      entries: ourEntries,95+      tombstones: ourTombstones,96+    });97+    const manifestIn = await receiveMessage(session, timer);98+    if (manifestIn.type !== "SYNC_MANIFEST") {99+      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);100+    }101+102+    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);103+    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });104+105+    const requestIn = await receiveMessage(session, timer);106+    if (requestIn.type !== "SYNC_REQUEST") {107+      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);108+    }109+    // Sanity: the peer's request must be a subset of what its manifest110+    // declared as newer. A malformed/over-eager request is a protocol111+    // violation but not a data-safety problem — we just send what they ask.112+113+    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);114+    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });115+    result.entriesSent = Object.keys(ourPayload).length;116+117+    if (wantFromPeer.length > 0) {118+      const entriesIn = await receiveMessage(session, timer);119+      if (entriesIn.type !== "SYNC_ENTRIES") {120+        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);121+      }122+      await vault.applyEntries(entriesIn.entries);123+      result.entriesReceived = Object.keys(entriesIn.entries).length;124+    }125+126+    // Tombstones: each side ships the tombstones that the other does not127+    // yet have a newer record for.128+    const newerTombstonesForPeer: Record<string, number> = {};129+    for (const [id, ts] of Object.entries(ourTombstones)) {130+      const peerTs = manifestIn.tombstones[id] ?? 0;131+      if (ts > peerTs) newerTombstonesForPeer[id] = ts;132+    }133+    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });134+    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;135+136+    const tombstonesIn = await receiveMessage(session, timer);137+    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {138+      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);139+    }140+    await vault.applyTombstones(tombstonesIn.tombstones);141+    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;142+143+    await sendMessage(session, { type: "SYNC_DONE" });144+    const done = await receiveMessage(session, timer);145+    if (done.type !== "SYNC_DONE") {146+      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);147+    }148+    result.status = "completed";149+    return result;150+  } catch (err) {151+    if ((err as Error).message === "chirp_sync_timeout") {152+      result.status = "timeout";153+      result.error = "session timed out after 30s";154+      return result;155+    }156+    return errorOutcome(result, (err as Error).message);157+  } finally {158+    timer.cancel();159+    await session.close().catch(() => undefined);160+  }161+}162+163+function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {164+  const want: string[] = [];165+  for (const [id, ts] of Object.entries(remote)) {166+    const ours = local[id];167+    if (ours === undefined || ts > ours) want.push(id);168+  }169+  return want;170+}171+172+interface Timer {173+  signal: AbortSignal;174+  cancel(): void;175+}176+177+function scheduleTimeout(ms: number): Timer {178+  const ctrl = new AbortController();179+  const t = setTimeout(() => ctrl.abort(new Error("chirp_sync_timeout")), ms);180+  return {181+    signal: ctrl.signal,182+    cancel: () => clearTimeout(t),183+  };184+}185+186+async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {187+  session.send(new TextEncoder().encode(JSON.stringify(msg)));188+}189+190+async function receiveMessage(session: ChirpSyncSession, timer: Timer): Promise<ChirpSyncMessage> {191+  const pending = session.receive();192+  const aborted = new Promise<never>((_, reject) => {193+    if (timer.signal.aborted) reject(new Error("chirp_sync_timeout"));194+    timer.signal.addEventListener("abort", () => reject(new Error("chirp_sync_timeout")));195+  });196+  const bytes = await Promise.race([pending, aborted]);197+  const parsed = JSON.parse(new TextDecoder().decode(bytes));198+  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");199+  return parsed as ChirpSyncMessage;200+}201+202+function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {203+  outcome.status = "errored";204+  outcome.error = error;205+  return outcome;206+}