Save the count of unread messages to event_push_summary

3 files changed, 53 insertions, 20 deletions

diff --git a/synapse/storage/data_stores/main/event_push_actions.py b/synapse/storage/data_stores/main/event_push_actions.py
index 52dcc7be47..2b56e1d104 100644
--- a/synapse/storage/data_stores/main/event_push_actions.py
+++ b/synapse/storage/data_stores/main/event_push_actions.py
@@ -144,14 +144,15 @@ class EventPushActionsWorkerStore(SQLBaseStore):
         unread_count = 0
         notify_count = 0
         for row in rows:
-            if row[1] == 0:
-                unread_count = row[0]
+            # We always increment unread_count because actions that notify also
+            # contribute to it.
+            unread_count += row[0]
             if row[1] == 1:
                 notify_count = row[0]
 
         txn.execute(
             """
-            SELECT notif_count FROM event_push_summary
+            SELECT notif_count, unread_count FROM event_push_summary
             WHERE room_id = ? AND user_id = ? AND stream_ordering > ?
         """,
             (room_id, user_id, stream_ordering),
@@ -159,10 +160,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
         rows = txn.fetchall()
         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
+            unread_count += rows[0][1]
 
         # Now get the number of highlights
         sql = (
@@ -841,23 +839,35 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         # Calculate the new counts that should be upserted into event_push_summary
         sql = """
             SELECT user_id, room_id,
-                coalesce(old.notif_count, 0) + upd.notif_count,
+                coalesce(old.%s, 0) + upd.%s,
                 upd.stream_ordering,
                 old.user_id
             FROM (
-                SELECT user_id, room_id, count(*) as notif_count,
+                SELECT user_id, room_id, count(*) as unread_count,
                     max(stream_ordering) as stream_ordering
                 FROM event_push_actions
                 WHERE ? <= stream_ordering AND stream_ordering < ?
-                    AND highlight = 0 AND notif = 1
+                    AND highlight = 0
+                    %s
                 GROUP BY user_id, room_id
             ) AS upd
             LEFT JOIN event_push_summary AS old USING (user_id, room_id)
         """
 
-        txn.execute(sql, (old_rotate_stream_ordering, rotate_to_stream_ordering))
+        # First get the count of unread messages.
+        txn.execute(
+            sql % ("unread_count", "unread_count", ""),
+            (old_rotate_stream_ordering, rotate_to_stream_ordering),
+        )
         rows = txn.fetchall()
 
+        # Then get the count of notifications.
+        txn.execute(
+            sql % ("notify_count", "notify_count", "notif = 1"),
+            (old_rotate_stream_ordering, rotate_to_stream_ordering),
+        )
+        notif_rows = txn.fetchall()
+
         logger.info("Rotating notifications, handling %d rows", len(rows))
 
         # If the `old.user_id` above is NULL then we know there isn't already an
@@ -868,22 +878,27 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             table="event_push_summary",
             values=[
                 {
-                    "user_id": row[0],
-                    "room_id": row[1],
-                    "notif_count": row[2],
-                    "stream_ordering": row[3],
+                    "user_id": rows[i][0],
+                    "room_id": rows[i][1],
+                    "notif_count": notif_rows[i][2],
+                    "unread_count": rows[i][2],
+                    "stream_ordering": rows[i][3],
                 }
-                for row in rows
-                if row[4] is None
+                for i, _ in enumerate(rows)
+                if rows[i][4] is None
             ],
         )
 
         txn.executemany(
             """
-                UPDATE event_push_summary SET notif_count = ?, stream_ordering = ?
+                UPDATE event_push_summary
+                SET notif_count = ?, unread_count = ?, stream_ordering = ?
                 WHERE user_id = ? AND room_id = ?
             """,
-            ((row[2], row[3], row[0], row[1]) for row in rows if row[4] is not None),
+            (
+                (notif_rows[i][2], rows[i][2], rows[i][3], rows[i][0], rows[i][1])
+                for i, _ in enumerate(rows) if rows[i][4] is not None
+            ),
         )
 
         txn.execute(
diff --git a/synapse/storage/data_stores/main/schema/delta/59/00push_summary_unread_count.sql b/synapse/storage/data_stores/main/schema/delta/59/00push_summary_unread_count.sql
new file mode 100644
index 0000000000..2985160201
--- /dev/null
+++ b/synapse/storage/data_stores/main/schema/delta/59/00push_summary_unread_count.sql
@@ -0,0 +1,18 @@
+/* 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.
+ */
+
+-- Store the number of unread messages, i.e. messages that triggered either a notify
+-- action or a mark_unread one.
+ALTER TABLE event_push_summary ADD COLUMN unread_count BIGINT NOT NULL;
diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py
index 9cc3b51fe6..bec8da7f62 100644
--- a/synapse/storage/prepare_database.py
+++ b/synapse/storage/prepare_database.py
@@ -34,7 +34,7 @@ logger = logging.getLogger(__name__)
 # XXX: If you're about to bump this to 59 (or higher) please create an update
 # that drops the unused `cache_invalidation_stream` table, as per #7436!
 # XXX: Also add an update to drop `account_data_max_stream_id` as per #7656!
-SCHEMA_VERSION = 58
+SCHEMA_VERSION = 59
 
 dir_path = os.path.abspath(os.path.dirname(__file__))