Repository: sugyan/atrium (language: rust)

Issue:
Subscribe repos example?
Regarding atrium-api/src/com/atproto/sync/subscribe_repos.rs

It would be great to see some sort of example on how to use this. From brief reading of the code, it seems to not be implemented yet, as I cannot find anything regarding reading streams, events or anything like that. Is that correct?


Discussion / hints:
Unfortunately, the `subscription` feature is not yet implemented.
I would like to use it too and will do my best to implement it. 😵‍💫 
@sugyan If there are notes on how to implement it (and you are accepting contributors) I would be interested in helping.  Thanks for all of your hard work on the library - really works well for being so early.
If it is helpful, I did some prototyping work with decoding the firehose here:  

https://github.com/timfpark/firehose-prototype/blob/main/src/main.rs

I've made it through the Header, Commit, and RepoOp structures after evaluating a bunch of CBOR deserializers and landing on `ciborium` in conjunction with `serde`. The principal challenge from here is decoding the CAR file in the commit. 

There is a CAR decoder here https://github.com/dapplion/rs-car/tree/master.

Hopefully this is helpful.
Oooh very cool prototype implementation, thank you!
I will try to implement something like this with your codes as a reference!
Bisky has methods for reading records from the firehose that are working, see here: 
https://github.com/jesopo/bisky/tree/jess/firehose

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 77e3a8ab..257c1f4a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,4 +16,6 @@ members = [
     "atrium-codegen",
     "atrium-lex",
     "atrium-xrpc",
+    "atrium-xrpc-server",
+    "examples/firehose",
 ]
diff --git a/atrium-api/Cargo.toml b/atrium-api/Cargo.toml
index e6f837d9..723f7570 100644
--- a/atrium-api/Cargo.toml
+++ b/atrium-api/Cargo.toml
@@ -13,8 +13,10 @@ keywords = ["atproto", "bluesky"]
 
 [dependencies]
 async-trait = "0.1.68"
+cid = { version = "0.10.1", features = ["serde-codec"] }
 http = "0.2.9"
 serde = { version = "1.0.160", features = ["derive"] }
+serde_bytes = "0.11.9"
 serde_json = "1.0.96"
 serde_urlencoded = "0.7.1"
 
diff --git a/atrium-api/src/app/bsky/actor/defs.rs b/atrium-api/src/app/bsky/actor/defs.rs
index 6044216f..7a79a055 100644
--- a/atrium-api/src/app/bsky/actor/defs.rs
+++ b/atrium-api/src/app/bsky/actor/defs.rs
@@ -15,7 +15,7 @@ pub struct ContentLabelPref {
     pub visibility: String,
 }
 #[doc = "`app.bsky.actor.defs#preferences`"]
-pub type Preferences = Vec<Box<PreferencesItem>>;
+pub type Preferences = Vec<PreferencesItem>;
 #[doc = "`app.bsky.actor.defs#profileView`"]
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
@@ -98,7 +98,7 @@ pub struct ViewerState {
 #[serde(tag = "$type")]
 pub enum PreferencesItem {
     #[serde(rename = "app.bsky.actor.defs#adultContentPref")]
-    AdultContentPref(AdultContentPref),
+    AdultContentPref(Box<AdultContentPref>),
     #[serde(rename = "app.bsky.actor.defs#contentLabelPref")]
-    ContentLabelPref(ContentLabelPref),
+    ContentLabelPref(Box<ContentLabelPref>),
 }
diff --git a/atrium-api/src/app/bsky/embed/record.rs b/atrium-api/src/app/bsky/embed/record.rs
index 998a1440..ac35c249 100644
--- a/atrium-api/src/app/bsky/embed/record.rs
+++ b/atrium-api/src/app/bsky/embed/record.rs
@@ -11,7 +11,7 @@ pub struct Main {
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct View {
-    pub record: Box<ViewRecordEnum>,
+    pub record: ViewRecordEnum,
 }
 #[doc = "`app.bsky.embed.record#viewBlocked`"]
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -32,7 +32,7 @@ pub struct ViewRecord {
     pub author: crate::app::bsky::actor::defs::ProfileViewBasic,
     pub cid: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub embeds: Option<Vec<Box<ViewRecordEmbedsItem>>>,
+    pub embeds: Option<Vec<ViewRecordEmbedsItem>>,
     pub indexed_at: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub labels: Option<Vec<crate::com::atproto::label::defs::Label>>,
@@ -43,21 +43,21 @@ pub struct ViewRecord {
 #[serde(tag = "$type")]
 pub enum ViewRecordEmbedsItem {
     #[serde(rename = "app.bsky.embed.images#view")]
-    AppBskyEmbedImagesView(crate::app::bsky::embed::images::View),
+    AppBskyEmbedImagesView(Box<crate::app::bsky::embed::images::View>),
     #[serde(rename = "app.bsky.embed.external#view")]
-    AppBskyEmbedExternalView(crate::app::bsky::embed::external::View),
+    AppBskyEmbedExternalView(Box<crate::app::bsky::embed::external::View>),
     #[serde(rename = "app.bsky.embed.record#view")]
-    AppBskyEmbedRecordView(crate::app::bsky::embed::record::View),
+    AppBskyEmbedRecordView(Box<crate::app::bsky::embed::record::View>),
     #[serde(rename = "app.bsky.embed.recordWithMedia#view")]
-    AppBskyEmbedRecordWithMediaView(crate::app::bsky::embed::record_with_media::View),
+    AppBskyEmbedRecordWithMediaView(Box<crate::app::bsky::embed::record_with_media::View>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ViewRecordEnum {
     #[serde(rename = "app.bsky.embed.record#viewRecord")]
-    ViewRecord(ViewRecord),
+    ViewRecord(Box<ViewRecord>),
     #[serde(rename = "app.bsky.embed.record#viewNotFound")]
-    ViewNotFound(ViewNotFound),
+    ViewNotFound(Box<ViewNotFound>),
     #[serde(rename = "app.bsky.embed.record#viewBlocked")]
-    ViewBlocked(ViewBlocked),
+    ViewBlocked(Box<ViewBlocked>),
 }
diff --git a/atrium-api/src/app/bsky/embed/record_with_media.rs b/atrium-api/src/app/bsky/embed/record_with_media.rs
index 6a1cb5af..28576fb0 100644
--- a/atrium-api/src/app/bsky/embed/record_with_media.rs
+++ b/atrium-api/src/app/bsky/embed/record_with_media.rs
@@ -5,29 +5,29 @@
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct Main {
-    pub media: Box<MainMediaEnum>,
+    pub media: MainMediaEnum,
     pub record: crate::app::bsky::embed::record::Main,
 }
 #[doc = "`app.bsky.embed.recordWithMedia#view`"]
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct View {
-    pub media: Box<ViewMediaEnum>,
+    pub media: ViewMediaEnum,
     pub record: crate::app::bsky::embed::record::View,
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum MainMediaEnum {
     #[serde(rename = "app.bsky.embed.images")]
-    AppBskyEmbedImagesMain(crate::app::bsky::embed::images::Main),
+    AppBskyEmbedImagesMain(Box<crate::app::bsky::embed::images::Main>),
     #[serde(rename = "app.bsky.embed.external")]
-    AppBskyEmbedExternalMain(crate::app::bsky::embed::external::Main),
+    AppBskyEmbedExternalMain(Box<crate::app::bsky::embed::external::Main>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ViewMediaEnum {
     #[serde(rename = "app.bsky.embed.images#view")]
-    AppBskyEmbedImagesView(crate::app::bsky::embed::images::View),
+    AppBskyEmbedImagesView(Box<crate::app::bsky::embed::images::View>),
     #[serde(rename = "app.bsky.embed.external#view")]
-    AppBskyEmbedExternalView(crate::app::bsky::embed::external::View),
+    AppBskyEmbedExternalView(Box<crate::app::bsky::embed::external::View>),
 }
diff --git a/atrium-api/src/app/bsky/feed/defs.rs b/atrium-api/src/app/bsky/feed/defs.rs
index 2aa60262..042de9cf 100644
--- a/atrium-api/src/app/bsky/feed/defs.rs
+++ b/atrium-api/src/app/bsky/feed/defs.rs
@@ -13,7 +13,7 @@ pub struct BlockedPost {
 pub struct FeedViewPost {
     pub post: crate::app::bsky::feed::defs::PostView,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub reason: Option<Box<FeedViewPostReasonEnum>>,
+    pub reason: Option<FeedViewPostReasonEnum>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reply: Option<ReplyRef>,
 }
@@ -31,7 +31,7 @@ pub struct PostView {
     pub author: crate::app::bsky::actor::defs::ProfileViewBasic,
     pub cid: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub embed: Option<Box<PostViewEmbedEnum>>,
+    pub embed: Option<PostViewEmbedEnum>,
     pub indexed_at: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub labels: Option<Vec<crate::com::atproto::label::defs::Label>>,
@@ -65,10 +65,10 @@ pub struct ReplyRef {
 #[serde(rename_all = "camelCase")]
 pub struct ThreadViewPost {
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub parent: Option<Box<ThreadViewPostParentEnum>>,
+    pub parent: Option<ThreadViewPostParentEnum>,
     pub post: PostView,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub replies: Option<Vec<Box<ThreadViewPostRepliesItem>>>,
+    pub replies: Option<Vec<ThreadViewPostRepliesItem>>,
 }
 #[doc = "`app.bsky.feed.defs#viewerState`"]
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -83,37 +83,37 @@ pub struct ViewerState {
 #[serde(tag = "$type")]
 pub enum FeedViewPostReasonEnum {
     #[serde(rename = "app.bsky.feed.defs#reasonRepost")]
-    ReasonRepost(ReasonRepost),
+    ReasonRepost(Box<ReasonRepost>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum PostViewEmbedEnum {
     #[serde(rename = "app.bsky.embed.images#view")]
-    AppBskyEmbedImagesView(crate::app::bsky::embed::images::View),
+    AppBskyEmbedImagesView(Box<crate::app::bsky::embed::images::View>),
     #[serde(rename = "app.bsky.embed.external#view")]
-    AppBskyEmbedExternalView(crate::app::bsky::embed::external::View),
+    AppBskyEmbedExternalView(Box<crate::app::bsky::embed::external::View>),
     #[serde(rename = "app.bsky.embed.record#view")]
-    AppBskyEmbedRecordView(crate::app::bsky::embed::record::View),
+    AppBskyEmbedRecordView(Box<crate::app::bsky::embed::record::View>),
     #[serde(rename = "app.bsky.embed.recordWithMedia#view")]
-    AppBskyEmbedRecordWithMediaView(crate::app::bsky::embed::record_with_media::View),
+    AppBskyEmbedRecordWithMediaView(Box<crate::app::bsky::embed::record_with_media::View>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ThreadViewPostParentEnum {
     #[serde(rename = "app.bsky.feed.defs#threadViewPost")]
-    ThreadViewPost(ThreadViewPost),
+    ThreadViewPost(Box<ThreadViewPost>),
     #[serde(rename = "app.bsky.feed.defs#notFoundPost")]
-    NotFoundPost(NotFoundPost),
+    NotFoundPost(Box<NotFoundPost>),
     #[serde(rename = "app.bsky.feed.defs#blockedPost")]
-    BlockedPost(BlockedPost),
+    BlockedPost(Box<BlockedPost>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ThreadViewPostRepliesItem {
     #[serde(rename = "app.bsky.feed.defs#threadViewPost")]
-    ThreadViewPost(ThreadViewPost),
+    ThreadViewPost(Box<ThreadViewPost>),
     #[serde(rename = "app.bsky.feed.defs#notFoundPost")]
-    NotFoundPost(NotFoundPost),
+    NotFoundPost(Box<NotFoundPost>),
     #[serde(rename = "app.bsky.feed.defs#blockedPost")]
-    BlockedPost(BlockedPost),
+    BlockedPost(Box<BlockedPost>),
 }
diff --git a/atrium-api/src/app/bsky/feed/get_post_thread.rs b/atrium-api/src/app/bsky/feed/get_post_thread.rs
index 548cd134..04dce36c 100644
--- a/atrium-api/src/app/bsky/feed/get_post_thread.rs
+++ b/atrium-api/src/app/bsky/feed/get_post_thread.rs
@@ -29,7 +29,7 @@ pub struct Parameters {
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct Output {
-    pub thread: Box<OutputThreadEnum>,
+    pub thread: OutputThreadEnum,
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "error", content = "message")]
@@ -40,9 +40,9 @@ pub enum Error {
 #[serde(tag = "$type")]
 pub enum OutputThreadEnum {
     #[serde(rename = "app.bsky.feed.defs#threadViewPost")]
-    AppBskyFeedDefsThreadViewPost(crate::app::bsky::feed::defs::ThreadViewPost),
+    AppBskyFeedDefsThreadViewPost(Box<crate::app::bsky::feed::defs::ThreadViewPost>),
     #[serde(rename = "app.bsky.feed.defs#notFoundPost")]
-    AppBskyFeedDefsNotFoundPost(crate::app::bsky::feed::defs::NotFoundPost),
+    AppBskyFeedDefsNotFoundPost(Box<crate::app::bsky::feed::defs::NotFoundPost>),
     #[serde(rename = "app.bsky.feed.defs#blockedPost")]
-    AppBskyFeedDefsBlockedPost(crate::app::bsky::feed::defs::BlockedPost),
+    AppBskyFeedDefsBlockedPost(Box<crate::app::bsky::feed::defs::BlockedPost>),
 }
diff --git a/atrium-api/src/app/bsky/feed/post.rs b/atrium-api/src/app/bsky/feed/post.rs
index 1d364b76..2c4d36ef 100644
--- a/atrium-api/src/app/bsky/feed/post.rs
+++ b/atrium-api/src/app/bsky/feed/post.rs
@@ -6,7 +6,7 @@
 pub struct Record {
     pub created_at: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub embed: Option<Box<RecordEmbedEnum>>,
+    pub embed: Option<RecordEmbedEnum>,
     #[doc = "Deprecated: replaced by app.bsky.richtext.facet."]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub entities: Option<Vec<Entity>>,
@@ -45,11 +45,11 @@ pub struct TextSlice {
 #[serde(tag = "$type")]
 pub enum RecordEmbedEnum {
     #[serde(rename = "app.bsky.embed.images")]
-    AppBskyEmbedImagesMain(crate::app::bsky::embed::images::Main),
+    AppBskyEmbedImagesMain(Box<crate::app::bsky::embed::images::Main>),
     #[serde(rename = "app.bsky.embed.external")]
-    AppBskyEmbedExternalMain(crate::app::bsky::embed::external::Main),
+    AppBskyEmbedExternalMain(Box<crate::app::bsky::embed::external::Main>),
     #[serde(rename = "app.bsky.embed.record")]
-    AppBskyEmbedRecordMain(crate::app::bsky::embed::record::Main),
+    AppBskyEmbedRecordMain(Box<crate::app::bsky::embed::record::Main>),
     #[serde(rename = "app.bsky.embed.recordWithMedia")]
-    AppBskyEmbedRecordWithMediaMain(crate::app::bsky::embed::record_with_media::Main),
+    AppBskyEmbedRecordWithMediaMain(Box<crate::app::bsky::embed::record_with_media::Main>),
 }
diff --git a/atrium-api/src/app/bsky/richtext/facet.rs b/atrium-api/src/app/bsky/richtext/facet.rs
index ce0bd46b..3631a968 100644
--- a/atrium-api/src/app/bsky/richtext/facet.rs
+++ b/atrium-api/src/app/bsky/richtext/facet.rs
@@ -4,7 +4,7 @@
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct Main {
-    pub features: Vec<Box<MainFeaturesItem>>,
+    pub features: Vec<MainFeaturesItem>,
     pub index: ByteSlice,
 }
 #[doc = "`app.bsky.richtext.facet#byteSlice`"]
@@ -33,7 +33,7 @@ pub struct Mention {
 #[serde(tag = "$type")]
 pub enum MainFeaturesItem {
     #[serde(rename = "app.bsky.richtext.facet#mention")]
-    Mention(Mention),
+    Mention(Box<Mention>),
     #[serde(rename = "app.bsky.richtext.facet#link")]
-    Link(Link),
+    Link(Box<Link>),
 }
diff --git a/atrium-api/src/com/atproto/admin/defs.rs b/atrium-api/src/com/atproto/admin/defs.rs
index ced16af8..348e9aca 100644
--- a/atrium-api/src/com/atproto/admin/defs.rs
+++ b/atrium-api/src/com/atproto/admin/defs.rs
@@ -29,7 +29,7 @@ pub struct ActionView {
     pub resolved_report_ids: Vec<i32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reversal: Option<ActionReversal>,
-    pub subject: Box<ActionViewSubjectEnum>,
+    pub subject: ActionViewSubjectEnum,
     pub subject_blob_cids: Vec<String>,
 }
 #[doc = "`com.atproto.admin.defs#actionViewCurrent`"]
@@ -55,7 +55,7 @@ pub struct ActionViewDetail {
     pub resolved_reports: Vec<ReportView>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reversal: Option<ActionReversal>,
-    pub subject: Box<ActionViewDetailSubjectEnum>,
+    pub subject: ActionViewDetailSubjectEnum,
     pub subject_blobs: Vec<BlobView>,
 }
 #[doc = "`com.atproto.admin.defs#blobView`"]
@@ -65,7 +65,7 @@ pub struct BlobView {
     pub cid: String,
     pub created_at: String,
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub details: Option<Box<BlobViewDetailsEnum>>,
+    pub details: Option<BlobViewDetailsEnum>,
     pub mime_type: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub moderation: Option<Moderation>,
@@ -179,7 +179,7 @@ pub struct ReportView {
     pub reason_type: crate::com::atproto::moderation::defs::ReasonType,
     pub reported_by: String,
     pub resolved_by_action_ids: Vec<i32>,
-    pub subject: Box<ReportViewSubjectEnum>,
+    pub subject: ReportViewSubjectEnum,
 }
 #[doc = "`com.atproto.admin.defs#reportViewDetail`"]
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -192,7 +192,7 @@ pub struct ReportViewDetail {
     pub reason_type: crate::com::atproto::moderation::defs::ReasonType,
     pub reported_by: String,
     pub resolved_by_actions: Vec<crate::com::atproto::admin::defs::ActionView>,
-    pub subject: Box<ReportViewDetailSubjectEnum>,
+    pub subject: ReportViewDetailSubjectEnum,
 }
 #[doc = "`com.atproto.admin.defs#takedown`"]
 #[doc = "Moderation action type: Takedown. Indicates that content should not be served by the PDS."]
@@ -209,39 +209,39 @@ pub struct VideoDetails {
 #[serde(tag = "$type")]
 pub enum ActionViewDetailSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoView")]
-    RepoView(RepoView),
+    RepoView(Box<RepoView>),
     #[serde(rename = "com.atproto.admin.defs#recordView")]
-    RecordView(RecordView),
+    RecordView(Box<RecordView>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ActionViewSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoRef")]
-    RepoRef(RepoRef),
+    RepoRef(Box<RepoRef>),
     #[serde(rename = "com.atproto.repo.strongRef")]
-    ComAtprotoRepoStrongRefMain(crate::com::atproto::repo::strong_ref::Main),
+    ComAtprotoRepoStrongRefMain(Box<crate::com::atproto::repo::strong_ref::Main>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum BlobViewDetailsEnum {
     #[serde(rename = "com.atproto.admin.defs#imageDetails")]
-    ImageDetails(ImageDetails),
+    ImageDetails(Box<ImageDetails>),
     #[serde(rename = "com.atproto.admin.defs#videoDetails")]
-    VideoDetails(VideoDetails),
+    VideoDetails(Box<VideoDetails>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ReportViewDetailSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoView")]
-    RepoView(RepoView),
+    RepoView(Box<RepoView>),
     #[serde(rename = "com.atproto.admin.defs#recordView")]
-    RecordView(RecordView),
+    RecordView(Box<RecordView>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum ReportViewSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoRef")]
-    RepoRef(RepoRef),
+    RepoRef(Box<RepoRef>),
     #[serde(rename = "com.atproto.repo.strongRef")]
-    ComAtprotoRepoStrongRefMain(crate::com::atproto::repo::strong_ref::Main),
+    ComAtprotoRepoStrongRefMain(Box<crate::com::atproto::repo::strong_ref::Main>),
 }
diff --git a/atrium-api/src/com/atproto/admin/take_moderation_action.rs b/atrium-api/src/com/atproto/admin/take_moderation_action.rs
index 6d0fec52..e2215e05 100644
--- a/atrium-api/src/com/atproto/admin/take_moderation_action.rs
+++ b/atrium-api/src/com/atproto/admin/take_moderation_action.rs
@@ -30,7 +30,7 @@ pub struct Input {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub negate_label_vals: Option<Vec<String>>,
     pub reason: String,
-    pub subject: Box<InputSubjectEnum>,
+    pub subject: InputSubjectEnum,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub subject_blob_cids: Option<Vec<String>>,
 }
@@ -44,7 +44,7 @@ pub enum Error {
 #[serde(tag = "$type")]
 pub enum InputSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoRef")]
-    ComAtprotoAdminDefsRepoRef(crate::com::atproto::admin::defs::RepoRef),
+    ComAtprotoAdminDefsRepoRef(Box<crate::com::atproto::admin::defs::RepoRef>),
     #[serde(rename = "com.atproto.repo.strongRef")]
-    ComAtprotoRepoStrongRefMain(crate::com::atproto::repo::strong_ref::Main),
+    ComAtprotoRepoStrongRefMain(Box<crate::com::atproto::repo::strong_ref::Main>),
 }
diff --git a/atrium-api/src/com/atproto/label/subscribe_labels.rs b/atrium-api/src/com/atproto/label/subscribe_labels.rs
index 8f79fc95..86624bc3 100644
--- a/atrium-api/src/com/atproto/label/subscribe_labels.rs
+++ b/atrium-api/src/com/atproto/label/subscribe_labels.rs
@@ -18,3 +18,11 @@ pub struct Labels {
     pub labels: Vec<crate::com::atproto::label::defs::Label>,
     pub seq: i32,
 }
+#[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
+#[serde(tag = "$type")]
+pub enum Message {
+    #[serde(rename = "com.atproto.label.subscribeLabels#labels")]
+    Labels(Box<Labels>),
+    #[serde(rename = "com.atproto.label.subscribeLabels#info")]
+    Info(Box<Info>),
+}
diff --git a/atrium-api/src/com/atproto/moderation/create_report.rs b/atrium-api/src/com/atproto/moderation/create_report.rs
index fd604885..590df8ab 100644
--- a/atrium-api/src/com/atproto/moderation/create_report.rs
+++ b/atrium-api/src/com/atproto/moderation/create_report.rs
@@ -23,7 +23,7 @@ pub struct Input {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reason: Option<String>,
     pub reason_type: crate::com::atproto::moderation::defs::ReasonType,
-    pub subject: Box<InputSubjectEnum>,
+    pub subject: InputSubjectEnum,
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
@@ -34,7 +34,7 @@ pub struct Output {
     pub reason: Option<String>,
     pub reason_type: crate::com::atproto::moderation::defs::ReasonType,
     pub reported_by: String,
-    pub subject: Box<OutputSubjectEnum>,
+    pub subject: OutputSubjectEnum,
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "error", content = "message")]
@@ -43,15 +43,15 @@ pub enum Error {}
 #[serde(tag = "$type")]
 pub enum InputSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoRef")]
-    ComAtprotoAdminDefsRepoRef(crate::com::atproto::admin::defs::RepoRef),
+    ComAtprotoAdminDefsRepoRef(Box<crate::com::atproto::admin::defs::RepoRef>),
     #[serde(rename = "com.atproto.repo.strongRef")]
-    ComAtprotoRepoStrongRefMain(crate::com::atproto::repo::strong_ref::Main),
+    ComAtprotoRepoStrongRefMain(Box<crate::com::atproto::repo::strong_ref::Main>),
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "$type")]
 pub enum OutputSubjectEnum {
     #[serde(rename = "com.atproto.admin.defs#repoRef")]
-    ComAtprotoAdminDefsRepoRef(crate::com::atproto::admin::defs::RepoRef),
+    ComAtprotoAdminDefsRepoRef(Box<crate::com::atproto::admin::defs::RepoRef>),
     #[serde(rename = "com.atproto.repo.strongRef")]
-    ComAtprotoRepoStrongRefMain(crate::com::atproto::repo::strong_ref::Main),
+    ComAtprotoRepoStrongRefMain(Box<crate::com::atproto::repo::strong_ref::Main>),
 }
diff --git a/atrium-api/src/com/atproto/repo/apply_writes.rs b/atrium-api/src/com/atproto/repo/apply_writes.rs
index 10bf45c2..accd5594 100644
--- a/atrium-api/src/com/atproto/repo/apply_writes.rs
+++ b/atrium-api/src/com/atproto/repo/apply_writes.rs
@@ -27,7 +27,7 @@ pub struct Input {
     #[doc = "Validate the records?"]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub validate: Option<bool>,
-    pub writes: Vec<Box<InputWritesItem>>,
+    pub writes: Vec<InputWritesItem>,
 }
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(tag = "error", content = "message")]
@@ -65,9 +65,9 @@ pub struct Update {
 #[serde(tag = "$type")]
 pub enum InputWritesItem {
     #[serde(rename = "com.atproto.repo.applyWrites#create")]
-    Create(Create),
+    Create(Box<Create>),
     #[serde(rename = "com.atproto.repo.applyWrites#update")]
-    Update(Update),
+    Update(Box<Update>),
     #[serde(rename = "com.atproto.repo.applyWrites#delete")]
-    Delete(Delete),
+    Delete(Box<Delete>),
 }
diff --git a/atrium-api/src/com/atproto/sync/subscribe_repos.rs b/atrium-api/src/com/atproto/sync/subscribe_repos.rs
index 7291cf88..8229989b 100644
--- a/atrium-api/src/com/atproto/sync/subscribe_repos.rs
+++ b/atrium-api/src/com/atproto/sync/subscribe_repos.rs
@@ -7,7 +7,14 @@ pub struct SubscribeRepos;
 #[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct Commit {
+    pub blobs: Vec<cid::Cid>,
+    #[doc = "CAR file containing relevant blocks"]
+    #[serde(with = "serde_bytes")]
+    pub blocks: Vec<u8>,
+    pub commit: cid::Cid,
     pub ops: Vec<RepoOp>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub prev: Option<cid::Cid>,
     pub rebase: bool,
     pub repo: String,
     pub seq: i32,
@@ -36,7 +43,8 @@ pub struct Info {
 #[serde(rename_all = "camelCase")]
 pub struct Migrate {
     pub did: String,
-    pub migrate_to: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub migrate_to: Option<String>,
     pub seq: i32,
     pub time: String,
 }
@@ -45,6 +53,8 @@ pub struct Migrate {
 #[serde(rename_all = "camelCase")]
 pub struct RepoOp {
     pub action: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cid: Option<cid::Cid>,
     pub path: String,
 }
 #[doc = "`com.atproto.sync.subscribeRepos#tombstone`"]
@@ -55,3 +65,17 @@ pub struct Tombstone {
     pub seq: i32,
     pub time: String,
 }
+#[derive(serde :: Serialize, serde :: Deserialize, Debug, Clone, PartialEq, Eq)]
+#[serde(tag = "$type")]
+pub enum Message {
+    #[serde(rename = "com.atproto.sync.subscribeRepos#commit")]
+    Commit(Box<Commit>),
+    #[serde(rename = "com.atproto.sync.subscribeRepos#handle")]
+    Handle(Box<Handle>),
+    #[serde(rename = "com.atproto.sync.subscribeRepos#migrate")]
+    Migrate(Box<Migrate>),
+    #[serde(rename = "com.atproto.sync.subscribeRepos#tombstone")]
+    Tombstone(Box<Tombstone>),
+    #[serde(rename = "com.atproto.sync.subscribeRepos#info")]
+    Info(Box<Info>),
+}
diff --git a/atrium-api/src/records.rs b/atrium-api/src/records.rs
index ffc56efb..dc74549d 100644
--- a/atrium-api/src/records.rs
+++ b/atrium-api/src/records.rs
@@ -4,19 +4,19 @@
 #[serde(tag = "$type")]
 pub enum Record {
     #[serde(rename = "app.bsky.actor.profile")]
-    AppBskyActorProfile(crate::app::bsky::actor::profile::Record),
+    AppBskyActorProfile(Box<crate::app::bsky::actor::profile::Record>),
     #[serde(rename = "app.bsky.feed.like")]
-    AppBskyFeedLike(crate::app::bsky::feed::like::Record),
+    AppBskyFeedLike(Box<crate::app::bsky::feed::like::Record>),
     #[serde(rename = "app.bsky.feed.post")]
-    AppBskyFeedPost(crate::app::bsky::feed::post::Record),
+    AppBskyFeedPost(Box<crate::app::bsky::feed::post::Record>),
     #[serde(rename = "app.bsky.feed.repost")]
-    AppBskyFeedRepost(crate::app::bsky::feed::repost::Record),
+    AppBskyFeedRepost(Box<crate::app::bsky::feed::repost::Record>),
     #[serde(rename = "app.bsky.graph.block")]
-    AppBskyGraphBlock(crate::app::bsky::graph::block::Record),
+    AppBskyGraphBlock(Box<crate::app::bsky::graph::block::Record>),
     #[serde(rename = "app.bsky.graph.follow")]
-    AppBskyGraphFollow(crate::app::bsky::graph::follow::Record),
+    AppBskyGraphFollow(Box<crate::app::bsky::graph::follow::Record>),
     #[serde(rename = "app.bsky.graph.list")]
-    AppBskyGraphList(crate::app::bsky::graph::list::Record),
+    AppBskyGraphList(Box<crate::app::bsky::graph::list::Record>),
     #[serde(rename = "app.bsky.graph.listitem")]
-    AppBskyGraphListitem(crate::app::bsky::graph::listitem::Record),
+    AppBskyGraphListitem(Box<crate::app::bsky::graph::listitem::Record>),
 }
diff --git a/atrium-codegen/src/schema.rs b/atrium-codegen/src/schema.rs
index eb13e389..b0388e02 100644
--- a/atrium-codegen/src/schema.rs
+++ b/atrium-codegen/src/schema.rs
@@ -29,11 +29,19 @@ pub(crate) fn find_ref_unions(defs: &HashMap<String, LexUserType>) -> Vec<(Strin
                     }
                 }
             }
-            LexUserType::XrpcSubscription(_) => {
-                // TODO
+            LexUserType::XrpcSubscription(subscription) => {
+                if let Some(message) = &subscription.message {
+                    if let Some(schema) = &message.schema {
+                        find_ref_unions_in_subscription_message_schema(
+                            schema,
+                            "Message",
+                            &mut unions,
+                        );
+                    }
+                }
             }
             LexUserType::Array(array) => {
-                find_ref_unions_in_array(array, &key.to_pascal_case(), &mut unions)
+                find_ref_unions_in_array(array, &key.to_pascal_case(), &mut unions);
             }
             LexUserType::Object(object) => {
                 find_ref_unions_in_object(object, &key.to_pascal_case(), &mut unions);
@@ -57,6 +65,22 @@ fn find_ref_unions_in_body_schema(
     }
 }
 
+fn find_ref_unions_in_subscription_message_schema(
+    schema: &LexXrpcSubscriptionMessageSchema,
+    name: &str,
+    unions: &mut Vec<(String, LexRefUnion)>,
+) {
+    match schema {
+        LexXrpcSubscriptionMessageSchema::Union(union) => {
+            unions.push((name.into(), union.clone()));
+        }
+        LexXrpcSubscriptionMessageSchema::Object(object) => {
+            find_ref_unions_in_object(object, name, unions)
+        }
+        _ => {}
+    }
+}
+
 fn find_ref_unions_in_array(array: &LexArray, name: &str, unions: &mut Vec<(String, LexRefUnion)>) {
     if let LexArrayItem::Union(union) = &array.items {
         unions.push((format!("{}Item", name), union.clone()));
diff --git a/atrium-codegen/src/token_stream.rs b/atrium-codegen/src/token_stream.rs
index fd5480b2..8121b788 100644
--- a/atrium-codegen/src/token_stream.rs
+++ b/atrium-codegen/src/token_stream.rs
@@ -334,11 +334,16 @@ impl LexConverter {
         let description = self.description(&object.description);
         let derives = self::derives()?;
         let struct_name = format_ident!("{}", name.to_pascal_case());
-        let required = if let Some(required) = &object.required {
+        let mut required = if let Some(required) = &object.required {
             HashSet::from_iter(required)
         } else {
             HashSet::new()
         };
+        if let Some(nullable) = &object.nullable {
+            for key in nullable {
+                required.remove(&key);
+            }
+        }
         let mut fields = Vec::new();
         if let Some(properties) = &object.properties {
             for key in properties.keys().sorted() {
@@ -366,7 +371,7 @@ impl LexConverter {
         is_required: bool,
         object_name: &str,
     ) -> Result<TokenStream> {
-        let (description, field_type) = match property {
+        let (description, mut field_type) = match property {
             LexObjectProperty::Ref(r#ref) => self.ref_type(r#ref)?,
             LexObjectProperty::Union(union) => self.union_type(
                 union,
@@ -399,17 +404,23 @@ impl LexConverter {
                 name.to_snake_case()
             }
         );
-        Ok(if is_required {
-            quote! {
-                #description
-                pub #field_name: #field_type,
-            }
-        } else {
-            quote! {
-                #description
+        let mut attributes = match property {
+            LexObjectProperty::Bytes(_) => quote! {
+                #[serde(with = "serde_bytes")]
+            },
+            _ => quote!(),
+        };
+        if !is_required {
+            field_type = quote!(Option<#field_type>);
+            attributes = quote! {
+                #attributes
                 #[serde(skip_serializing_if = "Option::is_none")]
-                pub #field_name: Option<#field_type>,
-            }
+            };
+        }
+        Ok(quote! {
+            #description
+            #attributes
+            pub #field_name: #field_type,
         })
     }
     fn string(&self, name: &str, string: &LexString) -> Result<TokenStream> {
@@ -431,18 +442,15 @@ impl LexConverter {
     ) -> Result<(TokenStream, TokenStream)> {
         let description = self.description(&union.description);
         let enum_type_name = format_ident!("{}", enum_name);
-        // Use `Box` to avoid recursive.
-        Ok((description, quote!(Box<#enum_type_name>)))
+        Ok((description, quote!(#enum_type_name)))
     }
     fn bytes_type(&self, bytes: &LexBytes) -> Result<(TokenStream, TokenStream)> {
         let description = self.description(&bytes.description);
-        // TODO
-        Ok((description, quote!()))
+        Ok((description, quote!(Vec<u8>)))
     }
     fn cid_link_type(&self, cid_link: &LexCidLink) -> Result<(TokenStream, TokenStream)> {
         let description = self.description(&cid_link.description);
-        // TODO
-        Ok((description, quote!()))
+        Ok((description, quote!(cid::Cid)))
     }
     fn array_type(
         &self,
@@ -539,7 +547,7 @@ pub(crate) fn refs_enum(
         let name = format_ident!("{}", parts.join(""));
         variants.push(quote! {
             #[serde(rename = #rename)]
-            #name(#path)
+            #name(Box<#path>)
         });
     }
     Ok(quote! {
diff --git a/atrium-xrpc-server/Cargo.toml b/atrium-xrpc-server/Cargo.toml
new file mode 100644
index 00000000..8bb12db6
--- /dev/null
+++ b/atrium-xrpc-server/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+name = "atrium-xrpc-server"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+atrium-api = { path = "../atrium-api" }
+libipld-core = { version = "0.16.0", features = ["serde-codec"] }
+serde_ipld_dagcbor = "0.3.0"
diff --git a/atrium-xrpc-server/src/lib.rs b/atrium-xrpc-server/src/lib.rs
new file mode 100644
index 00000000..baf29e06
--- /dev/null
+++ b/atrium-xrpc-server/src/lib.rs
@@ -0,0 +1,1 @@
+pub mod stream;
diff --git a/atrium-xrpc-server/src/stream.rs b/atrium-xrpc-server/src/stream.rs
new file mode 100644
index 00000000..b63fcef5
--- /dev/null
+++ b/atrium-xrpc-server/src/stream.rs
@@ -0,0 +1,1 @@
+pub mod frames;
diff --git a/atrium-xrpc-server/src/stream/frames.rs b/atrium-xrpc-server/src/stream/frames.rs
new file mode 100644
index 00000000..1c5dc830
--- /dev/null
+++ b/atrium-xrpc-server/src/stream/frames.rs
@@ -0,0 +1,164 @@
+use atrium_api::com::atproto::sync::subscribe_repos::{Commit, Message};
+use libipld_core::ipld::Ipld;
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+use std::io::Cursor;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct FrameTypeError;
+
+impl Display for FrameTypeError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "invalid frame type")
+    }
+}
+
+impl Error for FrameTypeError {}
+
+// original definition:
+//```
+// export enum FrameType {
+//   Message = 1,
+//   Error = -1,
+// }
+// export const messageFrameHeader = z.object({
+//   op: z.literal(FrameType.Message), // Frame op
+//   t: z.string().optional(), // Message body type discriminator
+// })
+// export type MessageFrameHeader = z.infer<typeof messageFrameHeader>
+// export const errorFrameHeader = z.object({
+//   op: z.literal(FrameType.Error),
+// })
+// export type ErrorFrameHeader = z.infer<typeof errorFrameHeader>
+// ```
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum FrameHeader {
+    Message(Option<String>),
+    Error,
+}
+
+impl TryFrom<Ipld> for FrameHeader {
+    type Error = FrameTypeError;
+
+    fn try_from(value: Ipld) -> Result<Self, <FrameHeader as TryFrom<Ipld>>::Error> {
+        if let Ipld::Map(map) = value {
+            if let Some(Ipld::Integer(i)) = map.get("op") {
+                match i {
+                    1 => {
+                        let t = if let Some(Ipld::String(s)) = map.get("t") {
+                            Some(s.clone())
+                        } else {
+                            None
+                        };
+                        return Ok(FrameHeader::Message(t));
+                    }
+                    -1 => return Ok(FrameHeader::Error),
+                    _ => {}
+                }
+            }
+        }
+        Err(FrameTypeError)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Frame {
+    Message(Box<MessageFrame>),
+    Error(ErrorFrame),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MessageFrame {
+    pub body: Message,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ErrorFrame {
+    // TODO
+    // body: Value,
+}
+
+impl TryFrom<&[u8]> for Frame {
+    type Error = Box<dyn Error>;
+
+    fn try_from(value: &[u8]) -> Result<Self, <Frame as TryFrom<&[u8]>>::Error> {
+        let mut cursor = Cursor::new(value);
+        let (left, right) = match serde_ipld_dagcbor::from_reader::<Ipld, _>(&mut cursor) {
+            Err(serde_ipld_dagcbor::DecodeError::TrailingData) => {
+                value.split_at(cursor.position() as usize)
+            }
+            _ => {
+                // TODO
+                return Err(Box::new(FrameTypeError));
+            }
+        };
+        let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(left)?;
+        let header = FrameHeader::try_from(ipld)?;
+        match header {
+            FrameHeader::Message(t) => match t.as_deref() {
+                Some("#commit") => Ok(Frame::Message(Box::new(MessageFrame {
+                    body: Message::Commit(Box::new(serde_ipld_dagcbor::from_slice::<Commit>(
+                        right,
+                    )?)),
+                }))),
+                _ => unimplemented!("{t:?}"),
+            },
+            FrameHeader::Error => Ok(Frame::Error(ErrorFrame {})),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn serialized_data(s: &str) -> Vec<u8> {
+        assert!(s.len() % 2 == 0);
+        let b2u = |b: u8| match b {
+            b'0'..=b'9' => b - b'0',
+            b'a'..=b'f' => b - b'a' + 10,
+            _ => unreachable!(),
+        };
+        s.as_bytes()
+            .chunks(2)
+            .map(|b| (b2u(b[0]) << 4) + b2u(b[1]))
+            .collect()
+    }
+
+    #[test]
+    fn deserialize_message_frame_header() {
+        // {"op": 1, "t": "#commit"}
+        let data = serialized_data("a2626f700161746723636f6d6d6974");
+        let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
+        assert_eq!(
+            FrameHeader::try_from(ipld),
+            Ok(FrameHeader::Message(Some(String::from("#commit"))))
+        );
+    }
+
+    #[test]
+    fn deserialize_error_frame_header() {
+        // {"op": -1}
+        let data = serialized_data("a1626f7020");
+        let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
+        assert_eq!(FrameHeader::try_from(ipld), Ok(FrameHeader::Error));
+    }
+
+    #[test]
+    fn deserialize_invalid_frame_header() {
+        {
+            // {"op": 2, "t": "#commit"}
+            let data = serialized_data("a2626f700261746723636f6d6d6974");
+            let ipld =
+                serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
+            assert_eq!(FrameHeader::try_from(ipld), Err(FrameTypeError));
+        }
+        {
+            // {"op": -2}
+            let data = serialized_data("a1626f7021");
+            let ipld =
+                serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
+            assert_eq!(FrameHeader::try_from(ipld), Err(FrameTypeError));
+        }
+    }
+}
diff --git a/examples/firehose/Cargo.toml b/examples/firehose/Cargo.toml
new file mode 100644
index 00000000..829971ef
--- /dev/null
+++ b/examples/firehose/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "firehose"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+atrium-api = { path = "../../atrium-api" }
+atrium-xrpc-server = { path = "../../atrium-xrpc-server" }
+ciborium = "0.2.1"
+futures = "0.3.28"
+rs-car = "0.4.1"
+tokio = { version = "1.28.1", features = ["full"] }
+tokio-tungstenite = { version = "0.18.0", features = ["native-tls"] }
diff --git a/examples/firehose/src/main.rs b/examples/firehose/src/main.rs
new file mode 100644
index 00000000..64aedb60
--- /dev/null
+++ b/examples/firehose/src/main.rs
@@ -0,0 +1,48 @@
+use atrium_api::app::bsky::feed::post::Record;
+use atrium_api::com::atproto::sync::subscribe_repos::Message;
+use atrium_xrpc_server::stream::frames::Frame;
+use futures::StreamExt;
+use tokio_tungstenite::{connect_async, tungstenite};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let (mut stream, _) =
+        connect_async("wss://bsky.social/xrpc/com.atproto.sync.subscribeRepos").await?;
+
+    while let Some(Ok(tungstenite::Message::Binary(message))) = stream.next().await {
+        process_message(&message).await?;
+    }
+    Ok(())
+}
+
+async fn process_message(message: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
+    match Frame::try_from(message)? {
+        Frame::Message(message) => {
+            match message.body {
+                Message::Commit(commit) => {
+                    for op in commit.ops {
+                        let collection = op.path.split('/').next().expect("op.path is empty");
+                        if op.action != "create" || collection != "app.bsky.feed.post" {
+                            continue;
+                        }
+                        let (items, _) =
+                            rs_car::car_read_all(&mut commit.blocks.as_slice(), true).await?;
+                        if let Some((_, item)) = items.iter().find(|(cid, _)| Some(*cid) == op.cid)
+                        {
+                            if let Ok(value) =
+                                ciborium::de::from_reader::<Record, _>(&mut item.as_slice())
+                            {
+                                println!("{}: {}", value.created_at, value.text);
+                            } else {
+                                // TODO
+                            }
+                        }
+                    }
+                }
+                _ => unimplemented!("{:?}", message.body),
+            }
+        }
+        Frame::Error(err) => panic!("{err:?}"),
+    }
+    Ok(())
+}


Test patch: