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 vault's monotonic version (`peach_last_modified_at`). */46  getLocalVersion(): Promise<number>;47  /** Returns the per-entry mtimes for our active entries. */48  buildEntryManifest(): Promise<Record<string, number>>;49  /** Returns the per-entry deletion timestamps for our tombstones. */50  buildTombstoneManifest(): Promise<Record<string, number>>;51  /** Returns base64 ciphertext for each requested entry id. */52  getEncryptedEntries(entryIds: string[]): Promise<Record<string, string>>;53  /** Apply incoming encrypted entries; vault decides per-entry overwrite. */54  applyEntries(entries: Record<string, string>): Promise<void>;55  /** Apply incoming tombstones; vault retains the latest deletion ts. */56  applyTombstones(tombstones: Record<string, number>): Promise<void>;57  /**58   * Optional — v2 brief §4.3 device manifest exchange. The extension (as59   * the trust anchor for the device graph) returns a signed manifest60   * enumerating every paired device it knows about; the peer verifies it61   * against the extension's known Ed25519 signing pubkey and upserts any62   * newly-vouched devices into its own registry.63   *64   * When this method is undefined, the coordinator skips the manifest65   * step entirely. Both ends must opt in for the exchange to occur.66   */67  buildOutgoingDeviceManifest?(): Promise<SignedDeviceManifest | null>;68  /** v2 brief §4.3 — handle a manifest received from the peer. */69  applyIncomingDeviceManifest?(manifest: SignedDeviceManifest): Promise<void>;70}7172export interface SyncOutcome {73  status: "completed" | "vault_mismatch" | "errored" | "timeout";74  entriesReceived: number;75  entriesSent: number;76  tombstonesReceived: number;77  tombstonesSent: number;78  error?: string;79}8081/**82 * Run one sync round as the initiator. The extension is always the83 * initiator (it is the WebSocket client). The peer's role is symmetric84 * after SYNC_HELLO, so a corresponding responder helper exists for the85 * mobile-side coordinator that calls this same protocol the other way.86 */87export async function runChirpSyncInitiator(88  session: ChirpSyncSession,89  vault: ChirpSyncVaultAdapter,90): Promise<SyncOutcome> {91  const result: SyncOutcome = {92    status: "errored",93    entriesReceived: 0,94    entriesSent: 0,95    tombstonesReceived: 0,96    tombstonesSent: 0,97  };9899  const timer = scheduleTimeout(SESSION_TIMEOUT_MS);100  try {101    const ourVaultId = await vault.computeVaultId();102    const ourDeviceId = await vault.getOurDeviceId();103    const localVersion = await vault.getLocalVersion();104    const localTime = Date.now();105106    await sendMessage(session, {107      type: "SYNC_HELLO",108      vaultId: ourVaultId,109      deviceId: ourDeviceId,110      version: localVersion,111      deviceTime: localTime,112    });113    const helloIn = await receiveMessage(session, timer);114    if (helloIn.type !== "SYNC_HELLO") {115      return errorOutcome(result, `expected SYNC_HELLO, got ${helloIn.type}`);116    }117    if (helloIn.vaultId !== ourVaultId) {118      await sendMessage(session, { type: "SYNC_ERROR", error: "vault_id_mismatch" });119      result.status = "vault_mismatch";120      return result;121    }122123    if (Math.abs(localTime - (helloIn.deviceTime ?? 0)) > 120_000) {124      console.warn(`[chirp-sync] clock skew detected: local=${localTime} remote=${helloIn.deviceTime ?? "unknown"}`);125    }126127    // Equal versions — nothing to exchange.128    if (typeof helloIn.version === "number" && helloIn.version === localVersion) {129      await sendMessage(session, { type: "SYNC_DONE", entriesReceived: 0, entriesSent: 0 });130      const done = await receiveMessage(session, timer);131      if (done.type !== "SYNC_DONE") {132        return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);133      }134      result.status = "completed";135      return result;136    }137138    const ourEntries = await vault.buildEntryManifest();139    const ourTombstones = await vault.buildTombstoneManifest();140    await sendMessage(session, {141      type: "SYNC_MANIFEST",142      entries: ourEntries,143      tombstones: ourTombstones,144    });145    const manifestIn = await receiveMessage(session, timer);146    if (manifestIn.type !== "SYNC_MANIFEST") {147      return errorOutcome(result, `expected SYNC_MANIFEST, got ${manifestIn.type}`);148    }149150    const wantFromPeer = diffNewer(manifestIn.entries, ourEntries);151    await sendMessage(session, { type: "SYNC_REQUEST", entryIds: wantFromPeer });152153    const requestIn = await receiveMessage(session, timer);154    if (requestIn.type !== "SYNC_REQUEST") {155      return errorOutcome(result, `expected SYNC_REQUEST, got ${requestIn.type}`);156    }157    // Sanity: the peer's request must be a subset of what its manifest158    // declared as newer. A malformed/over-eager request is a protocol159    // violation but not a data-safety problem — we just send what they ask.160161    const ourPayload = await vault.getEncryptedEntries(requestIn.entryIds);162    await sendMessage(session, { type: "SYNC_ENTRIES", entries: ourPayload });163    result.entriesSent = Object.keys(ourPayload).length;164165    if (wantFromPeer.length > 0) {166      const entriesIn = await receiveMessage(session, timer);167      if (entriesIn.type !== "SYNC_ENTRIES") {168        return errorOutcome(result, `expected SYNC_ENTRIES, got ${entriesIn.type}`);169      }170      await vault.applyEntries(entriesIn.entries);171      result.entriesReceived = Object.keys(entriesIn.entries).length;172    }173174    // Tombstones: each side ships the tombstones that the other does not175    // yet have a newer record for.176    const newerTombstonesForPeer: Record<string, number> = {};177    for (const [id, ts] of Object.entries(ourTombstones)) {178      const peerTs = manifestIn.tombstones[id] ?? 0;179      if (ts > peerTs) newerTombstonesForPeer[id] = ts;180    }181    await sendMessage(session, { type: "SYNC_TOMBSTONES", tombstones: newerTombstonesForPeer });182    result.tombstonesSent = Object.keys(newerTombstonesForPeer).length;183184    const tombstonesIn = await receiveMessage(session, timer);185    if (tombstonesIn.type !== "SYNC_TOMBSTONES") {186      return errorOutcome(result, `expected SYNC_TOMBSTONES, got ${tombstonesIn.type}`);187    }188    await vault.applyTombstones(tombstonesIn.tombstones);189    result.tombstonesReceived = Object.keys(tombstonesIn.tombstones).length;190191    // v2 brief §5.6 / §4.3 — device manifest exchange. Optional: both ends192    // must expose the adapter methods. Disabling on either side cleanly193    // bypasses the step. The exchange is symmetric on the wire; the194    // production adapter still treats the extension as the trust anchor and195    // only applies extension-issued manifests.196    if (vault.buildOutgoingDeviceManifest && vault.applyIncomingDeviceManifest) {197      const ours = await vault.buildOutgoingDeviceManifest();198      if (ours) {199        await sendMessage(session, { type: "SYNC_DEVICE_MANIFEST", manifest: ours });200      } else {201        // No manifest to send (e.g. master key unavailable) — still emit a202        // protocol frame so the peer's recv doesn't hang. Empty body is203        // serialised by the adapter when present; if it returned null we204        // synthesise a no-op shape the receiver can ignore.205        await sendMessage(session, {206          type: "SYNC_DEVICE_MANIFEST",207          manifest: EMPTY_MANIFEST_PLACEHOLDER,208        });209      }210      const manifestIn = await receiveMessage(session, timer);211      if (manifestIn.type !== "SYNC_DEVICE_MANIFEST") {212        return errorOutcome(result, `expected SYNC_DEVICE_MANIFEST, got ${manifestIn.type}`);213      }214      if (manifestIn.manifest && manifestIn.manifest.signatureHex !== EMPTY_MANIFEST_PLACEHOLDER.signatureHex) {215        await vault.applyIncomingDeviceManifest(manifestIn.manifest);216      }217    }218219    await sendMessage(session, {220      type: "SYNC_DONE",221      entriesReceived: result.entriesReceived,222      entriesSent: result.entriesSent,223    });224    const done = await receiveMessage(session, timer);225    if (done.type !== "SYNC_DONE") {226      return errorOutcome(result, `expected SYNC_DONE, got ${done.type}`);227    }228    result.status = "completed";229    return result;230  } catch (err) {231    if ((err as Error).message === "chirp_sync_timeout") {232      result.status = "timeout";233      result.error = "session timed out after 30s";234      return result;235    }236    return errorOutcome(result, (err as Error).message);237  } finally {238    // Session lifetime is owned by the runner (chirp-sync-runner.ts), which239    // closes it in its own finally block. The coordinator must not close it240    // here — the runner still needs to record the outcome after we return.241    timer.cancel();242  }243}244245function diffNewer(remote: Record<string, number>, local: Record<string, number>): string[] {246  const want: string[] = [];247  for (const [id, ts] of Object.entries(remote)) {248    const ours = local[id];249    if (ours === undefined || ts > ours) want.push(id);250  }251  return want;252}253254interface Timer {255  signal: AbortSignal;256  cancel(): void;257}258259function scheduleTimeout(ms: number): Timer {260  const ctrl = new AbortController();261  const t = setTimeout(() => ctrl.abort(new Error("chirp_sync_timeout")), ms);262  return {263    signal: ctrl.signal,264    cancel: () => clearTimeout(t),265  };266}267268async function sendMessage(session: ChirpSyncSession, msg: ChirpSyncMessage): Promise<void> {269  session.send(new TextEncoder().encode(JSON.stringify(msg)));270}271272async function receiveMessage(session: ChirpSyncSession, timer: Timer): Promise<ChirpSyncMessage> {273  const pending = session.receive();274  const aborted = new Promise<never>((_, reject) => {275    if (timer.signal.aborted) reject(new Error("chirp_sync_timeout"));276    timer.signal.addEventListener("abort", () => reject(new Error("chirp_sync_timeout")));277  });278  const bytes = await Promise.race([pending, aborted]);279  const parsed = JSON.parse(new TextDecoder().decode(bytes));280  if (typeof parsed?.type !== "string") throw new Error("malformed SyncMessage");281  return parsed as ChirpSyncMessage;282}283284function errorOutcome(outcome: SyncOutcome, error: string): SyncOutcome {285  outcome.status = "errored";286  outcome.error = error;287  return outcome;288}