+2

Cargo.toml

reviewed

··· 4 4 edition = "2024" 5 5 6 6 [features] 7 7 + default = ["backlinks"] 7 8 sync_all = [] 9 9 + backlinks = [] 8 10 9 11 [dependencies] 10 12 tokio = { version = "1.0", features = ["full"] }

+33

README.md

reviewed

··· 168 168 169 169 returns `{ count }`. 170 170 171 171 + #### blue.microcosm.links.* 172 172 + 173 173 + hydrant implements a subset of [microcosm constellation](https://constellation.microcosm.blue/) when it's built with the `backlinks` cargo feature (`cargo build --features backlinks`). 174 174 + 175 175 + when enabled, hydrant indexes all AT URI and DID references found inside stored records into a reverse index. this lets you efficiently answer "what records link to this subject?". 176 176 + 177 177 + ##### blue.microcosm.links.getBacklinks 178 178 + 179 179 + return records that link to a given subject. 180 180 + 181 181 + | param | required | description | 182 182 + | :--- | :--- | :--- | 183 183 + | `subject` | yes | AT URI or DID to look up backlinks for. | 184 184 + | `source` | no | filter by source collection, e.g. `app.bsky.feed.like`. also accepts `collection:path` form to further filter by field path, e.g. `app.bsky.feed.like:subject.uri`. the path is matched against the dotted field path within the record (`.` is prepended automatically). | 185 185 + | `limit` | no | max results to return (default 50, max 100). | 186 186 + | `cursor` | no | opaque pagination cursor from a previous response. | 187 187 + | `reverse` | no | if `true`, return results in reverse order (default `false`). | 188 188 + 189 189 + returns `{ backlinks: [{ uri, cid }], cursor? }`. 190 190 + 191 191 + results are ordered by source record rkey (ascending by default, descending when `reverse=true`). the cursor is stable across new insertions for TID rkey records. 192 192 + 193 193 + ##### blue.microcosm.links.getBacklinksCount 194 194 + 195 195 + return the number of records that link to a given subject. 196 196 + 197 197 + | param | required | description | 198 198 + | :--- | :--- | :--- | 199 199 + | `subject` | yes | AT URI or DID to count backlinks for. | 200 200 + | `source` | no | filter by source collection (same format as `getBacklinks`). | 201 201 + 202 202 + returns `{ count }`. 203 203 + 171 204 ### event stream 172 205 173 206 - `GET /stream`: subscribe to the event stream.

+10 -2

src/api/mod.rs

reviewed

··· 15 15 mod xrpc; 16 16 17 17 pub async fn serve(hydrant: Hydrant, port: u16) -> miette::Result<()> { 18 18 - let app = Router::new() 18 18 + #[allow(unused_mut)] 19 19 + let mut app = Router::new() 19 20 .route("/health", get(|| async { "OK" })) 20 21 .route("/stats", get(stats::get_stats)) 21 22 .nest("/stream", stream::router()) ··· 23 24 .merge(filter::router()) 24 25 .merge(repos::router()) 25 26 .merge(ingestion::router()) 26 26 - .merge(db::router()) 27 27 + .merge(db::router()); 28 28 + 29 29 + #[cfg(feature = "backlinks")] 30 30 + { 31 31 + app = app.merge(crate::backlinks::api::router()); 32 32 + } 33 33 + 34 34 + let app = app 27 35 .with_state(hydrant) 28 36 .layer(TraceLayer::new_for_http()) 29 37 .layer(CorsLayer::permissive());

+19

src/backfill/mod.rs

reviewed

··· 627 627 if !ephemeral { 628 628 batch.insert(&app_state.db.blocks, block_key.clone(), val.as_ref()); 629 629 batch.insert(&app_state.db.records, db_key, cid_raw); 630 630 + #[cfg(feature = "backlinks")] 631 631 + if let Ok(value) = serde_ipld_dagcbor::from_slice::<jacquard_common::Data>(val.as_ref()) { 632 632 + crate::backlinks::store::index_record( 633 633 + &mut batch, 634 634 + &app_state.db.backlinks, 635 635 + did.as_str(), 636 636 + collection, 637 637 + &rkey.to_smolstr(), 638 638 + &value, 639 639 + )?; 640 640 + } 630 641 } 631 642 632 643 added_blocks += 1; ··· 662 673 &app_state.db.records, 663 674 keys::record_key(&did, &collection, &rkey), 664 675 ); 676 676 + #[cfg(feature = "backlinks")] 677 677 + crate::backlinks::store::delete_record( 678 678 + &mut batch, 679 679 + &app_state.db.backlinks, 680 680 + did.as_str(), 681 681 + &collection, 682 682 + &rkey.to_smolstr(), 683 683 + )?; 665 684 666 685 let event_id = app_state.db.next_event_id.fetch_add(1, Ordering::SeqCst); 667 686 let evt = StoredEvent {

+115

src/backlinks/api.rs

reviewed

··· 1 1 + use axum::{Json, Router, extract::State, http::StatusCode, routing::get}; 2 2 + use serde::{Deserialize, Serialize}; 3 3 + use smol_str::SmolStr; 4 4 + 5 5 + use crate::control::Hydrant; 6 6 + 7 7 + pub fn router() -> Router<Hydrant> { 8 8 + Router::new() 9 9 + .route( 10 10 + "/xrpc/blue.microcosm.links.getBacklinks", 11 11 + get(handle_get_backlinks), 12 12 + ) 13 13 + .route( 14 14 + "/xrpc/blue.microcosm.links.getBacklinksCount", 15 15 + get(handle_get_backlinks_count), 16 16 + ) 17 17 + } 18 18 + 19 19 + #[derive(Deserialize)] 20 20 + pub struct GetBacklinksParams { 21 21 + pub subject: String, 22 22 + /// filter by source collection, optionally with a path suffix `collection:path` 23 23 + pub source: Option<String>, 24 24 + pub limit: Option<u64>, 25 25 + pub cursor: Option<String>, 26 26 + pub reverse: Option<bool>, 27 27 + } 28 28 + 29 29 + #[derive(Serialize)] 30 30 + pub struct Backlink { 31 31 + pub uri: SmolStr, 32 32 + pub cid: SmolStr, 33 33 + } 34 34 + 35 35 + #[derive(Serialize)] 36 36 + pub struct GetBacklinksOutput { 37 37 + pub backlinks: Vec<Backlink>, 38 38 + #[serde(skip_serializing_if = "Option::is_none")] 39 39 + pub cursor: Option<String>, 40 40 + } 41 41 + 42 42 + #[derive(Deserialize)] 43 43 + pub struct GetBacklinksCountParams { 44 44 + pub subject: String, 45 45 + pub source: Option<String>, 46 46 + } 47 47 + 48 48 + #[derive(Serialize)] 49 49 + pub struct GetBacklinksCountOutput { 50 50 + pub count: u64, 51 51 + } 52 52 + 53 53 + pub async fn handle_get_backlinks( 54 54 + State(hydrant): State<Hydrant>, 55 55 + axum::extract::Query(params): axum::extract::Query<GetBacklinksParams>, 56 56 + ) -> Result<Json<GetBacklinksOutput>, StatusCode> { 57 57 + let limit = params.limit.unwrap_or(50).min(100) as usize; 58 58 + let reverse = params.reverse.unwrap_or(false); 59 59 + 60 60 + let cursor_bytes = params 61 61 + .cursor 62 62 + .as_deref() 63 63 + .and_then(|c| data_encoding::BASE64URL_NOPAD.decode(c.as_bytes()).ok()); 64 64 + 65 65 + let mut fetch = hydrant 66 66 + .backlinks 67 67 + .fetch(params.subject) 68 68 + .limit(limit) 69 69 + .reverse(reverse); 70 70 + if let Some(ref src) = params.source { 71 71 + fetch = fetch.source(src); 72 72 + } 73 73 + if let Some(c) = cursor_bytes { 74 74 + fetch = fetch.cursor(c); 75 75 + } 76 76 + 77 77 + let page = fetch 78 78 + .run() 79 79 + .await 80 80 + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 81 81 + 82 82 + let next_cursor = page 83 83 + .next_cursor 84 84 + .map(|b| data_encoding::BASE64URL_NOPAD.encode(&b)); 85 85 + let backlinks = page 86 86 + .backlinks 87 87 + .into_iter() 88 88 + .map(|e| Backlink { 89 89 + uri: e.uri, 90 90 + cid: e.cid, 91 91 + }) 92 92 + .collect(); 93 93 + 94 94 + Ok(Json(GetBacklinksOutput { 95 95 + backlinks, 96 96 + cursor: next_cursor, 97 97 + })) 98 98 + } 99 99 + 100 100 + pub async fn handle_get_backlinks_count( 101 101 + State(hydrant): State<Hydrant>, 102 102 + axum::extract::Query(params): axum::extract::Query<GetBacklinksCountParams>, 103 103 + ) -> Result<Json<GetBacklinksCountOutput>, StatusCode> { 104 104 + let mut count_q = hydrant.backlinks.count(params.subject); 105 105 + if let Some(ref src) = params.source { 106 106 + count_q = count_q.source(src); 107 107 + } 108 108 + 109 109 + let count = count_q 110 110 + .run() 111 111 + .await 112 112 + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; 113 113 + 114 114 + Ok(Json(GetBacklinksCountOutput { count })) 115 115 + }

+206

src/backlinks/links.rs

reviewed

··· 1 1 + use jacquard_common::Data; 2 2 + use jacquard_common::types::string::AtprotoStr; 3 3 + use jacquard_common::types::uri::Uri; 4 4 + use smol_str::SmolStr; 5 5 + 6 6 + pub struct Link { 7 7 + /// dotted field path within the record, e.g. `.subject.uri` or `.facets[].features[app.bsky.richtext.facet#link].uri` 8 8 + pub path: SmolStr, 9 9 + pub target: SmolStr, 10 10 + } 11 11 + 12 12 + /// extract all link targets from a record, with their field paths. 13 13 + /// 14 14 + /// paths follow the same convention as constellation: 15 15 + /// - start with `.` (root-level field `subject` → `.subject`) 16 16 + /// - object keys: `{parent}.{key}` 17 17 + /// - plain array elements: `{parent}[]` 18 18 + /// - typed array elements (objects with `$type`): `{parent}[{$type}]` 19 19 + /// 20 20 + /// results are sorted by (path, target) and deduplicated. 21 21 + pub fn extract_links(value: &Data) -> Vec<Link> { 22 22 + let mut links = Vec::new(); 23 23 + walk(value, "", &mut links); 24 24 + links.sort_unstable_by(|a, b| a.path.cmp(&b.path).then(a.target.cmp(&b.target))); 25 25 + links.dedup_by(|a, b| a.path == b.path && a.target == b.target); 26 26 + links 27 27 + } 28 28 + 29 29 + fn walk(value: &Data<'_>, path: &str, links: &mut Vec<Link>) { 30 30 + match value { 31 31 + Data::String(s) => { 32 32 + let target = match s { 33 33 + AtprotoStr::AtUri(uri) => uri.as_str(), 34 34 + AtprotoStr::Did(did) => did.as_str(), 35 35 + AtprotoStr::Uri(Uri::At(uri)) => uri.as_str(), 36 36 + AtprotoStr::Uri(Uri::Did(did)) => did.as_str(), 37 37 + _ => return, 38 38 + }; 39 39 + links.push(Link { 40 40 + path: SmolStr::new(path), 41 41 + target: SmolStr::new(target), 42 42 + }); 43 43 + } 44 44 + Data::Array(arr) => { 45 45 + for item in &arr.0 { 46 46 + let child_path = match item { 47 47 + Data::Object(obj) => { 48 48 + let type_tag = obj.0.get("$type").and_then(|v| match v { 49 49 + Data::String(s) => Some(s.as_str()), 50 50 + _ => None, 51 51 + }); 52 52 + match type_tag { 53 53 + Some(t) => format!("{path}[{t}]"), 54 54 + None => format!("{path}[]"), 55 55 + } 56 56 + } 57 57 + _ => format!("{path}[]"), 58 58 + }; 59 59 + walk(item, &child_path, links); 60 60 + } 61 61 + } 62 62 + Data::Object(obj) => { 63 63 + for (k, v) in &obj.0 { 64 64 + walk(v, &format!("{path}.{k}"), links); 65 65 + } 66 66 + } 67 67 + _ => {} 68 68 + } 69 69 + } 70 70 + 71 71 + #[cfg(test)] 72 72 + mod tests { 73 73 + use super::*; 74 74 + 75 75 + fn make(s: &str) -> Data<'static> { 76 76 + Data::from_json_owned(serde_json::Value::String(s.to_string())).unwrap() 77 77 + } 78 78 + 79 79 + fn targets(links: &[Link]) -> Vec<&str> { 80 80 + links.iter().map(|l| l.target.as_str()).collect() 81 81 + } 82 82 + 83 83 + fn paths(links: &[Link]) -> Vec<&str> { 84 84 + links.iter().map(|l| l.path.as_str()).collect() 85 85 + } 86 86 + 87 87 + #[test] 88 88 + fn extracts_at_uri() { 89 89 + let val = make("at://did:plc:abc123/app.bsky.feed.post/tid123"); 90 90 + let links = extract_links(&val); 91 91 + assert_eq!( 92 92 + targets(&links), 93 93 + vec!["at://did:plc:abc123/app.bsky.feed.post/tid123"] 94 94 + ); 95 95 + assert_eq!(paths(&links), vec![""]); 96 96 + } 97 97 + 98 98 + #[test] 99 99 + fn extracts_did() { 100 100 + let val = make("did:plc:abc123"); 101 101 + let links = extract_links(&val); 102 102 + assert_eq!(targets(&links), vec!["did:plc:abc123"]); 103 103 + } 104 104 + 105 105 + #[test] 106 106 + fn skips_plain_string() { 107 107 + assert!(extract_links(&make("hello world")).is_empty()); 108 108 + } 109 109 + 110 110 + #[test] 111 111 + fn skips_handle() { 112 112 + assert!(extract_links(&make("alice.bsky.social")).is_empty()); 113 113 + } 114 114 + 115 115 + #[test] 116 116 + fn object_paths() { 117 117 + let json = serde_json::json!({ 118 118 + "subject": { 119 119 + "uri": "at://did:plc:abc/app.bsky.feed.post/tid", 120 120 + "cid": "bafyreiclp443lavogvhj3d2ob2cxbfuscni2k5jk7bebjzg7khl3esabwq" 121 121 + } 122 122 + }); 123 123 + let val = Data::from_json_owned(json).unwrap(); 124 124 + let links = extract_links(&val); 125 125 + assert_eq!(links.len(), 1); 126 126 + assert_eq!( 127 127 + links[0].target.as_str(), 128 128 + "at://did:plc:abc/app.bsky.feed.post/tid" 129 129 + ); 130 130 + assert_eq!(links[0].path.as_str(), ".subject.uri"); 131 131 + } 132 132 + 133 133 + #[test] 134 134 + fn array_paths_plain() { 135 135 + let json = serde_json::json!([ 136 136 + "at://did:plc:a/app.bsky.feed.post/tid1", 137 137 + "at://did:plc:b/app.bsky.feed.post/tid2" 138 138 + ]); 139 139 + let val = Data::from_json_owned(json).unwrap(); 140 140 + let links = extract_links(&val); 141 141 + assert_eq!(links.len(), 2); 142 142 + assert!(links.iter().all(|l| l.path.as_str() == "[]")); 143 143 + } 144 144 + 145 145 + #[test] 146 146 + fn array_paths_typed_object() { 147 147 + let json = serde_json::json!({ 148 148 + "features": [ 149 149 + { 150 150 + "$type": "app.bsky.richtext.facet#link", 151 151 + "uri": "at://did:plc:abc/app.bsky.feed.post/tid" 152 152 + } 153 153 + ] 154 154 + }); 155 155 + let val = Data::from_json_owned(json).unwrap(); 156 156 + let links = extract_links(&val); 157 157 + assert_eq!(links.len(), 1); 158 158 + assert_eq!( 159 159 + links[0].path.as_str(), 160 160 + ".features[app.bsky.richtext.facet#link].uri" 161 161 + ); 162 162 + } 163 163 + 164 164 + #[test] 165 165 + fn array_paths_untyped_object() { 166 166 + let json = serde_json::json!({ 167 167 + "items": [ 168 168 + { "subject": "did:plc:abc" } 169 169 + ] 170 170 + }); 171 171 + let val = Data::from_json_owned(json).unwrap(); 172 172 + let links = extract_links(&val); 173 173 + assert_eq!(links.len(), 1); 174 174 + assert_eq!(links[0].path.as_str(), ".items[].subject"); 175 175 + } 176 176 + 177 177 + #[test] 178 178 + fn deduplicates_same_path_and_target() { 179 179 + let uri = "at://did:plc:abc123/app.bsky.feed.post/tid123"; 180 180 + let json = serde_json::json!([uri, uri]); 181 181 + let val = Data::from_json_owned(json).unwrap(); 182 182 + assert_eq!(extract_links(&val).len(), 1); 183 183 + } 184 184 + 185 185 + #[test] 186 186 + fn keeps_same_target_at_different_paths() { 187 187 + let uri = "at://did:plc:abc/app.bsky.feed.post/tid"; 188 188 + let json = serde_json::json!({ 189 189 + "root": { "uri": uri }, 190 190 + "parent": { "uri": uri } 191 191 + }); 192 192 + let val = Data::from_json_owned(json).unwrap(); 193 193 + let links = extract_links(&val); 194 194 + assert_eq!(links.len(), 2); 195 195 + let ps: Vec<_> = links.iter().map(|l| l.path.as_str()).collect(); 196 196 + assert!(ps.contains(&".parent.uri")); 197 197 + assert!(ps.contains(&".root.uri")); 198 198 + } 199 199 + 200 200 + #[test] 201 201 + fn skips_cid_link() { 202 202 + let json = serde_json::json!({"$link": "bafyreiclp443lavogvhj3d2ob2cxbfuscni2k5jk7bebjzg7khl3esabwq"}); 203 203 + let val = Data::from_json_owned(json).unwrap(); 204 204 + assert!(extract_links(&val).is_empty()); 205 205 + } 206 206 + }

+232

src/backlinks/mod.rs

reviewed

··· 1 1 + pub mod links; 2 2 + pub mod store; 3 3 + 4 4 + pub(crate) mod api; 5 5 + 6 6 + use std::ops::Bound; 7 7 + use std::sync::Arc; 8 8 + 9 9 + use miette::{IntoDiagnostic, Result}; 10 10 + use smol_str::SmolStr; 11 11 + use tracing::warn; 12 12 + 13 13 + use crate::state::AppState; 14 14 + 15 15 + /// handle for querying the backlinks index. 16 16 + #[derive(Clone)] 17 17 + pub struct BacklinksControl(pub(crate) Arc<AppState>); 18 18 + 19 19 + pub struct BacklinkEntry { 20 20 + pub uri: SmolStr, 21 21 + pub cid: SmolStr, 22 22 + } 23 23 + 24 24 + pub struct BacklinksPage { 25 25 + pub backlinks: Vec<BacklinkEntry>, 26 26 + /// raw key bytes of the last returned entry, used as cursor for next page 27 27 + pub next_cursor: Option<Vec<u8>>, 28 28 + } 29 29 + 30 30 + impl BacklinksControl { 31 31 + /// begin building a paginated fetch of backlinks pointing to `subject`. 32 32 + pub fn fetch(&self, subject: impl Into<String>) -> BacklinksFetch { 33 33 + BacklinksFetch { 34 34 + state: self.0.clone(), 35 35 + subject: subject.into(), 36 36 + collection: None, 37 37 + path: None, 38 38 + limit: 50, 39 39 + reverse: false, 40 40 + cursor: None, 41 41 + } 42 42 + } 43 43 + 44 44 + /// begin building a backlinks count query for `subject`. 45 45 + pub fn count(&self, subject: impl Into<String>) -> BacklinksCount { 46 46 + BacklinksCount { 47 47 + state: self.0.clone(), 48 48 + subject: subject.into(), 49 49 + collection: None, 50 50 + path: None, 51 51 + } 52 52 + } 53 53 + } 54 54 + 55 55 + /// a paginated fetch against the backlinks reverse index. 56 56 + /// 57 57 + /// obtain via [`BacklinksControl::fetch`] and execute with [`BacklinksFetch::run`]. 58 58 + pub struct BacklinksFetch { 59 59 + state: Arc<AppState>, 60 60 + subject: String, 61 61 + collection: Option<String>, 62 62 + path: Option<String>, 63 63 + limit: usize, 64 64 + reverse: bool, 65 65 + cursor: Option<Vec<u8>>, 66 66 + } 67 67 + 68 68 + impl BacklinksFetch { 69 69 + /// filter results to the given source collection NSID. 70 70 + pub fn collection(mut self, collection: impl Into<String>) -> Self { 71 71 + self.collection = Some(collection.into()); 72 72 + self 73 73 + } 74 74 + 75 75 + /// filter results to a specific dotted field path within the record (e.g. `.subject.uri`). 76 76 + /// has no effect unless a collection is also set. 77 77 + pub fn path(mut self, path: impl Into<String>) -> Self { 78 78 + self.path = Some(path.into()); 79 79 + self 80 80 + } 81 81 + 82 82 + /// convenience: parse a `source` parameter of the form `collection` or `collection:path` 83 83 + /// and apply both filters. the path component (if present) has `.` prepended automatically. 84 84 + pub fn source(self, source: &str) -> Self { 85 85 + match source.split_once(':') { 86 86 + Some((col, p)) => self.collection(col).path(format!(".{p}")), 87 87 + None => self.collection(source), 88 88 + } 89 89 + } 90 90 + 91 91 + /// maximum number of results to return (default 50). 92 92 + pub fn limit(mut self, limit: usize) -> Self { 93 93 + self.limit = limit; 94 94 + self 95 95 + } 96 96 + 97 97 + /// if `true`, return results in reverse key order (default `false`). 98 98 + pub fn reverse(mut self, reverse: bool) -> Self { 99 99 + self.reverse = reverse; 100 100 + self 101 101 + } 102 102 + 103 103 + /// resume from the raw key bytes returned as `next_cursor` from a previous page. 104 104 + pub fn cursor(mut self, cursor: Vec<u8>) -> Self { 105 105 + self.cursor = Some(cursor); 106 106 + self 107 107 + } 108 108 + 109 109 + /// execute and return a page of backlink entries. 110 110 + pub async fn run(self) -> Result<BacklinksPage> { 111 111 + tokio::task::spawn_blocking(move || { 112 112 + let db = &self.state.db; 113 113 + let scan_prefix = store::reverse_scan_prefix( 114 114 + &self.subject, 115 115 + self.collection.as_deref(), 116 116 + self.path.as_deref(), 117 117 + ); 118 118 + 119 119 + let iter: Box<dyn Iterator<Item = _>> = if !self.reverse { 120 120 + if let Some(ref cursor_bytes) = self.cursor { 121 121 + Box::new(db.backlinks.range::<&[u8], _>(( 122 122 + Bound::Excluded(cursor_bytes.as_slice()), 123 123 + Bound::Unbounded, 124 124 + ))) 125 125 + } else { 126 126 + Box::new(db.backlinks.range(scan_prefix.as_slice()..)) 127 127 + } 128 128 + } else { 129 129 + // for reverse scans, bound the end at the cursor (exclusive) or the 130 130 + // prefix end (increment last byte to get an exclusive upper bound) 131 131 + let end: Vec<u8> = if let Some(ref cursor_bytes) = self.cursor { 132 132 + cursor_bytes.clone() 133 133 + } else { 134 134 + let mut end = scan_prefix.clone(); 135 135 + if let Some(last) = end.last_mut() { 136 136 + *last = last.saturating_add(1); 137 137 + } 138 138 + end 139 139 + }; 140 140 + Box::new( 141 141 + db.backlinks 142 142 + .range(scan_prefix.as_slice()..end.as_slice()) 143 143 + .rev(), 144 144 + ) 145 145 + }; 146 146 + 147 147 + let mut backlinks = Vec::with_capacity(self.limit); 148 148 + let mut next_cursor = None; 149 149 + 150 150 + for item in iter { 151 151 + let (key, _) = item.into_inner().into_diagnostic()?; 152 152 + if !key.starts_with(scan_prefix.as_slice()) { 153 153 + break; 154 154 + } 155 155 + if backlinks.len() >= self.limit { 156 156 + next_cursor = Some(key.to_vec()); 157 157 + break; 158 158 + } 159 159 + 160 160 + let Some((col, _path, did, rkey)) = store::source_from_reverse_key(&key) else { 161 161 + warn!("backlinks: could not extract source from reverse key"); 162 162 + continue; 163 163 + }; 164 164 + let Some(cid) = store::lookup_cid_from_ks(&db.records, did, col, rkey) else { 165 165 + // record deleted after backlink was written 166 166 + continue; 167 167 + }; 168 168 + let at_uri = format!("at://{did}/{col}/{rkey}"); 169 169 + backlinks.push(BacklinkEntry { 170 170 + uri: SmolStr::new(at_uri), 171 171 + cid: SmolStr::new(cid), 172 172 + }); 173 173 + } 174 174 + 175 175 + Ok(BacklinksPage { 176 176 + backlinks, 177 177 + next_cursor, 178 178 + }) 179 179 + }) 180 180 + .await 181 181 + .into_diagnostic()? 182 182 + } 183 183 + } 184 184 + 185 185 + /// a count query against the backlinks reverse index. 186 186 + /// 187 187 + /// obtain via [`BacklinksControl::count`] and execute with [`BacklinksCount::run`]. 188 188 + pub struct BacklinksCount { 189 189 + state: Arc<AppState>, 190 190 + subject: String, 191 191 + collection: Option<String>, 192 192 + path: Option<String>, 193 193 + } 194 194 + 195 195 + impl BacklinksCount { 196 196 + /// filter to the given source collection NSID. 197 197 + pub fn collection(mut self, collection: impl Into<String>) -> Self { 198 198 + self.collection = Some(collection.into()); 199 199 + self 200 200 + } 201 201 + 202 202 + /// filter to a specific dotted field path within the record (e.g. `.subject.uri`). 203 203 + /// has no effect unless a collection is also set. 204 204 + pub fn path(mut self, path: impl Into<String>) -> Self { 205 205 + self.path = Some(path.into()); 206 206 + self 207 207 + } 208 208 + 209 209 + /// parse a `source` parameter of the form `collection` or `collection:path` 210 210 + /// and apply both filters. the path component (if present) has `.` prepended automatically. 211 211 + pub fn source(self, source: &str) -> Self { 212 212 + match source.split_once(':') { 213 213 + Some((col, p)) => self.collection(col).path(format!(".{p}")), 214 214 + None => self.collection(source), 215 215 + } 216 216 + } 217 217 + 218 218 + /// execute and return the total count of matching entries. 219 219 + pub async fn run(self) -> Result<u64> { 220 220 + tokio::task::spawn_blocking(move || { 221 221 + let db = &self.state.db; 222 222 + let scan_prefix = store::reverse_scan_prefix( 223 223 + &self.subject, 224 224 + self.collection.as_deref(), 225 225 + self.path.as_deref(), 226 226 + ); 227 227 + Ok(db.backlinks.prefix(&scan_prefix).count() as u64) 228 228 + }) 229 229 + .await 230 230 + .into_diagnostic()? 231 231 + } 232 232 + }

+317

src/backlinks/store.rs

reviewed

··· 1 1 + use fjall::{Keyspace, OwnedWriteBatch}; 2 2 + use jacquard_common::types::string::Did; 3 3 + use jacquard_common::{Data, types::cid::Cid}; 4 4 + use miette::{IntoDiagnostic, Result}; 5 5 + 6 6 + use crate::db::keys::{self, SEP}; 7 7 + use crate::db::types::DbRkey; 8 8 + 9 9 + /// key format: `f|{did}|{collection}|{rkey}` 10 10 + /// 11 11 + /// value: msgpack-encoded `Vec<(String, String)>`, list of (path, target) pairs found in this record. 12 12 + pub fn forward_key(did: &str, collection: &str, rkey: &str) -> Vec<u8> { 13 13 + let mut key = Vec::with_capacity(2 + did.len() + 1 + collection.len() + 1 + rkey.len()); 14 14 + key.push(b'f'); 15 15 + key.push(SEP); 16 16 + key.extend_from_slice(did.as_bytes()); 17 17 + key.push(SEP); 18 18 + key.extend_from_slice(collection.as_bytes()); 19 19 + key.push(SEP); 20 20 + key.extend_from_slice(rkey.as_bytes()); 21 21 + key 22 22 + } 23 23 + 24 24 + /// key format: `r|{target}|{collection}|{path}|{did}|{rkey}` 25 25 + /// 26 26 + /// value: empty — the key itself carries all information. 27 27 + /// 28 28 + /// `|` separators are unambiguous because targets (AT URIs, DIDs), collection 29 29 + /// NSIDs, field paths, DIDs, and rkeys cannot contain `|`. 30 30 + /// prefix scans: 31 31 + /// - all backlinks to a target: `r|{target}|` 32 32 + /// - backlinks from a specific collection: `r|{target}|{collection}|` 33 33 + /// - backlinks from a specific collection+field path: `r|{target}|{collection}|{path}|` 34 34 + pub fn reverse_key(target: &str, collection: &str, path: &str, did: &str, rkey: &str) -> Vec<u8> { 35 35 + let mut key = Vec::with_capacity( 36 36 + 2 + target.len() + 1 + collection.len() + 1 + path.len() + 1 + did.len() + 1 + rkey.len(), 37 37 + ); 38 38 + key.push(b'r'); 39 39 + key.push(SEP); 40 40 + key.extend_from_slice(target.as_bytes()); 41 41 + key.push(SEP); 42 42 + key.extend_from_slice(collection.as_bytes()); 43 43 + key.push(SEP); 44 44 + key.extend_from_slice(path.as_bytes()); 45 45 + key.push(SEP); 46 46 + key.extend_from_slice(did.as_bytes()); 47 47 + key.push(SEP); 48 48 + key.extend_from_slice(rkey.as_bytes()); 49 49 + key 50 50 + } 51 51 + 52 52 + /// scan prefix for all reverse entries pointing to `target`. 53 53 + /// 54 54 + /// - no collection: `r|{target}|` 55 55 + /// - collection only: `r|{target}|{collection}|` (matches all paths for that collection) 56 56 + /// - collection + path: `r|{target}|{collection}|{path}|` (exact path match) 57 57 + /// 58 58 + /// path should include the leading `.` (e.g. `.subject.uri`). always ends with `|`. 59 59 + pub fn reverse_scan_prefix(target: &str, collection: Option<&str>, path: Option<&str>) -> Vec<u8> { 60 60 + let col_len = collection.map_or(0, |c| c.len() + 1); 61 61 + let path_len = if collection.is_some() { 62 62 + path.map_or(0, |p| p.len() + 1) 63 63 + } else { 64 64 + 0 65 65 + }; 66 66 + let mut prefix = Vec::with_capacity(2 + target.len() + 1 + col_len + path_len); 67 67 + prefix.push(b'r'); 68 68 + prefix.push(SEP); 69 69 + prefix.extend_from_slice(target.as_bytes()); 70 70 + prefix.push(SEP); 71 71 + if let Some(col) = collection { 72 72 + prefix.extend_from_slice(col.as_bytes()); 73 73 + prefix.push(SEP); 74 74 + if let Some(p) = path { 75 75 + prefix.extend_from_slice(p.as_bytes()); 76 76 + prefix.push(SEP); 77 77 + } 78 78 + } 79 79 + prefix 80 80 + } 81 81 + 82 82 + /// scan prefix for all forward entries from the given DID. 83 83 + pub fn forward_did_prefix(did: &Did) -> Vec<u8> { 84 84 + let did_str = did.as_str(); 85 85 + let mut prefix = Vec::with_capacity(2 + did_str.len() + 1); 86 86 + prefix.push(b'f'); 87 87 + prefix.push(SEP); 88 88 + prefix.extend_from_slice(did_str.as_bytes()); 89 89 + prefix.push(SEP); 90 90 + prefix 91 91 + } 92 92 + 93 93 + /// extract (collection, path, did, rkey) from a reverse key. 94 94 + /// 95 95 + /// reverse key format: `r|{target}|{collection}|{path}|{did}|{rkey}` 96 96 + /// parses the last four `|`-separated segments from the right. 97 97 + pub fn source_from_reverse_key(key: &[u8]) -> Option<(&str, &str, &str, &str)> { 98 98 + let rkey_sep = key.iter().rposition(|&b| b == SEP)?; 99 99 + let rkey = std::str::from_utf8(&key[rkey_sep + 1..]).ok()?; 100 100 + let did_sep = key[..rkey_sep].iter().rposition(|&b| b == SEP)?; 101 101 + let did = std::str::from_utf8(&key[did_sep + 1..rkey_sep]).ok()?; 102 102 + let path_sep = key[..did_sep].iter().rposition(|&b| b == SEP)?; 103 103 + let path = std::str::from_utf8(&key[path_sep + 1..did_sep]).ok()?; 104 104 + let col_sep = key[..path_sep].iter().rposition(|&b| b == SEP)?; 105 105 + let col = std::str::from_utf8(&key[col_sep + 1..path_sep]).ok()?; 106 106 + Some((col, path, did, rkey)) 107 107 + } 108 108 + 109 109 + /// index all links found in a record. 110 110 + /// 111 111 + /// on update, stale reverse entries from the previous version are removed first. 112 112 + /// on create with no links, no entries are written. 113 113 + pub fn index_record( 114 114 + batch: &mut OwnedWriteBatch, 115 115 + backlinks_ks: &Keyspace, 116 116 + did: &str, 117 117 + collection: &str, 118 118 + rkey: &str, 119 119 + value: &Data, 120 120 + ) -> Result<()> { 121 121 + let fwd_key = forward_key(did, collection, rkey); 122 122 + 123 123 + // remove reverse entries from the previous version of this record if any 124 124 + if let Some(old_bytes) = backlinks_ks.get(&fwd_key).into_diagnostic()? { 125 125 + let old_links: Vec<(String, String)> = 126 126 + rmp_serde::from_slice(&old_bytes).into_diagnostic()?; 127 127 + for (path, target) in &old_links { 128 128 + batch.remove( 129 129 + backlinks_ks, 130 130 + reverse_key(target, collection, path, did, rkey), 131 131 + ); 132 132 + } 133 133 + } 134 134 + 135 135 + let new_links = crate::backlinks::links::extract_links(value); 136 136 + if new_links.is_empty() { 137 137 + batch.remove(backlinks_ks, fwd_key); 138 138 + } else { 139 139 + for link in &new_links { 140 140 + batch.insert( 141 141 + backlinks_ks, 142 142 + reverse_key(&link.target, collection, &link.path, did, rkey), 143 143 + &[] as &[u8], 144 144 + ); 145 145 + } 146 146 + let fwd_pairs: Vec<(&str, &str)> = new_links 147 147 + .iter() 148 148 + .map(|l| (l.path.as_str(), l.target.as_str())) 149 149 + .collect(); 150 150 + let fwd_val = rmp_serde::to_vec(&fwd_pairs).into_diagnostic()?; 151 151 + batch.insert(backlinks_ks, fwd_key, fwd_val); 152 152 + } 153 153 + 154 154 + Ok(()) 155 155 + } 156 156 + 157 157 + /// remove all backlink index entries for a deleted record. 158 158 + pub fn delete_record( 159 159 + batch: &mut OwnedWriteBatch, 160 160 + backlinks_ks: &Keyspace, 161 161 + did: &str, 162 162 + collection: &str, 163 163 + rkey: &str, 164 164 + ) -> Result<()> { 165 165 + let fwd_key = forward_key(did, collection, rkey); 166 166 + let Some(old_bytes) = backlinks_ks.get(&fwd_key).into_diagnostic()? else { 167 167 + return Ok(()); 168 168 + }; 169 169 + let links: Vec<(String, String)> = rmp_serde::from_slice(&old_bytes).into_diagnostic()?; 170 170 + for (path, target) in &links { 171 171 + batch.remove( 172 172 + backlinks_ks, 173 173 + reverse_key(target, collection, path, did, rkey), 174 174 + ); 175 175 + } 176 176 + batch.remove(backlinks_ks, fwd_key); 177 177 + Ok(()) 178 178 + } 179 179 + 180 180 + /// remove all backlink index entries for every record in a deleted repo. 181 181 + /// 182 182 + /// scans the forward index by the `f|{did}|` prefix and removes both forward 183 183 + /// and reverse entries for each record found. 184 184 + pub fn delete_repo(batch: &mut OwnedWriteBatch, backlinks_ks: &Keyspace, did: &Did) -> Result<()> { 185 185 + let prefix = forward_did_prefix(did); 186 186 + let did_str = did.as_str(); 187 187 + 188 188 + for guard in backlinks_ks.prefix(&prefix) { 189 189 + let (fwd_key, fwd_val) = guard.into_inner().into_diagnostic()?; 190 190 + // fwd_key suffix after f|{did}| is {collection}|{rkey} 191 191 + let suffix = &fwd_key[prefix.len()..]; 192 192 + let Some(sep) = suffix.iter().position(|&b| b == SEP) else { 193 193 + tracing::warn!("backlinks: malformed forward key during delete_repo"); 194 194 + continue; 195 195 + }; 196 196 + let (Ok(col), Ok(rkey)) = ( 197 197 + std::str::from_utf8(&suffix[..sep]), 198 198 + std::str::from_utf8(&suffix[sep + 1..]), 199 199 + ) else { 200 200 + tracing::warn!("backlinks: invalid utf8 in forward key during delete_repo"); 201 201 + continue; 202 202 + }; 203 203 + 204 204 + let links: Vec<(String, String)> = match rmp_serde::from_slice(&fwd_val) { 205 205 + Ok(t) => t, 206 206 + Err(e) => { 207 207 + tracing::warn!( 208 208 + "backlinks: failed to decode forward value for {did_str}/{col}/{rkey}: {e}" 209 209 + ); 210 210 + continue; 211 211 + } 212 212 + }; 213 213 + for (path, target) in &links { 214 214 + batch.remove(backlinks_ks, reverse_key(target, col, path, did_str, rkey)); 215 215 + } 216 216 + batch.remove(backlinks_ks, fwd_key); 217 217 + } 218 218 + 219 219 + Ok(()) 220 220 + } 221 221 + 222 222 + /// look up the CID string for a record from the records keyspace. 223 223 + /// returns `None` if the record is not found (e.g. deleted after backlink was written). 224 224 + pub fn lookup_cid_from_ks( 225 225 + records: &Keyspace, 226 226 + did: &str, 227 227 + collection: &str, 228 228 + rkey: &str, 229 229 + ) -> Option<String> { 230 230 + let did = Did::new(did).ok()?; 231 231 + let db_rkey = DbRkey::new(rkey); 232 232 + let record_key = keys::record_key(&did, collection, &db_rkey); 233 233 + let cid_bytes = records.get(record_key).ok()??; 234 234 + Cid::new(&cid_bytes).ok().map(|c| c.as_str().to_string()) 235 235 + } 236 236 + 237 237 + #[cfg(test)] 238 238 + mod tests { 239 239 + use super::*; 240 240 + 241 241 + const TARGET: &str = "at://did:plc:abc/app.bsky.feed.post/tid"; 242 242 + const COLLECTION: &str = "app.bsky.feed.like"; 243 243 + const PATH: &str = ".subject.uri"; 244 244 + const SRC_DID: &str = "did:plc:xyz"; 245 245 + const SRC_RKEY: &str = "rkey1"; 246 246 + 247 247 + #[test] 248 248 + fn reverse_key_scan_prefix_no_collection() { 249 249 + let key = reverse_key(TARGET, COLLECTION, PATH, SRC_DID, SRC_RKEY); 250 250 + let prefix = reverse_scan_prefix(TARGET, None, None); 251 251 + assert!(key.starts_with(&prefix)); 252 252 + } 253 253 + 254 254 + #[test] 255 255 + fn reverse_key_scan_prefix_with_collection() { 256 256 + let key = reverse_key(TARGET, COLLECTION, PATH, SRC_DID, SRC_RKEY); 257 257 + let prefix = reverse_scan_prefix(TARGET, Some(COLLECTION), None); 258 258 + assert!(key.starts_with(&prefix)); 259 259 + } 260 260 + 261 261 + #[test] 262 262 + fn reverse_key_scan_prefix_with_collection_and_path() { 263 263 + let key = reverse_key(TARGET, COLLECTION, PATH, SRC_DID, SRC_RKEY); 264 264 + let prefix = reverse_scan_prefix(TARGET, Some(COLLECTION), Some(PATH)); 265 265 + assert!(key.starts_with(&prefix)); 266 266 + } 267 267 + 268 268 + #[test] 269 269 + fn reverse_key_collection_prefix_does_not_match_other_collection() { 270 270 + let collection2 = "app.bsky.feed.like2"; 271 271 + let key2 = reverse_key(TARGET, collection2, PATH, SRC_DID, SRC_RKEY); 272 272 + let prefix1 = reverse_scan_prefix(TARGET, Some(COLLECTION), None); 273 273 + assert!(!key2.starts_with(&prefix1)); 274 274 + } 275 275 + 276 276 + #[test] 277 277 + fn reverse_key_path_prefix_does_not_match_other_path() { 278 278 + let path2 = ".subject.uri2"; 279 279 + let key2 = reverse_key(TARGET, COLLECTION, path2, SRC_DID, SRC_RKEY); 280 280 + let prefix = reverse_scan_prefix(TARGET, Some(COLLECTION), Some(PATH)); 281 281 + assert!(!key2.starts_with(&prefix)); 282 282 + } 283 283 + 284 284 + #[test] 285 285 + fn collection_only_prefix_matches_all_paths() { 286 286 + let path2 = ".embed.uri"; 287 287 + let key2 = reverse_key(TARGET, COLLECTION, path2, SRC_DID, SRC_RKEY); 288 288 + let col_prefix = reverse_scan_prefix(TARGET, Some(COLLECTION), None); 289 289 + assert!(key2.starts_with(&col_prefix)); 290 290 + } 291 291 + 292 292 + #[test] 293 293 + fn source_from_reverse_key_roundtrip() { 294 294 + let key = reverse_key(TARGET, COLLECTION, PATH, SRC_DID, SRC_RKEY); 295 295 + let (col, path, did, rkey) = source_from_reverse_key(&key).unwrap(); 296 296 + assert_eq!(col, COLLECTION); 297 297 + assert_eq!(path, PATH); 298 298 + assert_eq!(did, SRC_DID); 299 299 + assert_eq!(rkey, SRC_RKEY); 300 300 + } 301 301 + 302 302 + #[test] 303 303 + fn forward_did_prefix_matches_key() { 304 304 + let did = jacquard_common::types::string::Did::new("did:plc:abc123").unwrap(); 305 305 + let prefix = forward_did_prefix(&did); 306 306 + let key = forward_key("did:plc:abc123", "app.bsky.feed.post", "tid123"); 307 307 + assert!(key.starts_with(&prefix)); 308 308 + } 309 309 + 310 310 + #[test] 311 311 + fn forward_did_prefix_does_not_match_other_did() { 312 312 + let did = jacquard_common::types::string::Did::new("did:plc:abc123").unwrap(); 313 313 + let prefix = forward_did_prefix(&did); 314 314 + let other_key = forward_key("did:plc:xyz999", "app.bsky.feed.post", "tid123"); 315 315 + assert!(!other_key.starts_with(&prefix)); 316 316 + } 317 317 + }

+9

src/config.rs

reviewed

··· 157 157 pub filter_signals: Option<Vec<String>>, 158 158 pub filter_collections: Option<Vec<String>>, 159 159 pub filter_excludes: Option<Vec<String>>, 160 160 + /// enable backlinks indexing (only meaningful in non-ephemeral mode). 161 161 + /// set via `HYDRANT_ENABLE_BACKLINKS=true`. 162 162 + pub enable_backlinks: bool, 160 163 /// crawler sources: each entry pairs a URL with a discovery mode. 161 164 /// 162 165 /// set via `HYDRANT_CRAWLER_URLS` as a comma-separated list of `[mode::]url` entries, ··· 300 303 .filter(|s| !s.is_empty()) 301 304 .collect() 302 305 }); 306 306 + 307 307 + let enable_backlinks: bool = cfg!("ENABLE_BACKLINKS", false); 303 308 304 309 let default_mode = CrawlerMode::default_for(full_network); 305 310 let crawler_sources = match std::env::var("HYDRANT_CRAWLER_URLS") { ··· 353 358 filter_signals, 354 359 filter_collections, 355 360 filter_excludes, 361 361 + enable_backlinks, 356 362 crawler_sources, 357 363 }) 358 364 } ··· 467 473 } 468 474 if let Some(excludes) = &self.filter_excludes { 469 475 config_line!(f, "filter excludes", format_args!("{:?}", excludes))?; 476 476 + } 477 477 + if self.enable_backlinks { 478 478 + config_line!(f, "backlinks", "enabled")?; 470 479 } 471 480 Ok(()) 472 481 }

+4

src/control.rs

reviewed

··· 74 74 pub filter: FilterControl, 75 75 pub repos: ReposControl, 76 76 pub db: DbControl, 77 77 + #[cfg(feature = "backlinks")] 78 78 + pub backlinks: crate::backlinks::BacklinksControl, 77 79 pub(crate) state: Arc<AppState>, 78 80 config: Arc<Config>, 79 81 started: Arc<AtomicBool>, ··· 148 150 filter: FilterControl(state.clone()), 149 151 repos: ReposControl(state.clone()), 150 152 db: DbControl(state.clone()), 153 153 + #[cfg(feature = "backlinks")] 154 154 + backlinks: crate::backlinks::BacklinksControl(state.clone()), 151 155 state, 152 156 config: Arc::new(config), 153 157 started: Arc::new(AtomicBool::new(false)),

+33 -9

src/db/mod.rs

reviewed

··· 48 48 pub counts: Keyspace, 49 49 pub filter: Keyspace, 50 50 pub crawler: Keyspace, 51 51 + #[cfg(feature = "backlinks")] 52 52 + pub backlinks: Keyspace, 51 53 pub event_tx: broadcast::Sender<BroadcastEvent>, 52 54 pub next_event_id: Arc<AtomicU64>, 53 55 pub counts_map: HashMap<SmolStr, u64>, ··· 99 101 }}; 100 102 } 101 103 104 104 + const fn kb(v: u32) -> u32 { 105 105 + v * 1024 106 106 + } 107 107 + const fn mb(v: u64) -> u64 { 108 108 + v * 1024 * 1024 109 109 + } 110 110 + 102 111 impl Db { 103 112 pub fn open(cfg: &crate::config::Config) -> Result<Self> { 104 104 - const fn kb(v: u32) -> u32 { 105 105 - v * 1024 106 106 - } 107 107 - const fn mb(v: u64) -> u64 { 108 108 - v * 1024 * 1024 109 109 - } 110 110 - 111 113 let db = Database::builder(&cfg.database_path) 112 114 .cache_size(cfg.cache_size * 2_u64.pow(20) / 2) 113 115 .manual_journal_persist(true) ··· 152 154 } 153 155 None 154 156 }; 155 155 - let dicts = ["repos", "blocks", "events"].into_iter().fold( 157 157 + let dicts = ["repos", "blocks", "events", "backlinks"].into_iter().fold( 156 158 std::collections::HashMap::new(), 157 159 |mut acc, name| { 158 160 let Some(dict) = load_dict(name) else { ··· 258 260 // cids arent compressable, most rkeys are TIDs so they will get compressed 259 261 // by prefix truncation anyway 260 262 .data_block_compression_policy(CompressionPolicy::disabled()) 261 261 - .data_block_restart_interval_policy(RestartIntervalPolicy::new([9, 18])), 263 263 + .data_block_restart_interval_policy(RestartIntervalPolicy::new([16, 32])), 262 264 )?; 263 265 let cursors = open_ks( 264 266 "cursors", ··· 354 356 .data_block_restart_interval_policy(RestartIntervalPolicy::all(2)), 355 357 )?; 356 358 359 359 + #[cfg(feature = "backlinks")] 360 360 + let backlinks = open_ks( 361 361 + "backlinks", 362 362 + opts() 363 363 + // lets assume we hit backlinks, getBacklinks will use iterator anyway 364 364 + // so we can disable bloom filter okay 365 365 + .expect_point_read_hits(true) 366 366 + .max_memtable_size(mb(cfg.db_records_memtable_size_mb)) 367 367 + // same as records basically 368 368 + .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(32)])) 369 369 + .data_block_compression_policy(CompressionPolicy::new([ 370 370 + CompressionType::None, 371 371 + get_compression("backlinks", 3), 372 372 + ])) 373 373 + .data_block_restart_interval_policy(RestartIntervalPolicy::new([16, 32])), 374 374 + )?; 375 375 + 357 376 // when adding new keyspaces, make sure to add them to the /stats endpoint 358 377 // and also update any relevant /debug/* endpoints 359 378 ··· 411 430 counts, 412 431 filter, 413 432 crawler, 433 433 + #[cfg(feature = "backlinks")] 434 434 + backlinks, 414 435 event_tx, 415 436 counts_map, 416 437 next_event_id: Arc::new(AtomicU64::new(last_id + 1)), ··· 422 443 "blocks" => &self.blocks, 423 444 "events" => &self.events, 424 445 "repos" => &self.repos, 446 446 + "backlinks" => &self.backlinks, 425 447 _ => miette::bail!("unknown keyspace for training: {ks_name}"), 426 448 }; 427 449 ··· 523 545 compact(self.filter.clone()), 524 546 compact(self.crawler.clone()), 525 547 )?; 548 548 + #[cfg(feature = "backlinks")] 549 549 + compact(self.backlinks.clone()).await?; 526 550 Ok(()) 527 551 } 528 552

+2

src/lib.rs

reviewed

··· 5 5 6 6 pub(crate) mod api; 7 7 pub(crate) mod backfill; 8 8 + #[cfg(feature = "backlinks")] 9 9 + pub(crate) mod backlinks; 8 10 pub(crate) mod crawler; 9 11 pub(crate) mod db; 10 12 pub(crate) mod ingest;

+26

src/ops.rs

reviewed

··· 2 2 use fjall::Slice; 3 3 4 4 use jacquard_common::CowStr; 5 5 + #[cfg(feature = "backlinks")] 6 6 + use jacquard_common::Data; 5 7 use jacquard_common::IntoStatic; 6 8 use jacquard_common::types::cid::Cid; 7 9 use jacquard_common::types::crypto::PublicKey; ··· 117 119 let k = guard.key().into_diagnostic()?; 118 120 batch.remove(&db.counts, k); 119 121 } 122 122 + 123 123 + // 5. remove backlinks for all records in this repo 124 124 + #[cfg(feature = "backlinks")] 125 125 + crate::backlinks::store::delete_repo(batch, &db.backlinks, did)?; 120 126 121 127 Ok(()) 122 128 } ··· 310 316 records_delta += 1; 311 317 *collection_deltas.entry(collection).or_default() += 1; 312 318 } 319 319 + #[cfg(feature = "backlinks")] 320 320 + if let Ok(value) = serde_ipld_dagcbor::from_slice::<Data>(bytes.as_ref()) { 321 321 + crate::backlinks::store::index_record( 322 322 + batch, 323 323 + &db.backlinks, 324 324 + did.as_str(), 325 325 + collection, 326 326 + &rkey.to_smolstr(), 327 327 + &value, 328 328 + )?; 329 329 + } 313 330 None 314 331 } else if action == DbAction::Create || action == DbAction::Update { 315 332 Some(bytes.clone()) ··· 324 341 // accumulate counts 325 342 records_delta -= 1; 326 343 *collection_deltas.entry(collection).or_default() -= 1; 344 344 + 345 345 + #[cfg(feature = "backlinks")] 346 346 + crate::backlinks::store::delete_record( 347 347 + batch, 348 348 + &db.backlinks, 349 349 + did.as_str(), 350 350 + collection, 351 351 + &rkey.to_smolstr(), 352 352 + )?; 327 353 } 328 354 329 355 None

+261

tests/backlinks_test.nu

reviewed

··· 1 1 + #!/usr/bin/env nu 2 2 + # tests the blue.microcosm.links backlinks API endpoints. 3 3 + # 4 4 + # backfills a single repo and verifies: 5 5 + # 1. getBacklinks returns a result for a subject URI extracted from an indexed like record 6 6 + # 2. getBacklinksCount is >= number of getBacklinks results for the same subject 7 7 + # 3. cursor pagination returns each result exactly once with no duplicates 8 8 + # 4. source filter restricts results to the specified collection 9 9 + # 5. reverse=true returns the same set as forward, with inverted order verified when 10 10 + # a subject with 2+ backlinks is found 11 11 + # 12 12 + # usage: nu tests/backlinks_test.nu 13 13 + use common.nu * 14 14 + 15 15 + # paginate through all backlinks for subject+source using limit=2, return all entries 16 16 + def get-all-backlinks [url: string, subject: string, source: string] { 17 17 + mut all = [] 18 18 + mut cursor = null 19 19 + loop { 20 20 + let params = if $cursor != null { 21 21 + $"subject=($subject | url encode)&source=($source)&limit=2&cursor=($cursor)" 22 22 + } else { 23 23 + $"subject=($subject | url encode)&source=($source)&limit=2" 24 24 + } 25 25 + let resp = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?($params)") 26 26 + $all = ($all | append $resp.backlinks) 27 27 + let next = ($resp | get --optional cursor) 28 28 + if ($next | is-empty) { break } 29 29 + $cursor = $next 30 30 + } 31 31 + $all 32 32 + } 33 33 + 34 34 + # scan up to `limit` reply posts looking for a thread root that has 2+ backlinks 35 35 + # from app.bsky.feed.post. returns a record {subject, count} or null. 36 36 + def find-multi-backlink-subject [url: string, did: string, limit: int] { 37 37 + let posts_resp = (try { 38 38 + http get $"($url)/xrpc/com.atproto.repo.listRecords?repo=($did)&collection=app.bsky.feed.post&limit=($limit)" 39 39 + } catch { 40 40 + return null 41 41 + }) 42 42 + 43 43 + for post in $posts_resp.records { 44 44 + let reply = ($post.value | get --optional reply) 45 45 + if ($reply | is-empty) { continue } 46 46 + let root_uri = $reply.root.uri 47 47 + let count_resp = (try { 48 48 + http get $"($url)/xrpc/blue.microcosm.links.getBacklinksCount?subject=($root_uri | url encode)" 49 49 + } catch { 50 50 + continue 51 51 + }) 52 52 + if $count_resp.count >= 2 { 53 53 + return {subject: $root_uri, count: $count_resp.count} 54 54 + } 55 55 + } 56 56 + 57 57 + null 58 58 + } 59 59 + 60 60 + # verify getBacklinks and getBacklinksCount return sensible results for sampled likes. 61 61 + # returns an error string on failure, or empty string on success. 62 62 + def check-basic [url: string, likes: list] { 63 63 + print "checking basic backlinks and count..." 64 64 + 65 65 + for like in $likes { 66 66 + let subject = $like.value.subject.uri 67 67 + print $" subject: ($subject)" 68 68 + 69 69 + let bl_resp = (try { 70 70 + http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&source=app.bsky.feed.like" 71 71 + } catch { 72 72 + return $"getBacklinks request failed for ($subject)" 73 73 + }) 74 74 + 75 75 + if ($bl_resp.backlinks | is-empty) { 76 76 + return $"expected at least 1 backlink for ($subject)" 77 77 + } 78 78 + let bl_count = ($bl_resp.backlinks | length) 79 79 + print $" getBacklinks: ($bl_count) results" 80 80 + 81 81 + let ct_resp = (try { 82 82 + http get $"($url)/xrpc/blue.microcosm.links.getBacklinksCount?subject=($subject | url encode)&source=app.bsky.feed.like" 83 83 + } catch { 84 84 + return $"getBacklinksCount request failed for ($subject)" 85 85 + }) 86 86 + 87 87 + # count may exceed backlinks results because stale index entries (records deleted 88 88 + # after indexing) are counted but skipped during result collection 89 89 + if $ct_resp.count < ($bl_resp.backlinks | length) { 90 90 + return $"count ($ct_resp.count) < backlinks result count ($bl_resp.backlinks | length) for ($subject)" 91 91 + } 92 92 + print $" getBacklinksCount: ($ct_resp.count)" 93 93 + } 94 94 + 95 95 + "" 96 96 + } 97 97 + 98 98 + # verify that cursor pagination returns every result exactly once with no duplicates. 99 99 + # returns an error string on failure, or empty string on success. 100 100 + def check-pagination [url: string, likes: list] { 101 101 + print "checking cursor pagination..." 102 102 + let subject = ($likes | first).value.subject.uri 103 103 + print $" subject: ($subject)" 104 104 + 105 105 + # fetch all in a single shot with a large limit 106 106 + let full = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&source=app.bsky.feed.like&limit=100") 107 107 + let full_uris = ($full.backlinks | each { |b| $b.uri } | sort) 108 108 + 109 109 + # paginate through the same subject with limit=2 110 110 + let paged = (get-all-backlinks $url $subject "app.bsky.feed.like") 111 111 + let paged_uris = ($paged | each { |b| $b.uri } | sort) 112 112 + 113 113 + print $" full fetch: ($full_uris | length), paginated: ($paged_uris | length)" 114 114 + 115 115 + if $full_uris != $paged_uris { 116 116 + return "paginated results differ from single-page results" 117 117 + } 118 118 + 119 119 + let unique_count = ($paged_uris | uniq | length) 120 120 + if $unique_count != ($paged_uris | length) { 121 121 + return "duplicate URIs found in paginated results" 122 122 + } 123 123 + 124 124 + print " pagination OK" 125 125 + "" 126 126 + } 127 127 + 128 128 + # verify source filter and that forward/reverse return the same set. 129 129 + # returns an error string on failure, or empty string on success. 130 130 + def check-source-filter [url: string, likes: list] { 131 131 + print "checking source filter..." 132 132 + let subject = ($likes | first).value.subject.uri 133 133 + print $" subject: ($subject)" 134 134 + 135 135 + let fwd = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&source=app.bsky.feed.like&limit=50") 136 136 + let rev = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&source=app.bsky.feed.like&limit=50&reverse=true") 137 137 + 138 138 + # source filter: all returned URIs must belong to app.bsky.feed.like 139 139 + let bad = ($fwd.backlinks | where { |b| not ($b.uri | str contains "/app.bsky.feed.like/") }) 140 140 + if not ($bad | is-empty) { 141 141 + return $"source filter returned non-like records: ($bad)" 142 142 + } 143 143 + let fwd_count = ($fwd.backlinks | length) 144 144 + print $" source filter OK: ($fwd_count) likes" 145 145 + 146 146 + # reverse must return the same set 147 147 + let fwd_sorted = ($fwd.backlinks | each { |b| $b.uri } | sort) 148 148 + let rev_sorted = ($rev.backlinks | each { |b| $b.uri } | sort) 149 149 + if $fwd_sorted != $rev_sorted { 150 150 + return "forward and reverse scans returned different sets" 151 151 + } 152 152 + 153 153 + print " reverse set equality OK" 154 154 + "" 155 155 + } 156 156 + 157 157 + # verify that reverse=true actually inverts the order using a subject with 2+ backlinks. 158 158 + # returns an error string on failure, or empty string on success. 159 159 + def check-reverse-ordering [url: string, subject: string, expected_count: int] { 160 160 + print $"checking reverse ordering — subject has ($expected_count) backlinks..." 161 161 + print $" subject: ($subject)" 162 162 + 163 163 + let fwd = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&limit=50") 164 164 + let rev = (http get $"($url)/xrpc/blue.microcosm.links.getBacklinks?subject=($subject | url encode)&limit=50&reverse=true") 165 165 + 166 166 + let fwd_uris = ($fwd.backlinks | each { |b| $b.uri }) 167 167 + let rev_uris = ($rev.backlinks | each { |b| $b.uri }) 168 168 + 169 169 + if ($fwd_uris | length) < 2 { 170 170 + return $"expected >= 2 forward results for ordering test, got ($fwd_uris | length)" 171 171 + } 172 172 + 173 173 + if $fwd_uris != ($rev_uris | reverse) { 174 174 + return "reverse order is not the inverse of forward order" 175 175 + } 176 176 + 177 177 + print $" order inversion verified with ($fwd_uris | length) results" 178 178 + "" 179 179 + } 180 180 + 181 181 + def main [] { 182 182 + let did = "did:plc:dfl62fgb7wtjj3fcbb72naae" 183 183 + let port = 3020 184 184 + let url = $"http://localhost:($port)" 185 185 + let db_path = (mktemp -d -t hydrant_backlinks_test.XXXXXX) 186 186 + 187 187 + print $"database path: ($db_path)" 188 188 + 189 189 + let binary = (build-hydrant-features "backlinks") 190 190 + let instance = (start-hydrant $binary $db_path $port) 191 191 + 192 192 + if not (wait-for-api $url) { 193 193 + print "ERROR: hydrant failed to start" 194 194 + try { kill -9 $instance.pid } 195 195 + rm -rf $db_path 196 196 + exit 1 197 197 + } 198 198 + 199 199 + print $"adding ($did) to tracking..." 200 200 + http put -t application/json $"($url)/repos" [{ did: $did }] 201 201 + 202 202 + if not (wait-for-backfill $url) { 203 203 + print "ERROR: backfill timed out or failed" 204 204 + try { kill -9 $instance.pid } 205 205 + rm -rf $db_path 206 206 + exit 1 207 207 + } 208 208 + 209 209 + # fetch a small set of like records to use as test subjects 210 210 + print "fetching likes from indexed repo..." 211 211 + let likes_resp = (try { 212 212 + http get $"($url)/xrpc/com.atproto.repo.listRecords?repo=($did)&collection=app.bsky.feed.like&limit=5" 213 213 + } catch { 214 214 + print "ERROR: could not fetch like records" 215 215 + try { kill -9 $instance.pid } 216 216 + rm -rf $db_path 217 217 + exit 1 218 218 + }) 219 219 + 220 220 + let likes = $likes_resp.records 221 221 + if ($likes | is-empty) { 222 222 + print "SKIP: no like records found for test DID. cannot verify backlinks" 223 223 + try { kill -9 $instance.pid } 224 224 + rm -rf $db_path 225 225 + exit 0 226 226 + } 227 227 + print $"found ($likes | length) likes for testing" 228 228 + 229 229 + # find a thread root with 2+ backlinks for ordering verification 230 230 + print "scanning posts for a thread root with 2+ backlinks..." 231 231 + let multi = (find-multi-backlink-subject $url $did 100) 232 232 + if ($multi | is-empty) { 233 233 + print " note: no thread root with 2+ backlinks found, ordering verification will be skipped" 234 234 + } else { 235 235 + print $" found subject with ($multi.count) backlinks: ($multi.subject)" 236 236 + } 237 237 + 238 238 + let basic_err = (check-basic $url $likes) 239 239 + let paging_err = (check-pagination $url $likes) 240 240 + let srcfilt_err = (check-source-filter $url $likes) 241 241 + let ordering_err = if not ($multi | is-empty) { 242 242 + check-reverse-ordering $url $multi.subject $multi.count 243 243 + } else { 244 244 + print "skipping reverse ordering check (no multi-backlink subject found)" 245 245 + "" 246 246 + } 247 247 + 248 248 + let failed = ([$basic_err, $paging_err, $srcfilt_err, $ordering_err] | where { |e| not ($e | is-empty) }) 249 249 + 250 250 + print "stopping hydrant..." 251 251 + try { kill -9 $instance.pid } 252 252 + rm -rf $db_path 253 253 + 254 254 + if ($failed | is-empty) { 255 255 + print "all backlinks tests PASSED" 256 256 + exit 0 257 257 + } else { 258 258 + for err in $failed { print $"FAILED: ($err)" } 259 259 + exit 1 260 260 + } 261 261 + }

+7

tests/common.nu

reviewed

··· 56 56 "target/debug/hydrant" 57 57 } 58 58 59 59 + # build the hydrant binary with extra cargo features (space-separated string) 60 60 + export def build-hydrant-features [features: string] { 61 61 + print $"building hydrant with features: ($features)..." 62 62 + cargo build --features $features 63 63 + "target/debug/hydrant" 64 64 + } 65 65 + 59 66 # start hydrant in the background 60 67 export def start-hydrant [binary: string, db_path: string, port: int] { 61 68 let log_file = $"($db_path)/hydrant.log"