use super::*;
#[cfg(feature = "indexer_stream")]
/// a stream of [`Event`]s. returned by [`Hydrant::subscribe`].
///
/// implements [`futures::Stream`] and can be used with `StreamExt::next`,
/// `while let Some(item) = stream.next().await`, `forward`, etc.
/// the stream terminates when hydrant shuts down, or when it can't read the next
/// events from the database: resubscribe with the last event's id as the cursor.
pub struct EventStream {
receiver: mpsc::Receiver<Result<Event, StreamError>>,
}
/// the `/stream` frames for [`Hydrant::subscribe_frames`]
#[cfg(feature = "indexer_stream")]
pub(crate) struct FrameStream {
receiver: mpsc::Receiver<Result<bytes::Bytes, StreamError>>,
}
#[cfg(feature = "indexer_stream")]
impl Stream for FrameStream {
type Item = Result<bytes::Bytes, StreamError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.receiver.poll_recv(cx)
}
}
#[cfg(feature = "indexer_stream")]
#[derive(Debug, Clone, thiserror::Error)]
pub enum StreamError {
#[error("stream consumer too slow: {reason}")]
ConsumerTooSlow { reason: String },
}
#[cfg(feature = "indexer_stream")]
impl StreamError {
pub fn code(&self) -> &'static str {
match self {
Self::ConsumerTooSlow { .. } => "ConsumerTooSlow",
}
}
}
#[cfg(feature = "indexer_stream")]
impl Stream for EventStream {
type Item = Result<Event, StreamError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.receiver.poll_recv(cx)
}
}
/// runtime control over the backfill worker component.
///
/// the backfill worker fetches full repo CAR files from each repo's PDS for any
/// repository in the pending queue, parses the MST, and inserts all matching records
/// into the database. concurrency is bounded by `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`.
#[derive(Clone)]
pub struct BackfillHandle(Arc<AppState>);
impl BackfillHandle {
pub(crate) fn new(state: Arc<AppState>) -> Self {
Self(state)
}
/// enable the backfill worker, no-op if already enabled.
pub fn enable(&self) {
self.0.backfill_enabled.send_replace(true);
}
/// disable the backfill worker, in-flight repos complete before pausing.
pub fn disable(&self) {
self.0.backfill_enabled.send_replace(false);
}
/// returns the current enabled state of the backfill worker.
pub fn is_enabled(&self) -> bool {
*self.0.backfill_enabled.borrow()
}
}
#[cfg(feature = "indexer_stream")]
impl Hydrant {
/// Return the highest committed durable indexer-stream event ID.
///
/// Consumers can snapshot this before subscribing and report readiness only
/// after their committed cursor reaches the snapshot.
pub fn stream_head(&self) -> miette::Result<Option<u64>> {
self.state.db.stream.event_head()
}
/// cursors and chunks only mean something on the stream they came from
pub fn stream_id(&self) -> &str {
self.state.db.stream.id()
}
pub(crate) fn stream_chunks(&self) -> Option<&crate::db::stream_chunks::StreamChunks> {
self.state.db.stream.chunks()
}
/// subscribe to the ordered event stream.
///
/// returns an [`EventStream`] that implements [`futures::Stream`].
///
/// - if `cursor` is `None`, streaming starts from the current head (live tail only).
/// - if `cursor` is `Some(id)`, all persisted `record` events from that ID onward are
/// replayed first, then the stream will switch to live tailing.
///
/// `identity` and `account` events are ephemeral and are never replayed from a cursor,
/// only live ones are delivered. use [`ReposControl::info`] to fetch current state for
/// a specific repository.
///
/// multiple concurrent subscribers each receive a full independent copy of the stream.
/// the stream ends when the `EventStream` is dropped. a subscriber that falls behind
/// the live broadcast catches up from the database, skipping only the ephemeral
/// events it fell behind on. one that stops reading for `HYDRANT_STREAM_SEND_TIMEOUT`
/// receives [`StreamError::ConsumerTooSlow`] after the events already queued, then the
/// stream ends.
///
/// the replay runs as fast as you read it. only `/stream` readers on the network are capped,
/// when `HYDRANT_THROTTLED_STREAMS` lists it.
pub fn subscribe(&self, cursor: Option<u64>) -> EventStream {
EventStream {
receiver: self
.spawn_stream::<Decoded>(cursor, stream::StreamOptions::from_config(&self.config)),
}
}
/// [`Self::subscribe`] as `/stream` sends it to a reader on the network: each event
/// already its frame, and the replay capped
pub(crate) fn subscribe_frames(&self, cursor: Option<u64>) -> FrameStream {
let opts = stream::StreamOptions::network(&self.config, crate::config::StreamKind::Indexer);
FrameStream {
receiver: self.spawn_stream::<Frame>(cursor, opts),
}
}
fn spawn_stream<F: stream::indexer::Form>(
&self,
cursor: Option<u64>,
opts: stream::StreamOptions,
) -> mpsc::Receiver<Result<F::Output, StreamError>> {
let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY);
let state = self.state.clone();
let runtime = tokio::runtime::Handle::current();
std::thread::Builder::new()
.name("hydrant-stream".into())
.spawn(move || {
let _g = runtime.enter();
event_stream_thread::<F>(state, tx, cursor, opts);
})
.expect("failed to spawn stream thread");
rx
}
#[cfg(feature = "indexer_stream")]
#[doc(hidden)]
pub fn seed_events_for_bench(&self, count: usize) -> usize {
use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid};
use crate::types::{StoredData, StoredEvent};
use jacquard_common::types::did::Did;
use jacquard_common::{CowStr, IntoStatic};
let db = &self.state.db;
let mut txn = crate::db::Txn::new(db);
let mut total_bytes = 0usize;
let did_str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let did: Did = Did::new_owned(did_str).expect("valid plc did");
let trimmed = TrimmedDid::from(&did).into_static();
let rev = DbTid::new_from_bytes([0u8; 8]);
let collection = CowStr::Borrowed("app.bsky.feed.post").into_static();
let block = bytes::Bytes::from(
serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": "app.bsky.feed.post",
"text": "seeded event",
"createdAt": "2026-01-01T00:00:00Z"
}))
.expect("fixture record must encode as DAG-CBOR"),
);
for i in 0..count {
let rkey = DbRkey::Str(smol_str::format_smolstr!("r{i}"));
let cid = crate::car::compute_cid(block.as_ref());
db.indexer.stage_head(
&mut txn.batch,
crate::db::keys::record_key(&did, &collection, &rkey),
block.as_ref(),
);
let evt = StoredEvent {
live: false,
did: trimmed.clone(),
rev,
collection: collection.clone(),
rkey,
action: DbAction::Create,
data: StoredData::Ptr(cid),
};
let bytes = rmp_serde::to_vec(&evt).expect("msgpack serialization cannot fail");
total_bytes += bytes.len();
txn.outbox.stream.push_record(bytes, None);
}
txn.commit().expect("failed to commit events batch");
total_bytes
}
}