Fix ops payment incident grouping

2026-06-01 13:49:47 +08:00
parent 49cb0bb42d
commit 31fb1b3f28
8 changed files with 410 additions and 32 deletions
@@ -33,7 +33,9 @@ export function runTests() {
      paymentsPage.includes("Intent 卡住") &&
      paymentsPage.includes("积分异常") &&
      paymentsPage.includes("推荐奖励结算") &&
-      paymentsPage.includes("月度邀请封顶"),
+      paymentsPage.includes("月度邀请封顶") &&
+      paymentsPage.includes("occurrence_count") &&
+      paymentsPage.includes("支付事件不匹配"),
    "ops payment page must surface trial, stuck intent, referral, points, and monthly-cap risk signals",
  );
  assert(
@@ -58,6 +58,8 @@ function paymentReasonLabel(reason?: string) {
  if (key === "tx_not_found") return "链上交易未找到";
  if (key === "tx_reverted") return "链上交易失败";
  if (key === "expired") return "订单已过期";
+  if (key === "event_mismatch") return "支付事件不匹配";
+  if (key === "direct_transfer_mismatch") return "直接转账不匹配";
  if (key === "unknown") return "未知原因";
  return key || "未知原因";
 }
@@ -134,7 +136,7 @@ export function PaymentsPageClient() {
  const reasonCounts: Record<string, number> = {};
  incidents.forEach((inc) => {
    const r = inc.reason || "unknown";
-    reasonCounts[r] = (reasonCounts[r] || 0) + 1;
+    reasonCounts[r] = (reasonCounts[r] || 0) + Math.max(1, Number(inc.occurrence_count ?? 1));
  });

  const incidentPieData = Object.entries(reasonCounts).map(([name, value]) => ({
@@ -318,6 +320,11 @@ export function PaymentsPageClient() {
                      <td className="py-2 pr-4 text-slate-500 font-mono">{inc.id}</td>
                      <td className="py-2 pr-4">
                        <div className="font-bold text-amber-600">{paymentReasonLabel(inc.reason)}</div>
+                        {Number(inc.occurrence_count ?? 1) > 1 ? (
+                          <div className="mt-0.5 text-[11px] font-semibold text-slate-400">
+                            同类重复 {Number(inc.occurrence_count).toLocaleString()} 次 · 最早 {compactDate(inc.first_seen_at)}
+                          </div>
+                        ) : null}
                        <div className="mt-0.5 max-w-xl truncate text-xs text-slate-500" title={inc.detail || inc.reason || ""}>
                          {inc.detail || inc.reason || "—"}
                        </div>
@@ -85,11 +85,14 @@ export type SourceHealthPayload = {
 };

 export type PaymentRuntimePayload = {
-  rpc?: string;
+  rpc?: Record<string, unknown> | string;
  chain_id?: number;
  receiver_contract?: string;
  last_scanned_block?: number;
  audit_events_count?: number;
+  checkout?: Record<string, unknown>;
+  event_loop_state?: Record<string, unknown>;
+  recent_audit_events?: Array<Record<string, unknown>>;
  recent_events?: Array<Record<string, unknown>>;
 };

@@ -106,6 +109,10 @@ export type PaymentIncident = {
  resolved?: boolean;
  resolved_at?: string;
  resolved_by?: string;
+  occurrence_count?: number;
+  event_ids?: number[];
+  first_seen_at?: string;
+  last_seen_at?: string;
 };

 export type IncidentsPayload = {
@@ -7,7 +7,7 @@ import threading
 import time
 from collections import Counter
 from datetime import datetime, timedelta
-from typing import Optional, Dict, Any, List, Set
+from typing import Optional, Dict, Any, List, Set, Tuple
 from urllib.parse import urlparse

 import requests
@@ -1291,6 +1291,124 @@ class DBManager:
                "created_at": row["created_at"],
            }

+    @staticmethod
+    def _payment_audit_resolution_key(
+        event_type: str,
+        payload: Dict[str, Any],
+    ) -> Tuple[str, str, str, str, str]:
+        confirm_failure = (
+            payload.get("confirm_failure")
+            if isinstance(payload.get("confirm_failure"), dict)
+            else {}
+        )
+        reason = str(
+            payload.get("reason")
+            or confirm_failure.get("reason")
+            or payload.get("error")
+            or "unknown"
+        ).strip().lower()
+        intent_id = str(
+            payload.get("intent_id")
+            or payload.get("payment_intent_id")
+            or confirm_failure.get("intent_id")
+            or ""
+        ).strip().lower()
+        user_id = str(payload.get("user_id") or "").strip().lower()
+        tx_hash = str(
+            payload.get("tx_hash")
+            or confirm_failure.get("tx_hash")
+            or ""
+        ).strip().lower()
+        return str(event_type or "").strip().lower(), reason, user_id, intent_id, tx_hash
+
+    def mark_related_payment_audit_events_resolved(
+        self,
+        event_id: int,
+        resolved_by: str,
+    ) -> List[Dict[str, Any]]:
+        safe_id = int(event_id or 0)
+        actor = str(resolved_by or "").strip().lower()
+        if safe_id <= 0 or not actor:
+            return []
+
+        with self._get_connection() as conn:
+            conn.row_factory = sqlite3.Row
+            target = conn.execute(
+                """
+                SELECT id, event_type, payload_json, created_at
+                FROM payment_audit_events
+                WHERE id = ?
+                LIMIT 1
+                """,
+                (safe_id,),
+            ).fetchone()
+            if not target:
+                return []
+
+            try:
+                target_payload = json.loads(str(target["payload_json"] or "{}"))
+            except Exception:
+                target_payload = {}
+            if not isinstance(target_payload, dict):
+                target_payload = {}
+
+            target_key = self._payment_audit_resolution_key(
+                str(target["event_type"] or ""),
+                target_payload,
+            )
+            if not (target_key[3] or target_key[4]):
+                single = self.mark_payment_audit_event_resolved(safe_id, actor)
+                return [single] if single else []
+
+            rows = conn.execute(
+                """
+                SELECT id, event_type, payload_json, created_at
+                FROM payment_audit_events
+                WHERE event_type = ?
+                ORDER BY id DESC
+                """,
+                (str(target["event_type"] or ""),),
+            ).fetchall()
+
+            resolved_at = datetime.now().isoformat()
+            resolved_rows: List[Dict[str, Any]] = []
+            for row in rows:
+                try:
+                    payload = json.loads(str(row["payload_json"] or "{}"))
+                except Exception:
+                    payload = {}
+                if not isinstance(payload, dict):
+                    payload = {}
+                if str(payload.get("resolved_at") or "").strip():
+                    continue
+                if self._payment_audit_resolution_key(
+                    str(row["event_type"] or ""),
+                    payload,
+                ) != target_key:
+                    continue
+
+                payload["resolved_at"] = resolved_at
+                payload["resolved_by"] = actor
+                conn.execute(
+                    """
+                    UPDATE payment_audit_events
+                    SET payload_json = ?
+                    WHERE id = ?
+                    """,
+                    (json.dumps(payload, ensure_ascii=False), int(row["id"])),
+                )
+                resolved_rows.append(
+                    {
+                        "id": int(row["id"]),
+                        "event_type": str(row["event_type"] or ""),
+                        "payload": payload,
+                        "created_at": row["created_at"],
+                    }
+                )
+
+            conn.commit()
+            return resolved_rows
+
    @staticmethod
    def _safe_week_key(value: str) -> str:
        text = str(value or "").strip()
@@ -548,6 +548,52 @@ def test_payment_runtime_state_and_audit_event_roundtrip(tmp_path):
    assert events[0]["payload"]["events"] == 2


+def test_mark_related_payment_audit_events_resolved(tmp_path):
+    db_path = tmp_path / "payments.db"
+    db = DBManager(str(db_path))
+
+    common_payload = {
+        "reason": "event_mismatch",
+        "intent_id": "intent-1",
+        "user_id": "user-1",
+        "tx_hash": "0x" + "1" * 64,
+    }
+    db.append_payment_audit_event("payment_intent_failed", dict(common_payload))
+    db.append_payment_audit_event("payment_intent_failed", dict(common_payload))
+    db.append_payment_audit_event(
+        "payment_intent_failed",
+        {
+            "reason": "event_mismatch",
+            "intent_id": "intent-2",
+            "user_id": "user-1",
+            "tx_hash": "0x" + "2" * 64,
+        },
+    )
+
+    target_id = db.list_payment_audit_events(
+        limit=10,
+        event_type="payment_intent_failed",
+    )[-1]["id"]
+
+    resolved = db.mark_related_payment_audit_events_resolved(
+        target_id,
+        "ops@example.com",
+    )
+    events = db.list_payment_audit_events(limit=10, event_type="payment_intent_failed")
+    grouped = [
+        event for event in events
+        if event["payload"].get("intent_id") == "intent-1"
+    ]
+    unrelated = [
+        event for event in events
+        if event["payload"].get("intent_id") == "intent-2"
+    ][0]
+
+    assert len(resolved) == 2
+    assert all(event["payload"].get("resolved_by") == "ops@example.com" for event in grouped)
+    assert not unrelated["payload"].get("resolved_at")
+
+
 def test_paid_subscription_replaces_active_trial_immediately(monkeypatch, tmp_path):
    _payment_env(monkeypatch, tmp_path)
    service = PaymentContractCheckoutService()
@@ -599,6 +599,84 @@ def test_ops_payment_incidents_expose_top_level_reason_and_filters_resolved(monk
    assert incident["resolved"] is False


+def test_ops_payment_incidents_group_duplicate_failures(monkeypatch):
+    from src.database.db_manager import DBManager
+
+    older = "2026-05-25T12:26:44"
+    newer = "2026-05-25T12:29:51"
+
+    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
+    monkeypatch.setattr(
+        DBManager,
+        "list_payment_audit_events",
+        lambda self, limit=50, event_type=None: [
+            {
+                "id": 275751,
+                "event_type": "payment_intent_failed",
+                "created_at": newer,
+                "payload": {
+                    "reason": "event_mismatch",
+                    "detail": "OrderPaid event mismatch",
+                    "intent_id": "intent-1",
+                    "user_id": "user-1",
+                    "tx_hash": "0x" + "1" * 64,
+                },
+            },
+            {
+                "id": 275730,
+                "event_type": "payment_intent_failed",
+                "created_at": older,
+                "payload": {
+                    "reason": "event_mismatch",
+                    "detail": "OrderPaid event mismatch",
+                    "intent_id": "intent-1",
+                    "user_id": "user-1",
+                    "tx_hash": "0x" + "1" * 64,
+                },
+            },
+        ],
+    )
+
+    payload = ops_api.list_ops_payment_incidents(None, limit=20)
+
+    assert payload["total"] == 1
+    assert payload["raw_total"] == 2
+    incident = payload["incidents"][0]
+    assert incident["id"] == 275751
+    assert incident["occurrence_count"] == 2
+    assert incident["event_ids"] == [275751, 275730]
+    assert incident["first_seen_at"] == older
+    assert incident["last_seen_at"] == newer
+
+
+def test_ops_resolve_payment_incident_marks_duplicate_group(monkeypatch):
+    from src.database.db_manager import DBManager
+
+    called = {}
+
+    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
+
+    def mark_related(self, event_id, resolved_by):
+        called["event_id"] = event_id
+        called["resolved_by"] = resolved_by
+        return [
+            {"id": 275751, "payload": {"resolved_at": "now"}},
+            {"id": 275730, "payload": {"resolved_at": "now"}},
+        ]
+
+    monkeypatch.setattr(
+        DBManager,
+        "mark_related_payment_audit_events_resolved",
+        mark_related,
+        raising=False,
+    )
+
+    payload = ops_api.resolve_ops_payment_incident(None, 275751)
+
+    assert called == {"event_id": 275751, "resolved_by": "ops@example.com"}
+    assert payload["resolved_count"] == 2
+
+
 def test_cities_endpoint_uses_denver_display_name_for_aurora_market():
    response = client.get("/api/cities")
    assert response.status_code == 200
@@ -1203,6 +1281,47 @@ def test_payment_runtime_endpoint_returns_shape():
    assert 'recent_audit_events' in payload


+def test_payment_runtime_endpoint_returns_ops_summary_fields(monkeypatch):
+    from src.database.db_manager import DBManager
+
+    monkeypatch.setattr(
+        routes.PAYMENT_CHECKOUT,
+        "get_config_payload",
+        lambda: {
+            "enabled": True,
+            "chain_id": 137,
+            "receiver_contract": "0x351a1bca5f49dd0046a7cf0bafa7e12fa6441c3a",
+        },
+    )
+    monkeypatch.setattr(
+        routes.PAYMENT_CHECKOUT,
+        "get_rpc_runtime_status",
+        lambda: {"connected": True, "chain_id": 137},
+    )
+    monkeypatch.setattr(
+        DBManager,
+        "get_payment_runtime_state",
+        lambda self, key: {"last_scanned_block": 123456},
+    )
+    monkeypatch.setattr(
+        DBManager,
+        "list_payment_audit_events",
+        lambda self, limit=20, event_type=None: [
+            {"id": 1, "event_type": "event_loop_cycle", "payload": {}, "created_at": "now"},
+            {"id": 2, "event_type": "payment_intent_failed", "payload": {}, "created_at": "now"},
+        ],
+    )
+
+    response = client.get("/api/payments/runtime")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["chain_id"] == 137
+    assert payload["receiver_contract"] == "0x351a1bca5f49dd0046a7cf0bafa7e12fa6441c3a"
+    assert payload["last_scanned_block"] == 123456
+    assert payload["audit_events_count"] == 2
+
+
 def test_payment_config_does_not_require_entitlement(monkeypatch):
    monkeypatch.setattr(
        routes,
@@ -4,7 +4,7 @@ from __future__ import annotations

 import os
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple

 from fastapi import HTTPException, Request
 import requests as _requests
@@ -425,20 +425,26 @@ def _normalize_payment_incident(item: Dict[str, Any]) -> Dict[str, Any]:
    }


-def list_ops_payment_incidents(
-    request: Request,
-    limit: int = 50,
+def _payment_incident_group_key(item: Dict[str, Any]) -> Tuple[str, str, str, str]:
+    reason = str(item.get("reason") or "unknown").strip().lower()
+    intent_id = str(item.get("intent_id") or "").strip().lower()
+    tx_hash = str(item.get("tx_hash") or "").strip().lower()
+    user_id = str(item.get("user_id") or "").strip().lower()
+    if intent_id or tx_hash:
+        return reason, user_id, intent_id, tx_hash
+    return reason, user_id, f"event:{item.get('id')}", ""
+
+
+def _group_payment_incidents(
+    incidents: List[Dict[str, Any]],
+    *,
    reason: str = "",
    include_resolved: bool = False,
 ) -> Dict[str, Any]:
-    _require_ops(request)
-    db = DBManager()
-    incidents = db.list_payment_audit_events(
-        limit=max(1, min(int(limit or 50), 200)),
-        event_type="payment_intent_failed",
-    )
    normalized_reason = str(reason or "").strip().lower()
-    filtered = []
+    groups: Dict[Tuple[str, str, str, str], Dict[str, Any]] = {}
+    raw_total = 0
+
    for item in incidents:
        normalized_item = _normalize_payment_incident(item)
        item_reason = str(normalized_item.get("reason") or "").strip().lower()
@@ -447,19 +453,86 @@ def list_ops_payment_incidents(
            continue
        if not include_resolved and resolved:
            continue
-        filtered.append(normalized_item)
-    return {"incidents": filtered, "total": len(filtered)}
+        raw_total += 1
+
+        key = _payment_incident_group_key(normalized_item)
+        created_at = str(normalized_item.get("created_at") or "").strip()
+        event_id = int(normalized_item.get("id") or 0)
+        existing = groups.get(key)
+        if existing is None:
+            grouped = {
+                **normalized_item,
+                "occurrence_count": 1,
+                "event_ids": [event_id] if event_id > 0 else [],
+                "first_seen_at": created_at,
+                "last_seen_at": created_at,
+            }
+            groups[key] = grouped
+            continue
+
+        existing["occurrence_count"] = int(existing.get("occurrence_count") or 1) + 1
+        if event_id > 0:
+            existing.setdefault("event_ids", []).append(event_id)
+        first_seen = str(existing.get("first_seen_at") or "").strip()
+        last_seen = str(existing.get("last_seen_at") or "").strip()
+        if created_at and (not first_seen or created_at < first_seen):
+            existing["first_seen_at"] = created_at
+        if created_at and (not last_seen or created_at > last_seen):
+            existing["last_seen_at"] = created_at
+
+    grouped_items = list(groups.values())
+    grouped_items.sort(
+        key=lambda item: (
+            str(item.get("last_seen_at") or item.get("created_at") or ""),
+            int(item.get("id") or 0),
+        ),
+        reverse=True,
+    )
+    return {
+        "incidents": grouped_items,
+        "raw_total": raw_total,
+        "total": len(grouped_items),
+    }
+
+
+def list_ops_payment_incidents(
+    request: Request,
+    limit: int = 50,
+    reason: str = "",
+    include_resolved: bool = False,
+) -> Dict[str, Any]:
+    _require_ops(request)
+    db = DBManager()
+    safe_limit = max(1, min(int(limit or 50), 200))
+    incidents = db.list_payment_audit_events(
+        limit=max(safe_limit, 500),
+        event_type="payment_intent_failed",
+    )
+    grouped = _group_payment_incidents(
+        incidents,
+        reason=reason,
+        include_resolved=include_resolved,
+    )
+    return {
+        **grouped,
+        "incidents": grouped["incidents"][:safe_limit],
+    }


 def resolve_ops_payment_incident(request: Request, event_id: int) -> Dict[str, Any]:
    admin = _require_ops(request) or {}
    db = DBManager()
-    resolved = db.mark_payment_audit_event_resolved(
+    resolved_group = db.mark_related_payment_audit_events_resolved(
        event_id, str(admin.get("email") or "")
    )
-    if not resolved:
+    if not resolved_group:
        raise HTTPException(status_code=404, detail="payment_incident_not_found")
-    return {"ok": True, "incident": resolved}
+    return {
+        "ok": True,
+        "incident": resolved_group[0],
+        "resolved_count": len(resolved_group),
+        "resolved_event_ids": [int(item.get("id") or 0) for item in resolved_group],
+    }


 def list_ops_payments(
@@ -913,15 +986,12 @@ def get_ops_billing_risk(
                )
            )

-    payment_incidents = db.list_payment_audit_events(
-        limit=safe_limit,
+    payment_incidents_raw = db.list_payment_audit_events(
+        limit=max(safe_limit, 500),
        event_type="payment_intent_failed",
    )
-    unresolved_incidents = [
-        item
-        for item in payment_incidents
-        if not str((item.get("payload") or {}).get("resolved_at") or "").strip()
-    ]
+    grouped_payment_incidents = _group_payment_incidents(payment_incidents_raw)
+    unresolved_incidents = grouped_payment_incidents["incidents"]

    issues.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True)
    recent_rewards = [
@@ -947,7 +1017,8 @@ def get_ops_billing_risk(
            "issues": len(issues),
            "stuck_intents": len(stuck_intents),
            "trial_gaps": len(trial_gaps),
-            "payment_incidents": len(unresolved_incidents),
+            "payment_incidents": grouped_payment_incidents["total"],
+            "payment_incident_events": grouped_payment_incidents["raw_total"],
            "points_discount_issues": len(points_issues),
            "referral_settlement_issues": len(referral_settlement_issues),
            "monthly_cap_hits": len(monthly_cap_hits),
@@ -47,11 +47,19 @@ def get_payment_runtime(request: Request) -> Dict[str, Any]:
    legacy_routes._assert_entitlement(request)
    try:
        db = DBManager()
+        checkout = legacy_routes.PAYMENT_CHECKOUT.get_config_payload()
+        rpc = legacy_routes.PAYMENT_CHECKOUT.get_rpc_runtime_status()
+        event_loop_state = db.get_payment_runtime_state("payment_event_loop") or {}
+        recent_audit_events = db.list_payment_audit_events(limit=20)
        return {
-            "checkout": legacy_routes.PAYMENT_CHECKOUT.get_config_payload(),
-            "rpc": legacy_routes.PAYMENT_CHECKOUT.get_rpc_runtime_status(),
-            "event_loop_state": db.get_payment_runtime_state("payment_event_loop") or {},
-            "recent_audit_events": db.list_payment_audit_events(limit=20),
+            "checkout": checkout,
+            "rpc": rpc,
+            "event_loop_state": event_loop_state,
+            "recent_audit_events": recent_audit_events,
+            "chain_id": checkout.get("chain_id") or rpc.get("chain_id"),
+            "receiver_contract": checkout.get("receiver_contract"),
+            "last_scanned_block": event_loop_state.get("last_scanned_block"),
+            "audit_events_count": len(recent_audit_events),
        }
    except legacy_routes.PaymentCheckoutError as exc:
        _raise_payment_error(exc)