Dispatch create-path notifications async to fix slow POST latency

dojo/api_v2/serializers.py

@@ -28,6 +28,7 @@
 import dojo.risk_acceptance.helper as ra_helper
 from dojo.authorization.authorization import user_has_permission
 from dojo.authorization.roles_permissions import Permissions
+from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.endpoint.utils import endpoint_filter, endpoint_meta_import
 from dojo.finding.helper import (
     save_endpoints_template,
@@ -116,7 +117,7 @@
     Vulnerability_Id,
     get_current_date,
 )
-from dojo.notifications.helper import create_notification
+from dojo.notifications.helper import async_create_notification
 from dojo.product_announcements import (
     LargeScanSizeProductAnnouncement,
     ScanTypeProductAnnouncement,
@@ -2086,10 +2087,11 @@ def create(self, validated_data):
             jira_helper.push_to_jira(new_finding)
 
         # Create a notification
-        create_notification(
+        dojo_dispatch_task(
+            async_create_notification,
             event="finding_added",
             title=_("Addition of %s") % new_finding.title,
-            finding=new_finding,
+            finding_id=new_finding.id,
             description=_('Finding "%s" was added by %s') % (new_finding.title, new_finding.reporter),
             url=reverse("view_finding", args=(new_finding.id,)),
             icon="exclamation-triangle",

dojo/engagement/signals.py

@@ -7,19 +7,30 @@
 from django.urls import reverse
 from django.utils.translation import gettext as _
 
+from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.file_uploads.helper import delete_related_files
 from dojo.models import Engagement, Product
 from dojo.notes.helper import delete_related_notes
-from dojo.notifications.helper import create_notification
+from dojo.notifications.helper import async_create_notification, create_notification
 from dojo.pghistory_models import DojoEvents
 
 
 @receiver(post_save, sender=Engagement)
 def engagement_post_save(sender, instance, created, **kwargs):
+    # raw=True is set by loaddata; skip dispatch so fixture loading doesn't require a live broker.
+    if kwargs.get("raw"):
+        return
     if created:
         title = _('Engagement created for "%(product)s": %(name)s') % {"product": instance.product, "name": instance.name}
-        create_notification(event="engagement_added", title=title, engagement=instance, product=instance.product,
-                            url=reverse("view_engagement", args=(instance.id,)), url_api=reverse("engagement-detail", args=(instance.id,)))
+        dojo_dispatch_task(
+            async_create_notification,
+            event="engagement_added",
+            title=title,
+            engagement_id=instance.id,
+            product_id=instance.product_id,
+            url=reverse("view_engagement", args=(instance.id,)),
+            url_api=reverse("engagement-detail", args=(instance.id,)),
+        )
 
 
 @receiver(pre_save, sender=Engagement)

dojo/importers/default_importer.py

@@ -18,7 +18,7 @@
     Test,
     Test_Import,
 )
-from dojo.notifications.helper import create_notification
+from dojo.notifications.helper import async_create_notification
 from dojo.utils import get_full_url, perform_product_grading
 from dojo.validators import clean_tags
 
@@ -140,12 +140,13 @@ def process_scan(
         )
         # Send out some notifications to the user
         logger.debug("IMPORT_SCAN: Generating notifications")
-        create_notification(
+        dojo_dispatch_task(
+            async_create_notification,
             event="test_added",
             title=f"Test created for {self.test.engagement.product}: {self.test.engagement.name}: {self.test}",
-            test=self.test,
-            engagement=self.test.engagement,
-            product=self.test.engagement.product,
+            test_id=self.test.id,
+            engagement_id=self.test.engagement_id,
+            product_id=self.test.engagement.product_id,
             url=reverse("view_test", args=(self.test.id,)),
             url_api=reverse("test-detail", args=(self.test.id,)),
         )

dojo/notifications/helper.py

@@ -911,6 +911,47 @@ def send_webhooks_notification(event: str, user_id: int | None = None, **kwargs:
     get_manager_class_instance()._get_manager_instance("webhooks").send_webhooks_notification(event, user=user, **kwargs)
 
 
+@app.task
+def async_create_notification(
+    event: str,
+    engagement_id: int | None = None,
+    product_id: int | None = None,
+    product_type_id: int | None = None,
+    finding_id: int | None = None,
+    test_id: int | None = None,
+    **kwargs: dict,
+) -> None:
+    # Re-fetch by id so the recipient-enumeration query and per-user Alert writes
+    # run in the worker rather than the request thread.
+    if engagement_id is not None:
+        engagement = Engagement.objects.filter(pk=engagement_id).select_related("product").first()
+        if engagement is None:
+            return
+        kwargs["engagement"] = engagement
+    if product_id is not None:
+        product = Product.objects.filter(pk=product_id).first()
+        if product is None:
+            return
+        kwargs["product"] = product
+    if product_type_id is not None:
+        product_type = Product_Type.objects.filter(pk=product_type_id).first()
+        if product_type is None:
+            return
+        kwargs["product_type"] = product_type
+    if finding_id is not None:
+        finding = Finding.objects.filter(pk=finding_id).select_related("test__engagement__product").first()
+        if finding is None:
+            return
+        kwargs["finding"] = finding
+    if test_id is not None:
+        test = Test.objects.filter(pk=test_id).select_related("engagement__product").first()
+        if test is None:
+            return
+        kwargs["test"] = test
+
+    create_notification(event=event, **kwargs)
+
+
 @app.task(ignore_result=True)
 def webhook_reactivation(endpoint_id: int, **_kwargs: dict) -> None:
     get_manager_class_instance()._get_manager_instance("webhooks")._webhook_reactivation(endpoint_id=endpoint_id)

dojo/product/signals.py

@@ -7,9 +7,10 @@
 from django.urls import reverse
 from django.utils.translation import gettext as _
 
+from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.labels import get_labels
 from dojo.models import Product
-from dojo.notifications.helper import create_notification
+from dojo.notifications.helper import async_create_notification, create_notification
 from dojo.pghistory_models import DojoEvents
 from dojo.utils import get_current_user
 
@@ -18,13 +19,18 @@
 
 @receiver(post_save, sender=Product)
 def product_post_save(sender, instance, created, **kwargs):
+    # raw=True is set by loaddata; skip dispatch so fixture loading doesn't require a live broker.
+    if kwargs.get("raw"):
+        return
     if created:
-        create_notification(event="product_added",
-                            title=instance.name,
-                            product=instance,
-                            url=reverse("view_product", args=(instance.id,)),
-                            url_api=reverse("product-detail", args=(instance.id,)),
-                        )
+        dojo_dispatch_task(
+            async_create_notification,
+            event="product_added",
+            title=instance.name,
+            product_id=instance.id,
+            url=reverse("view_product", args=(instance.id,)),
+            url_api=reverse("product-detail", args=(instance.id,)),
+        )
 
 
 @receiver(post_delete, sender=Product)

dojo/product_type/signals.py

@@ -8,23 +8,29 @@
 from django.urls import reverse
 from django.utils.translation import gettext as _
 
+from dojo.celery_dispatch import dojo_dispatch_task
 from dojo.labels import get_labels
 from dojo.models import Product_Type
-from dojo.notifications.helper import create_notification
+from dojo.notifications.helper import async_create_notification, create_notification
 from dojo.pghistory_models import DojoEvents
 
 labels = get_labels()
 
 
 @receiver(post_save, sender=Product_Type)
 def product_type_post_save(sender, instance, created, **kwargs):
+    # raw=True is set by loaddata; skip dispatch so fixture loading doesn't require a live broker.
+    if kwargs.get("raw"):
+        return
     if created:
-        create_notification(event="product_type_added",
-                            title=instance.name,
-                            product_type=instance,
-                            url=reverse("view_product_type", args=(instance.id,)),
-                            url_api=reverse("product_type-detail", args=(instance.id,)),
-                        )
+        dojo_dispatch_task(
+            async_create_notification,
+            event="product_type_added",
+            title=instance.name,
+            product_type_id=instance.id,
+            url=reverse("view_product_type", args=(instance.id,)),
+            url_api=reverse("product_type-detail", args=(instance.id,)),
+        )
 
 
 @receiver(post_delete, sender=Product_Type)

unittests/test_importers_performance.py

@@ -483,9 +483,9 @@ def test_deduplication_performance_pghistory_async(self):
 
         self._deduplication_performance(
             expected_num_queries1=74,
-            expected_num_async_tasks1=1,
-            expected_num_queries2=69,
-            expected_num_async_tasks2=1,
+            expected_num_async_tasks1=2,
+            expected_num_queries2=66,
+            expected_num_async_tasks2=2,
             check_duplicates=False,  # Async mode - deduplication happens later
         )
 
@@ -503,10 +503,10 @@ def test_deduplication_performance_pghistory_no_async(self):
         testuser.usercontactinfo.save()
 
         self._deduplication_performance(
-            expected_num_queries1=81,
-            expected_num_async_tasks1=1,
-            expected_num_queries2=77,
-            expected_num_async_tasks2=1,
+            expected_num_queries1=89,
+            expected_num_async_tasks1=2,
+            expected_num_queries2=82,
+            expected_num_async_tasks2=2,
         )
 
 
@@ -570,7 +570,7 @@ def test_import_reimport_reimport_performance_pghistory_async(self):
 
         self._import_reimport_performance(
             expected_num_queries1=1191,
-            expected_num_async_tasks1=6,
+            expected_num_async_tasks1=7,
             expected_num_queries2=716,
             expected_num_async_tasks2=17,
             expected_num_queries3=346,
@@ -593,8 +593,8 @@ def test_import_reimport_reimport_performance_pghistory_no_async(self):
         testuser.usercontactinfo.save()
 
         self._import_reimport_performance(
-            expected_num_queries1=1200,
-            expected_num_async_tasks1=6,
+            expected_num_queries1=1208,
+            expected_num_async_tasks1=7,
             expected_num_queries2=725,
             expected_num_async_tasks2=17,
             expected_num_queries3=355,
@@ -618,8 +618,8 @@ def test_import_reimport_reimport_performance_pghistory_no_async_with_product_gr
         self.system_settings(enable_product_grade=True)
 
         self._import_reimport_performance(
-            expected_num_queries1=1210,
-            expected_num_async_tasks1=8,
+            expected_num_queries1=1218,
+            expected_num_async_tasks1=9,
             expected_num_queries2=735,
             expected_num_async_tasks2=19,
             expected_num_queries3=359,
@@ -719,9 +719,9 @@ def test_deduplication_performance_pghistory_async(self):
 
         self._deduplication_performance(
             expected_num_queries1=1411,
-            expected_num_async_tasks1=7,
-            expected_num_queries2=1016,
-            expected_num_async_tasks2=7,
+            expected_num_async_tasks1=8,
+            expected_num_queries2=1013,
+            expected_num_async_tasks2=8,
             check_duplicates=False,  # Async mode - deduplication happens later
         )
 
@@ -738,8 +738,8 @@ def test_deduplication_performance_pghistory_no_async(self):
         testuser.usercontactinfo.save()
 
         self._deduplication_performance(
-            expected_num_queries1=1420,
-            expected_num_async_tasks1=7,
-            expected_num_queries2=1132,
-            expected_num_async_tasks2=7,
+            expected_num_queries1=1428,
+            expected_num_async_tasks1=8,
+            expected_num_queries2=1137,
+            expected_num_async_tasks2=8,
         )