import { buildCarFile, CID, cborEncode, cborEncodeDagCbor, cidToBytes, cidToString, createCid, } from '@pdsjs/core/repo'; import { describe, expect, it } from 'vitest'; import { commitWrites, decodeFrame } from '../src/firehose.js'; const COLLECTION = 'dev.pdsjs.git.repo'; const DID = 'did:plc:abc123'; /** * A #commit frame carrying one record write, built the way the server builds * it so the decoder is tested against the real encoding. * @param {{path: string, value: Record}[]} writes */ async function commitFrame(writes) { const blocks = []; const ops = []; for (const write of writes) { const data = cborEncodeDagCbor(write.value); const cid = cidToString(await createCid(data)); blocks.push({ cid, data }); ops.push({ action: 'create', path: write.path, cid: new CID(cidToBytes(cid)), }); } const header = cborEncode({ op: 1, t: '#commit' }); const body = cborEncodeDagCbor({ seq: 42, repo: DID, rev: '3ktest', blocks: blocks.length ? buildCarFile(blocks[0].cid, blocks) : new Uint8Array(0), ops, time: '2026-08-18T00:00:00.000Z', }); const frame = new Uint8Array(header.length + body.length); frame.set(header, 0); frame.set(body, header.length); return frame; } describe('decodeFrame', () => { it('splits the header from the body', async () => { const frame = await commitFrame([ { path: `${COLLECTION}/proj`, value: { name: 'proj' } }, ]); const { header, body } = decodeFrame(frame); expect(header).toEqual({ op: 1, t: '#commit' }); expect(body.seq).toBe(42); expect(body.repo).toBe(DID); }); it('decodes an error frame', () => { const header = cborEncode({ op: -1 }); const body = cborEncode({ error: 'FutureCursor' }); const frame = new Uint8Array(header.length + body.length); frame.set(header, 0); frame.set(body, header.length); expect(decodeFrame(frame).header.op).toBe(-1); expect(decodeFrame(frame).body.error).toBe('FutureCursor'); }); }); describe('commitWrites', () => { it('returns the record whole, with its uri and cid', async () => { const value = { name: 'proj', refs: [{ name: 'refs/heads/main', sha: 'aa' }], }; const frame = await commitFrame([{ path: `${COLLECTION}/proj`, value }]); const { body } = decodeFrame(frame); const writes = commitWrites(body, [COLLECTION]); expect(writes).toHaveLength(1); expect(writes[0].did).toBe(DID); expect(writes[0].collection).toBe(COLLECTION); expect(writes[0].rkey).toBe('proj'); expect(writes[0].uri).toBe(`at://${DID}/${COLLECTION}/proj`); expect(writes[0].seq).toBe(42); expect(writes[0].value.refs).toEqual(value.refs); expect(writes[0].cid).toMatch(/^b[a-z2-7]+$/); }); it('tags each write with the collection it landed in', async () => { const other = 'dev.pdsjs.git.checkRequest'; const frame = await commitFrame([ { path: `${COLLECTION}/proj`, value: { name: 'proj' } }, { path: `${other}/proj_refs_heads_main`, value: { repo: 'proj' } }, { path: 'app.bsky.feed.post/xyz', value: { text: 'hello' } }, ]); const writes = commitWrites(decodeFrame(frame).body, [COLLECTION, other]); expect(writes.map((write) => write.collection)).toEqual([ COLLECTION, other, ]); expect(writes[1].rkey).toBe('proj_refs_heads_main'); }); it('ignores writes to other collections', async () => { const frame = await commitFrame([ { path: 'app.bsky.feed.post/xyz', value: { text: 'hello' } }, ]); expect(commitWrites(decodeFrame(frame).body, [COLLECTION])).toEqual([]); }); it('ignores a delete, which carries no record', () => { const body = { repo: DID, seq: 1, ops: [{ action: 'delete', path: `${COLLECTION}/proj`, cid: null }], blocks: new Uint8Array(0), }; expect(commitWrites(body, [COLLECTION])).toEqual([]); }); it('skips an op whose block is absent, as in a tooBig commit', () => { const body = { repo: DID, seq: 1, ops: [ { action: 'create', path: `${COLLECTION}/proj`, cid: cidToBytes( 'bafyreib2rxk3rh6kzwq6rhcbo6ykvzvyq5zqjnvpsxbrnfvqvjqqbqrqvi', ), }, ], blocks: new Uint8Array(0), }; expect(commitWrites(body, [COLLECTION])).toEqual([]); }); });