fix(aci): Try our best in Project.transfer_to (#103704)

Detectors are project-associated, so they work automatically, but
DataSources, DataConditionGroups, and Workflows need to be updated.
Note that some of these (Workflows in particular) are org scoped, so
transferring may not make sense, so we handle the simple case and hope
for the best.

Author: kcons

src/sentry/models/project.py

@@ -9,7 +9,7 @@
 import sentry_sdk
 from django.conf import settings
 from django.db import IntegrityError, models, router, transaction
-from django.db.models import Q, QuerySet, Subquery
+from django.db.models import Count, Q, QuerySet, Subquery
 from django.db.models.signals import pre_delete
 from django.utils import timezone
 from django.utils.http import urlencode
@@ -55,6 +55,7 @@
 if TYPE_CHECKING:
     from sentry.models.options.project_option import ProjectOptionManager
     from sentry.models.options.project_template_option import ProjectTemplateOptionManager
+    from sentry.models.organization import Organization
     from sentry.users.models.user import User
 
 # NOTE:
@@ -499,7 +500,7 @@ def get_audit_log_data(self):
     def get_full_name(self):
         return self.slug
 
-    def transfer_to(self, organization):
+    def transfer_to(self, organization: Organization) -> None:
         from sentry.deletions.models.scheduleddeletion import RegionScheduledDeletion
         from sentry.incidents.models.alert_rule import AlertRule
         from sentry.integrations.models.external_issue import ExternalIssue
@@ -514,6 +515,7 @@ def transfer_to(self, organization):
         from sentry.models.rule import Rule
         from sentry.monitors.models import Monitor, MonitorEnvironment, MonitorStatus
         from sentry.snuba.models import SnubaQuery
+        from sentry.workflow_engine.models import DataConditionGroup, DataSource, Detector, Workflow
 
         old_org_id = self.organization_id
         org_changed = old_org_id != organization.id
@@ -637,6 +639,97 @@ def transfer_to(self, organization):
 
         AlertRule.objects.fetch_for_project(self).update(organization=organization)
 
+        # Transfer DataSource, Workflow, and DataConditionGroup objects for Detectors attached to this project.
+        # * DataSources link detectors to their data sources (QuerySubscriptions, Monitors, etc.).
+        # * Workflows are connected to detectors and define what actions to take.
+        # * DataConditionGroups are connected to workflows (unique 1:1 via WorkflowDataConditionGroup).
+        # Since Detectors are project-scoped and their DataSources are project-specific,
+        # we need to update all related organization-scoped workflow_engine models.
+        #
+        # IMPORTANT: Workflows and DataConditionGroups can be shared across multiple projects
+        # in the same organization. We only transfer them if they're exclusively used by
+        # detectors in this project. Shared workflows remain in the original organization.
+        # There are certainly more correct ways to do this, but this should cover most cases.
+
+        detector_ids = Detector.objects.filter(project_id=self.id).values_list("id", flat=True)
+        if detector_ids:
+            # Update DataSources
+            # DataSources are 1:1 with their source (e.g., QuerySubscription) so they always transfer
+            data_source_ids = (
+                DataSource.objects.filter(detectors__id__in=detector_ids)
+                .distinct()
+                .values_list("id", flat=True)
+            )
+            DataSource.objects.filter(id__in=data_source_ids).update(
+                organization_id=organization.id
+            )
+
+            # Update Workflows connected to these detectors
+            # Only transfer workflows that are exclusively used by detectors in this project
+            all_workflow_ids = (
+                Workflow.objects.filter(detectorworkflow__detector_id__in=detector_ids)
+                .distinct()
+                .values_list("id", flat=True)
+            )
+
+            # Find workflows that are ONLY connected to detectors in this project
+            exclusive_workflow_ids = (
+                Workflow.objects.filter(id__in=all_workflow_ids)
+                .annotate(
+                    detector_count=Count("detectorworkflow__detector"),
+                    project_detector_count=Count(
+                        "detectorworkflow__detector",
+                        filter=Q(detectorworkflow__detector_id__in=detector_ids),
+                    ),
+                )
+                .filter(detector_count=models.F("project_detector_count"))
+                .values_list("id", flat=True)
+            )
+
+            Workflow.objects.filter(id__in=exclusive_workflow_ids).update(
+                organization_id=organization.id
+            )
+
+            # Update DataConditionGroups connected to the transferred workflows
+            # These are linked via WorkflowDataConditionGroup with a unique constraint on condition_group
+            workflow_condition_group_ids = (
+                DataConditionGroup.objects.filter(
+                    workflowdataconditiongroup__workflow_id__in=exclusive_workflow_ids
+                )
+                .distinct()
+                .values_list("id", flat=True)
+            )
+            DataConditionGroup.objects.filter(id__in=workflow_condition_group_ids).update(
+                organization_id=organization.id
+            )
+
+            # Update DataConditionGroups that are directly owned by detectors
+            # These are linked via Detector.workflow_condition_group (unique FK)
+            # and are exclusively owned by the detector, so they always transfer
+            detector_condition_group_ids = (
+                Detector.objects.filter(
+                    id__in=detector_ids, workflow_condition_group_id__isnull=False
+                )
+                .values_list("workflow_condition_group_id", flat=True)
+                .distinct()
+            )
+            DataConditionGroup.objects.filter(id__in=detector_condition_group_ids).update(
+                organization_id=organization.id
+            )
+
+            # Update DataConditionGroups used as when_condition_group in transferred workflows
+            # DataConditionGroups are never shared, so transfer all when_condition_groups
+            when_condition_group_ids = (
+                Workflow.objects.filter(
+                    id__in=exclusive_workflow_ids, when_condition_group_id__isnull=False
+                )
+                .values_list("when_condition_group_id", flat=True)
+                .distinct()
+            )
+            DataConditionGroup.objects.filter(id__in=when_condition_group_ids).update(
+                organization_id=organization.id
+            )
+
         # Manually move over external issues to the new org
         linked_groups = GroupLink.objects.filter(project_id=self.id).values_list(
             "linked_id", flat=True

tests/sentry/models/test_project.py

@@ -36,7 +36,7 @@
 from sentry.types.actor import Actor
 from sentry.users.models.user import User
 from sentry.users.models.user_option import UserOption
-from sentry.workflow_engine.models import Detector
+from sentry.workflow_engine.models import Detector, DetectorWorkflow
 from sentry.workflow_engine.typings.grouptype import IssueStreamGroupType
 
 
@@ -482,6 +482,164 @@ def test_project_detectors(self) -> None:
         assert Detector.objects.filter(project=project, type=ErrorGroupType.slug).count() == 1
         assert Detector.objects.filter(project=project, type=IssueStreamGroupType.slug).count() == 1
 
+    def test_transfer_to_organization_with_metric_issue_detector_and_workflow(self) -> None:
+        from_org = self.create_organization()
+        team = self.create_team(organization=from_org)
+        to_org = self.create_organization()
+        project = self.create_project(teams=[team])
+
+        detector = self.create_detector(project=project)
+        data_source = self.create_data_source(organization=from_org)
+        data_source.detectors.add(detector)
+        workflow = self.create_workflow(organization=from_org)
+        self.create_detector_workflow(detector=detector, workflow=workflow)
+
+        project.transfer_to(organization=to_org)
+
+        project.refresh_from_db()
+        detector.refresh_from_db()
+        data_source.refresh_from_db()
+        workflow.refresh_from_db()
+
+        assert project.organization_id == to_org.id
+        assert detector.project_id == project.id
+        assert data_source.organization_id == to_org.id
+        assert workflow.organization_id == to_org.id
+        assert DetectorWorkflow.objects.filter(detector=detector, workflow=workflow).exists()
+
+    def test_transfer_to_organization_with_workflow_data_condition_groups(self) -> None:
+        from_org = self.create_organization()
+        team = self.create_team(organization=from_org)
+        to_org = self.create_organization()
+        project = self.create_project(teams=[team])
+
+        detector = self.create_detector(project=project)
+        workflow = self.create_workflow(organization=from_org)
+        self.create_detector_workflow(detector=detector, workflow=workflow)
+        condition_group = self.create_data_condition_group(organization=from_org)
+        self.create_workflow_data_condition_group(
+            workflow=workflow, condition_group=condition_group
+        )
+
+        project.transfer_to(organization=to_org)
+
+        project.refresh_from_db()
+        detector.refresh_from_db()
+        workflow.refresh_from_db()
+        condition_group.refresh_from_db()
+
+        assert project.organization_id == to_org.id
+        assert detector.project_id == project.id
+        assert workflow.organization_id == to_org.id
+        assert condition_group.organization_id == to_org.id
+        wdcg = condition_group.workflowdataconditiongroup_set.first()
+        assert wdcg is not None
+        assert wdcg.workflow_id == workflow.id
+
+    def test_transfer_to_organization_does_not_transfer_shared_workflows(self) -> None:
+        from_org = self.create_organization()
+        team = self.create_team(organization=from_org)
+        to_org = self.create_organization()
+
+        project_a = self.create_project(teams=[team], name="Project A")
+        project_b = self.create_project(teams=[team], organization=from_org, name="Project B")
+
+        detector_a = self.create_detector(project=project_a)
+        detector_b = self.create_detector(project=project_b)
+
+        shared_workflow = self.create_workflow(organization=from_org, name="Shared Workflow")
+        self.create_detector_workflow(detector=detector_a, workflow=shared_workflow)
+        self.create_detector_workflow(detector=detector_b, workflow=shared_workflow)
+
+        exclusive_workflow = self.create_workflow(organization=from_org, name="Exclusive Workflow")
+        self.create_detector_workflow(detector=detector_a, workflow=exclusive_workflow)
+
+        shared_dcg = self.create_data_condition_group(organization=from_org)
+        self.create_workflow_data_condition_group(
+            workflow=shared_workflow, condition_group=shared_dcg
+        )
+
+        exclusive_dcg = self.create_data_condition_group(organization=from_org)
+        self.create_workflow_data_condition_group(
+            workflow=exclusive_workflow, condition_group=exclusive_dcg
+        )
+
+        project_a.transfer_to(organization=to_org)
+
+        project_a.refresh_from_db()
+        project_b.refresh_from_db()
+        detector_a.refresh_from_db()
+        detector_b.refresh_from_db()
+        shared_workflow.refresh_from_db()
+        exclusive_workflow.refresh_from_db()
+        shared_dcg.refresh_from_db()
+        exclusive_dcg.refresh_from_db()
+
+        assert project_a.organization_id == to_org.id
+        assert project_b.organization_id == from_org.id
+        assert detector_a.project_id == project_a.id
+        assert detector_b.project_id == project_b.id
+        assert shared_workflow.organization_id == from_org.id
+        assert exclusive_workflow.organization_id == to_org.id
+        assert shared_dcg.organization_id == from_org.id
+        assert exclusive_dcg.organization_id == to_org.id
+        assert DetectorWorkflow.objects.filter(
+            detector=detector_a, workflow=shared_workflow
+        ).exists()
+        assert DetectorWorkflow.objects.filter(
+            detector=detector_b, workflow=shared_workflow
+        ).exists()
+        assert DetectorWorkflow.objects.filter(
+            detector=detector_a, workflow=exclusive_workflow
+        ).exists()
+
+    def test_transfer_to_organization_with_detector_workflow_condition_group(self) -> None:
+        from_org = self.create_organization()
+        team = self.create_team(organization=from_org)
+        to_org = self.create_organization()
+        project = self.create_project(teams=[team])
+
+        detector = self.create_detector(project=project)
+        workflow_condition_group = self.create_data_condition_group(organization=from_org)
+        detector.workflow_condition_group = workflow_condition_group
+        detector.save()
+
+        project.transfer_to(organization=to_org)
+
+        project.refresh_from_db()
+        detector.refresh_from_db()
+        workflow_condition_group.refresh_from_db()
+
+        assert project.organization_id == to_org.id
+        assert detector.project_id == project.id
+        assert workflow_condition_group.organization_id == to_org.id
+        assert detector.workflow_condition_group_id == workflow_condition_group.id
+
+    def test_transfer_to_organization_with_workflow_when_condition_groups(self) -> None:
+        from_org = self.create_organization()
+        to_org = self.create_organization()
+        team = self.create_team(organization=from_org)
+        project = self.create_project(teams=[team])
+
+        detector = self.create_detector(project=project)
+        when_condition_group = self.create_data_condition_group(organization=from_org)
+        workflow = self.create_workflow(
+            organization=from_org, when_condition_group=when_condition_group
+        )
+        self.create_detector_workflow(detector=detector, workflow=workflow)
+
+        project.transfer_to(organization=to_org)
+
+        project.refresh_from_db()
+        detector.refresh_from_db()
+        workflow.refresh_from_db()
+        when_condition_group.refresh_from_db()
+
+        assert project.organization_id == to_org.id
+        assert detector.project_id == project.id
+        assert workflow.organization_id == to_org.id
+        assert when_condition_group.organization_id == to_org.id
+
 
 class ProjectOptionsTests(TestCase):
     """