chadtmiller.com/pds.js

Switch repository

Loading…

import { describe, expect, it } from 'vitest';
import { jetstreamUrl, jetstreamWrite } from '../src/jetstream.js';

const COLLECTION = 'dev.pdsjs.git.pull';
const DID = 'did:plc:stranger';

/**
 * One v2 frame: an envelope naming a payload, whose $type names the kind.
 * @param {Partial<Record<string, unknown>>} commit
 * @param {Record<string, unknown>} [extra]
 */
const commitEvent = (commit, extra = {}) => ({
  $type: 'message',
  payload: {
    $type: 'network.bsky.jetstream.subscribeEvents#commit',
    did: DID,
    seq: 1000,
    time: '2026-08-13T06:47:43.959305Z',
    operation: 'create',
    collection: COLLECTION,
    rkey: '3kpull',
    rev: '3krev',
    record: { subject: { uri: 'at://did:plc:owner/dev.pdsjs.git.repo/proj' } },
    cid: 'bafypull',
    ...commit,
  },
  ...extra,
});

describe('jetstreamWrite', () => {
  it('maps a create onto a record write with the sequence', () => {
    expect(jetstreamWrite(commitEvent({}), [COLLECTION])).toEqual({
      did: DID,
      collection: COLLECTION,
      rkey: '3kpull',
      cid: 'bafypull',
      uri: `at://${DID}/${COLLECTION}/3kpull`,
      seq: 1000,
      value: {
        subject: { uri: 'at://did:plc:owner/dev.pdsjs.git.repo/proj' },
      },
    });
  });

  it('falls back to time_us when the payload names no sequence', () => {
    const write = jetstreamWrite(
      commitEvent(
        {},
        {
          payload: {
            ...commitEvent({}).payload,
            seq: undefined,
            time_us: 1000,
          },
        },
      ),
      [COLLECTION],
    );
    expect(write?.seq).toBe(1000);
  });

  it('carries a delete as a marker without a value', () => {
    expect(
      jetstreamWrite(commitEvent({ operation: 'delete' }), [COLLECTION]),
    ).toEqual({
      did: DID,
      collection: COLLECTION,
      rkey: '3kpull',
      uri: `at://${DID}/${COLLECTION}/3kpull`,
      seq: 1000,
      deleted: true,
    });
  });

  it('carries an update', () => {
    expect(
      jetstreamWrite(commitEvent({ operation: 'update' }), [COLLECTION]),
    ).not.toBe(null);
  });

  it('drops other collections, other kinds, and malformed frames', () => {
    expect(
      jetstreamWrite(commitEvent({ collection: 'app.bsky.feed.post' }), [
        COLLECTION,
      ]),
    ).toBe(null);
    expect(
      jetstreamWrite(
        {
          $type: 'message',
          payload: {
            $type: 'network.bsky.jetstream.subscribeEvents#identity',
            did: DID,
            seq: 1000,
            identity: { handle: 'stranger.example.com' },
          },
        },
        [COLLECTION],
      ),
    ).toBe(null);
    expect(jetstreamWrite(commitEvent({ record: null }), [COLLECTION])).toBe(
      null,
    );
    expect(jetstreamWrite(commitEvent({ cid: null }), [COLLECTION])).toBe(null);
    expect(jetstreamWrite(null, [COLLECTION])).toBe(null);
  });
});

describe('jetstreamUrl', () => {
  it('names the v2 endpoint, the collections, the accounts and the cursor', () => {
    const url = jetstreamUrl(
      'wss://jetstream.example',
      [COLLECTION, 'dev.pdsjs.git.issue'],
      42,
      [DID],
    );
    expect(url.pathname).toBe('/xrpc/network.bsky.jetstream.subscribeEvents');
    expect(url.searchParams.getAll('collections')).toEqual([
      COLLECTION,
      'dev.pdsjs.git.issue',
    ]);
    expect(url.searchParams.getAll('dids')).toEqual([DID]);
    expect(url.searchParams.get('cursor')).toBe('42');
  });

  it('omits a cursor of zero and no collections filter past the path', () => {
    expect(jetstreamUrl('wss://x', []).searchParams.has('cursor')).toBe(false);
    expect(jetstreamUrl('wss://x', [], 0).searchParams.has('cursor')).toBe(
      false,
    );
    expect(jetstreamUrl('http://x', [COLLECTION]).protocol).toBe('ws:');
  });
});