From ce74a6685d2fa57e9bbb54d0826344ee48ee7f57 Mon Sep 17 00:00:00 2001 From: Brendan Abolivier Date: Thu, 11 Jun 2020 17:58:26 +0100 Subject: [PATCH] Save the count of unread messages to event_push_summary --- .../data_stores/main/event_push_actions.py | 53 ++++++++++++------- .../delta/59/00push_summary_unread_count.sql | 18 +++++++ synapse/storage/prepare_database.py | 2 +- 3 files changed, 53 insertions(+), 20 deletions(-) create mode 100644 synapse/storage/data_stores/main/schema/delta/59/00push_summary_unread_count.sql diff --git a/synapse/storage/data_stores/main/event_push_actions.py b/synapse/storage/data_stores/main/event_push_actions.py index 52dcc7be4..2b56e1d10 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 000000000..298516020 --- /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 9cc3b51fe..bec8da7f6 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__))