MSC4140: support getting a single delayed event

changelog.d/19926.feature

@@ -0,0 +1 @@
+[MSC4140: Cancellable delayed events](https://github.com/matrix-org/matrix-spec-proposals/pull/4140): Add an endpoint for getting a single delayed event.

synapse/handlers/delayed_events.py

@@ -530,6 +530,19 @@ def _schedule_next_at(self, next_send_ts: Timestamp) -> None:
         else:
             self._next_delayed_event_call.reset(delay_duration.as_secs())
 
+    async def get_for_user(self, requester: Requester, delay_id: str) -> JsonDict:
+        """
+        Return the specified pending delayed event requested by the given user.
+
+        Raises:
+            NotFoundError: if no matching delayed event could be found.
+        """
+        await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)
+        return await self._store.get_delayed_event_for_user(
+            delay_id,
+            requester.user.localpart,
+        )
+
     async def get_all_for_user(self, requester: Requester) -> list[JsonDict]:
         """Return all pending delayed events requested by the given user."""
         await self._delayed_event_mgmt_ratelimiter.ratelimit(requester)

synapse/rest/client/delayed_events.py

@@ -134,6 +134,25 @@ async def on_POST(
         return 200, {}
 
 
+class DelayedEventServlet(RestServlet):
+    PATTERNS = client_patterns(
+        r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)$",
+        releases=(),
+    )
+    CATEGORY = "Delayed event management requests"
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        self.auth = hs.get_auth()
+        self.delayed_events_handler = hs.get_delayed_events_handler()
+
+    async def on_GET(
+        self, request: SynapseRequest, delay_id: str
+    ) -> tuple[int, JsonDict]:
+        requester = await self.auth.get_user_by_req(request)
+        return 200, await self.delayed_events_handler.get_for_user(requester, delay_id)
+
+
 class DelayedEventsServlet(RestServlet):
     PATTERNS = client_patterns(
         r"/org\.matrix\.msc4140/delayed_events$",
@@ -162,4 +181,5 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
         CancelDelayedEventServlet(hs).register(http_server)
         SendDelayedEventServlet(hs).register(http_server)
     RestartDelayedEventServlet(hs).register(http_server)
+    DelayedEventServlet(hs).register(http_server)
     DelayedEventsServlet(hs).register(http_server)

synapse/storage/databases/main/delayed_events.py

@@ -225,6 +225,39 @@ def _get_count_of_delayed_events(txn: LoggingTransaction) -> int:
             _get_count_of_delayed_events,
         )
 
+    async def get_delayed_event_for_user(
+        self,
+        delay_id: str,
+        user_localpart: str,
+    ) -> JsonDict:
+        """
+        Returns the specified pending delayed event owned by the given user.
+
+        Raises:
+            NotFoundError: if there is no matching delayed event.
+        """
+        row = await self.db_pool.simple_select_one(
+            table="delayed_events",
+            keyvalues={
+                "delay_id": delay_id,
+                "user_localpart": user_localpart,
+                "is_processed": False,
+            },
+            retcols=(
+                "room_id",
+                "event_type",
+                "state_key",
+                "delay",
+                "send_ts",
+                "content",
+            ),
+            allow_none=True,
+            desc="get_delayed_event_for_user",
+        )
+        if row is None:
+            raise NotFoundError("Delayed event not found")
+        return _row_to_delayed_event_dict((delay_id, *row))
+
     async def get_all_delayed_events_for_user(
         self,
         user_localpart: str,
@@ -248,18 +281,7 @@ async def get_all_delayed_events_for_user(
             """,
             user_localpart,
         )
-        return [
-            {
-                "delay_id": DelayID(row[0]),
-                "room_id": str(RoomID.from_string(row[1])),
-                "type": EventType(row[2]),
-                **({"state_key": StateKey(row[3])} if row[3] is not None else {}),
-                "delay": Delay(row[4]),
-                "running_since": Timestamp(row[5] - row[4]),
-                "content": db_to_json(row[6]),
-            }
-            for row in rows
-        ]
+        return [_row_to_delayed_event_dict(row) for row in rows]
 
     async def process_timeout_delayed_events(
         self, current_ts: Timestamp, reprocess_events: bool = False
@@ -573,6 +595,18 @@ def _get_next_delayed_event_send_ts_txn(
         return Timestamp(result) if result is not None else None
 
 
+def _row_to_delayed_event_dict(row: tuple) -> JsonDict:
+    return {
+        "delay_id": DelayID(row[0]),
+        "room_id": str(RoomID.from_string(row[1])),
+        "type": EventType(row[2]),
+        **({"state_key": StateKey(row[3])} if row[3] is not None else {}),
+        "delay": Delay(row[4]),
+        "running_since": Timestamp(row[5] - row[4]),
+        "content": db_to_json(row[6]),
+    }
+
+
 def _generate_delay_id() -> DelayID:
     """Generates an opaque string, for use as a delay ID"""
 

tests/rest/client/test_delayed_events.py

@@ -127,6 +127,48 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
     def test_delayed_events_empty_on_startup(self) -> None:
         self.assertListEqual([], self._get_delayed_events())
 
+    def test_delayed_event_lookup(self) -> None:
+        channel = self.make_request(
+            "POST",
+            _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 100000),
+            {},
+            self.user1_access_token,
+        )
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        delay_id = channel.json_body["delay_id"]
+
+        # Test that the scheduled delayed event can be retrieved
+        channel = self.make_request(
+            "GET",
+            f"{PATH_PREFIX}/{delay_id}",
+            access_token=self.user1_access_token,
+        )
+        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+
+        event = channel.json_body
+        self.assertEqual(delay_id, event["delay_id"])
+
+        # Test that the list lookup retrieves the same item
+        events = self._get_delayed_events()
+        self.assertEqual(1, len(events), events)
+        self.assertDictEqual(events[0], event)
+
+        # Test that a non-existent delayed event cannot be found
+        channel = self.make_request(
+            "GET",
+            f"{PATH_PREFIX}/{delay_id}-fake",
+            access_token=self.user1_access_token,
+        )
+        self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result)
+
+        # Test that other users cannot access this delayed event
+        channel = self.make_request(
+            "GET",
+            f"{PATH_PREFIX}/{delay_id}",
+            access_token=self.user2_access_token,
+        )
+        self.assertEqual(HTTPStatus.NOT_FOUND, channel.code, channel.result)
+
     def test_delayed_state_events_are_sent_on_timeout(self) -> None:
         state_key = "to_send_on_timeout"