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

+465 -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::nsid::Nsid; 32 33 use jacquard_common::types::string::AtUri; 33 34 use lasso::{Key, Spur, ThreadedRodeo}; 34 35 use scc::HashMap as SccMap; ··· 243 244 } 244 245 } 245 246 247 + #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] 248 + pub struct FilteredCount { 249 + pub count: u64, 250 + pub distinct_authors: u64, 251 + } 252 + 246 253 #[derive(Clone, Copy, Debug, Default)] 247 254 pub struct EdgeMemReport { 248 255 pub key_count: u64, ··· 419 426 sort_micros: u64, 420 427 } 421 428 429 + #[derive(Clone, Copy)] 430 + struct ProjectedState<K> { 431 + key: EdgeKeyId, 432 + kind: K, 433 + author: Option<AuthorId>, 434 + } 435 + 436 + struct CountBucket { 437 + count: u64, 438 + authors: HashMap<AuthorId, NonZeroU32, RuntimeHasher>, 439 + } 440 + 441 + struct StateCountInner<K> { 442 + projected: HashMap<SourceId, SmallVec<[ProjectedState<K>; 1]>, RuntimeHasher>, 443 + buckets: HashMap<(EdgeKeyId, K), CountBucket, RuntimeHasher>, 444 + } 445 + 446 + struct StateCountIndex<K> { 447 + inner: RwLock<StateCountInner<K>>, 448 + hasher: RuntimeHasher, 449 + } 450 + 451 + impl<K> StateCountIndex<K> { 452 + fn new(hasher: RuntimeHasher) -> Self { 453 + Self { 454 + inner: RwLock::new(StateCountInner { 455 + projected: HashMap::with_hasher(hasher.clone()), 456 + buckets: HashMap::with_hasher(hasher.clone()), 457 + }), 458 + hasher, 459 + } 460 + } 461 + 462 + fn heap_bytes(&self) -> u64 { 463 + let inner = self 464 + .inner 465 + .read() 466 + .expect("state-count index rwlock poisoned"); 467 + let projected = inner.projected.capacity() 468 + * (std::mem::size_of::<SourceId>() 469 + + std::mem::size_of::<SmallVec<[ProjectedState<K>; 1]>>() 470 + + 1) 471 + + inner 472 + .projected 473 + .values() 474 + .filter(|states| states.spilled()) 475 + .map(|states| states.capacity() * std::mem::size_of::<ProjectedState<K>>()) 476 + .sum::<usize>(); 477 + let buckets = inner.buckets.capacity() 478 + * (std::mem::size_of::<(EdgeKeyId, K)>() + std::mem::size_of::<CountBucket>() + 1); 479 + let author_slots = inner 480 + .buckets 481 + .values() 482 + .map(|bucket| { 483 + bucket.authors.capacity() 484 + * (std::mem::size_of::<AuthorId>() + std::mem::size_of::<NonZeroU32>() + 1) 485 + }) 486 + .sum::<usize>(); 487 + (projected + buckets + author_slots) as u64 488 + } 489 + } 490 + 422 491 pub struct EdgeStore { 423 492 source_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 424 493 did_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 425 494 collection_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>, 426 495 key_ids: SccMap<EdgeKey, EdgeKeyId, RuntimeHasher>, 496 + keys: SccMap<EdgeKeyId, EdgeKey, RuntimeHasher>, 427 497 next_key_id: AtomicU32, 428 498 forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>, 429 499 reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>, 500 + issue_counts: StateCountIndex<IssueStateKind>, 501 + pull_counts: StateCountIndex<PullStatusKind>, 430 502 hasher: RuntimeHasher, 431 503 writer: Mutex<()>, 432 504 } ··· 438 510 did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), 439 511 collection_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), 440 512 key_ids: SccMap::with_hasher(hasher.clone()), 513 + keys: SccMap::with_hasher(hasher.clone()), 441 514 next_key_id: AtomicU32::new(0), 442 515 forward: SccMap::with_hasher(hasher.clone()), 443 516 reverse: SccMap::with_hasher(hasher.clone()), 517 + issue_counts: StateCountIndex::new(hasher.clone()), 518 + pull_counts: StateCountIndex::new(hasher.clone()), 444 519 hasher, 445 520 writer: Mutex::new(()), 446 521 } 447 522 } 448 523 449 524 fn intern_key(&self, key: EdgeKey) -> EdgeKeyId { 450 - match self.key_ids.entry_sync(key) { 525 + match self.key_ids.entry_sync(key.clone()) { 451 526 Entry::Occupied(e) => *e.get(), 452 527 Entry::Vacant(e) => { 453 528 let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed)); 454 529 e.insert_entry(id); 530 + let _ = self.keys.insert_sync(id, key); 455 531 id 456 532 } 457 533 } ··· 650 726 }) 651 727 .unwrap_or(0) 652 728 } 729 + 730 + pub fn count_by_author(&self, key: &EdgeKey, author: &str) -> u64 { 731 + let Some(author) = self.did_interner.get(author).map(AuthorId::from_spur) else { 732 + return 0; 733 + }; 734 + self.lookup_key(key) 735 + .and_then(|id| { 736 + self.forward.read_sync(&id, |_, sources| match sources { 737 + Sources::Large(big) => big 738 + .authors 739 + .get(&author) 740 + .map_or(0, |count| count.get() as u64), 741 + Sources::Small(keys) => keys 742 + .iter() 743 + .filter(|key| self.author_of(key.source) == Some(author)) 744 + .count() as u64, 745 + }) 746 + }) 747 + .unwrap_or(0) 748 + } 749 + 750 + fn current_state_projection<K>( 751 + &self, 752 + states: &StateIndex<K>, 753 + entity: &AtUri<DefaultStr>, 754 + edge_kind: &str, 755 + ) -> (SourceId, SmallVec<[ProjectedState<K>; 1]>) 756 + where 757 + K: StateKind + Default, 758 + { 759 + let source = self.intern_source(entity); 760 + let author = self.author_of(source); 761 + let entity_author = source_authority_did(entity); 762 + let reverse = self 763 + .reverse 764 + .read_sync(&source, |_, entries| entries.clone()) 765 + .unwrap_or_default(); 766 + let projected = reverse 767 + .into_iter() 768 + .filter_map(|entry| { 769 + let key = self.keys.read_sync(&entry.key_id, |_, key| key.clone())?; 770 + if key.kind.as_ref() != edge_kind { 771 + return None; 772 + } 773 + let repo_owner = key.subject.as_did()?; 774 + let kind = states 775 + .latest_by(entity, |state_source| { 776 + let state_author = source_authority_did(state_source); 777 + state_author == entity_author || state_author == Some(repo_owner.as_ref()) 778 + }) 779 + .map(|(kind, _)| kind) 780 + .unwrap_or_default(); 781 + Some(ProjectedState { 782 + key: entry.key_id, 783 + kind, 784 + author, 785 + }) 786 + }) 787 + .collect(); 788 + (source, projected) 789 + } 790 + 791 + fn refresh_state_counts<K>( 792 + &self, 793 + index: &StateCountIndex<K>, 794 + states: &StateIndex<K>, 795 + entity: &AtUri<DefaultStr>, 796 + edge_kind: &str, 797 + ) where 798 + K: StateKind + Default, 799 + { 800 + // compute under the write lock so concurrent updates dont write a stale projection 801 + let mut inner = index 802 + .inner 803 + .write() 804 + .expect("state-count index rwlock poisoned"); 805 + let (source, current) = self.current_state_projection(states, entity, edge_kind); 806 + 807 + if let Some(previous) = inner.projected.remove(&source) { 808 + for projected in previous { 809 + let bucket_key = (projected.key, projected.kind); 810 + let remove = if let Some(bucket) = inner.buckets.get_mut(&bucket_key) { 811 + bucket.count -= 1; 812 + if let Some(author) = projected.author { 813 + drop_author(&mut bucket.authors, author); 814 + } 815 + bucket.count == 0 816 + } else { 817 + false 818 + }; 819 + if remove { 820 + inner.buckets.remove(&bucket_key); 821 + } 822 + } 823 + } 824 + 825 + for projected in current.iter().copied() { 826 + let bucket = inner 827 + .buckets 828 + .entry((projected.key, projected.kind)) 829 + .or_insert_with(|| CountBucket { 830 + count: 0, 831 + authors: HashMap::with_hasher(index.hasher.clone()), 832 + }); 833 + bucket.count += 1; 834 + if let Some(author) = projected.author { 835 + bump_author(&mut bucket.authors, author); 836 + } 837 + } 838 + if !current.is_empty() { 839 + inner.projected.insert(source, current); 840 + } 841 + } 842 + 843 + pub fn refresh_issue_counts( 844 + &self, 845 + states: &StateIndex<IssueStateKind>, 846 + entity: &AtUri<DefaultStr>, 847 + ) { 848 + self.refresh_state_counts(&self.issue_counts, states, entity, "sh.tangled.repo.issue"); 849 + } 850 + 851 + pub fn refresh_pull_counts( 852 + &self, 853 + states: &StateIndex<PullStatusKind>, 854 + entity: &AtUri<DefaultStr>, 855 + ) { 856 + self.refresh_state_counts(&self.pull_counts, states, entity, "sh.tangled.repo.pull"); 857 + } 858 + 859 + fn count_state<K>( 860 + &self, 861 + index: &StateCountIndex<K>, 862 + key: &EdgeKey, 863 + kind: K, 864 + author: Option<&str>, 865 + ) -> FilteredCount 866 + where 867 + K: StateKind, 868 + { 869 + let Some(key) = self.lookup_key(key) else { 870 + return FilteredCount::default(); 871 + }; 872 + let author = match author { 873 + Some(author) => match self.did_interner.get(author).map(AuthorId::from_spur) { 874 + Some(author) => Some(author), 875 + None => return FilteredCount::default(), 876 + }, 877 + None => None, 878 + }; 879 + let inner = index 880 + .inner 881 + .read() 882 + .expect("state-count index rwlock poisoned"); 883 + let Some(bucket) = inner.buckets.get(&(key, kind)) else { 884 + return FilteredCount::default(); 885 + }; 886 + match author { 887 + Some(author) => { 888 + let count = bucket 889 + .authors 890 + .get(&author) 891 + .map_or(0, |count| count.get() as u64); 892 + FilteredCount { 893 + count, 894 + distinct_authors: u64::from(count != 0), 895 + } 896 + } 897 + None => FilteredCount { 898 + count: bucket.count, 899 + distinct_authors: bucket.authors.len() as u64, 900 + }, 901 + } 902 + } 903 + 904 + pub fn count_issue_state( 905 + &self, 906 + key: &EdgeKey, 907 + kind: IssueStateKind, 908 + author: Option<&str>, 909 + ) -> FilteredCount { 910 + self.count_state(&self.issue_counts, key, kind, author) 911 + } 912 + 913 + pub fn count_pull_status( 914 + &self, 915 + key: &EdgeKey, 916 + kind: PullStatusKind, 917 + author: Option<&str>, 918 + ) -> FilteredCount { 919 + self.count_state(&self.pull_counts, key, kind, author) 920 + } 921 + 653 922 /// answers "did the viewer star/follow/etc. this subject, and with what rkey" 654 923 pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option<AtUri<DefaultStr>> { 655 924 let author_spur = self.did_interner.get(viewer)?; ··· 821 1090 822 1091 let mut edges_total = 0u64; 823 1092 let mut author_refs_total = 0u64; 824 - let mut forward_struct_bytes = 0u64; 1093 + let mut forward_struct_bytes = 1094 + self.issue_counts.heap_bytes() + self.pull_counts.heap_bytes(); 825 1095 let mut max_bucket = 0u64; 826 1096 let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT]; 827 1097 self.forward.iter_sync(|_, sources| { ··· 836 1106 true 837 1107 }); 838 1108 let key_interner_bytes = self.key_ids.len() as u64 839 - * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT); 1109 + * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT) 1110 + + self.keys.len() as u64 1111 + * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT); 840 1112 841 1113 let mut reverse_entries = 0u64; 842 1114 let mut reverse_cap = 0u64; ··· 867 1139 max_bucket, 868 1140 bucket_size_classes, 869 1141 } 1142 + } 1143 + } 1144 + 1145 + pub fn upsert_record_indexes( 1146 + edges: &EdgeStore, 1147 + issue_states: &StateIndex<IssueStateKind>, 1148 + pull_statuses: &StateIndex<PullStatusKind>, 1149 + source: &AtUri<DefaultStr>, 1150 + record_edges: Vec<Edge>, 1151 + record: &Record, 1152 + ) -> ApplyOutcome { 1153 + let previous_state_entity = match record { 1154 + Record::IssueState(_) => issue_states.entity_for_source(source), 1155 + Record::PullStatus(_) => pull_statuses.entity_for_source(source), 1156 + _ => None, 1157 + }; 1158 + edges.upsert_source(source, record_edges); 1159 + let outcome = apply_record_state(issue_states, pull_statuses, source, record); 1160 + 1161 + match record { 1162 + Record::Issue(_) => edges.refresh_issue_counts(issue_states, source), 1163 + Record::Pull(_) => edges.refresh_pull_counts(pull_statuses, source), 1164 + Record::IssueState(state) => { 1165 + if let Some(previous) = previous_state_entity.as_ref() 1166 + && previous != &state.issue 1167 + { 1168 + edges.refresh_issue_counts(issue_states, previous); 1169 + } 1170 + edges.refresh_issue_counts(issue_states, &state.issue); 1171 + } 1172 + Record::PullStatus(status) => { 1173 + if let Some(previous) = previous_state_entity.as_ref() 1174 + && previous != &status.pull 1175 + { 1176 + edges.refresh_pull_counts(pull_statuses, previous); 1177 + } 1178 + edges.refresh_pull_counts(pull_statuses, &status.pull); 1179 + } 1180 + _ => {} 1181 + } 1182 + outcome 1183 + } 1184 + 1185 + pub fn delete_record_indexes( 1186 + edges: &EdgeStore, 1187 + issue_states: &StateIndex<IssueStateKind>, 1188 + pull_statuses: &StateIndex<PullStatusKind>, 1189 + source: &AtUri<DefaultStr>, 1190 + nsid: &Nsid<DefaultStr>, 1191 + ) { 1192 + edges.remove_source(source); 1193 + match nsid.as_ref() { 1194 + "sh.tangled.repo.issue" => { 1195 + issue_states.remove_entity(source); 1196 + edges.refresh_issue_counts(issue_states, source); 1197 + } 1198 + "sh.tangled.repo.pull" => { 1199 + pull_statuses.remove_entity(source); 1200 + edges.refresh_pull_counts(pull_statuses, source); 1201 + } 1202 + "sh.tangled.repo.issue.state" => { 1203 + if let Some(entity) = issue_states.remove_source(source) { 1204 + edges.refresh_issue_counts(issue_states, &entity); 1205 + } 1206 + } 1207 + "sh.tangled.repo.pull.status" => { 1208 + if let Some(entity) = pull_statuses.remove_source(source) { 1209 + edges.refresh_pull_counts(pull_statuses, &entity); 1210 + } 1211 + } 1212 + _ => {} 870 1213 } 871 1214 } 872 1215 ··· 1182 1525 ); 1183 1526 assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0); 1184 1527 assert_eq!(store.count(&EdgeKey::new(kind, new_subject)), 1); 1528 + } 1529 + 1530 + #[test] 1531 + fn state_counts_reproject_after_state_first_ingest_and_rekey() { 1532 + let store = store(); 1533 + let states = StateIndex::new(RuntimeHasher::default()); 1534 + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1535 + let old_repo = did("did:plc:limpet"); 1536 + let new_repo = did("did:plc:scallop"); 1537 + let kind = nsid("sh.tangled.repo.issue"); 1538 + let old_key = EdgeKey::new(kind.clone(), SubjectRef::Did(old_repo.clone())); 1539 + let new_key = EdgeKey::new(kind.clone(), SubjectRef::Did(new_repo.clone())); 1540 + 1541 + states.upsert( 1542 + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), 1543 + issue.clone(), 1544 + 100, 1545 + IssueStateKind::Closed, 1546 + ); 1547 + store.upsert_source( 1548 + &issue, 1549 + vec![Edge { 1550 + kind: kind.clone(), 1551 + subject: SubjectRef::Did(old_repo), 1552 + source: issue.clone(), 1553 + sort_micros: 1, 1554 + }], 1555 + ); 1556 + store.refresh_issue_counts(&states, &issue); 1557 + assert_eq!( 1558 + store.count_issue_state(&old_key, IssueStateKind::Closed, None), 1559 + FilteredCount { 1560 + count: 1, 1561 + distinct_authors: 1, 1562 + } 1563 + ); 1564 + 1565 + store.upsert_source( 1566 + &issue, 1567 + vec![Edge { 1568 + kind, 1569 + subject: SubjectRef::Did(new_repo), 1570 + source: issue.clone(), 1571 + sort_micros: 2, 1572 + }], 1573 + ); 1574 + store.refresh_issue_counts(&states, &issue); 1575 + assert_eq!( 1576 + store.count_issue_state(&old_key, IssueStateKind::Closed, None), 1577 + FilteredCount::default() 1578 + ); 1579 + assert_eq!( 1580 + store.count_issue_state(&new_key, IssueStateKind::Open, Some("did:plc:nel")), 1581 + FilteredCount { 1582 + count: 1, 1583 + distinct_authors: 1, 1584 + }, 1585 + "the old repo owner's state stops being accepted after the rekey", 1586 + ); 1587 + } 1588 + 1589 + #[test] 1590 + fn record_index_mutations_move_materialized_state_counts() { 1591 + use bobbin_types::sh_tangled::repo::issue::state::{State as IssueStateRecord, StateState}; 1592 + 1593 + let store = store(); 1594 + let issue_states = StateIndex::new(RuntimeHasher::default()); 1595 + let pull_statuses = StateIndex::new(RuntimeHasher::default()); 1596 + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); 1597 + let repo = did("did:plc:limpet"); 1598 + let key = EdgeKey::new(nsid("sh.tangled.repo.issue"), SubjectRef::Did(repo.clone())); 1599 + store.upsert_source( 1600 + &issue, 1601 + vec![Edge { 1602 + kind: nsid("sh.tangled.repo.issue"), 1603 + subject: SubjectRef::Did(repo), 1604 + source: issue.clone(), 1605 + sort_micros: 1, 1606 + }], 1607 + ); 1608 + store.refresh_issue_counts(&issue_states, &issue); 1609 + 1610 + let state_source = at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"); 1611 + let state_record = Record::IssueState(IssueStateRecord { 1612 + issue: issue.clone(), 1613 + created_at: jacquard_common::types::string::Datetime::raw_str("2026-05-01T00:00:00Z"), 1614 + state: StateState::ShTangledRepoIssueStateClosed, 1615 + extra_data: None, 1616 + }); 1617 + upsert_record_indexes( 1618 + &store, 1619 + &issue_states, 1620 + &pull_statuses, 1621 + &state_source, 1622 + Vec::new(), 1623 + &state_record, 1624 + ); 1625 + assert_eq!( 1626 + store 1627 + .count_issue_state(&key, IssueStateKind::Closed, None) 1628 + .count, 1629 + 1, 1630 + ); 1631 + 1632 + delete_record_indexes( 1633 + &store, 1634 + &issue_states, 1635 + &pull_statuses, 1636 + &state_source, 1637 + &nsid("sh.tangled.repo.issue.state"), 1638 + ); 1639 + assert_eq!( 1640 + store 1641 + .count_issue_state(&key, IssueStateKind::Open, None) 1642 + .count, 1643 + 1, 1644 + ); 1185 1645 } 1186 1646 1187 1647 #[test]

+10 -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>) { ··· 220 221 pub fn source_count(&self) -> usize { 221 222 self.reverse.len() 222 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()) 228 + } 223 229 } 224 230 225 231 #[derive(Clone, Copy, Debug, Eq, PartialEq)] ··· 357 363 let pull_idx = StateIndex::<PullStatusKind>::new(RuntimeHasher::default()); 358 364 let rec = Record::IssueState(IssueStateRec { 359 365 issue: at("at://did:plc:limpet/sh.tangled.repo.issue/i1"), 366 + created_at: jacquard_common::types::string::Datetime::raw_str("2026-05-01T00:00:00Z"), 360 367 state: StateState::Other(SmolStr::new_static("sh.tangled.repo.issue.state.reopened")), 361 368 extra_data: None, 362 369 });

+19 -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; ··· 1367 1367 } = upsert; 1368 1368 let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; 1369 1369 cache_body(ctx.records, &source, cid, bytes); 1370 - ctx.store.upsert_source(&source, edges); 1371 - let outcome = apply_record_state(ctx.issue_states, ctx.pull_statuses, &source, &parsed); 1370 + let outcome = upsert_record_indexes( 1371 + ctx.store, 1372 + ctx.issue_states, 1373 + ctx.pull_statuses, 1374 + &source, 1375 + edges, 1376 + &parsed, 1377 + ); 1372 1378 log_unknown_state_variant(outcome, &source); 1373 1379 index_search(ctx.search, ctx.resolver, &source, parsed).await; 1374 1380 } ··· 1403 1409 edges, 1404 1410 } => { 1405 1411 cache_body(records, &source, cid, bytes); 1406 - store.upsert_source(&source, edges); 1407 - let outcome = apply_record_state(issue_states, pull_statuses, &source, &parsed); 1412 + let outcome = 1413 + upsert_record_indexes(store, issue_states, pull_statuses, &source, edges, &parsed); 1408 1414 log_unknown_state_variant(outcome, &source); 1409 1415 index_search(search, resolver, &source, *parsed).await; 1410 1416 } 1411 1417 PendingOp::Delete { source, nsid } => { 1412 - store.remove_source(&source); 1413 - apply_delete_to_state_index(issue_states, pull_statuses, &source, &nsid); 1418 + delete_record_indexes(store, issue_states, pull_statuses, &source, &nsid); 1414 1419 records.remove(&source); 1415 1420 search.remove(&source).await; 1416 1421 } ··· 1486 1491 } 1487 1492 } 1488 1493 1489 - fn apply_delete_to_state_index( 1490 - issue_states: &StateIndex<IssueStateKind>, 1491 - pull_statuses: &StateIndex<PullStatusKind>, 1492 - source: &AtUri<DefaultStr>, 1493 - nsid: &Nsid<DefaultStr>, 1494 - ) { 1495 - match nsid.as_ref() { 1496 - "sh.tangled.repo.issue" => issue_states.remove_entity(source), 1497 - "sh.tangled.repo.pull" => pull_statuses.remove_entity(source), 1498 - "sh.tangled.repo.issue.state" => issue_states.remove_source(source), 1499 - "sh.tangled.repo.pull.status" => pull_statuses.remove_source(source), 1500 - _ => {} 1501 - } 1502 - } 1503 - 1504 1494 fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { 1505 1495 PromotionSignal { 1506 1496 rev_micros: record.map(|r| r.rev.timestamp()), ··· 2929 2919 did_subj("did:plc:abalone"), 2930 2920 ); 2931 2921 assert_eq!(store.count(&key), 1); 2922 + assert_eq!( 2923 + store 2924 + .count_issue_state(&key, IssueStateKind::Open, None) 2925 + .count, 2926 + 1, 2927 + "hydrant ingest materializes the default state count", 2928 + ); 2932 2929 } 2933 2930 2934 2931 #[derive(Default)]

+54 -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; ··· 15 15 pub trait ListFilter: serde::de::DeserializeOwned + Send + Sync + 'static { 16 16 fn predicate(&self, state: &AppState, subject: &SubjectRef) -> FilterPredicate; 17 17 fn is_identity(&self) -> bool; 18 + } 19 + 20 + pub trait CountFilter: ListFilter { 21 + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount; 18 22 } 19 23 20 24 #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)] ··· 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.edges.count_issue_state( 78 + key, 79 + wanted.into(), 80 + self.author.as_ref().map(AsRef::as_ref), 81 + ), 82 + None => count_by_author(state, key, self.author.as_ref()), 83 + } 84 + } 85 + } 86 + 87 + #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq)] 71 88 #[serde(rename_all = "lowercase")] 72 89 pub enum PullStatus { 73 90 Open, ··· 103 120 104 121 fn is_identity(&self) -> bool { 105 122 self.author.is_none() && self.status.is_none() 123 + } 124 + } 125 + 126 + impl CountFilter for PullFilter { 127 + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount { 128 + match self.status { 129 + Some(wanted) => state.edges.count_pull_status( 130 + key, 131 + wanted.into(), 132 + self.author.as_ref().map(AsRef::as_ref), 133 + ), 134 + None => count_by_author(state, key, self.author.as_ref()), 135 + } 136 + } 137 + } 138 + 139 + fn count_by_author( 140 + state: &AppState, 141 + key: &EdgeKey, 142 + author: Option<&Did<DefaultStr>>, 143 + ) -> FilteredCount { 144 + match author { 145 + Some(author) => { 146 + let count = state.edges.count_by_author(key, author.as_ref()); 147 + FilteredCount { 148 + count, 149 + distinct_authors: u64::from(count != 0), 150 + } 151 + } 152 + None => FilteredCount { 153 + count: state.edges.count(key), 154 + distinct_authors: state.edges.count_distinct_authors(key), 155 + }, 106 156 } 107 157 } 108 158

+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, 1873 + distinct_authors: counted.distinct_authors, 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