austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

perf: cache broadcast audience counts

Browse Source
This commit is contained in:
3252a8
2026-06-07 23:15:44 +03:00
parent d1c4a6de80
commit 4b8f939a25
7 changed files with 244 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/app/web/admin_api_impl/broadcast.py
+89 -18
View File
@@ -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})
backend/config/settings.py
+1
View File
@@ -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)
backend/db/dal/user_dal.py
+73
View File
@@ -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
docs/configuration/env-vars.md
+1
View File
@@ -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-трафика для более ранней записи. |
frontend/src/admin/AdminPanel.svelte
+1
View File
@@ -518,6 +518,7 @@
if (typeof window !== "undefined") {
window.addEventListener("popstate", onPopState);
}
void broadcastStore.loadCounts();
return () => {
if (motionMql) motionMql.removeEventListener("change", onMotionChange);
if (compactMql) {
frontend/src/admin/sections/BroadcastSection.svelte
+11 -3
View File
@@ -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(() => {
frontend/src/lib/admin/stores/broadcastStore.js
+68 -7
View File
@@ -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 = "";