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 on the wire; the169    // production adapter still treats the extension as the trust anchor and170    // only applies extension-issued manifests.171    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {172      const ours = await vault.buildOutgoingDeviceManifest();173      if (ours) {174        await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });175      } else {176        // No manifest to send (e.g. master key unavailable) — still emit a177        // protocol frame so the peer's recv doesn't hang. Empty body is178        // serialised by the adapter when present; if it returned null we179        // synthesise a no-op shape the receiver can ignore.180        await sendMessage(session, {181          type: "SYNC_DEVICE_MANIFEST",182          manifest: EMPTY_MANIFEST_PLACEHOLDER,183        });184      }185      const manifestIn = await receiveMessage(session, timer);186      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {187        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);188      }189      if (manifestIn.manifest && manifestIn.manifest.signatureHex !== EMPTY_MANIFEST_PLACEHOLDER.signatureHex) {190        await vault.applyIncomingDeviceManifest(manifestIn.manifest);191      }192    }193194    await sendMessage(session, { type: "SYNC_DONE" });195    const done = await receiveMessage(session, timer);196    if (done.type !== "SYNC_DONE") {197      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);198    }199    result.status = "completed";200    return result;201  } catch (err) {202    if ((err as Error).message === "chirp_sync_timeout") {203      result.status = "timeout";204      result.error = "session timed out after 30s";205      return result;206    }207    return errorOutcome(result, (err as Error).message);208  } finally {209    timer.cancel();210    await session.close().catch(() => undefined);211  }212}213214function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {215  const want: string[] = [];216  for (const [id, ts] of Object.entries(remote)) {217    const ours = local[id];218    if (ours === undefined || ts > ours) want.push(id);219  }220  return want;221}222223interface Timer {224  signal: AbortSignal;225  cancel(): void;226}227228function scheduleTimeout(ms: number): Timer {229  const ctrl = new AbortController();230  const t = setTimeout(() => ctrl.abort(new Error("chirp_sync_timeout")), ms);231  return {232    signal: ctrl.signal,233    cancel: () => clearTimeout(t),234  };235}236237async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {238  session.send(new TextEncoder().encode(JSON.stringify(msg)));239}240241async function receiveMessage(session: ChirpSyncSession, timer: Timer): Promise<ChirpSyncMessage> {242  const pending = session.receive();243  const aborted = new Promise<never>((_, reject) => {244    if (timer.signal.aborted) reject(new Error("chirp_sync_timeout"));245    timer.signal.addEventListener("abort", () => reject(new Error("chirp_sync_timeout")));246  });247  const bytes = await Promise.race([pending, aborted]);248  const parsed = JSON.parse(new TextDecoder().decode(bytes));249  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");250  return parsed as ChirpSyncMessage;251}252253function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {254  outcome.status = "errored";255  outcome.error = error;256  return outcome;257}