// Sync handler tests - repo reads, CAR shape, proof paths, deactivated status import { describe, expect, it } from 'vitest'; import { parseCarFile } from '../packages/core/src/car.js'; import { generateKeyPair } from '../packages/core/src/crypto.js'; import { createSyncHandlers } from '../packages/core/src/handlers/xrpc-sync.js'; import { loadRepositoryFromCar } from '../packages/core/src/loader.js'; import { cborDecode, cidToString } from '../packages/core/src/repo.js'; import { buildRepositoryCar } from '../packages/readonly/src/build-car.js'; const DID = 'did:plc:synctest'; const COLLECTION = 'app.bsky.feed.post'; const RKEY = '3syncrec'; /** Actor storage over a real loaded repository. */ function createMockStorage() { /** @type {Map} */ const blocks = new Map(); /** @type {Map} */ const records = new Map(); /** @type {Array<{seq: number, cid: string, rev: string}>} */ const commits = []; return { blocks, commits, records, async getBlock(/** @type {string} */ cid) { return blocks.get(cid) || null; }, async putBlock(/** @type {string} */ cid, /** @type {Uint8Array} */ data) { blocks.set(cid, data); }, async getRecord(/** @type {string} */ uri) { return records.get(uri) || null; }, async putRecord( /** @type {string} */ uri, /** @type {string} */ cid, /** @type {string} */ _collection, /** @type {string} */ _rkey, /** @type {Uint8Array} */ value, ) { records.set(uri, { cid, value }); }, async listAllRecords() { return []; }, async getLatestCommit() { return commits.length > 0 ? commits[commits.length - 1] : null; }, async putCommit( /** @type {number} */ seq, /** @type {string} */ cid, /** @type {string} */ rev, ) { commits.push({ seq, cid, rev }); }, async putEvent() {}, async getDid() { return DID; }, async setDid() {}, async getHandle() { return 'sync.test'; }, async getPrivateKey() { return null; }, async setPrivateKey() {}, async getKemKey() { return null; }, async setKemKey() {}, }; } /** * Sync handlers over a real repository of one record, and `extraRecords` more. * @param {{did?: string|null, deactivated?: boolean, extraRecords?: number}} [options] */ async function createSync({ did = DID, deactivated = false, extraRecords = 0, } = {}) { const { privateKey } = await generateKeyPair(); const { carBytes } = await buildRepositoryCar({ did: DID, privateKey, records: [ { collection: COLLECTION, rkey: RKEY, record: { $type: COLLECTION, text: 'synced' }, }, ...Array.from({ length: extraRecords }, (_, i) => ({ collection: COLLECTION, rkey: `3extra${String(i).padStart(6, '0')}`, record: { $type: COLLECTION, text: `extra ${i}` }, })), ], }); const actorStorage = createMockStorage(); await loadRepositoryFromCar(carBytes, /** @type {any} */ (actorStorage)); const sync = createSyncHandlers({ actorStorage: /** @type {any} */ (actorStorage), getDid: async () => did, isDeactivated: async () => deactivated, }); return { sync, actorStorage }; } /** * @template T * @param {T|undefined} value * @param {string} what * @returns {T} */ function must(value, what) { if (value === undefined) throw new Error(`test setup has no ${what}`); return value; } /** @param {string} path */ function req(path) { return { request: new Request(`https://pds.example.com${path}`), url: new URL(`https://pds.example.com${path}`), }; } /** * @param {ReturnType} sync * @param {string} path */ async function call(sync, path) { const route = sync.routes[path.split('?')[0]]; const { request, url } = req(path); // The Route typedef declares `this: PersonalDataServer`. These handlers get // their dependencies from the factory context. They do not use `this`. const handler = /** @type {(request: Request, url: URL, auth: null) => Promise} */ ( route.handler ); return handler(request, url, null); } describe('listRepos', () => { it('reports the hosted repo with its head', async () => { const { sync, actorStorage } = await createSync(); const response = await call(sync, '/xrpc/com.atproto.sync.listRepos'); const body = await response.json(); expect(body.repos).toHaveLength(1); expect(body.repos[0].did).toBe(DID); expect(body.repos[0].head).toBe(actorStorage.commits[0].cid); expect(body.repos[0].active).toBe(true); expect(body.repos[0].status).toBeUndefined(); }); it('marks a deactivated repo and says so', async () => { const { sync } = await createSync({ deactivated: true }); const body = await ( await call(sync, '/xrpc/com.atproto.sync.listRepos') ).json(); expect(body.repos[0].active).toBe(false); expect(body.repos[0].status).toBe('deactivated'); }); it('is empty before an account exists', async () => { const { sync } = await createSync({ did: null }); const body = await ( await call(sync, '/xrpc/com.atproto.sync.listRepos') ).json(); expect(body.repos).toEqual([]); }); }); describe('getRepoStatus', () => { it('answers for the hosted did', async () => { const { sync, actorStorage } = await createSync(); const response = await call( sync, `/xrpc/com.atproto.sync.getRepoStatus?did=${DID}`, ); const body = await response.json(); expect(response.status).toBe(200); expect(body.active).toBe(true); expect(body.status).toBe('active'); expect(body.rev).toBe(actorStorage.commits[0].rev); }); it('reports deactivated status', async () => { const { sync } = await createSync({ deactivated: true }); const body = await ( await call(sync, '/xrpc/com.atproto.sync.getRepoStatus') ).json(); expect(body.active).toBe(false); expect(body.status).toBe('deactivated'); }); it('is a RepoNotFound for anyone else', async () => { const { sync } = await createSync(); const response = await call( sync, '/xrpc/com.atproto.sync.getRepoStatus?did=did:plc:someoneelse', ); expect(response.status).toBe(404); expect((await response.json()).error).toBe('RepoNotFound'); }); }); describe('getRepo', () => { it('serves the whole repo as a CAR', async () => { const { sync, actorStorage } = await createSync(); const response = await call(sync, '/xrpc/com.atproto.sync.getRepo'); expect(response.status).toBe(200); expect(response.headers.get('Content-Type')).toBe( 'application/vnd.ipld.car', ); const car = parseCarFile(new Uint8Array(await response.arrayBuffer())); expect(car.roots[0]).toBe(actorStorage.commits[0].cid); // The root must reach the commit, the MST root and the record. expect(car.blocks.size).toBeGreaterThanOrEqual(3); }); it('is a RepoNotFound for anyone else', async () => { const { sync } = await createSync(); const response = await call( sync, '/xrpc/com.atproto.sync.getRepo?did=did:plc:someoneelse', ); expect(response.status).toBe(404); }); it('lets a request that arrives meanwhile run before it ends', async () => { // More than one BLOCKS_PER_WAIT of blocks, so the walk waits at least once. const { sync } = await createSync({ extraRecords: 600 }); /** @type {string[]} */ const order = []; const served = call(sync, '/xrpc/com.atproto.sync.getRepo').then(() => { order.push('getRepo'); }); // A timer is a task, as an arriving request is. setTimeout(() => order.push('arrived'), 0); await served; expect(order).toEqual(['arrived', 'getRepo']); }); it('is a 500 naming the block when a descendant is missing', async () => { const { sync, actorStorage } = await createSync(); const commitCid = actorStorage.commits[0].cid; for (const cid of [...actorStorage.blocks.keys()]) { if (cid !== commitCid) actorStorage.blocks.delete(cid); } const response = await call(sync, '/xrpc/com.atproto.sync.getRepo'); expect(response.status).toBe(500); const body = await response.json(); expect(body.error).toBe('InternalServerError'); expect(body.message).toMatch(/Missing block/); }); }); describe('getRecord', () => { it('serves the record with its proof path', async () => { const { sync, actorStorage } = await createSync(); const response = await call( sync, `/xrpc/com.atproto.sync.getRecord?collection=${COLLECTION}&rkey=${RKEY}`, ); expect(response.status).toBe(200); const car = parseCarFile(new Uint8Array(await response.arrayBuffer())); expect(car.roots[0]).toBe(actorStorage.commits[0].cid); expect(car.blocks.size).toBeGreaterThanOrEqual(3); }); it('needs both collection and rkey', async () => { const { sync } = await createSync(); const response = await call( sync, `/xrpc/com.atproto.sync.getRecord?collection=${COLLECTION}`, ); expect(response.status).toBe(400); expect((await response.json()).error).toBe('InvalidRequest'); }); it('is a RecordNotFound for an rkey that is not there', async () => { const { sync } = await createSync(); const response = await call( sync, `/xrpc/com.atproto.sync.getRecord?collection=${COLLECTION}&rkey=nope`, ); expect(response.status).toBe(404); expect((await response.json()).error).toBe('RecordNotFound'); }); it('is a RepoNotFound for anyone else', async () => { const { sync } = await createSync(); const response = await call( sync, `/xrpc/com.atproto.sync.getRecord?did=did:plc:someoneelse&collection=${COLLECTION}&rkey=${RKEY}`, ); expect(response.status).toBe(404); expect((await response.json()).error).toBe('RepoNotFound'); }); }); describe('getLatestCommit', () => { it('answers with the head cid and rev', async () => { const { sync, actorStorage } = await createSync(); const body = await ( await call(sync, '/xrpc/com.atproto.sync.getLatestCommit') ).json(); expect(body.cid).toBe(actorStorage.commits[0].cid); expect(body.rev).toBe(actorStorage.commits[0].rev); }); it('is a RepoNotFound before the first commit', async () => { const { sync, actorStorage } = await createSync(); actorStorage.commits.length = 0; const response = await call(sync, '/xrpc/com.atproto.sync.getLatestCommit'); expect(response.status).toBe(404); expect((await response.json()).error).toBe('RepoNotFound'); }); }); describe('buildFullRepoCar', () => { it('is exported for the backup engine to reuse', async () => { const { sync, actorStorage } = await createSync(); const car = await sync.buildFullRepoCar(actorStorage.commits[0].cid); const parsed = parseCarFile(car); expect(parsed.roots[0]).toBe(actorStorage.commits[0].cid); expect(parsed.blocks.size).toBeGreaterThanOrEqual(3); }); it('throws naming the block when the root is unreachable', async () => { const { sync } = await createSync(); await expect(sync.buildFullRepoCar('bafyreimissingblock')).rejects.toThrow( /Missing block/, ); }); // A one-record repo has a single MST node, so it never reaches the recursive // paths. Enough records build subtrees in `l` and `t`. The two identical // records share a CID, which the block list must hold once. it('takes every record from a multi-level tree', async () => { const records = Array.from({ length: 60 }, (_, i) => ({ collection: COLLECTION, rkey: `3rec${String(i).padStart(4, '0')}`, record: { $type: COLLECTION, text: `post ${i}` }, })); records.push({ collection: COLLECTION, rkey: '3duplicatea', record: { $type: COLLECTION, text: 'same content' }, }); records.push({ collection: COLLECTION, rkey: '3duplicateb', record: { $type: COLLECTION, text: 'same content' }, }); const { privateKey } = await generateKeyPair(); const { carBytes } = await buildRepositoryCar({ did: DID, privateKey, records, }); const actorStorage = createMockStorage(); await loadRepositoryFromCar(carBytes, /** @type {any} */ (actorStorage)); const sync = createSyncHandlers({ actorStorage: /** @type {any} */ (actorStorage), getDid: async () => DID, isDeactivated: async () => false, }); const car = parseCarFile( await sync.buildFullRepoCar(actorStorage.commits[0].cid), ); // The duplicate pair is one block, so 62 records occupy 61 CIDs. const recordCids = new Set( [...actorStorage.records.values()].map((r) => r.cid), ); expect(recordCids.size).toBe(61); for (const cid of recordCids) { expect(car.blocks.has(cid), cid).toBe(true); } // Commit and record blocks aside, the rest are the MST nodes it walked. expect(car.blocks.size).toBeGreaterThan(recordCids.size + 1); }); it('throws when an MST node is absent', async () => { const { sync, actorStorage } = await createSync(); const commitCid = actorStorage.commits[0].cid; const mstRoot = cidToString( cborDecode(must(actorStorage.blocks.get(commitCid), 'commit block')).data, ); actorStorage.blocks.delete(mstRoot); await expect(sync.buildFullRepoCar(commitCid)).rejects.toThrow( new RegExp(`Missing block: ${mstRoot}`), ); }); // The walk reads `data`, `l` and `e` on the commit and on MST nodes. A record // is free to use those names for its own content, and those bytes are not // block CIDs. it('does not read link fields on a record', async () => { const { privateKey } = await generateKeyPair(); const { carBytes } = await buildRepositoryCar({ did: DID, privateKey, records: [ { collection: COLLECTION, rkey: RKEY, record: { $type: COLLECTION, text: 'synced', data: new Uint8Array([1, 2, 3, 4]), l: new Uint8Array([5, 6, 7, 8]), e: [{ v: new Uint8Array([9, 10]), t: new Uint8Array([11, 12]) }], }, }, ], }); const actorStorage = createMockStorage(); await loadRepositoryFromCar(carBytes, /** @type {any} */ (actorStorage)); const sync = createSyncHandlers({ actorStorage: /** @type {any} */ (actorStorage), getDid: async () => DID, isDeactivated: async () => false, }); const car = parseCarFile( await sync.buildFullRepoCar(actorStorage.commits[0].cid), ); expect(car.blocks.size).toBeGreaterThanOrEqual(3); }); it('throws when a record block is absent', async () => { const { sync, actorStorage } = await createSync(); const commitCid = actorStorage.commits[0].cid; const recordCid = must( actorStorage.records.get(`at://${DID}/${COLLECTION}/${RKEY}`), 'record', ).cid; actorStorage.blocks.delete(recordCid); await expect(sync.buildFullRepoCar(commitCid)).rejects.toThrow( new RegExp(`Missing block: ${recordCid}`), ); }); });