chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* A Jetstream consumer, narrowed to a set of collections.
*
* Jetstream v2 serves the network as JSON envelopes on one websocket,
* filtered by collection server-side. Each frame is `{$type: 'message',
* payload: {...}}`; the payload's `$type` names the event kind and its
* fields ride flat, not under a `commit` object. The v2 cursor is the
* event's sequence number, the server refuses one it can no longer serve
* with a pre-upgrade 400 named CursorTooOld, and deletes arrive as marker
* events the consumer must fold. This consumer maps every frame onto one
* of four shapes and hands it to the handler: a record write, a record
* delete, an identity event, or an account event.
*
* Workers-clean: no node imports. Without an injected `decompress` the
* subscription runs uncompressed; with one, it asks for zstd.
*/
/** Backoff bounds for reconnecting after the socket closes. */
const RECONNECT_MIN_MS = 1_000;
const RECONNECT_MAX_MS = 30_000;
/** Dead handshakes in a row before the stored cursor is given up on. */
const CURSOR_FAILURES = 3;
/** @typedef {import('./store.js').RecordWrite} RecordWrite */
/** @typedef {import('./store.js').RecordDelete} RecordDelete */
/**
* @typedef {Object} SubscribeOptions
* @property {string} service - Jetstream base URL, e.g. wss://jetstream.example
* @property {string[]} collections
* @property {AbortSignal} [signal]
* @property {(event: RecordWrite|RecordDelete|IdentityEvent|AccountEvent) => Promise<void>|void} onEvent
* @property {(seq: number) => Promise<void>|void} [onSeq] - every frame, so a
* restart resumes past the ones that indexed nothing
* @property {(message: string) => void} [onNotice]
* @property {string[]} [dids] - only these accounts
* @property {number} [cursor] - Jetstream sequence to resume after
* @property {(bytes: Uint8Array) => Promise<Uint8Array>} [decompress] - what
* unpacks a zstd frame; without it the subscription runs plain JSON
* @property {(refused: number) => Promise<number|undefined>|number|undefined} [onCursorTooOld]
* what resumes after the server refuses the stored cursor; its return
* value becomes the new cursor, and the tail continues from there
*/
/**
* The handle an account now answers to.
* @typedef {Object} IdentityEvent
* @property {'identity'} kind
* @property {string} did
* @property {string} handle
* @property {number} seq
*/
/**
* An account state change. `status: 'deleted'` is the one the fold needs:
* every record the account held is gone from the network.
* @typedef {Object} AccountEvent
* @property {'account'} kind
* @property {string} did
* @property {string|undefined} status
* @property {number} seq
*/
/**
* The event kind a frame payload names, by its `$type` suffix. A frame
* without one reads as kind 'commit' only when the v1 `kind` field says so.
* @param {Record<string, *>} event
* @returns {string}
*/
function kindOf(event) {
const type = typeof event.$type === 'string' ? event.$type : '';
const kind = type.slice(type.lastIndexOf('#') + 1);
if (kind) return kind;
return typeof event.kind === 'string' ? event.kind : '';
}
/**
* The record write one commit event carries, or null for anything else:
* other kinds and malformed frames. A delete becomes a RecordDelete so the
* fold can remove what its author withdrew.
* @param {unknown} body - one parsed Jetstream frame
* @param {string[]} collections
* @returns {RecordWrite|RecordDelete|null}
*/
export function jetstreamWrite(body, collections) {
const event = /** @type {Record<string, *>} */ (body ?? {});
const payload = event.payload ?? event;
if (kindOf(payload) !== 'commit' || typeof payload.did !== 'string') {
return null;
}
const { collection, rkey, operation } = payload;
if (typeof collection !== 'string' || !collections.includes(collection)) {
return null;
}
if (typeof rkey !== 'string') return null;
const uri = `at://${payload.did}/${collection}/${rkey}`;
const seq =
typeof payload.seq === 'number'
? payload.seq
: typeof payload.time_us === 'number'
? payload.time_us
: 0;
if (operation === 'delete') {
return { did: payload.did, collection, rkey, uri, seq, deleted: true };
}
if (operation !== 'create' && operation !== 'update') return null;
const { cid, record } = payload;
if (typeof cid !== 'string') return null;
if (!record || typeof record !== 'object') return null;
return { did: payload.did, collection, rkey, cid, uri, seq, value: record };
}
/**
* @param {string} service - Jetstream base URL, e.g. wss://host
* @param {string[]} collections
* @param {number} [cursor]
* @param {string[]} [dids]
* @returns {URL}
*/
export function jetstreamUrl(service, collections, cursor, dids) {
const url = new URL(
'/xrpc/network.bsky.jetstream.subscribeEvents',
service.replace(/^http/, 'ws'),
);
for (const collection of collections) {
url.searchParams.append('collections', collection);
}
for (const did of dids ?? []) {
url.searchParams.append('dids', did);
}
if (cursor !== undefined && cursor > 0) {
url.searchParams.set('cursor', String(cursor));
}
return url;
}
/**
* Consume a Jetstream endpoint until the signal aborts, reconnecting from
* the last sequence seen. Resolves only when aborted.
*
* A stored cursor the server refuses as too old (a v2 CursorTooOld arrives
* as a failed handshake, not a readable message) reads here as several
* connects in a row that carry no events. After that the consumer gives up
* the cursor and tails live; older records are the backfill's business.
* @param {SubscribeOptions} options
* @returns {Promise<void>}
*/
export async function subscribeJetstream(options) {
const {
service,
collections,
signal,
onEvent,
onSeq = () => {},
onNotice = () => {},
decompress,
} = options;
let cursor = options.cursor;
let backoff = RECONNECT_MIN_MS;
let deadHandshakes = 0;
while (!signal?.aborted) {
const closed = await new Promise((resolve) => {
const url = jetstreamUrl(service, collections, cursor, options.dids);
if (decompress) {
url.searchParams.set('requireCompression', 'true');
}
const socket = new WebSocket(url, decompress ? ['zstd'] : []);
/** @type {Promise<void>} */
let queue = Promise.resolve();
let sawEvent = false;
const abort = () => socket.close();
signal?.addEventListener('abort', abort, { once: true });
const finish = (/** @type {string|null} */ reason) => {
signal?.removeEventListener('abort', abort);
resolve(reason);
};
socket.addEventListener('open', () => {
backoff = RECONNECT_MIN_MS;
onNotice(`watching ${url.host} from ${cursor ?? 'now'}`);
});
// Each message is handled to completion before the next one starts,
// the same ordering promise the native consumer makes.
socket.addEventListener('message', (event) => {
queue = queue.then(async () => {
const body = JSON.parse(
String(await frameBody(event.data, decompress)),
);
const write = jetstreamWrite(body, collections);
if (write) {
sawEvent = true;
await onEvent(write);
}
const identity = identityOf(body);
if (identity) {
sawEvent = true;
await onEvent(identity);
}
const account = accountOf(body);
if (account) {
sawEvent = true;
await onEvent(account);
}
const at = seqOf(body);
if (typeof at === 'number') {
cursor = at;
await onSeq(at);
}
});
queue = queue.catch((err) => {
onNotice(
`event dropped: ${err instanceof Error ? err.message : err}`,
);
});
});
socket.addEventListener('error', () => finish('connection failed'));
socket.addEventListener('close', () => {
queue.then(() => finish(sawEvent ? null : 'no events'));
});
});
if (signal?.aborted) return;
if (closed) onNotice(`${service}: ${closed}`);
if (closed === 'no events' && cursor !== undefined) {
deadHandshakes += 1;
if (deadHandshakes >= CURSOR_FAILURES) {
if (options.onCursorTooOld) {
onNotice(`cursor ${cursor} refused; asking the caller to resume`);
const fresh = await options.onCursorTooOld(cursor);
cursor = fresh ?? undefined;
} else {
onNotice(`cursor ${cursor} refused; tailing live`);
cursor = undefined;
}
deadHandshakes = 0;
}
} else {
deadHandshakes = 0;
}
await new Promise((resolve) => setTimeout(resolve, backoff));
backoff = Math.min(backoff * 2, RECONNECT_MAX_MS);
}
}
/**
* The sequence one frame names: the payload's, the envelope's, then the
* v1 time fallback. An envelope carries its own cursor, which is the same
* number the payload names.
* @param {Record<string, *>} body
* @returns {number|undefined}
*/
function seqOf(body) {
const payload = body.payload ?? body;
if (typeof payload.seq === 'number') return payload.seq;
if (typeof body.cursor === 'number') return body.cursor;
if (typeof payload.time_us === 'number') return payload.time_us;
return undefined;
}
/**
* A body that arrives compressed is binary, handed to the injected
* decompressor. Plain JSON arrives as text.
* @param {unknown} data
* @param {((bytes: Uint8Array) => Promise<Uint8Array>)|undefined} decompress
* @returns {Promise<string>}
*/
async function frameBody(data, decompress) {
if (typeof data === 'string') return data;
const bytes =
data instanceof ArrayBuffer
? new Uint8Array(data)
: data instanceof Uint8Array
? data
: typeof Blob !== 'undefined' && data instanceof Blob
? new Uint8Array(await data.arrayBuffer())
: null;
if (!bytes) throw new Error('jetstream sent an unreadable frame');
if (!decompress) return new TextDecoder().decode(bytes);
return new TextDecoder().decode(await decompress(bytes));
}
/**
* The identity event behind a frame body, or null. The handle is what the
* index keeps; the rest of the identity body is not ours to interpret.
* @param {Record<string, *>} body
* @returns {IdentityEvent|null}
*/
function identityOf(body) {
const event = body.payload ?? body;
if (kindOf(event) !== 'identity' || typeof event.did !== 'string') {
return null;
}
const handle = event.identity?.handle;
if (typeof handle !== 'string' || !handle) return null;
return {
kind: 'identity',
did: event.did,
handle,
seq: typeof event.seq === 'number' ? event.seq : 0,
};
}
/**
* The account event behind a frame body, or null. Only the status rides
* along, and only the deleted one means anything to the fold.
* @param {Record<string, *>} body
* @returns {AccountEvent|null}
*/
function accountOf(body) {
const event = body.payload ?? body;
if (kindOf(event) !== 'account' || typeof event.did !== 'string') {
return null;
}
const status = event.status ?? event.account?.status;
return {
kind: 'account',
did: event.did,
status: typeof status === 'string' ? status : undefined,
seq: typeof event.seq === 'number' ? event.seq : 0,
};
}