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/rest/client/room.py

@@ -1070,6 +1070,32 @@ 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()
+
+    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._store.get_event_for_timestamp(
+            room_id, timestamp, direction
+        )
+
+        return 200, {
+            "event_id": event_id,
+        }
+
+
 class RoomSpaceSummaryRestServlet(RestServlet):
     PATTERNS = (
         re.compile(
@@ -1239,6 +1265,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/storage/databases/main/events_worker.py

@@ -34,7 +34,7 @@
 from twisted.internet import defer
 
 from synapse.api.constants import EventTypes
-from synapse.api.errors import NotFoundError, SynapseError
+from synapse.api.errors import Codes, NotFoundError, SynapseError
 from synapse.api.room_versions import (
     KNOWN_ROOM_VERSIONS,
     EventFormatVersions,
@@ -1626,3 +1626,55 @@ def _cleanup_old_transaction_ids_txn(txn):
             "_cleanup_old_transaction_ids",
             _cleanup_old_transaction_ids_txn,
         )
+
+    async def get_event_for_timestamp(
+        self, room_id: str, timestamp: int, direction: str
+    ) -> str:
+        sql_template = """
+            SELECT event_id FROM events
+            WHERE
+                origin_server_ts %s ?
+                AND room_id = ?
+            ORDER BY origin_server_ts %s
+            LIMIT 1;
+        """
+
+        def get_event_for_timestamp_txn(txn) -> str:
+
+            if direction == "b":
+                # Find closest event *before* a given timestamp. We use descending
+                # (which gives values largest to smallest) because we want the
+                # largest possible timestamp *before* the given timestamp.
+                comparison_operator = "<="
+                order = "DESC"
+            else:
+                # Find closest event *after* a given timestamp. We use ascending
+                # (which gives values smallest to largest) because we want the
+                # closest possible timestamp *after* the given timestamp.
+                comparison_operator = ">="
+                order = "ASC"
+
+            txn.execute(
+                sql_template % (comparison_operator, order), (timestamp, room_id)
+            )
+            row = txn.fetchone()
+            logger.info("row %s", row)
+            if row:
+                (event_id,) = row
+                return event_id
+
+            raise SynapseError(
+                404,
+                "Unable to find event from %s in direction %s" % (timestamp, direction),
+                errcode=Codes.NOT_FOUND,
+            )
+
+        if direction not in ("f", "b"):
+            raise SynapseError(
+                400, "Unknown direction: %s" % (direction,), errcode=Codes.INVALID_PARAM
+            )
+
+        return await self.db_pool.runInteraction(
+            "get_event_for_timestamp_txn",
+            get_event_for_timestamp_txn,
+        )