chadtmiller.com/pds.js

Switch repository

Loading…

/**
 * The merger: what puts a pull request's work on a repository's canonical
 * branch, given the ports its server holds.
 *
 * Two halves meet here. The pack format is this package's, and the authority
 * over the repository is the PDS's, so the PDS builds this from what it has
 * and calls it. The same seam the write guard uses, for the same reason:
 * neither package depends on the other.
 *
 * The work is in a branch of somebody else's copy on their own server, so
 * their objects are read across the network and the ones this copy lacks ride
 * in a new bundle. Nothing is written to their repository.
 *
 * Workers-clean: no node imports.
 */

import { parseBundle } from './bundle.js';
import { buildMerge } from './merge.js';
import { indexPacks } from './objects.js';
import {
  concatChunks,
  indexPartsOf,
  parseRepoRecord,
  repoNameOf,
  rkeyOf,
  splitIntoChunks,
} from './record.js';
import { createLazyStore } from './store.js';

/**
 * What a merge commit says it is, in git's own words.
 *
 * `git merge topic` writes "Merge branch 'topic'", and adds "of <where>" for a
 * branch it fetched from somewhere else. Both are true here: the branch came
 * from another account's copy, and the account is what stands where git puts a
 * URL. A squash takes the change's own subject instead, because a squash is
 * the change rather than a record of taking it.
 * @param {string} ref - full ref name
 * @param {string} [handle] - whose copy it came from
 * @returns {string}
 */
export function mergeSubject(ref, handle) {
  const branch = ref.replace(/^refs\/heads\//, '');
  return handle
    ? `Merge branch '${branch}' of ${handle}`
    : `Merge branch '${branch}'`;
}

/** What a bundle chunk and an index chunk are stored as. */
/**
 * One account's copy of a repository, from their own server: their record
 * key and the record. A copy is its holder's own record and carries their
 * own key, so the name is what the two copies share and what finds it.
 * @param {string} service
 * @param {string} did
 * @param {string} name
 * @returns {Promise<{rkey: string, record: import('./record.js').GitRepoRecord}|null>}
 */
async function readCopy(service, did, name) {
  const params = new URLSearchParams({
    repo: did,
    collection: 'dev.pdsjs.git.repo',
    limit: '100',
  });
  const answer = await fetch(
    `${service}/xrpc/com.atproto.repo.listRecords?${params}`,
  );
  if (!answer.ok) return null;
  const body =
    /** @type {{records?: Array<{uri?: unknown, value?: unknown}>}} */ (
      await answer.json()
    );
  for (const row of body.records ?? []) {
    if (typeof row.uri !== 'string') continue;
    const rkey = rkeyOf(row.uri);
    if (repoNameOf(row.value, rkey) !== name) continue;
    return { rkey, record: parseRepoRecord(row.value) };
  }
  return null;
}

const BUNDLE_MIME = 'application/x-git-bundle';
const INDEX_MIME = 'application/x-git-pack-index';

/**
 * An object store over one bundle chain, reading whichever way the chain
 * allows. A chain whose bundles all carry an index answers single objects out
 * of ranged reads; any other chain is downloaded whole.
 * @param {import('./record.js').GitRepoRecord} record
 * @param {(cid: string) => Promise<Uint8Array|null>} read
 * @returns {Promise<import('./merge.js').ObjectStore>}
 */
async function openObjects(record, read) {
  /** @param {string} cid */
  const need = async (cid) => {
    const bytes = await read(cid);
    if (!bytes) throw new Error(`bundle chunk ${cid} is not in blob storage`);
    return bytes;
  };

  const indexed =
    record.bundles.length > 0 &&
    record.bundles.every((bundle) => indexPartsOf(bundle).length > 0);
  if (indexed) {
    const store = await createLazyStore(record, need, async (cid, start, end) =>
      (await need(cid)).subarray(start, end),
    );
    return {
      meta: (sha) => store.meta(sha),
      get: (sha) => store.get(sha),
    };
  }

  /** @type {Uint8Array[]} */
  const packs = [];
  for (const bundle of record.bundles) {
    const chunks = await Promise.all(
      bundle.parts.map((part) => need(part.ref.$link)),
    );
    packs.push(parseBundle(concatChunks(chunks)).pack);
  }
  const objects = await indexPacks(packs);
  return {
    meta: (sha) => {
      const object = objects.get(sha);
      return object ? { type: object.type, size: object.data.length } : null;
    },
    get: async (sha) => objects.get(sha) ?? null,
  };
}

/**
 * @typedef {Object} MergerContext
 * @property {(did: string, collection: string, rkey: string) => Promise<Object|null>} readRecord
 * @property {(did: string, collection: string, rkey: string, value: Object) => Promise<void>} writeRecord
 * @property {(did: string, cid: string) => Promise<Uint8Array|null>} readBlob
 * @property {(did: string, data: Uint8Array, mimeType: string) => Promise<{ref: {$link: string}, mimeType: string, size: number}>} writeBlob
 * @property {(cid: string, recordUri: string) => Promise<void>} linkBlob
 * @property {number} chunkSize - the size a bundle is split to, which is the
 *   server's blob upload limit
 * @property {(did: string) => Promise<{did: string, service: string}>} locate -
 *   where an account's repository lives, from its DID document
 * @property {() => Promise<string|null>} handle - this account's handle, for
 *   the identity the merge commit records
 * @property {() => number} [now] - epoch milliseconds, for a test that wants
 *   the commit id to hold still
 */

/**
 * Build a merger over one server's ports.
 * @param {MergerContext} ctx
 */
export function createGitMerger(ctx) {
  const now = ctx.now ?? (() => Date.now());

  return {
    /**
     * Merge one pull request.
     * @param {Object} merge
     * @param {string} merge.did - the account that owns the repository
     * @param {string} merge.rkey - the repository's record key here
     * @param {string} merge.name - what the repository is called, which is
     *   what finds the copy holding the work
     * @param {string} merge.from - the account whose copy holds the work
     * @param {string} merge.ref - full ref name of their branch
     * @param {'merge'|'squash'} merge.mode
     * @param {string} [merge.handle] - the branch owner's handle, for the
     *   subject a merge writes
     * @param {string} [merge.subject] - a squash is the change, so its
     *   subject is the change's. A merge is not, and names the branch.
     * @param {string} [merge.body]
     */
    async merge({ did, rkey, name, from, ref, mode, handle, subject, body }) {
      const value = await ctx.readRecord(did, 'dev.pdsjs.git.repo', rkey);
      if (!value) throw new Error(`no such repository: ${name}`);
      const record = parseRepoRecord(value);
      const branch =
        /** @type {{defaultBranch?: string}} */ (value).defaultBranch || 'main';
      const canonicalRef = `refs/heads/${branch}`;
      const mainTip = record.refs.find(
        (entry) => entry.name === canonicalRef,
      )?.sha;
      if (!mainTip) throw new Error(`${canonicalRef} is not a branch here`);

      // Their copy, on their own server. The DID document is the only thing
      // that says where an account lives, and the copy is their own record
      // under their own key, so the name is what finds it.
      const where = await ctx.locate(from);
      const theirs = await readCopy(where.service, from, name);
      if (!theirs) throw new Error(`that account holds no copy of ${name}`);
      const branchTip = theirs.record.refs.find(
        (entry) => entry.name === ref,
      )?.sha;
      if (!branchTip) throw new Error(`${ref} is not a branch of that copy`);

      const built = await buildMerge({
        main: await openObjects(record, (cid) => ctx.readBlob(did, cid)),
        source: await openObjects(theirs.record, async (cid) => {
          const blob = await fetch(
            `${where.service}/xrpc/com.atproto.sync.getBlob?${new URLSearchParams(
              { did: from, cid },
            )}`,
          );
          return blob.ok ? new Uint8Array(await blob.arrayBuffer()) : null;
        }),
        mainTip,
        branchTip,
        canonicalRef,
        mode,
        from: { did: from, repo: theirs.rkey, ref },
        by: {
          name: (await ctx.handle()) || did,
          // A merge is signed by the account, and an account is a DID. The
          // reserved .invalid domain says plainly that no mail reaches it.
          email: `${did.replace(/^did:/, '')}@did.invalid`,
        },
        at: now(),
        subject: subject?.trim() || mergeSubject(ref, handle),
        body,
      });

      // Split the way a push over the remote helper splits, because the same
      // limit applies to both.
      const parts = [];
      for (const chunk of splitIntoChunks(built.bundle, ctx.chunkSize)) {
        parts.push(await ctx.writeBlob(did, chunk, BUNDLE_MIME));
      }
      const indexParts = [];
      for (const chunk of splitIntoChunks(built.index, ctx.chunkSize)) {
        indexParts.push(await ctx.writeBlob(did, chunk, INDEX_MIME));
      }

      await ctx.writeRecord(did, 'dev.pdsjs.git.repo', rkey, {
        .../** @type {Object} */ (value),
        refs: [
          ...record.refs.filter((entry) => entry.name !== canonicalRef),
          { name: canonicalRef, sha: built.sha },
        ],
        bundles: [...record.bundles, { heads: built.heads, parts, indexParts }],
      });
      // The link table is what keeps these blobs out of orphan cleanup, and
      // it holds them against the record's own address. A URI that named
      // anything else would leave the bundle bytes to be reaped.
      const uri = `at://${did}/dev.pdsjs.git.repo/${rkey}`;
      for (const part of [...parts, ...indexParts]) {
        await ctx.linkBlob(part.ref.$link, uri);
      }

      return {
        sha: built.sha,
        ref: canonicalRef,
        mode,
        objects: built.objects,
      };
    },
  };
}