From 0d68da96246354366818695e919c299c0462bd69 Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Mon, 25 May 2026 09:16:52 +0300 Subject: [PATCH] fix: handle Wata payment links --- backend/bot/app/web/webapp/billing.py | 26 ++ backend/bot/payment_providers/wata.py | 425 ++++++++++++++++++++------ tests/test_wata_webhook.py | 67 ++++ tests/test_webapp_payment_status.py | 25 ++ 4 files changed, 452 insertions(+), 91 deletions(-) diff --git a/backend/bot/app/web/webapp/billing.py b/backend/bot/app/web/webapp/billing.py index ed9b5e4..f937c73 100644 --- a/backend/bot/app/web/webapp/billing.py +++ b/backend/bot/app/web/webapp/billing.py @@ -715,6 +715,31 @@ async def _refresh_yookassa_payment_status( return payment +async def _refresh_wata_payment_status( + request: web.Request, + session: AsyncSession, + payment: Payment, +) -> Payment: + if str(getattr(payment, "provider", "") or "").lower() != "wata": + return payment + if not _payment_status_can_be_refreshed(payment): + return payment + + wata_service = request.app.get("wata_service") + if ( + not wata_service + or not getattr(wata_service, "configured", False) + or not hasattr(wata_service, "refresh_payment_status") + ): + return payment + + try: + return await wata_service.refresh_payment_status(session, payment) + except Exception: + logger.exception("Failed to refresh Wata payment %s status", payment.payment_id) + return payment + + async def payment_status_route(request: web.Request) -> web.Response: user_id = _require_user_id(request) try: @@ -728,6 +753,7 @@ async def payment_status_route(request: web.Request) -> web.Response: if not payment or payment.user_id != user_id: return _json_error(404, "not_found", "Payment not found") payment = await _refresh_yookassa_payment_status(request, session, payment) + payment = await _refresh_wata_payment_status(request, session, payment) if payment.status == "succeeded": await invalidate_webapp_user_caches(request.app["settings"], user_id) return web.json_response( diff --git a/backend/bot/payment_providers/wata.py b/backend/bot/payment_providers/wata.py index 1b27cb0..a20ba3a 100644 --- a/backend/bot/payment_providers/wata.py +++ b/backend/bot/payment_providers/wata.py @@ -2,7 +2,7 @@ import base64 import json import logging from datetime import datetime, timedelta, timezone -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Mapping, Optional, Tuple from aiogram import Bot, F, Router, types from aiohttp import web @@ -51,6 +51,7 @@ from .shared import ( payment_unavailable, post_json_request, render_link_or_fail, + safe_callback_answer, ) router = Router(name="user_subscription_payments_wata_router") @@ -58,6 +59,33 @@ _LOG = "wata" _WATA_IN_PROGRESS_STATUSES = {"created", "pending"} +def _wata_success_status(status: int, _body: Any) -> bool: + return 200 <= status < 300 + + +def _normalized_wata_status(payload: Optional[Mapping[str, Any]]) -> str: + if not payload: + return "" + return ( + str( + payload.get("transactionStatus") + or payload.get("status") + or payload.get("statusName") + or "" + ) + .strip() + .lower() + ) + + +def _wata_transaction_id(payload: Optional[Mapping[str, Any]]) -> Optional[str]: + return first_value(payload, "transactionId", "id") + + +def _wata_payment_link_id(payload: Optional[Mapping[str, Any]]) -> Optional[str]: + return first_value(payload, "paymentLinkId", "payment_link_id") + + class WataConfig(ProviderEnvConfig): model_config = SettingsConfigDict( env_file=provider_env_file(), @@ -142,7 +170,7 @@ class WataService(HttpClientMixin): self._default_return_url = default_return_url self._cached_public_key_pem = None # populated by webhook on first verify - self._init_http_client(total_timeout=20) + self._init_http_client(total_timeout=10) if not self.configured: logging.warning("WataService initialized but not fully configured. Payments disabled.") @@ -217,6 +245,84 @@ class WataService(HttpClientMixin): body=body, headers=self._auth_headers(), log_prefix="Wata create_payment_link", + is_success=_wata_success_status, + ) + + async def _get_json( + self, + url: str, + *, + params: Optional[Mapping[str, Any]] = None, + log_prefix: str, + ) -> Tuple[bool, Dict[str, Any]]: + if not self.configured: + logging.error("WataService is not configured. Cannot fetch provider state.") + return False, {"message": "service_not_configured"} + + session = await self._get_session() + try: + async with session.get( + url, + params=dict(params or {}), + headers=self._auth_headers(), + ) as response: + response_text = await response.text() + try: + response_data = json.loads(response_text) if response_text else {} + except json.JSONDecodeError: + logging.error("%s: invalid JSON response: %s", log_prefix, response_text) + return False, { + "status": response.status, + "message": "invalid_json", + "raw": response_text, + } + if not _wata_success_status(response.status, response_data): + logging.error( + "%s: API returned error (status=%s, body=%s)", + log_prefix, + response.status, + response_data, + ) + return False, {"status": response.status, "message": response_data} + return True, response_data + except Exception as exc: + logging.exception("%s: request failed.", log_prefix) + return False, {"message": str(exc)} + + async def get_payment_link(self, payment_link_id: str) -> Tuple[bool, Dict[str, Any]]: + return await self._get_json( + f"{self.base_url}/links/{payment_link_id}", + log_prefix="Wata get_payment_link", + ) + + async def get_transaction(self, transaction_id: str) -> Tuple[bool, Dict[str, Any]]: + return await self._get_json( + f"{self.base_url}/transactions/{transaction_id}", + log_prefix="Wata get_transaction", + ) + + async def search_transactions( + self, + *, + order_id: Optional[str] = None, + payment_link_id: Optional[str] = None, + status: Optional[str] = None, + limit: int = 5, + ) -> Tuple[bool, Dict[str, Any]]: + params: Dict[str, Any] = { + "skipCount": 0, + "maxResultCount": max(1, min(int(limit or 5), 1000)), + } + if order_id: + params["orderId"] = order_id + if payment_link_id: + params["paymentLinkId"] = payment_link_id + if status: + params["statuses"] = status + return await self._get_json( + f"{self.base_url}/transactions", + params=params, + log_prefix="Wata search_transactions", ) async def _get_public_key_pem(self) -> Optional[str]: @@ -257,6 +363,211 @@ class WataService(HttpClientMixin): logging.exception("Wata webhook: signature verification failed.") return False + def _transaction_matches_payment( + self, + payload: Mapping[str, Any], + payment: Any, + *, + provider_payment_id: Optional[str], + ) -> bool: + order_id = str(payload.get("orderId") or "").strip() + if order_id and order_id == str(payment.payment_id): + return True + + payment_link_id = _wata_payment_link_id(payload) + if payment_link_id and provider_payment_id and payment_link_id == provider_payment_id: + return True + + transaction_id = _wata_transaction_id(payload) + if transaction_id and provider_payment_id and transaction_id == provider_payment_id: + return True + + return False + + async def _find_transaction_for_payment( + self, + payment: Any, + *, + status: str, + ) -> Optional[Dict[str, Any]]: + provider_payment_id = str(getattr(payment, "provider_payment_id", "") or "").strip() + success, response_data = await self.search_transactions( + order_id=str(payment.payment_id), + status=status, + limit=5, + ) + if success: + for item in response_data.get("items") or []: + if not isinstance(item, dict): + continue + if _normalized_wata_status(item) != status.lower(): + continue + if self._transaction_matches_payment( + item, + payment, + provider_payment_id=provider_payment_id or None, + ): + return item + + return None + + async def _mark_paid_from_payload( + self, + session: AsyncSession, + payment: Any, + payload: Mapping[str, Any], + *, + log_prefix: str, + ) -> Optional[Any]: + current = await payment_dal.get_payment_by_db_id(session, payment.payment_id) + if current: + payment = current + if payment.status == "succeeded": + return payment + + transaction_id = _wata_transaction_id(payload) or str(payment.payment_id) + amount_raw = payload.get("amount") + currency = payload.get("currency") or self.settings.DEFAULT_CURRENCY_SYMBOL or "RUB" + + if amount_raw is not None: + try: + if not decimal_amounts_equal(amount_raw, payment.amount): + logging.warning( + "%s: amount mismatch for payment %s (expected %s, got %s)", + log_prefix, + payment.payment_id, + format_decimal_amount(payment.amount), + format_decimal_amount(amount_raw), + ) + except Exception as exc: + logging.warning( + "%s: failed to compare amounts for %s: %s", + log_prefix, + payment.payment_id, + exc, + ) + + try: + await payment_dal.update_provider_payment_and_status( + session, + payment.payment_id, + transaction_id, + "succeeded", + ) + await session.commit() + except Exception: + await session.rollback() + logging.exception( + "%s: failed to mark payment %s as succeeded.", + log_prefix, + transaction_id, + ) + return None + + payment_units = payment.purchased_gb or payment.subscription_duration_months or 1 + sale_mode = payment.sale_mode or ( + "traffic" if self.settings.traffic_sale_mode else "subscription" + ) + outcome = await finalize_successful_payment( + PaymentSuccessRequest( + bot=self.bot, + settings=self.settings, + i18n=self.i18n, + session=session, + subscription_service=self.subscription_service, + referral_service=self.referral_service, + payment=payment, + user_id=payment.user_id, + amount=float(payment.amount), + currency=str(currency), + sale_mode=sale_mode, + months=payment_units, + traffic_amount=float(payment_units), + provider_subscription="wata", + provider_notification="wata", + db_user=payment.user, + log_prefix=log_prefix, + ) + ) + if outcome is None: + return None + return await payment_dal.get_payment_by_db_id(session, payment.payment_id) or payment + + async def _mark_declined_from_payload( + self, + session: AsyncSession, + payment: Any, + payload: Mapping[str, Any], + *, + log_prefix: str, + notify_user: bool, + ) -> Optional[Any]: + transaction_id = _wata_transaction_id(payload) or str(payment.payment_id) + try: + await payment_dal.update_provider_payment_and_status( + session, + payment.payment_id, + transaction_id, + "failed", + ) + await session.commit() + except Exception: + await session.rollback() + logging.exception( + "%s: failed to mark payment %s as failed.", + log_prefix, + transaction_id, + ) + return None + + if notify_user: + await notify_user_payment_failed( + bot=self.bot, + settings=self.settings, + i18n=self.i18n, + session=session, + payment=payment, + ) + return await payment_dal.get_payment_by_db_id(session, payment.payment_id) or payment + + async def refresh_payment_status(self, session: AsyncSession, payment: Any) -> Any: + if str(getattr(payment, "provider", "") or "").lower() != "wata": + return payment + if not self.configured: + return payment + + current_status = str(getattr(payment, "status", "") or "").lower() + if current_status == "succeeded" or current_status in { + "failed", + "canceled", + "cancelled", + "failed_creation", + }: + return payment + + paid_payload = await self._find_transaction_for_payment(payment, status="Paid") + if paid_payload: + refreshed = await self._mark_paid_from_payload( + session, + payment, + paid_payload, + log_prefix="Wata status refresh", + ) + return refreshed or payment + + declined_payload = await self._find_transaction_for_payment(payment, status="Declined") + if declined_payload: + refreshed = await self._mark_declined_from_payload( + session, + payment, + declined_payload, + log_prefix="Wata status refresh", + notify_user=False, + ) + return refreshed or payment + + return payment + async def webhook_route(self, request: web.Request) -> web.Response: if not self.configured: return web.Response(status=503, text="wata_disabled") @@ -291,8 +602,6 @@ class WataService(HttpClientMixin): payment_link_id = str(payload.get("paymentLinkId") or payload.get("id") or "").strip() status = str(payload.get("transactionStatus") or "").strip().lower() order_id_raw = payload.get("orderId") - amount_raw = payload.get("amount") - currency = payload.get("currency") or self.settings.DEFAULT_CURRENCY_SYMBOL or "RUB" if not status or not (transaction_id or order_id_raw or payment_link_id): logging.error("Wata webhook: missing transaction status or ids: %s", payload) @@ -322,8 +631,6 @@ class WataService(HttpClientMixin): if payment.status == "succeeded": return web.Response(text="ok") - resolved_transaction_id = transaction_id or payment_link_id or str(payment.payment_id) - if status in _WATA_IN_PROGRESS_STATUSES: if transaction_id and payment.provider_payment_id != transaction_id: try: @@ -345,92 +652,24 @@ class WataService(HttpClientMixin): return web.Response(text="ok") if status == "paid": - if amount_raw is not None: - try: - if not decimal_amounts_equal(amount_raw, payment.amount): - logging.warning( - "Wata webhook: amount mismatch for payment %s " - "(expected %s, got %s)", - payment.payment_id, - format_decimal_amount(payment.amount), - format_decimal_amount(amount_raw), - ) - except Exception as exc: - logging.warning( - "Wata webhook: failed to compare amounts for %s: %s", - payment.payment_id, - exc, - ) - - try: - await payment_dal.update_provider_payment_and_status( - session, - payment.payment_id, - resolved_transaction_id, - "succeeded", - ) - await session.commit() - except Exception: - await session.rollback() - logging.exception( - "Wata webhook: failed to mark payment %s as succeeded.", - resolved_transaction_id, - ) - return web.Response(status=500, text="processing_error") - - payment_units = payment.purchased_gb or payment.subscription_duration_months or 1 - sale_mode = payment.sale_mode or ( - "traffic" if self.settings.traffic_sale_mode else "subscription" - ) - - outcome = await finalize_successful_payment( - PaymentSuccessRequest( - bot=self.bot, - settings=self.settings, - i18n=self.i18n, - session=session, - subscription_service=self.subscription_service, - referral_service=self.referral_service, - payment=payment, - user_id=payment.user_id, - amount=float(payment.amount), - currency=str(currency), - sale_mode=sale_mode, - months=payment_units, - traffic_amount=float(payment_units), - provider_subscription="wata", - provider_notification="wata", - db_user=payment.user, - log_prefix="Wata webhook", - ) - ) - if outcome is None: + if not await self._mark_paid_from_payload( + session, + payment, + payload, + log_prefix="Wata webhook", + ): return web.Response(status=500, text="processing_error") return web.Response(text="ok") if status == "declined": - try: - await payment_dal.update_provider_payment_and_status( - session, - payment.payment_id, - resolved_transaction_id, - "failed", - ) - await session.commit() - except Exception: - await session.rollback() - logging.exception( - "Wata webhook: failed to mark payment %s as failed.", - resolved_transaction_id, - ) + if not await self._mark_declined_from_payload( + session, + payment, + payload, + log_prefix="Wata webhook", + notify_user=True, + ): return web.Response(status=500, text="processing_error") - await notify_user_payment_failed( - bot=self.bot, - settings=self.settings, - i18n=self.i18n, - session=session, - payment=payment, - ) return web.Response(text="ok") logging.warning( @@ -492,6 +731,8 @@ async def pay_wata_callback_handler( await notify_payment_record_failure(callback, translator) return + await safe_callback_answer(callback) + success, response_data = await wata_service.create_payment_link( payment_db_id=payment_record.payment_id, amount=parts.price, @@ -507,8 +748,8 @@ async def pay_wata_callback_handler( session=session, payment=payment_record, api_success=success, - payment_url=first_value(response_data, "url"), - provider_payment_id=first_value(response_data, "id"), + payment_url=first_value(response_data, "url", "paymentUrl", "payment_url"), + provider_payment_id=first_value(response_data, "id", "paymentLinkId"), log_prefix=_LOG, ) @@ -543,8 +784,10 @@ async def create_webapp_payment(ctx: WebAppPaymentContext) -> web.Response: session=ctx.session, payment=payment, api_success=success, - payment_url=first_value(response_data, "url") if success else None, - provider_payment_id=first_value(response_data, "id"), + payment_url=first_value(response_data, "url", "paymentUrl", "payment_url") + if success + else None, + provider_payment_id=first_value(response_data, "id", "paymentLinkId"), log_prefix="Wata", ) diff --git a/tests/test_wata_webhook.py b/tests/test_wata_webhook.py index 332e26e..454fbbb 100644 --- a/tests/test_wata_webhook.py +++ b/tests/test_wata_webhook.py @@ -201,3 +201,70 @@ def test_wata_known_payment_with_unknown_status_still_acknowledges_webhook(monke assert response.status == 200 assert session.commits == 0 + + +def test_wata_refresh_finds_paid_transaction_by_order_id_and_finalizes(monkeypatch): + session = _FakeSession() + payment = _payment(provider="wata", provider_payment_id="link-id") + updates = [] + finalized = [] + service = _service(session) + + async def search_transactions(*, order_id=None, payment_link_id=None, status=None, limit=5): + assert order_id == "465" + assert payment_link_id is None + assert status == "Paid" + assert limit == 5 + return True, { + "items": [ + { + "id": "tx-paid", + "status": "Paid", + "orderId": "465", + "amount": 100, + "currency": "RUB", + "paymentLinkId": "link-id", + } + ] + } + + async def get_payment_by_db_id(_session, payment_id): + assert _session is session + assert payment_id == 465 + return payment + + async def update_provider_payment_and_status( + _session, + payment_id, + provider_payment_id, + status, + ): + updates.append((payment_id, provider_payment_id, status)) + payment.provider_payment_id = provider_payment_id + payment.status = status + + async def finalize_successful_payment(request): + finalized.append( + ( + request.payment.payment_id, + request.provider_subscription, + request.provider_notification, + ) + ) + return SimpleNamespace() + + service.search_transactions = search_transactions + monkeypatch.setattr(wata.payment_dal, "get_payment_by_db_id", get_payment_by_db_id) + monkeypatch.setattr( + wata.payment_dal, + "update_provider_payment_and_status", + update_provider_payment_and_status, + ) + monkeypatch.setattr(wata, "finalize_successful_payment", finalize_successful_payment) + + result = asyncio.run(service.refresh_payment_status(session, payment)) + + assert result is payment + assert updates == [(465, "tx-paid", "succeeded")] + assert finalized == [(465, "wata", "wata")] + assert session.commits == 1 diff --git a/tests/test_webapp_payment_status.py b/tests/test_webapp_payment_status.py index a87102e..ff15414 100644 --- a/tests/test_webapp_payment_status.py +++ b/tests/test_webapp_payment_status.py @@ -166,3 +166,28 @@ class WebAppPaymentStatusTests(IsolatedAsyncioTestCase): invalidate_cache.assert_awaited_once_with(settings, 1001) self.assertEqual(response.status, 200) + + async def test_wata_pending_payment_refresh_delegates_to_provider_service(self): + payment = SimpleNamespace( + payment_id=43, + user_id=1001, + provider="wata", + status="pending_wata", + ) + refreshed_payment = SimpleNamespace( + payment_id=43, + user_id=1001, + provider="wata", + status="succeeded", + ) + wata_service = SimpleNamespace( + configured=True, + refresh_payment_status=AsyncMock(return_value=refreshed_payment), + ) + request = SimpleNamespace(app={"wata_service": wata_service}) + session = AsyncMock() + + result = await billing_module._refresh_wata_payment_status(request, session, payment) + + self.assertIs(result, refreshed_payment) + wata_service.refresh_payment_status.assert_awaited_once_with(session, payment)