Allow room creator to send MSC2716 related events in existing room versions

changelog.d/10566.feature

@@ -0,0 +1 @@
+Allow room creators to send historical events specified by [MSC2716](https://github.com/matrix-org/matrix-doc/pull/2716) in existing room versions.

synapse/handlers/federation.py

@@ -934,9 +934,11 @@ async def _handle_marker_event(self, origin: str, marker_event: EventBase):
             return
 
         # Skip processing a marker event if the room version doesn't
-        # support it.
+        # support it or the event is not from the room creator.
         room_version = await self.store.get_room_version(marker_event.room_id)
-        if not room_version.msc2716_historical:
+        create_event = await self.store.get_create_event_for_room(marker_event.room_id)
+        room_creator = create_event.content.get("creator", None)
+        if not room_version.msc2716_historical or marker_event.sender != room_creator:
             return
 
         logger.debug("_handle_marker_event: received %s", marker_event)

synapse/storage/databases/main/events.py

@@ -1770,10 +1770,12 @@ def _handle_insertion_event(self, txn: LoggingTransaction, event: EventBase):
             # Not a insertion event
             return
 
-        # Skip processing a insertion event if the room version doesn't
-        # support it.
+        # Skip processing an insertion event if the room version doesn't
+        # support it or the event is not from the room creator.
         room_version = self.store.get_room_version_txn(txn, event.room_id)
-        if not room_version.msc2716_historical:
+        create_event = self.store.get_create_event_for_room_txn(txn, event.room_id)
+        room_creator = create_event.content.get("creator", None)
+        if not room_version.msc2716_historical or event.sender != room_creator:
             return
 
         next_chunk_id = event.content.get(EventContentFields.MSC2716_NEXT_CHUNK_ID)
@@ -1822,9 +1824,11 @@ def _handle_chunk_event(self, txn: LoggingTransaction, event: EventBase):
             return
 
         # Skip processing a chunk event if the room version doesn't
-        # support it.
+        # support it or the event is not from the room creator.
         room_version = self.store.get_room_version_txn(txn, event.room_id)
-        if not room_version.msc2716_historical:
+        create_event = self.store.get_create_event_for_room_txn(txn, event.room_id)
+        room_creator = create_event.content.get("creator", None)
+        if not room_version.msc2716_historical or event.sender != room_creator:
             return
 
         chunk_id = event.content.get(EventContentFields.MSC2716_CHUNK_ID)

synapse/storage/databases/main/events_worker.py

@@ -55,7 +55,7 @@
 from synapse.replication.tcp.streams import BackfillStream
 from synapse.replication.tcp.streams.events import EventsStream
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
-from synapse.storage.database import DatabasePool
+from synapse.storage.database import DatabasePool, LoggingTransaction
 from synapse.storage.engines import PostgresEngine
 from synapse.storage.util.id_generators import MultiWriterIdGenerator, StreamIdGenerator
 from synapse.storage.util.sequence import build_sequence_generator
@@ -223,6 +223,74 @@ async def get_received_ts(self, event_id: str) -> Optional[int]:
             desc="get_received_ts",
         )
 
+        # Inform mypy that if allow_none is False (the default) then get_event
+
+    # always returns an EventBase.
+    @overload
+    def get_event_txn(
+        self,
+        event_id: str,
+        allow_rejected: bool = False,
+        allow_none: Literal[False] = False,
+    ) -> EventBase:
+        ...
+
+    @overload
+    def get_event_txn(
+        self,
+        event_id: str,
+        allow_rejected: bool = False,
+        allow_none: Literal[True] = False,
+    ) -> Optional[EventBase]:
+        ...
+
+    def get_event_txn(
+        self,
+        txn: LoggingTransaction,
+        event_id: str,
+        allow_rejected: bool = False,
+        allow_none: bool = False,
+    ) -> Optional[EventBase]:
+        """Get an event from the database by event_id.
+        Args:
+            txn: Transaction object
+            event_id: The event_id of the event to fetch
+            get_prev_content: If True and event is a state event,
+                include the previous states content in the unsigned field.
+            allow_rejected: If True, return rejected events. Otherwise,
+                behave as per allow_none.
+            allow_none: If True, return None if no event found, if
+                False throw a NotFoundError
+            check_room_id: if not None, check the room of the found event.
+                If there is a mismatch, behave as per allow_none.
+        Returns:
+            The event, or None if the event was not found and allow_none=True
+        Raises:
+            NotFoundError: if the event_id was not found and allow_none=False
+        """
+        event_map = self._fetch_event_rows(txn, [event_id])
+        event_info = event_map[event_id]
+        if event_info is None and not allow_none:
+            raise NotFoundError("Could not find event %s" % (event_id,))
+
+        rejected_reason = event_info["rejected_reason"]
+        if not allow_rejected and rejected_reason:
+            return
+
+        d = db_to_json(event_info["json"])
+        internal_metadata = db_to_json(event_info["internal_metadata"])
+        room_version_id = event_info["room_version_id"]
+        room_version = KNOWN_ROOM_VERSIONS.get(room_version_id)
+
+        event = make_event_from_dict(
+            event_dict=d,
+            room_version=room_version,
+            internal_metadata_dict=internal_metadata,
+            rejected_reason=rejected_reason,
+        )
+
+        return event
+
     # Inform mypy that if allow_none is False (the default) then get_event
     # always returns an EventBase.
     @overload

synapse/storage/databases/main/state.py

@@ -178,15 +178,34 @@ async def get_create_event_for_room(self, room_id: str) -> EventBase:
         Raises:
             NotFoundError if the room is unknown
         """
-        state_ids = await self.get_current_state_ids(room_id)
+        return await self.db_pool.runInteraction(
+            "get_create_event_for_room_txn",
+            self.get_create_event_for_room_txn,
+            room_id,
+        )
+
+    def get_create_event_for_room_txn(
+        self, txn: LoggingTransaction, room_id: str
+    ) -> EventBase:
+        """Get the create state event for a room.
+        Args:
+            txn: Transaction object
+            room_id: The room ID.
+        Returns:
+            The room creation event.
+        Raises:
+            NotFoundError if the room is unknown
+        """
+
+        state_ids = self.get_current_state_ids_txn(txn, room_id)
         create_id = state_ids.get((EventTypes.Create, ""))
 
         # If we can't find the create event, assume we've hit a dead end
         if not create_id:
             raise NotFoundError("Unknown room %s" % (room_id,))
 
         # Retrieve the room's create event and return
-        create_event = await self.get_event(create_id)
+        create_event = self.get_event_txn(txn, create_id)
         return create_event
 
     @cached(max_entries=100000, iterable=True)
@@ -200,21 +219,35 @@ async def get_current_state_ids(self, room_id: str) -> StateMap[str]:
         Returns:
             The current state of the room.
         """
+        return await self.db_pool.runInteraction(
+            "get_current_state_ids_txn",
+            self.get_current_state_ids_txn,
+            room_id,
+        )
 
-        def _get_current_state_ids_txn(txn):
-            txn.execute(
-                """SELECT type, state_key, event_id FROM current_state_events
-                WHERE room_id = ?
-                """,
-                (room_id,),
-            )
+    def get_current_state_ids_txn(
+        self, txn: LoggingTransaction, room_id: str
+    ) -> StateMap[str]:
+        """Get the current state event ids for a room based on the
+        current_state_events table.
+
+        Args:
+            txn: Transaction object
+            room_id: The room to get the state IDs of.
 
-            return {(intern_string(r[0]), intern_string(r[1])): r[2] for r in txn}
+        Returns:
+            The current state of the room.
+        """
 
-        return await self.db_pool.runInteraction(
-            "get_current_state_ids", _get_current_state_ids_txn
+        txn.execute(
+            """SELECT type, state_key, event_id FROM current_state_events
+            WHERE room_id = ?
+            """,
+            (room_id,),
         )
 
+        return {(intern_string(r[0]), intern_string(r[1])): r[2] for r in txn}
+
     # FIXME: how should this be cached?
     async def get_filtered_current_state_ids(
         self, room_id: str, state_filter: Optional[StateFilter] = None