alice.pds.demo.boltless.dev
This repository has no description
71 kB
2132 lines
1use std::collections::{BTreeSet, HashMap};
2use std::marker::PhantomData;
3use std::num::NonZeroU32;
4use std::ops::{Bound, ControlFlow};
5use std::sync::atomic::{AtomicU32, Ordering};
6use std::sync::{Arc, Mutex, RwLock};
7
8const FILTER_SCAN_MULTIPLIER: usize = 64;
9const FILTER_SCAN_FLOOR: usize = 512;
10
11struct ScanState {
12 matched: Vec<(BucketKey, AtUri<DefaultStr>)>,
13 last_scanned: Option<BucketKey>,
14 scanned: usize,
15}
16
17impl ScanState {
18 fn with_capacity(cap: usize) -> Self {
19 Self {
20 matched: Vec::with_capacity(cap),
21 last_scanned: None,
22 scanned: 0,
23 }
24 }
25}
26
27use bobbin_runtime::RuntimeHasher;
28use bobbin_types::edges::{Edge, Record};
29use bobbin_types::ids::EdgeKey;
30use either::Either;
31use jacquard_common::DefaultStr;
32use jacquard_common::types::did::Did;
33use jacquard_common::types::nsid::Nsid;
34use jacquard_common::types::string::AtUri;
35use lasso::{Key, Spur, ThreadedRodeo};
36use scc::HashMap as SccMap;
37use scc::hash_map::Entry;
38use smallvec::SmallVec;
39use thiserror::Error;
40
41#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]
42struct SortMicros(u64);
43
44#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]
45struct BucketKey {
46 micros: SortMicros,
47 source: SourceId,
48}
49
50impl BucketKey {
51 fn new(micros: u64, source: SourceId) -> Self {
52 Self {
53 micros: SortMicros(micros),
54 source,
55 }
56 }
57
58 fn token(self) -> PageToken {
59 PageToken::new(self.micros.0, self.source.index())
60 }
61
62 fn from_token(tok: PageToken) -> Self {
63 Self {
64 micros: SortMicros(tok.micros),
65 source: SourceId::from_raw(tok.source),
66 }
67 }
68}
69
70pub mod coverage;
71pub mod state_index;
72pub use coverage::{Coverage, CoverageWatch, HydrantCursor, PromotionSignal};
73pub use state_index::{
74 ApplyOutcome, IssueStateKind, PullStatusKind, StateIndex, StateKind, apply_record_state,
75};
76
77#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
78pub struct SourceTag;
79#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
80struct AuthorTag;
81#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
82struct CollectionTag;
83
84#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
85pub struct Interned<T>(u32, PhantomData<T>);
86
87impl<T: Copy> Interned<T> {
88 fn from_spur(spur: Spur) -> Self {
89 Self(spur.into_usize() as u32, PhantomData)
90 }
91
92 fn from_raw(raw: u32) -> Self {
93 Self(raw, PhantomData)
94 }
95
96 fn index(self) -> u32 {
97 self.0
98 }
99
100 fn to_spur(self) -> Option<Spur> {
101 Spur::try_from_usize(self.0 as usize)
102 }
103}
104
105pub type SourceId = Interned<SourceTag>;
106type AuthorId = Interned<AuthorTag>;
107type CollectionId = Interned<CollectionTag>;
108
109#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
110pub struct PageToken {
111 micros: u64,
112 source: u32,
113}
114
115impl PageToken {
116 pub fn new(micros: u64, source: u32) -> Self {
117 Self { micros, source }
118 }
119
120 pub fn micros(self) -> u64 {
121 self.micros
122 }
123
124 pub fn source(self) -> u32 {
125 self.source
126 }
127
128 pub fn encode_token(self) -> String {
129 let mut bytes = [0u8; 12];
130 bytes[..8].copy_from_slice(&self.micros.to_be_bytes());
131 bytes[8..].copy_from_slice(&self.source.to_be_bytes());
132 encode_hex(&bytes)
133 }
134
135 pub fn decode_token(token: &str) -> Result<Self, CursorParseError> {
136 let bytes: [u8; 12] = decode_hex_array(token).ok_or(CursorParseError::Malformed)?;
137 let micros = u64::from_be_bytes(bytes[..8].try_into().unwrap());
138 let source = u32::from_be_bytes(bytes[8..].try_into().unwrap());
139 Ok(Self { micros, source })
140 }
141}
142
143fn encode_hex(bytes: &[u8]) -> String {
144 bytes
145 .iter()
146 .fold(String::with_capacity(bytes.len() * 2), |mut acc, b| {
147 acc.push(char::from_digit((b >> 4) as u32, 16).unwrap());
148 acc.push(char::from_digit((b & 0x0f) as u32, 16).unwrap());
149 acc
150 })
151}
152
153fn decode_hex_array<const N: usize>(token: &str) -> Option<[u8; N]> {
154 if token.len() != N * 2 {
155 return None;
156 }
157 let parsed: Vec<u8> = token
158 .as_bytes()
159 .chunks_exact(2)
160 .map(|pair| {
161 let hi = (pair[0] as char).to_digit(16)?;
162 let lo = (pair[1] as char).to_digit(16)?;
163 Some(((hi << 4) | lo) as u8)
164 })
165 .collect::<Option<Vec<u8>>>()?;
166 parsed.try_into().ok()
167}
168
169#[derive(Clone, Copy, Debug, Eq, PartialEq)]
170pub enum PageCursor {
171 Start,
172 After(PageToken),
173}
174
175#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
176pub enum SortDir {
177 Asc,
178 #[default]
179 Desc,
180}
181
182#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]
183pub enum CursorParseError {
184 #[error("cursor token must be a valid TID")]
185 Malformed,
186}
187
188impl PageCursor {
189 pub fn from_token(raw: Option<&str>) -> Result<Self, CursorParseError> {
190 raw.map_or(Ok(Self::Start), |t| {
191 PageToken::decode_token(t).map(Self::After)
192 })
193 }
194}
195
196#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
197pub struct PageLimit(u32);
198
199#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]
200pub enum PageLimitError {
201 #[error("page limit {value} below minimum {min}")]
202 TooSmall { value: u32, min: u32 },
203 #[error("page limit {value} above maximum {max}")]
204 TooLarge { value: u32, max: u32 },
205}
206
207impl PageLimit {
208 pub const MIN: u32 = 1;
209 pub const MAX: u32 = 1000;
210
211 pub fn new(value: u32) -> Result<Self, PageLimitError> {
212 match value {
213 v if v < Self::MIN => Err(PageLimitError::TooSmall {
214 value: v,
215 min: Self::MIN,
216 }),
217 v if v > Self::MAX => Err(PageLimitError::TooLarge {
218 value: v,
219 max: Self::MAX,
220 }),
221 v => Ok(Self(v)),
222 }
223 }
224
225 pub const fn get(self) -> u32 {
226 self.0
227 }
228}
229
230#[derive(Debug)]
231pub struct EdgePage {
232 pub items: Vec<EdgeItem>,
233 pub next: Option<PageToken>,
234}
235
236#[derive(Clone, Debug, Eq, PartialEq)]
237pub struct EdgeItem {
238 pub uri: AtUri<DefaultStr>,
239 pub sort_micros: u64,
240}
241
242impl AsRef<str> for EdgeItem {
243 fn as_ref(&self) -> &str {
244 self.uri.as_ref()
245 }
246}
247
248#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
249pub struct FilteredCount {
250 pub count: Count,
251 pub distinct_authors: DistinctAuthorCount,
252}
253
254#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
255pub struct Count(u64);
256
257impl 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)]
268pub struct DistinctAuthorCount(u64);
269
270impl 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
280impl 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
289#[derive(Clone, Copy, Debug, Default)]
290pub struct EdgeMemReport {
291 pub key_count: u64,
292 pub source_count: u64,
293 pub source_interner_bytes: u64,
294 pub did_interner_bytes: u64,
295 pub collection_interner_bytes: u64,
296 pub key_interner_bytes: u64,
297 pub edges_total: u64,
298 pub author_refs_total: u64,
299 pub reverse_entries: u64,
300 pub reverse_cap: u64,
301 pub forward_struct_bytes: u64,
302 pub reverse_struct_bytes: u64,
303 pub bucket_struct_bytes: u64,
304 pub max_bucket: u64,
305 pub bucket_size_classes: [u64; BUCKET_CLASS_COUNT],
306}
307
308const BUCKET_CLASS_BOUNDS: [u64; 16] = [
309 1,
310 2,
311 4,
312 8,
313 16,
314 32,
315 64,
316 128,
317 256,
318 512,
319 1024,
320 2048,
321 8192,
322 32768,
323 131072,
324 u64::MAX,
325];
326const BUCKET_CLASS_COUNT: usize = BUCKET_CLASS_BOUNDS.len();
327
328fn bucket_class(n: u64) -> usize {
329 BUCKET_CLASS_BOUNDS
330 .iter()
331 .position(|&bound| n <= bound)
332 .unwrap_or(BUCKET_CLASS_COUNT - 1)
333}
334
335impl EdgeMemReport {
336 pub fn bucket_histogram(&self) -> impl Iterator<Item = (u64, u64)> {
337 BUCKET_CLASS_BOUNDS
338 .into_iter()
339 .zip(self.bucket_size_classes)
340 }
341}
342
343#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
344struct EdgeKeyId(u32);
345
346fn bump_author(authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>, author: AuthorId) {
347 authors
348 .entry(author)
349 .and_modify(|c| *c = c.saturating_add(1))
350 .or_insert(NonZeroU32::MIN);
351}
352
353fn drop_author(authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>, author: AuthorId) {
354 match authors.get(&author).map(|c| c.get() - 1) {
355 Some(0) | None => {
356 authors.remove(&author);
357 }
358 Some(next) => {
359 authors.insert(author, NonZeroU32::new(next).unwrap());
360 }
361 }
362}
363
364struct LargeBucket {
365 keys: BTreeSet<BucketKey>,
366 authors: HashMap<AuthorId, NonZeroU32, RuntimeHasher>,
367}
368
369const BUCKET_PROMOTE_AT: usize = 256;
370const SOURCE_BYTES: u64 = 16;
371
372enum Sources {
373 Small(SmallVec<[BucketKey; 2]>),
374 Large(Box<LargeBucket>),
375}
376
377impl Default for Sources {
378 fn default() -> Self {
379 Self::Small(SmallVec::new())
380 }
381}
382
383impl Sources {
384 fn len(&self) -> usize {
385 match self {
386 Self::Small(v) => v.len(),
387 Self::Large(big) => big.keys.len(),
388 }
389 }
390
391 fn is_empty(&self) -> bool {
392 self.len() == 0
393 }
394
395 fn insert(&mut self, key: BucketKey, author: Option<AuthorId>) -> bool {
396 match self {
397 Self::Small(v) => match v.binary_search(&key) {
398 Ok(_) => false,
399 Err(pos) => {
400 v.insert(pos, key);
401 true
402 }
403 },
404 Self::Large(big) => {
405 let inserted = big.keys.insert(key);
406 if inserted && let Some(a) = author {
407 bump_author(&mut big.authors, a);
408 }
409 inserted
410 }
411 }
412 }
413
414 fn remove(&mut self, key: &BucketKey, author: Option<AuthorId>) {
415 match self {
416 Self::Small(v) => {
417 if let Ok(pos) = v.binary_search(key) {
418 v.remove(pos);
419 }
420 }
421 Self::Large(big) => {
422 if big.keys.remove(key)
423 && let Some(a) = author
424 {
425 drop_author(&mut big.authors, a);
426 }
427 }
428 }
429 }
430
431 fn directed(
432 &self,
433 cursor: PageCursor,
434 dir: SortDir,
435 ) -> Box<dyn Iterator<Item = BucketKey> + '_> {
436 match self {
437 Self::Small(v) => Box::new(directed_slice(v, cursor, dir)),
438 Self::Large(big) => Box::new(directed_tree(&big.keys, cursor, dir)),
439 }
440 }
441
442 fn heap_bytes(&self) -> u64 {
443 const BTREE_BYTES_PER_KEY: u64 = 32;
444 const HASHMAP_FIXED: u64 = 48;
445 const HASHMAP_PER_CAP: u64 = 9;
446 match self {
447 Self::Small(v) if v.spilled() => v.capacity() as u64 * SOURCE_BYTES,
448 Self::Small(_) => 0,
449 Self::Large(big) => {
450 std::mem::size_of::<LargeBucket>() as u64
451 + big.keys.len() as u64 * BTREE_BYTES_PER_KEY
452 + HASHMAP_FIXED
453 + big.authors.capacity() as u64 * HASHMAP_PER_CAP
454 }
455 }
456 }
457}
458
459#[derive(Clone, Copy, Debug, Eq, PartialEq)]
460struct ReverseEntry {
461 key_id: EdgeKeyId,
462 sort_micros: u64,
463}
464
465#[derive(Clone, Copy)]
466struct ProjectedState<K> {
467 key: EdgeKeyId,
468 kind: K,
469 author: Option<AuthorId>,
470}
471
472struct CountBucket {
473 count: u64,
474 authors: HashMap<AuthorId, NonZeroU32, RuntimeHasher>,
475}
476
477struct StateCountInner<K> {
478 projected: HashMap<SourceId, SmallVec<[ProjectedState<K>; 1]>, RuntimeHasher>,
479 buckets: HashMap<(EdgeKeyId, K), CountBucket, RuntimeHasher>,
480}
481
482struct StateCountIndex<K> {
483 inner: RwLock<StateCountInner<K>>,
484 hasher: RuntimeHasher,
485}
486
487impl<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
527pub struct EdgeStore {
528 source_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
529 did_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
530 collection_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
531 key_ids: SccMap<EdgeKey, EdgeKeyId, RuntimeHasher>,
532 keys: SccMap<EdgeKeyId, EdgeKey, RuntimeHasher>,
533 next_key_id: AtomicU32,
534 forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>,
535 reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>,
536 issue_counts: StateCountIndex<IssueStateKind>,
537 pull_counts: StateCountIndex<PullStatusKind>,
538 hasher: RuntimeHasher,
539 writer: Mutex<()>,
540}
541
542impl EdgeStore {
543 pub fn new(hasher: RuntimeHasher) -> Self {
544 Self {
545 source_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())),
546 did_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())),
547 collection_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())),
548 key_ids: SccMap::with_hasher(hasher.clone()),
549 keys: SccMap::with_hasher(hasher.clone()),
550 next_key_id: AtomicU32::new(0),
551 forward: SccMap::with_hasher(hasher.clone()),
552 reverse: SccMap::with_hasher(hasher.clone()),
553 issue_counts: StateCountIndex::new(hasher.clone()),
554 pull_counts: StateCountIndex::new(hasher.clone()),
555 hasher,
556 writer: Mutex::new(()),
557 }
558 }
559
560 fn intern_key(&self, key: EdgeKey) -> EdgeKeyId {
561 match self.key_ids.entry_sync(key.clone()) {
562 Entry::Occupied(e) => *e.get(),
563 Entry::Vacant(e) => {
564 let id = EdgeKeyId(self.next_key_id.fetch_add(1, Ordering::Relaxed));
565 e.insert_entry(id);
566 let _ = self.keys.insert_sync(id, key);
567 id
568 }
569 }
570 }
571
572 fn lookup_key(&self, key: &EdgeKey) -> Option<EdgeKeyId> {
573 self.key_ids.read_sync(key, |_, id| *id)
574 }
575
576 pub fn add(&self, edge: Edge) {
577 let _w = self
578 .writer
579 .lock()
580 .expect("edge-store writer mutex poisoned");
581 self.add_locked(edge);
582 }
583
584 pub fn intern_source(&self, source: &AtUri<DefaultStr>) -> SourceId {
585 let author = self.intern_author(source);
586 SourceId::from_spur(
587 self.source_interner
588 .get_or_intern(self.source_key(source, author)),
589 )
590 }
591
592 pub fn upsert_source(&self, source: &AtUri<DefaultStr>, edges: Vec<Edge>) {
593 let _w = self
594 .writer
595 .lock()
596 .expect("edge-store writer mutex poisoned");
597 self.clear_source_locked(source);
598 edges.into_iter().for_each(|e| self.add_locked(e));
599 }
600
601 pub fn remove_source(&self, source: &AtUri<DefaultStr>) {
602 let _w = self
603 .writer
604 .lock()
605 .expect("edge-store writer mutex poisoned");
606 self.clear_source_locked(source);
607 }
608
609 fn add_locked(&self, edge: Edge) {
610 let author = self.intern_author(&edge.source);
611 let source_key = self.source_key(&edge.source, author);
612 let id = SourceId::from_spur(self.source_interner.get_or_intern(source_key));
613 let sort_micros = edge.sort_micros;
614 let key_id = self.intern_key(EdgeKey::new(edge.kind, edge.subject));
615
616 let key = BucketKey::new(sort_micros, id);
617 let mut entry = self.forward.entry_sync(key_id).or_default();
618 let inserted = entry.get_mut().insert(key, author);
619 let promote_keys = match entry.get() {
620 Sources::Small(v) if v.len() > BUCKET_PROMOTE_AT => {
621 Some(v.iter().copied().collect::<Vec<_>>())
622 }
623 _ => None,
624 };
625 drop(entry);
626
627 if let Some(keys) = promote_keys {
628 let authors = self.build_author_map(&keys);
629 let large = LargeBucket {
630 keys: keys.into_iter().collect(),
631 authors,
632 };
633 if let Entry::Occupied(mut e) = self.forward.entry_sync(key_id) {
634 *e.get_mut() = Sources::Large(Box::new(large));
635 }
636 }
637
638 if inserted {
639 let mut rev = self.reverse.entry_sync(id).or_default();
640 rev.get_mut().push(ReverseEntry {
641 key_id,
642 sort_micros,
643 });
644 }
645 }
646
647 fn clear_source_locked(&self, source: &AtUri<DefaultStr>) {
648 let author = source_authority_did(source)
649 .and_then(|s| self.did_interner.get(s))
650 .map(AuthorId::from_spur);
651 let Some(source_spur) = self.source_interner.get(self.source_key(source, author)) else {
652 return;
653 };
654 let id = SourceId::from_spur(source_spur);
655 let Some((_, entries)) = self.reverse.remove_sync(&id) else {
656 return;
657 };
658 entries.into_iter().for_each(
659 |ReverseEntry {
660 key_id,
661 sort_micros,
662 }| {
663 self.forward.update_sync(&key_id, |_, sources| {
664 sources.remove(&BucketKey::new(sort_micros, id), author);
665 });
666 self.forward
667 .remove_if_sync(&key_id, |sources| sources.is_empty());
668 },
669 );
670 }
671
672 fn intern_author(&self, source: &AtUri<DefaultStr>) -> Option<AuthorId> {
673 let did = source_authority_did(source)?;
674 Some(AuthorId::from_spur(self.did_interner.get_or_intern(did)))
675 }
676
677 fn source_key(&self, source: &AtUri<DefaultStr>, author: Option<AuthorId>) -> String {
678 match (split_record_uri(source.as_ref()), author) {
679 (Some((_, collection, rkey)), Some(author)) => {
680 let collection =
681 CollectionId::from_spur(self.collection_interner.get_or_intern(collection));
682 format!("{}/{}/{}", author.index(), collection.index(), rkey)
683 }
684 _ => source.as_ref().to_owned(),
685 }
686 }
687
688 fn decode_source(&self, stored: &str) -> Option<String> {
689 if stored.starts_with("at://") {
690 return Some(stored.to_owned());
691 }
692 let mut parts = stored.splitn(3, '/');
693 let author = AuthorId::from_raw(parts.next()?.parse().ok()?);
694 let collection = CollectionId::from_raw(parts.next()?.parse().ok()?);
695 let rkey = parts.next()?;
696 let did = self.did_interner.try_resolve(&author.to_spur()?)?;
697 let collection = self
698 .collection_interner
699 .try_resolve(&collection.to_spur()?)?;
700 Some(format!("at://{did}/{collection}/{rkey}"))
701 }
702
703 fn author_of_stored(&self, stored: &str) -> Option<AuthorId> {
704 match stored.strip_prefix("at://") {
705 Some(rest) => {
706 let authority = rest.split('/').next().unwrap_or(rest);
707 authority
708 .starts_with("did:")
709 .then(|| self.did_interner.get(authority).map(AuthorId::from_spur))
710 .flatten()
711 }
712 None => stored
713 .split('/')
714 .next()?
715 .parse::<u32>()
716 .ok()
717 .map(AuthorId::from_raw),
718 }
719 }
720
721 fn author_of(&self, source: SourceId) -> Option<AuthorId> {
722 let spur = source.to_spur()?;
723 let stored = self.source_interner.try_resolve(&spur)?;
724 self.author_of_stored(stored)
725 }
726
727 fn distinct_authors_small(&self, keys: &[BucketKey]) -> u64 {
728 keys.iter()
729 .filter_map(|key| self.author_of(key.source))
730 .collect::<std::collections::HashSet<AuthorId>>()
731 .len() as u64
732 }
733
734 fn build_author_map(&self, keys: &[BucketKey]) -> HashMap<AuthorId, NonZeroU32, RuntimeHasher> {
735 keys.iter().fold(
736 HashMap::with_hasher(self.hasher.clone()),
737 |mut authors, key| {
738 if let Some(a) = self.author_of(key.source) {
739 bump_author(&mut authors, a);
740 }
741 authors
742 },
743 )
744 }
745
746 pub fn count(&self, key: &EdgeKey) -> u64 {
747 self.lookup_key(key)
748 .and_then(|id| {
749 self.forward
750 .read_sync(&id, |_, sources| sources.len() as u64)
751 })
752 .unwrap_or(0)
753 }
754
755 pub fn count_distinct_authors(&self, key: &EdgeKey) -> u64 {
756 self.lookup_key(key)
757 .and_then(|id| {
758 self.forward.read_sync(&id, |_, sources| match sources {
759 Sources::Large(big) => big.authors.len() as u64,
760 Sources::Small(v) => self.distinct_authors_small(v),
761 })
762 })
763 .unwrap_or(0)
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
956 /// answers "did the viewer star/follow/etc. this subject, and with what rkey"
957 pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option<AtUri<DefaultStr>> {
958 let author_spur = self.did_interner.get(viewer)?;
959 let author_id = AuthorId::from_spur(author_spur);
960 let key_id = self.lookup_key(key)?;
961
962 self.forward
963 .read_sync(&key_id, |_, sources| {
964 // large buckets track authors, so a missing author fast-fails the scan
965 if let Sources::Large(big) = sources
966 && !big.authors.contains_key(&author_id)
967 {
968 return None;
969 }
970 sources
971 .directed(PageCursor::Start, SortDir::Desc)
972 .find_map(|bucket| {
973 let spur = bucket.source.to_spur()?;
974 let stored = self.source_interner.try_resolve(&spur)?;
975 let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?;
976 (source_authority_did(&uri) == Some(viewer)).then_some(uri)
977 })
978 })
979 .flatten()
980 }
981
982 pub fn sources_for(&self, key: &EdgeKey) -> Vec<AtUri<DefaultStr>> {
983 self.lookup_key(key)
984 .and_then(|id| {
985 self.forward.read_sync(&id, |_, sources| {
986 sources
987 .directed(PageCursor::Start, SortDir::Desc)
988 .filter_map(|bucket| {
989 let spur = bucket.source.to_spur()?;
990 let stored = self.source_interner.try_resolve(&spur)?;
991 AtUri::new_owned(self.decode_source(stored)?).ok()
992 })
993 .collect::<Vec<_>>()
994 })
995 })
996 .unwrap_or_default()
997 }
998
999 pub fn list(
1000 &self,
1001 key: &EdgeKey,
1002 cursor: PageCursor,
1003 limit: PageLimit,
1004 dir: SortDir,
1005 ) -> EdgePage {
1006 let limit_usize = limit.get() as usize;
1007 self.lookup_key(key)
1008 .and_then(|id| {
1009 self.forward.read_sync(&id, |_, sources| {
1010 let iter = sources.directed(cursor, dir);
1011 let entries: Vec<BucketKey> = iter.take(limit_usize + 1).collect();
1012 let has_more = entries.len() > limit_usize;
1013 let page = &entries[..entries.len().min(limit_usize)];
1014 let items = page
1015 .iter()
1016 .filter_map(|&key| {
1017 let spur = key.source.to_spur()?;
1018 let stored = self.source_interner.try_resolve(&spur)?;
1019 let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?;
1020 Some(EdgeItem {
1021 uri,
1022 sort_micros: key.micros.0,
1023 })
1024 })
1025 .collect();
1026 let next = has_more
1027 .then(|| page.last().copied())
1028 .flatten()
1029 .map(BucketKey::token);
1030 EdgePage { items, next }
1031 })
1032 })
1033 .unwrap_or(EdgePage {
1034 items: Vec::new(),
1035 next: None,
1036 })
1037 }
1038
1039 pub fn list_filtered<F>(
1040 &self,
1041 key: &EdgeKey,
1042 cursor: PageCursor,
1043 limit: PageLimit,
1044 dir: SortDir,
1045 predicate: F,
1046 ) -> EdgePage
1047 where
1048 F: Fn(&AtUri<DefaultStr>) -> bool,
1049 {
1050 let limit_usize = limit.get() as usize;
1051 let scan_cap = limit_usize
1052 .saturating_mul(FILTER_SCAN_MULTIPLIER)
1053 .max(FILTER_SCAN_FLOOR);
1054 self.lookup_key(key)
1055 .and_then(|id| {
1056 self.forward.read_sync(&id, |_, sources| {
1057 let init = ScanState::with_capacity(limit_usize + 1);
1058 let outcome = sources
1059 .directed(cursor, dir)
1060 .try_fold(init, |mut state, key| {
1061 if state.scanned >= scan_cap && state.matched.len() <= limit_usize {
1062 return ControlFlow::Break(state);
1063 }
1064 state.scanned += 1;
1065 state.last_scanned = Some(key);
1066 if let Some(spur) = key.source.to_spur()
1067 && let Some(stored) = self.source_interner.try_resolve(&spur)
1068 && let Some(decoded) = self.decode_source(stored)
1069 && let Ok(uri) = AtUri::new_owned(decoded)
1070 && predicate(&uri)
1071 {
1072 state.matched.push((key, uri));
1073 if state.matched.len() > limit_usize {
1074 return ControlFlow::Break(state);
1075 }
1076 }
1077 ControlFlow::Continue(state)
1078 });
1079 let (state, bucket_exhausted) = match outcome {
1080 ControlFlow::Continue(s) => (s, true),
1081 ControlFlow::Break(s) => (s, false),
1082 };
1083 let has_more_matches = state.matched.len() > limit_usize;
1084 let visible_len = state.matched.len().min(limit_usize);
1085 let next = if has_more_matches && visible_len > 0 {
1086 Some(state.matched[visible_len - 1].0.token())
1087 } else if !bucket_exhausted {
1088 state.last_scanned.map(BucketKey::token)
1089 } else {
1090 None
1091 };
1092 let items = state
1093 .matched
1094 .into_iter()
1095 .take(visible_len)
1096 .map(|(key, uri)| EdgeItem {
1097 uri,
1098 sort_micros: key.micros.0,
1099 })
1100 .collect();
1101 EdgePage { items, next }
1102 })
1103 })
1104 .unwrap_or(EdgePage {
1105 items: Vec::new(),
1106 next: None,
1107 })
1108 }
1109
1110 pub fn key_count(&self) -> usize {
1111 self.forward.len()
1112 }
1113
1114 pub fn source_count(&self) -> usize {
1115 self.reverse.len()
1116 }
1117
1118 pub fn mem_report(&self) -> EdgeMemReport {
1119 const SCC_SLOT: u64 = 32;
1120 let edge_key = std::mem::size_of::<EdgeKey>() as u64;
1121 let rev_entry = std::mem::size_of::<ReverseEntry>() as u64;
1122 let rev_smallvec = std::mem::size_of::<SmallVec<[ReverseEntry; 1]>>() as u64;
1123 let bucket_struct = std::mem::size_of::<Sources>() as u64;
1124
1125 let mut edges_total = 0u64;
1126 let mut author_refs_total = 0u64;
1127 let mut forward_struct_bytes =
1128 self.issue_counts.heap_bytes() + self.pull_counts.heap_bytes();
1129 let mut max_bucket = 0u64;
1130 let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT];
1131 self.forward.iter_sync(|_, sources| {
1132 let bucket_len = sources.len() as u64;
1133 edges_total += bucket_len;
1134 max_bucket = max_bucket.max(bucket_len);
1135 bucket_size_classes[bucket_class(bucket_len)] += 1;
1136 if let Sources::Large(big) = sources {
1137 author_refs_total += big.authors.len() as u64;
1138 }
1139 forward_struct_bytes += SCC_SLOT + bucket_struct + sources.heap_bytes();
1140 true
1141 });
1142 let key_interner_bytes = self.key_ids.len() as u64
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);
1146
1147 let mut reverse_entries = 0u64;
1148 let mut reverse_cap = 0u64;
1149 let mut reverse_struct_bytes = 0u64;
1150 self.reverse.iter_sync(|_, refs| {
1151 let cap = refs.capacity() as u64;
1152 reverse_entries += refs.len() as u64;
1153 reverse_cap += cap;
1154 let heap = if refs.spilled() { cap * rev_entry } else { 0 };
1155 reverse_struct_bytes += SCC_SLOT + rev_smallvec + heap;
1156 true
1157 });
1158
1159 EdgeMemReport {
1160 key_count: self.forward.len() as u64,
1161 source_count: self.reverse.len() as u64,
1162 source_interner_bytes: self.source_interner.current_memory_usage() as u64,
1163 did_interner_bytes: self.did_interner.current_memory_usage() as u64,
1164 collection_interner_bytes: self.collection_interner.current_memory_usage() as u64,
1165 key_interner_bytes,
1166 edges_total,
1167 author_refs_total,
1168 reverse_entries,
1169 reverse_cap,
1170 forward_struct_bytes,
1171 reverse_struct_bytes,
1172 bucket_struct_bytes: bucket_struct,
1173 max_bucket,
1174 bucket_size_classes,
1175 }
1176 }
1177}
1178
1179pub 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
1219pub 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
1250fn directed_slice(
1251 sources: &[BucketKey],
1252 cursor: PageCursor,
1253 dir: SortDir,
1254) -> impl Iterator<Item = BucketKey> + '_ {
1255 match dir {
1256 SortDir::Asc => {
1257 let start = match cursor {
1258 PageCursor::Start => 0,
1259 PageCursor::After(tok) => {
1260 sources.partition_point(|k| *k <= BucketKey::from_token(tok))
1261 }
1262 };
1263 Either::Left(sources[start..].iter().copied())
1264 }
1265 SortDir::Desc => {
1266 let end = match cursor {
1267 PageCursor::Start => sources.len(),
1268 PageCursor::After(tok) => {
1269 sources.partition_point(|k| *k < BucketKey::from_token(tok))
1270 }
1271 };
1272 Either::Right(sources[..end].iter().rev().copied())
1273 }
1274 }
1275}
1276
1277fn directed_tree(
1278 keys: &BTreeSet<BucketKey>,
1279 cursor: PageCursor,
1280 dir: SortDir,
1281) -> impl Iterator<Item = BucketKey> + '_ {
1282 match dir {
1283 SortDir::Asc => {
1284 let lower = match cursor {
1285 PageCursor::Start => Bound::Unbounded,
1286 PageCursor::After(tok) => Bound::Excluded(BucketKey::from_token(tok)),
1287 };
1288 Either::Left(keys.range((lower, Bound::Unbounded)).copied())
1289 }
1290 SortDir::Desc => {
1291 let upper = match cursor {
1292 PageCursor::Start => Bound::Unbounded,
1293 PageCursor::After(tok) => Bound::Excluded(BucketKey::from_token(tok)),
1294 };
1295 Either::Right(keys.range((Bound::Unbounded, upper)).rev().copied())
1296 }
1297 }
1298}
1299
1300fn split_record_uri(source: &str) -> Option<(&str, &str, &str)> {
1301 let rest = source.strip_prefix("at://")?;
1302 let mut parts = rest.split('/');
1303 let authority = parts.next()?;
1304 let collection = parts.next()?;
1305 let rkey = parts.next()?;
1306 if parts.next().is_some() {
1307 return None;
1308 }
1309 (authority.starts_with("did:") && !collection.is_empty() && !rkey.is_empty())
1310 .then_some((authority, collection, rkey))
1311}
1312
1313fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<&str> {
1314 let rest = source.as_ref().strip_prefix("at://")?;
1315 let end = rest.find('/').unwrap_or(rest.len());
1316 let candidate = &rest[..end];
1317 candidate.starts_with("did:").then_some(candidate)
1318}
1319
1320#[cfg(test)]
1321mod tests {
1322 use super::*;
1323 use bobbin_types::ids::SubjectRef;
1324 use jacquard_common::types::did::Did;
1325 use jacquard_common::types::nsid::Nsid;
1326 use jacquard_common::types::string::AtUri;
1327
1328 fn store() -> EdgeStore {
1329 EdgeStore::new(RuntimeHasher::default())
1330 }
1331
1332 fn nsid(s: &'static str) -> Nsid<DefaultStr> {
1333 Nsid::new_static(s).unwrap()
1334 }
1335
1336 fn at(s: &str) -> AtUri<DefaultStr> {
1337 AtUri::new_owned(s).unwrap()
1338 }
1339
1340 fn did(s: &str) -> Did<DefaultStr> {
1341 Did::new_owned(s).unwrap()
1342 }
1343
1344 fn did_subj(s: &str) -> SubjectRef {
1345 SubjectRef::Did(did(s))
1346 }
1347
1348 fn limit(n: u32) -> PageLimit {
1349 PageLimit::new(n).unwrap()
1350 }
1351
1352 const NAMES: [&str; 5] = ["nel", "olaren", "teq", "lyna", "bailey"];
1353
1354 fn star_edge(source: AtUri<DefaultStr>, subject: Did<DefaultStr>) -> Edge {
1355 star_edge_at(source, subject, 0)
1356 }
1357
1358 fn star_edge_at(source: AtUri<DefaultStr>, subject: Did<DefaultStr>, sort_micros: u64) -> Edge {
1359 Edge {
1360 kind: nsid("sh.tangled.feed.star"),
1361 subject: SubjectRef::Did(subject),
1362 source,
1363 sort_micros,
1364 }
1365 }
1366
1367 fn shuffled_micros(i: usize) -> u64 {
1368 (i as u64).wrapping_mul(2_654_435_761) % 64
1369 }
1370
1371 fn source_uri(i: usize) -> String {
1372 format!(
1373 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1374 NAMES[i % NAMES.len()]
1375 )
1376 }
1377
1378 fn fill_subject(store: &EdgeStore, n: usize) -> EdgeKey {
1379 let subject = did("did:plc:squid");
1380 (0..n).for_each(|i| {
1381 store.add(star_edge_at(
1382 at(&source_uri(i)),
1383 subject.clone(),
1384 shuffled_micros(i),
1385 ));
1386 });
1387 EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:squid"))
1388 }
1389
1390 fn reference(kept: impl Iterator<Item = usize>) -> Vec<String> {
1391 let mut rows: Vec<(u64, usize)> = kept.map(|i| (shuffled_micros(i), i)).collect();
1392 rows.sort_unstable();
1393 rows.into_iter().map(|(_, i)| source_uri(i)).collect()
1394 }
1395
1396 fn paginate_all(store: &EdgeStore, key: &EdgeKey, dir: SortDir, page: u32) -> Vec<String> {
1397 std::iter::successors(
1398 Some(store.list(key, PageCursor::Start, limit(page), dir)),
1399 |prev| {
1400 prev.next
1401 .map(|tok| store.list(key, PageCursor::After(tok), limit(page), dir))
1402 },
1403 )
1404 .flat_map(|p| {
1405 p.items
1406 .into_iter()
1407 .map(|u| u.as_ref().to_owned())
1408 .collect::<Vec<_>>()
1409 })
1410 .collect()
1411 }
1412
1413 #[test]
1414 fn pagination_matches_reference_across_small_and_large() {
1415 [64usize, 1000].into_iter().for_each(|n| {
1416 let store = store();
1417 let key = fill_subject(&store, n);
1418 assert_eq!(store.count(&key), n as u64, "count mismatch at n={n}");
1419 assert_eq!(
1420 store.count_distinct_authors(&key),
1421 NAMES.len() as u64,
1422 "distinct authors at n={n}"
1423 );
1424
1425 let asc = reference(0..n);
1426 let desc: Vec<String> = asc.iter().rev().cloned().collect();
1427
1428 [3u32, 7, 50].into_iter().for_each(|page| {
1429 assert_eq!(
1430 paginate_all(&store, &key, SortDir::Asc, page),
1431 asc,
1432 "asc mismatch n={n} page={page}"
1433 );
1434 assert_eq!(
1435 paginate_all(&store, &key, SortDir::Desc, page),
1436 desc,
1437 "desc mismatch n={n} page={page}"
1438 );
1439 });
1440 });
1441 }
1442
1443 #[test]
1444 fn remove_from_large_bucket_keeps_pagination_exact() {
1445 let store = store();
1446 let n = 1000usize;
1447 let key = fill_subject(&store, n);
1448 (0..n).step_by(3).for_each(|i| {
1449 store.remove_source(&at(&source_uri(i)));
1450 });
1451 let expected = reference((0..n).filter(|i| i % 3 != 0));
1452 assert_eq!(store.count(&key), expected.len() as u64);
1453 assert_eq!(
1454 store.count_distinct_authors(&key),
1455 NAMES.len() as u64,
1456 "every author keeps sources after partial removal"
1457 );
1458 assert_eq!(paginate_all(&store, &key, SortDir::Asc, 7), expected);
1459 assert_eq!(
1460 paginate_all(&store, &key, SortDir::Desc, 11),
1461 expected.iter().rev().cloned().collect::<Vec<_>>()
1462 );
1463 }
1464
1465 #[test]
1466 fn add_then_count() {
1467 let store = store();
1468 let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1469
1470 store.add(star_edge(
1471 at("at://did:plc:nel/sh.tangled.feed.star/r1"),
1472 did("did:plc:abalone"),
1473 ));
1474 store.add(star_edge(
1475 at("at://did:plc:olaren/sh.tangled.feed.star/r2"),
1476 did("did:plc:abalone"),
1477 ));
1478 store.add(star_edge(
1479 at("at://did:plc:nel/sh.tangled.feed.star/r3"),
1480 did("did:plc:abalone"),
1481 ));
1482
1483 assert_eq!(store.count(&key), 3);
1484 assert_eq!(store.count_distinct_authors(&key), 2);
1485 }
1486
1487 #[test]
1488 fn duplicate_add_is_idempotent() {
1489 let store = store();
1490 let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1491 let edge = star_edge(
1492 at("at://did:plc:nel/sh.tangled.feed.star/r1"),
1493 did("did:plc:abalone"),
1494 );
1495 store.add(edge.clone());
1496 store.add(edge);
1497 assert_eq!(store.count(&key), 1);
1498 assert_eq!(store.count_distinct_authors(&key), 1);
1499 }
1500
1501 #[test]
1502 fn remove_source_clears_all_keys_for_that_source() {
1503 let store = store();
1504 let star_key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1505 let follow_key = EdgeKey::new(nsid("sh.tangled.graph.follow"), did_subj("did:plc:lyna"));
1506 let source = "at://did:plc:nel/sh.tangled.feed.star/r1";
1507
1508 store.add(Edge {
1509 kind: nsid("sh.tangled.feed.star"),
1510 subject: did_subj("did:plc:abalone"),
1511 source: at(source),
1512 sort_micros: 0,
1513 });
1514 store.add(Edge {
1515 kind: nsid("sh.tangled.graph.follow"),
1516 subject: did_subj("did:plc:lyna"),
1517 source: at(source),
1518 sort_micros: 0,
1519 });
1520
1521 assert_eq!(store.count(&star_key), 1);
1522 assert_eq!(store.count(&follow_key), 1);
1523
1524 store.remove_source(&at(source));
1525 assert_eq!(store.count(&star_key), 0);
1526 assert_eq!(store.count(&follow_key), 0);
1527 }
1528
1529 #[test]
1530 fn upsert_source_replaces_old_edges() {
1531 let store = store();
1532 let source = at("at://did:plc:teq/sh.tangled.feed.star/r1");
1533 let old_subject = did_subj("did:plc:abalone");
1534 let new_subject = did_subj("did:plc:uni");
1535 let kind = nsid("sh.tangled.feed.star");
1536
1537 store.upsert_source(
1538 &source,
1539 vec![Edge {
1540 kind: kind.clone(),
1541 subject: old_subject.clone(),
1542 source: source.clone(),
1543 sort_micros: 0,
1544 }],
1545 );
1546 assert_eq!(
1547 store.count(&EdgeKey::new(kind.clone(), old_subject.clone())),
1548 1
1549 );
1550
1551 store.upsert_source(
1552 &source,
1553 vec![Edge {
1554 kind: kind.clone(),
1555 subject: new_subject.clone(),
1556 source: source.clone(),
1557 sort_micros: 0,
1558 }],
1559 );
1560 assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0);
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 );
1675 }
1676
1677 #[test]
1678 fn list_pages_in_sort_order() {
1679 let store = store();
1680 let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1681 (0..5).for_each(|i| {
1682 store.add(star_edge_at(
1683 at(&format!(
1684 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1685 NAMES[i]
1686 )),
1687 did("did:plc:abalone"),
1688 1_000_000 + i as u64 * 1_000_000,
1689 ));
1690 });
1691
1692 let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
1693 assert_eq!(page1.items.len(), 2);
1694 let cursor = PageCursor::After(page1.next.expect("has next"));
1695
1696 let page2 = store.list(&key, cursor, limit(2), SortDir::Asc);
1697 assert_eq!(page2.items.len(), 2);
1698
1699 let cursor2 = PageCursor::After(page2.next.expect("has next"));
1700 let page3 = store.list(&key, cursor2, limit(2), SortDir::Asc);
1701 assert_eq!(page3.items.len(), 1);
1702 assert!(
1703 page3.next.is_none(),
1704 "final partial page should signal exhaustion"
1705 );
1706 }
1707
1708 #[test]
1709 fn list_exact_fill_signals_exhaustion() {
1710 let store = store();
1711 let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1712 (0..2).for_each(|i| {
1713 store.add(star_edge(
1714 at(&format!(
1715 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1716 NAMES[i]
1717 )),
1718 did("did:plc:abalone"),
1719 ));
1720 });
1721
1722 let page = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
1723 assert_eq!(page.items.len(), 2);
1724 assert!(page.next.is_none(), "exact-fill page must not promise more");
1725 }
1726
1727 #[test]
1728 fn list_on_unknown_key_is_empty() {
1729 let store = store();
1730 let page = store.list(
1731 &EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:periwinkle")),
1732 PageCursor::Start,
1733 limit(10),
1734 SortDir::Asc,
1735 );
1736 assert!(page.items.is_empty());
1737 assert!(page.next.is_none());
1738 }
1739
1740 #[test]
1741 fn distinct_authors_decreases_when_last_source_from_author_removed() {
1742 let store = store();
1743 let key = EdgeKey::new(nsid("sh.tangled.feed.star"), did_subj("did:plc:abalone"));
1744 let s1 = at("at://did:plc:nel/sh.tangled.feed.star/r1");
1745 let s2 = at("at://did:plc:nel/sh.tangled.feed.star/r2");
1746 let s3 = at("at://did:plc:olaren/sh.tangled.feed.star/r3");
1747
1748 store.add(star_edge(s1.clone(), did("did:plc:abalone")));
1749 store.add(star_edge(s2.clone(), did("did:plc:abalone")));
1750 store.add(star_edge(s3, did("did:plc:abalone")));
1751 assert_eq!(store.count_distinct_authors(&key), 2);
1752
1753 store.remove_source(&s1);
1754 assert_eq!(store.count_distinct_authors(&key), 2, "user1 still has s2");
1755
1756 store.remove_source(&s2);
1757 assert_eq!(store.count_distinct_authors(&key), 1, "user1 fully gone");
1758 }
1759
1760 #[test]
1761 fn page_limit_rejects_zero_and_oversize() {
1762 assert!(matches!(
1763 PageLimit::new(0),
1764 Err(PageLimitError::TooSmall { value: 0, min: 1 })
1765 ));
1766 assert!(matches!(
1767 PageLimit::new(PageLimit::MAX + 1),
1768 Err(PageLimitError::TooLarge { .. })
1769 ));
1770 assert_eq!(PageLimit::new(50).unwrap().get(), 50);
1771 }
1772
1773 #[test]
1774 fn cursor_token_round_trip() {
1775 let original = PageToken::new(1_730_000_000_000_000, 0x1234_abcd);
1776 let token = original.encode_token();
1777 assert_eq!(token.len(), 24, "12-byte cursor encodes to 24 hex chars");
1778 assert_eq!(PageToken::decode_token(&token).unwrap(), original);
1779 }
1780
1781 #[test]
1782 fn from_token_none_yields_start() {
1783 assert_eq!(PageCursor::from_token(None).unwrap(), PageCursor::Start);
1784 }
1785
1786 #[test]
1787 fn from_token_some_yields_after() {
1788 let original = PageToken::new(1_730_000_000_000_000, 42);
1789 let token = original.encode_token();
1790 assert_eq!(
1791 PageCursor::from_token(Some(&token)).unwrap(),
1792 PageCursor::After(original),
1793 );
1794 }
1795
1796 #[test]
1797 fn cursor_decode_rejects_malformed() {
1798 let bad = [
1799 "",
1800 "deadbeef",
1801 "no-hex!!aaaaaaaaaaaaaaaa",
1802 "12345",
1803 "this-string-is-way-too-long-to-be-a-valid-cursor",
1804 ];
1805 bad.into_iter().for_each(|s| {
1806 assert!(
1807 matches!(PageToken::decode_token(s), Err(CursorParseError::Malformed)),
1808 "expected malformed for {s:?}",
1809 );
1810 });
1811 }
1812
1813 #[test]
1814 fn cursor_token_is_hex_shape() {
1815 let token = PageToken::new(1_730_000_000_000_000, 7).encode_token();
1816 assert_eq!(token.len(), 24);
1817 assert!(token.chars().all(|c| c.is_ascii_hexdigit()));
1818 }
1819
1820 #[test]
1821 fn list_does_not_drop_entries_at_same_sort_micros_across_pages() {
1822 let store = store();
1823 let subject_did = did("did:plc:limpet");
1824 let key = EdgeKey::new(
1825 nsid("sh.tangled.feed.star"),
1826 SubjectRef::Did(subject_did.clone()),
1827 );
1828 (0..4).for_each(|i| {
1829 store.add(star_edge_at(
1830 at(&format!(
1831 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1832 NAMES[i]
1833 )),
1834 subject_did.clone(),
1835 42,
1836 ));
1837 });
1838
1839 let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
1840 assert_eq!(page1.items.len(), 2);
1841 let token = page1.next.expect("cursor must continue across ties");
1842
1843 let page2 = store.list(&key, PageCursor::After(token), limit(2), SortDir::Asc);
1844 assert_eq!(
1845 page2.items.len(),
1846 2,
1847 "remaining ties must surface on next page"
1848 );
1849 assert!(page2.next.is_none());
1850 let combined: std::collections::HashSet<String> = page1
1851 .items
1852 .iter()
1853 .chain(page2.items.iter())
1854 .map(|u| u.as_ref().to_owned())
1855 .collect();
1856 assert_eq!(combined.len(), 4, "every tied entry visible exactly once");
1857 }
1858
1859 #[test]
1860 fn list_filtered_narrows_by_predicate_and_paginates() {
1861 let store = store();
1862 let subject_did = did("did:plc:limpet");
1863 let key = EdgeKey::new(
1864 nsid("sh.tangled.repo.issue"),
1865 SubjectRef::Did(subject_did.clone()),
1866 );
1867 let by_nel = (0..3).map(|i| {
1868 star_edge_at(
1869 at(&format!("at://did:plc:nel/sh.tangled.repo.issue/r{i}")),
1870 subject_did.clone(),
1871 100 + i as u64,
1872 )
1873 });
1874 let by_olaren = (0..2).map(|i| {
1875 star_edge_at(
1876 at(&format!("at://did:plc:olaren/sh.tangled.repo.issue/o{i}")),
1877 subject_did.clone(),
1878 500 + i as u64,
1879 )
1880 });
1881 by_nel
1882 .chain(by_olaren)
1883 .map(|mut e| {
1884 e.kind = nsid("sh.tangled.repo.issue");
1885 e
1886 })
1887 .for_each(|e| store.add(e));
1888
1889 let only_nel = |u: &AtUri<DefaultStr>| u.as_ref().starts_with("at://did:plc:nel/");
1890 let page1 = store.list_filtered(&key, PageCursor::Start, limit(2), SortDir::Asc, only_nel);
1891 assert_eq!(page1.items.len(), 2);
1892 assert!(page1.next.is_some(), "cursor must allow more nel matches");
1893
1894 let page2 = store.list_filtered(
1895 &key,
1896 PageCursor::After(page1.next.unwrap()),
1897 limit(2),
1898 SortDir::Asc,
1899 only_nel,
1900 );
1901 assert_eq!(page2.items.len(), 1, "only one nel issue left");
1902 assert!(page2.next.is_none(), "tail page must not promise more");
1903 }
1904
1905 #[test]
1906 fn list_descending_returns_newest_first() {
1907 let store = store();
1908 let subject_did = did("did:plc:limpet");
1909 let key = EdgeKey::new(
1910 nsid("sh.tangled.feed.star"),
1911 SubjectRef::Did(subject_did.clone()),
1912 );
1913 (0..5).for_each(|i| {
1914 store.add(star_edge_at(
1915 at(&format!(
1916 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1917 NAMES[i]
1918 )),
1919 subject_did.clone(),
1920 1_000_000 + i as u64 * 1_000_000,
1921 ));
1922 });
1923
1924 let asc = store.list(&key, PageCursor::Start, limit(5), SortDir::Asc);
1925 let desc = store.list(&key, PageCursor::Start, limit(5), SortDir::Desc);
1926 assert_eq!(asc.items.len(), 5);
1927 assert_eq!(desc.items.len(), 5);
1928 let asc_uris: Vec<_> = asc.items.iter().map(|u| u.as_ref().to_owned()).collect();
1929 let mut reversed = asc_uris.clone();
1930 reversed.reverse();
1931 let desc_uris: Vec<_> = desc.items.iter().map(|u| u.as_ref().to_owned()).collect();
1932 assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc");
1933 }
1934
1935 #[test]
1936 fn list_descending_paginates_with_cursor() {
1937 let store = store();
1938 let subject_did = did("did:plc:whelk");
1939 let key = EdgeKey::new(
1940 nsid("sh.tangled.feed.star"),
1941 SubjectRef::Did(subject_did.clone()),
1942 );
1943 (0..5).for_each(|i| {
1944 store.add(star_edge_at(
1945 at(&format!(
1946 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
1947 NAMES[i]
1948 )),
1949 subject_did.clone(),
1950 1_000_000 + i as u64 * 1_000_000,
1951 ));
1952 });
1953
1954 let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Desc);
1955 assert_eq!(page1.items.len(), 2);
1956 let cursor = PageCursor::After(page1.next.expect("desc page1 must continue"));
1957 let page2 = store.list(&key, cursor, limit(2), SortDir::Desc);
1958 assert_eq!(page2.items.len(), 2);
1959 let cursor2 = PageCursor::After(page2.next.expect("desc page2 must continue"));
1960 let page3 = store.list(&key, cursor2, limit(2), SortDir::Desc);
1961 assert_eq!(page3.items.len(), 1);
1962 assert!(page3.next.is_none());
1963
1964 let combined: std::collections::HashSet<String> = page1
1965 .items
1966 .iter()
1967 .chain(page2.items.iter())
1968 .chain(page3.items.iter())
1969 .map(|u| u.as_ref().to_owned())
1970 .collect();
1971 assert_eq!(
1972 combined.len(),
1973 5,
1974 "desc pagination visits each item exactly once"
1975 );
1976 }
1977
1978 #[test]
1979 fn list_filtered_descending_respects_predicate() {
1980 let store = store();
1981 let subject_did = did("did:plc:scallop");
1982 let key = EdgeKey::new(
1983 nsid("sh.tangled.repo.issue"),
1984 SubjectRef::Did(subject_did.clone()),
1985 );
1986 let by_nel = (0..3).map(|i| {
1987 star_edge_at(
1988 at(&format!("at://did:plc:nel/sh.tangled.repo.issue/r{i}")),
1989 subject_did.clone(),
1990 100 + i as u64,
1991 )
1992 });
1993 let by_olaren = (0..2).map(|i| {
1994 star_edge_at(
1995 at(&format!("at://did:plc:olaren/sh.tangled.repo.issue/o{i}")),
1996 subject_did.clone(),
1997 500 + i as u64,
1998 )
1999 });
2000 by_nel
2001 .chain(by_olaren)
2002 .map(|mut e| {
2003 e.kind = nsid("sh.tangled.repo.issue");
2004 e
2005 })
2006 .for_each(|e| store.add(e));
2007
2008 let only_nel = |u: &AtUri<DefaultStr>| u.as_ref().starts_with("at://did:plc:nel/");
2009 let page = store.list_filtered(&key, PageCursor::Start, limit(5), SortDir::Desc, only_nel);
2010 assert_eq!(page.items.len(), 3, "all three nel issues visible");
2011 let last = page.items.last().unwrap().as_ref();
2012 let first = page.items.first().unwrap().as_ref();
2013 assert!(
2014 first > last,
2015 "desc order: first item rkey must be greater than last (got first={first}, last={last})",
2016 );
2017 }
2018
2019 #[test]
2020 fn non_did_source_round_trips_via_raw_fallback() {
2021 let store = store();
2022 let subject = did("did:plc:limpet");
2023 let key = EdgeKey::new(
2024 nsid("sh.tangled.feed.star"),
2025 SubjectRef::Did(subject.clone()),
2026 );
2027 let did_source = at("at://did:plc:nel/sh.tangled.feed.star/r1");
2028 let handle_source = at("at://witchcraft.systems/sh.tangled.feed.star/r2");
2029 store.add(star_edge_at(did_source.clone(), subject.clone(), 1));
2030 store.add(star_edge_at(handle_source.clone(), subject.clone(), 2));
2031
2032 let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc);
2033 let got: std::collections::HashSet<String> =
2034 page.items.iter().map(|u| u.as_ref().to_owned()).collect();
2035 assert!(
2036 got.contains(did_source.as_ref()),
2037 "did source must decode exactly"
2038 );
2039 assert!(
2040 got.contains(handle_source.as_ref()),
2041 "non-did authority must round-trip through the raw fallback"
2042 );
2043
2044 store.remove_source(&handle_source);
2045 assert_eq!(
2046 store.count(&key),
2047 1,
2048 "raw-keyed source removable by its uri"
2049 );
2050 }
2051
2052 #[test]
2053 fn distinct_collections_decode_with_their_own_collection() {
2054 let store = store();
2055 let subject = did("did:plc:limpet");
2056 let key = EdgeKey::new(
2057 nsid("sh.tangled.feed.star"),
2058 SubjectRef::Did(subject.clone()),
2059 );
2060 let star_src = at("at://did:plc:nel/sh.tangled.feed.star/aaa");
2061 let issue_src = at("at://did:plc:nel/sh.tangled.repo.issue/bbb");
2062 store.add(Edge {
2063 kind: nsid("sh.tangled.feed.star"),
2064 subject: SubjectRef::Did(subject.clone()),
2065 source: star_src.clone(),
2066 sort_micros: 1,
2067 });
2068 store.add(Edge {
2069 kind: nsid("sh.tangled.feed.star"),
2070 subject: SubjectRef::Did(subject.clone()),
2071 source: issue_src.clone(),
2072 sort_micros: 2,
2073 });
2074
2075 let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc);
2076 let got: std::collections::HashSet<String> =
2077 page.items.iter().map(|u| u.as_ref().to_owned()).collect();
2078 assert!(
2079 got.contains(star_src.as_ref()),
2080 "star-collection source decodes exactly"
2081 );
2082 assert!(
2083 got.contains(issue_src.as_ref()),
2084 "issue-collection source must keep its own collection, not borrow the star one"
2085 );
2086 }
2087
2088 #[test]
2089 fn author_refs_spill_preserves_distinct_count() {
2090 let store = store();
2091 let subject = did("did:plc:scallop");
2092 let key = EdgeKey::new(
2093 nsid("sh.tangled.feed.star"),
2094 SubjectRef::Did(subject.clone()),
2095 );
2096 (0..5).for_each(|i| {
2097 store.add(star_edge_at(
2098 at(&format!(
2099 "at://did:plc:{}/sh.tangled.feed.star/r{i}",
2100 NAMES[i]
2101 )),
2102 subject.clone(),
2103 i as u64,
2104 ));
2105 });
2106 assert_eq!(
2107 store.count_distinct_authors(&key),
2108 5,
2109 "five authors exceed the inline cap and spill to the map"
2110 );
2111
2112 store.add(star_edge_at(
2113 at("at://did:plc:nel/sh.tangled.feed.star/r99"),
2114 subject.clone(),
2115 99,
2116 ));
2117 assert_eq!(
2118 store.count_distinct_authors(&key),
2119 5,
2120 "second nel source adds no author"
2121 );
2122
2123 store.remove_source(&at("at://did:plc:nel/sh.tangled.feed.star/r0"));
2124 assert_eq!(
2125 store.count_distinct_authors(&key),
2126 5,
2127 "nel still present via r99"
2128 );
2129 store.remove_source(&at("at://did:plc:nel/sh.tangled.feed.star/r99"));
2130 assert_eq!(store.count_distinct_authors(&key), 4, "nel fully removed");
2131 }
2132}