Надёжный healthcheck ClickHouse и пример запроса в DEPLOY

- healthcheck ClickHouse переведён на wget по http://127.0.0.1:8123/ping без зависимости от переменных окружения
- wget подтверждён в образе clickhouse-server (используется его entrypoint)
- пример запроса к событиям в DEPLOY выполняется от пользователя контейнера, а не от хостовой оболочки
- версия 0.3.6

.env.example

@@ -28,3 +28,17 @@ ADMIN_IDS=
 
 # Часовой пояс для расписаний
 TIMEZONE=Europe/Moscow
+
+# Аналитика (ClickHouse). Логируются все действия бота (см. docs/events.md).
+# CLICKHOUSE_ENABLED=false — полностью выключить аналитику.
+CLICKHOUSE_ENABLED=true
+# Для локального запуска — localhost; в compose бот ходит по имени сервиса.
+CLICKHOUSE_HOST=localhost
+CLICKHOUSE_PORT=8123
+CLICKHOUSE_DB=analytics
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=
+# Батчинг записи событий: размер батча, интервал сброса (сек), предел буфера.
+ANALYTICS_BATCH_SIZE=200
+ANALYTICS_FLUSH_INTERVAL=5.0
+ANALYTICS_BUFFER_MAX=5000

DEPLOY.md

@@ -15,9 +15,13 @@ cd DispelEvilBot
 
 ```bash
 cp env/test.env.example env/test.env
-# заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.), ADMIN_IDS (опц.)
+# заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.),
+# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий)
 ```
 
+Порт ClickHouse на хосте (`CLICKHOUSE_HTTP_PORT`) у контуров должен различаться:
+`prod` — `8123`, `test` — `8124` (иначе контейнеры конфликтуют на одной VPS).
+
 ## 3. Логин в registry (если пакет приватный)
 
 ```bash
@@ -46,7 +50,62 @@ docker compose -p test --env-file env/test.env -f docker-compose.prod.yml logs -
 
 1. `postgres` поднимается и проходит healthcheck;
 2. `migrate` (**Flyway**) накатывает миграции из `sql/` — для пустой БД с нуля;
-3. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`).
+3. `clickhouse` поднимается и применяет `clickhouse/init/001_events.sql` —
+   **только при первой инициализации** пустого data-каталога;
+4. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же.
+
+Проверка контура (сервисы, миграции, Postgres, ClickHouse):
+
+```bash
+make check E=test
+```
+
+## ClickHouse (аналитика событий)
+
+События всех действий бота пишутся в `analytics.events` (контракт — [`docs/events.md`](docs/events.md)).
+ClickHouse наружу **не публикуется**: только loopback хоста (`127.0.0.1:${CLICKHOUSE_HTTP_PORT}`).
+
+```bash
+make ch E=test      # clickhouse-client контура
+# пример: события за сегодня по типам
+docker compose -p test --env-file env/test.env -f docker-compose.prod.yml exec -T clickhouse \
+  sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" \
+  --query "SELECT event_type, count() FROM analytics.events WHERE toDate(event_time) = today() GROUP BY event_type"'
+```
+
+Если ClickHouse недоступен, бот продолжает работать: события уходят в stdout (WARNING),
+полностью выключить можно флагом `CLICKHOUSE_ENABLED=false`.
+
+Изменения схемы: `clickhouse/init/*.sql` применяются лишь на пустом data-каталоге,
+поэтому на существующем контуре DDL нужно применить вручную (`make ch E=...`).
+
+## Бэкап событий на домашний сервер
+
+Сырые события живут на VPS ограниченное время: раз в сутки вчерашняя партиция
+выгружается в `Parquet.gz`, передаётся на домашний сервер, импортируется в тамошний
+ClickHouse и сверяется по количеству строк — **только после сверки** партиции старше
+`BACKUP_GRACE_DAYS` (по умолчанию 7) удаляются на VPS. На VPS остаются агрегаты
+(`events_daily`, `money_daily`).
+
+Разовая настройка:
+
+```bash
+cp env/backup.env.example env/backup.env   # BACKUP_REMOTE, BACKUP_DIR, BACKUP_SSH_KEY, ...
+# на домашнем сервере (в каталоге, куда кладутся архивы) разместить:
+#   scripts/clickhouse_import.sh
+#   clickhouse/init/001_events.sql  -> как 001_events.sql рядом со скриптом
+```
+
+Проверочный прогон (без cron): `make backup E=test`.
+
+Cron на VPS (вчерашний день уже закрыт, время московское):
+
+```cron
+0 4 * * * cd /opt/DispelEvilBot && ./scripts/clickhouse_backup.sh test >> /var/log/clickhouse-backup-test.log 2>&1
+```
+
+Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и пишет понятный лог.
+Импорт дня идемпотентен: повторный прогон не дублирует строки.
 
 ## Обновление и откат
 
@@ -67,7 +126,13 @@ docker compose -p <контур> --env-file env/<контур>.env -f docker-com
 
 ## Переменные окружения
 
-См. `env/prod.env.example` / `env/test.env.example`.
+См. `env/prod.env.example` / `env/test.env.example` (контуры) и `env/backup.env.example` (бэкап событий).
+
+## Дашборд (позже)
+
+Superset поднимается на домашнем сервере и читает **локальный** ClickHouse с полной
+историей событий (сюда ежедневно приезжают архивы). Отдельный read-only пользователь
+ClickHouse и сам дашборд — финальный этап, в этот контур не входят.
 
 ## Перенос данных на прод (позже, короткое окно простоя)
 

Makefile

@@ -9,7 +9,7 @@ COMPOSE := docker compose -p $(E) --env-file $(FILE) -f docker-compose.prod.yml
 .DEFAULT_GOAL := help
 .RECIPEPREFIX := >
 
-.PHONY: help up pull down restart logs ps images check psql
+.PHONY: help up pull down restart logs ps images check psql ch backup
 
 help:
 > @echo "Контур задаётся через E=test|prod (по умолчанию test):"
@@ -22,6 +22,8 @@ help:
 > @echo "  make images E=test   образы контура"
 > @echo "  make check E=test    проверка контура (скрипт)"
 > @echo "  make psql E=test     psql в БД контура"
+> @echo "  make ch E=test       clickhouse-client контура"
+> @echo "  make backup E=test   бэкап вчерашних событий на домашний сервер"
 
 up:
 > $(COMPOSE) up -d
@@ -49,3 +51,9 @@ check:
 
 psql:
 > $(COMPOSE) exec postgres sh -lc 'psql -U "$$POSTGRES_USER" -d "$$POSTGRES_DB"'
+
+ch:
+> $(COMPOSE) exec clickhouse sh -lc 'clickhouse-client --user "$$CLICKHOUSE_USER" --password "$$CLICKHOUSE_PASSWORD" --database analytics'
+
+backup:
+> ./scripts/clickhouse_backup.sh $(E)

README.md

@@ -6,9 +6,12 @@ Telegram-бот клана Dispel Evil (aiogram 3 + Postgres). Монорепо
 ## Структура
 
 ```
-common/     # общий пакет: config, logging, db, repositories, models, texts
+common/     # общий пакет: config, logging, db, repositories, models, texts, analytics
-bot/        # сервис бота: handlers, services, keyboards, notifications
+bot/        # сервис бота: handlers, services, keyboards, middlewares, notifications
 sql/        # миграции Flyway (V__*.sql, R__*.sql)
+clickhouse/ # init-скрипты ClickHouse (схема аналитики)
+docs/       # документация (docs/events.md — события аналитики)
+scripts/    # эксплуатационные скрипты (проверка контура, бэкап событий)
 images/     # статичные картинки (приветствие)
 deploy/     # материалы деплоя (Caddyfile, инструкции)
 tests/      # тесты
@@ -21,6 +24,7 @@ docker-compose.yml / docker-compose.prod.yml
 
 - Python 3.12
 - PostgreSQL 14+
+- ClickHouse 25.3 (аналитика событий; можно выключить — `CLICKHOUSE_ENABLED=false`)
 
 ## Настройка
 
@@ -44,9 +48,25 @@ APP_ENV=prod python -m bot    # prod
 Схема управляется миграциями **Flyway** в `sql/` (`V__*.sql` — versioned, `R__*.sql` — repeatable).
 При Docker-развёртывании накат выполняется автоматически сервисом `migrate` перед ботом.
 
+## Аналитика (ClickHouse)
+
+Все значимые действия бота пишутся в ClickHouse в единую таблицу `analytics.events`.
+Контракт событий — [`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
+
+- схема — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера);
+- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой ClickHouse не влияет на бота);
+- сырые события живут на VPS 7 дней и ежедневно уезжают архивом на домашний сервер,
+  на VPS остаются агрегаты (`events_daily`, `money_daily`) — см. [DEPLOY.md](DEPLOY.md);
+- выключается флагом `CLICKHOUSE_ENABLED=false`.
+
+## Уведомления в чат логов
+
+В `ALERT_CHAT_ID` (если задан) уходят: вход и выход участников, успешные очищения и
+ежечасный пинг «бот живой». Отправка не критична: ошибки только пишутся в лог.
+
 ## Docker
 
-Локальный запуск (postgres + migrate + bot):
+Локальный запуск (postgres + migrate + clickhouse + bot):
 
 ```bash
 cp .env.example .env      # заполнить
@@ -71,9 +91,20 @@ black --check .
 pytest
 ```
 
+## Бэкап событий
+
+Сырые события ClickHouse ежедневно выгружаются архивом на домашний сервер (по cron),
+на VPS остаются только агрегаты и последние дни сырых данных:
+
+```bash
+make backup E=test     # ./scripts/clickhouse_backup.sh test
+```
+
+Настройка и расписание — см. [`DEPLOY.md`](DEPLOY.md).
+
 ## Версионирование
 
-- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.2.5`.
+- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.3.6`.
 - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
 - Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
 - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.
@@ -90,14 +121,14 @@ pytest
 - Переход на «ветки + PR» и защиту ветки — когда появится второй разработчик.
 
 ```bash
-# обычная работа
+# обычная работа: сообщение без префиксов, по существу изменения
 git add -A
-git commit -m "feat: ..."
+git commit -m "Краткое описание изменения"
 git push origin master
 
 # выпуск релиза
 # 1) поднять __version__ в common/version.py
-git commit -am "chore: bump version to X.Y"
+git commit -am "Версия X.Y"
 git tag -a vX.Y -m "Release vX.Y"
 git push origin master --tags
 ```
@@ -109,4 +140,6 @@ git push origin master --tags
 
 ## Переменные окружения
 
-См. `.env.example`.
+- общие — `.env.example` (включая `CLICKHOUSE_*` и параметры батчинга `ANALYTICS_*`);
+- контуры — `env/prod.env.example`, `env/test.env.example`;
+- бэкап событий — `env/backup.env.example`.

bot/__main__.py

@@ -8,7 +8,9 @@ import logging
 from bot.bot_factory import create_bot, create_dispatcher
 from bot.services.heartbeat import run_heartbeat
 from bot.services.notifications import run_notifications
+from bot.services.runtime import uptime_seconds
 from common import __version__
+from common.analytics import EventType, close_analytics, init_analytics, record
 from common.config import get_settings
 from common.db.pool import close_pool, init_pool
 from common.logging_config import setup_logging
@@ -22,12 +24,14 @@ async def main() -> None:
     logger.info("Запуск DispelEvilBot v%s (APP_ENV=%s)", __version__, settings.app_env)
 
     await init_pool()
+    await init_analytics()
+    record(EventType.BOT_START, payload={"app_env": settings.app_env})
 
     bot = create_bot()
     dp = create_dispatcher()
 
     notifications_task = asyncio.create_task(run_notifications(bot))
-    heartbeat_task = asyncio.create_task(run_heartbeat())
+    heartbeat_task = asyncio.create_task(run_heartbeat(bot))
     try:
         await bot.delete_webhook(drop_pending_updates=True)
         await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
@@ -35,6 +39,8 @@ async def main() -> None:
         notifications_task.cancel()
         heartbeat_task.cancel()
         await asyncio.gather(notifications_task, heartbeat_task, return_exceptions=True)
+        record(EventType.BOT_STOP, payload={"app_env": settings.app_env, "uptime_seconds": uptime_seconds()})
+        await close_analytics()
         await close_pool()
         await bot.session.close()
 

bot/bot_factory.py

@@ -27,8 +27,11 @@ def create_dispatcher() -> Dispatcher:
     """
     from bot.handlers import commands, economy, member_events, shop
     from bot.handlers.errors import register_errors
+    from bot.middlewares import AnalyticsMiddleware
 
     dp = Dispatcher(storage=MemoryStorage())
+    # Внешний middleware: ловит все сообщения до фильтров (события `message`/`command`).
+    dp.message.outer_middleware(AnalyticsMiddleware())
     dp.include_router(member_events.router)
     dp.include_router(commands.router)
     dp.include_router(shop.router)

bot/handlers/economy.py

@@ -9,7 +9,7 @@ from aiogram import Router
 from aiogram.filters import Command
 from aiogram.types import Message
 
-from bot.services import access
+from bot.services import access, alerts
 from bot.services import economy as economy_service
 from common.config import get_settings
 from common.texts import economy, system
@@ -37,6 +37,12 @@ async def purification_handler(message: Message) -> None:
     else:
         context = {"money_add": result.money_add, "exp_add": result.exp_add}
         await message.answer(render(economy.MSG_PURIFICATION, context))
+        await alerts.notify_purification(
+            message.bot,
+            username=message.from_user.username,
+            user_id=message.from_user.id,
+            result=result,
+        )
 
 
 @router.message(Command("яички", ignore_case=True))

bot/handlers/errors.py

@@ -7,8 +7,13 @@ import logging
 from aiogram import Dispatcher
 from aiogram.types import ErrorEvent
 
+from common.analytics import EventType, record
+
 logger = logging.getLogger(__name__)
 
+#: Максимальная длина текста ошибки в payload события.
+ERROR_TEXT_MAX = 500
+
 
 def register_errors(dp: Dispatcher) -> None:
     """Зарегистрировать обработчик, чтобы одна ошибка не роняла поллинг."""
@@ -20,4 +25,15 @@ def register_errors(dp: Dispatcher) -> None:
             "Необработанная ошибка при обработке апдейта",
             exc_info=(type(exc), exc, exc.__traceback__),
         )
+        user = getattr(event.update, "event_from_user", None)
+        record(
+            EventType.ERROR,
+            user_id=getattr(user, "id", 0) or 0,
+            username=getattr(user, "username", None),
+            payload={
+                "error_type": type(exc).__name__,
+                "error": str(exc)[:ERROR_TEXT_MAX],
+                "update_type": str(getattr(event.update, "event_type", "") or ""),
+            },
+        )
         return True

bot/handlers/member_events.py

@@ -9,7 +9,9 @@ from aiogram.filters import JOIN_TRANSITION, LEAVE_TRANSITION, ChatMemberUpdated
 from aiogram.types import ChatMemberUpdated, FSInputFile
 from aiogram.utils.keyboard import InlineKeyboardBuilder
 
+from bot.services import alerts
 from bot.services import users as users_service
+from common.analytics import EventType, record
 from common.config import get_settings
 from common.texts import system, welcome
 from common.texts.render import render
@@ -39,6 +41,12 @@ async def welcome_handler(event: ChatMemberUpdated) -> None:
         return
 
     await users_service.register_from_event(event)
+    await alerts.notify_join(
+        event.bot,
+        username=user.username,
+        user_id=user.id,
+        chat_title=event.chat.title or "",
+    )
 
     builder = InlineKeyboardBuilder()
     builder.button(text=welcome.MSG_WELCOME_BUTTON_CAPTION, url=welcome.MSG_WELCOME_BUTTON_URL)
@@ -59,20 +67,23 @@ async def welcome_handler(event: ChatMemberUpdated) -> None:
 @router.chat_member(ChatMemberUpdatedFilter(member_status_changed=LEAVE_TRANSITION))
 async def goodbye_handler(event: ChatMemberUpdated) -> None:
     """Пользователь вышел из чата — удалить запись."""
-    settings = get_settings()
     user = event.from_user
     if user is None or user.is_bot:
         return
 
+    record(
+        EventType.LEAVE,
+        user_id=user.id,
+        username=user.username,
+        chat_id=event.chat.id,
+        payload={"chat_title": event.chat.title or ""},
+    )
     await users_service.delete_user(user.id)
     logger.info("Удалён пользователь %s после выхода из чата", user.id)
 
-    if not settings.alert_chat_id:
+    await alerts.notify_leave(
-        return
+        event.bot,
-    try:
+        username=user.username,
-        await event.bot.send_message(
+        user_id=user.id,
-            chat_id=settings.alert_chat_id,
+        chat_title=event.chat.title or "",
-            text=f"[goodbye] {user.username or user.id} покинул чат {event.chat.title}",
     )
-    except Exception:
-        logger.exception("Не удалось отправить уведомление о выходе пользователя")

bot/middlewares/__init__.py

@@ -0,0 +1,7 @@
+"""Middleware бота (транспортный слой)."""
+
+from __future__ import annotations
+
+from bot.middlewares.analytics import AnalyticsMiddleware
+
+__all__ = ["AnalyticsMiddleware"]

bot/middlewares/analytics.py

@@ -0,0 +1,66 @@
+"""Middleware аналитики: события ``message`` и ``command`` приходят с транспорта."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from aiogram import BaseMiddleware
+from aiogram.types import Message, TelegramObject
+
+from common.analytics import EventType, record, remember_user
+from common.analytics.messages import describe_command, describe_message
+from common.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+class AnalyticsMiddleware(BaseMiddleware):
+    """Логирует сообщения и команды основного чата.
+
+    Текст сообщения не сохраняется — только метаданные (см. ``docs/events.md``).
+    Аналитика не должна мешать обработке: любые сбои логируются и глушатся.
+    """
+
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        if isinstance(event, Message):
+            self._log_message(event)
+        return await handler(event, data)
+
+    @staticmethod
+    def _log_message(message: Message) -> None:
+        user = message.from_user
+        if user is None or user.is_bot:
+            return
+        if message.chat.id != get_settings().chat_id:
+            return
+
+        try:
+            remember_user(user.id, user.username)
+            record(
+                EventType.MESSAGE,
+                user_id=user.id,
+                username=user.username,
+                chat_id=message.chat.id,
+                payload=describe_message(message),
+            )
+            command = describe_command(message)
+            if command is not None:
+                record(
+                    EventType.COMMAND,
+                    user_id=user.id,
+                    username=user.username,
+                    chat_id=message.chat.id,
+                    payload=command,
+                )
+        except Exception:  # аналитика не должна ломать обработку сообщения
+            logger.exception("Не удалось записать события сообщения в аналитику")
+
+
+__all__ = ["AnalyticsMiddleware"]

bot/services/access.py

@@ -9,8 +9,12 @@ from aiogram import Bot
 from aiogram.enums import ChatMemberStatus
 from aiogram.types import Message
 
+from bot.services import alerts
 from bot.services import users as users_service
+from common.analytics import EventType, record, remember_user
 from common.config import get_settings
+from common.texts import alerts as alerts_text
+from common.texts.render import render
 
 logger = logging.getLogger(__name__)
 
@@ -48,8 +52,16 @@ async def check_user(message: Message) -> bool:
 
     if user.id in settings.admin_ids or await _is_member(message.bot, user.id):
         await users_service.register_from_message(message)
+        remember_user(user.id, user.username)
         return True
 
+    record(
+        EventType.ACCESS_DENIED,
+        user_id=user.id,
+        username=user.username,
+        chat_id=message.chat.id,
+        payload={"reason": "not_member"},
+    )
     await users_service.delete_user(user.id)
     logger.warning("Пользователь %s стучится в чат, доступ запрещён", user.id)
     await _notify_denied(message)
@@ -58,20 +70,8 @@ async def check_user(message: Message) -> bool:
 
 async def _notify_denied(message: Message) -> None:
     """Сообщить в чат логов (если задан) о попытке доступа извне."""
-    settings = get_settings()
+    user = message.from_user
-    if not settings.alert_chat_id:
+    if user is None:
         return
-
+    text = render(alerts_text.MSG_ACCESS_DENIED, {"user": alerts.user_tag(user.username, user.id)})
-    username = message.from_user.username if message.from_user else None
+    await alerts.notify_alert(message.bot, text, forward=message)
-    try:
-        await message.bot.send_message(
-            chat_id=settings.alert_chat_id,
-            text=f"[check_user]\nСтучится в чат {username}!",
-        )
-        await message.bot.forward_message(
-            chat_id=settings.alert_chat_id,
-            from_chat_id=message.chat.id,
-            message_id=message.message_id,
-        )
-    except Exception:  # уведомление не должно ломать обработку
-        logger.exception("Не удалось отправить уведомление о запрете доступа")

bot/services/alerts.py

@@ -0,0 +1,74 @@
+"""Уведомления в чат логов (``ALERT_CHAT_ID``).
+
+Единая точка отправки служебных сообщений бота: вход/выход участников,
+успешные очищения, пинг «бот живой». Уведомления **необязательные**:
+если чат не задан (``ALERT_CHAT_ID=0``) или отправка не удалась — это только
+запись в лог, обработка не прерывается.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from common.config import get_settings
+from common.texts import alerts as alerts_text
+from common.texts.render import render
+
+if TYPE_CHECKING:
+    from aiogram import Bot
+    from aiogram.types import Message
+
+    from common.models import PurificationResult
+
+logger = logging.getLogger(__name__)
+
+
+def user_tag(username: str | None, user_id: int) -> str:
+    """Представление пользователя для служебного сообщения (``@username`` или id)."""
+    return f"@{username}" if username else str(user_id)
+
+
+async def notify_alert(bot: Bot, text: str, *, forward: Message | None = None) -> bool:
+    """Отправить текст в чат логов (при необходимости — с пересылкой сообщения).
+
+    Возвращает ``False``, если чат логов не задан или отправка не удалась.
+    """
+    settings = get_settings()
+    if not settings.alert_chat_id:
+        return False
+
+    try:
+        await bot.send_message(chat_id=settings.alert_chat_id, text=text)
+        if forward is not None:
+            await bot.forward_message(
+                chat_id=settings.alert_chat_id,
+                from_chat_id=forward.chat.id,
+                message_id=forward.message_id,
+            )
+    except Exception:  # уведомление не должно ломать обработку
+        logger.exception("Не удалось отправить уведомление в чат логов")
+        return False
+    return True
+
+
+async def notify_join(bot: Bot, *, username: str | None, user_id: int, chat_title: str) -> bool:
+    """Уведомление о входе участника в чат."""
+    context = {"username": user_tag(username, user_id), "chat_title": chat_title}
+    return await notify_alert(bot, render(alerts_text.MSG_JOIN, context))
+
+
+async def notify_leave(bot: Bot, *, username: str | None, user_id: int, chat_title: str) -> bool:
+    """Уведомление о выходе участника из чата."""
+    context = {"username": user_tag(username, user_id), "chat_title": chat_title}
+    return await notify_alert(bot, render(alerts_text.MSG_LEAVE, context))
+
+
+async def notify_purification(bot: Bot, *, username: str | None, user_id: int, result: PurificationResult) -> bool:
+    """Уведомление об успешном очищении."""
+    context = {
+        "username": user_tag(username, user_id),
+        "exp_add": result.exp_add,
+        "money_add": result.money_add,
+    }
+    return await notify_alert(bot, render(alerts_text.MSG_PURIFICATION, context))

bot/services/economy.py

@@ -2,42 +2,84 @@
 
 from __future__ import annotations
 
+from common.analytics import EventType, record
 from common.models import EggsResult, PurificationResult
 from common.repositories import economy_repo
 
 #: id действия для начисления монеток.
 ACTION_COIN = 1
+#: id действия для награды за очищение.
+ACTION_PURIFICATION = 4
+#: id действия для обнуления баланса при удалении пользователя.
+ACTION_USER_DELETED = 5
 
 
 async def purify(id_user: int) -> PurificationResult:
-    """Выполнить ежедневное очищение."""
+    """Выполнить ежедневное очищение и записать события аналитики."""
     row = await economy_repo.purification(id_user)
     if not row:
         return PurificationResult(success=False)
-    return PurificationResult(
+
+    result = PurificationResult(
         success=int(row.get("res") or 0) != 0,
         exp_add=int(row.get("expr_add") or 0),
         money_add=int(row.get("mny_add") or 0),
         hours=int(row.get("h") or 0),
         minutes=int(row.get("m") or 0),
         seconds=int(row.get("s") or 0),
+        level_before=int(row.get("lvl_before") or 0),
+        level=int(row.get("lvl_after") or 0),
+        streak=int(row.get("streak") or 0),
     )
 
+    record(
+        EventType.PURIFICATION,
+        user_id=id_user,
+        payload={"success": result.success, "exp_add": result.exp_add, "money_add": result.money_add},
+    )
+    if not result.success:
+        return result
+
+    if result.exp_add:
+        record(EventType.EXP_ACCRUAL, user_id=id_user, payload={"amount": result.exp_add, "streak": result.streak})
+    if result.level_up:
+        record(
+            EventType.LEVEL_UP,
+            user_id=id_user,
+            payload={"level_from": result.level_before, "level_to": result.level},
+        )
+    if result.money_add:
+        record(
+            EventType.MONEY_ACCRUAL,
+            user_id=id_user,
+            payload={"amount": result.money_add, "reason": "purification", "action_id": ACTION_PURIFICATION},
+        )
+    return result
+
 
 async def collect_eggs(id_user: int) -> EggsResult:
     """Получить ежедневные яички."""
     row = await economy_repo.add_eggs(id_user)
     if not row:
         return EggsResult(success=False)
-    return EggsResult(
+
+    result = EggsResult(
         success=int(row.get("res") or 0) != 0,
         eggs=int(row.get("eggs_add") or 0),
         hours=int(row.get("h") or 0),
         minutes=int(row.get("m") or 0),
         seconds=int(row.get("s") or 0),
     )
+    if result.success:
+        record(EventType.EGGS, user_id=id_user, payload={"amount": result.eggs})
+    return result
 
 
 async def add_coins(id_user: int, amount: int) -> None:
-    """Начислить монеты."""
+    """Начислить монеты (монетка за активность)."""
     await economy_repo.money_action(id_user, amount, ACTION_COIN)
+    record(
+        EventType.MONEY_ACCRUAL,
+        user_id=id_user,
+        payload={"amount": amount, "reason": "coin", "action_id": ACTION_COIN},
+    )

bot/services/heartbeat.py

@@ -1,28 +1,72 @@
-"""Heartbeat для Docker healthcheck.
+"""Heartbeat для Docker healthcheck и ежечасный пинг «бот живой».
 
 Бот периодически обновляет mtime файла; healthcheck считает контейнер живым,
 если файл свежий. Если event loop завис — heartbeat перестаёт обновляться.
+
+Раз в час в чат логов (``ALERT_CHAT_ID``) уходит короткое сообщение о работе бота,
+и в ClickHouse пишется событие ``bot_alive``.
 """
 
 from __future__ import annotations
 
 import asyncio
 import logging
+from datetime import datetime
 from pathlib import Path
+from typing import TYPE_CHECKING
 
+from bot.services.alerts import notify_alert
+from bot.services.runtime import uptime_human, uptime_seconds
+from common import __version__
+from common.analytics import EventType, record
 from common.config import get_settings
+from common.texts import alerts as alerts_text
+from common.texts.render import render
+
+if TYPE_CHECKING:
+    from aiogram import Bot
 
 logger = logging.getLogger(__name__)
 
+#: Как часто обновлять heartbeat-файл.
 HEARTBEAT_INTERVAL = 30
 
+#: Час (в таймзоне бота), за который пинг уже отправлен.
+_last_ping_hour: datetime | None = None
 
-async def run_heartbeat() -> None:
+
-    """Обновлять файл-heartbeat каждые ``HEARTBEAT_INTERVAL`` секунд."""
+async def ping_hourly(bot: Bot, now: datetime | None = None) -> bool:
+    """Записать событие ``bot_alive`` и отправить пинг в чат логов раз в час.
+
+    Возвращает ``True``, если в этом часу пинг отправлен (или предпринята попытка).
+    Повторные вызовы в течение того же часа ничего не делают.
+    """
+    global _last_ping_hour
+
+    settings = get_settings()
+    current_hour = (now or datetime.now(settings.tzinfo)).replace(minute=0, second=0, microsecond=0)
+    if current_hour == _last_ping_hour:
+        return False
+
+    _last_ping_hour = current_hour
+    record(
+        EventType.BOT_ALIVE,
+        payload={"app_env": settings.app_env, "uptime_seconds": uptime_seconds()},
+    )
+    text = render(
+        alerts_text.MSG_ALIVE,
+        {"version": __version__, "uptime": uptime_human()},
+    )
+    return await notify_alert(bot, text)
+
+
+async def run_heartbeat(bot: Bot) -> None:
+    """Обновлять heartbeat-файл каждые ``HEARTBEAT_INTERVAL`` секунд и пинговать раз в час."""
     path = Path(get_settings().heartbeat_path)
     while True:
         try:
             path.touch()
         except OSError:
             logger.exception("Не удалось обновить heartbeat-файл %s", path)
+        await ping_hourly(bot)
         await asyncio.sleep(HEARTBEAT_INTERVAL)

bot/services/notifications.py

@@ -61,11 +61,6 @@ async def _send_broadcast(bot: Bot, template: str, subscribers: list[dict]) -> N
 
 
 async def _tick(bot: Bot, now: datetime) -> None:
-    settings = get_settings()
-    if now.minute == 0 and settings.alert_chat_id:
-        # Пинг «жив» в чат логов (если задан).
-        await bot.send_message(chat_id=settings.alert_chat_id, text="Я ЖЫФ!")
-
     for schedule in SCHEDULES:
         if schedule.matches(now):
             subscribers = await users_service.list_subscribers()

bot/services/runtime.py

@@ -0,0 +1,27 @@
+"""Общие runtime-данные процесса бота.
+
+Используется для событий аналитики (``bot_start``/``bot_stop``/``bot_alive``)
+и ежечасного пинга «бот живой» в чат логов.
+"""
+
+from __future__ import annotations
+
+import time
+
+#: Момент запуска процесса (монотонные секунды).
+STARTED_AT = time.monotonic()
+
+
+def uptime_seconds() -> int:
+    """Сколько секунд работает процесс бота."""
+    return int(time.monotonic() - STARTED_AT)
+
+
+def uptime_human() -> str:
+    """Аптайм в виде ``3ч 05м`` или ``12м`` (для уведомлений)."""
+    hours, remainder = divmod(uptime_seconds(), 3600)
+    minutes = remainder // 60
+    return f"{hours}ч {minutes:02d}м" if hours else f"{minutes}м"
+
+
+__all__ = ["STARTED_AT", "uptime_human", "uptime_seconds"]

bot/services/shop.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from common.analytics import EventType, record
 from common.models import Donat, Title
 from common.repositories import shop_repo
 
@@ -18,12 +19,48 @@ async def list_donats() -> list[Donat]:
 
 async def buy_title(id_user: int, id_title: int) -> bool:
     """Купить звание. Возвращает True при успехе."""
-    return await shop_repo.buy_title(id_user, id_title) == 1
+    price = await _title_price(id_title)
+    success = await shop_repo.buy_title(id_user, id_title) == 1
+
+    record(
+        EventType.PURCHASE_TITLE,
+        user_id=id_user,
+        payload={"title_id": id_title, "price": price, "success": success},
+    )
+    if success:
+        record(
+            EventType.MONEY_SPEND,
+            user_id=id_user,
+            payload={"amount": price, "reason": "title", "product_id": id_title},
+        )
+    return success
 
 
 async def buy_donat(id_user: int, id_donat: int) -> str | None:
     """Купить донат. Возвращает промокод или None при неуспехе."""
+    price = await _donat_price(id_donat)
     result = await shop_repo.buy_donat(id_user, id_donat)
-    if result == "-1":
+    promo = None if result == "-1" else result
-        return None
+
-    return result
+    record(
+        EventType.PURCHASE_DONAT,
+        user_id=id_user,
+        payload={"donat_id": id_donat, "price": price, "success": promo is not None},
+    )
+    if promo is not None:
+        record(
+            EventType.MONEY_SPEND,
+            user_id=id_user,
+            payload={"amount": price, "reason": "donat", "product_id": id_donat},
+        )
+    return promo
+
+
+async def _title_price(id_title: int) -> int:
+    """Цена звания для события покупки (0, если позиция недоступна)."""
+    return next((title.price for title in await list_titles() if title.id == id_title), 0)
+
+
+async def _donat_price(id_donat: int) -> int:
+    """Цена доната для события покупки (0, если позиция недоступна)."""
+    return next((donat.price for donat in await list_donats() if donat.id == id_donat), 0)

bot/services/subscriptions.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+from common.analytics import EventType, record
 from common.repositories import subscriptions_repo
 
 
 async def subscribe(id_user: int) -> None:
     await subscriptions_repo.subscribe(id_user)
+    record(EventType.SUBSCRIBE, user_id=id_user)
 
 
 async def unsubscribe(id_user: int) -> None:
     await subscriptions_repo.unsubscribe(id_user)
+    record(EventType.UNSUBSCRIBE, user_id=id_user)

bot/services/users.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 from aiogram.types import ChatMemberUpdated, Message, User
 
+from bot.services.economy import ACTION_USER_DELETED
+from common.analytics import EventType, record
 from common.models import NickEntry, UserInfo
-from common.repositories import users_repo
+from common.repositories import economy_repo, users_repo
 
 
 def _user_fields(user: User) -> tuple[int, str, str, str]:
@@ -31,14 +33,34 @@ async def register_from_event(event: ChatMemberUpdated) -> None:
     if user.is_bot:
         return
     await users_repo.add_user(*_user_fields(user))
+    record(
+        EventType.JOIN,
+        user_id=user.id,
+        username=user.username,
+        chat_id=event.chat.id,
+        payload={"chat_title": event.chat.title or ""},
+    )
 
 
 async def set_nick(id_user: int, nick: str) -> None:
     await users_repo.add_nick(id_user, nick)
+    record(EventType.NICK_SET, user_id=id_user, payload={"nick": nick})
 
 
 async def delete_user(id_user: int) -> None:
+    """Удалить пользователя и зафиксировать обнуление баланса.
+
+    Баланс читается **до** удаления: ``bot.del_user_by_id`` обнуляет его записью
+    в журнал ``money``.
+    """
+    balance = await economy_repo.get_money(id_user)
     await users_repo.del_user_by_id(id_user)
+    if balance:
+        record(
+            EventType.MONEY_SPEND,
+            user_id=id_user,
+            payload={"amount": abs(balance), "reason": "user_deleted", "action_id": ACTION_USER_DELETED},
+        )
 
 
 async def list_nicks() -> list[NickEntry]:

clickhouse/init/001_events.sql

@@ -0,0 +1,72 @@
+-- Схема аналитики ClickHouse.
+--
+-- Этот каталог монтируется в контейнер как /docker-entrypoint-initdb.d,
+-- поэтому скрипт применяется ТОЛЬКО при первой инициализации пустого data-каталога.
+-- Flyway у нас только для Postgres; изменения схемы здесь — новым файлом
+-- 00N_*.sql (и применением вручную на существующем контуре) плюс раздел в DEPLOY.md.
+
+CREATE DATABASE IF NOT EXISTS analytics;
+
+-- Единая широкая таблица событий (см. docs/events.md).
+-- Партиционирование по дню: сырые партиции удаляются после ежедневного бэкапа
+-- на домашний сервер (scripts/clickhouse_backup.sh), поэтому удаление должно
+-- быть дешёвым (DROP PARTITION, без mutation).
+CREATE TABLE IF NOT EXISTS analytics.events
+(
+    event_time  DateTime('UTC'),
+    event_type  LowCardinality(String),
+    user_id     Int64,
+    username    String,
+    chat_id     Int64,
+    payload     String,
+    bot_version String,
+    prod        UInt8
+)
+ENGINE = MergeTree
+PARTITION BY toYYYYMMDD(event_time)
+ORDER BY (event_time, event_type, user_id);
+
+-- Агрегат: события по дням и типам. Остаётся на VPS, когда сырые удалены.
+CREATE TABLE IF NOT EXISTS analytics.events_daily
+(
+    day        Date,
+    event_type LowCardinality(String),
+    user_id    Int64,
+    chat_id    Int64,
+    events     UInt64
+)
+ENGINE = SummingMergeTree
+ORDER BY (day, event_type, user_id, chat_id);
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.events_daily_mv TO analytics.events_daily AS
+SELECT
+    toDate(event_time) AS day,
+    event_type,
+    user_id,
+    chat_id,
+    count() AS events
+FROM analytics.events
+GROUP BY day, event_type, user_id, chat_id;
+
+-- Агрегат: денежные потоки по дням (сумма amount из payload событий экономики).
+CREATE TABLE IF NOT EXISTS analytics.money_daily
+(
+    day        Date,
+    event_type LowCardinality(String),
+    user_id    Int64,
+    amount_sum Int64,
+    events     UInt64
+)
+ENGINE = SummingMergeTree
+ORDER BY (day, event_type, user_id);
+
+CREATE MATERIALIZED VIEW IF NOT EXISTS analytics.money_daily_mv TO analytics.money_daily AS
+SELECT
+    toDate(event_time) AS day,
+    event_type,
+    user_id,
+    sum(JSONExtractInt(payload, 'amount')) AS amount_sum,
+    count() AS events
+FROM analytics.events
+WHERE event_type IN ('money_accrual', 'money_spend')
+GROUP BY day, event_type, user_id;

common/analytics/__init__.py

@@ -0,0 +1,33 @@
+"""Аналитика: клиент ClickHouse и каталог событий.
+
+Публичный интерфейс:
+
+- :func:`init_analytics` / :func:`close_analytics` — жизненный цикл (вызывает бот при старте/остановке);
+- :func:`record` — записать событие из сервисного слоя (неблокирующе);
+- :func:`remember_user` — запомнить ``username`` для обогащения событий;
+- :class:`EventType` / :data:`ALL_EVENT_TYPES` — каталог типов событий.
+"""
+
+from __future__ import annotations
+
+from common.analytics.client import (
+    AnalyticsClient,
+    close_analytics,
+    get_analytics,
+    init_analytics,
+    record,
+    remember_user,
+)
+from common.analytics.events import ALL_EVENT_TYPES, EVENT_COLUMNS, EventType
+
+__all__ = [
+    "ALL_EVENT_TYPES",
+    "EVENT_COLUMNS",
+    "AnalyticsClient",
+    "EventType",
+    "close_analytics",
+    "get_analytics",
+    "init_analytics",
+    "record",
+    "remember_user",
+]

common/analytics/client.py

@@ -0,0 +1,301 @@
+"""Клиент ClickHouse: батч-запись событий без блокировки бота.
+
+Свойства:
+- запись неблокирующая: :meth:`AnalyticsClient.record` только кладёт строку в буфер;
+- батчинг: отправка по размеру буфера и по таймеру;
+- недоступность ClickHouse **не роняет бота**: события попадают в stdout (fallback);
+- graceful-flush при остановке (:func:`close_analytics`).
+
+Импорт ``clickhouse_connect`` ленивый (внутри :meth:`AnalyticsClient._ensure_client`),
+поэтому пакет не является обязательным для импорта модуля (нужно для тестов/админки).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import logging
+from collections import deque
+from datetime import UTC, datetime
+from typing import Any, Protocol
+
+from common.analytics.events import EVENT_COLUMNS, EventType
+from common.config import Settings, get_settings
+from common.version import __version__
+
+logger = logging.getLogger(__name__)
+
+#: Ограничение длины JSON-payload (защита от «толстых» событий).
+PAYLOAD_MAX_LENGTH = 4000
+
+#: Ограничение кэша ``id_user -> username``.
+USERNAME_CACHE_MAX = 5000
+
+#: Сколько строк события выводить в лог при fallback.
+FALLBACK_SAMPLE = 5
+
+
+class ClickHouseClient(Protocol):
+    """Минимальный контракт клиента ClickHouse (используется и для подмены в тестах)."""
+
+    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> Any:
+        """Вставить строки в таблицу."""
+
+    async def close(self) -> None:
+        """Закрыть соединения."""
+
+
+class AnalyticsClient:
+    """Батч-запись событий в ClickHouse с деградацией в лог."""
+
+    def __init__(self, settings: Settings | None = None, *, client: ClickHouseClient | None = None) -> None:
+        self._settings = settings or get_settings()
+        self._client = client
+        self._rows: deque[list[Any]] = deque()
+        self._flush_now = asyncio.Event()
+        self._task: asyncio.Task[None] | None = None
+        self._usernames: dict[int, str] = {}
+        self._unavailable_logged = False
+        self._overflow = 0
+        self._table = f"{self._settings.clickhouse_db}.events"
+
+    @property
+    def enabled(self) -> bool:
+        """Включена ли аналитика (``CLICKHOUSE_ENABLED``)."""
+        return self._settings.clickhouse_enabled
+
+    @property
+    def pending(self) -> int:
+        """Сколько событий ждёт отправки (для тестов и диагностики)."""
+        return len(self._rows)
+
+    async def start(self) -> None:
+        """Подключиться к ClickHouse (best-effort) и запустить фоновый сброс буфера."""
+        if not self.enabled:
+            logger.info("Аналитика ClickHouse выключена (CLICKHOUSE_ENABLED=false)")
+            return
+
+        await self._ensure_client()
+        self._task = asyncio.create_task(self._run(), name="analytics-flush")
+        logger.info(
+            "Аналитика ClickHouse: %s:%s/%s (батч %s, интервал %sс)",
+            self._settings.clickhouse_host,
+            self._settings.clickhouse_port,
+            self._settings.clickhouse_db,
+            self._settings.analytics_batch_size,
+            self._settings.analytics_flush_interval,
+        )
+
+    async def _ensure_client(self) -> bool:
+        """Вернуть True, если клиент ClickHouse доступен."""
+        if self._client is not None:
+            return True
+
+        settings = self._settings
+        try:
+            import clickhouse_connect
+
+            self._client = await clickhouse_connect.get_async_client(
+                host=settings.clickhouse_host,
+                port=settings.clickhouse_port,
+                username=settings.clickhouse_user,
+                password=settings.clickhouse_password,
+                database=settings.clickhouse_db,
+            )
+        except Exception:
+            if not self._unavailable_logged:
+                self._unavailable_logged = True
+                logger.warning(
+                    "ClickHouse %s:%s недоступен: события пишутся в stdout (fallback)",
+                    settings.clickhouse_host,
+                    settings.clickhouse_port,
+                    exc_info=True,
+                )
+            return False
+
+        self._unavailable_logged = False
+        return True
+
+    def remember_user(self, user_id: int, username: str | None) -> None:
+        """Запомнить связку ``id_user -> username`` для обогащения будущих событий."""
+        if not user_id or not username:
+            return
+        if user_id not in self._usernames and len(self._usernames) >= USERNAME_CACHE_MAX:
+            self._usernames.pop(next(iter(self._usernames)))
+        self._usernames[user_id] = username
+
+    def record(
+        self,
+        event_type: EventType | str,
+        *,
+        user_id: int = 0,
+        username: str | None = None,
+        chat_id: int = 0,
+        payload: dict[str, Any] | None = None,
+    ) -> None:
+        """Положить событие в буфер (никогда не бросает исключений)."""
+        if not self.enabled:
+            return
+
+        if username is None:
+            username = self._usernames.get(user_id, "")
+
+        self._rows.append(
+            [
+                datetime.now(UTC).replace(tzinfo=None),
+                str(event_type),
+                int(user_id),
+                username or "",
+                int(chat_id),
+                self._dump_payload(payload),
+                __version__,
+                1 if self._settings.is_prod else 0,
+            ]
+        )
+
+        while len(self._rows) > self._settings.analytics_buffer_max:
+            self._rows.popleft()
+            self._overflow += 1
+
+        if len(self._rows) >= self._settings.analytics_batch_size:
+            self._flush_now.set()
+
+    async def _run(self) -> None:
+        """Периодически (или по достижении батча) сбрасывать буфер в ClickHouse."""
+        interval = self._settings.analytics_flush_interval
+        while True:
+            with contextlib.suppress(TimeoutError):
+                await asyncio.wait_for(self._flush_now.wait(), timeout=interval)
+            self._flush_now.clear()
+            await self.flush()
+
+    async def flush(self) -> None:
+        """Отправить накопленные события. При сбое — fallback в лог (без исключений)."""
+        overflow, self._overflow = self._overflow, 0
+        if overflow:
+            logger.warning("Буфер событий переполнен, потеряно %s событий (не записаны в ClickHouse)", overflow)
+
+        if not self._rows:
+            return
+
+        if not self.enabled:
+            self._rows.clear()
+            return
+
+        rows = list(self._rows)
+        self._rows.clear()
+
+        if not await self._ensure_client():
+            self._log_fallback(rows, "ClickHouse недоступен")
+            return
+
+        client = self._client
+        if client is None:
+            self._log_fallback(rows, "клиент ClickHouse не создан")
+            return
+
+        try:
+            await client.insert(self._table, rows, list(EVENT_COLUMNS))
+        except Exception:
+            logger.exception("Не удалось записать %s событий в ClickHouse", len(rows))
+            self._log_fallback(rows, "ошибка записи в ClickHouse")
+
+    async def close(self) -> None:
+        """Остановить фоновый таск, сбросить остаток и закрыть клиент."""
+        if self._task is not None:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._task
+            self._task = None
+
+        if self._client is not None:
+            await self.flush()
+            client, self._client = self._client, None
+            try:
+                await client.close()
+            except Exception:
+                logger.warning("Не удалось закрыть клиент ClickHouse", exc_info=True)
+        elif self._rows:
+            # Клиента нет (ClickHouse недоступен) — не пробуем подключиться на выходе.
+            rows = list(self._rows)
+            self._rows.clear()
+            self._log_fallback(rows, "остановка бота")
+
+    def _log_fallback(self, rows: list[list[Any]], reason: str) -> None:
+        """Вывести потерянные события в лог, чтобы данные не пропадали молча."""
+        if not rows:
+            return
+        lines = " | ".join(
+            f"{row[0]:%Y-%m-%d %H:%M:%S} {row[1]} user={row[2]} chat={row[4]} {row[5]}"
+            for row in rows[:FALLBACK_SAMPLE]
+        )
+        logger.warning(
+            "Fallback (%s): %s событий не записаны в ClickHouse: %s%s",
+            reason,
+            len(rows),
+            lines,
+            " ..." if len(rows) > FALLBACK_SAMPLE else "",
+        )
+
+    @staticmethod
+    def _dump_payload(payload: dict[str, Any] | None) -> str:
+        """Сериализовать payload в JSON-строку с обрезкой по длине."""
+        if not payload:
+            return "{}"
+        try:
+            text = json.dumps(payload, ensure_ascii=False, default=str)
+        except (TypeError, ValueError):
+            text = json.dumps({"_error": "payload не сериализуется"})
+        return text[:PAYLOAD_MAX_LENGTH]
+
+
+#: Активный клиент (если аналитика не инициализирована — None).
+_client: AnalyticsClient | None = None
+
+
+async def init_analytics(
+    settings: Settings | None = None,
+    *,
+    client: ClickHouseClient | None = None,
+) -> AnalyticsClient:
+    """Создать и запустить единственный клиент аналитики. Идемпотентно."""
+    global _client
+    if _client is not None:
+        return _client
+    _client = AnalyticsClient(settings, client=client)
+    await _client.start()
+    return _client
+
+
+async def close_analytics() -> None:
+    """Остановить клиент аналитики и сбросить буфер при остановке приложения."""
+    global _client
+    if _client is not None:
+        await _client.close()
+        _client = None
+
+
+def get_analytics() -> AnalyticsClient | None:
+    """Вернуть активный клиент аналитики (или None)."""
+    return _client
+
+
+def record(
+    event_type: EventType | str,
+    *,
+    user_id: int = 0,
+    username: str | None = None,
+    chat_id: int = 0,
+    payload: dict[str, Any] | None = None,
+) -> None:
+    """Записать событие. Если аналитика не инициализирована — no-op."""
+    if _client is None:
+        return
+    _client.record(event_type, user_id=user_id, username=username, chat_id=chat_id, payload=payload)
+
+
+def remember_user(user_id: int, username: str | None) -> None:
+    """Запомнить username пользователя (для обогащения событий из сервисов)."""
+    if _client is not None:
+        _client.remember_user(user_id, username)

common/analytics/events.py

@@ -0,0 +1,63 @@
+"""Каталог событий аналитики.
+
+Единственный источник правды по типам событий ClickHouse. Список синхронизирован
+с документацией (``docs/events.md``) — рассинхрон ловит ``tests/test_analytics.py``.
+
+Правило: добавляешь новое событие — добавь его в ``EventType`` и в ``docs/events.md``.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+
+class EventType(StrEnum):
+    """Типы событий (значение = колонка ``event_type`` в ClickHouse)."""
+
+    # --- Сообщения и команды ---
+    MESSAGE = "message"
+    COMMAND = "command"
+
+    # --- Экономика ---
+    PURIFICATION = "purification"
+    EXP_ACCRUAL = "exp_accrual"
+    LEVEL_UP = "level_up"
+    MONEY_ACCRUAL = "money_accrual"
+    MONEY_SPEND = "money_spend"
+    EGGS = "eggs"
+    PURCHASE_TITLE = "purchase_title"
+    PURCHASE_DONAT = "purchase_donat"
+
+    # --- Участники ---
+    JOIN = "join"
+    LEAVE = "leave"
+    NICK_SET = "nick_set"
+
+    # --- Подписки и доступ ---
+    SUBSCRIBE = "subscribe"
+    UNSUBSCRIBE = "unsubscribe"
+    ACCESS_DENIED = "access_denied"
+
+    # --- Жизненный цикл бота и ошибки ---
+    BOT_START = "bot_start"
+    BOT_STOP = "bot_stop"
+    BOT_ALIVE = "bot_alive"
+    ERROR = "error"
+
+
+#: Полный список типов событий (для проверки синхронности с документацией).
+ALL_EVENT_TYPES: frozenset[str] = frozenset(member.value for member in EventType)
+
+#: Колонки таблицы ``analytics.events`` в порядке вставки.
+EVENT_COLUMNS: tuple[str, ...] = (
+    "event_time",
+    "event_type",
+    "user_id",
+    "username",
+    "chat_id",
+    "payload",
+    "bot_version",
+    "prod",
+)
+
+__all__ = ["ALL_EVENT_TYPES", "EVENT_COLUMNS", "EventType"]

common/analytics/messages.py

@@ -0,0 +1,54 @@
+"""Метаданные сообщения для событий аналитики.
+
+Модуль намеренно **не зависит от aiogram**: работает с любым объектом, у которого
+есть нужные атрибуты (``Message`` в боте или заглушка в тестах). Текст сообщения
+не сохраняется — только метаданные (см. ``docs/events.md``).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+#: Максимальная длина имени команды в payload.
+COMMAND_MAX_LENGTH = 64
+
+#: Префикс, разделяющий команду и бота в групповом чате (``/команда@bot``).
+BOT_MENTION_SEPARATOR = "@"
+
+
+def _text(message: Any) -> str:
+    """Текст сообщения (или подпись) без лишних пробелов."""
+    text = getattr(message, "text", None) or getattr(message, "caption", None) or ""
+    return str(text).strip()
+
+
+def _enum_value(value: Any) -> str:
+    """Строковое значение enum-полей aiogram (``ChatType.PRIVATE`` -> ``private``)."""
+    return str(getattr(value, "value", value) or "")
+
+
+def describe_message(message: Any) -> dict[str, Any]:
+    """Метаданные сообщения для события ``message`` (без текста сообщения)."""
+    text = _text(message)
+    chat = getattr(message, "chat", None)
+    return {
+        "message_id": int(getattr(message, "message_id", 0) or 0),
+        "chat_type": _enum_value(getattr(chat, "type", "")) if chat is not None else "",
+        "content_type": _enum_value(getattr(message, "content_type", "")),
+        "text_len": len(text),
+        "is_command": text.startswith("/"),
+    }
+
+
+def describe_command(message: Any) -> dict[str, Any] | None:
+    """Метаданные команды для события ``command`` (``None``, если это не команда)."""
+    text = _text(message)
+    if not text.startswith("/"):
+        return None
+
+    parts = text.split()
+    name = parts[0].split(BOT_MENTION_SEPARATOR, 1)[0].lstrip("/").lower()[:COMMAND_MAX_LENGTH]
+    return {"command": name, "args_count": max(len(parts) - 1, 0)}
+
+
+__all__ = ["describe_command", "describe_message"]

common/config.py

@@ -53,6 +53,19 @@ class Settings(BaseSettings):
     #: Файл-heartbeat для Docker healthcheck (обновляется ботом периодически).
     heartbeat_path: str = "/tmp/dispelevilbot.heartbeat"
 
+    #: ClickHouse (аналитика событий). CLICKHOUSE_ENABLED=false — полностью выключить.
+    clickhouse_enabled: bool = True
+    clickhouse_host: str = "localhost"
+    clickhouse_port: int = 8123
+    clickhouse_db: str = "analytics"
+    clickhouse_user: str = "default"
+    clickhouse_password: str = ""
+
+    #: Батчинг записи событий: размер батча, интервал сброса и предел буфера.
+    analytics_batch_size: int = 200
+    analytics_flush_interval: float = 5.0
+    analytics_buffer_max: int = 5000
+
     @field_validator("chat_id", "alert_chat_id", mode="before")
     @classmethod
     def _empty_int_to_zero(cls, value: object) -> object:

common/models/economy.py

@@ -15,6 +15,16 @@ class PurificationResult:
     hours: int = 0
     minutes: int = 0
     seconds: int = 0
+    #: Уровень до и после очищения (для события `level_up`).
+    level_before: int = 0
+    level: int = 0
+    #: Серия ежедневных очищений (для события `exp_accrual`).
+    streak: int = 0
+
+    @property
+    def level_up(self) -> bool:
+        """Повысился ли уровень по итогам очищения."""
+        return self.level > self.level_before
 
 
 @dataclass(slots=True)

common/texts/alerts.py

@@ -0,0 +1,13 @@
+"""Тексты уведомлений в чат логов (``ALERT_CHAT_ID``).
+
+Служебные сообщения для админов (не для участников), поэтому в ``TEMPLATES``
+(редактируемые шаблоны Блока 4) они не входят.
+"""
+
+from __future__ import annotations
+
+MSG_JOIN = "[вход] {username} вошёл в чат «{chat_title}»"
+MSG_LEAVE = "[выход] {username} покинул чат «{chat_title}»"
+MSG_PURIFICATION = "[очищение] {username} — опыт +{exp_add}, монеты +{money_add}"
+MSG_ALIVE = "[бот живой] версия {version}, аптайм {uptime}"
+MSG_ACCESS_DENIED = "[доступ] {username} стучится в чат"

common/texts/render.py

@@ -42,6 +42,10 @@ TAG_REGISTRY: dict[str, str] = {
     "users_list_count": "Количество игроков в списке",
     "tops": "Топ игроков по сообщениям",
     "promo": "Промокод доната",
+    # --- Служебные теги уведомлений в чат логов (ALERT_CHAT_ID) ---
+    "chat_title": "Название чата",
+    "version": "Версия бота",
+    "uptime": "Время работы бота",
 }
 
 SUPPORTED_TAGS: frozenset[str] = frozenset(TAG_REGISTRY)

common/version.py

@@ -7,4 +7,4 @@
 
 from __future__ import annotations
 
-__version__ = "0.2.5"
+__version__ = "0.3.6"

docker-compose.prod.yml

@@ -46,17 +46,51 @@ services:
     environment:
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
     depends_on:
       postgres:
         condition: service_healthy
       migrate:
         condition: service_completed_successfully
+      # Без ожидания здоровья: аналитика некритична, бот работает и при недоступном ClickHouse.
+      clickhouse:
+        condition: service_started
     logging:
       driver: json-file
       options:
         max-size: "10m"
         max-file: "3"
 
+  # Аналитика событий (docs/events.md). Наружу не публикуется:
+  # только loopback хоста для бэкапа/туннеля; бот ходит внутрь сети compose.
+  clickhouse:
+    image: clickhouse/clickhouse-server:25.3
+    restart: unless-stopped
+    environment:
+      CLICKHOUSE_DB: ${CLICKHOUSE_DB:-analytics}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER:-de_bot}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:?задайте CLICKHOUSE_PASSWORD}
+      # Нужно для создания пользователей/грантов (в финале — read-only пользователь для Superset).
+      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
+    volumes:
+      - clickhousedata:/var/lib/clickhouse
+      # DDL применяется один раз при первой инициализации (пустой data-каталог).
+      - ./clickhouse/init:/docker-entrypoint-initdb.d:ro
+    ports:
+      # Порт контура задаётся в env (prod 8123, test 8124) — иначе конфликт на одной VPS.
+      - "127.0.0.1:${CLICKHOUSE_HTTP_PORT:-8123}:8123"
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    healthcheck:
+      test: ["CMD-SHELL", "wget --spider -q -T 5 http://127.0.0.1:8123/ping || exit 1"]
+      interval: 15s
+      timeout: 10s
+      retries: 10
+      start_period: 20s
+
   # --- Reverse proxy и TLS (Блок 2.4) ---
   # Включается, когда появится админка/Mini App (Блоки 4–5).
   # Раскомментировать и положить deploy/Caddyfile:
@@ -74,5 +108,6 @@ services:
 
 volumes:
   pgdata:
+  clickhousedata:
   # caddy_data:
   # caddy_config:

docker-compose.yml

@@ -51,16 +51,47 @@ services:
       # Внутри сети compose БД доступна по имени сервиса.
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
     depends_on:
       postgres:
         condition: service_healthy
       migrate:
         condition: service_completed_successfully
+      # Аналитика некритична: бот стартует, даже если ClickHouse ещё не готов.
+      clickhouse:
+        condition: service_started
     logging:
       driver: json-file
       options:
         max-size: "10m"
         max-file: "3"
 
+  # Аналитика событий (Блок 3). DDL — в clickhouse/init/.
+  clickhouse:
+    image: clickhouse/clickhouse-server:25.3
+    restart: unless-stopped
+    environment:
+      CLICKHOUSE_DB: ${CLICKHOUSE_DB:-analytics}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER:-de_bot}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:?задайте CLICKHOUSE_PASSWORD в .env}
+      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: "1"
+    volumes:
+      - clickhousedata:/var/lib/clickhouse
+      - ./clickhouse/init:/docker-entrypoint-initdb.d:ro
+    ports:
+      - "127.0.0.1:8123:8123"
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    healthcheck:
+      test: ["CMD-SHELL", "wget --spider -q -T 5 http://127.0.0.1:8123/ping || exit 1"]
+      interval: 15s
+      timeout: 10s
+      retries: 10
+      start_period: 20s
+
 volumes:
   pgdata:
+  clickhousedata:

docs/events.md

@@ -0,0 +1,124 @@
+# События аналитики (ClickHouse)
+
+Бот пишет все значимые действия в ClickHouse — единую широкую таблицу
+`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь
+описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`.
+
+> Источник правды в коде — `common/analytics/events.py` (`EventType`).
+> Расхождение со списком ниже ловит `tests/test_analytics.py`.
+> Добавляешь событие — добавь строку в таблицу.
+
+## Схема в ClickHouse
+
+DDL — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера).
+
+- `analytics.events` — сырые события, `MergeTree`, `PARTITION BY toYYYYMMDD(event_time)`,
+  `ORDER BY (event_time, event_type, user_id)`. Партиционирование по дню выбрано специально:
+  сырые партиции ежедневно удаляются после бэкапа (дешёвый `DROP PARTITION`, без mutation).
+- `analytics.events_daily` — агрегат (`SummingMergeTree`): `day`, `event_type`, `user_id`,
+  `chat_id`, `events`. Строится материализованным представлением `events_daily_mv`.
+- `analytics.money_daily` — агрегат по деньгам: `day`, `event_type`, `user_id`,
+  `amount_sum`, `events` (из `payload.amount` событий `money_accrual`/`money_spend`).
+
+Агрегаты остаются на VPS навсегда, поэтому активность, топ команд и денежные потоки
+доступны и после удаления сырых событий. Запросы к агрегатам суммируют `events`/`amount_sum`
+(фон слияний `SummingMergeTree` не гарантирует полную свёртку в любой момент).
+
+## Хранение и бэкап
+
+| Что | Где | Сколько |
+| --- | --- | --- |
+| Сырые события | VPS, `analytics.events` | последние `BACKUP_GRACE_DAYS` дней (по умолчанию 7) |
+| Агрегаты | VPS | постоянно |
+| Полная история | домашний сервер, свой ClickHouse | архивы `events_YYYY-MM-DD.parquet.gz` (ежедневно) |
+
+Каждую ночь (по Москве) `scripts/clickhouse_backup.sh` выгружает вчерашнюю партицию,
+передаёт её на домашний сервер, импортирует и **сверяет количество строк**; только после
+успешной сверки на VPS удаляются партиции старше грейса. Ошибка бэкапа данные не удаляет.
+Подробности — `DEPLOY.md`.
+
+## Уведомления в чат логов
+
+Помимо записи в ClickHouse часть событий дублируется сообщением в `ALERT_CHAT_ID`
+(если он задан; `0` — выключено). Отправка не критична: ошибки только логируются.
+
+| Уведомление | Когда | Событие в ClickHouse |
+| --- | --- | --- |
+| Вход участника | кто-то вошёл в основной чат | `join` |
+| Выход участника | кто-то вышел из основного чата | `leave` |
+| Очищение | успешное `/очищение` (`+опыт`, `+монеты`) | `purification` |
+| Пинг «бот живой» | раз в час | `bot_alive` |
+| Попытка доступа извне | не участник чата стучится в бота | `access_denied` |
+
+## Поля события
+
+| Поле | Тип | Описание |
+| --- | --- | --- |
+| **event_time** | `DateTime('UTC')` | Момент события (UTC, без таймзоны) |
+| **event_type** | `LowCardinality(String)` | Тип события (см. каталог ниже) |
+| **user_id** | `Int64` | Telegram-ID пользователя (0 — системное событие) |
+| **username** | `String` | `@username` на момент события (может быть пустым) |
+| **chat_id** | `Int64` | ID чата (обычно основной чат клана) |
+| **payload** | `String` | Дополнительные данные события в виде JSON |
+| **bot_version** | `String` | Версия бота из `common/version.py` |
+| **prod** | `UInt8` | 1 — прод-контур, 0 — тестовый |
+
+## Каталог событий
+
+| Тип события | Когда пишется | `payload` |
+| --- | --- | --- |
+| `message` | Любое сообщение в основном чате | `message_id`, `chat_type`, `content_type`, `text_len`, `is_command` |
+| `command` | Использование команды (`/инфо`, `/очищение`, …) | `command`, `args_count` |
+| `purification` | Ежедневное очищение (`/очищение`) | `success`, `exp_add`, `money_add` |
+| `exp_accrual` | Начисление опыта за очищение | `amount`, `streak` |
+| `level_up` | Повышение уровня по итогам очищения | `level_from`, `level_to` |
+| `money_accrual` | Начисление валюты (монетка за активность, награда за очищение) | `amount`, `reason`, `action_id` |
+| `money_spend` | Трата валюты (покупка, обнуление баланса при удалении) | `amount`, `reason`, `product_id` |
+| `eggs` | Получение яичек (`/яички`) | `amount` |
+| `purchase_title` | Покупка звания | `title_id`, `price`, `success` |
+| `purchase_donat` | Покупка доната | `donat_id`, `price`, `success` |
+| `join` | Вход участника в основной чат | `chat_title` |
+| `leave` | Выход участника из основного чата | `chat_title` |
+| `nick_set` | Смена ника (`/ник`) | `nick` |
+| `subscribe` | Подписка на рассылки (`/подписка`) | — |
+| `unsubscribe` | Отписка от рассылок (`/отписка`) | — |
+| `access_denied` | Отказ в доступе (не участник чата и не админ) | `reason` |
+| `bot_start` | Старт бота | `app_env` |
+| `bot_stop` | Остановка бота | `app_env`, `uptime_seconds` |
+| `bot_alive` | Ежечасный пинг «бот живой» | `app_env`, `uptime_seconds` |
+| `error` | Необработанная ошибка при обработке апдейта | `error_type`, `error`, `update_type` |
+
+## Поведение при недоступности ClickHouse
+
+Аналитика — **некритичная** подсистема: её сбой не должен влиять на бота.
+
+- события копятся в буфере и отправляются батчами (по размеру и по таймеру);
+- если ClickHouse недоступен, буфер не растёт бесконечно: при переполнении самые
+  старые записи отбрасываются со счётчиком в логе;
+- если записать батч не удалось, события выводятся в stdout (`WARNING`, первые
+  несколько строк + общее количество) и отбрасываются — данные не пропадают молча;
+- при остановке бота выполняется финальный сброс буфера;
+- выключение целиком — `CLICKHOUSE_ENABLED=false` (тогда `record()` — no-op).
+
+## Примеры запросов
+
+```sql
+-- активность по дням
+SELECT toDate(event_time) AS day, count() FROM analytics.events GROUP BY day ORDER BY day;
+
+-- топ команд за неделю
+SELECT JSONExtractString(payload, 'command') AS cmd, count() AS uses
+FROM analytics.events
+WHERE event_type = 'command' AND event_time >= now() - INTERVAL 7 DAY
+GROUP BY cmd ORDER BY uses DESC;
+
+-- покупки за месяц
+SELECT event_type, count(), sum(JSONExtractInt(payload, 'price'))
+FROM analytics.events
+WHERE event_type IN ('purchase_title', 'purchase_donat') AND event_time >= now() - INTERVAL 30 DAY
+GROUP BY event_type;
+
+-- ошибки
+SELECT event_time, JSONExtractString(payload, 'error_type'), JSONExtractString(payload, 'error')
+FROM analytics.events WHERE event_type = 'error' ORDER BY event_time DESC LIMIT 50;
+```

env/backup.env.example

@@ -0,0 +1,28 @@
+# Параметры ежедневного бэкапа событий ClickHouse с VPS на домашний сервер.
+# Скопируйте в env/backup.env (env/*.env в git не попадает) и заполните.
+#
+# Используется скриптами:
+#   scripts/clickhouse_backup.sh  (запускается на VPS по cron, см. DEPLOY.md)
+#   scripts/clickhouse_import.sh  (лежит на домашнем сервере)
+
+# Куда передавать архивы: пользователь@хост и каталог на домашнем сервере.
+BACKUP_REMOTE=user@home.example
+BACKUP_DIR=/srv/dispelevilbot/events
+
+# Приватный ключ для ssh/rsync (без пароля; на домашнем сервере — соответствующий public key).
+BACKUP_SSH_KEY=/root/.ssh/dispelevilbot_backup
+
+# Сколько дней сырых событий оставлять на VPS (остальное удаляется после успешной сверки).
+BACKUP_GRACE_DAYS=7
+
+# Таймзона расчёта «вчерашнего» дня.
+BACKUP_TZ=Europe/Moscow
+
+# Как запускать импорт на домашнем сервере (путь к scripts/clickhouse_import.sh).
+REMOTE_IMPORT_SCRIPT=./clickhouse_import.sh
+
+# ClickHouse домашнего сервера. Пароль лучше без кавычек и пробелов — он передаётся по ssh.
+REMOTE_CLICKHOUSE_HOST=localhost
+REMOTE_CLICKHOUSE_PORT=9000
+REMOTE_CLICKHOUSE_USER=default
+REMOTE_CLICKHOUSE_PASSWORD=

env/prod.env.example

@@ -7,7 +7,7 @@ ENV_FILE=env/prod.env
 
 APP_ENV=prod
 
-# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию, напр. 0.2.5)
+# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию)
 REGISTRY=git.mpakosrv.ru
 IMAGE_TAG=latest
 
@@ -30,3 +30,12 @@ ALERT_CHAT_ID=0
 # Telegram-ID админов (доступ в обход проверки членства), через запятую; можно пусто
 ADMIN_IDS=
 TIMEZONE=Europe/Moscow
+
+# ClickHouse (аналитика событий, см. docs/events.md).
+# Наружу не публикуется: только loopback хоста для бэкапа/туннеля;
+# бот ходит внутрь сети compose по имени сервиса clickhouse:8123.
+CLICKHOUSE_DB=analytics
+CLICKHOUSE_USER=de_bot
+CLICKHOUSE_PASSWORD=
+# Порт на хосте (для бэкапа). У test и prod он должен различаться.
+CLICKHOUSE_HTTP_PORT=8123

env/test.env.example

@@ -6,7 +6,7 @@ ENV_FILE=env/test.env
 
 APP_ENV=dev
 
-# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию, напр. 0.2.5)
+# Образ в реестре (latest = последняя сборка из master; можно закрепить <sha> или версию)
 REGISTRY=git.mpakosrv.ru
 IMAGE_TAG=latest
 
@@ -28,3 +28,12 @@ ALERT_CHAT_ID=0
 # Telegram-ID админов (доступ в обход проверки членства), через запятую; можно пусто
 ADMIN_IDS=
 TIMEZONE=Europe/Moscow
+
+# ClickHouse (аналитика событий, см. docs/events.md).
+# Наружу не публикуется: только loopback хоста для бэкапа/туннеля;
+# бот ходит внутрь сети compose по имени сервиса clickhouse:8123.
+CLICKHOUSE_DB=analytics
+CLICKHOUSE_USER=de_bot
+CLICKHOUSE_PASSWORD=
+# Порт на хосте: у test он отличается от prod (8123), иначе конфликт на одной VPS.
+CLICKHOUSE_HTTP_PORT=8124

requirements.txt

@@ -7,7 +7,7 @@ pydantic-settings>=2.3,<3.0
 python-dotenv>=1.0,<2.0
 psycopg[binary,pool]>=3.2,<4.0
 
-# Блок 3 (ClickHouse) — подключим позже:
+# ClickHouse (аналитика событий, Блок 3). Extras [async] — нативный asyncio-клиент.
-# clickhouse-connect>=0.7,<1.0
+clickhouse-connect[async]>=1.0,<2.0
 # Блок 4 (расписания из БД) — при необходимости:
 # APScheduler>=3.10,<4.0

scripts/check_contour.sh

@@ -25,7 +25,7 @@ echo "== проверки =="
 
 running="$($COMPOSE ps --services --status running || true)"
 fail=0
-for svc in postgres bot; do
+for svc in postgres clickhouse bot; do
     if echo "$running" | grep -qx "$svc"; then
         echo "OK   ${svc}: running"
     else
@@ -49,4 +49,18 @@ else
     fail=1
 fi
 
+# ClickHouse отвечает и схема аналитики применена (initdb применяется один раз,
+# при первой инициализации пустого data-каталога)
+if $COMPOSE exec -T clickhouse sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" --query "SELECT 1"' >/dev/null 2>&1; then
+    echo "OK   clickhouse: принимает запросы"
+    if $COMPOSE exec -T clickhouse sh -lc 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" --query "SELECT count() FROM analytics.events"' >/dev/null 2>&1; then
+        echo "OK   clickhouse: таблица analytics.events доступна"
+    else
+        echo "WARN clickhouse: нет доступа к analytics.events (см. clickhouse/init/001_events.sql)" >&2
+    fi
+else
+    echo "FAIL clickhouse: не отвечает" >&2
+    fail=1
+fi
+
 exit $fail

scripts/clickhouse_backup.sh

@@ -0,0 +1,109 @@
+#!/usr/bin/env bash
+# Ежедневный бэкап сырых событий ClickHouse с VPS на домашний сервер.
+#
+# Что делает:
+#   1) выгружает партицию «вчерашнего» дня из analytics.events в Parquet.gz;
+#   2) передаёт файл на домашний сервер (rsync) и импортирует его в тамошний ClickHouse;
+#   3) сверяет количество строк на VPS и на домашнем сервере;
+#   4) только после успешной сверки удаляет на VPS партиции старше BACKUP_GRACE_DAYS.
+#
+# Ошибка на любом шаге оставляет данные на VPS и возвращает ненулевой код.
+#
+# Запуск (на VPS, из каталога проекта):
+#   ./scripts/clickhouse_backup.sh test        # или prod
+#   make backup E=test
+#
+# Настройки (env/<контур>.env + env/backup.env, см. env/backup.env.example):
+#   CLICKHOUSE_*        доступ к ClickHouse контура
+#   BACKUP_REMOTE       user@host домашнего сервера
+#   BACKUP_DIR          каталог архива на домашнем сервере
+#   BACKUP_SSH_KEY      приватный ключ для ssh/rsync
+#   BACKUP_GRACE_DAYS   сколько дней сырых событий оставлять на VPS (по умолчанию 7)
+#   BACKUP_TZ           таймзона расчёта «вчера» (по умолчанию Europe/Moscow)
+
+set -euo pipefail
+
+E="${1:-test}"
+ENV_FILE="env/${E}.env"
+BACKUP_ENV="env/backup.env"
+COMPOSE="docker compose -p ${E} --env-file ${ENV_FILE} -f docker-compose.prod.yml"
+
+log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; }
+fail() {
+    echo "[$(date '+%Y-%m-%d %H:%M:%S')] ОШИБКА: $*" >&2
+    exit 1
+}
+
+[ -f "$ENV_FILE" ] || fail "нет файла окружения ${ENV_FILE}"
+# shellcheck disable=SC1090
+set -a && . "./${ENV_FILE}" && set +a
+if [ -f "$BACKUP_ENV" ]; then
+    # shellcheck disable=SC1090
+    set -a && . "./${BACKUP_ENV}" && set +a
+fi
+
+: "${BACKUP_REMOTE:?задайте BACKUP_REMOTE в env/backup.env}"
+: "${BACKUP_DIR:?задайте BACKUP_DIR в env/backup.env}"
+BACKUP_SSH_KEY="${BACKUP_SSH_KEY:-$HOME/.ssh/id_rsa}"
+BACKUP_GRACE_DAYS="${BACKUP_GRACE_DAYS:-7}"
+BACKUP_TZ="${BACKUP_TZ:-Europe/Moscow}"
+REMOTE_IMPORT_SCRIPT="${REMOTE_IMPORT_SCRIPT:-./clickhouse_import.sh}"
+REMOTE_CLICKHOUSE_HOST="${REMOTE_CLICKHOUSE_HOST:-localhost}"
+REMOTE_CLICKHOUSE_PORT="${REMOTE_CLICKHOUSE_PORT:-9000}"
+REMOTE_CLICKHOUSE_USER="${REMOTE_CLICKHOUSE_USER:-default}"
+REMOTE_CLICKHOUSE_PASSWORD="${REMOTE_CLICKHOUSE_PASSWORD:-}"
+SSH=(ssh -i "$BACKUP_SSH_KEY" -o BatchMode=yes "$BACKUP_REMOTE")
+
+ch() {
+    $COMPOSE exec -T clickhouse clickhouse-client \
+        --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" "$@"
+}
+
+DAY="$(TZ="$BACKUP_TZ" date -d 'yesterday' '+%Y-%m-%d')"
+PARTITION="${DAY//-/}"
+GRACE_CUTOFF="$(TZ="$BACKUP_TZ" date -d "${BACKUP_GRACE_DAYS} days ago" '+%Y%m%d')"
+TMP_DIR="$(mktemp -d)"
+trap 'rm -rf "$TMP_DIR"' EXIT
+
+log "Контур ${E}: бэкап событий за ${DAY} (грейс ${BACKUP_GRACE_DAYS} дней, порог ${GRACE_CUTOFF})"
+
+if ! $COMPOSE ps --services --status running 2>/dev/null | grep -qx clickhouse; then
+    fail "контейнер clickhouse контура ${E} не запущен"
+fi
+
+ROWS_LOCAL="$(ch --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}')" | tr -d '[:space:]')"
+log "На VPS строк за ${DAY}: ${ROWS_LOCAL}"
+if [ "$ROWS_LOCAL" = "0" ]; then
+    log "ВНИМАНИЕ: за ${DAY} событий нет — партиция пустая, продолжаю (сверка 0 = 0)"
+fi
+
+ARCHIVE="${TMP_DIR}/events_${DAY}.parquet.gz"
+log "Выгружаю партицию в ${ARCHIVE}"
+ch --query "SELECT * FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}') FORMAT Parquet" | gzip -9 >"$ARCHIVE"
+[ -s "$ARCHIVE" ] || fail "не удалось выгрузить архив ${DAY}"
+log "Размер архива: $(du -h "$ARCHIVE" | cut -f1)"
+
+log "Передаю архив на ${BACKUP_REMOTE}:${BACKUP_DIR}"
+rsync -az -e "ssh -i ${BACKUP_SSH_KEY} -o BatchMode=yes" "$ARCHIVE" "${BACKUP_REMOTE}:${BACKUP_DIR}/"
+
+log "Импортирую архив в ClickHouse домашнего сервера"
+ROWS_REMOTE="$("${SSH[@]}" \
+    "BACKUP_DIR='${BACKUP_DIR}' CLICKHOUSE_HOST='${REMOTE_CLICKHOUSE_HOST}' CLICKHOUSE_PORT='${REMOTE_CLICKHOUSE_PORT}' \
+CLICKHOUSE_USER='${REMOTE_CLICKHOUSE_USER}' CLICKHOUSE_PASSWORD='${REMOTE_CLICKHOUSE_PASSWORD}' \
+'${REMOTE_IMPORT_SCRIPT}' '${DAY}'" | tail -n1 | tr -d '[:space:]')"
+log "На домашнем сервере строк за ${DAY}: ${ROWS_REMOTE}"
+
+[ "$ROWS_LOCAL" = "$ROWS_REMOTE" ] || fail "расхождение строк (VPS ${ROWS_LOCAL}, домашний ${ROWS_REMOTE}) — партиции на VPS не удаляю"
+
+log "Удаляю на VPS партиции старше ${GRACE_CUTOFF}"
+OLDER="$(ch --query "SELECT DISTINCT partition FROM system.parts WHERE database='analytics' AND table='events' AND active AND partition < '${GRACE_CUTOFF}'")"
+if [ -z "$OLDER" ]; then
+    log "Удалять нечего"
+else
+    for partition in $OLDER; do
+        log "DROP PARTITION ID '${partition}'"
+        ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${partition}'"
+    done
+fi
+
+log "Готово: события за ${DAY} в архиве, на VPS остались агрегаты и последние ${BACKUP_GRACE_DAYS} дней сырых событий"

scripts/clickhouse_import.sh

@@ -0,0 +1,58 @@
+#!/usr/bin/env bash
+# Импорт дневного архива событий в ClickHouse домашнего сервера.
+#
+# Запускается на домашнем сервере (обычно по ssh из scripts/clickhouse_backup.sh).
+# Рядом должны лежать этот скрипт и канонический DDL кликхаус-схемы
+# (clickhouse/init/001_events.sql из репозитория).
+#
+# Использование: ./clickhouse_import.sh 2026-01-01
+# В stdout печатается количество строк загруженного дня (для сверки), диагностика — в stderr.
+#
+# Настройки (окружение):
+#   BACKUP_DIR          каталог, куда складываются архивы events_YYYY-MM-DD.parquet.gz
+#   SCHEMA_FILE         путь к DDL схемы (по умолчанию ./001_events.sql рядом со скриптом)
+#   CLICKHOUSE_CLIENT   бинарь клиента (по умолчанию clickhouse-client)
+#   CLICKHOUSE_HOST/PORT/USER/PASSWORD  доступ к ClickHouse домашнего сервера
+
+set -euo pipefail
+
+DAY="${1:?укажите дату в формате YYYY-MM-DD}"
+PARTITION="${DAY//-/}"
+ARCHIVE_DIR="${BACKUP_DIR:-.}"
+ARCHIVE="${ARCHIVE_DIR}/events_${DAY}.parquet.gz"
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+SCHEMA_FILE="${SCHEMA_FILE:-${SCRIPT_DIR}/001_events.sql}"
+
+CLIENT="${CLICKHOUSE_CLIENT:-clickhouse-client}"
+
+log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >&2; }
+
+ch() {
+    "$CLIENT" \
+        --host "${CLICKHOUSE_HOST:-localhost}" \
+        --port "${CLICKHOUSE_PORT:-9000}" \
+        --user "${CLICKHOUSE_USER:-default}" \
+        --password "${CLICKHOUSE_PASSWORD:-}" "$@"
+}
+
+[ -f "$ARCHIVE" ] || {
+    log "нет файла архива ${ARCHIVE}"
+    exit 1
+}
+[ -f "$SCHEMA_FILE" ] || {
+    log "нет файла схемы ${SCHEMA_FILE} (скопируйте clickhouse/init/001_events.sql на домашний сервер)"
+    exit 1
+}
+
+log "Применяю схему из ${SCHEMA_FILE}"
+ch --multiquery <"$SCHEMA_FILE"
+
+# Повторный импорт того же дня не должен дублировать строки.
+log "Очищаю партицию ${PARTITION} перед импортом"
+ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${PARTITION}'" 2>/dev/null || true
+
+log "Импортирую ${ARCHIVE} в analytics.events"
+gzip -dc "$ARCHIVE" | ch --query "INSERT INTO analytics.events FORMAT Parquet"
+
+log "Импорт завершён"
+ch --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate('${DAY}')"

sql/R__functions_economy.sql

@@ -35,12 +35,28 @@ BEGIN
 END;
 $$;
 
+-- Ежедневное очищение: опыт, монеты, уровень.
+-- Колонки lvl_before/lvl/streak нужны аналитике (события exp_accrual и level_up),
+-- поэтому сигнатуру пришлось изменить — сначала DROP (repeatable-миграция).
+DROP FUNCTION IF EXISTS bot.purification(bigint);
+
 CREATE OR REPLACE FUNCTION bot.purification(p_id_user bigint)
-    RETURNS TABLE(res integer, expr_add bigint, mny_add bigint, h integer, m integer, s integer)
+    RETURNS TABLE(
+        res integer,
+        expr_add bigint,
+        mny_add bigint,
+        h integer,
+        m integer,
+        s integer,
+        lvl_before bigint,
+        lvl_after bigint,
+        streak bigint
+    )
     LANGUAGE plpgsql
 AS $$
 DECLARE
     v_lvl bigint;
+    v_lvl_before bigint;
     v_expr bigint;
     v_expr_add bigint;
     v_streak bigint;
@@ -57,6 +73,8 @@ BEGIN
     FROM bot.de_users
     WHERE id_user = p_id_user;
 
+    v_lvl_before := v_lvl;
+
     BEGIN
         SELECT dt INTO v_dt_pur
         FROM (
@@ -113,7 +131,7 @@ BEGIN
         v_result := 1;
     END IF;
 
-    RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s;
+    RETURN QUERY SELECT v_result, v_expr_add, v_mny_add, v_h, v_m, v_s, v_lvl_before, v_lvl, v_streak;
 END;
 $$;
 

tests/conftest.py

@@ -0,0 +1,20 @@
+"""Общие фикстуры тестов."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+
+from common.config import Settings
+
+
+@pytest.fixture
+def settings_factory() -> Callable[..., Settings]:
+    """Фабрика ``Settings`` с обязательным ``BOT_TOKEN`` и без чтения ``.env``."""
+
+    def _factory(**overrides: Any) -> Settings:
+        return Settings(_env_file=None, bot_token="1:a", **overrides)
+
+    return _factory

tests/test_alerts.py

@@ -0,0 +1,179 @@
+"""Тесты уведомлений в чат логов и ежечасного пинга «бот живой»."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from bot.services import alerts, heartbeat
+from common.analytics import EventType, close_analytics, init_analytics
+from common.config import Settings
+from common.models import PurificationResult
+
+
+class FakeBot:
+    """Заглушка aiogram.Bot: записывает отправленные и пересланные сообщения."""
+
+    def __init__(self, *, fail: bool = False) -> None:
+        self.messages: list[tuple[int, str]] = []
+        self.forwarded: list[tuple[int, int, int]] = []
+        self._fail = fail
+
+    async def send_message(self, chat_id: int, text: str) -> None:
+        if self._fail:
+            raise RuntimeError("Telegram недоступен")
+        self.messages.append((chat_id, text))
+
+    async def forward_message(self, chat_id: int, from_chat_id: int, message_id: int) -> None:
+        if self._fail:
+            raise RuntimeError("Telegram недоступен")
+        self.forwarded.append((chat_id, from_chat_id, message_id))
+
+
+class FakeClickHouse:
+    """Заглушка клиента ClickHouse (см. тесты аналитики)."""
+
+    def __init__(self) -> None:
+        self.rows: list[list[Any]] = []
+
+    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
+        self.rows.extend(data)
+
+    async def close(self) -> None:
+        pass
+
+
+@pytest.fixture(autouse=True)
+async def _clean_singleton():
+    await close_analytics()
+    yield
+    await close_analytics()
+
+
+@pytest.fixture(autouse=True)
+def _reset_ping_hour(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пинг не должен «протекать» между тестами."""
+    monkeypatch.setattr(heartbeat, "_last_ping_hour", None)
+
+
+def _patch_settings(monkeypatch: pytest.MonkeyPatch, settings: Settings) -> Settings:
+    """Подменить настройки во всех модулях, которые их читают."""
+    monkeypatch.setattr("bot.services.alerts.get_settings", lambda: settings)
+    monkeypatch.setattr("bot.services.heartbeat.get_settings", lambda: settings)
+    return settings
+
+
+# --- отправка в чат логов ---
+
+
+async def test_notify_alert_skips_when_chat_disabled(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=0))
+    bot = FakeBot()
+
+    assert await alerts.notify_alert(bot, "текст") is False
+    assert bot.messages == []
+
+
+async def test_notify_alert_sends_to_alert_chat(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
+    bot = FakeBot()
+
+    assert await alerts.notify_alert(bot, "текст") is True
+    assert bot.messages == [(-100, "текст")]
+
+
+async def test_notify_alert_failure_is_logged(
+    settings_factory: Callable[..., Settings],
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
+
+    with caplog.at_level(logging.ERROR):
+        assert await alerts.notify_alert(FakeBot(fail=True), "текст") is False
+
+    assert any("не удалось отправить уведомление" in message.lower() for message in caplog.messages)
+
+
+async def test_notify_alert_forwards_message(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
+    bot = FakeBot()
+    message = SimpleNamespace(chat=SimpleNamespace(id=555), message_id=42)
+
+    assert await alerts.notify_alert(bot, "текст", forward=message) is True
+    assert bot.forwarded == [(-100, 555, 42)]
+
+
+async def test_notify_join_and_leave_text(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
+    bot = FakeBot()
+
+    await alerts.notify_join(bot, username="vasya", user_id=7, chat_title="Клан")
+    await alerts.notify_leave(bot, username=None, user_id=7, chat_title="Клан")
+
+    assert bot.messages[0] == (-100, "[вход] @vasya вошёл в чат «Клан»")
+    assert bot.messages[1] == (-100, "[выход] 7 покинул чат «Клан»")
+
+
+async def test_notify_purification_text(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100))
+    bot = FakeBot()
+
+    result = PurificationResult(success=True, exp_add=3, money_add=41)
+    await alerts.notify_purification(bot, username="vasya", user_id=7, result=result)
+
+    assert bot.messages == [(-100, "[очищение] @vasya — опыт +3, монеты +41")]
+
+
+# --- ежечасный пинг «бот живой» ---
+
+
+async def test_ping_hourly_sends_once_per_hour(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    settings = _patch_settings(monkeypatch, settings_factory(alert_chat_id=-100, app_env="prod"))
+    fake = FakeClickHouse()
+    await init_analytics(settings, client=fake)
+    bot = FakeBot()
+
+    assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is True
+    assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 30, tzinfo=UTC)) is False
+    assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 11, 5, tzinfo=UTC)) is True
+
+    await close_analytics()
+
+    assert len(bot.messages) == 2
+    assert "бот живой" in bot.messages[0][1]
+    assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE, EventType.BOT_ALIVE]
+
+
+async def test_ping_hourly_records_event_without_alert_chat(
+    settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    settings = _patch_settings(monkeypatch, settings_factory(alert_chat_id=0))
+    fake = FakeClickHouse()
+    await init_analytics(settings, client=fake)
+    bot = FakeBot()
+
+    assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is False
+    assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 45, tzinfo=UTC)) is False
+
+    await close_analytics()
+
+    assert bot.messages == []
+    assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE]

tests/test_analytics.py

@@ -0,0 +1,351 @@
+"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse.
+
+Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а
+``clickhouse_connect`` импортируется лениво внутри клиента.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from datetime import UTC, datetime
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from common.analytics import ALL_EVENT_TYPES, EventType, close_analytics, get_analytics, init_analytics, record
+from common.analytics.client import PAYLOAD_MAX_LENGTH, AnalyticsClient
+from common.analytics.messages import describe_command, describe_message
+from common.config import Settings
+from common.version import __version__
+
+_DOCS_PATH = Path(__file__).resolve().parents[1] / "docs" / "events.md"
+
+#: Строка каталога событий: первый столбец в обратных кавычках.
+_EVENT_ROW = re.compile(r"^\|\s*`([a-z_]+)`\s*\|", re.MULTILINE)
+
+
+class _FakeEnum:
+    """Заглушка enum-полей aiogram (у значений есть ``.value``)."""
+
+    def __init__(self, value: str) -> None:
+        self.value = value
+
+
+class FakeClickHouse:
+    """Заглушка клиента ClickHouse: пишет в список или падает."""
+
+    def __init__(self, *, fail: bool = False) -> None:
+        self.rows: list[list[Any]] = []
+        self.table: str | None = None
+        self.columns: list[str] | None = None
+        self.closed = False
+        self._fail = fail
+
+    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
+        if self._fail:
+            raise RuntimeError("ClickHouse недоступен")
+        self.table = table
+        self.columns = list(column_names)
+        self.rows.extend(data)
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+def _settings(**overrides: Any) -> Settings:
+    return Settings(_env_file=None, bot_token="1:a", **overrides)
+
+
+@pytest.fixture(autouse=True)
+async def _clean_singleton():
+    """Тесты не должны оставлять активный singleton аналитики."""
+    await close_analytics()
+    yield
+    await close_analytics()
+
+
+def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeClickHouse]:
+    fake = FakeClickHouse()
+    return AnalyticsClient(_settings(**overrides), client=fake), fake
+
+
+# --- каталог событий ---
+
+
+def test_event_types_are_unique_and_nonempty() -> None:
+    assert len(ALL_EVENT_TYPES) == len(EventType)
+    assert ALL_EVENT_TYPES
+
+
+def test_event_types_match_documentation() -> None:
+    documented = set(_EVENT_ROW.findall(_DOCS_PATH.read_text(encoding="utf-8")))
+    assert documented == set(ALL_EVENT_TYPES), "Список событий в common/analytics/events.py и docs/events.md разошёлся"
+
+
+# --- метаданные сообщения (без aiogram) ---
+
+
+def _fake_message(**overrides: Any) -> Any:
+    """Минимальная заглушка aiogram Message для describe_message/describe_command."""
+    fields: dict[str, Any] = {
+        "message_id": 77,
+        "text": "привет",
+        "caption": None,
+        "content_type": _FakeEnum("text"),
+        "chat": SimpleNamespace(type=_FakeEnum("supergroup")),
+    }
+    fields.update(overrides)
+    return SimpleNamespace(**fields)
+
+
+def test_describe_message_without_text() -> None:
+    payload = describe_message(_fake_message(text="привет"))
+
+    assert payload == {
+        "message_id": 77,
+        "chat_type": "supergroup",
+        "content_type": "text",
+        "text_len": 6,
+        "is_command": False,
+    }
+
+
+def test_describe_message_uses_caption_for_media() -> None:
+    payload = describe_message(_fake_message(text=None, caption="подпись", content_type=_FakeEnum("photo")))
+
+    assert payload["content_type"] == "photo"
+    assert payload["text_len"] == len("подпись")
+    assert payload["is_command"] is False
+
+
+def test_describe_command_bare_and_with_args() -> None:
+    assert describe_command(_fake_message(text="/очищение")) == {"command": "очищение", "args_count": 0}
+    assert describe_command(_fake_message(text="/ник Вася")) == {"command": "ник", "args_count": 1}
+
+
+def test_describe_command_strips_bot_mention() -> None:
+    assert describe_command(_fake_message(text="/Ник@DispelEvilBot Вася")) == {"command": "ник", "args_count": 1}
+
+
+def test_describe_command_returns_none_for_plain_message() -> None:
+    assert describe_command(_fake_message(text="просто текст")) is None
+    assert describe_command(_fake_message(text=None, caption="подпись")) is None
+
+
+# --- буферизация и запись ---
+
+
+async def test_record_without_init_is_noop() -> None:
+    assert get_analytics() is None
+    record(EventType.MESSAGE, user_id=1, chat_id=2, payload={"a": 1})  # не должно бросать
+
+
+async def test_client_buffers_and_flushes_rows() -> None:
+    client, fake = _client()
+    client.record(EventType.MESSAGE, user_id=11, username="vasya", chat_id=100, payload={"message_id": 5})
+    client.record(EventType.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "/инфо"})
+
+    assert client.pending == 2
+
+    await client.flush()
+
+    assert client.pending == 0
+    assert fake.table == "analytics.events"
+    assert fake.columns == [
+        "event_time",
+        "event_type",
+        "user_id",
+        "username",
+        "chat_id",
+        "payload",
+        "bot_version",
+        "prod",
+    ]
+    assert len(fake.rows) == 2
+
+    row = fake.rows[0]
+    assert isinstance(row[0], datetime)
+    assert row[1] == "message"
+    assert row[2] == 11
+    assert row[3] == "vasya"
+    assert row[4] == 100
+    assert '"message_id": 5' in row[5]
+    assert row[6] == __version__
+    assert row[7] == 0
+
+
+async def test_remember_user_enriches_username() -> None:
+    client, fake = _client()
+    client.remember_user(42, "nickname")
+    client.record(EventType.PURIFICATION, user_id=42, chat_id=1)
+
+    await client.flush()
+
+    assert fake.rows[0][3] == "nickname"
+
+
+async def test_record_triggers_flush_at_batch_size() -> None:
+    client, _ = _client(analytics_batch_size=3)
+
+    client.record(EventType.MESSAGE, user_id=1)
+    client.record(EventType.MESSAGE, user_id=1)
+    assert not client._flush_now.is_set()
+
+    client.record(EventType.MESSAGE, user_id=1)
+    assert client._flush_now.is_set()
+
+
+async def test_overflow_drops_oldest_events() -> None:
+    client, _ = _client(analytics_buffer_max=2, analytics_batch_size=100)
+
+    for user_id in range(1, 5):
+        client.record(EventType.MESSAGE, user_id=user_id)
+
+    assert client.pending == 2
+    assert [row[2] for row in client._rows] == [3, 4]
+
+
+async def test_disabled_analytics_ignores_events() -> None:
+    client, fake = _client(clickhouse_enabled=False)
+
+    client.record(EventType.MESSAGE, user_id=1)
+    await client.flush()
+
+    assert client.pending == 0
+    assert fake.rows == []
+
+
+# --- деградация без ClickHouse ---
+
+
+async def test_flush_failure_logs_fallback_and_drops(caplog: pytest.LogCaptureFixture) -> None:
+    fake = FakeClickHouse(fail=True)
+    client = AnalyticsClient(_settings(), client=fake)
+    client.record(EventType.MESSAGE, user_id=1, username="vasya", chat_id=100)
+
+    with caplog.at_level(logging.WARNING):
+        await client.flush()  # не должно бросать
+
+    assert client.pending == 0
+    assert any("Fallback" in message for message in caplog.messages)
+
+
+async def test_unavailable_client_falls_back_to_log(caplog: pytest.LogCaptureFixture) -> None:
+    settings = _settings(clickhouse_host="127.0.0.1", clickhouse_port=1, clickhouse_password="x")
+    client = AnalyticsClient(settings)
+    client.record(EventType.BOT_START, payload={"app_env": "dev"})
+
+    with caplog.at_level(logging.WARNING):
+        await client.flush()
+
+    assert client.pending == 0
+    assert any("Fallback" in message for message in caplog.messages)
+
+
+async def test_payload_is_truncated() -> None:
+    client, fake = _client()
+    client.record(EventType.MESSAGE, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)})
+
+    await client.flush()
+
+    assert len(fake.rows[0][5]) == PAYLOAD_MAX_LENGTH
+
+
+# --- жизненный цикл ---
+
+
+async def test_init_and_close_flush_pending_events() -> None:
+    fake = FakeClickHouse()
+    client = await init_analytics(_settings(), client=fake)
+
+    record(EventType.BOT_START, payload={"app_env": "dev"})
+    record(EventType.BOT_STOP, payload={"app_env": "dev"})
+
+    assert get_analytics() is client
+    assert client.pending == 2
+
+    await close_analytics()
+
+    assert get_analytics() is None
+    assert len(fake.rows) == 2
+    assert fake.closed is True
+
+
+# --- middleware (транспортные события) ---
+
+
+def _message(**overrides: Any) -> Any:
+    """Реальное сообщение aiogram для проверки middleware."""
+    from aiogram.types import Chat, Message, User
+
+    fields: dict[str, Any] = {
+        "message_id": 5,
+        "date": datetime.now(UTC),
+        "chat": Chat(id=-100, type="supergroup", title="Клан"),
+        "from_user": User(id=7, is_bot=False, first_name="Вася", username="vasya"),
+        "text": "/инфо",
+    }
+    fields.update(overrides)
+    return Message(**fields)
+
+
+async def _handler(event: Any, data: dict[str, Any]) -> str:
+    return "handled"
+
+
+async def test_middleware_records_message_and_command(monkeypatch: pytest.MonkeyPatch) -> None:
+    from bot.middlewares.analytics import AnalyticsMiddleware
+
+    settings = _settings(chat_id=-100)
+    monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
+
+    fake = FakeClickHouse()
+    client = await init_analytics(settings, client=fake)
+
+    assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"
+    await client.flush()
+
+    assert [row[1] for row in fake.rows] == ["message", "command"]
+    assert json.loads(fake.rows[0][5])["is_command"] is True
+    assert json.loads(fake.rows[1][5]) == {"command": "инфо", "args_count": 0}
+    assert fake.rows[0][3] == "vasya"
+
+
+async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyPatch) -> None:
+    from bot.middlewares.analytics import AnalyticsMiddleware
+
+    settings = _settings(chat_id=-100)
+    monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
+
+    fake = FakeClickHouse()
+    client = await init_analytics(settings, client=fake)
+
+    from aiogram.types import Chat, User
+
+    await AnalyticsMiddleware()(_handler, _message(chat=Chat(id=-999, type="supergroup")), {})
+    await AnalyticsMiddleware()(
+        _handler,
+        _message(from_user=User(id=7, is_bot=True, first_name="Bot")),
+        {},
+    )
+    await client.flush()
+
+    assert fake.rows == []
+
+
+async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) -> None:
+    from bot.middlewares.analytics import AnalyticsMiddleware
+
+    settings = _settings(chat_id=-100)
+    monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
+
+    def boom(*args: Any, **kwargs: Any) -> None:
+        raise RuntimeError("аналитика сломалась")
+
+    monkeypatch.setattr("bot.middlewares.analytics.record", boom)
+
+    assert await AnalyticsMiddleware()(_handler, _message(), {}) == "handled"

tests/test_version.py

@@ -7,4 +7,4 @@ from common import __version__
 
 def test_version_is_set():
     assert __version__
-    assert __version__ == "0.2.5"
+    assert __version__ == "0.3.6"