Restore subscriptions after relay reconnects

src/NostrPool.cpp

@@ -192,6 +192,10 @@ NostrString NostrPool::subscribeMany(std::initializer_list<NostrString> urls, Js
         this->subscriptions[subId].closeCallback = closeCallback;
         this->subscriptions[subId].eoseCallback = eoseCallback;
         this->subscriptions[subId].eventCallback = eventCallback;
+        this->subscriptions[subId].request = json;
+        for (auto url : urls) {
+            this->subscriptions[subId].relayUrls.push_back(url);
+        }
     }
 
     for (auto url : urls) {
@@ -222,6 +226,15 @@ void NostrPool::closeSubscription(NostrString subId) {
     this->subscriptions.erase(subId);
 }
 
+NostrSubscription *NostrPool::getSubscription(NostrString subId) {
+    auto it = this->subscriptions.find(subId);
+    return it == this->subscriptions.end() ? nullptr : &it->second;
+}
+
+bool NostrPool::hasSubscription(NostrString subId) {
+    return this->subscriptions.find(subId) != this->subscriptions.end();
+}
+
 NostrRelay *NostrPool::ensureRelay(NostrString url) {
     NostrRelay *relay = NULL;
     for (auto &r : this->relays) {
@@ -236,6 +249,19 @@ NostrRelay *NostrPool::ensureRelay(NostrString url) {
         relay = new NostrRelay(conn, url);
         this->relays.push_back(relay);
         relay->conn->addMessageListener([this, relay](NostrString message) { this->onEvent(relay, message); });
+        relay->conn->addConnectionStatusListener([this, relay](ConnectionStatus status) {
+            if (status != ConnectionStatus::CONNECTED) return;
+            for (auto &entry : this->subscriptions) {
+                NostrSubscription &subscription = entry.second;
+                for (const auto &subscriptionUrl : subscription.relayUrls) {
+                    if (NostrString_equals(subscriptionUrl, relay->url)) {
+                        subscription.eose = false;
+                        relay->send(subscription.request);
+                        break;
+                    }
+                }
+            }
+        });
     }
     return relay;
 }

src/NostrPool.h

@@ -32,6 +32,8 @@ namespace nostr {
         NostrEOSECallback eoseCallback = nullptr;
         NostrEventCallback eventCallback = nullptr;
         NostrEventStatusCallback statusCallback = nullptr;
+        NostrString request;
+        std::vector<NostrString> relayUrls;
     };
 
     typedef struct s_EventStatusCallbackEntry {
@@ -133,6 +135,16 @@ namespace nostr {
          */
         void closeSubscription(NostrString subId);
 
+        /**
+         * Get a subscription by ID, or nullptr if it does not exist
+         */
+        NostrSubscription *getSubscription(NostrString subId);
+
+        /**
+         * Check whether a subscription with the given ID exists
+         */
+        bool hasSubscription(NostrString subId);
+
         /**
          * Ensure the connection to the specified relay is open
          * @param url The relay URL

tests/unit/test_pool_transport.cpp

@@ -16,6 +16,7 @@ class MockConnection : public nostr::Connection {
     std::vector<String> sent;
     std::vector<String> incoming;
     std::vector<std::function<void(String)>> messageListeners;
+    std::vector<std::function<void(nostr::ConnectionStatus)>> statusListeners;
 
     void addMessageListener(std::function<void(String)> listener) override { messageListeners.push_back(listener); }
     void send(String message) override { sent.push_back(message); }
@@ -28,8 +29,13 @@ class MockConnection : public nostr::Connection {
         }
     }
     bool isReady() override { return ready; }
-    void addConnectionStatusListener(std::function<void(nostr::ConnectionStatus)>) override {}
+    void addConnectionStatusListener(std::function<void(nostr::ConnectionStatus)> listener) override {
+        statusListeners.push_back(listener);
+    }
     void queue(String message) { incoming.push_back(message); }
+    void status(nostr::ConnectionStatus value) {
+        for (const auto &listener : statusListeners) listener(value);
+    }
 };
 
 class MockTransport : public nostr::Transport {
@@ -82,10 +88,16 @@ bool testPoolQueueAndTransport() {
         nullptr,
         [&](const String &id) { eoseCalled = id == subId; });
     CHECK(subId.length() == 64);
+    CHECK(pool.hasSubscription(subId));
+    CHECK(pool.getSubscription(subId) != nullptr);
     CHECK(transport.lastConnection != nullptr);
     MockConnection *connection = transport.lastConnection;
     pool.loop();
     CHECK(!connection->sent.empty() && connection->sent.back().indexOf("\"REQ\"") >= 0);
+    const size_t sentBeforeReconnect = connection->sent.size();
+    connection->status(nostr::ConnectionStatus::DISCONNECTED);
+    connection->status(nostr::ConnectionStatus::CONNECTED);
+    CHECK(connection->sent.size() == sentBeforeReconnect + 1);
 
     nostr::UnsignedNostrEvent unsignedEvent(1, "pool event", 1700000000);
     nostr::SignedNostrEvent event = unsignedEvent.sign(
@@ -120,6 +132,8 @@ bool testPoolQueueAndTransport() {
     pool.closeSubscription(subId);
     pool.loop();
     CHECK(connection->sent.back().indexOf("\"CLOSE\"") >= 0);
+    CHECK(!pool.hasSubscription(subId));
+    CHECK(pool.getSubscription(subId) == nullptr);
     CHECK(pool.getRelays().size() == 1 && pool.getConnectedRelays()->size() == 1);
     pool.disconnectRelay("ws://relay.example");
     CHECK(pool.getRelays().empty() && transport.connections.empty());