feat(workflow_engine): Add registry hook for Activity creation (#93522)

## Description
Starting implementation for:
https://www.notion.so/sentry/Workflow-Status-Changes-1fa8b10e4b5d80a48acddb95d160da1f?source=copy_link#1fa8b10e4b5d80e6bb1aef39cab2c6dc

This will add a registry of handlers when an activity is created, we'll
use this in the [workflow engine to kick-off a
task](#93553) to execute
`process_workflows`.

This also adds a way to pass the detector_id through the
StatusMessageData all the way to a hook that we can invoke and read
directly from the StatusMessageData later. We are also passed the
`activity` so `workflow_engine` can trigger `.send_notification` (if
evaluated to do so).

Author: saponifi3d

src/sentry/issues/status_change_consumer.py

@@ -2,14 +2,15 @@
 
 import logging
 from collections import defaultdict
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from typing import Any
 
 from sentry_sdk.tracing import NoOpSpan, Span, Transaction
 
 from sentry.integrations.tasks.kick_off_status_syncs import kick_off_status_syncs
 from sentry.issues.escalating.escalating import manage_issue_states
 from sentry.issues.status_change_message import StatusChangeMessageData
+from sentry.models.activity import Activity
 from sentry.models.group import Group, GroupStatus
 from sentry.models.grouphash import GroupHash
 from sentry.models.groupinbox import (
@@ -23,11 +24,13 @@
 from sentry.types.activity import ActivityType
 from sentry.types.group import IGNORED_SUBSTATUS_CHOICES, GroupSubStatus
 from sentry.utils import metrics
+from sentry.utils.registry import Registry
 
 logger = logging.getLogger(__name__)
 
 
 def update_status(group: Group, status_change: StatusChangeMessageData) -> None:
+    activity_type: ActivityType | None = None
     new_status = status_change["new_status"]
     new_substatus = status_change["new_substatus"]
 
@@ -58,11 +61,12 @@ def update_status(group: Group, status_change: StatusChangeMessageData) -> None:
             return
 
     if new_status == GroupStatus.RESOLVED:
+        activity_type = ActivityType.SET_RESOLVED
         Group.objects.update_group_status(
             groups=[group],
             status=new_status,
             substatus=new_substatus,
-            activity_type=ActivityType.SET_RESOLVED,
+            activity_type=activity_type,
         )
         remove_group_from_inbox(group, action=GroupInboxRemoveAction.RESOLVED)
         kick_off_status_syncs.apply_async(
@@ -80,11 +84,12 @@ def update_status(group: Group, status_change: StatusChangeMessageData) -> None:
             )
             return
 
+        activity_type = ActivityType.SET_IGNORED
         Group.objects.update_group_status(
             groups=[group],
             status=new_status,
             substatus=new_substatus,
-            activity_type=ActivityType.SET_IGNORED,
+            activity_type=activity_type,
         )
         remove_group_from_inbox(group, action=GroupInboxRemoveAction.IGNORED)
         kick_off_status_syncs.apply_async(
@@ -136,6 +141,19 @@ def update_status(group: Group, status_change: StatusChangeMessageData) -> None:
             f"Unsupported status: {status_change['new_status']} {status_change['new_substatus']}"
         )
 
+    if activity_type is not None:
+        """
+        If we have set created an activity, then we'll also notify any registered handlers
+        that the group status has changed.
+
+        This is used to trigger the `workflow_engine` processing status changes.
+        """
+        latest_activity = Activity.objects.filter(
+            group_id=group.id, type=activity_type.value
+        ).order_by("-datetime")
+        for handler in group_status_update_registry.registrations.values():
+            handler(group, status_change, latest_activity[0])
+
 
 def get_group_from_fingerprint(project_id: int, fingerprint: Sequence[str]) -> Group | None:
     results = bulk_get_groups_from_fingerprints([(project_id, fingerprint)])
@@ -255,3 +273,7 @@ def process_status_change_message(
         update_status(group, status_change_data)
 
     return group
+
+
+GroupUpdateHandler = Callable[[Group, StatusChangeMessageData, Activity], None]
+group_status_update_registry = Registry[GroupUpdateHandler](enable_reverse_lookup=False)

src/sentry/issues/status_change_message.py

@@ -12,6 +12,7 @@ class StatusChangeMessageData(TypedDict):
     new_status: int
     new_substatus: int | None
     id: str
+    detector_id: int | None
 
 
 @dataclass(frozen=True)
@@ -20,6 +21,7 @@ class StatusChangeMessage:
     project_id: int
     new_status: int
     new_substatus: int | None
+    detector_id: int | None = None
     id: str = field(default_factory=lambda: uuid4().hex)
 
     def to_dict(
@@ -30,5 +32,6 @@ def to_dict(
             "project_id": self.project_id,
             "new_status": self.new_status,
             "new_substatus": self.new_substatus,
+            "detector_id": self.detector_id,
             "id": self.id,
         }

src/sentry/models/activity.py

@@ -98,6 +98,7 @@ def create_group_activity(
         if user_id is not None:
             activity_args["user_id"] = user_id
         activity = self.create(**activity_args)
+
         if send_notification:
             activity.send_notification()
 

src/sentry/workflow_engine/handlers/detector/stateful.py

@@ -365,6 +365,7 @@ def _create_resolve_message(self, group_key: DetectorGroupKey = None) -> StatusC
             project_id=self.detector.project_id,
             new_status=GroupStatus.RESOLVED,
             new_substatus=None,
+            detector_id=self.detector.id,
         )
 
     def _extract_value_from_packet(

tests/sentry/issues/test_status_change_consumer.py

@@ -4,7 +4,8 @@
 from unittest.mock import MagicMock, patch
 
 from sentry.issues.occurrence_consumer import _process_message
-from sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints
+from sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints, update_status
+from sentry.issues.status_change_message import StatusChangeMessageData
 from sentry.models.activity import Activity
 from sentry.models.group import Group, GroupStatus
 from sentry.models.grouphistory import GroupHistory, GroupHistoryStatus
@@ -20,13 +21,15 @@ def get_test_message_status_change(
     fingerprint: list[str] | None = None,
     new_status: int = GroupStatus.RESOLVED,
     new_substatus: int | None = None,
+    detector_id: int | None = None,
 ) -> dict[str, Any]:
     payload = {
         "project_id": project_id,
         "fingerprint": fingerprint or ["group-1"],
         "new_status": new_status,
         "new_substatus": new_substatus,
         "payload_type": "status_change",
+        "detector_id": detector_id,
     }
 
     return payload
@@ -345,3 +348,119 @@ def test_bulk_get_single_project_multiple_hash(self) -> None:
                 tuple([*other_occurrence.fingerprint, *self.occurrence.fingerprint]),
             ): other_group
         }
+
+
+class TestStatusChangeRegistry(IssueOccurrenceTestBase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.detector = self.create_detector()
+        self.group = self.create_group(
+            project=self.project,
+            status=GroupStatus.UNRESOLVED,
+            substatus=GroupSubStatus.ESCALATING,
+        )
+
+        status_change = get_test_message_status_change(
+            self.project.id,
+            new_status=GroupStatus.RESOLVED,
+            detector_id=self.detector.id,
+        )
+
+        self.message = StatusChangeMessageData(
+            id="test-id",
+            project_id=status_change["project_id"],
+            fingerprint=status_change["fingerprint"],
+            new_status=status_change["new_status"],
+            new_substatus=status_change.get("new_substatus"),
+            detector_id=status_change.get("detector_id"),
+        )
+
+    def get_latest_activity(self, activity_type: ActivityType) -> Activity:
+        latest_activity = (
+            Activity.objects.filter(group_id=self.group.id, type=activity_type.value)
+            .order_by("-datetime")
+            .first()
+        )
+
+        if latest_activity is None:
+            raise AssertionError(f"No activity found for type {activity_type}")
+
+        return latest_activity
+
+    def test_handler_is_called__resolved(self) -> None:
+        with patch(
+            "sentry.issues.status_change_consumer.group_status_update_registry",
+        ) as mock_registry:
+            mock_handler = MagicMock()
+            mock_registry.registrations = {
+                "test_status_change": mock_handler,
+            }
+
+            update_status(self.group, self.message)
+            latest_activity = self.get_latest_activity(ActivityType.SET_RESOLVED)
+
+            mock_handler.assert_called_once_with(self.group, self.message, latest_activity)
+
+    def test_handler_is_not_called__unresolved_escalating(self) -> None:
+        # There will be an issue occurrence that triggers this instead
+
+        self.message["new_status"] = GroupStatus.UNRESOLVED
+        self.message["new_substatus"] = GroupSubStatus.ESCALATING
+        with patch(
+            "sentry.issues.status_change_consumer.group_status_update_registry",
+        ) as mock_registry:
+            mock_handler = MagicMock()
+            mock_registry.registrations = {
+                "test_status_change": mock_handler,
+            }
+
+            update_status(self.group, self.message)
+            assert mock_handler.call_count == 0
+
+    def test_handler_is_called_unresolved_ongoing(self) -> None:
+        self.message["new_status"] = GroupStatus.UNRESOLVED
+        self.message["new_substatus"] = GroupSubStatus.ONGOING
+
+        with patch(
+            "sentry.issues.status_change_consumer.group_status_update_registry",
+        ) as mock_registry:
+            mock_handler = MagicMock()
+            mock_registry.registrations = {
+                "test_status_change": mock_handler,
+            }
+
+            update_status(self.group, self.message)
+            latest_activity = self.get_latest_activity(ActivityType.AUTO_SET_ONGOING)
+            mock_handler.assert_called_once_with(self.group, self.message, latest_activity)
+
+    def test_handler_is_called__unresolved_regressed(self) -> None:
+        self.message["new_status"] = GroupStatus.UNRESOLVED
+        self.message["new_substatus"] = GroupSubStatus.REGRESSED
+
+        with patch(
+            "sentry.issues.status_change_consumer.group_status_update_registry",
+        ) as mock_registry:
+            mock_handler = MagicMock()
+            mock_registry.registrations = {
+                "test_status_change": mock_handler,
+            }
+
+            update_status(self.group, self.message)
+            latest_activity = self.get_latest_activity(ActivityType.SET_REGRESSION)
+            mock_handler.assert_called_once_with(self.group, self.message, latest_activity)
+
+    def test_handler_is_called__ignored(self) -> None:
+        self.message["new_status"] = GroupStatus.IGNORED
+        self.message["new_substatus"] = GroupSubStatus.FOREVER
+
+        with patch(
+            "sentry.issues.status_change_consumer.group_status_update_registry",
+        ) as mock_registry:
+            mock_handler = MagicMock()
+            mock_registry.registrations = {
+                "test_status_change": mock_handler,
+            }
+
+            update_status(self.group, self.message)
+            latest_activity = self.get_latest_activity(ActivityType.SET_IGNORED)
+            mock_handler.assert_called_once_with(self.group, self.message, latest_activity)

tests/sentry/issues/test_utils.py

@@ -94,6 +94,7 @@ def build_statuschange_data(self, **overrides: Any) -> StatusChangeMessageData:
             "fingerprint": ["some-fingerprint"],
             "new_status": 1,
             "new_substatus": 1,
+            "detector_id": None,
         }
         kwargs.update(overrides)  # type: ignore[typeddict-item]
 

tests/sentry/workflow_engine/handlers/detector/test_stateful.py

@@ -218,6 +218,7 @@ def test_evaluate__resolve(self):
         assert evaluation_result
         assert evaluation_result.priority == DetectorPriorityLevel.OK
         assert isinstance(evaluation_result.result, StatusChangeMessage)
+        assert evaluation_result.result.detector_id == self.detector.id
 
     def test_evaluate__resolve__detector_state(self):
         self.handler.evaluate(self.data_packet)