microcosm.blue / microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
fork atom

Add new get_many_to_many XRPC endpoint #7

merged opened by seoul.systems targeting main from seoul.systems/microcosm-rs: xrpc_many2many

Added a new XRPC API endpoint to fetch joined record URIs, termed get_many_to_many (we talked about this briefly on Discord already). It is implemented and functions almost identical to the existing get_many_to_many_counts endpoint and handler. Some of its possible flaws like the two step lookup to verify a matching DID is indeed active are duplicated as well. On the plus side, this should make the PR pretty straightforward to review and make it easier to modify both endpoints later on when a more efficient way to validate the status of DIDs is possible.

If you have comments remarks etc. I am happy to work on some parts again.

Labels

None yet.

Participants 2
AT URI
at://did:plc:53wellrw53o7sw4zlpfenvuh/sh.tangled.repo.pull/3mbkyehqooh22
+813 -5
Diff #1
unified split
+7 -1
constellation/src/lib.rs
··· 31 31 } 32 32 } 33 33 34 - #[derive(Debug, PartialEq, Serialize, Deserialize)] 34 + #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)] 35 35 pub struct RecordId { 36 36 pub did: Did, 37 37 pub collection: String, ··· 48 48 pub fn rkey(&self) -> String { 49 49 self.rkey.clone() 50 50 } 51 + } 52 + 53 + #[derive(Debug, PartialEq, Serialize, Deserialize, Clone, Default)] 54 + pub struct RecordsBySubject { 55 + pub subject: String, 56 + pub records: Vec<RecordId>, 51 57 } 52 58 53 59 /// maybe the worst type in this repo, and there are some bad types
+103 -1
constellation/src/server/mod.rs
··· 18 18 use tokio_util::sync::CancellationToken; 19 19 20 20 use crate::storage::{LinkReader, StorageStats}; 21 - use crate::{CountsByCount, Did, RecordId}; 21 + use crate::{CountsByCount, Did, RecordId, RecordsBySubject}; 22 22 23 23 mod acceptable; 24 24 mod filters; ··· 83 83 let store = store.clone(); 84 84 move |accept, query| async { 85 85 spawn_blocking(|| count_distinct_dids(accept, query, store)) 86 + .await 87 + .map_err(to500)? 88 + } 89 + }), 90 + ) 91 + .route( 92 + "/xrpc/blue.microcosm.links.getManyToMany", 93 + get({ 94 + let store = store.clone(); 95 + move |accept, query| async { 96 + spawn_blocking(|| get_many_to_many(accept, query, store)) 86 97 .await 87 98 .map_err(to500)? 88 99 } ··· 580 591 accept, 581 592 GetLinkItemsResponse { 582 593 total: paged.total, 594 + linking_records: paged.items, 595 + cursor, 596 + query: (*query).clone(), 597 + }, 598 + )) 599 + } 600 + 601 + #[derive(Clone, Deserialize)] 602 + #[serde(rename_all = "camelCase")] 603 + struct GetManyToManyItemsQuery { 604 + subject: String, 605 + source: String, 606 + /// path to the secondary link in the linking record 607 + path_to_other: String, 608 + /// filter to linking records (join of the m2m) by these DIDs 609 + #[serde(default)] 610 + did: Vec<String>, 611 + /// filter to specific secondary records 612 + #[serde(default)] 613 + other_subject: Vec<String>, 614 + cursor: Option<OpaqueApiCursor>, 615 + #[serde(default = "get_default_cursor_limit")] 616 + limit: u64, 617 + } 618 + #[derive(Template, Serialize)] 619 + #[template(path = "get-many-to-many.html.j2")] 620 + struct GetManyToManyItemsResponse { 621 + linking_records: Vec<RecordsBySubject>, 622 + cursor: Option<OpaqueApiCursor>, 623 + #[serde(skip_serializing)] 624 + query: GetManyToManyItemsQuery, 625 + } 626 + fn get_many_to_many( 627 + accept: ExtractAccept, 628 + query: axum_extra::extract::Query<GetManyToManyItemsQuery>, // supports multiple param occurrences 629 + store: impl LinkReader, 630 + ) -> Result<impl IntoResponse, http::StatusCode> { 631 + let after = query 632 + .cursor 633 + .clone() 634 + .map(|oc| ApiKeyedCursor::try_from(oc).map_err(|_| http::StatusCode::BAD_REQUEST)) 635 + .transpose()? 636 + .map(|c| c.next); 637 + 638 + let limit = query.limit; 639 + if limit > DEFAULT_CURSOR_LIMIT_MAX { 640 + return Err(http::StatusCode::BAD_REQUEST); 641 + } 642 + 643 + let filter_dids: HashSet<Did> = HashSet::from_iter( 644 + query 645 + .did 646 + .iter() 647 + .map(|d| d.trim()) 648 + .filter(|d| !d.is_empty()) 649 + .map(|d| Did(d.to_string())), 650 + ); 651 + 652 + let filter_other_subjects: HashSet<String> = HashSet::from_iter( 653 + query 654 + .other_subject 655 + .iter() 656 + .map(|s| s.trim().to_string()) 657 + .filter(|s| !s.is_empty()), 658 + ); 659 + 660 + let Some((collection, path)) = query.source.split_once(':') else { 661 + return Err(http::StatusCode::BAD_REQUEST); 662 + }; 663 + let path = format!(".{path}"); 664 + 665 + let path_to_other = format!(".{}", query.path_to_other); 666 + 667 + let paged = store 668 + .get_many_to_many( 669 + &query.subject, 670 + collection, 671 + &path, 672 + &path_to_other, 673 + limit, 674 + after, 675 + &filter_dids, 676 + &filter_other_subjects, 677 + ) 678 + .map_err(|_| http::StatusCode::INTERNAL_SERVER_ERROR)?; 679 + 680 + let cursor = paged.next.map(|next| ApiKeyedCursor { next }.into()); 681 + 682 + Ok(acceptable( 683 + accept, 684 + GetManyToManyItemsResponse { 583 685 linking_records: paged.items, 584 686 cursor, 585 687 query: (*query).clone(),
+97 -1
constellation/src/storage/mem_store.rs
··· 1 1 use super::{ 2 2 LinkReader, LinkStorage, PagedAppendingCollection, PagedOrderedCollection, StorageStats, 3 3 }; 4 - use crate::{ActionableEvent, CountsByCount, Did, RecordId}; 4 + use crate::{ActionableEvent, CountsByCount, Did, RecordId, RecordsBySubject}; 5 5 use anyhow::Result; 6 6 use links::CollectedLink; 7 7 use std::collections::{HashMap, HashSet}; ··· 232 232 .map(|(did, _)| did) 233 233 .collect::<HashSet<_>>() 234 234 .len() as u64) 235 + } 236 + 237 + fn get_many_to_many( 238 + &self, 239 + target: &str, 240 + collection: &str, 241 + path: &str, 242 + path_to_other: &str, 243 + limit: u64, 244 + after: Option<String>, 245 + filter_dids: &HashSet<Did>, 246 + filter_to_targets: &HashSet<String>, 247 + ) -> Result<PagedOrderedCollection<RecordsBySubject, String>> { 248 + let empty_res = Ok(PagedOrderedCollection { 249 + items: Vec::new(), 250 + next: None, 251 + }); 252 + 253 + // struct MemStorageData { 254 + // dids: HashMap<Did, bool>, 255 + // targets: HashMap<Target, HashMap<Source, Linkers>>, 256 + // links: HashMap<Did, HashMap<RepoId, Vec<(RecordPath, Target)>>>, 257 + // } 258 + let data = self.0.lock().unwrap(); 259 + 260 + let Some(sources) = data.targets.get(&Target::new(target)) else { 261 + return empty_res; 262 + }; 263 + let Some(linkers) = sources.get(&Source::new(collection, path)) else { 264 + return empty_res; 265 + }; 266 + let path_to_other = RecordPath::new(path_to_other); 267 + 268 + // Convert filter_to_targets to Target objects for comparison 269 + let filter_to_target_objs: HashSet<Target> = 270 + HashSet::from_iter(filter_to_targets.iter().map(|s| Target::new(s))); 271 + 272 + let mut grouped_links: HashMap<Target, Vec<RecordId>> = HashMap::new(); 273 + for (did, rkey) in linkers.iter().flatten().cloned() { 274 + // Filter by DID if filter is provided 275 + if !filter_dids.is_empty() && !filter_dids.contains(&did) { 276 + continue; 277 + } 278 + if let Some(fwd_target) = data 279 + .links 280 + .get(&did) 281 + .unwrap_or(&HashMap::new()) 282 + .get(&RepoId { 283 + collection: collection.to_string(), 284 + rkey: rkey.clone(), 285 + }) 286 + .unwrap_or(&Vec::new()) 287 + .iter() 288 + .find_map(|(path, target)| { 289 + if *path == path_to_other 290 + && (filter_to_target_objs.is_empty() 291 + || filter_to_target_objs.contains(target)) 292 + { 293 + Some(target) 294 + } else { 295 + None 296 + } 297 + }) 298 + { 299 + let record_ids = grouped_links.entry(fwd_target.clone()).or_default(); 300 + record_ids.push(RecordId { 301 + did, 302 + collection: collection.to_string(), 303 + rkey: rkey.0, 304 + }); 305 + } 306 + } 307 + 308 + let mut items = grouped_links 309 + .into_iter() 310 + .map(|(t, r)| RecordsBySubject { 311 + subject: t.0, 312 + records: r, 313 + }) 314 + .collect::<Vec<_>>(); 315 + 316 + items.sort_by(|a, b| a.subject.cmp(&b.subject)); 317 + 318 + items = items 319 + .into_iter() 320 + .skip_while(|item| after.as_ref().map(|a| &item.subject <= a).unwrap_or(false)) 321 + .take(limit as usize) 322 + .collect(); 323 + 324 + let next = if items.len() as u64 >= limit { 325 + items.last().map(|item| item.subject.clone()) 326 + } else { 327 + None 328 + }; 329 + 330 + Ok(PagedOrderedCollection { items, next }) 235 331 } 236 332 237 333 fn get_links(
+241 -1
constellation/src/storage/mod.rs
··· 1 - use crate::{ActionableEvent, CountsByCount, Did, RecordId}; 1 + use crate::{ActionableEvent, CountsByCount, Did, RecordId, RecordsBySubject}; 2 2 use anyhow::Result; 3 3 use serde::{Deserialize, Serialize}; 4 4 use std::collections::{HashMap, HashSet}; ··· 103 103 104 104 fn get_all_record_counts(&self, _target: &str) 105 105 -> Result<HashMap<String, HashMap<String, u64>>>; 106 + 107 + fn get_many_to_many( 108 + &self, 109 + target: &str, 110 + collection: &str, 111 + path: &str, 112 + path_to_other: &str, 113 + limit: u64, 114 + after: Option<String>, 115 + filter_dids: &HashSet<Did>, 116 + filter_to_targets: &HashSet<String>, 117 + ) -> Result<PagedOrderedCollection<RecordsBySubject, String>>; 106 118 107 119 fn get_all_counts( 108 120 &self, ··· 1551 1563 next: None, 1552 1564 } 1553 1565 ); 1566 + }); 1567 + 1568 + test_each_storage!(get_m2m_empty, |storage| { 1569 + assert_eq!( 1570 + storage.get_many_to_many( 1571 + "a.com", 1572 + "a.b.c", 1573 + ".d.e", 1574 + ".f.g", 1575 + 10, 1576 + None, 1577 + &HashSet::new(), 1578 + &HashSet::new(), 1579 + )?, 1580 + PagedOrderedCollection { 1581 + items: vec![], 1582 + next: None, 1583 + } 1584 + ); 1585 + }); 1586 + 1587 + test_each_storage!(get_m2m_single, |storage| { 1588 + storage.push( 1589 + &ActionableEvent::CreateLinks { 1590 + record_id: RecordId { 1591 + did: "did:plc:asdf".into(), 1592 + collection: "app.t.c".into(), 1593 + rkey: "asdf".into(), 1594 + }, 1595 + links: vec![ 1596 + CollectedLink { 1597 + target: Link::Uri("a.com".into()), 1598 + path: ".abc.uri".into(), 1599 + }, 1600 + CollectedLink { 1601 + target: Link::Uri("b.com".into()), 1602 + path: ".def.uri".into(), 1603 + }, 1604 + CollectedLink { 1605 + target: Link::Uri("b.com".into()), 1606 + path: ".ghi.uri".into(), 1607 + }, 1608 + ], 1609 + }, 1610 + 0, 1611 + )?; 1612 + assert_eq!( 1613 + storage.get_many_to_many( 1614 + "a.com", 1615 + "app.t.c", 1616 + ".abc.uri", 1617 + ".def.uri", 1618 + 10, 1619 + None, 1620 + &HashSet::new(), 1621 + &HashSet::new(), 1622 + )?, 1623 + PagedOrderedCollection { 1624 + items: vec![RecordsBySubject { 1625 + subject: "b.com".to_string(), 1626 + records: vec![RecordId { 1627 + did: "did:plc:asdf".into(), 1628 + collection: "app.t.c".into(), 1629 + rkey: "asdf".into(), 1630 + }] 1631 + }], 1632 + next: None, 1633 + } 1634 + ); 1635 + }); 1636 + 1637 + test_each_storage!(get_m2m_filters, |storage| { 1638 + storage.push( 1639 + &ActionableEvent::CreateLinks { 1640 + record_id: RecordId { 1641 + did: "did:plc:asdf".into(), 1642 + collection: "app.t.c".into(), 1643 + rkey: "asdf".into(), 1644 + }, 1645 + links: vec![ 1646 + CollectedLink { 1647 + target: Link::Uri("a.com".into()), 1648 + path: ".abc.uri".into(), 1649 + }, 1650 + CollectedLink { 1651 + target: Link::Uri("b.com".into()), 1652 + path: ".def.uri".into(), 1653 + }, 1654 + ], 1655 + }, 1656 + 0, 1657 + )?; 1658 + storage.push( 1659 + &ActionableEvent::CreateLinks { 1660 + record_id: RecordId { 1661 + did: "did:plc:asdf".into(), 1662 + collection: "app.t.c".into(), 1663 + rkey: "asdf2".into(), 1664 + }, 1665 + links: vec![ 1666 + CollectedLink { 1667 + target: Link::Uri("a.com".into()), 1668 + path: ".abc.uri".into(), 1669 + }, 1670 + CollectedLink { 1671 + target: Link::Uri("b.com".into()), 1672 + path: ".def.uri".into(), 1673 + }, 1674 + ], 1675 + }, 1676 + 1, 1677 + )?; 1678 + storage.push( 1679 + &ActionableEvent::CreateLinks { 1680 + record_id: RecordId { 1681 + did: "did:plc:fdsa".into(), 1682 + collection: "app.t.c".into(), 1683 + rkey: "fdsa".into(), 1684 + }, 1685 + links: vec![ 1686 + CollectedLink { 1687 + target: Link::Uri("a.com".into()), 1688 + path: ".abc.uri".into(), 1689 + }, 1690 + CollectedLink { 1691 + target: Link::Uri("c.com".into()), 1692 + path: ".def.uri".into(), 1693 + }, 1694 + ], 1695 + }, 1696 + 2, 1697 + )?; 1698 + storage.push( 1699 + &ActionableEvent::CreateLinks { 1700 + record_id: RecordId { 1701 + did: "did:plc:fdsa".into(), 1702 + collection: "app.t.c".into(), 1703 + rkey: "fdsa2".into(), 1704 + }, 1705 + links: vec![ 1706 + CollectedLink { 1707 + target: Link::Uri("a.com".into()), 1708 + path: ".abc.uri".into(), 1709 + }, 1710 + CollectedLink { 1711 + target: Link::Uri("c.com".into()), 1712 + path: ".def.uri".into(), 1713 + }, 1714 + ], 1715 + }, 1716 + 3, 1717 + )?; 1718 + 1719 + // Test without filters - should get all records grouped by secondary target 1720 + let result = storage.get_many_to_many( 1721 + "a.com", 1722 + "app.t.c", 1723 + ".abc.uri", 1724 + ".def.uri", 1725 + 10, 1726 + None, 1727 + &HashSet::new(), 1728 + &HashSet::new(), 1729 + )?; 1730 + assert_eq!(result.items.len(), 2); 1731 + assert_eq!(result.next, None); 1732 + // Find b.com group 1733 + let b_group = result 1734 + .items 1735 + .iter() 1736 + .find(|group| group.subject == "b.com") 1737 + .unwrap(); 1738 + assert_eq!(b_group.subject, "b.com"); 1739 + assert_eq!(b_group.records.len(), 2); 1740 + assert!(b_group.records 1741 + .iter() 1742 + .any(|r| r.did.0 == "did:plc:asdf" && r.rkey == "asdf")); 1743 + assert!(b_group.records 1744 + .iter() 1745 + .any(|r| r.did.0 == "did:plc:asdf" && r.rkey == "asdf2")); 1746 + // Find c.com group 1747 + let c_group = result 1748 + .items 1749 + .iter() 1750 + .find(|group| group.subject == "c.com") 1751 + .unwrap(); 1752 + assert_eq!(c_group.subject, "c.com"); 1753 + assert_eq!(c_group.records.len(), 2); 1754 + assert!(c_group.records 1755 + .iter() 1756 + .any(|r| r.did.0 == "did:plc:fdsa" && r.rkey == "fdsa")); 1757 + assert!(c_group.records 1758 + .iter() 1759 + .any(|r| r.did.0 == "did:plc:fdsa" && r.rkey == "fdsa2")); 1760 + 1761 + // Test with DID filter - should only get records from did:plc:fdsa 1762 + let result = storage.get_many_to_many( 1763 + "a.com", 1764 + "app.t.c", 1765 + ".abc.uri", 1766 + ".def.uri", 1767 + 10, 1768 + None, 1769 + &HashSet::from_iter([Did("did:plc:fdsa".to_string())]), 1770 + &HashSet::new(), 1771 + )?; 1772 + assert_eq!(result.items.len(), 1); 1773 + let group = &result.items[0]; 1774 + assert_eq!(group.subject, "c.com"); 1775 + assert_eq!(group.records.len(), 2); 1776 + assert!(group.records.iter().all(|r| r.did.0 == "did:plc:fdsa")); 1777 + 1778 + // Test with target filter - should only get records linking to b.com 1779 + let result = storage.get_many_to_many( 1780 + "a.com", 1781 + "app.t.c", 1782 + ".abc.uri", 1783 + ".def.uri", 1784 + 10, 1785 + None, 1786 + &HashSet::new(), 1787 + &HashSet::from_iter(["b.com".to_string()]), 1788 + )?; 1789 + assert_eq!(result.items.len(), 1); 1790 + let group = &result.items[0]; 1791 + assert_eq!(group.subject, "b.com"); 1792 + assert_eq!(group.records.len(), 2); 1793 + assert!(group.records.iter().all(|r| r.did.0 == "did:plc:asdf")); 1554 1794 }); 1555 1795 }
+147 -1
constellation/src/storage/rocks_store.rs
··· 2 2 ActionableEvent, LinkReader, LinkStorage, PagedAppendingCollection, PagedOrderedCollection, 3 3 StorageStats, 4 4 }; 5 - use crate::{CountsByCount, Did, RecordId}; 5 + use crate::{CountsByCount, Did, RecordId, RecordsBySubject}; 6 6 use anyhow::{bail, Result}; 7 7 use bincode::Options as BincodeOptions; 8 8 use links::CollectedLink; ··· 1120 1120 } else { 1121 1121 Ok(0) 1122 1122 } 1123 + } 1124 + 1125 + fn get_many_to_many( 1126 + &self, 1127 + target: &str, 1128 + collection: &str, 1129 + path: &str, 1130 + path_to_other: &str, 1131 + limit: u64, 1132 + after: Option<String>, 1133 + filter_dids: &HashSet<Did>, 1134 + filter_to_targets: &HashSet<String>, 1135 + ) -> Result<PagedOrderedCollection<RecordsBySubject, String>> { 1136 + let collection = Collection(collection.to_string()); 1137 + let path = RPath(path.to_string()); 1138 + 1139 + let target_key = TargetKey(Target(target.to_string()), collection.clone(), path); 1140 + 1141 + let after = after.map(|s| s.parse::<u64>().map(TargetId)).transpose()?; 1142 + 1143 + let Some(target_id) = self.target_id_table.get_id_val(&self.db, &target_key)? else { 1144 + eprintln!("Target not found for {target_key:?}"); 1145 + return Ok(Default::default()); 1146 + }; 1147 + 1148 + let filter_did_ids: HashMap<DidId, bool> = filter_dids 1149 + .iter() 1150 + .filter_map(|did| self.did_id_table.get_id_val(&self.db, did).transpose()) 1151 + .collect::<Result<Vec<DidIdValue>>>()? 1152 + .into_iter() 1153 + .map(|DidIdValue(id, active)| (id, active)) 1154 + .collect(); 1155 + 1156 + let mut filter_to_target_ids: HashSet<TargetId> = HashSet::new(); 1157 + for t in filter_to_targets { 1158 + for (_, target_id) in self.iter_targets_for_target(&Target(t.to_string())) { 1159 + filter_to_target_ids.insert(target_id); 1160 + } 1161 + } 1162 + 1163 + let linkers = self.get_target_linkers(&target_id)?; 1164 + 1165 + // we want to provide many to many which effectively means that we want to show a specific 1166 + // list of reords that is linked to by a specific number of linkers 1167 + let mut grouped_links: BTreeMap<TargetId, Vec<RecordId>> = BTreeMap::new(); 1168 + for (did_id, rkey) in linkers.0 { 1169 + if did_id.is_empty() { 1170 + continue; 1171 + } 1172 + 1173 + if !filter_did_ids.is_empty() && filter_did_ids.get(&did_id) != Some(&true) { 1174 + continue; 1175 + } 1176 + 1177 + // Make sure the current did is active 1178 + let Some(did) = self.did_id_table.get_val_from_id(&self.db, did_id.0)? else { 1179 + eprintln!("failed to look up did from did_id {did_id:?}"); 1180 + continue; 1181 + }; 1182 + let Some(DidIdValue(_, active)) = self.did_id_table.get_id_val(&self.db, &did)? else { 1183 + eprintln!("failed to look up did_value from did_id {did_id:?}: {did:?}: data consistency bug?"); 1184 + continue; 1185 + }; 1186 + if !active { 1187 + continue; 1188 + } 1189 + 1190 + let record_link_key = RecordLinkKey(did_id, collection.clone(), rkey.clone()); 1191 + let Some(targets) = self.get_record_link_targets(&record_link_key)? else { 1192 + continue; 1193 + }; 1194 + 1195 + let Some(fwd_target) = targets 1196 + .0 1197 + .into_iter() 1198 + .filter_map(|RecordLinkTarget(rpath, target_id)| { 1199 + if rpath.0 == path_to_other 1200 + && (filter_to_target_ids.is_empty() 1201 + || filter_to_target_ids.contains(&target_id)) 1202 + { 1203 + Some(target_id) 1204 + } else { 1205 + None 1206 + } 1207 + }) 1208 + .take(1) 1209 + .next() 1210 + else { 1211 + eprintln!("no forward match found."); 1212 + continue; 1213 + }; 1214 + 1215 + // pagination logic mirrors what is currently done in get_many_to_many_counts 1216 + if after.as_ref().map(|a| fwd_target <= *a).unwrap_or(false) { 1217 + continue; 1218 + } 1219 + let page_is_full = grouped_links.len() as u64 >= limit; 1220 + if page_is_full { 1221 + let current_max = grouped_links.keys().next_back().unwrap(); 1222 + if fwd_target > *current_max { 1223 + continue; 1224 + } 1225 + } 1226 + 1227 + // pagination, continued 1228 + let mut should_evict = false; 1229 + let entry = grouped_links.entry(fwd_target.clone()).or_insert_with(|| { 1230 + should_evict = page_is_full; 1231 + Vec::default() 1232 + }); 1233 + entry.push(RecordId { 1234 + did, 1235 + collection: collection.0.clone(), 1236 + rkey: rkey.0, 1237 + }); 1238 + 1239 + if should_evict { 1240 + grouped_links.pop_last(); 1241 + } 1242 + } 1243 + 1244 + let mut items: Vec<RecordsBySubject> = Vec::with_capacity(grouped_links.len()); 1245 + for (fwd_target_id, records) in &grouped_links { 1246 + let Some(target_key) = self 1247 + .target_id_table 1248 + .get_val_from_id(&self.db, fwd_target_id.0)? 1249 + else { 1250 + eprintln!("failed to look up target from target_id {fwd_target_id:?}"); 1251 + continue; 1252 + }; 1253 + 1254 + let target_string = target_key.0 .0; 1255 + 1256 + items.push(RecordsBySubject { 1257 + subject: target_string, 1258 + records: records.clone(), 1259 + }); 1260 + } 1261 + 1262 + let next = if grouped_links.len() as u64 >= limit { 1263 + grouped_links.keys().next_back().map(|k| format!("{}", k.0)) 1264 + } else { 1265 + None 1266 + }; 1267 + 1268 + Ok(PagedOrderedCollection { items, next }) 1123 1269 } 1124 1270 1125 1271 fn get_links(
+60
constellation/templates/get-many-to-many.html.j2
··· 1 + {% extends "base.html.j2" %} 2 + {% import "try-it-macros.html.j2" as try_it %} 3 + 4 + {% block title %}Many-to-Many Links{% endblock %} 5 + {% block description %}All {{ query.source }} records with many-to-many links to {{ query.subject }} joining through {{ query.path_to_other }}{% endblock %} 6 + 7 + {% block content %} 8 + 9 + {% call try_it::get_many_to_many(query.subject, query.source, query.path_to_other, query.did, query.other_subject, query.limit) %} 10 + 11 + <h2> 12 + Many-to-many links to <code>{{ query.subject }}</code> 13 + {% if let Some(browseable_uri) = query.subject|to_browseable %} 14 + <small style="font-weight: normal; font-size: 1rem"><a href="{{ browseable_uri }}">browse record</a></small> 15 + {% endif %} 16 + </h2> 17 + 18 + <p><strong>Many-to-many links</strong> from <code>{{ query.source }}</code> joining through <code>{{ query.path_to_other }}</code></p> 19 + 20 + <ul> 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 + </ul> 23 + 24 + <h3>Many-to-many links, most recent first:</h3> 25 + 26 + {% for group in linking_records %} 27 + <h4>Target: <code>{{ group.subject }}</code> <small>(<a href="/links/all?target={{ group.subject|urlencode }}">view all links</a>)</small></h4> 28 + {% for record in group.records %} 29 + <pre style="display: block; margin: 1em 2em" class="code"><strong>DID</strong>: {{ record.did().0 }} 30 + <strong>Collection</strong>: {{ record.collection }} 31 + <strong>RKey</strong>: {{ record.rkey }} 32 + -> <a href="https://pdsls.dev/at://{{ record.did().0 }}/{{ record.collection }}/{{ record.rkey }}">browse record</a></pre> 33 + {% endfor %} 34 + {% endfor %} 35 + 36 + {% if let Some(c) = cursor %} 37 + <form method="get" action="/xrpc/blue.microcosm.links.getManyToMany"> 38 + <input type="hidden" name="subject" value="{{ query.subject }}" /> 39 + <input type="hidden" name="source" value="{{ query.source }}" /> 40 + <input type="hidden" name="pathToOther" value="{{ query.path_to_other }}" /> 41 + {% for did in query.did %} 42 + <input type="hidden" name="did" value="{{ did }}" /> 43 + {% endfor %} 44 + {% for other in query.other_subject %} 45 + <input type="hidden" name="otherSubject" value="{{ other }}" /> 46 + {% endfor %} 47 + <input type="hidden" name="limit" value="{{ query.limit }}" /> 48 + <input type="hidden" name="cursor" value={{ c|json|safe }} /> 49 + <button type="submit">next page&hellip;</button> 50 + </form> 51 + {% else %} 52 + <button disabled><em>end of results</em></button> 53 + {% endif %} 54 + 55 + <details> 56 + <summary>Raw JSON response</summary> 57 + <pre class="code">{{ self|tojson }}</pre> 58 + </details> 59 + 60 + {% endblock %}
+19
constellation/templates/hello.html.j2
··· 81 81 ) %} 82 82 83 83 84 + <h3 class="route"><code>GET /xrpc/blue.microcosm.links.getManyToMany</code></h3> 85 + 86 + <p>A list of many-to-many join records linking to a target and a secondary target.</p> 87 + 88 + <h4>Query parameters:</h4> 89 + 90 + <ul> 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 + <li><p><code>source</code>: required. Example: <code>app.bsky.feed.like:subject.uri</code></p></li> 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 + <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 + <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 + <li><p><code>limit</code>: optional. Default: <code>16</code>. Maximum: <code>100</code></p></li> 97 + </ul> 98 + 99 + <p style="margin-bottom: 0"><strong>Try it:</strong></p> 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 + 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 + {% macro get_many_to_many(subject, source, pathToOther, dids, otherSubjects, limit) %} 70 + <form method="get" action="/xrpc/blue.microcosm.links.getManyToMany"> 71 + <pre class="code"><strong>GET</strong> /xrpc/blue.microcosm.links.getManyToMany 72 + ?subject= <input type="text" name="subject" value="{{ subject }}" placeholder="at-uri, did, uri..." /> 73 + &source= <input type="text" name="source" value="{{ source }}" placeholder="app.bsky.feed.like:subject" /> 74 + &pathToOther= <input type="text" name="pathToOther" value="{{ pathToOther }}" placeholder="otherThing" /> 75 + {%- for did in dids %}{% if !did.is_empty() %} 76 + &did= <input type="text" name="did" value="{{ did }}" placeholder="did:plc:..." />{% endif %}{% endfor %} 77 + <span id="m2m-did-placeholder"></span> <button id="m2m-add-did">+ did filter</button> 78 + {%- for otherSubject in otherSubjects %}{% if !otherSubject.is_empty() %} 79 + &otherSubject= <input type="text" name="otherSubject" value="{{ otherSubject }}" placeholder="at-uri, did, uri..." />{% endif %}{% endfor %} 80 + <span id="m2m-other-placeholder"></span> <button id="m2m-add-other">+ other subject filter</button> 81 + &limit= <input type="number" name="limit" value="{{ limit }}" max="100" placeholder="100" /> <button type="submit">get many-to-many links</button></pre> 82 + </form> 83 + <script> 84 + const m2mAddDidButton = document.getElementById('m2m-add-did'); 85 + const m2mDidPlaceholder = document.getElementById('m2m-did-placeholder'); 86 + m2mAddDidButton.addEventListener('click', e => { 87 + e.preventDefault(); 88 + const i = document.createElement('input'); 89 + i.placeholder = 'did:plc:...'; 90 + i.name = "did" 91 + const p = m2mAddDidButton.parentNode; 92 + p.insertBefore(document.createTextNode('&did= '), m2mDidPlaceholder); 93 + p.insertBefore(i, m2mDidPlaceholder); 94 + p.insertBefore(document.createTextNode('\n '), m2mDidPlaceholder); 95 + }); 96 + </script> 97 + {% endmacro %} 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
+109
lexicons/blue.microcosm/links/getManyToMany.json
··· 1 + { 2 + "lexicon": 1, 3 + "id": "blue.microcosm.links.getManyToMany", 4 + "defs": { 5 + "main": { 6 + "type": "query", 7 + "description": "Get records that link to a primary subject, grouped by the secondary subjects they also reference", 8 + "parameters": { 9 + "type": "params", 10 + "required": ["subject", "source", "pathToOther"], 11 + "properties": { 12 + "subject": { 13 + "type": "string", 14 + "format": "uri", 15 + "description": "the primary target being linked to (at-uri, did, or uri)" 16 + }, 17 + "source": { 18 + "type": "string", 19 + "description": "collection and path specification for the primary link (e.g., 'app.bsky.feed.like:subject.uri')" 20 + }, 21 + "pathToOther": { 22 + "type": "string", 23 + "description": "path to the secondary link in the many-to-many record (e.g., 'otherThing.uri')" 24 + }, 25 + "did": { 26 + "type": "array", 27 + "description": "filter links to those from specific users", 28 + "items": { 29 + "type": "string", 30 + "format": "did" 31 + } 32 + }, 33 + "otherSubject": { 34 + "type": "array", 35 + "description": "filter secondary links to specific subjects", 36 + "items": { 37 + "type": "string" 38 + } 39 + }, 40 + "limit": { 41 + "type": "integer", 42 + "minimum": 1, 43 + "maximum": 100, 44 + "default": 16, 45 + "description": "number of results to return" 46 + } 47 + } 48 + }, 49 + "output": { 50 + "encoding": "application/json", 51 + "schema": { 52 + "type": "object", 53 + "required": ["linking_records"], 54 + "properties": { 55 + "linking_records": { 56 + "type": "array", 57 + "items": { 58 + "type": "ref", 59 + "ref": "#recordsBySubject" 60 + } 61 + }, 62 + "cursor": { 63 + "type": "string", 64 + "description": "pagination cursor" 65 + } 66 + } 67 + } 68 + } 69 + }, 70 + "recordsBySubject": { 71 + "type": "object", 72 + "required": ["subject", "records"], 73 + "properties": { 74 + "subject": { 75 + "type": "string", 76 + "description": "the secondary subject that these records link to" 77 + }, 78 + "records": { 79 + "type": "array", 80 + "items": { 81 + "type": "ref", 82 + "ref": "#linkRecord" 83 + } 84 + } 85 + } 86 + }, 87 + "linkRecord": { 88 + "type": "object", 89 + "required": ["did", "collection", "rkey"], 90 + "description": "A record identifier consisting of a DID, collection, and record key", 91 + "properties": { 92 + "did": { 93 + "type": "string", 94 + "format": "did", 95 + "description": "the DID of the linking record's repository" 96 + }, 97 + "collection": { 98 + "type": "string", 99 + "format": "nsid", 100 + "description": "the collection of the linking record" 101 + }, 102 + "rkey": { 103 + "type": "string", 104 + "format": "record-key" 105 + } 106 + } 107 + } 108 + } 109 + }

History

8 rounds 13 comments
sign up or login to add to the discussion
seoul.systems submitted #7
diff interdiff
11 commits
expand
45e798b8
wip: m2m
f3144f8d
Add tests for new get_many_to_many query handler
21057eca
Fix get_m2m_empty test
4d686c35
Replace tuple with RecordsBySubject struct
4e358f1a
Fix conflicts after rebasing on main
bb107c03
Use record_id/subject tuple as return type for get_many_to_many
90220d0b
Fix get_many_to_many pagination with composite cursor
373d0744
Fix get_many_to_many_counts pagination with fetch N+1
46f5fae4
wip
6d22f695
Fix rocks-store to match mem-store composite cursor
c2d01284
Address feedback from fig
expand 0 comments
pull request successfully merged
seoul.systems submitted #6
diff interdiff
10 commits
expand
45e798b8
wip: m2m
f3144f8d
Add tests for new get_many_to_many query handler
21057eca
Fix get_m2m_empty test
4d686c35
Replace tuple with RecordsBySubject struct
4e358f1a
Fix conflicts after rebasing on main
bb107c03
Use record_id/subject tuple as return type for get_many_to_many
90220d0b
Fix get_many_to_many pagination with composite cursor
373d0744
Fix get_many_to_many_counts pagination with fetch N+1
46f5fae4
wip
6d22f695
Fix rocks-store to match mem-store composite cursor
expand 0 comments
seoul.systems submitted #5
diff interdiff
8 commits
expand
45e798b8
wip: m2m
f3144f8d
Add tests for new get_many_to_many query handler
21057eca
Fix get_m2m_empty test
4d686c35
Replace tuple with RecordsBySubject struct
4e358f1a
Fix conflicts after rebasing on main
bb107c03
Use record_id/subject tuple as return type for get_many_to_many
90220d0b
Fix get_many_to_many pagination with composite cursor
373d0744
Fix get_many_to_many_counts pagination with fetch N+1
expand 1 comment
seoul.systems

Okay. I wrapped my head around the composite cursor you proposed and am working on refactoring both storage implementations towards that. I think I might re-submit another round tomorrow :)

seoul.systems submitted #4
diff interdiff
6 commits
expand
45e798b8
wip: m2m
f3144f8d
Add tests for new get_many_to_many query handler
21057eca
Fix get_m2m_empty test
4d686c35
Replace tuple with RecordsBySubject struct
4e358f1a
Fix conflicts after rebasing on main
bb107c03
Use record_id/subject tuple as return type for get_many_to_many
expand 3 comments
seoul.systems

Found a bug in how we handle some of the pagination logic in cases where the number of items and the user selected limit are identical to very close too each other (already working on a fix)

bad-example.com

thanks for the rebase! i tried to write things in the tiny text box but ended up needing to make a diagram: https://bsky.app/profile/did:plc:hdhoaan3xa3jiuq4fg4mefid/post/3mejuq44twc2t

key thing is that where the focus of getManyToManyCounts was the other subject (aggregation was against that, so grouping happened with it),

i think the focus of disagreggated many-to-many is on the linking records themselves

to me that takes me toward a few things

  • i don't think we should need to group the links by target (does the current code build up the full aggregation on every requested page? we should be able to avoid doing that)

  • i think the order of the response should actually be based on the linking record itself (since we have a row in the output), not the other subject, unlike with the aggregated/count version. this means you get eg. list items in order they were added instead of the order of the listed things being created. (i haven't fully wrapped my head around the grouping/ordering code here yet)

  • since any linking record can have a path_to_other with multiple links, i think a composite cursor could work here:

a 2-tuple of (backlink_vec_idx, forward_vec_idx).

for normal cases where the many-to-many record points to exactly one other subject, it would just be advancing backlink_vec_idx like normal backlinks

for cases where the many-to-many record actually has multiple foward links at the given path_to_other, the second part of the tuple would track progress through that list

i think that allows us to hold the necessary state between calls without needing to reconstruct too much in memory each time?

(also it's hard to write in this tiny tiny textbox and have a sense of whether what i'm saying makes sense)

seoul.systems

Interesting approach! I have to think through this for a bit to be honest. Maybe I tried to follow the existing counts implementation too closely

Having said that, I added a new composite cursor to fix a couple of bugs that would arrive when hitting a couple of possible edge-cases in the pagination logic. This affects both the new get-many-to-many endpoint as well as the existing get-many-to-many-counts endpoint. As the changes are split over two distinct commits things should be straightforward to review.

Your assumption is still correct in the sense that we do indeed have to build up the aggregation again for every request. I have to double-check the get-backlinks endpoint to get a better sense of where you're going at.

Finally, I agree that the interface here doesn't necessarily make the whole thing easier to understand, unfortunately

seoul.systems submitted #3
diff interdiff
6 commits
expand
dbbef261
wip: m2m
52039af4
Add tests for new get_many_to_many query handler
da6fb97c
Fix get_m2m_empty test
9bd3a217
Replace tuple with RecordsBySubject struct
6f1351b9
Fix conflicts after rebasing on main
235fb506
Use record_id/subject tuple as return type for get_many_to_many
expand 2 comments
bad-example.com

i think something got funky with a rebase or the way tangled is showing it -- some of my changes on main seem to be getting shown (reverted) in the diff.

i don't mind sorting it locally but will mostly get to it tomorrow, in case you want to see what's up before i do.

seoul.systems

That's one on me, sorry! Rebased again on main and now everything seems fine

seoul.systems submitted #2
diff interdiff
5 commits
expand
dbbef261
wip: m2m
52039af4
Add tests for new get_many_to_many query handler
da6fb97c
Fix get_m2m_empty test
9bd3a217
Replace tuple with RecordsBySubject struct
6f1351b9
Fix conflicts after rebasing on main
expand 5 comments
seoul.systems

Rebased on main. As we discussed in the PR for the order query parameter, I didn't include this here as it's not a particular sensible fit.

bad-example.com

i need to get into the code properly but my initial thought is that this endpoint should return a flat list of results, like

{
  "items": [
    {
      "link": { did, collection, rkey }, // the m2m link record
      "subject": "a.com"
    },
    {
      "link": { did, collection, rkey },
      "subject": "a.com"
    },
    {
      "link": { did, collection, rkey },
      "subject": "b.com"
    },
  ]
}

this will require a bit of tricks in the cursor to track pages across half-finished groups of links

(also this isn't an immediate change request, just getting it down for discussion!)

(and separately, i've also been wondering about moving more toward returning at-uris instead of broken-out did/collection/rkey objects. which isn't specifically about this PR, but if that happens then switching before releasing it is nice)

seoul.systems

Hmm, I wonder how this would then work with the path_to_other parameter. Currently we have this nested grouping in order to show and disambiguate different relationships between different links.

For instance take the following query and it's results:

http://localhost:6789/xrpc/blue.microcosm.links.getManyToMany?subject=at://did:plc:2w45zyhuklwihpdc7oj3mi63/app.bsky.feed.post/3mdbbkuq6t32y&source=app.bsky.feed.post:reply.root.uri&pathToOther=reply.parent.uri&limit=16

This query asks: "Show me all posts in this thread, grouped by who they're responding to."

A flat list would just give us all the posts in the thread. The nested structure answers a richer question: who's talking to whom? Some posts are direct responses to the original article. Others are replies to other commenters, forming side conversations that branch off from the main thread.

The pathToOther grouping preserves that distinction. Without it, we'd lose the information about who's talking to whom.

{
  "linking_records": [
    {
      "subject": "at://did:plc:2w45zyhuklwihpdc7oj3mi63/app.bsky.feed.post/3mdbbkuq6t32y",
      "records": [
        {
          "did": "did:plc:lznqwrsbnyf6fdxohikqj6h3",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd27pja7s2y"
        },
        {
          "did": "did:plc:uffx77au6hoauuuumkbuvqdr",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd2tt5efc2a"
        },
        {
          "did": "did:plc:y7qyxzo7dns5m54dlq3youu3",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd2wtjxgc2d"
        },
        {
          "did": "did:plc:yaakslxyqydb76ybgkhrr4jk",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd35hyads22"
        },
        {
          "did": "did:plc:fia7w2kbnrdjwp6zvxywt7qv",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd37j3ldk2m"
        },
        {
          "did": "did:plc:xtecipifublblkomwau5x2ok",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd3dbtbz22n"
        },
        {
          "did": "did:plc:hl5lhiy2qr4nf5e4eefldvme",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd42hpw7c2e"
        },
        {
          "did": "did:plc:fgquypfh32pewivn3bcmzseb",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd46jteoc2m"
        }
      ]
    },
    {
      "subject": "at://did:plc:3rhjxwwui6wwfokh4at3q2dl/app.bsky.feed.post/3mdczc7c4gk2i",
      "records": [
        {
          "did": "did:plc:3rhjxwwui6wwfokh4at3q2dl",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdczt7cwhk2i"
        }
      ]
    },
    {
      "subject": "at://did:plc:6buibzhkqr4vkqu75ezr2uv2/app.bsky.feed.post/3mdby25hbbk2v",
      "records": [
        {
          "did": "did:plc:fgeie2bmzlmx37iglj3xbzuj",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd26ulf4k2j"
        }
      ]
    },
    {
      "subject": "at://did:plc:lwgvv5oqh5stzb6dxa5d7z3n/app.bsky.feed.post/3mdcxqbkkfk2i",
      "records": [
        {
          "did": "did:plc:hl5lhiy2qr4nf5e4eefldvme",
          "collection": "app.bsky.feed.post",
          "rkey": "3mdd45u56sk2e"
        }
      ]
    }
  ],
  "cursor": null
}

Correct me if I'm somehow wrong here!

Regarding returning at-uris: I think this might be a nice idea as users might be able to split these up when they feel the need to any way and it feels conceptually more complete. But, it might be easier to do this in a different PR over all existing XRPC endpoints. This would allow us to add this new endpoint already while working on the updated return values in the meantime. I'd like to avoid doing too much distinct stuff in one PR. :)

bad-example.com

at-uris: totally fair, holding off for a follow-up.

flat list: i might have messed it up in my example but i think what i meant is actually equivalent to the grouped version: flattened, with the subject ("group by") included with every item in the flatted list.

clients can collect the flat list and group on subject to get back to your structured example, if they want.

my motivations are probably part sql-brain, part flat-list-enjoyer, and part cursor-related. i'm trying to disregard the first two, and i'm curious about your thoughts about how to handle the cursor:

with a flat list it's easy (from the client perspective at least) -- just keep chasing the cursor for as much of the data as you need. (cursors can happen in the middle of a subject)

with nested results grouped by subject it's less obvious to me. correct me if i'm wrong (need another block of time to actually get into the code) but i think the grouped item sub-list is unbounded size in the proposed code here? so cursors are only limiting the number of groups.

if we go with the grouped nested response, i think maybe we'd want something like:

  • a cursor at the end for fetching more groups, and
  • a cursor for each group-list that lets you fetch more items from just that group-list.

(i think this kind of nested paging is pretty neat!)

seoul.systems

Interesting. Now that you mention it I feel I kinda get where you're going at!

I think the whole cursor thing, albeit possible for sure, is kinda creating more unnecessary complexity so I'll probably go with your suggestion.

It seems easier to create custom groupings on their own for most users (having more freedom is always great) and I think from an ergonomic perspective the two cursors might create more friction.

seoul.systems submitted #1
interdiff
4 commits
expand
a4177589
wip: m2m
81b7df55
Add tests for new get_many_to_many query handler
c6742917
Fix get_m2m_empty test
f0b16b50
Replace tuple with RecordsBySubject struct
expand 1 comment
seoul.systems

Added the missing lexicon entry for the new endpoint and changed the return type as well. Commented this wrongly at the other PR that I was working on. Sorry about that lol.

seoul.systems submitted #0
diff
3 commits
expand
a4177589
wip: m2m
81b7df55
Add tests for new get_many_to_many query handler
c6742917
Fix get_m2m_empty test
expand 1 comment
seoul.systems

I think the existing get_many_to_many_counts handler and the new get_many_to_many handler are similar enough that we might extract the bulk of their logic in a shared piece of logic. Maybe a method that takes the existing identical function parameters and a new additional callback parameter (that handles what we do with found matches, i.e. calculate counts or join URIs) might be one way to go for it.

I am not too sure yet though if this is indeed the right thing to do as the new shared implementation might be a bit complicated. But given the strong similarities between the two I think it's worth at least considering.