Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272pub mod links;pub mod store;
pub mod api;
use std::ops::Bound;use std::sync::Arc;
use miette::{IntoDiagnostic, Result};use smol_str::SmolStr;use tracing::warn;
use crate::state::AppState;
/// handle for querying the backlinks index.#[derive(Clone)]pub struct BacklinksControl(pub(crate) Arc<AppState>);
pub struct BacklinkEntry { pub uri: SmolStr, pub cid: SmolStr,}
pub struct BacklinksPage { pub backlinks: Vec<BacklinkEntry>, /// raw key bytes of the last returned entry, used as cursor for next page pub next_cursor: Option<Vec<u8>>,}
impl BacklinksControl { /// begin building a paginated fetch of backlinks pointing to `subject`. pub fn fetch(&self, subject: impl Into<String>) -> BacklinksFetch { BacklinksFetch { state: self.0.clone(), subject: subject.into(), collection: None, path: None, dids: Vec::new(), limit: 50, reverse: false, cursor: None, } }
/// begin building a backlinks count query for `subject`. pub fn count(&self, subject: impl Into<String>) -> BacklinksCount { BacklinksCount { state: self.0.clone(), subject: subject.into(), collection: None, path: None, dids: Vec::new(), } }}
/// a paginated fetch against the backlinks reverse index.////// obtain via [`BacklinksControl::fetch`] and execute with [`BacklinksFetch::run`].pub struct BacklinksFetch { state: Arc<AppState>, subject: String, collection: Option<String>, path: Option<String>, dids: Vec<String>, limit: usize, reverse: bool, cursor: Option<Vec<u8>>,}
impl BacklinksFetch { /// filter results to the given source collection NSID. pub fn collection(mut self, collection: impl Into<String>) -> Self { self.collection = Some(collection.into()); self }
/// filter results to specific source author DIDs. pub fn dids(mut self, dids: Vec<String>) -> Self { self.dids = dids; self }
/// filter results to a specific dotted field path within the record (e.g. `.subject.uri`). /// has no effect unless a collection is also set. pub fn path(mut self, path: impl Into<String>) -> Self { self.path = Some(path.into()); self }
/// convenience: parse a `source` parameter of the form `collection` or `collection:path` /// and apply both filters. the path component (if present) has `.` prepended automatically. pub fn source(self, source: &str) -> Self { match source.split_once(':') { Some((col, p)) => self.collection(col).path(format!(".{p}")), None => self.collection(source), } }
/// maximum number of results to return (default 50). pub fn limit(mut self, limit: usize) -> Self { self.limit = limit; self }
/// if `true`, return results in reverse key order (default `false`). pub fn reverse(mut self, reverse: bool) -> Self { self.reverse = reverse; self }
/// resume from the raw key bytes returned as `next_cursor` from a previous page. pub fn cursor(mut self, cursor: Vec<u8>) -> Self { self.cursor = Some(cursor); self }
/// execute and return a page of backlink entries. pub async fn run(self) -> Result<BacklinksPage> { self.state .db .run(move |db| { let scan_prefix = store::reverse_scan_prefix( &self.subject, self.collection.as_deref(), self.path.as_deref(), );
let iter: Box<dyn Iterator<Item = _>> = if !self.reverse { if let Some(cursor_bytes) = &self.cursor { Box::new(db.backlinks.range::<&[u8], _>(( Bound::Excluded(cursor_bytes.as_slice()), Bound::Unbounded, ))) } else { Box::new(db.backlinks.range(scan_prefix.as_slice()..)) } } else { // for reverse scans, bound the end at the cursor (exclusive) or the // prefix end (increment last byte to get an exclusive upper bound) let end: Vec<u8> = if let Some(cursor_bytes) = &self.cursor { cursor_bytes.clone() } else { let mut end = scan_prefix.clone(); if let Some(last) = end.last_mut() { *last = last.saturating_add(1); } end }; Box::new( db.backlinks .range(scan_prefix.as_slice()..end.as_slice()) .rev(), ) };
let mut backlinks = Vec::with_capacity(self.limit); let mut next_cursor = None; let mut last_returned_key = None;
for item in iter { let key = item.key().into_diagnostic()?; if !key.starts_with(scan_prefix.as_slice()) { break; }
let Some((col, _path, did, rkey)) = store::source_from_reverse_key(&key) else { warn!("backlinks: could not extract source from reverse key"); continue; };
if !self.dids.is_empty() && !self.dids.iter().any(|d| d == did) { continue; }
let Some(cid) = store::lookup_cid(&db.indexer, did, col, rkey) else { // record deleted after backlink was written continue; }; if backlinks.len() >= self.limit { next_cursor = last_returned_key; break; } let at_uri = format!("at://{did}/{col}/{rkey}"); backlinks.push(BacklinkEntry { uri: SmolStr::new(at_uri), cid: SmolStr::new(cid), }); last_returned_key = Some(key.to_vec()); }
Ok(BacklinksPage { backlinks, next_cursor, }) }) .await }}
/// a count query against the backlinks reverse index.////// obtain via [`BacklinksControl::count`] and execute with [`BacklinksCount::run`].pub struct BacklinksCount { state: Arc<AppState>, subject: String, collection: Option<String>, path: Option<String>, dids: Vec<String>,}
impl BacklinksCount { /// filter to the given source collection NSID. pub fn collection(mut self, collection: impl Into<String>) -> Self { self.collection = Some(collection.into()); self }
/// filter results to specific source author DIDs. pub fn dids(mut self, dids: Vec<String>) -> Self { self.dids = dids; self }
/// filter to a specific dotted field path within the record (e.g. `.subject.uri`). /// has no effect unless a collection is also set. pub fn path(mut self, path: impl Into<String>) -> Self { self.path = Some(path.into()); self }
/// parse a `source` parameter of the form `collection` or `collection:path` /// and apply both filters. the path component (if present) has `.` prepended automatically. pub fn source(self, source: &str) -> Self { match source.split_once(':') { Some((col, p)) => self.collection(col).path(format!(".{p}")), None => self.collection(source), } }
/// execute and return the total count of matching entries. pub async fn run(self) -> Result<u64> { self.state .db .run(move |db| { let scan_prefix = store::reverse_scan_prefix( &self.subject, self.collection.as_deref(), self.path.as_deref(), );
if !self.dids.is_empty() { let mut count = 0; for item in db.backlinks.prefix(&scan_prefix) { let key = item.key().into_diagnostic()?; let Some((_col, _path, did, _rkey)) = store::source_from_reverse_key(&key) else { continue; }; if self.dids.iter().any(|d| d == did) { count += 1; } } Ok(count) } else { Ok(db.backlinks.prefix(&scan_prefix).count() as u64) } }) .await }}