bobbin: filter countIssues and countPulls by state · alice.pds.demo.boltless.dev/core@3df8829

+495 -5

bobbin/crates/edge-index/src/lib.rs

··· 3 3 use std::num::NonZeroU32; 4 4 use std::ops::{Bound, ControlFlow}; 5 5 use std::sync::atomic::{AtomicU32, Ordering}; 6 - use std::sync::{Arc, Mutex}; 6 + use std::sync::{Arc, Mutex, RwLock}; 7 7 8 8 const FILTER_SCAN_MULTIPLIER: usize = 64; 9 9 const FILTER_SCAN_FLOOR: usize = 512; ··· 25 25 } 26 26 27 27 use bobbin_runtime::RuntimeHasher; 28 - use bobbin_types::edges::Edge; 28 + use bobbin_types::edges::{Edge, Record}; 29 29 use bobbin_types::ids::EdgeKey; 30 30 use either::Either; 31 31 use jacquard_common::DefaultStr; 32 + use jacquard_common::types::did::Did; 33 + use jacquard_common::types::nsid::Nsid; 32 34 use jacquard_common::types::string::AtUri; 33 35 use lasso::{Key, Spur, ThreadedRodeo}; 34 36 use scc::HashMap as SccMap; ··· 243 245 } 244 246 } 245 247 248 + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] 249 + pub struct FilteredCount { 250 + pub count: Count, 251 + pub distinct_authors: DistinctAuthorCount, 252 + } 253 + 254 + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] 255 + pub struct Count(u64); 256 + 257 + impl Count { 258 + pub const fn new(value: u64) -> Self { 259 + Self(value) 260 + } 261 + 262 + pub const fn get(self) -> u64 { 263 + self.0 264 + } 265 + } 266 + 267 + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] 268 + pub struct DistinctAuthorCount(u64); 269 + 270 + impl DistinctAuthorCount { 271 + pub const fn new(value: u64) -> Self { 272 + Self(value) 273 + } 274 + 275 + pub const fn get(self) -> u64 { 276 + self.0 277 + } 278 + } 279 + 280 + impl FilteredCount { 281 + pub const fn new(count: u64, distinct_authors: u64) -> Self { 282 + Self { 283 + count: Count::new(count), 284 + distinct_authors: DistinctAuthorCount::new(distinct_authors), 285 + } 286 + } 287 + } 288 + 246 289 #[derive(Clone, Copy, Debug, Default)] 247 290 pub struct EdgeMemReport { 248 291 pub key_count: u64, ··· 419 462 sort_micros: u64, 420 463 } 421 464 465 + #[derive(Clone, Copy)] 466 + struct ProjectedState<K> { 467 + key: EdgeKeyId, 468 + kind: K, 469 + author: Option<AuthorId>, 470 + } 471 + 472 + struct CountBucket { 473 + count: u64, 474 + authors: HashMap<AuthorId, NonZeroU32, RuntimeHasher>, 475 + } 476 + 477 + struct StateCountInner<K> { 478 + projected: HashMap<SourceId, SmallVec<[ProjectedState<K>; 1]>, RuntimeHasher>, 479 + buckets: HashMap<(EdgeKeyId, K), CountBucket, RuntimeHasher>, 480 + } 481 + 482 + struct StateCountIndex<K> { 483 + inner: RwLock<StateCountInner<K>>, 484 + hasher: RuntimeHasher, 485 + } 486 + 487 + impl<K> StateCountIndex<K> { 488 + fn new(hasher: RuntimeHasher) -> Self { 489 + Self { 490 + inner: RwLock::new(StateCountInner { 491 + projected: HashMap::with_hasher(hasher.clone()), 492 + buckets: HashMap::with_hasher(hasher.clone()), 493 + }), 494 + hasher, 495 + } 496 + } 497 + 498 + fn heap_bytes(&self) -> u64 { 499 + let inner = self 500 + .inner 501 + .read() 502 + .expect("state-count index rwlock poisoned"); 503 + let projected = inner.projected.capacity() 504 + * (std::mem::size_of::<SourceId>() 505 + + std::mem::size_of::<SmallVec<[ProjectedState<K>; 1]>>() 506 + + 1) 507 + + inner 508 + .projected 509 + .values() 510 + .filter(|states| states.spilled()) 511 + .map(|states| states.capacity() * std::mem::size_of::<ProjectedState<K>>()) 512 + .sum::<usize>(); 513 + let buckets = inner.buckets.capacity() 514 + * (std::mem::size_of::<(EdgeKeyId, K)>() + std::mem::size_of::<CountBucket>() + 1); 515 + let author_slots = inner 516 + .buckets 517 + .values() 518 + .map(|bucket| { 519 + bucket.authors.capacity() 520 + * (std::mem::size_of::<AuthorId>() + std::mem::size_of::<NonZeroU32>() + 1) 521 + }) 522 + .sum::<usize>(); 523 + (projected + buckets + author_slots) as u64 524 + } 525 + } 526 + 422 527 pub struct EdgeStore { 423 528 source_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 424 529 did_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 425 530 collection_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 426 531 key_ids: SccMap<EdgeKey, EdgeKeyId, RuntimeHasher>, 532 + keys: SccMap<EdgeKeyId, EdgeKey, RuntimeHasher>, 427 533 next_key_id: AtomicU32, 428 534 forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>, 429 535 reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>, 536 + issue_counts: StateCountIndex<IssueStateKind>, 537 + pull_counts: StateCountIndex<PullStatusKind>, 430 538 hasher: RuntimeHasher, 431 539 writer: Mutex<()>, 432 540 } ··· 438 546 did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), 439 547 collection_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), 440 548 key_ids: SccMap::with_hasher(hasher.clone()), 549 + keys: SccMap::with_hasher(hasher.clone()), 441 550 next_key_id: AtomicU32::new(0), 442 551 forward: SccMap::with_hasher(hasher.clone()), 443 552 reverse: SccMap::with_hasher(hasher.clone()), 553 + issue_counts: StateCountIndex::new(hasher.clone()), 554 + pull_counts: StateCountIndex::new(hasher.clone()), 444 555 hasher, 445 556 writer: Mutex::new(()), 446 557 } 447 558 } 448 559 449 560 fn intern_key(&self, key: EdgeKey) -> EdgeKeyId { 450 - match self.key_ids.entry_sync(key) { 561 + match self.key_ids.entry_sync(key.clone()) { 451 562 Entry::Occupied(e) => *e.get(), 452 563 Entry::Vacant(e) => { 453 564 let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed)); 454 565 e.insert_entry(id); 566 + let _ = self.keys.insert_sync(id, key); 455 567 id 456 568 } 457 569 } ··· 650 762 }) 651 763 .unwrap_or(0) 652 764 } 765 + 766 + pub fn count_by_author(&self, key: &EdgeKey, author: &Did<DefaultStr>) -> u64 { 767 + let Some(author) = self 768 + .did_interner 769 + .get(author.as_ref()) 770 + .map(AuthorId::from_spur) 771 + else { 772 + return 0; 773 + }; 774 + self.lookup_key(key) 775 + .and_then(|id| { 776 + self.forward.read_sync(&id, |_, sources| match sources { 777 + Sources::Large(big) => big 778 + .authors 779 + .get(&author) 780 + .map_or(0, |count| count.get() as u64), 781 + Sources::Small(keys) => keys 782 + .iter() 783 + .filter(|key| self.author_of(key.source) == Some(author)) 784 + .count() as u64, 785 + }) 786 + }) 787 + .unwrap_or(0) 788 + } 789 + 790 + fn current_state_projection<K>( 791 + &self, 792 + states: &StateIndex<K>, 793 + entity: &AtUri<DefaultStr>, 794 + edge_kind: &str, 795 + ) -> (SourceId, SmallVec<[ProjectedState<K>; 1]>) 796 + where 797 + K: StateKind + Default, 798 + { 799 + let source = self.intern_source(entity); 800 + let author = self.author_of(source); 801 + let entity_author = source_authority_did(entity); 802 + let reverse = self 803 + .reverse 804 + .read_sync(&source, |_, entries| entries.clone()) 805 + .unwrap_or_default(); 806 + let projected = reverse 807 + .into_iter() 808 + .filter_map(|entry| { 809 + let key = self.keys.read_sync(&entry.key_id, |_, key| key.clone())?; 810 + if key.kind.as_ref() != edge_kind { 811 + return None; 812 + } 813 + let repo_owner = key.subject.as_did()?; 814 + let kind = states 815 + .latest_by(entity, |state_source| { 816 + let state_author = source_authority_did(state_source); 817 + state_author == entity_author || state_author == Some(repo_owner.as_ref()) 818 + }) 819 + .map(|(kind, _)| kind) 820 + .unwrap_or_default(); 821 + Some(ProjectedState { 822 + key: entry.key_id, 823 + kind, 824 + author, 825 + }) 826 + }) 827 + .collect(); 828 + (source, projected) 829 + } 830 + 831 + fn refresh_state_counts<K>( 832 + &self, 833 + index: &StateCountIndex<K>, 834 + states: &StateIndex<K>, 835 + entity: &AtUri<DefaultStr>, 836 + edge_kind: &str, 837 + ) where 838 + K: StateKind + Default, 839 + { 840 + // compute under the write lock so concurrent updates dont write a stale projection 841 + let mut inner = index 842 + .inner 843 + .write() 844 + .expect("state-count index rwlock poisoned"); 845 + let (source, current) = self.current_state_projection(states, entity, edge_kind); 846 + 847 + if let Some(previous) = inner.projected.remove(&source) { 848 + for projected in previous { 849 + let bucket_key = (projected.key, projected.kind); 850 + let remove = if let Some(bucket) = inner.buckets.get_mut(&bucket_key) { 851 + bucket.count -= 1; 852 + if let Some(author) = projected.author { 853 + drop_author(&mut bucket.authors, author); 854 + } 855 + bucket.count == 0 856 + } else { 857 + false 858 + }; 859 + if remove { 860 + inner.buckets.remove(&bucket_key); 861 + } 862 + } 863 + } 864 + 865 + for projected in current.iter().copied() { 866 + let bucket = inner 867 + .buckets 868 + .entry((projected.key, projected.kind)) 869 + .or_insert_with(|| CountBucket { 870 + count: 0, 871 + authors: HashMap::with_hasher(index.hasher.clone()), 872 + }); 873 + bucket.count += 1; 874 + if let Some(author) = projected.author { 875 + bump_author(&mut bucket.authors, author); 876 + } 877 + } 878 + if !current.is_empty() { 879 + inner.projected.insert(source, current); 880 + } 881 + } 882 + 883 + pub fn refresh_issue_counts( 884 + &self, 885 + states: &StateIndex<IssueStateKind>, 886 + entity: &AtUri<DefaultStr>, 887 + ) { 888 + self.refresh_state_counts(&self.issue_counts, states, entity, "sh.tangled.repo.issue"); 889 + } 890 + 891 + pub fn refresh_pull_counts( 892 + &self, 893 + states: &StateIndex<PullStatusKind>, 894 + entity: &AtUri<DefaultStr>, 895 + ) { 896 + self.refresh_state_counts(&self.pull_counts, states, entity, "sh.tangled.repo.pull"); 897 + } 898 + 899 + fn count_state<K>( 900 + &self, 901 + index: &StateCountIndex<K>, 902 + key: &EdgeKey, 903 + kind: K, 904 + author: Option<&Did<DefaultStr>>, 905 + ) -> FilteredCount 906 + where 907 + K: StateKind, 908 + { 909 + let Some(key) = self.lookup_key(key) else { 910 + return FilteredCount::default(); 911 + }; 912 + let author = match author { 913 + Some(author) => match self.did_interner.get(author).map(AuthorId::from_spur) { 914 + Some(author) => Some(author), 915 + None => return FilteredCount::default(), 916 + }, 917 + None => None, 918 + }; 919 + let inner = index 920 + .inner 921 + .read() 922 + .expect("state-count index rwlock poisoned"); 923 + let Some(bucket) = inner.buckets.get(&(key, kind)) else { 924 + return FilteredCount::default(); 925 + }; 926 + match author { 927 + Some(author) => { 928 + let count = bucket 929 + .authors 930 + .get(&author) 931 + .map_or(0, |count| count.get() as u64); 932 + FilteredCount::new(count, u64::from(count != 0)) 933 + } 934 + None => FilteredCount::new(bucket.count, bucket.authors.len() as u64), 935 + } 936 + } 937 + 938 + pub fn count_issue_state( 939 + &self, 940 + key: &EdgeKey, 941 + kind: IssueStateKind, 942 + author: Option<&Did<DefaultStr>>, 943 + ) -> FilteredCount { 944 + self.count_state(&self.issue_counts, key, kind, author) 945 + } 946 + 947 + pub fn count_pull_status( 948 + &self, 949 + key: &EdgeKey, 950 + kind: PullStatusKind, 951 + author: Option<&Did<DefaultStr>>, 952 + ) -> FilteredCount { 953 + self.count_state(&self.pull_counts, key, kind, author) 954 + } 955 + 653 956 /// answers "did the viewer star/follow/etc. this subject, and with what rkey" 654 957 pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option<AtUri<DefaultStr>> { 655 958 let author_spur = self.did_interner.get(viewer)?; ··· 821 1124 822 1125 let mut edges_total = 0u64; 823 1126 let mut author_refs_total = 0u64; 824 - let mut forward_struct_bytes = 0u64; 1127 + let mut forward_struct_bytes = 1128 + self.issue_counts.heap_bytes() + self.pull_counts.heap_bytes(); 825 1129 let mut max_bucket = 0u64; 826 1130 let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT]; 827 1131 self.forward.iter_sync(|_, sources| { ··· 836 1140 true 837 1141 }); 838 1142 let key_interner_bytes = self.key_ids.len() as u64 839 - * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT); 1143 + * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT) 1144 + + self.keys.len() as u64 1145 + * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT); 840 1146 841 1147 let mut reverse_entries = 0u64; 842 1148 let mut reverse_cap = 0u64; ··· 870 1176 } 871 1177 } 872 1178 1179 + pub fn upsert_record_indexes( 1180 + edges: &EdgeStore, 1181 + issue_states: &StateIndex<IssueStateKind>, 1182 + pull_statuses: &StateIndex<PullStatusKind>, 1183 + source: &AtUri<DefaultStr>, 1184 + record_edges: Vec<Edge>, 1185 + record: &Record, 1186 + ) -> ApplyOutcome { 1187 + let previous_state_entity = match record { 1188 + Record::IssueState(_) => issue_states.entity_for_source(source), 1189 + Record::PullStatus(_) => pull_statuses.entity_for_source(source), 1190 + _ => None, 1191 + }; 1192 + edges.upsert_source(source, record_edges); 1193 + let outcome = apply_record_state(issue_states, pull_statuses, source, record); 1194 + 1195 + match record { 1196 + Record::Issue(_) => edges.refresh_issue_counts(issue_states, source), 1197 + Record::Pull(_) => edges.refresh_pull_counts(pull_statuses, source), 1198 + Record::IssueState(state) => { 1199 + if let Some(previous) = previous_state_entity.as_ref() 1200 + && previous != &state.issue 1201 + { 1202 + edges.refresh_issue_counts(issue_states, previous); 1203 + } 1204 + edges.refresh_issue_counts(issue_states, &state.issue); 1205 + } 1206 + Record::PullStatus(status) => { 1207 + if let Some(previous) = previous_state_entity.as_ref() 1208 + && previous != &status.pull 1209 + { 1210 + edges.refresh_pull_counts(pull_statuses, previous); 1211 + } 1212 + edges.refresh_pull_counts(pull_statuses, &status.pull); 1213 + } 1214 + _ => {} 1215 + } 1216 + outcome 1217 + } 1218 + 1219 + pub fn delete_record_indexes( 1220 + edges: &EdgeStore, 1221 + issue_states: &StateIndex<IssueStateKind>, 1222 + pull_statuses: &StateIndex<PullStatusKind>, 1223 + source: &AtUri<DefaultStr>, 1224 + nsid: &Nsid<DefaultStr>, 1225 + ) { 1226 + edges.remove_source(source); 1227 + match nsid.as_ref() { 1228 + "sh.tangled.repo.issue" => { 1229 + issue_states.remove_entity(source); 1230 + edges.refresh_issue_counts(issue_states, source); 1231 + } 1232 + "sh.tangled.repo.pull" => { 1233 + pull_statuses.remove_entity(source); 1234 + edges.refresh_pull_counts(pull_statuses, source); 1235 + } 1236 + "sh.tangled.repo.issue.state" => { 1237 + if let Some(entity) = issue_states.remove_source(source) { 1238 + edges.refresh_issue_counts(issue_states, &entity); 1239 + } 1240 + } 1241 + "sh.tangled.repo.pull.status" => { 1242 + if let Some(entity) = pull_statuses.remove_source(source) { 1243 + edges.refresh_pull_counts(pull_statuses, &entity); 1244 + } 1245 + } 1246 + _ => {} 1247 + } 1248 + } 1249 + 873 1250 fn directed_slice( 874 1251 sources: &[BucketKey], 875 1252 cursor: PageCursor, ··· 1182 1559 ); 1183 1560 assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0); 1184 1561 assert_eq!(store.count(&EdgeKey::new(kind, new_subject)), 1); 1562 + } 1563 + 1564 + #[test] 1565 + fn state_counts_reproject_after_state_first_ingest_and_rekey() { 1566 + let store = store(); 1567 + let states = StateIndex::new(RuntimeHasher::default()); 1568 + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1569 + let old_repo = did("did:plc:limpet"); 1570 + let new_repo = did("did:plc:scallop"); 1571 + let kind = nsid("sh.tangled.repo.issue"); 1572 + let old_key = EdgeKey::new(kind.clone(), SubjectRef::Did(old_repo.clone())); 1573 + let new_key = EdgeKey::new(kind.clone(), SubjectRef::Did(new_repo.clone())); 1574 + 1575 + states.upsert( 1576 + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), 1577 + issue.clone(), 1578 + 100, 1579 + IssueStateKind::Closed, 1580 + ); 1581 + store.upsert_source( 1582 + &issue, 1583 + vec![Edge { 1584 + kind: kind.clone(), 1585 + subject: SubjectRef::Did(old_repo), 1586 + source: issue.clone(), 1587 + sort_micros: 1, 1588 + }], 1589 + ); 1590 + store.refresh_issue_counts(&states, &issue); 1591 + assert_eq!( 1592 + store.count_issue_state(&old_key, IssueStateKind::Closed, None), 1593 + FilteredCount::new(1, 1) 1594 + ); 1595 + 1596 + store.upsert_source( 1597 + &issue, 1598 + vec![Edge { 1599 + kind, 1600 + subject: SubjectRef::Did(new_repo), 1601 + source: issue.clone(), 1602 + sort_micros: 2, 1603 + }], 1604 + ); 1605 + store.refresh_issue_counts(&states, &issue); 1606 + assert_eq!( 1607 + store.count_issue_state(&old_key, IssueStateKind::Closed, None), 1608 + FilteredCount::default() 1609 + ); 1610 + assert_eq!( 1611 + store.count_issue_state(&new_key, IssueStateKind::Open, Some(&did("did:plc:nel")),), 1612 + FilteredCount::new(1, 1), 1613 + "the old repo owner's state stops being accepted after the rekey", 1614 + ); 1615 + } 1616 + 1617 + #[test] 1618 + fn record_index_mutations_move_materialized_state_counts() { 1619 + use bobbin_types::sh_tangled::repo::issue::state::{State as IssueStateRecord, StateState}; 1620 + 1621 + let store = store(); 1622 + let issue_states = StateIndex::new(RuntimeHasher::default()); 1623 + let pull_statuses = StateIndex::new(RuntimeHasher::default()); 1624 + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1625 + let repo = did("did:plc:limpet"); 1626 + let key = EdgeKey::new(nsid("sh.tangled.repo.issue"), SubjectRef::Did(repo.clone())); 1627 + store.upsert_source( 1628 + &issue, 1629 + vec![Edge { 1630 + kind: nsid("sh.tangled.repo.issue"), 1631 + subject: SubjectRef::Did(repo), 1632 + source: issue.clone(), 1633 + sort_micros: 1, 1634 + }], 1635 + ); 1636 + store.refresh_issue_counts(&issue_states, &issue); 1637 + 1638 + let state_source = at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"); 1639 + let state_record = Record::IssueState(IssueStateRecord { 1640 + issue: issue.clone(), 1641 + created_at: jacquard_common::types::string::Datetime::raw_str("2026-05-01T00:00:00Z"), 1642 + state: StateState::ShTangledRepoIssueStateClosed, 1643 + extra_data: None, 1644 + }); 1645 + upsert_record_indexes( 1646 + &store, 1647 + &issue_states, 1648 + &pull_statuses, 1649 + &state_source, 1650 + Vec::new(), 1651 + &state_record, 1652 + ); 1653 + assert_eq!( 1654 + store 1655 + .count_issue_state(&key, IssueStateKind::Closed, None) 1656 + .count 1657 + .get(), 1658 + 1, 1659 + ); 1660 + 1661 + delete_record_indexes( 1662 + &store, 1663 + &issue_states, 1664 + &pull_statuses, 1665 + &state_source, 1666 + &nsid("sh.tangled.repo.issue.state"), 1667 + ); 1668 + assert_eq!( 1669 + store 1670 + .count_issue_state(&key, IssueStateKind::Open, None) 1671 + .count 1672 + .get(), 1673 + 1, 1674 + ); 1185 1675 } 1186 1676 1187 1677 #[test]

+9 -3

bobbin/crates/edge-index/src/state_index.rs

View file

··· 10 10 use scc::HashMap as SccMap; 11 11 use scc::hash_map::Entry; 12 12 13 - pub trait StateKind: Copy + Eq + std::fmt::Debug + Send + Sync + 'static { 13 + pub trait StateKind: Copy + Eq + std::hash::Hash + std::fmt::Debug + Send + Sync + 'static { 14 14 fn wire(self) -> &'static str; 15 15 } 16 16 ··· 134 134 } 135 135 } 136 136 137 - pub fn remove_source(&self, source: &AtUri<DefaultStr>) { 137 + pub fn remove_source(&self, source: &AtUri<DefaultStr>) -> Option<AtUri<DefaultStr>> { 138 138 let _w = self 139 139 .writer 140 140 .lock() 141 141 .expect("state-index writer mutex poisoned"); 142 142 let Entry::Occupied(occ) = self.reverse.entry_sync(source.clone()) else { 143 - return; 143 + return None; 144 144 }; 145 145 let entity = occ.get().entity.clone(); 146 146 let sort_micros = occ.get().sort_micros; 147 147 let _ = occ.remove(); 148 148 self.remove_forward(&entity, sort_micros, source); 149 + Some(entity) 149 150 } 150 151 151 152 pub fn remove_entity(&self, entity: &AtUri<DefaultStr>) { ··· 219 220 220 221 pub fn source_count(&self) -> usize { 221 222 self.reverse.len() 223 + } 224 + 225 + pub fn entity_for_source(&self, source: &AtUri<DefaultStr>) -> Option<AtUri<DefaultStr>> { 226 + self.reverse 227 + .read_sync(source, |_, entry| entry.entity.clone()) 222 228 } 223 229 } 224 230

+20 -22

bobbin/crates/ingest/src/lib.rs

View file

··· 5 5 6 6 use bobbin_edge_index::{ 7 7 ApplyOutcome, Coverage, CoverageWatch, EdgeStore, HydrantCursor, IssueStateKind, 8 - PromotionSignal, PullStatusKind, StateIndex, apply_record_state, 8 + PromotionSignal, PullStatusKind, StateIndex, delete_record_indexes, upsert_record_indexes, 9 9 }; 10 10 use bobbin_knot_ingest::{CapabilityGate, KnotRegistry}; 11 11 use bobbin_record_lru::RecordStore; ··· 1358 1358 } = upsert; 1359 1359 let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; 1360 1360 cache_body(ctx.records, &source, cid, bytes); 1361 - ctx.store.upsert_source(&source, edges); 1362 - let outcome = apply_record_state(ctx.issue_states, ctx.pull_statuses, &source, &parsed); 1361 + let outcome = upsert_record_indexes( 1362 + ctx.store, 1363 + ctx.issue_states, 1364 + ctx.pull_statuses, 1365 + &source, 1366 + edges, 1367 + &parsed, 1368 + ); 1363 1369 log_unknown_state_variant(outcome, &source); 1364 1370 index_search(ctx.search, ctx.resolver, &source, parsed).await; 1365 1371 } ··· 1394 1400 edges, 1395 1401 } => { 1396 1402 cache_body(records, &source, cid, bytes); 1397 - store.upsert_source(&source, edges); 1398 - let outcome = apply_record_state(issue_states, pull_statuses, &source, &parsed); 1403 + let outcome = 1404 + upsert_record_indexes(store, issue_states, pull_statuses, &source, edges, &parsed); 1399 1405 log_unknown_state_variant(outcome, &source); 1400 1406 index_search(search, resolver, &source, *parsed).await; 1401 1407 } 1402 1408 PendingOp::Delete { source, nsid } => { 1403 - store.remove_source(&source); 1404 - apply_delete_to_state_index(issue_states, pull_statuses, &source, &nsid); 1409 + delete_record_indexes(store, issue_states, pull_statuses, &source, &nsid); 1405 1410 records.remove(&source); 1406 1411 search.remove(&source).await; 1407 1412 } ··· 1477 1482 } 1478 1483 } 1479 1484 1480 - fn apply_delete_to_state_index( 1481 - issue_states: &StateIndex<IssueStateKind>, 1482 - pull_statuses: &StateIndex<PullStatusKind>, 1483 - source: &AtUri<DefaultStr>, 1484 - nsid: &Nsid<DefaultStr>, 1485 - ) { 1486 - match nsid.as_ref() { 1487 - "sh.tangled.repo.issue" => issue_states.remove_entity(source), 1488 - "sh.tangled.repo.pull" => pull_statuses.remove_entity(source), 1489 - "sh.tangled.repo.issue.state" => issue_states.remove_source(source), 1490 - "sh.tangled.repo.pull.status" => pull_statuses.remove_source(source), 1491 - _ => {} 1492 - } 1493 - } 1494 - 1495 1485 fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { 1496 1486 PromotionSignal { 1497 1487 rev_micros: record.map(|r| r.rev.timestamp()), ··· 2920 2910 did_subj("did:plc:abalone"), 2921 2911 ); 2922 2912 assert_eq!(store.count(&key), 1); 2913 + assert_eq!( 2914 + store 2915 + .count_issue_state(&key, IssueStateKind::Open, None) 2916 + .count 2917 + .get(), 2918 + 1, 2919 + "hydrant ingest materializes the default state count", 2920 + ); 2923 2921 } 2924 2922 2925 2923 #[derive(Default)]

+47 -4

bobbin/crates/xrpc/src/filter.rs

View file

··· 1 1 use std::sync::Arc; 2 2 3 - use bobbin_edge_index::{IssueStateKind, PullStatusKind, StateIndex}; 4 - use bobbin_types::ids::SubjectRef; 3 + use bobbin_edge_index::{FilteredCount, IssueStateKind, PullStatusKind, StateIndex}; 4 + use bobbin_types::ids::{EdgeKey, SubjectRef}; 5 5 use jacquard_common::DefaultStr; 6 6 use jacquard_common::types::did::Did; 7 7 use jacquard_common::types::string::AtUri; ··· 17 17 fn is_identity(&self) -> bool; 18 18 } 19 19 20 + pub trait CountFilter: ListFilter { 21 + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount; 22 + } 23 + 20 24 #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)] 21 25 pub struct NoFilter; 22 26 ··· 30 34 } 31 35 } 32 36 33 - #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)] 37 + #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq)] 34 38 #[serde(rename_all = "lowercase")] 35 39 pub enum IssueState { 36 40 Open, ··· 67 71 } 68 72 } 69 73 70 - #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)] 74 + impl CountFilter for IssueFilter { 75 + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount { 76 + match self.state { 77 + Some(wanted) => state 78 + .edges 79 + .count_issue_state(key, wanted.into(), self.author.as_ref()), 80 + None => count_by_author(state, key, self.author.as_ref()), 81 + } 82 + } 83 + } 84 + 85 + #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq)] 71 86 #[serde(rename_all = "lowercase")] 72 87 pub enum PullStatus { 73 88 Open, ··· 103 118 104 119 fn is_identity(&self) -> bool { 105 120 self.author.is_none() && self.status.is_none() 121 + } 122 + } 123 + 124 + impl CountFilter for PullFilter { 125 + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount { 126 + match self.status { 127 + Some(wanted) => state 128 + .edges 129 + .count_pull_status(key, wanted.into(), self.author.as_ref()), 130 + None => count_by_author(state, key, self.author.as_ref()), 131 + } 132 + } 133 + } 134 + 135 + fn count_by_author( 136 + state: &AppState, 137 + key: &EdgeKey, 138 + author: Option<&Did<DefaultStr>>, 139 + ) -> FilteredCount { 140 + match author { 141 + Some(author) => { 142 + let count = state.edges.count_by_author(key, author); 143 + FilteredCount::new(count, u64::from(count != 0)) 144 + } 145 + None => FilteredCount::new( 146 + state.edges.count(key), 147 + state.edges.count_distinct_authors(key), 148 + ), 106 149 } 107 150 } 108 151

+30 -7

bobbin/crates/xrpc/src/lib.rs

View file

··· 105 105 pub use backpressure::{ 106 106 HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor, 107 107 }; 108 - use filter::{IssueFilter, ListFilter, NoFilter, PullFilter}; 108 + use filter::{CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter}; 109 109 110 110 const DEFAULT_LIMIT: u32 = 50; 111 111 const FETCH_CONCURRENCY: usize = 8; 112 - 113 112 #[derive(Clone)] 114 113 pub struct AppState { 115 114 pub records: Arc<dyn RecordStore>, ··· 682 681 } 683 682 684 683 #[derive(Debug, Deserialize)] 684 + struct TypedCountQuery<F> { 685 + subject: SubjectQuery, 686 + #[serde(flatten)] 687 + filter: F, 688 + } 689 + 690 + #[derive(Debug, Deserialize)] 685 691 struct GetEdgeQuery { 686 692 actor: Did<DefaultStr>, 687 693 subject: SubjectQuery, ··· 899 905 } 900 906 } 901 907 902 - #[derive(Serialize)] 908 + #[derive(Clone, Copy, Debug, Serialize)] 903 909 #[serde(rename_all = "camelCase")] 904 910 struct CountResponse { 905 911 count: u64, ··· 1851 1857 }) 1852 1858 } 1853 1859 1860 + fn count_typed_for<R, F>( 1861 + state: &AppState, 1862 + q: TypedCountQuery<F>, 1863 + ) -> Result<CountResponse, XrpcError> 1864 + where 1865 + R: XrpcResp + HasSubject, 1866 + F: CountFilter, 1867 + { 1868 + let subject = parse_subject(&q.subject, R::SHAPE)?; 1869 + let key = EdgeKey::new(nsid_static(R::NSID), subject); 1870 + let counted = q.filter.count(state, &key); 1871 + Ok(CountResponse { 1872 + count: counted.count.get(), 1873 + distinct_authors: counted.distinct_authors.get(), 1874 + }) 1875 + } 1876 + 1854 1877 // does `actor` have an edge of this kind pointing at `subject`, returns its own uri 1855 1878 fn get_for<R: XrpcResp + HasSubject>( 1856 1879 state: &AppState, ··· 1982 2005 1983 2006 async fn count_issues( 1984 2007 State(state): State<AppState>, 1985 - XrpcQuery(q): XrpcQuery<CountQuery>, 2008 + XrpcQuery(q): XrpcQuery<TypedCountQuery<IssueFilter>>, 1986 2009 ) -> Result<Json<CountResponse>, XrpcError> { 1987 - count_for::<IssueRecord>(&state, q).map(Json) 2010 + count_typed_for::<IssueRecord, _>(&state, q).map(Json) 1988 2011 } 1989 2012 1990 2013 async fn list_pulls( ··· 2007 2030 2008 2031 async fn count_pulls( 2009 2032 State(state): State<AppState>, 2010 - XrpcQuery(q): XrpcQuery<CountQuery>, 2033 + XrpcQuery(q): XrpcQuery<TypedCountQuery<PullFilter>>, 2011 2034 ) -> Result<Json<CountResponse>, XrpcError> { 2012 - count_for::<PullRecord>(&state, q).map(Json) 2035 + count_typed_for::<PullRecord, _>(&state, q).map(Json) 2013 2036 } 2014 2037 2015 2038 async fn list_feed_comments(

+311 -8

bobbin/crates/xrpc/tests/aggregation.rs

View file

··· 113 113 source: source.clone(), 114 114 sort_micros: next_sort_micros(), 115 115 }); 116 + match kind.as_ref() { 117 + "sh.tangled.repo.issue" => self 118 + .edges 119 + .refresh_issue_counts(&self.state.issue_states, source), 120 + "sh.tangled.repo.pull" => self 121 + .edges 122 + .refresh_pull_counts(&self.state.pull_statuses, source), 123 + _ => {} 124 + } 125 + } 126 + 127 + fn upsert_issue_state( 128 + &self, 129 + source: AtUri<DefaultStr>, 130 + issue: AtUri<DefaultStr>, 131 + sort_micros: u64, 132 + kind: IssueStateKind, 133 + ) { 134 + self.state 135 + .issue_states 136 + .upsert(source, issue.clone(), sort_micros, kind); 137 + self.edges 138 + .refresh_issue_counts(&self.state.issue_states, &issue); 139 + } 140 + 141 + fn upsert_pull_status( 142 + &self, 143 + source: AtUri<DefaultStr>, 144 + pull: AtUri<DefaultStr>, 145 + sort_micros: u64, 146 + kind: PullStatusKind, 147 + ) { 148 + self.state 149 + .pull_statuses 150 + .upsert(source, pull.clone(), sort_micros, kind); 151 + self.edges 152 + .refresh_pull_counts(&self.state.pull_statuses, &pull); 116 153 } 117 154 118 155 async fn mount( ··· 1717 1754 &at("at://did:plc:teq/sh.tangled.feed.comment/c2"), 1718 1755 ); 1719 1756 1720 - h.state.issue_states.upsert( 1757 + h.upsert_issue_state( 1721 1758 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), 1722 1759 issue_uri.clone(), 1723 1760 1_777_593_600_000_000, 1724 1761 IssueStateKind::Open, 1725 1762 ); 1726 - h.state.issue_states.upsert( 1763 + h.upsert_issue_state( 1727 1764 at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"), 1728 1765 issue_uri.clone(), 1729 1766 1_777_593_700_000_000, ··· 1886 1923 &pull_uri, 1887 1924 &at("at://did:plc:teq/sh.tangled.feed.comment/c1"), 1888 1925 ); 1889 - h.state.pull_statuses.upsert( 1926 + h.upsert_pull_status( 1890 1927 at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"), 1891 1928 pull_uri.clone(), 1892 1929 1_777_593_600_000_000, 1893 1930 PullStatusKind::Open, 1894 1931 ); 1895 - h.state.pull_statuses.upsert( 1932 + h.upsert_pull_status( 1896 1933 at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"), 1897 1934 pull_uri.clone(), 1898 1935 1_777_593_800_000_000, ··· 1952 1989 } 1953 1990 1954 1991 #[tokio::test] 1992 + async fn count_issues_splits_open_from_total() { 1993 + let h = Harness::new().await; 1994 + let repo = did("did:plc:limpet"); 1995 + let subject = at(&format!("at://{}", repo.as_ref())); 1996 + for rk in ["i1", "i2", "i3"] { 1997 + h.add_edge( 1998 + &nsid("sh.tangled.repo.issue"), 1999 + &subject, 2000 + &at(&format!("at://did:plc:nel/sh.tangled.repo.issue/{rk}")), 2001 + ); 2002 + } 2003 + // the repo owner closing it counts. with no record at all it is open 2004 + h.upsert_issue_state( 2005 + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), 2006 + at("at://did:plc:nel/sh.tangled.repo.issue/i1"), 2007 + 1_777_593_600_000_000, 2008 + IssueStateKind::Closed, 2009 + ); 2010 + 2011 + let app = router(h.state.clone()); 2012 + let counts = |args: &'static [(&'static str, &'static str)]| { 2013 + let app = app.clone(); 2014 + let subject = subject.clone(); 2015 + async move { 2016 + let (status, body) = json_response( 2017 + app.oneshot(list_request( 2018 + "sh.tangled.repo.countIssues", 2019 + subject.as_ref(), 2020 + args, 2021 + )) 2022 + .await 2023 + .unwrap(), 2024 + ) 2025 + .await; 2026 + assert_eq!(status, StatusCode::OK); 2027 + body["count"].as_u64().expect("count") 2028 + } 2029 + }; 2030 + 2031 + assert_eq!(counts(&[]).await, 3, "no filter is every issue"); 2032 + assert_eq!(counts(&[("state", "open")]).await, 2); 2033 + assert_eq!(counts(&[("state", "closed")]).await, 1); 2034 + } 2035 + 2036 + #[tokio::test] 2037 + async fn count_issues_open_ignores_third_party_state_source() { 2038 + let h = Harness::new().await; 2039 + let repo = did("did:plc:limpet"); 2040 + let subject = at(&format!("at://{}", repo.as_ref())); 2041 + let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 2042 + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); 2043 + h.upsert_issue_state( 2044 + at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), 2045 + issue_uri, 2046 + 1_777_593_800_000_000, 2047 + IssueStateKind::Closed, 2048 + ); 2049 + 2050 + let app = router(h.state.clone()); 2051 + let (status, body) = json_response( 2052 + app.oneshot(list_request( 2053 + "sh.tangled.repo.countIssues", 2054 + subject.as_ref(), 2055 + &[("state", "open")], 2056 + )) 2057 + .await 2058 + .unwrap(), 2059 + ) 2060 + .await; 2061 + assert_eq!(status, StatusCode::OK); 2062 + assert_eq!( 2063 + body["count"], 2064 + json!(1), 2065 + "a stranger closing an issue must not take it out of the open count", 2066 + ); 2067 + } 2068 + 2069 + #[tokio::test] 2070 + async fn filtered_count_follows_state_changes_and_repo_rekeying() { 2071 + let h = Harness::new().await; 2072 + let old_repo = did("did:plc:limpet"); 2073 + let new_repo = did("did:plc:scallop"); 2074 + let old_subject = at(&format!("at://{}", old_repo.as_ref())); 2075 + let new_subject = at(&format!("at://{}", new_repo.as_ref())); 2076 + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 2077 + let kind = nsid("sh.tangled.repo.issue"); 2078 + h.edges.upsert_source( 2079 + &issue, 2080 + vec![Edge { 2081 + kind: kind.clone(), 2082 + subject: SubjectRef::Did(old_repo), 2083 + source: issue.clone(), 2084 + sort_micros: next_sort_micros(), 2085 + }], 2086 + ); 2087 + h.edges.refresh_issue_counts(&h.state.issue_states, &issue); 2088 + 2089 + let app = router(h.state.clone()); 2090 + let count_open = |subject: AtUri<DefaultStr>| { 2091 + let app = app.clone(); 2092 + async move { 2093 + let (status, body) = json_response( 2094 + app.oneshot(list_request( 2095 + "sh.tangled.repo.countIssues", 2096 + subject.as_ref(), 2097 + &[("state", "open")], 2098 + )) 2099 + .await 2100 + .unwrap(), 2101 + ) 2102 + .await; 2103 + assert_eq!(status, StatusCode::OK); 2104 + body["count"].as_u64().expect("count") 2105 + } 2106 + }; 2107 + 2108 + assert_eq!(count_open(old_subject.clone()).await, 1); 2109 + assert_eq!(count_open(old_subject.clone()).await, 1, "repeated read"); 2110 + assert_eq!( 2111 + count_open(new_subject.clone()).await, 2112 + 0, 2113 + "the issue does not belong to the destination yet", 2114 + ); 2115 + 2116 + h.upsert_issue_state( 2117 + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), 2118 + issue.clone(), 2119 + 1_777_593_600_000_000, 2120 + IssueStateKind::Closed, 2121 + ); 2122 + assert_eq!( 2123 + count_open(old_subject.clone()).await, 2124 + 0, 2125 + "state changes are visible without cached classification", 2126 + ); 2127 + 2128 + h.edges.upsert_source( 2129 + &issue, 2130 + vec![Edge { 2131 + kind, 2132 + subject: SubjectRef::Did(new_repo), 2133 + source: issue.clone(), 2134 + sort_micros: next_sort_micros(), 2135 + }], 2136 + ); 2137 + h.edges.refresh_issue_counts(&h.state.issue_states, &issue); 2138 + assert_eq!(count_open(old_subject).await, 0); 2139 + assert_eq!( 2140 + count_open(new_subject).await, 2141 + 1, 2142 + "re-keying reprojects acceptance against the new repo owner", 2143 + ); 2144 + } 2145 + 2146 + #[tokio::test] 2147 + async fn count_pulls_splits_by_status() { 2148 + let h = Harness::new().await; 2149 + let repo = did("did:plc:limpet"); 2150 + let subject = at(&format!("at://{}", repo.as_ref())); 2151 + for rk in ["p1", "p2", "p3"] { 2152 + h.add_edge( 2153 + &nsid("sh.tangled.repo.pull"), 2154 + &subject, 2155 + &at(&format!("at://did:plc:nel/sh.tangled.repo.pull/{rk}")), 2156 + ); 2157 + } 2158 + h.upsert_pull_status( 2159 + at("at://did:plc:limpet/sh.tangled.repo.pull.status/s1"), 2160 + at("at://did:plc:nel/sh.tangled.repo.pull/p1"), 2161 + 1_777_593_600_000_000, 2162 + PullStatusKind::Merged, 2163 + ); 2164 + h.upsert_pull_status( 2165 + at("at://did:plc:limpet/sh.tangled.repo.pull.status/s2"), 2166 + at("at://did:plc:nel/sh.tangled.repo.pull/p2"), 2167 + 1_777_593_600_000_000, 2168 + PullStatusKind::Closed, 2169 + ); 2170 + 2171 + let app = router(h.state.clone()); 2172 + let counts = |args: &'static [(&'static str, &'static str)]| { 2173 + let app = app.clone(); 2174 + let subject = subject.clone(); 2175 + async move { 2176 + let (_, body) = json_response( 2177 + app.oneshot(list_request( 2178 + "sh.tangled.repo.countPulls", 2179 + subject.as_ref(), 2180 + args, 2181 + )) 2182 + .await 2183 + .unwrap(), 2184 + ) 2185 + .await; 2186 + body["count"].as_u64().expect("count") 2187 + } 2188 + }; 2189 + 2190 + assert_eq!(counts(&[]).await, 3); 2191 + assert_eq!(counts(&[("status", "open")]).await, 1); 2192 + assert_eq!(counts(&[("status", "closed")]).await, 1); 2193 + assert_eq!(counts(&[("status", "merged")]).await, 1); 2194 + } 2195 + 2196 + #[tokio::test] 2197 + async fn count_issues_distinct_authors_follows_the_filter() { 2198 + let h = Harness::new().await; 2199 + let repo = did("did:plc:limpet"); 2200 + let subject = at(&format!("at://{}", repo.as_ref())); 2201 + h.add_edge( 2202 + &nsid("sh.tangled.repo.issue"), 2203 + &subject, 2204 + &at("at://did:plc:nel/sh.tangled.repo.issue/i1"), 2205 + ); 2206 + h.add_edge( 2207 + &nsid("sh.tangled.repo.issue"), 2208 + &subject, 2209 + &at("at://did:plc:olaren/sh.tangled.repo.issue/i2"), 2210 + ); 2211 + h.upsert_issue_state( 2212 + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), 2213 + at("at://did:plc:olaren/sh.tangled.repo.issue/i2"), 2214 + 1_777_593_600_000_000, 2215 + IssueStateKind::Closed, 2216 + ); 2217 + 2218 + let app = router(h.state.clone()); 2219 + let (_, body) = json_response( 2220 + app.oneshot(list_request( 2221 + "sh.tangled.repo.countIssues", 2222 + subject.as_ref(), 2223 + &[("state", "open")], 2224 + )) 2225 + .await 2226 + .unwrap(), 2227 + ) 2228 + .await; 2229 + assert_eq!(body["count"], json!(1)); 2230 + assert_eq!( 2231 + body["distinctAuthors"], 2232 + json!(1), 2233 + "authors of filtered-out issues must not be counted", 2234 + ); 2235 + 2236 + for (args, count, distinct) in [ 2237 + (&[("author", "did:plc:nel")][..], 1, 1), 2238 + (&[("author", "did:plc:nel"), ("state", "open")][..], 1, 1), 2239 + (&[("author", "did:plc:olaren"), ("state", "open")][..], 0, 0), 2240 + ] { 2241 + let (_, body) = json_response( 2242 + router(h.state.clone()) 2243 + .oneshot(list_request( 2244 + "sh.tangled.repo.countIssues", 2245 + subject.as_ref(), 2246 + args, 2247 + )) 2248 + .await 2249 + .unwrap(), 2250 + ) 2251 + .await; 2252 + assert_eq!(body["count"], json!(count)); 2253 + assert_eq!(body["distinctAuthors"], json!(distinct)); 2254 + } 2255 + } 2256 + 2257 + #[tokio::test] 1955 2258 async fn list_issues_state_filter_ignores_third_party_state_source() { 1956 2259 let h = Harness::new().await; 1957 2260 let repo = did("did:plc:limpet"); ··· 1965 2268 issue_body(&repo, "open issue"), 1966 2269 ) 1967 2270 .await; 1968 - h.state.issue_states.upsert( 2271 + h.upsert_issue_state( 1969 2272 at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), 1970 2273 issue_uri.clone(), 1971 2274 1_777_593_800_000_000, ··· 2011 2314 pull_body(&repo, "wip"), 2012 2315 ) 2013 2316 .await; 2014 - h.state.pull_statuses.upsert( 2317 + h.upsert_pull_status( 2015 2318 at("at://did:plc:nautilus/sh.tangled.repo.pull.status/spoof"), 2016 2319 pull_uri.clone(), 2017 2320 1_777_593_800_000_000, ··· 2052 2355 issue_body(&repo_owner, "owner closed"), 2053 2356 ) 2054 2357 .await; 2055 - h.state.issue_states.upsert( 2358 + h.upsert_issue_state( 2056 2359 at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"), 2057 2360 issue_uri.clone(), 2058 2361 1_777_593_800_000_000, ··· 2183 2486 issue_body(&repo, "shut"), 2184 2487 ) 2185 2488 .await; 2186 - h.state.issue_states.upsert( 2489 + h.upsert_issue_state( 2187 2490 at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), 2188 2491 closed_uri.clone(), 2189 2492 1_777_593_800_000_000,

+10

lexicons/repo/countIssues.json

View file

··· 12 12 "type": "string", 13 13 "format": "did", 14 14 "description": "Repo DID to list issues for" 15 + }, 16 + "author": { 17 + "type": "string", 18 + "format": "did", 19 + "description": "Restrict to issues authored by this user DID." 20 + }, 21 + "state": { 22 + "type": "string", 23 + "knownValues": ["open", "closed"], 24 + "description": "Restrict to issues whose latest derived state matches." 15 25 } 16 26 } 17 27 },

+10

lexicons/repo/countPulls.json

View file

··· 12 12 "type": "string", 13 13 "format": "did", 14 14 "description": "Repo DID to list pulls for" 15 + }, 16 + "author": { 17 + "type": "string", 18 + "format": "did", 19 + "description": "Restrict to pulls authored by this user DID." 20 + }, 21 + "status": { 22 + "type": "string", 23 + "knownValues": ["open", "closed", "merged"], 24 + "description": "Restrict to pulls whose latest derived status matches." 15 25 } 16 26 } 17 27 },

Configure Feed

Configure Feed