Merge pull request #41 from machka-pasla/qrbsms-codex/add-user-notifications-for-expiration-webhooks

Handle panel webhook variants

.env.example

@@ -61,10 +61,11 @@ TRIBUTE_LINK_12_MONTHS=
 # API key for verifying Tribute webhook signatures
 TRIBUTE_API_KEY=
 
-# Subscription Expiration Notifications
+# Subscription Notifications
-SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS=7
+SUBSCRIPTION_NOTIFICATIONS_ENABLED=True
-SUBSCRIPTION_NOTIFICATION_HOUR_UTC=13
+SUBSCRIPTION_NOTIFY_ON_EXPIRE=True
-SUBSCRIPTION_NOTIFICATION_MINUTE_UTC=0
+SUBSCRIPTION_NOTIFY_AFTER_EXPIRE=True
+SUBSCRIPTION_NOTIFY_DAYS_BEFORE=3
 
 # Referral Bonus Days
 REFERRAL_BONUS_DAYS_1_MONTH=3
@@ -79,6 +80,7 @@ REFEREE_BONUS_DAYS_12_MONTHS=15
 # Panel API Configuration
 PANEL_API_URL=http://your_panel_api_url/api
 PANEL_API_KEY=your_panel_api_key
+PANEL_WEBHOOK_SECRET=   # secret used to verify panel webhook signatures
 
 # User traffic limits (applied for all users)
 # 0 means unlimited

README.md

@@ -102,6 +102,7 @@ This Telegram bot is designed to automate the sale and management of subscriptio
     * **Panel API Settings:**
         * `PANEL_API_URL`: Full URL to your Remnawave panel's API (e.g., `http://remnawave:3000/api` or `https://panel.yourdomain.com/api`).
         * `PANEL_API_KEY`: API Key for authenticating with the Remnawave panel.
+        * `PANEL_WEBHOOK_SECRET`: Secret key for verifying webhooks from the Remnawave panel.
     * `USER_INBOUND_UUIDS`: (Optional) Comma-separated list of inbound UUIDs from your panel to assign to users. If empty, `activateAllInbounds: true` (panel default) is used for new users.
     * `USER_TRAFFIC_LIMIT_GB` and `USER_TRAFFIC_STRATEGY`: Default traffic limit in gigabytes (0 for unlimited) and the reset strategy applied when updating users on the panel.
     * `TRIAL_ENABLED`, `TRIAL_DURATION_DAYS`, `TRIAL_TRAFFIC_LIMIT_GB`: Settings for the trial period.

bot/main_bot.py

@@ -16,7 +16,7 @@ from aiogram.client.default import DefaultBotProperties
 from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
 from aiogram.fsm.storage.memory import MemoryStorage
 from aiohttp import web
-from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from bot.services.panel_webhook_service import PanelWebhookService, panel_webhook_route
 from sqlalchemy.orm import sessionmaker
 
 from config.settings import Settings
@@ -31,7 +31,6 @@ from bot.handlers.user import user_router_aggregate
 from bot.handlers.admin import admin_router_aggregate
 from bot.filters.admin_filter import AdminFilter
 
-from bot.services.notification_service import schedule_subscription_notifications
 from bot.services.yookassa_service import YooKassaService
 from bot.services.panel_api_service import PanelApiService
 from bot.services.subscription_service import SubscriptionService
@@ -101,26 +100,6 @@ async def on_startup_configured(dispatcher: Dispatcher):
 
     logging.info("STARTUP: on_startup_configured executing...")
 
-    existing_scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler")
-
-    if existing_scheduler and existing_scheduler.running:
-        logging.warning("STARTUP: Scheduler already running, skipping initialization.")
-    else:
-        scheduler = AsyncIOScheduler(timezone="UTC")
-        try:
-            await schedule_subscription_notifications(
-                bot,
-                settings,
-                i18n_instance,
-                scheduler,
-                panel_service,
-                async_session_factory,
-            )
-            scheduler.start()
-            dispatcher["scheduler"] = scheduler
-            logging.info("STARTUP: APScheduler started.")
-        except Exception as e:
-            logging.error(f"STARTUP: Failed to start APScheduler: {e}", exc_info=True)
 
     telegram_webhook_url_to_set = getattr(settings, "TELEGRAM_WEBHOOK_BASE_URL", None)
     if telegram_webhook_url_to_set:
@@ -213,15 +192,6 @@ async def on_startup_configured(dispatcher: Dispatcher):
 async def on_shutdown_configured(dispatcher: Dispatcher):
     logging.warning("SHUTDOWN: on_shutdown_configured executing...")
 
-    scheduler: Optional[AsyncIOScheduler] = dispatcher.get("scheduler")
-    if scheduler and scheduler.running:
-        try:
-            scheduler.shutdown(wait=False)
-            logging.info("SHUTDOWN: APScheduler shut down.")
-        except Exception as e:
-            logging.error(
-                f"SHUTDOWN: Error shutting down APScheduler: {e}", exc_info=True
-            )
 
     panel_service: Optional[PanelApiService] = dispatcher.get("panel_service")
     if panel_service and hasattr(panel_service, "close_session"):
@@ -304,6 +274,12 @@ async def run_bot(settings_param: Settings):
         subscription_service,
         referral_service,
     )
+    panel_webhook_service = PanelWebhookService(
+        bot,
+        settings_param,
+        i18n_instance,
+        local_async_session_factory,
+    )
 
     dp["i18n_instance"] = i18n_instance
     dp["yookassa_service"] = yookassa_service
@@ -313,6 +289,7 @@ async def run_bot(settings_param: Settings):
     dp["promo_code_service"] = promo_code_service
     dp["stars_service"] = stars_service
     dp["tribute_service"] = tribute_service
+    dp["panel_webhook_service"] = panel_webhook_service
     dp["async_session_factory"] = local_async_session_factory
 
     dp.update.outer_middleware(DBSessionMiddleware(local_async_session_factory))
@@ -367,6 +344,7 @@ async def run_bot(settings_param: Settings):
         app["panel_service"] = panel_service
         app["stars_service"] = stars_service
         app["tribute_service"] = tribute_service
+        app["panel_webhook_service"] = panel_webhook_service
 
         setup_application(app, dp, bot=bot)
 
@@ -402,6 +380,11 @@ async def run_bot(settings_param: Settings):
             app.router.add_post(tribute_path, tribute_webhook_route)
             logging.info(f"Tribute webhook route configured at: [POST] {tribute_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(

bot/services/notification_service.py

@@ -2,138 +2,12 @@ import logging
 import asyncio
 from aiogram import Bot
 from aiogram.utils.text_decorations import html_decoration as hd
-from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from datetime import datetime, timezone
 
 from config.settings import Settings
 
 from sqlalchemy.orm import sessionmaker
 from bot.middlewares.i18n import JsonI18n
-from bot.services.panel_api_service import PanelApiService
-from bot.services.subscription_service import SubscriptionService
-
-
-async def send_expiration_warnings(bot: Bot, settings: Settings,
-                                   i18n: JsonI18n,
-                                   panel_service: PanelApiService,
-                                   async_session_factory: sessionmaker):
-
-    logging.info(
-        f"Scheduler job 'send_expiration_warnings' started at {datetime.now(timezone.utc)} UTC."
-    )
-
-    if async_session_factory is None:
-        logging.error(
-            "NotificationService: AsyncSessionFactory not provided to send_expiration_warnings!"
-        )
-        return
-
-    async with async_session_factory() as session:
-        try:
-
-            sub_service = SubscriptionService(settings, panel_service)
-
-            expiring_subs_details_list = await sub_service.get_subscriptions_ending_soon(
-                session, settings.SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS)
-
-            if not expiring_subs_details_list:
-                logging.info(
-                    "No subscriptions found ending soon for notification.")
-                return
-
-            logging.info(
-                f"Found {len(expiring_subs_details_list)} subscriptions for expiration warning."
-            )
-
-            for sub_details in expiring_subs_details_list:
-                user_id = sub_details['user_id']
-                user_lang = sub_details.get('language_code',
-                                            settings.DEFAULT_LANGUAGE)
-                first_name = hd.quote(sub_details.get('first_name', f"User {user_id}"))
-                end_date_str_for_msg = sub_details.get('end_date_str', "N/A")
-                days_left_display = sub_details.get('days_left', "N/A")
-
-                subscription_actual_end_date_obj: Optional[
-                    datetime] = sub_details.get(
-                        'subscription_end_date_iso_for_update')
-
-                _ = lambda key, **kwargs: i18n.gettext(user_lang, key, **kwargs
-                                                       )
-                message_text = _("subscription_ending_soon_notification",
-                                 user_name=first_name,
-                                 end_date=end_date_str_for_msg,
-                                 days_left=days_left_display)
-                try:
-                    await bot.send_message(user_id, message_text)
-                    logging.info(
-                        f"Sent expiration warning to user {user_id} for subscription ending {end_date_str_for_msg}."
-                    )
-
-                    if subscription_actual_end_date_obj:
-                        await sub_service.update_last_notification_sent(
-                            session, user_id, subscription_actual_end_date_obj)
-                    else:
-                        logging.warning(
-                            f"Could not find exact subscription end_date_obj for user {user_id} to update notification time."
-                        )
-
-                except Exception as e:
-                    logging.error(
-                        f"Failed to send expiration warning or update notification status for user {user_id}: {e}",
-                        exc_info=True)
-
-                await asyncio.sleep(0.1)
-
-            await session.commit()
-            logging.info(
-                "Finished processing expiration warnings. Session committed.")
-
-        except Exception as e_session:
-            logging.error(
-                f"Error during send_expiration_warnings session: {e_session}",
-                exc_info=True)
-            await session.rollback()
-            logging.info(
-                "Session rolled back due to error in send_expiration_warnings."
-            )
-
-
-async def schedule_subscription_notifications(
-        bot: Bot, settings: Settings, i18n: JsonI18n,
-        scheduler: AsyncIOScheduler, panel_service: PanelApiService,
-        async_session_factory: sessionmaker):
-
-    async def job_wrapper():
-
-        try:
-            await send_expiration_warnings(bot, settings, i18n, panel_service,
-                                           async_session_factory)
-        except Exception as e:
-            logging.error(
-                f"Unhandled error in scheduled job 'send_expiration_warnings' (job_wrapper): {e}",
-                exc_info=True)
-
-    try:
-        notification_hour = int(settings.SUBSCRIPTION_NOTIFICATION_HOUR_UTC)
-        notification_minute = int(
-            settings.SUBSCRIPTION_NOTIFICATION_MINUTE_UTC)
-    except (ValueError, TypeError):
-        logging.warning(
-            "SUBSCRIPTION_NOTIFICATION_HOUR_UTC or MINUTE_UTC is invalid in settings. Defaulting to 9:00 UTC."
-        )
-        notification_hour = 9
-        notification_minute = 0
-
-    scheduler.add_job(job_wrapper,
-                      'cron',
-                      hour=notification_hour,
-                      minute=notification_minute,
-                      name="daily_subscription_expiration_warnings_v2",
-                      misfire_grace_time=60 * 15,
-                      replace_existing=True)
-    logging.info(
-        f"Subscription expiration warning job scheduled daily at {notification_hour:02d}:{notification_minute:02d} UTC."
-    )
 
 
 async def notify_admins(bot: Bot, settings: Settings, i18n: JsonI18n,

bot/services/panel_webhook_service.py

@@ -0,0 +1,92 @@
+import json
+import logging
+import hmac
+import hashlib
+from aiohttp import web
+from aiogram import Bot
+from sqlalchemy.orm import sessionmaker
+from typing import Optional
+from config.settings import Settings
+from bot.middlewares.i18n import JsonI18n
+from db.dal import user_dal
+
+EVENT_MAP = {
+    "user.expires_in_72_hours": (3, "subscription_72h_notification"),
+    "user.expires_in_48_hours": (2, "subscription_48h_notification"),
+    "user.expires_in_24_hours": (1, "subscription_24h_notification"),
+}
+
+class PanelWebhookService:
+    def __init__(self, bot: Bot, settings: Settings, i18n: JsonI18n, async_session_factory: sessionmaker):
+        self.bot = bot
+        self.settings = settings
+        self.i18n = i18n
+        self.async_session_factory = async_session_factory
+
+    async def _send_message(self, user_id: int, lang: str, message_key: str, **kwargs):
+        _ = lambda k, **kw: self.i18n.gettext(lang, k, **kw)
+        try:
+            await self.bot.send_message(user_id, _(message_key, **kwargs))
+        except Exception as e:
+            logging.error(f"Failed to send notification to {user_id}: {e}")
+
+    async def handle_event(self, event_name: str, user_payload: dict):
+        telegram_id = user_payload.get("telegramId")
+        if not telegram_id:
+            logging.warning("Panel webhook without telegramId received")
+            return
+        user_id = int(telegram_id)
+
+        if not self.settings.SUBSCRIPTION_NOTIFICATIONS_ENABLED:
+            return
+
+        async with self.async_session_factory() as session:
+            db_user = await user_dal.get_user_by_id(session, user_id)
+            lang = db_user.language_code if db_user and db_user.language_code else self.settings.DEFAULT_LANGUAGE
+            first_name = db_user.first_name or f"User {user_id}" if db_user else f"User {user_id}"
+
+        if event_name in EVENT_MAP:
+            days_left, msg_key = EVENT_MAP[event_name]
+            if days_left == self.settings.SUBSCRIPTION_NOTIFY_DAYS_BEFORE:
+                await self._send_message(
+                    user_id,
+                    lang,
+                    msg_key,
+                    user_name=first_name,
+                    end_date=user_payload.get("expireAt", "")[:10],
+                )
+        elif event_name == "user.expired" and self.settings.SUBSCRIPTION_NOTIFY_ON_EXPIRE:
+            await self._send_message(user_id, lang, "subscription_expired_notification")
+        elif event_name == "user.expired_24_hours_ago" and self.settings.SUBSCRIPTION_NOTIFY_AFTER_EXPIRE:
+            await self._send_message(user_id, lang, "subscription_expired_yesterday_notification")
+
+    async def handle_webhook(self, raw_body: bytes, signature_header: Optional[str]) -> web.Response:
+        if self.settings.PANEL_WEBHOOK_SECRET:
+            if not signature_header:
+                return web.Response(status=403, text="no_signature")
+            expected_sig = hmac.new(
+                self.settings.PANEL_WEBHOOK_SECRET.encode(),
+                raw_body,
+                hashlib.sha256,
+            ).hexdigest()
+            if not hmac.compare_digest(expected_sig, signature_header):
+                return web.Response(status=403, text="invalid_signature")
+
+        try:
+            payload = json.loads(raw_body.decode())
+        except Exception:
+            return web.Response(status=400, text="bad_request")
+
+        event_name = payload.get("name") or payload.get("event")
+        user_data = payload.get("payload", {}).get("user") or payload.get("payload", {})
+        if not event_name:
+            return web.Response(status=200, text="ok_no_event")
+
+        await self.handle_event(event_name, user_data)
+        return web.Response(status=200, text="ok")
+
+async def panel_webhook_route(request: web.Request):
+    service: PanelWebhookService = request.app["panel_webhook_service"]
+    raw = await request.read()
+    signature_header = request.headers.get("X-Remnawave-Signature")
+    return await service.handle_webhook(raw, signature_header)

bot/services/tribute_service.py

@@ -76,7 +76,11 @@ class TributeService:
         event_name = payload.get('name')
         data = payload.get('payload', {})
         user_id = data.get('telegram_user_id')
-        price_val = data.get('price')
+        price_val = (
+            data.get('amount')
+            or data.get('amount_paid')
+            or data.get('price')
+        )
 
         if not user_id or price_val is None:
             return web.Response(status=200, text="ok_missing_fields")
@@ -168,18 +172,6 @@ class TributeService:
                     float(price_rub),
                     currency="RUB",
                 )
-            elif event_name == 'cancelled_subscription':
-                db_user = await user_dal.get_user_by_id(session, user_id)
-                lang = db_user.language_code if db_user and db_user.language_code else settings.DEFAULT_LANGUAGE
-                _ = lambda k, **kw: i18n.gettext(lang, k, **kw)
-                try:
-                    await bot.send_message(user_id, _("subscription_cancelled_notification"))
-                except Exception as e:
-                    logging.warning(
-                        f"Failed to notify user {user_id} about cancellation: {e}")
-                await subscription_dal.set_skip_notifications_for_provider(
-                    session, user_id, 'tribute', False)
-                await session.commit()
             else:
                 await session.commit()
         return web.Response(status=200, text="ok")

config/settings.py

@@ -61,10 +61,12 @@ class Settings(BaseSettings):
     TRIBUTE_LINK_6_MONTHS: Optional[str] = Field(default=None)
     TRIBUTE_LINK_12_MONTHS: Optional[str] = Field(default=None)
     TRIBUTE_API_KEY: Optional[str] = Field(default=None)
+    PANEL_WEBHOOK_SECRET: Optional[str] = Field(default=None)
 
-    SUBSCRIPTION_EXPIRATION_NOTIFICATION_DAYS: int = Field(default=7)
+    SUBSCRIPTION_NOTIFICATIONS_ENABLED: bool = Field(default=True)
-    SUBSCRIPTION_NOTIFICATION_HOUR_UTC: int = Field(default=9)
+    SUBSCRIPTION_NOTIFY_ON_EXPIRE: bool = Field(default=True)
-    SUBSCRIPTION_NOTIFICATION_MINUTE_UTC: int = Field(default=0)
+    SUBSCRIPTION_NOTIFY_AFTER_EXPIRE: bool = Field(default=True)
+    SUBSCRIPTION_NOTIFY_DAYS_BEFORE: int = Field(default=3)
 
     REFERRAL_BONUS_DAYS_INVITER_1_MONTH: Optional[int] = Field(
         default=3, alias="REFERRAL_BONUS_DAYS_1_MONTH")
@@ -184,6 +186,18 @@ class Settings(BaseSettings):
             return f"{self.YOOKASSA_WEBHOOK_BASE_URL.rstrip('/')}{self.tribute_webhook_path}"
         return None
 
+    @computed_field
+    @property
+    def panel_webhook_path(self) -> str:
+        return "/webhook/panel"
+
+    @computed_field
+    @property
+    def panel_full_webhook_url(self) -> Optional[str]:
+        if self.YOOKASSA_WEBHOOK_BASE_URL:
+            return f"{self.YOOKASSA_WEBHOOK_BASE_URL.rstrip('/')}{self.panel_webhook_path}"
+        return None
+
     @computed_field
     @property
     def subscription_options(self) -> Dict[int, float]:

locales/en.json

@@ -211,8 +211,11 @@
   "error_displaying_statistics": "Error displaying statistics.",
   "stub_page_display": "Page",
 
-  "subscription_ending_soon_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends on {end_date} (in {days_left} days).\n\nTo avoid interruption, please renew it in the main menu.",
+  "subscription_72h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 72 hours on {end_date}.\n\nPlease renew it in the main menu.",
-  "subscription_cancelled_notification": "Your recurring subscription was cancelled. You will keep access until the paid period ends.",
+  "subscription_48h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 48 hours on {end_date}.\n\nPlease renew it in the main menu.",
+  "subscription_24h_notification": "👋 Hi, {user_name}!\n\n⏳ Your VPN subscription ends in 24 hours on {end_date}.\n\nPlease renew it in the main menu.",
+  "subscription_expired_notification": "Your VPN subscription has expired.",
+  "subscription_expired_yesterday_notification": "Your VPN subscription expired 24 hours ago.",
 
   "admin_new_trial_notification": "\ud83c\udf21 User {user_id} activated a free trial until {end_date}.",
   "admin_new_payment_notification": "\ud83d\udcb3 Payment received from user {user_id}: {months} mo. for {amount} {currency}.",

locales/ru.json

@@ -211,8 +211,11 @@
   "error_displaying_statistics": "Ошибка отображения статистики.",
   "stub_page_display": "Страница",
 
-  "subscription_ending_soon_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает {end_date} (через {days_left} дн.).\n\nЧтобы не потерять доступ, пожалуйста, продлите ее заранее в главном меню бота.",
+  "subscription_72h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 72 часа ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.",
-  "subscription_cancelled_notification": "Ваша подписка отменена. Доступ сохранится до конца оплаченного периода.",
+  "subscription_48h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 48 часов ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.",
+  "subscription_24h_notification": "👋 Привет, {user_name}!\n\n⏳ Ваша подписка на VPN истекает через 24 часа ({end_date}).\n\nПожалуйста, продлите её в главном меню бота.",
+  "subscription_expired_notification": "Срок вашей подписки истек.",
+  "subscription_expired_yesterday_notification": "Ваша подписка истекла сутки назад.",
 
   "admin_new_trial_notification": "\ud83c\udf21 Пользователь {user_id} активировал пробный период до {end_date}.",
   "admin_new_payment_notification": "\ud83d\udcb3 Получен платеж от пользователя {user_id}: {months} мес. за {amount} {currency}.",

requirements.txt

@@ -4,8 +4,7 @@ aiohttp==3.10.11
 pydantic==2.7.1
 yookassa==3.5.0
 pycountry==23.12.11
-apscheduler==3.10.4
 pydantic_settings
 sqlalchemy[asyncio]==2.0.29
 asyncpg==0.29.0
-alembic==1.13.1
+alembic==1.13.1