very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust
fjall
at-protocol
atproto
indexer
[examples] comment statusphere a bit more
+15
-14
+15
-14
examples/statusphere.rs
+15
-14
examples/statusphere.rs
···
23
23
use hydrant::config::Config;
24
24
use hydrant::control::{EventStream, Hydrant, ReposControl};
25
25
use hydrant::filter::FilterMode;
26
+
use jacquard_common::types::did::Did;
26
27
use jacquard_common::types::tid::Tid;
27
28
use scc::HashMap;
28
29
···
101
102
}
102
103
103
104
async fn handle_stream(index: Arc<StatusIndex>, repos: ReposControl, mut stream: EventStream) {
105
+
// get handle of did through the hydrant api
106
+
let get_handle = async |did: &Did<'_>| {
107
+
repos
108
+
.get(did)
109
+
.await
110
+
.ok()
111
+
.flatten()
112
+
.and_then(|info| info.handle)
113
+
.unwrap_or_else(|| did.to_string())
114
+
};
104
115
while let Some(event) = stream.next().await {
105
116
if let Some(rec) = event.record {
106
117
let did = rec.did.as_str().to_owned();
···
119
130
.and_then(|v| v.as_str())
120
131
.unwrap_or("");
121
132
if index.set(did.clone(), emoji.clone(), created_at) {
122
-
let name = repos
123
-
.get(&rec.did)
124
-
.await
125
-
.ok()
126
-
.flatten()
127
-
.and_then(|info| info.handle)
128
-
.unwrap_or(did);
133
+
let name = get_handle(&rec.did).await;
129
134
println!("[{created_at}] {name}: {emoji}");
130
135
}
131
136
}
132
137
"delete" => {
133
-
let name = repos
134
-
.get(&rec.did)
135
-
.await
136
-
.ok()
137
-
.flatten()
138
-
.and_then(|info| info.handle)
139
-
.unwrap_or(did.clone());
138
+
let name = get_handle(&rec.did).await;
140
139
index.delete(&did);
140
+
// parse the tid to use as date since createdAt doesnt make sense here
141
141
let date = Tid::from_str(&rec.rkey)
142
142
.ok()
143
143
.and_then(|tid| DateTime::from_timestamp_micros(tid.timestamp() as i64))
···
183
183
184
184
let index = Arc::new(StatusIndex::new());
185
185
tokio::select! {
186
+
// this finally starts hydrant, so it will start crawling and backfilling etc.
186
187
r = hydrant.run() => r,
187
188
_ = run_ticker(index.clone()) => Ok(()),
188
189
_ = handle_stream(index.clone(), hydrant.repos.clone(), stream) => Ok(()),