feat(eventsub): reconnect (#6035)

CHANGELOG.md

@@ -37,7 +37,7 @@
 - Bugfix: Fixed color input thinking blue is also red. (#5982)
 - Bugfix: Fixed an issue where commands would sometimes reset if Chatterino was improperly shut down. (#6011)
 - Dev: Subscriptions to PubSub channel points redemption topics now use no auth token, making it continue to work during PubSub shutdown. (#5947)
-- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028)
+- Dev: Add initial experimental EventSub support. (#5837, #5895, #5897, #5904, #5910, #5903, #5915, #5916, #5930, #5935, #5932, #5943, #5952, #5953, #5968, #5973, #5974, #5980, #5981, #5985, #5990, #5992, #5993, #5996, #5995, #6000, #6001, #6002, #6003, #6005, #6007, #6010, #6008, #6012, #6013, #6015, #6017, #6027, #6028, #6035)
 - 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)

lib/twitch-eventsub-ws/include/twitch-eventsub-ws/listener.hpp

@@ -30,6 +30,10 @@ public:
     virtual void onNotification(const messages::Metadata &metadata,
                                 const boost::json::value &jv) = 0;
 
+    virtual void onClose(
+        std::unique_ptr<Listener> self,
+        const std::optional<std::string> & /* reconnectUrl */) {};
+
     // Subscription types
     virtual void onChannelBan(
         const messages::Metadata &metadata,

lib/twitch-eventsub-ws/include/twitch-eventsub-ws/payloads/session-welcome.hpp

@@ -24,6 +24,7 @@ namespace chatterino::eventsub::lib::payload::session_welcome {
 /// json_inner=session
 struct Payload {
     std::string id;
+    std::optional<std::string> reconnectURL;
 };
 
 #include "twitch-eventsub-ws/payloads/session-welcome.inc"

lib/twitch-eventsub-ws/include/twitch-eventsub-ws/session.hpp

@@ -22,20 +22,8 @@ boost::system::error_code handleMessage(
     std::unique_ptr<Listener> &listener,
     const boost::beast::flat_buffer &buffer);
 
-// Sends a WebSocket message and prints the response
 class Session : public std::enable_shared_from_this<Session>
 {
-    boost::asio::ip::tcp::resolver resolver;
-    boost::beast::websocket::stream<
-        boost::beast::ssl_stream<boost::beast::tcp_stream>>
-        ws;
-    boost::beast::flat_buffer buffer;
-    std::string host;
-    std::string port;
-    std::string path;
-    std::string userAgent;
-    std::unique_ptr<Listener> listener;
-
 public:
     // Resolver and socket require an io_context
     explicit Session(boost::asio::io_context &ioc,
@@ -65,6 +53,19 @@ private:
     void onRead(boost::beast::error_code ec, std::size_t bytes_transferred);
 
     void onClose(boost::beast::error_code ec);
+
+    void fail(boost::beast::error_code ec, std::string_view op);
+
+    boost::asio::ip::tcp::resolver resolver;
+    boost::beast::websocket::stream<
+        boost::beast::ssl_stream<boost::beast::tcp_stream>>
+        ws;
+    boost::beast::flat_buffer buffer;
+    std::string host;
+    std::string port;
+    std::string path;
+    std::string userAgent;
+    std::unique_ptr<Listener> listener;
 };
 
 }  // namespace chatterino::eventsub::lib

lib/twitch-eventsub-ws/src/generated/payloads/session-welcome.cpp

@@ -42,8 +42,23 @@ boost::json::result_for<Payload, boost::json::value>::type tag_invoke(
         return id.error();
     }
 
+    std::optional<std::string> reconnectURL = std::nullopt;
+    const auto *jvreconnectURL = root.if_contains("reconnect_url");
+    if (jvreconnectURL != nullptr && !jvreconnectURL->is_null())
+    {
+        auto treconnectURL =
+            boost::json::try_value_to<std::string>(*jvreconnectURL);
+
+        if (treconnectURL.has_error())
+        {
+            return treconnectURL.error();
+        }
+        reconnectURL = std::move(treconnectURL.value());
+    }
+
     return Payload{
         .id = std::move(id.value()),
+        .reconnectURL = std::move(reconnectURL),
     };
 }
 

lib/twitch-eventsub-ws/src/session.cpp

@@ -45,13 +45,6 @@ using MessageHandlers = std::unordered_map<
 
 namespace {
 
-    // Report a failure
-    void fail(beast::error_code ec, char const *what)
-    {
-        std::cerr << what << ": " << ec.message() << " (" << ec.location()
-                  << ")\n";
-    }
-
     template <class T>
     boost::system::result<T> parsePayload(const boost::json::value &jv)
     {
@@ -267,6 +260,21 @@ namespace {
                 return boost::system::error_code{};
             },
         },
+        {"session_reconnect",
+         [](const auto & /*metadata*/, const auto &jv, auto &listener,
+            const auto & /*notificationHandlers*/) {
+             auto oPayload =
+                 parsePayload<payload::session_welcome::Payload>(jv);
+             if (!oPayload)
+             {
+                 return oPayload.error();
+             }
+             const auto &payload = *oPayload;
+             auto *listenerPtr = listener.get();
+
+             listenerPtr->onClose(std::move(listener), payload.reconnectURL);
+             return boost::system::error_code{};
+         }},
         {
             "notification",
             [](const auto &metadata, const auto &jv, auto &listener,
@@ -389,7 +397,8 @@ void Session::onResolve(beast::error_code ec,
 {
     if (ec)
     {
-        return fail(ec, "resolve");
+        this->fail(ec, "resolve");
+        return;
     }
 
     // Set a timeout on the operation
@@ -407,7 +416,8 @@ void Session::onConnect(
 {
     if (ec)
     {
-        return fail(ec, "connect");
+        this->fail(ec, "connect");
+        return;
     }
 
     // Set a timeout on the operation
@@ -419,7 +429,8 @@ void Session::onConnect(
     {
         ec = beast::error_code(static_cast<int>(::ERR_get_error()),
                                boost::asio::error::get_ssl_category());
-        return fail(ec, "connect");
+        this->fail(ec, "connect");
+        return;
     }
 
     // Update the host_ string. This will provide the value of the
@@ -438,7 +449,8 @@ void Session::onSSLHandshake(beast::error_code ec)
 {
     if (ec)
     {
-        return fail(ec, "ssl_handshake");
+        this->fail(ec, "ssl_handshake");
+        return;
     }
 
     // Turn off the timeout on the tcp_stream, because
@@ -465,7 +477,8 @@ void Session::onHandshake(beast::error_code ec)
 {
     if (ec)
     {
-        return fail(ec, "handshake");
+        this->fail(ec, "handshake");
+        return;
     }
 
     this->ws.async_read(buffer, beast::bind_front_handler(&Session::onRead,
@@ -476,19 +489,31 @@ void Session::onRead(beast::error_code ec, std::size_t bytes_transferred)
 {
     boost::ignore_unused(bytes_transferred);
 
+    if (!this->listener)
+    {
+        return;
+    }
+
     if (ec)
     {
-        return fail(ec, "read");
+        this->fail(ec, "read");
+        return;
     }
 
     auto messageError = handleMessage(this->listener, this->buffer);
     if (messageError)
     {
-        fail(messageError, "handleMessage");
+        this->fail(messageError, "handleMessage");
     }
 
     this->buffer.clear();
 
+    if (!this->listener)
+    {
+        this->close();
+        return;
+    }
+
     this->ws.async_read(buffer, beast::bind_front_handler(&Session::onRead,
                                                           shared_from_this()));
 }
@@ -502,13 +527,24 @@ void Session::onClose(beast::error_code ec)
 {
     if (ec)
     {
-        return fail(ec, "close");
+        this->fail(ec, "close");
+        return;
     }
 
     // If we get here then the connection is closed gracefully
+    if (this->listener)
+    {
+        this->listener->onClose(std::move(this->listener), {});
+    }
+}
 
-    // The make_printable() function helps print a ConstBufferSequence
-    std::cout << beast::make_printable(buffer.data()) << std::endl;
+void Session::fail(beast::error_code ec, std::string_view op)
+{
+    std::cerr << op << ": " << ec.message() << " (" << ec.location() << ")\n";
+    if (!this->ws.is_open() && this->listener)
+    {
+        this->listener->onClose(std::move(this->listener), {});
+    }
 }
 
 }  // namespace chatterino::eventsub::lib

src/Application.cpp

@@ -592,7 +592,6 @@ pronouns::Pronouns *Application::getPronouns()
 
 eventsub::IController *Application::getEventSub()
 {
-    assertInGuiThread();
     assert(this->eventSub);
 
     return this->eventSub.get();

src/common/Args.cpp

@@ -140,6 +140,12 @@ Args::Args(const QApplication &app, const Paths &paths)
         "specified, Twitch is assumed.",
         "t:channel");
 
+#ifndef NDEBUG
+    QCommandLineOption useLocalEventsubOption(
+        "use-local-eventsub",
+        "Use the local eventsub server at 127.0.0.1:3012.");
+#endif
+
     parser.addOptions({
         {{"V", "version"}, "Displays version information."},
         crashRecoveryOption,
@@ -152,6 +158,9 @@ Args::Args(const QApplication &app, const Paths &paths)
         loginOption,
         channelLayout,
         activateOption,
+#ifndef NDEBUG
+        useLocalEventsubOption,
+#endif
     });
 
     if (!parser.parse(app.arguments()))
@@ -216,6 +225,13 @@ Args::Args(const QApplication &app, const Paths &paths)
             parseActivateOption(parser.value(activateOption));
     }
 
+#ifndef NDEBUG
+    if (parser.isSet(useLocalEventsubOption))
+    {
+        this->useLocalEventsub = true;
+    }
+#endif
+
     this->currentArguments_ = extractCommandLine(parser, {
                                                              verboseOption,
                                                              safeModeOption,

src/common/Args.hpp

@@ -64,6 +64,11 @@ public:
     bool verbose{};
     bool safeMode{};
 
+#ifndef NDEBUG
+    // twitch event websocket start-server --ssl --port 3012
+    bool useLocalEventsub = false;
+#endif
+
     QStringList currentArguments() const;
 
 private:

src/common/network/NetworkPrivate.hpp

@@ -53,6 +53,9 @@ public:
     /// By default, there's no explicit timeout for the request.
     /// To set a timeout, use NetworkRequest's timeout method
     std::optional<std::chrono::milliseconds> timeout{};
+#ifndef NDEBUG
+    bool ignoreSslErrors = false;  // for local eventsub
+#endif
 
     QString getHash();
 

src/common/network/NetworkRequest.cpp

@@ -215,4 +215,12 @@ NetworkRequest NetworkRequest::json(const QByteArray &payload) &&
         .header("Accept", "application/json");
 }
 
+#ifndef NDEBUG
+NetworkRequest NetworkRequest::ignoreSslErrors(bool ignore) &&
+{
+    this->data->ignoreSslErrors = ignore;
+    return std::move(*this);
+}
+#endif
+
 }  // namespace chatterino

src/common/network/NetworkRequest.hpp

@@ -76,6 +76,10 @@ public:
     NetworkRequest json(const QJsonDocument &document) &&;
     NetworkRequest json(const QByteArray &payload) &&;
 
+#ifndef NDEBUG
+    NetworkRequest ignoreSslErrors(bool ignore) &&;
+#endif
+
     void execute();
 
 private:

src/common/network/NetworkTask.cpp

@@ -61,6 +61,16 @@ void NetworkTask::run()
 
     QObject::connect(this->reply_, &QNetworkReply::finished, this,
                      &NetworkTask::finished);
+
+#ifndef NDEBUG
+    if (this->data_->ignoreSslErrors)
+    {
+        QObject::connect(this->reply_, &QNetworkReply::sslErrors, this,
+                         [this](const auto &errors) {
+                             this->reply_->ignoreSslErrors(errors);
+                         });
+    }
+#endif
 }
 
 QNetworkReply *NetworkTask::createReply()

src/providers/twitch/api/Helix.cpp

@@ -1,5 +1,7 @@
 #include "providers/twitch/api/Helix.hpp"
 
+#include "Application.hpp"
+#include "common/Args.hpp"
 #include "common/Literals.hpp"
 #include "common/network/NetworkRequest.hpp"
 #include "common/network/NetworkResult.hpp"
@@ -3393,7 +3395,16 @@ NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery,
         // return std::nullopt;
     }
 
-    const QString baseUrl("https://api.twitch.tv/helix/");
+    QString baseUrl("https://api.twitch.tv/helix/");
+
+#ifndef NDEBUG
+    bool ignoreSslErrors =
+        getApp()->getArgs().useLocalEventsub && url == "eventsub/subscriptions";
+    if (ignoreSslErrors)
+    {
+        baseUrl = "https://127.0.0.1:3012/";
+    }
+#endif
 
     QUrl fullUrl(baseUrl + url);
 
@@ -3403,7 +3414,11 @@ NetworkRequest Helix::makeRequest(const QString &url, const QUrlQuery &urlQuery,
         .timeout(5 * 1000)
         .header("Accept", "application/json")
         .header("Client-ID", this->clientId)
-        .header("Authorization", "Bearer " + this->oauthToken);
+        .header("Authorization", "Bearer " + this->oauthToken)
+#ifndef NDEBUG
+        .ignoreSslErrors(ignoreSslErrors)
+#endif
+        ;
 }
 
 NetworkRequest Helix::makeGet(const QString &url, const QUrlQuery &urlQuery)

src/providers/twitch/eventsub/Connection.cpp

@@ -6,6 +6,7 @@
 #include "controllers/highlights/HighlightController.hpp"
 #include "messages/Message.hpp"
 #include "messages/MessageBuilder.hpp"
+#include "providers/twitch/eventsub/Controller.hpp"
 #include "providers/twitch/eventsub/MessageBuilder.hpp"
 #include "providers/twitch/eventsub/MessageHandlers.hpp"
 #include "providers/twitch/PubSubActions.hpp"
@@ -18,7 +19,7 @@
 #include "util/PostToThread.hpp"
 
 #include <boost/json.hpp>
-#include <qdatetime.h>
+#include <QDateTime>
 #include <twitch-eventsub-ws/listener.hpp>
 #include <twitch-eventsub-ws/session.hpp>
 
@@ -71,6 +72,19 @@ void Connection::onNotification(const lib::messages::Metadata &metadata,
     qCDebug(LOG) << "on notification: " << jsonString.c_str();
 }
 
+void Connection::onClose(std::unique_ptr<lib::Listener> self,
+                         const std::optional<std::string> &reconnectURL)
+{
+    auto *app = tryGetApp();
+    if (!app)
+    {
+        return;
+    }
+
+    app->getEventSub()->reconnectConnection(std::move(self), reconnectURL,
+                                            this->subscriptions);
+}
+
 void Connection::onChannelBan(
     const lib::messages::Metadata &metadata,
     const lib::payload::channel_ban::v1::Payload &payload)
@@ -387,4 +401,9 @@ void Connection::markRequestSubscribed(const SubscriptionRequest &request)
     this->subscriptions.emplace(request);
 }
 
+void Connection::markRequestUnsubscribed(const SubscriptionRequest &request)
+{
+    this->subscriptions.erase(request);
+}
+
 }  // namespace chatterino::eventsub

src/providers/twitch/eventsub/Connection.hpp

@@ -21,6 +21,9 @@ public:
     void onNotification(const lib::messages::Metadata &metadata,
                         const boost::json::value &jv) override;
 
+    void onClose(std::unique_ptr<lib::Listener> self,
+                 const std::optional<std::string> &reconnectURL) override;
+
     void onChannelBan(
         const lib::messages::Metadata &metadata,
         const lib::payload::channel_ban::v1::Payload &payload) override;
@@ -85,7 +88,7 @@ public:
 
     bool isSubscribedTo(const SubscriptionRequest &request) const;
     void markRequestSubscribed(const SubscriptionRequest &request);
-    // TODO: Add an "markRequestUnsubscribed" method
+    void markRequestUnsubscribed(const SubscriptionRequest &request);
 
 private:
     QString sessionID;

src/providers/twitch/eventsub/Controller.cpp

@@ -1,5 +1,7 @@
 #include "providers/twitch/eventsub/Controller.hpp"
 
+#include "Application.hpp"
+#include "common/Args.hpp"
 #include "common/QLogging.hpp"
 #include "common/Version.hpp"
 #include "providers/twitch/api/Helix.hpp"
@@ -17,16 +19,16 @@
 
 namespace {
 
-/// Enable LOCAL_EVENTSUB when you want to debug eventsub with a local instance of the Twitch CLI
-/// twitch event websocket start-server --ssl --port 3012
-constexpr bool LOCAL_EVENTSUB = false;
+using namespace chatterino;
 
 std::tuple<std::string, std::string, std::string> getEventSubHost()
 {
-    if constexpr (LOCAL_EVENTSUB)
+#ifndef NDEBUG
+    if (getApp()->getArgs().useLocalEventsub)
     {
         return {"localhost", "3012", "/ws"};
     }
+#endif
 
     return {"eventsub.wss.twitch.tv", "443", "/ws"};
 }
@@ -222,6 +224,54 @@ SubscriptionHandle Controller::subscribe(const SubscriptionRequest &request)
     return handle;
 }
 
+void Controller::reconnectConnection(
+    std::unique_ptr<lib::Listener> connection,
+    const std::optional<std::string> &reconnectURL,
+    const std::unordered_set<SubscriptionRequest> &subs)
+{
+    this->clearConnections();
+    if (subs.empty())
+    {
+        return;
+    }
+
+    if (reconnectURL)
+    {
+        qCDebug(chatterinoTwitchEventSub) << "Using reconnect URL to reconnect";
+        // this is epic
+        QUrl url(QString::fromStdString(*reconnectURL));
+        this->createConnection(url.host(QUrl::FullyEncoded).toStdString(),
+                               std::to_string(url.port(443)),
+                               url.path(QUrl::FullyEncoded).toStdString(),
+                               std::move(connection));
+        return;
+    }
+
+    // no reconnect URL - something happened
+    // but first, clear the subscriptions
+    qCDebug(chatterinoTwitchEventSub)
+        << "Resubscribing to topics after connection failure";
+    {
+        std::lock_guard g(this->subscriptionsMutex);
+        for (const auto &sub : subs)
+        {
+            auto it = this->subscriptions.find(sub);
+            if (it != this->subscriptions.end())
+            {
+                qCDebug(chatterinoTwitchEventSub) << "Resetting" << it->first;
+                it->second.connection = {};
+                it->second.retryAttempts = 0;
+                it->second.state = Subscription::State::Subscribing;
+            }
+        }
+    }
+
+    for (const auto &sub : subs)
+    {
+        this->subscribe(sub, false);
+    }
+}
+
 void Controller::subscribe(const SubscriptionRequest &request, bool isRetry)
 {
     // 1. Flush dead connections (maybe this should not be done here)
@@ -355,8 +405,10 @@ std::optional<std::shared_ptr<lib::Session>> Controller::getViableConnection(
 
         auto *listener = dynamic_cast<Connection *>(connection->getListener());
 
-        assert(listener != nullptr && "Something goofy has gone wrong, Session "
-                                      "listener must be our Connection type");
+        if (!listener)
+        {
+            continue;  // dead connection
+        }
 
         if (listener->getSessionID().isEmpty())
         {
@@ -374,13 +426,23 @@ std::optional<std::shared_ptr<lib::Session>> Controller::getViableConnection(
 }
 
 void Controller::createConnection()
+{
+    this->createConnection(this->eventSubHost, this->eventSubPort,
+                           this->eventSubPath, std::make_unique<Connection>());
+}
+
+void Controller::createConnection(std::string host, std::string port,
+                                  std::string path,
+                                  std::unique_ptr<lib::Listener> listener)
 {
     try
     {
         boost::asio::ssl::context sslContext{
             boost::asio::ssl::context::tlsv12_client};
 
-        if constexpr (!LOCAL_EVENTSUB)
+#ifndef NDEBUG
+        if (!getApp()->getArgs().useLocalEventsub)
+#endif
         {
             sslContext.set_verify_mode(
                 boost::asio::ssl::verify_peer |
@@ -391,12 +453,12 @@ void Controller::createConnection()
         }
 
         auto connection = std::make_shared<lib::Session>(
-            this->ioContext, sslContext, std::make_unique<Connection>());
+            this->ioContext, sslContext, std::move(listener));
 
         this->registerConnection(connection);
 
-        connection->run(this->eventSubHost, this->eventSubPort,
-                        this->eventSubPath, this->userAgent);
+        connection->run(std::move(host), std::move(port), std::move(path),
+                        this->userAgent);
     }
     catch (std::exception &e)
     {
@@ -509,6 +571,20 @@ void Controller::markRequestSubscribed(const SubscriptionRequest &request,
 
     std::lock_guard lock(this->subscriptionsMutex);
 
+    auto strong = connection.lock();
+    if (!strong)
+    {
+        // we disconnected
+        return;
+    }
+    auto *listener = dynamic_cast<Connection *>(strong->getListener());
+    if (!listener)
+    {
+        // we disconnected
+        return;
+    }
+    listener->markRequestSubscribed(request);
+
     auto &subscription = this->subscriptions[request];
 
     assert((subscription.state == Subscription::State::Subscribing ||
@@ -560,6 +636,31 @@ void Controller::markRequestUnsubscribed(const SubscriptionRequest &request)
     qCDebug(LOG) << "Set state to unsubscribed" << request;
     subscription.state = Subscription::State::Unsubscribed;
     subscription.retryAttempts = 0;
+    auto conn = subscription.connection.lock();
+    if (conn)
+    {
+        auto *listener = dynamic_cast<Connection *>(conn->getListener());
+        if (listener)
+        {
+            listener->markRequestUnsubscribed(request);
+        }
+    }
+    subscription.connection = {};
+}
+
+void Controller::clearConnections()
+{
+    std::erase_if(this->connections, [](const auto &it) {
+        auto conn = it.lock();
+        return !conn || !conn->getListener();
+    });
+}
+
+void DummyController::reconnectConnection(
+    std::unique_ptr<lib::Listener> /* connection */,
+    const std::optional<std::string> & /* reconnectURL */,
+    const std::unordered_set<SubscriptionRequest> & /* subs */)
+{
 }
 
 }  // namespace chatterino::eventsub

src/providers/twitch/eventsub/Controller.hpp

@@ -17,6 +17,7 @@
 #include <optional>
 #include <string>
 #include <thread>
+#include <unordered_set>
 
 namespace chatterino::eventsub {
 
@@ -44,6 +45,11 @@ public:
     /// create a new connection and queue up the subscription to run again after X seconds.
     [[nodiscard]] virtual SubscriptionHandle subscribe(
         const SubscriptionRequest &request) = 0;
+
+    virtual void reconnectConnection(
+        std::unique_ptr<lib::Listener> connection,
+        const std::optional<std::string> &reconnectURL,
+        const std::unordered_set<SubscriptionRequest> &subs) = 0;
 };
 
 class Controller : public IController
@@ -59,10 +65,17 @@ public:
     [[nodiscard]] SubscriptionHandle subscribe(
         const SubscriptionRequest &request) override;
 
+    void reconnectConnection(
+        std::unique_ptr<lib::Listener> connection,
+        const std::optional<std::string> &reconnectURL,
+        const std::unordered_set<SubscriptionRequest> &subs) override;
+
 private:
     void subscribe(const SubscriptionRequest &request, bool isRetry);
 
     void createConnection();
+    void createConnection(std::string host, std::string port, std::string path,
+                          std::unique_ptr<lib::Listener> listener);
     void registerConnection(std::weak_ptr<lib::Session> &&connection);
 
     void retrySubscription(const SubscriptionRequest &request,
@@ -77,6 +90,8 @@ private:
 
     void markRequestUnsubscribed(const SubscriptionRequest &request);
 
+    void clearConnections();
+
     const std::string userAgent;
 
     std::string eventSubHost;
@@ -154,6 +169,11 @@ public:
         (void)request;
         return {};
     }
+
+    void reconnectConnection(
+        std::unique_ptr<lib::Listener> connection,
+        const std::optional<std::string> &reconnectURL,
+        const std::unordered_set<SubscriptionRequest> &subs) override;
 };
 
 }  // namespace chatterino::eventsub