Fix firehose

Cargo.toml

@@ -4,7 +4,6 @@ members = [
     "atrium-cli",
     "atrium-xrpc",
     "atrium-xrpc-client",
-    "atrium-xrpc-server",
 ]
 # Examples show how to use the latest published crates, not the workspace state.
 exclude = [
@@ -25,7 +24,6 @@ keywords = ["atproto", "bluesky"]
 atrium-api = { version = "0.18.1", path = "atrium-api" }
 atrium-xrpc = { version = "0.10.0", path = "atrium-xrpc" }
 atrium-xrpc-client = { version = "0.4.0", path = "atrium-xrpc-client" }
-atrium-xrpc-server = { version = "0.1.0", path = "atrium-xrpc-server" }
 
 # async in traits
 # Can be removed once MSRV is at least 1.75.0.

examples/firehose/Cargo.toml

@@ -6,10 +6,13 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-atrium-xrpc-server = { path = "../../atrium-xrpc-server" }
-atrium-api = "0.15"
-ciborium = "0.2.1"
+anyhow = "1.0.80"
+atrium-api = { version = "0.18.1", features = ["dag-cbor"] }
+chrono = "0.4.34"
 futures = "0.3.28"
+ipld-core = { version = "0.2.0", features = ["serde"] }
 rs-car = "0.4.1"
-tokio = { version = "1.28.1", features = ["full"] }
+serde_ipld_dagcbor = { git = "https://github.com/ipld/serde_ipld_dagcbor.git", rev = "297a6c26c8c89807e6602cab9803ef2c4ae8b459" }
+tokio = { version = "1.36.0", features = ["full"] }
 tokio-tungstenite = { version = "0.21.0", features = ["native-tls"] }
+trait-variant = "0.1.1"

examples/firehose/README.md

@@ -0,0 +1,3 @@
+# Firehose example
+
+https://github.com/sugyan/atrium/assets/80381/58245773-8a37-4fec-a299-2279b4d2e72f

examples/firehose/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod stream;
+pub mod subscription;

examples/firehose/src/main.rs

@@ -1,55 +1,82 @@
 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 atrium_api::com::atproto::sync::subscribe_repos::{Commit, NSID};
+use atrium_api::types::{CidLink, Collection};
+use chrono::Local;
+use firehose::stream::frames::Frame;
+use firehose::subscription::{CommitHandler, Subscription};
 use futures::StreamExt;
-use tokio_tungstenite::{connect_async, tungstenite};
+use tokio::net::TcpStream;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let (mut stream, _) =
-        connect_async("wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos").await?;
+struct RepoSubscription {
+    stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
+}
+
+impl RepoSubscription {
+    async fn new(bgs: &str) -> Result<Self, Box<dyn std::error::Error>> {
+        let (stream, _) = connect_async(format!("wss://{bgs}/xrpc/{NSID}")).await?;
+        Ok(RepoSubscription { stream })
+    }
+    async fn run(&mut self, handler: impl CommitHandler) -> Result<(), Box<dyn std::error::Error>> {
+        while let Some(result) = self.next().await {
+            if let Ok(Frame::Message(Some(t), message)) = result {
+                if t.as_str() == "#commit" {
+                    let commit = serde_ipld_dagcbor::from_reader(message.body.as_slice())?;
+                    handler.handle_commit(&commit).await?;
+                }
+            }
+        }
+        Ok(())
+    }
+}
 
-    while let Some(Ok(tungstenite::Message::Binary(message))) = stream.next().await {
-        process_message(&message).await.unwrap();
+impl Subscription for RepoSubscription {
+    async fn next(&mut self) -> Option<Result<Frame, <Frame as TryFrom<&[u8]>>::Error>> {
+        if let Some(Ok(Message::Binary(data))) = self.stream.next().await {
+            Some(Frame::try_from(data.as_slice()))
+        } else {
+            None
+        }
     }
-    Ok(())
 }
 
-async fn process_message(message: &[u8]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    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 {
-                                println!("FAILED: could not deserialize post from item of length: {}", item.len());
+struct Firehose;
 
-                            }
-                        } else {
-                            println!(
-                                "FAILED: could not find item with operation cid {:?} out of {} items",
-                                op.cid,
-                                items.len()
-                            );
-                        }
-                    }
+impl CommitHandler for Firehose {
+    async fn handle_commit(&self, commit: &Commit) -> Result<(), Box<dyn std::error::Error>> {
+        for op in &commit.ops {
+            let collection = op.path.split('/').next().expect("op.path is empty");
+            if op.action != "create" || collection != atrium_api::app::bsky::feed::Post::NSID {
+                continue;
+            }
+            let (items, _) = rs_car::car_read_all(&mut commit.blocks.as_slice(), true).await?;
+            if let Some((_, item)) = items.iter().find(|(cid, _)| Some(CidLink(*cid)) == op.cid) {
+                let record = serde_ipld_dagcbor::from_reader::<Record, _>(&mut item.as_slice())?;
+                println!(
+                    "{} - {}",
+                    record.created_at.as_ref().with_timezone(&Local),
+                    commit.repo.as_str()
+                );
+                for line in record.text.split('\n') {
+                    println!("  {line}");
                 }
-                _ => unimplemented!("{:?}", message.body),
+            } else {
+                panic!(
+                    "FAILED: could not find item with operation cid {:?} out of {} items",
+                    op.cid,
+                    items.len()
+                );
             }
         }
-        Frame::Error(err) => panic!("{err:?}"),
+        Ok(())
     }
-    Ok(())
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    RepoSubscription::new("bsky.network")
+        .await?
+        .run(Firehose)
+        .await
 }