From 0b282d33afc7fd375a30c8bd18946f146be89cd8 Mon Sep 17 00:00:00 2001
From: Mark Haines <mjark@negativecurvature.net>
Date: Tue, 19 Apr 2016 14:43:47 +0100
Subject: [PATCH] Add an HTTP API for removing rejected pushers.

When a push is rejected by the push gateway then synapse needs to
remove the pusher from the database. However we probably don't want
to do that directly from the slave, so we add an HTTP API to synapse
to remove the pusher from the database.
---
 synapse/app/pusher.py                  | 12 ++++++
 synapse/push/httppusher.py             |  2 +-
 synapse/replication/pusher_resource.py | 53 ++++++++++++++++++++++++++
 synapse/replication/resource.py        |  5 ++-
 synapse/server.py                      |  3 ++
 5 files changed, 72 insertions(+), 3 deletions(-)
 create mode 100644 synapse/replication/pusher_resource.py

diff --git a/synapse/app/pusher.py b/synapse/app/pusher.py
index 2fd988615a..fbed9763c0 100644
--- a/synapse/app/pusher.py
+++ b/synapse/app/pusher.py
@@ -89,6 +89,18 @@ class PusherServer(HomeServer):
         self.datastore = PusherSlaveStore(self.get_db_conn(), self)
         logger.info("Finished setting up.")
 
+    def remove_pusher(self, app_id, push_key, user_id):
+        http_client = self.get_simple_http_client()
+        replication_url = self.config.replication_url
+        url = replication_url + "/remove_pushers"
+        return http_client.post_json_get_json(url, {
+            "remove": [{
+                "app_id": app_id,
+                "push_key": push_key,
+                "user_id": user_id,
+            }]
+        })
+
     @defer.inlineCallbacks
     def replicate(self):
         http_client = self.get_simple_http_client()
diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py
index 6950a20632..3992804845 100644
--- a/synapse/push/httppusher.py
+++ b/synapse/push/httppusher.py
@@ -230,7 +230,7 @@ class HttpPusher(object):
                         "Pushkey %s was rejected: removing",
                         pk
                     )
-                    yield self.hs.get_pusherpool().remove_pusher(
+                    yield self.hs.remove_pusher(
                         self.app_id, pk, self.user_id
                     )
         defer.returnValue(True)
diff --git a/synapse/replication/pusher_resource.py b/synapse/replication/pusher_resource.py
new file mode 100644
index 0000000000..b87026d79a
--- /dev/null
+++ b/synapse/replication/pusher_resource.py
@@ -0,0 +1,53 @@
+# Copyright 2016 OpenMarket Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from synapse.http.server import respond_with_json_bytes, request_handler
+from synapse.http.servlet import parse_json_object_from_request
+
+from twisted.web.resource import Resource
+from twisted.web.server import NOT_DONE_YET
+from twisted.internet import defer
+
+
+class PusherResource(Resource):
+    """
+    HTTP endpoint for deleting rejected pushers
+    """
+
+    def __init__(self, hs):
+        Resource.__init__(self)  # Resource is old-style, so no super()
+
+        self.version_string = hs.version_string
+        self.store = hs.get_datastore()
+        self.notifier = hs.get_notifier()
+
+    def render_POST(self, request):
+        self._async_render_POST(request)
+        return NOT_DONE_YET
+
+    @request_handler
+    @defer.inlineCallbacks
+    def _async_render_POST(self, request):
+        content = parse_json_object_from_request(request)
+
+        for remove in content["remove"]:
+            yield self.store.delete_pusher_by_app_id_pushkey_user_id(
+                remove["app_id"],
+                remove["push_key"],
+                remove["user_id"],
+            )
+
+        self.notifier.on_new_replication_data()
+
+        respond_with_json_bytes(request, 200, "{}")
diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py
index af95f8ab22..e5c9a53929 100644
--- a/synapse/replication/resource.py
+++ b/synapse/replication/resource.py
@@ -15,6 +15,7 @@
 
 from synapse.http.servlet import parse_integer, parse_string
 from synapse.http.server import request_handler, finish_request
+from synapse.replication.pusher_resource import PusherResource
 
 from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
@@ -102,8 +103,6 @@ class ReplicationResource(Resource):
     long-polling this replication API for new data on those streams.
     """
 
-    isLeaf = True
-
     def __init__(self, hs):
         Resource.__init__(self)  # Resource is old-style, so no super()
 
@@ -114,6 +113,8 @@ class ReplicationResource(Resource):
         self.typing_handler = hs.get_handlers().typing_notification_handler
         self.notifier = hs.notifier
 
+        self.putChild("remove_pushers", PusherResource(hs))
+
     def render_GET(self, request):
         self._async_render_GET(request)
         return NOT_DONE_YET
diff --git a/synapse/server.py b/synapse/server.py
index 368d615576..ee138de756 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -193,6 +193,9 @@ class HomeServer(object):
             **self.db_config.get("args", {})
         )
 
+    def remove_pusher(self, app_id, push_key, user_id):
+        return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
+
 
 def _make_dependency_method(depname):
     def _get(hs):