[BE] add webhooks failsafe (#18)

* [BE] add webhooks failsafe

* Remove print

* Update README.rst

README.rst

@@ -124,6 +124,10 @@ Webhooks support
 All webhooks should be sent to ``/aa-stripe/webhooks`` url. Add ``STRIPE_WEBHOOK_ENDPOINT_SECRET`` to your settings to enable webhook verifications. Each received webhook is saved as StripeWebhook object in database. User need to add parsing webhooks depending on the project.
 Be advised. There might be times that Webhooks will not arrive because of some error or arrive in incorrect order. When parsing webhook it is also good to download the refered object to verify it's state.
 
+Stripe has the weird tendency to stop sending webhooks, and they have not fixed it yet on their side. To make sure all events have arrived into your system, the ``check_pending_webhooks`` management command should be run chronically.
+In case there is more pending webhooks than specified in the ``PENDING_EVENTS_THRESHOLD`` variable in settings (default: ``20``), an email to project admins will be sent with ids of the pending events, and also the command will fail raising an exception,
+so if you have some kind of error tracking service configured on your servers (for example: `Sentry <https://sentry.io>`_), you will be notified.
+
 Support
 =======
 * Django 1.11

aa_stripe/management/commands/check_pending_webhooks.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+import stripe
+from django.conf import settings
+from django.core.mail import mail_admins
+from django.core.management.base import BaseCommand
+from django.utils.timezone import now
+
+from aa_stripe.models import StripeWebhook
+
+
+def get_pending_events_threshold():
+    return getattr(settings, "PENDING_EVENTS_THRESHOLD", 20)
+
+
+class StripePendingWebooksLimitExceeded(Exception):
+    def __init__(self, pending_events):
+        self.message = "Pending events limit exceeded, current threshold is {}".format(get_pending_events_threshold)
+        # send email to admins
+        email_message = "Pending events at {now}:\n\n{events}".format(
+            now=now(), events="\n".join(event.id for event in pending_events)
+        )
+        mail_admins("Stripe webhooks pending threshold exceeded", email_message)
+        super(StripePendingWebooksLimitExceeded, self).__init__(self.message)
+
+
+class Command(BaseCommand):
+    help = "Check pending webhooks at Stripe API"
+
+    def handle(self, *args, **options):
+        stripe.api_key = settings.STRIPE_API_KEY
+        pending_events_threshold = get_pending_events_threshold()
+
+        pending_events = []
+        last_event = StripeWebhook.objects.latest("created").id
+        while True:
+            event_list = stripe.Event.list(starting_after=last_event, limit=100)  # 100 is the maximum
+            pending_events += event_list["data"]
+            if len(pending_events) > pending_events_threshold:
+                raise StripePendingWebooksLimitExceeded(pending_events)
+
+            if not event_list["has_more"]:
+                break
+            else:
+                last_event = event_list["data"][-1]

tests/test_webhooks.py

@@ -1,14 +1,40 @@
+import time
+from datetime import datetime
+from uuid import uuid4
+
 import mock
 import requests_mock
 import simplejson as json
+from django.core import mail
+from django.core.management import call_command
+from django.test import override_settings
 from rest_framework.reverse import reverse
 from tests.test_utils import BaseTestCase
 
 from aa_stripe.exceptions import StripeWebhookAlreadyParsed
+from aa_stripe.management.commands.check_pending_webhooks import StripePendingWebooksLimitExceeded
 from aa_stripe.models import StripeCoupon, StripeWebhook
 
 
 class TestWebhook(BaseTestCase):
+    def _create_ping_webhook(self):
+        payload = json.loads("""{
+          "id": "",
+          "object": "event",
+          "api_version": "2017-06-05",
+          "created": 1503474921,
+          "livemode": false,
+          "pending_webhooks": 0,
+          "request": {
+            "id": "",
+            "idempotency_key": null
+          },
+          "type": "ping"
+        }""")
+        payload["id"] = "evt_{}".format(uuid4())
+        payload["request"]["id"] = "req_{}".format(uuid4())
+        payload["created"] = int(time.mktime(datetime.now().timetuple()))
+        return StripeWebhook.objects.create(id=payload["id"], raw_data=payload)
 
     def test_subscription_creation(self):
         self.assertEqual(StripeWebhook.objects.count(), 0)
@@ -318,3 +344,48 @@ def test_coupon_delete(self):
             self.assertEqual(response.status_code, 201)
             mocked_signal.assert_called_with(event_action=None, event_model=None, event_type="ping",
                                              instance=StripeWebhook.objects.first(), sender=StripeWebhook)
+
+    @override_settings(PENDING_EVENTS_THRESHOLD=1, ADMINS=["admin@example.com"])
+    def test_check_pending_webhooks_command(self):
+        self._create_ping_webhook()
+        last_webhook = self._create_ping_webhook()
+
+        # create response with fake limits
+        base_stripe_response = json.loads("""{
+          "object": "list",
+          "url": "/v1/events",
+          "has_more": true,
+          "data": []
+        }""")
+        event1_data = last_webhook.raw_data.copy()
+        event1_data["id"] = "evt_1"
+        stripe_response_part1 = base_stripe_response.copy()
+        stripe_response_part1["data"] = [event1_data]
+
+        event2_data = event1_data.copy()
+        event2_data["id"] = "evt_2"
+        stripe_response_part2 = base_stripe_response.copy()
+        stripe_response_part2["data"] = [event2_data]
+        stripe_response_part2["has_more"] = False
+
+        with requests_mock.Mocker() as m:
+            m.register_uri(
+                "GET", "https://api.stripe.com/v1/events?starting_after={}&limit=100".format(last_webhook.id),
+                text=json.dumps(stripe_response_part1))
+            m.register_uri(
+                "GET", "https://api.stripe.com/v1/events?starting_after={}&limit=100".format(event1_data["id"]),
+                text=json.dumps(stripe_response_part2))
+
+            with self.assertRaises(StripePendingWebooksLimitExceeded):
+                call_command("check_pending_webhooks")
+                self.assertEqual(len(mail.outbox), 1)
+                message = mail.outbox[0]
+                self.assertEqual(message.to, "admin@example.com")
+                self.assertIn(event1_data["id"], message)
+                self.assertIn(event2_data["id"], message)
+                self.assertNotIn(last_webhook["id"], message)
+
+            mail.outbox = []
+            with override_settings(PENDING_EVENTS_THRESHOLD=20):
+                call_command("check_pending_webhooks")
+                self.assertEqual(len(mail.outbox), 0)