Repository: getsentry/sentry
Pull request: feat(issue-platform): Add STATUS_CHANGE payload and support RESOLVED statuses
Adds a new payload_type, STATUS_CHANGE, to the issue platform and sets it up to support the RESOLVED status. We add validation to ensure the provided (status, substatus) pair is valid before updating the group status, activity, and group history. 

Fixes https://github.com/getsentry/sentry/issues/58034
FIxes https://github.com/getsentry/sentry/issues/58035

Problem statement:
Add a new payload type to the Issue Platform API
The issue platform API should be extended to support a `status_update` payload type in addition to `occurrence`. 

The `status_update` message format is as follows:
* `fingerprint`: uniquely identifier for the occurrence
* `new_status`: status string, one of { `resolved` | `archived` | `unresolved`}
* `new_substatus` [optional]:
  * one of forever, until_escalating until_condition_met for `archived`
  * one of "escalating" or "ongoing" for `unresolved`


Support updating occurrence status to `Resolved` via API
Payload:
```
payload_type: "status_update",
status_update: {
  fingerprint: <>
  new_status: "resolved"
  new_susbtatus: None
```

Expectations:
- [ ] The occurrence with the given fingerprint has the status Resolved.
- [ ] The activity for the occurrence reflects the status change.
- [ ] The GroupHistory for the occurrence reflects the status change.
- [ ] A Resolved occurrence behaves similarly to a resolved issue - no alerts, does not appear on the issue stream, is searchable with `is:resolved`.

Note that this does not currently support the other Resolved statuses, i.e. Resolved in commit, resolved in next release.

Hints from issue discussion:

Commit message: Address PR comments

Patch to review:
diff --git a/src/sentry/issues/occurrence_status_change.py b/src/sentry/issues/occurrence_status_change.py
new file mode 100644
index 00000000000000..d2a45999984214
--- /dev/null
+++ b/src/sentry/issues/occurrence_status_change.py
@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Sequence
+
+from sentry.models.group import Group, GroupStatus
+from sentry.types.activity import ActivityType
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class StatusChangeOccurrence:
+    fingerprint: Sequence[str]
+    project_id: int
+    new_status: int
+    new_substatus: int | None
+
+    def update_status_for_occurrence(self, group):
+        if group.status == self.new_status and group.substatus == self.new_substatus:
+            return
+
+        # Validate the provided status and substatus - we only allow setting a substatus for unresolved or ignored groups.
+        if self.new_status in [GroupStatus.UNRESOLVED, GroupStatus.IGNORED]:
+            if self.new_substatus is None:
+                logger.error(
+                    "group.update_status.missing_substatus",
+                    extra={
+                        "project_id": self.project_id,
+                        "fingerprint": self.fingerprint,
+                        "new_status": self.new_status,
+                        "new_substatus": self.new_substatus,
+                    },
+                )
+                return
+        else:
+            if self.new_substatus is not None:
+                logger.error(
+                    "group.update_status.invalid_substatus",
+                    extra={
+                        "project_id": self.project_id,
+                        "fingerprint": self.fingerprint,
+                        "new_status": self.new_status,
+                        "new_substatus": self.new_substatus,
+                    },
+                )
+                return
+
+        if self.new_status == GroupStatus.RESOLVED:
+            Group.objects.update_group_status(
+                groups=[group],
+                status=self.new_status,
+                substatus=self.new_substatus,
+                activity_type=ActivityType.SET_RESOLVED,
+            )
+        else:
+            raise NotImplementedError(f"Unsupported status: {self.new_status} {self.new_substatus}")
diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py
index 4cf50b5b1beb2e..cb4c4e1073c654 100644
--- a/src/sentry/issues/producer.py
+++ b/src/sentry/issues/producer.py
@@ -1,20 +1,28 @@
 from __future__ import annotations
 
+import logging
 from typing import Any, Dict, MutableMapping, Optional, cast
 
 from arroyo import Topic
 from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
 from django.conf import settings
 
+from sentry import features
 from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.occurrence_status_change import StatusChangeOccurrence
+from sentry.models.grouphash import GroupHash
+from sentry.models.project import Project
 from sentry.services.hybrid_cloud import ValueEqualityEnum
 from sentry.utils import json
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
 
+logger = logging.getLogger(__name__)
+
 
 class PayloadType(ValueEqualityEnum):
     OCCURRENCE = "occurrence"
+    STATUS_CHANGE = "status_change"
 
 
 def _get_occurrence_producer() -> KafkaProducer:
@@ -31,13 +39,14 @@ def _get_occurrence_producer() -> KafkaProducer:
 
 
 def produce_occurrence_to_kafka(
-    payload_type: PayloadType | None = PayloadType.OCCURRENCE,
+    payload_type: PayloadType = PayloadType.OCCURRENCE,
     occurrence: IssueOccurrence | None = None,
+    status_change: StatusChangeOccurrence | None = None,
     event_data: Optional[Dict[str, Any]] = None,
 ) -> None:
     if payload_type == PayloadType.OCCURRENCE:
         if not occurrence:
-            raise ValueError("Occurrence must be provided")
+            raise ValueError("occurrence must be provided")
         if event_data and occurrence.event_id != event_data["event_id"]:
             raise ValueError("Event id on occurrence and event_data must be the same")
         if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
@@ -58,5 +67,32 @@ def produce_occurrence_to_kafka(
             payload_data["event"] = event_data
         payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
         _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
+    elif payload_type == PayloadType.STATUS_CHANGE:
+        if not status_change:
+            raise ValueError("status_change must be provided")
+
+        organization = Project.objects.get(id=status_change.project_id).organization
+        if not features.has("organizations:issue-platform-api-crons-sd", organization):
+            return
+
+        grouphash = (
+            GroupHash.objects.filter(
+                project=status_change.project_id,
+                hash=status_change.fingerprint[0],
+            )
+            .select_related("group")
+            .first()
+        )
+        if not grouphash:
+            logger.error(
+                "grouphash.not_found",
+                extra={
+                    "project_id": status_change.project_id,
+                    "fingerprint": status_change.fingerprint,
+                },
+            )
+            return
+
+        status_change.update_status_for_occurrence(grouphash.group)
     else:
         raise NotImplementedError(f"Unknown payload type {payload_type}")
diff --git a/tests/sentry/issues/test_producer.py b/tests/sentry/issues/test_producer.py
index df0dcbddefcd02..19f2f9557a6ef3 100644
--- a/tests/sentry/issues/test_producer.py
+++ b/tests/sentry/issues/test_producer.py
@@ -1,12 +1,22 @@
 import uuid
+from datetime import datetime
+from unittest.mock import MagicMock, patch
 
 import pytest
 
 from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.occurrence_status_change import StatusChangeOccurrence
 from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
+from sentry.models.activity import Activity
+from sentry.models.group import GroupStatus
+from sentry.models.grouphash import GroupHash
+from sentry.models.grouphistory import GroupHistory, GroupHistoryStatus
 from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers.datetime import before_now
+from sentry.testutils.helpers.datetime import before_now, iso_format
+from sentry.testutils.helpers.features import with_feature
 from sentry.testutils.skips import requires_snuba
+from sentry.types.activity import ActivityType
+from sentry.types.group import GroupSubStatus
 from sentry.utils.samples import load_data
 from tests.sentry.issues.test_utils import OccurrenceTestMixin
 
@@ -54,11 +64,170 @@ def test_with_only_occurrence(self) -> None:
         assert stored_occurrence
         assert occurrence.event_id == stored_occurrence.event_id
 
+    @with_feature("organizations:issue-platform-api-crons-sd")
+    def test_with_status_change(self):
+        event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "oh no",
+                "timestamp": iso_format(datetime.now()),
+                "fingerprint": ["group-1"],
+            },
+            project_id=self.project.id,
+        )
+        group = event.group
+        assert group
+        group_hash = GroupHash.objects.filter(group=group, project=self.project).first()
+        initial_status = group.status
+        initial_substatus = group.substatus
+
+        status_change_resolve = StatusChangeOccurrence(
+            fingerprint=[group_hash.hash],
+            project_id=group.project_id,
+            new_status=initial_status,
+            new_substatus=initial_substatus,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=status_change_resolve,
+        )
+        group.refresh_from_db()
+        assert group.status == initial_status
+        assert group.substatus == initial_substatus
+
+        assert not Activity.objects.filter(group=group).exists()
+        assert not GroupHistory.objects.filter(group=group).exists()
+
+        status_change_resolve = StatusChangeOccurrence(
+            fingerprint=[group_hash.hash],
+            project_id=group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=None,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=status_change_resolve,
+        )
+        group.refresh_from_db()
+        assert group.status == GroupStatus.RESOLVED
+        assert group.substatus is None
+
+        assert Activity.objects.filter(group=group, type=ActivityType.SET_RESOLVED.value).exists()
+        assert GroupHistory.objects.filter(group=group, status=GroupHistoryStatus.RESOLVED).exists()
+
+    @with_feature("organizations:issue-platform-api-crons-sd")
+    @patch("sentry.issues.occurrence_status_change.logger.error")
+    def test_with_invalid_status_change(self, mock_logger_error: MagicMock) -> None:
+        event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "oh no",
+                "timestamp": iso_format(datetime.now()),
+                "fingerprint": ["group-2"],
+            },
+            project_id=self.project.id,
+        )
+        group = event.group
+        assert group
+        group_hash = GroupHash.objects.filter(group=group, project=self.project).first()
+        initial_status = group.status
+        initial_substatus = group.substatus
+
+        bad_status_change_resolve = StatusChangeOccurrence(
+            fingerprint=[group_hash.hash],
+            project_id=group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=GroupSubStatus.FOREVER,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change_resolve,
+        )
+        group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "group.update_status.invalid_substatus",
+            extra={
+                "project_id": group.project_id,
+                "fingerprint": [group_hash.hash],
+                "new_status": GroupStatus.RESOLVED,
+                "new_substatus": GroupSubStatus.FOREVER,
+            },
+        )
+        assert group.status == initial_status
+        assert group.substatus == initial_substatus
+
+        bad_status_change_ignored = StatusChangeOccurrence(
+            fingerprint=[group_hash.hash],
+            project_id=group.project_id,
+            new_status=GroupStatus.IGNORED,
+            new_substatus=None,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change_ignored,
+        )
+        group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "group.update_status.missing_substatus",
+            extra={
+                "project_id": group.project_id,
+                "fingerprint": [group_hash.hash],
+                "new_status": GroupStatus.IGNORED,
+                "new_substatus": None,
+            },
+        )
+        assert group.status == initial_status
+        assert group.substatus == initial_substatus
+
+    @with_feature("organizations:issue-platform-api-crons-sd")
     def test_with_invalid_payloads(self) -> None:
-        with pytest.raises(ValueError):
+        with pytest.raises(ValueError, match="occurrence must be provided"):
             produce_occurrence_to_kafka(
                 payload_type=PayloadType.OCCURRENCE,
             )
 
-        with pytest.raises(NotImplementedError):
+        with pytest.raises(ValueError, match="status_change must be provided"):
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.STATUS_CHANGE,
+            )
+
+        with pytest.raises(NotImplementedError, match="Unknown payload type invalid"):
             produce_occurrence_to_kafka(payload_type="invalid")  # type: ignore
+
+    @patch("sentry.issues.producer.logger.error")
+    @with_feature("organizations:issue-platform-api-crons-sd")
+    def test_invalid_hashes(self, mock_logger_error) -> None:
+        event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "oh no",
+                "timestamp": iso_format(datetime.now()),
+                "fingerprint": ["group-2"],
+            },
+            project_id=self.project.id,
+        )
+        group = event.group
+        assert group
+        initial_status = group.status
+        initial_substatus = group.substatus
+
+        bad_status_change_resolve = StatusChangeOccurrence(
+            fingerprint=["wronghash"],
+            project_id=group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=GroupSubStatus.FOREVER,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change_resolve,
+        )
+        group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "grouphash.not_found",
+            extra={
+                "project_id": group.project_id,
+                "fingerprint": ["wronghash"],
+            },
+        )
+        assert group.status == initial_status
+        assert group.substatus == initial_substatus


Code review comments:
diff_hunk: @@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Sequence
+
+from sentry.models.group import Group, GroupStatus
+from sentry.types.activity import ActivityType
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class StatusChangeOccurrence:
path: src/sentry/issues/occurrence_status_change.py
text: @user1:
I'm not totally sure we should have occurrence in the name here. An occurrence is "an occurrence of an issue", basically an event. But we're changing the status of an issue, so I'm wondering if something like
`PlatformStatusChange`, `IssueStatusChange` or `StatusChangeMessage` make more sense.

diff_hunk: @@ -58,5 +67,32 @@ def produce_occurrence_to_kafka(
             payload_data["event"] = event_data
         payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
         _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
+    elif payload_type == PayloadType.STATUS_CHANGE:
+        if not status_change:
+            raise ValueError("status_change must be provided")
+
+        organization = Project.objects.get(id=status_change.project_id).organization
+        if not features.has("organizations:issue-platform-api-crons-sd", organization):
+            return
+
+        grouphash = (
+            GroupHash.objects.filter(
+                project=status_change.project_id,
+                hash=status_change.fingerprint[0],
+            )
+            .select_related("group")
+            .first()
+        )
+        if not grouphash:
+            logger.error(
+                "grouphash.not_found",
+                extra={
+                    "project_id": status_change.project_id,
+                    "fingerprint": status_change.fingerprint,
+                },
+            )
+            return
+
+        status_change.update_status_for_occurrence(grouphash.group)
path: src/sentry/issues/producer.py
text: @user1:
Is the idea that we'll send this message to Kafka in a later pr? This api should be accessible via Kafka so that any team that wants to change the status of an issue can do so, regardless of whether they're in the Sentry codebase.

@author:
Added that change here!