chadtmiller.com/pds.js

Switch repository

Loading…

// @pdsjs/core/handlers/xrpc-firehose - com.atproto.sync.subscribeRepos.
//
// A subscriber gets a websocket and every commit after the cursor it asked for,
// each as an AT Protocol frame: a header, then the event body, both dag-cbor.

import {
  CID,
  cborDecode,
  cborEncode,
  cborEncodeDagCbor,
  cidToBytes,
} from '../repo.js';
import { parseNumericCursor } from '../validation.js';

/**
 * @typedef {Object} FirehoseContext
 * @property {import('../ports.js').ActorStoragePort} actorStorage
 * @property {import('../ports.js').WebSocketPort} [webSocket] - Absent, subscribeRepos is not served
 */

/**
 * @param {FirehoseContext} ctx
 * @returns {{
 *   formatFirehoseEvent: (evt: {seq: number, commit_cid: string, evt: Uint8Array}) => Uint8Array,
 *   routes: import('../pds.js').Routes,
 * }}
 */
export function createFirehoseHandlers(ctx) {
  const { actorStorage, webSocket } = ctx;

  /**
   * Format event for firehose (AT Protocol frame format)
   * @param {{seq: number, commit_cid: string, evt: Uint8Array}} evt
   * @returns {Uint8Array}
   */
  function formatFirehoseEvent(evt) {
    // Decode stored event to get ops, blocks, rev, and time
    const evtData = cborDecode(evt.evt);

    // Typed (non-commit) events: #identity, #account, #sync
    if (evtData.type && evtData.type !== 'commit') {
      const typedHeader = cborEncode({ op: 1, t: `#${evtData.type}` });
      /** @type {Record<string, *>} */
      const typedBody = { seq: evt.seq, did: evtData.did, time: evtData.time };
      if (evtData.type === 'identity') {
        if (evtData.handle != null) typedBody.handle = evtData.handle;
      } else if (evtData.type === 'account') {
        typedBody.active = evtData.active;
        if (evtData.status != null) typedBody.status = evtData.status;
      } else if (evtData.type === 'sync') {
        typedBody.blocks = evtData.blocks || new Uint8Array(0);
        typedBody.rev = evtData.rev;
      }
      const encodedBody = cborEncodeDagCbor(typedBody);
      const typedFrame = new Uint8Array(
        typedHeader.length + encodedBody.length,
      );
      typedFrame.set(typedHeader, 0);
      typedFrame.set(encodedBody, typedHeader.length);
      return typedFrame;
    }

    // AT Protocol frame format: header + body
    const header = cborEncode({ op: 1, t: '#commit' });

    // Transform ops to wrap CIDs in CID class for DAG-CBOR tag 42 encoding
    // Stored events have cid as { $link: cidString } or null
    const ops = (evtData.ops || []).map(
      (
        /** @type {{action: string, path: string, cid?: {$link: string}|null, prev?: {$link: string}}} */ op,
      ) => ({
        ...op,
        cid: op.cid?.$link ? new CID(cidToBytes(op.cid.$link)) : null,
        ...(op.prev?.$link ? { prev: new CID(cidToBytes(op.prev.$link)) } : {}),
      }),
    );

    // Build body with blocks included
    // CIDs must be wrapped in CID class for proper DAG-CBOR tag 42 encoding
    // Use commit_cid from database row (not from decoded event) for reliability
    const body = cborEncodeDagCbor({
      seq: evt.seq,
      rebase: false,
      tooBig: false,
      repo: evtData.did,
      commit: new CID(cidToBytes(evt.commit_cid)),
      rev: evtData.rev,
      since: evtData.since ?? null,
      ...(evtData.prevData
        ? { prevData: new CID(cidToBytes(evtData.prevData)) }
        : {}),
      blocks: evtData.blocks || new Uint8Array(0),
      ops,
      blobs: [],
      time: evtData.time,
    });

    // Combine header and body
    const frame = new Uint8Array(header.length + body.length);
    frame.set(header, 0);
    frame.set(body, header.length);
    return frame;
  }

  /**
   * com.atproto.sync.subscribeRepos - WebSocket firehose
   * @param {Request} request
   * @param {URL} url
   * @returns {Response}
   */
  function handleSubscribeRepos(request, url) {
    // Check if WebSocket port is available
    if (!webSocket) {
      return Response.json(
        { error: 'NotImplemented', message: 'WebSocket not supported' },
        { status: 501 },
      );
    }

    // Check for WebSocket upgrade
    if (!webSocket.isUpgrade(request)) {
      return new Response('Expected WebSocket upgrade', { status: 426 });
    }

    // Validate cursor before upgrading
    const cursorParam = url.searchParams.get('cursor');
    const parsedCursor = parseNumericCursor(cursorParam);
    if (parsedCursor && !parsedCursor.valid) {
      return Response.json(
        { error: 'InvalidRequest', message: parsedCursor.error },
        { status: 400 },
      );
    }

    // Upgrade connection and send historical events
    return webSocket.upgrade(request, async (sender) => {
      if (parsedCursor?.valid) {
        const { events } = await actorStorage.getEvents(
          parsedCursor.value,
          1000,
        );
        for (const evt of events) {
          sender.send(formatFirehoseEvent(evt));
        }
      }
    });
  }
  return {
    formatFirehoseEvent,
    routes: {
      '/xrpc/com.atproto.sync.subscribeRepos': {
        handler: handleSubscribeRepos,
      },
    },
  };
}