feat(ecosystem): Implement cross-system issue synchronization

src/sentry/integrations/mixins/issues.py

@@ -12,6 +12,7 @@
 from sentry.eventstore.models import GroupEvent
 from sentry.integrations.base import IntegrationInstallation
 from sentry.integrations.models.external_issue import ExternalIssue
+from sentry.integrations.services.assignment_source import AssignmentSource
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.tasks.sync_status_inbound import (
     sync_status_inbound as sync_status_inbound_task,
@@ -62,7 +63,7 @@ def from_resolve_unresolve(
 
 
 class IssueBasicIntegration(IntegrationInstallation, ABC):
-    def should_sync(self, attribute):
+    def should_sync(self, attribute, sync_source: AssignmentSource | None = None):
         return False
 
     def get_group_title(self, group, event, **kwargs):
@@ -378,10 +379,17 @@ class IssueSyncIntegration(IssueBasicIntegration, ABC):
     outbound_assignee_key: ClassVar[str | None] = None
     inbound_assignee_key: ClassVar[str | None] = None
 
-    def should_sync(self, attribute: str) -> bool:
+    def should_sync(self, attribute: str, sync_source: AssignmentSource | None = None) -> bool:
         key = getattr(self, f"{attribute}_key", None)
         if key is None or self.org_integration is None:
             return False
+
+        # Check that the assignment source isn't this same integration in order to
+        # prevent sync-cycles from occurring. This should still allow other
+        # integrations to propagate changes outward.
+        if sync_source and sync_source.integration_id == self.org_integration.integration_id:
+            return False
+
         value: bool = self.org_integration.config.get(key, False)
         return value
 
@@ -400,7 +408,14 @@ def sync_assignee_outbound(
         raise NotImplementedError
 
     @abstractmethod
-    def sync_status_outbound(self, external_issue, is_resolved, project_id, **kwargs):
+    def sync_status_outbound(
+        self,
+        external_issue,
+        is_resolved,
+        project_id,
+        assignment_source: AssignmentSource | None = None,
+        **kwargs,
+    ):
         """
         Propagate a sentry issue's status to a linked issue's status.
         """

src/sentry/integrations/services/assignment_source.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from typing import TYPE_CHECKING, Any
+
+from django.utils import timezone
+
+if TYPE_CHECKING:
+    from sentry.integrations.models import Integration
+    from sentry.integrations.services.integration import RpcIntegration
+
+
+@dataclass(frozen=True)
+class AssignmentSource:
+    source_name: str
+    integration_id: int
+    queued: datetime = timezone.now()
+
+    @classmethod
+    def from_integration(cls, integration: Integration | RpcIntegration) -> AssignmentSource:
+        return AssignmentSource(
+            source_name=integration.name,
+            integration_id=integration.id,
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, input_dict: dict[str, Any]) -> AssignmentSource | None:
+        try:
+            return cls(**input_dict)
+        except (ValueError, TypeError):
+            return None

src/sentry/integrations/tasks/sync_assignee_outbound.py

@@ -1,6 +1,9 @@
+from typing import Any
+
 from sentry import analytics, features
 from sentry.integrations.models.external_issue import ExternalIssue
 from sentry.integrations.models.integration import Integration
+from sentry.integrations.services.assignment_source import AssignmentSource
 from sentry.integrations.services.integration import integration_service
 from sentry.models.organization import Organization
 from sentry.silo.base import SiloMode
@@ -24,7 +27,12 @@
         Organization.DoesNotExist,
     )
 )
-def sync_assignee_outbound(external_issue_id: int, user_id: int | None, assign: bool) -> None:
+def sync_assignee_outbound(
+    external_issue_id: int,
+    user_id: int | None,
+    assign: bool,
+    assignment_source_dict: dict[str, Any] | None = None,
+) -> None:
     # Sync Sentry assignee to an external issue.
     external_issue = ExternalIssue.objects.get(id=external_issue_id)
 
@@ -42,10 +50,15 @@ def sync_assignee_outbound(external_issue_id: int, user_id: int | None, assign:
     ):
         return
 
-    if installation.should_sync("outbound_assignee"):
+    parsed_assignment_source = (
+        AssignmentSource.from_dict(assignment_source_dict) if assignment_source_dict else None
+    )
+    if installation.should_sync("outbound_assignee", parsed_assignment_source):
         # Assume unassign if None.
         user = user_service.get_user(user_id) if user_id else None
-        installation.sync_assignee_outbound(external_issue, user, assign=assign)
+        installation.sync_assignee_outbound(
+            external_issue, user, assign=assign, assignment_source=parsed_assignment_source
+        )
         analytics.record(
             "integration.issue.assignee.synced",
             provider=integration.provider,

src/sentry/integrations/utils/sync.py

@@ -5,6 +5,7 @@
 from typing import TYPE_CHECKING
 
 from sentry import features
+from sentry.integrations.services.assignment_source import AssignmentSource
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.tasks.sync_assignee_outbound import sync_assignee_outbound
 from sentry.models.group import Group
@@ -92,7 +93,11 @@ def sync_group_assignee_inbound(
 
     if not assign:
         for group in affected_groups:
-            GroupAssignee.objects.deassign(group)
+            GroupAssignee.objects.deassign(
+                group,
+                assignment_source=AssignmentSource.from_integration(integration),
+            )
+
         return affected_groups
 
     users = user_service.get_many_by_email(emails=[email], is_verified=True)
@@ -104,14 +109,23 @@ def sync_group_assignee_inbound(
         user_id = get_user_id(projects_by_user, group)
         user = users_by_id.get(user_id)
         if user:
-            GroupAssignee.objects.assign(group, user)
+            GroupAssignee.objects.assign(
+                group,
+                user,
+                assignment_source=AssignmentSource.from_integration(integration),
+            )
             groups_assigned.append(group)
         else:
             logger.info("assignee-not-found-inbound", extra=log_context)
     return groups_assigned
 
 
-def sync_group_assignee_outbound(group: Group, user_id: int | None, assign: bool = True) -> None:
+def sync_group_assignee_outbound(
+    group: Group,
+    user_id: int | None,
+    assign: bool = True,
+    assignment_source: AssignmentSource | None = None,
+) -> None:
     from sentry.models.grouplink import GroupLink
 
     external_issue_ids = GroupLink.objects.filter(
@@ -120,5 +134,12 @@ def sync_group_assignee_outbound(group: Group, user_id: int | None, assign: bool
 
     for external_issue_id in external_issue_ids:
         sync_assignee_outbound.apply_async(
-            kwargs={"external_issue_id": external_issue_id, "user_id": user_id, "assign": assign}
+            kwargs={
+                "external_issue_id": external_issue_id,
+                "user_id": user_id,
+                "assign": assign,
+                "assignment_source_dict": assignment_source.to_dict()
+                if assignment_source
+                else None,
+            }
         )

src/sentry/models/groupassignee.py

@@ -12,6 +12,7 @@
 from sentry.db.models import FlexibleForeignKey, Model, region_silo_model, sane_repr
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
 from sentry.db.models.manager.base import BaseManager
+from sentry.integrations.services.assignment_source import AssignmentSource
 from sentry.models.grouphistory import GroupHistoryStatus, record_group_history
 from sentry.models.groupowner import GroupOwner
 from sentry.models.groupsubscription import GroupSubscription
@@ -134,6 +135,7 @@ def assign(
         create_only: bool = False,
         extra: dict[str, str] | None = None,
         force_autoassign: bool = False,
+        assignment_source: AssignmentSource | None = None,
     ):
         from sentry.integrations.utils import sync_group_assignee_outbound
         from sentry.models.activity import Activity
@@ -187,7 +189,9 @@ def assign(
             if assignee_type == "user" and features.has(
                 "organizations:integrations-issue-sync", group.organization, actor=acting_user
             ):
-                sync_group_assignee_outbound(group, assigned_to.id, assign=True)
+                sync_group_assignee_outbound(
+                    group, assigned_to.id, assign=True, assignment_source=assignment_source
+                )
 
             if not created:  # aka re-assignment
                 self.remove_old_assignees(group, assignee, assigned_to_id, assignee_type)
@@ -200,6 +204,7 @@ def deassign(
         acting_user: User | RpcUser | None = None,
         assigned_to: Team | RpcUser | None = None,
         extra: dict[str, str] | None = None,
+        assignment_source: AssignmentSource | None = None,
     ) -> None:
         from sentry.integrations.utils import sync_group_assignee_outbound
         from sentry.models.activity import Activity
@@ -230,7 +235,9 @@ def deassign(
             if features.has(
                 "organizations:integrations-issue-sync", group.organization, actor=acting_user
             ):
-                sync_group_assignee_outbound(group, None, assign=False)
+                sync_group_assignee_outbound(
+                    group, None, assign=False, assignment_source=assignment_source
+                )
 
             issue_unassigned.send_robust(
                 project=group.project, group=group, user=acting_user, sender=self.__class__

tests/sentry/integrations/services/test_assignment_source.py

@@ -0,0 +1,38 @@
+from typing import Any
+
+from sentry.integrations.services.assignment_source import AssignmentSource
+from sentry.testutils.cases import TestCase
+
+
+class TestAssignmentSource(TestCase):
+    def test_from_dict_empty_array(self):
+        data: dict[str, Any] = {}
+        result = AssignmentSource.from_dict(data)
+        assert result is None
+
+    def test_from_dict_inalid_data(self):
+        data = {
+            "foo": "bar",
+        }
+
+        result = AssignmentSource.from_dict(data)
+        assert result is None
+
+    def test_from_dict_valid_data(self):
+        data = {"source_name": "foo-source", "integration_id": 123}
+
+        result = AssignmentSource.from_dict(data)
+        assert result is not None
+        assert result.source_name == "foo-source"
+        assert result.integration_id == 123
+
+    def test_to_dict(self):
+        source = AssignmentSource(
+            source_name="foo-source",
+            integration_id=123,
+        )
+
+        result = source.to_dict()
+        assert result.get("queued") is not None
+        assert result.get("source_name") == "foo-source"
+        assert result.get("integration_id") == 123

tests/sentry/models/test_groupassignee.py

@@ -4,6 +4,7 @@
 
 from sentry.integrations.example.integration import ExampleIntegration
 from sentry.integrations.models.external_issue import ExternalIssue
+from sentry.integrations.services.assignment_source import AssignmentSource
 from sentry.integrations.utils import sync_group_assignee_inbound
 from sentry.models.activity import Activity
 from sentry.models.groupassignee import GroupAssignee
@@ -148,12 +149,77 @@ def test_assignee_sync_outbound_assign(self, mock_sync_assignee_outbound):
 
         with self.feature({"organizations:integrations-issue-sync": True}):
             with self.tasks():
-                GroupAssignee.objects.assign(self.group, self.user)
+                GroupAssignee.objects.assign(
+                    self.group,
+                    self.user,
+                )
 
                 mock_sync_assignee_outbound.assert_called_with(
-                    external_issue, user_service.get_user(self.user.id), assign=True
+                    external_issue,
+                    user_service.get_user(self.user.id),
+                    assign=True,
+                    assignment_source=None,
+                )
+
+                assert GroupAssignee.objects.filter(
+                    project=self.group.project,
+                    group=self.group,
+                    user_id=self.user.id,
+                    team__isnull=True,
+                ).exists()
+
+                activity = Activity.objects.get(
+                    project=self.group.project, group=self.group, type=ActivityType.ASSIGNED.value
+                )
+
+                assert activity.data["assignee"] == str(self.user.id)
+                assert activity.data["assigneeEmail"] == self.user.email
+                assert activity.data["assigneeType"] == "user"
+
+    @mock.patch.object(ExampleIntegration, "sync_assignee_outbound")
+    def test_assignee_sync_outbound_assign_with_matching_source_integration(
+        self, mock_sync_assignee_outbound
+    ):
+        group = self.group
+        integration = self.create_integration(
+            organization=group.organization,
+            external_id="123456",
+            provider="example",
+            oi_params={
+                "config": {
+                    "sync_comments": True,
+                    "sync_status_outbound": True,
+                    "sync_status_inbound": True,
+                    "sync_assignee_outbound": True,
+                    "sync_assignee_inbound": True,
+                }
+            },
+        )
+
+        external_issue = ExternalIssue.objects.create(
+            organization_id=group.organization.id, integration_id=integration.id, key="APP-123"
+        )
+
+        GroupLink.objects.create(
+            group_id=group.id,
+            project_id=group.project_id,
+            linked_type=GroupLink.LinkedType.issue,
+            linked_id=external_issue.id,
+            relationship=GroupLink.Relationship.references,
+        )
+
+        with self.feature({"organizations:integrations-issue-sync": True}):
+            with self.tasks():
+                # Assert that we don't perform an outbound assignment if
+                # the source of the assignment is the same target integration
+                GroupAssignee.objects.assign(
+                    self.group,
+                    self.user,
+                    assignment_source=AssignmentSource.from_integration(integration),
                 )
 
+                mock_sync_assignee_outbound.assert_not_called()
+
                 assert GroupAssignee.objects.filter(
                     project=self.group.project,
                     group=self.group,
@@ -205,7 +271,9 @@ def test_assignee_sync_outbound_unassign(self, mock_sync_assignee_outbound):
         with self.feature({"organizations:integrations-issue-sync": True}):
             with self.tasks():
                 GroupAssignee.objects.deassign(self.group)
-                mock_sync_assignee_outbound.assert_called_with(external_issue, None, assign=False)
+                mock_sync_assignee_outbound.assert_called_with(
+                    external_issue, None, assign=False, assignment_source=None
+                )
 
                 assert not GroupAssignee.objects.filter(
                     project=self.group.project,