From a7f298743de1f1bc9bffe32e8266554795e53857 Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Wed, 20 May 2026 23:04:29 +0300 Subject: [PATCH] refactor: improve startup and sync performance --- .env.example | 7 +- backend/bot/app/web/admin_api_impl/stats.py | 125 ++++++--- backend/bot/handlers/admin/sync_admin.py | 277 ++++++++++++++++---- backend/bot/middlewares/profile_sync.py | 46 ++++ backend/bot/services/panel_api_service.py | 73 ++++++ backend/bot/services/tariff_worker.py | 2 +- backend/config/settings.py | 7 +- scripts/perf_benchmarks.py | 269 ++++++++++++++++++- tests/test_admin_panel_stats_cache.py | 59 +++++ tests/test_admin_sync_performance.py | 71 +++++ tests/test_panel_api_service_logging.py | 60 +++++ tests/test_profile_sync_middleware.py | 74 ++++++ 12 files changed, 973 insertions(+), 97 deletions(-) create mode 100644 tests/test_admin_panel_stats_cache.py create mode 100644 tests/test_admin_sync_performance.py create mode 100644 tests/test_profile_sync_middleware.py diff --git a/.env.example b/.env.example index 93ebc23..82465e3 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,11 @@ REDIS_URL=redis://redis:6379/0 # REDIS_KEY_PREFIX=remnawave-tg-shop # Prefix for Redis keys WEBAPP_ME_CACHE_TTL_SECONDS=15 # Short TTL for /api/me payload cache WEBAPP_DEVICES_CACHE_TTL_SECONDS=5 # Short TTL for /api/devices payload cache +PANEL_USER_CACHE_TTL_SECONDS=5 # Short TTL for Remnawave /users/{uuid} cache +PANEL_DEVICES_CACHE_TTL_SECONDS=5 # Short TTL for Remnawave user devices cache +PANEL_ALL_USERS_CACHE_TTL_SECONDS=5 # Short TTL for concurrent Remnawave full user scans +ADMIN_PANEL_STATS_CACHE_TTL_SECONDS=15 # Short TTL for admin panel stats fetched from Remnawave +PROFILE_SYNC_CACHE_TTL_SECONDS=900 # Minimum seconds between Telegram profile sync checks per user WEBAPP_RATE_LIMIT_TTL_SECONDS=60 # Redis rate-limit window WEBAPP_RATE_LIMIT_MAX_REQUESTS=30 # Requests per window/action/user/IP WEBHOOK_QUEUE_NAME=webhook-events # Redis queue for heavy webhook processing @@ -24,7 +29,7 @@ WEBHOOK_QUEUE_CONCURRENCY=4 # WORKER_PANEL_SYNC_INTERVAL_SECONDS=900 # Worker panel sync interval TARIFF_WORKER_LOCK_TTL_SECONDS=240 # Redis lock TTL for tariff tick TARIFF_WORKER_TICK_SECONDS=300 # Tariff worker tick interval -TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD=200 # Active subs threshold to bulk-fetch panel users +TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD=50 # Active subs threshold to bulk-fetch panel users # Localization and Display DEFAULT_LANGUAGE="ru" # or "en" diff --git a/backend/bot/app/web/admin_api_impl/stats.py b/backend/bot/app/web/admin_api_impl/stats.py index 289f82f..703c033 100644 --- a/backend/bot/app/web/admin_api_impl/stats.py +++ b/backend/bot/app/web/admin_api_impl/stats.py @@ -1,5 +1,10 @@ # ruff: noqa: F401,F403,F405,I001 +import asyncio + from ._runtime import * # noqa: F403,F405 +from bot.utils.ttl_cache import AsyncTTLCache + +_ADMIN_PANEL_STATS_CACHES: Dict[tuple[int, int], AsyncTTLCache] = {} async def admin_me_route(request: web.Request) -> web.Response: @@ -36,47 +41,7 @@ async def admin_stats_route(request: web.Request) -> web.Response: panel_service = request.app.get("panel_service") if panel_service is not None: - try: - system = await panel_service.get_system_stats() - bandwidth = await panel_service.get_bandwidth_stats() - panel_body: Dict[str, Any] = { - "system": system or {}, - "bandwidth": bandwidth or {}, - } - try: - nodes = await panel_service.get_nodes_statistics() - panel_body["nodes"] = nodes or {} - except Exception as exc_nodes: # pragma: no cover - optional endpoint - logger.debug("Panel nodes stats unavailable: %s", exc_nodes) - panel_body["nodes"] = {} - try: - today = datetime.now(timezone.utc).date() - start_d = today - timedelta(days=7) - nodes_bw = await panel_service.get_nodes_bandwidth_usage( - start=start_d.isoformat(), - end=today.isoformat(), - top_nodes_limit=64, - ) - panel_body["nodes_bandwidth"] = nodes_bw or {} - except Exception as exc_nb: # pragma: no cover - optional endpoint - logger.debug("Panel nodes bandwidth range unavailable: %s", exc_nb) - panel_body["nodes_bandwidth"] = {} - try: - online_map = _panel_nodes_online_by_uuid(panel_body.get("nodes")) - lookups = await panel_service.get_nodes_online_lookups() - for k, v in lookups.get("byUuid", {}).items(): - online_map[k] = v - _enrich_bandwidth_nodes_with_online( - panel_body.get("nodes_bandwidth"), - online_map, - lookups.get("byName") or {}, - ) - except Exception as exc_merge: # pragma: no cover - logger.debug("Panel nodes online merge skipped: %s", exc_merge) - payload["panel"] = panel_body - except Exception as exc: - logger.debug("Panel stats unavailable: %s", exc) - payload["panel"] = {"error": "unavailable"} + payload["panel"] = await _load_admin_panel_stats(request, settings, panel_service) queue_manager = get_queue_manager() if queue_manager: @@ -87,3 +52,81 @@ async def admin_stats_route(request: web.Request) -> web.Response: payload["currency_symbol"] = settings.DEFAULT_CURRENCY_SYMBOL or "RUB" return _ok(payload) + + +async def _load_admin_panel_stats( + request: web.Request, + settings: Settings, + panel_service, +) -> Dict[str, Any]: + cache = _admin_panel_stats_cache(settings) + if cache is None: + return await _load_admin_panel_stats_uncached(panel_service) + return await cache.get_or_load("panel", lambda: _load_admin_panel_stats_uncached(panel_service)) + + +def _admin_panel_stats_cache(settings: Settings) -> Optional[AsyncTTLCache]: + ttl_seconds = int(getattr(settings, "ADMIN_PANEL_STATS_CACHE_TTL_SECONDS", 15) or 0) + if ttl_seconds <= 0: + return None + cache_key = (id(settings), ttl_seconds) + cache = _ADMIN_PANEL_STATS_CACHES.get(cache_key) + if cache is None: + cache = AsyncTTLCache( + ttl_seconds=ttl_seconds, + settings=settings, + namespace="admin:panel_stats", + ) + _ADMIN_PANEL_STATS_CACHES[cache_key] = cache + return cache + + +async def _load_admin_panel_stats_uncached(panel_service) -> Dict[str, Any]: + try: + today = datetime.now(timezone.utc).date() + start_d = today - timedelta(days=7) + system, bandwidth, nodes, nodes_bw, lookups = await asyncio.gather( + _safe_panel_call(panel_service.get_system_stats(), "system stats"), + _safe_panel_call(panel_service.get_bandwidth_stats(), "bandwidth stats"), + _safe_panel_call(panel_service.get_nodes_statistics(), "nodes stats"), + _safe_panel_call( + panel_service.get_nodes_bandwidth_usage( + start=start_d.isoformat(), + end=today.isoformat(), + top_nodes_limit=64, + ), + "nodes bandwidth range", + ), + _safe_panel_call(panel_service.get_nodes_online_lookups(), "nodes online lookups"), + ) + + panel_body: Dict[str, Any] = { + "system": system or {}, + "bandwidth": bandwidth or {}, + "nodes": nodes or {}, + "nodes_bandwidth": nodes_bw or {}, + } + if isinstance(lookups, dict): + try: + online_map = _panel_nodes_online_by_uuid(panel_body.get("nodes")) + for k, v in lookups.get("byUuid", {}).items(): + online_map[k] = v + _enrich_bandwidth_nodes_with_online( + panel_body.get("nodes_bandwidth"), + online_map, + lookups.get("byName") or {}, + ) + except Exception as exc_merge: # pragma: no cover + logger.debug("Panel nodes online merge skipped: %s", exc_merge) + return panel_body + except Exception as exc: + logger.debug("Panel stats unavailable: %s", exc) + return {"error": "unavailable"} + + +async def _safe_panel_call(awaitable, label: str) -> Any: + try: + return await awaitable + except Exception as exc: # pragma: no cover - optional panel endpoints + logger.debug("Panel %s unavailable: %s", label, exc) + return None diff --git a/backend/bot/handlers/admin/sync_admin.py b/backend/bot/handlers/admin/sync_admin.py index 5de5b62..25f6f27 100644 --- a/backend/bot/handlers/admin/sync_admin.py +++ b/backend/bot/handlers/admin/sync_admin.py @@ -1,11 +1,11 @@ import asyncio import logging from datetime import datetime, timezone -from typing import Optional, Union +from typing import Any, Optional, Union from aiogram import Bot, Router, types from aiogram.filters import Command -from sqlalchemy import or_, update +from sqlalchemy import func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from bot.middlewares.i18n import JsonI18n @@ -13,7 +13,7 @@ from bot.services.notification_service import NotificationService from bot.services.panel_api_service import PanelApiService from config.settings import Settings from db.dal import panel_sync_dal, subscription_dal, user_dal -from db.models import Subscription +from db.models import Subscription, User router = Router(name="admin_sync_router") @@ -28,6 +28,136 @@ def _normalize_panel_email(value: Optional[str]) -> Optional[str]: return email or None +def _coerce_panel_telegram_id(value: Any) -> Optional[int]: + if value in (None, ""): + return None + try: + return int(value) + except (TypeError, ValueError): + logging.warning("Panel user has non-numeric telegramId: %r", value) + return None + + +def _normalize_description(value: Optional[str]) -> str: + return "\n".join((value or "").split()).strip() + + +def _description_matches(current: Optional[str], desired: str) -> bool: + return _normalize_description(current) == _normalize_description(desired) + + +def _datetime_matches(current: Optional[datetime], desired: datetime) -> bool: + if current is None: + return False + current_dt = current if current.tzinfo else current.replace(tzinfo=timezone.utc) + desired_dt = desired if desired.tzinfo else desired.replace(tzinfo=timezone.utc) + delta = current_dt.astimezone(timezone.utc) - desired_dt.astimezone(timezone.utc) + return abs(delta.total_seconds()) < 1 + + +def _subscription_update_delta( + subscription: Subscription, desired: dict[str, Any] +) -> dict[str, Any]: + delta: dict[str, Any] = {} + for key, desired_value in desired.items(): + current_value = getattr(subscription, key) + if key == "end_date": + if not _datetime_matches(current_value, desired_value): + delta[key] = desired_value + elif current_value != desired_value: + delta[key] = desired_value + return delta + + +async def _prefetch_sync_indexes( + session: AsyncSession, panel_users_data: list[dict[str, Any]] +) -> dict[str, Any]: + telegram_ids: set[int] = set() + panel_uuids: set[str] = set() + emails: set[str] = set() + panel_subscription_uuids: set[str] = set() + panel_uuids_by_telegram_id: dict[int, set[str]] = {} + + for panel_user in panel_users_data: + telegram_id = _coerce_panel_telegram_id(panel_user.get("telegramId")) + panel_uuid = panel_user.get("uuid") + if telegram_id: + telegram_ids.add(telegram_id) + if panel_uuid: + panel_uuids_by_telegram_id.setdefault(telegram_id, set()).add(str(panel_uuid)) + if panel_uuid: + panel_uuids.add(panel_uuid) + email = _normalize_panel_email(panel_user.get("email")) + if email: + emails.add(email) + panel_subscription_uuid = panel_user.get("subscriptionUuid") or panel_user.get("shortUuid") + if panel_subscription_uuid: + panel_subscription_uuids.add(panel_subscription_uuid) + + users_by_telegram_id: dict[int, User] = {} + users_by_user_id: dict[int, User] = {} + users_by_panel_uuid: dict[str, User] = {} + users_by_email: dict[str, User] = {} + + user_filters = [] + if telegram_ids: + user_filters.append(User.telegram_id.in_(telegram_ids)) + user_filters.append(User.user_id.in_(telegram_ids)) + if panel_uuids: + user_filters.append(User.panel_user_uuid.in_(panel_uuids)) + if emails: + user_filters.append(func.lower(User.email).in_(emails)) + if user_filters: + result = await session.execute(select(User).where(or_(*user_filters))) + for user in result.scalars().unique().all(): + if user.telegram_id is not None: + users_by_telegram_id[int(user.telegram_id)] = user + users_by_user_id[int(user.user_id)] = user + if user.panel_user_uuid: + users_by_panel_uuid[user.panel_user_uuid] = user + if user.email: + users_by_email[user.email.strip().lower()] = user + + subscriptions_by_panel_uuid: dict[str, Subscription] = {} + if panel_subscription_uuids: + result = await session.execute( + select(Subscription).where( + Subscription.panel_subscription_uuid.in_(panel_subscription_uuids) + ) + ) + subscriptions_by_panel_uuid = { + str(sub.panel_subscription_uuid): sub + for sub in result.scalars().unique().all() + if sub.panel_subscription_uuid + } + + active_subscriptions_by_user_panel: dict[tuple[int, str], Subscription] = {} + if panel_uuids: + result = await session.execute( + select(Subscription) + .where( + Subscription.panel_user_uuid.in_(panel_uuids), + Subscription.is_active.is_(True), + Subscription.end_date > datetime.now(timezone.utc), + ) + .order_by(Subscription.end_date.desc()) + ) + for sub in result.scalars().unique().all(): + active_subscriptions_by_user_panel.setdefault( + (int(sub.user_id), sub.panel_user_uuid), sub + ) + + return { + "users_by_telegram_id": users_by_telegram_id, + "users_by_user_id": users_by_user_id, + "users_by_panel_uuid": users_by_panel_uuid, + "users_by_email": users_by_email, + "subscriptions_by_panel_uuid": subscriptions_by_panel_uuid, + "active_subscriptions_by_user_panel": active_subscriptions_by_user_panel, + "panel_uuids_by_telegram_id": panel_uuids_by_telegram_id, + } + + def _extract_lifetime_used_traffic_bytes(panel_user_data: dict) -> Optional[int]: user_traffic = panel_user_data.get("userTraffic") or {} raw_value = ( @@ -195,13 +325,23 @@ async def _perform_sync_impl( total_panel_users = len(panel_users_data) logging.info(f"Starting sync for {total_panel_users} panel users.") + sync_indexes = await _prefetch_sync_indexes(session, panel_users_data) + users_by_telegram_id = sync_indexes["users_by_telegram_id"] + users_by_user_id = sync_indexes["users_by_user_id"] + users_by_panel_uuid = sync_indexes["users_by_panel_uuid"] + users_by_email = sync_indexes["users_by_email"] + subscriptions_by_panel_uuid = sync_indexes["subscriptions_by_panel_uuid"] + active_subscriptions_by_user_panel = sync_indexes["active_subscriptions_by_user_panel"] + panel_uuids_by_telegram_id = sync_indexes["panel_uuids_by_telegram_id"] for panel_user_dict in panel_users_data: try: panel_records_checked += 1 panel_uuid = panel_user_dict.get("uuid") panel_user_dict.get("subscriptionUuid") or panel_user_dict.get("shortUuid") - telegram_id_from_panel = panel_user_dict.get("telegramId") + telegram_id_from_panel = _coerce_panel_telegram_id( + panel_user_dict.get("telegramId") + ) email_from_panel = _normalize_panel_email(panel_user_dict.get("email")) if not panel_uuid: @@ -218,20 +358,16 @@ async def _perform_sync_impl( # First, try to find by telegram ID if available if telegram_id_from_panel: - existing_user = await user_dal.get_user_by_telegram_id( - session, telegram_id_from_panel - ) - if not existing_user: - existing_user = await user_dal.get_user_by_id( - session, telegram_id_from_panel - ) + existing_user = users_by_telegram_id.get( + telegram_id_from_panel + ) or users_by_user_id.get(telegram_id_from_panel) if existing_user: logging.debug(f"Found user by telegramId {telegram_id_from_panel}") # If not found by telegram ID, try to find by panel UUID. # The panel UUID is the strongest local link for subscription sync. if not existing_user: - existing_user = await user_dal.get_user_by_panel_uuid(session, panel_uuid) + existing_user = users_by_panel_uuid.get(panel_uuid) if existing_user: logging.debug( f"Found user by panel UUID {panel_uuid}, telegramId: {existing_user.user_id}" # noqa: E501 @@ -248,7 +384,7 @@ async def _perform_sync_impl( # Finally, fall back to email. This mainly catches panel users that # were first imported as email-only identities. if not existing_user and email_from_panel: - existing_user = await user_dal.get_user_by_email(session, email_from_panel) + existing_user = users_by_email.get(email_from_panel) if existing_user: logging.debug(f"Found user by email {email_from_panel}") @@ -281,6 +417,12 @@ async def _perform_sync_impl( ) existing_user = new_user + users_by_user_id[int(new_user.user_id)] = new_user + if new_user.telegram_id is not None: + users_by_telegram_id[int(new_user.telegram_id)] = new_user + users_by_panel_uuid[panel_uuid] = new_user + if email_from_panel: + users_by_email[email_from_panel] = new_user except Exception as e_create: sync_errors.append( @@ -304,6 +446,9 @@ async def _perform_sync_impl( f"Created new email user {new_user.user_id} from panel sync with UUID {panel_uuid}" # noqa: E501 ) existing_user = new_user + users_by_user_id[int(new_user.user_id)] = new_user + users_by_panel_uuid[panel_uuid] = new_user + users_by_email[email_from_panel] = new_user except Exception as e_create_email: sync_errors.append( f"Error creating email user {email_from_panel}: {str(e_create_email)}" # noqa: E501 @@ -327,10 +472,26 @@ async def _perform_sync_impl( # Update panel UUID if different if existing_user.panel_user_uuid != panel_uuid: - existing_user.panel_user_uuid = panel_uuid - user_was_updated = True - users_uuid_updated += 1 - logging.info(f"Updated panel UUID for user {actual_user_id}: {panel_uuid}") + linked_uuid = existing_user.panel_user_uuid + linked_uuid_still_present = bool( + telegram_id_from_panel + and linked_uuid + and str(linked_uuid) + in panel_uuids_by_telegram_id.get(telegram_id_from_panel, set()) + ) + if linked_uuid_still_present: + logging.warning( + "Sync: duplicate panel users share telegramId %s; keeping local panel UUID %s and skipping duplicate panel UUID %s.", # noqa: E501 + telegram_id_from_panel, + linked_uuid, + panel_uuid, + ) + else: + existing_user.panel_user_uuid = panel_uuid + user_was_updated = True + users_uuid_updated += 1 + users_by_panel_uuid[panel_uuid] = existing_user + logging.info(f"Updated panel UUID for user {actual_user_id}: {panel_uuid}") existing_user, email_was_bound = await _bind_panel_email_to_user( session, existing_user=existing_user, @@ -339,9 +500,12 @@ async def _perform_sync_impl( ) if email_was_bound: user_was_updated = True + if email_from_panel: + users_by_email[email_from_panel] = existing_user if telegram_id_from_panel and existing_user.telegram_id != telegram_id_from_panel: existing_user.telegram_id = telegram_id_from_panel user_was_updated = True + users_by_telegram_id[telegram_id_from_panel] = existing_user lifetime_used = _extract_lifetime_used_traffic_bytes(panel_user_dict) if ( @@ -369,7 +533,9 @@ async def _perform_sync_impl( panel_user_dict.get("description") or "" ).strip() desired_description = description_text.strip() - if desired_description and desired_description != current_panel_description: + if desired_description and not _description_matches( + current_panel_description, desired_description + ): await panel_service.update_user_details_on_panel( panel_uuid, { @@ -427,28 +593,31 @@ async def _perform_sync_impl( ) # Try to find subscription by its panel_subscription_uuid first (idempotent) # noqa: E501 - existing_sub_by_uuid = ( - await subscription_dal.get_subscription_by_panel_subscription_uuid( - session, subscription_uuid_from_panel - ) + existing_sub_by_uuid = subscriptions_by_panel_uuid.get( + subscription_uuid_from_panel ) if existing_sub_by_uuid: - # Atomic update of all relevant fields - await subscription_dal.update_subscription( - session, - existing_sub_by_uuid.subscription_id, - { - "user_id": actual_user_id, - "panel_user_uuid": panel_uuid, - "end_date": panel_expire_at, - "is_active": panel_status == "ACTIVE", - "status_from_panel": panel_status, - }, + update_payload = { + "user_id": actual_user_id, + "panel_user_uuid": panel_uuid, + "end_date": panel_expire_at, + "is_active": panel_status == "ACTIVE", + "status_from_panel": panel_status, + } + update_delta = _subscription_update_delta( + existing_sub_by_uuid, update_payload ) + if update_delta: + # Atomic update of changed relevant fields + await subscription_dal.update_subscription( + session, + existing_sub_by_uuid.subscription_id, + update_delta, + ) + subscriptions_updated += 1 + user_was_updated = True subscriptions_synced_count += 1 - subscriptions_updated += 1 - user_was_updated = True logging.debug( f"Synced existing subscription {existing_sub_by_uuid.subscription_id} " # noqa: E501 f"for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" # noqa: E501 @@ -471,6 +640,15 @@ async def _perform_sync_impl( created_sub = await subscription_dal.upsert_subscription( session, sub_payload ) + subscriptions_by_panel_uuid[subscription_uuid_from_panel] = ( + created_sub + ) + if created_sub.is_active and created_sub.end_date > datetime.now( + timezone.utc + ): + active_subscriptions_by_user_panel[ + (int(created_sub.user_id), created_sub.panel_user_uuid) + ] = created_sub subscriptions_synced_count += 1 subscriptions_created += 1 user_was_updated = True @@ -480,22 +658,27 @@ async def _perform_sync_impl( ) else: # No subscription UUID from panel: only update an already active subscription for this user/panel UUID # noqa: E501 - active_sub = await subscription_dal.get_active_subscription_by_user_id( - session, actual_user_id, panel_uuid + active_sub = active_subscriptions_by_user_panel.get( + (actual_user_id, panel_uuid) ) if active_sub: - await subscription_dal.update_subscription( - session, - active_sub.subscription_id, - { - "end_date": panel_expire_at, - "is_active": panel_status == "ACTIVE", - "status_from_panel": panel_status, - }, + update_payload = { + "end_date": panel_expire_at, + "is_active": panel_status == "ACTIVE", + "status_from_panel": panel_status, + } + update_delta = _subscription_update_delta( + active_sub, update_payload ) + if update_delta: + await subscription_dal.update_subscription( + session, + active_sub.subscription_id, + update_delta, + ) + subscriptions_updated += 1 + user_was_updated = True subscriptions_synced_count += 1 - subscriptions_updated += 1 - user_was_updated = True logging.debug( f"Updated active subscription {active_sub.subscription_id} " f"for user {actual_user_id}: expires {panel_expire_at}, status {panel_status}" # noqa: E501 diff --git a/backend/bot/middlewares/profile_sync.py b/backend/bot/middlewares/profile_sync.py index cb6dedf..db10400 100644 --- a/backend/bot/middlewares/profile_sync.py +++ b/backend/bot/middlewares/profile_sync.py @@ -1,4 +1,5 @@ import logging +import time from typing import Any, Awaitable, Callable, Dict, Optional from aiogram import BaseMiddleware @@ -6,9 +7,13 @@ from aiogram.types import Update from aiogram.types import User as TgUser from sqlalchemy.ext.asyncio import AsyncSession +from bot.infra.redis import cache_get_json, cache_set_json, redis_key from bot.utils.text_sanitizer import sanitize_display_name, sanitize_username, username_for_display +from config.settings import Settings from db.dal import user_dal +_LOCAL_PROFILE_SYNC_CHECKS: Dict[int, float] = {} + class ProfileSyncMiddleware(BaseMiddleware): async def __call__( @@ -19,8 +24,12 @@ class ProfileSyncMiddleware(BaseMiddleware): ) -> Any: session: AsyncSession = data.get("session") tg_user: Optional[TgUser] = data.get("event_from_user") + settings: Optional[Settings] = data.get("settings") if session and tg_user: + if settings and await _profile_sync_recently_checked(settings, int(tg_user.id)): + return await handler(event, data) + try: db_user = await user_dal.get_user_by_telegram_id(session, tg_user.id) if not db_user: @@ -79,5 +88,42 @@ class ProfileSyncMiddleware(BaseMiddleware): f"ProfileSyncMiddleware: Failed to sync profile for user {getattr(tg_user, 'id', 'N/A')}: {e}", # noqa: E501 exc_info=True, ) + finally: + if settings: + await _mark_profile_sync_checked(settings, int(tg_user.id)) return await handler(event, data) + + +async def _profile_sync_recently_checked(settings: Settings, telegram_id: int) -> bool: + ttl_seconds = int(getattr(settings, "PROFILE_SYNC_CACHE_TTL_SECONDS", 900) or 0) + if ttl_seconds <= 0: + return False + + now = time.monotonic() + expires_at = _LOCAL_PROFILE_SYNC_CHECKS.get(telegram_id) + if expires_at and expires_at > now: + return True + + key = redis_key(settings, "cache", "profile-sync", telegram_id) + try: + cached = await cache_get_json(settings, key) + except Exception: + cached = None + if cached: + _LOCAL_PROFILE_SYNC_CHECKS[telegram_id] = now + ttl_seconds + return True + return False + + +async def _mark_profile_sync_checked(settings: Settings, telegram_id: int) -> None: + ttl_seconds = int(getattr(settings, "PROFILE_SYNC_CACHE_TTL_SECONDS", 900) or 0) + if ttl_seconds <= 0: + return + + _LOCAL_PROFILE_SYNC_CHECKS[telegram_id] = time.monotonic() + ttl_seconds + key = redis_key(settings, "cache", "profile-sync", telegram_id) + try: + await cache_set_json(settings, key, {"checked": True}, ttl_seconds) + except Exception: + pass diff --git a/backend/bot/services/panel_api_service.py b/backend/bot/services/panel_api_service.py index 2f0e980..0d6a8cf 100644 --- a/backend/bot/services/panel_api_service.py +++ b/backend/bot/services/panel_api_service.py @@ -41,6 +41,24 @@ class PanelApiService: settings=settings, namespace="panel:hosts", ) + self._users_cache: AsyncTTLCache = AsyncTTLCache( + ttl_seconds=max(0, int(getattr(settings, "PANEL_USER_CACHE_TTL_SECONDS", 5) or 0)), + settings=settings, + namespace="panel:users", + ) + self._devices_cache: AsyncTTLCache = AsyncTTLCache( + ttl_seconds=max(0, int(getattr(settings, "PANEL_DEVICES_CACHE_TTL_SECONDS", 5) or 0)), + settings=settings, + namespace="panel:devices", + ) + self._all_users_cache: AsyncTTLCache = AsyncTTLCache( + ttl_seconds=max( + 0, + int(getattr(settings, "PANEL_ALL_USERS_CACHE_TTL_SECONDS", 5) or 0), + ), + settings=settings, + namespace="panel:all_users", + ) async def __aenter__(self): """Context manager entry""" @@ -226,6 +244,18 @@ class PanelApiService: async def get_all_panel_users( self, page_size: int = 100, log_responses: bool = False + ) -> Optional[List[Dict[str, Any]]]: + if log_responses or page_size != 100 or self._all_users_cache.ttl_seconds <= 0: + return await self._get_all_panel_users_uncached( + page_size=page_size, log_responses=log_responses + ) + return await self._all_users_cache.get_or_load( + f"page_size:{page_size}", + lambda: self._get_all_panel_users_uncached(page_size=page_size, log_responses=False), + ) + + async def _get_all_panel_users_uncached( + self, page_size: int = 100, log_responses: bool = False ) -> Optional[List[Dict[str, Any]]]: all_users = [] start_offset = 0 @@ -253,6 +283,16 @@ class PanelApiService: async def get_user_by_uuid( self, user_uuid: str, log_response: bool = False + ) -> Optional[Dict[str, Any]]: + if log_response or self._users_cache.ttl_seconds <= 0: + return await self._get_user_by_uuid_uncached(user_uuid, log_response=log_response) + return await self._users_cache.get_or_load( + f"uuid:{user_uuid}", + lambda: self._get_user_by_uuid_uncached(user_uuid, log_response=False), + ) + + async def _get_user_by_uuid_uncached( + self, user_uuid: str, log_response: bool = False ) -> Optional[Dict[str, Any]]: endpoint = f"/users/{user_uuid}" full_response = await self._request("GET", endpoint, log_full_response=log_response) @@ -422,6 +462,7 @@ class PanelApiService: "POST", "/users", json=payload, log_full_response=log_response ) if response and not response.get("error") and "response" in response: + self._invalidate_all_users_cache() logging.info( f"Panel user '{username_on_panel}' created successfully (UUID: {response.get('response', {}).get('uuid')})." # noqa: E501 ) @@ -443,6 +484,8 @@ class PanelApiService: ) if full_response and not full_response.get("error") and "response" in full_response: logging.debug("User %s details updated on panel.", user_uuid) + self._invalidate_user_cache(user_uuid) + self._invalidate_all_users_cache() return full_response.get("response") logging.error( @@ -458,6 +501,8 @@ class PanelApiService: response_data = await self._request("POST", endpoint, log_full_response=log_response) if response_data and not response_data.get("error") and "response" in response_data: + self._invalidate_user_cache(user_uuid) + self._invalidate_all_users_cache() actual_status = response_data.get("response", {}).get("status") expected_status = "ACTIVE" if enable else "DISABLED" if actual_status == expected_status: @@ -494,11 +539,17 @@ class PanelApiService: logging.info( f"Panel user {user_uuid} already absent (errorCode {error_code}). Treating as deleted." # noqa: E501 ) + self._invalidate_user_cache(user_uuid) + self._invalidate_devices_cache(user_uuid) + self._invalidate_all_users_cache() return True logging.error(f"Failed to delete user {user_uuid} on panel. Response: {response_data}") return False logging.info(f"Panel user {user_uuid} deleted successfully.") + self._invalidate_user_cache(user_uuid) + self._invalidate_devices_cache(user_uuid) + self._invalidate_all_users_cache() return True async def get_subscription_link( @@ -513,6 +564,14 @@ class PanelApiService: return base_sub_url async def get_user_devices(self, user_uuid: str) -> Optional[List[Dict[str, Any]]]: + if self._devices_cache.ttl_seconds <= 0: + return await self._get_user_devices_uncached(user_uuid) + return await self._devices_cache.get_or_load( + f"user:{user_uuid}", + lambda: self._get_user_devices_uncached(user_uuid), + ) + + async def _get_user_devices_uncached(self, user_uuid: str) -> Optional[List[Dict[str, Any]]]: endpoint = f"/hwid/devices/{user_uuid}" response_data = await self._request("GET", endpoint, log_full_response=False) if response_data and not response_data.get("error") and "response" in response_data: @@ -525,6 +584,7 @@ class PanelApiService: payload = {"userUuid": user_uuid, "hwid": hwid} response_data = await self._request("POST", endpoint, json=payload, log_full_response=False) if response_data and not response_data.get("error") and "response" in response_data: + self._invalidate_devices_cache(user_uuid) return True logging.error( f"Failed to disconnect device {hwid} for user {user_uuid}. Payload: {payload}, Response: {response_data}" # noqa: E501 @@ -629,6 +689,19 @@ class PanelApiService: def _invalidate_squad_caches(self) -> None: self._squads_cache.invalidate() + def _invalidate_user_cache(self, user_uuid: Optional[str]) -> None: + if not user_uuid: + return + self._users_cache.invalidate(f"uuid:{user_uuid}") + + def _invalidate_all_users_cache(self) -> None: + self._all_users_cache.invalidate() + + def _invalidate_devices_cache(self, user_uuid: Optional[str]) -> None: + if not user_uuid: + return + self._devices_cache.invalidate(f"user:{user_uuid}") + async def get_internal_squads(self) -> Optional[List[Dict[str, Any]]]: return await self._squads_cache.get_or_load("list", self._get_internal_squads_uncached) diff --git a/backend/bot/services/tariff_worker.py b/backend/bot/services/tariff_worker.py index 86be457..999472d 100644 --- a/backend/bot/services/tariff_worker.py +++ b/backend/bot/services/tariff_worker.py @@ -29,7 +29,7 @@ PREMIUM_WARNING_DEPLETED_LEVEL = PREMIUM_WARNING_LEVEL_OFFSET + 100 # to avoid an N+1 serial chain to the Remnawave panel each tick. TARIFF_WORKER_BATCH_SIZE = 50 TARIFF_WORKER_PANEL_CONCURRENCY = 10 -TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD = 200 +TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD = 50 class TariffTrafficWorker: diff --git a/backend/config/settings.py b/backend/config/settings.py index 94446db..b09596a 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -96,6 +96,11 @@ class Settings(BaseSettings): REDIS_KEY_PREFIX: str = Field(default="remnawave-tg-shop") WEBAPP_ME_CACHE_TTL_SECONDS: int = Field(default=15) WEBAPP_DEVICES_CACHE_TTL_SECONDS: int = Field(default=5) + PANEL_USER_CACHE_TTL_SECONDS: int = Field(default=5) + PANEL_DEVICES_CACHE_TTL_SECONDS: int = Field(default=5) + PANEL_ALL_USERS_CACHE_TTL_SECONDS: int = Field(default=5) + ADMIN_PANEL_STATS_CACHE_TTL_SECONDS: int = Field(default=15) + PROFILE_SYNC_CACHE_TTL_SECONDS: int = Field(default=900) WEBAPP_RATE_LIMIT_TTL_SECONDS: int = Field(default=60) WEBAPP_RATE_LIMIT_MAX_REQUESTS: int = Field(default=30) WEBHOOK_QUEUE_NAME: str = Field(default="webhook-events") @@ -103,7 +108,7 @@ class Settings(BaseSettings): WORKER_PANEL_SYNC_INTERVAL_SECONDS: int = Field(default=900) TARIFF_WORKER_LOCK_TTL_SECONDS: int = Field(default=240) TARIFF_WORKER_TICK_SECONDS: int = Field(default=300) - TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD: int = Field(default=200) + TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD: int = Field(default=50) DEFAULT_LANGUAGE: str = Field(default="ru") DEFAULT_CURRENCY_SYMBOL: str = Field(default="RUB") diff --git a/scripts/perf_benchmarks.py b/scripts/perf_benchmarks.py index 83d51d3..cee0a9c 100644 --- a/scripts/perf_benchmarks.py +++ b/scripts/perf_benchmarks.py @@ -3,7 +3,6 @@ from __future__ import annotations import argparse import asyncio import json -import math import sys import time from pathlib import Path @@ -16,7 +15,15 @@ for path in (str(BACKEND), str(ROOT)): if path not in sys.path: sys.path.insert(0, path) +import bot.app.web.subscription_webapp # noqa: E402,F401 +from bot.app.web.admin_api_impl import stats as admin_stats_module # noqa: E402 +from bot.handlers.admin.sync_admin import ( # noqa: E402 + _description_matches, + _subscription_update_delta, +) +from bot.middlewares import profile_sync as profile_sync_module # noqa: E402 from bot.services import panel_api_service # noqa: E402 +from bot.services.panel_api_service import PanelApiService # noqa: E402 from bot.services.tariff_worker import TariffTrafficWorker # noqa: E402 from bot.utils import config_link # noqa: E402 from bot.utils.config_link import prepare_config_links # noqa: E402 @@ -25,6 +32,14 @@ from bot.utils.ttl_cache import AsyncTTLCache # noqa: E402 DEFAULT_USER_SIZES = (200, 500, 1000, 5000, 10000) +def estimated_panel_user_pages(users: int, page_size: int = 100) -> int: + if users <= 0: + return 1 + # get_all_panel_users stops on a short/empty page, so exact page multiples + # need one final empty-page request. + return users // page_size + 1 + + class FakePanel: def __init__(self, users: int): self.calls = 0 @@ -84,7 +99,7 @@ async def bench_premium_usage(users: int) -> dict: async def bench_panel_user_prefetch(users: int) -> dict: panel = FakeBulkPanel(users) worker = TariffTrafficWorker( - settings=SimpleNamespace(TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD=200), + settings=SimpleNamespace(TARIFF_WORKER_BULK_PANEL_FETCH_THRESHOLD=50), session_factory=SimpleNamespace(), panel_service=panel, subscription_service=SimpleNamespace(), @@ -98,7 +113,156 @@ async def bench_panel_user_prefetch(users: int) -> dict: "service_calls": panel.calls, "matched": len(by_uuid or {}), "legacy_user_get_calls": users, - "estimated_bulk_http_pages_at_100": math.ceil(users / 100), + "estimated_bulk_http_pages_at_100": estimated_panel_user_pages(users), + } + + +async def bench_panel_sync_startup(users: int) -> dict: + end_date = "2026-06-20T12:00:00+00:00" + from datetime import datetime, timezone + + parsed_end_date = datetime.fromisoformat(end_date).astimezone(timezone.utc) + started = time.perf_counter() + subscription_writes = 0 + description_patches = 0 + for index in range(users): + desired_description = f"user{index}@example.test\nusername_{index}" + current_description = f"user{index}@example.test username_{index}" + if not _description_matches(current_description, desired_description): + description_patches += 1 + subscription = SimpleNamespace( + user_id=index, + panel_user_uuid=f"panel-{index}", + end_date=parsed_end_date, + is_active=True, + status_from_panel="ACTIVE", + ) + delta = _subscription_update_delta( + subscription, + { + "user_id": index, + "panel_user_uuid": f"panel-{index}", + "end_date": parsed_end_date, + "is_active": True, + "status_from_panel": "ACTIVE", + }, + ) + if delta: + subscription_writes += 1 + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "panel_get_pages_estimate": estimated_panel_user_pages(users), + "legacy_user_lookup_queries_estimate": users * 3, + "optimized_user_lookup_queries_estimate": 1, + "legacy_subscription_lookup_queries_estimate": users, + "optimized_subscription_lookup_queries_estimate": 1, + "legacy_subscription_write_attempts": users, + "optimized_subscription_writes": subscription_writes, + "description_panel_patches": description_patches, + } + + +async def bench_panel_user_cache(users: int) -> dict: + settings = SimpleNamespace( + PANEL_API_URL="https://panel.example.test/api", + PANEL_API_KEY="key", + USER_HWID_DEVICE_LIMIT=None, + PANEL_USER_CACHE_TTL_SECONDS=60, + PANEL_DEVICES_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + service = PanelApiService(settings) + calls = 0 + + async def fake_request(method, endpoint, log_full_response=False, **kwargs): + nonlocal calls + calls += 1 + await asyncio.sleep(0.001) + return {"response": {"uuid": "panel-user", "username": "cached"}} + + service._request = fake_request + started = time.perf_counter() + await asyncio.gather(*(service.get_user_by_uuid("panel-user") for _ in range(users))) + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "panel_calls": calls, + "legacy_panel_calls": users, + } + + +async def bench_panel_all_users_cache(users: int) -> dict: + settings = SimpleNamespace( + PANEL_API_URL="https://panel.example.test/api", + PANEL_API_KEY="key", + USER_HWID_DEVICE_LIMIT=None, + PANEL_USER_CACHE_TTL_SECONDS=60, + PANEL_DEVICES_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + service = PanelApiService(settings) + panel_users = [{"uuid": f"panel-{index}"} for index in range(users)] + calls = 0 + + async def fake_request(method, endpoint, log_full_response=False, **kwargs): + nonlocal calls + calls += 1 + await asyncio.sleep(0.001) + params = kwargs.get("params") or {} + size = int(params.get("size", 100)) + start = int(params.get("start", 0)) + return {"response": {"users": panel_users[start : start + size]}} + + service._request = fake_request + started = time.perf_counter() + first, second = await asyncio.gather( + service.get_all_panel_users(), + service.get_all_panel_users(), + ) + elapsed = time.perf_counter() - started + pages = estimated_panel_user_pages(users) + return { + "seconds": elapsed, + "panel_calls": calls, + "legacy_panel_calls": pages * 2, + "users_first": len(first or []), + "users_second": len(second or []), + } + + +async def bench_panel_devices_cache(users: int) -> dict: + settings = SimpleNamespace( + PANEL_API_URL="https://panel.example.test/api", + PANEL_API_KEY="key", + USER_HWID_DEVICE_LIMIT=None, + PANEL_USER_CACHE_TTL_SECONDS=60, + PANEL_DEVICES_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + service = PanelApiService(settings) + calls = 0 + + async def fake_request(method, endpoint, log_full_response=False, **kwargs): + nonlocal calls + calls += 1 + await asyncio.sleep(0.001) + return {"response": [{"hwid": "device-1"}]} + + service._request = fake_request + started = time.perf_counter() + await asyncio.gather(*(service.get_user_devices("panel-user") for _ in range(users))) + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "panel_calls": calls, + "legacy_panel_calls": users, } @@ -132,6 +296,83 @@ async def bench_ttl_singleflight(users: int) -> dict: } +class FakeAdminStatsPanel: + def __init__(self): + self.calls = { + "system": 0, + "bandwidth": 0, + "nodes": 0, + "nodes_bandwidth": 0, + "online": 0, + } + + async def get_system_stats(self): + self.calls["system"] += 1 + await asyncio.sleep(0.001) + return {"users": {"totalUsers": 10}} + + async def get_bandwidth_stats(self): + self.calls["bandwidth"] += 1 + await asyncio.sleep(0.001) + return {"current": 123} + + async def get_nodes_statistics(self): + self.calls["nodes"] += 1 + await asyncio.sleep(0.001) + return {"nodes": []} + + async def get_nodes_bandwidth_usage(self, *, start: str, end: str, top_nodes_limit: int = 64): + self.calls["nodes_bandwidth"] += 1 + await asyncio.sleep(0.001) + return {"topNodes": []} + + async def get_nodes_online_lookups(self): + self.calls["online"] += 1 + await asyncio.sleep(0.001) + return {"byUuid": {}, "byName": {}} + + +async def bench_admin_stats_cache(users: int) -> dict: + admin_stats_module._ADMIN_PANEL_STATS_CACHES.clear() + settings = SimpleNamespace( + ADMIN_PANEL_STATS_CACHE_TTL_SECONDS=60, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + panel = FakeAdminStatsPanel() + started = time.perf_counter() + await asyncio.gather( + *(admin_stats_module._load_admin_panel_stats(None, settings, panel) for _ in range(users)) + ) + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "panel_endpoint_calls": sum(panel.calls.values()), + "legacy_panel_endpoint_calls": users * len(panel.calls), + } + + +async def bench_profile_sync_guard(users: int) -> dict: + profile_sync_module._LOCAL_PROFILE_SYNC_CHECKS.clear() + settings = SimpleNamespace( + PROFILE_SYNC_CACHE_TTL_SECONDS=900, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + allowed_checks = 0 + started = time.perf_counter() + for _ in range(users): + if not await profile_sync_module._profile_sync_recently_checked(settings, 42): + allowed_checks += 1 + await profile_sync_module._mark_profile_sync_checked(settings, 42) + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "profile_checks_allowed": allowed_checks, + "legacy_profile_checks": users, + } + + async def bench_crypt4(users: int) -> dict: config_link._CRYPT4_LINK_CACHES.clear() settings = SimpleNamespace( @@ -175,9 +416,15 @@ async def run_suite(user_sizes: tuple[int, ...]) -> dict: results: dict[str, dict] = {} for users in user_sizes: results[str(users)] = { + "panel_sync_startup": await bench_panel_sync_startup(users), "panel_user_bulk_prefetch": await bench_panel_user_prefetch(users), + "panel_all_users_cache": await bench_panel_all_users_cache(users), + "panel_user_cache": await bench_panel_user_cache(users), + "panel_devices_cache": await bench_panel_devices_cache(users), "premium_usage_1_node": await bench_premium_usage(users), "ttl_cache_cold_single_key": await bench_ttl_singleflight(users), + "admin_stats_cache": await bench_admin_stats_cache(users), + "profile_sync_guard": await bench_profile_sync_guard(users), "crypt4_same_link": await bench_crypt4(users), } return results @@ -186,16 +433,26 @@ async def run_suite(user_sizes: tuple[int, ...]) -> dict: def _print_table(results: dict[str, dict]) -> None: print( "users | bulk_pages_est | premium_usage_s | premium_panel_calls | " - "ttl_loader_calls | crypt4_panel_calls" + "sync_db_reads_est | sync_db_writes | user_cache_calls | " + "all_users_calls | device_cache_calls | admin_panel_calls | crypt4_panel_calls" ) - print("-" * 104) + print("-" * 154) for users, data in results.items(): + sync_optimized_reads = ( + data["panel_sync_startup"]["optimized_user_lookup_queries_estimate"] + + data["panel_sync_startup"]["optimized_subscription_lookup_queries_estimate"] + ) print( f"{users:>5} | " f"{data['panel_user_bulk_prefetch']['estimated_bulk_http_pages_at_100']:>14} | " f"{data['premium_usage_1_node']['seconds']:>15.6f} | " f"{data['premium_usage_1_node']['panel_calls']:>19} | " - f"{data['ttl_cache_cold_single_key']['loader_calls']:>16} | " + f"{sync_optimized_reads:>17} | " + f"{data['panel_sync_startup']['optimized_subscription_writes']:>14} | " + f"{data['panel_user_cache']['panel_calls']:>16} | " + f"{data['panel_all_users_cache']['panel_calls']:>15} | " + f"{data['panel_devices_cache']['panel_calls']:>18} | " + f"{data['admin_stats_cache']['panel_endpoint_calls']:>17} | " f"{data['crypt4_same_link']['panel_calls']:>18}" ) diff --git a/tests/test_admin_panel_stats_cache.py b/tests/test_admin_panel_stats_cache.py new file mode 100644 index 0000000..328a46e --- /dev/null +++ b/tests/test_admin_panel_stats_cache.py @@ -0,0 +1,59 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import bot.app.web.subscription_webapp # noqa: F401 +from bot.app.web.admin_api_impl import stats as stats_module + + +class AdminPanelStatsCacheTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + stats_module._ADMIN_PANEL_STATS_CACHES.clear() + + async def asyncTearDown(self): + stats_module._ADMIN_PANEL_STATS_CACHES.clear() + + def _settings(self): + return SimpleNamespace( + ADMIN_PANEL_STATS_CACHE_TTL_SECONDS=15, + REDIS_URL="redis://redis:6379/0", + REDIS_KEY_PREFIX="shop", + ) + + def _panel_service(self): + return SimpleNamespace( + get_system_stats=AsyncMock(return_value={"users": {"totalUsers": 10}}), + get_bandwidth_stats=AsyncMock(return_value={"current": 123}), + get_nodes_statistics=AsyncMock(return_value={"nodes": []}), + get_nodes_bandwidth_usage=AsyncMock(return_value={"topNodes": []}), + get_nodes_online_lookups=AsyncMock(return_value={"byUuid": {}, "byName": {}}), + ) + + async def test_admin_panel_stats_are_cached_between_requests(self): + settings = self._settings() + panel_service = self._panel_service() + cache_store = {} + + async def fake_get(_settings, key): + return cache_store.get(key) + + async def fake_set(_settings, key, value, ttl): + cache_store[key] = value + + with ( + patch("bot.infra.redis.cache_get_json", fake_get), + patch("bot.infra.redis.cache_set_json", fake_set), + ): + first = await stats_module._load_admin_panel_stats(None, settings, panel_service) + second = await stats_module._load_admin_panel_stats(None, settings, panel_service) + + self.assertEqual(first, second) + panel_service.get_system_stats.assert_awaited_once() + panel_service.get_bandwidth_stats.assert_awaited_once() + panel_service.get_nodes_statistics.assert_awaited_once() + panel_service.get_nodes_bandwidth_usage.assert_awaited_once() + panel_service.get_nodes_online_lookups.assert_awaited_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_admin_sync_performance.py b/tests/test_admin_sync_performance.py new file mode 100644 index 0000000..65f5823 --- /dev/null +++ b/tests/test_admin_sync_performance.py @@ -0,0 +1,71 @@ +from datetime import datetime, timedelta, timezone + +from bot.handlers.admin.sync_admin import ( + _coerce_panel_telegram_id, + _description_matches, + _subscription_update_delta, +) +from db.models import Subscription + + +def test_description_match_ignores_whitespace_shape(): + assert _description_matches("email@example.com username", "email@example.com\nusername") + + +def test_panel_telegram_id_is_coerced_to_int(): + assert _coerce_panel_telegram_id("12345") == 12345 + assert _coerce_panel_telegram_id("") is None + + +def test_subscription_update_delta_skips_unchanged_fields(): + end_date = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) + subscription = Subscription( + user_id=1, + panel_user_uuid="panel-1", + panel_subscription_uuid="sub-1", + end_date=end_date, + is_active=True, + status_from_panel="ACTIVE", + ) + + assert ( + _subscription_update_delta( + subscription, + { + "user_id": 1, + "panel_user_uuid": "panel-1", + "end_date": end_date + timedelta(milliseconds=500), + "is_active": True, + "status_from_panel": "ACTIVE", + }, + ) + == {} + ) + + +def test_subscription_update_delta_returns_only_changed_fields(): + end_date = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) + subscription = Subscription( + user_id=1, + panel_user_uuid="panel-1", + panel_subscription_uuid="sub-1", + end_date=end_date, + is_active=True, + status_from_panel="ACTIVE", + ) + + assert _subscription_update_delta( + subscription, + { + "user_id": 2, + "panel_user_uuid": "panel-1", + "end_date": end_date + timedelta(seconds=2), + "is_active": False, + "status_from_panel": "EXPIRED", + }, + ) == { + "user_id": 2, + "end_date": end_date + timedelta(seconds=2), + "is_active": False, + "status_from_panel": "EXPIRED", + } diff --git a/tests/test_panel_api_service_logging.py b/tests/test_panel_api_service_logging.py index c76b5a8..fb00de7 100644 --- a/tests/test_panel_api_service_logging.py +++ b/tests/test_panel_api_service_logging.py @@ -1,3 +1,4 @@ +import asyncio import unittest from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -46,6 +47,65 @@ class PanelApiServiceLoggingTests(unittest.IsolatedAsyncioTestCase): self.assertTrue(service._request.await_args.kwargs["log_full_response"]) + async def test_get_user_by_uuid_uses_short_ttl_cache_and_update_invalidates(self): + service = self._make_service() + service._request = AsyncMock(return_value={"response": {"uuid": "user-uuid"}}) + + first = await service.get_user_by_uuid("user-uuid") + second = await service.get_user_by_uuid("user-uuid") + + self.assertEqual(first, {"uuid": "user-uuid"}) + self.assertEqual(second, {"uuid": "user-uuid"}) + self.assertEqual(service._request.await_count, 1) + + await service.update_user_details_on_panel("user-uuid", {"description": "updated"}) + await service.get_user_by_uuid("user-uuid") + + self.assertEqual(service._request.await_count, 3) + + 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"}]}) + + first = await service.get_user_devices("user-uuid") + second = await service.get_user_devices("user-uuid") + + self.assertEqual(first, [{"hwid": "device-1"}]) + self.assertEqual(second, [{"hwid": "device-1"}]) + self.assertEqual(service._request.await_count, 1) + + await service.disconnect_device("user-uuid", "device-1") + await service.get_user_devices("user-uuid") + + self.assertEqual(service._request.await_count, 3) + + async def test_get_all_panel_users_uses_singleflight_cache_and_update_invalidates(self): + service = self._make_service() + get_calls = 0 + + async def fake_request(method, endpoint, **kwargs): + nonlocal get_calls + if method == "GET": + get_calls += 1 + return {"response": {"users": [{"uuid": "user-uuid"}]}} + return {"response": {"uuid": "user-uuid"}} + + service._request = AsyncMock(side_effect=fake_request) + + first, second = await asyncio.gather( + service.get_all_panel_users(), + service.get_all_panel_users(), + ) + + self.assertEqual(first, [{"uuid": "user-uuid"}]) + self.assertEqual(second, [{"uuid": "user-uuid"}]) + self.assertEqual(get_calls, 1) + + await service.update_user_details_on_panel("user-uuid", {"description": "updated"}) + await service.get_all_panel_users() + + self.assertEqual(get_calls, 2) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_profile_sync_middleware.py b/tests/test_profile_sync_middleware.py new file mode 100644 index 0000000..33b123a --- /dev/null +++ b/tests/test_profile_sync_middleware.py @@ -0,0 +1,74 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from bot.middlewares import profile_sync as profile_sync_module +from bot.middlewares.profile_sync import ProfileSyncMiddleware + + +class ProfileSyncMiddlewareCacheTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + profile_sync_module._LOCAL_PROFILE_SYNC_CHECKS.clear() + + async def asyncTearDown(self): + profile_sync_module._LOCAL_PROFILE_SYNC_CHECKS.clear() + + def _settings(self): + return SimpleNamespace( + PROFILE_SYNC_CACHE_TTL_SECONDS=900, + REDIS_URL="redis://redis:6379/0", + REDIS_KEY_PREFIX="shop", + ) + + async def test_profile_sync_skips_repeated_user_checks_inside_ttl(self): + middleware = ProfileSyncMiddleware() + handler = AsyncMock(return_value="ok") + event = SimpleNamespace() + tg_user = SimpleNamespace( + id=42, + username="alice", + first_name="Alice", + last_name="Smith", + ) + db_user = SimpleNamespace( + user_id=42, + telegram_id=42, + username="alice", + first_name="Alice", + last_name="Smith", + email=None, + panel_user_uuid=None, + ) + cache_store = {} + + async def fake_get(_settings, key): + return cache_store.get(key) + + async def fake_set(_settings, key, value, ttl): + cache_store[key] = value + + data = { + "session": AsyncMock(), + "event_from_user": tg_user, + "settings": self._settings(), + } + with ( + patch.object(profile_sync_module, "cache_get_json", fake_get), + patch.object(profile_sync_module, "cache_set_json", fake_set), + patch.object( + profile_sync_module.user_dal, + "get_user_by_telegram_id", + AsyncMock(return_value=db_user), + ) as get_user, + ): + first = await middleware(handler, event, data) + second = await middleware(handler, event, data) + + self.assertEqual(first, "ok") + self.assertEqual(second, "ok") + get_user.assert_awaited_once() + self.assertEqual(handler.await_count, 2) + + +if __name__ == "__main__": + unittest.main()