Something went wrong. Try again.
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
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081use std::sync::Arc;
use bytes::Bytes;use serde_json::value::RawValue;
/// pre-built commit data for live jetstream tailing./// `record` holds pre-serialized json so it can be embedded raw into each op's/// event bytes without re-encoding, and arc-shared across ops that reference the/// same cid so the serialization happens exactly once per unique block.#[derive(Clone)]pub(crate) struct JetstreamEphemeral { pub did: String, pub rev: String, pub operation: String, pub collection: String, pub rkey: String, pub record: Option<Arc<RawValue>>, pub cid: Option<String>, pub live: bool,}
impl JetstreamEphemeral { /// the live json for this commit op, stamped with its committed time. pub(crate) fn to_json(&self, time_us: i64) -> Option<Bytes> { let event = crate::control::stream::JetstreamEvent { did: &self.did, time_us, payload: crate::control::stream::JetstreamPayload::Commit { commit: crate::control::stream::JetstreamCommit { rev: &self.rev, operation: &self.operation, collection: &self.collection, rkey: &self.rkey, record: self.record.as_deref(), cid: self.cid.clone(), live: self.live, }, }, }; serde_json::to_vec(&event).ok().map(Bytes::from) }}
#[cfg(all(feature = "indexer_stream", feature = "jetstream"))]pub(crate) fn build_ephemeral_from_stored( did: &crate::db::types::TrimmedDid, rev: &str, operation: &str, collection: &str, rkey: &crate::db::types::DbRkey, data: &crate::types::StoredData, inline_block: Option<&Bytes>, live: bool,) -> Option<JetstreamEphemeral> { use crate::types::StoredData;
let (cid, record) = match data { StoredData::Ptr(cid) => { let record = crate::types::record_json(inline_block?).ok()?; (Some(cid.to_string()), Some(record)) } StoredData::Block(block) => { let cid = crate::car::compute_cid(block); let record = crate::types::record_json(block).ok()?; (Some(cid.to_string()), Some(record)) } StoredData::Nothing => (None, None), };
Some(JetstreamEphemeral { did: did.to_did().to_string(), rev: rev.to_string(), operation: operation.to_string(), collection: collection.to_string(), rkey: rkey.to_smolstr().to_string(), record, cid, live, })}