fix: устранение race conditions и атомарность платёжной системы

- SELECT FOR UPDATE блокировка во всех 9 провайдерах (кроме YooKassa — свой паттерн)
- create_transaction(commit=False) + единый db.commit() для атомарности
- emit_transaction_side_effects() для отложенных событий после коммита
- Все link_*_payment_to_transaction используют db.flush() вместо db.commit()
- Freekassa/KassaAI: прямое присвоение transaction_id + flush вместо update_status
- MulenPay: прямая мутация balance_kopeks вместо add_user_balance
- Platega: блокировка перед чтением metadata, инлайн обновления полей
- CloudPayments: int(round(amount * 100)) для корректного округления
- Heleket добавлен в SUPPORTED_AUTO_CHECK_METHODS
- Удалены PII из логов yookassa webhook (заголовки, IP)
- UniqueConstraint(external_id, payment_method) на транзакциях + миграция 0017
- Cabinet: PaymentService(bot=bot) внутри try блока
- verify_payment_amount утилита для проверки суммы webhook

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(

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(

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,

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)

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,

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)

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,

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

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

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

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))

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(

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(

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)

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)

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:

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)

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',

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
 

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

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
 

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

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:

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
 

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:

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)

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 платеж')

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".

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]]:
     """
     Возвращает список доступных способов оплаты с их настройками

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,
                 )
 

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')