chadtmiller.com/pds.js

Switch repository

Loading…

/**
 * The historical half of the appview: a first-boot walk of the records the
 * live tail will never replay on its own.
 *
 * The relay's `com.atproto.sync.listReposByCollection` names every account
 * holding a watched collection. Each account's PDS is read from its DID
 * document, and each collection is listed with the PDS's own repo API, one
 * page at a time, into the store. A DID that does not answer costs its
 * records, not the walk.
 *
 * The pass resumes: the pagination cursor of each collection is stored in
 * the same table the live cursor uses, under its own source name.
 */

/** How many accounts one collection listing answers per page. */
const LIST_PAGE = 1000;

/** How many records one account's collection answers per page. */
const RECORDS_PAGE = 100;

/** The most records read from one account for one collection. */
const MAX_RECORDS_PER_DID = 1000;

/** How many accounts read at once. */
const CONCURRENCY = 8;

/**
 * What carries a request; the global fetch unless one is injected.
 * @typedef {(input: string|URL, init?: RequestInit) => Promise<Response>} FetchLike
 */

/**
 * @typedef {Object} BackfillOptions
 * @property {import('./store.js').AppViewStore} store
 * @property {string[]} collections
 * @property {string} [listService] - the relay serving listReposByCollection
 * @property {string} [plcUrl] - the directory did:plc documents resolve from
 * @property {FetchLike} [fetch]
 * @property {AbortSignal} [signal]
 * @property {(message: string) => void} [onNotice]
 */

/**
 * @param {BackfillOptions} options
 * @returns {Promise<void>} resolves when every collection has been walked
 */
export async function backfill(options) {
  const { store, collections, signal } = options;
  const send =
    options.fetch ??
    /** @type {FetchLike} */ ((url, init) => globalThis.fetch(url, init));
  const onNotice = options.onNotice ?? (() => {});
  const listService = (options.listService ?? 'https://bsky.network').replace(
    /\/+$/,
    '',
  );
  const plcUrl = (options.plcUrl ?? 'https://plc.directory').replace(
    /\/+$/,
    '',
  );

  for (const collection of collections) {
    if (signal?.aborted) return;
    const source = `backfill:${collection}`;
    let cursor = store.textCursor(source) ?? '';
    let stopped = false;
    while (!stopped) {
      if (signal?.aborted) return;
      const params = new URLSearchParams({
        collection,
        limit: String(LIST_PAGE),
      });
      if (cursor) params.set('cursor', cursor);
      let body;
      try {
        const res = await send(
          `${listService}/xrpc/com.atproto.sync.listReposByCollection?${params}`,
          { signal },
        );
        if (!res.ok) {
          onNotice(`backfill ${collection}: list answered ${res.status}`);
          return;
        }
        body = await res.json();
      } catch (err) {
        if (signal?.aborted) return;
        onNotice(
          `backfill ${collection}: ${err instanceof Error ? err.message : err}`,
        );
        return;
      }
      const repos = /** @type {{repos?: Array<{did: string}>}} */ (body);
      const dids = (repos.repos ?? [])
        .map((row) => row?.did)
        .filter((did) => typeof did === 'string');
      await pool(dids, CONCURRENCY, async (did) => {
        await indexAccount({
          store,
          did,
          collection,
          plcUrl,
          fetch: send,
          signal,
          onNotice,
        });
      });
      const next =
        typeof (/** @type {{cursor?: unknown}} */ (body).cursor) === 'string'
          ? /** @type {{cursor: string}} */ (body).cursor
          : '';
      cursor = next;
      if (next) store.setTextCursor(source, next);
      if (!next) stopped = true;
    }
    onNotice(`backfill ${collection}: done`);
  }
}

/**
 * Every record of one collection one account holds, into the store. The
 * account's own handle rides along from its DID document, which the PDS
 * resolution already fetched.
 * @param {Object} ctx
 * @param {import('./store.js').AppViewStore} ctx.store
 * @param {string} ctx.did
 * @param {string} ctx.collection
 * @param {string} ctx.plcUrl
 * @param {FetchLike} ctx.fetch
 * @param {AbortSignal|undefined} ctx.signal
 * @param {(message: string) => void} ctx.onNotice
 */
async function indexAccount(ctx) {
  const { store, did, collection, fetch, signal } = ctx;
  const doc = await readDidDocument(did, ctx.plcUrl, fetch, signal);
  if (!doc) return;
  const pds = pdsOf(doc);
  const handle = handleOf(doc);
  if (handle) store.setIdentity(did, handle);

  let cursor = '';
  for (let page = 0; page * RECORDS_PAGE < MAX_RECORDS_PER_DID; page++) {
    if (signal?.aborted) return;
    const params = new URLSearchParams({
      repo: did,
      collection,
      limit: String(RECORDS_PAGE),
    });
    if (cursor) params.set('cursor', cursor);
    let body;
    try {
      const res = await fetch(
        `${pds}/xrpc/com.atproto.repo.listRecords?${params}`,
        { signal },
      );
      if (!res.ok) return;
      body = await res.json();
    } catch {
      return;
    }
    const rows =
      /** @type {{records?: Array<{uri: string, cid: string, value: Record<string, *>}>, cursor?: string}} */ (
        body
      );
    for (const row of rows.records ?? []) {
      if (
        typeof row?.uri !== 'string' ||
        typeof row?.cid !== 'string' ||
        !row?.value ||
        typeof row.value !== 'object'
      ) {
        continue;
      }
      const rkey = row.uri.split('/').pop() ?? '';
      store.putRecord({
        did,
        collection,
        rkey,
        cid: row.cid,
        uri: row.uri,
        seq: 0,
        value: row.value,
      });
    }
    cursor = typeof rows.cursor === 'string' ? rows.cursor : '';
    if (!cursor) return;
  }
}

/**
 * @param {string} did
 * @param {string} plcUrl
 * @param {FetchLike} fetch
 * @param {AbortSignal|undefined} signal
 * @returns {Promise<any|null>} the DID document, or null when unreadable
 */
async function readDidDocument(did, plcUrl, fetch, signal) {
  /** @type {string} */
  let url;
  if (did.startsWith('did:plc:')) {
    url = `${plcUrl}/${did}/data`;
  } else if (did.startsWith('did:web:')) {
    const host = did.slice('did:web:'.length).replaceAll(':', '/');
    url = `https://${host}/.well-known/did.json`;
  } else {
    return null;
  }
  try {
    const res = await fetch(url, { signal });
    if (!res.ok) return null;
    return await res.json();
  } catch {
    return null;
  }
}

/**
 * @param {any} doc
 * @returns {string|null}
 */
function pdsOf(doc) {
  const service = (doc?.service ?? []).find(
    (/** @type {{id: string}} */ entry) => entry.id.endsWith('#atproto_pds'),
  );
  return service?.serviceEndpoint ? String(service.serviceEndpoint) : null;
}

/**
 * @param {any} doc
 * @returns {string|null}
 */
function handleOf(doc) {
  const aka = (doc?.alsoKnownAs ?? []).find((/** @type {string} */ name) =>
    name.startsWith('at://'),
  );
  return aka ? aka.slice('at://'.length) : null;
}

/**
 * @template T
 * @param {T[]} items
 * @param {number} size
 * @param {(item: T) => Promise<void>} work
 * @returns {Promise<void>}
 */
async function pool(items, size, work) {
  for (let start = 0; start < items.length; start += size) {
    await Promise.all(items.slice(start, start + size).map(work));
  }
}