fix(workflow-engine): Link cron monitor detectors to shadow alert workflows (#103919)

When creating a cron monitor with alert notifications via the old APIs,
a "shadow" alert rule is created which generates a Workflow, but the
DetectorWorkflow link between the cron Detector and the Workflow was
not being created.

Changes:
- Modified IssueAlertMigrator._create_detector_lookups() to find cron
  detectors by extracting monitor.slug from rule conditions and querying
  for the associated detector via DataSource
- Removed conditional check that was skipping
_connect_default_detectors()
  for CRON_MONITOR rules, allowing detector linking to happen
- Used in_test_hide_transaction_boundary() to safely query across
databases

Part of [NEW-593: Cron Monitor Alerts are not linked to monitor when
created via the old
flow](https://linear.app/getsentry/issue/NEW-593/cron-monitor-alerts-are-not-linked-to-monitor-when-created-via-the-old)

Author: evanpurkhiser

src/sentry/workflow_engine/migration_helpers/issue_alert_migration.py

@@ -2,9 +2,12 @@
 from typing import Any
 
 from sentry.constants import ObjectStatus
+from sentry.db.postgres.transactions import in_test_hide_transaction_boundary
 from sentry.grouping.grouptype import ErrorGroupType
 from sentry.models.rule import Rule, RuleSource
 from sentry.models.rulesnooze import RuleSnooze
+from sentry.monitors.models import Monitor
+from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
 from sentry.rules.conditions.event_frequency import EventUniqueUserFrequencyConditionWithConditions
 from sentry.rules.conditions.every_event import EveryEventCondition
 from sentry.rules.processing.processor import split_conditions_and_filters
@@ -76,7 +79,32 @@ def run(self) -> Workflow:
 
     def _create_detector_lookups(self) -> list[Detector | None]:
         if self.rule.source == RuleSource.CRON_MONITOR:
-            return [None, None]
+            # Find the cron detector that was created before the rule
+            monitor_slug = None
+            for condition in self.data.get("conditions", []):
+                if condition.get("key") == "monitor.slug":
+                    monitor_slug = condition.get("value")
+                    break
+
+            if not monitor_slug:
+                return [None]
+
+            try:
+                with in_test_hide_transaction_boundary():
+                    monitor = Monitor.objects.get(
+                        slug=monitor_slug,
+                        organization_id=self.organization.id,
+                    )
+                    detector = Detector.objects.get(
+                        datasource__type=DATA_SOURCE_CRON_MONITOR,
+                        datasource__source_id=str(monitor.id),
+                        datasource__organization_id=self.organization.id,
+                    )
+                    return [detector]
+            except (Monitor.DoesNotExist, Detector.DoesNotExist):
+                pass
+
+            return [None]
 
         if self.is_dry_run:
             error_detector = Detector.objects.filter(
@@ -247,8 +275,7 @@ def _create_workflow_and_lookup(
         else:
             workflow = Workflow.objects.create(**kwargs)
             workflow.update(date_added=self.rule.date_added)
-            if not self.rule.source == RuleSource.CRON_MONITOR:
-                self._connect_default_detectors(workflow=workflow)
+            self._connect_default_detectors(workflow=workflow)
             AlertRuleWorkflow.objects.create(rule_id=self.rule.id, workflow=workflow)
 
         return workflow

tests/sentry/workflow_engine/migration_helpers/test_issue_alert_migration.py

@@ -7,6 +7,8 @@
 from sentry.grouping.grouptype import ErrorGroupType
 from sentry.models.rule import Rule, RuleSource
 from sentry.models.rulesnooze import RuleSnooze
+from sentry.monitors.models import Monitor, ScheduleType
+from sentry.monitors.utils import ensure_cron_detector, get_detector_for_monitor
 from sentry.rules.age import AgeComparisonType
 from sentry.rules.conditions.event_frequency import (
     ComparisonType,
@@ -439,6 +441,41 @@ def test_run__cron_rule(self) -> None:
         assert AlertRuleWorkflow.objects.filter(rule_id=self.issue_alert.id).exists()
         assert not DetectorWorkflow.objects.filter(workflow=workflow).exists()
 
+    def test_run__cron_rule_with_monitor(self) -> None:
+        """
+        Cron rule WITH monitor.slug filter should be connected to the cron detector
+        """
+        monitor = Monitor.objects.create(
+            organization_id=self.organization.id,
+            project_id=self.project.id,
+            name="Test Monitor",
+            slug="test-monitor",
+            config={"schedule_type": ScheduleType.CRONTAB, "schedule": "0 * * * *"},
+        )
+        ensure_cron_detector(monitor)
+
+        # Update rule to be cron monitor source with monitor.slug filter
+        self.issue_alert.source = RuleSource.CRON_MONITOR
+        self.issue_alert.data["conditions"].append(
+            {
+                "id": "sentry.rules.filters.tagged_event.TaggedEventFilter",
+                "key": "monitor.slug",
+                "match": "eq",
+                "value": "test-monitor",
+            }
+        )
+        self.issue_alert.save()
+
+        workflow = IssueAlertMigrator(self.issue_alert, self.user.id).run()
+
+        # Verify workflow created
+        assert AlertRuleWorkflow.objects.filter(rule_id=self.issue_alert.id).exists()
+
+        # Verify detector is linked to workflow
+        detector = get_detector_for_monitor(monitor)
+        assert detector is not None
+        assert DetectorWorkflow.objects.filter(detector=detector, workflow=workflow).exists()
+
     def test_dry_run(self) -> None:
         IssueAlertMigrator(self.issue_alert, self.user.id, is_dry_run=True).run()