[WIP] decoupled webhook processing

Author: swrichards

src/notifications/models.py

@@ -156,3 +156,14 @@ def deregister(self) -> None:
 
         self._subscription = ""
         self.save(update_fields=["_subscription"])
+
+class NotificationRecord(models.Model):
+    subscription = models.ForeignKey(Subscription, related_name="received_notifications", on_delete=models.CASCADE)
+    payload = models.BinaryField()
+    kanaal = models.CharField(max_length=256)
+    received_at = models.DateTimeField(auto_created=True)
+    last_processed_at = models.DateTimeField()
+    is_valid = models.BooleanField(default=False)
+    is_processed = models.BooleanField(default=False)
+    is_locked = models.BooleanField(default=False)
+    processing_output = models.TextField()

src/open_inwoner/openzaak/api/views.py

@@ -6,6 +6,7 @@
 from zgw_consumers.api_models.base import factory
 
 from notifications.api.serializers import NotificatieSerializer
+from notifications.models import NotificationRecord
 from open_inwoner.configurations.models import SiteConfiguration
 from open_inwoner.openzaak.api_models import Notification
 from open_inwoner.openzaak.auth import get_valid_subscription_from_request
@@ -40,6 +41,14 @@ def post(self, request):
                 status=status.HTTP_401_UNAUTHORIZED,
             )
 
+        # Create initial log, so
+        notification_record = NotificationRecord.objects.create(
+            subscription=subscription,
+            payload=request.content,
+            is_processed=False,
+            is_valid=False,
+        )
+
         # deserialize
         serializer = NotificatieSerializer(data=request.data)
         if not serializer.is_valid():
@@ -48,19 +57,33 @@ def post(self, request):
 
         notification = factory(Notification, serializer.validated_data)
 
-        # verify channel
+        # Update record with basic metadata
+        notification_record.is_valid = True
+        notification_record.kanaal = notification.kanaal
+        notification_record.save(
+            update_fields=(
+                "is_valid",
+                "kanaal",
+            )
+        )
+
+        # Handle test webhooks, and leave a record so it's visible for validation purposes
         if notification.kanaal == "test":
             self.log_webhook_test_channel(notification)
-            return Response(status=status.HTTP_204_NO_CONTENT)
 
+            # Test webhooks don't have to be processed
+            notification_record.is_processed = True
+            notification_record.processing_output = "test channel, no further processing done"
+            notification_record.save(update_fields=("is_processed", "processing_output", ))
+            return Response(status=status.HTTP_204_NO_CONTENT)
+            
         if notification.kanaal not in subscription.channels:
             self.log_webhook_channel_not_subscribed(notification)
-            return Response(
-                {
-                    "detail": f"notification channel '{notification.kanaal}' not subscribed to"
-                },
-                status=status.HTTP_400_BAD_REQUEST,
-            )
+
+            notification_record.is_processed = True
+            notification_record.processing_output = "channel not subscribed"
+            notification_record.save(update_fields=("is_processed", "processing_output", ))
+            return Response({"detail": "channel not subscribed"}, status=status.HTTP_400_BAD_REQUEST)
 
         if self.accept_channels and notification.kanaal not in self.accept_channels:
             self.log_webhook_channel_not_acceptable(notification)
@@ -97,5 +120,7 @@ def handle_notification(self, notification: Notification):
         if not config.notifications_cases_enabled:
             return
 
+        # If process eager, do now
+
         notification_data = dataclasses.asdict(notification)
         process_zaken_notification.delay(notification_data)

src/open_inwoner/openzaak/tasks.py

@@ -1,13 +1,17 @@
 import io
 
+import json
 from django.core.management import call_command
+from django.utils import timezone
 
 import structlog
 from zgw_consumers.api_models.base import factory
-
+from django.db import transaction
+from notifications.models import NotificationRecord
 from open_inwoner.celery import app
 from open_inwoner.openzaak.api_models import Notification
 from open_inwoner.openzaak.notifications import handle_zaken_notification
+from open_inwoner.utils.logentry import system_action
 
 logger = structlog.stdlib.get_logger(__name__)
 
@@ -25,8 +29,43 @@ def import_zgw_data():
     return out.getvalue()
 
 
+# TODO: Use Celery once for locking
 @app.task
-def process_zaken_notification(notification_data: dict):
+def process_zaken_notification(record_pk: int):
     logger.info("Started process_zaken_notification() task")
-    notification = factory(Notification, notification_data)
-    handle_zaken_notification(notification)
+
+    # The database guarantees that this WHERE clause evaluation
+    # + UPDATE happens atomically - no other transaction can see 
+    # or modify the row in between. This ensure only one task
+    # can process at a time
+    rows_updated = NotificationRecord.objects.filter(
+        pk=record_pk,
+        is_locked=False
+    ).update(is_locked=True, last_processed_at=timezone.now())
+
+    if rows_updated == 0:
+        # Another task already locked it. Log and raise, so Celery marks the
+        # task as failed
+        logger.info("Notification record already locked, skipping", record_pk=record_pk)
+        raise RuntimeError("This task is already being processed")
+    
+    try:
+        notification_record = NotificationRecord.objects.get(pk=record_pk)
+        notification_data = json.load(notification_record.payload)
+        notification = factory(Notification, notification_data)
+        handle_zaken_notification(notification)
+    except NotificationRecord.DoesNotExist:
+        system_action(
+            "Attempted to process unknown Notification Record",
+            log_level=logging.ERROR,
+            record_pk=record_pk,
+        )
+    except Exception:
+        msg = "Unknown error while processing notification"
+        system_action(msg)
+        logger.exception(msg)
+    finally:
+        NotificationRecord.objects.filter(pk=record_pk).update(
+            is_locked=False,
+            is_processed=True,
+        )