database optimization/ back button in the configurator/ query competition and more

core/bootstrap.py

@@ -1,4 +1,6 @@
 from database import async_session_maker
+from database.db import warm_pool
+from database.tariffs import initialize_all_tariff_weights
 
 from .settings.buttons_config import BUTTONS_CONFIG, load_buttons_config, update_buttons_config
 from .settings.management_config import MANAGEMENT_CONFIG, load_management_config, update_management_config
@@ -11,7 +13,9 @@ from .settings.tariffs_config import TARIFFS_CONFIG, load_tariffs_config, update
 
 
 async def bootstrap() -> None:
+    await warm_pool()
     async with async_session_maker() as session:
+        await initialize_all_tariff_weights(session)
         await load_buttons_config(session)
         await load_notifications_config(session)
         await load_modes_config(session)

database/coupons.py

@@ -113,6 +113,7 @@ async def create_coupon_usage(session: AsyncSession, coupon_id: int, user_id: in
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при сохранении использования купона: {e}")
         await session.rollback()
+        raise
 
 
 async def check_coupon_usage(session: AsyncSession, coupon_id: int, user_id: int) -> bool:
@@ -136,6 +137,7 @@ async def update_coupon_usage_count(session: AsyncSession, coupon_id: int):
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при обновлении купона {coupon_id}: {e}")
         await session.rollback()
+        raise
 
 
 def apply_percent_coupon(price_rub: int, coupon: Coupon) -> tuple[int, int]:

database/db.py

@@ -1,15 +1,24 @@
+import asyncio
+
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import declarative_base
 
-from config import DATABASE_URL
+from config import DATABASE_URL, DB_MAX_OVERFLOW, DB_POOL_SIZE
+
+
+CONCURRENT_UPDATES_LIMIT = DB_POOL_SIZE + DB_MAX_OVERFLOW
+MAX_UPDATE_AGE_SEC = 28
 
 engine = create_async_engine(
     DATABASE_URL,
     echo=False,
     future=True,
-    pool_size=100,
+    pool_size=DB_POOL_SIZE,
-    max_overflow=200,
+    max_overflow=DB_MAX_OVERFLOW,
     pool_timeout=60,
+    pool_pre_ping=True,
+    pool_recycle=300,
 )
 
 async_session_maker = async_sessionmaker(
@@ -19,3 +28,20 @@ async_session_maker = async_sessionmaker(
 )
 
 Base = declarative_base()
+
+WARM_POOL_COUNT = 10
+
+
+async def warm_pool() -> None:
+    """
+    Прогревает пул соединений при старте.
+    """
+
+    async def _one() -> None:
+        async with async_session_maker() as session:
+            await session.execute(text("SELECT 1"))
+
+    count = min(WARM_POOL_COUNT, DB_POOL_SIZE)
+    if count <= 0:
+        return
+    await asyncio.gather(*[asyncio.create_task(_one()) for _ in range(count)])

database/gifts.py

@@ -50,4 +50,4 @@ async def store_gift_link(
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при сохранении подарка {gift_id}: {e}")
         await session.rollback()
-        return False
+        raise

database/importer.py

@@ -92,8 +92,9 @@ async def import_keys_from_3xui_db(db_path: str, session: AsyncSession) -> tuple
                         updated_at=datetime.utcnow(),
                     )
                 )
-            except SQLAlchemyError:
+            except SQLAlchemyError as e:
-                continue
+                await session.rollback()
+                raise RuntimeError(f"Ошибка при импорте пользователя tg_id={tg_id}") from e
 
         key_exists = await session.execute(select(Key).where(Key.client_id == client_id))
         if key_exists.scalar():
@@ -119,8 +120,9 @@ async def import_keys_from_3xui_db(db_path: str, session: AsyncSession) -> tuple
                 )
             )
             imported += 1
-        except SQLAlchemyError:
+        except SQLAlchemyError as e:
-            continue
+            await session.rollback()
+            raise RuntimeError(f"Ошибка при импорте ключа client_id={client_id}") from e
 
     await session.commit()
     return imported, skipped

database/init_db.py

@@ -5,7 +5,6 @@ from sqlalchemy import select
 from config import ADMIN_ID
 from database.db import async_session_maker, engine
 from database.models import Admin, Base, User
-from database.tariffs import initialize_all_tariff_weights
 
 
 async def init_db():
@@ -35,5 +34,3 @@ async def init_db():
                     )
                 )
         await session.commit()
-
-        await initialize_all_tariff_weights(session)

database/keys.py

@@ -85,6 +85,7 @@ async def store_key(
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при сохранении ключа: {e}")
         await session.rollback()
+        raise
 
 
 async def get_keys(session: AsyncSession, tg_id: int):
@@ -155,10 +156,11 @@ async def get_key_count(session: AsyncSession, tg_id: int) -> int:
     return result.scalar() or 0
 
 
-async def delete_key(session: AsyncSession, identifier: int | str):
+async def delete_key(session: AsyncSession, identifier: int | str, commit: bool = True):
     stmt = delete(Key).where(Key.tg_id == identifier if str(identifier).isdigit() else Key.client_id == identifier)
     await session.execute(stmt)
-    await session.commit()
+    if commit:
+        await session.commit()
     logger.info(f"Ключ с идентификатором {identifier} удалён")
 
 

database/notifications.py

@@ -31,6 +31,7 @@ async def add_notification(session: AsyncSession, tg_id: int, notification_type:
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при добавлении уведомления: {e}")
         await session.rollback()
+        raise
 
 
 async def delete_notification(session: AsyncSession, tg_id: int, notification_type: str):
@@ -101,6 +102,7 @@ async def check_hot_lead_discount(session: AsyncSession, tg_id: int) -> dict:
 
     except Exception as e:
         logger.error(f"❌ Ошибка при проверке скидки горячего лида для {tg_id}: {e}")
+        await session.rollback()
         return {"available": False}
 
 
@@ -173,4 +175,5 @@ async def check_notifications_bulk(
 
     except Exception as e:
         logger.error(f"Ошибка при массовой проверке уведомлений типа {notification_type}: {e}")
+        await session.rollback()
         return []

database/payments.py

@@ -106,6 +106,7 @@ async def get_payment_by_id(session: AsyncSession, internal_id: int) -> dict | N
         }
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при поиске платежа id={internal_id}: {e}")
+        await session.rollback()
         return None
 
 
@@ -161,6 +162,7 @@ async def get_payment_by_payment_id(session: AsyncSession, pid: str) -> dict | N
         }
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при поиске платежа payment_id={pid}: {e}")
+        await session.rollback()
         return None
 
 

database/referrals.py

@@ -224,6 +224,7 @@ async def get_referral_stats(session: AsyncSession, referrer_tg_id: int):
 
     except Exception as e:
         logger.error(f"[ReferralStats] Ошибка при получении статистики для пользователя {referrer_tg_id}: {e}")
+        await session.rollback()
         raise
 
 

database/servers.py

@@ -100,6 +100,7 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
         return grouped
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при получении серверов: {e}")
+        await session.rollback()
         return {}
 
 
@@ -124,6 +125,7 @@ async def check_server_name_by_cluster(session: AsyncSession, server_name: str)
         return {"cluster_name": row[0]} if row else None
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при поиске кластера для сервера {server_name}: {e}")
+        await session.rollback()
         return None
 
 
@@ -161,6 +163,7 @@ async def get_server_by_name(session: AsyncSession, server_name: str) -> dict |
         return None
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при получении сервера {server_name}: {e}")
+        await session.rollback()
         return None
 
 
@@ -209,6 +212,7 @@ async def get_available_clusters(session: AsyncSession) -> list[str]:
         return [row[0] for row in result.all()]
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при получении списка кластеров: {e}")
+        await session.rollback()
         return []
 
 

database/tariffs.py

@@ -73,6 +73,7 @@ async def get_tariffs(
         return tariffs
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при получении тарифов: {e}")
+        await session.rollback()
         return []
 
 
@@ -83,6 +84,7 @@ async def get_tariff_by_id(session: AsyncSession, tariff_id: int):
         return dict(tariff.__dict__) if tariff else None
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при получении тарифа по ID {tariff_id}: {e}")
+        await session.rollback()
         return None
 
 
@@ -179,6 +181,7 @@ async def check_tariff_exists(session: AsyncSession, tariff_id: int):
         return False
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при проверке тарифа {tariff_id}: {e}")
+        await session.rollback()
         return False
 
 
@@ -195,6 +198,7 @@ async def get_tariff_sort_order(session: AsyncSession, tariff_id: int) -> int:
         return sort_order
     except SQLAlchemyError as e:
         logger.error(f"[TARIFF] Ошибка при получении sort_order для тарифа {tariff_id}: {e}")
+        await session.rollback()
         return None
 
 

database/temporary_data.py

@@ -25,6 +25,7 @@ async def create_temporary_data(session: AsyncSession, tg_id: int, state: str, d
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при сохранении временных данных для {tg_id}: {e}")
         await session.rollback()
+        raise
 
 
 async def get_temporary_data(session: AsyncSession, tg_id: int) -> dict | None:

database/tracking_sources.py

@@ -21,6 +21,7 @@ async def create_tracking_source(session: AsyncSession, name: str, code: str, ty
     except SQLAlchemyError as e:
         logger.error(f"❌ Ошибка при создании источника {code}: {e}")
         await session.rollback()
+        raise
 
 
 async def get_all_tracking_sources(session: AsyncSession) -> list[dict]:

database/users.py

@@ -100,6 +100,7 @@ async def set_user_balance(session: AsyncSession, tg_id: int, balance: float) ->
     except SQLAlchemyError as e:
         logger.error(f"Ошибка при установке баланса для пользователя {tg_id}: {e}")
         await session.rollback()
+        raise
 
 
 async def update_trial(session: AsyncSession, tg_id: int, status: int):
@@ -110,6 +111,7 @@ async def update_trial(session: AsyncSession, tg_id: int, status: int):
     except SQLAlchemyError as e:
         logger.error(f"[DB] Ошибка при обновлении триала пользователя {tg_id}: {e}")
         await session.rollback()
+        raise
 
 
 async def get_trial(session: AsyncSession, tg_id: int) -> int:
@@ -205,7 +207,7 @@ async def delete_user_data(session: AsyncSession, tg_id: int):
             delete(Referral).where(or_(Referral.referrer_tg_id == tg_id, Referral.referred_tg_id == tg_id))
         )
         await session.execute(delete(CouponUsage).where(CouponUsage.user_id == tg_id))
-        await delete_key(session, tg_id)
+        await delete_key(session, tg_id, commit=False)
         await session.execute(delete(TemporaryData).where(TemporaryData.tg_id == tg_id))
         await session.execute(delete(BlockedUser).where(BlockedUser.tg_id == tg_id))
         await session.execute(delete(User).where(User.tg_id == tg_id))

filters/admin.py

@@ -1,20 +1,48 @@
+import time
+
 from aiogram.filters import BaseFilter
 from aiogram.types import CallbackQuery, Message
-from sqlalchemy import select, exists
+from sqlalchemy import select
 
+from config import ADMIN_ID
 from database.db import async_session_maker
 from database.models import Admin
 
 
+_ADMIN_CACHE: dict[int, tuple[float, bool, bool]] = {}
+_ADMIN_CACHE_TTL = 60
+
+
+def _get_cached_admin(user_id: int) -> tuple[bool, bool] | None:
+    now = time.time()
+    entry = _ADMIN_CACHE.get(user_id)
+    if entry and entry[0] > now:
+        return entry[1], entry[2]
+    return None
+
+
+def _set_cached_admin(user_id: int, is_admin: bool, is_superadmin: bool) -> None:
+    _ADMIN_CACHE[user_id] = (time.time() + _ADMIN_CACHE_TTL, is_admin, is_superadmin)
+
+
 class IsAdminFilter(BaseFilter):
     async def __call__(self, event: Message | CallbackQuery) -> bool:
         if not event.from_user:
             return False
 
+        user_id = event.from_user.id
+        cached = _get_cached_admin(user_id)
+        if cached is not None:
+            return cached[0]
+
         try:
             async with async_session_maker() as session:
-                result = await session.execute(select(exists().where(Admin.tg_id == event.from_user.id)))
+                admin = (await session.execute(select(Admin).where(Admin.tg_id == user_id))).scalar_one_or_none()
-                return result.scalar()
+                admin_ids = (ADMIN_ID,) if isinstance(ADMIN_ID, int) else ADMIN_ID
+                is_admin = admin is not None or user_id in admin_ids
+                is_super = admin.role != "moderator" if admin else (user_id in admin_ids)
+                _set_cached_admin(user_id, is_admin, is_super)
+                return is_admin
         except (Exception,):
             return False
 
@@ -24,13 +52,19 @@ class IsSuperAdminFilter(BaseFilter):
         if not event.from_user:
             return False
 
+        user_id = event.from_user.id
+        cached = _get_cached_admin(user_id)
+        if cached is not None:
+            return cached[1]
+
         try:
             async with async_session_maker() as session:
-                admin = (
+                admin = (await session.execute(select(Admin).where(Admin.tg_id == user_id))).scalar_one_or_none()
-                    await session.execute(select(Admin).where(Admin.tg_id == event.from_user.id))
-                ).scalar_one_or_none()
                 if not admin:
+                    _set_cached_admin(user_id, False, False)
                     return False
-                return admin.role != "moderator"
+                is_super = admin.role != "moderator"
+                _set_cached_admin(user_id, True, is_super)
+                return is_super
         except (Exception,):
             return False

handlers/fallback_router.py

@@ -1,9 +1,10 @@
+from typing import Any
+
 from aiogram import F, Router
 from aiogram.types import InlineKeyboardButton, Message
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 
 from config import SUPPORT_CHAT_URL
-from database import async_session_maker
 from handlers.buttons import MAIN_MENU, SUPPORT
 from handlers.texts import FALLBACK_MESSAGE
 from hooks.hooks import run_hooks
@@ -13,23 +14,21 @@ fallback_router = Router()
 
 
 @fallback_router.message(F.text)
-async def handle_unhandled_messages(message: Message):
+async def handle_unhandled_messages(message: Message, session: Any):
-    async with async_session_maker() as session:
+    await run_hooks(
-        await run_hooks(
+        "user_message",
-            "user_message",
+        user_id=message.from_user.id,
-            user_id=message.from_user.id,
+        message_text=message.text,
-            message_text=message.text,
+        username=message.from_user.username,
-            username=message.from_user.username,
+        first_name=message.from_user.first_name,
-            first_name=message.from_user.first_name,
+        last_name=message.from_user.last_name,
-            last_name=message.from_user.last_name,
+        session=session,
-            session=session,
+        message=message,
-            message=message,
+    )
-        )
 
     keyboard = InlineKeyboardBuilder()
     keyboard.row(InlineKeyboardButton(text=SUPPORT, url=SUPPORT_CHAT_URL))
     keyboard.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
-
     await message.answer(
         FALLBACK_MESSAGE,
         reply_markup=keyboard.as_markup(),

handlers/keys/key_create.py

@@ -306,6 +306,7 @@ async def handle_key_creation(
             tg_id=tg_id,
             cluster_name=cluster_name,
             group_code=group_code,
+            tariff_subgroup_hash=None,
         )
         await state.set_state(Form.waiting_for_server_selection)
 
@@ -329,6 +330,8 @@ async def show_tariffs_in_subgroup_user(callback: CallbackQuery, state: FSMConte
         )
         return
 
+    await state.update_data(tariff_subgroup_hash=subgroup_hash)
+
     tariffs_for_cluster = await get_tariffs_for_cluster(session, cluster_name)
     filtered: list[dict[str, Any]] = []
 
@@ -383,6 +386,59 @@ async def back_to_tariff_group_list(callback: CallbackQuery, state: FSMContext,
     )
 
 
+@router.callback_query(F.data == "back_to_subgroup_tariffs")
+async def back_to_subgroup_tariffs(callback: CallbackQuery, state: FSMContext, session: AsyncSession):
+    """Возврат к списку тарифов текущей подгруппы (из конфигуратора)."""
+    data = await state.get_data()
+    subgroup_hash = data.get("tariff_subgroup_hash")
+    if not subgroup_hash:
+        await back_to_tariff_group_list(callback, state, session)
+        return
+    cluster_name = data.get("cluster_name")
+    group_code = data.get("group_code")
+
+    subgroup = await find_subgroup_by_hash(session, subgroup_hash, group_code)
+    if not subgroup:
+        await back_to_tariff_group_list(callback, state, session)
+        return
+
+    tariffs_for_cluster = await get_tariffs_for_cluster(session, cluster_name)
+    filtered: list[dict[str, Any]] = []
+    if tariffs_for_cluster:
+        gc = tariffs_for_cluster[0].get("group_code")
+        if gc:
+            tariffs = await get_tariffs(session, group_code=gc)
+            filtered = [
+                t for t in tariffs if t.get("subgroup_title") == subgroup and t.get("is_active")
+            ]
+
+    if not filtered:
+        await back_to_tariff_group_list(callback, state, session)
+        return
+
+    tg_id = callback.from_user.id
+    language_code = getattr(callback.from_user, "language_code", None)
+    builder = InlineKeyboardBuilder()
+    for tariff in filtered:
+        await add_tariff_button_generic(
+            builder=builder,
+            tariff=tariff,
+            session=session,
+            tg_id=tg_id,
+            language_code=language_code,
+            callback_prefix="select_tariff_plan",
+        )
+    builder.row(InlineKeyboardButton(text=BACK, callback_data="back_to_tariff_group_list"))
+    builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
+
+    await edit_or_send_message(
+        target_message=callback.message,
+        text=f"<b>{subgroup}</b>\n\nВыберите тариф:",
+        reply_markup=builder.as_markup(),
+    )
+    await callback.answer()
+
+
 async def create_key(
     tg_id: int,
     expiry_time: datetime,

handlers/payments/heleket/handlers.py

@@ -83,7 +83,7 @@ async def handle_custom_amount_input_heleket(
     method = enabled_methods[0]
 
     try:
-        payment_url = await generate_heleket_payment_link(amount, tg_id, method)
+        payment_url = await generate_heleket_payment_link(amount, tg_id, method, session)
 
         if not payment_url or payment_url == "https://heleket.com/":
             await edit_or_send_message(

handlers/payments/heleket/service.py

@@ -243,7 +243,7 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext,
             amount_rub = int(await to_rub(user_amount, "USD", session=session_http))
 
     await state.update_data(amount=amount_rub)
-    payment_url = await generate_heleket_payment_link(amount_rub, message.chat.id, method)
+    payment_url = await generate_heleket_payment_link(amount_rub, message.chat.id, method, session)
 
     if not payment_url or payment_url == "https://heleket.com/":
         await edit_or_send_message(
@@ -298,7 +298,7 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F
         return
 
     await state.update_data(amount=amount)
-    payment_url = await generate_heleket_payment_link(amount, callback_query.message.chat.id, method)
+    payment_url = await generate_heleket_payment_link(amount, callback_query.message.chat.id, method, session)
 
     if not payment_url or payment_url == "https://heleket.com/":
         await edit_or_send_message(
@@ -323,10 +323,13 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F
     await state.set_state(ReplenishBalanceHeleket.waiting_for_payment_confirmation)
 
 
-async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) -> str:
+async def generate_heleket_payment_link(
+    amount: int, tg_id: int, method: dict, session: AsyncSession | None = None
+) -> str:
     """
     Создание платежа в Heleket и получение ссылки на оплату.
     amount — сумма в RUB, method['currency'] — валюта провайдера (обычно USD).
+    session — сессия из хендлера; если не передана, создаётся своя (лишняя нагрузка на пул).
     """
     url = "https://api.heleket.com/v1/payment"
     unique_order_id = f"{int(time.time())}_{tg_id}"
@@ -371,9 +374,9 @@ async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) -
                         if resp_json.get("state") == 0:
                             payment_url = resp_json.get("result", {}).get("url")
                             if payment_url:
-                                async with async_session_maker() as dbs:
+                                if session is not None:
                                     await add_payment(
-                                        session=dbs,
+                                        session=session,
                                         tg_id=tg_id,
                                         amount=float(amount),
                                         payment_system="HELEKET",
@@ -381,6 +384,17 @@ async def generate_heleket_payment_link(amount: int, tg_id: int, method: dict) -
                                         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,
+                                        )
                                 logger.info(f"Heleket payment URL created for user {tg_id}")
                                 return payment_url
                             else:

handlers/payments/kassai/handlers.py

@@ -107,7 +107,7 @@ async def _handle_custom_amount_input_kassai(
         return
 
     try:
-        payment_url = await generate_kassai_payment_link(amount, tg_id, method)
+        payment_url = await generate_kassai_payment_link(amount, tg_id, method, session)
 
         if not payment_url or payment_url == "https://fk.life/":
             await edit_or_send_message(

handlers/payments/kassai/service.py

@@ -266,7 +266,7 @@ async def handle_custom_amount_input(message: types.Message, state: FSMContext,
             amount_rub = int(await to_rub(user_amount, "USD", session=session_http))
 
     await state.update_data(amount=amount_rub)
-    payment_url = await generate_kassai_payment_link(amount_rub, message.chat.id, method)
+    payment_url = await generate_kassai_payment_link(amount_rub, message.chat.id, method, session)
 
     if not payment_url or payment_url == "https://fk.life/":
         await edit_or_send_message(
@@ -328,7 +328,7 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F
         return
 
     await state.update_data(amount=amount)
-    payment_url = await generate_kassai_payment_link(amount, callback_query.message.chat.id, method)
+    payment_url = await generate_kassai_payment_link(amount, callback_query.message.chat.id, method, session)
 
     if not payment_url or payment_url == "https://fk.life/":
         await edit_or_send_message(
@@ -353,9 +353,12 @@ async def process_amount_selection(callback_query: types.CallbackQuery, state: F
     await state.set_state(ReplenishBalanceKassaiState.waiting_for_payment_confirmation)
 
 
-async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) -> str:
+async def generate_kassai_payment_link(
+    amount: int, tg_id: int, method: dict, session: AsyncSession | None = None
+) -> str:
     """
-    Создание заказа в KassaAI и получение ссылки на оплату
+    Создание заказа в KassaAI и получение ссылки на оплату.
+    session — сессия из хендлера; если не передана, создаётся своя (лишняя нагрузка на пул).
     """
     nonce = int(time.time())
     unique_payment_id = f"{nonce}_{tg_id}"
@@ -393,9 +396,9 @@ async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) ->
                         if resp_json.get("type") == "success":
                             payment_url = resp_json.get("location")
                             if payment_url:
-                                async with async_session_maker() as dbs:
+                                if session is not None:
                                     await add_payment(
-                                        session=dbs,
+                                        session=session,
                                         tg_id=tg_id,
                                         amount=float(amount),
                                         payment_system="KASSAI",
@@ -403,6 +406,17 @@ async def generate_kassai_payment_link(amount: int, tg_id: int, method: dict) ->
                                         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,
+                                        )
                                 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/profile.py

@@ -1,4 +1,3 @@
-import asyncio
 import os
 
 from aiogram import F, Router
@@ -63,44 +62,35 @@ async def process_callback_view_profile(
     chat_id = chat.id
     username = get_username(user or chat)
 
-    key_count, balance_rub, trial_status = await asyncio.gather(
+    key_count = await get_key_count(session, chat_id)
-        get_key_count(session, chat_id),
+    balance_rub = await get_balance(session, chat_id)
-        get_balance(session, chat_id),
+    trial_status = await get_trial(session, chat_id)
-        get_trial(session, chat_id),
-    )
     balance_rub = balance_rub or 0
 
-    balance_text_task = asyncio.create_task(
+    balance_text = await format_for_user(
-        format_for_user(
+        session,
-            session,
+        chat_id,
-            chat_id,
+        balance_rub,
-            balance_rub,
+        getattr(user, "language_code", None),
-            getattr(user, "language_code", None),
-        )
     )
-    profile_menu_buttons_task = asyncio.create_task(
+    profile_menu_buttons = await run_hooks(
-        run_hooks("profile_menu", chat_id=chat_id, admin=admin, session=session)
+        "profile_menu", chat_id=chat_id, admin=admin, session=session
     )
-    profile_text_hooks_task = asyncio.create_task(
+    text_hooks = await run_hooks(
-        run_hooks(
+        "profile_text",
-            "profile_text",
+        username=username,
-            username=username,
+        chat_id=chat_id,
-            chat_id=chat_id,
+        balance=int(balance_rub),
-            balance=int(balance_rub),
+        key_count=key_count,
-            key_count=key_count,
+        session=session,
-            session=session,
-        )
     )
 
-    balance_text = await balance_text_task
-
     profile_message = profile_message_send(username, chat_id, balance_text, key_count)
     if key_count == 0:
         profile_message += ADD_SUBSCRIPTION_HINT
     else:
         profile_message += f"\n<blockquote><i>{NEWS_MESSAGE}</i></blockquote>"
 
-    text_hooks = await profile_text_hooks_task
     if text_hooks:
         profile_message = text_hooks[0]
 
@@ -127,7 +117,6 @@ async def process_callback_view_profile(
     if extra_buttons:
         builder.row(*extra_buttons)
 
-    profile_menu_buttons = await profile_menu_buttons_task
     builder = insert_hook_buttons(builder, profile_menu_buttons)
 
     if BUTTONS_CONFIG.get("INSTRUCTIONS_BUTTON_ENABLE", INSTRUCTIONS_BUTTON):

handlers/tariffs/addons/key_addons_pack.py

@@ -16,7 +16,6 @@ from database import (
     get_key_details,
     get_tariff_by_id,
     save_key_config_with_mode,
-    update_balance,
 )
 from database.models import User
 from handlers.buttons import BACK, CONFIRM_ADDON_BUTTON_TEXT, PAYMENT

handlers/tariffs/buy/key_tariffs.py

@@ -12,7 +12,7 @@ from config import USE_NEW_PAYMENT_FLOW
 from core.settings.tariffs_config import normalize_tariff_config
 from database import get_balance, get_tariff_by_id
 from database.notifications import check_hot_lead_discount
-from handlers.buttons import CONFIG_PAY_BUTTON_TEXT, MAIN_MENU, PAYMENT
+from handlers.buttons import BACK, CONFIG_PAY_BUTTON_TEXT, MAIN_MENU, PAYMENT
 from handlers.payments.currency_rates import format_for_user
 from handlers.payments.fast_payment_flow import try_fast_payment_flow
 from handlers.tariffs.tariff_display import GB
@@ -237,6 +237,7 @@ async def proceed_purchase_with_values(
 
         builder = InlineKeyboardBuilder()
         builder.row(InlineKeyboardButton(text=PAYMENT, callback_data="pay"))
+        builder.row(InlineKeyboardButton(text=BACK, callback_data="back_to_tariff_group_list"))
         builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
         await edit_or_send_message(
             target_message=callback_query.message,
@@ -507,12 +508,18 @@ async def render_user_config_screen(
     is_renew_mode = data.get("renew_mode") == "renew"
     confirm_prefix = "cfg_renew_confirm" if is_renew_mode else "cfg_user_confirm"
 
+    back_callback = (
+        "back_to_subgroup_tariffs"
+        if data.get("tariff_subgroup_hash")
+        else "back_to_tariff_group_list"
+    )
     builder.row(
         InlineKeyboardButton(
             text=CONFIG_PAY_BUTTON_TEXT.format(amount=price_text),
             callback_data=f"{confirm_prefix}|{tariff_id}",
         )
     )
+    builder.row(InlineKeyboardButton(text=BACK, callback_data=back_callback))
     builder.row(InlineKeyboardButton(text=MAIN_MENU, callback_data="profile"))
 
     await state.update_data(
@@ -590,31 +597,17 @@ async def start_user_tariff_configurator(
     await render_user_config_screen(callback_query, state, session)
 
 
-async def show_price_and_confirm(callback_query: CallbackQuery, state: FSMContext, session: Any | None):
+async def show_price_and_confirm(callback_query: CallbackQuery, state: FSMContext, session: Any):
     """Обновляет экран конфигурации и показывает актуальную цену."""
-    if session is None:
-        from database import async_session_maker
-
-        async with async_session_maker() as new_session:
-            await show_price_and_confirm(callback_query, state, new_session)
-        return
-
     await render_user_config_screen(callback_query, state, session)
 
 
-async def finalize_config_and_purchase(callback_query: CallbackQuery, state: FSMContext, session: Any | None):
+async def finalize_config_and_purchase(callback_query: CallbackQuery, state: FSMContext, session: Any):
     """Фиксирует выбор пользователя и проводит оплату тарифа."""
     data = await state.get_data()
     tariff_id = data.get("config_tariff_id")
     cfg = data.get("tariff_config") or {}
 
-    if session is None:
-        from database import async_session_maker
-
-        async with async_session_maker() as new_session:
-            await finalize_config_and_purchase(callback_query, state, new_session)
-        return
-
     tariff = await get_tariff_by_id(session, int(tariff_id))
     if not tariff:
         await edit_or_send_message(

middlewares/__init__.py

@@ -8,6 +8,7 @@ from middlewares.subscription import SubscriptionMiddleware
 
 from .admin import AdminMiddleware
 from .answer import CallbackAnswerMiddleware
+from .concurrency import ConcurrencyLimiterMiddleware
 from .direct_start_blocker import DirectStartBlockerMiddleware
 from .loggings import LoggingMiddleware
 from .maintenance import MaintenanceModeMiddleware
@@ -34,6 +35,7 @@ def register_middleware(
         dispatcher.update.outer_middleware(StreamProbeMiddleware("global"))
 
     if sessionmaker:
+        dispatcher.update.outer_middleware(wrap(ConcurrencyLimiterMiddleware(), "concurrency"))
         dispatcher.update.outer_middleware(wrap(SessionMiddleware(sessionmaker), "session"))
 
     if DISABLE_DIRECT_START:

middlewares/answer.py

@@ -15,8 +15,14 @@ class CallbackAnswerMiddleware(BaseMiddleware):
         data: dict[str, Any],
     ) -> Any:
         if isinstance(event, CallbackQuery):
-            await event.answer()
+            try:
+                await event.answer()
+            except Exception:
+                pass
             if isinstance(event.message, InaccessibleMessage):
-                new_message = await bot.send_message(event.message.chat.id, "⏳")
+                try:
-                object.__setattr__(event, "message", new_message)
+                    new_message = await bot.send_message(event.message.chat.id, "⏳")
+                    object.__setattr__(event, "message", new_message)
+                except Exception:
+                    pass
         return await handler(event, data)

middlewares/concurrency.py

@@ -0,0 +1,59 @@
+import asyncio
+import time
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from aiogram import BaseMiddleware, Bot
+from aiogram.types import CallbackQuery, Message, TelegramObject
+
+from database.db import CONCURRENT_UPDATES_LIMIT, MAX_UPDATE_AGE_SEC
+
+
+class ConcurrencyLimiterMiddleware(BaseMiddleware):
+    """
+    Регистрируется до SessionMiddleware. Ограничивает число апдейтов, одновременно
+    получающих сессию, и отсекает апдейты, ждавшие слишком долго.
+    """
+
+    def __init__(self) -> None:
+        self._semaphore = asyncio.Semaphore(CONCURRENT_UPDATES_LIMIT)
+
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        data["request_time"] = time.monotonic()
+        await self._semaphore.acquire()
+        try:
+            age = time.monotonic() - data["request_time"]
+            if age > MAX_UPDATE_AGE_SEC:
+                await self._reject_stale(event, data)
+                return None
+            return await handler(event, data)
+        finally:
+            self._semaphore.release()
+
+    async def _reject_stale(self, event: TelegramObject, data: dict[str, Any]) -> None:
+        if isinstance(event, CallbackQuery):
+            bot: Bot = data.get("bot")
+            if bot:
+                try:
+                    await bot.answer_callback_query(
+                        event.id,
+                        text="Время ожидания истекло. Нажмите ещё раз.",
+                        show_alert=False,
+                    )
+                except Exception:
+                    pass
+        elif isinstance(event, Message) and event.text and event.chat:
+            bot: Bot = data.get("bot")
+            if bot:
+                try:
+                    await bot.send_message(
+                        event.chat.id,
+                        "Сейчас высокая нагрузка. Отправьте команду ещё раз через пару секунд.",
+                    )
+                except Exception:
+                    pass

middlewares/session.py

@@ -9,10 +9,20 @@ class SessionMiddleware(BaseMiddleware):
         if data.get("session"):
             return await handler(event, data)
 
-        async with self.sessionmaker() as session:
+        session = self.sessionmaker()
-            data["session"] = session
+        data["session"] = session
+        try:
+            result = await handler(event, data)
+            await session.commit()
+            return result
+        except Exception:
             try:
-                return await handler(event, data)
-            except Exception:
                 await session.rollback()
-                raise
+            except Exception:
+                pass
+            raise
+        finally:
+            try:
+                await session.close()
+            except Exception:
+                pass

utils/custom_emojis.py

@@ -71,8 +71,7 @@ async def _fetch_placeholder(emoji_id: str) -> str:
 
 
 async def _replace_markers(text: str) -> tuple[str, list[MessageEntity]]:
-    """Replace markers with placeholders and build custom emoji entities.
+    """Replace markers with placeholders and build custom emoji entities."""
-    """
     if not text:
         return text, []
 

utils/errors.py

@@ -1,3 +1,5 @@
+import html
+import re
 import traceback
 
 from aiogram import Bot, Dispatcher
@@ -11,6 +13,15 @@ from database import async_session_maker
 from logger import logger
 
 
+_OBFUSCATED_MIN_SEQ = 15
+_PLACEHOLDER = "<obfuscated>"
+
+
+def _sanitize_traceback(text: str) -> str:
+    """Убирает из текста длинные последовательности \\xNN (обфусцированный код)."""
+    return re.sub(r"(\\x[0-9a-fA-F]{2}){" + str(_OBFUSCATED_MIN_SEQ) + r",}", _PLACEHOLDER, text)
+
+
 def setup_error_handlers(dp: Dispatcher) -> None:
     @dp.errors(ExceptionTypeFilter(Exception))
     async def errors_handler(event: ErrorEvent, bot: Bot) -> bool:
@@ -27,11 +38,13 @@ def setup_error_handlers(dp: Dispatcher) -> None:
                 or "message to delete not found" in error_message
             ):
                 try:
-                    tb = "".join(
+                    tb = _sanitize_traceback(
-                        traceback.format_exception(
+                        "".join(
-                            type(event.exception),
+                            traceback.format_exception(
-                            event.exception,
+                                type(event.exception),
-                            event.exception.__traceback__,
+                                event.exception,
+                                event.exception.__traceback__,
+                            )
                         )
                     )
                     logger.warning(f"Показываем стартовое меню из-за TelegramBadRequest: {error_message}")
@@ -109,14 +122,16 @@ def setup_error_handlers(dp: Dispatcher) -> None:
             return True
 
         try:
+            tb_text = _sanitize_traceback(traceback.format_exc())
             for admin_id in ADMIN_ID:
+                exc_text = html.escape(str(event.exception)[:1021])
                 await bot.send_document(
                     chat_id=admin_id,
                     document=BufferedInputFile(
-                        traceback.format_exc().encode(),
+                        tb_text.encode(),
                         filename=f"error_{event.update.update_id}.txt",
                     ),
-                    caption=f"{hbold(type(event.exception).__name__)}: {str(event.exception)[:1021]}...",
+                    caption=f"{hbold(type(event.exception).__name__)}: {exc_text}...",
                 )
 
             from handlers.start import start_entry