chadtmiller.com/pds.js

Switch repository

Loading…

// A tiny XRPC client for talking to an arbitrary PDS during migration: the
// user's old server (cross-origin) and this one (same-origin). Deliberately
// standalone — the account /account/api helper is same-origin + session-cookie
// only, and migration has to reach another host with a bearer token.

const PUBLIC_APPVIEW = 'https://public.api.bsky.app';
const PLC_DIRECTORY = 'https://plc.directory';

// A JSON call that stops making progress is almost always a dropped socket, so
// give up quickly and let the caller retry. Binary transfers (repo CARs, blobs)
// can legitimately take longer, so they're bounded by a *stall* timeout — no new
// bytes for this long means the stream is wedged (a symptom we hit migrating a
// blob the source PDS could only half-read) — plus a generous overall backstop.
const JSON_TIMEOUT_MS = 30_000;
const BINARY_STALL_MS = 20_000;
const BINARY_OVERALL_MS = 300_000;

export class XrpcError extends Error {
  /**
   * @param {number} status
   * @param {string} [error]
   * @param {string} [message]
   */
  constructor(status, error, message) {
    super(message || error || `Request failed (${status})`);
    this.name = 'XrpcError';
    this.status = status;
    this.error = error || '';
  }
}

/**
 * True for the atproto "second factor required / wrong code" error.
 * @param {unknown} err
 */
export function isAuthFactorError(err) {
  return (
    err instanceof XrpcError &&
    (err.error === 'AuthFactorTokenRequired' ||
      /sign in code|auth ?factor/i.test(err.message))
  );
}

/**
 * True for a 429 from either side.
 * @param {unknown} err
 */
export function isRateLimitError(err) {
  return err instanceof XrpcError && err.status === 429;
}

/**
 * A client bound to one PDS base URL. `''` means this origin (same-origin).
 * @param {string} base
 */
export function xrpc(base) {
  const root = base.replace(/\/+$/, '');
  /**
   * @param {string} nsid
   * @param {{ method?: string, params?: Record<string, string>, body?: any, token?: string, binary?: boolean, contentType?: string }} [opts]
   */
  async function call(nsid, opts = {}) {
    const { method = 'GET', params, body, token, binary, contentType } = opts;
    let url = `${root}/xrpc/${nsid}`;
    if (params) url += `?${new URLSearchParams(params)}`;

    /** @type {Record<string, string>} */
    const headers = {};
    if (token) headers.Authorization = `Bearer ${token}`;
    /** @type {BodyInit | undefined} */
    let payload;
    if (body instanceof Uint8Array) {
      headers['Content-Type'] = contentType || 'application/octet-stream';
      payload = /** @type {Uint8Array<ArrayBuffer>} */ (body);
    } else if (body !== undefined) {
      headers['Content-Type'] = 'application/json';
      payload = JSON.stringify(body);
    }

    const controller = new AbortController();
    if (binary) {
      // Overall backstop; the finer stall guard lives in readWithStall.
      const overall = setTimeout(() => controller.abort(), BINARY_OVERALL_MS);
      try {
        const res = await fetch(url, {
          method,
          headers,
          body: payload,
          signal: controller.signal,
        });
        if (!res.ok) throw await toError(res);
        const bytes = await readWithStall(res, controller);
        return { bytes, res };
      } finally {
        clearTimeout(overall);
      }
    }
    const timer = setTimeout(() => controller.abort(), JSON_TIMEOUT_MS);
    try {
      const res = await fetch(url, {
        method,
        headers,
        body: payload,
        signal: controller.signal,
      });
      const data = await res.json().catch(() => ({}));
      if (!res.ok) throw new XrpcError(res.status, data.error, data.message);
      return data;
    } finally {
      clearTimeout(timer);
    }
  }
  return { base: root, call };
}

/**
 * Drain a binary response, aborting if the body goes quiet for BINARY_STALL_MS.
 * A stalled read (the server sent headers + some bytes, then wedged) otherwise
 * hangs forever; aborting turns it into a rejection the caller can retry. If a
 * Content-Length is present we stop as soon as we've read it all, so a server
 * that delivers every byte but never closes the stream still resolves.
 * @param {Response} res
 * @param {AbortController} controller
 */
async function readWithStall(res, controller) {
  if (!res.body) return new Uint8Array(await res.arrayBuffer());
  const lenHeader = res.headers.get('content-length');
  const expected = lenHeader ? Number(lenHeader) : null;
  const reader = res.body.getReader();
  /** @type {Uint8Array[]} */
  const chunks = [];
  let total = 0;
  /** @type {ReturnType<typeof setTimeout> | undefined} */
  let stall;
  const arm = () => {
    clearTimeout(stall);
    stall = setTimeout(() => controller.abort(), BINARY_STALL_MS);
  };
  try {
    arm();
    for (;;) {
      const { done, value } = await reader.read();
      if (done) break;
      if (value) {
        chunks.push(value);
        total += value.length;
        arm();
        if (expected != null && total >= expected) break;
      }
    }
  } finally {
    clearTimeout(stall);
  }
  const out = new Uint8Array(total);
  let offset = 0;
  for (const chunk of chunks) {
    out.set(chunk, offset);
    offset += chunk.length;
  }
  return out;
}

/** @param {Response} res */
async function toError(res) {
  const data = await res.json().catch(() => ({}));
  return new XrpcError(res.status, data.error, data.message);
}

/**
 * Resolve a DID to its DID document (did:plc via the directory, did:web via
 * well-known).
 * @param {string} did
 */
async function resolveDidDoc(did) {
  let url;
  if (did.startsWith('did:plc:')) {
    url = `${PLC_DIRECTORY}/${did}`;
  } else if (did.startsWith('did:web:')) {
    url = `https://${did.slice('did:web:'.length)}/.well-known/did.json`;
  } else {
    throw new Error(`Unsupported DID method: ${did}`);
  }
  const res = await fetch(url);
  if (!res.ok) throw new Error(`Could not resolve ${did}`);
  return res.json();
}

/**
 * Pull the atproto PDS endpoint out of a DID document.
 * @param {{service?: {id: string, type: string, serviceEndpoint?: string}[]}|null} didDoc
 */
export function pdsEndpointOf(didDoc) {
  const service = (didDoc?.service || []).find(
    (/** @type {{id: string, type: string}} */ s) =>
      s.id === '#atproto_pds' || s.type === 'AtprotoPersonalDataServer',
  );
  return service?.serviceEndpoint || null;
}

/**
 * Resolve a handle or DID to `{ did, pdsUrl }`. Handles resolve through the
 * public AppView (works for any handle in the network without DNS access from
 * the browser); the DID then resolves to its hosting PDS.
 * @param {string} identifier
 * @returns {Promise<{ did: string, pdsUrl: string }>}
 */
export async function resolveAccount(identifier) {
  const value = identifier.trim().replace(/^@/, '');
  let did = value;
  if (!value.startsWith('did:')) {
    const res = await fetch(
      `${PUBLIC_APPVIEW}/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(value)}`,
    );
    const data = await res.json().catch(() => ({}));
    if (!res.ok || !data.did) {
      throw new Error(`Could not find an account for "${value}".`);
    }
    did = data.did;
  }
  const doc = await resolveDidDoc(did);
  const pdsUrl = pdsEndpointOf(doc);
  if (!pdsUrl) throw new Error(`No PDS is listed for ${did}.`);
  return { did, pdsUrl };
}

/**
 * The most recent PDS endpoint in a DID's PLC history that isn't `exclude` —
 * used to rediscover the old PDS when resuming with only the new one known.
 * @param {string} did
 * @param {string} exclude
 */
export async function previousPdsEndpoint(did, exclude) {
  const res = await fetch(`${PLC_DIRECTORY}/${did}/log`);
  if (!res.ok) return null;
  const log = await res.json();
  for (let i = log.length - 1; i >= 0; i--) {
    const endpoint =
      log[i]?.operation?.services?.atproto_pds?.endpoint ||
      log[i]?.services?.atproto_pds?.endpoint;
    if (
      endpoint &&
      endpoint.replace(/\/+$/, '') !== exclude.replace(/\/+$/, '')
    )
      return endpoint;
  }
  return null;
}