From 72cc58d80d02f299f2c9a2421f7347a20a030a3c Mon Sep 17 00:00:00 2001 From: dawn Date: Sat, 25 Jul 2026 00:31:21 +0300 Subject: [PATCH] bobbin: filter countIssues and countPulls by state Signed-off-by: dawn --- bobbin/crates/edge-index/src/lib.rs | 500 +++++++++++++++++++- bobbin/crates/edge-index/src/state_index.rs | 12 +- bobbin/crates/ingest/src/lib.rs | 42 +- bobbin/crates/xrpc/src/filter.rs | 51 +- bobbin/crates/xrpc/src/lib.rs | 37 +- bobbin/crates/xrpc/tests/aggregation.rs | 319 ++++++++++++- lexicons/repo/countIssues.json | 10 + lexicons/repo/countPulls.json | 10 + 8 files changed, 932 insertions(+), 49 deletions(-) diff --git a/bobbin/crates/edge-index/src/lib.rs b/bobbin/crates/edge-index/src/lib.rs index 32f0354e..0e94d7b7 100644 --- a/bobbin/crates/edge-index/src/lib.rs +++ b/bobbin/crates/edge-index/src/lib.rs @@ -3,7 +3,7 @@ use std::marker::PhantomData; use std::num::NonZeroU32; use std::ops::{Bound, ControlFlow}; use std::sync::atomic::{AtomicU32, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, RwLock}; const FILTER_SCAN_MULTIPLIER: usize = 64; const FILTER_SCAN_FLOOR: usize = 512; @@ -25,10 +25,12 @@ impl ScanState { } use bobbin_runtime::RuntimeHasher; -use bobbin_types::edges::Edge; +use bobbin_types::edges::{Edge, Record}; use bobbin_types::ids::EdgeKey; use either::Either; use jacquard_common::DefaultStr; +use jacquard_common::types::did::Did; +use jacquard_common::types::nsid::Nsid; use jacquard_common::types::string::AtUri; use lasso::{Key, Spur, ThreadedRodeo}; use scc::HashMap as SccMap; @@ -243,6 +245,47 @@ impl AsRef for EdgeItem { } } +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct FilteredCount { + pub count: Count, + pub distinct_authors: DistinctAuthorCount, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct Count(u64); + +impl Count { + pub const fn new(value: u64) -> Self { + Self(value) + } + + pub const fn get(self) -> u64 { + self.0 + } +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct DistinctAuthorCount(u64); + +impl DistinctAuthorCount { + pub const fn new(value: u64) -> Self { + Self(value) + } + + pub const fn get(self) -> u64 { + self.0 + } +} + +impl FilteredCount { + pub const fn new(count: u64, distinct_authors: u64) -> Self { + Self { + count: Count::new(count), + distinct_authors: DistinctAuthorCount::new(distinct_authors), + } + } +} + #[derive(Clone, Copy, Debug, Default)] pub struct EdgeMemReport { pub key_count: u64, @@ -419,14 +462,79 @@ struct ReverseEntry { sort_micros: u64, } +#[derive(Clone, Copy)] +struct ProjectedState { + key: EdgeKeyId, + kind: K, + author: Option, +} + +struct CountBucket { + count: u64, + authors: HashMap, +} + +struct StateCountInner { + projected: HashMap; 1]>, RuntimeHasher>, + buckets: HashMap<(EdgeKeyId, K), CountBucket, RuntimeHasher>, +} + +struct StateCountIndex { + inner: RwLock>, + hasher: RuntimeHasher, +} + +impl StateCountIndex { + fn new(hasher: RuntimeHasher) -> Self { + Self { + inner: RwLock::new(StateCountInner { + projected: HashMap::with_hasher(hasher.clone()), + buckets: HashMap::with_hasher(hasher.clone()), + }), + hasher, + } + } + + fn heap_bytes(&self) -> u64 { + let inner = self + .inner + .read() + .expect("state-count index rwlock poisoned"); + let projected = inner.projected.capacity() + * (std::mem::size_of::() + + std::mem::size_of::; 1]>>() + + 1) + + inner + .projected + .values() + .filter(|states| states.spilled()) + .map(|states| states.capacity() * std::mem::size_of::>()) + .sum::(); + let buckets = inner.buckets.capacity() + * (std::mem::size_of::<(EdgeKeyId, K)>() + std::mem::size_of::() + 1); + let author_slots = inner + .buckets + .values() + .map(|bucket| { + bucket.authors.capacity() + * (std::mem::size_of::() + std::mem::size_of::() + 1) + }) + .sum::(); + (projected + buckets + author_slots) as u64 + } +} + pub struct EdgeStore { source_interner: Arc>, did_interner: Arc>, collection_interner: Arc>, key_ids: SccMap, + keys: SccMap, next_key_id: AtomicU32, forward: SccMap, reverse: SccMap, RuntimeHasher>, + issue_counts: StateCountIndex, + pull_counts: StateCountIndex, hasher: RuntimeHasher, writer: Mutex<()>, } @@ -438,20 +546,24 @@ impl EdgeStore { did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), collection_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())), key_ids: SccMap::with_hasher(hasher.clone()), + keys: SccMap::with_hasher(hasher.clone()), next_key_id: AtomicU32::new(0), forward: SccMap::with_hasher(hasher.clone()), reverse: SccMap::with_hasher(hasher.clone()), + issue_counts: StateCountIndex::new(hasher.clone()), + pull_counts: StateCountIndex::new(hasher.clone()), hasher, writer: Mutex::new(()), } } fn intern_key(&self, key: EdgeKey) -> EdgeKeyId { - match self.key_ids.entry_sync(key) { + match self.key_ids.entry_sync(key.clone()) { Entry::Occupied(e) => *e.get(), Entry::Vacant(e) => { let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed)); e.insert_entry(id); + let _ = self.keys.insert_sync(id, key); id } } @@ -650,6 +762,197 @@ impl EdgeStore { }) .unwrap_or(0) } + + pub fn count_by_author(&self, key: &EdgeKey, author: &Did) -> u64 { + let Some(author) = self + .did_interner + .get(author.as_ref()) + .map(AuthorId::from_spur) + else { + return 0; + }; + self.lookup_key(key) + .and_then(|id| { + self.forward.read_sync(&id, |_, sources| match sources { + Sources::Large(big) => big + .authors + .get(&author) + .map_or(0, |count| count.get() as u64), + Sources::Small(keys) => keys + .iter() + .filter(|key| self.author_of(key.source) == Some(author)) + .count() as u64, + }) + }) + .unwrap_or(0) + } + + fn current_state_projection( + &self, + states: &StateIndex, + entity: &AtUri, + edge_kind: &str, + ) -> (SourceId, SmallVec<[ProjectedState; 1]>) + where + K: StateKind + Default, + { + let source = self.intern_source(entity); + let author = self.author_of(source); + let entity_author = source_authority_did(entity); + let reverse = self + .reverse + .read_sync(&source, |_, entries| entries.clone()) + .unwrap_or_default(); + let projected = reverse + .into_iter() + .filter_map(|entry| { + let key = self.keys.read_sync(&entry.key_id, |_, key| key.clone())?; + if key.kind.as_ref() != edge_kind { + return None; + } + let repo_owner = key.subject.as_did()?; + let kind = states + .latest_by(entity, |state_source| { + let state_author = source_authority_did(state_source); + state_author == entity_author || state_author == Some(repo_owner.as_ref()) + }) + .map(|(kind, _)| kind) + .unwrap_or_default(); + Some(ProjectedState { + key: entry.key_id, + kind, + author, + }) + }) + .collect(); + (source, projected) + } + + fn refresh_state_counts( + &self, + index: &StateCountIndex, + states: &StateIndex, + entity: &AtUri, + edge_kind: &str, + ) where + K: StateKind + Default, + { + // compute under the write lock so concurrent updates dont write a stale projection + let mut inner = index + .inner + .write() + .expect("state-count index rwlock poisoned"); + let (source, current) = self.current_state_projection(states, entity, edge_kind); + + if let Some(previous) = inner.projected.remove(&source) { + for projected in previous { + let bucket_key = (projected.key, projected.kind); + let remove = if let Some(bucket) = inner.buckets.get_mut(&bucket_key) { + bucket.count -= 1; + if let Some(author) = projected.author { + drop_author(&mut bucket.authors, author); + } + bucket.count == 0 + } else { + false + }; + if remove { + inner.buckets.remove(&bucket_key); + } + } + } + + for projected in current.iter().copied() { + let bucket = inner + .buckets + .entry((projected.key, projected.kind)) + .or_insert_with(|| CountBucket { + count: 0, + authors: HashMap::with_hasher(index.hasher.clone()), + }); + bucket.count += 1; + if let Some(author) = projected.author { + bump_author(&mut bucket.authors, author); + } + } + if !current.is_empty() { + inner.projected.insert(source, current); + } + } + + pub fn refresh_issue_counts( + &self, + states: &StateIndex, + entity: &AtUri, + ) { + self.refresh_state_counts(&self.issue_counts, states, entity, "sh.tangled.repo.issue"); + } + + pub fn refresh_pull_counts( + &self, + states: &StateIndex, + entity: &AtUri, + ) { + self.refresh_state_counts(&self.pull_counts, states, entity, "sh.tangled.repo.pull"); + } + + fn count_state( + &self, + index: &StateCountIndex, + key: &EdgeKey, + kind: K, + author: Option<&Did>, + ) -> FilteredCount + where + K: StateKind, + { + let Some(key) = self.lookup_key(key) else { + return FilteredCount::default(); + }; + let author = match author { + Some(author) => match self.did_interner.get(author).map(AuthorId::from_spur) { + Some(author) => Some(author), + None => return FilteredCount::default(), + }, + None => None, + }; + let inner = index + .inner + .read() + .expect("state-count index rwlock poisoned"); + let Some(bucket) = inner.buckets.get(&(key, kind)) else { + return FilteredCount::default(); + }; + match author { + Some(author) => { + let count = bucket + .authors + .get(&author) + .map_or(0, |count| count.get() as u64); + FilteredCount::new(count, u64::from(count != 0)) + } + None => FilteredCount::new(bucket.count, bucket.authors.len() as u64), + } + } + + pub fn count_issue_state( + &self, + key: &EdgeKey, + kind: IssueStateKind, + author: Option<&Did>, + ) -> FilteredCount { + self.count_state(&self.issue_counts, key, kind, author) + } + + pub fn count_pull_status( + &self, + key: &EdgeKey, + kind: PullStatusKind, + author: Option<&Did>, + ) -> FilteredCount { + self.count_state(&self.pull_counts, key, kind, author) + } + /// answers "did the viewer star/follow/etc. this subject, and with what rkey" pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option> { let author_spur = self.did_interner.get(viewer)?; @@ -821,7 +1124,8 @@ impl EdgeStore { let mut edges_total = 0u64; let mut author_refs_total = 0u64; - let mut forward_struct_bytes = 0u64; + let mut forward_struct_bytes = + self.issue_counts.heap_bytes() + self.pull_counts.heap_bytes(); let mut max_bucket = 0u64; let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT]; self.forward.iter_sync(|_, sources| { @@ -836,7 +1140,9 @@ impl EdgeStore { true }); let key_interner_bytes = self.key_ids.len() as u64 - * (edge_key + std::mem::size_of::() as u64 + SCC_SLOT); + * (edge_key + std::mem::size_of::() as u64 + SCC_SLOT) + + self.keys.len() as u64 + * (edge_key + std::mem::size_of::() as u64 + SCC_SLOT); let mut reverse_entries = 0u64; let mut reverse_cap = 0u64; @@ -870,6 +1176,77 @@ impl EdgeStore { } } +pub fn upsert_record_indexes( + edges: &EdgeStore, + issue_states: &StateIndex, + pull_statuses: &StateIndex, + source: &AtUri, + record_edges: Vec, + record: &Record, +) -> ApplyOutcome { + let previous_state_entity = match record { + Record::IssueState(_) => issue_states.entity_for_source(source), + Record::PullStatus(_) => pull_statuses.entity_for_source(source), + _ => None, + }; + edges.upsert_source(source, record_edges); + let outcome = apply_record_state(issue_states, pull_statuses, source, record); + + match record { + Record::Issue(_) => edges.refresh_issue_counts(issue_states, source), + Record::Pull(_) => edges.refresh_pull_counts(pull_statuses, source), + Record::IssueState(state) => { + if let Some(previous) = previous_state_entity.as_ref() + && previous != &state.issue + { + edges.refresh_issue_counts(issue_states, previous); + } + edges.refresh_issue_counts(issue_states, &state.issue); + } + Record::PullStatus(status) => { + if let Some(previous) = previous_state_entity.as_ref() + && previous != &status.pull + { + edges.refresh_pull_counts(pull_statuses, previous); + } + edges.refresh_pull_counts(pull_statuses, &status.pull); + } + _ => {} + } + outcome +} + +pub fn delete_record_indexes( + edges: &EdgeStore, + issue_states: &StateIndex, + pull_statuses: &StateIndex, + source: &AtUri, + nsid: &Nsid, +) { + edges.remove_source(source); + match nsid.as_ref() { + "sh.tangled.repo.issue" => { + issue_states.remove_entity(source); + edges.refresh_issue_counts(issue_states, source); + } + "sh.tangled.repo.pull" => { + pull_statuses.remove_entity(source); + edges.refresh_pull_counts(pull_statuses, source); + } + "sh.tangled.repo.issue.state" => { + if let Some(entity) = issue_states.remove_source(source) { + edges.refresh_issue_counts(issue_states, &entity); + } + } + "sh.tangled.repo.pull.status" => { + if let Some(entity) = pull_statuses.remove_source(source) { + edges.refresh_pull_counts(pull_statuses, &entity); + } + } + _ => {} + } +} + fn directed_slice( sources: &[BucketKey], cursor: PageCursor, @@ -1184,6 +1561,119 @@ mod tests { assert_eq!(store.count(&EdgeKey::new(kind, new_subject)), 1); } + #[test] + fn state_counts_reproject_after_state_first_ingest_and_rekey() { + let store = store(); + let states = StateIndex::new(RuntimeHasher::default()); + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + let old_repo = did("did:plc:limpet"); + let new_repo = did("did:plc:scallop"); + let kind = nsid("sh.tangled.repo.issue"); + let old_key = EdgeKey::new(kind.clone(), SubjectRef::Did(old_repo.clone())); + let new_key = EdgeKey::new(kind.clone(), SubjectRef::Did(new_repo.clone())); + + states.upsert( + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), + issue.clone(), + 100, + IssueStateKind::Closed, + ); + store.upsert_source( + &issue, + vec![Edge { + kind: kind.clone(), + subject: SubjectRef::Did(old_repo), + source: issue.clone(), + sort_micros: 1, + }], + ); + store.refresh_issue_counts(&states, &issue); + assert_eq!( + store.count_issue_state(&old_key, IssueStateKind::Closed, None), + FilteredCount::new(1, 1) + ); + + store.upsert_source( + &issue, + vec![Edge { + kind, + subject: SubjectRef::Did(new_repo), + source: issue.clone(), + sort_micros: 2, + }], + ); + store.refresh_issue_counts(&states, &issue); + assert_eq!( + store.count_issue_state(&old_key, IssueStateKind::Closed, None), + FilteredCount::default() + ); + assert_eq!( + store.count_issue_state(&new_key, IssueStateKind::Open, Some(&did("did:plc:nel")),), + FilteredCount::new(1, 1), + "the old repo owner's state stops being accepted after the rekey", + ); + } + + #[test] + fn record_index_mutations_move_materialized_state_counts() { + use bobbin_types::sh_tangled::repo::issue::state::{State as IssueStateRecord, StateState}; + + let store = store(); + let issue_states = StateIndex::new(RuntimeHasher::default()); + let pull_statuses = StateIndex::new(RuntimeHasher::default()); + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + let repo = did("did:plc:limpet"); + let key = EdgeKey::new(nsid("sh.tangled.repo.issue"), SubjectRef::Did(repo.clone())); + store.upsert_source( + &issue, + vec![Edge { + kind: nsid("sh.tangled.repo.issue"), + subject: SubjectRef::Did(repo), + source: issue.clone(), + sort_micros: 1, + }], + ); + store.refresh_issue_counts(&issue_states, &issue); + + let state_source = at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"); + let state_record = Record::IssueState(IssueStateRecord { + issue: issue.clone(), + created_at: jacquard_common::types::string::Datetime::raw_str("2026-05-01T00:00:00Z"), + state: StateState::ShTangledRepoIssueStateClosed, + extra_data: None, + }); + upsert_record_indexes( + &store, + &issue_states, + &pull_statuses, + &state_source, + Vec::new(), + &state_record, + ); + assert_eq!( + store + .count_issue_state(&key, IssueStateKind::Closed, None) + .count + .get(), + 1, + ); + + delete_record_indexes( + &store, + &issue_states, + &pull_statuses, + &state_source, + &nsid("sh.tangled.repo.issue.state"), + ); + assert_eq!( + store + .count_issue_state(&key, IssueStateKind::Open, None) + .count + .get(), + 1, + ); + } + #[test] fn list_pages_in_sort_order() { let store = store(); diff --git a/bobbin/crates/edge-index/src/state_index.rs b/bobbin/crates/edge-index/src/state_index.rs index 536df7f7..c9694ea3 100644 --- a/bobbin/crates/edge-index/src/state_index.rs +++ b/bobbin/crates/edge-index/src/state_index.rs @@ -10,7 +10,7 @@ use jacquard_common::types::string::AtUri; use scc::HashMap as SccMap; use scc::hash_map::Entry; -pub trait StateKind: Copy + Eq + std::fmt::Debug + Send + Sync + 'static { +pub trait StateKind: Copy + Eq + std::hash::Hash + std::fmt::Debug + Send + Sync + 'static { fn wire(self) -> &'static str; } @@ -134,18 +134,19 @@ impl StateIndex { } } - pub fn remove_source(&self, source: &AtUri) { + pub fn remove_source(&self, source: &AtUri) -> Option> { let _w = self .writer .lock() .expect("state-index writer mutex poisoned"); let Entry::Occupied(occ) = self.reverse.entry_sync(source.clone()) else { - return; + return None; }; let entity = occ.get().entity.clone(); let sort_micros = occ.get().sort_micros; let _ = occ.remove(); self.remove_forward(&entity, sort_micros, source); + Some(entity) } pub fn remove_entity(&self, entity: &AtUri) { @@ -220,6 +221,11 @@ impl StateIndex { pub fn source_count(&self) -> usize { self.reverse.len() } + + pub fn entity_for_source(&self, source: &AtUri) -> Option> { + self.reverse + .read_sync(source, |_, entry| entry.entity.clone()) + } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] diff --git a/bobbin/crates/ingest/src/lib.rs b/bobbin/crates/ingest/src/lib.rs index 36941663..5924b630 100644 --- a/bobbin/crates/ingest/src/lib.rs +++ b/bobbin/crates/ingest/src/lib.rs @@ -5,7 +5,7 @@ use std::time::Duration; use bobbin_edge_index::{ ApplyOutcome, Coverage, CoverageWatch, EdgeStore, HydrantCursor, IssueStateKind, - PromotionSignal, PullStatusKind, StateIndex, apply_record_state, + PromotionSignal, PullStatusKind, StateIndex, delete_record_indexes, upsert_record_indexes, }; use bobbin_knot_ingest::{CapabilityGate, KnotRegistry}; use bobbin_record_lru::RecordStore; @@ -1359,8 +1359,14 @@ async fn finalize_drained( } = upsert; let edges = normalize_subjects(edges, ctx.resolver, ctx.coverage, None).await; cache_body(ctx.records, &source, cid, bytes); - ctx.store.upsert_source(&source, edges); - let outcome = apply_record_state(ctx.issue_states, ctx.pull_statuses, &source, &parsed); + let outcome = upsert_record_indexes( + ctx.store, + ctx.issue_states, + ctx.pull_statuses, + &source, + edges, + &parsed, + ); log_unknown_state_variant(outcome, &source); index_search(ctx.search, ctx.resolver, &source, parsed).await; } @@ -1395,14 +1401,13 @@ async fn commit_pending( edges, } => { cache_body(records, &source, cid, bytes); - store.upsert_source(&source, edges); - let outcome = apply_record_state(issue_states, pull_statuses, &source, &parsed); + let outcome = + upsert_record_indexes(store, issue_states, pull_statuses, &source, edges, &parsed); log_unknown_state_variant(outcome, &source); index_search(search, resolver, &source, *parsed).await; } PendingOp::Delete { source, nsid } => { - store.remove_source(&source); - apply_delete_to_state_index(issue_states, pull_statuses, &source, &nsid); + delete_record_indexes(store, issue_states, pull_statuses, &source, &nsid); records.remove(&source); search.remove(&source).await; } @@ -1478,21 +1483,6 @@ fn log_unknown_state_variant(outcome: ApplyOutcome, source: &AtUri) } } -fn apply_delete_to_state_index( - issue_states: &StateIndex, - pull_statuses: &StateIndex, - source: &AtUri, - nsid: &Nsid, -) { - match nsid.as_ref() { - "sh.tangled.repo.issue" => issue_states.remove_entity(source), - "sh.tangled.repo.pull" => pull_statuses.remove_entity(source), - "sh.tangled.repo.issue.state" => issue_states.remove_source(source), - "sh.tangled.repo.pull.status" => pull_statuses.remove_source(source), - _ => {} - } -} - fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { PromotionSignal { rev_micros: record.map(|r| r.rev.timestamp()), @@ -2922,6 +2912,14 @@ mod tests { did_subj("did:plc:abalone"), ); assert_eq!(store.count(&key), 1); + assert_eq!( + store + .count_issue_state(&key, IssueStateKind::Open, None) + .count + .get(), + 1, + "hydrant ingest materializes the default state count", + ); } #[derive(Default)] diff --git a/bobbin/crates/xrpc/src/filter.rs b/bobbin/crates/xrpc/src/filter.rs index 240efc6b..9ead9cb1 100644 --- a/bobbin/crates/xrpc/src/filter.rs +++ b/bobbin/crates/xrpc/src/filter.rs @@ -1,7 +1,7 @@ use std::sync::Arc; -use bobbin_edge_index::{IssueStateKind, PullStatusKind, StateIndex}; -use bobbin_types::ids::SubjectRef; +use bobbin_edge_index::{FilteredCount, IssueStateKind, PullStatusKind, StateIndex}; +use bobbin_types::ids::{EdgeKey, SubjectRef}; use jacquard_common::DefaultStr; use jacquard_common::types::did::Did; use jacquard_common::types::string::AtUri; @@ -17,6 +17,10 @@ pub trait ListFilter: serde::de::DeserializeOwned + Send + Sync + 'static { fn is_identity(&self) -> bool; } +pub trait CountFilter: ListFilter { + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount; +} + #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)] pub struct NoFilter; @@ -30,7 +34,7 @@ impl ListFilter for NoFilter { } } -#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq)] #[serde(rename_all = "lowercase")] pub enum IssueState { Open, @@ -67,7 +71,18 @@ impl ListFilter for IssueFilter { } } -#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)] +impl CountFilter for IssueFilter { + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount { + match self.state { + Some(wanted) => state + .edges + .count_issue_state(key, wanted.into(), self.author.as_ref()), + None => count_by_author(state, key, self.author.as_ref()), + } + } +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq)] #[serde(rename_all = "lowercase")] pub enum PullStatus { Open, @@ -106,6 +121,34 @@ impl ListFilter for PullFilter { } } +impl CountFilter for PullFilter { + fn count(&self, state: &AppState, key: &EdgeKey) -> FilteredCount { + match self.status { + Some(wanted) => state + .edges + .count_pull_status(key, wanted.into(), self.author.as_ref()), + None => count_by_author(state, key, self.author.as_ref()), + } + } +} + +fn count_by_author( + state: &AppState, + key: &EdgeKey, + author: Option<&Did>, +) -> FilteredCount { + match author { + Some(author) => { + let count = state.edges.count_by_author(key, author); + FilteredCount::new(count, u64::from(count != 0)) + } + None => FilteredCount::new( + state.edges.count(key), + state.edges.count_distinct_authors(key), + ), + } +} + fn compose_state_filter( author: Option>, want: Option, diff --git a/bobbin/crates/xrpc/src/lib.rs b/bobbin/crates/xrpc/src/lib.rs index 9c1b5c0e..0b3a7e7f 100644 --- a/bobbin/crates/xrpc/src/lib.rs +++ b/bobbin/crates/xrpc/src/lib.rs @@ -105,11 +105,10 @@ mod recordpath; pub use backpressure::{ HeavyLimiter, HeavyPermit, MaxInFlight, PerRequestAnonBytes, PressureVerdict, ReservedFloor, }; -use filter::{IssueFilter, ListFilter, NoFilter, PullFilter}; +use filter::{CountFilter, IssueFilter, ListFilter, NoFilter, PullFilter}; const DEFAULT_LIMIT: u32 = 50; const FETCH_CONCURRENCY: usize = 8; - #[derive(Clone)] pub struct AppState { pub records: Arc, @@ -681,6 +680,13 @@ struct CountQuery { subject: SubjectQuery, } +#[derive(Debug, Deserialize)] +struct TypedCountQuery { + subject: SubjectQuery, + #[serde(flatten)] + filter: F, +} + #[derive(Debug, Deserialize)] struct GetEdgeQuery { actor: Did, @@ -899,7 +905,7 @@ where } } -#[derive(Serialize)] +#[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "camelCase")] struct CountResponse { count: u64, @@ -1851,6 +1857,23 @@ fn count_for( }) } +fn count_typed_for( + state: &AppState, + q: TypedCountQuery, +) -> Result +where + R: XrpcResp + HasSubject, + F: CountFilter, +{ + let subject = parse_subject(&q.subject, R::SHAPE)?; + let key = EdgeKey::new(nsid_static(R::NSID), subject); + let counted = q.filter.count(state, &key); + Ok(CountResponse { + count: counted.count.get(), + distinct_authors: counted.distinct_authors.get(), + }) +} + // does `actor` have an edge of this kind pointing at `subject`, returns its own uri fn get_for( state: &AppState, @@ -1982,9 +2005,9 @@ async fn list_issues( async fn count_issues( State(state): State, - XrpcQuery(q): XrpcQuery, + XrpcQuery(q): XrpcQuery>, ) -> Result, XrpcError> { - count_for::(&state, q).map(Json) + count_typed_for::(&state, q).map(Json) } async fn list_pulls( @@ -2007,9 +2030,9 @@ async fn list_pulls( async fn count_pulls( State(state): State, - XrpcQuery(q): XrpcQuery, + XrpcQuery(q): XrpcQuery>, ) -> Result, XrpcError> { - count_for::(&state, q).map(Json) + count_typed_for::(&state, q).map(Json) } async fn list_feed_comments( diff --git a/bobbin/crates/xrpc/tests/aggregation.rs b/bobbin/crates/xrpc/tests/aggregation.rs index 9f43c320..14d82023 100644 --- a/bobbin/crates/xrpc/tests/aggregation.rs +++ b/bobbin/crates/xrpc/tests/aggregation.rs @@ -113,6 +113,43 @@ impl Harness { source: source.clone(), sort_micros: next_sort_micros(), }); + match kind.as_ref() { + "sh.tangled.repo.issue" => self + .edges + .refresh_issue_counts(&self.state.issue_states, source), + "sh.tangled.repo.pull" => self + .edges + .refresh_pull_counts(&self.state.pull_statuses, source), + _ => {} + } + } + + fn upsert_issue_state( + &self, + source: AtUri, + issue: AtUri, + sort_micros: u64, + kind: IssueStateKind, + ) { + self.state + .issue_states + .upsert(source, issue.clone(), sort_micros, kind); + self.edges + .refresh_issue_counts(&self.state.issue_states, &issue); + } + + fn upsert_pull_status( + &self, + source: AtUri, + pull: AtUri, + sort_micros: u64, + kind: PullStatusKind, + ) { + self.state + .pull_statuses + .upsert(source, pull.clone(), sort_micros, kind); + self.edges + .refresh_pull_counts(&self.state.pull_statuses, &pull); } async fn mount( @@ -1717,13 +1754,13 @@ async fn list_issues_includes_state_comment_count_and_state_updated_at() { &at("at://did:plc:teq/sh.tangled.feed.comment/c2"), ); - h.state.issue_states.upsert( + h.upsert_issue_state( at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), issue_uri.clone(), 1_777_593_600_000_000, IssueStateKind::Open, ); - h.state.issue_states.upsert( + h.upsert_issue_state( at("at://did:plc:nel/sh.tangled.repo.issue.state/s2"), issue_uri.clone(), 1_777_593_700_000_000, @@ -1886,13 +1923,13 @@ async fn list_pulls_includes_merged_state_and_comment_count() { &pull_uri, &at("at://did:plc:teq/sh.tangled.feed.comment/c1"), ); - h.state.pull_statuses.upsert( + h.upsert_pull_status( at("at://did:plc:nel/sh.tangled.repo.pull.status/s1"), pull_uri.clone(), 1_777_593_600_000_000, PullStatusKind::Open, ); - h.state.pull_statuses.upsert( + h.upsert_pull_status( at("at://did:plc:nel/sh.tangled.repo.pull.status/s2"), pull_uri.clone(), 1_777_593_800_000_000, @@ -1951,6 +1988,272 @@ async fn list_issues_state_filter_open_includes_records_without_state() { ); } +#[tokio::test] +async fn count_issues_splits_open_from_total() { + let h = Harness::new().await; + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + for rk in ["i1", "i2", "i3"] { + h.add_edge( + &nsid("sh.tangled.repo.issue"), + &subject, + &at(&format!("at://did:plc:nel/sh.tangled.repo.issue/{rk}")), + ); + } + // the repo owner closing it counts. with no record at all it is open + h.upsert_issue_state( + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), + at("at://did:plc:nel/sh.tangled.repo.issue/i1"), + 1_777_593_600_000_000, + IssueStateKind::Closed, + ); + + let app = router(h.state.clone()); + let counts = |args: &'static [(&'static str, &'static str)]| { + let app = app.clone(); + let subject = subject.clone(); + async move { + let (status, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.countIssues", + subject.as_ref(), + args, + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + body["count"].as_u64().expect("count") + } + }; + + assert_eq!(counts(&[]).await, 3, "no filter is every issue"); + assert_eq!(counts(&[("state", "open")]).await, 2); + assert_eq!(counts(&[("state", "closed")]).await, 1); +} + +#[tokio::test] +async fn count_issues_open_ignores_third_party_state_source() { + let h = Harness::new().await; + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + let issue_uri = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + h.add_edge(&nsid("sh.tangled.repo.issue"), &subject, &issue_uri); + h.upsert_issue_state( + at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), + issue_uri, + 1_777_593_800_000_000, + IssueStateKind::Closed, + ); + + let app = router(h.state.clone()); + let (status, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.countIssues", + subject.as_ref(), + &[("state", "open")], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + body["count"], + json!(1), + "a stranger closing an issue must not take it out of the open count", + ); +} + +#[tokio::test] +async fn filtered_count_follows_state_changes_and_repo_rekeying() { + let h = Harness::new().await; + let old_repo = did("did:plc:limpet"); + let new_repo = did("did:plc:scallop"); + let old_subject = at(&format!("at://{}", old_repo.as_ref())); + let new_subject = at(&format!("at://{}", new_repo.as_ref())); + let issue = at("at://did:plc:nel/sh.tangled.repo.issue/i1"); + let kind = nsid("sh.tangled.repo.issue"); + h.edges.upsert_source( + &issue, + vec![Edge { + kind: kind.clone(), + subject: SubjectRef::Did(old_repo), + source: issue.clone(), + sort_micros: next_sort_micros(), + }], + ); + h.edges.refresh_issue_counts(&h.state.issue_states, &issue); + + let app = router(h.state.clone()); + let count_open = |subject: AtUri| { + let app = app.clone(); + async move { + let (status, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.countIssues", + subject.as_ref(), + &[("state", "open")], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK); + body["count"].as_u64().expect("count") + } + }; + + assert_eq!(count_open(old_subject.clone()).await, 1); + assert_eq!(count_open(old_subject.clone()).await, 1, "repeated read"); + assert_eq!( + count_open(new_subject.clone()).await, + 0, + "the issue does not belong to the destination yet", + ); + + h.upsert_issue_state( + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), + issue.clone(), + 1_777_593_600_000_000, + IssueStateKind::Closed, + ); + assert_eq!( + count_open(old_subject.clone()).await, + 0, + "state changes are visible without cached classification", + ); + + h.edges.upsert_source( + &issue, + vec![Edge { + kind, + subject: SubjectRef::Did(new_repo), + source: issue.clone(), + sort_micros: next_sort_micros(), + }], + ); + h.edges.refresh_issue_counts(&h.state.issue_states, &issue); + assert_eq!(count_open(old_subject).await, 0); + assert_eq!( + count_open(new_subject).await, + 1, + "re-keying reprojects acceptance against the new repo owner", + ); +} + +#[tokio::test] +async fn count_pulls_splits_by_status() { + let h = Harness::new().await; + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + for rk in ["p1", "p2", "p3"] { + h.add_edge( + &nsid("sh.tangled.repo.pull"), + &subject, + &at(&format!("at://did:plc:nel/sh.tangled.repo.pull/{rk}")), + ); + } + h.upsert_pull_status( + at("at://did:plc:limpet/sh.tangled.repo.pull.status/s1"), + at("at://did:plc:nel/sh.tangled.repo.pull/p1"), + 1_777_593_600_000_000, + PullStatusKind::Merged, + ); + h.upsert_pull_status( + at("at://did:plc:limpet/sh.tangled.repo.pull.status/s2"), + at("at://did:plc:nel/sh.tangled.repo.pull/p2"), + 1_777_593_600_000_000, + PullStatusKind::Closed, + ); + + let app = router(h.state.clone()); + let counts = |args: &'static [(&'static str, &'static str)]| { + let app = app.clone(); + let subject = subject.clone(); + async move { + let (_, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.countPulls", + subject.as_ref(), + args, + )) + .await + .unwrap(), + ) + .await; + body["count"].as_u64().expect("count") + } + }; + + assert_eq!(counts(&[]).await, 3); + assert_eq!(counts(&[("status", "open")]).await, 1); + assert_eq!(counts(&[("status", "closed")]).await, 1); + assert_eq!(counts(&[("status", "merged")]).await, 1); +} + +#[tokio::test] +async fn count_issues_distinct_authors_follows_the_filter() { + let h = Harness::new().await; + let repo = did("did:plc:limpet"); + let subject = at(&format!("at://{}", repo.as_ref())); + h.add_edge( + &nsid("sh.tangled.repo.issue"), + &subject, + &at("at://did:plc:nel/sh.tangled.repo.issue/i1"), + ); + h.add_edge( + &nsid("sh.tangled.repo.issue"), + &subject, + &at("at://did:plc:olaren/sh.tangled.repo.issue/i2"), + ); + h.upsert_issue_state( + at("at://did:plc:limpet/sh.tangled.repo.issue.state/s1"), + at("at://did:plc:olaren/sh.tangled.repo.issue/i2"), + 1_777_593_600_000_000, + IssueStateKind::Closed, + ); + + let app = router(h.state.clone()); + let (_, body) = json_response( + app.oneshot(list_request( + "sh.tangled.repo.countIssues", + subject.as_ref(), + &[("state", "open")], + )) + .await + .unwrap(), + ) + .await; + assert_eq!(body["count"], json!(1)); + assert_eq!( + body["distinctAuthors"], + json!(1), + "authors of filtered-out issues must not be counted", + ); + + for (args, count, distinct) in [ + (&[("author", "did:plc:nel")][..], 1, 1), + (&[("author", "did:plc:nel"), ("state", "open")][..], 1, 1), + (&[("author", "did:plc:olaren"), ("state", "open")][..], 0, 0), + ] { + let (_, body) = json_response( + router(h.state.clone()) + .oneshot(list_request( + "sh.tangled.repo.countIssues", + subject.as_ref(), + args, + )) + .await + .unwrap(), + ) + .await; + assert_eq!(body["count"], json!(count)); + assert_eq!(body["distinctAuthors"], json!(distinct)); + } +} + #[tokio::test] async fn list_issues_state_filter_ignores_third_party_state_source() { let h = Harness::new().await; @@ -1965,7 +2268,7 @@ async fn list_issues_state_filter_ignores_third_party_state_source() { issue_body(&repo, "open issue"), ) .await; - h.state.issue_states.upsert( + h.upsert_issue_state( at("at://did:plc:nautilus/sh.tangled.repo.issue.state/spoof"), issue_uri.clone(), 1_777_593_800_000_000, @@ -2011,7 +2314,7 @@ async fn list_pulls_status_filter_ignores_third_party_status_source() { pull_body(&repo, "wip"), ) .await; - h.state.pull_statuses.upsert( + h.upsert_pull_status( at("at://did:plc:nautilus/sh.tangled.repo.pull.status/spoof"), pull_uri.clone(), 1_777_593_800_000_000, @@ -2052,7 +2355,7 @@ async fn list_issues_state_filter_accepts_repo_owner_state_source() { issue_body(&repo_owner, "owner closed"), ) .await; - h.state.issue_states.upsert( + h.upsert_issue_state( at("at://did:plc:limpet/sh.tangled.repo.issue.state/legit"), issue_uri.clone(), 1_777_593_800_000_000, @@ -2183,7 +2486,7 @@ async fn list_issues_by_state_filter_narrows_results() { issue_body(&repo, "shut"), ) .await; - h.state.issue_states.upsert( + h.upsert_issue_state( at("at://did:plc:nel/sh.tangled.repo.issue.state/s1"), closed_uri.clone(), 1_777_593_800_000_000, diff --git a/lexicons/repo/countIssues.json b/lexicons/repo/countIssues.json index 6179481c..f56ab0e4 100644 --- a/lexicons/repo/countIssues.json +++ b/lexicons/repo/countIssues.json @@ -12,6 +12,16 @@ "type": "string", "format": "did", "description": "Repo DID to list issues for" + }, + "author": { + "type": "string", + "format": "did", + "description": "Restrict to issues authored by this user DID." + }, + "state": { + "type": "string", + "knownValues": ["open", "closed"], + "description": "Restrict to issues whose latest derived state matches." } } }, diff --git a/lexicons/repo/countPulls.json b/lexicons/repo/countPulls.json index 616b8feb..b3acd9cb 100644 --- a/lexicons/repo/countPulls.json +++ b/lexicons/repo/countPulls.json @@ -12,6 +12,16 @@ "type": "string", "format": "did", "description": "Repo DID to list pulls for" + }, + "author": { + "type": "string", + "format": "did", + "description": "Restrict to pulls authored by this user DID." + }, + "status": { + "type": "string", + "knownValues": ["open", "closed", "merged"], + "description": "Restrict to pulls whose latest derived status matches." } } }, -- 2.51.2