We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
500s calling /messages on matrix.org
as seen by @schiessle on riot-android at https://github.com/vector-im/riot-android/issues/2802

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `synapse/handlers/pagination.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright 2014 - 2016 OpenMarket Ltd
3 # Copyright 2017 - 2018 New Vector Ltd
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 #     http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 import logging
17 
18 from twisted.internet import defer
19 from twisted.python.failure import Failure
20 
21 from synapse.api.constants import EventTypes, Membership
22 from synapse.api.errors import SynapseError
23 from synapse.events.utils import serialize_event
24 from synapse.storage.state import StateFilter
25 from synapse.types import RoomStreamToken
26 from synapse.util.async_helpers import ReadWriteLock
27 from synapse.util.logcontext import run_in_background
28 from synapse.util.stringutils import random_string
29 from synapse.visibility import filter_events_for_client
30 
31 logger = logging.getLogger(__name__)
32 
33 
34 class PurgeStatus(object):
35     """Object tracking the status of a purge request
36 
37     This class contains information on the progress of a purge request, for
38     return by get_purge_status.
39 
40     Attributes:
41         status (int): Tracks whether this request has completed. One of
42             STATUS_{ACTIVE,COMPLETE,FAILED}
43     """
44 
45     STATUS_ACTIVE = 0
46     STATUS_COMPLETE = 1
47     STATUS_FAILED = 2
48 
49     STATUS_TEXT = {
50         STATUS_ACTIVE: "active",
51         STATUS_COMPLETE: "complete",
52         STATUS_FAILED: "failed",
53     }
54 
55     def __init__(self):
56         self.status = PurgeStatus.STATUS_ACTIVE
57 
58     def asdict(self):
59         return {
60             "status": PurgeStatus.STATUS_TEXT[self.status]
61         }
62 
63 
64 class PaginationHandler(object):
65     """Handles pagination and purge history requests.
66 
67     These are in the same handler due to the fact we need to block clients
68     paginating during a purge.
69     """
70 
71     def __init__(self, hs):
72         self.hs = hs
73         self.auth = hs.get_auth()
74         self.store = hs.get_datastore()
75         self.clock = hs.get_clock()
76 
77         self.pagination_lock = ReadWriteLock()
78         self._purges_in_progress_by_room = set()
79         # map from purge id to PurgeStatus
80         self._purges_by_id = {}
81 
82     def start_purge_history(self, room_id, token,
83                             delete_local_events=False):
84         """Start off a history purge on a room.
85 
86         Args:
87             room_id (str): The room to purge from
88 
89             token (str): topological token to delete events before
90             delete_local_events (bool): True to delete local events as well as
91                 remote ones
92 
93         Returns:
94             str: unique ID for this purge transaction.
95         """
96         if room_id in self._purges_in_progress_by_room:
97             raise SynapseError(
98                 400,
99                 "History purge already in progress for %s" % (room_id, ),
100             )
101 
102         purge_id = random_string(16)
103 
104         # we log the purge_id here so that it can be tied back to the
105         # request id in the log lines.
106         logger.info("[purge] starting purge_id %s", purge_id)
107 
108         self._purges_by_id[purge_id] = PurgeStatus()
109         run_in_background(
110             self._purge_history,
111             purge_id, room_id, token, delete_local_events,
112         )
113         return purge_id
114 
115     @defer.inlineCallbacks
116     def _purge_history(self, purge_id, room_id, token,
117                        delete_local_events):
118         """Carry out a history purge on a room.
119 
120         Args:
121             purge_id (str): The id for this purge
122             room_id (str): The room to purge from
123             token (str): topological token to delete events before
124             delete_local_events (bool): True to delete local events as well as
125                 remote ones
126 
127         Returns:
128             Deferred
129         """
130         self._purges_in_progress_by_room.add(room_id)
131         try:
132             with (yield self.pagination_lock.write(room_id)):
133                 yield self.store.purge_history(
134                     room_id, token, delete_local_events,
135                 )
136             logger.info("[purge] complete")
137             self._purges_by_id[purge_id].status = PurgeStatus.STATUS_COMPLETE
138         except Exception:
139             logger.error("[purge] failed: %s", Failure().getTraceback().rstrip())
140             self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED
141         finally:
142             self._purges_in_progress_by_room.discard(room_id)
143 
144             # remove the purge from the list 24 hours after it completes
145             def clear_purge():
146                 del self._purges_by_id[purge_id]
147             self.hs.get_reactor().callLater(24 * 3600, clear_purge)
148 
149     def get_purge_status(self, purge_id):
150         """Get the current status of an active purge
151 
152         Args:
153             purge_id (str): purge_id returned by start_purge_history
154 
155         Returns:
156             PurgeStatus|None
157         """
158         return self._purges_by_id.get(purge_id)
159 
160     @defer.inlineCallbacks
161     def get_messages(self, requester, room_id=None, pagin_config=None,
162                      as_client_event=True, event_filter=None):
163         """Get messages in a room.
164 
165         Args:
166             requester (Requester): The user requesting messages.
167             room_id (str): The room they want messages from.
168             pagin_config (synapse.api.streams.PaginationConfig): The pagination
169                 config rules to apply, if any.
170             as_client_event (bool): True to get events in client-server format.
171             event_filter (Filter): Filter to apply to results or None
172         Returns:
173             dict: Pagination API results
174         """
175         user_id = requester.user.to_string()
176 
177         if pagin_config.from_token:
178             room_token = pagin_config.from_token.room_key
179         else:
180             pagin_config.from_token = (
181                 yield self.hs.get_event_sources().get_current_token_for_room(
182                     room_id=room_id
183                 )
184             )
185             room_token = pagin_config.from_token.room_key
186 
187         room_token = RoomStreamToken.parse(room_token)
188 
189         pagin_config.from_token = pagin_config.from_token.copy_and_replace(
190             "room_key", str(room_token)
191         )
192 
193         source_config = pagin_config.get_source_config("room")
194 
195         with (yield self.pagination_lock.read(room_id)):
196             membership, member_event_id = yield self.auth.check_in_room_or_world_readable(
197                 room_id, user_id
198             )
199 
200             if source_config.direction == 'b':
201                 # if we're going backwards, we might need to backfill. This
202                 # requires that we have a topo token.
203                 if room_token.topological:
204                     max_topo = room_token.topological
205                 else:
206                     max_topo = yield self.store.get_max_topological_token(
207                         room_id, room_token.stream
208                     )
209 
210                 if membership == Membership.LEAVE:
211                     # If they have left the room then clamp the token to be before
212                     # they left the room, to save the effort of loading from the
213                     # database.
214                     leave_token = yield self.store.get_topological_token_for_event(
215                         member_event_id
216                     )
217                     leave_token = RoomStreamToken.parse(leave_token)
218                     if leave_token.topological < max_topo:
219                         source_config.from_key = str(leave_token)
220 
221                 yield self.hs.get_handlers().federation_handler.maybe_backfill(
222                     room_id, max_topo
223                 )
224 
225             events, next_key = yield self.store.paginate_room_events(
226                 room_id=room_id,
227                 from_key=source_config.from_key,
228                 to_key=source_config.to_key,
229                 direction=source_config.direction,
230                 limit=source_config.limit,
231                 event_filter=event_filter,
232             )
233 
234             next_token = pagin_config.from_token.copy_and_replace(
235                 "room_key", next_key
236             )
237 
238         if not events:
239             defer.returnValue({
240                 "chunk": [],
241                 "start": pagin_config.from_token.to_string(),
242                 "end": next_token.to_string(),
243             })
244 
245         if event_filter:
246             events = event_filter.filter(events)
247 
248         events = yield filter_events_for_client(
249             self.store,
250             user_id,
251             events,
252             is_peeking=(member_event_id is None),
253         )
254 
255         state = None
256         if event_filter and event_filter.lazy_load_members():
257             # TODO: remove redundant members
258 
259             # FIXME: we also care about invite targets etc.
260             state_filter = StateFilter.from_types(
261                 (EventTypes.Member, event.sender)
262                 for event in events
263             )
264 
265             state_ids = yield self.store.get_state_ids_for_event(
266                 events[0].event_id, state_filter=state_filter,
267             )
268 
269             if state_ids:
270                 state = yield self.store.get_events(list(state_ids.values()))
271                 state = state.values()
272 
273         time_now = self.clock.time_msec()
274 
275         chunk = {
276             "chunk": [
277                 serialize_event(e, time_now, as_client_event)
278                 for e in events
279             ],
280             "start": pagin_config.from_token.to_string(),
281             "end": next_token.to_string(),
282         }
283 
284         if state:
285             chunk["state"] = [
286                 serialize_event(e, time_now, as_client_event)
287                 for e in state
288             ]
289 
290         defer.returnValue(chunk)
291
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py
--- a/synapse/handlers/pagination.py
+++ b/synapse/handlers/pagination.py
@@ -235,6 +235,17 @@
                 "room_key", next_key
             )
 
+        if events:
+            if event_filter:
+                events = event_filter.filter(events)
+
+            events = yield filter_events_for_client(
+                self.store,
+                user_id,
+                events,
+                is_peeking=(member_event_id is None),
+            )
+
         if not events:
             defer.returnValue({
                 "chunk": [],
@@ -242,16 +253,6 @@
                 "end": next_token.to_string(),
             })
 
-        if event_filter:
-            events = event_filter.filter(events)
-
-        events = yield filter_events_for_client(
-            self.store,
-            user_id,
-            events,
-            is_peeking=(member_event_id is None),
-        )
-
         state = None
         if event_filter and event_filter.lazy_load_members():
             # TODO: remove redundant members

{"golden_diff": "diff --git a/synapse/handlers/pagination.py b/synapse/handlers/pagination.py\n--- a/synapse/handlers/pagination.py\n+++ b/synapse/handlers/pagination.py\n@@ -235,6 +235,17 @@\n                 \"room_key\", next_key\n             )\n \n+        if events:\n+            if event_filter:\n+                events = event_filter.filter(events)\n+\n+            events = yield filter_events_for_client(\n+                self.store,\n+                user_id,\n+                events,\n+                is_peeking=(member_event_id is None),\n+            )\n+\n         if not events:\n             defer.returnValue({\n                 \"chunk\": [],\n@@ -242,16 +253,6 @@\n                 \"end\": next_token.to_string(),\n             })\n \n-        if event_filter:\n-            events = event_filter.filter(events)\n-\n-        events = yield filter_events_for_client(\n-            self.store,\n-            user_id,\n-            events,\n-            is_peeking=(member_event_id is None),\n-        )\n-\n         state = None\n         if event_filter and event_filter.lazy_load_members():\n             # TODO: remove redundant members\n", "issue": "500s calling /messages on matrix.org\nas seen by @schiessle on riot-android at https://github.com/vector-im/riot-android/issues/2802\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2014 - 2016 OpenMarket Ltd\n# Copyright 2017 - 2018 New Vector Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport logging\n\nfrom twisted.internet import defer\nfrom twisted.python.failure import Failure\n\nfrom synapse.api.constants import EventTypes, Membership\nfrom synapse.api.errors import SynapseError\nfrom synapse.events.utils import serialize_event\nfrom synapse.storage.state import StateFilter\nfrom synapse.types import RoomStreamToken\nfrom synapse.util.async_helpers import ReadWriteLock\nfrom synapse.util.logcontext import run_in_background\nfrom synapse.util.stringutils import random_string\nfrom synapse.visibility import filter_events_for_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass PurgeStatus(object):\n    \"\"\"Object tracking the status of a purge request\n\n    This class contains information on the progress of a purge request, for\n    return by get_purge_status.\n\n    Attributes:\n        status (int): Tracks whether this request has completed. One of\n            STATUS_{ACTIVE,COMPLETE,FAILED}\n    \"\"\"\n\n    STATUS_ACTIVE = 0\n    STATUS_COMPLETE = 1\n    STATUS_FAILED = 2\n\n    STATUS_TEXT = {\n        STATUS_ACTIVE: \"active\",\n        STATUS_COMPLETE: \"complete\",\n        STATUS_FAILED: \"failed\",\n    }\n\n    def __init__(self):\n        self.status = PurgeStatus.STATUS_ACTIVE\n\n    def asdict(self):\n        return {\n            \"status\": PurgeStatus.STATUS_TEXT[self.status]\n        }\n\n\nclass PaginationHandler(object):\n    \"\"\"Handles pagination and purge history requests.\n\n    These are in the same handler due to the fact we need to block clients\n    paginating during a purge.\n    \"\"\"\n\n    def __init__(self, hs):\n        self.hs = hs\n        self.auth = hs.get_auth()\n        self.store = hs.get_datastore()\n        self.clock = hs.get_clock()\n\n        self.pagination_lock = ReadWriteLock()\n        self._purges_in_progress_by_room = set()\n        # map from purge id to PurgeStatus\n        self._purges_by_id = {}\n\n    def start_purge_history(self, room_id, token,\n                            delete_local_events=False):\n        \"\"\"Start off a history purge on a room.\n\n        Args:\n            room_id (str): The room to purge from\n\n            token (str): topological token to delete events before\n            delete_local_events (bool): True to delete local events as well as\n                remote ones\n\n        Returns:\n            str: unique ID for this purge transaction.\n        \"\"\"\n        if room_id in self._purges_in_progress_by_room:\n            raise SynapseError(\n                400,\n                \"History purge already in progress for %s\" % (room_id, ),\n            )\n\n        purge_id = random_string(16)\n\n        # we log the purge_id here so that it can be tied back to the\n        # request id in the log lines.\n        logger.info(\"[purge] starting purge_id %s\", purge_id)\n\n        self._purges_by_id[purge_id] = PurgeStatus()\n        run_in_background(\n            self._purge_history,\n            purge_id, room_id, token, delete_local_events,\n        )\n        return purge_id\n\n    @defer.inlineCallbacks\n    def _purge_history(self, purge_id, room_id, token,\n                       delete_local_events):\n        \"\"\"Carry out a history purge on a room.\n\n        Args:\n            purge_id (str): The id for this purge\n            room_id (str): The room to purge from\n            token (str): topological token to delete events before\n            delete_local_events (bool): True to delete local events as well as\n                remote ones\n\n        Returns:\n            Deferred\n        \"\"\"\n        self._purges_in_progress_by_room.add(room_id)\n        try:\n            with (yield self.pagination_lock.write(room_id)):\n                yield self.store.purge_history(\n                    room_id, token, delete_local_events,\n                )\n            logger.info(\"[purge] complete\")\n            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_COMPLETE\n        except Exception:\n            logger.error(\"[purge] failed: %s\", Failure().getTraceback().rstrip())\n            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED\n        finally:\n            self._purges_in_progress_by_room.discard(room_id)\n\n            # remove the purge from the list 24 hours after it completes\n            def clear_purge():\n                del self._purges_by_id[purge_id]\n            self.hs.get_reactor().callLater(24 * 3600, clear_purge)\n\n    def get_purge_status(self, purge_id):\n        \"\"\"Get the current status of an active purge\n\n        Args:\n            purge_id (str): purge_id returned by start_purge_history\n\n        Returns:\n            PurgeStatus|None\n        \"\"\"\n        return self._purges_by_id.get(purge_id)\n\n    @defer.inlineCallbacks\n    def get_messages(self, requester, room_id=None, pagin_config=None,\n                     as_client_event=True, event_filter=None):\n        \"\"\"Get messages in a room.\n\n        Args:\n            requester (Requester): The user requesting messages.\n            room_id (str): The room they want messages from.\n            pagin_config (synapse.api.streams.PaginationConfig): The pagination\n                config rules to apply, if any.\n            as_client_event (bool): True to get events in client-server format.\n            event_filter (Filter): Filter to apply to results or None\n        Returns:\n            dict: Pagination API results\n        \"\"\"\n        user_id = requester.user.to_string()\n\n        if pagin_config.from_token:\n            room_token = pagin_config.from_token.room_key\n        else:\n            pagin_config.from_token = (\n                yield self.hs.get_event_sources().get_current_token_for_room(\n                    room_id=room_id\n                )\n            )\n            room_token = pagin_config.from_token.room_key\n\n        room_token = RoomStreamToken.parse(room_token)\n\n        pagin_config.from_token = pagin_config.from_token.copy_and_replace(\n            \"room_key\", str(room_token)\n        )\n\n        source_config = pagin_config.get_source_config(\"room\")\n\n        with (yield self.pagination_lock.read(room_id)):\n            membership, member_event_id = yield self.auth.check_in_room_or_world_readable(\n                room_id, user_id\n            )\n\n            if source_config.direction == 'b':\n                # if we're going backwards, we might need to backfill. This\n                # requires that we have a topo token.\n                if room_token.topological:\n                    max_topo = room_token.topological\n                else:\n                    max_topo = yield self.store.get_max_topological_token(\n                        room_id, room_token.stream\n                    )\n\n                if membership == Membership.LEAVE:\n                    # If they have left the room then clamp the token to be before\n                    # they left the room, to save the effort of loading from the\n                    # database.\n                    leave_token = yield self.store.get_topological_token_for_event(\n                        member_event_id\n                    )\n                    leave_token = RoomStreamToken.parse(leave_token)\n                    if leave_token.topological < max_topo:\n                        source_config.from_key = str(leave_token)\n\n                yield self.hs.get_handlers().federation_handler.maybe_backfill(\n                    room_id, max_topo\n                )\n\n            events, next_key = yield self.store.paginate_room_events(\n                room_id=room_id,\n                from_key=source_config.from_key,\n                to_key=source_config.to_key,\n                direction=source_config.direction,\n                limit=source_config.limit,\n                event_filter=event_filter,\n            )\n\n            next_token = pagin_config.from_token.copy_and_replace(\n                \"room_key\", next_key\n            )\n\n        if not events:\n            defer.returnValue({\n                \"chunk\": [],\n                \"start\": pagin_config.from_token.to_string(),\n                \"end\": next_token.to_string(),\n            })\n\n        if event_filter:\n            events = event_filter.filter(events)\n\n        events = yield filter_events_for_client(\n            self.store,\n            user_id,\n            events,\n            is_peeking=(member_event_id is None),\n        )\n\n        state = None\n        if event_filter and event_filter.lazy_load_members():\n            # TODO: remove redundant members\n\n            # FIXME: we also care about invite targets etc.\n            state_filter = StateFilter.from_types(\n                (EventTypes.Member, event.sender)\n                for event in events\n            )\n\n            state_ids = yield self.store.get_state_ids_for_event(\n                events[0].event_id, state_filter=state_filter,\n            )\n\n            if state_ids:\n                state = yield self.store.get_events(list(state_ids.values()))\n                state = state.values()\n\n        time_now = self.clock.time_msec()\n\n        chunk = {\n            \"chunk\": [\n                serialize_event(e, time_now, as_client_event)\n                for e in events\n            ],\n            \"start\": pagin_config.from_token.to_string(),\n            \"end\": next_token.to_string(),\n        }\n\n        if state:\n            chunk[\"state\"] = [\n                serialize_event(e, time_now, as_client_event)\n                for e in state\n            ]\n\n        defer.returnValue(chunk)\n", "path": "synapse/handlers/pagination.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright 2014 - 2016 OpenMarket Ltd\n# Copyright 2017 - 2018 New Vector Ltd\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport logging\n\nfrom twisted.internet import defer\nfrom twisted.python.failure import Failure\n\nfrom synapse.api.constants import EventTypes, Membership\nfrom synapse.api.errors import SynapseError\nfrom synapse.events.utils import serialize_event\nfrom synapse.storage.state import StateFilter\nfrom synapse.types import RoomStreamToken\nfrom synapse.util.async_helpers import ReadWriteLock\nfrom synapse.util.logcontext import run_in_background\nfrom synapse.util.stringutils import random_string\nfrom synapse.visibility import filter_events_for_client\n\nlogger = logging.getLogger(__name__)\n\n\nclass PurgeStatus(object):\n    \"\"\"Object tracking the status of a purge request\n\n    This class contains information on the progress of a purge request, for\n    return by get_purge_status.\n\n    Attributes:\n        status (int): Tracks whether this request has completed. One of\n            STATUS_{ACTIVE,COMPLETE,FAILED}\n    \"\"\"\n\n    STATUS_ACTIVE = 0\n    STATUS_COMPLETE = 1\n    STATUS_FAILED = 2\n\n    STATUS_TEXT = {\n        STATUS_ACTIVE: \"active\",\n        STATUS_COMPLETE: \"complete\",\n        STATUS_FAILED: \"failed\",\n    }\n\n    def __init__(self):\n        self.status = PurgeStatus.STATUS_ACTIVE\n\n    def asdict(self):\n        return {\n            \"status\": PurgeStatus.STATUS_TEXT[self.status]\n        }\n\n\nclass PaginationHandler(object):\n    \"\"\"Handles pagination and purge history requests.\n\n    These are in the same handler due to the fact we need to block clients\n    paginating during a purge.\n    \"\"\"\n\n    def __init__(self, hs):\n        self.hs = hs\n        self.auth = hs.get_auth()\n        self.store = hs.get_datastore()\n        self.clock = hs.get_clock()\n\n        self.pagination_lock = ReadWriteLock()\n        self._purges_in_progress_by_room = set()\n        # map from purge id to PurgeStatus\n        self._purges_by_id = {}\n\n    def start_purge_history(self, room_id, token,\n                            delete_local_events=False):\n        \"\"\"Start off a history purge on a room.\n\n        Args:\n            room_id (str): The room to purge from\n\n            token (str): topological token to delete events before\n            delete_local_events (bool): True to delete local events as well as\n                remote ones\n\n        Returns:\n            str: unique ID for this purge transaction.\n        \"\"\"\n        if room_id in self._purges_in_progress_by_room:\n            raise SynapseError(\n                400,\n                \"History purge already in progress for %s\" % (room_id, ),\n            )\n\n        purge_id = random_string(16)\n\n        # we log the purge_id here so that it can be tied back to the\n        # request id in the log lines.\n        logger.info(\"[purge] starting purge_id %s\", purge_id)\n\n        self._purges_by_id[purge_id] = PurgeStatus()\n        run_in_background(\n            self._purge_history,\n            purge_id, room_id, token, delete_local_events,\n        )\n        return purge_id\n\n    @defer.inlineCallbacks\n    def _purge_history(self, purge_id, room_id, token,\n                       delete_local_events):\n        \"\"\"Carry out a history purge on a room.\n\n        Args:\n            purge_id (str): The id for this purge\n            room_id (str): The room to purge from\n            token (str): topological token to delete events before\n            delete_local_events (bool): True to delete local events as well as\n                remote ones\n\n        Returns:\n            Deferred\n        \"\"\"\n        self._purges_in_progress_by_room.add(room_id)\n        try:\n            with (yield self.pagination_lock.write(room_id)):\n                yield self.store.purge_history(\n                    room_id, token, delete_local_events,\n                )\n            logger.info(\"[purge] complete\")\n            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_COMPLETE\n        except Exception:\n            logger.error(\"[purge] failed: %s\", Failure().getTraceback().rstrip())\n            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED\n        finally:\n            self._purges_in_progress_by_room.discard(room_id)\n\n            # remove the purge from the list 24 hours after it completes\n            def clear_purge():\n                del self._purges_by_id[purge_id]\n            self.hs.get_reactor().callLater(24 * 3600, clear_purge)\n\n    def get_purge_status(self, purge_id):\n        \"\"\"Get the current status of an active purge\n\n        Args:\n            purge_id (str): purge_id returned by start_purge_history\n\n        Returns:\n            PurgeStatus|None\n        \"\"\"\n        return self._purges_by_id.get(purge_id)\n\n    @defer.inlineCallbacks\n    def get_messages(self, requester, room_id=None, pagin_config=None,\n                     as_client_event=True, event_filter=None):\n        \"\"\"Get messages in a room.\n\n        Args:\n            requester (Requester): The user requesting messages.\n            room_id (str): The room they want messages from.\n            pagin_config (synapse.api.streams.PaginationConfig): The pagination\n                config rules to apply, if any.\n            as_client_event (bool): True to get events in client-server format.\n            event_filter (Filter): Filter to apply to results or None\n        Returns:\n            dict: Pagination API results\n        \"\"\"\n        user_id = requester.user.to_string()\n\n        if pagin_config.from_token:\n            room_token = pagin_config.from_token.room_key\n        else:\n            pagin_config.from_token = (\n                yield self.hs.get_event_sources().get_current_token_for_room(\n                    room_id=room_id\n                )\n            )\n            room_token = pagin_config.from_token.room_key\n\n        room_token = RoomStreamToken.parse(room_token)\n\n        pagin_config.from_token = pagin_config.from_token.copy_and_replace(\n            \"room_key\", str(room_token)\n        )\n\n        source_config = pagin_config.get_source_config(\"room\")\n\n        with (yield self.pagination_lock.read(room_id)):\n            membership, member_event_id = yield self.auth.check_in_room_or_world_readable(\n                room_id, user_id\n            )\n\n            if source_config.direction == 'b':\n                # if we're going backwards, we might need to backfill. This\n                # requires that we have a topo token.\n                if room_token.topological:\n                    max_topo = room_token.topological\n                else:\n                    max_topo = yield self.store.get_max_topological_token(\n                        room_id, room_token.stream\n                    )\n\n                if membership == Membership.LEAVE:\n                    # If they have left the room then clamp the token to be before\n                    # they left the room, to save the effort of loading from the\n                    # database.\n                    leave_token = yield self.store.get_topological_token_for_event(\n                        member_event_id\n                    )\n                    leave_token = RoomStreamToken.parse(leave_token)\n                    if leave_token.topological < max_topo:\n                        source_config.from_key = str(leave_token)\n\n                yield self.hs.get_handlers().federation_handler.maybe_backfill(\n                    room_id, max_topo\n                )\n\n            events, next_key = yield self.store.paginate_room_events(\n                room_id=room_id,\n                from_key=source_config.from_key,\n                to_key=source_config.to_key,\n                direction=source_config.direction,\n                limit=source_config.limit,\n                event_filter=event_filter,\n            )\n\n            next_token = pagin_config.from_token.copy_and_replace(\n                \"room_key\", next_key\n            )\n\n        if events:\n            if event_filter:\n                events = event_filter.filter(events)\n\n            events = yield filter_events_for_client(\n                self.store,\n                user_id,\n                events,\n                is_peeking=(member_event_id is None),\n            )\n\n        if not events:\n            defer.returnValue({\n                \"chunk\": [],\n                \"start\": pagin_config.from_token.to_string(),\n                \"end\": next_token.to_string(),\n            })\n\n        state = None\n        if event_filter and event_filter.lazy_load_members():\n            # TODO: remove redundant members\n\n            # FIXME: we also care about invite targets etc.\n            state_filter = StateFilter.from_types(\n                (EventTypes.Member, event.sender)\n                for event in events\n            )\n\n            state_ids = yield self.store.get_state_ids_for_event(\n                events[0].event_id, state_filter=state_filter,\n            )\n\n            if state_ids:\n                state = yield self.store.get_events(list(state_ids.values()))\n                state = state.values()\n\n        time_now = self.clock.time_msec()\n\n        chunk = {\n            \"chunk\": [\n                serialize_event(e, time_now, as_client_event)\n                for e in events\n            ],\n            \"start\": pagin_config.from_token.to_string(),\n            \"end\": next_token.to_string(),\n        }\n\n        if state:\n            chunk[\"state\"] = [\n                serialize_event(e, time_now, as_client_event)\n                for e in state\n            ]\n\n        defer.returnValue(chunk)\n", "path": "synapse/handlers/pagination.py"}]}