Engine de Pipeline configurável

O processamento do pedido (e, no futuro, outros fluxos) não é lógica inline — é um grafo de nós configurado por organização, executado por um runtime genérico. Novos comportamentos comuns são configuração (um novo pipeline); só uma capacidade nova de nó exige código. Decisão em ADR-019; princípios em .claude/skills/event-driven.

Esta é a fatia síncrona (E-6.1). A execução assíncrona (order.post, outbox, filas, retry/DLQ) está no E-6.2 — ver Pós-processamento assíncrono. Webhooks no E-6.3; editor visual no E-6.4.

Modelo

• Pipeline — definição persistida por organização e trigger (order.pre, order.post, webhook.received… — string extensível, sem migração). Guarda um grafo definition = { nodes, edges } (JSONB), enabled e version (incrementa a cada alteração da definição; o histórico preserva a versão usada).
• PipelineExecution — uma execução: status (RUNNING/SUCCEEDED/FAILED), tempos, subjectType/subjectId (ex.: order + hashId), correlationId, input/output.
• PipelineStepExecution — uma linha por nó executado: status, duração, branch seguido, output, erro, logs. É a fonte das métricas de desempenho por step.

Gatilhos (trigger)

O trigger é um nome livre, não um enum — é o que permite criar gatilhos novos sem migração e sem tocar no código. A convenção evento → trigger vive em apps/api/src/events/event-queue.ts (triggerForEvent):

Origem	Trigger	Quem dispara
Pedido, antes de confirmar	order.pre	OrdersService
Evento OrderConfirmed	order.post	worker do outbox
Webhook de entrada com eventType sem ponto	webhook.received (default) ou event.<Tipo>	ingestão de webhook
Webhook de entrada com eventType com ponto	o próprio eventType (ex.: webhook.sap-order-status, erp.order.created)	ingestão de webhook
Evento de domínio emitido (inclusive pelo nó emit-event)	event.<Tipo>	worker do outbox

Dois pipelines habilitados no mesmo trigger: roda só o de key menor (ordem alfabética). A tela de criação avisa quando isso vai acontecer.

Nós (tipos disponíveis)

Tipo	O que faz	Config principal
start	Ponto de entrada do grafo (no-op).	—
validate	Avalia um predicado estruturado; passa ou rejeita.	conditions, message
if	Desvia por uma condição: saída then (passou) ou else (não passou). Nunca falha.	conditions
switch	Roteia por casos, na ordem: sai pelo primeiro que casar, senão por default. Nunca falha.	cases: [{ label, when }]
transform	Define/copia campos no contexto (enriquecimento/cálculo).	set: [{ target, from? , value? }]
set-status	Acumula uma entrada de histórico de status.	statusTag, name, phase?
emit-event	Registra um evento pretendido (publicação real = E-6.2).	eventType, payload?/payloadFrom?
http-request	Chamada externa (validar/enriquecer contra ERP), com timeout.	method, url, headers?, body?, saveAs?
apply-order-status	Casa um pedido por código externo e grava no histórico de status.	matchField?, matchFrom?, statusTag, name, identifierFrom?

Um novo tipo de nó é um handler registrado no código (buildNodeRegistry). A combinação dos nós é configuração.

Predicado estruturado (sem linguagem de expressão)

validate usa condições declarativas — seguras por construção (sem eval), ideais para o editor no-code:

{
  "all": [
    { "field": "order.totalAmount", "op": "gte", "value": 100 },
    { "any": [
      { "field": "order.clientExternalCode", "op": "exists" },
      { "field": "order.attributes.allowNoClient", "op": "eq", "value": true }
    ]}
  ]
}

Operadores: eq ne gt gte lt lte in nin exists empty contains regex. Composição: all / any / not. Campos por dot-path no contexto (ex.: order.items.length).

Execução

O executor acha o nó inicial (start, ou o único nó sem entrada) e percorre o grafo seguindo as arestas pelo rótulo que o nó devolve: success, error, default, then/else (do if) ou o nome de um caso (do switch). Uma aresta com branch específico tem prioridade; uma aresta sem branch é catch-all.

Política de erro por nó (onError):

• abort (padrão) — para o pipeline e marca a execução como FAILED (o motivo vira mensagem de rejeição).
• branch — segue a aresta error (fluxo de compensação/alternativo).
• continue — ignora o erro e segue o caminho normal.

Desvio de caminho: if e switch (ADR-046)

Ramificar por regra de negócio é trabalho do if/switch, não do onError. Os dois sempre sucedem: escolher caminho não é falha, então nada entra nos motivos de rejeição e nenhum passo aparece como FAILED no histórico — o ramo error volta a significar só o que quebrou.

{ "id": "vip", "type": "if",
  "config": { "conditions": { "field": "order.customer.tier", "op": "eq", "value": "VIP" } } }

Saídas: then (condição passou) e else (não passou).

{ "id": "rota", "type": "switch", "config": { "cases": [
  { "label": "aprovado", "when": { "field": "sap.status", "op": "eq", "value": "O" } },
  { "label": "pendente", "when": { "field": "sap.status", "op": "eq", "value": "P" } }
] } }

Saídas: aprovado, pendente e default (nenhum caso casou). A ordem importa — vence o primeiro caso que casar. No editor, cada saída é uma porta nomeada no cartão: arraste a bolinha da saída até o próximo passo e a aresta já nasce com o rótulo certo.

Um orçamento de passos protege contra ciclos. O contexto tipado flui em memória entre os nós (data, statusEntries, emittedEvents) — sem re-leituras do banco a cada passo.

Ligação com o pedido (order.pre)

Ao criar um pedido, o OrdersService roda o pipeline order.pre da organização (se houver um habilitado) sobre o rascunho já com os totais calculados, antes de persistir:

• Sem pipeline → comportamento idêntico ao de antes (nada muda).
• validate que aborta → o pedido é rejeitado com HTTP 422 e os motivos.
• transform pode enriquecer order.attributes (mesclado no pedido).
• set-status alimenta o histórico de status do pedido.

order.post (assíncrono, pós-confirmação) não roda aqui — é o E-6.2.

Endpoints (API NestJS)

Escopados por organização (ActiveOrgGuard) + RBAC.

Método	Rota	Permissão	Descrição
GET	/api/pipelines	pipeline:read	Lista pipelines (resumo).
GET	/api/pipelines/:id	pipeline:read	Pipeline com a definição do grafo.
POST	/api/pipelines	pipeline:write	Cria (valida a estrutura do grafo).
PATCH	/api/pipelines/:id	pipeline:write	Atualiza (nova versão se a definição mudar).
DELETE	/api/pipelines/:id	pipeline:write	Remove (e o histórico de execuções).
POST	/api/pipelines/:id/test	pipeline:run	Roda com input de amostra; devolve o trace.
GET	/api/pipelines/:id/executions	pipeline:read	Histórico de execuções (com steps).

Schemas completos na Referência da API (tag Pipeline · Engine).

Exemplo de definição

{
  "nodes": [
    { "id": "start", "type": "start" },
    { "id": "min", "type": "validate", "name": "Valor mínimo",
      "config": { "conditions": { "field": "order.totalAmount", "op": "gte", "value": 100 },
                  "message": "Pedido abaixo do mínimo de R$100" },
      "onError": "abort" },
    { "id": "flag", "type": "transform",
      "config": { "set": [{ "target": "order.attributes.preApproved", "value": true }] } },
    { "id": "st", "type": "set-status",
      "config": { "statusTag": "success", "name": "Pré-validado", "phase": "order.pre" } }
  ],
  "edges": [
    { "from": "start", "to": "min" },
    { "from": "min", "to": "flag", "branch": "success" },
    { "from": "flag", "to": "st" }
  ]
}

Segurança

O nó http-request faz egress configurável — risco de SSRF. Mitigação atual: só http/https, timeout curto (10s, teto 30s), sem seguir redirects. A permissão pipeline:write é sensível (quem edita um pipeline pode disparar chamadas externas). Endurecimento planejado: allowlist de hosts por organização / egress-proxy antes de liberar o nó amplamente.