chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { describe, expect, it } from 'vitest';
import { LtHash } from '../src/lthash.js';
import { createMemorySpaceStorage } from '../src/memory-storage.js';
import { formatSetHashElement } from '../src/path.js';
import { makeSpaceRow } from '../src/space-row.js';
import {
applyWrites,
MAX_WRITES_PER_COMMIT,
SpaceRecordAlreadyExistsError,
SpaceRecordNotFoundError,
} from '../src/writer.js';
const SPACE = 'at://did:plc:aaaa/space/com.example.forum/general';
const COL = 'com.example.post';
const create = (/** @type {string} */ rkey, /** @type {string} */ text) => ({
action: /** @type {const} */ ('create'),
collection: COL,
rkey,
record: { $type: COL, text },
});
describe('applyWrites', () => {
it('materialises the space on first write, not owned', async () => {
const store = createMemorySpaceStorage();
expect(await store.getSpace(SPACE)).toBeNull();
await applyWrites(store, { space: SPACE, writes: [create('a', 'hi')] });
const row = await store.getSpace(SPACE);
expect(row?.uri).toBe(SPACE);
expect(row?.isOwner).toBe(false);
// Components come from the URI, so listSpaces filters work on it.
expect(row?.spaceDid).toBe('did:plc:aaaa');
expect(row?.spaceType).toBe('com.example.forum');
});
it('creates a record and advances the repo', async () => {
const store = createMemorySpaceStorage();
const { rev, results } = await applyWrites(store, {
space: SPACE,
writes: [create('a', 'hi')],
});
expect(results).toHaveLength(1);
expect(results[0].action).toBe('create');
expect(results[0].prev).toBeNull();
expect(results[0].cid).toMatch(/^b[a-z2-7]+$/);
const row = await store.getSpaceRecord(SPACE, COL, 'a');
expect(row?.cid).toBe(results[0].cid);
expect(row?.repoRev).toBe(rev);
expect((await store.getSpaceRepo(SPACE))?.rev).toBe(rev);
});
it('produces a set hash matching the records it holds', async () => {
const store = createMemorySpaceStorage();
const { setHash } = await applyWrites(store, {
space: SPACE,
writes: [create('a', 'one'), create('b', 'two')],
});
const all = await store.listAllSpaceRecords(SPACE);
const expected = new LtHash();
for (const r of all) {
expected.add(formatSetHashElement(r.collection, r.rkey, r.cid));
}
expect(await new LtHash(setHash).digest()).toEqual(await expected.digest());
});
it('shares one rev across a batch and numbers the oplog in order', async () => {
const store = createMemorySpaceStorage();
const { rev } = await applyWrites(store, {
space: SPACE,
writes: [create('a', 'one'), create('b', 'two')],
});
const ops = await store.listSpaceOps(SPACE, null, 50);
expect(ops.map((o) => [o.rev, o.idx, o.rkey])).toEqual([
[rev, 0, 'a'],
[rev, 1, 'b'],
]);
});
it('rejects creating a record that already exists', async () => {
const store = createMemorySpaceStorage();
await applyWrites(store, { space: SPACE, writes: [create('a', 'one')] });
await expect(
applyWrites(store, { space: SPACE, writes: [create('a', 'again')] }),
).rejects.toBeInstanceOf(SpaceRecordAlreadyExistsError);
});
it('rejects updating or deleting a missing record', async () => {
const store = createMemorySpaceStorage();
await expect(
applyWrites(store, {
space: SPACE,
writes: [
{ action: 'update', collection: COL, rkey: 'x', record: { a: 1 } },
],
}),
).rejects.toBeInstanceOf(SpaceRecordNotFoundError);
await expect(
applyWrites(store, {
space: SPACE,
writes: [{ action: 'delete', collection: COL, rkey: 'x' }],
}),
).rejects.toBeInstanceOf(SpaceRecordNotFoundError);
});
it('resolves put to create then update', async () => {
const store = createMemorySpaceStorage();
const first = await applyWrites(store, {
space: SPACE,
writes: [{ action: 'put', collection: COL, rkey: 'a', record: { v: 1 } }],
});
expect(first.results[0].action).toBe('create');
const second = await applyWrites(store, {
space: SPACE,
writes: [{ action: 'put', collection: COL, rkey: 'a', record: { v: 2 } }],
});
expect(second.results[0].action).toBe('update');
expect(second.results[0].prev).toBe(first.results[0].cid);
});
it('honours swapCid: matching CID passes, stale CID rejects', async () => {
const store = createMemorySpaceStorage();
const first = await applyWrites(store, {
space: SPACE,
writes: [{ action: 'put', collection: COL, rkey: 'a', record: { v: 1 } }],
});
const second = await applyWrites(store, {
space: SPACE,
writes: [
{
action: 'put',
collection: COL,
rkey: 'a',
record: { v: 2 },
swapCid: first.results[0].cid,
},
],
});
expect(second.results[0].action).toBe('update');
await expect(
applyWrites(store, {
space: SPACE,
writes: [
{
action: 'put',
collection: COL,
rkey: 'a',
record: { v: 3 },
swapCid: first.results[0].cid,
},
],
}),
).rejects.toMatchObject({ code: 'InvalidSwap' });
});
it('honours swapCid null: the record must not exist', async () => {
const store = createMemorySpaceStorage();
const created = await applyWrites(store, {
space: SPACE,
writes: [
{
action: 'put',
collection: COL,
rkey: 'a',
record: { v: 1 },
swapCid: null,
},
],
});
expect(created.results[0].action).toBe('create');
await expect(
applyWrites(store, {
space: SPACE,
writes: [
{
action: 'put',
collection: COL,
rkey: 'a',
record: { v: 2 },
swapCid: null,
},
],
}),
).rejects.toMatchObject({ code: 'InvalidSwap' });
});
it('sees earlier writes in the same batch', async () => {
const store = createMemorySpaceStorage();
const { results } = await applyWrites(store, {
space: SPACE,
writes: [
create('a', 'one'),
{ action: 'update', collection: COL, rkey: 'a', record: { v: 2 } },
],
});
expect(results.map((r) => r.action)).toEqual(['create', 'update']);
expect(results[1].prev).toBe(results[0].cid);
});
it('catches a repeated create within one batch', async () => {
const store = createMemorySpaceStorage();
await expect(
applyWrites(store, {
space: SPACE,
writes: [create('a', 'one'), create('a', 'two')],
}),
).rejects.toBeInstanceOf(SpaceRecordAlreadyExistsError);
});
it('leaves the repo untouched when a batch fails partway', async () => {
const store = createMemorySpaceStorage();
const before = await applyWrites(store, {
space: SPACE,
writes: [create('a', 'one')],
});
await expect(
applyWrites(store, {
space: SPACE,
writes: [create('b', 'two'), create('a', 'conflict')],
}),
).rejects.toBeInstanceOf(SpaceRecordAlreadyExistsError);
expect(await store.getSpaceRecord(SPACE, COL, 'b')).toBeNull();
expect((await store.getSpaceRepo(SPACE))?.rev).toBe(before.rev);
const ops = await store.listSpaceOps(SPACE, null, 50);
expect(ops.map((o) => o.rev)).toEqual([before.rev]);
});
it('a delete removes the record and folds it out of the hash', async () => {
const store = createMemorySpaceStorage();
await applyWrites(store, { space: SPACE, writes: [create('a', 'one')] });
const { setHash } = await applyWrites(store, {
space: SPACE,
writes: [{ action: 'delete', collection: COL, rkey: 'a' }],
});
expect(await store.getSpaceRecord(SPACE, COL, 'a')).toBeNull();
expect(new LtHash(setHash).isEmpty()).toBe(true);
});
it('refuses the authority a write into its own space that does not exist', async () => {
const store = createMemorySpaceStorage();
const owner = 'did:plc:aaaa';
await expect(
applyWrites(store, { space: SPACE, writes: [create('a', 'x')], owner }),
).rejects.toMatchObject({ code: 'SpaceNotFound' });
expect(await store.getSpace(SPACE)).toBeNull();
await store.putSpace(makeSpaceRow(SPACE, { isOwner: true }));
await applyWrites(store, {
space: SPACE,
writes: [create('a', 'x')],
owner,
});
await store.deleteSpace(SPACE, new Date().toISOString());
await expect(
applyWrites(store, { space: SPACE, writes: [create('b', 'x')], owner }),
).rejects.toMatchObject({ code: 'SpaceNotFound' });
});
it('lets a member write into a space another authority holds', async () => {
const store = createMemorySpaceStorage();
await applyWrites(store, {
space: SPACE,
writes: [create('a', 'x')],
owner: 'did:plc:member',
});
expect(await store.getSpaceRecord(SPACE, COL, 'a')).not.toBeNull();
});
it('refuses a record over 1 MiB', async () => {
const store = createMemorySpaceStorage();
const big = {
action: /** @type {const} */ ('create'),
collection: COL,
rkey: 'big',
record: { $type: COL, text: 'x'.repeat(1024 * 1024) },
};
await expect(
applyWrites(store, { space: SPACE, writes: [big] }),
).rejects.toMatchObject({ code: 'InvalidRequest' });
});
it('rejects a batch over the commit limit', async () => {
const store = createMemorySpaceStorage();
const writes = Array.from({ length: MAX_WRITES_PER_COMMIT + 1 }, (_, i) =>
create(`r${i}`, 'x'),
);
await expect(applyWrites(store, { space: SPACE, writes })).rejects.toThrow(
/Too many writes/,
);
});
it('serialises concurrent commits to one space', async () => {
const store = createMemorySpaceStorage();
// Interleaved, these would each fold into the same starting hash and one
// contribution would be lost.
await Promise.all([
applyWrites(store, { space: SPACE, writes: [create('a', 'one')] }),
applyWrites(store, { space: SPACE, writes: [create('b', 'two')] }),
applyWrites(store, { space: SPACE, writes: [create('c', 'three')] }),
]);
const repo = await store.getSpaceRepo(SPACE);
const all = await store.listAllSpaceRecords(SPACE);
expect(all).toHaveLength(3);
const expected = new LtHash();
for (const r of all) {
expected.add(formatSetHashElement(r.collection, r.rkey, r.cid));
}
expect(await new LtHash(repo?.setHash ?? null).digest()).toEqual(
await expected.digest(),
);
});
it('a failed commit does not block later writes to the same space', async () => {
const store = createMemorySpaceStorage();
await applyWrites(store, { space: SPACE, writes: [create('a', 'one')] });
await expect(
applyWrites(store, { space: SPACE, writes: [create('a', 'dup')] }),
).rejects.toBeInstanceOf(SpaceRecordAlreadyExistsError);
const after = await applyWrites(store, {
space: SPACE,
writes: [create('b', 'two')],
});
expect(after.results[0].action).toBe('create');
});
});