add chat plane

crates/concord/src/store.rs

@@ -1,20 +1,23 @@
-use std::collections::BTreeMap;
+use std::cmp::Reverse;
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::LazyLock;
 
 use anyhow::{Result, anyhow};
 use nostr_sdk::prelude::*;
 use serde::{Deserialize, Serialize};
 
+use crate::chat::{self, ChatRumor, plane_keys};
 use crate::control::{
     ChannelMetadata, CommunityGenesis, CommunityMetadata, ControlFold, ROOT_EPOCH,
 };
 use crate::derive::control_signer_group_key;
 use crate::edition::{EntityHead, Floors, ParsedEdition, vsk};
-use crate::stream::OpenedStream;
-use crate::{ChannelId, CommunityId, Epoch};
+use crate::stream::{KIND_WRAP_EPHEMERAL, OpenedStream};
+use crate::{ChannelId, CommunityId, Epoch, GroupKey};
 
 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";
@@ -265,18 +268,179 @@ where
     }
 }
 
+pub async fn backfill(
+    client: &Client,
+    database: &dyn NostrDatabase,
+    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 {
+            cache_rumor(database, 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)) = chat::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 nostr_memory::MemoryDatabase;
 
     use super::*;
     use crate::Epoch;
+    use crate::chat::{build_message, seal_rumor};
     use crate::derive::channel_group_key;
     use crate::stream::{
         KIND_WRAP, SealForm, build_rumor_ms, build_seal, channel_binding_tags, open_wrap, wrap_seal,
     };
 
     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");
+
+        // 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);
+            relay.insert(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();
+        assert_eq!(
+            contents,
+            ["after the rekey", "still before", "before the rekey"]
+        );
+    }
 
     #[test]
     fn rumors_read_back_after_a_restart() {