Расширенное руководство по разработке экшнов

Руководство для сторонних разработчиков. Описывает контракт между вашим сервисом и платформой Accelera (модули Flow и Segments): очереди, форматы сообщений, регистрацию, интерфейс настройки, а также правила, без которых действие работает на стенде, но ломается в бою.

Содержание

1. Что такое кастомное действие — виды, жизненный цикл, глоссарий, быстрый старт
2. Требования к окружению
3. Топология RabbitMQ
4. Регистрация действия
5. Интерфейс настройки (html, script, webhooks)
6. Flow-действие
7. Segment-действие
8. Контрольные сигналы (stop / pause / resume)
9. Надёжность: ack, повторы, идемпотентность, падения
10. Эксплуатация: логи, метрики, секреты, остановка
11. Изменение и версионирование действия
12. Локальная разработка и тестирование
13. Что согласовать с администратором установки
14. FAQ
15. Частые ошибки
16. Чек-лист перед запуском
17. Скелеты реализации

Как читать. Разделы 1–4 обязательны всем. Дальше: делаете flow-действие — 5, 6, 9; segment-действие — 5, 7, 8, 9. Перед выкладкой — 12, 13, 16. Справочник форматов всех сообщений собран в приложении A.

1. Что такое кастомное действие

Кастомное действие (custom action, «экшен») — это блок, который бизнес-пользователь ставит в сценарий или в обработку сегмента, а работу за этим блоком выполняет ваш сервис. Платформа отвечает за редактор, запуск и учёт; ваш сервис — за то, что происходит во внешней системе (отправить SMS, выгрузить аудиторию, записать данные в кеш, вызвать API заказчика).

Связь только через RabbitMQ. Язык и стек — любые, нужна библиотека AMQP 0-9-1 (список клиентов). Действующие реализации написаны на Node.js и Go.

Есть два вида действий, и устроены они по-разному:

Один сервис может реализовывать оба вида и сколько угодно действий каждого вида.

Жизненный цикл

 Ваш сервис                         RabbitMQ                          Платформа
 ──────────                         ────────                          ─────────
 старт ── карточка действия ──────▶ FLOWS.*-custom-actions-register ─▶ бэкофис: блок в палитре
                                                                       │
                                                                       │ пользователь настраивает
                                                                       │ блок (ваш html/script)
                                                                       ▼
 consume ◀──────── задача ───────── ex-custom-actions (flow)  ◀──────── Flow: событие дошло до блока
                                    ex-segment-custom-actions.{svc} ◀── Segments: запуск сегмента
   │
   │ работа во внешней системе
   ▼
 flow:    событие ─────────────────▶ ex-triggers ─────────────────────▶ Flow: сценарий продолжается
 segment: статус  ─────────────────▶ FLOWS.custom-actions-statuses ───▶ Segments: шаг завершён
                                                                       │
 consume ◀──── stop ─────────────── ex-segments-control.{svc} ◀─────── пользователь нажал «Стоп»

Глоссарий

Быстрый старт

Минимальный путь от нуля до работающего блока:

1. Получите у администратора параметры установки (полный список — раздел 13): адрес RabbitMQ, учётные данные, SEGMENTS_SERVICE_NAME, режим очередей, доступ к БД сегментов.
2. Выберите имя действия с префиксом своей компании (acme-sms) и имена своих очередей.
3. Возьмите скелет из раздела 17 для своего вида действия.
4. Поднимите локальный стенд (раздел 12) и прогоните задачи руками: свою, чужую, битую, тестовую, со stop.
5. Запустите сервис против тестовой установки — блок появится в палитре редактора.
6. Соберите в бэкофисе тестовый сценарий/сегмент, настройте блок, запустите.
7. Пройдите чек-лист и выкладывайте.

2. Требования к окружению

3. Топология RabbitMQ

3.1 Сводная таблица

{svc} — значение SEGMENTS_SERVICE_NAME. {q} — суффикс .quorum, если установка работает на quorum-очередях (к обменникам не применяется).

Публикация в очереди FLOWS.* идёт через обменник по умолчанию (""), ключ маршрутизации — имя очереди.

Объявляйте (**assert**/**declare**) всё, во что публикуете, на каждом новом канале, с теми же параметрами, что у платформы. Если после переподключения вы попадёте на ноду, где очереди ещё нет, публикация без объявления молча пропадёт.

3.2 Свои очереди: имя решает всё

Обменники задач — fanout. Это значит:

• каждая очередь, привязанная к обменнику, получает копию каждой задачи — в том числе задачи чужих действий других сервисов;
• инстансы, объявившие очередь с одним и тем же именем, делят задачи между собой (round-robin) — так масштабируется ваш сервис;
• инстансы с разными именами очередей получат каждую задачу по разу каждый — получится дублирование.

Отсюда правила:

1. Имя очереди задач — уникальное для вашего сервиса и одинаковое для всех его инстансов. Например acme-actions для flow и acme-segment-actions.{svc} для сегментов.
2. Никогда не используйте имя чужой очереди (например, utility-actions). Вы начнёте отбирать задачи у другого сервиса и подтверждать их как «чужие» — задачи будут теряться.
3. Имя actions для очереди flow-задач запрещено — оно занято платформой.
4. Очереди задач — durable.

Очередь контрольных сигналов — наоборот, своя у каждого инстанса (см. раздел 8).

3.3 Classic и quorum

Установка работает в одном из режимов, и ваш сервис обязан работать в том же:

Последствия несовпадения:

• если платформа читает FLOWS.custom-actions-statuses.quorum, а вы пишете в FLOWS.custom-actions-statuses, статусы копятся в очереди, которую никто не читает, и сегменты «висят» в обработке;
• если очередь с таким именем уже существует с другими аргументами, объявление падает с PRECONDITION_FAILED, и сервис не стартует.

Сделайте режим настройкой (USE_QUORUM_QUEUES=true|false), а не константой.

Переход classic → quorum. Новая *.quorum-очередь привязывается к тому же fanout-обменнику, а старая classic-очередь остаётся привязанной и продолжает получать копии всех задач, которые никто не читает. При старте в quorum-режиме удаляйте свою старую classic-очередь (отдельным каналом, ошибку удаления не считайте фатальной). При обратном переходе — наоборот.

4. Регистрация действия

Действие появляется в палитре бэкофиса после того, как ваш сервис опубликует его карточку в очередь регистрации. Карточка публикуется при каждом старте сервиса: платформа обновляет запись по name (upsert), так что новая версия интерфейса доезжает до пользователей с перезапуском.

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

4.1 Общие поля карточки

Служебные флаги в data:

Даже если форма целиком строится вашим script, все параметры должны быть перечислены в **data** с дефолтными значениями.

4.2 Карточка flow-действия

Очередь: FLOWS.flows-custom-actions-register{q}.

{
  "name": "acme-sms",
  "title": "SMS (ACME)",
  "class": "stateless-action",
  "icon": "<svg width=\"18\" height=\"18\" class=\"fe\" viewBox=\"0 0 24 24\">...</svg>",
  "html": "<div>...разметка блока...</div>",
  "inputs": 1,
  "outputs": 0,
  "data": {
    "svgIcon": true,
    "isCustom": true,
    "isCustomJS": true,
    "isCustomHTML": true,
    "returnable": false,
    "phone": "",
    "text": "",
    "eventname": ""
  },
  "paramsDescription": "{}",
  "script": "$(function () { ... })",
  "webhooks": null
}

4.3 Карточка segment-действия

Очередь: FLOWS.segments-custom-actions-register{q}.

Интерфейс настройки можно описать декларативно через parameters (бэкофис сам построит форму) или полностью своим html + script.

{
  "name": "acme-audience-upload",
  "title": "Выгрузка в ACME Audience",
  "icon": "fe fe-user-plus",
  "data": {
    "audienceName": "",
    "idColumn": "",
    "mode": "append",
    "throttle": ""
  },
  "parameters": [
    { "name": "audienceName", "label": "Название аудитории", "type": "input" },
    { "name": "idColumn",     "label": "Колонка с ID клиента", "type": "inputSingleColumn" },
    { "name": "mode",         "label": "Режим", "type": "dropdown", "options": ["append", "replace"] },
    { "name": "throttle",     "label": "Строк в секунду", "type": "number" }
  ],
  "webhooks": null
}

parameters[].name должен совпадать с ключом в data.

Типы полей parameters:

Пример list и dynamicList:

{
  "name": "channels",
  "type": "dynamicList",
  "fields": [
    { "name": "Channel",  "type": "dropdown", "options": ["sms", "push", "email"] },
    { "name": "Template", "type": "number" }
  ]
}

Рекомендуемые CSS-иконки: fe fe-user-plus, fe fe-user-minus, fe fe-user-x, fe fe-user-check, fe fe-link, fe fe-rss, fe fe-mail, fe fe-message-circle.

4.4 Правила регистрации

• Публикуйте с persistent: true и дожидайтесь подтверждения брокера (publisher confirms). Если регистрация не прошла — это повод не стартовать: без карточки задачи всё равно не придут.
• Имя, под которым действие зарегистрировано, и имя, по которому вы фильтруете задачи, должны браться из одной константы. Иначе карточка будет в палитре, а задачи с её именем ваш сервис будет считать чужими и молча выбрасывать.
• Если вы включаете действия по списку из конфигурации, проверяйте этот список при старте и громко падайте на неизвестном имени — опечатка в списке иначе молча выключает действие.
• Не регистрируйте заглушки и демо-действия в боевых установках: пользователь видит их наравне с рабочими.

5. Интерфейс настройки

5.1 Как это устроено

Редактор бэкофиса построен на Drawflow, jQuery и Bootstrap 4. Ваш html — это разметка блока на холсте, ваш script выполняется в странице редактора при загрузке и обычно:

1. добавляет модальное окно настройки в контейнер .drawflow-custom-modals;
2. при открытии окна заполняет поля сохранёнными значениями;
3. по кнопке «Сохранить» собирает значения и передаёт их платформе.

Модальное окно должно иметь класс custom-modal-{name} — по нему бэкофис открывает окно при клике на блок.

Скрипт исполняется в общей странице вместе со скриптами всех остальных действий:

• оборачивайте всё в $(function () { ... }) и не объявляйте глобальных переменных;
• все селекторы делайте с префиксом .custom-modal-{name}, иначе зацепите чужие формы;
• стили — с тем же префиксом;
• если скрипт собирается из шаблона, значение **name** внутри скрипта должно совпадать с **name** карточки (типовая ошибка при копировании: форма открывается, но «Сохранить» не срабатывает, потому что селектор указывает на модалку другого действия).

5.2 Функции бэкофиса: flow и segment отличаются

Перепутать формат сохранения — самая частая ошибка: данные не сохранятся или сохранятся не туда.

Flow:

$('.custom-modal-acme-sms').on('shown.bs.modal', function () {
    const params = get_data_custom_action();
    $('.custom-modal-acme-sms .phone').val(params.data.phone || '');
});

$('.custom-modal-acme-sms .save').on('click', function () {
    save_data_custom_action([
        { Parameter: 'phone',     Value: $('.custom-modal-acme-sms .phone').val() },
        { Parameter: 'text',      Value: $('.custom-modal-acme-sms .text').val() },
        { Parameter: 'eventname', Value: $('.custom-modal-acme-sms .eventname').val() },
    ]);
});

Segment:

$(document).on('show.bs.modal', '.custom-modal-acme-audience-upload', function () {
    const saved = editor.getSplitNode(node_selected).data || {};
    const columns = (check_inputs_data(node_selected) || {}).columns || [];
    const $col = $('.custom-modal-acme-audience-upload .idColumn').empty();
    columns.forEach(c => $col.append(new Option(c, c)));
    if (saved.idColumn) $col.val(saved.idColumn);
});

$('.custom-modal-acme-audience-upload .save').on('click', function () {
    save_data_segment_custom_action({
        isCustomJS: true,
        idColumn: $('.custom-modal-acme-audience-upload .idColumn').val(),
        throttle: $('.custom-modal-acme-audience-upload .throttle').val(),
    });
});

5.3 Валидация в форме

Проверяйте обязательные поля до сохранения: в обработчике кнопки вызывайте e.preventDefault(); e.stopPropagation(), чтобы окно с data-dismiss="modal" не закрылось, и показывайте ошибку. Но проверяйте те же поля и на стороне сервиса: значения могли быть сохранены старой версией формы или не заполнены вовсе.

5.4 Webhooks

Если форме нужны данные из вашей системы (список аудиторий, шаблонов, кампаний), укажите в карточке webhooks — базовый URL вашего HTTP-сервиса. Скрипт формы вызывает:

custom_webhooks('acme-audience-upload', { query: 'list-audiences' })
    .then(res => { /* заполнить выпадающий список */ });

Бэкофис отправит POST {webhooks}/webhooks/{name} с телом-объектом и вернёт ответ в промис.

Требования к обработчику:

• всегда отвечать: на неизвестное действие — 404, на ошибку — 500 с текстом. Не ответить — значит подвесить форму пользователя до таймаута бэкофиса;
• отвечать быстро (единицы секунд), тяжёлое — кешировать;
• слушать на адресе, доступном бэкофису (0.0.0.0, а не 127.0.0.1);
• аутентификации у вызова нет — не отдавайте через webhook ничего, что нельзя показать любому пользователю бэкофиса, и не открывайте этот порт наружу.

Можно также ходить из скрипта формы в API самого бэкофиса (например, справочники /flow/api/v1/data-dictionary/list) — запрос идёт от имени пользователя.

Если значения для формы зависят от конфигурации вашего сервиса (например, имена справочников), их можно подставить в текст script на стороне сервиса при формировании карточки — скрипт публикуется заново при каждом старте.

5.5 Разметка блока (html)

Это то, как блок выглядит на холсте. Используйте шаблоны ниже — они повторяют стиль встроенных блоков. Атрибуты df-<ключ> Drawflow связывает с полем data.<ключ> узла: df-header хранит подпись, которую пользователь пишет прямо на блоке. Не используйте df- атрибуты для своих параметров, если не хотите, чтобы они редактировались на холсте.

Flow-блок (клик по «…» открывает форму):

<div>
  <span class="badge badge-pill bg-primary text-white">0</span>
  <div class="custom-action-form">
    <div class="custom-action-item-bg success-item-bg"></div>
    <div class="title-box editable-headers custom-action-header">
      <i class="fe icon-custom-action"><!-- SVG иконки 18×18, stroke="currentColor" --></i>
      <textarea spellcheck="false" rows="1" class="pl-3" wrap="soft"
                placeholder="Write a description" df-header="">SMS (ACME)</textarea>
      <i class="fe fe-more-horizontal fe-16 pl-4 text-primary custom-action-setting"
         style="cursor: pointer" onclick="configure_custom_request(event)"></i>
    </div>
  </div>
</div>

Segment-блок (двойной клик открывает форму):

<div>
  <span class="badge badge-pill bg-primary text-white">-</span>
  <div class="text-center" ondblclick="configure_custom_action(event)">
    <i class="fe fe-user-plus ml-1 text-primary" style="font-size: 25px; cursor: grab;"
       data-toggle="tooltip" data-placement="top" title="Выгрузка в ACME Audience"></i>
    <div class="box" style="display: none"></div>
  </div>
</div>

Значок badge заполняет платформа (счётчик). SVG-иконки рисуйте с stroke="currentColor" / fill="currentColor", чтобы они работали в светлой и тёмной теме.

5.6 Полный пример script для flow-действия

$(function () {
    const action_name = 'acme-sms';               // = name карточки
    const root = '.custom-modal-' + action_name;  // префикс всех селекторов

    const modal_html = `
    <div class="modal custom-modal-${action_name} fade" tabindex="-1" role="dialog" aria-hidden="true">
      <div class="modal-dialog" role="document">
        <div class="modal-content">
          <div class="modal-header">
            <h5 class="modal-title">SMS (ACME)</h5>
            <button type="button" class="close" data-dismiss="modal" aria-label="Close">
              <span aria-hidden="true">×</span>
            </button>
          </div>
          <div class="modal-body">
            <form class="pb-0">
              <div class="form-group">
                <label class="col-form-label">Телефон</label>
                <input type="text" class="form-control mb-2 phone" placeholder="{{clients.phone}}"/>
                <small class="form-text text-muted">Можно использовать подстановки {{поле.контекста}}</small>
              </div>
              <div class="form-group">
                <label class="col-form-label">Текст</label>
                <textarea class="form-control mb-2 text" rows="4"></textarea>
                <div class="invalid-feedback text-error">Текст обязателен</div>
              </div>
              <div class="form-group">
                <label class="col-form-label">Имя события-результата</label>
                <input type="text" class="form-control mb-2 eventname" placeholder="acme-sms"/>
                <small class="form-text text-muted">
                  В сценарий вернётся &lt;имя&gt;-sent или &lt;имя&gt;-failed
                </small>
              </div>
            </form>
          </div>
          <div class="modal-footer">
            <button type="button" class="btn mb-2 btn-primary save" data-dismiss="modal">Сохранить</button>
            <button type="button" class="btn mb-2 btn-secondary" data-dismiss="modal">Закрыть</button>
          </div>
        </div>
      </div>
    </div>`;

    $('.drawflow-custom-modals').append(modal_html);

    // Заполнение сохранёнными значениями при каждом открытии
    $(root).on('shown.bs.modal', function () {
        const params = get_data_custom_action() || {};
        const data = params.data || {};
        $(root + ' .phone').val(data.phone || '');
        $(root + ' .text').val(data.text || '').removeClass('is-invalid');
        $(root + ' .eventname').val(data.eventname || '');
    });

    $(root + ' .save').on('click', function (e) {
        const text = $(root + ' .text').val();
        if (!text || !text.trim()) {
            e.preventDefault();
            e.stopPropagation();          // не закрывать окно
            $(root + ' .text').addClass('is-invalid');
            return;
        }
        save_data_custom_action([
            { Parameter: 'phone',     Value: $(root + ' .phone').val() },
            { Parameter: 'text',      Value: text },
            { Parameter: 'eventname', Value: $(root + ' .eventname').val() },
        ]);
    });
});

5.7 Полный пример script для segment-действия

$(function () {
    const action_name = 'acme-audience-upload';
    const root = '.custom-modal-' + action_name;

    const modal_html = `
    <div class="modal custom-modal-${action_name} fade" tabindex="-1" role="dialog" aria-hidden="true">
      <div class="modal-dialog" role="document">
        <div class="modal-content">
          <div class="modal-header">
            <h5 class="modal-title">Выгрузка в ACME Audience</h5>
            <button type="button" class="close" data-dismiss="modal" aria-label="Close">
              <span aria-hidden="true">×</span>
            </button>
          </div>
          <div class="modal-body">
            <form class="pb-0">
              <div class="form-group">
                <label class="col-form-label">Аудитория</label>
                <select class="form-control mb-2 audience"></select>
              </div>
              <div class="form-group">
                <label class="col-form-label">Колонка с ID клиента</label>
                <select class="form-control mb-2 idColumn"></select>
                <div class="invalid-feedback">Выберите колонку</div>
              </div>
              <div class="form-group">
                <label class="col-form-label">Строк в секунду</label>
                <input type="number" min="1" class="form-control mb-2 throttle" placeholder="300"/>
              </div>
            </form>
          </div>
          <div class="modal-footer">
            <button type="button" class="btn mb-2 btn-primary save" data-dismiss="modal">Сохранить</button>
            <button type="button" class="btn mb-2 btn-secondary" data-dismiss="modal">Закрыть</button>
          </div>
        </div>
      </div>
    </div>`;

    $('.drawflow-custom-modals').append(modal_html);

    $(document).on('show.bs.modal', root, function () {
        const saved = editor.getSplitNode(node_selected).data || {};

        // Колонки датасета предыдущего шага
        const columns = (check_inputs_data(node_selected) || {}).columns || [];
        const $col = $(root + ' .idColumn').empty().removeClass('is-invalid');
        columns.forEach(c => $col.append(new Option(c, c)));
        if (saved.idColumn) $col.val(saved.idColumn);

        // Справочник из вашего сервиса через webhook
        const $aud = $(root + ' .audience').empty();
        custom_webhooks(action_name, { query: 'list-audiences' }).then(res => {
            const list = typeof res === 'string' ? JSON.parse(res) : res;
            (list.items || []).forEach(a => $aud.append(new Option(a.name, a.id)));
            if (saved.audienceId) $aud.val(saved.audienceId);
        });

        $(root + ' .throttle').val(saved.throttle || '');
    });

    $(root + ' .save').on('click', function (e) {
        const idColumn = $(root + ' .idColumn').val();
        if (!idColumn) {
            e.preventDefault();
            e.stopPropagation();
            $(root + ' .idColumn').addClass('is-invalid');
            return;
        }
        save_data_segment_custom_action({
            isCustomJS: true,
            audienceId: $(root + ' .audience').val(),
            idColumn,
            throttle: $(root + ' .throttle').val(),
        });
    });
});

Для этой формы в карточке нужны data.isCustomJS: true, script и webhooks; parameters можно не передавать.

6. Flow-действие

6.1 Задача

Подпишитесь: своя durable-очередь (см. 3.2), привязанная к ex-custom-actions.

{
  "id": "79161234567",
  "event": "purchase",
  "flow_id": "9b2f...",
  "flow_name": "Welcome-цепочка",
  "context": {
    "clients": { "client_id": "79161234567", "name": "Иван" },
    "amount": 1500
  },
  "action": {
    "id": 14,
    "name": "acme-sms",
    "class": "stateless-action",
    "data": {
      "phone": "{{clients.client_id}}",
      "text": "{{clients.name}}, спасибо за покупку!",
      "eventname": "acme-sms-sent"
    }
  }
}

В очередь приходят задачи всех flow-действий установки. Сообщение с чужим action.name — штатная ситуация: подтвердите (ack) и пропустите, без nack и без ошибок в логе.

6.2 Шаблоны в параметрах

Пользователи пишут в параметрах подстановки {{путь}} — путь внутри context: {{clients.client_id}}, {{amount}}. Разрешать их — задача вашего действия.

• Если параметр целиком состоит из подстановки, сохраняйте тип значения (число остаётся числом).
• Если подставляете в строку, а потом собираете из неё JSON — экранируйте значения как JSON. Кавычка или перевод строки в имени клиента иначе сломают разбор, и сообщение не уйдёт.
• Не используйте подставляемое значение как шаблон замены (в JavaScript String.replace интерпретирует $&, $' в строке замены — используйте функцию-заменитель).
• Отсутствующий путь — явная ошибка с понятным текстом, а не пустая строка, которая уйдёт клиенту.

6.3 Возврат в сценарий: событие в ex-triggers

Задача flow-действия не предполагает ответа. Чтобы сценарий узнал результат и продолжился, действие публикует событие в обменник ex-triggers:

• тип обменника x-consistent-hash, durable;
• ключ маршрутизации — **id** участника строкой (consistent hash по нему гарантирует, что все события одного участника обрабатываются по порядку);
• тело:

{
  "id": "79161234567",
  "event_id": "79161234567",
  "event": "acme-sms-sent",
  "context": { "...": "исходный context, дополненный результатом" },
  "flowId": "9b2f..."
}

Правила именования событий:

• два разных события на успех и на неудачу: acme-sms-sent / acme-sms-failed. Сценарий должен уметь отличить одно от другого;
• уникальный префикс вашего действия. Если несколько ваших действий шлют одно и то же push-sent, в сценарии их не различить;
• дайте пользователю переопределить имя события в форме (параметр eventname), но имейте дефолт.

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

6.4 Аналитика (необязательно)

Действующие реализации после обработки пишут запись в журнал выполнения узлов: сообщение {"table": "log_table", "body": "<JSON строкой>"} в очередь FLOWS.analytics{q}. В body: id, flow_name, flow_id, previous_node и node (= action.id строкой), message (текст ошибки или пусто), processed_successfully (1 или -1), action, action_class, action_data и context (JSON строкой), event, timestamp (мс), date, time, datetime.

Если пишете — processed_successfully должен отражать реальный результат, а не просто «обработчик не бросил исключение».

6.5 Порядок обработки

1. Разобрать JSON. Не разбирается — ack, лог, выход.
2. Чужое action.name — ack, выход.
3. Провалидировать и разрешить параметры.
4. Выполнить работу во внешней системе с таймаутом.
5. Опубликовать событие успеха или неудачи, дождаться подтверждения брокера.
6. Ack задачи.

Повторов на уровне брокера для flow-задач нет: если хотите повторять вызов внешней системы, делайте это внутри шага 4 с ограниченным бюджетом времени, а по исчерпании публикуйте событие неудачи.

7. Segment-действие

7.1 Задача

Подпишитесь: своя durable-очередь, привязанная к ex-segment-custom-actions.{svc}.

{
  "name": "acme-audience-upload",
  "data": {
    "audienceName": "Купившие в октябре",
    "idColumn": "client_id",
    "throttle": "500"
  },
  "meta": {
    "segmentId": "a1b2c3",
    "startupId": "1760521234567",
    "trigger": "start",
    "nodeId": "7",
    "nodeClass": "acme-audience-upload",
    "segment_name": "Купившие в октябре",
    "initiator": "ivanov",
    "inputTable": "segments.s_a1b2c3_1760521234567_6"
  }
}

7.2 Тестовый запуск

Если meta.trigger == "test", ничего не делайте во внешней системе и сразу отправьте статус success с counter: 0. Платформа так проверяет, что действие доступно и правильно зарегистрировано.

Тестовый запуск не должен открывать соединений с БД, которые потом не закроются, и не должен оставлять следов в журналах выполнения.

7.3 Чтение входной таблицы

Таблица может содержать миллионы строк. Не загружайте её в память целиком.

• Потоком (курсор/stream): один запрос, строки читаются по мере обработки. Просто, но держит открытую транзакцию на всё время обработки — на часах работы это мешает очистке (vacuum) в PostgreSQL.
• Порциями: короткие независимые запросы. В PostgreSQL 14+ эффективнее всего диапазоны ctid (WHERE ctid >= '(N,0)'::tid AND ctid < '(M,0)'::tid); универсальный вариант — keyset по уникальной колонке (WHERE id > :last ORDER BY id LIMIT :n).
• Не используйте **LIMIT … OFFSET …** без **ORDER BY**: порядок строк между запросами не гарантирован, часть строк продублируется, часть потеряется. С ORDER BY же OFFSET на больших таблицах деградирует квадратично.

Прочее:

• В Oracle имя таблицы приводите к верхнему регистру.
• inputTable подставляется в SQL как идентификатор — проверяйте его регулярным выражением вида ^[A-Za-z0-9_]+(\.[A-Za-z0-9_]+)?$ и/или экранируйте как идентификатор.
• Читайте только нужные колонки; ожидаемые колонки проверьте до начала обработки и падайте с понятной ошибкой («в датасете нет колонки client_id»), а не на тысячной строке.
• Пропуск строк (пустой ID, невалидное значение) — считайте и показывайте в итоговом статусе, а не отбрасывайте молча.
• Ставьте таймаут выполнения запроса на стороне БД (statement_timeout) — отмена запроса со стороны клиента срабатывает не всегда.
• Закрывайте соединение/пул на всех путях выхода, включая тестовый запуск, ошибку и остановку.

7.4 Троттлинг и параллелизм

Внешние системы имеют лимиты. Ограничивайте скорость отправки параметром действия (throttle, строк в секунду) с разумным дефолтом из конфигурации.

Ограничивайте две величины:

• Скорость — сколько отправок начинается в секунду (token bucket / окно в 1 с).
• Число одновременных запросов в полёте. Если отправки «выстрелил и забыл», а внешняя система тормозит, при ограничении только скорости число незавершённых запросов растёт без предела — до исчерпания памяти и сокетов. Ставьте предел (например, 2 × throttle) и приостанавливайте чтение, пока он превышен.

В конце обработки дождитесь завершения всех запросов в полёте (с таймаутом) — и только потом считайте итог и отправляйте статус.

Если throttle в вашем действии означает не «строк в секунду», а, например, «размер пачки», назовите параметр иначе (batchSize): одинаковое имя с разным смыслом путает пользователей.

7.5 Статус выполнения

По завершении отправьте в очередь FLOWS.custom-actions-statuses{q} исходное сообщение задачи, дополнив meta:

{
  "name": "acme-audience-upload",
  "data": { "...": "как в задаче" },
  "meta": {
    "...": "все поля из задачи",
    "status": "success",
    "counter": 128034,
    "statement": "SELECT client_id FROM segments.s_a1b2c3_1760521234567_6",
    "error": null
  }
}

Правила:

• Ровно один итоговый статус на запуск. Ни одного — сегмент навсегда «в обработке». Два противоречивых — непредсказуемое состояние.
• Без промежуточных статусов: прогресс платформа не принимает, отправляйте его в логи и метрики.
• **success** означает, что работа сделана, а не что строки прочитаны. Если 30 % отправок во внешнюю систему упали — это не успех. Действующие реализации расходятся в том, где проводить черту; разумный дефолт — error при любой неудаче с подробностями в тексте: "обработано 128034, ошибок 4102 (HTTP 429: 3990, таймаут: 112)". Если для вашего случая частичный результат приемлем, сделайте порог настраиваемым и всегда пишите число ошибок в текст.
• Не добавляйте в возвращаемое сообщение служебные поля своей реализации — оно уходит в платформу как есть.
• Отправляйте с persistent: true и подтверждением брокера; при неудаче — повторить с паузой. Статус должен дойти даже при остановке сервиса (не отменяйте его отправку вместе с остальной работой при SIGTERM).

7.6 Порядок обработки

• Разобрать JSON; не разбирается — ack, лог, выход.
• Чужое name — ack, выход.
• Зарегистрировать запуск в реестре активных и подписаться на stop (до любых await — см. раздел 8). Проверить, не пришёл ли stop заранее.
• Тестовый запуск — статус success, выход.
• Провалидировать параметры; ошибка — статус error с понятным текстом.
• Проверить колонки входной таблицы.
• Читать, отправлять с троттлингом, проверяя флаг остановки на каждой строке/порции.
• Дождаться запросов в полёте.
• Отправить итоговый статус.
• На всех путях: снять запуск с реестра, отписаться от сигналов, закрыть соединения.

Когда подтверждать (ack) саму задачу — см. раздел 9.

7.7 Сегмент → сценарий: запуск участников в Flow

Segment-действие может отправлять клиентов сегмента в сценарий — публиковать по событию на каждую строку в ex-triggers (формат как в 6.3):

Особенности:

• троттлинг обязателен — без него миллион событий одномоментно ляжет на модуль Flow;
• строки без идентификатора пропускайте и считайте в статусе;
• если публикация события не прошла (брокер недоступен), прекращайте обработку со статусом error: продолжать «в пустоту» бессмысленно, а счётчик покажет, сколько ушло;
• counter в статусе — число опубликованных событий.

8. Контрольные сигналы

Пользователь может остановить обработку сегмента. Сигнал приходит в fanout-обменник ex-segments-control.{svc}.

8.1 Подписка

Каждый инстанс создаёт собственную очередь и привязывает её к обменнику:

• имя с уникальным суффиксом: acme-actions-control.{svc}.{random} (или пустое имя — брокер сгенерирует сам);
• exclusive: true, autoDelete: true, x-message-ttl: 300000;
• не durable и не quorum: очередь нужна, пока жив процесс.

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

Читайте сигналы через отдельное соединение, не через то, что занято задачами. Обработка сегмента может идти часами, и сигнал не должен ждать в одной очереди с ней.

8.2 Формат

{
  "signal": "stop",
  "segmentId": "a1b2c3",
  "startupId": "1760521234567",
  "reason": "Остановлено пользователем",
  "timestamp": 1760521299000
}

signal: stop, pause, resume. На практике платформа и все действующие реализации используют только stop; pause и resume логируйте как неподдерживаемые и подтверждайте. timestamp встречается и числом, и строкой — не полагайтесь на тип.

8.3 Обработка

• Сигналы всегда подтверждайте (ack), включая неразборчивые. nack с повтором зациклит сигнал.
• Найдите у себя запуск по segmentId + startupId. В одном запуске может работать несколько ваших узлов (nodeId) — остановите все.
• Нет такого запуска — не выбрасывайте сигнал сразу (см. 8.4).
• Есть — выставьте флаг остановки / отмените контекст. Обработка на ближайшей строке прекращает чтение, закрывает поток и соединение, дожидается (или отменяет) запросов в полёте и отправляет статус error с текстом вида "Остановлено пользователем: <reason>" и counter = обработано строк.

Статуса «отменено» нет: остановка — это error с причиной в тексте. Уже отправленное во внешнюю систему не откатывается — опишите это пользователям.

8.4 Stop раньше задачи

Задача и сигнал идут разными очередями, порядок между ними не гарантирован. Пользователь может нажать «Стоп» сразу после запуска, пока задача ещё в очереди или пока действие открывает соединение с БД. Если подписываться на stop только после подключения к БД, такой stop теряется, и сегмент отрабатывает целиком.

Как избежать:

• регистрируйте запуск и подписывайтесь на сигнал синхронно, до первого **await**;
• на сигнал по неизвестному запуску сохраняйте «надгробие» segmentId:startupId с TTL несколько минут. Пришедшая позже задача проверяет его первым делом и сразу завершается статусом error без какой-либо работы;
• если инстансов несколько, храните надгробие в общем хранилище (например, Redis с TTL) — сигнал мог получить инстанс, которому задача ещё не досталась.

8.5 Утечки

Подписки на сигналы (обработчики событий, записи в реестре активных запусков) удаляйте на всех путях выхода: успех, ошибка, остановка, исключение. Иначе реестр растёт, а сигнал по давно завершённому запуску «находит» мёртвую запись.

9. Надёжность

9.1 prefetch и таймауты

Ставьте prefetch(1) (или небольшое число) на очередях задач: задачи распределяются между инстансами равномерно, а не скапливаются у одного.

Обратная сторона: пока задача не подтверждена, инстанс не берёт следующую. Значит:

• у каждого внешнего вызова должен быть таймаут. Один зависший HTTP/SOAP-вызов без таймаута останавливает обработку всей очереди;
• у segment-действия должен быть общий таймаут выполнения (например, 15–60 минут, настраиваемый), по истечении которого — статус error с текстом "превышено время выполнения N мин";
• выполнять пользовательский код (eval и т. п.) в основном процессе нельзя — бесконечный цикл в нём остановит всё.

9.2 Когда подтверждать задачу

Flow-задача — после публикации события в ex-triggers (с подтверждением брокера). Падение до ack приведёт к повторной доставке — действие должно это пережить (см. 9.3).

Segment-задача — две стратегии, выбирайте по природе действия:

Журнал активных запусков (для ack сразу): при старте обработки запишите в общее хранилище segmentId:startupId → {задача, id инстанса}, при завершении удалите. При старте сервиса пройдите журнал: записи, принадлежащие этому инстансу или инстансам, которых больше нет, закройте статусом error с текстом «инстанс, выполнявший обработку, перезапущен» и удалите. Не закрывайте записи живых соседних инстансов — иначе рестарт одного пода помечает ошибкой сегменты, которые прямо сейчас успешно идут на другом.

Счётчик попыток (для ack после статуса): в classic-очередях есть только флаг redelivered, считайте попытки сами по детерминированному ключу name|segmentId|startupId (в quorum — заголовок x-delivery-count). После N попыток — статус error и nack без повтора, чтобы задача-«отравитель» не крутилась вечно.

9.3 Идемпотентность

Любая задача может прийти повторно: переподключение до ack, рестарт, сетевой сбой.

• Операции записи делайте как UPSERT по естественному ключу (клиент + сущность).
• Для отправок во внешние системы передавайте детерминированный ключ идемпотентности, если система его поддерживает: для flow — производный от flow_id + action.id + id + отметки события; для segment — от startupId + nodeId + ID строки.
• Если система не поддерживает идемпотентность, выбирайте стратегию ack «сразу» и честно отражайте в статусе потерю запуска при падении.

9.4 Гарантия ровно одного статуса

Оберните обработку segment-задачи так, чтобы итоговый статус гарантированно уходил ровно один раз:

• исключение и паника внутри действия → error с текстом;
• остановка → error с причиной остановки;
• таймаут → error с указанием таймаута;
• действие вернулось без статуса → success с его счётчиком;
• если статус уже отправлен самим действием — второй не отправлять.

Если отправить статус не удалось (брокер недоступен), не отправляйте вместо него противоположный. При стратегии «ack после статуса» верните задачу в очередь с паузой; при стратегии «ack сразу» — повторяйте отправку с ограниченным бюджетом и оставьте запись в журнале активных запусков, чтобы её закрыло восстановление.

9.5 Публикация: подтверждения и mandatory

• Используйте publisher confirms и ждите подтверждения на события, статусы и регистрацию.
• Ставьте mandatory: true. Тонкость протокола: сообщение, которое никуда не смаршрутизировалось, всё равно получает положительное подтверждение, а basic.return приходит до него. Проверяйте возвраты после подтверждения и считайте возврат ошибкой.
• Не публикуйте через внутренние поля обёрток над каналом (_channel и т. п.): после переподключения их может не быть.

9.6 Переподключение

• Переподключайтесь с экспоненциальной паузой (1 с, 2 с, 4 с … до 30 с).
• После переподключения заново объявите очереди, обменники, привязки и prefetch.
• Дождитесь завершения обработчика со старого канала, прежде чем начинать читать на новом: его ack на мёртвом канале не пройдёт, задача придёт повторно и выполнится дважды.
• При OAuth получайте свежий токен при каждом переподключении; учитывайте срок жизни токена (кешируйте до ~90 % TTL).

9.7 Масштабирование

Несколько инстансов — нормальный режим:

• очереди задач — одно имя на все инстансы (раздел 3.2);
• очередь сигналов — своя у каждого (раздел 8.1);
• всё общее состояние (журнал запусков, надгробия stop, счётчики попыток при необходимости) — во внешнем хранилище, ключи с учётом инстанса там, где это важно (раздел 9.2);
• не полагайтесь на то, что stop придёт тому же инстансу, что и задача.

10. Эксплуатация

10.1 Логи

• Структурированные (JSON), с полями корреляции: для flow — action, flow_id, event_id (id); для segment — action, segment_id, startup_id, node_id, input_table, trigger.
• Прогресс сегмента — раз в порцию/раз в N секунд на уровне info, не построчно.
• Не логируйте персональные данные построчно (телефоны, ИИН, email, тексты сообщений). Миллион строк сегмента = миллион записей с ПДн в системе логов.
• Чужие задачи из fanout-очереди — на уровне debug и без метрик: это штатный трафик.

10.2 Метрики

Минимальный набор (Prometheus):

Эндпоинт метрик слушайте на 0.0.0.0, иначе скрейпер из соседнего контейнера его не увидит.

10.3 Healthcheck

/ready должен проверять реальную готовность: соединение с RabbitMQ поднято, консьюмеры активны, БД сегментов доступна. Заглушка, всегда отвечающая 200, скрывает падения.

10.4 Остановка (SIGTERM)

• Перестать брать новые задачи (cancel consumer).
• Дать текущим flow-задачам завершиться (секунды).
• Segment-обработку прервать: отправить статус error («сервис остановлен») либо, при стратегии «ack после статуса», не подтверждать задачу — она вернётся в очередь.
• Отправку итоговых статусов не прерывать вместе с остальным.
• Закрыть каналы, соединения, пулы БД.

Также: перехватывайте необработанные исключения/rejection в процессе — в Node.js 15+ необработанный rejection завершает процесс.

10.5 Конфигурация и секреты

• Все адреса, ключи, лимиты — через переменные окружения или секреты оркестратора. Никаких ключей в коде, в примерах .env, в script формы (он виден в браузере любому пользователю бэкофиса).
• Флаги сравнивайте строго (=== 'true'): непустая строка "false" — тоже истина при проверке на непустоту.
• Ведите полный пример конфигурации (.env.example) со всеми переменными и дефолтами.
• Если сервис создаёт свои таблицы — только миграциями, которые никогда не делают DROP данных при повторном применении.

11. Изменение и версионирование действия

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

Логируйте при старте версию сервиса и список зарегистрированных действий — это первое, что спросят при разборе инцидента.

12. Локальная разработка и тестирование

12.1 Стенд

RabbitMQ с консолью управления и нужным плагином:

docker run -d --name rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management
docker exec rabbit rabbitmq-plugins enable rabbitmq_consistent_hash_exchange
# консоль: http://localhost:15672, guest / guest

PostgreSQL с тестовой «входной таблицей» сегмента:

docker run -d --name pg -p 5432:5432 -e POSTGRES_PASSWORD=pg postgres:16
docker exec -i pg psql -U postgres <<'SQL'
CREATE SCHEMA segments;
CREATE TABLE segments.s_test_1 AS
  SELECT 'client_' || g AS client_id, (random() * 1000)::int AS amount
  FROM generate_series(1, 100000) g;
SQL

Чтобы видеть, что ваш сервис публикует, создайте в консоли RabbitMQ (вкладка Queues) отладочные очереди и привяжите их:

12.2 Ручная отправка задач

В консоли: Exchanges → нужный обменник → Publish message, тело — JSON. Или rabbitmqadmin:

# flow-задача
rabbitmqadmin publish exchange=ex-custom-actions routing_key="" \
  payload='{"id":"client_1","event":"purchase","flow_id":"f1","flow_name":"test",
            "context":{"clients":{"phone":"79160000000","name":"Иван"}},
            "action":{"id":1,"name":"acme-sms","class":"stateless-action",
                      "data":{"phone":"{{clients.phone}}","text":"Привет, {{clients.name}}","eventname":""}}}'

# segment-задача
rabbitmqadmin publish exchange=ex-segment-custom-actions.segments-service routing_key="" \
  payload='{"name":"acme-audience-upload","data":{"idColumn":"client_id","throttle":"1000"},
            "meta":{"segmentId":"seg1","startupId":"1","trigger":"start","nodeId":"3",
                    "nodeClass":"acme-audience-upload","segment_name":"test","initiator":"dev",
                    "inputTable":"segments.s_test_1"}}'

# stop
rabbitmqadmin publish exchange=ex-segments-control.segments-service routing_key="" \
  payload='{"signal":"stop","segmentId":"seg1","startupId":"1","reason":"test","timestamp":1760521299000}'

Обменники объявляет ваш сервис при старте; если публикуете до его запуска — объявите их сами в консоли с теми же типами.

12.3 Обязательные сценарии проверки

Flow

Segment

Форма

12.4 Проверка на установке

• Запустите сервис, проверьте в логах, что карточки опубликованы.
• Откройте редактор сценариев/сегментов — блок есть в палитре с правильной иконкой и подписью.
• Соберите минимальную схему, настройте блок, сохраните.
• Сегмент: сначала тестовый запуск (S1), затем боевой на маленькой выборке.
• Сценарий: отправьте событие одного тестового клиента и убедитесь, что сценарий продолжился по событию-результату.

13. Что согласовать с администратором установки

Получить от администратора:

Сообщить администратору:

14. FAQ

На каком языке писать? На любом, где есть клиент AMQP 0-9-1 и драйвер БД сегментов. Действующие реализации — на Node.js и Go.

Можно ли в одном сервисе несколько действий, и flow, и segment? Да. Одна очередь flow-задач и одна segment-задач на сервис; диспетчеризация по action.name / name внутри.

Платформа как-то подтверждает регистрацию? Нет, ответа нет. Признак успеха — подтверждение брокера на публикацию и блок в палитре редактора.

Блок есть в палитре, но задачи не приходят. Проверьте по порядку: очередь привязана к нужному обменнику (ex-custom-actions / ex-segment-custom-actions.{svc} с правильным {svc}); фильтруете по тому же имени, что в карточке; режим quorum/classic совпадает; в консоли RabbitMQ у вашей очереди есть консьюмер и сообщения не копятся.

Ко мне приходят задачи чужих действий. Это нормально? Да, обменники fanout. Ack и пропустить.

Сегмент «висит» в обработке. Статус не дошёл: не отправлен на каком-то пути выхода, отправлен в очередь без .quorum (или наоборот), процесс упал после ack без восстановления. См. 7.5 и 9.2.

Можно ли сообщать прогресс выполнения сегмента? Нет, платформа принимает только итоговый статус. Прогресс — в логи и метрики.

Как показать пользователю, что часть строк не обработана? Через текст meta.error при статусе error, с числами по видам ошибок.

Работают ли pause / resume? Сигналы предусмотрены форматом, но платформой не используются. Логируйте и игнорируйте; поддерживать их сейчас не нужно.

Можно ли менять **context** во flow-действии? Добавлять — да, в своё поле с префиксом действия. Удалять или перезаписывать чужие поля не нужно: они могут понадобиться дальше по сценарию.

Числа в **data** приходят строками. Да, значения из формы часто строковые. Разбирайте оба варианта.

Можно ли обрабатывать несколько сегментов одновременно в одном инстансе? Можно (prefetch > 1 или ack сразу и обработка в фоне), но тогда сами ограничивайте общий параллелизм и нагрузку на БД и внешние системы. Проще масштабироваться инстансами.

Нужно ли поддерживать Oracle? Только если установка на Oracle — уточните у администратора. Отличия: имя таблицы в верхнем регистре, другой драйвер, нет ctid.

Как удалить действие из палитры? Через администратора платформы. Перестать регистрировать недостаточно.

Можно ли из flow-действия запустить сегмент? Это делает встроенное действие платформы; своё писать не нужно.

Где брать справочники для формы? Свои — через webhooks; справочники платформы — запросом из скрипта к API бэкофиса.

15. Частые ошибки

Все пункты взяты из реальных реализаций.

16. Чек-лист перед запуском

Топология

• Очереди задач с уникальными именами сервиса, одинаковыми на всех инстансах, durable
• Очередь сигналов своя на инстанс: exclusive, autoDelete, TTL, отдельное соединение
• Режим quorum/classic совпадает с установкой; суффикс .quorum только у очередей
• Все очереди/обменники, в которые публикуете, объявляются на каждом канале

Регистрация

• name уникален и с префиксом компании; та же константа используется для фильтрации
• Все параметры перечислены в data с дефолтами
• Регистрация при каждом старте, с подтверждением брокера
• Форма: селекторы с префиксом .custom-modal-{name}, правильная функция сохранения, заполнение сохранённых значений при открытии, валидация

Flow

• Фильтрация по action.name, чужие — ack
• Шаблоны {{...}} разрешаются с экранированием
• Событие успеха и неудачи в ex-triggers, ключ = id, flowId передан
• Таймаут внешнего вызова; ack после публикации события

Segment

• Фильтрация по name, чужие — ack
• Тестовый запуск: success, counter: 0, без побочных эффектов и утечек
• Чтение потоком или порциями без OFFSET; имя таблицы проверяется
• Троттлинг скорости и предел запросов в полёте; ожидание хвоста перед статусом
• Ровно один статус на всех путях; error — текстом; success отражает реальную доставку
• Стратегия ack выбрана осознанно; журнал запусков или идемпотентность + счётчик попыток
• Общий таймаут выполнения
• Stop: подписка до первого await, надгробия, остановка всех узлов запуска, очистка подписок

Эксплуатация

• Структурированные логи с корреляцией, без построчных ПДн
• Метрики и честный healthcheck
• Graceful shutdown; перехват необработанных ошибок
• Секреты вне кода; полный .env.example

17. Скелеты реализации

Минимальные примеры на «голом» клиенте AMQP, без фреймворков. Это отправная точка, а не готовый сервис: переподключение, метрики, журнал запусков и прочее из разделов 9–10 сюда не включены.

17.1 Node.js: общее

// common.js
const amqp = require('amqplib');
const crypto = require('crypto');

const QUORUM = process.env.USE_QUORUM_QUEUES === 'true';
const SVC = (process.env.SEGMENTS_SERVICE_NAME || 'segments-service').trim().split(' ').join('_');

const q = (name) => (QUORUM ? `${name}.quorum` : name);
const queueOptions = () => (QUORUM
    ? { durable: true, arguments: { 'x-queue-type': 'quorum' } }
    : { durable: true });

const NAMES = {
    FLOW_REGISTER: q('FLOWS.flows-custom-actions-register'),
    SEGMENT_REGISTER: q('FLOWS.segments-custom-actions-register'),
    STATUS: q('FLOWS.custom-actions-statuses'),
    FLOW_TASKS_EXCHANGE: 'ex-custom-actions',
    SEGMENT_TASKS_EXCHANGE: `ex-segment-custom-actions.${SVC}`,
    CONTROL_EXCHANGE: `ex-segments-control.${SVC}`,
    TRIGGERS_EXCHANGE: 'ex-triggers',
    // Свои имена: уникальны для сервиса, одинаковы для всех инстансов
    MY_FLOW_QUEUE: q(process.env.QUEUE_NAME || 'acme-actions'),
    MY_SEGMENT_QUEUE: q(`acme-segment-actions.${SVC}`),
    MY_CONTROL_QUEUE: `acme-actions-control.${SVC}.${crypto.randomBytes(4).toString('hex')}`,
};

async function createPublisher(conn) {
    const ch = await conn.createConfirmChannel();
    await ch.assertQueue(NAMES.FLOW_REGISTER, queueOptions());
    await ch.assertQueue(NAMES.SEGMENT_REGISTER, queueOptions());
    await ch.assertQueue(NAMES.STATUS, queueOptions());
    await ch.assertExchange(NAMES.TRIGGERS_EXCHANGE, 'x-consistent-hash', { durable: true });

    let returned = false;
    ch.on('return', () => { returned = true; });

    // Публикация с подтверждением брокера; возврат (unroutable) считаем ошибкой
    async function publish(exchange, routingKey, body) {
        returned = false;
        ch.publish(exchange, routingKey, Buffer.from(JSON.stringify(body)),
            { persistent: true, mandatory: true, contentType: 'application/json' });
        await ch.waitForConfirms();
        if (returned) throw new Error(`unroutable: ${exchange}/${routingKey}`);
    }

    return {
        registerFlow: (card) => publish('', NAMES.FLOW_REGISTER, card),
        registerSegment: (card) => publish('', NAMES.SEGMENT_REGISTER, card),
        status: (task, status, { counter = 0, statement = null, error = null } = {}) =>
            publish('', NAMES.STATUS, { ...task, meta: { ...task.meta, status, counter, statement, error } }),
        event: (id, event, context, flowId) =>
            publish(NAMES.TRIGGERS_EXCHANGE, String(id),
                { id, event_id: id, event, context, ...(flowId ? { flowId } : {}) }),
    };
}

module.exports = { amqp, NAMES, queueOptions, createPublisher };

waitForConfirms ждёт все неподтверждённые публикации канала; при параллельной публикации из многих обработчиков используйте колбэк publish(..., cb) на каждое сообщение.

17.2 Node.js: flow-действие

// flow.js
const { amqp, NAMES, queueOptions, createPublisher } = require('./common');

const CARD = {
    name: 'acme-sms',
    title: 'SMS (ACME)',
    class: 'stateless-action',
    icon: 'fe fe-message-circle',
    html: '<div>...</div>',
    inputs: 1,
    outputs: 0,
    data: { isCustom: true, isCustomJS: true, isCustomHTML: true, returnable: false,
            phone: '', text: '', eventname: '' },
    paramsDescription: '{}',
    script: require('fs').readFileSync(__dirname + '/ui/acme-sms.js', 'utf8'),
    webhooks: null,
};

// {{a.b}} из context; значение целиком — с сохранением типа
function resolve(value, context) {
    if (typeof value !== 'string') return value;
    const get = (path) => {
        const v = path.trim().split('.').reduce((o, k) => (o == null ? undefined : o[k]), context);
        if (v === undefined) throw new Error(`в контексте нет поля ${path.trim()}`);
        return v;
    };
    const whole = value.match(/^\{\{(.+?)\}\}$/);
    if (whole) return get(whole[1]);
    return value.replace(/\{\{(.+?)\}\}/g, (_, p) => String(get(p)));
}

async function sendSms(phone, text) {
    const res = await fetch(process.env.ACME_SMS_URL, {
        method: 'POST',
        headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.ACME_TOKEN}` },
        body: JSON.stringify({ phone, text }),   // JSON.stringify сам экранирует значения
        signal: AbortSignal.timeout(5000),
    });
    if (!res.ok) throw new Error(`ACME API ответил ${res.status}`);
    return res.json();
}

(async () => {
    const conn = await amqp.connect(process.env.AMQP_URL);
    const pub = await createPublisher(conn);
    await pub.registerFlow(CARD);

    const ch = await conn.createChannel();
    await ch.assertExchange(NAMES.FLOW_TASKS_EXCHANGE, 'fanout', { durable: true });
    await ch.assertQueue(NAMES.MY_FLOW_QUEUE, queueOptions());
    await ch.bindQueue(NAMES.MY_FLOW_QUEUE, NAMES.FLOW_TASKS_EXCHANGE, '');
    await ch.prefetch(1);

    await ch.consume(NAMES.MY_FLOW_QUEUE, async (msg) => {
        let task;
        try { task = JSON.parse(msg.content.toString()); } catch { return ch.ack(msg); }
        if (task?.action?.name !== CARD.name) return ch.ack(msg);   // чужая задача — штатно

        const data = task.action.data || {};
        const context = { ...(task.context || {}) };
        const base = data.eventname || 'acme-sms';
        let event;
        try {
            const result = await sendSms(resolve(data.phone, context), resolve(data.text, context));
            context.acme_sms = { message_id: result.id };
            event = `${base}-sent`;
        } catch (e) {
            context.acme_sms = { error: e.message };
            event = `${base}-failed`;
        }

        try {
            await pub.event(task.id, event, context, task.flow_id);
            ch.ack(msg);
        } catch (e) {
            console.error('не удалось вернуть событие в сценарий', e);
            ch.nack(msg, false, true);   // вернуть задачу; повтор должен быть идемпотентным
        }
    });
})();

17.3 Node.js: segment-действие с сигналами

// segment.js
const { Pool } = require('pg');
const Cursor = require('pg-cursor');
const { amqp, NAMES, queueOptions, createPublisher } = require('./common');

const CARD = {
    name: 'acme-audience-upload',
    title: 'Выгрузка в ACME Audience',
    icon: 'fe fe-user-plus',
    data: { audienceName: '', idColumn: '', throttle: '' },
    parameters: [
        { name: 'audienceName', label: 'Название аудитории', type: 'input' },
        { name: 'idColumn', label: 'Колонка с ID клиента', type: 'inputSingleColumn' },
        { name: 'throttle', label: 'Строк в секунду', type: 'number' },
    ],
};

const TABLE_RE = /^[A-Za-z0-9_]+(\.[A-Za-z0-9_]+)?$/;
const COLUMN_RE = /^[A-Za-z0-9_]+$/;
const BATCH = 1000;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// Реестр запусков и «надгробия» для stop, пришедшего раньше задачи.
// При нескольких инстансах надгробия держите в Redis с TTL.
const running = new Map();     // `${segmentId}:${startupId}` -> { stopReason }
const tombstones = new Map();  // key -> { reason, expires }

function onStop(signal) {
    const key = `${signal.segmentId}:${signal.startupId}`;
    const run = running.get(key);
    if (run) run.stopReason = signal.reason || 'Остановлено пользователем';
    else tombstones.set(key, { reason: signal.reason, expires: Date.now() + 5 * 60_000 });
}

async function processSegment(task, pub, pool) {
    const { segmentId, startupId, inputTable, trigger } = task.meta;
    const key = `${segmentId}:${startupId}`;

    // Регистрация запуска — синхронно, до первого await
    const run = { stopReason: null };
    const tomb = tombstones.get(key);
    if (tomb && tomb.expires > Date.now()) run.stopReason = tomb.reason || 'Остановлено пользователем';
    tombstones.delete(key);
    running.set(key, run);

    let counter = 0, failed = 0, statement = null;
    let client;
    try {
        if (run.stopReason) return pub.status(task, 'error', { error: `Остановлено пользователем: ${run.stopReason}` });
        if (trigger === 'test') return pub.status(task, 'success', { counter: 0 });

        const idColumn = String(task.data.idColumn || '').replace(/[{}]/g, '');
        if (!TABLE_RE.test(inputTable)) throw new Error(`недопустимое имя таблицы: ${inputTable}`);
        if (!COLUMN_RE.test(idColumn)) throw new Error('не выбрана колонка с ID клиента');
        const throttle = parseInt(task.data.throttle, 10) || Number(process.env.DEFAULT_THROTTLE) || 300;

        statement = `SELECT "${idColumn}" AS id FROM ${inputTable}`;
        client = await pool.connect();
        const cursor = client.query(new Cursor(statement));

        let windowStart = Date.now(), inWindow = 0;
        while (!run.stopReason) {
            const rows = await cursor.read(BATCH);
            if (rows.length === 0) break;
            const ids = rows.map((r) => r.id).filter((id) => id != null);
            failed += rows.length - ids.length;

            // Здесь отправка порции во внешнюю систему, с таймаутом и учётом ошибок
            // failed += await uploadToAudience(task.data.audienceName, ids);
            counter += rows.length;

            inWindow += rows.length;
            if (inWindow >= throttle) {
                const left = 1000 - (Date.now() - windowStart);
                if (left > 0) await sleep(left);
                windowStart = Date.now(); inWindow = 0;
            }
        }
        await cursor.close();

        if (run.stopReason) {
            return pub.status(task, 'error', { counter, statement, error: `Остановлено пользователем: ${run.stopReason}` });
        }
        if (failed > 0) {
            return pub.status(task, 'error', { counter, statement, error: `обработано ${counter}, ошибок ${failed}` });
        }
        return pub.status(task, 'success', { counter, statement });
    } catch (e) {
        return pub.status(task, 'error', { counter, statement, error: e.message });
    } finally {
        running.delete(key);
        if (client) client.release();
    }
}

(async () => {
    const pool = new Pool({ connectionString: process.env.SEGMENTS_DB_URL, statement_timeout: 15 * 60_000 });
    const conn = await amqp.connect(process.env.AMQP_URL);
    const pub = await createPublisher(conn);
    await pub.registerSegment(CARD);

    // Сигналы — отдельное соединение и своя очередь на инстанс
    const controlConn = await amqp.connect(process.env.AMQP_URL);
    const ctl = await controlConn.createChannel();
    await ctl.assertExchange(NAMES.CONTROL_EXCHANGE, 'fanout', { durable: true });
    await ctl.assertQueue(NAMES.MY_CONTROL_QUEUE,
        { exclusive: true, autoDelete: true, arguments: { 'x-message-ttl': 300000 } });
    await ctl.bindQueue(NAMES.MY_CONTROL_QUEUE, NAMES.CONTROL_EXCHANGE, '');
    await ctl.consume(NAMES.MY_CONTROL_QUEUE, (msg) => {
        try {
            const s = JSON.parse(msg.content.toString());
            if (s.signal === 'stop') onStop(s);
            else console.info('неподдерживаемый сигнал', s.signal);
        } catch (e) { console.warn('неразборчивый сигнал', e.message); }
        ctl.ack(msg);   // всегда ack
    });

    const ch = await conn.createChannel();
    await ch.assertExchange(NAMES.SEGMENT_TASKS_EXCHANGE, 'fanout', { durable: true });
    await ch.assertQueue(NAMES.MY_SEGMENT_QUEUE, queueOptions());
    await ch.bindQueue(NAMES.MY_SEGMENT_QUEUE, NAMES.SEGMENT_TASKS_EXCHANGE, '');
    await ch.prefetch(1);

    await ch.consume(NAMES.MY_SEGMENT_QUEUE, async (msg) => {
        let task;
        try { task = JSON.parse(msg.content.toString()); } catch { return ch.ack(msg); }
        if (task?.name !== CARD.name) return ch.ack(msg);

        // Стратегия «ack сразу»: требует журнала запусков и восстановления при старте (9.2)
        ch.ack(msg);
        await processSegment(task, pub, pool);
    });
})();

17.4 Go: каркас segment-действия

Ключевые структуры и обработка; соединение, переподключение и confirms — по разделу 9.

type SegmentTask struct {
    Name string                 `json:"name"`
    Data map[string]interface{} `json:"data"`
    Meta map[string]interface{} `json:"meta"`
    // Внутренние поля — только с `json:"-"`: статус возвращает сообщение как есть
    RunID string `json:"-"`
}

type ControlSignal struct {
    Signal    string          `json:"signal"`
    SegmentID string          `json:"segmentId"`
    StartupID string          `json:"startupId"`
    Reason    string          `json:"reason"`
    Timestamp json.RawMessage `json:"timestamp"` // встречается и числом, и строкой
}

func (s *Service) handleSegment(d amqp.Delivery) {
    var task SegmentTask
    if err := json.Unmarshal(d.Body, &task); err != nil {
        d.Ack(false)
        return
    }
    if task.Name != ActionName { // чужая задача из fanout — штатно
        d.Ack(false)
        return
    }

    key := fmt.Sprintf("%v:%v", task.Meta["segmentId"], task.Meta["startupId"])
    ctx, cancel := context.WithTimeout(context.Background(), s.cfg.ActionTimeout)
    defer cancel()
    s.registry.Register(key, cancel) // до любой работы; учитывает надгробия stop
    defer s.registry.Unregister(key)

    counter, err := s.run(ctx, &task) // ctx отменяется по stop и по таймауту

    status, errText := "success", ""
    switch {
    case errors.Is(ctx.Err(), context.Canceled):
        status, errText = "error", "Остановлено пользователем: "+s.registry.Reason(key)
    case errors.Is(ctx.Err(), context.DeadlineExceeded):
        status, errText = "error", fmt.Sprintf("превышено время выполнения %s", s.cfg.ActionTimeout)
    case err != nil:
        status, errText = "error", err.Error()
    }

    task.Meta["status"] = status
    task.Meta["counter"] = counter
    task.Meta["error"] = nilIfEmpty(errText)

    // Статус отправляем на контексте, не зависящем от остановки сервиса
    pubCtx, pubCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
    defer pubCancel()
    if err := s.publisher.PublishConfirmed(pubCtx, "", s.names.Status, task); err != nil {
        // Стратегия «ack после статуса»: вернуть задачу, а не слать противоположный статус
        d.Nack(false, true)
        return
    }
    d.Ack(false)
}

Приложение A. Справочник форматов сообщений

Обозначения: О — обязательно, Н — необязательно, П — присылает платформа (вам только читать).

A.1 Карточка flow-действия → FLOWS.flows-custom-actions-register{q}

A.2 Карточка segment-действия → FLOWS.segments-custom-actions-register{q}

A.3 Flow-задача ← ex-custom-actions

A.4 Событие → ex-triggers (ключ маршрутизации = String(id))

A.5 Segment-задача ← ex-segment-custom-actions.{svc}

A.6 Статус → FLOWS.custom-actions-statuses{q}

Исходная задача целиком (A.5) плюс в meta:

A.7 Сигнал ← ex-segments-control.{svc}

A.8 Запись аналитики → FLOWS.analytics{q} (необязательно)

{"table": "log_table", "body": "<JSON-строка записи>"}, состав записи — в 6.4.

A.9 Webhook ← бэкофис

POST {webhooks}/webhooks/{name}, тело — объект, переданный в custom_webhooks(name, body). Ответ — JSON, 200; ошибка — 4xx/5xx с текстом.

Вопросы по контракту и согласование имён очередей и действий для вашей установки — к администратору платформы Accelera.