We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix logic bug where Group.Status is archived_until_escalating but the GroupInbox.Status is escalating


![Image](https://user-images.githubusercontent.com/10491193/236284761-303ee7f8-c596-4034-b21f-0f44e779e129.png)

## Objective:
We are not transitioning the GroupStatus to escalating but transitioned GroupInboxStatus.

--- 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/issues/escalating.py`
Content:
```
1 """This module has the logic for querying Snuba for the hourly event count for a list of groups.
2 This is later used for generating group forecasts for determining when a group may be escalating.
3 """
4 
5 import logging
6 from collections import defaultdict
7 from datetime import datetime, timedelta
8 from typing import Dict, List, Optional, Sequence, Tuple, TypedDict
9 
10 from snuba_sdk import (
11     Column,
12     Condition,
13     Direction,
14     Entity,
15     Function,
16     Limit,
17     Offset,
18     Op,
19     OrderBy,
20     Query,
21     Request,
22 )
23 
24 from sentry import analytics
25 from sentry.issues.escalating_group_forecast import EscalatingGroupForecast
26 from sentry.issues.escalating_issues_alg import GroupCount
27 from sentry.issues.grouptype import GroupCategory
28 from sentry.models import Group
29 from sentry.models.group import GroupStatus
30 from sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox
31 from sentry.snuba.dataset import Dataset, EntityKey
32 from sentry.types.group import GroupSubStatus
33 from sentry.utils.cache import cache
34 from sentry.utils.snuba import raw_snql_query
35 
36 logger = logging.getLogger(__name__)
37 
38 __all__ = ["query_groups_past_counts", "parse_groups_past_counts"]
39 
40 REFERRER = "sentry.issues.escalating"
41 ELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba
42 # The amount of data needed to generate a group forecast
43 BUCKETS_PER_GROUP = 7 * 24
44 ONE_WEEK_DURATION = 7
45 IS_ESCALATING_REFERRER = "sentry.issues.escalating.is_escalating"
46 GROUP_HOURLY_COUNT_TTL = 60
47 
48 GroupsCountResponse = TypedDict(
49     "GroupsCountResponse",
50     {"group_id": int, "hourBucket": str, "count()": int, "project_id": int},
51 )
52 
53 ParsedGroupsCount = Dict[int, GroupCount]
54 
55 
56 def query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:
57     """Query Snuba for the counts for every group bucketed into hours.
58 
59     It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.
60     This is important for Snuba as the data is stored in blocks related to the project id.
61 
62     We maximize the number of projects and groups to reduce the total number of Snuba queries.
63     Each project may not have enough groups in order to reach the max number of returned
64     elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until
65     we get at least a certain number of groups.
66 
67     NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less
68     than 7 days old) will skew the optimization since we may only get one page and less elements than the max
69     ELEMENTS_PER_SNUBA_PAGE.
70     """
71     all_results = []  # type: ignore[var-annotated]
72     if not groups:
73         return all_results
74 
75     start_date, end_date = _start_and_end_dates()
76 
77     # Error groups use the events dataset while profile and perf groups use the issue platform dataset
78     error_groups: List[Group] = []
79     other_groups: List[Group] = []
80     for g in groups:
81         if g.issue_category == GroupCategory.ERROR:
82             error_groups.append(g)
83         else:
84             other_groups.append(g)
85 
86     all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)
87     all_results += _process_groups(other_groups, start_date, end_date)
88 
89     return all_results
90 
91 
92 def _process_groups(
93     groups: Sequence[Group],
94     start_date: datetime,
95     end_date: datetime,
96     category: Optional[GroupCategory] = None,
97 ) -> List[GroupsCountResponse]:
98     """Given a list of groups, query Snuba for their hourly bucket count.
99     The category defines which Snuba dataset and entity we query."""
100     all_results = []  # type: ignore[var-annotated]
101     if not groups:
102         return all_results
103 
104     group_ids_by_project = _extract_project_and_group_ids(groups)
105     proj_ids, group_ids = [], []
106     processed_projects = 0
107     total_projects_count = len(group_ids_by_project)
108     organization_id = groups[0].project.organization.id
109 
110     # This iteration guarantees that all groups for a project will be queried in the same call
111     # and only one page where the groups could be mixed with groups from another project
112     # Iterating over the sorted keys guarantees results for tests
113     for proj_id in sorted(group_ids_by_project.keys()):
114         _group_ids = group_ids_by_project[proj_id]
115         # Add them to the list of projects and groups to query
116         proj_ids.append(proj_id)
117         group_ids += _group_ids
118         processed_projects += 1
119         potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP
120         # This is trying to maximize the number of groups on the first page
121         if (
122             processed_projects < total_projects_count
123             and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE
124         ):
125             continue
126 
127         # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids
128         all_results += _query_with_pagination(
129             organization_id, proj_ids, group_ids, start_date, end_date, category
130         )
131         # We're ready for a new set of projects and ids
132         proj_ids, group_ids = [], []
133 
134     return all_results
135 
136 
137 def _query_with_pagination(
138     organization_id: int,
139     project_ids: Sequence[int],
140     group_ids: Sequence[int],
141     start_date: datetime,
142     end_date: datetime,
143     category: Optional[GroupCategory],
144 ) -> List[GroupsCountResponse]:
145     """Query Snuba for event counts for the given list of project ids and groups ids in
146     a time range."""
147     all_results = []
148     offset = 0
149     while True:
150         query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)
151         request = Request(
152             dataset=_issue_category_dataset(category),
153             app_id=REFERRER,
154             query=query,
155             tenant_ids={"referrer": REFERRER, "organization_id": organization_id},
156         )
157         results = raw_snql_query(request, referrer=REFERRER)["data"]
158         all_results += results
159         offset += ELEMENTS_PER_SNUBA_PAGE
160         if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:
161             break
162 
163     return all_results
164 
165 
166 def _generate_query(
167     project_ids: Sequence[int],
168     group_ids: Sequence[int],
169     offset: int,
170     start_date: datetime,
171     end_date: datetime,
172     category: Optional[GroupCategory],
173 ) -> Query:
174     """This simply generates a query based on the passed parameters"""
175     group_id_col = Column("group_id")
176     proj_id_col = Column("project_id")
177     return Query(
178         match=Entity(_issue_category_entity(category)),
179         select=[
180             proj_id_col,
181             group_id_col,
182             Function("toStartOfHour", [Column("timestamp")], "hourBucket"),
183             Function("count", []),
184         ],
185         groupby=[proj_id_col, group_id_col, Column("hourBucket")],
186         where=[
187             Condition(proj_id_col, Op.IN, Function("tuple", project_ids)),
188             Condition(Column("group_id"), Op.IN, Function("tuple", group_ids)),
189             Condition(Column("timestamp"), Op.GTE, start_date),
190             Condition(Column("timestamp"), Op.LT, end_date),
191         ],
192         limit=Limit(ELEMENTS_PER_SNUBA_PAGE),
193         offset=Offset(offset),
194         orderby=[
195             OrderBy(proj_id_col, Direction.ASC),
196             OrderBy(group_id_col, Direction.ASC),
197             OrderBy(Column("hourBucket"), Direction.ASC),
198         ],
199     )
200 
201 
202 def _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:
203     """Return the start and end date of N hours time range."""
204     end_datetime = datetime.now()
205     return end_datetime - timedelta(hours=hours), end_datetime
206 
207 
208 def _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:
209     """Return all project and group IDs from a list of Group"""
210     group_ids_by_project: Dict[int, List[int]] = defaultdict(list)
211     for group in groups:
212         group_ids_by_project[group.project_id].append(group.id)
213 
214     return group_ids_by_project
215 
216 
217 def get_group_hourly_count(group: Group) -> int:
218     """Return the number of events a group has had today in the last hour"""
219     key = f"hourly-group-count:{group.project.id}:{group.id}"
220     hourly_count = cache.get(key)
221 
222     if hourly_count is None:
223         now = datetime.now()
224         current_hour = now.replace(minute=0, second=0, microsecond=0)
225         query = Query(
226             match=Entity(_issue_category_entity(group.issue_category)),
227             select=[
228                 Function("count", []),
229             ],
230             where=[
231                 Condition(Column("project_id"), Op.EQ, group.project.id),
232                 Condition(Column("group_id"), Op.EQ, group.id),
233                 Condition(Column("timestamp"), Op.GTE, current_hour),
234                 Condition(Column("timestamp"), Op.LT, now),
235             ],
236         )
237         request = Request(
238             dataset=_issue_category_dataset(group.issue_category),
239             app_id=IS_ESCALATING_REFERRER,
240             query=query,
241             tenant_ids={
242                 "referrer": IS_ESCALATING_REFERRER,
243                 "organization_id": group.project.organization.id,
244             },
245         )
246         hourly_count = int(
247             raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)["data"][0]["count()"]
248         )
249         cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)
250     return int(hourly_count)
251 
252 
253 def is_escalating(group: Group) -> bool:
254     """Return boolean depending on if the group is escalating or not"""
255     group_hourly_count = get_group_hourly_count(group)
256     forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)
257     # Check if current event occurance is greater than forecast for today's date
258     if group_hourly_count > forecast_today:
259         group.substatus = GroupSubStatus.ESCALATING
260         group.status = GroupStatus.UNRESOLVED
261         add_group_to_inbox(group, GroupInboxReason.ESCALATING)
262 
263         analytics.record(
264             "issue.escalating",
265             organization_id=group.project.organization.id,
266             project_id=group.project.id,
267             group_id=group.id,
268         )
269         return True
270     return False
271 
272 
273 def parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:
274     """
275     Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.
276     ParsedGroupCount is of the form {<group_id>: {"intervals": [str], "data": [int]}}.
277 
278     `response`: Snuba response for group event counts
279     """
280     group_counts: ParsedGroupsCount = {}
281     group_ids_list = group_counts.keys()
282     for data in response:
283         group_id = data["group_id"]
284         if group_id not in group_ids_list:
285             group_counts[group_id] = {
286                 "intervals": [data["hourBucket"]],
287                 "data": [data["count()"]],
288             }
289         else:
290             group_counts[group_id]["intervals"].append(data["hourBucket"])
291             group_counts[group_id]["data"].append(data["count()"])
292     return group_counts
293 
294 
295 def _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:
296     return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value
297 
298 
299 def _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:
300     return (
301         EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value
302     )
303
```
--- 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/issues/escalating.py b/src/sentry/issues/escalating.py
--- a/src/sentry/issues/escalating.py
+++ b/src/sentry/issues/escalating.py
@@ -258,6 +258,7 @@
     if group_hourly_count > forecast_today:
         group.substatus = GroupSubStatus.ESCALATING
         group.status = GroupStatus.UNRESOLVED
+        group.save()
         add_group_to_inbox(group, GroupInboxReason.ESCALATING)
 
         analytics.record(

{"golden_diff": "diff --git a/src/sentry/issues/escalating.py b/src/sentry/issues/escalating.py\n--- a/src/sentry/issues/escalating.py\n+++ b/src/sentry/issues/escalating.py\n@@ -258,6 +258,7 @@\n     if group_hourly_count > forecast_today:\n         group.substatus = GroupSubStatus.ESCALATING\n         group.status = GroupStatus.UNRESOLVED\n+        group.save()\n         add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n \n         analytics.record(\n", "issue": "Fix logic bug where Group.Status is archived_until_escalating but the GroupInbox.Status is escalating\n\n\n![Image](https://user-images.githubusercontent.com/10491193/236284761-303ee7f8-c596-4034-b21f-0f44e779e129.png)\n\n## Objective:\nWe are not transitioning the GroupStatus to escalating but transitioned GroupInboxStatus.\n", "before_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Optional, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry import analytics\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.issues.grouptype import GroupCategory\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_HOURLY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n\n    # Error groups use the events dataset while profile and perf groups use the issue platform dataset\n    error_groups: List[Group] = []\n    other_groups: List[Group] = []\n    for g in groups:\n        if g.issue_category == GroupCategory.ERROR:\n            error_groups.append(g)\n        else:\n            other_groups.append(g)\n\n    all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)\n    all_results += _process_groups(other_groups, start_date, end_date)\n\n    return all_results\n\n\ndef _process_groups(\n    groups: Sequence[Group],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory] = None,\n) -> List[GroupsCountResponse]:\n    \"\"\"Given a list of groups, query Snuba for their hourly bucket count.\n    The category defines which Snuba dataset and entity we query.\"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date, category\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)\n        request = Request(\n            dataset=_issue_category_dataset(category),\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(_issue_category_entity(category)),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_hourly_count(group: Group) -> int:\n    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n    key = f\"hourly-group-count:{group.project.id}:{group.id}\"\n    hourly_count = cache.get(key)\n\n    if hourly_count is None:\n        now = datetime.now()\n        current_hour = now.replace(minute=0, second=0, microsecond=0)\n        query = Query(\n            match=Entity(_issue_category_entity(group.issue_category)),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, group.project.id),\n                Condition(Column(\"group_id\"), Op.EQ, group.id),\n                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=_issue_category_dataset(group.issue_category),\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\n                \"referrer\": IS_ESCALATING_REFERRER,\n                \"organization_id\": group.project.organization.id,\n            },\n        )\n        hourly_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n    return int(hourly_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_hourly_count = get_group_hourly_count(group)\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_hourly_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n\n        analytics.record(\n            \"issue.escalating\",\n            organization_id=group.project.organization.id,\n            project_id=group.project.id,\n            group_id=group.id,\n        )\n        return True\n    return False\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:\n    return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value\n\n\ndef _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:\n    return (\n        EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value\n    )\n", "path": "src/sentry/issues/escalating.py"}], "after_files": [{"content": "\"\"\"This module has the logic for querying Snuba for the hourly event count for a list of groups.\nThis is later used for generating group forecasts for determining when a group may be escalating.\n\"\"\"\n\nimport logging\nfrom collections import defaultdict\nfrom datetime import datetime, timedelta\nfrom typing import Dict, List, Optional, Sequence, Tuple, TypedDict\n\nfrom snuba_sdk import (\n    Column,\n    Condition,\n    Direction,\n    Entity,\n    Function,\n    Limit,\n    Offset,\n    Op,\n    OrderBy,\n    Query,\n    Request,\n)\n\nfrom sentry import analytics\nfrom sentry.issues.escalating_group_forecast import EscalatingGroupForecast\nfrom sentry.issues.escalating_issues_alg import GroupCount\nfrom sentry.issues.grouptype import GroupCategory\nfrom sentry.models import Group\nfrom sentry.models.group import GroupStatus\nfrom sentry.models.groupinbox import GroupInboxReason, add_group_to_inbox\nfrom sentry.snuba.dataset import Dataset, EntityKey\nfrom sentry.types.group import GroupSubStatus\nfrom sentry.utils.cache import cache\nfrom sentry.utils.snuba import raw_snql_query\n\nlogger = logging.getLogger(__name__)\n\n__all__ = [\"query_groups_past_counts\", \"parse_groups_past_counts\"]\n\nREFERRER = \"sentry.issues.escalating\"\nELEMENTS_PER_SNUBA_PAGE = 10000  # This is the maximum value for Snuba\n# The amount of data needed to generate a group forecast\nBUCKETS_PER_GROUP = 7 * 24\nONE_WEEK_DURATION = 7\nIS_ESCALATING_REFERRER = \"sentry.issues.escalating.is_escalating\"\nGROUP_HOURLY_COUNT_TTL = 60\n\nGroupsCountResponse = TypedDict(\n    \"GroupsCountResponse\",\n    {\"group_id\": int, \"hourBucket\": str, \"count()\": int, \"project_id\": int},\n)\n\nParsedGroupsCount = Dict[int, GroupCount]\n\n\ndef query_groups_past_counts(groups: Sequence[Group]) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for the counts for every group bucketed into hours.\n\n    It optimizes the query by guaranteeing that we look at group_ids that are from the same project id.\n    This is important for Snuba as the data is stored in blocks related to the project id.\n\n    We maximize the number of projects and groups to reduce the total number of Snuba queries.\n    Each project may not have enough groups in order to reach the max number of returned\n    elements (ELEMENTS_PER_SNUBA_PAGE), thus, projects with few groups should be grouped together until\n    we get at least a certain number of groups.\n\n    NOTE: Groups with less than the maximum number of buckets (think of groups with just 1 event or less\n    than 7 days old) will skew the optimization since we may only get one page and less elements than the max\n    ELEMENTS_PER_SNUBA_PAGE.\n    \"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    start_date, end_date = _start_and_end_dates()\n\n    # Error groups use the events dataset while profile and perf groups use the issue platform dataset\n    error_groups: List[Group] = []\n    other_groups: List[Group] = []\n    for g in groups:\n        if g.issue_category == GroupCategory.ERROR:\n            error_groups.append(g)\n        else:\n            other_groups.append(g)\n\n    all_results += _process_groups(error_groups, start_date, end_date, GroupCategory.ERROR)\n    all_results += _process_groups(other_groups, start_date, end_date)\n\n    return all_results\n\n\ndef _process_groups(\n    groups: Sequence[Group],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory] = None,\n) -> List[GroupsCountResponse]:\n    \"\"\"Given a list of groups, query Snuba for their hourly bucket count.\n    The category defines which Snuba dataset and entity we query.\"\"\"\n    all_results = []  # type: ignore[var-annotated]\n    if not groups:\n        return all_results\n\n    group_ids_by_project = _extract_project_and_group_ids(groups)\n    proj_ids, group_ids = [], []\n    processed_projects = 0\n    total_projects_count = len(group_ids_by_project)\n    organization_id = groups[0].project.organization.id\n\n    # This iteration guarantees that all groups for a project will be queried in the same call\n    # and only one page where the groups could be mixed with groups from another project\n    # Iterating over the sorted keys guarantees results for tests\n    for proj_id in sorted(group_ids_by_project.keys()):\n        _group_ids = group_ids_by_project[proj_id]\n        # Add them to the list of projects and groups to query\n        proj_ids.append(proj_id)\n        group_ids += _group_ids\n        processed_projects += 1\n        potential_num_elements = len(_group_ids) * BUCKETS_PER_GROUP\n        # This is trying to maximize the number of groups on the first page\n        if (\n            processed_projects < total_projects_count\n            and potential_num_elements < ELEMENTS_PER_SNUBA_PAGE\n        ):\n            continue\n\n        # TODO: Write this as a dispatcher type task and fire off a separate task per proj_ids\n        all_results += _query_with_pagination(\n            organization_id, proj_ids, group_ids, start_date, end_date, category\n        )\n        # We're ready for a new set of projects and ids\n        proj_ids, group_ids = [], []\n\n    return all_results\n\n\ndef _query_with_pagination(\n    organization_id: int,\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> List[GroupsCountResponse]:\n    \"\"\"Query Snuba for event counts for the given list of project ids and groups ids in\n    a time range.\"\"\"\n    all_results = []\n    offset = 0\n    while True:\n        query = _generate_query(project_ids, group_ids, offset, start_date, end_date, category)\n        request = Request(\n            dataset=_issue_category_dataset(category),\n            app_id=REFERRER,\n            query=query,\n            tenant_ids={\"referrer\": REFERRER, \"organization_id\": organization_id},\n        )\n        results = raw_snql_query(request, referrer=REFERRER)[\"data\"]\n        all_results += results\n        offset += ELEMENTS_PER_SNUBA_PAGE\n        if not results or len(results) < ELEMENTS_PER_SNUBA_PAGE:\n            break\n\n    return all_results\n\n\ndef _generate_query(\n    project_ids: Sequence[int],\n    group_ids: Sequence[int],\n    offset: int,\n    start_date: datetime,\n    end_date: datetime,\n    category: Optional[GroupCategory],\n) -> Query:\n    \"\"\"This simply generates a query based on the passed parameters\"\"\"\n    group_id_col = Column(\"group_id\")\n    proj_id_col = Column(\"project_id\")\n    return Query(\n        match=Entity(_issue_category_entity(category)),\n        select=[\n            proj_id_col,\n            group_id_col,\n            Function(\"toStartOfHour\", [Column(\"timestamp\")], \"hourBucket\"),\n            Function(\"count\", []),\n        ],\n        groupby=[proj_id_col, group_id_col, Column(\"hourBucket\")],\n        where=[\n            Condition(proj_id_col, Op.IN, Function(\"tuple\", project_ids)),\n            Condition(Column(\"group_id\"), Op.IN, Function(\"tuple\", group_ids)),\n            Condition(Column(\"timestamp\"), Op.GTE, start_date),\n            Condition(Column(\"timestamp\"), Op.LT, end_date),\n        ],\n        limit=Limit(ELEMENTS_PER_SNUBA_PAGE),\n        offset=Offset(offset),\n        orderby=[\n            OrderBy(proj_id_col, Direction.ASC),\n            OrderBy(group_id_col, Direction.ASC),\n            OrderBy(Column(\"hourBucket\"), Direction.ASC),\n        ],\n    )\n\n\ndef _start_and_end_dates(hours: int = BUCKETS_PER_GROUP) -> Tuple[datetime, datetime]:\n    \"\"\"Return the start and end date of N hours time range.\"\"\"\n    end_datetime = datetime.now()\n    return end_datetime - timedelta(hours=hours), end_datetime\n\n\ndef _extract_project_and_group_ids(groups: Sequence[Group]) -> Dict[int, List[int]]:\n    \"\"\"Return all project and group IDs from a list of Group\"\"\"\n    group_ids_by_project: Dict[int, List[int]] = defaultdict(list)\n    for group in groups:\n        group_ids_by_project[group.project_id].append(group.id)\n\n    return group_ids_by_project\n\n\ndef get_group_hourly_count(group: Group) -> int:\n    \"\"\"Return the number of events a group has had today in the last hour\"\"\"\n    key = f\"hourly-group-count:{group.project.id}:{group.id}\"\n    hourly_count = cache.get(key)\n\n    if hourly_count is None:\n        now = datetime.now()\n        current_hour = now.replace(minute=0, second=0, microsecond=0)\n        query = Query(\n            match=Entity(_issue_category_entity(group.issue_category)),\n            select=[\n                Function(\"count\", []),\n            ],\n            where=[\n                Condition(Column(\"project_id\"), Op.EQ, group.project.id),\n                Condition(Column(\"group_id\"), Op.EQ, group.id),\n                Condition(Column(\"timestamp\"), Op.GTE, current_hour),\n                Condition(Column(\"timestamp\"), Op.LT, now),\n            ],\n        )\n        request = Request(\n            dataset=_issue_category_dataset(group.issue_category),\n            app_id=IS_ESCALATING_REFERRER,\n            query=query,\n            tenant_ids={\n                \"referrer\": IS_ESCALATING_REFERRER,\n                \"organization_id\": group.project.organization.id,\n            },\n        )\n        hourly_count = int(\n            raw_snql_query(request, referrer=IS_ESCALATING_REFERRER)[\"data\"][0][\"count()\"]\n        )\n        cache.set(key, hourly_count, GROUP_HOURLY_COUNT_TTL)\n    return int(hourly_count)\n\n\ndef is_escalating(group: Group) -> bool:\n    \"\"\"Return boolean depending on if the group is escalating or not\"\"\"\n    group_hourly_count = get_group_hourly_count(group)\n    forecast_today = EscalatingGroupForecast.fetch_todays_forecast(group.project.id, group.id)\n    # Check if current event occurance is greater than forecast for today's date\n    if group_hourly_count > forecast_today:\n        group.substatus = GroupSubStatus.ESCALATING\n        group.status = GroupStatus.UNRESOLVED\n        group.save()\n        add_group_to_inbox(group, GroupInboxReason.ESCALATING)\n\n        analytics.record(\n            \"issue.escalating\",\n            organization_id=group.project.organization.id,\n            project_id=group.project.id,\n            group_id=group.id,\n        )\n        return True\n    return False\n\n\ndef parse_groups_past_counts(response: Sequence[GroupsCountResponse]) -> ParsedGroupsCount:\n    \"\"\"\n    Return the parsed snuba response for groups past counts to be used in generate_issue_forecast.\n    ParsedGroupCount is of the form {<group_id>: {\"intervals\": [str], \"data\": [int]}}.\n\n    `response`: Snuba response for group event counts\n    \"\"\"\n    group_counts: ParsedGroupsCount = {}\n    group_ids_list = group_counts.keys()\n    for data in response:\n        group_id = data[\"group_id\"]\n        if group_id not in group_ids_list:\n            group_counts[group_id] = {\n                \"intervals\": [data[\"hourBucket\"]],\n                \"data\": [data[\"count()\"]],\n            }\n        else:\n            group_counts[group_id][\"intervals\"].append(data[\"hourBucket\"])\n            group_counts[group_id][\"data\"].append(data[\"count()\"])\n    return group_counts\n\n\ndef _issue_category_dataset(category: Optional[GroupCategory]) -> Dataset:\n    return Dataset.Events.value if category == GroupCategory.ERROR else Dataset.IssuePlatform.value\n\n\ndef _issue_category_entity(category: Optional[GroupCategory]) -> EntityKey:\n    return (\n        EntityKey.Events.value if category == GroupCategory.ERROR else EntityKey.IssuePlatform.value\n    )\n", "path": "src/sentry/issues/escalating.py"}]}