Merge pull request #2397 from BEDOLAGA-DEV/dev5

Update subscription.py

.env.example

@@ -581,6 +581,23 @@ FREEKASSA_PAYMENT_SYSTEM_ID=
 # Использовать API для создания заказов (обязательно для NSPK СБП)
 FREEKASSA_USE_API=false
 
+# ===== KASSA AI (api.fk.life) =====
+# Отдельная платёжная система, работает параллельно с Freekassa
+KASSA_AI_ENABLED=false
+KASSA_AI_SHOP_ID=
+KASSA_AI_API_KEY=
+# Секретное слово 2 (для webhook)
+KASSA_AI_SECRET_WORD_2=
+KASSA_AI_DISPLAY_NAME=KassaAI
+KASSA_AI_CURRENCY=RUB
+KASSA_AI_MIN_AMOUNT_KOPEKS=10000
+KASSA_AI_MAX_AMOUNT_KOPEKS=100000000
+KASSA_AI_WEBHOOK_PATH=/kassa-ai-webhook
+KASSA_AI_WEBHOOK_HOST=0.0.0.0
+KASSA_AI_WEBHOOK_PORT=8089
+# Способ оплаты: 44 = СБП (QR), 36 = Карты РФ, 43 = SberPay
+KASSA_AI_PAYMENT_SYSTEM_ID=44
+
 # ===== WATA =====
 WATA_ENABLED=false
 WATA_BASE_URL=https://api.wata.pro

README.md

@@ -1,6 +1,6 @@
 # 🚀 Remnawave Bedolaga Bot
 
-> **🆕 Новый веб-кабинет (Cabinet WebApp)**
+> **🆕 Новый веб-кабинет (Cabinet WebApp) https://github.com/BEDOLAGA-DEV/bedolaga-cabinet/**
 >
 > Вышла новая версия личного кабинета пользователя — веб-интерфейс для управления подписками!
 >

app/cabinet/routes/admin_promo_offers.py

@@ -2,13 +2,21 @@
 
 from __future__ import annotations
 
+import asyncio
+import logging
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+from aiogram import Bot
+from aiogram.client.default import DefaultBotProperties
+from aiogram.enums import ParseMode
+from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.config import settings
 from app.database.crud.discount_offer import (
     count_discount_offers,
     list_discount_offers,
@@ -24,9 +32,12 @@ from app.database.crud.promo_offer_template import (
 from app.database.crud.user import get_user_by_telegram_id
 from app.database.models import DiscountOffer, PromoOfferLog, PromoOfferTemplate, User
 from app.handlers.admin.messages import get_custom_users, get_target_users
+from app.utils.miniapp_buttons import build_miniapp_or_callback_button
 
 from ..dependencies import get_cabinet_db, get_current_admin_user
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/admin/promo-offers", tags=["Admin Promo Offers"])
 
 
@@ -110,12 +121,18 @@ class PromoOfferBroadcastRequest(BaseModel):
     target: Optional[str] = None
     user_id: Optional[int] = None
     telegram_id: Optional[int] = None
+    # Telegram notification options
+    send_notification: bool = Field(False, description="Send Telegram notification to users")
+    message_text: Optional[str] = Field(None, description="Custom message text (HTML)")
+    button_text: Optional[str] = Field(None, description="Button text")
 
 
 class PromoOfferBroadcastResponse(BaseModel):
     created_offers: int
     user_ids: List[int]
     target: Optional[str] = None
+    notifications_sent: int = 0
+    notifications_failed: int = 0
 
 
 class PromoOfferLogOfferInfo(BaseModel):
@@ -344,13 +361,137 @@ async def list_offers(
     )
 
 
+def _get_bot() -> Bot:
+    """Create bot instance for sending notifications."""
+    return Bot(
+        token=settings.BOT_TOKEN,
+        default=DefaultBotProperties(parse_mode=ParseMode.HTML),
+    )
+
+
+def _build_default_promo_message(
+    discount_percent: int,
+    bonus_amount_kopeks: int,
+    valid_hours: int,
+) -> str:
+    """Build default promo notification message."""
+    lines = ["🎁 <b>Специальное предложение для вас!</b>\n"]
+
+    if discount_percent > 0:
+        lines.append(f"🔥 Скидка <b>{discount_percent}%</b> на подписку")
+    if bonus_amount_kopeks > 0:
+        bonus_rub = bonus_amount_kopeks / 100
+        lines.append(f"💰 Бонус <b>{bonus_rub:.0f}₽</b> на баланс")
+
+    lines.append(f"\n⏰ Предложение действует <b>{valid_hours} ч.</b>")
+    lines.append("\nНажмите кнопку ниже, чтобы активировать!")
+
+    return "\n".join(lines)
+
+
+async def _send_promo_notifications(
+    offers_to_notify: List[tuple[User, DiscountOffer]],
+    message_text: Optional[str],
+    button_text: Optional[str],
+    discount_percent: int,
+    bonus_amount_kopeks: int,
+    valid_hours: int,
+) -> tuple[int, int]:
+    """Send Telegram notifications for promo offers.
+
+    Returns:
+        Tuple of (sent_count, failed_count)
+    """
+    if not offers_to_notify:
+        return 0, 0
+
+    bot = _get_bot()
+    sent = 0
+    failed = 0
+
+    # Build message text
+    text = message_text or _build_default_promo_message(
+        discount_percent=discount_percent,
+        bonus_amount_kopeks=bonus_amount_kopeks,
+        valid_hours=valid_hours,
+    )
+
+    # Default button text
+    btn_text = button_text or "🎁 Получить"
+
+    semaphore = asyncio.Semaphore(20)
+
+    async def send_single(user: User, offer: DiscountOffer) -> bool:
+        async with semaphore:
+            try:
+                keyboard = InlineKeyboardMarkup(
+                    inline_keyboard=[
+                        [
+                            build_miniapp_or_callback_button(
+                                text=btn_text,
+                                callback_data=f"claim_discount_{offer.id}",
+                            )
+                        ],
+                        [
+                            InlineKeyboardButton(
+                                text="❌ Закрыть",
+                                callback_data="promo_offer_close",
+                            )
+                        ],
+                    ]
+                )
+
+                await bot.send_message(
+                    chat_id=user.telegram_id,
+                    text=text,
+                    reply_markup=keyboard,
+                )
+                return True
+            except (TelegramForbiddenError, TelegramBadRequest) as exc:
+                logger.warning(
+                    "Failed to send promo notification to user %s: %s",
+                    user.telegram_id,
+                    exc,
+                )
+                return False
+            except Exception as exc:
+                logger.error(
+                    "Error sending promo notification to user %s: %s",
+                    user.telegram_id,
+                    exc,
+                )
+                return False
+
+    # Send in batches
+    batch_size = 50
+    for i in range(0, len(offers_to_notify), batch_size):
+        batch = offers_to_notify[i : i + batch_size]
+        tasks = [send_single(user, offer) for user, offer in batch]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        for result in results:
+            if isinstance(result, bool) and result:
+                sent += 1
+            else:
+                failed += 1
+
+        # Small delay between batches
+        if i + batch_size < len(offers_to_notify):
+            await asyncio.sleep(0.1)
+
+    # Close bot session
+    await bot.session.close()
+
+    return sent, failed
+
+
 @router.post("/broadcast", response_model=PromoOfferBroadcastResponse, status_code=status.HTTP_201_CREATED)
 async def broadcast_offer(
     payload: PromoOfferBroadcastRequest,
     admin: User = Depends(get_current_admin_user),
     db: AsyncSession = Depends(get_cabinet_db),
 ) -> PromoOfferBroadcastResponse:
-    """Broadcast promo offer to users."""
+    """Broadcast promo offer to users with optional Telegram notification."""
     recipients: dict[int, User] = {}
 
     # Resolve target segment
@@ -386,8 +527,10 @@ async def broadcast_offer(
             "No recipients: specify target or user",
         )
 
-    # Create offers for all recipients
+    # Create offers for all recipients and collect (user, offer) pairs
     created_offers = 0
+    offers_to_notify: List[tuple[User, DiscountOffer]] = []
+
     for recipient in recipients.values():
         offer = await upsert_discount_offer(
             db,
@@ -402,11 +545,28 @@ async def broadcast_offer(
         )
         if offer:
             created_offers += 1
+            offers_to_notify.append((recipient, offer))
+
+    # Send Telegram notifications if requested
+    notifications_sent = 0
+    notifications_failed = 0
+
+    if payload.send_notification and offers_to_notify:
+        notifications_sent, notifications_failed = await _send_promo_notifications(
+            offers_to_notify=offers_to_notify,
+            message_text=payload.message_text,
+            button_text=payload.button_text,
+            discount_percent=payload.discount_percent,
+            bonus_amount_kopeks=payload.bonus_amount_kopeks,
+            valid_hours=payload.valid_hours,
+        )
 
     return PromoOfferBroadcastResponse(
         created_offers=created_offers,
         user_ids=list(recipients.keys()),
         target=payload.target,
+        notifications_sent=notifications_sent,
+        notifications_failed=notifications_failed,
     )
 
 

app/cabinet/routes/auth.py

@@ -1,5 +1,6 @@
 """Authentication routes for cabinet."""
 
+import asyncio
 import hashlib
 import logging
 from datetime import datetime
@@ -296,11 +297,12 @@ async def register_email(
 
     await db.commit()
 
-    # Send verification email
-    if email_service.is_configured():
+    # Send verification email asynchronously (smtplib is blocking)
+    if settings.is_cabinet_email_verification_enabled() and email_service.is_configured():
         # TODO: Get actual verification URL from settings
         verification_url = "https://example.com/cabinet/verify-email"
-        email_service.send_verification_email(
+        await asyncio.to_thread(
+            email_service.send_verification_email,
             to_email=request.email,
             verification_token=verification_token,
             verification_url=verification_url,
@@ -375,15 +377,26 @@ async def resend_verification(
 
     await db.commit()
 
-    # Send verification email
-    if email_service.is_configured():
+    # Send verification email asynchronously (smtplib is blocking)
+    if settings.is_cabinet_email_verification_enabled() and email_service.is_configured():
         verification_url = "https://example.com/cabinet/verify-email"
-        email_service.send_verification_email(
+        await asyncio.to_thread(
+            email_service.send_verification_email,
             to_email=user.email,
             verification_token=verification_token,
             verification_url=verification_url,
             username=user.first_name,
         )
+    elif not settings.is_cabinet_email_verification_enabled():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Email verification is disabled",
+        )
+    elif not email_service.is_configured():
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Email service is not configured",
+        )
 
     return {"message": "Verification email sent"}
 
@@ -548,10 +561,11 @@ async def forgot_password(
 
     await db.commit()
 
-    # Send reset email
+    # Send reset email asynchronously (smtplib is blocking)
     if email_service.is_configured():
         reset_url = "https://example.com/cabinet/reset-password"
-        email_service.send_password_reset_email(
+        await asyncio.to_thread(
+            email_service.send_password_reset_email,
             to_email=user.email,
             reset_token=reset_token,
             reset_url=reset_url,

app/cabinet/routes/balance.py

@@ -650,6 +650,7 @@ async def create_topup(
                 user_id=user.id,
                 amount_kopeks=request.amount_kopeks,
                 description=settings.get_balance_payment_description(request.amount_kopeks),
+                telegram_id=user.telegram_id,
                 language=getattr(user, 'language', None) or settings.DEFAULT_LANGUAGE,
             )
 

app/cabinet/routes/subscription.py

@@ -34,6 +34,7 @@ from app.services.subscription_purchase_service import (
 )
 from app.services.user_cart_service import user_cart_service
 from app.utils.cache import cache, cache_key, RateLimitCache
+from app.utils.promo_offer import get_user_active_promo_discount_percent
 
 from ..dependencies import get_cabinet_db, get_current_cabinet_user
 from ..schemas.subscription import (
@@ -331,13 +332,27 @@ async def renew_subscription(
             detail="Invalid renewal period",
         )
 
-    # Apply discount
-    discount_percent = 0
+    # Apply promo group discount
+    original_price_kopeks = price_kopeks
+    promo_group_discount_percent = 0
     if hasattr(user, "get_promo_discount"):
-        discount_percent = user.get_promo_discount("period", request.period_days)
+        promo_group_discount_percent = user.get_promo_discount("period", request.period_days)
 
-    if discount_percent > 0:
-        price_kopeks = int(price_kopeks * (100 - discount_percent) / 100)
+    if promo_group_discount_percent > 0:
+        price_kopeks = int(price_kopeks * (100 - promo_group_discount_percent) / 100)
+
+    # Apply promo offer discount (temporary discount from promo offers)
+    promo_offer_discount_percent = get_user_active_promo_discount_percent(user)
+    promo_offer_discount_value = 0
+    if promo_offer_discount_percent > 0:
+        promo_offer_discount_value = price_kopeks * promo_offer_discount_percent // 100
+        price_kopeks = price_kopeks - promo_offer_discount_value
+
+    # Combined discount percent for display
+    discount_percent = promo_group_discount_percent
+    if promo_offer_discount_percent > 0 and original_price_kopeks > 0:
+        total_discount = original_price_kopeks - price_kopeks
+        discount_percent = int(total_discount * 100 / original_price_kopeks)
 
     # Check balance
     if user.balance_kopeks < price_kopeks:
@@ -400,13 +415,17 @@ async def renew_subscription(
     # Deduct balance and extend subscription
     user.balance_kopeks -= price_kopeks
 
+    # Consume promo offer discount if it was used
+    if promo_offer_discount_value > 0:
+        user.promo_offer_discount_percent = 0
+        user.promo_offer_discount_source = None
+        user.promo_offer_discount_expires_at = None
+
     # Extend from end_date or now if expired
     now = datetime.utcnow()
     if user.subscription.end_date and user.subscription.end_date > now:
-        from datetime import timedelta
         user.subscription.end_date = user.subscription.end_date + timedelta(days=request.period_days)
     else:
-        from datetime import timedelta
         user.subscription.end_date = now + timedelta(days=request.period_days)
         user.subscription.start_date = now
 
@@ -415,12 +434,20 @@ async def renew_subscription(
 
     await db.commit()
 
-    return {
+    response = {
         "message": "Subscription renewed successfully",
         "new_end_date": user.subscription.end_date.isoformat(),
         "amount_paid_kopeks": price_kopeks,
     }
 
+    # Add discount info to response
+    if promo_offer_discount_value > 0:
+        response["promo_discount_percent"] = promo_offer_discount_percent
+        response["promo_discount_amount_kopeks"] = promo_offer_discount_value
+        response["original_price_kopeks"] = original_price_kopeks
+
+    return response
+
 
 @router.get("/traffic-packages", response_model=List[TrafficPackageResponse])
 async def get_traffic_packages(
@@ -952,8 +979,8 @@ async def _build_tariff_response(
     periods = []
     if tariff.period_prices:
         for period_str, price_kopeks in sorted(tariff.period_prices.items(), key=lambda x: int(x[0])):
-            if int(price_kopeks) <= 0:
-                continue  # Skip disabled periods
+            if int(price_kopeks) < 0:
+                continue  # Skip disabled periods (negative price)
             period_days = int(period_str)
 
             # Apply promo group discount for this period
@@ -1363,6 +1390,14 @@ async def purchase_tariff(
             traffic_limit_gb = request.traffic_gb
             price_kopeks += traffic_price_kopeks
 
+        # Apply promo offer discount (temporary discount from promo offers)
+        price_before_promo_offer = price_kopeks
+        promo_offer_discount_percent = get_user_active_promo_discount_percent(user)
+        promo_offer_discount_value = 0
+        if promo_offer_discount_percent > 0:
+            promo_offer_discount_value = price_kopeks * promo_offer_discount_percent // 100
+            price_kopeks = price_kopeks - promo_offer_discount_value
+
         # Check balance
         if user.balance_kopeks < price_kopeks:
             missing = price_kopeks - user.balance_kopeks
@@ -1438,6 +1473,8 @@ async def purchase_tariff(
             description = f"Покупка тарифа '{tariff.name}' на {period_days} дней"
         if discount_percent > 0:
             description += f" (скидка {discount_percent}%)"
+        if promo_offer_discount_value > 0:
+            description += f" (промо -{promo_offer_discount_percent}%)"
         success = await subtract_user_balance(db, user, price_kopeks, description)
         if not success:
             raise HTTPException(
@@ -1445,6 +1482,12 @@ async def purchase_tariff(
                 detail="Failed to charge balance",
             )
 
+        # Consume promo offer discount if it was used
+        if promo_offer_discount_value > 0:
+            user.promo_offer_discount_percent = 0
+            user.promo_offer_discount_source = None
+            user.promo_offer_discount_expires_at = None
+
         # Create transaction
         await create_transaction(
             db=db,
@@ -1541,11 +1584,18 @@ async def purchase_tariff(
             response["discount_percent"] = discount_percent
             response["original_price_kopeks"] = original_price
             response["original_price_label"] = settings.format_price(original_price)
-            response["discount_amount_kopeks"] = original_price - price_kopeks
-            response["discount_label"] = settings.format_price(original_price - price_kopeks)
+            response["discount_amount_kopeks"] = original_price - price_before_promo_offer
+            response["discount_label"] = settings.format_price(original_price - price_before_promo_offer)
             if promo_group:
                 response["promo_group_name"] = promo_group.name
 
+        # Add promo offer discount info if it was applied
+        if promo_offer_discount_value > 0:
+            response["promo_offer_discount_percent"] = promo_offer_discount_percent
+            response["promo_offer_discount_amount_kopeks"] = promo_offer_discount_value
+            response["promo_offer_discount_label"] = settings.format_price(promo_offer_discount_value)
+            response["price_before_promo_offer_kopeks"] = price_before_promo_offer
+
         return response
 
     except HTTPException:

app/config.py

@@ -488,6 +488,21 @@ class Settings(BaseSettings):
     # Публичный IP сервера для Freekassa API (если не задан - определяется автоматически)
     SERVER_PUBLIC_IP: Optional[str] = None
 
+    # KassaAI (api.fk.life) - отдельная платёжка
+    KASSA_AI_ENABLED: bool = False
+    KASSA_AI_SHOP_ID: Optional[int] = None
+    KASSA_AI_API_KEY: Optional[str] = None
+    KASSA_AI_SECRET_WORD_2: Optional[str] = None  # Для webhook
+    KASSA_AI_DISPLAY_NAME: str = "KassaAI"
+    KASSA_AI_CURRENCY: str = "RUB"
+    KASSA_AI_MIN_AMOUNT_KOPEKS: int = 10000  # 100 руб
+    KASSA_AI_MAX_AMOUNT_KOPEKS: int = 100000000  # 1 000 000 руб
+    KASSA_AI_WEBHOOK_PATH: str = "/kassa-ai-webhook"
+    KASSA_AI_WEBHOOK_HOST: str = "0.0.0.0"
+    KASSA_AI_WEBHOOK_PORT: int = 8089
+    # Способ оплаты: 44 = СБП (QR код), 36 = Карты РФ, 43 = SberPay
+    KASSA_AI_PAYMENT_SYSTEM_ID: int = 44
+
     MAIN_MENU_MODE: str = "default"
     CONNECT_BUTTON_MODE: str = "guide"
     MINIAPP_CUSTOM_URL: str = ""
@@ -1683,6 +1698,21 @@ class Settings(BaseSettings):
     def get_freekassa_display_name_html(self) -> str:
         return html.escape(self.get_freekassa_display_name())
 
+    def is_kassa_ai_enabled(self) -> bool:
+        return (
+            self.KASSA_AI_ENABLED
+            and self.KASSA_AI_SHOP_ID is not None
+            and self.KASSA_AI_API_KEY is not None
+            and self.KASSA_AI_SECRET_WORD_2 is not None
+        )
+
+    def get_kassa_ai_display_name(self) -> str:
+        name = (self.KASSA_AI_DISPLAY_NAME or "").strip()
+        return name if name else "KassaAI"
+
+    def get_kassa_ai_display_name_html(self) -> str:
+        return html.escape(self.get_kassa_ai_display_name())
+
     def is_payment_verification_auto_check_enabled(self) -> bool:
         return self.PAYMENT_VERIFICATION_AUTO_CHECK_ENABLED
 

app/database/crud/kassa_ai.py

@@ -0,0 +1,161 @@
+"""CRUD операции для платежей KassaAI."""
+
+import json
+import logging
+from datetime import datetime
+from typing import Optional, List
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database.models import KassaAiPayment
+
+logger = logging.getLogger(__name__)
+
+
+async def create_kassa_ai_payment(
+    db: AsyncSession,
+    *,
+    user_id: int,
+    order_id: str,
+    amount_kopeks: int,
+    currency: str = "RUB",
+    description: Optional[str] = None,
+    payment_url: Optional[str] = None,
+    payment_system_id: Optional[int] = None,
+    expires_at: Optional[datetime] = None,
+    metadata_json: Optional[str] = None,
+) -> KassaAiPayment:
+    """Создает запись о платеже KassaAI."""
+    payment = KassaAiPayment(
+        user_id=user_id,
+        order_id=order_id,
+        amount_kopeks=amount_kopeks,
+        currency=currency,
+        description=description,
+        payment_url=payment_url,
+        payment_system_id=payment_system_id,
+        expires_at=expires_at,
+        metadata_json=json.loads(metadata_json) if metadata_json else None,
+        status="pending",
+        is_paid=False,
+    )
+    db.add(payment)
+    await db.commit()
+    await db.refresh(payment)
+    logger.info(f"Создан платеж KassaAI: order_id={order_id}, user_id={user_id}")
+    return payment
+
+
+async def get_kassa_ai_payment_by_order_id(
+    db: AsyncSession, order_id: str
+) -> Optional[KassaAiPayment]:
+    """Получает платеж по order_id."""
+    result = await db.execute(
+        select(KassaAiPayment).where(KassaAiPayment.order_id == order_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def get_kassa_ai_payment_by_external_order_id(
+    db: AsyncSession, kassa_ai_order_id: str
+) -> Optional[KassaAiPayment]:
+    """Получает платеж по ID от KassaAI (orderId)."""
+    result = await db.execute(
+        select(KassaAiPayment).where(
+            KassaAiPayment.kassa_ai_order_id == kassa_ai_order_id
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+async def get_kassa_ai_payment_by_id(
+    db: AsyncSession, payment_id: int
+) -> Optional[KassaAiPayment]:
+    """Получает платеж по ID."""
+    result = await db.execute(
+        select(KassaAiPayment).where(KassaAiPayment.id == payment_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def update_kassa_ai_payment_status(
+    db: AsyncSession,
+    payment: KassaAiPayment,
+    *,
+    status: str,
+    is_paid: bool = False,
+    kassa_ai_order_id: Optional[str] = None,
+    payment_system_id: Optional[int] = None,
+    callback_payload: Optional[dict] = None,
+    transaction_id: Optional[int] = None,
+) -> KassaAiPayment:
+    """Обновляет статус платежа."""
+    payment.status = status
+    payment.is_paid = is_paid
+    payment.updated_at = datetime.utcnow()
+
+    if is_paid:
+        payment.paid_at = datetime.utcnow()
+    if kassa_ai_order_id:
+        payment.kassa_ai_order_id = kassa_ai_order_id
+    if payment_system_id is not None:
+        payment.payment_system_id = payment_system_id
+    if callback_payload:
+        payment.callback_payload = callback_payload
+    if transaction_id:
+        payment.transaction_id = transaction_id
+
+    await db.commit()
+    await db.refresh(payment)
+    logger.info(
+        f"Обновлен статус платежа KassaAI: order_id={payment.order_id}, "
+        f"status={status}, is_paid={is_paid}"
+    )
+    return payment
+
+
+async def get_pending_kassa_ai_payments(
+    db: AsyncSession, user_id: int
+) -> List[KassaAiPayment]:
+    """Получает незавершенные платежи пользователя."""
+    result = await db.execute(
+        select(KassaAiPayment).where(
+            KassaAiPayment.user_id == user_id,
+            KassaAiPayment.status == "pending",
+            KassaAiPayment.is_paid == False,
+        )
+    )
+    return list(result.scalars().all())
+
+
+async def get_user_kassa_ai_payments(
+    db: AsyncSession,
+    user_id: int,
+    limit: int = 10,
+    offset: int = 0,
+) -> List[KassaAiPayment]:
+    """Получает платежи пользователя с пагинацией."""
+    result = await db.execute(
+        select(KassaAiPayment)
+        .where(KassaAiPayment.user_id == user_id)
+        .order_by(KassaAiPayment.created_at.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    return list(result.scalars().all())
+
+
+async def get_expired_pending_kassa_ai_payments(
+    db: AsyncSession,
+) -> List[KassaAiPayment]:
+    """Получает просроченные платежи в статусе pending."""
+    now = datetime.utcnow()
+    result = await db.execute(
+        select(KassaAiPayment).where(
+            KassaAiPayment.status == "pending",
+            KassaAiPayment.is_paid == False,
+            KassaAiPayment.expires_at < now,
+        )
+    )
+    return list(result.scalars().all())

app/database/crud/subscription.py

@@ -1606,6 +1606,15 @@ async def check_and_update_subscription_status(
     if (subscription.status == SubscriptionStatus.ACTIVE.value and
         subscription.end_date <= current_time):
 
+        # Детальное логирование для отладки проблемы с деактивацией
+        time_diff = current_time - subscription.end_date
+        logger.warning(
+            f"⏰ DEACTIVATION: подписка {subscription.id} (user_id={subscription.user_id}) "
+            f"деактивируется в check_and_update_subscription_status. "
+            f"end_date={subscription.end_date}, current_time={current_time}, "
+            f"просрочена на {time_diff}"
+        )
+
         subscription.status = SubscriptionStatus.EXPIRED.value
         subscription.updated_at = current_time
 

app/database/crud/transaction.py

@@ -9,6 +9,23 @@ from app.database.models import Transaction, TransactionType, PaymentMethod, Use
 
 logger = logging.getLogger(__name__)
 
+# Реальные платёжные методы для подсчёта дохода
+# Исключены: MANUAL (админские), BALANCE (оплата с баланса), NULL (колесо, промокоды, бонусы)
+REAL_PAYMENT_METHODS = [
+    PaymentMethod.TELEGRAM_STARS.value,
+    PaymentMethod.TRIBUTE.value,
+    PaymentMethod.YOOKASSA.value,
+    PaymentMethod.CRYPTOBOT.value,
+    PaymentMethod.HELEKET.value,
+    PaymentMethod.MULENPAY.value,
+    PaymentMethod.PAL24.value,
+    PaymentMethod.WATA.value,
+    PaymentMethod.PLATEGA.value,
+    PaymentMethod.CLOUDPAYMENTS.value,
+    PaymentMethod.FREEKASSA.value,
+    PaymentMethod.KASSA_AI.value,
+]
+
 
 async def create_transaction(
     db: AsyncSession,
@@ -209,6 +226,7 @@ async def get_transactions_statistics(
     if not end_date:
         end_date = datetime.utcnow()
     
+    # Доход считаем только по реальным платежам (исключаем колесо, промокоды, админские пополнения)
     income_result = await db.execute(
         select(func.coalesce(func.sum(Transaction.amount_kopeks), 0))
         .where(
@@ -216,7 +234,8 @@ async def get_transactions_statistics(
                 Transaction.type == TransactionType.DEPOSIT.value,
                 Transaction.is_completed == True,
                 Transaction.created_at >= start_date,
-                Transaction.created_at <= end_date
+                Transaction.created_at <= end_date,
+                Transaction.payment_method.in_(REAL_PAYMENT_METHODS)
             )
         )
     )
@@ -297,13 +316,15 @@ async def get_transactions_statistics(
     )
     transactions_today = today_result.scalar()
     
+    # Доход за сегодня - только реальные платежи
     today_income_result = await db.execute(
         select(func.coalesce(func.sum(Transaction.amount_kopeks), 0))
         .where(
             and_(
                 Transaction.type == TransactionType.DEPOSIT.value,
                 Transaction.is_completed == True,
-                Transaction.created_at >= today
+                Transaction.created_at >= today,
+                Transaction.payment_method.in_(REAL_PAYMENT_METHODS)
             )
         )
     )
@@ -333,9 +354,9 @@ async def get_revenue_by_period(
     db: AsyncSession,
     days: int = 30
 ) -> List[dict]:
-    
+    """Доход по дням - только реальные платежи."""
     start_date = datetime.utcnow() - timedelta(days=days)
-    
+
     result = await db.execute(
         select(
             func.date(Transaction.created_at).label('date'),
@@ -345,13 +366,14 @@ async def get_revenue_by_period(
             and_(
                 Transaction.type == TransactionType.DEPOSIT.value,
                 Transaction.is_completed == True,
-                Transaction.created_at >= start_date
+                Transaction.created_at >= start_date,
+                Transaction.payment_method.in_(REAL_PAYMENT_METHODS)
             )
         )
         .group_by(func.date(Transaction.created_at))
         .order_by(func.date(Transaction.created_at))
     )
-    
+
     return [{"date": row.date, "amount_kopeks": row.amount} for row in result]
 
 

app/database/models.py

@@ -108,6 +108,7 @@ class PaymentMethod(Enum):
     PLATEGA = "platega"
     CLOUDPAYMENTS = "cloudpayments"
     FREEKASSA = "freekassa"
+    KASSA_AI = "kassa_ai"
     MANUAL = "manual"
     BALANCE = "balance"
 
@@ -649,6 +650,74 @@ class FreekassaPayment(Base):
         )
 
 
+class KassaAiPayment(Base):
+    """Платежи через KassaAI (api.fk.life)."""
+    __tablename__ = "kassa_ai_payments"
+
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+
+    # Идентификаторы
+    order_id = Column(String(64), unique=True, nullable=False, index=True)  # Наш ID заказа
+    kassa_ai_order_id = Column(String(64), unique=True, nullable=True, index=True)  # orderId от KassaAI
+
+    # Суммы
+    amount_kopeks = Column(Integer, nullable=False)
+    currency = Column(String(10), nullable=False, default="RUB")
+    description = Column(Text, nullable=True)
+
+    # Статусы
+    status = Column(String(32), nullable=False, default="pending")  # pending, success, failed, expired
+    is_paid = Column(Boolean, default=False)
+
+    # Данные платежа
+    payment_url = Column(Text, nullable=True)
+    payment_system_id = Column(Integer, nullable=True)  # ID платежной системы (44=СБП, 36=Карты, 43=SberPay)
+
+    # Метаданные
+    metadata_json = Column(JSON, nullable=True)
+    callback_payload = Column(JSON, nullable=True)
+
+    # Временные метки
+    paid_at = Column(DateTime, nullable=True)
+    expires_at = Column(DateTime, nullable=True)
+    created_at = Column(DateTime, default=func.now())
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    # Связь с транзакцией
+    transaction_id = Column(Integer, ForeignKey("transactions.id"), nullable=True)
+
+    # Relationships
+    user = relationship("User", backref="kassa_ai_payments")
+    transaction = relationship("Transaction", backref="kassa_ai_payment")
+
+    @property
+    def amount_rubles(self) -> float:
+        return self.amount_kopeks / 100
+
+    @property
+    def is_pending(self) -> bool:
+        return self.status == "pending"
+
+    @property
+    def is_success(self) -> bool:
+        return self.status == "success" and self.is_paid
+
+    @property
+    def is_failed(self) -> bool:
+        return self.status in ["failed", "expired"]
+
+    def __repr__(self) -> str:  # pragma: no cover - debug helper
+        return (
+            "<KassaAiPayment(id={0}, order_id={1}, amount={2}₽, status={3})>".format(
+                self.id,
+                self.order_id,
+                self.amount_rubles,
+                self.status,
+            )
+        )
+
+
 class PromoGroup(Base):
     __tablename__ = "promo_groups"
 

app/database/universal_migration.py

@@ -1401,6 +1401,118 @@ async def create_freekassa_payments_table():
         return False
 
 
+async def create_kassa_ai_payments_table():
+    """Создаёт таблицу kassa_ai_payments для платежей через KassaAI."""
+    table_exists = await check_table_exists('kassa_ai_payments')
+    if table_exists:
+        logger.info("Таблица kassa_ai_payments уже существует")
+        return True
+
+    try:
+        async with engine.begin() as conn:
+            db_type = await get_database_type()
+
+            if db_type == 'sqlite':
+                create_sql = """
+                CREATE TABLE kassa_ai_payments (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    user_id INTEGER NOT NULL,
+                    order_id VARCHAR(64) NOT NULL UNIQUE,
+                    kassa_ai_order_id VARCHAR(64) NULL UNIQUE,
+                    amount_kopeks INTEGER NOT NULL,
+                    currency VARCHAR(10) NOT NULL DEFAULT 'RUB',
+                    description TEXT NULL,
+                    status VARCHAR(32) NOT NULL DEFAULT 'pending',
+                    is_paid BOOLEAN NOT NULL DEFAULT 0,
+                    payment_url TEXT NULL,
+                    payment_system_id INTEGER NULL,
+                    metadata_json JSON NULL,
+                    callback_payload JSON NULL,
+                    paid_at DATETIME NULL,
+                    expires_at DATETIME NULL,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    transaction_id INTEGER NULL,
+                    FOREIGN KEY (user_id) REFERENCES users(id),
+                    FOREIGN KEY (transaction_id) REFERENCES transactions(id)
+                );
+
+                CREATE INDEX idx_kassa_ai_user_id ON kassa_ai_payments(user_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_order_id ON kassa_ai_payments(order_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_kai_order_id ON kassa_ai_payments(kassa_ai_order_id);
+                """
+
+            elif db_type == 'postgresql':
+                create_sql = """
+                CREATE TABLE kassa_ai_payments (
+                    id SERIAL PRIMARY KEY,
+                    user_id INTEGER NOT NULL REFERENCES users(id),
+                    order_id VARCHAR(64) NOT NULL UNIQUE,
+                    kassa_ai_order_id VARCHAR(64) NULL UNIQUE,
+                    amount_kopeks INTEGER NOT NULL,
+                    currency VARCHAR(10) NOT NULL DEFAULT 'RUB',
+                    description TEXT NULL,
+                    status VARCHAR(32) NOT NULL DEFAULT 'pending',
+                    is_paid BOOLEAN NOT NULL DEFAULT FALSE,
+                    payment_url TEXT NULL,
+                    payment_system_id INTEGER NULL,
+                    metadata_json JSON NULL,
+                    callback_payload JSON NULL,
+                    paid_at TIMESTAMP NULL,
+                    expires_at TIMESTAMP NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    transaction_id INTEGER NULL REFERENCES transactions(id)
+                );
+
+                CREATE INDEX idx_kassa_ai_user_id ON kassa_ai_payments(user_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_order_id ON kassa_ai_payments(order_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_kai_order_id ON kassa_ai_payments(kassa_ai_order_id);
+                """
+
+            elif db_type == 'mysql':
+                create_sql = """
+                CREATE TABLE kassa_ai_payments (
+                    id INT AUTO_INCREMENT PRIMARY KEY,
+                    user_id INT NOT NULL,
+                    order_id VARCHAR(64) NOT NULL UNIQUE,
+                    kassa_ai_order_id VARCHAR(64) NULL UNIQUE,
+                    amount_kopeks INT NOT NULL,
+                    currency VARCHAR(10) NOT NULL DEFAULT 'RUB',
+                    description TEXT NULL,
+                    status VARCHAR(32) NOT NULL DEFAULT 'pending',
+                    is_paid BOOLEAN NOT NULL DEFAULT 0,
+                    payment_url TEXT NULL,
+                    payment_system_id INT NULL,
+                    metadata_json JSON NULL,
+                    callback_payload JSON NULL,
+                    paid_at DATETIME NULL,
+                    expires_at DATETIME NULL,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+                    transaction_id INT NULL,
+                    FOREIGN KEY (user_id) REFERENCES users(id),
+                    FOREIGN KEY (transaction_id) REFERENCES transactions(id)
+                );
+
+                CREATE INDEX idx_kassa_ai_user_id ON kassa_ai_payments(user_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_order_id ON kassa_ai_payments(order_id);
+                CREATE UNIQUE INDEX idx_kassa_ai_kai_order_id ON kassa_ai_payments(kassa_ai_order_id);
+                """
+
+            else:
+                logger.error(f"Неподдерживаемый тип БД для таблицы kassa_ai_payments: {db_type}")
+                return False
+
+            await conn.execute(text(create_sql))
+            logger.info("Таблица kassa_ai_payments успешно создана")
+            return True
+
+    except Exception as e:
+        logger.error(f"Ошибка создания таблицы kassa_ai_payments: {e}")
+        return False
+
+
 async def create_discount_offers_table():
     table_exists = await check_table_exists('discount_offers')
     if table_exists:
@@ -6333,6 +6445,13 @@ async def run_universal_migration():
         else:
             logger.warning("⚠️ Проблемы с таблицей Freekassa payments")
 
+        logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ KASSA_AI ===")
+        kassa_ai_created = await create_kassa_ai_payments_table()
+        if kassa_ai_created:
+            logger.info("✅ Таблица KassaAI payments готова")
+        else:
+            logger.warning("⚠️ Проблемы с таблицей KassaAI payments")
+
         logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ DISCOUNT_OFFERS ===")
         discount_created = await create_discount_offers_table()
         if discount_created:

app/handlers/balance/kassa_ai.py

@@ -0,0 +1,391 @@
+"""Handler for KassaAI balance top-up."""
+
+import logging
+
+from aiogram import types
+from aiogram.fsm.context import FSMContext
+from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.database.models import User
+from app.keyboards.inline import get_back_keyboard
+from app.localization.texts import get_texts
+from app.services.payment_service import PaymentService
+from app.states import BalanceStates
+from app.utils.decorators import error_handler
+
+logger = logging.getLogger(__name__)
+
+
+async def _create_kassa_ai_payment_and_respond(
+    message_or_callback,
+    db_user: User,
+    db: AsyncSession,
+    amount_kopeks: int,
+    edit_message: bool = False,
+):
+    """
+    Common logic for creating KassaAI payment and sending response.
+
+    Args:
+        message_or_callback: Either a Message or CallbackQuery object
+        db_user: User object
+        db: Database session
+        amount_kopeks: Amount in kopeks
+        edit_message: Whether to edit existing message or send new one
+    """
+    texts = get_texts(db_user.language)
+    amount_rub = amount_kopeks / 100
+
+    # Create payment
+    payment_service = PaymentService()
+
+    description = settings.PAYMENT_BALANCE_TEMPLATE.format(
+        service_name=settings.PAYMENT_SERVICE_NAME,
+        description="Пополнение баланса",
+    )
+
+    result = await payment_service.create_kassa_ai_payment(
+        db=db,
+        user_id=db_user.id,
+        amount_kopeks=amount_kopeks,
+        description=description,
+        email=getattr(db_user, "email", None),
+        language=db_user.language,
+    )
+
+    if not result:
+        error_text = texts.t(
+            "PAYMENT_CREATE_ERROR",
+            "Не удалось создать платёж. Попробуйте позже.",
+        )
+        if edit_message:
+            await message_or_callback.edit_text(
+                error_text,
+                reply_markup=get_back_keyboard(db_user.language),
+                parse_mode="HTML",
+            )
+        else:
+            await message_or_callback.answer(
+                error_text,
+                parse_mode="HTML",
+            )
+        return
+
+    payment_url = result.get("payment_url")
+    display_name = settings.get_kassa_ai_display_name()
+
+    # Create keyboard with payment button
+    keyboard = InlineKeyboardMarkup(
+        inline_keyboard=[
+            [
+                InlineKeyboardButton(
+                    text=texts.t(
+                        "PAY_BUTTON",
+                        "💳 Оплатить {amount}₽",
+                    ).format(amount=f"{amount_rub:.0f}"),
+                    url=payment_url,
+                )
+            ],
+            [
+                InlineKeyboardButton(
+                    text=texts.t("BACK_BUTTON", "◀️ Назад"),
+                    callback_data="menu_balance",
+                )
+            ],
+        ]
+    )
+
+    response_text = texts.t(
+        "KASSA_AI_PAYMENT_CREATED",
+        "💳 <b>Оплата через {name}</b>\n\n"
+        "Сумма: <b>{amount}₽</b>\n\n"
+        "Нажмите кнопку ниже для оплаты.\n"
+        "После успешной оплаты баланс будет пополнен автоматически.",
+    ).format(name=display_name, amount=f"{amount_rub:.2f}")
+
+    if edit_message:
+        await message_or_callback.edit_text(
+            response_text,
+            reply_markup=keyboard,
+            parse_mode="HTML",
+        )
+    else:
+        await message_or_callback.answer(
+            response_text,
+            reply_markup=keyboard,
+            parse_mode="HTML",
+        )
+
+    logger.info(
+        "KassaAI payment created: user=%s, amount=%s₽",
+        db_user.telegram_id,
+        amount_rub,
+    )
+
+
+@error_handler
+async def process_kassa_ai_payment_amount(
+    message: types.Message,
+    db_user: User,
+    db: AsyncSession,
+    amount_kopeks: int,
+    state: FSMContext,
+):
+    """
+    Process payment amount directly (called from quick_amount handlers).
+    """
+    texts = get_texts(db_user.language)
+
+    # Проверка ограничения на пополнение
+    if getattr(db_user, "restriction_topup", False):
+        reason = (
+            getattr(db_user, "restriction_reason", None)
+            or "Действие ограничено администратором"
+        )
+        support_url = settings.get_support_contact_url()
+        keyboard = []
+        if support_url:
+            keyboard.append(
+                [InlineKeyboardButton(text="🆘 Обжаловать", url=support_url)]
+            )
+        keyboard.append(
+            [InlineKeyboardButton(text=texts.BACK, callback_data="menu_balance")]
+        )
+
+        await message.answer(
+            f"🚫 <b>Пополнение ограничено</b>\n\n{reason}",
+            parse_mode="HTML",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard),
+        )
+        await state.clear()
+        return
+
+    # Validate amount
+    min_amount = settings.KASSA_AI_MIN_AMOUNT_KOPEKS
+    max_amount = settings.KASSA_AI_MAX_AMOUNT_KOPEKS
+
+    if amount_kopeks < min_amount:
+        await message.answer(
+            texts.t(
+                "PAYMENT_AMOUNT_TOO_LOW",
+                "Минимальная сумма пополнения: {min_amount}₽",
+            ).format(min_amount=min_amount // 100),
+            parse_mode="HTML",
+        )
+        return
+
+    if amount_kopeks > max_amount:
+        await message.answer(
+            texts.t(
+                "PAYMENT_AMOUNT_TOO_HIGH",
+                "Максимальная сумма пополнения: {max_amount}₽",
+            ).format(max_amount=max_amount // 100),
+            parse_mode="HTML",
+        )
+        return
+
+    await state.clear()
+
+    await _create_kassa_ai_payment_and_respond(
+        message_or_callback=message,
+        db_user=db_user,
+        db=db,
+        amount_kopeks=amount_kopeks,
+        edit_message=False,
+    )
+
+
+@error_handler
+async def start_kassa_ai_topup(
+    callback: types.CallbackQuery,
+    db_user: User,
+    db: AsyncSession,
+    state: FSMContext,
+):
+    """
+    Start KassaAI top-up process - ask for amount.
+    """
+    texts = get_texts(db_user.language)
+
+    # Проверка ограничения на пополнение
+    if getattr(db_user, "restriction_topup", False):
+        reason = (
+            getattr(db_user, "restriction_reason", None)
+            or "Действие ограничено администратором"
+        )
+        support_url = settings.get_support_contact_url()
+        keyboard = []
+        if support_url:
+            keyboard.append(
+                [InlineKeyboardButton(text="🆘 Обжаловать", url=support_url)]
+            )
+        keyboard.append(
+            [InlineKeyboardButton(text=texts.BACK, callback_data="menu_balance")]
+        )
+
+        await callback.message.edit_text(
+            f"🚫 <b>Пополнение ограничено</b>\n\n{reason}",
+            parse_mode="HTML",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard),
+        )
+        return
+
+    await state.set_state(BalanceStates.waiting_for_amount)
+    await state.update_data(payment_method="kassa_ai")
+
+    min_amount = settings.KASSA_AI_MIN_AMOUNT_KOPEKS // 100
+    max_amount = settings.KASSA_AI_MAX_AMOUNT_KOPEKS // 100
+    display_name = settings.get_kassa_ai_display_name()
+
+    keyboard = InlineKeyboardMarkup(
+        inline_keyboard=[
+            [
+                InlineKeyboardButton(
+                    text=texts.t("BACK_BUTTON", "◀️ Назад"),
+                    callback_data="menu_balance",
+                )
+            ]
+        ]
+    )
+
+    await callback.message.edit_text(
+        texts.t(
+            "KASSA_AI_ENTER_AMOUNT",
+            "💳 <b>Пополнение через {name}</b>\n\n"
+            "Введите сумму пополнения в рублях.\n\n"
+            "Минимум: {min_amount}₽\n"
+            "Максимум: {max_amount}₽",
+        ).format(
+            name=display_name,
+            min_amount=min_amount,
+            max_amount=f"{max_amount:,}".replace(",", " "),
+        ),
+        parse_mode="HTML",
+        reply_markup=keyboard,
+    )
+
+
+@error_handler
+async def process_kassa_ai_custom_amount(
+    message: types.Message,
+    db_user: User,
+    db: AsyncSession,
+    state: FSMContext,
+):
+    """
+    Process custom amount input for KassaAI payment.
+    """
+    data = await state.get_data()
+    if data.get("payment_method") != "kassa_ai":
+        return
+
+    texts = get_texts(db_user.language)
+
+    try:
+        amount_text = message.text.replace(",", ".").replace(" ", "").strip()
+        amount_rubles = float(amount_text)
+        amount_kopeks = int(amount_rubles * 100)
+    except (ValueError, TypeError):
+        await message.answer(
+            texts.t(
+                "PAYMENT_INVALID_AMOUNT",
+                "Введите корректную сумму числом.",
+            ),
+            parse_mode="HTML",
+        )
+        return
+
+    await process_kassa_ai_payment_amount(
+        message=message,
+        db_user=db_user,
+        db=db,
+        amount_kopeks=amount_kopeks,
+        state=state,
+    )
+
+
+@error_handler
+async def process_kassa_ai_quick_amount(
+    callback: types.CallbackQuery,
+    db_user: User,
+    db: AsyncSession,
+    state: FSMContext,
+):
+    """
+    Process quick amount selection for KassaAI payment.
+    Called when user clicks a predefined amount button.
+    """
+    texts = get_texts(db_user.language)
+
+    if not settings.is_kassa_ai_enabled():
+        await callback.answer(
+            texts.t("KASSA_AI_NOT_AVAILABLE", "KassaAI временно недоступен"),
+            show_alert=True,
+        )
+        return
+
+    # Extract amount from callback data: topup_amount|kassa_ai|{amount_kopeks}
+    try:
+        parts = callback.data.split("|")
+        if len(parts) >= 3:
+            amount_kopeks = int(parts[2])
+        else:
+            await callback.answer("Invalid callback data", show_alert=True)
+            return
+    except (ValueError, IndexError):
+        await callback.answer("Invalid amount", show_alert=True)
+        return
+
+    # Проверка ограничения на пополнение
+    if getattr(db_user, "restriction_topup", False):
+        reason = (
+            getattr(db_user, "restriction_reason", None)
+            or "Действие ограничено администратором"
+        )
+        support_url = settings.get_support_contact_url()
+        keyboard = []
+        if support_url:
+            keyboard.append(
+                [InlineKeyboardButton(text="🆘 Обжаловать", url=support_url)]
+            )
+        keyboard.append(
+            [InlineKeyboardButton(text=texts.BACK, callback_data="menu_balance")]
+        )
+
+        await callback.message.edit_text(
+            f"🚫 <b>Пополнение ограничено</b>\n\n{reason}",
+            parse_mode="HTML",
+            reply_markup=InlineKeyboardMarkup(inline_keyboard=keyboard),
+        )
+        return
+
+    # Validate amount
+    min_amount = settings.KASSA_AI_MIN_AMOUNT_KOPEKS
+    max_amount = settings.KASSA_AI_MAX_AMOUNT_KOPEKS
+
+    if amount_kopeks < min_amount:
+        await callback.answer(
+            texts.t("AMOUNT_TOO_LOW_SHORT", "Сумма слишком мала"),
+            show_alert=True,
+        )
+        return
+
+    if amount_kopeks > max_amount:
+        await callback.answer(
+            texts.t("AMOUNT_TOO_HIGH_SHORT", "Сумма слишком велика"),
+            show_alert=True,
+        )
+        return
+
+    await callback.answer()
+    await state.clear()
+
+    await _create_kassa_ai_payment_and_respond(
+        message_or_callback=callback.message,
+        db_user=db_user,
+        db=db,
+        amount_kopeks=amount_kopeks,
+        edit_message=True,
+    )

app/handlers/balance/main.py

@@ -118,6 +118,12 @@ async def route_payment_by_method(
             await process_freekassa_payment_amount(message, db_user, db, amount_kopeks, state)
         return True
 
+    if payment_method == "kassa_ai":
+        from .kassa_ai import process_kassa_ai_payment_amount
+        async with AsyncSessionLocal() as db:
+            await process_kassa_ai_payment_amount(message, db_user, db, amount_kopeks, state)
+        return True
+
     return False
 
 
@@ -990,6 +996,16 @@ def register_balance_handlers(dp: Dispatcher):
         F.data.startswith("topup_amount|freekassa|")
     )
 
+    from .kassa_ai import start_kassa_ai_topup, process_kassa_ai_quick_amount
+    dp.callback_query.register(
+        start_kassa_ai_topup,
+        F.data == "topup_kassa_ai"
+    )
+    dp.callback_query.register(
+        process_kassa_ai_quick_amount,
+        F.data.startswith("topup_amount|kassa_ai|")
+    )
+
     from .mulenpay import check_mulenpay_payment_status
     dp.callback_query.register(
         check_mulenpay_payment_status,

app/handlers/start.py

@@ -48,6 +48,10 @@ from app.utils.promo_offer import (
 )
 from app.utils.timezone import format_local_datetime
 from app.database.crud.user_message import get_random_active_message
+from app.middlewares.channel_checker import (
+    get_pending_payload_from_redis,
+    delete_pending_payload_from_redis,
+)
 from app.database.crud.subscription import decrement_subscription_server_counts
 from app.services.blacklist_service import blacklist_service
 
@@ -327,6 +331,19 @@ async def cmd_start(message: types.Message, state: FSMContext, db: AsyncSession,
     campaign_notification_sent = data.pop("campaign_notification_sent", False)
     state_needs_update = had_pending_payload or had_campaign_notification_flag
 
+    # Если в FSM state нет payload, пробуем получить из Redis (резервный механизм)
+    if not pending_start_payload:
+        redis_payload = await get_pending_payload_from_redis(message.from_user.id)
+        if redis_payload:
+            pending_start_payload = redis_payload
+            state_needs_update = True
+            logger.info(
+                "📦 START: Payload '%s' восстановлен из Redis (fallback)",
+                pending_start_payload,
+            )
+            # Очищаем Redis после получения
+            await delete_pending_payload_from_redis(message.from_user.id)
+
     referral_code = None
     campaign = None
     start_args = message.text.split()
@@ -1832,6 +1849,17 @@ async def required_sub_channel_check(
         state_data = await state.get_data() or {}
 
         pending_start_payload = state_data.pop("pending_start_payload", None)
+
+        # Если в FSM state нет payload, пробуем получить из Redis (резервный механизм)
+        if not pending_start_payload:
+            redis_payload = await get_pending_payload_from_redis(query.from_user.id)
+            if redis_payload:
+                pending_start_payload = redis_payload
+                logger.info(
+                    "📦 CHANNEL CHECK: Payload '%s' восстановлен из Redis (fallback)",
+                    pending_start_payload,
+                )
+
         state_updated = pending_start_payload is not None
 
         if pending_start_payload:
@@ -1840,27 +1868,27 @@ async def required_sub_channel_check(
                 pending_start_payload,
             )
 
-            if "campaign_id" not in state_data and "referral_code" not in state_data:
-                campaign = await get_campaign_by_start_parameter(
-                    db,
-                    pending_start_payload,
-                    only_active=True,
-                )
+            # Очищаем Redis после получения payload
+            await delete_pending_payload_from_redis(query.from_user.id)
 
-                if campaign:
-                    state_data["campaign_id"] = campaign.id
-                    logger.info(
-                        "📣 CHANNEL CHECK: Кампания %s восстановлена из payload",
-                        campaign.id,
-                    )
-                else:
-                    state_data["referral_code"] = pending_start_payload
-                    logger.info(
-                        "🎯 CHANNEL CHECK: Payload интерпретирован как реферальный код",
-                    )
+            # Всегда обновляем referral_code если есть новый payload
+            # (исправление бага с устаревшими данными в state)
+            campaign = await get_campaign_by_start_parameter(
+                db,
+                pending_start_payload,
+                only_active=True,
+            )
+
+            if campaign:
+                state_data["campaign_id"] = campaign.id
+                logger.info(
+                    "📣 CHANNEL CHECK: Кампания %s восстановлена из payload",
+                    campaign.id,
+                )
             else:
-                logger.debug(
-                    "ℹ️ CHANNEL CHECK: Payload уже обработан ранее, пропускаем восстановление",
+                state_data["referral_code"] = pending_start_payload
+                logger.info(
+                    "🎯 CHANNEL CHECK: Payload интерпретирован как реферальный код",
                 )
 
         if state_updated:

app/handlers/subscription/purchase.py

@@ -599,7 +599,18 @@ async def show_trial_offer(
 
     texts = get_texts(db_user.language)
 
-    if db_user.subscription or db_user.has_had_paid_subscription:
+    # Проверяем, использовал ли пользователь триал
+    # PENDING триальные подписки не считаются - пользователь может повторить оплату
+    trial_blocked = False
+    if db_user.has_had_paid_subscription:
+        trial_blocked = True
+    elif db_user.subscription:
+        sub = db_user.subscription
+        # Разрешаем если это PENDING триальная подписка (повторная попытка оплаты)
+        if not (sub.status == SubscriptionStatus.PENDING.value and sub.is_trial):
+            trial_blocked = True
+
+    if trial_blocked:
         await callback.message.edit_text(
             texts.TRIAL_ALREADY_USED,
             reply_markup=get_back_keyboard(db_user.language)
@@ -814,7 +825,18 @@ async def activate_trial(
         await callback.answer()
         return
 
-    if db_user.subscription or db_user.has_had_paid_subscription:
+    # Проверяем, использовал ли пользователь триал
+    # PENDING триальные подписки не считаются - пользователь может повторить оплату
+    trial_blocked = False
+    if db_user.has_had_paid_subscription:
+        trial_blocked = True
+    elif db_user.subscription:
+        sub = db_user.subscription
+        # Разрешаем если это PENDING триальная подписка (повторная попытка оплаты)
+        if not (sub.status == SubscriptionStatus.PENDING.value and sub.is_trial):
+            trial_blocked = True
+
+    if trial_blocked:
         await callback.message.edit_text(
             texts.TRIAL_ALREADY_USED,
             reply_markup=get_back_keyboard(db_user.language)
@@ -3360,7 +3382,17 @@ async def handle_trial_pay_with_balance(
     texts = get_texts(db_user.language)
 
     # Проверяем права на триал
-    if db_user.subscription or db_user.has_had_paid_subscription:
+    # PENDING триальные подписки не считаются - пользователь может повторить оплату
+    trial_blocked = False
+    if db_user.has_had_paid_subscription:
+        trial_blocked = True
+    elif db_user.subscription:
+        sub = db_user.subscription
+        # Разрешаем если это PENDING триальная подписка (повторная попытка оплаты)
+        if not (sub.status == SubscriptionStatus.PENDING.value and sub.is_trial):
+            trial_blocked = True
+
+    if trial_blocked:
         await callback.message.edit_text(
             texts.TRIAL_ALREADY_USED,
             reply_markup=get_back_keyboard(db_user.language)
@@ -3695,7 +3727,17 @@ async def handle_trial_payment_method(
     texts = get_texts(db_user.language)
 
     # Проверяем права на триал
-    if db_user.subscription or db_user.has_had_paid_subscription:
+    # PENDING триальные подписки не считаются - пользователь может повторить оплату
+    trial_blocked = False
+    if db_user.has_had_paid_subscription:
+        trial_blocked = True
+    elif db_user.subscription:
+        sub = db_user.subscription
+        # Разрешаем если это PENDING триальная подписка (повторная попытка оплаты)
+        if not (sub.status == SubscriptionStatus.PENDING.value and sub.is_trial):
+            trial_blocked = True
+
+    if trial_blocked:
         await callback.message.edit_text(
             texts.TRIAL_ALREADY_USED,
             reply_markup=get_back_keyboard(db_user.language)
@@ -3808,11 +3850,12 @@ async def handle_trial_payment_method(
         elif payment_method == "yookassa":
             # Оплата через YooKassa карта
             payment_result = await payment_service.create_yookassa_payment(
+                db=db,
+                user_id=db_user.id,
                 amount_kopeks=trial_price_kopeks,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
                 metadata={
                     "type": "trial",
                     "subscription_id": pending_subscription.id,
@@ -3840,20 +3883,37 @@ async def handle_trial_payment_method(
 
         elif payment_method == "cryptobot":
             # Оплата через CryptoBot
+            # Конвертируем копейки в USD
+            from app.utils.currency_converter import currency_converter
+            try:
+                usd_rate = await currency_converter.get_usd_to_rub_rate()
+            except Exception as rate_error:
+                logger.warning("Не удалось получить курс USD: %s", rate_error)
+                usd_rate = 95.0
+
+            amount_rubles = trial_price_kopeks / 100
+            amount_usd = round(amount_rubles / usd_rate, 2)
+            if amount_usd < 1:
+                amount_usd = 1.0
+
             payment_result = await payment_service.create_cryptobot_payment(
-                amount_kopeks=trial_price_kopeks,
+                db=db,
+                user_id=db_user.id,
+                amount_usd=amount_usd,
+                asset=settings.CRYPTOBOT_DEFAULT_ASSET,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
-                metadata={
-                    "type": "trial",
-                    "subscription_id": pending_subscription.id,
-                    "user_id": db_user.id,
-                },
+                payload=f"trial_{pending_subscription.id}_{db_user.id}",
             )
 
-            if not payment_result or not payment_result.get("pay_url"):
+            payment_url = (
+                payment_result.get("mini_app_invoice_url")
+                or payment_result.get("bot_invoice_url")
+                or payment_result.get("web_app_invoice_url")
+            ) if payment_result else None
+
+            if not payment_result or not payment_url:
                 await callback.answer("❌ Не удалось создать платеж. Попробуйте позже.", show_alert=True)
                 return
 
@@ -3865,7 +3925,7 @@ async def handle_trial_payment_method(
                     "💰 Сумма: {amount}"
                 ).format(amount=settings.format_price(trial_price_kopeks)),
                 reply_markup=InlineKeyboardMarkup(inline_keyboard=[
-                    [InlineKeyboardButton(text="🪙 Оплатить", url=payment_result["pay_url"])],
+                    [InlineKeyboardButton(text="🪙 Оплатить", url=payment_url)],
                     [InlineKeyboardButton(
                         text=texts.t("CHECK_PAYMENT", "🔄 Проверить оплату"),
                         callback_data=f"check_trial_cryptobot_{pending_subscription.id}"
@@ -3878,19 +3938,16 @@ async def handle_trial_payment_method(
         elif payment_method == "heleket":
             # Оплата через Heleket
             payment_result = await payment_service.create_heleket_payment(
+                db=db,
+                user_id=db_user.id,
                 amount_kopeks=trial_price_kopeks,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
-                metadata={
-                    "type": "trial",
-                    "subscription_id": pending_subscription.id,
-                    "user_id": db_user.id,
-                },
+                language=db_user.language,
             )
 
-            if not payment_result or not payment_result.get("pay_url"):
+            if not payment_result or not payment_result.get("payment_url"):
                 await callback.answer("❌ Не удалось создать платеж. Попробуйте позже.", show_alert=True)
                 return
 
@@ -3902,7 +3959,7 @@ async def handle_trial_payment_method(
                     "💰 Сумма: {amount}"
                 ).format(amount=settings.format_price(trial_price_kopeks)),
                 reply_markup=InlineKeyboardMarkup(inline_keyboard=[
-                    [InlineKeyboardButton(text="🪙 Оплатить", url=payment_result["pay_url"])],
+                    [InlineKeyboardButton(text="🪙 Оплатить", url=payment_result["payment_url"])],
                     [InlineKeyboardButton(
                         text=texts.t("CHECK_PAYMENT", "🔄 Проверить оплату"),
                         callback_data=f"check_trial_heleket_{pending_subscription.id}"
@@ -3915,19 +3972,16 @@ async def handle_trial_payment_method(
         elif payment_method == "mulenpay":
             # Оплата через MulenPay
             payment_result = await payment_service.create_mulenpay_payment(
+                db=db,
+                user_id=db_user.id,
                 amount_kopeks=trial_price_kopeks,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
-                metadata={
-                    "type": "trial",
-                    "subscription_id": pending_subscription.id,
-                    "user_id": db_user.id,
-                },
+                language=db_user.language,
             )
 
-            if not payment_result or not payment_result.get("pay_url"):
+            if not payment_result or not payment_result.get("payment_url"):
                 await callback.answer("❌ Не удалось создать платеж. Попробуйте позже.", show_alert=True)
                 return
 
@@ -3940,7 +3994,7 @@ async def handle_trial_payment_method(
                     "💰 Сумма: {amount}"
                 ).format(name=mulenpay_name, amount=settings.format_price(trial_price_kopeks)),
                 reply_markup=InlineKeyboardMarkup(inline_keyboard=[
-                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["pay_url"])],
+                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["payment_url"])],
                     [InlineKeyboardButton(
                         text=texts.t("CHECK_PAYMENT", "🔄 Проверить оплату"),
                         callback_data=f"check_trial_mulenpay_{pending_subscription.id}"
@@ -3953,19 +4007,16 @@ async def handle_trial_payment_method(
         elif payment_method == "pal24":
             # Оплата через PAL24
             payment_result = await payment_service.create_pal24_payment(
+                db=db,
+                user_id=db_user.id,
                 amount_kopeks=trial_price_kopeks,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
-                metadata={
-                    "type": "trial",
-                    "subscription_id": pending_subscription.id,
-                    "user_id": db_user.id,
-                },
+                language=db_user.language,
             )
 
-            if not payment_result or not payment_result.get("pay_url"):
+            if not payment_result or not payment_result.get("payment_url"):
                 await callback.answer("❌ Не удалось создать платеж. Попробуйте позже.", show_alert=True)
                 return
 
@@ -3977,7 +4028,7 @@ async def handle_trial_payment_method(
                     "💰 Сумма: {amount}"
                 ).format(amount=settings.format_price(trial_price_kopeks)),
                 reply_markup=InlineKeyboardMarkup(inline_keyboard=[
-                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["pay_url"])],
+                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["payment_url"])],
                     [InlineKeyboardButton(
                         text=texts.t("CHECK_PAYMENT", "🔄 Проверить оплату"),
                         callback_data=f"check_trial_pal24_{pending_subscription.id}"
@@ -3990,19 +4041,16 @@ async def handle_trial_payment_method(
         elif payment_method == "wata":
             # Оплата через WATA
             payment_result = await payment_service.create_wata_payment(
+                db=db,
+                user_id=db_user.id,
                 amount_kopeks=trial_price_kopeks,
                 description=texts.t("PAID_TRIAL_PAYMENT_DESC", "Пробная подписка на {days} дней").format(
                     days=settings.TRIAL_DURATION_DAYS
                 ),
-                user_id=db_user.id,
-                metadata={
-                    "type": "trial",
-                    "subscription_id": pending_subscription.id,
-                    "user_id": db_user.id,
-                },
+                language=db_user.language,
             )
 
-            if not payment_result or not payment_result.get("pay_url"):
+            if not payment_result or not payment_result.get("payment_url"):
                 await callback.answer("❌ Не удалось создать платеж. Попробуйте позже.", show_alert=True)
                 return
 
@@ -4014,7 +4062,7 @@ async def handle_trial_payment_method(
                     "💰 Сумма: {amount}"
                 ).format(amount=settings.format_price(trial_price_kopeks)),
                 reply_markup=InlineKeyboardMarkup(inline_keyboard=[
-                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["pay_url"])],
+                    [InlineKeyboardButton(text="💳 Оплатить", url=payment_result["payment_url"])],
                     [InlineKeyboardButton(
                         text=texts.t("CHECK_PAYMENT", "🔄 Проверить оплату"),
                         callback_data=f"check_trial_wata_{pending_subscription.id}"

app/keyboards/inline.py

@@ -1471,6 +1471,16 @@ def get_payment_methods_keyboard(amount_kopeks: int, language: str = DEFAULT_LAN
         ])
         has_direct_payment_methods = True
 
+    if settings.is_kassa_ai_enabled():
+        kassa_ai_name = settings.get_kassa_ai_display_name()
+        keyboard.append([
+            InlineKeyboardButton(
+                text=texts.t("PAYMENT_KASSA_AI", f"💳 {kassa_ai_name}"),
+                callback_data=_build_callback("kassa_ai")
+            )
+        ])
+        has_direct_payment_methods = True
+
     if settings.is_support_topup_enabled():
         keyboard.append([
             InlineKeyboardButton(

app/localization/locales/en.json

@@ -1187,6 +1187,8 @@
   "PAYMENT_METHOD_YOOKASSA_NAME": "💳 <b>Bank card</b>",
   "PAYMENT_METHOD_YOOKASSA_SBP_DESCRIPTION": "via YooKassa Fast Payment System",
   "PAYMENT_METHOD_YOOKASSA_SBP_NAME": "🏦 <b>SBP (YooKassa)</b>",
+  "PAYMENT_METHOD_WATA_DESCRIPTION": "via WATA",
+  "PAYMENT_METHOD_WATA_NAME": "💳 <b>Bank card (WATA)</b>",
   "PAYMENT_HELEKET_MARKUP_LABEL": "Provider markup",
   "PAYMENT_HELEKET_DISCOUNT_LABEL": "Provider discount",
   "PAYMENT_RETURN_HOME_BUTTON": "🏠 Main menu",

app/localization/locales/ru.json

@@ -1204,6 +1204,8 @@
   "PAYMENT_METHOD_YOOKASSA_NAME": "💳 <b>Банковская карта</b>",
   "PAYMENT_METHOD_YOOKASSA_SBP_DESCRIPTION": "через систему быстрых платежей YooKassa",
   "PAYMENT_METHOD_YOOKASSA_SBP_NAME": "🏦 <b>СБП (YooKassa)</b>",
+  "PAYMENT_METHOD_WATA_DESCRIPTION": "через WATA",
+  "PAYMENT_METHOD_WATA_NAME": "💳 <b>Банковская карта (WATA)</b>",
   "PAYMENT_HELEKET_MARKUP_LABEL": "Наценка провайдера",
   "PAYMENT_HELEKET_DISCOUNT_LABEL": "Скидка провайдера",
   "PAYMENT_RETURN_HOME_BUTTON": "🏠 На главную",

app/localization/locales/ua.json

@@ -1125,6 +1125,8 @@
 "PAYMENT_METHOD_YOOKASSA_NAME": "💳 <b>Банківська картка</b>",
  "PAYMENT_METHOD_YOOKASSA_SBP_DESCRIPTION": "через систему швидких платежів YooKassa",
  "PAYMENT_METHOD_YOOKASSA_SBP_NAME": "🏦 <b>СБП (YooKassa)</b>",
+ "PAYMENT_METHOD_WATA_DESCRIPTION": "через WATA",
+ "PAYMENT_METHOD_WATA_NAME": "💳 <b>Банківська картка (WATA)</b>",
  "PAYMENT_HELEKET_MARKUP_LABEL": "Націнка провайдера",
  "PAYMENT_HELEKET_DISCOUNT_LABEL": "Знижка провайдера",
  "PAYMENT_RETURN_HOME_BUTTON": "🏠 На головну",

app/localization/locales/zh.json

@@ -1123,6 +1123,8 @@
 "PAYMENT_METHOD_YOOKASSA_NAME":"💳<b>银行卡</b>",
 "PAYMENT_METHOD_YOOKASSA_SBP_DESCRIPTION":"通过YooKassa快速支付系统",
 "PAYMENT_METHOD_YOOKASSA_SBP_NAME":"🏦<b>SBP(YooKassa)</b>",
+"PAYMENT_METHOD_WATA_DESCRIPTION":"通过WATA",
+"PAYMENT_METHOD_WATA_NAME":"💳<b>银行卡(WATA)</b>",
 "PAYMENT_HELEKET_MARKUP_LABEL":"服务商加价",
 "PAYMENT_HELEKET_DISCOUNT_LABEL":"服务商折扣",
 "PAYMENT_RETURN_HOME_BUTTON":"🏠返回首页",

app/middlewares/channel_checker.py

@@ -7,6 +7,7 @@ from aiogram.fsm.context import FSMContext
 from aiogram.types import TelegramObject, Update, Message, CallbackQuery
 from aiogram.enums import ChatMemberStatus
 from sqlalchemy.ext.asyncio import AsyncSession
+import redis.asyncio as aioredis
 
 from app.config import settings
 from app.database.database import AsyncSessionLocal
@@ -23,6 +24,58 @@ from app.services.admin_notification_service import AdminNotificationService
 
 logger = logging.getLogger(__name__)
 
+# Ключ для хранения pending_start_payload в Redis (резервный механизм)
+REDIS_PAYLOAD_KEY_PREFIX = "pending_start_payload:"
+REDIS_PAYLOAD_TTL = 3600  # 1 час
+
+
+async def save_pending_payload_to_redis(telegram_id: int, payload: str) -> bool:
+    """Сохраняет pending_start_payload в Redis напрямую (резервный механизм)."""
+    try:
+        redis_client = aioredis.from_url(settings.REDIS_URL)
+        key = f"{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}"
+        await redis_client.set(key, payload, ex=REDIS_PAYLOAD_TTL)
+        await redis_client.aclose()
+        logger.info(
+            "💾 [Redis fallback] Сохранен payload '%s' для пользователя %s",
+            payload,
+            telegram_id,
+        )
+        return True
+    except Exception as e:
+        logger.error(
+            "❌ [Redis fallback] Ошибка сохранения payload для %s: %s",
+            telegram_id,
+            e,
+        )
+        return False
+
+
+async def get_pending_payload_from_redis(telegram_id: int) -> Optional[str]:
+    """Получает pending_start_payload из Redis (резервный механизм)."""
+    try:
+        redis_client = aioredis.from_url(settings.REDIS_URL)
+        key = f"{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}"
+        payload = await redis_client.get(key)
+        await redis_client.aclose()
+        if payload:
+            return payload.decode("utf-8") if isinstance(payload, bytes) else payload
+        return None
+    except Exception as e:
+        logger.debug("❌ [Redis fallback] Ошибка получения payload для %s: %s", telegram_id, e)
+        return None
+
+
+async def delete_pending_payload_from_redis(telegram_id: int) -> None:
+    """Удаляет pending_start_payload из Redis."""
+    try:
+        redis_client = aioredis.from_url(settings.REDIS_URL)
+        key = f"{REDIS_PAYLOAD_KEY_PREFIX}{telegram_id}"
+        await redis_client.delete(key)
+        await redis_client.aclose()
+    except Exception:
+        pass
+
 
 class ChannelCheckerMiddleware(BaseMiddleware):
     """
@@ -170,8 +223,11 @@ class ChannelCheckerMiddleware(BaseMiddleware):
         event: TelegramObject,
         bot: Optional[Bot] = None,
     ) -> None:
-        if not state:
-            return
+        telegram_id = None
+        if isinstance(event, Message):
+            telegram_id = event.from_user.id if event.from_user else None
+        elif isinstance(event, CallbackQuery):
+            telegram_id = event.from_user.id if event.from_user else None
 
         message: Optional[Message] = None
         if isinstance(event, Message):
@@ -194,11 +250,26 @@ class ChannelCheckerMiddleware(BaseMiddleware):
 
         payload = parts[1]
 
-        state_data = await state.get_data() or {}
-        if state_data.get("pending_start_payload") != payload:
-            state_data["pending_start_payload"] = payload
-            await state.set_data(state_data)
-            logger.debug("💾 Сохранен start payload %s для последующей обработки", payload)
+        # Сохраняем в FSM state
+        if state:
+            state_data = await state.get_data() or {}
+            if state_data.get("pending_start_payload") != payload:
+                state_data["pending_start_payload"] = payload
+                await state.set_data(state_data)
+                logger.info(
+                    "💾 Сохранен start payload '%s' для пользователя %s (FSM)",
+                    payload,
+                    telegram_id,
+                )
+        else:
+            logger.warning(
+                "⚠️ _capture_start_payload: state=None для пользователя %s",
+                telegram_id,
+            )
+
+        # Также сохраняем в Redis как резерв (на случай потери FSM state)
+        if telegram_id:
+            await save_pending_payload_to_redis(telegram_id, payload)
 
         if bot and message.from_user:
             await self._try_send_campaign_visit_notification(

app/middlewares/subscription_checker.py

@@ -1,6 +1,6 @@
 import logging
 from typing import Callable, Dict, Any, Awaitable
-from datetime import datetime
+from datetime import datetime, timedelta
 from aiogram import BaseMiddleware
 from aiogram.types import TelegramObject
 
@@ -8,12 +8,18 @@ from app.database.models import SubscriptionStatus
 
 logger = logging.getLogger(__name__)
 
+# Буфер времени перед деактивацией (защита от race condition при продлении)
+EXPIRATION_BUFFER_MINUTES = 5
+
 
 class SubscriptionStatusMiddleware(BaseMiddleware):
     """
     Проверяет статус подписки пользователя.
     ВАЖНО: Использует db и db_user из data, которые уже загружены в AuthMiddleware.
     Не создаёт дополнительных сессий БД.
+
+    Деактивирует подписку только если она истекла более чем на EXPIRATION_BUFFER_MINUTES минут.
+    Это защищает от race conditions при продлении подписки.
     """
 
     async def __call__(
@@ -35,11 +41,26 @@ class SubscriptionStatusMiddleware(BaseMiddleware):
                     subscription.end_date and
                     subscription.end_date <= current_time):
 
-                    subscription.status = SubscriptionStatus.EXPIRED.value
-                    subscription.updated_at = current_time
-                    await db.commit()
+                    # Вычисляем насколько давно истекла подписка
+                    time_since_expiry = current_time - subscription.end_date
 
-                    logger.info(f"⏰ Middleware: Статус подписки пользователя {user.id} изменен на 'expired' (время истекло)")
+                    # Деактивируем только если прошло больше буфера (защита от race condition)
+                    if time_since_expiry > timedelta(minutes=EXPIRATION_BUFFER_MINUTES):
+                        subscription.status = SubscriptionStatus.EXPIRED.value
+                        subscription.updated_at = current_time
+                        await db.commit()
+
+                        logger.warning(
+                            f"⏰ Middleware DEACTIVATION: подписка {subscription.id} "
+                            f"(user_id={user.id}) деактивирована. "
+                            f"end_date={subscription.end_date}, просрочена на {time_since_expiry}"
+                        )
+                    else:
+                        # Подписка только что истекла - не деактивируем сразу (может быть продление)
+                        logger.debug(
+                            f"⏰ Middleware: подписка пользователя {user.id} истекла недавно "
+                            f"({time_since_expiry}), ждём буфер {EXPIRATION_BUFFER_MINUTES} мин"
+                        )
 
             except Exception as e:
                 logger.error(f"Ошибка проверки статуса подписки: {e}")

app/services/cloudpayments_service.py

@@ -8,6 +8,7 @@ import hmac
 import logging
 import time
 from typing import Any, Dict, Optional
+from urllib.parse import unquote_plus
 
 import httpx
 
@@ -278,9 +279,15 @@ class CloudPaymentsService:
         """
         Verify CloudPayments webhook signature.
 
+        CloudPayments uses two different HMAC headers:
+        - Content-HMAC: calculated from URL-encoded body (raw)
+        - X-Content-HMAC: calculated from URL-decoded body
+
+        This method tries both variants to ensure compatibility.
+
         Args:
             body: Raw request body bytes
-            signature: Signature from X-Content-HMAC header
+            signature: Signature from X-Content-HMAC or Content-HMAC header
             api_secret: CloudPayments API secret
 
         Returns:
@@ -289,15 +296,37 @@ class CloudPaymentsService:
         if not signature or not api_secret:
             return False
 
-        calculated = base64.b64encode(
-            hmac.new(
-                api_secret.encode(),
-                body,
-                hashlib.sha256,
-            ).digest()
-        ).decode()
+        def calc_hmac(data: bytes) -> str:
+            return base64.b64encode(
+                hmac.new(
+                    api_secret.encode(),
+                    data,
+                    hashlib.sha256,
+                ).digest()
+            ).decode()
 
-        return hmac.compare_digest(calculated, signature)
+        # Try with raw (URL-encoded) body first (for Content-HMAC)
+        calculated_raw = calc_hmac(body)
+        if hmac.compare_digest(calculated_raw, signature):
+            return True
+
+        # Try with URL-decoded body (for X-Content-HMAC)
+        calculated_decoded = None
+        try:
+            decoded_body = unquote_plus(body.decode("utf-8")).encode("utf-8")
+            calculated_decoded = calc_hmac(decoded_body)
+            if hmac.compare_digest(calculated_decoded, signature):
+                return True
+        except Exception:
+            pass
+
+        logger.warning(
+            "CloudPayments signature mismatch: expected_raw=%s..., expected_decoded=%s..., got=%s...",
+            calculated_raw[:20],
+            calculated_decoded[:20] if calculated_decoded else "N/A",
+            signature[:20],
+        )
+        return False
 
     @staticmethod
     def parse_webhook_data(form_data: Dict[str, Any]) -> Dict[str, Any]:

app/services/kassa_ai_service.py

@@ -0,0 +1,293 @@
+"""Сервис для работы с API KassaAI (api.fk.life)."""
+
+import hashlib
+import hmac
+import time
+import logging
+import asyncio
+from typing import Optional, Dict, Any, Set
+
+import aiohttp
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+# Кэш для публичного IP
+_cached_public_ip: Optional[str] = None
+_ip_fetch_lock = asyncio.Lock()
+
+API_BASE_URL = "https://api.fk.life/v1"
+
+# Сервисы для определения публичного IP
+IP_SERVICES = [
+    "https://api.ipify.org",
+    "https://ifconfig.me/ip",
+    "https://icanhazip.com",
+    "https://ipinfo.io/ip",
+]
+
+
+async def get_public_ip() -> str:
+    """
+    Получает публичный IP сервера.
+    1. Проверяет переменную окружения SERVER_PUBLIC_IP
+    2. Если нет - запрашивает через внешние сервисы и кэширует
+    """
+    global _cached_public_ip
+
+    env_ip = getattr(settings, 'SERVER_PUBLIC_IP', None)
+    if env_ip:
+        return env_ip
+
+    if _cached_public_ip:
+        return _cached_public_ip
+
+    async with _ip_fetch_lock:
+        if _cached_public_ip:
+            return _cached_public_ip
+
+        async with aiohttp.ClientSession() as session:
+            for service_url in IP_SERVICES:
+                try:
+                    async with session.get(
+                        service_url,
+                        timeout=aiohttp.ClientTimeout(total=5)
+                    ) as response:
+                        if response.status == 200:
+                            ip = (await response.text()).strip()
+                            if ip and len(ip.split('.')) == 4:
+                                _cached_public_ip = ip
+                                logger.info(f"KassaAI: определён публичный IP сервера: {ip}")
+                                return ip
+                except Exception as e:
+                    logger.debug(f"KassaAI: не удалось получить IP от {service_url}: {e}")
+                    continue
+
+        fallback_ip = "127.0.0.1"
+        logger.warning(f"KassaAI: не удалось определить публичный IP, используем fallback: {fallback_ip}")
+        _cached_public_ip = fallback_ip
+        return fallback_ip
+
+
+class KassaAiService:
+    """Сервис для работы с API KassaAI."""
+
+    def __init__(self):
+        self._shop_id: Optional[int] = None
+        self._api_key: Optional[str] = None
+        self._secret2: Optional[str] = None
+
+    @property
+    def shop_id(self) -> int:
+        if self._shop_id is None:
+            self._shop_id = settings.KASSA_AI_SHOP_ID
+        return self._shop_id or 0
+
+    @property
+    def api_key(self) -> str:
+        if self._api_key is None:
+            self._api_key = settings.KASSA_AI_API_KEY
+        return self._api_key or ""
+
+    @property
+    def secret2(self) -> str:
+        if self._secret2 is None:
+            self._secret2 = settings.KASSA_AI_SECRET_WORD_2
+        return self._secret2 or ""
+
+    def _generate_hmac_signature(self, params: Dict[str, Any]) -> str:
+        """
+        Генерирует подпись для API запроса (HMAC-SHA256).
+        Сортирует ключи, соединяет значения через |
+        """
+        sign_data = {k: v for k, v in params.items() if k != "signature"}
+        sorted_keys = sorted(sign_data.keys())
+        msg = "|".join(str(sign_data[k]) for k in sorted_keys)
+
+        return hmac.new(
+            self.api_key.encode("utf-8"),
+            msg.encode("utf-8"),
+            hashlib.sha256
+        ).hexdigest()
+
+    def verify_webhook_signature(
+        self, shop_id: int, amount: float, order_id: str, sign: str
+    ) -> bool:
+        """
+        Проверяет подпись webhook уведомления.
+        Формат: MD5(shop_id:amount:secret2:order_id)
+        """
+        try:
+            # Приводим amount к строке без лишних нулей
+            if isinstance(amount, float) and amount.is_integer():
+                amount_str = str(int(amount))
+            else:
+                amount_str = str(amount)
+
+            sign_str = f"{shop_id}:{amount_str}:{self.secret2}:{order_id}"
+            expected_sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest()
+
+            return expected_sign.lower() == sign.lower()
+        except Exception as e:
+            logger.error(f"KassaAI webhook verify error: {e}")
+            return False
+
+    async def create_order(
+        self,
+        order_id: str,
+        amount: float,
+        currency: str = "RUB",
+        email: Optional[str] = None,
+        ip: Optional[str] = None,
+        payment_system_id: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        """
+        Создает заказ через API KassaAI.
+        POST /orders/create
+
+        payment_system_id:
+        - 44 = СБП (QR код)
+        - 36 = Банковские карты РФ
+        - 43 = SberPay
+        """
+        # Приводим amount к int, если это целое число
+        final_amount = int(amount) if float(amount).is_integer() else amount
+
+        # Payment system из настроек или default (44 = СБП)
+        ps_id = payment_system_id or settings.KASSA_AI_PAYMENT_SYSTEM_ID or 44
+
+        # Email: используем telegram-формат если не указан
+        target_email = email or f"user_{order_id}@telegram.org"
+
+        # Определяем публичный IP сервера
+        server_ip = ip or await get_public_ip()
+
+        params = {
+            "shopId": self.shop_id,
+            "nonce": int(time.time_ns()),
+            "paymentId": str(order_id),
+            "i": ps_id,
+            "email": target_email,
+            "ip": server_ip,
+            "amount": final_amount,
+            "currency": currency,
+        }
+
+        # Генерируем подпись HMAC-SHA256
+        params["signature"] = self._generate_hmac_signature(params)
+
+        logger.info(f"KassaAI API create_order: shop_id={self.shop_id}, order_id={order_id}, amount={final_amount}, ps_id={ps_id}")
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{API_BASE_URL}/orders/create",
+                    json=params,
+                    headers={"Content-Type": "application/json"},
+                    timeout=aiohttp.ClientTimeout(total=30),
+                ) as response:
+                    text = await response.text()
+                    logger.info(f"KassaAI API response: {text}")
+
+                    data = await response.json()
+
+                    # Проверяем на ошибку
+                    if data.get("type") == "error":
+                        error_msg = data.get("error") or data.get("message") or "Unknown error"
+                        logger.error(f"KassaAI create_order error: {error_msg}")
+                        raise Exception(f"KassaAI API error: {error_msg}")
+
+                    if data.get("type") == "success":
+                        return {
+                            "location": data.get("location"),
+                            "orderId": data.get("orderId"),
+                            "paymentId": data.get("paymentId"),
+                        }
+
+                    # Неизвестный формат ответа
+                    logger.error(f"KassaAI unexpected response: {data}")
+                    raise Exception(f"KassaAI unexpected response format")
+
+        except aiohttp.ClientError as e:
+            logger.exception(f"KassaAI API connection error: {e}")
+            raise
+
+    async def create_order_and_get_url(
+        self,
+        order_id: str,
+        amount: float,
+        currency: str = "RUB",
+        email: Optional[str] = None,
+        ip: Optional[str] = None,
+        payment_system_id: Optional[int] = None,
+    ) -> str:
+        """
+        Создает заказ через API и возвращает URL для оплаты.
+        """
+        result = await self.create_order(
+            order_id=order_id,
+            amount=amount,
+            currency=currency,
+            email=email,
+            ip=ip,
+            payment_system_id=payment_system_id,
+        )
+        location = result.get("location")
+        if not location:
+            raise Exception("KassaAI API did not return payment URL (location)")
+        return location
+
+    async def get_order_status(self, order_id: str) -> Dict[str, Any]:
+        """
+        Получает статус заказа.
+        POST /orders
+        """
+        params = {
+            "shopId": self.shop_id,
+            "nonce": int(time.time_ns()),
+            "paymentId": str(order_id),
+        }
+        params["signature"] = self._generate_hmac_signature(params)
+
+        logger.debug(f"KassaAI get_order_status: order_id={order_id}")
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{API_BASE_URL}/orders",
+                    json=params,
+                    headers={"Content-Type": "application/json"},
+                    timeout=aiohttp.ClientTimeout(total=30),
+                ) as response:
+                    text = await response.text()
+                    logger.debug(f"KassaAI get_order_status response: {text}")
+                    return await response.json()
+        except aiohttp.ClientError as e:
+            logger.exception(f"KassaAI API connection error: {e}")
+            raise
+
+    async def get_balance(self) -> Dict[str, Any]:
+        """Получает баланс магазина."""
+        params = {
+            "shopId": self.shop_id,
+            "nonce": int(time.time_ns()),
+        }
+        params["signature"] = self._generate_hmac_signature(params)
+
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    f"{API_BASE_URL}/balance",
+                    json=params,
+                    headers={"Content-Type": "application/json"},
+                    timeout=aiohttp.ClientTimeout(total=30),
+                ) as response:
+                    return await response.json()
+        except aiohttp.ClientError as e:
+            logger.exception(f"KassaAI API connection error: {e}")
+            raise
+
+
+# Singleton instance
+kassa_ai_service = KassaAiService()

app/services/payment/__init__.py

@@ -16,6 +16,7 @@ from .platega import PlategaPaymentMixin
 from .wata import WataPaymentMixin
 from .cloudpayments import CloudPaymentsPaymentMixin
 from .freekassa import FreekassaPaymentMixin
+from .kassa_ai import KassaAiPaymentMixin
 
 __all__ = [
     "PaymentCommonMixin",
@@ -30,4 +31,5 @@ __all__ = [
     "WataPaymentMixin",
     "CloudPaymentsPaymentMixin",
     "FreekassaPaymentMixin",
+    "KassaAiPaymentMixin",
 ]

app/services/payment/kassa_ai.py

@@ -0,0 +1,490 @@
+"""Mixin для интеграции с KassaAI (api.fk.life)."""
+
+from __future__ import annotations
+
+import json
+import uuid
+import logging
+from datetime import datetime, timedelta
+from importlib import import_module
+from typing import Any, Dict, Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.database.models import PaymentMethod, TransactionType
+from app.services.kassa_ai_service import kassa_ai_service
+from app.services.subscription_auto_purchase_service import (
+    auto_activate_subscription_after_topup,
+    auto_purchase_saved_cart_after_topup,
+)
+from app.utils.user_utils import format_referrer_info
+from app.utils.payment_logger import payment_logger as logger
+
+
+class KassaAiPaymentMixin:
+    """Mixin для работы с платежами KassaAI."""
+
+    async def create_kassa_ai_payment(
+        self,
+        db: AsyncSession,
+        *,
+        user_id: int,
+        amount_kopeks: int,
+        description: str = "Пополнение баланса",
+        email: Optional[str] = None,
+        language: str = "ru",
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Создает платеж KassaAI.
+
+        Args:
+            db: Сессия БД
+            user_id: ID пользователя
+            amount_kopeks: Сумма в копейках
+            description: Описание платежа
+            email: Email пользователя
+            language: Язык интерфейса
+
+        Returns:
+            Словарь с данными платежа или None при ошибке
+        """
+        if not settings.is_kassa_ai_enabled():
+            logger.error("KassaAI не настроен")
+            return None
+
+        # Валидация лимитов
+        if amount_kopeks < settings.KASSA_AI_MIN_AMOUNT_KOPEKS:
+            logger.warning(
+                "KassaAI: сумма %s меньше минимальной %s",
+                amount_kopeks,
+                settings.KASSA_AI_MIN_AMOUNT_KOPEKS,
+            )
+            return None
+
+        if amount_kopeks > settings.KASSA_AI_MAX_AMOUNT_KOPEKS:
+            logger.warning(
+                "KassaAI: сумма %s больше максимальной %s",
+                amount_kopeks,
+                settings.KASSA_AI_MAX_AMOUNT_KOPEKS,
+            )
+            return None
+
+        # Генерируем уникальный order_id
+        order_id = f"kai_{user_id}_{uuid.uuid4().hex[:12]}"
+        amount_rubles = amount_kopeks / 100
+        currency = settings.KASSA_AI_CURRENCY
+
+        # Срок действия платежа (1 час по умолчанию)
+        expires_at = datetime.utcnow() + timedelta(hours=1)
+
+        # Метаданные
+        metadata = {
+            "user_id": user_id,
+            "amount_kopeks": amount_kopeks,
+            "description": description,
+            "language": language,
+            "type": "balance_topup",
+        }
+
+        try:
+            # Используем API для создания заказа
+            result = await kassa_ai_service.create_order(
+                order_id=order_id,
+                amount=amount_rubles,
+                currency=currency,
+                email=email,
+                payment_system_id=settings.KASSA_AI_PAYMENT_SYSTEM_ID,
+            )
+
+            payment_url = result.get("location")
+            if not payment_url:
+                logger.error("KassaAI API не вернул URL платежа")
+                return None
+
+            logger.info(
+                "KassaAI API: создан заказ order_id=%s, url=%s",
+                order_id,
+                payment_url,
+            )
+
+            # Импортируем CRUD модуль
+            kassa_ai_crud = import_module("app.database.crud.kassa_ai")
+
+            # Сохраняем в БД
+            local_payment = await kassa_ai_crud.create_kassa_ai_payment(
+                db=db,
+                user_id=user_id,
+                order_id=order_id,
+                amount_kopeks=amount_kopeks,
+                currency=currency,
+                description=description,
+                payment_url=payment_url,
+                payment_system_id=settings.KASSA_AI_PAYMENT_SYSTEM_ID,
+                expires_at=expires_at,
+                metadata_json=json.dumps(metadata, ensure_ascii=False),
+            )
+
+            logger.info(
+                "KassaAI: создан платеж order_id=%s, user_id=%s, amount=%s %s",
+                order_id,
+                user_id,
+                amount_rubles,
+                currency,
+            )
+
+            return {
+                "order_id": order_id,
+                "amount_kopeks": amount_kopeks,
+                "amount_rubles": amount_rubles,
+                "currency": currency,
+                "payment_url": payment_url,
+                "expires_at": expires_at.isoformat(),
+                "local_payment_id": local_payment.id,
+            }
+
+        except Exception as e:
+            logger.exception("KassaAI: ошибка создания платежа: %s", e)
+            return None
+
+    async def process_kassa_ai_webhook(
+        self,
+        db: AsyncSession,
+        *,
+        merchant_id: int,
+        amount: float,
+        order_id: str,
+        sign: str,
+        intid: str,
+        cur_id: Optional[int] = None,
+    ) -> bool:
+        """
+        Обрабатывает webhook от KassaAI.
+
+        Args:
+            db: Сессия БД
+            merchant_id: ID магазина (MERCHANT_ID)
+            amount: Сумма платежа (AMOUNT)
+            order_id: Номер заказа (MERCHANT_ORDER_ID)
+            sign: Подпись (SIGN)
+            intid: ID транзакции KassaAI
+            cur_id: ID валюты/платежной системы (CUR_ID)
+
+        Returns:
+            True если платеж успешно обработан
+        """
+        try:
+            # Проверка подписи
+            if not kassa_ai_service.verify_webhook_signature(
+                merchant_id, amount, order_id, sign
+            ):
+                logger.warning(
+                    "KassaAI webhook: неверная подпись для order_id=%s", order_id
+                )
+                return False
+
+            # Импортируем CRUD модуль
+            kassa_ai_crud = import_module("app.database.crud.kassa_ai")
+
+            # Получаем платеж из БД
+            payment = await kassa_ai_crud.get_kassa_ai_payment_by_order_id(
+                db, order_id
+            )
+            if not payment:
+                logger.warning(
+                    "KassaAI webhook: платеж не найден order_id=%s", order_id
+                )
+                return False
+
+            # Проверка дублирования
+            if payment.is_paid:
+                logger.info(
+                    "KassaAI webhook: платеж уже обработан order_id=%s", order_id
+                )
+                return True
+
+            # Проверка суммы
+            expected_amount = payment.amount_kopeks / 100
+            if abs(amount - expected_amount) > 0.01:
+                logger.warning(
+                    "KassaAI webhook: несоответствие суммы ожидалось=%s, получено=%s",
+                    expected_amount,
+                    amount,
+                )
+                return False
+
+            # Обновляем статус платежа
+            callback_payload = {
+                "merchant_id": merchant_id,
+                "amount": amount,
+                "order_id": order_id,
+                "intid": intid,
+                "cur_id": cur_id,
+            }
+
+            payment = await kassa_ai_crud.update_kassa_ai_payment_status(
+                db=db,
+                payment=payment,
+                status="success",
+                is_paid=True,
+                kassa_ai_order_id=intid,
+                payment_system_id=cur_id,
+                callback_payload=callback_payload,
+            )
+
+            # Финализируем платеж (начисляем баланс, создаем транзакцию)
+            return await self._finalize_kassa_ai_payment(
+                db, payment, intid=intid, trigger="webhook"
+            )
+
+        except Exception as e:
+            logger.exception("KassaAI webhook: ошибка обработки: %s", e)
+            return False
+
+    async def _finalize_kassa_ai_payment(
+        self,
+        db: AsyncSession,
+        payment: Any,
+        *,
+        intid: Optional[str],
+        trigger: str,
+    ) -> bool:
+        """Создаёт транзакцию, начисляет баланс и отправляет уведомления."""
+        payment_module = import_module("app.services.payment_service")
+
+        if payment.transaction_id:
+            logger.info(
+                "KassaAI платеж %s уже привязан к транзакции (trigger=%s)",
+                payment.order_id,
+                trigger,
+            )
+            return True
+
+        # Получаем пользователя
+        user = await payment_module.get_user_by_id(db, payment.user_id)
+        if not user:
+            logger.error(
+                "Пользователь %s не найден для KassaAI платежа %s (trigger=%s)",
+                payment.user_id,
+                payment.order_id,
+                trigger,
+            )
+            return False
+
+        # Создаем транзакцию
+        transaction = await payment_module.create_transaction(
+            db,
+            user_id=payment.user_id,
+            type=TransactionType.DEPOSIT,
+            amount_kopeks=payment.amount_kopeks,
+            description=f"Пополнение через KassaAI (#{intid or payment.order_id})",
+            payment_method=PaymentMethod.KASSA_AI,
+            external_id=str(intid) if intid else payment.order_id,
+            is_completed=True,
+        )
+
+        # Связываем платеж с транзакцией
+        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,
+        )
+
+        old_balance = user.balance_kopeks
+        was_first_topup = not user.has_made_first_topup
+
+        # Начисляем баланс
+        user.balance_kopeks += payment.amount_kopeks
+        user.updated_at = datetime.utcnow()
+
+        promo_group = user.get_primary_promo_group()
+        subscription = getattr(user, "subscription", None)
+        referrer_info = format_referrer_info(user)
+        topup_status = "Первое пополнение" if was_first_topup else "Пополнение"
+
+        await db.commit()
+
+        # Обработка реферального пополнения
+        try:
+            from app.services.referral_service import process_referral_topup
+
+            await process_referral_topup(
+                db, user.id, payment.amount_kopeks, getattr(self, "bot", None)
+            )
+        except Exception as error:
+            logger.error(
+                "Ошибка обработки реферального пополнения KassaAI: %s", error
+            )
+
+        if was_first_topup and not user.has_made_first_topup:
+            user.has_made_first_topup = True
+            await db.commit()
+
+        await db.refresh(user)
+        await db.refresh(payment)
+
+        # Отправка уведомления админам
+        if getattr(self, "bot", None):
+            try:
+                from app.services.admin_notification_service import (
+                    AdminNotificationService,
+                )
+
+                notification_service = AdminNotificationService(self.bot)
+                await notification_service.send_balance_topup_notification(
+                    user,
+                    transaction,
+                    old_balance,
+                    topup_status=topup_status,
+                    referrer_info=referrer_info,
+                    subscription=subscription,
+                    promo_group=promo_group,
+                    db=db,
+                )
+            except Exception as error:
+                logger.error(
+                    "Ошибка отправки админ уведомления KassaAI: %s", error
+                )
+
+        # Отправка уведомления пользователю
+        if getattr(self, "bot", None):
+            try:
+                keyboard = await self.build_topup_success_keyboard(user)
+                display_name = settings.get_kassa_ai_display_name()
+                await self.bot.send_message(
+                    user.telegram_id,
+                    (
+                        "✅ <b>Пополнение успешно!</b>\n\n"
+                        f"💰 Сумма: {settings.format_price(payment.amount_kopeks)}\n"
+                        f"💳 Способ: {display_name}\n"
+                        f"🆔 Транзакция: {transaction.id}\n\n"
+                        "Баланс пополнен автоматически!"
+                    ),
+                    parse_mode="HTML",
+                    reply_markup=keyboard,
+                )
+            except Exception as error:
+                logger.error(
+                    "Ошибка отправки уведомления пользователю KassaAI: %s", error
+                )
+
+        # Автопокупка подписки
+        try:
+            from app.services.user_cart_service import user_cart_service
+            from aiogram import types
+
+            has_saved_cart = await user_cart_service.has_user_cart(user.id)
+            auto_purchase_success = False
+
+            if has_saved_cart:
+                try:
+                    auto_purchase_success = await auto_purchase_saved_cart_after_topup(
+                        db,
+                        user,
+                        bot=getattr(self, "bot", None),
+                    )
+                except Exception as auto_error:
+                    logger.error(
+                        "Ошибка автоматической покупки подписки для пользователя %s: %s",
+                        user.id,
+                        auto_error,
+                        exc_info=True,
+                    )
+
+                if auto_purchase_success:
+                    has_saved_cart = False
+
+            # Умная автоактивация если автопокупка не сработала
+            activation_notification_sent = False
+            if not auto_purchase_success:
+                try:
+                    _, activation_notification_sent = await auto_activate_subscription_after_topup(
+                        db, user, bot=getattr(self, "bot", None), topup_amount=payment.amount_kopeks
+                    )
+                except Exception as auto_activate_error:
+                    logger.error(
+                        "Ошибка умной автоактивации для пользователя %s: %s",
+                        user.id,
+                        auto_activate_error,
+                        exc_info=True,
+                    )
+
+            # Отправляем уведомление только если его ещё не отправили
+            if has_saved_cart and getattr(self, "bot", None) and not activation_notification_sent:
+                from app.localization.texts import get_texts
+
+                texts = get_texts(user.language)
+                cart_message = texts.t(
+                    "BALANCE_TOPUP_CART_REMINDER",
+                    "У вас есть незавершенное оформление подписки. Вернуться?",
+                )
+
+                keyboard = types.InlineKeyboardMarkup(
+                    inline_keyboard=[
+                        [
+                            types.InlineKeyboardButton(
+                                text=texts.t(
+                                    "BALANCE_TOPUP_CART_BUTTON",
+                                    "🛒 Продолжить оформление",
+                                ),
+                                callback_data="return_to_saved_cart",
+                            )
+                        ],
+                        [
+                            types.InlineKeyboardButton(
+                                text="🏠 Главное меню",
+                                callback_data="back_to_menu",
+                            )
+                        ],
+                    ]
+                )
+
+                await self.bot.send_message(
+                    chat_id=user.telegram_id,
+                    text=(
+                        "✅ Баланс пополнен на "
+                        f"{settings.format_price(payment.amount_kopeks)}!\n\n"
+                        f"{cart_message}"
+                    ),
+                    reply_markup=keyboard,
+                )
+        except Exception as error:
+            logger.error(
+                "Ошибка при работе с сохраненной корзиной для пользователя %s: %s",
+                user.id,
+                error,
+                exc_info=True,
+            )
+
+        logger.info(
+            "✅ Обработан KassaAI платеж %s для пользователя %s (trigger=%s)",
+            payment.order_id,
+            payment.user_id,
+            trigger,
+        )
+
+        return True
+
+    async def check_kassa_ai_payment_status(
+        self,
+        db: AsyncSession,
+        order_id: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Проверяет статус платежа через API.
+
+        Args:
+            db: Сессия БД
+            order_id: Номер заказа
+
+        Returns:
+            Данные о статусе платежа
+        """
+        try:
+            status_data = await kassa_ai_service.get_order_status(order_id)
+            return status_data
+        except Exception as e:
+            logger.exception("KassaAI: ошибка проверки статуса: %s", e)
+            return None

app/services/payment_service.py

@@ -30,6 +30,7 @@ from app.services.payment import (
 )
 from app.services.payment.cloudpayments import CloudPaymentsPaymentMixin
 from app.services.payment.freekassa import FreekassaPaymentMixin
+from app.services.payment.kassa_ai import KassaAiPaymentMixin
 from app.services.yookassa_service import YooKassaService
 from app.services.wata_service import WataService
 from app.services.cloudpayments_service import CloudPaymentsService
@@ -171,6 +172,11 @@ async def get_wata_payment_by_id(*args, **kwargs):
     return await wata_crud.get_wata_payment_by_id(*args, **kwargs)
 
 
+# Алиас для совместимости с хендлерами
+async def get_wata_payment_by_local_id(*args, **kwargs):
+    return await get_wata_payment_by_id(*args, **kwargs)
+
+
 async def get_wata_payment_by_order_id(*args, **kwargs):
     wata_crud = import_module("app.database.crud.wata")
     return await wata_crud.get_wata_payment_by_order_id(*args, **kwargs)
@@ -299,6 +305,7 @@ class PaymentService(
     WataPaymentMixin,
     CloudPaymentsPaymentMixin,
     FreekassaPaymentMixin,
+    KassaAiPaymentMixin,
 ):
     """Основной интерфейс платежей, делегирующий работу специализированным mixin-ам."""
 

app/services/remnawave_service.py

@@ -1430,13 +1430,30 @@ class RemnaWaveService:
                                 logger.warning(f"⚠️ Не удалось удалить серверы подписки: {servers_error}")
                             
                             from app.database.models import SubscriptionStatus
-                            
+
+                            # Проверяем, была ли это платная подписка
+                            was_paid = (
+                                not subscription.is_trial
+                                or getattr(db_user, 'has_had_paid_subscription', False)
+                            )
+
                             subscription.status = SubscriptionStatus.DISABLED.value
-                            subscription.is_trial = True 
-                            subscription.end_date = datetime.utcnow()
-                            subscription.traffic_limit_gb = 0
-                            subscription.traffic_used_gb = 0.0
-                            subscription.device_limit = 1
+
+                            if was_paid:
+                                # Для платных подписок - НЕ сбрасываем is_trial и end_date!
+                                # Сохраняем оригинальные значения чтобы можно было восстановить
+                                logger.warning(
+                                    f"⚠️ ПЛАТНАЯ подписка пользователя {telegram_id} отключена (нет в панели), "
+                                    f"но is_trial={subscription.is_trial} и end_date={subscription.end_date} СОХРАНЕНЫ"
+                                )
+                            else:
+                                # Для триальных подписок - сбрасываем как раньше
+                                subscription.is_trial = True
+                                subscription.end_date = datetime.utcnow()
+                                subscription.traffic_limit_gb = 0
+                                subscription.traffic_used_gb = 0.0
+                                subscription.device_limit = 1
+
                             subscription.connected_squads = []
                             subscription.autopay_enabled = False
                             subscription.remnawave_short_uuid = None
@@ -1620,21 +1637,43 @@ class RemnaWaveService:
             
             if expire_at_str:
                 expire_at = self._parse_remnawave_date(expire_at_str)
-                
-                if abs((subscription.end_date - expire_at).total_seconds()) > 60: 
-                    subscription.end_date = expire_at
-                    logger.debug(f"Обновлена дата окончания подписки до {expire_at}")
+
+                # КРИТИЧНО: НЕ перезаписываем end_date если локальная дата ПОЗЖЕ
+                # Это защищает от ситуации когда подписка была продлена в боте,
+                # но RemnaWave ещё не получил обновление или вернул старую дату
+                if abs((subscription.end_date - expire_at).total_seconds()) > 60:
+                    if expire_at > subscription.end_date:
+                        # RemnaWave имеет более позднюю дату - обновляем
+                        subscription.end_date = expire_at
+                        logger.debug(f"Обновлена дата окончания подписки до {expire_at}")
+                    else:
+                        # Локальная дата позже - НЕ перезаписываем, логируем предупреждение
+                        logger.warning(
+                            f"⚠️ Sync: пропускаем обновление end_date для user {getattr(user, 'telegram_id', '?')}: "
+                            f"локальная дата ({subscription.end_date}) позже чем в RemnaWave ({expire_at})"
+                        )
             
             current_time = self._now_utc()
             if panel_status == 'ACTIVE' and subscription.end_date > current_time:
                 new_status = SubscriptionStatus.ACTIVE.value
-            elif subscription.end_date <= current_time:
-                new_status = SubscriptionStatus.EXPIRED.value
             elif panel_status == 'DISABLED':
                 new_status = SubscriptionStatus.DISABLED.value
+            elif subscription.end_date <= current_time:
+                # КРИТИЧНО: НЕ деактивируем если текущий статус ACTIVE
+                # Это защищает от race condition когда sync использует старую end_date из памяти,
+                # а реальная end_date уже обновлена продлением
+                if subscription.status == SubscriptionStatus.ACTIVE.value:
+                    logger.warning(
+                        f"⚠️ Sync: пропускаем деактивацию подписки user {getattr(user, 'telegram_id', '?')}: "
+                        f"статус ACTIVE, end_date в памяти ({subscription.end_date}) <= now. "
+                        f"Деактивация будет выполнена через middleware с буфером."
+                    )
+                    new_status = subscription.status  # Сохраняем текущий статус
+                else:
+                    new_status = SubscriptionStatus.EXPIRED.value
             else:
-                new_status = subscription.status 
-            
+                new_status = subscription.status
+
             if subscription.status != new_status:
                 subscription.status = new_status
                 logger.debug(f"Обновлен статус подписки: {new_status}")
@@ -2143,7 +2182,25 @@ class RemnaWaveService:
             return False
 
     async def force_cleanup_user_data(self, db: AsyncSession, user: User) -> bool:
+        """
+        ОПАСНАЯ ФУНКЦИЯ: Полностью сбрасывает все данные пользователя включая баланс!
+        Используйте только для полной очистки пользователя.
+        """
         try:
+            # Предупреждение для платных пользователей
+            was_paid = (
+                user.has_had_paid_subscription
+                or (user.subscription and not user.subscription.is_trial)
+                or user.balance_kopeks > 0
+            )
+            if was_paid:
+                logger.warning(
+                    f"⚠️ ВНИМАНИЕ: force_cleanup_user_data вызвана для ПЛАТНОГО пользователя {user.telegram_id}! "
+                    f"has_had_paid_subscription={user.has_had_paid_subscription}, "
+                    f"balance={user.balance_kopeks}, "
+                    f"is_trial={user.subscription.is_trial if user.subscription else 'N/A'}"
+                )
+
             logger.info(f"🗑️ ПРИНУДИТЕЛЬНАЯ полная очистка данных пользователя {user.telegram_id}")
             
             if user.remnawave_uuid:
@@ -2384,8 +2441,15 @@ class RemnaWaveService:
                         issues_fixed = 0
                     
                         current_time = self._now_utc()
-                        if subscription.end_date <= current_time and subscription.status == SubscriptionStatus.ACTIVE.value:
-                            logger.info(f"🔧 Исправляем статус просроченной подписки {user.telegram_id}")
+                        # Добавляем буфер 5 минут для защиты от race condition при продлении
+                        expiry_buffer = timedelta(minutes=5)
+                        if (subscription.end_date + expiry_buffer <= current_time and
+                            subscription.status == SubscriptionStatus.ACTIVE.value):
+                            time_since_expiry = current_time - subscription.end_date
+                            logger.warning(
+                                f"🔧 fix_data_issues: деактивируем подписку {subscription.id} "
+                                f"(user={user.telegram_id}), просрочена на {time_since_expiry}"
+                            )
                             subscription.status = SubscriptionStatus.EXPIRED.value
                             issues_fixed += 1
                 

app/services/reporting_service.py

@@ -16,6 +16,7 @@ from sqlalchemy.sql import false, true
 
 from app.config import settings
 from app.database.crud.subscription import get_subscriptions_statistics
+from app.database.crud.transaction import REAL_PAYMENT_METHODS
 from app.database.database import AsyncSessionLocal
 from app.database.models import (
     PaymentMethod,
@@ -491,6 +492,10 @@ class ReportingService:
         )
 
     def _deposit_query_excluding_referrals(self, start_utc: datetime, end_utc: datetime):
+        """Запрос депозитов только по реальным платежам.
+
+        Исключаются: колесо удачи, промокоды, админские пополнения, оплата с баланса.
+        """
         return select(
             func.count(Transaction.id),
             func.coalesce(func.sum(Transaction.amount_kopeks), 0),
@@ -500,11 +505,8 @@ class ReportingService:
             Transaction.created_at >= start_utc,
             Transaction.created_at < end_utc,
             self._exclude_referral_deposits_condition(),
-            # Исключаем ручные (админские) пополнения из статистики
-            or_(
-                Transaction.payment_method.is_(None),
-                Transaction.payment_method != PaymentMethod.MANUAL.value,
-            ),
+            # Только реальные платежи (исключаем колесо, промокоды, админские, баланс)
+            Transaction.payment_method.in_(REAL_PAYMENT_METHODS),
         )
 
     async def _get_top_referrers(

app/services/subscription_service.py

@@ -328,17 +328,19 @@ class SubscriptionService:
                 pass  # tariff может быть None или уже загружен
 
             current_time = datetime.utcnow()
+            # Определяем актуальный статус для отправки в RemnaWave
+            # НЕ меняем статус подписки здесь - это задача scheduled job
             is_actually_active = (subscription.status == SubscriptionStatus.ACTIVE.value and
                                  subscription.end_date > current_time)
 
+            # Логируем если статус и end_date не согласованы (для отладки)
             if (subscription.status == SubscriptionStatus.ACTIVE.value and
                 subscription.end_date <= current_time):
-
-                subscription.status = SubscriptionStatus.EXPIRED.value
-                subscription.updated_at = current_time
-                await db.commit()
-                is_actually_active = False
-                logger.info(f"🔔 Статус подписки {subscription.id} автоматически изменен на 'expired'")
+                logger.warning(
+                    f"⚠️ update_remnawave_user: подписка {subscription.id} имеет статус ACTIVE, "
+                    f"но end_date ({subscription.end_date}) <= now ({current_time}). "
+                    f"Отправляем в RemnaWave как EXPIRED, но НЕ меняем статус в БД."
+                )
 
             user_tag = self._resolve_user_tag(subscription)
 

app/services/system_settings_service.py

@@ -85,6 +85,7 @@ class BotConfigurationService:
         "HELEKET": "🪙 Heleket",
         "CLOUDPAYMENTS": "💳 CloudPayments",
         "FREEKASSA": "💳 Freekassa",
+        "KASSA_AI": "💳 KassaAI",
         "YOOKASSA": "🟣 YooKassa",
         "PLATEGA": "💳 {platega_name}",
         "TRIBUTE": "🎁 Tribute",
@@ -143,6 +144,7 @@ class BotConfigurationService:
         "HELEKET": "Heleket: криптоплатежи, ключи мерчанта и вебхуки.",
         "CLOUDPAYMENTS": "CloudPayments: оплата банковскими картами, Public ID, API Secret и вебхуки.",
         "FREEKASSA": "Freekassa: ID магазина, API ключ, секретные слова и вебхуки.",
+        "KASSA_AI": "KassaAI: отдельная платёжка api.fk.life с СБП, картами и SberPay.",
         "PLATEGA": "{platega_name}: merchant ID, секрет, ссылки возврата и методы оплаты.",
         "MULENPAY": "Платежи {mulenpay_name} и параметры магазина.",
         "PAL24": "PAL24 / PayPalych подключения и лимиты.",
@@ -337,6 +339,7 @@ class BotConfigurationService:
         "HELEKET_": "HELEKET",
         "CLOUDPAYMENTS_": "CLOUDPAYMENTS",
         "FREEKASSA_": "FREEKASSA",
+        "KASSA_AI_": "KASSA_AI",
         "PLATEGA_": "PLATEGA",
         "MULENPAY_": "MULENPAY",
         "PAL24_": "PAL24",

app/webserver/payments.py

@@ -679,18 +679,39 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
         @router.post(settings.CLOUDPAYMENTS_WEBHOOK_PATH + "/check")
         async def cloudpayments_check_webhook(request: Request) -> JSONResponse:
             """Check webhook - вызывается перед списанием, можно отклонить платёж."""
-            raw_body = await request.body()
+            try:
+                raw_body = await request.body()
 
-            # Проверяем подпись
-            signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
-            if settings.CLOUDPAYMENTS_API_SECRET and not cloudpayments_service.verify_webhook_signature(
-                raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
-            ):
-                logger.warning("CloudPayments check webhook: invalid signature")
-                return JSONResponse({"code": 13})  # Отклонить
+                # Логируем для диагностики
+                logger.info(
+                    "CloudPayments check webhook received, body_len=%d, all_headers=%s",
+                    len(raw_body),
+                    dict(request.headers),
+                )
 
-            # Разрешаем платёж
-            return JSONResponse({"code": 0})
+                # Проверяем подпись только если она пришла и API_SECRET настроен
+                # CloudPayments использует заголовок X-Content-HMAC или Content-HMAC
+                signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
+                if settings.CLOUDPAYMENTS_API_SECRET and signature:
+                    if not cloudpayments_service.verify_webhook_signature(
+                        raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
+                    ):
+                        logger.warning(
+                            "CloudPayments check webhook: invalid signature, sig=%s...",
+                            signature[:20] if signature else "empty",
+                        )
+                        return JSONResponse({"code": 13})  # Отклонить
+                elif settings.CLOUDPAYMENTS_API_SECRET and not signature:
+                    # Подпись не пришла, но API_SECRET настроен - пропускаем проверку с предупреждением
+                    logger.warning("CloudPayments check webhook: no signature header, skipping verification")
+
+                # Разрешаем платёж
+                logger.info("CloudPayments check webhook: allowing payment, returning code=0")
+                return JSONResponse({"code": 0})
+            except Exception as e:
+                logger.exception("CloudPayments check webhook error: %s", e)
+                # В случае ошибки всё равно разрешаем платёж
+                return JSONResponse({"code": 0})
 
         # CloudPayments Pay webhook (успешная оплата)
         @router.post(settings.CLOUDPAYMENTS_WEBHOOK_PATH + "/pay")
@@ -698,13 +719,14 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
             """Pay webhook - вызывается после успешной оплаты."""
             raw_body = await request.body()
 
-            # Проверяем подпись
+            # Проверяем подпись только если она пришла и API_SECRET настроен
             signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
-            if settings.CLOUDPAYMENTS_API_SECRET and not cloudpayments_service.verify_webhook_signature(
-                raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
-            ):
-                logger.warning("CloudPayments pay webhook: invalid signature")
-                return JSONResponse({"code": 13})
+            if settings.CLOUDPAYMENTS_API_SECRET and signature:
+                if not cloudpayments_service.verify_webhook_signature(
+                    raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
+                ):
+                    logger.warning("CloudPayments pay webhook: invalid signature")
+                    return JSONResponse({"code": 13})
 
             # Парсим данные формы
             try:
@@ -729,13 +751,14 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
             """Fail webhook - вызывается при неуспешной оплате."""
             raw_body = await request.body()
 
-            # Проверяем подпись
+            # Проверяем подпись только если она пришла и API_SECRET настроен
             signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
-            if settings.CLOUDPAYMENTS_API_SECRET and not cloudpayments_service.verify_webhook_signature(
-                raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
-            ):
-                logger.warning("CloudPayments fail webhook: invalid signature")
-                return JSONResponse({"code": 13})
+            if settings.CLOUDPAYMENTS_API_SECRET and signature:
+                if not cloudpayments_service.verify_webhook_signature(
+                    raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
+                ):
+                    logger.warning("CloudPayments fail webhook: invalid signature")
+                    return JSONResponse({"code": 13})
 
             # Парсим данные формы
             try:
@@ -758,43 +781,60 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
         @router.post(settings.CLOUDPAYMENTS_WEBHOOK_PATH)
         async def cloudpayments_webhook(request: Request) -> JSONResponse:
             """Универсальный webhook endpoint."""
-            raw_body = await request.body()
-
-            # Проверяем подпись
-            signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
-            if settings.CLOUDPAYMENTS_API_SECRET and not cloudpayments_service.verify_webhook_signature(
-                raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
-            ):
-                logger.warning("CloudPayments webhook: invalid signature")
-                return JSONResponse({"code": 13})
-
-            # Парсим данные формы
             try:
-                form_data = await request.form()
-                webhook_data = cloudpayments_service.parse_webhook_data(dict(form_data))
-            except Exception as error:
-                logger.error("CloudPayments webhook parse error: %s", error)
+                raw_body = await request.body()
+
+                # Логируем для диагностики
+                logger.info(
+                    "CloudPayments universal webhook received, body_len=%d, headers=%s",
+                    len(raw_body),
+                    dict(request.headers),
+                )
+
+                # Проверяем подпись только если она пришла и API_SECRET настроен
+                signature = request.headers.get("X-Content-HMAC") or request.headers.get("Content-HMAC") or ""
+                if settings.CLOUDPAYMENTS_API_SECRET and signature:
+                    if not cloudpayments_service.verify_webhook_signature(
+                        raw_body, signature, settings.CLOUDPAYMENTS_API_SECRET
+                    ):
+                        logger.warning("CloudPayments webhook: invalid signature")
+                        return JSONResponse({"code": 13})
+
+                # Парсим данные формы
+                try:
+                    form_data = await request.form()
+                    webhook_data = cloudpayments_service.parse_webhook_data(dict(form_data))
+                    logger.info("CloudPayments webhook parsed data: %s", webhook_data)
+                except Exception as error:
+                    logger.error("CloudPayments webhook parse error: %s", error)
+                    # Может быть это Check уведомление - просто разрешаем
+                    return JSONResponse({"code": 0})
+
+                # Определяем тип webhook по статусу
+                status_value = webhook_data.get("status", "")
+
+                if status_value in ("Completed", "Authorized"):
+                    # Успешная оплата
+                    await _process_payment_service_callback(
+                        payment_service,
+                        webhook_data,
+                        "process_cloudpayments_pay_webhook",
+                    )
+                elif status_value in ("Declined", "Cancelled"):
+                    # Неуспешная оплата
+                    await _process_payment_service_callback(
+                        payment_service,
+                        webhook_data,
+                        "process_cloudpayments_fail_webhook",
+                    )
+                else:
+                    # Check или другой тип уведомления - просто разрешаем
+                    logger.info("CloudPayments webhook: status=%s, allowing (code=0)", status_value)
+
+                return JSONResponse({"code": 0})
+            except Exception as e:
+                logger.exception("CloudPayments universal webhook error: %s", e)
                 return JSONResponse({"code": 0})
-
-            # Определяем тип webhook по статусу
-            status_value = webhook_data.get("status", "")
-
-            if status_value in ("Completed", "Authorized"):
-                # Успешная оплата
-                await _process_payment_service_callback(
-                    payment_service,
-                    webhook_data,
-                    "process_cloudpayments_pay_webhook",
-                )
-            elif status_value in ("Declined", "Cancelled"):
-                # Неуспешная оплата
-                await _process_payment_service_callback(
-                    payment_service,
-                    webhook_data,
-                    "process_cloudpayments_fail_webhook",
-                )
-
-            return JSONResponse({"code": 0})
 
         routes_registered = True
 
@@ -892,6 +932,77 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
 
         routes_registered = True
 
+    # KassaAI webhook
+    if settings.is_kassa_ai_enabled():
+        @router.get(settings.KASSA_AI_WEBHOOK_PATH)
+        async def kassa_ai_health() -> JSONResponse:
+            return JSONResponse(
+                {
+                    "status": "ok",
+                    "service": "kassa_ai_webhook",
+                    "enabled": settings.is_kassa_ai_enabled(),
+                }
+            )
+
+        @router.post(settings.KASSA_AI_WEBHOOK_PATH)
+        async def kassa_ai_webhook(request: Request) -> Response:
+            # Получаем данные формы
+            try:
+                form_data = await request.form()
+            except Exception:
+                logger.error("KassaAI webhook: не удалось прочитать данные формы")
+                return Response("Error reading form data", status_code=status.HTTP_400_BAD_REQUEST)
+
+            # Извлекаем параметры (те же что и у Freekassa)
+            merchant_id = form_data.get("MERCHANT_ID")
+            amount = form_data.get("AMOUNT")
+            order_id = form_data.get("MERCHANT_ORDER_ID")
+            sign = form_data.get("SIGN")
+            intid = form_data.get("intid")
+            cur_id = form_data.get("CUR_ID")
+
+            if not all([merchant_id, amount, order_id, sign, intid]):
+                logger.warning("KassaAI webhook: отсутствуют обязательные параметры")
+                return Response("Missing parameters", status_code=status.HTTP_400_BAD_REQUEST)
+
+            try:
+                merchant_id_int = int(merchant_id)
+                amount_float = float(amount)
+                cur_id_int = int(cur_id) if cur_id else None
+            except (ValueError, TypeError) as e:
+                logger.error("KassaAI webhook: некорректные параметры - %s", e)
+                return Response("Invalid parameters", status_code=status.HTTP_400_BAD_REQUEST)
+
+            # Обрабатываем webhook
+            db_generator = get_db()
+            try:
+                db = await db_generator.__anext__()
+            except StopAsyncIteration:
+                return Response("DB Error", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+            try:
+                success = await payment_service.process_kassa_ai_webhook(
+                    db,
+                    merchant_id=merchant_id_int,
+                    amount=amount_float,
+                    order_id=order_id,
+                    sign=sign,
+                    intid=intid,
+                    cur_id=cur_id_int,
+                )
+            finally:
+                try:
+                    await db_generator.__anext__()
+                except StopAsyncIteration:
+                    pass
+
+            if success:
+                return Response("YES", status_code=status.HTTP_200_OK)
+
+            return Response("Error", status_code=status.HTTP_400_BAD_REQUEST)
+
+        routes_registered = True
+
     if routes_registered:
         @router.get("/health/payment-webhooks")
         async def payment_webhooks_health() -> JSONResponse:
@@ -908,6 +1019,7 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
                     "platega_enabled": settings.is_platega_enabled(),
                     "cloudpayments_enabled": settings.is_cloudpayments_enabled(),
                     "freekassa_enabled": settings.is_freekassa_enabled(),
+                    "kassa_ai_enabled": settings.is_kassa_ai_enabled(),
                 }
             )
 

tests/middlewares/__init__.py

@@ -0,0 +1 @@
+# Middlewares tests package

tests/middlewares/test_channel_checker_payload.py

@@ -0,0 +1,263 @@
+"""Тесты для функций сохранения/получения pending_start_payload в channel_checker."""
+
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch, create_autospec
+
+import pytest
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from aiogram.types import Message, User
+
+
+class TestRedisPayloadFunctions:
+    """Тесты для Redis-функций сохранения payload."""
+
+    async def test_save_pending_payload_to_redis_success(self, monkeypatch):
+        """Тест успешного сохранения payload в Redis."""
+        from app.middlewares import channel_checker
+
+        mock_redis = AsyncMock()
+        mock_redis.set = AsyncMock(return_value=True)
+        mock_redis.aclose = AsyncMock()
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(return_value=mock_redis)
+
+            result = await channel_checker.save_pending_payload_to_redis(123456, "ref_test123")
+
+            assert result is True
+            mock_redis.set.assert_awaited_once()
+            call_args = mock_redis.set.await_args
+            assert "pending_start_payload:123456" in call_args.args[0]
+            assert call_args.args[1] == "ref_test123"
+            assert call_args.kwargs.get("ex") == 3600
+            mock_redis.aclose.assert_awaited_once()
+
+    async def test_save_pending_payload_to_redis_failure(self, monkeypatch):
+        """Тест обработки ошибки при сохранении в Redis."""
+        from app.middlewares import channel_checker
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(side_effect=Exception("Redis connection failed"))
+
+            result = await channel_checker.save_pending_payload_to_redis(123456, "ref_test123")
+
+            assert result is False
+
+    async def test_get_pending_payload_from_redis_success(self, monkeypatch):
+        """Тест успешного получения payload из Redis."""
+        from app.middlewares import channel_checker
+
+        mock_redis = AsyncMock()
+        mock_redis.get = AsyncMock(return_value=b"ref_test123")
+        mock_redis.aclose = AsyncMock()
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(return_value=mock_redis)
+
+            result = await channel_checker.get_pending_payload_from_redis(123456)
+
+            assert result == "ref_test123"
+            mock_redis.get.assert_awaited_once()
+            mock_redis.aclose.assert_awaited_once()
+
+    async def test_get_pending_payload_from_redis_not_found(self, monkeypatch):
+        """Тест когда payload не найден в Redis."""
+        from app.middlewares import channel_checker
+
+        mock_redis = AsyncMock()
+        mock_redis.get = AsyncMock(return_value=None)
+        mock_redis.aclose = AsyncMock()
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(return_value=mock_redis)
+
+            result = await channel_checker.get_pending_payload_from_redis(123456)
+
+            assert result is None
+
+    async def test_get_pending_payload_from_redis_failure(self, monkeypatch):
+        """Тест обработки ошибки при получении из Redis."""
+        from app.middlewares import channel_checker
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(side_effect=Exception("Redis connection failed"))
+
+            result = await channel_checker.get_pending_payload_from_redis(123456)
+
+            assert result is None
+
+    async def test_delete_pending_payload_from_redis(self, monkeypatch):
+        """Тест удаления payload из Redis."""
+        from app.middlewares import channel_checker
+
+        mock_redis = AsyncMock()
+        mock_redis.delete = AsyncMock(return_value=1)
+        mock_redis.aclose = AsyncMock()
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(return_value=mock_redis)
+
+            # Не должно бросать исключение
+            await channel_checker.delete_pending_payload_from_redis(123456)
+
+            mock_redis.delete.assert_awaited_once()
+
+    async def test_delete_pending_payload_from_redis_handles_error(self, monkeypatch):
+        """Тест что удаление не бросает исключение при ошибке."""
+        from app.middlewares import channel_checker
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(side_effect=Exception("Redis error"))
+
+            # Не должно бросать исключение
+            await channel_checker.delete_pending_payload_from_redis(123456)
+
+
+def _create_mock_message(text: str, user_id: int):
+    """Создаёт мок Message с нужными атрибутами."""
+    mock_msg = MagicMock(spec=Message)
+    mock_msg.text = text
+    mock_msg.from_user = SimpleNamespace(id=user_id)
+    return mock_msg
+
+
+class TestCaptureStartPayload:
+    """Тесты для метода _capture_start_payload."""
+
+    async def test_capture_saves_to_fsm_state(self, monkeypatch):
+        """Тест сохранения payload в FSM state."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        mock_state = AsyncMock()
+        mock_state.get_data = AsyncMock(return_value={})
+        mock_state.set_data = AsyncMock()
+
+        mock_message = _create_mock_message("/start ref_abc123", 123456)
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock) as mock_save_redis:
+            await middleware._capture_start_payload(mock_state, mock_message, None)
+
+            mock_state.set_data.assert_awaited_once()
+            saved_data = mock_state.set_data.await_args.args[0]
+            assert saved_data["pending_start_payload"] == "ref_abc123"
+
+            # Также должен сохраняться в Redis
+            mock_save_redis.assert_awaited_once_with(123456, "ref_abc123")
+
+    async def test_capture_saves_to_redis_when_state_none(self, monkeypatch):
+        """Тест сохранения payload в Redis когда FSM state недоступен."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        mock_message = _create_mock_message("/start ref_xyz789", 999888)
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock) as mock_save_redis:
+            await middleware._capture_start_payload(None, mock_message, None)
+
+            # Должен сохраняться в Redis даже если state=None
+            mock_save_redis.assert_awaited_once_with(999888, "ref_xyz789")
+
+    async def test_capture_ignores_message_without_payload(self, monkeypatch):
+        """Тест что сообщение без payload игнорируется."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        mock_state = AsyncMock()
+        mock_state.get_data = AsyncMock(return_value={})
+        mock_state.set_data = AsyncMock()
+
+        mock_message = _create_mock_message("/start", 123456)  # Без payload
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock) as mock_save_redis:
+            await middleware._capture_start_payload(mock_state, mock_message, None)
+
+            mock_state.set_data.assert_not_awaited()
+            mock_save_redis.assert_not_awaited()
+
+    async def test_capture_ignores_non_start_message(self, monkeypatch):
+        """Тест что не-start сообщения игнорируются."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        mock_state = AsyncMock()
+        mock_state.get_data = AsyncMock(return_value={})
+        mock_state.set_data = AsyncMock()
+
+        mock_message = _create_mock_message("/help something", 123456)  # Не /start
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock) as mock_save_redis:
+            await middleware._capture_start_payload(mock_state, mock_message, None)
+
+            mock_state.set_data.assert_not_awaited()
+            mock_save_redis.assert_not_awaited()
+
+    async def test_capture_does_not_overwrite_same_payload(self, monkeypatch):
+        """Тест что одинаковый payload не перезаписывается в FSM state."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        mock_state = AsyncMock()
+        mock_state.get_data = AsyncMock(return_value={"pending_start_payload": "ref_same"})
+        mock_state.set_data = AsyncMock()
+
+        mock_message = _create_mock_message("/start ref_same", 123456)  # Тот же payload
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock) as mock_save_redis:
+            await middleware._capture_start_payload(mock_state, mock_message, None)
+
+            # FSM state не должен перезаписываться
+            mock_state.set_data.assert_not_awaited()
+            # Но в Redis всё равно сохраняем (для надёжности)
+            mock_save_redis.assert_awaited_once()
+
+
+class TestPayloadIntegration:
+    """Интеграционные тесты для потока сохранения/восстановления payload."""
+
+    async def test_full_flow_fsm_state_works(self, monkeypatch):
+        """Тест полного потока когда FSM state работает корректно."""
+        from app.middlewares.channel_checker import ChannelCheckerMiddleware
+
+        middleware = ChannelCheckerMiddleware()
+
+        # Сохраняем payload
+        state_storage = {}
+
+        mock_state = AsyncMock()
+        mock_state.get_data = AsyncMock(return_value=state_storage)
+        mock_state.set_data = AsyncMock(side_effect=lambda d: state_storage.update(d))
+
+        mock_message = _create_mock_message("/start ref_flow_test", 111222)
+
+        with patch("app.middlewares.channel_checker.save_pending_payload_to_redis", new_callable=AsyncMock):
+            await middleware._capture_start_payload(mock_state, mock_message, None)
+
+        # Проверяем что payload сохранён
+        assert state_storage.get("pending_start_payload") == "ref_flow_test"
+
+    async def test_payload_retrieved_from_redis_fallback(self, monkeypatch):
+        """Тест что payload восстанавливается из Redis если в FSM state его нет."""
+        from app.middlewares.channel_checker import get_pending_payload_from_redis
+
+        mock_redis = AsyncMock()
+        mock_redis.get = AsyncMock(return_value=b"ref_from_redis")
+        mock_redis.aclose = AsyncMock()
+
+        with patch("app.middlewares.channel_checker.aioredis") as mock_aioredis:
+            mock_aioredis.from_url = MagicMock(return_value=mock_redis)
+
+            result = await get_pending_payload_from_redis(333444)
+
+            assert result == "ref_from_redis"