chore(aci): migration to fix fallthrough_type on email Actions

migrations_lockfile.txt

@@ -39,4 +39,4 @@ tempest: 0003_use_encrypted_char_field
 
 uptime: 0051_add_assertion
 
-workflow_engine: 0105_add_incident_identifer_index
+workflow_engine: 0106_fix_email_action_fallthrough_type

src/sentry/workflow_engine/migrations/0106_fix_email_action_fallthrough_type.py

@@ -0,0 +1,226 @@
+# Generated by Django 5.2.8 on 2026-01-22 18:25
+
+import logging
+from typing import Any
+
+from django.db import migrations
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from django.db.migrations.state import StateApps
+
+from sentry.new_migrations.migrations import CheckedMigration
+from sentry.utils.query import RangeQuerySetWrapper
+
+logger = logging.getLogger(__name__)
+
+EMAIL_ACTION_REGISTRY_ID = "sentry.mail.actions.NotifyEmailAction"
+
+# Target type mappings (from ActionTarget enum)
+TARGET_TYPE_USER = 1
+TARGET_TYPE_TEAM = 2
+TARGET_TYPE_ISSUE_OWNERS = 4
+
+# Maps Rule action targetType string to Action config target_type int
+TARGET_TYPE_STRING_TO_INT = {
+    "Member": TARGET_TYPE_USER,
+    "Team": TARGET_TYPE_TEAM,
+    "IssueOwners": TARGET_TYPE_ISSUE_OWNERS,
+}
+
+DEFAULT_FALLTHROUGH_TYPE = "ActiveMembers"
+
+
+def translate_email_action(rule_action: dict[str, Any]) -> dict[str, Any]:
+    """
+    Translate a Rule email action blob to Action model fields.
+    Returns dict with: data, config, integration_id
+    """
+    target_type_str = rule_action.get("targetType")
+    if not target_type_str:
+        raise ValueError("targetType is required for email actions")
+
+    target_type_int = TARGET_TYPE_STRING_TO_INT.get(target_type_str)
+    if target_type_int is None:
+        raise ValueError(f"Unknown targetType: {target_type_str}")
+
+    # Build config
+    config: dict[str, Any] = {
+        "target_type": target_type_int,
+        "target_display": None,
+    }
+
+    # target_identifier is only set for Member/Team, not IssueOwners
+    if target_type_str in ("Member", "Team"):
+        target_identifier = rule_action.get("targetIdentifier")
+        config["target_identifier"] = str(target_identifier) if target_identifier else None
+    else:
+        config["target_identifier"] = None
+
+    # Build data - only IssueOwners has fallthrough_type
+    data: dict[str, Any] = {}
+    if target_type_str == "IssueOwners":
+        # Check both possible keys for fallthrough type
+        fallthrough_type = rule_action.get("fallthrough_type")
+        if fallthrough_type is None:
+            fallthrough_type = rule_action.get("fallthroughType", DEFAULT_FALLTHROUGH_TYPE)
+        data["fallthrough_type"] = str(fallthrough_type)
+
+    return {
+        "data": data,
+        "config": config,
+        "integration_id": None,  # Email actions don't use integrations
+    }
+
+
+def fix_email_action_fallthrough_type(
+    apps: StateApps, schema_editor: BaseDatabaseSchemaEditor
+) -> None:
+    """
+    Fix email actions that were incorrectly migrated with missing fallthrough_type.
+
+    This migration:
+    1. Finds all Workflows that have at least 1 Action of type="email"
+    2. For each workflow, fetches the Rule through AlertRuleWorkflow
+    3. Gets the NotifyEmailAction actions from Rule.data["actions"]
+    4. Gets the email actions for the workflow
+    5. Overwrites each Action with translated data from the rule action
+    """
+    Rule = apps.get_model("sentry", "Rule")
+    Workflow = apps.get_model("workflow_engine", "Workflow")
+    Action = apps.get_model("workflow_engine", "Action")
+    AlertRuleWorkflow = apps.get_model("workflow_engine", "AlertRuleWorkflow")
+
+    # Iterate directly through workflows with email actions
+    workflows_with_email_actions = Workflow.objects.filter(
+        workflowdataconditiongroup__condition_group__dataconditiongroupaction__action__type="email"
+    ).distinct()
+
+    for workflow in RangeQuerySetWrapper(workflows_with_email_actions):
+        try:
+            # Get the Rule for this workflow through AlertRuleWorkflow
+            alert_rule_workflow = AlertRuleWorkflow.objects.filter(workflow_id=workflow.id).first()
+            if not alert_rule_workflow or not alert_rule_workflow.rule_id:
+                logger.info(
+                    "workflow.no_rule_link",
+                    extra={"workflow_id": workflow.id},
+                )
+                continue
+
+            try:
+                rule = Rule.objects.get(id=alert_rule_workflow.rule_id)
+            except Rule.DoesNotExist:
+                logger.info(
+                    "workflow.rule_not_found",
+                    extra={
+                        "workflow_id": workflow.id,
+                        "rule_id": alert_rule_workflow.rule_id,
+                    },
+                )
+                continue
+
+            # Rule email actions
+            rule_actions = rule.data.get("actions", [])
+            rule_email_actions = [
+                action for action in rule_actions if action.get("id") == EMAIL_ACTION_REGISTRY_ID
+            ]
+
+            if not rule_email_actions:
+                logger.info(
+                    "workflow.no_rule_email_actions",
+                    extra={"workflow_id": workflow.id, "rule_id": rule.id},
+                )
+                continue
+
+            # Workflow email actions
+            workflow_email_actions = list(
+                Action.objects.filter(
+                    dataconditiongroupaction__condition_group__workflowdataconditiongroup__workflow_id=workflow.id,
+                    type="email",
+                )
+                .distinct()
+                .order_by("id")
+            )
+
+            if not workflow_email_actions:
+                logger.info(
+                    "workflow.no_workflow_email_actions",
+                    extra={"workflow_id": workflow.id},
+                )
+                continue
+
+            # Translate rule email actions
+            translated_actions = []
+            for rule_action in rule_email_actions:
+                try:
+                    translated_actions.append(translate_email_action(rule_action))
+                except ValueError:
+                    logger.exception(
+                        "workflow.translate_action_error",
+                        extra={"workflow_id": workflow.id, "rule_id": rule.id},
+                    )
+                    continue
+
+            if len(translated_actions) != len(workflow_email_actions):
+                logger.warning(
+                    "workflow.action_count_mismatch",
+                    extra={
+                        "workflow_id": workflow.id,
+                        "rule_id": rule.id,
+                        "translated_count": len(translated_actions),
+                        "workflow_count": len(workflow_email_actions),
+                    },
+                )
+
+            # Overwrite each workflow action with the translated data
+            for i, workflow_action in enumerate(workflow_email_actions):
+                if i >= len(translated_actions):
+                    break
+
+                translated = translated_actions[i]
+
+                workflow_action.data = translated["data"]
+                workflow_action.config = translated["config"]
+                workflow_action.integration_id = translated["integration_id"]
+                workflow_action.save()
+
+            logger.info(
+                "workflow.actions_updated",
+                extra={
+                    "workflow_id": workflow.id,
+                    "action_ids": [action.id for action in workflow_email_actions],
+                },
+            )
+
+        except Exception:
+            logger.exception(
+                "workflow.migration_error",
+                extra={"workflow_id": workflow.id},
+            )
+            continue
+
+
+class Migration(CheckedMigration):
+    # This flag is used to mark that a migration shouldn't be automatically run in production.
+    # This should only be used for operations where it's safe to run the migration after your
+    # code has deployed. So this should not be used for most operations that alter the schema
+    # of a table.
+    # Here are some things that make sense to mark as post deployment:
+    # - Large data migrations. Typically we want these to be run manually so that they can be
+    #   monitored and not block the deploy for a long period of time while they run.
+    # - Adding indexes to large tables. Since this can take a long time, we'd generally prefer to
+    #   run this outside deployments so that we don't block them. Note that while adding an index
+    #   is a schema change, it's completely safe to run the operation after the code has deployed.
+    # Once deployed, run these manually via: https://develop.sentry.dev/database-migrations/#migration-deployment
+
+    is_post_deployment = True
+
+    dependencies = [
+        ("workflow_engine", "0105_add_incident_identifer_index"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            fix_email_action_fallthrough_type,
+            migrations.RunPython.noop,
+            hints={"tables": ["workflow_engine_action", "workflow_engine_workflow", "sentry_rule"]},
+        )
+    ]

tests/sentry/workflow_engine/migrations/test_0106_fix_email_action_fallthrough_type.py

@@ -0,0 +1,158 @@
+from sentry.testutils.cases import TestMigrations
+from sentry.workflow_engine.migration_helpers.issue_alert_migration import IssueAlertMigrator
+from sentry.workflow_engine.migration_helpers.rule_action import (
+    translate_rule_data_actions_to_notification_actions,
+)
+from sentry.workflow_engine.models import Action
+
+EMAIL_ACTION_REGISTRY_ID = "sentry.mail.actions.NotifyEmailAction"
+
+
+class FixEmailActionFallthroughTypeTest(TestMigrations):
+    migrate_from = "0105_add_incident_identifer_index"
+    migrate_to = "0106_fix_email_action_fallthrough_type"
+    app = "workflow_engine"
+
+    def setup_initial_state(self) -> None:
+        self.org = self.create_organization(name="test-org")
+        self.project = self.create_project(organization=self.org)
+
+        # Create rule with 2 email actions (different fallthroughType) + 1 plugin action
+        self.rule1 = self.create_project_rule(
+            project=self.project,
+            action_data=[
+                {
+                    "id": "sentry.mail.actions.NotifyEmailAction",
+                    "targetType": "IssueOwners",
+                    "fallthroughType": "ActiveMembers",
+                },
+                {
+                    "id": "sentry.mail.actions.NotifyEmailAction",
+                    "targetType": "IssueOwners",
+                    "fallthroughType": "NoOne",
+                },
+                {
+                    "id": "sentry.rules.actions.notify_event.NotifyEventAction",
+                },
+            ],
+            condition_data=[
+                {"id": "sentry.rules.conditions.first_seen_event.FirstSeenEventCondition"},
+            ],
+        )
+
+        # Migrate rule1 to workflow
+        self.workflow1 = IssueAlertMigrator(self.rule1).run()
+
+        # Simulate the bug: update all email actions to have wrong data/config
+        email_actions_1 = Action.objects.filter(
+            dataconditiongroupaction__condition_group__workflowdataconditiongroup__workflow_id=self.workflow1.id,
+            type="email",
+        ).order_by("id")
+
+        for action in email_actions_1:
+            # Overwrite with wrong data to simulate the bug
+            action.data.update({"fallthrough_type": "ActiveMembers"})
+            action.save()
+
+        # Store original plugin action data for comparison
+        self.plugin_action_1 = Action.objects.filter(
+            dataconditiongroupaction__condition_group__workflowdataconditiongroup__workflow_id=self.workflow1.id,
+            type="plugin",
+        ).first()
+        self.plugin_action_1_original_data = (
+            self.plugin_action_1.data.copy() if self.plugin_action_1 else {}
+        )
+        self.plugin_action_1_original_config = (
+            self.plugin_action_1.config.copy() if self.plugin_action_1 else {}
+        )
+
+        # Create another rule with a webhook action (NotifyEventServiceAction)
+        self.rule2 = self.create_project_rule(
+            project=self.project,
+            action_data=[
+                {
+                    "id": "sentry.rules.actions.notify_event_service.NotifyEventServiceAction",
+                    "service": "mail",
+                },
+            ],
+            condition_data=[
+                {"id": "sentry.rules.conditions.first_seen_event.FirstSeenEventCondition"},
+            ],
+        )
+
+        # Migrate rule2 to workflow
+        self.workflow2 = IssueAlertMigrator(self.rule2).run()
+
+        # Store original webhook action data for comparison
+        self.webhook_action = Action.objects.filter(
+            dataconditiongroupaction__condition_group__workflowdataconditiongroup__workflow_id=self.workflow2.id,
+            type="webhook",
+        ).first()
+        self.webhook_action_original_data = (
+            self.webhook_action.data.copy() if self.webhook_action else {}
+        )
+        self.webhook_action_original_config = (
+            self.webhook_action.config.copy() if self.webhook_action else {}
+        )
+
+    def test_migration(self) -> None:
+        self._test_migration_fixes_email_actions()
+        self._test_migration_does_not_change_non_email_actions()
+
+    def _test_migration_fixes_email_actions(self) -> None:
+        """
+        Verify the migration produces the same Action state as translate_rule_data_actions_to_notification_actions.
+        """
+        # Get the email actions for workflow1 after migration
+        email_actions = list(
+            Action.objects.filter(
+                dataconditiongroupaction__condition_group__workflowdataconditiongroup__workflow_id=self.workflow1.id,
+                type="email",
+            ).order_by("id")
+        )
+
+        # Get the rule's email actions and translate them using the helper function
+        rule_email_actions = [
+            action
+            for action in self.rule1.data["actions"]
+            if action.get("id") == EMAIL_ACTION_REGISTRY_ID
+        ]
+        expected_actions = translate_rule_data_actions_to_notification_actions(
+            rule_email_actions, skip_failures=True
+        )
+
+        assert len(email_actions) == len(expected_actions) == 2
+
+        # Compare each migrated action to what the helper function produces
+        for i, (migrated_action, expected_action) in enumerate(
+            zip(email_actions, expected_actions)
+        ):
+            assert migrated_action.type == expected_action.type, (
+                f"Action {i}: type mismatch - "
+                f"got {migrated_action.type}, expected {expected_action.type}"
+            )
+            assert migrated_action.data == expected_action.data, (
+                f"Action {i}: data mismatch - "
+                f"got {migrated_action.data}, expected {expected_action.data}"
+            )
+            assert migrated_action.config == expected_action.config, (
+                f"Action {i}: config mismatch - "
+                f"got {migrated_action.config}, expected {expected_action.config}"
+            )
+            assert migrated_action.integration_id == expected_action.integration_id, (
+                f"Action {i}: integration_id mismatch - "
+                f"got {migrated_action.integration_id}, expected {expected_action.integration_id}"
+            )
+
+    def _test_migration_does_not_change_non_email_actions(self) -> None:
+        # Check that the plugin action in workflow1 is unchanged
+        if self.plugin_action_1:
+            self.plugin_action_1.refresh_from_db()
+            assert self.plugin_action_1.data == self.plugin_action_1_original_data
+            assert self.plugin_action_1.config == self.plugin_action_1_original_config
+
+        # Check that the webhook action in workflow2 is unchanged
+        if self.webhook_action:
+            self.webhook_action.refresh_from_db()
+            assert self.webhook_action.data == self.webhook_action_original_data
+            assert self.webhook_action.config == self.webhook_action_original_config