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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375use std::sync::Arc;use tracing::error;
use miette::{IntoDiagnostic, Result};
use crate::db::filter as db_filter;use crate::filter::FilterMode;use crate::patch::SetUpdate;use crate::state::AppState;
/// a point-in-time snapshot of the filter configuration. returned by all [`FilterControl`] methods.////// because the filter is stored in the database and loaded on demand, this snapshot/// may be stale if another caller modifies the filter concurrently. for the authoritative/// live config use [`FilterControl::get`].#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]pub struct FilterSnapshot { pub mode: FilterMode, pub signals: Vec<String>, pub collections: Vec<String>, pub excludes: Vec<String>,}
/// runtime control over the indexing filter.////// the filter has two orthogonal axes:////// **mode** controls discovery:/// - [`FilterMode::Filter`]: only indexes repos whose firehose commits touch a collection/// matching a configured `signal`. explicit [`ReposControl::track`] always works regardless./// - [`FilterMode::Full`]: indexes the entire network. `signals` are ignored for discovery/// but `collections` and `excludes` still apply.////// **sets** are each independently configurable:/// - `signals`: NSID patterns that trigger auto-discovery in `filter` mode (e.g. `app.bsky.feed.post`, `app.bsky.graph.*`)/// - `collections`: NSID patterns that filter which records are *stored*. empty means store all./// - `excludes`: DIDs that are always skipped regardless of mode.////// NSID patterns support an optional `.*` suffix to match an entire namespace./// all mutations are persisted to the database and take effect immediately.#[derive(Clone)]pub struct FilterControl(pub(super) Arc<AppState>);
impl FilterControl { /// return the current filter configuration from the database. pub async fn get(&self) -> Result<FilterSnapshot> { self.0 .db .run(move |db| { let hot = db_filter::load(&db.filter)?; let excludes = db_filter::read_set(&db.filter, db_filter::EXCLUDE_PREFIX)?; Ok(FilterSnapshot { mode: hot.mode, signals: hot.signals.iter().map(|s| s.to_string()).collect(), collections: hot.collections.iter().map(|s| s.to_string()).collect(), excludes, }) }) .await }
/// set the indexing mode. see [`FilterControl`] for mode semantics. pub fn set_mode(&self, mode: FilterMode) -> FilterPatch { FilterPatch::new(self).set_mode(mode) }
/// replace the entire signals set. existing signals are removed. pub fn set_signals(&self, signals: impl IntoIterator<Item = impl Into<String>>) -> FilterPatch { FilterPatch::new(self).set_signals(signals) }
/// add multiple signals without disturbing existing ones. pub fn append_signals( &self, signals: impl IntoIterator<Item = impl Into<String>>, ) -> FilterPatch { FilterPatch::new(self).append_signals(signals) }
/// add a single signal. no-op if already present. pub fn add_signal(&self, signal: impl Into<String>) -> FilterPatch { FilterPatch::new(self).add_signal(signal) }
/// remove a single signal. no-op if not present. pub fn remove_signal(&self, signal: impl Into<String>) -> FilterPatch { FilterPatch::new(self).remove_signal(signal) }
/// replace the entire collections set. pass an empty iterator to store all collections. pub fn set_collections( &self, collections: impl IntoIterator<Item = impl Into<String>>, ) -> FilterPatch { FilterPatch::new(self).set_collections(collections) }
/// add multiple collections without disturbing existing ones. pub fn append_collections( &self, collections: impl IntoIterator<Item = impl Into<String>>, ) -> FilterPatch { FilterPatch::new(self).append_collections(collections) }
/// add a single collection filter. no-op if already present. pub fn add_collection(&self, collection: impl Into<String>) -> FilterPatch { FilterPatch::new(self).add_collection(collection) }
/// remove a single collection filter. no-op if not present. pub fn remove_collection(&self, collection: impl Into<String>) -> FilterPatch { FilterPatch::new(self).remove_collection(collection) }
/// replace the entire excludes set. pub fn set_excludes( &self, excludes: impl IntoIterator<Item = impl Into<String>>, ) -> FilterPatch { FilterPatch::new(self).set_excludes(excludes) }
/// add multiple DIDs to the excludes set without disturbing existing ones. pub fn append_excludes( &self, excludes: impl IntoIterator<Item = impl Into<String>>, ) -> FilterPatch { FilterPatch::new(self).append_excludes(excludes) }
/// add a single DID to the excludes set. no-op if already excluded. pub fn add_exclude(&self, did: impl Into<String>) -> FilterPatch { FilterPatch::new(self).add_exclude(did) }
/// remove a single DID from the excludes set. no-op if not present. pub fn remove_exclude(&self, did: impl Into<String>) -> FilterPatch { FilterPatch::new(self).remove_exclude(did) }}
/// a staged set of filter mutations. all methods accumulate changes without touching/// the database. call [`FilterPatch::apply`] to commit the entire patch atomically.////// obtain an instance by calling any mutation method on [`FilterControl`], or via/// [`FilterPatch::new`] to start from a blank patch.pub struct FilterPatch { state: Arc<AppState>, /// if set, replaces the current indexing mode. pub mode: Option<FilterMode>, /// if set, replaces or patches the signals set. pub(crate) signals: Option<SetUpdate>, /// if set, replaces or patches the collections set. pub(crate) collections: Option<SetUpdate>, /// if set, replaces or patches the excludes set. pub(crate) excludes: Option<SetUpdate>,}
impl FilterPatch { /// create a new blank patch associated with the given [`FilterControl`]. pub fn new(control: &FilterControl) -> Self { Self { state: control.0.clone(), mode: None, signals: None, collections: None, excludes: None, } }
/// set the indexing mode. see [`FilterControl`] for mode semantics. pub fn set_mode(mut self, mode: FilterMode) -> Self { self.mode = Some(mode); self }
/// replace the entire signals set. existing signals are removed. pub fn set_signals(mut self, signals: impl IntoIterator<Item = impl Into<String>>) -> Self { self.signals = Some(SetUpdate::Set( signals.into_iter().map(Into::into).collect(), )); self }
/// add multiple signals without disturbing existing ones. pub fn append_signals(mut self, signals: impl IntoIterator<Item = impl Into<String>>) -> Self { merge_set_update_append(&mut self.signals, signals.into_iter().map(Into::into)); self }
/// add a single signal. no-op if already present. pub fn add_signal(mut self, signal: impl Into<String>) -> Self { merge_set_update_add(&mut self.signals, signal.into()); self }
/// remove a single signal. no-op if not present. pub fn remove_signal(mut self, signal: impl Into<String>) -> Self { merge_set_update_remove(&mut self.signals, signal.into()); self }
/// replace the entire collections set. pass an empty iterator to store all collections. pub fn set_collections( mut self, collections: impl IntoIterator<Item = impl Into<String>>, ) -> Self { self.collections = Some(SetUpdate::Set( collections.into_iter().map(Into::into).collect(), )); self }
/// add multiple collections without disturbing existing ones. pub fn append_collections( mut self, collections: impl IntoIterator<Item = impl Into<String>>, ) -> Self { merge_set_update_append( &mut self.collections, collections.into_iter().map(Into::into), ); self }
/// add a single collection filter. no-op if already present. pub fn add_collection(mut self, collection: impl Into<String>) -> Self { merge_set_update_add(&mut self.collections, collection.into()); self }
/// remove a single collection filter. no-op if not present. pub fn remove_collection(mut self, collection: impl Into<String>) -> Self { merge_set_update_remove(&mut self.collections, collection.into()); self }
/// replace the entire excludes set. pub fn set_excludes(mut self, excludes: impl IntoIterator<Item = impl Into<String>>) -> Self { self.excludes = Some(SetUpdate::Set( excludes.into_iter().map(Into::into).collect(), )); self }
/// add multiple DIDs to the excludes set without disturbing existing ones. pub fn append_excludes( mut self, excludes: impl IntoIterator<Item = impl Into<String>>, ) -> Self { merge_set_update_append(&mut self.excludes, excludes.into_iter().map(Into::into)); self }
/// add a single DID to the excludes set. no-op if already excluded. pub fn add_exclude(mut self, did: impl Into<String>) -> Self { merge_set_update_add(&mut self.excludes, did.into()); self }
/// remove a single DID from the excludes set. no-op if not present. pub fn remove_exclude(mut self, did: impl Into<String>) -> Self { merge_set_update_remove(&mut self.excludes, did.into()); self }
/// commit the patch atomically to the database and update the in-memory filter. /// returns the updated [`FilterSnapshot`]. pub async fn apply(self) -> Result<FilterSnapshot> { let filter_handle = self.state.filter.clone(); let state = self.state.clone(); let mode = self.mode; let signals = self.signals; let collections = self.collections; let excludes = self.excludes;
let new_filter = 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()?; db.persist()?; db_filter::load(&db.filter) }) .await .map_err(|e| { error!(err = %e, "failed to apply filter patch"); e })?;
let exclude_list = state .db .run(move |db| db_filter::read_set(&db.filter, db_filter::EXCLUDE_PREFIX)) .await?;
let snapshot = FilterSnapshot { mode: new_filter.mode, signals: new_filter.signals.iter().map(|s| s.to_string()).collect(), collections: new_filter .collections .iter() .map(|s| s.to_string()) .collect(), excludes: exclude_list, };
filter_handle.store(Arc::new(new_filter)); Ok(snapshot) }}
fn merge_set_update_append(opt: &mut Option<SetUpdate>, items: impl IntoIterator<Item = String>) { let new_items: Vec<String> = items.into_iter().collect(); if new_items.is_empty() { return; } match opt { Some(SetUpdate::Set(vec)) => { for item in new_items { if !vec.contains(&item) { vec.push(item); } } } Some(SetUpdate::Patch(map)) => { for item in new_items { map.insert(item, true); } } None => { let map = new_items.into_iter().map(|item| (item, true)).collect(); *opt = Some(SetUpdate::Patch(map)); } }}
fn merge_set_update_add(opt: &mut Option<SetUpdate>, item: String) { match opt { Some(SetUpdate::Set(vec)) => { if !vec.contains(&item) { vec.push(item); } } Some(SetUpdate::Patch(map)) => { map.insert(item, true); } None => { *opt = Some(SetUpdate::Patch([(item, true)].into())); } }}
fn merge_set_update_remove(opt: &mut Option<SetUpdate>, item: String) { match opt { Some(SetUpdate::Set(vec)) => { vec.retain(|x| x != &item); } Some(SetUpdate::Patch(map)) => { map.insert(item, false); } None => { *opt = Some(SetUpdate::Patch([(item, false)].into())); } }}