feat: add community ui (#52)

Reviewed-on: #52

crates/community/src/cache.rs

@@ -0,0 +1,331 @@
+use std::collections::BTreeMap;
+use std::sync::LazyLock;
+
+use anyhow::{Result, anyhow};
+use concord::cord01::OpenedStream;
+use concord::state::{CommunityState, STATE_PREFIX, state_identifier};
+use concord::{ChannelId, CommunityId, cord03};
+use nostr_sdk::prelude::*;
+
+static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(|| {
+    Keys::new(SecretKey::from_slice(&[0x43; 32]).expect("a fixed 32-byte scalar is a valid key"))
+});
+
+const CHANNEL_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_C;
+const MARK_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_T;
+const MARK_VALUE: &str = "concord";
+const WRAP_TAG: &str = "e";
+const KIND_TAG: SingleLetterTag = SingleLetterTag::LOWERCASE_K;
+
+/// An already-expired rumor is refused at ingest. Returns whether it was kept.
+pub async fn cache_rumor(
+    client: &Client,
+    channel: &ChannelId,
+    opened: &OpenedStream,
+) -> Result<bool> {
+    let at = Timestamp::from_secs(opened.at_ms / 1000);
+
+    if cord03::expiration_of(&opened.rumor)?
+        .is_some_and(|expiration| expiration <= Timestamp::now())
+    {
+        return Ok(false);
+    }
+
+    let tags = vec![
+        Tag::identifier(opened.rumor_id),
+        Tag::custom(KIND_TAG.as_str(), [opened.rumor.kind.to_string()]),
+        Tag::custom(WRAP_TAG, [opened.wrapper_id.to_string()]),
+        Tag::custom(MARK_TAG.as_str(), [MARK_VALUE]),
+        Tag::custom(CHANNEL_TAG.as_str(), [channel.to_hex()]),
+        Tag::public_key(opened.author),
+    ];
+
+    let event = EventBuilder::new(Kind::ApplicationSpecificData, opened.rumor.as_json())
+        .tags(tags)
+        .custom_created_at(at)
+        .finalize_async(&*LOCAL_KEYS)
+        .await?;
+
+    client.database().save_event(&event).await?;
+
+    Ok(true)
+}
+
+pub async fn purge_expired(client: &Client, channel: &ChannelId, now: Timestamp) -> Result<usize> {
+    let filter = Filter::new()
+        .kind(Kind::ApplicationSpecificData)
+        .custom_tag(MARK_TAG, MARK_VALUE)
+        .custom_tag(CHANNEL_TAG, channel.to_hex());
+
+    let mut expired = Vec::new();
+
+    for event in client.database().query(filter).await? {
+        let Ok(rumor) = UnsignedEvent::from_json(&event.content) else {
+            continue;
+        };
+
+        let Ok(Some(expiration)) = cord03::expiration_of(&rumor) else {
+            continue;
+        };
+
+        if expiration <= now {
+            expired.push(event.id);
+        }
+    }
+
+    let purged = expired.len();
+
+    if purged > 0 {
+        client.database().delete(Filter::new().ids(expired)).await?;
+    }
+
+    Ok(purged)
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Observed {
+    pub author: PublicKey,
+    pub at_ms: u64,
+}
+
+/// The cached rumors of `channel`, keyed by the wrap they were opened from.
+pub async fn wrapper_index(
+    client: &Client,
+    channel: &ChannelId,
+) -> Result<BTreeMap<EventId, Observed>> {
+    let filter = Filter::new()
+        .kind(Kind::ApplicationSpecificData)
+        .custom_tag(MARK_TAG, MARK_VALUE)
+        .custom_tag(CHANNEL_TAG, channel.to_hex());
+
+    let mut index = BTreeMap::new();
+
+    for event in client.database().query(filter).await? {
+        let (Some(wrapper_id), Some(author)) = (
+            event.tags.event_ids().next(),
+            event.tags.public_keys().next(),
+        ) else {
+            continue;
+        };
+
+        index.insert(
+            wrapper_id,
+            Observed {
+                author,
+                at_ms: event.created_at.as_secs().saturating_mul(1000),
+            },
+        );
+    }
+
+    Ok(index)
+}
+
+/// Cached rumors for `channel`, newest first, deduplicated by rumor id.
+pub async fn query_rumors(
+    client: &Client,
+    channel: &ChannelId,
+    until: Option<Timestamp>,
+    limit: usize,
+    kinds: Option<&[u16]>,
+) -> Result<Vec<UnsignedEvent>> {
+    let mut filter = Filter::new()
+        .kind(Kind::ApplicationSpecificData)
+        .custom_tag(MARK_TAG, MARK_VALUE)
+        .custom_tag(CHANNEL_TAG, channel.to_hex());
+
+    if let Some(kinds) = kinds {
+        filter = filter.custom_tags(KIND_TAG, kinds.iter().map(u16::to_string));
+    }
+
+    if let Some(until) = until {
+        filter = filter.until(until);
+    }
+
+    let mut newest: BTreeMap<String, Event> = BTreeMap::new();
+    for event in client.database().query(filter).await? {
+        let Some(rumor_id) = event.tags.identifier() else {
+            continue;
+        };
+
+        match newest.get(&rumor_id) {
+            Some(existing) if existing.created_at >= event.created_at => {}
+            _ => {
+                newest.insert(rumor_id, event);
+            }
+        }
+    }
+
+    let mut events: Vec<Event> = newest.into_values().collect();
+    events.sort_by_key(|event| std::cmp::Reverse(event.created_at));
+    events.truncate(limit);
+
+    let mut rumors = Vec::with_capacity(events.len());
+    for event in events {
+        let rumor = UnsignedEvent::from_json(event.content)
+            .map_err(|error| anyhow!("cached rumor is not a valid event: {error}"))?;
+        rumors.push(rumor);
+    }
+
+    Ok(rumors)
+}
+
+pub async fn save_state(client: &Client, state: &CommunityState) -> Result<()> {
+    let event = EventBuilder::new(Kind::ApplicationSpecificData, serde_json::to_string(state)?)
+        .tags([Tag::identifier(state.identifier())])
+        .finalize_async(&*LOCAL_KEYS)
+        .await?;
+
+    client.database().save_event(&event).await?;
+
+    Ok(())
+}
+
+pub async fn load_state(client: &Client, id: &CommunityId) -> Result<Option<CommunityState>> {
+    let filter = Filter::new()
+        .kind(Kind::ApplicationSpecificData)
+        .identifier(state_identifier(id))
+        .limit(1);
+
+    match client.database().query(filter).await?.into_iter().next() {
+        Some(event) => Ok(Some(serde_json::from_str(&event.content)?)),
+        None => Ok(None),
+    }
+}
+
+/// The newest state document per community carried in the local database.
+pub async fn load_states(client: &Client) -> Result<Vec<CommunityState>> {
+    let filter = Filter::new().kind(Kind::ApplicationSpecificData);
+    let mut newest: BTreeMap<CommunityId, Event> = BTreeMap::new();
+
+    for event in client.database().query(filter).await? {
+        let Some(id) = state_document_of(&event) else {
+            continue;
+        };
+
+        match newest.get(&id) {
+            Some(existing) if existing.created_at >= event.created_at => {}
+            _ => {
+                newest.insert(id, event);
+            }
+        }
+    }
+
+    let mut states = Vec::with_capacity(newest.len());
+
+    for event in newest.into_values() {
+        match serde_json::from_str::<CommunityState>(&event.content) {
+            Ok(state) => states.push(state),
+            Err(error) => log::warn!("ignoring malformed community state {}: {error}", event.id),
+        }
+    }
+
+    Ok(states)
+}
+
+fn state_document_of(event: &Event) -> Option<CommunityId> {
+    let identifier = event.tags.identifier()?;
+    let hex = identifier.strip_prefix(STATE_PREFIX)?;
+    hex.parse().ok()
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeSet;
+
+    use concord::Epoch;
+    use nostr_memory::MemoryDatabase;
+
+    use super::*;
+
+    fn client() -> Client {
+        ClientBuilder::default()
+            .database(MemoryDatabase::unbounded())
+            .build()
+    }
+
+    #[test]
+    fn load_states_reads_one_document_per_community_and_ignores_other_documents() {
+        smol::block_on(async {
+            let client = client();
+
+            let state = CommunityState {
+                id: CommunityId::from_bytes([0x42; 32]),
+                name: Some("Anime and Manga".to_owned()),
+                owner: Keys::generate().public_key(),
+                owner_salt: [0x01; 32],
+                community_root: [0x02; 32],
+                root_epoch: Epoch(0),
+                control_root: None,
+                control_pks: BTreeMap::new(),
+                channels: Vec::new(),
+                relays: Vec::new(),
+                heads: Vec::new(),
+                banned: BTreeSet::new(),
+                cursors: BTreeMap::new(),
+                held_roots: Vec::new(),
+                channel_cuts: BTreeMap::new(),
+                refounders: BTreeSet::new(),
+                removed_at: None,
+                stranded: false,
+                dissolved: false,
+                added_at_ms: 7,
+            };
+
+            save_state(&client, &state).await.expect("saves");
+
+            // A cached rumor is also an application-specific document, but not a
+            // state document, so the prefix keeps it out of the state scan.
+            let other = EventBuilder::new(Kind::ApplicationSpecificData, "{}")
+                .tags([Tag::identifier("deadbeef")])
+                .finalize(&*LOCAL_KEYS)
+                .expect("builds");
+            client.database().save_event(&other).await.expect("saves");
+
+            let loaded = load_states(&client).await.expect("loads");
+
+            assert_eq!(loaded, vec![state]);
+        });
+    }
+
+    #[test]
+    fn caching_the_same_rumor_twice_leaves_one_row() {
+        smol::block_on(async {
+            let client = client();
+            let channel = ChannelId::from_bytes([0x9c; 32]);
+            let keys = Keys::generate();
+            let group = concord::derive::channel_group_key(&[0x07; 32], &channel, Epoch(0))
+                .expect("a group key");
+
+            let rumor = concord::cord03::build_message(
+                keys.public_key(),
+                &channel,
+                Epoch(0),
+                "twice",
+                None,
+                1_700_000_000_000,
+                None,
+            );
+            let (wrap, _) = concord::cord03::seal_rumor(&rumor, &group, &keys, false)
+                .await
+                .expect("seals");
+            let (opened, _) =
+                concord::cord03::open(&wrap, &group, &channel, Epoch(0)).expect("opens");
+
+            assert!(
+                cache_rumor(&client, &channel, &opened)
+                    .await
+                    .expect("caches")
+            );
+            assert!(
+                cache_rumor(&client, &channel, &opened)
+                    .await
+                    .expect("caches")
+            );
+
+            let cached = query_rumors(&client, &channel, None, 10, None)
+                .await
+                .expect("reads");
+            assert_eq!(cached.len(), 1);
+        });
+    }
+}

crates/community/src/history.rs

@@ -0,0 +1,764 @@
+use std::collections::{BTreeMap, BTreeSet, HashMap};
+use std::future::Future;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::{Duration, Instant};
+
+use anyhow::{Result, bail};
+use concord::cord01::{KIND_WRAP_EPHEMERAL, OpenedStream};
+use concord::cord03::{self, ChatRumor};
+use concord::derive::channel_group_key;
+use concord::state::{ChannelCursor, HeldKey};
+use concord::{ChannelId, GroupKey};
+use futures::future::{Either, select};
+use nostr_sdk::prelude::*;
+
+use crate::cache::cache_rumor;
+use crate::sync::connect_relays;
+
+/// How long one relay is given to answer one page of history.
+const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
+/// How far below a cursor a warm window reaches back.
+pub const CURSOR_OVERLAP: Duration = Duration::from_secs(60);
+
+/// The region of history to read.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct Window {
+    pub until: Option<Timestamp>,
+    pub since: Option<Timestamp>,
+}
+
+impl Window {
+    /// The newest wraps, with no older bound.
+    pub fn newest() -> Self {
+        Self::default()
+    }
+
+    /// The wraps strictly older than `oldest`.
+    pub fn older_than(oldest: Timestamp) -> Self {
+        Self {
+            until: Some(oldest - 1u64),
+            since: None,
+        }
+    }
+
+    /// The region between `since` and `oldest`, both inclusive.
+    pub fn between(since: Timestamp, oldest: Timestamp) -> Self {
+        Self {
+            until: Some(oldest - 1u64),
+            since: Some(since),
+        }
+    }
+
+    /// The window a channel is opened with.
+    pub fn opening(cursor: ChannelCursor) -> Self {
+        match cursor.newest {
+            Some(newest) => Self {
+                since: Some(newest - CURSOR_OVERLAP),
+                until: None,
+            },
+            None => Self::default(),
+        }
+    }
+}
+
+/// What a relay said about one page subscription.
+#[derive(Debug, Clone)]
+pub enum Settled {
+    Replayed,
+    Refused(String),
+}
+
+/// One relay's verdict on one page subscription.
+#[derive(Debug, Clone)]
+pub struct PageReport {
+    pub relay: RelayUrl,
+    pub outcome: Settled,
+}
+
+/// The page subscriptions in flight, by subscription id.
+pub struct PageRegistry {
+    pages: Arc<Mutex<HashMap<SubscriptionId, flume::Sender<PageReport>>>>,
+}
+
+impl Default for PageRegistry {
+    fn default() -> Self {
+        Self {
+            pages: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+}
+
+impl Clone for PageRegistry {
+    fn clone(&self) -> Self {
+        Self {
+            pages: Arc::clone(&self.pages),
+        }
+    }
+}
+
+impl PageRegistry {
+    pub fn register(&self, id: SubscriptionId, sender: flume::Sender<PageReport>) {
+        self.lock().insert(id, sender);
+    }
+
+    pub fn unregister(&self, id: &SubscriptionId) {
+        self.lock().remove(id);
+    }
+
+    /// Forget every page still in flight, because its round is gone.
+    pub fn clear(&self) {
+        self.lock().clear();
+    }
+
+    /// Hand a relay's verdict to the page that owns `id`, when it is still waiting.
+    pub fn deliver(&self, id: &SubscriptionId, relay: RelayUrl, outcome: Settled) {
+        let sender = self.lock().get(id).cloned();
+
+        let Some(sender) = sender else {
+            return;
+        };
+
+        if let Err(error) = sender.try_send(PageReport { relay, outcome }) {
+            log::debug!("community: a page report was not delivered: {error}");
+        }
+    }
+
+    fn lock(&self) -> MutexGuard<'_, HashMap<SubscriptionId, flume::Sender<PageReport>>> {
+        match self.pages.lock() {
+            Ok(pages) => pages,
+            Err(poisoned) => poisoned.into_inner(),
+        }
+    }
+}
+
+/// What one paged fetch saw.
+#[derive(Debug, Clone, Default)]
+pub struct WrapPage {
+    pub opened: Vec<ChatRumor>,
+    pub raw: usize,
+    /// Wraps that reached us under a held plane but that no held key could open.
+    pub unreadable: usize,
+    pub newest: Option<Timestamp>,
+    pub oldest: Option<Timestamp>,
+    pub exhausted: bool,
+    pub failed: bool,
+    pub errors: usize,
+}
+
+/// Walks a channel's history back over the community's own relays.
+#[allow(clippy::too_many_arguments)]
+pub async fn page(
+    client: &Client,
+    pages: &PageRegistry,
+    channel: &ChannelId,
+    held: &[HeldKey],
+    relays: &[RelayUrl],
+    window: Window,
+    max_pages: usize,
+    limit: usize,
+) -> Result<WrapPage> {
+    let planes: Vec<(HeldKey, GroupKey)> = held
+        .iter()
+        .map(|key| Ok((*key, channel_group_key(&key.key, channel, key.epoch)?)))
+        .collect::<Result<Vec<_>>>()?;
+    let authors: Vec<PublicKey> = planes.iter().map(|(_, group)| group.pk()).collect();
+
+    if authors.is_empty() || relays.is_empty() || limit == 0 {
+        return Ok(WrapPage {
+            failed: true,
+            ..WrapPage::default()
+        });
+    }
+
+    // A REQ can only target a relay the pool already knows about.
+    connect_relays(client, relays).await;
+
+    let mut walk = Walk::new(relays, window);
+    let mut opened = Vec::new();
+
+    for _ in 0..max_pages {
+        if walk.is_done() {
+            break;
+        }
+
+        let filter = wrap_filter(&authors, walk.region(), limit);
+        let asked: Vec<(usize, RelayUrl)> = walk
+            .live()
+            .map(|index| (index, walk.url(index).clone()))
+            .collect();
+
+        let answers = ask_page(client, pages, &asked, &filter).await;
+
+        for (index, url) in &asked {
+            match answers.get(url) {
+                Some(Settled::Replayed) => {}
+                Some(Settled::Refused(reason)) => {
+                    log::warn!("community: relay {url} refused a history page: {reason}");
+                    walk.reject(*index);
+                }
+                None => {
+                    log::warn!("community: relay {url} did not finish a history page");
+                    walk.reject(*index);
+                }
+            }
+        }
+
+        let answered = client.database().query(filter).await?;
+
+        for wrap in walk.accept(answered, limit) {
+            let Some((held, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
+            else {
+                continue;
+            };
+
+            let Ok((stream, rumor)) = read_under(&wrap, held, group, channel) else {
+                walk.unreadable += 1;
+                continue;
+            };
+
+            if cache_rumor(client, channel, &stream).await? {
+                opened.push(rumor);
+            }
+        }
+    }
+
+    Ok(walk.finish(opened))
+}
+
+/// Opens one wrap under a held key.
+fn read_under(
+    wrap: &Event,
+    held: &HeldKey,
+    group: &GroupKey,
+    channel: &ChannelId,
+) -> Result<(OpenedStream, ChatRumor)> {
+    if held
+        .retired_at
+        .is_some_and(|retired| wrap.created_at > retired)
+    {
+        bail!("sealed after the key that reads it was retired");
+    }
+
+    Ok(cord03::open(wrap, group, channel, held.epoch)?)
+}
+
+/// The one filter a page is asked for.
+fn wrap_filter(authors: &[PublicKey], window: Window, limit: usize) -> Filter {
+    let mut filter = Filter::new()
+        .kinds([Kind::GiftWrap, Kind::Custom(KIND_WRAP_EPHEMERAL)])
+        .authors(authors.iter().copied())
+        .limit(limit);
+
+    if let Some(until) = window.until {
+        filter = filter.until(until);
+    }
+
+    if let Some(since) = window.since {
+        filter = filter.since(since);
+    }
+
+    filter
+}
+
+/// One page's verdicts, by relay.
+type PageAnswers = BTreeMap<RelayUrl, Settled>;
+
+async fn ask_page(
+    client: &Client,
+    pages: &PageRegistry,
+    asked: &[(usize, RelayUrl)],
+    filter: &Filter,
+) -> PageAnswers {
+    let mut answers = PageAnswers::new();
+    let distinct: BTreeSet<RelayUrl> = asked.iter().map(|(_, url)| url.clone()).collect();
+    let mut targets: Vec<(RelayUrl, Vec<Filter>)> = Vec::with_capacity(distinct.len());
+
+    for url in &distinct {
+        match client.relay(url).await {
+            Ok(Some(_)) => targets.push((url.clone(), vec![filter.clone()])),
+            Ok(None) => {
+                log::warn!("community: relay {url} is not in the pool for a history page");
+                answers.insert(
+                    url.clone(),
+                    Settled::Refused("not in the relay pool".to_owned()),
+                );
+            }
+            Err(error) => {
+                log::warn!("community: relay {url} could not be looked up: {error}");
+                answers.insert(url.clone(), Settled::Refused(error.to_string()));
+            }
+        }
+    }
+
+    if targets.is_empty() {
+        return answers;
+    }
+
+    let id = history_subscription();
+    let (sender, receiver) = flume::bounded(distinct.len());
+    pages.register(id.clone(), sender);
+
+    let options = SubscribeAutoCloseOptions::default()
+        .exit_policy(ReqExitPolicy::ExitOnEOSE)
+        .timeout(Some(PAGE_TIMEOUT));
+
+    match client
+        .subscribe(ReqTarget::manual(targets))
+        .with_id(id.clone())
+        .close_on(options)
+        .await
+    {
+        Ok(output) => {
+            for (url, reason) in output.failed {
+                answers.insert(url, Settled::Refused(reason));
+            }
+
+            let deadline = Instant::now() + PAGE_TIMEOUT;
+
+            while answers.len() < distinct.len() {
+                let remaining = deadline.saturating_duration_since(Instant::now());
+
+                let Some(report) = within(remaining, receiver.recv_async()).await else {
+                    break;
+                };
+
+                match report {
+                    Ok(report) => {
+                        answers.insert(report.relay, report.outcome);
+                    }
+                    Err(error) => {
+                        log::warn!("community: a history page's reports were lost: {error}");
+                        break;
+                    }
+                }
+            }
+        }
+        Err(error) => {
+            log::warn!("community: a history page REQ was refused: {error}");
+        }
+    }
+
+    pages.unregister(&id);
+
+    answers
+}
+
+pub(crate) fn auth_required(reason: &str) -> bool {
+    matches!(
+        MachineReadablePrefix::parse(reason),
+        Some(MachineReadablePrefix::AuthRequired)
+    )
+}
+
+fn history_subscription() -> SubscriptionId {
+    static NEXT: AtomicU64 = AtomicU64::new(0);
+    SubscriptionId::new(format!("history-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
+}
+
+async fn within<F>(limit: Duration, future: F) -> Option<F::Output>
+where
+    F: Future,
+{
+    let future = std::pin::pin!(future);
+    let deadline = std::pin::pin!(smol::Timer::after(limit));
+
+    match select(future, deadline).await {
+        Either::Left((output, _)) => Some(output),
+        Either::Right(_) => None,
+    }
+}
+
+/// One pass over a channel's history, page by page.
+#[derive(Debug)]
+struct Walk {
+    relays: Vec<Walker>,
+    since: Option<Timestamp>,
+    /// The inclusive upper bound of the next page.
+    cursor: Option<Timestamp>,
+    seen: BTreeSet<EventId>,
+    newest: Option<Timestamp>,
+    oldest: Option<Timestamp>,
+    raw: usize,
+    errors: usize,
+    /// Wraps the caller could not read under any held key.
+    unreadable: usize,
+    /// A short page ended the walk.
+    bottom: bool,
+}
+
+/// One relay's standing in a walk.
+#[derive(Debug)]
+struct Walker {
+    url: RelayUrl,
+    dead: bool,
+}
+
+impl Walk {
+    fn new(relays: &[RelayUrl], window: Window) -> Self {
+        Self {
+            relays: relays
+                .iter()
+                .cloned()
+                .map(|url| Walker { url, dead: false })
+                .collect(),
+            since: window.since,
+            cursor: window.until,
+            seen: BTreeSet::new(),
+            newest: None,
+            oldest: None,
+            raw: 0,
+            errors: 0,
+            unreadable: 0,
+            bottom: false,
+        }
+    }
+
+    fn is_done(&self) -> bool {
+        self.bottom || self.relays.iter().all(|walker| walker.dead)
+    }
+
+    /// The region the next page asks for.
+    fn region(&self) -> Window {
+        Window {
+            until: self.cursor,
+            since: self.since,
+        }
+    }
+
+    fn live(&self) -> impl Iterator<Item = usize> + '_ {
+        (0..self.relays.len()).filter(|&index| !self.relays[index].dead)
+    }
+
+    fn url(&self, index: usize) -> &RelayUrl {
+        &self.relays[index].url
+    }
+
+    fn reject(&mut self, index: usize) {
+        self.relays[index].dead = true;
+        self.errors += 1;
+    }
+
+    fn accept(&mut self, page: BTreeSet<Event>, limit: usize) -> Vec<Event> {
+        if page.len() < limit {
+            self.bottom = true;
+        }
+
+        let mut oldest: Option<Timestamp> = None;
+        let mut events = Vec::with_capacity(page.len());
+
+        for event in page {
+            let at = event.created_at;
+            self.newest = Some(self.newest.map_or(at, |newest| newest.max(at)));
+            oldest = Some(oldest.map_or(at, |oldest| oldest.min(at)));
+
+            if self.seen.insert(event.id) {
+                self.raw += 1;
+                events.push(event);
+            }
+        }
+
+        // The walk's floor, which the round resumes the older pass from.
+        if let Some(oldest) = oldest {
+            self.oldest = Some(self.oldest.map_or(oldest, |held| held.min(oldest)));
+        }
+
+        match oldest {
+            Some(oldest) if !oldest.is_zero() => self.cursor = Some(oldest - 1u64),
+            Some(_) => self.bottom = true,
+            None => {}
+        }
+
+        events
+    }
+
+    fn finish(self, opened: Vec<ChatRumor>) -> WrapPage {
+        let swept = self.bottom && self.errors == 0;
+
+        WrapPage {
+            opened,
+            raw: self.raw,
+            unreadable: self.unreadable,
+            newest: self.newest,
+            oldest: self.oldest,
+            exhausted: swept && self.raw > 0,
+            failed: self.errors > 0 || (self.bottom && self.raw == 0),
+            errors: self.errors,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::cmp::Reverse;
+
+    use concord::Epoch;
+    use concord::cord03::{build_message, seal_rumor};
+    use concord::derive::channel_group_key;
+
+    use super::*;
+
+    const SECRET: [u8; 32] = [0x07u8; 32];
+    const NEXT_SECRET: [u8; 32] = [0x11u8; 32];
+
+    fn serve_page(database: &BTreeSet<Event>, window: Window, limit: usize) -> BTreeSet<Event> {
+        let mut events: Vec<Event> = database
+            .iter()
+            .filter(|event| {
+                window.until.is_none_or(|until| event.created_at <= until)
+                    && window.since.is_none_or(|since| event.created_at >= since)
+            })
+            .cloned()
+            .collect();
+
+        events.sort_by_key(|event| Reverse(event.created_at));
+        events.truncate(limit);
+        events.into_iter().collect()
+    }
+
+    fn relay_url(host: &str) -> RelayUrl {
+        RelayUrl::parse(&format!("wss://{host}.example.com")).expect("parses")
+    }
+
+    #[test]
+    fn a_walk_pages_back_across_a_rekey() {
+        let channel = ChannelId::from_bytes([0x9cu8; 32]);
+        let author = Keys::generate();
+        let held = [
+            HeldKey {
+                epoch: Epoch(0),
+                key: SECRET,
+                retired_at: None,
+            },
+            HeldKey {
+                epoch: Epoch(1),
+                key: NEXT_SECRET,
+                retired_at: None,
+            },
+        ];
+        let planes: Vec<(HeldKey, GroupKey)> = held
+            .iter()
+            .map(|key| {
+                (
+                    *key,
+                    channel_group_key(&key.key, &channel, key.epoch).expect("derives"),
+                )
+            })
+            .collect();
+
+        // Three messages a second apart: a page boundary falls between each.
+        let base = 1_700_000_000_000;
+        let mut relay: BTreeSet<Event> = BTreeSet::new();
+
+        for (content, secret, epoch, at_ms) in [
+            ("before the rekey", &SECRET, Epoch(0), base),
+            ("still before", &SECRET, Epoch(0), base + 1_000),
+            ("after the rekey", &NEXT_SECRET, Epoch(1), base + 2_000),
+        ] {
+            let group = channel_group_key(secret, &channel, epoch).expect("derives");
+            let rumor = build_message(
+                author.public_key(),
+                &channel,
+                epoch,
+                content,
+                None,
+                at_ms,
+                None,
+            );
+            relay.insert(
+                smol::block_on(seal_rumor(&rumor, &group, &author, false))
+                    .expect("seals")
+                    .0,
+            );
+        }
+
+        let mut walk = Walk::new(&[relay_url("history")], Window::newest());
+        let mut found = Vec::new();
+        let mut pages = 0;
+
+        while !walk.is_done() && pages < 10 {
+            pages += 1;
+            let page = serve_page(&relay, walk.region(), 2);
+
+            for wrap in walk.accept(page, 2) {
+                let Some((held, group)) =
+                    planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
+                else {
+                    continue;
+                };
+
+                let (_, rumor) = cord03::open(&wrap, group, &channel, held.epoch).expect("opens");
+                found.push(rumor);
+            }
+        }
+
+        found.sort_by_key(|rumor| (Reverse(rumor.at_ms), rumor.id));
+
+        let contents: Vec<&str> = found.iter().map(|rumor| rumor.content.as_str()).collect();
+        assert_eq!(
+            contents,
+            ["after the rekey", "still before", "before the rekey"]
+        );
+
+        let page = walk.finish(Vec::new());
+        assert!(page.exhausted);
+        assert!(!page.failed);
+        assert_eq!(page.raw, 3);
+        assert_eq!(page.newest, Some(Timestamp::from_secs(1_700_000_002)));
+        assert_eq!(
+            page.oldest,
+            Some(Timestamp::from_secs(1_700_000_000)),
+            "the walk reports its floor, which the older pass resumes below"
+        );
+    }
+
+    /// A wrap that reaches us and still will not open is history we cannot read,
+    /// not history that does not exist.
+    #[test]
+    fn a_wrap_no_held_key_can_open_reads_as_unreadable() {
+        let channel = ChannelId::from_bytes([0x9cu8; 32]);
+        let other = ChannelId::from_bytes([0x9du8; 32]);
+        let author = Keys::generate();
+        let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives");
+        let held = HeldKey {
+            epoch: Epoch(0),
+            key: SECRET,
+            retired_at: Some(Timestamp::from_secs(1_000)),
+        };
+
+        let wrap_at = |channel: &ChannelId, at_ms: u64| {
+            let rumor = build_message(
+                author.public_key(),
+                channel,
+                Epoch(0),
+                "sealed",
+                None,
+                at_ms,
+                None,
+            );
+            smol::block_on(seal_rumor(&rumor, &group, &author, false))
+                .expect("seals")
+                .0
+        };
+
+        // Sealed before the rotation superseded this key, so it still reads.
+        let before = wrap_at(&channel, 999_000);
+        assert!(read_under(&before, &held, &group, &channel).is_ok());
+
+        // Sealed after the cutoff the rotation set on that key.
+        let after = wrap_at(&channel, 1_001_000);
+        assert!(read_under(&after, &held, &group, &channel).is_err());
+
+        // Sealed to this plane but bound to another channel.
+        let misbound = wrap_at(&other, 999_000);
+        assert!(read_under(&misbound, &held, &group, &channel).is_err());
+    }
+
+    #[test]
+    fn an_empty_answer_never_seals_the_channel() {
+        let database: BTreeSet<Event> = BTreeSet::new();
+        let mut walk = Walk::new(&[relay_url("history")], Window::newest());
+
+        let page = serve_page(&database, walk.region(), 50);
+        assert!(walk.accept(page, 50).is_empty());
+
+        let page = walk.finish(Vec::new());
+        assert!(page.failed);
+        assert!(!page.exhausted);
+        assert_eq!(page.raw, 0);
+        assert_eq!(page.oldest, None);
+    }
+
+    #[test]
+    fn a_silent_relay_blocks_the_bottom() {
+        let database: BTreeSet<Event> = BTreeSet::new();
+        let mut walk = Walk::new(
+            &[relay_url("history"), relay_url("archive")],
+            Window::newest(),
+        );
+
+        // One relay answered the empty page; the other never answered at all, so
+        // its share of the region was never read and the walk must not seal.
+        walk.reject(1);
+        let page = serve_page(&database, walk.region(), 50);
+        assert!(walk.accept(page, 50).is_empty());
+
+        let page = walk.finish(Vec::new());
+        assert!(page.failed);
+        assert!(!page.exhausted);
+        assert_eq!(page.errors, 1);
+    }
+
+    #[test]
+    fn a_page_boundary_is_exclusive() {
+        let database = BTreeSet::from([
+            event_at(Timestamp::from_secs(1_700_000_000)),
+            event_at(Timestamp::from_secs(1_700_000_001)),
+        ]);
+        let mut walk = Walk::new(&[relay_url("history")], Window::newest());
+
+        let first = walk.accept(serve_page(&database, walk.region(), 1), 1);
+        let second = walk.accept(serve_page(&database, walk.region(), 1), 1);
+
+        assert_eq!(first.len(), 1);
+        assert_eq!(second.len(), 1);
+        assert_ne!(first[0].id, second[0].id);
+
+        let oldest = second
+            .iter()
+            .map(|event| event.created_at)
+            .min()
+            .expect("one wrap");
+        assert_eq!(oldest, Timestamp::from_secs(1_700_000_000));
+    }
+
+    fn event_at(at: Timestamp) -> Event {
+        let keys = Keys::generate();
+        EventBuilder::new(Kind::TextNote, "page")
+            .custom_created_at(at)
+            .finalize(&keys)
+            .expect("signs")
+    }
+
+    /// A cold channel asks wide; a warm one resumes at the overlap above its
+    /// cursor, and `older_than` never includes the boundary event itself.
+    #[test]
+    fn a_cold_window_is_open_and_a_warm_one_resumes_at_the_overlap() {
+        assert_eq!(Window::opening(ChannelCursor::default()), Window::default());
+
+        let warm = Window::opening(ChannelCursor {
+            newest: Some(Timestamp::from_secs(2_000_000)),
+            oldest: Some(Timestamp::from_secs(1_000)),
+            exhausted: false,
+        });
+        assert_eq!(
+            warm,
+            Window {
+                since: Some(Timestamp::from_secs(2_000_000) - CURSOR_OVERLAP),
+                until: None,
+            }
+        );
+
+        assert_eq!(
+            Window::older_than(Timestamp::from_secs(1_000)),
+            Window {
+                since: None,
+                until: Some(Timestamp::from_secs(999)),
+            }
+        );
+    }
+
+    /// A page whose round has moved on is simply a report nobody reads.
+    #[test]
+    fn a_page_that_moved_on_receives_nothing() {
+        let pages = PageRegistry::default();
+        let id = SubscriptionId::new("concord-history-9");
+        let (sender, receiver) = flume::bounded(1);
+
+        pages.register(id.clone(), sender);
+        pages.unregister(&id);
+        pages.deliver(&id, relay_url("history"), Settled::Replayed);
+
+        assert!(receiver.try_recv().is_err());
+    }
+}

crates/community/src/lib.rs

@@ -0,0 +1,850 @@
+use std::collections::{BTreeSet, HashMap};
+use std::time::{Duration, Instant};
+
+use anyhow::Result;
+pub use concord::cord02::CommunityMetadata;
+pub use concord::cord03::{ChatMessage, ReplyRef};
+use concord::state::CommunityState;
+pub use concord::{ChannelId, CommunityId, Epoch};
+use futures::future::{Either, select};
+use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, Window};
+use nostr_sdk::prelude::*;
+use smallvec::{SmallVec, smallvec};
+use state::NostrRegistry;
+
+use crate::history::{PageRegistry, Settled, auth_required};
+use crate::rekey::WatchRegistry;
+
+pub mod cache;
+mod community;
+pub mod history;
+mod rekey;
+mod sync;
+
+pub use community::*;
+pub use sync::*;
+
+/// How long a burst of relay notifications is collected before it is folded.
+const PUMP_WINDOW: Duration = Duration::from_millis(200);
+/// How long a community's relays may deliver nothing before
+/// its standing subscription is torn down and re-issued.
+const LIVE_ROTATE: Duration = Duration::from_secs(90);
+
+pub fn init(cx: &mut App) {
+    CommunityRegistry::set_global(cx.new(CommunityRegistry::new), cx);
+}
+
+struct GlobalCommunityRegistry(Entity<CommunityRegistry>);
+
+impl Global for GlobalCommunityRegistry {}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum Signal {
+    Event(CommunityId),
+    List,
+    Rekey(CommunityId),
+}
+
+/// Which standing subscription an id belongs to.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Route {
+    List,
+    Community(CommunityId),
+}
+
+fn route_of(id: &SubscriptionId) -> Option<Route> {
+    if sync::is_list_subscription(id) {
+        return Some(Route::List);
+    }
+
+    sync::community_of(id).map(Route::Community)
+}
+
+/// What a window of relay notifications saw, waiting to be folded once.
+#[derive(Default)]
+struct Batch {
+    list: bool,
+    communities: BTreeSet<CommunityId>,
+    rekeys: BTreeSet<CommunityId>,
+}
+
+/// Whether the pump should keep listening.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Flow {
+    Continue,
+    Stop,
+}
+
+/// Fold one notification into the window, or settle a page it belongs to.
+fn route(
+    notification: ClientNotification,
+    pages: &PageRegistry,
+    watches: &WatchRegistry,
+    batch: &mut Batch,
+) -> Flow {
+    match notification {
+        ClientNotification::Event {
+            subscription_id, ..
+        } => match route_of(&subscription_id) {
+            Some(Route::List) => batch.list = true,
+            Some(Route::Community(id)) => {
+                batch.communities.insert(id);
+            }
+            None => {
+                if let Some(id) = watches.community_of(&subscription_id) {
+                    batch.rekeys.insert(id);
+                }
+            }
+        },
+        ClientNotification::Message { relay_url, message } => match *message {
+            RelayMessage::EndOfStoredEvents(id) => {
+                pages.deliver(&id, relay_url, Settled::Replayed);
+            }
+            RelayMessage::Closed {
+                subscription_id,
+                message,
+            } if !auth_required(&message) => {
+                pages.deliver(
+                    &subscription_id,
+                    relay_url,
+                    Settled::Refused(message.into_owned()),
+                );
+            }
+            _ => {}
+        },
+        ClientNotification::Shutdown => return Flow::Stop,
+    }
+
+    Flow::Continue
+}
+
+/// Hand the window's signals to the foreground consumer, one per community.
+async fn flush(tx: &flume::Sender<Signal>, batch: &mut Batch) -> Result<()> {
+    for id in std::mem::take(&mut batch.communities) {
+        tx.send_async(Signal::Event(id)).await?;
+    }
+
+    for id in std::mem::take(&mut batch.rekeys) {
+        tx.send_async(Signal::Rekey(id)).await?;
+    }
+
+    if std::mem::take(&mut batch.list) {
+        tx.send_async(Signal::List).await?;
+    }
+
+    Ok(())
+}
+
+impl EventEmitter<CommunityEvent> for CommunityRegistry {}
+
+pub struct CommunityRegistry {
+    communities: Vec<Entity<Community>>,
+    index: HashMap<CommunityId, Entity<Community>>,
+    /// The plane set each community was last subscribed with
+    synced: HashMap<CommunityId, SubscriptionKey>,
+    /// When a relay last delivered something for a community,
+    /// which is the only evidence the standing subscription is alive.
+    last_event: HashMap<CommunityId, Instant>,
+    /// One observer per tracked community, dropped on reset
+    observers: HashMap<CommunityId, Subscription>,
+    signal_tx: flume::Sender<Signal>,
+    signal_rx: flume::Receiver<Signal>,
+    /// The page subscriptions in flight, shared with the notification pump.
+    pages: PageRegistry,
+    /// The rekey watch subscriptions, resolved to their community.
+    watches: WatchRegistry,
+    tasks: SmallVec<[Task<Result<()>>; 2]>,
+    /// Notification listener task (cancelled on signer change)
+    notification_listener: Option<Task<Result<()>>>,
+    /// Signal consumer task (cancelled on signer change)
+    signal_consumer: Option<Task<Result<()>>>,
+    /// The round scheduler (cancelled on signer change)
+    scheduler: Option<Task<Result<()>>>,
+    _subscriptions: SmallVec<[Subscription; 2]>,
+}
+
+impl CommunityRegistry {
+    pub fn global(cx: &App) -> Entity<Self> {
+        cx.global::<GlobalCommunityRegistry>().0.clone()
+    }
+
+    fn set_global(state: Entity<Self>, cx: &mut App) {
+        cx.set_global(GlobalCommunityRegistry(state));
+    }
+
+    fn new(cx: &mut Context<Self>) -> Self {
+        let entity = cx.entity().downgrade();
+        let nostr = NostrRegistry::global(cx);
+        let (tx, rx) = flume::bounded::<Signal>(256);
+        let mut subscriptions = smallvec![];
+
+        subscriptions.push(cx.subscribe(&nostr, |this, _nostr, event, cx| {
+            if event.signer_changed() {
+                this.reset(cx);
+                this.handle_notifications(cx);
+                this.subscribe_list(cx);
+                this.load(cx);
+            }
+        }));
+
+        cx.defer(move |cx| {
+            entity
+                .update(cx, |this, cx| {
+                    this.handle_notifications(cx);
+                    if nostr.read(cx).current_user().is_some() {
+                        this.subscribe_list(cx);
+                        this.load(cx);
+                    }
+                })
+                .ok();
+        });
+
+        Self {
+            communities: Vec::new(),
+            index: HashMap::new(),
+            synced: HashMap::new(),
+            last_event: HashMap::new(),
+            observers: HashMap::new(),
+            signal_tx: tx,
+            signal_rx: rx,
+            pages: PageRegistry::default(),
+            watches: WatchRegistry::default(),
+            tasks: smallvec![],
+            notification_listener: None,
+            signal_consumer: None,
+            scheduler: None,
+            _subscriptions: subscriptions,
+        }
+    }
+
+    pub fn communities(&self) -> &[Entity<Community>] {
+        &self.communities
+    }
+
+    pub fn community(&self, id: &CommunityId) -> Option<Entity<Community>> {
+        self.index.get(id).cloned()
+    }
+
+    /// Ask the workspace to open a community's panel.
+    pub fn emit_community(
+        &mut self,
+        community: &Entity<Community>,
+        window: &mut Window,
+        cx: &mut Context<Self>,
+    ) {
+        let id = community.read(cx).id();
+
+        cx.defer_in(window, move |_this, _window, cx| {
+            cx.emit(CommunityEvent::Open(id));
+        });
+    }
+
+    /// Ask the workspace to close a community's panel.
+    pub fn emit_close(&mut self, id: CommunityId, window: &mut Window, cx: &mut Context<Self>) {
+        cx.defer_in(window, move |_this, _window, cx| {
+            cx.emit(CommunityEvent::Close(id));
+        });
+    }
+
+    /// Create a community owned by the current account and begin tracking it.
+    pub fn create(&mut self, metadata: CommunityMetadata, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let current_user = nostr.read(cx).current_user();
+
+        if current_user.is_none() {
+            cx.emit(CommunityEvent::Error(
+                "cannot create a community without an account".to_owned(),
+            ));
+            return;
+        }
+
+        let signer = nostr.read(cx).signer();
+        let client = nostr.read(cx).client();
+
+        let task =
+            cx.background_spawn(async move { sync::create(&client, &signer, &metadata).await });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            match task.await {
+                Ok(_state) => this.update(cx, |this, cx| this.load(cx))?,
+                Err(error) => {
+                    this.update(cx, |_this, cx| {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    })?;
+                }
+            }
+
+            Ok(())
+        }));
+    }
+
+    /// Forget the current account and cancel everything in flight.
+    pub fn reset(&mut self, cx: &mut Context<Self>) {
+        self.notification_listener = None;
+        self.signal_consumer = None;
+        self.scheduler = None;
+        self.tasks.clear();
+        self.observers.clear();
+        self.pages.clear();
+        self.watches.clear();
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+        let ids: Vec<CommunityId> = self.index.keys().copied().collect();
+
+        for id in ids {
+            let client = client.clone();
+            let subscription = sync::subscription_id(&id);
+            let rekey = rekey::subscription_id(&id);
+
+            self.tasks.push(cx.background_spawn(async move {
+                client.unsubscribe(&subscription).await?;
+                client.unsubscribe(&rekey).await?;
+                Ok(())
+            }));
+        }
+
+        self.communities.clear();
+        self.index.clear();
+        self.synced.clear();
+        self.last_event.clear();
+
+        cx.notify();
+    }
+
+    /// Subscribe to the account's community list.
+    fn subscribe_list(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let signer = nostr.read(cx).signer();
+        let client = nostr.read(cx).client();
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            let self_pk = signer.get_public_key_async().await?;
+
+            if let Err(error) = sync::subscribe_list(&client, self_pk).await {
+                this.update(cx, |_this, cx| {
+                    cx.emit(CommunityEvent::Error(error.to_string()));
+                })?;
+            }
+
+            Ok(())
+        }));
+    }
+
+    /// Discover the account's communities in the local database.
+    fn load(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+        let signer = nostr.read(cx).signer();
+        let client = nostr.read(cx).client();
+
+        let task = cx.background_spawn(async move {
+            let self_pk = signer.get_public_key_async().await?;
+            sync::load(&client, &signer, self_pk).await
+        });
+
+        self.tasks.push(cx.spawn(async move |this, cx| {
+            match task.await {
+                Ok(states) => {
+                    this.update(cx, |this, cx| this.track(states, cx))?;
+                }
+                Err(error) => {
+                    this.update(cx, |_this, cx| {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    })?;
+                }
+            }
+
+            Ok(())
+        }));
+    }
+
+    /// Replace the tracked communities with a freshly loaded set.
+    ///
+    /// A community that survives the reload keeps its entity, so an open panel
+    /// and a browsing sidebar stay pointed at a live community.
+    fn track(&mut self, states: Vec<CommunityState>, cx: &mut Context<Self>) {
+        let mut communities = Vec::with_capacity(states.len());
+
+        for state in states {
+            let id = state.id;
+
+            let community = match self.index.remove(&id) {
+                Some(community) => {
+                    // The list can carry plane material the store does not.
+                    if community.read(cx).state() != &state {
+                        community.update(cx, |community, _cx| community.adopt(state));
+                    }
+
+                    community
+                }
+                None => {
+                    let community = cx.new(|_| Community::new(state, self.pages.clone()));
+
+                    self.observers.insert(
+                        id,
+                        cx.observe(&community, |this, _community, cx| {
+                            this.sync_subscriptions(cx);
+                            cx.notify();
+                        }),
+                    );
+
+                    community
+                }
+            };
+
+            communities.push((id, community));
+        }
+
+        // Whatever the index still holds is no longer in the list.
+        let dropped: Vec<CommunityId> = self.index.keys().copied().collect();
+
+        for id in dropped {
+            self.observers.remove(&id);
+            self.synced.remove(&id);
+            self.last_event.remove(&id);
+        }
+
+        self.communities = communities
+            .iter()
+            .map(|(_, community)| community.clone())
+            .collect();
+        self.index = communities.into_iter().collect();
+
+        self.sync_subscriptions(cx);
+
+        // A backlog already in the database produces no notification, so fold it once.
+        for community in self.communities.clone() {
+            community.update(cx, |community, cx| community.refresh(cx));
+        }
+
+        cx.notify();
+    }
+
+    fn refresh(&mut self, id: CommunityId, cx: &mut Context<Self>) {
+        let Some(community) = self.index.get(&id).cloned() else {
+            return;
+        };
+
+        community.update(cx, |community, cx| community.refresh(cx));
+    }
+
+    /// Adopt whatever a community's rekey watch has delivered.
+    fn rekey(&mut self, id: CommunityId, cx: &mut Context<Self>) {
+        let Some(community) = self.index.get(&id).cloned() else {
+            return;
+        };
+
+        community.update(cx, |community, cx| community.rekey(cx));
+    }
+
+    /// One scheduler pass: every community repairs itself if it has gone stale,
+    /// and any whose relays have gone quiet is re-subscribed.
+    fn tick(&mut self, cx: &mut Context<Self>) {
+        for community in self.communities.clone() {
+            community.update(cx, |community, cx| community.tick(cx));
+        }
+        self.rotate_quiet(cx);
+    }
+
+    /// Re-issue the standing subscription of every community that has been quiet for `LIVE_ROTATE`
+    fn rotate_quiet(&mut self, cx: &mut Context<Self>) {
+        let now = Instant::now();
+        let ids: Vec<CommunityId> = self.index.keys().copied().collect();
+        let mut rotated = false;
+
+        for id in ids {
+            let quiet = self
+                .last_event
+                .get(&id)
+                .is_none_or(|at| now.duration_since(*at) >= LIVE_ROTATE);
+
+            if !quiet {
+                continue;
+            }
+
+            self.synced.remove(&id);
+            rotated = true;
+        }
+
+        if rotated {
+            self.sync_subscriptions(cx);
+        }
+    }
+
+    /// Re-subscribe every community whose held planes moved.
+    fn sync_subscriptions(&mut self, cx: &mut Context<Self>) {
+        let nostr = NostrRegistry::global(cx);
+
+        for community in self.communities.clone() {
+            let client = nostr.read(cx).client();
+
+            let (id, key, state) = {
+                let community = community.read(cx);
+                (
+                    community.id(),
+                    community.subscription_key(),
+                    community.state().clone(),
+                )
+            };
+
+            if self.synced.get(&id) == Some(&key) {
+                continue;
+            }
+
+            let planes = match sync::planes(&state) {
+                Ok(planes) => planes,
+                Err(error) => {
+                    cx.emit(CommunityEvent::Error(error.to_string()));
+                    continue;
+                }
+            };
+
+            let subscription = sync::subscription_id(&id);
+            let filter = sync::live_filter(&planes, sync::live_window(&state, Timestamp::now()));
+            let relays = key.relays().to_vec();
+
+            // A fresh REQ counts as evidence of life for this community.
+            self.last_event.insert(id, Instant::now());
+
+            // The rekey watch is a second standing REQ over the same relays.
+            let watch = match rekey::watches(&state) {
+                Ok(watches) => rekey::watch_filter(&watches),
+                Err(error) => {
+                    cx.emit(CommunityEvent::Error(error.to_string()));
+                    continue;
+                }
+            };
+
+            let rekey_subscription = rekey::subscription_id(&id);
+            self.watches.register(rekey_subscription.clone(), id);
+
+            self.synced.insert(id, key);
+
+            self.tasks.push(cx.spawn(async move |this, cx| {
+                if let Err(error) = subscribe(&client, &subscription, &relays, filter).await {
+                    this.update(cx, |_this, cx| {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    })?;
+                }
+
+                if let Err(error) = subscribe(&client, &rekey_subscription, &relays, watch).await {
+                    this.update(cx, |_this, cx| {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    })?;
+                }
+
+                Ok(())
+            }));
+        }
+    }
+
+    fn handle_notifications(&mut self, cx: &mut Context<Self>) {
+        self.notification_listener = None;
+        self.signal_consumer = None;
+        self.scheduler = None;
+
+        let nostr = NostrRegistry::global(cx);
+        let client = nostr.read(cx).client();
+
+        let tx = self.signal_tx.clone();
+        let rx = self.signal_rx.clone();
+        let pages = self.pages.clone();
+        let watches = self.watches.clone();
+        let executor = cx.background_executor().clone();
+
+        self.notification_listener = Some(cx.background_spawn(async move {
+            let mut notifications = client.notifications();
+            let mut batch = Batch::default();
+
+            'outer: loop {
+                match notifications.next().await {
+                    Some(notification) => {
+                        if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
+                            flush(&tx, &mut batch).await?;
+                            break 'outer;
+                        }
+                    }
+                    None => break 'outer,
+                }
+
+                let deadline = Instant::now() + PUMP_WINDOW;
+
+                loop {
+                    let now = Instant::now();
+
+                    if now >= deadline {
+                        break;
+                    }
+
+                    let timer = executor.timer(deadline - now);
+                    let next = notifications.next();
+                    futures::pin_mut!(timer);
+                    futures::pin_mut!(next);
+
+                    match select(next, timer).await {
+                        Either::Left((Some(notification), _)) => {
+                            if route(notification, &pages, &watches, &mut batch) == Flow::Stop {
+                                flush(&tx, &mut batch).await?;
+                                break 'outer;
+                            }
+                        }
+                        Either::Left((None, _)) => break 'outer,
+                        Either::Right(_) => break,
+                    }
+                }
+
+                flush(&tx, &mut batch).await?;
+            }
+
+            Ok(())
+        }));
+
+        self.signal_consumer = Some(cx.spawn(async move |this, cx| {
+            while let Ok(signal) = rx.recv_async().await {
+                match signal {
+                    Signal::Event(id) => this.update(cx, |this, cx| {
+                        // The only proof a community's subscription is still
+                        // delivering anything.
+                        this.last_event.insert(id, Instant::now());
+                        this.refresh(id, cx);
+                    })?,
+                    Signal::Rekey(id) => this.update(cx, |this, cx| this.rekey(id, cx))?,
+                    Signal::List => this.update(cx, |this, cx| this.load(cx))?,
+                }
+            }
+            Ok(())
+        }));
+
+        self.scheduler = Some(cx.spawn(async move |this, cx| {
+            loop {
+                cx.background_executor().timer(MIN_ROUND_INTERVAL).await;
+
+                if let Some(registry) = this.upgrade() {
+                    registry.update(cx, |this, cx| {
+                        this.tick(cx);
+                    });
+                } else {
+                    break;
+                }
+            }
+
+            Ok(())
+        }));
+    }
+}
+
+async fn subscribe(
+    client: &Client,
+    id: &SubscriptionId,
+    relays: &[RelayUrl],
+    filter: Filter,
+) -> Result<()> {
+    client.unsubscribe(id).await?;
+
+    if relays.is_empty() {
+        log::warn!("community {id}: no relay to subscribe to");
+        return Ok(());
+    }
+
+    let mut targets: Vec<(RelayUrl, Vec<Filter>)> = Vec::with_capacity(relays.len());
+
+    for url in relays {
+        if let Err(error) = client.add_relay(url).and_connect().await {
+            log::warn!("community {id}: failed to add relay {url}: {error}");
+        }
+
+        match client.relay(url).await {
+            Ok(Some(_)) => targets.push((url.clone(), vec![filter.clone()])),
+            Ok(None) => log::warn!("community {id}: relay {url} is not in the pool"),
+            Err(error) => log::warn!("community {id}: relay {url} could not be looked up: {error}"),
+        }
+    }
+
+    if targets.is_empty() {
+        log::warn!("community {id}: no relay accepted the standing subscription");
+        return Ok(());
+    }
+
+    let output = client
+        .subscribe(ReqTarget::manual(targets))
+        .with_id(id.clone())
+        .await?;
+
+    if !output.failed.is_empty() {
+        log::warn!(
+            "community {id}: {} relay(s) rejected the subscription",
+            output.failed.len()
+        );
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn relay() -> RelayUrl {
+        RelayUrl::parse("wss://relay.example").expect("a url")
+    }
+
+    fn event(subscription_id: SubscriptionId) -> ClientNotification {
+        let keys = Keys::generate();
+        let event = EventBuilder::new(Kind::TextNote, "hi")
+            .finalize(&keys)
+            .expect("signs");
+
+        ClientNotification::Event {
+            relay_url: relay(),
+            subscription_id,
+            event: Box::new(event),
+        }
+    }
+
+    fn message(message: RelayMessage<'static>) -> ClientNotification {
+        ClientNotification::Message {
+            relay_url: relay(),
+            message: Box::new(message),
+        }
+    }
+
+    /// A burst within one window folds each community once, and the list once,
+    /// however many events arrived.
+    #[test]
+    fn a_burst_collapses_to_one_signal_per_community() {
+        let pages = PageRegistry::default();
+        let watches = WatchRegistry::default();
+        let community = CommunityId::from_bytes([0x42; 32]);
+        let mut batch = Batch::default();
+
+        let plane = event(sync::subscription_id(&community));
+        for _ in 0..50 {
+            assert_eq!(
+                route(plane.clone(), &pages, &watches, &mut batch),
+                Flow::Continue
+            );
+        }
+
+        route(
+            event(sync::list_subscription_id()),
+            &pages,
+            &watches,
+            &mut batch,
+        );
+
+        // A page's event is folded from the database later, not routed here.
+        route(
+            event(SubscriptionId::new("concord-history-7")),
+            &pages,
+            &watches,
+            &mut batch,
+        );
+
+        assert!(batch.list);
+        assert_eq!(batch.communities, BTreeSet::from([community]));
+        assert!(batch.rekeys.is_empty());
+    }
+
+    /// A rekey watch's wraps put themselves in the database; the pump's only job
+    /// is to wake the adoption pass, and it resolves the community by id.
+    #[test]
+    fn a_rekey_watch_event_wakes_its_community() {
+        let pages = PageRegistry::default();
+        let watches = WatchRegistry::default();
+        let community = CommunityId::from_bytes([0x42; 32]);
+        let id = rekey::subscription_id(&community);
+        watches.register(id.clone(), community);
+
+        let mut batch = Batch::default();
+        route(event(id), &pages, &watches, &mut batch);
+
+        assert_eq!(batch.rekeys, BTreeSet::from([community]));
+        assert!(batch.communities.is_empty());
+    }
+
+    #[test]
+    fn an_eose_settles_only_the_page_that_owns_the_id() {
+        let pages = PageRegistry::default();
+        let mine = SubscriptionId::new("concord-history-1");
+        let other = SubscriptionId::new("concord-history-2");
+        let (mine_tx, mine_rx) = flume::bounded(1);
+        let (other_tx, other_rx) = flume::bounded(1);
+        pages.register(mine.clone(), mine_tx);
+        pages.register(other, other_tx);
+
+        let mut batch = Batch::default();
+        let flow = route(
+            message(RelayMessage::eose(mine)),
+            &pages,
+            &WatchRegistry::default(),
+            &mut batch,
+        );
+
+        assert_eq!(flow, Flow::Continue);
+        assert!(matches!(
+            mine_rx.try_recv().expect("a report").outcome,
+            Settled::Replayed
+        ));
+        assert!(other_rx.try_recv().is_err());
+    }
+
+    #[test]
+    fn a_refused_page_settles_its_relay_as_refused() {
+        let pages = PageRegistry::default();
+        let id = SubscriptionId::new("concord-history-1");
+        let (sender, receiver) = flume::bounded(1);
+        pages.register(id.clone(), sender);
+
+        let mut batch = Batch::default();
+        route(
+            message(RelayMessage::closed(id, "blocked: not allowed")),
+            &pages,
+            &WatchRegistry::default(),
+            &mut batch,
+        );
+
+        match receiver.try_recv().expect("a report").outcome {
+            Settled::Refused(reason) => assert!(reason.contains("blocked")),
+            other => panic!("expected a refusal, got {other:?}"),
+        }
+    }
+
+    /// The SDK re-issues an `auth-required` REQ under the same id after AUTH, so
+    /// the page keeps waiting rather than writing the relay off.
+    #[test]
+    fn an_auth_required_close_settles_nothing() {
+        let pages = PageRegistry::default();
+        let id = SubscriptionId::new("concord-history-1");
+        let (sender, receiver) = flume::bounded(1);
+        pages.register(id.clone(), sender);
+
+        let mut batch = Batch::default();
+        route(
+            message(RelayMessage::closed(
+                id,
+                "auth-required: please authenticate",
+            )),
+            &pages,
+            &WatchRegistry::default(),
+            &mut batch,
+        );
+
+        assert!(receiver.try_recv().is_err());
+    }
+
+    #[test]
+    fn a_shutdown_stops_the_pump() {
+        let pages = PageRegistry::default();
+        let mut batch = Batch::default();
+
+        assert_eq!(
+            route(
+                ClientNotification::Shutdown,
+                &pages,
+                &WatchRegistry::default(),
+                &mut batch
+            ),
+            Flow::Stop
+        );
+    }
+}