chirp-sync-coordinator.ts

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.2021import type { ChirpSyncMessage } from "../shared/sync-types";22import type { ChirpSyncSession } from "../shared/chirp-sync-transport";2324const SESSION_TIMEOUT_MS = 30_000;2526/**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 */32export 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}4647export 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}5556/**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 */62export 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  };7374  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);75  try {76    const ourVaultId = await vault.computeVaultId();77    const ourDeviceId = await vault.getOurDeviceId();7879    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    }8990    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    }101102    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);103    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });104105    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.112113    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);114    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });115    result.entriesSent = Object.keys(ourPayload).length;116117    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    }125126    // 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;135136    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;142143    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}162163function 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}171172interface Timer {173  signal: AbortSignal;174  cancel(): void;175}176177function 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}185186async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {187  session.send(new TextEncoder().encode(JSON.stringify(msg)));188}189190async 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}201202function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {203  outcome.status = "errored";204  outcome.error = error;205  return outcome;206}