···11+[package]
22+name = "jetstream-oxide"
33+version = "0.1.0"
44+edition = "2021"
55+66+[dependencies]
77+async-trait = "0.1.83"
88+atrium-api = { version = "0.24.7", default-features = false, features = ["namespace-appbsky"] }
99+tokio = { version = "1.41.1", features = ["full", "sync"] }
1010+tokio-tungstenite = { version = "0.24.0", features = ["connect", "native-tls-vendored", "url"] }
1111+futures-util = "0.3.31"
1212+url = "2.5.3"
1313+serde = { version = "1.0.215", features = ["derive"] }
1414+serde_json = "1.0.132"
1515+chrono = "0.4.38"
1616+zstd = "0.13.2"
1717+thiserror = "2.0.3"
1818+flume = "0.11.1"
1919+log = "0.4.22"
2020+2121+[dev-dependencies]
2222+anyhow = "1.0.93"
2323+clap = { version = "4.5.20", features = ["derive"] }
+29
README.md
···11+# jetstream-oxide
22+33+A typed Rust library for easily interacting with and consuming the
44+Bluesky [Jetstream](https://github.com/bluesky-social/jetstream)
55+service.
66+77+# Example
88+99+A small example CLI utility to show how to use this crate can be found in the `examples` directory. To run it, use the
1010+following command:
1111+1212+```sh
1313+cargo run --example basic -- --nsid "app.bsky.feed.post"
1414+```
1515+1616+This will display a real-time feed of every single post that is being made or deleted in the entire Bluesky network,
1717+right in your terminal!
1818+1919+You can filter it down to just specific accounts like this:
2020+2121+```sh
2222+cargo run --example basic -- \
2323+--nsid "app.bsky.feed.post" \
2424+--did "did:plc:inze6wrmsm7pjl7yta3oig77"
2525+```
2626+2727+This listens for posts that *I personally make*. You can substitute your own DID and make a few test posts yourself if
2828+you'd
2929+like of course!
+72
examples/basic.rs
···11+//! A very basic example of how to listen for create/delete events on a specific DID and NSID.
22+33+use atrium_api::{
44+ record::KnownRecord::AppBskyFeedPost,
55+ types::string,
66+};
77+use clap::Parser;
88+use jetstream_oxide::{
99+ events::{
1010+ commit::CommitEvent,
1111+ JetstreamEvent::Commit,
1212+ },
1313+ DefaultJetstreamEndpoints,
1414+ JetstreamCompression,
1515+ JetstreamConfig,
1616+ JetstreamConnector,
1717+};
1818+1919+#[derive(Parser, Debug)]
2020+#[command(version, about, long_about = None)]
2121+struct Args {
2222+ /// The DIDs to listen for events on, if not provided we will listen for all DIDs.
2323+ #[arg(short, long)]
2424+ did: Option<Vec<string::Did>>,
2525+ /// The NSID for the collection to listen for (e.g. `app.bsky.feed.post`).
2626+ #[arg(short, long)]
2727+ nsid: string::Nsid,
2828+}
2929+3030+#[tokio::main]
3131+async fn main() -> anyhow::Result<()> {
3232+ let args = Args::parse();
3333+3434+ let dids = args.did.unwrap_or_default();
3535+ let config = JetstreamConfig {
3636+ endpoint: DefaultJetstreamEndpoints::USEastOne.into(),
3737+ wanted_collections: vec![args.nsid.clone()],
3838+ wanted_dids: dids.clone(),
3939+ compression: JetstreamCompression::Zstd,
4040+ cursor: None,
4141+ };
4242+4343+ let jetstream = JetstreamConnector::new(config)?;
4444+ let (receiver, _) = jetstream.connect().await?;
4545+4646+ println!(
4747+ "Listening for '{}' events on DIDs: {:?}",
4848+ args.nsid.to_string(),
4949+ dids,
5050+ );
5151+5252+ while let Ok(event) = receiver.recv_async().await {
5353+ if let Commit(commit) = event {
5454+ match commit {
5555+ CommitEvent::Create { info: _, commit } => {
5656+ if let AppBskyFeedPost(record) = commit.record {
5757+ println!(
5858+ "New post created! ({})\n\n'{}'",
5959+ commit.info.rkey, record.text
6060+ );
6161+ }
6262+ }
6363+ CommitEvent::Delete { info: _, commit } => {
6464+ println!("A post has been deleted. ({})", commit.rkey);
6565+ }
6666+ _ => {}
6767+ }
6868+ }
6969+ }
7070+7171+ Ok(())
7272+}
···11+///! Various error types.
22+use std::io;
33+44+55+use thiserror::Error;
66+77+/// Possible errors that can occur when a [JetstreamConfig](crate::JetstreamConfig) that is passed
88+/// to a [JetstreamConnector](crate::JetstreamConnector) is invalid.
99+#[derive(Error, Debug)]
1010+pub enum ConfigValidationError {
1111+ #[error("too many wanted collections: {0} > 100")]
1212+ TooManyWantedCollections(usize),
1313+ #[error("too many wanted DIDs: {0} > 10,000")]
1414+ TooManyDids(usize),
1515+}
1616+1717+/// Possible errors that can occur in the process of connecting to a Jetstream instance over
1818+/// WebSockets.
1919+///
2020+/// See [JetstreamConnector::connect](crate::JetstreamConnector::connect).
2121+#[derive(Error, Debug)]
2222+pub enum ConnectionError {
2323+ #[error("invalid endpoint: {0}")]
2424+ InvalidEndpoint(#[from] url::ParseError),
2525+ #[error("failed to connect to Jetstream instance: {0}")]
2626+ WebSocketFailure(#[from] tokio_tungstenite::tungstenite::Error),
2727+ #[error("the Jetstream config is invalid (this really should not happen here): {0}")]
2828+ InvalidConfig(#[from] ConfigValidationError),
2929+}
3030+3131+/// Possible errors that can occur when receiving events from a Jetstream instance over WebSockets.
3232+///
3333+/// See [websocket_task](crate::websocket_task).
3434+#[derive(Error, Debug)]
3535+pub enum JetstreamEventError {
3636+ #[error("received websocket message that could not be deserialized as JSON: {0}")]
3737+ ReceivedMalformedJSON(#[from] serde_json::Error),
3838+ #[error("failed to load built-in zstd dictionary for decoding: {0}")]
3939+ CompressionDictionaryError(io::Error),
4040+ #[error("failed to decode zstd-compressed message: {0}")]
4141+ CompressionDecoderError(io::Error),
4242+ #[error("all receivers were dropped but the websocket connection failed to close cleanly")]
4343+ WebSocketCloseFailure,
4444+}
+40
src/events/account.rs
···11+use chrono::Utc;
22+use serde::Deserialize;
33+44+use crate::{
55+ events::EventInfo,
66+ exports,
77+};
88+99+/// An event representing a change to an account.
1010+#[derive(Deserialize, Debug)]
1111+pub struct AccountEvent {
1212+ /// Basic metadata included with every event.
1313+ #[serde(flatten)]
1414+ pub info: EventInfo,
1515+ /// Account specific data bundled with this event.
1616+ pub account: AccountData,
1717+}
1818+1919+/// Account specific data bundled with an account event.
2020+#[derive(Deserialize, Debug)]
2121+pub struct AccountData {
2222+ /// Whether the account is currently active.
2323+ pub active: bool,
2424+ /// The DID of the account.
2525+ pub did: exports::Did,
2626+ pub seq: u64,
2727+ pub time: chrono::DateTime<Utc>,
2828+ /// If `active` is `false` this will be present to explain why the account is inactive.
2929+ pub status: Option<AccountStatus>,
3030+}
3131+3232+/// The possible reasons an account might be listed as inactive.
3333+#[derive(Deserialize, Debug)]
3434+#[serde(rename_all = "lowercase")]
3535+pub enum AccountStatus {
3636+ Deactivated,
3737+ Deleted,
3838+ Suspended,
3939+ TakenDown,
4040+}
+61
src/events/commit.rs
···11+use atrium_api::record::KnownRecord;
22+use serde::Deserialize;
33+44+use crate::{
55+ events::EventInfo,
66+ exports,
77+};
88+99+/// An event representing a repo commit, which can be a `create`, `update`, or `delete` operation.
1010+#[derive(Deserialize, Debug)]
1111+#[serde(untagged, rename_all = "snake_case")]
1212+pub enum CommitEvent {
1313+ Create {
1414+ #[serde(flatten)]
1515+ info: EventInfo,
1616+ commit: CommitData,
1717+ },
1818+ Update {
1919+ #[serde(flatten)]
2020+ info: EventInfo,
2121+ commit: CommitData,
2222+ },
2323+ Delete {
2424+ #[serde(flatten)]
2525+ info: EventInfo,
2626+ commit: CommitInfo,
2727+ },
2828+}
2929+3030+/// The type of commit operation that was performed.
3131+#[derive(Deserialize, Debug)]
3232+#[serde(rename_all = "snake_case")]
3333+pub enum CommitType {
3434+ Create,
3535+ Update,
3636+ Delete,
3737+}
3838+3939+/// Basic commit specific info bundled with every event, also the only data included with a `delete`
4040+/// operation.
4141+#[derive(Deserialize, Debug)]
4242+pub struct CommitInfo {
4343+ /// The type of commit operation that was performed.
4444+ pub operation: CommitType,
4545+ pub rev: String,
4646+ pub rkey: String,
4747+ /// The NSID of the record type that this commit is associated with.
4848+ pub collection: exports::Nsid,
4949+}
5050+5151+/// Detailed data bundled with a commit event. This data is only included when the event is
5252+/// `create` or `update`.
5353+#[derive(Deserialize, Debug)]
5454+pub struct CommitData {
5555+ #[serde(flatten)]
5656+ pub info: CommitInfo,
5757+ /// The CID of the record that was operated on.
5858+ pub cid: exports::Cid,
5959+ /// The record that was operated on.
6060+ pub record: KnownRecord,
6161+}
+28
src/events/identity.rs
···11+use chrono::Utc;
22+use serde::Deserialize;
33+44+use crate::{
55+ events::EventInfo,
66+ exports,
77+};
88+99+/// An event representing a change to an identity.
1010+#[derive(Deserialize, Debug)]
1111+pub struct IdentityEvent {
1212+ /// Basic metadata included with every event.
1313+ #[serde(flatten)]
1414+ pub info: EventInfo,
1515+ /// Identity specific data bundled with this event.
1616+ pub identity: IdentityData,
1717+}
1818+1919+/// Identity specific data bundled with an identity event.
2020+#[derive(Deserialize, Debug)]
2121+pub struct IdentityData {
2222+ /// The DID of the identity.
2323+ pub did: exports::Did,
2424+ /// The handle associated with the identity.
2525+ pub handle: Option<exports::Handle>,
2626+ pub seq: u64,
2727+ pub time: chrono::DateTime<Utc>,
2828+}
+31
src/events/mod.rs
···11+pub mod account;
22+pub mod commit;
33+pub mod identity;
44+55+use serde::Deserialize;
66+77+use crate::exports;
88+99+/// Basic data that is included with every event.
1010+#[derive(Deserialize, Debug)]
1111+pub struct EventInfo {
1212+ pub did: exports::Did,
1313+ pub time_us: u64,
1414+ pub kind: EventKind,
1515+}
1616+1717+#[derive(Deserialize, Debug)]
1818+#[serde(untagged)]
1919+pub enum JetstreamEvent {
2020+ Commit(commit::CommitEvent),
2121+ Identity(identity::IdentityEvent),
2222+ Account(account::AccountEvent),
2323+}
2424+2525+#[derive(Deserialize, Debug)]
2626+#[serde(rename_all = "snake_case")]
2727+pub enum EventKind {
2828+ Commit,
2929+ Identity,
3030+ Account,
3131+}
+8
src/exports.rs
···11+//! Useful exports for third-party crates used by this project.
22+33+pub use atrium_api::types::string::{
44+ Cid,
55+ Did,
66+ Handle,
77+ Nsid,
88+};
+305
src/lib.rs
···11+pub mod error;
22+pub mod events;
33+pub mod exports;
44+55+use std::io::{
66+ Cursor,
77+ Read,
88+};
99+1010+use chrono::Utc;
1111+use futures_util::stream::StreamExt;
1212+use tokio::{
1313+ net::TcpStream,
1414+ task::JoinHandle,
1515+};
1616+use tokio_tungstenite::{
1717+ connect_async,
1818+ tungstenite::Message,
1919+ MaybeTlsStream,
2020+ WebSocketStream,
2121+};
2222+use url::Url;
2323+use zstd::dict::DecoderDictionary;
2424+2525+use crate::{
2626+ error::{
2727+ ConfigValidationError,
2828+ ConnectionError,
2929+ JetstreamEventError,
3030+ },
3131+ events::JetstreamEvent,
3232+};
3333+3434+/// The Jetstream endpoints officially provided by Bluesky themselves.
3535+///
3636+/// There are no guarantees that these endpoints will always be available, but you are free
3737+/// to run your own Jetstream instance in any case.
3838+pub enum DefaultJetstreamEndpoints {
3939+ /// `jetstream1.us-east.bsky.network`
4040+ USEastOne,
4141+ /// `jetstream2.us-east.bsky.network`
4242+ USEastTwo,
4343+ /// `jetstream1.us-west.bsky.network`
4444+ USWestOne,
4545+ /// `jetstream2.us-west.bsky.network`
4646+ USWestTwo,
4747+}
4848+4949+impl From<DefaultJetstreamEndpoints> for String {
5050+ fn from(endpoint: DefaultJetstreamEndpoints) -> Self {
5151+ match endpoint {
5252+ DefaultJetstreamEndpoints::USEastOne => {
5353+ "wss://jetstream1.us-east.bsky.network/subscribe".to_owned()
5454+ }
5555+ DefaultJetstreamEndpoints::USEastTwo => {
5656+ "wss://jetstream2.us-east.bsky.network/subscribe".to_owned()
5757+ }
5858+ DefaultJetstreamEndpoints::USWestOne => {
5959+ "wss://jetstream1.us-west.bsky.network/subscribe".to_owned()
6060+ }
6161+ DefaultJetstreamEndpoints::USWestTwo => {
6262+ "wss://jetstream2.us-west.bsky.network/subscribe".to_owned()
6363+ }
6464+ }
6565+ }
6666+}
6767+6868+/// The maximum number of wanted collections that can be requested on a single Jetstream connection.
6969+const MAX_WANTED_COLLECTIONS: usize = 100;
7070+/// The maximum number of wanted DIDs that can be requested on a single Jetstream connection.
7171+const MAX_WANTED_DIDS: usize = 10_000;
7272+7373+/// The custom `zstd` dictionary used for decoding compressed Jetstream messages.
7474+///
7575+/// Sourced from the [official Bluesky Jetstream repo.](https://github.com/bluesky-social/jetstream/tree/main/pkg/models)
7676+const JETSTREAM_ZSTD_DICTIONARY: &[u8] = include_bytes!("../zstd/dictionary");
7777+7878+/// A receiver channel for consuming Jetstream events.
7979+pub type JetstreamReceiver = flume::Receiver<JetstreamEvent>;
8080+8181+/// An internal sender channel for sending Jetstream events to [JetstreamReceiver]'s.
8282+type JetstreamSender = flume::Sender<JetstreamEvent>;
8383+8484+/// A wrapper connector type for working with a WebSocket connection to a Jetstream instance to
8585+/// receive and consume events. See [JetstreamConnector::connect] for more info.
8686+pub struct JetstreamConnector {
8787+ /// The configuration for the Jetstream connection.
8888+ config: JetstreamConfig,
8989+}
9090+9191+pub enum JetstreamCompression {
9292+ /// No compression, just raw plaintext JSON.
9393+ None,
9494+ /// Use the `zstd` compression algorithm, which can result in a ~56% smaller messages on
9595+ /// average. See [here](https://github.com/bluesky-social/jetstream?tab=readme-ov-file#compression) for more info.
9696+ Zstd,
9797+}
9898+9999+impl From<JetstreamCompression> for bool {
100100+ fn from(compression: JetstreamCompression) -> Self {
101101+ match compression {
102102+ JetstreamCompression::None => false,
103103+ JetstreamCompression::Zstd => true,
104104+ }
105105+ }
106106+}
107107+108108+pub struct JetstreamConfig {
109109+ /// A Jetstream endpoint to connect to with a WebSocket Scheme i.e.
110110+ /// `wss://jetstream1.us-east.bsky.network/subscribe`.
111111+ pub endpoint: String,
112112+ /// A list of collection [NSIDs](https://atproto.com/specs/nsid) to filter events for.
113113+ ///
114114+ /// An empty list will receive events for *all* collections.
115115+ ///
116116+ /// Regardless of desired collections, all subscribers receive
117117+ /// [AccountEvent](events::account::AccountEvent) and
118118+ /// [IdentityEvent](events::identity::Identity) events.
119119+ pub wanted_collections: Vec<exports::Nsid>,
120120+ /// A list of repo [DIDs](https://atproto.com/specs/did) to filter events for.
121121+ ///
122122+ /// An empty list will receive events for *all* repos, which is a lot of events!
123123+ pub wanted_dids: Vec<exports::Did>,
124124+ /// The compression algorithm to request and use for the WebSocket connection (if any).
125125+ pub compression: JetstreamCompression,
126126+ /// An optional timestamp to begin playback from.
127127+ ///
128128+ /// An absent cursor or a cursor from the future will result in live-tail operation.
129129+ ///
130130+ /// When reconnecting, use the time_us from your most recently processed event and maybe
131131+ /// provide a negative buffer (i.e. subtract a few seconds) to ensure gapless playback.
132132+ pub cursor: Option<chrono::DateTime<Utc>>,
133133+}
134134+135135+impl Default for JetstreamConfig {
136136+ fn default() -> Self {
137137+ JetstreamConfig {
138138+ endpoint: DefaultJetstreamEndpoints::USEastOne.into(),
139139+ wanted_collections: Vec::new(),
140140+ wanted_dids: Vec::new(),
141141+ compression: JetstreamCompression::None,
142142+ cursor: None,
143143+ }
144144+ }
145145+}
146146+147147+impl JetstreamConfig {
148148+ /// Constructs a new endpoint URL with the given [JetstreamConfig] applied.
149149+ pub fn construct_endpoint(&self, endpoint: &String) -> Result<Url, url::ParseError> {
150150+ let did_search_query = self
151151+ .wanted_dids
152152+ .iter()
153153+ .map(|s| ("wantedDids", s.to_string()));
154154+155155+ let collection_search_query = self
156156+ .wanted_collections
157157+ .iter()
158158+ .map(|s| ("wantedCollections", s.to_string()));
159159+160160+ let compression = (
161161+ "compress",
162162+ match self.compression {
163163+ JetstreamCompression::None => "false".to_owned(),
164164+ JetstreamCompression::Zstd => "true".to_owned(),
165165+ },
166166+ );
167167+168168+ let cursor = self
169169+ .cursor
170170+ .map(|c| ("cursor", c.timestamp_micros().to_string()));
171171+172172+ let params = did_search_query
173173+ .chain(collection_search_query)
174174+ .chain(std::iter::once(compression))
175175+ .chain(cursor.into_iter())
176176+ .collect::<Vec<(&str, String)>>();
177177+178178+ Url::parse_with_params(&endpoint, params)
179179+ }
180180+181181+ /// Validates the configuration to make sure it is within the limits of the Jetstream API.
182182+ ///
183183+ /// # Constants
184184+ /// The following constants are used to validate the configuration and should only be changed
185185+ /// if the Jetstream API has itself changed.
186186+ /// - [MAX_WANTED_COLLECTIONS]
187187+ /// - [MAX_WANTED_DIDS]
188188+ pub fn validate(&self) -> Result<(), ConfigValidationError> {
189189+ let collections = self.wanted_collections.len();
190190+ let dids = self.wanted_dids.len();
191191+192192+ if collections > MAX_WANTED_COLLECTIONS {
193193+ return Err(ConfigValidationError::TooManyWantedCollections(collections));
194194+ }
195195+196196+ if dids > MAX_WANTED_DIDS {
197197+ return Err(ConfigValidationError::TooManyDids(dids));
198198+ }
199199+200200+ Ok(())
201201+ }
202202+}
203203+204204+impl JetstreamConnector {
205205+ /// Create a Jetstream connector with a valid [JetstreamConfig].
206206+ ///
207207+ /// After creation, you can call [connect] to connect to the provided Jetstream instance.
208208+ pub fn new(config: JetstreamConfig) -> Result<Self, ConfigValidationError> {
209209+ // We validate the configuration here so any issues are caught early.
210210+ config.validate()?;
211211+ Ok(JetstreamConnector { config })
212212+ }
213213+214214+ /// Connects to a Jetstream instance as defined in the [JetstreamConfig].
215215+ ///
216216+ /// A [JetstreamReceiver] is returned which can be used to respond to events. When all instances
217217+ /// of this receiver are dropped, the connection and task are automatically closed.
218218+ pub async fn connect(
219219+ &self,
220220+ ) -> Result<
221221+ (
222222+ JetstreamReceiver,
223223+ JoinHandle<Result<(), JetstreamEventError>>,
224224+ ),
225225+ ConnectionError,
226226+ > {
227227+ // We validate the config again for good measure. Probably not necessary but it can't hurt.
228228+ self.config
229229+ .validate()
230230+ .map_err(ConnectionError::InvalidConfig)?;
231231+232232+ // TODO: Run some benchmarks and look into using a bounded channel instead.
233233+ let (send_channel, receive_channel) = flume::unbounded();
234234+235235+ let configured_endpoint = self
236236+ .config
237237+ .construct_endpoint(&self.config.endpoint)
238238+ .map_err(ConnectionError::InvalidEndpoint)?;
239239+240240+ let (ws_stream, _) = connect_async(&configured_endpoint)
241241+ .await
242242+ .map_err(ConnectionError::WebSocketFailure)?;
243243+244244+ let dict = DecoderDictionary::copy(JETSTREAM_ZSTD_DICTIONARY);
245245+246246+ // TODO: Internally creating and returning a tokio task might not be the best idea(?)
247247+ let handle = tokio::task::spawn(websocket_task(dict, ws_stream, send_channel));
248248+249249+ Ok((receive_channel, handle))
250250+ }
251251+}
252252+253253+/// The main task that handles the WebSocket connection and sends [JetstreamEvent]'s to any
254254+/// receivers that are listening for them.
255255+async fn websocket_task(
256256+ dictionary: DecoderDictionary<'_>,
257257+ ws: WebSocketStream<MaybeTlsStream<TcpStream>>,
258258+ send_channel: JetstreamSender,
259259+) -> Result<(), JetstreamEventError> {
260260+ // TODO: Use the write half to allow the user to change configuration settings on the fly.
261261+ let (_, mut read) = ws.split();
262262+ loop {
263263+ if let Some(Ok(message)) = read.next().await {
264264+ match message {
265265+ Message::Text(json) => {
266266+ let event = serde_json::from_str::<JetstreamEvent>(&json)
267267+ .map_err(JetstreamEventError::ReceivedMalformedJSON)?;
268268+269269+ if let Err(_) = send_channel.send(event) {
270270+ // We can assume that all receivers have been dropped, so we can close the
271271+ // connection and exit the task.
272272+ log::info!(
273273+ "All receivers for the Jetstream connection have been dropped, closing connection."
274274+ );
275275+ return Ok(());
276276+ }
277277+ }
278278+ Message::Binary(zstd_json) => {
279279+ let mut cursor = Cursor::new(zstd_json);
280280+ let mut decoder =
281281+ zstd::stream::Decoder::with_prepared_dictionary(&mut cursor, &dictionary)
282282+ .map_err(JetstreamEventError::CompressionDictionaryError)?;
283283+284284+ let mut json = String::new();
285285+ decoder
286286+ .read_to_string(&mut json)
287287+ .map_err(JetstreamEventError::CompressionDecoderError)?;
288288+289289+ let event = serde_json::from_str::<JetstreamEvent>(&json)
290290+ .map_err(JetstreamEventError::ReceivedMalformedJSON)?;
291291+292292+ if let Err(_) = send_channel.send(event) {
293293+ // We can assume that all receivers have been dropped, so we can close the
294294+ // connection and exit the task.
295295+ log::info!(
296296+ "All receivers for the Jetstream connection have been dropped, closing connection..."
297297+ );
298298+ return Ok(());
299299+ }
300300+ }
301301+ _ => {}
302302+ }
303303+ }
304304+ }
305305+}