Add endpoint to get an event at a given timestamp - MSC3030

scripts-dev/complement.sh

@@ -65,4 +65,4 @@ if [[ -n "$1" ]]; then
 fi
 
 # Run the tests!
-go test -v -tags synapse_blacklist,msc2946,msc3083,msc2403,msc2716 -count=1 "${EXTRA_COMPLEMENT_ARGS[@]}" ./tests/...
+go test -v -tags synapse_blacklist,msc2946,msc3083,msc2403,msc2716,msc3030 -count=1 "${EXTRA_COMPLEMENT_ARGS[@]}" ./tests/...

synapse/federation/federation_server.py

@@ -200,6 +200,22 @@ async def on_backfill_request(
 
         return 200, res
 
+    async def on_timestamp_to_event_request(
+        self, origin: str, room_id: str, timestamp: int, direction: str
+    ) -> Tuple[int, Dict[str, Any]]:
+        with (await self._server_linearizer.queue((origin, room_id))):
+            origin_host, _ = parse_server_name(origin)
+            await self.check_server_matches_acl(origin_host, room_id)
+
+            # We only try to fetch data from the local database
+            event_id = await self.store.get_event_for_timestamp(
+                room_id, timestamp, direction
+            )
+
+        return 200, {
+            "event_id": event_id,
+        }
+
     async def on_incoming_transaction(
         self,
         origin: str,

synapse/federation/transport/client.py

@@ -148,6 +148,21 @@ async def backfill(
             destination, path=path, args=args, try_trailing_slash_on_400=True
         )
 
+    @log_function
+    async def timestamp_to_event(
+        self, destination: str, room_id: str, timestamp: int, direction: str
+    ) -> Optional[JsonDict]:
+        """
+        TODO
+        """
+        path = _create_v1_path("/timestamp_to_event/%s", room_id)
+
+        args = {"ts": [str(timestamp)], "dir": [direction]}
+
+        return await self.client.get_json(
+            destination, path=path, args=args, try_trailing_slash_on_400=True
+        )
+
     @log_function
     async def send_transaction(
         self,

synapse/federation/transport/server/federation.py

@@ -174,6 +174,26 @@ async def on_GET(
         return await self.handler.on_backfill_request(origin, room_id, versions, limit)
 
 
+class FederationTimestampLookupServlet(BaseFederationServerServlet):
+    PATH = "/timestamp_to_event/(?P<room_id>[^/]*)/?"
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: Dict[bytes, List[bytes]],
+        room_id: str,
+    ) -> Tuple[int, JsonDict]:
+        timestamp = parse_integer_from_args(query, "ts", None)
+        direction = parse_string_from_args(
+            query, "dir", default="f", allowed_values=["f", "b"]
+        )
+
+        return await self.handler.on_timestamp_to_event_request(
+            origin, room_id, timestamp, direction
+        )
+
+
 class FederationQueryServlet(BaseFederationServerServlet):
     PATH = "/query/(?P<query_type>[^/]*)"
 
@@ -680,6 +700,7 @@ async def on_GET(
     FederationStateV1Servlet,
     FederationStateIdsServlet,
     FederationBackfillServlet,
+    FederationTimestampLookupServlet,
     FederationQueryServlet,
     FederationMakeJoinServlet,
     FederationMakeLeaveServlet,

synapse/handlers/federation.py

@@ -68,6 +68,37 @@
 logger = logging.getLogger(__name__)
 
 
+def get_domains_from_state(state: StateMap[EventBase]) -> List[Tuple[str, int]]:
+    """Get joined domains from state
+
+    Args:
+        state: State map from type/state key to event.
+
+    Returns:
+        Returns a list of servers with the lowest depth of their joins.
+            Sorted by lowest depth first.
+    """
+    joined_users = [
+        (state_key, int(event.depth))
+        for (e_type, state_key), event in state.items()
+        if e_type == EventTypes.Member and event.membership == Membership.JOIN
+    ]
+
+    joined_domains: Dict[str, int] = {}
+    for u, d in joined_users:
+        try:
+            dom = get_domain_from_id(u)
+            old_d = joined_domains.get(dom)
+            if old_d:
+                joined_domains[dom] = min(d, old_d)
+            else:
+                joined_domains[dom] = d
+        except Exception:
+            pass
+
+    return sorted(joined_domains.items(), key=lambda d: d[1])
+
+
 class FederationHandler:
     """Handles general incoming federation requests
 
@@ -268,36 +299,6 @@ async def _maybe_backfill_inner(
 
         curr_state = await self.state_handler.get_current_state(room_id)
 
-        def get_domains_from_state(state: StateMap[EventBase]) -> List[Tuple[str, int]]:
-            """Get joined domains from state
-
-            Args:
-                state: State map from type/state key to event.
-
-            Returns:
-                Returns a list of servers with the lowest depth of their joins.
-                 Sorted by lowest depth first.
-            """
-            joined_users = [
-                (state_key, int(event.depth))
-                for (e_type, state_key), event in state.items()
-                if e_type == EventTypes.Member and event.membership == Membership.JOIN
-            ]
-
-            joined_domains: Dict[str, int] = {}
-            for u, d in joined_users:
-                try:
-                    dom = get_domain_from_id(u)
-                    old_d = joined_domains.get(dom)
-                    if old_d:
-                        joined_domains[dom] = min(d, old_d)
-                    else:
-                        joined_domains[dom] = d
-                except Exception:
-                    pass
-
-            return sorted(joined_domains.items(), key=lambda d: d[1])
-
         curr_domains = get_domains_from_state(curr_state)
 
         likely_domains = [

synapse/handlers/room.py

@@ -76,6 +76,7 @@
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.util.stringutils import parse_and_validate_server_name
 from synapse.visibility import filter_events_for_client
+from synapse.handlers.federation import get_domains_from_state
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -1217,6 +1218,80 @@ async def filter_evts(events: List[EventBase]) -> List[EventBase]:
         return results
 
 
+class TimestampLookupHandler:
+    def __init__(self, hs: "HomeServer"):
+        self.server_name = hs.hostname
+        self.store = hs.get_datastore()
+        self.state_handler = hs.get_state_handler()
+        self.transport_layer = hs.get_federation_transport_client()
+
+    async def get_event_for_timestamp(
+        self,
+        requester: Requester,
+        room_id: str,
+        timestamp: int,
+        direction: str,
+    ) -> Optional[JsonDict]:
+        event_id = await self.store.get_event_for_timestamp(
+            room_id, timestamp, direction
+        )
+
+        if not event_id:
+            raise SynapseError(
+                404,
+                "Unable to find event from %s in direction %s" % (timestamp, direction),
+                errcode=Codes.NOT_FOUND,
+            )
+
+        # If we found an extremity, we should probably ask another homeserver
+        # first about more history in between
+        is_extremity = await self.store.check_if_event_is_extremity(room_id, event_id)
+        logger.info(
+            "get_event_for_timestamp: locally, we found event=%s closest to timestamp=%s and is is_extremity=%s",
+            event_id,
+            timestamp,
+            is_extremity,
+        )
+        if is_extremity:
+            logger.info(
+                "get_event_for_timestamp: locally, we found event=%s closest to timestamp=%s is an extremity so we're asking other homeservers first",
+                event_id,
+                timestamp,
+            )
+
+            curr_state = await self.state_handler.get_current_state(room_id)
+            curr_domains = get_domains_from_state(curr_state)
+            likely_domains = [
+                domain for domain, depth in curr_domains if domain != self.server_name
+            ]
+
+            for domain in likely_domains:
+                try:
+                    remote_response = await self.transport_layer.timestamp_to_event(
+                        domain, room_id, timestamp, direction
+                    )
+                    logger.info(
+                        "get_event_for_timestamp: response from domain(%s)=%s",
+                        domain,
+                        remote_response,
+                    )
+                    remote_event_id = remote_response.get("event_id", None)
+                    if remote_event_id:
+                        # TODO: Do we want to persist this as an extremity?
+                        return remote_event_id
+
+                    continue
+                except Exception as e:
+                    logger.exception(
+                        "Failed to fetch /timestamp_to_event from %s because %s",
+                        domain,
+                        e,
+                    )
+                    continue
+
+        return event_id
+
+
 class RoomEventSource(EventSource[RoomStreamToken, EventBase]):
     def __init__(self, hs: "HomeServer"):
         self.store = hs.get_datastore()

synapse/rest/client/room.py

@@ -1070,6 +1070,33 @@ def register_txn_path(
         )
 
 
+class TimestampLookupRestServlet(RestServlet):
+    PATTERNS = client_patterns("/rooms/(?P<room_id>[^/]*)/timestamp_to_event$")
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self._auth = hs.get_auth()
+        self._store = hs.get_datastore()
+        self.timestamp_lookup_handler = hs.get_timestamp_lookup_handler()
+
+    async def on_GET(
+        self, request: SynapseRequest, room_id: str
+    ) -> Tuple[int, JsonDict]:
+        requester = await self._auth.get_user_by_req(request)
+        await self._auth.check_user_in_room(room_id, requester.user.to_string())
+
+        timestamp = parse_integer(request, "ts")
+        direction = parse_string(request, "dir", default="f", allowed_values=["f", "b"])
+
+        event_id = await self.timestamp_lookup_handler.get_event_for_timestamp(
+            requester, room_id, timestamp, direction
+        )
+
+        return 200, {
+            "event_id": event_id,
+        }
+
+
 class RoomSpaceSummaryRestServlet(RestServlet):
     PATTERNS = (
         re.compile(
@@ -1239,6 +1266,7 @@ def register_servlets(
     RoomAliasListServlet(hs).register(http_server)
     SearchRestServlet(hs).register(http_server)
     RoomCreateRestServlet(hs).register(http_server)
+    TimestampLookupRestServlet(hs).register(http_server)
 
     # Some servlets only get registered for the main process.
     if not is_worker:

synapse/server.py

@@ -96,6 +96,7 @@
     RoomContextHandler,
     RoomCreationHandler,
     RoomShutdownHandler,
+    TimestampLookupHandler,
 )
 from synapse.handlers.room_batch import RoomBatchHandler
 from synapse.handlers.room_list import RoomListHandler
@@ -725,6 +726,10 @@ def get_pagination_handler(self) -> PaginationHandler:
     def get_room_context_handler(self) -> RoomContextHandler:
         return RoomContextHandler(self)
 
+    @cache_in_self
+    def get_timestamp_lookup_handler(self) -> TimestampLookupHandler:
+        return TimestampLookupHandler(self)
+
     @cache_in_self
     def get_registration_handler(self) -> RegistrationHandler:
         return RegistrationHandler(self)