remnawave-minishop/backend/bot/main_bot.py

import asyncio
import logging
from typing import Awaitable, Callable, Optional

from aiogram import Bot, Dispatcher
from aiogram.exceptions import TelegramNetworkError
from aiogram.types import BotCommand, MenuButtonDefault, MenuButtonWebApp, WebAppInfo

from bot.app.controllers.dispatcher_controller import build_dispatcher
from bot.app.factories.build_services import build_core_services
from bot.app.web.web_server import build_and_start_web_app
from bot.infra.redis import close_redis
from bot.middlewares.i18n import JsonI18n
from bot.routers import build_root_router
from bot.services.locale_override_service import load_locale_overrides
from bot.utils.message_queue import init_queue_manager
from config.settings import Settings
from db.database_setup import init_db, init_db_connection

TELEGRAM_STARTUP_RETRY_DELAY_SECONDS = 2.0


def redact_token(value: str, token: Optional[str]) -> str:
    if not value or not token:
        return value
    return value.replace(token, "***")


def _telegram_network_error_detail(exc: TelegramNetworkError) -> str:
    root_cause = exc.__cause__ or exc.__context__
    detail = str(exc)
    if root_cause:
        root_detail = f"{type(root_cause).__name__}: {root_cause}"
        if root_detail not in detail:
            detail = f"{detail} ({root_detail})"
    return detail


async def _run_telegram_startup_step(
    action: str,
    step: Callable[[], Awaitable[object]],
    unexpected_log_message: str,
    *,
    attempts: Optional[int] = None,
    retry_delay_seconds: float = TELEGRAM_STARTUP_RETRY_DELAY_SECONDS,
) -> bool:
    attempt = 1
    max_attempts = max(1, attempts) if attempts is not None else None
    while True:
        try:
            await step()
            if attempt > 1:
                logging.info(
                    "STARTUP: Telegram step succeeded while %s on attempt %s%s.",
                    action,
                    attempt,
                    f"/{max_attempts}" if max_attempts is not None else "",
                )
            return True
        except TelegramNetworkError as exc:
            detail = _telegram_network_error_detail(exc)
            attempt_label = (
                f"{attempt}/{max_attempts}" if max_attempts is not None else str(attempt)
            )
            if max_attempts is not None and attempt >= max_attempts:
                logging.warning(
                    "STARTUP: Telegram network error while %s after %s attempts: %s.",
                    action,
                    max_attempts,
                    detail,
                )
                return False
            logging.warning(
                "STARTUP: Telegram network error while %s on attempt %s: %s. "
                "Retrying in %.1fs and will keep trying until Telegram is reachable.",
                action,
                attempt_label,
                detail,
                retry_delay_seconds,
            )
            attempt += 1
            await asyncio.sleep(retry_delay_seconds)
            continue
        except Exception:
            logging.exception(unexpected_log_message)
            return False


async def register_all_routers(dp: Dispatcher, settings: Settings):
    dp.include_router(build_root_router(settings))
    logging.info("All application routers registered.")


async def configure_telegram_webhook(dispatcher: Dispatcher) -> None:
    bot: Bot = dispatcher["bot_instance"]
    settings: Settings = dispatcher["settings"]

    telegram_webhook_url_to_set = settings.WEBHOOK_BASE_URL
    if telegram_webhook_url_to_set:
        full_telegram_webhook_url = (
            f"{str(telegram_webhook_url_to_set).rstrip('/')}{settings.telegram_webhook_path}"
        )

        logging.info(
            "STARTUP: Attempting to set Telegram webhook to: %s",
            redact_token(full_telegram_webhook_url, settings.BOT_TOKEN),
        )

        async def _configure_webhook() -> None:
            current_webhook_info = await bot.get_webhook_info()
            logging.info(
                f"STARTUP: Current Telegram webhook info BEFORE setting: {current_webhook_info.model_dump_json(exclude_none=True, indent=2)}"  # noqa: E501
            )

            set_success = await bot.set_webhook(
                url=full_telegram_webhook_url,
                secret_token=settings.WEBHOOK_SECRET_TOKEN,
                drop_pending_updates=True,
                allowed_updates=dispatcher.resolve_used_update_types(),
            )
            if set_success:
                logging.info(
                    "STARTUP: bot.set_webhook to %s returned SUCCESS (True).",
                    redact_token(full_telegram_webhook_url, settings.BOT_TOKEN),
                )
            else:
                logging.error(
                    "STARTUP: bot.set_webhook to %s returned FAILURE (False).",
                    redact_token(full_telegram_webhook_url, settings.BOT_TOKEN),
                )

            new_webhook_info = await bot.get_webhook_info()
            logging.info(
                f"STARTUP: Telegram Webhook info AFTER setting: {new_webhook_info.model_dump_json(exclude_none=True, indent=2)}"  # noqa: E501
            )
            if not new_webhook_info.url:
                logging.error(
                    "STARTUP: CRITICAL - Telegram Webhook URL is EMPTY after set attempt. Check bot token and URL validity."  # noqa: E501
                )

        await _run_telegram_startup_step(
            "configuring Telegram webhook",
            _configure_webhook,
            "STARTUP: EXCEPTION during set/get Telegram webhook.",
        )
    else:
        logging.error(
            "STARTUP: WEBHOOK_BASE_URL not set in environment. Webhook mode is required. Exiting."
        )
        raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")


async def on_startup_configured(dispatcher: Dispatcher):
    bot: Bot = dispatcher["bot_instance"]
    settings: Settings = dispatcher["settings"]
    i18n_instance: JsonI18n = dispatcher["i18n_instance"]

    logging.info("STARTUP: on_startup_configured executing...")

    if settings.SUBSCRIPTION_MINI_APP_URL:

        async def _configure_mini_app_menu() -> None:
            menu_text = i18n_instance.gettext(
                settings.DEFAULT_LANGUAGE,
                "menu_personal_account_button",
            )
            await bot.set_chat_menu_button(
                menu_button=MenuButtonWebApp(
                    text=menu_text,
                    web_app=WebAppInfo(url=settings.SUBSCRIPTION_MINI_APP_URL),
                )
            )
            await bot.set_chat_menu_button(menu_button=MenuButtonDefault())
            logging.info("STARTUP: Mini app domain registered and default menu button restored.")

        await _run_telegram_startup_step(
            "registering mini app menu button",
            _configure_mini_app_menu,
            "STARTUP: Failed to register mini app domain.",
        )

    async def _configure_bot_commands() -> None:
        bot_commands = [
            BotCommand(command="tg", description="Интерфейс в боте"),
        ]
        if settings.START_COMMAND_DESCRIPTION:
            bot_commands.insert(
                0,
                BotCommand(command="start", description=settings.START_COMMAND_DESCRIPTION),
            )
        await bot.set_my_commands(bot_commands)
        logging.info("STARTUP: bot command descriptions set.")

    await _run_telegram_startup_step(
        "setting bot commands",
        _configure_bot_commands,
        "STARTUP: Failed to set bot commands.",
    )

    # Initialize message queue manager
    try:
        queue_manager = init_queue_manager(bot)
        dispatcher["queue_manager"] = queue_manager
        logging.info("STARTUP: Message queue manager initialized")
    except Exception:
        logging.exception("STARTUP: Failed to initialize message queue manager.")

    logging.info("STARTUP: Bot on_startup_configured completed.")


async def on_shutdown_configured(dispatcher: Dispatcher):
    logging.warning("SHUTDOWN: on_shutdown_configured executing...")

    async def close_service(key: str) -> None:
        service = dispatcher.get(key)
        if not service:
            return
        close_coro = getattr(service, "close", None)
        if callable(close_coro):
            try:
                await close_coro()
                logging.info(f"{key} closed on shutdown.")
            except Exception as e:
                logging.warning(f"Failed to close {key}: {e}")
        else:
            close_session = getattr(service, "close_session", None)
            if callable(close_session):
                try:
                    await close_session()
                    logging.info(f"{key} session closed on shutdown.")
                except Exception as e:
                    logging.warning(f"Failed to close session for {key}: {e}")

    from bot.payment_providers import iter_service_keys

    for service_key in (
        "panel_service",
        "panel_webhook_service",
        "lknpd_service",
        "promo_code_service",
        "subscription_service",
        "referral_service",
        "support_service",
        "notification_service",
        "email_auth_service",
        *iter_service_keys(),
    ):
        await close_service(service_key)

    bot: Bot = dispatcher["bot_instance"]
    if bot and bot.session:
        try:
            await bot.session.close()
            logging.info("SHUTDOWN: Aiogram Bot session closed.")
        except Exception as e:
            logging.warning(f"SHUTDOWN: Failed to close bot session: {e}")

    from db.database_setup import async_engine as global_async_engine

    if global_async_engine:
        logging.info("SHUTDOWN: Disposing SQLAlchemy engine...")
        await global_async_engine.dispose()
        logging.info("SHUTDOWN: SQLAlchemy engine disposed.")
    await close_redis()

    logging.info("SHUTDOWN: Bot on_shutdown_configured completed.")


async def run_bot(settings_param: Settings):
    local_async_session_factory = init_db_connection(settings_param)
    if local_async_session_factory is None:
        logging.critical("Failed to initialize database connection and session factory. Exiting.")
        return
    await init_db(settings_param, local_async_session_factory)
    dp, bot, extra = build_dispatcher(settings_param, local_async_session_factory)
    i18n_instance = extra["i18n_instance"]
    await load_locale_overrides(i18n_instance, local_async_session_factory)

    # Get bot username for YooKassa default return URL if needed
    actual_bot_username = "your_bot_username"

    async def _resolve_bot_username() -> None:
        nonlocal actual_bot_username
        bot_info = await bot.get_me()
        if bot_info.username:
            actual_bot_username = bot_info.username
            dp["bot_username"] = actual_bot_username
            logging.info(f"Bot username resolved: @{actual_bot_username}")
        else:
            logging.warning("Bot username is empty; Telegram Login Widget will be unavailable.")

    bot_username_resolved = await _run_telegram_startup_step(
        "getting bot info from Telegram",
        _resolve_bot_username,
        f"Failed to get bot info (e.g., for YooKassa default URL). Using fallback: {actual_bot_username}",  # noqa: E501
    )
    if not bot_username_resolved:
        logging.warning("Using fallback bot username: %s", actual_bot_username)

    services = build_core_services(
        settings_param,
        bot,
        local_async_session_factory,
        i18n_instance,
        actual_bot_username,
    )
    for key, service in services.items():
        dp[key] = service
    dp["panel_service"] = services["panel_service"]
    dp["async_session_factory"] = local_async_session_factory

    # Wrap startup/shutdown handlers to satisfy aiogram event signature (no args passed)
    async def _on_startup_wrapper():
        await on_startup_configured(dp)

    async def _on_shutdown_wrapper():
        await on_shutdown_configured(dp)

    dp.startup.register(_on_startup_wrapper)
    dp.shutdown.register(_on_shutdown_wrapper)

    await register_all_routers(dp, settings_param)

    if not settings_param.WEBHOOK_BASE_URL:
        logging.error("WEBHOOK_BASE_URL is required. Polling mode is disabled. Exiting.")
        await dp.emit_shutdown()
        raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")

    from bot.payment_providers import get_provider_spec

    _yk_spec = get_provider_spec("yookassa")
    _yk_path = _yk_spec.webhook_path(settings_param) if _yk_spec and _yk_spec.webhook_path else "-"
    logging.info(
        "Starting AIOHTTP server: webhook_base=%s yookassa_path=%s",
        settings_param.WEBHOOK_BASE_URL,
        _yk_path,
    )

    async def _after_webhooks_started() -> None:
        await configure_telegram_webhook(dp)

    async def web_server_task():
        await build_and_start_web_app(
            dp,
            bot,
            settings_param,
            local_async_session_factory,
            after_webhooks_started=_after_webhooks_started,
        )

    main_tasks = [asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")]

    try:
        await asyncio.gather(*main_tasks)
    except (KeyboardInterrupt, SystemExit, asyncio.CancelledError) as e:
        logging.info(f"Main bot loop interrupted/cancelled: {type(e).__name__} - {e}")
    finally:
        logging.info("Initiating final bot shutdown sequence...")
        for task in main_tasks:
            if task and not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    logging.info(f"Task '{task.get_name()}' was cancelled successfully.")
                except Exception as e_task_cancel:
                    logging.error(
                        f"Error during cancellation of task '{task.get_name()}': {e_task_cancel}",
                        exc_info=True,
                    )

        await dp.emit_shutdown()
        logging.info("Dispatcher shutdown sequence emitted.")

        logging.info("Bot run_bot function finished.")