chadtmiller.com/pds.js

Switch repository

Loading…

// Drives a full account migration onto this PDS, following the same ordered
// dance the atproto reference tools use. The destination is this server
// (same-origin); the source is the user's old PDS (cross-origin, bearer auth).
//
// The two human pauses — the old server's 2FA email and the PLC confirmation
// email — are `await`ed hooks the UI resolves when the code is typed.

import {
  isAuthFactorError,
  isRateLimitError,
  resolveAccount,
  XrpcError,
  xrpc,
} from '#/lib/atproto.js';

const sleep = (/** @type {number} */ ms) =>
  new Promise((resolve) => setTimeout(resolve, ms));

/** Progress caption for the blob step, mentioning skips only when they happen. */
function blobDetail(
  /** @type {number} */ moved,
  /** @type {string[]} */ skipped,
) {
  const base = `${moved} copied`;
  return skipped.length ? `${base}, ${skipped.length} unreadable` : base;
}

/** @typedef {ReturnType<typeof import('./atproto.js').xrpc>} Client */

/**
 * Pull the source repo as a single CAR and import it into the destination.
 * Idempotent: re-importing a newer CAR advances the destination to the source's
 * latest commit, which is what makes the pre-identity catch-up work.
 * @param {Client} source
 * @param {Client} dest
 * @param {string} did
 * @param {string} destToken
 * @returns {Promise<number>} the CAR size in bytes
 */
async function importRepo(source, dest, did, destToken) {
  const { bytes: car } = await source.call('com.atproto.sync.getRepo', {
    params: { did },
    binary: true,
  });
  await dest.call('com.atproto.repo.importRepo', {
    method: 'POST',
    body: car,
    contentType: 'application/vnd.ipld.car',
    token: destToken,
  });
  return car.length;
}

/**
 * Copy every blob the destination still reports missing, retrying each and
 * skipping any the source genuinely cannot serve. Progress and the running
 * count are reported against `stepKey` so the same routine can drive both the
 * initial "Copy your blobs" pass and the later catch-up. Mutates
 * `state.moved` / `state.skipped`.
 * @param {Client} source
 * @param {Client} dest
 * @param {string} did
 * @param {string} destToken
 * @param {{moved: number, skipped: string[]}} state
 * @param {(key: string, patch: object) => void} step
 * @param {string} stepKey
 */
async function copyMissingBlobs(
  source,
  dest,
  did,
  destToken,
  state,
  step,
  stepKey,
) {
  for (;;) {
    const { blobs } = await dest.call('com.atproto.repo.listMissingBlobs', {
      params: { limit: '100' },
      token: destToken,
    });
    // Anything still "missing" that we've already given up on would loop
    // forever; excluding it lets the loop terminate.
    const todo = (blobs || []).filter(
      (/** @type {{cid: string}} */ b) => !state.skipped.includes(b.cid),
    );
    if (todo.length === 0) break;
    for (const blob of todo) {
      try {
        // Each blob independently retried: one dropped connection out of
        // hundreds should not abandon the whole transfer.
        await withRetry(async () => {
          const { bytes } = await source.call('com.atproto.sync.getBlob', {
            params: { did, cid: blob.cid },
            binary: true,
          });
          await dest.call('com.atproto.repo.uploadBlob', {
            method: 'POST',
            body: bytes,
            token: destToken,
          });
        });
        state.moved += 1;
      } catch (err) {
        // A blob the source can't serve (e.g. damaged storage that only
        // half-reads) would block the whole move. Skip it, note it, press on —
        // the record still imports; only that one attachment is absent.
        if (isRateLimitError(err)) throw err;
        state.skipped.push(blob.cid);
      }
      step(stepKey, {
        status: 'active',
        detail: blobDetail(state.moved, state.skipped),
      });
    }
  }
}

/**
 * Retry a transient failure (a dropped connection surfaces as "Failed to
 * fetch") with exponential backoff. A real rate limit is surfaced immediately —
 * hammering it would only make it worse — and so is a final give-up, so the
 * migration pauses with the source still intact rather than half-moving.
 * @template T
 * @param {() => Promise<T>} fn
 * @returns {Promise<T>}
 */
async function withRetry(fn) {
  const maxAttempts = 5;
  let lastError;
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err) {
      if (isRateLimitError(err)) throw err;
      lastError = err;
      if (attempt < maxAttempts) {
        await sleep(Math.min(500 * 2 ** (attempt - 1), 8000));
      }
    }
  }
  throw lastError;
}

/** The ordered steps, for the UI to render as a checklist. */
export const MIGRATION_STEPS = [
  { key: 'resolve', label: 'Find your account' },
  { key: 'login', label: 'Sign in to your old server' },
  { key: 'create', label: 'Create your account here' },
  { key: 'repo', label: 'Copy your repository' },
  { key: 'blobs', label: 'Copy your blobs' },
  { key: 'prefs', label: 'Copy your preferences' },
  { key: 'sync', label: 'Catch up any last changes' },
  { key: 'identity', label: 'Move your identity' },
  { key: 'activate', label: 'Activate here, retire the old one' },
];

/**
 * @param {{identifier: string, password: string, newHandle: string, email: string, serverPassword: string, recoveryKey?: (string|null)}} input
 * @param {{ onStep: (key: string, patch: object) => void, requestTwoFactor: () => Promise<string>, requestPlcCode: () => Promise<string> }} hooks
 */
export async function runMigration(input, hooks) {
  const {
    identifier,
    password,
    newHandle,
    email,
    serverPassword,
    recoveryKey,
  } = input;
  const step = hooks.onStep;
  const dest = xrpc(''); // same origin — this PDS

  // 1. Resolve the old identity to a DID + its current PDS.
  step('resolve', { status: 'active' });
  const { did, pdsUrl } = await resolveAccount(identifier);
  const source = xrpc(pdsUrl);
  step('resolve', {
    status: 'done',
    detail: `${did} at ${new URL(pdsUrl).host}`,
  });

  // This server's did:web identity — the audience for the service token.
  const destDesc = await dest.call('com.atproto.server.describeServer');
  const destDid = destDesc.did;

  // 2. Sign in to the source. A token-less attempt is what makes the old
  //    server email its 2FA code, so a first failure is expected, not fatal.
  step('login', { status: 'active' });
  const sourceSession = await loginSource(source, did, password, hooks);
  step('login', { status: 'done' });

  // 3. Create the account here (idempotent — a resumed run finds it exists),
  //    then open a working session against this server. createSession also
  //    checks the server password, so this is the operator's confirmation.
  step('create', { status: 'active' });
  const { token: serviceJwt } = await source.call(
    'com.atproto.server.getServiceAuth',
    {
      params: { aud: destDid, lxm: 'com.atproto.server.createAccount' },
      token: sourceSession.accessJwt,
    },
  );
  try {
    await dest.call('com.atproto.server.createAccount', {
      method: 'POST',
      body: { did, handle: newHandle, email: email || undefined },
      token: serviceJwt,
    });
  } catch (err) {
    if (!(err instanceof XrpcError && /already/i.test(err.message))) throw err;
  }
  const destSession = await dest.call('com.atproto.server.createSession', {
    method: 'POST',
    body: { identifier: did, password: serverPassword },
  });
  const destToken = destSession.accessJwt;
  step('create', { status: 'done' });

  // 4. Import the repository (a single CAR).
  step('repo', { status: 'active' });
  const carLen = await importRepo(source, dest, did, destToken);
  step('repo', { status: 'done', detail: `${carLen} bytes` });

  // 5. Blobs — driven by the destination's missing-blob list. That makes it
  //    idempotent (a retry only fetches what's still absent) and covers every
  //    blob a record references.
  step('blobs', { status: 'active', detail: '0 copied' });
  const blobState = { moved: 0, /** @type {string[]} */ skipped: [] };
  await copyMissingBlobs(
    source,
    dest,
    did,
    destToken,
    blobState,
    step,
    'blobs',
  );
  step('blobs', {
    status: blobState.skipped.length ? 'warn' : 'done',
    detail:
      blobState.moved || blobState.skipped.length
        ? blobDetail(blobState.moved, blobState.skipped)
        : 'none needed',
  });

  // 6. Preferences (best effort — an appview convenience, not identity).
  step('prefs', { status: 'active' });
  try {
    const { preferences } = await source.call('app.bsky.actor.getPreferences', {
      token: sourceSession.accessJwt,
    });
    await dest.call('app.bsky.actor.putPreferences', {
      method: 'POST',
      body: { preferences: preferences || [] },
      token: destToken,
    });
    step('prefs', { status: 'done' });
  } catch {
    step('prefs', { status: 'done', detail: 'skipped' });
  }

  // 6b. Catch up. Copying the repo happened minutes ago, and the source stayed
  //     live throughout the blob transfer, so it may have committed new records
  //     since. Re-import until the destination's latest commit matches the
  //     source's — nothing written mid-migration should be stranded. If the rev
  //     never settles, the account is still actively in use; stop and tell the
  //     user to quiet it rather than move a repo that's still a moving target.
  step('sync', { status: 'active' });
  let caughtUp = false;
  for (let pass = 0; pass < 4; pass++) {
    const [{ rev: srcRev }, { rev: dstRev }] = await Promise.all([
      source.call('com.atproto.sync.getLatestCommit', { params: { did } }),
      dest.call('com.atproto.sync.getRepoStatus', { params: { did } }),
    ]);
    if (srcRev === dstRev) {
      caughtUp = true;
      break;
    }
    step('sync', { status: 'active', detail: 'catching up new changes' });
    await importRepo(source, dest, did, destToken);
    await copyMissingBlobs(
      source,
      dest,
      did,
      destToken,
      blobState,
      step,
      'sync',
    );
  }
  if (!caughtUp) {
    throw new Error(
      'Your old account is still receiving new records, so its data keeps ' +
        'changing under the migration. Sign out of any apps still using it ' +
        '(Bluesky and the like), then click “Try again” to finish the move.',
    );
  }
  step('sync', { status: 'done' });

  // 7. Move the identity. The source still holds the DID's rotation key, so it
  //    signs; the emailed token proves the request is the account owner's.
  step('identity', {
    status: 'active',
    detail: 'emailing a confirmation code',
  });
  const creds = await dest.call(
    'com.atproto.identity.getRecommendedDidCredentials',
    { token: destToken },
  );
  // This server lists its own rotation key first (see getRecommendedDidCredentials),
  // then any keys already on the DID. When the user minted a recovery key we want
  // a clean set — the key they hold, plus this server's — and deliberately drop
  // the old PDS's keys (retired) and any stale keys they no longer control.
  // Without a recovery key we keep the server's full recommendation, which
  // preserves whatever was already there rather than narrowing it.
  const rotationKeys = recoveryKey
    ? [recoveryKey, creds.rotationKeys[0]]
    : creds.rotationKeys;
  // No body: these methods take no input, and a strict PDS rejects an
  // unexpected request body ("A request body was provided when none was
  // expected").
  await source.call('com.atproto.identity.requestPlcOperationSignature', {
    method: 'POST',
    token: sourceSession.accessJwt,
  });
  const code = await hooks.requestPlcCode();
  const signed = await source.call('com.atproto.identity.signPlcOperation', {
    method: 'POST',
    body: {
      token: code,
      rotationKeys,
      verificationMethods: creds.verificationMethods,
      alsoKnownAs: creds.alsoKnownAs,
      services: creds.services,
    },
    token: sourceSession.accessJwt,
  });
  await dest.call('com.atproto.identity.submitPlcOperation', {
    method: 'POST',
    body: { operation: signed.operation },
    token: destToken,
  });
  step('identity', { status: 'done' });

  // 8. Activate here, then retire the old account — never the other way, so
  //    the DID never points at a deactivated home.
  step('activate', { status: 'active' });
  await dest.call('com.atproto.server.activateAccount', {
    method: 'POST',
    token: destToken,
  });
  try {
    await source.call('com.atproto.server.deactivateAccount', {
      method: 'POST',
      token: sourceSession.accessJwt,
    });
  } catch {
    // The move is already complete; a failure to stand down the old account
    // is cosmetic and the user can do it later.
  }
  step('activate', { status: 'done' });

  return { did, handle: newHandle, skippedBlobs: blobState.skipped };
}

/**
 * Log in to the source PDS, prompting for a 2FA code if the server demands one.
 * @param {Client} source
 * @param {string} identifier
 * @param {string} password
 * @param {{requestTwoFactor: () => Promise<string|null>}} hooks
 */
async function loginSource(source, identifier, password, hooks) {
  try {
    return await source.call('com.atproto.server.createSession', {
      method: 'POST',
      body: { identifier, password },
    });
  } catch (err) {
    if (!isAuthFactorError(err)) throw err;
    const authFactorToken = await hooks.requestTwoFactor();
    return await source.call('com.atproto.server.createSession', {
      method: 'POST',
      body: { identifier, password, authFactorToken },
    });
  }
}