perf(uptime): Bulk load data needed for process_detectors, and call it directly from the uptime result consumer (#96269)

This saves us an extra query reloading things. Previously we did this,
but hadn't loaded the required related objects, so it was less
efficient.

<!-- Describe your PR here. -->

Author: wedamija

src/sentry/uptime/consumers/results_consumer.py

@@ -45,13 +45,13 @@
     send_uptime_config_deletion,
     update_remote_uptime_subscription,
 )
-from sentry.uptime.types import DATA_SOURCE_UPTIME_SUBSCRIPTION, IncidentStatus, UptimeMonitorMode
+from sentry.uptime.types import IncidentStatus, UptimeMonitorMode
 from sentry.utils import metrics
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
 from sentry.workflow_engine.models.data_source import DataPacket
 from sentry.workflow_engine.models.detector import Detector
-from sentry.workflow_engine.processors.data_packet import process_data_packet
+from sentry.workflow_engine.processors import process_detectors
 
 logger = logging.getLogger(__name__)
 
@@ -310,9 +310,8 @@ def handle_active_result(
             subscription=uptime_subscription,
             metric_tags=metric_tags,
         )
-        process_data_packet(
-            DataPacket(source_id=str(uptime_subscription.id), packet=packet),
-            DATA_SOURCE_UPTIME_SUBSCRIPTION,
+        process_detectors(
+            DataPacket(source_id=str(uptime_subscription.id), packet=packet), [detector]
         )
 
         # Bail if we're doing issue creation via detectors, we don't want to
@@ -444,7 +443,7 @@ def handle_result(self, subscription: UptimeSubscription | None, result: CheckRe
         if should_run_region_checks(subscription, result):
             try_check_and_update_regions(subscription, subscription_regions)
 
-        detector = get_detector(subscription)
+        detector = get_detector(subscription, prefetch_workflow_data=True)
 
         # Nothing to do if there's an orphaned project subscription
         if not detector:

src/sentry/uptime/models.py

@@ -348,7 +348,9 @@ def get_current_instance_count(org: Organization) -> int:
         raise NotImplementedError
 
 
-def get_detector(uptime_subscription: UptimeSubscription) -> Detector | None:
+def get_detector(
+    uptime_subscription: UptimeSubscription, prefetch_workflow_data=False
+) -> Detector | None:
     """
     Fetches a workflow_engine Detector given an existing uptime_subscription.
     This is used during the transition period moving uptime to detector.
@@ -358,9 +360,15 @@ def get_detector(uptime_subscription: UptimeSubscription) -> Detector | None:
             type=DATA_SOURCE_UPTIME_SUBSCRIPTION,
             source_id=str(uptime_subscription.id),
         )
-        return Detector.objects.select_related("project", "project__organization").get(
+        qs = Detector.objects.filter(
             type=GROUP_TYPE_UPTIME_DOMAIN_CHECK_FAILURE, data_sources=data_source[:1]
         )
+        select_related = ["project", "project__organization"]
+        if prefetch_workflow_data:
+            select_related.append("workflow_condition_group")
+            qs = qs.prefetch_related("workflow_condition_group__conditions")
+        qs = qs.select_related(*select_related)
+        return qs.get()
     except Detector.DoesNotExist:
         return None