+1 -1

constellation/src/lib.rs

··· 31 31 } 32 32 } 33 33 34 34 - #[derive(Debug, PartialEq, Serialize, Deserialize)] 34 34 + #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)] 35 35 pub struct RecordId { 36 36 pub did: Did, 37 37 pub collection: String,

+102

constellation/src/server/mod.rs

··· 89 89 }), 90 90 ) 91 91 .route( 92 92 + "/xrpc/blue.microcosm.links.getManyToMany", 93 93 + get({ 94 94 + let store = store.clone(); 95 95 + move |accept, query| async { 96 96 + spawn_blocking(|| get_many_to_many(accept, query, store)) 97 97 + .await 98 98 + .map_err(to500)? 99 99 + } 100 100 + }), 101 101 + ) 102 102 + .route( 92 103 "/xrpc/blue.microcosm.links.getBacklinks", 93 104 get({ 94 105 let store = store.clone(); ··· 580 591 accept, 581 592 GetLinkItemsResponse { 582 593 total: paged.total, 594 594 + linking_records: paged.items, 595 595 + cursor, 596 596 + query: (*query).clone(), 597 597 + }, 598 598 + )) 599 599 + } 600 600 + 601 601 + #[derive(Clone, Deserialize)] 602 602 + #[serde(rename_all = "camelCase")] 603 603 + struct GetManyToManyItemsQuery { 604 604 + subject: String, 605 605 + source: String, 606 606 + /// path to the secondary link in the linking record 607 607 + path_to_other: String, 608 608 + /// filter to linking records (join of the m2m) by these DIDs 609 609 + #[serde(default)] 610 610 + did: Vec<String>, 611 611 + /// filter to specific secondary records 612 612 + #[serde(default)] 613 613 + other_subject: Vec<String>, 614 614 + cursor: Option<OpaqueApiCursor>, 615 615 + #[serde(default = "get_default_cursor_limit")] 616 616 + limit: u64, 617 617 + } 618 618 + #[derive(Template, Serialize)] 619 619 + #[template(path = "get-many-to-many.html.j2")] 620 620 + struct GetManyToManyItemsResponse { 621 621 + linking_records: Vec<(String, Vec<RecordId>)>, 622 622 + cursor: Option<OpaqueApiCursor>, 623 623 + #[serde(skip_serializing)] 624 624 + query: GetManyToManyItemsQuery, 625 625 + } 626 626 + fn get_many_to_many( 627 627 + accept: ExtractAccept, 628 628 + query: axum_extra::extract::Query<GetManyToManyItemsQuery>, // supports multiple param occurrences 629 629 + store: impl LinkReader, 630 630 + ) -> Result<impl IntoResponse, http::StatusCode> { 631 631 + let after = query 632 632 + .cursor 633 633 + .clone() 634 634 + .map(|oc| ApiKeyedCursor::try_from(oc).map_err(|_| http::StatusCode::BAD_REQUEST)) 635 635 + .transpose()? 636 636 + .map(|c| c.next); 637 637 + 638 638 + let limit = query.limit; 639 639 + if limit > DEFAULT_CURSOR_LIMIT_MAX { 640 640 + return Err(http::StatusCode::BAD_REQUEST); 641 641 + } 642 642 + 643 643 + let filter_dids: HashSet<Did> = HashSet::from_iter( 644 644 + query 645 645 + .did 646 646 + .iter() 647 647 + .map(|d| d.trim()) 648 648 + .filter(|d| !d.is_empty()) 649 649 + .map(|d| Did(d.to_string())), 650 650 + ); 651 651 + 652 652 + let filter_other_subjects: HashSet<String> = HashSet::from_iter( 653 653 + query 654 654 + .other_subject 655 655 + .iter() 656 656 + .map(|s| s.trim().to_string()) 657 657 + .filter(|s| !s.is_empty()), 658 658 + ); 659 659 + 660 660 + let Some((collection, path)) = query.source.split_once(':') else { 661 661 + return Err(http::StatusCode::BAD_REQUEST); 662 662 + }; 663 663 + let path = format!(".{path}"); 664 664 + 665 665 + let path_to_other = format!(".{}", query.path_to_other); 666 666 + 667 667 + let paged = store 668 668 + .get_many_to_many( 669 669 + &query.subject, 670 670 + collection, 671 671 + &path, 672 672 + &path_to_other, 673 673 + limit, 674 674 + after, 675 675 + &filter_dids, 676 676 + &filter_other_subjects, 677 677 + ) 678 678 + .map_err(|_| http::StatusCode::INTERNAL_SERVER_ERROR)?; 679 679 + 680 680 + let cursor = paged.next.map(|next| ApiKeyedCursor { next }.into()); 681 681 + 682 682 + Ok(acceptable( 683 683 + accept, 684 684 + GetManyToManyItemsResponse { 583 685 linking_records: paged.items, 584 686 cursor, 585 687 query: (*query).clone(),

+93

constellation/src/storage/mem_store.rs

··· 234 234 .len() as u64) 235 235 } 236 236 237 237 + fn get_many_to_many( 238 238 + &self, 239 239 + target: &str, 240 240 + collection: &str, 241 241 + path: &str, 242 242 + path_to_other: &str, 243 243 + limit: u64, 244 244 + after: Option<String>, 245 245 + filter_dids: &HashSet<Did>, 246 246 + filter_to_targets: &HashSet<String>, 247 247 + ) -> Result<PagedOrderedCollection<(String, Vec<RecordId>), String>> { 248 248 + let empty_res = Ok(PagedOrderedCollection { 249 249 + items: Vec::new(), 250 250 + next: Some("".to_string()), 251 251 + }); 252 252 + 253 253 + // struct MemStorageData { 254 254 + // dids: HashMap<Did, bool>, 255 255 + // targets: HashMap<Target, HashMap<Source, Linkers>>, 256 256 + // links: HashMap<Did, HashMap<RepoId, Vec<(RecordPath, Target)>>>, 257 257 + // } 258 258 + let data = self.0.lock().unwrap(); 259 259 + 260 260 + let Some(sources) = data.targets.get(&Target::new(target)) else { 261 261 + return empty_res; 262 262 + }; 263 263 + let Some(linkers) = sources.get(&Source::new(collection, path)) else { 264 264 + return empty_res; 265 265 + }; 266 266 + let path_to_other = RecordPath::new(path_to_other); 267 267 + 268 268 + // Convert filter_to_targets to Target objects for comparison 269 269 + let filter_to_target_objs: HashSet<Target> = 270 270 + HashSet::from_iter(filter_to_targets.iter().map(|s| Target::new(s))); 271 271 + 272 272 + let mut grouped_links: HashMap<Target, Vec<RecordId>> = HashMap::new(); 273 273 + for (did, rkey) in linkers.iter().flatten().cloned() { 274 274 + // Filter by DID if filter is provided 275 275 + if !filter_dids.is_empty() && !filter_dids.contains(&did) { 276 276 + continue; 277 277 + } 278 278 + if let Some(fwd_target) = data 279 279 + .links 280 280 + .get(&did) 281 281 + .unwrap_or(&HashMap::new()) 282 282 + .get(&RepoId { 283 283 + collection: collection.to_string(), 284 284 + rkey: rkey.clone(), 285 285 + }) 286 286 + .unwrap_or(&Vec::new()) 287 287 + .iter() 288 288 + .find_map(|(path, target)| { 289 289 + if *path == path_to_other 290 290 + && (filter_to_target_objs.is_empty() 291 291 + || filter_to_target_objs.contains(target)) 292 292 + { 293 293 + Some(target) 294 294 + } else { 295 295 + None 296 296 + } 297 297 + }) 298 298 + { 299 299 + let record_ids = grouped_links.entry(fwd_target.clone()).or_default(); 300 300 + record_ids.push(RecordId { 301 301 + did, 302 302 + collection: collection.to_string(), 303 303 + rkey: rkey.0, 304 304 + }); 305 305 + } 306 306 + } 307 307 + 308 308 + let mut items = grouped_links 309 309 + .into_iter() 310 310 + .map(|(t, r)| (t.0, r)) 311 311 + .collect::<Vec<_>>(); 312 312 + 313 313 + items.sort_by(|(a, _), (b, _)| a.cmp(b)); 314 314 + 315 315 + items = items 316 316 + .into_iter() 317 317 + .skip_while(|(t, _)| after.as_ref().map(|a| t <= a).unwrap_or(false)) 318 318 + .take(limit as usize) 319 319 + .collect(); 320 320 + 321 321 + let next = if items.len() as u64 >= limit { 322 322 + items.last().map(|(t, _)| t.clone()) 323 323 + } else { 324 324 + None 325 325 + }; 326 326 + 327 327 + Ok(PagedOrderedCollection { items, next }) 328 328 + } 329 329 + 237 330 fn get_links( 238 331 &self, 239 332 target: &str,

+12

constellation/src/storage/mod.rs

··· 104 104 fn get_all_record_counts(&self, _target: &str) 105 105 -> Result<HashMap<String, HashMap<String, u64>>>; 106 106 107 107 + fn get_many_to_many( 108 108 + &self, 109 109 + target: &str, 110 110 + collection: &str, 111 111 + path: &str, 112 112 + path_to_other: &str, 113 113 + limit: u64, 114 114 + after: Option<String>, 115 115 + filter_dids: &HashSet<Did>, 116 116 + filter_to_targets: &HashSet<String>, 117 117 + ) -> Result<PagedOrderedCollection<(String, Vec<RecordId>), String>>; 118 118 + 107 119 fn get_all_counts( 108 120 &self, 109 121 _target: &str,

+143

constellation/src/storage/rocks_store.rs

··· 1122 1122 } 1123 1123 } 1124 1124 1125 1125 + fn get_many_to_many( 1126 1126 + &self, 1127 1127 + target: &str, 1128 1128 + collection: &str, 1129 1129 + path: &str, 1130 1130 + path_to_other: &str, 1131 1131 + limit: u64, 1132 1132 + after: Option<String>, 1133 1133 + filter_dids: &HashSet<Did>, 1134 1134 + filter_to_targets: &HashSet<String>, 1135 1135 + ) -> Result<PagedOrderedCollection<(String, Vec<RecordId>), String>> { 1136 1136 + let collection = Collection(collection.to_string()); 1137 1137 + let path = RPath(path.to_string()); 1138 1138 + 1139 1139 + let target_key = TargetKey(Target(target.to_string()), collection.clone(), path); 1140 1140 + 1141 1141 + let after = after.map(|s| s.parse::<u64>().map(TargetId)).transpose()?; 1142 1142 + 1143 1143 + let Some(target_id) = self.target_id_table.get_id_val(&self.db, &target_key)? else { 1144 1144 + eprintln!("Target not found for {target_key:?}"); 1145 1145 + return Ok(Default::default()); 1146 1146 + }; 1147 1147 + 1148 1148 + let filter_did_ids: HashMap<DidId, bool> = filter_dids 1149 1149 + .iter() 1150 1150 + .filter_map(|did| self.did_id_table.get_id_val(&self.db, did).transpose()) 1151 1151 + .collect::<Result<Vec<DidIdValue>>>()? 1152 1152 + .into_iter() 1153 1153 + .map(|DidIdValue(id, active)| (id, active)) 1154 1154 + .collect(); 1155 1155 + 1156 1156 + let mut filter_to_target_ids: HashSet<TargetId> = HashSet::new(); 1157 1157 + for t in filter_to_targets { 1158 1158 + for (_, target_id) in self.iter_targets_for_target(&Target(t.to_string())) { 1159 1159 + filter_to_target_ids.insert(target_id); 1160 1160 + } 1161 1161 + } 1162 1162 + 1163 1163 + let linkers = self.get_target_linkers(&target_id)?; 1164 1164 + 1165 1165 + // we want to provide many to many which effectively means that we want to show a specific 1166 1166 + // list of reords that is linked to by a specific number of linkers 1167 1167 + let mut grouped_links: BTreeMap<TargetId, Vec<RecordId>> = BTreeMap::new(); 1168 1168 + for (did_id, rkey) in linkers.0 { 1169 1169 + if did_id.is_empty() { 1170 1170 + continue; 1171 1171 + } 1172 1172 + 1173 1173 + if !filter_did_ids.is_empty() && filter_did_ids.get(&did_id) != Some(&true) { 1174 1174 + continue; 1175 1175 + } 1176 1176 + 1177 1177 + // Make sure the current did is active 1178 1178 + let Some(did) = self.did_id_table.get_val_from_id(&self.db, did_id.0)? else { 1179 1179 + eprintln!("failed to look up did from did_id {did_id:?}"); 1180 1180 + continue; 1181 1181 + }; 1182 1182 + let Some(DidIdValue(_, active)) = self.did_id_table.get_id_val(&self.db, &did)? else { 1183 1183 + eprintln!("failed to look up did_value from did_id {did_id:?}: {did:?}: data consistency bug?"); 1184 1184 + continue; 1185 1185 + }; 1186 1186 + if !active { 1187 1187 + continue; 1188 1188 + } 1189 1189 + 1190 1190 + let record_link_key = RecordLinkKey(did_id, collection.clone(), rkey.clone()); 1191 1191 + let Some(targets) = self.get_record_link_targets(&record_link_key)? else { 1192 1192 + continue; 1193 1193 + }; 1194 1194 + 1195 1195 + let Some(fwd_target) = targets 1196 1196 + .0 1197 1197 + .into_iter() 1198 1198 + .filter_map(|RecordLinkTarget(rpath, target_id)| { 1199 1199 + if rpath.0 == path_to_other 1200 1200 + && (filter_to_target_ids.is_empty() 1201 1201 + || filter_to_target_ids.contains(&target_id)) 1202 1202 + { 1203 1203 + Some(target_id) 1204 1204 + } else { 1205 1205 + None 1206 1206 + } 1207 1207 + }) 1208 1208 + .take(1) 1209 1209 + .next() 1210 1210 + else { 1211 1211 + eprintln!("no forward match found."); 1212 1212 + continue; 1213 1213 + }; 1214 1214 + 1215 1215 + // pagination logic mirrors what is currently done in get_many_to_many_counts 1216 1216 + if after.as_ref().map(|a| fwd_target <= *a).unwrap_or(false) { 1217 1217 + continue; 1218 1218 + } 1219 1219 + let page_is_full = grouped_links.len() as u64 >= limit; 1220 1220 + if page_is_full { 1221 1221 + let current_max = grouped_links.keys().next_back().unwrap(); 1222 1222 + if fwd_target > *current_max { 1223 1223 + continue; 1224 1224 + } 1225 1225 + } 1226 1226 + 1227 1227 + // pagination, continued 1228 1228 + let mut should_evict = false; 1229 1229 + let entry = grouped_links.entry(fwd_target.clone()).or_insert_with(|| { 1230 1230 + should_evict = page_is_full; 1231 1231 + Vec::default() 1232 1232 + }); 1233 1233 + entry.push(RecordId { 1234 1234 + did, 1235 1235 + collection: collection.0.clone(), 1236 1236 + rkey: rkey.0, 1237 1237 + }); 1238 1238 + 1239 1239 + if should_evict { 1240 1240 + grouped_links.pop_last(); 1241 1241 + } 1242 1242 + } 1243 1243 + 1244 1244 + let mut items: Vec<(String, Vec<RecordId>)> = Vec::with_capacity(grouped_links.len()); 1245 1245 + for (fwd_target_id, records) in &grouped_links { 1246 1246 + let Some(target_key) = self 1247 1247 + .target_id_table 1248 1248 + .get_val_from_id(&self.db, fwd_target_id.0)? 1249 1249 + else { 1250 1250 + eprintln!("failed to look up target from target_id {fwd_target_id:?}"); 1251 1251 + continue; 1252 1252 + }; 1253 1253 + 1254 1254 + let target_string = target_key.0 .0; 1255 1255 + 1256 1256 + items.push((target_string, records.clone())); 1257 1257 + } 1258 1258 + 1259 1259 + let next = if grouped_links.len() as u64 >= limit { 1260 1260 + grouped_links.keys().next_back().map(|k| format!("{}", k.0)) 1261 1261 + } else { 1262 1262 + None 1263 1263 + }; 1264 1264 + 1265 1265 + Ok(PagedOrderedCollection { items, next }) 1266 1266 + } 1267 1267 + 1125 1268 fn get_links( 1126 1269 &self, 1127 1270 target: &str,

+60

constellation/templates/get-many-to-many.html.j2

··· 1 1 + {% extends "base.html.j2" %} 2 2 + {% import "try-it-macros.html.j2" as try_it %} 3 3 + 4 4 + {% block title %}Many-to-Many Links{% endblock %} 5 5 + {% block description %}All {{ query.source }} records with many-to-many links to {{ query.subject }} joining through {{ query.path_to_other }}{% endblock %} 6 6 + 7 7 + {% block content %} 8 8 + 9 9 + {% call try_it::get_many_to_many(query.subject, query.source, query.path_to_other, query.did, query.other_subject, query.limit) %} 10 10 + 11 11 + <h2> 12 12 + Many-to-many links to <code>{{ query.subject }}</code> 13 13 + {% if let Some(browseable_uri) = query.subject|to_browseable %} 14 14 + <small style="font-weight: normal; font-size: 1rem"><a href="{{ browseable_uri }}">browse record</a></small> 15 15 + {% endif %} 16 16 + </h2> 17 17 + 18 18 + <p><strong>Many-to-many links</strong> from <code>{{ query.source }}</code> joining through <code>{{ query.path_to_other }}</code></p> 19 19 + 20 20 + <ul> 21 21 + <li>See all links to this target at <code>/links/all</code>: <a href="/links/all?target={{ query.subject|urlencode }}">/links/all?target={{ query.subject }}</a></li> 22 22 + </ul> 23 23 + 24 24 + <h3>Many-to-many links, most recent first:</h3> 25 25 + 26 26 + {% for (target, records) in linking_records %} 27 27 + <h4>Target: <code>{{ target }}</code> <small>(<a href="/links/all?target={{ target|urlencode }}">view all links</a>)</small></h4> 28 28 + {% for record in records %} 29 29 + <pre style="display: block; margin: 1em 2em" class="code"><strong>DID</strong>: {{ record.did().0 }} 30 30 + <strong>Collection</strong>: {{ record.collection }} 31 31 + <strong>RKey</strong>: {{ record.rkey }} 32 32 + -> <a href="https://pdsls.dev/at://{{ record.did().0 }}/{{ record.collection }}/{{ record.rkey }}">browse record</a></pre> 33 33 + {% endfor %} 34 34 + {% endfor %} 35 35 + 36 36 + {% if let Some(c) = cursor %} 37 37 + <form method="get" action="/xrpc/blue.microcosm.links.getManyToMany"> 38 38 + <input type="hidden" name="subject" value="{{ query.subject }}" /> 39 39 + <input type="hidden" name="source" value="{{ query.source }}" /> 40 40 + <input type="hidden" name="pathToOther" value="{{ query.path_to_other }}" /> 41 41 + {% for did in query.did %} 42 42 + <input type="hidden" name="did" value="{{ did }}" /> 43 43 + {% endfor %} 44 44 + {% for other in query.other_subject %} 45 45 + <input type="hidden" name="otherSubject" value="{{ other }}" /> 46 46 + {% endfor %} 47 47 + <input type="hidden" name="limit" value="{{ query.limit }}" /> 48 48 + <input type="hidden" name="cursor" value={{ c|json|safe }} /> 49 49 + <button type="submit">next page&hellip;</button> 50 50 + </form> 51 51 + {% else %} 52 52 + <button disabled><em>end of results</em></button> 53 53 + {% endif %} 54 54 + 55 55 + <details> 56 56 + <summary>Raw JSON response</summary> 57 57 + <pre class="code">{{ self|tojson }}</pre> 58 58 + </details> 59 59 + 60 60 + {% endblock %}

+19

constellation/templates/hello.html.j2

··· 81 81 ) %} 82 82 83 83 84 84 + <h3 class="route"><code>GET /xrpc/blue.microcosm.links.getManyToMany</code></h3> 85 85 + 86 86 + <p>A list of many-to-many join records linking to a target and a secondary target.</p> 87 87 + 88 88 + <h4>Query parameters:</h4> 89 89 + 90 90 + <ul> 91 91 + <li><p><code>subject</code>: required, must url-encode. Example: <code>at://did:plc:vc7f4oafdgxsihk4cry2xpze/app.bsky.feed.post/3lgwdn7vd722r</code></p></li> 92 92 + <li><p><code>source</code>: required. Example: <code>app.bsky.feed.like:subject.uri</code></p></li> 93 93 + <li><p><code>pathToOther</code>: required. Path to the secondary link in the many-to-many record. Example: <code>otherThing.uri</code></p></li> 94 94 + <li><p><code>did</code>: optional, filter links to those from specific users. Include multiple times to filter by multiple users. Example: <code>did=did:plc:vc7f4oafdgxsihk4cry2xpze&did=did:plc:vc7f4oafdgxsihk4cry2xpze</code></p></li> 95 95 + <li><p><code>otherSubject</code>: optional, filter secondary links to specific subjects. Include multiple times to filter by multiple subjects. Example: <code>at://did:plc:vc7f4oafdgxsihk4cry2xpze/app.bsky.feed.post/3lgwdn7vd722r</code></p></li> 96 96 + <li><p><code>limit</code>: optional. Default: <code>16</code>. Maximum: <code>100</code></p></li> 97 97 + </ul> 98 98 + 99 99 + <p style="margin-bottom: 0"><strong>Try it:</strong></p> 100 100 + {% call try_it::get_many_to_many("at://did:plc:a4pqq234yw7fqbddawjo7y35/app.bsky.feed.post/3m237ilwc372e", "app.bsky.feed.like:subject.uri", "reply.parent.uri", [""], [""], 16) %} 101 101 + 102 102 + 84 103 <h3 class="route"><code>GET /links</code></h3> 85 104 86 105 <p>A list of records linking to a target.</p>

+30

constellation/templates/try-it-macros.html.j2

··· 66 66 </script> 67 67 {% endmacro %} 68 68 69 69 + {% macro get_many_to_many(subject, source, pathToOther, dids, otherSubjects, limit) %} 70 70 + <form method="get" action="/xrpc/blue.microcosm.links.getManyToMany"> 71 71 + <pre class="code"><strong>GET</strong> /xrpc/blue.microcosm.links.getManyToMany 72 72 + ?subject= <input type="text" name="subject" value="{{ subject }}" placeholder="at-uri, did, uri..." /> 73 73 + &source= <input type="text" name="source" value="{{ source }}" placeholder="app.bsky.feed.like:subject" /> 74 74 + &pathToOther= <input type="text" name="pathToOther" value="{{ pathToOther }}" placeholder="otherThing" /> 75 75 + {%- for did in dids %}{% if !did.is_empty() %} 76 76 + &did= <input type="text" name="did" value="{{ did }}" placeholder="did:plc:..." />{% endif %}{% endfor %} 77 77 + <span id="m2m-did-placeholder"></span> <button id="m2m-add-did">+ did filter</button> 78 78 + {%- for otherSubject in otherSubjects %}{% if !otherSubject.is_empty() %} 79 79 + &otherSubject= <input type="text" name="otherSubject" value="{{ otherSubject }}" placeholder="at-uri, did, uri..." />{% endif %}{% endfor %} 80 80 + <span id="m2m-other-placeholder"></span> <button id="m2m-add-other">+ other subject filter</button> 81 81 + &limit= <input type="number" name="limit" value="{{ limit }}" max="100" placeholder="100" /> <button type="submit">get many-to-many links</button></pre> 82 82 + </form> 83 83 + <script> 84 84 + const m2mAddDidButton = document.getElementById('m2m-add-did'); 85 85 + const m2mDidPlaceholder = document.getElementById('m2m-did-placeholder'); 86 86 + m2mAddDidButton.addEventListener('click', e => { 87 87 + e.preventDefault(); 88 88 + const i = document.createElement('input'); 89 89 + i.placeholder = 'did:plc:...'; 90 90 + i.name = "did" 91 91 + const p = m2mAddDidButton.parentNode; 92 92 + p.insertBefore(document.createTextNode('&did= '), m2mDidPlaceholder); 93 93 + p.insertBefore(i, m2mDidPlaceholder); 94 94 + p.insertBefore(document.createTextNode('\n '), m2mDidPlaceholder); 95 95 + }); 96 96 + </script> 97 97 + {% endmacro %} 98 98 + 69 99 {% macro links(target, collection, path, dids, limit) %} 70 100 <form method="get" action="/links"> 71 101 <pre class="code"><strong>GET</strong> /links