Redis/ Query optimization/ multithreading/ and more

Dockerfile

@@ -20,7 +20,9 @@ RUN rm -rf /app/venv \
     && /app/venv/bin/pip install --upgrade pip \
     && /app/venv/bin/pip install -r requirements.txt
 
-RUN adduser --disabled-password --gecos "" appuser && chown -R appuser:appuser /app
+RUN adduser --disabled-password --gecos "" appuser \
+    && mkdir -p /app/backups \
+    && chown -R appuser:appuser /app
 USER appuser
 
 CMD ["/app/venv/bin/python", "main.py"]

api/depends.py

@@ -12,7 +12,12 @@ from database.models import Admin
 
 async def get_session() -> AsyncGenerator[AsyncSession, None]:
     async with async_session_maker() as session:
-        yield session
+        try:
+            yield session
+            await session.commit()
+        except Exception:
+            await session.rollback()
+            raise
 
 
 def hash_token(token: str) -> str:
@@ -65,6 +70,7 @@ async def verify_identity_admin_short(
     """Проверка админа с короткой сессией (для broadcast и др.), чтобы не держать соединение с БД."""
     async with async_session_maker() as session:
         identity = await idb.verify_identity_token(session, x_identity_id, token)
+        await session.commit()
     if not identity:
         raise HTTPException(status_code=401, detail="Unauthorized")
     if not identity.is_admin:
@@ -81,6 +87,7 @@ async def verify_admin_token_short(
     async with async_session_maker() as session:
         result = await session.execute(select(Admin).where(Admin.tg_id == admin_id, Admin.token == hashed))
         admin = result.scalar_one_or_none()
+        await session.commit()
     if not admin:
         raise HTTPException(status_code=401, detail="Unauthorized")
     return admin

api/v1/routes/management.py

@@ -18,6 +18,7 @@ from api.depends import get_session, verify_admin_token, verify_admin_token_shor
 from database import async_session_maker
 from config import API_TOKEN, BOT_SERVICE
 from core.bootstrap import MANAGEMENT_CONFIG
+from core.executor import get_thread_pool
 from core.settings.management_config import update_management_config
 from database.models import Key, User
 from database.models import Server
@@ -64,9 +65,13 @@ async def _restart_bot() -> None:
         is_systemd = parent and "systemd" in parent.name().lower()
 
         if is_systemd:
-            subprocess.run(
-                ["sudo", "systemctl", "restart", BOT_SERVICE],
-                check=True,
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(
+                get_thread_pool(),
+                lambda: subprocess.run(
+                    ["sudo", "systemctl", "restart", BOT_SERVICE],
+                    check=True,
+                ),
             )
         else:
             python_exe = sys.executable
@@ -195,6 +200,7 @@ async def launch_broadcast(
 
     async with async_session_maker() as session:
         tg_ids, total_users = await get_recipients(session, payload.send_to, (payload.cluster_name or None))
+        await session.commit()
     if not tg_ids:
         return {"success": False, "message": "No recipients found", "stats": {"total_messages": 0}}
 

api/v1/routes/settings.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.depends import get_session, verify_admin_token
 from api.v1.schemas.settings import SettingResponse, SettingUpsert
-from database import settings_cache
+from database.settings_cache import settings_cache
 from core.settings.buttons_config import BUTTONS_CONFIG, update_buttons_config
 from core.settings.modes_config import MODES_CONFIG, update_modes_config
 from core.settings.money_config import MONEY_CONFIG, update_money_config

api/v1/routes/users.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from api.depends import get_session, verify_admin_token
 from api.v1.routes.base_crud import generate_crud_router
 from api.v1.schemas.users import UserBase, UserResponse, UserUpdate
-from database import delete_user_data, get_servers
+from database import async_session_maker, delete_user_data, get_servers
 from database.models import Key, User
 from handlers.keys.operations import delete_key_from_cluster
 from logger import logger
@@ -33,18 +33,24 @@ async def delete_user(
         result = await session.execute(select(Key.email, Key.client_id).where(Key.tg_id == tg_id))
         key_records = result.all()
 
-        async def delete_keys_from_servers():
-            try:
-                servers = await get_servers(session=session)
-                tasks = []
-                for email, client_id in key_records:
-                    for cluster_id in servers:
-                        tasks.append(delete_key_from_cluster(cluster_id, email, client_id, session))
-                await asyncio.gather(*tasks, return_exceptions=True)
-            except Exception as e:
-                logger.error(f"[DELETE] Ошибка при удалении ключей с серверов для пользователя {tg_id}: {e}")
+        async with async_session_maker() as s:
+            servers = await get_servers(session=s)
+        cluster_ids = list(servers.keys())
+
+        async def _delete_one(cluster_id: str, email: str, client_id: str):
+            async with async_session_maker() as s:
+                await delete_key_from_cluster(cluster_id, email, client_id, s)
+
+        try:
+            tasks = [
+                _delete_one(cluster_id, email, client_id)
+                for email, client_id in key_records
+                for cluster_id in cluster_ids
+            ]
+            await asyncio.gather(*tasks, return_exceptions=True)
+        except Exception as e:
+            logger.error(f"[DELETE] Ошибка при удалении ключей с серверов для пользователя {tg_id}: {e}")
 
-        await delete_keys_from_servers()
         await delete_user_data(session, tg_id)
 
         return {"detail": f"Пользователь {tg_id} и его ключи успешно удалены."}

api/v2/routes/management.py

@@ -18,6 +18,7 @@ from api.depends import get_session, verify_identity_admin, verify_identity_admi
 from database import async_session_maker
 from config import API_TOKEN, BOT_SERVICE
 from core.bootstrap import MANAGEMENT_CONFIG
+from core.executor import get_thread_pool
 from core.settings.management_config import update_management_config
 from database.models import Key, Server, User
 from handlers.admin.sender.sender_service import BroadcastService
@@ -63,7 +64,14 @@ async def _restart_bot() -> None:
         parent = psutil.Process(os.getpid()).parent()
         is_systemd = parent and "systemd" in parent.name().lower()
         if is_systemd:
-            subprocess.run(["sudo", "systemctl", "restart", BOT_SERVICE], check=True)
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(
+                get_thread_pool(),
+                lambda: subprocess.run(
+                    ["sudo", "systemctl", "restart", BOT_SERVICE],
+                    check=True,
+                ),
+            )
         else:
             python_exe = sys.executable
             script_path = os.path.abspath(sys.argv[0])
@@ -191,6 +199,7 @@ async def launch_broadcast(
         raise HTTPException(status_code=400, detail=f"Message too long. Max {max_len} symbols")
     async with async_session_maker() as session:
         tg_ids, total_users = await get_recipients(session, payload.send_to, (payload.cluster_name or None))
+        await session.commit()
     if not tg_ids:
         return {"success": False, "message": "No recipients found", "stats": {"total_messages": 0}}
     bot = _get_broadcast_bot()

api/v2/routes/settings.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.depends import get_session, verify_identity_admin
 from api.v2.schemas import SettingResponse, SettingUpsert
-from database import settings_cache
+from database.settings_cache import settings_cache
 from core.settings.buttons_config import BUTTONS_CONFIG, update_buttons_config
 from core.settings.modes_config import MODES_CONFIG, update_modes_config
 from core.settings.money_config import MONEY_CONFIG, update_money_config

api/v2/routes/users.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from api.depends import get_session, verify_identity_admin
 from api.v2.schemas import UserBase, UserResponse, UserUpdate
 from api.v2.base_crud import generate_crud_router
-from database import delete_user_data, get_servers
+from database import async_session_maker, delete_user_data, get_servers
 from database.models import Key, User
 from handlers.keys.operations import delete_key_from_cluster
 from logger import logger
@@ -33,18 +33,24 @@ async def delete_user(
         result = await session.execute(select(Key.email, Key.client_id).where(Key.tg_id == tg_id))
         key_records = result.all()
 
-        async def delete_keys_from_servers():
-            try:
-                servers = await get_servers(session=session)
-                tasks = []
-                for email, client_id in key_records:
-                    for cluster_id in servers:
-                        tasks.append(delete_key_from_cluster(cluster_id, email, client_id, session))
-                await asyncio.gather(*tasks, return_exceptions=True)
-            except Exception as e:
-                logger.error(f"[DELETE] Ошибка при удалении ключей с серверов для пользователя {tg_id}: {e}")
+        async with async_session_maker() as s:
+            servers = await get_servers(session=s)
+        cluster_ids = list(servers.keys())
+
+        async def _delete_one(cluster_id: str, email: str, client_id: str):
+            async with async_session_maker() as s:
+                await delete_key_from_cluster(cluster_id, email, client_id, s)
+
+        try:
+            tasks = [
+                _delete_one(cluster_id, email, client_id)
+                for email, client_id in key_records
+                for cluster_id in cluster_ids
+            ]
+            await asyncio.gather(*tasks, return_exceptions=True)
+        except Exception as e:
+            logger.error(f"[DELETE] Ошибка при удалении ключей с серверов для пользователя {tg_id}: {e}")
 
-        await delete_keys_from_servers()
         await delete_user_data(session, tg_id)
         return {"detail": f"Пользователь {tg_id} и его ключи успешно удалены."}
     except Exception as e:

bot.py

@@ -1,3 +1,6 @@
+import os
+from importlib import import_module
+
 from aiogram import Bot, Dispatcher
 from aiogram.client.default import DefaultBotProperties
 from aiogram.enums import ParseMode
@@ -14,6 +17,17 @@ apply_button_icons_patch()
 
 bot = Bot(token=API_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
 storage = MemoryStorage()
+redis_url = os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0")
+
+try:
+    RedisStorage = import_module("aiogram.fsm.storage.redis").RedisStorage
+    redis_from_url = import_module("redis.asyncio").from_url
+
+    redis = redis_from_url(redis_url, encoding="utf-8", decode_responses=True)
+    storage = RedisStorage(redis=redis)
+except Exception:
+    storage = MemoryStorage()
+
 dp = Dispatcher(bot=bot, storage=storage)
 
 dp.include_router(modules_hub)

core/bootstrap.py

@@ -1,4 +1,5 @@
-from database import async_session_maker, settings_cache
+from database import async_session_maker
+from database.settings_cache import settings_cache
 from database.db import warm_pool
 from database.tariffs import initialize_all_tariff_weights
 
@@ -9,6 +10,7 @@ from .settings.money_config import MONEY_CONFIG, load_money_config, update_money
 from .settings.notifications_config import NOTIFICATIONS_CONFIG, load_notifications_config, update_notifications_config
 from .settings.payments_config import PAYMENTS_CONFIG, load_payments_config, update_payments_config
 from .settings.providers_order_config import PROVIDERS_ORDER, load_providers_order, update_providers_order
+from .settings.runtime_sync import publish_runtime_snapshot
 from .settings.tariffs_config import TARIFFS_CONFIG, load_tariffs_config, update_tariffs_config
 
 
@@ -26,3 +28,4 @@ async def bootstrap() -> None:
         await load_tariffs_config(session)
         await session.commit()
         await settings_cache.load(session)
+        await publish_runtime_snapshot()

core/cache_config.py

@@ -0,0 +1,49 @@
+UPDATE_STALE_AGE_SEC = 60
+
+SUBSCRIPTION_CACHE_SUBSCRIBED_MAXSIZE = 200_000
+SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC = 300
+SUBSCRIPTION_CACHE_UNSUBSCRIBED_MAXSIZE = 100_000
+SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC = 60
+
+CONCURRENCY_REJECT_NOTICE_CACHE_MAXSIZE = 100_000
+CONCURRENCY_REJECT_NOTICE_TTL_SEC = 5
+
+THROTTLE_CACHE_MAXSIZE = 50_000
+THROTTLE_CACHE_TTL_SEC = 1.0
+THROTTLE_NOTICE_CACHE_MAXSIZE = 50_000
+THROTTLE_NOTICE_TTL_SEC = 1.0
+
+START_UTM_EXISTS_CACHE_MAXSIZE = 20_000
+START_UTM_EXISTS_TTL_SEC = 300
+
+USER_SNAPSHOT_CACHE_MAXSIZE = 150_000
+USER_SNAPSHOT_CACHE_TTL_SEC = 30
+USER_EXISTS_CACHE_MAXSIZE = 150_000
+USER_EXISTS_CACHE_TTL_SEC = 60
+
+BAN_CACHE_MAXSIZE = 50_000
+BAN_CACHE_TTL_SEC = 30
+
+DIRECT_START_USER_EXISTS_CACHE_MAXSIZE = 50_000
+DIRECT_START_USER_EXISTS_CACHE_TTL_SEC = 20
+
+ADMIN_CACHE_MAXSIZE = 10_000
+ADMIN_CACHE_TTL_SEC = 60
+
+REMNAWAVE_SERVER_CACHE_MAXSIZE = 50_000
+REMNAWAVE_SERVER_CACHE_TTL_SEC = 300
+REMNAWAVE_PROFILE_CACHE_MAXSIZE = 200_000
+REMNAWAVE_PROFILE_CACHE_TTL_SEC = 20
+REMNAWAVE_PROFILE_TIMEOUT_SEC = 3.0
+REMNAWAVE_ACTION_TIMEOUT_SEC = 5.0
+REMNAWAVE_MAX_CONCURRENCY = 20
+
+RUNTIME_CONFIG_SYNC_PULL_INTERVAL_SEC = 1.0
+RUNTIME_CONFIG_SYNC_TTL_SEC = 86_400
+
+SUBSCRIPTION_RESPONSE_CACHE_TTL_SEC = 20
+SUBSCRIPTION_HANDLER_CONCURRENCY = 80
+
+SERVERS_CACHE_TTL_SEC = 60
+TARIFF_BY_ID_CACHE_TTL_SEC = 120
+TARIFFS_FOR_CLUSTER_CACHE_TTL_SEC = 120

core/executor.py

@@ -0,0 +1,73 @@
+import atexit
+import signal
+import multiprocessing
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
+
+from logger import logger
+
+_thread_pool: ThreadPoolExecutor | None = None
+_process_pool: ProcessPoolExecutor | None = None
+
+
+def _atexit_shutdown_pools() -> None:
+    """Очистка пулов при выходе из процесса (в т.ч. по atexit), уменьшает предупреждения resource_tracker."""
+    shutdown_process_pool()
+    shutdown_thread_pool()
+
+
+class _IgnoreSIGINTProcess(multiprocessing.Process):
+    """Процесс, игнорирующий SIGINT в воркере, чтобы Ctrl+C не обрывал queue.get() с трейсбеком."""
+
+    def run(self) -> None:
+        signal.signal(signal.SIGINT, signal.SIG_IGN)
+        super().run()
+
+
+def get_thread_pool() -> ThreadPoolExecutor:
+    """Возвращает общий пул потоков (создаёт при первом вызове)."""
+    global _thread_pool
+    if _thread_pool is None:
+        from config import EXECUTOR_POOL_SIZE
+        size = max(1, int(EXECUTOR_POOL_SIZE))
+        _thread_pool = ThreadPoolExecutor(max_workers=size, thread_name_prefix="bot-thread")
+        logger.debug("Thread pool started (workers=%s)", size)
+    return _thread_pool
+
+
+def shutdown_thread_pool() -> None:
+    """Останавливает пул потоков (вызывать при shutdown приложения)."""
+    global _thread_pool
+    if _thread_pool is not None:
+        _thread_pool.shutdown(wait=True)
+        _thread_pool = None
+        logger.debug("Thread pool shut down")
+
+
+def get_process_pool() -> ProcessPoolExecutor:
+    """
+    Возвращает пул процессов для тяжёлых задач (бэкап и т.д.).
+    Задачи выполняются в отдельных процессах и могут использовать другие ядра CPU.
+    """
+    global _process_pool
+    if _process_pool is None:
+        from config import PROCESS_POOL_SIZE
+        size = max(1, min(int(PROCESS_POOL_SIZE), multiprocessing.cpu_count() or 4))
+        ctx = multiprocessing.get_context("spawn")
+        ctx.Process = _IgnoreSIGINTProcess
+        _process_pool = ProcessPoolExecutor(max_workers=size, mp_context=ctx)
+        atexit.register(_atexit_shutdown_pools)
+        logger.debug("Process pool started (workers=%s)", size)
+    return _process_pool
+
+
+def shutdown_process_pool() -> None:
+    """Останавливает пул процессов (вызывать при shutdown приложения)."""
+    global _process_pool
+    if _process_pool is not None:
+        try:
+            atexit.unregister(_atexit_shutdown_pools)
+        except Exception:
+            pass
+        _process_pool.shutdown(wait=True)
+        _process_pool = None
+        logger.debug("Process pool shut down")

core/redis_cache.py

@@ -0,0 +1,111 @@
+import json
+import os
+import time
+from importlib import import_module
+from typing import Any
+
+_REDIS_CLIENT = None
+_REDIS_UNAVAILABLE_UNTIL = 0.0
+_REDIS_BACKOFF_SEC = 5.0
+
+
+def _now() -> float:
+    return time.monotonic()
+
+
+async def _get_redis() -> Any | None:
+    global _REDIS_CLIENT, _REDIS_UNAVAILABLE_UNTIL
+
+    if _REDIS_CLIENT is not None:
+        return _REDIS_CLIENT
+    if _REDIS_UNAVAILABLE_UNTIL > _now():
+        return None
+
+    try:
+        redis_url = os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0")
+        redis_from_url = import_module("redis.asyncio").from_url
+        client = redis_from_url(redis_url, encoding="utf-8", decode_responses=True)
+        await client.ping()
+        _REDIS_CLIENT = client
+        return _REDIS_CLIENT
+    except Exception:
+        _REDIS_UNAVAILABLE_UNTIL = _now() + _REDIS_BACKOFF_SEC
+        _REDIS_CLIENT = None
+        return None
+
+
+def cache_key(prefix: str, *parts: Any) -> str:
+    tail = ":".join(str(p) for p in parts)
+    return f"{prefix}:{tail}" if tail else prefix
+
+
+async def cache_get(key: str) -> Any | None:
+    client = await _get_redis()
+    if client is None:
+        return None
+    try:
+        raw = await client.get(key)
+        if raw is None:
+            return None
+        return json.loads(raw)
+    except Exception:
+        return None
+
+
+async def cache_set(key: str, value: Any, ttl_sec: float) -> bool:
+    client = await _get_redis()
+    if client is None:
+        return False
+    try:
+        ttl = max(1, int(ttl_sec))
+        await client.set(key, json.dumps(value, ensure_ascii=False), ex=ttl)
+        return True
+    except Exception:
+        return False
+
+
+async def cache_delete(key: str) -> None:
+    client = await _get_redis()
+    if client is None:
+        return
+    try:
+        await client.delete(key)
+    except Exception:
+        return
+
+
+async def cache_setnx(key: str, value: Any, ttl_sec: float) -> bool:
+    client = await _get_redis()
+    if client is None:
+        return False
+    try:
+        ttl = max(1, int(ttl_sec))
+        return bool(await client.set(key, json.dumps(value, ensure_ascii=False), ex=ttl, nx=True))
+    except Exception:
+        return False
+
+
+async def cache_incr(key: str, ttl_sec: float) -> int:
+    client = await _get_redis()
+    if client is None:
+        return 1
+    try:
+        value = await client.incr(key)
+        if value == 1:
+            await client.expire(key, max(1, int(ttl_sec)))
+        return int(value)
+    except Exception:
+        return 1
+
+
+async def cache_delete_pattern(pattern: str) -> int:
+    client = await _get_redis()
+    if client is None:
+        return 0
+    deleted = 0
+    try:
+        async for key in client.scan_iter(match=pattern, count=200):
+            deleted += int(await client.delete(key))
+    except Exception:
+        return deleted
+    return deleted

core/settings/buttons_config.py

@@ -3,13 +3,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_BUTTONS_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 BUTTONS_CONFIG: dict[str, bool] = DEFAULT_BUTTONS_CONFIG.copy()
 BUTTONS_CONFIG.setdefault("ANDROID_TV_BUTTON_ENABLE", False)
 BUTTONS_CONFIG.setdefault("COUPON_BUTTON_ENABLE", True)
+register_runtime_config("BUTTONS_CONFIG", BUTTONS_CONFIG)
 
 
 async def load_buttons_config(session: AsyncSession) -> None:
@@ -65,3 +67,4 @@ async def update_buttons_config(session: AsyncSession, new_values: dict[str, boo
     BUTTONS_CONFIG.clear()
     BUTTONS_CONFIG.update(buttons_config)
     settings_cache.update("BUTTONS_CONFIG", buttons_config)
+    await publish_runtime_config("BUTTONS_CONFIG", buttons_config)

core/settings/management_config.py

@@ -5,12 +5,14 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_MANAGEMENT_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 MANAGEMENT_CONFIG: dict[str, Any] = DEFAULT_MANAGEMENT_CONFIG.copy()
 MANAGEMENT_SETTING_KEY = "MANAGEGENT_CONFIG"
+register_runtime_config(MANAGEMENT_SETTING_KEY, MANAGEMENT_CONFIG)
 
 
 async def load_management_config(session: AsyncSession) -> None:
@@ -60,3 +62,4 @@ async def update_management_config(session: AsyncSession, new_values: dict[str,
     MANAGEMENT_CONFIG.clear()
     MANAGEMENT_CONFIG.update(management_config)
     settings_cache.update(MANAGEMENT_SETTING_KEY, management_config)
+    await publish_runtime_config(MANAGEMENT_SETTING_KEY, management_config)

core/settings/modes_config.py

@@ -3,11 +3,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_MODES_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 MODES_CONFIG: dict[str, bool] = DEFAULT_MODES_CONFIG.copy()
+register_runtime_config("MODES_CONFIG", MODES_CONFIG)
 
 
 async def load_modes_config(session: AsyncSession) -> None:
@@ -57,3 +59,4 @@ async def update_modes_config(session: AsyncSession, new_values: dict[str, bool]
     MODES_CONFIG.clear()
     MODES_CONFIG.update(modes_config)
     settings_cache.update("MODES_CONFIG", modes_config)
+    await publish_runtime_config("MODES_CONFIG", modes_config)

core/settings/money_config.py

@@ -5,11 +5,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_MONEY_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 MONEY_CONFIG: dict[str, Any] = DEFAULT_MONEY_CONFIG.copy()
+register_runtime_config("MONEY_CONFIG", MONEY_CONFIG)
 
 
 def get_currency_mode() -> tuple[str, bool]:
@@ -75,3 +77,4 @@ async def update_money_config(session: AsyncSession, new_values: dict[str, Any])
     MONEY_CONFIG.clear()
     MONEY_CONFIG.update(money_config)
     settings_cache.update("MONEY_CONFIG", money_config)
+    await publish_runtime_config("MONEY_CONFIG", money_config)

core/settings/notifications_config.py

@@ -5,11 +5,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_NOTIFICATIONS_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 NOTIFICATIONS_CONFIG: dict[str, Any] = DEFAULT_NOTIFICATIONS_CONFIG.copy()
+register_runtime_config("NOTIFICATIONS_CONFIG", NOTIFICATIONS_CONFIG)
 
 
 async def load_notifications_config(session: AsyncSession) -> None:
@@ -59,3 +61,4 @@ async def update_notifications_config(session: AsyncSession, new_values: dict[st
     NOTIFICATIONS_CONFIG.clear()
     NOTIFICATIONS_CONFIG.update(notifications_config)
     settings_cache.update("NOTIFICATIONS_CONFIG", notifications_config)
+    await publish_runtime_config("NOTIFICATIONS_CONFIG", notifications_config)

core/settings/payments_config.py

@@ -3,11 +3,13 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
 from ..defaults import DEFAULT_PAYMENTS_CONFIG
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 PAYMENTS_CONFIG: dict[str, bool] = DEFAULT_PAYMENTS_CONFIG.copy()
+register_runtime_config("PAYMENTS_CONFIG", PAYMENTS_CONFIG)
 
 
 async def load_payments_config(session: AsyncSession) -> None:
@@ -57,3 +59,4 @@ async def update_payments_config(session: AsyncSession, new_values: dict[str, bo
     PAYMENTS_CONFIG.clear()
     PAYMENTS_CONFIG.update(payments_config)
     settings_cache.update("PAYMENTS_CONFIG", payments_config)
+    await publish_runtime_config("PAYMENTS_CONFIG", payments_config)

core/settings/providers_order_config.py

@@ -3,9 +3,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 PROVIDERS_ORDER: dict[str, int] = {}
+register_runtime_config("PROVIDERS_ORDER", PROVIDERS_ORDER)
 
 
 async def load_providers_order(session: AsyncSession) -> None:
@@ -39,3 +41,4 @@ async def update_providers_order(session: AsyncSession, new_order: dict[str, int
     PROVIDERS_ORDER.clear()
     PROVIDERS_ORDER.update(new_order)
     settings_cache.update("PROVIDERS_ORDER", new_order)
+    await publish_runtime_config("PROVIDERS_ORDER", new_order)

core/settings/runtime_sync.py

@@ -0,0 +1,98 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Any
+
+from core.cache_config import (
+    RUNTIME_CONFIG_SYNC_PULL_INTERVAL_SEC,
+    RUNTIME_CONFIG_SYNC_TTL_SEC,
+)
+from core.redis_cache import cache_get, cache_key, cache_set
+
+_RUNTIME_CONFIGS_KEY = cache_key("runtime_configs")
+_REGISTRY: dict[str, dict[str, Any]] = {}
+_LOCAL_VERSION = 0.0
+_LAST_PULL_MONOTONIC = 0.0
+_PULL_LOCK = asyncio.Lock()
+
+
+def register_runtime_config(name: str, config_ref: dict[str, Any]) -> None:
+    _REGISTRY[name] = config_ref
+
+
+def _snapshot_from_registry() -> dict[str, dict[str, Any]]:
+    return {name: dict(config_ref) for name, config_ref in _REGISTRY.items()}
+
+
+def _apply_runtime_config(name: str, raw_value: Any) -> None:
+    target = _REGISTRY.get(name)
+    if target is None or not isinstance(raw_value, dict):
+        return
+    target.clear()
+    target.update(raw_value)
+
+
+async def publish_runtime_snapshot() -> None:
+    global _LOCAL_VERSION
+    version = time.time()
+    payload = {
+        "version": version,
+        "configs": _snapshot_from_registry(),
+    }
+    await cache_set(_RUNTIME_CONFIGS_KEY, payload, RUNTIME_CONFIG_SYNC_TTL_SEC)
+    _LOCAL_VERSION = max(_LOCAL_VERSION, float(version))
+
+
+async def publish_runtime_config(name: str, config_value: dict[str, Any]) -> None:
+    global _LOCAL_VERSION
+    if not isinstance(config_value, dict):
+        return
+
+    merged_configs: dict[str, dict[str, Any]] = {}
+    cached = await cache_get(_RUNTIME_CONFIGS_KEY)
+    if isinstance(cached, dict):
+        raw_configs = cached.get("configs")
+        if isinstance(raw_configs, dict):
+            for key, value in raw_configs.items():
+                if isinstance(value, dict):
+                    merged_configs[key] = dict(value)
+
+    merged_configs[name] = dict(config_value)
+    version = time.time()
+    payload = {"version": version, "configs": merged_configs}
+    await cache_set(_RUNTIME_CONFIGS_KEY, payload, RUNTIME_CONFIG_SYNC_TTL_SEC)
+    _LOCAL_VERSION = max(_LOCAL_VERSION, float(version))
+
+
+async def maybe_sync_runtime_configs(force: bool = False) -> bool:
+    global _LOCAL_VERSION, _LAST_PULL_MONOTONIC
+    now = time.monotonic()
+    if not force and (now - _LAST_PULL_MONOTONIC) < RUNTIME_CONFIG_SYNC_PULL_INTERVAL_SEC:
+        return False
+
+    async with _PULL_LOCK:
+        now = time.monotonic()
+        if not force and (now - _LAST_PULL_MONOTONIC) < RUNTIME_CONFIG_SYNC_PULL_INTERVAL_SEC:
+            return False
+        _LAST_PULL_MONOTONIC = now
+
+        payload = await cache_get(_RUNTIME_CONFIGS_KEY)
+        if not isinstance(payload, dict):
+            return False
+
+        try:
+            remote_version = float(payload.get("version") or 0.0)
+        except (TypeError, ValueError):
+            remote_version = 0.0
+        if remote_version <= _LOCAL_VERSION:
+            return False
+
+        raw_configs = payload.get("configs")
+        if not isinstance(raw_configs, dict):
+            return False
+
+        for name, raw_value in raw_configs.items():
+            _apply_runtime_config(name, raw_value)
+        _LOCAL_VERSION = remote_version
+        return True

core/settings/tariffs_config.py

@@ -6,7 +6,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database.models import Setting
 
-from database import settings_cache
+from database.settings_cache import settings_cache
+from .runtime_sync import publish_runtime_config, register_runtime_config
 
 
 TARIFFS_CONFIG: dict[str, Any] = {
@@ -14,6 +15,7 @@ TARIFFS_CONFIG: dict[str, Any] = {
     "KEY_ADDONS_PACK_MODE": "all",
     "KEY_ADDONS_PRICE_BASE_MODE": "current",
 }
+register_runtime_config("TARIFFS_CONFIG", TARIFFS_CONFIG)
 
 
 async def load_tariffs_config(session: AsyncSession) -> None:
@@ -65,6 +67,7 @@ async def update_tariffs_config(session: AsyncSession, new_values: dict[str, Any
     TARIFFS_CONFIG.clear()
     TARIFFS_CONFIG.update(tariffs_config)
     settings_cache.update("TARIFFS_CONFIG", tariffs_config)
+    await publish_runtime_config("TARIFFS_CONFIG", tariffs_config)
 
 
 def calc_extra_devices_price(tariff: dict[str, Any], device_limit: int) -> int:

database/db.py

@@ -5,10 +5,11 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from sqlalchemy.orm import declarative_base
 
 from config import DATABASE_URL, DB_MAX_OVERFLOW, DB_POOL_SIZE
+from core.cache_config import UPDATE_STALE_AGE_SEC
 
 
 CONCURRENT_UPDATES_LIMIT = DB_POOL_SIZE + DB_MAX_OVERFLOW
-MAX_UPDATE_AGE_SEC = 15
+MAX_UPDATE_AGE_SEC = UPDATE_STALE_AGE_SEC
 
 engine = create_async_engine(
     DATABASE_URL,

database/models.py

@@ -86,7 +86,7 @@ class User(DictLikeMixin, Base):
 class Key(DictLikeMixin, Base):
     __tablename__ = "keys"
 
-    tg_id = Column(BigInteger, ForeignKey("users.tg_id"), nullable=False)
+    tg_id = Column(BigInteger, ForeignKey("users.tg_id"), nullable=False, index=True)
     client_id = Column(String, primary_key=True)
     email = Column(String, unique=True)
     created_at = Column(BigInteger)

database/notifications.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timedelta
 
-from sqlalchemy import and_, delete, func, select
+from sqlalchemy import and_, delete, func, select, tuple_
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -56,6 +56,39 @@ async def check_notification_time(session: AsyncSession, tg_id: int, notificatio
     return datetime.utcnow() - last_time > timedelta(hours=hours)
 
 
+async def check_notification_time_bulk(
+    session: AsyncSession,
+    items: list[tuple[int, str]],
+    hours: int,
+) -> set[tuple[int, str]]:
+    """
+    За один запрос определяет, кому из (tg_id, notification_type) можно слать уведомление
+    (прошло больше hours с последней отправки или не слали никогда).
+    Возвращает множество пар (tg_id, notification_type), которым можно слать.
+    """
+    if not items:
+        return set()
+    now = datetime.utcnow()
+    threshold = now - timedelta(hours=hours)
+    stmt = select(
+        Notification.tg_id,
+        Notification.notification_type,
+        Notification.last_notification_time,
+    ).where(tuple_(Notification.tg_id, Notification.notification_type).in_(items))
+    result = await session.execute(stmt)
+    rows = result.all()
+    can_notify = set()
+    found = set()
+    for row in rows:
+        found.add((row.tg_id, row.notification_type))
+        if row.last_notification_time is None or row.last_notification_time < threshold:
+            can_notify.add((row.tg_id, row.notification_type))
+    for pair in items:
+        if pair not in found:
+            can_notify.add(pair)
+    return can_notify
+
+
 async def get_last_notification_time(session: AsyncSession, tg_id: int, notification_type: str) -> int | None:
     stmt = select(Notification.last_notification_time).where(
         Notification.tg_id == tg_id, Notification.notification_type == notification_type

database/servers.py

@@ -2,10 +2,16 @@ from sqlalchemy import delete, func, insert, select, update
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.cache_config import SERVERS_CACHE_TTL_SEC
+from core.redis_cache import cache_delete_pattern, cache_get, cache_key, cache_set
 from database.models import Key, Server, ServerSpecialgroup, ServerSubgroup, Tariff
 from logger import logger
 
 
+async def _invalidate_servers_cache() -> None:
+    await cache_delete_pattern("servers:*")
+
+
 async def create_server(
     session: AsyncSession,
     cluster_name: str,
@@ -24,6 +30,7 @@ async def create_server(
         )
         await session.execute(stmt)
         await session.commit()
+        await _invalidate_servers_cache()
         logger.info(f"✅ Сервер {server_name} добавлен в кластер {cluster_name}")
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при добавлении сервера {server_name}: {e}")
@@ -36,6 +43,7 @@ async def delete_server(session: AsyncSession, server_name: str):
         stmt = delete(Server).where(Server.server_name == server_name)
         await session.execute(stmt)
         await session.commit()
+        await _invalidate_servers_cache()
         logger.info(f"🗑 Сервер {server_name} удалён")
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при удалении сервера {server_name}: {e}")
@@ -46,6 +54,11 @@ async def delete_server(session: AsyncSession, server_name: str):
 async def get_servers(session: AsyncSession, include_enabled: bool = False) -> dict:
     from handlers.utils import ALLOWED_GROUP_CODES
 
+    cache_key_servers = cache_key("servers", int(include_enabled))
+    cached = await cache_get(cache_key_servers)
+    if isinstance(cached, dict):
+        return cached
+
     try:
         stmt = select(Server)
         result = await session.execute(stmt)
@@ -97,6 +110,7 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
                 "cluster_name": cluster,
                 "server_id": s.id,
             })
+        await cache_set(cache_key_servers, grouped, SERVERS_CACHE_TTL_SEC)
         return grouped
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при получении серверов: {e}")
@@ -172,6 +186,7 @@ async def update_server_field(session: AsyncSession, server_name: str, field: st
         stmt = update(Server).where(Server.server_name == server_name).values(**{field: value})
         await session.execute(stmt)
         await session.commit()
+        await _invalidate_servers_cache()
         logger.info(f"✅ Поле {field} сервера {server_name} обновлено на {value}")
         return True
     except SQLAlchemyError as e:
@@ -197,6 +212,7 @@ async def update_server_name_with_keys(session: AsyncSession, old_name: str, new
         await session.execute(stmt_keys)
 
         await session.commit()
+        await _invalidate_servers_cache()
         logger.info(f"✅ Сервер переименован с {old_name} на {new_name}")
         return True
     except SQLAlchemyError as e:
@@ -256,6 +272,7 @@ async def update_server_cluster(session: AsyncSession, server_name: str, new_clu
             )
 
         await session.commit()
+        await _invalidate_servers_cache()
         logger.info(
             f"✅ Сервер {server_name} перемещен в кластер {new_cluster} с обновлением тарифной группы и привязок подгрупп"
         )

database/tariffs.py

@@ -7,10 +7,32 @@ from sqlalchemy import delete, func, insert, select, update
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.cache_config import TARIFF_BY_ID_CACHE_TTL_SEC, TARIFFS_FOR_CLUSTER_CACHE_TTL_SEC
+from core.redis_cache import cache_delete, cache_delete_pattern, cache_get, cache_key, cache_set
 from database.models import Server, Tariff
 from logger import logger
 
 
+def _row_to_cache_dict(row_dict: dict) -> dict:
+    """Делает dict строки БД пригодным для JSON/Redis (datetime → str, убирает _sa_instance_state)."""
+    out = {}
+    for k, v in row_dict.items():
+        if k.startswith("_"):
+            continue
+        if isinstance(v, datetime):
+            out[k] = v.isoformat()
+        else:
+            out[k] = v
+    return out
+
+
+async def _invalidate_tariff_cache(tariff_id: int | None = None) -> None:
+    """Сброс кэша тарифов при изменении (по id и списков по кластерам)."""
+    if tariff_id is not None:
+        await cache_delete(cache_key("tariff", tariff_id))
+    await cache_delete_pattern("tariffs_cluster:*")
+
+
 def create_subgroup_hash(subgroup_title: str, group_code: str) -> str:
     if not subgroup_title:
         return ""
@@ -78,10 +100,18 @@ async def get_tariffs(
 
 
 async def get_tariff_by_id(session: AsyncSession, tariff_id: int):
+    key = cache_key("tariff", tariff_id)
+    cached = await cache_get(key)
+    if isinstance(cached, dict):
+        return cached
     try:
         result = await session.execute(select(Tariff).where(Tariff.id == tariff_id))
         tariff = result.scalar_one_or_none()
-        return dict(tariff.__dict__) if tariff else None
+        if not tariff:
+            return None
+        row = _row_to_cache_dict(dict(tariff.__dict__))
+        await cache_set(key, row, TARIFF_BY_ID_CACHE_TTL_SEC)
+        return row
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при получении тарифа по ID {tariff_id}: {e}")
         await session.rollback()
@@ -89,6 +119,10 @@ async def get_tariff_by_id(session: AsyncSession, tariff_id: int):
 
 
 async def get_tariffs_for_cluster(session: AsyncSession, cluster_name: str):
+    key = cache_key("tariffs_cluster", cluster_name)
+    cached = await cache_get(key)
+    if isinstance(cached, list):
+        return cached
     try:
         server_row = await session.execute(
             select(Server.tariff_group).where(Server.cluster_name == cluster_name).limit(1)
@@ -110,7 +144,9 @@ async def get_tariffs_for_cluster(session: AsyncSession, cluster_name: str):
             .where(Tariff.group_code == group_code, Tariff.is_active.is_(True))
             .order_by(Tariff.sort_order, Tariff.id)
         )
-        return [dict(r.__dict__) for r in result.scalars().all()]
+        rows = [_row_to_cache_dict(dict(r.__dict__)) for r in result.scalars().all()]
+        await cache_set(key, rows, TARIFFS_FOR_CLUSTER_CACHE_TTL_SEC)
+        return rows
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при получении тарифов для кластера {cluster_name}: {e}")
         return []
@@ -139,6 +175,7 @@ async def create_tariff(session: AsyncSession, data: dict):
         stmt = insert(Tariff).values(**data).returning(Tariff)
         result = await session.execute(stmt)
         await session.commit()
+        await _invalidate_tariff_cache()
         return result.scalar_one()
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при создании тарифа: {e}")
@@ -153,6 +190,7 @@ async def update_tariff(session: AsyncSession, tariff_id: int, updates: dict):
         updates["updated_at"] = datetime.utcnow()
         await session.execute(update(Tariff).where(Tariff.id == tariff_id).values(**updates))
         await session.commit()
+        await _invalidate_tariff_cache(tariff_id)
         return True
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при обновлении тарифа ID={tariff_id}: {e}")
@@ -164,6 +202,7 @@ async def delete_tariff(session: AsyncSession, tariff_id: int):
     try:
         await session.execute(delete(Tariff).where(Tariff.id == tariff_id))
         await session.commit()
+        await _invalidate_tariff_cache(tariff_id)
         return True
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при удалении тарифа ID={tariff_id}: {e}")
@@ -193,6 +232,7 @@ async def get_tariff_sort_order(session: AsyncSession, tariff_id: int) -> int:
         if sort_order is None:
             await session.execute(update(Tariff).where(Tariff.id == tariff_id).values(sort_order=1))
             await session.commit()
+            await _invalidate_tariff_cache(tariff_id)
             return 1
 
         return sort_order
@@ -209,6 +249,7 @@ async def move_tariff_up(session: AsyncSession, tariff_id: int) -> bool:
 
         await session.execute(update(Tariff).where(Tariff.id == tariff_id).values(sort_order=new_order))
         await session.commit()
+        await _invalidate_tariff_cache(tariff_id)
         return True
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при перемещении тарифа {tariff_id} вверх: {e}")
@@ -223,6 +264,7 @@ async def move_tariff_down(session: AsyncSession, tariff_id: int) -> bool:
 
         await session.execute(update(Tariff).where(Tariff.id == tariff_id).values(sort_order=new_order))
         await session.commit()
+        await _invalidate_tariff_cache(tariff_id)
         return True
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при перемещении тарифа {tariff_id} вниз: {e}")
@@ -243,6 +285,7 @@ async def initialize_tariff_sort_orders(session: AsyncSession, group_code: str)
             await session.execute(update(Tariff).where(Tariff.id == tariff.id).values(sort_order=new_sort_order))
 
         await session.commit()
+        await _invalidate_tariff_cache()
         return True
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при инициализации sort_order для группы {group_code}: {e}")
@@ -262,6 +305,7 @@ async def initialize_all_tariff_weights(session: AsyncSession) -> bool:
             await session.execute(update(Tariff).where(Tariff.id == tariff.id).values(sort_order=1))
 
         await session.commit()
+        await _invalidate_tariff_cache()
         return True
 
     except SQLAlchemyError as e:

database/users.py

@@ -1,11 +1,15 @@
 from datetime import datetime
 
-from cachetools import TTLCache
 from sqlalchemy import delete, exists, func, or_, select, update
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.cache_config import (
+    USER_EXISTS_CACHE_TTL_SEC,
+    USER_SNAPSHOT_CACHE_TTL_SEC,
+)
+from core.redis_cache import cache_delete, cache_get, cache_key, cache_set
 from database.models import (
     BlockedUser,
     CouponUsage,
@@ -20,12 +24,15 @@ from database.models import (
 )
 from logger import logger
 
-_SNAPSHOT_CACHE: TTLCache[int, tuple[int, int]] = TTLCache(maxsize=150_000, ttl=30)
-_EXISTS_CACHE: TTLCache[int, bool] = TTLCache(maxsize=150_000, ttl=60)
-
 
 def invalidate_user_snapshot(tg_id: int) -> None:
-    _SNAPSHOT_CACHE.pop(tg_id, None)
+    import asyncio
+
+    try:
+        loop = asyncio.get_running_loop()
+        loop.create_task(cache_delete(cache_key("user_snapshot", tg_id)))
+    except RuntimeError:
+        return
 
 
 async def add_user(
@@ -60,7 +67,7 @@ async def add_user(
             return False
         if commit:
             await session.commit()
-        _EXISTS_CACHE[tg_id] = True
+        await cache_set(cache_key("user_exists", tg_id), True, USER_EXISTS_CACHE_TTL_SEC)
         logger.info(f"[DB] Новый пользователь добавлен: {tg_id} (source: {source_code})")
         return True
     except SQLAlchemyError as e:
@@ -90,14 +97,13 @@ async def update_balance(session: AsyncSession, tg_id: int, amount: float) -> No
 
 
 async def check_user_exists(session: AsyncSession, tg_id: int) -> bool:
-    try:
-        return _EXISTS_CACHE[tg_id]
-    except KeyError:
-        pass
+    cached = await cache_get(cache_key("user_exists", tg_id))
+    if isinstance(cached, bool):
+        return cached
     stmt = select(exists().where(User.tg_id == tg_id))
     result = await session.execute(stmt)
     value = result.scalar()
-    _EXISTS_CACHE[tg_id] = value
+    await cache_set(cache_key("user_exists", tg_id), bool(value), USER_EXISTS_CACHE_TTL_SEC)
     return value
 
 
@@ -173,7 +179,7 @@ async def upsert_user(
             if row is None:
                 return None
             await session.commit()
-            _EXISTS_CACHE[tg_id] = True
+            await cache_set(cache_key("user_exists", tg_id), True, USER_EXISTS_CACHE_TTL_SEC)
             return dict(row)
 
         res = await session.execute(
@@ -203,7 +209,7 @@ async def upsert_user(
         )
         row = res.mappings().one()
         await session.commit()
-        _EXISTS_CACHE[tg_id] = True
+        await cache_set(cache_key("user_exists", tg_id), True, USER_EXISTS_CACHE_TTL_SEC)
         return dict(row)
     except SQLAlchemyError as e:
         logger.error(f"[DB] Ошибка при UPSERT пользователя {tg_id}: {e}")
@@ -213,7 +219,6 @@ async def upsert_user(
 
 async def delete_user_data(session: AsyncSession, tg_id: int):
     try:
-        # Local import breaks circular dependency with database.keys <-> database.users
         from database.keys import delete_key
 
         await session.execute(delete(Notification).where(Notification.tg_id == tg_id))
@@ -246,17 +251,18 @@ async def mark_trial_extended(tg_id: int, session: AsyncSession):
 
 
 async def get_user_snapshot(session: AsyncSession, tg_id: int) -> tuple[int, int] | None:
-    try:
-        return _SNAPSHOT_CACHE[tg_id]
-    except KeyError:
-        pass
+    cached = await cache_get(cache_key("user_snapshot", tg_id))
+    if isinstance(cached, list) and len(cached) == 2:
+        return (int(cached[0]), int(cached[1]))
+    if isinstance(cached, tuple) and len(cached) == 2:
+        return (int(cached[0]), int(cached[1]))
     keys_count_sq = select(func.count(Key.client_id)).where(Key.tg_id == tg_id).scalar_subquery()
     res = await session.execute(select(func.coalesce(User.trial, 0), keys_count_sq).where(User.tg_id == tg_id))
     row = res.first()
     if row is None:
         return None
     value = (int(row[0]), int(row[1]))
-    _SNAPSHOT_CACHE[tg_id] = value
+    await cache_set(cache_key("user_snapshot", tg_id), [value[0], value[1]], USER_SNAPSHOT_CACHE_TTL_SEC)
     return value
 
 

docker-compose.yml

@@ -4,6 +4,32 @@ services:
     build: .
     restart: unless-stopped
     network_mode: host
+    depends_on:
+      redis:
+        condition: service_healthy
+    environment:
+      REDIS_URL: redis://127.0.0.1:6379/0
+      BACK_DIR: /app/backups
     volumes:
       - /:/host:ro
+      - backups_data:/app/backups
+
+  redis:
+    image: redis:7-alpine
+    container_name: solobot-redis
+    restart: unless-stopped
+    command: ["redis-server", "--appendonly", "yes", "--maxmemory", "256mb", "--maxmemory-policy", "allkeys-lru"]
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+
+volumes:
+  redis_data:
+  backups_data:
 

filters/admin.py

@@ -42,6 +42,7 @@ class IsAdminFilter(BaseFilter):
                 is_admin = admin is not None or user_id in admin_ids
                 is_super = admin.role != "moderator" if admin else (user_id in admin_ids)
                 _set_cached_admin(user_id, is_admin, is_super)
+                await session.commit()
                 return is_admin
         except (Exception,):
             return False
@@ -62,9 +63,11 @@ class IsSuperAdminFilter(BaseFilter):
                 admin = (await session.execute(select(Admin).where(Admin.tg_id == user_id))).scalar_one_or_none()
                 if not admin:
                     _set_cached_admin(user_id, False, False)
+                    await session.commit()
                     return False
                 is_super = admin.role != "moderator"
                 _set_cached_admin(user_id, True, is_super)
+                await session.commit()
                 return is_super
         except (Exception,):
             return False

handlers/admin/clusters/cluster_manage.py

@@ -198,6 +198,7 @@ async def handle_days_input(message: Message, state: FSMContext, session: AsyncS
                 f"✅ Время подписки продлено на <b>{days} дней</b> для <b>{affected}</b> пользователей в кластере <b>{cluster_name}</b>."
             )
         else:
+            await session.release_early()
             for key in keys:
                 new_expiry = key.expiry_time + add_ms
 

handlers/admin/restart/restart_handler.py

@@ -6,6 +6,7 @@ import sys
 import psutil
 
 from aiogram import F, Router
+from core.executor import get_thread_pool
 from aiogram.types import CallbackQuery
 
 from filters.admin import IsAdminFilter
@@ -32,9 +33,13 @@ async def restart_bot():
         is_systemd = parent and "systemd" in parent.name().lower()
 
         if is_systemd:
-            subprocess.run(
-                ["sudo", "systemctl", "restart", "bot.service"],
-                check=True,
+            loop = asyncio.get_running_loop()
+            await loop.run_in_executor(
+                get_thread_pool(),
+                lambda: subprocess.run(
+                    ["sudo", "systemctl", "restart", "bot.service"],
+                    check=True,
+                ),
             )
         else:
             python_exe = sys.executable

handlers/admin/users/users_hwid.py

@@ -2,10 +2,13 @@ from aiogram import F, Router, types
 from aiogram.types import CallbackQuery
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD
-from database import get_client_id_by_email, get_servers
+from panels.remnawave_runtime import (
+    invalidate_remnawave_profile,
+    resolve_remnawave_api_url,
+    with_remnawave_api,
+)
+from database import get_client_id_by_email
 from filters.admin import IsAdminFilter
-from panels.remnawave import RemnawaveAPI
 
 from .keyboard import AdminUserEditorCallback, build_editor_kb, build_hwid_menu_kb
 
@@ -30,30 +33,25 @@ async def handle_hwid_menu(
         await callback_query.message.edit_text("🚫 Не удалось найти client_id по email.")
         return
 
-    servers = await get_servers(session=session)
-    remna_server = None
-    for cluster_servers in servers.values():
-        for server in cluster_servers:
-            if server.get("panel_type", "") == "remnawave":
-                remna_server = server
-                break
-        if remna_server:
-            break
-
-    if not remna_server:
+    remna_api_url = await resolve_remnawave_api_url(session, "", fallback_any=True)
+    if not remna_api_url:
         await callback_query.message.edit_text(
             "🚫 Нет доступного сервера Remnawave.",
             reply_markup=build_editor_kb(tg_id),
         )
         return
 
-    api = RemnawaveAPI(remna_server["api_url"])
-    if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
+    async def _fetch_info_and_devices(api):
+        user_info = await api.get_user_by_uuid(client_id)
+        devices = await api.get_user_hwid_devices(client_id)
+        return user_info, devices
+
+    result = await with_remnawave_api(session, "", _fetch_info_and_devices, fallback_any=True, timeout_sec=8.0)
+    if result is None:
         await callback_query.message.edit_text("❌ Ошибка авторизации в Remnawave.")
         return
 
-    user_info = await api.get_user_by_uuid(client_id)
-    devices = await api.get_user_hwid_devices(client_id)
+    user_info, devices = result
 
     status_emoji = "🟢"
     status_text = "Онлайн"
@@ -131,42 +129,44 @@ async def handle_hwid_reset(
         await callback_query.message.edit_text("🚫 Не удалось найти client_id по email.")
         return
 
-    servers = await get_servers(session=session)
-    remna_server = None
-    for cluster_servers in servers.values():
-        for server in cluster_servers:
-            if server.get("panel_type", "") == "remnawave":
-                remna_server = server
-                break
-        if remna_server:
-            break
-
-    if not remna_server:
+    remna_api_url = await resolve_remnawave_api_url(session, "", fallback_any=True)
+    if not remna_api_url:
         await callback_query.message.edit_text(
             "🚫 Нет доступного сервера Remnawave.",
             reply_markup=build_editor_kb(tg_id),
         )
         return
 
-    api = RemnawaveAPI(remna_server["api_url"])
-    if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
+    async def _reset_devices(api):
+        devices = await api.get_user_hwid_devices(client_id)
+        if not devices:
+            return 0, 0
+        deleted = 0
+        for device in devices:
+            if await api.delete_user_hwid_device(client_id, device["hwid"]):
+                deleted += 1
+        return len(devices), deleted
+
+    reset_result = await with_remnawave_api(session, "", _reset_devices, fallback_any=True, timeout_sec=12.0)
+    if reset_result is None:
         await callback_query.message.edit_text("❌ Ошибка авторизации в Remnawave.")
         return
 
-    devices = await api.get_user_hwid_devices(client_id)
-    if not devices:
+    total, deleted = reset_result
+    await invalidate_remnawave_profile(
+        session,
+        "",
+        str(client_id),
+        fallback_any=True,
+    )
+    if total == 0:
         await callback_query.message.edit_text(
             "ℹ️ У пользователя нет привязанных устройств.",
             reply_markup=build_editor_kb(tg_id, True),
         )
         return
 
-    deleted = 0
-    for device in devices:
-        if await api.delete_user_hwid_device(client_id, device["hwid"]):
-            deleted += 1
-
     await callback_query.message.edit_text(
-        f"✅ Удалено HWID-устройств: <b>{deleted}</b> из <b>{len(devices)}</b>.",
+        f"✅ Удалено HWID-устройств: <b>{deleted}</b> из <b>{total}</b>.",
         reply_markup=build_editor_kb(tg_id, True),
     )

handlers/admin/users/users_keys.py

@@ -815,6 +815,7 @@ async def handle_delete_key_confirm(
 
     if client_id:
         clusters = await get_servers(session=session)
+        await session.release_early()
 
         async def delete_key_from_servers():
             tasks = []
@@ -862,6 +863,7 @@ async def handle_delete_user_confirm(
 
     result = await session.execute(select(Key.email, Key.client_id).where(Key.tg_id == tg_id))
     key_records = result.all()
+    await session.release_early()
 
     async def delete_keys_from_servers():
         try:
@@ -1283,6 +1285,7 @@ async def handle_admin_unfreeze_subscription(
 
         await mark_key_as_unfrozen(session, record["tg_id"], client_id, new_expiry_time)
         await session.commit()
+        await session.release_early()
 
         await renew_key_in_cluster(
             cluster_id=cluster_id,
@@ -1361,6 +1364,9 @@ async def change_expiry_time(expiry_time: int, email: str, session: AsyncSession
         if not target_cluster:
             return ValueError(f"No suitable cluster found for server {server_id}")
 
+    from middlewares.session import release_session_early
+    await release_session_early(session)
+
     await renew_key_in_cluster(
         cluster_id=target_cluster,
         email=email,
@@ -1747,6 +1753,7 @@ async def handle_cfg_save(callback_query: CallbackQuery, state: FSMContext, sess
         return
 
     try:
+        await session.release_early()
         await renew_key_in_cluster(
             cluster_id=key_obj.server_id,
             email=email,

handlers/admin/users/users_tariffs.py

@@ -333,6 +333,7 @@ async def handle_user_renew_confirm(
         )
     )
     await session.commit()
+    await session.release_early()
 
     try:
         ok = await renew_key_in_cluster(
@@ -664,6 +665,7 @@ async def handle_cfg_renew_apply(callback_query: CallbackQuery, session: AsyncSe
         )
     )
     await session.commit()
+    await session.release_early()
 
     try:
         ok = await renew_key_in_cluster(

handlers/coupons.py

@@ -249,6 +249,7 @@ async def handle_key_extension(
         if tariff:
             key_subgroup = tariff.get("subgroup_title")
 
+        await session.release_early()
         await renew_key_in_cluster(
             cluster_id=key.server_id,
             email=key.email,

handlers/keys/key_freeze.py

@@ -102,6 +102,7 @@ async def process_callback_unfreeze_subscription_confirm(callback_query: Callbac
 
         await mark_key_as_unfrozen(session, record["tg_id"], client_id, new_expiry_time)
         await session.commit()
+        await session.release_early()
 
         max(leftover / (1000 * 86400), 0.01)
         logger.info(

handlers/keys/key_renew.py

@@ -825,6 +825,7 @@ async def complete_key_renewal(
             logger.error(f"[Error] Кластер для {server_or_cluster} не найден.")
             return
 
+        await session.release_early()
         await renew_key_in_cluster(
             cluster_id=cluster_id,
             email=email,

handlers/keys/key_view.py

@@ -20,15 +20,19 @@ from config import (
     HAPP_CRYPTOLINK,
     HWID_RESET_BUTTON,
     QRCODE,
-    REMNAWAVE_LOGIN,
-    REMNAWAVE_PASSWORD,
     REMNAWAVE_WEBAPP,
     REMNAWAVE_WEBAPP_OPEN_IN_BROWSER,
     TOGGLE_CLIENT,
     USE_COUNTRY_SELECTION,
 )
 from core.bootstrap import BUTTONS_CONFIG, MODES_CONFIG
-from database import get_key_details, get_keys, get_servers
+from panels.remnawave_runtime import (
+    get_remnawave_profile,
+    invalidate_remnawave_profile,
+    resolve_remnawave_api_url,
+    with_remnawave_api,
+)
+from database import get_key_details, get_keys
 from database.models import Key
 from handlers.buttons import (
     ADDONS_BUTTON_DEVICES,
@@ -73,9 +77,6 @@ from hooks.processors import (
     process_view_key_menu,
 )
 from logger import logger
-from panels.remnawave import RemnawaveAPI
-
-
 router = Router()
 moscow_tz = pytz.timezone("Europe/Moscow")
 
@@ -360,37 +361,17 @@ async def build_key_view_payload(session: AsyncSession, key_name: str):
     hwid_count = 0
     remna_used_gb = None
     if is_full_remnawave and client_id:
-        try:
-            servers = await get_servers(session)
-            remna_server = None
-            for cluster_name, cluster_servers in servers.items():
-                for srv in cluster_servers:
-                    if (srv.get("server_name") == server_name or cluster_name == server_name) and srv.get(
-                        "panel_type"
-                    ) == "remnawave":
-                        remna_server = srv
-                        break
-                if remna_server:
-                    break
-
-            if remna_server:
-                api = RemnawaveAPI(remna_server["api_url"])
-                if await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-                    devices = await api.get_user_hwid_devices(client_id)
-                    hwid_count = len(devices or [])
-                    user_data = await api.get_user_by_uuid(client_id)
-                    if user_data:
-                        user_traffic = user_data.get("userTraffic", {})
-                        used_bytes = user_traffic.get("usedTrafficBytes", 0)
-                        remna_used_gb = round(used_bytes / GB, 1)
-                        traffic_limit_bytes_actual = user_data.get("trafficLimitBytes")
-                        if traffic_limit_bytes_actual is not None:
-                            if traffic_limit_bytes_actual > 0:
-                                traffic_limit_gb = int(traffic_limit_bytes_actual / GB)
-                            else:
-                                traffic_limit_gb = 0
-        except Exception as error:
-            logger.error(f"Ошибка при получении данных Remnawave для {client_id}: {error}")
+        profile = await get_remnawave_profile(session, str(server_name), client_id)
+        if profile:
+            hwid_count = int(profile.get("hwid_count") or 0)
+            remna_used_gb = profile.get("used_gb")
+            traffic_limit_bytes_actual = profile.get("traffic_limit_bytes")
+            if traffic_limit_bytes_actual is not None:
+                try:
+                    traffic_limit_bytes_actual = int(traffic_limit_bytes_actual)
+                    traffic_limit_gb = int(traffic_limit_bytes_actual / GB) if traffic_limit_bytes_actual > 0 else 0
+                except (TypeError, ValueError):
+                    pass
 
     country_selection_enabled = bool(MODES_CONFIG.get("COUNTRY_SELECTION_ENABLED", USE_COUNTRY_SELECTION))
     remnawave_webapp_enabled = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_ENABLED", REMNAWAVE_WEBAPP))
@@ -507,25 +488,42 @@ async def handle_reset_hwid(callback_query: CallbackQuery, session: AsyncSession
         await callback_query.answer("❌ У ключа отсутствует client_id.", show_alert=True)
         return
 
-    servers = await get_servers(session=session)
-    remna_server = next((srv for cl in servers.values() for srv in cl if srv.get("panel_type") == "remnawave"), None)
-    if not remna_server:
+    remna_api_url = await resolve_remnawave_api_url(session, str(record.get("server_id") or ""), fallback_any=True)
+    if not remna_api_url:
         await callback_query.answer("❌ Remnawave-сервер не найден.", show_alert=True)
         return
 
-    api = RemnawaveAPI(remna_server["api_url"])
-    if not await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
+    async def _reset_devices(api):
+        devices = await api.get_user_hwid_devices(client_id)
+        if not devices:
+            return 0, 0
+        deleted_local = 0
+        for device in devices:
+            if await api.delete_user_hwid_device(client_id, device["hwid"]):
+                deleted_local += 1
+        return len(devices), deleted_local
+
+    reset_result = await with_remnawave_api(
+        session,
+        str(record.get("server_id") or ""),
+        _reset_devices,
+        fallback_any=True,
+        timeout_sec=12.0,
+    )
+    if reset_result is None:
         await callback_query.answer("❌ Авторизация в Remnawave не удалась.", show_alert=True)
         return
 
-    devices = await api.get_user_hwid_devices(client_id)
-    if not devices:
+    total, deleted = reset_result
+    await invalidate_remnawave_profile(
+        session,
+        str(record.get("server_id") or ""),
+        str(client_id),
+        fallback_any=True,
+    )
+    if total == 0:
         await callback_query.answer("✅ Устройства не были привязаны.", show_alert=True)
     else:
-        deleted = 0
-        for device in devices:
-            if await api.delete_user_hwid_device(client_id, device["hwid"]):
-                deleted += 1
         await callback_query.answer(f"✅ Устройства сброшены ({deleted})", show_alert=True)
 
     if await process_after_hwid_reset(

handlers/keys/keys.py

@@ -69,6 +69,7 @@ async def process_callback_confirm_delete(callback_query: CallbackQuery, session
             keyboard = types.InlineKeyboardMarkup(inline_keyboard=[[back_button]])
 
             await delete_key(session, client_id)
+            await session.release_early()
 
             await edit_or_send_message(
                 target_message=callback_query.message,

handlers/keys/operations/aggregated_links.py

@@ -2,15 +2,15 @@ import asyncio
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import HAPP_CRYPTOLINK, LEGACY_LINKS, PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
+from config import HAPP_CRYPTOLINK, LEGACY_LINKS, PUBLIC_LINK, SUPERNODE
 from core.bootstrap import MODES_CONFIG
+from panels.remnawave_runtime import with_remnawave_api
 from database import filter_cluster_by_subgroup, get_key_details, get_tariff_by_id
 from logger import logger
 from panels._3xui import get_vless_link_for_client, get_xui_instance
-from panels.remnawave import RemnawaveAPI
 from servers import extract_host
 
-from .utils import is_plan_vless, score_vless_url, split_by_panel
+from .utils import is_plan_vless, split_by_panel
 
 
 async def _is_vless_tariff(session: AsyncSession, email: str) -> bool:
@@ -23,40 +23,46 @@ async def _is_vless_tariff(session: AsyncSession, email: str) -> bool:
     return is_plan_vless(tariff)
 
 
-async def _try_build_remna_vless(servers: list, email: str) -> tuple[str | None, str | None, str | None]:
+async def _try_build_remna_vless(
+    session: AsyncSession,
+    servers: list,
+    email: str,
+) -> tuple[str | None, str | None, str | None]:
     happ_cryptolink_enabled = bool(MODES_CONFIG.get("HAPP_CRYPTOLINK_ENABLED", HAPP_CRYPTOLINK))
 
     si = servers[0]
-    remna = RemnawaveAPI(si["api_url"])
-    ok = await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD)
-    if not ok:
+    server_ref = si.get("server_name") or si.get("cluster_name") or si.get("api_url") or ""
+
+    async def _build(api):
+        data = await api.get_subscription_by_username(email)
+        if not data:
+            return None
+
+        sub_url = data.get("subscriptionUrl") or None
+        links = data.get("links") or []
+
+        best_vless = None
+        for link in links:
+            if isinstance(link, str) and link.lower().startswith("vless://"):
+                best_vless = link
+                logger.debug(f"[Remnawave] Found VLESS in links: {link[:60]}...")
+                break
+
+        happ_link = None
+        if happ_cryptolink_enabled and sub_url:
+            try:
+                happ_link = await api.encrypt_happ_crypto_link(sub_url)
+            except Exception as e:
+                logger.warning(f"[Remnawave] happ encrypt failed: {e}")
+                happ_link = None
+
+        return best_vless, sub_url, happ_link
+
+    result = await with_remnawave_api(session, str(server_ref), _build, fallback_any=True, timeout_sec=8.0)
+    if result is None:
         logger.warning("[Remnawave] login failed")
         return None, None, None
-
-    data = await remna.get_subscription_by_username(email)
-    if not data:
-        logger.warning("[Remnawave] by-username empty")
-        return None, None, None
-
-    sub_url = data.get("subscriptionUrl") or None
-    links = data.get("links") or []
-
-    best_vless = None
-    for link in links:
-        if isinstance(link, str) and link.lower().startswith("vless://"):
-            best_vless = link
-            logger.debug(f"[Remnawave] Found VLESS in links: {link[:60]}...")
-            break
-
-    happ_link = None
-    if happ_cryptolink_enabled and sub_url:
-        try:
-            happ_link = await remna.encrypt_happ_crypto_link(sub_url)
-        except Exception as e:
-            logger.warning(f"[Remnawave] happ encrypt failed: {e}")
-            happ_link = None
-
-    return best_vless, sub_url, happ_link
+    return result
 
 
 async def _try_build_3xui_vless(servers: list, email: str) -> str | None:
@@ -143,7 +149,7 @@ async def make_aggregated_link(
                 logger.info("[agg_link] choose 3x-ui VLESS")
                 return xui_link
         if remna:
-            best_vless, sub_url, happ_link = await _try_build_remna_vless(remna, email)
+            best_vless, sub_url, happ_link = await _try_build_remna_vless(session, remna, email)
             if best_vless:
                 logger.info("[agg_link] choose Remnawave VLESS")
                 return best_vless
@@ -168,7 +174,7 @@ async def make_aggregated_link(
         if legacy_links_enabled:
             logger.info("[agg_link] LEGACY non-vless -> base link")
             return f"{base}/{email}/{tg_id}"
-        best_vless, sub_url, happ_link = await _try_build_remna_vless(remna, email)
+        best_vless, sub_url, happ_link = await _try_build_remna_vless(session, remna, email)
         if remna_link_override and (
             remna_link_override.lower().startswith("vless://") or remna_link_override.startswith(("http", "happ://"))
         ):

handlers/keys/operations/renewal.py

@@ -4,7 +4,8 @@ from datetime import datetime
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
+from config import SUPERNODE
+from panels.remnawave_runtime import invalidate_remnawave_profile, with_remnawave_api
 from database import (
     delete_notification,
     filter_cluster_by_subgroup,
@@ -24,7 +25,6 @@ from logger import (
     PANEL_XUI,
 )
 from panels._3xui import extend_client_key, get_xui_instance
-from panels.remnawave import RemnawaveAPI
 
 from .aggregated_links import make_aggregated_link
 from ...tariffs.subgroup_migration import migrate_between_subgroups
@@ -70,21 +70,9 @@ async def renew_on_remnawave(
             :1
         ]
 
-    remna = RemnawaveAPI(remnawave_nodes[0]["api_url"])
-    if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-        logger.error(f"{PANEL_REMNA} Не удалось войти в Remnawave API")
-        return False
+    server_ref = remnawave_nodes[0].get("server_name") or remnawave_nodes[0].get("api_url") or ""
     hwid_device_limit = int(hwid_device_limit or 0)
 
-    if old_device_limit is not None and hwid_device_limit < old_device_limit:
-        try:
-            await remna.clear_all_hwid_devices(client_id)
-            logger.info(
-                f"{PANEL_REMNA} HWID устройства сброшены для {client_id} (лимит {old_device_limit} → {hwid_device_limit})"
-            )
-        except Exception as e:
-            logger.warning(f"{PANEL_REMNA} Ошибка сброса HWID: {e}")
-
     expire_iso = datetime.utcfromtimestamp(new_expiry_time // 1000).isoformat() + "Z"
     traffic_limit_bytes = total_gb * 1024 * 1024 * 1024 if total_gb else 0
     active_inbounds = [s["inbound_id"] for s in remnawave_nodes]
@@ -98,13 +86,39 @@ async def renew_on_remnawave(
         "external_squad_uuid": external_squad_uuid,
     }
 
-    updated = await remna.update_user(**update_kwargs)
-    if updated:
-        if reset_traffic:
+    async def _renew(api):
+        if old_device_limit is not None and hwid_device_limit < old_device_limit:
             try:
-                await remna.reset_user_traffic(client_id)
+                await api.clear_all_hwid_devices(client_id)
+                logger.info(
+                    f"{PANEL_REMNA} HWID устройства сброшены для {client_id} (лимит {old_device_limit} → {hwid_device_limit})"
+                )
+            except Exception as e:
+                logger.warning(f"{PANEL_REMNA} Ошибка сброса HWID: {e}")
+
+        updated_local = await api.update_user(**update_kwargs)
+        if updated_local and reset_traffic:
+            try:
+                await api.reset_user_traffic(client_id)
             except Exception as e:
                 logger.warning(f"{PANEL_REMNA} reset_user_traffic: {e}")
+        if updated_local:
+            await invalidate_remnawave_profile(
+                session,
+                str(server_ref),
+                str(client_id),
+                fallback_any=True,
+            )
+        return bool(updated_local)
+
+    updated = await with_remnawave_api(
+        session,
+        str(server_ref),
+        _renew,
+        fallback_any=True,
+        timeout_sec=12.0,
+    )
+    if updated:
         logger.info(f"{PANEL_REMNA} Подписка {client_id} успешно продлена")
         return True
     logger.debug(f"{PANEL_REMNA} Не удалось продлить {client_id}. Автосоздание отключено.")

handlers/keys/operations/traffic.py

@@ -5,12 +5,16 @@ from typing import Any
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
+from config import SUPERNODE
+from panels.remnawave_runtime import (
+    get_remnawave_profile,
+    invalidate_remnawave_profile,
+    with_remnawave_api,
+)
 from database import get_servers
 from database.models import Key, Server
 from logger import logger
 from panels._3xui import get_client_traffic, get_xui_instance
-from panels.remnawave import RemnawaveAPI
 
 
 async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dict[str, Any]:
@@ -54,7 +58,7 @@ async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dic
 
     remnawave_client_id = None
     remnawave_checked = False
-    remnawave_api_url = None
+    remnawave_server_ref = None
 
     async def fetch_traffic(server_info: dict, client_id: str) -> tuple[str, Any]:
         server_name = server_info["server_name"]
@@ -88,7 +92,7 @@ async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dic
 
             if panel_type == "remnawave" and not remnawave_checked:
                 remnawave_client_id = client_id
-                remnawave_api_url = server_info["api_url"]
+                remnawave_server_ref = server_info.get("server_name") or server_info.get("cluster_name")
                 remnawave_checked = True
             elif panel_type == "3x-ui":
                 tasks.append(fetch_traffic(server_info, client_id))
@@ -97,22 +101,13 @@ async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dic
     for server, result in results:
         user_traffic_data[server] = result
 
-    if remnawave_client_id and remnawave_api_url:
-        try:
-            remna = RemnawaveAPI(remnawave_api_url)
-            if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-                user_traffic_data["Remnawave (общий)"] = "Не удалось авторизоваться"
-            else:
-                user_data = await remna.get_user_by_uuid(remnawave_client_id)
-                if not user_data:
-                    user_traffic_data["Remnawave (общий)"] = "Клиент не найден"
-                else:
-                    user_traffic = user_data.get("userTraffic", {})
-                    used_bytes = user_traffic.get("usedTrafficBytes", 0)
-                    used_gb = round(used_bytes / 1073741824, 2)
-                    user_traffic_data["Remnawave (общий)"] = used_gb
-        except Exception as e:
-            user_traffic_data["Remnawave (общий)"] = f"Ошибка: {e}"
+    if remnawave_client_id and remnawave_server_ref:
+        profile = await get_remnawave_profile(session, str(remnawave_server_ref), remnawave_client_id, fallback_any=True)
+        if not profile:
+            user_traffic_data["Remnawave (общий)"] = "Данные недоступны"
+        else:
+            used_gb = profile.get("used_gb")
+            user_traffic_data["Remnawave (общий)"] = round(float(used_gb), 2) if used_gb is not None else 0
 
     return {"status": "success", "traffic": user_traffic_data}
 
@@ -154,12 +149,18 @@ async def reset_traffic_in_cluster(cluster_id: str, email: str, session: AsyncSe
 
                 client_id = row[0]
 
-                remna = RemnawaveAPI(api_url)
-                if not await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-                    logger.warning(f"[Reset Traffic] Не удалось авторизоваться в Remnawave ({server_name})")
-                    continue
+                async def _reset(api):
+                    done = await api.reset_user_traffic(client_id)
+                    if done:
+                        await invalidate_remnawave_profile(
+                            session,
+                            str(server_name or cluster_id),
+                            str(client_id),
+                            fallback_any=True,
+                        )
+                    return done
 
-                tasks.append(remna.reset_user_traffic(client_id))
+                tasks.append(with_remnawave_api(session, server_name or cluster_id, _reset, fallback_any=True))
                 remnawave_done = True
                 continue
 

handlers/keys/operations/update.py

@@ -5,7 +5,8 @@ from datetime import datetime, timezone
 from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
+from config import PUBLIC_LINK, SUPERNODE
+from panels.remnawave_runtime import invalidate_remnawave_profile, with_remnawave_api
 from database import filter_cluster_by_subgroup, filter_cluster_by_tariff, get_servers, get_tariff_by_id, store_key
 from handlers.utils import ALLOWED_GROUP_CODES
 from database.models import Key, Tariff
@@ -17,7 +18,6 @@ from logger import (
     PANEL_XUI,
 )
 from panels._3xui import ClientConfig, add_client, get_xui_instance
-from panels.remnawave import RemnawaveAPI
 
 from .aggregated_links import make_aggregated_link
 from .deletion import delete_key_from_cluster
@@ -92,58 +92,67 @@ async def update_key_on_cluster(
 
         if remnawave_servers:
             inbound_ids = [s["inbound_id"] for s in remnawave_servers if s.get("inbound_id")]
-            remna = RemnawaveAPI(remnawave_servers[0]["api_url"])
-            if await remna.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD):
-                await remna.delete_user(client_id)
+            group_code = remnawave_servers[0].get("tariff_group")
+            if not group_code:
+                raise ValueError("У Remnawave-сервера отсутствует tariff_group")
 
-                group_code = remnawave_servers[0].get("tariff_group")
-                if not group_code:
-                    raise ValueError("У Remnawave-сервера отсутствует tariff_group")
+            _ = await session.execute(
+                select(Tariff)
+                .where(Tariff.group_code == group_code, Tariff.is_active.is_(True))
+                .order_by(Tariff.duration_days.desc())
+                .limit(1)
+            )
 
-                _ = await session.execute(
-                    select(Tariff)
-                    .where(Tariff.group_code == group_code, Tariff.is_active.is_(True))
-                    .order_by(Tariff.duration_days.desc())
-                    .limit(1)
+            short_uuid = None
+            if remnawave_link and "/" in remnawave_link:
+                short_uuid = remnawave_link.rstrip("/").split("/")[-1]
+                logger.debug(f"{PANEL_REMNA} Извлечен short_uuid: {short_uuid}")
+
+            user_data = {
+                "username": email,
+                "trafficLimitStrategy": "NO_RESET",
+                "expireAt": expire_iso,
+                "telegramId": tg_id,
+                "activeInternalSquads": inbound_ids,
+                "uuid": client_id,
+            }
+
+            if external_squad_uuid:
+                user_data["activeExternalSquads"] = [external_squad_uuid]
+                user_data["activeExternalSquadUuids"] = [external_squad_uuid]
+                user_data["externalSquadUuid"] = external_squad_uuid
+
+            if traffic_limit is not None:
+                user_data["trafficLimitBytes"] = traffic_limit * 1024**3
+            if device_limit is not None:
+                user_data["hwidDeviceLimit"] = device_limit
+            if short_uuid:
+                user_data["shortUuid"] = short_uuid
+                logger.debug(f"{PANEL_REMNA} Добавлен short_uuid: {short_uuid}")
+
+            async def _recreate(api):
+                await api.delete_user(client_id)
+                return await api.create_user(user_data)
+
+            remna_result = await with_remnawave_api(
+                session,
+                str(remnawave_servers[0].get("server_name") or cluster_id),
+                _recreate,
+                fallback_any=True,
+                timeout_sec=12.0,
+            )
+            if remna_result:
+                remnawave_client_id = remna_result.get("uuid")
+                remnawave_link_value = remna_result.get("subscriptionUrl")
+                await invalidate_remnawave_profile(
+                    session,
+                    str(remnawave_servers[0].get("server_name") or cluster_id),
+                    str(remnawave_client_id or client_id),
+                    fallback_any=True,
                 )
-
-                short_uuid = None
-                if remnawave_link and "/" in remnawave_link:
-                    short_uuid = remnawave_link.rstrip("/").split("/")[-1]
-                    logger.debug(f"{PANEL_REMNA} Извлечен short_uuid: {short_uuid}")
-
-                user_data = {
-                    "username": email,
-                    "trafficLimitStrategy": "NO_RESET",
-                    "expireAt": expire_iso,
-                    "telegramId": tg_id,
-                    "activeInternalSquads": inbound_ids,
-                    "uuid": client_id,
-                }
-
-                if external_squad_uuid:
-                    user_data["activeExternalSquads"] = [external_squad_uuid]
-                    user_data["activeExternalSquadUuids"] = [external_squad_uuid]
-                    user_data["externalSquadUuid"] = external_squad_uuid
-
-                if traffic_limit is not None:
-                    user_data["trafficLimitBytes"] = traffic_limit * 1024**3
-                if device_limit is not None:
-                    user_data["hwidDeviceLimit"] = device_limit
-                if short_uuid:
-                    user_data["shortUuid"] = short_uuid
-                    logger.debug(f"{PANEL_REMNA} Добавлен short_uuid: {short_uuid}")
-
-                result = await remna.create_user(user_data)
-                if result:
-                    remnawave_client_id = result.get("uuid")
-                    remnawave_link_value = result.get("subscriptionUrl")
-
-                    logger.info(f"{PANEL_REMNA} Клиент заново создан, uuid={remnawave_client_id}")
-                else:
-                    logger.error(f"{PANEL_REMNA} Ошибка создания клиента")
+                logger.info(f"{PANEL_REMNA} Клиент заново создан, uuid={remnawave_client_id}")
             else:
-                logger.error(f"{PANEL_REMNA} Не удалось авторизоваться")
+                logger.error(f"{PANEL_REMNA} Не удалось авторизоваться/создать клиента")
 
         if not remnawave_client_id:
             logger.warning(f"{PANEL_REMNA} client_id не получен, используем исходный {client_id}")
@@ -244,6 +253,8 @@ async def update_subscription(
     else:
         logger.warning("[LOG] update_subscription: tariff_id отсутствует!")
 
+    from middlewares.session import release_session_early
+    await release_session_early(session)
     await delete_key_from_cluster(old_cluster_id, email, client_id, session=session)
     await session.execute(delete(Key).where(Key.tg_id == tg_id, Key.email == email))
     await session.commit()

handlers/keys/subscriptions.py

@@ -20,12 +20,19 @@ from config import (
     USE_COUNTRY_SELECTION,
 )
 from core.bootstrap import MODES_CONFIG
+from core.cache_config import (
+    SUBSCRIPTION_HANDLER_CONCURRENCY,
+    SUBSCRIPTION_RESPONSE_CACHE_TTL_SEC,
+)
+from core.redis_cache import cache_get, cache_key, cache_set
 from database import get_key_details, get_servers
 from database.models import Server
 from handlers.texts import HAPP_ANNOUNCE, HIDDIFY_PROFILE_TITLE, SUBSCRIPTION_INFO_TEXT, V2RAYTUN_ANNOUNCE
 from handlers.utils import convert_to_bytes
 from logger import logger
 
+_subscription_semaphore = asyncio.Semaphore(SUBSCRIPTION_HANDLER_CONCURRENCY)
+
 
 async def fetch_url_content(url: str, identifier: str) -> tuple[list[str], dict[str, str]]:
     try:
@@ -246,41 +253,55 @@ async def handle_subscription(request: web.Request) -> web.Response:
     if not email or not tg_id:
         return web.Response(text="❌ Неверные параметры запроса.", status=400)
 
+    cache_key_sub = cache_key("sub_response", email, tg_id)
+    cached = await cache_get(cache_key_sub)
+    if isinstance(cached, dict) and "b" in cached and "h" in cached:
+        return web.Response(text=cached["b"], headers=cached["h"])
+
     sessionmaker = request.app["sessionmaker"]
 
-    async with sessionmaker() as session:
-        try:
-            key = await get_key_details(session, email)
-            if not key:
-                return web.Response(text="❌ Клиент с таким email не найден.", status=404)
+    async with _subscription_semaphore:
+        async with sessionmaker() as session:
+            try:
+                key = await get_key_details(session, email)
+                if not key:
+                    return web.Response(text="❌ Клиент с таким email не найден.", status=404)
 
-            if int(tg_id) != int(key["tg_id"]):
-                return web.Response(text="❌ Неверные данные. Получите свой ключ в боте.", status=403)
+                if int(tg_id) != int(key["tg_id"]):
+                    return web.Response(text="❌ Неверные данные. Получите свой ключ в боте.", status=403)
 
-            expiry_time_ms = key["expiry_time"]
-            server_id = key["server_id"]
-            remnawave_link = key["remnawave_link"]
+                expiry_time_ms = key["expiry_time"]
+                server_id = key["server_id"]
+                remnawave_link = key["remnawave_link"]
 
-            time_left = format_time_left(expiry_time_ms)
+                time_left = format_time_left(expiry_time_ms)
 
-            urls = await get_subscription_urls(server_id, email, session, include_remnawave_key=remnawave_link)
-            if not urls:
-                return web.Response(text="❌ Сервер не найден.", status=404)
+                urls = await get_subscription_urls(server_id, email, session, include_remnawave_key=remnawave_link)
+                if not urls:
+                    return web.Response(text="❌ Сервер не найден.", status=404)
 
-            query_string = request.query_string
-            combined_subscriptions, headers_list = await combine_unique_lines(urls, tg_id or email, query_string)
+                query_string = request.query_string
+                combined_subscriptions, headers_list = await combine_unique_lines(urls, tg_id or email, query_string)
 
-            cleaned_subscriptions = [clean_subscription_line(line) for line in combined_subscriptions]
+                cleaned_subscriptions = [clean_subscription_line(line) for line in combined_subscriptions]
 
-            base64_encoded = base64.b64encode("\n".join(cleaned_subscriptions).encode("utf-8")).decode("utf-8")
-            subscription_info = SUBSCRIPTION_INFO_TEXT.format(email=email, time_left=time_left)
+                base64_encoded = base64.b64encode("\n".join(cleaned_subscriptions).encode("utf-8")).decode("utf-8")
+                subscription_info = SUBSCRIPTION_INFO_TEXT.format(email=email, time_left=time_left)
 
-            user_agent = request.headers.get("User-Agent", "")
-            subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms, headers_list)
-            headers = prepare_headers(user_agent, PROJECT_NAME, subscription_info, subscription_userinfo)
+                user_agent = request.headers.get("User-Agent", "")
+                subscription_userinfo = calculate_traffic(cleaned_subscriptions, expiry_time_ms, headers_list)
+                headers = prepare_headers(user_agent, PROJECT_NAME, subscription_info, subscription_userinfo)
 
-            return web.Response(text=base64_encoded, headers=headers)
+                await session.commit()
 
-        except Exception as e:
-            logger.error(f"Ошибка в handle_subscription: {e}", exc_info=True)
-            return web.Response(text=f"❌ Ошибка сервера: {e}", status=500)
+                await cache_set(
+                    cache_key_sub,
+                    {"b": base64_encoded, "h": dict(headers)},
+                    SUBSCRIPTION_RESPONSE_CACHE_TTL_SEC,
+                )
+                return web.Response(text=base64_encoded, headers=headers)
+
+            except Exception as e:
+                await session.rollback()
+                logger.error(f"Ошибка в handle_subscription: {e}", exc_info=True)
+                return web.Response(text=f"❌ Ошибка сервера: {e}", status=500)

handlers/notifications/general_notifications.py

@@ -9,6 +9,7 @@ from sqlalchemy import select, text, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from config import (
+    EXECUTOR_POOL_SIZE,
     NOTIFICATION_TIME,
     NOTIFY_10H_ENABLED,
     NOTIFY_10H_HOURS,
@@ -26,6 +27,7 @@ from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG
 from database import (
     add_notification,
     check_notification_time,
+    check_notification_time_bulk,
     check_notifications_bulk,
     delete_key,
     delete_notification,
@@ -60,9 +62,15 @@ from handlers.texts import (
 from handlers.utils import format_hours, format_minutes, get_russian_month
 from hooks.hooks import run_hooks
 from logger import logger
+from middlewares.session import release_session_early, wrap_session
 
 from .hot_leads_notifications import notify_hot_leads
-from .notify_utils import prepare_key_expiry_data, send_messages_with_limit, send_notification
+from .notify_utils import (
+    NotificationRateLimiter,
+    prepare_key_expiry_data,
+    send_messages_with_limit,
+    send_notification,
+)
 from .special_notifications import notify_inactive_trial_users, notify_users_no_traffic
 
 
@@ -366,6 +374,7 @@ async def try_auto_renew(ctx: NotificationContext, key) -> tuple[bool, Optional[
 
     key_subgroup = current_tariff.get("subgroup_title")
 
+    await release_session_early(ctx.session)
     await renew_key_in_cluster(
         cluster_id=server_id,
         email=email,
@@ -416,6 +425,7 @@ async def notify_expiring_keys(
     notify_type: str,
     photo: str,
     notify_renew_enabled: bool,
+    sessionmaker: Optional[async_sessionmaker] = None,
 ):
     if min_hours > 0:
         logger.info(f"Начало проверки подписок, истекающих через {min_hours}-{max_hours} часов.")
@@ -436,7 +446,15 @@ async def notify_expiring_keys(
     allowed = await check_notifications_bulk(ctx.session, notify_type, max_hours, tg_ids=tg_ids, emails=emails)
     allowed_set = {(user["tg_id"], user["email"]) for user in allowed}
 
+    notify_pairs = [
+        (key.tg_id, f"{(key.email or '')}_{notify_type}")
+        for key in expiring_keys
+        if (key.tg_id, key.email or "") in allowed_set
+    ]
+    can_notify_set = await check_notification_time_bulk(ctx.session, notify_pairs, max_hours)
+
     messages = []
+    renew_candidates: list[tuple[Any, str]] = []
 
     for key in expiring_keys:
         tg_id = key.tg_id
@@ -446,24 +464,11 @@ async def notify_expiring_keys(
             continue
 
         notification_id = f"{email}_{notify_type}"
-
-        can_notify = await check_notification_time(ctx.session, tg_id, notification_id, hours=max_hours)
-        if not can_notify:
+        if (tg_id, notification_id) not in can_notify_set:
             continue
 
         if notify_renew_enabled:
-            try:
-                renewed, tariff, new_expiry = await try_auto_renew(ctx, key)
-
-                if renewed and tariff and new_expiry:
-                    await send_renewed_notification(ctx, key, tariff, new_expiry)
-                    await add_notification(ctx.session, tg_id, notification_id)
-                else:
-                    await send_cannot_renew(ctx, key, photo)
-                    await add_notification(ctx.session, tg_id, notification_id)
-
-            except Exception as error:
-                logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}")
+            renew_candidates.append((key, notification_id))
         else:
             expiry_data = await prepare_key_expiry_data(key, ctx.session, ctx.current_time)
             notification_text = KEY_EXPIRY.format(
@@ -483,6 +488,70 @@ async def notify_expiring_keys(
                 "email": email,
             })
 
+    
+    renew_results: list[tuple[Any, str, bool, Optional[dict], Optional[int]]] = [] 
+    use_parallel = (
+        notify_renew_enabled
+        and renew_candidates
+        and sessionmaker is not None
+        and EXECUTOR_POOL_SIZE > 1
+    )
+    if use_parallel:
+        semaphore = asyncio.Semaphore(EXECUTOR_POOL_SIZE)
+
+        async def do_one_renew(key: Any, notification_id: str) -> tuple[Any, str, bool, Optional[dict], Optional[int]]:
+            async with semaphore:
+                async with sessionmaker() as session:
+                    session = wrap_session(session, sessionmaker)
+                    ctx_key = NotificationContext(
+                        bot=ctx.bot,
+                        session=session,
+                        current_time=ctx.current_time,
+                        preload_data=ctx.preload_data,
+                        bulk_updates=None,
+                    )
+                    try:
+                        renewed, tariff, new_expiry = await try_auto_renew(ctx_key, key)
+                        await session.commit()
+                        return (key, notification_id, bool(renewed), tariff, new_expiry)
+                    except Exception as error:
+                        logger.error(
+                            "Ошибка авто-продления для пользователя %s (%s): %s",
+                            key.tg_id,
+                            getattr(key, "email", ""),
+                            error,
+                        )
+                        return (key, notification_id, False, None, None)
+
+        tasks = [do_one_renew(key, nid) for key, nid in renew_candidates]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        for r in results:
+            if isinstance(r, Exception):
+                logger.error("Ошибка в задаче продления: %s", r)
+                continue
+            renew_results.append(r)
+    else:
+
+        for key, notification_id in renew_candidates:
+            tg_id = key.tg_id
+            try:
+                renewed, tariff, new_expiry = await try_auto_renew(ctx, key)
+                renew_results.append((key, notification_id, bool(renewed), tariff, new_expiry))
+            except Exception as error:
+                logger.error(f"Ошибка авто-продления/уведомления для пользователя {tg_id}: {error}")
+                renew_results.append((key, notification_id, False, None, None))
+
+    renew_rate_limiter = NotificationRateLimiter(max_rate=30, window=1.0)
+    for key, notification_id, renewed, tariff, new_expiry in renew_results:
+        tg_id = key.tg_id
+        await renew_rate_limiter.acquire()
+        if renewed and tariff and new_expiry:
+            await send_renewed_notification(ctx, key, tariff, new_expiry)
+            await add_notification(ctx.session, tg_id, notification_id)
+        else:
+            await send_cannot_renew(ctx, key, photo)
+            await add_notification(ctx.session, tg_id, notification_id)
+
     if messages:
         results = await send_messages_with_limit(ctx.bot, messages, session=ctx.session)
         sent_count = 0
@@ -569,6 +638,8 @@ async def handle_expired_keys(ctx: NotificationContext, keys: list):
 
 
 async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
+    from middlewares.session import wrap_session
+
     while True:
         notification_interval = int(NOTIFICATIONS_CONFIG.get("BASE_NOTIFICATION_MINUTE", NOTIFICATION_TIME))
 
@@ -579,14 +650,23 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
 
         async with notification_lock:
             try:
-                async with sessionmaker() as session:
+
+                current_time = int(datetime.now(moscow_tz).timestamp() * 1000)
+                start_time = datetime.now()
+                preload_data = None
+                keys = []
+                bulk_updates = {
+                    "balance_changes": {},
+                    "key_expiry_updates": [],
+                    "key_tariff_updates": [],
+                    "notifications_to_add": [],
+                    "notifications_to_delete": [],
+                }
+
+                async with sessionmaker() as preload_session:
                     logger.info("Запуск обработки уведомлений")
-
-                    current_time = int(datetime.now(moscow_tz).timestamp() * 1000)
-                    start_time = datetime.now()
-
                     try:
-                        preload_data = await preload_notification_data(session)
+                        preload_data = await preload_notification_data(preload_session)
                         keys_data = preload_data["keys_data"]
                         keys = [data["key"] for data in keys_data.values()]
                         preload_time = (datetime.now() - start_time).total_seconds()
@@ -594,30 +674,21 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
                             f"Предзагружено данных: {len(keys)} ключей, "
                             f"{len(preload_data['tariffs_cache'])} тарифов за {preload_time:.2f}s"
                         )
-
-                        bulk_updates = {
-                            "balance_changes": {},
-                            "key_expiry_updates": [],
-                            "key_tariff_updates": [],
-                            "notifications_to_add": [],
-                            "notifications_to_delete": [],
-                        }
-
                     except Exception as error:
                         logger.error(f"Ошибка при предварительной загрузке данных: {error}")
                         try:
-                            keys = await get_all_keys(session=session)
+                            keys = await get_all_keys(session=preload_session)
                             keys = [k for k in keys if not k.is_frozen]
                             preload_data = None
-                            bulk_updates = None
                             preload_time = (datetime.now() - start_time).total_seconds()
                             logger.info(f"Fallback: получено {len(keys)} ключей за {preload_time:.2f}s")
                         except Exception as fallback_error:
                             logger.error(f"Ошибка fallback получения ключей: {fallback_error}")
                             keys = []
                             preload_data = None
-                            bulk_updates = None
 
+                async with sessionmaker() as session:
+                    session = wrap_session(session, sessionmaker)
                     ctx = NotificationContext(
                         bot=bot,
                         session=session,
@@ -653,6 +724,7 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
                                 notify_type="key_24h",
                                 photo="notify_24h.jpg",
                                 notify_renew_enabled=notify_renew_enabled,
+                                sessionmaker=sessionmaker,
                             )
                         except Exception as error:
                             logger.error(f"Ошибка в notify_expiring_keys (24h): {error}")
@@ -667,6 +739,7 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
                                 notify_type="key_10h",
                                 photo="notify_10h.jpg",
                                 notify_renew_enabled=notify_renew_enabled,
+                                sessionmaker=sessionmaker,
                             )
                         except Exception as error:
                             logger.error(f"Ошибка в notify_expiring_keys (10h): {error}")
@@ -710,6 +783,7 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
 
                     total_time = (datetime.now() - start_time).total_seconds()
                     logger.info(f"Уведомления завершены за {total_time:.2f}s")
+                    await session.commit()
 
             except Exception as error:
                 logger.error(f"Ошибка в periodic_notifications: {error}")

handlers/start.py

@@ -22,6 +22,8 @@ from config import (
     TRIAL_TIME_DISABLE,
 )
 from core.bootstrap import BUTTONS_CONFIG, MODES_CONFIG
+from core.cache_config import START_UTM_EXISTS_TTL_SEC
+from core.redis_cache import cache_get, cache_key, cache_set
 from database import (
     add_user,
     get_coupon_by_code,
@@ -261,8 +263,13 @@ async def prompt_subscription(callback: CallbackQuery):
 
 
 async def handle_utm_link(utm_code: str, message: Message, state: FSMContext, session: AsyncSession, user_data: dict):
-    res = await session.execute(select(TrackingSource).where(TrackingSource.code == utm_code))
-    if not res.scalar_one_or_none():
+    is_known = await cache_get(cache_key("utm_exists", utm_code))
+    if is_known is None:
+        res = await session.execute(select(TrackingSource).where(TrackingSource.code == utm_code))
+        is_known = res.scalar_one_or_none() is not None
+        await cache_set(cache_key("utm_exists", utm_code), bool(is_known), START_UTM_EXISTS_TTL_SEC)
+
+    if not is_known:
         await message.answer("❌ UTM ссылка не найдена.")
         return
     await upsert_source_if_empty(session, user_data["tg_id"], utm_code)

handlers/tariffs/addons/key_addons_main.py

@@ -851,6 +851,7 @@ async def handle_addons_confirm(callback: CallbackQuery, state: FSMContext, sess
             f"old_subgroup={old_subgroup}"
         )
 
+        await session.release_early()
         await renew_key_in_cluster(
             cluster_id=server_id,
             email=email,

handlers/tariffs/addons/key_addons_pack.py

@@ -827,6 +827,7 @@ async def handle_addons_confirm(callback: CallbackQuery, state: FSMContext, sess
                 f"old_subgroup={old_subgroup}"
             )
 
+            await session.release_early()
             await renew_key_in_cluster(
                 cluster_id=server_id,
                 email=email,

handlers/tariffs/tariff_display.py

@@ -2,9 +2,9 @@ from typing import Any
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD
+from panels.remnawave_runtime import get_remnawave_profile
 from core.settings.tariffs_config import TARIFFS_CONFIG, normalize_tariff_config
-from database import get_servers, get_tariff_by_id
+from database import get_tariff_by_id
 from database.models import Key
 from handlers.texts import key_message_success
 from logger import logger
@@ -191,56 +191,26 @@ async def get_key_tariff_display(
 
     if server_cluster_id and client_id:
         try:
-            servers = await get_servers(session)
-            cluster_servers = servers.get(server_cluster_id) or servers.get(str(server_cluster_id)) or []
-            remna_server = next((srv for srv in cluster_servers if srv.get("panel_type") == "remnawave"), None)
-            if not remna_server:
-                remna_server = next(
-                    (srv for cl in servers.values() for srv in cl if srv.get("panel_type") == "remnawave"),
-                    None,
-                )
+            profile = await get_remnawave_profile(session, str(server_cluster_id), client_id, fallback_any=True)
+            if profile:
+                panel_traffic_limit_bytes = profile.get("traffic_limit_bytes")
+                panel_device_limit = profile.get("hwid_device_limit")
 
-            if remna_server:
-                from panels.remnawave import RemnawaveAPI
-
-                api = RemnawaveAPI(remna_server["api_url"])
-                try:
-                    ok = await api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD)
-                except Exception as e:
-                    logger.warning(f"[KeyTariffDisplay] Remnawave login error for {client_id}: {e}")
-                    ok = False
-
-                if ok:
+                if panel_traffic_limit_bytes is not None:
                     try:
-                        user_data = await api.get_user_by_uuid(client_id)
-                    except Exception as e:
-                        logger.warning(f"[KeyTariffDisplay] Remnawave get_user_by_uuid error for {client_id}: {e}")
-                        user_data = None
+                        traffic_limit_bytes = int(panel_traffic_limit_bytes)
+                    except (TypeError, ValueError):
+                        logger.warning(
+                            f"[KeyTariffDisplay] Invalid trafficLimitBytes from Remnawave for {client_id}: {panel_traffic_limit_bytes}"
+                        )
 
-                    if user_data:
-                        panel_traffic_limit_bytes = user_data.get("trafficLimitBytes")
-                        panel_device_limit = user_data.get("hwidDeviceLimit")
-
-                        if panel_traffic_limit_bytes is not None:
-                            try:
-                                traffic_limit_bytes = int(panel_traffic_limit_bytes)
-                            except (TypeError, ValueError):
-                                logger.warning(
-                                    f"[KeyTariffDisplay] Invalid trafficLimitBytes from Remnawave for {client_id}: {panel_traffic_limit_bytes}"
-                                )
-
-                        if panel_device_limit is not None:
-                            try:
-                                device_limit = int(panel_device_limit)
-                            except (TypeError, ValueError):
-                                logger.warning(
-                                    f"[KeyTariffDisplay] Invalid hwidDeviceLimit from Remnawave for {client_id}: {panel_device_limit}"
-                                )
-
-                try:
-                    await api.aclose()
-                except Exception:
-                    pass
+                if panel_device_limit is not None:
+                    try:
+                        device_limit = int(panel_device_limit)
+                    except (TypeError, ValueError):
+                        logger.warning(
+                            f"[KeyTariffDisplay] Invalid hwidDeviceLimit from Remnawave for {client_id}: {panel_device_limit}"
+                        )
         except Exception as e:
             logger.warning(f"[KeyTariffDisplay] Error while overriding limits from panel: {e}")
 

middlewares/__init__.py

@@ -2,6 +2,7 @@ from collections.abc import Iterable
 
 from aiogram import BaseMiddleware, Dispatcher
 
+from core.bootstrap import MODES_CONFIG
 from middlewares.ban_checker import BanCheckerMiddleware
 from middlewares.subscription import SubscriptionMiddleware
 
@@ -12,6 +13,7 @@ from .direct_start_blocker import DirectStartBlockerMiddleware
 from .loggings import LoggingMiddleware
 from .maintenance import MaintenanceModeMiddleware
 from .probe import MiddlewareProbe, StreamProbeMiddleware, TailHandlerProbe
+from .runtime_config_sync import RuntimeConfigSyncMiddleware
 from .session import SessionMiddleware
 from .throttling import ThrottlingMiddleware
 from .user import UserMiddleware
@@ -30,18 +32,47 @@ def register_middleware(
     def wrap(mw, name: str):
         return MiddlewareProbe(mw, name) if PROBE_LOGGING else mw
 
+    exclude_set = set(exclude or [])
+
+    flag_by_name = {
+        "runtime_config_sync": "RUNTIME_CONFIG_SYNC_MIDDLEWARE_ENABLED",
+        "concurrency": "CONCURRENCY_MIDDLEWARE_ENABLED",
+        "subscription": "SUBSCRIPTION_MIDDLEWARE_ENABLED",
+        "session": "SESSION_MIDDLEWARE_ENABLED",
+        "direct_start_blocker": "DIRECT_START_BLOCKER_MIDDLEWARE_ENABLED",
+        "ban_checker": "BAN_CHECKER_MIDDLEWARE_ENABLED",
+        "admin": "ADMIN_MIDDLEWARE_ENABLED",
+        "maintenance": "MAINTENANCE_MIDDLEWARE_ENABLED",
+        "logging": "LOGGING_MIDDLEWARE_ENABLED",
+        "throttling": "THROTTLING_MIDDLEWARE_ENABLED",
+        "user": "USER_MIDDLEWARE_ENABLED",
+        "answer": "ANSWER_MIDDLEWARE_ENABLED",
+    }
+
+    def middleware_enabled(name: str) -> bool:
+        if name in exclude_set:
+            return False
+        flag_name = flag_by_name.get(name)
+        if not flag_name:
+            return True
+        return bool(MODES_CONFIG.get(flag_name, True))
+
     if PROBE_LOGGING:
         dispatcher.update.outer_middleware(StreamProbeMiddleware("global"))
 
-    if sessionmaker:
+    if middleware_enabled("runtime_config_sync"):
+        dispatcher.update.outer_middleware(wrap(RuntimeConfigSyncMiddleware(), "runtime_config_sync"))
+    if sessionmaker and middleware_enabled("concurrency"):
         dispatcher.update.outer_middleware(wrap(ConcurrencyLimiterMiddleware(), "concurrency"))
+    if middleware_enabled("subscription"):
+        dispatcher.update.outer_middleware(wrap(SubscriptionMiddleware(), "subscription"))
+    if sessionmaker and middleware_enabled("session"):
         dispatcher.update.outer_middleware(wrap(SessionMiddleware(sessionmaker), "session"))
+    if middleware_enabled("direct_start_blocker"):
+        dispatcher.update.outer_middleware(wrap(DirectStartBlockerMiddleware(), "direct_start_blocker"))
 
-    dispatcher.update.outer_middleware(wrap(DirectStartBlockerMiddleware(), "direct_start_blocker"))
-
-    dispatcher.update.outer_middleware(wrap(SubscriptionMiddleware(), "subscription"))
-
-    dispatcher.update.outer_middleware(wrap(BanCheckerMiddleware(), "ban_checker"))
+    if middleware_enabled("ban_checker"):
+        dispatcher.update.outer_middleware(wrap(BanCheckerMiddleware(), "ban_checker"))
 
     if middlewares is None:
         available_middlewares = {
@@ -52,8 +83,9 @@ def register_middleware(
             "user": UserMiddleware(),
             "answer": CallbackAnswerMiddleware(),
         }
-        exclude_set = set(exclude or [])
-        middlewares = [wrap(mw, name) for name, mw in available_middlewares.items() if name not in exclude_set]
+        middlewares = [
+            wrap(mw, name) for name, mw in available_middlewares.items() if middleware_enabled(name)
+        ]
     else:
         wrapped = []
         for mw in middlewares:

middlewares/admin.py

@@ -3,16 +3,16 @@ from typing import Any
 
 from aiogram import BaseMiddleware
 from aiogram.types import CallbackQuery, Message, TelegramObject
-from cachetools import TTLCache
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import ADMIN_ID
+from core.cache_config import ADMIN_CACHE_TTL_SEC
+from core.redis_cache import cache_get, cache_key, cache_set
 from database.models import Admin
 
 
-_ADMIN_CACHE_TTL = 60
-_admin_cache: TTLCache[int, bool] = TTLCache(maxsize=10_000, ttl=_ADMIN_CACHE_TTL)
+_ADMIN_CACHE_TTL = ADMIN_CACHE_TTL_SEC
 
 
 class AdminMiddleware(BaseMiddleware):
@@ -55,16 +55,17 @@ class AdminMiddleware(BaseMiddleware):
             if user_id in self._admin_ids:
                 return True
 
-            if user_id in _admin_cache:
-                return _admin_cache[user_id]
+            cached = await cache_get(cache_key("admin_access", user_id))
+            if isinstance(cached, bool):
+                return cached
 
             if not session:
-                _admin_cache[user_id] = False
+                await cache_set(cache_key("admin_access", user_id), False, _ADMIN_CACHE_TTL)
                 return False
 
             result = await session.execute(select(Admin).where(Admin.tg_id == user_id))
             is_admin = result.scalar_one_or_none() is not None
-            _admin_cache[user_id] = is_admin
+            await cache_set(cache_key("admin_access", user_id), bool(is_admin), _ADMIN_CACHE_TTL)
             return is_admin
         except Exception:
             return False

middlewares/ban_checker.py

@@ -4,24 +4,48 @@ from typing import Any
 
 from aiogram import BaseMiddleware
 from aiogram.types import CallbackQuery, Message, TelegramObject, Update
-from cachetools import TTLCache
 from pytz import timezone
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import ADMIN_ID, SUPPORT_CHAT_URL
+from core.cache_config import BAN_CACHE_TTL_SEC
+from core.redis_cache import cache_get, cache_key, cache_set
+from database import async_session_maker
 from database.models import ManualBan
 from logger import logger
 
 
 TZ = timezone("Europe/Moscow")
-_BAN_CACHE_TTL = 30
-_ban_cache: TTLCache[int, tuple[float, dict | None]] = TTLCache(maxsize=50_000, ttl=_BAN_CACHE_TTL)
+_BAN_CACHE_TTL = BAN_CACHE_TTL_SEC
 
 
 class BanCheckerMiddleware(BaseMiddleware):
     """Проверка банов."""
 
+    async def _load_ban_info(self, session: AsyncSession, tg_id: int) -> dict[str, Any] | None:
+        query = (
+            select(ManualBan.reason, ManualBan.until)
+            .where(
+                ManualBan.tg_id == tg_id,
+                (ManualBan.until.is_(None)) | (ManualBan.until > datetime.utcnow()),
+            )
+            .limit(1)
+        )
+        result = await session.execute(query)
+        row = result.first()
+        if row:
+            reason, until = row
+            await cache_set(
+                cache_key("ban_status", tg_id),
+                {"has_ban": True, "reason": reason or "не указана", "until": until.isoformat() if until else None},
+                _BAN_CACHE_TTL,
+            )
+            return {"reason": reason or "не указана", "until": until}
+
+        await cache_set(cache_key("ban_status", tg_id), {"has_ban": False}, _BAN_CACHE_TTL)
+        return None
+
     async def __call__(
         self,
         handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
@@ -45,33 +69,30 @@ class BanCheckerMiddleware(BaseMiddleware):
         if tg_id is None:
             return await handler(event, data)
 
-        now_ts = datetime.utcnow().timestamp()
-        cached = _ban_cache.get(tg_id)
-        if cached and cached[0] > now_ts:
-            ban_info = cached[1]
+        cached = await cache_get(cache_key("ban_status", tg_id))
+        if isinstance(cached, dict):
+            if not cached.get("has_ban"):
+                ban_info = None
+            else:
+                until_raw = cached.get("until")
+                until_parsed = None
+                if isinstance(until_raw, str):
+                    try:
+                        until_parsed = datetime.fromisoformat(until_raw)
+                    except ValueError:
+                        until_parsed = None
+                ban_info = {
+                    "reason": cached.get("reason") or "не указана",
+                    "until": until_parsed,
+                }
         else:
             session = data.get("session")
-            if not isinstance(session, AsyncSession):
-                logger.error("[BanChecker] session отсутствует в data")
-                return await handler(event, data)
-
-            query = (
-                select(ManualBan.reason, ManualBan.until)
-                .where(
-                    ManualBan.tg_id == tg_id,
-                    (ManualBan.until.is_(None)) | (ManualBan.until > datetime.utcnow()),
-                )
-                .limit(1)
-            )
-            result = await session.execute(query)
-            row = result.first()
-            if row:
-                reason, until = row
-                ban_info = {"reason": reason or "не указана", "until": until}
+            if session is not None and getattr(session, "execute", None) is not None:
+                ban_info = await self._load_ban_info(session, tg_id)
             else:
-                ban_info = None
-
-            _ban_cache[tg_id] = (now_ts + _BAN_CACHE_TTL, ban_info)
+                async with async_session_maker() as short_session:
+                    ban_info = await self._load_ban_info(short_session, tg_id)
+                    await short_session.commit()
 
         if not ban_info:
             return await handler(event, data)

middlewares/concurrency.py

@@ -6,6 +6,8 @@ from typing import Any
 from aiogram import BaseMiddleware, Bot
 from aiogram.types import CallbackQuery, Message, TelegramObject
 
+from core.cache_config import CONCURRENCY_REJECT_NOTICE_TTL_SEC
+from core.redis_cache import cache_key, cache_setnx
 from database.db import CONCURRENT_UPDATES_LIMIT, MAX_UPDATE_AGE_SEC
 
 
@@ -17,6 +19,7 @@ class ConcurrencyLimiterMiddleware(BaseMiddleware):
 
     def __init__(self) -> None:
         self._semaphore = asyncio.Semaphore(CONCURRENT_UPDATES_LIMIT)
+        self._notice_ttl = CONCURRENCY_REJECT_NOTICE_TTL_SEC
 
     async def __call__(
         self,
@@ -38,7 +41,13 @@ class ConcurrencyLimiterMiddleware(BaseMiddleware):
     async def _reject_stale(self, event: TelegramObject, data: dict[str, Any]) -> None:
         if isinstance(event, CallbackQuery):
             bot: Bot = data.get("bot")
-            if bot:
+            uid = event.from_user.id if event.from_user else None
+            should_notify = (
+                bot
+                and uid is not None
+                and await cache_setnx(cache_key("concurrency_notice", uid), 1, self._notice_ttl)
+            )
+            if should_notify:
                 try:
                     await bot.answer_callback_query(
                         event.id,
@@ -49,7 +58,13 @@ class ConcurrencyLimiterMiddleware(BaseMiddleware):
                     pass
         elif isinstance(event, Message) and event.text and event.chat:
             bot: Bot = data.get("bot")
-            if bot:
+            uid = event.from_user.id if event.from_user else None
+            should_notify = (
+                bot
+                and uid is not None
+                and await cache_setnx(cache_key("concurrency_notice", uid), 1, self._notice_ttl)
+            )
+            if should_notify:
                 try:
                     await bot.send_message(
                         event.chat.id,

middlewares/direct_start_blocker.py

@@ -1,20 +1,18 @@
-import time
 from collections.abc import Awaitable, Callable
 from typing import Any
 
 from aiogram import BaseMiddleware
 from aiogram.types import Message, Update
-from cachetools import TTLCache
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import ADMIN_ID, DISABLE_DIRECT_START
 from core.bootstrap import MODES_CONFIG
+from core.cache_config import DIRECT_START_USER_EXISTS_CACHE_TTL_SEC
+from core.redis_cache import cache_get, cache_key, cache_set
 from database import check_user_exists
 from logger import logger
 
 
-_TTL = 20
-_cache_user_exists: TTLCache[int, tuple[float, bool]] = TTLCache(maxsize=50_000, ttl=_TTL)
+_TTL = DIRECT_START_USER_EXISTS_CACHE_TTL_SEC
 
 
 class DirectStartBlockerMiddleware(BaseMiddleware):
@@ -50,24 +48,23 @@ class DirectStartBlockerMiddleware(BaseMiddleware):
                 return await handler(event, data)
 
         session = data.get("session")
-        if not isinstance(session, AsyncSession):
+        if session is None or not hasattr(session, "execute"):
             return await handler(event, data)
 
         tg_id = message.from_user.id
         text = message.text.strip()
-        now = time.time()
         user_in_data = bool(data.get("user"))
 
         async def user_exists_cached() -> bool:
             if user_in_data:
                 return True
 
-            cached = _cache_user_exists.get(tg_id)
-            if cached is not None and cached[0] > now:
-                return cached[1]
+            cached = await cache_get(cache_key("direct_start_user_exists", tg_id))
+            if isinstance(cached, bool):
+                return cached
 
             exists = await check_user_exists(session, tg_id)
-            _cache_user_exists[tg_id] = (now + _TTL, exists)
+            await cache_set(cache_key("direct_start_user_exists", tg_id), bool(exists), _TTL)
             return exists
 
         if not text.startswith("/"):

middlewares/runtime_config_sync.py

@@ -0,0 +1,21 @@
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject
+
+from core.settings.runtime_sync import maybe_sync_runtime_configs
+
+
+class RuntimeConfigSyncMiddleware(BaseMiddleware):
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        try:
+            await maybe_sync_runtime_configs()
+        except Exception:
+            pass
+        return await handler(event, data)

middlewares/session.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import time
 from typing import Any
 
@@ -18,6 +20,11 @@ async def release_session_early(session: Any) -> bool:
     return False
 
 
+def wrap_session(session: AsyncSession, maker) -> "_SessionProxy":
+    """Оборачивает сессию в прокси с release_early (для фоновых задач вроде periodic_notifications)."""
+    return _SessionProxy(session, maker, {})
+
+
 class _SessionProxy:
     __slots__ = ("_session", "_maker", "_released", "_data")
 
@@ -47,10 +54,15 @@ class _SessionProxy:
         import asyncio
 
         async with self._maker() as s:
-            result = getattr(s, method)(*args, **kwargs)
-            if asyncio.iscoroutine(result):
-                return await result
-            return result
+            try:
+                result = getattr(s, method)(*args, **kwargs)
+                if asyncio.iscoroutine(result):
+                    result = await result
+                await s.commit()
+                return result
+            except Exception:
+                await s.rollback()
+                raise
 
     def __getattr__(self, name: str):
         if name in ("_session", "_maker", "_released", "_data", "release_early", "_with_short_session"):
@@ -92,6 +104,7 @@ class SessionMiddleware(BaseMiddleware):
             proxy = _SessionProxy(session, self.sessionmaker, data)
             data["session"] = proxy
             committed = False
+            rolled_back = False
             try:
                 result = await handler(event, data)
                 if data.get("_session_released_early"):
@@ -111,6 +124,7 @@ class SessionMiddleware(BaseMiddleware):
                         exc_info=True,
                     )
                     await self._rollback(session, "commit failure")
+                    rolled_back = True
                     return result
             except Exception as e:
                 logger.warning(
@@ -122,9 +136,10 @@ class SessionMiddleware(BaseMiddleware):
                     exc_info=True,
                 )
                 await self._rollback(session, "handler failure")
+                rolled_back = True
                 raise
             finally:
-                if not committed and not data.get("_session_released_early"):
+                if not committed and not rolled_back and not data.get("_session_released_early"):
                     try:
                         await session.rollback()
                     except Exception:

middlewares/subscription.py

@@ -8,9 +8,13 @@ from aiogram.fsm.context import FSMContext
 from aiogram.types import InlineKeyboardButton, Message, Update
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 
-from bot import bot
 from config import CHANNEL_EXISTS, CHANNEL_ID, CHANNEL_REQUIRED, CHANNEL_URL
 from core.bootstrap import MODES_CONFIG
+from core.cache_config import (
+    SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC,
+    SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC,
+)
+from core.redis_cache import cache_delete, cache_get, cache_key, cache_set
 from handlers.buttons import SUB_CHANELL, SUB_CHANELL_DONE
 from handlers.texts import SUBSCRIPTION_REQUIRED_MSG
 from handlers.utils import edit_or_send_message
@@ -18,6 +22,10 @@ from logger import logger
 
 
 class SubscriptionMiddleware(BaseMiddleware):
+    def __init__(self) -> None:
+        self._subscribed_ttl = SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC
+        self._unsubscribed_ttl = SUBSCRIPTION_CACHE_UNSUBSCRIBED_TTL_SEC
+
     async def __call__(
         self,
         handler: Callable[[Update, dict[str, Any]], Awaitable[Any]],
@@ -57,19 +65,51 @@ class SubscriptionMiddleware(BaseMiddleware):
         else:
             return await handler(event, data)
 
+        cached_status = await self._get_cached_status(tg_id)
+        if cached_status is False:
+            logger.info(f"[SubMiddleware] Пользователь {tg_id} не подписан (cache)")
+            await self._store_user_state(data, message, from_user)
+            return await self._ask_to_subscribe(message)
+        if cached_status is True:
+            return await handler(event, data)
+
+        bot = data.get("bot")
+        if bot is None:
+            return await handler(event, data)
+
         try:
             member = await bot.get_chat_member(CHANNEL_ID, tg_id)
-            if member.status not in ("member", "administrator", "creator"):
+            is_subscribed = member.status in ("member", "administrator", "creator")
+            await self._cache_status(tg_id, is_subscribed)
+            if not is_subscribed:
                 logger.info(f"[SubMiddleware] Пользователь {tg_id} не подписан")
                 await self._store_user_state(data, message, from_user)
                 return await self._ask_to_subscribe(message)
         except (TelegramBadRequest, TelegramForbiddenError) as e:
-            logger.warning(f"[SubMiddleware] Ошибка при проверке подписки {tg_id}: {e}")
-            await self._store_user_state(data, message, from_user)
-            return await self._ask_to_subscribe(message)
+            logger.warning(f"[SubMiddleware] Ошибка проверки подписки {tg_id}, пропускаем: {e}")
+            return await handler(event, data)
 
         return await handler(event, data)
 
+    async def _get_cached_status(self, tg_id: int) -> bool | None:
+        subscribed_key = cache_key("subscribed", tg_id)
+        unsubscribed_key = cache_key("unsubscribed", tg_id)
+        if await cache_get(subscribed_key) is not None:
+            return True
+        if await cache_get(unsubscribed_key) is not None:
+            return False
+        return None
+
+    async def _cache_status(self, tg_id: int, is_subscribed: bool) -> None:
+        subscribed_key = cache_key("subscribed", tg_id)
+        unsubscribed_key = cache_key("unsubscribed", tg_id)
+        if is_subscribed:
+            await cache_delete(unsubscribed_key)
+            await cache_set(subscribed_key, 1, self._subscribed_ttl)
+        else:
+            await cache_delete(subscribed_key)
+            await cache_set(unsubscribed_key, 1, self._unsubscribed_ttl)
+
     async def _store_user_state(self, data: dict, message: Message, from_user):
         state: FSMContext = data.get("state")
         if not state or not from_user or from_user.is_bot:

middlewares/throttling.py

@@ -1,12 +1,18 @@
 from aiogram import BaseMiddleware, Bot
 from aiogram.types import CallbackQuery
-from cachetools import TTLCache
+
+from core.cache_config import (
+    THROTTLE_CACHE_TTL_SEC,
+    THROTTLE_NOTICE_TTL_SEC,
+)
+from core.redis_cache import cache_incr, cache_key, cache_setnx
+from hashlib import sha1
 
 
 class ThrottlingMiddleware(BaseMiddleware):
     def __init__(self) -> None:
-        self.cache = TTLCache(maxsize=50_000, ttl=1.0)
-        self.throttle_notice_cache = TTLCache(maxsize=50_000, ttl=1.0)
+        self._counter_ttl = THROTTLE_CACHE_TTL_SEC
+        self._notice_ttl = THROTTLE_NOTICE_TTL_SEC
 
     async def __call__(self, handler, event, data):
         if not isinstance(event, CallbackQuery):
@@ -17,11 +23,13 @@ class ThrottlingMiddleware(BaseMiddleware):
             return await handler(event, data)
 
         key = (user_id, event.data or "")
-        current_count = self.cache.get(key, 0)
+        key_hash = sha1(key[1].encode("utf-8")).hexdigest()
+        counter_key = cache_key("throttle_counter", user_id, key_hash)
+        current_count = await cache_incr(counter_key, self._counter_ttl)
 
         if current_count >= 2:
-            if key not in self.throttle_notice_cache:
-                self.throttle_notice_cache[key] = None
+            notice_key = cache_key("throttle_notice", user_id, key_hash)
+            if await cache_setnx(notice_key, 1, self._notice_ttl):
                 bot: Bot = data["bot"]
                 await bot.answer_callback_query(
                     callback_query_id=event.id,
@@ -30,5 +38,4 @@ class ThrottlingMiddleware(BaseMiddleware):
                 )
             return
 
-        self.cache[key] = current_count + 1
         return await handler(event, data)

middlewares/user.py

@@ -5,10 +5,10 @@ from typing import Any
 
 from aiogram import BaseMiddleware
 from aiogram.types import TelegramObject, User
-from cachetools import TTLCache
 from sqlalchemy import update
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.redis_cache import cache_get, cache_key, cache_set
 from database import upsert_user
 from database.models import User as DbUser
 from logger import logger
@@ -17,9 +17,7 @@ from logger import logger
 class UserMiddleware(BaseMiddleware):
     def __init__(self, debounce_sec: float = 60.0, cache_maxsize: int = 100_000) -> None:
         self._debounce = float(debounce_sec)
-        self._cache: TTLCache[int, tuple[str, float, float, dict | None]] = TTLCache(
-            maxsize=cache_maxsize, ttl=debounce_sec * 2
-        )
+        self._cache_ttl = debounce_sec * 2
 
     async def __call__(
         self,
@@ -31,7 +29,7 @@ class UserMiddleware(BaseMiddleware):
             user: User | None = data.get("event_from_user")
             if user and not user.is_bot:
                 session = data.get("session")
-                if isinstance(session, AsyncSession):
+                if session is not None and getattr(session, "execute", None) is not None:
                     db_user = await self._process_user(user, session)
                     if db_user:
                         data["user"] = db_user
@@ -43,15 +41,28 @@ class UserMiddleware(BaseMiddleware):
         uid = user.id
         fingerprint = self._fingerprint(user)
         now = monotonic()
+        key = cache_key("user_middleware", uid)
 
-        cached = self._cache.get(uid)
-        if cached:
-            cached_fingerprint, profile_ts, touch_ts, cached_db_user = cached
+        cached = await cache_get(key)
+        if isinstance(cached, dict):
+            cached_fingerprint = str(cached.get("fingerprint") or "")
+            profile_ts = float(cached.get("profile_ts") or 0.0)
+            touch_ts = float(cached.get("touch_ts") or 0.0)
+            cached_db_user = cached.get("db_user")
 
             if fingerprint == cached_fingerprint:
                 if now - touch_ts >= self._debounce:
                     db_user = await self._touch_user(uid, session)
-                    self._cache[uid] = (cached_fingerprint, profile_ts, now, db_user or cached_db_user)
+                    await cache_set(
+                        key,
+                        {
+                            "fingerprint": cached_fingerprint,
+                            "profile_ts": profile_ts,
+                            "touch_ts": now,
+                            "db_user": db_user or cached_db_user,
+                        },
+                        self._cache_ttl,
+                    )
                     return db_user or cached_db_user
 
                 if now - profile_ts < self._debounce:
@@ -67,7 +78,16 @@ class UserMiddleware(BaseMiddleware):
             session=session,
             only_if_exists=True,
         )
-        self._cache[uid] = (fingerprint, now, now, db_user)
+        await cache_set(
+            key,
+            {
+                "fingerprint": fingerprint,
+                "profile_ts": now,
+                "touch_ts": now,
+                "db_user": db_user,
+            },
+            self._cache_ttl,
+        )
         return db_user
 
     async def _touch_user(self, tg_id: int, session: AsyncSession) -> dict | None:

panels/remnawave_runtime.py

@@ -0,0 +1,208 @@
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from config import REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, REMNAWAVE_TOKEN_LOGIN_ENABLED
+from core.cache_config import (
+    REMNAWAVE_MAX_CONCURRENCY,
+    REMNAWAVE_ACTION_TIMEOUT_SEC,
+    REMNAWAVE_PROFILE_CACHE_TTL_SEC,
+    REMNAWAVE_PROFILE_TIMEOUT_SEC,
+    REMNAWAVE_SERVER_CACHE_TTL_SEC,
+)
+from core.redis_cache import cache_delete_pattern, cache_get, cache_key, cache_set
+from database import get_servers
+from logger import logger
+from panels.remnawave import RemnawaveAPI
+
+_remnawave_semaphore = asyncio.Semaphore(REMNAWAVE_MAX_CONCURRENCY)
+
+
+def invalidate_remnawave_profile_cache(*, api_url: str | None = None, client_id: str | None = None) -> None:
+    """Invalidate cached Remnawave profiles by api_url/client_id (or both)."""
+    import asyncio
+
+    async def _invalidate_async() -> None:
+        if api_url is None and client_id is None:
+            await cache_delete_pattern("remna_profile:*")
+            return
+        if api_url is not None and client_id is not None:
+            await cache_delete_pattern(f"remna_profile:{api_url}:{client_id}")
+            return
+        if api_url is not None:
+            await cache_delete_pattern(f"remna_profile:{api_url}:*")
+            return
+        await cache_delete_pattern(f"remna_profile:*:{client_id}")
+
+    try:
+        loop = asyncio.get_running_loop()
+        loop.create_task(_invalidate_async())
+    except RuntimeError:
+        return
+
+
+async def resolve_remnawave_api_url(
+    session: AsyncSession,
+    server_ref: str,
+    *,
+    fallback_any: bool = False,
+) -> str | None:
+    ckey = cache_key("remna_server", str(server_ref), int(bool(fallback_any)))
+    cached_api_url = await cache_get(ckey)
+    if isinstance(cached_api_url, str) or cached_api_url is None:
+        if cached_api_url is not None:
+            return cached_api_url
+
+    servers = await get_servers(session)
+    ref = str(server_ref)
+    remna_server = None
+
+    cluster_servers = servers.get(ref) or servers.get(str(ref)) or []
+    remna_server = next((srv for srv in cluster_servers if srv.get("panel_type") == "remnawave"), None)
+
+    if remna_server is None:
+        for cluster_name, cluster in servers.items():
+            for srv in cluster:
+                if (srv.get("server_name") == ref or str(cluster_name) == ref) and srv.get("panel_type") == "remnawave":
+                    remna_server = srv
+                    break
+            if remna_server:
+                break
+
+    if remna_server is None and fallback_any:
+        remna_server = next((srv for cluster in servers.values() for srv in cluster if srv.get("panel_type") == "remnawave"), None)
+
+    api_url = remna_server.get("api_url") if remna_server else None
+    await cache_set(ckey, api_url, REMNAWAVE_SERVER_CACHE_TTL_SEC)
+    return api_url
+
+
+async def get_remnawave_profile(
+    session: AsyncSession,
+    server_ref: str,
+    client_id: str,
+    *,
+    fallback_any: bool = False,
+) -> dict[str, Any] | None:
+    api_url = await resolve_remnawave_api_url(session, server_ref, fallback_any=fallback_any)
+    if not api_url:
+        return None
+
+    pkey = cache_key("remna_profile", api_url, client_id)
+    cached_profile = await cache_get(pkey)
+    if isinstance(cached_profile, dict) or cached_profile is None:
+        if cached_profile is not None:
+            return cached_profile
+
+    profile: dict[str, Any] | None = None
+
+    async with _remnawave_semaphore:
+        api = RemnawaveAPI(api_url)
+        try:
+            logged_in = True
+            if not REMNAWAVE_TOKEN_LOGIN_ENABLED:
+                logged_in = await asyncio.wait_for(
+                    api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD),
+                    timeout=REMNAWAVE_PROFILE_TIMEOUT_SEC,
+                )
+            if not logged_in:
+                await cache_set(pkey, None, REMNAWAVE_PROFILE_CACHE_TTL_SEC)
+                return None
+
+            devices = await asyncio.wait_for(
+                api.get_user_hwid_devices(client_id),
+                timeout=REMNAWAVE_PROFILE_TIMEOUT_SEC,
+            )
+            user_data = await asyncio.wait_for(
+                api.get_user_by_uuid(client_id),
+                timeout=REMNAWAVE_PROFILE_TIMEOUT_SEC,
+            )
+
+            hwid_count = len(devices or [])
+            used_gb = None
+            traffic_limit_bytes = None
+            hwid_device_limit = None
+
+            if user_data:
+                user_traffic = user_data.get("userTraffic", {})
+                used_bytes = user_traffic.get("usedTrafficBytes", 0)
+                used_gb = round(used_bytes / 1073741824, 1)
+                traffic_limit_bytes = user_data.get("trafficLimitBytes")
+                hwid_device_limit = user_data.get("hwidDeviceLimit")
+
+            profile = {
+                "api_url": api_url,
+                "hwid_count": hwid_count,
+                "used_gb": used_gb,
+                "traffic_limit_bytes": traffic_limit_bytes,
+                "hwid_device_limit": hwid_device_limit,
+            }
+        except asyncio.TimeoutError:
+            logger.warning(f"[Remnawave] Таймаут профиля для client_id={client_id}")
+            profile = None
+        except Exception as e:
+            logger.warning(f"[Remnawave] Ошибка профиля для client_id={client_id}: {e}")
+            profile = None
+        finally:
+            if hasattr(api, "aclose"):
+                try:
+                    await api.aclose()
+                except Exception:
+                    pass
+
+    await cache_set(pkey, profile, REMNAWAVE_PROFILE_CACHE_TTL_SEC)
+    return profile
+
+
+async def invalidate_remnawave_profile(
+    session: AsyncSession,
+    server_ref: str,
+    client_id: str,
+    *,
+    fallback_any: bool = False,
+) -> None:
+    api_url = await resolve_remnawave_api_url(session, server_ref, fallback_any=fallback_any)
+    if api_url:
+        invalidate_remnawave_profile_cache(api_url=api_url, client_id=client_id)
+    else:
+        invalidate_remnawave_profile_cache(client_id=client_id)
+
+
+async def with_remnawave_api(
+    session: AsyncSession,
+    server_ref: str,
+    operation: Callable[[RemnawaveAPI], Awaitable[Any]],
+    *,
+    fallback_any: bool = False,
+    timeout_sec: float = REMNAWAVE_ACTION_TIMEOUT_SEC,
+) -> Any | None:
+    api_url = await resolve_remnawave_api_url(session, server_ref, fallback_any=fallback_any)
+    if not api_url:
+        return None
+
+    async with _remnawave_semaphore:
+        api = RemnawaveAPI(api_url)
+        try:
+            logged_in = True
+            if not REMNAWAVE_TOKEN_LOGIN_ENABLED:
+                logged_in = await asyncio.wait_for(
+                    api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD),
+                    timeout=timeout_sec,
+                )
+            if not logged_in:
+                return None
+            return await asyncio.wait_for(operation(api), timeout=timeout_sec)
+        except asyncio.TimeoutError:
+            logger.warning(f"[Remnawave] Таймаут операции для server_ref={server_ref}")
+            return None
+        except Exception as e:
+            logger.warning(f"[Remnawave] Ошибка операции для server_ref={server_ref}: {e}")
+            return None
+        finally:
+            if hasattr(api, "aclose"):
+                try:
+                    await api.aclose()
+                except Exception:
+                    pass

requirements.txt

@@ -50,6 +50,7 @@ python-dateutil==2.9.0.post0
 pytz==2025.1
 qrcode==8.2
 requests==2.32.4
+redis==5.2.0
 rich==14.1.0
 robokassa==0.3.2
 ruff==0.9.5

servers.py

@@ -7,11 +7,11 @@ from datetime import datetime, timedelta
 from aiogram.types import InlineKeyboardButton
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 from ping3 import ping
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from bot import bot
 from config import ADMIN_ID, PING_TIME
-from database import get_servers
+from core.executor import get_thread_pool
+from database import async_session_maker, get_servers
 from handlers.admin.servers.keyboard import AdminServerCallback
 from logger import logger
 
@@ -22,10 +22,19 @@ notified_servers = set()
 PING_SEMAPHORE = asyncio.Semaphore(3)
 
 
+def _sync_ping(server_ip: str, timeout: float = 3):
+    """Синхронный ping для вызова в пуле потоков."""
+    return ping(server_ip, timeout=timeout)
+
+
 async def ping_server(server_ip: str) -> bool:
     async with PING_SEMAPHORE:
         try:
-            response = await asyncio.to_thread(ping, server_ip, timeout=3)
+            loop = asyncio.get_running_loop()
+            response = await loop.run_in_executor(
+                get_thread_pool(),
+                lambda: _sync_ping(server_ip, 3),
+            )
             if response is not None and response is not False:
                 return True
             return await check_tcp_connection(server_ip, 443)
@@ -86,13 +95,15 @@ async def notify_admin(server_name: str, status: str, down_duration: timedelta =
         await bot.send_message(admin_id, message, reply_markup=builder.as_markup())
 
 
-async def check_servers(session: AsyncSession):
+async def check_servers(sessionmaker=None):
     """
     Периодическая проверка серверов.
-    Использует asyncio.gather() для ускорения.
+    Использует короткую сессию на итерацию, чтобы не держать транзакцию во время ping.
     """
+    maker = sessionmaker or async_session_maker
     while True:
-        servers = await get_servers(session=session)
+        async with maker() as session:
+            servers = await get_servers(session=session)
         current_time = datetime.now()
 
         tasks = []

utils/backup.py

@@ -10,7 +10,6 @@ import aiofiles
 from aiogram import Bot
 from aiogram.types import BufferedInputFile
 
-from bot import bot
 from config import (
     ADMIN_ID,
     BACKUP_CAPTION,
@@ -36,17 +35,23 @@ from logger import logger
 async def backup_database() -> Exception | None:
     """
     Создает резервную копию базы данных (или полный архив) и отправляет его администраторам.
+    Блокирующие операции (pg_dump и т.д.) выполняются в пуле процессов, не блокируя event loop и используя другие ядра CPU.
 
     Returns:
         Optional[Exception]: Исключение в случае ошибки или None при успешном выполнении
     """
+    import asyncio
+    from core.executor import get_process_pool
+
+    loop = asyncio.get_event_loop()
+    pool = get_process_pool()
     if BACKUP_CREATE_ARCHIVE:
         if not any([BACKUP_INCLUDE_DB, BACKUP_INCLUDE_CONFIG, BACKUP_INCLUDE_TEXTS, BACKUP_INCLUDE_IMG]):
-            backup_file_path, exception = _create_database_backup()
+            backup_file_path, exception = await loop.run_in_executor(pool, _create_database_backup)
         else:
-            backup_file_path, exception = _create_backup_archive()
+            backup_file_path, exception = await loop.run_in_executor(pool, _create_backup_archive)
     else:
-        backup_file_path, exception = _create_database_backup()
+        backup_file_path, exception = await loop.run_in_executor(pool, _create_database_backup)
 
     if exception:
         logger.error(f"Ошибка при создании бэкапа: {exception}")
@@ -74,11 +79,12 @@ def _create_database_backup() -> tuple[str | None, Exception | None]:
         Tuple[Optional[str], Optional[Exception]]: Путь к файлу бэкапа и исключение (если произошла ошибка)
     """
     date_formatted = datetime.now().strftime("%Y-%m-%d-%H%M%S")
+    pid_suffix = os.getpid()
 
     backup_dir = Path(BACK_DIR)
     backup_dir.mkdir(parents=True, exist_ok=True)
 
-    filename = backup_dir / f"{DB_NAME}-backup-{date_formatted}.sql"
+    filename = backup_dir / f"{DB_NAME}-backup-{date_formatted}-{pid_suffix}.sql"
 
     try:
         os.environ["PGPASSWORD"] = DB_PASSWORD
@@ -123,10 +129,11 @@ def _create_backup_archive() -> tuple[str | None, Exception | None]:
         Tuple[Optional[str], Optional[Exception]]: Путь к файлу архива и исключение (если произошла ошибка)
     """
     date_formatted = datetime.now().strftime("%Y-%m-%d-%H%M%S")
+    pid_suffix = os.getpid()
     backup_dir = Path(BACK_DIR)
     backup_dir.mkdir(parents=True, exist_ok=True)
 
-    archive_path = backup_dir / f"{DB_NAME}-full-backup-{date_formatted}.tar.gz"
+    archive_path = backup_dir / f"{DB_NAME}-full-backup-{date_formatted}-{pid_suffix}.tar.gz"
     project_root = Path(__file__).parent.parent
     archive_folder = f"backup-{date_formatted}"
 
@@ -242,6 +249,8 @@ async def _send_backup_to_admins(backup_file_path: str) -> None:
     if not backup_file_path or not os.path.exists(backup_file_path):
         raise FileNotFoundError(f"Файл бэкапа не найден: {backup_file_path}")
 
+    from bot import bot
+
     async def send_default():
         for admin_id in ADMIN_ID:
             try:

utils/button_icons.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import aiogram.types
 from pydantic import ConfigDict
 
-
 _UniqueGiftColors = getattr(aiogram.types, "UniqueGiftColors", None)
 if _UniqueGiftColors is not None:
     _cfg = getattr(_UniqueGiftColors, "model_config", None)

utils/errors.py

@@ -97,6 +97,7 @@ def setup_error_handlers(dp: Dispatcher) -> None:
                                 admin=False,
                                 captcha=False,
                             )
+                            await session.commit()
                     elif event.update.callback_query:
                         fsm_context = dp.fsm.get_context(
                             bot=bot,
@@ -111,6 +112,7 @@ def setup_error_handlers(dp: Dispatcher) -> None:
                                 admin=False,
                                 captcha=False,
                             )
+                            await session.commit()
                 except Exception as e:
                     logger.error(f"Ошибка при показе стартового меню после ошибки: {e}", exc_info=True)
 
@@ -150,6 +152,7 @@ def setup_error_handlers(dp: Dispatcher) -> None:
                         admin=False,
                         captcha=False,
                     )
+                    await session.commit()
             elif event.update.callback_query:
                 fsm_context = dp.fsm.get_context(
                     bot=bot,
@@ -164,6 +167,7 @@ def setup_error_handlers(dp: Dispatcher) -> None:
                         admin=False,
                         captcha=False,
                     )
+                    await session.commit()
 
         except TelegramBadRequest as exception:
             logger.warning(f"Не удалось отправить детали ошибки: {exception}")