chadtmiller.com/pds.js

Switch repository

Loading…

// 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<typeof createMemoryStorage>} */
  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);
    }
  });
});