Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141use std::sync::Arc;use std::sync::atomic::Ordering;
use bytes::Bytes;use fjall::OwnedWriteBatch;use miette::{IntoDiagnostic, Result};use serde_json::value::RawValue;
use crate::db::{Db, keys};use crate::types::{JetstreamBroadcast, StoredJetstreamEvent};
/// 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,}
pub(crate) fn stage_event( batch: &mut OwnedWriteBatch, db: &Db, event: StoredJetstreamEvent<'_>, ephemeral: Option<JetstreamEphemeral>,) -> Result<JetstreamBroadcast> { let id = db.jetstream.next_id.fetch_add(1, Ordering::SeqCst); let time_us = next_time_us(db); let ephemeral = ephemeral.and_then(|data| { let json_event = crate::control::stream::JetstreamEvent { did: &data.did, time_us, payload: crate::control::stream::JetstreamPayload::Commit { commit: crate::control::stream::JetstreamCommit { rev: &data.rev, operation: &data.operation, collection: &data.collection, rkey: &data.rkey, record: data.record.as_deref(), cid: data.cid, live: data.live, }, }, }; serde_json::to_vec(&json_event).ok().map(Bytes::from) }); let event = event.into_static(); let bytes = rmp_serde::to_vec(&event).into_diagnostic()?; batch.insert( &db.jetstream.events, keys::jetstream_event_key(time_us as u64, id), bytes, ); Ok(JetstreamBroadcast { id, time_us, event, ephemeral, })}
fn next_time_us(db: &Db) -> i64 { loop { let last = db.jetstream.last_time_us.load(Ordering::SeqCst); let now = chrono::Utc::now().timestamp_micros(); let next = now.max(last.saturating_add(1)); if db .jetstream .last_time_us .compare_exchange(last, next, Ordering::SeqCst, Ordering::SeqCst) .is_ok() { return next; } }}
#[cfg(all(feature = "indexer_stream", feature = "jetstream"))]pub(crate) fn build_ephemeral_from_stored( did: &str, rev: &str, operation: &str, collection: &str, rkey: &str, data: &crate::types::StoredData, inline_block: Option<&Bytes>, live: bool,) -> Option<JetstreamEphemeral> { use crate::types::StoredData; use jacquard_common::types::cid::{ATP_CID_HASH, IpldCid}; use jacquard_repo::DAG_CBOR_CID_CODEC; use sha2::{Digest, Sha256};
let (cid, record) = match data { StoredData::Ptr(cid) => { if let Some(bytes) = inline_block { match serde_ipld_dagcbor::from_slice::<jacquard_common::RawData>(bytes) { Ok(val) => ( Some(cid.to_string()), Some(Arc::from(serde_json::value::to_raw_value(&val).ok()?)), ), Err(_) => return None, } } else { return None; } } StoredData::Block(block) => { let digest = Sha256::digest(block); let hash = cid::multihash::Multihash::wrap(ATP_CID_HASH, &digest).ok()?; let cid = IpldCid::new_v1(DAG_CBOR_CID_CODEC, hash); match serde_ipld_dagcbor::from_slice::<jacquard_common::RawData>(block) { Ok(val) => ( Some(cid.to_string()), Some(Arc::from(serde_json::value::to_raw_value(&val).ok()?)), ), Err(_) => return None, } } StoredData::Nothing => (None, None), };
Some(JetstreamEphemeral { did: did.to_string(), rev: rev.to_string(), operation: operation.to_string(), collection: collection.to_string(), rkey: rkey.to_string(), record, cid, live, })}