def account_data(self, writer, current_token, limit):
        current_position = current_token.account_data

        user_account_data = parse_integer(writer.request, "user_account_data")
        room_account_data = parse_integer(writer.request, "room_account_data")
        tag_account_data = parse_integer(writer.request, "tag_account_data")

        if user_account_data is not None or room_account_data is not None:
            if user_account_data is None:
                user_account_data = current_position
            if room_account_data is None:
                room_account_data = current_position
            user_rows, room_rows = yield self.store.get_all_updated_account_data(
                user_account_data, room_account_data, current_position, limit
            )
            writer.write_header_and_rows("user_account_data", user_rows, (
                "position", "user_id", "type", "content"
            ))
            writer.write_header_and_rows("room_account_data", room_rows, (
                "position", "user_id", "room_id", "type", "content"
            ))

        if tag_account_data is not None:
            tag_rows = yield self.store.get_all_updated_tags(
                tag_account_data, current_position, limit
            )
            writer.write_header_and_rows("tag_account_data", tag_rows, (
                "position", "user_id", "room_id", "tags"
            ))

    def _async_render_GET(self, request):
        server_name, media_id, _ = parse_media_id(request)
        width = parse_integer(request, "width")
        height = parse_integer(request, "height")
        method = parse_string(request, "method", "scale")
        m_type = parse_string(request, "type", "image/png")

        if server_name == self.server_name:
            if self.dynamic_thumbnails:
                yield self._select_or_generate_local_thumbnail(
                    request, media_id, width, height, method, m_type
                )
            else:
                yield self._respond_local_thumbnail(
                    request, media_id, width, height, method, m_type
                )
        else:
            if self.dynamic_thumbnails:
                yield self._select_or_generate_remote_thumbnail(
                    request, server_name, media_id,
                    width, height, method, m_type
                )
            else:
                yield self._respond_remote_thumbnail(
                    request, server_name, media_id,
                    width, height, method, m_type
                )

    def on_GET(self, request, target_user_id):
        """Get request to get specific number of users from Synapse.
        This needs user to have administrator access in Synapse.
        """
        target_user = UserID.from_string(target_user_id)
        requester = yield self.auth.get_user_by_req(request)
        is_admin = yield self.auth.is_server_admin(requester.user)

        if not is_admin:
            raise AuthError(403, "You are not a server admin")

        # To allow all users to get the users list
        # if not is_admin and target_user != auth_user:
        #     raise AuthError(403, "You are not a server admin")

        if not self.hs.is_mine(target_user):
            raise SynapseError(400, "Can only users a local user")

        order = "name"  # order by name in user table
        start = parse_integer(request, "start", required=True)
        limit = parse_integer(request, "limit", required=True)

        logger.info("limit: %s, start: %s", limit, start)

        ret = yield self.handlers.admin_handler.get_users_paginate(
            order, start, limit
        )
        defer.returnValue((200, ret))

    def _async_render_GET(self, request):
        limit = parse_integer(request, "limit", 100)
        timeout = parse_integer(request, "timeout", 10 * 1000)

        request.setHeader(b"Content-Type", b"application/json")

        request_streams = {
            name: parse_integer(request, name)
            for names in STREAM_NAMES for name in names
        }
        request_streams["streams"] = parse_string(request, "streams")

        def replicate():
            return self.replicate(request_streams, limit)

        result = yield self.notifier.wait_for_replication(replicate, timeout)

        for stream_name, stream_content in result.items():
            logger.info(
                "Replicating %d rows of %s from %s -> %s",
                len(stream_content["rows"]),
                stream_name,
                request_streams.get(stream_name),
                stream_content["position"],
            )

        request.write(json.dumps(result, ensure_ascii=False))
        finish_request(request)

    def _async_render_GET(self, request):
        limit = parse_integer(request, "limit", 100)
        timeout = parse_integer(request, "timeout", 10 * 1000)

        request.setHeader(b"Content-Type", b"application/json")
        writer = _Writer(request)

        @defer.inlineCallbacks
        def replicate():
            current_token = yield self.current_replication_token()
            logger.info("Replicating up to %r", current_token)

            yield self.account_data(writer, current_token, limit)
            yield self.events(writer, current_token, limit)
            yield self.presence(writer, current_token)  # TODO: implement limit
            yield self.typing(writer, current_token)  # TODO: implement limit
            yield self.receipts(writer, current_token, limit)
            yield self.push_rules(writer, current_token, limit)
            yield self.pushers(writer, current_token, limit)
            self.streams(writer, current_token)

            logger.info("Replicated %d rows", writer.total)
            defer.returnValue(writer.total)

        yield self.notifier.wait_for_replication(replicate, timeout)

        writer.finish()

    def events(self, writer, current_token, limit):
        request_events = parse_integer(writer.request, "events")
        request_backfill = parse_integer(writer.request, "backfill")

        if request_events is not None or request_backfill is not None:
            if request_events is None:
                request_events = current_token.events
            if request_backfill is None:
                request_backfill = current_token.backfill
            events_rows, backfill_rows = yield self.store.get_all_new_events(
                request_backfill, request_events, current_token.backfill, current_token.events, limit
            )
            writer.write_header_and_rows("events", events_rows, ("position", "internal", "json"))
            writer.write_header_and_rows("backfill", backfill_rows, ("position", "internal", "json"))

    def on_GET(self, request):
        user, client = yield self.auth.get_user_by_req(request)

        timeout = parse_integer(request, "timeout", default=0)
        limit = parse_integer(request, "limit", required=True)
        gap = parse_boolean(request, "gap", default=True)
        sort = parse_string(request, "sort", default="timeline,asc", allowed_values=self.ALLOWED_SORT)
        since = parse_string(request, "since")
        set_presence = parse_string(request, "set_presence", default="online", allowed_values=self.ALLOWED_PRESENCE)
        backfill = parse_boolean(request, "backfill", default=False)
        filter_id = parse_string(request, "filter", default=None)

        logger.info(
            "/sync: user=%r, timeout=%r, limit=%r, gap=%r, sort=%r, since=%r,"
            " set_presence=%r, backfill=%r, filter_id=%r"
            % (user, timeout, limit, gap, sort, since, set_presence, backfill, filter_id)
        )

        # TODO(mjark): Load filter and apply overrides.
        try:
            filter = yield self.filtering.get_user_filter(user.localpart, filter_id)
        except:
            filter = Filter({})
        # filter = filter.apply_overrides(http_request)
        # if filter.matches(event):
        #   # stuff

        sync_config = SyncConfig(
            user=user, client_info=client, gap=gap, limit=limit, sort=sort, backfill=backfill, filter=filter
        )

        if since is not None:
            since_token = StreamToken.from_string(since)
        else:
            since_token = None

        sync_result = yield self.sync_handler.wait_for_sync_for_user(
            sync_config, since_token=since_token, timeout=timeout
        )

        time_now = self.clock.time_msec()

        response_content = {
            "public_user_data": self.encode_user_data(sync_result.public_user_data, filter, time_now),
            "private_user_data": self.encode_user_data(sync_result.private_user_data, filter, time_now),
            "rooms": self.encode_rooms(sync_result.rooms, filter, time_now, client.token_id),
            "next_batch": sync_result.next_batch.to_string(),
        }

        defer.returnValue((200, response_content))

    def pushers(self, writer, current_token, limit):
        current_position = current_token.pushers

        pushers = parse_integer(writer.request, "pushers")
        if pushers is not None:
            updated, deleted = yield self.store.get_all_updated_pushers(pushers, current_position, limit)
            writer.write_header_and_rows(
                "pushers",
                updated,
                (
                    "position",
                    "user_id",
                    "access_token",
                    "profile_tag",
                    "kind",
                    "app_id",
                    "app_display_name",
                    "device_display_name",
                    "pushkey",
                    "ts",
                    "lang",
                    "data",
                ),
            )
            writer.write_header_and_rows("deleted", deleted, ("position", "user_id", "app_id", "pushkey"))

    def from_request(cls, request, raise_invalid_params=True,
                     default_limit=None):
        direction = parse_string(request, "dir", default='f', allowed_values=['f', 'b'])

        from_tok = parse_string(request, "from")
        to_tok = parse_string(request, "to")

        try:
            if from_tok == "END":
                from_tok = None  # For backwards compat.
            elif from_tok:
                from_tok = StreamToken.from_string(from_tok)
        except Exception:
            raise SynapseError(400, "'from' paramater is invalid")

        try:
            if to_tok:
                to_tok = StreamToken.from_string(to_tok)
        except Exception:
            raise SynapseError(400, "'to' paramater is invalid")

        limit = parse_integer(request, "limit", default=default_limit)

        if limit and limit < 0:
            raise SynapseError(400, "Limit must be 0 or above")

        try:
            return PaginationConfig(from_tok, to_tok, direction, limit)
        except Exception:
            logger.exception("Failed to create pagination config")
            raise SynapseError(400, "Invalid request.")

    def on_GET(self, request, room_id, parent_id, relation_type=None, event_type=None):
        requester = yield self.auth.get_user_by_req(request, allow_guest=True)

        yield self.auth.check_in_room_or_world_readable(
            room_id, requester.user.to_string()
        )

        # This checks that a) the event exists and b) the user is allowed to
        # view it.
        yield self.event_handler.get_event(requester.user, room_id, parent_id)

        if relation_type not in (RelationTypes.ANNOTATION, None):
            raise SynapseError(400, "Relation type must be 'annotation'")

        limit = parse_integer(request, "limit", default=5)
        from_token = parse_string(request, "from")
        to_token = parse_string(request, "to")

        if from_token:
            from_token = AggregationPaginationToken.from_string(from_token)

        if to_token:
            to_token = AggregationPaginationToken.from_string(to_token)

        res = yield self.store.get_aggregation_groups_for_event(
            event_id=parent_id,
            event_type=event_type,
            limit=limit,
            from_token=from_token,
            to_token=to_token,
        )

        defer.returnValue((200, res.to_dict()))

    def on_GET(self, request):
        server = parse_string(request, "server", default=None)

        try:
            yield self.auth.get_user_by_req(request, allow_guest=True)
        except AuthError as e:
            # We allow people to not be authed if they're just looking at our
            # room list, but require auth when we proxy the request.
            # In both cases we call the auth function, as that has the side
            # effect of logging who issued this request if an access token was
            # provided.
            if server:
                raise e
            else:
                pass

        limit = parse_integer(request, "limit", 0)
        since_token = parse_string(request, "since", None)

        handler = self.hs.get_room_list_handler()
        if server:
            data = yield handler.get_remote_public_room_list(
                server,
                limit=limit,
                since_token=since_token,
            )
        else:
            data = yield handler.get_local_public_room_list(
                limit=limit,
                since_token=since_token,
            )

        defer.returnValue((200, data))

    def typing(self, writer, current_token):
        current_position = current_token.presence

        request_typing = parse_integer(writer.request, "typing")

        if request_typing is not None:
            typing_rows = yield self.typing_handler.get_all_typing_updates(request_typing, current_position)
            writer.write_header_and_rows("typing", typing_rows, ("position", "room_id", "typing"))

 def on_GET(self, request, user_id, device_id, algorithm):
     yield self.auth.get_user_by_req(request)
     timeout = parse_integer(request, "timeout", 10 * 1000)
     result = yield self.e2e_keys_handler.claim_one_time_keys(
         {"one_time_keys": {user_id: {device_id: algorithm}}},
         timeout,
     )
     defer.returnValue((200, result))

 def on_POST(self, request, user_id, device_id, algorithm):
     yield self.auth.get_user_by_req(request)
     timeout = parse_integer(request, "timeout", 10 * 1000)
     body = parse_json_object_from_request(request)
     result = yield self.e2e_keys_handler.claim_one_time_keys(
         body,
         timeout,
     )
     defer.returnValue((200, result))

    def on_POST(self, request):
        yield assert_requester_is_admin(self.auth, request)

        before_ts = parse_integer(request, "before_ts", required=True)
        logger.info("before_ts: %r", before_ts)

        ret = yield self.media_repository.delete_old_remote_media(before_ts)

        defer.returnValue((200, ret))

    def receipts(self, writer, current_token, limit):
        current_position = current_token.receipts

        request_receipts = parse_integer(writer.request, "receipts")

        if request_receipts is not None:
            receipts_rows = yield self.store.get_all_updated_receipts(request_receipts, current_position, limit)
            writer.write_header_and_rows(
                "receipts", receipts_rows, ("position", "room_id", "receipt_type", "user_id", "event_id", "data")
            )

    def on_GET(self, request):
        requester = yield self.auth.get_user_by_req(request)
        user_id = requester.user.to_string()

        from_token = parse_string(request, "from", required=False)
        limit = parse_integer(request, "limit", default=50)
        only = parse_string(request, "only", required=False)

        limit = min(limit, 500)

        push_actions = yield self.store.get_push_actions_for_user(
            user_id, from_token, limit, only_highlight=(only == "highlight")
        )

        receipts_by_room = yield self.store.get_receipts_for_user_with_orderings(
            user_id, 'm.read'
        )

        notif_event_ids = [pa["event_id"] for pa in push_actions]
        notif_events = yield self.store.get_events(notif_event_ids)

        returned_push_actions = []

        next_token = None

        for pa in push_actions:
            returned_pa = {
                "room_id": pa["room_id"],
                "profile_tag": pa["profile_tag"],
                "actions": pa["actions"],
                "ts": pa["received_ts"],
                "event": serialize_event(
                    notif_events[pa["event_id"]],
                    self.clock.time_msec(),
                    event_format=format_event_for_client_v2_without_room_id,
                ),
            }

            if pa["room_id"] not in receipts_by_room:
                returned_pa["read"] = False
            else:
                receipt = receipts_by_room[pa["room_id"]]

                returned_pa["read"] = (
                    receipt["topological_ordering"], receipt["stream_ordering"]
                ) >= (
                    pa["topological_ordering"], pa["stream_ordering"]
                )
            returned_push_actions.append(returned_pa)
            next_token = str(pa["stream_ordering"])

        defer.returnValue((200, {
            "notifications": returned_push_actions,
            "next_token": next_token,
        }))

    def on_GET(self, request, target_user_id):
        """Get request to get specific number of users from Synapse.
        This needs user to have administrator access in Synapse.
        """
        yield assert_requester_is_admin(self.auth, request)

        target_user = UserID.from_string(target_user_id)

        if not self.hs.is_mine(target_user):
            raise SynapseError(400, "Can only users a local user")

        order = "name"  # order by name in user table
        start = parse_integer(request, "start", required=True)
        limit = parse_integer(request, "limit", required=True)

        logger.info("limit: %s, start: %s", limit, start)

        ret = yield self.handlers.admin_handler.get_users_paginate(
            order, start, limit
        )
        defer.returnValue((200, ret))

    def push_rules(self, writer, current_token, limit):
        current_position = current_token.push_rules

        push_rules = parse_integer(writer.request, "push_rules")

        if push_rules is not None:
            rows = yield self.store.get_all_push_rule_updates(
                push_rules, current_position, limit
            )
            writer.write_header_and_rows("push_rules", rows, (
                "position", "event_stream_ordering", "user_id", "rule_id", "op",
                "priority_class", "priority", "conditions", "actions"
            ))

    def on_POST(self, request):
        requester = yield self.auth.get_user_by_req(request)
        is_admin = yield self.auth.is_server_admin(requester.user)

        if not is_admin:
            raise AuthError(403, "You are not a server admin")

        before_ts = parse_integer(request, "before_ts", required=True)
        logger.info("before_ts: %r", before_ts)

        ret = yield self.media_repository.delete_old_remote_media(before_ts)

        defer.returnValue((200, ret))