Automatically load recent messages on reconnect (#3878)

* Add working reconnect recent messages

* Rename method to messagesUpdated

* Use audo declarations

* Add docs to new LimitedQueue methods

* Add more documentation, try atomic loading flag

* Update CHANGELOG.md

* Remove unused include

* Rename 'reconnected' signal to 'connected'

* Reserve before filtering on arbitrary update

* Extract recent messages fetching to own class

* Use std::atomic_flag instead of std::atomic_bool

* Add PostToThread include

* Add chatterino.recentmessages logging

* Remove unneeded parameters, lambda move capture

* Remove TwitchChannel::buildRecentMessages

* Add documentation, use more clear method name

* Reword changelog entry

I think it sounds better like this :)

* Rework how filling in missing messages is handled

This should hopefully prevent issues with filtered channels with old messages
that no longer exist in the underlying channel

* Check existing messages when looking for reply

* Clean up string distribution in file

* Try to improve documentation

* Use std::function for RecentMessagesApi

* Only trigger filledInMessages if we inserted

* Remove old unused lines

* Use make_shared<MessageLayout> instead of new MessageLayout

* Alphabetize QLogging categories

* Reorder CHANGELOG.md

src/providers/RecentMessagesApi.cpp

@@ -0,0 +1,229 @@
+#include "RecentMessagesApi.hpp"
+
+#include "common/Channel.hpp"
+#include "common/Common.hpp"
+#include "common/Env.hpp"
+#include "common/NetworkRequest.hpp"
+#include "common/QLogging.hpp"
+#include "providers/twitch/IrcMessageHandler.hpp"
+#include "providers/twitch/TwitchChannel.hpp"
+#include "providers/twitch/TwitchMessageBuilder.hpp"
+#include "singletons/Settings.hpp"
+#include "util/FormatTime.hpp"
+#include "util/PostToThread.hpp"
+
+#include <IrcMessage>
+#include <QJsonArray>
+#include <QJsonObject>
+#include <QUrl>
+
+namespace chatterino {
+
+namespace {
+
+    // convertClearchatToNotice takes a Communi::IrcMessage that is a CLEARCHAT
+    // command and converts it to a readable NOTICE message. This has
+    // historically been done in the Recent Messages API, but this functionality
+    // has been moved to Chatterino instead.
+    auto convertClearchatToNotice(Communi::IrcMessage *message)
+    {
+        auto channelName = message->parameter(0);
+        QString noticeMessage{};
+        if (message->tags().contains("target-user-id"))
+        {
+            auto target = message->parameter(1);
+
+            if (message->tags().contains("ban-duration"))
+            {
+                // User was timed out
+                noticeMessage =
+                    QString("%1 has been timed out for %2.")
+                        .arg(target)
+                        .arg(formatTime(
+                            message->tag("ban-duration").toString()));
+            }
+            else
+            {
+                // User was permanently banned
+                noticeMessage =
+                    QString("%1 has been permanently banned.").arg(target);
+            }
+        }
+        else
+        {
+            // Chat was cleared
+            noticeMessage = "Chat has been cleared by a moderator.";
+        }
+
+        // rebuild the raw IRC message so we can convert it back to an ircmessage again!
+        // this could probably be done in a smarter way
+
+        auto s = QString(":tmi.twitch.tv NOTICE %1 :%2")
+                     .arg(channelName)
+                     .arg(noticeMessage);
+
+        auto newMessage = Communi::IrcMessage::fromData(s.toUtf8(), nullptr);
+        newMessage->setTags(message->tags());
+
+        return newMessage;
+    }
+
+    // Parse the IRC messages returned in JSON form into Communi messages
+    std::vector<Communi::IrcMessage *> parseRecentMessages(
+        const QJsonObject &jsonRoot)
+    {
+        QJsonArray jsonMessages = jsonRoot.value("messages").toArray();
+        std::vector<Communi::IrcMessage *> messages;
+
+        if (jsonMessages.empty())
+            return messages;
+
+        for (const auto jsonMessage : jsonMessages)
+        {
+            auto content = jsonMessage.toString();
+            content.replace(COMBINED_FIXER, ZERO_WIDTH_JOINER);
+
+            auto message =
+                Communi::IrcMessage::fromData(content.toUtf8(), nullptr);
+
+            if (message->command() == "CLEARCHAT")
+            {
+                message = convertClearchatToNotice(message);
+            }
+
+            messages.emplace_back(std::move(message));
+        }
+
+        return messages;
+    }
+
+    // Build Communi messages retrieved from the recent messages API into
+    // proper chatterino messages.
+    std::vector<MessagePtr> buildRecentMessages(
+        std::vector<Communi::IrcMessage *> &messages, Channel *channel)
+    {
+        auto &handler = IrcMessageHandler::instance();
+        std::vector<MessagePtr> allBuiltMessages;
+
+        for (auto message : messages)
+        {
+            if (message->tags().contains("rm-received-ts"))
+            {
+                QDate msgDate =
+                    QDateTime::fromMSecsSinceEpoch(
+                        message->tags().value("rm-received-ts").toLongLong())
+                        .date();
+
+                // Check if we need to insert a message stating that a new day began
+                if (msgDate != channel->lastDate_)
+                {
+                    channel->lastDate_ = msgDate;
+                    auto msg = makeSystemMessage(
+                        QLocale().toString(msgDate, QLocale::LongFormat),
+                        QTime(0, 0));
+                    msg->flags.set(MessageFlag::RecentMessage);
+                    allBuiltMessages.emplace_back(msg);
+                }
+            }
+
+            auto builtMessages = handler.parseMessageWithReply(
+                channel, message, allBuiltMessages);
+
+            for (auto builtMessage : builtMessages)
+            {
+                builtMessage->flags.set(MessageFlag::RecentMessage);
+                allBuiltMessages.emplace_back(builtMessage);
+            }
+        }
+
+        return allBuiltMessages;
+    }
+
+    // Returns the URL to be used for querying the Recent Messages API for the
+    // given channel.
+    QUrl constructRecentMessagesUrl(const QString &name)
+    {
+        QUrl url(Env::get().recentMessagesApiUrl.arg(name));
+        QUrlQuery urlQuery(url);
+        if (!urlQuery.hasQueryItem("limit"))
+        {
+            urlQuery.addQueryItem(
+                "limit",
+                QString::number(getSettings()->twitchMessageHistoryLimit));
+        }
+        url.setQuery(urlQuery);
+        return url;
+    }
+
+}  // namespace
+
+void RecentMessagesApi::loadRecentMessages(const QString &channelName,
+                                           std::weak_ptr<Channel> channelPtr,
+                                           ResultCallback onLoaded,
+                                           ErrorCallback onError)
+{
+    qCDebug(chatterinoRecentMessages)
+        << "Loading recent messages for" << channelName;
+
+    QUrl url = constructRecentMessagesUrl(channelName);
+
+    NetworkRequest(url)
+        .onSuccess([channelPtr, onLoaded](NetworkResult result) -> Outcome {
+            auto shared = channelPtr.lock();
+            if (!shared)
+                return Failure;
+
+            qCDebug(chatterinoRecentMessages)
+                << "Successfully loaded recent messages for"
+                << shared->getName();
+
+            auto root = result.parseJson();
+            auto parsedMessages = parseRecentMessages(root);
+
+            // build the Communi messages into chatterino messages
+            auto builtMessages =
+                buildRecentMessages(parsedMessages, shared.get());
+
+            postToThread([shared = std::move(shared), root = std::move(root),
+                          messages = std::move(builtMessages),
+                          onLoaded]() mutable {
+                // Notify user about a possible gap in logs if it returned some messages
+                // but isn't currently joined to a channel
+                if (QString errorCode = root.value("error_code").toString();
+                    !errorCode.isEmpty())
+                {
+                    qCDebug(chatterinoRecentMessages)
+                        << QString("Got error from API: error_code=%1, "
+                                   "channel=%2")
+                               .arg(errorCode, shared->getName());
+                    if (errorCode == "channel_not_joined" && !messages.empty())
+                    {
+                        shared->addMessage(makeSystemMessage(
+                            "Message history service recovering, there may "
+                            "be gaps in the message history."));
+                    }
+                }
+
+                onLoaded(messages);
+            });
+
+            return Success;
+        })
+        .onError([channelPtr, onError](NetworkResult result) {
+            auto shared = channelPtr.lock();
+            if (!shared)
+                return;
+
+            qCDebug(chatterinoRecentMessages)
+                << "Failed to load recent messages for" << shared->getName();
+
+            shared->addMessage(makeSystemMessage(
+                QString("Message history service unavailable (Error %1)")
+                    .arg(result.status())));
+
+            onError();
+        })
+        .execute();
+}
+
+}  // namespace chatterino

src/providers/RecentMessagesApi.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include "ForwardDecl.hpp"
+
+#include <QString>
+
+#include <functional>
+#include <memory>
+#include <vector>
+
+namespace chatterino {
+
+class RecentMessagesApi
+{
+public:
+    using ResultCallback = std::function<void(const std::vector<MessagePtr> &)>;
+    using ErrorCallback = std::function<void()>;
+
+    /**
+     * @brief Loads recent messages for a channel using the Recent Messages API
+     * 
+     * @param channelName Name of Twitch channel
+     * @param channelPtr Weak pointer to Channel to use to build messages
+     * @param onLoaded Callback taking the built messages as a const std::vector<MessagePtr> &
+     * @param onError Callback called when the network request fails
+     */
+    static void loadRecentMessages(const QString &channelName,
+                                   std::weak_ptr<Channel> channelPtr,
+                                   ResultCallback onLoaded,
+                                   ErrorCallback onError);
+};
+
+}  // namespace chatterino

src/providers/irc/AbstractIrcServer.cpp

@@ -327,10 +327,13 @@ void AbstractIrcServer::onReadConnected(IrcConnection *connection)
         if (replaceMessage)
         {
             chan->replaceMessage(snapshot[snapshot.size() - 1], reconnected);
-            continue;
+        }
+        else
+        {
+            chan->addMessage(connectedMsg);
         }
 
-        chan->addMessage(connectedMsg);
+        chan->connected.invoke();
     }
 
     this->falloffCounter_ = 1;

src/providers/twitch/IrcMessageHandler.cpp

@@ -304,6 +304,8 @@ void IrcMessageHandler::populateReply(
             }
         }
 
+        MessagePtr foundMessage;
+
         // Thread does not yet exist, find root reply and create thread.
         // Linear search is justified by the infrequent use of replies
         for (auto &otherMsg : otherLoaded)
@@ -311,15 +313,29 @@ void IrcMessageHandler::populateReply(
             if (otherMsg->id == replyID)
             {
                 // Found root reply message
-                std::shared_ptr<MessageThread> newThread =
-                    std::make_shared<MessageThread>(otherMsg);
-
-                builder.setThread(newThread);
-                // Store weak reference to thread in channel
-                channel->addReplyThread(newThread);
+                foundMessage = otherMsg;
                 break;
             }
         }
+
+        if (!foundMessage)
+        {
+            // We didn't find the reply root message in the otherLoaded messages
+            // which are typically the already-parsed recent messages from the
+            // Recent Messages API. We could have a really old message that
+            // still exists being replied to, so check for that here.
+            foundMessage = channel->findMessage(replyID);
+        }
+
+        if (foundMessage)
+        {
+            std::shared_ptr<MessageThread> newThread =
+                std::make_shared<MessageThread>(foundMessage);
+
+            builder.setThread(newThread);
+            // Store weak reference to thread in channel
+            channel->addReplyThread(newThread);
+        }
     }
 }
 

src/providers/twitch/TwitchChannel.cpp

@@ -7,6 +7,7 @@
 #include "controllers/accounts/AccountController.hpp"
 #include "controllers/notifications/NotificationController.hpp"
 #include "messages/Message.hpp"
+#include "providers/RecentMessagesApi.hpp"
 #include "providers/bttv/BttvEmotes.hpp"
 #include "providers/bttv/LoadBttvChannelEmote.hpp"
 #include "providers/twitch/IrcMessageHandler.hpp"
@@ -19,7 +20,6 @@
 #include "singletons/Settings.hpp"
 #include "singletons/Toasts.hpp"
 #include "singletons/WindowManager.hpp"
-#include "util/FormatTime.hpp"
 #include "util/PostToThread.hpp"
 #include "util/QStringHash.hpp"
 #include "widgets/Window.hpp"
@@ -48,79 +48,6 @@ namespace {
     const QString LOGIN_PROMPT_TEXT("Click here to add your account again.");
     const Link ACCOUNTS_LINK(Link::OpenAccountsPage, QString());
 
-    // convertClearchatToNotice takes a Communi::IrcMessage that is a CLEARCHAT command and converts it to a readable NOTICE message
-    // This has historically been done in the Recent Messages API, but this functionality is being moved to Chatterino instead
-    auto convertClearchatToNotice(Communi::IrcMessage *message)
-    {
-        auto channelName = message->parameter(0);
-        QString noticeMessage{};
-        if (message->tags().contains("target-user-id"))
-        {
-            auto target = message->parameter(1);
-
-            if (message->tags().contains("ban-duration"))
-            {
-                // User was timed out
-                noticeMessage =
-                    QString("%1 has been timed out for %2.")
-                        .arg(target)
-                        .arg(formatTime(
-                            message->tag("ban-duration").toString()));
-            }
-            else
-            {
-                // User was permanently banned
-                noticeMessage =
-                    QString("%1 has been permanently banned.").arg(target);
-            }
-        }
-        else
-        {
-            // Chat was cleared
-            noticeMessage = "Chat has been cleared by a moderator.";
-        }
-
-        // rebuild the raw IRC message so we can convert it back to an ircmessage again!
-        // this could probably be done in a smarter way
-
-        auto s = QString(":tmi.twitch.tv NOTICE %1 :%2")
-                     .arg(channelName)
-                     .arg(noticeMessage);
-
-        auto newMessage = Communi::IrcMessage::fromData(s.toUtf8(), nullptr);
-        newMessage->setTags(message->tags());
-
-        return newMessage;
-    }
-
-    // parseRecentMessages takes a json object and returns a vector of
-    // Communi IrcMessages
-    auto parseRecentMessages(const QJsonObject &jsonRoot, ChannelPtr channel)
-    {
-        QJsonArray jsonMessages = jsonRoot.value("messages").toArray();
-        std::vector<Communi::IrcMessage *> messages;
-
-        if (jsonMessages.empty())
-            return messages;
-
-        for (const auto jsonMessage : jsonMessages)
-        {
-            auto content = jsonMessage.toString();
-            content.replace(COMBINED_FIXER, ZERO_WIDTH_JOINER);
-
-            auto message =
-                Communi::IrcMessage::fromData(content.toUtf8(), nullptr);
-
-            if (message->command() == "CLEARCHAT")
-            {
-                message = convertClearchatToNotice(message);
-            }
-
-            messages.emplace_back(std::move(message));
-        }
-
-        return messages;
-    }
     std::pair<Outcome, std::unordered_set<QString>> parseChatters(
         const QJsonObject &jsonRoot)
     {
@@ -143,6 +70,7 @@ namespace {
 
         return {Success, std::move(usernames)};
     }
+
 }  // namespace
 
 TwitchChannel::TwitchChannel(const QString &name)
@@ -181,6 +109,19 @@ TwitchChannel::TwitchChannel(const QString &name)
         this->refreshBTTVChannelEmotes(false);
     });
 
+    this->connected.connect([this]() {
+        if (this->roomId().isEmpty())
+        {
+            // If we get a reconnected event when the room id is not set, we
+            // just connected for the first time. After receiving the first
+            // message from a channel, setRoomId is called and further
+            // invocations of this event will load recent messages.
+            return;
+        }
+
+        this->loadRecentMessagesReconnect();
+    });
+
     this->messageRemovedFromStart.connect([this](MessagePtr &msg) {
         if (msg->replyThread)
         {
@@ -819,93 +760,77 @@ void TwitchChannel::loadRecentMessages()
         return;
     }
 
-    QUrl url(Env::get().recentMessagesApiUrl.arg(this->getName()));
-    QUrlQuery urlQuery(url);
-    if (!urlQuery.hasQueryItem("limit"))
+    if (this->loadingRecentMessages_.test_and_set())
     {
-        urlQuery.addQueryItem(
-            "limit", QString::number(getSettings()->twitchMessageHistoryLimit));
+        return;  // already loading
     }
-    url.setQuery(urlQuery);
 
     auto weak = weakOf<Channel>(this);
-
-    NetworkRequest(url)
-        .onSuccess([this, weak](NetworkResult result) -> Outcome {
-            auto shared = weak.lock();
-            if (!shared)
-                return Failure;
-
-            auto root = result.parseJson();
-            auto messages = parseRecentMessages(root, shared);
-
-            auto &handler = IrcMessageHandler::instance();
-
-            std::vector<MessagePtr> allBuiltMessages;
-
-            for (auto message : messages)
-            {
-                if (message->tags().contains("rm-received-ts"))
-                {
-                    QDate msgDate = QDateTime::fromMSecsSinceEpoch(
-                                        message->tags()
-                                            .value("rm-received-ts")
-                                            .toLongLong())
-                                        .date();
-                    if (msgDate != shared.get()->lastDate_)
-                    {
-                        shared.get()->lastDate_ = msgDate;
-                        auto msg = makeSystemMessage(
-                            QLocale().toString(msgDate, QLocale::LongFormat),
-                            QTime(0, 0));
-                        msg->flags.set(MessageFlag::RecentMessage);
-                        allBuiltMessages.emplace_back(msg);
-                    }
-                }
-
-                auto builtMessages = handler.parseMessageWithReply(
-                    shared.get(), message, allBuiltMessages);
-
-                for (auto builtMessage : builtMessages)
-                {
-                    builtMessage->flags.set(MessageFlag::RecentMessage);
-                    allBuiltMessages.emplace_back(builtMessage);
-                }
-            }
-
-            postToThread([this, shared, root,
-                          messages = std::move(allBuiltMessages)]() mutable {
-                shared->addMessagesAtStart(messages);
-
-                // Notify user about a possible gap in logs if it returned some messages
-                // but isn't currently joined to a channel
-                if (QString errorCode = root.value("error_code").toString();
-                    !errorCode.isEmpty())
-                {
-                    qCDebug(chatterinoTwitch)
-                        << QString("rm error_code=%1, channel=%2")
-                               .arg(errorCode, this->getName());
-                    if (errorCode == "channel_not_joined" && !messages.empty())
-                    {
-                        shared->addMessage(makeSystemMessage(
-                            "Message history service recovering, there may be "
-                            "gaps in the message history."));
-                    }
-                }
-            });
-
-            return Success;
-        })
-        .onError([weak](NetworkResult result) {
+    RecentMessagesApi::loadRecentMessages(
+        this->getName(), weak,
+        [weak](const auto &messages) {
             auto shared = weak.lock();
             if (!shared)
                 return;
 
-            shared->addMessage(makeSystemMessage(
-                QString("Message history service unavailable (Error %1)")
-                    .arg(result.status())));
-        })
-        .execute();
+            auto tc = dynamic_cast<TwitchChannel *>(shared.get());
+            if (!tc)
+                return;
+
+            tc->addMessagesAtStart(messages);
+            tc->loadingRecentMessages_.clear();
+        },
+        [weak]() {
+            auto shared = weak.lock();
+            if (!shared)
+                return;
+
+            auto tc = dynamic_cast<TwitchChannel *>(shared.get());
+            if (!tc)
+                return;
+
+            tc->loadingRecentMessages_.clear();
+        });
+}
+
+void TwitchChannel::loadRecentMessagesReconnect()
+{
+    if (!getSettings()->loadTwitchMessageHistoryOnConnect)
+    {
+        return;
+    }
+
+    if (this->loadingRecentMessages_.test_and_set())
+    {
+        return;  // already loading
+    }
+
+    auto weak = weakOf<Channel>(this);
+    RecentMessagesApi::loadRecentMessages(
+        this->getName(), weak,
+        [weak](const auto &messages) {
+            auto shared = weak.lock();
+            if (!shared)
+                return;
+
+            auto tc = dynamic_cast<TwitchChannel *>(shared.get());
+            if (!tc)
+                return;
+
+            tc->fillInMissingMessages(messages);
+            tc->loadingRecentMessages_.clear();
+        },
+        [weak]() {
+            auto shared = weak.lock();
+            if (!shared)
+                return;
+
+            auto tc = dynamic_cast<TwitchChannel *>(shared.get());
+            if (!tc)
+                return;
+
+            tc->loadingRecentMessages_.clear();
+        });
 }
 
 void TwitchChannel::refreshPubSub()

src/providers/twitch/TwitchChannel.hpp

@@ -21,6 +21,7 @@
 #include <boost/signals2.hpp>
 #include <pajlada/signals/signalholder.hpp>
 
+#include <atomic>
 #include <mutex>
 #include <unordered_map>
 
@@ -163,6 +164,7 @@ private:
     void refreshBadges();
     void refreshCheerEmotes();
     void loadRecentMessages();
+    void loadRecentMessagesReconnect();
     void fetchDisplayName();
     void cleanUpReplyThreads();
     void showLoginMessage();
@@ -188,6 +190,7 @@ private:
     int chatterCount_;
     UniqueAccess<StreamStatus> streamStatus_;
     UniqueAccess<RoomModes> roomModes_;
+    std::atomic_flag loadingRecentMessages_ = ATOMIC_FLAG_INIT;
     std::unordered_map<QString, std::weak_ptr<MessageThread>> threads_;
 
 protected: