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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395use futures::FutureExt;use miette::{IntoDiagnostic, Result};use std::future::Future;use std::sync::Arc;use std::sync::atomic::{AtomicBool, Ordering};use tokio::sync::watch;use tracing::{debug, error, info, warn};use url::Url;
use crate::config::{Config, SignatureVerification};use crate::db::{self, filter as db_filter, load_persisted_firehose_sources};use crate::filter::FilterMode;use crate::state::AppState;
use super::firehose::{FirehoseHandle, FirehoseShared};use super::{DbControl, FilterControl, ReposControl, pds::PdsControl};
#[cfg(feature = "indexer")]use super::{BackfillHandle, crawler};#[cfg(feature = "indexer")]use crate::backfill::BackfillWorker;#[cfg(feature = "indexer")]use crate::db::load_persisted_crawler_sources;#[cfg(feature = "indexer")]use crate::ingest::indexer::FirehoseWorker;
#[cfg(feature = "backlinks")]use crate::backlinks::BacklinksControl;
use super::seed;
pub(crate) mod run;
/// the top-level handle to a hydrant instance.////// `Hydrant` is cheaply cloneable. all sub-handles share the same underlying state./// construct it via [`Hydrant::new`] or [`Hydrant::from_env`], configure the filter/// and repos as needed, then call [`Hydrant::run`] to start all background components.#[derive(Clone)]pub struct Hydrant { #[cfg(feature = "indexer")] pub crawler: crawler::CrawlerHandle, pub firehose: FirehoseHandle, #[cfg(feature = "indexer")] pub backfill: BackfillHandle, pub filter: FilterControl, pub pds: PdsControl, pub repos: ReposControl, pub db: DbControl, #[cfg(feature = "backlinks")] pub backlinks: crate::backlinks::BacklinksControl, pub(crate) state: Arc<AppState>, pub(crate) config: Arc<Config>, pub(crate) started: Arc<AtomicBool>, _priv: (),}
impl Hydrant { /// Returns a reference to the internal DID resolver. pub fn resolver(&self) -> &crate::resolver::Resolver { &self.state.resolver }
#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] pub(crate) fn stream_send_timeout(&self) -> std::time::Duration { self.config.stream_send_timeout }
/// open the database and configure hydrant from `config`. /// /// this sets up the database, applies any filter configuration from `config`, and /// initializes all sub-handles. no background tasks are started yet: call /// [`run`](Self::run) to start all components and drive the instance. pub async fn new(config: Config) -> Result<Self> { info!("{config}");
#[cfg(feature = "relay")] if config.only_index_links { miette::bail!("HYDRANT_ONLY_INDEX_LINKS is not supported in relay mode"); }
// 1. open database and construct AppState let state = AppState::new(&config)?;
// 2. apply any filter config from env variables if config.full_network || config.filter_signals.is_some() || config.filter_collections.is_some() || config.filter_excludes.is_some() { let mode = config.full_network.then_some(FilterMode::Full); let signals = config .filter_signals .clone() .map(crate::patch::SetUpdate::Set); let collections = config .filter_collections .clone() .map(crate::patch::SetUpdate::Set); let excludes = config .filter_excludes .clone() .map(crate::patch::SetUpdate::Set);
state .db .run(move |db| { let mut batch = db.inner.batch(); db_filter::apply_patch( &mut batch, &db.filter, mode, signals, collections, excludes, )?; batch.commit().into_diagnostic() }) .await?;
// 3. reload the live filter into the hot-path arc-swap let new_filter = state.db.run(move |db| db_filter::load(&db.filter)).await?; state.filter.store(Arc::new(new_filter)); }
#[cfg(feature = "indexer")] { // 4. set crawler enabled state from config, evaluated against the post-patch filter let post_patch_crawler = match config.enable_crawler { Some(b) => b, None => { state.filter.load().mode == FilterMode::Full || !config.crawler_sources.is_empty() } }; state.crawler_enabled.send_replace(post_patch_crawler); }
let state = Arc::new(state);
Ok(Self { firehose: FirehoseHandle::new(state.clone()), filter: FilterControl(state.clone()), pds: PdsControl(state.clone()), repos: ReposControl(state.clone()), db: DbControl(state.clone()), #[cfg(feature = "indexer")] crawler: crawler::CrawlerHandle { state: state.clone(), shared: Arc::new(std::sync::OnceLock::new()), tasks: Arc::new(scc::HashMap::new()), persisted: Arc::new(scc::HashSet::new()), }, #[cfg(feature = "indexer")] backfill: BackfillHandle::new(state.clone()), #[cfg(feature = "backlinks")] backlinks: crate::backlinks::BacklinksControl(state.clone()), state, config: Arc::new(config), started: Arc::new(AtomicBool::new(false)), _priv: (), }) }
/// reads config from environment variables and calls [`Hydrant::new`]. pub async fn from_env() -> Result<Self> { Self::new(Config::from_env()?).await }
/// returns a future that runs the HTTP management API server on the given bind addresses. /// /// the server exposes all management endpoints (`/filter`, `/repos`, `/ingestion`, /// `/stream`, `/stats`, `/db/*`, `/xrpc/*`). it runs indefinitely and resolves /// only on error. /// /// intended for `tokio::spawn` or inclusion in a `select!` / task list. the clone /// of `self` is deferred until the future is first polled. /// /// to disable the HTTP API entirely, simply don't call this method. pub fn serve(&self, binds: super::ApiBinds) -> impl Future<Output = Result<()>> { let hydrant = self.clone(); async move { crate::api::serve(hydrant, binds).await } }
/// returns a future that runs the debug HTTP API server on `127.0.0.1:{port}`. /// /// exposes internal inspection endpoints (`/debug/get`, `/debug/iter`, etc.). /// binds only to loopback. pub fn serve_debug(&self, port: u16) -> impl Future<Output = Result<()>> { let state = self.state.clone(); async move { crate::api::serve_debug(state, port).await } }
/// iterate sequentially over every block stored for `collection`, ordered by CID. /// /// values are raw DAG-CBOR. deleted or updated records may leave orphan blocks /// behind since the CAS is content-addressed, not record-referenced. #[cfg(feature = "indexer")] pub fn scan_collection_blocks( &self, collection: &str, ) -> impl Iterator<Item = Result<ScannedBlock>> + 'static { let db = self.state.db.indexer.clone(); let collection = collection.to_string();
let mut start = Vec::with_capacity(collection.len() + 1); start.extend_from_slice(collection.as_bytes()); start.push(b'|');
let mut end = Vec::with_capacity(collection.len() + 1); end.extend_from_slice(collection.as_bytes()); end.push(b'|' + 1);
let iter = db.block_range(( std::ops::Bound::Included(start), std::ops::Bound::Excluded(end), ));
iter.map(|item| { let (key, value) = item.into_inner().into_diagnostic()?; Ok(ScannedBlock { key, value }) }) }}
#[cfg(feature = "indexer")]/// a raw block entry yielded by [`Hydrant::scan_collection_blocks`].////// key and value are kept as raw slices; use the accessor methods for lazy parsing.#[derive(Clone, Debug)]pub struct ScannedBlock { pub(crate) key: fjall::Slice, pub(crate) value: fjall::Slice,}
#[cfg(feature = "indexer")]impl ScannedBlock { /// the collection name part of the block's key. pub fn collection(&self) -> &str { let sep_idx = self .key .iter() .position(|&b| b == b'|') .expect("malformed block key: missing separator"); std::str::from_utf8(&self.key[..sep_idx]).expect("collection name must be valid UTF-8") }
/// the raw CID bytes after the separator. pub fn cid(&self) -> &[u8] { let sep_idx = self .key .iter() .position(|&b| b == b'|') .expect("malformed block key: missing separator"); &self.key[sep_idx + 1..] }
/// the key's CID, parsed. pub fn parsed_cid(&self) -> Result<cid::Cid, miette::Report> { cid::Cid::read_bytes(self.cid()).map_err(|e| miette::miette!("failed to parse CID: {e}")) }
/// the key's CID as a jacquard CID. pub fn jacquard_cid( &self, ) -> Result<jacquard_common::types::cid::Cid<'static>, miette::Report> { let parsed = cid::Cid::read_bytes(self.cid()) .map_err(|e| miette::miette!("failed to parse CID: {e}"))?; Ok(jacquard_common::types::cid::Cid::ipld(parsed)) }
/// the raw DAG-CBOR block bytes. pub fn value(&self) -> &[u8] { &self.value }}
impl axum::extract::FromRef<Hydrant> for Arc<AppState> { fn from_ref(h: &Hydrant) -> Self { h.state.clone() }}
#[cfg(all(test, feature = "indexer"))]mod tests { use super::*; use crate::config::Config; use crate::db::keys; use cid::Cid; use cid::multihash::Multihash; use tempfile::tempdir;
const SHA2_256: u64 = 0x12; const DAG_CBOR: u64 = 0x71;
fn test_config(path: &std::path::Path) -> Config { Config { database_path: path.to_path_buf(), ..Default::default() } }
#[tokio::test] async fn test_scan_collection_blocks() -> Result<()> { let tmp = tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(tmp.path())).await?; let state = hydrant.state.clone();
let hash1 = [1u8; 32]; let mh1 = Multihash::<64>::wrap(SHA2_256, &hash1).unwrap(); let valid_cid1 = Cid::new_v1(DAG_CBOR, mh1); let cid_bytes1 = valid_cid1.to_bytes();
let hash2 = [2u8; 32]; let mh2 = Multihash::<64>::wrap(SHA2_256, &hash2).unwrap(); let valid_cid2 = Cid::new_v1(DAG_CBOR, mh2); let cid_bytes2 = valid_cid2.to_bytes();
let hash3 = [3u8; 32]; let mh3 = Multihash::<64>::wrap(SHA2_256, &hash3).unwrap(); let valid_cid3 = Cid::new_v1(DAG_CBOR, mh3); let cid_bytes3 = valid_cid3.to_bytes();
let cid_bytes1_val = cid_bytes1.clone(); let cid_bytes2_val = cid_bytes2.clone(); let cid_bytes3_val = cid_bytes3.clone();
// Insert a couple of blocks for two collections: "app.bsky.feed.post" and "app.bsky.feed.like" state .db .run(move |db| -> Result<()> { let mut batch = db.inner.batch();
let post_key = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes1_val); let post_val = b"post_cbor_value"; db.indexer .stage_block(&mut batch, post_key, post_val.to_vec());
let like_key = keys::indexer::block_key("app.bsky.feed.like", &cid_bytes2_val); let like_val = b"like_cbor_value"; db.indexer .stage_block(&mut batch, like_key, like_val.to_vec());
// Let's add one more post to test multiple blocks in same collection let post_key2 = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes3_val); let post_val2 = b"post_cbor_value_2"; db.indexer .stage_block(&mut batch, post_key2, post_val2.to_vec());
batch.commit().into_diagnostic()?; Ok(()) }) .await?;
// Scan posts let post_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.post") .collect::<Result<Vec<_>>>()?;
assert_eq!(post_blocks.len(), 2); assert_eq!(post_blocks[0].collection(), "app.bsky.feed.post"); assert_eq!(post_blocks[0].cid(), cid_bytes1); assert_eq!(post_blocks[0].parsed_cid().unwrap(), valid_cid1); assert_eq!( post_blocks[0].jacquard_cid().unwrap(), jacquard_common::types::cid::Cid::ipld(valid_cid1) ); assert_eq!(post_blocks[0].value(), b"post_cbor_value");
assert_eq!(post_blocks[1].collection(), "app.bsky.feed.post"); assert_eq!(post_blocks[1].cid(), cid_bytes3); assert_eq!(post_blocks[1].parsed_cid().unwrap(), valid_cid3); assert_eq!(post_blocks[1].value(), b"post_cbor_value_2");
// Scan likes let like_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.like") .collect::<Result<Vec<_>>>()?;
assert_eq!(like_blocks.len(), 1); assert_eq!(like_blocks[0].collection(), "app.bsky.feed.like"); assert_eq!(like_blocks[0].cid(), cid_bytes2); assert_eq!(like_blocks[0].parsed_cid().unwrap(), valid_cid2); assert_eq!(like_blocks[0].value(), b"like_cbor_value");
// Scan non-existent let empty_blocks: Vec<ScannedBlock> = hydrant .scan_collection_blocks("app.bsky.feed.repost") .collect::<Result<Vec<_>>>()?; assert!(empty_blocks.is_empty());
Ok(()) }}