chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* Jetstream replay: the archive, paged, downloaded and folded.
*
* Jetstream v2 keeps the whole network in sealed segment files and serves
* it through one plan-then-download surface. `planSnapshot` returns the
* segment files or block ranges that may hold matching events, `getBlock`
* and `getSegment` stream them, and the fold converges to network truth
* the same way the live tail does: creates apply, deletes and account
* deletions remove. The archive is at-least-once, so every write here is
* idempotent.
*
* On Bluesky's public instances the replay HTTP endpoints are metered and
* take a Bearer API key; a 429 names a Retry-After in seconds. The block
* and segment payloads are exact on-disk bytes, so the columnar block
* format is decoded here and a zstd decompressor rides in on the same
* injection the live tail uses.
*/
import { cborDecode } from '@pdsjs/core/repo';
/** The kind column values the on-disk format names. */
const KIND_CREATE = 1;
const KIND_UPDATE = 2;
const KIND_DELETE = 3;
const KIND_IDENTITY = 4;
const KIND_ACCOUNT = 5;
const KIND_SYNC = 6;
const KIND_CREATE_RESYNC = 7;
/** The sealed segment's fixed header, at file offset zero. */
const HEADER_BLOCK_COUNT = 14;
const HEADER_BLOCK_INDEX = 58;
/** One block index entry, from the block index at the footer offset. */
const INDEX_ENTRY_OFFSET = 0;
const INDEX_ENTRY_SIZE = 52;
const INDEX_COMPRESSED_SIZE = 8;
/** The length prefix before a block's frame body. */
const BLOCK_LENGTH_PREFIX = 8;
/**
* What carries an archive request; the global fetch unless one is injected.
* @typedef {(input: string|URL|Request, init?: RequestInit) => Promise<Response>} FetchLike
*/
/**
* @param {number} seconds
* @param {AbortSignal} [signal]
* @returns {Promise<void>}
*/
function sleep(seconds, signal) {
return new Promise((resolve) => {
const timer = setTimeout(resolve, seconds * 1000);
signal?.addEventListener('abort', () => {
clearTimeout(timer);
resolve();
});
});
}
/**
* The headers the metered archive endpoints want. A 429 answer names a
* Retry-After in seconds; the caller backs off rather than failing.
* @param {string|undefined} apiKey
* @returns {{'content-type': string, authorization?: string}}
*/
function archiveHeaders(apiKey) {
return {
'content-type': 'application/json',
...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}),
};
}
/**
* One page of the download plan for a data pattern. `afterSeq` is
* exclusive; `beforeSeq` caps the archive tip so pagination never floats.
* @param {Object} options
* @param {string} options.service - the Jetstream base URL
* @param {string[]} [options.collections]
* @param {string[]} [options.dids]
* @param {string[]} [options.kinds]
* @param {number} [options.afterSeq]
* @param {number} [options.beforeSeq]
* @param {string} [options.apiKey]
* @param {FetchLike} [options.fetch]
* @param {AbortSignal} [options.signal]
* @returns {Promise<PlanPage>}
*/
export async function planSnapshot(options) {
const {
service,
collections = [],
dids = [],
kinds = [],
apiKey,
signal,
} = options;
/** @type {Record<string, unknown>} */
const body = { collections, dids, kinds };
if (options.afterSeq !== undefined) body.afterSeq = options.afterSeq;
if (options.beforeSeq !== undefined) body.beforeSeq = options.beforeSeq;
const send = options.fetch ?? globalThis.fetch;
const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.planSnapshot`;
for (;;) {
const res = await send(url, {
method: 'POST',
headers: archiveHeaders(apiKey),
body: JSON.stringify(body),
signal,
});
if (res.status === 429) {
const retry = Number(res.headers.get('retry-after'));
if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
continue;
}
if (!res.ok) {
throw new Error(
`replay plan refused (${res.status}): ${await res.text()}`,
);
}
return /** @type {PlanPage} */ (await res.json());
}
}
/**
* One block of a sealed segment, exactly as the archive stored it: a zstd
* frame over the columnar body. The plan names it by index.
* @param {Object} options
* @param {string} options.service
* @param {string} options.segment - the segment file name
* @param {number} options.blockIndex
* @param {string} [options.apiKey]
* @param {FetchLike} [options.fetch]
* @param {AbortSignal} [options.signal]
* @returns {Promise<Uint8Array>}
*/
export async function getBlock(options) {
const { service, segment, blockIndex, apiKey, signal } = options;
const send = options.fetch ?? globalThis.fetch;
const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.getBlock?${new URLSearchParams(
{ segment, blockIndex: String(blockIndex) },
)}`;
for (;;) {
const res = await send(url, {
headers: archiveHeaders(apiKey),
signal,
});
if (res.status === 429) {
const retry = Number(res.headers.get('retry-after'));
if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
continue;
}
if (!res.ok) {
throw new Error(
`replay block refused (${res.status}): ${await res.text()}`,
);
}
return new Uint8Array(await res.arrayBuffer());
}
}
/**
* One sealed segment file, whole. Only a plan entry in `segment` mode
* fetches this; the header and block index pick the frames out of it.
* @param {Object} options
* @param {string} options.service
* @param {string} options.segment
* @param {string} [options.apiKey]
* @param {FetchLike} [options.fetch]
* @param {AbortSignal} [options.signal]
* @returns {Promise<Uint8Array>}
*/
export async function getSegment(options) {
const { service, segment, apiKey, signal } = options;
const send = options.fetch ?? globalThis.fetch;
const url = `${service.replace(/\/+$/, '')}/xrpc/network.bsky.jetstream.getSegment?${new URLSearchParams(
{ name: segment },
)}`;
for (;;) {
const res = await send(url, {
headers: archiveHeaders(apiKey),
signal,
});
if (res.status === 429) {
const retry = Number(res.headers.get('retry-after'));
if (Number.isFinite(retry) && retry > 0) await sleep(retry, signal);
continue;
}
if (!res.ok) {
throw new Error(
`replay segment refused (${res.status}): ${await res.text()}`,
);
}
return new Uint8Array(await res.arrayBuffer());
}
}
/**
* The events one block frame holds, in stored order. The frame is a zstd
* compression over a columnar body; without an injected `decompress` it is
* read uncompressed.
* @param {Uint8Array} frame
* @param {(bytes: Uint8Array) => Promise<Uint8Array>} [decompress]
* @returns {Promise<ArchiveRow[]>}
*/
export async function decodeBlock(frame, decompress) {
const bytes = decompress ? await decompress(frame) : frame;
return decodeBlockColumns(bytes);
}
/**
* The columnar block body, exactly as `encodeBlock` writes it: one uint32
* event count, then each fixed column whole (seq, witnessed_at, indexed_at,
* kind, and the four length columns), then the concatenated variable
* columns in the same order. Columns span every event, in stored order.
* @param {Uint8Array} bytes
* @returns {ArchiveRow[]}
*/
export function decodeBlockColumns(bytes) {
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
const count = view.getUint32(0, true);
// seq, witnessed_at, indexed_at, kind, and the four length columns.
const fixedPerEvent = 8 + 8 + 8 + 1 + 1 + 2 + 1 + 1 + 4;
if (bytes.byteLength < 4 + count * fixedPerEvent) {
throw new Error('replay block is shorter than its header claims');
}
const str = (/** @type {number} */ offset, /** @type {number} */ length) =>
new TextDecoder().decode(bytes.subarray(offset, offset + length));
const seqColumn = 4;
const witnessedColumn = seqColumn + count * 8;
const indexedColumn = witnessedColumn + count * 8;
const kindColumn = indexedColumn + count * 8;
const collLenColumn = kindColumn + count;
const didLenColumn = collLenColumn + count;
const rkeyLenColumn = didLenColumn + count * 2;
const revLenColumn = rkeyLenColumn + count;
const payloadLenColumn = revLenColumn + count;
/** @type {number[]} */
const collLens = [];
/** @type {number[]} */
const didLens = [];
/** @type {number[]} */
const rkeyLens = [];
/** @type {number[]} */
const revLens = [];
/** @type {number[]} */
const payloadLens = [];
for (let i = 0; i < count; i++) {
collLens.push(view.getUint8(collLenColumn + i));
didLens.push(view.getUint16(didLenColumn + i * 2, true));
rkeyLens.push(view.getUint8(rkeyLenColumn + i));
revLens.push(view.getUint8(revLenColumn + i));
payloadLens.push(view.getUint32(payloadLenColumn + i * 4, true));
}
const sum = (/** @type {number[]} */ list) =>
list.reduce((total, length) => total + length, 0);
// Each variable column is one concatenated blob, in column order.
const varBase = 4 + count * fixedPerEvent;
const base = [varBase];
for (const lens of [collLens, didLens, rkeyLens, revLens, payloadLens]) {
base.push(base[base.length - 1] + sum(lens));
}
if (base[5] !== bytes.byteLength) {
throw new Error('replay block carries trailing bytes');
}
/** @type {number[]} */
const offset = [0, 0, 0, 0, 0];
/** @type {ArchiveRow[]} */
const rows = [];
for (let i = 0; i < count; i++) {
const column = (/** @type {number} */ n) => {
const at = base[n] + offset[n];
offset[n] += [collLens, didLens, rkeyLens, revLens, payloadLens][n][i];
return at;
};
const payloadStart = column(4);
rows.push({
seq: Number(view.getBigUint64(seqColumn + i * 8, true)),
kind: view.getUint8(kindColumn + i),
did: str(column(1), didLens[i]),
collection: str(column(0), collLens[i]),
rkey: str(column(2), rkeyLens[i]),
rev: str(column(3), revLens[i]),
payload: bytes.subarray(payloadStart, payloadStart + payloadLens[i]),
});
}
return rows;
}
/**
* Every block frame a sealed segment file holds, decoded in stored order.
* @param {Uint8Array} file - the .jss bytes
* @param {(bytes: Uint8Array) => Promise<Uint8Array>} decompress
* @returns {Promise<ArchiveRow[][]>}
*/
export async function decodeSegment(file, decompress) {
if (!decompress) {
throw new Error('replay needs a zstd decompressor');
}
const view = new DataView(file.buffer, file.byteOffset, file.byteLength);
const blockCount = view.getUint32(HEADER_BLOCK_COUNT, true);
const blockIndex = Number(view.getBigUint64(HEADER_BLOCK_INDEX, true));
/** @type {ArchiveRow[][]} */
const blocks = [];
for (let i = 0; i < blockCount; i++) {
const entry = blockIndex + i * INDEX_ENTRY_SIZE;
const offset = Number(view.getBigUint64(entry + INDEX_ENTRY_OFFSET, true));
const compressed = view.getUint32(entry + INDEX_COMPRESSED_SIZE, true);
const frame = file.subarray(
offset + BLOCK_LENGTH_PREFIX,
offset + BLOCK_LENGTH_PREFIX + compressed,
);
blocks.push(await decodeBlock(frame, decompress));
}
return blocks;
}
/**
* Fold one decoded row through the same shapes the live tail hands onEvent.
* Rows the archive cannot speak to (an identity event with no handle, an
* account event with no status) fold to nothing rather than a guess.
* @param {ArchiveRow} row
* @returns {RecordWrite|RecordDelete|IdentityEvent|AccountEvent|null}
*/
export function rowToEvent(row) {
const uri =
row.collection && row.rkey
? `at://${row.did}/${row.collection}/${row.rkey}`
: null;
switch (row.kind) {
case KIND_CREATE:
case KIND_UPDATE:
case KIND_CREATE_RESYNC: {
if (!uri) return null;
let value;
try {
value = cborDecode(row.payload);
} catch {
return null;
}
return {
did: row.did,
collection: row.collection,
rkey: row.rkey,
cid: null,
uri,
seq: row.seq,
value,
};
}
case KIND_DELETE:
return uri
? {
did: row.did,
collection: row.collection,
rkey: row.rkey,
uri,
seq: row.seq,
deleted: true,
}
: null;
case KIND_IDENTITY: {
let handle = '';
try {
const value = cborDecode(row.payload);
handle = typeof value?.handle === 'string' ? value.handle : '';
} catch {
// A row with no readable payload cannot name a handle.
}
return handle
? { kind: 'identity', did: row.did, handle, seq: row.seq }
: null;
}
case KIND_ACCOUNT: {
let status;
try {
const value = cborDecode(row.payload);
status = typeof value?.status === 'string' ? value.status : undefined;
} catch {
// A row with no readable payload cannot name a status.
}
return status
? { kind: 'account', did: row.did, status, seq: row.seq }
: null;
}
default:
return null;
}
}
/**
* Replay the archive from `afterSeq` to the sealed tip, folding each
* matching event and checkpointing each page's `plannedThroughSeq`. The
* tip is pinned on the first page, so the range never floats as new data
* arrives. Resolves with the sealed tip the caller tails live from.
* @param {Object} options
* @param {string} options.service
* @param {string[]} [options.collections]
* @param {string[]} [options.dids]
* @param {string[]} [options.kinds]
* @param {number} [options.afterSeq] - exclusive; the archive starts here
* @param {string} [options.apiKey]
* @param {FetchLike} [options.fetch]
* @param {(bytes: Uint8Array) => Promise<Uint8Array>} [options.decompress]
* @param {AbortSignal} [options.signal]
* @param {(event: RecordWrite|RecordDelete|IdentityEvent|AccountEvent) => Promise<void>|void} options.onEvent
* @param {(seq: number) => Promise<void>|void} [options.onSeq] - each page's
* continuation cursor, which a caller checkpoints between pages
* @param {(message: string) => void} [options.onNotice]
* @returns {Promise<number>} the sealed tip to resume live from
*/
export async function replayArchive(options) {
const {
service,
collections = [],
dids = [],
kinds = [],
apiKey,
decompress,
signal,
onEvent,
onSeq = () => {},
onNotice = () => {},
} = options;
const sends = options.fetch ?? globalThis.fetch;
if (!decompress) {
throw new Error('replay needs a zstd decompressor');
}
let after = options.afterSeq ?? 0;
let sealed = after;
/** @param {ArchiveRow} row */
const foldRow = async (row) => {
if (row.seq <= after) return;
if (kinds.length > 0 && !kinds.includes(kindName(row.kind))) return;
if (dids.length > 0 && !dids.includes(row.did)) return;
if (
row.collection &&
collections.length > 0 &&
!collections.includes(row.collection)
) {
return;
}
const event = rowToEvent(row);
if (event) await onEvent(event);
};
/** @param {Uint8Array} frame */
const foldFrame = async (frame) => {
for (const row of await decodeBlock(frame, decompress)) {
await foldRow(row);
}
};
/** @param {Uint8Array} file */
const foldFile = async (file) => {
for (const block of await decodeSegment(file, decompress)) {
for (const row of block) {
await foldRow(row);
}
}
};
for (;;) {
if (signal?.aborted) return sealed;
const plan = await planSnapshot({
service,
collections,
dids,
kinds,
afterSeq: after,
beforeSeq: sealed > after ? sealed : undefined,
apiKey,
fetch: sends,
signal,
});
sealed = plan.sealedTipSeq;
onNotice(
`replay ${after}..${sealed} over ${plan.segments.length} plan entries`,
);
for (const segment of plan.segments) {
if (signal?.aborted) return sealed;
if (segment.mode === 'segment') {
await foldFile(
await getSegment({
service,
segment: segment.name,
apiKey,
fetch: sends,
signal,
}),
);
} else {
for (const range of segment.blocks ?? []) {
for (let index = range.first; index <= range.last; index++) {
await foldFrame(
await getBlock({
service,
segment: segment.name,
blockIndex: index,
apiKey,
fetch: sends,
signal,
}),
);
}
}
}
}
await onSeq(plan.plannedThroughSeq);
if (plan.plannedThroughSeq >= sealed) return sealed;
after = plan.plannedThroughSeq;
}
}
/**
* The wire name of a row kind, for the kinds filter a plan accepts.
* @param {number} kind
* @returns {string}
*/
function kindName(kind) {
switch (kind) {
case KIND_CREATE:
case KIND_CREATE_RESYNC:
return 'commit';
case KIND_UPDATE:
return 'commit';
case KIND_DELETE:
return 'commit';
case KIND_IDENTITY:
return 'identity';
case KIND_ACCOUNT:
return 'account';
case KIND_SYNC:
return 'sync';
default:
return 'commit';
}
}
/** @typedef {import('./store.js').RecordWrite} RecordWrite */
/** @typedef {import('./store.js').RecordDelete} RecordDelete */
/** @typedef {import('./jetstream.js').IdentityEvent} IdentityEvent */
/** @typedef {import('./jetstream.js').AccountEvent} AccountEvent */
/**
* One decoded row of a block.
* @typedef {Object} ArchiveRow
* @property {number} seq
* @property {number} kind
* @property {string} did
* @property {string} collection
* @property {string} rkey
* @property {string} rev
* @property {Uint8Array} payload
*/
/**
* One page of a download plan.
* @typedef {Object} PlanPage
* @property {number} sealedTipSeq
* @property {number} plannedThroughSeq
* @property {PlanSegment[]} segments
*/
/**
* @typedef {Object} PlanSegment
* @property {string} name
* @property {number} index
* @property {string} mode - 'segment' or 'blocks'
* @property {Array<{first: number, last: number}>} [blocks]
*/