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>} commit * @param {Record} [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:'); }); });