Room: fix the read markers/receipts confusion

This turns the design changes laid out in #464 comments to code:
- readMarker() now returns the fully read marker, unlike
  readMarker(User*) that returns a read receipt, even when called for
  the local user.
- Private::setLastReadEvent() -> setLastReadReceipt(), incorporating
  the "promotion" logic from promoteReadReceipt().
- The above makes promoteReadReceipt() unneeded; the remaining piece
  of logic that recalculates the number of unread messages is put to
  its own method - Private::recalculateUnreadCount().
- Private::updateUnreadCount() is only slightly refreshed, continues
  to use the fully read marker position (as it used to).
- Now that read receipts and fully read markers are managed separately,
  Private::setLastReadReceipt() has got its counterpart,
  Private::setFullyReadMarker(); both only update their respective
  markers locally (emitting signals as needed), without interaction
  with the homeserver.
- Private::markMessagesAsRead() now delegates updating the fully read
  marker to setFullyReadMarker() and on top of that sends the new
  fully read marker to the homeserver.
- Private::serverReadMarker -> fullyReadUntilEventId (to be really clear
  what it stores).
- The hand-written PostReadMarkersJob is replaced with the generated
  SetReadMarkerJob that does the same thing (and can update the read
  receipt on top, though the current code doesn't use that).

1 files changed, 177 insertions, 153 deletions

diff --git a/lib/room.cpp b/lib/room.cpp
index 85eadd67..92768aff 100644
--- a/lib/room.cpp
+++ b/lib/room.cpp
@@ -31,6 +31,7 @@
 #include "csapi/kicking.h"
 #include "csapi/leaving.h"
 #include "csapi/receipts.h"
+#include "csapi/read_markers.h"
 #include "csapi/redaction.h"
 #include "csapi/room_send.h"
 #include "csapi/room_state.h"
@@ -55,7 +56,6 @@
 #include "events/roompowerlevelsevent.h"
 #include "jobs/downloadfilejob.h"
 #include "jobs/mediathumbnailjob.h"
-#include "jobs/postreadmarkersjob.h"
 #include "events/roomcanonicalaliasevent.h"
 
 #include <QtCore/QDir>
@@ -135,7 +135,7 @@ public:
     QString firstDisplayedEventId;
     QString lastDisplayedEventId;
     QHash<const User*, QString> lastReadEventIds;
-    QString serverReadMarker;
+    QString fullyReadUntilEventId;
     TagsMap tags;
     UnorderedMap<QString, EventPtr> accountData;
     QString prevBatch;
@@ -311,11 +311,12 @@ public:
      */
     void dropDuplicateEvents(RoomEvents& events) const;
 
-    Changes setLastReadEvent(User* u, QString eventId);
-    void updateUnreadCount(const rev_iter_t& from, const rev_iter_t& to);
-    Changes promoteReadMarker(User* u, const rev_iter_t& newMarker, bool force = false);
-
-    Changes markMessagesAsRead(rev_iter_t upToMarker);
+    void setLastReadReceipt(User* u, rev_iter_t newMarker,
+                            QString newEvtId = {});
+    Changes setFullyReadMarker(const QString &eventId);
+    Changes updateUnreadCount(const rev_iter_t& from, const rev_iter_t& to);
+    Changes recalculateUnreadCount();
+    void markMessagesAsRead(const rev_iter_t &upToMarker);
 
     void getAllMembers();
 
@@ -637,139 +638,162 @@ void Room::setJoinState(JoinState state)
     emit joinStateChanged(oldState, state);
 }
 
-Room::Changes Room::Private::setLastReadEvent(User* u, QString eventId)
+void Room::Private::setLastReadReceipt(User* u, rev_iter_t newMarker,
+                                       QString newEvtId)
 {
+    if (!u) {
+        Q_ASSERT(u != nullptr); // For Debug builds
+        qCCritical(MAIN) << "Empty user, skipping read receipt registration";
+        return; // For Release builds
+    }
+    if (q->memberJoinState(u) != JoinState::Join) {
+        qCWarning(MAIN) << "Won't record read receipt for non-member" << u->id();
+        return;
+    }
+
+    if (newMarker == timeline.crend() && !newEvtId.isEmpty())
+        newMarker = q->findInTimeline(newEvtId);
+    if (newMarker != timeline.crend()) {
+        // NB: with reverse iterators, timeline history >= sync edge
+        if (newMarker >= q->readMarker(u))
+            return;
+
+        // Try to auto-promote the read marker over the user's own messages
+        // (switch to direct iterators for that).
+        const auto eagerMarker = find_if(newMarker.base(), timeline.cend(),
+                                         [=](const TimelineItem& ti) {
+                                             return ti->senderId() != u->id();
+                                         })
+                                 - 1;
+        newEvtId = (*eagerMarker)->id();
+        if (eagerMarker != newMarker.base() - 1) // &*(rIt.base() - 1) === &*rIt
+            qCDebug(EPHEMERAL) << "Auto-promoted read receipt for" << u->id()
+                               << "to" << newEvtId;
+    }
+
     auto& storedId = lastReadEventIds[u];
-    if (storedId == eventId)
-        return Change::NoChange;
+    if (storedId == newEvtId)
+        return;
+    // Finally make the change
     eventIdReadUsers.remove(storedId, u);
-    eventIdReadUsers.insert(eventId, u);
-    swap(storedId, eventId);
+    eventIdReadUsers.insert(newEvtId, u);
+    swap(storedId, newEvtId); // Now newEvtId actually stores the old eventId
     emit q->lastReadEventChanged(u);
-    emit q->readMarkerForUserMoved(u, eventId, storedId);
-    if (isLocalUser(u)) {
-        if (storedId != serverReadMarker)
-            connection->callApi<PostReadMarkersJob>(BackgroundRequest, id,
-                                                    storedId);
-        emit q->readMarkerMoved(eventId, storedId);
-        return Change::ReadMarkerChange;
-    }
-    return Change::NoChange;
+    if (!isLocalUser(u))
+        emit q->readMarkerForUserMoved(u, newEvtId, storedId);
 }
 
-void Room::Private::updateUnreadCount(const rev_iter_t& from,
-                                      const rev_iter_t& to)
+Room::Changes Room::Private::updateUnreadCount(const rev_iter_t& from,
+                                               const rev_iter_t& to)
 {
     Q_ASSERT(from >= timeline.crbegin() && from <= timeline.crend());
     Q_ASSERT(to >= from && to <= timeline.crend());
 
-    // Catch a special case when the last read event id refers to an event
-    // that has just arrived. In this case we should recalculate
-    // unreadMessages and might need to promote the read marker further
-    // over local-origin messages.
-    auto readMarker = q->readMarker();
-    if (readMarker == timeline.crend() && q->allHistoryLoaded())
-        --readMarker; // Read marker not found in the timeline, initialise it
-    if (readMarker >= from && readMarker < to) {
-        promoteReadMarker(q->localUser(), readMarker, true);
-        return;
-    }
-
-    Q_ASSERT(to <= readMarker);
+    auto fullyReadMarker = q->readMarker();
+    if (fullyReadMarker <= from)
+        return NoChange; // What's arrived is already fully read
+
+    if (fullyReadMarker == timeline.crend() && q->allHistoryLoaded())
+        --fullyReadMarker; // No read marker in the whole room, initialise it
+    if (fullyReadMarker < to) {
+        // Catch a special case when the last fully read event id refers to an
+        // event that has just arrived. In this case we should recalculate
+        // unreadMessages to get an exact number instead of an estimation
+        // (see https://github.com/quotient-im/libQuotient/wiki/unread_count).
+        // For the same reason (switching from the estimation to the exact
+        // number) this branch always returns UnreadNotifsChange, even if
+        // the estimation luckily matched the exact result.
+        recalculateUnreadCount();
+        return UnreadNotifsChange;
+    }
+
+    // Fully read marker is somewhere beyond the most historical message from
+    // the arrived batch - add up newly arrived messages to the current counter,
+    // instead of a complete recalculation.
+    Q_ASSERT(to <= fullyReadMarker);
 
     QElapsedTimer et;
     et.start();
     const auto newUnreadMessages =
-        count_if(from, to, std::bind(&Room::Private::isEventNotable, this, _1));
+        count_if(from, to,
+                 std::bind(&Room::Private::isEventNotable, this, _1));
     if (et.nsecsElapsed() > profilerMinNsecs() / 10)
         qCDebug(PROFILER) << "Counting gained unread messages took" << et;
 
-    if (newUnreadMessages > 0) {
-        // See https://github.com/quotient-im/libQuotient/wiki/unread_count
-        if (unreadMessages < 0)
-            unreadMessages = 0;
-
-        unreadMessages += newUnreadMessages;
-        qCDebug(MESSAGES) << "Room" << q->objectName() << "has gained"
-                          << newUnreadMessages << "unread message(s),"
-                          << (q->readMarker() == timeline.crend()
-                                  ? "in total at least"
-                                  : "in total")
-                          << unreadMessages << "unread message(s)";
-        emit q->unreadMessagesChanged(q);
-    }
+    if (newUnreadMessages == 0)
+        return NoChange;
+
+    // See https://github.com/quotient-im/libQuotient/wiki/unread_count
+    if (unreadMessages < 0)
+        unreadMessages = 0;
+
+    unreadMessages += newUnreadMessages;
+    qCDebug(MESSAGES) << "Room" << q->objectName() << "has gained"
+                      << newUnreadMessages << "unread message(s),"
+                      << (q->readMarker() == timeline.crend()
+                              ? "in total at least"
+                              : "in total")
+                      << unreadMessages << "unread message(s)";
+    emit q->unreadMessagesChanged(q);
+    return UnreadNotifsChange;
 }
 
-Room::Changes Room::Private::promoteReadMarker(User* u,
-                                               const rev_iter_t& newMarker,
-                                               bool force)
+Room::Changes Room::Private::recalculateUnreadCount()
 {
-    Q_ASSERT_X(u, __FUNCTION__, "User* should not be nullptr");
-    Q_ASSERT(newMarker >= timeline.crbegin() && newMarker <= timeline.crend());
+    // Recalculate unread messages
+    const auto oldUnreadCount = unreadMessages;
+    QElapsedTimer et;
+    et.start();
+    unreadMessages =
+        int(count_if(timeline.crbegin(), q->readMarker(),
+                     [this](const auto& ti) { return isEventNotable(ti); }));
+    if (et.nsecsElapsed() > profilerMinNsecs() / 10)
+        qCDebug(PROFILER) << "Recounting unread messages took" << et;
 
-    const auto prevMarker = q->readMarker(u);
-    if (!force && prevMarker <= newMarker) // Remember, we deal with reverse
-                                           // iterators
-        return Change::NoChange;
+    // See https://github.com/quotient-im/libQuotient/wiki/unread_count
+    if (unreadMessages == 0)
+        unreadMessages = -1;
 
-    Q_ASSERT(newMarker < timeline.crend());
+    if (unreadMessages == oldUnreadCount)
+        return NoChange;
 
-    // Try to auto-promote the read marker over the user's own messages
-    // (switch to direct iterators for that).
-    auto eagerMarker =
-        find_if(newMarker.base(), timeline.cend(), [=](const TimelineItem& ti) {
-            return ti->senderId() != u->id();
-        });
+    if (unreadMessages == -1)
+        qCDebug(MESSAGES)
+            << "Room" << displayname << "has no more unread messages";
+    else
+        qCDebug(MESSAGES) << "Room" << displayname << "still has"
+                          << unreadMessages << "unread message(s)";
+    emit q->unreadMessagesChanged(q);
+    return UnreadNotifsChange;
+}
 
-    auto changes = setLastReadEvent(u, (*(eagerMarker - 1))->id());
-    if (isLocalUser(u)) {
-        const auto oldUnreadCount = unreadMessages;
-        QElapsedTimer et;
-        et.start();
-        unreadMessages =
-            int(count_if(eagerMarker, timeline.cend(),
-                         [this](const auto& ti) { return isEventNotable(ti); }));
-        if (et.nsecsElapsed() > profilerMinNsecs() / 10)
-            qCDebug(PROFILER) << "Recounting unread messages took" << et;
-
-        // See https://github.com/quotient-im/libQuotient/wiki/unread_count
-        if (unreadMessages == 0)
-            unreadMessages = -1;
-
-        if (force || unreadMessages != oldUnreadCount) {
-            if (unreadMessages == -1) {
-                qCDebug(MESSAGES)
-                    << "Room" << displayname << "has no more unread messages";
-            } else
-                qCDebug(MESSAGES) << "Room" << displayname << "still has"
-                                  << unreadMessages << "unread message(s)";
-            emit q->unreadMessagesChanged(q);
-            changes |= Change::UnreadNotifsChange;
-        }
+Room::Changes Room::Private::setFullyReadMarker(const QString& eventId)
+{
+    if (fullyReadUntilEventId == eventId)
+        return NoChange;
+
+    const auto prevFullyReadId = std::exchange(fullyReadUntilEventId, eventId);
+    qCDebug(MESSAGES) << "Fully read marker moved to" << fullyReadUntilEventId;
+    emit q->readMarkerMoved(prevFullyReadId, fullyReadUntilEventId);
+
+    Changes changes = ReadMarkerChange;
+    if (const auto rm = q->readMarker(); rm != timeline.crend()) {
+        // Pull read receipt if it's behind
+        if (auto rr = q->readMarker(q->localUser()); rr > rm)
+            setLastReadReceipt(q->localUser(), rm);
+
+        changes |= recalculateUnreadCount();
     }
     return changes;
 }
 
-Room::Changes Room::Private::markMessagesAsRead(rev_iter_t upToMarker)
+void Room::Private::markMessagesAsRead(const rev_iter_t &upToMarker)
 {
-    const auto prevMarker = q->readMarker();
-    auto changes = promoteReadMarker(q->localUser(), upToMarker);
-    if (prevMarker != upToMarker)
-        qCDebug(MESSAGES) << "Marked messages as read until" << *q->readMarker();
-
-    // We shouldn't send read receipts for the local user's own messages - so
-    // search earlier messages for the latest message not from the local user
-    // until the previous last-read message, whichever comes first.
-    for (; upToMarker < prevMarker; ++upToMarker) {
-        if ((*upToMarker)->senderId() != q->localUser()->id()) {
-            connection->callApi<PostReceiptJob>(BackgroundRequest,
-                                                id, QStringLiteral("m.read"),
-                                                QUrl::toPercentEncoding(
-                                                    (*upToMarker)->id()));
-            break;
-        }
+    if (upToMarker < q->readMarker()) {
+        setFullyReadMarker((*upToMarker)->id());
+        connection->callApi<SetReadMarkerJob>(BackgroundRequest, id,
+                                              fullyReadUntilEventId);
     }
-    return changes;
 }
 
 void Room::markMessagesAsRead(QString uptoEventId)
@@ -924,6 +948,11 @@ void Room::setFirstDisplayedEventId(const QString& eventId)
     if (d->firstDisplayedEventId == eventId)
         return;
 
+    if (findInTimeline(eventId) == historyEdge())
+        qCWarning(MESSAGES)
+            << eventId
+            << "is marked as first displayed but doesn't seem to be loaded";
+
     d->firstDisplayedEventId = eventId;
     emit firstDisplayedEventChanged();
 }
@@ -946,8 +975,18 @@ void Room::setLastDisplayedEventId(const QString& eventId)
     if (d->lastDisplayedEventId == eventId)
         return;
 
+    const auto marker = findInTimeline(eventId);
+    if (marker == historyEdge())
+        qCWarning(MESSAGES)
+            << eventId
+            << "is marked as last displayed but doesn't seem to be loaded";
+
     d->lastDisplayedEventId = eventId;
     emit lastDisplayedEventChanged();
+    if (d->displayed && marker < readMarker(localUser()))
+        connection()->callApi<PostReceiptJob>(BackgroundRequest, id(),
+                                              QStringLiteral("m.read"),
+                                              QUrl::toPercentEncoding(eventId));
 }
 
 void Room::setLastDisplayedEvent(TimelineItem::index_t index)
@@ -962,11 +1001,14 @@ Room::rev_iter_t Room::readMarker(const User* user) const
     return findInTimeline(d->lastReadEventIds.value(user));
 }
 
-Room::rev_iter_t Room::readMarker() const { return readMarker(localUser()); }
+Room::rev_iter_t Room::readMarker() const
+{
+    return findInTimeline(d->fullyReadUntilEventId);
+}
 
 QString Room::readMarkerEventId() const
 {
-    return d->lastReadEventIds.value(localUser());
+    return d->fullyReadUntilEventId;
 }
 
 QList<User*> Room::usersAtEventId(const QString& eventId)
@@ -2347,22 +2389,15 @@ Room::Changes Room::Private::addNewMessageEvents(RoomEvents&& events)
         // The first event in the just-added batch (referred to by `from`)
         // defines whose read marker can possibly be promoted any further over
         // the same author's events newly arrived. Others will need explicit
-        // read receipts from the server (or, for the local user,
-        // markMessagesAsRead() invocation) to promote their read markers over
+        // read receipts from the server - or, for the local user, calling
+        // setLastDisplayedEventId() - to promote their read receipts over
         // the new message events.
-        if (const auto senderId = (*from)->senderId(); !senderId.isEmpty()) {
-            auto* const firstWriter = q->user(senderId);
-            if (firstWriter && q->readMarker(firstWriter) != timeline.crend()) {
-                roomChanges |=
-                    promoteReadMarker(firstWriter, rev_iter_t(from) - 1);
-                qCDebug(MESSAGES)
-                    << "Auto-promoted read marker for" << senderId
-                    << "to" << *q->readMarker(firstWriter);
-            }
+        if (auto* const firstWriter = q->user((*from)->senderId())) {
+            const auto firstEventId = (*from)->id();
+            if (lastReadEventIds.value(firstWriter) == firstEventId)
+                setLastReadReceipt(firstWriter, rev_iter_t(from + 1));
         }
-
-        updateUnreadCount(timeline.crbegin(), rev_iter_t(from));
-        roomChanges |= Change::UnreadNotifsChange;
+        roomChanges |= updateUnreadCount(timeline.crbegin(), rev_iter_t(from));
     }
 
     Q_ASSERT(timeline.size() == timelineSize + totalInserted);
@@ -2407,8 +2442,7 @@ void Room::Private::addHistoricalMessageEvents(RoomEvents&& events)
             emit q->updatedEvent(relation.eventId);
         }
     }
-    if (from <= q->readMarker())
-        updateUnreadCount(from, timeline.crend());
+    updateUnreadCount(from, timeline.crend());
 
     Q_ASSERT(timeline.size() == timelineSize + insertedSize);
     if (insertedSize > 9 || et.nsecsElapsed() >= profilerMinNsecs())
@@ -2612,7 +2646,7 @@ Room::Changes Room::processEphemeralEvent(EventPtr&& event)
         d->usersTyping.clear();
         for (const QString& userId : qAsConst(evt->users())) {
             auto* const u = user(userId);
-            if (u && memberJoinState(u) == JoinState::Join)
+            if (memberJoinState(u) == JoinState::Join)
                 d->usersTyping.append(u);
         }
         if (evt->users().size() > 3 || et.nsecsElapsed() >= profilerMinNsecs())
@@ -2633,25 +2667,21 @@ Room::Changes Room::processEphemeralEvent(EventPtr&& event)
                                        << p.receipts.size() << "users";
             }
             const auto newMarker = findInTimeline(p.evtId);
-            if (newMarker != timelineEdge())
-                qCDebug(EPHEMERAL) << "Event" << p.evtId
-                                   << "not found; saving read receipts anyway";
-            for (const Receipt& r : p.receipts) {
-                auto* const u = user(r.userId);
-                if (u == localUser())
-                    continue; // The local user has m.fully_read but FIXME: #464
-                if (u && memberJoinState(u) == JoinState::Join) {
+            if (newMarker == historyEdge())
+                qCDebug(EPHEMERAL) << "Event of the read receipt(s) is not "
+                                      "found; saving them anyway";
+            for (const Receipt& r : p.receipts)
+                if (auto* const u = user(r.userId);
+                    memberJoinState(u) == JoinState::Join) {
                     // If the event is not found (most likely, because it's
                     // too old and hasn't been fetched from the server yet)
                     // but there is a previous marker for a user, keep
-                    // the previous marker (XXX: why??). Otherwise, blindly
-                    // store the event id for this user.
-                    if (newMarker != timelineEdge())
-                        changes |= d->promoteReadMarker(u, newMarker);
-                    else if (readMarker(u) == timelineEdge())
-                        changes |= d->setLastReadEvent(u, p.evtId);
+                    // the previous marker because read receipts are not
+                    // supposed to move backwards. Otherwise, blindly
+                    // store the event id for this user and update the read
+                    // marker when/if the event is fetched later on.
+                    d->setLastReadReceipt(u, newMarker, p.evtId);
                 }
-            }
         }
         if (evt->eventsWithReceipts().size() > 3 || totalReceipts > 10
             || et.nsecsElapsed() >= profilerMinNsecs())
@@ -2671,15 +2701,9 @@ Room::Changes Room::processAccountDataEvent(EventPtr&& event)
         changes |= Change::TagsChange;
     }
 
-    if (auto* evt = eventCast<ReadMarkerEvent>(event)) {
-        auto readEventId = evt->event_id();
-        qCDebug(STATE) << "Server-side read marker at" << readEventId;
-        d->serverReadMarker = readEventId;
-        const auto newMarker = findInTimeline(readEventId);
-        changes |= newMarker != timelineEdge()
-                       ? d->markMessagesAsRead(newMarker)
-                       : d->setLastReadEvent(localUser(), readEventId);
-    }
+    if (auto* evt = eventCast<const ReadMarkerEvent>(event))
+        changes |= d->setFullyReadMarker(evt->event_id());
+
     // For all account data events
     auto& currentData = d->accountData[event->matrixType()];
     // A polymorphic event-specific comparison might be a bit more