feat(explorer): Add RPC for issue stats (#102693)

Adds RPC to get issue stats, needed for
getsentry/seer#3891

Author: roaga

src/sentry/seer/assisted_query/issues_tools.py

@@ -492,7 +492,7 @@ def execute_issues_query(
         limit: Number of results to return (default 25)
 
     Returns:
-        Issue data from the API response, or None if organization doesn't exist
+        List of issues, or None if organization doesn't exist
     """
     try:
         organization = Organization.objects.get(id=org_id)
@@ -523,3 +523,57 @@ def execute_issues_query(
     )
 
     return resp.data
+
+
+def get_issues_stats(
+    *,
+    org_id: int,
+    issue_ids: list[str],
+    project_ids: list[int],
+    query: str,
+    stats_period: str = "7d",
+) -> list[dict[str, Any]] | None:
+    """
+    Get stats for specific issues by calling the issues-stats endpoint.
+
+    This endpoint provides count, userCount, firstSeen, lastSeen, and
+    timeseries data for each issue.
+
+    Args:
+        org_id: Organization ID
+        issue_ids: List of issue IDs to get stats for
+        project_ids: List of project IDs
+        query: Search query string (e.g., "is:unresolved")
+        stats_period: Time period for the query (e.g., "24h", "7d", "14d"). Defaults to "7d".
+
+    Returns:
+        List of issue stats, or None if organization doesn't exist.
+        Each item contains: id, count, userCount, firstSeen, lastSeen, stats, lifetime
+    """
+    try:
+        organization = Organization.objects.get(id=org_id)
+    except Organization.DoesNotExist:
+        logger.warning("Organization not found", extra={"org_id": org_id})
+        return None
+
+    if not issue_ids:
+        return []
+
+    api_key = ApiKey(organization_id=organization.id, scope_list=API_KEY_SCOPES)
+
+    params: dict[str, Any] = {
+        "project": project_ids,
+        "groups": issue_ids,
+        "query": query,
+        "statsPeriod": stats_period,
+        "referrer": Referrer.SEER_RPC,
+    }
+
+    resp = client.get(
+        auth=api_key,
+        user=None,
+        path=f"/organizations/{organization.slug}/issues-stats/",
+        params=params,
+    )
+
+    return resp.data

src/sentry/seer/endpoints/seer_rpc.py

@@ -71,6 +71,7 @@
     execute_issues_query,
     get_filter_key_values,
     get_issue_filter_keys,
+    get_issues_stats,
 )
 from sentry.seer.autofix.autofix_tools import get_error_event_details, get_profile_details
 from sentry.seer.autofix.coding_agent import launch_coding_agents_for_run
@@ -1134,6 +1135,7 @@ def check_repository_integrations_status(*, repository_integrations: list[dict[s
     "get_issue_filter_keys": get_issue_filter_keys,
     "get_filter_key_values": get_filter_key_values,
     "execute_issues_query": execute_issues_query,
+    "get_issues_stats": get_issues_stats,
     #
     # Explorer
     "get_transactions_for_project": rpc_get_transactions_for_project,

tests/sentry/seer/assisted_query/test_issues_tools.py

@@ -1,9 +1,12 @@
+from datetime import datetime
+
 import pytest
 
 from sentry.seer.assisted_query.issues_tools import (
     execute_issues_query,
     get_filter_key_values,
     get_issue_filter_keys,
+    get_issues_stats,
 )
 from sentry.testutils.cases import APITestCase, SnubaTestCase
 from sentry.testutils.helpers.datetime import before_now
@@ -662,3 +665,230 @@ def test_execute_issues_query_multiple_projects(self):
         project_ids = {issue["project"]["id"] for issue in result}
         assert str(self.project.id) in project_ids
         assert str(project2.id) in project_ids
+
+
+@pytest.mark.django_db(databases=["default", "control"])
+class TestGetIssuesStats(APITestCase, SnubaTestCase):
+    databases = {"default", "control"}
+
+    def setUp(self):
+        super().setUp()
+        self.min_ago = before_now(minutes=1)
+
+    def test_get_issues_stats_success(self):
+        """Test that get_issues_stats returns stats for issues"""
+        # Store two events to create issues
+        event1 = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "First error",
+                "timestamp": self.min_ago.isoformat(),
+            },
+            project_id=self.project.id,
+        )
+        event2 = self.store_event(
+            data={
+                "event_id": "b" * 32,
+                "message": "Second error",
+                "timestamp": self.min_ago.isoformat(),
+            },
+            project_id=self.project.id,
+        )
+
+        issue_ids = [str(event1.group_id), str(event2.group_id)]
+
+        result = get_issues_stats(
+            org_id=self.organization.id,
+            issue_ids=issue_ids,
+            project_ids=[self.project.id],
+            query="is:unresolved",
+            stats_period="24h",
+        )
+
+        assert result is not None
+        assert isinstance(result, list)
+        assert len(result) == 2
+
+        # Verify each stat has the expected fields
+        for stat in result:
+            assert "id" in stat
+            assert "count" in stat
+            assert "userCount" in stat
+            assert "firstSeen" in stat
+            assert "lastSeen" in stat
+            assert stat["id"] in issue_ids
+
+            # Verify stats field structure
+            # stats should be a dict with stats_period keys (e.g., "24h", "14d")
+            # Each value is an array of (timestamp, count) tuples
+            assert "stats" in stat
+            assert isinstance(stat["stats"], dict)
+            # Should have the stats_period key we passed ("24h")
+            assert "24h" in stat["stats"]
+            # The value should be an array of tuples
+            assert isinstance(stat["stats"]["24h"], list)
+            # Each element should be a tuple of (timestamp, count) where both are numbers
+            for data_point in stat["stats"]["24h"]:
+                assert isinstance(data_point, tuple)
+                assert len(data_point) == 2
+                assert isinstance(data_point[0], (int, float))  # timestamp
+                assert isinstance(data_point[1], (int, float))  # count
+
+            # Verify lifetime field structure
+            # lifetime should be a dict with count, userCount, firstSeen, lastSeen
+            assert "lifetime" in stat
+            assert isinstance(stat["lifetime"], dict)
+            assert "count" in stat["lifetime"]
+            assert "userCount" in stat["lifetime"]
+            assert "firstSeen" in stat["lifetime"]
+            assert "lastSeen" in stat["lifetime"]
+            # count should be a string representation of the number
+            assert isinstance(stat["lifetime"]["count"], str)
+            # userCount should be an integer
+            assert isinstance(stat["lifetime"]["userCount"], int)
+            # firstSeen and lastSeen are datetime objects or None
+            assert stat["lifetime"]["firstSeen"] is None or isinstance(
+                stat["lifetime"]["firstSeen"], datetime
+            )
+            assert stat["lifetime"]["lastSeen"] is None or isinstance(
+                stat["lifetime"]["lastSeen"], datetime
+            )
+
+    def test_get_issues_stats_with_multiple_projects(self):
+        """Test that get_issues_stats works with multiple project IDs"""
+        project2 = self.create_project(organization=self.organization)
+
+        event1 = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "Project 1 error",
+                "timestamp": self.min_ago.isoformat(),
+            },
+            project_id=self.project.id,
+        )
+        event2 = self.store_event(
+            data={
+                "event_id": "b" * 32,
+                "message": "Project 2 error",
+                "timestamp": self.min_ago.isoformat(),
+            },
+            project_id=project2.id,
+        )
+
+        issue_ids = [str(event1.group_id), str(event2.group_id)]
+
+        result = get_issues_stats(
+            org_id=self.organization.id,
+            issue_ids=issue_ids,
+            project_ids=[self.project.id, project2.id],
+            query="is:unresolved",
+            stats_period="24h",
+        )
+
+        assert result is not None
+        assert isinstance(result, list)
+        # Should return stats for both issues
+        assert len(result) >= 2
+        returned_issue_ids = {stat["id"] for stat in result}
+        assert str(event1.group_id) in returned_issue_ids
+        assert str(event2.group_id) in returned_issue_ids
+
+    def test_get_issues_stats_nonexistent_org(self):
+        """Test that get_issues_stats returns None for nonexistent org"""
+        result = get_issues_stats(
+            org_id=999999,
+            issue_ids=["123"],
+            project_ids=[self.project.id],
+            query="is:unresolved",
+            stats_period="24h",
+        )
+
+        assert result is None
+
+    def test_get_issues_stats_empty_issue_ids(self):
+        """Test that get_issues_stats handles empty issue IDs"""
+        result = get_issues_stats(
+            org_id=self.organization.id,
+            issue_ids=[],
+            project_ids=[self.project.id],
+            query="is:unresolved",
+            stats_period="24h",
+        )
+
+        assert result is not None
+        assert isinstance(result, list)
+        assert len(result) == 0
+
+    def test_get_issues_stats_stats_and_lifetime_structure(self):
+        """Test that stats and lifetime fields have the correct structure"""
+        # Create an issue
+        event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "Test error",
+                "timestamp": self.min_ago.isoformat(),
+            },
+            project_id=self.project.id,
+        )
+
+        result = get_issues_stats(
+            org_id=self.organization.id,
+            issue_ids=[str(event.group_id)],
+            project_ids=[self.project.id],
+            query="is:unresolved",
+            stats_period="24h",
+        )
+
+        assert result is not None
+        assert len(result) == 1
+
+        stat = result[0]
+
+        # Verify stats structure:
+        # stats is a dict where keys are stats_period strings (e.g., "24h", "14d")
+        # and values are arrays of (timestamp, count) tuples
+        assert "stats" in stat
+        assert isinstance(stat["stats"], dict)
+        assert "24h" in stat["stats"]
+        assert isinstance(stat["stats"]["24h"], list)
+        # Each element should be a tuple of (timestamp, count)
+        for timepoint in stat["stats"]["24h"]:
+            assert isinstance(timepoint, tuple)
+            assert len(timepoint) == 2
+            timestamp, count = timepoint
+            assert isinstance(timestamp, (int, float))
+            assert isinstance(count, (int, float))
+            # Timestamp should be a reasonable Unix timestamp (seconds since epoch)
+            # For 24h period, should be within last day
+            assert timestamp > 0
+
+        # Verify lifetime structure:
+        # lifetime is a dict with count (string), userCount (int), firstSeen (datetime), lastSeen (datetime)
+        assert "lifetime" in stat
+        assert isinstance(stat["lifetime"], dict)
+        lifetime = stat["lifetime"]
+
+        # Required fields
+        assert "count" in lifetime
+        assert "userCount" in lifetime
+        assert "firstSeen" in lifetime
+        assert "lastSeen" in lifetime
+
+        # Field types
+        assert isinstance(lifetime["count"], str)
+        # Count should be a numeric string representing the total times seen
+        # (e.g., "1", "42", "1000")
+        assert lifetime["count"].isdigit()
+
+        assert isinstance(lifetime["userCount"], int)
+
+        # firstSeen and lastSeen are datetime objects or None
+        if lifetime["firstSeen"] is not None:
+            assert isinstance(lifetime["firstSeen"], datetime)
+
+        if lifetime["lastSeen"] is not None:
+            assert isinstance(lifetime["lastSeen"], datetime)
+
+        # Optional stats field in lifetime (currently None in implementation)
+        if "stats" in lifetime:
+            assert lifetime["stats"] is None or isinstance(lifetime["stats"], dict)