feat(metric-issues): Configure workflow notifications by group type

src/sentry/issues/grouptype.py

@@ -174,6 +174,9 @@ class GroupType:
     notification_config: NotificationConfig = NotificationConfig()
     detector_handler: type[DetectorHandler] | None = None
     detector_validator: type[BaseGroupTypeDetectorValidator] | None = None
+    # Controls whether status change (i.e. resolved, regressed) workflow notifications are enabled.
+    # Defaults to true to maintain the default workflow notification behavior as it exists for error group types.
+    enable_status_change_workflow_notifications: bool = True
     detector_config_schema: ClassVar[dict[str, Any]] = {}
 
     def __init_subclass__(cls: type[GroupType], **kwargs: Any) -> None:
@@ -636,6 +639,7 @@ class MetricIssuePOC(GroupType):
     default_priority = PriorityLevel.HIGH
     enable_auto_resolve = False
     enable_escalation_detection = False
+    enable_status_change_workflow_notifications = False
 
 
 def should_create_group(

src/sentry/models/activity.py

@@ -23,8 +23,9 @@
 )
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.db.models.manager.base import BaseManager
+from sentry.issues.grouptype import get_group_type_by_type_id
 from sentry.tasks import activity
-from sentry.types.activity import CHOICES, ActivityType
+from sentry.types.activity import CHOICES, STATUS_CHANGE_ACTIVITY_TYPES, ActivityType
 from sentry.types.group import PriorityLevel
 
 if TYPE_CHECKING:
@@ -191,6 +192,18 @@ def delete(self, *args, **kwargs):
                 )
 
     def send_notification(self):
+        if self.group:
+            group_type = get_group_type_by_type_id(self.group.type)
+            has_status_change_notifications = group_type.enable_status_change_workflow_notifications
+            is_status_change = self.type in {
+                activity.value for activity in STATUS_CHANGE_ACTIVITY_TYPES
+            }
+
+            # Skip sending the activity notification if the group type does not
+            # support status change workflow notifications
+            if is_status_change and not has_status_change_notifications:
+                return
+
         activity.send_activity_notifications.delay(self.id)
 
 

src/sentry/testutils/factories.py

@@ -1003,6 +1003,9 @@ def store_event(
     @staticmethod
     @assume_test_silo_mode(SiloMode.REGION)
     def create_group(project, **kwargs):
+        from sentry.models.group import GroupStatus
+        from sentry.types.group import GroupSubStatus
+
         kwargs.setdefault("message", "Hello world")
         kwargs.setdefault("data", {})
         if "type" not in kwargs["data"]:
@@ -1012,6 +1015,10 @@ def create_group(project, **kwargs):
         if "metadata" in kwargs:
             metadata = kwargs.pop("metadata")
             kwargs["data"].setdefault("metadata", {}).update(metadata)
+        if "status" not in kwargs:
+            kwargs["status"] = GroupStatus.UNRESOLVED
+            kwargs["substatus"] = GroupSubStatus.NEW
+
         return Group.objects.create(project=project, **kwargs)
 
     @staticmethod

src/sentry/types/activity.py

@@ -67,3 +67,16 @@ class ActivityType(Enum):
         ActivityType.DELETED_ATTACHMENT,  # 27
     ]
 )
+
+
+STATUS_CHANGE_ACTIVITY_TYPES = (
+    ActivityType.SET_RESOLVED,
+    ActivityType.SET_UNRESOLVED,
+    ActivityType.SET_IGNORED,
+    ActivityType.SET_REGRESSION,
+    ActivityType.SET_RESOLVED_IN_RELEASE,
+    ActivityType.SET_RESOLVED_BY_AGE,
+    ActivityType.SET_RESOLVED_IN_COMMIT,
+    ActivityType.SET_RESOLVED_IN_PULL_REQUEST,
+    ActivityType.SET_ESCALATING,
+)

tests/sentry/issues/test_status_change.py

@@ -106,7 +106,7 @@ def test_unresolve_ignored_issue(self, issue_unignored: Any) -> None:
             self.projects,
             self.project_lookup,
             acting_user=self.user,
-            is_bulk=True,
+            is_bulk=False,
             status_details={},
             new_status=GroupStatus.UNRESOLVED,
             new_substatus=GroupSubStatus.ONGOING,
@@ -131,7 +131,7 @@ def test_unresolve_resolved_issue(self, issue_unresolved: Any) -> None:
             acting_user=self.user,
             new_status=GroupStatus.UNRESOLVED,
             new_substatus=GroupSubStatus.ONGOING,
-            is_bulk=True,
+            is_bulk=False,
             status_details={},
             sender=self,
         )
@@ -154,7 +154,7 @@ def test_ignore_new_issue(self, issue_ignored: Any) -> None:
             acting_user=self.user,
             new_status=GroupStatus.IGNORED,
             new_substatus=None,
-            is_bulk=True,
+            is_bulk=False,
             status_details={"ignoreDuration": 30},
             sender=self,
         )
@@ -177,7 +177,7 @@ def test_ignore_until_escalating(self, issue_ignored: Any) -> None:
             acting_user=self.user,
             new_status=GroupStatus.IGNORED,
             new_substatus=None,
-            is_bulk=True,
+            is_bulk=False,
             status_details={"ignoreUntilEscalating": True},
             sender=self,
         )

tests/sentry/models/test_activity.py

@@ -1,6 +1,8 @@
 import logging
+from unittest.mock import MagicMock, patch
 
 from sentry.event_manager import EventManager
+from sentry.issues.grouptype import MetricIssuePOC
 from sentry.models.activity import Activity
 from sentry.testutils.cases import TestCase
 from sentry.types.activity import ActivityType
@@ -319,3 +321,26 @@ def test_get_activities_for_group_flip_flop(self):
         for pair in chunked(act_for_group[:-1], 2):
             assert pair[0].type == ActivityType.SET_IGNORED.value
             assert pair[1].type == ActivityType.SET_UNRESOLVED.value
+
+    @patch("sentry.tasks.activity.send_activity_notifications.delay")
+    def test_skips_status_change_notifications_if_disabled(
+        self, mock_send_activity_notifications: MagicMock
+    ):
+        project = self.create_project(name="test_activities_group")
+        group = self.create_group(project)
+
+        # Create an activity that would normally trigger a notification
+        activity = Activity.objects.create_group_activity(
+            group=group, type=ActivityType.SET_UNRESOLVED, data=None, send_notification=True
+        )
+
+        mock_send_activity_notifications.assert_called_once_with(activity.id)
+        mock_send_activity_notifications.reset_mock()
+
+        group.type = MetricIssuePOC.type_id
+        group.save()
+        _ = Activity.objects.create_group_activity(
+            group=group, type=ActivityType.SET_RESOLVED, data=None, send_notification=True
+        )
+
+        mock_send_activity_notifications.assert_not_called()