Как устроен мой Telegram-бот на aiogram 3 и FastAPI, какие грабли встретились по дороге и где заканчиваются возможности платформы

Сразу о том, что вы читаете: я автор сервиса, о котором пойдёт речь, и это разбор его устройства, а не приглашение им пользоваться. Меня самого в чужих статьях интересует не «что умеет продукт», а «на чём он сломался и как это починили», поэтому и здесь упор на второе.

Исходная задача простая. В Telegram у сообщения нет истории правок и нет корзины. Собеседник может удалить сообщение или тихо исправить в нём сумму, и у вас остаётся только память. Мне нужно было автоматически вести архив таких событий: копию удалённого, разницу между старой и новой версией, время, через которое это произошло.

Что Telegram разрешает на самом деле

Никаких клиентских сессий, кодов из СМС и обходов здесь нет. Бот работает через официальную функцию Telegram Business: в настройках аккаунта есть раздел «Автоматизация чатов», куда можно подключить бота к выбранным личным чатам. После этого Telegram присылает боту три типа событий:

• business_message — новое сообщение в подключённом чате;

• edited_business_message — сообщение изменили;

• deleted_business_messages — одно или несколько сообщений удалили.

Из этого вытекает ключевое ограничение, на котором держится вся архитектура. Событие об удалении не содержит текста. В нём только идентификаторы удалённых сообщений. Значит, восстановить удалённое можно единственным способом: сохранить сообщение в момент получения. Бот не видит прошлое, он ведёт архив с момента подключения. Эту оговорку я выношу в самое первое сообщение пользователю, потому что от неё зависят ожидания от сервиса.

Стек и общая схема

Python 3.12, aiogram 3, FastAPI, PostgreSQL, Redis с очередью ARQ, S3-совместимое хранилище для медиа. Всё живёт на одном небольшом VPS с 2 ГБ памяти.

Порядок обработки такой:

1. Telegram присылает апдейт на вебхук (FastAPI).

2. Апдейт сразу записывается в таблицу-«почтовый ящик» (inbox) в транзакции.

3. Только после успешной записи сервер отвечает 200.

4. Отдельный воркер ARQ разбирает inbox: сохраняет сообщение, скачивает медиа, шлёт уведомление.

Упрощённый вариант обработчика вебхука:

@app.post("/webhook/{secret}")
async def webhook(secret: str, request: Request):
    if secret != settings.webhook_secret:
        raise HTTPException(status_code=404)

    update = await request.json()
    try:
        async with db.transaction():
            await db.execute(
                """
                INSERT INTO inbox (update_id, payload)
                VALUES ($1, $2)
                ON CONFLICT (update_id) DO NOTHING
                """,
                update["update_id"], json.dumps(update),
            )
    except Exception:
        # не смогли записать: пусть Telegram повторит доставку
        return Response(status_code=503)

    await queue.enqueue_job("process_update", update["update_id"])
    return Response(status_code=200)

Здесь заложены два принципа, которые себя оправдали.

Сначала записать, потом отвечать. Если ответить 200 до записи и упасть, событие потеряно навсегда: Telegram считает его доставленным. Если записать и только потом ответить, любой сбой превращается в 503, и Telegram сам повторит попытку. Для архива, где потеря события означает потерянное удалённое сообщение, другого варианта я не вижу.

Идемпотентность. Повторная доставка того же update_id ничего не дублирует: ON CONFLICT DO NOTHING отсекает дубль на уровне базы, а воркер проверяет, что уведомление по этому событию ещё не отправлялось. Без этого каждый ретрай от Telegram превращался бы в повторное уведомление пользователю.

Хранение данных и шифрование

Раз сервис хранит переписку, вопрос защиты стоит остро. Что сделано:

• Тексты, профили и файлы шифруются на уровне приложения (Fernet) до записи в БД и в S3.

• Срок хранения выбирает сам пользователь: 1, 7, 30 или 90 дней, после этого автоочистка.

• Есть команда, которая удаляет все данные пользователя сразу.

Честная оговорка, которую я так же прямо написал в политике конфиденциальности: это не сквозное шифрование. Ключ находится у оператора, то есть технически у меня. Шифрование на уровне приложения защищает от утечки дампа базы или бакета, но не от самого оператора. Выдавать это за «end-to-end» было бы неправдой, поэтому формулировка в документах такая, какая есть.

Грабли №1: «зелёный» сервис, который молча терял сообщения

Самая неприятная проблема за всё время. Проверка /health зелёная, сервис отвечает, а у Telegram копится очередь из сотен неотправленных событий.

Причина оказалась в мелочи. aiogram подставляет в LinkPreviewOptions служебные значения по умолчанию, которые Pydantic не умеет превратить в JSON. Любое сообщение со ссылкой падало на этапе сохранения, сервер отвечал 503, и Telegram аккуратно копил повторы.

Лечится одним параметром сериализации:

payload = update.model_dump(mode="json", exclude_none=True, exclude_unset=True)

Но главный вывод не про Pydantic, а про мониторинг. Смотреть нужно не только на /health своего сервиса, но и на то, что видит Telegram:

curl "https://api.telegram.org/bot<TOKEN>/getWebhookInfo"

В ответе важны поля pending_update_count (сколько событий ждёт доставки) и last_error_message (на чём последний раз упали). Если число растёт, а healthcheck зелёный, значит, ваш сервис отвечает, но не тем, что нужно Telegram. После этого случая у меня появился регрессионный тест «сообщение со ссылкой» и алерт на рост pending_update_count.

Грабли №2: смещения в entities считаются в UTF-16

Хотелось аккуратное уведомление: ссылка на собеседника, цитата, зачёркнутое и жирное внутри цитаты. Форматирование в Bot API описывается через MessageEntity со смещением (offset) и длиной (length). Загвоздка в том, что они измеряются в единицах UTF-16, а не в символах Python.

Эмодзи занимает две единицы UTF-16, но один символ в строке Python. Если не учитывать это, все смещения после первого эмодзи уезжают, и форматирование ложится не на те слова. Базовая функция тривиальна:

def utf16_len(text: str) -> int:
    return len(text.encode("utf-16-le")) // 2

Но на её основе пришлось собрать небольшой билдер RichText, который накапливает текст и сам считает смещения при каждом добавлении фрагмента. Второй его задачей стала нарезка длинных сообщений на части, чтобы лимит Telegram на длину не рвал сущности посередине: если граница попадает внутрь форматированного куска, билдер закрывает сущность в конце одной части и открывает заново в начале следующей.

Подсветка правок

Когда сообщение изменили, пользователю показывается старая и новая версия, где удалённое зачёркнуто, а добавленное выделено жирным. Разбор идёт по словам через difflib.SequenceMatcher:

from difflib import SequenceMatcher

def diff_words(old: str, new: str):
    a, b = old.split(), new.split()
    matcher = SequenceMatcher(None, a, b)
    if matcher.ratio() < 0.3:
        return None  # текст переписан почти целиком, подсветка бесполезна
    return matcher.get_opcodes(), a, b

Порог введён эмпирически. Если текст переписан больше чем на 70%, подсветка превращается в кашу из зачёркнутого и жирного, и читать такое хуже, чем две цельные версии подряд. В этом случае бот просто показывает старый и новый текст без разметки.

Фото с таймером: что известно и чего нет

Самый частый вопрос: сохраняет ли бот одноразовые фото. Ответ неоднозначный, и я описываю его пользователям именно так.

Telegram не присылает боту исчезающие фото отдельным событием. По моим наблюдениям, если ответить на такое сообщение до открытия, то в событии с вашим ответом внутри reply_to_message приходит вложение, и бот его забирает. Это работает только по ответу, и Telegram такого поведения официально не гарантирует: завтра оно может измениться без предупреждения. Поэтому в сервисе это описано как приём, а не как стабильная функция.

Отдельно стоит сказать об этической стороне. Отправитель исчезающего фото рассчитывает, что оно не сохранится. Технически архив обходит эти ожидания, и это та точка, где «можем» и «стоит ли» расходятся. Я оставил эту возможность, но считаю честным называть её прямо.

Сертификат

API банка подписан Russian Trusted Root CA, которого нет в стандартных CA-бандлах базового Docker-образа. Результат: CERTIFICATE_VERIFY_FAILED. Решение — добавить корневой сертификат Минцифры, причём только для HTTP-клиента, который ходит в банк, и сверить отпечаток с цепочкой. Подкладывать его во все исходящие соединения сервиса я не стал, чтобы не расширять доверие там, где оно не нужно.

СБП без платёжной формы

Метод GetQr возвращает ссылку, которая сразу открывает приложение банка пользователя. Но способ нужно отдельно включить на терминале, иначе придёт ошибка 3001.

Автопродление

Первый платёж идёт картой с Recurrent=Y, из уведомления сохраняется RebillId, а дальше воркер за сутки до списания напоминает пользователю и затем вызывает Charge. Платёж через СБП я делаю отдельным разовым, чтобы он не привязывал счёт для автосписаний. Условия автосписания показываются перед оплатой: это требование эквайринга, и обходить его нельзя.

Ограничения, о которых стоит знать заранее

• Работают только личные чаты. Так устроен Business API, группы недоступны.

• Сохраняется только то, что пришло после подключения. История до этого момента недоступна.

• Telegram не сообщает, кто именно удалил сообщение, поэтому в уведомлениях формулировки осторожные.

• Платформа может изменить поведение Business API или то, как приходят вложения, и сервис придётся подстраивать.

Вопрос согласия

Сервис хранит сообщения собеседников, а они об этом, как правило, не знают. Технически это законно работающая схема, но у неё есть очевидное напряжение: человек пишет в личной переписке и не ожидает, что правку или удаление зафиксируют. Это не отменяет пользы архива (спор с подрядчиком, подтверждение договорённости), но заставляет относиться к хранению серьёзно.

Что я сделал со своей стороны: сроки хранения выбирает пользователь, данные шифруются, любую информацию можно удалить в один клик, есть политика обработки персональных данных. Если вы делаете что-то подобное, сразу продумайте юридическую часть: вы фактически становитесь оператором чужих персональных данных, и в России это подразумевает вполне конкретные обязанности по закону о персональных данных. Это отдельная тема для консультации с юристом, а не для догадок.

Что я вынес

Пара технических выводов, применимых за пределами этого бота.

1. Для любого сервиса, принимающего вебхуки, «записать сначала, ответить потом» плюс идемпотентность по идентификатору события снимают целый класс проблем с потерями и дублями. Это дешевле, чем разбираться с ними постфактум.

2. Здоровье сервиса надо проверять глазами внешней системы. Мой healthcheck был зелёным всё время, пока Telegram копил очередь. Метрики вроде pending_update_count показывают то, чего не видно изнутри.

3. Единицы измерения в API стоит проверять сразу. Смещения в UTF-16 не сломают ничего, пока у вас нет эмодзи, и сломают всё, как только они появятся.

Сервис называется «Эхо диалогов», бот доступен как @udaleniechatov_bot. Упоминаю его как пример реализации описанного выше. Вопросы и критику, особенно по архитектуре очереди и работе с платёжным API, буду рад увидеть в комментариях.