From 31fb1b3f289284fa81c9f8437419be534f1f8b04 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 1 Jun 2026 13:49:47 +0800 Subject: [PATCH] Fix ops payment incident grouping --- .../ops/__tests__/opsBillingRisk.test.ts | 4 +- .../ops/payments/PaymentsPageClient.tsx | 9 +- frontend/types/ops.ts | 9 +- src/database/db_manager.py | 120 +++++++++++++++++- tests/test_payments_runtime.py | 46 +++++++ tests/test_web_observability.py | 119 +++++++++++++++++ web/services/ops_api.py | 119 +++++++++++++---- web/services/payment_api.py | 16 ++- 8 files changed, 410 insertions(+), 32 deletions(-) diff --git a/frontend/components/ops/__tests__/opsBillingRisk.test.ts b/frontend/components/ops/__tests__/opsBillingRisk.test.ts index 2f68e09d..48e411d1 100644 --- a/frontend/components/ops/__tests__/opsBillingRisk.test.ts +++ b/frontend/components/ops/__tests__/opsBillingRisk.test.ts @@ -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( diff --git a/frontend/components/ops/payments/PaymentsPageClient.tsx b/frontend/components/ops/payments/PaymentsPageClient.tsx index 8e82f516..f6cd0c1b 100644 --- a/frontend/components/ops/payments/PaymentsPageClient.tsx +++ b/frontend/components/ops/payments/PaymentsPageClient.tsx @@ -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 = {}; 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() { {inc.id}
{paymentReasonLabel(inc.reason)}
+ {Number(inc.occurrence_count ?? 1) > 1 ? ( +
+ 同类重复 {Number(inc.occurrence_count).toLocaleString()} 次 · 最早 {compactDate(inc.first_seen_at)} +
+ ) : null}
{inc.detail || inc.reason || "—"}
diff --git a/frontend/types/ops.ts b/frontend/types/ops.ts index 55381dc1..37bc6696 100644 --- a/frontend/types/ops.ts +++ b/frontend/types/ops.ts @@ -85,11 +85,14 @@ export type SourceHealthPayload = { }; export type PaymentRuntimePayload = { - rpc?: string; + rpc?: Record | string; chain_id?: number; receiver_contract?: string; last_scanned_block?: number; audit_events_count?: number; + checkout?: Record; + event_loop_state?: Record; + recent_audit_events?: Array>; recent_events?: Array>; }; @@ -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 = { diff --git a/src/database/db_manager.py b/src/database/db_manager.py index b3b93e66..9d5e58f1 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -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() diff --git a/tests/test_payments_runtime.py b/tests/test_payments_runtime.py index f6369d0c..5597fd32 100644 --- a/tests/test_payments_runtime.py +++ b/tests/test_payments_runtime.py @@ -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() diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 943de5b7..f91ba962 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -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, diff --git a/web/services/ops_api.py b/web/services/ops_api.py index d77ee4b2..627b7359 100644 --- a/web/services/ops_api.py +++ b/web/services/ops_api.py @@ -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), diff --git a/web/services/payment_api.py b/web/services/payment_api.py index a595969c..52c50fde 100644 --- a/web/services/payment_api.py +++ b/web/services/payment_api.py @@ -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)