feat(awooop): show recurring alert links

apps/api/src/api/v1/platform/events.py

@@ -16,6 +16,7 @@ from pydantic import BaseModel
 from src.services.channel_event_dossier_service import (
     fetch_channel_event_dossier,
     fetch_channel_event_dossier_coverage,
+    fetch_channel_event_dossier_recurrence,
 )
 from src.services.platform_operator_service import list_recent_channel_events
 
@@ -115,6 +116,50 @@ class ChannelEventDossierCoverageResponse(BaseModel):
     providers: list[ChannelEventProviderCoverage]
 
 
+class ChannelEventRecurrenceSummary(BaseModel):
+    source_event_total: int
+    recurrence_group_total: int
+    recurrent_group_total: int
+    duplicate_event_total: int
+    linked_run_total: int
+    unlinked_event_total: int
+    latest_received_at: datetime | None
+
+
+class ChannelEventRecurrenceItem(BaseModel):
+    recurrence_key: str
+    provider: str | None
+    alertname: str | None
+    severity: str | None
+    namespace: str | None
+    target_resource: str | None
+    fingerprint: str | None
+    latest_event_id: UUID | None
+    latest_provider_event_id: str | None
+    latest_content_preview: str | None
+    latest_run_id: UUID | None
+    latest_run_state: str | None
+    latest_agent_id: str | None
+    occurrence_total: int
+    duplicate_total: int
+    linked_run_total: int
+    source_ref_total: int
+    missing_source_refs_total: int
+    sentry_ref_total: int
+    signoz_ref_total: int
+    alert_ref_total: int
+    run_state_counts: dict[str, int]
+    first_received_at: datetime | None
+    latest_received_at: datetime | None
+
+
+class ChannelEventRecurrenceResponse(BaseModel):
+    project_id: str
+    limit: int
+    summary: ChannelEventRecurrenceSummary
+    items: list[ChannelEventRecurrenceItem]
+
+
 @router.get(
     "/events/dossier",
     response_model=ChannelEventDossierResponse,
@@ -159,6 +204,27 @@ async def get_event_dossier_coverage(
     )
 
 
+@router.get(
+    "/events/dossier/recurrence",
+    response_model=ChannelEventRecurrenceResponse,
+    summary="查詢 Channel Event 重複發生與關聯 Run 狀態",
+    description=(
+        "將近期 inbound source events 依 fingerprint / alertname / namespace / target 分組，"
+        "顯示重複發生次數、去重數、source refs 與最新 linked run 狀態。"
+    ),
+)
+async def get_event_dossier_recurrence(
+    project_id: str | None = Query(None, description="租戶 ID（可選）"),
+    provider: str | None = Query(None, description="provider（可選，如 alertmanager / sentry / signoz）"),
+    limit: int = Query(100, ge=1, le=300, description="最多納入統計筆數"),
+) -> dict[str, Any]:
+    return await fetch_channel_event_dossier_recurrence(
+        project_id=project_id,
+        provider=provider,
+        limit=limit,
+    )
+
+
 @router.get(
     "/events/recent",
     response_model=RecentEventsResponse,

apps/api/src/services/channel_event_dossier_service.py

@@ -17,6 +17,7 @@ from src.db.base import get_db_context
 
 _MAX_DOSSIER_EVENTS = 50
 _MAX_COVERAGE_EVENTS = 200
+_MAX_RECURRENCE_EVENTS = 300
 
 
 def _as_dict(value: Any) -> dict[str, Any]:
@@ -40,6 +41,135 @@ def _ref_count(source_refs: dict[str, Any], key: str) -> int:
     return 1 if value else 0
 
 
+def _recurrence_key(event: dict[str, Any]) -> str:
+    fingerprint = str(event.get("fingerprint") or "").strip()
+    if fingerprint:
+        return f"fingerprint:{fingerprint}"
+
+    provider = str(event.get("provider") or event.get("channel_type") or "unknown")
+    alertname = str(event.get("alertname") or "").strip()
+    namespace = str(event.get("namespace") or "").strip()
+    target = str(event.get("target_resource") or "").strip()
+    if alertname or namespace or target:
+        return f"alert:{provider}:{alertname}:{namespace}:{target}"
+
+    return f"event:{provider}:{event.get('provider_event_id')}"
+
+
+def build_dossier_recurrence(
+    rows: list[dict[str, Any]],
+    *,
+    project_id: str,
+    limit: int,
+) -> dict[str, Any]:
+    """Group recent source events into recurrence buckets with linked run state."""
+    groups: dict[str, dict[str, Any]] = {}
+
+    for row in rows:
+        event = build_dossier_event(row)
+        key = _recurrence_key(event)
+        source_ref_count = int(event.get("source_ref_count") or 0)
+        source_refs = _as_dict(event.get("source_refs"))
+        run_id = row.get("run_id")
+        run_state = row.get("run_state")
+        received_at = event.get("received_at")
+
+        group = groups.setdefault(
+            key,
+            {
+                "recurrence_key": key,
+                "provider": event.get("provider"),
+                "alertname": event.get("alertname"),
+                "severity": event.get("severity"),
+                "namespace": event.get("namespace"),
+                "target_resource": event.get("target_resource"),
+                "fingerprint": event.get("fingerprint"),
+                "latest_event_id": event.get("event_id"),
+                "latest_provider_event_id": event.get("provider_event_id"),
+                "latest_content_preview": event.get("content_preview"),
+                "latest_run_id": run_id,
+                "latest_run_state": run_state,
+                "latest_agent_id": row.get("run_agent_id"),
+                "occurrence_total": 0,
+                "duplicate_total": 0,
+                "linked_run_total": 0,
+                "source_ref_total": 0,
+                "missing_source_refs_total": 0,
+                "sentry_ref_total": 0,
+                "signoz_ref_total": 0,
+                "alert_ref_total": 0,
+                "run_state_counts": {},
+                "first_received_at": received_at,
+                "latest_received_at": received_at,
+                "_run_ids": set(),
+            },
+        )
+
+        group["occurrence_total"] += 1
+        group["source_ref_total"] += source_ref_count
+        if source_ref_count <= 0:
+            group["missing_source_refs_total"] += 1
+        if event.get("is_duplicate"):
+            group["duplicate_total"] += 1
+
+        group["sentry_ref_total"] += _ref_count(source_refs, "sentry_issue_ids")
+        group["signoz_ref_total"] += _ref_count(source_refs, "signoz_alerts")
+        group["alert_ref_total"] += _ref_count(source_refs, "alert_ids")
+
+        if run_id:
+            group["_run_ids"].add(str(run_id))
+            if group.get("latest_run_id") is None:
+                group["latest_run_id"] = run_id
+                group["latest_run_state"] = run_state
+                group["latest_agent_id"] = row.get("run_agent_id")
+        if run_state:
+            state_counts = group["run_state_counts"]
+            state_counts[str(run_state)] = int(state_counts.get(str(run_state), 0)) + 1
+
+        if received_at and (
+            group.get("first_received_at") is None
+            or str(received_at) < str(group.get("first_received_at"))
+        ):
+            group["first_received_at"] = received_at
+        if received_at and (
+            group.get("latest_received_at") is None
+            or str(received_at) > str(group.get("latest_received_at"))
+        ):
+            group["latest_received_at"] = received_at
+
+    items = []
+    linked_run_total = 0
+    for group in groups.values():
+        run_ids = group.pop("_run_ids")
+        group["linked_run_total"] = len(run_ids)
+        linked_run_total += len(run_ids)
+        items.append(group)
+
+    items.sort(key=lambda item: str(item.get("latest_received_at") or ""), reverse=True)
+    items.sort(key=lambda item: int(item.get("occurrence_total") or 0), reverse=True)
+    latest_received_at = max(
+        (item.get("latest_received_at") for item in items if item.get("latest_received_at")),
+        default=None,
+    )
+
+    return {
+        "project_id": project_id,
+        "limit": limit,
+        "summary": {
+            "source_event_total": len(rows),
+            "recurrence_group_total": len(items),
+            "recurrent_group_total": sum(
+                1 for item in items if int(item.get("occurrence_total") or 0) > 1
+            ),
+            "duplicate_event_total": sum(int(item.get("duplicate_total") or 0) for item in items),
+            "linked_run_total": linked_run_total,
+            "unlinked_event_total": sum(1 for row in rows if not row.get("run_id")),
+            "latest_received_at": latest_received_at,
+        },
+        "items": items,
+    }
+
+
 def build_dossier_coverage(
     rows: list[dict[str, Any]],
     *,
@@ -290,3 +420,62 @@ async def fetch_channel_event_dossier_coverage(
         project_id=effective_project_id,
         limit=safe_limit,
     )
+
+
+async def fetch_channel_event_dossier_recurrence(
+    *,
+    project_id: str | None,
+    provider: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    """Fetch recurrence groups and linked run state for recent source events."""
+    effective_project_id = project_id or "awoooi"
+    safe_limit = max(1, min(limit, _MAX_RECURRENCE_EVENTS))
+    where_clauses = ["e.project_id = :project_id"]
+    params: dict[str, Any] = {
+        "project_id": effective_project_id,
+        "limit": safe_limit,
+    }
+    if provider:
+        where_clauses.append(
+            "COALESCE(NULLIF(e.source_envelope->>'provider', ''), "
+            "split_part(e.provider_event_id, ':', 1), e.channel_type) = :provider"
+        )
+        params["provider"] = provider
+
+    async with get_db_context(effective_project_id) as db:
+        result = await db.execute(
+            text(f"""
+                SELECT
+                    e.event_id,
+                    e.project_id,
+                    e.channel_type,
+                    e.provider_event_id,
+                    e.content_hash,
+                    e.content_preview,
+                    e.content_redacted,
+                    e.redaction_version,
+                    e.source_envelope,
+                    e.is_duplicate,
+                    e.provider_ts,
+                    e.received_at,
+                    e.run_id,
+                    r.state AS run_state,
+                    r.agent_id AS run_agent_id
+                FROM awooop_conversation_event e
+                LEFT JOIN awooop_run_state r
+                  ON r.project_id = e.project_id
+                 AND r.run_id = e.run_id
+                WHERE {" AND ".join(where_clauses)}
+                ORDER BY e.received_at DESC
+                LIMIT :limit
+            """),
+            params,
+        )
+        rows = [dict(row) for row in result.mappings().all()]
+
+    return build_dossier_recurrence(
+        rows,
+        project_id=effective_project_id,
+        limit=safe_limit,
+    )