sync-coordinator.ts

1/**2 * Blob sync session coordinator.3 *4 * Replaces the per-entry SYNC_HELLO/MANIFEST/REQUEST/ENTRIES/DONE state machine5 * with a single bidirectional vault-blob exchange. The encrypted vault blob is6 * already AES-256-GCM ciphertext under the user's master password, so the7 * Noise NK transport adds a second layer and the sync layer never sees8 * plaintext.9 *10 * Last-write-wins at the vault level: an incoming blob is applied only if its11 * `sent_at` is strictly newer than the local `peach_last_sync_at`. Vault12 * history snapshots before every apply so the user can restore an older state13 * if a sync overwrites the wrong vault.14 */1516import { storageGet, storageSet } from "../shared/storage";17import { decodeSyncMessage, encodeSyncMessage } from "../shared/sync-wire";18import type { SyncMessage } from "../shared/sync-types";19import { takeVaultSnapshot } from "./vault";2021const LAST_SYNC_KEY = "peach_last_sync_at";22const SYNC_TIMEOUT_MS = 30_000;2324export interface SyncCoordinatorDeps {25  isPro: () => Promise<boolean>;26  isUnlocked: () => boolean;27  getVaultBlobs: () => Promise<{ vault_a: string; vault_b: string }>;28}2930export interface SyncSessionTransport {31  send: (frame: string) => void | Promise<void>;32  onFrame: (handler: (frame: string) => void) => () => void;33  onError: (handler: (message: string) => void) => () => void;34  onClose: (handler: () => void) => () => void;35  close: () => void;36}3738export type SyncRole = "initiator" | "responder";3940export interface SyncSessionResult {41  status: "completed" | "errored" | "closed";42  error?: string;43}4445/**46 * Persist an incoming vault blob if its `sent_at` beats the local last-sync47 * timestamp. Returns 'applied' when the blob was written, 'ignored' when the48 * incoming was older. Always snapshots BEFORE writing so the user can roll49 * back via vault history if a sync overwrites the wrong state.50 */51export async function applyIncomingVault(52  incomingVaultA: string,53  incomingVaultB: string,54  sentAt: number,55): Promise<"applied" | "ignored"> {56  const lastRaw = await storageGet(LAST_SYNC_KEY);57  const lastTs = lastRaw ? Number.parseInt(lastRaw, 10) : 0;5859  if (!Number.isFinite(sentAt) || sentAt <= lastTs) {60    return "ignored";61  }6263  // History capture happens BEFORE the write. Failures are swallowed inside64  // takeVaultSnapshot — sync must never block on history.65  await takeVaultSnapshot("sync_received");6667  await storageSet("peach_vault_a", incomingVaultA);68  await storageSet("peach_vault_b", incomingVaultB);69  await storageSet(LAST_SYNC_KEY, String(sentAt));7071  return "applied";72}7374/**75 * Run one bidirectional sync session over an authenticated transport.76 *77 * Wire flow:78 *   1. Initiator sends SYNC_VAULT_SEND with its blobs + sent_at79 *   2. Responder receives, applies if newer, replies with SYNC_VAULT_ACK80 *   3. Responder then sends ITS own SYNC_VAULT_SEND so the initiator gets81 *      whichever vault is newer82 *   4. Initiator receives, applies if newer, replies with SYNC_VAULT_ACK83 *   5. Both sides close84 *85 * Either side can short-circuit with SYNC_ERROR (pro_required, vault_locked,86 * timeout). The session times out after 30 seconds total.87 */88export async function runBlobSyncSession(89  role: SyncRole,90  transport: SyncSessionTransport,91  deps: SyncCoordinatorDeps,92): Promise<SyncSessionResult> {93  if (!(await deps.isPro())) {94    await transport.send(95      encodeSyncMessage({ type: "SYNC_ERROR", reason: "pro_required" }),96    );97    transport.close();98    return { status: "errored", error: "pro_required" };99  }100  if (!deps.isUnlocked()) {101    await transport.send(102      encodeSyncMessage({ type: "SYNC_ERROR", reason: "vault_locked" }),103    );104    transport.close();105    return { status: "errored", error: "vault_locked" };106  }107108  return new Promise<SyncSessionResult>((resolve) => {109    let settled = false;110    const unsubscribers: Array<() => void> = [];111112    const finish = (result: SyncSessionResult): void => {113      if (settled) return;114      settled = true;115      clearTimeout(timeoutHandle);116      for (const fn of unsubscribers) {117        try {118          fn();119        } catch {120          // unsubscribe must not throw121        }122      }123      try {124        transport.close();125      } catch {126        // close must not throw127      }128      resolve(result);129    };130131    const timeoutHandle = setTimeout(132      () => finish({ status: "errored", error: "timeout" }),133      SYNC_TIMEOUT_MS,134    );135136    unsubscribers.push(transport.onClose(() => finish({ status: "closed" })));137    unsubscribers.push(138      transport.onError((message) =>139        finish({ status: "errored", error: message || "transport_error" }),140      ),141    );142143    unsubscribers.push(144      transport.onFrame((frame) => {145        void (async () => {146          let msg: SyncMessage;147          try {148            msg = decodeSyncMessage(frame);149          } catch {150            await transport.send(151              encodeSyncMessage({ type: "SYNC_ERROR", reason: "invalid_frame" }),152            );153            finish({ status: "errored", error: "invalid_frame" });154            return;155          }156157          if (msg.type === "SYNC_VAULT_SEND") {158            try {159              await applyIncomingVault(msg.vault_a, msg.vault_b, msg.sent_at);160            } catch (err) {161              const reason = err instanceof Error ? err.message : "apply_failed";162              await transport.send(163                encodeSyncMessage({ type: "SYNC_ERROR", reason }),164              );165              finish({ status: "errored", error: reason });166              return;167            }168169            await transport.send(170              encodeSyncMessage({171                type: "SYNC_VAULT_ACK",172                received_at: Date.now(),173              }),174            );175176            if (role === "responder") {177              // Send our blob back so the initiator ends up with whichever is newer.178              const blobs = await deps.getVaultBlobs();179              await transport.send(180                encodeSyncMessage({181                  type: "SYNC_VAULT_SEND",182                  vault_a: blobs.vault_a,183                  vault_b: blobs.vault_b,184                  sent_at: Date.now(),185                }),186              );187              // Now wait for initiator's ACK.188            } else {189              // Initiator received responder's vault and ACK'd — done.190              finish({ status: "completed" });191            }192          } else if (msg.type === "SYNC_VAULT_ACK") {193            // Peer received our vault. If we are the responder we still need194            // to wait for the initiator's ACK after we send our blob back; in195            // practice that flow ends via the initiator's SYNC_VAULT_SEND196            // branch above. The simple case: we sent first, peer ACK'd, we197            // can close.198            if (role === "initiator") {199              // Wait — initiator only completes after receiving responder's200              // SYNC_VAULT_SEND. A naked ACK without the responder vault means201              // the responder chose not to send anything back; treat as done.202              finish({ status: "completed" });203            } else {204              finish({ status: "completed" });205            }206          } else if (msg.type === "SYNC_ERROR") {207            finish({ status: "errored", error: msg.reason });208          }209          // PAIR_* messages are out-of-band for the data-sync session.210        })().catch((err) => {211          finish({212            status: "errored",213            error: err instanceof Error ? err.message : String(err),214          });215        });216      }),217    );218219    if (role === "initiator") {220      void (async () => {221        try {222          const blobs = await deps.getVaultBlobs();223          await transport.send(224            encodeSyncMessage({225              type: "SYNC_VAULT_SEND",226              vault_a: blobs.vault_a,227              vault_b: blobs.vault_b,228              sent_at: Date.now(),229            }),230          );231        } catch (err) {232          finish({233            status: "errored",234            error: err instanceof Error ? err.message : String(err),235          });236        }237      })();238    }239  });240}241242/**243 * Build a transport adapter that bridges a native-host connection (via sync.ts)244 * into the SyncSessionTransport interface runBlobSyncSession consumes. The245 * adapter scopes frame/error/disconnect events to a single connection id so246 * concurrent sync sessions on different connections don't cross-talk.247 */248export interface NativeSyncTransportBindings {249  sendSyncData: (connectionId: string, data: string) => void;250  onSyncMessage: (251    handler: (connectionId: string, data: string) => void,252  ) => () => void;253  onError: (254    handler: (message: string, connectionId?: string) => void,255  ) => () => void;256  onDisconnected: (handler: (connectionId?: string) => void) => () => void;257  disconnectFromPeer: (connectionId: string) => void;258}259260async function loadNativeSyncBindings(): Promise<NativeSyncTransportBindings> {261  const sync = await import("./sync");262  return {263    sendSyncData: sync.sendSyncData,264    onSyncMessage: sync.onSyncMessage,265    onError: sync.onError,266    onDisconnected: sync.onDisconnected,267    disconnectFromPeer: sync.disconnectFromPeer,268  };269}270271/**272 * Wait for the next sync frame on a specific connection or reject after273 * `timeoutMs`. Used by the pairing flow to await a single PAIR_ACCEPT /274 * PAIR_REJECT response. Resolves with the decoded SyncMessage; throws on275 * timeout, decode failure, or transport disconnect.276 */277export async function waitForPairingResponse(278  connectionId: string,279  timeoutMs: number,280  bindings?: NativeSyncTransportBindings,281): Promise<SyncMessage> {282  const ready = bindings ?? (await loadNativeSyncBindings());283  return new Promise<SyncMessage>((resolve, reject) => {284    let settled = false;285    const cleanup: Array<() => void> = [];286    const finish = (action: () => void) => {287      if (settled) return;288      settled = true;289      for (const fn of cleanup) {290        try {291          fn();292        } catch {293          // unsubscribe must not throw294        }295      }296      action();297    };298299    const handle = setTimeout(300      () => finish(() => reject(new Error("pairing_response_timeout"))),301      timeoutMs,302    );303    cleanup.push(() => clearTimeout(handle));304305    cleanup.push(306      ready.onSyncMessage((id, frame) => {307        if (id !== connectionId) return;308        try {309          const msg = decodeSyncMessage(frame);310          finish(() => resolve(msg));311        } catch (err) {312          finish(() => reject(err instanceof Error ? err : new Error(String(err))));313        }314      }),315    );316317    cleanup.push(318      ready.onDisconnected((id) => {319        if (id === undefined || id === connectionId) {320          finish(() => reject(new Error("pairing_connection_closed")));321        }322      }),323    );324325    cleanup.push(326      ready.onError((message, id) => {327        if (id === undefined || id === connectionId) {328          finish(() => reject(new Error(message || "pairing_transport_error")));329        }330      }),331    );332  });333}334335export async function createNativeSyncTransport(336  connectionId: string,337  bindings?: NativeSyncTransportBindings,338): Promise<SyncSessionTransport> {339  const ready = bindings ?? (await loadNativeSyncBindings());340  return {341    send: (frame) => ready.sendSyncData(connectionId, frame),342    onFrame: (handler) =>343      ready.onSyncMessage((id, frame) => {344        if (id === connectionId) handler(frame);345      }),346    onError: (handler) =>347      ready.onError((message, id) => {348        if (id === undefined || id === connectionId) handler(message);349      }),350    onClose: (handler) =>351      ready.onDisconnected((id) => {352        if (id === undefined || id === connectionId) handler();353      }),354    close: () => ready.disconnectFromPeer(connectionId),355  };356}