chadtmiller.com/pds.js

Switch repository

Loading…

/**
 * The issue statements about one repository, read on the repository's own
 * server.
 *
 * An issue is a record in its filer's repo, a reply is a record naming its
 * parent, and the owner's disposition record is what a reader shows as the
 * state. Every one of them names the repository at `subject.uri`. The
 * listed voices are read directly from their own servers; anybody else is
 * found through the discovery port and read from theirs, so a lying index
 * can point at nothing but cannot forge a word.
 *
 * The server answers with the statements. Threading and state are the
 * reader's, in issues.js, and the page computes them.
 *
 * Workers-clean: no node imports.
 */

import { collaboratorsOf, reviewersOf } from './rules.js';

/** @typedef {import('./pull-requests.js').Contributor} Contributor */
/** @typedef {import('./discovery.js').DiscoveryPort} DiscoveryPort */

const REPO_COLLECTION = 'dev.pdsjs.git.repo';
const ISSUE_COLLECTION = 'dev.pdsjs.git.issue';

/** Records read per listing, the most a PDS answers in one page. */
const LIST_LIMIT = 100;

/** Records followed per discovery answer, so one noisy index stays bounded. */
const MAX_DISCOVERED = 100;

const READ_TIMEOUT_MS = 5000;

/**
 * @typedef {Object} IssueReaderContext
 * @property {(did: string) => Promise<Contributor|null>} resolveContributor
 * @property {(did: string, collection: string, rkey: string) => Promise<Object|null>} readRecord -
 *   this account's own records
 * @property {DiscoveryPort} discovery
 * @property {typeof fetch} [fetch] - what carries a request to another server; the global by default
 */

/**
 * One statement, as the page reads it: the record's fields, its address,
 * and who said it.
 * @typedef {Record<string, unknown> & {uri: string, cid: string, by: Contributor, indexed?: boolean}} IssueStatement
 */

/**
 * @param {IssueReaderContext} ctx
 */
export function createIssueReader(ctx) {
  const send = ctx.fetch ?? ((input, init) => globalThis.fetch(input, init));

  /**
   * @param {string} url
   * @returns {Promise<Record<string, unknown>|null>}
   */
  async function readJson(url) {
    try {
      const res = await send(url, {
        signal: AbortSignal.timeout(READ_TIMEOUT_MS),
      });
      if (!res.ok) return null;
      return /** @type {Record<string, unknown>} */ (await res.json());
    } catch {
      // An unreachable server costs its words, not the thread.
      return null;
    }
  }

  /**
   * One voice's issue statements about the subject, from their own server.
   * @param {Contributor} who
   * @param {string} subjectUri
   * @returns {Promise<IssueStatement[]>}
   */
  async function listVoice(who, subjectUri) {
    const params = new URLSearchParams({
      repo: who.did,
      collection: ISSUE_COLLECTION,
      limit: String(LIST_LIMIT),
    });
    const body = await readJson(
      `${who.pds}/xrpc/com.atproto.repo.listRecords?${params}`,
    );
    const rows = /** @type {unknown[]} */ (body?.records ?? []);
    /** @type {IssueStatement[]} */
    const found = [];
    for (const entry of rows) {
      const row = /** @type {Record<string, unknown>} */ (entry ?? {});
      const value = /** @type {Record<string, unknown>} */ (row.value ?? {});
      const subject = /** @type {{uri?: unknown}} */ (value.subject ?? {});
      if (subject.uri !== subjectUri) continue;
      if (typeof row.uri !== 'string' || typeof row.cid !== 'string') continue;
      found.push({ ...value, uri: row.uri, cid: row.cid, by: who });
    }
    return found;
  }

  /**
   * The statements the discovery port names, each read from its author.
   * @param {string} subjectUri
   * @param {Set<string>} except - DIDs read directly already
   * @returns {Promise<IssueStatement[]>}
   */
  async function strangers(subjectUri, except) {
    const entries = (await ctx.discovery.about(subjectUri, ISSUE_COLLECTION))
      .filter((entry) => !except.has(entry.did))
      .slice(0, MAX_DISCOVERED);
    const found = await Promise.all(
      entries.map(async (entry) => {
        const who = await ctx.resolveContributor(entry.did);
        if (!who) return null;
        const params = new URLSearchParams({
          repo: who.did,
          collection: ISSUE_COLLECTION,
          rkey: entry.rkey,
        });
        const body = await readJson(
          `${who.pds}/xrpc/com.atproto.repo.getRecord?${params}`,
        );
        if (
          !body ||
          typeof body.uri !== 'string' ||
          typeof body.cid !== 'string'
        ) {
          return null;
        }
        const value = /** @type {Record<string, unknown>} */ (body.value ?? {});
        const subject = /** @type {{uri?: unknown}} */ (value.subject ?? {});
        // The index said the record names this repository. The record is
        // what says so.
        if (subject.uri !== subjectUri) return null;
        /** @type {IssueStatement} */
        const statement = {
          ...value,
          uri: body.uri,
          cid: body.cid,
          by: who,
          indexed: true,
        };
        return statement;
      }),
    );
    return found.filter((one) => one !== null);
  }

  return {
    /**
     * Everything said in this repository's issue threads.
     * @param {string} did - the repository owner
     * @param {string} rkey - the repository's record key
     * @returns {Promise<{statements: IssueStatement[]}|null>} null for no such repository
     */
    async listIssues(did, rkey) {
      if (!(await ctx.readRecord(did, REPO_COLLECTION, rkey))) return null;
      const config = await ctx.readRecord(did, 'dev.pdsjs.git.config', rkey);
      const subjectUri = `at://${did}/${REPO_COLLECTION}/${rkey}`;
      const dids = [
        ...new Set([did, ...collaboratorsOf(config), ...reviewersOf(config)]),
      ];
      const voices = (
        await Promise.all(dids.map((one) => ctx.resolveContributor(one)))
      ).filter((who) => who !== null);

      const [listed, found] = await Promise.all([
        Promise.all(voices.map((who) => listVoice(who, subjectUri))),
        strangers(subjectUri, new Set(dids)),
      ]);
      const held = new Set(listed.flat().map((one) => one.uri));
      return {
        statements: [
          ...listed.flat(),
          ...found.filter((one) => !held.has(one.uri)),
        ],
      };
    },
  };
}