feat(workflow_engine): Schedule Deletion for Workflow (#85665)

# Description
Fix scheduled deletions for the `Workflow` model - this will now ensure
that all the subsequent models are cleaned up correctly.

src/sentry/deletions/__init__.py

@@ -111,7 +111,7 @@ def load_defaults(manager: DeletionTaskManager) -> None:
     from sentry.sentry_apps.models.sentry_app_installation_token import SentryAppInstallationToken
     from sentry.sentry_apps.models.servicehook import ServiceHook
     from sentry.snuba import models as snuba_models
-    from sentry.workflow_engine.models import DataSource, Detector
+    from sentry.workflow_engine.models import DataSource, Detector, Workflow
 
     from . import defaults
 
@@ -182,6 +182,7 @@ def load_defaults(manager: DeletionTaskManager) -> None:
     manager.register(models.ArtifactBundle, defaults.ArtifactBundleDeletionTask)
     manager.register(models.Rule, defaults.RuleDeletionTask)
     manager.register(RuleFireHistory, defaults.RuleFireHistoryDeletionTask)
+    manager.register(Workflow, defaults.WorkflowDeletionTask)
 
 
 _default_manager = None

src/sentry/deletions/defaults/__init__.py

@@ -30,3 +30,4 @@
 from .sentry_app_installation_token import *  # noqa: F401,F403
 from .service_hook import *  # noqa: F401,F403
 from .team import *  # noqa: F401,F403
+from .workflow import *  # noqa: F401,F403

src/sentry/deletions/defaults/workflow.py

@@ -0,0 +1,27 @@
+from sentry.deletions.base import BaseRelation, ModelDeletionTask, ModelRelation
+from sentry.workflow_engine.models import Action, DataConditionGroup, Workflow
+
+
+class WorkflowDeletionTask(ModelDeletionTask[Workflow]):
+    def get_child_relations(self, instance: Workflow) -> list[BaseRelation]:
+        model_relations: list[BaseRelation] = []
+
+        action_filter_ids = DataConditionGroup.objects.filter(
+            workflowdataconditiongroup__workflow_id=instance.id
+        ).values_list("id", flat=True)
+
+        if action_filter_ids:
+            model_relations.append(
+                ModelRelation(
+                    Action, {"dataconditiongroupaction__condition_group_id__in": action_filter_ids}
+                )
+            )
+
+            model_relations.append(ModelRelation(DataConditionGroup, {"id__in": action_filter_ids}))
+
+        if instance.when_condition_group:
+            model_relations.append(
+                ModelRelation(DataConditionGroup, {"id": instance.when_condition_group.id})
+            )
+
+        return model_relations

tests/sentry/deletions/test_workflow.py

@@ -0,0 +1,82 @@
+import pytest
+
+from sentry.deletions.tasks.scheduled import run_scheduled_deletions
+from sentry.testutils.factories import Factories
+from sentry.testutils.helpers import TaskRunner
+from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
+from sentry.testutils.pytest.fixtures import django_db_all
+
+
+@django_db_all
+class TestDeleteWorkflow(HybridCloudTestMixin):
+    def tasks(self):
+        return TaskRunner()
+
+    @pytest.fixture(autouse=True)
+    def setUp(self):
+        self.organization = Factories.create_organization()
+        self.project = Factories.create_project(organization=self.organization)
+
+        self.workflow = Factories.create_workflow()
+        self.workflow_trigger = Factories.create_data_condition_group(
+            organization=self.organization
+        )
+        self.workflow.when_condition_group = self.workflow_trigger
+        self.workflow.save()
+
+        self.action_filter = Factories.create_data_condition_group(organization=self.organization)
+        self.action = Factories.create_action()
+        self.action_and_filter = Factories.create_data_condition_group_action(
+            condition_group=self.action_filter,
+            action=self.action,
+        )
+
+        self.second_action = Factories.create_action()
+        self.second_action_and_filter = Factories.create_data_condition_group_action(
+            condition_group=self.action_filter,
+            action=self.second_action,
+        )
+
+        self.workflow_actions = Factories.create_workflow_data_condition_group(
+            workflow=self.workflow,
+            condition_group=self.action_filter,
+        )
+
+        self.trigger_condition = Factories.create_data_condition(
+            condition_group=self.workflow_trigger,
+            comparison=1,
+            condition_result=True,
+        )
+
+        self.action_condition = Factories.create_data_condition(
+            condition_group=self.action_filter,
+            comparison=1,
+            condition_result=True,
+        )
+
+    @pytest.mark.parametrize(
+        "instance_attr",
+        [
+            "workflow",
+            "action",
+            "second_action",
+            "workflow_trigger",
+            "action_filter",
+            "action_and_filter",
+            "second_action_and_filter",
+            "workflow_actions",
+            "trigger_condition",
+            "action_condition",
+        ],
+    )
+    def test_delete_workflow(self, instance_attr):
+        instance = getattr(self, instance_attr)
+        instance_id = instance.id
+        cls = instance.__class__
+
+        self.ScheduledDeletion.schedule(instance=self.workflow, days=0)
+
+        with self.tasks():
+            run_scheduled_deletions()
+
+        assert not cls.objects.filter(id=instance_id).exists()