Ensure live status requests are always batched (#4713)

src/controllers/twitch/LiveController.cpp

@@ -0,0 +1,190 @@
+#include "controllers/twitch/LiveController.hpp"
+
+#include "common/QLogging.hpp"
+#include "providers/twitch/api/Helix.hpp"
+#include "providers/twitch/TwitchChannel.hpp"
+#include "util/Helpers.hpp"
+
+#include <QDebug>
+
+namespace {
+
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+const auto &LOG = chatterinoTwitchLiveController;
+
+}  // namespace
+
+namespace chatterino {
+
+TwitchLiveController::TwitchLiveController()
+{
+    QObject::connect(&this->refreshTimer, &QTimer::timeout, [this] {
+        this->request();
+    });
+    this->refreshTimer.start(TwitchLiveController::REFRESH_INTERVAL);
+
+    QObject::connect(&this->immediateRequestTimer, &QTimer::timeout, [this] {
+        QStringList channelIDs;
+
+        {
+            std::unique_lock immediateRequestsLock(
+                this->immediateRequestsMutex);
+            for (const auto &channelID : this->immediateRequests)
+            {
+                channelIDs.append(channelID);
+            }
+            this->immediateRequests.clear();
+        }
+
+        if (channelIDs.isEmpty())
+        {
+            return;
+        }
+
+        this->request(channelIDs);
+    });
+    this->immediateRequestTimer.start(
+        TwitchLiveController::IMMEDIATE_REQUEST_INTERVAL);
+}
+
+void TwitchLiveController::add(const std::shared_ptr<TwitchChannel> &newChannel)
+{
+    assert(newChannel != nullptr);
+
+    const auto channelID = newChannel->roomId();
+    assert(!channelID.isEmpty());
+
+    {
+        std::unique_lock lock(this->channelsMutex);
+        this->channels[channelID] = newChannel;
+    }
+
+    {
+        std::unique_lock immediateRequestsLock(this->immediateRequestsMutex);
+        this->immediateRequests.emplace(channelID);
+    }
+}
+
+void TwitchLiveController::request(std::optional<QStringList> optChannelIDs)
+{
+    QStringList channelIDs;
+
+    if (optChannelIDs)
+    {
+        channelIDs = *optChannelIDs;
+    }
+    else
+    {
+        std::shared_lock lock(this->channelsMutex);
+
+        for (const auto &channelList : this->channels)
+        {
+            channelIDs.append(channelList.first);
+        }
+    }
+
+    if (channelIDs.isEmpty())
+    {
+        return;
+    }
+
+    auto batches =
+        splitListIntoBatches(channelIDs, TwitchLiveController::BATCH_SIZE);
+
+    qCDebug(LOG) << "Make" << batches.size() << "requests";
+
+    for (const auto &batch : batches)
+    {
+        // TODO: Explore making this concurrent
+        getHelix()->fetchStreams(
+            batch, {},
+            [this, batch{batch}](const auto &streams) {
+                std::unordered_map<QString, std::optional<HelixStream>> results;
+
+                for (const auto &channelID : batch)
+                {
+                    results[channelID] = std::nullopt;
+                }
+
+                for (const auto &stream : streams)
+                {
+                    results[stream.userId] = stream;
+                }
+
+                QStringList deadChannels;
+
+                {
+                    std::shared_lock lock(this->channelsMutex);
+                    for (const auto &result : results)
+                    {
+                        auto it = this->channels.find(result.first);
+                        if (it != channels.end())
+                        {
+                            if (auto channel = it->second.lock(); channel)
+                            {
+                                channel->updateStreamStatus(result.second);
+                            }
+                            else
+                            {
+                                deadChannels.append(result.first);
+                            }
+                        }
+                    }
+                }
+
+                if (!deadChannels.isEmpty())
+                {
+                    std::unique_lock lock(this->channelsMutex);
+                    for (const auto &deadChannel : deadChannels)
+                    {
+                        this->channels.erase(deadChannel);
+                    }
+                }
+            },
+            [] {
+                qCWarning(LOG) << "Failed stream check request";
+            },
+            [] {});
+
+        // TODO: Explore making this concurrent
+        getHelix()->fetchChannels(
+            batch,
+            [this, batch{batch}](const auto &helixChannels) {
+                QStringList deadChannels;
+
+                {
+                    std::shared_lock lock(this->channelsMutex);
+                    for (const auto &helixChannel : helixChannels)
+                    {
+                        auto it = this->channels.find(helixChannel.userId);
+                        if (it != this->channels.end())
+                        {
+                            if (auto channel = it->second.lock(); channel)
+                            {
+                                channel->updateStreamTitle(helixChannel.title);
+                                channel->updateDisplayName(helixChannel.name);
+                            }
+                            else
+                            {
+                                deadChannels.append(helixChannel.userId);
+                            }
+                        }
+                    }
+                }
+
+                if (!deadChannels.isEmpty())
+                {
+                    std::unique_lock lock(this->channelsMutex);
+                    for (const auto &deadChannel : deadChannels)
+                    {
+                        this->channels.erase(deadChannel);
+                    }
+                }
+            },
+            [] {
+                qCWarning(LOG) << "Failed stream check request";
+            });
+    }
+}
+
+}  // namespace chatterino

src/controllers/twitch/LiveController.hpp

@@ -0,0 +1,89 @@
+#pragma once
+
+#include "common/Singleton.hpp"
+#include "util/QStringHash.hpp"
+
+#include <QString>
+#include <QTimer>
+
+#include <chrono>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <shared_mutex>
+#include <unordered_map>
+#include <unordered_set>
+
+namespace chatterino {
+
+class TwitchChannel;
+
+class ITwitchLiveController
+{
+public:
+    virtual ~ITwitchLiveController() = default;
+
+    virtual void add(const std::shared_ptr<TwitchChannel> &newChannel) = 0;
+};
+
+class TwitchLiveController : public ITwitchLiveController, public Singleton
+{
+public:
+    // Controls how often all channels have their stream status refreshed
+    static constexpr std::chrono::seconds REFRESH_INTERVAL{30};
+
+    // Controls how quickly new channels have their stream status loaded
+    static constexpr std::chrono::seconds IMMEDIATE_REQUEST_INTERVAL{1};
+
+    /**
+     * How many channels to include in a single request
+     *
+     * Should not be more than 100
+     **/
+    static constexpr int BATCH_SIZE{100};
+
+    TwitchLiveController();
+
+    // Add a Twitch channel to be queried for live status
+    // A request is made within a few seconds if this is the first time this channel is added
+    void add(const std::shared_ptr<TwitchChannel> &newChannel) override;
+
+private:
+    /**
+     * Run batched Helix Channels & Stream requests for channels
+     *
+     * If a list of channel IDs is passed to request, we only make a request for those channels
+     *
+     * If no list of channels is passed to request (the default behaviour), we make requests for all channels
+     * in the `channels` map.
+     **/
+    void request(std::optional<QStringList> optChannelIDs = std::nullopt);
+
+    /**
+     * List of channel IDs pointing to their Twitch Channel
+     *
+     * These channels will have their stream status updated every REFRESH_INTERVAL seconds
+     **/
+    std::unordered_map<QString, std::weak_ptr<TwitchChannel>> channels;
+    std::shared_mutex channelsMutex;
+
+    /**
+     * List of channels that need an immediate live status update
+     *
+     * These channels will have their stream status updated after at most IMMEDIATE_REQUEST_INTERVAL seconds
+     **/
+    std::unordered_set<QString> immediateRequests;
+    std::mutex immediateRequestsMutex;
+
+    /**
+     * Timer responsible for refreshing `channels`
+     **/
+    QTimer refreshTimer;
+
+    /**
+     * Timer responsible for refreshing `immediateRequests`
+     **/
+    QTimer immediateRequestTimer;
+};
+
+}  // namespace chatterino