feat(aci): Make Workflow.when_condition_group unique

migrations_lockfile.txt

@@ -39,4 +39,4 @@ tempest: 0003_use_encrypted_char_field
 
 uptime: 0049_cleanup_failed_safe_deletes
 
-workflow_engine: 0102_cleanup_failed_safe_deletes
+workflow_engine: 0103_add_unique_constraint

src/sentry/workflow_engine/migrations/0103_add_unique_constraint.py

@@ -0,0 +1,76 @@
+# Generated by Django 5.2.8 on 2025-11-20 21:39
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+import sentry.db.models.fields.foreignkey
+from sentry.new_migrations.migrations import CheckedMigration
+
+
+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 = [
+        ("sentry", "1007_cleanup_failed_safe_deletes"),
+        ("workflow_engine", "0102_cleanup_failed_safe_deletes"),
+    ]
+
+    operations = [
+        migrations.SeparateDatabaseAndState(
+            database_operations=[
+                # First add the unique constraint (creates a unique index)
+                migrations.AddConstraint(
+                    model_name="workflow",
+                    constraint=models.UniqueConstraint(
+                        fields=("when_condition_group_id",),
+                        name="workflow_engine_workflow_when_condition_group_id_11d9ba05_uniq",
+                    ),
+                ),
+                # Then drop the old regular index (db_index=False)
+                migrations.AlterField(
+                    model_name="workflow",
+                    name="when_condition_group",
+                    field=sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        blank=True,
+                        db_index=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="workflow_engine.dataconditiongroup",
+                    ),
+                ),
+            ],
+            state_operations=[
+                # Update Django's state to reflect the final model state
+                migrations.AlterField(
+                    model_name="workflow",
+                    name="when_condition_group",
+                    field=sentry.db.models.fields.foreignkey.FlexibleForeignKey(
+                        blank=True,
+                        db_index=False,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="workflow_engine.dataconditiongroup",
+                    ),
+                ),
+                migrations.AddConstraint(
+                    model_name="workflow",
+                    constraint=models.UniqueConstraint(
+                        fields=("when_condition_group_id",),
+                        name="workflow_engine_workflow_when_condition_group_id_11d9ba05_uniq",
+                    ),
+                ),
+            ],
+        ),
+    ]

src/sentry/workflow_engine/models/workflow.py

@@ -69,7 +69,7 @@ class Workflow(DefaultFieldsModel, OwnerModel, JSONConfigBase):
 
     # Required as the 'when' condition for the workflow, this evaluates states emitted from the detectors
     when_condition_group = FlexibleForeignKey(
-        "workflow_engine.DataConditionGroup", null=True, blank=True
+        "workflow_engine.DataConditionGroup", null=True, blank=True, db_index=False
     )
 
     environment = FlexibleForeignKey("sentry.Environment", null=True, blank=True)
@@ -99,6 +99,12 @@ class Workflow(DefaultFieldsModel, OwnerModel, JSONConfigBase):
     class Meta:
         app_label = "workflow_engine"
         db_table = "workflow_engine_workflow"
+        constraints = [
+            models.UniqueConstraint(
+                fields=["when_condition_group_id"],
+                name="workflow_engine_workflow_when_condition_group_id_11d9ba05_uniq",
+            ),
+        ]
 
     def get_audit_log_data(self) -> dict[str, Any]:
         return {"name": self.name}

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -189,18 +189,19 @@ def test_same_environment_only(self) -> None:
             workflow=non_matching_env_workflow,
         )
 
-        dcg = self.create_data_condition_group()
+        matching_dcg = self.create_data_condition_group()
         matching_env_workflow = self.create_workflow(
-            when_condition_group=dcg,
+            when_condition_group=matching_dcg,
             environment=env,
         )
         self.create_detector_workflow(
             detector=self.error_detector,
             workflow=matching_env_workflow,
         )
 
+        mismatched_dcg = self.create_data_condition_group()
         mismatched_env_workflow = self.create_workflow(
-            when_condition_group=dcg, environment=other_env
+            when_condition_group=mismatched_dcg, environment=other_env
         )
         self.create_detector_workflow(
             detector=self.error_detector,