comprehensive_example/
comprehensive-example.rs

1use async_trait::async_trait;
2use rocketman::{
3    connection::JetstreamConnection,
4    handler::{self, Ingestors},
5    ingestion::LexiconIngestor,
6    options::JetstreamOptions,
7    types::event::{Account, Commit, Event, Identity},
8};
9use serde_json::Value;
10use std::{sync::Arc, sync::Mutex};
11
12#[tokio::main]
13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18
19    // init the builder with wanted collections
20    let opts = JetstreamOptions::builder()
21        .wanted_collections(vec![
22            "app.bsky.feed.post".to_string(),
23            "xyz.statusphere.status".to_string(),
24        ])
25        .build();
26
27    // create the jetstream connector
28    let jetstream = JetstreamConnection::new(opts);
29
30    // create your ingestors
31    let mut ingestors = Ingestors::new();
32
33    // register commit ingestors
34    ingestors.commits.insert(
35        "xyz.statusphere.status".to_string(),
36        Box::new(StatusphereIngestor),
37    );
38
39    // register identity ingestor
40    ingestors.identity = Some(Box::new(IdentityIngestor));
41
42    // register account ingestor
43    ingestors.account = Some(Box::new(AccountIngestor));
44
45    // tracks the last message we've processed
46    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
47
48    // get channels
49    let msg_rx = jetstream.get_msg_rx();
50    let reconnect_tx = jetstream.get_reconnect_tx();
51
52    // spawn a task to process messages from the queue.
53    let c_cursor = cursor.clone();
54    tokio::spawn(async move {
55        while let Ok(message) = msg_rx.recv().await {
56            if let Err(e) =
57                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
58                    .await
59            {
60                eprintln!("Error processing message: {}", e);
61            };
62        }
63    });
64
65    // connect to jetstream
66    if let Err(e) = jetstream.connect(cursor.clone()).await {
67        eprintln!("Failed to connect to Jetstream: {}", e);
68        std::process::exit(1);
69    }
70}
71
72/// Handles Statusphere status updates
73pub struct StatusphereIngestor;
74
75#[async_trait]
76impl LexiconIngestor for StatusphereIngestor {
77    async fn ingest(&self, message: Event<Value>) -> anyhow::Result<()> {
78        if let Some(Commit {
79            record: Some(record),
80            operation,
81            ..
82        }) = message.commit
83        {
84            if let Some(Value::String(status)) = record.get("status") {
85                println!("[STATUSPHERE] [{operation:?}] {status:?}");
86            }
87        }
88        Ok(())
89    }
90}
91
92/// Handles identity events (handle changes, etc.)
93pub struct IdentityIngestor;
94
95#[async_trait]
96impl LexiconIngestor for IdentityIngestor {
97    async fn ingest(&self, message: Event<Value>) -> anyhow::Result<()> {
98        if let Some(Identity {
99            did,
100            handle,
101            seq,
102            time,
103        }) = message.identity
104        {
105            println!("[IDENTITY] seq={seq} did={did} handle={handle:?} time={time}");
106        }
107        Ok(())
108    }
109}
110
111/// Handles account events (active/inactive status)
112pub struct AccountIngestor;
113
114#[async_trait]
115impl LexiconIngestor for AccountIngestor {
116    async fn ingest(&self, message: Event<Value>) -> anyhow::Result<()> {
117        if let Some(Account {
118            did,
119            handle,
120            seq,
121            time,
122            status,
123        }) = message.account
124        {
125            let handle_str = handle
126                .as_ref()
127                .map(|h| format!(" handle={h}"))
128                .unwrap_or_default();
129            let status_str = status
130                .as_ref()
131                .map(|s| format!(" status={s:?}"))
132                .unwrap_or_default();
133            println!("[ACCOUNT] seq={seq} did={did}{handle_str}{status_str} time={time}");
134        }
135        Ok(())
136    }
137}
comprehensive_example/comprehensive-example.rs

comprehensive_example/
comprehensive-example.rs
