redis cache/ webhook abuse/ separate streams for sending and requesting panels/ increased security and more

core/cache_config.py

@@ -1,7 +1,7 @@
 UPDATE_STALE_AGE_SEC = 60
 
 CONCURRENCY_MAX_WAIT_SEC = 300
-CONCURRENCY_LIMIT = 25
+CONCURRENCY_LIMIT = 200
 
 SUBSCRIPTION_CACHE_SUBSCRIBED_MAXSIZE = 200_000
 SUBSCRIPTION_CACHE_SUBSCRIBED_TTL_SEC = 300
@@ -25,7 +25,7 @@ USER_EXISTS_CACHE_MAXSIZE = 150_000
 USER_EXISTS_CACHE_TTL_SEC = 60
 
 BAN_CACHE_MAXSIZE = 50_000
-BAN_CACHE_TTL_SEC = 30
+BAN_CACHE_TTL_SEC = 60
 
 DIRECT_START_USER_EXISTS_CACHE_MAXSIZE = 50_000
 DIRECT_START_USER_EXISTS_CACHE_TTL_SEC = 20
@@ -37,7 +37,8 @@ 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_PROFILE_TIMEOUT_SEC = 10.0
+REMNAWAVE_PROFILE_ERROR_CACHE_TTL_SEC = 45
 REMNAWAVE_ACTION_TIMEOUT_SEC = 5.0
 REMNAWAVE_MAX_CONCURRENCY = 20
 
@@ -51,7 +52,19 @@ SERVERS_CACHE_TTL_SEC = 60
 TARIFF_BY_ID_CACHE_TTL_SEC = 120
 TARIFFS_FOR_CLUSTER_CACHE_TTL_SEC = 120
 
+KEY_DETAILS_CACHE_TTL_SEC = 45
+KEYS_LIST_CACHE_TTL_SEC = 25
+KEY_COUNT_CACHE_TTL_SEC = 25
+
+BALANCE_CACHE_TTL_SEC = 25
+PROFILE_DATA_CACHE_TTL_SEC = 25
+
+PAYMENT_PENDING_CACHE_TTL_SEC = 3600
 
 ERROR_THROTTLE_WINDOW_SEC = 60
 ERROR_THROTTLE_MAX_KEYS = 500
 ERROR_THROTTLE_MESSAGE_MAX_LEN = 120
+
+WEBHOOK_ABUSE_FAIL_THRESHOLD = 10
+WEBHOOK_ABUSE_FAIL_WINDOW_SEC = 60
+WEBHOOK_ABUSE_BLOCK_TTL_SEC = 300

core/executor.py

@@ -77,6 +77,20 @@ def shutdown_process_pool() -> None:
         logger.debug("[Executor] Пул процессов остановлен")
 
 
+def should_run_heavy_tasks_separately() -> bool:
+    """
+    True, если есть запас по ядрам/потокам — тогда рассылка и уведомления
+    можно выносить в отдельный поток/ядро.
+    """
+    try:
+        from config import EXECUTOR_POOL_SIZE
+        pool_size = max(1, int(EXECUTOR_POOL_SIZE))
+    except Exception:
+        pool_size = 1
+    cpu_count = multiprocessing.cpu_count() or 1
+    return cpu_count >= 2 or pool_size >= 2
+
+
 async def run_io(fn: Callable[..., T], *args: object) -> T:
     """Выполняет fn(*args) в пуле потоков (I/O). Один вызов для всех блокирующих операций."""
     loop = asyncio.get_running_loop()

core/webhook_abuse.py

@@ -0,0 +1,52 @@
+from aiohttp import web
+
+from core.cache_config import (
+    WEBHOOK_ABUSE_BLOCK_TTL_SEC,
+    WEBHOOK_ABUSE_FAIL_THRESHOLD,
+    WEBHOOK_ABUSE_FAIL_WINDOW_SEC,
+)
+from core.redis_cache import cache_delete, cache_get, cache_incr, cache_key, cache_set
+
+
+def get_webhook_client_ip(request: web.Request) -> str:
+    """IP клиента: X-Forwarded-For (первый) или X-Real-IP, иначе request.remote."""
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        return forwarded.split(",")[0].strip() or "unknown"
+    real = request.headers.get("X-Real-IP")
+    if real:
+        return real.strip() or "unknown"
+    if request.remote:
+        s = str(request.remote)
+        if "%" in s:
+            s = s.split("%")[0]
+        if ":" in s:
+            s = s.rsplit(":", 1)[0]
+        return s or "unknown"
+    return "unknown"
+
+
+async def is_webhook_ip_blocked(ip: str) -> bool:
+    """True, если IP временно заблокирован из‑за множества невалидных подписей."""
+    if not ip or ip == "unknown":
+        return False
+    try:
+        block_key = cache_key("webhook_abuse_block", ip)
+        return (await cache_get(block_key)) is not None
+    except Exception:
+        return False
+
+
+async def record_webhook_signature_failure(ip: str) -> None:
+    """Увеличивает счётчик неудачных проверок подписи для IP; при превышении порога блокирует IP."""
+    if not ip or ip == "unknown":
+        return
+    try:
+        fail_key = cache_key("webhook_abuse_fail", ip)
+        count = await cache_incr(fail_key, WEBHOOK_ABUSE_FAIL_WINDOW_SEC)
+        if count >= WEBHOOK_ABUSE_FAIL_THRESHOLD:
+            block_key = cache_key("webhook_abuse_block", ip)
+            await cache_set(block_key, 1, WEBHOOK_ABUSE_BLOCK_TTL_SEC)
+            await cache_delete(fail_key)
+    except Exception:
+        pass

database/keys.py

@@ -1,14 +1,47 @@
 from datetime import datetime
+from types import SimpleNamespace
 
 from sqlalchemy import delete, func, select, text, update
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.cache_config import (
+    KEY_COUNT_CACHE_TTL_SEC,
+    KEY_DETAILS_CACHE_TTL_SEC,
+    KEYS_LIST_CACHE_TTL_SEC,
+)
+from core.redis_cache import cache_delete, cache_get, cache_key, cache_set
 from database.models import Key, User
-from database.users import invalidate_user_snapshot
+from database.users import invalidate_profile_cache, invalidate_user_snapshot
 from logger import logger
 
 
+async def invalidate_key_details(email: str) -> None:
+    await cache_delete(cache_key("key_details", email))
+
+
+async def invalidate_key_email(client_id: str) -> None:
+    await cache_delete(cache_key("key_email", client_id))
+
+
+async def invalidate_keys_list(tg_id: int) -> None:
+    await cache_delete(cache_key("keys_list", tg_id))
+    await cache_delete(cache_key("key_count", tg_id))
+    await invalidate_profile_cache(tg_id)
+
+
+async def invalidate_key_details_by_client_id(session: AsyncSession, client_id: str) -> None:
+    email = await cache_get(cache_key("key_email", client_id))
+    await cache_delete(cache_key("key_email", client_id))
+    if email:
+        await invalidate_key_details(str(email))
+    else:
+        res = await session.execute(select(Key.email).where(Key.client_id == client_id).limit(1))
+        row = res.scalar_one_or_none()
+        if row is not None:
+            await invalidate_key_details(str(row))
+
+
 async def store_key(
     session: AsyncSession,
     tg_id: int,
@@ -83,6 +116,8 @@ async def store_key(
 
         await session.commit()
         invalidate_user_snapshot(tg_id)
+        await invalidate_keys_list(tg_id)
+        await invalidate_key_details(email)
 
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при сохранении ключа: {e}")
@@ -90,9 +125,29 @@ async def store_key(
         raise
 
 
+def _key_to_cache_dict(k: Key) -> dict:
+    return {
+        "email": k.email,
+        "alias": k.alias,
+        "client_id": k.client_id,
+        "expiry_time": int(k.expiry_time) if k.expiry_time is not None else 0,
+        "created_at": int(k.created_at) if k.created_at is not None else 0,
+        "tariff_id": k.tariff_id,
+        "server_id": k.server_id,
+        "is_frozen": bool(k.is_frozen) if k.is_frozen is not None else False,
+    }
+
+
 async def get_keys(session: AsyncSession, tg_id: int):
+    ckey = cache_key("keys_list", tg_id)
+    cached = await cache_get(ckey)
+    if isinstance(cached, list):
+        return [SimpleNamespace(**d) for d in cached]
     result = await session.execute(select(Key).where(Key.tg_id == tg_id))
-    return result.scalars().all()
+    rows = result.scalars().all()
+    serialized = [_key_to_cache_dict(k) for k in rows]
+    await cache_set(ckey, serialized, KEYS_LIST_CACHE_TTL_SEC)
+    return rows
 
 
 async def get_all_keys(session: AsyncSession):
@@ -107,7 +162,12 @@ async def get_key_by_server(session: AsyncSession, tg_id: int, client_id: str):
 
 
 async def get_key_details(session: AsyncSession, email: str) -> dict | None:
-    """Возвращает подробную информацию о ключе по email."""
+    """Возвращает подробную информацию о ключе по email. Горячие данные кэшируются в Redis."""
+    ckey = cache_key("key_details", email)
+    cached = await cache_get(ckey)
+    if isinstance(cached, dict):
+        return cached
+
     stmt = select(Key, User).join(User, Key.tg_id == User.tg_id).where(Key.email == email)
     result = await session.execute(stmt)
     row = result.first()
@@ -127,7 +187,7 @@ async def get_key_details(session: AsyncSession, email: str) -> dict | None:
         hours_left = time_left.seconds // 3600
         days_left_message = f"Осталось часов: <b>{hours_left}</b>"
 
-    return {
+    out = {
         "key": key.key,
         "remnawave_link": key.remnawave_link,
         "server_id": key.server_id,
@@ -151,18 +211,36 @@ async def get_key_details(session: AsyncSession, email: str) -> dict | None:
         "current_device_limit": key.current_device_limit,
         "current_traffic_limit": key.current_traffic_limit,
     }
+    await cache_set(ckey, out, KEY_DETAILS_CACHE_TTL_SEC)
+    if key.client_id:
+        await cache_set(cache_key("key_email", key.client_id), email, KEY_DETAILS_CACHE_TTL_SEC)
+    return out
 
 
 async def get_key_count(session: AsyncSession, tg_id: int) -> int:
+    cached = await cache_get(cache_key("key_count", tg_id))
+    if cached is not None:
+        try:
+            return int(cached)
+        except (TypeError, ValueError):
+            pass
     result = await session.execute(select(func.count()).select_from(Key).where(Key.tg_id == tg_id))
-    return result.scalar() or 0
+    count = result.scalar() or 0
+    await cache_set(cache_key("key_count", tg_id), count, KEY_COUNT_CACHE_TTL_SEC)
+    return count
 
 
 async def delete_key(session: AsyncSession, identifier: int | str, commit: bool = True):
     tg_id_for_cache = None
+    email_for_cache = None
     if isinstance(identifier, str):
-        res = await session.execute(select(Key.tg_id).where(Key.client_id == identifier).limit(1))
-        tg_id_for_cache = res.scalar_one_or_none()
+        res = await session.execute(
+            select(Key.tg_id, Key.email).where(Key.client_id == identifier).limit(1)
+        )
+        row = res.first()
+        if row:
+            tg_id_for_cache, email_for_cache = row[0], row[1]
+        await cache_delete(cache_key("key_email", identifier))
     else:
         tg_id_for_cache = identifier
     stmt = delete(Key).where(Key.tg_id == identifier if isinstance(identifier, int) else Key.client_id == identifier)
@@ -171,12 +249,16 @@ async def delete_key(session: AsyncSession, identifier: int | str, commit: bool
         await session.commit()
     if tg_id_for_cache is not None:
         invalidate_user_snapshot(tg_id_for_cache)
+        await invalidate_keys_list(tg_id_for_cache)
+    if email_for_cache is not None:
+        await invalidate_key_details(str(email_for_cache))
     logger.info(f"Ключ с идентификатором {identifier} удалён")
 
 
 async def update_key_expiry(session: AsyncSession, client_id: str, new_expiry_time: int):
     await session.execute(update(Key).where(Key.client_id == client_id).values(expiry_time=new_expiry_time))
     await session.commit()
+    await invalidate_key_details_by_client_id(session, client_id)
     logger.info(f"Срок действия ключа {client_id} обновлён до {new_expiry_time}")
 
 
@@ -188,6 +270,8 @@ async def get_client_id_by_email(session: AsyncSession, email: str):
 async def update_key_notified(session: AsyncSession, tg_id: int, client_id: str):
     await session.execute(update(Key).where(Key.tg_id == tg_id, Key.client_id == client_id).values(notified=True))
     await session.commit()
+    await invalidate_keys_list(tg_id)
+    await invalidate_key_details_by_client_id(session, client_id)
 
 
 async def mark_key_as_frozen(session: AsyncSession, tg_id: int, client_id: str, time_left: int):
@@ -203,6 +287,8 @@ async def mark_key_as_frozen(session: AsyncSession, tg_id: int, client_id: str,
         ),
         {"expiry": time_left, "tg_id": tg_id, "client_id": client_id},
     )
+    await invalidate_keys_list(tg_id)
+    await invalidate_key_details_by_client_id(session, client_id)
 
 
 async def mark_key_as_unfrozen(
@@ -223,11 +309,14 @@ async def mark_key_as_unfrozen(
         ),
         {"expiry": new_expiry_time, "tg_id": tg_id, "client_id": client_id},
     )
+    await invalidate_keys_list(tg_id)
+    await invalidate_key_details_by_client_id(session, client_id)
 
 
 async def update_key_tariff(session: AsyncSession, client_id: str, tariff_id: int):
     await session.execute(update(Key).where(Key.client_id == client_id).values(tariff_id=tariff_id))
     await session.commit()
+    await invalidate_key_details_by_client_id(session, client_id)
     logger.info(f"Тариф ключа {client_id} обновлён на {tariff_id}")
 
 
@@ -239,6 +328,7 @@ async def get_subscription_link(session: AsyncSession, email: str) -> str | None
 async def update_key_client_id(session: AsyncSession, email: str, new_client_id: str):
     await session.execute(update(Key).where(Key.email == email).values(client_id=new_client_id))
     await session.commit()
+    await invalidate_key_details(email)
     logger.info(f"client_id обновлён для {email} -> {new_client_id}")
 
 
@@ -246,7 +336,10 @@ async def update_key_link(session: AsyncSession, email: str, link: str) -> bool:
     q = update(Key).where(Key.email == email).values(key=link).returning(Key.client_id)
     res = await session.execute(q)
     await session.commit()
-    return res.scalar_one_or_none() is not None
+    ok = res.scalar_one_or_none() is not None
+    if ok:
+        await invalidate_key_details(email)
+    return ok
 
 
 async def save_key_config_with_mode(
@@ -280,6 +373,7 @@ async def save_key_config_with_mode(
         return
 
     await session.execute(update(Key).where(Key.email == email).values(**values))
+    await invalidate_key_details(email)
 
 
 async def reset_key_current_limits_to_selected(session: AsyncSession, client_id: str):
@@ -296,4 +390,5 @@ async def reset_key_current_limits_to_selected(session: AsyncSession, client_id:
         {"client_id": client_id},
     )
     await session.commit()
+    await invalidate_key_details_by_client_id(session, client_id)
     logger.info(f"Текущие лимиты ключа {client_id} сброшены к выбранным")

database/notifications.py

@@ -110,6 +110,30 @@ async def get_last_notification_time(session: AsyncSession, tg_id: int, notifica
     return None
 
 
+async def get_last_notification_times_bulk(
+    session: AsyncSession, pairs: list[tuple[int, str]]
+) -> dict[tuple[int, str], int]:
+    """
+    Один запрос: последние времена уведомлений для списка (tg_id, notification_type).
+    Возвращает dict[(tg_id, notification_type)] -> timestamp_ms.
+    """
+    if not pairs:
+        return {}
+    from sqlalchemy import tuple_
+
+    stmt = select(
+        Notification.tg_id,
+        Notification.notification_type,
+        Notification.last_notification_time,
+    ).where(tuple_(Notification.tg_id, Notification.notification_type).in_(pairs))
+    result = await session.execute(stmt)
+    out = {}
+    for tg_id, ntype, last_time in result.all():
+        if last_time:
+            out[(tg_id, ntype)] = int(last_time.timestamp() * 1000)
+    return out
+
+
 async def check_hot_lead_discount(session: AsyncSession, tg_id: int) -> dict:
     try:
         result = await session.execute(

database/payments.py

@@ -5,6 +5,8 @@ from sqlalchemy import and_, insert, select, update
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from core.cache_config import PAYMENT_PENDING_CACHE_TTL_SEC
+from core.redis_cache import cache_delete, cache_get, cache_key, cache_set
 from database.models import Payment
 from logger import logger
 
@@ -12,6 +14,42 @@ from logger import logger
 MOSCOW_TZ = timezone("Europe/Moscow")
 
 
+def _payment_cache_key(pid: str) -> str:
+    return cache_key("payment_pending", pid)
+
+
+async def register_pending_payment(
+    payment_id: str,
+    tg_id: int,
+    amount: float,
+    payment_system: str,
+    *,
+    currency: str = "RUB",
+    metadata: dict | None = None,
+    original_amount: float | None = None,
+) -> bool:
+    """Регистрирует ожидающий платёж только в Redis. В БД пишем при success/fail из вебхука."""
+    data = {
+        "tg_id": tg_id,
+        "amount": amount,
+        "currency": currency,
+        "status": "pending",
+        "payment_system": payment_system,
+        "payment_id": payment_id,
+        "metadata": metadata,
+        "original_amount": original_amount,
+    }
+    ok = await cache_set(_payment_cache_key(payment_id), data, PAYMENT_PENDING_CACHE_TTL_SEC)
+    if ok:
+        logger.debug(f"[Payments] Pending в кэше: payment_id={payment_id}, tg_id={tg_id}")
+    return ok
+
+
+async def invalidate_payment_cache(payment_id: str) -> None:
+    """Вызвать после сохранения платежа в БД (success/fail) из вебхука."""
+    await cache_delete(_payment_cache_key(payment_id))
+
+
 async def add_payment(
     session: AsyncSession,
     tg_id: int,
@@ -142,6 +180,21 @@ async def update_payment_status(
 
 
 async def get_payment_by_payment_id(session: AsyncSession, pid: str) -> dict | None:
+    """Сначала Redis (pending), затем БД. Из кэша возвращается запись без id — вебхук делает add_payment."""
+    cached = await cache_get(_payment_cache_key(pid))
+    if cached is not None:
+        return {
+            "id": None,
+            "tg_id": cached["tg_id"],
+            "amount": cached["amount"],
+            "currency": cached.get("currency", "RUB"),
+            "status": cached.get("status", "pending"),
+            "payment_system": cached["payment_system"],
+            "payment_id": cached["payment_id"],
+            "created_at": None,
+            "metadata": cached.get("metadata"),
+            "original_amount": cached.get("original_amount"),
+        }
     try:
         result = await session.execute(select(Payment).where(Payment.payment_id == pid).limit(1))
         payment = result.scalar_one_or_none()

database/tariffs.py

@@ -99,6 +99,35 @@ async def get_tariffs(
         return []
 
 
+async def get_tariff_names_groups_subgroups_durations(
+    session: AsyncSession, tariff_ids: list[int]
+) -> tuple[dict[int, str], dict[int, str], dict[int, str | None], dict[int, int]]:
+    """Один запрос: id, name, group_code, subgroup_title, duration_days → четыре словаря."""
+    if not tariff_ids:
+        return {}, {}, {}, {}
+
+    result = await session.execute(
+        select(
+            Tariff.id,
+            Tariff.name,
+            Tariff.group_code,
+            Tariff.subgroup_title,
+            Tariff.duration_days,
+        ).where(Tariff.id.in_(tariff_ids))
+    )
+    rows = result.all()
+    names = {}
+    groups = {}
+    subgroups = {}
+    durations = {}
+    for tid, name, group_code, subgroup_title, duration_days in rows:
+        names[tid] = name
+        groups[tid] = group_code
+        subgroups[tid] = subgroup_title
+        durations[tid] = duration_days
+    return names, groups, subgroups, durations
+
+
 async def get_tariff_by_id(session: AsyncSession, tariff_id: int):
     key = cache_key("tariff", tariff_id)
     cached = await cache_get(key)

database/users.py

@@ -6,6 +6,7 @@ from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.cache_config import (
+    BALANCE_CACHE_TTL_SEC,
     USER_EXISTS_CACHE_TTL_SEC,
     USER_SNAPSHOT_CACHE_TTL_SEC,
 )
@@ -76,6 +77,14 @@ async def add_user(
         raise
 
 
+async def invalidate_balance_cache(tg_id: int) -> None:
+    await cache_delete(cache_key("balance", tg_id))
+
+
+async def invalidate_profile_cache(tg_id: int) -> None:
+    await cache_delete(cache_key("profile_data", tg_id))
+
+
 async def update_balance(session: AsyncSession, tg_id: int, amount: float) -> None:
     try:
         res = await session.execute(
@@ -91,6 +100,8 @@ async def update_balance(session: AsyncSession, tg_id: int, amount: float) -> No
             logger.info(f"[DB] Баланс пользователя {tg_id} обновлён: {old_balance} → {new_balance}")
         else:
             logger.info(f"[DB] Баланс пользователя {tg_id} не изменён: пользователь не найден")
+        await invalidate_balance_cache(tg_id)
+        await invalidate_profile_cache(tg_id)
     except SQLAlchemyError as e:
         logger.error(f"[DB] Ошибка при обновлении баланса пользователя {tg_id}: {e}")
         await session.rollback()
@@ -108,15 +119,25 @@ async def check_user_exists(session: AsyncSession, tg_id: int) -> bool:
 
 
 async def get_balance(session: AsyncSession, tg_id: int) -> float:
+    cached = await cache_get(cache_key("balance", tg_id))
+    if cached is not None:
+        try:
+            return round(float(cached), 1)
+        except (TypeError, ValueError):
+            pass
     result = await session.execute(select(func.coalesce(User.balance, 0.0)).where(User.tg_id == tg_id))
     balance = result.scalar_one_or_none()
-    return round(float(balance or 0.0), 1)
+    value = round(float(balance or 0.0), 1)
+    await cache_set(cache_key("balance", tg_id), value, BALANCE_CACHE_TTL_SEC)
+    return value
 
 
 async def set_user_balance(session: AsyncSession, tg_id: int, balance: float) -> None:
     try:
         await session.execute(update(User).where(User.tg_id == tg_id).values(balance=balance))
         await session.commit()
+        await invalidate_balance_cache(tg_id)
+        await invalidate_profile_cache(tg_id)
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при установке баланса для пользователя {tg_id}: {e}")
         await session.rollback()
@@ -127,6 +148,7 @@ async def update_trial(session: AsyncSession, tg_id: int, status: int):
     try:
         await session.execute(update(User).where(User.tg_id == tg_id).values(trial=status))
         await session.commit()
+        await invalidate_profile_cache(tg_id)
         invalidate_user_snapshot(tg_id)
         logger.info(f"[DB] Триал статус обновлён для пользователя {tg_id}: {status}")
     except SQLAlchemyError as e:
@@ -141,6 +163,45 @@ async def get_trial(session: AsyncSession, tg_id: int) -> int:
     return int(trial or 0)
 
 
+async def get_balance_and_trial(session: AsyncSession, tg_id: int) -> tuple[float, int]:
+    """Один запрос к БД для баланса и триала (профиль при промахе кэша)."""
+    result = await session.execute(
+        select(
+            func.coalesce(User.balance, 0.0),
+            func.coalesce(User.trial, 0),
+        ).where(User.tg_id == tg_id)
+    )
+    row = result.one_or_none()
+    if row is None:
+        return 0.0, 0
+    balance, trial = row
+    return round(float(balance or 0.0), 1), int(trial or 0)
+
+
+async def get_balance_trial_key_count(session: AsyncSession, tg_id: int) -> tuple[float, int, int]:
+    """
+    Один запрос: баланс, триал и число ключей пользователя (для профиля при промахе кэша).
+    Возвращает (balance_rub, trial_status, key_count).
+    """
+    key_count_subq = select(func.count()).select_from(Key).where(Key.tg_id == User.tg_id).scalar_subquery()
+    result = await session.execute(
+        select(
+            func.coalesce(User.balance, 0.0),
+            func.coalesce(User.trial, 0),
+            key_count_subq,
+        ).where(User.tg_id == tg_id)
+    )
+    row = result.one_or_none()
+    if row is None:
+        return 0.0, 0, 0
+    balance, trial, key_count = row
+    return (
+        round(float(balance or 0.0), 1),
+        int(trial or 0),
+        int(key_count or 0),
+    )
+
+
 async def upsert_user(
     session: AsyncSession,
     tg_id: int,

handlers/admin/bans/bans_handler.py

@@ -16,6 +16,7 @@ from database import delete_user_data
 from database.models import BlockedUser, Key, ManualBan
 from filters.admin import IsAdminFilter
 from logger import logger
+from middlewares.ban_checker import invalidate_ban_cache
 
 from ..panel.keyboard import AdminPanelCallback
 from .keyboard import (
@@ -245,8 +246,12 @@ async def handle_clear_shadow_bans(callback_query: CallbackQuery, session: Async
             )
             return
 
+        tg_ids_result = await session.execute(select(ManualBan.tg_id).where(ManualBan.reason == "shadow"))
+        tg_ids_to_invalidate = [r[0] for r in tg_ids_result.all()]
         await session.execute(delete(ManualBan).where(ManualBan.reason == "shadow"))
         await session.commit()
+        for uid in tg_ids_to_invalidate:
+            await invalidate_ban_cache(uid)
 
         await callback_query.message.answer(
             text=f"🗑️ Очищено {total_count} записей теневых банов из базы данных.",
@@ -279,8 +284,14 @@ async def handle_clear_manual_bans(callback_query: CallbackQuery, session: Async
             )
             return
 
+        tg_ids_result = await session.execute(
+            select(ManualBan.tg_id).where(or_(ManualBan.reason != "shadow", ManualBan.reason.is_(None)))
+        )
+        tg_ids_to_invalidate = [r[0] for r in tg_ids_result.all()]
         await session.execute(delete(ManualBan).where(or_(ManualBan.reason != "shadow", ManualBan.reason.is_(None))))
         await session.commit()
+        for uid in tg_ids_to_invalidate:
+            await invalidate_ban_cache(uid)
 
         await callback_query.message.answer(
             text=f"🗑️ Очищено {total_count} записей ручных банов из базы данных.",
@@ -358,6 +369,8 @@ async def handle_preemptive_ids_input(message: Message, state: FSMContext, sessi
 
     await session.execute(stmt)
     await session.commit()
+    for uid in tg_ids:
+        await invalidate_ban_cache(uid)
 
     await message.answer(
         f"✅ Успешно добавлено в теневой бан: <b>{len(tg_ids)}</b> пользователей.",

handlers/admin/sender/sender_handler.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from aiogram import F, Router
 from aiogram.exceptions import TelegramBadRequest
 from aiogram.fsm.context import FSMContext
@@ -5,13 +7,15 @@ from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMar
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from config import API_TOKEN
+from core.executor import run_io, should_run_heavy_tasks_separately
 from database.models import Server
 from filters.admin import IsAdminFilter
 from logger import logger
 
 from ..panel.keyboard import AdminPanelCallback, build_admin_back_kb
 from .keyboard import AdminSenderCallback, build_clusters_kb, build_sender_kb
-from .sender_service import BroadcastService
+from .sender_service import BroadcastService, run_broadcast_in_thread
 from .sender_states import AdminSender
 from .sender_utils import get_recipients, parse_message_buttons
 
@@ -192,33 +196,65 @@ async def handle_broadcast_confirm(callback_query: CallbackQuery, state: FSMCont
         _broadcast_progress_text(0, total_users_for_bar, 0, 0),
     )
 
-    messages = []
-    for tg_id in tg_ids:
-        message_data = {"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard}
-        messages.append(message_data)
-
     bot = callback_query.bot
+    state_keyboard_data = data.get("keyboard")
 
-    async def on_progress(completed: int, total: int, sent: int, failed: int) -> None:
-        text = _broadcast_progress_text(completed, total, sent, failed)
-        try:
-            await bot.edit_message_text(
-                chat_id=status_message.chat.id,
-                message_id=status_message.message_id,
-                text=text,
+    if should_run_heavy_tasks_separately():
+        main_loop = asyncio.get_running_loop()
+
+        async def _edit_progress(completed: int, total: int, sent: int, failed: int) -> None:
+            text = _broadcast_progress_text(completed, total, sent, failed)
+            try:
+                await bot.edit_message_text(
+                    chat_id=status_message.chat.id,
+                    message_id=status_message.message_id,
+                    text=text,
+                )
+            except TelegramBadRequest as e:
+                if "message is not modified" not in str(e).lower():
+                    logger.debug(f"[Sender] Обновление прогресса: {e}")
+
+        def progress_cb(completed: int, total: int, sent: int, failed: int) -> None:
+            main_loop.call_soon_threadsafe(
+                lambda c=completed, t=total, s=sent, f=failed: asyncio.ensure_future(
+                    _edit_progress(c, t, s, f), loop=main_loop
+                )
             )
-        except TelegramBadRequest as e:
-            if "message is not modified" not in str(e).lower():
-                logger.debug(f"[Sender] Обновление прогресса: {e}")
 
-    broadcast_service = BroadcastService(bot=bot, session=session, messages_per_second=35)
+        stats = await run_io(
+            run_broadcast_in_thread,
+            API_TOKEN,
+            tg_ids,
+            text_message,
+            photo,
+            state_keyboard_data,
+            progress_cb,
+        )
+    else:
+        messages = []
+        for tg_id in tg_ids:
+            message_data = {"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard}
+            messages.append(message_data)
 
-    stats = await broadcast_service.broadcast(
-        messages,
-        workers=5,
-        on_progress=on_progress,
-        progress_interval=2.0,
-    )
+        async def on_progress(completed: int, total: int, sent: int, failed: int) -> None:
+            text = _broadcast_progress_text(completed, total, sent, failed)
+            try:
+                await bot.edit_message_text(
+                    chat_id=status_message.chat.id,
+                    message_id=status_message.message_id,
+                    text=text,
+                )
+            except TelegramBadRequest as e:
+                if "message is not modified" not in str(e).lower():
+                    logger.debug(f"[Sender] Обновление прогресса: {e}")
+
+        broadcast_service = BroadcastService(bot=bot, session=session, messages_per_second=35)
+        stats = await broadcast_service.broadcast(
+            messages,
+            workers=5,
+            on_progress=on_progress,
+            progress_interval=2.0,
+        )
 
     duration_minutes = int(stats["total_duration"] // 60)
     duration_seconds = int(stats["total_duration"] % 60)

handlers/admin/sender/sender_service.py

@@ -6,13 +6,60 @@ from collections import deque
 from typing import Any
 
 from aiogram import Bot
+from aiogram.client.default import DefaultBotProperties
+from aiogram.enums import ParseMode
 from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError, TelegramRetryAfter
+from aiogram.types import InlineKeyboardMarkup
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from database import async_session_maker
 from logger import logger
 
 
+def run_broadcast_in_thread(
+    api_token: str,
+    tg_ids: list[int],
+    text_message: str,
+    photo: str | None,
+    keyboard_data: dict | None,
+    progress_cb: Callable[[int, int, int, int], None] | None = None,
+) -> dict:
+    """
+    Синхронная обёртка: запускает рассылку в отдельном event loop в текущем потоке.
+    """
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    bot = None
+    try:
+        bot = Bot(token=api_token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
+        keyboard = InlineKeyboardMarkup.model_validate(keyboard_data) if keyboard_data else None
+        messages = [
+            {"tg_id": tg_id, "text": text_message, "photo": photo, "keyboard": keyboard}
+            for tg_id in tg_ids
+        ]
+        service = BroadcastService(bot=bot, session=None, messages_per_second=35)
+
+        async def on_progress(completed: int, total: int, sent: int, failed: int) -> None:
+            if progress_cb:
+                progress_cb(completed, total, sent, failed)
+
+        return loop.run_until_complete(
+            service.broadcast(
+                messages,
+                workers=5,
+                on_progress=on_progress,
+                progress_interval=2.0,
+            )
+        )
+    finally:
+        if bot is not None and bot.session is not None:
+            try:
+                loop.run_until_complete(bot.session.close())
+            except Exception:
+                pass
+        loop.close()
+
+
 class BroadcastMessage:
     def __init__(self, tg_id: int, text: str, photo: str | None = None, keyboard: Any = None) -> None:
         self.tg_id = tg_id

handlers/admin/stats/stats_handler.py

@@ -22,10 +22,7 @@ from database import (
     count_users_registered_since,
     count_users_updated_today,
     get_tariff_distribution,
-    get_tariff_durations,
-    get_tariff_groups,
-    get_tariff_names,
-    get_tariff_subgroups,
+    get_tariff_names_groups_subgroups_durations,
     sum_payments_between,
     sum_payments_since,
     sum_total_payments,
@@ -54,21 +51,15 @@ async def handle_stats(callback_query: CallbackQuery, session: AsyncSession):
         now = datetime.now(moscow_tz)
         today = now.date()
 
-        total_users = await count_total_users(session)
         today_start = moscow_tz.localize(datetime.combine(today, datetime.min.time()))
         today_start_utc = today_start.astimezone(pytz.UTC).replace(tzinfo=None)
 
-        users_updated_today = await count_users_updated_today(session, today_start_utc)
-        registrations_today = await count_users_registered_since(session, today_start_utc)
-
         yesterday_date = today - timedelta(days=1)
         yesterday_start = moscow_tz.localize(datetime.combine(yesterday_date, datetime.min.time()))
         yesterday_end = moscow_tz.localize(datetime.combine(today, datetime.min.time()))
         yesterday_start_utc = yesterday_start.astimezone(pytz.UTC).replace(tzinfo=None)
         yesterday_end_utc = yesterday_end.astimezone(pytz.UTC).replace(tzinfo=None)
 
-        registrations_yesterday = await count_users_registered_between(session, yesterday_start_utc, yesterday_end_utc)
-
         week_start_date = today - timedelta(days=today.weekday())
         week_start = moscow_tz.localize(datetime.combine(week_start_date, datetime.min.time()))
         week_start_utc = week_start.astimezone(pytz.UTC).replace(tzinfo=None)
@@ -77,32 +68,35 @@ async def handle_stats(callback_query: CallbackQuery, session: AsyncSession):
         month_start = moscow_tz.localize(datetime.combine(month_start_date, datetime.min.time()))
         month_start_utc = month_start.astimezone(pytz.UTC).replace(tzinfo=None)
 
-        registrations_week = await count_users_registered_since(session, week_start_utc)
-        registrations_month = await count_users_registered_since(session, month_start_utc)
-
         last_month_start_date = (today.replace(day=1) - timedelta(days=1)).replace(day=1)
         this_month_start_date = today.replace(day=1)
-
         last_month_start = moscow_tz.localize(datetime.combine(last_month_start_date, datetime.min.time()))
         last_month_end = moscow_tz.localize(datetime.combine(this_month_start_date, datetime.min.time()))
         last_month_start_utc = last_month_start.astimezone(pytz.UTC).replace(tzinfo=None)
         last_month_end_utc = last_month_end.astimezone(pytz.UTC).replace(tzinfo=None)
 
+        total_users = await count_total_users(session)
+        users_updated_today = await count_users_updated_today(session, today_start_utc)
+        registrations_today = await count_users_registered_since(session, today_start_utc)
+        registrations_yesterday = await count_users_registered_between(
+            session, yesterday_start_utc, yesterday_end_utc
+        )
+        registrations_week = await count_users_registered_since(session, week_start_utc)
+        registrations_month = await count_users_registered_since(session, month_start_utc)
         registrations_last_month = await count_users_registered_between(
             session, last_month_start_utc, last_month_end_utc
         )
-
         total_keys = await count_total_keys(session)
         active_keys = await count_active_keys(session)
         active_paid_keys = await count_active_paid_keys(session)
         active_trial_keys = await count_active_trial_keys(session)
-        expired_keys = total_keys - active_keys
-
         tariff_counts, no_tariff_keys = await get_tariff_distribution(session, include_unbound=True)
-        tariff_names = await get_tariff_names(session, [tid for tid, _ in tariff_counts])
-        tariff_groups = await get_tariff_groups(session, [tid for tid, _ in tariff_counts])
-        tariff_subgroups = await get_tariff_subgroups(session, [tid for tid, _ in tariff_counts])
-        tariff_durations = await get_tariff_durations(session, [tid for tid, _ in tariff_counts])
+
+        expired_keys = total_keys - active_keys
+        tariff_ids = [tid for tid, _ in tariff_counts]
+        tariff_names, tariff_groups, tariff_subgroups, tariff_durations = (
+            await get_tariff_names_groups_subgroups_durations(session, tariff_ids)
+        )
 
         grouped_tariffs = {}
         for tid, count in tariff_counts:

handlers/admin/users/users_bans.py

@@ -11,6 +11,7 @@ from handlers.buttons import BACK
 
 from database.models import ManualBan
 from filters.admin import IsAdminFilter
+from middlewares.ban_checker import invalidate_ban_cache
 
 from .keyboard import AdminUserEditorCallback, build_editor_btn, build_editor_kb, build_user_ban_type_kb
 from .users_states import BanUserStates
@@ -81,6 +82,7 @@ async def handle_ban_forever_reason_input(message: Message, state: FSMContext, s
 
     await session.execute(stmt)
     await session.commit()
+    await invalidate_ban_cache(tg_id)
     await state.clear()
 
     await message.answer(
@@ -161,6 +163,7 @@ async def handle_ban_duration_input(message: Message, state: FSMContext, session
 
         await session.execute(stmt)
         await session.commit()
+        await invalidate_ban_cache(tg_id)
 
         text = (
             f"✅ Пользователь <code>{tg_id}</code> временно забанен до <b>{until:%Y-%m-%d %H:%M}</b> по UTC."
@@ -200,6 +203,7 @@ async def handle_ban_shadow(callback: CallbackQuery, callback_data: AdminUserEdi
     )
     await session.execute(stmt)
     await session.commit()
+    await invalidate_ban_cache(callback_data.tg_id)
 
     await callback.message.edit_text(
         text=f"👻 Пользователь <code>{callback_data.tg_id}</code> получил теневой бан.",
@@ -218,6 +222,7 @@ async def handle_user_unban(
 ):
     await session.execute(delete(ManualBan).where(ManualBan.tg_id == callback_data.tg_id))
     await session.commit()
+    await invalidate_ban_cache(callback_data.tg_id)
 
     text = (
         f"✅ Пользователь <code>{callback_data.tg_id}</code> разблокирован. Нажмите кнопку ниже для возврата в профиль."

handlers/instructions/instructions.py

@@ -42,6 +42,7 @@ from handlers.texts import (
     ROUTER_MESSAGE,
     SUBSCRIPTION_DETAILS_TEXT,
 )
+from handlers.keys.utils import key_owned_by_user
 from handlers.utils import edit_or_send_message, is_full_remnawave_cluster
 from hooks.processors import process_remnawave_webapp_override
 
@@ -75,6 +76,10 @@ async def send_instructions(callback_query_or_message: CallbackQuery | Message):
 @router.callback_query(F.data.startswith("connect_pc|"))
 async def process_connect_pc(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     key_link = await get_subscription_link(session, key_name)
     if not key_link:
         builder = InlineKeyboardBuilder()
@@ -103,6 +108,10 @@ async def process_connect_pc(callback_query: CallbackQuery, session: Any):
 @router.callback_query(F.data.startswith("windows_menu|"))
 async def process_windows_menu(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     key_link = await get_subscription_link(session, key_name)
     if not key_link:
         await callback_query.message.answer("❌ Ошибка: ключ не найден.")
@@ -136,6 +145,10 @@ async def process_windows_menu(callback_query: CallbackQuery, session: Any):
 @router.callback_query(F.data.startswith("macos_menu|"))
 async def process_macos_menu(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     key_link = await get_subscription_link(session, key_name)
     if not key_link:
         await callback_query.message.answer("❌ Ошибка: ключ не найден.")
@@ -171,6 +184,9 @@ async def process_connect_tv(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|", 1)[1]
 
     record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     final_link = None
     is_full_remnawave = False
     use_webapp = False
@@ -221,6 +237,10 @@ async def process_connect_tv(callback_query: CallbackQuery, session: Any):
 @router.callback_query(F.data.startswith("continue_tv|"))
 async def process_continue_tv(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     key_link = await get_subscription_link(session, key_name)
     message_text = SUBSCRIPTION_DETAILS_TEXT.format(subscription_link=key_link)
 
@@ -239,6 +259,10 @@ async def process_continue_tv(callback_query: CallbackQuery, session: Any):
 @router.callback_query(F.data.startswith("connect_router|"))
 async def process_connect_router(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     key_link = await get_subscription_link(session, key_name)
     if not key_link:
         builder = InlineKeyboardBuilder()

handlers/keys/key_connect.py

@@ -19,7 +19,7 @@ from config import (
     DOWNLOAD_IOS,
     INSTRUCTIONS_BUTTON,
 )
-from database import Key, get_subscription_link
+from database import Key, get_key_details, get_subscription_link
 from handlers.buttons import (
     ANDROID,
     BACK,
@@ -39,6 +39,7 @@ from handlers.texts import (
     IOS_DESCRIPTION_TEMPLATE,
     SUBSCRIPTION_DESCRIPTION,
 )
+from handlers.keys.utils import key_owned_by_user
 from handlers.utils import edit_or_send_message
 from hooks.hook_buttons import insert_hook_buttons
 from hooks.processors import process_connect_device_menu
@@ -67,6 +68,10 @@ def generate_key_qr_file(qr_data: str, email: str) -> str:
 async def handle_connect_device(callback_query: CallbackQuery, session: AsyncSession):
     try:
         key_name = callback_query.data.split("|")[1]
+        record = await get_key_details(session, key_name)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         builder = InlineKeyboardBuilder()
         builder.row(InlineKeyboardButton(text=IPHONE, callback_data=f"connect_ios|{key_name}"))
@@ -102,6 +107,10 @@ async def process_callback_connect_phone(callback_query: CallbackQuery, session:
     email = callback_query.data.split("|")[1]
 
     try:
+        record = await get_key_details(session, email)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         key_link = await get_subscription_link(session, email)
         if not key_link:
             await callback_query.message.answer("❌ Ошибка: ключ не найден.")
@@ -148,6 +157,10 @@ async def process_callback_connect_ios(callback_query: CallbackQuery, session: A
     email = callback_query.data.split("|")[1]
 
     try:
+        record = await get_key_details(session, email)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         key_link = await get_subscription_link(session, email)
         if not key_link:
             await callback_query.message.answer("❌ Ошибка: ключ не найден.")
@@ -188,6 +201,10 @@ async def process_callback_connect_android(callback_query: CallbackQuery, sessio
     email = callback_query.data.split("|")[1]
 
     try:
+        record = await get_key_details(session, email)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         key_link = await get_subscription_link(session, email)
         if not key_link:
             await callback_query.message.answer("❌ Ошибка: ключ не найден.")
@@ -235,6 +252,9 @@ async def show_qr_code(callback_query: types.CallbackQuery, session: AsyncSessio
         if not record:
             await callback_query.message.answer("❌ Подписка не найдена.")
             return
+        if record.tg_id != callback_query.from_user.id:
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         qr_data = record.key or record.remnawave_link
         if not qr_data:

handlers/keys/key_freeze.py

@@ -21,6 +21,7 @@ from handlers.texts import (
     SUBSCRIPTION_UNFROZEN_MSG,
     UNFREEZE_SUBSCRIPTION_CONFIRM_MSG,
 )
+from handlers.keys.utils import key_owned_by_user
 from handlers.utils import edit_or_send_message, handle_error
 from middlewares.session import release_session_early
 from logger import logger
@@ -32,6 +33,10 @@ router = Router()
 @router.callback_query(F.data.startswith("unfreeze_subscription|"))
 async def process_callback_unfreeze_subscription(callback_query: CallbackQuery, session: Any):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     confirm_text = UNFREEZE_SUBSCRIPTION_CONFIRM_MSG
 
     builder = InlineKeyboardBuilder()
@@ -66,6 +71,9 @@ async def process_callback_unfreeze_subscription_confirm(callback_query: Callbac
         if not record:
             await callback_query.message.answer("Ключ не найден.")
             return
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         email = record["email"]
         client_id = record["client_id"]
@@ -137,6 +145,10 @@ async def process_callback_freeze_subscription(callback_query: CallbackQuery, se
     Показывает пользователю диалог подтверждения заморозки (отключения) подписки.
     """
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
 
     confirm_text = FREEZE_SUBSCRIPTION_CONFIRM_MSG
 
@@ -172,6 +184,9 @@ async def process_callback_freeze_subscription_confirm(callback_query: CallbackQ
         if not record:
             await callback_query.message.answer("Ключ не найден.")
             return
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         email = record["email"]
         client_id = record["client_id"]

handlers/keys/key_mode/key_country_mode.py

@@ -264,6 +264,9 @@ async def change_location_callback(callback_query: CallbackQuery, session: Any):
         if not record:
             await callback_query.answer("❌ Ключ не найден", show_alert=True)
             return
+        if record.get("tg_id") != callback_query.from_user.id:
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         expiry_timestamp = record["expiry_time"]
         ts = int(expiry_timestamp / 1000)

handlers/keys/key_renew.py

@@ -56,7 +56,7 @@ from hooks.processors import (
 )
 from logger import logger
 
-from .utils import add_tariff_button_generic
+from .utils import add_tariff_button_generic, key_owned_by_user
 
 
 router = Router()
@@ -86,6 +86,9 @@ async def process_callback_renew_key(callback_query: CallbackQuery, state: FSMCo
         if not record:
             await callback_query.message.answer("<b>Ключ не найден.</b>")
             return
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
 
         client_id = record["client_id"]
         expiry_time_raw = record["expiry_time"]

handlers/keys/key_view.py

@@ -62,6 +62,7 @@ from handlers.texts import (
     RENAME_KEY_PROMPT,
     key_message,
 )
+from handlers.keys.utils import key_owned_by_user
 from handlers.utils import (
     edit_or_send_message,
     format_days,
@@ -227,8 +228,12 @@ async def build_keys_response(records: list[Key] | None, session: AsyncSession,
 
 
 @router.callback_query(F.data.startswith("rename_key|"))
-async def handle_rename_key(callback: CallbackQuery, state: FSMContext):
+async def handle_rename_key(callback: CallbackQuery, state: FSMContext, session: AsyncSession):
     client_id = callback.data.split("|")[1]
+    key_row = (await session.execute(select(Key).where(Key.client_id == client_id))).scalar_one_or_none()
+    if not key_row or key_row.tg_id != callback.from_user.id:
+        await callback.answer("Доступ запрещён.", show_alert=True)
+        return
     await state.set_state(RenameKeyState.waiting_for_new_alias)
     await state.update_data(client_id=client_id)
 
@@ -282,6 +287,10 @@ async def handle_new_alias_input(message: Message, state: FSMContext, session: A
 @router.callback_query(F.data.startswith("view_key|"))
 async def process_callback_view_key(callback_query: CallbackQuery, session: AsyncSession):
     key_name = callback_query.data.split("|")[1]
+    record = await get_key_details(session, key_name)
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
     image_path = os.path.join("img", "pic_view.jpg")
     await render_key_info(callback_query.message, session, key_name, image_path)
 
@@ -482,6 +491,9 @@ async def handle_reset_hwid(callback_query: CallbackQuery, session: AsyncSession
     if not record:
         await callback_query.answer("❌ Ключ не найден.", show_alert=True)
         return
+    if not key_owned_by_user(record, callback_query.from_user.id):
+        await callback_query.answer("Доступ запрещён.", show_alert=True)
+        return
 
     client_id = record.get("client_id")
     if not client_id:

handlers/keys/keys.py

@@ -7,6 +7,7 @@ from database import delete_key, get_key_details
 from handlers.buttons import APPLY, BACK, CANCEL
 from handlers.keys.key_view import process_callback_view_key
 from handlers.keys.operations import delete_key_from_cluster, update_subscription
+from handlers.keys.utils import key_owned_by_user
 from handlers.texts import DELETE_KEY_CONFIRM_MSG, KEY_DELETED_MSG_SIMPLE
 from handlers.utils import edit_or_send_message, handle_error
 from middlewares.session import release_session_early
@@ -22,6 +23,10 @@ async def process_callback_update_subscription(callback_query: CallbackQuery, se
     email = callback_query.data.split("|")[1]
 
     try:
+        record = await get_key_details(session, email)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         try:
             await callback_query.message.delete()
         except TelegramBadRequest as e:
@@ -36,12 +41,16 @@ async def process_callback_update_subscription(callback_query: CallbackQuery, se
 
 
 @router.callback_query(F.data.startswith("delete_key|"))
-async def process_callback_delete_key(callback_query: CallbackQuery):
-    client_id = callback_query.data.split("|")[1]
+async def process_callback_delete_key(callback_query: CallbackQuery, session: AsyncSession):
+    key_identifier = callback_query.data.split("|")[1]
     try:
+        record = await get_key_details(session, key_identifier)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         confirmation_keyboard = types.InlineKeyboardMarkup(
             inline_keyboard=[
-                [types.InlineKeyboardButton(text=APPLY, callback_data=f"confirm_delete|{client_id}")],
+                [types.InlineKeyboardButton(text=APPLY, callback_data=f"confirm_delete|{key_identifier}")],
                 [types.InlineKeyboardButton(text=CANCEL, callback_data="view_keys")],
             ]
         )
@@ -54,7 +63,7 @@ async def process_callback_delete_key(callback_query: CallbackQuery):
             await callback_query.message.edit_text(text=DELETE_KEY_CONFIRM_MSG, reply_markup=confirmation_keyboard)
 
     except Exception as e:
-        logger.error(f"Ошибка при подготовке удаления ключа {client_id}: {e}")
+        logger.error(f"Ошибка при подготовке удаления ключа {key_identifier}: {e}")
 
 
 @router.callback_query(F.data.startswith("confirm_delete|"))
@@ -62,6 +71,9 @@ async def process_callback_confirm_delete(callback_query: CallbackQuery, session
     email = callback_query.data.split("|")[1]
     try:
         record = await get_key_details(session, email)
+        if not key_owned_by_user(record, callback_query.from_user.id):
+            await callback_query.answer("Доступ запрещён.", show_alert=True)
+            return
         if record:
             client_id = record["client_id"]
             server_id = record["server_id"]

handlers/keys/operations/traffic.py

@@ -2,7 +2,7 @@ import asyncio
 
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import SUPERNODE
@@ -21,37 +21,37 @@ async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dic
     """
     Получает трафик пользователя на всех серверах, где у него есть ключ (3x-ui и Remnawave).
     Для Remnawave трафик считается один раз и отображается как "Remnawave (общий):".
+    Один запрос: Key + Server через join.
     """
-    result = await session.execute(select(Key.client_id, Key.server_id).where(Key.tg_id == tg_id, Key.email == email))
+    join_cond = or_(
+        Key.server_id == Server.server_name,
+        Key.server_id == Server.cluster_name,
+    )
+    result = await session.execute(
+        select(Key.client_id, Key.server_id, Server)
+        .select_from(Key)
+        .join(Server, join_cond)
+        .where(Server.enabled.is_(True), Key.tg_id == tg_id, Key.email == email)
+    )
     rows = result.all()
     if not rows:
         return {"status": "error", "message": "У пользователя нет активных ключей."}
 
-    server_ids = {row.server_id for row in rows}
-    server_id = list(server_ids)[0]
 
-    result = await session.execute(
-        select(Server)
-        .where(Server.enabled.is_(True))
-        .where(Server.server_name.in_(server_ids) | Server.cluster_name.in_(server_ids))
-    )
-    server_rows = result.scalars().all()
-    if not server_rows:
-        logger.error(f"Не найдено серверов для: {server_ids}")
-        return {
-            "status": "error",
-            "message": f"Серверы не найдены: {', '.join(server_ids)}",
-        }
-
-    servers_map = {
-        s.server_name: {
-            "server_name": s.server_name,
-            "cluster_name": s.cluster_name,
-            "api_url": s.api_url,
-            "panel_type": s.panel_type,
-        }
-        for s in server_rows
-    }
+    seen_pairs = set()
+    unique_rows = []
+    servers_map = {}
+    for client_id, server_id, server in rows:
+        if (client_id, server_id) not in seen_pairs:
+            seen_pairs.add((client_id, server_id))
+            unique_rows.append((client_id, server_id))
+        if server.server_name not in servers_map:
+            servers_map[server.server_name] = {
+                "server_name": server.server_name,
+                "cluster_name": server.cluster_name,
+                "api_url": server.api_url,
+                "panel_type": server.panel_type,
+            }
 
     user_traffic_data = {}
     tasks = []
@@ -80,10 +80,7 @@ async def get_user_traffic(session: AsyncSession, tg_id: int, email: str) -> dic
         except Exception as e:
             return server_name, f"Ошибка: {e}"
 
-    for row in rows:
-        client_id = row.client_id
-        server_id = row.server_id
-
+    for client_id, server_id in unique_rows:
         matched_servers = [
             s for s in servers_map.values() if s["server_name"] == server_id or s["cluster_name"] == server_id
         ]

handlers/keys/utils.py

@@ -7,6 +7,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from handlers.payments.currency_rates import format_for_user
 
 
+def key_owned_by_user(record: dict | None, user_id: int) -> bool:
+    """Проверка, что ключ принадлежит пользователю (защита от пересылки callback)."""
+    return record is not None and record.get("tg_id") == user_id
+
+
 async def add_tariff_button_generic(
     builder: InlineKeyboardBuilder,
     tariff: dict[str, Any],

handlers/notifications/general_notifications.py

@@ -34,6 +34,7 @@ from database import (
     get_all_keys,
     get_balance,
     get_last_notification_time,
+    get_last_notification_times_bulk,
     update_balance,
     update_key_expiry,
     update_key_tariff,
@@ -583,6 +584,9 @@ async def handle_expired_keys(ctx: NotificationContext, keys: list):
     notify_delete_key_enabled = bool(NOTIFICATIONS_CONFIG.get("DELETE_KEY_ENABLED", NOTIFY_DELETE_KEY))
     delete_key_delay_minutes = int(NOTIFICATIONS_CONFIG.get("DELETE_KEY_DELAY_MINUTES", NOTIFY_DELETE_DELAY))
 
+    notification_pairs = [(key.tg_id, f"{key.email or ''}_key_expired") for key in expired_keys]
+    last_times = await get_last_notification_times_bulk(ctx.session, notification_pairs)
+
     for key in expired_keys:
         tg_id = key.tg_id
         email = key.email or ""
@@ -590,7 +594,7 @@ async def handle_expired_keys(ctx: NotificationContext, keys: list):
         server_id = key.server_id
         notification_id = f"{email}_key_expired"
 
-        last_notification_time = await get_last_notification_time(ctx.session, tg_id, notification_id)
+        last_notification_time = last_times.get((tg_id, notification_id))
 
         if notify_renew_expired_enabled:
             try:
@@ -700,7 +704,7 @@ async def periodic_notifications(bot: Bot, *, sessionmaker: async_sessionmaker):
                     trial_time_disable = bool(MODES_CONFIG.get("TRIAL_TIME_DISABLED", TRIAL_TIME_DISABLE))
                     if not trial_time_disable:
                         try:
-                            await notify_inactive_trial_users(bot, session)
+                            await notify_inactive_trial_users(bot, session, sessionmaker=sessionmaker)
                         except Exception as error:
                             logger.error(f"Ошибка в notify_inactive_trial_users: {error}")
 

handlers/notifications/special_notifications.py

@@ -6,7 +6,7 @@ from aiogram import Bot, Router, types
 from aiogram.types import InlineKeyboardButton, WebAppInfo
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 from sqlalchemy import update
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from config import (
     NOTIFY_EXTRA_DAYS,
@@ -38,7 +38,9 @@ router = Router()
 moscow_tz = pytz.timezone("Europe/Moscow")
 
 
-async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
+async def notify_inactive_trial_users(
+    bot: Bot, session: AsyncSession, *, sessionmaker: async_sessionmaker | None = None
+):
     logger.info("Проверка пользователей, не активировавших пробный период...")
 
     inactive_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_USER_ENABLED", NOTIFY_INACTIVE))
@@ -119,8 +121,13 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
                 sent_tg_ids.append(msg["tg_id"])
 
         if sent_tg_ids:
-            for tg_id in sent_tg_ids:
-                await add_notification(session, tg_id, "inactive_trial")
+            if sessionmaker is not None:
+                async with sessionmaker() as fresh_session:
+                    for tg_id in sent_tg_ids:
+                        await add_notification(fresh_session, tg_id, "inactive_trial")
+            else:
+                for tg_id in sent_tg_ids:
+                    await add_notification(session, tg_id, "inactive_trial")
             logger.info(f"Отправлено {len(sent_tg_ids)} уведомлений неактивным пользователям.")
 
     logger.info("Проверка пользователей с неактивным пробным периодом завершена.")

handlers/payments/freekassa/freekassa_pay.py

@@ -16,6 +16,11 @@ from config import (
     FREEKASSA_SECRET2,
     FREEKASSA_SHOP_ID,
 )
+from core.webhook_abuse import (
+    get_webhook_client_ip,
+    is_webhook_ip_blocked,
+    record_webhook_signature_failure,
+)
 from database import (
     add_payment,
     add_user,
@@ -25,6 +30,7 @@ from database import (
     get_key_count,
     get_payment_by_payment_id,
     get_temporary_data,
+    invalidate_payment_cache,
     update_balance,
 )
 from handlers.buttons import BACK, PAY_2
@@ -198,6 +204,9 @@ def verify_signature(params: dict) -> bool:
 
 async def freekassa_webhook(request: web.Request):
     try:
+        ip = get_webhook_client_ip(request)
+        if await is_webhook_ip_blocked(ip):
+            return web.Response(status=429)
         params = dict(request.query)
         logger.info(f"Received Freekassa webhook: {params}")
 
@@ -213,6 +222,7 @@ async def freekassa_webhook(request: web.Request):
 
         if not verify_signature(params):
             logger.error("Invalid signature in webhook")
+            await record_webhook_signature_failure(ip)
             return web.Response(status=400, text="Invalid signature")
 
         if str(merchant_id) != str(FREEKASSA_SHOP_ID):
@@ -244,6 +254,7 @@ async def freekassa_webhook(request: web.Request):
             await send_payment_success_notification(tg_id_int, amount_float, session)
             await add_payment(session, tg_id_int, amount_float, "freekassa", payment_id=merchant_order_id)
             await clear_temporary_data(session, tg_id_int)
+            await invalidate_payment_cache(merchant_order_id)
 
         logger.info(f"Payment processed successfully. User: {tg_id_int}, Amount: {amount_float}")
         return web.Response(text="YES")

handlers/payments/heleket/service.py

@@ -21,7 +21,7 @@ from config import (
     HELEKET_SUCCESS_URL,
     PROVIDERS_ENABLED,
 )
-from database import add_payment, async_session_maker
+from database import async_session_maker, register_pending_payment
 from database.models import User
 from handlers.buttons import BACK, HELEKET, PAY_2
 from handlers.payments.currency_rates import (
@@ -240,7 +240,8 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext,
     if currency == "RUB":
         amount_rub = user_amount
     else:
-        async with aiohttp.ClientSession() as session_http:
+        timeout = aiohttp.ClientTimeout(total=30, connect=10)
+        async with aiohttp.ClientSession(timeout=timeout) as session_http:
             amount_rub = int(await to_rub(user_amount, "USD", session=session_http))
 
     await state.update_data(amount=amount_rub)
@@ -336,8 +337,9 @@ async def generate_heleket_payment_link(
     unique_order_id = f"{int(time.time())}_{tg_id}"
     db_session = session
 
+    timeout = aiohttp.ClientTimeout(total=30, connect=10)
     try:
-        async with aiohttp.ClientSession() as http_session:
+        async with aiohttp.ClientSession(timeout=timeout) as http_session:
             pay_cur = str(method["currency"]).upper()
 
             if pay_cur == "RUB":
@@ -376,27 +378,13 @@ async def generate_heleket_payment_link(
                         if resp_json.get("state") == 0:
                             payment_url = resp_json.get("result", {}).get("url")
                             if payment_url:
-                                if db_session is not None:
-                                    await add_payment(
-                                        session=db_session,
-                                        tg_id=tg_id,
-                                        amount=float(amount),
-                                        payment_system="HELEKET",
-                                        status="pending",
-                                        currency="RUB",
-                                        payment_id=unique_order_id,
-                                    )
-                                else:
-                                    async with async_session_maker() as dbs:
-                                        await add_payment(
-                                            session=dbs,
-                                            tg_id=tg_id,
-                                            amount=float(amount),
-                                            payment_system="HELEKET",
-                                            status="pending",
-                                            currency="RUB",
-                                            payment_id=unique_order_id,
-                                        )
+                                await register_pending_payment(
+                                    payment_id=unique_order_id,
+                                    tg_id=tg_id,
+                                    amount=float(amount),
+                                    payment_system="heleket",
+                                    currency="RUB",
+                                )
                                 logger.info(f"Heleket payment URL created for user {tg_id}")
                                 return payment_url
                             else:

handlers/payments/heleket/webhook.py

@@ -5,10 +5,16 @@ import json
 from aiohttp import web
 
 from config import HELEKET_API_KEY
+from core.webhook_abuse import (
+    get_webhook_client_ip,
+    is_webhook_ip_blocked,
+    record_webhook_signature_failure,
+)
 from database import (
     add_payment,
     async_session_maker,
     get_payment_by_payment_id,
+    invalidate_payment_cache,
     update_balance,
     update_payment_status,
 )
@@ -106,18 +112,30 @@ async def process_heleket_webhook(data: dict) -> bool:
                     if payment.get("status") == "success":
                         logger.info(f"Heleket: платёж {order_id} уже обработан")
                         return True
-                    ok = await update_payment_status(
-                        session=session, internal_id=int(payment["id"]), new_status="success"
-                    )
-                    if not ok:
-                        logger.error(f"Heleket: не удалось обновить статус платежа {order_id}")
-                        return False
+                    if payment.get("id") is not None:
+                        ok = await update_payment_status(
+                            session=session, internal_id=int(payment["id"]), new_status="success"
+                        )
+                        if not ok:
+                            logger.error(f"Heleket: не удалось обновить статус платежа {order_id}")
+                            return False
+                    else:
+                        await add_payment(
+                            session=session,
+                            tg_id=tg_id,
+                            amount=balance_amount,
+                            payment_system="HELEKET",
+                            status="success",
+                            currency="USD",
+                            payment_id=order_id,
+                            metadata=None,
+                        )
                 else:
                     await add_payment(
                         session=session,
                         tg_id=tg_id,
                         amount=balance_amount,
-                        payment_system="HELEKET",
+                        payment_system="heleket",
                         status="success",
                         currency="USD",
                         payment_id=order_id,
@@ -126,6 +144,7 @@ async def process_heleket_webhook(data: dict) -> bool:
 
                 await update_balance(session, tg_id, balance_amount)
                 await send_payment_success_notification(tg_id, balance_amount, session)
+                await invalidate_payment_cache(order_id)
             logger.info(
                 f"Heleket: платёж {order_id} для пользователя {tg_id} "
                 f"успешно обработан, баланс пополнен на {balance_amount} RUB"
@@ -136,13 +155,14 @@ async def process_heleket_webhook(data: dict) -> bool:
 
             async with async_session_maker() as session:
                 payment = await get_payment_by_payment_id(session, order_id)
-                if payment:
+                if payment and payment.get("id") is not None:
                     await update_payment_status(
                         session=session,
                         internal_id=int(payment["id"]),
                         new_status="failed",
                     )
                     await session.commit()
+                await invalidate_payment_cache(order_id)
             return True
         else:
             logger.info(f"Heleket: промежуточный статус {status} для платежа {order_id}")
@@ -155,11 +175,15 @@ async def process_heleket_webhook(data: dict) -> bool:
 async def heleket_webhook(request: web.Request):
     """Обработчик webhook от Heleket для aiohttp."""
     try:
+        ip = get_webhook_client_ip(request)
+        if await is_webhook_ip_blocked(ip):
+            return web.Response(status=429)
         data = await request.json()
         logger.info(f"Heleket webhook received from {request.remote}")
 
         if not verify_heleket_signature(data):
             logger.error("Heleket webhook: неверная подпись")
+            await record_webhook_signature_failure(ip)
             return web.Response(status=400, text="Invalid signature")
 
         success = await process_heleket_webhook(data)

handlers/payments/kassai/service.py

@@ -20,7 +20,7 @@ from config import (
     KASSAI_SUCCESS_URL,
     PROVIDERS_ENABLED,
 )
-from database import add_payment, async_session_maker
+from database import async_session_maker, register_pending_payment
 from database.models import User
 from handlers.buttons import BACK, KASSAI_CARDS, KASSAI_SBP, PAY_2
 from handlers.payments.currency_rates import (
@@ -263,7 +263,8 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext,
     if currency == "RUB":
         amount_rub = user_amount
     else:
-        async with aiohttp.ClientSession() as session_http:
+        timeout = aiohttp.ClientTimeout(total=30, connect=10)
+        async with aiohttp.ClientSession(timeout=timeout) as session_http:
             amount_rub = int(await to_rub(user_amount, "USD", session=session_http))
 
     await state.update_data(amount=amount_rub)
@@ -390,8 +391,9 @@ async def generate_kassai_payment_link(
 
     db_session = session
 
+    timeout = aiohttp.ClientTimeout(total=60, connect=10)
     try:
-        async with aiohttp.ClientSession() as http_session:
+        async with aiohttp.ClientSession(timeout=timeout) as http_session:
             async with http_session.post(url, headers=headers, json=data, timeout=60) as resp:
                 if resp.status == 200:
                     try:
@@ -399,27 +401,13 @@ async def generate_kassai_payment_link(
                         if resp_json.get("type") == "success":
                             payment_url = resp_json.get("location")
                             if payment_url:
-                                if db_session is not None:
-                                    await add_payment(
-                                        session=db_session,
-                                        tg_id=tg_id,
-                                        amount=float(amount),
-                                        payment_system="KASSAI",
-                                        status="pending",
-                                        currency="RUB",
-                                        payment_id=unique_payment_id,
-                                    )
-                                else:
-                                    async with async_session_maker() as dbs:
-                                        await add_payment(
-                                            session=dbs,
-                                            tg_id=tg_id,
-                                            amount=float(amount),
-                                            payment_system="KASSAI",
-                                            status="pending",
-                                            currency="RUB",
-                                            payment_id=unique_payment_id,
-                                        )
+                                await register_pending_payment(
+                                    payment_id=unique_payment_id,
+                                    tg_id=tg_id,
+                                    amount=float(amount),
+                                    payment_system="kassai",
+                                    currency="RUB",
+                                )
                                 logger.info(f"KassaAI payment URL created for user {tg_id}")
                                 return payment_url
                             logger.error(f"KassaAI: No location in response: {resp_json}")

handlers/payments/kassai/webhook.py

@@ -3,10 +3,16 @@ import hashlib
 from aiohttp import web
 
 from config import KASSAI_SECRET_KEY, KASSAI_SHOP_ID, KASSAI_WEBHOOK_RESPONSE
+from core.webhook_abuse import (
+    get_webhook_client_ip,
+    is_webhook_ip_blocked,
+    record_webhook_signature_failure,
+)
 from database import (
     add_payment,
     async_session_maker,
     get_payment_by_payment_id,
+    invalidate_payment_cache,
     update_balance,
     update_payment_status,
 )
@@ -44,14 +50,19 @@ def verify_kassai_signature(data: dict, signature: str) -> bool:
 async def kassai_webhook(request: web.Request):
     """Обработчик webhook от KassaAI для aiohttp."""
     try:
+        ip = get_webhook_client_ip(request)
+        if await is_webhook_ip_blocked(ip):
+            return web.Response(status=429)
         data = await request.post()
         logger.info(f"KassaAI webhook received: {dict(data)}")
         signature = data.get("SIGN", "")
         if not signature:
             logger.error("KassaAI webhook: отсутствует подпись")
+            await record_webhook_signature_failure(ip)
             return web.Response(status=400)
         if not verify_kassai_signature(data, signature):
             logger.error("KassaAI webhook: неверная подпись")
+            await record_webhook_signature_failure(ip)
             return web.Response(status=400)
 
         amount_raw = data.get("AMOUNT")
@@ -77,10 +88,24 @@ async def kassai_webhook(request: web.Request):
                 if payment.get("status") == "success":
                     logger.info(f"KassaAI: платёж {order_id} уже обработан")
                     return web.Response(text=KASSAI_WEBHOOK_RESPONSE)
-                ok = await update_payment_status(session=session, internal_id=int(payment["id"]), new_status="success")
-                if not ok:
-                    logger.error(f"KassaAI: не удалось обновить статус платежа {order_id}")
-                    return web.Response(status=500)
+                if payment.get("id") is not None:
+                    ok = await update_payment_status(
+                        session=session, internal_id=int(payment["id"]), new_status="success"
+                    )
+                    if not ok:
+                        logger.error(f"KassaAI: не удалось обновить статус платежа {order_id}")
+                        return web.Response(status=500)
+                else:
+                    await add_payment(
+                        session=session,
+                        tg_id=tg_id,
+                        amount=amount,
+                        payment_system="kassai",
+                        status="success",
+                        currency="RUB",
+                        payment_id=order_id,
+                        metadata=None,
+                    )
             else:
                 await add_payment(
                     session=session,
@@ -95,6 +120,7 @@ async def kassai_webhook(request: web.Request):
 
             await update_balance(session, tg_id, amount)
             await send_payment_success_notification(tg_id, amount, session)
+            await invalidate_payment_cache(order_id)
         logger.info(
             f"KassaAI: платёж {order_id} успешно обработан, баланс пользователя {tg_id} пополнен на {amount} RUB"
         )

handlers/payments/robokassa/service.py

@@ -8,7 +8,7 @@ from urllib.parse import quote_plus, urlencode
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import ROBOKASSA_LOGIN, ROBOKASSA_PASSWORD1, ROBOKASSA_PASSWORD2, ROBOKASSA_TEST_MODE
-from database import add_payment
+from database import register_pending_payment
 from handlers.payments.payment_links import register_payment_creator
 
 
@@ -61,15 +61,12 @@ async def create_and_store_robokassa_payment(
     session: AsyncSession, tg_id: int, amount: int | float, description: str, inv_id: int = 0
 ) -> tuple[str, str]:
     url, pid = generate_payment_link(amount, inv_id, description, tg_id)
-    await add_payment(
-        session=session,
+    await register_pending_payment(
+        payment_id=pid,
         tg_id=tg_id,
         amount=float(amount),
         payment_system="robokassa",
-        status="pending",
         currency="RUB",
-        payment_id=pid,
-        metadata=None,
     )
     return url, pid
 

handlers/payments/robokassa/webhook.py

@@ -1,6 +1,18 @@
 from aiohttp import web
 
-from database import add_payment, async_session_maker, get_payment_by_payment_id, update_balance, update_payment_status
+from core.webhook_abuse import (
+    get_webhook_client_ip,
+    is_webhook_ip_blocked,
+    record_webhook_signature_failure,
+)
+from database import (
+    add_payment,
+    async_session_maker,
+    get_payment_by_payment_id,
+    invalidate_payment_cache,
+    update_balance,
+    update_payment_status,
+)
 from handlers.payments.utils import send_payment_success_notification
 from logger import logger
 
@@ -9,8 +21,12 @@ from .service import check_payment_signature
 
 async def robokassa_webhook(request: web.Request):
     try:
+        ip = get_webhook_client_ip(request)
+        if await is_webhook_ip_blocked(ip):
+            return web.Response(status=429)
         params = await request.post()
         if not check_payment_signature(params):
+            await record_webhook_signature_failure(ip)
             return web.Response(status=400)
 
         amount_raw = params.get("OutSum")
@@ -29,15 +45,29 @@ async def robokassa_webhook(request: web.Request):
             if payment:
                 if payment.get("status") == "success":
                     return web.Response(text=f"OK{inv_id}")
-                ok = await update_payment_status(session=session, internal_id=int(payment["id"]), new_status="success")
-                if not ok:
-                    return web.Response(status=500)
+                if payment.get("id") is not None:
+                    ok = await update_payment_status(
+                        session=session, internal_id=int(payment["id"]), new_status="success"
+                    )
+                    if not ok:
+                        return web.Response(status=500)
+                else:
+                    await add_payment(
+                        session=session,
+                        tg_id=tg_id,
+                        amount=amount,
+                        payment_system="robokassa",
+                        status="success",
+                        currency="RUB",
+                        payment_id=shp_pid,
+                        metadata=None,
+                    )
             else:
                 await add_payment(
                     session=session,
                     tg_id=tg_id,
                     amount=amount,
-                    payment_system="ROBOKASSA",
+                    payment_system="robokassa",
                     status="success",
                     currency="RUB",
                     payment_id=shp_pid,
@@ -46,6 +76,7 @@ async def robokassa_webhook(request: web.Request):
 
             await update_balance(session, tg_id, amount)
             await send_payment_success_notification(tg_id, amount, session)
+            await invalidate_payment_cache(shp_pid)
 
         return web.Response(text=f"OK{inv_id}")
     except Exception as e:

handlers/profile.py

@@ -15,7 +15,9 @@ from config import (
     TRIAL_TIME_DISABLE,
 )
 from core.bootstrap import BUTTONS_CONFIG, MODES_CONFIG
-from database import get_balance, get_key_count, get_trial
+from core.cache_config import BALANCE_CACHE_TTL_SEC, KEY_COUNT_CACHE_TTL_SEC, PROFILE_DATA_CACHE_TTL_SEC
+from core.redis_cache import cache_get, cache_key, cache_set
+from database import get_balance_trial_key_count
 from handlers.buttons import (
     ABOUT_VPN,
     ADD_SUB,
@@ -33,6 +35,7 @@ from handlers.payments.currency_rates import format_for_user
 from handlers.texts import ADD_SUBSCRIPTION_HINT
 from hooks.hook_buttons import insert_hook_buttons
 from hooks.hooks import run_hooks
+from middlewares.session import release_session_early
 
 from .admin.panel.keyboard import AdminPanelCallback
 from .texts import profile_message_send
@@ -62,10 +65,21 @@ async def process_callback_view_profile(
     chat_id = chat.id
     username = get_username(user or chat)
 
-    key_count = await get_key_count(session, chat_id)
-    balance_rub = await get_balance(session, chat_id)
-    trial_status = await get_trial(session, chat_id)
-    balance_rub = balance_rub or 0
+    cached = await cache_get(cache_key("profile_data", chat_id))
+    if isinstance(cached, dict) and "key_count" in cached and "balance_rub" in cached and "trial_status" in cached:
+        key_count = int(cached["key_count"])
+        balance_rub = float(cached.get("balance_rub") or 0)
+        trial_status = int(cached.get("trial_status") or 0)
+    else:
+        balance_rub, trial_status, key_count = await get_balance_trial_key_count(session, chat_id)
+        balance_rub = balance_rub or 0
+        await cache_set(cache_key("balance", chat_id), balance_rub, BALANCE_CACHE_TTL_SEC)
+        await cache_set(cache_key("key_count", chat_id), key_count, KEY_COUNT_CACHE_TTL_SEC)
+        await cache_set(
+            cache_key("profile_data", chat_id),
+            {"key_count": key_count, "balance_rub": balance_rub, "trial_status": trial_status},
+            PROFILE_DATA_CACHE_TTL_SEC,
+        )
 
     balance_text = await format_for_user(
         session,
@@ -134,6 +148,7 @@ async def process_callback_view_profile(
     else:
         builder.row(InlineKeyboardButton(text=BACK, callback_data="start"))
 
+    await release_session_early(session)
     await edit_or_send_message(
         target_message=message,
         text=profile_message,

handlers/start.py

@@ -155,9 +155,20 @@ async def process_start_logic(
 
     await state.update_data(original_text=text, user_data=user_data)
 
+    _MAX_START_PAYLOAD_LEN = 256
+    _MAX_START_PARTS = 20
+    if text and len(text) > _MAX_START_PAYLOAD_LEN:
+        text = text[:_MAX_START_PAYLOAD_LEN]
+    parts = text.split("-") if text else []
+    if len(parts) > _MAX_START_PARTS:
+        parts = parts[:_MAX_START_PARTS]
+
     gift_detected = False
-    if text:
-        for part in text.split("-"):
+    if parts:
+        for part in parts:
+            part = part.strip()
+            if not part:
+                continue
             await run_hooks("start_link", message=message, state=state, session=session, user_data=user_data, part=part)
             if "coupons" in part:
                 await handle_coupon_link(part, message, state, session, admin, user_data)
@@ -171,6 +182,8 @@ async def process_start_logic(
             if "utm" in part:
                 await handle_utm_link(part, message, state, session, user_data)
 
+    text = "-".join(parts) if parts else (text or "")
+
     await state.clear()
     if gift_detected:
         return

handlers/tariffs/addons/key_addons_main.py

@@ -318,6 +318,9 @@ async def start_key_addons(callback: CallbackQuery, state: FSMContext, session:
         logger.warning(f"[ADDONS] Подписка {email} не найдена")
         await callback.message.answer("❌ Подписка не найдена.")
         return
+    if record.get("tg_id") != callback.from_user.id:
+        await callback.answer("Доступ запрещён.", show_alert=True)
+        return
 
     tariff_id = record.get("tariff_id")
     if not tariff_id:

handlers/tariffs/addons/key_addons_pack.py

@@ -396,6 +396,9 @@ async def start_key_addons(callback: CallbackQuery, state: FSMContext, session:
         logger.warning(f"[ADDONS] PACK_MODE: подписка {email} не найдена")
         await callback.message.answer("❌ Подписка не найдена.")
         return
+    if record.get("tg_id") != callback.from_user.id:
+        await callback.answer("Доступ запрещён.", show_alert=True)
+        return
 
     tariff_id = record.get("tariff_id")
     if not tariff_id:

hooks/constants.py

@@ -1 +1 @@
-DEFAULT_HOOK_TIMEOUT = 4.0
+DEFAULT_HOOK_TIMEOUT = 8.0

middlewares/ban_checker.py

@@ -10,7 +10,7 @@ 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 core.redis_cache import cache_delete, cache_get, cache_key, cache_set
 from database import async_session_maker
 from database.models import ManualBan
 from logger import logger
@@ -20,6 +20,11 @@ TZ = timezone("Europe/Moscow")
 _BAN_CACHE_TTL = BAN_CACHE_TTL_SEC
 
 
+async def invalidate_ban_cache(tg_id: int) -> None:
+    """Сбросить кэш статуса бана после добавления/снятия бана."""
+    await cache_delete(cache_key("ban_status", tg_id))
+
+
 class BanCheckerMiddleware(BaseMiddleware):
     """Проверка банов."""
 
@@ -81,10 +86,14 @@ class BanCheckerMiddleware(BaseMiddleware):
                         until_parsed = datetime.fromisoformat(until_raw)
                     except ValueError:
                         until_parsed = None
-                ban_info = {
-                    "reason": cached.get("reason") or "не указана",
-                    "until": until_parsed,
-                }
+                if until_parsed is not None and until_parsed < datetime.utcnow():
+                    ban_info = None
+                    await cache_delete(cache_key("ban_status", tg_id))
+                else:
+                    ban_info = {
+                        "reason": cached.get("reason") or "не указана",
+                        "until": until_parsed,
+                    }
         else:
             session = data.get("session")
             if session is not None and getattr(session, "execute", None) is not None:

middlewares/loggings.py

@@ -1,3 +1,4 @@
+import asyncio
 from collections.abc import Awaitable, Callable
 from typing import Any, TypedDict
 
@@ -13,8 +14,18 @@ class UserInfo(TypedDict):
     action: str | None
 
 
+def _log_activity_sync(user_info: UserInfo) -> None:
+    """Синхронный вывод в лог, чтобы не блокировать event loop в create_task."""
+    logger.info(
+        f"Активность пользователя │ "
+        f"ID: {str(user_info['user_id']).ljust(10)} │ "
+        f"Имя: {user_info['username'] or '—':<15} │ "
+        f"Действие: {user_info['action'] or '—'}"
+    )
+
+
 class LoggingMiddleware(BaseMiddleware):
-    """Middleware для логирования действий пользователя."""
+    """Middleware для логирования действий пользователя. Лог пишется в фоне, не задерживая обработчик."""
 
     async def __call__(
         self,
@@ -25,12 +36,7 @@ class LoggingMiddleware(BaseMiddleware):
         user_info = self._extract_user_info(event)
 
         if user_info["user_id"]:
-            logger.info(
-                f"Активность пользователя │ "
-                f"ID: {str(user_info['user_id']).ljust(10)} │ "
-                f"Имя: {user_info['username'] or '—':<15} │ "
-                f"Действие: {user_info['action'] or '—'}"
-            )
+            asyncio.create_task(asyncio.to_thread(_log_activity_sync, user_info))
 
         return await handler(event, data)
 

middlewares/session.py

@@ -115,6 +115,10 @@ class SessionMiddleware(BaseMiddleware):
         t0 = time.perf_counter() if LOG_SESSION_DURATION else None
 
         async with self.sessionmaker() as session:
+            try:
+                await session.rollback()
+            except Exception:
+                pass
             proxy = _SessionProxy(session, self.sessionmaker, data)
             data["session"] = proxy
             committed = False
@@ -160,7 +164,7 @@ class SessionMiddleware(BaseMiddleware):
                 rolled_back = True
                 raise
             finally:
-                if not committed and not rolled_back and not data.get("_session_released_early"):
+                if not committed and not data.get("_session_released_early"):
                     try:
                         await session.rollback()
                     except Exception:

panels/remnawave_runtime.py

@@ -9,9 +9,11 @@ from core.cache_config import (
     REMNAWAVE_MAX_CONCURRENCY,
     REMNAWAVE_ACTION_TIMEOUT_SEC,
     REMNAWAVE_PROFILE_CACHE_TTL_SEC,
+    REMNAWAVE_PROFILE_ERROR_CACHE_TTL_SEC,
     REMNAWAVE_PROFILE_TIMEOUT_SEC,
     REMNAWAVE_SERVER_CACHE_TTL_SEC,
 )
+from core.executor import run_io
 from core.redis_cache import cache_delete_pattern, cache_get, cache_key, cache_set
 from database import get_servers
 from logger import logger
@@ -20,6 +22,93 @@ from panels.remnawave import RemnawaveAPI
 _remnawave_semaphore = asyncio.Semaphore(REMNAWAVE_MAX_CONCURRENCY)
 
 
+async def _fetch_profile_http_only(api_url: str, client_id: str) -> dict[str, Any] | None:
+    """Только HTTP к панели: логин + устройства + юзер. Без кэша и без resolve. Вызывается из потока."""
+    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:
+            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")
+        return {
+            "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, Exception):
+        return None
+    finally:
+        if hasattr(api, "aclose"):
+            try:
+                await api.aclose()
+            except Exception:
+                pass
+
+
+def _run_profile_http_in_thread(api_url: str, client_id: str) -> dict[str, Any] | None:
+    """Синхронная обёртка: свой event loop в потоке, чтобы не блокировать основной цикл бота."""
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        return loop.run_until_complete(_fetch_profile_http_only(api_url, client_id))
+    finally:
+        loop.close()
+
+
+def _run_with_api_in_thread(
+    api_url: str,
+    operation: Callable[[RemnawaveAPI], Awaitable[Any]],
+    timeout_sec: float,
+) -> Any:
+    """Синхронная обёртка: логин + operation(api) в отдельном event loop в потоке."""
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    api = RemnawaveAPI(api_url)
+    try:
+        logged_in = True
+        if not REMNAWAVE_TOKEN_LOGIN_ENABLED:
+            logged_in = loop.run_until_complete(
+                asyncio.wait_for(api.login(REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD), timeout=timeout_sec)
+            )
+        if not logged_in:
+            return None
+        coro = operation(api)
+        return loop.run_until_complete(asyncio.wait_for(coro, timeout=timeout_sec))
+    except (asyncio.TimeoutError, Exception):
+        return None
+    finally:
+        if hasattr(api, "aclose"):
+            try:
+                loop.run_until_complete(api.aclose())
+            except Exception:
+                pass
+        loop.close()
+
+
 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
@@ -96,63 +185,13 @@ async def get_remnawave_profile(
         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
+        profile = await run_io(_run_profile_http_in_thread, api_url, client_id)
+        if profile is None:
+            logger.warning(f"[Remnawave] Таймаут или ошибка профиля для client_id={client_id}")
 
-            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)
+    ttl = REMNAWAVE_PROFILE_CACHE_TTL_SEC if profile else REMNAWAVE_PROFILE_ERROR_CACHE_TTL_SEC
+    await cache_set(pkey, profile, ttl)
     return profile
 
 
@@ -183,26 +222,7 @@ async def with_remnawave_api(
         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
+        result = await run_io(_run_with_api_in_thread, api_url, operation, timeout_sec)
+        if result is None:
+            logger.warning(f"[Remnawave] Таймаут или ошибка операции для server_ref={server_ref}")
+        return result