[Draft] Feature/1206 failed notif email

requirements/base.in

@@ -27,7 +27,7 @@ mozilla-django-oidc-db
 # API libraries
 djangorestframework
 gemma-zds-client<1.0
-git+https://github.com/VNG-Realisatie/vng-api-common.git@43779a141cfba3ecd5f90959c5d6b98c80386281#egg=vng_api_common
+git+https://github.com/VNG-Realisatie/vng-api-common.git@c870aec8ccabb8a7556426262f3c2f55800b08a8#egg=vng_api_common
 
 # WSGI servers & monitoring - production oriented
 uwsgi

requirements/base.txt

@@ -287,7 +287,7 @@ vine==5.0.0
     #   amqp
     #   celery
     #   kombu
-vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@43779a141cfba3ecd5f90959c5d6b98c80386281
+vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@c870aec8ccabb8a7556426262f3c2f55800b08a8
     # via -r requirements/base.in
 wcwidth==0.2.5
     # via prompt-toolkit

requirements/ci.txt

@@ -467,7 +467,7 @@ vine==5.0.0
     #   amqp
     #   celery
     #   kombu
-vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@43779a141cfba3ecd5f90959c5d6b98c80386281
+vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@c870aec8ccabb8a7556426262f3c2f55800b08a8
     # via -r requirements/base.txt
 waitress==2.1.1
     # via webtest

requirements/dev.txt

@@ -520,7 +520,7 @@ vine==5.0.0
     #   amqp
     #   celery
     #   kombu
-vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@43779a141cfba3ecd5f90959c5d6b98c80386281
+vng-api-common @ git+https://github.com/VNG-Realisatie/vng-api-common.git@c870aec8ccabb8a7556426262f3c2f55800b08a8
     # via -r requirements/ci.txt
 waitress==2.1.1
     # via

src/nrc/api/tasks.py

@@ -1,7 +1,16 @@
+import inspect
 import json
 import logging
+from datetime import datetime
+from urllib.parse import urlparse
+from uuid import UUID
 
+from django.conf import settings
+from django.core.cache import cache
+from django.core.mail import send_mail
 from django.core.serializers.json import DjangoJSONEncoder
+from django.urls import reverse
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
 import requests
@@ -16,6 +25,9 @@
 logger = logging.getLogger(__name__)
 
 
+LATEST_EMAIL_CACHE_PREFIX = "latest_notification_email"
+
+
 def add_autoretry_behaviour(task, **options):
     """
     Adapted from celery to use admin configurable autoretry settings
@@ -29,7 +41,7 @@ def add_autoretry_behaviour(task, **options):
     if autoretry_for and not hasattr(task, "_orig_run"):
 
         @wraps(task.run)
-        def run(*args, **kwargs):
+        def run(sub_id: int, msg: dict, **kwargs):
             config = NotificationsConfig.get_solo()
             max_retries = config.notification_delivery_max_retries
             retry_backoff = config.notification_delivery_retry_backoff
@@ -38,7 +50,7 @@ def run(*args, **kwargs):
             task.max_retries = max_retries
 
             try:
-                return task._orig_run(*args, **kwargs)
+                return task._orig_run(sub_id, msg, **kwargs)
             except Ignore:
                 # If Ignore signal occurs task shouldn't be retried,
                 # even if it suits autoretry_for list
@@ -59,11 +71,17 @@ def run(*args, **kwargs):
                         task, "override_max_retries", max_retries
                     )
 
-                ret = task.retry(exc=exc, **retry_kwargs)
-                # Stop propagation
-                if hasattr(task, "override_max_retries"):
-                    delattr(task, "override_max_retries")
-                raise ret
+                try:
+                    ret = task.retry(exc=exc, **retry_kwargs)
+                    # Stop propagation
+                    if hasattr(task, "override_max_retries"):
+                        delattr(task, "override_max_retries")
+                    raise ret
+                except autoretry_for:
+                    # Final retry failed, send email
+                    send_email_to_admins.delay(sub_id, msg["uuid"])
+                except Exception:
+                    raise
 
         task._orig_run, task.run = task.run, run
 
@@ -72,6 +90,60 @@ class NotificationException(Exception):
     pass
 
 
+def cooldown_period(func):
+    def inner(sub_id: int, notificatie_uuid: UUID):
+        subscription = Abonnement.objects.get(pk=sub_id)
+        cache_key = f"{LATEST_EMAIL_CACHE_PREFIX}:{subscription.uuid}"
+        current_time = timezone.now()
+        latest_timestamp = cache.get(cache_key)
+        if (
+            not latest_timestamp
+            or (
+                current_time
+                - timezone.make_aware(datetime.fromtimestamp(latest_timestamp))
+            ).seconds
+            / 3600
+            >= 24
+        ):
+            cache.set(cache_key, current_time.timestamp())
+            return func(sub_id, notificatie_uuid)
+
+    return inner
+
+
+@app.task
+@cooldown_period
+def send_email_to_admins(sub_id: int, notificatie_uuid: UUID) -> None:
+    subscription = Abonnement.objects.get(pk=sub_id)
+    config = NotificationsConfig.get_solo()
+
+    if not config.failed_notification_admin_recipients:
+        return
+
+    parsed = urlparse(config.api_root)
+    notifications_changelist = reverse("admin:datamodel_notificatie_changelist")
+
+    body = inspect.cleandoc(
+        """
+        Notification {notificatie_uuid} to subscriber {sub_uuid} failed.
+
+        See {admin_url} for more info
+    """.format(
+            notificatie_uuid=notificatie_uuid,
+            sub_uuid=subscription.uuid,
+            admin_url=f"{parsed.scheme}://{parsed.netloc}{notifications_changelist}",
+        )
+    )
+
+    send_mail(
+        f"Failed notification - {notificatie_uuid}",
+        body,
+        settings.DEFAULT_FROM_EMAIL,
+        config.failed_notification_admin_recipients,
+        fail_silently=False,
+    )
+
+
 @app.task
 def deliver_message(sub_id: int, msg: dict, **kwargs) -> None:
     """
@@ -117,7 +189,7 @@ def deliver_message(sub_id: int, msg: dict, **kwargs) -> None:
                 notificatie_id=notificatie_id,
                 abonnement=sub,
                 attempt=kwargs.get("attempt", 1),
-                **response_init_kwargs
+                **response_init_kwargs,
             )
 
 

src/nrc/api/tests/test_notificatie.py

@@ -1,18 +1,28 @@
+import inspect
+from datetime import datetime
 from unittest.mock import patch
 
+from django.core import mail
+from django.core.cache import cache
 from django.test import TestCase, override_settings
 from django.utils.timezone import now
 
 import requests_mock
 from celery.exceptions import Retry
+from freezegun import freeze_time
 from rest_framework import status
 from rest_framework.reverse import reverse
 from rest_framework.test import APITestCase
 from vng_api_common.conf.api import BASE_REST_FRAMEWORK
 from vng_api_common.notifications.models import NotificationsConfig
 from vng_api_common.tests import JWTAuthMixin
 
-from nrc.api.tasks import deliver_message
+from nrc.api.tasks import (
+    LATEST_EMAIL_CACHE_PREFIX,
+    NotificationException,
+    deliver_message,
+    send_email_to_admins,
+)
 from nrc.datamodel.models import Notificatie
 from nrc.datamodel.tests.factories import (
     AbonnementFactory,
@@ -159,6 +169,7 @@ def test_notificatie_send_empty_kenmerk_value(self, mock_task):
         )
 
 
+@freeze_time("2022-01-01T12:00:00Z")
 @patch("nrc.api.tasks.get_exponential_backoff_interval")
 @patch("nrc.api.tasks.NotificationsConfig.get_solo")
 @patch("nrc.api.serializers.deliver_message.retry")
@@ -210,3 +221,145 @@ def test_notificatie_retry_use_global_config(
             full_jitter=False,
         )
         self.assertEqual(deliver_message.max_retries, 4)
+
+    @patch("nrc.api.tasks.send_email_to_admins.delay", side_effect=send_email_to_admins)
+    def test_notificatie_retry_email(
+        self,
+        mock_email,
+        mock_retry,
+        mock_config,
+        mock_get_exponential_backoff,
+    ):
+        """
+        Verify that an email is sent after all retries are done
+        """
+        mock_config.return_value = NotificationsConfig(
+            api_root="https://nrc.com/api/v1/",
+            notification_delivery_max_retries=4,
+            notification_delivery_retry_backoff=4,
+            notification_delivery_retry_backoff_max=28,
+            failed_notification_admin_recipients=["foo@bar.nl", "bar@baz.nl"],
+        )
+        kanaal = KanaalFactory.create(
+            naam="zaken", filters=["bron", "zaaktype", "vertrouwelijkheidaanduiding"]
+        )
+        abon = AbonnementFactory.create(callback_url="https://example.com/callback")
+        filter_group = FilterGroupFactory.create(kanaal=kanaal, abonnement=abon)
+        FilterFactory.create(
+            filter_group=filter_group, key="bron", value="082096752011"
+        )
+        msg = {
+            "uuid": "920fc3b4-622c-45c9-b656-dee6cd463627",
+            "kanaal": "zaken",
+            "hoofdObject": "https://ref.tst.vng.cloud/zrc/api/v1/zaken/d7a22",
+            "resource": "status",
+            "resourceUrl": "https://ref.tst.vng.cloud/zrc/api/v1/statussen/d7a22/721c9",
+            "actie": "create",
+            "aanmaakdatum": now(),
+            "kenmerken": {
+                "bron": "082096752011",
+                "zaaktype": "example.com/api/v1/zaaktypen/5aa5c",
+                "vertrouwelijkheidaanduiding": "openbaar",
+            },
+        }
+
+        # Mock that max retries have been exceeded
+        mock_retry.side_effect = NotificationException()
+        with requests_mock.Mocker() as m:
+            m.post(abon.callback_url, status_code=404)
+            deliver_message(abon.id, msg)
+
+        mock_email.assert_called_once_with(
+            abon.pk, "920fc3b4-622c-45c9-b656-dee6cd463627"
+        )
+
+        self.assertEqual(
+            cache.get(f"{LATEST_EMAIL_CACHE_PREFIX}:{abon.uuid}"),
+            datetime(2022, 1, 1, 12, 0, 0).timestamp(),
+        )
+        self.assertEqual(len(mail.outbox), 1)
+
+        outbound_mail = mail.outbox[0]
+        notifications_changelist = reverse("admin:datamodel_notificatie_changelist")
+        admin_url = f"https://nrc.com{notifications_changelist}"
+
+        self.assertEqual(
+            outbound_mail.subject,
+            "Failed notification - 920fc3b4-622c-45c9-b656-dee6cd463627",
+        )
+        self.assertEqual(
+            outbound_mail.body,
+            inspect.cleandoc(
+                f"""
+            Notification 920fc3b4-622c-45c9-b656-dee6cd463627 to subscriber {abon.uuid} failed.
+
+            See {admin_url} for more info
+        """
+            ),
+        )
+        self.assertEqual(outbound_mail.from_email, "opennotificaties@example.com")
+        self.assertEqual(outbound_mail.to, ["foo@bar.nl", "bar@baz.nl"])
+
+    @patch("nrc.api.tasks.send_email_to_admins.delay", side_effect=send_email_to_admins)
+    def test_notificatie_retry_email_cooldown_period(
+        self,
+        mock_email,
+        mock_retry,
+        mock_config,
+        mock_get_exponential_backoff,
+    ):
+        """
+        Verify that an email is sent after all retries are done
+        """
+        mock_config.return_value = NotificationsConfig(
+            api_root="https://nrc.com/api/v1/",
+            notification_delivery_max_retries=4,
+            notification_delivery_retry_backoff=4,
+            notification_delivery_retry_backoff_max=28,
+            failed_notification_admin_recipients=["foo@bar.nl", "bar@baz.nl"],
+        )
+        kanaal = KanaalFactory.create(
+            naam="zaken", filters=["bron", "zaaktype", "vertrouwelijkheidaanduiding"]
+        )
+        abon = AbonnementFactory.create(callback_url="https://example.com/callback")
+        filter_group = FilterGroupFactory.create(kanaal=kanaal, abonnement=abon)
+        FilterFactory.create(
+            filter_group=filter_group, key="bron", value="082096752011"
+        )
+        msg = {
+            "uuid": "920fc3b4-622c-45c9-b656-dee6cd463627",
+            "kanaal": "zaken",
+            "hoofdObject": "https://ref.tst.vng.cloud/zrc/api/v1/zaken/d7a22",
+            "resource": "status",
+            "resourceUrl": "https://ref.tst.vng.cloud/zrc/api/v1/statussen/d7a22/721c9",
+            "actie": "create",
+            "aanmaakdatum": now(),
+            "kenmerken": {
+                "bron": "082096752011",
+                "zaaktype": "example.com/api/v1/zaaktypen/5aa5c",
+                "vertrouwelijkheidaanduiding": "openbaar",
+            },
+        }
+
+        cache.set(
+            f"{LATEST_EMAIL_CACHE_PREFIX}:{abon.uuid}",
+            datetime(2022, 1, 1, 6, 0, 0).timestamp(),
+        )
+
+        # Mock that max retries have been exceeded
+        mock_retry.side_effect = NotificationException()
+        with requests_mock.Mocker() as m:
+            m.post(abon.callback_url, status_code=404)
+            deliver_message(abon.id, msg)
+
+        mock_email.assert_called_once_with(
+            abon.pk, "920fc3b4-622c-45c9-b656-dee6cd463627"
+        )
+
+        # Latest notification remains the same
+        self.assertEqual(
+            cache.get(f"{LATEST_EMAIL_CACHE_PREFIX}:{abon.uuid}"),
+            datetime(2022, 1, 1, 6, 0, 0).timestamp(),
+        )
+        # No emails should be sent, since the cooldown period is active
+        self.assertEqual(len(mail.outbox), 0)