diff --git a/app/cabinet/routes/admin_payments.py b/app/cabinet/routes/admin_payments.py index a736bbb6..b8c71254 100644 --- a/app/cabinet/routes/admin_payments.py +++ b/app/cabinet/routes/admin_payments.py @@ -4,10 +4,14 @@ import math from datetime import datetime import structlog +from aiogram import Bot +from aiogram.client.default import DefaultBotProperties +from aiogram.enums import ParseMode from fastapi import APIRouter, Depends, HTTPException, Query, status from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession +from app.config import settings from app.database.models import PaymentMethod, User from app.services.payment_service import PaymentService from app.services.payment_verification_service import ( @@ -390,8 +394,12 @@ async def check_payment_status( old_is_paid = record.is_paid # Run manual check - payment_service = PaymentService() - updated = await run_manual_check(db, payment_method, payment_id, payment_service) + bot = Bot(token=settings.BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) + try: + payment_service = PaymentService(bot=bot) + updated = await run_manual_check(db, payment_method, payment_id, payment_service) + finally: + await bot.session.close() if not updated: return ManualCheckResponse( diff --git a/app/cabinet/routes/balance.py b/app/cabinet/routes/balance.py index 4e3709d4..822e33f9 100644 --- a/app/cabinet/routes/balance.py +++ b/app/cabinet/routes/balance.py @@ -6,6 +6,9 @@ from decimal import ROUND_HALF_UP, Decimal, InvalidOperation import httpx import structlog +from aiogram import Bot +from aiogram.client.default import DefaultBotProperties +from aiogram.enums import ParseMode from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy import desc, func, select from sqlalchemy.ext.asyncio import AsyncSession @@ -1041,8 +1044,12 @@ async def check_payment_status( old_is_paid = record.is_paid # Run manual check - payment_service = PaymentService() - updated = await run_manual_check(db, payment_method, payment_id, payment_service) + bot = Bot(token=settings.BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) + try: + payment_service = PaymentService(bot=bot) + updated = await run_manual_check(db, payment_method, payment_id, payment_service) + finally: + await bot.session.close() if not updated: return ManualCheckResponse( diff --git a/app/database/crud/cloudpayments.py b/app/database/crud/cloudpayments.py index 1b03a84f..bdde4694 100644 --- a/app/database/crud/cloudpayments.py +++ b/app/database/crud/cloudpayments.py @@ -92,6 +92,16 @@ async def get_cloudpayments_payment_by_id( return result.scalars().first() +async def get_cloudpayments_payment_by_id_for_update( + db: AsyncSession, + payment_id: int, +) -> CloudPaymentsPayment | None: + result = await db.execute( + select(CloudPaymentsPayment).where(CloudPaymentsPayment.id == payment_id).with_for_update() + ) + return result.scalar_one_or_none() + + async def get_cloudpayments_payment_by_transaction_id( db: AsyncSession, transaction_id_cp: int, diff --git a/app/database/crud/cryptobot.py b/app/database/crud/cryptobot.py index 30af3e46..566e7e21 100644 --- a/app/database/crud/cryptobot.py +++ b/app/database/crud/cryptobot.py @@ -67,6 +67,11 @@ async def get_cryptobot_payment_by_id(db: AsyncSession, payment_id: int) -> Cryp return result.scalar_one_or_none() +async def get_cryptobot_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> CryptoBotPayment | None: + result = await db.execute(select(CryptoBotPayment).where(CryptoBotPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def update_cryptobot_payment_status( db: AsyncSession, invoice_id: str, status: str, paid_at: datetime | None = None ) -> CryptoBotPayment | None: @@ -99,7 +104,7 @@ async def link_cryptobot_payment_to_transaction( payment.transaction_id = transaction_id payment.updated_at = datetime.now(UTC) - await db.commit() + await db.flush() await db.refresh(payment) logger.info('Связан CryptoBot платеж с транзакцией', invoice_id=invoice_id, transaction_id=transaction_id) diff --git a/app/database/crud/freekassa.py b/app/database/crud/freekassa.py index b6ed4a07..8140684b 100644 --- a/app/database/crud/freekassa.py +++ b/app/database/crud/freekassa.py @@ -63,6 +63,11 @@ async def get_freekassa_payment_by_id(db: AsyncSession, payment_id: int) -> Free return result.scalar_one_or_none() +async def get_freekassa_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> FreekassaPayment | None: + result = await db.execute(select(FreekassaPayment).where(FreekassaPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def update_freekassa_payment_status( db: AsyncSession, payment: FreekassaPayment, diff --git a/app/database/crud/heleket.py b/app/database/crud/heleket.py index 3889c1a9..ad226e63 100644 --- a/app/database/crud/heleket.py +++ b/app/database/crud/heleket.py @@ -91,6 +91,11 @@ async def get_heleket_payment_by_id( return result.scalar_one_or_none() +async def get_heleket_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> HeleketPayment | None: + result = await db.execute(select(HeleketPayment).where(HeleketPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def update_heleket_payment( db: AsyncSession, uuid: str, @@ -159,7 +164,7 @@ async def link_heleket_payment_to_transaction( payment.transaction_id = transaction_id payment.updated_at = datetime.now(UTC) - await db.commit() + await db.flush() await db.refresh(payment) logger.info('Heleket платеж связан с транзакцией', uuid=uuid, transaction_id=transaction_id) diff --git a/app/database/crud/kassa_ai.py b/app/database/crud/kassa_ai.py index 16225c2a..16c4112b 100644 --- a/app/database/crud/kassa_ai.py +++ b/app/database/crud/kassa_ai.py @@ -65,6 +65,11 @@ async def get_kassa_ai_payment_by_id(db: AsyncSession, payment_id: int) -> Kassa return result.scalar_one_or_none() +async def get_kassa_ai_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> KassaAiPayment | None: + result = await db.execute(select(KassaAiPayment).where(KassaAiPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def update_kassa_ai_payment_status( db: AsyncSession, payment: KassaAiPayment, diff --git a/app/database/crud/mulenpay.py b/app/database/crud/mulenpay.py index 97dbc3a8..059a9850 100644 --- a/app/database/crud/mulenpay.py +++ b/app/database/crud/mulenpay.py @@ -57,6 +57,11 @@ async def get_mulenpay_payment_by_local_id(db: AsyncSession, payment_id: int) -> return result.scalar_one_or_none() +async def get_mulenpay_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> MulenPayPayment | None: + result = await db.execute(select(MulenPayPayment).where(MulenPayPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def get_mulenpay_payment_by_uuid(db: AsyncSession, uuid: str) -> MulenPayPayment | None: result = await db.execute(select(MulenPayPayment).where(MulenPayPayment.uuid == uuid)) return result.scalar_one_or_none() @@ -117,6 +122,6 @@ async def link_mulenpay_payment_to_transaction( ) -> MulenPayPayment: payment.transaction_id = transaction_id payment.updated_at = datetime.now(UTC) - await db.commit() + await db.flush() await db.refresh(payment) return payment diff --git a/app/database/crud/pal24.py b/app/database/crud/pal24.py index 50c042bd..514d8592 100644 --- a/app/database/crud/pal24.py +++ b/app/database/crud/pal24.py @@ -66,6 +66,11 @@ async def get_pal24_payment_by_id(db: AsyncSession, payment_id: int) -> Pal24Pay return result.scalar_one_or_none() +async def get_pal24_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> Pal24Payment | None: + result = await db.execute(select(Pal24Payment).where(Pal24Payment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def get_pal24_payment_by_bill_id(db: AsyncSession, bill_id: str) -> Pal24Payment | None: result = await db.execute(select(Pal24Payment).where(Pal24Payment.bill_id == bill_id)) return result.scalar_one_or_none() @@ -143,7 +148,7 @@ async def link_pal24_payment_to_transaction( transaction_id: int, ) -> Pal24Payment: await db.execute(update(Pal24Payment).where(Pal24Payment.id == payment.id).values(transaction_id=transaction_id)) - await db.commit() + await db.flush() await db.refresh(payment) logger.info('Pal24 платеж привязан к транзакции', bill_id=payment.bill_id, transaction_id=transaction_id) return payment diff --git a/app/database/crud/platega.py b/app/database/crud/platega.py index 6d676f9c..ade0226a 100644 --- a/app/database/crud/platega.py +++ b/app/database/crud/platega.py @@ -130,6 +130,6 @@ async def link_platega_payment_to_transaction( ) -> PlategaPayment: payment.transaction_id = transaction_id payment.updated_at = datetime.now(UTC) - await db.commit() + await db.flush() await db.refresh(payment) return payment diff --git a/app/database/crud/promocode.py b/app/database/crud/promocode.py index e739d2b1..d454fe7f 100644 --- a/app/database/crud/promocode.py +++ b/app/database/crud/promocode.py @@ -84,7 +84,6 @@ async def create_promocode( return promocode - async def check_user_promocode_usage(db: AsyncSession, user_id: int, promocode_id: int) -> bool: result = await db.execute( select(PromoCodeUse).where(and_(PromoCodeUse.user_id == user_id, PromoCodeUse.promocode_id == promocode_id)) diff --git a/app/database/crud/transaction.py b/app/database/crud/transaction.py index 2499d90f..fff3bbcc 100644 --- a/app/database/crud/transaction.py +++ b/app/database/crud/transaction.py @@ -38,6 +38,8 @@ async def create_transaction( external_id: str | None = None, is_completed: bool = True, created_at: datetime | None = None, + *, + commit: bool = True, ) -> Transaction: # SUBSCRIPTION_PAYMENT — always store as negative (debit from user balance) # Keep original for downstream consumers (events, contests) @@ -58,7 +60,10 @@ async def create_transaction( ) db.add(transaction) - await db.commit() + if commit: + await db.commit() + else: + await db.flush() await db.refresh(transaction) logger.info( @@ -68,7 +73,69 @@ async def create_transaction( user_id=user_id, ) - # Отправляем событие о транзакции + # Side-effects skipped when commit=False to preserve caller's transaction atomicity. + # Callers using commit=False should call emit_transaction_side_effects() after their own db.commit(). + if commit: + try: + from app.services.event_emitter import event_emitter + + await event_emitter.emit( + 'payment.completed' if type == TransactionType.DEPOSIT else 'transaction.created', + { + 'transaction_id': transaction.id, + 'user_id': user_id, + 'type': type.value, + 'amount_kopeks': abs(amount_kopeks), + 'amount_rubles': abs(amount_kopeks) / 100, + 'payment_method': payment_method.value if payment_method else None, + 'external_id': external_id, + 'is_completed': is_completed, + 'description': description, + }, + db=db, + ) + except Exception as error: + logger.warning('Failed to emit transaction event', error=error) + + try: + from app.services.promo_group_assignment import ( + maybe_assign_promo_group_by_total_spent, + ) + + await maybe_assign_promo_group_by_total_spent(db, user_id) + except Exception as exc: + logger.debug('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc) + if type == TransactionType.SUBSCRIPTION_PAYMENT and is_completed: + try: + from app.services.referral_contest_service import referral_contest_service + + await referral_contest_service.on_subscription_payment( + db, + user_id, + abs(amount_kopeks), + ) + except Exception as exc: + logger.debug('Не удалось записать событие конкурса для пользователя', user_id=user_id, exc=exc) + + return transaction + + +async def emit_transaction_side_effects( + db: AsyncSession, + transaction: Transaction, + *, + amount_kopeks: int, + user_id: int, + type: TransactionType, + payment_method: PaymentMethod | None = None, + external_id: str | None = None, + is_completed: bool = True, + description: str = '', +) -> None: + """Fire side-effects that were deferred when create_transaction(commit=False) was used. + + Call this AFTER db.commit() to emit events and run promo checks. + """ try: from app.services.event_emitter import event_emitter @@ -88,7 +155,7 @@ async def create_transaction( db=db, ) except Exception as error: - logger.warning('Failed to emit transaction event', error=error) + logger.warning('Failed to emit deferred transaction event', error=error) try: from app.services.promo_group_assignment import ( @@ -98,6 +165,7 @@ async def create_transaction( await maybe_assign_promo_group_by_total_spent(db, user_id) except Exception as exc: logger.debug('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc) + if type == TransactionType.SUBSCRIPTION_PAYMENT and is_completed: try: from app.services.referral_contest_service import referral_contest_service @@ -110,8 +178,6 @@ async def create_transaction( except Exception as exc: logger.debug('Не удалось записать событие конкурса для пользователя', user_id=user_id, exc=exc) - return transaction - async def get_transaction_by_id(db: AsyncSession, transaction_id: int) -> Transaction | None: result = await db.execute( diff --git a/app/database/crud/wata.py b/app/database/crud/wata.py index 8cf2bf31..e35219c7 100644 --- a/app/database/crud/wata.py +++ b/app/database/crud/wata.py @@ -71,6 +71,11 @@ async def get_wata_payment_by_id( return result.scalar_one_or_none() +async def get_wata_payment_by_id_for_update(db: AsyncSession, payment_id: int) -> WataPayment | None: + result = await db.execute(select(WataPayment).where(WataPayment.id == payment_id).with_for_update()) + return result.scalar_one_or_none() + + async def get_wata_payment_by_link_id( db: AsyncSession, payment_link_id: str, @@ -143,7 +148,7 @@ async def link_wata_payment_to_transaction( transaction_id: int, ) -> WataPayment: await db.execute(update(WataPayment).where(WataPayment.id == payment.id).values(transaction_id=transaction_id)) - await db.commit() + await db.flush() await db.refresh(payment) logger.info( diff --git a/app/database/crud/yookassa.py b/app/database/crud/yookassa.py index 0841e56c..d77728f9 100644 --- a/app/database/crud/yookassa.py +++ b/app/database/crud/yookassa.py @@ -127,7 +127,7 @@ async def link_yookassa_payment_to_transaction( .where(YooKassaPayment.yookassa_payment_id == yookassa_payment_id) .values(transaction_id=transaction_id, updated_at=datetime.now(UTC)) ) - await db.commit() + await db.flush() result = await db.execute( select(YooKassaPayment) diff --git a/app/database/models.py b/app/database/models.py index 98783441..87fae4e8 100644 --- a/app/database/models.py +++ b/app/database/models.py @@ -1393,6 +1393,7 @@ class TrafficPurchase(Base): class Transaction(Base): __tablename__ = 'transactions' __table_args__ = ( + UniqueConstraint('external_id', 'payment_method', name='uq_transaction_external_id_method'), Index('ix_transactions_type_created_completed', 'type', 'created_at', 'is_completed'), Index('ix_transactions_user_created', 'user_id', 'created_at'), Index('ix_transactions_type_method_created', 'type', 'payment_method', 'created_at'), @@ -1502,9 +1503,7 @@ class PromoCode(Base): class PromoCodeUse(Base): __tablename__ = 'promocode_uses' - __table_args__ = ( - UniqueConstraint('user_id', 'promocode_id', name='uq_promocode_uses_user_promo'), - ) + __table_args__ = (UniqueConstraint('user_id', 'promocode_id', name='uq_promocode_uses_user_promo'),) id = Column(Integer, primary_key=True, index=True) promocode_id = Column(Integer, ForeignKey('promocodes.id'), nullable=False) diff --git a/app/external/yookassa_webhook.py b/app/external/yookassa_webhook.py index 4d75be4a..f47c0d6e 100644 --- a/app/external/yookassa_webhook.py +++ b/app/external/yookassa_webhook.py @@ -207,7 +207,6 @@ class YooKassaWebhookHandler: async def handle_webhook(self, request: web.Request) -> web.Response: try: logger.info('📥 Получен YooKassa webhook', method=request.method, path=request.path) - logger.info('📋 Headers', value=dict(request.headers)) header_ip_candidates = collect_yookassa_ip_candidates( request.headers.get('X-Forwarded-For'), @@ -242,7 +241,7 @@ class YooKassaWebhookHandler: logger.warning('⚠️ Получен пустой webhook от YooKassa') return web.Response(status=400, text='Empty body') - logger.info('📄 Body', body=body) + logger.debug('📄 Body received', length=len(body)) signature = request.headers.get('Signature') or request.headers.get('X-YooKassa-Signature') if signature: diff --git a/app/handlers/start.py b/app/handlers/start.py index dda88be8..fb00bef1 100644 --- a/app/handlers/start.py +++ b/app/handlers/start.py @@ -1168,11 +1168,6 @@ async def complete_registration_from_callback(callback: types.CallbackQuery, sta language = data.get('language', DEFAULT_LANGUAGE) texts = get_texts(language) - campaign_id = data.get('campaign_id') - is_new_user_registration = existing_user is None or ( - existing_user and existing_user.status == UserStatus.DELETED.value - ) - referrer_id = data.get('referrer_id') if not referrer_id and data.get('referral_code'): referrer = await get_user_by_referral_code(db, data['referral_code']) @@ -1279,16 +1274,7 @@ async def complete_registration_from_callback(callback: types.CallbackQuery, sta offer_text = await get_welcome_text_for_user(db, callback.from_user) - skip_welcome_offer = bool(campaign_id) and is_new_user_registration - - if skip_welcome_offer: - logger.info( - 'ℹ️ Пропускаем приветственное предложение для нового пользователя из рекламной кампании', - telegram_id=user.telegram_id, - campaign_id=campaign_id, - ) - - if offer_text and not skip_welcome_offer: + if offer_text: try: await callback.message.answer( offer_text, @@ -1433,11 +1419,6 @@ async def complete_registration(message: types.Message, state: FSMContext, db: A language = data.get('language', DEFAULT_LANGUAGE) texts = get_texts(language) - campaign_id = data.get('campaign_id') - is_new_user_registration = existing_user is None or ( - existing_user and existing_user.status == UserStatus.DELETED.value - ) - referrer_id = data.get('referrer_id') if not referrer_id and data.get('referral_code'): referrer = await get_user_by_referral_code(db, data['referral_code']) @@ -1573,16 +1554,7 @@ async def complete_registration(message: types.Message, state: FSMContext, db: A offer_text = await get_welcome_text_for_user(db, message.from_user) - skip_welcome_offer = bool(campaign_id) and is_new_user_registration - - if skip_welcome_offer: - logger.info( - 'ℹ️ Пропускаем приветственное предложение для нового пользователя из рекламной кампании', - telegram_id=user.telegram_id, - campaign_id=campaign_id, - ) - - if offer_text and not skip_welcome_offer: + if offer_text: try: # Если у пользователя уже есть подписка (например, от промокода), не предлагаем триал user_has_subscription = user.subscription and getattr(user.subscription, 'is_active', False) diff --git a/app/services/payment/cloudpayments.py b/app/services/payment/cloudpayments.py index 231fa6a4..49daac40 100644 --- a/app/services/payment/cloudpayments.py +++ b/app/services/payment/cloudpayments.py @@ -138,7 +138,7 @@ class CloudPaymentsPaymentMixin: invoice_id = webhook_data.get('invoice_id') transaction_id_cp = webhook_data.get('transaction_id') amount = webhook_data.get('amount', 0) - amount_kopeks = int(amount * 100) + amount_kopeks = int(round(amount * 100)) account_id = webhook_data.get('account_id', '') token = webhook_data.get('token') test_mode = webhook_data.get('test_mode', False) @@ -186,11 +186,32 @@ class CloudPaymentsPaymentMixin: logger.error('Не удалось создать запись платежа') return False - # Check if already processed - if payment.is_paid: + # Lock payment row to prevent concurrent double-processing + from app.database.crud.cloudpayments import get_cloudpayments_payment_by_id_for_update + + locked = await get_cloudpayments_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('CloudPayments: не удалось заблокировать платёж', payment_id=payment.id) + return False + payment = locked + + # Check if already processed (under lock) + if payment.is_paid or payment.transaction_id: logger.info('CloudPayments платёж уже обработан: invoice', invoice_id=invoice_id) return True + # Verify webhook amount matches stored amount + from app.utils.payment_utils import verify_payment_amount + + if not verify_payment_amount(amount_kopeks, payment.amount_kopeks): + logger.warning( + 'CloudPayments: несоответствие суммы', + invoice_id=invoice_id, + received_kopeks=amount_kopeks, + expected_kopeks=payment.amount_kopeks, + ) + return False + # Update payment record payment.transaction_id_cp = transaction_id_cp payment.status = 'completed' @@ -205,10 +226,8 @@ class CloudPaymentsPaymentMixin: payment.test_mode = test_mode payment.callback_payload = webhook_data - await db.flush() - # Get user - from app.database.crud.user import add_user_balance, get_user_by_id + from app.database.crud.user import get_user_by_id user = await get_user_by_id(db, payment.user_id) @@ -216,8 +235,9 @@ class CloudPaymentsPaymentMixin: logger.error('Пользователь не найден: id', user_id=payment.user_id) return False - # Add balance (без автоматической транзакции - создадим ниже с external_id) - await add_user_balance(db, user, amount_kopeks, create_transaction=False) + # Credit balance directly (not via add_user_balance which commits) + user.balance_kopeks += amount_kopeks + user.updated_at = datetime.now(UTC) # Create transaction record from app.database.crud.transaction import create_transaction @@ -232,11 +252,26 @@ class CloudPaymentsPaymentMixin: external_id=str(transaction_id_cp) if transaction_id_cp else invoice_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) payment.transaction_id = transaction.id await db.commit() + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=amount_kopeks, + user_id=user.id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.CLOUDPAYMENTS, + external_id=str(transaction_id_cp) if transaction_id_cp else invoice_id, + description=payment.description or settings.CLOUDPAYMENTS_DESCRIPTION, + ) + user_id_display = user.telegram_id or user.email or f'#{user.id}' logger.info( 'CloudPayments платёж успешно обработан: invoice amount=₽, user', diff --git a/app/services/payment/cryptobot.py b/app/services/payment/cryptobot.py index e1193fcf..080e5ea0 100644 --- a/app/services/payment/cryptobot.py +++ b/app/services/payment/cryptobot.py @@ -195,6 +195,12 @@ class CryptoBotPaymentMixin: if renewal_handled: return True + locked = await cryptobot_crud.get_cryptobot_payment_by_id_for_update(db, updated_payment.id) + if not locked: + logger.error('CryptoBot: не удалось заблокировать платёж', payment_id=updated_payment.id) + return False + updated_payment = locked + if not updated_payment.transaction_id: amount_usd = updated_payment.amount_float @@ -241,6 +247,7 @@ class CryptoBotPaymentMixin: external_id=invoice_id, is_completed=True, created_at=getattr(updated_payment, 'created_at', None), + commit=False, ) await cryptobot_crud.link_cryptobot_payment_to_transaction(db, invoice_id, transaction.id) @@ -262,6 +269,19 @@ class CryptoBotPaymentMixin: await db.commit() + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=amount_kopeks, + user_id=updated_payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.CRYPTOBOT, + external_id=invoice_id, + ) + try: from app.services.referral_service import process_referral_topup diff --git a/app/services/payment/freekassa.py b/app/services/payment/freekassa.py index 8a0e0b3c..0ab392e5 100644 --- a/app/services/payment/freekassa.py +++ b/app/services/payment/freekassa.py @@ -250,6 +250,13 @@ class FreekassaPaymentMixin: """Создаёт транзакцию, начисляет баланс и отправляет уведомления.""" payment_module = import_module('app.services.payment_service') + freekassa_lock_crud = import_module('app.database.crud.freekassa') + locked = await freekassa_lock_crud.get_freekassa_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('Freekassa: не удалось заблокировать платёж', payment_id=payment.id) + return False + payment = locked + if payment.transaction_id: logger.info( 'Freekassa платеж уже привязан к транзакции (trigger=)', order_id=payment.order_id, trigger=trigger @@ -278,16 +285,13 @@ class FreekassaPaymentMixin: external_id=str(intid) if intid else payment.order_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) - # Связываем платеж с транзакцией - freekassa_crud = import_module('app.database.crud.freekassa') - await freekassa_crud.update_freekassa_payment_status( - db=db, - payment=payment, - status=payment.status, - transaction_id=transaction.id, - ) + # Связываем платеж с транзакцией (без commit, чтобы сохранить атомарность) + payment.transaction_id = transaction.id + payment.updated_at = datetime.now(UTC) + await db.flush() old_balance = user.balance_kopeks was_first_topup = not user.has_made_first_topup @@ -303,6 +307,19 @@ class FreekassaPaymentMixin: await db.commit() + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.FREEKASSA, + external_id=str(intid) if intid else payment.order_id, + ) + # Обработка реферального пополнения try: from app.services.referral_service import process_referral_topup diff --git a/app/services/payment/heleket.py b/app/services/payment/heleket.py index f7589ac3..e70f329f 100644 --- a/app/services/payment/heleket.py +++ b/app/services/payment/heleket.py @@ -276,6 +276,13 @@ class HeleketPaymentMixin: except Exception as error: # pragma: no cover - diagnostics logger.warning('Не удалось обновить метаданные Heleket после удаления счёта', error=error) + heleket_lock_crud = import_module('app.database.crud.heleket') + locked = await heleket_lock_crud.get_heleket_payment_by_id_for_update(db, updated_payment.id) + if not locked: + logger.error('Heleket: не удалось заблокировать платёж', payment_id=updated_payment.id) + return None + updated_payment = locked + if updated_payment.transaction_id: logger.info( 'Heleket платеж уже связан с транзакцией', @@ -309,6 +316,7 @@ class HeleketPaymentMixin: external_id=updated_payment.uuid, is_completed=True, created_at=getattr(updated_payment, 'created_at', None), + commit=False, ) linked_payment = await heleket_crud.link_heleket_payment_to_transaction( @@ -334,6 +342,19 @@ class HeleketPaymentMixin: await db.commit() await db.refresh(user) + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=amount_kopeks, + user_id=updated_payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.HELEKET, + external_id=updated_payment.uuid, + ) + try: from app.services.referral_service import process_referral_topup diff --git a/app/services/payment/kassa_ai.py b/app/services/payment/kassa_ai.py index 02ed7501..ad4f9670 100644 --- a/app/services/payment/kassa_ai.py +++ b/app/services/payment/kassa_ai.py @@ -236,6 +236,13 @@ class KassaAiPaymentMixin: """Создаёт транзакцию, начисляет баланс и отправляет уведомления.""" payment_module = import_module('app.services.payment_service') + kassa_ai_lock_crud = import_module('app.database.crud.kassa_ai') + locked = await kassa_ai_lock_crud.get_kassa_ai_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('KassaAI: не удалось заблокировать платёж', payment_id=payment.id) + return False + payment = locked + if payment.transaction_id: logger.info( 'KassaAI платеж уже привязан к транзакции (trigger=)', order_id=payment.order_id, trigger=trigger @@ -264,16 +271,13 @@ class KassaAiPaymentMixin: external_id=str(intid) if intid else payment.order_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) - # Связываем платеж с транзакцией - kassa_ai_crud = import_module('app.database.crud.kassa_ai') - await kassa_ai_crud.update_kassa_ai_payment_status( - db=db, - payment=payment, - status=payment.status, - transaction_id=transaction.id, - ) + # Связываем платеж с транзакцией (без commit, чтобы сохранить атомарность) + payment.transaction_id = transaction.id + payment.updated_at = datetime.now(UTC) + await db.flush() old_balance = user.balance_kopeks was_first_topup = not user.has_made_first_topup @@ -289,6 +293,19 @@ class KassaAiPaymentMixin: await db.commit() + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.KASSA_AI, + external_id=str(intid) if intid else payment.order_id, + ) + # Обработка реферального пополнения try: from app.services.referral_service import process_referral_topup diff --git a/app/services/payment/mulenpay.py b/app/services/payment/mulenpay.py index dadaa325..abaee07a 100644 --- a/app/services/payment/mulenpay.py +++ b/app/services/payment/mulenpay.py @@ -3,6 +3,7 @@ from __future__ import annotations import uuid +from datetime import UTC, datetime from importlib import import_module from typing import Any @@ -225,6 +226,13 @@ class MulenPayPaymentMixin: metadata=metadata, ) + mulenpay_lock_crud = import_module('app.database.crud.mulenpay') + locked = await mulenpay_lock_crud.get_mulenpay_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('MulenPay: не удалось заблокировать платёж', payment_id=payment.id) + return False + payment = locked + if payment.transaction_id: logger.info('Для платежа уже создана транзакция', display_name=display_name, uuid=payment.uuid) return True @@ -245,6 +253,7 @@ class MulenPayPaymentMixin: external_id=payment.uuid, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) await payment_module.link_mulenpay_payment_to_transaction( @@ -263,12 +272,23 @@ class MulenPayPaymentMixin: old_balance = user.balance_kopeks was_first_topup = not user.has_made_first_topup - await payment_module.add_user_balance( + # Начисляем баланс напрямую (без add_user_balance, который делает db.commit()) + user.balance_kopeks += payment.amount_kopeks + user.updated_at = datetime.now(UTC) + + await db.commit() + + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( db, - user, - payment.amount_kopeks, - f'Пополнение {display_name}: {payment.amount_kopeks // 100}₽', - create_transaction=False, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.MULENPAY, + external_id=payment.uuid, ) try: diff --git a/app/services/payment/pal24.py b/app/services/payment/pal24.py index fbf86d27..680349bb 100644 --- a/app/services/payment/pal24.py +++ b/app/services/payment/pal24.py @@ -346,6 +346,13 @@ class Pal24PaymentMixin: except Exception as error: # pragma: no cover - diagnostics logger.warning('Не удалось обновить метаданные PayPalych после удаления счёта', error=error) + pal24_lock_crud = import_module('app.database.crud.pal24') + locked = await pal24_lock_crud.get_pal24_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('Pal24: не удалось заблокировать платёж', payment_id=payment.id) + return False + payment = locked + if payment.transaction_id: logger.info('Pal24 платеж уже привязан к транзакции (trigger=)', bill_id=payment.bill_id, trigger=trigger) return True @@ -370,6 +377,7 @@ class Pal24PaymentMixin: external_id=str(payment_id) if payment_id else payment.bill_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) await payment_module.link_pal24_payment_to_transaction(db, payment, transaction.id) @@ -387,6 +395,19 @@ class Pal24PaymentMixin: await db.commit() + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.PAL24, + external_id=payment.bill_id, + ) + try: from app.services.referral_service import process_referral_topup diff --git a/app/services/payment/platega.py b/app/services/payment/platega.py index 0873bfef..c4d9d7c5 100644 --- a/app/services/payment/platega.py +++ b/app/services/payment/platega.py @@ -177,7 +177,10 @@ class PlategaPaymentMixin: payment=payment, **update_kwargs, ) - await self._finalize_platega_payment(db, payment, payload) + result = await self._finalize_platega_payment(db, payment, payload) + if result is None: + logger.error('Platega webhook: финализация не удалась', payment_id=payment.id) + return False return True if status_raw in self._FAILED_STATUSES: @@ -242,7 +245,9 @@ class PlategaPaymentMixin: status=remote_status, callback_payload=remote_payload, ) - await self._finalize_platega_payment(db, payment, remote_payload) + result = await self._finalize_platega_payment(db, payment, remote_payload) + if result is not None: + payment = result return { 'payment': payment, @@ -259,10 +264,6 @@ class PlategaPaymentMixin: ) -> Any: payment_module = import_module('app.services.payment_service') - metadata = dict(getattr(payment, 'metadata_json', {}) or {}) - if payload is not None: - metadata['webhook'] = payload - paid_at = None if isinstance(payload, dict): paid_at_raw = payload.get('paidAt') or payload.get('confirmedAt') @@ -273,21 +274,38 @@ class PlategaPaymentMixin: except ValueError: paid_at = None - payment = await payment_module.update_platega_payment( - db, - payment=payment, - status='CONFIRMED', - is_paid=True, - paid_at=paid_at, - metadata=metadata, - callback_payload=payload, - ) + # Lock FIRST, then read fresh state + platega_lock_crud = import_module('app.database.crud.platega') + locked = await platega_lock_crud.get_platega_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('Platega: не удалось заблокировать платёж', payment_id=payment.id) + return None + payment = locked - locked_payment = await payment_module.get_platega_payment_by_id_for_update(db, payment.id) - if locked_payment: - payment = locked_payment + if payment.transaction_id: + logger.info( + 'Platega платеж уже связан с транзакцией', + correlation_id=payment.correlation_id, + transaction_id=payment.transaction_id, + ) + return payment + # Read fresh metadata AFTER lock to avoid stale data metadata = dict(getattr(payment, 'metadata_json', {}) or {}) + if payload is not None: + metadata['webhook'] = payload + + # Inline field assignments instead of update_platega_payment() which commits + # and would release the FOR UPDATE lock prematurely + payment.status = 'CONFIRMED' + payment.is_paid = True + if paid_at is not None: + payment.paid_at = paid_at + payment.metadata_json = metadata + if payload is not None: + payment.callback_payload = payload + payment.updated_at = datetime.now(UTC) + balance_already_credited = bool(metadata.get('balance_credited')) invoice_message = metadata.get('invoice_message') or {} @@ -302,14 +320,6 @@ class PlategaPaymentMixin: else: metadata.pop('invoice_message', None) - if payment.transaction_id: - logger.info( - 'Platega платеж уже связан с транзакцией', - correlation_id=payment.correlation_id, - transaction_id=payment.transaction_id, - ) - return payment - user = await payment_module.get_user_by_id(db, payment.user_id) if not user: logger.error('Пользователь не найден для Platega', user_id=payment.user_id) @@ -361,6 +371,7 @@ class PlategaPaymentMixin: external_id=transaction_external_id or payment.correlation_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) created_transaction = True @@ -379,6 +390,20 @@ class PlategaPaymentMixin: user.updated_at = datetime.now(UTC) await db.commit() await db.refresh(user) + + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.PLATEGA, + external_id=transaction_external_id or payment.correlation_id, + ) + topup_status = '🆕 Первое пополнение' if was_first_topup else '🔄 Пополнение' try: diff --git a/app/services/payment/wata.py b/app/services/payment/wata.py index 3b0d20ec..1b4c5f80 100644 --- a/app/services/payment/wata.py +++ b/app/services/payment/wata.py @@ -423,6 +423,13 @@ class WataPaymentMixin: metadata=existing_metadata, ) + wata_lock_crud = import_module('app.database.crud.wata') + locked = await wata_lock_crud.get_wata_payment_by_id_for_update(db, payment.id) + if not locked: + logger.error('WATA: не удалось заблокировать платёж', payment_id=payment.id) + return None + payment = locked + if payment.transaction_id: logger.info( 'WATA платеж уже привязан к транзакции', @@ -449,6 +456,7 @@ class WataPaymentMixin: external_id=transaction_external_id or payment.payment_link_id, is_completed=True, created_at=getattr(payment, 'created_at', None), + commit=False, ) await payment_module.link_wata_payment_to_transaction(db, payment, transaction.id) @@ -459,6 +467,20 @@ class WataPaymentMixin: user.balance_kopeks += payment.amount_kopeks user.updated_at = datetime.now(UTC) await db.commit() + + # Emit deferred side-effects after atomic commit + from app.database.crud.transaction import emit_transaction_side_effects + + await emit_transaction_side_effects( + db, + transaction, + amount_kopeks=payment.amount_kopeks, + user_id=payment.user_id, + type=TransactionType.DEPOSIT, + payment_method=PaymentMethod.WATA, + external_id=transaction_external_id or payment.payment_link_id, + ) + user = await payment_module.get_user_by_id(db, user.id) if not user: logger.error('Пользователь не найден после коммита WATA', user_id=payment.user_id) diff --git a/app/services/payment/yookassa.py b/app/services/payment/yookassa.py index b114c535..c2963622 100644 --- a/app/services/payment/yookassa.py +++ b/app/services/payment/yookassa.py @@ -450,6 +450,7 @@ class YooKassaPaymentMixin: exc_info=True, ) + await db.commit() return True payment_metadata: dict[str, Any] = {} @@ -560,6 +561,7 @@ class YooKassaPaymentMixin: exc_info=True, ) + await db.commit() return True payment_description = getattr(payment, 'description', 'YooKassa платеж') diff --git a/app/services/payment_verification_service.py b/app/services/payment_verification_service.py index 580d219d..8ec409de 100644 --- a/app/services/payment_verification_service.py +++ b/app/services/payment_verification_service.py @@ -83,6 +83,7 @@ SUPPORTED_AUTO_CHECK_METHODS: frozenset[PaymentMethod] = frozenset( PaymentMethod.PAL24, PaymentMethod.CRYPTOBOT, PaymentMethod.PLATEGA, + PaymentMethod.HELEKET, # CloudPayments removed - API returns "Completed" during authorization # before final result, causing premature balance credits. Webhooks work correctly. # WATA removed - API returns 429 "Use webhook – polling is rate-limited". diff --git a/app/utils/payment_utils.py b/app/utils/payment_utils.py index bda21adc..94e1b2f2 100644 --- a/app/utils/payment_utils.py +++ b/app/utils/payment_utils.py @@ -2,6 +2,15 @@ from app.config import settings from app.localization.texts import get_texts +def verify_payment_amount( + received_kopeks: int, + expected_kopeks: int, + tolerance_kopeks: int = 1, +) -> bool: + """Check that the received amount matches the expected amount within tolerance.""" + return abs(received_kopeks - expected_kopeks) <= tolerance_kopeks + + def get_available_payment_methods() -> list[dict[str, str]]: """ Возвращает список доступных способов оплаты с их настройками diff --git a/app/webserver/payments.py b/app/webserver/payments.py index 8f0a7ee1..c758ea94 100644 --- a/app/webserver/payments.py +++ b/app/webserver/payments.py @@ -392,21 +392,13 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute if client_ip is None: return JSONResponse( - { - 'status': 'error', - 'reason': 'unknown_ip', - 'candidates': header_ip_candidates + ([remote_ip] if remote_ip else []), - }, + {'status': 'error', 'reason': 'unknown_ip'}, status_code=status.HTTP_403_FORBIDDEN, ) if not yookassa_webhook_module.is_yookassa_ip_allowed(client_ip): return JSONResponse( - { - 'status': 'error', - 'reason': 'forbidden_ip', - 'ip': str(client_ip), - }, + {'status': 'error', 'reason': 'forbidden_ip'}, status_code=status.HTTP_403_FORBIDDEN, ) diff --git a/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py b/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py new file mode 100644 index 00000000..7e61b708 --- /dev/null +++ b/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py @@ -0,0 +1,46 @@ +"""add unique constraint on transactions(external_id, payment_method) + +Revision ID: 0017 +Revises: 0016 +Create Date: 2026-03-06 + +Prevents duplicate transaction records for the same payment provider +external ID, which could cause double-crediting of user balance. +NULL external_id values do not violate the constraint in PostgreSQL. +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = '0017' +down_revision: Union[str, None] = '0016' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Deduplicate any existing rows with same (external_id, payment_method) + # where external_id is not NULL. Keep the row with the lowest id, + # suffix duplicates with _dup_{id} to preserve audit trail. + op.execute(""" + UPDATE transactions + SET external_id = external_id || '_dup_' || id::text + WHERE external_id IS NOT NULL + AND id NOT IN ( + SELECT MIN(id) + FROM transactions + WHERE external_id IS NOT NULL + GROUP BY external_id, payment_method + ) + """) + + op.create_unique_constraint( + 'uq_transaction_external_id_method', + 'transactions', + ['external_id', 'payment_method'], + ) + + +def downgrade() -> None: + op.drop_constraint('uq_transaction_external_id_method', 'transactions', type_='unique')