feat(eventsub): properly unsubscribe once no more handles are interested (#5943)

CHANGELOG.md

@@ -22,7 +22,7 @@
 - Bugfix: Fixed the reply button showing for inline whispers and announcements. (#5863)
 - Bugfix: Fixed suspicious user treatment update messages not being searchable. (#5865)
 - Bugfix: Ensure miniaudio backend exits even if it doesn't exit cleanly. (#5896)
-- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932)
+- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943)
 - Dev: Remove unneeded platform specifier for toasts. (#5914)
 - Dev: Highlight checks now use non-capturing groups for the boundaries. (#5784)
 - Dev: Removed unused PubSub whisper code. (#5898)

mocks/include/mocks/Helix.hpp

@@ -436,6 +436,11 @@ public:
                       failureCallback)),
                 (override));
 
+    MOCK_METHOD(void, deleteEventSubSubscription,
+                (const QString &request, ResultCallback<> successCallback,
+                 (FailureCallback<QString> failureCallback)),
+                (override));
+
     MOCK_METHOD(void, update, (QString clientId, QString oauthToken),
                 (override));
 

src/providers/twitch/api/Helix.cpp

@@ -3334,6 +3334,46 @@ QDebug &operator<<(QDebug &dbg,
     return dbg;
 }
 
+void Helix::deleteEventSubSubscription(const QString &subscriptionID,
+                                       ResultCallback<> successCallback,
+                                       FailureCallback<QString> failureCallback)
+{
+    QUrlQuery query;
+    query.addQueryItem("id", subscriptionID);
+
+    this->makeDelete("eventsub/subscriptions", query)
+        .onSuccess([successCallback](const auto &result) {
+            if (result.status() != 204)
+            {
+                qCWarning(chatterinoTwitchEventSub)
+                    << "Success result for deleting eventsub subscription was "
+                    << result.formatError() << "but we expected it to be 204";
+            }
+
+            successCallback();
+        })
+        .onError([failureCallback](const NetworkResult &result) {
+            if (!result.status())
+            {
+                failureCallback(result.formatError());
+                return;
+            }
+
+            const auto obj = result.parseJson();
+            auto message = obj["message"].toString();
+
+            if (message.isEmpty())
+            {
+                failureCallback("Twitch internal server error");
+            }
+            else
+            {
+                failureCallback(message);
+            }
+        })
+        .execute();
+}
+
 NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery,
                                   NetworkRequestType type)
 {

src/providers/twitch/api/Helix.hpp

@@ -1217,6 +1217,11 @@ public:
         FailureCallback<HelixCreateEventSubSubscriptionError, QString>
             failureCallback) = 0;
 
+    // https://dev.twitch.tv/docs/api/reference/#delete-eventsub-subscription
+    virtual void deleteEventSubSubscription(
+        const QString &subscriptionID, ResultCallback<> successCallback,
+        FailureCallback<QString> failureCallback) = 0;
+
     virtual void update(QString clientId, QString oauthToken) = 0;
 
 protected:
@@ -1567,6 +1572,11 @@ public:
         FailureCallback<HelixCreateEventSubSubscriptionError, QString>
             failureCallback) final;
 
+    // https://dev.twitch.tv/docs/api/reference/#delete-eventsub-subscription
+    void deleteEventSubSubscription(
+        const QString &subscriptionID, ResultCallback<> successCallback,
+        FailureCallback<QString> failureCallback) final;
+
     void update(QString clientId, QString oauthToken) final;
 
     static void initialize();

src/providers/twitch/eventsub/Controller.cpp

@@ -60,7 +60,6 @@ Controller::Controller()
 Controller::~Controller()
 {
     qCInfo(LOG) << "Controller dtor start";
-    this->queuedSubscriptions.clear();
 
     for (const auto &weakConnection : this->connections)
     {
@@ -73,6 +72,8 @@ Controller::~Controller()
         connection->close();
     }
 
+    this->subscriptions.clear();
+
     this->work.reset();
 
     if (this->thread->joinable())
@@ -91,15 +92,35 @@ void Controller::removeRef(const SubscriptionRequest &request)
 {
     std::lock_guard lock(this->subscriptionsMutex);
 
-    assert(this->activeSubscriptions.contains(request));
+    assert(this->subscriptions.contains(request));
 
-    auto &xd = this->activeSubscriptions[request];
+    auto &subscription = this->subscriptions[request];
-    xd.refCount--;
+    subscription.refCount--;
-    qCInfo(LOG) << "Remove ref for" << request << xd.refCount;
+    qCInfo(LOG) << "Removed ref for" << request << subscription.refCount;
-    // todo use actual atomic things here to be smart
+
-    if (xd.refCount <= 0)
+    if (subscription.refCount <= 0)
     {
-        qCInfo(LOG) << "TODO: Unsubscribe from" << request;
+        if (subscription.subscriptionID.isEmpty())
+        {
+            qCWarning(LOG) << "Refcount fell to zero for" << request
+                           << "but we had no subscription ID attached - was a "
+                              "successful subscription never made?";
+            return;
+        }
+
+        qCInfo(LOG) << "Unsubscribing from" << request;
+        getHelix()->deleteEventSubSubscription(
+            subscription.subscriptionID,
+            [request] {
+                qCInfo(LOG) << "Successfully unsubscribed from" << request;
+            },
+            [request](const auto &errorMessage) {
+                qCInfo(LOG)
+                    << "An error occurred while attempting to unsubscribe from"
+                    << request << errorMessage;
+            });
+
+        subscription.subscriptionID.clear();
     }
 }
 
@@ -110,13 +131,13 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
     {
         std::lock_guard lock(this->subscriptionsMutex);
 
-        auto &xd = this->activeSubscriptions[request];
+        auto &subscription = this->subscriptions[request];
-        if (xd.refCount == 0)
+        if (subscription.refCount == 0)
         {
             needToSubscribe = true;
         }
-        xd.refCount++;
+        subscription.refCount++;
-        qCInfo(LOG) << "Add ref for" << request << xd.refCount;
+        qCInfo(LOG) << "Added ref for" << request << subscription.refCount;
     }
 
     auto handle = std::make_unique<RawSubscriptionHandle>(request);
@@ -131,24 +152,29 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
     return handle;
 }
 
-void Controller::subscribe(const SubscriptionRequest &request, bool isQueued)
+void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
 {
     qCInfo(LOG) << "Subscribe request for" << request.subscriptionType;
-    boost::asio::post(this->ioContext, [this, request, isQueued] {
+    boost::asio::post(this->ioContext, [this, request, isRetry] {
         // 1. Flush dead connections (maybe this should not be done here)
         // TODO: implement
 
-        if (isQueued)
         {
-            qCInfo(LOG) << "Removing subscription from queued list";
+            std::lock_guard lock(this->subscriptionsMutex);
-            this->queuedSubscriptions.erase(request);
+            auto &subscription = this->subscriptions[request];
-        }
+            if (isRetry)
+            {
+                qCInfo(LOG) << "Removing subscription from queued list";
 
-        if (this->queuedSubscriptions.contains(request))
+                subscription.retryTimer.reset();
-        {
+            }
-            qCWarning(LOG) << "We already have a queued subscription for this, "
+            else if (subscription.retryTimer)
-                              "let's chill :)";
+            {
-            return;
+                qCWarning(LOG)
+                    << "We already have a queued subscription for this, "
+                       "let's chill :)";
+                return;
+            }
         }
 
         uint32_t openButNotReadyConnections = 0;
@@ -188,7 +214,8 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued)
                 [this, request, connection,
                  weakConnection{weakConnection}](const auto &res) {
                     qCInfo(LOG) << "success" << res;
-                    this->markRequestSubscribed(request, weakConnection);
+                    this->markRequestSubscribed(request, weakConnection,
+                                                res.subscriptionID);
                 },
                 [this, request](const auto &error, const auto &errorString) {
                     using Error = HelixCreateEventSubSubscriptionError;
@@ -204,6 +231,10 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued)
 
                         case Error::Forbidden:
                             qCWarning(LOG) << "Forbidden" << errorString;
+                            boost::asio::post(this->ioContext, [this, request] {
+                                this->retrySubscription(
+                                    request, boost::posix_time::seconds(2), 5);
+                            });
                             break;
 
                         case Error::Conflict:
@@ -221,8 +252,8 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued)
                                 << "Unhandled error, retrying subscription"
                                 << errorString;
                             boost::asio::post(this->ioContext, [this, request] {
-                                this->queueSubscription(
+                                this->retrySubscription(
-                                    request, boost::posix_time::seconds(2));
+                                    request, boost::posix_time::seconds(2), 5);
                             });
                             break;
                     }
@@ -234,18 +265,20 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isQueued)
         {
             // No connection was available to handle this subscription request, create a new connection
             this->createConnection();
-            this->queueSubscription(request, boost::posix_time::millisec(500));
+            this->retrySubscription(request, boost::posix_time::millisec(500),
+                                    10);
         }
         else
         {
             if (openButNotReadyConnections > 1)
             {
                 qCWarning(LOG) << "We have" << openButNotReadyConnections
-                               << "open but not ready connections, hmmm";
+                               << "open but no ready connections, hmmm";
             }
 
             // At least one connection is open, but it has not gotten the welcome message yet
-            this->queueSubscription(request, boost::posix_time::millisec(250));
+            this->retrySubscription(request, boost::posix_time::millisec(250),
+                                    10);
         }
     });
 }
@@ -288,15 +321,30 @@ void Controller::registerConnection(std::weak_ptr<lib::Session> &&connection)
     this->connections.emplace_back(std::move(connection));
 }
 
-void Controller::queueSubscription(const SubscriptionRequest &request,
+void Controller::retrySubscription(const SubscriptionRequest &request,
-                                   boost::posix_time::time_duration delay)
+                                   boost::posix_time::time_duration delay,
+                                   int32_t maxAttempts)
 {
-    this->threadGuard->guard();
+    std::lock_guard lock(this->subscriptionsMutex);
 
-    auto resubTimer =
+    auto &connection = this->subscriptions[request];
+
+    if (connection.retryAttempts <= 0)
+    {
+        connection.retryAttempts = maxAttempts;
+    }
+    else if (--connection.retryAttempts == 0)
+    {
+        qCWarning(LOG) << "Reached max amount of retries for" << request;
+        return;
+    }
+
+    qCInfo(LOG) << "Retrying subscription" << request << " - attempt"
+                << connection.retryAttempts;
+    auto retryTimer =
         std::make_unique<boost::asio::deadline_timer>(this->ioContext);
-    resubTimer->expires_from_now(delay);
+    retryTimer->expires_from_now(delay);
-    resubTimer->async_wait([this, request](const auto &ec) {
+    retryTimer->async_wait([this, request](const auto &ec) {
         if (!ec)
         {
             // The timer passed naturally
@@ -304,15 +352,22 @@ void Controller::queueSubscription(const SubscriptionRequest &request,
         }
     });
 
-    this->queuedSubscriptions.emplace(request, std::move(resubTimer));
+    assert(connection.retryTimer == nullptr &&
+           "Timer should not already be set");
+
+    connection.retryTimer = std::move(retryTimer);
 }
 
 void Controller::markRequestSubscribed(const SubscriptionRequest &request,
-                                       std::weak_ptr<lib::Session> connection)
+                                       std::weak_ptr<lib::Session> connection,
+                                       const QString &subscriptionID)
 {
     std::lock_guard lock(this->subscriptionsMutex);
 
-    this->activeSubscriptions[request].connection = std::move(connection);
+    auto &subscription = this->subscriptions[request];
+
+    subscription.connection = std::move(connection);
+    subscription.subscriptionID = subscriptionID;
 }
 
 }  // namespace chatterino::eventsub

src/providers/twitch/eventsub/Controller.hpp

@@ -22,6 +22,9 @@ class IController
 public:
     virtual ~IController() = default;
 
+    /// Removes one reference for the given subscription request
+    ///
+    /// Should realistically only be called in the dtor of SubscriptionHandle
     virtual void removeRef(const SubscriptionRequest &request) = 0;
 
     /// Subscribe will make a request to each open connection and ask them to
@@ -47,16 +50,18 @@ public:
         const SubscriptionRequest &request) override;
 
 private:
-    void subscribe(const SubscriptionRequest &request, bool isQueued);
+    void subscribe(const SubscriptionRequest &request, bool isRetry);
 
     void createConnection();
     void registerConnection(std::weak_ptr<lib::Session> &&connection);
 
-    void queueSubscription(const SubscriptionRequest &request,
+    void retrySubscription(const SubscriptionRequest &request,
-                           boost::posix_time::time_duration delay);
+                           boost::posix_time::time_duration delay,
+                           int32_t maxAttempts);
 
     void markRequestSubscribed(const SubscriptionRequest &request,
-                               std::weak_ptr<lib::Session> connection);
+                               std::weak_ptr<lib::Session> connection,
+                               const QString &subscriptionID);
 
     const std::string userAgent;
 
@@ -72,17 +77,20 @@ private:
 
     std::vector<std::weak_ptr<lib::Session>> connections;
 
-    struct XD {
+    struct Subscription {
         int32_t refCount = 0;
         std::weak_ptr<lib::Session> connection;
+
+        /// The ID of the subscription the Twitch Helix API has given us
+        QString subscriptionID;
+
+        /// The timer, if any, for retrying the subscription creation
+        std::unique_ptr<boost::asio::deadline_timer> retryTimer;
+        int32_t retryAttempts = 0;
     };
 
     std::mutex subscriptionsMutex;
-    std::unordered_map<SubscriptionRequest, XD> activeSubscriptions;
+    std::unordered_map<SubscriptionRequest, Subscription> subscriptions;
-
-    std::unordered_map<SubscriptionRequest,
-                       std::unique_ptr<boost::asio::deadline_timer>>
-        queuedSubscriptions;
 };
 
 class DummyController : public IController