fix: don't use same websocket connection for multiple accounts (#6348)

CHANGELOG.md

@@ -52,6 +52,7 @@
 - Bugfix: Fixed some minor typos. (#6196)
 - Bugfix: Fixed inconsistent spaces in messages when using fractional scaling. (#6231, #6254)
 - Bugfix: Fixed eventsub message delete notifications not being affected by "Show deletions of single messages". (#6233)
+- Bugfix: Fixed an issue where eventsub subscriptions might fail after changing user. (#6348)
 - Bugfix: Don't add reply buttons to messages that are invalid reply targets. (#6119)
 - Bugfix: Fixed invalid commands from being forwarded to Helix, making it possible for information to leak (e.g. if you typed `/bann username ban reason` it would be seen by others in chat as `username ban reason`). (#6272, #6330)
 - Bugfix: Emotes that failed to load their images now show as text. (#6355)

src/controllers/commands/CommandController.cpp

@@ -471,6 +471,8 @@ CommandController::CommandController(const Paths &paths)
     this->registerCommand("/debug-invalidate-buffers",
                           &commands::invalidateBuffers);
 
+    this->registerCommand("/debug-eventsub", &commands::eventsub);
+
     this->registerCommand("/debug-test", &commands::debugTest);
 
     this->registerCommand("/shield", &commands::shieldModeOn);

src/controllers/commands/builtin/chatterino/Debugging.cpp

@@ -10,6 +10,7 @@
 #include "messages/Message.hpp"
 #include "messages/MessageBuilder.hpp"
 #include "messages/MessageElement.hpp"
+#include "providers/twitch/eventsub/Controller.hpp"
 #include "providers/twitch/TwitchChannel.hpp"
 #include "providers/twitch/TwitchIrcServer.hpp"
 #include "singletons/Theme.hpp"
@@ -158,6 +159,12 @@ QString invalidateBuffers(const CommandContext & /*ctx*/)
     return {};
 }
 
+QString eventsub(const CommandContext & /*ctx*/)
+{
+    getApp()->getEventSub()->debug();
+    return {};
+}
+
 QString debugTest(const CommandContext &ctx)
 {
     if (!ctx.channel)

src/controllers/commands/builtin/chatterino/Debugging.hpp

@@ -28,6 +28,8 @@ QString incrementImageGeneration(const CommandContext &ctx);
 
 QString invalidateBuffers(const CommandContext &ctx);
 
+QString eventsub(const CommandContext &ctx);
+
 QString debugTest(const CommandContext &ctx);
 
 }  // namespace chatterino::commands

src/providers/twitch/TwitchAccountManager.cpp

@@ -342,6 +342,9 @@ void TwitchAccountManager::load()
 
     this->currentUsername.connect([this](const QString &newUsername) {
         auto user = this->findUserByUsername(newUsername);
+
+        this->currentUserAboutToChange.invoke(this->currentUser_, user);
+
         if (user)
         {
             qCDebug(chatterinoTwitch)

src/providers/twitch/TwitchAccountManager.hpp

@@ -55,7 +55,15 @@ public:
 
     pajlada::Settings::Setting<QString> currentUsername{"/accounts/current",
                                                         ""};
-    // pajlada::Signals::NoArgSignal currentUserChanged;
+
+    /// This signal fires after we've figured out what the new account is, but before
+    /// any updates to Helix have been made.
+    ///
+    /// Useful for scenarios where you have to call Helix using the previous account.
+    pajlada::Signals::Signal<std::shared_ptr<TwitchAccount>,
+                             std::shared_ptr<TwitchAccount>>
+        currentUserAboutToChange;
+
     boost::signals2::signal<void()> currentUserChanged;
     pajlada::Signals::NoArgSignal userListUpdated;
 

src/providers/twitch/TwitchChannel.cpp

@@ -125,6 +125,18 @@ TwitchChannel::TwitchChannel(const QString &name)
 {
     qCDebug(chatterinoTwitch) << "[TwitchChannel" << name << "] Opened";
 
+    this->signalHolder_.managedConnect(
+        getApp()->getAccounts()->twitch.currentUserAboutToChange,
+        [this](const auto & /*oldAccount*/, const auto & /*newAccount*/) {
+            this->eventSubChannelChatUserMessageHoldHandle.reset();
+            this->eventSubChannelChatUserMessageUpdateHandle.reset();
+            this->eventSubChannelModerateHandle.reset();
+            this->eventSubAutomodMessageHoldHandle.reset();
+            this->eventSubAutomodMessageUpdateHandle.reset();
+            this->eventSubSuspiciousUserMessageHandle.reset();
+            this->eventSubSuspiciousUserUpdateHandle.reset();
+        });
+
     this->bSignals_.emplace_back(
         getApp()->getAccounts()->twitch.currentUserChanged.connect([this] {
             this->setMod(false);
@@ -1547,12 +1559,15 @@ void TwitchChannel::refreshPubSub()
 
     auto currentAccount = getApp()->getAccounts()->twitch.getCurrent();
 
+    const auto &currentTwitchUserID = currentAccount->getUserId();
+
     if (this->hasModRights())
     {
         this->eventSubChannelModerateHandle =
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "channel.moderate",
                 .subscriptionVersion = "2",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1561,7 +1576,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "moderator_user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1569,6 +1584,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "automod.message.hold",
                 .subscriptionVersion = "2",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1577,7 +1593,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "moderator_user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1585,6 +1601,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "automod.message.update",
                 .subscriptionVersion = "2",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1593,7 +1610,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "moderator_user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1601,6 +1618,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "channel.suspicious_user.message",
                 .subscriptionVersion = "1",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1609,7 +1627,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "moderator_user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1617,6 +1635,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "channel.suspicious_user.update",
                 .subscriptionVersion = "1",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1625,7 +1644,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "moderator_user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1645,6 +1664,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "channel.chat.user_message_hold",
                 .subscriptionVersion = "1",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1653,7 +1673,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });
@@ -1662,6 +1682,7 @@ void TwitchChannel::refreshPubSub()
             getApp()->getEventSub()->subscribe(eventsub::SubscriptionRequest{
                 .subscriptionType = "channel.chat.user_message_update",
                 .subscriptionVersion = "1",
+                .ownerTwitchUserID = currentTwitchUserID,
                 .conditions =
                     {
                         {
@@ -1670,7 +1691,7 @@ void TwitchChannel::refreshPubSub()
                         },
                         {
                             "user_id",
-                            currentAccount->getUserId(),
+                            currentTwitchUserID,
                         },
                     },
             });

src/providers/twitch/eventsub/Connection.cpp

@@ -9,6 +9,7 @@
 #include "providers/twitch/eventsub/Controller.hpp"
 #include "providers/twitch/eventsub/MessageBuilder.hpp"
 #include "providers/twitch/eventsub/MessageHandlers.hpp"
+#include "providers/twitch/PubSubManager.hpp"
 #include "providers/twitch/TwitchBadge.hpp"
 #include "providers/twitch/TwitchChannel.hpp"
 #include "providers/twitch/TwitchIrcServer.hpp"
@@ -416,12 +417,42 @@ bool Connection::isSubscribedTo(const SubscriptionRequest &request) const
 
 void Connection::markRequestSubscribed(const SubscriptionRequest &request)
 {
+    assert((this->twitchUserID.isEmpty() ||
+            this->twitchUserID == request.ownerTwitchUserID) &&
+           "A subscription was made when another user's subscriptions were "
+           "still active");
+
+    this->twitchUserID = request.ownerTwitchUserID;
+
     this->subscriptions.emplace(request);
 }
 
 void Connection::markRequestUnsubscribed(const SubscriptionRequest &request)
 {
     this->subscriptions.erase(request);
+
+    if (this->subscriptions.empty())
+    {
+        // TODO: Verify that it's fine for us to reuse a connection for another
+        // user after all old subscriptions are gone
+        this->twitchUserID.clear();
+    }
+}
+
+bool Connection::canHandleSubscriptionFrom(
+    const QString &otherTwitchUserID) const
+{
+    return this->twitchUserID.isEmpty() ||
+           this->twitchUserID == otherTwitchUserID;
+}
+
+void Connection::debug()
+{
+    for (const auto &request : this->subscriptions)
+    {
+        qCInfo(LOG).noquote().nospace()
+            << this->getSessionID() << " -> " << request;
+    }
 }
 
 }  // namespace chatterino::eventsub

src/providers/twitch/eventsub/Connection.hpp

@@ -90,9 +90,17 @@ public:
     void markRequestSubscribed(const SubscriptionRequest &request);
     void markRequestUnsubscribed(const SubscriptionRequest &request);
 
+    bool canHandleSubscriptionFrom(const QString &otherTwitchUserID) const;
+
+    void debug();
+
 private:
     QString sessionID;
 
+    /// The Twitch User ID that's attached to all subscriptions of this connection.
+    /// If no subscriptions are attached yet, this is an empty string, meaning it's open for any subscription.
+    QString twitchUserID;
+
     std::unordered_set<SubscriptionRequest> subscriptions;
 };
 

src/providers/twitch/eventsub/Controller.cpp

@@ -6,6 +6,7 @@
 #include "common/Version.hpp"
 #include "providers/twitch/api/Helix.hpp"
 #include "providers/twitch/eventsub/Connection.hpp"
+#include "util/QMagicEnum.hpp"
 #include "util/RenameThread.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -169,7 +170,7 @@ void Controller::removeRef(const SubscriptionRequest &request)
                 << "Refcount fell to zero for" << request
                 << "but we had no subscription ID attached - a "
                    "successful subscription was never made. From state "
-                << static_cast<uint8_t>(subscription.state);
+                << qmagicenum::enumName(subscription.state);
             return;
         }
 
@@ -203,6 +204,9 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
     assert(!this->quitting &&
            "Subscribe cannot be called while we are quitting");
 
+    assert(!request.ownerTwitchUserID.isEmpty() &&
+           "Subscription requests must include a Twitch User ID");
+
     bool needToSubscribe = false;
 
     {
@@ -251,7 +255,7 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
         subscription.refCount++;
         qCDebug(LOG) << "Added ref for" << request << subscription.refCount
                      << needToSubscribe
-                     << "state:" << static_cast<uint8_t>(subscription.state);
+                     << "state:" << qmagicenum::enumName(subscription.state);
     }
 
     auto handle = std::make_unique<RawSubscriptionHandle>(request);
@@ -314,6 +318,59 @@ void Controller::reconnectConnection(
     }
 }
 
+void Controller::debug()
+{
+    std::lock_guard g(this->subscriptionsMutex);
+    for (const auto &[request, subscription] : this->subscriptions)
+    {
+        QString sessionID;
+        auto connection = subscription.connection.lock();
+        if (connection)
+        {
+            auto *connection2 =
+                dynamic_cast<Connection *>(connection->getListener());
+            if (connection2)
+            {
+                sessionID = connection2->getSessionID();
+            }
+            else
+            {
+                sessionID = "BAD";
+            }
+        }
+        else
+        {
+            sessionID = "DEAD";
+        }
+
+        qCInfo(LOG).noquote().nospace()
+            << request << " (" << qmagicenum::enumName(subscription.state)
+            << ") -> " << sessionID;
+    }
+
+    boost::asio::post(this->ioContext, [this] {
+        for (const auto &weakConnection : this->connections)
+        {
+            auto connection = weakConnection.lock();
+            if (connection)
+            {
+                auto *connection2 =
+                    dynamic_cast<Connection *>(connection->getListener());
+                if (connection2)
+                {
+                    // qCInfo(LOG)
+                    //     << "Connected to" << connection2->getSessionID();
+                    connection2->debug();
+                }
+            }
+            else
+            {
+                qCInfo(LOG) << "Dead connection";
+            }
+        }
+    });
+}
+
 void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
 {
     // 1. Flush dead connections (maybe this should not be done here)
@@ -341,8 +398,8 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
     uint32_t openButNotReadyConnections = 0;
 
     // 2. Check if any currently open connection can handle this subscription
-    auto viableConnection =
+    auto viableConnection = this->getViableConnection(
-        this->getViableConnection(openButNotReadyConnections);
+        request.ownerTwitchUserID, openButNotReadyConnections);
 
     if (viableConnection.has_value())
     {
@@ -436,7 +493,7 @@ void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
 }
 
 std::optional<std::shared_ptr<lib::Session>> Controller::getViableConnection(
-    uint32_t &openButNotReadyConnections)
+    const QString &ownerTwitchUserID, uint32_t &openButNotReadyConnections)
 {
     for (const auto &weakConnection : this->connections)
     {
@@ -461,6 +518,11 @@ std::optional<std::shared_ptr<lib::Session>> Controller::getViableConnection(
             continue;
         }
 
+        if (!listener->canHandleSubscriptionFrom(ownerTwitchUserID))
+        {
+            continue;  // Connection is active with another Twitch User's subscriptions
+        }
+
         // TODO: Check if this listener has room for another subscription
 
         return connection;
@@ -536,9 +598,12 @@ void Controller::retrySubscription(const SubscriptionRequest &request)
         qCDebug(LOG) << "No one is interested in this subscription anymore, "
                         "stop trying"
                      << request << "from state"
-                     << static_cast<uint8_t>(subscription.state);
+                     << qmagicenum::enumName(subscription.state);
         qCDebug(LOG) << "Set state to unsubscribed" << request;
         subscription.state = Subscription::State::Unsubscribed;
+
+        this->subscriptions.erase(request);
+
         return;
     }
 
@@ -656,7 +721,7 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request)
     auto &subscription = this->subscriptions[request];
 
     qCDebug(LOG) << "Request" << request << "marked as unsubscribed from state"
-                 << static_cast<uint8_t>(subscription.state);
+                 << qmagicenum::enumName(subscription.state);
 
     assert(subscription.state == Subscription::State::Unsubscribing ||
            subscription.state == Subscription::State::Retrying);
@@ -679,6 +744,7 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request)
     if (subscription.refCount == 0)
     {
         // we could remove the subscription here
+        this->subscriptions.erase(request);
         return;
     }
 

src/providers/twitch/eventsub/Controller.hpp

@@ -53,6 +53,8 @@ public:
         std::unique_ptr<lib::Listener> connection,
         const std::optional<std::string> &reconnectURL,
         const std::unordered_set<SubscriptionRequest> &subs) = 0;
+
+    virtual void debug() = 0;
 };
 
 class Controller : public IController
@@ -73,6 +75,8 @@ public:
         const std::optional<std::string> &reconnectURL,
         const std::unordered_set<SubscriptionRequest> &subs) override;
 
+    void debug() override;
+
 private:
     void subscribe(const SubscriptionRequest &request, bool isRetry);
 
@@ -110,7 +114,8 @@ private:
     std::vector<std::weak_ptr<lib::Session>> connections;
 
     [[nodiscard]] std::optional<std::shared_ptr<lib::Session>>
-        getViableConnection(uint32_t &openButNotReadyConnections);
+        getViableConnection(const QString &ownerTwitchUserID,
+                            uint32_t &openButNotReadyConnections);
 
     struct Subscription {
         enum class State : uint8_t {
@@ -179,6 +184,10 @@ public:
         std::unique_ptr<lib::Listener> connection,
         const std::optional<std::string> &reconnectURL,
         const std::unordered_set<SubscriptionRequest> &subs) override;
+
+    void debug() override
+    {
+    }
 };
 
 }  // namespace chatterino::eventsub

src/providers/twitch/eventsub/SubscriptionRequest.hpp

@@ -18,6 +18,12 @@ struct SubscriptionRequest {
     // maybe this should be part of the enum later
     QString subscriptionVersion;
 
+    /// The ID of the Twitch User that wants to create this subscription
+    ///
+    /// Since a single websocket connection cannot share connections between users,
+    /// we need to key them by the user id even if it's not used in the condition.
+    QString ownerTwitchUserID;
+
     /// Optional list of conditions for the subscription
     std::vector<std::pair<QString, QString>> conditions;