feat: add personal webhook notification backend

engine/apps/api/views/features.py

@@ -27,6 +27,7 @@ class Feature(enum.StrEnum):
     LABELS = "labels"
     GOOGLE_OAUTH2 = "google_oauth2"
     SERVICE_DEPENDENCIES = "service_dependencies"
+    PERSONAL_WEBHOOK = "personal_webhook"
 
 
 class FeaturesAPIView(APIView):
@@ -76,4 +77,7 @@ def _get_enabled_features(self, request):
         if settings.FEATURE_SERVICE_DEPENDENCIES_ENABLED:
             enabled_features.append(Feature.SERVICE_DEPENDENCIES)
 
+        if settings.FEATURE_PERSONAL_WEBHOOK_ENABLED:
+            enabled_features.append(Feature.PERSONAL_WEBHOOK)
+
         return enabled_features

engine/apps/webhooks/backend.py

@@ -0,0 +1,35 @@
+import typing
+
+from django.core.exceptions import ObjectDoesNotExist
+
+from apps.base.messaging import BaseMessagingBackend
+
+if typing.TYPE_CHECKING:
+    from apps.alerts.models import AlertGroup
+    from apps.base.models import UserNotificationPolicy
+    from apps.user_management.models import User
+
+
+class PersonalWebhookBackend(BaseMessagingBackend):
+    backend_id = "WEBHOOK"
+    label = "Webhook"
+    short_label = "Webhook"
+    available_for_use = True
+
+    def serialize_user(self, user: "User"):
+        try:
+            personal_webhook = user.personal_webhook
+        except ObjectDoesNotExist:
+            return {"name": None}
+        return {"name": personal_webhook.webhook.name}
+
+    def notify_user(
+        self, user: "User", alert_group: "AlertGroup", notification_policy: typing.Optional["UserNotificationPolicy"]
+    ):
+        from apps.webhooks.tasks import notify_user_async
+
+        notify_user_async.delay(
+            user_pk=user.pk,
+            alert_group_pk=alert_group.pk,
+            notification_policy_pk=notification_policy.pk if notification_policy else None,
+        )

engine/apps/webhooks/migrations/0018_alter_webhook_trigger_type_and_more.py

@@ -0,0 +1,35 @@
+# Generated by Django 4.2.16 on 2025-01-27 18:46
+
+from django.db import migrations, models
+import django.db.models.deletion
+import mirage.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('user_management', '0029_remove_organization_general_log_channel_id_db'),
+        ('webhooks', '0017_alter_webhook_trigger_type_and_more'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='webhook',
+            name='trigger_type',
+            field=models.IntegerField(choices=[(0, 'Manual or escalation step'), (1, 'Alert Group Created'), (2, 'Acknowledged'), (3, 'Resolved'), (4, 'Silenced'), (5, 'Unsilenced'), (6, 'Unresolved'), (7, 'Unacknowledged'), (8, 'Status change'), (9, 'Personal notification')], default=0, null=True),
+        ),
+        migrations.AlterField(
+            model_name='webhookresponse',
+            name='trigger_type',
+            field=models.IntegerField(choices=[(0, 'Manual or escalation step'), (1, 'Alert Group Created'), (2, 'Acknowledged'), (3, 'Resolved'), (4, 'Silenced'), (5, 'Unsilenced'), (6, 'Unresolved'), (7, 'Unacknowledged'), (8, 'Status change'), (9, 'Personal notification')]),
+        ),
+        migrations.CreateModel(
+            name='PersonalNotificationWebhook',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('additional_context_data', mirage.fields.EncryptedTextField(null=True)),
+                ('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='personal_webhook', to='user_management.user')),
+                ('webhook', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='personal_channels', to='webhooks.webhook')),
+            ],
+        ),
+    ]

engine/apps/webhooks/models/__init__.py

@@ -1 +1 @@
-from .webhook import Webhook, WebhookResponse  # noqa: F401
+from .webhook import PersonalNotificationWebhook, Webhook, WebhookResponse  # noqa: F401

engine/apps/webhooks/models/webhook.py

@@ -88,7 +88,8 @@ class Webhook(models.Model):
         TRIGGER_UNRESOLVE,
         TRIGGER_UNACKNOWLEDGE,
         TRIGGER_STATUS_CHANGE,
-    ) = range(9)
+        TRIGGER_PERSONAL_NOTIFICATION,
+    ) = range(10)
 
     # Must be the same order as previous
     TRIGGER_TYPES = (
@@ -101,6 +102,7 @@ class Webhook(models.Model):
         (TRIGGER_UNRESOLVE, "Unresolved"),
         (TRIGGER_UNACKNOWLEDGE, "Unacknowledged"),
         (TRIGGER_STATUS_CHANGE, "Status change"),
+        (TRIGGER_PERSONAL_NOTIFICATION, "Personal notification"),
     )
 
     ALL_TRIGGER_TYPES = [i[0] for i in TRIGGER_TYPES]
@@ -123,6 +125,7 @@ class Webhook(models.Model):
         TRIGGER_UNRESOLVE: "unresolve",
         TRIGGER_UNACKNOWLEDGE: "unacknowledge",
         TRIGGER_STATUS_CHANGE: "status change",
+        TRIGGER_PERSONAL_NOTIFICATION: "personal notification",
     }
 
     PUBLIC_ALL_TRIGGER_TYPES = [i for i in PUBLIC_TRIGGER_TYPES_MAP.values()]
@@ -363,3 +366,27 @@ def webhook_response_post_save(sender, instance, created, *args, **kwargs):
     source_alert_receive_channel = instance.webhook.get_source_alert_receive_channel()
     if source_alert_receive_channel and hasattr(source_alert_receive_channel.config, "on_webhook_response_created"):
         source_alert_receive_channel.config.on_webhook_response_created(instance, source_alert_receive_channel)
+
+
+class PersonalNotificationWebhook(models.Model):
+    user = models.OneToOneField(
+        "user_management.User",
+        on_delete=models.CASCADE,
+        related_name="personal_webhook",
+    )
+    webhook = models.ForeignKey(
+        "webhooks.Webhook",
+        on_delete=models.CASCADE,
+        related_name="personal_channels",
+    )
+    # only visible to owner
+    additional_context_data = mirage_fields.EncryptedTextField(null=True)
+
+    @property
+    def context_data(self):
+        return json.loads(self.additional_context_data) if self.additional_context_data else {}
+
+    @context_data.setter
+    def context_data(self, value):
+        self.additional_context_data = json.dumps(value) if value else None
+        self.save(update_fields=["additional_context_data"])

engine/apps/webhooks/tasks/__init__.py

@@ -1,2 +1,3 @@
 from .alert_group_status import alert_group_created, alert_group_status_change  # noqa: F401
+from .notify_user import notify_user_async  # noqa: F401
 from .trigger_webhook import execute_webhook, send_webhook_event  # noqa: F401

engine/apps/webhooks/tasks/notify_user.py

@@ -0,0 +1,67 @@
+from celery.utils.log import get_task_logger
+from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
+
+from apps.alerts.models import AlertGroup
+from apps.user_management.models import User
+from apps.webhooks.models import Webhook
+from common.custom_celery_tasks import shared_dedicated_queue_retry_task
+
+MAX_RETRIES = 1 if settings.DEBUG else 10
+logger = get_task_logger(__name__)
+
+
+@shared_dedicated_queue_retry_task(autoretry_for=(Exception,), retry_backoff=True, max_retries=MAX_RETRIES)
+def notify_user_async(user_pk, alert_group_pk, notification_policy_pk):
+    # imported here to avoid circular import error
+    from apps.base.models import UserNotificationPolicy, UserNotificationPolicyLogRecord
+    from apps.webhooks.tasks import execute_webhook
+
+    try:
+        user = User.objects.get(pk=user_pk)
+    except User.DoesNotExist:
+        logger.warning(f"User {user_pk} does not exist")
+        return
+
+    try:
+        alert_group = AlertGroup.objects.get(pk=alert_group_pk)
+    except AlertGroup.DoesNotExist:
+        logger.warning(f"Alert group {alert_group_pk} does not exist")
+        return
+
+    try:
+        notification_policy = UserNotificationPolicy.objects.get(pk=notification_policy_pk)
+    except UserNotificationPolicy.DoesNotExist:
+        logger.warning(f"User notification policy {notification_policy_pk} does not exist")
+        return
+
+    try:
+        personal_webhook = user.personal_webhook
+    except ObjectDoesNotExist:
+        logger.warning(f"Personal webhook is not set for user {user_pk}")
+        # record log notification error
+        UserNotificationPolicyLogRecord.objects.create(
+            author=user,
+            type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_FAILED,
+            notification_policy=notification_policy,
+            alert_group=alert_group,
+            notification_step=notification_policy.step,
+            notification_channel=notification_policy.notify_by,
+        )
+        return
+
+    # trigger webhook via task
+    execute_webhook.apply_async(
+        (personal_webhook.pk, alert_group.pk, user.pk, notification_policy.pk),
+        kwargs={"trigger_type": Webhook.TRIGGER_PERSONAL_NOTIFICATION},
+    )
+
+    # record log notification success
+    UserNotificationPolicyLogRecord.objects.create(
+        author=user,
+        type=UserNotificationPolicyLogRecord.TYPE_PERSONAL_NOTIFICATION_SUCCESS,
+        notification_policy=notification_policy,
+        alert_group=alert_group,
+        notification_step=notification_policy.step,
+        notification_channel=notification_policy.notify_by,
+    )

engine/apps/webhooks/tasks/trigger_webhook.py

@@ -46,6 +46,7 @@
     Webhook.TRIGGER_MANUAL: "escalation",
     Webhook.TRIGGER_UNACKNOWLEDGE: "unacknowledge",
     Webhook.TRIGGER_STATUS_CHANGE: "status change",
+    Webhook.TRIGGER_PERSONAL_NOTIFICATION: "personal notification",
 }
 
 
@@ -107,9 +108,14 @@ def _build_payload(
     elif payload_trigger_type == Webhook.TRIGGER_SILENCE:
         event["time"] = _isoformat_date(alert_group.silenced_at)
         event["until"] = _isoformat_date(alert_group.silenced_until)
-    elif payload_trigger_type == Webhook.TRIGGER_MANUAL:
+    elif payload_trigger_type in (Webhook.TRIGGER_MANUAL, Webhook.TRIGGER_PERSONAL_NOTIFICATION):
         event["time"] = _isoformat_date(timezone.now())
 
+    # if this is a personal notification triggered webhook, event will include additional user data
+    if payload_trigger_type == Webhook.TRIGGER_PERSONAL_NOTIFICATION:
+        user_context_data = user.personal_webhook.context_data if user.personal_webhook else {}
+        event["user"] = user_context_data
+
     # include latest response data per webhook in the event input data
     # exclude past responses from webhook being executed
     responses_data = {}
@@ -179,6 +185,9 @@ def make_request(
                 status["request_data"] = json.dumps(request_kwargs["json"])
             else:
                 status["request_data"] = request_kwargs.get("data")
+            if webhook.trigger_type == Webhook.TRIGGER_PERSONAL_NOTIFICATION:
+                # mask data for personal webhooks
+                status["request_data"] = WEBHOOK_FIELD_PLACEHOLDER
             response = webhook.make_request(status["url"], request_kwargs)
             status["status_code"] = response.status_code
             content_length = len(response.content)

engine/apps/webhooks/tests/factories.py

@@ -1,7 +1,7 @@
 import factory
 import pytz
 
-from apps.webhooks.models import Webhook, WebhookResponse
+from apps.webhooks.models import PersonalNotificationWebhook, Webhook, WebhookResponse
 from common.utils import UniqueFaker
 
 
@@ -13,6 +13,11 @@ class Meta:
         model = Webhook
 
 
+class PersonalNotificationWebhookFactory(factory.DjangoModelFactory):
+    class Meta:
+        model = PersonalNotificationWebhook
+
+
 class WebhookResponseFactory(factory.DjangoModelFactory):
     timestamp = factory.Faker("date_time", tzinfo=pytz.UTC)
 

engine/apps/webhooks/tests/test_backend.py

@@ -0,0 +1,70 @@
+from unittest.mock import patch
+
+import pytest
+from django.conf import settings
+
+from apps.base.models import UserNotificationPolicy
+from apps.webhooks.backend import PersonalWebhookBackend
+from apps.webhooks.models import Webhook
+
+
+@pytest.mark.django_db
+def test_serialize_user(
+    make_organization, make_user_for_organization, make_custom_webhook, make_personal_notification_webhook
+):
+    organization = make_organization()
+    user = make_user_for_organization(organization)
+
+    backend = PersonalWebhookBackend()
+
+    # by default, there is no personal webhook set
+    assert backend.serialize_user(user) == {"name": None}
+
+    # set personal webhook
+    webhook = make_custom_webhook(
+        organization=organization,
+        trigger_type=Webhook.TRIGGER_PERSONAL_NOTIFICATION,
+    )
+    make_personal_notification_webhook(user=user, webhook=webhook)
+
+    assert backend.serialize_user(user) == {"name": user.personal_webhook.webhook.name}
+
+
+@pytest.mark.django_db
+def test_notify_user_triggers_task(
+    make_organization,
+    make_user_for_organization,
+    make_user_notification_policy,
+    make_alert_receive_channel,
+    make_alert_group,
+    make_custom_webhook,
+    make_personal_notification_webhook,
+):
+    organization = make_organization()
+    user = make_user_for_organization(organization)
+    alert_receive_channel = make_alert_receive_channel(organization)
+    alert_group = make_alert_group(alert_receive_channel)
+
+    backend = PersonalWebhookBackend()
+    # set personal webhook
+    webhook = make_custom_webhook(
+        organization=organization,
+        trigger_type=Webhook.TRIGGER_PERSONAL_NOTIFICATION,
+    )
+    make_personal_notification_webhook(user=user, webhook=webhook)
+
+    notification_policy = make_user_notification_policy(
+        user,
+        UserNotificationPolicy.Step.NOTIFY,
+        notify_by=settings.PERSONAL_WEBHOOK_BACKEND_ID,
+        important=False,
+    )
+
+    with patch("apps.webhooks.tasks.notify_user_async") as mock_notify_user_async:
+        backend.notify_user(user, alert_group, notification_policy)
+
+    mock_notify_user_async.delay.assert_called_once_with(
+        user_pk=user.pk,
+        alert_group_pk=alert_group.pk,
+        notification_policy_pk=notification_policy.pk,
+    )