Generate historic pagination token for `/messages` when no `?from` token provided (#12370)

2 files changed, 10 insertions, 3 deletions

diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py
index 876b879483..7ee3340373 100644
--- a/synapse/handlers/pagination.py
+++ b/synapse/handlers/pagination.py
@@ -441,7 +441,14 @@ class PaginationHandler:
         if pagin_config.from_token:
             from_token = pagin_config.from_token
         else:
-            from_token = self.hs.get_event_sources().get_current_token_for_pagination()
+            from_token = (
+                await self.hs.get_event_sources().get_current_token_for_pagination(
+                    room_id
+                )
+            )
+            # We expect `/messages` to use historic pagination tokens by default but
+            # `/messages` should still works with live tokens when manually provided.
+            assert from_token.room_key.topological
 
         if pagin_config.limit is None:
             # This shouldn't happen as we've set a default limit before this
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index 65d4aea9af..b31f00b517 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -1444,8 +1444,8 @@ class RoomEventSource(EventSource[RoomStreamToken, EventBase]):
     def get_current_key(self) -> RoomStreamToken:
         return self.store.get_room_max_token()
 
-    def get_current_key_for_room(self, room_id: str) -> Awaitable[str]:
-        return self.store.get_room_events_max_id(room_id)
+    def get_current_key_for_room(self, room_id: str) -> Awaitable[RoomStreamToken]:
+        return self.store.get_current_room_stream_token_for_room_id(room_id)
 
 
 class ShutdownRoomResponse(TypedDict):