sdk: add subscription EOSE query and atomic reissue

nostr-sdk/CHANGELOG.md

@@ -35,6 +35,7 @@
 - Redesign `Client::try_connect_relay` API
 - Return `NotificationStream<T>` from client and relay `notifications()` methods instead of a boxed stream.
 - Remove `RelayStatus::Pending` and rename `RelayStatus::Terminated` to `RelayStatus::Idle` (https://github.com/nostrdevkit/nostr/pull/1491).
+- Return `SubscriptionData` from `Relay::subscriptions`, `Relay::subscription`, `Client::subscriptions` and `Client::subscription` (one per relay) instead of the filters. It also tells whether the current `REQ` was answered with `EOSE` and whether the relay closed the subscription.
 
 ### Changed
 
@@ -47,6 +48,7 @@
 - Add `LocalRelayBuilderNip42::relay_url` to make it possible to configure a custom `relay_url` for the nip42 challenge validation (https://github.com/nostrdevkit/nostr/pull/1476)
 - Add `LocalRelayBuilder::new_event_channel_size` for customizing the size of the channel used to notify new received events
 - Add opt-in receiver gap reporting with `NotificationStream::with_gaps()`.
+- Add `Relay::reissue_subscription` to queue `CLOSE` and `REQ` for a long-lived subscription as a single unit, so the relay sends its stored events again.
 
 ### Fixed
 
@@ -60,6 +62,8 @@
 - Restore subscriptions after reconnections within the same second  (https://github.com/nostrdevkit/nostr/pull/1491).
 - Reset subscription EOSE and event-limit tracking when a subscription is restored, so a relay isn't flagged for resending stored events.
 - Avoid duplicate subscription `REQ`s after reconnecting.
+- Reset subscription EOSE and event-limit tracking when a `REQ` is sent again on the same connection.
+- Avoid a duplicate subscription `REQ` when another `REQ` for it is queued while it's being restored after a reconnection.
 
 ### Deprecated
 

nostr-sdk/src/client/api/subscribe.rs

@@ -70,20 +70,61 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
     use std::time::Duration;
+    use std::{future, slice};
 
     use futures::StreamExt;
     use nostr::event::{EventBuilder, FinalizeEvent, Kind};
     use nostr::filter::Filter;
     use nostr::key::Keys;
-    use nostr::message::SubscriptionId;
+    use nostr::message::{RelayMessage, SubscriptionId};
+    use nostr::types::RelayUrl;
 
     use crate::authenticator::SignerAuthenticator;
     use crate::client::ClientNotification;
+    use crate::local_relay::MockRelay;
+    use crate::relay::SubscriptionData;
     use crate::test_utils::{
         setup_client, setup_client_with_authenticator, setup_nip42_read_local_relay,
     };
 
+    #[tokio::test]
+    async fn test_client_subscription_returns_data_registered_on_each_relay() {
+        let mock = MockRelay::run().await.unwrap();
+        let url: RelayUrl = mock.url().await;
+        let client = setup_client(url.clone()).await;
+        let filter: Filter = Filter::new().kind(Kind::TextNote);
+        let id: SubscriptionId = SubscriptionId::new("client-subscription-data");
+
+        let notifications = client.notifications();
+
+        client
+            .subscribe(filter.clone())
+            .with_id(id.clone())
+            .await
+            .unwrap();
+
+        let eose = notifications.any(|notification| {
+            future::ready(matches!(notification, ClientNotification::Message { ref message, .. }
+                if matches!(message.as_ref(), RelayMessage::EndOfStoredEvents(received) if received.as_ref() == &id)))
+        });
+        assert!(
+            tokio::time::timeout(Duration::from_secs(5), eose)
+                .await
+                .unwrap()
+        );
+
+        let subscription: HashMap<RelayUrl, SubscriptionData> = client.subscription(&id).await;
+        let data: &SubscriptionData = subscription.get(&url).unwrap();
+        assert_eq!(data.filters(), slice::from_ref(&filter));
+        assert!(data.received_eose());
+        assert!(!data.is_closed());
+
+        let subscriptions = client.subscriptions().await;
+        assert_eq!(subscriptions[&id][&url].filters(), [filter]);
+    }
+
     #[tokio::test]
     async fn test_client_subscribe_dont_resubscribes_after_auth_required_closed_without_authenticator()
      {

nostr-sdk/src/client/mod.rs

@@ -29,7 +29,8 @@ use crate::pool::{RelayPool, RelayPoolBuilder};
 #[cfg(not(target_arch = "wasm32"))]
 use crate::proxy::Proxy;
 use crate::relay::{
-    Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SyncOptions,
+    Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SubscriptionData,
+    SyncOptions,
 };
 use crate::stream::NotificationStream;
 
@@ -591,15 +592,17 @@ impl Client {
         self.pool().disconnect().await
     }
 
-    /// Get subscriptions
+    /// Get long-lived subscriptions, with the data registered on each relay
     #[inline]
-    pub async fn subscriptions(&self) -> HashMap<SubscriptionId, HashMap<RelayUrl, Vec<Filter>>> {
+    pub async fn subscriptions(
+        &self,
+    ) -> HashMap<SubscriptionId, HashMap<RelayUrl, SubscriptionData>> {
         self.pool().subscriptions().await
     }
 
-    /// Get subscription
+    /// Get subscription by [`SubscriptionId`], with the data registered on each relay
     #[inline]
-    pub async fn subscription(&self, id: &SubscriptionId) -> HashMap<RelayUrl, Vec<Filter>> {
+    pub async fn subscription(&self, id: &SubscriptionId) -> HashMap<RelayUrl, SubscriptionData> {
         self.pool().subscription(id).await
     }
 

nostr-sdk/src/pool/mod.rs

@@ -25,7 +25,7 @@ use crate::monitor::Monitor;
 use crate::policy::AdmitStatus;
 use crate::relay::{
     AtomicRelayCapabilities, Relay, RelayCapabilities, RelayOptions, RelayStreamEvent,
-    ReqExitPolicy, SubscribeAutoCloseOptions, SyncOptions,
+    ReqExitPolicy, SubscribeAutoCloseOptions, SubscriptionData, SyncOptions,
 };
 use crate::shared::SharedState;
 use crate::stream::ReceiverStream;
@@ -388,45 +388,48 @@ impl RelayPool {
     #[inline]
     pub(crate) async fn subscriptions(
         &self,
-    ) -> HashMap<SubscriptionId, HashMap<RelayUrl, Vec<Filter>>> {
+    ) -> HashMap<SubscriptionId, HashMap<RelayUrl, SubscriptionData>> {
         // Lock with read shared access
         let relays = self.relays.read().await;
 
-        let mut subscriptions: HashMap<SubscriptionId, HashMap<RelayUrl, Vec<Filter>>> =
+        let mut subscriptions: HashMap<SubscriptionId, HashMap<RelayUrl, SubscriptionData>> =
             HashMap::new();
 
         for (url, relay) in relays.iter() {
             // Get relay subscriptions
             let relay_subscriptions = relay.subscriptions().await;
 
             // Iterate relay subscriptions and populate the general subscriptions map
-            for (id, list) in relay_subscriptions.into_iter() {
+            for (id, data) in relay_subscriptions.into_iter() {
                 subscriptions
                     .entry(id)
                     .or_default()
-                    .insert(url.clone(), list);
+                    .insert(url.clone(), data);
             }
         }
 
         subscriptions
     }
 
     #[inline]
-    pub(crate) async fn subscription(&self, id: &SubscriptionId) -> HashMap<RelayUrl, Vec<Filter>> {
+    pub(crate) async fn subscription(
+        &self,
+        id: &SubscriptionId,
+    ) -> HashMap<RelayUrl, SubscriptionData> {
         // Lock with read shared access
         let relays = self.relays.read().await;
 
-        let mut filters: HashMap<RelayUrl, Vec<Filter>> = HashMap::new();
+        let mut subscription: HashMap<RelayUrl, SubscriptionData> = HashMap::new();
 
-        // Iterate relays and populate filters
+        // Iterate relays and collect the subscription registered on each
         for (url, relay) in relays.iter() {
             // try to get subscription by ID from the relay
-            if let Some(list) = relay.subscription(id).await {
-                filters.insert(url.clone(), list);
+            if let Some(data) = relay.subscription(id).await {
+                subscription.insert(url.clone(), data);
             }
         }
 
-        filters
+        subscription
     }
 
     pub(crate) async fn send_msg(

nostr-sdk/src/relay/api/mod.rs

@@ -1,5 +1,6 @@
 mod count_events;
 mod fetch_events;
+mod reissue_subscription;
 mod send_event;
 mod send_msg;
 mod stream_events;
@@ -11,6 +12,7 @@ mod unsubscribe_all;
 
 pub use self::count_events::*;
 pub use self::fetch_events::*;
+pub use self::reissue_subscription::*;
 pub use self::send_event::*;
 pub use self::send_msg::*;
 pub use self::stream_events::*;

nostr-sdk/src/relay/api/reissue_subscription.rs

@@ -0,0 +1,33 @@
+use std::future::IntoFuture;
+
+use nostr::message::SubscriptionId;
+
+use crate::error::Error;
+use crate::future::BoxedFuture;
+use crate::relay::Relay;
+
+/// Reissue a long-lived subscription
+#[must_use = "Does nothing unless you await!"]
+pub struct ReissueSubscription<'relay, 'id> {
+    relay: &'relay Relay,
+    id: &'id SubscriptionId,
+}
+
+impl<'relay, 'id> ReissueSubscription<'relay, 'id> {
+    #[inline]
+    pub(crate) fn new(relay: &'relay Relay, id: &'id SubscriptionId) -> Self {
+        Self { relay, id }
+    }
+}
+
+impl<'relay, 'id> IntoFuture for ReissueSubscription<'relay, 'id>
+where
+    'id: 'relay,
+{
+    type Output = Result<(), Error>;
+    type IntoFuture = BoxedFuture<'relay, Self::Output>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        Box::pin(async move { self.relay.inner.reissue_subscription(self.id).await })
+    }
+}

nostr-sdk/src/relay/api/subscribe.rs

@@ -313,7 +313,10 @@ mod tests {
 
         assert_eq!(err.kind(), ErrorKind::Invalid);
         assert_eq!(err.to_string(), "subscription ID already exists");
-        assert_eq!(relay.subscription(&id).await.unwrap(), vec![first_filter]);
+        assert_eq!(
+            relay.subscription(&id).await.unwrap().filters(),
+            [first_filter]
+        );
     }
 
     #[tokio::test]
@@ -348,7 +351,10 @@ mod tests {
 
         assert_eq!(err.kind(), ErrorKind::Invalid);
         assert_eq!(err.to_string(), "subscription ID already exists");
-        assert_eq!(relay.subscription(&id).await.unwrap(), vec![first_filter]);
+        assert_eq!(
+            relay.subscription(&id).await.unwrap().filters(),
+            [first_filter]
+        );
     }
 
     #[tokio::test]