From 1a251d5211f1217049661c7bcae06d99c5103d7f Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Tue, 6 Aug 2024 15:18:34 -0500 Subject: [PATCH] Fill in `sliding_sync_non_join_memberships` when current state changes --- synapse/storage/databases/main/events.py | 107 ++++++++++++++++-- .../delta/87/01_sliding_sync_memberships.sql | 30 +++-- 2 files changed, 122 insertions(+), 15 deletions(-) diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 1ce7578160..3a0eebe411 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -53,6 +53,7 @@ from synapse.storage.database import ( DatabasePool, LoggingDatabaseConnection, LoggingTransaction, + make_tuple_in_list_sql_clause, ) from synapse.storage.databases.main.event_federation import EventFederationStore from synapse.storage.databases.main.events_worker import EventCacheEntry @@ -1159,12 +1160,11 @@ class PersistEventsStore: # We find out which membership events we may have deleted # and which we have added, then we invalidate the caches for all # those users. - members_changed = { + members_to_cache_bust = { state_key for ev_type, state_key in itertools.chain(to_delete, to_insert) if ev_type == EventTypes.Member } - members_to_cache_bust = members_changed.copy() if delta_state.no_longer_in_room: # Server is no longer in the room so we delete the room from @@ -1317,7 +1317,9 @@ class PersistEventsStore: ) sliding_sync_joined_rooms_insert_map["is_encrypted"] = is_encrypted elif event_id == room_name_event_id: - room_name = event_json.get("content", {}).get("name") + room_name = event_json.get("content", {}).get( + EventContentFields.ROOM_NAME + ) sliding_sync_joined_rooms_insert_map["room_name"] = room_name else: raise AssertionError( @@ -1325,13 +1327,14 @@ class PersistEventsStore: ) # Update the `sliding_sync_joined_rooms` table - insert_keys, insert_values = sliding_sync_joined_rooms_insert_map.items() + insert_keys = sliding_sync_joined_rooms_insert_map.keys() + insert_values = sliding_sync_joined_rooms_insert_map.values() if len(insert_keys) > 0: # TODO: Should we add `bump_stamp` on insert? txn.execute( f""" INSERT INTO sliding_sync_joined_rooms - (room_id, stream_ordering, {", ".join(insert_keys)}) + (room_id, event_stream_ordering, {", ".join(insert_keys)}) VALUES ( ?, (SELECT stream_ordering FROM events WHERE event_id = ?) @@ -1381,9 +1384,97 @@ class PersistEventsStore: # We now update `sliding_sync_non_join_memberships`. We do this regardless of # whether the server is still in the room or not because we still want a row if - # we just left/kicked or got banned from the room. - # - # TODO + # a local user was just left/kicked or got banned from the room. + if to_insert: + membership_event_ids: List[str] = [] + for state_key, event_id in to_insert.items(): + if state_key[0] == EventTypes.Member and self.is_mine_id(state_key[1]): + membership_event_ids.append(event_id) + + # Fetch the events from the database + ( + event_type_and_state_key_in_list_clause, + event_type_and_state_key_args, + ) = make_tuple_in_list_sql_clause( + self.database_engine, + ("type", "state_key"), + [ + (EventTypes.Create, ""), + (EventTypes.RoomEncryption, ""), + (EventTypes.Name, ""), + ], + ) + txn.execute( + f""" + SELECT event_id, type, state_key, json + FROM current_state_events + INNER JOIN event_json USING (event_id) + WHERE + room_id = ? + AND {event_type_and_state_key_in_list_clause} + """, + [room_id] + event_type_and_state_key_args, + ) + + # Parse the raw event JSON + sliding_sync_non_joined_rooms_insert_map: Dict[ + str, Optional[Union[str, bool]] + ] = {} + for row in txn: + event_id, event_type, state_key, json = row + event_json = db_to_json(json) + + if event_type == EventTypes.Create: + room_type = event_json.get("content", {}).get( + EventContentFields.ROOM_TYPE + ) + sliding_sync_non_joined_rooms_insert_map["room_type"] = room_type + elif event_type == EventTypes.RoomEncryption: + is_encrypted = event_json.get("content", {}).get( + EventContentFields.ENCRYPTION_ALGORITHM + ) + sliding_sync_non_joined_rooms_insert_map["is_encrypted"] = ( + is_encrypted + ) + elif event_type == EventTypes.Name: + room_name = event_json.get("content", {}).get( + EventContentFields.ROOM_NAME + ) + sliding_sync_non_joined_rooms_insert_map["room_name"] = room_name + else: + raise AssertionError( + f"Unexpected event (we should not be fetching extra events): ({event_type}, {state_key})" + ) + + # Update the `sliding_sync_non_join_memberships` table + insert_keys = sliding_sync_non_joined_rooms_insert_map.keys() + insert_values = sliding_sync_non_joined_rooms_insert_map.values() + # We `DO NOTHING` on conflict because if the row is already in the database, + # we just assume that it was already processed (values should be the same anyways). + txn.execute_batch( + f""" + INSERT INTO sliding_sync_non_join_memberships + (room_id, membership_event_id, user_id, membership, event_stream_ordering, {", ".join(insert_keys)}) + VALUES ( + ?, ?, ?, + (SELECT membership FROM room_memberships WHERE event_id = ?), + (SELECT stream_ordering FROM events WHERE event_id = ?) + {", ".join("?" for _ in insert_values)} + ) + ON CONFLICT (room_id) + DO NOTHING + """, + [ + ( + room_id, + membership_event_id, + state_key[1], + membership_event_id, + membership_event_id, + ) + for membership_event_id in membership_event_ids + ], + ) txn.call_after( self.store._curr_state_delta_stream_cache.entity_has_changed, diff --git a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql index be1b27fdfd..efbc5ebd1e 100644 --- a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql +++ b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql @@ -11,26 +11,42 @@ -- See the GNU Affero General Public License for more details: -- . --- Kept in sync with `current_state_events` +-- We store the join memberships in a separate table from +-- `sliding_sync_non_join_memberships` because the information can be shared across +-- everyone who is joined. +-- +-- This table is kept in sync with `current_state_events` CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms( FOREIGN KEY(room_id) REFERENCES rooms(room_id), - room_type TEXT, - room_name TEXT, - is_encrypted BOOLEAN, - stream_ordering: BIGINT, + -- The `stream_ordering` of the latest event in the room + event_stream_ordering BIGINT REFERENCES events(stream_ordering) + -- The `stream_ordering` of the last event according to the `bump_event_types` bump_stamp: BIGINT, + -- `m.room.create` -> `content.type` + room_type TEXT, + -- `m.room.name` -> `content.name` + room_name TEXT, + -- `m.room.encryption` -> `content.algorithm` + is_encrypted BOOLEAN, PRIMARY KEY (room_id) ); +-- We don't include `bump_stamp` here because we can just use the `stream_ordering` from +-- the membership event itself as the `bump_stamp`. CREATE TABLE IF NOT EXISTS sliding_sync_non_join_memberships( FOREIGN KEY(room_id) REFERENCES rooms(room_id), FOREIGN KEY(membership_event_id) REFERENCES events(event_id), + user_id TEXT NOT NULL, + membership TEXT NOT NULL, + -- `stream_ordering` of the `membership_event_id` + event_stream_ordering BIGINT REFERENCES events(stream_ordering) + -- `m.room.create` -> `content.type` room_type TEXT, + -- `m.room.name` -> `content.name` room_name TEXT, + -- `m.room.encryption` -> `content.algorithm` is_encrypted BOOLEAN, - stream_ordering: BIGINT, - bump_stamp: BIGINT, PRIMARY KEY (room_id, membership_event_id) );