@atcute/jetstream#

lightweight Jetstream subscriber for AT Protocol.

npm install @atcute/jetstream

Jetstream is a streaming service that delivers a filtered firehose of events from the AT Protocol network over WebSocket. this package provides a simple client to subscribe to these events.

usage#

subscribing to events#

create a subscription and iterate over events with for await:

import { JetstreamSubscription } from '@atcute/jetstream';

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
});

for await (const event of subscription) {
	console.log(event.kind, event.did);
}

the connection opens when you start iterating and closes when you break out of the loop. the underlying WebSocket automatically reconnects on disconnection.

filtering by collection#

use wantedCollections to receive only events for specific record types:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	wantedCollections: ['app.bsky.feed.post', 'app.bsky.feed.like'],
});

for await (const event of subscription) {
	if (event.kind === 'commit') {
		console.log(event.commit.collection, event.commit.operation);
		// -> "app.bsky.feed.post" "create"
	}
}

filtering by account#

use wantedDids to receive only events from specific accounts:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	wantedDids: ['did:plc:z72i7hdynmk6r22z27h6tvur'], // @bsky.app
});

handling event types#

jetstream delivers three kinds of events:

for await (const event of subscription) {
	switch (event.kind) {
		case 'commit': {
			// record was created, updated, or deleted
			const { collection, operation, rkey, rev } = event.commit;

			if (operation === 'create' || operation === 'update') {
				// record and cid are available on create/update
				console.log(event.commit.record);
			}

			break;
		}

		case 'identity': {
			// handle or DID document changed
			const { did, handle, seq, time } = event.identity;
			break;
		}

		case 'account': {
			// account status changed (activated, deactivated, etc.)
			const { did, active, seq, time } = event.account;
			break;
		}
	}
}

validating records#

jetstream events include the raw record data. use is() from @atcute/lexicons to validate and narrow the type:

import { JetstreamSubscription } from '@atcute/jetstream';
import { is } from '@atcute/lexicons';

import { AppBskyFeedPost } from '@atcute/bluesky';

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	wantedCollections: ['app.bsky.feed.post'],
});

for await (const event of subscription) {
	if (event.kind !== 'commit') {
		continue;
	}

	const commit = event.commit;
	if (commit.operation !== 'create') {
		continue;
	}

	// validate the record against the schema
	if (!is(AppBskyFeedPost.mainSchema, commit.record)) {
		console.warn('invalid record', commit.record);
		continue;
	}

	// commit.record is now typed as AppBskyFeedPost.$record
	console.log(`@${event.did}: ${commit.record.text}`);
}

resuming from a cursor#

jetstream supports cursors for resuming from a specific point. the cursor is a timestamp in microseconds:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	// resume from a saved cursor
	cursor: 1699900000000000,
});

// save the cursor periodically to resume later
setInterval(() => {
	localStorage.setItem('jetstream-cursor', String(subscription.cursor));
}, 5_000);

when switching between jetstream instances (e.g., when using multiple URLs for failover), the client automatically rolls back the cursor by 10 seconds to avoid missing events due to clock differences.

using multiple servers#

pass an array of URLs for automatic failover. the client randomly selects one on each connection:

const subscription = new JetstreamSubscription({
	url: [
		'wss://jetstream1.us-east.bsky.network',
		'wss://jetstream2.us-east.bsky.network',
		'wss://jetstream1.us-west.bsky.network',
		'wss://jetstream2.us-west.bsky.network',
	],
});

updating options at runtime#

change filters without reconnecting using updateOptions():

// start with all collections
const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
});

// later, filter to only posts
subscription.updateOptions({
	wantedCollections: ['app.bsky.feed.post'],
});

// add accounts to filter
subscription.updateOptions({
	wantedDids: ['did:plc:...'],
});

changes to wantedCollections and wantedDids are sent to the server without reconnecting. other option changes trigger a reconnection.

connection lifecycle callbacks#

handle connection events for logging or UI updates:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	onConnectionOpen(event) {
		console.log('connected to jetstream');
	},
	onConnectionClose(event) {
		console.log('disconnected from jetstream', event.code, event.reason);
	},
	onConnectionError(event) {
		console.error('jetstream error', event.error);
	},
});

disabling event validation#

by default, jetstream events are validated. disable this for slightly better performance if you trust the server:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	validateEvents: false,
});

note: this only disables validation of the event envelope. you should still validate records using is() from @atcute/lexicons.

WebSocket options#

pass options to the underlying partysocket WebSocket for custom reconnection behavior:

const subscription = new JetstreamSubscription({
	url: 'wss://jetstream2.us-east.bsky.network',
	ws: {
		maxRetries: 10,
		minReconnectionDelay: 1000,
		maxReconnectionDelay: 30000,
	},
});