use std::collections::{HashMap, HashSet};
use std::marker::PhantomData;
use std::num::NonZeroU32;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Mutex};

use bobbin_runtime::RuntimeHasher;
use bobbin_types::edges::{Edge, Record, TICKET_ISSUE_KIND, TICKET_PULL_KIND};
use bobbin_types::ids::{EdgeKey, SubjectRef};
use either::Either;
use itertools::Itertools;
use jacquard_common::DefaultStr;
use jacquard_common::deps::smol_str::SmolStr;
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;
use scc::hash_map::Entry;
use smallvec::SmallVec;
use thiserror::Error;

#[repr(C)]
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]
pub(crate) struct SortMicros {
    high: u32,
    low: u32,
}

impl SortMicros {
    const fn new(value: u64) -> Self {
        Self {
            high: (value >> 32) as u32,
            low: value as u32,
        }
    }

    const fn get(self) -> u64 {
        ((self.high as u64) << 32) | self.low as u64
    }
}

#[repr(C)]
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Ord, PartialOrd)]
pub(crate) struct BucketKey {
    micros: SortMicros,
    source: SourceId,
}

impl BucketKey {
    pub(crate) const fn new(micros: u64, source: SourceId) -> Self {
        Self {
            micros: SortMicros::new(micros),
            source,
        }
    }

    fn token(self) -> PageToken {
        PageToken::new(self.micros.get(), self.source.index())
    }

    fn from_token(tok: PageToken) -> Self {
        Self::new(tok.micros, SourceId::from_raw(tok.source))
    }
}

pub mod claims;
pub mod coverage;
pub mod invites;
pub mod labels;
pub mod settle;
mod sorted_blocks;
pub mod state_index;
mod state_view;
pub mod stream;
pub use claims::{ClaimDelta, ClaimIndex, RepoTransition, Rewrite};
pub use coverage::{Coverage, CoverageWatch, HydrantCursor, PromotionSignal};
pub use invites::{InviteIndex, InviteScope, Offer};
pub use labels::{
    IncomingLabel, LabelEntry, LabelIndex, LabelTarget, LabelVal, ModerationLabel,
    UnsupportedTarget,
};
pub use settle::{ParsedCid, RecordOutcome, Rejection, Settlement, Settlements, Watch};
use sorted_blocks::{PageKey, SortedBlocks};
pub use state_index::{StateKind, TicketStateKind};
use state_view::{ProjectedState, StateViewIndex};
pub use stream::{StreamHealth, StreamHealthSnapshot, StreamState};

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct SourceTag;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
struct AuthorTag;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
struct CollectionTag;

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct Interned<T>(u32, PhantomData<T>);

impl<T: Copy> Interned<T> {
    fn from_spur(spur: Spur) -> Self {
        Self(spur.into_usize() as u32, PhantomData)
    }

    fn from_raw(raw: u32) -> Self {
        Self(raw, PhantomData)
    }

    fn index(self) -> u32 {
        self.0
    }

    fn to_spur(self) -> Option<Spur> {
        Spur::try_from_usize(self.0 as usize)
    }
}

pub type SourceId = Interned<SourceTag>;
pub(crate) type AuthorId = Interned<AuthorTag>;
type CollectionId = Interned<CollectionTag>;

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PageToken {
    micros: u64,
    source: u32,
}

impl PageToken {
    pub fn new(micros: u64, source: u32) -> Self {
        Self { micros, source }
    }

    pub fn micros(self) -> u64 {
        self.micros
    }

    pub fn source(self) -> u32 {
        self.source
    }

    pub fn encode_token(self) -> String {
        let mut bytes = [0u8; 12];
        bytes[..8].copy_from_slice(&self.micros.to_be_bytes());
        bytes[8..].copy_from_slice(&self.source.to_be_bytes());
        encode_hex(&bytes)
    }

    pub fn decode_token(token: &str) -> Result<Self, CursorParseError> {
        let bytes: [u8; 12] = decode_hex_array(token).ok_or(CursorParseError::Malformed)?;
        let micros = u64::from_be_bytes(bytes[..8].try_into().unwrap());
        let source = u32::from_be_bytes(bytes[8..].try_into().unwrap());
        Ok(Self { micros, source })
    }
}

fn encode_hex(bytes: &[u8]) -> String {
    bytes
        .iter()
        .fold(String::with_capacity(bytes.len() * 2), |mut acc, b| {
            acc.push(char::from_digit((b >> 4) as u32, 16).unwrap());
            acc.push(char::from_digit((b & 0x0f) as u32, 16).unwrap());
            acc
        })
}

fn decode_hex_array<const N: usize>(token: &str) -> Option<[u8; N]> {
    if token.len() != N * 2 {
        return None;
    }
    let parsed: Vec<u8> = token
        .as_bytes()
        .chunks_exact(2)
        .map(|pair| {
            let hi = (pair[0] as char).to_digit(16)?;
            let lo = (pair[1] as char).to_digit(16)?;
            Some(((hi << 4) | lo) as u8)
        })
        .collect::<Option<Vec<u8>>>()?;
    parsed.try_into().ok()
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PageCursor {
    Start,
    After(PageToken),
}

#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    serde::Deserialize,
    serde::Serialize,
)]
#[serde(transparent)]
pub struct PageOffset(u64);

impl PageOffset {
    pub const ZERO: Self = Self(0);

    pub const fn new(value: u64) -> Self {
        Self(value)
    }

    pub const fn get(self) -> u64 {
        self.0
    }

    pub const fn as_usize(self) -> usize {
        self.0 as usize
    }
}

impl From<u64> for PageOffset {
    fn from(v: u64) -> Self {
        Self(v)
    }
}

impl From<usize> for PageOffset {
    fn from(v: usize) -> Self {
        Self(v as u64)
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PageRank(usize);

impl PageRank {
    pub const ZERO: Self = Self(0);

    pub const fn new(value: usize) -> Self {
        Self(value)
    }

    pub const fn get(self) -> usize {
        self.0
    }
}

impl From<usize> for PageRank {
    fn from(v: usize) -> Self {
        Self(v)
    }
}

#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    serde::Deserialize,
    serde::Serialize,
)]
#[serde(transparent)]
pub struct TotalCount(u64);

impl TotalCount {
    pub const ZERO: Self = Self(0);

    pub const fn new(value: u64) -> Self {
        Self(value)
    }

    pub const fn get(self) -> u64 {
        self.0
    }
}

impl From<u64> for TotalCount {
    fn from(v: u64) -> Self {
        Self(v)
    }
}

impl From<usize> for TotalCount {
    fn from(v: usize) -> Self {
        Self(v as u64)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PageStart {
    Cursor(PageCursor),
    Offset(PageOffset),
}

impl From<PageCursor> for PageStart {
    fn from(c: PageCursor) -> Self {
        Self::Cursor(c)
    }
}

impl From<PageOffset> for PageStart {
    fn from(o: PageOffset) -> Self {
        Self::Offset(o)
    }
}

impl From<u64> for PageStart {
    fn from(v: u64) -> Self {
        Self::Offset(PageOffset::new(v))
    }
}

#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum SortDir {
    Asc,
    #[default]
    Desc,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]
pub enum CursorParseError {
    #[error("cursor token must be a valid TID")]
    Malformed,
}

impl PageCursor {
    pub fn from_token(raw: Option<&str>) -> Result<Self, CursorParseError> {
        raw.map_or(Ok(Self::Start), |t| {
            PageToken::decode_token(t).map(Self::After)
        })
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PageLimit(u32);

#[derive(Clone, Copy, Debug, Eq, PartialEq, Error)]
pub enum PageLimitError {
    #[error("page limit {value} below minimum {min}")]
    TooSmall { value: u32, min: u32 },
    #[error("page limit {value} above maximum {max}")]
    TooLarge { value: u32, max: u32 },
}

impl PageLimit {
    pub const MIN: u32 = 1;
    pub const MAX: u32 = 1000;

    pub fn new(value: u32) -> Result<Self, PageLimitError> {
        match value {
            v if v < Self::MIN => Err(PageLimitError::TooSmall {
                value: v,
                min: Self::MIN,
            }),
            v if v > Self::MAX => Err(PageLimitError::TooLarge {
                value: v,
                max: Self::MAX,
            }),
            v => Ok(Self(v)),
        }
    }

    pub const fn get(self) -> u32 {
        self.0
    }
}

#[derive(Debug)]
pub struct EdgePage {
    pub items: Vec<EdgeItem>,
    pub next: Option<PageToken>,
    pub total: Option<TotalCount>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct EdgeItem {
    pub uri: AtUri<DefaultStr>,
    pub sort_micros: u64,
    pub sort_source: u32,
}

impl EdgeItem {
    pub fn sort_token(&self) -> PageToken {
        PageToken::new(self.sort_micros, self.sort_source)
    }
}

impl AsRef<str> for EdgeItem {
    fn as_ref(&self) -> &str {
        self.uri.as_ref()
    }
}

#[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,
    pub source_count: u64,
    pub source_interner_bytes: u64,
    pub did_interner_bytes: u64,
    pub collection_interner_bytes: u64,
    pub key_interner_bytes: u64,
    pub edges_total: u64,
    pub author_refs_total: u64,
    pub reverse_entries: u64,
    pub reverse_cap: u64,
    pub forward_struct_bytes: u64,
    pub reverse_struct_bytes: u64,
    pub bucket_struct_bytes: u64,
    pub max_bucket: u64,
    pub bucket_size_classes: [u64; BUCKET_CLASS_COUNT],
}

const BUCKET_CLASS_BOUNDS: [u64; 16] = [
    1,
    2,
    4,
    8,
    16,
    32,
    64,
    128,
    256,
    512,
    1024,
    2048,
    8192,
    32768,
    131072,
    u64::MAX,
];
const BUCKET_CLASS_COUNT: usize = BUCKET_CLASS_BOUNDS.len();

fn bucket_class(n: u64) -> usize {
    BUCKET_CLASS_BOUNDS
        .iter()
        .position(|&bound| n <= bound)
        .unwrap_or(BUCKET_CLASS_COUNT - 1)
}

impl EdgeMemReport {
    pub fn bucket_histogram(&self) -> impl Iterator<Item = (u64, u64)> {
        BUCKET_CLASS_BOUNDS
            .into_iter()
            .zip(self.bucket_size_classes)
    }
}

#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub(crate) struct EdgeKeyId(u32);

pub(crate) fn bump_author(
    authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>,
    author: AuthorId,
) {
    authors
        .entry(author)
        .and_modify(|c| *c = c.saturating_add(1))
        .or_insert(NonZeroU32::MIN);
}

pub(crate) fn drop_author(
    authors: &mut HashMap<AuthorId, NonZeroU32, RuntimeHasher>,
    author: AuthorId,
) {
    match authors.get(&author).map(|c| c.get() - 1) {
        Some(0) | None => {
            authors.remove(&author);
        }
        Some(next) => {
            authors.insert(author, NonZeroU32::new(next).unwrap());
        }
    }
}

enum AuthorHistory {
    Small(SmallVec<[BucketKey; 4]>),
    Large(BlockStore),
}

impl AuthorHistory {
    fn pair(left: BucketKey, right: BucketKey) -> Self {
        let mut keys = SmallVec::from_slice(&[left, right]);
        keys.sort_unstable();
        Self::Small(keys)
    }

    fn len(&self) -> usize {
        match self {
            Self::Small(keys) => keys.len(),
            Self::Large(keys) => keys.len(),
        }
    }

    fn newest(&self) -> BucketKey {
        match self {
            Self::Small(keys) => *keys.last().unwrap(),
            Self::Large(keys) => keys.last().expect("nonempty author history"),
        }
    }

    fn insert(&mut self, key: BucketKey) -> bool {
        match self {
            Self::Small(keys) => match keys.binary_search(&key) {
                Ok(_) => false,
                Err(position) if keys.len() < keys.inline_size() => {
                    keys.insert(position, key);
                    true
                }
                Err(position) => {
                    let mut promoted = keys.to_vec();
                    promoted.insert(position, key);
                    *self = Self::Large(BlockStore::from_sorted(promoted));
                    true
                }
            },
            Self::Large(keys) => keys.insert(key),
        }
    }

    fn remove(&mut self, key: &BucketKey) -> bool {
        match self {
            Self::Small(keys) => keys
                .binary_search(key)
                .map(|position| {
                    keys.remove(position);
                })
                .is_ok(),
            Self::Large(keys) => {
                if !keys.remove(key) {
                    return false;
                }
                if keys.len() <= 4 {
                    *self = Self::Small(keys.directed(PageCursor::Start, SortDir::Asc).collect());
                }
                true
            }
        }
    }

    fn heap_bytes(&self) -> u64 {
        match self {
            Self::Small(keys) if keys.spilled() => keys.capacity() as u64 * BUCKET_KEY_BYTES,
            Self::Small(_) => 0,
            Self::Large(keys) => keys.heap_bytes(),
        }
    }
}

fn estimated_hashmap_heap_bytes<K, V, S>(map: &HashMap<K, V, S>) -> u64 {
    let slot = std::mem::size_of::<(K, V)>() as u64 + 1;
    map.capacity() as u64 * slot
}

struct AuthorIndex {
    newest: HashMap<AuthorId, BucketKey, RuntimeHasher>,
    histories: HashMap<AuthorId, Box<AuthorHistory>, RuntimeHasher>,
    representatives: BlockStore,
}

impl AuthorIndex {
    fn new(hasher: RuntimeHasher) -> Self {
        Self {
            newest: HashMap::with_hasher(hasher.clone()),
            histories: HashMap::with_hasher(hasher),
            representatives: BlockStore::from_sorted(Vec::new()),
        }
    }

    fn len(&self) -> usize {
        self.newest.len()
    }

    fn contains(&self, author: &AuthorId) -> bool {
        self.newest.contains_key(author)
    }

    fn count(&self, author: &AuthorId) -> usize {
        self.newest
            .contains_key(author)
            .then(|| {
                self.histories
                    .get(author)
                    .map_or(1, |history| history.len())
            })
            .unwrap_or(0)
    }

    fn replace_representative(&mut self, previous: BucketKey, next: BucketKey) {
        if self.representatives.replace_last(previous, next) {
            return;
        }
        let removed = self.representatives.remove(&previous);
        let inserted = self.representatives.insert(next);
        debug_assert!(removed && inserted);
    }

    fn insert(&mut self, author: AuthorId, key: BucketKey) {
        let Some(previous) = self.newest.get(&author).copied() else {
            let inserted = self.representatives.insert(key);
            debug_assert!(inserted);
            self.newest.insert(author, key);
            return;
        };
        let inserted = match self.histories.get_mut(&author) {
            Some(history) => history.insert(key),
            None => {
                self.histories
                    .insert(author, Box::new(AuthorHistory::pair(previous, key)));
                true
            }
        };
        debug_assert!(inserted);
        if key > previous {
            self.replace_representative(previous, key);
            self.newest.insert(author, key);
        }
    }

    fn remove(&mut self, author: AuthorId, key: &BucketKey) {
        let Some(newest) = self.newest.get(&author).copied() else {
            debug_assert!(false, "large bucket missing author");
            return;
        };
        let Some(history) = self.histories.get_mut(&author) else {
            debug_assert_eq!(*key, newest, "single author key is not newest");
            self.newest.remove(&author);
            let removed = self.representatives.remove(&newest);
            debug_assert!(removed);
            return;
        };
        let removed = history.remove(key);
        debug_assert!(removed);
        let next = history.newest();
        if history.len() == 1 {
            self.histories.remove(&author);
        }
        if *key == newest {
            self.replace_representative(newest, next);
            self.newest.insert(author, next);
        }
    }

    fn heap_bytes(&self) -> u64 {
        estimated_hashmap_heap_bytes(&self.newest)
            + estimated_hashmap_heap_bytes(&self.histories)
            + self.representatives.heap_bytes()
            + self
                .histories
                .values()
                .map(|history| std::mem::size_of::<AuthorHistory>() as u64 + history.heap_bytes())
                .sum::<u64>()
    }
}

struct LargeBucket {
    keys: BlockStore,
    authors: AuthorIndex,
}

impl LargeBucket {
    fn insert(&mut self, key: BucketKey, author: Option<AuthorId>) -> bool {
        if !self.keys.insert(key) {
            return false;
        }
        if let Some(author) = author {
            self.authors.insert(author, key);
        }
        true
    }

    fn remove(&mut self, key: &BucketKey, author: Option<AuthorId>) -> bool {
        if !self.keys.remove(key) {
            return false;
        }
        if let Some(author) = author {
            self.authors.remove(author, key);
        }
        true
    }
}

const BUCKET_PROMOTE_AT: usize = 256;
const BUCKET_KEY_BYTES: u64 = std::mem::size_of::<BucketKey>() as u64;

type BlockStore = SortedBlocks<BucketKey>;

impl PageKey for BucketKey {
    fn token(self) -> PageToken {
        self.token()
    }

    fn from_token(tok: PageToken) -> Self {
        Self::from_token(tok)
    }
}

enum Sources {
    Small(SmallVec<[BucketKey; 2]>),
    Large(Box<LargeBucket>),
}

impl Default for Sources {
    fn default() -> Self {
        Self::Small(SmallVec::new())
    }
}

impl Sources {
    fn len(&self) -> usize {
        match self {
            Self::Small(v) => v.len(),
            Self::Large(big) => big.keys.len(),
        }
    }

    fn is_empty(&self) -> bool {
        self.len() == 0
    }

    fn insert(&mut self, key: BucketKey, author: Option<AuthorId>) -> bool {
        match self {
            Self::Small(v) => match v.binary_search(&key) {
                Ok(_) => false,
                Err(pos) => {
                    v.insert(pos, key);
                    true
                }
            },
            Self::Large(big) => big.insert(key, author),
        }
    }

    fn remove(&mut self, key: &BucketKey, author: Option<AuthorId>) {
        match self {
            Self::Small(v) => {
                if let Ok(pos) = v.binary_search(key) {
                    v.remove(pos);
                }
            }
            Self::Large(big) => {
                big.remove(key, author);
            }
        }
    }

    fn directed(
        &self,
        cursor: PageCursor,
        dir: SortDir,
    ) -> Box<dyn Iterator<Item = BucketKey> + '_> {
        match self {
            Self::Small(v) => Box::new(directed_slice(v, cursor, dir)),
            Self::Large(big) => big.keys.directed(cursor, dir),
        }
    }

    fn select(&self, rank: PageRank) -> Option<BucketKey> {
        match self {
            Self::Small(v) => v.get(rank.get()).copied(),
            Self::Large(big) => big.keys.select(rank),
        }
    }

    fn resolve_start(&self, start: PageStart, dir: SortDir) -> Option<PageCursor> {
        match start {
            PageStart::Cursor(c) => Some(c),
            PageStart::Offset(o) if o.get() == 0 => Some(PageCursor::Start),
            PageStart::Offset(o) => {
                let len = self.len() as u64;
                let o = o.get();
                if o >= len {
                    return None;
                }
                let pred_rank = match dir {
                    SortDir::Asc => o - 1,
                    SortDir::Desc => len - o,
                };
                self.select(PageRank::new(pred_rank as usize))
                    .map(|k| PageCursor::After(k.token()))
            }
        }
    }

    fn heap_bytes(&self) -> u64 {
        match self {
            Self::Small(v) if v.spilled() => v.capacity() as u64 * BUCKET_KEY_BYTES,
            Self::Small(_) => 0,
            Self::Large(big) => {
                std::mem::size_of::<LargeBucket>() as u64
                    + big.keys.heap_bytes()
                    + big.authors.heap_bytes()
            }
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ReverseEntry {
    key_id: EdgeKeyId,
    sort_micros: SortMicros,
}

#[cfg(target_pointer_width = "64")]
const _: () = {
    assert!(std::mem::size_of::<SortMicros>() == 8);
    assert!(std::mem::align_of::<SortMicros>() == 4);
    assert!(std::mem::size_of::<BucketKey>() == 12);
    assert!(std::mem::align_of::<BucketKey>() == 4);
    assert!(std::mem::size_of::<ReverseEntry>() == 12);
    assert!(std::mem::size_of::<(AuthorId, BucketKey)>() == 16);
    assert!(std::mem::size_of::<(AuthorId, Box<AuthorHistory>)>() == 16);
};

pub struct EdgeStore {
    source_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
    did_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
    collection_interner: Arc<ThreadedRodeo<Spur, RuntimeHasher>>,
    key_ids: SccMap<EdgeKey, EdgeKeyId, RuntimeHasher>,
    keys: SccMap<EdgeKeyId, EdgeKey, RuntimeHasher>,
    next_key_id: AtomicU32,
    forward: SccMap<EdgeKeyId, Sources, RuntimeHasher>,
    reverse: SccMap<SourceId, SmallVec<[ReverseEntry; 1]>, RuntimeHasher>,
    source_collections: SccMap<SourceId, Vec<SmolStr>, RuntimeHasher>,
    ticket_states: StateViewIndex<TicketStateKind>,
    hasher: RuntimeHasher,
    writer: Mutex<()>,
}

impl EdgeStore {
    pub fn new(hasher: RuntimeHasher) -> Self {
        Self {
            source_interner: Arc::new(ThreadedRodeo::with_hasher(hasher.clone())),
            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()),
            source_collections: SccMap::with_hasher(hasher.clone()),
            ticket_states: StateViewIndex::new(hasher.clone()),
            hasher,
            writer: Mutex::new(()),
        }
    }

    fn intern_key(&self, key: EdgeKey) -> EdgeKeyId {
        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
            }
        }
    }

    fn lookup_key(&self, key: &EdgeKey) -> Option<EdgeKeyId> {
        self.key_ids.read_sync(key, |_, id| *id)
    }

    pub fn add(&self, edge: Edge) {
        let _w = self
            .writer
            .lock()
            .expect("edge-store writer mutex poisoned");
        self.add_locked(edge);
    }

    pub fn intern_source(&self, source: &AtUri<DefaultStr>) -> SourceId {
        let author = self.intern_author(source);
        SourceId::from_spur(
            self.source_interner
                .get_or_intern(self.source_key(source, author)),
        )
    }

    /// a source's id without interning anything, `None` if the index never saw it
    pub fn lookup_source(&self, source: &AtUri<DefaultStr>) -> Option<SourceId> {
        let key = match (
            split_record_uri(source.as_ref()),
            source_authority_did(source),
        ) {
            (Some((_, collection, rkey)), Some(did)) => {
                let author = self.lookup_did(did)?;
                let collection =
                    CollectionId::from_spur(self.collection_interner.get(collection)?);
                format!("{}/{}/{}", author.index(), collection.index(), rkey)
            }
            _ => source.as_ref().to_owned(),
        };
        self.source_interner.get(key).map(SourceId::from_spur)
    }

    pub(crate) fn intern_did(&self, did: &str) -> AuthorId {
        AuthorId::from_spur(self.did_interner.get_or_intern(did))
    }

    pub(crate) fn lookup_did(&self, did: &str) -> Option<AuthorId> {
        self.did_interner.get(did).map(AuthorId::from_spur)
    }

    pub fn upsert_source(&self, source: &AtUri<DefaultStr>, edges: Vec<Edge>) {
        let _w = self
            .writer
            .lock()
            .expect("edge-store writer mutex poisoned");
        self.clear_source_locked(source);
        edges.into_iter().for_each(|e| self.add_locked(e));
    }

    pub fn remove_source(&self, source: &AtUri<DefaultStr>) {
        let _w = self
            .writer
            .lock()
            .expect("edge-store writer mutex poisoned");
        self.clear_source_locked(source);
    }

    /// Store collection filters for a source. None or empty = all collections (removes entry).
    /// Call only after the source has been interned via add/upsert_source.
    pub fn set_source_collections(
        &self,
        source: &AtUri<DefaultStr>,
        collections: Option<Vec<SmolStr>>,
    ) {
        let author = source_authority_did(source)
            .and_then(|s| self.did_interner.get(s))
            .map(AuthorId::from_spur);
        let Some(spur) = self.source_interner.get(self.source_key(source, author)) else {
            return;
        };
        let id = SourceId::from_spur(spur);
        if let Some(cols) = collections
            && !cols.is_empty()
        {
            let _ = self.source_collections.insert_sync(id, cols);
        } else {
            self.source_collections.remove_sync(&id);
        }
    }

    /// Returns None if no collection filter is stored (meaning all collections).
    pub fn source_collections_for(&self, source: &AtUri<DefaultStr>) -> Option<Vec<SmolStr>> {
        let author = source_authority_did(source)
            .and_then(|s| self.did_interner.get(s))
            .map(AuthorId::from_spur);
        let spur = self.source_interner.get(self.source_key(source, author))?;
        let id = SourceId::from_spur(spur);
        self.source_collections
            .read_sync(&id, |_, cols| cols.clone())
    }
    fn add_locked(&self, edge: Edge) {
        let author = self.intern_author(&edge.source);
        let source_key = self.source_key(&edge.source, author);
        let id = SourceId::from_spur(self.source_interner.get_or_intern(source_key));
        let sort_micros = edge.sort_micros;
        let key_id = self.intern_key(EdgeKey::new(edge.kind, edge.subject));

        let key = BucketKey::new(sort_micros, id);
        let mut entry = self.forward.entry_sync(key_id).or_default();
        let inserted = entry.get_mut().insert(key, author);
        let promote_keys = match entry.get() {
            Sources::Small(v) if v.len() > BUCKET_PROMOTE_AT => {
                Some(v.iter().copied().collect::<Vec<_>>())
            }
            _ => None,
        };
        drop(entry);

        if let Some(keys) = promote_keys {
            let authors = self.build_author_index(&keys);
            let large = LargeBucket {
                keys: BlockStore::from_sorted(keys),
                authors,
            };
            if let Entry::Occupied(mut e) = self.forward.entry_sync(key_id) {
                *e.get_mut() = Sources::Large(Box::new(large));
            }
        }

        if inserted {
            let mut rev = self.reverse.entry_sync(id).or_default();
            rev.get_mut().push(ReverseEntry {
                key_id,
                sort_micros: SortMicros::new(sort_micros),
            });
        }
    }

    fn clear_source_locked(&self, source: &AtUri<DefaultStr>) {
        let author = source_authority_did(source)
            .and_then(|s| self.did_interner.get(s))
            .map(AuthorId::from_spur);
        let Some(source_spur) = self.source_interner.get(self.source_key(source, author)) else {
            return;
        };
        let id = SourceId::from_spur(source_spur);
        self.source_collections.remove_sync(&id);
        let Some((_, entries)) = self.reverse.remove_sync(&id) else {
            return;
        };
        entries.into_iter().for_each(
            |ReverseEntry {
                 key_id,
                 sort_micros,
             }| {
                self.forward.update_sync(&key_id, |_, sources| {
                    sources.remove(&BucketKey::new(sort_micros.get(), id), author);
                });
                self.forward
                    .remove_if_sync(&key_id, |sources| sources.is_empty());
            },
        );
    }

    fn intern_author(&self, source: &AtUri<DefaultStr>) -> Option<AuthorId> {
        let did = source_authority_did(source)?;
        Some(AuthorId::from_spur(self.did_interner.get_or_intern(did)))
    }

    fn source_key(&self, source: &AtUri<DefaultStr>, author: Option<AuthorId>) -> String {
        match (split_record_uri(source.as_ref()), author) {
            (Some((_, collection, rkey)), Some(author)) => {
                let collection =
                    CollectionId::from_spur(self.collection_interner.get_or_intern(collection));
                format!("{}/{}/{}", author.index(), collection.index(), rkey)
            }
            _ => source.as_ref().to_owned(),
        }
    }

    fn decode_source(&self, stored: &str) -> Option<String> {
        if stored.starts_with("at://") {
            return Some(stored.to_owned());
        }
        let mut parts = stored.splitn(3, '/');
        let author = AuthorId::from_raw(parts.next()?.parse().ok()?);
        let collection = CollectionId::from_raw(parts.next()?.parse().ok()?);
        let rkey = parts.next()?;
        let did = self.did_interner.try_resolve(&author.to_spur()?)?;
        let collection = self
            .collection_interner
            .try_resolve(&collection.to_spur()?)?;
        Some(format!("at://{did}/{collection}/{rkey}"))
    }

    fn edge_item(&self, key: BucketKey) -> Option<EdgeItem> {
        let spur = key.source.to_spur()?;
        let stored = self.source_interner.try_resolve(&spur)?;
        let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?;
        Some(EdgeItem {
            uri,
            sort_micros: key.micros.get(),
            sort_source: key.token().source(),
        })
    }

    fn edge_page(
        &self,
        total: usize,
        keys: impl Iterator<Item = BucketKey>,
        limit: PageLimit,
    ) -> EdgePage {
        let limit = limit.get() as usize;
        let entries = keys.take(limit + 1).collect::<Vec<_>>();
        let has_more = entries.len() > limit;
        let page = &entries[..entries.len().min(limit)];
        let items = page.iter().filter_map(|&key| self.edge_item(key)).collect();
        let next = has_more
            .then(|| page.last().copied())
            .flatten()
            .map(BucketKey::token);
        EdgePage {
            items,
            next,
            total: Some(TotalCount::from(total)),
        }
    }

    fn author_of_stored(&self, stored: &str) -> Option<AuthorId> {
        match stored.strip_prefix("at://") {
            Some(rest) => {
                let authority = rest.split('/').next().unwrap_or(rest);
                authority
                    .starts_with("did:")
                    .then(|| self.did_interner.get(authority).map(AuthorId::from_spur))
                    .flatten()
            }
            None => stored
                .split('/')
                .next()?
                .parse::<u32>()
                .ok()
                .map(AuthorId::from_raw),
        }
    }

    fn author_of(&self, source: SourceId) -> Option<AuthorId> {
        let spur = source.to_spur()?;
        let stored = self.source_interner.try_resolve(&spur)?;
        self.author_of_stored(stored)
    }

    fn distinct_authors_small(&self, keys: &[BucketKey]) -> u64 {
        keys.iter()
            .filter_map(|key| self.author_of(key.source))
            .collect::<std::collections::HashSet<AuthorId>>()
            .len() as u64
    }

    fn build_author_index(&self, keys: &[BucketKey]) -> AuthorIndex {
        debug_assert!(keys.windows(2).all(|pair| pair[0] < pair[1]));
        let mut authors = AuthorIndex::new(self.hasher.clone());
        keys.iter().for_each(|&key| {
            if let Some(author) = self.author_of(key.source) {
                authors.insert(author, key);
            }
        });
        authors
    }
    fn distinct_authors(&self, sources: &Sources) -> u64 {
        match sources {
            Sources::Large(big) => big.authors.len() as u64,
            Sources::Small(keys) => self.distinct_authors_small(keys),
        }
    }

    fn distinct_authors_since(&self, sources: &Sources, since_micros: u64) -> u64 {
        sources
            .directed(PageCursor::Start, SortDir::Asc)
            .filter(|key| key.micros.get() >= since_micros)
            .filter_map(|key| self.author_of(key.source))
            .collect::<std::collections::HashSet<AuthorId>>()
            .len() as u64
    }

    fn ranked_dids(&self, kind: &str, since_micros: Option<u64>, limit: usize) -> Vec<Did> {
        let mut ranked = Vec::with_capacity(self.forward.len());
        self.forward.iter_sync(|key_id, sources| {
            let Some(key) = self.keys.read_sync(key_id, |_, key| key.clone()) else {
                return true;
            };
            let SubjectRef::Did(did) = key.subject else {
                return true;
            };
            if key.kind.as_ref() != kind {
                return true;
            }
            let count = since_micros.map_or_else(
                || self.distinct_authors(sources),
                |since| self.distinct_authors_since(sources, since),
            );
            if count > 0 {
                ranked.push((count, did));
            }
            true
        });
        ranked.sort_unstable_by(|(left_count, left_did), (right_count, right_did)| {
            right_count
                .cmp(left_count)
                .then_with(|| left_did.as_str().cmp(right_did.as_str()))
        });
        ranked.into_iter().take(limit).map(|(_, did)| did).collect()
    }

    pub fn most_followed(&self, limit: usize) -> Vec<Did> {
        self.ranked_dids("org.tangled.graph.follow", None, limit)
    }

    pub fn top_starred_since(&self, since_micros: u64, limit: usize) -> Vec<Did> {
        self.ranked_dids("org.tangled.feed.star", Some(since_micros), limit)
    }

    pub fn count(&self, key: &EdgeKey) -> u64 {
        self.lookup_key(key)
            .and_then(|id| {
                self.forward
                    .read_sync(&id, |_, sources| sources.len() as u64)
            })
            .unwrap_or(0)
    }

    pub fn count_distinct_authors(&self, key: &EdgeKey) -> u64 {
        self.lookup_key(key)
            .and_then(|id| {
                self.forward
                    .read_sync(&id, |_, sources| self.distinct_authors(sources))
            })
            .unwrap_or(0)
    }

    pub fn count_by_author(&self, key: &EdgeKey, author: &Did<DefaultStr>) -> 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.count(&author) as u64,
                    Sources::Small(keys) => keys
                        .iter()
                        .filter(|key| self.author_of(key.source) == Some(author))
                        .count() as u64,
                })
            })
            .unwrap_or(0)
    }

    pub fn refresh_ticket_counts(&self, entity: &AtUri<DefaultStr>, kind: TicketStateKind) {
        // compute under the write lock so concurrent updates dont write a stale projection
        let mut inner = self.ticket_states.write_inner();
        let source = self.intern_source(entity);
        let author = self.author_of(source);
        let reverse = self
            .reverse
            .read_sync(&source, |_, entries| entries.clone())
            .unwrap_or_default();
        let current = reverse
            .into_iter()
            .filter_map(|entry| {
                let key = self.keys.read_sync(&entry.key_id, |_, key| key.clone())?;
                [TICKET_ISSUE_KIND, TICKET_PULL_KIND]
                    .contains(&key.kind.as_ref())
                    .then(|| ProjectedState {
                        key: entry.key_id,
                        kind,
                        author,
                        micros: entry.sort_micros.get(),
                        // tickets are authored by the repo, so author sub-views would mirror the bucket
                        by_bucket: true,
                    })
            })
            .collect();
        StateViewIndex::apply_projection(&mut inner, source, &current);
    }

    pub fn count_ticket_state(&self, key: &EdgeKey, kind: TicketStateKind) -> FilteredCount {
        let Some(key) = self.lookup_key(key) else {
            return FilteredCount::default();
        };
        let (count, distinct) = self.ticket_states.counts(key, kind, None);
        FilteredCount::new(count, distinct)
    }

    pub fn page_ticket_state(
        &self,
        key: &EdgeKey,
        kind: TicketStateKind,
        start: impl Into<PageStart>,
        limit: PageLimit,
        dir: SortDir,
    ) -> EdgePage {
        let page = self.lookup_key(key).and_then(|key_id| {
            self.ticket_states
                .page(key_id, Some(kind), start.into(), limit, dir, None)
        });
        let Some(page) = page else {
            return EdgePage {
                items: Vec::new(),
                next: None,
                total: Some(TotalCount::ZERO),
            };
        };
        let items = page
            .keys
            .iter()
            .filter_map(|&key| self.edge_item(key))
            .collect();
        EdgePage {
            items,
            next: page.next,
            total: Some(page.total),
        }
    }

    /// answers "did the viewer star/follow/etc. this subject, and with what rkey"
    pub fn viewer_source(&self, key: &EdgeKey, viewer: &str) -> Option<AtUri<DefaultStr>> {
        let author_spur = self.did_interner.get(viewer)?;
        let author_id = AuthorId::from_spur(author_spur);
        let key_id = self.lookup_key(key)?;

        self.forward
            .read_sync(&key_id, |_, sources| {
                // large buckets track authors, so a missing author fast-fails the scan
                if let Sources::Large(big) = sources
                    && !big.authors.contains(&author_id)
                {
                    return None;
                }
                sources
                    .directed(PageCursor::Start, SortDir::Desc)
                    .find_map(|bucket| {
                        let spur = bucket.source.to_spur()?;
                        let stored = self.source_interner.try_resolve(&spur)?;
                        let uri = AtUri::new_owned(self.decode_source(stored)?).ok()?;
                        (source_authority_did(&uri) == Some(viewer)).then_some(uri)
                    })
            })
            .flatten()
    }

    pub fn sources_for(&self, key: &EdgeKey) -> Vec<AtUri<DefaultStr>> {
        self.lookup_key(key)
            .and_then(|id| {
                self.forward.read_sync(&id, |_, sources| {
                    sources
                        .directed(PageCursor::Start, SortDir::Desc)
                        .filter_map(|bucket| {
                            let spur = bucket.source.to_spur()?;
                            let stored = self.source_interner.try_resolve(&spur)?;
                            AtUri::new_owned(self.decode_source(stored)?).ok()
                        })
                        .collect::<Vec<_>>()
                })
            })
            .unwrap_or_default()
    }

    pub fn list(
        &self,
        key: &EdgeKey,
        start: impl Into<PageStart>,
        limit: PageLimit,
        dir: SortDir,
    ) -> EdgePage {
        let start = start.into();
        self.lookup_key(key)
            .and_then(|id| {
                self.forward.read_sync(&id, |_, sources| {
                    let total = sources.len();
                    let Some(cursor) = sources.resolve_start(start, dir) else {
                        return EdgePage {
                            items: Vec::new(),
                            next: None,
                            total: Some(TotalCount::from(total)),
                        };
                    };
                    self.edge_page(total, sources.directed(cursor, dir), limit)
                })
            })
            .unwrap_or(EdgePage {
                items: Vec::new(),
                next: None,
                total: None,
            })
    }

    pub fn list_distinct_authors(
        &self,
        key: &EdgeKey,
        start: impl Into<PageStart>,
        limit: PageLimit,
        dir: SortDir,
    ) -> EdgePage {
        let start = start.into();
        self.lookup_key(key)
            .and_then(|id| {
                self.forward.read_sync(&id, |_, sources| match sources {
                    Sources::Small(keys) => {
                        let mut seen = HashSet::with_hasher(self.hasher.clone());
                        let mut representatives = keys
                            .iter()
                            .rev()
                            .filter(|key| {
                                self.author_of(key.source)
                                    .is_some_and(|author| seen.insert(author))
                            })
                            .copied()
                            .collect::<SmallVec<[BucketKey; 2]>>();
                        representatives.reverse();
                        let representatives = Sources::Small(representatives);
                        let total = representatives.len();
                        let Some(cursor) = representatives.resolve_start(start, dir) else {
                            return EdgePage {
                                items: Vec::new(),
                                next: None,
                                total: Some(TotalCount::from(total)),
                            };
                        };
                        self.edge_page(total, representatives.directed(cursor, dir), limit)
                    }
                    Sources::Large(big) => {
                        let total = big.authors.representatives.len();
                        debug_assert_eq!(total, big.authors.len());
                        let Some(cursor) = big.authors.representatives.resolve_start(start, dir)
                        else {
                            return EdgePage {
                                items: Vec::new(),
                                next: None,
                                total: Some(TotalCount::from(total)),
                            };
                        };
                        self.edge_page(
                            total,
                            big.authors.representatives.directed(cursor, dir),
                            limit,
                        )
                    }
                })
            })
            .unwrap_or(EdgePage {
                items: Vec::new(),
                next: None,
                total: None,
            })
    }

    /// Merge a page across many buckets in one global time order.
    pub fn list_multi(
        &self,
        keys: &[EdgeKey],
        cursor: PageCursor,
        limit: PageLimit,
        dir: SortDir,
    ) -> EdgePage {
        let n = limit.get() as usize + 1;
        let is_first = |x: &BucketKey, y: &BucketKey| match dir {
            SortDir::Asc => x <= y,
            SortDir::Desc => x >= y,
        };
        let mut top: Vec<BucketKey> = Vec::with_capacity(n);
        for key in keys {
            let Some(id) = self.lookup_key(key) else {
                continue;
            };
            let bucket: Vec<BucketKey> = self
                .forward
                .read_sync(&id, |_, sources| {
                    sources.directed(cursor, dir).take(n).collect()
                })
                .unwrap_or_default();
            top = top
                .into_iter()
                .merge_by(bucket, &is_first)
                .take(n)
                .collect();
        }
        top.dedup_by_key(|k| k.source);

        let has_more = top.len() > limit.get() as usize;
        let page = &top[..top.len().min(limit.get() as usize)];
        let items = page.iter().filter_map(|&key| self.edge_item(key)).collect();
        let next = has_more
            .then(|| page.last().copied())
            .flatten()
            .map(BucketKey::token);
        EdgePage {
            items,
            next,
            total: None,
        }
    }

    pub fn key_count(&self) -> usize {
        self.forward.len()
    }

    pub fn source_count(&self) -> usize {
        self.reverse.len()
    }

    pub fn mem_report(&self) -> EdgeMemReport {
        const SCC_SLOT: u64 = 32;
        let edge_key = std::mem::size_of::<EdgeKey>() as u64;
        let rev_entry = std::mem::size_of::<ReverseEntry>() as u64;
        let rev_smallvec = std::mem::size_of::<SmallVec<[ReverseEntry; 1]>>() as u64;
        let bucket_struct = std::mem::size_of::<Sources>() as u64;

        let mut edges_total = 0u64;
        let mut author_refs_total = 0u64;
        let mut forward_struct_bytes = self.ticket_states.heap_bytes();
        let mut max_bucket = 0u64;
        let mut bucket_size_classes = [0u64; BUCKET_CLASS_COUNT];
        self.forward.iter_sync(|_, sources| {
            let bucket_len = sources.len() as u64;
            edges_total += bucket_len;
            max_bucket = max_bucket.max(bucket_len);
            bucket_size_classes[bucket_class(bucket_len)] += 1;
            if let Sources::Large(big) = sources {
                author_refs_total += big.authors.len() as u64;
            }
            forward_struct_bytes += SCC_SLOT + bucket_struct + sources.heap_bytes();
            true
        });
        let key_interner_bytes = self.key_ids.len() as u64
            * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT)
            + self.keys.len() as u64
                * (edge_key + std::mem::size_of::<EdgeKeyId>() as u64 + SCC_SLOT);

        let mut reverse_entries = 0u64;
        let mut reverse_cap = 0u64;
        let mut reverse_struct_bytes = 0u64;
        self.reverse.iter_sync(|_, refs| {
            let cap = refs.capacity() as u64;
            reverse_entries += refs.len() as u64;
            reverse_cap += cap;
            let heap = if refs.spilled() { cap * rev_entry } else { 0 };
            reverse_struct_bytes += SCC_SLOT + rev_smallvec + heap;
            true
        });

        EdgeMemReport {
            key_count: self.forward.len() as u64,
            source_count: self.reverse.len() as u64,
            source_interner_bytes: self.source_interner.current_memory_usage() as u64,
            did_interner_bytes: self.did_interner.current_memory_usage() as u64,
            collection_interner_bytes: self.collection_interner.current_memory_usage() as u64,
            key_interner_bytes,
            edges_total,
            author_refs_total,
            reverse_entries,
            reverse_cap,
            forward_struct_bytes,
            reverse_struct_bytes,
            bucket_struct_bytes: bucket_struct,
            max_bucket,
            bucket_size_classes,
        }
    }
}

pub fn upsert_record_indexes(
    edges: &EdgeStore,
    claims: &ClaimIndex,
    source: &AtUri<DefaultStr>,
    record_edges: Vec<Edge>,
    record: &Record,
) -> Vec<RepoTransition> {
    // manifests only get their edges once claimed, and a declaration can flip a manifest it points at
    let transitions = match record {
        Record::Repo(m) => apply_claims(
            edges,
            claims.upsert_manifest(
                source,
                m.declaration.owner.as_ref(),
                m.declaration.name.0.as_str(),
                record_edges,
            ),
        ),
        Record::Declaration(d) => {
            apply_claims(edges, claims.upsert_declaration(source, d.repo.as_ref()))
        }
        _ => {
            edges.upsert_source(source, record_edges);
            Vec::new()
        }
    };
    if let Record::Ticket(t) = record {
        // an unknown ticket state counts as open
        edges.refresh_ticket_counts(
            source,
            TicketStateKind::from_ticket(t.state.as_ref()).unwrap_or_default(),
        );
    }
    transitions
}

fn apply_claims(edges: &EdgeStore, delta: ClaimDelta) -> Vec<RepoTransition> {
    for rewrite in delta.rewrites {
        edges.upsert_source(&rewrite.source, rewrite.edges);
    }
    delta.transitions
}

pub fn delete_record_indexes(
    edges: &EdgeStore,
    claims: &ClaimIndex,
    source: &AtUri<DefaultStr>,
    nsid: &Nsid<DefaultStr>,
) {
    edges.remove_source(source);
    match nsid.as_ref() {
        "org.tangled.repo.manifest" | "org.tangled.repo.declaration" => {
            apply_claims(edges, claims.delete(source));
        }
        "org.tangled.track.ticket" => edges.refresh_ticket_counts(source, TicketStateKind::Open),
        _ => {}
    }
}

fn directed_slice(
    sources: &[BucketKey],
    cursor: PageCursor,
    dir: SortDir,
) -> impl Iterator<Item = BucketKey> + '_ {
    match dir {
        SortDir::Asc => {
            let start = match cursor {
                PageCursor::Start => 0,
                PageCursor::After(tok) => {
                    sources.partition_point(|k| *k <= BucketKey::from_token(tok))
                }
            };
            Either::Left(sources[start..].iter().copied())
        }
        SortDir::Desc => {
            let end = match cursor {
                PageCursor::Start => sources.len(),
                PageCursor::After(tok) => {
                    sources.partition_point(|k| *k < BucketKey::from_token(tok))
                }
            };
            Either::Right(sources[..end].iter().rev().copied())
        }
    }
}

fn split_record_uri(source: &str) -> Option<(&str, &str, &str)> {
    let rest = source.strip_prefix("at://")?;
    let mut parts = rest.split('/');
    let authority = parts.next()?;
    let collection = parts.next()?;
    let rkey = parts.next()?;
    if parts.next().is_some() {
        return None;
    }
    (authority.starts_with("did:") && !collection.is_empty() && !rkey.is_empty())
        .then_some((authority, collection, rkey))
}

fn source_authority_did(source: &AtUri<DefaultStr>) -> Option<&str> {
    let rest = source.as_ref().strip_prefix("at://")?;
    let end = rest.find('/').unwrap_or(rest.len());
    let candidate = &rest[..end];
    candidate.starts_with("did:").then_some(candidate)
}

#[cfg(test)]
mod tests {
    use super::*;
    use bobbin_types::ids::SubjectRef;
    use jacquard_common::types::did::Did;
    use jacquard_common::types::nsid::Nsid;
    use jacquard_common::types::string::AtUri;

    fn store() -> EdgeStore {
        EdgeStore::new(RuntimeHasher::default())
    }

    fn nsid(s: &'static str) -> Nsid<DefaultStr> {
        Nsid::new_static(s).unwrap()
    }

    fn at(s: &str) -> AtUri<DefaultStr> {
        AtUri::new_owned(s).unwrap()
    }

    fn did(s: &str) -> Did<DefaultStr> {
        Did::new_owned(s).unwrap()
    }

    fn did_subj(s: &str) -> SubjectRef {
        SubjectRef::Did(did(s))
    }

    fn limit(n: u32) -> PageLimit {
        PageLimit::new(n).unwrap()
    }

    const NAMES: [&str; 5] = ["nel", "olaren", "teq", "lyna", "bailey"];

    fn star_edge(source: AtUri<DefaultStr>, subject: Did<DefaultStr>) -> Edge {
        star_edge_at(source, subject, 0)
    }

    fn star_edge_at(source: AtUri<DefaultStr>, subject: Did<DefaultStr>, sort_micros: u64) -> Edge {
        Edge {
            kind: nsid("org.tangled.feed.star"),
            subject: SubjectRef::Did(subject),
            source,
            sort_micros,
        }
    }

    fn follow_edge_at(
        source: AtUri<DefaultStr>,
        subject: Did<DefaultStr>,
        sort_micros: u64,
    ) -> Edge {
        Edge {
            kind: nsid("org.tangled.graph.follow"),
            subject: SubjectRef::Did(subject),
            source,
            sort_micros,
        }
    }

    fn shuffled_micros(i: usize) -> u64 {
        (i as u64).wrapping_mul(2_654_435_761) % 64
    }

    fn source_uri(i: usize) -> String {
        format!(
            "at://did:plc:{}/org.tangled.feed.star/r{i}",
            NAMES[i % NAMES.len()]
        )
    }

    fn fill_subject(store: &EdgeStore, n: usize) -> EdgeKey {
        let subject = did("did:plc:squid");
        (0..n).for_each(|i| {
            store.add(star_edge_at(
                at(&source_uri(i)),
                subject.clone(),
                shuffled_micros(i),
            ));
        });
        EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:squid"))
    }

    fn reference(kept: impl Iterator<Item = usize>) -> Vec<String> {
        let mut rows: Vec<(u64, usize)> = kept.map(|i| (shuffled_micros(i), i)).collect();
        rows.sort_unstable();
        rows.into_iter().map(|(_, i)| source_uri(i)).collect()
    }

    fn paginate_all(
        store: &EdgeStore,
        key: &EdgeKey,
        dir: SortDir,
        page: PageLimit,
    ) -> Vec<String> {
        std::iter::successors(
            Some(store.list(key, PageCursor::Start, page, dir)),
            |prev| {
                prev.next
                    .map(|tok| store.list(key, PageCursor::After(tok), page, dir))
            },
        )
        .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned()))
        .collect()
    }

    fn paginate_distinct_authors(
        store: &EdgeStore,
        key: &EdgeKey,
        dir: SortDir,
        page: PageLimit,
    ) -> Vec<String> {
        std::iter::successors(
            Some(store.list_distinct_authors(key, PageCursor::Start, page, dir)),
            |prev| {
                prev.next.map(|token| {
                    store.list_distinct_authors(key, PageCursor::After(token), page, dir)
                })
            },
        )
        .flat_map(|page| page.items.into_iter().map(|item| item.as_ref().to_owned()))
        .collect()
    }

    fn newest_author_reference(rows: impl IntoIterator<Item = String>) -> Vec<String> {
        let mut authors = HashSet::new();
        let mut newest = rows
            .into_iter()
            .collect::<Vec<_>>()
            .into_iter()
            .rev()
            .filter(|uri| {
                let author = split_record_uri(uri).unwrap().0;
                authors.insert(author.to_owned())
            })
            .collect::<Vec<_>>();
        newest.reverse();
        newest
    }

    #[test]
    fn pagination_matches_reference_across_small_and_large() {
        [64usize, 1000].into_iter().for_each(|n| {
            let store = store();
            let key = fill_subject(&store, n);
            assert_eq!(store.count(&key), n as u64, "count mismatch at n={n}");
            assert_eq!(
                store.count_distinct_authors(&key),
                NAMES.len() as u64,
                "distinct authors at n={n}"
            );

            let asc = reference(0..n);
            let desc: Vec<String> = asc.iter().rev().cloned().collect();

            [3u32, 7, 50].into_iter().for_each(|page| {
                assert_eq!(
                    paginate_all(&store, &key, SortDir::Asc, limit(page)),
                    asc,
                    "asc mismatch n={n} page={page}"
                );
                assert_eq!(
                    paginate_all(&store, &key, SortDir::Desc, limit(page)),
                    desc,
                    "desc mismatch n={n} page={page}"
                );
            });
        });
    }

    fn paginate_by_offsets(
        store: &EdgeStore,
        key: &EdgeKey,
        dir: SortDir,
        page: PageLimit,
    ) -> Vec<String> {
        std::iter::successors(
            Some((
                PageOffset::ZERO,
                store.list(key, PageOffset::ZERO, page, dir),
            )),
            |(offset, prev)| {
                prev.next.map(|_| {
                    let next = PageOffset::from(offset.get() + page.get() as u64);
                    (next, store.list(key, next, page, dir))
                })
            },
        )
        .flat_map(|(_, p)| p.items.into_iter().map(|u| u.as_ref().to_owned()))
        .collect()
    }

    #[test]
    fn distinct_author_pagination_matches_reference_across_small_and_large() {
        [64usize, 1000].into_iter().for_each(|n| {
            let store = store();
            let key = fill_subject(&store, n);
            let asc = newest_author_reference(reference(0..n));
            let desc = asc.iter().rev().cloned().collect::<Vec<_>>();

            [1u32, 2, 5].into_iter().for_each(|page| {
                assert_eq!(
                    paginate_distinct_authors(&store, &key, SortDir::Asc, limit(page)),
                    asc,
                    "asc distinct mismatch n={n} page={page}",
                );
                assert_eq!(
                    paginate_distinct_authors(&store, &key, SortDir::Desc, limit(page)),
                    desc,
                    "desc distinct mismatch n={n} page={page}",
                );
            });

            let offset =
                store.list_distinct_authors(&key, PageOffset::new(2), limit(2), SortDir::Asc);
            assert_eq!(
                offset
                    .items
                    .iter()
                    .map(|item| item.as_ref().to_owned())
                    .collect::<Vec<_>>(),
                asc.iter().skip(2).take(2).cloned().collect::<Vec<_>>(),
            );
            assert_eq!(
                offset.total,
                Some(TotalCount::from(NAMES.len())),
                "distinct total n={n}",
            );
        });
    }

    #[test]
    fn large_distinct_index_tracks_newest_representative() {
        let store = store();
        let subject = did("did:plc:large-subject");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject.clone()),
        );
        (0..255).for_each(|index| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:author{index}/org.tangled.feed.star/r{index}"
                )),
                subject.clone(),
                index as u64,
            ));
        });
        let older = at("at://did:plc:repeat/org.tangled.feed.star/older");
        let newest = at("at://did:plc:repeat/org.tangled.feed.star/newest");
        store.add(star_edge_at(older.clone(), subject.clone(), 1_000));
        store.add(star_edge_at(newest.clone(), subject.clone(), 2_000));

        let listed = || {
            store
                .list_distinct_authors(
                    &key,
                    PageCursor::Start,
                    limit(PageLimit::MAX),
                    SortDir::Desc,
                )
                .items
                .into_iter()
                .map(|item| item.uri)
                .collect::<HashSet<_>>()
        };
        assert_eq!(store.count(&key), 257);
        assert_eq!(store.count_distinct_authors(&key), 256);
        assert!(listed().contains(&newest));
        assert!(!listed().contains(&older));

        store.remove_source(&newest);
        assert!(listed().contains(&older));

        store.add(star_edge_at(newest.clone(), subject, 3_000));
        store.remove_source(&older);
        assert!(listed().contains(&newest));
        assert_eq!(store.count_distinct_authors(&key), 256);
    }

    #[test]
    fn author_history_promotes_and_shrinks_without_losing_the_newest() {
        let store = store();
        let subject = did("did:plc:large-subject");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject.clone()),
        );
        (0..255).for_each(|index| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:author{index}/org.tangled.feed.star/r{index}"
                )),
                subject.clone(),
                index as u64,
            ));
        });
        let repeated = (0..6)
            .map(|index| {
                at(&format!(
                    "at://did:plc:repeat/org.tangled.feed.star/r{index}"
                ))
            })
            .collect::<Vec<_>>();
        repeated.iter().enumerate().for_each(|(index, source)| {
            store.add(star_edge_at(
                source.clone(),
                subject.clone(),
                1_000 + index as u64,
            ));
        });
        let newest = || {
            store
                .list_distinct_authors(&key, PageCursor::Start, limit(1), SortDir::Desc)
                .items[0]
                .uri
                .clone()
        };

        repeated
            .iter()
            .enumerate()
            .rev()
            .for_each(|(index, source)| {
                assert_eq!(newest(), *source);
                store.remove_source(source);
                assert_eq!(
                    store.count_distinct_authors(&key),
                    255 + u64::from(index != 0),
                );
            });
    }

    #[test]
    fn offset_pagination_matches_reference_across_small_and_large() {
        [64usize, 1000].into_iter().for_each(|n| {
            let store = store();
            let key = fill_subject(&store, n);

            let asc = reference(0..n);
            let desc: Vec<String> = asc.iter().rev().cloned().collect();

            [3u32, 7, 50].into_iter().for_each(|page| {
                assert_eq!(
                    paginate_by_offsets(&store, &key, SortDir::Asc, limit(page)),
                    asc,
                    "asc offset walk mismatch n={n} page={page}"
                );
                assert_eq!(
                    paginate_by_offsets(&store, &key, SortDir::Desc, limit(page)),
                    desc,
                    "desc offset walk mismatch n={n} page={page}"
                );
            });

            for o in [1u64, 5, 63, 255, 256, 257, 999] {
                let o = o.min(n as u64 - 1);
                let page = store.list(
                    &key,
                    PageStart::Offset(PageOffset::from(o)),
                    limit(10),
                    SortDir::Asc,
                );
                let want: Vec<String> = asc.iter().skip(o as usize).take(10).cloned().collect();
                let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect();
                assert_eq!(got, want, "asc offset {o} mismatch n={n}");
                assert_eq!(page.total, Some(TotalCount::from(n as u64)), "total n={n}");
            }

            let page = store.list(
                &key,
                PageStart::Offset(PageOffset::from(n as u64)),
                limit(10),
                SortDir::Asc,
            );
            assert!(page.items.is_empty(), "offset==len must be empty n={n}");
            assert!(page.next.is_none());
            assert_eq!(page.total, Some(TotalCount::from(n as u64)));
        });
    }

    #[test]
    fn offset_page_cursor_continues_sequentially() {
        let n = 1000usize;
        let store = store();
        let key = fill_subject(&store, n);
        let asc = reference(0..n);

        let first = store.list(
            &key,
            PageStart::Offset(PageOffset::new(600)),
            limit(7),
            SortDir::Asc,
        );
        let got: Vec<String> = std::iter::successors(Some(first), |prev| {
            prev.next
                .map(|tok| store.list(&key, PageCursor::After(tok), limit(7), SortDir::Asc))
        })
        .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned()))
        .collect();
        assert_eq!(got, asc.into_iter().skip(600).collect::<Vec<_>>());
    }

    #[test]
    fn offset_desc_matches_cursor_desc_small_bucket() {
        let store = store();
        let key = fill_subject(&store, 5);
        let desc: Vec<String> = reference(0..5).into_iter().rev().collect();

        let first = store.list(
            &key,
            PageStart::Offset(PageOffset::new(2)),
            limit(3),
            SortDir::Desc,
        );
        let got: Vec<String> = std::iter::successors(Some(first), |prev| {
            prev.next
                .map(|tok| store.list(&key, PageCursor::After(tok), limit(3), SortDir::Desc))
        })
        .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned()))
        .collect();
        assert_eq!(got, desc.into_iter().skip(2).collect::<Vec<_>>());
    }

    #[test]
    fn offset_desc_matches_cursor_desc_large_bucket() {
        let n = 1000usize;
        let store = store();
        let key = fill_subject(&store, n);
        let desc: Vec<String> = reference(0..n).into_iter().rev().collect();

        let first = store.list(
            &key,
            PageStart::Offset(PageOffset::new(600)),
            limit(7),
            SortDir::Desc,
        );
        let got: Vec<String> = std::iter::successors(Some(first), |prev| {
            prev.next
                .map(|tok| store.list(&key, PageCursor::After(tok), limit(7), SortDir::Desc))
        })
        .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned()))
        .collect();
        assert_eq!(got, desc.into_iter().skip(600).collect::<Vec<_>>());
    }

    #[test]
    fn offset_boundary_matrix_asc_and_desc() {
        // offsets {0, 1, len-1, len} across asc and desc pin resolve_start pred_rank math
        for &n in &[5usize, 300] {
            let store = store();
            let key = fill_subject(&store, n);
            let asc = reference(0..n);
            let desc: Vec<String> = asc.iter().rev().cloned().collect();

            for &(dir, ref_order, label) in
                &[(SortDir::Asc, &asc, "asc"), (SortDir::Desc, &desc, "desc")]
            {
                let p = store.list(
                    &key,
                    PageStart::Offset(PageOffset::ZERO),
                    limit(n as u32),
                    dir,
                );
                let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect();
                assert_eq!(got, ref_order[..n].to_vec(), "n={n} {label} offset=0");

                let p = store.list(
                    &key,
                    PageStart::Offset(PageOffset::new(1)),
                    limit(n as u32),
                    dir,
                );
                let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect();
                assert_eq!(got, ref_order[1..].to_vec(), "n={n} {label} offset=1");

                let p = store.list(
                    &key,
                    PageStart::Offset(PageOffset::from(n as u64 - 1)),
                    limit(n as u32),
                    dir,
                );
                let got: Vec<String> = p.items.iter().map(|u| u.as_ref().to_owned()).collect();
                assert_eq!(
                    got,
                    vec![ref_order[n - 1].clone()],
                    "n={n} {label} offset=len-1"
                );
                assert!(p.next.is_none(), "last item has no continuation");

                let p = store.list(
                    &key,
                    PageStart::Offset(PageOffset::from(n as u64)),
                    limit(n as u32),
                    dir,
                );
                assert!(p.items.is_empty(), "n={n} {label} offset=len is empty");
                assert_eq!(
                    p.total,
                    Some(TotalCount::from(n as u64)),
                    "past-end page still reports total"
                );
            }
        }
    }
    #[test]
    fn descending_cursor_block_boundary_edges() {
        let store = store();
        let n = 600usize;
        let key = fill_subject(&store, n);
        let asc = reference(0..n);
        let desc: Vec<String> = asc.iter().rev().cloned().collect();

        for target_idx in [255usize, 256, 257, 511, 512] {
            let p_single = store.list(
                &key,
                PageStart::Offset(PageOffset::from(target_idx as u64)),
                limit(1),
                SortDir::Asc,
            );
            let tok = p_single.next.unwrap();
            let page = store.list(
                &key,
                PageStart::Cursor(PageCursor::After(tok)),
                limit(600),
                SortDir::Desc,
            );
            let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect();
            let expected = &desc[n - target_idx..];
            assert_eq!(got, expected, "descending cursor after index {target_idx}");
        }

        let page = store.list(
            &key,
            PageStart::Offset(PageOffset::new(256)),
            limit(10),
            SortDir::Desc,
        );
        let got: Vec<String> = page.items.iter().map(|u| u.as_ref().to_owned()).collect();
        assert_eq!(got, desc[256..266].to_vec(), "descending offset 256");
    }
    #[test]
    fn empty_store_descending_after_cursor_does_not_underflow() {
        let store = store();
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(did("did:plc:empty_test")),
        );
        let tok = PageToken::new(12345, 1);
        let page = store.list(
            &key,
            PageStart::Cursor(PageCursor::After(tok)),
            limit(10),
            SortDir::Desc,
        );
        assert!(page.items.is_empty());
    }

    #[test]
    fn property_random_offsets_match_reference() {
        // seeded random inserts with duplicate attempts swept across offsets
        fn xorshift(mut x: u64) -> u64 {
            x ^= x << 13;
            x ^= x >> 7;
            x ^= x << 17;
            x
        }

        let store = store();
        let key = EdgeKey::new(
            nsid("sh.tangled.repo.issue"),
            SubjectRef::Did(did("did:plc:prop")),
        );

        let mut rng = 0xdead_beef_cafe_babe_u64;
        let mut order: Vec<usize> = (0..500).collect();
        for i in (1..order.len()).rev() {
            rng = xorshift(rng);
            let j = rng as usize % (i + 1);
            order.swap(i, j);
        }
        for _ in 0..100 {
            rng = xorshift(rng);
            order.push(rng as usize % 500);
        }

        for &i in &order {
            store.add(Edge {
                kind: nsid("sh.tangled.repo.issue"),
                subject: SubjectRef::Did(did("did:plc:prop")),
                source: at(&format!("at://did:plc:prop/sh.tangled.repo.issue/i{i}")),
                sort_micros: (i as u64 + 1) * 1000,
            });
        }

        let asc: Vec<String> = (0..500)
            .map(|i| format!("at://did:plc:prop/sh.tangled.repo.issue/i{i}"))
            .collect();
        let desc: Vec<String> = asc.iter().rev().cloned().collect();
        let n = 500usize;

        let mut offsets: Vec<u64> = (0..n as u64).step_by(50).collect();
        offsets.extend([n as u64 - 1, n as u64]);
        offsets.sort();
        offsets.dedup();

        for o in offsets {
            for &(dir, ref_order, label) in
                &[(SortDir::Asc, &asc, "asc"), (SortDir::Desc, &desc, "desc")]
            {
                let first =
                    store.list(&key, PageStart::Offset(PageOffset::from(o)), limit(25), dir);
                let got: Vec<String> = std::iter::successors(Some(first), |prev| {
                    prev.next
                        .map(|tok| store.list(&key, PageCursor::After(tok), limit(25), dir))
                })
                .flat_map(|p| p.items.into_iter().map(|u| u.as_ref().to_owned()))
                .collect();
                let expected = if (o as usize) < n {
                    ref_order[o as usize..].to_vec()
                } else {
                    vec![]
                };
                assert_eq!(got, expected, "random n=500 {label} offset={o}");
            }
        }
    }

    #[test]
    fn block_store_matches_sorted_vec_oracle() {
        let mut blocks = BlockStore::from_sorted(Vec::new());
        let mut oracle: Vec<BucketKey> = Vec::new();
        let key_at = |i: usize| BucketKey::new(shuffled_micros(i), SourceId::from_raw(i as u32));

        let mut inserted: Vec<usize> = (0..1500).collect();
        inserted.sort_by_key(|i| shuffled_micros(*i ^ 0x5a5a));
        for i in inserted {
            let key = key_at(i);
            assert!(blocks.insert(key), "insert {i}");
            let pos = oracle.partition_point(|k| *k < key);
            oracle.insert(pos, key);
        }
        assert_eq!(blocks.len(), oracle.len());

        // remove every third key in a different shuffled order
        let mut removed: Vec<BucketKey> = oracle.iter().step_by(3).copied().collect();
        removed.sort_by_key(|k| shuffled_micros(k.source.index() as usize ^ 0x33));
        for key in &removed {
            assert!(blocks.remove(key), "remove present key");
            let pos = oracle.partition_point(|k| *k < *key);
            oracle.remove(pos);
        }
        assert_eq!(blocks.len(), oracle.len());
        assert!(!blocks.remove(&removed[0]), "double remove must be false");

        for rank in [
            0usize,
            1,
            255,
            256,
            300,
            511,
            oracle.len() / 2,
            oracle.len() - 1,
        ] {
            assert_eq!(
                blocks.select(PageRank::new(rank)),
                Some(oracle[rank]),
                "select rank {rank}"
            );
        }
        assert_eq!(
            blocks.select(PageRank::new(oracle.len())),
            None,
            "select past end"
        );

        let asc: Vec<BucketKey> = blocks.directed(PageCursor::Start, SortDir::Asc).collect();
        assert_eq!(asc, oracle, "asc full iteration");
        let desc: Vec<BucketKey> = blocks.directed(PageCursor::Start, SortDir::Desc).collect();
        assert_eq!(
            desc,
            oracle.iter().rev().copied().collect::<Vec<_>>(),
            "desc full iteration"
        );

        // cursor-resume from a mid-block token in both directions
        let pivot = oracle[oracle.len() / 2];
        let after_asc: Vec<BucketKey> = blocks
            .directed(PageCursor::After(pivot.token()), SortDir::Asc)
            .collect();
        let want_asc: Vec<BucketKey> = oracle.iter().copied().filter(|k| *k > pivot).collect();
        assert_eq!(after_asc, want_asc, "asc after pivot");
        let after_desc: Vec<BucketKey> = blocks
            .directed(PageCursor::After(pivot.token()), SortDir::Desc)
            .collect();
        let want_desc: Vec<BucketKey> = oracle
            .iter()
            .rev()
            .copied()
            .filter(|k| *k < pivot)
            .collect();
        assert_eq!(after_desc, want_desc, "desc after pivot");
    }

    // Build three subject buckets with interleaved micros so the global time
    // order crosses buckets, and return the three keys.
    fn fill_three_subjects(store: &EdgeStore) -> Vec<EdgeKey> {
        [
            ("alpha", "r0", 10u64),
            ("bravo", "r1", 50),
            ("charlie", "r2", 20),
            ("alpha", "r3", 40),
            ("bravo", "r4", 30),
            ("charlie", "r5", 60),
        ]
        .into_iter()
        .map(|(subj, rkey, micros)| {
            star_edge_at(
                at(&format!("at://did:plc:x/org.tangled.feed.star/{rkey}")),
                did(&format!("did:plc:{subj}")),
                micros,
            )
        })
        .for_each(|edge| store.add(edge));
        ["alpha", "bravo", "charlie"]
            .iter()
            .map(|s| {
                EdgeKey::new(
                    nsid("org.tangled.feed.star"),
                    did_subj(&format!("did:plc:{s}")),
                )
            })
            .collect()
    }

    #[test]
    fn list_multi_merges_subjects_newest_first() {
        let store = store();
        let keys = fill_three_subjects(&store);
        let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc);
        let micros: Vec<u64> = page.items.iter().map(|it| it.sort_micros).collect();
        assert_eq!(micros, vec![60, 50, 40, 30, 20, 10]);
        assert!(page.next.is_none());
    }

    #[test]
    fn list_multi_paginates_disjoint_across_buckets() {
        let store = store();
        let keys = fill_three_subjects(&store);
        // limit=4 with 6 items across 3 buckets exercises the bounded top-N merge.
        let p1 = store.list_multi(&keys, PageCursor::Start, limit(4), SortDir::Desc);
        let m1: Vec<u64> = p1.items.iter().map(|it| it.sort_micros).collect();
        assert_eq!(m1, vec![60, 50, 40, 30]);
        let tok = p1.next.expect("first page has more");
        let p2 = store.list_multi(&keys, PageCursor::After(tok), limit(4), SortDir::Desc);
        let m2: Vec<u64> = p2.items.iter().map(|it| it.sort_micros).collect();
        assert_eq!(m2, vec![20, 10]);
        assert!(p2.next.is_none());
    }

    #[test]
    fn list_multi_dedups_source_under_two_keys() {
        let store = store();
        // Same record (source) indexed under two subject keys, same createdAt.
        let shared = at("at://did:plc:x/org.tangled.feed.star/shared");
        store.add(star_edge_at(shared.clone(), did("did:plc:alpha"), 100));
        store.add(star_edge_at(shared, did("did:plc:bravo"), 100));
        let keys = vec![
            EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:alpha")),
            EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:bravo")),
        ];
        let page = store.list_multi(&keys, PageCursor::Start, limit(10), SortDir::Desc);
        assert_eq!(page.items.len(), 1);
        assert!(page.next.is_none());
    }

    #[test]
    fn remove_from_large_bucket_keeps_pagination_exact() {
        let store = store();
        let n = 1000usize;
        let key = fill_subject(&store, n);
        (0..n).step_by(3).for_each(|i| {
            store.remove_source(&at(&source_uri(i)));
        });
        let expected = reference((0..n).filter(|i| i % 3 != 0));
        assert_eq!(store.count(&key), expected.len() as u64);
        assert_eq!(
            store.count_distinct_authors(&key),
            NAMES.len() as u64,
            "every author keeps sources after partial removal"
        );
        assert_eq!(paginate_all(&store, &key, SortDir::Asc, limit(7)), expected);
        assert_eq!(
            paginate_all(&store, &key, SortDir::Desc, limit(11)),
            expected.iter().rev().cloned().collect::<Vec<_>>()
        );
    }

    #[test]
    fn add_then_count() {
        let store = store();
        let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));

        store.add(star_edge(
            at("at://did:plc:nel/org.tangled.feed.star/r1"),
            did("did:plc:abalone"),
        ));
        store.add(star_edge(
            at("at://did:plc:olaren/org.tangled.feed.star/r2"),
            did("did:plc:abalone"),
        ));
        store.add(star_edge(
            at("at://did:plc:nel/org.tangled.feed.star/r3"),
            did("did:plc:abalone"),
        ));

        assert_eq!(store.count(&key), 3);
        assert_eq!(store.count_distinct_authors(&key), 2);
    }

    #[test]
    fn duplicate_add_is_idempotent() {
        let store = store();
        let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
        let edge = star_edge(
            at("at://did:plc:nel/org.tangled.feed.star/r1"),
            did("did:plc:abalone"),
        );
        store.add(edge.clone());
        store.add(edge);
        assert_eq!(store.count(&key), 1);
        assert_eq!(store.count_distinct_authors(&key), 1);
    }

    #[test]
    fn remove_source_clears_all_keys_for_that_source() {
        let store = store();
        let star_key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
        let follow_key = EdgeKey::new(nsid("org.tangled.graph.follow"), did_subj("did:plc:lyna"));
        let source = "at://did:plc:nel/org.tangled.feed.star/r1";

        store.add(Edge {
            kind: nsid("org.tangled.feed.star"),
            subject: did_subj("did:plc:abalone"),
            source: at(source),
            sort_micros: 0,
        });
        store.add(Edge {
            kind: nsid("org.tangled.graph.follow"),
            subject: did_subj("did:plc:lyna"),
            source: at(source),
            sort_micros: 0,
        });

        assert_eq!(store.count(&star_key), 1);
        assert_eq!(store.count(&follow_key), 1);

        store.remove_source(&at(source));
        assert_eq!(store.count(&star_key), 0);
        assert_eq!(store.count(&follow_key), 0);
    }

    #[test]
    fn upsert_source_replaces_old_edges() {
        let store = store();
        let source = at("at://did:plc:teq/org.tangled.feed.star/r1");
        let old_subject = did_subj("did:plc:abalone");
        let new_subject = did_subj("did:plc:uni");
        let kind = nsid("org.tangled.feed.star");

        store.upsert_source(
            &source,
            vec![Edge {
                kind: kind.clone(),
                subject: old_subject.clone(),
                source: source.clone(),
                sort_micros: 0,
            }],
        );
        assert_eq!(
            store.count(&EdgeKey::new(kind.clone(), old_subject.clone())),
            1
        );

        store.upsert_source(
            &source,
            vec![Edge {
                kind: kind.clone(),
                subject: new_subject.clone(),
                source: source.clone(),
                sort_micros: 0,
            }],
        );
        assert_eq!(store.count(&EdgeKey::new(kind.clone(), old_subject)), 0);
        assert_eq!(store.count(&EdgeKey::new(kind, new_subject)), 1);
    }

    fn seed_tickets(store: &EdgeStore, repo: &str, micros: &[u64]) -> Vec<AtUri<DefaultStr>> {
        let kind = nsid(TICKET_ISSUE_KIND);
        micros
            .iter()
            .map(|micros| {
                let ticket = at(&format!("at://{repo}/org.tangled.track.ticket/{micros}"));
                store.upsert_source(
                    &ticket,
                    vec![Edge {
                        kind: kind.clone(),
                        subject: SubjectRef::Did(did(repo)),
                        source: ticket.clone(),
                        sort_micros: *micros,
                    }],
                );
                store.refresh_ticket_counts(&ticket, TicketStateKind::Open);
                ticket
            })
            .collect()
    }

    #[test]
    fn state_view_pages_default_open_by_rank_with_total() {
        let store = store();
        let key = EdgeKey::new(
            nsid(TICKET_ISSUE_KIND),
            SubjectRef::Did(did("did:plc:limpet")),
        );
        let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2, 3, 4, 5]);

        let page = store.page_ticket_state(
            &key,
            TicketStateKind::Open,
            PageStart::Offset(PageOffset::new(2)),
            limit(2),
            SortDir::Asc,
        );
        assert_eq!(page.total, Some(TotalCount::new(5)),);
        assert_eq!(
            page.items
                .iter()
                .map(|i| i.uri.as_ref())
                .collect::<Vec<_>>(),
            vec![issues[2].as_ref(), issues[3].as_ref()],
        );
        let follow = store.page_ticket_state(
            &key,
            TicketStateKind::Open,
            PageStart::Cursor(PageCursor::After(page.next.unwrap())),
            limit(2),
            SortDir::Asc,
        );
        assert_eq!(follow.items.len(), 1);
        assert_eq!(follow.items[0].uri, issues[4]);
    }

    #[test]
    fn state_view_transitions_repage_across_kinds() {
        let store = store();
        let key = EdgeKey::new(
            nsid(TICKET_ISSUE_KIND),
            SubjectRef::Did(did("did:plc:limpet")),
        );
        let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2]);

        store.refresh_ticket_counts(&issues[0], TicketStateKind::Closed);

        let open = store.page_ticket_state(
            &key,
            TicketStateKind::Open,
            PageStart::Offset(PageOffset::ZERO),
            limit(10),
            SortDir::Asc,
        );
        assert_eq!(open.total, Some(TotalCount::new(1)));
        assert_eq!(open.items[0].uri, issues[1]);
        let closed = store.page_ticket_state(
            &key,
            TicketStateKind::Closed,
            PageStart::Offset(PageOffset::ZERO),
            limit(10),
            SortDir::Asc,
        );
        assert_eq!(closed.total, Some(TotalCount::new(1)));
        assert_eq!(closed.items[0].uri, issues[0]);

        store.refresh_ticket_counts(&issues[1], TicketStateKind::Completed);
        let completed = store.count_ticket_state(&key, TicketStateKind::Completed);
        assert_eq!(completed.count.get(), 1);
        let open = store.count_ticket_state(&key, TicketStateKind::Open);
        assert_eq!(open.count.get(), 0);
    }

    #[test]
    fn state_view_offset_desc_matches_cursor_desc() {
        // views resolve offsets through SortedBlocks::resolve_start directly:
        // issues i1..i5 ascending page as i5 i4 i3 i2 i1 descending, offset 2
        // starts at i3
        let store = store();
        let key = EdgeKey::new(
            nsid(TICKET_ISSUE_KIND),
            SubjectRef::Did(did("did:plc:limpet")),
        );
        let issues = seed_tickets(&store, "did:plc:limpet", &[1, 2, 3, 4, 5]);

        let page = store.page_ticket_state(
            &key,
            TicketStateKind::Open,
            PageStart::Offset(PageOffset::new(2)),
            limit(2),
            SortDir::Desc,
        );
        assert_eq!(page.total, Some(TotalCount::new(5)));
        assert_eq!(
            page.items
                .iter()
                .map(|i| i.uri.as_ref())
                .collect::<Vec<_>>(),
            vec![issues[2].as_ref(), issues[1].as_ref()],
            "desc offset starts at the third-newest",
        );

        let tail = store.page_ticket_state(
            &key,
            TicketStateKind::Open,
            PageCursor::After(page.next.unwrap()),
            limit(5),
            SortDir::Desc,
        );
        assert_eq!(
            tail.items
                .iter()
                .map(|i| i.uri.as_ref())
                .collect::<Vec<_>>(),
            vec![issues[0].as_ref()],
            "cursor continuation finishes the desc walk",
        );
        assert!(tail.next.is_none());
    }

    #[test]
    fn record_index_mutations_move_materialized_state_counts() {
        let store = store();
        let ticket = at("at://did:plc:limpet/org.tangled.track.ticket/1");
        let nsid_ticket = nsid("org.tangled.track.ticket");
        let repo = || SubjectRef::Did(did("did:plc:limpet"));
        let issues = EdgeKey::new(nsid(TICKET_ISSUE_KIND), repo());
        let pulls = EdgeKey::new(nsid(TICKET_PULL_KIND), repo());
        let upsert = |state: &str, target: &str| {
            let body = format!(
                r#"{{"$type":"org.tangled.track.ticket","title":{{"text":"bug"}},"state":"{state}",{target}"createdAt":"2026-05-01T00:00:00Z"}}"#
            );
            let record = Record::from_json_bytes(&nsid_ticket, body.as_bytes()).unwrap();
            let edges = record.extract_edges(&ticket).unwrap();
            upsert_record_indexes(&store, &ClaimIndex::new(), &ticket, edges, &record);
        };
        let count = |key, kind| store.count_ticket_state(key, kind).count.get();
        use TicketStateKind::{Closed, Completed, Open};

        upsert("closed", "");
        assert_eq!((count(&issues, Open), count(&issues, Closed)), (0, 1));
        upsert("open", "");
        assert_eq!((count(&issues, Open), count(&issues, Closed)), (1, 0));

        // gaining a target branch moves the ticket from issues to pulls
        upsert("completed", r#""targetBranch":"main","#);
        assert_eq!(count(&issues, Open), 0);
        assert_eq!(count(&pulls, Completed), 1);

        delete_record_indexes(&store, &ClaimIndex::new(), &ticket, &nsid_ticket);
        assert_eq!(count(&pulls, Completed), 0);
    }

    #[test]
    fn list_pages_in_sort_order() {
        let store = store();
        let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
        (0..5).for_each(|i| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                did("did:plc:abalone"),
                1_000_000 + i as u64 * 1_000_000,
            ));
        });

        let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
        assert_eq!(page1.items.len(), 2);
        let cursor = PageCursor::After(page1.next.expect("has next"));

        let page2 = store.list(&key, cursor, limit(2), SortDir::Asc);
        assert_eq!(page2.items.len(), 2);

        let cursor2 = PageCursor::After(page2.next.expect("has next"));
        let page3 = store.list(&key, cursor2, limit(2), SortDir::Asc);
        assert_eq!(page3.items.len(), 1);
        assert!(
            page3.next.is_none(),
            "final partial page should signal exhaustion"
        );
    }

    #[test]
    fn list_exact_fill_signals_exhaustion() {
        let store = store();
        let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
        (0..2).for_each(|i| {
            store.add(star_edge(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                did("did:plc:abalone"),
            ));
        });

        let page = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
        assert_eq!(page.items.len(), 2);
        assert!(page.next.is_none(), "exact-fill page must not promise more");
    }

    #[test]
    fn list_on_unknown_key_is_empty() {
        let store = store();
        let page = store.list(
            &EdgeKey::new(
                nsid("org.tangled.feed.star"),
                did_subj("did:plc:periwinkle"),
            ),
            PageCursor::Start,
            limit(10),
            SortDir::Asc,
        );
        assert!(page.items.is_empty());
        assert!(page.next.is_none());
    }

    #[test]
    fn distinct_authors_decreases_when_last_source_from_author_removed() {
        let store = store();
        let key = EdgeKey::new(nsid("org.tangled.feed.star"), did_subj("did:plc:abalone"));
        let s1 = at("at://did:plc:nel/org.tangled.feed.star/r1");
        let s2 = at("at://did:plc:nel/org.tangled.feed.star/r2");
        let s3 = at("at://did:plc:olaren/org.tangled.feed.star/r3");

        store.add(star_edge(s1.clone(), did("did:plc:abalone")));
        store.add(star_edge(s2.clone(), did("did:plc:abalone")));
        store.add(star_edge(s3, did("did:plc:abalone")));
        assert_eq!(store.count_distinct_authors(&key), 2);

        store.remove_source(&s1);
        assert_eq!(store.count_distinct_authors(&key), 2, "user1 still has s2");

        store.remove_source(&s2);
        assert_eq!(store.count_distinct_authors(&key), 1, "user1 fully gone");
    }

    #[test]
    fn packed_sort_micros_preserves_u64_ordering_and_cursor_values() {
        let values = [
            0,
            u32::MAX as u64,
            u32::MAX as u64 + 1,
            1_730_000_000_000_000,
            u64::MAX,
        ];
        let keys = values.map(|micros| BucketKey::new(micros, SourceId::from_raw(7)));
        assert!(keys.windows(2).all(|pair| pair[0] < pair[1]));
        keys.into_iter().zip(values).for_each(|(key, micros)| {
            assert_eq!(key.micros.get(), micros);
            assert_eq!(key.token(), PageToken::new(micros, 7));
        });
    }

    #[test]
    fn author_index_separates_and_rejoins_history() {
        let author = AuthorId::from_raw(1);
        let older = BucketKey::new(10, SourceId::from_raw(10));
        let middle = BucketKey::new(20, SourceId::from_raw(20));
        let newest = BucketKey::new(30, SourceId::from_raw(30));
        let mut index = AuthorIndex::new(RuntimeHasher::default());

        index.insert(author, newest);
        assert_eq!(index.newest[&author], newest);
        assert!(index.histories.is_empty());

        index.insert(author, older);
        index.insert(author, middle);
        assert_eq!(index.newest[&author], newest);
        assert_eq!(index.histories[&author].len(), 3);

        index.remove(author, &older);
        assert_eq!(index.histories[&author].len(), 2);
        assert_eq!(index.newest[&author], newest);

        index.remove(author, &newest);
        assert!(index.histories.is_empty());
        assert_eq!(index.newest[&author], middle);
        assert_eq!(index.representatives.last(), Some(middle));

        index.remove(author, &middle);
        assert!(index.newest.is_empty());
        assert!(index.representatives.last().is_none());
    }

    #[test]
    fn page_limit_rejects_zero_and_oversize() {
        assert!(matches!(
            PageLimit::new(0),
            Err(PageLimitError::TooSmall { value: 0, min: 1 })
        ));
        assert!(matches!(
            PageLimit::new(PageLimit::MAX + 1),
            Err(PageLimitError::TooLarge { .. })
        ));
        assert_eq!(PageLimit::new(50).unwrap().get(), 50);
    }

    #[test]
    fn cursor_token_round_trip() {
        let original = PageToken::new(1_730_000_000_000_000, 0x1234_abcd);
        let token = original.encode_token();
        assert_eq!(token.len(), 24, "12-byte cursor encodes to 24 hex chars");
        assert_eq!(PageToken::decode_token(&token).unwrap(), original);
    }

    #[test]
    fn from_token_none_yields_start() {
        assert_eq!(PageCursor::from_token(None).unwrap(), PageCursor::Start);
    }

    #[test]
    fn from_token_some_yields_after() {
        let original = PageToken::new(1_730_000_000_000_000, 42);
        let token = original.encode_token();
        assert_eq!(
            PageCursor::from_token(Some(&token)).unwrap(),
            PageCursor::After(original),
        );
    }

    #[test]
    fn cursor_decode_rejects_malformed() {
        let bad = [
            "",
            "deadbeef",
            "no-hex!!aaaaaaaaaaaaaaaa",
            "12345",
            "this-string-is-way-too-long-to-be-a-valid-cursor",
        ];
        bad.into_iter().for_each(|s| {
            assert!(
                matches!(PageToken::decode_token(s), Err(CursorParseError::Malformed)),
                "expected malformed for {s:?}",
            );
        });
    }

    #[test]
    fn cursor_token_is_hex_shape() {
        let token = PageToken::new(1_730_000_000_000_000, 7).encode_token();
        assert_eq!(token.len(), 24);
        assert!(token.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn list_does_not_drop_entries_at_same_sort_micros_across_pages() {
        let store = store();
        let subject_did = did("did:plc:limpet");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject_did.clone()),
        );
        (0..4).for_each(|i| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                subject_did.clone(),
                42,
            ));
        });

        let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Asc);
        assert_eq!(page1.items.len(), 2);
        let token = page1.next.expect("cursor must continue across ties");

        let page2 = store.list(&key, PageCursor::After(token), limit(2), SortDir::Asc);
        assert_eq!(
            page2.items.len(),
            2,
            "remaining ties must surface on next page"
        );
        assert!(page2.next.is_none());
        let combined: std::collections::HashSet<String> = page1
            .items
            .iter()
            .chain(page2.items.iter())
            .map(|u| u.as_ref().to_owned())
            .collect();
        assert_eq!(combined.len(), 4, "every tied entry visible exactly once");
    }

    #[test]
    fn list_descending_returns_newest_first() {
        let store = store();
        let subject_did = did("did:plc:limpet");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject_did.clone()),
        );
        (0..5).for_each(|i| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                subject_did.clone(),
                1_000_000 + i as u64 * 1_000_000,
            ));
        });

        let asc = store.list(&key, PageCursor::Start, limit(5), SortDir::Asc);
        let desc = store.list(&key, PageCursor::Start, limit(5), SortDir::Desc);
        assert_eq!(asc.items.len(), 5);
        assert_eq!(desc.items.len(), 5);
        let asc_uris: Vec<_> = asc.items.iter().map(|u| u.as_ref().to_owned()).collect();
        let mut reversed = asc_uris.clone();
        reversed.reverse();
        let desc_uris: Vec<_> = desc.items.iter().map(|u| u.as_ref().to_owned()).collect();
        assert_eq!(desc_uris, reversed, "desc must be exact reverse of asc");
    }

    #[test]
    fn list_descending_paginates_with_cursor() {
        let store = store();
        let subject_did = did("did:plc:whelk");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject_did.clone()),
        );
        (0..5).for_each(|i| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                subject_did.clone(),
                1_000_000 + i as u64 * 1_000_000,
            ));
        });

        let page1 = store.list(&key, PageCursor::Start, limit(2), SortDir::Desc);
        assert_eq!(page1.items.len(), 2);
        let cursor = PageCursor::After(page1.next.expect("desc page1 must continue"));
        let page2 = store.list(&key, cursor, limit(2), SortDir::Desc);
        assert_eq!(page2.items.len(), 2);
        let cursor2 = PageCursor::After(page2.next.expect("desc page2 must continue"));
        let page3 = store.list(&key, cursor2, limit(2), SortDir::Desc);
        assert_eq!(page3.items.len(), 1);
        assert!(page3.next.is_none());

        let combined: std::collections::HashSet<String> = page1
            .items
            .iter()
            .chain(page2.items.iter())
            .chain(page3.items.iter())
            .map(|u| u.as_ref().to_owned())
            .collect();
        assert_eq!(
            combined.len(),
            5,
            "desc pagination visits each item exactly once"
        );
    }

    #[test]
    fn non_did_source_round_trips_via_raw_fallback() {
        let store = store();
        let subject = did("did:plc:limpet");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject.clone()),
        );
        let did_source = at("at://did:plc:nel/org.tangled.feed.star/r1");
        let handle_source = at("at://witchcraft.systems/org.tangled.feed.star/r2");
        store.add(star_edge_at(did_source.clone(), subject.clone(), 1));
        store.add(star_edge_at(handle_source.clone(), subject.clone(), 2));

        let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc);
        let got: std::collections::HashSet<String> =
            page.items.iter().map(|u| u.as_ref().to_owned()).collect();
        assert!(
            got.contains(did_source.as_ref()),
            "did source must decode exactly"
        );
        assert!(
            got.contains(handle_source.as_ref()),
            "non-did authority must round-trip through the raw fallback"
        );

        store.remove_source(&handle_source);
        assert_eq!(
            store.count(&key),
            1,
            "raw-keyed source removable by its uri"
        );
    }

    #[test]
    fn distinct_collections_decode_with_their_own_collection() {
        let store = store();
        let subject = did("did:plc:limpet");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject.clone()),
        );
        let star_src = at("at://did:plc:nel/org.tangled.feed.star/aaa");
        let issue_src = at("at://did:plc:nel/sh.tangled.repo.issue/bbb");
        store.add(Edge {
            kind: nsid("org.tangled.feed.star"),
            subject: SubjectRef::Did(subject.clone()),
            source: star_src.clone(),
            sort_micros: 1,
        });
        store.add(Edge {
            kind: nsid("org.tangled.feed.star"),
            subject: SubjectRef::Did(subject.clone()),
            source: issue_src.clone(),
            sort_micros: 2,
        });

        let page = store.list(&key, PageCursor::Start, limit(10), SortDir::Asc);
        let got: std::collections::HashSet<String> =
            page.items.iter().map(|u| u.as_ref().to_owned()).collect();
        assert!(
            got.contains(star_src.as_ref()),
            "star-collection source decodes exactly"
        );
        assert!(
            got.contains(issue_src.as_ref()),
            "issue-collection source must keep its own collection, not borrow the star one"
        );
    }

    #[test]
    fn most_followed_ranks_distinct_current_followers() {
        let store = store();
        let alpha = did("did:plc:alpha");
        let beta = did("did:plc:beta");
        let gamma = did("did:plc:gamma");
        [
            ("did:plc:nel", "a1", alpha.clone()),
            ("did:plc:nel", "a2", alpha.clone()),
            ("did:plc:olaren", "a3", alpha.clone()),
            ("did:plc:nel", "b1", beta.clone()),
            ("did:plc:olaren", "b2", beta.clone()),
            ("did:plc:teq", "b3", beta.clone()),
            ("did:plc:nel", "g1", gamma.clone()),
            ("did:plc:olaren", "g2", gamma.clone()),
        ]
        .into_iter()
        .enumerate()
        .for_each(|(index, (author, rkey, subject))| {
            store.add(follow_edge_at(
                at(&format!("at://{author}/org.tangled.graph.follow/{rkey}")),
                subject,
                index as u64,
            ));
        });

        assert_eq!(store.most_followed(3), vec![beta, alpha, gamma]);
    }

    #[test]
    fn top_starred_since_counts_distinct_recent_starrers() {
        let store = store();
        let recent = did("did:plc:recent");
        let one = did("did:plc:one");
        let old = did("did:plc:old");
        [
            ("did:plc:nel", "r1", recent.clone(), 100),
            ("did:plc:nel", "r2", recent.clone(), 110),
            ("did:plc:olaren", "r3", recent.clone(), 120),
            ("did:plc:teq", "o1", one.clone(), 130),
            ("did:plc:nel", "x1", old.clone(), 1),
            ("did:plc:olaren", "x2", old.clone(), 2),
            ("did:plc:teq", "x3", old.clone(), 3),
        ]
        .into_iter()
        .for_each(|(author, rkey, subject, micros)| {
            store.add(star_edge_at(
                at(&format!("at://{author}/org.tangled.feed.star/{rkey}")),
                subject,
                micros,
            ));
        });

        assert_eq!(store.top_starred_since(100, 5), vec![recent, one]);
        assert_eq!(store.top_starred_since(100, 1).len(), 1);
    }

    #[test]
    fn author_refs_spill_preserves_distinct_count() {
        let store = store();
        let subject = did("did:plc:scallop");
        let key = EdgeKey::new(
            nsid("org.tangled.feed.star"),
            SubjectRef::Did(subject.clone()),
        );
        (0..5).for_each(|i| {
            store.add(star_edge_at(
                at(&format!(
                    "at://did:plc:{}/org.tangled.feed.star/r{i}",
                    NAMES[i]
                )),
                subject.clone(),
                i as u64,
            ));
        });
        assert_eq!(
            store.count_distinct_authors(&key),
            5,
            "five authors exceed the inline cap and spill to the map"
        );

        store.add(star_edge_at(
            at("at://did:plc:nel/org.tangled.feed.star/r99"),
            subject.clone(),
            99,
        ));
        assert_eq!(
            store.count_distinct_authors(&key),
            5,
            "second nel source adds no author"
        );

        store.remove_source(&at("at://did:plc:nel/org.tangled.feed.star/r0"));
        assert_eq!(
            store.count_distinct_authors(&key),
            5,
            "nel still present via r99"
        );
        store.remove_source(&at("at://did:plc:nel/org.tangled.feed.star/r99"));
        assert_eq!(store.count_distinct_authors(&key), 4, "nel fully removed");
    }

    #[test]
    fn source_collections_set_and_clear() {
        let store = store();
        let source = at("at://did:plc:bob/org.tangled.feed.subscription/r1");
        let edge = Edge {
            kind: nsid("org.tangled.feed.subscription"),
            subject: did_subj("did:plc:repo"),
            source: source.clone(),
            sort_micros: 1,
        };
        store.add(edge);

        // Initially no filter
        assert_eq!(store.source_collections_for(&source), None);

        // Set a filter
        store.set_source_collections(
            &source,
            Some(vec![SmolStr::new_static("sh.tangled.repo.issue")]),
        );
        let stored = store.source_collections_for(&source).unwrap();
        assert_eq!(stored, vec![SmolStr::new_static("sh.tangled.repo.issue")]);

        // Update to empty = unrestricted
        store.set_source_collections(&source, Some(vec![]));
        assert_eq!(store.source_collections_for(&source), None);

        // Set a different filter
        store.set_source_collections(
            &source,
            Some(vec![SmolStr::new_static("sh.tangled.repo.pull")]),
        );
        let stored = store.source_collections_for(&source).unwrap();
        assert_eq!(stored, vec![SmolStr::new_static("sh.tangled.repo.pull")]);

        // Set to None = unrestricted
        store.set_source_collections(&source, None);
        assert_eq!(store.source_collections_for(&source), None);
    }

    #[test]
    fn remove_source_cleans_up_collection_filter() {
        let store = store();
        let source = at("at://did:plc:bob/org.tangled.feed.subscription/r1");
        let edge = Edge {
            kind: nsid("org.tangled.feed.subscription"),
            subject: did_subj("did:plc:repo"),
            source: source.clone(),
            sort_micros: 1,
        };
        store.add(edge);
        store.set_source_collections(
            &source,
            Some(vec![SmolStr::new_static("sh.tangled.repo.issue")]),
        );
        assert!(store.source_collections_for(&source).is_some());

        store.remove_source(&source);
        assert_eq!(store.source_collections_for(&source), None);
    }
}