Implement unread counter (MSC2625)

changelog.d/7673.feature

@@ -0,0 +1 @@
+Add a per-room counter for unread messages in responses to `/sync` requests.

synapse/handlers/sync.py

@@ -1895,6 +1895,9 @@ async def _generate_room_entry(
                 if notifs is not None:
                     unread_notifications["notification_count"] = notifs["notify_count"]
                     unread_notifications["highlight_count"] = notifs["highlight_count"]
+                    unread_notifications["org.matrix.msc2625.unread_count"] = notifs[
+                        "unread_count"
+                    ]
 
                 sync_result_builder.joined.append(room_sync)
 

synapse/push/push_tools.py

@@ -39,7 +39,10 @@ def get_badge_count(store, user_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
+            # We're populating this badge using the unread_count (instead of the
+            # notify_count) as this badge is the number of missed messages, not the
+            # number of missed notifications.
+            badge += 1 if notifs["unread_count"] else 0
     return badge
 
 

synapse/rest/client/v1/push_rule.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2014-2016 OpenMarket Ltd
+# Copyright 2014-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.
@@ -267,7 +267,7 @@ def _check_actions(actions):
         raise InvalidRuleException("No actions found")
 
     for a in actions:
-        if a in ["notify", "dont_notify", "coalesce"]:
+        if a in ["notify", "dont_notify", "coalesce", "org.matrix.msc2625.mark_unread"]:
             pass
         elif isinstance(a, dict) and "set_tweak" in a:
             pass

synapse/storage/data_stores/main/event_push_actions.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2015 OpenMarket Ltd
-# Copyright 2018 New Vector Ltd
+# Copyright 2015-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.
@@ -124,21 +123,31 @@ def _get_unread_counts_by_receipt_txn(
 
     def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, stream_ordering):
 
-        # First get number of notifications.
-        # We don't need to put a notif=1 clause as all rows always have
-        # notif=1
+        # First get number of actions, grouped on whether the action notifies.
         sql = (
-            "SELECT count(*)"
+            "SELECT count(*), notif"
             " FROM event_push_actions ea"
             " WHERE"
             " user_id = ?"
             " AND room_id = ?"
             " AND stream_ordering > ?"
+            " GROUP BY notif"
         )
-
         txn.execute(sql, (user_id, room_id, stream_ordering))
-        row = txn.fetchone()
-        notify_count = row[0] if row else 0
+        rows = txn.fetchall()
+
+        # We should get a maximum number of two rows: one for notif = 0, which is the
+        # number of actions that contribute to the unread_count but not to the
+        # notify_count, and one for notif = 1, which is the number of actions that
+        # contribute to both counters. If one or both rows don't appear, then the
+        # value for the matching counter should be 0.
+        unread_count = 0
+        notify_count = 0
+        for row in rows:
+            if row[1] == 0:
+                unread_count = row[0]
+            if row[1] == 1:
+                notify_count = row[0]
 
         txn.execute(
             """
@@ -151,6 +160,10 @@ def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, stream_ordering):
         if rows:
             notify_count += rows[0][0]
 
+        # Now that we've got the final notify_count, add it to unread_count, as notify
+        # actions also contribute to the unread count.
+        unread_count += notify_count
+
         # Now get the number of highlights
         sql = (
             "SELECT count(*)"
@@ -166,7 +179,11 @@ def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, stream_ordering):
         row = txn.fetchone()
         highlight_count = row[0] if row else 0
 
-        return {"notify_count": notify_count, "highlight_count": highlight_count}
+        return {
+            "unread_count": unread_count,
+            "notify_count": notify_count,
+            "highlight_count": highlight_count,
+        }
 
     @defer.inlineCallbacks
     def get_push_action_users_in_range(self, min_stream_ordering, max_stream_ordering):
@@ -224,6 +241,7 @@ def get_after_receipt(txn):
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
+                "   AND ep.notif = 1"
                 " ORDER BY ep.stream_ordering ASC LIMIT ?"
             )
             args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
@@ -252,6 +270,7 @@ def get_no_receipt(txn):
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
+                "   AND ep.notif = 1"
                 " ORDER BY ep.stream_ordering ASC LIMIT ?"
             )
             args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
@@ -324,6 +343,7 @@ def get_after_receipt(txn):
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
+                "   AND ep.notif = 1"
                 " ORDER BY ep.stream_ordering DESC LIMIT ?"
             )
             args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
@@ -352,6 +372,7 @@ def get_no_receipt(txn):
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
+                "   AND ep.notif = 1"
                 " ORDER BY ep.stream_ordering DESC LIMIT ?"
             )
             args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
@@ -401,7 +422,7 @@ def get_if_maybe_push_in_range_for_user(self, user_id, min_stream_ordering):
         def _get_if_maybe_push_in_range_for_user_txn(txn):
             sql = """
                 SELECT 1 FROM event_push_actions
-                WHERE user_id = ? AND stream_ordering > ?
+                WHERE user_id = ? AND stream_ordering > ? AND notif = 1
                 LIMIT 1
             """
 
@@ -430,14 +451,15 @@ def add_push_actions_to_staging(self, event_id, user_id_actions):
             return
 
         # This is a helper function for generating the necessary tuple that
-        # can be used to inert into the `event_push_actions_staging` table.
+        # can be used to insert into the `event_push_actions_staging` table.
         def _gen_entry(user_id, actions):
             is_highlight = 1 if _action_has_highlight(actions) else 0
+            notif = 0 if "org.matrix.msc2625.mark_unread" in actions else 1
             return (
                 event_id,  # event_id column
                 user_id,  # user_id column
                 _serialize_action(actions, is_highlight),  # actions column
-                1,  # notif column
+                notif,  # notif column
                 is_highlight,  # highlight column
             )
 
@@ -827,7 +849,7 @@ def _rotate_notifs_before_txn(self, txn, rotate_to_stream_ordering):
                     max(stream_ordering) as stream_ordering
                 FROM event_push_actions
                 WHERE ? <= stream_ordering AND stream_ordering < ?
-                    AND highlight = 0
+                    AND highlight = 0 AND notif = 1
                 GROUP BY user_id, room_id
             ) AS upd
             LEFT JOIN event_push_summary AS old USING (user_id, room_id)

tests/replication/slave/storage/test_events.py

@@ -160,7 +160,7 @@ def test_push_actions_for_user(self):
         self.check(
             "get_unread_event_push_actions_by_room_for_user",
             [ROOM_ID, USER_ID_2, event1.event_id],
-            {"highlight_count": 0, "notify_count": 0},
+            {"highlight_count": 0, "notify_count": 0, "unread_count": 0},
         )
 
         self.persist(
@@ -173,7 +173,7 @@ def test_push_actions_for_user(self):
         self.check(
             "get_unread_event_push_actions_by_room_for_user",
             [ROOM_ID, USER_ID_2, event1.event_id],
-            {"highlight_count": 0, "notify_count": 1},
+            {"highlight_count": 0, "notify_count": 1, "unread_count": 1},
         )
 
         self.persist(
@@ -188,7 +188,7 @@ def test_push_actions_for_user(self):
         self.check(
             "get_unread_event_push_actions_by_room_for_user",
             [ROOM_ID, USER_ID_2, event1.event_id],
-            {"highlight_count": 1, "notify_count": 2},
+            {"highlight_count": 1, "notify_count": 2, "unread_count": 2},
         )
 
     def test_get_rooms_for_user_with_stream_ordering(self):

tests/storage/test_event_push_actions.py

@@ -55,13 +55,17 @@ def test_count_aggregation(self):
         user_id = "@user1235:example.com"
 
         @defer.inlineCallbacks
-        def _assert_counts(noitf_count, highlight_count):
+        def _assert_counts(unread_count, notif_count, highlight_count):
             counts = yield self.store.db.runInteraction(
                 "", self.store._get_unread_counts_by_pos_txn, room_id, user_id, 0
             )
             self.assertEquals(
                 counts,
-                {"notify_count": noitf_count, "highlight_count": highlight_count},
+                {
+                    "unread_count": unread_count,
+                    "notify_count": notif_count,
+                    "highlight_count": highlight_count,
+                },
             )
 
         @defer.inlineCallbacks
@@ -96,23 +100,23 @@ def _mark_read(stream, depth):
                 stream,
             )
 
-        yield _assert_counts(0, 0)
+        yield _assert_counts(0, 0, 0)
         yield _inject_actions(1, PlAIN_NOTIF)
-        yield _assert_counts(1, 0)
+        yield _assert_counts(1, 1, 0)
         yield _rotate(2)
-        yield _assert_counts(1, 0)
+        yield _assert_counts(1, 1, 0)
 
         yield _inject_actions(3, PlAIN_NOTIF)
-        yield _assert_counts(2, 0)
+        yield _assert_counts(2, 2, 0)
         yield _rotate(4)
-        yield _assert_counts(2, 0)
+        yield _assert_counts(2, 2, 0)
 
         yield _inject_actions(5, PlAIN_NOTIF)
         yield _mark_read(3, 3)
-        yield _assert_counts(1, 0)
+        yield _assert_counts(1, 1, 0)
 
         yield _mark_read(5, 5)
-        yield _assert_counts(0, 0)
+        yield _assert_counts(0, 0, 0)
 
         yield _inject_actions(6, PlAIN_NOTIF)
         yield _rotate(7)
@@ -121,17 +125,17 @@ def _mark_read(stream, depth):
             table="event_push_actions", keyvalues={"1": 1}, desc=""
         )
 
-        yield _assert_counts(1, 0)
+        yield _assert_counts(1, 1, 0)
 
         yield _mark_read(7, 7)
-        yield _assert_counts(0, 0)
+        yield _assert_counts(0, 0, 0)
 
         yield _inject_actions(8, HIGHLIGHT)
-        yield _assert_counts(1, 1)
+        yield _assert_counts(1, 1, 1)
         yield _rotate(9)
-        yield _assert_counts(1, 1)
+        yield _assert_counts(1, 1, 1)
         yield _rotate(10)
-        yield _assert_counts(1, 1)
+        yield _assert_counts(1, 1, 1)
 
     @defer.inlineCallbacks
     def test_find_first_stream_ordering_after_ts(self):