event stream subscriptions and jetstream work!

- fixed issues in the codegen for subscription types,
- fixed decode to account for the header
- added jetstream as manual impl of the traits.
- added examples for both.

Cargo.lock

···
 ]
 
 [[package]]
+name = "core-foundation"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
 name = "core-foundation-sys"
 version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.60.2",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
  "jacquard-derive 0.5.4",
  "miette",
  "serde",
+ "serde_ipld_dagcbor",
  "thiserror 2.0.17",
 ]
 
···
  "bon",
  "bytes",
  "chrono",
+ "ciborium",
  "cid",
  "ed25519-dalek",
  "futures",
···
  "tracing",
  "trait-variant",
  "url",
+ "zstd",
 ]
 
 [[package]]
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
 dependencies = [
- "windows-sys 0.60.2",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
 version = "1.70.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a4895175b425cb1f87721b59f0f286c2092bd4af812243672510e1ac53e2e0ad"
+
+[[package]]
+name = "openssl-probe"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
 
 [[package]]
 name = "ouroboros"
···
  "errno",
  "libc",
  "linux-raw-sys 0.11.0",
- "windows-sys 0.60.2",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "rustls-native-certs"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9980d917ebb0c0536119ba501e90834767bffc3d60641457fd84a1f3fd337923"
+dependencies = [
+ "openssl-probe",
+ "rustls-pki-types",
+ "schannel",
+ "security-framework",
+]
+
+[[package]]
 name = "rustls-pki-types"
 version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502"
 dependencies = [
  "winapi-util",
+]
+
+[[package]]
+name = "schannel"
+version = "0.1.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "891d81b926048e76efe18581bf793546b4c0eaf8448d72be8de2bbee5fd166e1"
+dependencies = [
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "security-framework"
+version = "3.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef"
+dependencies = [
+ "bitflags",
+ "core-foundation 0.10.1",
+ "core-foundation-sys",
+ "libc",
+ "security-framework-sys",
+]
+
+[[package]]
+name = "security-framework-sys"
+version = "2.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc1f0cbffaac4852523ce30d8bd3c5cdc873501d96ff467ca09b6767bb8cd5c0"
+dependencies = [
+ "core-foundation-sys",
+ "libc",
+]
+
+[[package]]
 name = "semver"
 version = "1.0.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b"
 dependencies = [
  "bitflags",
- "core-foundation",
+ "core-foundation 0.9.4",
  "system-configuration-sys",
 ]
 
···
  "getrandom 0.3.4",
  "once_cell",
  "rustix 1.1.2",
- "windows-sys 0.60.2",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
 dependencies = [
  "futures-util",
  "log",
+ "rustls",
+ "rustls-native-certs",
+ "rustls-pki-types",
  "tokio",
+ "tokio-rustls",
  "tungstenite",
 ]
 
···
  "http",
  "httparse",
  "js-sys",
+ "rustls",
  "thiserror 1.0.69",
  "tokio",
  "tokio-tungstenite",
···
  "httparse",
  "log",
  "rand 0.8.5",
+ "rustls",
+ "rustls-pki-types",
  "sha1",
  "thiserror 1.0.69",
  "utf-8",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "db67ae75a9405634f5882791678772c94ff5f16a66535aae186e26aa0841fc8b"
 dependencies = [
- "core-foundation",
+ "core-foundation 0.9.4",
  "home",
  "jni",
  "log",
···
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.60.2",
+ "windows-sys 0.61.1",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "windows-sys"
+version = "0.61.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6f109e41dd4a3c848907eb83d5a42ea98b3769495597450cf6d153507b166f0f"
+dependencies = [
+ "windows-link 0.2.0",
+]
+
+[[package]]
 name = "windows-targets"
 version = "0.42.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "proc-macro2",
  "quote",
  "syn 2.0.106",
+]
+
+[[package]]
+name = "zstd"
+version = "0.13.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
+dependencies = [
+ "zstd-safe",
+]
+
+[[package]]
+name = "zstd-safe"
+version = "7.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
+dependencies = [
+ "zstd-sys",
+]
+
+[[package]]
+name = "zstd-sys"
+version = "2.0.16+zstd.1.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748"
+dependencies = [
+ "cc",
+ "pkg-config",
 ]
 
 [[package]]

crates/jacquard-api/Cargo.toml

···
 jacquard-derive = { version = "0.5", path = "../jacquard-derive" }
 miette.workspace = true
 serde.workspace = true
+serde_ipld_dagcbor.workspace = true
 thiserror.workspace = true
 
 

crates/jacquard-api/src/com_atproto/label.rs

···
 // Any manual changes will be overwritten on the next regeneration.
 
 pub mod query_labels;
-#[cfg(feature = "streaming")]
 pub mod subscribe_labels;
 
 /// Metadata tag on an atproto resource (eg, repo or record).
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Label<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct LabelValueDefinition<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct LabelValueDefinitionStrings<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    Default,
+    Default
 )]
 #[serde(rename_all = "camelCase")]
 pub struct SelfLabel<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder,
+    bon::Builder
 )]
 #[serde(rename_all = "camelCase")]
 pub struct SelfLabels<'a> {
     #[serde(borrow)]
     pub values: Vec<crate::com_atproto::label::SelfLabel<'a>>,
-}
+}

crates/jacquard-api/src/com_atproto/label/subscribe_labels.rs

···
     Info(Box<crate::com_atproto::label::subscribe_labels::Info<'a>>),
 }
 
+impl<'a> SubscribeLabelsMessage<'a> {
+    /// Decode a framed DAG-CBOR message (header + body).
+    pub fn decode_framed<'de: 'a>(
+        bytes: &'de [u8],
+    ) -> Result<SubscribeLabelsMessage<'a>, jacquard_common::error::DecodeError> {
+        let (header, body) = jacquard_common::xrpc::subscription::parse_event_header(
+            bytes,
+        )?;
+        match header.t.as_str() {
+            "#labels" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Labels(Box::new(variant)))
+            }
+            "#info" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Info(Box::new(variant)))
+            }
+            unknown => {
+                Err(
+                    jacquard_common::error::DecodeError::UnknownEventType(unknown.into()),
+                )
+            }
+        }
+    }
+}
+
 #[jacquard_derive::open_union]
 #[derive(
     serde::Serialize,
···
     const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
     type Message<'de> = SubscribeLabelsMessage<'de>;
     type Error<'de> = SubscribeLabelsError<'de>;
+    fn decode_message<'de>(
+        bytes: &'de [u8],
+    ) -> Result<Self::Message<'de>, jacquard_common::error::DecodeError> {
+        SubscribeLabelsMessage::decode_framed(bytes)
+    }
 }
 
 impl jacquard_common::xrpc::XrpcSubscription for SubscribeLabels {

crates/jacquard-api/src/com_atproto/sync.rs

···
 pub mod list_repos_by_collection;
 pub mod notify_of_update;
 pub mod request_crawl;
-#[cfg(feature = "streaming")]
 pub mod subscribe_repos;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
···
             HostStatus::Other(v) => HostStatus::Other(v.into_static()),
         }
     }
-}
+}

crates/jacquard-api/src/com_atproto/sync/subscribe_repos.rs

···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Account<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Commit<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Identity<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    Default
+    Default,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Info<'a> {
···
     PartialEq,
     Eq,
     bon::Builder,
-    jacquard_derive::IntoStatic
+    jacquard_derive::IntoStatic,
 )]
 #[builder(start_fn = new)]
 #[serde(rename_all = "camelCase")]
···
 
 #[jacquard_derive::open_union]
 #[derive(
-    serde::Serialize,
-    serde::Deserialize,
-    Debug,
-    Clone,
-    PartialEq,
-    Eq,
-    jacquard_derive::IntoStatic
+    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic,
 )]
 #[serde(tag = "$type")]
 #[serde(bound(deserialize = "'de: 'a"))]
···
     Info(Box<crate::com_atproto::sync::subscribe_repos::Info<'a>>),
 }
 
+impl<'a> SubscribeReposMessage<'a> {
+    /// Decode a framed DAG-CBOR message (header + body).
+    pub fn decode_framed<'de: 'a>(
+        bytes: &'de [u8],
+    ) -> Result<SubscribeReposMessage<'a>, jacquard_common::error::DecodeError> {
+        let (header, body) = jacquard_common::xrpc::subscription::parse_event_header(bytes)?;
+        match header.t.as_str() {
+            "#commit" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Commit(Box::new(variant)))
+            }
+            "#sync" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Sync(Box::new(variant)))
+            }
+            "#identity" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Identity(Box::new(variant)))
+            }
+            "#account" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Account(Box::new(variant)))
+            }
+            "#info" => {
+                let variant = serde_ipld_dagcbor::from_slice(body)?;
+                Ok(Self::Info(Box::new(variant)))
+            }
+            unknown => Err(jacquard_common::error::DecodeError::UnknownEventType(
+                unknown.into(),
+            )),
+        }
+    }
+}
+
 #[jacquard_derive::open_union]
 #[derive(
     serde::Serialize,
···
     Eq,
     thiserror::Error,
     miette::Diagnostic,
-    jacquard_derive::IntoStatic
+    jacquard_derive::IntoStatic,
 )]
 #[serde(tag = "error", content = "message")]
 #[serde(bound(deserialize = "'de: 'a"))]
···
 pub struct SubscribeReposStream;
 impl jacquard_common::xrpc::SubscriptionResp for SubscribeReposStream {
     const NSID: &'static str = "com.atproto.sync.subscribeRepos";
-    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
     type Message<'de> = SubscribeReposMessage<'de>;
     type Error<'de> = SubscribeReposError<'de>;
+    fn decode_message<'de>(
+        bytes: &'de [u8],
+    ) -> Result<Self::Message<'de>, jacquard_common::error::DecodeError> {
+        SubscribeReposMessage::decode_framed(bytes)
+    }
 }
 
 impl jacquard_common::xrpc::XrpcSubscription for SubscribeRepos {
     const NSID: &'static str = "com.atproto.sync.subscribeRepos";
-    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
     type Stream = SubscribeReposStream;
 }
 
 pub struct SubscribeReposEndpoint;
 impl jacquard_common::xrpc::SubscriptionEndpoint for SubscribeReposEndpoint {
     const PATH: &'static str = "/xrpc/com.atproto.sync.subscribeRepos";
-    const ENCODING: jacquard_common::xrpc::MessageEncoding = jacquard_common::xrpc::MessageEncoding::DagCbor;
+    const ENCODING: jacquard_common::xrpc::MessageEncoding =
+        jacquard_common::xrpc::MessageEncoding::DagCbor;
     type Params<'de> = SubscribeRepos;
     type Stream = SubscribeReposStream;
 }
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct RepoOp<'a> {
···
     PartialEq,
     Eq,
     jacquard_derive::IntoStatic,
-    bon::Builder
+    bon::Builder,
 )]
 #[serde(rename_all = "camelCase")]
 pub struct Sync<'a> {
···
     pub seq: i64,
     /// Timestamp of when this message was originally broadcast.
     pub time: jacquard_common::types::string::Datetime,
-}
+}

crates/jacquard-api/src/place_stream/live/get_profile_card.rs

···
     PartialEq,
     Eq,
     bon::Builder,
-    jacquard_derive::IntoStatic,
+    jacquard_derive::IntoStatic
 )]
 #[builder(start_fn = new)]
 #[serde(rename_all = "camelCase")]
···
 }
 
 #[derive(
-    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq, jacquard_derive::IntoStatic,
+    serde::Serialize,
+    serde::Deserialize,
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    jacquard_derive::IntoStatic
 )]
 #[serde(rename_all = "camelCase")]
 pub struct GetProfileCardOutput {
···
     Eq,
     thiserror::Error,
     miette::Diagnostic,
-    jacquard_derive::IntoStatic,
+    jacquard_derive::IntoStatic
 )]
 #[serde(tag = "error", content = "message")]
 #[serde(bound(deserialize = "'de: 'a"))]
···
     const METHOD: jacquard_common::xrpc::XrpcMethod = jacquard_common::xrpc::XrpcMethod::Query;
     type Request<'de> = GetProfileCard<'de>;
     type Response = GetProfileCardResponse;
-}
+}

crates/jacquard-common/Cargo.toml

···
 license.workspace = true
 
 [features]
-default = ["service-auth", "reqwest-client", "crypto"]
+default = ["service-auth", "reqwest-client", "crypto", "websocket", "zstd"]
 crypto = []
 crypto-ed25519 = ["crypto", "dep:ed25519-dalek"]
 crypto-k256 = ["crypto", "dep:k256", "k256/ecdsa"]
···
 reqwest-client = ["dep:reqwest"]
 tracing = ["dep:tracing"]
 streaming = ["n0-future", "futures"]
-websocket = ["streaming", "tokio-tungstenite-wasm"]
+websocket = ["streaming", "tokio-tungstenite-wasm", "dep:ciborium"]
+zstd = ["dep:zstd"]
 
 [dependencies]
 trait-variant.workspace = true
···
 # Streaming support (optional)
 n0-future = { workspace = true, optional = true }
 futures = { version = "0.3", optional = true }
-tokio-tungstenite-wasm = { version = "0.4", optional = true }
+tokio-tungstenite-wasm = { version = "0.4", features = ["rustls-tls-native-roots"], optional = true }
+ciborium = {version = "0.2.0", optional = true }
 genawaiter = { version = "0.99.1", features = ["futures03"] }
+zstd = { version = "0.13", optional = true }
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 getrandom = { version = "0.3.4", features = ["wasm_js"] }

crates/jacquard-common/src/error.rs

···
         #[source]
         serde_ipld_dagcbor::DecodeError<std::convert::Infallible>,
     ),
+    #[cfg(feature = "websocket")]
+    #[error("Failed to deserialize cbor header: {0}")]
+    CborHeader(
+        #[from]
+        #[source]
+        ciborium::de::Error<std::io::Error>,
+    ),
+
+    /// Unknown event type in framed message
+    #[cfg(feature = "websocket")]
+    #[error("Unknown event type: {0}")]
+    UnknownEventType(smol_str::SmolStr),
 }
 
 /// HTTP error response (non-200 status codes outside of XRPC error handling)

crates/jacquard-common/src/jetstream.rs

···
+//! Jetstream subscription support
+//!
+//! Jetstream is a simplified JSON-based alternative to the atproto firehose.
+//! Unlike subscribeRepos which uses DAG-CBOR, Jetstream uses JSON encoding.
+
+use crate::types::string::{Datetime, Did, Handle};
+use crate::xrpc::{MessageEncoding, SubscriptionResp, XrpcSubscription};
+use crate::{CowStr, Data, IntoStatic};
+use serde::{Deserialize, Serialize};
+
+/// Parameters for subscribing to Jetstream
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, bon::Builder)]
+#[serde(rename_all = "camelCase")]
+#[builder(start_fn = new)]
+pub struct JetstreamParams<'a> {
+    /// Filter by collection NSIDs (max 100)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    #[builder(into)]
+    pub wanted_collections: Option<Vec<crate::CowStr<'a>>>,
+
+    /// Filter by DIDs (max 10,000)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    #[builder(into)]
+    pub wanted_dids: Option<Vec<crate::CowStr<'a>>>,
+
+    /// Unix microseconds timestamp to start playback
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<i64>,
+
+    /// Maximum payload size in bytes
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_message_size_bytes: Option<u64>,
+
+    /// Enable zstd compression
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub compress: Option<bool>,
+
+    /// Pause stream until first options update
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub require_hello: Option<bool>,
+}
+
+/// Commit operation type
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum CommitOperation {
+    /// Create a new record
+    Create,
+    /// Update an existing record
+    Update,
+    /// Delete a record
+    Delete,
+}
+
+/// Commit event details
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JetstreamCommit<'a> {
+    /// Revision string
+    #[serde(borrow)]
+    pub rev: CowStr<'a>,
+    /// Operation type
+    pub operation: CommitOperation,
+    /// Collection NSID
+    #[serde(borrow)]
+    pub collection: CowStr<'a>,
+    /// Record key
+    #[serde(borrow)]
+    pub rkey: CowStr<'a>,
+    /// Record data (present for create/update)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    pub record: Option<Data<'a>>,
+    /// Content identifier
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    pub cid: Option<CowStr<'a>>,
+}
+
+/// Identity event details
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JetstreamIdentity<'a> {
+    /// DID
+    #[serde(borrow)]
+    pub did: Did<'a>,
+    /// Handle
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    pub handle: Option<Handle<'a>>,
+    /// Sequence number
+    pub seq: i64,
+    /// Timestamp
+    pub time: Datetime,
+}
+
+/// Account event details
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JetstreamAccount<'a> {
+    /// Account active status
+    pub active: bool,
+    /// DID
+    #[serde(borrow)]
+    pub did: Did<'a>,
+    /// Sequence number
+    pub seq: i64,
+    /// Timestamp
+    pub time: Datetime,
+    /// Optional status message
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(borrow)]
+    pub status: Option<CowStr<'a>>,
+}
+
+/// Jetstream event message
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind")]
+#[serde(rename_all = "lowercase")]
+pub enum JetstreamMessage<'a> {
+    /// Commit event
+    Commit {
+        /// DID
+        #[serde(borrow)]
+        did: Did<'a>,
+        /// Unix microseconds timestamp
+        time_us: i64,
+        /// Commit details
+        #[serde(borrow)]
+        commit: JetstreamCommit<'a>,
+    },
+    /// Identity event
+    Identity {
+        /// DID
+        #[serde(borrow)]
+        did: Did<'a>,
+        /// Unix microseconds timestamp
+        time_us: i64,
+        /// Identity details
+        #[serde(borrow)]
+        identity: JetstreamIdentity<'a>,
+    },
+    /// Account event
+    Account {
+        /// DID
+        #[serde(borrow)]
+        did: Did<'a>,
+        /// Unix microseconds timestamp
+        time_us: i64,
+        /// Account details
+        #[serde(borrow)]
+        account: JetstreamAccount<'a>,
+    },
+}
+
+impl IntoStatic for CommitOperation {
+    type Output = CommitOperation;
+
+    fn into_static(self) -> Self::Output {
+        self
+    }
+}
+
+impl IntoStatic for JetstreamCommit<'_> {
+    type Output = JetstreamCommit<'static>;
+
+    fn into_static(self) -> Self::Output {
+        JetstreamCommit {
+            rev: self.rev.into_static(),
+            operation: self.operation,
+            collection: self.collection.into_static(),
+            rkey: self.rkey.into_static(),
+            record: self.record.map(|r| r.into_static()),
+            cid: self.cid.map(|c| c.into_static()),
+        }
+    }
+}
+
+impl IntoStatic for JetstreamIdentity<'_> {
+    type Output = JetstreamIdentity<'static>;
+
+    fn into_static(self) -> Self::Output {
+        JetstreamIdentity {
+            did: self.did.into_static(),
+            handle: self.handle.map(|h| h.into_static()),
+            seq: self.seq,
+            time: self.time,
+        }
+    }
+}
+
+impl IntoStatic for JetstreamAccount<'_> {
+    type Output = JetstreamAccount<'static>;
+
+    fn into_static(self) -> Self::Output {
+        JetstreamAccount {
+            active: self.active,
+            did: self.did.into_static(),
+            seq: self.seq,
+            time: self.time,
+            status: self.status.map(|s| s.into_static()),
+        }
+    }
+}
+
+impl IntoStatic for JetstreamMessage<'_> {
+    type Output = JetstreamMessage<'static>;
+
+    fn into_static(self) -> Self::Output {
+        match self {
+            JetstreamMessage::Commit {
+                did,
+                time_us,
+                commit,
+            } => JetstreamMessage::Commit {
+                did: did.into_static(),
+                time_us,
+                commit: commit.into_static(),
+            },
+            JetstreamMessage::Identity {
+                did,
+                time_us,
+                identity,
+            } => JetstreamMessage::Identity {
+                did: did.into_static(),
+                time_us,
+                identity: identity.into_static(),
+            },
+            JetstreamMessage::Account {
+                did,
+                time_us,
+                account,
+            } => JetstreamMessage::Account {
+                did: did.into_static(),
+                time_us,
+                account: account.into_static(),
+            },
+        }
+    }
+}
+
+/// Stream response type for Jetstream subscriptions
+pub struct JetstreamStream;
+
+impl SubscriptionResp for JetstreamStream {
+    const NSID: &'static str = "jetstream";
+    const ENCODING: MessageEncoding = MessageEncoding::Json;
+
+    /// Typed Jetstream message
+    type Message<'de> = JetstreamMessage<'de>;
+
+    /// Generic error type
+    type Error<'de> = crate::xrpc::GenericError<'de>;
+}
+
+impl<'a> XrpcSubscription for JetstreamParams<'a> {
+    const NSID: &'static str = "jetstream";
+    const ENCODING: MessageEncoding = MessageEncoding::Json;
+    const CUSTOM_PATH: Option<&'static str> = Some("/subscribe");
+    type Stream = JetstreamStream;
+}
+
+impl IntoStatic for JetstreamParams<'_> {
+    type Output = JetstreamParams<'static>;
+
+    fn into_static(self) -> Self::Output {
+        JetstreamParams {
+            wanted_collections: self
+                .wanted_collections
+                .map(|v| v.into_iter().map(|s| s.into_static()).collect()),
+            wanted_dids: self
+                .wanted_dids
+                .map(|v| v.into_iter().map(|s| s.into_static()).collect()),
+            cursor: self.cursor,
+            max_message_size_bytes: self.max_message_size_bytes,
+            compress: self.compress,
+            require_hello: self.require_hello,
+        }
+    }
+}

crates/jacquard-common/src/lib.rs

···
 pub mod websocket;
 
 #[cfg(feature = "websocket")]
+pub mod jetstream;
+
+#[cfg(feature = "websocket")]
 pub use websocket::{
     tungstenite_client::TungsteniteClient, CloseCode, CloseFrame, WebSocketClient,
     WebSocketConnection, WsMessage, WsSink, WsStream, WsText,

crates/jacquard-common/src/types/cid.rs

···
     }
 }
 
-// TODO: take another look at this, see if we can do more borrowed and such
 impl<'de, 'a> Deserialize<'de> for Cid<'a>
 where
     'de: 'a,
···
     where
         D: Deserializer<'de>,
     {
-        struct CidVisitor;
+        if deserializer.is_human_readable() {
+            // JSON: always a string
+            struct StrVisitor;
 
-        impl<'de> Visitor<'de> for CidVisitor {
-            type Value = Cid<'de>;
+            impl<'de> Visitor<'de> for StrVisitor {
+                type Value = Cid<'de>;
 
-            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
-                formatter.write_str("either valid IPLD CID bytes or a str")
-            }
+                fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+                    formatter.write_str("a CID string")
+                }
 
-            fn visit_borrowed_str<E>(self, v: &'de str) -> Result<Self::Value, E>
-            where
-                E: serde::de::Error,
-            {
-                Ok(Cid::str(v))
-            }
+                fn visit_borrowed_str<E>(self, v: &'de str) -> Result<Self::Value, E>
+                where
+                    E: serde::de::Error,
+                {
+                    Ok(Cid::str(v))
+                }
 
-            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
-            where
-                E: serde::de::Error,
-            {
-                Ok(FromStr::from_str(v).unwrap())
+                fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+                where
+                    E: serde::de::Error,
+                {
+                    Ok(FromStr::from_str(v).unwrap())
+                }
             }
 
-            fn visit_bytes<E>(self, v: &[u8]) -> Result<Self::Value, E>
-            where
-                E: serde::de::Error,
-            {
-                let hash = cid::multihash::Multihash::from_bytes(v).map_err(|e| E::custom(e))?;
-                Ok(Cid::ipld(IpldCid::new_v1(ATP_CID_CODEC, hash)))
-            }
+            deserializer.deserialize_str(StrVisitor)
+        } else {
+            // CBOR: use IpldCid's deserializer which handles CBOR tag 42
+            let cid = IpldCid::deserialize(deserializer)?;
+            Ok(Cid::ipld(cid))
         }
-
-        deserializer.deserialize_any(CidVisitor)
     }
 }
 

crates/jacquard-common/src/xrpc/subscription.rs

···
 use std::marker::PhantomData;
 use url::Url;
 
+use crate::error::DecodeError;
 use crate::stream::StreamError;
 use crate::websocket::{WebSocketClient, WebSocketConnection, WsSink, WsStream};
 use crate::{CowStr, Data, IntoStatic, RawData, WsMessage};
···
 
     /// Error union type
     type Error<'de>: Error + Deserialize<'de> + IntoStatic;
+
+    /// Decode a message from bytes.
+    ///
+    /// Default implementation uses simple deserialization via serde.
+    /// Subscriptions that use framed encoding (header + body) can override
+    /// this to do two-stage deserialization.
+    fn decode_message<'de>(bytes: &'de [u8]) -> Result<Self::Message<'de>, DecodeError> {
+        match Self::ENCODING {
+            MessageEncoding::Json => serde_json::from_slice(bytes).map_err(DecodeError::from),
+            MessageEncoding::DagCbor => {
+                serde_ipld_dagcbor::from_slice(bytes).map_err(DecodeError::from)
+            }
+        }
+    }
 }
 
 /// XRPC subscription (WebSocket)
···
     /// Message encoding (JSON or DAG-CBOR)
     const ENCODING: MessageEncoding;
 
+    /// Custom path override (e.g., "/subscribe" for Jetstream).
+    /// If None, defaults to "/xrpc/{NSID}"
+    const CUSTOM_PATH: Option<&'static str> = None;
+
     /// Stream response type (marker struct)
     type Stream: SubscriptionResp;
 
···
     }
 }
 
+#[derive(Debug, serde::Deserialize)]
+pub struct EventHeader {
+    pub op: i64,
+    pub t: smol_str::SmolStr, // type discriminator like "#commit"
+}
+
+pub fn parse_event_header<'a>(bytes: &'a [u8]) -> Result<(EventHeader, &'a [u8]), DecodeError> {
+    let mut cursor = std::io::Cursor::new(bytes);
+    let header: EventHeader = ciborium::de::from_reader(&mut cursor)?;
+    let position = cursor.position() as usize;
+    drop(cursor); // explicit drop before reborrowing bytes
+
+    Ok((header, &bytes[position..]))
+}
+
 /// Decode JSON messages from a WebSocket stream
-fn decode_json_msg<S: SubscriptionResp>(
+pub fn decode_json_msg<S: SubscriptionResp>(
     msg_result: Result<crate::websocket::WsMessage, StreamError>,
 ) -> Option<Result<StreamMessage<'static, S>, StreamError>>
 where
···
 {
     use crate::websocket::WsMessage;
 
-    fn parse_msg<'a, S: SubscriptionResp>(
-        bytes: &'a [u8],
-    ) -> Result<S::Message<'a>, serde_json::Error> {
-        serde_json::from_slice(bytes)
-    }
-
     match msg_result {
         Ok(WsMessage::Text(text)) => Some(
-            parse_msg::<S>(text.as_ref())
+            S::decode_message(text.as_ref())
                 .map(|v| v.into_static())
                 .map_err(StreamError::decode),
         ),
-        Ok(WsMessage::Binary(bytes)) => Some(
-            parse_msg::<S>(&bytes)
-                .map(|v| v.into_static())
-                .map_err(StreamError::decode),
-        ),
+        Ok(WsMessage::Binary(bytes)) => {
+            #[cfg(feature = "zstd")]
+            {
+                // Try to decompress with zstd first (Jetstream uses zstd compression)
+                match decompress_zstd(&bytes) {
+                    Ok(decompressed) => Some(
+                        S::decode_message(&decompressed)
+                            .map(|v| v.into_static())
+                            .map_err(StreamError::decode),
+                    ),
+                    Err(_) => {
+                        // Not zstd-compressed, try direct decode
+                        Some(
+                            S::decode_message(&bytes)
+                                .map(|v| v.into_static())
+                                .map_err(StreamError::decode),
+                        )
+                    }
+                }
+            }
+            #[cfg(not(feature = "zstd"))]
+            {
+                Some(
+                    S::decode_message(&bytes)
+                        .map(|v| v.into_static())
+                        .map_err(StreamError::decode),
+                )
+            }
+        }
         Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())),
         Err(e) => Some(Err(e)),
     }
 }
 
+#[cfg(feature = "zstd")]
+fn decompress_zstd(bytes: &[u8]) -> Result<Vec<u8>, std::io::Error> {
+    use std::sync::OnceLock;
+    use zstd::stream::decode_all;
+
+    static DICTIONARY: OnceLock<Vec<u8>> = OnceLock::new();
+
+    let dict = DICTIONARY.get_or_init(|| {
+        include_bytes!("../../zstd_dictionary").to_vec()
+    });
+
+    decode_all(std::io::Cursor::new(bytes))
+        .or_else(|_| {
+            // Try with dictionary
+            let mut decoder = zstd::Decoder::with_dictionary(std::io::Cursor::new(bytes), dict)?;
+            let mut result = Vec::new();
+            std::io::Read::read_to_end(&mut decoder, &mut result)?;
+            Ok(result)
+        })
+}
+
 /// Decode CBOR messages from a WebSocket stream
-fn decode_cbor_msg<S: SubscriptionResp>(
+pub fn decode_cbor_msg<S: SubscriptionResp>(
     msg_result: Result<crate::websocket::WsMessage, StreamError>,
 ) -> Option<Result<StreamMessage<'static, S>, StreamError>>
 where
···
 {
     use crate::websocket::WsMessage;
 
-    fn parse_cbor<'a, S: SubscriptionResp>(
-        bytes: &'a [u8],
-    ) -> Result<S::Message<'a>, serde_ipld_dagcbor::DecodeError<std::convert::Infallible>> {
-        serde_ipld_dagcbor::from_slice(bytes)
-    }
-
     match msg_result {
         Ok(WsMessage::Binary(bytes)) => Some(
-            parse_cbor::<S>(&bytes)
+            S::decode_message(&bytes)
                 .map(|v| v.into_static())
-                .map_err(|e| StreamError::decode(crate::error::DecodeError::from(e))),
+                .map_err(StreamError::decode),
         ),
         Ok(WsMessage::Text(_)) => Some(Err(StreamError::wrong_message_format(
             "expected binary frame for CBOR, got text",
···
                             .map(|v| v.into_static())
                             .map_err(StreamError::decode),
                     ),
-                    Ok(WsMessage::Binary(bytes)) => Some(
-                        parse_msg(&bytes)
-                            .map(|v| v.into_static())
-                            .map_err(StreamError::decode),
-                    ),
+                    Ok(WsMessage::Binary(bytes)) => {
+                        #[cfg(feature = "zstd")]
+                        {
+                            match decompress_zstd(&bytes) {
+                                Ok(decompressed) => Some(
+                                    parse_msg(&decompressed)
+                                        .map(|v| v.into_static())
+                                        .map_err(StreamError::decode),
+                                ),
+                                Err(_) => Some(
+                                    parse_msg(&bytes)
+                                        .map(|v| v.into_static())
+                                        .map_err(StreamError::decode),
+                                ),
+                            }
+                        }
+                        #[cfg(not(feature = "zstd"))]
+                        {
+                            Some(
+                                parse_msg(&bytes)
+                                    .map(|v| v.into_static())
+                                    .map_err(StreamError::decode),
+                            )
+                        }
+                    }
                     Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())),
                     Err(e) => Some(Err(e)),
                 })
···
                             .map(|v| v.into_static())
                             .map_err(StreamError::decode),
                     ),
-                    Ok(WsMessage::Binary(bytes)) => Some(
-                        parse_msg(&bytes)
-                            .map(|v| v.into_static())
-                            .map_err(StreamError::decode),
-                    ),
+                    Ok(WsMessage::Binary(bytes)) => {
+                        #[cfg(feature = "zstd")]
+                        {
+                            match decompress_zstd(&bytes) {
+                                Ok(decompressed) => Some(
+                                    parse_msg(&decompressed)
+                                        .map(|v| v.into_static())
+                                        .map_err(StreamError::decode),
+                                ),
+                                Err(_) => Some(
+                                    parse_msg(&bytes)
+                                        .map(|v| v.into_static())
+                                        .map_err(StreamError::decode),
+                                ),
+                            }
+                        }
+                        #[cfg(not(feature = "zstd"))]
+                        {
+                            Some(
+                                parse_msg(&bytes)
+                                    .map(|v| v.into_static())
+                                    .map_err(StreamError::decode),
+                            )
+                        }
+                    }
                     Ok(WsMessage::Close(_)) => Some(Err(StreamError::closed())),
                     Err(e) => Some(Err(e)),
                 })
···
         Sub: XrpcSubscription,
     {
         let mut url = self.base.clone();
+
+        // Use custom path if provided, otherwise construct from NSID
         let mut path = url.path().trim_end_matches('/').to_owned();
-        path.push_str("/xrpc/");
-        path.push_str(Sub::NSID);
+        if let Some(custom_path) = Sub::CUSTOM_PATH {
+            path.push_str(custom_path);
+        } else {
+            path.push_str("/xrpc/");
+            path.push_str(Sub::NSID);
+        }
         url.set_path(&path);
 
         let query_params = params.query_params();

crates/jacquard-lexicon/src/codegen/xrpc.rs

···
                 let enum_ident = syn::Ident::new(&enum_name, proc_macro2::Span::call_site());
 
                 let mut variants = Vec::new();
+                let mut decode_arms = Vec::new();
+
                 for ref_str in &union.refs {
                     let ref_str_s = ref_str.as_ref();
 
···
                         #[serde(rename = #ref_str_s)]
                         #variant_ident(Box<#type_path>)
                     });
+
+                    // Generate decode arm for framed decoding
+                    decode_arms.push(quote! {
+                        #ref_str_s => {
+                            let variant = serde_ipld_dagcbor::from_slice(body)?;
+                            Ok(Self::#variant_ident(Box::new(variant)))
+                        }
+                    });
                 }
 
                 let doc = self.generate_doc_comment(union.description.as_ref());
 
+                // Generate decode_framed method for DAG-CBOR subscriptions
+                let decode_framed_impl = quote! {
+                    impl<'a> #enum_ident<'a> {
+                        /// Decode a framed DAG-CBOR message (header + body).
+                        pub fn decode_framed<'de: 'a>(bytes: &'de [u8]) -> Result<#enum_ident<'a>, jacquard_common::error::DecodeError> {
+                            let (header, body) = jacquard_common::xrpc::subscription::parse_event_header(bytes)?;
+                            match header.t.as_str() {
+                                #(#decode_arms)*
+                                unknown => Err(jacquard_common::error::DecodeError::UnknownEventType(
+                                    unknown.into()
+                                )),
+                            }
+                        }
+                    }
+                };
+
                 Ok(quote! {
                     #doc
                     #[jacquard_derive::open_union]
···
                     pub enum #enum_ident<'a> {
                         #(#variants,)*
                     }
+
+                    #decode_framed_impl
                 })
             }
             LexXrpcSubscriptionMessageSchema::Object(obj) => {
···
 
         // Determine encoding from nsid convention
         // ATProto subscriptions use DAG-CBOR, community ones might use JSON
-        let encoding = if nsid.starts_with("com.atproto") {
+        let is_dag_cbor = nsid.starts_with("com.atproto");
+        let encoding = if is_dag_cbor {
             quote! { jacquard_common::xrpc::MessageEncoding::DagCbor }
         } else {
             quote! { jacquard_common::xrpc::MessageEncoding::Json }
         };
 
         // Generate SubscriptionResp impl
+        // For DAG-CBOR subscriptions, override decode_message to use framed decoding
+        let decode_message_override = if is_dag_cbor && has_message {
+            let msg_ident = syn::Ident::new(
+                &format!("{}Message", type_base),
+                proc_macro2::Span::call_site(),
+            );
+            quote! {
+                fn decode_message<'de>(bytes: &'de [u8]) -> Result<Self::Message<'de>, jacquard_common::error::DecodeError> {
+                    #msg_ident::decode_framed(bytes)
+                }
+            }
+        } else {
+            quote! {}
+        };
+
         let stream_resp_impl = quote! {
             #[doc = "Stream response type for "]
             #[doc = #nsid]
···
 
                 type Message<'de> = #message_type;
                 type Error<'de> = #error_type;
+
+                #decode_message_override
             }
         };
 

crates/jacquard-oauth/src/client.rs

···
         Sub: XrpcSubscription + Send + Sync,
     {
         use jacquard_common::xrpc::SubscriptionExt;
-        let base = self.base_uri();
+        let base = self.base_uri().await;
         self.subscription(base)
             .with_options(opts)
             .subscribe(params)

crates/jacquard/Cargo.toml

···
 path = "../../examples/app_password_create_post.rs"
 required-features = ["api_bluesky"]
 
+[[example]]
+name = "subscribe_repos"
+path = "../../examples/subscribe_repos.rs"
+required-features = ["api_bluesky", "streaming"]
+
+[[example]]
+name = "subscribe_jetstream"
+path = "../../examples/subscribe_jetstream.rs"
+required-features = ["streaming"]
+
 
 [dependencies]
 jacquard-api = { version = "0.5", path = "../jacquard-api" }
···
 [dev-dependencies]
 clap.workspace = true
 miette = { workspace = true, features = ["fancy"] }
+tokio = { workspace = true, features = ["signal"] }
 viuer = { version =  "0.9", features = ["print-file", "sixel"] }
 tiff = { version = "0.6.0-alpha" }
 image = { version = "0.25" }

crates/jacquard/src/client/credential_session.rs

···
     T: Send + Sync + 'static,
     W: WebSocketClient + Send + Sync,
 {
-    fn base_uri(&self) -> Url {
-        #[cfg(not(target_arch = "wasm32"))]
-        if tokio::runtime::Handle::try_current().is_ok() {
-            tokio::task::block_in_place(|| {
-                self.endpoint.blocking_read().clone().unwrap_or(
-                    Url::parse("https://public.bsky.app")
-                        .expect("public appview should be valid url"),
-                )
-            })
-        } else {
-            self.endpoint.blocking_read().clone().unwrap_or(
-                Url::parse("https://public.bsky.app").expect("public appview should be valid url"),
-            )
-        }
-
-        #[cfg(target_arch = "wasm32")]
-        {
-            self.endpoint.blocking_read().clone().unwrap_or(
-                Url::parse("https://public.bsky.app").expect("public appview should be valid url"),
-            )
-        }
+    async fn base_uri(&self) -> Url {
+        self.endpoint.read().await.clone().unwrap_or(
+            Url::parse("https://public.bsky.app").expect("public appview should be valid url"),
+        )
     }
 
     async fn subscription_opts(&self) -> jacquard_common::xrpc::SubscriptionOptions<'_> {
···
         Sub: XrpcSubscription + Send + Sync,
     {
         use jacquard_common::xrpc::SubscriptionExt;
-        let base = self.base_uri();
+        let base = self.base_uri().await;
         self.subscription(base)
             .with_options(opts)
             .subscribe(params)

examples/subscribe_jetstream.rs

···
+//! Example: Subscribe to Jetstream firehose
+//!
+//! Jetstream is a JSON-based alternative to the standard DAG-CBOR firehose.
+//! It streams all public network updates in a simplified format.
+//!
+//! Usage:
+//!   cargo run --example subscribe_jetstream
+//!   cargo run --example subscribe_jetstream -- jetstream2.us-west.bsky.network
+
+use clap::Parser;
+use jacquard_common::jetstream::{CommitOperation, JetstreamMessage, JetstreamParams};
+use jacquard_common::xrpc::{SubscriptionClient, TungsteniteSubscriptionClient};
+use miette::IntoDiagnostic;
+use n0_future::StreamExt;
+use url::Url;
+
+#[derive(Parser, Debug)]
+#[command(author, version, about = "Subscribe to Jetstream firehose")]
+struct Args {
+    /// Jetstream URL (e.g., jetstream1.us-east.fire.hose.cam)
+    #[arg(default_value = "jetstream1.us-east.fire.hose.cam")]
+    jetstream_url: String,
+}
+
+fn normalize_url(input: &str) -> Result<Url, url::ParseError> {
+    // Strip any existing scheme
+    let without_scheme = input
+        .trim_start_matches("https://")
+        .trim_start_matches("http://")
+        .trim_start_matches("wss://")
+        .trim_start_matches("ws://");
+
+    // Prepend wss:// and parse
+    Url::parse(&format!("wss://{}", without_scheme))
+}
+
+fn print_message(msg: &JetstreamMessage) {
+    match msg {
+        JetstreamMessage::Commit {
+            did,
+            time_us,
+            commit,
+        } => {
+            let op = match commit.operation {
+                CommitOperation::Create => "create",
+                CommitOperation::Update => "update",
+                CommitOperation::Delete => "delete",
+            };
+            println!(
+                "Commit | did={} time_us={} op={} collection={} rkey={} cid={:?}",
+                did, time_us, op, commit.collection, commit.rkey, commit.cid
+            );
+        }
+        JetstreamMessage::Identity {
+            did,
+            time_us,
+            identity,
+        } => {
+            println!(
+                "Identity | did={} time_us={} handle={:?} seq={} time={}",
+                did, time_us, identity.handle, identity.seq, identity.time
+            );
+        }
+        JetstreamMessage::Account {
+            did,
+            time_us,
+            account,
+        } => {
+            println!(
+                "Account | did={} time_us={} active={} seq={} time={} status={:?}",
+                did, time_us, account.active, account.seq, account.time, account.status
+            );
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> miette::Result<()> {
+    let args = Args::parse();
+
+    let base_url = normalize_url(&args.jetstream_url).into_diagnostic()?;
+    println!("Connecting to {}", base_url);
+
+    // Create subscription client
+    let client = TungsteniteSubscriptionClient::from_base_uri(base_url);
+
+    // Subscribe with no filters (firehose mode)
+    let mut params_builder = JetstreamParams::new();
+
+    // Enable compression if zstd feature is available
+    #[cfg(feature = "zstd")]
+    {
+        params_builder = params_builder.compress(true);
+    }
+
+    let params = params_builder.build();
+    let stream = client.subscribe(&params).await.into_diagnostic()?;
+
+    println!("Connected! Streaming messages (Ctrl-C to stop)...\n");
+
+    // Set up Ctrl-C handler
+    let (tx, mut rx) = tokio::sync::oneshot::channel();
+    tokio::spawn(async move {
+        tokio::signal::ctrl_c().await.ok();
+        let _ = tx.send(());
+    });
+
+    // Convert to typed message stream
+    let (_sink, mut messages) = stream.into_stream();
+
+    let mut count = 0u64;
+
+    loop {
+        tokio::select! {
+            Some(result) = messages.next() => {
+                match result {
+                    Ok(msg) => {
+                        count += 1;
+                        print_message(&msg);
+                    }
+                    Err(e) => eprintln!("Error: {}", e),
+                }
+            }
+            _ = &mut rx => {
+                println!("\nReceived {} messages", count);
+                println!("Shutting down...");
+                break;
+            }
+        }
+    }
+
+    Ok(())
+}

examples/subscribe_repos.rs

···
+//! Example: Subscribe to a PDS's subscribeRepos endpoint
+//!
+//! This demonstrates consuming the repo event stream directly from a PDS,
+//! which is what a Relay does to ingest updates from PDSes.
+//!
+//! Usage:
+//!   cargo run --example subscribe_repos -- atproto.systems
+
+use clap::Parser;
+use jacquard::api::com_atproto::sync::subscribe_repos::{SubscribeRepos, SubscribeReposMessage};
+use jacquard_common::xrpc::{SubscriptionClient, TungsteniteSubscriptionClient};
+use miette::IntoDiagnostic;
+use n0_future::StreamExt;
+use url::Url;
+
+#[derive(Parser, Debug)]
+#[command(author, version, about = "Subscribe to a PDS's subscribeRepos endpoint")]
+struct Args {
+    /// PDS URL (e.g., atproto.systems or https://atproto.systems)
+    pds_url: String,
+
+    /// Starting cursor position
+    #[arg(short, long)]
+    cursor: Option<i64>,
+}
+
+fn normalize_url(input: &str) -> Result<Url, url::ParseError> {
+    // Strip any existing scheme
+    let without_scheme = input
+        .trim_start_matches("https://")
+        .trim_start_matches("http://")
+        .trim_start_matches("wss://")
+        .trim_start_matches("ws://");
+
+    // Prepend wss://
+    Url::parse(&format!("wss://{}", without_scheme))
+}
+
+fn print_message(msg: &SubscribeReposMessage) {
+    match msg {
+        SubscribeReposMessage::Commit(commit) => {
+            println!(
+                "Commit | repo={} seq={} time={} rev={} commit={} ops={} prev={}",
+                commit.repo,
+                commit.seq,
+                commit.time,
+                commit.rev,
+                commit.commit,
+                commit.ops.len(),
+                commit.since,
+            );
+        }
+        SubscribeReposMessage::Identity(identity) => {
+            println!(
+                "Identity | did={} seq={} time={} handle={:?}",
+                identity.did, identity.seq, identity.time, identity.handle
+            );
+        }
+        SubscribeReposMessage::Account(account) => {
+            println!(
+                "Account | did={} seq={} time={} active={} status={:?}",
+                account.did, account.seq, account.time, account.active, account.status
+            );
+        }
+        SubscribeReposMessage::Sync(sync) => {
+            println!(
+                "Sync | did={} seq={} time={} rev={} blocks={}b",
+                sync.did,
+                sync.seq,
+                sync.time,
+                sync.rev,
+                sync.blocks.len()
+            );
+        }
+        SubscribeReposMessage::Info(info) => {
+            println!("Info | name={} message={:?}", info.name, info.message);
+        }
+        SubscribeReposMessage::Unknown(data) => {
+            println!("Unknown message: {:?}", data);
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> miette::Result<()> {
+    let args = Args::parse();
+
+    let base_url = normalize_url(&args.pds_url).into_diagnostic()?;
+    println!("Connecting to {}", base_url);
+
+    // Create subscription client
+    let client = TungsteniteSubscriptionClient::from_base_uri(base_url);
+
+    // Subscribe with optional cursor
+    let params = if let Some(cursor) = args.cursor {
+        SubscribeRepos::new().cursor(cursor).build()
+    } else {
+        SubscribeRepos::new().build()
+    };
+    let stream = client.subscribe(&params).await.into_diagnostic()?;
+
+    println!("Connected! Streaming messages (Ctrl-C to stop)...\n");
+
+    // Set up Ctrl-C handler
+    let (tx, mut rx) = tokio::sync::oneshot::channel();
+    tokio::spawn(async move {
+        tokio::signal::ctrl_c().await.ok();
+        let _ = tx.send(());
+    });
+
+    // Convert to typed message stream
+    let (_sink, mut messages) = stream.into_stream();
+
+    loop {
+        tokio::select! {
+            Some(result) = messages.next() => {
+                match result {
+                    Ok(msg) => print_message(&msg),
+                    Err(e) => eprintln!("Error: {}", e),
+                }
+            }
+            _ = &mut rx => {
+                println!("\nShutting down...");
+                break;
+            }
+        }
+    }
+
+    Ok(())
+}

justfile

···
 example NAME *ARGS:
     #!/usr/bin/env bash
     if [ -f "examples/{{NAME}}.rs" ]; then
-        cargo run -p jacquard --features=api_bluesky --example {{NAME}} -- {{ARGS}}
+        cargo run -p jacquard --features=api_bluesky,streaming --example {{NAME}} -- {{ARGS}}
     elif cargo metadata --format-version=1 --no-deps | \
          jq -e '.packages[] | select(.name == "jacquard-axum") | .targets[] | select(.kind[] == "example" and .name == "{{NAME}}")' > /dev/null; then
         cargo run -p jacquard-axum --example {{NAME}}  -- {{ARGS}}