Merge pull request #16 from Fr1ngg/FixDbCollisions

Fix db collisions

app/database/crud/subscription.py

@@ -19,6 +19,8 @@ async def get_subscription_by_user_id(db: AsyncSession, user_id: int) -> Optiona
         select(Subscription)
         .options(selectinload(Subscription.user))
         .where(Subscription.user_id == user_id)
+        .order_by(Subscription.created_at.desc())
+        .limit(1) 
     )
     subscription = result.scalar_one_or_none()
     
@@ -98,17 +100,30 @@ async def extend_subscription(
     subscription: Subscription,
     days: int
 ) -> Subscription:
+    current_time = datetime.utcnow()
     
-    subscription.extend_subscription(days)
+    logger.info(f"🔄 Продление подписки {subscription.id} на {days} дней")
+    logger.info(f"📊 Текущие параметры: статус={subscription.status}, окончание={subscription.end_date}")
+    
+    if subscription.end_date > current_time:
+        subscription.end_date = subscription.end_date + timedelta(days=days)
+        logger.info(f"📅 Подписка активна, добавляем {days} дней к текущей дате окончания")
+    else:
+        subscription.end_date = current_time + timedelta(days=days)
+        logger.info(f"📅 Подписка истекла, устанавливаем новую дату окончания")
     
     if subscription.status == SubscriptionStatus.EXPIRED.value:
         subscription.status = SubscriptionStatus.ACTIVE.value
+        logger.info(f"🔄 Статус изменён с EXPIRED на ACTIVE")
+    
+    subscription.updated_at = current_time
     
-    subscription.updated_at = datetime.utcnow()
     await db.commit()
     await db.refresh(subscription)
     
-    logger.info(f"⏰ Подписка пользователя {subscription.user_id} продлена на {days} дней")
+    logger.info(f"✅ Подписка продлена до: {subscription.end_date}")
+    logger.info(f"📊 Новые параметры: статус={subscription.status}, окончание={subscription.end_date}")
+    
     return subscription
 
 

app/database/models.py

@@ -104,7 +104,7 @@ class Subscription(Base):
     __tablename__ = "subscriptions"
     
     id = Column(Integer, primary_key=True, index=True)
-    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, unique=True)
     
     status = Column(String(20), default=SubscriptionStatus.TRIAL.value)
     is_trial = Column(Boolean, default=True)

app/database/universal_migration.py

@@ -0,0 +1,185 @@
+import logging
+from sqlalchemy import text, inspect
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.database.database import engine
+
+logger = logging.getLogger(__name__)
+
+async def get_database_type():
+    """Определяет тип базы данных"""
+    return engine.dialect.name
+
+async def check_unique_constraint_exists():
+    """Проверяет, существует ли ограничение уникальности на user_id"""
+    try:
+        async with engine.begin() as conn:
+            db_type = await get_database_type()
+            
+            if db_type == 'sqlite':
+                result = await conn.execute(text("PRAGMA table_info(subscriptions)"))
+                columns = result.fetchall()
+                
+                check_result = await conn.execute(text("""
+                    SELECT user_id, COUNT(*) as count 
+                    FROM subscriptions 
+                    GROUP BY user_id 
+                    HAVING COUNT(*) > 1
+                    LIMIT 1
+                """))
+                
+                duplicates = check_result.fetchall()
+                return len(duplicates) == 0 
+                
+            elif db_type == 'postgresql':
+                result = await conn.execute(text("""
+                    SELECT constraint_name 
+                    FROM information_schema.table_constraints 
+                    WHERE table_name = 'subscriptions' 
+                    AND constraint_type = 'UNIQUE'
+                    AND constraint_name LIKE '%user_id%'
+                """))
+                constraints = result.fetchall()
+                return len(constraints) > 0
+                
+            elif db_type == 'mysql':
+                result = await conn.execute(text("""
+                    SELECT CONSTRAINT_NAME 
+                    FROM information_schema.TABLE_CONSTRAINTS 
+                    WHERE TABLE_NAME = 'subscriptions' 
+                    AND CONSTRAINT_TYPE = 'UNIQUE'
+                    AND CONSTRAINT_NAME LIKE '%user_id%'
+                """))
+                constraints = result.fetchall()
+                return len(constraints) > 0
+                
+            return False
+            
+    except Exception as e:
+        logger.error(f"Ошибка проверки ограничения уникальности: {e}")
+        return False
+
+async def fix_subscription_duplicates_universal():
+    """Универсальная функция очистки дубликатов для разных типов БД"""
+    
+    async with engine.begin() as conn:
+        db_type = await get_database_type()
+        logger.info(f"Обнаружен тип базы данных: {db_type}")
+        
+        try:
+            result = await conn.execute(text("""
+                SELECT user_id, COUNT(*) as count 
+                FROM subscriptions 
+                GROUP BY user_id 
+                HAVING COUNT(*) > 1
+            """))
+            
+            duplicates = result.fetchall()
+            
+            if not duplicates:
+                logger.info("Дублирующихся подписок не найдено")
+                return 0
+                
+            logger.info(f"Найдено {len(duplicates)} пользователей с дублирующимися подписками")
+            
+            total_deleted = 0
+            
+            for user_id_row, count in duplicates:
+                user_id = user_id_row
+                
+                if db_type == 'sqlite':
+                    delete_result = await conn.execute(text("""
+                        DELETE FROM subscriptions 
+                        WHERE user_id = :user_id AND id NOT IN (
+                            SELECT MAX(id) 
+                            FROM subscriptions 
+                            WHERE user_id = :user_id
+                        )
+                    """), {"user_id": user_id})
+                    
+                elif db_type in ['postgresql', 'mysql']:
+                    delete_result = await conn.execute(text("""
+                        DELETE FROM subscriptions 
+                        WHERE user_id = :user_id AND id NOT IN (
+                            SELECT max_id FROM (
+                                SELECT MAX(id) as max_id
+                                FROM subscriptions 
+                                WHERE user_id = :user_id
+                            ) as subquery
+                        )
+                    """), {"user_id": user_id})
+                
+                else:
+                    subs_result = await conn.execute(text("""
+                        SELECT id FROM subscriptions 
+                        WHERE user_id = :user_id 
+                        ORDER BY created_at DESC, id DESC
+                    """), {"user_id": user_id})
+                    
+                    sub_ids = [row[0] for row in subs_result.fetchall()]
+                    
+                    if len(sub_ids) > 1:
+                        ids_to_delete = sub_ids[1:]
+                        for sub_id in ids_to_delete:
+                            await conn.execute(text("""
+                                DELETE FROM subscriptions WHERE id = :id
+                            """), {"id": sub_id})
+                        delete_result = type('Result', (), {'rowcount': len(ids_to_delete)})()
+                    else:
+                        delete_result = type('Result', (), {'rowcount': 0})()
+                
+                deleted_count = delete_result.rowcount
+                total_deleted += deleted_count
+                logger.info(f"Удалено {deleted_count} дублирующихся подписок для пользователя {user_id}")
+            
+            logger.info(f"Всего удалено дублирующихся подписок: {total_deleted}")
+            return total_deleted
+            
+        except Exception as e:
+            logger.error(f"Ошибка при очистке дублирующихся подписок: {e}")
+            raise
+
+async def run_universal_migration():
+    """Запускает универсальную миграцию"""
+    
+    logger.info("=== НАЧАЛО УНИВЕРСАЛЬНОЙ МИГРАЦИИ ПОДПИСОК ===")
+    
+    try:
+        db_type = await get_database_type()
+        logger.info(f"Тип базы данных: {db_type}")
+        
+        async with engine.begin() as conn:
+            total_subs = await conn.execute(text("SELECT COUNT(*) FROM subscriptions"))
+            unique_users = await conn.execute(text("SELECT COUNT(DISTINCT user_id) FROM subscriptions"))
+            
+            total_count = total_subs.fetchone()[0]
+            unique_count = unique_users.fetchone()[0]
+            
+            logger.info(f"Всего подписок: {total_count}")
+            logger.info(f"Уникальных пользователей: {unique_count}")
+            
+            if total_count == unique_count:
+                logger.info("База данных уже в корректном состоянии")
+                return True
+        
+        deleted_count = await fix_subscription_duplicates_universal()
+        
+        async with engine.begin() as conn:
+            final_check = await conn.execute(text("""
+                SELECT user_id, COUNT(*) as count 
+                FROM subscriptions 
+                GROUP BY user_id 
+                HAVING COUNT(*) > 1
+            """))
+            
+            remaining_duplicates = final_check.fetchall()
+            
+            if remaining_duplicates:
+                logger.warning(f"Остались дубликаты у {len(remaining_duplicates)} пользователей")
+                return False
+            else:
+                logger.info("=== МИГРАЦИЯ ЗАВЕРШЕНА УСПЕШНО ===")
+                return True
+                
+    except Exception as e:
+        logger.error(f"=== ОШИБКА ВЫПОЛНЕНИЯ МИГРАЦИИ: {e} ===")
+        return False

app/handlers/subscription.py

@@ -906,19 +906,30 @@ async def confirm_extend_subscription(
     db_user: User,
     db: AsyncSession
 ):
-    
     days = int(callback.data.split('_')[2])
     texts = get_texts(db_user.language)
     subscription = db_user.subscription
     
+    if not subscription:
+        await callback.answer("❌ У вас нет активной подписки", show_alert=True)
+        return
+    
     subscription_service = SubscriptionService()
-    price = await subscription_service.calculate_renewal_price(subscription, days, db)
+    
+    try:
+        price = await subscription_service.calculate_renewal_price(subscription, days, db)
+    except Exception as e:
+        logger.error(f"❌ ОШИБКА РАСЧЕТА ЦЕНЫ: {e}")
+        await callback.answer("❌ Ошибка расчета стоимости", show_alert=True)
+        return
     
     if db_user.balance_kopeks < price:
         await callback.answer("❌ Недостаточно средств на балансе", show_alert=True)
         return
     
     try:
+        logger.info(f"🔄 Начинаем продление подписки {subscription.id} на {days} дней за {price/100}₽")
+        
         success = await subtract_user_balance(
             db, db_user, price,
             f"Продление подписки на {days} дней"
@@ -928,18 +939,40 @@ async def confirm_extend_subscription(
             await callback.answer("❌ Ошибка списания средств", show_alert=True)
             return
         
-        await extend_subscription(db, subscription, days)
+        current_time = datetime.utcnow()
         
-        subscription_service = SubscriptionService()
-        await subscription_service.update_remnawave_user(db, subscription)
+        if subscription.end_date > current_time:
+            subscription.end_date = subscription.end_date + timedelta(days=days)
+        else:
+            subscription.end_date = current_time + timedelta(days=days)
         
-        await create_transaction(
-            db=db,
-            user_id=db_user.id,
-            type=TransactionType.SUBSCRIPTION_PAYMENT,
-            amount_kopeks=price,
-            description=f"Продление подписки на {days} дней"
-        )
+        subscription.status = SubscriptionStatus.ACTIVE.value
+        subscription.updated_at = current_time
+        
+        await db.commit()
+        await db.refresh(subscription)
+        await db.refresh(db_user)
+        
+        try:
+            remnawave_result = await subscription_service.update_remnawave_user(db, subscription)
+            if remnawave_result:
+                logger.info(f"✅ RemnaWave обновлен успешно")
+            else:
+                logger.error(f"❌ ОШИБКА ОБНОВЛЕНИЯ REMNAWAVE")
+        except Exception as e:
+            logger.error(f"❌ ИСКЛЮЧЕНИЕ ПРИ ОБНОВЛЕНИИ REMNAWAVE: {e}")
+        
+        try:
+            transaction = await create_transaction(
+                db=db,
+                user_id=db_user.id,
+                type=TransactionType.SUBSCRIPTION_PAYMENT,
+                amount_kopeks=price,
+                description=f"Продление подписки на {days} дней"
+            )
+            logger.info(f"✅ Транзакция создана: ID {transaction.id}")
+        except Exception as e:
+            logger.error(f"❌ ОШИБКА СОЗДАНИЯ ТРАНЗАКЦИИ: {e}")
         
         try:
             await process_referral_purchase(
@@ -948,11 +981,9 @@ async def confirm_extend_subscription(
                 purchase_amount_kopeks=price,
                 transaction_id=None
             )
+            logger.info(f"✅ Рефералы обработаны")
         except Exception as e:
-            logger.error(f"Ошибка обработки реферальной покупки: {e}")
-        
-        await db.refresh(db_user)
-        await db.refresh(subscription)
+            logger.error(f"❌ ОШИБКА ОБРАБОТКИ РЕФЕРАЛОВ: {e}")
         
         await callback.message.edit_text(
             f"✅ Подписка успешно продлена!\n\n"
@@ -965,9 +996,12 @@ async def confirm_extend_subscription(
         logger.info(f"✅ Пользователь {db_user.telegram_id} продлил подписку на {days} дней за {price/100}₽")
         
     except Exception as e:
-        logger.error(f"Ошибка продления подписки: {e}")
+        logger.error(f"❌ КРИТИЧЕСКАЯ ОШИБКА ПРОДЛЕНИЯ: {e}")
+        import traceback
+        logger.error(f"TRACEBACK: {traceback.format_exc()}")
+        
         await callback.message.edit_text(
-            texts.ERROR,
+            "❌ Произошла ошибка при продлении подписки. Обратитесь в поддержку.",
             reply_markup=get_back_keyboard(db_user.language)
         )
     
@@ -1429,27 +1463,29 @@ async def confirm_purchase(
         
         existing_subscription = db_user.subscription
         
-        if existing_subscription and existing_subscription.is_trial:
-            logger.info(f"🔄 Обновляем триальную подписку пользователя {db_user.telegram_id}")
+        if existing_subscription:
+            logger.info(f"🔄 Обновляем существующую подписку пользователя {db_user.telegram_id}")
             
             existing_subscription.is_trial = False
             existing_subscription.status = SubscriptionStatus.ACTIVE.value
-            
             existing_subscription.traffic_limit_gb = final_traffic_gb
             existing_subscription.device_limit = data['devices']
             existing_subscription.connected_squads = data['countries']
             
-            existing_subscription.extend_subscription(data['period_days'])
+            existing_subscription.start_date = datetime.utcnow()
+            existing_subscription.end_date = datetime.utcnow() + timedelta(days=data['period_days'])
             existing_subscription.updated_at = datetime.utcnow()
             
+            existing_subscription.traffic_used_gb = 0.0
+            
             await db.commit()
             await db.refresh(existing_subscription)
             subscription = existing_subscription
             
-            logger.info(f"✅ Триальная подписка обновлена до платной. Новая дата окончания: {subscription.end_date}")
+            logger.info(f"✅ Подписка обновлена. Новая дата окончания: {subscription.end_date}")
             
         else:
-            logger.info(f"🆕 Создаем новую платную подписку для пользователя {db_user.telegram_id}")
+            logger.info(f"🆕 Создаем новую подписку для пользователя {db_user.telegram_id}")
             subscription = await create_paid_subscription_with_traffic_mode(
                 db=db,
                 user_id=db_user.id,
@@ -1479,7 +1515,7 @@ async def confirm_purchase(
         subscription_service = SubscriptionService()
         
         if db_user.remnawave_uuid:
-            logger.info(f"📝 Обновляем существующего RemnaWave пользователя {db_user.remnawave_uuid}")
+            logger.info(f"🔄 Обновляем существующего RemnaWave пользователя {db_user.remnawave_uuid}")
             remnawave_user = await subscription_service.update_remnawave_user(db, subscription)
         else:
             logger.info(f"🆕 Создаем нового RemnaWave пользователя для {db_user.telegram_id}")

app/utils/subscription_utils.py

@@ -0,0 +1,97 @@
+import logging
+from datetime import datetime
+from typing import Optional
+from sqlalchemy import select, delete
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.database.models import Subscription, User
+
+logger = logging.getLogger(__name__)
+
+
+async def ensure_single_subscription(db: AsyncSession, user_id: int) -> Optional[Subscription]:
+    result = await db.execute(
+        select(Subscription)
+        .where(Subscription.user_id == user_id)
+        .order_by(Subscription.created_at.desc())
+    )
+    subscriptions = result.scalars().all()
+    
+    if len(subscriptions) <= 1:
+        return subscriptions[0] if subscriptions else None
+    
+    latest_subscription = subscriptions[0]
+    old_subscriptions = subscriptions[1:]
+    
+    logger.warning(f"🚨 Обнаружено {len(subscriptions)} подписок у пользователя {user_id}. Удаляем {len(old_subscriptions)} старых.")
+    
+    for old_sub in old_subscriptions:
+        await db.delete(old_sub)
+        logger.info(f"🗑️ Удалена подписка ID {old_sub.id} от {old_sub.created_at}")
+    
+    await db.commit()
+    await db.refresh(latest_subscription)
+    
+    logger.info(f"✅ Оставлена подписка ID {latest_subscription.id} от {latest_subscription.created_at}")
+    return latest_subscription
+
+
+async def update_or_create_subscription(
+    db: AsyncSession,
+    user_id: int,
+    **subscription_data
+) -> Subscription:
+    existing_subscription = await ensure_single_subscription(db, user_id)
+    
+    if existing_subscription:
+        for key, value in subscription_data.items():
+            if hasattr(existing_subscription, key):
+                setattr(existing_subscription, key, value)
+        
+        existing_subscription.updated_at = datetime.utcnow()
+        await db.commit()
+        await db.refresh(existing_subscription)
+        
+        logger.info(f"🔄 Обновлена существующая подписка ID {existing_subscription.id}")
+        return existing_subscription
+    
+    else:
+        new_subscription = Subscription(
+            user_id=user_id,
+            **subscription_data
+        )
+        
+        db.add(new_subscription)
+        await db.commit()
+        await db.refresh(new_subscription)
+        
+        logger.info(f"🆕 Создана новая подписка ID {new_subscription.id}")
+        return new_subscription
+
+
+async def cleanup_duplicate_subscriptions(db: AsyncSession) -> int:
+    result = await db.execute(
+        select(Subscription.user_id)
+        .group_by(Subscription.user_id)
+        .having(func.count(Subscription.id) > 1)
+    )
+    users_with_duplicates = result.scalars().all()
+    
+    total_deleted = 0
+    
+    for user_id in users_with_duplicates:
+        subscriptions_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.user_id == user_id)
+            .order_by(Subscription.created_at.desc())
+        )
+        subscriptions = subscriptions_result.scalars().all()
+        
+        for old_subscription in subscriptions[1:]:
+            await db.delete(old_subscription)
+            total_deleted += 1
+            logger.info(f"🗑️ Удалена дублирующаяся подписка ID {old_subscription.id} пользователя {user_id}")
+    
+    await db.commit()
+    logger.info(f"🧹 Очищено {total_deleted} дублирующихся подписок")
+    
+    return total_deleted

main.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import sys
+import os
 from pathlib import Path
 
 sys.path.append(str(Path(__file__).parent))
@@ -10,6 +11,7 @@ from app.config import settings
 from app.database.database import init_db
 from app.services.monitoring_service import monitoring_service
 from app.external.webhook_server import WebhookServer
+from app.database.universal_migration import run_universal_migration
 
 
 async def main():
@@ -23,7 +25,7 @@ async def main():
     )
     
     logger = logging.getLogger(__name__)
-    logger.info("🚀 Запуск VPN бота...")
+    logger.info("🚀 Запуск Bedolaga Remnawave Bot...")
     
     webhook_server = None
     
@@ -31,12 +33,29 @@ async def main():
         logger.info("📊 Инициализация базы данных...")
         await init_db()
         
+        skip_migration = os.getenv('SKIP_MIGRATION', 'false').lower() == 'true'
+        
+        if not skip_migration:
+            logger.info("🔧 Выполняем проверку и миграцию базы данных...")
+            try:
+                migration_success = await run_universal_migration()
+                
+                if migration_success:
+                    logger.info("✅ Миграция базы данных завершена успешно")
+                else:
+                    logger.warning("⚠️ Миграция завершилась с предупреждениями, но продолжаем запуск")
+                    
+            except Exception as migration_error:
+                logger.error(f"❌ Ошибка выполнения миграции: {migration_error}")
+                logger.warning("⚠️ Продолжаем запуск без миграции")
+        else:
+            logger.info("ℹ️ Миграция пропущена (SKIP_MIGRATION=true)")
+        
         logger.info("🤖 Настройка бота...")
         bot, dp = await setup_bot()
         
         monitoring_service.bot = bot
         
-        # Инициализируем webhook сервер если Tribute включен
         if settings.TRIBUTE_ENABLED:
             logger.info("🌐 Запуск webhook сервера для Tribute...")
             webhook_server = WebhookServer(bot)
@@ -78,4 +97,4 @@ if __name__ == "__main__":
         print("\n🛑 Бот остановлен пользователем")
     except Exception as e:
         print(f"❌ Критическая ошибка: {e}")
-        sys.exit(1)
+        sys.exit(1)