Опрос событий

Лента событий сообщает о двух фактах: лиду отправлено первое сообщение и лид ответил. Цикл ниже не теряет и не дублирует события, даже после сбоев и простоев.

#Как устроена лента

GET /v1/events возвращает события в порядке их появления в ленте. Типов два:

Свойства ленты:

• Порядок по появлению, а не по времени факта. Поле timestamp показывает, когда произошел факт. Событие с более ранним timestamp может прийти позже соседнего. Не используйте timestamp как курсор.
• Доставка не реже одного раза. После сбоя страница может прийти повторно. Ключ дедупликации: event.id вида evt_ и 32 шестнадцатеричных знака.
• Идентификаторы совпадают с вебхуками. Если в кабинете настроены вебхуки, у того же факта тот же id. Ленту и вебхуки можно использовать вместе, дедуплицируя по id.
• Курсор есть всегда. next_cursor никогда не равен null, даже у пустой страницы. Каждый ответ выдает новый курсор, его нужно сохранить.
• Хранение 30 дней. События старше 30 дней из ленты уходят. Курсор живет 30 суток с момента выдачи и продлевается каждым запросом.

Что важно знать о данных события:

• lead.replied приходит на каждый засчитанный ответ, у одного лида их может быть несколько.
• data.reply.automation_stopped сейчас всегда null: остановку цепочки событие не подтверждает. Поле data.reply.is_first_reply сейчас не передается.
• data.first_message_sent_at у события ответа часто равен null. Время первого сообщения надежнее брать из события lead.first_message_sent или из карточки лида.
• data.lead.fields.record_id сейчас равен null, record_id_status равен unknown. Как связать событие с CRM, описано в руководстве Связь с CRM.
• data.lead.username приходит в нижнем регистре.

#Первый запуск

Есть два способа начать:

• С текущего момента. Запрос с start=now возвращает пустую страницу и курсор. Лента начнется с событий, которые появятся после него. Прошлое состояние заберите обходом кампаний и лидов.
• С доступной истории. Запрос без cursor и без start начинает с самого раннего события, которое хранит лента, но не раньше 30 дней и не раньше подключения кабинета к сбору событий.

Для интеграции с CRM обычно удобнее первый способ: сначала получить курсор, затем сверить карточки, затем читать ленту от курсора. Так ни одно изменение во время сверки не потеряется.

#Надежный цикл

1. Загрузите сохраненный курсор. Если его нет, выполните первый запуск.
2. Запросите GET /v1/events?cursor=<курсор>&limit=100.
3. В одной транзакции вашей базы примените события страницы и сохраните новый next_cursor. Пропускайте события, id которых уже применены.
4. Если has_more равен true, сразу запросите следующую страницу.
5. Если has_more равен false, подождите минуту и повторите с шага 2.

Почему транзакция: если процесс упадет между применением событий и сохранением курсора, после перезапуска он повторит ту же страницу, а дедупликация по id не даст применить событие дважды. Если сохранить курсор раньше, чем применить события, при сбое они потеряются.

#Ошибки при опросе

#Восстановление после простоя

Восстановление нужно после ответа 410 или cursor_mismatch, после простоя дольше 30 дней и когда meta.coverage.status сообщает о возможной неполноте.

1. Отметьте у себя разрыв: время, когда поток прервался.
2. Получите новый якорь: GET /v1/events?start=now. Сохраните его курсор только после шага 3.
3. Обойдите текущие кампании и лидов и обновите карточки в своей системе. Поля first_message_sent_at, first_reply_at и status лида покажут, что произошло за время разрыва.
4. Сохраните курсор якоря и продолжайте обычный цикл.

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

#Полнота ленты

Каждая страница содержит meta:

• meta.available_from: с какого момента в ленте есть события.
• meta.coverage.status: complete, partial или unknown. В раннем доступе обычно приходит unknown или partial: полнота ленты не гарантируется. Для важных процессов периодически, например раз в сутки, сверяйте карточки лидов.
• meta.coverage.gaps: интервалы, за которые события могли собраться не полностью. После такого интервала сверьте лидов затронутых кампаний.
• meta.collection.state: состояние сбора. degraded означает, что данные могут быть неполными.

#Тексты сообщений

С параметром include_text=true событие содержит текст: у lead.first_message_sent в data.first_message, у lead.replied в data.reply. Нужны право messages:read и включенный у ключа «Текст сообщений». Текст есть, только если сохранен и его принадлежность переписке доказана; иначе text равен null, а причина в text_unavailable_reason. Запрос с текстом весит 3 вместо 2.

#Пример на Node.js

Функции хранилища замените своими. meliora() с повторами при 429 и 503 описана в руководстве Лимиты и повторы.

const POLL_INTERVAL_MS = 60_000;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export async function runEventLoop(db) {
  let cursor = await db.loadCursor();
  if (!cursor) cursor = await resync(db);

  for (;;) {
    let page;
    try {
      page = await meliora('/events', { cursor, limit: 100 });
    } catch (err) {
      if (err.status === 410 || err.code === 'cursor_mismatch') {
        cursor = await resync(db);
        continue;
      }
      throw err; // 401 и 403: остановиться и проверить ключ
    }

    await db.transaction(async (tx) => {
      for (const event of page.data) {
        if (await tx.isApplied(event.id)) continue;
        if (event.type === 'lead.first_message_sent' || event.type === 'lead.replied') {
          await applyToCrm(tx, event);
        }
        await tx.markApplied(event.id); // незнакомый тип тоже отмечаем
      }
      await tx.saveCursor(page.next_cursor);
    });
    cursor = page.next_cursor;

    if (!page.has_more) await sleep(POLL_INTERVAL_MS);
  }
}

async function resync(db) {
  await db.recordGap(new Date());
  const anchor = await meliora('/events', { start: 'now' });
  await reconcileLeads(db); // обход /v1/leads и обновление карточек
  await db.saveCursor(anchor.next_cursor);
  return anchor.next_cursor;
}

#Пример на Python

import time
from datetime import datetime, timezone

POLL_INTERVAL = 60


def run_event_loop(db):
    cursor = db.load_cursor() or resync(db)
    while True:
        try:
            page = meliora("/events", {"cursor": cursor, "limit": 100})
        except MelioraError as err:
            if err.status == 410 or err.code == "cursor_mismatch":
                cursor = resync(db)
                continue
            raise  # 401 и 403: остановиться и проверить ключ

        with db.transaction() as tx:
            for event in page["data"]:
                if tx.is_applied(event["id"]):
                    continue
                if event["type"] in ("lead.first_message_sent", "lead.replied"):
                    apply_to_crm(tx, event)
                tx.mark_applied(event["id"])  # незнакомый тип тоже отмечаем
            tx.save_cursor(page["next_cursor"])
        cursor = page["next_cursor"]

        if not page["has_more"]:
            time.sleep(POLL_INTERVAL)


def resync(db):
    db.record_gap(datetime.now(timezone.utc))
    anchor = meliora("/events", {"start": "now"})
    reconcile_leads(db)  # обход /v1/leads и обновление карточек
    db.save_cursor(anchor["next_cursor"])
    return anchor["next_cursor"]

Функция meliora() и класс MelioraError приведены в руководстве Лимиты и повторы. Как сопоставить событие с записью CRM, описано в руководстве Связь с CRM.

#Как часто опрашивать

Раз в минуту достаточно для большинства интеграций, это 2 весовые единицы в минуту из лимита ключа. Реже тоже можно, например раз в час: курсор живет 30 суток, а лента хранит события 30 дней. Пока has_more равен true, забирайте страницы подряд без паузы. Более частый опрос не ускоряет появление событий: они попадают в ленту по мере сбора данных, и время этой задержки не гарантируется.