fix(eventsub): check keepalive (#6058)

lib/twitch-eventsub-ws/benchmarks/src/parse.cpp

@@ -171,12 +171,16 @@ void BM_ParseAndHandleMessages(benchmark::State &state)
     auto messages = readMessages();
 
     std::unique_ptr<Listener> listener = std::make_unique<NoopListener>();
+    boost::asio::io_context ioc;
+    boost::asio::ssl::context ssl(
+        boost::asio::ssl::context::method::tls_client);
+    Session sess(ioc, ssl, std::move(listener));
 
     for (auto _ : state)
     {
         for (const auto &msg : messages)
         {
-            boost::system::error_code ec = handleMessage(listener, msg);
+            boost::system::error_code ec = sess.handleMessage(msg);
             assert(!ec);
         }
     }

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

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

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

@@ -7,21 +7,16 @@
 #include <boost/beast/websocket/ssl.hpp>
 #include <boost/json.hpp>
 
+namespace chatterino::eventsub::lib::messages {
+
+struct Metadata;
+
+}  // namespace chatterino::eventsub::lib::messages
+
 namespace chatterino::eventsub::lib {
 
 class Listener;
 
-/**
- * handleMessage takes the incoming message in the buffer, parses it
- * as JSON then forwards it to the listener, if applicable.
- *
- * This is called from the Session, and is only provided if you are interested
- * in building your own boost asio framework thing
- **/
-boost::system::error_code handleMessage(
-    std::unique_ptr<Listener> &listener,
-    const boost::beast::flat_buffer &buffer);
-
 class Session : public std::enable_shared_from_this<Session>
 {
 public:
@@ -38,13 +33,17 @@ public:
 
     Listener *getListener();
 
+    // public for testing
+    boost::system::error_code handleMessage(
+        const boost::beast::flat_buffer &buffer);
+
 private:
     void onResolve(boost::beast::error_code ec,
-                   boost::asio::ip::tcp::resolver::results_type results);
+                   const boost::asio::ip::tcp::resolver::results_type &results);
 
     void onConnect(
         boost::beast::error_code ec,
-        boost::asio::ip::tcp::resolver::results_type::endpoint_type ep);
+        const boost::asio::ip::tcp::resolver::results_type::endpoint_type &ep);
 
     void onSSLHandshake(boost::beast::error_code ec);
 
@@ -56,6 +55,14 @@ private:
 
     void fail(boost::beast::error_code ec, std::string_view op);
 
+    boost::system::error_code onSessionWelcome(
+        const messages::Metadata &metadata, const boost::json::value &jv);
+    boost::system::error_code onSessionReconnect(const boost::json::value &jv);
+    boost::system::error_code onNotification(const messages::Metadata &metadata,
+                                             const boost::json::value &jv);
+
+    void checkKeepalive();
+
     boost::asio::ip::tcp::resolver resolver;
     boost::beast::websocket::stream<
         boost::beast::ssl_stream<boost::beast::tcp_stream>>
@@ -66,6 +73,10 @@ private:
     std::string path;
     std::string userAgent;
     std::unique_ptr<Listener> listener;
+
+    std::chrono::seconds keepaliveTimeout{0};
+    bool receivedMessage = false;
+    std::unique_ptr<boost::asio::system_timer> keepaliveTimer;
 };
 
 }  // namespace chatterino::eventsub::lib

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

@@ -56,9 +56,29 @@ boost::json::result_for<Payload, boost::json::value>::type tag_invoke(
         reconnectURL = std::move(treconnectURL.value());
     }
 
+    static_assert(
+        std::is_trivially_copyable_v<std::remove_reference_t<
+            decltype(std::declval<Payload>().keepaliveTimeoutSeconds)>>);
+    std::optional<int> keepaliveTimeoutSeconds = std::nullopt;
+    const auto *jvkeepaliveTimeoutSeconds =
+        root.if_contains("keepalive_timeout_seconds");
+    if (jvkeepaliveTimeoutSeconds != nullptr &&
+        !jvkeepaliveTimeoutSeconds->is_null())
+    {
+        auto tkeepaliveTimeoutSeconds =
+            boost::json::try_value_to<int>(*jvkeepaliveTimeoutSeconds);
+
+        if (tkeepaliveTimeoutSeconds.has_error())
+        {
+            return tkeepaliveTimeoutSeconds.error();
+        }
+        keepaliveTimeoutSeconds = tkeepaliveTimeoutSeconds.value();
+    }
+
     return Payload{
         .id = std::move(id.value()),
         .reconnectURL = std::move(reconnectURL),
+        .keepaliveTimeoutSeconds = keepaliveTimeoutSeconds,
     };
 }
 

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

@@ -14,11 +14,9 @@
 #include <boost/container_hash/hash.hpp>
 #include <boost/json.hpp>
 
-#include <array>
 #include <chrono>
 #include <iostream>
 #include <memory>
-#include <sstream>
 #include <unordered_map>
 
 namespace beast = boost::beast;
@@ -37,12 +35,6 @@ using NotificationHandlers = std::unordered_map<
                                             std::unique_ptr<Listener> &)>,
     boost::hash<EventSubSubscription>>;
 
-using MessageHandlers = std::unordered_map<
-    std::string,
-    std::function<boost::system::error_code(
-        const messages::Metadata &, const boost::json::value &,
-        std::unique_ptr<Listener> &, const NotificationHandlers &)>>;
-
 namespace {
 
     template <class T>
@@ -234,124 +226,8 @@ namespace {
         // Add your new subscription types above this line
     };
 
-    const MessageHandlers MESSAGE_HANDLERS{
-        {
-            "session_welcome",
-            [](const auto &metadata, const auto &jv, auto &listener,
-               const auto & /*notificationHandlers*/) {
-                auto oPayload =
-                    parsePayload<payload::session_welcome::Payload>(jv);
-                if (!oPayload)
-                {
-                    // TODO: error handling
-                    return oPayload.error();
-                }
-                const auto &payload = *oPayload;
-
-                listener->onSessionWelcome(metadata, payload);
-                return boost::system::error_code{};
-            },
-        },
-        {
-            "session_keepalive",
-            [](const auto &metadata, const auto &jv, auto &listener,
-               const auto &notificationHandlers) {
-                // TODO: should we do something here?
-                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,
-               const auto &notificationHandlers) {
-                listener->onNotification(metadata, jv);
-
-                if (!metadata.subscriptionType || !metadata.subscriptionVersion)
-                {
-                    // TODO: error handling
-                    return boost::system::error_code{};
-                }
-
-                auto it =
-                    notificationHandlers.find({*metadata.subscriptionType,
-                                               *metadata.subscriptionVersion});
-                if (it == notificationHandlers.end())
-                {
-                    EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler);
-                }
-
-                return it->second(metadata, jv, listener);
-            },
-        },
-    };
-
 }  // namespace
 
-boost::system::error_code handleMessage(std::unique_ptr<Listener> &listener,
-                                        const beast::flat_buffer &buffer)
-{
-    boost::system::error_code parseError;
-    auto jv =
-        boost::json::parse(beast::buffers_to_string(buffer.data()), parseError);
-    if (parseError)
-    {
-        // TODO: wrap error?
-        return parseError;
-    }
-
-    const auto *jvObject = jv.if_object();
-    if (jvObject == nullptr)
-    {
-        EVENTSUB_BAIL_HERE(error::Kind::ExpectedObject);
-    }
-
-    const auto *metadataV = jvObject->if_contains("metadata");
-    if (metadataV == nullptr)
-    {
-        EVENTSUB_BAIL_HERE(error::Kind::FieldMissing);
-    }
-    auto metadataResult =
-        boost::json::try_value_to<messages::Metadata>(*metadataV);
-    if (metadataResult.has_error())
-    {
-        // TODO: wrap error?
-        return metadataResult.error();
-    }
-
-    const auto &metadata = metadataResult.value();
-
-    auto handler = MESSAGE_HANDLERS.find(metadata.messageType);
-
-    if (handler == MESSAGE_HANDLERS.end())
-    {
-        EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler);
-    }
-
-    const auto *payloadV = jvObject->if_contains("payload");
-    if (payloadV == nullptr)
-    {
-        EVENTSUB_BAIL_HERE(error::Kind::FieldMissing);
-    }
-
-    return handler->second(metadata, *payloadV, listener,
-                           NOTIFICATION_HANDLERS);
-}
-
 // Resolver and socket require an io_context
 Session::Session(boost::asio::io_context &ioc, boost::asio::ssl::context &ctx,
                  std::unique_ptr<Listener> listener)
@@ -392,8 +268,9 @@ Listener *Session::getListener()
     return this->listener.get();
 }
 
-void Session::onResolve(beast::error_code ec,
-                        boost::asio::ip::tcp::resolver::results_type results)
+void Session::onResolve(
+    beast::error_code ec,
+    const boost::asio::ip::tcp::resolver::results_type &results)
 {
     if (ec)
     {
@@ -412,7 +289,7 @@ void Session::onResolve(beast::error_code ec,
 
 void Session::onConnect(
     beast::error_code ec,
-    boost::asio::ip::tcp::resolver::results_type::endpoint_type ep)
+    const boost::asio::ip::tcp::resolver::results_type::endpoint_type &ep)
 {
     if (ec)
     {
@@ -500,7 +377,8 @@ void Session::onRead(beast::error_code ec, std::size_t bytes_transferred)
         return;
     }
 
-    auto messageError = handleMessage(this->listener, this->buffer);
+    this->receivedMessage = true;
+    auto messageError = this->handleMessage(this->buffer);
     if (messageError)
     {
         this->fail(messageError, "handleMessage");
@@ -543,8 +421,161 @@ 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)
     {
+        if (this->keepaliveTimer)
+        {
+            this->keepaliveTimer.reset();
+        }
+
         this->listener->onClose(std::move(this->listener), {});
     }
 }
 
+boost::system::error_code Session::handleMessage(
+    const beast::flat_buffer &buffer)
+{
+    boost::system::error_code parseError;
+    auto jv =
+        boost::json::parse(beast::buffers_to_string(buffer.data()), parseError);
+    if (parseError)
+    {
+        // TODO: wrap error?
+        return parseError;
+    }
+
+    const auto *jvObject = jv.if_object();
+    if (jvObject == nullptr)
+    {
+        EVENTSUB_BAIL_HERE(error::Kind::ExpectedObject);
+    }
+
+    const auto *metadataV = jvObject->if_contains("metadata");
+    if (metadataV == nullptr)
+    {
+        EVENTSUB_BAIL_HERE(error::Kind::FieldMissing);
+    }
+    auto metadataResult =
+        boost::json::try_value_to<messages::Metadata>(*metadataV);
+    if (metadataResult.has_error())
+    {
+        // TODO: wrap error?
+        return metadataResult.error();
+    }
+
+    const auto &metadata = metadataResult.value();
+
+    const auto *payloadV = jvObject->if_contains("payload");
+    if (payloadV == nullptr)
+    {
+        EVENTSUB_BAIL_HERE(error::Kind::FieldMissing);
+    }
+
+    if (metadata.messageType == "notification")
+    {
+        return this->onNotification(metadata, *payloadV);
+    }
+    if (metadata.messageType == "session_welcome")
+    {
+        return this->onSessionWelcome(metadata, *payloadV);
+    }
+    if (metadata.messageType == "session_keepalive")
+    {
+        return {};  // nothing to do
+    }
+    if (metadata.messageType == "session_reconnect")
+    {
+        return this->onSessionReconnect(*payloadV);
+    }
+    EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler);
+}
+
+boost::system::error_code Session::onSessionWelcome(
+    const messages::Metadata &metadata, const boost::json::value &jv)
+{
+    auto oPayload = parsePayload<payload::session_welcome::Payload>(jv);
+    if (!oPayload)
+    {
+        // TODO: error handling
+        return oPayload.error();
+    }
+    const auto &payload = *oPayload;
+
+    listener->onSessionWelcome(metadata, payload);
+
+    // we're graceful with the keepalive timeout
+    this->keepaliveTimeout =
+        std::chrono::seconds{payload.keepaliveTimeoutSeconds.value_or(60)} * 2;
+    assert(!this->keepaliveTimer);
+    std::cerr << "Keepalive: " << this->keepaliveTimeout.count() << 's';
+    this->checkKeepalive();
+
+    return {};
+}
+
+boost::system::error_code Session::onSessionReconnect(
+    const boost::json::value &jv)
+{
+    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 Session::onNotification(
+    const messages::Metadata &metadata, const boost::json::value &jv)
+{
+    listener->onNotification(metadata, jv);
+
+    if (!metadata.subscriptionType || !metadata.subscriptionVersion)
+    {
+        // TODO: error handling
+        return boost::system::error_code{};
+    }
+
+    auto it = NOTIFICATION_HANDLERS.find(
+        {*metadata.subscriptionType, *metadata.subscriptionVersion});
+    if (it == NOTIFICATION_HANDLERS.end())
+    {
+        EVENTSUB_BAIL_HERE(error::Kind::NoMessageHandler);
+    }
+
+    return it->second(metadata, jv, listener);
+}
+
+void Session::checkKeepalive()
+{
+    if (!this->receivedMessage)
+    {
+        std::cerr << "Keepalive timeout, closing\n";
+        if (this->listener)
+        {
+            this->listener->onClose(std::move(this->listener), {});
+        }
+        return;
+    }
+    this->receivedMessage = false;
+
+    if (this->keepaliveTimeout.count() == 0)
+    {
+        return;
+    }
+
+    this->keepaliveTimer =
+        std::make_unique<boost::asio::system_timer>(this->ws.get_executor());
+    this->keepaliveTimer->expires_after(this->keepaliveTimeout);
+    this->keepaliveTimer->async_wait([this](boost::system::error_code ec) {
+        if (ec)
+        {
+            std::cerr << "Keepalive timer cancelled: " << ec.message() << '\n';
+            return;
+        }
+        this->checkKeepalive();
+    });
+}
+
 }  // namespace chatterino::eventsub::lib

lib/twitch-eventsub-ws/tests/src/parse.cpp

@@ -154,7 +154,11 @@ TEST_P(TestHandleMessageP, Run)
     auto buf = readToFlatBuffer(filePath(GetParam() + ".json"));
 
     std::unique_ptr<Listener> listener = std::make_unique<NoOpListener>();
-    auto ec = handleMessage(listener, buf);
+    boost::asio::io_context ioc;
+    boost::asio::ssl::context ssl(
+        boost::asio::ssl::context::method::tls_client);
+    Session sess(ioc, ssl, std::move(listener));
+    auto ec = sess.handleMessage(buf);
     ASSERT_FALSE(ec.failed())
         << ec.what() << ec.message() << ec.location().to_string();
 }