feat: Implement payment processing with contract auditing, event loops, and web observability endpoints.

src/database/db_manager.py

@@ -3,7 +3,7 @@ import os
 import hashlib
 import json
 from datetime import datetime, timedelta
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, List
 from loguru import logger
 
 
@@ -83,9 +83,27 @@ class DBManager:
                     pro_granted INTEGER DEFAULT 0,
                     pro_error TEXT,
                     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
-                    PRIMARY KEY (week_key, telegram_id)
+                PRIMARY KEY (week_key, telegram_id)
                 )
             """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS payment_runtime_state (
+                    state_key TEXT PRIMARY KEY,
+                    payload_json TEXT NOT NULL,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS payment_audit_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    event_type TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_payment_audit_events_created_at ON payment_audit_events(created_at DESC)"
+            )
             conn.execute("""
                 CREATE TABLE IF NOT EXISTS supabase_bindings (
                     supabase_user_id TEXT PRIMARY KEY,
@@ -123,6 +141,91 @@ class DBManager:
             conn.commit()
             logger.info(f"Database initialized successfully path={self.db_path}")
 
+    def get_payment_runtime_state(self, state_key: str) -> Optional[Dict[str, Any]]:
+        key = str(state_key or "").strip()
+        if not key:
+            return None
+        with self._get_connection() as conn:
+            conn.row_factory = sqlite3.Row
+            row = conn.execute(
+                """
+                SELECT payload_json
+                FROM payment_runtime_state
+                WHERE state_key = ?
+                LIMIT 1
+                """,
+                (key,),
+            ).fetchone()
+            if not row:
+                return None
+            try:
+                payload = json.loads(str(row["payload_json"] or "{}"))
+            except Exception:
+                return None
+            return payload if isinstance(payload, dict) else None
+
+    def set_payment_runtime_state(self, state_key: str, payload: Dict[str, Any]) -> None:
+        key = str(state_key or "").strip()
+        if not key:
+            return
+        body = payload if isinstance(payload, dict) else {}
+        with self._get_connection() as conn:
+            conn.execute(
+                """
+                INSERT INTO payment_runtime_state (state_key, payload_json, updated_at)
+                VALUES (?, ?, ?)
+                ON CONFLICT(state_key) DO UPDATE SET
+                    payload_json = excluded.payload_json,
+                    updated_at = excluded.updated_at
+                """,
+                (key, json.dumps(body, ensure_ascii=False), datetime.now().isoformat()),
+            )
+            conn.commit()
+
+    def append_payment_audit_event(self, event_type: str, payload: Dict[str, Any]) -> None:
+        kind = str(event_type or "").strip().lower()
+        if not kind:
+            return
+        body = payload if isinstance(payload, dict) else {}
+        with self._get_connection() as conn:
+            conn.execute(
+                """
+                INSERT INTO payment_audit_events (event_type, payload_json, created_at)
+                VALUES (?, ?, ?)
+                """,
+                (kind, json.dumps(body, ensure_ascii=False), datetime.now().isoformat()),
+            )
+            conn.commit()
+
+    def list_payment_audit_events(self, limit: int = 50) -> List[Dict[str, Any]]:
+        safe_limit = max(1, min(int(limit or 50), 500))
+        with self._get_connection() as conn:
+            conn.row_factory = sqlite3.Row
+            rows = conn.execute(
+                """
+                SELECT id, event_type, payload_json, created_at
+                FROM payment_audit_events
+                ORDER BY id DESC
+                LIMIT ?
+                """,
+                (safe_limit,),
+            ).fetchall()
+            out = []
+            for row in rows:
+                try:
+                    payload = json.loads(str(row["payload_json"] or "{}"))
+                except Exception:
+                    payload = {}
+                out.append(
+                    {
+                        "id": int(row["id"]),
+                        "event_type": str(row["event_type"] or ""),
+                        "payload": payload if isinstance(payload, dict) else {},
+                        "created_at": row["created_at"],
+                    }
+                )
+            return out
+
     @staticmethod
     def _safe_week_key(value: str) -> str:
         text = str(value or "").strip()

src/payments/confirm_loop.py

@@ -7,8 +7,11 @@ from typing import Any, Dict
 
 from loguru import logger
 
+from src.database.db_manager import DBManager
 from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError
 
+_DB = DBManager()
+
 
 def _env_bool(name: str, default: bool) -> bool:
     raw = os.getenv(name)
@@ -46,6 +49,13 @@ def _short_hash(tx_hash: str) -> str:
     return f"{text[:10]}...{text[-6:]}"
 
 
+def _append_audit_event(event_type: str, payload: Dict[str, Any]) -> None:
+    try:
+        _DB.append_payment_audit_event(event_type, payload)
+    except Exception as exc:
+        logger.debug(f"payment confirm audit append failed: {exc}")
+
+
 def _runner() -> None:
     enabled = _env_bool("POLYWEATHER_PAYMENT_CONFIRM_LOOP_ENABLED", True)
     if not enabled:
@@ -67,6 +77,15 @@ def _runner() -> None:
         PAYMENT_CHECKOUT.chain_id,
         PAYMENT_CHECKOUT.confirmations,
     )
+    _append_audit_event(
+        "confirm_loop_started",
+        {
+            "interval_sec": interval_sec,
+            "batch_size": batch_size,
+            "chain_id": PAYMENT_CHECKOUT.chain_id,
+            "confirmations": PAYMENT_CHECKOUT.confirmations,
+        },
+    )
 
     while True:
         try:
@@ -114,6 +133,13 @@ def _runner() -> None:
                     )
 
             if scanned and (confirmed or already_confirmed or failed):
+                cycle_summary = {
+                    "scanned": scanned,
+                    "confirmed": confirmed,
+                    "already_confirmed": already_confirmed,
+                    "pending": pending,
+                    "failed": failed,
+                }
                 logger.info(
                     "payment confirm cycle scanned={} confirmed={} already={} pending={} failed={}",
                     scanned,
@@ -122,8 +148,10 @@ def _runner() -> None:
                     pending,
                     failed,
                 )
+                _append_audit_event("confirm_loop_cycle", cycle_summary)
         except Exception as exc:
             logger.warning(f"payment confirm cycle failed: {exc}")
+            _append_audit_event("confirm_loop_error", {"error": str(exc)})
         time.sleep(interval_sec)
 
 
@@ -135,4 +163,3 @@ def start_payment_confirm_loop():
     )
     thread.start()
     return thread
-

src/payments/contract_audit.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import os
+import re
+from typing import Any, Dict, List
+
+
+def _has(pattern: str, text: str) -> bool:
+    return re.search(pattern, text, re.MULTILINE | re.DOTALL) is not None
+
+
+def analyze_checkout_contract(source_path: str) -> Dict[str, Any]:
+    path = os.path.abspath(source_path)
+    with open(path, "r", encoding="utf-8") as fh:
+        source = fh.read()
+
+    checks = {
+        "has_only_owner_modifier": _has(r"modifier\s+onlyOwner\s*\(", source),
+        "owner_set_in_constructor": _has(r"owner\s*=\s*msg\.sender\s*;", source),
+        "set_treasury_only_owner": _has(
+            r"function\s+setTreasury\s*\([^)]*\)\s*external\s+onlyOwner", source
+        ),
+        "set_token_allowed_only_owner": _has(
+            r"function\s+setTokenAllowed\s*\([^)]*\)\s*external\s+onlyOwner", source
+        ),
+        "zero_address_guard_in_constructor": _has(
+            r"constructor\s*\([^)]*\)\s*\{\s*require\(\s*_token\s*!=\s*address\(0\)\s*&&\s*_treasury\s*!=\s*address\(0\)",
+            source,
+        ),
+        "zero_address_guard_in_setters": _has(
+            r"function\s+setTreasury[\s\S]*?require\(\s*_treasury\s*!=\s*address\(0\)",
+            source,
+        )
+        and _has(
+            r"function\s+setTokenAllowed[\s\S]*?require\(\s*token\s*!=\s*address\(0\)",
+            source,
+        ),
+        "allowed_token_check": _has(r"require\(\s*allowedToken\[token\]", source),
+        "amount_non_zero_check": _has(r"require\(\s*amount\s*>\s*0", source),
+        "duplicate_order_check": _has(r"require\(\s*!paidOrder\[orderId\]", source),
+        "paid_order_written_before_transfer": _has(
+            r"paidOrder\[orderId\]\s*=\s*true\s*;\s*require\(IERC20\(token\)\.transferFrom",
+            source,
+        ),
+        "emits_order_paid": _has(r"emit\s+OrderPaid\s*\(", source),
+        "uses_safe_erc20": _has(r"SafeERC20", source),
+        "has_pause_switch": _has(r"\bpaused\b|\bPausable\b|\bwhenNotPaused\b", source),
+        "has_rescue_function": _has(
+            r"function\s+(rescue|sweep|withdraw|recover)", source
+        ),
+        "binds_plan_amount_onchain": _has(
+            r"mapping\s*\(\s*uint256\s*=>[\s\S]*plan|planAmount|require\(\s*amount\s*==",
+            source,
+        ),
+    }
+
+    strengths: List[str] = []
+    risks: List[Dict[str, Any]] = []
+
+    if checks["has_only_owner_modifier"] and checks["set_treasury_only_owner"]:
+        strengths.append("关键管理函数受 onlyOwner 保护。")
+    if checks["allowed_token_check"]:
+        strengths.append("支付代币有 allowlist，避免任意 token 进入收款流程。")
+    if checks["duplicate_order_check"] and checks["paid_order_written_before_transfer"]:
+        strengths.append("订单去重状态在外部 transferFrom 前写入，能拦住同订单重复支付与典型重入重放。")
+    if checks["emits_order_paid"]:
+        strengths.append("链上事件 OrderPaid 明确，可作为链下审计与补单的唯一确认源。")
+
+    if checks["uses_safe_erc20"]:
+        strengths.append("使用了 SafeERC20 包装，兼容性更稳。")
+    else:
+        risks.append(
+            {
+                "id": "erc20_transfer_assumption",
+                "severity": "medium",
+                "title": "依赖 IERC20.transferFrom 直接返回 bool",
+                "detail": "当前合约直接调用 IERC20.transferFrom。对非标准 ERC20 的兼容性弱于 SafeERC20，建议如未来升级合约时改为 OpenZeppelin SafeERC20。",
+            }
+        )
+
+    if not checks["has_pause_switch"]:
+        risks.append(
+            {
+                "id": "no_pause_switch",
+                "severity": "medium",
+                "title": "缺少紧急暂停开关",
+                "detail": "一旦发现代币配置错误、接收地址异常或链上风险，当前合约无法直接暂停 pay。建议升级版合约加入 Pausable。",
+            }
+        )
+
+    if not checks["binds_plan_amount_onchain"]:
+        risks.append(
+            {
+                "id": "offchain_price_enforcement",
+                "severity": "medium",
+                "title": "套餐金额与 planId 绑定主要靠链下校验",
+                "detail": "合约事件只记录 planId 与 amount，本身不校验 planId 对应价格。当前依赖后端 intent/confirm 流程校验，后续升级可考虑链上 plan 配置或签名校验。",
+            }
+        )
+
+    if not checks["has_rescue_function"]:
+        risks.append(
+            {
+                "id": "no_rescue_function",
+                "severity": "low",
+                "title": "缺少误转资产救援函数",
+                "detail": "当前合约把资金直接转 treasury，不太容易残留余额，但若未来支持更多资产或误转到合约地址，缺少救援路径。",
+            }
+        )
+
+    risks.append(
+        {
+            "id": "single_owner_admin",
+            "severity": "medium",
+            "title": "owner 为单地址管理模型",
+            "detail": "setTreasury 和 setTokenAllowed 由单一 owner 控制。生产建议用多签地址持有 owner，降低单点密钥失窃风险。",
+        }
+    )
+
+    runtime_controls = [
+        "后端只认链上 OrderPaid 事件，不认前端自报支付成功。",
+        "payment event loop 与 confirm loop 已写入 SQLite 审计事件，可做对账与回放。",
+        "支持 POLYWEATHER_PAYMENT_RPC_URLS 多 RPC 容灾，单节点故障时可轮换。",
+    ]
+
+    recommendations = [
+        "生产 owner 建议迁移到多签钱包。",
+        "下一版合约优先补 SafeERC20 与 Pausable。",
+        "若要进一步收紧授权边界，可把 planId/amount/token 绑定做进链上或 EIP-712 签名校验。",
+        "每次合约地址或 allowed token 变更后，都运行静态检查与链上回放脚本。",
+    ]
+
+    return {
+        "contract_path": path,
+        "contract_name": "PolyWeatherCheckout",
+        "summary": {
+            "strength_count": len(strengths),
+            "risk_count": len(risks),
+            "highest_severity": "medium" if risks else "none",
+        },
+        "checks": checks,
+        "strengths": strengths,
+        "runtime_controls": runtime_controls,
+        "risks": risks,
+        "recommendations": recommendations,
+    }
+

src/payments/contract_checkout.py

@@ -247,6 +247,9 @@ class PaymentContractCheckoutService:
         self.chain_id = _env_int("POLYWEATHER_PAYMENT_CHAIN_ID", DEFAULT_POLYGON_CHAIN_ID)
         self.token_decimals = _env_int("POLYWEATHER_PAYMENT_TOKEN_DECIMALS", 6)
         self.rpc_url = str(os.getenv("POLYWEATHER_PAYMENT_RPC_URL") or "").strip()
+        self.rpc_urls = self._load_rpc_urls(
+            os.getenv("POLYWEATHER_PAYMENT_RPC_URLS") or self.rpc_url
+        )
         legacy_receiver_contract = _normalize_address(
             os.getenv("POLYWEATHER_PAYMENT_RECEIVER_CONTRACT") or ""
         )
@@ -315,6 +318,7 @@ class PaymentContractCheckoutService:
         )
         self._w3_lock = threading.Lock()
         self._w3: Optional[Web3] = None
+        self._w3_url: str = ""
         self._event_topic = Web3.keccak(
             text="OrderPaid(bytes32,address,uint256,address,uint256)"
         ).hex()
@@ -332,7 +336,7 @@ class PaymentContractCheckoutService:
         return bool(
             self.supabase_url
             and self.supabase_service_role_key
-            and self.rpc_url
+            and bool(self.rpc_urls)
             and has_valid_token_routes
         )
 
@@ -348,6 +352,14 @@ class PaymentContractCheckoutService:
                 ),
             )
 
+    def _load_rpc_urls(self, raw: str) -> List[str]:
+        out: List[str] = []
+        for part in str(raw or "").split(","):
+            url = str(part or "").strip()
+            if url and url not in out:
+                out.append(url)
+        return out
+
     def _default_token_meta(self, address: str) -> Dict[str, str]:
         normalized = _normalize_address(address)
         if normalized == _normalize_address(DEFAULT_NATIVE_USDC_ADDRESS):
@@ -821,15 +833,48 @@ class PaymentContractCheckoutService:
         result["discount_usdc"] = _format_decimal(discount_usdc)
         return result
 
-    def _get_web3(self) -> Web3:
+    def _build_web3(self, rpc_url: str) -> Web3:
+        return Web3(
+            Web3.HTTPProvider(rpc_url, request_kwargs={"timeout": self.timeout_sec})
+        )
+
+    def _try_connect_rpc(self, rpc_url: str) -> Optional[Web3]:
+        try:
+            w3 = self._build_web3(rpc_url)
+            if not w3.is_connected():
+                return None
+            if int(w3.eth.chain_id) != int(self.chain_id):
+                return None
+            return w3
+        except Exception:
+            return None
+
+    def _rotate_rpc(self) -> Optional[Web3]:
+        for rpc_url in self.rpc_urls:
+            w3 = self._try_connect_rpc(rpc_url)
+            if w3 is not None:
+                self._w3 = w3
+                self._w3_url = rpc_url
+                return w3
+        self._w3 = None
+        self._w3_url = ""
+        return None
+
+    def _get_web3(self, force_refresh: bool = False) -> Web3:
         with self._w3_lock:
-            if self._w3 is None:
-                self._w3 = Web3(
-                    Web3.HTTPProvider(self.rpc_url, request_kwargs={"timeout": self.timeout_sec})
-                )
+            if self._w3 is None or force_refresh:
+                self._rotate_rpc()
         assert self._w3 is not None
         return self._w3
 
+    def get_rpc_runtime_status(self) -> Dict[str, Any]:
+        candidates = list(self.rpc_urls)
+        return {
+            "configured_rpc_count": len(candidates),
+            "active_rpc_url": self._w3_url or (candidates[0] if candidates else ""),
+            "all_rpc_urls": candidates,
+        }
+
     def _get_contract(self, receiver_address: Optional[str] = None):
         w3 = self._get_web3()
         contract_address = _normalize_address(receiver_address or self.receiver_contract)
@@ -1561,15 +1606,25 @@ class PaymentContractCheckoutService:
     def _wait_receipt(self, tx_hash: str) -> Any:
         import time as _time
 
-        w3 = self._get_web3()
         start = _now_utc()
         while (_now_utc() - start).total_seconds() < self.max_wait_sec:
             try:
+                w3 = self._get_web3()
                 receipt = w3.eth.get_transaction_receipt(tx_hash)
             except Exception:
-                receipt = None
+                try:
+                    w3 = self._get_web3(force_refresh=True)
+                    receipt = w3.eth.get_transaction_receipt(tx_hash)
+                except Exception:
+                    receipt = None
             if receipt and receipt.get("blockNumber"):
                 return receipt
+            try:
+                latest_w3 = self._get_web3()
+                if not latest_w3.is_connected():
+                    self._get_web3(force_refresh=True)
+            except Exception:
+                receipt = None
             _time.sleep(self.poll_interval_sec)
         raise PaymentCheckoutError(408, "tx receipt timeout")
 

src/payments/event_loop.py

@@ -10,8 +10,11 @@ from typing import Any, Dict, List, Optional
 from loguru import logger
 from web3 import Web3
 
+from src.database.db_manager import DBManager
 from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError
 
+_DB = DBManager()
+
 
 def _env_bool(name: str, default: bool) -> bool:
     raw = os.getenv(name)
@@ -70,6 +73,9 @@ def _state_file() -> str:
 
 
 def _load_state(path: str) -> Dict[str, Any]:
+    db_state = _DB.get_payment_runtime_state("payment_event_loop")
+    if isinstance(db_state, dict) and db_state:
+        return db_state
     if not os.path.exists(path):
         return {}
     try:
@@ -82,6 +88,7 @@ def _load_state(path: str) -> Dict[str, Any]:
 
 
 def _save_state(path: str, state: Dict[str, Any]) -> None:
+    _DB.set_payment_runtime_state("payment_event_loop", state)
     os.makedirs(os.path.dirname(path), exist_ok=True)
     tmp_path = f"{path}.tmp"
     with open(tmp_path, "w", encoding="utf-8") as fh:
@@ -89,6 +96,13 @@ def _save_state(path: str, state: Dict[str, Any]) -> None:
     os.replace(tmp_path, path)
 
 
+def _append_audit_event(event_type: str, payload: Dict[str, Any]) -> None:
+    try:
+        _DB.append_payment_audit_event(event_type, payload)
+    except Exception as exc:
+        logger.debug(f"payment event audit append failed: {exc}")
+
+
 def _is_pending_confirm_error(exc: PaymentCheckoutError) -> bool:
     detail = str(exc.detail or "").lower()
     if exc.status_code in {404, 408, 502, 503}:
@@ -233,6 +247,17 @@ def _runner() -> None:
         len(receiver_contracts),
         PAYMENT_CHECKOUT.chain_id,
     )
+    _append_audit_event(
+        "event_loop_started",
+        {
+            "interval_sec": interval_sec,
+            "lookback_blocks": lookback_blocks,
+            "step_blocks": step_blocks,
+            "max_events": max_events,
+            "receiver_contracts": receiver_contracts,
+            "chain_id": PAYMENT_CHECKOUT.chain_id,
+        },
+    )
 
     while True:
         cycle_started = time.time()
@@ -392,6 +417,18 @@ def _runner() -> None:
                 cursor = to_block + 1
 
             if scanned_blocks > 0:
+                cycle_summary = {
+                    "blocks": scanned_blocks,
+                    "events": scanned_events,
+                    "matched": matched_intents,
+                    "submitted": submitted,
+                    "confirmed": confirmed,
+                    "already": already,
+                    "pending": pending,
+                    "failed": failed,
+                    "ignored": ignored,
+                    "last_scanned_block": int(state.get("last_scanned_block") or 0),
+                }
                 logger.info(
                     "payment event cycle blocks={} events={} matched={} submitted={} "
                     "confirmed={} already={} pending={} failed={} ignored={}",
@@ -405,8 +442,10 @@ def _runner() -> None:
                     failed,
                     ignored,
                 )
+                _append_audit_event("event_loop_cycle", cycle_summary)
         except Exception as exc:
             logger.warning(f"payment event cycle failed: {exc}")
+            _append_audit_event("event_loop_error", {"error": str(exc)})
 
         elapsed = time.time() - cycle_started
         time.sleep(max(0.0, interval_sec - elapsed))