Merge pull request #76 from machka-pasla/dev

add new referral param and broadcast for groups of users

.env.example

@@ -76,6 +76,8 @@ SUBSCRIPTION_NOTIFY_ON_EXPIRE=True
 SUBSCRIPTION_NOTIFY_AFTER_EXPIRE=True
 SUBSCRIPTION_NOTIFY_DAYS_BEFORE=3
 
+
+REFERRAL_ONE_BONUS_PER_REFEREE=False
 # Referral Bonus Days
 REFERRAL_BONUS_DAYS_1_MONTH=3
 REFERRAL_BONUS_DAYS_3_MONTHS=7

README.md

@@ -28,7 +28,6 @@
 -   **aiohttp:** Для запуска веб-сервера (вебхуки).
 -   **SQLAlchemy 2.x & asyncpg:** Асинхронная работа с базой данных PostgreSQL.
 -   **YooKassa, aiocryptopay:** SDK для интеграции с платежными системами.
--   **APScheduler:** Для выполнения отложенных задач (например, уведомления об окончании подписки).
 -   **Pydantic:** Для управления настройками из `.env` файла.
 -   **Docker & Docker Compose:** Для контейнеризации и развертывания.
 

bot/app/controllers/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/controllers/dispatcher_controller.py

@@ -0,0 +1,38 @@
+import logging
+from typing import Dict
+
+from aiogram import Bot, Dispatcher
+from aiogram.enums import ParseMode
+from aiogram.client.default import DefaultBotProperties
+from aiogram.fsm.storage.memory import MemoryStorage
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+from bot.middlewares.db_session import DBSessionMiddleware
+from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n
+from bot.middlewares.ban_check_middleware import BanCheckMiddleware
+from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
+from bot.middlewares.profile_sync import ProfileSyncMiddleware
+
+
+def build_dispatcher(settings: Settings, async_session_factory: sessionmaker) -> tuple[Dispatcher, Bot, Dict]:
+    storage = MemoryStorage()
+    default_props = DefaultBotProperties(parse_mode=ParseMode.HTML)
+    bot = Bot(token=settings.BOT_TOKEN, default=default_props)
+
+    dp = Dispatcher(storage=storage, settings=settings, bot_instance=bot)
+
+    i18n_instance = get_i18n_instance(path="locales", default=settings.DEFAULT_LANGUAGE)
+
+    dp["i18n_instance"] = i18n_instance
+    dp["async_session_factory"] = async_session_factory
+
+    dp.update.outer_middleware(DBSessionMiddleware(async_session_factory))
+    dp.update.outer_middleware(I18nMiddleware(i18n=i18n_instance, settings=settings))
+    dp.update.outer_middleware(ProfileSyncMiddleware())
+    dp.update.outer_middleware(BanCheckMiddleware(settings=settings, i18n_instance=i18n_instance))
+    dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings))
+
+    return dp, bot, {"i18n_instance": i18n_instance}
+
+

bot/app/factories/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/factories/build_services.py

@@ -0,0 +1,69 @@
+from aiogram import Bot
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+from bot.middlewares.i18n import JsonI18n
+from bot.services.yookassa_service import YooKassaService
+from bot.services.panel_api_service import PanelApiService
+from bot.services.subscription_service import SubscriptionService
+from bot.services.referral_service import ReferralService
+from bot.services.promo_code_service import PromoCodeService
+from bot.services.stars_service import StarsService
+from bot.services.tribute_service import TributeService
+from bot.services.crypto_pay_service import CryptoPayService
+from bot.services.panel_webhook_service import PanelWebhookService
+
+
+def build_core_services(
+    settings: Settings,
+    bot: Bot,
+    async_session_factory: sessionmaker,
+    i18n: JsonI18n,
+    bot_username_for_default_return: str,
+):
+    panel_service = PanelApiService(settings)
+    subscription_service = SubscriptionService(settings, panel_service, bot, i18n)
+    referral_service = ReferralService(settings, subscription_service, bot, i18n)
+    promo_code_service = PromoCodeService(settings, subscription_service, bot, i18n)
+    stars_service = StarsService(bot, settings, i18n, subscription_service, referral_service)
+    cryptopay_service = CryptoPayService(
+        settings.CRYPTOPAY_TOKEN,
+        settings.CRYPTOPAY_NETWORK,
+        bot,
+        settings,
+        i18n,
+        async_session_factory,
+        subscription_service,
+        referral_service,
+    )
+    tribute_service = TributeService(
+        bot,
+        settings,
+        i18n,
+        async_session_factory,
+        panel_service,
+        subscription_service,
+        referral_service,
+    )
+    panel_webhook_service = PanelWebhookService(bot, settings, i18n, async_session_factory)
+    yookassa_service = YooKassaService(
+        shop_id=settings.YOOKASSA_SHOP_ID,
+        secret_key=settings.YOOKASSA_SECRET_KEY,
+        configured_return_url=settings.YOOKASSA_RETURN_URL,
+        bot_username_for_default_return=bot_username_for_default_return,
+        settings_obj=settings,
+    )
+
+    return {
+        "panel_service": panel_service,
+        "subscription_service": subscription_service,
+        "referral_service": referral_service,
+        "promo_code_service": promo_code_service,
+        "stars_service": stars_service,
+        "cryptopay_service": cryptopay_service,
+        "tribute_service": tribute_service,
+        "panel_webhook_service": panel_webhook_service,
+        "yookassa_service": yookassa_service,
+    }
+
+

bot/app/web/__init__.py

@@ -0,0 +1,2 @@
+
+

bot/app/web/web_server.py

@@ -0,0 +1,91 @@
+import asyncio
+import logging
+from aiohttp import web
+from aiogram import Bot, Dispatcher
+from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
+from sqlalchemy.orm import sessionmaker
+
+from config.settings import Settings
+
+
+async def build_and_start_web_app(
+    dp: Dispatcher,
+    bot: Bot,
+    settings: Settings,
+    async_session_factory: sessionmaker,
+):
+    app = web.Application()
+    app["bot"] = bot
+    app["dp"] = dp
+    app["settings"] = settings
+    app["async_session_factory"] = async_session_factory
+    # Inject shared instances used by webhook handlers
+    app["i18n"] = dp.get("i18n_instance")
+    for key in (
+        "yookassa_service",
+        "subscription_service",
+        "referral_service",
+        "panel_service",
+        "stars_service",
+        "cryptopay_service",
+        "tribute_service",
+        "panel_webhook_service",
+    ):
+        # Access dispatcher workflow_data directly to avoid sequence protocol issues
+        if hasattr(dp, "workflow_data") and key in dp.workflow_data:  # type: ignore
+            app[key] = dp.workflow_data[key]  # type: ignore
+
+    setup_application(app, dp, bot=bot)
+
+    telegram_uses_webhook_mode = bool(settings.WEBHOOK_BASE_URL)
+
+    if telegram_uses_webhook_mode:
+        telegram_webhook_path = f"/{settings.BOT_TOKEN}"
+        app.router.add_post(telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot))
+        logging.info(
+            f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
+        )
+
+    from bot.handlers.user.payment import yookassa_webhook_route
+    from bot.services.tribute_service import tribute_webhook_route
+    from bot.services.crypto_pay_service import cryptopay_webhook_route
+    from bot.services.panel_webhook_service import panel_webhook_route
+
+    tribute_path = settings.tribute_webhook_path
+    if tribute_path.startswith("/"):
+        app.router.add_post(tribute_path, tribute_webhook_route)
+        logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
+
+    cp_path = settings.cryptopay_webhook_path
+    if cp_path.startswith("/"):
+        app.router.add_post(cp_path, cryptopay_webhook_route)
+        logging.info(f"CryptoPay webhook route configured at: [POST] {cp_path}")
+
+    # YooKassa webhook (register only when base URL present and path configured)
+    yk_path = settings.yookassa_webhook_path
+    if settings.WEBHOOK_BASE_URL and yk_path and yk_path.startswith("/"):
+        app.router.add_post(yk_path, yookassa_webhook_route)
+        logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}")
+
+    panel_path = settings.panel_webhook_path
+    if panel_path.startswith("/"):
+        app.router.add_post(panel_path, panel_webhook_route)
+        logging.info(f"Panel webhook route configured at: [POST] {panel_path}")
+
+    web_app_runner = web.AppRunner(app)
+    await web_app_runner.setup()
+    site = web.TCPSite(
+        web_app_runner,
+        host=settings.WEB_SERVER_HOST,
+        port=settings.WEB_SERVER_PORT,
+    )
+
+    await site.start()
+    logging.info(
+        f"AIOHTTP server started on http://{settings.WEB_SERVER_HOST}:{settings.WEB_SERVER_PORT}"
+    )
+
+    # Run until cancelled
+    await asyncio.Event().wait()
+
+

bot/handlers/admin/broadcast.py

@@ -113,17 +113,58 @@ async def process_broadcast_message_handler(
     await state.update_data(
         broadcast_text=text,
         broadcast_entities=entities,
+        broadcast_target="all",
     )
 
     confirmation_prompt = _("admin_broadcast_confirm_prompt", message_preview=text)
 
     await message.answer(
         confirmation_prompt,
-        reply_markup=get_broadcast_confirmation_keyboard(current_lang, i18n),
+        reply_markup=get_broadcast_confirmation_keyboard(current_lang, i18n, target="all"),
     )
     await state.set_state(AdminStates.confirming_broadcast)
 
 
+@router.callback_query(
+    F.data.startswith("broadcast_target:"),
+    AdminStates.confirming_broadcast,
+)
+async def change_broadcast_target_handler(
+    callback: types.CallbackQuery,
+    state: FSMContext,
+    i18n_data: dict,
+    settings: Settings,
+):
+    current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
+    i18n: Optional[JsonI18n] = i18n_data.get("i18n_instance")
+    if not i18n or not callback.message:
+        await callback.answer("Error updating selection.", show_alert=True)
+        return
+
+    new_target = callback.data.split(":")[1]
+    if new_target not in {"all", "active", "inactive"}:
+        await callback.answer("Unknown target.", show_alert=True)
+        return
+
+    await state.update_data(broadcast_target=new_target)
+    user_fsm_data = await state.get_data()
+    text = user_fsm_data.get("broadcast_text", "")
+    _ = lambda key, **kwargs: i18n.gettext(current_lang, key, **kwargs)
+    confirmation_prompt = _(
+        "admin_broadcast_confirm_prompt", message_preview=text
+    )
+    try:
+        await callback.message.edit_text(
+            confirmation_prompt,
+            reply_markup=get_broadcast_confirmation_keyboard(
+                current_lang, i18n, target=new_target
+            ),
+        )
+    except Exception:
+        pass
+    await callback.answer()
+
+
 @router.callback_query(
     F.data == "admin_action:main", AdminStates.waiting_for_broadcast_message
 )
@@ -194,7 +235,13 @@ async def confirm_broadcast_callback_handler(
         await callback.message.edit_text(_("admin_broadcast_sending_started"), reply_markup=None)
         await callback.answer()
 
-        user_ids = await user_dal.get_all_active_user_ids_for_broadcast(session)
+        target = user_fsm_data.get("broadcast_target", "all")
+        if target == "active":
+            user_ids = await user_dal.get_user_ids_with_active_subscription(session)
+        elif target == "inactive":
+            user_ids = await user_dal.get_user_ids_without_active_subscription(session)
+        else:
+            user_ids = await user_dal.get_all_active_user_ids_for_broadcast(session)
 
         sent_count = 0
         failed_count = 0

bot/handlers/user/payment.py

@@ -123,7 +123,12 @@ async def process_successful_payment(session: AsyncSession, bot: Bot,
             "applied_promo_bonus_days", 0)
 
         referral_bonus_info = await referral_service.apply_referral_bonuses_for_payment(
-            session, user_id, subscription_months)
+            session,
+            user_id,
+            subscription_months,
+            current_payment_db_id=payment_db_id,
+            skip_if_active_before_payment=False,
+        )
         applied_referee_bonus_days_from_referral: Optional[int] = None
         if referral_bonus_info and referral_bonus_info.get(
                 "referee_new_end_date"):

bot/handlers/user/start.py

@@ -194,8 +194,15 @@ async def start_command_handler(message: types.Message,
         update_payload = {}
         if db_user.language_code != current_lang:
             update_payload["language_code"] = current_lang
+        # Set referral only if not already set AND user is not currently active.
+        # This allows previously subscribed but currently inactive users to be attributed.
         if referred_by_user_id and db_user.referred_by_id is None:
-            update_payload["referred_by_id"] = referred_by_user_id
+            try:
+                is_active_now = await subscription_service.has_active_subscription(session, user_id)
+            except Exception:
+                is_active_now = False
+            if not is_active_now:
+                update_payload["referred_by_id"] = referred_by_user_id
         if user.username != db_user.username:
             update_payload["username"] = user.username
         if user.first_name != db_user.first_name:

bot/keyboards/inline/admin_keyboards.py

@@ -260,12 +260,47 @@ def get_confirmation_keyboard(yes_callback_data: str, no_callback_data: str,
 
 
 def get_broadcast_confirmation_keyboard(lang: str,
-                                        i18n_instance) -> InlineKeyboardMarkup:
+                                        i18n_instance,
+                                        target: str = "all") -> InlineKeyboardMarkup:
     _ = lambda key, **kwargs: i18n_instance.gettext(lang, key, **kwargs)
     builder = InlineKeyboardBuilder()
-    builder.button(text=_(key="confirm_broadcast_send_button"),
+
+    # Row: target selection (all / active / inactive)
+    target_all_label = _(
+        key="broadcast_target_all_button",
+        default="👥 Все"
+    )
+    target_active_label = _(
+        key="broadcast_target_active_button",
+        default="✅ Активные"
+    )
+    target_inactive_label = _(
+        key="broadcast_target_inactive_button",
+        default="⌛ Неактивные"
+    )
+
+    # Highlight current selection with a prefix
+    def mark_selected(label: str, is_selected: bool) -> str:
+        return ("• " + label) if is_selected else label
+
+    builder.button(
+        text=mark_selected(target_all_label, target == "all"),
+        callback_data="broadcast_target:all",
+    )
+    builder.button(
+        text=mark_selected(target_active_label, target == "active"),
+        callback_data="broadcast_target:active",
+    )
+    builder.button(
+        text=mark_selected(target_inactive_label, target == "inactive"),
+        callback_data="broadcast_target:inactive",
+    )
+    builder.adjust(3)
+
+    # Row: confirmation
+    builder.button(text=_(key="confirm_broadcast_send_button", default="🚀 Отправить"),
                    callback_data="broadcast_final_action:send")
-    builder.button(text=_(key="cancel_broadcast_button"),
+    builder.button(text=_(key="cancel_broadcast_button", default="❌ Отмена"),
                    callback_data="broadcast_final_action:cancel")
     builder.adjust(2)
     return builder.as_markup()

bot/main_bot.py

@@ -20,6 +20,10 @@ from bot.middlewares.i18n import I18nMiddleware, get_i18n_instance, JsonI18n
 from bot.middlewares.db_session import DBSessionMiddleware
 from bot.middlewares.ban_check_middleware import BanCheckMiddleware
 from bot.middlewares.action_logger_middleware import ActionLoggerMiddleware
+from bot.middlewares.profile_sync import ProfileSyncMiddleware
+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.routers import build_root_router
 
@@ -224,19 +228,16 @@ async def on_shutdown_configured(dispatcher: Dispatcher):
 
 
 async def run_bot(settings_param: Settings):
-    storage = MemoryStorage()
-    default_props = DefaultBotProperties(parse_mode=ParseMode.HTML)
-    bot = Bot(token=settings_param.BOT_TOKEN, default=default_props)
-
     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
+    dp, bot, extra = build_dispatcher(settings_param, local_async_session_factory)
+    i18n_instance = extra["i18n_instance"]
 
-    dp = Dispatcher(storage=storage, settings=settings_param, bot_instance=bot)
-
+    # Get bot username for YooKassa default return URL if needed
     actual_bot_username = "your_bot_username"
     try:
         bot_info = await bot.get_me()
@@ -247,81 +248,25 @@ async def run_bot(settings_param: Settings):
             f"Failed to get bot info (e.g., for YooKassa default URL): {e}. Using fallback: {actual_bot_username}"
         )
 
-    i18n_instance = get_i18n_instance(
-        path="locales", default=settings_param.DEFAULT_LANGUAGE
-    )
-
-    yookassa_service = YooKassaService(
-        shop_id=settings_param.YOOKASSA_SHOP_ID,
-        secret_key=settings_param.YOOKASSA_SECRET_KEY,
-        configured_return_url=settings_param.YOOKASSA_RETURN_URL,
-        bot_username_for_default_return=actual_bot_username,
-        settings_obj=settings_param,
-    )
-    panel_service = PanelApiService(settings_param)
-
-    subscription_service = SubscriptionService(
-        settings_param, panel_service, bot, i18n_instance
-    )
-    referral_service = ReferralService(
-        settings_param, subscription_service, bot, i18n_instance
-    )
-    promo_code_service = PromoCodeService(
-        settings_param, subscription_service, bot, i18n_instance
-    )
-    stars_service = StarsService(
-        bot, settings_param, i18n_instance, subscription_service, referral_service
-    )
-    cryptopay_service = CryptoPayService(
-        settings_param.CRYPTOPAY_TOKEN,
-        settings_param.CRYPTOPAY_NETWORK,
-        bot,
+    services = build_core_services(
         settings_param,
-        i18n_instance,
-        local_async_session_factory,
-        subscription_service,
-        referral_service,
-    )
-    tribute_service = TributeService(
         bot,
-        settings_param,
-        i18n_instance,
         local_async_session_factory,
-        panel_service,
-        subscription_service,
-        referral_service,
-    )
-    panel_webhook_service = PanelWebhookService(
-        bot,
-        settings_param,
         i18n_instance,
-        local_async_session_factory,
+        actual_bot_username,
     )
-
-    dp["i18n_instance"] = i18n_instance
-    dp["yookassa_service"] = yookassa_service
-    dp["panel_service"] = panel_service
-    dp["subscription_service"] = subscription_service
-    dp["referral_service"] = referral_service
-    dp["promo_code_service"] = promo_code_service
-    dp["stars_service"] = stars_service
-    dp["cryptopay_service"] = cryptopay_service
-    dp["tribute_service"] = tribute_service
-    dp["panel_webhook_service"] = panel_webhook_service
+    for key, service in services.items():
+        dp[key] = service
+    dp["panel_service"] = services["panel_service"]
     dp["async_session_factory"] = local_async_session_factory
 
-    dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory))
-    dp.update.outer_middleware(
-        I18nMiddleware(i18n=i18n_instance, settings=settings_param)
-    )
-    dp.update.outer_middleware(
-        BanCheckMiddleware(settings=settings_param, i18n_instance=i18n_instance)
-    )
-    dp.update.outer_middleware(ActionLoggerMiddleware(settings=settings_param))
-
-    dp.startup.register(on_startup_configured)
-    # Register shutdown callback directly so Dispatcher instance is provided
-    dp.shutdown.register(on_shutdown_configured)
+    # 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)
 
@@ -350,88 +295,10 @@ async def run_bot(settings_param: Settings):
     main_tasks = []
 
     if should_run_aiohttp_server:
-        app = web.Application()
-        app["bot"] = bot
-        app["dp"] = dp
-        app["settings"] = settings_param
-        app["i18n"] = i18n_instance
-        app["async_session_factory"] = local_async_session_factory
-
-        app["yookassa_service"] = yookassa_service
-        app["subscription_service"] = subscription_service
-        app["referral_service"] = referral_service
-        app["panel_service"] = panel_service
-        app["stars_service"] = stars_service
-        app["cryptopay_service"] = cryptopay_service
-        app["tribute_service"] = tribute_service
-        app["panel_webhook_service"] = panel_webhook_service
-
-        setup_application(app, dp, bot=bot)
-
-        if telegram_uses_webhook_mode:
-            telegram_webhook_path = f"/{settings_param.BOT_TOKEN}"
-            if not telegram_webhook_path.startswith("/"):
-                telegram_webhook_path = "/" + telegram_webhook_path
-            app.router.add_post(
-                telegram_webhook_path, SimpleRequestHandler(dispatcher=dp, bot=bot)
-            )
-            logging.info(
-                f"Telegram webhook route configured at: [POST] {telegram_webhook_path} (relative to base URL)"
-            )
-
-        if yk_webhook_base and settings_param.yookassa_webhook_path:
-            yk_path = settings_param.yookassa_webhook_path
-            if not yk_path or not isinstance(yk_path, str):
-                logging.error(
-                    f"YooKassa webhook path is invalid or not configured in settings: {yk_path}. Skipping YooKassa webhook setup."
-                )
-            elif not yk_path.startswith("/"):
-                logging.error(
-                    f"CRITICAL: YooKassa webhook path '{yk_path}' from settings does not start with '/'. Correct settings.py or .env. Skipping YooKassa webhook."
-                )
-            else:
-                app.router.add_post(
-                    yk_path, user_payment_webhook_module.yookassa_webhook_route
-                )
-                logging.info(f"YooKassa webhook route configured at: [POST] {yk_path}")
-
-        tribute_path = settings_param.tribute_webhook_path
-        if tribute_path.startswith("/"):
-            app.router.add_post(tribute_path, tribute_webhook_route)
-            logging.info(f"Tribute webhook route configured at: [POST] {tribute_path}")
-
-        cp_path = settings_param.cryptopay_webhook_path
-        if cp_path.startswith("/"):
-            app.router.add_post(cp_path, cryptopay_webhook_route)
-            logging.info(f"CryptoPay webhook route configured at: [POST] {cp_path}")
-
-        panel_path = settings_param.panel_webhook_path
-        if panel_path.startswith("/"):
-            app.router.add_post(panel_path, panel_webhook_route)
-            logging.info(f"Panel webhook route configured at: [POST] {panel_path}")
-
-        web_app_runner = web.AppRunner(app)
-        await web_app_runner.setup()
-        site = web.TCPSite(
-            web_app_runner,
-            host=settings_param.WEB_SERVER_HOST,
-            port=settings_param.WEB_SERVER_PORT,
-        )
-
         async def web_server_task():
-            await site.start()
-            logging.info(
-                f"AIOHTTP server started on http://{settings_param.WEB_SERVER_HOST}:{settings_param.WEB_SERVER_PORT}"
-            )
-            (
-                await asyncio.Event().wait()
-                if not run_telegram_polling
-                else await asyncio.sleep(31536000)
-            )
+            await build_and_start_web_app(dp, bot, settings_param, local_async_session_factory)
 
-        main_tasks.append(
-            asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")
-        )
+        main_tasks.append(asyncio.create_task(web_server_task(), name="AIOHTTPServerTask"))
 
     if run_telegram_polling:
         logging.info("Starting bot in Telegram Polling mode...")

bot/middlewares/profile_sync.py

@@ -0,0 +1,47 @@
+import logging
+from typing import Callable, Dict, Any, Awaitable, Optional
+
+from aiogram import BaseMiddleware
+from aiogram.types import Update, User as TgUser
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.dal import user_dal
+
+
+class ProfileSyncMiddleware(BaseMiddleware):
+
+    async def __call__(
+        self,
+        handler: Callable[[Update, Dict[str, Any]], Awaitable[Any]],
+        event: Update,
+        data: Dict[str, Any],
+    ) -> Any:
+        session: AsyncSession = data.get("session")
+        tg_user: Optional[TgUser] = data.get("event_from_user")
+
+        if session and tg_user:
+            try:
+                db_user = await user_dal.get_user_by_id(session, tg_user.id)
+                if db_user:
+                    update_payload: Dict[str, Any] = {}
+                    if db_user.username != tg_user.username:
+                        update_payload["username"] = tg_user.username
+                    if db_user.first_name != tg_user.first_name:
+                        update_payload["first_name"] = tg_user.first_name
+                    if db_user.last_name != tg_user.last_name:
+                        update_payload["last_name"] = tg_user.last_name
+
+                    if update_payload:
+                        await user_dal.update_user(session, tg_user.id, update_payload)
+                        logging.info(
+                            f"ProfileSyncMiddleware: Updated user {tg_user.id} profile fields: {list(update_payload.keys())}"
+                        )
+            except Exception as e:
+                logging.error(
+                    f"ProfileSyncMiddleware: Failed to sync profile for user {getattr(tg_user, 'id', 'N/A')}: {e}",
+                    exc_info=True,
+                )
+
+        return await handler(event, data)
+
+

bot/services/crypto_pay_service.py

@@ -142,7 +142,11 @@ class CryptoPayService:
                     provider="cryptopay",
                 )
                 referral_bonus = await referral_service.apply_referral_bonuses_for_payment(
-                    session, user_id, months
+                    session,
+                    user_id,
+                    months,
+                    current_payment_db_id=payment_db_id,
+                    skip_if_active_before_payment=False,
                 )
                 await session.commit()
             except Exception as e:

bot/services/panel_webhook_service.py

@@ -42,12 +42,16 @@ class PanelWebhookService:
             logging.error(f"Failed to send notification to {user_id}: {e}")
 
     async def _handle_expired_subscription(self, session, user_id: int, user_payload: dict, 
-                                         lang: str, markup, first_name: str):
-        """Handle expired subscription - auto-renew tribute users if no cancellation was received"""
+                                         lang: str, markup, first_name: str) -> bool:
+        """Handle expired subscription - auto-renew tribute users if no cancellation was received.
+
+        Returns True if an auto-renewal was performed (and renewal message sent), False otherwise.
+        """
         from db.dal import subscription_dal, payment_dal
         from datetime import datetime, timezone, timedelta
         
         try:
+            auto_renewed = False
             # Check if user has tribute subscriptions that weren't cancelled
             user_subs = await subscription_dal.get_active_subscriptions_for_user(session, user_id)
             
@@ -96,14 +100,17 @@ class PanelWebhookService:
                             reply_markup=markup,
                             parse_mode="HTML"
                         )
+                        auto_renewed = True
                     except Exception as e:
                         logging.error(f"Failed to send auto-renewal notification to user {user_id}: {e}")
                         
             await session.commit()
+            return auto_renewed
             
         except Exception as e:
             logging.error(f"Error handling expired subscription for user {user_id}: {e}")
             await session.rollback()
+            return False
 
     async def handle_event(self, event_name: str, user_payload: dict):
         telegram_id = user_payload.get("telegramId")
@@ -135,10 +142,10 @@ class PanelWebhookService:
                 )
         elif event_name == "user.expired":
             # Check if this is a tribute user that should be auto-renewed (regardless of notification settings)
-            await self._handle_expired_subscription(session, user_id, user_payload, lang, markup, first_name)
+            auto_renewed = await self._handle_expired_subscription(session, user_id, user_payload, lang, markup, first_name)
             
-            # Send notification only if enabled
-            if self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
+            # If auto-renewed via Tribute, suppress expiration notification. Otherwise, send it if enabled.
+            if not auto_renewed and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
                 await self._send_message(
                     user_id,
                     lang,

bot/services/referral_service.py

@@ -7,6 +7,7 @@ from datetime import datetime, timezone, timedelta
 
 from config.settings import Settings
 from db.dal import user_dal
+from db.dal import payment_dal
 from db.models import User
 from db.dal import subscription_dal
 from bot.middlewares.i18n import JsonI18n
@@ -24,8 +25,12 @@ class ReferralService:
         self.i18n = i18n
 
     async def apply_referral_bonuses_for_payment(
-            self, session: AsyncSession, referee_user_id: int,
-            purchased_subscription_months: int) -> Dict[str, Any]:
+            self,
+            session: AsyncSession,
+            referee_user_id: int,
+            purchased_subscription_months: int,
+            current_payment_db_id: Optional[int] = None,
+            skip_if_active_before_payment: bool = True) -> Dict[str, Any]:
 
         referee_final_end_date: Optional[datetime] = None
         referee_bonus_applied_days: Optional[int] = None
@@ -43,6 +48,37 @@ class ReferralService:
                     "referee_new_end_date": None
                 }
 
+            # If configured to apply referral bonuses only once per invited user,
+            # check if the referee already has succeeded payments.
+            if self.settings.REFERRAL_ONE_BONUS_PER_REFEREE:
+                try:
+                    succeeded_count = await payment_dal.count_user_succeeded_payments(
+                        session, referee_user_id, exclude_payment_id=current_payment_db_id
+                    )
+                    if succeeded_count and succeeded_count > 0:
+                        logging.info(
+                            f"Referral bonuses skipped for user {referee_user_id}: already has {succeeded_count} succeeded payments.")
+                        return {
+                            "referee_bonus_applied_days": None,
+                            "referee_new_end_date": None
+                        }
+                except Exception as e_cnt:
+                    logging.error(f"Failed counting succeeded payments for user {referee_user_id}: {e_cnt}")
+
+            # Additionally, do not award referral bonuses if the user was active at payment time
+            # (has an active subscription now). This avoids giving bonuses to already active users.
+            if skip_if_active_before_payment:
+                try:
+                    if await self.subscription_service.has_active_subscription(session, referee_user_id):
+                        logging.info(
+                            f"Referral bonuses skipped for user {referee_user_id}: user currently has an active subscription.")
+                        return {
+                            "referee_bonus_applied_days": None,
+                            "referee_new_end_date": None
+                        }
+                except Exception as e_sub:
+                    logging.error(f"Failed to check active subscription for {referee_user_id}: {e_sub}")
+
             inviter_user_id = referee_user_model.referred_by_id
             inviter_user_model = await user_dal.get_user_by_id(
                 session, inviter_user_id)

bot/services/stars_service.py

@@ -96,7 +96,12 @@ class StarsService:
             return
 
         referral_bonus = await self.referral_service.apply_referral_bonuses_for_payment(
-            session, message.from_user.id, months)
+            session,
+            message.from_user.id,
+            months,
+            current_payment_db_id=payment_db_id,
+            skip_if_active_before_payment=False,
+        )
         await session.commit()
 
         applied_days = referral_bonus.get("referee_bonus_applied_days") if referral_bonus else None

bot/services/subscription_service.py

@@ -37,6 +37,22 @@ class SubscriptionService:
     async def has_had_any_subscription(self, session: AsyncSession, user_id: int) -> bool:
         return await subscription_dal.has_any_subscription_for_user(session, user_id)
 
+    async def has_active_subscription(self, session: AsyncSession, user_id: int) -> bool:
+        """Return True if user currently has an active subscription (end_date in future)."""
+        try:
+            user_record = await user_dal.get_user_by_id(session, user_id)
+            if not user_record or not user_record.panel_user_uuid:
+                return False
+            active_sub = await subscription_dal.get_active_subscription_by_user_id(
+                session, user_id, user_record.panel_user_uuid
+            )
+            if not active_sub or not active_sub.end_date:
+                return False
+            from datetime import datetime, timezone
+            return active_sub.is_active and active_sub.end_date > datetime.now(timezone.utc)
+        except Exception:
+            return False
+
     async def _notify_admin_panel_user_creation_failed(self, user_id: int):
         if not self.bot or not self.i18n or not self.settings.ADMIN_IDS:
             return

bot/services/tribute_service.py

@@ -157,7 +157,12 @@ class TributeService:
                     provider="tribute",
                 )
                 referral_bonus = await referral_service.apply_referral_bonuses_for_payment(
-                    session, int(user_id), months)
+                    session,
+                    int(user_id),
+                    months,
+                    current_payment_db_id=payment_record.payment_id,
+                    skip_if_active_before_payment=False,
+                )
                 await session.commit()
 
                 db_user = await user_dal.get_user_by_id(session, int(user_id))

config/settings.py

@@ -93,6 +93,12 @@ class Settings(BaseSettings):
     REFERRAL_BONUS_DAYS_REFEREE_12_MONTHS: Optional[int] = Field(
         default=15, alias="REFEREE_BONUS_DAYS_12_MONTHS")
 
+    # Referral program configuration
+    REFERRAL_ONE_BONUS_PER_REFEREE: bool = Field(
+        default=True,
+        description="When true, referral bonuses (for inviter and referee) are applied only once per invited user – on their first successful payment."
+    )
+
     PANEL_API_URL: Optional[str] = None
     PANEL_API_KEY: Optional[str] = None
     USER_TRAFFIC_LIMIT_GB: Optional[float] = Field(default=0.0)

db/dal/payment_dal.py

@@ -138,6 +138,23 @@ async def get_all_succeeded_payments_with_user(session: AsyncSession) -> List[Pa
     return result.scalars().all()
 
 
+async def count_user_succeeded_payments(
+    session: AsyncSession, user_id: int, exclude_payment_id: Optional[int] = None
+) -> int:
+    """Count succeeded payments for a specific user.
+
+    If exclude_payment_id is provided, that specific payment will be excluded
+    from the count. Useful to check "prior" payments while processing the
+    current payment in the same transaction.
+    """
+    conditions = [Payment.user_id == user_id, Payment.status == 'succeeded']
+    if exclude_payment_id is not None:
+        conditions.append(Payment.payment_id != exclude_payment_id)
+    stmt = select(func.count(Payment.payment_id)).where(and_(*conditions))
+    result = await session.execute(stmt)
+    return result.scalar() or 0
+
+
 async def update_provider_payment_and_status(
         session: AsyncSession, payment_db_id: int,
         provider_payment_id: str, new_status: str) -> Optional[Payment]:

db/dal/user_dal.py

@@ -93,9 +93,10 @@ async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]:
 
 async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
     """Get comprehensive user statistics including active users, trial users, etc."""
-    from datetime import datetime, timedelta
+    from datetime import datetime, timezone
     
-    now = datetime.utcnow()
+    # Use timezone-aware UTC to avoid naive/aware comparison issues in SQL queries
+    now = datetime.now(timezone.utc)
     today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
     
     # Total users
@@ -106,13 +107,11 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
     banned_users_stmt = select(func.count(User.user_id)).where(User.is_banned == True)
     banned_users = (await session.execute(banned_users_stmt)).scalar() or 0
     
-    # Active users today (users with login activity - for now using registration as proxy)
-    active_today_stmt = select(func.count(User.user_id)).where(
-        User.registration_date >= today_start
-    )
+    # Active users today (proxy: registered today)
+    active_today_stmt = select(func.count(User.user_id)).where(User.registration_date >= today_start)
     active_today = (await session.execute(active_today_stmt)).scalar() or 0
     
-    # Users with active paid subscriptions
+    # Users with active paid subscriptions (non-trial providers only)
     paid_subs_stmt = (
         select(func.count(func.distinct(Subscription.user_id)))
         .join(User, Subscription.user_id == User.user_id)
@@ -156,3 +155,52 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]:
         "inactive_users": max(0, inactive_users),
         "referral_users": referral_users
     }
+
+
+async def get_user_ids_with_active_subscription(session: AsyncSession) -> List[int]:
+    """Return non-banned user IDs who have an active subscription (paid or trial)."""
+    from datetime import datetime, timezone
+    now = datetime.now(timezone.utc)
+
+    stmt = (
+        select(func.distinct(Subscription.user_id))
+        .join(User, Subscription.user_id == User.user_id)
+        .where(
+            and_(
+                User.is_banned == False,
+                Subscription.is_active == True,
+                Subscription.end_date > now,
+            )
+        )
+    )
+    result = await session.execute(stmt)
+    return result.scalars().all()
+
+
+async def get_user_ids_without_active_subscription(session: AsyncSession) -> List[int]:
+    """Return non-banned user IDs who do NOT have any active subscription."""
+    from datetime import datetime, timezone
+    now = datetime.now(timezone.utc)
+
+    # Subquery for users with active subscription
+    active_subs_subq = (
+        select(Subscription.user_id)
+        .where(
+            and_(
+                Subscription.is_active == True,
+                Subscription.end_date > now,
+            )
+        )
+    ).scalar_subquery()
+
+    stmt = (
+        select(User.user_id)
+        .where(
+            and_(
+                User.is_banned == False,
+                ~User.user_id.in_(active_subs_subq),
+            )
+        )
+    )
+    result = await session.execute(stmt)
+    return result.scalars().all()

locales/en.json

@@ -155,6 +155,9 @@
 
   "admin_broadcast_enter_message": "Enter the broadcast message (HTML supported):",
   "admin_broadcast_confirm_prompt": "You are about to send the following message:\n\n{message_preview}\n\nConfirm sending?",
+  "broadcast_target_all_button": "👥 All",
+  "broadcast_target_active_button": "✅ Active",
+  "broadcast_target_inactive_button": "⌛ Inactive",
   "confirm_broadcast_send_button": "✅ Send",
   "cancel_broadcast_button": "❌ Cancel",
   "admin_broadcast_sending_started": "Starting broadcast...",

locales/ru.json

@@ -155,6 +155,9 @@
 
   "admin_broadcast_enter_message": "Введите сообщение для рассылки (HTML поддерживается):",
   "admin_broadcast_confirm_prompt": "Вы собираетесь отправить следующее сообщение:\n\n{message_preview}\n\nПодтверждаете отправку?",
+  "broadcast_target_all_button": "👥 Все",
+  "broadcast_target_active_button": "✅ Активные",
+  "broadcast_target_inactive_button": "⌛ Неактивные",
   "confirm_broadcast_send_button": "✅ Отправить",
   "cancel_broadcast_button": "❌ Отмена",
   "admin_broadcast_sending_started": "Начинаю рассылку...",