Как я сделал инкрементальный архив Telegram-каналов: gramjs, аварийные чекпоинты и один fetch на два хранилища

Мне понадобился локальный архив нескольких Telegram-каналов: читаемый человеком, пригодный для SQL-запросов и способный обновляться по расписанию без дублей. Не очередной облачный сервис и не одноразовый экспорт, а процесс, который можно безопасно запускать каждые несколько минут.

Так появился getmes-tg — открытая CLI-утилита на Node.js и gramjs. Она выгружает доступную аккаунту историю в Markdown и MySQL, умеет восстанавливать прогресс после сбоя и собирается через Bun в самостоятельные бинарники для Windows, Linux и macOS.

В этой статье — не пересказ README, а инженерная часть проекта: почему для инкрементальной выгрузки нужны одновременно minId и offsetId, почему служебный state нельзя считать единственным источником истины и как один проход по Telegram безопасно разделить между двумя хранилищами.

Задача и инварианты

Одноразово скачать историю несложно. Сложность начинается со второго запуска.

Я сформулировал пять инвариантов:

1. Повторный запуск не создаёт дублей.

2. Сбой между записью результата и сохранением прогресса не теряет сообщения.

3. Файловый архив и база могут работать независимо.

4. Совместный режим читает канал один раз, даже если пишет в два места.

5. Telegram-разметка сохраняется в Markdown, включая случаи с эмодзи и вложенными сущностями.

Получилась такая схема:

                         ┌──────────────► output/<канал>.md
Telegram API ─► fetch ───┤
                         └──────────────► MySQL
                              │
                              └─────────► state/export.yaml
                                          state/db.yaml

Снаружи это четыре команды:

minId и offsetId решают разные задачи

Наивное описание пагинации часто сводят к фразе «используйте minId вместо offsetId». Это неточно: в getmes-tg используются оба параметра, но для разных целей.

minId задаёт нижнюю границу инкрементальной выборки: нас интересуют только сообщения с id больше последнего сохранённого.

const params = { limit: 100 };

if (lastMessageId) {
  params.minId = lastMessageId;
}

offsetId нужен уже внутри пагинации: Telegram возвращает сообщения от новых к старым, поэтому после каждой страницы курсор сдвигается к последнему элементу батча.

params.offsetId = batch[batch.length - 1].id;
await new Promise((resolve) => setTimeout(resolve, 500));

Иными словами:

• minId отвечает на вопрос «где заканчиваются уже сохранённые данные»;

• offsetId — «с какой точки продолжать обход текущей истории».

После получения батча сообщения дополнительно фильтруются по id, а итоговый массив сортируется от старых к новым. Для первой выгрузки вместо чекпоинта используется заданная в конфиге startDate.

Пауза между страницами намеренная: первоначальный импорт большого канала не должен превращаться в соревнование с FloodWait.

Почему одного state-файла недостаточно

Предположим, программа выполнила два действия:

1. записала сообщение #120 в Markdown;

2. сохранила lastMessageId: 120 в служебном файле.

Если процесс завершится между этими действиями, возможны две плохие стратегии:

• записывать state первым — тогда при сбое сообщение потеряется;

• записывать результат первым и верить только старому state — тогда сообщение придёт повторно.

В проекте результат всегда записывается первым, а эффективный чекпоинт вычисляется как максимум из state и самого артефакта:

export function effectiveExportCheckpoint(entry, fileMax = 0) {
  const fromState = entry?.lastMessageId
    ? Number(entry.lastMessageId)
    : 0;
  const fromFile = fileMax ? Number(fileMax) : 0;

  return Math.max(fromState, fromFile);
}

Для Markdown вершина восстанавливается по заголовкам ## Сообщение #N. Для базы используется MAX(message_id) по каналу.

Таким образом, state — ускоряющая метаинформация, но не единственный источник истины. Если удалить каталог state/, следующий запуск восстановит границы из файлов и базы. Если приложение упадёт после записи артефакта, но до обновления state, более свежий артефакт победит.

Сами YAML-файлы записываются атомарно через временный файл и rename:

const tmp = `${file}.tmp`;
fs.writeFileSync(tmp, stringifyYaml(state), "utf-8");
fs.renameSync(tmp, file);

У export и db разные state-файлы. Это важно: пользователь может несколько дней вести только файловый архив, а потом включить базу — каждый режим продолжит со своей границы.

База защищает от дублей независимо от приложения

Проверки в коде полезны, но гарантию лучше опускать на уровень хранилища. В MySQL идентичность сообщения задаёт составной первичный ключ:

PRIMARY KEY (channel_id, message_id)

Запись выполняется через ON DUPLICATE KEY UPDATE. Поэтому повторная доставка сообщения не создаст вторую строку даже при ошибке в логике приложения или ручном перезапуске.

На этом же свойстве основан режим --repair. Он игнорирует чекпоинт и заново читает историю с startDate:

• MySQL получает полный идемпотентный upsert;

• Markdown предварительно собирает множество уже существующих id и дописывает только отсутствующие.

Ремонт получается не отдельной магией, а следствием тех же инвариантов.

Один fetch для файла и базы

Первоначально режимы export и db можно запускать независимо. Но последовательный запуск означает две авторизации, две пагинации и две серии пауз.

В full сначала вычисляются два чекпоинта. Общая выборка начинается с меньшего из них:

const fetchFrom = cpExport > 0 && cpDb > 0
  ? Math.min(cpExport, cpDb)
  : 0;

const messages = await fetchChannelMessages(client, {
  ...channelConfig,
  lastMessageId: fetchFrom,
});

const { forFile, forDb } = partitionByCheckpoints(
  messages,
  cpExport,
  cpDb,
);

Если файл отстаёт от базы, общая выборка начинается с файлового чекпоинта, но в MySQL проходят только сообщения новее его собственной границы. Если отстаёт база — происходит обратное.

Критически важно, что две части сохраняются независимо. Ошибка записи файла не отменяет успешный upsert в БД, и наоборот; чекпоинт продвигается только у успешно завершившегося потребителя. При частичном сбое full возвращает ненулевой exit code, чтобы cron или мониторинг увидели проблему.

Telegram-разметка и UTF-16

Текст Telegram — это исходная строка плюс массив сущностей: bold, italic, URL, code, mention и другие. Их offset и length заданы в координатах UTF-16. Поэтому эмодзи перед форматируемым фрагментом ломает алгоритм, который считает позиции как Unicode code points.

Есть и вторая ловушка gramjs: msg.text уже может содержать Markdown. Если поверх него ещё раз наложить entities, появляются конструкции вроде ****текст****. В проекте базой служит msg.rawText.

Сущности преобразуются в события открытия и закрытия тегов. Они сортируются по позиции, причём на одной позиции закрывающие события идут раньше открывающих, а внешняя сущность открывается раньше внутренней.

events.sort((a, b) => {
  if (a.pos !== b.pos) return a.pos - b.pos;
  if (a.kind !== b.kind) return a.kind === "close" ? -1 : 1;
  if (a.span !== b.span) {
    return a.kind === "open" ? b.span - a.span : a.span - b.span;
  }
  return a.kind === "open" ? b.idx - a.idx : a.idx - b.idx;
});

После этого строка собирается одним проходом. Такой подход не меняет исходные координаты во время обработки и корректно работает с соседними и вложенными сущностями.

Что обнаружилось в gramjs

В процессе проявились несколько неочевидных особенностей библиотеки.

Первая: client.connect() при неудаче может вернуть false, а не бросить исключение. Если результат не проверить, следующий этап авторизации ведёт себя как зависший, а процесс способен завершиться с кодом 0.

Вторая: для окончательного завершения клиента используется client.destroy(). Обычный disconnect() оставлял внутренний цикл и задерживал завершение процесса примерно на 30 секунд.

Третья — уже упомянутая разница между msg.text и msg.rawText.

Эти детали кажутся локальными, но именно из них складывается CLI, которому можно доверить cron: он должен не только получить данные, но и честно сообщить автоматике, что произошло.

Эксплуатация и проверка

У меня утилита работает в Docker Compose: контейнер с cron запускает файловую и MySQL-выгрузку каждые три минуты. Конфиг, сессия, state, логи и результаты монтируются с хоста; MySQL находится в той же изолированной сети.

Фрагмент реального вывода одного из запусков:

РИА Новости: +3 → output/РИА Новости.md
MarketTwits: нет новых
Готово: 8 каналов, +3 сообщений за 4.7 с

Время зависит от сети и числа страниц, поэтому воспринимать эту строку как бенчмарк не стоит.

Тестовый набор содержит 203 проверки в шести файлах. На момент подготовки статьи npm test проходит полностью. Тесты не ходят в Telegram и живую MySQL: они проверяют преобразование entities, CLI и exit codes, конфигурацию БД и прокси, логгер, восстановление state и разделение общей выборки между режимами.

Сетевые сценарии остаются интеграционной проверкой. Это сознательное ограничение, а не попытка выдать unit-тесты за end-to-end.

Что проект пока не решает

• Два параллельных запуска одного режима не поддерживаются: у них общие state и артефакты. Cron нужно остановить перед ручным запуском того же режима.

• Инструмент работает от имени вашего аккаунта и только с теми каналами, которые ему доступны. Ограничения Telegram он не обходит.

• Медиа сейчас фиксируется в архиве как класс вложения, а не скачивается целиком.

• Сборки ARM64 есть только как закомментированные цели Makefile; опубликованы x64-архивы.

• MySQL нужен лишь для db и full; простой Markdown-архив обходится без базы.

Отдельно о безопасности: session.txt фактически даёт доступ к Telegram-сессии. Его нельзя коммитить, отправлять вместе с архивом или хранить в общедоступной папке. В репозитории сессия и реальные конфиги исключены через .gitignore и .dockerignore.

Как попробовать

Самый короткий путь — взять архив для своей платформы из предварительного Release0. Там опубликованы x64-сборки для Windows, Linux и macOS Intel с SHA-256 digest.

Для запуска из исходников:

git clone https://github.com/fdima/getmes-tg
cd getmes-tg
npm install
cp config.yaml.example config.yaml
node export.js export

В config.yaml понадобятся apiId и apiHash, полученные на my.telegram.org, список каналов и дата начала истории. Первый запуск интерактивно создаст локальную сессию, последующие будут инкрементальными.

Для самостоятельной сборки одного файла:

bun build ./export.js --compile \
  --target=bun-linux-x64 \
  --outfile dist/getmes-tg-linux-x64

Подробные команды, конфигурация MySQL, Docker Compose и режимы full/--repair описаны в README. Код распространяется по лицензии MIT.

Выводы

Главный результат проекта для меня — не конкретный CLI, а три более общих решения:

1. Прогресс инкрементального процесса стоит восстанавливать из результата, а не только из служебного state.

2. Идемпотентность лучше обеспечивать одновременно алгоритмом и ограничениями хранилища.

3. Если два потребителя читают один дорогой источник, выгоднее сделать общую выборку и разделить её по независимым чекпоинтам.

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