Combine duplicated code for calculating an event ID from a txn ID (#16023)

Refactoring related to stabilization of MSC3970, refactor to combine
code which has the same logic.

changelog.d/16023.misc

@@ -0,0 +1 @@
+Combine duplicated code.

synapse/handlers/message.py

@@ -878,6 +878,53 @@ class EventCreationHandler:
                 return prev_event
         return None
 
+    async def get_event_id_from_transaction(
+        self,
+        requester: Requester,
+        txn_id: str,
+        room_id: str,
+    ) -> Optional[str]:
+        """For the given transaction ID and room ID, check if there is a matching event ID.
+
+        Args:
+            requester: The requester making the request in the context of which we want
+                to fetch the event.
+            txn_id: The transaction ID.
+            room_id: The room ID.
+
+        Returns:
+            An event ID if one could be found, None otherwise.
+        """
+        existing_event_id = None
+
+        if self._msc3970_enabled and requester.device_id:
+            # When MSC3970 is enabled, we lookup for events sent by the same device first,
+            # and fallback to the old behaviour if none were found.
+            existing_event_id = (
+                await self.store.get_event_id_from_transaction_id_and_device_id(
+                    room_id,
+                    requester.user.to_string(),
+                    requester.device_id,
+                    txn_id,
+                )
+            )
+            if existing_event_id:
+                return existing_event_id
+
+        # Pre-MSC3970, we looked up for events that were sent by the same session by
+        # using the access token ID.
+        if requester.access_token_id:
+            existing_event_id = (
+                await self.store.get_event_id_from_transaction_id_and_token_id(
+                    room_id,
+                    requester.user.to_string(),
+                    requester.access_token_id,
+                    txn_id,
+                )
+            )
+
+        return existing_event_id
+
     async def get_event_from_transaction(
         self,
         requester: Requester,
@@ -896,35 +943,11 @@ class EventCreationHandler:
         Returns:
             An event if one could be found, None otherwise.
         """
-
+        existing_event_id = await self.get_event_id_from_transaction(
-        if self._msc3970_enabled and requester.device_id:
+            requester, txn_id, room_id
-            # When MSC3970 is enabled, we lookup for events sent by the same device first,
+        )
-            # and fallback to the old behaviour if none were found.
+        if existing_event_id:
-            existing_event_id = (
+            return await self.store.get_event(existing_event_id)
-                await self.store.get_event_id_from_transaction_id_and_device_id(
-                    room_id,
-                    requester.user.to_string(),
-                    requester.device_id,
-                    txn_id,
-                )
-            )
-            if existing_event_id:
-                return await self.store.get_event(existing_event_id)
-
-        # Pre-MSC3970, we looked up for events that were sent by the same session by
-        # using the access token ID.
-        if requester.access_token_id:
-            existing_event_id = (
-                await self.store.get_event_id_from_transaction_id_and_token_id(
-                    room_id,
-                    requester.user.to_string(),
-                    requester.access_token_id,
-                    txn_id,
-                )
-            )
-            if existing_event_id:
-                return await self.store.get_event(existing_event_id)
-
         return None
 
     async def create_and_send_nonmember_event(

synapse/handlers/room_member.py

@@ -176,8 +176,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
         self.request_ratelimiter = hs.get_request_ratelimiter()
         hs.get_notifier().add_new_join_in_room_callback(self._on_user_joined_room)
 
-        self._msc3970_enabled = hs.config.experimental.msc3970_enabled
-
     def _on_user_joined_room(self, event_id: str, room_id: str) -> None:
         """Notify the rate limiter that a room join has occurred.
 
@@ -418,29 +416,11 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
         # do this check just before we persist an event as well, but may as well
         # do it up front for efficiency.)
         if txn_id:
-            existing_event_id = None
+            existing_event_id = (
-            if self._msc3970_enabled and requester.device_id:
+                await self.event_creation_handler.get_event_id_from_transaction(
-                # When MSC3970 is enabled, we lookup for events sent by the same device
+                    requester, txn_id, room_id
-                # first, and fallback to the old behaviour if none were found.
-                existing_event_id = (
-                    await self.store.get_event_id_from_transaction_id_and_device_id(
-                        room_id,
-                        requester.user.to_string(),
-                        requester.device_id,
-                        txn_id,
-                    )
                 )
-
+            )
-            if requester.access_token_id and not existing_event_id:
-                existing_event_id = (
-                    await self.store.get_event_id_from_transaction_id_and_token_id(
-                        room_id,
-                        requester.user.to_string(),
-                        requester.access_token_id,
-                        txn_id,
-                    )
-                )
-
             if existing_event_id:
                 event_pos = await self.store.get_position_for_event(existing_event_id)
                 return existing_event_id, event_pos.stream