chadtmiller.com/pds.js

Switch repository

Loading…

// 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<string, Uint8Array>} */
  const blocks = new Map();
  /** @type {Map<string, {cid: string, value: Uint8Array}>} */
  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<typeof createSyncHandlers>} 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<Response>} */ (
      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}`),
    );
  });
});