chadtmiller.com/pds.js

Switch repository

Loading…

/**
 * The live half of the appview: jetstream events folded into the store.
 *
 * Every write is idempotent, because the stream is at-least-once: a record
 * upserts by uri, a delete removes, and replaying an event already indexed
 * changes nothing. The cursor advances on a timer rather than per event, so
 * a quiet network still moves the checkpoint forward.
 *
 * A `replay` option hands the history to the archive: on a first boot the
 * archive is replayed from the configured floor and the tail resumes at the
 * sealed tip; on a restart whose stored cursor the server refuses, the gap
 * is replayed before the tail tries again. Archive and live share one
 * cursor, the last folded sequence.
 */

import { replayArchive } from './archive.js';
import { subscribeJetstream } from './jetstream.js';

/** How often the checkpoint is written while events flow. */
const CHECKPOINT_MS = 5_000;

/**
 * @typedef {Object} WatchOptions
 * @property {import('./store.js').AppViewStore} store
 * @property {string} service - jetstream base URL
 * @property {string[]} collections - the collections to index
 * @property {string[]} [dids] - only these accounts
 * @property {number} [cursor] - resume position; the store's is used when absent
 * @property {AbortSignal} [signal]
 * @property {(message: string) => void} [onNotice]
 * @property {(event: import('./store.js').RecordWrite|import('./store.js').RecordDelete|{kind: 'identity', did: string, handle: string, seq: number}|{kind: 'account', did: string, status: string|undefined, seq: number}) => void} [onEvent]
 *   what the fold did, for metrics and tests
 * @property {(seq: number) => void} [onSeq] - every frame, for tests and
 *   callers that decide for themselves when they have seen enough
 * @property {ReplayOptions} [replay] - the archive to catch up from, when
 *   the caller wants history the live lookback window cannot serve
 */

/**
 * @typedef {Object} ReplayOptions
 * @property {string} [service] - the archive's Jetstream base URL; the
 *   live service is used when absent
 * @property {string} [apiKey] - the archive's bearer token, when metered
 * @property {number} [afterSeq] - the first-boot floor; 0 replays everything
 * @property {(bytes: Uint8Array) => Promise<Uint8Array>} decompress - what
 *   unpacks the archive's zstd block frames
 */

/**
 * Fold one event into the store: a write applies, a delete removes, an
 * identity updates the handle, a deleted account takes its records with it.
 * @param {import('./store.js').AppViewStore} store
 * @param {import('./store.js').RecordWrite|import('./store.js').RecordDelete|{kind: 'identity', did: string, handle: string, seq: number}|{kind: 'account', did: string, status: string|undefined, seq: number}} event
 */
export function foldEvent(store, event) {
  if ('deleted' in event) {
    store.deleteRecord(event.uri);
  } else if ('kind' in event) {
    if (event.kind === 'identity') {
      store.setIdentity(event.did, event.handle);
    } else if (event.status === 'deleted') {
      store.clearAccount(event.did);
    }
  } else {
    store.putRecord(event);
  }
}

/**
 * Watch the network until the signal aborts, folding every record write,
 * handle change and account deletion into the store. Resolves on abort.
 * @param {WatchOptions} options
 * @returns {Promise<void>}
 */
export async function watch(options) {
  const { store, signal, onNotice = () => {} } = options;
  const fold = (/** @type {Parameters<typeof foldEvent>[1]} */ event) => {
    foldEvent(store, event);
    options.onEvent?.(event);
  };

  /** The last folded sequence, persisted on a timer. @type {number|undefined} */
  let written;
  /** The highest sequence seen since the last write. @type {number|undefined} */
  let pending;
  /** @param {number} seq */
  const checkpoint = (seq) => {
    pending = seq;
    options.onSeq?.(seq);
  };
  const timer = setInterval(() => {
    if (pending !== undefined && pending !== written) {
      store.setCursor('jetstream', pending);
      written = pending;
      pending = undefined;
    }
  }, CHECKPOINT_MS);
  if (typeof timer === 'object' && timer && 'unref' in timer) {
    /** @type {{unref: () => void}} */ (timer).unref();
  }

  try {
    if (options.replay) {
      await watchWithReplay(options, fold, checkpoint);
    } else {
      await subscribeJetstream({
        service: options.service,
        collections: options.collections,
        dids: options.dids,
        cursor: options.cursor ?? store.cursor('jetstream'),
        signal,
        onNotice,
        onEvent: fold,
        onSeq: checkpoint,
      });
    }
  } finally {
    clearInterval(timer);
    if (pending !== undefined) store.setCursor('jetstream', pending);
  }
}

/**
 * Catch up from the archive, then tail live. The archive is the history
 * the live lookback window cannot serve; a cursor the server refuses sends
 * the tail back to it.
 * @param {WatchOptions} options
 * @param {(event: any) => void} fold
 * @param {(seq: number) => void} checkpoint
 * @returns {Promise<void>}
 */
async function watchWithReplay(options, fold, checkpoint) {
  const { store, signal, onNotice = () => {} } = options;
  const replay = /** @type {ReplayOptions} */ (options.replay);
  const service = replay.service ?? options.service;

  /**
   * Replay the archive from one sequence and fold the gap. The sealed tip
   * returned is the resume point the tail starts at.
   * @param {number} from
   * @returns {Promise<number>}
   */
  const gap = async (from) => {
    onNotice(`replaying the archive from ${from}`);
    return replayArchive({
      service,
      collections: options.collections,
      dids: options.dids,
      apiKey: replay.apiKey,
      decompress: replay.decompress,
      afterSeq: from,
      signal,
      onNotice,
      onEvent: fold,
      onSeq: checkpoint,
    });
  };

  let cursor = options.cursor ?? store.cursor('jetstream');
  if (cursor === undefined) {
    cursor = await gap(replay.afterSeq ?? 0);
    store.setCursor('jetstream', cursor);
  }

  await subscribeJetstream({
    service: options.service,
    collections: options.collections,
    dids: options.dids,
    cursor,
    signal,
    onNotice,
    onEvent: fold,
    onSeq: checkpoint,
    // A refused cursor replays the gap from the refused sequence, then
    // hands the tail the new tip to resume from.
    onCursorTooOld: async (refused) => {
      const tip = await gap(refused);
      store.setCursor('jetstream', tip);
      return tip;
    },
  });
}