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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360use crate::api::AppState;#[cfg(feature = "indexer")]use crate::db::keys;use crate::db::registry;use axum::routing::{get, post};use axum::{ Json, extract::{Query, State}, http::StatusCode,};#[cfg(feature = "indexer")]use jacquard_common::types::ident::AtIdentifier;use miette::IntoDiagnostic;use serde::{Deserialize, Serialize};use serde_json::Value;use std::sync::Arc;
#[cfg(feature = "indexer")]#[derive(Deserialize)]pub struct DebugCountRequest { pub did: String, pub collection: String,}
#[cfg(feature = "indexer")]#[derive(Serialize)]pub struct DebugCountResponse { pub count: usize,}
pub fn router() -> axum::Router<Arc<AppState>> { let r = axum::Router::new() .route("/debug/get", get(handle_debug_get)) .route("/debug/iter", get(handle_debug_iter)) .route("/debug/compact", post(handle_debug_compact));
#[cfg(any(feature = "indexer_stream", feature = "relay"))] let r = r .route( "/debug/ephemeral_ttl_tick", post(handle_debug_ephemeral_ttl_tick), ) .route("/debug/seed_watermark", post(handle_debug_seed_watermark)) .route("/debug/seed_events", post(handle_debug_seed_events));
#[cfg(feature = "indexer")] let r = r.route("/debug/count", get(handle_debug_count));
r}
#[cfg(feature = "indexer")]pub async fn handle_debug_count( State(state): State<Arc<AppState>>, Query(req): Query<DebugCountRequest>,) -> Result<Json<DebugCountResponse>, StatusCode> { let did = state .resolver .resolve_did(&AtIdentifier::new(req.did.as_str()).map_err(|_| StatusCode::BAD_REQUEST)?) .await .map_err(|_| StatusCode::BAD_REQUEST)?;
let prefix = keys::record_prefix_collection(&did, &req.collection);
let count = state .db .run(move |db| { let ks = db .keyspace_by_name("records") .expect("records keyspace exists in indexer mode");
let start_key = prefix.clone(); // rkeys cannot contain 0xff, so every key in the collection sorts // below `prefix ff` let mut end_key = prefix.clone(); end_key.push(0xff);
Ok::<_, miette::Report>(ks.range(start_key..end_key).count()) }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(DebugCountResponse { count }))}
#[derive(Deserialize)]pub struct DebugGetRequest { pub partition: String, pub key: String,}
#[derive(Serialize)]pub struct DebugGetResponse { pub value: Option<Value>,}
pub async fn handle_debug_get( State(state): State<Arc<AppState>>, Query(req): Query<DebugGetRequest>,) -> Result<Json<DebugGetResponse>, StatusCode> { let ks = get_keyspace_by_name(&state.db, &req.partition)?;
let key = registry::debug_parse_key(&req.partition, &req.key).ok_or(StatusCode::BAD_REQUEST)?;
let partition = req.partition.clone(); let value = state .db .run(move |_| { ks.get(key) .inspect_err(crate::db::check_poisoned) .into_diagnostic() }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .map(|v| registry::debug_value(&partition, &v));
Ok(Json(DebugGetResponse { value }))}
#[derive(Deserialize)]pub struct DebugIterRequest { pub partition: String, pub start: Option<String>, pub end: Option<String>, pub limit: Option<usize>, pub reverse: Option<bool>,}
#[derive(Serialize)]pub struct DebugIterResponse { pub items: Vec<(String, Value)>,}
pub async fn handle_debug_iter( State(state): State<Arc<AppState>>, Query(req): Query<DebugIterRequest>,) -> Result<Json<DebugIterResponse>, StatusCode> { let partition = req.partition.clone();
let parse_bound = |s: Option<String>| -> Result<Option<Vec<u8>>, StatusCode> { s.map(|s| registry::debug_parse_key(&partition, &s).ok_or(StatusCode::BAD_REQUEST)) .transpose() };
let start = parse_bound(req.start)?; let end = parse_bound(req.end)?;
let items = state .db .run(move |db| { let ks = get_keyspace_by_name(db, &req.partition) .map_err(|_| miette::miette!("bad request"))?; let limit = req.limit.unwrap_or(50);
let collect = |iter: &mut dyn Iterator<Item = fjall::Guard>| { let mut items = Vec::new(); for guard in iter.take(limit) { let (k, v) = guard .into_inner() .map_err(|_| miette::miette!("internal error"))?;
let key_str = registry::debug_render_key(&partition, &k);
items.push((key_str, registry::debug_value(&partition, &v))); } Ok::<_, miette::Report>(items) };
let start_bound = if let Some(s) = &start { std::ops::Bound::Included(s.as_slice()) } else { std::ops::Bound::Unbounded };
let end_bound = if let Some(e) = &end { std::ops::Bound::Included(e.as_slice()) } else { std::ops::Bound::Unbounded };
if req.reverse == Some(true) { collect( &mut ks .range::<&[u8], (std::ops::Bound<&[u8]>, std::ops::Bound<&[u8]>)>(( start_bound, end_bound, )) .rev(), ) } else { collect( &mut ks.range::<&[u8], (std::ops::Bound<&[u8]>, std::ops::Bound<&[u8]>)>(( start_bound, end_bound, )), ) } }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(DebugIterResponse { items }))}
fn get_keyspace_by_name(db: &crate::db::Db, name: &str) -> Result<fjall::Keyspace, StatusCode> { db.keyspace_by_name(name).ok_or(StatusCode::BAD_REQUEST)}
#[derive(Deserialize)]pub struct DebugCompactRequest { pub partition: String,}
pub async fn handle_debug_compact( State(state): State<Arc<AppState>>, Query(req): Query<DebugCompactRequest>,) -> Result<StatusCode, StatusCode> { state .db .run(move |db| { let ks = get_keyspace_by_name(db, &req.partition) .map_err(|_| miette::miette!("bad request"))?; ks.remove(b"dummy_tombstone123").into_diagnostic()?; db.inner .persist(fjall::PersistMode::Buffer) .into_diagnostic()?; ks.rotate_memtable_and_wait().into_diagnostic()?; ks.major_compact().into_diagnostic()?; Ok::<_, miette::Report>(()) }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]pub async fn handle_debug_ephemeral_ttl_tick( State(state): State<Arc<AppState>>,) -> Result<StatusCode, StatusCode> { let ephemeral_ttl = state.ephemeral_ttl; state .db .run(move |db| { #[cfg(feature = "indexer_stream")] crate::db::ephemeral::ephemeral_ttl_tick(db, &ephemeral_ttl)?; #[cfg(feature = "relay")] crate::db::ephemeral::relay_events_ttl_tick(db, &ephemeral_ttl)?; #[cfg(feature = "jetstream")] crate::db::ephemeral::jetstream_events_ttl_tick(db, &ephemeral_ttl)?; Ok::<_, miette::Report>(()) }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)}
#[derive(Deserialize)]#[cfg(any(feature = "indexer_stream", feature = "relay"))]pub struct DebugSeedWatermarkRequest { /// unix timestamp (seconds) to write the watermark at pub ts: u64, /// event_id the watermark points to, all events before this id will be pruned pub event_id: u64,}
/// writes an event watermark entry directly to the cursors keyspace, using identical/// key/value encoding to the real TTL worker. used in tests to plant a past watermark/// so the real `ephemeral_ttl_tick` code path is exercised without waiting 3600 seconds.#[cfg(any(feature = "indexer_stream", feature = "relay"))]pub async fn handle_debug_seed_watermark( State(state): State<Arc<AppState>>, Query(req): Query<DebugSeedWatermarkRequest>,) -> Result<StatusCode, StatusCode> { state .db .run(move |db| { #[cfg(feature = "indexer_stream")] db.cursors .insert( crate::db::keys::event_watermark_key(req.ts), req.event_id.to_be_bytes(), ) .into_diagnostic()?; #[cfg(feature = "relay")] db.cursors .insert( crate::db::keys::relay_event_watermark_key(req.ts), req.event_id.to_be_bytes(), ) .into_diagnostic()?; Ok::<_, miette::Report>(()) }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)}
#[derive(Deserialize)]#[cfg(any(feature = "indexer_stream", feature = "relay"))]pub struct DebugSeedEventsRequest { pub partition: String, pub count: u64,}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]pub async fn handle_debug_seed_events( State(state): State<Arc<AppState>>, Query(req): Query<DebugSeedEventsRequest>,) -> Result<StatusCode, StatusCode> { if req.partition != "events" && req.partition != "relay_events" { return Err(StatusCode::BAD_REQUEST); }
state .db .run(move |db| { let mut batch = db.inner.batch(); if req.partition == "events" { #[cfg(feature = "indexer_stream")] { for _ in 0..req.count { let seq = db .stream .next_event_id .fetch_add(1, std::sync::atomic::Ordering::SeqCst); db.stream.stage_event( &mut batch, crate::db::keys::event_key(seq), b"dummy", ); } } } else if req.partition == "relay_events" { #[cfg(feature = "relay")] { for _ in 0..req.count { let seq = db .relay .next_seq .fetch_add(1, std::sync::atomic::Ordering::SeqCst); batch.insert( &db.relay.events, crate::db::keys::relay_event_key(seq), b"dummy", ); } } } batch.commit().into_diagnostic()?; Ok::<_, miette::Report>(()) }) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::OK)}