sync-runner.ts

1// Production sync runner — composes the live vault adapter, WebSocket2// transport, and coordinator into the function the dispatcher injects via3// `setSyncRunner`. The runner is responsible for opening one Noise KK4// session against a single paired device, driving the sync protocol, and5// surfacing a `SyncRunResult` (or throwing on failure). The dispatcher in6// sync-trigger.ts iterates over hosts and aggregates results.7//8// Connection strategy:9//   1. Try `getLastKnownIp(deviceLongTermPubkey)` if cached — fast path.10//   2. Fall back to `probeSubnetForPeer` over the local /24. The subnet11//      probe is itself rate-limited (60s per-pubkey cooldown) inside the12//      transport, so we don't add another layer here.13//   3. On a successful session, persist the IP we connected on so the next14//      sync round skips the probe.15//16// We do NOT verify vault_id here — the coordinator's SYNC_HELLO exchange17// does that (returns `status: "vault_mismatch"` if the peer holds a18// different vault). The runner just translates the coordinator's outcome19// into the throw-or-return contract the dispatcher expects.2021import { runSyncInitiator, type SyncVaultAdapter } from "./sync-coordinator";22import { createSyncVaultAdapter } from "./sync-vault-adapter";23import {24  getLastKnownIp,25  setLastKnownIp,26} from "./paired-devices";27import {28  openSyncSession,29  probeSubnetForPeer,30  type ProbedSyncEndpoint,31  type SyncSession,32  type SyncEndpoint,33  type SyncPeerIdentity,34} from "../shared/sync-transport";35import { hexToPubkey } from "../shared/sync-identity";36import { getLocalSubnet } from "../shared/local-network";37import { buildLocalIdentity, deriveLocalSyncKeypair } from "./device-identity";38import { PAIRING_PORT } from "../shared/pairing-dial";39import { getRootKeyHex, isUnlocked } from "./vault";40import type { PairedDevice } from "../shared/sync-types";41import type { SyncRunResult } from "./sync-trigger";42import {43  dialingRendezvousIds,44  rendezvousSecretFromHex,45} from "../shared/rendezvous";4647/** Same fixed port pairing dials — SyncReconnectListener.SYNC_PORT on Android. */48const DEFAULT_PORT = PAIRING_PORT;4950/**51 * Test/injection seam. Production wires in the real `openSyncSession`;52 * unit tests inject a fake to avoid real WebSocket + Noise KK.53 */54export interface SyncRunnerDeps {55  openSession: (56    endpoint: SyncEndpoint,57    ourIdentity: SyncPeerIdentity,58    expectedRemoteHex: string,59    rendezvousId: string,60  ) => Promise<SyncSession>;61  probeSubnet: (62    subnet: string,63    port: number,64    ourIdentity: SyncPeerIdentity,65    expectedRemoteHex: string,66    rendezvousIds: readonly string[],67  ) => Promise<ProbedSyncEndpoint | null>;68  getSubnet: () => Promise<string | null>;69  createAdapter: (context?: { deviceName?: string }) => SyncVaultAdapter;70}7172const productionDeps: SyncRunnerDeps = {73  openSession: (endpoint, identity, expected, rendezvousId) =>74    openSyncSession(endpoint, identity, expected, rendezvousId),75  probeSubnet: (subnet, port, identity, expected, rendezvousIds) =>76    probeSubnetForPeer(subnet, port, identity, expected, rendezvousIds),77  getSubnet: getLocalSubnet,78  createAdapter: (context) => createSyncVaultAdapter(context),79};8081export async function productionSyncRunner(device: PairedDevice): Promise<SyncRunResult> {82  return runWithDeps(device, productionDeps);83}8485/**86 * Inner runner with explicit deps — exported so tests can inject fakes.87 */88export async function runWithDeps(89  device: PairedDevice,90  deps: SyncRunnerDeps,91): Promise<SyncRunResult> {92  if (!isUnlocked()) return { ok: false, error: "VAULT_LOCKED" };93  if (!device.longTermPublicKeyHex) {94    throw new Error("paired_device_missing_pubkey");95  }96  if (!device.rendezvousSecretHex) {97    throw new Error("paired_device_missing_rendezvous_secret");98  }99  const expectedRemoteHex = device.longTermPublicKeyHex;100  const reconnectIds = dialingRendezvousIds(101    rendezvousSecretFromHex(device.rendezvousSecretHex),102    Date.now() / 1000,103  );104  const port = device.wsPort ?? DEFAULT_PORT;105106  // Sync identity derives from the stable root key, not the password-derived107  // key. Reconnect uses both halves directly in standard Noise KK.108  const masterKeyHex = getRootKeyHex();109  if (!masterKeyHex) return { ok: false, error: "VAULT_LOCKED" };110  const masterKey = hexToBytesLocal(masterKeyHex);111  const publicIdentity = await buildLocalIdentity(masterKey);112  const localKeys = deriveLocalSyncKeypair(masterKey, publicIdentity.deviceId);113  if (publicIdentity.longTermPublicKeyHex !== bytesToHex(localKeys.publicKey)) {114    masterKey.fill(0);115    localKeys.secretKey.fill(0);116    throw new Error("derived reconnect identity mismatch");117  }118  const ourIdentity: SyncPeerIdentity = {119    ...publicIdentity,120    longTermPrivateKey: localKeys.secretKey,121  };122123  // Sanity — the remote pubkey must be 32-byte hex so noise can compare it.124  void hexToPubkey(expectedRemoteHex);125126  // HV-1b: pass the peer being synced so remote-mutation alerts can name it.127  const adapter = deps.createAdapter({ deviceName: device.deviceName });128  let session: SyncSession;129  try {130    session = await connect(expectedRemoteHex, port, ourIdentity, reconnectIds, deps);131  } finally {132    masterKey.fill(0);133    localKeys.secretKey.fill(0);134  }135136  try {137    const outcome = await runSyncInitiator(session, adapter);138    if (outcome.status === "completed") {139      return {140        ok: true,141        entriesReceived: outcome.entriesReceived,142        entriesSent: outcome.entriesSent,143      };144    }145    if (outcome.status === "vault_mismatch") {146      throw new Error("vault_id_mismatch");147    }148    if (outcome.status === "timeout") {149      throw new Error("sync_timeout");150    }151    throw new Error(outcome.error ?? "sync_errored");152  } finally {153    // The coordinator does NOT close the session (it returns and leaves the154    // session open). This runner is the sole owner of session lifetime.155    await session.close().catch(() => undefined);156  }157}158159async function connect(160  expectedRemoteHex: string,161  port: number,162  ourIdentity: SyncPeerIdentity,163  rendezvousIds: string[],164  deps: SyncRunnerDeps,165): Promise<SyncSession> {166  const cached = await getLastKnownIp(expectedRemoteHex);167  if (cached) {168    for (const rendezvousId of rendezvousIds) {169      try {170        const session = await deps.openSession(171          { ip: cached, port },172          ourIdentity,173          expectedRemoteHex,174          rendezvousId,175        );176        return session;177      } catch {178        // Try the next clock-skew window before treating the IP as stale.179      }180    }181  }182183  const subnet = await deps.getSubnet();184  if (!subnet) {185    throw new Error(cached ? "host_unreachable" : "subnet_unknown");186  }187188  const found = await deps.probeSubnet(189    subnet,190    port,191    ourIdentity,192    expectedRemoteHex,193    rendezvousIds,194  );195  if (!found) throw new Error("host_unreachable");196  const session = await deps.openSession(197    { ip: found.ip, port: found.port },198    ourIdentity,199    expectedRemoteHex,200    found.rendezvousId,201  );202  await setLastKnownIp(expectedRemoteHex, found.ip).catch(() => undefined);203  return session;204}205206function hexToBytesLocal(hex: string): Uint8Array {207  const out = new Uint8Array(hex.length / 2);208  for (let i = 0; i < out.length; i++) {209    out[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16);210  }211  return out;212}213214function bytesToHex(bytes: Uint8Array): string {215  return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join("");216}