.env.example

@@ -35,9 +35,15 @@ ADMIN_IDS=
 # Часовой пояс для расписаний
 TIMEZONE=Europe/Moscow
 
-# Аналитика (Postgres, схема analytics). Логируются действия бота (см. docs/events.md).
-# ANALYTICS_ENABLED=false — полностью выключить аналитику.
-ANALYTICS_ENABLED=true
+# Аналитика (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

DEPLOY.md

@@ -16,14 +16,16 @@ cd DispelEvilBot
 ```bash
 cp env/test.env.example env/test.env
 # заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.),
-# ADMIN_IDS (опц.)
+# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий)
 # плюс для админки: BOT_USERNAME, ADMIN_PUBLIC_URL (HTTPS-домен),
 # ADMIN_SESSION_SECRET (openssl rand -hex 32)
 ```
 
-Порт админки (`ADMIN_HTTP_PORT`) у контуров должен различаться: `prod` — `8101`,
-`test` — `8100` (8000 на VPS занят Portainer). `ADMIN_BIND_IP` по умолчанию `127.0.0.1`;
-чтобы nginx-proxy-manager (контейнер) видел админку, ставьте `172.17.0.1`.
+Порт ClickHouse на хосте (`CLICKHOUSE_HTTP_PORT`) у контуров должен различаться:
+`prod` — `8123`, `test` — `8124` (иначе контейнеры конфликтуют на одной VPS).
+Порт админки (`ADMIN_HTTP_PORT`) — так же: `prod` — `8101`, `test` — `8100`
+(8000 на VPS занят Portainer). `ADMIN_BIND_IP` по умолчанию `127.0.0.1`; чтобы
+nginx-proxy-manager (контейнер) видел админку, ставьте `172.17.0.1`.
 
 ## 3. Логин в registry (если пакет приватный)
 
@@ -53,36 +55,35 @@ 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=...)`), аналитика и рассылки — в нём же;
-4. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста).
+3. `clickhouse` поднимается и применяет `clickhouse/init/001_events.sql` —
+   **только при первой инициализации** пустого data-каталога;
+4. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же;
+5. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста).
 
-Проверка контура (сервисы, миграции, Postgres, схема аналитики):
+Проверка контура (сервисы, миграции, Postgres, ClickHouse):
 
 ```bash
 make check E=test
 ```
 
-## Аналитика (Postgres)
+## ClickHouse (аналитика событий)
 
-События и дневные счётчики сообщений пишутся в схему `analytics` той же БД Postgres
-(контракт — [`docs/events.md`](docs/events.md)). Отдельного сервиса нет.
+События всех действий бота пишутся в `analytics.events` (контракт — [`docs/events.md`](docs/events.md)).
+ClickHouse наружу **не публикуется**: только loopback хоста (`127.0.0.1:${CLICKHOUSE_HTTP_PORT}`).
 
 ```bash
-make psql E=test
+make ch E=test      # clickhouse-client контура
 # пример: события за сегодня по типам
-SELECT event_type, count(*) FROM analytics.events
-WHERE event_time >= current_date GROUP BY event_type;
-
-# сообщения по дням
-SELECT day, sum(count) FROM analytics.messages_daily
-WHERE day >= current_date - 7 GROUP BY day ORDER BY day;
+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"'
 ```
 
-Если Postgres недоступен, бот продолжает работать: батч аналитики уходит в stdout
-(WARNING) и отбрасывается. Полностью выключить можно флагом `ANALYTICS_ENABLED=false`.
+Если ClickHouse недоступен, бот продолжает работать: события уходят в stdout (WARNING),
+полностью выключить можно флагом `CLICKHOUSE_ENABLED=false`.
 
-Партиции создаются приложением при старте и раз в 6 часов; старые **не удаляются**.
-Схема создаётся миграцией `V26__analytics_schema.sql`.
+Изменения схемы: `clickhouse/init/*.sql` применяются лишь на пустом data-каталоге,
+поэтому на существующем контуре DDL нужно применить вручную (`make ch E=...`).
 
 ## Админка (FastAPI)
 
@@ -118,11 +119,51 @@ reverse proxy. Сессия живёт 12 часов (`ADMIN_SESSION_MAX_AGE`),
 Роль администратора перечитывается из БД на каждом запросе — отключение пользователя
 (`enabled = false`) или удаление записи сразу закрывает доступ.
 
-## Бэкап
+## Бэкап событий (забирает домашний сервер)
 
-Аналитика теперь лежит в Postgres, поэтому отдельного бэкапа событий больше нет.
-Бэкап БД — стандартными средствами Postgres (`pg_dump`/`pg_basebackup`); при росте
-истории следите за размером `analytics` и диском.
+Сырые события живут на VPS ограниченное время: раз в сутки домашний сервер **сам забирает**
+вчерашнюю партицию (`pull`), складывает её в `Parquet.gz`, импортирует в свой ClickHouse и
+сверяет количество строк — **только после сверки** на VPS удаляются партиции старше
+`BACKUP_GRACE_DAYS` (по умолчанию 7). На VPS остаются агрегаты (`events_daily`, `money_daily`).
+
+Входящий SSH на домашний сервер не нужен: он инициирует соединение сам.
+
+Разовая настройка:
+
+```bash
+# на домашнем сервере
+git clone ssh://git@<gitea-host>:2222/mpakobecex/DispelEvilBot.git
+cd DispelEvilBot
+cp env/backup.env.example env/backup.env   # VPS_SSH, VPS_PORT, VPS_PROJECT_DIR, VPS_SSH_KEY, BACKUP_DIR, CLICKHOUSE_*
+```
+
+Требования: на VPS — пользователь с доступом к `docker` и каталог с репозиторием
+(`VPS_PROJECT_DIR`, там запускается `docker compose`); на домашнем сервере — `clickhouse-client`
+и локальный ClickHouse. Настройка SSH-доступа — `deploy/ssh-access.md`.
+
+Если ставить `clickhouse-client` на хост не хочется, укажите в `env/backup.env` переменную
+`CLICKHOUSE_CLIENT` — путь к обёртке, которая выполняет клиент внутри контейнера:
+
+```bash
+#!/usr/bin/env bash
+set -euo pipefail
+cd "$(dirname "$0")/.."
+exec docker compose --env-file env/backup.env -f docker-compose.yml exec -T clickhouse clickhouse-client "$@"
+```
+
+Проверочный прогон (без cron): `make backup E=test` или `./scripts/clickhouse_backup.sh test`.
+
+Cron на домашнем сервере (вчерашний день уже закрыт, время московское):
+
+```cron
+0 4 * * * cd /srv/DispelEvilBot && ./scripts/clickhouse_backup.sh test >> /var/log/clickhouse-backup-test.log 2>&1
+```
+
+Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и пишет понятный лог.
+Импорт дня идемпотентен: повторный прогон не дублирует строки.
+
+Если контур на VPS не запущен (нет работающего `clickhouse` или не найден каталог проекта),
+скрипт пишет об этом в лог и завершается с кодом 0 — для cron это не ошибка.
 
 ## Обновление и откат
 
@@ -143,13 +184,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 поднимается отдельно и читает Postgres (схема `analytics`) с полной историей
-событий. Отдельный read-only пользователь Postgres и сам дашборд — финальный этап,
-в этот контур не входят.
+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   бэкап событий с VPS (запускать на домашнем сервере)"
 
 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

@@ -9,9 +9,10 @@ Telegram-бот клана Dispel Evil (aiogram 3 + Postgres). Монорепо
 common/     # общий пакет: config, logging, db, repositories, models, texts, analytics
 bot/        # сервис бота: handlers, services, keyboards, middlewares, notifications
 admin/      # сервис админки: FastAPI + Jinja2/HTMX (Блок 4), свой Dockerfile
-sql/        # миграции Flyway (V__*.sql, R__*.sql), включая схему analytics
+sql/        # миграции Flyway (V__*.sql, R__*.sql)
+clickhouse/ # init-скрипты ClickHouse (схема аналитики)
 docs/       # документация (docs/events.md — события аналитики)
-scripts/    # эксплуатационные скрипты (проверка контура)
+scripts/    # эксплуатационные скрипты (проверка контура, бэкап событий)
 images/     # статичные картинки (приветствие)
 deploy/     # материалы деплоя (Caddyfile, инструкции)
 tests/      # тесты
@@ -23,7 +24,8 @@ docker-compose.yml / docker-compose.prod.yml
 ## Требования
 
 - Python 3.12
-- PostgreSQL 14+ (аналитика — в схеме `analytics`)
+- PostgreSQL 14+
+- ClickHouse 25.3 (аналитика событий; можно выключить — `CLICKHOUSE_ENABLED=false`)
 
 ## Настройка
 
@@ -86,10 +88,10 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR
 видны боту (он читает `bot.titles`). Удалить можно только звание, которое ни у кого
 не стоит; используемое — скрывается (`visible = false`).
 
-Раздел «Логи» — события из Postgres и отдельная вкладка «Сообщения по дням»: фильтры
-по типу (для событий), пользователю (ID или username) и датам, пагинация по 50.
-Время показывается в `TIMEZONE`, payload раскрывается как JSON, ID пользователя ведёт
-в его карточку. Если Postgres недоступен, страница показывает ошибку, а не падает.
+Раздел «Логи» — события из ClickHouse: фильтры по типу, пользователю (ID или username)
+и датам, пагинация по 50. Время показывается в `TIMEZONE`, payload раскрывается как JSON,
+ID пользователя ведёт в его карточку. ClickHouse читается отдельным ленивым клиентом;
+если он недоступен, страница показывает ошибку, а не падает.
 
 Раздел «Админы» — выдача и отзыв доступа к панели по Telegram-ID, роли и вкл/выкл.
 В карточке пользователя есть исключение из чата (бан + разбан) с удалением данных и
@@ -111,19 +113,16 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR
 редактором, картинка и URL-кнопки, предпросмотр и подтверждение. Каждая отправка
 (успешная или нет) пишется в `bot.admin_messages`.
 
-## Аналитика (Postgres)
+## Аналитика (ClickHouse)
 
-Все значимые действия бота пишутся в Postgres, схема `analytics`:
-`analytics.events` (события, партиции по дням) и `analytics.messages_daily`
-(количество сообщений пользователя по дням). Контракт событий —
-[`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
+Все значимые действия бота пишутся в ClickHouse в единую таблицу `analytics.events`.
+Контракт событий — [`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
 
-- схема — `sql/V26__analytics_schema.sql`;
-- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой Postgres не влияет на бота);
-- сообщения не пишутся поштучно: на каждое сообщение инкрементится дневной счётчик;
-- построчная `bot.messages` удалена (V27), старые данные перенесены в `analytics.messages_daily`;
-- партиции создаются приложением и **не удаляются** — история хранится бессрочно;
-- выключается флагом `ANALYTICS_ENABLED=false`.
+- схема — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера);
+- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой ClickHouse не влияет на бота);
+- сырые события живут на VPS 7 дней и ежедневно уезжают архивом на домашний сервер,
+  на VPS остаются агрегаты (`events_daily`, `money_daily`) — см. [DEPLOY.md](DEPLOY.md);
+- выключается флагом `CLICKHOUSE_ENABLED=false`.
 
 ## Уведомления в чат логов
 
@@ -132,7 +131,7 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR
 
 ## Docker
 
-Локальный запуск (postgres + migrate + bot + admin):
+Локальный запуск (postgres + migrate + clickhouse + bot + admin):
 
 ```bash
 cp .env.example .env      # заполнить
@@ -163,17 +162,22 @@ black --check .
 pytest
 ```
 
-## Бэкап
+## Бэкап событий
 
-Аналитика теперь в Postgres, отдельного бэкапа событий нет. Бэкап БД — стандартными
-средствами Postgres; история `analytics` не чистится, поэтому следите за размером БД
-и диском. Подробности — [`DEPLOY.md`](DEPLOY.md).
+Сырые события ClickHouse ежедневно забираются с VPS домашним сервером (`pull`), на VPS
+остаются только агрегаты и последние дни сырых данных:
+
+```bash
+make backup E=test     # на домашнем сервере: ./scripts/clickhouse_backup.sh test
+```
+
+Настройка, cron и SSH-доступ — см. [`DEPLOY.md`](DEPLOY.md).
 
 ## Версионирование
 
-- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.7.1`.
+- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.9`.
 - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
-- Docker-образы тегируются версией (Блок 2), события аналитики получают `bot_version`.
+- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
 - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.
 
 ## Процесс работы с git
@@ -227,5 +231,6 @@ Telegram не принимает кириллицу в именах команд
 
 ## Переменные окружения
 
-- общие — `.env.example` (включая `ANALYTICS_*`);
-- контуры — `env/prod.env.example`, `env/test.env.example`.
+- общие — `.env.example` (включая `CLICKHOUSE_*` и параметры батчинга `ANALYTICS_*`);
+- контуры — `env/prod.env.example`, `env/test.env.example`;
+- бэкап событий — `env/backup.env.example`.

admin/app.py

@@ -16,7 +16,7 @@ from admin.media import ensure_media
 from admin.middleware import AdminAuthMiddleware
 from admin.routes import admins, auth, health, logs, pages, preview, schedules, send, sequences, titles, users
 from common import __version__
-from common.analytics.partitions import ensure_partitions
+from common.analytics import read as analytics_read
 from common.config import AdminSettings, get_admin_settings
 from common.db.pool import close_pool, init_pool
 from common.logging_config import setup_logging
@@ -42,10 +42,6 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) -
         if init_db:
             await init_pool(app.state.settings)
             ensure_media(app.state.settings)
-            try:
-                await ensure_partitions(app.state.settings)
-            except Exception:
-                logger.warning("Не удалось создать партиции аналитики", exc_info=True)
             try:
                 inserted = await admins_repo.ensure_admins(app.state.settings.admin_ids)
                 if inserted:
@@ -60,6 +56,7 @@ def create_app(settings: AdminSettings | None = None, *, init_db: bool = True) -
         finally:
             if init_db:
                 await close_pool()
+            await analytics_read.close_client()
 
     app = FastAPI(
         title="DispelEvilBot Admin",

admin/routes/logs.py

@@ -1,4 +1,4 @@
-"""Раздел «Логи»: события и дневные счётчики сообщений из Postgres."""
+"""Раздел «Логи»: события из ClickHouse с фильтрами и пагинацией."""
 
 from __future__ import annotations
 
@@ -19,18 +19,16 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/logs", tags=["logs"])
 
-#: Сколько строк показывать на странице.
+#: Сколько событий показывать на странице.
 PAGE_SIZE = 50
 
 
-def _pretty_payload(payload: Any) -> str:
+def _pretty_payload(payload: str) -> str:
     """Отформатировать JSON-payload для просмотра; битый JSON — как есть."""
-    if isinstance(payload, dict):
-        return json.dumps(payload, ensure_ascii=False, indent=2)
     try:
-        return json.dumps(json.loads(str(payload or "")), ensure_ascii=False, indent=2)
+        return json.dumps(json.loads(payload), ensure_ascii=False, indent=2)
     except (TypeError, ValueError):
-        return str(payload or "")
+        return payload or ""
 
 
 def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]:
@@ -46,85 +44,66 @@ def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]:
     return {
         **row,
         "time_local": time_local,
-        "payload_pretty": _pretty_payload(row.get("payload")),
-    }
-
-
-def _decorate_message(row: dict[str, Any]) -> dict[str, Any]:
-    day = row.get("day")
-    return {
-        **row,
-        "day_local": day.strftime("%d.%m.%Y") if hasattr(day, "strftime") else str(day or "—"),
+        "payload_pretty": _pretty_payload(str(row.get("payload") or "")),
     }
 
 
 @router.get("", response_class=HTMLResponse)
 async def logs_page(
     request: Request,
-    view: str = "events",
     event_type: str = "",
     user: str = "",
     date_from: str = "",
     date_to: str = "",
     page: int = 1,
 ) -> HTMLResponse:
-    """Показать события или дневные счётчики сообщений с фильтрами."""
+    """Показать события с фильтрами по типу, пользователю и датам."""
     settings = request.app.state.settings
-    view = view if view in {"events", "messages"} else "events"
     today = datetime.now(settings.tzinfo).date()
     date_from = date_from or (today - timedelta(days=6)).isoformat()
     date_to = date_to or today.isoformat()
     page = max(page, 1)
 
     events: list[dict[str, Any]] = []
-    messages: list[dict[str, Any]] = []
     total = 0
     error: str | None = None
 
     try:
-        if view == "messages":
-            messages, total = await analytics_read.fetch_messages(
-                settings,
-                user=user,
-                date_from=date_from,
-                date_to=date_to,
-                limit=PAGE_SIZE,
-                offset=(page - 1) * PAGE_SIZE,
-            )
-        else:
-            events, total = await analytics_read.fetch_events(
-                settings,
-                event_type=event_type,
-                user=user,
-                date_from=date_from,
-                date_to=date_to,
-                limit=PAGE_SIZE,
-                offset=(page - 1) * PAGE_SIZE,
-            )
+        events, total = await analytics_read.fetch_events(
+            settings,
+            event_type=event_type,
+            user=user,
+            date_from=date_from,
+            date_to=date_to,
+            limit=PAGE_SIZE,
+            offset=(page - 1) * PAGE_SIZE,
+        )
     except ValueError:
         error = "Даты должны быть в формате ГГГГ-ММ-ДД."
     except Exception:
-        logger.exception("Не удалось прочитать аналитику из Postgres")
-        error = "Postgres недоступен — логи временно не читаются."
+        logger.exception("Не удалось прочитать события из ClickHouse")
+        error = "ClickHouse недоступен — логи временно не читаются."
 
     pages = max((total + PAGE_SIZE - 1) // PAGE_SIZE, 1)
-    query_params = {
-        "view": view if view != "events" else "",
-        "event_type": event_type if view == "events" else "",
-        "user": user,
-        "date_from": date_from,
-        "date_to": date_to,
-    }
-    query_base = urlencode({key: value for key, value in query_params.items() if value})
+    query_base = urlencode(
+        {
+            key: value
+            for key, value in {
+                "event_type": event_type,
+                "user": user,
+                "date_from": date_from,
+                "date_to": date_to,
+            }.items()
+            if value
+        }
+    )
 
     return request.app.state.templates.TemplateResponse(
         request=request,
         name="logs/list.html",
         context={
             "version": __version__,
-            "view": view,
             "events": [_decorate(row, settings) for row in events],
-            "messages": [_decorate_message(row) for row in messages],
             "event_types": sorted(ALL_EVENT_TYPES),
             "event_type": event_type,
             "user": user,

admin/routes/pages.py

@@ -38,7 +38,7 @@ SECTIONS: tuple[dict[str, str | None], ...] = (
     },
     {
         "title": "Логи",
-        "description": "События и сообщения из Postgres с фильтрами",
+        "description": "События из ClickHouse с фильтрами",
         "url": "/logs",
     },
     {

admin/templates/logs/list.html

@@ -6,33 +6,11 @@
 <div class="flex flex-wrap items-end justify-between gap-4">
   <div>
     <h1 class="text-2xl font-semibold">Логи</h1>
-    <p class="mt-1 text-sm text-slate-600">
-      {% if view == "messages" %}Сообщения по дням{% else %}События из Postgres{% endif %}
-      · найдено: {{ total }}
-    </p>
+    <p class="mt-1 text-sm text-slate-600">События из ClickHouse · найдено: {{ total }}</p>
   </div>
 </div>
 
-<div class="mt-4 flex gap-1 border-b border-slate-200">
-  <a
-    href="/logs?date_from={{ date_from }}&date_to={{ date_to }}"
-    class="rounded-t-lg px-4 py-2 text-sm {% if view == 'events' %}border border-b-0 border-slate-200 bg-white font-medium text-slate-900{% else %}text-slate-600 hover:text-slate-900{% endif %}"
-  >
-    События
-  </a>
-  <a
-    href="/logs?view=messages&date_from={{ date_from }}&date_to={{ date_to }}"
-    class="rounded-t-lg px-4 py-2 text-sm {% if view == 'messages' %}border border-b-0 border-slate-200 bg-white font-medium text-slate-900{% else %}text-slate-600 hover:text-slate-900{% endif %}"
-  >
-    Сообщения по дням
-  </a>
-</div>
-
-<form method="get" action="/logs" class="flex flex-wrap items-end gap-3 rounded-b-lg border border-t-0 border-slate-200 bg-white p-4">
-  {% if view == "messages" %}
-  <input type="hidden" name="view" value="messages" />
-  {% endif %}
-  {% if view == "events" %}
+<form method="get" action="/logs" class="mt-4 flex flex-wrap items-end gap-3 rounded-lg border border-slate-200 bg-white p-4">
   <div>
     <label for="event_type" class="text-xs text-slate-500">Тип</label>
     <select
@@ -46,7 +24,6 @@
       {% endfor %}
     </select>
   </div>
-  {% endif %}
   <div>
     <label for="user" class="text-xs text-slate-500">Пользователь</label>
     <input
@@ -54,7 +31,7 @@
       name="user"
       type="text"
       value="{{ user }}"
-      placeholder="ID или username"
+      placeholder="ID или @username"
       class="mt-1 block w-48 rounded-lg border border-slate-300 px-3 py-2 text-sm focus:border-sky-500 focus:outline-none"
     />
   </div>
@@ -81,46 +58,13 @@
   <button type="submit" class="rounded-lg bg-sky-600 px-4 py-2 text-sm font-medium text-white hover:bg-sky-700">
     Показать
   </button>
-  <a href="/logs{% if view == 'messages' %}?view=messages{% endif %}" class="text-sm text-slate-600 hover:underline">Сбросить</a>
+  <a href="/logs" class="text-sm text-slate-600 hover:underline">Сбросить</a>
 </form>
 
 {% if error %}
 <div class="mt-4 rounded-lg border border-amber-200 bg-amber-50 p-3 text-sm text-amber-800">{{ error }}</div>
 {% endif %}
 
-{% if view == "messages" %}
-<div class="mt-6 overflow-x-auto rounded-lg border border-slate-200 bg-white">
-  <table class="min-w-full divide-y divide-slate-200 text-sm">
-    <thead class="bg-slate-50 text-left text-slate-600">
-      <tr>
-        <th class="px-4 py-3 font-medium">День</th>
-        <th class="px-4 py-3 font-medium">Пользователь</th>
-        <th class="hidden px-4 py-3 font-medium lg:table-cell">Чат</th>
-        <th class="px-4 py-3 text-right font-medium">Сообщений</th>
-      </tr>
-    </thead>
-    <tbody class="divide-y divide-slate-100">
-      {% for row in messages %}
-      <tr class="hover:bg-slate-50">
-        <td class="whitespace-nowrap px-4 py-3 text-slate-600">{{ row.day_local }}</td>
-        <td class="px-4 py-3">
-          {% if row.username %}<div>@{{ row.username }}</div>{% endif %}
-          <a href="/users/{{ row.user_id }}" class="font-mono text-xs text-sky-700 hover:underline">{{ row.user_id }}</a>
-        </td>
-        <td class="hidden px-4 py-3 font-mono text-xs text-slate-500 lg:table-cell">{{ row.chat_id or "—" }}</td>
-        <td class="px-4 py-3 text-right font-mono">{{ row.count }}</td>
-      </tr>
-      {% else %}
-      <tr>
-        <td colspan="4" class="px-4 py-6 text-center text-slate-500">
-          {% if error %}Данные не загрузились.{% else %}За выбранный период сообщений нет.{% endif %}
-        </td>
-      </tr>
-      {% endfor %}
-    </tbody>
-  </table>
-</div>
-{% else %}
 <div class="mt-6 overflow-x-auto rounded-lg border border-slate-200 bg-white">
   <table class="min-w-full divide-y divide-slate-200 text-sm">
     <thead class="bg-slate-50 text-left text-slate-600">
@@ -172,7 +116,6 @@
     </tbody>
   </table>
 </div>
-{% endif %}
 
 {% if pages > 1 %}
 <div class="mt-4 flex items-center justify-between text-sm text-slate-600">

admin/templates/users/card.html

@@ -68,7 +68,7 @@
     <div class="text-xs uppercase tracking-wide text-slate-500">Сообщения</div>
     <div class="mt-1 text-xl font-semibold">{{ user.msg_count }}</div>
     <div class="text-xs text-slate-500">
-      последняя активность: {{ user.last_message_at.strftime("%d.%m.%Y") if user.last_message_at else "—" }}
+      последнее: {{ user.last_message_at.strftime("%d.%m.%Y %H:%M") if user.last_message_at else "—" }}
     </div>
   </div>
   <div class="rounded-lg border border-slate-200 bg-white p-4">

bot/__main__.py

@@ -12,7 +12,6 @@ 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.analytics.partitions import ensure_partitions, run_partitions
 from common.config import get_settings
 from common.db.pool import close_pool, init_pool
 from common.logging_config import setup_logging
@@ -27,10 +26,6 @@ async def main() -> None:
 
     await init_pool()
     await init_analytics()
-    try:
-        await ensure_partitions(settings)
-    except Exception:
-        logger.warning("Не удалось создать партиции аналитики", exc_info=True)
     record(EventType.BOT_START, payload={"app_env": settings.app_env})
 
     bot = create_bot()
@@ -41,20 +36,13 @@ async def main() -> None:
 
     notifications_task = asyncio.create_task(run_notifications(bot))
     heartbeat_task = asyncio.create_task(run_heartbeat(bot))
-    partitions_task = asyncio.create_task(run_partitions(settings))
     try:
         await bot.delete_webhook(drop_pending_updates=True)
         await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
     finally:
         notifications_task.cancel()
         heartbeat_task.cancel()
-        partitions_task.cancel()
-        await asyncio.gather(
-            notifications_task,
-            heartbeat_task,
-            partitions_task,
-            return_exceptions=True,
-        )
+        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()

bot/middlewares/analytics.py

@@ -9,8 +9,8 @@ from typing import Any
 from aiogram import BaseMiddleware
 from aiogram.types import Message, TelegramObject
 
-from common.analytics import EventType, record, record_message, remember_user
-from common.analytics.messages import describe_command
+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__)
@@ -43,7 +43,13 @@ class AnalyticsMiddleware(BaseMiddleware):
 
         try:
             remember_user(user.id, user.username)
-            record_message(user_id=user.id, username=user.username, chat_id=message.chat.id)
+            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(

bot/services/heartbeat.py

@@ -4,7 +4,7 @@
 если файл свежий. Если event loop завис — heartbeat перестаёт обновляться.
 
 Раз в час в чат логов (``ALERT_CHAT_ID``) уходит короткое сообщение о работе бота,
-и в аналитику пишется событие ``bot_alive``.
+и в ClickHouse пишется событие ``bot_alive``.
 """
 
 from __future__ import annotations

clickhouse/config.d/disable-trace-log.xml

@@ -0,0 +1,8 @@
+<clickhouse>
+    <!--
+      system.trace_log отключён: на тестовом контуре он генерировал
+      ~1.3 ГБ в сутки (~6.6 ГБ за 5 дней). Профилировщик здесь не нужен.
+      Чтобы вернуть: удалить этот файл и перезапустить контейнер.
+    -->
+    <trace_log remove="1"/>
+</clickhouse>

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

@@ -1,4 +1,4 @@
-"""Аналитика: клиент Postgres и каталог событий.
+"""Аналитика: клиент ClickHouse и каталог событий.
 
 Публичный интерфейс:
 
@@ -16,7 +16,6 @@ from common.analytics.client import (
     get_analytics,
     init_analytics,
     record,
-    record_message,
     remember_user,
 )
 from common.analytics.events import ALL_EVENT_TYPES, EVENT_COLUMNS, EventType
@@ -30,6 +29,5 @@ __all__ = [
     "get_analytics",
     "init_analytics",
     "record",
-    "record_message",
     "remember_user",
 ]

common/analytics/client.py

@@ -1,10 +1,13 @@
-"""Аналитика в Postgres: батч-запись событий и дневных счётчиков сообщений.
+"""Клиент ClickHouse: батч-запись событий без блокировки бота.
 
 Свойства:
-- запись неблокирующая: ``record``/``record_message`` только кладут данные в буфер;
-- батчинг по размеру/таймеру;
-- сбой Postgres не роняет бота: батч уходит в лог (fallback) и отбрасывается;
-- сообщения не пишутся поштучно: инкрементится ``analytics.messages_daily``.
+- запись неблокирующая: :meth:`AnalyticsClient.record` только кладёт строку в буфер;
+- батчинг: отправка по размеру буфера и по таймеру;
+- недоступность ClickHouse **не роняет бота**: события попадают в stdout (fallback);
+- graceful-flush при остановке (:func:`close_analytics`).
+
+Импорт ``clickhouse_connect`` ленивый (внутри :meth:`AnalyticsClient._ensure_client`),
+поэтому пакет не является обязательным для импорта модуля (нужно для тестов/админки).
 """
 
 from __future__ import annotations
@@ -14,15 +17,11 @@ import contextlib
 import json
 import logging
 from collections import deque
-from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime
-from typing import Any
+from typing import Any, Protocol
 
-from psycopg.types.json import Jsonb
-
-from common.analytics.events import EventType
+from common.analytics.events import EVENT_COLUMNS, EventType
 from common.config import Settings, get_settings
-from common.db.pool import get_pool
 from common.version import __version__
 
 logger = logging.getLogger(__name__)
@@ -36,90 +35,90 @@ USERNAME_CACHE_MAX = 5000
 #: Сколько строк события выводить в лог при fallback.
 FALLBACK_SAMPLE = 5
 
-#: Строка события: (time, type, user_id, username, chat_id, payload, version, prod).
-EventRow = list[Any]
-#: (day, user_id, chat_id) -> (username, count)
-MessageCounts = dict[tuple[Any, int, int], list[Any]]
 
-#: Куда сбрасывать батч (подменяется в тестах).
-Writer = Callable[[list[EventRow], MessageCounts], Awaitable[None]]
+class ClickHouseClient(Protocol):
+    """Минимальный контракт клиента ClickHouse (используется и для подмены в тестах)."""
 
+    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> Any:
+        """Вставить строки в таблицу."""
 
-async def _write_db(events: list[EventRow], messages: MessageCounts) -> None:
-    """Записать батч в Postgres."""
-    async with get_pool().connection() as conn:
-        if events:
-            await conn.executemany(
-                """
-                INSERT INTO analytics.events
-                    (event_time, event_type, user_id, username, chat_id, payload, bot_version, prod)
-                VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
-                """,
-                [
-                    [
-                        row[0],
-                        row[1],
-                        row[2],
-                        row[3],
-                        row[4],
-                        Jsonb(row[5]),
-                        row[6],
-                        row[7],
-                    ]
-                    for row in events
-                ],
-            )
-
-        for (day, user_id, chat_id), (username, count) in messages.items():
-            await conn.execute(
-                """
-                INSERT INTO analytics.messages_daily (day, user_id, username, chat_id, count)
-                VALUES (%s, %s, %s, %s, %s)
-                ON CONFLICT (day, user_id, chat_id) DO UPDATE
-                SET count = analytics.messages_daily.count + EXCLUDED.count,
-                    username = EXCLUDED.username
-                """,
-                (day, user_id, username, chat_id, count),
-            )
+    async def close(self) -> None:
+        """Закрыть соединения."""
 
 
 class AnalyticsClient:
-    """Батч-запись событий и счётчиков в Postgres с деградацией в лог."""
+    """Батч-запись событий в ClickHouse с деградацией в лог."""
 
-    def __init__(
-        self,
-        settings: Settings | None = None,
-        *,
-        writer: Writer | None = None,
-    ) -> None:
+    def __init__(self, settings: Settings | None = None, *, client: ClickHouseClient | None = None) -> None:
         self._settings = settings or get_settings()
-        self._writer = writer or _write_db
-        self._rows: deque[EventRow] = deque()
-        self._messages: MessageCounts = {}
+        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:
-        return self._settings.analytics_enabled
+        """Включена ли аналитика (``CLICKHOUSE_ENABLED``)."""
+        return self._settings.clickhouse_enabled
 
     @property
     def pending(self) -> int:
-        return len(self._rows) + sum(int(value[1]) for value in self._messages.values())
+        """Сколько событий ждёт отправки (для тестов и диагностики)."""
+        return len(self._rows)
 
     async def start(self) -> None:
+        """Подключиться к ClickHouse (best-effort) и запустить фоновый сброс буфера."""
         if not self.enabled:
-            logger.info("Аналитика выключена (ANALYTICS_ENABLED=false)")
+            logger.info("Аналитика ClickHouse выключена (CLICKHOUSE_ENABLED=false)")
             return
+
+        await self._ensure_client()
         self._task = asyncio.create_task(self._run(), name="analytics-flush")
         logger.info(
-            "Аналитика Postgres: батч %s, интервал %sс",
+            "Аналитика 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:
@@ -144,41 +143,26 @@ class AnalyticsClient:
 
         self._rows.append(
             [
-                datetime.now(UTC),
+                datetime.now(UTC).replace(tzinfo=None),
                 str(event_type),
                 int(user_id),
                 username or "",
                 int(chat_id),
-                self._payload(payload),
+                self._dump_payload(payload),
                 __version__,
-                bool(self._settings.is_prod),
+                1 if self._settings.is_prod else 0,
             ]
         )
 
         while len(self._rows) > self._settings.analytics_buffer_max:
             self._rows.popleft()
+            self._overflow += 1
 
-        if self.pending >= self._settings.analytics_batch_size:
-            self._flush_now.set()
-
-    def record_message(self, *, user_id: int, username: str | None, chat_id: int) -> None:
-        """Увеличить дневной счётчик сообщений пользователя."""
-        if not self.enabled or not user_id:
-            return
-
-        day = datetime.now(self._settings.tzinfo).date()
-        key = (day, int(user_id), int(chat_id))
-        entry = self._messages.get(key)
-        if entry is None:
-            self._messages[key] = [username or self._usernames.get(user_id, ""), 1]
-        else:
-            entry[0] = username or entry[0]
-            entry[1] = int(entry[1]) + 1
-
-        if self.pending >= self._settings.analytics_batch_size:
+        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):
@@ -187,74 +171,105 @@ class AnalyticsClient:
             await self.flush()
 
     async def flush(self) -> None:
-        """Отправить накопленное. При сбое — fallback в лог (без исключений)."""
-        if not self._rows and not self._messages:
+        """Отправить накопленные события. При сбое — fallback в лог (без исключений)."""
+        overflow, self._overflow = self._overflow, 0
+        if overflow:
+            logger.warning("Буфер событий переполнен, потеряно %s событий (не записаны в ClickHouse)", overflow)
+
+        if not self._rows:
             return
 
-        events = list(self._rows)
-        messages = self._messages
-        self._rows.clear()
-        self._messages = {}
-
         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 self._writer(events, messages)
+            await client.insert(self._table, rows, list(EVENT_COLUMNS))
         except Exception:
-            self._log_fallback(events, messages)
+            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
-        await self.flush()
 
-    def _log_fallback(self, events: list[EventRow], messages: MessageCounts) -> None:
-        sample = " | ".join(
-            f"{row[0]:%Y-%m-%d %H:%M:%S} {row[1]} user={row[2]} chat={row[4]}" for row in events[:FALLBACK_SAMPLE]
+        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: не записано в Postgres событий=%s, счётчиков=%s%s",
-            len(events),
-            len(messages),
-            f": {sample}" if sample else "",
+            "Fallback (%s): %s событий не записаны в ClickHouse: %s%s",
+            reason,
+            len(rows),
+            lines,
+            " ..." if len(rows) > FALLBACK_SAMPLE else "",
         )
 
     @staticmethod
-    def _payload(payload: dict[str, Any] | None) -> dict[str, Any]:
+    def _dump_payload(payload: dict[str, Any] | None) -> str:
+        """Сериализовать payload в JSON-строку с обрезкой по длине."""
         if not payload:
-            return {}
+            return "{}"
         try:
             text = json.dumps(payload, ensure_ascii=False, default=str)
         except (TypeError, ValueError):
-            return {"_error": "payload не сериализуется"}
-        if len(text) > PAYLOAD_MAX_LENGTH:
-            return {"_truncated": text[:PAYLOAD_MAX_LENGTH]}
-        # Пересобираем из JSON: так в jsonb не попадёт объект, который не сериализуется.
-        return json.loads(text)
+            text = json.dumps({"_error": "payload не сериализуется"})
+        return text[:PAYLOAD_MAX_LENGTH]
 
 
+#: Активный клиент (если аналитика не инициализирована — None).
 _client: AnalyticsClient | None = None
 
 
 async def init_analytics(
     settings: Settings | None = None,
     *,
-    writer: Writer | None = None,
+    client: ClickHouseClient | None = None,
 ) -> AnalyticsClient:
     """Создать и запустить единственный клиент аналитики. Идемпотентно."""
     global _client
     if _client is not None:
         return _client
-    _client = AnalyticsClient(settings, writer=writer)
+    _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()
@@ -262,6 +277,7 @@ async def close_analytics() -> None:
 
 
 def get_analytics() -> AnalyticsClient | None:
+    """Вернуть активный клиент аналитики (или None)."""
     return _client
 
 
@@ -273,17 +289,13 @@ def record(
     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 record_message(*, user_id: int, username: str | None, chat_id: int) -> None:
-    if _client is None:
-        return
-    _client.record_message(user_id=user_id, username=username, chat_id=chat_id)
-
-
 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

@@ -1,6 +1,6 @@
 """Каталог событий аналитики.
 
-Единственный источник правды по типам событий. Список синхронизирован
+Единственный источник правды по типам событий ClickHouse. Список синхронизирован
 с документацией (``docs/events.md``) — рассинхрон ловит ``tests/test_analytics.py``.
 
 Правило: добавляешь новое событие — добавь его в ``EventType`` и в ``docs/events.md``.
@@ -12,9 +12,10 @@ from enum import StrEnum
 
 
 class EventType(StrEnum):
-    """Типы событий (значение = колонка ``event_type`` в Postgres)."""
+    """Типы событий (значение = колонка ``event_type`` в ClickHouse)."""
 
     # --- Сообщения и команды ---
+    MESSAGE = "message"
     COMMAND = "command"
 
     # --- Экономика ---

common/analytics/partitions.py

@@ -1,64 +0,0 @@
-"""Партиции аналитики по дням.
-
-Создаются заранее (вчера/сегодня/два дня вперёд), старые **не удаляются**:
-история событий и дневных счётчиков хранится бессрочно.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-from datetime import UTC, date, datetime, timedelta
-
-from common.config import Settings
-from common.db.pool import get_pool
-
-logger = logging.getLogger(__name__)
-
-#: Сколько дней вперёд создавать партиции.
-DAYS_AHEAD = 2
-#: Сколько дней назад проверять (на случай простоя).
-DAYS_BACK = 1
-#: Интервал фоновой проверки партиций.
-PARTITION_INTERVAL = 6 * 60 * 60
-
-
-def _event_partition(day: date) -> str:
-    start = f"{day.isoformat()} 00:00:00+00"
-    end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00+00"
-    return (
-        f"CREATE TABLE IF NOT EXISTS analytics.events_{day:%Y%m%d} "
-        f"PARTITION OF analytics.events FOR VALUES FROM ('{start}') TO ('{end}')"
-    )
-
-
-def _messages_partition(day: date) -> str:
-    start = day.isoformat()
-    end = (day + timedelta(days=1)).isoformat()
-    return (
-        f"CREATE TABLE IF NOT EXISTS analytics.messages_daily_{day:%Y%m%d} "
-        f"PARTITION OF analytics.messages_daily FOR VALUES FROM ('{start}') TO ('{end}')"
-    )
-
-
-async def ensure_partitions(settings: Settings) -> None:
-    """Создать недостающие партиции на несколько дней вперёд."""
-    utc_today = datetime.now(UTC).date()
-    local_today = datetime.now(settings.tzinfo).date()
-
-    async with get_pool().connection() as conn:
-        for offset in range(-DAYS_BACK, DAYS_AHEAD + 1):
-            await conn.execute(_event_partition(utc_today + timedelta(days=offset)))
-            await conn.execute(_messages_partition(local_today + timedelta(days=offset)))
-
-
-async def run_partitions(settings: Settings) -> None:
-    """Периодически проверять партиции; ошибки не роняют бота."""
-    while True:
-        try:
-            await ensure_partitions(settings)
-        except asyncio.CancelledError:
-            raise
-        except Exception:
-            logger.warning("Не удалось создать партиции аналитики", exc_info=True)
-        await asyncio.sleep(PARTITION_INTERVAL)

common/analytics/read.py

@@ -1,14 +1,17 @@
-"""Чтение аналитики из Postgres для админки."""
+"""Чтение событий ClickHouse для админки (Блок 4.3).
+
+Клиент создаётся лениво и только на чтение; недоступность ClickHouse не роняет
+админку — страница покажет ошибку. Запись событий живёт в ``client.py``.
+"""
 
 from __future__ import annotations
 
 import logging
-from datetime import UTC, date, datetime, time, timedelta
+from datetime import UTC, datetime, time, timedelta
 from typing import Any
 
 from common.analytics.events import ALL_EVENT_TYPES
 from common.config import Settings
-from common.db.pool import get_pool
 
 logger = logging.getLogger(__name__)
 
@@ -24,29 +27,62 @@ EVENT_FIELDS = (
     "prod",
 )
 
-
-def _dates(date_from: str, date_to: str, settings: Settings) -> tuple[date, date]:
-    today = datetime.now(settings.tzinfo).date()
-    start = datetime.strptime(date_from, "%Y-%m-%d").date() if date_from else today - timedelta(days=6)
-    end = datetime.strptime(date_to, "%Y-%m-%d").date() if date_to else today
-    return start, end
+_client: Any | None = None
 
 
-def _event_bounds(start: date, end: date, settings: Settings) -> tuple[datetime, datetime]:
-    start_dt = datetime.combine(start, time.min, tzinfo=settings.tzinfo).astimezone(UTC)
-    end_dt = datetime.combine(end + timedelta(days=1), time.min, tzinfo=settings.tzinfo).astimezone(UTC)
-    return start_dt, end_dt
+async def _ensure_client(settings: Settings) -> Any:
+    """Создать (один раз) асинхронный клиент ClickHouse."""
+    global _client
+    if _client is not None:
+        return _client
+
+    import clickhouse_connect
+
+    _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,
+    )
+    return _client
 
 
-def _user_clause(user: str, params: list[Any], column: str = "user_id") -> str:
+async def close_client() -> None:
+    """Закрыть клиент при остановке админки."""
+    global _client
+    if _client is not None:
+        client, _client = _client, None
+        try:
+            await client.close()
+        except Exception:
+            logger.warning("Не удалось закрыть клиент ClickHouse", exc_info=True)
+
+
+def _where(
+    *,
+    event_type: str,
+    user: str,
+    start: datetime,
+    end: datetime,
+) -> tuple[str, dict[str, Any]]:
+    clauses = ["event_time >= {start:DateTime}", "event_time < {end:DateTime}"]
+    params: dict[str, Any] = {"start": start, "end": end}
+
+    if event_type and event_type in ALL_EVENT_TYPES:
+        clauses.append("event_type = {event_type:String}")
+        params["event_type"] = event_type
+
     user = user.strip()
-    if not user:
-        return ""
-    if user.isdigit():
-        params.append(int(user))
-        return f" AND {column} = %s"
-    params.append(f"%{user}%")
-    return " AND username ILIKE %s"
+    if user:
+        if user.isdigit():
+            clauses.append("user_id = {user_id:Int64}")
+            params["user_id"] = int(user)
+        else:
+            clauses.append("positionCaseInsensitive(username, {user_text:String}) > 0")
+            params["user_text"] = user
+
+    return " WHERE " + " AND ".join(clauses), params
 
 
 async def fetch_events(
@@ -59,71 +95,31 @@ async def fetch_events(
     limit: int = 50,
     offset: int = 0,
 ) -> tuple[list[dict[str, Any]], int]:
-    """Вернуть события и общее количество по фильтрам."""
-    start, end = _dates(date_from, date_to, settings)
-    start_dt, end_dt = _event_bounds(start, end, settings)
+    """Вернуть события и общее количество по фильтрам.
 
-    params: list[Any] = [start_dt, end_dt]
-    where = " WHERE event_time >= %s AND event_time < %s"
-    if event_type and event_type in ALL_EVENT_TYPES:
-        where += " AND event_type = %s"
-        params.append(event_type)
-    where += _user_clause(user, params)
+    Даты — ``YYYY-MM-DD`` в таймзоне админки; внутри переводятся в UTC.
+    """
+    today = datetime.now(settings.tzinfo).date()
+    start_date = datetime.strptime(date_from, "%Y-%m-%d").date() if date_from else today - timedelta(days=6)
+    end_date = datetime.strptime(date_to, "%Y-%m-%d").date() if date_to else today
 
-    async with get_pool().connection() as conn:
-        cur = await conn.execute(f"SELECT count(*) AS total FROM analytics.events{where}", tuple(params))
-        row = await cur.fetchone()
-        total = int(row["total"]) if row else 0
+    start = datetime.combine(start_date, time.min, tzinfo=settings.tzinfo).astimezone(UTC).replace(tzinfo=None)
+    end = (
+        datetime.combine(end_date + timedelta(days=1), time.min, tzinfo=settings.tzinfo)
+        .astimezone(UTC)
+        .replace(tzinfo=None)
+    )
 
-        cur = await conn.execute(
-            f"SELECT {', '.join(EVENT_FIELDS)} FROM analytics.events{where}"
-            " ORDER BY event_time DESC LIMIT %s OFFSET %s",
-            (*params, limit, offset),
-        )
-        rows = await cur.fetchall()
-    return rows, total
-
-
-async def fetch_messages(
-    settings: Settings,
-    *,
-    user: str = "",
-    date_from: str = "",
-    date_to: str = "",
-    limit: int = 50,
-    offset: int = 0,
-) -> tuple[list[dict[str, Any]], int]:
-    """Вернуть дневные счётчики сообщений и общее количество строк."""
-    start, end = _dates(date_from, date_to, settings)
-
-    params: list[Any] = [start, end]
-    where = " WHERE day >= %s AND day <= %s"
-    where += _user_clause(user, params)
-
-    async with get_pool().connection() as conn:
-        cur = await conn.execute(
-            f"""
-            SELECT count(*) AS total
-            FROM (
-                SELECT 1 FROM analytics.messages_daily{where}
-                GROUP BY day, user_id
-            ) AS grouped
-            """,
-            tuple(params),
-        )
-        row = await cur.fetchone()
-        total = int(row["total"]) if row else 0
-
-        cur = await conn.execute(
-            f"""
-            SELECT day, user_id, max(username) AS username, max(chat_id) AS chat_id,
-                   sum(count)::bigint AS count
-            FROM analytics.messages_daily{where}
-            GROUP BY day, user_id
-            ORDER BY day DESC, sum(count) DESC
-            LIMIT %s OFFSET %s
-            """,
-            (*params, limit, offset),
-        )
-        rows = await cur.fetchall()
+    where, params = _where(event_type=event_type, user=user, start=start, end=end)
+    client = await _ensure_client(settings)
+
+    count_result = await client.query(f"SELECT count() FROM analytics.events{where}", parameters=params)
+    total = int(count_result.result_rows[0][0]) if count_result.result_rows else 0
+
+    result = await client.query(
+        f"SELECT {', '.join(EVENT_FIELDS)} FROM analytics.events{where}"
+        " ORDER BY event_time DESC LIMIT {limit:UInt32} OFFSET {offset:UInt32}",
+        parameters={**params, "limit": limit, "offset": offset},
+    )
+    rows = [dict(zip(result.column_names, row, strict=False)) for row in result.result_rows]
     return rows, total

common/config.py

@@ -60,8 +60,13 @@ class Settings(BaseSettings):
     #: Публичный адрес админки (deep-link вход, ссылки из бота).
     admin_public_url: str = "http://localhost:8000"
 
-    #: Аналитика событий в Postgres. ANALYTICS_ENABLED=false — полностью выключить.
-    analytics_enabled: bool = True
+    #: 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
@@ -130,7 +135,7 @@ def get_settings() -> Settings:
 class AdminSettings(Settings):
     """Настройки сервиса админки.
 
-    Наследует общие поля (окружение, Postgres, таймзона), но не требует
+    Наследует общие поля (окружение, Postgres, таймзона, ClickHouse), но не требует
     ``BOT_TOKEN``: на этапе каркаса он не нужен, а появится в проверке подписи
     Telegram-логина (Блок 4.2).
     """

common/repositories/users_admin_repo.py

@@ -16,9 +16,9 @@ _FROM = """
         LEFT JOIN bot.de_users AS du ON du.id_user = tg.id_user
         LEFT JOIN bot.titles AS t ON t.id_title = du.id_title
         LEFT JOIN (
-            SELECT user_id, sum(count) AS cnt, max(day) AS last_dt
-            FROM analytics.messages_daily GROUP BY user_id
-        ) AS msg ON msg.user_id = tg.id_user
+            SELECT id_user, count(*) AS cnt, max(dt) AS last_dt
+            FROM bot.messages GROUP BY id_user
+        ) AS msg ON msg.id_user = tg.id_user
         LEFT JOIN (
             SELECT id_user, sum(amount) AS balance
             FROM bot.money GROUP BY id_user

common/repositories/users_repo.py

@@ -8,7 +8,7 @@ from common.db.pool import get_pool
 
 
 async def add_user(id_user: int, username: str, first_name: str, last_name: str) -> None:
-    """Создать/обновить пользователя (счётчик сообщений ведёт аналитика)."""
+    """Создать/обновить пользователя и добавить запись о сообщении."""
     async with get_pool().connection() as conn, conn.cursor() as cur:
         await cur.execute(
             "CALL bot.add_user(%s, %s, %s, %s)",

common/version.py

@@ -2,9 +2,9 @@
 
 Единый источник правды. Git-тег совпадает с версией с префиксом ``v``
 (например, версия ``0.1`` -> тег ``v0.1``). Docker-образы в Блоке 2 будут
-тегироваться этой версией, а события аналитики получат поле ``bot_version``.
+тегироваться этой версией, а события в ClickHouse получат поле ``bot_version``.
 """
 
 from __future__ import annotations
 
-__version__ = "0.7.1"
+__version__ = "0.6.9"

docker-compose.prod.yml

@@ -46,6 +46,8 @@ services:
     environment:
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
     volumes:
       - media:/app/media
     depends_on:
@@ -53,6 +55,9 @@ services:
         condition: service_healthy
       migrate:
         condition: service_completed_successfully
+      # Без ожидания здоровья: аналитика некритична, бот работает и при недоступном ClickHouse.
+      clickhouse:
+        condition: service_started
     logging:
       driver: json-file
       options:
@@ -69,6 +74,8 @@ services:
     environment:
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
       ADMIN_HOST: 0.0.0.0
       ADMIN_PORT: "8000"
     volumes:
@@ -88,6 +95,37 @@ services:
         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
+      # trace_log не нужен и на тестовом контуре съедал гигабайты в сутки.
+      - ./clickhouse/config.d/disable-trace-log.xml:/etc/clickhouse-server/config.d/disable-trace-log.xml: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:
@@ -105,6 +143,7 @@ services:
 
 volumes:
   pgdata:
+  clickhousedata:
   media:
   # caddy_data:
   # caddy_config:

docker-compose.yml

@@ -51,6 +51,8 @@ services:
       # Внутри сети compose БД доступна по имени сервиса.
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
     volumes:
       - media:/app/media
     depends_on:
@@ -58,6 +60,9 @@ services:
         condition: service_healthy
       migrate:
         condition: service_completed_successfully
+      # Аналитика некритична: бот стартует, даже если ClickHouse ещё не готов.
+      clickhouse:
+        condition: service_started
     logging:
       driver: json-file
       options:
@@ -77,6 +82,8 @@ services:
     environment:
       DB_HOST: postgres
       DB_PORT: "5432"
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_PORT: "8123"
       ADMIN_HOST: 0.0.0.0
       ADMIN_PORT: "8000"
     volumes:
@@ -94,6 +101,34 @@ services:
         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
+      # trace_log не нужен: на тестовом контуре съедал гигабайты в сутки.
+      - ./clickhouse/config.d/disable-trace-log.xml:/etc/clickhouse-server/config.d/disable-trace-log.xml: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:
   media:

docs/events.md

@@ -1,45 +1,48 @@
-# События аналитики (Postgres)
+# События аналитики (ClickHouse)
 
-Бот пишет значимые действия в Postgres, схема `analytics`. Дашборд (Superset) и ETL —
-отдельный финальный этап; здесь описан **контракт** событий: какие бывают `event_type`
-и что лежит в `payload`.
+Бот пишет все значимые действия в ClickHouse — единую широкую таблицу
+`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь
+описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`.
 
 > Источник правды в коде — `common/analytics/events.py` (`EventType`).
 > Расхождение со списком ниже ловит `tests/test_analytics.py`.
 > Добавляешь событие — добавь строку в таблицу.
 
-## Схема в Postgres
+## Схема в ClickHouse
 
-DDL — `sql/V26__analytics_schema.sql`.
+DDL — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера).
 
-- `analytics.events` — события, `PARTITION BY RANGE (event_time)` по дням.
-  Партиции создаёт приложение (бот и админка) на несколько дней вперёд;
-  **старые партиции не удаляются** — история хранится бессрочно.
-- `analytics.messages_daily` — количество сообщений пользователя по дням:
-  `day`, `user_id`, `username`, `chat_id`, `count`; PK `(day, user_id, chat_id)`.
-  На каждое сообщение выполняется `INSERT ... ON CONFLICT DO UPDATE count + 1`,
-  поэтому вместо строки на каждое сообщение растёт только счётчик за день.
+- `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`).
 
-Сообщения **не** пишутся в `analytics.events` отдельным событием `message`.
-Построчная таблица `bot.messages` удалена миграцией `V27`: старые строки один раз
-пересчитаны в `analytics.messages_daily`, новые пишутся только счётчиком.
+Агрегаты остаются на VPS навсегда, поэтому активность, топ команд и денежные потоки
+доступны и после удаления сырых событий. Запросы к агрегатам суммируют `events`/`amount_sum`
+(фон слияний `SummingMergeTree` не гарантирует полную свёртку в любой момент).
 
-## Хранение
+## Хранение и бэкап
 
 | Что | Где | Сколько |
 | --- | --- | --- |
-| События | Postgres, `analytics.events` | бессрочно (партиции по дням) |
-| Счётчики сообщений | Postgres, `analytics.messages_daily` | бессрочно (партиции по дням) |
+| Сырые события | VPS, `analytics.events` | последние `BACKUP_GRACE_DAYS` дней (по умолчанию 7) |
+| Агрегаты | VPS | постоянно |
+| Полная история | домашний сервер, свой ClickHouse | архивы `events_YYYY-MM-DD.parquet.gz` (ежедневно) |
 
-Бэкап — общий бэкап Postgres. Отдельного пайплайна ClickHouse/Parquet больше нет.
-При росте базы нужно следить за диском: автоматической очистки нет.
+Каждую ночь (по Москве) домашний сервер **сам забирает** вчерашнюю партицию с VPS (`pull`
+по ssh), импортирует её в свой ClickHouse и **сверяет количество строк**; только после
+успешной сверки он просит VPS удалить партиции старше грейса. Ошибка бэкапа данные не
+удаляет. Подробности — `DEPLOY.md` и `deploy/ssh-access.md`.
 
 ## Уведомления в чат логов
 
-Помимо записи в Postgres часть событий дублируется сообщением в `ALERT_CHAT_ID`
+Помимо записи в ClickHouse часть событий дублируется сообщением в `ALERT_CHAT_ID`
 (если он задан; `0` — выключено). Отправка не критична: ошибки только логируются.
 
-| Уведомление | Когда | Событие |
+| Уведомление | Когда | Событие в ClickHouse |
 | --- | --- | --- |
 | Вход участника | кто-то вошёл в основной чат | `join` |
 | Выход участника | кто-то вышел из основного чата | `leave` |
@@ -51,19 +54,20 @@ DDL — `sql/V26__analytics_schema.sql`.
 
 | Поле | Тип | Описание |
 | --- | --- | --- |
-| **event_time** | `timestamptz` | Момент события (UTC) |
-| **event_type** | `text` | Тип события (см. каталог ниже) |
-| **user_id** | `bigint` | Telegram-ID пользователя (0 — системное событие) |
-| **username** | `text` | `@username` на момент события (может быть пустым) |
-| **chat_id** | `bigint` | ID чата (обычно основной чат клана) |
-| **payload** | `jsonb` | Дополнительные данные события |
-| **bot_version** | `text` | Версия бота из `common/version.py` |
-| **prod** | `boolean` | `true` — прод-контур, `false` — тестовый |
+| **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` |
@@ -84,44 +88,37 @@ DDL — `sql/V26__analytics_schema.sql`.
 | `bot_alive` | Ежечасный пинг «бот живой» | `app_env`, `uptime_seconds` |
 | `error` | Необработанная ошибка при обработке апдейта | `error_type`, `error`, `update_type` |
 
-## Поведение при недоступности Postgres
+## Поведение при недоступности ClickHouse
 
 Аналитика — **некритичная** подсистема: её сбой не должен влиять на бота.
 
-- события и счётчики копятся в буфере и записываются батчами (по размеру и по таймеру);
-- если Postgres недоступен, буфер не растёт бесконечно: при переполнении самые
-  старые события отбрасываются;
-- если записать батч не удалось, он выводится в stdout (`WARNING`, первые несколько
-  строк + общее количество) и отбрасывается — данные не пропадают молча;
+- события копятся в буфере и отправляются батчами (по размеру и по таймеру);
+- если ClickHouse недоступен, буфер не растёт бесконечно: при переполнении самые
+  старые записи отбрасываются со счётчиком в логе;
+- если записать батч не удалось, события выводятся в stdout (`WARNING`, первые
+  несколько строк + общее количество) и отбрасываются — данные не пропадают молча;
 - при остановке бота выполняется финальный сброс буфера;
-- выключение целиком — `ANALYTICS_ENABLED=false` (тогда `record()`/`record_message()` — no-op).
+- выключение целиком — `CLICKHOUSE_ENABLED=false` (тогда `record()` — no-op).
 
 ## Примеры запросов
 
 ```sql
--- события по дням
-SELECT date_trunc('day', event_time) AS day, count(*)
-FROM analytics.events GROUP BY day ORDER BY day;
+-- активность по дням
+SELECT toDate(event_time) AS day, count() FROM analytics.events GROUP BY day ORDER BY day;
 
 -- топ команд за неделю
-SELECT payload->>'command' AS cmd, count(*) AS uses
+SELECT JSONExtractString(payload, 'command') AS cmd, count() AS uses
 FROM analytics.events
-WHERE event_type = 'command' AND event_time >= now() - interval '7 days'
+WHERE event_type = 'command' AND event_time >= now() - INTERVAL 7 DAY
 GROUP BY cmd ORDER BY uses DESC;
 
 -- покупки за месяц
-SELECT event_type, count(*), sum((payload->>'price')::int)
+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 days'
+WHERE event_type IN ('purchase_title', 'purchase_donat') AND event_time >= now() - INTERVAL 30 DAY
 GROUP BY event_type;
 
--- сообщения по дням
-SELECT day, sum(count) FROM analytics.messages_daily
-WHERE day >= current_date - 30 GROUP BY day ORDER BY day;
-
 -- ошибки
-SELECT event_time, payload->>'error_type', payload->>'error'
-FROM analytics.events WHERE event_type = 'error'
-ORDER BY event_time DESC LIMIT 50;
+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,32 @@
+# Параметры бэкапа событий ClickHouse (pull: домашний сервер забирает данные с VPS).
+# Скопируйте в env/backup.env (env/*.env в git не попадает) и заполните.
+#
+# Запускается НА ДОМАШНЕМ СЕРВЕРЕ по cron:
+#   scripts/clickhouse_backup.sh test|prod
+#
+# SSH-доступ к VPS настраивается отдельно (инструкция — deploy/ssh-access.md).
+
+# --- VPS ---
+# user@host (хост указывайте без скобок; порт задаётся отдельно)
+VPS_SSH=deploy@vps.example
+VPS_PORT=27
+# Каталог с репозиторием на VPS (там запускается docker compose)
+VPS_PROJECT_DIR=/opt/DispelEvilBot
+# Приватный ключ для ssh на VPS (публичная часть добавлена в authorized_keys на VPS)
+VPS_SSH_KEY=/root/.ssh/id_ed25519_vps
+
+# --- Домашний сервер ---
+# Куда складывать архивы events_YYYY-MM-DD.parquet.gz
+BACKUP_DIR=/srv/dispelevilbot/events
+
+# --- Хранение ---
+# Сколько дней сырых событий остаётся на VPS (остальное удаляется после успешной сверки)
+BACKUP_GRACE_DAYS=7
+# Таймзона расчёта «вчерашнего» дня
+BACKUP_TZ=Europe/Moscow
+
+# --- ClickHouse домашнего сервера (сюда импортируются архивы) ---
+CLICKHOUSE_HOST=localhost
+CLICKHOUSE_PORT=9000
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=

env/prod.env.example

@@ -37,8 +37,14 @@ ALERT_CHAT_ID=0
 ADMIN_IDS=
 TIMEZONE=Europe/Moscow
 
-# Аналитика (Postgres, схема analytics; см. docs/events.md).
-ANALYTICS_ENABLED=true
+# ClickHouse (аналитика событий, см. docs/events.md).
+# Наружу не публикуется: только loopback хоста для бэкапа/туннеля;
+# бот ходит внутрь сети compose по имени сервиса clickhouse:8123.
+CLICKHOUSE_DB=analytics
+CLICKHOUSE_USER=de_bot
+CLICKHOUSE_PASSWORD=
+# Порт на хосте (для бэкапа). У test и prod он должен различаться.
+CLICKHOUSE_HTTP_PORT=8123
 
 # Админка (FastAPI). Порт на хосте; у test и prod должен различаться
 # (test 8100, prod 8101 — 8000 занят Portainer).

env/test.env.example

@@ -35,8 +35,14 @@ ALERT_CHAT_ID=0
 ADMIN_IDS=
 TIMEZONE=Europe/Moscow
 
-# Аналитика (Postgres, схема analytics; см. docs/events.md).
-ANALYTICS_ENABLED=true
+# 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
 
 # Админка (FastAPI). Порт на хосте; у test и prod должен различаться
 # (test 8100, prod 8101 — 8000 занят Portainer).

requirements-admin.txt

@@ -17,6 +17,9 @@ tzdata>=2024.1
 # Postgres (async psycopg v3) — понадобится разделам шаблонов и пользователей.
 psycopg[binary,pool]>=3.2,<4.0
 
+# ClickHouse — чтение логов в админке (раздел «Логи»).
+clickhouse-connect[async]>=1.0,<2.0
+
 # Telegram Bot API для модерации (исключение из чата) и уведомлений.
 httpx>=0.27,<1.0
 

requirements.txt

@@ -10,5 +10,7 @@ python-dotenv>=1.0,<2.0
 tzdata>=2024.1
 psycopg[binary,pool]>=3.2,<4.0
 
+# ClickHouse (аналитика событий, Блок 3). Extras [async] — нативный asyncio-клиент.
+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 admin; do
+for svc in postgres clickhouse bot; do
     if echo "$running" | grep -qx "$svc"; then
         echo "OK   ${svc}: running"
     else
@@ -49,11 +49,17 @@ else
     fail=1
 fi
 
-# Схема аналитики в Postgres (создаётся миграцией V26)
-if $COMPOSE exec -T postgres sh -lc 'psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -tAc "SELECT to_regclass('"'"'analytics.events'"'"') IS NOT NULL AND to_regclass('"'"'analytics.messages_daily'"'"') IS NOT NULL"' | grep -qx t; then
-    echo "OK   postgres: схема analytics доступна"
+# 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 postgres: нет таблиц analytics.events/messages_daily (см. sql/V26__analytics_schema.sql)" >&2
+    echo "FAIL clickhouse: не отвечает" >&2
     fail=1
 fi
 

scripts/clickhouse_backup.sh

@@ -0,0 +1,129 @@
+#!/usr/bin/env bash
+# Бэкап сырых событий ClickHouse: домашний сервер забирает данные с VPS (pull).
+#
+# Запускается НА ДОМАШНЕМ СЕРВЕРЕ по cron. Входящий SSH на домашний сервер не нужен:
+# скрипт сам ходит на VPS по ключу и запускает там docker compose.
+#
+# Что делает:
+#   1) по ssh выгружает на VPS вчерашнюю партицию analytics.events в Parquet;
+#   2) складывает events_YYYY-MM-DD.parquet.gz в BACKUP_DIR;
+#   3) импортирует архив в локальный ClickHouse (идемпотентно, scripts/clickhouse_import.sh);
+#   4) сверяет количество строк на VPS и локально;
+#   5) только после успешной сверки просит VPS удалить партиции старше BACKUP_GRACE_DAYS.
+#
+# Ошибка на любом шаге оставляет данные на VPS (партиции не удаляются) и возвращает ненулевой код.
+#
+# Использование:
+#   ./scripts/clickhouse_backup.sh test      # или prod
+#   make backup E=test
+#
+# Настройки — env/backup.env (см. env/backup.env.example).
+
+set -euo pipefail
+
+E="${1:-test}"
+BACKUP_ENV="${BACKUP_ENV:-env/backup.env}"
+
+log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; }
+fail() {
+    echo "[$(date '+%Y-%m-%d %H:%M:%S')] ОШИБКА: $*" >&2
+    exit 1
+}
+
+[ -f "$BACKUP_ENV" ] || fail "нет файла настроек ${BACKUP_ENV} (скопируйте env/backup.env.example)"
+# shellcheck disable=SC1090
+set -a && . "./${BACKUP_ENV}" && set +a
+
+: "${VPS_SSH:?задайте VPS_SSH в env/backup.env}"
+: "${BACKUP_DIR:?задайте BACKUP_DIR в env/backup.env}"
+VPS_PORT="${VPS_PORT:-22}"
+VPS_PROJECT_DIR="${VPS_PROJECT_DIR:?задайте VPS_PROJECT_DIR в env/backup.env}"
+VPS_SSH_KEY="${VPS_SSH_KEY:-$HOME/.ssh/id_ed25519_vps}"
+VPS_COMPOSE_FILE="${VPS_COMPOSE_FILE:-docker-compose.prod.yml}"
+BACKUP_GRACE_DAYS="${BACKUP_GRACE_DAYS:-7}"
+BACKUP_TZ="${BACKUP_TZ:-Europe/Moscow}"
+IMPORT_SCRIPT="${IMPORT_SCRIPT:-$(dirname "$0")/clickhouse_import.sh}"
+SCHEMA_FILE="${SCHEMA_FILE:-$(dirname "$0")/../clickhouse/init/001_events.sql}"
+
+SSH=(ssh -i "$VPS_SSH_KEY" -p "$VPS_PORT" -o BatchMode=yes -o ServerAliveInterval=30 "$VPS_SSH")
+
+# Запустить команду на VPS. Пароль ClickHouse читается из окружения контейнера,
+# поэтому клиент вызывается через `sh -c` уже внутри контейнера.
+# ssh склеивает аргументы в одну строку и отдаёт её удалённому shell, поэтому каждый
+# аргумент экранируем: в SQL есть пробелы, скобки и кавычки.
+vps_ch() {
+    local quoted=() arg
+    for arg in "$@"; do
+        quoted+=("$(printf '%q' "$arg")")
+    done
+    "${SSH[@]}" bash -s "$VPS_PROJECT_DIR" "$E" "$VPS_COMPOSE_FILE" "${quoted[@]}" <<'REMOTE'
+set -euo pipefail
+project_dir="$1"; contour="$2"; compose_file="$3"; shift 3
+cd "$project_dir"
+exec docker compose -p "$contour" --env-file "env/$contour.env" -f "$compose_file" exec -T clickhouse \
+    sh -c 'clickhouse-client --user "$CLICKHOUSE_USER" --password "$CLICKHOUSE_PASSWORD" "$@"' sh "$@"
+REMOTE
+}
+
+# Какие сервисы контура запущены на VPS (пусто, если проекта/контура нет).
+vps_running_services() {
+    "${SSH[@]}" bash -s "$VPS_PROJECT_DIR" "$E" "$VPS_COMPOSE_FILE" <<'REMOTE'
+set -uo pipefail
+if [ ! -d "$1" ]; then
+    echo "каталог проекта $1 на VPS не найден" >&2
+    exit 0
+fi
+cd "$1" || exit 0
+docker compose -p "$2" --env-file "env/$2.env" -f "$3" ps --services --status running 2>/dev/null || true
+REMOTE
+}
+
+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')"
+
+log "Бэкап событий контура ${E} за ${DAY} (грейс ${BACKUP_GRACE_DAYS} дней, порог ${GRACE_CUTOFF})"
+
+# Бот может быть не запущен (или ещё не развёрнут) — это не ошибка, просто пропускаем прогон.
+if ! RUNNING="$(vps_running_services)"; then
+    fail "не удалось получить состояние контура ${E} на VPS (проверьте ssh и VPS_SSH)"
+fi
+if ! grep -qx 'clickhouse' <<<"$RUNNING"; then
+    log "Контур ${E} на VPS не запущен — бэкап пропущен, на VPS ничего не меняем"
+    exit 0
+fi
+
+mkdir -p "$BACKUP_DIR"
+ARCHIVE="${BACKUP_DIR}/events_${DAY}.parquet.gz"
+
+ROWS_VPS="$(vps_ch --param_day="$DAY" --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate({day:String})" | tr -d '[:space:]')"
+log "На VPS строк за ${DAY}: ${ROWS_VPS}"
+if [ "$ROWS_VPS" = "0" ]; then
+    log "ВНИМАНИЕ: за ${DAY} событий нет — партиция пустая, продолжаю (сверка 0 = 0)"
+fi
+
+log "Выгружаю партицию в ${ARCHIVE}"
+vps_ch --param_day="$DAY" --query "SELECT * FROM analytics.events WHERE toDate(event_time) = toDate({day:String}) FORMAT Parquet" |
+    gzip -9 >"${ARCHIVE}.part"
+[ -s "${ARCHIVE}.part" ] || fail "не удалось выгрузить архив ${DAY}"
+mv "${ARCHIVE}.part" "$ARCHIVE"
+log "Размер архива: $(du -h "$ARCHIVE" | cut -f1)"
+
+log "Импортирую архив в ClickHouse домашнего сервера"
+ROWS_HOME="$(BACKUP_DIR="$BACKUP_DIR" SCHEMA_FILE="$SCHEMA_FILE" "$IMPORT_SCRIPT" "$DAY" | tail -n1 | tr -d '[:space:]')"
+log "На домашнем сервере строк за ${DAY}: ${ROWS_HOME}"
+
+[ "$ROWS_VPS" = "$ROWS_HOME" ] || fail "расхождение строк (VPS ${ROWS_VPS}, домашний ${ROWS_HOME}) — партиции на VPS не удаляю"
+
+log "Удаляю на VPS партиции старше ${GRACE_CUTOFF}"
+DROPPED="$(vps_ch --query "SELECT DISTINCT partition FROM system.parts WHERE database='analytics' AND table='events' AND active AND partition < '${GRACE_CUTOFF}'")"
+if [ -z "$DROPPED" ]; then
+    log "Удалять нечего"
+else
+    for partition in $DROPPED; do
+        log "DROP PARTITION ID '${partition}'"
+        vps_ch --query "ALTER TABLE analytics.events DROP PARTITION ID '${partition}'"
+    done
+fi
+
+log "Готово: события за ${DAY} в архиве ${ARCHIVE}, на VPS остались агрегаты и последние ${BACKUP_GRACE_DAYS} дней сырых событий"

scripts/clickhouse_import.sh

@@ -0,0 +1,58 @@
+#!/usr/bin/env bash
+# Импорт дневного архива событий в ClickHouse домашнего сервера.
+#
+# Запускается на домашнем сервере; обычно вызывается из scripts/clickhouse_backup.sh,
+# но можно и вручную — например, чтобы повторно залить архив после сбоя.
+#
+# Использование:
+#   ./scripts/clickhouse_import.sh 2026-01-01
+# В stdout печатается количество строк загруженного дня (для сверки), диагностика — в stderr.
+#
+# Настройки (окружение):
+#   BACKUP_DIR          каталог с архивами events_YYYY-MM-DD.parquet.gz
+#   SCHEMA_FILE         DDL схемы (по умолчанию clickhouse/init/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}/../clickhouse/init/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} (канонический DDL — 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 --param_day="$DAY" --query "SELECT count() FROM analytics.events WHERE toDate(event_time) = toDate({day:String})"

sql/R__functions_stats.sql

@@ -6,10 +6,10 @@ CREATE OR REPLACE FUNCTION bot.get_top_list()
 AS $$
 BEGIN
     RETURN QUERY
-        SELECT tu.username, du.nick, sum(m.count)::bigint AS cnt
-        FROM analytics.messages_daily m
-            INNER JOIN bot.tg_users tu ON m.user_id = tu.id_user
-            INNER JOIN bot.de_users du ON m.user_id = du.id_user
+        SELECT tu.username, du.nick, count(*) AS cnt
+        FROM bot.messages m
+            INNER JOIN bot.tg_users tu ON m.id_user = tu.id_user
+            INNER JOIN bot.de_users du ON m.id_user = du.id_user
         GROUP BY tu.username, du.nick
         ORDER BY 3 DESC
         LIMIT 10;

sql/R__functions_users.sql

@@ -75,15 +75,15 @@ BEGIN
             LEFT JOIN bot.de_users de ON tg.id_user = de.id_user
             LEFT JOIN bot.titles t ON de.id_title = t.id_title
             LEFT JOIN (
-                SELECT user_id, sum(count)::bigint AS msg_count
-                FROM analytics.messages_daily
-                GROUP BY user_id
-            ) m ON tg.id_user = m.user_id
+                SELECT id_user, count(*) AS msg_count
+                FROM bot.messages
+                GROUP BY id_user
+            ) m ON tg.id_user = m.id_user
         WHERE tg.id_user = p_id_user;
 END;
 $$;
 
--- Создать/обновить пользователя.
+-- Создать/обновить пользователя и записать сообщение.
 CREATE OR REPLACE PROCEDURE bot.add_user(
     IN p_id_user bigint,
     IN p_username character varying,
@@ -102,6 +102,8 @@ BEGIN
     INSERT INTO bot.de_users (id_user, nick)
     VALUES (p_id_user, NULL)
     ON CONFLICT (id_user) DO NOTHING;
+
+    INSERT INTO bot.messages (id_user) VALUES (p_id_user);
 END;
 $$;
 
@@ -125,7 +127,7 @@ DECLARE
 BEGIN
     DELETE FROM bot.tg_users WHERE id_user = p_id_user;
     DELETE FROM bot.de_users WHERE id_user = p_id_user;
-    DELETE FROM analytics.messages_daily WHERE user_id = p_id_user;
+    DELETE FROM bot.messages WHERE id_user = p_id_user;
     DELETE FROM bot.purifications WHERE id_user = p_id_user;
 
     v_money := bot.get_money(p_id_user);

sql/V26__analytics_schema.sql

@@ -1,33 +0,0 @@
--- Аналитика в Postgres: события и дневные счётчики сообщений.
--- Партиции по дням создаёт приложение; старые партиции не удаляются.
-CREATE SCHEMA IF NOT EXISTS analytics;
-
-CREATE TABLE IF NOT EXISTS analytics.events (
-    event_time  timestamptz NOT NULL,
-    event_type  text NOT NULL,
-    user_id     bigint NOT NULL DEFAULT 0,
-    username    text NOT NULL DEFAULT '',
-    chat_id     bigint NOT NULL DEFAULT 0,
-    payload     jsonb NOT NULL DEFAULT '{}'::jsonb,
-    bot_version text NOT NULL DEFAULT '',
-    prod        boolean NOT NULL DEFAULT false
-) PARTITION BY RANGE (event_time);
-
-CREATE INDEX IF NOT EXISTS idx_events_time ON analytics.events (event_time);
-CREATE INDEX IF NOT EXISTS idx_events_type_time ON analytics.events (event_type, event_time);
-CREATE INDEX IF NOT EXISTS idx_events_user_time ON analytics.events (user_id, event_time);
-
-COMMENT ON TABLE analytics.events IS 'События бота (партиции по дням, без автоудаления)';
-
-CREATE TABLE IF NOT EXISTS analytics.messages_daily (
-    day      date NOT NULL,
-    user_id  bigint NOT NULL,
-    username text NOT NULL DEFAULT '',
-    chat_id  bigint NOT NULL DEFAULT 0,
-    count    bigint NOT NULL DEFAULT 0,
-    PRIMARY KEY (day, user_id, chat_id)
-) PARTITION BY RANGE (day);
-
-CREATE INDEX IF NOT EXISTS idx_messages_daily_user_day ON analytics.messages_daily (user_id, day);
-
-COMMENT ON TABLE analytics.messages_daily IS 'Количество сообщений пользователя по дням (upsert)';

sql/V27__drop_message_rows.sql

@@ -1,51 +0,0 @@
--- Переход на дневные счётчики: переносим старые построчные сообщения
--- в analytics.messages_daily и удаляем журнал bot.messages.
---
--- Для старых строк chat_id неизвестен (в bot.messages его не было), поэтому
--- пишем 0. Чтение группирует по (day, user_id) и суммирует count, так что
--- старые и новые записи одного дня не дублируются в интерфейсе.
-DO $$
-DECLARE
-    min_day date;
-    max_day date;
-    d date;
-BEGIN
-    SELECT min((dt AT TIME ZONE 'UTC' AT TIME ZONE 'Europe/Moscow')::date),
-           max((dt AT TIME ZONE 'UTC' AT TIME ZONE 'Europe/Moscow')::date)
-      INTO min_day, max_day
-      FROM bot.messages
-     WHERE dt IS NOT NULL;
-
-    IF min_day IS NULL THEN
-        RETURN;
-    END IF;
-
-    -- Партиции на все исторические дни, иначе вставка ниже не найдёт партицию.
-    FOR d IN
-        SELECT min_day + i
-        FROM generate_series(0, max_day - min_day) AS i
-    LOOP
-        EXECUTE format(
-            'CREATE TABLE IF NOT EXISTS analytics.messages_daily_%s '
-            'PARTITION OF analytics.messages_daily '
-            'FOR VALUES FROM (%L) TO (%L)',
-            to_char(d, 'YYYYMMDD'), d, d + 1
-        );
-    END LOOP;
-END $$;
-
-INSERT INTO analytics.messages_daily (day, user_id, username, chat_id, count)
-SELECT (m.dt AT TIME ZONE 'UTC' AT TIME ZONE 'Europe/Moscow')::date AS day,
-       m.id_user,
-       coalesce(tg.username, '') AS username,
-       0 AS chat_id,
-       count(*) AS count
-FROM bot.messages AS m
-    LEFT JOIN bot.tg_users AS tg ON tg.id_user = m.id_user
-WHERE m.dt IS NOT NULL
-GROUP BY 1, 2, 3
-ON CONFLICT (day, user_id, chat_id) DO UPDATE
-SET count = analytics.messages_daily.count + EXCLUDED.count,
-    username = EXCLUDED.username;
-
-DROP TABLE IF EXISTS bot.messages;

tests/test_admin_logs.py

@@ -1,8 +1,8 @@
-"""Тесты раздела «Логи» (Postgres не требуется — чтение подменяется)."""
+"""Тесты раздела «Логи» (без ClickHouse — чтение подменяется)."""
 
 from __future__ import annotations
 
-from datetime import date, datetime
+from datetime import datetime
 from typing import Any
 
 import pytest
@@ -10,39 +10,26 @@ import pytest
 from common.analytics import read as analytics_read
 
 EVENT: dict[str, Any] = {
-    "event_time": datetime(2026, 10, 18, 17, 0, 0),  # UTC без tz — как отдаёт psycopg
+    "event_time": datetime(2026, 10, 18, 17, 0, 0),  # UTC, без tz — как отдаёт clickhouse-connect
     "event_type": "command",
     "user_id": 42,
     "username": "test_user",
     "chat_id": -100123,
-    "payload": {"command": "инфо", "args_count": 0},
+    "payload": '{"command": "инфо", "args_count": 0}',
     "bot_version": "0.4.8",
-    "prod": False,
-}
-
-MESSAGE: dict[str, Any] = {
-    "day": date(2026, 10, 18),
-    "user_id": 42,
-    "username": "test_user",
-    "chat_id": -100123,
-    "count": 17,
+    "prod": 0,
 }
 
 
 @pytest.fixture
 def fake_read(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
-    state: dict[str, Any] = {"total": 1, "messages_total": 1}
+    state: dict[str, Any] = {"total": 1}
 
     async def fetch_events(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]:
         state["kwargs"] = kwargs
         return [EVENT], state["total"]
 
-    async def fetch_messages(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]:
-        state["message_kwargs"] = kwargs
-        return [MESSAGE], state["messages_total"]
-
     monkeypatch.setattr(analytics_read, "fetch_events", fetch_events)
-    monkeypatch.setattr(analytics_read, "fetch_messages", fetch_messages)
     return state
 
 
@@ -56,16 +43,6 @@ def test_logs_page(fake_read: dict[str, Any], client_factory) -> None:
     assert "20:00:00" in response.text  # 17:00 UTC -> 20:00 МСК
 
 
-def test_logs_messages_view(fake_read: dict[str, Any], client_factory) -> None:
-    response = client_factory().get("/logs", params={"view": "messages"})
-
-    assert response.status_code == 200
-    assert "Сообщения по дням" in response.text
-    assert "18.10.2026" in response.text
-    assert "17" in response.text
-    assert fake_read["message_kwargs"]["limit"] == 50
-
-
 def test_logs_filters_passed(fake_read: dict[str, Any], client_factory) -> None:
     client_factory().get(
         "/logs",
@@ -96,16 +73,16 @@ def test_logs_pagination(fake_read: dict[str, Any], client_factory) -> None:
     assert fake_read["kwargs"]["offset"] == 50
 
 
-def test_logs_postgres_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_logs_clickhouse_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
     async def fetch_events(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]:
-        raise RuntimeError("Postgres недоступен")
+        raise RuntimeError("ClickHouse недоступен")
 
     monkeypatch.setattr(analytics_read, "fetch_events", fetch_events)
 
     response = client_factory().get("/logs")
 
     assert response.status_code == 200
-    assert "Postgres недоступен" in response.text
+    assert "ClickHouse недоступен" in response.text
 
 
 def test_logs_bad_date(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:

tests/test_alerts.py

@@ -35,14 +35,17 @@ class FakeBot:
         self.forwarded.append((chat_id, from_chat_id, message_id))
 
 
-class FakeWriter:
-    """Заглушка записи аналитики в Postgres (см. тесты аналитики)."""
+class FakeClickHouse:
+    """Заглушка клиента ClickHouse (см. тесты аналитики)."""
 
     def __init__(self) -> None:
-        self.events: list[list[Any]] = []
+        self.rows: list[list[Any]] = []
 
-    async def __call__(self, events: list[list[Any]], messages: dict[Any, Any]) -> None:
-        self.events.extend(events)
+    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)
@@ -144,8 +147,8 @@ 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 = FakeWriter()
-    await init_analytics(settings, writer=fake)
+    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
@@ -156,15 +159,15 @@ async def test_ping_hourly_sends_once_per_hour(
 
     assert len(bot.messages) == 2
     assert "бот живой" in bot.messages[0][1]
-    assert [row[1] for row in fake.events] == [EventType.BOT_ALIVE, EventType.BOT_ALIVE]
+    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 = FakeWriter()
-    await init_analytics(settings, writer=fake)
+    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
@@ -173,4 +176,4 @@ async def test_ping_hourly_records_event_without_alert_chat(
     await close_analytics()
 
     assert bot.messages == []
-    assert [row[1] for row in fake.events] == [EventType.BOT_ALIVE]
+    assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE]

tests/test_analytics.py

@@ -1,10 +1,13 @@
-"""Тесты аналитики: каталог событий, буферизация и запись в Postgres.
+"""Тесты аналитики: каталог событий, буферизация и деградация без ClickHouse.
 
-Postgres не требуется: writer подменяется фейком.
+Тесты не требуют ни ClickHouse, ни aiogram: клиент подменяется фейком, а
+``clickhouse_connect`` импортируется лениво внутри клиента.
 """
 
 from __future__ import annotations
 
+import json
+import logging
 import re
 from datetime import UTC, datetime
 from pathlib import Path
@@ -13,15 +16,7 @@ from typing import Any
 
 import pytest
 
-from common.analytics import (
-    ALL_EVENT_TYPES,
-    EventType,
-    close_analytics,
-    get_analytics,
-    init_analytics,
-    record,
-    record_message,
-)
+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
@@ -40,24 +35,25 @@ class _FakeEnum:
         self.value = value
 
 
-class FakeWriter:
-    """Заглушка записи в Postgres: складывает батчи в списки."""
+class FakeClickHouse:
+    """Заглушка клиента ClickHouse: пишет в список или падает."""
 
     def __init__(self, *, fail: bool = False) -> None:
-        self.events: list[list[Any]] = []
-        self.messages: dict[tuple[Any, int, int], list[Any]] = {}
+        self.rows: list[list[Any]] = []
+        self.table: str | None = None
+        self.columns: list[str] | None = None
+        self.closed = False
         self._fail = fail
 
-    async def __call__(self, events: list[list[Any]], messages: dict[tuple[Any, int, int], list[Any]]) -> None:
+    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
         if self._fail:
-            raise RuntimeError("Postgres недоступен")
-        self.events.extend(events)
-        for key, value in messages.items():
-            if key in self.messages:
-                self.messages[key][1] += value[1]
-                self.messages[key][0] = value[0]
-            else:
-                self.messages[key] = list(value)
+            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:
@@ -72,9 +68,9 @@ async def _clean_singleton():
     await close_analytics()
 
 
-def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeWriter]:
-    fake = FakeWriter()
-    return AnalyticsClient(_settings(**overrides), writer=fake), fake
+def _client(**overrides: Any) -> tuple[AnalyticsClient, FakeClickHouse]:
+    fake = FakeClickHouse()
+    return AnalyticsClient(_settings(**overrides), client=fake), fake
 
 
 # --- каталог событий ---
@@ -145,46 +141,41 @@ def test_describe_command_returns_none_for_plain_message() -> None:
 
 async def test_record_without_init_is_noop() -> None:
     assert get_analytics() is None
-    record(EventType.COMMAND, user_id=1, chat_id=2, payload={"a": 1})  # не должно бросать
-    record_message(user_id=1, username="vasya", chat_id=2)
+    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.COMMAND, user_id=11, username="vasya", chat_id=100, payload={"command": "инфо"})
-    client.record(EventType.PURIFICATION, user_id=11, username="vasya", chat_id=100, payload={"success": True})
+    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 len(fake.events) == 2
+    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.events[0]
+    row = fake.rows[0]
     assert isinstance(row[0], datetime)
-    assert row[1] == "command"
+    assert row[1] == "message"
     assert row[2] == 11
     assert row[3] == "vasya"
     assert row[4] == 100
-    assert row[5] == {"command": "инфо"}
+    assert '"message_id": 5' in row[5]
     assert row[6] == __version__
-    assert row[7] is False
-
-
-async def test_message_counter_is_aggregated_per_day() -> None:
-    client, fake = _client()
-    client.record_message(user_id=11, username="vasya", chat_id=100)
-    client.record_message(user_id=11, username="vasya", chat_id=100)
-    client.record_message(user_id=12, username=None, chat_id=100)
-
-    assert client.pending == 3
-    await client.flush()
-
-    assert client.pending == 0
-    assert len(fake.messages) == 2
-    vasya = next(value for key, value in fake.messages.items() if key[1] == 11)
-    assert vasya == ["vasya", 2]
+    assert row[7] == 0
 
 
 async def test_remember_user_enriches_username() -> None:
@@ -194,17 +185,17 @@ async def test_remember_user_enriches_username() -> None:
 
     await client.flush()
 
-    assert fake.events[0][3] == "nickname"
+    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.COMMAND, user_id=1)
-    client.record(EventType.COMMAND, user_id=1)
+    client.record(EventType.MESSAGE, user_id=1)
+    client.record(EventType.MESSAGE, user_id=1)
     assert not client._flush_now.is_set()
 
-    client.record(EventType.COMMAND, user_id=1)
+    client.record(EventType.MESSAGE, user_id=1)
     assert client._flush_now.is_set()
 
 
@@ -212,56 +203,64 @@ 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.COMMAND, user_id=user_id)
+        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(analytics_enabled=False)
+    client, fake = _client(clickhouse_enabled=False)
 
-    client.record(EventType.COMMAND, user_id=1)
-    client.record_message(user_id=1, username="vasya", chat_id=1)
+    client.record(EventType.MESSAGE, user_id=1)
     await client.flush()
 
     assert client.pending == 0
-    assert fake.events == []
-    assert fake.messages == {}
+    assert fake.rows == []
 
 
-# --- деградация без Postgres ---
+# --- деградация без ClickHouse ---
 
 
 async def test_flush_failure_logs_fallback_and_drops(caplog: pytest.LogCaptureFixture) -> None:
-    fake = FakeWriter(fail=True)
-    client = AnalyticsClient(_settings(), writer=fake)
-    client.record(EventType.COMMAND, user_id=1, username="vasya", chat_id=100)
+    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("WARNING"):
+    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.COMMAND, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)})
+    client.record(EventType.MESSAGE, user_id=1, payload={"text": "x" * (PAYLOAD_MAX_LENGTH * 2)})
 
     await client.flush()
 
-    payload = fake.events[0][5]
-    assert "_truncated" in payload
-    assert len(payload["_truncated"]) == PAYLOAD_MAX_LENGTH
+    assert len(fake.rows[0][5]) == PAYLOAD_MAX_LENGTH
 
 
 # --- жизненный цикл ---
 
 
 async def test_init_and_close_flush_pending_events() -> None:
-    fake = FakeWriter()
-    client = await init_analytics(_settings(), writer=fake)
+    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"})
@@ -272,7 +271,8 @@ async def test_init_and_close_flush_pending_events() -> None:
     await close_analytics()
 
     assert get_analytics() is None
-    assert len(fake.events) == 2
+    assert len(fake.rows) == 2
+    assert fake.closed is True
 
 
 # --- middleware (транспортные события) ---
@@ -297,23 +297,22 @@ async def _handler(event: Any, data: dict[str, Any]) -> str:
     return "handled"
 
 
-async def test_middleware_counts_message_and_records_command(monkeypatch: pytest.MonkeyPatch) -> None:
+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 = FakeWriter()
-    client = await init_analytics(settings, writer=fake)
+    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.events] == ["command"]
-    assert fake.events[0][5] == {"command": "инфо", "args_count": 0}
-    assert fake.events[0][3] == "vasya"
-    assert len(fake.messages) == 1
-    assert next(iter(fake.messages.values())) == ["vasya", 1]
+    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:
@@ -322,8 +321,8 @@ async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyP
     settings = _settings(chat_id=-100)
     monkeypatch.setattr("bot.middlewares.analytics.get_settings", lambda: settings)
 
-    fake = FakeWriter()
-    client = await init_analytics(settings, writer=fake)
+    fake = FakeClickHouse()
+    client = await init_analytics(settings, client=fake)
 
     from aiogram.types import Chat, User
 
@@ -335,8 +334,7 @@ async def test_middleware_skips_other_chats_and_bots(monkeypatch: pytest.MonkeyP
     )
     await client.flush()
 
-    assert fake.events == []
-    assert fake.messages == {}
+    assert fake.rows == []
 
 
 async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -348,6 +346,6 @@ async def test_middleware_never_breaks_handling(monkeypatch: pytest.MonkeyPatch)
     def boom(*args: Any, **kwargs: Any) -> None:
         raise RuntimeError("аналитика сломалась")
 
-    monkeypatch.setattr("bot.middlewares.analytics.record_message", boom)
+    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.7.1"
+    assert __version__ == "0.6.9"