From 4b8f939a2552e58073dde440d56f1a6fc4097ba2 Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Sun, 7 Jun 2026 23:15:44 +0300 Subject: [PATCH] perf: cache broadcast audience counts --- .../bot/app/web/admin_api_impl/broadcast.py | 107 +++++++++++++++--- backend/config/settings.py | 1 + backend/db/dal/user_dal.py | 73 ++++++++++++ docs/configuration/env-vars.md | 1 + frontend/src/admin/AdminPanel.svelte | 1 + .../admin/sections/BroadcastSection.svelte | 14 ++- .../src/lib/admin/stores/broadcastStore.js | 75 ++++++++++-- 7 files changed, 244 insertions(+), 28 deletions(-) diff --git a/backend/bot/app/web/admin_api_impl/broadcast.py b/backend/bot/app/web/admin_api_impl/broadcast.py index 3c7c094..986988a 100644 --- a/backend/bot/app/web/admin_api_impl/broadcast.py +++ b/backend/bot/app/web/admin_api_impl/broadcast.py @@ -5,6 +5,8 @@ from .common import _panel_user_connection_activity import asyncio from collections import defaultdict +from bot.utils.ttl_cache import AsyncTTLCache + BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED = "active_never_connected" BROADCAST_TARGETS = { @@ -16,6 +18,7 @@ BROADCAST_TARGETS = { BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED, } PANEL_ACTIVITY_LOOKUP_CONCURRENCY = 10 +_ADMIN_BROADCAST_AUDIENCE_COUNT_CACHES: Dict[tuple[int, int], AsyncTTLCache] = {} def _resolve_panel_service(request: web.Request) -> Any: @@ -73,14 +76,92 @@ async def _user_ids_with_active_subscription_never_connected( async with semaphore: return await _panel_connection_status(panel_service, panel_uuid) + panel_uuids = list( + dict.fromkeys( + panel_uuid + for user_panel_uuids in panel_uuids_by_user.values() + for panel_uuid in user_panel_uuids + ) + ) + statuses_by_uuid = dict( + zip( + panel_uuids, + await asyncio.gather(*(lookup(uuid) for uuid in panel_uuids)), + ) + ) + user_ids: List[int] = [] for user_id, panel_uuids in panel_uuids_by_user.items(): - statuses = await asyncio.gather(*(lookup(panel_uuid) for panel_uuid in panel_uuids)) + statuses = [statuses_by_uuid.get(panel_uuid, "unknown") for panel_uuid in panel_uuids] if statuses and all(status == "never" for status in statuses): user_ids.append(user_id) return user_ids +def _admin_broadcast_audience_counts_cache(settings: Settings) -> Optional[AsyncTTLCache]: + ttl_seconds = int( + getattr(settings, "ADMIN_BROADCAST_AUDIENCE_COUNTS_CACHE_TTL_SECONDS", 30) or 0 + ) + if ttl_seconds <= 0: + return None + cache_key = (id(settings), ttl_seconds) + cache = _ADMIN_BROADCAST_AUDIENCE_COUNT_CACHES.get(cache_key) + if cache is None: + cache = AsyncTTLCache( + ttl_seconds=ttl_seconds, + settings=settings, + namespace="admin:broadcast_audience_counts", + ) + _ADMIN_BROADCAST_AUDIENCE_COUNT_CACHES[cache_key] = cache + return cache + + +async def _load_broadcast_audience_counts( + settings: Settings, + async_session_factory: sessionmaker, + panel_service: Any, +) -> Dict[str, Optional[int]]: + cache = _admin_broadcast_audience_counts_cache(settings) + if cache is None: + return await _load_broadcast_audience_counts_uncached( + async_session_factory, + panel_service, + ) + cache_key = "with-panel" if panel_service is not None else "without-panel" + return await cache.get_or_load( + cache_key, + lambda: _load_broadcast_audience_counts_uncached( + async_session_factory, + panel_service, + ), + ) + + +async def _load_broadcast_audience_counts_uncached( + async_session_factory: sessionmaker, + panel_service: Any, +) -> Dict[str, Optional[int]]: + async with async_session_factory() as session: + counts: Dict[str, Optional[int]] = { + "all": await user_dal.count_all_active_users_for_broadcast(session), + "active": await user_dal.count_users_with_active_subscription_for_broadcast(session), + "inactive": await user_dal.count_users_without_active_subscription_for_broadcast( + session + ), + "expired": await user_dal.count_users_with_expired_subscription_for_broadcast(session), + "never": await user_dal.count_users_without_any_subscription_for_broadcast(session), + BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED: None, + } + if panel_service is not None: + counts[BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED] = len( + await _user_ids_with_active_subscription_never_connected( + session, + panel_service, + ) + ) + return counts + + async def admin_broadcast_route(request: web.Request) -> web.Response: actor_id = _require_admin_user_id(request) payload = await _read_json(request) @@ -149,23 +230,13 @@ async def admin_broadcast_audience_counts_route(request: web.Request) -> web.Res """Return how many users each broadcast audience currently resolves to.""" _require_admin_user_id(request) + settings: Settings = request.app["settings"] async_session_factory: sessionmaker = request.app["async_session_factory"] - async with async_session_factory() as session: - panel_service = _resolve_panel_service(request) - counts = { - "all": len(await user_dal.get_all_active_user_ids_for_broadcast(session)), - "active": len(await user_dal.get_user_ids_with_active_subscription(session)), - "inactive": len(await user_dal.get_user_ids_without_active_subscription(session)), - "expired": len(await user_dal.get_user_ids_with_expired_subscription(session)), - "never": len(await user_dal.get_user_ids_without_any_subscription(session)), - BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED: None, - } - if panel_service is not None: - counts[BROADCAST_TARGET_ACTIVE_NEVER_CONNECTED] = len( - await _user_ids_with_active_subscription_never_connected( - session, - panel_service, - ) - ) + panel_service = _resolve_panel_service(request) + counts = await _load_broadcast_audience_counts( + settings, + async_session_factory, + panel_service, + ) return _ok({"counts": counts}) diff --git a/backend/config/settings.py b/backend/config/settings.py index 43795a3..fc05dd6 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -215,6 +215,7 @@ class Settings(BaseSettings): 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) + ADMIN_BROADCAST_AUDIENCE_COUNTS_CACHE_TTL_SECONDS: int = Field(default=30) 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) diff --git a/backend/db/dal/user_dal.py b/backend/db/dal/user_dal.py index 0acc533..6467d89 100644 --- a/backend/db/dal/user_dal.py +++ b/backend/db/dal/user_dal.py @@ -715,6 +715,12 @@ async def get_all_active_user_ids_for_broadcast(session: AsyncSession) -> List[i return result.scalars().all() +async def count_all_active_users_for_broadcast(session: AsyncSession) -> int: + stmt = select(func.count(User.user_id)).where(User.is_banned == False) + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + async def get_all_users_with_panel_uuid(session: AsyncSession) -> List[User]: stmt = select(User).where(User.panel_user_uuid.is_not(None)) result = await session.execute(stmt) @@ -890,6 +896,27 @@ async def get_user_ids_with_active_subscription(session: AsyncSession) -> List[i return result.scalars().all() +async def count_users_with_active_subscription_for_broadcast(session: AsyncSession) -> int: + """Count non-banned users who have any active subscription.""" + from datetime import datetime, timezone + + now = datetime.now(timezone.utc) + + stmt = ( + select(func.count(func.distinct(Subscription.user_id))) + .join(User, Subscription.user_id == User.user_id) + .where( + and_( + User.is_banned == False, + Subscription.is_active == True, + Subscription.end_date > now, + ) + ) + ) + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + async def get_user_ids_without_active_subscription(session: AsyncSession) -> List[int]: """Return non-banned user IDs who do NOT have any active subscription.""" from datetime import datetime, timezone @@ -919,6 +946,20 @@ async def get_user_ids_without_active_subscription(session: AsyncSession) -> Lis return result.scalars().all() +async def count_users_without_active_subscription_for_broadcast(session: AsyncSession) -> int: + """Count non-banned users who do NOT have any active subscription.""" + from datetime import datetime, timezone + + now = datetime.now(timezone.utc) + + stmt = select(func.count(User.user_id)).where( + User.is_banned == False, + ~_active_subscription_exists_for_user(now), + ) + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + async def get_user_ids_without_any_subscription(session: AsyncSession) -> List[int]: """Return non-banned user IDs who never had any subscription or trial. @@ -942,6 +983,24 @@ async def get_user_ids_without_any_subscription(session: AsyncSession) -> List[i return result.scalars().all() +async def count_users_without_any_subscription_for_broadcast(session: AsyncSession) -> int: + """Count non-banned users who never had any subscription or trial.""" + any_sub = aliased(Subscription) + + stmt = ( + select(func.count(User.user_id)) + .outerjoin(any_sub, any_sub.user_id == User.user_id) + .where( + and_( + User.is_banned == False, + any_sub.user_id.is_(None), + ) + ) + ) + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + def _expired_subscription_exists_for_user(now: datetime): expired_subs = aliased(Subscription) normalized_status = func.lower(func.coalesce(expired_subs.status_from_panel, "")) @@ -988,6 +1047,20 @@ async def count_users_with_expired_subscription(session: AsyncSession) -> int: return int(result.scalar_one() or 0) +async def count_users_with_expired_subscription_for_broadcast(session: AsyncSession) -> int: + """Count non-banned users with an expired subscription and no active one.""" + from datetime import datetime, timezone + + now = datetime.now(timezone.utc) + stmt = select(func.count(User.user_id)).where( + User.is_banned == False, + _expired_subscription_exists_for_user(now), + ~_active_subscription_exists_for_user(now), + ) + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + async def get_user_ids_with_expired_subscription(session: AsyncSession) -> List[int]: """Return non-banned user IDs with an expired subscription and no active one.""" from datetime import datetime, timezone diff --git a/docs/configuration/env-vars.md b/docs/configuration/env-vars.md index b2e43f2..47f5656 100644 --- a/docs/configuration/env-vars.md +++ b/docs/configuration/env-vars.md @@ -66,6 +66,7 @@ | `ADMIN_PANEL_STATS_CACHE_TTL_SECONDS` | TTL статистики Remnawave в админке. | | `ADMIN_DB_STATS_CACHE_TTL_SECONDS` | TTL дорогих DB-агрегатов админки. | | `ADMIN_USERS_LIST_CACHE_TTL_SECONDS` | TTL списка пользователей админки. | +| `ADMIN_BROADCAST_AUDIENCE_COUNTS_CACHE_TTL_SECONDS` | TTL счетчиков целевых групп рассылки в админке. | | `PROFILE_SYNC_CACHE_TTL_SECONDS` | Минимальная пауза между sync Telegram-профиля пользователя. | | `PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS` | Минимальная пауза записи lifetime-трафика. | | `PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES` | Дельта lifetime-трафика для более ранней записи. | diff --git a/frontend/src/admin/AdminPanel.svelte b/frontend/src/admin/AdminPanel.svelte index b81b27f..df27457 100644 --- a/frontend/src/admin/AdminPanel.svelte +++ b/frontend/src/admin/AdminPanel.svelte @@ -518,6 +518,7 @@ if (typeof window !== "undefined") { window.addEventListener("popstate", onPopState); } + void broadcastStore.loadCounts(); return () => { if (motionMql) motionMql.removeEventListener("change", onMotionChange); if (compactMql) { diff --git a/frontend/src/admin/sections/BroadcastSection.svelte b/frontend/src/admin/sections/BroadcastSection.svelte index 849bac0..02b3b61 100644 --- a/frontend/src/admin/sections/BroadcastSection.svelte +++ b/frontend/src/admin/sections/BroadcastSection.svelte @@ -8,15 +8,23 @@ export let at; const broadcastStore = getContext("broadcastStore"); - $: ({ broadcastTarget, broadcastText, broadcastBusy, broadcastResult, broadcastCounts } = - $broadcastStore); + $: ({ + broadcastTarget, + broadcastText, + broadcastBusy, + broadcastResult, + broadcastCounts, + broadcastCountsLoading, + } = $broadcastStore); const BROADCAST_TARGET_OPTIONS = broadcastStore.BROADCAST_TARGET_OPTIONS; // Append the resolved audience size to each option once counts are loaded. $: targetOptions = BROADCAST_TARGET_OPTIONS.map((option) => { const count = broadcastCounts?.[option.value]; - return count == null ? option : { ...option, label: `${option.label} (${count})` }; + if (count != null) return { ...option, label: `${option.label} (${count})` }; + if (broadcastCountsLoading) return { ...option, label: `${option.label} (...)` }; + return option; }); onMount(() => { diff --git a/frontend/src/lib/admin/stores/broadcastStore.js b/frontend/src/lib/admin/stores/broadcastStore.js index e431be1..8547879 100644 --- a/frontend/src/lib/admin/stores/broadcastStore.js +++ b/frontend/src/lib/admin/stores/broadcastStore.js @@ -1,12 +1,20 @@ import { writable } from "svelte/store"; export function createBroadcastStore({ api, onToast, at }) { + const COUNTS_CACHE_TTL_MS = 30_000; + const COUNTS_DISPLAY_CACHE_TTL_MS = 5 * 60_000; + const COUNTS_STORAGE_KEY = "remnawave-admin:broadcast-audience-counts"; + let countsPromise = null; + const cachedCounts = readStoredCounts(); + const state = writable({ broadcastTarget: "all", broadcastText: "", broadcastBusy: false, broadcastResult: null, - broadcastCounts: null, + broadcastCounts: cachedCounts?.counts || null, + broadcastCountsLoading: false, + broadcastCountsLoadedAt: cachedCounts?.loadedAt || 0, }); const BROADCAST_TARGET_OPTIONS = [ @@ -28,17 +36,70 @@ export function createBroadcastStore({ api, onToast, at }) { }, ]; - async function loadCounts() { + function countsAreFresh(stateSnapshot) { + return ( + stateSnapshot.broadcastCounts && + Date.now() - Number(stateSnapshot.broadcastCountsLoadedAt || 0) < COUNTS_CACHE_TTL_MS + ); + } + + function readStoredCounts() { try { - const res = await api("/admin/broadcast/audience-counts"); - if (res?.ok && res.counts) { - state.update((s) => ({ ...s, broadcastCounts: res.counts })); - } + if (typeof window === "undefined" || !window.sessionStorage) return null; + const raw = window.sessionStorage.getItem(COUNTS_STORAGE_KEY); + if (!raw) return null; + const payload = JSON.parse(raw); + const loadedAt = Number(payload?.loadedAt || 0); + if (!payload?.counts || Date.now() - loadedAt > COUNTS_DISPLAY_CACHE_TTL_MS) return null; + return { counts: payload.counts, loadedAt }; } catch { - // Counts are advisory; ignore failures and keep plain labels. + return null; } } + function writeStoredCounts(counts, loadedAt) { + try { + if (typeof window === "undefined" || !window.sessionStorage) return; + window.sessionStorage.setItem(COUNTS_STORAGE_KEY, JSON.stringify({ counts, loadedAt })); + } catch { + // Ignore storage quota/privacy errors; in-memory counts still work. + } + } + + async function loadCounts({ force = false } = {}) { + let shouldLoad = false; + state.update((s) => { + if (!force && countsAreFresh(s)) return s; + if (countsPromise || s.broadcastCountsLoading) return s; + shouldLoad = true; + return { ...s, broadcastCountsLoading: true }; + }); + + if (!shouldLoad) return countsPromise || Promise.resolve(); + + countsPromise = (async () => { + try { + const res = await api("/admin/broadcast/audience-counts"); + if (res?.ok && res.counts) { + const loadedAt = Date.now(); + state.update((s) => ({ + ...s, + broadcastCounts: res.counts, + broadcastCountsLoadedAt: loadedAt, + })); + writeStoredCounts(res.counts, loadedAt); + } + } catch { + // Counts are advisory; ignore failures and keep existing/plain labels. + } finally { + state.update((s) => ({ ...s, broadcastCountsLoading: false })); + countsPromise = null; + } + })(); + + return countsPromise; + } + async function runBroadcast() { let text = ""; let target = "";