add pins, disappearing messages and hardenin

crates/concord/src/store.rs

@@ -25,11 +25,18 @@ const WRAP_TAG: &str = "e";
 const KIND_TAG: &str = "k";
 const STATE_PREFIX: &str = "concord/";
 
+/// CORD-08 §3: an already-expired rumor is refused at ingest, never stored.
+/// Returns whether the rumor was kept.
 pub async fn cache_rumor(
     database: &dyn NostrDatabase,
     channel: &ChannelId,
     opened: &OpenedStream,
-) -> Result<()> {
+) -> Result<bool> {
+    if chat::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()]),
@@ -47,7 +54,42 @@ pub async fn cache_rumor(
 
     database.save_event(&event).await?;
 
-    Ok(())
+    Ok(true)
+}
+
+pub async fn purge_expired(
+    database: &dyn NostrDatabase,
+    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 database.query(filter).await? {
+        let Ok(rumor) = UnsignedEvent::from_json(&event.content) else {
+            continue;
+        };
+
+        let Ok(Some(expiration)) = chat::expiration_of(&rumor) else {
+            continue;
+        };
+
+        if expiration <= now {
+            expired.push(event.id);
+        }
+    }
+
+    let purged = expired.len();
+
+    if purged > 0 {
+        database.delete(Filter::new().ids(expired)).await?;
+    }
+
+    Ok(purged)
 }
 
 pub async fn query_rumors(
@@ -300,8 +342,9 @@ pub async fn backfill(
         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);
+            if cache_rumor(database, channel, &opened).await? {
+                found.push(rumor);
+            }
         }
 
         match next {
@@ -420,7 +463,15 @@ mod tests {
             ("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);
+            let rumor = build_message(
+                author.public_key(),
+                &channel,
+                epoch,
+                content,
+                None,
+                at_ms,
+                None,
+            );
             relay.insert(seal_rumor(&rumor, &group, &author, false).expect("seals").0);
         }
 
@@ -505,4 +556,93 @@ mod tests {
             assert_eq!(capped[0].content, "second");
         });
     }
+
+    #[test]
+    fn an_expired_rumor_is_refused_at_ingest_and_purged_by_the_sweep() {
+        let database = MemoryDatabase::unbounded();
+        let channel = ChannelId::from_bytes([0x77u8; 32]);
+        let author = Keys::generate();
+        let group = channel_group_key(&SECRET, &channel, Epoch(0)).expect("derives");
+        let now = Timestamp::now().as_secs();
+
+        smol::block_on(async {
+            // A live timer is stored; one that already elapsed is refused at ingest.
+            assert!(
+                cache(
+                    &database,
+                    &group,
+                    &channel,
+                    &author,
+                    "live",
+                    Some(3_600),
+                    now
+                )
+                .await
+            );
+            assert!(
+                !cache(
+                    &database,
+                    &group,
+                    &channel,
+                    &author,
+                    "gone",
+                    Some(1),
+                    now - 120
+                )
+                .await
+            );
+
+            let stored = query_rumors(&database, &channel, None, 10)
+                .await
+                .expect("queries");
+            assert_eq!(stored.len(), 1);
+            assert_eq!(stored[0].content, "live");
+
+            // Hiding is not disappearing: the sweep removes the row itself,
+            // judged on the rumor's own signed tag.
+            let purged = purge_expired(&database, &channel, Timestamp::from_secs(now + 7_200))
+                .await
+                .expect("sweeps");
+            assert_eq!(purged, 1);
+            assert!(
+                query_rumors(&database, &channel, None, 10)
+                    .await
+                    .expect("queries")
+                    .is_empty()
+            );
+
+            // An untimed rumor is never swept, whatever the clock says.
+            assert!(cache(&database, &group, &channel, &author, "timeless", None, now).await);
+            let purged = purge_expired(&database, &channel, Timestamp::from_secs(now + 86_400))
+                .await
+                .expect("sweeps");
+            assert_eq!(purged, 0);
+        });
+    }
+
+    async fn cache(
+        database: &MemoryDatabase,
+        group: &GroupKey,
+        channel: &ChannelId,
+        author: &Keys,
+        content: &str,
+        timer: Option<u64>,
+        at_secs: u64,
+    ) -> bool {
+        let rumor = build_message(
+            author.public_key(),
+            channel,
+            Epoch(0),
+            content,
+            None,
+            at_secs * 1_000,
+            timer,
+        );
+        let (wrap, _) = seal_rumor(&rumor, group, author, false).expect("seals");
+        let opened = open_wrap(&wrap, group).expect("opens");
+
+        cache_rumor(database, channel, &opened)
+            .await
+            .expect("caches")
+    }
 }