From 02e35a3a5f0928bf32dc9c8ee9be9bc493dece80 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sat, 14 Mar 2026 10:35:30 +0800 Subject: [PATCH] feat: Implement a new contract-based payment system with wallet binding, payment intents, and subscription plan management. --- .env.example | 4 + .vscode/settings.json | 5 + .../api/payments/intents/[intentId]/route.ts | 47 ++ frontend/components/account/AccountCenter.tsx | 66 ++- src/bot/runtime_coordinator.py | 49 ++- .../polymarket_wallet_activity_watcher.py | 11 +- src/payments/contract_checkout.py | 75 ++++ src/payments/event_loop.py | 416 ++++++++++++++++++ src/utils/telegram_chat_ids.py | 20 + web/app.py | 14 + 10 files changed, 698 insertions(+), 9 deletions(-) create mode 100644 .vscode/settings.json create mode 100644 frontend/app/api/payments/intents/[intentId]/route.ts create mode 100644 src/payments/event_loop.py diff --git a/.env.example b/.env.example index 431438a8..7d11cfdc 100644 --- a/.env.example +++ b/.env.example @@ -141,6 +141,10 @@ POLYGON_WALLET_WATCH_POLYMARKET_CONTRACTS= # Polymarket Wallet Activity Watcher (all markets, not weather-only) POLYMARKET_WALLET_ACTIVITY_ENABLED=false POLYMARKET_WALLET_ACTIVITY_USERS=0x0000000000000000000000000000000000000000 +# Optional dedicated chat targets for wallet activity push (recommended) +# If unset, fallback to TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID. +POLYMARKET_WALLET_ACTIVITY_CHAT_ID= +POLYMARKET_WALLET_ACTIVITY_CHAT_IDS= # Optional wallet nicknames: # - CSV: 0xabc...=Whale_A,0xdef...=Main_Account # - JSON: {"0xabc...":"Whale A","0xdef...":"Main Account"} diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 00000000..80abd191 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,5 @@ +{ + "css.validate": false, + "scss.validate": false, + "less.validate": false +} diff --git a/frontend/app/api/payments/intents/[intentId]/route.ts b/frontend/app/api/payments/intents/[intentId]/route.ts new file mode 100644 index 00000000..6d9ddcac --- /dev/null +++ b/frontend/app/api/payments/intents/[intentId]/route.ts @@ -0,0 +1,47 @@ +import { NextRequest, NextResponse } from "next/server"; +import { + applyAuthResponseCookies, + buildBackendRequestHeaders, +} from "@/lib/backend-auth"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + +export async function GET( + req: NextRequest, + context: { params: Promise<{ intentId: string }> }, +) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { status: 500 }, + ); + } + const { intentId } = await context.params; + try { + const auth = await buildBackendRequestHeaders(req); + const res = await fetch( + `${API_BASE}/api/payments/intents/${encodeURIComponent(intentId)}`, + { + method: "GET", + headers: auth.headers, + cache: "no-store", + }, + ); + if (!res.ok) { + const raw = await res.text(); + const response = NextResponse.json( + { error: `Backend returned ${res.status}`, detail: raw.slice(0, 350) }, + { status: res.status }, + ); + return applyAuthResponseCookies(response, auth.response); + } + const data = await res.json(); + const response = NextResponse.json(data); + return applyAuthResponseCookies(response, auth.response); + } catch (error) { + return NextResponse.json( + { error: "Failed to fetch payment intent", detail: String(error) }, + { status: 500 }, + ); + } +} diff --git a/frontend/components/account/AccountCenter.tsx b/frontend/components/account/AccountCenter.tsx index eee36a1e..de1f9095 100644 --- a/frontend/components/account/AccountCenter.tsx +++ b/frontend/components/account/AccountCenter.tsx @@ -125,6 +125,14 @@ type CreatedIntent = { }; }; +type IntentStatusResponse = { + intent?: { + intent_id?: string; + status?: string; + tx_hash?: string | null; + }; +}; + declare global { interface Window { ethereum?: EvmProvider; @@ -398,7 +406,7 @@ function normalizePaymentError(error: unknown): NormalizedPaymentError { ?.trim(); const lower = String(rawMessage || "").toLowerCase(); - if (lower.includes("confirm pending")) { + if (lower.includes("confirm pending") || lower.includes("payment pending timeout")) { return { message: "链上交易已提交，正在确认中，请稍后刷新查看状态。", pending: true, @@ -940,6 +948,55 @@ export function AccountCenter() { throw new Error(`transaction confirmation timeout: ${txHash}`); }; + const pollIntentUntilConfirmed = useCallback( + async ( + intentId: string, + authHeaders: Record, + txHashHint = "", + timeoutMs = 180000, + pollMs = 5000, + ) => { + const startedAt = Date.now(); + const shortTx = shortAddress(txHashHint); + while (Date.now() - startedAt < timeoutMs) { + const statusRes = await fetch(`/api/payments/intents/${intentId}`, { + method: "GET", + headers: authHeaders, + cache: "no-store", + }); + if (!statusRes.ok) { + if (statusRes.status >= 500 || statusRes.status === 429) { + await new Promise((resolve) => setTimeout(resolve, pollMs)); + continue; + } + const raw = (await statusRes.text()).slice(0, 260); + throw new Error(`query intent failed: ${raw}`); + } + + const statusJson = (await statusRes.json()) as IntentStatusResponse; + const intent = statusJson.intent || {}; + const status = String(intent.status || "").toLowerCase(); + const txHash = String(intent.tx_hash || txHashHint || "").toLowerCase(); + if (status === "confirmed") { + setPaymentError(""); + setPaymentInfo(`支付确认成功，交易: ${shortAddress(txHash)}`); + await loadSnapshot(); + await loadPaymentSnapshot(); + return; + } + if (status === "failed" || status === "cancelled" || status === "expired") { + throw new Error(`payment ${status}`); + } + setPaymentInfo( + `交易已提交: ${shortTx}，正在链上确认（状态: ${status || "submitted"}）...`, + ); + await new Promise((resolve) => setTimeout(resolve, pollMs)); + } + throw new Error("payment pending timeout"); + }, + [loadPaymentSnapshot, loadSnapshot], + ); + const signBindMessage = async ( eth: EvmProvider, address: string, @@ -1296,8 +1353,11 @@ export function AccountCenter() { (lowerRaw.includes("confirmations not enough") || lowerRaw.includes("tx indexed partially"))); if (maybePending) { - setPaymentInfo(`交易已提交: ${shortAddress(txHashNorm)}，等待确认中。`); - throw new Error(`confirm pending: ${raw}`); + setPaymentInfo( + `交易已提交: ${shortAddress(txHashNorm)}，等待链上确认中...`, + ); + await pollIntentUntilConfirmed(intentId, authHeaders, txHashNorm); + return; } throw new Error(`confirm failed: ${raw}`); } diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index ebe52ee6..2ca76ee3 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -6,7 +6,10 @@ from datetime import datetime from importlib import import_module from typing import Any, Callable, Dict, List, Optional -from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env +from src.utils.telegram_chat_ids import ( + get_polymarket_wallet_activity_chat_ids_from_env, + get_telegram_chat_ids_from_env, +) def _env_bool(name: str, default: bool) -> bool: @@ -87,6 +90,7 @@ class StartupCoordinator: self._start_polygon_wallet_loop(), self._start_polymarket_wallet_activity_loop(), self._start_weekly_reward_loop(), + self._start_payment_event_loop(), self._start_payment_confirm_loop(), ] self._runtime_status = RuntimeStatus( @@ -210,7 +214,7 @@ class StartupCoordinator: def _start_polymarket_wallet_activity_loop(self) -> LoopStatus: enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False) - chat_ids = get_telegram_chat_ids_from_env() + chat_ids = get_polymarket_wallet_activity_chat_ids_from_env() users_count = _parse_csv_count(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS")) poll = max(5, _env_int("POLYMARKET_WALLET_ACTIVITY_INTERVAL_SEC", 20)) details = { @@ -221,7 +225,9 @@ class StartupCoordinator: } validation_error = None if not chat_ids: - validation_error = "missing_TELEGRAM_CHAT_IDS" + validation_error = ( + "missing_POLYMARKET_WALLET_ACTIVITY_CHAT_IDS_or_TELEGRAM_CHAT_IDS" + ) elif users_count == 0: validation_error = "missing_POLYMARKET_WALLET_ACTIVITY_USERS" return self._start_with_validation( @@ -300,6 +306,43 @@ class StartupCoordinator: ).start_payment_confirm_loop(), ) + def _start_payment_event_loop(self) -> LoopStatus: + enabled = _env_bool("POLYWEATHER_PAYMENT_EVENT_LOOP_ENABLED", True) + details = { + "interval_sec": max( + 5, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_INTERVAL_SEC", 20) + ), + "lookback_blocks": max( + 500, + _env_int( + "POLYWEATHER_PAYMENT_EVENT_LOOP_START_LOOKBACK_BLOCKS", + 5000, + ), + ), + "step_blocks": min( + 49999, + max(100, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_STEP_BLOCKS", 2000)), + ), + "max_events": max( + 10, _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_MAX_EVENTS_PER_CYCLE", 200) + ), + "payment_enabled": _env_bool("POLYWEATHER_PAYMENT_ENABLED", False), + "chain_id": _env_int("POLYWEATHER_PAYMENT_CHAIN_ID", 137), + } + validation_error = None + if not bool(details["payment_enabled"]): + validation_error = "payment_service_disabled" + return self._start_with_validation( + key="payment_event", + label="支付事件监听", + configured_enabled=enabled, + details=details, + validation_error=validation_error, + starter=lambda: import_module( + "src.payments.event_loop" + ).start_payment_event_loop(), + ) + def render_runtime_status_html(status: RuntimeStatus) -> str: lines = [ diff --git a/src/onchain/polymarket_wallet_activity_watcher.py b/src/onchain/polymarket_wallet_activity_watcher.py index a3243179..74b75feb 100644 --- a/src/onchain/polymarket_wallet_activity_watcher.py +++ b/src/onchain/polymarket_wallet_activity_watcher.py @@ -9,7 +9,9 @@ from typing import Any, Dict, List, Optional, Tuple import requests from loguru import logger -from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env +from src.utils.telegram_chat_ids import ( + get_polymarket_wallet_activity_chat_ids_from_env, +) def _env_bool(name: str, default: bool) -> bool: @@ -577,7 +579,7 @@ def _filter_changes_by_position_value( def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread]: enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False) - chat_ids = get_telegram_chat_ids_from_env() + chat_ids = get_polymarket_wallet_activity_chat_ids_from_env() users = _parse_addresses(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS")) user_aliases = _parse_address_aliases( os.getenv("POLYMARKET_WALLET_ACTIVITY_USER_ALIASES") @@ -591,7 +593,10 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread logger.info("polymarket wallet activity watcher disabled") return None if not chat_ids: - logger.warning("polymarket wallet activity watcher skipped: TELEGRAM_CHAT_IDS is not set") + logger.warning( + "polymarket wallet activity watcher skipped: " + "POLYMARKET_WALLET_ACTIVITY_CHAT_IDS/CHAT_ID and TELEGRAM_CHAT_IDS/CHAT_ID are empty" + ) return None if not users: logger.warning("polymarket wallet activity watcher skipped: POLYMARKET_WALLET_ACTIVITY_USERS is empty") diff --git a/src/payments/contract_checkout.py b/src/payments/contract_checkout.py index 59b3ab52..07ee19d2 100644 --- a/src/payments/contract_checkout.py +++ b/src/payments/contract_checkout.py @@ -103,6 +103,21 @@ def _normalize_address(address: Any) -> str: return Web3.to_checksum_address(text).lower() +def _normalize_order_id_hex(order_id_hex: Any) -> str: + text = str(order_id_hex or "").strip().lower() + if not text: + return "" + if not text.startswith("0x"): + text = f"0x{text}" + if len(text) != 66: + return "" + try: + int(text[2:], 16) + except Exception: + return "" + return text + + def _now_utc() -> datetime: return datetime.now(timezone.utc) @@ -1321,6 +1336,66 @@ class PaymentContractCheckoutService: ) return out + def list_open_intents_by_order_id( + self, + order_id_hex: str, + limit: int = 10, + ) -> List[Dict[str, Any]]: + """ + Find intents by on-chain order id for event-driven reconciliation. + Includes created/submitted intents; confirmed intents are returned too for idempotent skip. + """ + self._ensure_enabled() + normalized_order = _normalize_order_id_hex(order_id_hex) + if not normalized_order: + return [] + safe_limit = max(1, min(int(limit or 10), 50)) + rows = self._rest( + "GET", + "payment_intents", + params={ + "select": ( + "id,user_id,status,tx_hash,order_id_hex,plan_id,token_address," + "amount_units,payment_mode,allowed_wallet,chain_id,created_at,updated_at" + ), + "order_id_hex": f"eq.{normalized_order}", + "chain_id": f"eq.{self.chain_id}", + "status": "in.(created,submitted,confirmed)", + "order": "created_at.desc", + "limit": str(safe_limit), + }, + allowed_status=[200], + ) + if not isinstance(rows, list): + return [] + + out: List[Dict[str, Any]] = [] + for row in rows: + if not isinstance(row, dict): + continue + intent_id = str(row.get("id") or "").strip() + user_id = str(row.get("user_id") or "").strip() + status = str(row.get("status") or "").strip().lower() + if not intent_id or not user_id or not status: + continue + out.append( + { + "intent_id": intent_id, + "user_id": user_id, + "status": status, + "tx_hash": str(row.get("tx_hash") or "").strip().lower(), + "order_id_hex": str(row.get("order_id_hex") or "").strip().lower(), + "plan_id": int(row.get("plan_id") or 0), + "token_address": _normalize_address(row.get("token_address")), + "amount_units": int(row.get("amount_units") or 0), + "payment_mode": str(row.get("payment_mode") or "").strip().lower(), + "allowed_wallet": _normalize_address(row.get("allowed_wallet")), + "created_at": row.get("created_at"), + "updated_at": row.get("updated_at"), + } + ) + return out + def submit_intent_tx( self, user_id: str, diff --git a/src/payments/event_loop.py b/src/payments/event_loop.py new file mode 100644 index 00000000..5063c194 --- /dev/null +++ b/src/payments/event_loop.py @@ -0,0 +1,416 @@ +from __future__ import annotations + +import json +import os +import threading +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from loguru import logger +from web3 import Web3 + +from src.payments import PAYMENT_CHECKOUT, PaymentCheckoutError + + +def _env_bool(name: str, default: bool) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + +def _env_int(name: str, default: int, min_value: int = 0) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + value = int(raw) + except Exception: + return default + return max(min_value, value) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _normalize_address(address: Any) -> str: + text = str(address or "").strip() + if not text or not Web3.is_address(text): + return "" + return Web3.to_checksum_address(text).lower() + + +def _to_hex(value: Any) -> str: + try: + return str(Web3.to_hex(value or b"")).lower() + except Exception: + return "" + + +def _state_file() -> str: + custom = str( + os.getenv("POLYWEATHER_PAYMENT_EVENT_LOOP_STATE_PATH") or "" + ).strip() + if custom: + return custom + runtime_dir = str(os.getenv("POLYWEATHER_RUNTIME_DATA_DIR") or "").strip() + if runtime_dir: + return os.path.join(runtime_dir, "payment_event_loop_state.json") + root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + return os.path.join(root, "data", "payment_event_loop_state.json") + + +def _load_state(path: str) -> Dict[str, Any]: + if not os.path.exists(path): + return {} + try: + with open(path, "r", encoding="utf-8") as fh: + raw = json.load(fh) + return raw if isinstance(raw, dict) else {} + except Exception as exc: + logger.warning(f"payment event loop state load failed: {exc}") + return {} + + +def _save_state(path: str, state: Dict[str, Any]) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp_path = f"{path}.tmp" + with open(tmp_path, "w", encoding="utf-8") as fh: + json.dump(state, fh, ensure_ascii=False, indent=2) + os.replace(tmp_path, path) + + +def _is_pending_confirm_error(exc: PaymentCheckoutError) -> bool: + detail = str(exc.detail or "").lower() + if exc.status_code in {404, 408, 502, 503}: + return True + if exc.status_code == 409 and ( + "confirmations not enough" in detail or "tx indexed partially" in detail + ): + return True + return False + + +def _is_nonfatal_submit_error(exc: PaymentCheckoutError) -> bool: + detail = str(exc.detail or "").lower() + if exc.status_code in {502, 503, 408}: + return True + if exc.status_code == 409 and ( + "intent status is submitted" in detail + or "intent status is confirmed" in detail + or "cannot submit" in detail + ): + return True + return False + + +def _decode_order_paid_log(log_item: Any) -> Optional[Dict[str, Any]]: + log_get = getattr(log_item, "get", None) + if not callable(log_get): + return None + address = _normalize_address(log_get("address")) + if not address: + return None + try: + contract = PAYMENT_CHECKOUT._get_contract(address) # noqa: SLF001 + event_obj = contract.events.OrderPaid().process_log(log_item) + except Exception: + return None + + event_get = getattr(event_obj, "get", None) + args = event_get("args") if callable(event_get) else getattr(event_obj, "args", None) + if not args: + return None + args_get = getattr(args, "get", None) + if not callable(args_get): + return None + + order_id_hex = _to_hex(args_get("orderId")) + payer = _normalize_address(args_get("payer")) + token = _normalize_address(args_get("token")) + plan_id = int(args_get("planId") or 0) + amount_units = int(args_get("amount") or 0) + tx_hash = _to_hex(log_get("transactionHash")) + block_number = int(log_get("blockNumber") or 0) + log_index = int(log_get("logIndex") or 0) + + if not (order_id_hex and payer and token and tx_hash and plan_id > 0 and amount_units > 0): + return None + return { + "order_id_hex": order_id_hex, + "payer": payer, + "plan_id": plan_id, + "token_address": token, + "amount_units": amount_units, + "receiver_contract": address, + "tx_hash": tx_hash, + "block_number": block_number, + "log_index": log_index, + } + + +def _event_key(event_row: Dict[str, Any]) -> str: + return f"{event_row.get('tx_hash')}:{int(event_row.get('log_index') or 0)}" + + +def _select_matching_intents( + intents: List[Dict[str, Any]], + event_row: Dict[str, Any], +) -> List[Dict[str, Any]]: + out: List[Dict[str, Any]] = [] + for row in intents: + if not isinstance(row, dict): + continue + if int(row.get("plan_id") or 0) != int(event_row.get("plan_id") or 0): + continue + if _normalize_address(row.get("token_address")) != _normalize_address( + event_row.get("token_address") + ): + continue + if int(row.get("amount_units") or 0) != int(event_row.get("amount_units") or 0): + continue + out.append(row) + if out: + return out + return intents + + +def _runner() -> None: + enabled = _env_bool("POLYWEATHER_PAYMENT_EVENT_LOOP_ENABLED", True) + if not enabled: + logger.info("payment event loop disabled") + return + if not PAYMENT_CHECKOUT.enabled: + logger.info("payment event loop skipped: payment service disabled") + return + + interval_sec = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_INTERVAL_SEC", 20, 5) + lookback_blocks = _env_int( + "POLYWEATHER_PAYMENT_EVENT_LOOP_START_LOOKBACK_BLOCKS", 5000, 500 + ) + step_blocks = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_STEP_BLOCKS", 2000, 100) + step_blocks = min(step_blocks, 49999) + max_events = _env_int("POLYWEATHER_PAYMENT_EVENT_LOOP_MAX_EVENTS_PER_CYCLE", 200, 10) + state_path = _state_file() + + receiver_contracts = sorted( + { + _normalize_address(token.receiver_contract) + for token in PAYMENT_CHECKOUT.supported_tokens.values() + if _normalize_address(token.receiver_contract) + } + ) + if not receiver_contracts: + logger.warning("payment event loop skipped: no receiver contract configured") + return + + topic0 = str(PAYMENT_CHECKOUT._event_topic or "").strip().lower() # noqa: SLF001 + if topic0 and not topic0.startswith("0x"): + topic0 = f"0x{topic0}" + if not topic0: + topic0 = ( + "0x" + + Web3.keccak( + text="OrderPaid(bytes32,address,uint256,address,uint256)" + ).hex().lower().replace("0x", "") + ) + + logger.info( + "payment event loop started interval={}s lookback={} step={} max_events={} " + "contracts={} chain_id={}", + interval_sec, + lookback_blocks, + step_blocks, + max_events, + len(receiver_contracts), + PAYMENT_CHECKOUT.chain_id, + ) + + while True: + cycle_started = time.time() + try: + w3 = PAYMENT_CHECKOUT._get_web3() # noqa: SLF001 + if not w3.is_connected(): + logger.warning("payment event loop skipped: rpc not connected") + time.sleep(interval_sec) + continue + if int(w3.eth.chain_id) != int(PAYMENT_CHECKOUT.chain_id): + logger.warning( + "payment event loop skipped: chain mismatch rpc={} expected={}", + int(w3.eth.chain_id), + int(PAYMENT_CHECKOUT.chain_id), + ) + time.sleep(interval_sec) + continue + + latest_block = int(w3.eth.block_number) + safe_latest = latest_block - max(0, int(PAYMENT_CHECKOUT.confirmations) - 1) + if safe_latest <= 0: + time.sleep(interval_sec) + continue + + state = _load_state(state_path) + last_scanned = int(state.get("last_scanned_block") or 0) + start_block = ( + last_scanned + 1 + if last_scanned > 0 + else max(0, safe_latest - lookback_blocks + 1) + ) + if start_block > safe_latest: + time.sleep(interval_sec) + continue + + scanned_blocks = 0 + scanned_events = 0 + matched_intents = 0 + submitted = 0 + confirmed = 0 + already = 0 + pending = 0 + failed = 0 + ignored = 0 + seen_events: set[str] = set() + cursor = start_block + + while cursor <= safe_latest and scanned_events < max_events: + to_block = min(cursor + step_blocks - 1, safe_latest) + params: Dict[str, Any] = { + "fromBlock": cursor, + "toBlock": to_block, + "topics": [topic0], + } + params["address"] = ( + receiver_contracts + if len(receiver_contracts) > 1 + else receiver_contracts[0] + ) + + logs = w3.eth.get_logs(params) + scanned_blocks += max(0, to_block - cursor + 1) + state["last_scanned_block"] = to_block + state["updated_at"] = _now_iso() + _save_state(state_path, state) + + if logs: + for log_item in logs: + event_row = _decode_order_paid_log(log_item) + if not event_row: + continue + event_key = _event_key(event_row) + if event_key in seen_events: + continue + seen_events.add(event_key) + scanned_events += 1 + + intents = PAYMENT_CHECKOUT.list_open_intents_by_order_id( + event_row["order_id_hex"], + limit=10, + ) + if not intents: + ignored += 1 + if scanned_events >= max_events: + break + continue + + candidates = _select_matching_intents(intents, event_row) + for row in candidates: + status = str(row.get("status") or "").strip().lower() + if status == "confirmed": + already += 1 + continue + + user_id = str(row.get("user_id") or "").strip() + intent_id = str(row.get("intent_id") or "").strip() + if not user_id or not intent_id: + continue + matched_intents += 1 + + if status == "created" or not str(row.get("tx_hash") or "").strip(): + try: + PAYMENT_CHECKOUT.submit_intent_tx( + user_id=user_id, + intent_id=intent_id, + tx_hash=event_row["tx_hash"], + from_address=event_row["payer"], + ) + submitted += 1 + except PaymentCheckoutError as exc: + if not _is_nonfatal_submit_error(exc): + failed += 1 + logger.warning( + "payment event submit failed intent={} user={} tx={} status={} detail={}", + intent_id, + user_id, + event_row["tx_hash"], + exc.status_code, + exc.detail, + ) + continue + + try: + result = PAYMENT_CHECKOUT.confirm_intent_tx( + user_id=user_id, + intent_id=intent_id, + tx_hash=event_row["tx_hash"], + ) + if bool(result.get("already_confirmed")): + already += 1 + else: + confirmed += 1 + logger.info( + "payment event-confirmed intent={} user={} tx={} block={}", + intent_id, + user_id, + event_row["tx_hash"], + int(event_row.get("block_number") or 0), + ) + except PaymentCheckoutError as exc: + if _is_pending_confirm_error(exc): + pending += 1 + continue + failed += 1 + logger.warning( + "payment event confirm failed intent={} user={} tx={} status={} detail={}", + intent_id, + user_id, + event_row["tx_hash"], + exc.status_code, + exc.detail, + ) + + if scanned_events >= max_events: + break + + cursor = to_block + 1 + + if scanned_blocks > 0: + logger.info( + "payment event cycle blocks={} events={} matched={} submitted={} " + "confirmed={} already={} pending={} failed={} ignored={}", + scanned_blocks, + scanned_events, + matched_intents, + submitted, + confirmed, + already, + pending, + failed, + ignored, + ) + except Exception as exc: + logger.warning(f"payment event cycle failed: {exc}") + + elapsed = time.time() - cycle_started + time.sleep(max(0.0, interval_sec - elapsed)) + + +def start_payment_event_loop() -> threading.Thread: + thread = threading.Thread( + target=_runner, + daemon=True, + name="payment-event-loop", + ) + thread.start() + return thread diff --git a/src/utils/telegram_chat_ids.py b/src/utils/telegram_chat_ids.py index 77261d54..02cb2264 100644 --- a/src/utils/telegram_chat_ids.py +++ b/src/utils/telegram_chat_ids.py @@ -40,6 +40,26 @@ def get_telegram_chat_ids_from_env() -> List[str]: ) +def get_polymarket_wallet_activity_chat_ids_from_env() -> List[str]: + """ + Preferred envs for wallet activity push: + - POLYMARKET_WALLET_ACTIVITY_CHAT_IDS (comma-separated) + - POLYMARKET_WALLET_ACTIVITY_CHAT_ID (single) + + Falls back to global TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID for compatibility. + """ + dedicated = parse_telegram_chat_ids( + os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_IDS"), + os.getenv("POLYMARKET_WALLET_ACTIVITY_CHAT_ID"), + ) + if dedicated: + return dedicated + return parse_telegram_chat_ids( + os.getenv("TELEGRAM_CHAT_IDS"), + os.getenv("TELEGRAM_CHAT_ID"), + ) + + def get_primary_telegram_chat_id_from_env() -> str: chat_ids = get_telegram_chat_ids_from_env() return chat_ids[0] if chat_ids else "" diff --git a/web/app.py b/web/app.py index a82621ce..c4b37f85 100644 --- a/web/app.py +++ b/web/app.py @@ -1275,6 +1275,20 @@ async def payment_create_intent(request: Request, body: CreatePaymentIntentReque raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc +@app.get("/api/payments/intents/{intent_id}") +async def payment_get_intent(request: Request, intent_id: str): + _assert_entitlement(request) + identity = _require_supabase_identity(request) + try: + intent = PAYMENT_CHECKOUT.get_intent( + user_id=identity["user_id"], + intent_id=intent_id, + ) + return {"intent": intent.__dict__} + except PaymentCheckoutError as exc: + raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc + + @app.post("/api/payments/intents/{intent_id}/submit") async def payment_submit_tx( request: Request,