diff --git a/app/cabinet/routes/subscription.py b/app/cabinet/routes/subscription.py index a4aaa0e0..f36fcc86 100644 --- a/app/cabinet/routes/subscription.py +++ b/app/cabinet/routes/subscription.py @@ -1000,7 +1000,10 @@ async def purchase_devices_legacy( # Lock subscription row to prevent concurrent device purchases exceeding the limit result = await db.execute( - select(Subscription).where(Subscription.user_id == user.id).with_for_update() + select(Subscription) + .where(Subscription.user_id == user.id) + .with_for_update() + .execution_options(populate_existing=True) ) subscription = result.scalar_one_or_none() @@ -1092,9 +1095,33 @@ async def purchase_devices_legacy( detail='Insufficient funds', ) - # Add devices - subscription.device_limit = new_devices + # Re-lock subscription after subtract_user_balance committed (which released all locks). + # Re-validate max device limit to prevent concurrent purchases exceeding the limit. + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + actual_current = subscription.device_limit or 1 + actual_new = actual_current + request.devices + if max_devices > 0 and actual_new > max_devices: + # Concurrent purchase already exceeded limit — refund balance + user_refund = await db.execute( + select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += total_price + await db.commit() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f'Maximum device limit is {max_devices}. Balance refunded.', + ) + + # Add devices (under lock) + subscription.device_limit = actual_new await db.commit() await db.refresh(user) @@ -1111,10 +1138,10 @@ async def purchase_devices_legacy( await notification_service.send_subscription_update_notification( db=db, user=user, - subscription=user.subscription, + subscription=subscription, update_type='devices', old_value=current_devices, - new_value=new_devices, + new_value=actual_new, price_paid=total_price, ) finally: @@ -1125,7 +1152,7 @@ async def purchase_devices_legacy( response = { 'message': 'Devices added successfully', 'devices_added': request.devices, - 'new_device_limit': new_devices, + 'new_device_limit': actual_new, 'amount_paid_kopeks': total_price, } @@ -2298,7 +2325,10 @@ async def purchase_devices( try: # Lock subscription row to prevent concurrent device purchases exceeding the limit result = await db.execute( - select(Subscription).where(Subscription.user_id == user.id).with_for_update() + select(Subscription) + .where(Subscription.user_id == user.id) + .with_for_update() + .execution_options(populate_existing=True) ) subscription = result.scalar_one_or_none() @@ -2430,8 +2460,33 @@ async def purchase_devices( detail='Insufficient funds', ) - # Increase device limit - subscription.device_limit += request.devices + # Re-lock subscription after subtract_user_balance committed (which released all locks). + # Re-validate max device limit to prevent concurrent purchases exceeding the limit. + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + + actual_current = subscription.device_limit or 1 + actual_new = actual_current + request.devices + if max_device_limit and actual_new > max_device_limit: + # Concurrent purchase already exceeded limit — refund balance + user_refund = await db.execute( + select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price_kopeks + await db.commit() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f'Максимальное количество устройств: {max_device_limit}. Баланс возвращён.', + ) + + # Increase device limit (under lock) + subscription.device_limit = actual_new await db.commit() await db.refresh(subscription) @@ -3720,16 +3775,21 @@ async def reduce_devices( detail='Invalid new_device_limit', ) - await db.refresh(user, ['subscription']) + # Lock subscription to prevent concurrent device modifications + result = await db.execute( + select(Subscription) + .where(Subscription.user_id == user.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = result.scalar_one_or_none() - if not user.subscription: + if not subscription: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail='No subscription found', ) - subscription = user.subscription - if subscription.is_trial: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, diff --git a/app/database/crud/subscription.py b/app/database/crud/subscription.py index c33ad8bc..063af465 100644 --- a/app/database/crud/subscription.py +++ b/app/database/crud/subscription.py @@ -613,7 +613,29 @@ async def add_subscription_traffic(db: AsyncSession, subscription: Subscription, async def add_subscription_devices(db: AsyncSession, subscription: Subscription, devices: int) -> Subscription: - subscription.device_limit += devices + # Lock subscription to prevent concurrent modifications + locked_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = locked_result.scalar_one() + + # Check max device limit + max_devices = settings.MAX_DEVICES_LIMIT + new_limit = (subscription.device_limit or 1) + devices + if max_devices > 0 and new_limit > max_devices: + logger.warning( + '📱 Попытка превысить лимит устройств', + user_id=subscription.user_id, + current=subscription.device_limit, + requested=devices, + max_devices=max_devices, + ) + new_limit = max_devices + + subscription.device_limit = new_limit subscription.updated_at = datetime.now(UTC) await db.commit() diff --git a/app/database/crud/user.py b/app/database/crud/user.py index 9170471a..05670354 100644 --- a/app/database/crud/user.py +++ b/app/database/crud/user.py @@ -515,7 +515,9 @@ async def subtract_user_balance( logger.info('📝 Описание', description=description) # Lock the user row to prevent concurrent balance race conditions - locked_result = await db.execute(select(User).where(User.id == user.id).with_for_update()) + locked_result = await db.execute( + select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True) + ) user = locked_result.scalar_one() log_context: dict[str, object] | None = None diff --git a/app/handlers/subscription/devices.py b/app/handlers/subscription/devices.py index c219b823..806c1bcc 100644 --- a/app/handlers/subscription/devices.py +++ b/app/handlers/subscription/devices.py @@ -2,13 +2,13 @@ import html as html_mod from datetime import UTC, datetime from aiogram import types +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings -from app.database.crud.subscription import add_subscription_devices from app.database.crud.transaction import create_transaction from app.database.crud.user import subtract_user_balance -from app.database.models import TransactionType, User +from app.database.models import Subscription, TransactionType, User from app.keyboards.inline import ( get_app_selection_keyboard, get_back_keyboard, @@ -554,6 +554,51 @@ async def execute_change_devices(callback: types.CallbackQuery, db_user: User, d description=f'Изменение устройств с {current_devices} до {new_devices_count} на {charged_months} мес', ) + # Re-lock subscription after subtract_user_balance committed (released all locks) + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + + # Re-validate: prevent double-charge and max-limit violation + if new_devices_count > current_devices: + max_devices = settings.MAX_DEVICES_LIMIT + if max_devices > 0 and new_devices_count > max_devices: + if price > 0: + user_refund = await db.execute( + select(User) + .where(User.id == db_user.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price + await db.commit() + await callback.answer( + f'⚠️ Лимит устройств ({max_devices}) превышен. Баланс возвращён.', + show_alert=True, + ) + return + # Check if concurrent request already applied the same change + if price > 0 and subscription.device_limit >= new_devices_count: + user_refund = await db.execute( + select(User) + .where(User.id == db_user.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price + await db.commit() + await callback.answer( + '⚠️ Изменение уже применено. Баланс возвращён.', + show_alert=True, + ) + return + subscription.device_limit = new_devices_count subscription.updated_at = datetime.now(UTC) @@ -1196,7 +1241,36 @@ async def confirm_add_devices(callback: types.CallbackQuery, db_user: User, db: await callback.answer('⚠️ Ошибка списания средств', show_alert=True) return - await add_subscription_devices(db, subscription, devices_count) + # Re-lock subscription after subtract_user_balance committed (released all locks) + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + + # Re-validate max device limit after re-lock + actual_current = subscription.device_limit or 1 + actual_new = actual_current + devices_count + max_devices = settings.MAX_DEVICES_LIMIT + if max_devices > 0 and actual_new > max_devices: + # Concurrent purchase exceeded limit — refund + user_refund = await db.execute( + select(User).where(User.id == db_user.id).with_for_update().execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price + await db.commit() + await callback.answer( + f'⚠️ Лимит устройств ({max_devices}) превышен. Баланс возвращён.', + show_alert=True, + ) + return + + subscription.device_limit = actual_new + subscription.updated_at = datetime.now(UTC) + await db.commit() subscription_service = SubscriptionService() await subscription_service.update_remnawave_user(db, subscription) diff --git a/app/services/subscription_auto_purchase_service.py b/app/services/subscription_auto_purchase_service.py index c7153c35..14b954c6 100644 --- a/app/services/subscription_auto_purchase_service.py +++ b/app/services/subscription_auto_purchase_service.py @@ -8,6 +8,7 @@ from datetime import UTC, datetime, timedelta import structlog from aiogram import Bot from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings @@ -1181,7 +1182,6 @@ async def _auto_add_devices( """Auto-purchase devices from saved cart after balance topup.""" from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup - from app.database.crud.subscription import get_subscription_by_user_id from app.database.crud.user import subtract_user_balance from app.database.models import PaymentMethod @@ -1207,8 +1207,14 @@ async def _auto_add_devices( ) return False - # Проверяем подписку - subscription = await get_subscription_by_user_id(db, user.id) + # Проверяем подписку (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) + ) + subscription = locked_result.scalar_one_or_none() if not subscription: logger.warning('🔁 Автопокупка устройств: у пользователя нет подписки', format_user_id=_format_user_id(user)) await user_cart_service.delete_user_cart(user.id) @@ -1223,6 +1229,21 @@ async def _auto_add_devices( await user_cart_service.delete_user_cart(user.id) return False + # Check max device limit before charging + old_device_limit = subscription.device_limit or 1 + new_device_limit = old_device_limit + devices_to_add + max_devices = settings.MAX_DEVICES_LIMIT + if max_devices > 0 and new_device_limit > max_devices: + logger.warning( + '🔁 Автопокупка устройств: превышен лимит устройств', + format_user_id=_format_user_id(user), + current=old_device_limit, + requested=new_device_limit, + max_devices=max_devices, + ) + await user_cart_service.delete_user_cart(user.id) + return False + # Списываем баланс description = f'Покупка {devices_to_add} доп. устройств' try: @@ -1248,9 +1269,35 @@ async def _auto_add_devices( ) return False - # Добавляем устройства + # Re-lock subscription after subtract_user_balance committed (released locks) + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + old_device_limit = subscription.device_limit or 1 - subscription.device_limit = old_device_limit + devices_to_add + new_device_limit = old_device_limit + devices_to_add + + if max_devices > 0 and new_device_limit > max_devices: + # Concurrent modification exceeded limit — refund + user_refund = await db.execute( + select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price_kopeks + await db.commit() + logger.warning( + '🔁 Автопокупка устройств: лимит превышен после оплаты, баланс возвращён', + format_user_id=_format_user_id(user), + ) + await user_cart_service.delete_user_cart(user.id) + return False + + # Добавляем устройства (under lock) + subscription.device_limit = new_device_limit try: await db.commit() diff --git a/app/webapi/routes/miniapp.py b/app/webapi/routes/miniapp.py index 2f6c1344..c0d8a39e 100644 --- a/app/webapi/routes/miniapp.py +++ b/app/webapi/routes/miniapp.py @@ -6115,7 +6115,10 @@ async def update_subscription_devices_endpoint( # Re-read subscription under row lock to prevent concurrent device purchases exceeding limit locked_result = await db.execute( - select(Subscription).where(Subscription.id == subscription.id).with_for_update() + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) ) subscription = locked_result.scalar_one() @@ -6192,6 +6195,45 @@ async def update_subscription_devices_endpoint( description=f'{description} на {charged_months or get_remaining_months(subscription.end_date)} мес', ) + if price_to_charge > 0: + # Re-lock subscription after subtract_user_balance committed (which released all locks). + # Re-validate to prevent concurrent device purchases from exceeding the limit or double-charging. + relock_result = await db.execute( + select(Subscription) + .where(Subscription.id == subscription.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + subscription = relock_result.scalar_one() + + actual_current = subscription.device_limit or 1 + actual_delta = new_devices - actual_current + max_devices_limit = settings.MAX_DEVICES_LIMIT + + if actual_delta <= 0 or (max_devices_limit > 0 and new_devices > max_devices_limit): + # Concurrent request already applied the change or pushed limit beyond max — refund + user_refund = await db.execute( + select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True) + ) + refund_user = user_refund.scalar_one() + refund_user.balance_kopeks += price_to_charge + await db.commit() + if actual_delta <= 0: + raise HTTPException( + status.HTTP_409_CONFLICT, + detail={ + 'code': 'already_applied', + 'message': 'Изменение уже применено параллельным запросом. Баланс возвращён.', + }, + ) + raise HTTPException( + status.HTTP_409_CONFLICT, + detail={ + 'code': 'devices_limit_exceeded', + 'message': f'Превышен максимальный лимит устройств ({max_devices_limit}). Баланс возвращён.', + }, + ) + subscription.device_limit = new_devices subscription.updated_at = datetime.now(UTC) await db.commit()