chadtmiller.com/pds.js

Switch repository

Loading…

// @pdsjs/spaces/notify - telling a space's syncers that a repo advanced.
//
// A space's writes are never broadcast, which is the point of one. So a service
// that wants to follow a space registers with the authority through
// registerNotify, and the authority forwards each write notice it learns of. A
// syncer that receives one holds a session and can read; the notice itself
// carries no records, only the repo rev reached and the space revision the
// authority gave the update.

import { LtHash } from './lthash.js';
import { NOTICE_TIMEOUT_MS } from './outbox.js';
import { recordWriterUpdate } from './sequence.js';
import { createServiceAuth } from './service-auth.js';
import { parseSpaceUri } from './uri.js';

/**
 * Forward a write notice to every service registered for this space.
 *
 * Best effort, and deliberately so: sync correctness rests on comparing set
 * hashes, so a syncer that misses a notice finds out on the next one or on its
 * own next walk. One unreachable recipient must not fail the others, and none of
 * them may fail the write that produced the notice.
 *
 * The write handler runs this after its response through the host's
 * `background`, so no recipient delays the write.
 *
 * @param {Object} ctx
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
 * @param {() => Promise<import('./token.js').SpaceSigner>} ctx.getSigner
 * @param {typeof fetch} [ctx.fetch]
 * @param {Object} notice
 * @param {string} notice.authorityDid - the space's authority, which signs
 * @param {string} notice.space
 * @param {string} notice.repo - the account whose repo advanced
 * @param {string} notice.repoRev
 * @param {Uint8Array} notice.data - the repo's commit data after the write
 * @param {string} notice.spaceRev
 * @param {string|null} notice.prevSpaceRev - null for the space's first update
 * @returns {Promise<void>}
 */
export async function forwardToSyncers(
  ctx,
  { authorityDid, space, repo, repoRev, data, spaceRev, prevSpaceRev },
) {
  let recipients;
  try {
    recipients = await ctx.spaceStorage.listCredentialRecipients(space);
  } catch {
    return;
  }
  if (recipients.length === 0) return;

  const body = JSON.stringify({
    space,
    repo,
    repoRev,
    data: toJsonBytes(data),
    spaceRev,
    ...(prevSpaceRev ? { prevSpaceRev } : {}),
  });
  const doFetch = ctx.fetch ?? fetch;
  const signer = await ctx.getSigner();

  await Promise.all(
    recipients.map(async (recipient) => {
      try {
        const token = await createServiceAuth({
          iss: authorityDid,
          aud: recipient.serviceDid,
          lxm: 'com.atproto.space.notifyWrite',
          signer,
        });
        await doFetch(
          `${recipient.serviceEndpoint}/xrpc/com.atproto.space.notifyWrite`,
          {
            method: 'POST',
            headers: {
              'content-type': 'application/json',
              authorization: `Bearer ${token}`,
            },
            body,
            signal: AbortSignal.timeout(NOTICE_TIMEOUT_MS),
          },
        );
      } catch {
        // Best effort.
      }
    }),
  );
}

/**
 * @param {Uint8Array} bytes
 * @returns {{$bytes: string}}
 */
function toJsonBytes(bytes) {
  let binary = '';
  for (const b of bytes) binary += String.fromCharCode(b);
  return { $bytes: btoa(binary).replace(/=+$/, '') };
}

/**
 * Build the function that tells a space's authority this repo advanced, so it
 * can keep the writer set that listRepos enumerates and notify syncers.
 *
 * The function never fails a write, since the commit has already happened.
 *
 * @param {Object} ctx
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} ctx.spaceStorage
 * @param {() => Promise<string|null>} ctx.getDid
 * @param {() => Promise<import('./token.js').SpaceSigner>} ctx.getSigner
 * @param {typeof fetch} [ctx.fetch]
 * @param {(work: Promise<unknown>) => void} [ctx.background]
 * @param {ReturnType<typeof import('./outbox.js').createNotifyOutbox>} outbox
 * @returns {(space: string, commit: {rev: string, setHash: Uint8Array}) => Promise<void>}
 */
export function createWriteNotifier(ctx, outbox) {
  return async (space, commit) => {
    try {
      const { spaceDid } = parseSpaceUri(space);
      const did = await ctx.getDid();
      if (!did) return;
      const data = await new LtHash(commit.setHash).digest();
      // Writing into our own space: this server is both the repo host and the
      // authority, so it does what the authority does on receiving the
      // notification. That is both halves. A syncer registered for this space
      // is owed the notice however the write reached it, and a Durable Object
      // cannot post to itself.
      if (spaceDid === did) {
        const sequenced = await recordWriterUpdate(ctx.spaceStorage, {
          space,
          did,
          repoRev: commit.rev,
          data,
        });
        if (!sequenced) return;
        const forwarded = forwardToSyncers(ctx, {
          authorityDid: did,
          space,
          repo: did,
          repoRev: commit.rev,
          data,
          ...sequenced,
        });
        if (ctx.background) ctx.background(forwarded);
        else await forwarded;
        return;
      }
      await outbox.deliver(space, { rev: commit.rev, data });
    } catch {
      // The outbox holds the notice when delivery can be retried.
    }
  };
}