Пайплайны и графовые воркфлоу

Автоматизируйте cron-расписания, DAG с fan-out, ветвления, ретраи, подтверждения человека и воркфлоу по вебхукам с помощью графовых пайплайнов — визуального холста узлов и рёбер, который использует те же serverless-функции, что и ваши HTTP-маршруты.

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

Визуальный редактор графа

Редактор графа рассчитан на HTTP-воркфлоу: триггеры ведут в Lambda и узлы управления потоком, узел Map формирует JSON, а Respond отдаёт HTTP-ответ клиенту.

• В разделе Пайплайны откройте карточку визуального пайплайна и нажмите Открыть визуальный редактор (или создайте пайплайн через Новый визуальный пайплайн).
• Перетаскивайте типы узлов из палитры слева: Триггеры (Manual, HTTP, Cron), Шаги (Lambda, If, Set, Map, Respond, Human), группа Parallel (узлы Parallel и Merge) и Note. Соединяйте выходы и входы, чтобы задать порядок выполнения.
• Клик по узлу открывает его настройки в боковой панели (привязка функции, выражения, JSON-шаблоны, коды ответа и т.д.).
• Узлы Заметка только для документации на схеме: они не выполняются и не должны иметь рёбер — это стикеры на холсте.

В визуальном редакторе соединяют триггеры Manual, HTTP и Cron с узлами Lambda, If, Map, Respond и Human; на скриншоте ниже — холст в консоли.

Типы узлов графа (справочник)

Соединяйте нижние ручки (выходы) с верхней ручкой (входом) следующего узла. У триггеров, Parallel, Map, Set и Merge один выход по умолчанию. У Lambda два выхода — ok и err; у If — true и false. Human gate в режиме approve имеет выходы ok и no (ручки approve/reject); в режиме question — один выход по умолчанию. У Respond и Note исходящих ручек нет. В графе всегда должен быть хотя бы один триггер.

Ручной триггер

Запускает выполнение из UI пайплайна или через API execute. Дополнительной настройки не требует. Используйте этот узел (или другой триггер) как начало пути.

HTTP-триггер

Запускает выполнение, когда публичный HTTP-вход пайплайна получает запрос с настроенным методом и путём (например, POST /hook). Узел также хранит настройки авторизации для этого входа (без проверки, API-ключ или bearer); при добавлении из палитры подставляются значения по умолчанию.

Cron-триггер

Запускает выполнение по расписанию. Задайте пятипольное cron-выражение и строку часового пояса, например UTC. Минимальный интервал — <b>одна минута</b>, а планировщик опрашивает расписание примерно раз в 30 секунд, поэтому запуск срабатывает в пределах примерно полуминуты от намеченного времени. Некорректное cron-выражение — или выражение, срабатывающее чаще минимума, — отклоняется при сохранении или публикации.

Lambda (функция)

Вызывает функцию рабочего пространства; выберите её в панели свойств. Настройка поведения при ошибке определяет, что происходит при сбое функции: failPipeline останавливает запуск; continue продолжает по ветке ok, фиксируя сбой; errorBranch направляет на выход err — подключите его для сценариев восстановления. Оставьте переопределение входа пустым, чтобы передать выход предыдущего узла, тело триггера сразу после триггера или структурированный объект ошибки, если предыдущий шаг упал и граф продолжился. JSON переопределения использует те же шаблонные плейсхолдеры, что и другие поля, но input в этом поле не подставляется — используйте пути trigger, vars и steps. Исполнитель поддерживает retryPolicy в JSON графа; в UI настраиваются функция, onError и переопределение входа.

If (ветвление)

Делит поток по булеву выражению. Подключите обе исходящие ручки — <b>true</b> и <b>false</b>. Условие вычисляется как JavaScript-выражение в изолированном песочнице и оборачивается в <code>Boolean(…)</code>, поэтому работают сравнения (<code>===</code>, <code>!==</code>, <code>==</code>, <code>!=</code> и числовые <code>&gt;</code> / <code>&gt;=</code> / <code>&lt;</code> / <code>&lt;=</code>), логические операторы <code>&amp;&amp;</code> и <code>||</code>, тернарные выражения и один путь для проверки «истинности». Ради безопасности исходник отклоняется заранее, если содержит точки с запятой, переводы строк, обратные кавычки, комментарии <code>//</code> или <code>/*</code>, escape-последовательности <code>\u</code>/<code>\x</code> или идентификаторы <code>import</code>, <code>require</code>, <code>eval</code>, <code>Function</code>, <code>constructor</code>. В выражении доступны привязки <code>input</code>, <code>trigger</code>, <code>vars</code>, <code>steps</code>, <code>inputs</code> и <code>ifResult</code>. <code>input.*</code> строится из выхода предыдущего шага: строка, похожая на JSON-объект или массив, разбирается, поэтому её поля работают как JSON-тело Lambda. Значения из узлов Set поверхностно сливаются в тот же <code>input</code> (при совпадении ключей побеждают vars); можно по-прежнему явно писать <code>vars.ИМЯ</code> или <code>steps.NODE_ID.output…</code>. По умолчанию выход шага — это неизменённый входной payload: значение условия хранится отдельно как <code>ifEvaluated</code> на шаге и не смешивается с output. Чтобы передать булево значение в JSON дальше по графу, задайте <b>форму выхода</b> (outputMapping), например с <code>{{ifResult}}</code>. Необязательная форма выхода принимает <code>{{input}}</code>, <code>{{ifResult}}</code>, <code>{{vars…}}</code> и <code>{{steps…}}</code>.

Пишите булево выражение на JavaScript по <code>input</code>, <code>trigger</code>, <code>vars</code>, <code>steps</code> и <code>inputs</code>. Комбинируйте условия через <code>&amp;&amp;</code> / <code>||</code> и тернарные выражения — например <code>input.score &gt; 50 &amp;&amp; input.tier == 'pro'</code>. Не оборачивайте выражение в <code>{{ }}</code> — такой синтаксис нужен только в JSON-шаблонах. Вместо <code>NODE_ID</code> подставьте реальный идентификатор узла графа. Примеры:

• Строка из upstream — input.tier == 'pro' (литерал в одинарных или двойных кавычках).
• Число из upstream больше порога — input.score > 50
• Строгое равенство с булевым — input.ok === true (справа также false или null).
• Выход другого узла — steps.NODE_ID.output.status == 'done'
• Переменная пайплайна из Set — vars.environment == 'production'
• Число «больше или равно» — input.retryCount >= 3
• Истинность без оператора: один путь, например input.recordId или steps.NODE_ID.output.items — ветка true, если значение truthy.
• Неравенство — input.code != 404 или input.code !== 404 (== и != допускают приведение типов, === и !== — нет).

Parallel

Разветвляет одно входящее ребро на несколько исходящих с ручки parallel. Каждая нижестоящая ветка выполняется независимо, пока их не соберёт Merge (или другой способ объединения).

Merge

Собирает ветки после Parallel. Режим all (единственный поддерживаемый) ждёт завершения каждой непосредственно предшествующей ветки и затем выполняется один раз. Без outputMapping одна входящая ветка передаёт свой payload без изменений, а несколько веток дают объект с картой inputs, где ключи — идентификаторы исходных узлов веток. С outputMapping используйте {{inputs}} для всей карты или {{inputs.someNodeId}} для одной ветки; при единственном входящем ребре также доступен {{input}} с payload этой ветки — как у Map.

Set

Записывает переменные пайплайна для последующих шаблонных плейсхолдеров. Задайте JSON-объект; значения могут содержать шаблонные строки. Разрешённые записи сливаются в vars. Выход узла объединяет эти значения с upstream-объектом, если тот является обычным объектом.

Map

Строит JSON для следующего шага через outputMapping и подстановку шаблонов; upstream-payload доступен в шаблонах как input (см. раздел о JSON-шаблонах ниже). Если outputMapping не задан, движок передаёт input без изменений.

Respond

Завершает запуск по HTTP-триггеру: задайте код ответа, необязательные шаблоны тела ответа и необязательные заголовки (JSON-объект значений заголовков; каждое значение может использовать те же плейсхолдеры, что и тело). Оставьте тело ответа пустым — без outputMapping и без устаревшего bodyMapping, — чтобы вернуть выход предыдущего шага без изменений или тело триггера, если Respond идёт сразу после триггера. Пример объекта заголовков: Content-Type со значением application/json и X-Request-Id со значением {{vars.requestId}}. У узла нет исходящих рёбер. Если клиент должен дождаться этого ответа, предпочитайте синхронный режим выполнения.

Human gate

Приостанавливает запуск до вызова POST /api/pipeline-graph-executions/:executionId/resume. Режим Approve требует двух исходящих рёбер с метками approve и reject; тело API — { "decision": "approve" } или { "decision": "reject" }. Режим Question использует одно стандартное ребро; тело — { "answer": <любой JSON> }. Необязательный promptTemplate — JSON-шаблон ({{input}} — это upstream-payload), сохраняемый на ожидающем шаге для операторов. Выход завершённого шага объединяет upstream-payload с humanGate: { mode, decision или answer, optional prompt }.

Заметка (Note)

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

Как запускаются прогоны

В графе всегда должен быть хотя бы один узел-триггер; триггер определяет, как начинается прогон:

• Ручной триггер — запуск из UI пайплайна или по HTTP через POST /pipeline-graphs/:id/execute с JSON-телом (тело становится payload триггера прогона).
• HTTP-триггер — даёт пайплайну собственный публичный вход на методе и пути (например, POST /hook) с необязательной авторизацией по API-ключу или bearer; подходящий запрос запускает прогон.
• Cron-триггер — запуск по расписанию (например, 0 */6 * * *) с учётом минимального интервала в одну минуту.

Подтверждение человеком

Чтобы поставить прогон на паузу для человека, добавьте узел Human gate (см. справочник узлов выше). В режиме approve он ждёт, пока кто-нибудь не вызовет POST /api/pipeline-graph-executions/:executionId/resume с { "decision": "approve" } или { "decision": "reject" }, и направляет на соответствующую ручку; в режиме question собирает ответ и продолжает по стандартному ребру.

Графовые пайплайны: JSON-шаблоны

В визуальном редакторе графа часть узлов принимает JSON, в строковых значениях которого можно использовать плейсхолдеры. Каждый фрагмент, записанный в синтаксисе двойных фигурных скобок (шаблоны ниже), движок заменяет данными текущего запуска.

• {{input}} — Полезная нагрузка с предыдущего узла (или тело триггера после триггера). Вложенные поля: {{input.field}}. Подставляется для Map, Respond (тело и значения заголовков), необязательного выходного JSON узла If и outputMapping узла Merge, если входящая ветка одна.
• {{trigger.body}}, {{trigger.type}} и т.д. — данные триггера запуска (доступны во всех шаблонных полях этих узлов).
• {{vars.name}} — значения, заданные узлами Set ранее в графе.
• {{steps.NODE_ID.output}} — выход завершённого шага; вместо NODE_ID укажите id узла из графа. Работают вложенные пути (например {{steps.abc.output.score}}).
• {{inputs}} — при разрешении outputMapping узла Merge это объект с ключами по id исходного узла каждой входящей ветки. Используйте {{inputs.idУзла}} для полезной нагрузки одной ветки (подставьте реальный id из графа).
• {{ifResult}} — только при разрешении необязательного outputMapping узла If: булево значение условия. В последующих узлах то же значение: {{steps.ID_УЗЛА_IF.ifEvaluated}} (id узла If). В тексте самого условия недоступно.

Если строка состоит только из одного плейсхолдера, подставленное значение сохраняет JSON-тип (объект, массив, число). Внутри более длинной строки значения превращаются в текст.

Как плейсхолдеры применяются к узлам:

• Lambda — оставьте переопределение входа пустым, чтобы передать выход предыдущего узла (или payload триггера) без изменений. Если задаёте JSON вручную, используйте {{trigger…}}, {{vars…}} и {{steps…}}. Для переопределения Lambda {{input}} не используется.
• Merge — поведение без outputMapping описано выше в разделе Merge. Произвольный outputMapping использует {{inputs…}}; при одном входящем ребре работает и {{input}}.
• If — поле условия без {{ }}. Движок собирает input из выхода предыдущего шага (строки JSON разбираются) и сливает vars из Set. Используйте input.поле, steps.id.output…, vars…. Необязательная форма выхода — JSON как у Map и {{ifResult}}.
• Map — поле outputMapping задаёт объект, который отдаёт узел; {{input}} — вход с предыдущего узла (если поле опущено в JSON, движок по умолчанию пробрасывает {{input}}).
• Respond — необязательный outputMapping формирует тело HTTP; если нет ни outputMapping, ни устаревшего bodyMapping, тело ответа совпадает с входом с предыдущего узла. Значения в headersMapping подчиняются тем же правилам (например Content-Type и application/json).
• Заметка — не участвует в выполнении и в шаблонах; движок её игнорирует. Не подключайте к заметкам рёбра.

// In Map, Respond (body/headers), and Lambda input override, values may be JSON
// with strings like "{{trigger.body.field}}" or "{{steps.lambdaNodeId.output}}".

// Map node — outputMapping (default shape is pass-through of upstream via {{input}})
{
  "data": "{{input}}",
  "userId": "{{input.userId}}",
  "enriched": "{{steps.analyze.output}}"
}

// Respond node — optional outputMapping; omit it to return the upstream body unchanged
{
  "result": "{{input}}",
  "meta": { "source": "pipeline" }
}

// Respond headers (same rules; header values are strings after resolution)
{
  "X-Request-Id": "{{steps.firstLambda.output.requestId}}"
}

// Lambda — input payload override only when you need a custom shape (otherwise leave empty for chain input)
{
  "item": "{{trigger.body}}",
  "prior": "{{steps.stepA.output}}"
}