use 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,
})
}