.

Cargo.lock

@@ -1281,6 +1281,7 @@ dependencies = [
  "anyhow",
  "concord",
  "flume 0.11.1",
+ "futures",
  "gpui-pre",
  "log",
  "nostr-memory",
@@ -1299,7 +1300,6 @@ dependencies = [
  "common",
  "community",
  "gpui-pre",
- "log",
  "nostr-sdk",
  "person",
  "settings",
@@ -1337,9 +1337,7 @@ dependencies = [
  "data-encoding",
  "hkdf",
  "hmac 0.12.1",
- "log",
  "nostr",
- "nostr-memory",
  "nostr-sdk",
  "rand 0.10.2",
  "serde",

crates/community/Cargo.toml

@@ -13,10 +13,11 @@ nostr-sdk.workspace = true
 
 anyhow.workspace = true
 flume.workspace = true
+futures.workspace = true
 log.workspace = true
 serde_json.workspace = true
 smallvec.workspace = true
+smol.workspace = true
 
 [dev-dependencies]
 nostr-memory.workspace = true
-smol.workspace = true

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

@@ -1,22 +1,35 @@
-use std::collections::HashMap;
+use std::collections::{BTreeSet, HashMap};
+use std::time::{Duration, Instant};
 
 use anyhow::Result;
-use concord::cord01::KIND_WRAP;
 pub use concord::cord02::CommunityMetadata;
 pub use concord::cord03::{ChatMessage, ReplyRef};
-use concord::store::CommunityState;
-pub use concord::{ChannelId, CommunityId};
+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);
 }
@@ -29,6 +42,97 @@ impl Global for GlobalCommunityRegistry {}
 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 {}
@@ -38,15 +142,24 @@ pub struct CommunityRegistry {
     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]>,
 }
 
@@ -90,12 +203,16 @@ impl CommunityRegistry {
             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,
         }
     }
@@ -165,8 +282,11 @@ impl CommunityRegistry {
     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();
@@ -175,9 +295,11 @@ impl CommunityRegistry {
         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(())
             }));
         }
@@ -185,6 +307,8 @@ impl CommunityRegistry {
         self.communities.clear();
         self.index.clear();
         self.synced.clear();
+        self.last_event.clear();
+
         cx.notify();
     }
 
@@ -254,7 +378,7 @@ impl CommunityRegistry {
                     community
                 }
                 None => {
-                    let community = cx.new(|_| Community::new(state));
+                    let community = cx.new(|_| Community::new(state, self.pages.clone()));
 
                     self.observers.insert(
                         id,
@@ -277,6 +401,7 @@ impl CommunityRegistry {
         for id in dropped {
             self.observers.remove(&id);
             self.synced.remove(&id);
+            self.last_event.remove(&id);
         }
 
         self.communities = communities
@@ -303,6 +428,49 @@ impl CommunityRegistry {
         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);
@@ -332,9 +500,24 @@ impl CommunityRegistry {
             };
 
             let subscription = sync::subscription_id(&id);
-            let filter = sync::subscription_filter(&planes);
+            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| {
@@ -344,6 +527,12 @@ impl CommunityRegistry {
                     })?;
                 }
 
+                if let Err(error) = subscribe(&client, &rekey_subscription, &relays, watch).await {
+                    this.update(cx, |_this, cx| {
+                        cx.emit(CommunityEvent::Error(error.to_string()));
+                    })?;
+                }
+
                 Ok(())
             }));
         }
@@ -352,40 +541,59 @@ impl CommunityRegistry {
     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();
 
-            while let Some(notification) = notifications.next().await {
-                let ClientNotification::Event {
-                    subscription_id,
-                    event,
-                    ..
-                } = notification
-                else {
-                    continue;
-                };
-
-                if sync::is_list_subscription(&subscription_id) {
-                    tx.send_async(Signal::List).await?;
-                    continue;
+            '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,
                 }
 
-                if event.kind != Kind::from(KIND_WRAP) {
-                    continue;
+                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,
+                    }
                 }
 
-                let Some(id) = sync::community_of(&subscription_id) else {
-                    continue;
-                };
-
-                tx.send_async(Signal::Event(id)).await?;
+                flush(&tx, &mut batch).await?;
             }
 
             Ok(())
@@ -394,12 +602,34 @@ impl CommunityRegistry {
         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| this.refresh(id, cx))?,
+                    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(())
+        }));
     }
 }
 
@@ -411,24 +641,34 @@ async fn subscribe(
 ) -> 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}"),
+        }
     }
 
-    let target = if relays.is_empty() {
-        ReqTarget::auto(vec![filter])
-    } else {
-        ReqTarget::manual(
-            relays
-                .iter()
-                .map(|url| (url.clone(), vec![filter.clone()]))
-                .collect::<Vec<_>>(),
-        )
-    };
+    if targets.is_empty() {
+        log::warn!("community {id}: no relay accepted the standing subscription");
+        return Ok(());
+    }
 
-    let output = client.subscribe(target).with_id(id.clone()).await?;
+    let output = client
+        .subscribe(ReqTarget::manual(targets))
+        .with_id(id.clone())
+        .await?;
 
     if !output.failed.is_empty() {
         log::warn!(
@@ -439,3 +679,188 @@ async fn subscribe(
 
     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),
+        }
+    }
+
+    #[test]
+    fn only_the_list_and_a_community_id_have_a_route() {
+        let id = CommunityId::from_bytes([0x42; 32]);
+
+        assert_eq!(route_of(&sync::list_subscription_id()), Some(Route::List));
+        assert_eq!(
+            route_of(&sync::subscription_id(&id)),
+            Some(Route::Community(id))
+        );
+        assert_eq!(route_of(&SubscriptionId::new("concord-history-3")), None);
+        assert_eq!(
+            route_of(&SubscriptionId::new("some/other/subscription")),
+            None
+        );
+    }
+
+    /// 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
+        );
+    }
+}

crates/community_ui/Cargo.toml

@@ -17,4 +17,3 @@ gpui.workspace = true
 nostr-sdk.workspace = true
 anyhow.workspace = true
 smallvec.workspace = true
-log.workspace = true

crates/community_ui/src/lib.rs

@@ -1,11 +1,18 @@
+use std::collections::HashMap;
+use std::fmt;
+
 use anyhow::Result;
-use community::{ChannelId, ChatMessage, Community, CommunityEvent};
+use community::{
+    ChannelId, ChatMessage, Community, CommunityEvent, Epoch, Intent, LOAD_OLDER_PAGES,
+    TIMELINE_PAGE, Timeline,
+};
 use gpui::prelude::FluentBuilder;
 use gpui::{
-    AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable,
-    IntoElement, ListAlignment, ListState, ParentElement, Render, SharedString, Styled,
-    Subscription, Task, WeakEntity, Window, div, list, px,
+    AnyElement, App, AppContext, Context, Entity, EventEmitter, FocusHandle, Focusable, FollowMode,
+    IntoElement, ListAlignment, ListScrollEvent, ListState, ParentElement, Render, SharedString,
+    Styled, Subscription, Task, WeakEntity, Window, div, list, px,
 };
+use nostr_sdk::prelude::EventId;
 use smallvec::{SmallVec, smallvec};
 use theme::ActiveTheme;
 use ui::avatar::Avatar;
@@ -14,10 +21,66 @@ use ui::dock::{Panel, PanelEvent};
 use ui::input::{InputEvent, Textarea, TextareaState};
 use ui::notification::Notification;
 use ui::scroll::Scrollbar;
-use ui::{IconName, Sizable, WindowExtension, h_flex, v_flex};
+use ui::{Disableable, IconName, Sizable, WindowExtension, h_flex, v_flex};
 
 mod message;
 
+/// How near the top row a scroll has to come before the panel splices older history in.
+const LOAD_OLDER_THRESHOLD: usize = 20;
+
+/// Why a room is not showing the messages it knows about.
+///
+/// "No messages yet" is a claim, and a room we cannot read is not an empty one.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Notice {
+    Stranded,
+    Removed(Epoch),
+    ChannelRemoved(Epoch),
+    MissingKey(Epoch),
+    Unreachable,
+    Unreadable(usize),
+}
+
+impl fmt::Display for Notice {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Notice::Stranded => formatter.write_str(
+                "This invite is stale, the community has rotated past the epoch it names",
+            ),
+            Notice::Removed(epoch) => write!(
+                formatter,
+                "You were removed from this community at epoch {}. Its history stays readable",
+                epoch.0
+            ),
+            Notice::ChannelRemoved(epoch) => write!(
+                formatter,
+                "A rotation removed you from this channel at epoch {}",
+                epoch.0
+            ),
+            Notice::MissingKey(epoch) => write!(
+                formatter,
+                "Messages here can't be read yet — this channel's key for epoch {} is missing",
+                epoch.0
+            ),
+            Notice::Unreachable => formatter.write_str("Couldn't reach the community's relays"),
+            Notice::Unreadable(1) => {
+                formatter.write_str("1 message here can't be read yet — no key we hold opens it")
+            }
+            Notice::Unreadable(count) => write!(
+                formatter,
+                "{count} messages here can't be read yet — no key we hold opens them"
+            ),
+        }
+    }
+}
+
+impl Notice {
+    /// Whether the room still accepts writes, rather than only being readable.
+    fn writable(self) -> bool {
+        matches!(self, Notice::Unreachable | Notice::Unreadable(_))
+    }
+}
+
 pub fn init(
     community: Entity<Community>,
     window: &mut Window,
@@ -30,25 +93,22 @@ pub fn init(
 pub struct CommunityPanel {
     id: SharedString,
     focus_handle: FocusHandle,
-
     /// Community
     community: WeakEntity<Community>,
-
     /// The selected channel
     channel: Option<ChannelId>,
-
     /// The selected channel's timeline (oldest first)
-    messages: Vec<ChatMessage>,
-
+    rows: Vec<ChatMessage>,
+    /// Whether the store holds rows older than `rows`
+    has_more: bool,
+    /// A round or a page read is in flight
+    loading: bool,
     /// Message list state
     list_state: ListState,
-
     /// Message input state
     input: Entity<TextareaState>,
-
-    /// Async operations
-    tasks: Vec<Task<Result<()>>>,
-
+    /// Spawned reads and publishes, cancelled when the panel closes
+    tasks: SmallVec<[Task<Result<()>>; 4]>,
     /// Event subscriptions
     _subscriptions: SmallVec<[Subscription; 2]>,
 }
@@ -85,20 +145,24 @@ impl CommunityPanel {
         subscriptions.push(cx.subscribe_in(
             &community,
             window,
-            |_this, _community, event, window, cx| match event {
-                // The fold holds the community, so reload once it is released.
-                CommunityEvent::Updated(_) => {
-                    cx.defer_in(window, |this, window, cx| this.reload(window, cx));
-                }
-                // The sidebar picked another channel while the community was updating.
-                CommunityEvent::Channel(..) => {
-                    cx.defer_in(window, |this, window, cx| this.load(window, cx));
-                }
-                CommunityEvent::Error(error) => {
-                    window
-                        .push_notification(Notification::error(error.clone()).autohide(false), cx);
-                }
-                CommunityEvent::Open(_) | CommunityEvent::Close(_) => {}
+            |_this, _community, event, window, cx| {
+                match event {
+                    CommunityEvent::Updated(_)
+                    | CommunityEvent::Unreadable(_)
+                    | CommunityEvent::Failed(_) => {
+                        cx.defer_in(window, |this, window, cx| this.reload(window, cx));
+                    }
+                    CommunityEvent::Channel(..) => {
+                        cx.defer_in(window, |this, window, cx| this.load(window, cx));
+                    }
+                    CommunityEvent::Error(error) => {
+                        window.push_notification(
+                            Notification::error(error.clone()).autohide(false),
+                            cx,
+                        );
+                    }
+                    CommunityEvent::Open(_) | CommunityEvent::Close(_) => {}
+                };
             },
         ));
 
@@ -107,14 +171,27 @@ impl CommunityPanel {
             focus_handle: cx.focus_handle(),
             community: community.downgrade(),
             channel,
-            messages: Vec::new(),
+            rows: Vec::new(),
+            has_more: false,
+            loading: false,
             list_state: ListState::new(0, ListAlignment::Bottom, px(1024.)),
             input,
-            tasks: Vec::new(),
+            tasks: smallvec![],
             _subscriptions: subscriptions,
         };
 
-        cx.defer_in(window, |this, window, cx| this.load(window, cx));
+        panel.list_state.set_follow_mode(FollowMode::Tail);
+        panel.list_state.set_scroll_handler(cx.listener(
+            |this, event: &ListScrollEvent, window, cx| {
+                if event.visible_range.start <= LOAD_OLDER_THRESHOLD {
+                    this.load_older(window, cx);
+                }
+            },
+        ));
+
+        cx.defer_in(window, |this, window, cx| {
+            this.load(window, cx);
+        });
 
         panel
     }
@@ -129,60 +206,163 @@ impl CommunityPanel {
 
         if channel != self.channel {
             self.channel = channel;
-            self.messages.clear();
-            self.list_state.reset(0);
+            self.rows.clear();
+            self.has_more = false;
+            self.loading = false;
+            self.list_state.reset(1);
         }
 
         channel
     }
 
-    /// Page the selected channel's history into the cache, then read it back.
+    /// The list's item count: the load-older row, then every message row.
+    fn item_count(&self) -> usize {
+        self.rows.len() + 1
+    }
+
+    /// A timeline read, `before_ms` exclusive, or `None` for the newest rows.
+    fn read(
+        &self,
+        channel: ChannelId,
+        before_ms: Option<u64>,
+        cx: &App,
+    ) -> Option<Task<Result<Timeline>>> {
+        self.community
+            .read_with(cx, |community, cx| {
+                community.timeline(&channel, before_ms, TIMELINE_PAGE, cx)
+            })
+            .ok()
+    }
+
+    /// A channel round, or `None` once the community is gone.
+    fn sync(
+        &self,
+        channel: ChannelId,
+        intent: Intent,
+        cx: &mut App,
+    ) -> Option<Task<Result<community::Progress>>> {
+        self.community
+            .update(cx, |community, cx| {
+                community.sync_channel(&channel, intent, cx)
+            })
+            .ok()
+    }
+
+    /// Paint the selected channel's cache, then catch it up from the relays.
     fn load(&mut self, window: &mut Window, cx: &mut Context<Self>) {
         let Some(channel) = self.resolve_channel(cx) else {
             return;
         };
 
-        let Ok(backfill) = self
-            .community
-            .read_with(cx, |community, cx| community.backfill(&channel, cx))
-        else {
+        self.reload(window, cx);
+
+        // Opening a channel twice in a breath asks the relays once.
+        if self.due(channel, cx) {
+            self.round(channel, Intent::CatchUp, window, cx);
+        }
+    }
+
+    /// Whether the community would actually round `channel`, or just serve it.
+    fn due(&self, channel: ChannelId, cx: &App) -> bool {
+        self.community
+            .read_with(cx, |community, _cx| community.due(&channel))
+            .unwrap_or(true)
+    }
+
+    /// Why the room is not showing messages, when it is not simply empty.
+    fn notice(&self, cx: &App) -> Option<Notice> {
+        let channel = self.channel?;
+
+        self.community
+            .read_with(cx, |community, _cx| {
+                if community.stranded() {
+                    return Some(Notice::Stranded);
+                }
+
+                if let Some(epoch) = community.removed_at() {
+                    return Some(Notice::Removed(epoch));
+                }
+
+                if let Some(epoch) = community.channel_removed_at(&channel) {
+                    return Some(Notice::ChannelRemoved(epoch));
+                }
+
+                if let Some(epoch) = community.missing_key(&channel) {
+                    return Some(Notice::MissingKey(epoch));
+                }
+
+                if community
+                    .progress(&channel)
+                    .is_some_and(|progress| progress.failed && progress.errors > 0)
+                {
+                    return Some(Notice::Unreachable);
+                }
+
+                let unreadable = community.unreadable(&channel);
+
+                (unreadable > 0).then_some(Notice::Unreadable(unreadable))
+            })
+            .ok()
+            .flatten()
+    }
+
+    /// Run a round for `channel`, its completion re-reads the timeline.
+    fn round(
+        &mut self,
+        channel: ChannelId,
+        intent: Intent,
+        window: &mut Window,
+        cx: &mut Context<Self>,
+    ) {
+        let Some(round) = self.sync(channel, intent, cx) else {
             return;
         };
 
-        self.tasks.push(cx.spawn_in(window, async move |this, cx| {
-            if let Err(error) = backfill.await {
-                log::warn!("community panel: backfill failed: {error}");
-            }
+        self.loading = true;
+        cx.notify();
 
-            this.update_in(cx, |this, window, cx| this.reload(window, cx))?;
+        let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
+            let result = round.await;
+
+            this.update_in(cx, |this, window, cx| {
+                this.loading = false;
+
+                if let Err(error) = result {
+                    window.push_notification(
+                        Notification::error(error.to_string()).autohide(false),
+                        cx,
+                    );
+                }
+            })?;
 
             Ok(())
-        }));
+        });
+
+        self.tasks.push(task);
     }
 
-    /// Replace the timeline with the selected channel's folded messages.
+    /// Run the newest round again after a relay failure.
+    fn retry(&mut self, window: &mut Window, cx: &mut Context<Self>) {
+        let Some(channel) = self.channel else {
+            return;
+        };
+
+        self.round(channel, Intent::CatchUp, window, cx);
+    }
+
+    /// Read the newest page and fold it into what is on screen.
     fn reload(&mut self, window: &mut Window, cx: &mut Context<Self>) {
         let Some(channel) = self.resolve_channel(cx) else {
             return;
         };
 
-        let Ok(messages) = self
-            .community
-            .read_with(cx, |community, cx| community.messages(&channel, cx))
-        else {
+        let Some(timeline) = self.read(channel, None, cx) else {
             return;
         };
 
-        self.tasks.push(cx.spawn_in(window, async move |this, cx| {
-            match messages.await {
-                Ok(messages) => {
-                    this.update(cx, |this, cx| {
-                        this.messages = messages;
-                        this.list_state.reset(this.messages.len());
-                        this.list_state.scroll_to_end();
-                        cx.notify();
-                    })?;
-                }
+        let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
+            match timeline.await {
+                Ok(timeline) => this.update(cx, |this, cx| this.apply(channel, timeline, cx))?,
                 Err(error) => {
                     this.update_in(cx, |_this, window, cx| {
                         window.push_notification(
@@ -194,7 +374,180 @@ impl CommunityPanel {
             }
 
             Ok(())
-        }));
+        });
+
+        self.tasks.push(task);
+    }
+
+    /// Splice the page of history above the oldest row on screen.
+    fn load_older(&mut self, window: &mut Window, cx: &mut Context<Self>) {
+        if self.loading || !self.has_more {
+            return;
+        }
+
+        let Some(channel) = self.channel else {
+            return;
+        };
+
+        let Some(before_ms) = self
+            .rows
+            .first()
+            .map(|message| message.at_ms.saturating_sub(1))
+        else {
+            return;
+        };
+
+        let Some(page) = self.read(channel, Some(before_ms), cx) else {
+            return;
+        };
+
+        self.loading = true;
+
+        let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
+            let timeline = match page.await {
+                Ok(timeline) => timeline,
+                Err(error) => {
+                    this.update_in(cx, |this, window, cx| {
+                        this.loading = false;
+                        window.push_notification(
+                            Notification::error(error.to_string()).autohide(false),
+                            cx,
+                        );
+                    })?;
+                    return Ok(());
+                }
+            };
+
+            let swept = this.update(cx, |this, cx| {
+                this.prepend(channel, timeline, cx);
+                !this.has_more
+            })?;
+
+            if !swept {
+                this.update(cx, |this, _cx| this.loading = false)?;
+                return Ok(());
+            }
+
+            let round = this.update(cx, |this, cx| {
+                this.sync(
+                    channel,
+                    Intent::Older {
+                        pages: LOAD_OLDER_PAGES,
+                    },
+                    cx,
+                )
+            })?;
+
+            let Some(round) = round else {
+                this.update(cx, |this, _cx| this.loading = false)?;
+                return Ok(());
+            };
+
+            match round.await {
+                Ok(_) => {
+                    let page =
+                        this.update(cx, |this, cx| this.read(channel, Some(before_ms), cx))?;
+
+                    if let Some(page) = page {
+                        match page.await {
+                            Ok(timeline) => {
+                                this.update(cx, |this, cx| this.prepend(channel, timeline, cx))?
+                            }
+                            Err(error) => this.update_in(cx, |_this, window, cx| {
+                                window.push_notification(
+                                    Notification::error(error.to_string()).autohide(false),
+                                    cx,
+                                );
+                            })?,
+                        }
+                    }
+
+                    this.update(cx, |this, _cx| this.loading = false)?;
+                }
+                Err(error) => {
+                    this.update_in(cx, |this, window, cx| {
+                        this.loading = false;
+                        window.push_notification(
+                            Notification::error(error.to_string()).autohide(false),
+                            cx,
+                        );
+                    })?;
+                }
+            }
+
+            Ok(())
+        });
+
+        self.tasks.push(task);
+    }
+
+    /// Fold a freshly read window into the rows on screen.
+    fn apply(&mut self, channel: ChannelId, timeline: Timeline, cx: &mut Context<Self>) {
+        if self.channel != Some(channel) {
+            return;
+        }
+
+        let Timeline { messages, has_more } = timeline;
+
+        let connected = self
+            .rows
+            .last()
+            .is_some_and(|last| messages.iter().any(|message| message.id == last.id));
+
+        if !connected {
+            self.rows = messages;
+            self.has_more = has_more;
+            self.list_state.reset(self.item_count());
+            cx.notify();
+            return;
+        }
+
+        self.merge(messages);
+        cx.notify();
+    }
+
+    /// Splice older rows in above what is on screen.
+    fn prepend(&mut self, channel: ChannelId, timeline: Timeline, cx: &mut Context<Self>) {
+        if self.channel != Some(channel) {
+            return;
+        }
+
+        self.has_more = timeline.has_more;
+        self.merge(timeline.messages);
+        cx.notify();
+    }
+
+    /// Fold read rows into what is on screen, keeping the list in time order.
+    fn merge(&mut self, messages: Vec<ChatMessage>) {
+        let mut shown: HashMap<EventId, usize> = self
+            .rows
+            .iter()
+            .enumerate()
+            .map(|(ix, message)| (message.id, ix))
+            .collect();
+
+        let mut fresh = Vec::new();
+
+        for message in messages {
+            match shown.get(&message.id).copied() {
+                Some(ix) => self.rows[ix] = message,
+                None => fresh.push(message),
+            }
+        }
+
+        for message in fresh {
+            if shown.contains_key(&message.id) {
+                continue;
+            }
+
+            let at = self
+                .rows
+                .partition_point(|row| (row.at_ms, row.id) <= (message.at_ms, message.id));
+
+            shown.insert(message.id, at);
+            self.rows.insert(at, message);
+            self.list_state.splice(at + 1..at + 1, 1);
+        }
     }
 
     fn send(&mut self, window: &mut Window, cx: &mut Context<Self>) {
@@ -209,6 +562,11 @@ impl CommunityPanel {
             return;
         };
 
+        if let Some(notice) = self.notice(cx).filter(|notice| !notice.writable()) {
+            window.push_notification(Notification::error(notice.to_string()).autohide(false), cx);
+            return;
+        }
+
         let Ok(send) = self.community.read_with(cx, |community, cx| {
             community.send(&channel, &content, None, cx)
         }) else {
@@ -224,7 +582,7 @@ impl CommunityPanel {
             input.set_value("", window, cx);
         });
 
-        self.tasks.push(cx.spawn_in(window, async move |this, cx| {
+        let task = cx.spawn_in::<_, Result<()>>(window, async move |this, cx| {
             match send.await {
                 Ok(_) => {
                     this.update_in(cx, |this, window, cx| this.reload(window, cx))?;
@@ -240,7 +598,59 @@ impl CommunityPanel {
             }
 
             Ok(())
-        }));
+        });
+
+        self.tasks.push(task);
+    }
+
+    /// The honest reason this room has nothing to show, with a way out of it.
+    fn render_notice(&self, notice: Notice, cx: &mut Context<Self>) -> AnyElement {
+        h_flex()
+            .w_full()
+            .justify_center()
+            .items_center()
+            .gap_2()
+            .px_3()
+            .py_2()
+            .text_sm()
+            .text_color(cx.theme().text_placeholder)
+            .child(notice.to_string())
+            .when(notice == Notice::Unreachable, |this| {
+                this.child(
+                    Button::new("retry-round")
+                        .label("Retry")
+                        .ghost()
+                        .small()
+                        .loading(self.loading)
+                        .on_click(cx.listener(|this, _event, window, cx| this.retry(window, cx))),
+                )
+            })
+            .into_any_element()
+    }
+
+    /// The row at index 0: the affordance that pages older history in.
+    fn render_older(&self, cx: &mut Context<Self>) -> AnyElement {
+        if !self.has_more {
+            return div().into_any_element();
+        }
+
+        h_flex()
+            .w_full()
+            .justify_center()
+            .py_2()
+            .child(
+                Button::new("load-older")
+                    .label(if self.loading {
+                        "Loading earlier messages…"
+                    } else {
+                        "Load earlier messages"
+                    })
+                    .ghost()
+                    .small()
+                    .loading(self.loading)
+                    .on_click(cx.listener(|this, _event, window, cx| this.load_older(window, cx))),
+            )
+            .into_any_element()
     }
 
     fn render_message(
@@ -249,13 +659,20 @@ impl CommunityPanel {
         _window: &mut Window,
         cx: &mut Context<Self>,
     ) -> AnyElement {
-        let Some(message) = self.messages.get(ix) else {
+        if ix == 0 {
+            return self.render_older(cx);
+        }
+
+        let Some(message) = self.rows.get(ix - 1) else {
             return div().into_any_element();
         };
+
         message::render(ix, message, cx)
     }
 
     fn render_composer(&self, cx: &mut Context<Self>) -> impl IntoElement {
+        let writable = self.notice(cx).is_none_or(|notice| notice.writable());
+
         h_flex()
             .flex_shrink_0()
             .w_full()
@@ -269,6 +686,7 @@ impl CommunityPanel {
                     .tooltip("Send")
                     .ghost()
                     .large()
+                    .disabled(!writable)
                     .on_click(cx.listener(|this, _event, window, cx| {
                         this.send(window, cx);
                     })),
@@ -323,17 +741,26 @@ impl Render for CommunityPanel {
                     .min_h_0()
                     .relative()
                     .map(|this| {
-                        if self.messages.is_empty() {
+                        let notice = self.notice(cx);
+
+                        if self.rows.is_empty() {
                             this.child(
-                                h_flex()
-                                    .size_full()
-                                    .justify_center()
-                                    .text_sm()
-                                    .text_color(cx.theme().text_placeholder)
-                                    .child("No messages yet"),
+                                v_flex().size_full().justify_center().child(match notice {
+                                    Some(notice) => self.render_notice(notice, cx),
+                                    None => h_flex()
+                                        .size_full()
+                                        .justify_center()
+                                        .text_sm()
+                                        .text_color(cx.theme().text_placeholder)
+                                        .child("No messages yet")
+                                        .into_any_element(),
+                                }),
                             )
                         } else {
-                            this.child(
+                            this.when_some(notice, |this, notice| {
+                                this.child(self.render_notice(notice, cx))
+                            })
+                            .child(
                                 list(
                                     self.list_state.clone(),
                                     cx.processor(move |this, ix, window, cx| {

crates/concord/Cargo.toml

@@ -18,8 +18,6 @@ rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 anyhow.workspace = true
-log.workspace = true
 
 [dev-dependencies]
-nostr-memory.workspace = true
 smol.workspace = true

crates/concord/src/cords/cord02/guestbook.rs

@@ -187,10 +187,15 @@ pub fn open(
     Ok((opened, rumor))
 }
 
+/// Coalesce the guestbook flat: one final state per npub, the latest entry
+/// winning by millisecond time, ties broken by the lower rumor id.
+///
+/// `snapshot_authorities` are the npubs whose refounding is known to have minted an epoch this client reads.
+/// A snapshot chunk is honored only from one of them, and an empty set honors no snapshot at all.
 pub fn coalesce(
     rumors: &[GuestbookRumor],
     now_ms: u64,
-    snapshot_authority: Option<&PublicKey>,
+    snapshot_authorities: &BTreeSet<PublicKey>,
     can_kick: impl Fn(&PublicKey, &PublicKey, Option<&AuthorityCitation>) -> bool,
 ) -> BTreeMap<PublicKey, MemberState> {
     let mut states: BTreeMap<PublicKey, (u64, Reverse<EventId>, MemberState)> = BTreeMap::new();
@@ -250,7 +255,7 @@ pub fn coalesce(
                 at_ms,
                 ..
             } => {
-                if snapshot_authority != Some(refounder) {
+                if !snapshot_authorities.contains(refounder) {
                     continue;
                 }
 
@@ -457,6 +462,11 @@ mod tests {
         CommunityId::from_bytes([0x11u8; 32])
     }
 
+    /// The refounders a fold is told about: a snapshot seeds members on theirs alone.
+    fn refounders(keys: &[&Keys]) -> BTreeSet<PublicKey> {
+        keys.iter().map(|keys| keys.public_key()).collect()
+    }
+
     fn group() -> GroupKey {
         guestbook_group_key(&ROOT, &community(), Epoch(0)).expect("derives")
     }
@@ -533,7 +543,7 @@ mod tests {
                 citation.is_some() && actor == &carol.public_key() && target != &owner.public_key()
             };
 
-        let states = coalesce(&rumors, AT + 8_000, Some(&carol.public_key()), can_kick);
+        let states = coalesce(&rumors, AT + 8_000, &refounders(&[&carol]), can_kick);
 
         assert_eq!(
             states.get(&alice.public_key()),
@@ -562,7 +572,7 @@ mod tests {
 
         let reversed: Vec<GuestbookRumor> = rumors.iter().rev().cloned().collect();
         assert_eq!(
-            coalesce(&reversed, AT + 8_000, Some(&carol.public_key()), can_kick),
+            coalesce(&reversed, AT + 8_000, &refounders(&[&carol]), can_kick),
             states,
             "arrival order cannot change the fold"
         );
@@ -649,7 +659,7 @@ mod tests {
             ),
         ];
 
-        let states = coalesce(&rumors, AT + 1_000, None, can_kick);
+        let states = coalesce(&rumors, AT + 1_000, &BTreeSet::new(), can_kick);
 
         assert_eq!(
             states.get(&kicked.public_key()),
@@ -686,21 +696,21 @@ mod tests {
         )
         .remove(0);
 
-        for authority in [None, Some(refounder.public_key())] {
+        for authority in [BTreeSet::new(), refounders(&[&refounder])] {
             let states = coalesce(
                 &[
                     publish(&by_refounder, &refounder),
                     publish(&by_impostor, &impostor),
                 ],
                 AT + 1_000,
-                authority.as_ref(),
+                &authority,
                 |_, _, _| true,
             );
 
             assert_eq!(
                 states.contains_key(&seeded.public_key()),
-                authority.is_some(),
-                "only the epoch's refounder seeds, and there is no owner fallback"
+                !authority.is_empty(),
+                "only a known refounder seeds, and there is no owner fallback"
             );
             assert!(
                 !states.contains_key(&smuggled.public_key()),
@@ -724,11 +734,11 @@ mod tests {
             &member,
         );
         assert!(
-            coalesce(&[future], AT, None, |_, _, _| true).is_empty(),
+            coalesce(&[future], AT, &BTreeSet::new(), |_, _, _| true).is_empty(),
             "an entry more than an hour ahead is dropped"
         );
         assert_eq!(
-            coalesce(&[horizon], AT, None, |_, _, _| true).len(),
+            coalesce(&[horizon], AT, &BTreeSet::new(), |_, _, _| true).len(),
             1,
             "the horizon itself is skew, not forgery"
         );

crates/concord/src/cords/cord02/mod.rs

@@ -781,7 +781,7 @@ mod tests {
     use crate::cord04::pins;
     use crate::cord04::roles::{Grant, MAX_BANLIST, MAX_ROLES_PER_MEMBER, Role, RoleScope};
     use crate::derive::{channel_group_key, grant_locator};
-    use crate::store::CommunityState;
+    use crate::state::CommunityState;
     use crate::{Extra, RoleId};
 
     const AT: u64 = 1_700_000_000;

crates/concord/src/cords/cord03.rs

@@ -25,6 +25,9 @@ pub const KIND_TIMER_NOTICE: u16 = 1740;
 pub const KIND_WEBXDC: u16 = 3310;
 pub const KIND_TYPING: u16 = 23311;
 
+pub const ROW_KINDS: [u16; 4] = [KIND_MESSAGE, KIND_FILE, KIND_COMMENT, KIND_TIMER_NOTICE];
+pub const SIDE_KINDS: [u16; 3] = [KIND_DELETE, KIND_REACTION, KIND_EDIT];
+
 const TAG_QUOTE: &str = "q";
 const TAG_TARGET: &str = "e";
 const TAG_TARGET_KIND: &str = "k";

crates/concord/src/cords/cord06.rs

@@ -534,6 +534,43 @@ pub fn plan_refounding(epoch: Epoch) -> Result<Refounding> {
     })
 }
 
+/// The material one rotation delivers, by scope.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RotationPlan {
+    /// A refounding: a fresh root, with the root that reads and signs under it.
+    Base(Refounding),
+    /// A channel: a fresh channel key, and nothing beside it.
+    Channel { epoch: Epoch, new_key: [u8; 32] },
+}
+
+impl RotationPlan {
+    pub fn epoch(&self) -> Epoch {
+        match self {
+            Self::Base(refounding) => refounding.epoch,
+            Self::Channel { epoch, .. } => *epoch,
+        }
+    }
+
+    /// The secret the rotation delivers, which becomes the scope's new read key.
+    pub fn new_key(&self) -> [u8; 32] {
+        match self {
+            Self::Base(refounding) => refounding.new_root,
+            Self::Channel { new_key, .. } => *new_key,
+        }
+    }
+}
+
+/// Mint what a rotation of `scope` delivers at `epoch`.
+pub fn plan_rotation(scope: RekeyScope, epoch: Epoch) -> Result<RotationPlan> {
+    match scope {
+        RekeyScope::Base => Ok(RotationPlan::Base(plan_refounding(epoch)?)),
+        RekeyScope::Channel(_) => Ok(RotationPlan::Channel {
+            epoch,
+            new_key: random_32()?,
+        }),
+    }
+}
+
 /// Carries the settled heads across a refounding.
 pub fn compact(
     seals: &[Event],

crates/concord/src/lib.rs

@@ -2,11 +2,11 @@ mod cords;
 mod types;
 mod utils;
 
-pub mod store;
+pub mod state;
 
 pub use cords::{cord01, cord02, cord03, cord04, cord05, cord06};
-pub use types::{ChannelId, CommunityId, Epoch, RoleId};
-pub use utils::derive::{self, GroupKey};
-
 pub(crate) use types::Extra;
-pub(crate) use utils::{decode_hex_32, decode_hex_lower, fill_random, random_32};
+pub use types::{ChannelId, CommunityId, Epoch, RoleId};
+pub use utils::decode_hex_32;
+pub use utils::derive::{self, GroupKey};
+pub(crate) use utils::{decode_hex_lower, fill_random, random_32};

crates/concord/src/state.rs

@@ -1,140 +1,43 @@
-use std::cmp::Reverse;
 use std::collections::{BTreeMap, BTreeSet};
-use std::sync::LazyLock;
 
-use anyhow::{Result, anyhow};
+use anyhow::Result;
 use data_encoding::HEXLOWER;
 use nostr_sdk::prelude::*;
 use serde::{Deserialize, Serialize};
 
-use crate::cord01::{KIND_WRAP_EPHEMERAL, OpenedStream};
 use crate::cord02::list::{CommunityListEntry, JoinMaterial};
 use crate::cord02::{
     ChannelMetadata, CommunityGenesis, CommunityMetadata, ControlFold, ROOT_EPOCH,
 };
-use crate::cord03::{self, ChatRumor, plane_keys};
 use crate::cord04::{EntityHead, Floors, ParsedEdition, vsk};
 use crate::cord05::ChannelGrant;
 use crate::derive::control_signer_group_key;
-use crate::{ChannelId, CommunityId, Epoch, Extra, GroupKey, decode_hex_32};
+use crate::{ChannelId, CommunityId, Epoch, Extra, decode_hex_32};
 
-static LOCAL_KEYS: LazyLock<Keys> = LazyLock::new(Keys::generate);
-
-const MAX_PAGES: usize = 8;
-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: &str = "k";
 /// The `concord/` namespace for locally-keyed documents.
 pub const STATE_PREFIX: &str = "concord/";
 
-/// 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, [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)
+/// A key epoch the client still holds, retained so history stays readable.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct HeldKey {
+    pub epoch: Epoch,
+    pub key: [u8; 32],
+    /// The publish time of the rotation that superseded this key.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retired_at: Option<Timestamp>,
 }
 
-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)
-}
-
-pub async fn query_rumors(
-    client: &Client,
-    channel: &ChannelId,
-    until: Option<Timestamp>,
-    limit: usize,
-) -> 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(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)
+/// A community root epoch the client still holds, retained for the same reason.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct HeldRoot {
+    pub epoch: Epoch,
+    pub key: [u8; 32],
+    /// The epoch's Control Plane signer, when the rotation delivered one.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub control_pk: Option<PublicKey>,
+    /// The publish time of the rotation that superseded this root.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retired_at: Option<Timestamp>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -144,10 +47,58 @@ pub struct ChannelKeyRef {
     pub private: bool,
     pub epoch: Epoch,
     /// The channel's read secret when the member was granted it.
-    ///
-    /// A public channel derives its key from the `community_root` and carries none.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub key: Option<[u8; 32]>,
+    /// Keys this one superseded, retained so a rotation never blanks history.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub priors: Vec<HeldKey>,
+}
+
+impl ChannelKeyRef {
+    /// The write coordinate: only the current epoch is ever published under.
+    pub fn current(&self) -> Option<(Epoch, [u8; 32])> {
+        self.key.map(|key| (self.epoch, key))
+    }
+}
+
+/// How far a channel's history sync has reached, in wrap times.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ChannelCursor {
+    /// The newest wrap ingested, so a live subscription knows where to resume.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub newest: Option<Timestamp>,
+    /// The oldest wrap paged back to, so the next round resumes below it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub oldest: Option<Timestamp>,
+    /// History verifiably swept to the bottom.
+    #[serde(default)]
+    pub exhausted: bool,
+}
+
+impl ChannelCursor {
+    pub fn merge(self, round: Self) -> Self {
+        Self {
+            newest: later(self.newest, round.newest),
+            oldest: earlier(self.oldest, round.oldest),
+            exhausted: self.exhausted || round.exhausted,
+        }
+    }
+}
+
+fn later(held: Option<Timestamp>, round: Option<Timestamp>) -> Option<Timestamp> {
+    match (held, round) {
+        (Some(held), Some(round)) => Some(held.max(round)),
+        (held, None) => held,
+        (None, round) => round,
+    }
+}
+
+fn earlier(held: Option<Timestamp>, round: Option<Timestamp>) -> Option<Timestamp> {
+    match (held, round) {
+        (Some(held), Some(round)) => Some(held.min(round)),
+        (held, None) => held,
+        (None, round) => round,
+    }
 }
 
 /// One local document per community, keyed by `concord/<community_id>`.
@@ -171,8 +122,31 @@ pub struct CommunityState {
     pub heads: Vec<EntityHead>,
     #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
     pub banned: BTreeSet<PublicKey>,
+    /// Where each channel's history sync has reached.
+    #[serde(
+        default,
+        rename = "channel_cursors",
+        skip_serializing_if = "BTreeMap::is_empty"
+    )]
+    pub cursors: BTreeMap<ChannelId, ChannelCursor>,
+    /// Root epochs the community has rotated past that this client still holds.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub held_roots: Vec<HeldRoot>,
+    /// The epoch a channel rotation removed us at.
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    pub channel_cuts: BTreeMap<ChannelId, Epoch>,
+    /// The npubs whose rotation minted an epoch of this community we verified.
+    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
+    pub refounders: BTreeSet<PublicKey>,
+    /// The base epoch we were excluded at.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub removed_at: Option<Epoch>,
+    /// A complete rotation ahead of our epoch predates our join and carries no blob.
+    #[serde(default)]
+    pub stranded: bool,
     #[serde(default)]
     pub dissolved: bool,
+    /// When this community joined the member's list, in milliseconds.
     pub added_at_ms: u64,
 }
 
@@ -214,6 +188,7 @@ impl CommunityState {
                         private: metadata.private,
                         epoch: ROOT_EPOCH,
                         key: None,
+                        priors: Vec::new(),
                     });
                 }
                 _ => {}
@@ -243,6 +218,12 @@ impl CommunityState {
             relays,
             heads,
             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,
         })
@@ -255,6 +236,7 @@ impl CommunityState {
         };
 
         let mut channels = Vec::with_capacity(material.channels.len());
+
         for grant in &material.channels {
             let key = match &grant.key {
                 Some(key) => Some(decode_hex_32(key)?),
@@ -267,6 +249,7 @@ impl CommunityState {
                 private: key.is_some(),
                 epoch: grant.epoch,
                 key,
+                priors: Vec::new(),
             });
         }
 
@@ -290,6 +273,12 @@ impl CommunityState {
                 .collect(),
             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,
         })
@@ -299,6 +288,56 @@ impl CommunityState {
         state_identifier(&self.id)
     }
 
+    /// Every root epoch we hold, the current one first.
+    pub fn roots(&self) -> Vec<HeldRoot> {
+        let mut roots = Vec::with_capacity(self.held_roots.len() + 1);
+        roots.push(HeldRoot {
+            epoch: self.root_epoch,
+            key: self.community_root,
+            control_pk: self.control_pks.get(&self.root_epoch.0).copied(),
+            retired_at: None,
+        });
+        roots.extend(self.held_roots.iter().copied());
+        roots
+    }
+
+    /// Every secret held for a channel, newest epoch first.
+    pub fn held_keys(&self, channel: &ChannelId) -> Vec<HeldKey> {
+        let Some(held) = self.channels.iter().find(|held| held.id == *channel) else {
+            return Vec::new();
+        };
+
+        if held.private {
+            let mut keys: Vec<HeldKey> = held
+                .key
+                .map(|key| HeldKey {
+                    epoch: held.epoch,
+                    key,
+                    retired_at: None,
+                })
+                .into_iter()
+                .collect();
+            keys.extend(held.priors.iter().copied());
+            return keys;
+        }
+
+        self.roots()
+            .into_iter()
+            .map(|root| HeldKey {
+                epoch: root.epoch,
+                key: root.key,
+                retired_at: root.retired_at,
+            })
+            .collect()
+    }
+
+    /// Whether a channel rotation removed us at or after `epoch`.
+    pub fn channel_cut(&self, channel: &ChannelId, epoch: Epoch) -> bool {
+        self.channel_cuts
+            .get(channel)
+            .is_some_and(|cut| epoch <= *cut)
+    }
+
     pub fn floors(&self) -> Floors {
         self.heads
             .iter()
@@ -342,6 +381,7 @@ impl CommunityState {
                     private: false,
                     epoch: self.root_epoch,
                     key: None,
+                    priors: Vec::new(),
                 }),
                 None => {}
             }
@@ -388,249 +428,118 @@ fn label(name: &str) -> Option<String> {
     (!trimmed.is_empty()).then(|| trimmed.to_owned())
 }
 
-fn state_identifier(id: &CommunityId) -> String {
+/// The local document key a community's state is stored under.
+pub fn state_identifier(id: &CommunityId) -> String {
     format!("{STATE_PREFIX}{}", id.to_hex())
 }
 
-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()
-}
-
-pub async fn backfill(
-    client: &Client,
-    channel: &ChannelId,
-    held: &[(Epoch, [u8; 32])],
-    until: Option<Timestamp>,
-    limit: usize,
-) -> Result<Vec<ChatRumor>> {
-    let planes = plane_keys(held, channel)?;
-    let authors: Vec<PublicKey> = planes.iter().map(|(_, group)| group.pk()).collect();
-
-    let mut cursor = until;
-    let mut seen: BTreeSet<EventId> = BTreeSet::new();
-    let mut found: Vec<ChatRumor> = Vec::new();
-
-    for _ in 0..MAX_PAGES {
-        let page = fetch_page(client, &authors, cursor, limit).await?;
-
-        if page.is_empty() {
-            break;
-        }
-
-        let (fresh, next) = advance(&page, &planes, channel, cursor, limit, &mut seen);
-
-        for (opened, rumor) in fresh {
-            if cache_rumor(client, channel, &opened).await? {
-                found.push(rumor);
-            }
-        }
-
-        match next {
-            Some(next) => cursor = Some(next),
-            None => break,
-        }
-    }
-
-    found.sort_by_key(|rumor| (Reverse(rumor.at_ms), rumor.id));
-    found.truncate(limit);
-
-    Ok(found)
-}
-
-fn advance(
-    page: &BTreeSet<Event>,
-    planes: &[(Epoch, GroupKey)],
-    channel: &ChannelId,
-    cursor: Option<Timestamp>,
-    limit: usize,
-    seen: &mut BTreeSet<EventId>,
-) -> (Vec<(OpenedStream, ChatRumor)>, Option<Timestamp>) {
-    let mut fresh = Vec::new();
-
-    for wrap in page {
-        let Some((epoch, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
-        else {
-            continue;
-        };
-
-        let Ok((opened, rumor)) = cord03::open(wrap, group, channel, *epoch) else {
-            continue;
-        };
-
-        if seen.insert(rumor.id) {
-            fresh.push((opened, rumor));
-        }
-    }
-
-    if fresh.is_empty() || page.len() < limit {
-        return (fresh, None);
-    }
-
-    let oldest = page.iter().map(|event| event.created_at).min();
-
-    match oldest {
-        Some(oldest) if cursor != Some(oldest) => (fresh, Some(oldest)),
-        _ => (fresh, None),
-    }
-}
-
-async fn fetch_page(
-    client: &Client,
-    authors: &[PublicKey],
-    until: Option<Timestamp>,
-    limit: usize,
-) -> Result<BTreeSet<Event>> {
-    let mut filter = Filter::new()
-        .kinds([Kind::GiftWrap, Kind::Custom(KIND_WRAP_EPHEMERAL)])
-        .authors(authors.iter().copied())
-        .limit(limit);
-
-    if let Some(until) = until {
-        filter = filter.until(until);
-    }
-
-    Ok(client.fetch_events(filter).await?)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::cord03::{build_message, seal_rumor};
-    use crate::cord05::ChannelGrant;
-    use crate::derive::channel_group_key;
-    use crate::{Epoch, Extra};
-
-    const SECRET: [u8; 32] = [0x07u8; 32];
-    const NEXT_SECRET: [u8; 32] = [0x11u8; 32];
-
-    /// What a relay does with an inclusive `until` and a `limit`.
-    fn serve_page(
-        relay: &BTreeSet<Event>,
-        cursor: Option<Timestamp>,
-        limit: usize,
-    ) -> BTreeSet<Event> {
-        let mut events: Vec<Event> = relay
-            .iter()
-            .filter(|event| cursor.is_none_or(|cursor| event.created_at <= cursor))
-            .cloned()
-            .collect();
-
-        events.sort_by_key(|event| Reverse(event.created_at));
-        events.truncate(limit);
-        events.into_iter().collect()
-    }
 
     #[test]
-    fn history_pages_back_across_a_rekey() {
-        let channel = ChannelId::from_bytes([0x9cu8; 32]);
-        let author = Keys::generate();
-        let held = [(Epoch(0), SECRET), (Epoch(1), NEXT_SECRET)];
-        let planes = plane_keys(&held, &channel).expect("derives");
+    fn a_cursor_merge_only_moves_forward_and_never_seals() {
+        let held = ChannelCursor {
+            newest: Some(Timestamp::from_secs(1_000)),
+            oldest: Some(Timestamp::from_secs(5_000)),
+            exhausted: false,
+        };
 
-        // 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();
+        // An incomplete round reports nothing and moves neither bound.
+        assert_eq!(held.merge(ChannelCursor::default()), held);
 
-        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 seen = BTreeSet::new();
-        let mut found = Vec::new();
-        let mut cursor = None;
-
-        for _ in 0..3 {
-            let page = serve_page(&relay, cursor, 2);
-            let (fresh, next) = advance(&page, &planes, &channel, cursor, 2, &mut seen);
-
-            found.extend(fresh.into_iter().map(|(_, rumor)| rumor));
-
-            match next {
-                Some(next) => cursor = Some(next),
-                None => break,
-            }
-        }
-
-        found.sort_by_key(|rumor| (Reverse(rumor.at_ms), rumor.id));
-
-        let contents: Vec<&str> = found.iter().map(|rumor| rumor.content.as_str()).collect();
+        let merged = held.merge(ChannelCursor {
+            newest: Some(Timestamp::from_secs(2_000)),
+            oldest: Some(Timestamp::from_secs(3_000)),
+            exhausted: true,
+        });
         assert_eq!(
-            contents,
-            ["after the rekey", "still before", "before the rekey"]
+            merged,
+            ChannelCursor {
+                newest: Some(Timestamp::from_secs(2_000)),
+                oldest: Some(Timestamp::from_secs(3_000)),
+                exhausted: true,
+            }
         );
+
+        // A later round that learned less cannot walk either bound back.
+        assert_eq!(
+            merged.merge(ChannelCursor {
+                newest: Some(Timestamp::from_secs(1_500)),
+                oldest: Some(Timestamp::from_secs(4_000)),
+                exhausted: false,
+            }),
+            merged
+        );
+    }
+
+    /// A cursor stored when the boundaries were milliseconds must not be read as
+    /// seconds. The key it was stored under is gone, so the document's counters
+    /// are ignored and the channel re-syncs rather than being sealed off by an
+    /// `exhausted` that outlived the bounds it was earned against.
+    #[test]
+    fn a_cursor_stored_in_the_old_unit_is_dropped_rather_than_reinterpreted() {
+        let channel = ChannelId::from_bytes([0x9c; 32]);
+        let cursor = ChannelCursor {
+            newest: Some(Timestamp::from_secs(1_700_000_000)),
+            exhausted: true,
+            ..ChannelCursor::default()
+        };
+        let mut state = CommunityState {
+            id: CommunityId::from_bytes([0x42; 32]),
+            name: None,
+            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,
+        };
+
+        // The document a version that stored milliseconds wrote: its own key,
+        // and boundaries padded by a thousand.
+        let mut stored = serde_json::Map::new();
+        stored.insert(
+            channel.to_hex(),
+            serde_json::json!({
+                "newest_ms": 1_700_000_000_000u64,
+                "oldest_ms": 1_699_999_000_000u64,
+                "exhausted": true
+            }),
+        );
+
+        let mut legacy = serde_json::to_value(&state).expect("serializes");
+        legacy
+            .as_object_mut()
+            .expect("a document")
+            .insert("cursors".to_owned(), serde_json::Value::Object(stored));
+
+        let read: CommunityState = serde_json::from_value(legacy).expect("deserializes");
+
+        assert!(
+            read.cursors.is_empty(),
+            "a millisecond cursor is not a seconds cursor"
+        );
+
+        // A typed cursor still round-trips under the key it is written with.
+        state.cursors.insert(channel, cursor);
+
+        let document = serde_json::to_value(&state).expect("serializes");
+        let read: CommunityState = serde_json::from_value(document).expect("deserializes");
+
+        assert_eq!(read.cursors.get(&channel), Some(&cursor));
     }
 
     #[test]
@@ -715,44 +624,4 @@ mod tests {
         assert_eq!(state.control_root, None);
         assert_eq!(state.control_pks, BTreeMap::from([(3, control_pk)]));
     }
-
-    #[test]
-    fn load_states_reads_one_document_per_community_and_ignores_other_documents() {
-        smol::block_on(async {
-            let client = ClientBuilder::default()
-                .database(nostr_memory::MemoryDatabase::unbounded())
-                .build();
-
-            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(),
-                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]);
-        });
-    }
 }

crates/concord/src/utils/mod.rs

@@ -9,8 +9,10 @@ use serde::Serialize;
 
 use crate::Extra;
 
+/// Decode a 64-character lowercase-hex string into 32 bytes.
+///
 /// Uppercase and other non-canonical spellings are rejected.
-pub(crate) fn decode_hex_32(value: &str) -> Result<[u8; 32]> {
+pub fn decode_hex_32(value: &str) -> Result<[u8; 32]> {
     decode_hex_lower::<32>(value)
 }
 

crates/workspace/src/sidebar/mod.rs

@@ -836,6 +836,7 @@ fn channel_row(
 
     div()
         .flex_shrink_0()
+        .rounded(cx.theme().radius)
         .child(
             TreeRow::new(
                 ElementId::Name(SharedString::from(format!(

docs/community-sidebar-plan.md

@@ -1,98 +0,0 @@
-# Community sidebar
-
-When a community is opened from the sidebar, the sidebar's content becomes the
-community's own channel/member browser. Going back restores the normal tabs.
-
-```
-+------------------+--------------------------------+
-| [<-] community   |                                |
-| banner           |                                |
-| v Channels  3    |   messages                     |
-|   # general      |                                |
-|   # random       |                                |
-| v Admins    1    |                                |
-|   @owner         |                                |
-| v Members   2    +--------------------------------+
-|   @alice         |   [ composer ]                 |
-|   @bob           |                                |
-+------------------+--------------------------------+
-```
-
-## Current state
-
-* `community_ui::CommunityPanel` renders its own 220px left column (channels +
-  members) next to the timeline.
-* `Sidebar` renders the Recents/Chats/Communities tabs, and `Sidebar::open_community`
-  records the community as recent and asks `CommunityRegistry` to emit
-  `CommunityEvent::Open`, which `Workspace` turns into a center dock panel.
-* `Community` already folds channels (`state().channels`), members, the control
-  roster (`control().roles`), the community icon, and the banner (`metadata.banner`).
-
-## Plan
-
-### 1. Backend: community crate
-
-`crates/community/src/community.rs`
-
-* Add `CommunityEvent::Close(CommunityId)` and `CommunityEvent::Channel(CommunityId, ChannelId)`.
-* Add `active: Option<ChannelId>` to `Community`, with:
-  * `pub fn active_channel(&self) -> Option<ChannelId>` — the selected channel,
-    defaulting to the first one.
-  * `pub fn set_active_channel(&mut self, ChannelId, &mut Context<Self>)` — stores
-    it and emits `Channel`.
-* Add `banner: Option<PathBuf>` resolved from `metadata.banner` the same way the
-  icon already is, exposed as `pub fn banner(&self) -> Option<PathBuf>`.
-
-`crates/community/src/lib.rs`
-
-* Add `CommunityRegistry::emit_close(&mut self, CommunityId, &mut Window, &mut Context<Self>)`,
-  mirroring `emit_community`, emitting `Close`.
-
-`crates/community/src/sync.rs`
-
-* Rename `resolve_icon` to `resolve_image` (it takes any `ImageRef`).
-
-### 2. `community_ui`: panel becomes timeline-only
-
-* Drop the internal channel/member column and its helpers.
-* Resolve the shown channel from `Community::active_channel` on every load and
-  reload, and subscribe to `CommunityEvent::Channel` to follow sidebar clicks
-  (deferred, because the community is being updated when it emits).
-
-### 3. `workspace`: sidebar renders the community
-
-`crates/workspace/src/sidebar/mod.rs`
-
-* New state: `community: Option<WeakEntity<Community>>` and three section flags.
-* `open_community` also sets `self.community`; new `close_community` clears it and
-  asks the registry to emit `Close`.
-* `render` picks between the existing tabs and `render_community`.
-* `render_user` keeps the account button as usual and adds a back button on the
-  right when a community is open.
-* `render_community` pins the optional banner image and scrolls everything below
-  it: collapsible Channels, Admins (members where `control().roles.is_staff`), and
-  Members sections. Every row is a `TreeRow` — the component the sidebar's own
-  lists use — so height, padding, typography, hover, and selection match, and each
-  row is wrapped `flex_shrink_0` so the list scrolls instead of squashing.
-* Observe the registry so member/roster changes re-render the lists.
-
-`crates/workspace/src/lib.rs`
-
-* Remember the opened `CommunityPanel` so `CommunityEvent::Close` can close it:
-  activate it, focus the group, then dispatch `ClosePanel`.
-
-### 4. `community`: a list refresh keeps entities
-
-`CommunityRegistry::track` used to rebuild every community, so an open panel and a
-browsing sidebar went stale on any list signal — and `sync::load` merges join
-material into a loaded `CommunityState`, so the state really can change. `track`
-now keeps the entity for an id it still tracks and hands it the new state through
-`Community::adopt`, which also cancels a fold in flight that would otherwise write
-the pre-adoption state back. Observers move to a
-`HashMap<CommunityId, Subscription>` so the survivors keep theirs, and the ids the
-list dropped lose their observer and their synced key.
-
-## Out of scope
-
-Moderation, invites, community management, channel icons, unread badges, member
-search, and persisting the selected channel across restarts.

docs/concord-usage.md

@@ -29,11 +29,16 @@ vocabulary are shared substrate — every document calls them — so they live o
 | `cord05` | Invite bundles, links, the Direct Invite, the Invite List |
 | `cord06` | Key rotations, refounding, compaction, dissolution |
 | `derive` | Every frozen HKDF derivation and coordinate |
-| `store` | Local rumor cache, the community state document, relay paging |
+| `state` | The community state document and its pure folds — no I/O |
 
 `CommunityId`, `ChannelId`, `RoleId`, `Epoch` and `Extra` (crate-internal) come from
 the private `types` module and are re-exported at the crate root.
 
+Nothing in `concord` touches a relay or a database. The local rumor cache, the
+state documents and the relay paging walk live one crate up, in
+`community::cache` and `community::history`, and are what a client actually calls
+(read on below).
+
 CORD-07 (audio/video) is unimplemented. CORD-08's timer has no file of its own: it
 lives in the metadata it reads (`cord02`) and the fold it filters (`cord03`).
 
@@ -44,7 +49,8 @@ Read `CommunityId` as "this community", `ChannelId` as "this channel", `Epoch` a
 
 ```rust
 use concord::cord02::{self, CommunityMetadata};
-use concord::store::{self, CommunityState, save_state};
+use concord::state::CommunityState;
+use community::cache::save_state;
 
 let metadata = CommunityMetadata { name: "Room".into(), ..Default::default() };
 let minted = cord02::genesis(&owner_keys, &metadata, now_secs).await?;
@@ -169,8 +175,11 @@ let (wrap, wrap_key) = cord03::seal_rumor(&rumor, &plane, &my_keys, false).await
 client.send_event(&wrap).to(&relays).await?;
 ```
 
-- `epoch` is the channel's current epoch (`state.channels` carries it). A private
-  channel derives from its own key instead of `community_root`.
+- `epoch` is the **root** epoch for a public channel: its plane is
+  `channel_group_key(community_root, channel, root_epoch)`, so a Refounding moves
+  it along with the root. A private channel passes its own current channel epoch
+  (`state.channels` carries it) and derives from its own key instead of
+  `community_root`.
 - `timer` is `control.community.message_expiration`; pass `None` when it is off.
   The builder attaches the NIP-40 tag and `seal_rumor` mirrors it onto the wrap,
   so relays drop the ciphertext too.
@@ -187,6 +196,7 @@ about an existing `EventId` rather than a mutation.
 
 ```rust
 use concord::cord03::{self, fold, plane_keys};
+use community::cache;
 
 let planes = plane_keys(&held, &channel)?;            // &[(Epoch, secret)]
 let mut rumors = Vec::new();
@@ -198,7 +208,7 @@ for wrap in &wraps {
     let Ok((opened, rumor)) = cord03::open(wrap, group, &channel, *epoch) else {
         continue;
     };
-    store::cache_rumor(&client, &channel, &opened).await?;
+    cache::cache_rumor(&client, &channel, &opened).await?;
     rumors.push(rumor);
 }
 
@@ -215,15 +225,85 @@ let messages = fold(&rumors, Timestamp::now(), |actor, citation, author| {
 Relay history pages through the local cache:
 
 ```rust
-let page = store::backfill(client, &channel, &held, until, 50).await?;
-let cached = store::query_rumors(&client, &channel, None, 50).await?;
+use community::history::{self, PageRegistry, Window};
+
+// Every key the client still holds for the channel, newest epoch first: the
+// state's current key (`ChannelKeyRef::key` at `epoch`) plus every `priors`
+// entry a rotation stepped off. A public channel derives one per held root.
+let held: Vec<concord::state::HeldKey> = state.held_keys(&channel);
+
+// The same registry `CommunityRegistry` shares with its notification pump: it is
+// what carries each page's EOSE and CLOSED back to the round waiting on it.
+let page = history::page(
+    client,
+    &pages,
+    &channel,
+    &held,
+    &relays,
+    Window::opening(cursor),
+    20,
+    50,
+)
+.await?;
+let cached = cache::query_rumors(&client, &channel, None, 50, Some(&cord03::ROW_KINDS)).await?;
 ```
 
-`backfill` walks newest-first across every held epoch, caches what it opens, and
-stops on a short page. `query_rumors` is the read path when the group keys are
-gone. Run `store::purge_expired(client, &channel, now)` on the same cadence as
-any other local sweep — the timer is cooperative, so the local store is the
-artifact that has to forget.
+A key's `retired_at` (a `Timestamp`, set when a rotation supersedes it) is a read
+cutoff: a wrap at that epoch with a later `created_at` is refused, so a retired
+epoch is history and never a live plane an ejected holder can keep writing into.
+
+A relay is only ever *asked*: `history::page` installs **one REQ per page** over
+`ReqTarget::manual` for the community's own relays (which is what makes the client
+verify a wrap, deduplicate it and persist it), waits for every relay to settle,
+and then reads the page back out of the local database — no wrap is ever consumed
+straight off the wire, and `fetch_events` is not used anywhere. The wrap lands in
+the client's shared event store through that ordinary ingest path; `cache_rumor`
+then puts the decrypted rumor beside it, and only the rumor is ever served to a
+reader.
+
+EOSE and CLOSED are **not** watched inside the page. The registry's notification
+pump is the only consumer of `client.notifications()`; it delivers one
+`PageReport` per settling relay into the `PageRegistry` under the page's
+subscription id, and the page waits on that channel with `PAGE_TIMEOUT`. The id
+is opaque (`concord-history-<n>`) because a NIP-01 id is capped at 64 characters,
+and the registry is what maps it back to the waiting round.
+
+A page ends on EOSE. A CLOSED ends it as unanswered, *except* NIP-42's
+`auth-required`: the SDK re-issues that REQ under the same subscription id once
+the handshake completes, so the page waits for the resubscribed answer rather
+than writing off a relay that only wanted to authenticate.
+
+`history::page` walks newest-first across every held epoch, caches what it opens, and
+reports what it saw: `oldest`/`newest` feed the caller's `ChannelCursor`,
+`exhausted` is earned only by a short page *after* history was seen, and an
+all-empty answer sets `failed` so a later round re-asks instead of sealing the
+channel at "no more history". `unreadable` counts the wraps the page reached that
+no held key could open — sealed past the cutoff their key's rotation set, or bound
+to another channel — because those are history the reader is missing, not history
+that is not there. Page down with `Window::older_than(seen.oldest)`,
+open a channel with `Window::opening(cursor)` (wide cold, `newest - 60s` warm),
+and read the region between two cursors with `Window::between(..)`. A `Window`
+carries `Timestamp`s, so its bounds go straight into a NIP-01 filter with no unit
+conversion; only a reader's millisecond `at_ms` narrows to a second at the query.
+`query_rumors` is the read path when the group keys are gone; pass `kinds` to
+budget rows apart from the events that only decorate them. `cache::wrapper_index`
+reads the cached rows back keyed by the wrap they came from, which is what lets
+`sync::fold` observe author and message times without re-opening a wrap it already
+cached, and `cache::purge_expired(client, &channel, now)` runs at the top of every
+round — the timer is cooperative, so the local store is the artifact that has to
+forget.
+
+`sync::fold` counts the same thing over the whole store, per channel, in
+`Snapshot.unreadable`. `sync_round` sums a round's pages into
+`Progress.unreadable`, and `Community` keeps both per channel: `progress(channel)`
+is the last completed round, `unreadable(channel)` is the count, and
+`missing_key(channel)`, `channel_removed_at(channel)`, `removed_at()` and
+`stranded()` are the rest of the honest read surface a panel needs to tell an
+empty room from one it cannot read. `Community::due(channel)` is the other half —
+whether a round is worth asking for yet (a round for `Older` and an explicit retry
+never ask it), and `Community::tick`, driven by `CommunityRegistry` once per
+`community::MIN_ROUND_INTERVAL`, re-folds and re-rounds the active channel of a
+community whose last round is older than `STALE_AFTER`.
 
 `ChatAction::TimerNotice { seconds }` is a policy notice, not a message: render it
 as an inline row only when its author passes
@@ -341,25 +421,25 @@ whether a link still stands, and `fits()` is the write gate.
 ## Rekeys, refounding and dissolution
 
 A rotation is authority plus delivery: `rekey_authorized(&control.roles, &owner, &me, permission, &removed)`
-gates it, `plan_refounding(epoch)` mints the new pair, and `build_rekey_chunks`
-seals one blob per remaining member:
+gates it, `plan_rotation(scope, epoch)` mints what it delivers, and
+`build_rekey_chunks` seals one blob per remaining member:
 
 ```rust
 use concord::derive::epoch_key_commitment;
-use concord::cord06::{self, RekeyScope};
+use concord::cord06::{self, RekeyScope, RotationPlan};
 
 let scope = RekeyScope::Channel(channel_id);                    // or RekeyScope::Base
-let plan = cord06::plan_refounding(Epoch(epoch + 1))?;
+let plan = cord06::plan_rotation(scope, Epoch(epoch + 1))?;
+let new_key = plan.new_key();
 
 // A base rotation delivers the new control-plane keys beside the root; a channel
 // rotation delivers only that channel's fresh key.
-let new_key = plan.new_root;
-let (control_pk, control_root) = match scope {
-    RekeyScope::Base => {
-        let pk = plan.signer(&community_id)?.pk().to_bytes();
-        (Some(pk), is_staff.then_some(&plan.new_control_root))
+let (control_pk, control_root) = match &plan {
+    RotationPlan::Base(refounding) => {
+        let pk = refounding.signer(&community_id)?.pk().to_bytes();
+        (Some(pk), is_staff.then_some(&refounding.new_control_root))
     }
-    RekeyScope::Channel(_) => (None, None),
+    RotationPlan::Channel { .. } => (None, None),
 };
 
 let mut blobs = Vec::with_capacity(members.len());
@@ -367,18 +447,24 @@ let mut blobs = Vec::with_capacity(members.len());
 for member in &members {
     blobs.push(
         cord06::build_blob(
-            &my_keys, member, scope, plan.epoch, &new_key, control_pk.as_ref(), control_root,
+            &my_keys,
+            member,
+            scope,
+            plan.epoch(),
+            &new_key,
+            control_pk.as_ref(),
+            control_root,
         )
         .await?,
     );
 }
 
-let rekey_group = cord06::rekey_group(scope, &community_root, &community_id, plan.epoch)?;
+let rekey_group = cord06::rekey_group(scope, &community_root, &community_id, plan.epoch())?;
 let wraps = cord06::build_rekey_chunks(
     &my_keys,
     &rekey_group,
     scope,
-    plan.epoch,
+    plan.epoch(),
     Epoch(epoch),
     &epoch_key_commitment(Epoch(epoch), &community_root),
     &blobs,
@@ -396,6 +482,24 @@ member finds their delivery with `find_my_blobs` / `open_blob`, and adopts the k
 only if the plaintext binds to the scope and epoch they expect and its `prevcommit`
 matches the key they already hold. Two concurrent rotations settle on `fork_winner`.
 
+`community::rekey::adopt` is that receiver, run against the local database: it
+walks a scope forward one epoch at a time off the key actually held (never
+waiving a gap, never adopting a fork), keeps each stepped-off key as a prior
+with the rotation's publish time as its read cutoff, and reports a removal or a
+strand when a complete rotation carries no blob for the member.
+
+`community::rekey::rotate` is the sender, run against the held state: it resolves
+the epoch and key the scope is stepping off, refuses a rotation that would skip or
+cut off its own rotator, delivers one blob per recipient (`Rewrite.recipients`,
+which the caller names because a private channel's audience is not in the local
+state), marks the rotation severed when it excludes somebody, and — for a base
+scope — carries the settled Control Plane heads onto the new epoch's groups with
+`cord06::compact`. `Community::rotate` checks `Rewrite::authorized` first, the same
+`rekey_authorized` under the same permissions `adopt` applies, publishes to the
+community's relays, and then adopts what it published through `rekey::adopt`
+rather than by construction: a rotation this client wrote and one it received take
+one path.
+
 The blob plaintext is a fixed-width binary record, but a signer's NIP-44 is
 text-only, so `build_blob` carries it base64-encoded inside the envelope.
 `open_blob` mirrors that, so the record layout and the `locator` are unchanged.
@@ -419,7 +523,7 @@ A member's own memberships, synced across their devices:
 ```rust
 use concord::cord02::list;
 
-let entry = concord::store::list_entry(&state, &metadata.name);  // state → §8 material
+let entry = concord::state::list_entry(&state, &metadata.name);  // state → §8 material
 let held = list::parse_list_event(&my_keys, &event).await?;       // validates the d tag
 let mine = held.joined(entry);                                    // community_id-keyed union
 let event = list::build_list_event(&my_keys, &mine, 0, now_secs).await?;  // kind 33302, d = 0
@@ -431,7 +535,7 @@ locally *before* it resolves targets, so the fragment is available to the
 `concord/list` read path even if every relay is unreachable.
 
 `join_material(&invite, control_root)` makes the §8 material from a CORD-05
-invite; `store::list_entry(state, name)` makes it from a `CommunityState`, and is
+invite; `concord::state::list_entry(state, name)` makes it from a `CommunityState`, and is
 what a write path uses after a create, join or rename. `joined` and `tombstoned`
 are the two mutations: both are `community_id`-keyed unions, so neither an append
 nor a leave can lose a membership the other writer has.
@@ -538,7 +642,7 @@ self.ingress = Some(cx.background_spawn(async move {
             continue;
         };
         let (opened, rumor) = cord03::open(wrap, &plane.group, &plane.channel, plane.epoch)?;
-        store::cache_rumor(&client, &plane.channel, &opened).await?;
+        cache::cache_rumor(&client, &plane.channel, &opened).await?;
         signal_tx.send_async(Signal::Chat { channel: plane.channel, rumor }).await?;
     }
     Ok(())
@@ -553,7 +657,7 @@ self.consumer = Some(cx.spawn(async move |this, cx| {
 }));
 ```
 
-- Every store function takes the `&Client` and reaches the database through
+- Every cache function takes the `&Client` and reaches the database through
   `client.database()`, so clone the `Client` into the background task.
 - Keep long-lived tasks in fields — dropping a `Task` cancels it. Assign `None`
   to an `Option<Task<_>>` before respawning it; a signer change replaces both
@@ -585,10 +689,24 @@ client.subscribe(filter).with_id(sub_id).await?;
   rekey fold changes it.
 - Route inbound events by `subscription_id` from `RelayMessage::Event`, never by
   kind.
-- Watch one epoch ahead: while holding `root_N`, subscribe to
-  `base_rekey_group_key(&root_N, &community_id, Epoch(N + 1))` and to
-  `channel_rekey_group_key(&root_N, &channel, Epoch(N + 1))` for each private
-  channel. A second epoch ahead is not derivable until the new root arrives.
+- Watch one epoch ahead for the base, and a **window** of channel epochs under
+  every held root for private channels: `base_rekey_group_key(&root_N, &id,
+  Epoch(N + 1))`, plus `channel_rekey_group_key(&root, &channel, Epoch(channel_epoch
+  + ahead))` for `ahead in 1..=8` and each root in `state.roots()`. A Refounding
+  seals its channel rekeys under the root current when it was minted, so a member
+  who adopted the base rotation first must still ask under the prior root; the
+  window is what lets a member who missed a rotation catch up, or learn they were
+  cut. This is `community::rekey::watches`, installed as a second standing REQ
+  whose id resolves back to its community through `rekey::WatchRegistry` (the id
+  cannot carry a 64-hex community id within the NIP-01 length cap).
+- `community::rekey::adopt` then reads those wraps **from the database** and walks
+  each scope forward one epoch at a time: a complete, authorized rotation whose
+  `prevcommit` extends the key actually held hands over the next key (raced
+  rotations settle on the lowest); a gap is fetched, never waived; a fork is never
+  adopted. An addressed-but-unverifiable rotation is neither adopted nor read as a
+  removal. Only a complete rotation at/after the join, from a rotator who outranks
+  the member, with no blob for them, is a removal; one that predates the join is a
+  strand — a stale invite landed the member on a superseded epoch.
 
 ### Tests
 
@@ -627,15 +745,22 @@ client.subscribe(filter).with_id(sub_id).await?;
   app `UniversalSigner` both work. The NIP-59 paths (`build_direct_invite`,
   `unwrap_direct_invite`) stay `Sized` because the SDK's gift-wrap helpers are.
   Group-key and locally-held-secret writers (`cord01` wrap functions,
-  `cord05::build_bundle_event`, `store`) still take the raw key material they
+  `cord05::build_bundle_event`, `community::cache`) still take the raw key material they
   genuinely need.
 - **`crates/chat/src/lib.rs::handle_notifications` treats every kind 1059 event as
   a NIP-59 gift wrap for the current user.** Concord wraps are kind 1059 too, so
   that handler must route by subscription id before any concord subscription goes
   live, or every stream wrap lands in the DM trash and raises a toast.
-- **Rotation-delivered plane keys cannot be persisted yet.** `CommunityState` has
-  nowhere to keep a key a rotation delivered, so a client can verify a rotation
-  and still lose it on restart — history under a prior root or a prior channel
-  epoch is unreadable until that schema change lands. (A granted private-channel
-  key does now have a home: `ChannelKeyRef.key`, filled by
-  `CommunityState::from_join_material`.)
+- **Rotation-delivered plane keys are persisted, and history spans them.**
+  `CommunityState.held_roots` keeps every root a rotation stepped off (with the
+  retired root's Control signer and the publish time that retires it), and each
+  `ChannelKeyRef.priors` keeps every channel key a rotation stepped off. The read
+  side derives planes from all of them, and a retired key's `retired_at` is a hard
+  read cutoff at both `history::page` and `sync::fold`. The state
+  `Community::removed_at`/`Community::stranded`/`channel_removed_at` carries is
+  rendered as a notice **and** enforced at write time: `Community::send` refuses
+  when `channel_secret` is `None` (a removal, a strand, a channel cut, or a key we
+  never held), because a wrap sealed under a superseded root would reach nobody who
+  rotated. A rotation this client publishes itself (`Community::rotate`) is adopted
+  through the same `rekey::adopt` an arriving one goes through, so the held state
+  and the wire cannot diverge.