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, SignedDeviceManifest } from "../shared/sync-types";22import type { ChirpSyncSession } from "../shared/chirp-sync-transport";2324const SESSION_TIMEOUT_MS = 30_000;2526// Sentinel signature used when an adapter can't build a real manifest (e.g.27// the master key isn't accessible from the coordinator context). The peer28// recognises this exact value and skips signature verification. 64 hex29// chars of zero is impossible to produce as a real Ed25519 signature for a30// fresh body, so collisions are negligible.31const EMPTY_MANIFEST_PLACEHOLDER: SignedDeviceManifest = {32  body: { manifestVersion: 1, issuerDeviceId: "", issuedAt: 0, devices: [] },33  signatureHex: "0".repeat(128),34};3536/**37 * Vault-side adapter — the coordinator's dependency on the vault layer.38 * The vault holds the canonical entry/tombstone tables; the coordinator39 * pulls a manifest, requests entries by id, applies remote entries, and40 * applies remote tombstones.41 */42export interface ChirpSyncVaultAdapter {43  computeVaultId(): Promise<string>;44  getOurDeviceId(): Promise<string>;45  /** Returns the per-entry mtimes for our active entries. */46  buildEntryManifest(): Promise<Record<string, number>>;47  /** Returns the per-entry deletion timestamps for our tombstones. */48  buildTombstoneManifest(): Promise<Record<string, number>>;49  /** Returns base64 ciphertext for each requested entry id. */50  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;51  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */52  applyEntries(entries: Record<string, string>): Promise<void>;53  /** Apply incoming tombstones; vault retains the latest deletion ts. */54  applyTombstones(tombstones: Record<string, number>): Promise<void>;55  /**56   * Optional — v2 brief §4.3 device manifest exchange. The extension (as57   * the trust anchor for the device graph) returns a signed manifest58   * enumerating every paired device it knows about; the peer verifies it59   * against the extension's known Ed25519 signing pubkey and upserts any60   * newly-vouched devices into its own registry.61   *62   * When this method is undefined, the coordinator skips the manifest63   * step entirely. Both ends must opt in for the exchange to occur.64   */65  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;66  /** v2 brief §4.3 — handle a manifest received from the peer. */67  applyIncomingDeviceManifest?(manifest: SignedDeviceManifest): Promise<void>;68}6970export interface SyncOutcome {71  status: "completed" | "vault_mismatch" | "errored" | "timeout";72  entriesReceived: number;73  entriesSent: number;74  tombstonesReceived: number;75  tombstonesSent: number;76  error?: string;77}7879/**80 * Run one sync round as the initiator. The extension is always the81 * initiator (it is the WebSocket client). The peer's role is symmetric82 * after SYNC_HELLO, so a corresponding responder helper exists for the83 * mobile-side coordinator that calls this same protocol the other way.84 */85export async function runChirpSyncInitiator(86  session: ChirpSyncSession,87  vault: ChirpSyncVaultAdapter,88): Promise<SyncOutcome> {89  const result: SyncOutcome = {90    status: "errored",91    entriesReceived: 0,92    entriesSent: 0,93    tombstonesReceived: 0,94    tombstonesSent: 0,95  };9697  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);98  try {99    const ourVaultId = await vault.computeVaultId();100    const ourDeviceId = await vault.getOurDeviceId();101102    await sendMessage(session, { type: "SYNC_HELLO", vaultId: ourVaultId, deviceId: ourDeviceId });103    const helloIn = await receiveMessage(session, timer);104    if (helloIn.type !== "SYNC_HELLO") {105      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);106    }107    if (helloIn.vaultId !== ourVaultId) {108      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });109      result.status = "vault_mismatch";110      return result;111    }112113    const ourEntries = await vault.buildEntryManifest();114    const ourTombstones = await vault.buildTombstoneManifest();115    await sendMessage(session, {116      type: "SYNC_MANIFEST",117      entries: ourEntries,118      tombstones: ourTombstones,119    });120    const manifestIn = await receiveMessage(session, timer);121    if (manifestIn.type !== "SYNC_MANIFEST") {122      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);123    }124125    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);126    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });127128    const requestIn = await receiveMessage(session, timer);129    if (requestIn.type !== "SYNC_REQUEST") {130      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);131    }132    // Sanity: the peer's request must be a subset of what its manifest133    // declared as newer. A malformed/over-eager request is a protocol134    // violation but not a data-safety problem — we just send what they ask.135136    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);137    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });138    result.entriesSent = Object.keys(ourPayload).length;139140    if (wantFromPeer.length > 0) {141      const entriesIn = await receiveMessage(session, timer);142      if (entriesIn.type !== "SYNC_ENTRIES") {143        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);144      }145      await vault.applyEntries(entriesIn.entries);146      result.entriesReceived = Object.keys(entriesIn.entries).length;147    }148149    // Tombstones: each side ships the tombstones that the other does not150    // yet have a newer record for.151    const newerTombstonesForPeer: Record<string, number> = {};152    for (const [id, ts] of Object.entries(ourTombstones)) {153      const peerTs = manifestIn.tombstones[id] ?? 0;154      if (ts > peerTs) newerTombstonesForPeer[id] = ts;155    }156    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });157    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;158159    const tombstonesIn = await receiveMessage(session, timer);160    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {161      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);162    }163    await vault.applyTombstones(tombstonesIn.tombstones);164    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;165166    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends167    // must expose the adapter methods. Disabling on either side cleanly168    // bypasses the step. The exchange is symmetric (each side sends one169    // and applies the other's) even though in the current device graph170    // only the extension actually has anything to vouch for; mobiles still171    // produce a signed manifest, just with an empty `devices[]`, so the172    // protocol stays uniform.173    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {174      const ours = await vault.buildOutgoingDeviceManifest();175      if (ours) {176        await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });177      } else {178        // No manifest to send (e.g. master key unavailable) — still emit a179        // protocol frame so the peer's recv doesn't hang. Empty body is180        // serialised by the adapter when present; if it returned null we181        // synthesise a no-op shape the receiver can ignore.182        await sendMessage(session, {183          type: "SYNC_DEVICE_MANIFEST",184          manifest: EMPTY_MANIFEST_PLACEHOLDER,185        });186      }187      const manifestIn = await receiveMessage(session, timer);188      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {189        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);190      }191      if (manifestIn.manifest && manifestIn.manifest.signatureHex !== EMPTY_MANIFEST_PLACEHOLDER.signatureHex) {192        await vault.applyIncomingDeviceManifest(manifestIn.manifest);193      }194    }195196    await sendMessage(session, { type: "SYNC_DONE" });197    const done = await receiveMessage(session, timer);198    if (done.type !== "SYNC_DONE") {199      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);200    }201    result.status = "completed";202    return result;203  } catch (err) {204    if ((err as Error).message === "chirp_sync_timeout") {205      result.status = "timeout";206      result.error = "session timed out after 30s";207      return result;208    }209    return errorOutcome(result, (err as Error).message);210  } finally {211    timer.cancel();212    await session.close().catch(() => undefined);213  }214}215216function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {217  const want: string[] = [];218  for (const [id, ts] of Object.entries(remote)) {219    const ours = local[id];220    if (ours === undefined || ts > ours) want.push(id);221  }222  return want;223}224225interface Timer {226  signal: AbortSignal;227  cancel(): void;228}229230function scheduleTimeout(ms: number): Timer {231  const ctrl = new AbortController();232  const t = setTimeout(() => ctrl.abort(new Error("chirp_sync_timeout")), ms);233  return {234    signal: ctrl.signal,235    cancel: () => clearTimeout(t),236  };237}238239async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {240  session.send(new TextEncoder().encode(JSON.stringify(msg)));241}242243async function receiveMessage(session: ChirpSyncSession, timer: Timer): Promise<ChirpSyncMessage> {244  const pending = session.receive();245  const aborted = new Promise<never>((_, reject) => {246    if (timer.signal.aborted) reject(new Error("chirp_sync_timeout"));247    timer.signal.addEventListener("abort", () => reject(new Error("chirp_sync_timeout")));248  });249  const bytes = await Promise.race([pending, aborted]);250  const parsed = JSON.parse(new TextDecoder().decode(bytes));251  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");252  return parsed as ChirpSyncMessage;253}254255function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {256  outcome.status = "errored";257  outcome.error = error;258  return outcome;259}