diff --git a/.env.example b/.env.example index 87075f1..56bf837 100644 --- a/.env.example +++ b/.env.example @@ -22,6 +22,7 @@ PANEL_DEVICES_CACHE_TTL_SECONDS=5 # PANEL_ALL_USERS_CACHE_TTL_SECONDS=5 # Short TTL for concurrent Remnawave full user scans 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 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/stats.py b/backend/bot/app/web/admin_api_impl/stats.py index 703c033..2072786 100644 --- a/backend/bot/app/web/admin_api_impl/stats.py +++ b/backend/bot/app/web/admin_api_impl/stats.py @@ -2,9 +2,12 @@ import asyncio from ._runtime import * # noqa: F403,F405 +from .auth import _require_admin_user_id +from .common import _ok, _serialize_payment from bot.utils.ttl_cache import AsyncTTLCache _ADMIN_PANEL_STATS_CACHES: Dict[tuple[int, int], AsyncTTLCache] = {} +_ADMIN_DB_STATS_CACHES: Dict[tuple[int, int], AsyncTTLCache] = {} async def admin_me_route(request: web.Request) -> web.Response: @@ -18,26 +21,7 @@ async def admin_stats_route(request: web.Request) -> web.Response: settings: Settings = request.app["settings"] async_session_factory: sessionmaker = request.app["async_session_factory"] - async with async_session_factory() as session: - user_stats = await user_dal.get_enhanced_user_statistics(session) - financial_stats = await payment_dal.get_financial_statistics(session) - sync_status = await panel_sync_dal.get_panel_sync_status(session) - recent_payments = await payment_dal.get_recent_payment_logs_with_user(session, limit=10) - - payload = { - "users": user_stats, - "financial": financial_stats, - "panel_sync": { - "status": sync_status.status if sync_status else "never_run", - "last_sync_time": sync_status.last_sync_time.isoformat() - if sync_status and sync_status.last_sync_time - else None, - "details": sync_status.details if sync_status else None, - "users_processed": sync_status.users_processed_from_panel if sync_status else 0, - "subscriptions_synced": sync_status.subscriptions_synced if sync_status else 0, - }, - "recent_payments": [_serialize_payment(p) for p in recent_payments], - } + payload = dict(await _load_admin_db_stats(settings, async_session_factory)) panel_service = request.app.get("panel_service") if panel_service is not None: @@ -54,6 +38,58 @@ async def admin_stats_route(request: web.Request) -> web.Response: return _ok(payload) +async def _load_admin_db_stats( + settings: Settings, + async_session_factory: sessionmaker, +) -> Dict[str, Any]: + cache = _admin_db_stats_cache(settings) + if cache is None: + return await _load_admin_db_stats_uncached(async_session_factory) + return await cache.get_or_load( + "db", + lambda: _load_admin_db_stats_uncached(async_session_factory), + ) + + +async def _load_admin_db_stats_uncached(async_session_factory: sessionmaker) -> Dict[str, Any]: + async with async_session_factory() as session: + user_stats = await user_dal.get_enhanced_user_statistics(session) + financial_stats = await payment_dal.get_financial_statistics(session) + sync_status = await panel_sync_dal.get_panel_sync_status(session) + recent_payments = await payment_dal.get_recent_payment_logs_with_user(session, limit=10) + + return { + "users": user_stats, + "financial": financial_stats, + "panel_sync": { + "status": sync_status.status if sync_status else "never_run", + "last_sync_time": sync_status.last_sync_time.isoformat() + if sync_status and sync_status.last_sync_time + else None, + "details": sync_status.details if sync_status else None, + "users_processed": sync_status.users_processed_from_panel if sync_status else 0, + "subscriptions_synced": sync_status.subscriptions_synced if sync_status else 0, + }, + "recent_payments": [_serialize_payment(p) for p in recent_payments], + } + + +def _admin_db_stats_cache(settings: Settings) -> Optional[AsyncTTLCache]: + ttl_seconds = int(getattr(settings, "ADMIN_DB_STATS_CACHE_TTL_SECONDS", 5) or 0) + if ttl_seconds <= 0: + return None + cache_key = (id(settings), ttl_seconds) + cache = _ADMIN_DB_STATS_CACHES.get(cache_key) + if cache is None: + cache = AsyncTTLCache( + ttl_seconds=ttl_seconds, + settings=settings, + namespace="admin:db_stats", + ) + _ADMIN_DB_STATS_CACHES[cache_key] = cache + return cache + + async def _load_admin_panel_stats( request: web.Request, settings: Settings, diff --git a/backend/config/settings.py b/backend/config/settings.py index 230b76d..b16340f 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -101,6 +101,7 @@ class Settings(BaseSettings): PANEL_ALL_USERS_CACHE_TTL_SECONDS: int = Field(default=5) 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) 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/payment_dal.py b/backend/db/dal/payment_dal.py index f1bb997..6f6bdde 100644 --- a/backend/db/dal/payment_dal.py +++ b/backend/db/dal/payment_dal.py @@ -1,10 +1,10 @@ import logging from typing import Any, Dict, List, Optional -from sqlalchemy import Date, and_, cast, func +from sqlalchemy import Date, and_, case, cast, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select -from sqlalchemy.orm import selectinload +from sqlalchemy.orm import joinedload, selectinload from db.models import Payment, User @@ -80,7 +80,7 @@ async def get_payment_by_db_id(session: AsyncSession, payment_db_id: int) -> Opt stmt = ( select(Payment) .where(Payment.payment_id == payment_db_id) - .options(selectinload(Payment.user), selectinload(Payment.promo_code_used)) + .options(joinedload(Payment.user), joinedload(Payment.promo_code_used)) ) result = await session.execute(stmt) return result.scalar_one_or_none() @@ -108,7 +108,7 @@ async def get_recent_payment_logs_with_user( ) -> List[Payment]: stmt = ( select(Payment) - .options(selectinload(Payment.user)) + .options(joinedload(Payment.user)) .where(Payment.status == "succeeded") .order_by(Payment.created_at.desc()) .limit(limit) @@ -211,45 +211,31 @@ async def get_financial_statistics(session: AsyncSession) -> Dict[str, Any]: """Get comprehensive financial statistics.""" from datetime import datetime, timedelta - from sqlalchemy import and_ - now = datetime.utcnow() today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) week_start = today_start - timedelta(days=7) month_start = today_start - timedelta(days=30) - # Today's revenue - stmt_today = select(func.sum(Payment.amount)).where( - and_(Payment.status == "succeeded", Payment.created_at >= today_start) - ) - today_revenue = await session.execute(stmt_today) - today_amount = today_revenue.scalar() or 0 - - # Week revenue - stmt_week = select(func.sum(Payment.amount)).where( - and_(Payment.status == "succeeded", Payment.created_at >= week_start) - ) - week_revenue = await session.execute(stmt_week) - week_amount = week_revenue.scalar() or 0 - - # Month revenue - stmt_month = select(func.sum(Payment.amount)).where( - and_(Payment.status == "succeeded", Payment.created_at >= month_start) - ) - month_revenue = await session.execute(stmt_month) - month_amount = month_revenue.scalar() or 0 - - # All time revenue - stmt_all = select(func.sum(Payment.amount)).where(Payment.status == "succeeded") - all_revenue = await session.execute(stmt_all) - all_amount = all_revenue.scalar() or 0 - - # Count of successful payments today - stmt_count_today = select(func.count(Payment.payment_id)).where( - and_(Payment.status == "succeeded", Payment.created_at >= today_start) - ) - today_count = await session.execute(stmt_count_today) - today_payments_count = today_count.scalar() or 0 + revenue_stmt = select( + func.coalesce( + func.sum(case((Payment.created_at >= today_start, Payment.amount), else_=0)), 0 + ), + func.coalesce( + func.sum(case((Payment.created_at >= week_start, Payment.amount), else_=0)), 0 + ), + func.coalesce( + func.sum(case((Payment.created_at >= month_start, Payment.amount), else_=0)), + 0, + ), + func.coalesce(func.sum(Payment.amount), 0), + func.coalesce(func.sum(case((Payment.created_at >= today_start, 1), else_=0)), 0), + ).where(Payment.status == "succeeded") + revenue_row = (await session.execute(revenue_stmt)).one() + today_amount = revenue_row[0] or 0 + week_amount = revenue_row[1] or 0 + month_amount = revenue_row[2] or 0 + all_amount = revenue_row[3] or 0 + today_payments_count = int(revenue_row[4] or 0) # Longer tail for admin dashboard charts (presets up to 1y + custom range on the client). daily_series = await _daily_revenue_series_utc(session, days=730) diff --git a/backend/db/dal/user_dal.py b/backend/db/dal/user_dal.py index fab6a07..a65d6d7 100644 --- a/backend/db/dal/user_dal.py +++ b/backend/db/dal/user_dal.py @@ -4,7 +4,7 @@ import string from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional, Tuple -from sqlalchemy import and_, delete, desc, func, or_, update +from sqlalchemy import and_, case, delete, desc, func, or_, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select @@ -591,55 +591,46 @@ async def get_enhanced_user_statistics(session: AsyncSession) -> Dict[str, Any]: now = datetime.now(timezone.utc) today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) - # Total users - total_users_stmt = select(func.count(User.user_id)) - total_users = (await session.execute(total_users_stmt)).scalar() or 0 - - # Banned users - banned_users_stmt = select(func.count(User.user_id)).where(User.is_banned == True) - banned_users = (await session.execute(banned_users_stmt)).scalar() or 0 - - # Active users today (proxy: registered today) - active_today_stmt = select(func.count(User.user_id)).where( - User.registration_date >= today_start + user_counts_stmt = select( + func.count(User.user_id), + func.coalesce(func.sum(case((User.is_banned == True, 1), else_=0)), 0), + func.coalesce(func.sum(case((User.registration_date >= today_start, 1), else_=0)), 0), + func.coalesce(func.sum(case((User.referred_by_id.is_not(None), 1), else_=0)), 0), ) - active_today = (await session.execute(active_today_stmt)).scalar() or 0 + user_counts = (await session.execute(user_counts_stmt)).one() + total_users = int(user_counts[0] or 0) + banned_users = int(user_counts[1] or 0) + active_today = int(user_counts[2] or 0) + referral_users = int(user_counts[3] or 0) - # Users with active paid subscriptions (non-trial providers only) - paid_subs_stmt = ( - select(func.count(func.distinct(Subscription.user_id))) + subscription_counts_stmt = ( + select( + func.count( + func.distinct( + case((Subscription.provider.is_not(None), Subscription.user_id), else_=None) + ) + ), + func.count( + func.distinct( + case((Subscription.provider.is_(None), Subscription.user_id), else_=None) + ) + ), + ) .join(User, Subscription.user_id == User.user_id) .where( and_( Subscription.is_active == True, Subscription.end_date > now, - Subscription.provider.is_not(None), # Not trial ) ) ) - paid_subs_users = (await session.execute(paid_subs_stmt)).scalar() or 0 - - # Users on trial period - trial_subs_stmt = ( - select(func.count(func.distinct(Subscription.user_id))) - .join(User, Subscription.user_id == User.user_id) - .where( - and_( - Subscription.is_active == True, - Subscription.end_date > now, - Subscription.provider.is_(None), # Trial subscriptions - ) - ) - ) - trial_users = (await session.execute(trial_subs_stmt)).scalar() or 0 + subscription_counts = (await session.execute(subscription_counts_stmt)).one() + paid_subs_users = int(subscription_counts[0] or 0) + trial_users = int(subscription_counts[1] or 0) # Inactive users (no active subscription) inactive_users = total_users - paid_subs_users - trial_users - banned_users - # Users attracted via referral - referral_users_stmt = select(func.count(User.user_id)).where(User.referred_by_id.is_not(None)) - referral_users = (await session.execute(referral_users_stmt)).scalar() or 0 - return { "total_users": total_users, "banned_users": banned_users, diff --git a/scripts/perf_benchmarks.py b/scripts/perf_benchmarks.py index f8947ce..f2eb8d9 100644 --- a/scripts/perf_benchmarks.py +++ b/scripts/perf_benchmarks.py @@ -355,6 +355,99 @@ async def bench_admin_stats_cache(users: int) -> dict: } +async def bench_admin_db_stats_cache(users: int) -> dict: + admin_stats_module._ADMIN_DB_STATS_CACHES.clear() + settings = SimpleNamespace( + ADMIN_DB_STATS_CACHE_TTL_SECONDS=60, + REDIS_URL=None, + REDIS_KEY_PREFIX="bench", + ) + dal_calls = 0 + + class FakeSessionFactory: + def __call__(self): + return self + + async def __aenter__(self): + return SimpleNamespace() + + async def __aexit__(self, exc_type, exc, tb): + return None + + async def fake_user_stats(session): + nonlocal dal_calls + dal_calls += 1 + await asyncio.sleep(0.001) + return { + "total_users": users, + "banned_users": 0, + "active_today": 0, + "paid_subscriptions": users, + "trial_users": 0, + "inactive_users": 0, + "referral_users": 0, + } + + async def fake_financial_stats(session): + nonlocal dal_calls + dal_calls += 1 + await asyncio.sleep(0.001) + return { + "today_revenue": 0.0, + "week_revenue": 0.0, + "month_revenue": 0.0, + "all_time_revenue": 0.0, + "today_payments_count": 0, + "daily_series": [], + } + + async def fake_sync_status(session): + nonlocal dal_calls + dal_calls += 1 + await asyncio.sleep(0.001) + return SimpleNamespace( + status="success", + last_sync_time=None, + details=None, + users_processed_from_panel=users, + subscriptions_synced=users, + ) + + async def fake_recent_payments(session, limit=10): + nonlocal dal_calls + dal_calls += 1 + await asyncio.sleep(0.001) + return [] + + started = time.perf_counter() + with ( + patch.object(admin_stats_module.user_dal, "get_enhanced_user_statistics", fake_user_stats), + patch.object( + admin_stats_module.payment_dal, "get_financial_statistics", fake_financial_stats + ), + patch.object(admin_stats_module.panel_sync_dal, "get_panel_sync_status", fake_sync_status), + patch.object( + admin_stats_module.payment_dal, + "get_recent_payment_logs_with_user", + fake_recent_payments, + ), + ): + await asyncio.gather( + *( + admin_stats_module._load_admin_db_stats(settings, FakeSessionFactory()) + for _ in range(users) + ) + ) + elapsed = time.perf_counter() - started + return { + "seconds": elapsed, + "dal_loader_calls": dal_calls, + "optimized_db_round_trips_per_miss_estimate": 6, + "optimized_db_round_trips_with_cache_estimate": 6, + "legacy_db_round_trips_estimate": users * 15, + } + + async def bench_profile_sync_guard(users: int) -> dict: profile_sync_module._LOCAL_PROFILE_SYNC_CHECKS.clear() settings = SimpleNamespace( @@ -427,6 +520,7 @@ async def run_suite(user_sizes: tuple[int, ...]) -> dict: "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), + "admin_db_stats_cache": await bench_admin_db_stats_cache(users), "profile_sync_guard": await bench_profile_sync_guard(users), "crypt4_same_link": await bench_crypt4(users), } @@ -437,9 +531,10 @@ def _print_table(results: dict[str, dict]) -> None: print( "users | bulk_pages_est | premium_usage_s | premium_panel_calls | " "sync_db_reads_est | sync_db_writes | user_cache_calls | " - "all_users_calls | device_cache_calls | admin_panel_calls | crypt4_panel_calls" + "all_users_calls | device_cache_calls | admin_panel_calls | admin_db_reads_est | " + "crypt4_panel_calls" ) - print("-" * 154) + print("-" * 177) for users, data in results.items(): sync_optimized_reads = ( data["panel_sync_startup"]["optimized_user_lookup_queries_estimate"] @@ -456,6 +551,7 @@ def _print_table(results: dict[str, dict]) -> None: 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['admin_db_stats_cache']['optimized_db_round_trips_with_cache_estimate']:>18} | " 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 index 328a46e..e37abb7 100644 --- a/tests/test_admin_panel_stats_cache.py +++ b/tests/test_admin_panel_stats_cache.py @@ -9,9 +9,11 @@ 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() + stats_module._ADMIN_DB_STATS_CACHES.clear() async def asyncTearDown(self): stats_module._ADMIN_PANEL_STATS_CACHES.clear() + stats_module._ADMIN_DB_STATS_CACHES.clear() def _settings(self): return SimpleNamespace( @@ -55,5 +57,113 @@ class AdminPanelStatsCacheTests(unittest.IsolatedAsyncioTestCase): panel_service.get_nodes_online_lookups.assert_awaited_once() +class AdminDbStatsCacheTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + stats_module._ADMIN_DB_STATS_CACHES.clear() + + async def asyncTearDown(self): + stats_module._ADMIN_DB_STATS_CACHES.clear() + + def _settings(self): + return SimpleNamespace( + ADMIN_DB_STATS_CACHE_TTL_SECONDS=15, + REDIS_URL="redis://redis:6379/0", + REDIS_KEY_PREFIX="shop", + ) + + class _SessionFactory: + def __call__(self): + return self + + async def __aenter__(self): + return SimpleNamespace() + + async def __aexit__(self, exc_type, exc, tb): + return None + + async def test_admin_db_stats_are_cached_between_requests(self): + settings = self._settings() + 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 + + user_stats = AsyncMock( + return_value={ + "total_users": 10, + "banned_users": 1, + "active_today": 2, + "paid_subscriptions": 7, + "trial_users": 1, + "inactive_users": 1, + "referral_users": 3, + } + ) + financial_stats = AsyncMock( + return_value={ + "today_revenue": 1.0, + "week_revenue": 2.0, + "month_revenue": 3.0, + "all_time_revenue": 4.0, + "today_payments_count": 1, + "daily_series": [], + } + ) + sync_status = AsyncMock( + return_value=SimpleNamespace( + status="success", + last_sync_time=None, + details=None, + users_processed_from_panel=10, + subscriptions_synced=7, + ) + ) + recent_payments = AsyncMock( + return_value=[ + SimpleNamespace( + payment_id=1, + user_id=10, + provider="test", + provider_payment_id="provider-1", + amount=123.45, + currency="RUB", + status="succeeded", + description="Test payment", + subscription_duration_months=1, + sale_mode=None, + tariff_key=None, + purchased_gb=None, + purchased_hwid_devices=None, + created_at=None, + ) + ] + ) + + with ( + patch("bot.infra.redis.cache_get_json", fake_get), + patch("bot.infra.redis.cache_set_json", fake_set), + patch.object(stats_module.user_dal, "get_enhanced_user_statistics", user_stats), + patch.object(stats_module.payment_dal, "get_financial_statistics", financial_stats), + patch.object(stats_module.panel_sync_dal, "get_panel_sync_status", sync_status), + patch.object( + stats_module.payment_dal, + "get_recent_payment_logs_with_user", + recent_payments, + ), + ): + first = await stats_module._load_admin_db_stats(settings, self._SessionFactory()) + second = await stats_module._load_admin_db_stats(settings, self._SessionFactory()) + + self.assertEqual(first, second) + self.assertEqual(first["recent_payments"][0]["payment_id"], 1) + user_stats.assert_awaited_once() + financial_stats.assert_awaited_once() + sync_status.assert_awaited_once() + recent_payments.assert_awaited_once() + + if __name__ == "__main__": unittest.main()