sdk: try to fetch only newer events when updating gossip lists

Closes https://gitworkshop.dev/yukikishimoto.com/nostr/issues/note15mhvsqdntu8w3njkh7urf92jlvak5rxdq7z9snfjtf2678kd9qvsgut0hz

Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

crates/nostr-sdk/CHANGELOG.md

@@ -33,6 +33,7 @@
 ### Changed
 
 - Fetch gossip NIP-17 list only if really needed (https://github.com/rust-nostr/nostr/pull/1090)
+- Try to fetch only newer events when updating gossip lists (https://github.com/rust-nostr/nostr/pull/1090)
 
 ### Added
 

crates/nostr-sdk/src/client/mod.rs

@@ -10,11 +10,11 @@ use std::iter;
 use std::sync::Arc;
 use std::time::Duration;
 
-use async_utility::futures_util::stream::BoxStream;
+use async_utility::futures_util::stream::{BoxStream, FuturesUnordered};
 use nostr::prelude::*;
 use nostr_database::prelude::*;
 use nostr_relay_pool::prelude::*;
-use tokio::sync::broadcast;
+use tokio::sync::{broadcast, Semaphore};
 
 pub mod builder;
 mod error;
@@ -1342,13 +1342,13 @@ impl Client {
         // Get kind
         let kind: Kind = kind.to_event_kind();
 
-        // Compose filters
-        let filter: Filter = Filter::default()
+        // Compose database filter
+        let db_filter: Filter = Filter::default()
             .authors(outdated_public_keys.clone())
             .kind(kind);
 
-        // Query from database
-        let stored_events: Events = self.database().query(filter.clone()).await?;
+        // Get events from database
+        let stored_events: Events = self.database().query(db_filter).await?;
 
         // Get DISCOVERY and READ relays
         let urls: Vec<RelayUrl> = self
@@ -1359,16 +1359,69 @@ impl Client {
             )
             .await;
 
-        // Get events from discovery and read relays
-        let events: Events = self
-            .fetch_events_from(urls, filter, Duration::from_secs(10))
+        let semaphore = Arc::new(Semaphore::new(10)); // Allow at max 10 concurrent requests
+        let mut futures = FuturesUnordered::new();
+
+        // Try to fetch from relays only the newer events (last created_at + 1)
+        for event in stored_events.iter() {
+            let author = event.pubkey;
+            let created_at = event.created_at;
+            let urls = urls.clone();
+            let semaphore = semaphore.clone();
+
+            futures.push(async move {
+                // Acquire permit
+                let _permit = semaphore.acquire().await;
+
+                // Construct filter
+                let filter: Filter = Filter::new()
+                    .author(author)
+                    .kind(kind)
+                    .since(created_at + Duration::from_secs(1))
+                    .limit(1);
+
+                // Fetch the event
+                let events: Events = self
+                    .fetch_events_from(urls, filter, Duration::from_secs(10))
+                    .await?;
+
+                Ok::<_, Error>(events)
+            });
+        }
+
+        // Keep track of the missing public keys
+        let mut missing_public_keys: HashSet<PublicKey> = outdated_public_keys;
+
+        // Keep track of the updated events
+        let mut updated_events = Events::default();
+
+        while let Some(result) = futures.next().await {
+            if let Ok(events) = result {
+                if let Some(event) = events.first() {
+                    // Remove from missing set
+                    missing_public_keys.remove(&event.pubkey);
+
+                    // Update the last check for this public key
+                    self.gossip.update_last_check([event.pubkey]).await;
+                }
+
+                updated_events = updated_events.merge(events);
+            }
+        }
+
+        // Get the missing events
+        let missing_filter: Filter = Filter::default()
+            .authors(missing_public_keys.clone())
+            .kind(kind);
+        let missing_events: Events = self
+            .fetch_events_from(urls, missing_filter, Duration::from_secs(10))
             .await?;
 
-        // Update last check for these public keys
-        self.gossip.update_last_check(outdated_public_keys).await;
+        // Update the last check for the missing public keys
+        self.gossip.update_last_check(missing_public_keys).await;
 
-        // Merge database and relays events
-        let merged: Events = events.merge(stored_events);
+        // Merge all the events
+        let merged: Events = stored_events.merge(updated_events).merge(missing_events);
 
         // Update gossip graph
         self.gossip.update(merged).await;