sync-coordinator.ts

1/**2 * Dormant SYN-001 coordinator helpers.3 *4 * This module owns extension-side sync preflight and cancellation rules before5 * any live background path enables native messaging. Entry payload exchange is6 * intentionally left to the next phase so the encrypted-entry envelope can be7 * reviewed on its own.8 */910import {11  buildConflictedCopy,12  buildSyncManifest,13  chooseConflictWinner,14  diffSyncManifests,15  selectEntriesForSync,16} from "../shared/sync-protocol";17import type { SyncDiff } from "../shared/sync-protocol";18import {19  decryptSyncEntry,20  encryptSyncEntry,21  importSyncEntryKey,22} from "../shared/sync-entry-payload";23import type { SyncManifestEntry, SyncMessage } from "../shared/sync-types";24import { decodeSyncMessage, encodeSyncMessage } from "../shared/sync-wire";25import type { VaultEntry } from "../shared/types";2627export type SyncPreflightError = "pro_required" | "vault_locked";2829export interface SyncPreflightResult {30  ok: boolean;31  error?: SyncPreflightError;32}3334export interface SyncSnapshot {35  entries: VaultEntry[];36  manifest: SyncManifestEntry[];37}3839export interface SyncPlan {40  localManifest: SyncManifestEntry[];41  diff: SyncDiff;42}4344export interface SyncCoordinatorDeps {45  isProUser: () => Promise<boolean>;46  isUnlocked: () => boolean;47  getEntries: () => VaultEntry[];48  restoreEntries: (49    entries: VaultEntry[],50    mode: "replace" | "merge"51  ) => Promise<{ added: number; updated: number; skipped: number }>;52  sendSyncData: (connectionId: string, data: string) => void;53}5455async function loadDefaultDeps(): Promise<SyncCoordinatorDeps> {56  const [license, sync, vault] = await Promise.all([57    import("./license"),58    import("./sync"),59    import("./vault"),60  ]);61  return {62    isProUser: license.isProUser,63    isUnlocked: vault.isUnlocked,64    getEntries: vault.getEntries,65    restoreEntries: vault.restoreEntries,66    sendSyncData: sync.sendSyncData,67  };68}6970export interface NativeSyncTransportBindings {71  sendSyncData: (connectionId: string, data: string) => void;72  onSyncMessage: (73    handler: (connectionId: string, data: string) => void74  ) => () => void;75  onError: (76    handler: (message: string, connectionId?: string) => void77  ) => () => void;78  onDisconnected: (handler: (connectionId?: string) => void) => () => void;79  disconnectFromPeer: (connectionId: string) => void;80  getNativeSessionKey: (connectionId: string) => Promise<Uint8Array>;81}8283async function loadNativeSyncBindings(): Promise<NativeSyncTransportBindings> {84  const sync = await import("./sync");85  return {86    sendSyncData: sync.sendSyncData,87    onSyncMessage: sync.onSyncMessage,88    onError: sync.onError,89    onDisconnected: sync.onDisconnected,90    disconnectFromPeer: sync.disconnectFromPeer,91    getNativeSessionKey: sync.getNativeSessionKey,92  };93}9495export async function createNativeSyncTransport(96  connectionId: string,97  bindings?: NativeSyncTransportBindings98): Promise<SyncSessionTransport> {99  const ready = bindings ?? (await loadNativeSyncBindings());100  return {101    send: (frame) => ready.sendSyncData(connectionId, frame),102    onFrame: (handler) =>103      ready.onSyncMessage((id, frame) => {104        if (id === connectionId) handler(frame);105      }),106    onError: (handler) =>107      ready.onError((message, id) => {108        if (id === undefined || id === connectionId) handler(message);109      }),110    onClose: (handler) =>111      ready.onDisconnected((id) => {112        if (id === undefined || id === connectionId) handler();113      }),114    close: () => ready.disconnectFromPeer(connectionId),115    getSessionKey: () => ready.getNativeSessionKey(connectionId),116  };117}118119export async function checkSyncPreflight(120  deps?: Pick<SyncCoordinatorDeps, "isProUser" | "isUnlocked">121): Promise<SyncPreflightResult> {122  const readyDeps = deps ?? (await loadDefaultDeps());123  if (!(await readyDeps.isProUser())) return { ok: false, error: "pro_required" };124  if (!readyDeps.isUnlocked()) return { ok: false, error: "vault_locked" };125  return { ok: true };126}127128export async function buildLocalSyncSnapshot(129  deps?: Pick<SyncCoordinatorDeps, "isProUser" | "isUnlocked" | "getEntries">130): Promise<SyncSnapshot> {131  const readyDeps = deps ?? (await loadDefaultDeps());132  const preflight = await checkSyncPreflight(readyDeps);133  if (!preflight.ok) throw new Error(preflight.error);134135  const entries = readyDeps.getEntries();136  return {137    entries,138    manifest: await buildSyncManifest(entries),139  };140}141142export async function planManifestSync(143  remoteManifest: SyncManifestEntry[],144  deps?: Pick<SyncCoordinatorDeps, "isProUser" | "isUnlocked" | "getEntries">145): Promise<SyncPlan> {146  const snapshot = await buildLocalSyncSnapshot(deps);147  return {148    localManifest: snapshot.manifest,149    diff: diffSyncManifests(snapshot.manifest, remoteManifest),150  };151}152153export async function buildSyncEntriesMessage(154  ids: string[],155  rawEntryKey: Uint8Array,156  deps?: Pick<SyncCoordinatorDeps, "isProUser" | "isUnlocked" | "getEntries">157): Promise<Extract<SyncMessage, { type: "SYNC_ENTRIES" }>> {158  const readyDeps = deps ?? (await loadDefaultDeps());159  const preflight = await checkSyncPreflight(readyDeps);160  if (!preflight.ok) throw new Error(preflight.error);161162  const key = await importSyncEntryKey(rawEntryKey);163  const selected = selectEntriesForSync(readyDeps.getEntries(), ids);164  return {165    type: "SYNC_ENTRIES",166    entries: await Promise.all(selected.map((entry) => encryptSyncEntry(entry, key))),167  };168}169170export async function applySyncEntriesMessage(171  message: Extract<SyncMessage, { type: "SYNC_ENTRIES" }>,172  rawEntryKey: Uint8Array,173  deps?: Pick<SyncCoordinatorDeps, "isProUser" | "isUnlocked" | "restoreEntries">174): Promise<{ added: number; updated: number; skipped: number }> {175  const readyDeps = deps ?? (await loadDefaultDeps());176  const preflight = await checkSyncPreflight(readyDeps);177  if (!preflight.ok) throw new Error(preflight.error);178179  const key = await importSyncEntryKey(rawEntryKey);180  const entries = await Promise.all(181    message.entries.map((payload) => decryptSyncEntry(payload, key))182  );183  return readyDeps.restoreEntries(entries, "merge");184}185186export interface SyncFrameDispatchResult {187  outbound: string[];188  merge?: { added: number; updated: number; skipped: number };189  done?: boolean;190  error?: { code: string; message: string };191}192193export async function dispatchSyncFrame(194  frame: string,195  rawEntryKey: Uint8Array,196  deps?: Pick<197    SyncCoordinatorDeps,198    "isProUser" | "isUnlocked" | "getEntries" | "restoreEntries"199  >200): Promise<SyncFrameDispatchResult> {201  const readyDeps = deps ?? (await loadDefaultDeps());202  const preflight = await checkSyncPreflight(readyDeps);203  if (!preflight.ok) {204    const message =205      preflight.error === "pro_required"206        ? "Sync requires a Pro license."207        : "Vault is locked; unlock Peach before syncing.";208    return {209      outbound: [210        encodeSyncMessage({211          type: "SYNC_ERROR",212          code: preflight.error ?? "preflight_failed",213          message,214        }),215      ],216      done: true,217      error: { code: preflight.error ?? "preflight_failed", message },218    };219  }220221  let message: SyncMessage;222  try {223    message = decodeSyncMessage(frame);224  } catch {225    return {226      outbound: [227        encodeSyncMessage({228          type: "SYNC_ERROR",229          code: "invalid_message",230          message: "Could not decode sync frame.",231        }),232      ],233      done: true,234      error: { code: "invalid_message", message: "Could not decode sync frame." },235    };236  }237238  const key = await importSyncEntryKey(rawEntryKey);239240  switch (message.type) {241    case "SYNC_HELLO": {242      const manifest = await buildSyncManifest(readyDeps.getEntries());243      return {244        outbound: [encodeSyncMessage({ type: "SYNC_MANIFEST", entries: manifest })],245      };246    }247    case "SYNC_MANIFEST": {248      const localEntries = readyDeps.getEntries();249      const localManifest = await buildSyncManifest(localEntries);250      const diff = diffSyncManifests(localManifest, message.entries);251252      const requestIds = new Set(diff.requestFromRemote);253      const offerIds = new Set(diff.offerToRemote);254      const archiveCopies: VaultEntry[] = [];255256      if (diff.conflicts.length > 0) {257        const localManifestById = new Map(258          localManifest.map((entry) => [entry.id, entry])259        );260        const remoteManifestById = new Map(261          message.entries.map((entry) => [entry.id, entry])262        );263        const localEntryById = new Map(localEntries.map((entry) => [entry.id, entry]));264265        for (const id of diff.conflicts) {266          const localManifestEntry = localManifestById.get(id);267          const remoteManifestEntry = remoteManifestById.get(id);268          const localEntry = localEntryById.get(id);269          if (!localManifestEntry || !remoteManifestEntry || !localEntry) continue;270271          const winner = chooseConflictWinner(272            localManifestEntry.hash,273            remoteManifestEntry.hash274          );275          if (winner === "local") {276            offerIds.add(id);277          } else {278            requestIds.add(id);279            archiveCopies.push(buildConflictedCopy(localEntry, localManifestEntry.hash));280          }281        }282      }283284      // Persist any losing-side conflicted copies before the peer's winning285      // entries overwrite the original ids on the next SYNC_ENTRIES message.286      let merge: { added: number; updated: number; skipped: number } | undefined;287      if (archiveCopies.length > 0) {288        merge = await readyDeps.restoreEntries(archiveCopies, "merge");289      }290291      const sortedRequestIds = [...requestIds].sort();292      const sortedOfferIds = [...offerIds].sort();293      const offered = selectEntriesForSync(localEntries, sortedOfferIds);294      const outbound: string[] = [];295      if (sortedRequestIds.length > 0) {296        outbound.push(297          encodeSyncMessage({ type: "SYNC_REQUEST", ids: sortedRequestIds })298        );299      }300      if (offered.length > 0) {301        outbound.push(302          encodeSyncMessage({303            type: "SYNC_ENTRIES",304            entries: await Promise.all(305              offered.map((entry) => encryptSyncEntry(entry, key))306            ),307          })308        );309      }310      if (outbound.length === 0) {311        outbound.push(encodeSyncMessage({ type: "SYNC_DONE" }));312      }313      return merge ? { outbound, merge } : { outbound };314    }315    case "SYNC_REQUEST": {316      const offered = selectEntriesForSync(readyDeps.getEntries(), message.ids);317      return {318        outbound: [319          encodeSyncMessage({320            type: "SYNC_ENTRIES",321            entries: await Promise.all(322              offered.map((entry) => encryptSyncEntry(entry, key))323            ),324          }),325        ],326      };327    }328    case "SYNC_ENTRIES": {329      const decrypted = await Promise.all(330        message.entries.map((payload) => decryptSyncEntry(payload, key))331      );332      const merge = await readyDeps.restoreEntries(decrypted, "merge");333      return { outbound: [encodeSyncMessage({ type: "SYNC_DONE" })], merge };334    }335    case "SYNC_DONE":336      return { outbound: [], done: true };337    case "SYNC_ERROR":338      return {339        outbound: [],340        done: true,341        error: { code: message.code, message: message.message },342      };343    case "PAIR_REQUEST":344    case "PAIR_ACCEPT":345    case "PAIR_REJECT":346      // Pairing flow is handled outside the data-sync state machine.347      return { outbound: [] };348  }349}350351export interface SyncSessionTransport {352  send: (frame: string) => void;353  onFrame: (handler: (frame: string) => void) => () => void;354  onError: (handler: (message: string) => void) => () => void;355  onClose: (handler: () => void) => () => void;356  close: () => void;357  getSessionKey: () => Promise<Uint8Array>;358}359360export interface SyncSessionResult {361  status: "completed" | "errored" | "closed";362  merges: { added: number; updated: number; skipped: number };363  error?: { code: string; message: string };364}365366export type SyncRole = "initiator" | "responder";367368export async function runSyncSession(options: {369  role: SyncRole;370  deviceId: string;371  vaultVersion: number;372  transport: SyncSessionTransport;373  deps?: Pick<374    SyncCoordinatorDeps,375    "isProUser" | "isUnlocked" | "getEntries" | "restoreEntries"376  >;377}): Promise<SyncSessionResult> {378  const readyDeps = options.deps ?? (await loadDefaultDeps());379  const merges = { added: 0, updated: 0, skipped: 0 };380381  const preflight = await checkSyncPreflight(readyDeps);382  if (!preflight.ok) {383    const message =384      preflight.error === "pro_required"385        ? "Sync requires a Pro license."386        : "Vault is locked; unlock Peach before syncing.";387    options.transport.send(388      encodeSyncMessage({389        type: "SYNC_ERROR",390        code: preflight.error ?? "preflight_failed",391        message,392      })393    );394    options.transport.close();395    return {396      status: "errored",397      merges,398      error: { code: preflight.error ?? "preflight_failed", message },399    };400  }401402  const sessionKey = await options.transport.getSessionKey();403404  return new Promise<SyncSessionResult>((resolve) => {405    let settled = false;406    const unsubscribers: Array<() => void> = [];407408    const finish = (result: SyncSessionResult): void => {409      if (settled) return;410      settled = true;411      for (const fn of unsubscribers) fn();412      options.transport.close();413      resolve(result);414    };415416    unsubscribers.push(417      options.transport.onClose(() =>418        finish({ status: "closed", merges })419      )420    );421    unsubscribers.push(422      options.transport.onError((message) =>423        finish({424          status: "errored",425          merges,426          error: { code: "transport_error", message },427        })428      )429    );430431    unsubscribers.push(432      options.transport.onFrame((frame) => {433        void (async () => {434          const result = await dispatchSyncFrame(frame, sessionKey, readyDeps);435          if (settled) return;436          if (result.merge) {437            merges.added += result.merge.added;438            merges.updated += result.merge.updated;439            merges.skipped += result.merge.skipped;440          }441          for (const outbound of result.outbound) options.transport.send(outbound);442          if (result.error) {443            finish({ status: "errored", merges, error: result.error });444            return;445          }446          if (result.done) {447            finish({ status: "completed", merges });448          }449        })();450      })451    );452453    if (options.role === "initiator") {454      options.transport.send(455        encodeSyncMessage({456          type: "SYNC_HELLO",457          deviceId: options.deviceId,458          clockMs: Date.now(),459          vaultVersion: options.vaultVersion,460        })461      );462    }463  });464}465466export async function cancelSyncIfVaultLocked(467  connectionId: string,468  deps?: Pick<SyncCoordinatorDeps, "isUnlocked" | "sendSyncData">469): Promise<boolean> {470  const readyDeps = deps ?? (await loadDefaultDeps());471  if (readyDeps.isUnlocked()) return false;472473  readyDeps.sendSyncData(474    connectionId,475    encodeSyncMessage({476      type: "SYNC_ERROR",477      code: "vault_locked",478      message: "Vault is locked; unlock Peach before syncing.",479    })480  );481  return true;482}