chadtmiller.com/pds.js

Switch repository

Loading…

// Firehose tests - frame format per event type, and the subscribe handshake
import { describe, expect, it } from 'vitest';
import { createFirehoseHandlers } from '../packages/core/src/handlers/xrpc-firehose.js';
import { cborDecode, cborEncodeDagCbor } from '../packages/core/src/repo.js';

const DID = 'did:plc:firehosetestaccount2';
const COMMIT_CID =
  'bafyreiald5e2ib3gzzgreu7vp2dsrjpphhehnlh46lrk53mw7z2na5eoiq';

/**
 * @param {{events?: Array<Object>, withSocket?: boolean, isUpgrade?: boolean}} [options]
 */
function createFirehose({
  events = [],
  withSocket = true,
  isUpgrade = true,
} = {}) {
  /** @type {Uint8Array[]} */
  const sent = [];
  /** @type {Array<{cursor: number, limit: number}>} */
  const queries = [];

  const webSocket = {
    isUpgrade: () => isUpgrade,
    upgrade: (
      /** @type {Request} */ _request,
      /** @type {(sender: {send: (b: Uint8Array) => void}) => Promise<void>} */ onOpen,
    ) => {
      // A real port answers 101, which Response refuses to construct here.
      onOpen({ send: (b) => sent.push(b) });
      return new Response(null, { headers: { 'x-upgraded': 'yes' } });
    },
  };

  const firehose = createFirehoseHandlers({
    actorStorage: /** @type {any} */ ({
      async getEvents(
        /** @type {number} */ cursor,
        /** @type {number} */ limit,
      ) {
        queries.push({ cursor, limit });
        return { events };
      },
    }),
    webSocket: withSocket ? /** @type {any} */ (webSocket) : undefined,
  });
  return { firehose, sent, queries };
}

/** The two dag-cbor values a frame is made of, read back out. */
function readFrame(/** @type {Uint8Array} */ frame) {
  // cborDecode reads the first value; the body follows it in the same buffer.
  const header = cborDecode(frame);
  // Re-encode the header to learn its length, then decode the remainder.
  let offset = 1;
  while (offset < frame.length) {
    try {
      const body = cborDecode(frame.slice(offset));
      if (body && typeof body === 'object' && 'seq' in body) {
        return { header, body };
      }
    } catch {
      // keep scanning
    }
    offset += 1;
  }
  throw new Error('no body found in frame');
}

/** @param {Object} evtData */
function storedEvent(evtData, seq = 1) {
  return { seq, commit_cid: COMMIT_CID, evt: cborEncodeDagCbor(evtData) };
}

const SUBSCRIBE = '/xrpc/com.atproto.sync.subscribeRepos';

/**
 * @param {ReturnType<typeof createFirehoseHandlers>} firehose
 * @param {string} query
 */
function call(firehose, query = '') {
  const route = firehose.routes[SUBSCRIBE];
  const url = new URL(`https://pds.example.com${SUBSCRIBE}${query}`);
  const request = new Request(url);
  const handler =
    /** @type {(request: Request, url: URL, auth: null) => Response} */ (
      route.handler
    );
  return handler(request, url, null);
}

describe('formatFirehoseEvent', () => {
  it('frames a commit with its ops and cid', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        did: DID,
        rev: 'rev1',
        time: '2026-01-01T00:00:00.000Z',
        blocks: new Uint8Array([1, 2, 3]),
        ops: [
          {
            action: 'create',
            path: 'app.bsky.feed.post/abc',
            cid: { $link: COMMIT_CID },
          },
        ],
      }),
    );

    const { header, body } = readFrame(frame);
    expect(header).toEqual({ op: 1, t: '#commit' });
    expect(body.seq).toBe(1);
    expect(body.repo).toBe(DID);
    expect(body.rev).toBe('rev1');
    expect(body.ops).toHaveLength(1);
    expect(body.ops[0].action).toBe('create');
    expect(body.rebase).toBe(false);
    expect(body.tooBig).toBe(false);
    expect(body.blobs).toEqual([]);
  });

  it('carries a null cid for a delete', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        did: DID,
        rev: 'rev2',
        time: '2026-01-01T00:00:00.000Z',
        ops: [{ action: 'delete', path: 'app.bsky.feed.post/abc', cid: null }],
      }),
    );

    const { body } = readFrame(frame);
    expect(body.ops[0].cid).toBe(null);
  });

  it('frames an identity event with its handle', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        type: 'identity',
        did: DID,
        time: '2026-01-01T00:00:00.000Z',
        handle: 'alice.example.com',
      }),
    );

    const { header, body } = readFrame(frame);
    expect(header).toEqual({ op: 1, t: '#identity' });
    expect(body.did).toBe(DID);
    expect(body.handle).toBe('alice.example.com');
  });

  it('omits a handle an identity event does not carry', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        type: 'identity',
        did: DID,
        time: '2026-01-01T00:00:00.000Z',
      }),
    );

    const { body } = readFrame(frame);
    expect('handle' in body).toBe(false);
  });

  it('frames an account event with its active flag and status', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        type: 'account',
        did: DID,
        time: '2026-01-01T00:00:00.000Z',
        active: false,
        status: 'deactivated',
      }),
    );

    const { header, body } = readFrame(frame);
    expect(header).toEqual({ op: 1, t: '#account' });
    expect(body.active).toBe(false);
    expect(body.status).toBe('deactivated');
  });

  it('frames a sync event with its blocks and rev', () => {
    const { firehose } = createFirehose();
    const frame = firehose.formatFirehoseEvent(
      storedEvent({
        type: 'sync',
        did: DID,
        time: '2026-01-01T00:00:00.000Z',
        blocks: new Uint8Array([9, 8]),
        rev: 'rev3',
      }),
    );

    const { header, body } = readFrame(frame);
    expect(header).toEqual({ op: 1, t: '#sync' });
    expect(body.rev).toBe('rev3');
    expect(body.blocks).toEqual(new Uint8Array([9, 8]));
  });
});

describe('subscribeRepos', () => {
  it('is a 501 with no websocket port', () => {
    const { firehose } = createFirehose({ withSocket: false });
    const response = call(firehose);

    expect(response.status).toBe(501);
  });

  it('is a 426 without an upgrade header', () => {
    const { firehose } = createFirehose({ isUpgrade: false });
    const response = call(firehose);

    expect(response.status).toBe(426);
  });

  it('refuses a cursor that is not a number', () => {
    const { firehose } = createFirehose();
    const response = call(firehose, '?cursor=abc');

    expect(response.status).toBe(400);
  });

  it('upgrades and replays the events after the cursor', async () => {
    const { firehose, sent, queries } = createFirehose({
      events: [
        storedEvent(
          {
            did: DID,
            rev: 'rev1',
            time: '2026-01-01T00:00:00.000Z',
            ops: [],
          },
          5,
        ),
      ],
    });

    const response = call(firehose, '?cursor=4');
    expect(response.headers.get('x-upgraded')).toBe('yes');
    // The upgrade callback runs on its own, so wait a turn for the replay.
    await new Promise((r) => setTimeout(r, 0));

    expect(queries).toEqual([{ cursor: 4, limit: 1000 }]);
    expect(sent).toHaveLength(1);
    expect(readFrame(sent[0]).body.seq).toBe(5);
  });

  it('sends no history without a cursor', async () => {
    const { firehose, sent, queries } = createFirehose({
      events: [storedEvent({ did: DID, rev: 'r', time: 't', ops: [] })],
    });

    call(firehose);
    await new Promise((r) => setTimeout(r, 0));

    expect(queries).toHaveLength(0);
    expect(sent).toHaveLength(0);
  });
});