Workflow Engine's process_data_sources pipeline method (#78526)

## Description
📜 [Tech
Spec](https://www.notion.so/sentry/Alerts-Create-Issues-9114a498098143178839d584c293fe75?pvs=4#0f513ed0ba46424fa79cd0ba0788c6f2)

- Adds `DataPacket` as a Protocol, so we can enforce a base set of
attributes
- Adds `process_data_sources` method
- Adds telemetry
  - Tracking by data packet type
  - adds a span to the DB queries so we can ensure performanec.

Author: saponifi3d

src/sentry/workflow_engine/models/__init__.py

@@ -3,6 +3,7 @@
     "DataCondition",
     "DataConditionGroup",
     "DataConditionGroupAction",
+    "DataPacket",
     "DataSource",
     "DataSourceDetector",
     "Detector",
@@ -16,7 +17,7 @@
 from .data_condition import DataCondition
 from .data_condition_group import DataConditionGroup
 from .data_condition_group_action import DataConditionGroupAction
-from .data_source import DataSource
+from .data_source import DataPacket, DataSource
 from .data_source_detector import DataSourceDetector
 from .detector import Detector
 from .detector_state import DetectorState

src/sentry/workflow_engine/models/data_source.py

@@ -1,3 +1,5 @@
+from typing import Protocol
+
 from django.db import models
 
 from sentry.backup.scopes import RelocationScope
@@ -10,12 +12,17 @@
 from sentry.workflow_engine.models.data_source_detector import DataSourceDetector
 
 
+class DataPacket(Protocol):
+    query_id: int
+
+
 @region_silo_model
 class DataSource(DefaultFieldsModel):
     __relocation_scope__ = RelocationScope.Organization
 
     class Type(models.IntegerChoices):
         SNUBA_QUERY_SUBSCRIPTION = 1
+        SNUBA_QUERY = 2
 
     organization = FlexibleForeignKey("sentry.Organization")
     query_id = BoundedBigIntegerField()

src/sentry/workflow_engine/processors/__init__.py

@@ -0,0 +1,5 @@
+__all__ = [
+    "process_data_sources",
+]
+
+from .data_source import process_data_sources

src/sentry/workflow_engine/processors/data_source.py

@@ -0,0 +1,45 @@
+import logging
+
+import sentry_sdk
+from django.db.models import Prefetch
+
+from sentry.utils import metrics
+from sentry.workflow_engine.models import DataPacket, DataSource, Detector
+
+logger = logging.getLogger("sentry.workflow_engine.process_data_source")
+
+
+def process_data_sources(
+    data_packets: list[DataPacket], query_type: DataSource.Type = DataSource.Type.SNUBA_QUERY
+) -> list[tuple[DataPacket, list[Detector]]]:
+    metrics.incr("sentry.workflow_engine.process_data_sources", tags={"query_type": query_type})
+
+    data_packet_ids = {packet.query_id for packet in data_packets}
+
+    # Fetch all data sources and associated detectors for the given data packets
+    with sentry_sdk.start_span(op="sentry.workflow_engine.process_data_sources.fetch_data_sources"):
+        data_sources = DataSource.objects.filter(
+            query_id__in=data_packet_ids, type=query_type
+        ).prefetch_related(Prefetch("detectors"))
+
+    # Build a lookup dict for query_id to detectors
+    query_id_to_detectors = {ds.query_id: list(ds.detectors.all()) for ds in data_sources}
+
+    # Create the result tuples
+    result = []
+    for packet in data_packets:
+        detectors = query_id_to_detectors.get(packet.query_id)
+
+        if detectors:
+            data_packet_tuple = (packet, detectors)
+            result.append(data_packet_tuple)
+        else:
+            logger.warning(
+                "No detectors found", extra={"query_id": packet.query_id, "query_type": query_type}
+            )
+            metrics.incr(
+                "sentry.workflow_engine.process_data_sources.no_detectors",
+                tags={"query_type": query_type},
+            )
+
+    return result

tests/sentry/workflow_engine/models/test_workflow.py

@@ -1,5 +1,5 @@
 from sentry.testutils.cases import TestCase
-from sentry.workflow_engine.models.workflow import Workflow
+from sentry.workflow_engine.models import Workflow
 
 
 class WorkflowTest(TestCase):

tests/sentry/workflow_engine/processors/test_data_sources.py

@@ -0,0 +1,95 @@
+from sentry.snuba.models import SnubaQuery
+from sentry.testutils.cases import TestCase
+from sentry.workflow_engine.models import DataSource
+from sentry.workflow_engine.processors import process_data_sources
+
+
+class TestProcessDataSources(TestCase):
+    def create_snuba_query(self, **kwargs):
+        return SnubaQuery.objects.create(
+            type=SnubaQuery.Type.ERROR.value,
+            dataset="events",
+            aggregate="count()",
+            time_window=60,
+            resolution=60,
+            **kwargs,
+        )
+
+    def setUp(self):
+        self.query = self.create_snuba_query()
+        self.query_two = self.create_snuba_query()
+
+        self.detector_one = self.create_detector(name="test_detector1")
+        self.detector_two = self.create_detector(name="test_detector2")
+
+        self.ds1 = self.create_data_source(query_id=self.query.id, type=DataSource.Type.SNUBA_QUERY)
+        self.ds1.detectors.set([self.detector_one])
+
+        self.ds2 = self.create_data_source(
+            query_id=self.query_two.id, type=DataSource.Type.SNUBA_QUERY
+        )
+        self.ds2.detectors.set([self.detector_two])
+
+        self.packet = self.query
+        self.packet_two = self.query_two
+
+        # turn a query into a data packet, "simulating" the result from the snuba query
+        self.packet.query_id = self.query.id
+        self.packet_two.query_id = self.query_two.id
+
+        self.data_packets = [self.query, self.query_two]
+
+    def test_single_data_packet(self):
+        self.data_packets = [self.query]
+        assert process_data_sources(self.data_packets, DataSource.Type.SNUBA_QUERY) == [
+            (self.query, [self.detector_one])
+        ]
+
+    def test_multiple_data_packets(self):
+        assert process_data_sources(self.data_packets, DataSource.Type.SNUBA_QUERY) == [
+            (self.query, [self.detector_one]),
+            (self.query_two, [self.detector_two]),
+        ]
+
+    def test_multiple_detectors(self):
+        self.detector_three = self.create_detector(name="test_detector3")
+        self.detector_four = self.create_detector(name="test_detector4")
+        self.detector_five = self.create_detector(name="test_detector5")
+        self.detector_five = self.create_detector(name="test_detector5")
+
+        self.ds2.detectors.add(self.detector_three)
+        self.ds2.detectors.add(self.detector_four)
+        self.ds2.detectors.add(self.detector_five)
+
+        assert process_data_sources(self.data_packets, DataSource.Type.SNUBA_QUERY) == [
+            (self.query, [self.detector_one]),
+            (
+                self.query_two,
+                [self.detector_two, self.detector_three, self.detector_four, self.detector_five],
+            ),
+        ]
+
+    def test_no_results(self):
+        self.ds1.detectors.clear()
+        self.ds2.detectors.clear()
+
+        assert process_data_sources(self.data_packets, DataSource.Type.SNUBA_QUERY) == []
+
+    def test_different_data_packet_type__no_results(self):
+        assert (
+            process_data_sources(self.data_packets, DataSource.Type.SNUBA_QUERY_SUBSCRIPTION) == []
+        )
+
+    def test_different_data_packet_type__with_results(self):
+        self.ds1.type = DataSource.Type.SNUBA_QUERY_SUBSCRIPTION
+        self.ds1.save()
+
+        self.ds2.type = DataSource.Type.SNUBA_QUERY_SUBSCRIPTION
+        self.ds2.save()
+
+        assert process_data_sources(
+            self.data_packets, DataSource.Type.SNUBA_QUERY_SUBSCRIPTION
+        ) == [
+            (self.query, [self.detector_one]),
+            (self.query_two, [self.detector_two]),
+        ]