diff --git a/.env.example b/.env.example index 56bf837..cde0a82 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,7 @@ PANEL_ALL_USERS_CACHE_TTL_SECONDS=5 # PANEL_ALL_USERS_PAGE_SIZE=1000 # Remnawave /users page size with fallback to 100 ADMIN_PANEL_STATS_CACHE_TTL_SECONDS=15 # Short TTL for admin panel stats fetched from Remnawave ADMIN_DB_STATS_CACHE_TTL_SECONDS=5 # Short TTL for expensive admin dashboard DB aggregates +ADMIN_USERS_LIST_CACHE_TTL_SECONDS=3 # Short TTL for admin users list queries PROFILE_SYNC_CACHE_TTL_SECONDS=900 # Minimum seconds between Telegram profile sync checks per user PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600 # Min seconds between local lifetime traffic writes per user PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=104857600 # Write lifetime traffic sooner when delta is at least this many bytes diff --git a/backend/bot/app/web/admin_api_impl/users.py b/backend/bot/app/web/admin_api_impl/users.py index 95073d2..9db7c7d 100644 --- a/backend/bot/app/web/admin_api_impl/users.py +++ b/backend/bot/app/web/admin_api_impl/users.py @@ -1,13 +1,21 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 +import hashlib from html import escape as html_escape from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup +from bot.app.web.webapp.cache_helpers import invalidate_webapp_user_caches +from bot.infra.redis import cache_delete_pattern, redis_key +from bot.utils.ttl_cache import AsyncTTLCache + +_ADMIN_USERS_LIST_CACHES: Dict[tuple[int, int], AsyncTTLCache] = {} + async def admin_users_list_route(request: web.Request) -> web.Response: _require_admin_user_id(request) + settings: Settings = request.app["settings"] async_session_factory: sessionmaker = request.app["async_session_factory"] page = max(0, int(request.query.get("page", 0) or 0)) @@ -18,6 +26,79 @@ async def admin_users_list_route(request: web.Request) -> web.Response: premium_traffic = (request.query.get("premium_traffic") or "all").lower() sort_value = (request.query.get("sort") or "registered_desc").lower() + payload = await _load_admin_users_list_payload( + settings, + async_session_factory, + page=page, + page_size=page_size, + query=query, + filter_value=filter_value, + panel_status=panel_status, + premium_traffic=premium_traffic, + sort_value=sort_value, + ) + return _ok(payload) + + +async def _load_admin_users_list_payload( + settings: Settings, + async_session_factory: sessionmaker, + *, + page: int, + page_size: int, + query: str, + filter_value: str, + panel_status: str, + premium_traffic: str, + sort_value: str, +) -> Dict[str, Any]: + cache = _admin_users_list_cache(settings) + cache_key = _admin_users_list_cache_key( + page=page, + page_size=page_size, + query=query, + filter_value=filter_value, + panel_status=panel_status, + premium_traffic=premium_traffic, + sort_value=sort_value, + ) + if cache is None: + return await _load_admin_users_list_payload_uncached( + async_session_factory, + page=page, + page_size=page_size, + query=query, + filter_value=filter_value, + panel_status=panel_status, + premium_traffic=premium_traffic, + sort_value=sort_value, + ) + return await cache.get_or_load( + cache_key, + lambda: _load_admin_users_list_payload_uncached( + async_session_factory, + page=page, + page_size=page_size, + query=query, + filter_value=filter_value, + panel_status=panel_status, + premium_traffic=premium_traffic, + sort_value=sort_value, + ), + ) + + +async def _load_admin_users_list_payload_uncached( + async_session_factory: sessionmaker, + *, + page: int, + page_size: int, + query: str, + filter_value: str, + panel_status: str, + premium_traffic: str, + sort_value: str, +) -> Dict[str, Any]: async with async_session_factory() as session: users, total = await _filter_and_sort_users( session, @@ -51,14 +132,58 @@ async def admin_users_list_route(request: web.Request) -> web.Response: payload["premium_traffic"] = _premium_traffic_list_payload(active_subs.get(user.user_id)) serialized.append(payload) - return _ok( - { - "users": serialized, - "page": page, - "page_size": page_size, - "total": total, - } - ) + return { + "users": serialized, + "page": page, + "page_size": page_size, + "total": total, + } + + +def _admin_users_list_cache(settings: Settings) -> Optional[AsyncTTLCache]: + ttl_seconds = int(getattr(settings, "ADMIN_USERS_LIST_CACHE_TTL_SECONDS", 3) or 0) + if ttl_seconds <= 0: + return None + cache_key = (id(settings), ttl_seconds) + cache = _ADMIN_USERS_LIST_CACHES.get(cache_key) + if cache is None: + cache = AsyncTTLCache( + ttl_seconds=ttl_seconds, + settings=settings, + namespace="admin:users_list", + ) + _ADMIN_USERS_LIST_CACHES[cache_key] = cache + return cache + + +def _admin_users_list_cache_key(**params: Any) -> str: + raw = json.dumps(params, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +async def _invalidate_admin_users_list_cache(settings: Settings) -> None: + for settings_id, _ttl in tuple(_ADMIN_USERS_LIST_CACHES): + if settings_id == id(settings): + _ADMIN_USERS_LIST_CACHES[(settings_id, _ttl)].invalidate() + try: + await cache_delete_pattern(settings, redis_key(settings, "cache", "admin:users_list", "*")) + except Exception: + return + + +async def _invalidate_after_admin_user_mutation( + settings: Settings, + user_id: Optional[int] = None, + *, + include_devices: bool = True, +) -> None: + await _invalidate_admin_users_list_cache(settings) + if user_id is not None: + await invalidate_webapp_user_caches( + settings, + user_id, + include_devices=include_devices, + ) async def _bulk_user_statuses( @@ -530,6 +655,7 @@ async def admin_user_ban_route(request: web.Request) -> web.Response: payload = await _read_json(request) desired = bool(payload.get("banned")) + settings: Settings = request.app["settings"] async_session_factory: sessionmaker = request.app["async_session_factory"] async with async_session_factory() as session: user = await user_dal.get_user_by_id(session, target_id) @@ -538,6 +664,7 @@ async def admin_user_ban_route(request: web.Request) -> web.Response: user.is_banned = bool(desired) await session.commit() await session.refresh(user) + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok({"user": _serialize_user(user)}) @@ -733,6 +860,7 @@ async def admin_user_delete_route(request: web.Request) -> web.Response: actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] async_session_factory: sessionmaker = request.app["async_session_factory"] async with async_session_factory() as session: ok = await user_dal.delete_user_and_relations(session, target_id) @@ -749,12 +877,14 @@ async def admin_user_delete_route(request: web.Request) -> web.Response: }, ) await session.commit() + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok({}) async def admin_user_reset_trial_route(request: web.Request) -> web.Response: actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] panel_service = request.app.get("panel_service") subscription_service = request.app.get("subscription_service") if panel_service is None or subscription_service is None: @@ -781,6 +911,7 @@ async def admin_user_reset_trial_route(request: web.Request) -> web.Response: }, ) await session.commit() + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok({}) @@ -788,6 +919,7 @@ async def admin_user_premium_override_route(request: web.Request) -> web.Respons """Premium-squad traffic overrides only (unlimited toggle + bonus GB).""" actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] payload = await _read_json(request) subscription_service = request.app.get("subscription_service") @@ -839,6 +971,7 @@ async def admin_user_premium_override_route(request: web.Request) -> web.Respons await session.commit() await session.refresh(active) + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok({"subscription": _serialize_subscription(active)}) @@ -846,6 +979,7 @@ async def admin_user_regular_traffic_override_route(request: web.Request) -> web """Main (regular) traffic: unlimited-style ceiling + admin bonus GB.""" actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] payload = await _read_json(request) unlimited = bool(payload.get("unlimited")) @@ -896,6 +1030,7 @@ async def admin_user_regular_traffic_override_route(request: web.Request) -> web await session.commit() await session.refresh(active) + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok({"subscription": _serialize_subscription(active)}) @@ -910,6 +1045,7 @@ async def admin_user_traffic_grant_route(request: web.Request) -> web.Response: """ actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] payload = await _read_json(request) kind = str(payload.get("kind") or "regular").strip().lower() @@ -970,6 +1106,7 @@ async def admin_user_traffic_grant_route(request: web.Request) -> web.Response: refreshed = await subscription_dal.get_active_subscription_by_user_id(session, target_id) + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok( { "subscription": _serialize_subscription(refreshed) if refreshed else None, @@ -985,6 +1122,7 @@ async def admin_user_traffic_grant_route(request: web.Request) -> web.Response: async def admin_user_extend_route(request: web.Request) -> web.Response: actor_id = _require_admin_user_id(request) target_id = int(request.match_info["user_id"]) + settings: Settings = request.app["settings"] payload = await _read_json(request) try: days = int(payload.get("days") or 0) @@ -1023,6 +1161,7 @@ async def admin_user_extend_route(request: web.Request) -> web.Response: refreshed = await subscription_dal.get_active_subscription_by_user_id(session, target_id) + await _invalidate_after_admin_user_mutation(settings, target_id) return _ok( { "subscription": _serialize_subscription(refreshed) if refreshed else None, diff --git a/backend/bot/app/web/webapp/account.py b/backend/bot/app/web/webapp/account.py index 826829e..67259ec 100644 --- a/backend/bot/app/web/webapp/account.py +++ b/backend/bot/app/web/webapp/account.py @@ -1,5 +1,7 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 + +from bot.app.web.webapp.cache_helpers import webapp_cached_user_payload from .auth import _hash_email_password from .common import _invalidate_webapp_user_caches @@ -433,12 +435,13 @@ async def account_telegram_link_route(request: web.Request) -> web.Response: async def me_route(request: web.Request) -> web.Response: user_id = _require_user_id(request) settings: Settings = request.app["settings"] - cache_key = redis_key(settings, "cache", "webapp", "me", user_id) - cached = await cache_get_json(settings, cache_key) - if cached: - return web.json_response({"ok": True, **cached}) - data = await _build_user_payload(request, user_id) - await cache_set_json(settings, cache_key, data, settings.WEBAPP_ME_CACHE_TTL_SECONDS) + data = await webapp_cached_user_payload( + settings, + "me", + user_id, + int(getattr(settings, "WEBAPP_ME_CACHE_TTL_SECONDS", 15) or 0), + lambda: _build_user_payload(request, user_id), + ) return web.json_response({"ok": True, **data}) diff --git a/backend/bot/app/web/webapp/assets.py b/backend/bot/app/web/webapp/assets.py index db47a3d..be63e0e 100644 --- a/backend/bot/app/web/webapp/assets.py +++ b/backend/bot/app/web/webapp/assets.py @@ -9,6 +9,11 @@ from config.webapp_themes_config import ( public_themes_catalog_payload, ) +_TEXT_FILE_CACHE: Dict[tuple[str, bool], tuple[int, int, str]] = {} +_ASSET_NAME_CACHE: Dict[tuple[str, str], tuple[float, str]] = {} +_I18N_PAYLOAD_CACHE: Dict[tuple[int, str, tuple[tuple[str, int, int], ...]], Dict[str, Any]] = {} +_ASSET_NAME_CACHE_TTL_SECONDS = 30.0 + async def health_route(request: web.Request) -> web.Response: return web.json_response({"ok": True}) @@ -910,11 +915,28 @@ def _normalize_i18n_scope(raw_scope: object) -> str: return scope if scope in WEBAPP_I18N_SCOPES else "webapp" +def _i18n_cache_fingerprint( + locales_data: Dict[str, Any], +) -> tuple[tuple[str, int, int], ...]: + return tuple( + sorted( + (str(lang), id(messages), len(messages)) + for lang, messages in locales_data.items() + if isinstance(messages, dict) + ) + ) + + def _filter_webapp_i18n_payload(locales_data: object, scope: str = "webapp") -> Dict[str, Any]: if not isinstance(locales_data, dict): return {} normalized_scope = _normalize_i18n_scope(scope) + cache_key = (id(locales_data), normalized_scope, _i18n_cache_fingerprint(locales_data)) + cached = _I18N_PAYLOAD_CACHE.get(cache_key) + if cached is not None: + return cached + payload: Dict[str, Any] = {} for lang, messages in locales_data.items(): if not isinstance(messages, dict): @@ -928,6 +950,9 @@ def _filter_webapp_i18n_payload(locales_data: object, scope: str = "webapp") -> ): filtered[key_text] = value payload[str(lang)] = filtered + if len(_I18N_PAYLOAD_CACHE) > 32: + _I18N_PAYLOAD_CACHE.clear() + _I18N_PAYLOAD_CACHE[cache_key] = payload return payload @@ -1005,7 +1030,7 @@ async def index_route(request: web.Request) -> web.Response: if not settings.WEBAPP_ENABLED: raise web.HTTPNotFound(text="webapp_disabled") - html = TEMPLATE_PATH.read_text(encoding="utf-8") + html = _read_template_text_cached(TEMPLATE_PATH) cached = _get_cached_webapp_settings(request) themes_catalog = settings.webapp_themes_catalog primary_color = settings.WEBAPP_PRIMARY_COLOR or "#00fe7a" @@ -1082,6 +1107,17 @@ async def _serve_template_asset( raise web.HTTPNotFound(text="webapp_disabled") path = ASSET_DIR / filename + text = _read_template_text_cached(path, strip_dev_mock=strip_dev_mock) + return web.Response(text=text, content_type=content_type, charset="utf-8") + + +def _read_template_text_cached(path: Path, *, strip_dev_mock: bool = False) -> str: + stat = path.stat() + key = (str(path.resolve()), strip_dev_mock) + cached = _TEXT_FILE_CACHE.get(key) + if cached and cached[0] == stat.st_mtime_ns and cached[1] == stat.st_size: + return cached[2] + text = path.read_text(encoding="utf-8") if strip_dev_mock: text = _strip_marked_block( @@ -1089,10 +1125,17 @@ async def _serve_template_asset( "/* WEBAPP_DEV_MOCK_START */", "/* WEBAPP_DEV_MOCK_END */", ) - return web.Response(text=text, content_type=content_type, charset="utf-8") + _TEXT_FILE_CACHE[key] = (stat.st_mtime_ns, stat.st_size, text) + if len(_TEXT_FILE_CACHE) > 24: + _TEXT_FILE_CACHE.clear() + _TEXT_FILE_CACHE[key] = (stat.st_mtime_ns, stat.st_size, text) + return text def _resolve_webapp_js_asset_name() -> str: + cached = _get_cached_asset_name("js") + if cached: + return cached minified_assets = [] for path in ASSET_DIR.glob("subscription_webapp.min.*.js"): try: @@ -1101,11 +1144,14 @@ def _resolve_webapp_js_asset_name() -> str: continue if minified_assets: minified_assets.sort(reverse=True) - return minified_assets[0][1] - return "subscription_webapp.js" + return _set_cached_asset_name("js", minified_assets[0][1]) + return _set_cached_asset_name("js", "subscription_webapp.js") def _resolve_webapp_css_asset_name() -> str: + cached = _get_cached_asset_name("css") + if cached: + return cached hashed_assets = [] for path in ASSET_DIR.glob("subscription_webapp.*.css"): if not re.fullmatch(r"subscription_webapp\.[0-9a-f]{8}\.css", path.name): @@ -1116,8 +1162,25 @@ def _resolve_webapp_css_asset_name() -> str: continue if hashed_assets: hashed_assets.sort(reverse=True) - return hashed_assets[0][1] - return "subscription_webapp.css" + return _set_cached_asset_name("css", hashed_assets[0][1]) + return _set_cached_asset_name("css", "subscription_webapp.css") + + +def _get_cached_asset_name(kind: str) -> Optional[str]: + key = (str(ASSET_DIR.resolve()), kind) + cached = _ASSET_NAME_CACHE.get(key) + if not cached: + return None + cached_at, filename = cached + if time.monotonic() - cached_at >= _ASSET_NAME_CACHE_TTL_SECONDS: + return None + return filename + + +def _set_cached_asset_name(kind: str, filename: str) -> str: + key = (str(ASSET_DIR.resolve()), kind) + _ASSET_NAME_CACHE[key] = (time.monotonic(), filename) + return filename _INITIAL_THEME_TOKEN_CSS_MAP = { diff --git a/backend/bot/app/web/webapp/cache_helpers.py b/backend/bot/app/web/webapp/cache_helpers.py new file mode 100644 index 0000000..65ef57d --- /dev/null +++ b/backend/bot/app/web/webapp/cache_helpers.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from typing import Any, Awaitable, Callable, Optional + +from bot.infra.redis import cache_delete, redis_key +from bot.utils.ttl_cache import AsyncTTLCache +from config.settings import Settings + +_WEBAPP_USER_PAYLOAD_CACHES: dict[tuple[int, str, int], AsyncTTLCache] = {} + + +def _webapp_user_payload_cache( + settings: Settings, + namespace: str, + ttl_seconds: int, +) -> Optional[AsyncTTLCache]: + ttl = max(0, int(ttl_seconds or 0)) + if ttl <= 0: + return None + cache_key = (id(settings), namespace, ttl) + cache = _WEBAPP_USER_PAYLOAD_CACHES.get(cache_key) + if cache is None: + cache = AsyncTTLCache( + ttl_seconds=ttl, + settings=settings, + namespace=f"webapp:{namespace}", + ) + _WEBAPP_USER_PAYLOAD_CACHES[cache_key] = cache + return cache + + +async def webapp_cached_user_payload( + settings: Settings, + namespace: str, + user_id: int, + ttl_seconds: int, + loader: Callable[[], Awaitable[Any]], +) -> Any: + cache = _webapp_user_payload_cache(settings, namespace, ttl_seconds) + if cache is None: + return await loader() + return await cache.get_or_load(str(int(user_id)), loader) + + +def invalidate_local_webapp_user_payload( + settings: Settings, + namespace: str, + user_id: int, +) -> None: + key = str(int(user_id)) + for (settings_id, cache_namespace, _ttl), cache in tuple( + _WEBAPP_USER_PAYLOAD_CACHES.items() + ): + if settings_id == id(settings) and cache_namespace == namespace: + cache.invalidate(key) + + +async def invalidate_webapp_user_caches( + settings: Settings, + *user_ids: Optional[int], + include_devices: bool = False, +) -> None: + keys: list[str] = [] + seen: set[int] = set() + for raw_user_id in user_ids: + if raw_user_id is None: + continue + try: + user_id = int(raw_user_id) + except (TypeError, ValueError): + continue + if user_id in seen: + continue + seen.add(user_id) + keys.append(redis_key(settings, "cache", "webapp", "me", user_id)) + invalidate_local_webapp_user_payload(settings, "me", user_id) + if include_devices: + keys.append(redis_key(settings, "cache", "webapp", "devices", user_id)) + invalidate_local_webapp_user_payload(settings, "devices", user_id) + if keys: + await cache_delete(settings, *keys) diff --git a/backend/bot/app/web/webapp/common.py b/backend/bot/app/web/webapp/common.py index 459724a..f55217f 100644 --- a/backend/bot/app/web/webapp/common.py +++ b/backend/bot/app/web/webapp/common.py @@ -1,6 +1,10 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 +from bot.app.web.webapp.cache_helpers import ( + invalidate_local_webapp_user_payload, +) + async def _read_json(request: web.Request) -> Dict[str, Any]: try: @@ -35,8 +39,10 @@ async def _invalidate_webapp_user_caches( continue seen.add(user_id) keys.append(redis_key(settings, "cache", "webapp", "me", user_id)) + invalidate_local_webapp_user_payload(settings, "me", user_id) if include_devices: keys.append(redis_key(settings, "cache", "webapp", "devices", user_id)) + invalidate_local_webapp_user_payload(settings, "devices", user_id) if keys: await cache_delete(settings, *keys) diff --git a/backend/bot/app/web/webapp/devices.py b/backend/bot/app/web/webapp/devices.py index 88f00d1..6d72641 100644 --- a/backend/bot/app/web/webapp/devices.py +++ b/backend/bot/app/web/webapp/devices.py @@ -1,6 +1,8 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 +from bot.app.web.webapp.cache_helpers import webapp_cached_user_payload + async def devices_route(request: web.Request) -> web.Response: user_id = _require_user_id(request) @@ -15,44 +17,80 @@ async def devices_route(request: web.Request) -> web.Response: if not db_user or db_user.is_banned: return _json_error(403, "access_denied", "Access denied") - cache_key = redis_key(settings, "cache", "webapp", "devices", user_id) - cached = await cache_get_json(settings, cache_key) - if isinstance(cached, dict): - return web.json_response({"ok": True, **cached}) + result = await webapp_cached_user_payload( + settings, + "devices", + user_id, + int(getattr(settings, "WEBAPP_DEVICES_CACHE_TTL_SECONDS", 5) or 0), + lambda: _load_devices_payload(subscription_service, session, user_id), + ) + if isinstance(result, dict) and result.get("ok") is True: + return web.json_response({"ok": True, **(result.get("payload") or {})}) + if isinstance(result, dict) and not result.get("error"): + # Backward-compatible with payloads written by older versions under + # the same Redis cache key. + return web.json_response({"ok": True, **result}) + if not isinstance(result, dict): + result = {} + if not result.get("ok"): + return _json_error( + int(result.get("status") or 500), + str(result.get("error") or "devices_load_failed"), + str(result.get("message") or "Failed to load devices"), + ) + return web.json_response({"ok": True, **(result.get("payload") or {})}) - active = await subscription_service.get_active_subscription_details(session, user_id) - panel_user_uuid = active.get("user_id") if active else None - if not panel_user_uuid: - return _json_error(400, "subscription_not_active", "Subscription is not active") - panel_service = getattr(subscription_service, "panel_service", None) - if not panel_service: - return _json_error(503, "panel_unavailable", "Panel service unavailable") +async def _load_devices_payload( + subscription_service: SubscriptionService, + session: AsyncSession, + user_id: int, +) -> Dict[str, Any]: + active = await subscription_service.get_active_subscription_details(session, user_id) + panel_user_uuid = active.get("user_id") if active else None + if not panel_user_uuid: + return { + "ok": False, + "status": 400, + "error": "subscription_not_active", + "message": "Subscription is not active", + } - try: - devices_response = await panel_service.get_user_devices(panel_user_uuid) - except Exception: - logger.exception("Failed to load WebApp devices for user %s", user_id) - return _json_error(502, "devices_load_failed", "Failed to load devices") + panel_service = getattr(subscription_service, "panel_service", None) + if not panel_service: + return { + "ok": False, + "status": 503, + "error": "panel_unavailable", + "message": "Panel service unavailable", + } + + try: + devices_response = await panel_service.get_user_devices(panel_user_uuid) + except Exception: + logger.exception("Failed to load WebApp devices for user %s", user_id) + return { + "ok": False, + "status": 502, + "error": "devices_load_failed", + "message": "Failed to load devices", + } devices = _normalize_devices_response(devices_response) max_devices = _coerce_int_or_none(active.get("max_devices")) if active else None - payload = { - "enabled": True, - "current_devices": len(devices), - "max_devices": max_devices, - "max_devices_label": _format_devices_limit(max_devices), - "devices": [ - _serialize_device(device, index) for index, device in enumerate(devices, start=1) - ], + return { + "ok": True, + "payload": { + "enabled": True, + "current_devices": len(devices), + "max_devices": max_devices, + "max_devices_label": _format_devices_limit(max_devices), + "devices": [ + _serialize_device(device, index) + for index, device in enumerate(devices, start=1) + ], + }, } - await cache_set_json( - settings, - cache_key, - payload, - max(1, int(getattr(settings, "WEBAPP_DEVICES_CACHE_TTL_SECONDS", 5) or 5)), - ) - return web.json_response({"ok": True, **payload}) async def disconnect_device_route(request: web.Request) -> web.Response: diff --git a/backend/bot/utils/ttl_cache.py b/backend/bot/utils/ttl_cache.py index 278e492..18408fd 100644 --- a/backend/bot/utils/ttl_cache.py +++ b/backend/bot/utils/ttl_cache.py @@ -15,6 +15,7 @@ class AsyncTTLCache: self.namespace = namespace self._data: Dict[str, Tuple[float, Any]] = {} self._locks: Dict[str, asyncio.Lock] = {} + self._inflight: Dict[str, asyncio.Task] = {} def _is_fresh(self, expires_at: float) -> bool: return time.monotonic() < expires_at @@ -46,37 +47,49 @@ class AsyncTTLCache: cached = self.get_fresh(key) if cached is not None: return cached + task = self._inflight.get(key) + if task is None: + task = asyncio.create_task(self._load_and_store(key, loader)) + self._inflight[key] = task - cache_key = None - if self.settings is not None and self.namespace: + def _forget_inflight(done_task: asyncio.Task) -> None: + if self._inflight.get(key) is done_task: + self._inflight.pop(key, None) + + task.add_done_callback(_forget_inflight) + return await task + + async def _load_and_store(self, key: str, loader: Callable[[], Awaitable[Any]]) -> Any: + cache_key = None + if self.settings is not None and self.namespace: + try: + from bot.infra.redis import cache_get_json, redis_key + + cache_key = redis_key(self.settings, "cache", self.namespace, key) + cached = await cache_get_json(self.settings, cache_key) + if cached is not None: + if self._is_cacheable(cached): + self._data[key] = (time.monotonic() + self.ttl_seconds, cached) + return cached + except Exception: + cache_key = None + + value = await loader() + if self._is_cacheable(value): + self._data[key] = (time.monotonic() + self.ttl_seconds, value) + if cache_key is not None: try: - from bot.infra.redis import cache_get_json, redis_key + from bot.infra.redis import cache_set_json - cache_key = redis_key(self.settings, "cache", self.namespace, key) - cached = await cache_get_json(self.settings, cache_key) - if cached is not None: - if self._is_cacheable(cached): - self._data[key] = (time.monotonic() + self.ttl_seconds, cached) - return cached + await cache_set_json( + self.settings, + cache_key, + value, + max(1, int(self.ttl_seconds)), + ) except Exception: - cache_key = None - - value = await loader() - if self._is_cacheable(value): - self._data[key] = (time.monotonic() + self.ttl_seconds, value) - if cache_key is not None: - try: - from bot.infra.redis import cache_set_json - - await cache_set_json( - self.settings, - cache_key, - value, - max(1, int(self.ttl_seconds)), - ) - except Exception: - pass - return value + pass + return value def invalidate(self, key: Optional[str] = None) -> None: if key is None: diff --git a/backend/config/settings.py b/backend/config/settings.py index b16340f..70589e1 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -102,6 +102,7 @@ class Settings(BaseSettings): PANEL_ALL_USERS_PAGE_SIZE: int = Field(default=1000) ADMIN_PANEL_STATS_CACHE_TTL_SECONDS: int = Field(default=15) ADMIN_DB_STATS_CACHE_TTL_SECONDS: int = Field(default=5) + ADMIN_USERS_LIST_CACHE_TTL_SECONDS: int = Field(default=3) PROFILE_SYNC_CACHE_TTL_SECONDS: int = Field(default=900) PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS: int = Field(default=3600) PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES: int = Field(default=104857600)