RemnaWaveBOT/app/services/payment_service.py

from __future__ import annotations

import html
import logging
import unicodedata
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from math import ceil
from uuid import uuid4

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.config import PaymentPlanSettings, Settings
from app.db.base import utcnow
from app.db.models import PaymentOrder, ReferralBonus, ReferralInvite, TelegramUser
from app.schemas.remnawave import RemnawaveUser
from app.services.bot_config_service import BotConfigSnapshot, StaticBotConfigService
from app.services.remnawave_client import RemnawaveApiClient, RemnawaveApiError
from app.services.sync_service import SyncService


PENDING_PAYMENT_STATUS = "PENDING"
REVIEW_PAYMENT_STATUS = "REVIEW"
PROCESSING_PAYMENT_STATUS = "PROCESSING"
REJECTED_PAYMENT_STATUS = "REJECTED"
FULFILLED_PAYMENT_STATUS = "FULFILLED"
REFERRAL_BONUS_PENDING_STATUS = "PENDING"
REFERRAL_BONUS_PROCESSING_STATUS = "PROCESSING"
REFERRAL_BONUS_APPLIED_STATUS = "APPLIED"

logger = logging.getLogger(__name__)

_CYRILLIC_TRANSLITERATION = str.maketrans(
    {
        "\u0410": "A",
        "\u0411": "B",
        "\u0412": "V",
        "\u0413": "G",
        "\u0414": "D",
        "\u0415": "E",
        "\u0401": "E",
        "\u0416": "Zh",
        "\u0417": "Z",
        "\u0418": "I",
        "\u0419": "Y",
        "\u041a": "K",
        "\u041b": "L",
        "\u041c": "M",
        "\u041d": "N",
        "\u041e": "O",
        "\u041f": "P",
        "\u0420": "R",
        "\u0421": "S",
        "\u0422": "T",
        "\u0423": "U",
        "\u0424": "F",
        "\u0425": "Kh",
        "\u0426": "Ts",
        "\u0427": "Ch",
        "\u0428": "Sh",
        "\u0429": "Sch",
        "\u042a": "",
        "\u042b": "Y",
        "\u042c": "",
        "\u042d": "E",
        "\u042e": "Yu",
        "\u042f": "Ya",
        "\u0430": "a",
        "\u0431": "b",
        "\u0432": "v",
        "\u0433": "g",
        "\u0434": "d",
        "\u0435": "e",
        "\u0451": "e",
        "\u0436": "zh",
        "\u0437": "z",
        "\u0438": "i",
        "\u0439": "y",
        "\u043a": "k",
        "\u043b": "l",
        "\u043c": "m",
        "\u043d": "n",
        "\u043e": "o",
        "\u043f": "p",
        "\u0440": "r",
        "\u0441": "s",
        "\u0442": "t",
        "\u0443": "u",
        "\u0444": "f",
        "\u0445": "kh",
        "\u0446": "ts",
        "\u0447": "ch",
        "\u0448": "sh",
        "\u0449": "sch",
        "\u044a": "",
        "\u044b": "y",
        "\u044c": "",
        "\u044d": "e",
        "\u044e": "yu",
        "\u044f": "ya",
    }
)


@dataclass(slots=True)
class AppliedDiscount:
    code: str = ""
    source: str = ""
    discount_percent: int = 0
    plan_code: str | None = None
    expires_at: datetime | None = None

    def applies_to_plan(self, plan_code: str, *, days: int | None = None) -> bool:
        if not self.code:
            return False
        if self.plan_code is None:
            return True
        # "all_30" matches any product with 30 days, including single-duration product codes.
        if self.plan_code.startswith("all_"):
            raw_days = self.plan_code.removeprefix("all_")
            return raw_days.isdigit() and (
                plan_code.endswith(f"_{raw_days}") or days == int(raw_days)
            )
        # Exact match or product-prefix match (e.g. promo plan_code="vpn_white" matches plan "vpn_white_30")
        return self.plan_code == plan_code or plan_code.startswith(f"{self.plan_code}_")


@dataclass(slots=True)
class PaymentPlan:
    code: str
    label: str
    title: str
    description: str
    duration_days: int
    amount_rub: int
    discount_percent: int
    original_amount_rub: int
    applied_referral_code: str = ""
    discount_code: str = ""
    discount_source: str = ""
    discount_scope_plan_code: str | None = None
    traffic_limit_bytes: int = 0
    traffic_limit_strategy: str = "NO_RESET"
    internal_squad_uuids: list[str] | None = None
    external_squad_uuid: str | None = None

    @property
    def is_ready(self) -> bool:
        return bool(self.internal_squad_uuids) and self.duration_days > 0 and self.amount_rub > 0


@dataclass(slots=True)
class CreatedPaymentOrder:
    order_uuid: str
    provision_username: str
    plan: PaymentPlan
    transfer_text: str
    extends_existing_access: bool = False


@dataclass(slots=True)
class StoredPaymentOrder:
    order_uuid: str
    telegram_id: int
    plan_title: str
    plan_duration_days: int
    amount_rub: int
    provision_username: str
    status: str


@dataclass(slots=True)
class IssuedAccess:
    telegram_id: int
    username: str
    subscription_url: str
    expire_at: datetime
    short_uuid: str
    remnawave_user_uuid: str
    traffic_limit_bytes: int
    referral_bonus: ReferralBonusGrant | None = None
    is_renewal: bool = False


@dataclass(slots=True)
class ReferralBonusGrant:
    inviter_telegram_id: int
    bonus_days: int
    status: str
    total_applied_days: int = 0
    expire_at: datetime | None = None
    applied_to_user_uuid: str | None = None


@dataclass(slots=True)
class AppliedReferralBonus:
    telegram_id: int
    total_bonus_days: int
    expire_at: datetime
    applied_to_user_uuid: str


@dataclass(slots=True)
class PaymentAdminStats:
    pending_orders: int = 0
    review_orders: int = 0
    pending_referral_bonuses: int = 0


class PaymentService:
    def __init__(
        self,
        *,
        settings: Settings,
        session_factory: async_sessionmaker[AsyncSession],
        remnawave_client: RemnawaveApiClient,
        sync_service: SyncService,
        config_service: StaticBotConfigService | None = None,
    ) -> None:
        self._settings = settings
        self._session_factory = session_factory
        self._remnawave_client = remnawave_client
        self._sync_service = sync_service
        self._config_service = config_service or StaticBotConfigService(settings)

    async def _get_config(self) -> BotConfigSnapshot:
        return await self._config_service.get_snapshot()

    async def get_available_plans(self, *, telegram_id: int) -> list[PaymentPlan]:
        config = await self._get_config()
        applied_discount = await self._get_applied_discount(
            telegram_id,
            config=config,
        )
        return [
            self._build_plan(
                plan,
                applied_discount=applied_discount,
                config=config,
            )
            for plan in config.payment_plans
        ]

    async def create_order(
        self,
        *,
        telegram_id: int,
        username: str | None,
        first_name: str | None,
        last_name: str | None,
        language_code: str | None,
        plan_code: str,
    ) -> CreatedPaymentOrder:
        config = await self._get_config()
        applied_discount = await self._get_applied_discount(
            telegram_id,
            config=config,
        )
        plan = self._get_plan_by_code(
            plan_code,
            applied_discount=applied_discount,
            config=config,
        )
        if not plan.is_ready:
            raise ValueError(
                "Покупка не настроена. Проверьте PAYMENT_PLANS и PAYMENT_INTERNAL_SQUAD_UUIDS."
            )

        existing_accesses = await self._sync_service.get_cached_users_for_telegram(telegram_id)
        renewal_target_record = self._select_cached_renewal_target(existing_accesses)
        renewal_target_uuid: str | None = None
        renewal_target_username: str | None = None
        if renewal_target_record is not None:
            renewal_target_uuid = str(renewal_target_record.rw_uuid)
            renewal_target_username = renewal_target_record.username
        else:
            try:
                remote_users = await self._remnawave_client.get_users_by_telegram_id(
                    telegram_id
                )
            except RemnawaveApiError:
                remote_users = []
            remote_user = self._select_referral_bonus_target(remote_users)
            if remote_user is not None:
                renewal_target_uuid = str(remote_user.uuid)
                renewal_target_username = remote_user.username

        order_uuid = str(uuid4())
        invoice_payload = f"manual:{plan.code}:{order_uuid}"
        provision_username = (
            renewal_target_username
            or self._build_provision_username(
                config=config,
                order_uuid=order_uuid,
                telegram_id=telegram_id,
                username=username,
                first_name=first_name,
            )
        )

        async with self._session_factory() as session:
            user = await self._upsert_telegram_user(
                session,
                telegram_id=telegram_id,
                username=username,
                first_name=first_name,
                last_name=last_name,
                language_code=language_code,
            )
            session.add(
                PaymentOrder(
                    order_uuid=order_uuid,
                    telegram_user_id=user.id,
                    telegram_id=telegram_id,
                    plan_code=plan.code,
                    plan_title=plan.title,
                    plan_duration_days=plan.duration_days,
                    traffic_limit_bytes=plan.traffic_limit_bytes,
                    traffic_limit_strategy=plan.traffic_limit_strategy,
                    amount_stars=plan.amount_rub,
                    currency="RUB",
                    status=PENDING_PAYMENT_STATUS,
                    invoice_payload=invoice_payload,
                    provision_username=provision_username,
                    remnawave_user_uuid=renewal_target_uuid,
                    error_message=plan.discount_code or None,
                )
            )
            await session.commit()

        return CreatedPaymentOrder(
            order_uuid=order_uuid,
            provision_username=provision_username,
            plan=plan,
            transfer_text=config.payment_transfer_text.strip(),
            extends_existing_access=bool(existing_accesses) or renewal_target_uuid is not None,
        )

    async def get_order_for_user(self, *, order_uuid: str, telegram_id: int) -> StoredPaymentOrder:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder).where(
                    PaymentOrder.order_uuid == order_uuid,
                    PaymentOrder.telegram_id == telegram_id,
                )
            )
            if order is None:
                raise ValueError("Заказ не найден.")

            return StoredPaymentOrder(
                order_uuid=order.order_uuid,
                telegram_id=order.telegram_id,
                plan_title=order.plan_title,
                plan_duration_days=order.plan_duration_days,
                amount_rub=order.amount_stars,
                provision_username=order.provision_username,
                status=order.status,
            )

    async def mark_order_under_review(self, *, order_uuid: str, telegram_id: int) -> StoredPaymentOrder:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder)
                .where(
                    PaymentOrder.order_uuid == order_uuid,
                    PaymentOrder.telegram_id == telegram_id,
                )
                .with_for_update()
            )
            if order is None:
                raise ValueError("Заказ не найден.")
            if order.status == REVIEW_PAYMENT_STATUS:
                raise ValueError("Чек уже отправлен на проверку.")
            if order.status == PROCESSING_PAYMENT_STATUS:
                raise ValueError("Заказ уже обрабатывается модератором.")
            if order.status == FULFILLED_PAYMENT_STATUS:
                raise ValueError("Этот заказ уже подтверждён.")
            if order.status == REJECTED_PAYMENT_STATUS:
                raise ValueError("Этот заказ уже отклонён. Создайте новый платёж.")
            if order.status != PENDING_PAYMENT_STATUS:
                raise ValueError("Некорректный статус заказа для отправки чека.")

            order.status = REVIEW_PAYMENT_STATUS
            order.updated_at = utcnow()
            await session.commit()

            return StoredPaymentOrder(
                order_uuid=order.order_uuid,
                telegram_id=order.telegram_id,
                plan_title=order.plan_title,
                plan_duration_days=order.plan_duration_days,
                amount_rub=order.amount_stars,
                provision_username=order.provision_username,
                status=order.status,
            )

    async def revert_order_to_pending(self, *, order_uuid: str, telegram_id: int) -> None:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder)
                .where(
                    PaymentOrder.order_uuid == order_uuid,
                    PaymentOrder.telegram_id == telegram_id,
                )
                .with_for_update()
            )
            if order is None or order.status != REVIEW_PAYMENT_STATUS:
                return

            order.status = PENDING_PAYMENT_STATUS
            order.updated_at = utcnow()
            await session.commit()

    async def approve_order(self, *, order_uuid: str) -> IssuedAccess:
        order = await self._claim_order_for_approval(order_uuid)
        is_renewal = False

        try:
            config = await self._get_config()
            plan = self._get_plan_by_code(
                order.plan_code,
                applied_discount=AppliedDiscount(),
                config=config,
            )
            remote_user: RemnawaveUser | None = None
            if order.remnawave_user_uuid:
                try:
                    remote_user = await self._remnawave_client.get_user_by_uuid(
                        order.remnawave_user_uuid
                    )
                except RemnawaveApiError as exc:
                    if exc.status_code != 404:
                        raise
                    remote_user = None

            if remote_user is None:
                try:
                    remote_user = await self._remnawave_client.get_user_by_username(
                        order.provision_username
                    )
                except RemnawaveApiError as exc:
                    if exc.status_code != 404:
                        raise

            if remote_user is None:
                try:
                    remote_users = await self._remnawave_client.get_users_by_telegram_id(
                        order.telegram_id
                    )
                    remote_user = self._select_referral_bonus_target(remote_users)
                except Exception:
                    logger.warning(
                        "Failed to look up user by telegram_id=%s, will try to create",
                        order.telegram_id,
                        exc_info=True,
                    )

            if remote_user is None:
                remote_user = await self._create_remnawave_user(
                    order,
                    plan=plan,
                    config=config,
                )
            else:
                is_renewal = True
                remote_user = await self._extend_remnawave_user(
                    order,
                    remote_user,
                    plan=plan,
                    config=config,
                )
        except ValueError as exc:
            await self._restore_review_after_failed_approval(order.order_uuid, str(exc))
            raise
        except RemnawaveApiError as exc:
            await self._restore_review_after_failed_approval(order.order_uuid, exc.message)
            raise
        except Exception as exc:
            error_msg = f"Unexpected error: {exc.__class__.__name__}: {exc}"
            logger.error(
                "Unexpected error during order approval %s: %s",
                order.order_uuid,
                error_msg,
                exc_info=True,
            )
            await self._restore_review_after_failed_approval(order.order_uuid, error_msg)
            raise RemnawaveApiError(
                status_code=0,
                message=f"Ошибка подключения к панели. Попробуйте ещё раз. ({exc.__class__.__name__})",
            )

        await self._save_fulfilled_order(order.order_uuid, remote_user)

        # Burn the applied discount (promo or referral) so it cannot be reused
        try:
            await self._sync_service.consume_discount_for_user(order.telegram_id)
        except Exception:
            logger.warning(
                "Could not consume discount for telegram_id=%s after order=%s",
                order.telegram_id,
                order.order_uuid,
                exc_info=True,
            )

        own_bonus = None
        try:
            own_bonus = await self.apply_pending_referral_bonuses(
                telegram_id=order.telegram_id,
                refresh_cache=False,
            )
        except RemnawaveApiError as exc:
            logger.warning(
                "Could not apply pending referral bonus for telegram_id=%s: %s",
                order.telegram_id,
                exc.message,
            )

        try:
            await self._sync_service.refresh_cached_users_for_telegram(
                telegram_id=order.telegram_id,
            )
        except RemnawaveApiError:
            # Access is already issued at this point; cache can be refreshed later.
            pass

        expire_at = remote_user.expire_at
        if own_bonus is not None and own_bonus.applied_to_user_uuid == str(remote_user.uuid):
            expire_at = own_bonus.expire_at

        referral_bonus = await self._grant_referral_bonus(order, config=config)

        return IssuedAccess(
            telegram_id=order.telegram_id,
            username=remote_user.username,
            subscription_url=remote_user.subscription_url,
            expire_at=expire_at,
            short_uuid=remote_user.short_uuid,
            remnawave_user_uuid=str(remote_user.uuid),
            traffic_limit_bytes=remote_user.traffic_limit_bytes,
            referral_bonus=referral_bonus,
            is_renewal=is_renewal,
        )

    async def reject_order(self, *, order_uuid: str, reason: str | None = None) -> StoredPaymentOrder:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder)
                .where(PaymentOrder.order_uuid == order_uuid)
                .with_for_update()
            )
            if order is None:
                raise ValueError("Заказ не найден.")
            if order.status == FULFILLED_PAYMENT_STATUS:
                raise ValueError("Подтверждённый заказ нельзя отклонить.")
            if order.status == PROCESSING_PAYMENT_STATUS:
                raise ValueError("Заказ уже обрабатывается другим модератором.")
            if order.status == PENDING_PAYMENT_STATUS:
                raise ValueError("Чек ещё не отправлен на проверку.")
            if order.status == REJECTED_PAYMENT_STATUS:
                raise ValueError("Заказ уже отклонён.")

            order.status = REJECTED_PAYMENT_STATUS
            order.error_message = reason or "Платёж отклонён"
            order.updated_at = utcnow()
            await session.commit()

            return StoredPaymentOrder(
                order_uuid=order.order_uuid,
                telegram_id=order.telegram_id,
                plan_title=order.plan_title,
                plan_duration_days=order.plan_duration_days,
                amount_rub=order.amount_stars,
                provision_username=order.provision_username,
                status=order.status,
            )

    async def get_referral_code_for_order(self, *, order_uuid: str) -> str:
        order = await self._get_order(order_uuid)
        return order.error_message or ""

    def build_transfer_text(self, *, order: CreatedPaymentOrder) -> str:
        lines = [
            "<b>Оплата тарифа</b>",
            "",
            f"<b>{html.escape(order.plan.title)}</b>",
            html.escape(order.plan.description),
            "",
            f"Сумма к переводу: <b>{order.plan.amount_rub} ₽</b>",
        ]
        if order.plan.discount_code:
            discount_label = (
                "Промокод"
                if order.plan.discount_source == "promo"
                else "Реферальный код"
            )
            lines.append(
                f"{discount_label}: <code>{order.plan.discount_code}</code> "
                f"(<b>-{order.plan.discount_percent}%</b>)"
            )
        if order.extends_existing_access:
            lines.extend(
                [
                    "",
                    (
                        "<blockquote>"
                        "У вас уже есть доступ. После подтверждения оплаты бот автоматически "
                        "продлит текущую подписку на выбранный срок."
                        "</blockquote>"
                    ),
                ]
            )
        lines.extend(
            [
                "",
                "<b>Реквизиты</b>",
                (
                    "<blockquote>"
                    f"{html.escape(order.transfer_text or 'Укажите реквизиты в PAYMENT_TRANSFER_TEXT.')}"
                    "</blockquote>"
                ),
                "",
                "После перевода нажмите кнопку ниже и отправьте чек следующим сообщением.",
            ]
        )
        return "\n".join(lines)

    def _get_plan_by_code(
        self,
        plan_code: str,
        *,
        applied_discount: AppliedDiscount,
        config: BotConfigSnapshot,
    ) -> PaymentPlan:
        for plan in config.payment_plans:
            if plan.code == plan_code:
                return self._build_plan(
                    plan,
                    applied_discount=applied_discount,
                    config=config,
                )

        raise ValueError("Тариф не найден.")

    def _build_plan(
        self,
        plan_settings: PaymentPlanSettings,
        *,
        applied_discount: AppliedDiscount,
        config: BotConfigSnapshot,
    ) -> PaymentPlan:
        resolved_discount = (
            applied_discount
            if applied_discount.applies_to_plan(
                plan_settings.code,
                days=plan_settings.days,
            )
            else AppliedDiscount()
        )
        discount_percent = max(resolved_discount.discount_percent, 0)
        discounted_amount = self._apply_discount(
            amount_rub=plan_settings.amount_rub,
            discount_percent=discount_percent,
        )
        traffic_limit_bytes = max(config.payment_plan_traffic_limit_gb, 0) * 1024**3
        return PaymentPlan(
            code=plan_settings.code,
            label=plan_settings.title or f"{plan_settings.days} дней",
            title=(
                f"{config.bot_brand_name} · {plan_settings.title}"
                if plan_settings.title
                else f"{config.bot_brand_name} на {plan_settings.days} дней"
            ),
            description=plan_settings.description or f"Доступ к VPN на {plan_settings.days} дней",
            duration_days=plan_settings.days,
            amount_rub=discounted_amount,
            discount_percent=discount_percent,
            original_amount_rub=plan_settings.amount_rub,
            applied_referral_code=(
                resolved_discount.code if resolved_discount.source == "referral" else ""
            ),
            discount_code=resolved_discount.code,
            discount_source=resolved_discount.source,
            discount_scope_plan_code=resolved_discount.plan_code,
            traffic_limit_bytes=traffic_limit_bytes,
            traffic_limit_strategy=config.payment_plan_traffic_reset_period.strip() or "NO_RESET",
            internal_squad_uuids=self._resolve_plan_internal_squad_uuids(
                plan_settings,
                config=config,
            ),
            external_squad_uuid=config.payment_external_squad_uuid_normalized,
        )

    @staticmethod
    def _resolve_plan_internal_squad_uuids(
        plan_settings: PaymentPlanSettings,
        *,
        config: BotConfigSnapshot,
    ) -> list[str]:
        if not plan_settings.squad_groups:
            return config.payment_internal_squad_uuids

        values: list[str] = []
        for group in plan_settings.squad_groups:
            if group == "vpn":
                group_values = config.payment_vpn_squad_uuids
            elif group == "white":
                group_values = config.payment_white_squad_uuids
            else:
                group_values = []

            for squad_uuid in group_values:
                if squad_uuid not in values:
                    values.append(squad_uuid)

        return values

    @staticmethod
    def _apply_discount(*, amount_rub: int, discount_percent: int) -> int:
        if discount_percent <= 0:
            return amount_rub
        return max(1, ceil(amount_rub * (100 - discount_percent) / 100))

    async def get_active_discount_for_user(self, *, telegram_id: int) -> AppliedDiscount:
        config = await self._get_config()
        return await self._get_applied_discount(telegram_id, config=config)

    async def _get_applied_discount(
        self,
        telegram_id: int,
        *,
        config: BotConfigSnapshot | None = None,
    ) -> AppliedDiscount:
        resolved_config = config or await self._get_config()
        promo_getter = getattr(self._sync_service, "get_active_promo_for_user", None)
        if callable(promo_getter):
            promo = await promo_getter(telegram_id)
            if promo is not None:
                return AppliedDiscount(
                    code=str(promo.code).strip().upper(),
                    source="promo",
                    discount_percent=max(int(promo.discount_percent), 0),
                    plan_code=promo.plan_code,
                    expires_at=promo.expires_at,
                )

        if not resolved_config.referral_enabled:
            return AppliedDiscount()

        summary = await self._sync_service.get_referral_summary(telegram_id)
        referral_code = summary.applied_referral_code.strip().upper()
        if not referral_code:
            return AppliedDiscount()

        # Check if referral discount was already consumed by a previous payment
        discount_used_checker = getattr(
            self._sync_service, "is_referral_discount_used", None
        )
        if callable(discount_used_checker) and await discount_used_checker(telegram_id):
            return AppliedDiscount()

        return AppliedDiscount(
            code=referral_code,
            source="referral",
            discount_percent=max(resolved_config.referral_discount_percent, 0),
        )

    async def apply_pending_referral_bonuses(
        self,
        *,
        telegram_id: int,
        refresh_cache: bool = True,
    ) -> AppliedReferralBonus | None:
        config = await self._get_config()
        if not config.referral_enabled:
            return None

        claimed_bonuses = await self._claim_pending_referral_bonuses(telegram_id)
        if not claimed_bonuses:
            return None

        total_bonus_days = sum(max(bonus.bonus_days, 0) for bonus in claimed_bonuses)
        if total_bonus_days <= 0:
            await self._restore_claimed_referral_bonuses(
                [bonus.id for bonus in claimed_bonuses],
                status=REFERRAL_BONUS_PENDING_STATUS,
            )
            return None

        remote_users = await self._remnawave_client.get_users_by_telegram_id(telegram_id)
        target_user = self._select_referral_bonus_target(remote_users)
        if target_user is None:
            await self._restore_claimed_referral_bonuses(
                [bonus.id for bonus in claimed_bonuses],
                status=REFERRAL_BONUS_PENDING_STATUS,
            )
            return None

        current_expire_at = target_user.expire_at
        if current_expire_at.tzinfo is None:
            current_expire_at = current_expire_at.replace(tzinfo=timezone.utc)
        next_expire_at = max(current_expire_at, datetime.now(timezone.utc)) + timedelta(
            days=total_bonus_days
        )

        try:
            updated_user = await self._remnawave_client.update_user(
                {
                    "uuid": str(target_user.uuid),
                    "expireAt": next_expire_at.isoformat().replace("+00:00", "Z"),
                    "description": (
                        f"Referral bonus applied by Telegram bot "
                        f"({total_bonus_days} days)"
                    ),
                }
            )
        except RemnawaveApiError:
            await self._restore_claimed_referral_bonuses(
                [bonus.id for bonus in claimed_bonuses],
                status=REFERRAL_BONUS_PENDING_STATUS,
            )
            raise

        await self._mark_referral_bonuses_applied(
            [bonus.id for bonus in claimed_bonuses],
            applied_to_user_uuid=str(updated_user.uuid),
        )

        if refresh_cache:
            try:
                await self._sync_service.refresh_cached_users_for_telegram(
                    telegram_id=telegram_id,
                )
            except RemnawaveApiError:
                pass

        return AppliedReferralBonus(
            telegram_id=telegram_id,
            total_bonus_days=total_bonus_days,
            expire_at=updated_user.expire_at,
            applied_to_user_uuid=str(updated_user.uuid),
        )

    async def get_admin_stats(self) -> PaymentAdminStats:
        async with self._session_factory() as session:
            pending_orders = await session.scalar(
                select(func.count(PaymentOrder.id)).where(
                    PaymentOrder.status == PENDING_PAYMENT_STATUS
                )
            )
            review_orders = await session.scalar(
                select(func.count(PaymentOrder.id)).where(
                    PaymentOrder.status == REVIEW_PAYMENT_STATUS
                )
            )
            pending_referral_bonuses = await session.scalar(
                select(func.count(ReferralBonus.id)).where(
                    ReferralBonus.status == REFERRAL_BONUS_PENDING_STATUS
                )
            )

        return PaymentAdminStats(
            pending_orders=int(pending_orders or 0),
            review_orders=int(review_orders or 0),
            pending_referral_bonuses=int(pending_referral_bonuses or 0),
        )

    async def _get_order(self, order_uuid: str) -> PaymentOrder:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder).where(PaymentOrder.order_uuid == order_uuid)
            )
            if order is None:
                raise ValueError("Заказ не найден.")
            return order

    async def _claim_order_for_approval(self, order_uuid: str) -> PaymentOrder:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder)
                .where(PaymentOrder.order_uuid == order_uuid)
                .with_for_update()
            )
            if order is None:
                raise ValueError("Заказ не найден.")
            if order.status == PENDING_PAYMENT_STATUS:
                raise ValueError("Чек ещё не отправлен на проверку.")
            if order.status == PROCESSING_PAYMENT_STATUS:
                # If the order has been stuck in PROCESSING for more than 60s,
                # it means the previous approval attempt crashed (e.g. ConnectError).
                # Allow re-claiming it instead of blocking forever.
                stale_threshold = utcnow() - timedelta(seconds=60)
                if order.updated_at and order.updated_at > stale_threshold:
                    raise ValueError("Заказ уже обрабатывается другим модератором.")
                logger.warning(
                    "Order %s was stuck in PROCESSING since %s, reclaiming",
                    order_uuid,
                    order.updated_at,
                )
                # Already in PROCESSING — just refresh the timestamp and proceed
                order.updated_at = utcnow()
                await session.commit()
                return order
            if order.status == REJECTED_PAYMENT_STATUS:
                raise ValueError("Заказ уже отклонён.")
            if order.status == FULFILLED_PAYMENT_STATUS:
                raise ValueError("Этот заказ уже подтверждён.")
            if order.status != REVIEW_PAYMENT_STATUS:
                raise ValueError("Некорректный статус заказа для подтверждения.")

            order.status = PROCESSING_PAYMENT_STATUS
            order.updated_at = utcnow()
            await session.commit()
            return order

    async def _create_remnawave_user(
        self,
        order: PaymentOrder,
        *,
        plan: PaymentPlan,
        config: BotConfigSnapshot,
    ) -> RemnawaveUser:
        expire_at = datetime.now(timezone.utc) + timedelta(days=order.plan_duration_days)
        body: dict[str, object] = {
            "username": order.provision_username,
            "expireAt": expire_at.isoformat().replace("+00:00", "Z"),
            "trafficLimitBytes": order.traffic_limit_bytes,
            "trafficLimitStrategy": order.traffic_limit_strategy,
            "telegramId": order.telegram_id,
            "description": f"Created by Telegram bot manual payment {order.order_uuid}",
            "activeInternalSquads": plan.internal_squad_uuids or [],
        }
        if config.payment_user_tag_normalized:
            body["tag"] = config.payment_user_tag_normalized
        if config.payment_external_squad_uuid_normalized:
            body["externalSquadUuid"] = config.payment_external_squad_uuid_normalized

        try:
            return await self._remnawave_client.create_user(body)
        except RemnawaveApiError as exc:
            if exc.status_code != 400:
                raise
            # Creation failed (likely duplicate telegramId) — fall back to
            # finding the existing user and extending their subscription.
            logger.warning(
                "POST /api/users returned 400 for order=%s telegram_id=%s: %s. "
                "Falling back to extend existing user.",
                order.order_uuid,
                order.telegram_id,
                exc.message,
            )
            remote_users = await self._remnawave_client.get_users_by_telegram_id(
                order.telegram_id
            )
            existing_user = self._select_referral_bonus_target(remote_users)
            if existing_user is None:
                raise  # re-raise original 400 if we truly can't find anyone
            return await self._extend_remnawave_user(
                order,
                existing_user,
                plan=plan,
                config=config,
            )

    async def _extend_remnawave_user(
        self,
        order: PaymentOrder,
        remote_user: RemnawaveUser,
        *,
        plan: PaymentPlan,
        config: BotConfigSnapshot,
    ) -> RemnawaveUser:
        current_expire_at = remote_user.expire_at
        if current_expire_at.tzinfo is None:
            current_expire_at = current_expire_at.replace(tzinfo=timezone.utc)

        base_expire_at = max(current_expire_at, datetime.now(timezone.utc))
        next_expire_at = base_expire_at + timedelta(days=order.plan_duration_days)

        traffic_limit_bytes = order.traffic_limit_bytes
        if traffic_limit_bytes <= 0:
            traffic_limit_bytes = remote_user.traffic_limit_bytes

        body: dict[str, object] = {
            "uuid": str(remote_user.uuid),
            "username": remote_user.username,
            "expireAt": next_expire_at.isoformat().replace("+00:00", "Z"),
            "trafficLimitBytes": traffic_limit_bytes,
            "trafficLimitStrategy": order.traffic_limit_strategy,
            "telegramId": order.telegram_id,
            "description": f"Updated by Telegram bot manual payment {order.order_uuid}",
            "activeInternalSquads": plan.internal_squad_uuids or [],
        }
        if config.payment_user_tag_normalized:
            body["tag"] = config.payment_user_tag_normalized
        if config.payment_external_squad_uuid_normalized:
            body["externalSquadUuid"] = config.payment_external_squad_uuid_normalized
        return await self._remnawave_client.update_user(body)

    async def _save_fulfilled_order(self, order_uuid: str, remote_user: RemnawaveUser) -> None:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder).where(PaymentOrder.order_uuid == order_uuid)
            )
            if order is None:
                return
            order.status = FULFILLED_PAYMENT_STATUS
            order.remnawave_user_uuid = str(remote_user.uuid)
            order.subscription_url = remote_user.subscription_url
            order.error_message = None
            order.paid_at = utcnow()
            order.fulfilled_at = utcnow()
            order.updated_at = utcnow()
            await session.commit()

    async def _save_order_error(self, order_uuid: str, error_message: str) -> None:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder).where(PaymentOrder.order_uuid == order_uuid)
            )
            if order is None:
                return
            order.error_message = error_message
            order.updated_at = utcnow()
            await session.commit()

    async def _restore_review_after_failed_approval(
        self,
        order_uuid: str,
        error_message: str,
    ) -> None:
        async with self._session_factory() as session:
            order = await session.scalar(
                select(PaymentOrder)
                .where(PaymentOrder.order_uuid == order_uuid)
                .with_for_update()
            )
            if order is None:
                return

            if order.status == PROCESSING_PAYMENT_STATUS:
                order.status = REVIEW_PAYMENT_STATUS
            order.error_message = error_message
            order.updated_at = utcnow()
            await session.commit()

    async def _grant_referral_bonus(
        self,
        order: PaymentOrder,
        *,
        config: BotConfigSnapshot,
    ) -> ReferralBonusGrant | None:
        if not config.referral_enabled:
            return None

        bonus_days = max(config.referral_bonus_days, 0)
        if bonus_days <= 0:
            return None

        inviter_telegram_id = await self._get_inviter_telegram_id(order.telegram_id)
        if inviter_telegram_id is None or inviter_telegram_id == order.telegram_id:
            return None

        await self._create_referral_bonus(
            inviter_telegram_id=inviter_telegram_id,
            invited_telegram_id=order.telegram_id,
            source_order_uuid=order.order_uuid,
            bonus_days=bonus_days,
        )

        try:
            applied_bonus = await self.apply_pending_referral_bonuses(
                telegram_id=inviter_telegram_id,
            )
        except RemnawaveApiError as exc:
            logger.warning(
                "Could not apply referral bonus for inviter=%s after order=%s: %s",
                inviter_telegram_id,
                order.order_uuid,
                exc.message,
            )
            applied_bonus = None

        if applied_bonus is None:
            return ReferralBonusGrant(
                inviter_telegram_id=inviter_telegram_id,
                bonus_days=bonus_days,
                status=REFERRAL_BONUS_PENDING_STATUS,
            )

        return ReferralBonusGrant(
            inviter_telegram_id=inviter_telegram_id,
            bonus_days=bonus_days,
            status=REFERRAL_BONUS_APPLIED_STATUS,
            total_applied_days=applied_bonus.total_bonus_days,
            expire_at=applied_bonus.expire_at,
            applied_to_user_uuid=applied_bonus.applied_to_user_uuid,
        )

    async def _get_inviter_telegram_id(self, invited_telegram_id: int) -> int | None:
        async with self._session_factory() as session:
            return await session.scalar(
                select(ReferralInvite.inviter_telegram_id).where(
                    ReferralInvite.invited_telegram_id == invited_telegram_id
                )
            )

    async def _create_referral_bonus(
        self,
        *,
        inviter_telegram_id: int,
        invited_telegram_id: int,
        source_order_uuid: str,
        bonus_days: int,
    ) -> None:
        async with self._session_factory() as session:
            existing = await session.scalar(
                select(ReferralBonus).where(
                    ReferralBonus.source_order_uuid == source_order_uuid
                )
            )
            if existing is not None:
                return

            session.add(
                ReferralBonus(
                    inviter_telegram_id=inviter_telegram_id,
                    invited_telegram_id=invited_telegram_id,
                    source_order_uuid=source_order_uuid,
                    bonus_days=bonus_days,
                    status=REFERRAL_BONUS_PENDING_STATUS,
                )
            )
            await session.commit()

    async def _claim_pending_referral_bonuses(self, telegram_id: int) -> list[ReferralBonus]:
        async with self._session_factory() as session:
            bonuses = list(
                (
                    await session.scalars(
                        select(ReferralBonus)
                        .where(
                            ReferralBonus.inviter_telegram_id == telegram_id,
                            ReferralBonus.status == REFERRAL_BONUS_PENDING_STATUS,
                        )
                        .order_by(ReferralBonus.created_at.asc(), ReferralBonus.id.asc())
                        .with_for_update()
                    )
                ).all()
            )
            if not bonuses:
                return []

            for bonus in bonuses:
                bonus.status = REFERRAL_BONUS_PROCESSING_STATUS
                bonus.updated_at = utcnow()
            await session.commit()
            return bonuses

    async def _restore_claimed_referral_bonuses(
        self,
        bonus_ids: list[int],
        *,
        status: str,
    ) -> None:
        if not bonus_ids:
            return

        async with self._session_factory() as session:
            bonuses = list(
                (
                    await session.scalars(
                        select(ReferralBonus)
                        .where(ReferralBonus.id.in_(bonus_ids))
                        .with_for_update()
                    )
                ).all()
            )
            for bonus in bonuses:
                bonus.status = status
                bonus.updated_at = utcnow()
            await session.commit()

    async def _mark_referral_bonuses_applied(
        self,
        bonus_ids: list[int],
        *,
        applied_to_user_uuid: str,
    ) -> None:
        if not bonus_ids:
            return

        async with self._session_factory() as session:
            bonuses = list(
                (
                    await session.scalars(
                        select(ReferralBonus)
                        .where(ReferralBonus.id.in_(bonus_ids))
                        .with_for_update()
                    )
                ).all()
            )
            for bonus in bonuses:
                bonus.status = REFERRAL_BONUS_APPLIED_STATUS
                bonus.applied_to_user_uuid = applied_to_user_uuid
                bonus.applied_at = utcnow()
                bonus.updated_at = utcnow()
            await session.commit()

    @staticmethod
    def _select_referral_bonus_target(
        remote_users: list[RemnawaveUser],
    ) -> RemnawaveUser | None:
        if not remote_users:
            return None

        def _sort_key(user: RemnawaveUser) -> tuple[int, datetime, str]:
            expire_at = user.expire_at
            if expire_at.tzinfo is None:
                expire_at = expire_at.replace(tzinfo=timezone.utc)
            return (0 if user.status.upper() == "ACTIVE" else 1, expire_at, user.username)

        return sorted(remote_users, key=_sort_key)[0]

    @staticmethod
    def _select_cached_renewal_target(cached_users: list[object]) -> object | None:
        records = [
            record
            for access in cached_users
            if (record := getattr(access, "record", None)) is not None
            and getattr(record, "rw_uuid", None)
            and getattr(record, "username", None)
        ]
        if not records:
            return None

        def _sort_key(record: object) -> tuple[int, datetime, str]:
            expire_at = getattr(record, "expire_at", None) or datetime.min
            if expire_at.tzinfo is None:
                expire_at = expire_at.replace(tzinfo=timezone.utc)
            status = str(getattr(record, "status", "")).upper()
            username = str(getattr(record, "username", ""))
            return (0 if status == "ACTIVE" else 1, expire_at, username)

        return sorted(records, key=_sort_key)[0]

    async def _upsert_telegram_user(
        self,
        session: AsyncSession,
        *,
        telegram_id: int,
        username: str | None,
        first_name: str | None,
        last_name: str | None,
        language_code: str | None,
    ) -> TelegramUser:
        record = await session.scalar(
            select(TelegramUser).where(TelegramUser.telegram_id == telegram_id)
        )

        if record is None:
            record = TelegramUser(telegram_id=telegram_id)
            session.add(record)

        record.username = username
        record.first_name = first_name
        record.last_name = last_name
        record.language_code = language_code
        record.is_admin = self._settings.is_admin(telegram_id)
        record.last_seen_at = utcnow()
        record.updated_at = utcnow()

        await session.flush()
        return record

    def _build_provision_username(
        self,
        *,
        config: BotConfigSnapshot,
        order_uuid: str,
        telegram_id: int,
        username: str | None,
        first_name: str | None,
    ) -> str:
        base_name = username or first_name or "user"
        normalized_name = self._normalize_provision_name(base_name)

        prefix = config.payment_username_prefix_normalized or "Oreol"
        unique_suffix = order_uuid.replace("-", "")[:6]
        provision_username = f"{prefix}-{telegram_id}-{unique_suffix}-{normalized_name}"
        return provision_username[:36]

    @staticmethod
    def _normalize_provision_name(raw_name: str) -> str:
        transliterated = raw_name.translate(_CYRILLIC_TRANSLITERATION)
        decomposed = unicodedata.normalize("NFKD", transliterated)
        ascii_name = decomposed.encode("ascii", "ignore").decode("ascii")
        normalized_name = "".join(
            ch for ch in ascii_name if ch.isalnum() or ch in "_-"
        ).strip("-_")
        return normalized_name or "user"