tangled.org / core
Monorepo for Tangled tangled.org
1
Fork 31

Configure Feed

Select the types of activity you want to include in your feed.

bobbin/xrpc,bobbin/types/lexicons: implement query.enrichResponse

Signed-off-by: dawn <dawn@tangled.org>

author
dawn date (Jul 23, 2026, 6:50 PM +0300) commit 6e01d5d9 parent 97cc2c7a change-id tyzttztr
+1380 -175
+1 -1
bobbin/crates/xrpc/Cargo.toml
··· 23 23 serde = { workspace = true } 24 24 serde_json = { workspace = true } 25 25 thiserror = { workspace = true } 26 + tower = { workspace = true } 26 27 tower-http = { workspace = true } 27 28 tracing = { workspace = true } 28 29 url = { workspace = true } ··· 31 32 bobbin-runtime = { workspace = true } 32 33 http = { workspace = true } 33 34 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } 34 - tower = { workspace = true } 35 35 url = { workspace = true } 36 36 wiremock = { workspace = true }
+372
bobbin/crates/xrpc/src/enrich.rs
··· 1 + use std::collections::HashSet; 2 + 3 + use axum::{ 4 + Json, 5 + body::{Body, to_bytes}, 6 + extract::State, 7 + http::{Request, StatusCode}, 8 + response::Response, 9 + }; 10 + use bobbin_types::ids::{EdgeKey, SubjectRef, nsid_static}; 11 + use jacquard_common::DefaultStr; 12 + use jacquard_common::types::did::Did; 13 + use jacquard_common::types::ident::AtIdentifier; 14 + use jacquard_common::types::string::AtUri; 15 + use serde::Deserialize; 16 + use serde_json::{Map, Value, json}; 17 + use tower::ServiceExt; 18 + 19 + use crate::recordpath::{parse_record_path, walk_path}; 20 + use crate::{AppState, SubjectShape, XrpcError, mirror_kind, subject_shape}; 21 + 22 + #[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] 23 + #[serde(rename_all = "camelCase")] 24 + pub enum Aggregation { 25 + Count, 26 + DistinctAuthors, 27 + Viewer, 28 + } 29 + 30 + impl Aggregation { 31 + fn key(self) -> &'static str { 32 + match self { 33 + Aggregation::Count => "count", 34 + Aggregation::DistinctAuthors => "distinctAuthors", 35 + Aggregation::Viewer => "viewer", 36 + } 37 + } 38 + } 39 + 40 + #[derive(Debug, Deserialize)] 41 + pub struct LinkDescriptor { 42 + /// constellation link source 43 + source: String, 44 + #[serde(rename = "type")] 45 + aggregation: Aggregation, 46 + } 47 + 48 + impl LinkDescriptor { 49 + fn parts(&self) -> Result<(&str, &str, Aggregation), XrpcError> { 50 + match self.source.split_once(':') { 51 + Some((collection, path)) if !collection.is_empty() && !path.is_empty() => { 52 + Ok((collection, path, self.aggregation)) 53 + } 54 + _ => Err(XrpcError::InvalidParams(format!( 55 + "enrich {:?}: expected \"collection:path\"", 56 + self.source 57 + ))), 58 + } 59 + } 60 + } 61 + 62 + #[derive(Debug, Deserialize)] 63 + pub struct EnrichInput { 64 + xrpc: String, 65 + #[serde(default)] 66 + params: Option<Map<String, Value>>, 67 + enrich: Vec<LinkDescriptor>, 68 + #[serde(default)] 69 + sources: Option<Vec<String>>, 70 + #[serde(default)] 71 + viewer: Option<Did>, 72 + } 73 + 74 + pub async fn enrich( 75 + State(state): State<AppState>, 76 + Json(input): Json<EnrichInput>, 77 + ) -> Result<Json<Value>, XrpcError> { 78 + for descriptor in &input.enrich { 79 + validate_descriptor(descriptor)?; 80 + if descriptor.aggregation == Aggregation::Viewer && input.viewer.is_none() { 81 + return Err(descriptor_error( 82 + descriptor, 83 + "viewer aggregation requires a viewer param", 84 + )); 85 + } 86 + } 87 + 88 + let inner = run_inner(&state, &input.xrpc, input.params.unwrap_or_default()).await?; 89 + 90 + let mut refs: Vec<SubjectRef> = Vec::new(); 91 + let mut seen: HashSet<SubjectRef> = HashSet::new(); 92 + match &input.sources { 93 + Some(sources) => { 94 + for (i, path) in sources.iter().enumerate() { 95 + let segs = parse_record_path(path) 96 + .map_err(|e| XrpcError::InvalidParams(format!("sources[{i}] {path:?}: {e}")))?; 97 + for node in walk_path(&segs, [&inner]) { 98 + collect_ref(node, &mut refs, &mut seen); 99 + } 100 + } 101 + } 102 + None => discover_refs(&inner, &mut refs, &mut seen), 103 + } 104 + 105 + let mut stats = Map::new(); 106 + for reference in &refs { 107 + let mut per_ref = Map::new(); 108 + for descriptor in &input.enrich { 109 + let (.., aggregation) = descriptor.parts()?; 110 + let Some(kind) = resolve_descriptor(descriptor, reference)? else { 111 + continue; 112 + }; 113 + let key = EdgeKey::new(kind, reference.clone()); 114 + let result = match aggregation { 115 + Aggregation::Count => Value::from(state.edges.count(&key)), 116 + Aggregation::DistinctAuthors => { 117 + Value::from(state.edges.count_distinct_authors(&key)) 118 + } 119 + Aggregation::Viewer => { 120 + // viewer presence is validated up front 121 + let viewer = input.viewer.as_ref().expect("viewer param present"); 122 + match state.edges.viewer_source(&key, viewer.as_str()) { 123 + Some(uri) => Value::String(uri.to_string()), 124 + None => Value::Null, 125 + } 126 + } 127 + }; 128 + let entry = per_ref 129 + .entry(descriptor.source.clone()) 130 + .or_insert_with(|| Value::Object(Map::new())); 131 + if let Value::Object(counts) = entry { 132 + counts.insert(aggregation.key().to_owned(), result); 133 + } 134 + } 135 + if !per_ref.is_empty() { 136 + stats.insert(subject_string(reference), Value::Object(per_ref)); 137 + } 138 + } 139 + 140 + Ok(Json(json!({ "output": inner, "stats": stats }))) 141 + } 142 + 143 + fn descriptor_error(descriptor: &LinkDescriptor, msg: &str) -> XrpcError { 144 + XrpcError::InvalidParams(format!("enrich {}: {msg}", descriptor.source)) 145 + } 146 + 147 + fn validate_descriptor(descriptor: &LinkDescriptor) -> Result<(), XrpcError> { 148 + let (collection, path, _) = descriptor.parts()?; 149 + match path { 150 + "subject" => subject_shape(collection) 151 + .map(|_| ()) 152 + .ok_or_else(|| descriptor_error(descriptor, "unknown collection")), 153 + ".repo" => mirror_kind(collection) 154 + .map(|_| ()) 155 + .ok_or_else(|| descriptor_error(descriptor, "collection has no author index")), 156 + _ if path.starts_with('.') => Err(descriptor_error( 157 + descriptor, 158 + "envelope field not index-backed; try .repo", 159 + )), 160 + _ => Err(descriptor_error( 161 + descriptor, 162 + "path not index-backed; only `subject` and `.repo` are supported", 163 + )), 164 + } 165 + } 166 + 167 + /// the edge kind to look up for one reference, or None if the descriptor's shape 168 + /// doesn't apply to this ref kind, eg. a did-only descriptor asked about an at-uri 169 + fn resolve_descriptor( 170 + descriptor: &LinkDescriptor, 171 + reference: &SubjectRef, 172 + ) -> Result<Option<jacquard_common::types::nsid::Nsid<DefaultStr>>, XrpcError> { 173 + let (collection, path, _) = descriptor.parts()?; 174 + match path { 175 + "subject" => { 176 + let Some((nsid, shape)) = subject_shape(collection) else { 177 + return Err(descriptor_error(descriptor, "unknown collection")); 178 + }; 179 + Ok(shape_accepts(shape, reference).then(|| nsid_static(nsid))) 180 + } 181 + ".repo" => { 182 + let Some(kind) = mirror_kind(collection) else { 183 + return Err(descriptor_error( 184 + descriptor, 185 + "collection has no author index", 186 + )); 187 + }; 188 + Ok(matches!(reference, SubjectRef::Did(_)).then(|| nsid_static(kind))) 189 + } 190 + _ => unreachable!("validated up front"), 191 + } 192 + } 193 + 194 + /// mismatch means skip, not reject 195 + fn shape_accepts(shape: SubjectShape, reference: &SubjectRef) -> bool { 196 + match (shape, reference) { 197 + (SubjectShape::BareDid, SubjectRef::Did(_)) => true, 198 + (SubjectShape::Collection(expected), SubjectRef::Uri(uri)) => { 199 + uri.collection().is_some_and(|c| c.as_ref() == expected) 200 + } 201 + (SubjectShape::OneOfCollections(allowed), SubjectRef::Uri(uri)) 202 + | (SubjectShape::BareDidOrOneOfCollections(allowed), SubjectRef::Uri(uri)) => uri 203 + .collection() 204 + .is_some_and(|c| allowed.contains(&c.as_ref())), 205 + (SubjectShape::BareDidOrOneOfCollections(_), SubjectRef::Did(_)) => true, 206 + (SubjectShape::AnyAtUri, SubjectRef::Uri(_)) => true, 207 + _ => false, 208 + } 209 + } 210 + 211 + fn subject_string(reference: &SubjectRef) -> String { 212 + match reference { 213 + SubjectRef::Did(d) => d.as_ref().to_owned(), 214 + SubjectRef::Uri(u) => u.as_ref().to_owned(), 215 + } 216 + } 217 + 218 + /// a value counts as a reference if it is a did string or an at-uri string 219 + /// with collection and rkey, or a {uri, cid} strong ref 220 + fn collect_ref(value: &Value, refs: &mut Vec<SubjectRef>, seen: &mut HashSet<SubjectRef>) { 221 + let candidate = match value { 222 + Value::String(s) => Some(s.as_str()), 223 + Value::Object(o) => o 224 + .get("uri") 225 + .and_then(Value::as_str) 226 + .filter(|_| o.contains_key("cid")), 227 + _ => None, 228 + }; 229 + let Some(candidate) = candidate else { return }; 230 + let reference = if candidate.starts_with("did:") { 231 + Did::<DefaultStr>::new_owned(candidate) 232 + .ok() 233 + .map(SubjectRef::Did) 234 + } else if candidate.starts_with("at://") { 235 + AtUri::<DefaultStr>::new_owned(candidate) 236 + .ok() 237 + .filter(|u| { 238 + matches!(u.authority(), AtIdentifier::Did(_)) 239 + && u.collection().is_some() 240 + && u.rkey().is_some() 241 + }) 242 + .map(SubjectRef::Uri) 243 + } else { 244 + None 245 + }; 246 + if let Some(reference) = reference 247 + && seen.insert(reference.clone()) 248 + { 249 + refs.push(reference); 250 + } 251 + } 252 + 253 + fn discover_refs(value: &Value, refs: &mut Vec<SubjectRef>, seen: &mut HashSet<SubjectRef>) { 254 + collect_ref(value, refs, seen); 255 + match value { 256 + Value::Array(items) => { 257 + for item in items { 258 + discover_refs(item, refs, seen); 259 + } 260 + } 261 + Value::Object(map) => { 262 + for v in map.values() { 263 + discover_refs(v, refs, seen); 264 + } 265 + } 266 + _ => {} 267 + } 268 + } 269 + 270 + async fn run_inner( 271 + state: &AppState, 272 + nsid: &str, 273 + params: Map<String, Value>, 274 + ) -> Result<Value, XrpcError> { 275 + let qs = encode_params(&params); 276 + let uri = format!("/xrpc/{nsid}?{qs}"); 277 + let request = Request::builder() 278 + .method("GET") 279 + .uri(&uri) 280 + .body(Body::empty()) 281 + .map_err(|e| XrpcError::Internal(format!("inner request: {e}")))?; 282 + let response = state 283 + .self_router() 284 + .oneshot(request) 285 + .await 286 + .map_err(|e| XrpcError::Internal(format!("inner dispatch: {e}")))?; 287 + if response.status() == StatusCode::NOT_FOUND { 288 + let bytes = to_bytes(response.into_body(), usize::MAX) 289 + .await 290 + .map_err(|e| XrpcError::Internal(format!("inner response: {e}")))?; 291 + return Err(bytes 292 + .is_empty() 293 + .then(|| XrpcError::InvalidParams(format!("unknown or unenrichable query: {nsid}"))) 294 + .unwrap_or(XrpcError::NotFound)); 295 + } 296 + finish(response).await 297 + } 298 + 299 + fn encode_params(params: &Map<String, Value>) -> String { 300 + let mut out = url::form_urlencoded::Serializer::new(String::new()); 301 + for (key, value) in params { 302 + match value { 303 + Value::Array(items) => { 304 + for item in items { 305 + if let Some(scalar) = scalar_str(item) { 306 + out.append_pair(key, &scalar); 307 + } 308 + } 309 + } 310 + _ => { 311 + if let Some(scalar) = scalar_str(value) { 312 + out.append_pair(key, &scalar); 313 + } 314 + } 315 + } 316 + } 317 + out.finish() 318 + } 319 + 320 + fn scalar_str(value: &Value) -> Option<String> { 321 + match value { 322 + Value::String(s) => Some(s.clone()), 323 + Value::Number(n) => Some(n.to_string()), 324 + Value::Bool(b) => Some(b.to_string()), 325 + _ => None, 326 + } 327 + } 328 + 329 + async fn finish(resp: Response) -> Result<Value, XrpcError> { 330 + let status = resp.status(); 331 + let bytes = to_bytes(resp.into_body(), usize::MAX) 332 + .await 333 + .map_err(|e| XrpcError::Internal(format!("inner response: {e}")))?; 334 + if !status.is_success() { 335 + let msg = String::from_utf8_lossy(&bytes).into_owned(); 336 + return Err(match status { 337 + StatusCode::BAD_REQUEST => XrpcError::InvalidParams(msg), 338 + StatusCode::NOT_FOUND => XrpcError::NotFound, 339 + StatusCode::SERVICE_UNAVAILABLE => XrpcError::Overloaded, 340 + _ => XrpcError::UpstreamUnavailable(format!("inner query ({status}): {msg}")), 341 + }); 342 + } 343 + serde_json::from_slice(&bytes) 344 + .map_err(|e| XrpcError::Internal(format!("inner response decode: {e}"))) 345 + } 346 + 347 + #[cfg(test)] 348 + mod tests { 349 + use super::*; 350 + use serde_json::json; 351 + 352 + #[test] 353 + fn collects_dids_uris_and_strong_refs() { 354 + let mut refs = Vec::new(); 355 + let mut seen = HashSet::new(); 356 + collect_ref(&json!("did:plc:abc"), &mut refs, &mut seen); 357 + collect_ref( 358 + &json!("at://did:plc:abc/sh.tangled.repo/x"), 359 + &mut refs, 360 + &mut seen, 361 + ); 362 + collect_ref( 363 + &json!({"uri": "at://did:plc:abc/sh.tangled.repo/y", "cid": "bafy"}), 364 + &mut refs, 365 + &mut seen, 366 + ); 367 + collect_ref(&json!("at://did:plc:abc"), &mut refs, &mut seen); 368 + collect_ref(&json!("oppi.li"), &mut refs, &mut seen); 369 + collect_ref(&json!("did:plc:abc"), &mut refs, &mut seen); 370 + assert_eq!(refs.len(), 3); 371 + } 372 + }
+68 -174
bobbin/crates/xrpc/src/lib.rs
··· 95 95 use tracing::{Level, Span}; 96 96 97 97 mod backpressure; 98 + mod enrich; 98 99 mod filter; 100 + mod recordpath; 99 101 100 102 pub use backpressure::{ 101 103 HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor, ··· 117 119 pub search: Arc<dyn SearchReader>, 118 120 pub resolver: Arc<RepoIdResolver>, 119 121 pub limiter: Option<Arc<HeavyLimiter>>, 122 + enrich_router: Arc<std::sync::OnceLock<Router>>, 120 123 } 121 124 122 125 impl AppState { ··· 143 146 search, 144 147 resolver, 145 148 limiter: None, 149 + enrich_router: Arc::new(std::sync::OnceLock::new()), 146 150 } 147 151 } 148 152 149 153 pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self { 150 154 self.limiter = limiter; 151 155 self 156 + } 157 + 158 + /// for internal xrpc dispatch [`enrich`] 159 + pub fn self_router(&self) -> Router { 160 + self.enrich_router 161 + .get_or_init(|| router(self.clone())) 162 + .clone() 152 163 } 153 164 154 165 fn heavy_permit(&self) -> Result<Option<HeavyPermit>, XrpcError> { ··· 385 396 .route("/xrpc/sh.tangled.string.listStrings", get(list_strings)) 386 397 .route("/xrpc/sh.tangled.string.countStrings", get(count_strings)) 387 398 .route("/xrpc/sh.tangled.search.query", get(search_query)) 399 + .route( 400 + "/xrpc/sh.tangled.query.enrichResponse", 401 + axum::routing::post(enrich::enrich), 402 + ) 388 403 .route("/xrpc/sh.tangled.bobbin.getCoverage", get(get_coverage)) 389 404 .route( 390 405 "/xrpc/com.bad-example.identity.resolveMiniDoc", ··· 916 931 const SHAPE: SubjectShape; 917 932 } 918 933 919 - pub struct StarBy; 920 - pub struct ReactionBy; 921 - pub struct FollowBy; 922 - pub struct VouchBy; 923 - pub struct RefUpdateBy; 924 - pub struct KnotMemberBy; 925 - pub struct LabelOpBy; 926 - pub struct PipelineBy; 927 - pub struct PipelineStatusBy; 928 - pub struct ArtifactBy; 929 - pub struct CollaboratorBy; 930 - pub struct FeedCommentBy; 931 - pub struct IssueBy; 932 - pub struct IssueStateBy; 933 - pub struct PullBy; 934 - pub struct PullStatusBy; 935 - pub struct SpindleMemberBy; 934 + macro_rules! edge_kinds { 935 + ($($collection:literal => $record:ty, $shape:expr $(, mirror $by:ident)? ;)*) => { 936 + $( 937 + impl HasSubject for $record { 938 + const SHAPE: SubjectShape = $shape; 939 + } 940 + )* 941 + $($( 942 + pub struct $by; 943 + impl MirrorOf for $by { 944 + type Record = $record; 945 + const EDGE_KIND: &'static str = concat!($collection, ".by"); 946 + const SHAPE: SubjectShape = SubjectShape::BareDid; 947 + } 948 + )?)* 936 949 937 - impl MirrorOf for StarBy { 938 - type Record = StarRecord; 939 - const EDGE_KIND: &'static str = "sh.tangled.feed.star.by"; 940 - const SHAPE: SubjectShape = SubjectShape::BareDid; 941 - } 942 - impl MirrorOf for ReactionBy { 943 - type Record = ReactionRecord; 944 - const EDGE_KIND: &'static str = "sh.tangled.feed.reaction.by"; 945 - const SHAPE: SubjectShape = SubjectShape::BareDid; 946 - } 947 - impl MirrorOf for FollowBy { 948 - type Record = FollowRecord; 949 - const EDGE_KIND: &'static str = "sh.tangled.graph.follow.by"; 950 - const SHAPE: SubjectShape = SubjectShape::BareDid; 951 - } 952 - impl MirrorOf for VouchBy { 953 - type Record = VouchRecord; 954 - const EDGE_KIND: &'static str = "sh.tangled.graph.vouch.by"; 955 - const SHAPE: SubjectShape = SubjectShape::BareDid; 956 - } 957 - impl MirrorOf for RefUpdateBy { 958 - type Record = RefUpdateRecord; 959 - const EDGE_KIND: &'static str = "sh.tangled.git.refUpdate.by"; 960 - const SHAPE: SubjectShape = SubjectShape::BareDid; 961 - } 962 - impl MirrorOf for KnotMemberBy { 963 - type Record = KnotMemberRecord; 964 - const EDGE_KIND: &'static str = "sh.tangled.knot.member.by"; 965 - const SHAPE: SubjectShape = SubjectShape::BareDid; 966 - } 967 - impl MirrorOf for LabelOpBy { 968 - type Record = LabelOpRecord; 969 - const EDGE_KIND: &'static str = "sh.tangled.label.op.by"; 970 - const SHAPE: SubjectShape = SubjectShape::BareDid; 971 - } 972 - impl MirrorOf for PipelineBy { 973 - type Record = PipelineRecord; 974 - const EDGE_KIND: &'static str = "sh.tangled.pipeline.by"; 975 - const SHAPE: SubjectShape = SubjectShape::BareDid; 976 - } 977 - impl MirrorOf for PipelineStatusBy { 978 - type Record = PipelineStatusRecord; 979 - const EDGE_KIND: &'static str = "sh.tangled.pipeline.status.by"; 980 - const SHAPE: SubjectShape = SubjectShape::BareDid; 981 - } 982 - impl MirrorOf for ArtifactBy { 983 - type Record = ArtifactRecord; 984 - const EDGE_KIND: &'static str = "sh.tangled.repo.artifact.by"; 985 - const SHAPE: SubjectShape = SubjectShape::BareDid; 986 - } 987 - impl MirrorOf for CollaboratorBy { 988 - type Record = CollaboratorRecord; 989 - const EDGE_KIND: &'static str = "sh.tangled.repo.collaborator.by"; 990 - const SHAPE: SubjectShape = SubjectShape::BareDid; 991 - } 992 - impl MirrorOf for FeedCommentBy { 993 - type Record = FeedCommentRecord; 994 - const EDGE_KIND: &'static str = "sh.tangled.feed.comment.by"; 995 - const SHAPE: SubjectShape = SubjectShape::BareDid; 996 - } 997 - impl MirrorOf for IssueBy { 998 - type Record = IssueRecord; 999 - const EDGE_KIND: &'static str = "sh.tangled.repo.issue.by"; 1000 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1001 - } 1002 - impl MirrorOf for IssueStateBy { 1003 - type Record = IssueStateRecord; 1004 - const EDGE_KIND: &'static str = "sh.tangled.repo.issue.state.by"; 1005 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1006 - } 1007 - impl MirrorOf for PullBy { 1008 - type Record = PullRecord; 1009 - const EDGE_KIND: &'static str = "sh.tangled.repo.pull.by"; 1010 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1011 - } 1012 - impl MirrorOf for PullStatusBy { 1013 - type Record = PullStatusRecord; 1014 - const EDGE_KIND: &'static str = "sh.tangled.repo.pull.status.by"; 1015 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1016 - } 1017 - impl MirrorOf for SpindleMemberBy { 1018 - type Record = SpindleMemberRecord; 1019 - const EDGE_KIND: &'static str = "sh.tangled.spindle.member.by"; 1020 - const SHAPE: SubjectShape = SubjectShape::BareDid; 950 + pub(crate) fn subject_shape(collection: &str) -> Option<(&'static str, SubjectShape)> { 951 + Some(match collection { 952 + $($collection => ($collection, <$record as HasSubject>::SHAPE),)* 953 + _ => return None, 954 + }) 955 + } 956 + 957 + pub(crate) fn mirror_kind(collection: &str) -> Option<&'static str> { 958 + Some(match collection { 959 + $($($collection => <$by as MirrorOf>::EDGE_KIND,)?)* 960 + _ => return None, 961 + }) 962 + } 963 + }; 1021 964 } 1022 965 1023 - impl HasSubject for StarRecord { 1024 - const SHAPE: SubjectShape = SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]); 1025 - } 1026 - impl HasSubject for FollowRecord { 1027 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1028 - } 1029 - impl HasSubject for IssueRecord { 1030 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1031 - } 1032 - impl HasSubject for PullRecord { 1033 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1034 - } 1035 - impl HasSubject for FeedCommentRecord { 1036 - const SHAPE: SubjectShape = SubjectShape::OneOfCollections(&[ 1037 - "sh.tangled.repo.issue", 1038 - "sh.tangled.repo.pull", 1039 - "sh.tangled.string", 1040 - ]); 1041 - } 1042 - impl HasSubject for LabelDefinitionRecord { 1043 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1044 - } 1045 - impl HasSubject for LabelOpRecord { 1046 - const SHAPE: SubjectShape = 1047 - SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]); 1048 - } 1049 - impl HasSubject for PipelineRecord { 1050 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1051 - } 1052 - impl HasSubject for PipelineStatusRecord { 1053 - const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.pipeline"); 1054 - } 1055 - impl HasSubject for ArtifactRecord { 1056 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1057 - } 1058 - impl HasSubject for KnotMemberRecord { 1059 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1060 - } 1061 - impl HasSubject for SpindleMemberRecord { 1062 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1063 - } 1064 - impl HasSubject for TangledStringRecord { 1065 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1066 - } 1067 - impl HasSubject for ReactionRecord { 1068 - const SHAPE: SubjectShape = SubjectShape::AnyAtUri; 1069 - } 1070 - impl HasSubject for RefUpdateRecord { 1071 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1072 - } 1073 - impl HasSubject for CollaboratorRecord { 1074 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1075 - } 1076 - impl HasSubject for IssueStateRecord { 1077 - const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.issue"); 1078 - } 1079 - impl HasSubject for PullStatusRecord { 1080 - const SHAPE: SubjectShape = SubjectShape::Collection("sh.tangled.repo.pull"); 1081 - } 1082 - impl HasSubject for KnotRecord { 1083 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1084 - } 1085 - impl HasSubject for SpindleRecord { 1086 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1087 - } 1088 - impl HasSubject for PublicKeyRecord { 1089 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1090 - } 1091 - impl HasSubject for RepoRecord { 1092 - const SHAPE: SubjectShape = SubjectShape::BareDid; 1093 - } 1094 - impl HasSubject for VouchRecord { 1095 - const SHAPE: SubjectShape = SubjectShape::BareDid; 966 + edge_kinds! { 967 + "sh.tangled.feed.star" => StarRecord, SubjectShape::BareDidOrOneOfCollections(&["sh.tangled.string"]), mirror StarBy; 968 + "sh.tangled.feed.comment" => FeedCommentRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull", "sh.tangled.string"]), mirror FeedCommentBy; 969 + "sh.tangled.feed.reaction" => ReactionRecord, SubjectShape::AnyAtUri, mirror ReactionBy; 970 + "sh.tangled.graph.follow" => FollowRecord, SubjectShape::BareDid, mirror FollowBy; 971 + "sh.tangled.graph.vouch" => VouchRecord, SubjectShape::BareDid, mirror VouchBy; 972 + "sh.tangled.git.refUpdate" => RefUpdateRecord, SubjectShape::BareDid, mirror RefUpdateBy; 973 + "sh.tangled.knot" => KnotRecord, SubjectShape::BareDid; 974 + "sh.tangled.knot.member" => KnotMemberRecord, SubjectShape::BareDid, mirror KnotMemberBy; 975 + "sh.tangled.label.definition" => LabelDefinitionRecord, SubjectShape::BareDid; 976 + "sh.tangled.label.op" => LabelOpRecord, SubjectShape::OneOfCollections(&["sh.tangled.repo.issue", "sh.tangled.repo.pull"]), mirror LabelOpBy; 977 + "sh.tangled.pipeline" => PipelineRecord, SubjectShape::BareDid, mirror PipelineBy; 978 + "sh.tangled.pipeline.status" => PipelineStatusRecord, SubjectShape::Collection("sh.tangled.pipeline"), mirror PipelineStatusBy; 979 + "sh.tangled.publicKey" => PublicKeyRecord, SubjectShape::BareDid; 980 + "sh.tangled.repo" => RepoRecord, SubjectShape::BareDid; 981 + "sh.tangled.repo.artifact" => ArtifactRecord, SubjectShape::BareDid, mirror ArtifactBy; 982 + "sh.tangled.repo.collaborator" => CollaboratorRecord, SubjectShape::BareDid, mirror CollaboratorBy; 983 + "sh.tangled.repo.issue" => IssueRecord, SubjectShape::BareDid, mirror IssueBy; 984 + "sh.tangled.repo.issue.state" => IssueStateRecord, SubjectShape::Collection("sh.tangled.repo.issue"), mirror IssueStateBy; 985 + "sh.tangled.repo.pull" => PullRecord, SubjectShape::BareDid, mirror PullBy; 986 + "sh.tangled.repo.pull.status" => PullStatusRecord, SubjectShape::Collection("sh.tangled.repo.pull"), mirror PullStatusBy; 987 + "sh.tangled.spindle" => SpindleRecord, SubjectShape::BareDid; 988 + "sh.tangled.spindle.member" => SpindleMemberRecord, SubjectShape::BareDid, mirror SpindleMemberBy; 989 + "sh.tangled.string" => TangledStringRecord, SubjectShape::BareDid; 1096 990 } 1097 991 1098 992 fn parse_subject(raw: &SubjectQuery, shape: SubjectShape) -> Result<SubjectRef, XrpcError> {
+301
bobbin/crates/xrpc/src/recordpath.rs
··· 1 + //! subset of microcosm.blue/RecordPath 2 + 3 + use serde_json::Value; 4 + 5 + #[derive(Debug, Clone, PartialEq)] 6 + pub(crate) enum Modifier { 7 + /// `[]` 8 + Elements, 9 + /// `[nsid]` 10 + UnionElements(String), 11 + /// `{nsid}` 12 + Union(String), 13 + } 14 + 15 + #[derive(Debug, Clone, PartialEq)] 16 + pub(crate) struct Segment { 17 + field: String, 18 + modifier: Option<Modifier>, 19 + } 20 + 21 + pub(crate) fn parse_record_path(path: &str) -> Result<Vec<Segment>, String> { 22 + let chars: Vec<char> = path.chars().collect(); 23 + let mut segments = Vec::new(); 24 + let mut field = String::new(); 25 + let mut i = 0; 26 + let mut saw_any = false; 27 + 28 + let unescape = |chars: &[char], i: &mut usize| -> Result<char, String> { 29 + *i += 1; 30 + match chars.get(*i) { 31 + Some(&c @ ('.' | '[' | ']' | '{' | '}' | '!')) => Ok(c), 32 + Some(c) => Err(format!("invalid escape !{c}")), 33 + None => Err("trailing ! escape".to_owned()), 34 + } 35 + }; 36 + 37 + while i < chars.len() { 38 + let c = chars[i]; 39 + match c { 40 + '.' => { 41 + if !saw_any { 42 + return Err("empty path segment".to_owned()); 43 + } 44 + segments.push(Segment { 45 + field: std::mem::take(&mut field), 46 + modifier: None, 47 + }); 48 + saw_any = false; 49 + i += 1; 50 + } 51 + '!' => { 52 + field.push(unescape(&chars, &mut i)?); 53 + saw_any = true; 54 + i += 1; 55 + } 56 + '[' | '{' => { 57 + let (close, is_union_brace) = if c == '[' { (']', false) } else { ('}', true) }; 58 + let mut inner = String::new(); 59 + i += 1; 60 + loop { 61 + match chars.get(i) { 62 + None => return Err(format!("unclosed {c}")), 63 + Some(&close_c) if close_c == close => break, 64 + Some(&'!') => inner.push(unescape(&chars, &mut i)?), 65 + Some(&ch) => inner.push(ch), 66 + } 67 + i += 1; 68 + } 69 + i += 1; // consume closer 70 + let modifier = match (is_union_brace, inner.is_empty()) { 71 + (false, true) => Modifier::Elements, 72 + (false, false) => Modifier::UnionElements(inner), 73 + (true, true) => return Err("empty {} union ref".to_owned()), 74 + (true, false) => Modifier::Union(inner), 75 + }; 76 + if field.is_empty() && !saw_any { 77 + return Err("modifier without a field".to_owned()); 78 + } 79 + segments.push(Segment { 80 + field: std::mem::take(&mut field), 81 + modifier: Some(modifier), 82 + }); 83 + saw_any = false; 84 + // a modified segment must end the path or be followed by '.' 85 + match chars.get(i) { 86 + None => break, 87 + Some('.') => { 88 + i += 1; 89 + } 90 + Some(other) => { 91 + return Err(format!("expected '.' after {c}{close}, got {other:?}")); 92 + } 93 + } 94 + } 95 + _ => { 96 + field.push(c); 97 + saw_any = true; 98 + i += 1; 99 + } 100 + } 101 + } 102 + if saw_any { 103 + segments.push(Segment { 104 + field, 105 + modifier: None, 106 + }); 107 + } 108 + if segments.is_empty() { 109 + return Err("empty path".to_owned()); 110 + } 111 + Ok(segments) 112 + } 113 + 114 + pub(crate) fn type_tag(value: &Value) -> Option<&str> { 115 + value.get("$type").and_then(Value::as_str) 116 + } 117 + 118 + pub(crate) fn walk_path<'a>( 119 + segments: &[Segment], 120 + roots: impl IntoIterator<Item = &'a Value>, 121 + ) -> Vec<&'a Value> { 122 + let mut nodes: Vec<&Value> = roots.into_iter().collect(); 123 + for segment in segments { 124 + let mut next: Vec<&Value> = nodes 125 + .into_iter() 126 + .filter_map(|node| node.get(&segment.field)) 127 + .collect(); 128 + if let Some(modifier) = &segment.modifier { 129 + next = match modifier { 130 + Modifier::Elements => next 131 + .into_iter() 132 + .flat_map(|node| node.as_array().into_iter().flatten()) 133 + .collect(), 134 + Modifier::UnionElements(nsid) => next 135 + .into_iter() 136 + .flat_map(|node| node.as_array().into_iter().flatten()) 137 + .filter(|item| type_tag(item) == Some(nsid.as_str())) 138 + .collect(), 139 + Modifier::Union(nsid) => next 140 + .into_iter() 141 + .filter(|node| type_tag(node) == Some(nsid.as_str())) 142 + .collect(), 143 + }; 144 + } 145 + nodes = next; 146 + } 147 + nodes 148 + } 149 + 150 + #[cfg(test)] 151 + mod tests { 152 + use super::*; 153 + use serde_json::json; 154 + 155 + fn parse_ok(path: &str) -> Vec<Segment> { 156 + parse_record_path(path).unwrap_or_else(|e| panic!("{path}: {e}")) 157 + } 158 + 159 + #[test] 160 + fn parses_field_paths() { 161 + assert_eq!( 162 + parse_ok("subject.uri"), 163 + vec![ 164 + Segment { 165 + field: "subject".into(), 166 + modifier: None 167 + }, 168 + Segment { 169 + field: "uri".into(), 170 + modifier: None 171 + }, 172 + ] 173 + ); 174 + } 175 + 176 + #[test] 177 + fn parses_array_descents() { 178 + assert_eq!( 179 + parse_ok("repos[]"), 180 + vec![Segment { 181 + field: "repos".into(), 182 + modifier: Some(Modifier::Elements) 183 + }] 184 + ); 185 + assert_eq!( 186 + parse_ok("facets[].features[app.bsky.richtext.facet#mention].did"), 187 + vec![ 188 + Segment { 189 + field: "facets".into(), 190 + modifier: Some(Modifier::Elements) 191 + }, 192 + Segment { 193 + field: "features".into(), 194 + modifier: Some(Modifier::UnionElements( 195 + "app.bsky.richtext.facet#mention".into() 196 + )) 197 + }, 198 + Segment { 199 + field: "did".into(), 200 + modifier: None 201 + }, 202 + ] 203 + ); 204 + assert_eq!( 205 + parse_ok("embed{app.bsky.embed.record}.record.uri"), 206 + vec![ 207 + Segment { 208 + field: "embed".into(), 209 + modifier: Some(Modifier::Union("app.bsky.embed.record".into())) 210 + }, 211 + Segment { 212 + field: "record".into(), 213 + modifier: None 214 + }, 215 + Segment { 216 + field: "uri".into(), 217 + modifier: None 218 + }, 219 + ] 220 + ); 221 + } 222 + 223 + #[test] 224 + fn parses_escaped_field_names() { 225 + assert_eq!( 226 + parse_ok("meta.dot!.name"), 227 + vec![Segment { 228 + field: "meta".into(), 229 + modifier: None 230 + }] 231 + .into_iter() 232 + .chain([Segment { 233 + field: "dot.name".into(), 234 + modifier: None 235 + }]) 236 + .collect::<Vec<_>>() 237 + ); 238 + assert_eq!( 239 + parse_ok("meta.a!!b"), 240 + vec![ 241 + Segment { 242 + field: "meta".into(), 243 + modifier: None 244 + }, 245 + Segment { 246 + field: "a!b".into(), 247 + modifier: None 248 + }, 249 + ] 250 + ); 251 + assert_eq!( 252 + parse_ok("meta.$unknown"), 253 + vec![ 254 + Segment { 255 + field: "meta".into(), 256 + modifier: None 257 + }, 258 + Segment { 259 + field: "$unknown".into(), 260 + modifier: None 261 + }, 262 + ] 263 + ); 264 + } 265 + 266 + #[test] 267 + fn rejects_malformed_paths() { 268 + for bad in [ 269 + "", ".", "a..b", "a!", "a!x", "a[", "a[]b", "a{}", "[did]", "a[nsid]b", 270 + ] { 271 + assert!(parse_record_path(bad).is_err(), "{bad} should fail"); 272 + } 273 + } 274 + 275 + #[test] 276 + fn walks_vector_matches() { 277 + let doc = json!({ 278 + "repos": [ 279 + {"uri": "at://did:plc:a/sh.tangled.repo/x"}, 280 + {"uri": "at://did:plc:b/sh.tangled.repo/y"}, 281 + ], 282 + "owner": "did:plc:z", 283 + }); 284 + let segs = parse_ok("repos[].uri"); 285 + let found = walk_path(&segs, [&doc]); 286 + assert_eq!(found.len(), 2); 287 + } 288 + 289 + #[test] 290 + fn walks_union_filters() { 291 + let doc = json!({ 292 + "items": [ 293 + {"$type": "sh.tangled.repo", "uri": "at://did:plc:a/sh.tangled.repo/x"}, 294 + {"$type": "sh.tangled.actor.profile", "did": "did:plc:b"}, 295 + ] 296 + }); 297 + let segs = parse_ok("items[sh.tangled.repo].uri"); 298 + let found = walk_path(&segs, [&doc]); 299 + assert_eq!(found, vec![&json!("at://did:plc:a/sh.tangled.repo/x")]); 300 + } 301 + }
+441
bobbin/crates/xrpc/tests/enrich.rs
··· 1 + use std::sync::Arc; 2 + 3 + use axum::body::{Body, to_bytes}; 4 + use bobbin_edge_index::{CoverageWatch, EdgeStore, IssueStateKind, PullStatusKind, StateIndex}; 5 + use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig}; 6 + use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 7 + use bobbin_resolver::RepoIdResolver; 8 + use bobbin_runtime::{RuntimeHasher, SystemClock}; 9 + use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; 10 + use bobbin_slingshot_client::SlingshotClient; 11 + use bobbin_types::edges::Edge; 12 + use bobbin_types::ids::SubjectRef; 13 + use bobbin_xrpc::{AppState, router}; 14 + use http::{Request, StatusCode}; 15 + use jacquard_common::DefaultStr; 16 + use jacquard_common::types::did::Did; 17 + use jacquard_common::types::nsid::Nsid; 18 + use jacquard_common::types::string::AtUri; 19 + use serde_json::{Value, json}; 20 + use tower::ServiceExt; 21 + use url::Url; 22 + use wiremock::matchers::{method, path, query_param}; 23 + use wiremock::{Mock, MockServer, ResponseTemplate}; 24 + 25 + const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; 26 + 27 + fn at(s: &str) -> AtUri<DefaultStr> { 28 + AtUri::new_owned(s).unwrap() 29 + } 30 + 31 + fn did(s: &str) -> Did<DefaultStr> { 32 + Did::new_owned(s).unwrap() 33 + } 34 + 35 + fn nsid(s: &'static str) -> Nsid<DefaultStr> { 36 + Nsid::new_static(s).unwrap() 37 + } 38 + 39 + static EDGE_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); 40 + 41 + fn next_sort_micros() -> u64 { 42 + EDGE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed) 43 + } 44 + 45 + struct Harness { 46 + server: MockServer, 47 + edges: Arc<EdgeStore>, 48 + state: AppState, 49 + } 50 + 51 + impl Harness { 52 + async fn new() -> Self { 53 + let server = MockServer::start().await; 54 + let edges = Arc::new(EdgeStore::new(RuntimeHasher::default())); 55 + let coverage = Arc::new(CoverageWatch::new()); 56 + let state = AppState::new( 57 + Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 58 + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap(), 59 + edges.clone(), 60 + Arc::new(StateIndex::<IssueStateKind>::new(RuntimeHasher::default())), 61 + Arc::new(StateIndex::<PullStatusKind>::new(RuntimeHasher::default())), 62 + coverage.clone(), 63 + Arc::new( 64 + KnotProxy::new( 65 + KnotProxyConfig::default(), 66 + KnotHttpConfig::default(), 67 + Arc::new(SystemClock::new()), 68 + RuntimeHasher::default(), 69 + ) 70 + .unwrap(), 71 + ), 72 + Arc::new( 73 + SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, Arc::new(SystemClock::new())).unwrap(), 74 + ) as Arc<dyn SearchReader>, 75 + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), 76 + ); 77 + Self { 78 + server, 79 + edges, 80 + state, 81 + } 82 + } 83 + 84 + fn add_edge(&self, kind: &'static str, subject: SubjectRef, source: &AtUri<DefaultStr>) { 85 + self.edges.add(Edge { 86 + kind: nsid(kind), 87 + subject, 88 + source: source.clone(), 89 + sort_micros: next_sort_micros(), 90 + }); 91 + } 92 + 93 + async fn mount(&self, did: &Did<DefaultStr>, collection: &str, rkey: &str, value: Value) { 94 + let uri = format!("at://{}/{}/{}", did.as_ref(), collection, rkey); 95 + let body = json!({ "uri": uri, "cid": CID, "value": value }); 96 + Mock::given(method("GET")) 97 + .and(path("/xrpc/com.atproto.repo.getRecord")) 98 + .and(query_param("repo", did.as_ref())) 99 + .and(query_param("collection", collection)) 100 + .and(query_param("rkey", rkey)) 101 + .respond_with(ResponseTemplate::new(200).set_body_json(body)) 102 + .mount(&self.server) 103 + .await; 104 + } 105 + } 106 + 107 + fn enrich_request(body: Value) -> Request<Body> { 108 + Request::builder() 109 + .method("POST") 110 + .uri("/xrpc/sh.tangled.query.enrichResponse") 111 + .header("content-type", "application/json") 112 + .body(Body::from(serde_json::to_vec(&body).unwrap())) 113 + .unwrap() 114 + } 115 + 116 + async fn json_response(resp: axum::response::Response) -> (StatusCode, Value) { 117 + let status = resp.status(); 118 + let bytes = to_bytes(resp.into_body(), 1 << 20).await.unwrap(); 119 + let parsed: Value = serde_json::from_slice(&bytes).expect("JSON body"); 120 + (status, parsed) 121 + } 122 + 123 + fn repo_body(name: &str, repo_did: &Did<DefaultStr>) -> Value { 124 + json!({ 125 + "$type": "sh.tangled.repo", 126 + "name": name, 127 + "knot": "oyster.cafe", 128 + "repoDid": repo_did.as_ref(), 129 + "createdAt": "2026-05-01T00:00:00Z" 130 + }) 131 + } 132 + 133 + fn follow_body(subject: &Did<DefaultStr>) -> Value { 134 + json!({ 135 + "$type": "sh.tangled.graph.follow", 136 + "createdAt": "2026-05-01T00:00:00Z", 137 + "subject": subject.as_ref() 138 + }) 139 + } 140 + 141 + /// one repo owned by `owner`, with `stars`/`issues` counts against its repo did 142 + async fn repo_fixture(h: &Harness, owner: &Did<DefaultStr>, repo_did: &Did<DefaultStr>) { 143 + let repo_uri = at(&format!("at://{}/sh.tangled.repo/reef", owner.as_ref())); 144 + h.add_edge("sh.tangled.repo", SubjectRef::Did(owner.clone()), &repo_uri); 145 + h.mount( 146 + owner, 147 + "sh.tangled.repo", 148 + "reef", 149 + repo_body("reef", repo_did), 150 + ) 151 + .await; 152 + for (i, stargazer) in ["did:plc:a", "did:plc:b", "did:plc:a"].iter().enumerate() { 153 + h.add_edge( 154 + "sh.tangled.feed.star", 155 + SubjectRef::Did(repo_did.clone()), 156 + &at(&format!("at://{stargazer}/sh.tangled.feed.star/s{i}")), 157 + ); 158 + } 159 + h.add_edge( 160 + "sh.tangled.repo.issue", 161 + SubjectRef::Did(repo_did.clone()), 162 + &at("at://did:plc:a/sh.tangled.repo.issue/i0"), 163 + ); 164 + } 165 + 166 + #[tokio::test] 167 + async fn zero_config_counts_stars_and_issues_for_repo_did() { 168 + let h = Harness::new().await; 169 + let owner = did("did:plc:nel"); 170 + let repo_did = did("did:plc:limpet"); 171 + repo_fixture(&h, &owner, &repo_did).await; 172 + 173 + let app = router(h.state.clone()); 174 + let (status, body) = json_response( 175 + app.oneshot(enrich_request(json!({ 176 + "xrpc": "sh.tangled.repo.listRepos", 177 + "params": { "subject": owner.as_ref() }, 178 + "enrich": [ 179 + { "source": "sh.tangled.feed.star:subject", "type": "count" }, 180 + { "source": "sh.tangled.feed.star:subject", "type": "distinctAuthors" }, 181 + { "source": "sh.tangled.repo.issue:subject", "type": "count" } 182 + ] 183 + }))) 184 + .await 185 + .unwrap(), 186 + ) 187 + .await; 188 + 189 + assert_eq!(status, StatusCode::OK, "{body}"); 190 + assert_eq!(body["output"]["items"].as_array().unwrap().len(), 1); 191 + let stats = &body["stats"]; 192 + assert_eq!( 193 + stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["count"], 194 + json!(3) 195 + ); 196 + assert_eq!( 197 + stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["distinctAuthors"], 198 + json!(2) 199 + ); 200 + assert_eq!( 201 + stats["did:plc:limpet"]["sh.tangled.repo.issue:subject"]["count"], 202 + json!(1) 203 + ); 204 + assert!(stats["at://did:plc:nel/sh.tangled.repo/reef"].is_null()); 205 + } 206 + 207 + #[tokio::test] 208 + async fn follow_counts_cover_both_directions() { 209 + let h = Harness::new().await; 210 + let owner = did("did:plc:nel"); 211 + // followers, edges pointing at owner 212 + for (i, fan) in ["did:plc:a", "did:plc:b"].iter().enumerate() { 213 + h.add_edge( 214 + "sh.tangled.graph.follow", 215 + SubjectRef::Did(owner.clone()), 216 + &at(&format!("at://{fan}/sh.tangled.graph.follow/f{i}")), 217 + ); 218 + h.mount( 219 + &did(fan), 220 + "sh.tangled.graph.follow", 221 + &format!("f{i}"), 222 + follow_body(&owner), 223 + ) 224 + .await; 225 + } 226 + // following, via the .by mirror edge since owner is the author here 227 + h.add_edge( 228 + "sh.tangled.graph.follow.by", 229 + SubjectRef::Did(owner.clone()), 230 + &at("at://did:plc:nel/sh.tangled.graph.follow/f0"), 231 + ); 232 + 233 + let app = router(h.state.clone()); 234 + let (status, body) = json_response( 235 + app.oneshot(enrich_request(json!({ 236 + "xrpc": "sh.tangled.graph.listFollows", 237 + "params": { "subject": owner.as_ref() }, 238 + "enrich": [{ "source": "sh.tangled.graph.follow:subject", "type": "count" }, { "source": "sh.tangled.graph.follow:.repo", "type": "count" }] 239 + }))) 240 + .await 241 + .unwrap(), 242 + ) 243 + .await; 244 + 245 + assert_eq!(status, StatusCode::OK, "{body}"); 246 + let nel = &body["stats"]["did:plc:nel"]; 247 + assert_eq!( 248 + nel["sh.tangled.graph.follow:subject"]["count"], 249 + json!(2), 250 + "{body}" 251 + ); 252 + assert_eq!( 253 + nel["sh.tangled.graph.follow:.repo"]["count"], 254 + json!(1), 255 + "{body}" 256 + ); 257 + } 258 + 259 + #[tokio::test] 260 + async fn sources_scope_which_refs_get_enriched() { 261 + let h = Harness::new().await; 262 + let owner = did("did:plc:nel"); 263 + let repo_did = did("did:plc:limpet"); 264 + repo_fixture(&h, &owner, &repo_did).await; 265 + 266 + let app = router(h.state.clone()); 267 + let (status, body) = json_response( 268 + app.oneshot(enrich_request(json!({ 269 + "xrpc": "sh.tangled.repo.listRepos", 270 + "params": { "subject": owner.as_ref() }, 271 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }], 272 + "sources": ["items[].value.repoDid"] 273 + }))) 274 + .await 275 + .unwrap(), 276 + ) 277 + .await; 278 + assert_eq!(status, StatusCode::OK, "{body}"); 279 + let stats = &body["stats"]; 280 + assert_eq!( 281 + stats["did:plc:limpet"]["sh.tangled.feed.star:subject"]["count"], 282 + json!(3) 283 + ); 284 + assert_eq!(stats.as_object().unwrap().len(), 1, "{body}"); 285 + 286 + // a path matching nothing is empty stats, not an error, since selection is vector-matched 287 + let app = router(h.state.clone()); 288 + let (status, body) = json_response( 289 + app.oneshot(enrich_request(json!({ 290 + "xrpc": "sh.tangled.repo.listRepos", 291 + "params": { "subject": owner.as_ref() }, 292 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }], 293 + "sources": ["items[].value.nope"] 294 + }))) 295 + .await 296 + .unwrap(), 297 + ) 298 + .await; 299 + assert_eq!(status, StatusCode::OK, "{body}"); 300 + assert_eq!(body["stats"], json!({})); 301 + } 302 + 303 + #[tokio::test] 304 + async fn inner_record_miss_passes_through_as_404() { 305 + let h = Harness::new().await; 306 + let app = router(h.state.clone()); 307 + let (status, body) = json_response( 308 + app.oneshot(enrich_request(json!({ 309 + "xrpc": "sh.tangled.repo.getRepo", 310 + "params": { "repo": "at://did:plc:nel/sh.tangled.repo/absent" }, 311 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }] 312 + }))) 313 + .await 314 + .unwrap(), 315 + ) 316 + .await; 317 + assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); 318 + assert_eq!(body["error"], json!("RecordNotFound")); 319 + } 320 + 321 + #[tokio::test] 322 + async fn rejects_bad_requests() { 323 + let h = Harness::new().await; 324 + let cases = [ 325 + json!({ "xrpc": "sh.tangled.nope.nope", "enrich": [] }), 326 + json!({ 327 + "xrpc": "sh.tangled.repo.countRepos", 328 + "params": { "subject": "did:plc:nel" }, 329 + "enrich": [{ "source": "sh.tangled.nope:subject", "type": "count" }] 330 + }), 331 + json!({ 332 + "xrpc": "sh.tangled.repo.countRepos", 333 + "params": { "subject": "did:plc:nel" }, 334 + "enrich": [{ "source": "sh.tangled.feed.star:subject.uri", "type": "count" }] 335 + }), 336 + json!({ 337 + "xrpc": "sh.tangled.repo.countRepos", 338 + "params": { "subject": "did:plc:nel" }, 339 + "enrich": [{ "source": "sh.tangled.feed.star:.rkey", "type": "count" }] 340 + }), 341 + json!({ 342 + "xrpc": "sh.tangled.repo.countRepos", 343 + "params": { "subject": "did:plc:nel" }, 344 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "count" }], 345 + "sources": ["items["] 346 + }), 347 + // no defaults, a source without a colon is rejected 348 + json!({ 349 + "xrpc": "sh.tangled.repo.countRepos", 350 + "params": { "subject": "did:plc:nel" }, 351 + "enrich": [{ "source": "sh.tangled.feed.star", "type": "count" }] 352 + }), 353 + ]; 354 + for case in cases { 355 + let app = router(h.state.clone()); 356 + let (status, body) = json_response(app.oneshot(enrich_request(case)).await.unwrap()).await; 357 + assert_eq!(status, StatusCode::BAD_REQUEST, "{body}"); 358 + assert_eq!(body["error"], json!("InvalidRequest"), "{body}"); 359 + } 360 + 361 + // serde rejects a missing type before the handler sees it 362 + // that returns plain-text 422, not our 400 json body 363 + let app = router(h.state.clone()); 364 + let response = app 365 + .oneshot(enrich_request(json!({ 366 + "xrpc": "sh.tangled.repo.countRepos", 367 + "params": { "subject": "did:plc:nel" }, 368 + "enrich": [{ "source": "sh.tangled.feed.star:subject" }] 369 + }))) 370 + .await 371 + .unwrap(); 372 + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); 373 + } 374 + 375 + #[tokio::test] 376 + async fn viewer_aggregation_uses_explicit_viewer_param() { 377 + let h = Harness::new().await; 378 + let owner = did("did:plc:abc"); 379 + let repo_did = did("did:plc:limpet"); 380 + repo_fixture(&h, &owner, &repo_did).await; 381 + 382 + // the viewer already starred this repo, for the checks below 383 + let subject = SubjectRef::Did(repo_did.clone()); 384 + h.state.edges.add(Edge { 385 + kind: nsid("sh.tangled.feed.star"), 386 + subject, 387 + source: at("at://did:plc:nel/sh.tangled.feed.star/r99"), 388 + sort_micros: 99, 389 + }); 390 + 391 + let app = router(h.state.clone()); 392 + 393 + // missing viewer param is a 400, viewer descriptors require it 394 + let no_viewer = json!({ 395 + "xrpc": "sh.tangled.repo.listRepos", 396 + "params": { "subject": owner.as_ref() }, 397 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }] 398 + }); 399 + let (status, _) = json_response( 400 + app.clone() 401 + .oneshot(enrich_request(no_viewer)) 402 + .await 403 + .unwrap(), 404 + ) 405 + .await; 406 + assert_eq!(status, StatusCode::BAD_REQUEST); 407 + 408 + // viewer who starred it gets their own star uri back 409 + let starred_viewer = json!({ 410 + "xrpc": "sh.tangled.repo.listRepos", 411 + "params": { "subject": owner.as_ref() }, 412 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }], 413 + "viewer": "did:plc:nel" 414 + }); 415 + let (status, resp) = json_response( 416 + app.clone() 417 + .oneshot(enrich_request(starred_viewer)) 418 + .await 419 + .unwrap(), 420 + ) 421 + .await; 422 + assert_eq!(status, StatusCode::OK); 423 + let stats = &resp["stats"][repo_did.as_str()]["sh.tangled.feed.star:subject"]; 424 + assert_eq!( 425 + stats["viewer"], 426 + json!("at://did:plc:nel/sh.tangled.feed.star/r99") 427 + ); 428 + 429 + // a viewer who never starred it gets an explicit null, not absent 430 + let other_viewer = json!({ 431 + "xrpc": "sh.tangled.repo.listRepos", 432 + "params": { "subject": owner.as_ref() }, 433 + "enrich": [{ "source": "sh.tangled.feed.star:subject", "type": "viewer" }], 434 + "viewer": "did:plc:someoneelse" 435 + }); 436 + let (status, resp) = 437 + json_response(app.oneshot(enrich_request(other_viewer)).await.unwrap()).await; 438 + assert_eq!(status, StatusCode::OK); 439 + let stats = &resp["stats"][repo_did.as_str()]["sh.tangled.feed.star:subject"]; 440 + assert_eq!(stats["viewer"], Value::Null); 441 + }
+104
lexicons/query/enrichResponse.json
··· 1 + { 2 + "lexicon": 1, 3 + "id": "sh.tangled.query.enrichResponse", 4 + "defs": { 5 + "main": { 6 + "type": "procedure", 7 + "description": "like calling an xrpc query directly, but the response comes back with a stats sidecar: aggregate counts (stars, follows, issues...) for every did / at-uri / strong-ref found in the response. counts are computed synchronously from the server's local index, so there are no pending states to resolve later. any GET query the server implements can be hydrated, including queries it proxies elsewhere.", 8 + "input": { 9 + "encoding": "application/json", 10 + "schema": { 11 + "type": "object", 12 + "required": [ 13 + "xrpc", 14 + "enrich" 15 + ], 16 + "properties": { 17 + "xrpc": { 18 + "type": "string", 19 + "format": "nsid", 20 + "description": "NSID of the inner query to run, e.g. sh.tangled.repo.listRepos." 21 + }, 22 + "params": { 23 + "type": "unknown", 24 + "description": "Parameters for the inner query, exactly as it declares them." 25 + }, 26 + "enrich": { 27 + "type": "array", 28 + "items": { 29 + "type": "ref", 30 + "ref": "#linkDescriptor" 31 + }, 32 + "description": "Counts to compute for each found reference. A descriptor only applies to references it can meaningfully point at: a did can carry author-side counts (.repo), an at-uri can carry subject-side counts." 33 + }, 34 + "sources": { 35 + "type": "array", 36 + "items": { 37 + "type": "string" 38 + }, 39 + "description": "RecordPaths into the inner response restricting which references get counted. When omitted, every did / at-uri / strong-ref in the response is counted. Paths that match nothing yield empty stats rather than an error." 40 + }, 41 + "viewer": { 42 + "type": "string", 43 + "format": "did", 44 + "description": "DID whose own relation to each reference is looked up for viewer-type descriptors, e.g. whether this did starred the found repo and with which record. Required when any enrich descriptor uses type viewer." 45 + } 46 + } 47 + } 48 + }, 49 + "output": { 50 + "encoding": "application/json", 51 + "schema": { 52 + "type": "object", 53 + "required": [ 54 + "output", 55 + "stats" 56 + ], 57 + "properties": { 58 + "output": { 59 + "type": "unknown", 60 + "description": "The inner query's response, unmodified." 61 + }, 62 + "stats": { 63 + "type": "unknown", 64 + "description": "Reference value (did or at-uri) -> link source (echoed verbatim from the request) -> aggregation results. Example: stats[\"did:plc:...\"][\"sh.tangled.graph.follow:subject\"] = {\"count\": 12, \"distinctAuthors\": 11}." 65 + } 66 + } 67 + } 68 + }, 69 + "errors": [ 70 + { 71 + "name": "UnknownQuery", 72 + "description": "The xrpc parameter does not name a query this server can execute." 73 + }, 74 + { 75 + "name": "InvalidSourcePath", 76 + "description": "A sources entry is not valid RecordPath syntax." 77 + }, 78 + { 79 + "name": "InvalidLinkDescriptor", 80 + "description": "An enrich entry names an unknown collection, or a path the server's index does not support." 81 + } 82 + ] 83 + }, 84 + "linkDescriptor": { 85 + "type": "object", 86 + "required": [ 87 + "source", 88 + "type" 89 + ], 90 + "description": "one aggregate count: records whose reference field equals the found reference, addressed by a constellation-style link source (\"collection:recordpath\"). envelope paths (.repo etc.) address the record's own metadata instead of its contents.", 91 + "properties": { 92 + "source": { 93 + "type": "string", 94 + "description": "Link source: the collection whose records are counted, a colon, then a RecordPath naming the reference field (e.g. sh.tangled.feed.star:subject, sh.tangled.feed.star:subject.uri), or an envelope field: .repo (authoring repo), .collection, .rkey, or bare . (the record's own at-uri)." 95 + }, 96 + "type": { 97 + "type": "string", 98 + "knownValues": ["count", "distinctAuthors", "viewer"], 99 + "description": "What to compute over matching records: how many records (count), how many distinct authors (distinctAuthors), or the viewer's own record among the matches (viewer, requires the top-level viewer param). Results land in the stats sidecar at stats[ref][source][type]." 100 + } 101 + } 102 + } 103 + } 104 + }
+1
web/lex.config.ts
··· 20 20 "../lexicons/markup/**/*.json", 21 21 "../lexicons/pipeline/**/*.json", 22 22 "../lexicons/pulls/**/*.json", 23 + "../lexicons/query/**/*.json", 23 24 "../lexicons/repo/**/*.json", 24 25 "../lexicons/spindle/**/*.json", 25 26 "../lexicons/string/**/*.json",
+1
web/src/lib/api/lexicons/index.ts
··· 68 68 export * as ShTangledPipelineStatus from "./types/sh/tangled/pipeline/status.js"; 69 69 export * as ShTangledPublicKey from "./types/sh/tangled/publicKey.js"; 70 70 export * as ShTangledPublicKeyListKeys from "./types/sh/tangled/publicKey/listKeys.js"; 71 + export * as ShTangledQueryEnrichResponse from "./types/sh/tangled/query/enrichResponse.js"; 71 72 export * as ShTangledRepo from "./types/sh/tangled/repo.js"; 72 73 export * as ShTangledRepoAddCollaborator from "./types/sh/tangled/repo/addCollaborator.js"; 73 74 export * as ShTangledRepoAddSecret from "./types/sh/tangled/repo/addSecret.js";
+91
web/src/lib/api/lexicons/types/sh/tangled/query/enrichResponse.ts
··· 1 + import type {} from "@atcute/lexicons"; 2 + import * as v from "@atcute/lexicons/validations"; 3 + import type {} from "@atcute/lexicons/ambient"; 4 + 5 + const _linkDescriptorSchema = /*#__PURE__*/ v.object({ 6 + $type: /*#__PURE__*/ v.optional( 7 + /*#__PURE__*/ v.literal("sh.tangled.query.enrichResponse#linkDescriptor"), 8 + ), 9 + /** 10 + * Link source: the collection whose records are counted, a colon, then a RecordPath naming the reference field (e.g. sh.tangled.feed.star:subject, sh.tangled.feed.star:subject.uri), or an envelope field: .repo (authoring repo), .collection, .rkey, or bare . (the record's own at-uri). 11 + */ 12 + source: /*#__PURE__*/ v.string(), 13 + /** 14 + * What to compute over matching records: how many records (count), how many distinct authors (distinctAuthors), or the viewer's own record among the matches (viewer, requires the top-level viewer param). Results land in the stats sidecar at stats[ref][source][type]. 15 + */ 16 + type: /*#__PURE__*/ v.string< 17 + "count" | "distinctAuthors" | "viewer" | (string & {}) 18 + >(), 19 + }); 20 + const _mainSchema = /*#__PURE__*/ v.procedure( 21 + "sh.tangled.query.enrichResponse", 22 + { 23 + params: null, 24 + input: { 25 + type: "lex", 26 + schema: /*#__PURE__*/ v.object({ 27 + /** 28 + * Counts to compute for each found reference. A descriptor only applies to references it can meaningfully point at: a did can carry author-side counts (.repo), an at-uri can carry subject-side counts. 29 + */ 30 + get enrich() { 31 + return /*#__PURE__*/ v.array(linkDescriptorSchema); 32 + }, 33 + /** 34 + * Parameters for the inner query, exactly as it declares them. 35 + */ 36 + params: /*#__PURE__*/ v.optional(/*#__PURE__*/ v.unknown()), 37 + /** 38 + * RecordPaths into the inner response restricting which references get counted. When omitted, every did / at-uri / strong-ref in the response is counted. Paths that match nothing yield empty stats rather than an error. 39 + */ 40 + sources: /*#__PURE__*/ v.optional( 41 + /*#__PURE__*/ v.array(/*#__PURE__*/ v.string()), 42 + ), 43 + /** 44 + * DID whose own relation to each reference is looked up for viewer-type descriptors, e.g. whether this did starred the found repo and with which record. Required when any enrich descriptor uses type viewer. 45 + */ 46 + viewer: /*#__PURE__*/ v.optional(/*#__PURE__*/ v.didString()), 47 + /** 48 + * NSID of the inner query to run, e.g. sh.tangled.repo.listRepos. 49 + */ 50 + xrpc: /*#__PURE__*/ v.nsidString(), 51 + }), 52 + }, 53 + output: { 54 + type: "lex", 55 + schema: /*#__PURE__*/ v.object({ 56 + /** 57 + * The inner query's response, unmodified. 58 + */ 59 + output: /*#__PURE__*/ v.unknown(), 60 + /** 61 + * Reference value (did or at-uri) -> link source (echoed verbatim from the request) -> aggregation results. Example: stats["did:plc:..."]["sh.tangled.graph.follow:subject"] = {"count": 12, "distinctAuthors": 11}. 62 + */ 63 + stats: /*#__PURE__*/ v.unknown(), 64 + }), 65 + }, 66 + }, 67 + ); 68 + 69 + type linkDescriptor$schematype = typeof _linkDescriptorSchema; 70 + type main$schematype = typeof _mainSchema; 71 + 72 + export interface linkDescriptorSchema extends linkDescriptor$schematype {} 73 + export interface mainSchema extends main$schematype {} 74 + 75 + export const linkDescriptorSchema = 76 + _linkDescriptorSchema as linkDescriptorSchema; 77 + export const mainSchema = _mainSchema as mainSchema; 78 + 79 + export interface LinkDescriptor extends v.InferInput< 80 + typeof linkDescriptorSchema 81 + > {} 82 + 83 + export interface $params {} 84 + export interface $input extends v.InferXRPCBodyInput<mainSchema["input"]> {} 85 + export interface $output extends v.InferXRPCBodyInput<mainSchema["output"]> {} 86 + 87 + declare module "@atcute/lexicons/ambient" { 88 + interface XRPCProcedures { 89 + "sh.tangled.query.enrichResponse": mainSchema; 90 + } 91 + }