From 001e54cfe2bb1c7c4c5d60743204e629005c313a Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Fri, 29 May 2026 21:46:42 +0300 Subject: [PATCH] fix: preserve subscriptions on panel lookup failures --- backend/bot/services/panel_api_service.py | 104 ++++++++++- .../subscription_lifecycle_notifications.py | 47 +++++ .../subscription_service_impl/lifecycle.py | 164 +++++++++++++++++- tests/test_panel_api_service_logging.py | 53 ++++++ ...st_subscription_lifecycle_notifications.py | 60 +++++++ tests/test_subscription_service_behavior.py | 143 ++++++++++++++- 6 files changed, 564 insertions(+), 7 deletions(-) diff --git a/backend/bot/services/panel_api_service.py b/backend/bot/services/panel_api_service.py index c30a060..4ff5c43 100644 --- a/backend/bot/services/panel_api_service.py +++ b/backend/bot/services/panel_api_service.py @@ -328,12 +328,112 @@ class PanelApiService: async def _get_user_by_uuid_uncached( self, user_uuid: str, log_response: bool = False ) -> Optional[Dict[str, Any]]: + lookup = await self.get_user_by_uuid_lookup(user_uuid, log_response=log_response) + if lookup.get("ok") and isinstance(lookup.get("user"), dict): + return lookup["user"] + return None + + @staticmethod + def _panel_response_details(response_data: Optional[Dict[str, Any]]) -> Dict[str, Any]: + if not isinstance(response_data, dict): + return {} + details = response_data.get("details") + return details if isinstance(details, dict) else {} + + @classmethod + def _panel_response_error_code(cls, response_data: Optional[Dict[str, Any]]) -> Optional[str]: + if not isinstance(response_data, dict): + return None + details = cls._panel_response_details(response_data) + error_code = ( + response_data.get("errorCode") + or response_data.get("code") + or details.get("errorCode") + or details.get("code") + ) + return str(error_code) if error_code else None + + @classmethod + def _panel_response_message(cls, response_data: Optional[Dict[str, Any]]) -> Optional[str]: + if not isinstance(response_data, dict): + return None + details = cls._panel_response_details(response_data) + message = ( + response_data.get("message") + or details.get("message") + or details.get("error") + or details.get("raw_response_text") + ) + if message is None: + return None + message = str(message).replace("\n", " ").strip() + return message[:500] if message else None + + @classmethod + def _is_user_not_found_response(cls, response_data: Optional[Dict[str, Any]]) -> bool: + if not isinstance(response_data, dict): + return False + status_code = response_data.get("status_code") + error_code = cls._panel_response_error_code(response_data) + if error_code in {"A040", "A062", "USER_NOT_FOUND", "NOT_FOUND"}: + return True + return status_code == 404 + + @classmethod + def _describe_user_lookup_failure( + cls, + response_data: Optional[Dict[str, Any]], + *, + not_found: bool, + ) -> str: + if not isinstance(response_data, dict): + return "classification=panel_lookup_failed response=empty" + + classification = "confirmed_not_found" if not_found else "panel_lookup_failed" + parts = [f"classification={classification}"] + status_code = response_data.get("status_code") + if status_code is not None: + parts.append(f"status_code={status_code}") + error_code = cls._panel_response_error_code(response_data) + if error_code: + parts.append(f"error_code={error_code}") + message = cls._panel_response_message(response_data) + if message: + parts.append(f"message={message}") + return " ".join(parts) + + async def get_user_by_uuid_lookup( + self, user_uuid: str, log_response: bool = False + ) -> Dict[str, Any]: + """Fetch a panel user and preserve whether a miss was confirmed. + + ``get_user_by_uuid`` historically returned ``None`` both for a real + 404/not-found and for transient panel/API failures. Callers that may + mutate local state need this richer result to avoid treating an outage + as a deleted panel user. + """ endpoint = f"/users/{user_uuid}" full_response = await self._request("GET", endpoint, log_full_response=log_response) if full_response and not full_response.get("error") and "response" in full_response: - return full_response.get("response") + return { + "ok": True, + "user": full_response.get("response"), + "not_found": False, + "failure_reason": None, + "response": full_response, + } - return None + not_found = self._is_user_not_found_response(full_response) + return { + "ok": False, + "user": None, + "not_found": not_found, + "failure_reason": self._describe_user_lookup_failure( + full_response, + not_found=not_found, + ), + "response": full_response, + } async def get_user( self, diff --git a/backend/bot/services/subscription_lifecycle_notifications.py b/backend/bot/services/subscription_lifecycle_notifications.py index 13385db..f590628 100644 --- a/backend/bot/services/subscription_lifecycle_notifications.py +++ b/backend/bot/services/subscription_lifecycle_notifications.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone from typing import Optional from aiogram import Bot +from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError from aiogram.types import InlineKeyboardMarkup from sqlalchemy.ext.asyncio import AsyncSession @@ -125,6 +126,35 @@ class SubscriptionLifecycleNotificationService: return False try: await self.bot.send_message(chat_id, message_text, reply_markup=markup) + except (TelegramBadRequest, TelegramForbiddenError) as exc: + if self._is_terminal_telegram_delivery_error(exc): + logging.warning( + "Skipping subscription notification %s for unreachable Telegram user %s: %s", + stage.key, + chat_id, + exc, + ) + try: + await subscription_dal.record_subscription_notification( + session, + sub.subscription_id, + self._channel_key(stage.key, "telegram"), + sent_at=sent_at, + ) + except Exception: + logging.exception( + "Failed to record skipped subscription notification %s " + "for Telegram user %s", + stage.key, + chat_id, + ) + return False + logging.exception( + "Failed to send subscription notification %s to Telegram user %s", + stage.key, + chat_id, + ) + return False except Exception: logging.exception( "Failed to send subscription notification %s to Telegram user %s", @@ -230,6 +260,23 @@ class SubscriptionLifecycleNotificationService: return chat_id return None + @staticmethod + def _is_terminal_telegram_delivery_error( + exc: TelegramBadRequest | TelegramForbiddenError, + ) -> bool: + if isinstance(exc, TelegramForbiddenError): + return True + message = str(exc).lower() + return any( + token in message + for token in ( + "chat not found", + "bot was blocked", + "bot can't initiate conversation", + "user is deactivated", + ) + ) + @staticmethod def _as_utc(value: Optional[datetime]) -> Optional[datetime]: if value is None: diff --git a/backend/bot/services/subscription_service_impl/lifecycle.py b/backend/bot/services/subscription_service_impl/lifecycle.py index f5e05a1..3968a2c 100644 --- a/backend/bot/services/subscription_service_impl/lifecycle.py +++ b/backend/bot/services/subscription_service_impl/lifecycle.py @@ -3,6 +3,142 @@ from ._runtime import * # noqa: F403,F405 class SubscriptionLifecycleMixin: + async def _lookup_panel_user_for_subscription_details( + self, + panel_user_uuid: str, + ) -> Tuple[Optional[Dict[str, Any]], bool, str]: + lookup_method = getattr(self.panel_service, "get_user_by_uuid_lookup", None) + if callable(lookup_method): + try: + lookup = await lookup_method(panel_user_uuid, log_response=False) + except TypeError: + try: + lookup = await lookup_method(panel_user_uuid) + except Exception as exc: + logging.exception( + "Failed to fetch panel user %s for subscription details", + panel_user_uuid, + ) + return None, False, self._panel_lookup_exception_reason(exc) + except Exception as exc: + logging.exception( + "Failed to fetch panel user %s for subscription details", + panel_user_uuid, + ) + return None, False, self._panel_lookup_exception_reason(exc) + + if isinstance(lookup, dict) and ("ok" in lookup or "not_found" in lookup): + user = lookup.get("user") + if lookup.get("ok") and isinstance(user, dict): + return user, False, "" + reason = str(lookup.get("failure_reason") or "classification=panel_lookup_failed") + return None, bool(lookup.get("not_found")), reason + + try: + panel_user = await self.panel_service.get_user_by_uuid(panel_user_uuid) + except Exception as exc: + logging.exception( + "Failed to fetch panel user %s for subscription details", + panel_user_uuid, + ) + return None, False, self._panel_lookup_exception_reason(exc) + return (panel_user if isinstance(panel_user, dict) else None), False, "" + + @staticmethod + def _panel_lookup_exception_reason(exc: Exception) -> str: + message = str(exc).replace("\n", " ").strip() + if len(message) > 300: + message = f"{message[:300]}..." + reason = f"classification=panel_lookup_failed exception={type(exc).__name__}" + if message: + reason = f"{reason} message={message}" + return reason + + async def _local_active_subscription_details_fallback( + self, + db_user: User, + local_active_sub: Subscription, + ) -> Dict[str, Any]: + panel_sub_id = str(local_active_sub.panel_subscription_uuid or "").strip() + config_link_raw = ( + await self.panel_service.get_subscription_link(panel_sub_id) if panel_sub_id else None + ) + display_link, connect_button_url = await prepare_config_links( + self.settings, + config_link_raw, + ) + tariff = None + if local_active_sub.tariff_key and self._tariffs_config(): + try: + tariff = self._resolve_tariff(local_active_sub.tariff_key) + except Exception: + tariff = None + language = db_user.language_code or self.settings.DEFAULT_LANGUAGE + premium_access = ( + await self.premium_access_for_tariff(tariff) + if tariff + else {"squad_uuids": [], "squad_labels": [], "node_labels": []} + ) + premium_baseline = int(local_active_sub.premium_baseline_bytes or 0) + premium_topup_balance = int(local_active_sub.premium_topup_balance_bytes or 0) + premium_topup_used = int(getattr(local_active_sub, "premium_topup_used_bytes", 0) or 0) + premium_bonus_bytes = int(getattr(local_active_sub, "premium_bonus_bytes", 0) or 0) + return { + "user_id": db_user.panel_user_uuid, + "panel_subscription_uuid": local_active_sub.panel_subscription_uuid, + "panel_short_uuid": local_active_sub.panel_subscription_uuid, + "end_date": local_active_sub.end_date, + "status_from_panel": local_active_sub.status_from_panel or "LOCAL_CACHE", + "config_link": display_link, + "connect_button_url": connect_button_url, + "traffic_limit_bytes": local_active_sub.traffic_limit_bytes, + "traffic_used_bytes": local_active_sub.traffic_used_bytes, + "traffic_limit_strategy": "", + "tariff_key": local_active_sub.tariff_key, + "tariff_name": tariff.name(language) if tariff else None, + "tariff_description": tariff.description(language) if tariff else None, + "premium_title": tariff.premium_name(language) if tariff else None, + "billing_model": tariff.billing_model + if tariff + else ("traffic" if getattr(self.settings, "traffic_sale_mode", False) else "period"), + "tier_baseline_bytes": local_active_sub.tier_baseline_bytes, + "topup_balance_bytes": local_active_sub.topup_balance_bytes, + "regular_bonus_bytes": int(getattr(local_active_sub, "regular_bonus_bytes", 0) or 0), + "regular_unlimited_override": bool( + getattr(local_active_sub, "regular_unlimited_override", False) + ), + "premium_baseline_bytes": premium_baseline, + "premium_topup_balance_bytes": premium_topup_balance, + "premium_topup_used_bytes": premium_topup_used, + "premium_used_bytes": local_active_sub.premium_used_bytes, + "premium_bonus_bytes": premium_bonus_bytes, + "premium_unlimited_override": bool( + getattr(local_active_sub, "premium_unlimited_override", False) + ), + "premium_limit_bytes": self._premium_effective_limit_bytes( + premium_baseline, + premium_topup_balance, + premium_topup_used, + premium_bonus_bytes, + ), + "premium_is_limited": bool(local_active_sub.premium_is_limited), + "premium_period_start_at": getattr(local_active_sub, "premium_period_start_at", None), + "premium_squad_labels": premium_access.get("squad_labels") or [], + "premium_node_labels": premium_access.get("node_labels") or [], + "period_start_at": local_active_sub.period_start_at, + "is_throttled": bool(local_active_sub.is_throttled), + "base_hwid_device_limit": local_active_sub.hwid_device_limit, + "extra_hwid_devices": int(local_active_sub.extra_hwid_devices or 0), + "extra_hwid_devices_valid_until": None, + "extra_hwid_devices_next_valid_from": None, + "user_bot_username": db_user.username, + "is_panel_data": False, + "max_devices": self._effective_hwid_limit( + local_active_sub.hwid_device_limit, + int(local_active_sub.extra_hwid_devices or 0), + ), + } + async def switch_tariff_without_payment( self, session: AsyncSession, @@ -663,14 +799,34 @@ class SubscriptionLifecycleMixin: local_active_sub = await subscription_dal.get_active_subscription_by_user_id( session, user_id, panel_user_uuid ) - panel_user_data = await self.panel_service.get_user_by_uuid(panel_user_uuid) + panel_user_data, panel_user_confirmed_absent, panel_lookup_failure_reason = ( + await self._lookup_panel_user_for_subscription_details(panel_user_uuid) + ) if not panel_user_data: + if panel_user_confirmed_absent: + logging.warning( + "Panel user %s confirmed absent on panel for user %s. " + "Clearing local linkage. reason=%s", + panel_user_uuid, + user_id, + panel_lookup_failure_reason, + ) + await subscription_dal.deactivate_all_user_subscriptions(session, user_id) + await user_dal.update_user(session, user_id, {"panel_user_uuid": None}) + return None logging.warning( - f"Panel user {panel_user_uuid} not found on panel for user {user_id}. Clearing local linkage." # noqa: E501 + "Panel user %s lookup failed for user %s; treating it as a panel access/API " + "problem and preserving local linkage/subscription. reason=%s", + panel_user_uuid, + user_id, + panel_lookup_failure_reason, ) - await subscription_dal.deactivate_all_user_subscriptions(session, user_id) - await user_dal.update_user(session, user_id, {"panel_user_uuid": None}) + if local_active_sub: + return await self._local_active_subscription_details_fallback( + db_user, + local_active_sub, + ) return None panel_lifetime_used = self._extract_lifetime_used_traffic(panel_user_data) diff --git a/tests/test_panel_api_service_logging.py b/tests/test_panel_api_service_logging.py index ee4decc..4691bb5 100644 --- a/tests/test_panel_api_service_logging.py +++ b/tests/test_panel_api_service_logging.py @@ -77,6 +77,59 @@ class PanelApiServiceLoggingTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(service._request.await_count, 3) + async def test_get_user_by_uuid_lookup_returns_success_payload(self): + service = self._make_service() + service._request = AsyncMock(return_value={"response": {"uuid": "user-uuid"}}) + + result = await service.get_user_by_uuid_lookup("user-uuid") + + self.assertTrue(result["ok"]) + self.assertFalse(result["not_found"]) + self.assertIsNone(result["failure_reason"]) + self.assertEqual(result["user"], {"uuid": "user-uuid"}) + service._request.assert_awaited_once_with( + "GET", + "/users/user-uuid", + log_full_response=False, + ) + + async def test_get_user_by_uuid_lookup_keeps_transient_errors_separate_from_not_found(self): + service = self._make_service() + transient_response = { + "error": True, + "status_code": -1, + "message": "Connection error", + } + service._request = AsyncMock(return_value=transient_response) + + result = await service.get_user_by_uuid_lookup("user-uuid") + + self.assertFalse(result["ok"]) + self.assertFalse(result["not_found"]) + self.assertIsNone(result["user"]) + self.assertIn("classification=panel_lookup_failed", result["failure_reason"]) + self.assertIn("status_code=-1", result["failure_reason"]) + self.assertIn("message=Connection error", result["failure_reason"]) + self.assertEqual(result["response"], transient_response) + + async def test_get_user_by_uuid_lookup_marks_confirmed_not_found(self): + service = self._make_service() + cases = [ + {"error": True, "status_code": 404}, + {"error": True, "status_code": 400, "details": {"errorCode": "A062"}}, + ] + + for response in cases: + with self.subTest(response=response): + service._request = AsyncMock(return_value=response) + + result = await service.get_user_by_uuid_lookup("missing-user") + + self.assertFalse(result["ok"]) + self.assertTrue(result["not_found"]) + self.assertIsNone(result["user"]) + self.assertIn("classification=confirmed_not_found", result["failure_reason"]) + async def test_get_user_devices_uses_short_ttl_cache_and_disconnect_invalidates(self): service = self._make_service() service._request = AsyncMock(return_value={"response": [{"hwid": "device-1"}]}) diff --git a/tests/test_subscription_lifecycle_notifications.py b/tests/test_subscription_lifecycle_notifications.py index a0c6295..bc60f0f 100644 --- a/tests/test_subscription_lifecycle_notifications.py +++ b/tests/test_subscription_lifecycle_notifications.py @@ -2,6 +2,9 @@ import asyncio from datetime import datetime, timezone from types import SimpleNamespace +from aiogram.exceptions import TelegramBadRequest +from aiogram.methods import SendMessage + from bot.services import subscription_lifecycle_notifications as lifecycle from bot.services.subscription_lifecycle_notifications import ( SubscriptionLifecycleNotificationService, @@ -41,6 +44,18 @@ class FakeBot: ) +class ChatNotFoundBot: + def __init__(self): + self.calls = [] + + async def send_message(self, chat_id, text, reply_markup=None): + self.calls.append((chat_id, text, reply_markup)) + raise TelegramBadRequest( + method=SendMessage(chat_id=chat_id, text=text), + message="Bad Request: chat not found", + ) + + class FakeEmailService: def __init__(self): self.messages = [] @@ -172,3 +187,48 @@ def test_legacy_stage_key_suppresses_only_telegram(monkeypatch): assert bot.messages == [] assert email_service.messages[0]["email"] == "user@example.test" assert recorded == ["before_3d", "before_3d:email"] + + +def test_terminal_telegram_failure_is_recorded_to_avoid_retry_spam(monkeypatch): + recorded = [] + + async def fake_has(session, subscription_id, notification_key): + return notification_key in recorded + + async def fake_record(session, subscription_id, notification_key, *, sent_at=None): + recorded.append(notification_key) + + monkeypatch.setattr(lifecycle.subscription_dal, "has_subscription_notification", fake_has) + monkeypatch.setattr(lifecycle.subscription_dal, "record_subscription_notification", fake_record) + + bot = ChatNotFoundBot() + settings = _settings() + settings.email_auth_configured = False + service = SubscriptionLifecycleNotificationService( + settings, + bot, + FakeI18n(), + ) + user = _user() + user.telegram_id = 777 + user.email = "" + + async def run(): + return await service.send_stage( + object(), + _subscription(), + SubscriptionNotificationStage( + key="before_3d", + message_key="subscription_72h_notification", + days_left=3, + ), + user=user, + telegram_markup="markup", + ) + + delivery = asyncio.run(run()) + + assert delivery.telegram_sent is False + assert delivery.email_sent is False + assert bot.calls[0][0] == 777 + assert recorded == ["before_3d:telegram"] diff --git a/tests/test_subscription_service_behavior.py b/tests/test_subscription_service_behavior.py index 3820186..592b901 100644 --- a/tests/test_subscription_service_behavior.py +++ b/tests/test_subscription_service_behavior.py @@ -1,7 +1,7 @@ import json import tempfile import unittest -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -393,6 +393,147 @@ class SubscriptionServiceActivationDispatchTests(unittest.IsolatedAsyncioTestCas self.assertEqual(kwargs["payment_db_id"], 12) +class SubscriptionServiceActiveDetailsTests(unittest.IsolatedAsyncioTestCase): + def _local_active_sub(self) -> SimpleNamespace: + return SimpleNamespace( + subscription_id=7, + user_id=42, + panel_user_uuid="panel-user", + panel_subscription_uuid="short-uuid", + end_date=datetime.now(timezone.utc) + timedelta(days=10), + is_active=True, + status_from_panel="ACTIVE", + traffic_limit_bytes=1000, + traffic_used_bytes=100, + tariff_key=None, + tier_baseline_bytes=None, + topup_balance_bytes=0, + regular_bonus_bytes=0, + regular_unlimited_override=False, + premium_baseline_bytes=0, + premium_topup_balance_bytes=0, + premium_topup_used_bytes=0, + premium_used_bytes=0, + premium_bonus_bytes=0, + premium_unlimited_override=False, + premium_is_limited=False, + premium_period_start_at=None, + period_start_at=None, + is_throttled=False, + hwid_device_limit=None, + extra_hwid_devices=0, + ) + + async def test_get_active_subscription_details_preserves_local_subscription_on_panel_error( + self, + ): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(_tariffs_config_payload(), tmpdir) + service = _make_service(settings) + service.panel_service.get_user_by_uuid_lookup = AsyncMock( + return_value={ + "ok": False, + "user": None, + "not_found": False, + "failure_reason": "classification=panel_lookup_failed status_code=-1 " + "message=Connection error", + "response": {"error": True, "status_code": -1}, + } + ) + service.panel_service.get_subscription_link = AsyncMock( + return_value="https://panel.example.test/sub/short-uuid" + ) + session = AsyncMock() + db_user = SimpleNamespace( + user_id=42, + panel_user_uuid="panel-user", + username="alice", + language_code="en", + ) + local_sub = self._local_active_sub() + + with ( + patch( + "bot.services.subscription_service_impl.lifecycle.user_dal.get_user_by_id", + AsyncMock(return_value=db_user), + ), + patch( + "bot.services.subscription_service_impl.lifecycle.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=local_sub), + ), + patch( + "bot.services.subscription_service_impl.lifecycle.subscription_dal.deactivate_all_user_subscriptions", + AsyncMock(), + ) as deactivate_all, + patch( + "bot.services.subscription_service_impl.lifecycle.user_dal.update_user", + AsyncMock(), + ) as update_user, + patch( + "bot.services.subscription_service_impl.lifecycle.logging.warning", + ) as warning_log, + ): + result = await service.get_active_subscription_details(session, user_id=42) + + self.assertIsNotNone(result) + self.assertFalse(result["is_panel_data"]) + self.assertEqual(result["end_date"], local_sub.end_date) + self.assertEqual(result["config_link"], "https://panel.example.test/sub/short-uuid") + deactivate_all.assert_not_awaited() + update_user.assert_not_awaited() + warning_text = " ".join(str(call) for call in warning_log.call_args_list) + self.assertIn("panel access/API problem", warning_text) + self.assertIn("status_code=-1", warning_text) + self.assertIn("Connection error", warning_text) + + async def test_get_active_subscription_details_clears_link_only_when_panel_confirms_absent( + self, + ): + with tempfile.TemporaryDirectory() as tmpdir: + settings = _make_settings(_tariffs_config_payload(), tmpdir) + service = _make_service(settings) + service.panel_service.get_user_by_uuid_lookup = AsyncMock( + return_value={ + "ok": False, + "user": None, + "not_found": True, + "failure_reason": "classification=confirmed_not_found status_code=404", + "response": {"error": True, "status_code": 404}, + } + ) + session = AsyncMock() + db_user = SimpleNamespace( + user_id=42, + panel_user_uuid="panel-user", + username="alice", + language_code="en", + ) + + with ( + patch( + "bot.services.subscription_service_impl.lifecycle.user_dal.get_user_by_id", + AsyncMock(return_value=db_user), + ), + patch( + "bot.services.subscription_service_impl.lifecycle.subscription_dal.get_active_subscription_by_user_id", + AsyncMock(return_value=self._local_active_sub()), + ), + patch( + "bot.services.subscription_service_impl.lifecycle.subscription_dal.deactivate_all_user_subscriptions", + AsyncMock(), + ) as deactivate_all, + patch( + "bot.services.subscription_service_impl.lifecycle.user_dal.update_user", + AsyncMock(), + ) as update_user, + ): + result = await service.get_active_subscription_details(session, user_id=42) + + self.assertIsNone(result) + deactivate_all.assert_awaited_once_with(session, 42) + update_user.assert_awaited_once_with(session, 42, {"panel_user_uuid": None}) + + class SubscriptionDalPayloadTests(unittest.TestCase): def test_subscription_model_payload_drops_panel_only_keys(self): payload = _subscription_model_payload(