perf(core): Wave 2 part1 — 复合索引(只生成不apply) + 消息中心 N+1 治理

索引(只 makemigrations 看 SQL,远程 MySQL apply 留用户低峰 / rule 3):
- Asset (team,category,-created_at)+(team,asset_type)、CreditLedger (team,-created_at)+
  (team,ledger_type,created_at)、Project (team,-updated_at)、AITask (team,-created_at)、
  Product (team,-created_at);迁移 assets0005/billing0002/projects0004/ai0008/products0002

消息中心 N+1(ops/views.py):
- type_counts 原 6 次 count(每请求)→ 一次 aggregate(Count(filter=...)) 条件聚合
- ensure_team_notifications 项目花费原逐项目 aggregate(N+1)→ 一次 values('project').annotate(Sum)

验证: apps.ops 测试 3/3 OK;全量 6fail+1err 均为预存 image provider 路由失败(stash 跑 baseline
  复现)= 零新增回归;索引迁移在 sqlite 测试库成功 apply

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/backend/apps/ops/views.py

@@ -1,7 +1,7 @@
 from datetime import timedelta
 
 from django.core.cache import cache
-from django.db.models import Q, Sum
+from django.db.models import Count, Q, Sum
 from django.utils import timezone
 from rest_framework import status
 from rest_framework.decorators import action
@@ -107,12 +107,20 @@ def ensure_team_notifications(team, user):
         metadata={"timeline": [[_step_time(timezone.now()), "团队接入 AirShelf,真实消息中心已启用"]]},
     )
 
-    for project in Project.objects.filter(team=team).select_related("product", "created_by").order_by("-updated_at")[:5]:
+    recent_projects = list(
+        Project.objects.filter(team=team).select_related("product", "created_by").order_by("-updated_at")[:5]
+    )
+    # 5 个项目的累计扣费一次聚合(原先每个项目各跑一次 aggregate = N+1)
+    spend_by_project = dict(
+        CreditLedger.objects.filter(project__in=recent_projects, ledger_type=CreditLedger.Type.CHARGE)
+        .values("project")
+        .annotate(total=Sum("amount"))
+        .values_list("project", "total")
+    )
+    for project in recent_projects:
         product_title = project.product.title if project.product_id else "未绑定商品"
         # 项目累计花费 = 该项目所有 AI 扣费流水之和(脚本/图片/视频/导出),无扣费则显示「-」
-        project_spend = project.credit_ledgers.filter(
-            ledger_type=CreditLedger.Type.CHARGE
-        ).aggregate(total=Sum("amount"))["total"]
+        project_spend = spend_by_project.get(project.id)
         project_cost = f"¥{project_spend:.2f}" if project_spend else "-"
         create_once(
             f"project:{project.id}:status:{project.status}:{project.current_stage}",
@@ -280,15 +288,24 @@ class NotificationViewSet(TeamScopedViewSetMixin, ModelViewSet):
         if isinstance(data, dict):
             # 分类 chip 计数取绝对总数(忽略当前 tab/搜索),与设计稿一致
             base = self._recipient_scope()
-            unread_count = base.filter(is_read=False).count()
+            # 原先 6 次独立 count(每次 list 请求都跑)合成一次条件聚合,接口少 5 个 DB 往返
+            counts = base.aggregate(
+                all=Count("id"),
+                unread=Count("id", filter=Q(is_read=False)),
+                task=Count("id", filter=Q(notification_type="task")),
+                team=Count("id", filter=Q(notification_type="team")),
+                billing=Count("id", filter=Q(notification_type="billing")),
+                system=Count("id", filter=Q(notification_type="system")),
+            )
+            unread_count = counts["unread"]
             data["unread_count"] = unread_count
             data["type_counts"] = {
-                "all": base.count(),
-                "unread": unread_count,
-                "task": base.filter(notification_type="task").count(),
-                "team": base.filter(notification_type="team").count(),
-                "billing": base.filter(notification_type="billing").count(),
-                "system": base.filter(notification_type="system").count(),
+                "all": counts["all"],
+                "unread": counts["unread"],
+                "task": counts["task"],
+                "team": counts["team"],
+                "billing": counts["billing"],
+                "system": counts["system"],
             }
         return response