fix: atomicity refactor, review fixes, and DELETED recovery logging

- subtract_user_balance: add commit=False parameter for atomic balance+subscription ops
- extend_subscription: add commit=False parameter, propagate to clear_notifications
- wata_service: wire _MIN_EXPIRATION_MINUTES constant to actual usage
- admin_users: fix no-op ternary in sync_user_from_panel timezone normalization
- start.py: log warning when DELETED recovery zeros non-zero balance (3 locations)
- remnawave_service: preserve PromoCodeUse records and used_promocodes in force_cleanup

app/cabinet/routes/admin_users.py

@@ -2515,7 +2515,9 @@ async def sync_user_from_panel(
                 if panel_user.expire_at:
                     panel_expire_utc = panel_datetime_to_utc(panel_user.expire_at)
 
-                    sub_end_utc = sub.end_date if sub.end_date and sub.end_date.tzinfo else sub.end_date
+                    sub_end_utc = sub.end_date
+                    if sub_end_utc is not None and sub_end_utc.tzinfo is None:
+                        sub_end_utc = sub_end_utc.replace(tzinfo=UTC)
                     if sub_end_utc != panel_expire_utc:
                         # Предупреждаем если локальная дата новее панельной
                         # (например, автопокупка уже продлила подписку)

app/database/crud/subscription.py

@@ -382,6 +382,7 @@ async def extend_subscription(
     traffic_limit_gb: int | None = None,
     device_limit: int | None = None,
     connected_squads: list[str] | None = None,
+    commit: bool = True,
 ) -> Subscription:
     """Продлевает подписку на указанное количество дней.
 
@@ -594,9 +595,13 @@ async def extend_subscription(
 
     subscription.updated_at = current_time
 
-    await db.commit()
-    await db.refresh(subscription, ['tariff'])
-    await clear_notifications(db, subscription.id)
+    if commit:
+        await db.commit()
+        await db.refresh(subscription, ['tariff'])
+    else:
+        await db.flush()
+
+    await clear_notifications(db, subscription.id, commit=commit)
 
     logger.info('✅ Подписка продлена до', end_date=subscription.end_date)
     logger.info('📊 Новые параметры: статус=, окончание', status=subscription.status, end_date=subscription.end_date)

app/database/crud/user.py

@@ -501,6 +501,7 @@ async def subtract_user_balance(
     transaction_type: TransactionType = TransactionType.WITHDRAWAL,
     consume_promo_offer: bool = False,
     mark_as_paid_subscription: bool = False,
+    commit: bool = True,
 ) -> bool:
     user_id_display = user.telegram_id or user.email or f'#{user.id}'
     logger.info('💸 ОТЛАДКА subtract_user_balance:')
@@ -572,8 +573,6 @@ async def subtract_user_balance(
                 create_transaction as create_trans,
             )
 
-            # create_trans commits the session, atomically persisting
-            # both the balance change and the transaction record
             await create_trans(
                 db=db,
                 user_id=user.id,
@@ -581,11 +580,15 @@ async def subtract_user_balance(
                 amount_kopeks=amount_kopeks,
                 description=description,
                 payment_method=payment_method,
+                commit=commit,
             )
-        else:
+        elif commit:
             await db.commit()
+        else:
+            await db.flush()
 
-        await db.refresh(user)
+        if commit:
+            await db.refresh(user)
 
         if consume_promo_offer and log_context:
             try:
@@ -598,18 +601,20 @@ async def subtract_user_balance(
                     percent=log_context.get('percent'),
                     effect_type=log_context.get('effect_type'),
                     details=log_context.get('details'),
+                    commit=commit,
                 )
             except Exception as log_error:  # pragma: no cover - defensive logging
                 logger.warning(
                     'Failed to record promo offer consumption log for user', user_id=user.id, log_error=log_error
                 )
-                try:
-                    await db.rollback()
-                except Exception as rollback_error:  # pragma: no cover - defensive logging
-                    logger.warning(
-                        'Failed to rollback session after promo offer consumption log failure',
-                        rollback_error=rollback_error,
-                    )
+                if commit:
+                    try:
+                        await db.rollback()
+                    except Exception as rollback_error:  # pragma: no cover - defensive logging
+                        logger.warning(
+                            'Failed to rollback session after promo offer consumption log failure',
+                            rollback_error=rollback_error,
+                        )
 
         logger.info('✅ Средства списаны: →', old_balance=old_balance, balance_kopeks=user.balance_kopeks)
         return True

app/handlers/start.py

@@ -754,6 +754,13 @@ async def cmd_start(message: types.Message, state: FSMContext, db: AsyncSession,
 
             await db.execute(delete(Transaction).where(Transaction.user_id == user.id))
 
+            if user.balance_kopeks > 0:
+                logger.warning(
+                    '⚠️ DELETED-восстановление: обнуляем ненулевой баланс',
+                    telegram_id=user.telegram_id,
+                    balance_kopeks=user.balance_kopeks,
+                )
+
             user.status = UserStatus.ACTIVE.value
             user.balance_kopeks = 0
             user.remnawave_uuid = None
@@ -1344,6 +1351,13 @@ async def complete_registration_from_callback(callback: types.CallbackQuery, sta
         # Prevent self-referral when partner re-registers via own campaign link
         safe_referrer_id = referrer_id if referrer_id != existing_user.id else None
 
+        if existing_user.balance_kopeks > 0:
+            logger.warning(
+                '⚠️ DELETED-восстановление: обнуляем ненулевой баланс',
+                telegram_id=existing_user.telegram_id,
+                balance_kopeks=existing_user.balance_kopeks,
+            )
+
         existing_user.username = callback.from_user.username
         existing_user.first_name = callback.from_user.first_name
         existing_user.last_name = callback.from_user.last_name
@@ -1639,6 +1653,13 @@ async def complete_registration(message: types.Message, state: FSMContext, db: A
         # Prevent self-referral when partner re-registers via own campaign link
         safe_referrer_id = referrer_id if referrer_id != existing_user.id else None
 
+        if existing_user.balance_kopeks > 0:
+            logger.warning(
+                '⚠️ DELETED-восстановление: обнуляем ненулевой баланс',
+                telegram_id=existing_user.telegram_id,
+                balance_kopeks=existing_user.balance_kopeks,
+            )
+
         existing_user.username = message.from_user.username
         existing_user.first_name = message.from_user.first_name
         existing_user.last_name = message.from_user.last_name

app/services/remnawave_service.py

@@ -2425,7 +2425,6 @@ class RemnaWaveService:
                 from sqlalchemy import delete
 
                 from app.database.models import (
-                    PromoCodeUse,
                     ReferralEarning,
                     SubscriptionServer,
                     SubscriptionStatus,
@@ -2447,8 +2446,8 @@ class RemnaWaveService:
                 await db.execute(delete(ReferralEarning).where(ReferralEarning.referral_id == user.id))
                 logger.info('🗑️ Удалены реферальные доходы для', user_id_display=user_id_display)
 
-                await db.execute(delete(PromoCodeUse).where(PromoCodeUse.user_id == user.id))
-                logger.info('🗑️ Удалены использования промокодов для', user_id_display=user_id_display)
+                # PromoCodeUse НЕ удаляем — история промокодов постоянна,
+                # иначе пользователь может повторно активировать промокоды
 
             except Exception as records_error:
                 logger.error('❌ Ошибка удаления связанных записей', records_error=records_error)
@@ -2461,7 +2460,6 @@ class RemnaWaveService:
                         balance_kopeks=user.balance_kopeks,
                     )
                 user.remnawave_uuid = None
-                user.used_promocodes = 0
                 user.updated_at = self._now_utc()
 
                 if user.subscription:

app/services/wata_service.py

@@ -15,6 +15,10 @@ from app.config import settings
 
 logger = structlog.get_logger(__name__)
 
+# WATA API rejects expirationDateTime <= now + 10 minutes (exclusive lower bound).
+# 15 minutes provides a 5-minute buffer against clock skew and request latency.
+_MIN_EXPIRATION_MINUTES = 15
+
 
 class WataAPIError(RuntimeError):
     """Raised when the WATA API returns an error response."""
@@ -196,7 +200,7 @@ class WataService:
         if expiration_minutes:
             # WATA API требует expirationDateTime строго > now + 10 минут.
             # Принудительный минимум 15 минут, чтобы не попасть на границу.
-            safe_minutes = max(expiration_minutes, 15)
+            safe_minutes = max(expiration_minutes, _MIN_EXPIRATION_MINUTES)
             expiration_time = datetime.now(UTC) + timedelta(minutes=safe_minutes)
             payload['expirationDateTime'] = self._format_datetime(expiration_time)