feat(reports): Cache per-project weekly report metrics (#116739)

Resolves
[ID-1589](https://linear.app/getsentry/issue/ID-1589/cache-total-project-errors-and-total-project-transactions-for-weekly).

Goal: Cache per-project total errors and total transactions during
weekly report generation so the frontend can display week-over-week
percentage change without re-querying Snuba.

**Redis Cache Layer**
- Caches total errors and total transactions by Org Id, Project Id (no
timestamp)
- 10 day TTL

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: amy-chen23

src/sentry/tasks/summaries/weekly_report_cache.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from typing import Any
+
+from django.conf import settings
+from sentry_redis_tools.clients import RedisCluster, StrictRedis
+
+from sentry.utils import json, metrics, redis
+
+CACHE_TTL_SEC = 10 * 24 * 60 * 60  # 10 days
+KEY_PREFIX = "wr:proj_metrics"
+
+
+def _make_cache_key(org_id: int, project_id: int) -> str:
+    return f"{KEY_PREFIX}:{org_id}:{project_id}"
+
+
+def _get_redis_client() -> RedisCluster[str] | StrictRedis[str]:
+    return redis.redis_clusters.get(settings.SENTRY_WEEKLY_REPORTS_REDIS_CLUSTER)
+
+
+def cache_project_metrics(
+    org_id: int,
+    project_metrics: dict[int, dict[str, int]],
+) -> None:
+    client = _get_redis_client()
+    pipeline = client.pipeline()
+
+    for project_id, values in project_metrics.items():
+        key = _make_cache_key(org_id, project_id)
+        pipeline.set(key, json.dumps(values), ex=CACHE_TTL_SEC)
+
+    pipeline.execute()
+
+
+def read_project_metrics(
+    org_id: int,
+    project_ids: list[int],
+) -> dict[int, dict[str, Any]]:
+    if not project_ids:
+        return {}
+
+    client = _get_redis_client()
+    pipeline = client.pipeline()
+
+    for project_id in project_ids:
+        pipeline.get(_make_cache_key(org_id, project_id))
+
+    results = pipeline.execute()
+
+    result_map: dict[int, dict[str, Any]] = {}
+    for i, project_id in enumerate(project_ids):
+        raw = results[i]
+        if raw is None:
+            metrics.incr("weekly_report.cache.miss")
+        else:
+            result_map[project_id] = json.loads(raw)
+
+    return result_map

src/sentry/tasks/summaries/weekly_reports.py

@@ -37,6 +37,7 @@
     OrganizationReportContextFactory,
 )
 from sentry.tasks.summaries.utils import ONE_DAY, OrganizationReportContext
+from sentry.tasks.summaries.weekly_report_cache import cache_project_metrics
 from sentry.taskworker.namespaces import reports_tasks
 from sentry.types.group import GroupSubStatus
 from sentry.users.services.user_option import user_option_service
@@ -218,6 +219,20 @@ def prepare_organization_report(
             lifecycle.record_halt(WeeklyReportHaltReason.EMPTY_REPORT)
             return
 
+    if not dry_run:
+        try:
+            project_metrics: dict[int, dict[str, int]] = {}
+            for project_id, project_ctx in ctx.projects_context_map.items():
+                if not project_ctx.check_if_project_is_empty():
+                    project_metrics[project_id] = {
+                        "e": project_ctx.accepted_error_count,
+                        "t": project_ctx.accepted_transaction_count,
+                    }
+            if project_metrics:
+                cache_project_metrics(organization_id, project_metrics)
+        except Exception:
+            sentry_sdk.capture_exception()
+
     # Finally, deliver the reports
     batch = OrganizationReportBatch(ctx, batch_id, dry_run, target_user, email_override)
     with sentry_sdk.start_span(op="weekly_reports.deliver_reports"):

tests/sentry/tasks/test_weekly_report_cache.py

@@ -0,0 +1,53 @@
+from sentry.tasks.summaries.weekly_report_cache import (
+    _make_cache_key,
+    cache_project_metrics,
+    read_project_metrics,
+)
+from sentry.testutils.cases import TestCase
+
+
+class WeeklyReportCacheTest(TestCase):
+    def test_make_cache_key(self) -> None:
+        key = _make_cache_key(org_id=1, project_id=2)
+        assert key == "wr:proj_metrics:1:2"
+
+    def test_write_and_read(self) -> None:
+        org_id = self.organization.id
+        project = self.create_project(organization=self.organization)
+
+        cache_project_metrics(org_id, {project.id: {"e": 500, "t": 3000}})
+
+        result = read_project_metrics(org_id=org_id, project_ids=[project.id])
+
+        assert result[project.id] == {"e": 500, "t": 3000}
+
+    def test_read_empty_cache(self) -> None:
+        result = read_project_metrics(org_id=self.organization.id, project_ids=[999])
+
+        assert result == {}
+
+    def test_read_empty_project_ids(self) -> None:
+        result = read_project_metrics(org_id=self.organization.id, project_ids=[])
+
+        assert result == {}
+
+    def test_write_empty_metrics_is_noop(self) -> None:
+        cache_project_metrics(self.organization.id, {})
+
+    def test_multiple_projects(self) -> None:
+        org_id = self.organization.id
+        p1 = self.create_project(organization=self.organization)
+        p2 = self.create_project(organization=self.organization)
+
+        cache_project_metrics(
+            org_id,
+            {
+                p1.id: {"e": 100, "t": 200},
+                p2.id: {"e": 300, "t": 400},
+            },
+        )
+
+        result = read_project_metrics(org_id=org_id, project_ids=[p1.id, p2.id])
+
+        assert result[p1.id] == {"e": 100, "t": 200}
+        assert result[p2.id] == {"e": 300, "t": 400}