Построчный журнал сообщений больше не нужен

Счётчики сообщений теперь живут только в analytics.messages_daily: /инфо, /рейтинг и админка суммируют дневные значения, а bot.add_user больше не пишет строку на каждое сообщение. Старые строки bot.messages один раз перенесены в дневную таблицу и сама таблица удалена; в карточке пользователя «последнее сообщение» стало датой.

.env.example

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

DEPLOY.md

@@ -16,16 +16,14 @@ cd DispelEvilBot
 ```bash
 cp env/test.env.example env/test.env
 # заполнить: BOT_TOKEN, DB_PASSWORD, CHAT_ID (основной чат), ALERT_CHAT_ID (опц.),
-# ADMIN_IDS (опц.), CLICKHOUSE_PASSWORD (аналитика событий)
+# ADMIN_IDS (опц.)
 # плюс для админки: BOT_USERNAME, ADMIN_PUBLIC_URL (HTTPS-домен),
 # ADMIN_SESSION_SECRET (openssl rand -hex 32)
 ```
 
-Порт 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`.
+Порт админки (`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 (если пакет приватный)
 
@@ -55,35 +53,36 @@ docker compose -p test --env-file env/test.env -f docker-compose.prod.yml logs -
 
 1. `postgres` поднимается и проходит healthcheck;
 2. `migrate` (**Flyway**) накатывает миграции из `sql/` — для пустой БД с нуля;
-3. `clickhouse` поднимается и применяет `clickhouse/init/001_events.sql` —
-   **только при первой инициализации** пустого data-каталога;
-4. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же;
-5. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста).
+3. `bot` стартует (`Запуск DispelEvilBot v<версия> (APP_ENV=...)`), аналитика и рассылки — в нём же;
+4. `admin` стартует и отвечает на `GET /healthz` (порт на loopback хоста).
 
-Проверка контура (сервисы, миграции, Postgres, ClickHouse):
+Проверка контура (сервисы, миграции, Postgres, схема аналитики):
 
 ```bash
 make check E=test
 ```
 
-## ClickHouse (аналитика событий)
+## Аналитика (Postgres)
 
-События всех действий бота пишутся в `analytics.events` (контракт — [`docs/events.md`](docs/events.md)).
-ClickHouse наружу **не публикуется**: только loopback хоста (`127.0.0.1:${CLICKHOUSE_HTTP_PORT}`).
+События и дневные счётчики сообщений пишутся в схему `analytics` той же БД Postgres
+(контракт — [`docs/events.md`](docs/events.md)). Отдельного сервиса нет.
 
 ```bash
-make ch E=test      # clickhouse-client контура
+make psql E=test
 # пример: события за сегодня по типам
-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"'
+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;
 ```
 
-Если ClickHouse недоступен, бот продолжает работать: события уходят в stdout (WARNING),
-полностью выключить можно флагом `CLICKHOUSE_ENABLED=false`.
+Если Postgres недоступен, бот продолжает работать: батч аналитики уходит в stdout
+(WARNING) и отбрасывается. Полностью выключить можно флагом `ANALYTICS_ENABLED=false`.
 
-Изменения схемы: `clickhouse/init/*.sql` применяются лишь на пустом data-каталоге,
-поэтому на существующем контуре DDL нужно применить вручную (`make ch E=...`).
+Партиции создаются приложением при старте и раз в 6 часов; старые **не удаляются**.
+Схема создаётся миграцией `V26__analytics_schema.sql`.
 
 ## Админка (FastAPI)
 
@@ -119,51 +118,11 @@ reverse proxy. Сессия живёт 12 часов (`ADMIN_SESSION_MAX_AGE`),
 Роль администратора перечитывается из БД на каждом запросе — отключение пользователя
 (`enabled = false`) или удаление записи сразу закрывает доступ.
 
-## Бэкап событий (забирает домашний сервер)
+## Бэкап
 
-Сырые события живут на 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 это не ошибка.
+Аналитика теперь лежит в Postgres, поэтому отдельного бэкапа событий больше нет.
+Бэкап БД — стандартными средствами Postgres (`pg_dump`/`pg_basebackup`); при росте
+истории следите за размером `analytics` и диском.
 
 ## Обновление и откат
 
@@ -184,13 +143,13 @@ docker compose -p <контур> --env-file env/<контур>.env -f docker-com
 
 ## Переменные окружения
 
-См. `env/prod.env.example` / `env/test.env.example` (контуры) и `env/backup.env.example` (бэкап событий).
+См. `env/prod.env.example` / `env/test.env.example` (контуры).
 
 ## Дашборд (позже)
 
-Superset поднимается на домашнем сервере и читает **локальный** ClickHouse с полной
-историей событий (сюда ежедневно приезжают архивы). Отдельный read-only пользователь
-ClickHouse и сам дашборд — финальный этап, в этот контур не входят.
+Superset поднимается отдельно и читает Postgres (схема `analytics`) с полной историей
+событий. Отдельный read-only пользователь Postgres и сам дашборд — финальный этап,
+в этот контур не входят.
 
 ## Перенос данных на прод (позже, короткое окно простоя)
 

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 ch backup
+.PHONY: help up pull down restart logs ps images check psql
 
 help:
 > @echo "Контур задаётся через E=test|prod (по умолчанию test):"
@@ -22,8 +22,6 @@ 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
@@ -51,9 +49,3 @@ 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,10 +9,9 @@ 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)
-clickhouse/ # init-скрипты ClickHouse (схема аналитики)
+sql/        # миграции Flyway (V__*.sql, R__*.sql), включая схему analytics
 docs/       # документация (docs/events.md — события аналитики)
-scripts/    # эксплуатационные скрипты (проверка контура, бэкап событий)
+scripts/    # эксплуатационные скрипты (проверка контура)
 images/     # статичные картинки (приветствие)
 deploy/     # материалы деплоя (Caddyfile, инструкции)
 tests/      # тесты
@@ -24,8 +23,7 @@ docker-compose.yml / docker-compose.prod.yml
 ## Требования
 
 - Python 3.12
-- PostgreSQL 14+
-- ClickHouse 25.3 (аналитика событий; можно выключить — `CLICKHOUSE_ENABLED=false`)
+- PostgreSQL 14+ (аналитика — в схеме `analytics`)
 
 ## Настройка
 
@@ -88,10 +86,10 @@ HttpOnly-cookie (`ADMIN_SESSION_SECRET`), все формы защищены CSR
 видны боту (он читает `bot.titles`). Удалить можно только звание, которое ни у кого
 не стоит; используемое — скрывается (`visible = false`).
 
-Раздел «Логи» — события из ClickHouse: фильтры по типу, пользователю (ID или username)
-и датам, пагинация по 50. Время показывается в `TIMEZONE`, payload раскрывается как JSON,
-ID пользователя ведёт в его карточку. ClickHouse читается отдельным ленивым клиентом;
-если он недоступен, страница показывает ошибку, а не падает.
+Раздел «Логи» — события из Postgres и отдельная вкладка «Сообщения по дням»: фильтры
+по типу (для событий), пользователю (ID или username) и датам, пагинация по 50.
+Время показывается в `TIMEZONE`, payload раскрывается как JSON, ID пользователя ведёт
+в его карточку. Если Postgres недоступен, страница показывает ошибку, а не падает.
 
 Раздел «Админы» — выдача и отзыв доступа к панели по Telegram-ID, роли и вкл/выкл.
 В карточке пользователя есть исключение из чата (бан + разбан) с удалением данных и
@@ -113,16 +111,19 @@ ID пользователя ведёт в его карточку. ClickHouse ч
 редактором, картинка и URL-кнопки, предпросмотр и подтверждение. Каждая отправка
 (успешная или нет) пишется в `bot.admin_messages`.
 
-## Аналитика (ClickHouse)
+## Аналитика (Postgres)
 
-Все значимые действия бота пишутся в ClickHouse в единую таблицу `analytics.events`.
-Контракт событий — [`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
+Все значимые действия бота пишутся в Postgres, схема `analytics`:
+`analytics.events` (события, партиции по дням) и `analytics.messages_daily`
+(количество сообщений пользователя по дням). Контракт событий —
+[`docs/events.md`](docs/events.md) (типы, payload, примеры запросов).
 
-- схема — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера);
-- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой ClickHouse не влияет на бота);
-- сырые события живут на VPS 7 дней и ежедневно уезжают архивом на домашний сервер,
-  на VPS остаются агрегаты (`events_daily`, `money_daily`) — см. [DEPLOY.md](DEPLOY.md);
-- выключается флагом `CLICKHOUSE_ENABLED=false`.
+- схема — `sql/V26__analytics_schema.sql`;
+- клиент и батчинг — `common/analytics/` (запись неблокирующая, сбой Postgres не влияет на бота);
+- сообщения не пишутся поштучно: на каждое сообщение инкрементится дневной счётчик;
+- построчная `bot.messages` удалена (V27), старые данные перенесены в `analytics.messages_daily`;
+- партиции создаются приложением и **не удаляются** — история хранится бессрочно;
+- выключается флагом `ANALYTICS_ENABLED=false`.
 
 ## Уведомления в чат логов
 
@@ -131,7 +132,7 @@ ID пользователя ведёт в его карточку. ClickHouse ч
 
 ## Docker
 
-Локальный запуск (postgres + migrate + clickhouse + bot + admin):
+Локальный запуск (postgres + migrate + bot + admin):
 
 ```bash
 cp .env.example .env      # заполнить
@@ -162,22 +163,17 @@ black --check .
 pytest
 ```
 
-## Бэкап событий
+## Бэкап
 
-Сырые события ClickHouse ежедневно забираются с VPS домашним сервером (`pull`), на VPS
-остаются только агрегаты и последние дни сырых данных:
-
-```bash
-make backup E=test     # на домашнем сервере: ./scripts/clickhouse_backup.sh test
-```
-
-Настройка, cron и SSH-доступ — см. [`DEPLOY.md`](DEPLOY.md).
+Аналитика теперь в Postgres, отдельного бэкапа событий нет. Бэкап БД — стандартными
+средствами Postgres; история `analytics` не чистится, поэтому следите за размером БД
+и диском. Подробности — [`DEPLOY.md`](DEPLOY.md).
 
 ## Версионирование
 
-- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.6.9`.
+- Единый источник версии — `common/version.py` (`__version__`), текущая — `0.7.1`.
 - Схема — SemVer-подобная (`MAJOR.MINOR.PATCH`); git-тег = `v` + версия (`v0.1`).
-- Docker-образы тегируются версией (Блок 2), события ClickHouse получают `bot_version` (Блок 3).
+- Docker-образы тегируются версией (Блок 2), события аналитики получают `bot_version`.
 - Выпуск: изменить `common/version.py` → коммит → аннотированный тег `vX.Y` → `git push --tags`.
 
 ## Процесс работы с git
@@ -231,6 +227,5 @@ Telegram не принимает кириллицу в именах команд
 
 ## Переменные окружения
 
-- общие — `.env.example` (включая `CLICKHOUSE_*` и параметры батчинга `ANALYTICS_*`);
-- контуры — `env/prod.env.example`, `env/test.env.example`;
-- бэкап событий — `env/backup.env.example`.
+- общие — `.env.example` (включая `ANALYTICS_*`);
+- контуры — `env/prod.env.example`, `env/test.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 import read as analytics_read
+from common.analytics.partitions import ensure_partitions
 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,6 +42,10 @@ 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:
@@ -56,7 +60,6 @@ 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 @@
-"""Раздел «Логи»: события из ClickHouse с фильтрами и пагинацией."""
+"""Раздел «Логи»: события и дневные счётчики сообщений из Postgres."""
 
 from __future__ import annotations
 
@@ -19,16 +19,18 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/logs", tags=["logs"])
 
-#: Сколько событий показывать на странице.
+#: Сколько строк показывать на странице.
 PAGE_SIZE = 50
 
 
-def _pretty_payload(payload: str) -> str:
+def _pretty_payload(payload: Any) -> str:
     """Отформатировать JSON-payload для просмотра; битый JSON — как есть."""
+    if isinstance(payload, dict):
+        return json.dumps(payload, ensure_ascii=False, indent=2)
     try:
-        return json.dumps(json.loads(payload), ensure_ascii=False, indent=2)
+        return json.dumps(json.loads(str(payload or "")), ensure_ascii=False, indent=2)
     except (TypeError, ValueError):
-        return payload or ""
+        return str(payload or "")
 
 
 def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]:
@@ -44,31 +46,52 @@ def _decorate(row: dict[str, Any], settings: Any) -> dict[str, Any]:
     return {
         **row,
         "time_local": time_local,
-        "payload_pretty": _pretty_payload(str(row.get("payload") or "")),
+        "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 "—"),
     }
 
 
 @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,
@@ -81,29 +104,27 @@ async def logs_page(
     except ValueError:
         error = "Даты должны быть в формате ГГГГ-ММ-ДД."
     except Exception:
-        logger.exception("Не удалось прочитать события из ClickHouse")
-        error = "ClickHouse недоступен — логи временно не читаются."
+        logger.exception("Не удалось прочитать аналитику из Postgres")
+        error = "Postgres недоступен — логи временно не читаются."
 
     pages = max((total + PAGE_SIZE - 1) // PAGE_SIZE, 1)
-    query_base = urlencode(
-        {
-            key: value
-            for key, value in {
-                "event_type": event_type,
+    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,
-            }.items()
-            if value
     }
-    )
+    query_base = urlencode({key: value for key, value in query_params.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": "События из ClickHouse с фильтрами",
+        "description": "События и сообщения из Postgres с фильтрами",
         "url": "/logs",
     },
     {

admin/templates/logs/list.html

@@ -6,11 +6,33 @@
 <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">События из ClickHouse · найдено: {{ total }}</p>
+    <p class="mt-1 text-sm text-slate-600">
+      {% if view == "messages" %}Сообщения по дням{% else %}События из Postgres{% endif %}
+      · найдено: {{ total }}
+    </p>
   </div>
 </div>
 
-<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 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" %}
   <div>
     <label for="event_type" class="text-xs text-slate-500">Тип</label>
     <select
@@ -24,6 +46,7 @@
       {% endfor %}
     </select>
   </div>
+  {% endif %}
   <div>
     <label for="user" class="text-xs text-slate-500">Пользователь</label>
     <input
@@ -31,7 +54,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>
@@ -58,13 +81,46 @@
   <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" class="text-sm text-slate-600 hover:underline">Сбросить</a>
+  <a href="/logs{% if view == 'messages' %}?view=messages{% endif %}" 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">
@@ -116,6 +172,7 @@
     </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 %H:%M") if user.last_message_at else "—" }}
+      последняя активность: {{ user.last_message_at.strftime("%d.%m.%Y") if user.last_message_at else "—" }}
     </div>
   </div>
   <div class="rounded-lg border border-slate-200 bg-white p-4">

bot/__main__.py

@@ -12,6 +12,7 @@ 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
@@ -26,6 +27,10 @@ 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()
@@ -36,13 +41,20 @@ 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()
-        await asyncio.gather(notifications_task, heartbeat_task, return_exceptions=True)
+        partitions_task.cancel()
+        await asyncio.gather(
+            notifications_task,
+            heartbeat_task,
+            partitions_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, remember_user
-from common.analytics.messages import describe_command, describe_message
+from common.analytics import EventType, record, record_message, remember_user
+from common.analytics.messages import describe_command
 from common.config import get_settings
 
 logger = logging.getLogger(__name__)
@@ -43,13 +43,7 @@ class AnalyticsMiddleware(BaseMiddleware):
 
         try:
             remember_user(user.id, user.username)
-            record(
-                EventType.MESSAGE,
-                user_id=user.id,
-                username=user.username,
-                chat_id=message.chat.id,
-                payload=describe_message(message),
-            )
+            record_message(user_id=user.id, username=user.username, chat_id=message.chat.id)
             command = describe_command(message)
             if command is not None:
                 record(

bot/services/heartbeat.py

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

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

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

clickhouse/init/001_events.sql

@@ -1,72 +0,0 @@
--- Схема аналитики 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 @@
-"""Аналитика: клиент ClickHouse и каталог событий.
+"""Аналитика: клиент Postgres и каталог событий.
 
 Публичный интерфейс:
 
@@ -16,6 +16,7 @@ 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
@@ -29,5 +30,6 @@ __all__ = [
     "get_analytics",
     "init_analytics",
     "record",
+    "record_message",
     "remember_user",
 ]

common/analytics/client.py

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

common/analytics/partitions.py

@@ -0,0 +1,64 @@
+"""Партиции аналитики по дням.
+
+Создаются заранее (вчера/сегодня/два дня вперёд), старые **не удаляются**:
+история событий и дневных счётчиков хранится бессрочно.
+"""
+
+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,17 +1,14 @@
-"""Чтение событий ClickHouse для админки (Блок 4.3).
-
-Клиент создаётся лениво и только на чтение; недоступность ClickHouse не роняет
-админку — страница покажет ошибку. Запись событий живёт в ``client.py``.
-"""
+"""Чтение аналитики из Postgres для админки."""
 
 from __future__ import annotations
 
 import logging
-from datetime import UTC, datetime, time, timedelta
+from datetime import UTC, date, 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__)
 
@@ -27,62 +24,29 @@ EVENT_FIELDS = (
     "prod",
 )
 
-_client: Any | None = None
+
+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
 
 
-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 _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 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
-
+def _user_clause(user: str, params: list[Any], column: str = "user_id") -> str:
     user = user.strip()
-    if user:
+    if not user:
+        return ""
     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
+        params.append(int(user))
+        return f" AND {column} = %s"
+    params.append(f"%{user}%")
+    return " AND username ILIKE %s"
 
 
 async def fetch_events(
@@ -95,31 +59,71 @@ 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)
 
-    Даты — ``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
+    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)
 
-    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)
-    )
+    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
 
-    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(
+        cur = await conn.execute(
             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},
+            " ORDER BY event_time DESC LIMIT %s OFFSET %s",
+            (*params, limit, offset),
         )
-    rows = [dict(zip(result.column_names, row, strict=False)) for row in result.result_rows]
+        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()
     return rows, total

common/config.py

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

docker-compose.prod.yml

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

docker-compose.yml

@@ -51,8 +51,6 @@ services:
       # Внутри сети compose БД доступна по имени сервиса.
       DB_HOST: postgres
       DB_PORT: "5432"
-      CLICKHOUSE_HOST: clickhouse
-      CLICKHOUSE_PORT: "8123"
     volumes:
       - media:/app/media
     depends_on:
@@ -60,9 +58,6 @@ services:
         condition: service_healthy
       migrate:
         condition: service_completed_successfully
-      # Аналитика некритична: бот стартует, даже если ClickHouse ещё не готов.
-      clickhouse:
-        condition: service_started
     logging:
       driver: json-file
       options:
@@ -82,8 +77,6 @@ services:
     environment:
       DB_HOST: postgres
       DB_PORT: "5432"
-      CLICKHOUSE_HOST: clickhouse
-      CLICKHOUSE_PORT: "8123"
       ADMIN_HOST: 0.0.0.0
       ADMIN_PORT: "8000"
     volumes:
@@ -101,34 +94,6 @@ 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,48 +1,45 @@
-# События аналитики (ClickHouse)
+# События аналитики (Postgres)
 
-Бот пишет все значимые действия в ClickHouse — единую широкую таблицу
-`analytics.events`. Дашборд (Superset) и ETL — отдельный финальный этап; здесь
-описан **контракт** событий: какие бывают `event_type` и что лежит в `payload`.
+Бот пишет значимые действия в Postgres, схема `analytics`. Дашборд (Superset) и ETL —
+отдельный финальный этап; здесь описан **контракт** событий: какие бывают `event_type`
+и что лежит в `payload`.
 
 > Источник правды в коде — `common/analytics/events.py` (`EventType`).
 > Расхождение со списком ниже ловит `tests/test_analytics.py`.
 > Добавляешь событие — добавь строку в таблицу.
 
-## Схема в ClickHouse
+## Схема в Postgres
 
-DDL — `clickhouse/init/001_events.sql` (применяется при первой инициализации контейнера).
+DDL — `sql/V26__analytics_schema.sql`.
 
-- `analytics.events` — сырые события, `MergeTree`, `PARTITION BY toYYYYMMDD(event_time)`,
-  `ORDER BY (event_time, event_type, user_id)`. Партиционирование по дню выбрано специально:
-  сырые партиции ежедневно удаляются после бэкапа (дешёвый `DROP PARTITION`, без mutation).
-- `analytics.events_daily` — агрегат (`SummingMergeTree`): `day`, `event_type`, `user_id`,
-  `chat_id`, `events`. Строится материализованным представлением `events_daily_mv`.
-- `analytics.money_daily` — агрегат по деньгам: `day`, `event_type`, `user_id`,
-  `amount_sum`, `events` (из `payload.amount` событий `money_accrual`/`money_spend`).
+- `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`,
+  поэтому вместо строки на каждое сообщение растёт только счётчик за день.
 
-Агрегаты остаются на VPS навсегда, поэтому активность, топ команд и денежные потоки
-доступны и после удаления сырых событий. Запросы к агрегатам суммируют `events`/`amount_sum`
-(фон слияний `SummingMergeTree` не гарантирует полную свёртку в любой момент).
+Сообщения **не** пишутся в `analytics.events` отдельным событием `message`.
+Построчная таблица `bot.messages` удалена миграцией `V27`: старые строки один раз
+пересчитаны в `analytics.messages_daily`, новые пишутся только счётчиком.
 
-## Хранение и бэкап
+## Хранение
 
 | Что | Где | Сколько |
 | --- | --- | --- |
-| Сырые события | VPS, `analytics.events` | последние `BACKUP_GRACE_DAYS` дней (по умолчанию 7) |
-| Агрегаты | VPS | постоянно |
-| Полная история | домашний сервер, свой ClickHouse | архивы `events_YYYY-MM-DD.parquet.gz` (ежедневно) |
+| События | Postgres, `analytics.events` | бессрочно (партиции по дням) |
+| Счётчики сообщений | Postgres, `analytics.messages_daily` | бессрочно (партиции по дням) |
 
-Каждую ночь (по Москве) домашний сервер **сам забирает** вчерашнюю партицию с VPS (`pull`
-по ssh), импортирует её в свой ClickHouse и **сверяет количество строк**; только после
-успешной сверки он просит VPS удалить партиции старше грейса. Ошибка бэкапа данные не
-удаляет. Подробности — `DEPLOY.md` и `deploy/ssh-access.md`.
+Бэкап — общий бэкап Postgres. Отдельного пайплайна ClickHouse/Parquet больше нет.
+При росте базы нужно следить за диском: автоматической очистки нет.
 
 ## Уведомления в чат логов
 
-Помимо записи в ClickHouse часть событий дублируется сообщением в `ALERT_CHAT_ID`
+Помимо записи в Postgres часть событий дублируется сообщением в `ALERT_CHAT_ID`
 (если он задан; `0` — выключено). Отправка не критична: ошибки только логируются.
 
-| Уведомление | Когда | Событие в ClickHouse |
+| Уведомление | Когда | Событие |
 | --- | --- | --- |
 | Вход участника | кто-то вошёл в основной чат | `join` |
 | Выход участника | кто-то вышел из основного чата | `leave` |
@@ -54,20 +51,19 @@ DDL — `clickhouse/init/001_events.sql` (применяется при перв
 
 | Поле | Тип | Описание |
 | --- | --- | --- |
-| **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 — тестовый |
+| **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` — тестовый |
 
 ## Каталог событий
 
 | Тип события | Когда пишется | `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` |
@@ -88,37 +84,44 @@ DDL — `clickhouse/init/001_events.sql` (применяется при перв
 | `bot_alive` | Ежечасный пинг «бот живой» | `app_env`, `uptime_seconds` |
 | `error` | Необработанная ошибка при обработке апдейта | `error_type`, `error`, `update_type` |
 
-## Поведение при недоступности ClickHouse
+## Поведение при недоступности Postgres
 
 Аналитика — **некритичная** подсистема: её сбой не должен влиять на бота.
 
-- события копятся в буфере и отправляются батчами (по размеру и по таймеру);
-- если ClickHouse недоступен, буфер не растёт бесконечно: при переполнении самые
-  старые записи отбрасываются со счётчиком в логе;
-- если записать батч не удалось, события выводятся в stdout (`WARNING`, первые
-  несколько строк + общее количество) и отбрасываются — данные не пропадают молча;
+- события и счётчики копятся в буфере и записываются батчами (по размеру и по таймеру);
+- если Postgres недоступен, буфер не растёт бесконечно: при переполнении самые
+  старые события отбрасываются;
+- если записать батч не удалось, он выводится в stdout (`WARNING`, первые несколько
+  строк + общее количество) и отбрасывается — данные не пропадают молча;
 - при остановке бота выполняется финальный сброс буфера;
-- выключение целиком — `CLICKHOUSE_ENABLED=false` (тогда `record()` — no-op).
+- выключение целиком — `ANALYTICS_ENABLED=false` (тогда `record()`/`record_message()` — no-op).
 
 ## Примеры запросов
 
 ```sql
--- активность по дням
-SELECT toDate(event_time) AS day, count() FROM analytics.events GROUP BY day ORDER BY day;
+-- события по дням
+SELECT date_trunc('day', event_time) AS day, count(*)
+FROM analytics.events GROUP BY day ORDER BY day;
 
 -- топ команд за неделю
-SELECT JSONExtractString(payload, 'command') AS cmd, count() AS uses
+SELECT payload->>'command' AS cmd, count(*) AS uses
 FROM analytics.events
-WHERE event_type = 'command' AND event_time >= now() - INTERVAL 7 DAY
+WHERE event_type = 'command' AND event_time >= now() - interval '7 days'
 GROUP BY cmd ORDER BY uses DESC;
 
 -- покупки за месяц
-SELECT event_type, count(), sum(JSONExtractInt(payload, 'price'))
+SELECT event_type, count(*), sum((payload->>'price')::int)
 FROM analytics.events
-WHERE event_type IN ('purchase_title', 'purchase_donat') AND event_time >= now() - INTERVAL 30 DAY
+WHERE event_type IN ('purchase_title', 'purchase_donat')
+  AND event_time >= now() - interval '30 days'
 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, JSONExtractString(payload, 'error_type'), JSONExtractString(payload, 'error')
-FROM analytics.events WHERE event_type = 'error' ORDER BY event_time DESC LIMIT 50;
+SELECT event_time, payload->>'error_type', payload->>'error'
+FROM analytics.events WHERE event_type = 'error'
+ORDER BY event_time DESC LIMIT 50;
 ```

env/backup.env.example

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

env/test.env.example

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

requirements-admin.txt

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

scripts/clickhouse_backup.sh

@@ -1,129 +0,0 @@
-#!/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

@@ -1,58 +0,0 @@
-#!/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, 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
+        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
         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 id_user, count(*) AS msg_count
-                FROM bot.messages
-                GROUP BY id_user
-            ) m ON tg.id_user = m.id_user
+                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
         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,8 +102,6 @@ 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;
 $$;
 
@@ -127,7 +125,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 bot.messages WHERE id_user = p_id_user;
+    DELETE FROM analytics.messages_daily WHERE user_id = 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

@@ -0,0 +1,33 @@
+-- Аналитика в 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

@@ -0,0 +1,51 @@
+-- Переход на дневные счётчики: переносим старые построчные сообщения
+-- в 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 @@
-"""Тесты раздела «Логи» (без ClickHouse — чтение подменяется)."""
+"""Тесты раздела «Логи» (Postgres не требуется — чтение подменяется)."""
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import date, datetime
 from typing import Any
 
 import pytest
@@ -10,26 +10,39 @@ 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 — как отдаёт clickhouse-connect
+    "event_time": datetime(2026, 10, 18, 17, 0, 0),  # UTC без tz — как отдаёт psycopg
     "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": 0,
+    "prod": False,
+}
+
+MESSAGE: dict[str, Any] = {
+    "day": date(2026, 10, 18),
+    "user_id": 42,
+    "username": "test_user",
+    "chat_id": -100123,
+    "count": 17,
 }
 
 
 @pytest.fixture
 def fake_read(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
-    state: dict[str, Any] = {"total": 1}
+    state: dict[str, Any] = {"total": 1, "messages_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
 
 
@@ -43,6 +56,16 @@ 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",
@@ -73,16 +96,16 @@ def test_logs_pagination(fake_read: dict[str, Any], client_factory) -> None:
     assert fake_read["kwargs"]["offset"] == 50
 
 
-def test_logs_clickhouse_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
+def test_logs_postgres_error(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:
     async def fetch_events(settings: Any, **kwargs: Any) -> tuple[list[dict[str, Any]], int]:
-        raise RuntimeError("ClickHouse недоступен")
+        raise RuntimeError("Postgres недоступен")
 
     monkeypatch.setattr(analytics_read, "fetch_events", fetch_events)
 
     response = client_factory().get("/logs")
 
     assert response.status_code == 200
-    assert "ClickHouse недоступен" in response.text
+    assert "Postgres недоступен" in response.text
 
 
 def test_logs_bad_date(client_factory, monkeypatch: pytest.MonkeyPatch) -> None:

tests/test_alerts.py

@@ -35,17 +35,14 @@ class FakeBot:
         self.forwarded.append((chat_id, from_chat_id, message_id))
 
 
-class FakeClickHouse:
-    """Заглушка клиента ClickHouse (см. тесты аналитики)."""
+class FakeWriter:
+    """Заглушка записи аналитики в Postgres (см. тесты аналитики)."""
 
     def __init__(self) -> None:
-        self.rows: list[list[Any]] = []
+        self.events: list[list[Any]] = []
 
-    async def insert(self, table: str, data: list[list[Any]], column_names: list[str]) -> None:
-        self.rows.extend(data)
-
-    async def close(self) -> None:
-        pass
+    async def __call__(self, events: list[list[Any]], messages: dict[Any, Any]) -> None:
+        self.events.extend(events)
 
 
 @pytest.fixture(autouse=True)
@@ -147,8 +144,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 = FakeClickHouse()
-    await init_analytics(settings, client=fake)
+    fake = FakeWriter()
+    await init_analytics(settings, writer=fake)
     bot = FakeBot()
 
     assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is True
@@ -159,15 +156,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.rows] == [EventType.BOT_ALIVE, EventType.BOT_ALIVE]
+    assert [row[1] for row in fake.events] == [EventType.BOT_ALIVE, EventType.BOT_ALIVE]
 
 
 async def test_ping_hourly_records_event_without_alert_chat(
     settings_factory: Callable[..., Settings], monkeypatch: pytest.MonkeyPatch
 ) -> None:
     settings = _patch_settings(monkeypatch, settings_factory(alert_chat_id=0))
-    fake = FakeClickHouse()
-    await init_analytics(settings, client=fake)
+    fake = FakeWriter()
+    await init_analytics(settings, writer=fake)
     bot = FakeBot()
 
     assert await heartbeat.ping_hourly(bot, now=datetime(2026, 1, 1, 10, 0, tzinfo=UTC)) is False
@@ -176,4 +173,4 @@ async def test_ping_hourly_records_event_without_alert_chat(
     await close_analytics()
 
     assert bot.messages == []
-    assert [row[1] for row in fake.rows] == [EventType.BOT_ALIVE]
+    assert [row[1] for row in fake.events] == [EventType.BOT_ALIVE]

tests/test_analytics.py

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