feat: create notifications to subscribed users upon suggestion creation

Author: florentc

src/shared/listeners/notify_users.py

@@ -0,0 +1,76 @@
+import logging
+
+import pgpubsub
+from django.contrib.auth.models import User
+
+from shared.channels import CVEDerivationClusterProposalChannel
+from shared.models.linkage import CVEDerivationClusterProposal
+from webview.models import Notification
+
+logger = logging.getLogger(__name__)
+
+
+def create_package_subscription_notifications(
+    suggestion: CVEDerivationClusterProposal,
+) -> None:
+    """
+    Create notifications for users subscribed to packages affected by the suggestion.
+    """
+    # Extract all affected package names from the suggestion
+    affected_packages = list(
+        suggestion.derivations.values_list("attribute", flat=True).distinct()
+    )
+
+    if not affected_packages:
+        logger.debug(f"No packages found for suggestion {suggestion.pk}")
+        return
+
+    # Find users subscribed to ANY of these packages
+    subscribed_users = User.objects.filter(
+        profile__package_subscriptions__overlap=affected_packages
+    ).select_related("profile")
+
+    if not subscribed_users.exists():
+        logger.debug(f"No subscribed users found for packages: {affected_packages}")
+        return
+
+    logger.info(
+        f"Creating notifications for {subscribed_users.count()} users for CVE {suggestion.cve.cve_id}"
+    )
+
+    for user in subscribed_users:
+        # Find which of their subscribed packages are actually affected
+        user_affected_packages = [
+            pkg
+            for pkg in user.profile.package_subscriptions
+            if pkg in affected_packages
+        ]
+
+        # Create notification
+        try:
+            Notification.objects.create_for_user(
+                user=user,
+                title=f"New security suggestion affects: {', '.join(user_affected_packages)}",
+                message=f"CVE {suggestion.cve.cve_id} may affect packages you're subscribed to. "
+                f"Affected packages: {', '.join(user_affected_packages)}. ",
+            )
+            logger.debug(
+                f"Created notification for user {user.username} for packages: {user_affected_packages}"
+            )
+        except Exception as e:
+            logger.error(f"Failed to create notification for user {user.username}: {e}")
+
+
+@pgpubsub.post_insert_listener(CVEDerivationClusterProposalChannel)
+def notify_subscribed_users_following_suggestion_insert(
+    old: CVEDerivationClusterProposal, new: CVEDerivationClusterProposal
+) -> None:
+    """
+    Notify users subscribed to packages when a new security suggestion is created.
+    """
+    try:
+        create_package_subscription_notifications(new)
+    except Exception as e:
+        logger.error(
+            f"Failed to create package subscription notifications for suggestion {new.pk}: {e}"
+        )

src/webview/tests/test_subscriptions.py

@@ -3,6 +3,17 @@
 from django.test import Client, TestCase
 from django.urls import reverse
 
+from shared.listeners.automatic_linkage import build_new_links
+from shared.listeners.notify_users import create_package_subscription_notifications
+from shared.models.cve import (
+    AffectedProduct,
+    CveRecord,
+    Description,
+    Metric,
+    Organization,
+    Version,
+)
+from shared.models.linkage import CVEDerivationClusterProposal
 from shared.models.nix_evaluation import (
     NixChannel,
     NixDerivation,
@@ -338,3 +349,54 @@ def test_user_unsubscribes_from_empty_package_name_fails_htmx(self) -> None:
         # Verify empty subscriptions in context
         self.assertIn("package_subscriptions", response.context)
         self.assertEqual(response.context["package_subscriptions"], [])
+
+    def test_user_receives_notification_for_subscribed_package_suggestion(self) -> None:
+        """Test that users receive notifications when suggestions affect their subscribed packages"""
+        # User subscribes to firefox package
+        add_url = reverse("webview:subscriptions:add")
+        self.client.post(add_url, {"package_name": "firefox"}, HTTP_HX_REQUEST="true")
+
+        # Create CVE and container - this should trigger automatic linkage and then notifications
+        assigner = Organization.objects.create(uuid=1, short_name="test_org")
+        cve_record = CveRecord.objects.create(
+            cve_id="CVE-2025-0001",
+            assigner=assigner,
+        )
+
+        description = Description.objects.create(value="Test firefox vulnerability")
+        metric = Metric.objects.create(format="cvssV3_1", raw_cvss_json={})
+        affected_product = AffectedProduct.objects.create(package_name="firefox")
+        affected_product.versions.add(
+            Version.objects.create(status=Version.Status.AFFECTED, version="120.0")
+        )
+
+        container = cve_record.container.create(
+            provider=assigner,
+            title="Firefox Security Issue",
+        )
+
+        container.affected.set([affected_product])
+        container.descriptions.set([description])
+        container.metrics.set([metric])
+
+        # Trigger the linkage and notification system manually since pgpubsub triggers won't work in tests
+        linkage_created = build_new_links(container)
+
+        if linkage_created:
+            # Get the created proposal and trigger notifications
+            suggestion = CVEDerivationClusterProposal.objects.get(cve=cve_record)
+            create_package_subscription_notifications(suggestion)
+
+        # Verify notification appears in notification center context
+        response = self.client.get(reverse("webview:notifications:center"))
+        self.assertEqual(response.status_code, 200)
+
+        # Check that notification appears in context
+        notifications = response.context["notifications"]
+        self.assertEqual(len(notifications), 1)
+
+        notification = notifications[0]
+        self.assertEqual(notification.user, self.user)
+        self.assertIn("firefox", notification.title)
+        self.assertIn("CVE-2025-0001", notification.message)
+        self.assertFalse(notification.is_read)  # Should be unread initially