remnawave-bedolaga-telegram-bot/app/database/crud/user.py

import secrets
import string
from datetime import UTC, datetime, timedelta

import structlog
from sqlalchemy import and_, case, exists, func, nullslast, or_, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.config import settings
from app.database.crud.discount_offer import get_latest_claimed_offer_for_user
from app.database.crud.promo_group import get_default_promo_group
from app.database.crud.promo_offer_log import log_promo_offer_action
from app.database.models import (
    AdvertisingCampaign,
    AdvertisingCampaignRegistration,
    PaymentMethod,
    PromoGroup,
    Subscription,
    SubscriptionStatus,
    Transaction,
    TransactionType,
    User,
    UserPromoGroup,
    UserStatus,
)
from app.utils.validators import sanitize_telegram_name


logger = structlog.get_logger(__name__)


def _normalize_language_code(language: str | None, fallback: str = 'ru') -> str:
    normalized = (language or '').strip().lower()
    if '-' in normalized:
        normalized = normalized.split('-', 1)[0]
    return normalized or fallback


def _build_spending_stats_select():
    """
    Возвращает базовый SELECT для статистики трат пользователей.

    Используется в:
    - get_users_list() для сортировки по тратам/покупкам
    - get_users_spending_stats() для получения статистики

    Returns:
        Tuple колонок (user_id, total_spent, purchase_count)
    """

    return (
        Transaction.user_id.label('user_id'),
        func.coalesce(
            func.sum(
                case(
                    (
                        Transaction.type == TransactionType.SUBSCRIPTION_PAYMENT.value,
                        func.abs(Transaction.amount_kopeks),
                    ),
                    else_=0,
                )
            ),
            0,
        ).label('total_spent'),
        func.coalesce(
            func.sum(
                case(
                    (
                        Transaction.type == TransactionType.SUBSCRIPTION_PAYMENT.value,
                        1,
                    ),
                    else_=0,
                )
            ),
            0,
        ).label('purchase_count'),
    )


def generate_referral_code() -> str:
    alphabet = string.ascii_letters + string.digits
    code_suffix = ''.join(secrets.choice(alphabet) for _ in range(8))
    return f'ref{code_suffix}'


async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.referrer),
            selectinload(User.promo_group),
        )
        .where(User.id == user_id)
    )
    user = result.scalar_one_or_none()

    if user and user.subscription:
        # Загружаем дополнительные зависимости для subscription
        _ = user.subscription.is_active

    return user


async def get_user_by_telegram_id(db: AsyncSession, telegram_id: int) -> User | None:
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.referrer),
            selectinload(User.promo_group),
        )
        .where(User.telegram_id == telegram_id)
    )
    user = result.scalar_one_or_none()

    if user and user.subscription:
        # Загружаем дополнительные зависимости для subscription
        _ = user.subscription.is_active

    return user


async def find_phantom_user_by_username(db: AsyncSession, username: str) -> User | None:
    """Find a phantom user created by guest purchase (no telegram_id, auth_type=telegram).

    Used during /start to reconcile phantom users with real Telegram accounts.
    """
    if not username:
        return None

    normalized = username.lower()
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
        )
        .where(
            User.telegram_id.is_(None),
            User.auth_type == 'telegram',
            User.status != UserStatus.DELETED.value,
            func.lower(User.username) == normalized,
        )
        .with_for_update()
    )
    return result.scalars().first()


async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
    if not username:
        return None

    normalized = username.lower()

    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.referrer),
            selectinload(User.promo_group),
        )
        .where(func.lower(User.username) == normalized)
    )

    user = result.scalar_one_or_none()

    if user and user.subscription:
        # Загружаем дополнительные зависимости для subscription
        _ = user.subscription.is_active

    return user


async def get_user_by_referral_code(db: AsyncSession, referral_code: str) -> User | None:
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.promo_group),
            selectinload(User.referrer),
        )
        .where(User.referral_code == referral_code)
    )
    user = result.scalar_one_or_none()

    if user and user.subscription:
        # Загружаем дополнительные зависимости для subscription
        _ = user.subscription.is_active

    return user


async def get_user_by_remnawave_uuid(db: AsyncSession, remnawave_uuid: str) -> User | None:
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.promo_group),
            selectinload(User.referrer),
        )
        .where(User.remnawave_uuid == remnawave_uuid)
    )
    user = result.scalar_one_or_none()

    # Multi-tariff: UUID lives on Subscription, not User
    if not user and settings.is_multi_tariff_enabled():
        from app.database.models import Subscription as _Subscription

        sub_result = await db.execute(
            select(_Subscription)
            .options(
                selectinload(_Subscription.user).selectinload(User.subscriptions).selectinload(_Subscription.tariff)
            )
            .where(_Subscription.remnawave_uuid == remnawave_uuid)
        )
        sub = sub_result.scalar_one_or_none()
        if sub and sub.user:
            user = sub.user

    if user and user.subscription:
        # Загружаем дополнительные зависимости для subscription
        _ = user.subscription.is_active

    return user


async def create_unique_referral_code(db: AsyncSession) -> str:
    max_attempts = 10

    for _ in range(max_attempts):
        code = generate_referral_code()
        existing_user = await get_user_by_referral_code(db, code)
        if not existing_user:
            return code

    timestamp = str(int(datetime.now(UTC).timestamp()))[-6:]
    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_no_commit(
    db: AsyncSession,
    telegram_id: int,
    username: str = None,
    first_name: str = None,
    last_name: str = None,
    language: str = 'ru',
    referred_by_id: int = None,
    referral_code: str = None,
) -> User:
    """
    Создает пользователя без немедленного коммита для пакетной обработки
    """

    if not referral_code:
        referral_code = await create_unique_referral_code(db)
    normalized_language = _normalize_language_code(language)

    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=normalized_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)

    # Обязательно выполняем flush, чтобы получить присвоенный первичный ключ
    await db.flush()

    # Сохраняем ссылку на группу, чтобы дальнейшие операции могли её использовать
    user.promo_group = default_group

    # Не коммитим сразу, оставляем для пакетной обработки
    logger.info(
        '✅ Подготовлен пользователь с реферальным кодом (ожидает коммита)',
        telegram_id=telegram_id,
        referral_code=referral_code,
    )
    return user


async def create_user(
    db: AsyncSession,
    telegram_id: int,
    username: str = None,
    first_name: str = None,
    last_name: str = None,
    language: str = 'ru',
    referred_by_id: int = None,
    referral_code: str = None,
) -> User:
    if not referral_code:
        referral_code = await create_unique_referral_code(db)
    normalized_language = _normalize_language_code(language)

    # If no referrer provided, check Redis for pending referral from /start
    if not referred_by_id and telegram_id:
        try:
            from app.services.referral_service import clear_pending_referral, get_pending_referral

            pending = await get_pending_referral(telegram_id)
            if pending and pending.get('referrer_id'):
                referred_by_id = pending['referrer_id']
                logger.info(
                    'Resolved referral from Redis pending_referral',
                    telegram_id=telegram_id,
                    referrer_id=referred_by_id,
                )
                await clear_pending_referral(telegram_id)
        except Exception as e:
            logger.warning('Failed to check pending referral from Redis', error=e)

    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=normalized_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)

        try:
            await db.commit()
            await db.refresh(user)

            user.promo_group = default_group
            logger.info(
                '✅ Создан пользователь с реферальным кодом', telegram_id=telegram_id, referral_code=referral_code
            )

            # Отправляем событие о создании пользователя
            try:
                from app.services.event_emitter import event_emitter

                await event_emitter.emit(
                    'user.created',
                    {
                        'user_id': user.id,
                        'telegram_id': user.telegram_id,
                        'username': user.username,
                        'first_name': user.first_name,
                        'last_name': user.last_name,
                        'referral_code': user.referral_code,
                        'referred_by_id': user.referred_by_id,
                    },
                    db=db,
                )
            except Exception as error:
                logger.warning('Failed to emit user.created event', error=error)

            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 при создании пользователя . Выполняем повторную синхронизацию (попытка /)',
                    telegram_id=telegram_id,
                    attempt=attempt,
                    attempts=attempts,
                )
                await _sync_users_sequence(db)
                continue

            raise

    raise RuntimeError('Не удалось создать пользователя после синхронизации последовательности')


async def update_user(db: AsyncSession, user: User, **kwargs) -> User:
    from app.utils.validators import sanitize_telegram_name

    for field, value in kwargs.items():
        if field in ('first_name', 'last_name'):
            value = sanitize_telegram_name(value)
        if field == 'language':
            value = _normalize_language_code(value)
        if hasattr(user, field):
            setattr(user, field, value)

    user.updated_at = datetime.now(UTC)
    await db.commit()
    await db.refresh(user)

    return user


async def lock_user_for_update(db: AsyncSession, user: User) -> User:
    """Lock user row with SELECT FOR UPDATE to prevent concurrent balance modifications.

    Returns the refreshed user object with current DB values.
    Must be called within an active transaction before modifying balance_kopeks.
    Eagerly loads key relationships to avoid MissingGreenlet in async context.
    """
    result = await db.execute(
        select(User)
        .where(User.id == user.id)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.promo_group),
            selectinload(User.referrer),
        )
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    return result.scalar_one()


async def add_user_balance(
    db: AsyncSession,
    user: User,
    amount_kopeks: int,
    description: str = 'Пополнение баланса',
    create_transaction: bool = True,
    transaction_type: TransactionType = TransactionType.DEPOSIT,
    bot=None,
    payment_method: PaymentMethod | None = None,
    commit: bool = True,
) -> bool:
    try:
        # Lock the user row to prevent concurrent balance race conditions
        # Eagerly load key relationships to avoid MissingGreenlet in async context
        locked_result = await db.execute(
            select(User)
            .where(User.id == user.id)
            .options(
                selectinload(User.subscriptions).selectinload(Subscription.tariff),
                selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
                selectinload(User.promo_group),
                selectinload(User.referrer),
            )
            .with_for_update()
            .execution_options(populate_existing=True)
        )
        user = locked_result.scalar_one()

        if amount_kopeks < 0:
            logger.error(
                'add_user_balance вызван с отрицательной суммой — используйте subtract_user_balance',
                amount_kopeks=amount_kopeks,
                user_id=user.id,
            )
            return False

        old_balance = user.balance_kopeks
        user.balance_kopeks += amount_kopeks
        user.updated_at = datetime.now(UTC)

        if create_transaction:
            from app.database.crud.transaction import create_transaction as create_trans

            await create_trans(
                db=db,
                user_id=user.id,
                type=transaction_type,
                amount_kopeks=amount_kopeks,
                description=description,
                payment_method=payment_method,
            )

        if commit:
            await db.commit()
            await db.refresh(user)

        user_id_display = user.telegram_id or user.email or f'#{user.id}'
        logger.info(
            '💰 Баланс пользователя изменен: → (изменение: +)',
            user_id_display=user_id_display,
            old_balance=old_balance,
            balance_kopeks=user.balance_kopeks,
            amount_kopeks=amount_kopeks,
        )

        # Авто-возобновление суточной подписки НЕ делаем здесь —
        # это обязанность try_resume_disabled_daily_after_topup (через send_cart_notification_after_topup)
        # и DailySubscriptionService.process_auto_resume (30-минутный цикл).
        # Они корректно списывают суточную плату при возобновлении.

        return True

    except Exception as e:
        logger.error('Ошибка изменения баланса пользователя', user_id=user.id, error=e)
        if commit:
            await db.rollback()
        return False


async def add_user_balance_by_id(
    db: AsyncSession,
    telegram_id: int,
    amount_kopeks: int,
    description: str = 'Пополнение баланса',
    transaction_type: TransactionType = TransactionType.DEPOSIT,
    payment_method: PaymentMethod | None = None,
) -> bool:
    try:
        user = await get_user_by_telegram_id(db, telegram_id)
        if not user:
            logger.error('Пользователь с telegram_id не найден', telegram_id=telegram_id)
            return False

        return await add_user_balance(
            db,
            user,
            amount_kopeks,
            description,
            transaction_type=transaction_type,
            payment_method=payment_method,
        )

    except Exception as e:
        logger.error('Ошибка пополнения баланса пользователя', telegram_id=telegram_id, error=e)
        return False


async def lock_user_for_pricing(db: AsyncSession, user_id: int) -> User:
    """Lock user row with FOR UPDATE and return refreshed instance.

    Call BEFORE computing prices that depend on promo offer state
    to prevent TOCTOU race conditions where two concurrent requests
    both read the same promo offer discount and charge a discounted price.
    """
    result = await db.execute(
        select(User)
        .where(User.id == user_id)
        .options(
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.promo_group),
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
        )
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    return result.scalar_one()


async def subtract_user_balance(
    db: AsyncSession,
    user: User,
    amount_kopeks: int,
    description: str,
    create_transaction: bool = False,
    payment_method: PaymentMethod | None = None,
    *,
    transaction_type: TransactionType = TransactionType.WITHDRAWAL,
    consume_promo_offer: bool = False,
    mark_as_paid_subscription: bool = False,
    commit: bool = True,
) -> bool:
    if amount_kopeks < 0:
        logger.error('subtract_user_balance called with negative amount', amount_kopeks=amount_kopeks, user_id=user.id)
        return False

    logger.debug(
        'subtract_user_balance called',
        user_id=user.id,
        balance_kopeks=user.balance_kopeks,
        amount_kopeks=amount_kopeks,
        description=description,
    )

    # Lock the user row to prevent concurrent balance race conditions
    # Eagerly load key relationships to avoid MissingGreenlet in async context
    locked_result = await db.execute(
        select(User)
        .where(User.id == user.id)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.promo_group),
            selectinload(User.referrer),
        )
        .with_for_update()
        .execution_options(populate_existing=True)
    )
    user = locked_result.scalar_one()

    log_context: dict[str, object] | None = None
    if consume_promo_offer:
        try:
            current_percent = int(getattr(user, 'promo_offer_discount_percent', 0) or 0)
        except (TypeError, ValueError):
            current_percent = 0

        if current_percent > 0:
            source = getattr(user, 'promo_offer_discount_source', None)
            log_context = {
                'offer_id': None,
                'percent': current_percent,
                'source': source,
                'effect_type': None,
                'details': {
                    'reason': 'manual_charge',
                    'description': description,
                    'amount_kopeks': amount_kopeks,
                },
            }
            try:
                offer = await get_latest_claimed_offer_for_user(db, user.id, source)
            except Exception as lookup_error:  # pragma: no cover - defensive logging
                logger.warning(
                    'Failed to fetch latest claimed promo offer for user', user_id=user.id, lookup_error=lookup_error
                )
                offer = None

            if offer:
                log_context['offer_id'] = offer.id
                log_context['effect_type'] = offer.effect_type
                if not log_context['percent'] and offer.discount_percent:
                    log_context['percent'] = offer.discount_percent

    if user.balance_kopeks < amount_kopeks:
        logger.error('   ❌ НЕДОСТАТОЧНО СРЕДСТВ!')
        return False

    try:
        old_balance = user.balance_kopeks
        user.balance_kopeks -= amount_kopeks

        if consume_promo_offer and getattr(user, 'promo_offer_discount_percent', 0):
            user.promo_offer_discount_percent = 0
            user.promo_offer_discount_source = None
            user.promo_offer_discount_expires_at = None

        if mark_as_paid_subscription:
            user.has_had_paid_subscription = True

        user.updated_at = datetime.now(UTC)

        if create_transaction:
            from app.database.crud.transaction import (
                create_transaction as create_trans,
            )

            await create_trans(
                db=db,
                user_id=user.id,
                type=transaction_type,
                amount_kopeks=amount_kopeks,
                description=description,
                payment_method=payment_method,
                commit=commit,
            )
        elif commit:
            await db.commit()
        else:
            await db.flush()

        if commit:
            await db.refresh(user)

        if consume_promo_offer and log_context:
            # Пишем лог в ОТДЕЛЬНОЙ сессии, чтобы его commit/rollback не касался
            # основной сессии caller'а. Иначе rollback в случае фейла логирования
            # экспайрит объекты сессии и следующее обращение к subscription/user
            # attrs у caller'а падает с MissingGreenlet.
            if commit:
                try:
                    from app.database.database import AsyncSessionLocal

                    async with AsyncSessionLocal() as log_db:
                        await log_promo_offer_action(
                            log_db,
                            user_id=user.id,
                            offer_id=log_context.get('offer_id'),
                            action='consumed',
                            source=log_context.get('source'),
                            percent=log_context.get('percent'),
                            effect_type=log_context.get('effect_type'),
                            details=log_context.get('details'),
                            commit=True,
                        )
                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,
                    )
            else:
                # Caller управляет транзакцией — пишем в его сессию без commit.
                try:
                    await log_promo_offer_action(
                        db,
                        user_id=user.id,
                        offer_id=log_context.get('offer_id'),
                        action='consumed',
                        source=log_context.get('source'),
                        percent=log_context.get('percent'),
                        effect_type=log_context.get('effect_type'),
                        details=log_context.get('details'),
                        commit=False,
                    )
                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,
                    )

        logger.info('✅ Средства списаны: →', old_balance=old_balance, balance_kopeks=user.balance_kopeks)
        return True

    except Exception as e:
        logger.error('❌ ОШИБКА СПИСАНИЯ', error=e)
        if commit:
            await db.rollback()
            return False
        raise


async def cleanup_expired_promo_offer_discounts(db: AsyncSession) -> int:
    now = datetime.now(UTC)
    result = await db.execute(
        select(User).where(
            User.promo_offer_discount_percent > 0,
            User.promo_offer_discount_expires_at.isnot(None),
            User.promo_offer_discount_expires_at <= now,
        )
    )
    users = result.scalars().all()
    if not users:
        return 0

    log_payloads: list[dict[str, object]] = []

    for user in users:
        try:
            percent = int(getattr(user, 'promo_offer_discount_percent', 0) or 0)
        except (TypeError, ValueError):
            percent = 0

        source = getattr(user, 'promo_offer_discount_source', None)
        offer_id = None
        effect_type = None

        if source:
            try:
                offer = await get_latest_claimed_offer_for_user(db, user.id, source)
            except Exception as lookup_error:  # pragma: no cover - defensive logging
                logger.warning(
                    'Failed to fetch latest claimed promo offer for user during expiration cleanup',
                    user_id=user.id,
                    lookup_error=lookup_error,
                )
                offer = None

            if offer:
                offer_id = offer.id
                effect_type = offer.effect_type
                if not percent and offer.discount_percent:
                    percent = offer.discount_percent

        log_payloads.append(
            {
                'user_id': user.id,
                'offer_id': offer_id,
                'source': source,
                'percent': percent,
                'effect_type': effect_type,
            }
        )

        user.promo_offer_discount_percent = 0
        user.promo_offer_discount_source = None
        user.promo_offer_discount_expires_at = None
        user.updated_at = now

    await db.commit()

    for payload in log_payloads:
        user_id = payload.get('user_id')
        if not user_id:
            continue
        try:
            await log_promo_offer_action(
                db,
                user_id=user_id,
                offer_id=payload.get('offer_id'),
                action='disabled',
                source=payload.get('source'),
                percent=payload.get('percent'),
                effect_type=payload.get('effect_type'),
                details={'reason': 'offer_expired'},
            )
        except Exception as log_error:  # pragma: no cover - defensive logging
            logger.warning('Failed to log promo offer expiration 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 expiration log failure', rollback_error=rollback_error
                )

    return len(users)


async def get_users_list(
    db: AsyncSession,
    offset: int = 0,
    limit: int = 50,
    search: str | None = None,
    email: str | None = None,
    status: UserStatus | None = None,
    subscription_status: str | None = None,
    tariff_ids: list[int] | None = None,
    promo_group_id: int | None = None,
    campaign_id: int | None = None,
    partner_id: int | None = None,
    order_by_balance: bool = False,
    order_by_traffic: bool = False,
    order_by_last_activity: bool = False,
    order_by_total_spent: bool = False,
    order_by_purchase_count: bool = False,
) -> list[User]:
    query = select(User).options(
        selectinload(User.subscriptions).selectinload(Subscription.tariff),
        selectinload(User.promo_group),
        selectinload(User.referrer),
    )

    if status:
        query = query.where(User.status == status.value)

    # Subscription-level filters via subquery
    if subscription_status or tariff_ids:
        sub_conditions = []
        if subscription_status:
            sub_conditions.append(Subscription.status == subscription_status)
        if tariff_ids:
            sub_conditions.append(Subscription.tariff_id.in_(tariff_ids))
        sub_query = select(Subscription.user_id).where(and_(*sub_conditions)).distinct().scalar_subquery()
        query = query.where(User.id.in_(sub_query))

    if promo_group_id:
        query = query.where(User.promo_group_id == promo_group_id)

    if campaign_id:
        query = query.where(
            exists(
                select(AdvertisingCampaignRegistration.id).where(
                    AdvertisingCampaignRegistration.user_id == User.id,
                    AdvertisingCampaignRegistration.campaign_id == campaign_id,
                )
            )
        )

    if partner_id:
        query = query.where(
            exists(
                select(AdvertisingCampaignRegistration.id)
                .join(AdvertisingCampaign, AdvertisingCampaign.id == AdvertisingCampaignRegistration.campaign_id)
                .where(
                    AdvertisingCampaignRegistration.user_id == User.id,
                    AdvertisingCampaign.partner_user_id == partner_id,
                )
            )
        )

    if search:
        search_term = f'%{search}%'
        conditions = [
            User.first_name.ilike(search_term),
            User.last_name.ilike(search_term),
            User.username.ilike(search_term),
        ]

        if search.isdigit():
            try:
                search_int = int(search)
                # Добавляем условие поиска по telegram_id, который является BigInteger
                # и может содержать большие значения, в отличие от User.id (INTEGER)
                conditions.append(User.telegram_id == search_int)
            except ValueError:
                # Если не удалось преобразовать в int, просто ищем по текстовым полям
                pass

        query = query.where(or_(*conditions))

    if email:
        query = query.where(User.email.ilike(f'%{email}%'))

    sort_flags = [
        order_by_balance,
        order_by_traffic,
        order_by_last_activity,
        order_by_total_spent,
        order_by_purchase_count,
    ]
    if sum(int(flag) for flag in sort_flags) > 1:
        logger.debug(
            'Выбрано несколько сортировок пользователей — применяется приоритет: трафик > траты > покупки > баланс > активность'
        )

    transactions_stats = None
    if order_by_total_spent or order_by_purchase_count:
        from app.database.models import Transaction

        transactions_stats = (
            select(*_build_spending_stats_select())
            .where(Transaction.is_completed.is_(True))
            .group_by(Transaction.user_id)
            .subquery()
        )
        query = query.outerjoin(transactions_stats, transactions_stats.c.user_id == User.id)

    if order_by_traffic:
        traffic_sort = func.coalesce(Subscription.traffic_used_gb, 0.0)
        query = query.outerjoin(Subscription, Subscription.user_id == User.id)
        query = query.order_by(traffic_sort.desc(), User.created_at.desc())
    elif order_by_total_spent:
        order_column = func.coalesce(transactions_stats.c.total_spent, 0)
        query = query.order_by(order_column.desc(), User.created_at.desc())
    elif order_by_purchase_count:
        order_column = func.coalesce(transactions_stats.c.purchase_count, 0)
        query = query.order_by(order_column.desc(), User.created_at.desc())
    elif order_by_balance:
        query = query.order_by(User.balance_kopeks.desc(), User.created_at.desc())
    elif order_by_last_activity:
        query = query.order_by(nullslast(User.last_activity.desc()), User.created_at.desc())
    else:
        query = query.order_by(User.created_at.desc())

    query = query.offset(offset).limit(limit)

    result = await db.execute(query)
    users = result.scalars().unique().all()

    # Загружаем дополнительные зависимости для всех пользователей
    for user in users:
        if user and user.subscription:
            # Загружаем дополнительные зависимости для subscription
            _ = user.subscription.is_active

    return users


async def get_users_count(
    db: AsyncSession,
    status: UserStatus | None = None,
    search: str | None = None,
    email: str | None = None,
    subscription_status: str | None = None,
    tariff_ids: list[int] | None = None,
    promo_group_id: int | None = None,
    campaign_id: int | None = None,
    partner_id: int | None = None,
) -> int:
    query = select(func.count(User.id))

    if status:
        query = query.where(User.status == status.value)

    if subscription_status or tariff_ids:
        sub_conditions = []
        if subscription_status:
            sub_conditions.append(Subscription.status == subscription_status)
        if tariff_ids:
            sub_conditions.append(Subscription.tariff_id.in_(tariff_ids))
        sub_query = select(Subscription.user_id).where(and_(*sub_conditions)).distinct().scalar_subquery()
        query = query.where(User.id.in_(sub_query))

    if promo_group_id:
        query = query.where(User.promo_group_id == promo_group_id)

    if campaign_id:
        query = query.where(
            exists(
                select(AdvertisingCampaignRegistration.id).where(
                    AdvertisingCampaignRegistration.user_id == User.id,
                    AdvertisingCampaignRegistration.campaign_id == campaign_id,
                )
            )
        )

    if partner_id:
        query = query.where(
            exists(
                select(AdvertisingCampaignRegistration.id)
                .join(AdvertisingCampaign, AdvertisingCampaign.id == AdvertisingCampaignRegistration.campaign_id)
                .where(
                    AdvertisingCampaignRegistration.user_id == User.id,
                    AdvertisingCampaign.partner_user_id == partner_id,
                )
            )
        )

    if search:
        search_term = f'%{search}%'
        conditions = [
            User.first_name.ilike(search_term),
            User.last_name.ilike(search_term),
            User.username.ilike(search_term),
        ]

        if search.isdigit():
            try:
                search_int = int(search)
                # Добавляем условие поиска по telegram_id, который является BigInteger
                # и может содержать большие значения, в отличие от User.id (INTEGER)
                conditions.append(User.telegram_id == search_int)
            except ValueError:
                # Если не удалось преобразовать в int, просто ищем по текстовым полям
                pass

        query = query.where(or_(*conditions))

    if email:
        query = query.where(User.email.ilike(f'%{email}%'))

    result = await db.execute(query)
    return result.scalar()


async def get_users_spending_stats(db: AsyncSession, user_ids: list[int]) -> dict[int, dict[str, int]]:
    """
    Получает статистику трат для списка пользователей.

    Args:
        db: Сессия базы данных
        user_ids: Список ID пользователей

    Returns:
        Словарь {user_id: {"total_spent": int, "purchase_count": int}}
    """
    if not user_ids:
        return {}

    stats_query = (
        select(*_build_spending_stats_select())
        .where(
            Transaction.user_id.in_(user_ids),
            Transaction.is_completed.is_(True),
        )
        .group_by(Transaction.user_id)
    )

    result = await db.execute(stats_query)
    rows = result.all()

    return {
        row.user_id: {
            'total_spent': int(row.total_spent or 0),
            'purchase_count': int(row.purchase_count or 0),
        }
        for row in rows
    }


async def get_referrals(db: AsyncSession, user_id: int) -> list[User]:
    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.referrer),
            selectinload(User.promo_group),
        )
        .where(User.referred_by_id == user_id)
        .order_by(User.created_at.desc())
    )
    users = result.scalars().all()

    # Загружаем дополнительные зависимости для всех пользователей
    for user in users:
        if user and user.subscription:
            # Загружаем дополнительные зависимости для subscription
            _ = user.subscription.is_active

    return users


async def get_users_for_promo_segment(db: AsyncSession, segment: str) -> list[User]:
    now = datetime.now(UTC)

    base_query = (
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.promo_group),
            selectinload(User.referrer),
        )
        .where(User.status == UserStatus.ACTIVE.value)
    )

    if segment == 'no_subscription':
        query = base_query.outerjoin(Subscription, Subscription.user_id == User.id).where(Subscription.id.is_(None))
    else:
        query = base_query.join(Subscription)

        if segment == 'paid_active':
            query = query.where(
                Subscription.is_trial == False,
                Subscription.status == SubscriptionStatus.ACTIVE.value,
                Subscription.end_date > now,
            )
        elif segment == 'paid_expired':
            query = query.where(
                Subscription.is_trial == False,
                or_(
                    Subscription.status == SubscriptionStatus.EXPIRED.value,
                    Subscription.end_date <= now,
                ),
            )
        elif segment == 'trial_active':
            query = query.where(
                Subscription.is_trial == True,
                Subscription.status == SubscriptionStatus.ACTIVE.value,
                Subscription.end_date > now,
            )
        elif segment == 'trial_expired':
            query = query.where(
                Subscription.is_trial == True,
                or_(
                    Subscription.status == SubscriptionStatus.EXPIRED.value,
                    Subscription.end_date <= now,
                ),
            )
        else:
            logger.warning('Неизвестный сегмент для промо', segment=segment)
            return []

    result = await db.execute(query.order_by(User.id))
    users = result.scalars().unique().all()

    # Загружаем дополнительные зависимости для всех пользователей
    for user in users:
        if user and user.subscription:
            # Загружаем дополнительные зависимости для subscription
            _ = user.subscription.is_active

    return users


async def get_inactive_users(db: AsyncSession, months: int = 3) -> list[User]:
    threshold_date = datetime.now(UTC) - timedelta(days=months * 30)

    # Подзапрос: пользователи, у которых есть подписка с end_date >= threshold
    # (активная или недавно истёкшая) — таких удалять нельзя
    users_with_recent_subs = (
        select(Subscription.user_id).where(Subscription.end_date >= threshold_date).distinct().scalar_subquery()
    )

    result = await db.execute(
        select(User)
        .options(
            selectinload(User.subscriptions).selectinload(Subscription.tariff),
            selectinload(User.user_promo_groups).selectinload(UserPromoGroup.promo_group),
            selectinload(User.referrer),
            selectinload(User.promo_group),
        )
        .where(
            and_(
                User.last_activity < threshold_date,
                User.status == UserStatus.ACTIVE.value,
                User.id.not_in(users_with_recent_subs),
            )
        )
    )
    users = result.scalars().all()

    # Загружаем дополнительные зависимости для всех пользователей
    for user in users:
        if user and user.subscription:
            # Загружаем дополнительные зависимости для subscription
            _ = user.subscription.is_active

    return users


async def delete_user(db: AsyncSession, user: User) -> bool:
    user.status = UserStatus.DELETED.value
    user.updated_at = datetime.now(UTC)

    await db.commit()
    user_id_display = user.telegram_id or user.email or f'#{user.id}'
    logger.info('🗑️ Пользователь помечен как удаленный', user_id_display=user_id_display)
    return True


async def get_users_statistics(db: AsyncSession) -> dict:
    total_result = await db.execute(select(func.count(User.id)))
    total_users = total_result.scalar()

    active_result = await db.execute(select(func.count(User.id)).where(User.status == UserStatus.ACTIVE.value))
    active_users = active_result.scalar()

    today = datetime.now(UTC).date()
    today_result = await db.execute(
        select(func.count(User.id)).where(and_(User.created_at >= today, User.status == UserStatus.ACTIVE.value))
    )
    new_today = today_result.scalar()

    week_ago = datetime.now(UTC) - timedelta(days=7)
    week_result = await db.execute(
        select(func.count(User.id)).where(and_(User.created_at >= week_ago, User.status == UserStatus.ACTIVE.value))
    )
    new_week = week_result.scalar()

    month_ago = datetime.now(UTC) - timedelta(days=30)
    month_result = await db.execute(
        select(func.count(User.id)).where(and_(User.created_at >= month_ago, User.status == UserStatus.ACTIVE.value))
    )
    new_month = month_result.scalar()

    return {
        'total_users': total_users,
        'active_users': active_users,
        'blocked_users': total_users - active_users,
        'new_today': new_today,
        'new_week': new_week,
        'new_month': new_month,
    }


async def get_users_with_active_subscriptions(db: AsyncSession) -> list[User]:
    """
    Получает список пользователей с активными подписками.
    Используется для мониторинга трафика.

    Returns:
        Список пользователей с активными подписками и remnawave_uuid
    """
    current_time = datetime.now(UTC)

    result = await db.execute(
        select(User)
        .join(Subscription, User.id == Subscription.user_id)
        .where(
            and_(
                User.remnawave_uuid.isnot(None),
                User.status == UserStatus.ACTIVE.value,
                Subscription.status == SubscriptionStatus.ACTIVE.value,
                Subscription.end_date > current_time,
            )
        )
        .options(selectinload(User.subscriptions).selectinload(Subscription.tariff))
    )

    return result.scalars().unique().all()


async def create_user_by_email(
    db: AsyncSession,
    email: str,
    password_hash: str,
    first_name: str | None = None,
    language: str = 'ru',
    referred_by_id: int | None = None,
) -> User:
    """
    Создать пользователя через email регистрацию (без Telegram).

    Args:
        db: Database session
        email: Email address (will be unverified initially)
        password_hash: Hashed password
        first_name: Optional first name
        language: User language
        referred_by_id: Referrer user ID

    Returns:
        Created User object
    """
    referral_code = await create_unique_referral_code(db)
    normalized_language = _normalize_language_code(language)
    default_group = await _get_or_create_default_promo_group(db)

    user = User(
        telegram_id=None,  # Email-only user
        auth_type='email',
        email=email,
        email_verified=False,
        password_hash=password_hash,
        username=None,
        first_name=sanitize_telegram_name(first_name) if first_name else None,
        last_name=None,
        language=normalized_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=default_group.id,
    )

    db.add(user)
    await db.commit()
    await db.refresh(user)

    user.promo_group = default_group
    logger.info('✅ Создан email-пользователь с id', email=email, user_id=user.id)

    # Emit event
    try:
        from app.services.event_emitter import event_emitter

        await event_emitter.emit(
            'user.created',
            {
                'user_id': user.id,
                'email': user.email,
                'auth_type': 'email',
                'first_name': user.first_name,
                'referral_code': user.referral_code,
                'referred_by_id': user.referred_by_id,
            },
            db=db,
        )
    except Exception as error:
        logger.warning('Failed to emit user.created event', error=error)

    return user


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    """Get user by email address (case-insensitive)."""
    if not email or not email.strip():
        return None
    email_lower = email.strip().lower()
    result = await db.execute(select(User).where(func.lower(User.email) == email_lower))
    return result.scalar_one_or_none()


async def is_email_taken(db: AsyncSession, email: str, exclude_user_id: int | None = None) -> bool:
    """
    Check if email is already taken by another user.

    Args:
        db: Database session
        email: Email to check
        exclude_user_id: User ID to exclude from check (for current user)

    Returns:
        True if email is taken, False otherwise
    """
    if not email or not email.strip():
        return False
    email_lower = email.strip().lower()
    query = select(User.id).where(func.lower(User.email) == email_lower)
    if exclude_user_id:
        query = query.where(User.id != exclude_user_id)
    result = await db.execute(query)
    return result.scalar_one_or_none() is not None


async def set_email_change_pending(
    db: AsyncSession,
    user: User,
    new_email: str,
    code: str,
    expires_at: datetime,
) -> User:
    """
    Set pending email change for user.

    Args:
        db: Database session
        user: User object
        new_email: New email address
        code: 6-digit verification code
        expires_at: Code expiration datetime

    Returns:
        Updated User object
    """
    user.email_change_new = new_email
    user.email_change_code = code
    user.email_change_expires = expires_at
    user.updated_at = datetime.now(UTC)

    await db.commit()
    await db.refresh(user)

    logger.info('Email change pending for user', user_id=user.id, email=user.email, new_email=new_email)
    return user


async def verify_and_apply_email_change(db: AsyncSession, user: User, code: str) -> tuple[bool, str]:
    """
    Verify email change code and apply the change.

    Args:
        db: Database session
        user: User object
        code: Verification code from user

    Returns:
        Tuple of (success: bool, message: str)
    """
    if not user.email_change_new or not user.email_change_code:
        return False, 'No pending email change'

    if user.email_change_expires and datetime.now(UTC) > user.email_change_expires:
        # Clear expired data
        user.email_change_new = None
        user.email_change_code = None
        user.email_change_expires = None
        await db.commit()
        return False, 'Verification code has expired'

    if user.email_change_code != code:
        return False, 'Invalid verification code'

    # Check if new email is still available
    existing = await get_user_by_email(db, user.email_change_new)
    if existing and existing.id != user.id:
        user.email_change_new = None
        user.email_change_code = None
        user.email_change_expires = None
        await db.commit()
        return False, 'This email is already taken'

    old_email = user.email
    new_email = user.email_change_new

    # Apply the change
    user.email = new_email
    user.email_verified = True
    user.email_verified_at = datetime.now(UTC)
    user.email_change_new = None
    user.email_change_code = None
    user.email_change_expires = None
    user.updated_at = datetime.now(UTC)

    await db.commit()
    await db.refresh(user)

    logger.info('Email changed for user', user_id=user.id, old_email=old_email, new_email=new_email)
    return True, 'Email changed successfully'


async def clear_email_change_pending(db: AsyncSession, user: User) -> None:
    """
    Clear pending email change data.

    Args:
        db: Database session
        user: User object
    """
    user.email_change_new = None
    user.email_change_code = None
    user.email_change_expires = None
    user.updated_at = datetime.now(UTC)

    await db.commit()
    logger.info('Email change cancelled for user', user_id=user.id)


# --- OAuth provider functions ---

# Single source of truth: provider name → User model column name.
# Imported by account_linking.py and account_merge_service.py.
OAUTH_PROVIDER_COLUMNS: dict[str, str] = {
    'google': 'google_id',
    'yandex': 'yandex_id',
    'discord': 'discord_id',
    'vk': 'vk_id',
}


async def get_user_by_oauth_provider(db: AsyncSession, provider: str, provider_id: str) -> User | None:
    """Find a user by OAuth provider ID."""
    column_name = OAUTH_PROVIDER_COLUMNS.get(provider)
    if not column_name:
        logger.warning('Unknown OAuth provider in lookup', provider=provider)
        return None
    column = getattr(User, column_name)
    # VK uses BigInteger, so convert
    value: str | int = int(provider_id) if provider == 'vk' else provider_id
    result = await db.execute(select(User).where(column == value))
    return result.scalar_one_or_none()


async def set_user_oauth_provider_id(db: AsyncSession, user: User, provider: str, provider_id: str) -> None:
    """Link an OAuth provider ID to an existing user."""
    column_name = OAUTH_PROVIDER_COLUMNS.get(provider)
    if not column_name:
        logger.warning('Unknown OAuth provider in set', provider=provider, user_id=user.id)
        return
    value: str | int = int(provider_id) if provider == 'vk' else provider_id
    setattr(user, column_name, value)
    user.updated_at = datetime.now(UTC)
    logger.info('OAuth provider linked to user', provider=provider, provider_id=provider_id, user_id=user.id)


async def clear_user_oauth_provider_id(db: AsyncSession, user: User, provider: str) -> None:
    """Unlink an OAuth provider from an existing user (set column to None)."""
    column_name = OAUTH_PROVIDER_COLUMNS.get(provider)
    if not column_name:
        logger.warning('Unknown OAuth provider in clear', provider=provider, user_id=user.id)
        return
    setattr(user, column_name, None)
    user.updated_at = datetime.now(UTC)
    logger.info('Unlinked OAuth provider from user', provider=provider, user_id=user.id)


async def create_user_by_oauth(
    db: AsyncSession,
    provider: str,
    provider_id: str,
    email: str | None = None,
    email_verified: bool = False,
    first_name: str | None = None,
    last_name: str | None = None,
    username: str | None = None,
    language: str = 'ru',
    referred_by_id: int | None = None,
) -> User:
    """Create a new user via OAuth provider."""
    referral_code = await create_unique_referral_code(db)
    normalized_language = _normalize_language_code(language)
    default_group = await _get_or_create_default_promo_group(db)

    column_name = OAUTH_PROVIDER_COLUMNS.get(provider)
    provider_value: str | int = int(provider_id) if provider == 'vk' else provider_id

    user = User(
        telegram_id=None,
        auth_type=provider,
        email=email,
        email_verified=email_verified,
        password_hash=None,
        username=sanitize_telegram_name(username) if username else None,
        first_name=sanitize_telegram_name(first_name) if first_name else None,
        last_name=sanitize_telegram_name(last_name) if last_name else None,
        language=normalized_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=default_group.id,
    )
    if column_name:
        setattr(user, column_name, provider_value)

    db.add(user)
    await db.flush()
    await db.refresh(user)

    user.promo_group = default_group
    logger.info(
        'Created OAuth user via (provider_id=) with id', provider=provider, provider_id=provider_id, user_id=user.id
    )

    try:
        from app.services.event_emitter import event_emitter

        await event_emitter.emit(
            'user.created',
            {
                'user_id': user.id,
                'email': user.email,
                'auth_type': provider,
                'first_name': user.first_name,
                'referral_code': user.referral_code,
            },
            db=db,
        )
    except Exception as error:
        logger.warning('Failed to emit user.created event', error=error)

    return user


async def lock_user_subscriptions_for_update(db: AsyncSession, user_id: int) -> list[Subscription]:
    """Lock all subscriptions for a user using SELECT FOR UPDATE."""
    result = await db.execute(
        select(Subscription)
        .where(Subscription.user_id == user_id)
        .with_for_update()
        .order_by(Subscription.created_at.desc())
    )
    return list(result.scalars().all())