+7 -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, ··· 48 48 pub fn rkey(&self) -> String { 49 49 self.rkey.clone() 50 50 } 51 51 + } 52 52 + 53 53 + #[derive(Debug, PartialEq, Serialize, Deserialize, Clone, Default)] 54 54 + pub struct RecordsBySubject { 55 55 + pub subject: String, 56 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 21 - use crate::{CountsByCount, Did, RecordId}; 21 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 86 + .await 87 87 + .map_err(to500)? 88 88 + } 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)) 86 97 .await 87 98 .map_err(to500)? 88 99 } ··· 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<RecordsBySubject>, 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(),

+97 -1

constellation/src/storage/mem_store.rs

··· 1 1 use super::{ 2 2 LinkReader, LinkStorage, PagedAppendingCollection, PagedOrderedCollection, StorageStats, 3 3 }; 4 4 - use crate::{ActionableEvent, CountsByCount, Did, RecordId}; 4 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 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<RecordsBySubject, String>> { 248 248 + let empty_res = Ok(PagedOrderedCollection { 249 249 + items: Vec::new(), 250 250 + next: None, 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)| RecordsBySubject { 311 311 + subject: t.0, 312 312 + records: r, 313 313 + }) 314 314 + .collect::<Vec<_>>(); 315 315 + 316 316 + items.sort_by(|a, b| a.subject.cmp(&b.subject)); 317 317 + 318 318 + items = items 319 319 + .into_iter() 320 320 + .skip_while(|item| after.as_ref().map(|a| &item.subject <= a).unwrap_or(false)) 321 321 + .take(limit as usize) 322 322 + .collect(); 323 323 + 324 324 + let next = if items.len() as u64 >= limit { 325 325 + items.last().map(|item| item.subject.clone()) 326 326 + } else { 327 327 + None 328 328 + }; 329 329 + 330 330 + Ok(PagedOrderedCollection { items, next }) 235 331 } 236 332 237 333 fn get_links(

+241 -1

constellation/src/storage/mod.rs

··· 1 1 - use crate::{ActionableEvent, CountsByCount, Did, RecordId}; 1 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 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<RecordsBySubject, String>>; 106 118 107 119 fn get_all_counts( 108 120 &self, ··· 1551 1563 next: None, 1552 1564 } 1553 1565 ); 1566 1566 + }); 1567 1567 + 1568 1568 + test_each_storage!(get_m2m_empty, |storage| { 1569 1569 + assert_eq!( 1570 1570 + storage.get_many_to_many( 1571 1571 + "a.com", 1572 1572 + "a.b.c", 1573 1573 + ".d.e", 1574 1574 + ".f.g", 1575 1575 + 10, 1576 1576 + None, 1577 1577 + &HashSet::new(), 1578 1578 + &HashSet::new(), 1579 1579 + )?, 1580 1580 + PagedOrderedCollection { 1581 1581 + items: vec![], 1582 1582 + next: None, 1583 1583 + } 1584 1584 + ); 1585 1585 + }); 1586 1586 + 1587 1587 + test_each_storage!(get_m2m_single, |storage| { 1588 1588 + storage.push( 1589 1589 + &ActionableEvent::CreateLinks { 1590 1590 + record_id: RecordId { 1591 1591 + did: "did:plc:asdf".into(), 1592 1592 + collection: "app.t.c".into(), 1593 1593 + rkey: "asdf".into(), 1594 1594 + }, 1595 1595 + links: vec![ 1596 1596 + CollectedLink { 1597 1597 + target: Link::Uri("a.com".into()), 1598 1598 + path: ".abc.uri".into(), 1599 1599 + }, 1600 1600 + CollectedLink { 1601 1601 + target: Link::Uri("b.com".into()), 1602 1602 + path: ".def.uri".into(), 1603 1603 + }, 1604 1604 + CollectedLink { 1605 1605 + target: Link::Uri("b.com".into()), 1606 1606 + path: ".ghi.uri".into(), 1607 1607 + }, 1608 1608 + ], 1609 1609 + }, 1610 1610 + 0, 1611 1611 + )?; 1612 1612 + assert_eq!( 1613 1613 + storage.get_many_to_many( 1614 1614 + "a.com", 1615 1615 + "app.t.c", 1616 1616 + ".abc.uri", 1617 1617 + ".def.uri", 1618 1618 + 10, 1619 1619 + None, 1620 1620 + &HashSet::new(), 1621 1621 + &HashSet::new(), 1622 1622 + )?, 1623 1623 + PagedOrderedCollection { 1624 1624 + items: vec![RecordsBySubject { 1625 1625 + subject: "b.com".to_string(), 1626 1626 + records: vec![RecordId { 1627 1627 + did: "did:plc:asdf".into(), 1628 1628 + collection: "app.t.c".into(), 1629 1629 + rkey: "asdf".into(), 1630 1630 + }] 1631 1631 + }], 1632 1632 + next: None, 1633 1633 + } 1634 1634 + ); 1635 1635 + }); 1636 1636 + 1637 1637 + test_each_storage!(get_m2m_filters, |storage| { 1638 1638 + storage.push( 1639 1639 + &ActionableEvent::CreateLinks { 1640 1640 + record_id: RecordId { 1641 1641 + did: "did:plc:asdf".into(), 1642 1642 + collection: "app.t.c".into(), 1643 1643 + rkey: "asdf".into(), 1644 1644 + }, 1645 1645 + links: vec![ 1646 1646 + CollectedLink { 1647 1647 + target: Link::Uri("a.com".into()), 1648 1648 + path: ".abc.uri".into(), 1649 1649 + }, 1650 1650 + CollectedLink { 1651 1651 + target: Link::Uri("b.com".into()), 1652 1652 + path: ".def.uri".into(), 1653 1653 + }, 1654 1654 + ], 1655 1655 + }, 1656 1656 + 0, 1657 1657 + )?; 1658 1658 + storage.push( 1659 1659 + &ActionableEvent::CreateLinks { 1660 1660 + record_id: RecordId { 1661 1661 + did: "did:plc:asdf".into(), 1662 1662 + collection: "app.t.c".into(), 1663 1663 + rkey: "asdf2".into(), 1664 1664 + }, 1665 1665 + links: vec![ 1666 1666 + CollectedLink { 1667 1667 + target: Link::Uri("a.com".into()), 1668 1668 + path: ".abc.uri".into(), 1669 1669 + }, 1670 1670 + CollectedLink { 1671 1671 + target: Link::Uri("b.com".into()), 1672 1672 + path: ".def.uri".into(), 1673 1673 + }, 1674 1674 + ], 1675 1675 + }, 1676 1676 + 1, 1677 1677 + )?; 1678 1678 + storage.push( 1679 1679 + &ActionableEvent::CreateLinks { 1680 1680 + record_id: RecordId { 1681 1681 + did: "did:plc:fdsa".into(), 1682 1682 + collection: "app.t.c".into(), 1683 1683 + rkey: "fdsa".into(), 1684 1684 + }, 1685 1685 + links: vec![ 1686 1686 + CollectedLink { 1687 1687 + target: Link::Uri("a.com".into()), 1688 1688 + path: ".abc.uri".into(), 1689 1689 + }, 1690 1690 + CollectedLink { 1691 1691 + target: Link::Uri("c.com".into()), 1692 1692 + path: ".def.uri".into(), 1693 1693 + }, 1694 1694 + ], 1695 1695 + }, 1696 1696 + 2, 1697 1697 + )?; 1698 1698 + storage.push( 1699 1699 + &ActionableEvent::CreateLinks { 1700 1700 + record_id: RecordId { 1701 1701 + did: "did:plc:fdsa".into(), 1702 1702 + collection: "app.t.c".into(), 1703 1703 + rkey: "fdsa2".into(), 1704 1704 + }, 1705 1705 + links: vec![ 1706 1706 + CollectedLink { 1707 1707 + target: Link::Uri("a.com".into()), 1708 1708 + path: ".abc.uri".into(), 1709 1709 + }, 1710 1710 + CollectedLink { 1711 1711 + target: Link::Uri("c.com".into()), 1712 1712 + path: ".def.uri".into(), 1713 1713 + }, 1714 1714 + ], 1715 1715 + }, 1716 1716 + 3, 1717 1717 + )?; 1718 1718 + 1719 1719 + // Test without filters - should get all records grouped by secondary target 1720 1720 + let result = storage.get_many_to_many( 1721 1721 + "a.com", 1722 1722 + "app.t.c", 1723 1723 + ".abc.uri", 1724 1724 + ".def.uri", 1725 1725 + 10, 1726 1726 + None, 1727 1727 + &HashSet::new(), 1728 1728 + &HashSet::new(), 1729 1729 + )?; 1730 1730 + assert_eq!(result.items.len(), 2); 1731 1731 + assert_eq!(result.next, None); 1732 1732 + // Find b.com group 1733 1733 + let b_group = result 1734 1734 + .items 1735 1735 + .iter() 1736 1736 + .find(|group| group.subject == "b.com") 1737 1737 + .unwrap(); 1738 1738 + assert_eq!(b_group.subject, "b.com"); 1739 1739 + assert_eq!(b_group.records.len(), 2); 1740 1740 + assert!(b_group.records 1741 1741 + .iter() 1742 1742 + .any(|r| r.did.0 == "did:plc:asdf" && r.rkey == "asdf")); 1743 1743 + assert!(b_group.records 1744 1744 + .iter() 1745 1745 + .any(|r| r.did.0 == "did:plc:asdf" && r.rkey == "asdf2")); 1746 1746 + // Find c.com group 1747 1747 + let c_group = result 1748 1748 + .items 1749 1749 + .iter() 1750 1750 + .find(|group| group.subject == "c.com") 1751 1751 + .unwrap(); 1752 1752 + assert_eq!(c_group.subject, "c.com"); 1753 1753 + assert_eq!(c_group.records.len(), 2); 1754 1754 + assert!(c_group.records 1755 1755 + .iter() 1756 1756 + .any(|r| r.did.0 == "did:plc:fdsa" && r.rkey == "fdsa")); 1757 1757 + assert!(c_group.records 1758 1758 + .iter() 1759 1759 + .any(|r| r.did.0 == "did:plc:fdsa" && r.rkey == "fdsa2")); 1760 1760 + 1761 1761 + // Test with DID filter - should only get records from did:plc:fdsa 1762 1762 + let result = storage.get_many_to_many( 1763 1763 + "a.com", 1764 1764 + "app.t.c", 1765 1765 + ".abc.uri", 1766 1766 + ".def.uri", 1767 1767 + 10, 1768 1768 + None, 1769 1769 + &HashSet::from_iter([Did("did:plc:fdsa".to_string())]), 1770 1770 + &HashSet::new(), 1771 1771 + )?; 1772 1772 + assert_eq!(result.items.len(), 1); 1773 1773 + let group = &result.items[0]; 1774 1774 + assert_eq!(group.subject, "c.com"); 1775 1775 + assert_eq!(group.records.len(), 2); 1776 1776 + assert!(group.records.iter().all(|r| r.did.0 == "did:plc:fdsa")); 1777 1777 + 1778 1778 + // Test with target filter - should only get records linking to b.com 1779 1779 + let result = storage.get_many_to_many( 1780 1780 + "a.com", 1781 1781 + "app.t.c", 1782 1782 + ".abc.uri", 1783 1783 + ".def.uri", 1784 1784 + 10, 1785 1785 + None, 1786 1786 + &HashSet::new(), 1787 1787 + &HashSet::from_iter(["b.com".to_string()]), 1788 1788 + )?; 1789 1789 + assert_eq!(result.items.len(), 1); 1790 1790 + let group = &result.items[0]; 1791 1791 + assert_eq!(group.subject, "b.com"); 1792 1792 + assert_eq!(group.records.len(), 2); 1793 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 5 - use crate::{CountsByCount, Did, RecordId}; 5 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 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<RecordsBySubject, 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<RecordsBySubject> = 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(RecordsBySubject { 1257 1257 + subject: target_string, 1258 1258 + records: records.clone(), 1259 1259 + }); 1260 1260 + } 1261 1261 + 1262 1262 + let next = if grouped_links.len() as u64 >= limit { 1263 1263 + grouped_links.keys().next_back().map(|k| format!("{}", k.0)) 1264 1264 + } else { 1265 1265 + None 1266 1266 + }; 1267 1267 + 1268 1268 + Ok(PagedOrderedCollection { items, next }) 1123 1269 } 1124 1270 1125 1271 fn get_links(

+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 group in linking_records %} 27 27 + <h4>Target: <code>{{ group.subject }}</code> <small>(<a href="/links/all?target={{ group.subject|urlencode }}">view all links</a>)</small></h4> 28 28 + {% for record in group.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

+109

lexicons/blue.microcosm/links/getManyToMany.json

··· 1 1 + { 2 2 + "lexicon": 1, 3 3 + "id": "blue.microcosm.links.getManyToMany", 4 4 + "defs": { 5 5 + "main": { 6 6 + "type": "query", 7 7 + "description": "Get records that link to a primary subject, grouped by the secondary subjects they also reference", 8 8 + "parameters": { 9 9 + "type": "params", 10 10 + "required": ["subject", "source", "pathToOther"], 11 11 + "properties": { 12 12 + "subject": { 13 13 + "type": "string", 14 14 + "format": "uri", 15 15 + "description": "the primary target being linked to (at-uri, did, or uri)" 16 16 + }, 17 17 + "source": { 18 18 + "type": "string", 19 19 + "description": "collection and path specification for the primary link (e.g., 'app.bsky.feed.like:subject.uri')" 20 20 + }, 21 21 + "pathToOther": { 22 22 + "type": "string", 23 23 + "description": "path to the secondary link in the many-to-many record (e.g., 'otherThing.uri')" 24 24 + }, 25 25 + "did": { 26 26 + "type": "array", 27 27 + "description": "filter links to those from specific users", 28 28 + "items": { 29 29 + "type": "string", 30 30 + "format": "did" 31 31 + } 32 32 + }, 33 33 + "otherSubject": { 34 34 + "type": "array", 35 35 + "description": "filter secondary links to specific subjects", 36 36 + "items": { 37 37 + "type": "string" 38 38 + } 39 39 + }, 40 40 + "limit": { 41 41 + "type": "integer", 42 42 + "minimum": 1, 43 43 + "maximum": 100, 44 44 + "default": 16, 45 45 + "description": "number of results to return" 46 46 + } 47 47 + } 48 48 + }, 49 49 + "output": { 50 50 + "encoding": "application/json", 51 51 + "schema": { 52 52 + "type": "object", 53 53 + "required": ["linking_records"], 54 54 + "properties": { 55 55 + "linking_records": { 56 56 + "type": "array", 57 57 + "items": { 58 58 + "type": "ref", 59 59 + "ref": "#recordsBySubject" 60 60 + } 61 61 + }, 62 62 + "cursor": { 63 63 + "type": "string", 64 64 + "description": "pagination cursor" 65 65 + } 66 66 + } 67 67 + } 68 68 + } 69 69 + }, 70 70 + "recordsBySubject": { 71 71 + "type": "object", 72 72 + "required": ["subject", "records"], 73 73 + "properties": { 74 74 + "subject": { 75 75 + "type": "string", 76 76 + "description": "the secondary subject that these records link to" 77 77 + }, 78 78 + "records": { 79 79 + "type": "array", 80 80 + "items": { 81 81 + "type": "ref", 82 82 + "ref": "#linkRecord" 83 83 + } 84 84 + } 85 85 + } 86 86 + }, 87 87 + "linkRecord": { 88 88 + "type": "object", 89 89 + "required": ["did", "collection", "rkey"], 90 90 + "description": "A record identifier consisting of a DID, collection, and record key", 91 91 + "properties": { 92 92 + "did": { 93 93 + "type": "string", 94 94 + "format": "did", 95 95 + "description": "the DID of the linking record's repository" 96 96 + }, 97 97 + "collection": { 98 98 + "type": "string", 99 99 + "format": "nsid", 100 100 + "description": "the collection of the linking record" 101 101 + }, 102 102 + "rkey": { 103 103 + "type": "string", 104 104 + "format": "record-key" 105 105 + } 106 106 + } 107 107 + } 108 108 + } 109 109 + }