Add local_current_membership table

synapse/storage/data_stores/main/events.py

@@ -128,6 +128,7 @@ def _censor_redactions():
             hs.get_clock().looping_call(_censor_redactions, 5 * 60 * 1000)
 
         self._ephemeral_messages_enabled = hs.config.enable_ephemeral_messages
+        self.is_mine_id = hs.is_mine_id
 
     @defer.inlineCallbacks
     def _read_forward_extremities(self):
@@ -547,6 +548,34 @@ def _update_current_state_txn(self, txn, state_delta_by_room, stream_id):
                 ],
             )
 
+            # Note: Do we really want to delete rows here (that we do
+            # subsequently reinsert below)? While technically correct it means
+            # we have no record of the fact the user *was* a member of the
+            # room but got, say, state reset out of it.
+            if to_delete or to_insert:
+                txn.executemany(
+                    "DELETE FROM local_current_membership"
+                    " WHERE room_id = ? AND user_id = ?",
+                    (
+                        (room_id, state_key)
+                        for etype, state_key in itertools.chain(to_delete, to_insert)
+                        if etype == EventTypes.Member and self.is_mine_id(state_key)
+                    ),
+                )
+
+            if to_insert:
+                txn.executemany(
+                    """INSERT INTO local_current_membership
+                        (room_id, user_id, event_id, membership)
+                    VALUES (?, ?, ?, (SELECT membership FROM room_memberships WHERE event_id = ?))
+                    """,
+                    [
+                        (room_id, key[1], ev_id, ev_id)
+                        for key, ev_id in to_insert.items()
+                        if key[0] == EventTypes.Member and self.is_mine_id(key[1])
+                    ],
+                )
+
             txn.call_after(
                 self._curr_state_delta_stream_cache.entity_has_changed,
                 room_id,
@@ -1724,6 +1753,7 @@ def _purge_room_txn(self, txn, room_id):
             "local_invites",
             "room_account_data",
             "room_tags",
+            "local_current_membership",
         ):
             logger.info("[purge] removing %s from %s", room_id, table)
             txn.execute("DELETE FROM %s WHERE room_id=?" % (table,), (room_id,))

synapse/storage/data_stores/main/roommember.py

@@ -1064,6 +1064,23 @@ def _store_room_members_txn(self, txn, events, backfilled):
                         ),
                     )
 
+                # We also update the `local_current_membership` table with
+                # latest invite info. This will usually get updated by the
+                # `current_state_events` handling, unless its an outlier.
+                if event.internal_metadata.is_outlier():
+                    self.db.simple_upsert_txn(
+                        txn,
+                        table="local_current_membership",
+                        keyvalues={
+                            "room_id": event.room_id,
+                            "user_id": event.state_key,
+                        },
+                        values={
+                            "event_id": event.event_id,
+                            "membership": event.membership,
+                        },
+                    )
+
     @defer.inlineCallbacks
     def locally_reject_invite(self, user_id, room_id):
         sql = (
@@ -1075,6 +1092,15 @@ def locally_reject_invite(self, user_id, room_id):
         def f(txn, stream_ordering):
             txn.execute(sql, (stream_ordering, True, room_id, user_id))
 
+            # We also clear this entry from `local_current_membership`.
+            # Ideally we'd point to a leave event, but we don't have one, so
+            # nevermind.
+            self.db.simple_delete_txn(
+                txn,
+                table="local_current_membership",
+                keyvalues={"room_id": room_id, "user_id": user_id},
+            )
+
         with self._stream_id_gen.get_next() as stream_ordering:
             yield self.db.runInteraction("locally_reject_invite", f, stream_ordering)
 

synapse/storage/data_stores/main/schema/delta/57/local_current_membership.sql

@@ -0,0 +1,38 @@
+/* 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.
+ */
+
+-- Stores the latest membership state of local users in rooms, which helps
+-- track leaves/bans/etc even if the server has left the room (and so has
+-- deleted the room from `current_state_events`).
+--
+-- This may take a bit of time for large servers (e.g. 40s for matrix.org) but
+-- means we avoid a lots of book keeping required to do it as a background
+-- update.
+--
+-- We join against `room_memberships` as `current_state_events.membership` may
+-- not have been filled in yet when migrating from old schemas.
+CREATE TABLE local_current_membership AS
+    SELECT room_id, state_key AS user_id, event_id, room_memberships.membership
+    FROM current_state_events
+    INNER JOIN users ON (name = state_key)
+    INNER JOIN room_memberships USING (room_id, event_id)
+    WHERE type = 'm.room.member';
+
+-- Adds the appropriate indices
+INSERT INTO background_updates (update_name, progress_json) VALUES
+  ('local_current_membership_idx', '{}');
+
+INSERT INTO background_updates (update_name, progress_json) VALUES
+  ('local_current_membership_rm_idx', '{}');

synapse/storage/data_stores/main/state.py

@@ -293,6 +293,8 @@ class MainStateBackgroundUpdateStore(SQLBaseStore):
 
     CURRENT_STATE_INDEX_UPDATE_NAME = "current_state_members_idx"
     EVENT_STATE_GROUP_INDEX_UPDATE_NAME = "event_to_state_groups_sg_index"
+    LOCAL_CURRENT_MEMBERSHIP_UPDATE_NAME = "local_current_membership_idx"
+    LOCAL_CURRENT_MEMBERSHIP_ROOM_UPDATE_NAME = "local_current_membership_rm_idx"
 
     def __init__(self, database: Database, db_conn, hs):
         super(MainStateBackgroundUpdateStore, self).__init__(database, db_conn, hs)
@@ -310,6 +312,19 @@ def __init__(self, database: Database, db_conn, hs):
             table="event_to_state_groups",
             columns=["state_group"],
         )
+        self.db.updates.register_background_index_update(
+            self.LOCAL_CURRENT_MEMBERSHIP_UPDATE_NAME,
+            index_name="local_current_membership_idx",
+            table="local_current_membership",
+            columns=["user_id", "room_id"],
+            unique=True,
+        )
+        self.db.updates.register_background_index_update(
+            self.LOCAL_CURRENT_MEMBERSHIP_ROOM_UPDATE_NAME,
+            index_name="local_current_membership_room_idx",
+            table="local_current_membership",
+            columns=["room_id"],
+        )
 
 
 class StateStore(StateGroupWorkerStore, MainStateBackgroundUpdateStore):

synapse/storage/database.py

@@ -73,6 +73,7 @@
     "device_lists_remote_extremeties": "device_lists_remote_extremeties_unique_idx",
     "device_lists_remote_cache": "device_lists_remote_cache_unique_idx",
     "event_search": "event_search_event_id_idx",
+    "local_current_membership": "local_current_membership_idx",
 }
 
 

synapse/storage/prepare_database.py

@@ -29,7 +29,7 @@
 
 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 56
+SCHEMA_VERSION = 57
 
 dir_path = os.path.abspath(os.path.dirname(__file__))