chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { zstdDecompressSync } from 'node:zlib';
import { cborEncodeDagCbor } from '@pdsjs/core/repo';
import { describe, expect, it } from 'vitest';
import {
decodeBlock,
decodeBlockColumns,
decodeSegment,
getBlock,
planSnapshot,
replayArchive,
rowToEvent,
} from '../src/archive.js';
import { blockFrame, encodeBlockColumns } from './helpers/archive-block.js';
const COLLECTION = 'dev.pdsjs.git.pull';
const DID = 'did:plc:stranger';
/**
* @param {Uint8Array} bytes
* @returns {Promise<Uint8Array>}
*/
const decompress = async (bytes) =>
new Uint8Array(zstdDecompressSync(Buffer.from(bytes)));
const RECORD = {
ref: 'refs/heads/fix',
subject: { uri: 'at://did:plc:owner/dev.pdsjs.git.repo/proj' },
};
describe('decodeBlockColumns', () => {
it('round-trips the rows the encoder wrote', () => {
const rows = [
{
seq: 1000,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
payload: new Uint8Array([1, 2]),
},
{
seq: 1001,
kind: 3,
did: DID,
collection: COLLECTION,
rkey: '3kdel',
rev: '',
payload: new Uint8Array(),
},
];
const decoded = decodeBlockColumns(encodeBlockColumns(rows));
expect(decoded).toHaveLength(2);
expect(decoded[0]).toMatchObject({
seq: 1000,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
});
expect([...decoded[0].payload]).toEqual([1, 2]);
expect(decoded[1].kind).toBe(3);
expect(decoded[1].rkey).toBe('3kdel');
});
it('decodes a zstd frame through the injected decompressor', async () => {
const rows = [
{
seq: 5,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
payload: new Uint8Array(),
},
];
const decoded = await decodeBlock(blockFrame(rows), decompress);
expect(decoded[0].seq).toBe(5);
});
});
describe('decodeSegment', () => {
it('picks the block frames out of a sealed segment file', async () => {
const rows = [
{
seq: 7,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
payload: new Uint8Array(),
},
];
const frame = blockFrame(rows);
const header = 256;
const footer = header + 8 + frame.length;
const file = new Uint8Array(footer + 52);
const view = new DataView(file.buffer);
file.set(new TextEncoder().encode('jss0'), 0);
view.setUint32(14, 1, true);
view.setBigUint64(58, BigInt(footer), true);
view.setUint32(header, frame.length, true);
file.set(frame, header + 8);
view.setBigUint64(footer, BigInt(header), true);
view.setUint32(footer + 8, frame.length, true);
const blocks = await decodeSegment(file, decompress);
expect(blocks).toHaveLength(1);
expect(blocks[0][0].seq).toBe(7);
});
});
describe('rowToEvent', () => {
it('maps a create onto a record write whose value is the CBOR record', () => {
const row = {
seq: 9,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
payload: cborEncodeDagCbor(RECORD),
};
expect(rowToEvent(row)).toEqual({
did: DID,
collection: COLLECTION,
rkey: '3kpull',
cid: null,
uri: `at://${DID}/${COLLECTION}/3kpull`,
seq: 9,
value: RECORD,
});
});
it('maps a resync create onto a record write', () => {
const row = {
seq: 10,
kind: 7,
did: DID,
collection: COLLECTION,
rkey: '3kresync',
rev: '3krev',
payload: cborEncodeDagCbor(RECORD),
};
expect(rowToEvent(row)).toMatchObject({
uri: expect.stringContaining('3kresync'),
});
});
it('maps a delete onto a marker', () => {
const row = {
seq: 11,
kind: 3,
did: DID,
collection: COLLECTION,
rkey: '3kdel',
rev: '',
payload: new Uint8Array(),
};
expect(rowToEvent(row)).toEqual({
did: DID,
collection: COLLECTION,
rkey: '3kdel',
uri: `at://${DID}/${COLLECTION}/3kdel`,
seq: 11,
deleted: true,
});
});
it('reads the handle from an identity payload and skips one without', () => {
expect(
rowToEvent({
seq: 12,
kind: 4,
did: DID,
collection: '',
rkey: '',
rev: '',
payload: cborEncodeDagCbor({ handle: 'stranger.example.com' }),
}),
).toEqual({
kind: 'identity',
did: DID,
handle: 'stranger.example.com',
seq: 12,
});
expect(
rowToEvent({
seq: 13,
kind: 4,
did: DID,
collection: '',
rkey: '',
rev: '',
payload: new Uint8Array(),
}),
).toBe(null);
});
it('reads the status from an account payload and skips one without', () => {
expect(
rowToEvent({
seq: 14,
kind: 5,
did: DID,
collection: '',
rkey: '',
rev: '',
payload: cborEncodeDagCbor({ active: false, status: 'deleted' }),
}),
).toEqual({ kind: 'account', did: DID, status: 'deleted', seq: 14 });
expect(
rowToEvent({
seq: 15,
kind: 5,
did: DID,
collection: '',
rkey: '',
rev: '',
payload: new Uint8Array(),
}),
).toBe(null);
});
});
describe('planSnapshot', () => {
it('sends the filter and window and returns the plan', async () => {
/** @type {Array<{url: string, init: RequestInit}>} */
const calls = [];
const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
async (input, init) => {
calls.push({ url: String(input), init: init ?? {} });
return new Response(
JSON.stringify({
sealedTipSeq: 2000,
plannedThroughSeq: 2000,
segments: [],
}),
{ status: 200 },
);
}
);
const plan = await planSnapshot({
service: 'https://jetstream.example',
collections: [COLLECTION],
afterSeq: 42,
beforeSeq: 100,
apiKey: 'key',
fetch,
});
expect(plan.sealedTipSeq).toBe(2000);
const body = JSON.parse(String(calls[0].init.body));
expect(body).toMatchObject({
collections: [COLLECTION],
afterSeq: 42,
beforeSeq: 100,
});
const headers = /** @type {Record<string, string>} */ (
calls[0].init.headers
);
expect(headers.authorization).toBe('Bearer key');
});
it('backs off a 429 and retries after Retry-After', async () => {
let tries = 0;
const fetch = async () => {
tries += 1;
if (tries === 1) {
return new Response('{"error":"byte limit exceeded"}', {
status: 429,
headers: { 'retry-after': '0' },
});
}
return new Response(
JSON.stringify({ sealedTipSeq: 1, plannedThroughSeq: 1, segments: [] }),
{ status: 200 },
);
};
await planSnapshot({
service: 'https://jetstream.example',
fetch,
});
expect(tries).toBe(2);
});
});
describe('getBlock', () => {
it('downloads the frame for one block index', async () => {
const frame = new Uint8Array([1, 2, 3]);
const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
async (input) => {
expect(String(input)).toContain('segment=seg_0000000000.jss');
expect(String(input)).toContain('blockIndex=3');
return new Response(/** @type {BodyInit} */ (frame), { status: 200 });
}
);
expect(
await getBlock({
service: 'https://jetstream.example',
segment: 'seg_0000000000.jss',
blockIndex: 3,
fetch,
}),
).toEqual(frame);
});
});
describe('replayArchive', () => {
it('plans, downloads, folds and checkpoints to the sealed tip', async () => {
const rows = [
{
seq: 10,
kind: 1,
did: DID,
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
payload: cborEncodeDagCbor(RECORD),
},
{
seq: 11,
kind: 3,
did: DID,
collection: COLLECTION,
rkey: '3kdel',
rev: '',
payload: new Uint8Array(),
},
{
seq: 12,
kind: 1,
did: 'did:plc:other',
collection: 'app.bsky.feed.post',
rkey: '3kpost',
rev: '',
payload: cborEncodeDagCbor({}),
},
];
const frame = blockFrame(rows);
const plans = [
{
sealedTipSeq: 2000,
plannedThroughSeq: 1000,
segments: [
{
name: 'seg_0000000000.jss',
index: 0,
mode: 'blocks',
blocks: [{ first: 0, last: 0 }],
},
],
},
{
sealedTipSeq: 2000,
plannedThroughSeq: 2000,
segments: [],
},
];
let planCalls = 0;
const fetch = /** @type {import('../src/archive.js').FetchLike} */ (
async (input, _init) => {
if (String(input).includes('planSnapshot')) {
const plan = plans[planCalls];
planCalls += 1;
return new Response(JSON.stringify(plan), { status: 200 });
}
return new Response(/** @type {BodyInit} */ (frame), { status: 200 });
}
);
/** @type {any[]} */
const events = [];
/** @type {number[]} */
const checkpoints = [];
const tip = await replayArchive({
service: 'https://jetstream.example',
collections: [COLLECTION],
afterSeq: 0,
fetch,
decompress,
onEvent: (event) => {
events.push(event);
},
onSeq: (seq) => {
checkpoints.push(seq);
},
});
expect(tip).toBe(2000);
expect(events.map((event) => event.uri)).toEqual([
`at://${DID}/${COLLECTION}/3kpull`,
`at://${DID}/${COLLECTION}/3kdel`,
]);
expect(events[0].value).toEqual(RECORD);
expect(events[1].deleted).toBe(true);
expect(checkpoints).toEqual([1000, 2000]);
expect(planCalls).toBe(2);
});
});