diff --git a/migrations_lockfile.txt b/migrations_lockfile.txt index 862967fbdccf0d..614aae9710ebf4 100644 --- a/migrations_lockfile.txt +++ b/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 diff --git a/src/sentry/workflow_engine/migrations/0106_fix_email_action_fallthrough_type.py b/src/sentry/workflow_engine/migrations/0106_fix_email_action_fallthrough_type.py new file mode 100644 index 00000000000000..3ec73dffea7767 --- /dev/null +++ b/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"]}, + ) + ] diff --git a/tests/sentry/workflow_engine/migrations/test_0106_fix_email_action_fallthrough_type.py b/tests/sentry/workflow_engine/migrations/test_0106_fix_email_action_fallthrough_type.py new file mode 100644 index 00000000000000..c2aeb64d49baaf --- /dev/null +++ b/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