// A commit drops the blocks it leaves unreachable, getRepo sweeps what was // stored before, and then reads the block table as the repo. import { PersonalDataServer } from '@pdsjs/core'; import { parseCarFile } from '@pdsjs/core/car'; import { cborDecode, cborEncodeDagCbor, cidToString } from '@pdsjs/core/repo'; import { beforeEach, describe, expect, test } from 'vitest'; import { createMemorySharedStorage, createMemoryStorage, } from './helpers/memory-storage.js'; const DID = 'did:plc:collecttestaaaaaaaaaaaaa'; const PASSWORD = 'test-password'; const COLLECTION = 'app.bsky.feed.post'; describe('collecting blocks', () => { /** @type {ReturnType} */ let storage; /** @type {PersonalDataServer} */ let pds; /** @type {string} */ let token; beforeEach(async () => { storage = createMemoryStorage(); pds = new PersonalDataServer({ actorStorage: /** @type {any} */ (storage), sharedStorage: /** @type {any} */ (createMemorySharedStorage()), blobs: { async get() { return null; }, async put() {}, async delete() {}, }, jwtSecret: 'test-secret', hostname: 'pds.example.com', password: PASSWORD, }); await pds.fetch( new Request(`https://pds.example.com/init?did=${DID}`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ did: DID, privateKey: Buffer.from( crypto.getRandomValues(new Uint8Array(32)), ).toString('hex'), handle: 'collect.test', password: PASSWORD, }), }), ); const res = await pds.fetch( new Request( 'https://pds.example.com/xrpc/com.atproto.server.createSession', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ identifier: DID, password: PASSWORD }), }, ), ); token = (await res.json()).accessJwt; expect(token, 'session setup failed').toBeTruthy(); }); /** * @param {string} method * @param {object} body */ async function write(method, body) { const res = await pds.fetch( new Request(`https://pds.example.com/xrpc/com.atproto.repo.${method}`, { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${token}`, }, body: JSON.stringify({ repo: DID, collection: COLLECTION, ...body }), }), ); expect(res.status, `${method} ${JSON.stringify(body)}`).toBe(200); } /** @param {string} text */ const post = (text) => ({ $type: COLLECTION, text, createdAt: '2026-10-29T00:00:00.000Z', }); /** The CIDs of every block the head reaches, by walking its tree. */ async function reached() { const head = /** @type {{cid: string}} */ (await storage.getLatestCommit()); const car = await pds._sync.buildFullRepoCar(head.cid); return [...parseCarFile(car).blocks.keys()].sort(); } const stored = () => [...storage._blocks.keys()].sort(); async function getRepo() { const res = await pds.fetch( new Request( `https://pds.example.com/xrpc/com.atproto.sync.getRepo?did=${DID}`, ), ); expect(res.status).toBe(200); return parseCarFile(new Uint8Array(await res.arrayBuffer())); } /** Writes enough to split and merge the tree: creates, updates, deletes. */ async function churn() { for (let i = 0; i < 60; i++) { await write('createRecord', { rkey: `r${i}`, record: post(`p${i}`) }); } for (let i = 0; i < 60; i += 3) { await write('putRecord', { rkey: `r${i}`, record: post(`edit ${i}`) }); } for (let i = 1; i < 60; i += 2) { await write('deleteRecord', { rkey: `r${i}` }); } } test('a commit keeps only what its head reaches', async () => { await storage.setBlocksCollected(true); await churn(); expect(stored()).toEqual(await reached()); }); test('a deleted record is no longer readable by its CID', async () => { await storage.setBlocksCollected(true); await write('createRecord', { rkey: 'gone', record: post('secret') }); const cid = /** @type {{cid: string}} */ ( await storage.getRecord(`at://${DID}/${COLLECTION}/gone`) ).cid; expect(await storage.getBlock(cid)).not.toBeNull(); await write('deleteRecord', { rkey: 'gone' }); expect(await storage.getBlock(cid)).toBeNull(); }); test('a record another record shares stays when one goes', async () => { await storage.setBlocksCollected(true); await write('createRecord', { rkey: 'one', record: post('same') }); await write('createRecord', { rkey: 'two', record: post('same') }); const cid = /** @type {{cid: string}} */ ( await storage.getRecord(`at://${DID}/${COLLECTION}/two`) ).cid; await write('deleteRecord', { rkey: 'one' }); expect(await storage.getBlock(cid)).not.toBeNull(); expect(stored()).toEqual(await reached()); }); test('getRepo sweeps once, then reads the table as the repo', async () => { await churn(); // As a store from before commits collected: an old tree, a stray block. const stray = cborEncodeDagCbor({ stray: true }); storage._blocks.set( 'bafyreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku', stray, ); expect(stored().length).toBeGreaterThan((await reached()).length); const car = await getRepo(); expect(await storage.getBlocksCollected()).toBe(true); expect(stored()).toEqual(await reached()); expect([...car.blocks.keys()].sort()).toEqual(await reached()); const head = /** @type {{cid: string}} */ (await storage.getLatestCommit()); expect(car.roots).toEqual([head.cid]); // Read from the table now: a block the walk would miss shows up. storage._blocks.set( 'bafyreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku', stray, ); expect((await getRepo()).blocks.size).toBe(car.blocks.size + 1); }); test('a rebuilt tree is swept by the next getRepo', async () => { await churn(); await getRepo(); // Drift the tree so the rebuild changes the root. await storage.putRecord( `at://${DID}/${COLLECTION}/zzz`, 'bafyreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku', COLLECTION, 'zzz', cborEncodeDagCbor({ stray: true }), ); storage._blocks.set( 'bafyreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku', cborEncodeDagCbor({ stray: true }), ); const rebuilt = await pds.rebuildMst(); expect(rebuilt?.changed).toBe(true); expect(await storage.getBlocksCollected()).toBe(false); expect(stored().length).toBeGreaterThan((await reached()).length); await getRepo(); expect(stored()).toEqual(await reached()); }); test('sync.getRecord carries the path to the record, not the tree', async () => { await churn(); const res = await pds.fetch( new Request( `https://pds.example.com/xrpc/com.atproto.sync.getRecord?did=${DID}&collection=${COLLECTION}&rkey=r0`, ), ); const car = parseCarFile(new Uint8Array(await res.arrayBuffer())); const all = (await reached()).length; expect(car.blocks.size).toBeLessThan(all / 4); // The path leads from the commit to the record. const commit = cborDecode( /** @type {Uint8Array} */ (car.blocks.get(car.roots[0])), ); let cid = cidToString(commit.data); const key = `${COLLECTION}/r0`; for (;;) { const node = cborDecode(/** @type {Uint8Array} */ (car.blocks.get(cid))); let prevKey = ''; let next = node.l ? cidToString(node.l) : null; let found = null; for (const entry of node.e) { const k = prevKey.slice(0, entry.p) + new TextDecoder().decode(entry.k); prevKey = k; if (k === key) found = cidToString(entry.v); if (k < key) next = entry.t ? cidToString(entry.t) : null; } if (found) { expect(car.blocks.has(found)).toBe(true); break; } expect(next, 'the path ends before the record').not.toBeNull(); cid = /** @type {string} */ (next); } }); });