Re-implement unread counts

changelog.d/7736.feature

@@ -0,0 +1 @@
+Add unread messages count to sync responses.

synapse/handlers/sync.py

@@ -103,6 +103,7 @@ class JoinedSyncResult:
     account_data = attr.ib(type=List[JsonDict])
     unread_notifications = attr.ib(type=JsonDict)
     summary = attr.ib(type=Optional[JsonDict])
+    unread_count = attr.ib(type=int)
 
     def __nonzero__(self) -> bool:
         """Make the result appear empty if there are no updates. This is used
@@ -951,6 +952,23 @@ async def unread_notifs_for_room_id(
         # count is whatever it was last time.
         return None
 
+    async def unread_messages_for_room_id(
+        self, room_id: str, sync_config: SyncConfig,
+    ) -> int:
+        """Retrieve the count of unread message for the current user in the given room.
+        """
+        with Measure(self.clock, "unread_messages_for_room_id"):
+            last_unread_event_id = await self.store.get_last_receipt_event_id_for_user(
+                user_id=sync_config.user.to_string(),
+                room_id=room_id,
+                receipt_type="m.read",
+            )
+
+            count = await self.store.get_unread_message_count_for_user(
+                sync_config.user.to_string(), room_id, last_unread_event_id
+            )
+            return count
+
     async def generate_sync_result(
         self,
         sync_config: SyncConfig,
@@ -1877,6 +1895,8 @@ async def _generate_room_entry(
 
         if room_builder.rtype == "joined":
             unread_notifications = {}  # type: Dict[str, str]
+
+            unread_count = await self.unread_messages_for_room_id(room_id, sync_config)
             room_sync = JoinedSyncResult(
                 room_id=room_id,
                 timeline=batch,
@@ -1885,6 +1905,7 @@ async def _generate_room_entry(
                 account_data=account_data_events,
                 unread_notifications=unread_notifications,
                 summary=summary,
+                unread_count=unread_count,
             )
 
             if room_sync or always_include:

synapse/push/push_tools.py

@@ -32,14 +32,14 @@ def get_badge_count(store, user_id):
         if room_id in my_receipts_by_room:
             last_unread_event_id = my_receipts_by_room[room_id]
 
-            notifs = yield (
-                store.get_unread_event_push_actions_by_room_for_user(
-                    room_id, user_id, last_unread_event_id
+            unread_count = yield defer.ensureDeferred(
+                store.get_unread_message_count_for_user(
+                    user_id, room_id, last_unread_event_id
                 )
             )
             # return one badge count per conversation, as count per
             # message is so noisy as to be almost useless
-            badge += 1 if notifs["notify_count"] else 0
+            badge += 1 if unread_count else 0
     return badge
 
 

synapse/rest/client/v2_alpha/sync.py

@@ -417,6 +417,7 @@ def serialize(events):
             result["ephemeral"] = {"events": ephemeral_events}
             result["unread_notifications"] = room.unread_notifications
             result["summary"] = room.summary
+            result["org.matrix.msc2654.unread_count"] = room.unread_count
 
         return result
 

synapse/storage/data_stores/main/events.py

@@ -25,6 +25,7 @@
 from canonicaljson import json
 from prometheus_client import Counter
 
+from twisted.enterprise.adbapi import Connection
 from twisted.internet import defer
 
 import synapse.metrics
@@ -61,6 +62,13 @@
     ["type", "origin_type", "origin_entity"],
 )
 
+STATE_EVENT_TYPES_TO_MARK_UNREAD = [
+    EventTypes.PowerLevels,
+    EventTypes.Topic,
+    EventTypes.Name,
+    EventTypes.RoomAvatar,
+]
+
 
 def encode_json(json_object):
     """
@@ -977,7 +985,7 @@ def _update_metadata_tables_txn(
             txn, events=[event for event, _ in events_and_contexts]
         )
 
-        for event, _ in events_and_contexts:
+        for event, context in events_and_contexts:
             if event.type == EventTypes.Name:
                 # Insert into the event_search table.
                 self._store_room_name_txn(txn, event)
@@ -990,6 +998,8 @@ def _update_metadata_tables_txn(
             elif event.type == EventTypes.Redaction and event.redacts is not None:
                 # Insert into the redactions table.
                 self._store_redaction(txn, event)
+                # If the redacted event was unread, revert that.
+                self._handle_redacted_unread_event_txn(txn, event)
             elif event.type == EventTypes.Retention:
                 # Update the room_retention table.
                 self._store_retention_policy_for_room_txn(txn, event)
@@ -1009,6 +1019,8 @@ def _update_metadata_tables_txn(
                 if isinstance(expiry_ts, int) and not event.is_state():
                     self._insert_event_expiry_txn(txn, event.event_id, expiry_ts)
 
+            self._maybe_insert_unread_event_txn(txn, event, context)
+
         # Insert into the room_memberships table.
         self._store_room_members_txn(
             txn,
@@ -1614,3 +1626,87 @@ def f(txn, stream_ordering):
             await self.db.runInteraction("locally_reject_invite", f, stream_ordering)
 
         return stream_ordering
+
+    def _maybe_insert_unread_event_txn(
+        self, txn: Connection, event: EventBase, context: EventContext,
+    ):
+        """Mark the event as unread for every current member of the room if it passes the
+        conditions for that.
+
+        These conditions are: the event must either have a non-empty string body, be an
+        encrypted message, or be either a power levels event, a room name event or a room
+        topic event, and must be neither rejected or soft-failed nor an edit or a notice.
+
+        Args:
+            txn: The transaction to use to retrieve room members and to mark the event
+                as unread.
+            event: The event to evaluate and maybe mark as unread.
+            context: The context in which the event was sent (used to figure out whether
+                the event has been rejected).
+        """
+        content = event.content
+
+        is_edit = (
+            content.get("m.relates_to", {}).get("rel_type") == RelationTypes.REPLACE
+        )
+        is_notice = not event.is_state() and content.get("msgtype") == "m.notice"
+
+        # We don't want rejected or soft-failed events, edits or notices to be marked
+        # unread.
+        if (
+            context.rejected
+            or is_edit
+            or is_notice
+            or event.internal_metadata.is_soft_failed()
+        ):
+            return
+
+        body_exists = isinstance(content.get("body"), str)
+        is_state_event_to_mark_unread = (
+            event.is_state() and event.type in STATE_EVENT_TYPES_TO_MARK_UNREAD
+        )
+        is_encrypted_message = (
+            not event.is_state() and event.type == EventTypes.Encrypted
+        )
+
+        # We want to mark unread messages with a non-empty string body, some state events
+        # (power levels, room name, room topic, room avatar) and encrypted messages.
+        if not (body_exists or is_state_event_to_mark_unread or is_encrypted_message):
+            return
+
+        # Get the list of users that are currently joined to the room.
+        users_in_room = self.db.simple_select_onecol_txn(
+            txn=txn,
+            table="current_state_events",
+            keyvalues={"membership": Membership.JOIN, "room_id": event.room_id},
+            retcol="state_key",
+        )  # type: list
+
+        # Only insert rows for local users.
+        local_users_in_room = list(
+            filter(lambda user_id: self.hs.is_mine_id(user_id), users_in_room)
+        )
+
+        # Mark the message as unread for every user currently in the room, except the
+        # sender of the event (because even if they haven't sent a read receipt for the
+        # event, it seems dumb to show it as unread to its sender).
+        self.db.simple_insert_many_txn(
+            txn=txn,
+            table="unread_messages",
+            values=[
+                {
+                    "user_id": user_id,
+                    "stream_ordering": event.internal_metadata.stream_ordering,
+                    "room_id": event.room_id,
+                    "event_id": event.event_id,
+                }
+                for user_id in local_users_in_room
+                if user_id != event.sender
+            ],
+        )
+
+    def _handle_redacted_unread_event_txn(self, txn: Connection, event: EventBase):
+        # Redact every row for this event in the unread_messages table.
+        self.db.simple_delete_txn(
+            txn=txn, table="unread_messages", keyvalues={"event_id": event.redacts}
+        )

synapse/storage/data_stores/main/events_worker.py

@@ -41,7 +41,7 @@
 from synapse.replication.tcp.streams import BackfillStream
 from synapse.replication.tcp.streams.events import EventsStream
 from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause
-from synapse.storage.database import Database
+from synapse.storage.database import Database, LoggingTransaction
 from synapse.storage.util.id_generators import StreamIdGenerator
 from synapse.types import get_domain_from_id
 from synapse.util.caches.descriptors import Cache, cached, cachedInlineCallbacks
@@ -1362,6 +1362,62 @@ def get_next_event_to_expire_txn(txn):
             desc="get_next_event_to_expire", func=get_next_event_to_expire_txn
         )
 
+    async def get_unread_message_count_for_user(
+        self, user_id: str, room_id: str, last_read_event_id: str,
+    ):
+        return await self.db.runInteraction(
+            "get_unread_message_count_for_user",
+            self._get_unread_message_count_for_user_txn,
+            user_id,
+            room_id,
+            last_read_event_id,
+        )
+
+    def _get_unread_message_count_for_user_txn(
+        self,
+        txn: LoggingTransaction,
+        user_id: str,
+        room_id: str,
+        last_read_event_id: Optional[str],
+    ):
+        if last_read_event_id:
+            # Get the stream ordering for the last read event.
+            stream_ordering = self.db.simple_select_one_onecol_txn(
+                txn=txn,
+                table="events",
+                keyvalues={"room_id": room_id, "event_id": last_read_event_id},
+                retcol="stream_ordering",
+            )
+        else:
+            # If there's no read receipt for that room, it probably means the user hasn't
+            # opened it yet, in which case use the stream ID of their join event.
+            # We can't just set it to 0 otherwise messages from other local users from
+            # before this user joined will be counted as well.
+            txn.execute(
+                """
+                SELECT stream_ordering FROM room_memberships
+                LEFT JOIN events USING (event_id, room_id)
+                WHERE membership = 'join'
+                    AND user_id = ?
+                    AND room_id = ?
+                """,
+                (user_id, room_id),
+            )
+            row = txn.fetchone()
+            stream_ordering = row[0]
+
+        # Count the messages that qualify as unread after the stream ordering we've just
+        # retrieved.
+        sql = """
+            SELECT COUNT(*) FROM unread_messages
+            WHERE user_id = ? AND room_id = ? AND stream_ordering > ?
+        """
+
+        txn.execute(sql, (user_id, room_id, stream_ordering))
+        row = txn.fetchone()
+
+        return row[0] if row else 0
+
 
 AllNewEventsResult = namedtuple(
     "AllNewEventsResult",

synapse/storage/data_stores/main/schema/delta/58/09unread_messages.sql

@@ -0,0 +1,22 @@
+/* Copyright 2020 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE IF NOT EXISTS unread_messages(
+    user_id TEXT NOT NULL,  -- The user for which the message is unread.
+    stream_ordering BIGINT NOT NULL,  -- The position of the message in the event stream.
+    room_id TEXT NOT NULL,
+    event_id TEXT NOT NULL,  -- The ID of the message, we need it to handle redactions.
+    UNIQUE (user_id, stream_ordering)
+);

tests/rest/client/v1/utils.py

@@ -143,6 +143,26 @@ def send_event(
 
         return channel.json_body
 
+    def redact(self, room_id, event_id, txn_id=None, tok=None, expect_code=200):
+        if txn_id is None:
+            txn_id = "m%s" % (str(time.time()))
+
+        path = "/_matrix/client/r0/rooms/%s/redact/%s/%s" % (room_id, event_id, txn_id)
+        if tok:
+            path = path + "?access_token=%s" % tok
+
+        request, channel = make_request(
+            self.hs.get_reactor(), "PUT", path, json.dumps({}).encode("utf8")
+        )
+        render(request, self.resource, self.hs.get_reactor())
+
+        assert int(channel.result["code"]) == expect_code, (
+            "Expected: %d, got: %d, resp: %r"
+            % (expect_code, int(channel.result["code"]), channel.result["body"])
+        )
+
+        return channel.json_body
+
     def _read_write_state(
         self,
         room_id: str,