Merge pull request #2773 from BEDOLAGA-DEV/dev

Dev

app/cabinet/routes/__init__.py

@@ -20,6 +20,7 @@ from .admin_pinned_messages import router as admin_pinned_messages_router
 from .admin_policies import router as admin_policies_router
 from .admin_promo_offers import router as admin_promo_offers_router
 from .admin_promocodes import promo_groups_router as admin_promo_groups_router, router as admin_promocodes_router
+from .admin_referral_network import router as admin_referral_network_router
 from .admin_remnawave import router as admin_remnawave_router
 from .admin_roles import router as admin_roles_router
 from .admin_sales_stats import router as admin_sales_stats_router
@@ -99,6 +100,7 @@ router.include_router(admin_wheel_router)
 router.include_router(admin_tariffs_router)
 router.include_router(admin_servers_router)
 router.include_router(admin_stats_router)
+router.include_router(admin_referral_network_router)
 router.include_router(admin_sales_stats_router)
 router.include_router(admin_ban_system_router)
 router.include_router(admin_broadcasts_router)

app/cabinet/routes/admin_tickets.py

@@ -5,7 +5,7 @@ from datetime import UTC, datetime
 
 import structlog
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 from sqlalchemy import desc, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
@@ -90,6 +90,19 @@ class AdminReplyRequest(BaseModel):
     """Admin reply to ticket."""
 
     message: str = Field(..., min_length=1, max_length=4000, description='Reply message')
+    media_type: str | None = Field(None, description='Media type: photo, video, or document')
+    media_file_id: str | None = Field(None, max_length=255, description='Telegram file_id from media upload')
+    media_caption: str | None = Field(None, max_length=1000, description='Caption for media')
+
+    @model_validator(mode='after')
+    def validate_media_fields(self) -> 'AdminReplyRequest':
+        if self.media_file_id and not self.media_type:
+            raise ValueError('media_type is required when media_file_id is provided')
+        if self.media_type and not self.media_file_id:
+            raise ValueError('media_file_id is required when media_type is provided')
+        if self.media_type and self.media_type not in {'photo', 'video', 'document'}:
+            raise ValueError('media_type must be one of: photo, video, document')
+        return self
 
 
 class AdminStatusUpdateRequest(BaseModel):
@@ -443,11 +456,16 @@ async def reply_to_ticket(
         )
 
     # Create admin message
+    has_media = bool(request.media_file_id)
     message = TicketMessage(
         ticket_id=ticket.id,
         user_id=ticket.user_id,
         message_text=request.message,
         is_from_admin=True,
+        has_media=has_media,
+        media_type=request.media_type if has_media else None,
+        media_file_id=request.media_file_id if has_media else None,
+        media_caption=request.media_caption if has_media else None,
         created_at=datetime.now(UTC),
     )
     db.add(message)

app/database/crud/transaction.py

@@ -108,7 +108,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)
+            logger.warning('Не удалось проверить автовыдачу промогруппы для пользователя', 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
@@ -168,7 +168,7 @@ async def emit_transaction_side_effects(
 
         await maybe_assign_promo_group_by_total_spent(db, user_id)
     except Exception as exc:
-        logger.debug('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
+        logger.warning('Не удалось проверить автовыдачу промогруппы для пользователя', user_id=user_id, exc=exc)
 
     if type == TransactionType.SUBSCRIPTION_PAYMENT and is_completed:
         try:
@@ -253,7 +253,7 @@ async def complete_transaction(db: AsyncSession, transaction: Transaction) -> Tr
 
         await maybe_assign_promo_group_by_total_spent(db, transaction.user_id)
     except Exception as exc:
-        logger.debug(
+        logger.warning(
             'Не удалось проверить автовыдачу промогруппы для пользователя', user_id=transaction.user_id, exc=exc
         )
 

app/database/crud/user_promo_group.py

@@ -3,7 +3,7 @@
 from datetime import UTC, datetime
 
 import structlog
-from sqlalchemy import and_, desc, select
+from sqlalchemy import and_, desc, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -24,7 +24,7 @@ async def _sync_user_primary_promo_group(
             select(UserPromoGroup.promo_group_id)
             .join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
             .where(UserPromoGroup.user_id == user_id)
-            .order_by(desc(PromoGroup.priority), PromoGroup.id)
+            .order_by(desc(PromoGroup.priority), desc(PromoGroup.id))
         )
 
         first = result.first()
@@ -53,7 +53,12 @@ async def sync_user_primary_promo_group(
 
 
 async def add_user_to_promo_group(
-    db: AsyncSession, user_id: int, promo_group_id: int, assigned_by: str = 'admin'
+    db: AsyncSession,
+    user_id: int,
+    promo_group_id: int,
+    assigned_by: str = 'admin',
+    *,
+    commit: bool = True,
 ) -> UserPromoGroup | None:
     """
     Добавляет пользователю промогруппу.
@@ -63,6 +68,7 @@ async def add_user_to_promo_group(
         user_id: ID пользователя
         promo_group_id: ID промогруппы
         assigned_by: Кто назначил ('admin', 'system', 'auto', 'promocode')
+        commit: Коммитить транзакцию (False для батчевых операций)
 
     Returns:
         UserPromoGroup или None если уже существует
@@ -85,8 +91,9 @@ async def add_user_to_promo_group(
 
         await _sync_user_primary_promo_group(db, user_id)
 
-        await db.commit()
-        await db.refresh(user_promo_group)
+        if commit:
+            await db.commit()
+            await db.refresh(user_promo_group)
 
         logger.info(
             'Пользователю добавлена промогруппа',
@@ -98,11 +105,19 @@ async def add_user_to_promo_group(
 
     except Exception as error:
         logger.error('Ошибка добавления промогруппы пользователю', error=error)
-        await db.rollback()
-        return None
+        if commit:
+            await db.rollback()
+            return None
+        raise
 
 
-async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_group_id: int) -> bool:
+async def remove_user_from_promo_group(
+    db: AsyncSession,
+    user_id: int,
+    promo_group_id: int,
+    *,
+    commit: bool = True,
+) -> bool:
     """
     Удаляет промогруппу у пользователя.
 
@@ -110,6 +125,7 @@ async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_gro
         db: Сессия БД
         user_id: ID пользователя
         promo_group_id: ID промогруппы
+        commit: Коммитить транзакцию (False для батчевых операций)
 
     Returns:
         True если удалено, False если связи не было
@@ -133,15 +149,18 @@ async def remove_user_from_promo_group(db: AsyncSession, user_id: int, promo_gro
 
         await _sync_user_primary_promo_group(db, user_id)
 
-        await db.commit()
+        if commit:
+            await db.commit()
 
         logger.info('У пользователя удалена промогруппа', user_id=user_id, promo_group_id=promo_group_id)
         return True
 
     except Exception as error:
         logger.error('Ошибка удаления промогруппы у пользователя', error=error)
-        await db.rollback()
-        return False
+        if commit:
+            await db.rollback()
+            return False
+        raise
 
 
 async def get_user_promo_groups(db: AsyncSession, user_id: int) -> list[UserPromoGroup]:
@@ -155,19 +174,14 @@ async def get_user_promo_groups(db: AsyncSession, user_id: int) -> list[UserProm
     Returns:
         Список UserPromoGroup с загруженными PromoGroup, отсортированный по приоритету DESC
     """
-    try:
-        result = await db.execute(
-            select(UserPromoGroup)
-            .options(selectinload(UserPromoGroup.promo_group))
-            .where(UserPromoGroup.user_id == user_id)
-            .join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
-            .order_by(desc(PromoGroup.priority), PromoGroup.id)
-        )
-        return list(result.scalars().all())
-
-    except Exception as error:
-        logger.error('Ошибка получения промогрупп пользователя', user_id=user_id, error=error)
-        return []
+    result = await db.execute(
+        select(UserPromoGroup)
+        .options(selectinload(UserPromoGroup.promo_group))
+        .where(UserPromoGroup.user_id == user_id)
+        .join(PromoGroup, UserPromoGroup.promo_group_id == PromoGroup.id)
+        .order_by(desc(PromoGroup.priority), desc(PromoGroup.id))
+    )
+    return list(result.scalars().all())
 
 
 async def get_primary_user_promo_group(db: AsyncSession, user_id: int) -> PromoGroup | None:
@@ -181,19 +195,14 @@ async def get_primary_user_promo_group(db: AsyncSession, user_id: int) -> PromoG
     Returns:
         PromoGroup с максимальным приоритетом или None
     """
-    try:
-        user_promo_groups = await get_user_promo_groups(db, user_id)
+    user_promo_groups = await get_user_promo_groups(db, user_id)
 
-        if not user_promo_groups:
-            return None
-
-        # Первая в списке имеет максимальный приоритет (список уже отсортирован)
-        return user_promo_groups[0].promo_group or None
-
-    except Exception as error:
-        logger.error('Ошибка получения primary промогруппы пользователя', user_id=user_id, error=error)
+    if not user_promo_groups:
         return None
 
+    # Первая в списке имеет максимальный приоритет (список уже отсортирован)
+    return user_promo_groups[0].promo_group or None
+
 
 async def has_user_promo_group(db: AsyncSession, user_id: int, promo_group_id: int) -> bool:
     """
@@ -207,17 +216,12 @@ async def has_user_promo_group(db: AsyncSession, user_id: int, promo_group_id: i
     Returns:
         True если пользователь уже имеет эту промогруппу
     """
-    try:
-        result = await db.execute(
-            select(UserPromoGroup).where(
-                and_(UserPromoGroup.user_id == user_id, UserPromoGroup.promo_group_id == promo_group_id)
-            )
+    result = await db.execute(
+        select(UserPromoGroup).where(
+            and_(UserPromoGroup.user_id == user_id, UserPromoGroup.promo_group_id == promo_group_id)
         )
-        return result.scalar_one_or_none() is not None
-
-    except Exception as error:
-        logger.error('Ошибка проверки промогруппы пользователя', error=error)
-        return False
+    )
+    return result.scalar_one_or_none() is not None
 
 
 async def count_user_promo_groups(db: AsyncSession, user_id: int) -> int:
@@ -232,8 +236,10 @@ async def count_user_promo_groups(db: AsyncSession, user_id: int) -> int:
         Количество промогрупп
     """
     try:
-        result = await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
-        return len(list(result.scalars().all()))
+        result = await db.execute(
+            select(func.count()).select_from(UserPromoGroup).where(UserPromoGroup.user_id == user_id)
+        )
+        return result.scalar_one()
 
     except Exception as error:
         logger.error('Ошибка подсчета промогрупп пользователя', error=error)
@@ -257,15 +263,18 @@ async def replace_user_promo_groups(
     """
     try:
         # Удаляем все текущие промогруппы
-        await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
         result = await db.execute(select(UserPromoGroup).where(UserPromoGroup.user_id == user_id))
         for upg in result.scalars().all():
             await db.delete(upg)
+        await db.flush()
 
         # Добавляем новые
         for promo_group_id in promo_group_ids:
             user_promo_group = UserPromoGroup(user_id=user_id, promo_group_id=promo_group_id, assigned_by=assigned_by)
             db.add(user_promo_group)
+        await db.flush()
+
+        await _sync_user_primary_promo_group(db, user_id)
 
         await db.commit()
         logger.info('Промогруппы пользователя заменены на', user_id=user_id, promo_group_ids=promo_group_ids)

app/database/models.py

@@ -1574,6 +1574,7 @@ class Transaction(Base):
         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'),
+        Index('ix_transactions_user_type_completed_amount', 'user_id', 'type', 'is_completed', 'amount_kopeks'),
     )
 
     id = Column(Integer, primary_key=True, index=True)
@@ -2490,7 +2491,10 @@ class AdvertisingCampaign(Base):
 
 class AdvertisingCampaignRegistration(Base):
     __tablename__ = 'advertising_campaign_registrations'
-    __table_args__ = (UniqueConstraint('campaign_id', 'user_id', name='uq_campaign_user'),)
+    __table_args__ = (
+        UniqueConstraint('campaign_id', 'user_id', name='uq_campaign_user'),
+        Index('ix_campaign_reg_user_created', 'user_id', 'created_at'),
+    )
 
     id = Column(Integer, primary_key=True, index=True)
     campaign_id = Column(Integer, ForeignKey('advertising_campaigns.id', ondelete='CASCADE'), nullable=False)
@@ -3283,6 +3287,7 @@ class GuestPurchase(Base):
     cabinet_password = Column(Text, nullable=True)
     auto_login_token = Column(Text, nullable=True)
     recipient_warning = Column(String(50), nullable=True)
+    retry_count = Column(Integer, nullable=False, default=0, server_default='0')
 
     landing = relationship('LandingPage', back_populates='guest_purchases', lazy='selectin')
     tariff = relationship('Tariff', lazy='selectin')

app/handlers/stars_payments.py

@@ -354,10 +354,6 @@ async def _handle_guest_purchase_payment(
                 stars_amount=stars_amount,
                 purchase_token_prefix=purchase_token[:5],
             )
-        elif result is False:
-            await message.answer(
-                '❌ Произошла ошибка при обработке подарочной подписки. Обратитесь в поддержку.',
-            )
         else:
             logger.error('try_fulfill_guest_purchase returned None for Stars gift', payload=payload)
             await message.answer('❌ Ошибка обработки платежа. Обратитесь в поддержку.')

app/services/guest_purchase_service.py

@@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta
 from typing import Literal
 
 import structlog
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -392,24 +392,41 @@ async def fulfill_purchase(
     return purchase
 
 
+def _resolve_base_payment_method(method_str: str | None) -> str:
+    """Resolve base payment method string by stripping sub-option suffixes.
+
+    'yookassa_sbp' → 'yookassa', 'kassa_ai' → 'kassa_ai' (enum match keeps it),
+    'platega_2' → 'platega'.
+    """
+    if not method_str:
+        return ''
+    # If exact enum match, return as-is (handles 'telegram_stars', 'kassa_ai', etc.)
+    try:
+        PaymentMethod(method_str)
+        return method_str
+    except ValueError:
+        pass
+    # Strip sub-option suffix
+    if '_' in method_str:
+        base = method_str.rsplit('_', 1)[0]
+        try:
+            PaymentMethod(base)
+            return base
+        except ValueError:
+            pass
+    return method_str
+
+
 def _resolve_payment_method(method_str: str | None) -> PaymentMethod | None:
     """Convert payment method string from GuestPurchase to PaymentMethod enum."""
     if not method_str:
         return None
-    # Try exact match first (handles 'telegram_stars', 'kassa_ai', 'yookassa', etc.)
+    base = _resolve_base_payment_method(method_str)
     try:
-        return PaymentMethod(method_str)
+        return PaymentMethod(base)
     except ValueError:
-        pass
-    # Strip sub-option suffix ('yookassa_sbp' → 'yookassa', 'platega_2' → 'platega')
-    if '_' in method_str:
-        base_method = method_str.split('_')[0]
-        try:
-            return PaymentMethod(base_method)
-        except ValueError:
-            pass
-    logger.debug('Unknown payment method for transaction', method=method_str)
-    return None
+        logger.debug('Unknown payment method for transaction', method=method_str)
+        return None
 
 
 def _mask_email(email: str) -> str:
@@ -1002,14 +1019,16 @@ async def retry_stuck_paid_purchases(
     stale_minutes: int = 5,
     limit: int = 10,
     max_age_hours: int = 24,
+    max_retries: int = 20,
 ) -> int:
     """Retry fulfillment for purchases stuck in PAID status.
 
     Finds purchases that have been in PAID status for longer than stale_minutes
-    (but not older than max_age_hours) and attempts to fulfill them in isolated
-    sessions. Returns the number of successfully retried purchases.
+    (but not older than max_age_hours, and with retry_count < max_retries) and
+    attempts to fulfill them in isolated sessions.
 
-    Purchases older than max_age_hours are left for manual investigation.
+    Purchases exceeding max_retries are marked FAILED and an admin alert is sent.
+    Returns the number of successfully retried purchases.
     """
     from app.database.database import AsyncSessionLocal
 
@@ -1018,10 +1037,12 @@ async def retry_stuck_paid_purchases(
 
     # Collect tokens only — each retry gets its own session.
     # NULL paid_at is included via or_() as a safety net for data anomalies.
+    # Filter retry_count < max_retries in SQL to avoid wasting LIMIT slots.
     result = await db.execute(
         select(GuestPurchase.token)
         .where(
             GuestPurchase.status == GuestPurchaseStatus.PAID.value,
+            GuestPurchase.retry_count < max_retries,
             or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)),
             or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
             # Exclude code-only gifts — they stay PAID intentionally until activated
@@ -1032,6 +1053,9 @@ async def retry_stuck_paid_purchases(
     )
     tokens = result.scalars().all()
 
+    # Separately fail exhausted purchases (retry_count >= max_retries)
+    await _fail_exhausted_purchases_batch(db, GuestPurchaseStatus.PAID, max_retries, max_age)
+
     if not tokens:
         return 0
 
@@ -1039,6 +1063,7 @@ async def retry_stuck_paid_purchases(
     for token in tokens:
         try:
             async with AsyncSessionLocal() as retry_db:
+                await _increment_retry_count(retry_db, token)
                 await fulfill_purchase(retry_db, token)
                 retried += 1
                 logger.info('Retried stuck purchase successfully', token_prefix=token[:5])
@@ -1053,12 +1078,15 @@ async def retry_stuck_pending_activation(
     stale_minutes: int = 10,
     limit: int = 10,
     max_age_hours: int = 24,
+    max_retries: int = 20,
 ) -> int:
     """Retry activation for purchases stuck in PENDING_ACTIVATION status.
 
     This handles the case where activate_purchase() failed after the status
     was already transitioned to PENDING_ACTIVATION (e.g., Remnawave panel was
     temporarily down). Each retry runs in an isolated session.
+
+    Purchases exceeding max_retries are marked FAILED and an admin alert is sent.
     """
     from app.database.database import AsyncSessionLocal
 
@@ -1069,6 +1097,7 @@ async def retry_stuck_pending_activation(
         select(GuestPurchase.token)
         .where(
             GuestPurchase.status == GuestPurchaseStatus.PENDING_ACTIVATION.value,
+            GuestPurchase.retry_count < max_retries,
             or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)),
             or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
             GuestPurchase.user_id.isnot(None),
@@ -1078,6 +1107,9 @@ async def retry_stuck_pending_activation(
     )
     tokens = result.scalars().all()
 
+    # Separately fail exhausted purchases (retry_count >= max_retries)
+    await _fail_exhausted_purchases_batch(db, GuestPurchaseStatus.PENDING_ACTIVATION, max_retries, max_age)
+
     if not tokens:
         return 0
 
@@ -1085,6 +1117,7 @@ async def retry_stuck_pending_activation(
     for token in tokens:
         try:
             async with AsyncSessionLocal() as retry_db:
+                await _increment_retry_count(retry_db, token)
                 await activate_purchase(retry_db, token)
                 retried += 1
                 logger.info('Retried stuck pending_activation successfully', token_prefix=token[:5])
@@ -1092,3 +1125,369 @@ async def retry_stuck_pending_activation(
             logger.exception('Failed to retry stuck pending_activation', token_prefix=token[:5])
 
     return retried
+
+
+async def _increment_retry_count(db: AsyncSession, purchase_token: str) -> None:
+    """Atomically increment retry_count via UPDATE statement (no SELECT, no identity map pollution)."""
+    await db.execute(
+        update(GuestPurchase)
+        .where(GuestPurchase.token == purchase_token)
+        .values(retry_count=GuestPurchase.retry_count + 1)
+    )
+    await db.commit()
+
+
+async def _fail_exhausted_purchases_batch(
+    db: AsyncSession,
+    status: GuestPurchaseStatus,
+    max_retries: int,
+    max_age: datetime,
+) -> None:
+    """Find and mark exhausted purchases as FAILED, then send admin alerts."""
+    from app.database.crud.landing import update_purchase_status
+    from app.database.database import AsyncSessionLocal
+
+    result = await db.execute(
+        select(GuestPurchase.token, GuestPurchase.retry_count)
+        .where(
+            GuestPurchase.status == status.value,
+            GuestPurchase.retry_count >= max_retries,
+            or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
+        )
+        .limit(10)
+    )
+    exhausted = result.all()
+
+    for token, retry_count in exhausted:
+        # Collect alert data before closing the session
+        alert_data: dict | None = None
+        try:
+            async with AsyncSessionLocal() as fail_db:
+                row = await fail_db.execute(select(GuestPurchase).where(GuestPurchase.token == token).with_for_update())
+                purchase = row.scalars().first()
+                if purchase and purchase.status not in (
+                    GuestPurchaseStatus.DELIVERED.value,
+                    GuestPurchaseStatus.FAILED.value,
+                ):
+                    # Capture alert data before commit expires attributes
+                    alert_data = {
+                        'id': purchase.id,
+                        'token': purchase.token,
+                        'amount_kopeks': purchase.amount_kopeks,
+                        'payment_method': purchase.payment_method,
+                        'payment_id': purchase.payment_id,
+                        'contact_type': purchase.contact_type,
+                        'contact_value': purchase.contact_value,
+                        'created_at': purchase.created_at,
+                    }
+                    await update_purchase_status(fail_db, token, GuestPurchaseStatus.FAILED)
+                    logger.error(
+                        'Purchase exceeded max retries — marked FAILED',
+                        token_prefix=token[:5],
+                        retry_count=retry_count,
+                        phase=status.value,
+                    )
+        except Exception:
+            logger.exception('Failed to mark exhausted purchase as FAILED', token_prefix=token[:5])
+
+        # Send alert OUTSIDE the session (no row lock held)
+        if alert_data:
+            await _send_stuck_purchase_alert(alert_data, retry_count, status.value)
+
+
+async def _send_stuck_purchase_alert(data: dict, retry_count: int, phase: str) -> None:
+    """Send admin notification about a purchase that exhausted all retries.
+
+    Accepts a plain dict (not ORM object) so it can be called after the session is closed.
+    """
+    if not getattr(settings, 'ADMIN_NOTIFICATIONS_ENABLED', False) or not settings.BOT_TOKEN:
+        return
+    try:
+        import html as html_mod
+
+        from aiogram import Bot
+
+        from app.services.admin_notification_service import AdminNotificationService, NotificationCategory
+
+        amount_rub = data['amount_kopeks'] / 100
+        contact_value = html_mod.escape(str(data.get('contact_value', '?')))
+        contact_type = html_mod.escape(str(data.get('contact_type', '?')))
+        text = (
+            f'<b>STUCK PURCHASE — retries exhausted</b>\n\n'
+            f'Token: <code>{data["token"][:8]}...</code>\n'
+            f'Status: <code>{phase}</code> → <code>FAILED</code>\n'
+            f'Retries: <b>{retry_count}</b>\n'
+            f'Amount: <b>{amount_rub:.0f} ₽</b>\n'
+            f'Payment: <code>{html_mod.escape(str(data.get("payment_method") or "?"))}</code>\n'
+            f'Payment ID: <code>{html_mod.escape(str(data.get("payment_id") or "?"))}</code>\n'
+            f'Contact: {contact_type}: <code>{contact_value}</code>\n'
+            f'Created: {data["created_at"]:%Y-%m-%d %H:%M UTC}\n\n'
+            f'Requires manual investigation.'
+        )
+
+        async with Bot(token=settings.BOT_TOKEN) as bot:
+            service = AdminNotificationService(bot)
+            await service.send_admin_notification(text, category=NotificationCategory.ERRORS)
+    except Exception:
+        logger.warning('Failed to send stuck purchase admin alert', purchase_id=data.get('id'), exc_info=True)
+
+
+async def _send_amount_mismatch_alert(
+    purchase: GuestPurchase,
+    provider_amount_kopeks: int,
+    provider_payment_id: str,
+    payment_method: str | None,
+) -> None:
+    """Send admin alert when recovery detects an amount mismatch (possible fraud or bug)."""
+    if not getattr(settings, 'ADMIN_NOTIFICATIONS_ENABLED', False) or not settings.BOT_TOKEN:
+        return
+    try:
+        import html as html_mod
+
+        from aiogram import Bot
+
+        from app.services.admin_notification_service import AdminNotificationService, NotificationCategory
+
+        text = (
+            f'<b>AMOUNT MISMATCH — purchase marked FAILED</b>\n\n'
+            f'Token: <code>{purchase.token[:8]}...</code>\n'
+            f'Expected: <b>{purchase.amount_kopeks / 100:.0f} ₽</b>\n'
+            f'Provider: <b>{provider_amount_kopeks / 100:.0f} ₽</b>\n'
+            f'Payment: <code>{html_mod.escape(str(payment_method or "?"))}</code>\n'
+            f'Payment ID: <code>{html_mod.escape(str(provider_payment_id))}</code>\n'
+            f'Contact: {html_mod.escape(str(purchase.contact_type))}: '
+            f'<code>{html_mod.escape(str(purchase.contact_value))}</code>\n\n'
+            f'Requires manual investigation.'
+        )
+
+        async with Bot(token=settings.BOT_TOKEN) as bot:
+            service = AdminNotificationService(bot)
+            await service.send_admin_notification(text, category=NotificationCategory.ERRORS)
+    except Exception:
+        logger.warning('Failed to send amount mismatch alert', purchase_id=purchase.id, exc_info=True)
+
+
+async def recover_stuck_pending_purchases(
+    db: AsyncSession,
+    stale_minutes: int = 10,
+    limit: int = 10,
+    max_age_hours: int = 24,
+) -> int:
+    """Recover purchases stuck in PENDING by checking provider payment status.
+
+    Queries all payment provider tables (YooKassa, Heleket, CryptoBot, etc.)
+    for succeeded payments matching the purchase_token. If a provider payment
+    is confirmed but the GuestPurchase is still PENDING (webhook was lost or
+    processing failed), marks the purchase as PAID so retry_stuck_paid_purchases
+    can fulfill it. Includes amount verification.
+
+    Returns the number of recovered purchases.
+    """
+    from app.database.database import AsyncSessionLocal
+
+    cutoff = datetime.now(UTC) - timedelta(minutes=stale_minutes)
+    max_age = datetime.now(UTC) - timedelta(hours=max_age_hours)
+
+    # Find PENDING purchases older than stale_minutes but younger than max_age_hours
+    result = await db.execute(
+        select(GuestPurchase.token, GuestPurchase.payment_method)
+        .where(
+            GuestPurchase.status == GuestPurchaseStatus.PENDING.value,
+            GuestPurchase.created_at < cutoff,
+            GuestPurchase.created_at > max_age,
+        )
+        .order_by(GuestPurchase.created_at.asc())
+        .limit(limit)
+    )
+    pending_purchases = result.all()
+
+    if not pending_purchases:
+        return 0
+
+    recovered = 0
+    for token, payment_method in pending_purchases:
+        try:
+            async with AsyncSessionLocal() as recover_db:
+                paid = await _check_and_recover_pending_purchase(recover_db, token, payment_method)
+                if paid:
+                    recovered += 1
+        except Exception:
+            logger.exception('Failed to recover pending purchase', token_prefix=token[:5])
+
+    return recovered
+
+
+async def _find_succeeded_provider_payment(
+    db: AsyncSession,
+    base_method: str,
+    purchase_token: str,
+) -> tuple[str, int | None] | None:
+    """Query provider payment tables for a succeeded payment matching purchase_token.
+
+    Returns ``(provider_payment_id, amount_kopeks)`` or ``None``.
+    ``amount_kopeks`` is ``None`` when the amount check should be skipped
+    (e.g., CryptoBot where USD→RUB conversion introduces imprecision).
+    """
+    from sqlalchemy import cast
+    from sqlalchemy.types import JSON as SA_JSON
+
+    from app.database.models import (
+        CloudPaymentsPayment,
+        CryptoBotPayment,
+        FreekassaPayment,
+        HeleketPayment,
+        KassaAiPayment,
+        MulenPayPayment,
+        Pal24Payment,
+        PlategaPayment,
+        RioPayPayment,
+        SeverPayPayment,
+        WataPayment,
+        YooKassaPayment,
+    )
+
+    # --- CryptoBot: special case — payload field (text JSON), skip amount check ---
+    if base_method == 'cryptobot':
+        result = await db.execute(
+            select(CryptoBotPayment).where(
+                CryptoBotPayment.status == 'paid',
+                cast(CryptoBotPayment.payload, SA_JSON)['purchase_token'].as_string() == purchase_token,
+            )
+        )
+        p = result.scalars().first()
+        return (p.invoice_id, None) if p else None
+
+    # --- All other providers: metadata_json['purchase_token'] + is_paid/status filters ---
+    model = None
+    payment_id_attr: str = ''
+    extra_conditions: list = []
+
+    if base_method.startswith('yookassa'):
+        model = YooKassaPayment
+        payment_id_attr = 'yookassa_payment_id'
+        extra_conditions = [YooKassaPayment.status == 'succeeded', YooKassaPayment.is_paid.is_(True)]
+    elif base_method == 'heleket':
+        model = HeleketPayment
+        payment_id_attr = 'uuid'
+        extra_conditions = [HeleketPayment.status.in_(['paid', 'paid_over'])]
+    elif base_method == 'mulenpay':
+        model = MulenPayPayment
+        payment_id_attr = 'uuid'
+        extra_conditions = [MulenPayPayment.is_paid.is_(True)]
+    elif base_method == 'pal24':
+        model = Pal24Payment
+        payment_id_attr = 'bill_id'
+        extra_conditions = [Pal24Payment.is_paid.is_(True)]
+    elif base_method == 'wata':
+        model = WataPayment
+        payment_id_attr = 'payment_link_id'
+        extra_conditions = [WataPayment.is_paid.is_(True)]
+    elif base_method == 'platega':
+        model = PlategaPayment
+        payment_id_attr = 'platega_transaction_id'
+        extra_conditions = [PlategaPayment.is_paid.is_(True)]
+    elif base_method == 'cloudpayments':
+        model = CloudPaymentsPayment
+        payment_id_attr = 'invoice_id'
+        extra_conditions = [CloudPaymentsPayment.status == 'completed', CloudPaymentsPayment.is_paid.is_(True)]
+    elif base_method == 'freekassa':
+        model = FreekassaPayment
+        payment_id_attr = 'order_id'
+        extra_conditions = [FreekassaPayment.status == 'success', FreekassaPayment.is_paid.is_(True)]
+    elif base_method == 'kassa_ai':
+        model = KassaAiPayment
+        payment_id_attr = 'order_id'
+        extra_conditions = [KassaAiPayment.status == 'success', KassaAiPayment.is_paid.is_(True)]
+    elif base_method == 'riopay':
+        model = RioPayPayment
+        payment_id_attr = 'order_id'
+        extra_conditions = [RioPayPayment.status == 'success', RioPayPayment.is_paid.is_(True)]
+    elif base_method == 'severpay':
+        model = SeverPayPayment
+        payment_id_attr = 'order_id'
+        extra_conditions = [SeverPayPayment.status == 'success', SeverPayPayment.is_paid.is_(True)]
+
+    if model is None:
+        return None
+
+    result = await db.execute(
+        select(model).where(
+            model.metadata_json['purchase_token'].as_string() == purchase_token,
+            *extra_conditions,
+        )
+    )
+    p = result.scalars().first()
+    if p is None:
+        return None
+
+    payment_id = str(getattr(p, payment_id_attr))
+    # amount_kopeks: Integer column for most providers, @property for Heleket
+    amount = getattr(p, 'amount_kopeks', None)
+    return (payment_id, amount)
+
+
+async def _check_and_recover_pending_purchase(
+    db: AsyncSession,
+    purchase_token: str,
+    payment_method: str | None,
+) -> bool:
+    """Check if a PENDING purchase has a succeeded payment and transition to PAID.
+
+    Uses SELECT ... FOR UPDATE on the GuestPurchase row to prevent concurrent
+    webhook processing from racing with the recovery.
+    Verifies amount match between provider payment and guest purchase.
+    """
+    from app.database.crud.landing import update_purchase_status
+
+    # Lock the row to prevent TOCTOU race with concurrent webhook processing
+    result = await db.execute(select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update())
+    purchase = result.scalars().first()
+    if purchase is None or purchase.status != GuestPurchaseStatus.PENDING.value:
+        return False
+
+    # Resolve base method: 'yookassa_sbp' → 'yookassa', 'kassa_ai' stays 'kassa_ai'
+    base_method = _resolve_base_payment_method(payment_method)
+
+    match = await _find_succeeded_provider_payment(db, base_method, purchase_token)
+    if match is None:
+        if base_method:
+            logger.debug(
+                'No succeeded provider payment found for PENDING purchase',
+                token_prefix=purchase_token[:5],
+                payment_method=payment_method,
+            )
+        return False
+
+    provider_payment_id, provider_amount_kopeks = match
+
+    # Amount verification (skip when provider_amount_kopeks is None, e.g., crypto)
+    if provider_amount_kopeks is not None and provider_amount_kopeks != purchase.amount_kopeks:
+        logger.error(
+            'Amount mismatch during PENDING recovery — skipping',
+            token_prefix=purchase_token[:5],
+            provider_amount=provider_amount_kopeks,
+            purchase_amount=purchase.amount_kopeks,
+            payment_method=payment_method,
+        )
+        # Mark FAILED to prevent repeated mismatch logs every cycle
+        from app.database.crud.landing import update_purchase_status as _update_status
+
+        await _update_status(db, purchase_token, GuestPurchaseStatus.FAILED)
+        await _send_amount_mismatch_alert(purchase, provider_amount_kopeks, provider_payment_id, payment_method)
+        return False
+
+    # Transition PENDING → PAID for retry_stuck_paid_purchases to handle
+    await update_purchase_status(
+        db,
+        purchase_token,
+        GuestPurchaseStatus.PAID,
+        payment_id=provider_payment_id,
+        paid_at=datetime.now(UTC),
+    )
+    logger.info(
+        'Recovered stuck PENDING purchase → PAID',
+        token_prefix=purchase_token[:5],
+        payment_method=payment_method,
+        provider_payment_id=provider_payment_id,
+    )
+    return True

app/services/monitoring_service.py

@@ -1742,18 +1742,35 @@ class MonitoringService:
             )
 
     async def _retry_stuck_guest_purchases(self, db: AsyncSession):
-        try:
-            from app.services.guest_purchase_service import retry_stuck_paid_purchases, retry_stuck_pending_activation
+        from app.services.guest_purchase_service import (
+            recover_stuck_pending_purchases,
+            retry_stuck_paid_purchases,
+            retry_stuck_pending_activation,
+        )
 
+        # Phase 1: Recover PENDING purchases where provider payment already succeeded
+        try:
+            recovered = await recover_stuck_pending_purchases(db, stale_minutes=10, limit=10)
+            if recovered:
+                logger.info('Recovered stuck PENDING purchases', recovered=recovered)
+        except Exception:
+            logger.error('Error recovering stuck PENDING guest purchases', exc_info=True)
+
+        # Phase 2: Retry fulfillment for purchases in PAID status
+        try:
             retried = await retry_stuck_paid_purchases(db, stale_minutes=5, limit=10)
             if retried:
                 logger.info('Retried stuck guest purchases', retried=retried)
+        except Exception:
+            logger.error('Error retrying stuck PAID guest purchases', exc_info=True)
 
+        # Phase 3: Retry activation for purchases in PENDING_ACTIVATION status
+        try:
             retried_pa = await retry_stuck_pending_activation(db, stale_minutes=10, limit=10)
             if retried_pa:
                 logger.info('Retried stuck pending_activation purchases', retried=retried_pa)
         except Exception:
-            logger.error('Error retrying stuck guest purchases', exc_info=True)
+            logger.error('Error retrying stuck PENDING_ACTIVATION guest purchases', exc_info=True)
 
     async def _cleanup_inactive_users(self, db: AsyncSession):
         try:

app/services/payment/common.py

@@ -476,8 +476,7 @@ async def try_fulfill_guest_purchase(
             introduces imprecision.
 
     Returns:
-        ``True``  -- guest purchase was detected and successfully fulfilled.
-        ``False`` -- guest purchase was detected but fulfillment failed.
+        ``True``  -- guest purchase was detected and consumed (fulfilled or queued for retry).
         ``None``  -- this is NOT a guest purchase (caller should proceed normally).
     """
     purchase_token = _extract_guest_purchase_token(metadata)
@@ -485,7 +484,7 @@ async def try_fulfill_guest_purchase(
         return None
 
     from app.database.crud.landing import get_purchase_by_token, update_purchase_status
-    from app.database.models import GuestPurchaseStatus
+    from app.database.models import GuestPurchase, GuestPurchaseStatus
     from app.services.guest_purchase_service import fulfill_purchase
 
     try:
@@ -558,13 +557,26 @@ async def try_fulfill_guest_purchase(
             provider=provider_name,
             error=guest_error,
         )
-        # Mark as FAILED so it doesn't get retried forever
+        # Mark as PAID (not FAILED) so retry_stuck_paid_purchases can pick it up.
+        # Use a fresh session to avoid tainted-session issues after rollback.
+        # The monitoring service retries PAID purchases every 5 minutes for up to 24 hours.
         try:
-            await update_purchase_status(
-                db,
-                purchase_token,
-                GuestPurchaseStatus.FAILED,
-            )
+            from app.database.database import AsyncSessionLocal
+
+            async with AsyncSessionLocal() as recovery_db:
+                # Use FOR UPDATE to prevent TOCTOU race with concurrent webhook.
+                row = await recovery_db.execute(
+                    select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update()
+                )
+                current = row.scalars().first()
+                if current and current.status in (
+                    GuestPurchaseStatus.PENDING.value,
+                    GuestPurchaseStatus.PAID.value,
+                ):
+                    current.status = GuestPurchaseStatus.PAID.value
+                    current.payment_id = provider_payment_id
+                    current.paid_at = datetime.now(UTC)
+                    await recovery_db.commit()
         except Exception:
-            logger.exception('Failed to mark guest purchase as FAILED')
-        return False
+            logger.exception('Failed to mark guest purchase as PAID for retry')
+        return True

app/services/payment_service.py

@@ -721,7 +721,7 @@ class PaymentService(
                 payment_system_id=ps_id,
             )
             if result:
-                await _patch_guest_metadata(result['local_payment_id'], payment_method)
+                await _patch_guest_metadata(result['local_payment_id'], 'kassa_ai')
                 return {
                     'payment_url': result.get('payment_url'),
                     'payment_id': result.get('order_id'),

app/services/promo_group_assignment.py

@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.database.crud.transaction import get_user_total_spent_kopeks
+from app.database.crud.user import lock_user_for_update
 from app.database.models import PromoGroup, User
 from app.services.admin_notification_service import AdminNotificationService
 
@@ -90,7 +91,9 @@ async def maybe_assign_promo_group_by_total_spent(
 ) -> PromoGroup | None:
     from app.database.crud.user_promo_group import (
         add_user_to_promo_group,
+        get_user_promo_groups,
         has_user_promo_group,
+        remove_user_from_promo_group,
         sync_user_primary_promo_group,
     )
 
@@ -99,6 +102,9 @@ async def maybe_assign_promo_group_by_total_spent(
         logger.debug('Не удалось найти пользователя для автовыдачи промогруппы', user_id=user_id)
         return None
 
+    # Блокируем строку пользователя для предотвращения гонок при конкурентных вебхуках
+    user = await lock_user_for_update(db, user)
+
     # Получаем текущую primary промогруппу
     old_group = user.get_primary_promo_group()
 
@@ -108,60 +114,68 @@ async def maybe_assign_promo_group_by_total_spent(
 
     previous_threshold = user.auto_promo_group_threshold_kopeks or 0
 
-    target_group = await _get_best_group_for_spending(
-        db,
-        total_spent,
-        min_threshold_kopeks=previous_threshold,
-    )
+    # Находим группу, соответствующую текущим тратам (без порогового фильтра,
+    # чтобы промокод-группы всегда очищались при покупке)
+    target_group = await _get_best_group_for_spending(db, total_spent)
     if not target_group:
         return None
 
     try:
         target_threshold = target_group.auto_assign_total_spent_kopeks or 0
 
-        if target_threshold <= previous_threshold:
-            logger.debug(
-                "Порог промогруппы '' не превышает ранее назначенный для пользователя",
-                target_group_name=target_group.name,
-                target_threshold=target_threshold,
-                previous_threshold=previous_threshold,
-                telegram_id=user.telegram_id,
-            )
-            return None
+        # Фаза 1: Удаляем старые auto/promocode группы, отличные от целевой
+        current_groups = await get_user_promo_groups(db, user_id)
+        removed_any = False
+        for upg in current_groups:
+            if upg.promo_group_id != target_group.id and upg.assigned_by in ('auto', 'promocode'):
+                await remove_user_from_promo_group(db, user_id, upg.promo_group_id, commit=False)
+                removed_any = True
+                logger.info(
+                    'Удалена старая промогруппа перед автоназначением',
+                    telegram_id=user.telegram_id,
+                    old_group_name=upg.promo_group.name if upg.promo_group else upg.promo_group_id,
+                    old_assigned_by=upg.assigned_by,
+                )
 
-        # Проверяем, есть ли уже эта группа у пользователя
+        if removed_any:
+            await db.flush()
+            await db.refresh(user)
+
+        # Проверяем, есть ли уже целевая группа у пользователя
         already_has_group = await has_user_promo_group(db, user_id, target_group.id)
 
-        if user.auto_promo_group_assigned and already_has_group:
+        if user.auto_promo_group_assigned and already_has_group and not removed_any:
             logger.debug(
-                "Пользователь уже имеет промогруппу '', повторная выдача не требуется",
+                'Пользователь уже имеет промогруппу, повторная выдача не требуется',
                 telegram_id=user.telegram_id,
                 target_group_name=target_group.name,
             )
-            await sync_user_primary_promo_group(db, user_id)
             if target_threshold > previous_threshold:
                 user.auto_promo_group_threshold_kopeks = target_threshold
                 user.updated_at = datetime.now(UTC)
-                await db.commit()
-                await db.refresh(user)
+            await db.commit()
+            await db.refresh(user)
             return target_group
 
         user.auto_promo_group_assigned = True
-        user.auto_promo_group_threshold_kopeks = target_threshold
+        if target_threshold > previous_threshold:
+            user.auto_promo_group_threshold_kopeks = target_threshold
         user.updated_at = datetime.now(UTC)
 
+        newly_added = False
         if not already_has_group:
-            # Добавляем новую промогруппу к существующим
-            await add_user_to_promo_group(db, user_id, target_group.id, assigned_by='auto')
+            await add_user_to_promo_group(db, user_id, target_group.id, assigned_by='auto', commit=False)
+            newly_added = True
             logger.info(
-                "🤖 Пользователю добавлена промогруппа '' за траты ₽",
+                'Пользователю назначена промогруппа за траты',
                 telegram_id=user.telegram_id,
                 target_group_name=target_group.name,
                 total_spent=total_spent / 100,
             )
         else:
+            await sync_user_primary_promo_group(db, user_id)
             logger.info(
-                "🤖 Пользователь уже имеет промогруппу '', отмечаем автоприсвоение",
+                'Пользователь уже имеет промогруппу, синхронизировано',
                 telegram_id=user.telegram_id,
                 target_group_name=target_group.name,
             )
@@ -169,7 +183,7 @@ async def maybe_assign_promo_group_by_total_spent(
         await db.commit()
         await db.refresh(user)
 
-        if not already_has_group:
+        if newly_added:
             await _notify_admins_about_auto_assignment(
                 db,
                 user,

app/services/promocode_service.py

@@ -119,7 +119,7 @@ class PromoCodeService:
                         if promo_group:
                             # Add promo group to user
                             await add_user_to_promo_group(
-                                db, user_id, promocode.promo_group_id, assigned_by='promocode'
+                                db, user_id, promocode.promo_group_id, assigned_by='promocode', commit=False
                             )
 
                             logger.info(
@@ -393,7 +393,7 @@ class PromoCodeService:
 
                     has_group = await has_user_promo_group(db, user_id, promocode.promo_group_id)
                     if has_group:
-                        await remove_user_from_promo_group(db, user_id, promocode.promo_group_id)
+                        await remove_user_from_promo_group(db, user_id, promocode.promo_group_id, commit=False)
                         logger.info(
                             'Снята промогруппа ID у пользователя при деактивации промокода',
                             promo_group_id=promocode.promo_group_id,

app/utils/cache.py

@@ -400,9 +400,16 @@ return c
             return fail_closed
 
     @staticmethod
-    async def is_rate_limited(user_id: int, action: str, limit: int, window: int) -> bool:
+    async def is_rate_limited(
+        user_id: int,
+        action: str,
+        limit: int,
+        window: int,
+        *,
+        fail_closed: bool = False,
+    ) -> bool:
         key = cache_key('rate_limit', user_id, action)
-        return await RateLimitCache._atomic_rate_check(key, limit, window)
+        return await RateLimitCache._atomic_rate_check(key, limit, window, fail_closed=fail_closed)
 
     @staticmethod
     async def reset_rate_limit(user_id: int, action: str) -> bool:

migrations/alembic/env.py

@@ -45,7 +45,11 @@ def run_migrations_offline() -> None:
 
 
 def do_run_migrations(connection: Connection) -> None:
-    context.configure(connection=connection, target_metadata=target_metadata)
+    context.configure(
+        connection=connection,
+        target_metadata=target_metadata,
+        transaction_per_migration=True,
+    )
 
     with context.begin_transaction():
         context.run_migrations()

migrations/alembic/versions/0041_add_performance_indexes.py

@@ -0,0 +1,56 @@
+"""add performance indexes for referral network queries
+
+Revision ID: 0041
+Revises: 0040
+Create Date: 2026-03-20
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = '0041'
+down_revision: Union[str, None] = '0040'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # CREATE INDEX CONCURRENTLY cannot run inside a transaction.
+    # autocommit_block() temporarily disables the transaction wrapper.
+    #
+    # NOTE: If a concurrent index creation fails midway, PostgreSQL leaves behind
+    # an INVALID index. Check with:
+    #   SELECT indexrelname FROM pg_stat_user_indexes
+    #   JOIN pg_index ON pg_index.indexrelid = pg_stat_user_indexes.indexrelid
+    #   WHERE NOT pg_index.indisvalid;
+    # Then drop the invalid index and re-run the migration.
+    with op.get_context().autocommit_block():
+        # Index on advertising_campaign_registrations(user_id, created_at)
+        # Fixes sequential scan in _fetch_campaign_registrations which filters by user_id
+        # and uses ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at)
+        op.create_index(
+            'ix_campaign_reg_user_created',
+            'advertising_campaign_registrations',
+            ['user_id', 'created_at'],
+            if_not_exists=True,
+            postgresql_concurrently=True,
+        )
+
+        # Covering composite index on transactions(user_id, type, is_completed, amount_kopeks)
+        # Enables index-only scans for aggregation queries in referral network stats:
+        # _fetch_personal_spent, _fetch_branch_revenue, _fetch_campaign_stats
+        op.create_index(
+            'ix_transactions_user_type_completed_amount',
+            'transactions',
+            ['user_id', 'type', 'is_completed', 'amount_kopeks'],
+            if_not_exists=True,
+            postgresql_concurrently=True,
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute('DROP INDEX CONCURRENTLY IF EXISTS ix_transactions_user_type_completed_amount')
+        op.execute('DROP INDEX CONCURRENTLY IF EXISTS ix_campaign_reg_user_created')

migrations/alembic/versions/0042_add_retry_count_and_payment_recovery_indexes.py

@@ -0,0 +1,88 @@
+"""add retry_count to guest_purchases and expression indexes for payment recovery
+
+Revision ID: 0042
+Revises: 0041
+Create Date: 2026-03-20
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = '0042'
+down_revision: Union[str, None] = '0041'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# Payment tables with metadata_json + is_paid column.
+_TABLES_WITH_IS_PAID = [
+    'yookassa_payments',
+    'mulenpay_payments',
+    'pal24_payments',
+    'wata_payments',
+    'platega_payments',
+    'cloudpayments_payments',
+    'freekassa_payments',
+    'kassa_ai_payments',
+    'riopay_payments',
+    'severpay_payments',
+]
+
+# All tables that get a metadata purchase_token index (for downgrade)
+_ALL_METADATA_TABLES = [*_TABLES_WITH_IS_PAID, 'heleket_payments']
+
+
+def upgrade() -> None:
+    # 1. Add retry_count column to guest_purchases (safe: has server_default)
+    op.add_column(
+        'guest_purchases',
+        sa.Column('retry_count', sa.Integer(), nullable=False, server_default='0'),
+    )
+
+    # 2. Create expression indexes for payment recovery queries.
+    #    These allow efficient lookup of succeeded payments by purchase_token
+    #    stored inside the metadata_json column.
+    with op.get_context().autocommit_block():
+        # Tables with is_paid boolean column
+        for table in _TABLES_WITH_IS_PAID:
+            idx_name = f'ix_{table}_metadata_purchase_token'
+            op.execute(
+                sa.text(
+                    f'CREATE INDEX CONCURRENTLY IF NOT EXISTS {idx_name} '
+                    f"ON {table} ((metadata_json ->> 'purchase_token')) "
+                    f'WHERE is_paid = TRUE'
+                )
+            )
+
+        # Heleket: no is_paid column (it's a Python @property), use status filter
+        op.execute(
+            sa.text(
+                'CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_heleket_payments_metadata_purchase_token '
+                "ON heleket_payments ((metadata_json ->> 'purchase_token')) "
+                "WHERE status IN ('paid', 'paid_over')"
+            )
+        )
+
+        # CryptoBot: payload (text) column with JSON inside, no metadata_json
+        op.execute(
+            sa.text(
+                'CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_cryptobot_payments_payload_purchase_token '
+                "ON cryptobot_payments ((CAST(payload AS json) ->> 'purchase_token')) "
+                "WHERE status = 'paid'"
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        for table in _ALL_METADATA_TABLES:
+            idx_name = f'ix_{table}_metadata_purchase_token'
+            op.execute(sa.text(f'DROP INDEX CONCURRENTLY IF EXISTS {idx_name}'))
+
+        op.execute(
+            sa.text('DROP INDEX CONCURRENTLY IF EXISTS ix_cryptobot_payments_payload_purchase_token')
+        )
+
+    op.drop_column('guest_purchases', 'retry_count')

uv.lock

@@ -1115,7 +1115,7 @@ wheels = [
 
 [[package]]
 name = "remnawave-bedolaga-telegram-bot"
-version = "3.33.0"
+version = "3.34.1"
 source = { virtual = "." }
 dependencies = [
     { name = "aiogram" },