chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { zstdCompressSync, zstdDecompressSync } from 'node:zlib';
import { afterEach, describe, expect, it } from 'vitest';
import { WebSocketServer } from 'ws';
import { subscribeJetstream } from '../src/jetstream.js';
const COLLECTION = 'dev.pdsjs.git.pull';
const DID = 'did:plc:stranger';
/**
* @param {Partial<Record<string, unknown>>} commit
* @param {Record<string, unknown>} [extra]
*/
const frame = (commit, extra = {}) => ({
$type: 'message',
payload: {
$type: 'network.bsky.jetstream.subscribeEvents#commit',
did: DID,
seq: 1000,
time: '2026-08-13T06:47:43.959305Z',
operation: 'create',
collection: COLLECTION,
rkey: '3kpull',
rev: '3krev',
record: { ref: 'refs/heads/fix' },
cid: 'bafypull',
...commit,
},
...extra,
});
describe('subscribeJetstream', () => {
/** @type {WebSocketServer} */
let server;
afterEach(() => {
server?.close();
});
it('hands each event to the handler and advances the sequence', async () => {
server = new WebSocketServer({ port: 0 });
const port = /** @type {import('node:net').AddressInfo} */ (
server.address()
).port;
server.on('connection', (socket) => {
socket.send(JSON.stringify(frame({})));
socket.send(JSON.stringify(frame({ operation: 'delete', seq: 2000 })));
socket.send(
JSON.stringify({
$type: 'message',
payload: {
$type: 'network.bsky.jetstream.subscribeEvents#identity',
did: DID,
seq: 3000,
time: '2026-08-13T06:47:43.959305Z',
identity: { handle: 'stranger.example.com' },
},
}),
);
});
const controller = new AbortController();
/** @type {any[]} */
const events = [];
/** @type {number[]} */
const seqs = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
onEvent: (event) => {
events.push(event);
},
onSeq: (seq) => {
seqs.push(seq);
if (seq === 3000) controller.abort();
},
});
expect(events).toHaveLength(3);
expect(events[0].cid).toBe('bafypull');
expect(events[1].deleted).toBe(true);
expect(events[2]).toEqual({
kind: 'identity',
did: DID,
handle: 'stranger.example.com',
seq: 3000,
});
expect(seqs).toEqual([1000, 2000, 3000]);
});
it('reconnects from the last sequence after a drop', async () => {
/** @type {WebSocketServer} */
const sockets = new WebSocketServer({ port: 0 }, () => {});
server = sockets;
const port = /** @type {import('node:net').AddressInfo} */ (
sockets.address()
).port;
/** @type {import('ws').WebSocket|undefined} */
let client;
sockets.on('connection', (socket) => {
client = socket;
socket.send(JSON.stringify(frame({ seq: 500 })));
// The drop lands once the first event is out.
setTimeout(() => socket.close(), 50);
});
const controller = new AbortController();
/** @type {string[]} */
const notices = [];
const done = subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
onNotice: (message) => notices.push(message),
onEvent: () => {},
onSeq: () => {
// The second connection carries the resume cursor.
if (notices.some((message) => message.includes('from 500'))) {
controller.abort();
}
},
});
await done;
expect(notices.some((message) => message.includes('from 500'))).toBe(true);
client?.close();
});
it('asks the caller how to resume a refused cursor and tails from its answer', async () => {
const sockets = new WebSocketServer({ port: 0 });
server = sockets;
const port = /** @type {import('node:net').AddressInfo} */ (
sockets.address()
).port;
let refused = true;
sockets.on('connection', (socket) => {
if (refused) {
socket.close(1008, 'CursorTooOld');
} else {
socket.send(JSON.stringify(frame({ seq: 200 })));
}
});
const controller = new AbortController();
/** @type {number[]} */
const refusedSeqs = [];
/** @type {string[]} */
const notices = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
cursor: 42,
signal: controller.signal,
onNotice: (message) => notices.push(message),
onEvent: () => {},
onSeq: (seq) => {
if (seq === 200) controller.abort();
},
onCursorTooOld: (refusedSeq) => {
refusedSeqs.push(refusedSeq);
refused = false;
return 100;
},
});
expect(refusedSeqs).toEqual([42]);
expect(notices.some((message) => message.includes('refused'))).toBe(true);
});
it('reads compressed frames through the injected decompressor', async () => {
const sockets = new WebSocketServer({
port: 0,
handleProtocols: () => 'zstd',
});
server = sockets;
const port = /** @type {import('node:net').AddressInfo} */ (
sockets.address()
).port;
/** @type {string[]} */
const protocols = [];
/** @type {URL[]} */
const asked = [];
sockets.on('connection', (socket, request) => {
protocols.push(socket.protocol);
asked.push(new URL(request.url ?? '', 'ws://127.0.0.1'));
socket.send(zstdCompressSync(Buffer.from(JSON.stringify(frame({})))));
});
const controller = new AbortController();
/** @type {any[]} */
const events = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
decompress: async (bytes) =>
new Uint8Array(zstdDecompressSync(Buffer.from(bytes))),
onEvent: (event) => {
events.push(event);
},
onSeq: () => controller.abort(),
});
expect(events).toHaveLength(1);
expect(events[0].cid).toBe('bafypull');
expect(protocols).toEqual(['zstd']);
expect(asked[0].searchParams.get('requireCompression')).toBe('true');
});
it('reads a binary frame that is plain JSON', async () => {
server = new WebSocketServer({ port: 0 });
const port = /** @type {import('node:net').AddressInfo} */ (
server.address()
).port;
server.on('connection', (socket) => {
socket.send(Buffer.from(JSON.stringify(frame({}))), { binary: true });
});
const controller = new AbortController();
/** @type {any[]} */
const events = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
onEvent: (event) => {
events.push(event);
},
onSeq: () => controller.abort(),
});
expect(events).toHaveLength(1);
});
it('takes the sequence from the envelope, then from the v1 timestamp', async () => {
server = new WebSocketServer({ port: 0 });
const port = /** @type {import('node:net').AddressInfo} */ (
server.address()
).port;
server.on('connection', (socket) => {
socket.send(
JSON.stringify(frame({ seq: undefined, rkey: '3kone' }, { cursor: 7 })),
);
socket.send(
JSON.stringify(frame({ seq: undefined, time_us: 8, rkey: '3ktwo' })),
);
// No sequence anywhere: the frame indexes, the cursor holds.
socket.send(JSON.stringify(frame({ seq: undefined, rkey: '3kthree' })));
});
const controller = new AbortController();
/** @type {any[]} */
const events = [];
/** @type {number[]} */
const seqs = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
onEvent: (event) => {
const write = /** @type {any} */ (event);
events.push(write);
if (write.rkey === '3kthree') controller.abort();
},
onSeq: (seq) => {
seqs.push(seq);
},
});
expect(events.map((event) => event.rkey)).toEqual([
'3kone',
'3ktwo',
'3kthree',
]);
expect(events[1].seq).toBe(8);
expect(seqs).toEqual([7, 8]);
});
it('notices an event the handler threw on and keeps reading', async () => {
server = new WebSocketServer({ port: 0 });
const port = /** @type {import('node:net').AddressInfo} */ (
server.address()
).port;
server.on('connection', (socket) => {
socket.send(JSON.stringify(frame({ rkey: '3kbad' })));
socket.send(JSON.stringify(frame({ rkey: '3kgood', seq: 1001 })));
});
const controller = new AbortController();
/** @type {string[]} */
const notices = [];
/** @type {string[]} */
const seen = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
signal: controller.signal,
onNotice: (message) => notices.push(message),
onEvent: (event) => {
const write = /** @type {any} */ (event);
if (write.rkey === '3kbad') throw new Error('index is full');
seen.push(write.rkey);
},
onSeq: (seq) => {
if (seq === 1001) controller.abort();
},
});
expect(seen).toEqual(['3kgood']);
expect(notices).toContain('event dropped: index is full');
});
it('gives up a refused cursor and tails live when no resume is offered', async () => {
const sockets = new WebSocketServer({ port: 0 });
server = sockets;
const port = /** @type {import('node:net').AddressInfo} */ (
sockets.address()
).port;
let connections = 0;
/** @type {Array<string|null>} */
const cursors = [];
sockets.on('connection', (socket, request) => {
connections += 1;
cursors.push(
new URL(request.url ?? '', 'ws://127.0.0.1').searchParams.get('cursor'),
);
if (connections <= 3) socket.close();
else socket.send(JSON.stringify(frame({ seq: 200 })));
});
const controller = new AbortController();
/** @type {string[]} */
const notices = [];
await subscribeJetstream({
service: `ws://127.0.0.1:${port}`,
collections: [COLLECTION],
cursor: 42,
signal: controller.signal,
onNotice: (message) => notices.push(message),
onEvent: () => {},
onSeq: (seq) => {
if (seq === 200) controller.abort();
},
});
expect(cursors).toEqual(['42', '42', '42', null]);
expect(notices).toContain('cursor 42 refused; tailing live');
});
});