fix: multi-tariff Stage 5 fixes — auth sync, notifications, cart, race guard

HIGH fixes: - auth.py: profile description sync now iterates all per-subscription remnawave_uuids in multi-tariff mode - admin_users: sync_from_panel uses subscription UUIDs for panel lookup, does not overwrite user.remnawave_uuid in multi-tariff MEDIUM fixes: - monitoring_service: _send_subscription_expired_notification now takes subscription param, uses se:{sub_id} in multi-tariff - remnawave_webhook_service: _get_renew_keyboard accepts subscription_id, all 7 callers pass it - recurrent_payment_service: _build_extend_keyboard with subscription_id - user_service: balance notification keyboards use menu_subscription in multi-tariff instead of bare subscription_extend - autopay.py + purchase.py: per-subscription cart deletion instead of global delete_user_cart where subscription context available - subscription_auto_purchase_service: 60-sec race guard changed from per-user to per-subscription (checks subscription.updated_at)
2026-03-26 09:01:36 +03:00
parent a49e52cc92
commit 948e4791f4
9 changed files with 210 additions and 74 deletions
@@ -2720,7 +2720,18 @@ async def sync_user_from_panel(
            # Find user in panel: UUID → telegram_id → email
            panel_user = None

-            if user.remnawave_uuid:
+            if settings.is_multi_tariff_enabled():
+                # In multi-tariff mode, user.remnawave_uuid is None; UUIDs live on subscriptions
+                sub_uuids = [
+                    s.remnawave_uuid
+                    for s in (getattr(user, 'subscriptions', None) or [])
+                    if s.remnawave_uuid
+                ]
+                for _uuid in sub_uuids:
+                    panel_user = await api.get_user_by_uuid(_uuid)
+                    if panel_user:
+                        break
+            elif user.remnawave_uuid:
                panel_user = await api.get_user_by_uuid(user.remnawave_uuid)

            if not panel_user and user.telegram_id:
@@ -2763,7 +2774,8 @@ async def sync_user_from_panel(
            )

            # Update remnawave_uuid if different
-            if user.remnawave_uuid != panel_user.uuid:
+            # In multi-tariff mode the UUID belongs to the subscription, not the user
+            if not settings.is_multi_tariff_enabled() and user.remnawave_uuid != panel_user.uuid:
                changes['remnawave_uuid'] = {'old': user.remnawave_uuid, 'new': panel_user.uuid}
                user.remnawave_uuid = panel_user.uuid

@@ -301,8 +301,31 @@ async def handle_subscription_cancel(callback: types.CallbackQuery, state: FSMCo
    await state.clear()
    await clear_subscription_checkout_draft(db_user.id)

-    # Удаляем сохраненную корзину, чтобы не показывать кнопку возврата
-    await user_cart_service.delete_user_cart(db_user.id)
+    # Multi-tariff safe: delete only the cart for the current subscription
+    # to avoid nuking carts belonging to other subscriptions.
+    cart_data = await user_cart_service.get_user_cart(db_user.id)
+    cart_sub_id = None
+    if cart_data:
+        try:
+            raw = cart_data.get('subscription_id')
+            if raw is not None:
+                cart_sub_id = int(raw)
+        except (TypeError, ValueError):
+            pass
+
+    if cart_sub_id is not None:
+        await user_cart_service.delete_subscription_cart(db_user.id, cart_sub_id)
+        # Clean up global key only if it still references this subscription
+        global_cart = await user_cart_service.get_user_cart(db_user.id)
+        if global_cart and global_cart.get('subscription_id') is not None:
+            try:
+                if int(global_cart['subscription_id']) == cart_sub_id:
+                    await user_cart_service.delete_global_cart_only(db_user.id)
+            except (TypeError, ValueError):
+                pass
+    else:
+        # No subscription_id in cart -- safe to delete the global cart
+        await user_cart_service.delete_user_cart(db_user.id)

    from app.handlers.menu import show_main_menu

@@ -1471,7 +1471,27 @@ async def return_to_saved_cart(callback: types.CallbackQuery, state: FSMContext,

    if 'period_days' not in prepared_cart_data:
        await callback.answer('❌ Корзина повреждена. Оформите подписку заново.', show_alert=True)
-        await user_cart_service.delete_user_cart(db_user.id)
+        # Multi-tariff safe: try per-subscription deletion to avoid nuking other carts
+        corrupted_sub_id = None
+        try:
+            raw = cart_data.get('subscription_id')
+            if raw is not None:
+                corrupted_sub_id = int(raw)
+        except (TypeError, ValueError):
+            pass
+
+        if corrupted_sub_id is not None:
+            await user_cart_service.delete_subscription_cart(db_user.id, corrupted_sub_id)
+            global_cart = await user_cart_service.get_user_cart(db_user.id)
+            if global_cart and global_cart.get('subscription_id') is not None:
+                try:
+                    if int(global_cart['subscription_id']) == corrupted_sub_id:
+                        await user_cart_service.delete_global_cart_only(db_user.id)
+                except (TypeError, ValueError):
+                    pass
+        else:
+            # Cart corrupted beyond reading subscription_id -- global cleanup
+            await user_cart_service.delete_user_cart(db_user.id)
        return

    if not settings.is_devices_selection_enabled():
@@ -2919,7 +2939,10 @@ async def handle_subscription_settings(callback: types.CallbackQuery, db_user: U


 async def clear_saved_cart(callback: types.CallbackQuery, state: FSMContext, db_user: User, db: AsyncSession):
-    # Очищаем как FSM, так и Redis
+    # Очищаем как FSM, так и Redis.
+    # NOTE: Intentionally deletes ALL carts (global + per-subscription cascade)
+    # because this is an explicit user action ("clear my cart").  In multi-tariff
+    # mode the user expects a full reset, not per-subscription cleanup.
    await state.clear()
    await user_cart_service.delete_user_cart(db_user.id)

@@ -212,6 +212,23 @@ class AuthMiddleware(BaseMiddleware):
                            )
                        )

+                    # Multi-tariff: sync all per-subscription panel users
+                    if settings.is_multi_tariff_enabled():
+                        description = settings.format_remnawave_user_description(
+                            full_name=db_user.full_name,
+                            username=db_user.username,
+                            telegram_id=db_user.telegram_id,
+                        )
+                        for sub in getattr(db_user, 'subscriptions', None) or []:
+                            if sub.remnawave_uuid and sub.remnawave_uuid != db_user.remnawave_uuid:
+                                asyncio.create_task(
+                                    _refresh_remnawave_description(
+                                        remnawave_uuid=sub.remnawave_uuid,
+                                        description=description,
+                                        telegram_id=db_user.telegram_id,
+                                    )
+                                )
+
                data['db'] = db
                data['db_user'] = db_user
                data['is_admin'] = settings.is_admin(user.id)
@@ -298,7 +298,7 @@ class MonitoringService:

                user = await get_user_by_id(db, subscription.user_id)
                if user and self.bot:
-                    await self._send_subscription_expired_notification(user)
+                    await self._send_subscription_expired_notification(user, subscription)

                logger.info(
                    "🔴 Подписка пользователя истекла и статус изменен на 'expired'", user_id=subscription.user_id
@@ -1303,7 +1303,7 @@ class MonitoringService:
        except Exception as e:
            logger.error('Ошибка обработки автоплатежей', error=e)

-    async def _send_subscription_expired_notification(self, user: User) -> bool:
+    async def _send_subscription_expired_notification(self, user: User, subscription: Subscription) -> bool:
        try:
            message = """
 ⛔ <b>Подписка истекла</b>
@@ -1315,9 +1315,14 @@ class MonitoringService:

            from aiogram.types import InlineKeyboardMarkup

+            extend_callback = (
+                f'se:{subscription.id}'
+                if settings.is_multi_tariff_enabled()
+                else 'subscription_extend'
+            )
            keyboard = InlineKeyboardMarkup(
                inline_keyboard=[
-                    [build_miniapp_or_callback_button(text='💎 Купить подписку', callback_data='menu_buy')],
+                    [build_miniapp_or_callback_button(text='💎 Продлить подписку', callback_data=extend_callback)],
                    [build_miniapp_or_callback_button(text='💳 Пополнить баланс', callback_data='balance_topup')],
                ]
            )
@@ -52,14 +52,19 @@ class _DailyGuard:
 _daily_guard = _DailyGuard()


-def _build_extend_keyboard(texts) -> InlineKeyboardMarkup:
+def _build_extend_keyboard(texts, subscription_id: int | None = None) -> InlineKeyboardMarkup:
    """Клавиатура с кнопкой продления подписки для уведомлений."""
+    extend_callback = (
+        f'se:{subscription_id}'
+        if settings.is_multi_tariff_enabled() and subscription_id
+        else 'subscription_extend'
+    )
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text=texts.t('SUBSCRIPTION_EXTEND', '💎 Продлить подписку'),
-                    callback_data='subscription_extend',
+                    callback_data=extend_callback,
                )
            ],
        ]
@@ -355,7 +360,7 @@ async def _process_single_subscription(
                texts = get_texts(user.language)
                payment_status = result.get('status', '')
                if result.get('paid'):
-                    keyboard = _build_extend_keyboard(texts)
+                    keyboard = _build_extend_keyboard(texts, subscription.id)
                    msg = texts.t(
                        'RECURRENT_TOPUP_SUCCESS',
                        '✅ <b>Автоплатёж выполнен</b>\n\nБаланс пополнен на {amount} для продления подписки.',
@@ -383,7 +388,7 @@ async def _process_single_subscription(
            from app.localization.texts import get_texts

            texts = get_texts(user.language)
-            keyboard = _build_extend_keyboard(texts)
+            keyboard = _build_extend_keyboard(texts, subscription.id)
            msg = texts.t(
                'RECURRENT_TOPUP_FAILED',
                '❌ <b>Автоплатёж не удался</b>\n\nНе удалось списать {amount} ни с одной сохранённой карты для продления подписки.\n\nПополните баланс вручную, чтобы подписка не прервалась.',
@@ -443,12 +443,17 @@ class RemnaWaveWebhookService:
            return False
        return bool(re.match(r'^[a-zA-Z][a-zA-Z0-9+\-.]*://', value))

-    def _get_renew_keyboard(self, user: User) -> InlineKeyboardMarkup:
+    def _get_renew_keyboard(self, user: User, subscription_id: int | None = None) -> InlineKeyboardMarkup:
        texts = get_texts(user.language)
        button_text = texts.get('WEBHOOK_RENEW_BUTTON', 'Renew subscription')
+        extend_callback = (
+            f'se:{subscription_id}'
+            if settings.is_multi_tariff_enabled() and subscription_id
+            else 'subscription_extend'
+        )
        return InlineKeyboardMarkup(
            inline_keyboard=[
-                [build_miniapp_or_callback_button(text=button_text, callback_data='subscription_extend')],
+                [build_miniapp_or_callback_button(text=button_text, callback_data=extend_callback)],
            ]
        )

@@ -592,7 +597,11 @@ class RemnaWaveWebhookService:
            else:
                await db.commit()

-        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRED', reply_markup=self._get_renew_keyboard(user))
+        await self._notify_user(
+            user,
+            'WEBHOOK_SUB_EXPIRED',
+            reply_markup=self._get_renew_keyboard(user, getattr(subscription, 'id', None) if subscription else None),
+        )

    async def _handle_user_disabled(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
@@ -916,9 +925,17 @@ class RemnaWaveWebhookService:
                subscription.connected_squads = []
                subscription.updated_at = datetime.now(UTC)
                await db.commit()
-                await self._notify_user(user, 'WEBHOOK_SUB_DELETED', reply_markup=self._get_renew_keyboard(user))
+                await self._notify_user(
+                    user,
+                    'WEBHOOK_SUB_DELETED',
+                    reply_markup=self._get_renew_keyboard(user, getattr(subscription, 'id', None) if subscription else None),
+                )
        else:
-            await self._notify_user(user, 'WEBHOOK_SUB_DELETED', reply_markup=self._get_renew_keyboard(user))
+            await self._notify_user(
+                user,
+                'WEBHOOK_SUB_DELETED',
+                reply_markup=self._get_renew_keyboard(user, getattr(subscription, 'id', None) if subscription else None),
+            )

    async def _attempt_panel_recreation(self, db: AsyncSession, user: User, subscription: Subscription) -> bool:
        """Re-create user in RemnaWave panel after spurious user.deleted webhook.
@@ -1010,22 +1027,26 @@ class RemnaWaveWebhookService:
    async def _handle_expires_in_72h(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
    ) -> None:
-        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_72H', reply_markup=self._get_renew_keyboard(user))
+        sub_id = getattr(subscription, 'id', None) if subscription else None
+        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_72H', reply_markup=self._get_renew_keyboard(user, sub_id))

    async def _handle_expires_in_48h(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
    ) -> None:
-        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_48H', reply_markup=self._get_renew_keyboard(user))
+        sub_id = getattr(subscription, 'id', None) if subscription else None
+        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_48H', reply_markup=self._get_renew_keyboard(user, sub_id))

    async def _handle_expires_in_24h(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
    ) -> None:
-        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_24H', reply_markup=self._get_renew_keyboard(user))
+        sub_id = getattr(subscription, 'id', None) if subscription else None
+        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRES_24H', reply_markup=self._get_renew_keyboard(user, sub_id))

    async def _handle_expired_24h_ago(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
    ) -> None:
-        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRED_24H_AGO', reply_markup=self._get_renew_keyboard(user))
+        sub_id = getattr(subscription, 'id', None) if subscription else None
+        await self._notify_user(user, 'WEBHOOK_SUB_EXPIRED_24H_AGO', reply_markup=self._get_renew_keyboard(user, sub_id))

    async def _handle_first_connected(
        self, db: AsyncSession, user: User, subscription: Subscription | None, data: dict
@@ -1416,12 +1416,23 @@ async def _auto_add_devices(
        return False

    # Проверяем подписку (with lock to prevent concurrent device modifications)
-    locked_result = await db.execute(
-        select(Subscription)
-        .where(Subscription.user_id == user.id)
-        .with_for_update()
-        .execution_options(populate_existing=True)
-    )
+    _cart_sub_id_devices = _safe_int(cart_data.get('subscription_id'))
+    if settings.is_multi_tariff_enabled() and _cart_sub_id_devices:
+        locked_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.id == _cart_sub_id_devices, Subscription.user_id == user.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+    else:
+        locked_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.user_id == user.id)
+            .order_by(Subscription.created_at.desc())
+            .limit(1)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
    subscription = locked_result.scalar_one_or_none()
    if not subscription:
        logger.warning('🔁 Автопокупка устройств: у пользователя нет подписки', format_user_id=_format_user_id(user))
@@ -2062,7 +2073,6 @@ async def try_auto_extend_expired_after_topup(
    """
    from app.cabinet.routes.websocket import notify_user_subscription_renewed
    from app.database.crud.subscription import get_subscription_by_user_id
-    from app.database.crud.transaction import get_user_transactions

    if not user or not getattr(user, 'id', None):
        return False
@@ -2166,25 +2176,24 @@ async def try_auto_extend_expired_after_topup(
        )
        return False

-    # Race condition guard: skip if a subscription payment was made in the last 60 seconds
+    # Race condition guard (per-subscription): skip if THIS subscription was
+    # modified in the last 60 seconds (indicates a concurrent renewal just landed).
    try:
-        recent_transactions = await get_user_transactions(db, user.id, limit=1)
-        if recent_transactions:
-            last_tx = recent_transactions[0]
-            if (
-                last_tx.type == TransactionType.SUBSCRIPTION_PAYMENT
-                and last_tx.created_at
-                and (datetime.now(UTC) - last_tx.created_at) < timedelta(seconds=60)
-            ):
-                logger.info(
-                    '🔄 Автопродление expired: пропуск — подписка оплачена секунд назад',
-                    format_user_id=_format_user_id(user),
-                    total_seconds=(datetime.now(UTC) - last_tx.created_at).total_seconds(),
-                )
-                return False
+        await db.refresh(subscription, attribute_names=['updated_at'])
+        if (
+            subscription.updated_at
+            and (datetime.now(UTC) - subscription.updated_at) < timedelta(seconds=60)
+        ):
+            logger.info(
+                '🔄 Автопродление expired: пропуск — подписка обновлена секунд назад',
+                format_user_id=_format_user_id(user),
+                subscription_id=subscription.id,
+                total_seconds=(datetime.now(UTC) - subscription.updated_at).total_seconds(),
+            )
+            return False
    except Exception as check_error:
        logger.warning(
-            '🔄 Автопродление expired: ошибка проверки последней транзакции',
+            '🔄 Автопродление expired: ошибка проверки updated_at подписки',
            format_user_id=_format_user_id(user),
            check_error=check_error,
        )
@@ -2511,26 +2520,24 @@ async def try_resume_disabled_daily_after_topup(
        )
        return False

-    # Race condition guard: skip if a subscription payment was made in the last 60 seconds
-    from app.database.crud.transaction import get_user_transactions
-
+    # Race condition guard (per-subscription): skip if THIS daily subscription
+    # was modified in the last 60 seconds (indicates a concurrent charge just landed).
    try:
-        recent_transactions = await get_user_transactions(db, user.id, limit=1)
-        if recent_transactions:
-            last_tx = recent_transactions[0]
-            if (
-                last_tx.type == TransactionType.SUBSCRIPTION_PAYMENT
-                and last_tx.created_at
-                and (datetime.now(UTC) - last_tx.created_at) < timedelta(seconds=60)
-            ):
-                logger.info(
-                    '🔄 Авто-возобновление daily: пропуск — оплата секунд назад',
-                    format_user_id=_format_user_id(user),
-                )
-                return False
+        await db.refresh(subscription, attribute_names=['updated_at'])
+        if (
+            subscription.updated_at
+            and (datetime.now(UTC) - subscription.updated_at) < timedelta(seconds=60)
+        ):
+            logger.info(
+                '🔄 Авто-возобновление daily: пропуск — подписка обновлена секунд назад',
+                format_user_id=_format_user_id(user),
+                subscription_id=subscription.id,
+                total_seconds=(datetime.now(UTC) - subscription.updated_at).total_seconds(),
+            )
+            return False
    except Exception as check_error:
        logger.warning(
-            '🔄 Авто-возобновление daily: ошибка проверки последней транзакции',
+            '🔄 Авто-возобновление daily: ошибка проверки updated_at подписки',
            format_user_id=_format_user_id(user),
            check_error=check_error,
        )
@@ -2853,23 +2860,43 @@ async def _process_single_cart(
        await clear_subscription_checkout_draft(user.id)
        return False

-    # Race condition guard: subscription paid in the last 60 seconds
+    # Race condition guard (per-subscription): skip if THIS subscription was
+    # modified in the last 60 seconds (indicates a concurrent purchase just landed).
+    # When cart_sub_id is available we check the specific subscription's updated_at;
+    # otherwise fall back to the user-global last transaction check.
    if cart_mode in ('extend', 'tariff_purchase', 'daily_tariff_purchase'):
        try:
-            recent_transactions = await get_user_transactions(db, user.id, limit=1)
-            if recent_transactions:
-                last_tx = recent_transactions[0]
+            if cart_sub_id:
+                from app.database.crud.subscription import get_subscription_by_id_for_user
+
+                target_sub = await get_subscription_by_id_for_user(db, cart_sub_id, user.id)
                if (
-                    last_tx.type == TransactionType.SUBSCRIPTION_PAYMENT
-                    and last_tx.created_at
-                    and (datetime.now(UTC) - last_tx.created_at) < timedelta(seconds=60)
+                    target_sub
+                    and target_sub.updated_at
+                    and (datetime.now(UTC) - target_sub.updated_at) < timedelta(seconds=60)
                ):
                    logger.info(
-                        'Автопокупка: пропускаем -- подписка уже куплена секунд назад',
+                        'Автопокупка: пропускаем -- подписка обновлена секунд назад',
                        format_user_id=_format_user_id(user),
-                        total_seconds=(datetime.now(UTC) - last_tx.created_at).total_seconds(),
+                        subscription_id=cart_sub_id,
+                        total_seconds=(datetime.now(UTC) - target_sub.updated_at).total_seconds(),
                    )
                    return False
+            else:
+                recent_transactions = await get_user_transactions(db, user.id, limit=1)
+                if recent_transactions:
+                    last_tx = recent_transactions[0]
+                    if (
+                        last_tx.type == TransactionType.SUBSCRIPTION_PAYMENT
+                        and last_tx.created_at
+                        and (datetime.now(UTC) - last_tx.created_at) < timedelta(seconds=60)
+                    ):
+                        logger.info(
+                            'Автопокупка: пропускаем -- подписка уже куплена секунд назад',
+                            format_user_id=_format_user_id(user),
+                            total_seconds=(datetime.now(UTC) - last_tx.created_at).total_seconds(),
+                        )
+                        return False
        except Exception as check_error:
            logger.warning(
                'Автопокупка: ошибка проверки последней транзакции',
@@ -96,12 +96,13 @@ class UserService:
                f'💳 Текущий баланс: {settings.format_price(user.balance_kopeks)}\n\n'
                f'Спасибо за использование нашего сервиса! 🎉'
            )
+            extend_callback = 'menu_subscription' if settings.is_multi_tariff_enabled() else 'subscription_extend'
            keyboard = types.InlineKeyboardMarkup(
                inline_keyboard=[
                    [
                        types.InlineKeyboardButton(
                            text=texts.t('SUBSCRIPTION_EXTEND', '💎 Продлить подписку'),
-                            callback_data='subscription_extend',
+                            callback_data=extend_callback,
                        )
                    ]
                ]
@@ -117,10 +118,11 @@ class UserService:
                f'Пополнение баланса НЕ активирует подписку автоматически!\n\n'
                f'👇 <b>Выберите действие:</b>'
            )
+            extend_callback = 'menu_subscription' if settings.is_multi_tariff_enabled() else 'subscription_extend'
            keyboard = types.InlineKeyboardMarkup(
                inline_keyboard=[
                    [types.InlineKeyboardButton(text='🚀 АКТИВИРОВАТЬ ПОДПИСКУ', callback_data='subscription_buy')],
-                    [types.InlineKeyboardButton(text='💎 ПРОДЛИТЬ ПОДПИСКУ', callback_data='subscription_extend')],
+                    [types.InlineKeyboardButton(text='💎 ПРОДЛИТЬ ПОДПИСКУ', callback_data=extend_callback)],
                    [
                        types.InlineKeyboardButton(
                            text='📱 ДОБАВИТЬ УСТРОЙСТВА', callback_data='subscription_add_devices'
@@ -169,11 +171,12 @@ class UserService:
        subs = getattr(user, 'subscriptions', None) or []
        has_extendable = any(sub.status in {'active', 'expired', 'trial'} for sub in subs)
        if has_extendable:
+            extend_callback = 'menu_subscription' if settings.is_multi_tariff_enabled() else 'subscription_extend'
            keyboard_rows.append(
                [
                    types.InlineKeyboardButton(
                        text=get_texts(user.language).t('SUBSCRIPTION_EXTEND', '💎 Продлить подписку'),
-                        callback_data='subscription_extend',
+                        callback_data=extend_callback,
                    )
                ]
            )