We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix typing for `sentry.rules.processor`
Many files related to post-process are skipping type checks.

* Remove the file from [sentry/pyproject.toml](https://github.com/getsentry/sentry/blob/b7810a9dca4b57afd2858903a6a9ec7ab50cdead/pyproject.toml)
* Run `mypy sentry.rules.processor` and fix the typing errors

From https://github.com/getsentry/sentry/issues/55193

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/sentry/rules/processor.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 import uuid
5 from datetime import timedelta
6 from random import randrange
7 from typing import (
8     Any,
9     Callable,
10     Collection,
11     List,
12     Mapping,
13     MutableMapping,
14     Optional,
15     Sequence,
16     Set,
17     Tuple,
18 )
19 
20 from django.core.cache import cache
21 from django.utils import timezone
22 
23 from sentry import analytics
24 from sentry.eventstore.models import GroupEvent
25 from sentry.models.environment import Environment
26 from sentry.models.grouprulestatus import GroupRuleStatus
27 from sentry.models.rule import Rule
28 from sentry.models.rulesnooze import RuleSnooze
29 from sentry.rules import EventState, history, rules
30 from sentry.rules.conditions.base import EventCondition
31 from sentry.types.rules import RuleFuture
32 from sentry.utils.hashlib import hash_values
33 from sentry.utils.safe import safe_execute
34 
35 SLOW_CONDITION_MATCHES = ["event_frequency"]
36 
37 
38 def get_match_function(match_name: str) -> Callable[..., bool] | None:
39     if match_name == "all":
40         return all
41     elif match_name == "any":
42         return any
43     elif match_name == "none":
44         return lambda bool_iter: not any(bool_iter)
45     return None
46 
47 
48 def is_condition_slow(condition: Mapping[str, str]) -> bool:
49     for slow_conditions in SLOW_CONDITION_MATCHES:
50         if slow_conditions in condition["id"]:
51             return True
52     return False
53 
54 
55 class RuleProcessor:
56     logger = logging.getLogger("sentry.rules")
57 
58     def __init__(
59         self,
60         event: GroupEvent,
61         is_new: bool,
62         is_regression: bool,
63         is_new_group_environment: bool,
64         has_reappeared: bool,
65     ) -> None:
66         self.event = event
67         self.group = event.group
68         self.project = event.project
69 
70         self.is_new = is_new
71         self.is_regression = is_regression
72         self.is_new_group_environment = is_new_group_environment
73         self.has_reappeared = has_reappeared
74 
75         self.grouped_futures: MutableMapping[
76             str, Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]
77         ] = {}
78 
79     def get_rules(self) -> Sequence[Rule]:
80         """Get all of the rules for this project from the DB (or cache)."""
81         rules_: Sequence[Rule] = Rule.get_for_project(self.project.id)
82         return rules_
83 
84     def _build_rule_status_cache_key(self, rule_id: int) -> str:
85         return "grouprulestatus:1:%s" % hash_values([self.group.id, rule_id])
86 
87     def bulk_get_rule_status(self, rules: Sequence[Rule]) -> Mapping[int, GroupRuleStatus]:
88         keys = [self._build_rule_status_cache_key(rule.id) for rule in rules]
89         cache_results: Mapping[str, GroupRuleStatus] = cache.get_many(keys)
90         missing_rule_ids: Set[int] = set()
91         rule_statuses: MutableMapping[int, GroupRuleStatus] = {}
92         for key, rule in zip(keys, rules):
93             rule_status = cache_results.get(key)
94             if not rule_status:
95                 missing_rule_ids.add(rule.id)
96             else:
97                 rule_statuses[rule.id] = rule_status
98 
99         if missing_rule_ids:
100             # If not cached, attempt to fetch status from the database
101             statuses = GroupRuleStatus.objects.filter(
102                 group=self.group, rule_id__in=missing_rule_ids
103             )
104             to_cache: List[GroupRuleStatus] = list()
105             for status in statuses:
106                 rule_statuses[status.rule_id] = status
107                 missing_rule_ids.remove(status.rule_id)
108                 to_cache.append(status)
109 
110             # We might need to create some statuses if they don't already exist
111             if missing_rule_ids:
112                 # We use `ignore_conflicts=True` here to avoid race conditions where the statuses
113                 # might be created between when we queried above and attempt to create the rows now.
114                 GroupRuleStatus.objects.bulk_create(
115                     [
116                         GroupRuleStatus(rule_id=rule_id, group=self.group, project=self.project)
117                         for rule_id in missing_rule_ids
118                     ],
119                     ignore_conflicts=True,
120                 )
121                 # Using `ignore_conflicts=True` prevents the pk from being set on the model
122                 # instances. Re-query the database to fetch the rows, they should all exist at this
123                 # point.
124                 statuses = GroupRuleStatus.objects.filter(
125                     group=self.group, rule_id__in=missing_rule_ids
126                 )
127                 for status in statuses:
128                     rule_statuses[status.rule_id] = status
129                     missing_rule_ids.remove(status.rule_id)
130                     to_cache.append(status)
131 
132                 if missing_rule_ids:
133                     # Shouldn't happen, but log just in case
134                     self.logger.error(
135                         "Failed to fetch some GroupRuleStatuses in RuleProcessor",
136                         extra={"missing_rule_ids": missing_rule_ids, "group_id": self.group.id},
137                     )
138             if to_cache:
139                 cache.set_many(
140                     {self._build_rule_status_cache_key(item.rule_id): item for item in to_cache}
141                 )
142 
143         return rule_statuses
144 
145     def condition_matches(
146         self, condition: Mapping[str, Any], state: EventState, rule: Rule
147     ) -> bool | None:
148         condition_cls = rules.get(condition["id"])
149         if condition_cls is None:
150             self.logger.warning("Unregistered condition %r", condition["id"])
151             return None
152 
153         condition_inst: EventCondition = condition_cls(self.project, data=condition, rule=rule)
154         passes: bool = safe_execute(
155             condition_inst.passes, self.event, state, _with_transaction=False
156         )
157         return passes
158 
159     def get_rule_type(self, condition: Mapping[str, Any]) -> str | None:
160         rule_cls = rules.get(condition["id"])
161         if rule_cls is None:
162             self.logger.warning("Unregistered condition or filter %r", condition["id"])
163             return None
164 
165         rule_type: str = rule_cls.rule_type
166         return rule_type
167 
168     def get_state(self) -> EventState:
169         return EventState(
170             is_new=self.is_new,
171             is_regression=self.is_regression,
172             is_new_group_environment=self.is_new_group_environment,
173             has_reappeared=self.has_reappeared,
174         )
175 
176     def apply_rule(self, rule: Rule, status: GroupRuleStatus) -> None:
177         """
178         If all conditions and filters pass, execute every action.
179 
180         :param rule: `Rule` object
181         :return: void
182         """
183         logging_details = {
184             "rule_id": rule.id,
185             "group_id": self.group.id,
186             "event_id": self.event.event_id,
187             "project_id": self.project.id,
188             "is_new": self.is_new,
189             "is_regression": self.is_regression,
190             "has_reappeared": self.has_reappeared,
191             "new_group_environment": self.is_new_group_environment,
192         }
193 
194         condition_match = rule.data.get("action_match") or Rule.DEFAULT_CONDITION_MATCH
195         filter_match = rule.data.get("filter_match") or Rule.DEFAULT_FILTER_MATCH
196         rule_condition_list = rule.data.get("conditions", ())
197         frequency = rule.data.get("frequency") or Rule.DEFAULT_FREQUENCY
198         try:
199             environment = self.event.get_environment()
200         except Environment.DoesNotExist:
201             return
202 
203         if rule.environment_id is not None and environment.id != rule.environment_id:
204             return
205 
206         now = timezone.now()
207         freq_offset = now - timedelta(minutes=frequency)
208         if status.last_active and status.last_active > freq_offset:
209             return
210 
211         state = self.get_state()
212 
213         condition_list = []
214         filter_list = []
215         for rule_cond in rule_condition_list:
216             if self.get_rule_type(rule_cond) == "condition/event":
217                 condition_list.append(rule_cond)
218             else:
219                 filter_list.append(rule_cond)
220 
221         # Sort `condition_list` so that most expensive conditions run last.
222         condition_list.sort(key=lambda condition: is_condition_slow(condition))
223 
224         for predicate_list, match, name in (
225             (filter_list, filter_match, "filter"),
226             (condition_list, condition_match, "condition"),
227         ):
228             if not predicate_list:
229                 continue
230             predicate_iter = (self.condition_matches(f, state, rule) for f in predicate_list)
231             predicate_func = get_match_function(match)
232             if predicate_func:
233                 if not predicate_func(predicate_iter):
234                     return
235             else:
236                 self.logger.error(
237                     f"Unsupported {name}_match {match!r} for rule {rule.id}",
238                     filter_match,
239                     rule.id,
240                     extra={**logging_details},
241                 )
242                 return
243 
244         updated = (
245             GroupRuleStatus.objects.filter(id=status.id)
246             .exclude(last_active__gt=freq_offset)
247             .update(last_active=now)
248         )
249 
250         if not updated:
251             return
252 
253         if randrange(10) == 0:
254             analytics.record(
255                 "issue_alert.fired",
256                 issue_id=self.group.id,
257                 project_id=rule.project.id,
258                 organization_id=rule.project.organization.id,
259                 rule_id=rule.id,
260             )
261 
262         notification_uuid = str(uuid.uuid4())
263         history.record(rule, self.group, self.event.event_id, notification_uuid)
264         self.activate_downstream_actions(rule, notification_uuid)
265 
266     def activate_downstream_actions(
267         self, rule: Rule, notification_uuid: Optional[str] = None
268     ) -> None:
269         state = self.get_state()
270         for action in rule.data.get("actions", ()):
271             action_cls = rules.get(action["id"])
272             if action_cls is None:
273                 self.logger.warning("Unregistered action %r", action["id"])
274                 continue
275 
276             action_inst = action_cls(self.project, data=action, rule=rule)
277 
278             results = safe_execute(
279                 action_inst.after,
280                 event=self.event,
281                 state=state,
282                 _with_transaction=False,
283                 notification_uuid=notification_uuid,
284             )
285             if results is None:
286                 self.logger.warning("Action %s did not return any futures", action["id"])
287                 continue
288 
289             for future in results:
290                 key = future.key if future.key is not None else future.callback
291                 rule_future = RuleFuture(rule=rule, kwargs=future.kwargs)
292 
293                 if key not in self.grouped_futures:
294                     self.grouped_futures[key] = (future.callback, [rule_future])
295                 else:
296                     self.grouped_futures[key][1].append(rule_future)
297 
298     def apply(
299         self,
300     ) -> Collection[Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]]:
301         # we should only apply rules on unresolved issues
302         if not self.event.group.is_unresolved():
303             return {}.values()
304 
305         self.grouped_futures.clear()
306         rules = self.get_rules()
307         snoozed_rules = RuleSnooze.objects.filter(rule__in=rules, user_id=None).values_list(
308             "rule", flat=True
309         )
310         rule_statuses = self.bulk_get_rule_status(rules)
311         for rule in rules:
312             if rule.id not in snoozed_rules:
313                 self.apply_rule(rule, rule_statuses[rule.id])
314 
315         return self.grouped_futures.values()
316
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/sentry/rules/processor.py b/src/sentry/rules/processor.py
--- a/src/sentry/rules/processor.py
+++ b/src/sentry/rules/processor.py
@@ -27,7 +27,9 @@
 from sentry.models.rule import Rule
 from sentry.models.rulesnooze import RuleSnooze
 from sentry.rules import EventState, history, rules
+from sentry.rules.actions.base import EventAction
 from sentry.rules.conditions.base import EventCondition
+from sentry.rules.filters.base import EventFilter
 from sentry.types.rules import RuleFuture
 from sentry.utils.hashlib import hash_values
 from sentry.utils.safe import safe_execute
@@ -143,16 +145,22 @@
         return rule_statuses
 
     def condition_matches(
-        self, condition: Mapping[str, Any], state: EventState, rule: Rule
+        self, condition: dict[str, Any], state: EventState, rule: Rule
     ) -> bool | None:
         condition_cls = rules.get(condition["id"])
         if condition_cls is None:
             self.logger.warning("Unregistered condition %r", condition["id"])
             return None
 
-        condition_inst: EventCondition = condition_cls(self.project, data=condition, rule=rule)
+        condition_inst = condition_cls(self.project, data=condition, rule=rule)
+        if not isinstance(condition_inst, (EventCondition, EventFilter)):
+            self.logger.warning("Unregistered condition %r", condition["id"])
+            return None
         passes: bool = safe_execute(
-            condition_inst.passes, self.event, state, _with_transaction=False
+            condition_inst.passes,
+            self.event,
+            state,
+            _with_transaction=False,
         )
         return passes
 
@@ -274,6 +282,9 @@
                 continue
 
             action_inst = action_cls(self.project, data=action, rule=rule)
+            if not isinstance(action_inst, EventAction):
+                self.logger.warning("Unregistered action %r", action["id"])
+                continue
 
             results = safe_execute(
                 action_inst.after,

{"golden_diff": "diff --git a/src/sentry/rules/processor.py b/src/sentry/rules/processor.py\n--- a/src/sentry/rules/processor.py\n+++ b/src/sentry/rules/processor.py\n@@ -27,7 +27,9 @@\n from sentry.models.rule import Rule\n from sentry.models.rulesnooze import RuleSnooze\n from sentry.rules import EventState, history, rules\n+from sentry.rules.actions.base import EventAction\n from sentry.rules.conditions.base import EventCondition\n+from sentry.rules.filters.base import EventFilter\n from sentry.types.rules import RuleFuture\n from sentry.utils.hashlib import hash_values\n from sentry.utils.safe import safe_execute\n@@ -143,16 +145,22 @@\n         return rule_statuses\n \n     def condition_matches(\n-        self, condition: Mapping[str, Any], state: EventState, rule: Rule\n+        self, condition: dict[str, Any], state: EventState, rule: Rule\n     ) -> bool | None:\n         condition_cls = rules.get(condition[\"id\"])\n         if condition_cls is None:\n             self.logger.warning(\"Unregistered condition %r\", condition[\"id\"])\n             return None\n \n-        condition_inst: EventCondition = condition_cls(self.project, data=condition, rule=rule)\n+        condition_inst = condition_cls(self.project, data=condition, rule=rule)\n+        if not isinstance(condition_inst, (EventCondition, EventFilter)):\n+            self.logger.warning(\"Unregistered condition %r\", condition[\"id\"])\n+            return None\n         passes: bool = safe_execute(\n-            condition_inst.passes, self.event, state, _with_transaction=False\n+            condition_inst.passes,\n+            self.event,\n+            state,\n+            _with_transaction=False,\n         )\n         return passes\n \n@@ -274,6 +282,9 @@\n                 continue\n \n             action_inst = action_cls(self.project, data=action, rule=rule)\n+            if not isinstance(action_inst, EventAction):\n+                self.logger.warning(\"Unregistered action %r\", action[\"id\"])\n+                continue\n \n             results = safe_execute(\n                 action_inst.after,\n", "issue": "Fix typing for `sentry.rules.processor`\nMany files related to post-process are skipping type checks.\n\n* Remove the file from [sentry/pyproject.toml](https://github.com/getsentry/sentry/blob/b7810a9dca4b57afd2858903a6a9ec7ab50cdead/pyproject.toml)\n* Run `mypy sentry.rules.processor` and fix the typing errors\n\nFrom https://github.com/getsentry/sentry/issues/55193\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport uuid\nfrom datetime import timedelta\nfrom random import randrange\nfrom typing import (\n    Any,\n    Callable,\n    Collection,\n    List,\n    Mapping,\n    MutableMapping,\n    Optional,\n    Sequence,\n    Set,\n    Tuple,\n)\n\nfrom django.core.cache import cache\nfrom django.utils import timezone\n\nfrom sentry import analytics\nfrom sentry.eventstore.models import GroupEvent\nfrom sentry.models.environment import Environment\nfrom sentry.models.grouprulestatus import GroupRuleStatus\nfrom sentry.models.rule import Rule\nfrom sentry.models.rulesnooze import RuleSnooze\nfrom sentry.rules import EventState, history, rules\nfrom sentry.rules.conditions.base import EventCondition\nfrom sentry.types.rules import RuleFuture\nfrom sentry.utils.hashlib import hash_values\nfrom sentry.utils.safe import safe_execute\n\nSLOW_CONDITION_MATCHES = [\"event_frequency\"]\n\n\ndef get_match_function(match_name: str) -> Callable[..., bool] | None:\n    if match_name == \"all\":\n        return all\n    elif match_name == \"any\":\n        return any\n    elif match_name == \"none\":\n        return lambda bool_iter: not any(bool_iter)\n    return None\n\n\ndef is_condition_slow(condition: Mapping[str, str]) -> bool:\n    for slow_conditions in SLOW_CONDITION_MATCHES:\n        if slow_conditions in condition[\"id\"]:\n            return True\n    return False\n\n\nclass RuleProcessor:\n    logger = logging.getLogger(\"sentry.rules\")\n\n    def __init__(\n        self,\n        event: GroupEvent,\n        is_new: bool,\n        is_regression: bool,\n        is_new_group_environment: bool,\n        has_reappeared: bool,\n    ) -> None:\n        self.event = event\n        self.group = event.group\n        self.project = event.project\n\n        self.is_new = is_new\n        self.is_regression = is_regression\n        self.is_new_group_environment = is_new_group_environment\n        self.has_reappeared = has_reappeared\n\n        self.grouped_futures: MutableMapping[\n            str, Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]\n        ] = {}\n\n    def get_rules(self) -> Sequence[Rule]:\n        \"\"\"Get all of the rules for this project from the DB (or cache).\"\"\"\n        rules_: Sequence[Rule] = Rule.get_for_project(self.project.id)\n        return rules_\n\n    def _build_rule_status_cache_key(self, rule_id: int) -> str:\n        return \"grouprulestatus:1:%s\" % hash_values([self.group.id, rule_id])\n\n    def bulk_get_rule_status(self, rules: Sequence[Rule]) -> Mapping[int, GroupRuleStatus]:\n        keys = [self._build_rule_status_cache_key(rule.id) for rule in rules]\n        cache_results: Mapping[str, GroupRuleStatus] = cache.get_many(keys)\n        missing_rule_ids: Set[int] = set()\n        rule_statuses: MutableMapping[int, GroupRuleStatus] = {}\n        for key, rule in zip(keys, rules):\n            rule_status = cache_results.get(key)\n            if not rule_status:\n                missing_rule_ids.add(rule.id)\n            else:\n                rule_statuses[rule.id] = rule_status\n\n        if missing_rule_ids:\n            # If not cached, attempt to fetch status from the database\n            statuses = GroupRuleStatus.objects.filter(\n                group=self.group, rule_id__in=missing_rule_ids\n            )\n            to_cache: List[GroupRuleStatus] = list()\n            for status in statuses:\n                rule_statuses[status.rule_id] = status\n                missing_rule_ids.remove(status.rule_id)\n                to_cache.append(status)\n\n            # We might need to create some statuses if they don't already exist\n            if missing_rule_ids:\n                # We use `ignore_conflicts=True` here to avoid race conditions where the statuses\n                # might be created between when we queried above and attempt to create the rows now.\n                GroupRuleStatus.objects.bulk_create(\n                    [\n                        GroupRuleStatus(rule_id=rule_id, group=self.group, project=self.project)\n                        for rule_id in missing_rule_ids\n                    ],\n                    ignore_conflicts=True,\n                )\n                # Using `ignore_conflicts=True` prevents the pk from being set on the model\n                # instances. Re-query the database to fetch the rows, they should all exist at this\n                # point.\n                statuses = GroupRuleStatus.objects.filter(\n                    group=self.group, rule_id__in=missing_rule_ids\n                )\n                for status in statuses:\n                    rule_statuses[status.rule_id] = status\n                    missing_rule_ids.remove(status.rule_id)\n                    to_cache.append(status)\n\n                if missing_rule_ids:\n                    # Shouldn't happen, but log just in case\n                    self.logger.error(\n                        \"Failed to fetch some GroupRuleStatuses in RuleProcessor\",\n                        extra={\"missing_rule_ids\": missing_rule_ids, \"group_id\": self.group.id},\n                    )\n            if to_cache:\n                cache.set_many(\n                    {self._build_rule_status_cache_key(item.rule_id): item for item in to_cache}\n                )\n\n        return rule_statuses\n\n    def condition_matches(\n        self, condition: Mapping[str, Any], state: EventState, rule: Rule\n    ) -> bool | None:\n        condition_cls = rules.get(condition[\"id\"])\n        if condition_cls is None:\n            self.logger.warning(\"Unregistered condition %r\", condition[\"id\"])\n            return None\n\n        condition_inst: EventCondition = condition_cls(self.project, data=condition, rule=rule)\n        passes: bool = safe_execute(\n            condition_inst.passes, self.event, state, _with_transaction=False\n        )\n        return passes\n\n    def get_rule_type(self, condition: Mapping[str, Any]) -> str | None:\n        rule_cls = rules.get(condition[\"id\"])\n        if rule_cls is None:\n            self.logger.warning(\"Unregistered condition or filter %r\", condition[\"id\"])\n            return None\n\n        rule_type: str = rule_cls.rule_type\n        return rule_type\n\n    def get_state(self) -> EventState:\n        return EventState(\n            is_new=self.is_new,\n            is_regression=self.is_regression,\n            is_new_group_environment=self.is_new_group_environment,\n            has_reappeared=self.has_reappeared,\n        )\n\n    def apply_rule(self, rule: Rule, status: GroupRuleStatus) -> None:\n        \"\"\"\n        If all conditions and filters pass, execute every action.\n\n        :param rule: `Rule` object\n        :return: void\n        \"\"\"\n        logging_details = {\n            \"rule_id\": rule.id,\n            \"group_id\": self.group.id,\n            \"event_id\": self.event.event_id,\n            \"project_id\": self.project.id,\n            \"is_new\": self.is_new,\n            \"is_regression\": self.is_regression,\n            \"has_reappeared\": self.has_reappeared,\n            \"new_group_environment\": self.is_new_group_environment,\n        }\n\n        condition_match = rule.data.get(\"action_match\") or Rule.DEFAULT_CONDITION_MATCH\n        filter_match = rule.data.get(\"filter_match\") or Rule.DEFAULT_FILTER_MATCH\n        rule_condition_list = rule.data.get(\"conditions\", ())\n        frequency = rule.data.get(\"frequency\") or Rule.DEFAULT_FREQUENCY\n        try:\n            environment = self.event.get_environment()\n        except Environment.DoesNotExist:\n            return\n\n        if rule.environment_id is not None and environment.id != rule.environment_id:\n            return\n\n        now = timezone.now()\n        freq_offset = now - timedelta(minutes=frequency)\n        if status.last_active and status.last_active > freq_offset:\n            return\n\n        state = self.get_state()\n\n        condition_list = []\n        filter_list = []\n        for rule_cond in rule_condition_list:\n            if self.get_rule_type(rule_cond) == \"condition/event\":\n                condition_list.append(rule_cond)\n            else:\n                filter_list.append(rule_cond)\n\n        # Sort `condition_list` so that most expensive conditions run last.\n        condition_list.sort(key=lambda condition: is_condition_slow(condition))\n\n        for predicate_list, match, name in (\n            (filter_list, filter_match, \"filter\"),\n            (condition_list, condition_match, \"condition\"),\n        ):\n            if not predicate_list:\n                continue\n            predicate_iter = (self.condition_matches(f, state, rule) for f in predicate_list)\n            predicate_func = get_match_function(match)\n            if predicate_func:\n                if not predicate_func(predicate_iter):\n                    return\n            else:\n                self.logger.error(\n                    f\"Unsupported {name}_match {match!r} for rule {rule.id}\",\n                    filter_match,\n                    rule.id,\n                    extra={**logging_details},\n                )\n                return\n\n        updated = (\n            GroupRuleStatus.objects.filter(id=status.id)\n            .exclude(last_active__gt=freq_offset)\n            .update(last_active=now)\n        )\n\n        if not updated:\n            return\n\n        if randrange(10) == 0:\n            analytics.record(\n                \"issue_alert.fired\",\n                issue_id=self.group.id,\n                project_id=rule.project.id,\n                organization_id=rule.project.organization.id,\n                rule_id=rule.id,\n            )\n\n        notification_uuid = str(uuid.uuid4())\n        history.record(rule, self.group, self.event.event_id, notification_uuid)\n        self.activate_downstream_actions(rule, notification_uuid)\n\n    def activate_downstream_actions(\n        self, rule: Rule, notification_uuid: Optional[str] = None\n    ) -> None:\n        state = self.get_state()\n        for action in rule.data.get(\"actions\", ()):\n            action_cls = rules.get(action[\"id\"])\n            if action_cls is None:\n                self.logger.warning(\"Unregistered action %r\", action[\"id\"])\n                continue\n\n            action_inst = action_cls(self.project, data=action, rule=rule)\n\n            results = safe_execute(\n                action_inst.after,\n                event=self.event,\n                state=state,\n                _with_transaction=False,\n                notification_uuid=notification_uuid,\n            )\n            if results is None:\n                self.logger.warning(\"Action %s did not return any futures\", action[\"id\"])\n                continue\n\n            for future in results:\n                key = future.key if future.key is not None else future.callback\n                rule_future = RuleFuture(rule=rule, kwargs=future.kwargs)\n\n                if key not in self.grouped_futures:\n                    self.grouped_futures[key] = (future.callback, [rule_future])\n                else:\n                    self.grouped_futures[key][1].append(rule_future)\n\n    def apply(\n        self,\n    ) -> Collection[Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]]:\n        # we should only apply rules on unresolved issues\n        if not self.event.group.is_unresolved():\n            return {}.values()\n\n        self.grouped_futures.clear()\n        rules = self.get_rules()\n        snoozed_rules = RuleSnooze.objects.filter(rule__in=rules, user_id=None).values_list(\n            \"rule\", flat=True\n        )\n        rule_statuses = self.bulk_get_rule_status(rules)\n        for rule in rules:\n            if rule.id not in snoozed_rules:\n                self.apply_rule(rule, rule_statuses[rule.id])\n\n        return self.grouped_futures.values()\n", "path": "src/sentry/rules/processor.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport uuid\nfrom datetime import timedelta\nfrom random import randrange\nfrom typing import (\n    Any,\n    Callable,\n    Collection,\n    List,\n    Mapping,\n    MutableMapping,\n    Optional,\n    Sequence,\n    Set,\n    Tuple,\n)\n\nfrom django.core.cache import cache\nfrom django.utils import timezone\n\nfrom sentry import analytics\nfrom sentry.eventstore.models import GroupEvent\nfrom sentry.models.environment import Environment\nfrom sentry.models.grouprulestatus import GroupRuleStatus\nfrom sentry.models.rule import Rule\nfrom sentry.models.rulesnooze import RuleSnooze\nfrom sentry.rules import EventState, history, rules\nfrom sentry.rules.actions.base import EventAction\nfrom sentry.rules.conditions.base import EventCondition\nfrom sentry.rules.filters.base import EventFilter\nfrom sentry.types.rules import RuleFuture\nfrom sentry.utils.hashlib import hash_values\nfrom sentry.utils.safe import safe_execute\n\nSLOW_CONDITION_MATCHES = [\"event_frequency\"]\n\n\ndef get_match_function(match_name: str) -> Callable[..., bool] | None:\n    if match_name == \"all\":\n        return all\n    elif match_name == \"any\":\n        return any\n    elif match_name == \"none\":\n        return lambda bool_iter: not any(bool_iter)\n    return None\n\n\ndef is_condition_slow(condition: Mapping[str, str]) -> bool:\n    for slow_conditions in SLOW_CONDITION_MATCHES:\n        if slow_conditions in condition[\"id\"]:\n            return True\n    return False\n\n\nclass RuleProcessor:\n    logger = logging.getLogger(\"sentry.rules\")\n\n    def __init__(\n        self,\n        event: GroupEvent,\n        is_new: bool,\n        is_regression: bool,\n        is_new_group_environment: bool,\n        has_reappeared: bool,\n    ) -> None:\n        self.event = event\n        self.group = event.group\n        self.project = event.project\n\n        self.is_new = is_new\n        self.is_regression = is_regression\n        self.is_new_group_environment = is_new_group_environment\n        self.has_reappeared = has_reappeared\n\n        self.grouped_futures: MutableMapping[\n            str, Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]\n        ] = {}\n\n    def get_rules(self) -> Sequence[Rule]:\n        \"\"\"Get all of the rules for this project from the DB (or cache).\"\"\"\n        rules_: Sequence[Rule] = Rule.get_for_project(self.project.id)\n        return rules_\n\n    def _build_rule_status_cache_key(self, rule_id: int) -> str:\n        return \"grouprulestatus:1:%s\" % hash_values([self.group.id, rule_id])\n\n    def bulk_get_rule_status(self, rules: Sequence[Rule]) -> Mapping[int, GroupRuleStatus]:\n        keys = [self._build_rule_status_cache_key(rule.id) for rule in rules]\n        cache_results: Mapping[str, GroupRuleStatus] = cache.get_many(keys)\n        missing_rule_ids: Set[int] = set()\n        rule_statuses: MutableMapping[int, GroupRuleStatus] = {}\n        for key, rule in zip(keys, rules):\n            rule_status = cache_results.get(key)\n            if not rule_status:\n                missing_rule_ids.add(rule.id)\n            else:\n                rule_statuses[rule.id] = rule_status\n\n        if missing_rule_ids:\n            # If not cached, attempt to fetch status from the database\n            statuses = GroupRuleStatus.objects.filter(\n                group=self.group, rule_id__in=missing_rule_ids\n            )\n            to_cache: List[GroupRuleStatus] = list()\n            for status in statuses:\n                rule_statuses[status.rule_id] = status\n                missing_rule_ids.remove(status.rule_id)\n                to_cache.append(status)\n\n            # We might need to create some statuses if they don't already exist\n            if missing_rule_ids:\n                # We use `ignore_conflicts=True` here to avoid race conditions where the statuses\n                # might be created between when we queried above and attempt to create the rows now.\n                GroupRuleStatus.objects.bulk_create(\n                    [\n                        GroupRuleStatus(rule_id=rule_id, group=self.group, project=self.project)\n                        for rule_id in missing_rule_ids\n                    ],\n                    ignore_conflicts=True,\n                )\n                # Using `ignore_conflicts=True` prevents the pk from being set on the model\n                # instances. Re-query the database to fetch the rows, they should all exist at this\n                # point.\n                statuses = GroupRuleStatus.objects.filter(\n                    group=self.group, rule_id__in=missing_rule_ids\n                )\n                for status in statuses:\n                    rule_statuses[status.rule_id] = status\n                    missing_rule_ids.remove(status.rule_id)\n                    to_cache.append(status)\n\n                if missing_rule_ids:\n                    # Shouldn't happen, but log just in case\n                    self.logger.error(\n                        \"Failed to fetch some GroupRuleStatuses in RuleProcessor\",\n                        extra={\"missing_rule_ids\": missing_rule_ids, \"group_id\": self.group.id},\n                    )\n            if to_cache:\n                cache.set_many(\n                    {self._build_rule_status_cache_key(item.rule_id): item for item in to_cache}\n                )\n\n        return rule_statuses\n\n    def condition_matches(\n        self, condition: dict[str, Any], state: EventState, rule: Rule\n    ) -> bool | None:\n        condition_cls = rules.get(condition[\"id\"])\n        if condition_cls is None:\n            self.logger.warning(\"Unregistered condition %r\", condition[\"id\"])\n            return None\n\n        condition_inst = condition_cls(self.project, data=condition, rule=rule)\n        if not isinstance(condition_inst, (EventCondition, EventFilter)):\n            self.logger.warning(\"Unregistered condition %r\", condition[\"id\"])\n            return None\n        passes: bool = safe_execute(\n            condition_inst.passes,\n            self.event,\n            state,\n            _with_transaction=False,\n        )\n        return passes\n\n    def get_rule_type(self, condition: Mapping[str, Any]) -> str | None:\n        rule_cls = rules.get(condition[\"id\"])\n        if rule_cls is None:\n            self.logger.warning(\"Unregistered condition or filter %r\", condition[\"id\"])\n            return None\n\n        rule_type: str = rule_cls.rule_type\n        return rule_type\n\n    def get_state(self) -> EventState:\n        return EventState(\n            is_new=self.is_new,\n            is_regression=self.is_regression,\n            is_new_group_environment=self.is_new_group_environment,\n            has_reappeared=self.has_reappeared,\n        )\n\n    def apply_rule(self, rule: Rule, status: GroupRuleStatus) -> None:\n        \"\"\"\n        If all conditions and filters pass, execute every action.\n\n        :param rule: `Rule` object\n        :return: void\n        \"\"\"\n        logging_details = {\n            \"rule_id\": rule.id,\n            \"group_id\": self.group.id,\n            \"event_id\": self.event.event_id,\n            \"project_id\": self.project.id,\n            \"is_new\": self.is_new,\n            \"is_regression\": self.is_regression,\n            \"has_reappeared\": self.has_reappeared,\n            \"new_group_environment\": self.is_new_group_environment,\n        }\n\n        condition_match = rule.data.get(\"action_match\") or Rule.DEFAULT_CONDITION_MATCH\n        filter_match = rule.data.get(\"filter_match\") or Rule.DEFAULT_FILTER_MATCH\n        rule_condition_list = rule.data.get(\"conditions\", ())\n        frequency = rule.data.get(\"frequency\") or Rule.DEFAULT_FREQUENCY\n        try:\n            environment = self.event.get_environment()\n        except Environment.DoesNotExist:\n            return\n\n        if rule.environment_id is not None and environment.id != rule.environment_id:\n            return\n\n        now = timezone.now()\n        freq_offset = now - timedelta(minutes=frequency)\n        if status.last_active and status.last_active > freq_offset:\n            return\n\n        state = self.get_state()\n\n        condition_list = []\n        filter_list = []\n        for rule_cond in rule_condition_list:\n            if self.get_rule_type(rule_cond) == \"condition/event\":\n                condition_list.append(rule_cond)\n            else:\n                filter_list.append(rule_cond)\n\n        # Sort `condition_list` so that most expensive conditions run last.\n        condition_list.sort(key=lambda condition: is_condition_slow(condition))\n\n        for predicate_list, match, name in (\n            (filter_list, filter_match, \"filter\"),\n            (condition_list, condition_match, \"condition\"),\n        ):\n            if not predicate_list:\n                continue\n            predicate_iter = (self.condition_matches(f, state, rule) for f in predicate_list)\n            predicate_func = get_match_function(match)\n            if predicate_func:\n                if not predicate_func(predicate_iter):\n                    return\n            else:\n                self.logger.error(\n                    f\"Unsupported {name}_match {match!r} for rule {rule.id}\",\n                    filter_match,\n                    rule.id,\n                    extra={**logging_details},\n                )\n                return\n\n        updated = (\n            GroupRuleStatus.objects.filter(id=status.id)\n            .exclude(last_active__gt=freq_offset)\n            .update(last_active=now)\n        )\n\n        if not updated:\n            return\n\n        if randrange(10) == 0:\n            analytics.record(\n                \"issue_alert.fired\",\n                issue_id=self.group.id,\n                project_id=rule.project.id,\n                organization_id=rule.project.organization.id,\n                rule_id=rule.id,\n            )\n\n        notification_uuid = str(uuid.uuid4())\n        history.record(rule, self.group, self.event.event_id, notification_uuid)\n        self.activate_downstream_actions(rule, notification_uuid)\n\n    def activate_downstream_actions(\n        self, rule: Rule, notification_uuid: Optional[str] = None\n    ) -> None:\n        state = self.get_state()\n        for action in rule.data.get(\"actions\", ()):\n            action_cls = rules.get(action[\"id\"])\n            if action_cls is None:\n                self.logger.warning(\"Unregistered action %r\", action[\"id\"])\n                continue\n\n            action_inst = action_cls(self.project, data=action, rule=rule)\n            if not isinstance(action_inst, EventAction):\n                self.logger.warning(\"Unregistered action %r\", action[\"id\"])\n                continue\n\n            results = safe_execute(\n                action_inst.after,\n                event=self.event,\n                state=state,\n                _with_transaction=False,\n                notification_uuid=notification_uuid,\n            )\n            if results is None:\n                self.logger.warning(\"Action %s did not return any futures\", action[\"id\"])\n                continue\n\n            for future in results:\n                key = future.key if future.key is not None else future.callback\n                rule_future = RuleFuture(rule=rule, kwargs=future.kwargs)\n\n                if key not in self.grouped_futures:\n                    self.grouped_futures[key] = (future.callback, [rule_future])\n                else:\n                    self.grouped_futures[key][1].append(rule_future)\n\n    def apply(\n        self,\n    ) -> Collection[Tuple[Callable[[GroupEvent, Sequence[RuleFuture]], None], List[RuleFuture]]]:\n        # we should only apply rules on unresolved issues\n        if not self.event.group.is_unresolved():\n            return {}.values()\n\n        self.grouped_futures.clear()\n        rules = self.get_rules()\n        snoozed_rules = RuleSnooze.objects.filter(rule__in=rules, user_id=None).values_list(\n            \"rule\", flat=True\n        )\n        rule_statuses = self.bulk_get_rule_status(rules)\n        for rule in rules:\n            if rule.id not in snoozed_rules:\n                self.apply_rule(rule, rule_statuses[rule.id])\n\n        return self.grouped_futures.values()\n", "path": "src/sentry/rules/processor.py"}]}