feat: add support tickets and imrpove web app loading

backend/bot/app/factories/build_services.py

@@ -7,12 +7,15 @@ from bot.payment_providers import (
     build_provider_configs,
     build_provider_services,
 )
+from bot.services.email_auth_service import EmailAuthService
 from bot.services.lknpd_service import LknpdService
+from bot.services.notification_service import NotificationService
 from bot.services.panel_api_service import PanelApiService
 from bot.services.panel_webhook_service import PanelWebhookService
 from bot.services.promo_code_service import PromoCodeService
 from bot.services.referral_service import ReferralService
 from bot.services.subscription_service import SubscriptionService
+from bot.services.support_service import SupportService
 from config.settings import Settings
 
 
@@ -27,6 +30,23 @@ def build_core_services(
     subscription_service = SubscriptionService(settings, panel_service, bot, i18n)
     referral_service = ReferralService(settings, subscription_service, bot, i18n)
     promo_code_service = PromoCodeService(settings, subscription_service, bot, i18n)
+    email_auth_service = EmailAuthService(settings)
+    notification_service = NotificationService(
+        bot,
+        settings,
+        i18n,
+        session_factory=async_session_factory,
+        email_auth_service=email_auth_service,
+        bot_username=bot_username_for_default_return,
+    )
+    support_service = SupportService(
+        async_session_factory,
+        settings,
+        bot,
+        i18n,
+        notification_service,
+        email_auth_service,
+    )
     panel_webhook_service = PanelWebhookService(
         bot, settings, i18n, async_session_factory, panel_service
     )
@@ -58,6 +78,9 @@ def build_core_services(
         "subscription_service": subscription_service,
         "referral_service": referral_service,
         "promo_code_service": promo_code_service,
+        "notification_service": notification_service,
+        "email_auth_service": email_auth_service,
+        "support_service": support_service,
         "panel_webhook_service": panel_webhook_service,
         "lknpd_service": lknpd_service,
     }

backend/bot/app/web/admin_api.py

@@ -15,6 +15,7 @@ from bot.app.web.admin_api_impl import (
     routes as _routes,
     settings as _settings,
     stats as _stats,
+    support as _support,
     sync as _sync,
     tariffs as _tariffs,
     themes as _themes,
@@ -30,6 +31,7 @@ _MODULES = (
     _payments,
     _promos,
     _logs,
+    _support,
     _broadcast,
     _sync,
     _ads,

backend/bot/app/web/admin_api_impl/routes.py

@@ -45,6 +45,16 @@ def setup_admin_routes(app: web.Application) -> None:
 
     router.add_get("/api/admin/logs", admin_logs_route)
 
+    router.add_get("/api/admin/support/tickets", admin_support_tickets_route)
+    router.add_get("/api/admin/support/tickets/{id:\\d+}", admin_support_ticket_detail_route)
+    router.add_post(
+        "/api/admin/support/tickets/{id:\\d+}/messages",
+        admin_support_ticket_reply_route,
+    )
+    router.add_patch("/api/admin/support/tickets/{id:\\d+}", admin_support_ticket_patch_route)
+    router.add_post("/api/admin/support/tickets/{id:\\d+}/read", admin_support_ticket_read_route)
+    router.add_get("/api/admin/support/stats", admin_support_stats_route)
+
     router.add_post("/api/admin/broadcast", admin_broadcast_route)
     router.add_post("/api/admin/sync", admin_sync_route)
 

backend/bot/app/web/admin_api_impl/support.py

@@ -0,0 +1,251 @@
+# ruff: noqa: F401,F403,F405,I001
+from ._runtime import *  # noqa: F403,F405
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, constr, field_validator
+
+from bot.services.support_service import TicketNotFound
+from db.dal import support_dal, user_dal
+from db.models import SupportTicket, SupportTicketMessage
+
+
+class AdminTicketReplyPayload(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    body: constr(min_length=1, max_length=4000)
+    is_internal_note: bool = False
+
+    @field_validator("body")
+    @classmethod
+    def _strip_body(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("empty_text")
+        return stripped
+
+
+class AdminTicketPatchPayload(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    status: Optional[Literal["open", "awaiting_user", "awaiting_admin", "resolved", "closed"]] = (
+        None
+    )
+    priority: Optional[Literal["low", "normal", "high", "urgent"]] = None
+    category: Optional[Literal["billing", "technical", "account", "other"]] = None
+    assigned_admin_id: Optional[int] = None
+
+
+def _validate_model_payload(model_cls, payload: Dict[str, Any]):
+    try:
+        return model_cls.model_validate(payload), None
+    except ValidationError:
+        return None, _error(400, "invalid_request", "Invalid request")
+
+
+def _support_ticket_payload(ticket: SupportTicket) -> Dict[str, Any]:
+    return {
+        "ticket_id": ticket.ticket_id,
+        "user_id": ticket.user_id,
+        "subject": ticket.subject,
+        "category": ticket.category,
+        "priority": ticket.priority,
+        "status": ticket.status,
+        "assigned_admin_id": ticket.assigned_admin_id,
+        "last_message_at": ticket.last_message_at.isoformat() if ticket.last_message_at else None,
+        "last_message_role": ticket.last_message_role,
+        "unread_user_count": int(ticket.unread_user_count or 0),
+        "unread_admin_count": int(ticket.unread_admin_count or 0),
+        "created_at": ticket.created_at.isoformat() if ticket.created_at else None,
+        "updated_at": ticket.updated_at.isoformat() if ticket.updated_at else None,
+        "closed_at": ticket.closed_at.isoformat() if ticket.closed_at else None,
+    }
+
+
+def _user_display_name(user) -> Optional[str]:
+    if not user:
+        return None
+    name = " ".join(
+        part.strip() for part in [user.first_name, user.last_name] if part and part.strip()
+    ).strip()
+    return name or user.username or user.email or str(user.user_id)
+
+
+def _support_message_payload(
+    message: SupportTicketMessage,
+    *,
+    authors: Optional[Dict[int, Any]] = None,
+) -> Dict[str, Any]:
+    author = authors.get(message.author_user_id) if authors and message.author_user_id else None
+    return {
+        "message_id": message.message_id,
+        "ticket_id": message.ticket_id,
+        "author_role": message.author_role,
+        "author_user_id": message.author_user_id,
+        "author_name": _user_display_name(author),
+        "body": message.body,
+        "is_internal_note": bool(message.is_internal_note),
+        "created_at": message.created_at.isoformat() if message.created_at else None,
+        "read_by_user_at": message.read_by_user_at.isoformat() if message.read_by_user_at else None,
+        "read_by_admin_at": message.read_by_admin_at.isoformat()
+        if message.read_by_admin_at
+        else None,
+    }
+
+
+def _admin_support_user_payload(user) -> Dict[str, Any]:
+    if not user:
+        return {}
+    return {
+        "user_id": user.user_id,
+        "telegram_id": user.telegram_id,
+        "username": user.username,
+        "first_name": user.first_name,
+        "last_name": user.last_name,
+        "email": user.email,
+        "telegram_photo_url": user.telegram_photo_url,
+        "is_banned": bool(user.is_banned),
+        "registration_date": user.registration_date.isoformat() if user.registration_date else None,
+    }
+
+
+def _support_limit_offset(request: web.Request) -> tuple[int, int]:
+    limit = max(1, min(100, int(request.query.get("limit", 25) or 25)))
+    offset = max(0, int(request.query.get("offset", 0) or 0))
+    return limit, offset
+
+
+async def admin_support_tickets_route(request: web.Request) -> web.Response:
+    _require_admin_user_id(request)
+    limit, offset = _support_limit_offset(request)
+    assigned_raw = request.query.get("assigned")
+    assigned_admin_id = None
+    if assigned_raw and assigned_raw not in {"all", "any"}:
+        assigned_admin_id = int(assigned_raw)
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        tickets = await support_dal.list_admin_tickets(
+            session,
+            status=request.query.get("status") or None,
+            priority=request.query.get("priority") or None,
+            category=request.query.get("category") or None,
+            assigned_admin_id=assigned_admin_id,
+            search=request.query.get("search") or None,
+            sort=request.query.get("sort") or "updated_desc",
+            limit=limit,
+            offset=offset,
+        )
+    return web.json_response(
+        {
+            "ok": True,
+            "tickets": [
+                {
+                    **_support_ticket_payload(ticket),
+                    "user": _admin_support_user_payload(getattr(ticket, "user", None)),
+                }
+                for ticket in tickets
+            ],
+        }
+    )
+
+
+async def admin_support_ticket_detail_route(request: web.Request) -> web.Response:
+    _require_admin_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    service = request.app["support_service"]
+    async with async_session_factory() as session:
+        ticket, messages = await support_dal.get_ticket(session, ticket_id, include_internal=True)
+        if not ticket:
+            return _error(404, "not_found", "Ticket not found")
+        user = await user_dal.get_user_by_id(session, ticket.user_id)
+        snapshot = await service.build_user_snapshot(user, session=session) if user else {}
+        author_ids = {m.author_user_id for m in messages if m.author_user_id is not None}
+        authors = {}
+        for author_id in author_ids:
+            author = await user_dal.get_user_by_id(session, author_id)
+            if author:
+                authors[author_id] = author
+    return web.json_response(
+        {
+            "ok": True,
+            "ticket": {
+                **_support_ticket_payload(ticket),
+                "user": _admin_support_user_payload(user),
+            },
+            "messages": [_support_message_payload(m, authors=authors) for m in messages],
+            "user_snapshot": snapshot,
+        }
+    )
+
+
+async def admin_support_ticket_reply_route(request: web.Request) -> web.Response:
+    admin_id = _require_admin_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    payload, error = _validate_model_payload(AdminTicketReplyPayload, await _read_json(request))
+    if error:
+        return error
+    try:
+        ticket, message = await request.app["support_service"].reply_as_admin(
+            admin_id,
+            ticket_id,
+            payload.body,
+            is_internal_note=payload.is_internal_note,
+        )
+    except TicketNotFound:
+        return _error(404, "not_found", "Ticket not found")
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        admin = await user_dal.get_user_by_id(session, admin_id)
+    return web.json_response(
+        {
+            "ok": True,
+            "ticket": _support_ticket_payload(ticket),
+            "message": _support_message_payload(
+                message, authors={admin_id: admin} if admin else {}
+            ),
+        }
+    )
+
+
+async def admin_support_ticket_patch_route(request: web.Request) -> web.Response:
+    admin_id = _require_admin_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    payload, error = _validate_model_payload(AdminTicketPatchPayload, await _read_json(request))
+    if error:
+        return error
+    updates = payload.model_dump(exclude_unset=True)
+    try:
+        if updates.get("status") == "closed":
+            ticket = await request.app["support_service"].close_ticket(admin_id, ticket_id)
+            updates.pop("status", None)
+            if updates:
+                ticket = await request.app["support_service"]._update_and_audit(
+                    admin_id,
+                    ticket_id,
+                    **updates,
+                )
+        else:
+            ticket = await request.app["support_service"]._update_and_audit(
+                admin_id,
+                ticket_id,
+                **updates,
+            )
+    except TicketNotFound:
+        return _error(404, "not_found", "Ticket not found")
+    return web.json_response({"ok": True, "ticket": _support_ticket_payload(ticket)})
+
+
+async def admin_support_ticket_read_route(request: web.Request) -> web.Response:
+    _require_admin_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    await request.app["support_service"].mark_read_as_admin(ticket_id)
+    return web.json_response({"ok": True})
+
+
+async def admin_support_stats_route(request: web.Request) -> web.Response:
+    _require_admin_user_id(request)
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        stats = await support_dal.admin_stats(session)
+    return web.json_response({"ok": True, "stats": stats})

backend/bot/app/web/admin_settings_manifest.py

@@ -235,6 +235,7 @@ SETTINGS_MANIFEST: List[SettingField] = [
     ),
     SettingField("LOG_NEW_USERS", "bool", "notifications", "Логировать новых пользователей"),
     SettingField("LOG_PAYMENTS", "bool", "notifications", "Логировать платежи"),
+    SettingField("LOG_SUPPORT", "bool", "notifications", "Логировать тикеты поддержки"),
     SettingField(
         "LOG_PROMO_ACTIVATIONS", "bool", "notifications", "Логировать активации промокодов"
     ),
@@ -248,8 +249,8 @@ SETTINGS_MANIFEST: List[SettingField] = [
         "notifications",
         "Логировать действия администраторов",
         "Если выключено, события от пользователей из ADMIN_IDS не записываются в message logs.",
-        i18n_label_key="settings_field_log_admin_actions_label",
-        i18n_description_key="settings_field_log_admin_actions_description",
+        i18n_label_key="admin_settings_field_log_admin_actions_label",
+        i18n_description_key="admin_settings_field_log_admin_actions_description",
     ),
     SettingField(
         "LOG_LEVEL",
@@ -260,6 +261,73 @@ SETTINGS_MANIFEST: List[SettingField] = [
     ),
     SettingField("LOG_CHAT_ID", "int", "notifications", "ID чата для логов"),
     SettingField("LOG_THREAD_ID", "int", "notifications", "ID треда (для супергрупп)"),
+    SettingField(
+        "LOG_SUPPORT_THREAD_ID",
+        "int",
+        "notifications",
+        "ID треда поддержки",
+        "Тред лог-чата для уведомлений о тикетах поддержки.",
+    ),
+    SettingField(
+        "SUPPORT_TICKETS_ENABLED",
+        "bool",
+        "support",
+        "Тикеты поддержки включены",
+        "Показывает раздел поддержки в ЛК и включает создание тикетов.",
+    ),
+    SettingField(
+        "SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED",
+        "bool",
+        "support",
+        "Email-уведомления админам",
+        (
+            "Если выключено, новые тикеты и ответы пользователей останутся "
+            "только в Telegram и лог-чате."
+        ),
+    ),
+    SettingField(
+        "SUPPORT_ADMIN_NOTIFICATION_COOLDOWN_SECONDS",
+        "int",
+        "support",
+        "Пауза Telegram-уведомлений",
+        (
+            "Минимум секунд между повторными Telegram/log уведомлениями "
+            "по одному непрочитанному тикету."
+        ),
+        min=0,
+    ),
+    SettingField(
+        "SUPPORT_ADMIN_EMAIL_COOLDOWN_SECONDS",
+        "int",
+        "support",
+        "Пауза email-уведомлений",
+        "Минимум секунд между повторными email-уведомлениями по одному непрочитанному тикету.",
+        min=0,
+    ),
+    SettingField(
+        "SUPPORT_TICKET_MAX_BODY_LENGTH",
+        "int",
+        "support",
+        "Макс. длина сообщения",
+        "Максимальное количество символов в сообщении тикета.",
+        min=1,
+    ),
+    SettingField(
+        "SUPPORT_TICKET_MAX_SUBJECT_LENGTH",
+        "int",
+        "support",
+        "Макс. длина темы",
+        "Максимальное количество символов в теме тикета.",
+        min=1,
+    ),
+    SettingField(
+        "SUPPORT_TICKET_RATE_LIMIT_PER_HOUR",
+        "int",
+        "support",
+        "Лимит тикетов в час",
+        "Сколько новых тикетов пользователь может создать за час. 0 — без лимита.",
+        min=0,
+    ),
     # ─── Devices ───────────────────────────────────────────────────
     SettingField("MY_DEVICES_SECTION_ENABLED", "bool", "devices", "Раздел «Мои устройства»"),
     SettingField(
@@ -371,12 +439,15 @@ def manifest_payload() -> List[dict]:
         "trial": 5,
         "referral": 6,
         "notifications": 7,
-        "devices": 8,
+        "support": 8,
+        "devices": 9,
     }
     items: List[dict] = []
     for field in aggregated_manifest():
-        auto_label_i18n_key = f"settings_field_{field.key.lower()}_label"
-        auto_description_i18n_key = f"settings_field_{field.key.lower()}_description"
+        auto_label_i18n_key = f"admin_settings_field_{field.key.lower()}_label"
+        auto_description_i18n_key = (
+            f"admin_settings_field_{field.key.lower()}_description"
+        )
 
         default_value: Optional[str] = None
         owner = find_manifest_owner(field.key)

backend/bot/app/web/subscription_webapp.py

@@ -14,6 +14,7 @@ from bot.app.web.webapp import (
     payloads as _payloads,
     routes as _routes,
     serializers as _serializers,
+    support as _support,
 )
 
 _MODULES = (
@@ -26,6 +27,7 @@ _MODULES = (
     _serializers,
     _billing,
     _devices,
+    _support,
     _routes,
     _application,
 )

backend/bot/app/web/templates/subscription_webapp.html

@@ -1,25 +1,60 @@
 <!doctype html>
 <html lang="ru">
+  <head>
+    <meta charset="utf-8" />
+    <meta
+      name="viewport"
+      content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no, viewport-fit=cover"
+    />
+    <meta name="robots" content="noindex, nofollow" />
+    <meta name="theme-color" content="#03070b" />
+    <link id="app-favicon" rel="icon" href="data:," sizes="any" />
+    <title>/minishop</title>
+    <link rel="stylesheet" href="/subscription_webapp.css" />
+    <style>
+      .app-boot-fallback {
+        min-height: 100dvh;
+        display: grid;
+        place-items: center;
+        padding: 24px;
+        background: #03070b;
+      }
 
-<head>
-  <meta charset="utf-8">
-  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no, viewport-fit=cover">
-  <meta name="robots" content="noindex, nofollow">
-  <meta name="theme-color" content="#03070b">
-  <link id="app-favicon" rel="icon" href="data:," sizes="any">
-  <title>/minishop</title>
-  <link rel="stylesheet" href="/subscription_webapp.css">
-</head>
+      .app-boot-fallback__spinner {
+        width: 28px;
+        height: 28px;
+        border: 2px solid rgba(242, 247, 244, 0.18);
+        border-top-color: #00fe7a;
+        border-radius: 999px;
+        animation: appBootSpin 0.8s linear infinite;
+      }
 
-<body>
-  <main id="app"></main>
+      @media (prefers-reduced-motion: reduce) {
+        .app-boot-fallback__spinner {
+          animation: none;
+        }
+      }
 
-  <!-- WEBAPP_I18N_SCRIPT -->
-  <!-- WEBAPP_CONFIG_SCRIPT -->
-  <!-- WEBAPP_JS_SCRIPT -->
-  <!-- WEBAPP_DEV_MOCK_START -->
-  <script src="/subscription_webapp.js" defer></script>
-  <!-- WEBAPP_DEV_MOCK_END -->
-</body>
+      @keyframes appBootSpin {
+        to {
+          transform: rotate(360deg);
+        }
+      }
+    </style>
+  </head>
 
+  <body>
+    <main id="app">
+      <div class="app-boot-fallback" role="status" aria-label="Загрузка">
+        <div class="app-boot-fallback__spinner" aria-hidden="true"></div>
+      </div>
+    </main>
+
+    <!-- WEBAPP_I18N_SCRIPT -->
+    <!-- WEBAPP_CONFIG_SCRIPT -->
+    <!-- WEBAPP_JS_SCRIPT -->
+    <!-- WEBAPP_DEV_MOCK_START -->
+    <script src="/subscription_webapp.js" defer></script>
+    <!-- WEBAPP_DEV_MOCK_END -->
+  </body>
 </html>

backend/bot/app/web/webapp/_runtime.py

@@ -51,7 +51,7 @@ from bot.utils.config_link import prepare_config_links
 from bot.utils.request_security import parse_ip_entries, request_client_ip
 from bot.utils.text_sanitizer import sanitize_display_name, sanitize_username
 from config.settings import Settings
-from db.dal import payment_dal, security_dal, subscription_dal, user_dal
+from db.dal import payment_dal, security_dal, subscription_dal, support_dal, user_dal
 from db.dal.user_dal import UserMergeConflictError
 from db.models import Payment, User, UserTelegramAvatar
 

backend/bot/app/web/webapp/application.py

@@ -44,6 +44,9 @@ def create_subscription_webapp_application(
         "subscription_service",
         "promo_code_service",
         "referral_service",
+        "support_service",
+        "notification_service",
+        "email_auth_service",
         "panel_service",
         *iter_service_keys(),
     ):

backend/bot/app/web/webapp/assets.py

@@ -15,7 +15,13 @@ async def health_route(request: web.Request) -> web.Response:
 
 
 async def css_asset_route(request: web.Request) -> web.Response:
-    return await _serve_template_asset(request, "subscription_webapp.css", "text/css")
+    asset_hash = request.match_info.get("asset_hash")
+    filename = f"subscription_webapp.{asset_hash}.css" if asset_hash else "subscription_webapp.css"
+    response = await _serve_template_asset(request, filename, "text/css")
+    response.headers["Cache-Control"] = (
+        "public, max-age=31536000, immutable" if asset_hash else "no-cache"
+    )
+    return response
 
 
 def _safe_theme_css_relative_path(raw_path: str) -> Optional[Path]:
@@ -121,9 +127,7 @@ async def theme_asset_route(request: web.Request) -> web.Response:
     query = getattr(request, "query", {})
     response = web.Response(body=body, content_type=content_type)
     response.headers["Cache-Control"] = (
-        "public, max-age=31536000, immutable"
-        if query.get("v")
-        else "public, max-age=3600"
+        "public, max-age=31536000, immutable" if query.get("v") else "public, max-age=3600"
     )
     return response
 
@@ -887,6 +891,41 @@ async def js_asset_route(request: web.Request) -> web.Response:
     return response
 
 
+WEBAPP_BOOTSTRAP_I18N_PREFIXES = ("wa_",)
+WEBAPP_BOOTSTRAP_I18N_KEYS = {"menu_support_button"}
+WEBAPP_I18N_SCOPES = {"webapp", "admin"}
+
+
+def _is_webapp_bootstrap_i18n_key(key: str) -> bool:
+    return key in WEBAPP_BOOTSTRAP_I18N_KEYS or key.startswith(WEBAPP_BOOTSTRAP_I18N_PREFIXES)
+
+
+def _normalize_i18n_scope(raw_scope: object) -> str:
+    scope = str(raw_scope or "webapp").strip().lower()
+    return scope if scope in WEBAPP_I18N_SCOPES else "webapp"
+
+
+def _filter_webapp_i18n_payload(locales_data: object, scope: str = "webapp") -> Dict[str, Any]:
+    if not isinstance(locales_data, dict):
+        return {}
+
+    normalized_scope = _normalize_i18n_scope(scope)
+    payload: Dict[str, Any] = {}
+    for lang, messages in locales_data.items():
+        if not isinstance(messages, dict):
+            continue
+        filtered: Dict[str, Any] = {}
+        for key, value in messages.items():
+            key_text = str(key)
+            is_bootstrap_key = _is_webapp_bootstrap_i18n_key(key_text)
+            if (normalized_scope == "webapp" and is_bootstrap_key) or (
+                normalized_scope == "admin" and not is_bootstrap_key
+            ):
+                filtered[key_text] = value
+        payload[str(lang)] = filtered
+    return payload
+
+
 def _build_webapp_bootstrap_payload(request: web.Request) -> Dict[str, Any]:
     settings: Settings = request.app["settings"]
     cached = _get_cached_webapp_settings(request)
@@ -897,6 +936,8 @@ def _build_webapp_bootstrap_payload(request: web.Request) -> Dict[str, Any]:
     if preview_theme is None or not preview_theme.enabled:
         preview_key = ""
     i18n_instance: Optional[object] = request.app.get("i18n")
+    i18n_scope = _normalize_i18n_scope(request.query.get("i18n_scope") or "webapp")
+    locales_data = getattr(i18n_instance, "locales_data", {}) if i18n_instance else {}
     return {
         "config": {
             "title": settings.WEBAPP_TITLE,
@@ -929,12 +970,29 @@ def _build_webapp_bootstrap_payload(request: web.Request) -> Dict[str, Any]:
             "appVersion": _resolve_app_version(),
             "appRepositoryUrl": APP_REPOSITORY_URL,
         },
-        "i18n": getattr(i18n_instance, "locales_data", {}) if i18n_instance else {},
+        "i18n": _filter_webapp_i18n_payload(locales_data, i18n_scope),
     }
 
 
 async def bootstrap_route(request: web.Request) -> web.Response:
-    return web.json_response({"ok": True, **_build_webapp_bootstrap_payload(request)})
+    response = web.json_response({"ok": True, **_build_webapp_bootstrap_payload(request)})
+    response.headers["Cache-Control"] = "no-cache"
+    return response
+
+
+async def i18n_route(request: web.Request) -> web.Response:
+    i18n_instance: Optional[object] = request.app.get("i18n")
+    scope = _normalize_i18n_scope(request.query.get("scope") or "webapp")
+    locales_data = getattr(i18n_instance, "locales_data", {}) if i18n_instance else {}
+    response = web.json_response(
+        {
+            "ok": True,
+            "scope": scope,
+            "i18n": _filter_webapp_i18n_payload(locales_data, scope),
+        }
+    )
+    response.headers["Cache-Control"] = "no-cache"
+    return response
 
 
 async def index_route(request: web.Request) -> web.Response:
@@ -950,6 +1008,11 @@ async def index_route(request: web.Request) -> web.Response:
     bootstrap = _build_webapp_bootstrap_payload(request)
     config = bootstrap["config"]
     html = _strip_marked_block(html, DEV_MOCK_START_MARKER, DEV_MOCK_END_MARKER)
+    html = html.replace(
+        'href="/subscription_webapp.css"',
+        f'href="/{_resolve_webapp_css_asset_name()}"',
+        1,
+    )
     initial_theme_markup = _initial_theme_head_markup(request, initial_theme, primary_color)
     if initial_theme_markup:
         html = html.replace("</head>", f"{initial_theme_markup}\n</head>", 1)
@@ -997,7 +1060,9 @@ async def index_route(request: web.Request) -> web.Response:
             ),
             1,
         )
-    return web.Response(text=html, content_type="text/html", charset="utf-8")
+    response = web.Response(text=html, content_type="text/html", charset="utf-8")
+    response.headers["Cache-Control"] = "no-cache"
+    return response
 
 
 async def _serve_template_asset(
@@ -1035,6 +1100,21 @@ def _resolve_webapp_js_asset_name() -> str:
     return "subscription_webapp.js"
 
 
+def _resolve_webapp_css_asset_name() -> str:
+    hashed_assets = []
+    for path in ASSET_DIR.glob("subscription_webapp.*.css"):
+        if not re.fullmatch(r"subscription_webapp\.[0-9a-f]{8}\.css", path.name):
+            continue
+        try:
+            hashed_assets.append((path.stat().st_mtime, path.name))
+        except OSError:
+            continue
+    if hashed_assets:
+        hashed_assets.sort(reverse=True)
+        return hashed_assets[0][1]
+    return "subscription_webapp.css"
+
+
 _INITIAL_THEME_TOKEN_CSS_MAP = {
     "accent": "--accent",
     "bg": "--bg",

backend/bot/app/web/webapp/payloads.py

@@ -1,6 +1,8 @@
 # ruff: noqa: F401,F403,F405,I001
 from ._runtime import *  # noqa: F403,F405
 
+from typing import Literal
+
 
 class WebAppEmailPayload(BaseModel):
     model_config = ConfigDict(extra="ignore")
@@ -69,3 +71,52 @@ class WebAppDeviceDisconnectPayload(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
     token: constr(min_length=8, max_length=128)
+
+
+SupportCategory = Literal["billing", "technical", "account", "other"]
+SupportPriority = Literal["low", "normal", "high", "urgent"]
+SupportStatus = Literal["open", "awaiting_user", "awaiting_admin", "resolved", "closed"]
+
+
+class CreateTicketPayload(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    subject: constr(min_length=1, max_length=160)
+    category: SupportCategory = "other"
+    priority: Literal["normal", "high"] = "normal"
+    body: constr(min_length=1, max_length=4000)
+
+    @field_validator("subject", "body")
+    @classmethod
+    def _strip_required_text(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("empty_text")
+        return stripped
+
+
+class TicketReplyPayload(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    body: constr(min_length=1, max_length=4000)
+
+    @field_validator("body")
+    @classmethod
+    def _strip_body(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped:
+            raise ValueError("empty_text")
+        return stripped
+
+
+class AdminTicketReplyPayload(TicketReplyPayload):
+    is_internal_note: bool = False
+
+
+class AdminTicketPatchPayload(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    status: Optional[SupportStatus] = None
+    priority: Optional[SupportPriority] = None
+    category: Optional[SupportCategory] = None
+    assigned_admin_id: Optional[int] = None

backend/bot/app/web/webapp/routes.py

@@ -9,15 +9,18 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
     app.router.add_get("/invite", index_route)
     app.router.add_get("/devices", index_route)
     app.router.add_get("/settings", index_route)
+    app.router.add_get("/support", index_route)
+    app.router.add_get("/support/{ticket_id:\\d+}", index_route)
     app.router.add_get("/admin", index_route)
     app.router.add_get(
         (
             "/admin/{section:stats|users|payments|promos|ads|broadcast|logs|tariffs|"
-            "appearance|settings}"
+            "appearance|settings|support}"
         ),
         index_route,
     )
     app.router.add_get("/admin/users/{user_id:-?[0-9]+}", index_route)
+    app.router.add_get("/admin/support/{ticket_id:\\d+}", index_route)
     app.router.add_get("/auth/telegram/start", telegram_oauth_start_route)
     app.router.add_get("/auth/telegram/callback", telegram_oauth_callback_route)
     app.router.add_get("/health", health_route)
@@ -34,6 +37,7 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
         r"/webapp-emoji/{codepoints:[0-9a-f_]+}/512.{ext:gif|webp}",
         webapp_animated_emoji_route,
     )
+    app.router.add_get("/subscription_webapp.{asset_hash:[0-9a-f]{8}}.css", css_asset_route)
     app.router.add_get("/subscription_webapp.css", css_asset_route)
     app.router.add_get(r"/webapp-theme-css/{path:.+}", theme_css_asset_route)
     app.router.add_get(r"/webapp-theme-assets/{path:.+}", theme_asset_route)
@@ -47,6 +51,7 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
     app.router.add_post("/api/auth/email/password", email_password_auth_route)
     app.router.add_post("/api/auth/logout", logout_route)
     app.router.add_get("/api/bootstrap", bootstrap_route)
+    app.router.add_get("/api/i18n", i18n_route)
     app.router.add_get("/api/me", me_route)
     app.router.add_get("/api/account/avatar", account_avatar_route)
     app.router.add_post("/api/account/language", account_language_route)
@@ -60,6 +65,12 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
     app.router.add_get("/api/devices", devices_route)
     app.router.add_post("/api/devices/disconnect", disconnect_device_route)
     app.router.add_get("/api/devices/topup-options", device_topup_options_route)
+    app.router.add_get("/api/support/tickets", support_tickets_route)
+    app.router.add_post("/api/support/tickets", support_create_ticket_route)
+    app.router.add_get("/api/support/tickets/{id:\\d+}", support_ticket_detail_route)
+    app.router.add_post("/api/support/tickets/{id:\\d+}/messages", support_ticket_reply_route)
+    app.router.add_post("/api/support/tickets/{id:\\d+}/read", support_ticket_read_route)
+    app.router.add_get("/api/support/unread", support_unread_route)
     app.router.add_get("/api/tariffs/topup-options", tariff_topup_options_route)
     app.router.add_get("/api/tariffs/change-options", tariff_change_options_route)
     app.router.add_post("/api/tariffs/change", tariff_change_route)

backend/bot/app/web/webapp/serializers.py

@@ -38,6 +38,11 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
             if referral_service
             else {"invited_count": 0, "purchased_count": 0}
         )
+        support_unread_count = (
+            await support_dal.count_user_unread(session, user_id)
+            if settings.SUPPORT_TICKETS_ENABLED
+            else 0
+        )
         local_sub = (
             await subscription_dal.get_active_subscription_by_user_id(
                 session,
@@ -106,8 +111,14 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
             settings.WEBAPP_PRIMARY_COLOR or "#00fe7a",
             enabled_only=True,
         ),
+        "support_unread_count": int(support_unread_count or 0),
         "settings": {
             "support_url": settings.SUPPORT_LINK,
+            "support_tickets_enabled": bool(settings.SUPPORT_TICKETS_ENABLED),
+            "support_ticket_max_body_length": int(settings.SUPPORT_TICKET_MAX_BODY_LENGTH or 4000),
+            "support_ticket_max_subject_length": int(
+                settings.SUPPORT_TICKET_MAX_SUBJECT_LENGTH or 160
+            ),
             "traffic_mode": bool(settings.traffic_sale_mode),
             "my_devices_enabled": bool(settings.MY_DEVICES_SECTION_ENABLED),
             "user_hwid_device_limit": (

backend/bot/app/web/webapp/support.py

@@ -0,0 +1,152 @@
+# ruff: noqa: F401,F403,F405,I001
+from ._runtime import *  # noqa: F403,F405
+
+from bot.services.support_service import TicketForbidden, TicketNotFound, TicketRateLimited
+from db.dal import support_dal, user_dal
+from db.models import SupportTicket, SupportTicketMessage
+
+
+def _support_ticket_payload(ticket: SupportTicket) -> Dict[str, Any]:
+    return {
+        "ticket_id": ticket.ticket_id,
+        "user_id": ticket.user_id,
+        "subject": ticket.subject,
+        "category": ticket.category,
+        "priority": ticket.priority,
+        "status": ticket.status,
+        "assigned_admin_id": ticket.assigned_admin_id,
+        "last_message_at": ticket.last_message_at.isoformat() if ticket.last_message_at else None,
+        "last_message_role": ticket.last_message_role,
+        "unread_user_count": int(ticket.unread_user_count or 0),
+        "unread_admin_count": int(ticket.unread_admin_count or 0),
+        "created_at": ticket.created_at.isoformat() if ticket.created_at else None,
+        "updated_at": ticket.updated_at.isoformat() if ticket.updated_at else None,
+        "closed_at": ticket.closed_at.isoformat() if ticket.closed_at else None,
+    }
+
+
+def _support_message_payload(message: SupportTicketMessage) -> Dict[str, Any]:
+    return {
+        "message_id": message.message_id,
+        "ticket_id": message.ticket_id,
+        "author_role": message.author_role,
+        "author_user_id": message.author_user_id,
+        "body": message.body,
+        "is_internal_note": bool(message.is_internal_note),
+        "created_at": message.created_at.isoformat() if message.created_at else None,
+        "read_by_user_at": message.read_by_user_at.isoformat() if message.read_by_user_at else None,
+        "read_by_admin_at": message.read_by_admin_at.isoformat()
+        if message.read_by_admin_at
+        else None,
+    }
+
+
+def _support_limit_offset(request: web.Request) -> tuple[int, int]:
+    limit = max(1, min(100, int(request.query.get("limit", 25) or 25)))
+    offset = max(0, int(request.query.get("offset", 0) or 0))
+    return limit, offset
+
+
+async def support_tickets_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    limit, offset = _support_limit_offset(request)
+    status_filter = request.query.get("status") or None
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        tickets = await support_dal.list_user_tickets(
+            session,
+            user_id,
+            limit=limit,
+            offset=offset,
+            status_filter=status_filter,
+        )
+        counts = await support_dal.user_ticket_counts(session, user_id)
+    return web.json_response(
+        {
+            "ok": True,
+            "tickets": [_support_ticket_payload(t) for t in tickets],
+            "counts": counts,
+        }
+    )
+
+
+async def support_create_ticket_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    payload, error = _validate_model_payload(CreateTicketPayload, await _read_json(request))
+    if error:
+        return error
+    service = request.app["support_service"]
+    try:
+        ticket = await service.create_ticket(
+            user_id,
+            payload.subject,
+            payload.category,
+            payload.priority,
+            payload.body,
+        )
+    except TicketForbidden:
+        return _json_error(403, "ticket_forbidden", "Support ticket action is forbidden")
+    except TicketRateLimited:
+        return _json_error(429, "ticket_rate_limited", "Too many support tickets")
+    return web.json_response({"ok": True, "ticket": _support_ticket_payload(ticket)})
+
+
+async def support_ticket_detail_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        ticket, messages = await support_dal.get_ticket(session, ticket_id, include_internal=False)
+        if not ticket or ticket.user_id != user_id:
+            return _json_error(404, "not_found", "Ticket not found")
+    return web.json_response(
+        {
+            "ok": True,
+            "ticket": _support_ticket_payload(ticket),
+            "messages": [_support_message_payload(m) for m in messages],
+        }
+    )
+
+
+async def support_ticket_reply_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    payload, error = _validate_model_payload(TicketReplyPayload, await _read_json(request))
+    if error:
+        return error
+    service = request.app["support_service"]
+    try:
+        ticket, message = await service.reply_as_user(user_id, ticket_id, payload.body)
+    except TicketForbidden:
+        return _json_error(403, "ticket_forbidden", "Support ticket action is forbidden")
+    except TicketNotFound:
+        return _json_error(404, "not_found", "Ticket not found")
+    return web.json_response(
+        {
+            "ok": True,
+            "ticket": _support_ticket_payload(ticket),
+            "message": _support_message_payload(message),
+        }
+    )
+
+
+async def support_ticket_read_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    ticket_id = int(request.match_info["id"])
+    service = request.app["support_service"]
+    try:
+        await service.mark_read_as_user(user_id, ticket_id)
+    except TicketNotFound:
+        return _json_error(404, "not_found", "Ticket not found")
+    return web.json_response({"ok": True})
+
+
+async def support_unread_route(request: web.Request) -> web.Response:
+    user_id = _require_user_id(request)
+    async_session_factory: sessionmaker = request.app["async_session_factory"]
+    async with async_session_factory() as session:
+        user = await user_dal.get_user_by_id(session, user_id)
+        if user and user.is_banned:
+            return _json_error(403, "ticket_forbidden", "Support ticket action is forbidden")
+        unread = await support_dal.count_user_unread(session, user_id)
+    return web.json_response({"ok": True, "unread": unread})

backend/bot/handlers/user/start.py

@@ -387,11 +387,12 @@ async def ensure_required_channel_subscription(
 )
 @router.message(CommandStart(magic=F.args.regexp(r"^promo_(\w+)$").as_("promo_match")))
 @router.message(CommandStart(magic=F.args.regexp(r"^admin_user_(\d+)$").as_("admin_user_match")))
+@router.message(CommandStart(magic=F.args.regexp(r"^ticket_(\d+)$").as_("ticket_match")))
 @router.message(CommandStart(magic=F.args.regexp(r"^page_ref$").as_("page_ref_match")))
 @router.message(
     CommandStart(
         magic=F.args.regexp(
-            r"^(?!ref_|promo_|admin_user_|page_ref$|webapp_auth_)([A-Za-z0-9_\-]{2,64})$"
+            r"^(?!ref_|promo_|admin_user_|ticket_|page_ref$|webapp_auth_)([A-Za-z0-9_\-]{2,64})$"
         ).as_("ad_param_match")
     )
 )
@@ -408,6 +409,7 @@ async def start_command_handler(
     page_ref_match: Optional[re.Match] = None,
     ad_param_match: Optional[re.Match] = None,
     admin_user_match: Optional[re.Match] = None,
+    ticket_match: Optional[re.Match] = None,
 ):
     await state.clear()
     current_lang = i18n_data.get("current_language", settings.DEFAULT_LANGUAGE)
@@ -465,6 +467,31 @@ async def start_command_handler(
             await message.answer(_("admin_user_card_error"))
             return
 
+    if ticket_match:
+        ticket_id = int(ticket_match.group(1))
+        base_url = (settings.SUBSCRIPTION_MINI_APP_URL or "").strip()
+        if base_url:
+            ticket_url = f"{base_url.rstrip('/')}/support/{ticket_id}"
+            keyboard = types.InlineKeyboardMarkup(
+                inline_keyboard=[
+                    [
+                        types.InlineKeyboardButton(
+                            text=i18n.gettext(current_lang, "wa_support_open_ticket")
+                            if i18n
+                            else "Открыть тикет",
+                            web_app=types.WebAppInfo(url=ticket_url),
+                        )
+                    ]
+                ]
+            )
+            await message.answer(
+                i18n.gettext(current_lang, "wa_support_open_ticket_hint")
+                if i18n
+                else "Откройте тикет в Mini App.",
+                reply_markup=keyboard,
+            )
+            return
+
     referred_by_user_id: Optional[int] = None
     promo_code_to_apply: Optional[str] = None
     should_open_referral_from_start = False

backend/bot/main_bot.py

@@ -196,6 +196,9 @@ async def on_shutdown_configured(dispatcher: Dispatcher):
         "promo_code_service",
         "subscription_service",
         "referral_service",
+        "support_service",
+        "notification_service",
+        "email_auth_service",
         *iter_service_keys(),
     ):
         await close_service(service_key)

backend/bot/services/email_templates.py

@@ -492,3 +492,155 @@ def render_subscription_expiring(
         footer_html=footer,
     )
     return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered)
+
+
+def _support_email(
+    settings: Settings,
+    i18n: Optional[JsonI18n],
+    language: Optional[str],
+    *,
+    subject: str,
+    heading: str,
+    intro: str,
+    rows: Sequence[Tuple[str, str]],
+    body_preview: str,
+    ticket_url: Optional[str],
+    cta_label: str,
+) -> EmailContent:
+    lang = _normalize_lang(language, settings)
+    brand = _brand_title(settings)
+    accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR)
+    safe_url = (ticket_url or "").strip()
+    footer = _t_html(_resolve_i18n(i18n), lang, "email_footer_auto", brand=brand)
+    preview_block = (
+        f'<div style="margin:0 0 16px 0;background:{_BG};border:1px solid {_BORDER};'
+        f'border-radius:14px;padding:14px 16px;font-size:14px;line-height:1.55;color:{_TEXT};'
+        f'white-space:pre-wrap;">{html.escape(body_preview or "")}</div>'
+    )
+    body_parts = [_info_rows_html(rows), preview_block]
+    if safe_url:
+        body_parts.append(_cta_button_html(label=cta_label, url=safe_url, accent=accent))
+    rendered = _layout(
+        settings=settings,
+        preheader=intro,
+        heading=heading,
+        intro_html=html.escape(intro),
+        body_html="".join(body_parts),
+        footer_html=footer,
+    )
+    text_lines = [intro, "", *[f"{label}: {value}" for label, value in rows], "", body_preview]
+    if safe_url:
+        text_lines.extend(["", safe_url])
+    return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered)
+
+
+def render_support_new_ticket_admin(
+    settings: Settings,
+    i18n: Optional[JsonI18n],
+    language: Optional[str],
+    *,
+    ticket_id: int,
+    user_display: str,
+    subject: str,
+    body_preview: str,
+    snapshot_rows: Sequence[Tuple[str, str]],
+    ticket_url: Optional[str],
+) -> EmailContent:
+    rows = [
+        ("Ticket", f"#{ticket_id}"),
+        ("User", user_display),
+        ("Subject", subject),
+        *snapshot_rows,
+    ]
+    return _support_email(
+        settings,
+        i18n,
+        language,
+        subject=f"New support ticket #{ticket_id}",
+        heading=f"New support ticket #{ticket_id}",
+        intro="A user opened a new support ticket.",
+        rows=rows,
+        body_preview=body_preview,
+        ticket_url=ticket_url,
+        cta_label="Open ticket",
+    )
+
+
+def render_support_user_reply_admin(
+    settings: Settings,
+    i18n: Optional[JsonI18n],
+    language: Optional[str],
+    *,
+    ticket_id: int,
+    user_display: str,
+    subject: str,
+    body_preview: str,
+    snapshot_rows: Sequence[Tuple[str, str]],
+    ticket_url: Optional[str],
+) -> EmailContent:
+    rows = [
+        ("Ticket", f"#{ticket_id}"),
+        ("User", user_display),
+        ("Subject", subject),
+        *snapshot_rows,
+    ]
+    return _support_email(
+        settings,
+        i18n,
+        language,
+        subject=f"New user reply in ticket #{ticket_id}",
+        heading=f"User replied in ticket #{ticket_id}",
+        intro="A user sent a new support message.",
+        rows=rows,
+        body_preview=body_preview,
+        ticket_url=ticket_url,
+        cta_label="Open ticket",
+    )
+
+
+def render_support_admin_reply_user(
+    settings: Settings,
+    i18n: Optional[JsonI18n],
+    language: Optional[str],
+    *,
+    ticket_id: int,
+    subject: str,
+    body_preview: str,
+    ticket_url: Optional[str],
+) -> EmailContent:
+    return _support_email(
+        settings,
+        i18n,
+        language,
+        subject=f"New reply for ticket #{ticket_id}",
+        heading=f"New reply for ticket #{ticket_id}",
+        intro="Support has replied to your ticket.",
+        rows=[("Ticket", f"#{ticket_id}"), ("Subject", subject)],
+        body_preview=body_preview,
+        ticket_url=ticket_url,
+        cta_label="Open in Mini App",
+    )
+
+
+def render_support_ticket_closed_user(
+    settings: Settings,
+    i18n: Optional[JsonI18n],
+    language: Optional[str],
+    *,
+    ticket_id: int,
+    subject: str,
+    body_preview: str = "",
+    ticket_url: Optional[str],
+) -> EmailContent:
+    return _support_email(
+        settings,
+        i18n,
+        language,
+        subject=f"Ticket #{ticket_id} was closed",
+        heading=f"Ticket #{ticket_id} was closed",
+        intro="Your support ticket has been closed.",
+        rows=[("Ticket", f"#{ticket_id}"), ("Subject", subject)],
+        body_preview=body_preview or "The ticket is closed.",
+        ticket_url=ticket_url,
+        cta_label="Open in Mini App",
+    )

backend/bot/services/notification_service.py

@@ -4,10 +4,18 @@ from typing import Any, Callable, Dict, Optional
 
 from aiogram import Bot
 from aiogram.exceptions import TelegramBadRequest
-from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, WebAppInfo
 from aiogram.utils.text_decorations import html_decoration as hd
 
 from bot.middlewares.i18n import JsonI18n
+from bot.services.email_auth_service import EmailAuthService
+from bot.services.email_templates import (
+    render_support_admin_reply_user,
+    render_support_new_ticket_admin,
+    render_support_ticket_closed_user,
+    render_support_user_reply_admin,
+)
+from bot.utils import MessageContent, send_message_via_queue
 from bot.utils.message_queue import get_queue_manager
 from bot.utils.telegram_markup import (
     is_profile_link_error,
@@ -18,15 +26,30 @@ from bot.utils.text_sanitizer import (
     username_for_display,
 )
 from config.settings import Settings
+from db.dal import app_settings_dal, user_dal
+
+SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_KEY = "SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED"
 
 
 class NotificationService:
     """Enhanced notification service for sending messages to admins and log channels"""
 
-    def __init__(self, bot: Bot, settings: Settings, i18n: Optional[JsonI18n] = None):
+    def __init__(
+        self,
+        bot: Bot,
+        settings: Settings,
+        i18n: Optional[JsonI18n] = None,
+        *,
+        session_factory=None,
+        email_auth_service: Optional[EmailAuthService] = None,
+        bot_username: Optional[str] = None,
+    ):
         self.bot = bot
         self.settings = settings
         self.i18n = i18n
+        self.session_factory = session_factory
+        self.email_auth_service = email_auth_service
+        self.bot_username = bot_username or ""
 
     @staticmethod
     def _format_user_display(
@@ -149,7 +172,11 @@ class NotificationService:
                 "Failed to queue notification to log channel %s.", self.settings.LOG_CHAT_ID
             )
 
-    async def _send_to_admins(self, message: str):
+    async def _send_to_admins(
+        self,
+        message: str,
+        reply_markup: Optional[InlineKeyboardMarkup] = None,
+    ):
         """Send message to all admin users using message queue"""
         if not self.settings.ADMIN_IDS:
             return
@@ -164,6 +191,7 @@ class NotificationService:
                         text=message,
                         parse_mode="HTML",
                         disable_web_page_preview=True,
+                        reply_markup=reply_markup,
                     )
                 except Exception:
                     logging.exception("Failed to send notification to admin %s.", admin_id)
@@ -172,11 +200,343 @@ class NotificationService:
         for admin_id in self.settings.ADMIN_IDS:
             try:
                 await queue_manager.send_message(
-                    chat_id=admin_id, text=message, parse_mode="HTML", disable_web_page_preview=True
+                    chat_id=admin_id,
+                    text=message,
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
+                    reply_markup=reply_markup,
                 )
             except Exception:
                 logging.exception("Failed to queue notification to admin %s.", admin_id)
 
+    def _support_webapp_url(self, path: str) -> Optional[str]:
+        base_url = str(getattr(self.settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip()
+        if not base_url:
+            return None
+        normalized_path = f"/{str(path or '').lstrip('/')}"
+        return f"{base_url.rstrip('/')}{normalized_path}"
+
+    def _support_ticket_url(self, ticket_id: int, *, admin: bool = True) -> str:
+        path = f"/admin/support/{ticket_id}" if admin else f"/support/{ticket_id}"
+        webapp_url = self._support_webapp_url(path)
+        if webapp_url:
+            return webapp_url
+        bot_username = self.bot_username.strip().lstrip("@")
+        if bot_username:
+            return f"https://t.me/{bot_username}?startapp=ticket_{ticket_id}"
+        return "https://t.me/"
+
+    def _support_mini_app_button(
+        self,
+        *,
+        text: str,
+        path: str,
+        fallback_url: str,
+    ) -> InlineKeyboardButton:
+        webapp_url = self._support_webapp_url(path)
+        if webapp_url:
+            return InlineKeyboardButton(text=text, web_app=WebAppInfo(url=webapp_url))
+        return InlineKeyboardButton(text=text, url=fallback_url)
+
+    def _support_text(self, language: Optional[str], key: str, fallback: str) -> str:
+        if not self.i18n:
+            return fallback
+        return self.i18n.gettext(language or self.settings.DEFAULT_LANGUAGE, key) or fallback
+
+    @staticmethod
+    def _support_preview(body: str, limit: int = 700) -> str:
+        text = (body or "").strip()
+        return text if len(text) <= limit else f"{text[: limit - 1]}…"
+
+    @staticmethod
+    def _support_user_display(user) -> str:
+        name = " ".join(
+            part for part in [user.first_name, getattr(user, "last_name", None)] if part
+        )
+        if user.username:
+            return f"{name or user.username} (@{user.username})"
+        return name or getattr(user, "email", None) or f"ID {user.user_id}"
+
+    @staticmethod
+    def _support_snapshot_rows(snapshot: Optional[dict]) -> list[tuple[str, str]]:
+        if not snapshot:
+            return []
+        rows = []
+        for key, label in (
+            ("tariff", "Tariff"),
+            ("end_date", "End date"),
+            ("remaining", "Remaining"),
+            ("panel_status", "Panel status"),
+        ):
+            value = snapshot.get(key)
+            if value:
+                rows.append((label, str(value)))
+        return rows
+
+    @staticmethod
+    def _coerce_bool_setting(value: Any, default: bool) -> bool:
+        if value is None:
+            return default
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, (int, float)):
+            return bool(value)
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in {"1", "true", "yes", "on"}:
+                return True
+            if normalized in {"0", "false", "no", "off"}:
+                return False
+        return bool(value)
+
+    async def support_admin_email_notifications_enabled(self) -> bool:
+        enabled = bool(
+            getattr(self.settings, SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_KEY, False)
+        )
+        if not self.session_factory:
+            return enabled
+        try:
+            async with self.session_factory() as session:
+                found, raw_value = await app_settings_dal.get_override_value(
+                    session,
+                    SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_KEY,
+                )
+        except Exception:
+            logging.exception("Failed to read support admin email notification override.")
+            return enabled
+        if not found:
+            return enabled
+        return self._coerce_bool_setting(raw_value, enabled)
+
+    def _support_keyboard(self, ticket, user, *, admin: bool = True) -> InlineKeyboardMarkup:
+        ticket_path = (
+            f"/admin/support/{ticket.ticket_id}" if admin else f"/support/{ticket.ticket_id}"
+        )
+        rows = [
+            [
+                self._support_mini_app_button(
+                    text="Открыть тикет",
+                    path=ticket_path,
+                    fallback_url=self._support_ticket_url(ticket.ticket_id, admin=admin),
+                )
+            ]
+        ]
+        if admin:
+            profile_row = []
+            if getattr(user, "user_id", 0) and int(user.user_id) > 0:
+                profile_row.append(
+                    InlineKeyboardButton(text="Профиль", url=f"tg://user?id={user.user_id}")
+                )
+            user_card_path = f"/admin/users/{user.user_id}"
+            if self._support_webapp_url(user_card_path):
+                profile_row.append(
+                    self._support_mini_app_button(
+                        text="Карточка пользователя",
+                        path=user_card_path,
+                        fallback_url=self._support_ticket_url(ticket.ticket_id, admin=True),
+                    )
+                )
+            if profile_row:
+                rows.append(profile_row)
+        return InlineKeyboardMarkup(inline_keyboard=rows)
+
+    def _support_user_keyboard(self, ticket, user) -> InlineKeyboardMarkup:
+        button_text = self._support_text(
+            getattr(user, "language_code", None),
+            "wa_support_open_ticket",
+            "Открыть тикет",
+        )
+        webapp_url = self._support_webapp_url(f"/support/{ticket.ticket_id}")
+        if webapp_url:
+            button = InlineKeyboardButton(text=button_text, web_app=WebAppInfo(url=webapp_url))
+        else:
+            button = InlineKeyboardButton(
+                text=button_text,
+                url=self._support_ticket_url(ticket.ticket_id, admin=False),
+            )
+        return InlineKeyboardMarkup(inline_keyboard=[[button]])
+
+    async def _admin_email_users(self):
+        if not self.session_factory:
+            return []
+        async with self.session_factory() as session:
+            users = []
+            for admin_id in self.settings.ADMIN_IDS:
+                user = await user_dal.get_user_by_id(session, int(admin_id))
+                if user and user.email:
+                    users.append(user)
+            return users
+
+    async def _send_admin_support_email(self, renderer, **kwargs) -> None:
+        if not await self.support_admin_email_notifications_enabled():
+            return
+        if not self.email_auth_service:
+            return
+        for admin in await self._admin_email_users():
+            try:
+                content = renderer(
+                    self.settings,
+                    self.i18n,
+                    getattr(admin, "language_code", None) or self.settings.DEFAULT_LANGUAGE,
+                    **kwargs,
+                )
+                await self.email_auth_service.send_rendered_email(
+                    email=admin.email,
+                    content=content,
+                )
+            except Exception:
+                logging.exception("Failed to send support email to admin %s.", admin.user_id)
+
+    async def notify_new_support_ticket(self, ticket, user, first_message: str, snapshot: dict):
+        if not getattr(self.settings, "LOG_SUPPORT", True):
+            return
+        priority_emoji = {"low": "🟢", "normal": "🟡", "high": "🟠", "urgent": "🔴"}.get(
+            ticket.priority,
+            "🟡",
+        )
+        preview = self._support_preview(first_message)
+        user_display = self._support_user_display(user)
+        message = (
+            f"🆘 <b>Новый тикет #{ticket.ticket_id}</b>\n"
+            f"{priority_emoji} <b>{hd.quote(ticket.priority)}</b> · {hd.quote(ticket.category)}\n\n"
+            f"<b>Пользователь</b>\n{hd.quote(user_display)}\nID: <code>{user.user_id}</code>\n\n"
+            f"<b>Подписка</b>\n"
+            f"{hd.quote(str(snapshot.get('tariff') or '—'))}, "
+            f"до {hd.quote(str(snapshot.get('end_date') or '—'))}, "
+            f"осталось {hd.quote(str(snapshot.get('remaining') or '—'))}\n"
+            f"статус: {hd.quote(str(snapshot.get('panel_status') or '—'))}\n\n"
+            f"<b>Текст обращения</b>\n{hd.quote(preview)}"
+        )
+        keyboard = self._support_keyboard(ticket, user, admin=True)
+        await self._send_to_admins(message, reply_markup=keyboard)
+        await self._send_to_log_channel(
+            message,
+            thread_id=getattr(self.settings, "LOG_SUPPORT_THREAD_ID", None),
+            reply_markup=keyboard,
+        )
+        await self._send_admin_support_email(
+            render_support_new_ticket_admin,
+            ticket_id=ticket.ticket_id,
+            user_display=user_display,
+            subject=ticket.subject,
+            body_preview=preview,
+            snapshot_rows=self._support_snapshot_rows(snapshot),
+            ticket_url=self._support_ticket_url(ticket.ticket_id, admin=True),
+        )
+
+    async def notify_support_user_reply(
+        self,
+        ticket,
+        message,
+        user,
+        snapshot: dict,
+        *,
+        unread_count: Optional[int] = None,
+        send_telegram: bool = True,
+        send_email: bool = True,
+    ):
+        if not getattr(self.settings, "LOG_SUPPORT", True):
+            return
+        preview = self._support_preview(message.body)
+        user_display = self._support_user_display(user)
+        unread_line = (
+            f"\n<b>Unread:</b> {int(unread_count)}"
+            if unread_count is not None and int(unread_count or 0) > 1
+            else ""
+        )
+        text = (
+            f"💬 <b>Ответ пользователя в тикете #{ticket.ticket_id}</b>\n"
+            f"{hd.quote(user_display)}{unread_line}\n\n{hd.quote(preview)}"
+        )
+        keyboard = self._support_keyboard(ticket, user, admin=True)
+        if send_telegram and getattr(self.settings, "LOG_SUPPORT", True):
+            await self._send_to_admins(text, reply_markup=keyboard)
+            await self._send_to_log_channel(
+                text,
+                thread_id=getattr(self.settings, "LOG_SUPPORT_THREAD_ID", None),
+                reply_markup=keyboard,
+            )
+        if send_email:
+            await self._send_admin_support_email(
+                render_support_user_reply_admin,
+                ticket_id=ticket.ticket_id,
+                user_display=user_display,
+                subject=ticket.subject,
+                body_preview=preview,
+                snapshot_rows=self._support_snapshot_rows(snapshot),
+                ticket_url=self._support_ticket_url(ticket.ticket_id, admin=True),
+            )
+
+    async def notify_support_admin_reply(self, ticket, message, user):
+        preview = self._support_preview(message.body, limit=500)
+        url = self._support_ticket_url(ticket.ticket_id, admin=False)
+        text = f"💬 <b>Новый ответ по тикету #{ticket.ticket_id}</b>\n\n{hd.quote(preview)}"
+        keyboard = self._support_user_keyboard(ticket, user)
+        if int(user.user_id) > 0:
+            queue_manager = get_queue_manager()
+            if queue_manager:
+                await send_message_via_queue(
+                    queue_manager,
+                    int(user.user_id),
+                    MessageContent(content_type="text", text=text),
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
+                    reply_markup=keyboard,
+                )
+            else:
+                await self.bot.send_message(
+                    chat_id=int(user.user_id),
+                    text=text,
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
+                    reply_markup=keyboard,
+                )
+        if self.email_auth_service and getattr(user, "email", None):
+            content = render_support_admin_reply_user(
+                self.settings,
+                self.i18n,
+                getattr(user, "language_code", None),
+                ticket_id=ticket.ticket_id,
+                subject=ticket.subject,
+                body_preview=preview,
+                ticket_url=url,
+            )
+            await self.email_auth_service.send_rendered_email(email=user.email, content=content)
+
+    async def notify_support_ticket_closed(self, ticket, user, closing_admin):
+        url = self._support_ticket_url(ticket.ticket_id, admin=False)
+        text = f"✅ <b>Тикет #{ticket.ticket_id} закрыт</b>\n\n{hd.quote(ticket.subject)}"
+        keyboard = self._support_user_keyboard(ticket, user)
+        if int(user.user_id) > 0:
+            queue_manager = get_queue_manager()
+            if queue_manager:
+                await send_message_via_queue(
+                    queue_manager,
+                    int(user.user_id),
+                    MessageContent(content_type="text", text=text),
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
+                    reply_markup=keyboard,
+                )
+            else:
+                await self.bot.send_message(
+                    chat_id=int(user.user_id),
+                    text=text,
+                    parse_mode="HTML",
+                    disable_web_page_preview=True,
+                    reply_markup=keyboard,
+                )
+        if self.email_auth_service and getattr(user, "email", None):
+            content = render_support_ticket_closed_user(
+                self.settings,
+                self.i18n,
+                getattr(user, "language_code", None),
+                ticket_id=ticket.ticket_id,
+                subject=ticket.subject,
+                ticket_url=url,
+            )
+            await self.email_auth_service.send_rendered_email(email=user.email, content=content)
+
     async def notify_new_user_registration(
         self,
         user_id: int,

backend/bot/services/support_service.py

@@ -0,0 +1,465 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from aiogram import Bot
+from sqlalchemy.orm import sessionmaker
+
+from bot.middlewares.i18n import JsonI18n
+from bot.services.email_auth_service import EmailAuthService
+from bot.services.notification_service import NotificationService
+from config.settings import Settings
+from db.dal import message_log_dal, subscription_dal, support_dal, user_dal
+from db.models import SupportTicket, SupportTicketMessage, User
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class AdminNotificationDecision:
+    send_telegram: bool
+    send_email: bool
+
+
+def _as_utc(value: Optional[datetime]) -> Optional[datetime]:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def _notification_due(
+    last_sent_at: Optional[datetime],
+    *,
+    now: datetime,
+    cooldown_seconds: int,
+) -> bool:
+    last_sent_at = _as_utc(last_sent_at)
+    if last_sent_at is None:
+        return True
+    cooldown = max(0, int(cooldown_seconds or 0))
+    if cooldown <= 0:
+        return True
+    return (now - last_sent_at).total_seconds() >= cooldown
+
+
+def _support_admin_notification_decision(
+    ticket: SupportTicket,
+    settings: Settings,
+    *,
+    now: Optional[datetime] = None,
+    admin_email_notifications_enabled: Optional[bool] = None,
+) -> AdminNotificationDecision:
+    now = _as_utc(now) or datetime.now(timezone.utc)
+    unread_count = max(0, int(getattr(ticket, "unread_admin_count", 0) or 0))
+    if unread_count <= 0:
+        return AdminNotificationDecision(send_telegram=False, send_email=False)
+
+    first_unread = unread_count <= 1
+    send_telegram = first_unread or _notification_due(
+        getattr(ticket, "admin_last_notified_at", None),
+        now=now,
+        cooldown_seconds=getattr(settings, "SUPPORT_ADMIN_NOTIFICATION_COOLDOWN_SECONDS", 300),
+    )
+    if admin_email_notifications_enabled is None:
+        admin_email_notifications_enabled = bool(
+            getattr(settings, "SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED", False)
+        )
+    send_email = bool(admin_email_notifications_enabled) and (
+        first_unread
+        or _notification_due(
+            getattr(ticket, "admin_last_emailed_at", None),
+            now=now,
+            cooldown_seconds=getattr(settings, "SUPPORT_ADMIN_EMAIL_COOLDOWN_SECONDS", 1800),
+        )
+    )
+    return AdminNotificationDecision(send_telegram=send_telegram, send_email=send_email)
+
+
+def _format_support_remaining(seconds: int, lang: str) -> str:
+    if seconds <= 0:
+        return "Subscription inactive" if lang == "en" else "Подписка не активна"
+    days, rem = divmod(seconds, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes = rem // 60
+    if lang == "en":
+        if days > 0:
+            return f"{days} d. {hours} h."
+        if hours > 0:
+            return f"{hours} h. {minutes} min."
+        return f"{max(1, minutes)} min."
+    if days > 0:
+        return f"{days} д. {hours} ч."
+    if hours > 0:
+        return f"{hours} ч. {minutes} мин."
+    return f"{max(1, minutes)} мин."
+
+
+class TicketForbidden(PermissionError):
+    pass
+
+
+class TicketRateLimited(RuntimeError):
+    pass
+
+
+class TicketNotFound(LookupError):
+    pass
+
+
+class SupportService:
+    def __init__(
+        self,
+        session_factory: sessionmaker,
+        settings: Settings,
+        bot: Bot,
+        i18n: Optional[JsonI18n],
+        notification_service: Optional[NotificationService] = None,
+        email_auth_service: Optional[EmailAuthService] = None,
+    ):
+        self.session_factory = session_factory
+        self.settings = settings
+        self.bot = bot
+        self.i18n = i18n
+        self.email_auth_service = email_auth_service or EmailAuthService(settings)
+        self.notification_service = notification_service or NotificationService(
+            bot,
+            settings,
+            i18n,
+            session_factory=session_factory,
+            email_auth_service=self.email_auth_service,
+        )
+
+    async def _ensure_user_allowed(self, session, user_id: int) -> User:
+        user = await user_dal.get_user_by_id(session, user_id)
+        if not user or user.is_banned or not self.settings.SUPPORT_TICKETS_ENABLED:
+            raise TicketForbidden("ticket_forbidden")
+        return user
+
+    async def create_ticket(
+        self,
+        user_id: int,
+        subject: str,
+        category: str,
+        priority: str,
+        first_message_body: str,
+    ) -> SupportTicket:
+        async with self.session_factory() as session:
+            user = await self._ensure_user_allowed(session, user_id)
+            limit = max(0, int(self.settings.SUPPORT_TICKET_RATE_LIMIT_PER_HOUR or 0))
+            if limit:
+                recent = await support_dal.count_recent_tickets_for_user(session, user_id, 3600)
+                if recent >= limit:
+                    raise TicketRateLimited("ticket_rate_limited")
+            ticket = await support_dal.create_ticket(
+                session,
+                user_id,
+                subject[: self.settings.SUPPORT_TICKET_MAX_SUBJECT_LENGTH],
+                category,
+                priority,
+                first_message_body[: self.settings.SUPPORT_TICKET_MAX_BODY_LENGTH],
+            )
+            snapshot = await self.build_user_snapshot(user, session=session)
+            await session.commit()
+
+        try:
+            await self.notification_service.notify_new_support_ticket(
+                ticket,
+                user,
+                first_message_body,
+                snapshot,
+            )
+        except Exception:
+            logger.exception("Failed to notify about support ticket %s", ticket.ticket_id)
+        return ticket
+
+    async def reply_as_user(
+        self,
+        user_id: int,
+        ticket_id: int,
+        body: str,
+    ) -> tuple[SupportTicket, SupportTicketMessage]:
+        async with self.session_factory() as session:
+            user = await self._ensure_user_allowed(session, user_id)
+            ticket, _messages = await support_dal.get_ticket(session, ticket_id)
+            if not ticket or ticket.user_id != user_id:
+                raise TicketNotFound("not_found")
+            message = await support_dal.add_message(
+                session,
+                ticket_id,
+                "user",
+                user_id,
+                body[: self.settings.SUPPORT_TICKET_MAX_BODY_LENGTH],
+            )
+            if message is None:
+                raise TicketNotFound("not_found")
+            await session.refresh(ticket)
+            notification_at = datetime.now(timezone.utc)
+            admin_email_notifications_enabled = (
+                await self.notification_service.support_admin_email_notifications_enabled()
+            )
+            notification_decision = _support_admin_notification_decision(
+                ticket,
+                self.settings,
+                now=notification_at,
+                admin_email_notifications_enabled=admin_email_notifications_enabled,
+            )
+            await support_dal.record_admin_notification(
+                session,
+                ticket_id,
+                notified_at=notification_at if notification_decision.send_telegram else None,
+                emailed_at=notification_at if notification_decision.send_email else None,
+            )
+            snapshot = await self.build_user_snapshot(user, session=session)
+            await session.commit()
+
+        if notification_decision.send_telegram or notification_decision.send_email:
+            try:
+                await self.notification_service.notify_support_user_reply(
+                    ticket,
+                    message,
+                    user,
+                    snapshot,
+                    unread_count=int(ticket.unread_admin_count or 0),
+                    send_telegram=notification_decision.send_telegram,
+                    send_email=notification_decision.send_email,
+                )
+            except Exception:
+                logger.exception("Failed to notify about support user reply %s", ticket_id)
+        return ticket, message
+
+    async def reply_as_admin(
+        self,
+        admin_id: int,
+        ticket_id: int,
+        body: str,
+        *,
+        is_internal_note: bool = False,
+    ) -> tuple[SupportTicket, SupportTicketMessage]:
+        async with self.session_factory() as session:
+            ticket, _messages = await support_dal.get_ticket(
+                session,
+                ticket_id,
+                include_internal=True,
+            )
+            if not ticket:
+                raise TicketNotFound("not_found")
+            user = await user_dal.get_user_by_id(session, ticket.user_id)
+            message = await support_dal.add_message(
+                session,
+                ticket_id,
+                "admin",
+                admin_id,
+                body[: self.settings.SUPPORT_TICKET_MAX_BODY_LENGTH],
+                is_internal_note=is_internal_note,
+            )
+            if message is None:
+                raise TicketNotFound("not_found")
+            await support_dal.mark_read(session, ticket_id, "admin")
+            if not is_internal_note:
+                await message_log_dal.create_message_log_no_commit(
+                    session,
+                    {
+                        "user_id": admin_id,
+                        "event_type": "support_admin_reply",
+                        "content": f"Support ticket #{ticket_id} reply",
+                        "is_admin_event": True,
+                        "target_user_id": ticket.user_id,
+                    },
+                )
+            await session.refresh(ticket)
+            await session.commit()
+
+        if user and not is_internal_note:
+            try:
+                await self.notification_service.notify_support_admin_reply(ticket, message, user)
+            except Exception:
+                logger.exception("Failed to notify user about support admin reply %s", ticket_id)
+        return ticket, message
+
+    async def change_status(self, admin_id: int, ticket_id: int, status: str) -> SupportTicket:
+        return await self._update_and_audit(admin_id, ticket_id, status=status)
+
+    async def change_priority(self, admin_id: int, ticket_id: int, priority: str) -> SupportTicket:
+        return await self._update_and_audit(admin_id, ticket_id, priority=priority)
+
+    async def change_category(self, admin_id: int, ticket_id: int, category: str) -> SupportTicket:
+        return await self._update_and_audit(admin_id, ticket_id, category=category)
+
+    async def assign_admin(
+        self,
+        admin_id: int,
+        ticket_id: int,
+        assigned_admin_id: Optional[int],
+    ) -> SupportTicket:
+        return await self._update_and_audit(
+            admin_id,
+            ticket_id,
+            assigned_admin_id=assigned_admin_id,
+        )
+
+    async def close_ticket(self, admin_id: int, ticket_id: int) -> SupportTicket:
+        ticket = await self._update_and_audit(
+            admin_id,
+            ticket_id,
+            status="closed",
+            closed_by_admin_id=admin_id,
+        )
+        async with self.session_factory() as session:
+            user = await user_dal.get_user_by_id(session, ticket.user_id)
+        if user:
+            try:
+                await self.notification_service.notify_support_ticket_closed(ticket, user, admin_id)
+            except Exception:
+                logger.exception("Failed to notify user about support close %s", ticket_id)
+        return ticket
+
+    async def _update_and_audit(
+        self,
+        admin_id: int,
+        ticket_id: int,
+        **updates: Any,
+    ) -> SupportTicket:
+        async with self.session_factory() as session:
+            ticket = await support_dal.update_ticket(session, ticket_id, **updates)
+            if not ticket:
+                raise TicketNotFound("not_found")
+            await message_log_dal.create_message_log_no_commit(
+                session,
+                {
+                    "user_id": admin_id,
+                    "event_type": "support_ticket_update",
+                    "content": f"Support ticket #{ticket_id}: {updates}",
+                    "is_admin_event": True,
+                    "target_user_id": ticket.user_id,
+                },
+            )
+            await session.commit()
+            return ticket
+
+    async def mark_read_as_user(self, user_id: int, ticket_id: int) -> None:
+        async with self.session_factory() as session:
+            ticket, _messages = await support_dal.get_ticket(session, ticket_id)
+            if not ticket or ticket.user_id != user_id:
+                raise TicketNotFound("not_found")
+            await support_dal.mark_read(session, ticket_id, "user")
+            await session.commit()
+
+    async def mark_read_as_admin(self, ticket_id: int) -> None:
+        async with self.session_factory() as session:
+            await support_dal.mark_read(session, ticket_id, "admin")
+            await session.commit()
+
+    async def build_user_snapshot(self, user: User, *, session=None) -> dict:
+        owns_session = session is None
+        if owns_session:
+            session = self.session_factory()
+            await session.__aenter__()
+        try:
+            sub = None
+            if getattr(user, "panel_user_uuid", None):
+                sub = await subscription_dal.get_active_subscription_by_user_id(
+                    session,
+                    int(user.user_id),
+                    user.panel_user_uuid,
+                )
+            lang = (
+                getattr(user, "language_code", None) or self.settings.DEFAULT_LANGUAGE or "ru"
+            ).split("-")[0]
+            tariff_name = ""
+            if sub and getattr(sub, "tariff_key", None) and self.settings.tariffs_config:
+                try:
+                    tariff = self.settings.tariffs_config.require(str(sub.tariff_key))
+                    tariff_name = tariff.name(lang)
+                except Exception:
+                    tariff_name = str(sub.tariff_key or "")
+            end_date = getattr(sub, "end_date", None)
+            seconds_left = (
+                max(0, int((end_date - datetime.now(timezone.utc)).total_seconds()))
+                if end_date
+                else 0
+            )
+            return {
+                "user_id": int(user.user_id),
+                "name": " ".join(
+                    part for part in [user.first_name, getattr(user, "last_name", None)] if part
+                )
+                or user.username
+                or str(user.user_id),
+                "username": user.username,
+                "email": user.email,
+                "telegram_id": user.telegram_id,
+                "language": lang,
+                "registration_date": (
+                    user.registration_date.isoformat() if user.registration_date else None
+                ),
+                "email_login": bool(user.email and user.email_verified_at),
+                "subscription_active": bool(sub),
+                "panel_status": getattr(sub, "status_from_panel", None) if sub else None,
+                "end_date": end_date.isoformat() if end_date else None,
+                "remaining": _format_support_remaining(seconds_left, lang),
+                "tariff": tariff_name or (getattr(sub, "tariff_key", None) if sub else ""),
+                "traffic_regular": self._traffic_snapshot(
+                    getattr(sub, "traffic_used_bytes", 0) if sub else 0,
+                    self._regular_limit(sub),
+                ),
+                "traffic_premium": self._traffic_snapshot(
+                    getattr(sub, "premium_used_bytes", 0) if sub else 0,
+                    self._premium_limit(sub),
+                ),
+                "is_throttled": bool(getattr(sub, "is_throttled", False)) if sub else False,
+                "topup_balance_bytes": int(getattr(sub, "topup_balance_bytes", 0) or 0)
+                if sub
+                else 0,
+                "premium_topup_balance_bytes": int(
+                    getattr(sub, "premium_topup_balance_bytes", 0) or 0
+                )
+                if sub
+                else 0,
+                "lifetime_used_traffic_bytes": int(user.lifetime_used_traffic_bytes or 0),
+            }
+        finally:
+            if owns_session:
+                await session.__aexit__(None, None, None)
+
+    @staticmethod
+    def _regular_limit(sub) -> int:
+        if not sub:
+            return 0
+        if getattr(sub, "regular_unlimited_override", False):
+            return 0
+        return int(
+            (sub.traffic_limit_bytes or 0)
+            + (sub.topup_balance_bytes or 0)
+            + (getattr(sub, "regular_bonus_bytes", 0) or 0)
+        )
+
+    @staticmethod
+    def _premium_limit(sub) -> int:
+        if not sub:
+            return 0
+        if getattr(sub, "premium_unlimited_override", False):
+            return 0
+        return int(
+            (sub.premium_baseline_bytes or 0)
+            + (sub.premium_topup_balance_bytes or 0)
+            + (getattr(sub, "premium_topup_used_bytes", 0) or 0)
+            + (getattr(sub, "premium_bonus_bytes", 0) or 0)
+        )
+
+    @staticmethod
+    def _traffic_snapshot(used: Any, limit: Any) -> dict:
+        used_value = max(0, int(used or 0))
+        limit_value = max(0, int(limit or 0))
+        percent = round((used_value / limit_value) * 100, 2) if limit_value else 0
+        left = max(0, limit_value - used_value) if limit_value else 0
+        return {
+            "used_bytes": used_value,
+            "limit_bytes": limit_value,
+            "percent": percent,
+            "left_bytes": left,
+        }

backend/config/settings.py

@@ -336,6 +336,13 @@ class Settings(BaseSettings):
         description="Log updates/events triggered by users from ADMIN_IDS.",
     )
 
+    SUPPORT_TICKETS_ENABLED: bool = Field(default=True)
+    SUPPORT_TICKET_MAX_BODY_LENGTH: int = Field(default=4000)
+    SUPPORT_TICKET_MAX_SUBJECT_LENGTH: int = Field(default=160)
+    SUPPORT_TICKET_RATE_LIMIT_PER_HOUR: int = Field(default=5)
+    SUPPORT_ADMIN_EMAIL_NOTIFICATIONS_ENABLED: bool = Field(default=False)
+    SUPPORT_ADMIN_NOTIFICATION_COOLDOWN_SECONDS: int = Field(default=5 * 60)
+    SUPPORT_ADMIN_EMAIL_COOLDOWN_SECONDS: int = Field(default=30 * 60)
     SUBSCRIPTION_MINI_APP_URL: Optional[str] = Field(default=None)
 
     START_COMMAND_DESCRIPTION: Optional[str] = Field(default=None)
@@ -805,6 +812,9 @@ class Settings(BaseSettings):
     LOG_THREAD_ID: Optional[int] = Field(
         default=None, description="Thread ID for supergroup messages (optional)"
     )
+    LOG_SUPPORT_THREAD_ID: Optional[int] = Field(
+        default=None, description="Thread ID for support ticket log messages"
+    )
 
     @field_validator("LOG_LEVEL", mode="before")
     @classmethod
@@ -835,7 +845,7 @@ class Settings(BaseSettings):
             return v
         return secrets.token_urlsafe(32)
 
-    @field_validator("LOG_CHAT_ID", "LOG_THREAD_ID", mode="before")
+    @field_validator("LOG_CHAT_ID", "LOG_THREAD_ID", "LOG_SUPPORT_THREAD_ID", mode="before")
     @classmethod
     def validate_optional_int_fields(cls, v):
         """Convert empty strings to None for optional integer fields"""
@@ -890,6 +900,7 @@ class Settings(BaseSettings):
     LOG_SUSPICIOUS_ACTIVITY: bool = Field(
         default=True, description="Send notifications for suspicious promo attempts"
     )
+    LOG_SUPPORT: bool = Field(default=True, description="Send support ticket notifications")
 
     model_config = SettingsConfigDict(
         env_file=".env", env_file_encoding="utf-8", extra="ignore", populate_by_name=True

backend/db/dal/__init__.py

@@ -7,6 +7,7 @@ from . import (
     promo_code_dal,
     security_dal,
     subscription_dal,
+    support_dal,
     user_billing_dal,
     user_dal,
 )
@@ -22,4 +23,5 @@ __all__ = (
     "ad_dal",
     "security_dal",
     "app_settings_dal",
+    "support_dal",
 )

backend/db/dal/app_settings_dal.py

@@ -37,6 +37,17 @@ async def get_all_overrides(session: AsyncSession) -> Dict[str, Any]:
     return {row.key: _decode(row.value) for row in rows}
 
 
+async def get_override_value(session: AsyncSession, key: str) -> Tuple[bool, Any]:
+    row = (
+        await session.execute(
+            select(AppSettingOverride).where(AppSettingOverride.key == key).limit(1)
+        )
+    ).scalar_one_or_none()
+    if row is None:
+        return False, None
+    return True, _decode(row.value)
+
+
 async def get_overrides_with_meta(session: AsyncSession) -> List[Dict[str, Any]]:
     rows = (await session.execute(select(AppSettingOverride))).scalars().all()
     items: List[Dict[str, Any]] = []

backend/db/dal/support_dal.py

@@ -0,0 +1,375 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from sqlalchemy import and_, case, desc, func, or_, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy.orm import selectinload
+
+from ..models import SupportTicket, SupportTicketMessage, User
+
+ACTIVE_STATUSES = {"open", "awaiting_user", "awaiting_admin"}
+CLOSED_STATUSES = {"resolved", "closed"}
+_UNSET = object()
+
+
+def _status_condition(status: Optional[str]):
+    normalized = (status or "").strip().lower()
+    if not normalized or normalized in {"all", "any"}:
+        return None
+    if normalized == "active":
+        return SupportTicket.status.in_(ACTIVE_STATUSES)
+    if normalized == "closed":
+        return SupportTicket.status.in_(CLOSED_STATUSES)
+    return SupportTicket.status == normalized
+
+
+async def create_ticket(
+    session: AsyncSession,
+    user_id: int,
+    subject: str,
+    category: str,
+    priority: str,
+    first_message_body: str,
+) -> SupportTicket:
+    now = datetime.now(timezone.utc)
+    ticket = SupportTicket(
+        user_id=user_id,
+        subject=subject,
+        category=category,
+        priority=priority,
+        status="awaiting_admin",
+        last_message_at=now,
+        last_message_role="user",
+        unread_admin_count=1,
+        unread_user_count=0,
+        admin_last_notified_at=now,
+        admin_last_emailed_at=now,
+    )
+    session.add(ticket)
+    await session.flush()
+    message = SupportTicketMessage(
+        ticket_id=ticket.ticket_id,
+        author_role="user",
+        author_user_id=user_id,
+        body=first_message_body,
+        is_internal_note=False,
+        created_at=now,
+    )
+    session.add(message)
+    await session.flush()
+    await session.refresh(ticket)
+    return ticket
+
+
+async def add_message(
+    session: AsyncSession,
+    ticket_id: int,
+    author_role: str,
+    author_user_id: Optional[int],
+    body: str,
+    is_internal_note: bool = False,
+) -> Optional[SupportTicketMessage]:
+    stmt = select(SupportTicket).where(SupportTicket.ticket_id == ticket_id).with_for_update()
+    result = await session.execute(stmt)
+    ticket = result.scalar_one_or_none()
+    if not ticket:
+        return None
+
+    now = datetime.now(timezone.utc)
+    message = SupportTicketMessage(
+        ticket_id=ticket_id,
+        author_role=author_role,
+        author_user_id=author_user_id,
+        body=body,
+        is_internal_note=bool(is_internal_note),
+        created_at=now,
+    )
+    session.add(message)
+
+    ticket.last_message_at = now
+    ticket.last_message_role = author_role
+    ticket.updated_at = now
+    if author_role == "user":
+        ticket.unread_admin_count = int(ticket.unread_admin_count or 0) + 1
+        if ticket.status not in CLOSED_STATUSES:
+            ticket.status = "awaiting_admin"
+    elif author_role == "admin" and not is_internal_note:
+        ticket.unread_user_count = int(ticket.unread_user_count or 0) + 1
+        if ticket.status not in CLOSED_STATUSES:
+            ticket.status = "awaiting_user"
+
+    await session.flush()
+    await session.refresh(message)
+    return message
+
+
+async def record_admin_notification(
+    session: AsyncSession,
+    ticket_id: int,
+    *,
+    notified_at: Optional[datetime] = None,
+    emailed_at: Optional[datetime] = None,
+) -> None:
+    values = {}
+    if notified_at is not None:
+        values["admin_last_notified_at"] = notified_at
+    if emailed_at is not None:
+        values["admin_last_emailed_at"] = emailed_at
+    if not values:
+        return
+    await session.execute(
+        update(SupportTicket).where(SupportTicket.ticket_id == ticket_id).values(**values)
+    )
+    await session.flush()
+
+
+async def get_ticket(
+    session: AsyncSession,
+    ticket_id: int,
+    *,
+    include_internal: bool = False,
+) -> tuple[Optional[SupportTicket], list[SupportTicketMessage]]:
+    stmt = (
+        select(SupportTicket)
+        .where(SupportTicket.ticket_id == ticket_id)
+        .options(selectinload(SupportTicket.user))
+    )
+    result = await session.execute(stmt)
+    ticket = result.scalar_one_or_none()
+    if not ticket:
+        return None, []
+
+    msg_stmt = select(SupportTicketMessage).where(SupportTicketMessage.ticket_id == ticket_id)
+    if not include_internal:
+        msg_stmt = msg_stmt.where(SupportTicketMessage.is_internal_note.is_(False))
+    msg_stmt = msg_stmt.order_by(
+        SupportTicketMessage.created_at.asc(),
+        SupportTicketMessage.message_id.asc(),
+    )
+    msg_result = await session.execute(msg_stmt)
+    return ticket, list(msg_result.scalars().all())
+
+
+async def list_user_tickets(
+    session: AsyncSession,
+    user_id: int,
+    *,
+    limit: int,
+    offset: int,
+    status_filter: Optional[str] = None,
+) -> list[SupportTicket]:
+    stmt = select(SupportTicket).where(SupportTicket.user_id == user_id)
+    status_cond = _status_condition(status_filter)
+    if status_cond is not None:
+        stmt = stmt.where(status_cond)
+    stmt = (
+        stmt.order_by(desc(SupportTicket.last_message_at), desc(SupportTicket.ticket_id))
+        .limit(limit)
+        .offset(offset)
+    )
+    result = await session.execute(stmt)
+    return list(result.scalars().all())
+
+
+async def list_admin_tickets(
+    session: AsyncSession,
+    *,
+    status: Optional[str] = None,
+    priority: Optional[str] = None,
+    category: Optional[str] = None,
+    assigned_admin_id: Optional[int] = None,
+    search: Optional[str] = None,
+    sort: str = "updated_desc",
+    limit: int,
+    offset: int,
+) -> list[SupportTicket]:
+    stmt = select(SupportTicket).join(User, User.user_id == SupportTicket.user_id)
+    status_cond = _status_condition(status)
+    if status_cond is not None:
+        stmt = stmt.where(status_cond)
+    if priority:
+        stmt = stmt.where(SupportTicket.priority == priority)
+    if category:
+        stmt = stmt.where(SupportTicket.category == category)
+    if assigned_admin_id is not None:
+        stmt = stmt.where(SupportTicket.assigned_admin_id == assigned_admin_id)
+    if search:
+        pattern = f"%{search.strip().lower()}%"
+        stmt = stmt.where(
+            or_(
+                func.lower(SupportTicket.subject).like(pattern),
+                func.lower(User.username).like(pattern),
+                func.lower(User.first_name).like(pattern),
+                func.lower(User.email).like(pattern),
+            )
+        )
+    priority_rank = case(
+        (SupportTicket.priority == "urgent", 4),
+        (SupportTicket.priority == "high", 3),
+        (SupportTicket.priority == "normal", 2),
+        (SupportTicket.priority == "low", 1),
+        else_=0,
+    )
+    sort_key = (sort or "updated_desc").strip().lower()
+    sort_map = {
+        "updated_desc": (SupportTicket.last_message_at.desc().nullslast(),),
+        "updated_asc": (SupportTicket.last_message_at.asc().nullslast(),),
+        "created_desc": (SupportTicket.created_at.desc().nullslast(),),
+        "created_asc": (SupportTicket.created_at.asc().nullslast(),),
+        "importance_desc": (
+            priority_rank.desc(),
+            SupportTicket.last_message_at.desc().nullslast(),
+        ),
+        "importance_asc": (
+            priority_rank.asc(),
+            SupportTicket.last_message_at.desc().nullslast(),
+        ),
+    }
+    order_by = sort_map.get(sort_key, sort_map["updated_desc"])
+    stmt = stmt.options(selectinload(SupportTicket.user)).order_by(
+        *order_by,
+        desc(SupportTicket.ticket_id),
+    )
+    stmt = stmt.limit(limit).offset(offset)
+    result = await session.execute(stmt)
+    return list(result.scalars().unique().all())
+
+
+async def user_ticket_counts(session: AsyncSession, user_id: int) -> dict:
+    stmt = (
+        select(SupportTicket.status, func.count())
+        .where(SupportTicket.user_id == user_id)
+        .group_by(SupportTicket.status)
+    )
+    result = await session.execute(stmt)
+    by_status = {str(status): int(count or 0) for status, count in result.all()}
+    active = sum(by_status.get(status, 0) for status in ACTIVE_STATUSES)
+    closed = sum(by_status.get(status, 0) for status in CLOSED_STATUSES)
+    return {
+        **by_status,
+        "active": active,
+        "closed": closed,
+        "total": active + closed,
+    }
+
+
+async def mark_read(session: AsyncSession, ticket_id: int, role: str) -> None:
+    now = datetime.now(timezone.utc)
+    if role == "user":
+        await session.execute(
+            update(SupportTicket)
+            .where(SupportTicket.ticket_id == ticket_id)
+            .values(unread_user_count=0, updated_at=now)
+        )
+        await session.execute(
+            update(SupportTicketMessage)
+            .where(
+                and_(
+                    SupportTicketMessage.ticket_id == ticket_id,
+                    SupportTicketMessage.author_role == "admin",
+                    SupportTicketMessage.is_internal_note.is_(False),
+                    SupportTicketMessage.read_by_user_at.is_(None),
+                )
+            )
+            .values(read_by_user_at=now)
+        )
+    elif role == "admin":
+        await session.execute(
+            update(SupportTicket)
+            .where(SupportTicket.ticket_id == ticket_id)
+            .values(unread_admin_count=0, updated_at=now)
+        )
+        await session.execute(
+            update(SupportTicketMessage)
+            .where(
+                and_(
+                    SupportTicketMessage.ticket_id == ticket_id,
+                    SupportTicketMessage.author_role == "user",
+                    SupportTicketMessage.read_by_admin_at.is_(None),
+                )
+            )
+            .values(read_by_admin_at=now)
+        )
+    await session.flush()
+
+
+async def update_ticket(
+    session: AsyncSession,
+    ticket_id: int,
+    *,
+    status: Optional[str] = None,
+    priority: Optional[str] = None,
+    category: Optional[str] = None,
+    assigned_admin_id: object = _UNSET,
+    closed_by_admin_id: Optional[int] = None,
+) -> Optional[SupportTicket]:
+    ticket = await session.get(SupportTicket, ticket_id)
+    if not ticket:
+        return None
+    now = datetime.now(timezone.utc)
+    if status is not None:
+        ticket.status = status
+        if status == "closed":
+            ticket.closed_at = now
+            ticket.closed_by_admin_id = closed_by_admin_id
+        elif status != "closed":
+            ticket.closed_at = None
+            ticket.closed_by_admin_id = None
+    if priority is not None:
+        ticket.priority = priority
+    if category is not None:
+        ticket.category = category
+    if assigned_admin_id is not _UNSET:
+        ticket.assigned_admin_id = assigned_admin_id
+    ticket.updated_at = now
+    await session.flush()
+    await session.refresh(ticket)
+    return ticket
+
+
+async def count_user_unread(session: AsyncSession, user_id: int) -> int:
+    stmt = select(func.coalesce(func.sum(SupportTicket.unread_user_count), 0)).where(
+        SupportTicket.user_id == user_id
+    )
+    result = await session.execute(stmt)
+    return int(result.scalar_one() or 0)
+
+
+async def admin_stats(session: AsyncSession) -> dict:
+    status_result = await session.execute(
+        select(SupportTicket.status, func.count()).group_by(SupportTicket.status)
+    )
+    by_status = {str(status): int(count or 0) for status, count in status_result.all()}
+    unread_result = await session.execute(
+        select(func.coalesce(func.sum(SupportTicket.unread_admin_count), 0))
+    )
+    active = sum(by_status.get(status, 0) for status in ACTIVE_STATUSES)
+    closed = sum(by_status.get(status, 0) for status in CLOSED_STATUSES)
+    return {
+        **by_status,
+        "active": active,
+        "closed": closed,
+        "total": active + closed,
+        "open": by_status.get("open", 0),
+        "awaiting_admin": by_status.get("awaiting_admin", 0),
+        "awaiting_user": by_status.get("awaiting_user", 0),
+        "total_unread_admin": int(unread_result.scalar_one() or 0),
+    }
+
+
+async def count_recent_tickets_for_user(
+    session: AsyncSession,
+    user_id: int,
+    window_seconds: int,
+) -> int:
+    cutoff = datetime.now(timezone.utc) - timedelta(seconds=max(1, int(window_seconds)))
+    stmt = (
+        select(func.count())
+        .select_from(SupportTicket)
+        .where(SupportTicket.user_id == user_id, SupportTicket.created_at >= cutoff)
+    )
+    result = await session.execute(stmt)
+    return int(result.scalar_one() or 0)

backend/db/migrator.py

@@ -754,6 +754,126 @@ def _migration_0023_add_email_password_auth_fields(connection: Connection) -> No
         connection.execute(text("ALTER TABLE users ADD COLUMN password_set_at TIMESTAMPTZ"))
 
 
+def _migration_0024_add_support_tickets(connection: Connection) -> None:
+    connection.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS support_tickets (
+                ticket_id SERIAL PRIMARY KEY,
+                user_id BIGINT NOT NULL REFERENCES users(user_id),
+                subject VARCHAR(160) NOT NULL,
+                category VARCHAR(32) NOT NULL DEFAULT 'other',
+                priority VARCHAR(16) NOT NULL DEFAULT 'normal',
+                status VARCHAR(24) NOT NULL DEFAULT 'open',
+                assigned_admin_id BIGINT NULL,
+                last_message_at TIMESTAMPTZ DEFAULT NOW(),
+                last_message_role VARCHAR(16) NULL,
+                unread_user_count INTEGER NOT NULL DEFAULT 0,
+                unread_admin_count INTEGER NOT NULL DEFAULT 0,
+                admin_last_notified_at TIMESTAMPTZ NULL,
+                admin_last_emailed_at TIMESTAMPTZ NULL,
+                created_at TIMESTAMPTZ DEFAULT NOW(),
+                updated_at TIMESTAMPTZ NULL,
+                closed_at TIMESTAMPTZ NULL,
+                closed_by_admin_id BIGINT NULL
+            )
+            """
+        )
+    )
+    connection.execute(
+        text(
+            """
+            CREATE TABLE IF NOT EXISTS support_ticket_messages (
+                message_id SERIAL PRIMARY KEY,
+                ticket_id INTEGER NOT NULL REFERENCES support_tickets(ticket_id) ON DELETE CASCADE,
+                author_role VARCHAR(16) NOT NULL,
+                author_user_id BIGINT NULL REFERENCES users(user_id),
+                body TEXT NOT NULL,
+                is_internal_note BOOLEAN NOT NULL DEFAULT FALSE,
+                created_at TIMESTAMPTZ DEFAULT NOW(),
+                read_by_user_at TIMESTAMPTZ NULL,
+                read_by_admin_at TIMESTAMPTZ NULL
+            )
+            """
+        )
+    )
+
+    inspector = inspect(connection)
+    ticket_columns: Set[str] = {col["name"] for col in inspector.get_columns("support_tickets")}
+    ticket_column_sql = {
+        "user_id": "BIGINT NOT NULL REFERENCES users(user_id)",
+        "subject": "VARCHAR(160) NOT NULL DEFAULT ''",
+        "category": "VARCHAR(32) NOT NULL DEFAULT 'other'",
+        "priority": "VARCHAR(16) NOT NULL DEFAULT 'normal'",
+        "status": "VARCHAR(24) NOT NULL DEFAULT 'open'",
+        "assigned_admin_id": "BIGINT NULL",
+        "last_message_at": "TIMESTAMPTZ DEFAULT NOW()",
+        "last_message_role": "VARCHAR(16) NULL",
+        "unread_user_count": "INTEGER NOT NULL DEFAULT 0",
+        "unread_admin_count": "INTEGER NOT NULL DEFAULT 0",
+        "admin_last_notified_at": "TIMESTAMPTZ NULL",
+        "admin_last_emailed_at": "TIMESTAMPTZ NULL",
+        "created_at": "TIMESTAMPTZ DEFAULT NOW()",
+        "updated_at": "TIMESTAMPTZ NULL",
+        "closed_at": "TIMESTAMPTZ NULL",
+        "closed_by_admin_id": "BIGINT NULL",
+    }
+    for column, definition in ticket_column_sql.items():
+        if column not in ticket_columns:
+            connection.execute(
+                text(f"ALTER TABLE support_tickets ADD COLUMN {column} {definition}")
+            )
+
+    message_columns: Set[str] = {
+        col["name"] for col in inspector.get_columns("support_ticket_messages")
+    }
+    message_column_sql = {
+        "ticket_id": "INTEGER NOT NULL REFERENCES support_tickets(ticket_id) ON DELETE CASCADE",
+        "author_role": "VARCHAR(16) NOT NULL DEFAULT 'user'",
+        "author_user_id": "BIGINT NULL REFERENCES users(user_id)",
+        "body": "TEXT NOT NULL DEFAULT ''",
+        "is_internal_note": "BOOLEAN NOT NULL DEFAULT FALSE",
+        "created_at": "TIMESTAMPTZ DEFAULT NOW()",
+        "read_by_user_at": "TIMESTAMPTZ NULL",
+        "read_by_admin_at": "TIMESTAMPTZ NULL",
+    }
+    for column, definition in message_column_sql.items():
+        if column not in message_columns:
+            connection.execute(
+                text(f"ALTER TABLE support_ticket_messages ADD COLUMN {column} {definition}")
+            )
+
+    index_statements = [
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_user_id ON support_tickets (user_id)",
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_category ON support_tickets (category)",
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_priority ON support_tickets (priority)",
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_status ON support_tickets (status)",
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_assigned_admin_id ON support_tickets (assigned_admin_id)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_last_message_at ON support_tickets (last_message_at)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_tickets_status_last_msg ON support_tickets (status, last_message_at)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_ticket_messages_ticket_id ON support_ticket_messages (ticket_id)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_ticket_messages_author_user_id ON support_ticket_messages (author_user_id)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_ticket_messages_is_internal_note ON support_ticket_messages (is_internal_note)",  # noqa: E501
+        "CREATE INDEX IF NOT EXISTS ix_support_ticket_messages_created_at ON support_ticket_messages (created_at)",  # noqa: E501
+    ]
+    for stmt in index_statements:
+        connection.execute(text(stmt))
+
+
+def _migration_0025_add_support_notification_timestamps(connection: Connection) -> None:
+    inspector = inspect(connection)
+    ticket_columns: Set[str] = {col["name"] for col in inspector.get_columns("support_tickets")}
+    column_sql = {
+        "admin_last_notified_at": "TIMESTAMPTZ NULL",
+        "admin_last_emailed_at": "TIMESTAMPTZ NULL",
+    }
+    for column, definition in column_sql.items():
+        if column not in ticket_columns:
+            connection.execute(
+                text(f"ALTER TABLE support_tickets ADD COLUMN {column} {definition}")
+            )
+
+
 MIGRATIONS: List[Migration] = [
     Migration(
         id="0001_add_channel_subscription_fields",
@@ -881,6 +1001,16 @@ MIGRATIONS: List[Migration] = [
         description="Store hashed passwords for optional email password login",
         upgrade=_migration_0023_add_email_password_auth_fields,
     ),
+    Migration(
+        id="0024_add_support_tickets",
+        description="Add support ticket inbox and messages",
+        upgrade=_migration_0024_add_support_tickets,
+    ),
+    Migration(
+        id="0025_add_support_notification_timestamps",
+        description="Track support ticket admin notification cooldown timestamps",
+        upgrade=_migration_0025_add_support_notification_timestamps,
+    ),
 ]
 
 

backend/db/models.py

@@ -374,6 +374,62 @@ class MessageLog(Base):
     )
 
 
+class SupportTicket(Base):
+    __tablename__ = "support_tickets"
+
+    ticket_id = Column(Integer, primary_key=True, autoincrement=True)
+    user_id = Column(BigInteger, ForeignKey("users.user_id"), nullable=False, index=True)
+    subject = Column(String(160), nullable=False)
+    category = Column(String(32), nullable=False, default="other", index=True)
+    priority = Column(String(16), nullable=False, default="normal", index=True)
+    status = Column(String(24), nullable=False, default="open", index=True)
+    assigned_admin_id = Column(BigInteger, nullable=True, index=True)
+    last_message_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    last_message_role = Column(String(16), nullable=True)
+    unread_user_count = Column(Integer, nullable=False, default=0)
+    unread_admin_count = Column(Integer, nullable=False, default=0)
+    admin_last_notified_at = Column(DateTime(timezone=True), nullable=True)
+    admin_last_emailed_at = Column(DateTime(timezone=True), nullable=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+    closed_at = Column(DateTime(timezone=True), nullable=True)
+    closed_by_admin_id = Column(BigInteger, nullable=True)
+
+    user = relationship("User")
+    messages = relationship(
+        "SupportTicketMessage",
+        back_populates="ticket",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
+
+    __table_args__ = (
+        Index("ix_support_tickets_status_last_msg", "status", "last_message_at"),
+    )
+
+
+class SupportTicketMessage(Base):
+    __tablename__ = "support_ticket_messages"
+
+    message_id = Column(Integer, primary_key=True, autoincrement=True)
+    ticket_id = Column(
+        Integer,
+        ForeignKey("support_tickets.ticket_id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    author_role = Column(String(16), nullable=False)
+    author_user_id = Column(BigInteger, ForeignKey("users.user_id"), nullable=True, index=True)
+    body = Column(Text, nullable=False)
+    is_internal_note = Column(Boolean, nullable=False, default=False, index=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    read_by_user_at = Column(DateTime(timezone=True), nullable=True)
+    read_by_admin_at = Column(DateTime(timezone=True), nullable=True)
+
+    ticket = relationship("SupportTicket", back_populates="messages")
+    author_user = relationship("User")
+
+
 class PanelSyncStatus(Base):
     __tablename__ = "panel_sync_status"