diff --git a/app/database/crud/user.py b/app/database/crud/user.py index 98ee1f75..69b091b8 100644 --- a/app/database/crud/user.py +++ b/app/database/crud/user.py @@ -3,9 +3,10 @@ import secrets import string from datetime import datetime, timedelta from typing import Optional, List, Dict -from sqlalchemy import select, and_, or_, func, case, nullslast +from sqlalchemy import select, and_, or_, func, case, nullslast, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload +from sqlalchemy.exc import IntegrityError from app.database.models import ( User, @@ -85,6 +86,37 @@ async def create_unique_referral_code(db: AsyncSession) -> str: return f"ref{timestamp}" +async def _sync_users_sequence(db: AsyncSession) -> None: + """Ensure the users.id sequence matches the current max ID.""" + await db.execute( + text( + "SELECT setval('users_id_seq', " + "COALESCE((SELECT MAX(id) FROM users), 0) + 1, false)" + ) + ) + await db.commit() + logger.warning( + "🔄 Последовательность users_id_seq была синхронизирована с текущим максимумом id" + ) + + +async def _get_or_create_default_promo_group(db: AsyncSession) -> PromoGroup: + default_group = await get_default_promo_group(db) + if default_group: + return default_group + + default_group = PromoGroup( + name="Базовый юзер", + server_discount_percent=0, + traffic_discount_percent=0, + device_discount_percent=0, + is_default=True, + ) + db.add(default_group) + await db.flush() + return default_group + + async def create_user( db: AsyncSession, telegram_id: int, @@ -99,46 +131,61 @@ async def create_user( if not referral_code: referral_code = await create_unique_referral_code(db) - default_group = await get_default_promo_group(db) - if not default_group: - default_group = PromoGroup( - name="Базовый юзер", - server_discount_percent=0, - traffic_discount_percent=0, - device_discount_percent=0, - is_default=True, + attempts = 3 + + for attempt in range(1, attempts + 1): + default_group = await _get_or_create_default_promo_group(db) + promo_group_id = default_group.id + + safe_first = sanitize_telegram_name(first_name) + safe_last = sanitize_telegram_name(last_name) + user = User( + telegram_id=telegram_id, + username=username, + first_name=safe_first, + last_name=safe_last, + language=language, + referred_by_id=referred_by_id, + referral_code=referral_code, + balance_kopeks=0, + has_had_paid_subscription=False, + has_made_first_topup=False, + promo_group_id=promo_group_id, ) - db.add(default_group) - await db.flush() - promo_group_id = default_group.id + db.add(user) - safe_first = sanitize_telegram_name(first_name) - safe_last = sanitize_telegram_name(last_name) - user = User( - telegram_id=telegram_id, - username=username, - first_name=safe_first, - last_name=safe_last, - language=language, - referred_by_id=referred_by_id, - referral_code=referral_code, - balance_kopeks=0, - has_had_paid_subscription=False, - has_made_first_topup=False, - promo_group_id=promo_group_id, - ) - - db.add(user) - await db.commit() - await db.refresh(user) - - if default_group: - user.promo_group = default_group + try: + await db.commit() + await db.refresh(user) - logger.info(f"✅ Создан пользователь {telegram_id} с реферальным кодом {referral_code}") + user.promo_group = default_group + logger.info( + f"✅ Создан пользователь {telegram_id} с реферальным кодом {referral_code}" + ) + return user - return user + except IntegrityError as exc: + await db.rollback() + + if ( + isinstance(getattr(exc, "orig", None), Exception) + and "users_pkey" in str(exc.orig) + and attempt < attempts + ): + logger.warning( + "⚠️ Обнаружено несоответствие последовательности users_id_seq при создании пользователя %s. " + "Выполняем повторную синхронизацию (попытка %s/%s)", + telegram_id, + attempt, + attempts, + ) + await _sync_users_sequence(db) + continue + + raise + + raise RuntimeError("Не удалось создать пользователя после синхронизации последовательности") async def update_user( diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py index 04d20494..f56ded44 100644 --- a/app/database/universal_migration.py +++ b/app/database/universal_migration.py @@ -72,13 +72,8 @@ async def sync_postgres_sequences() -> bool: seq_name = seq_name.strip('"') current_result = await conn.execute( text( - """ - SELECT last_value, is_called - FROM pg_sequences - WHERE schemaname = :schema AND sequencename = :sequence - """ - ), - {"schema": seq_schema, "sequence": seq_name}, + f'SELECT last_value, is_called FROM "{seq_schema}"."{seq_name}"' + ) ) current_row = current_result.fetchone() diff --git a/app/services/subscription_service.py b/app/services/subscription_service.py index e398b286..012bf6f8 100644 --- a/app/services/subscription_service.py +++ b/app/services/subscription_service.py @@ -1,4 +1,5 @@ import logging +from contextlib import asynccontextmanager from datetime import datetime, timedelta from typing import Optional, List, Tuple from sqlalchemy.ext.asyncio import AsyncSession @@ -6,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.database.models import Subscription, User, SubscriptionStatus, PromoGroup from app.external.remnawave_api import ( - RemnaWaveAPI, RemnaWaveUser, UserStatus, + RemnaWaveAPI, RemnaWaveUser, UserStatus, TrafficLimitStrategy, RemnaWaveAPIError ) from app.database.crud.user import get_user_by_id @@ -75,16 +76,54 @@ def get_traffic_reset_strategy(): class SubscriptionService: - + def __init__(self): auth_params = settings.get_remnawave_auth_params() - self.api = RemnaWaveAPI( - base_url=auth_params["base_url"], - api_key=auth_params["api_key"], - secret_key=auth_params["secret_key"], - username=auth_params["username"], - password=auth_params["password"] - ) + base_url = (auth_params.get("base_url") or "").strip() + api_key = (auth_params.get("api_key") or "").strip() + + self._config_error: Optional[str] = None + + if not base_url: + self._config_error = "REMNAWAVE_API_URL не настроен" + elif not api_key: + self._config_error = "REMNAWAVE_API_KEY не настроен" + + if self._config_error: + logger.warning( + "RemnaWave API недоступен: %s. Подписочный сервис будет работать в оффлайн-режиме.", + self._config_error + ) + self.api = None + else: + self.api = RemnaWaveAPI( + base_url=base_url, + api_key=api_key, + secret_key=auth_params.get("secret_key"), + username=auth_params.get("username"), + password=auth_params.get("password") + ) + + @property + def is_configured(self) -> bool: + return self._config_error is None + + @property + def configuration_error(self) -> Optional[str]: + return self._config_error + + def _ensure_configured(self) -> None: + if not self.api or not self.is_configured: + raise RemnaWaveAPIError( + self._config_error or "RemnaWave API не настроен" + ) + + @asynccontextmanager + async def get_api_client(self): + self._ensure_configured() + assert self.api is not None + async with self.api as api: + yield api async def create_remnawave_user( self, @@ -106,7 +145,7 @@ class SubscriptionService: logger.error(f"Ошибка валидации подписки для пользователя {user.telegram_id}") return None - async with self.api as api: + async with self.get_api_client() as api: existing_users = await api.get_user_by_telegram_id(user.telegram_id) if existing_users: logger.info(f"🔄 Найден существующий пользователь в панели для {user.telegram_id}") @@ -216,7 +255,7 @@ class SubscriptionService: is_actually_active = False logger.info(f"🔔 Статус подписки {subscription.id} автоматически изменен на 'expired'") - async with self.api as api: + async with self.get_api_client() as api: updated_user = await api.update_user( uuid=user.remnawave_uuid, status=UserStatus.ACTIVE if is_actually_active else UserStatus.EXPIRED, @@ -281,7 +320,7 @@ class SubscriptionService: async def disable_remnawave_user(self, user_uuid: str) -> bool: try: - async with self.api as api: + async with self.get_api_client() as api: await api.disable_user(user_uuid) logger.info(f"✅ Отключен RemnaWave пользователь {user_uuid}") return True @@ -301,7 +340,7 @@ class SubscriptionService: if not user or not user.remnawave_uuid: return None - async with self.api as api: + async with self.get_api_client() as api: updated_user = await api.revoke_user_subscription(user.remnawave_uuid) subscription.remnawave_short_uuid = updated_user.short_uuid @@ -319,7 +358,7 @@ class SubscriptionService: async def get_subscription_info(self, short_uuid: str) -> Optional[dict]: try: - async with self.api as api: + async with self.get_api_client() as api: info = await api.get_subscription_info(short_uuid) return info @@ -338,7 +377,7 @@ class SubscriptionService: if not user or not user.remnawave_uuid: return False - async with self.api as api: + async with self.get_api_client() as api: remnawave_user = await api.get_user_by_uuid(user.remnawave_uuid) if not remnawave_user: return False @@ -585,7 +624,7 @@ class SubscriptionService: if user.remnawave_uuid: try: - async with self.api as api: + async with self.get_api_client() as api: remnawave_user = await api.get_user_by_uuid(user.remnawave_uuid) if not remnawave_user: