update backend

crates/community/src/cache.rs

@@ -0,0 +1,288 @@
+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)
+}
+
+/// 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(),
+                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/community.rs

@@ -1,4 +1,4 @@
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::path::PathBuf;
 
 use anyhow::Result;
@@ -6,15 +6,63 @@ use concord::cord02::{ControlFold, ImageRef};
 use concord::cord03::{self, ChatMessage, ReplyRef};
 use concord::cord04::roles::{Permissions, citation_ok};
 use concord::derive::channel_group_key;
-use concord::store::{self, ChannelKeyRef, CommunityState};
+use concord::state::{ChannelCursor, ChannelKeyRef, CommunityState};
 use concord::{ChannelId, CommunityId, Epoch};
 use gpui::{App, AppContext, Context, EventEmitter, Task};
 use nostr_sdk::prelude::*;
 use state::NostrRegistry;
 
+use crate::cache;
+use crate::history::{self, Window, WrapPage};
 use crate::sync::{self, Snapshot};
 
-const MESSAGE_LIMIT: usize = 200;
+/// Wraps one relay returns for one page request.
+const PAGE_WRAPS: usize = 50;
+/// Pages a catch-up round walks down a channel's history on its own.
+const CATCH_UP_PAGES: usize = 20;
+/// Pages one explicit "load older" fetches from the relays.
+pub const LOAD_OLDER_PAGES: usize = 6;
+/// Rows one timeline read returns before the caller asks for more.
+pub const TIMELINE_PAGE: usize = 100;
+/// Side events read per row, so a reaction flood cannot displace the rows it
+/// decorates.
+const SIDE_EVENT_FACTOR: usize = 4;
+
+/// Which direction a sync round reads.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Intent {
+    CatchUp,
+    Older { pages: usize },
+}
+
+/// What one round saw, for the caller to report and act on.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct Progress {
+    pub fetched: usize,
+    pub opened: usize,
+    pub exhausted: bool,
+    pub failed: bool,
+    pub errors: usize,
+}
+
+/// A channel's timeline, folded from the local cache.
+#[derive(Debug, Clone, Default)]
+pub struct Timeline {
+    /// Oldest first, ready for a bottom-aligned list.
+    pub messages: Vec<ChatMessage>,
+    /// The cache holds rows older than `messages`.
+    pub has_more: bool,
+}
+
+/// One channel's round in flight, coalescing requests that arrive while it runs.
+#[derive(Default)]
+struct Round {
+    running: bool,
+    /// The next round to run once this one lands.
+    queued: Option<Intent>,
+    /// Everyone waiting on the outcome.
+    waiters: Vec<flume::Sender<Result<Progress, String>>>,
+}
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SubscriptionKey {
@@ -64,6 +112,7 @@ pub struct Community {
     refresh_task: Option<Task<Result<()>>>,
     icon_task: Option<Task<Result<()>>>,
     banner_task: Option<Task<Result<()>>>,
+    rounds: HashMap<ChannelId, Round>,
 }
 
 impl EventEmitter<CommunityEvent> for Community {}
@@ -83,6 +132,7 @@ impl Community {
             refresh_task: None,
             icon_task: None,
             banner_task: None,
+            rounds: HashMap::new(),
         }
     }
 
@@ -162,31 +212,168 @@ impl Community {
         Some((held.epoch, self.state.community_root))
     }
 
-    /// Page a channel's history into the local cache, once per channel.
-    pub fn backfill(&self, channel: &ChannelId, cx: &App) -> Task<Result<()>> {
-        let Some((epoch, secret)) = self.channel_secret(channel) else {
-            return Task::ready(Ok(()));
-        };
+    /// Every secret the client holds for a channel, newest epoch first.
+    fn held_keys(&self, channel: &ChannelId) -> Vec<(Epoch, [u8; 32])> {
+        self.channel_secret(channel).into_iter().collect()
+    }
 
-        let client = NostrRegistry::global(cx).read(cx).client();
+    pub fn sync_channel(
+        &mut self,
+        channel: &ChannelId,
+        intent: Intent,
+        cx: &mut Context<Self>,
+    ) -> Task<Result<Progress>> {
         let channel = *channel;
+        let (sender, receiver) = flume::bounded(1);
 
-        cx.background_spawn(async move {
-            if !store::query_rumors(&client, &channel, None, 1)
-                .await?
-                .is_empty()
-            {
-                return Ok(());
+        let start = {
+            let round = self.rounds.entry(channel).or_default();
+            round.waiters.push(sender);
+
+            let start = !round.running;
+
+            if !start {
+                round.queued = Some(intent);
             }
 
-            store::backfill(&client, &channel, &[(epoch, secret)], None, MESSAGE_LIMIT).await?;
+            start
+        };
 
-            Ok(())
+        if start {
+            self.start_round(channel, intent, cx);
+        }
+
+        cx.background_spawn(async move {
+            let progress = match receiver.recv_async().await {
+                Ok(progress) => progress,
+                Err(error) => {
+                    log::warn!("community: channel round cancelled: {error}");
+                    return Ok(Progress {
+                        failed: true,
+                        ..Progress::default()
+                    });
+                }
+            };
+            progress.map_err(anyhow::Error::msg)
         })
     }
 
+    fn start_round(&mut self, channel: ChannelId, intent: Intent, cx: &mut Context<Self>) {
+        let client = NostrRegistry::global(cx).read(cx).client();
+        let held = self.held_keys(&channel);
+        let relays = self.state.relays.clone();
+
+        let saved = self
+            .state
+            .cursors
+            .get(&channel)
+            .copied()
+            .unwrap_or_default();
+
+        if let Some(round) = self.rounds.get_mut(&channel) {
+            round.running = true;
+            round.queued = None;
+        }
+
+        let round = cx.background_spawn(async move {
+            sync_round(&client, &channel, &held, &relays, saved, intent).await
+        });
+
+        cx.spawn(async move |this, cx| {
+            let result = round.await;
+
+            if let Err(error) = this.update(cx, |this, cx| this.finish_round(channel, result, cx)) {
+                log::warn!("community: a channel round outlived its community: {error}");
+            }
+        })
+        .detach();
+    }
+
+    fn finish_round(
+        &mut self,
+        channel: ChannelId,
+        result: Result<RoundOutcome>,
+        cx: &mut Context<Self>,
+    ) {
+        let outcome = match result {
+            Ok((progress, cursor)) => {
+                self.merge_cursor(channel, cursor, cx);
+                Ok(progress)
+            }
+            Err(error) => Err(error.to_string()),
+        };
+
+        let Some(round) = self.rounds.get_mut(&channel) else {
+            return;
+        };
+
+        round.running = false;
+        let queued = round.queued.take();
+        let waiters = std::mem::take(&mut round.waiters);
+
+        for waiter in waiters {
+            if let Err(error) = waiter.try_send(outcome.clone()) {
+                log::warn!("community: a channel round result was not delivered: {error}");
+            }
+        }
+
+        if let Some(intent) = queued {
+            self.start_round(channel, intent, cx);
+        }
+
+        if let Ok(progress) = &outcome
+            && progress.failed
+            && progress.errors > 0
+            && progress.fetched == 0
+        {
+            cx.emit(CommunityEvent::Error(format!(
+                "could not reach {} of the community's relays",
+                progress.errors
+            )));
+        }
+
+        cx.emit(CommunityEvent::Updated(self.state.id));
+    }
+
+    /// Fold a round's cursor findings in, monotonically, and persist them.
+    fn merge_cursor(&mut self, channel: ChannelId, cursor: ChannelCursor, cx: &mut Context<Self>) {
+        let held = self
+            .state
+            .cursors
+            .get(&channel)
+            .copied()
+            .unwrap_or_default();
+        let merged = held.merge(cursor);
+
+        if merged == held {
+            return;
+        }
+
+        self.state.cursors.insert(channel, merged);
+        self.persist(cx);
+    }
+
+    /// Write the local state document out.
+    fn persist(&self, cx: &Context<Self>) {
+        let client = NostrRegistry::global(cx).read(cx).client();
+        let state = self.state.clone();
+
+        cx.background_spawn(async move {
+            if let Err(error) = cache::save_state(&client, &state).await {
+                log::warn!("community: failed to persist the local state: {error}");
+            }
+        })
+        .detach();
+    }
+
     /// The channel's timeline, folded from the local cache, oldest first.
-    pub fn messages(&self, channel: &ChannelId, cx: &App) -> Task<Result<Vec<ChatMessage>>> {
+    pub fn timeline(
+        &self,
+        channel: &ChannelId,
+        before_ms: Option<u64>,
+        limit: usize,
+        cx: &App,
+    ) -> Task<Result<Timeline>> {
         let client = NostrRegistry::global(cx).read(cx).client();
         let channel = *channel;
         let owner = self.state.owner;
@@ -195,10 +382,31 @@ impl Community {
         let roles = self.control.roles.clone();
 
         cx.background_spawn(async move {
-            let cached = store::query_rumors(&client, &channel, None, MESSAGE_LIMIT).await?;
-            let mut rumors = Vec::with_capacity(cached.len());
+            let until = before_ms.map(|before_ms| Timestamp::from_secs(before_ms / 1000));
 
-            for rumor in &cached {
+            let rows = cache::query_rumors(
+                &client,
+                &channel,
+                until,
+                limit.saturating_add(1),
+                Some(&cord03::ROW_KINDS),
+            )
+            .await?;
+
+            let has_more = rows.len() > limit;
+
+            let sides = cache::query_rumors(
+                &client,
+                &channel,
+                until,
+                limit.saturating_mul(SIDE_EVENT_FACTOR),
+                Some(&cord03::SIDE_KINDS),
+            )
+            .await?;
+
+            let mut rumors = Vec::with_capacity(rows.len() + sides.len());
+
+            for rumor in rows.iter().take(limit).chain(sides.iter()) {
                 match cord03::parse_rumor(rumor) {
                     Ok(chat) => rumors.push(chat),
                     Err(error) => {
@@ -220,7 +428,7 @@ impl Community {
 
             messages.reverse();
 
-            Ok(messages)
+            Ok(Timeline { messages, has_more })
         })
     }
 
@@ -264,7 +472,7 @@ impl Community {
             let (wrap, _) = cord03::seal_rumor(&rumor, &group, &signer, false).await?;
 
             let (opened, _) = cord03::open(&wrap, &group, &channel, epoch)?;
-            store::cache_rumor(&client, &channel, &opened).await?;
+            cache::cache_rumor(&client, &channel, &opened).await?;
 
             sync::connect_relays(&client, &relays).await;
             sync::publish_wrap(&client, &wrap, &relays).await;
@@ -308,7 +516,9 @@ impl Community {
 
         match result {
             Ok(Some(snapshot)) => {
-                self.state = snapshot.state;
+                let mut state = snapshot.state;
+                state.cursors = std::mem::take(&mut self.state.cursors);
+                self.state = state;
                 self.control = snapshot.control;
                 self.members = snapshot.members;
                 self.load_images(cx);
@@ -388,3 +598,112 @@ impl Community {
         }));
     }
 }
+
+/// One round's progress and the cursor material it earned.
+type RoundOutcome = (Progress, ChannelCursor);
+
+/// Read a channel's history from the community's relays, in three passes.
+async fn sync_round(
+    client: &Client,
+    channel: &ChannelId,
+    held: &[(Epoch, [u8; 32])],
+    relays: &[RelayUrl],
+    saved: ChannelCursor,
+    intent: Intent,
+) -> Result<RoundOutcome> {
+    let mut progress = Progress::default();
+    let mut round = ChannelCursor::default();
+
+    let newest = match intent {
+        Intent::CatchUp => {
+            let page = history::page(
+                client,
+                channel,
+                held,
+                relays,
+                Window::newest(),
+                1,
+                PAGE_WRAPS,
+            )
+            .await?;
+            absorb(&mut progress, &page);
+            page
+        }
+        Intent::Older { .. } => WrapPage::default(),
+    };
+
+    let bridge = match (intent, newest.oldest_ms, saved.newest_ms) {
+        (Intent::CatchUp, Some(oldest), Some(saved_newest)) if oldest > saved_newest => {
+            let page = history::page(
+                client,
+                channel,
+                held,
+                relays,
+                Window::between(saved_newest, oldest),
+                CATCH_UP_PAGES,
+                PAGE_WRAPS,
+            )
+            .await?;
+            absorb(&mut progress, &page);
+            page
+        }
+        // Nothing to bridge, so the newest region is already complete.
+        _ => WrapPage {
+            exhausted: true,
+            ..WrapPage::default()
+        },
+    };
+
+    let resume = match intent {
+        Intent::CatchUp => saved.oldest_ms.or(newest.oldest_ms),
+        Intent::Older { .. } => saved.oldest_ms,
+    };
+
+    let pages = match intent {
+        Intent::CatchUp => CATCH_UP_PAGES,
+        Intent::Older { pages } => pages,
+    };
+
+    let older = match resume {
+        Some(until) => {
+            let page = history::page(
+                client,
+                channel,
+                held,
+                relays,
+                Window::older_than(until),
+                pages,
+                PAGE_WRAPS,
+            )
+            .await?;
+            absorb(&mut progress, &page);
+            page
+        }
+        None => WrapPage::default(),
+    };
+
+    if intent == Intent::CatchUp {
+        let complete = !newest.failed && bridge.exhausted;
+        let top = newest
+            .newest_ms
+            .unwrap_or(0)
+            .max(bridge.newest_ms.unwrap_or(0));
+
+        if complete && top > 0 {
+            round.newest_ms = Some(top);
+        }
+    }
+
+    round.oldest_ms = older.oldest_ms.or(newest.oldest_ms);
+    round.exhausted = older.exhausted;
+
+    Ok((progress, round))
+}
+
+fn absorb(progress: &mut Progress, page: &WrapPage) {
+    progress.fetched += page.raw;
+    progress.opened += page.opened.len();
+    progress.exhausted |= page.exhausted;
+    progress.failed |= page.failed;
+    progress.errors += page.errors;
+}

crates/community/src/history.rs

@@ -0,0 +1,516 @@
+use std::collections::BTreeSet;
+use std::future::Future;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+use anyhow::{Result, anyhow};
+use concord::cord01::KIND_WRAP_EPHEMERAL;
+use concord::cord03::{self, ChatRumor, plane_keys};
+use concord::{ChannelId, Epoch};
+use futures::future::{Either, join_all, select};
+use nostr_sdk::prelude::*;
+
+use crate::cache::cache_rumor;
+
+/// How long one relay is given to answer one page of history.
+const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// The region of history to read.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct Window {
+    pub until_ms: Option<u64>,
+    pub since_ms: Option<u64>,
+}
+
+impl Window {
+    /// The newest wraps, with no older bound.
+    pub fn newest() -> Self {
+        Self::default()
+    }
+
+    /// The wraps strictly older than `oldest_ms`.
+    pub fn older_than(oldest_ms: u64) -> Self {
+        Self {
+            until_ms: Some(oldest_ms.saturating_sub(1)),
+            since_ms: None,
+        }
+    }
+
+    /// The region between `since_ms` and `oldest_ms`, both inclusive.
+    pub fn between(since_ms: u64, oldest_ms: u64) -> Self {
+        Self {
+            until_ms: Some(oldest_ms.saturating_sub(1)),
+            since_ms: Some(since_ms),
+        }
+    }
+}
+
+/// What one paged fetch saw.
+#[derive(Debug, Clone, Default)]
+pub struct WrapPage {
+    pub opened: Vec<ChatRumor>,
+    pub raw: usize,
+    pub newest_ms: Option<u64>,
+    pub oldest_ms: Option<u64>,
+    pub exhausted: bool,
+    pub failed: bool,
+    pub errors: usize,
+}
+
+/// Walks a channel's history back over the community's own relays.
+pub async fn page(
+    client: &Client,
+    channel: &ChannelId,
+    held: &[(Epoch, [u8; 32])],
+    relays: &[RelayUrl],
+    window: Window,
+    max_pages: usize,
+    limit: usize,
+) -> Result<WrapPage> {
+    let planes = plane_keys(held, channel)?;
+    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()
+        });
+    }
+
+    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 = join_all(
+            asked
+                .iter()
+                .map(|(_, url)| ingest_page(client, url, &filter)),
+        )
+        .await;
+
+        for ((index, url), answer) in asked.iter().zip(answers) {
+            if let Err(error) = answer {
+                log::warn!("community: relay {url} did not answer a history page: {error}");
+                walk.reject(*index);
+            }
+        }
+
+        let answered = client.database().query(filter).await?;
+
+        for wrap in walk.accept(answered, limit) {
+            let Some((epoch, group)) = planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
+            else {
+                continue;
+            };
+
+            let Ok((stream, rumor)) = cord03::open(&wrap, group, channel, *epoch) else {
+                continue;
+            };
+
+            if cache_rumor(client, channel, &stream).await? {
+                opened.push(rumor);
+            }
+        }
+    }
+
+    Ok(walk.finish(opened))
+}
+
+/// 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_ms) = window.until_ms {
+        filter = filter.until(Timestamp::from_secs(until_ms / 1000));
+    }
+
+    if let Some(since_ms) = window.since_ms {
+        filter = filter.since(Timestamp::from_secs(since_ms / 1000));
+    }
+
+    filter
+}
+
+/// Ask one relay for one page and wait until it has answered.
+async fn ingest_page(client: &Client, url: &RelayUrl, filter: &Filter) -> Result<()> {
+    let relay = match client.relay(url).await? {
+        Some(relay) => relay,
+        None => {
+            client.add_relay(url).and_connect().await?;
+            client
+                .relay(url)
+                .await?
+                .ok_or_else(|| anyhow!("the relay was not added to the pool"))?
+        }
+    };
+
+    let id = history_subscription();
+    let mut notifications = relay.notifications();
+
+    // The SDK closes and unregisters the subscription itself once the relay says
+    // it is done, so nothing has to be unwound here.
+    relay
+        .subscribe(vec![filter.clone()])
+        .with_id(id.clone())
+        .close_on(
+            SubscribeAutoCloseOptions::default()
+                .exit_policy(ReqExitPolicy::ExitOnEOSE)
+                .timeout(Some(PAGE_TIMEOUT)),
+        )
+        .await?;
+
+    // One deadline for the whole page, so a relay that keeps talking cannot
+    // extend the wait past it.
+    let deadline = Instant::now() + PAGE_TIMEOUT;
+
+    loop {
+        let remaining = deadline.saturating_duration_since(Instant::now());
+
+        let Some(notification) = within(remaining, notifications.next()).await else {
+            return Err(anyhow!("the relay did not finish the page"));
+        };
+
+        let Some(notification) = notification else {
+            return Err(anyhow!("the relay's notification stream ended"));
+        };
+
+        let RelayNotification::Message { message } = notification else {
+            continue;
+        };
+
+        match *message {
+            RelayMessage::EndOfStoredEvents(ended) => {
+                if ended.as_ref() == &id {
+                    return Ok(());
+                }
+            }
+            RelayMessage::Closed {
+                subscription_id,
+                message,
+            } if subscription_id.as_ref() == &id && !auth_required(&message) => {
+                return Err(anyhow!("the relay closed the page: {message}"));
+            }
+            _ => {}
+        }
+    }
+}
+
+/// Whether a CLOSED reason is the NIP-42 `auth-required` one, the only reason the
+/// SDK recovers from on its own.
+fn auth_required(reason: &str) -> bool {
+    matches!(
+        MachineReadablePrefix::parse(reason),
+        Some(MachineReadablePrefix::AuthRequired)
+    )
+}
+
+/// A subscription id for one page of one relay, unique so a page's answer is
+/// never confused with the community's standing subscription or another page's.
+fn history_subscription() -> SubscriptionId {
+    static NEXT: AtomicU64 = AtomicU64::new(0);
+
+    SubscriptionId::new(format!(
+        "concord-history-{}",
+        NEXT.fetch_add(1, Ordering::Relaxed)
+    ))
+}
+
+/// Await `future`, giving up after `limit`.
+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_ms: Option<u64>,
+    /// The inclusive upper bound of the next page.
+    cursor: Option<u64>,
+    seen: BTreeSet<EventId>,
+    newest_ms: Option<u64>,
+    oldest_ms: Option<u64>,
+    raw: usize,
+    errors: 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_ms: window.since_ms,
+            cursor: window.until_ms,
+            seen: BTreeSet::new(),
+            newest_ms: None,
+            oldest_ms: None,
+            raw: 0,
+            errors: 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_ms: self.cursor,
+            since_ms: self.since_ms,
+        }
+    }
+
+    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<u64> = None;
+        let mut events = Vec::with_capacity(page.len());
+
+        for event in page {
+            let at_ms = event.created_at.as_secs().saturating_mul(1000);
+            self.newest_ms = Some(self.newest_ms.map_or(at_ms, |newest| newest.max(at_ms)));
+            oldest = Some(oldest.map_or(at_ms, |oldest| oldest.min(at_ms)));
+
+            if self.seen.insert(event.id) {
+                self.raw += 1;
+                events.push(event);
+            }
+        }
+
+        match oldest {
+            Some(oldest) if oldest > 0 => self.cursor = Some(oldest - 1),
+            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,
+            newest_ms: self.newest_ms,
+            oldest_ms: self.oldest_ms,
+            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::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| {
+                let at_ms = event.created_at.as_secs().saturating_mul(1000);
+                window.until_ms.is_none_or(|until| at_ms <= until)
+                    && window.since_ms.is_none_or(|since| at_ms >= 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 = [(Epoch(0), SECRET), (Epoch(1), NEXT_SECRET)];
+        let planes = plane_keys(&held, &channel).expect("derives");
+
+        // 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((epoch, group)) =
+                    planes.iter().find(|(_, group)| group.pk() == wrap.pubkey)
+                else {
+                    continue;
+                };
+
+                let (_, rumor) = cord03::open(&wrap, group, &channel, *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);
+    }
+
+    #[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_ms, 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(1_700_000_000_000), event_at(1_700_000_001_000)]);
+        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.as_secs() * 1000)
+            .min()
+            .expect("one wrap");
+        assert_eq!(oldest, 1_700_000_000_000);
+    }
+
+    fn event_at(at_ms: u64) -> Event {
+        let keys = Keys::generate();
+        EventBuilder::new(Kind::TextNote, "page")
+            .custom_created_at(Timestamp::from_secs(at_ms / 1000))
+            .finalize(&keys)
+            .expect("signs")
+    }
+}

crates/community/src/lib.rs

@@ -4,14 +4,16 @@ use anyhow::Result;
 use concord::cord01::KIND_WRAP;
 pub use concord::cord02::CommunityMetadata;
 pub use concord::cord03::{ChatMessage, ReplyRef};
-use concord::store::CommunityState;
+use concord::state::CommunityState;
 pub use concord::{ChannelId, CommunityId};
 use gpui::{App, AppContext, Context, Entity, EventEmitter, Global, Subscription, Task, Window};
 use nostr_sdk::prelude::*;
 use smallvec::{SmallVec, smallvec};
 use state::NostrRegistry;
 
+pub mod cache;
 mod community;
+pub mod history;
 mod sync;
 
 pub use community::*;

crates/community/src/sync.rs

@@ -10,12 +10,14 @@ use concord::cord04::roles::{Permissions, citation_ok};
 use concord::derive::{
     channel_group_key, control_group_key, control_signer_group_key, guestbook_group_key,
 };
-use concord::store::{self, CommunityState};
+use concord::state::{CommunityState, list_entry};
 use concord::{ChannelId, CommunityId, Epoch, GroupKey};
 use gpui::AsyncApp;
 use nostr_sdk::prelude::*;
 use state::UniversalSigner;
 
+use crate::cache;
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum PlaneKind {
     Control(Epoch),
@@ -126,7 +128,7 @@ where
     }
 
     let state = CommunityState::from_genesis(&genesis, &editions, at_secs.saturating_mul(1000))?;
-    store::save_state(client, &state).await?;
+    cache::save_state(client, &state).await?;
 
     publish_wraps(client, &genesis.wraps, &state.relays).await;
 
@@ -199,7 +201,7 @@ where
         return Ok(());
     }
 
-    let entry = store::list_entry(state, name);
+    let entry = list_entry(state, name);
     let list = match held {
         Some(held) => held.joined(entry),
         None => CommunityList::default().joined(entry),
@@ -282,10 +284,10 @@ pub async fn load(
 ) -> Result<Vec<CommunityState>> {
     let list = match load_list(client, signer, self_pk).await? {
         Some(list) => list,
-        None => return store::load_states(client).await,
+        None => return cache::load_states(client).await,
     };
 
-    let mut held: BTreeMap<CommunityId, CommunityState> = store::load_states(client)
+    let mut held: BTreeMap<CommunityId, CommunityState> = cache::load_states(client)
         .await?
         .into_iter()
         .map(|state| (state.id, state))
@@ -314,7 +316,7 @@ pub async fn load(
             None => fresh,
         };
 
-        store::save_state(client, &state).await?;
+        cache::save_state(client, &state).await?;
         held.insert(entry.community_id, state);
     }
 
@@ -458,7 +460,7 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
                 if let Ok((opened, rumor)) =
                     concord::cord03::open(wrap, &plane.group, &channel, epoch)
                 {
-                    store::cache_rumor(client, &channel, &opened).await?;
+                    cache::cache_rumor(client, &channel, &opened).await?;
                     observe(&mut observed, rumor.author, rumor.at_ms);
                 }
             }
@@ -507,7 +509,7 @@ pub async fn fold(client: &Client, state: &CommunityState) -> Result<Option<Snap
 
     let mut state = state.clone();
     state.apply_fold(&control);
-    store::save_state(client, &state).await?;
+    cache::save_state(client, &state).await?;
 
     Ok(Some(Snapshot {
         state,
@@ -551,14 +553,14 @@ mod tests {
             control_root: None,
             control_pks: BTreeMap::from([(0, control_pk)]),
             channels: vec![
-                concord::store::ChannelKeyRef {
+                concord::state::ChannelKeyRef {
                     id: general,
                     name: "general".to_owned(),
                     private: false,
                     epoch: Epoch(0),
                     key: None,
                 },
-                concord::store::ChannelKeyRef {
+                concord::state::ChannelKeyRef {
                     id: ChannelId::from_bytes([0x9d; 32]),
                     name: "staff".to_owned(),
                     private: true,
@@ -569,6 +571,7 @@ mod tests {
             relays: vec![RelayUrl::parse("wss://relay.example").expect("a url")],
             heads: Vec::new(),
             banned: BTreeSet::new(),
+            cursors: BTreeMap::new(),
             dissolved: false,
             added_at_ms: 0,
         };
@@ -611,7 +614,7 @@ mod tests {
             root_epoch: Epoch(0),
             control_root: Some([0x03; 32]),
             control_pks: BTreeMap::from([(0, control_pk)]),
-            channels: vec![concord::store::ChannelKeyRef {
+            channels: vec![concord::state::ChannelKeyRef {
                 id: ChannelId::from_bytes([0x9c; 32]),
                 name: "general".to_owned(),
                 private: false,
@@ -621,6 +624,7 @@ mod tests {
             relays: Vec::new(),
             heads: Vec::new(),
             banned: BTreeSet::new(),
+            cursors: BTreeMap::new(),
             dissolved: false,
             added_at_ms: 1_700_000_000_000,
         }
@@ -806,11 +810,11 @@ mod tests {
                 CommunityId::from_bytes([0x42; 32]),
                 Keys::generate().public_key(),
             );
-            let list = CommunityList::default().joined(store::list_entry(&listed, "coop"));
+            let list = CommunityList::default().joined(list_entry(&listed, "coop"));
             store_fragment(&client, &signer, &list).await;
 
             assert!(
-                store::load_state(&client, &listed.id)
+                cache::load_state(&client, &listed.id)
                     .await
                     .expect("reads")
                     .is_none()
@@ -833,7 +837,7 @@ mod tests {
 
             // Discovery writes the document, so the next load is warm.
             assert_eq!(
-                store::load_state(&client, &listed.id)
+                cache::load_state(&client, &listed.id)
                     .await
                     .expect("reads")
                     .map(|state| state.id),
@@ -860,9 +864,9 @@ mod tests {
                 Keys::generate().public_key(),
             );
 
-            let list = CommunityList::default().joined(store::list_entry(&listed, "listed"));
+            let list = CommunityList::default().joined(list_entry(&listed, "listed"));
             store_fragment(&client, &signer, &list).await;
-            store::save_state(&client, &local).await.expect("saves");
+            cache::save_state(&client, &local).await.expect("saves");
 
             let loaded = load(&client, &signer, keys.public_key())
                 .await
@@ -886,7 +890,7 @@ mod tests {
                 CommunityId::from_bytes([0x42; 32]),
                 Keys::generate().public_key(),
             );
-            store::save_state(&client, &local).await.expect("saves");
+            cache::save_state(&client, &local).await.expect("saves");
 
             let list = CommunityList::default().tombstoned(local.id, u64::MAX);
             store_fragment(&client, &signer, &list).await;