Fluxo inbound completo

Da mensagem chegar no gateway até o engine executar o flow — passo a passo com decisões arquiteturais.

---

Visão geral

O caminho de uma mensagem inbound atravessa três serviços em sequência, comunicando-se exclusivamente via Redis Streams. O primeiro salto depende do canal: três gateways distintos publicam no mesmo stream:incoming (IncomingStreamEvent):

• flow-ai-meta-api — WhatsApp Cloud API (HTTP POST /webhook)
• flow-ai-ig-api — Instagram DMs e comentários (HTTP POST /webhook)
• flow-ai-webchat-gateway — WebChat (Socket.io)

WhatsApp / Instagram / WebChat (usuário)
    │  HTTP webhook (WA/IG) ou Socket.io (WebChat)
    ▼
flow-ai-meta-api | flow-ai-ig-api | flow-ai-webchat-gateway
    │  stream:incoming (IncomingStreamEvent)
    ▼
flow-ai-orchestrator      ← persiste, gerencia sessão, roteia por canal
    │  stream:flow | stream:helpdesk | stream:agent
    ▼
flow-ai-engine            ← executa a máquina de estados do flow
    │  stream:outgoing-meta | stream:outgoing-chat | stream:outgoing-ig
    ▼
gateway do canal          ← envia resposta ao usuário

Nenhum serviço chama outro via HTTP. As únicas exceções são as fronteiras externas: a Meta Cloud API e o Instagram Graph API chegam via HTTP nos respectivos gateways, o WebChat chega via Socket.io no flow-ai-webchat-gateway, e os gateways chamam de volta as APIs externas para enviar. Tudo mais é assíncrono via streams.

O fluxo abaixo detalha o caminho WhatsApp (canal de referência); as seções marcadas cobrem as diferenças de WebChat e Instagram.

---

Fase 1 — recepção e validação (gateways)

O caminho descrito aqui é o do WhatsApp (flow-ai-meta-api). O flow-ai-ig-api e o flow-ai-webchat-gateway seguem a mesma ideia — validar, normalizar, publicar em stream:incoming — com as diferenças descritas em Diferenças por canal.

1.1 Handshake do webhook (GET)

Antes de qualquer mensagem, a Meta faz um GET com hub.verify_token para validar o endpoint. O meta-api:

1. Lê hub.verify_token da query string.
2. Busca meta:verify-token:{verifyToken} no Redis → recebe { routerId }.
3. Se existir, responde com hub.challenge. Caso contrário, 403.

Por que o token fica no Redis e não no .env? Um mesmo processo do flow-ai-meta-api pode servir múltiplos routers, cada um com seu próprio verify token. Centralizar no .env tornaria inviável multi-tenancy no mesmo processo. O flow-ai-core é o único responsável por gravar esses caches quando um Router é criado ou atualizado.

1.2 Validação de assinatura (POST)

Para cada POST de webhook, a Meta envia o header x-hub-signature-256 com HMAC-SHA256 do corpo usando o app_secret daquele número.

O meta-api:

1. Extrai phoneNumberId do payload (entry[0].changes[0].value.metadata.phone_number_id).
2. Busca wa:webhook:{phoneNumberId} no Redis → recebe { phoneNumberId, routerId, appSecret }.
3. Calcula HMAC-SHA256 do raw body com appSecret.
4. Compara com o header. Se divergir, 403.

Por que não usar o app_secret do .env? Mesmo motivo: cada Router tem seu próprio app secret, e o meta-api não tem conhecimento de routers. Ele só sabe sobre phoneNumberId. O flow-ai-core mantém esse cache decifrado (o valor no Postgres é cifrado via AES-256-CBC).

1.3 Normalização da mensagem

Após validação, a função toIncomingMessage() converte cada mensagem do payload bruto da Meta para o tipo interno WhatsAppIncomingMessage. O conjunto de tipos suportados (SUPPORTED_MESSAGE_TYPES em webhook.ts) cobre 11 variantes:

type WhatsAppIncomingMessage =
  | WhatsAppIncomingTextMessage
  | WhatsAppIncomingImageMessage
  | WhatsAppIncomingAudioMessage
  | WhatsAppIncomingVideoMessage
  | WhatsAppIncomingDocumentMessage
  | WhatsAppIncomingStickerMessage
  | WhatsAppIncomingInteractiveMessage  // button_reply e list_reply
  | WhatsAppIncomingLocationMessage
  | WhatsAppIncomingReactionMessage
  | WhatsAppIncomingContactsMessage
  | WhatsAppIncomingButtonMessage       // botão de template (payload)

Só tipos genuinamente desconhecidos (fora do conjunto acima, ex: unsupported) são descartados silenciosamente — o meta-api nunca bloqueia nem retorna erro para a Meta por causa de tipo desconhecido, pois a Meta exige resposta HTTP 200 rápida independente do que for feito com a mensagem. location, reaction, contacts e button são normalizados e persistidos (o orchestrator os trata em extractContent()).

1.4 Publicação em stream:incoming

O meta-api agrupa todas as mensagens do mesmo POST por (phoneNumberId, from) e publica um único IncomingStreamEvent por grupo:

type IncomingStreamEvent = {
  channelType: "whatsapp" | "webchat" | "instagram"
  sessionId: string        // formato depende do canal (ver abaixo)
  phoneNumberId: string    // WA: phoneNumberId | WC: wc-{channelId} | IG: igUserId
  from: string             // WA: E.164 sem + | WC: userId | IG: IGSID
  contactName: string
  messages: WhatsAppIncomingMessage[]  // contrato sempre WhatsApp-shaped
  timestamp: number
}

O sessionId é derivado deterministicamente por canal — no WhatsApp, wa-{phoneNumberId}-{from}. Isso garante que todas as mensagens do mesmo usuário, para o mesmo número, sempre caiam na mesma sessão — sem coordenação externa. Os formatos de cada canal:

Canal	Formato do sessionId
WhatsApp	wa-{phoneNumberId}-{from}
WebChat	wc-{channelId}-{userId}
Instagram	ig-{igUserId}-{from}

Nota: o JSDoc de IncomingStreamEvent (stream-events.ts) descreve o Instagram como ig-{igUserId}, mas o flow-ai-ig-api emite de fato ig-{igUserId}-{from} (webhook.ts) — inclusive from sendo o IGSID do contato. O formato com -{from} é o que roda em produção.

O messages é sempre WhatsAppIncomingMessage[] independente do canal: o contrato inbound é WhatsApp-shaped, e WebChat/Instagram reaproveitam os campos (phoneNumberId, from, whatsApp.contact) como slots genéricos de transporte.

Por que agrupar antes de publicar? A Meta pode entregar múltiplas mensagens no mesmo webhook POST quando o usuário manda várias rapidamente. Publicar como um evento único garante que o orchestrator processe o conjunto atomicamente, sem risco de criar sessões duplicadas em paralelo.

O meta-api também publica stream:status para cada statuses[] recebido (delivered, read, failed etc.), mas isso pertence ao fluxo de status e não ao inbound.

---

Fase 1b — Diferenças por canal (WebChat e Instagram)

Os três gateways publicam o mesmo IncomingStreamEvent em stream:incoming, mas validam e normalizam de formas distintas:

Aspecto	WhatsApp (meta-api)	Instagram (ig-api)	WebChat (webchat-gateway)
Transporte	HTTP POST /webhook	HTTP POST /webhook	Socket.io (/webchat/socket.io)
Handshake / verify token	Redis meta:verify-token:{token}	env IG_VERIFY_TOKEN	— (sem handshake)
Validação de assinatura	HMAC-SHA256, appSecret via Redis wa:webhook:{phoneNumberId}	HMAC-SHA256, appSecret via Redis ig:webhook:{igUserId}	canal existe via wc:channel:{channelId}; sessionKey validada por bcrypt contra WebChatIdentity.sessionKeyHash (Postgres)
channelType	"whatsapp"	"instagram"	"webchat"

Instagram — o flow-ai-ig-api normaliza DMs em toIncomingMessages(): postbacks e quick replies viram interactive.button_reply (id = payload), texto vira text, anexos viram mídia; echoes (is_echo) são ignorados. Comentários de post/live chegam por entry.changes (não por messaging) e seguem outro caminho — são publicados em stream:ig-comments como IgCommentStreamEvent, não em stream:incoming. Note a assimetria em relação ao WhatsApp: o verify token vem do env (IG_VERIFY_TOKEN), enquanto o app secret da assinatura HMAC vem do Redis (ig:webhook:{igUserId}).

WebChat — sem HMAC nem verify token. O browser autentica via auth ({ channelId, userId?, sessionKey? }); o gateway confirma que o canal existe consultando wc:channel:{channelId} e delega a identidade a negotiateIdentity() — que valida a sessionKey por bcrypt contra WebChatIdentity.sessionKeyHash (Postgres) ou cria uma identidade nova. Em seguida coloca o socket na sala wc-{channelId}-{userId}. Cada message/message_interactive vira um IncomingStreamEvent publicado em stream:incoming.

---

Fase 2 — flow-ai-orchestrator: persistência e sessão

O orchestrator consome stream:incoming no grupo orchestrator. É o único consumer desse stream.

2.1 Deduplicação

Ao persistir cada mensagem, o orchestrator verifica se ela já foi processada buscando whatsappMessageId (o id da mensagem) no banco. A Meta garante entrega "at least once" — duplicatas são normais em falhas de rede. O whatsappMessageId funciona como idempotency key: mensagens já persistidas são puladas, e se todas forem duplicadas o orchestrator retorna sem publicar em stream:flow/stream:helpdesk/stream:agent.

2.2 Resolução de Contact e Chat

O orchestrator resolve (ou cria) as entidades de domínio, escopando o Chat pelo routerId resolvido:

• Contact: identificado por phoneNumber. No WhatsApp é o número E.164; no Instagram é ig-{from} (IGSID sintético); no WebChat é wc-{channelId}-{userId} (sintético). Armazena nome e origin (whatsapp | instagram | webchat).
• Chat: container de todas as mensagens daquela conversa, escopado por contactId + routerId + status: "open".

No WhatsApp e no Instagram, o Contact é criado sob demanda (findUnique ?? create). No WebChat, porém, o Contact é pré-criado pelo próprio gateway: se o orchestrator não o encontrar, registra erro e retorna sem processar (não faz create). O Chat, em todos os canais, é reaproveitado se já houver um open para aquele contato+router, senão é criado.

2.3 Resolução de sessão

A sessão (SessionState) vive no Redis com TTL. O orchestrator tenta carregar a sessão existente:

Sessão existente → atualiza lastInteractionAt e metaWindowExpiresAt, mantém o restante.

Sessão inexistente → precisa descobrir qual flow executar. A resolução do roteamento inicial é por canal (channelType):

Canal	Cache consultado	Helper	Config
WhatsApp	wa:router:{phoneNumberId}	getWaRouter	WhatsAppRouterConfig
Instagram	ig:router:{igUserId}	getIgRouter	InstagramRouterConfig
WebChat	wc:channel:{channelId}	getWcChannel	WebChatChannelConfig

// WhatsApp e Instagram têm o mesmo shape (com a chave de identidade do canal)
type WhatsAppRouterConfig = {
  phoneNumberId: string
  routerId: string
  initialFlowId: string
  sessionExpiryMs?: number   // TTL configurável por router/flow
}

type InstagramRouterConfig = {
  igUserId: string
  routerId: string
  initialFlowId: string
  sessionExpiryMs?: number
}

// WebChat carrega o initialFlowId do router (pode ser null)
type WebChatChannelConfig = {
  id: string
  name: string
  routerId: string
  initialFlowId: string | null
  isTestChannel: boolean
}

Com routerId e initialFlowId em mãos, cria o SessionState inicial (trecho — a criação real também preenche source, whatsApp, contactIdentity, sessionHistory):

{
  mode: "flow",
  routerId: "...",
  flowId: "...",        // = initialFlowId
  contactId: "...",
  chatId: "...",
  startedAt: Date,
  lastInteractionAt: Date,
  metaWindowExpiresAt: Date,       // = lastInteractionAt + 24h (janela de conversação Meta)
  currentBlockId: undefined,       // preenchido pelo engine no primeiro run
  variables: {},
  expiryTimeout: sessionExpiryMs,  // do router/flow; default DEFAULT_SESSION_TTL_MS (30 min)
  source: "whatsapp",              // "whatsapp" | "webchat" | "instagram"
  whatsApp: { contact: { name, phoneNumber } },
}

O expiryTimeout não é mais fixo em 30 minutos: ele vem do sessionExpiryMs do router/flow, com DEFAULT_SESSION_TTL_MS (30 min, definido em orchestrator/src/constants.ts) apenas como fallback quando o cache não traz o valor.

Por que a sessão fica no Redis e não no Postgres? Sessões são lidas e escritas em cada mensagem — latência importa. O Redis fornece acesso em <1ms. O Postgres tem os dados históricos (mensagens, tickets), mas o estado vivo da sessão precisa ser acessado e atualizado atomicamente pelo engine sem round-trip de banco.

O que acontece se o cache de roteamento não existir? A mensagem é persistida no Chat, mas nenhum flow é iniciado. O orchestrator registra o erro e faz ACK da entry (não há retry útil — o cache precisa ser preenchido pelo flow-ai-core). O mesmo vale quando o router não tem initialFlowId.

2.4 Roteamento da sessão

Com a sessão resolvida, o orchestrator decide para onde publicar conforme session.mode:

session.mode	Destino	Evento
"flow"	stream:flow	FlowStreamUserInputEvent (ou FlowStreamExternalJumpEvent)
"human"	stream:helpdesk	HelpdeskMessageEvent
"agent"	stream:agent	AgentStreamEvent

// Quando mode = "flow"
type FlowStreamUserInputEvent = {
  kind: "userInput"
  sessionId: string
  chatId: string           // necessário para persistir respostas
  to: string               // destinatário (E.164/IGSID/userId)
  phoneNumberId: string
  messages: WhatsAppIncomingMessage[]
  timestamp: number
}

O mode: "agent" (agente de IA) é o terceiro destino: quando session.agentState existe, o orchestrator publica um AgentStreamEvent em stream:agent, consumido pelo flow-ai-agent para rodar o loop de tool-calling do LLM.

Por que o orchestrator toma essa decisão e não o engine? O engine não conhece o estado de helpdesk nem o loop do agente. Separar a preocupação de roteamento (orchestrator) da execução (engine/agent) mantém o engine simples: ele só sabe executar flows, nunca precisa saber se aquela sessão está em atendimento humano ou sob um agente de IA.

Gatilho de comentário do Instagram (externalJump). No canal Instagram, um comentário pode iniciar uma sessão de flow. Ao criar a sessão, o orchestrator consome (GETDEL) o destino pendente gravado pelo flow-ai-ig-api:

• inbound sintético (id começa com ig-comment-, caminho legado): consome ig:comment-intent:{igUserId}:{from} via takeIgCommentIntent e marca a 1ª mensagem de saída como Private Reply (setIgPendingCommentReply);
• primeira resposta real do contato (DM): consome ig:comment-destination:{igUserId}:{from} via takeIgCommentDestination (a abertura já foi enviada pelo ig-api).

Quando o destino é um bloco específico (flow_block), o orchestrator publica um FlowStreamExternalJumpEvent com executeOnEntry: true em vez de userInput — assim o engine entra no bloco de destino e publica o conteúdo dele, em vez de tratar a mensagem como input do bloco.

2.5 Política de ACK

O orchestrator só faz ACK após sucesso completo do handler. Se o handler lançar exceção (falha de banco, Redis indisponível, etc.), a entry fica na pending list e será reentregue automaticamente. Isso implementa at-least-once com retry sem infraestrutura adicional.

---

Fase 3 — flow-ai-engine: execução do flow

O engine consome stream:flow no grupo engine. É o único consumer desse stream.

3.1 Bootstrap e cache de flows

No startup, o engine:

1. Chama bootstrapFlowsCache() — carrega todos os flows publicados do Postgres para uma única chave Redis flows (FLOWS_CACHE_KEY), um payload agregado lido inteiro por getFlowsCache().
2. Subscreve ao canal pub/sub bot:invalidate — quando o flow-ai-core publica um novo flow, o engine recarrega o cache sem restart.

Por que cache de flows no Redis em vez de buscar do Postgres a cada execução? Um flow pode ter centenas de blocos e ser executado centenas de vezes por minuto. Buscar do Postgres a cada execução adicionaria latência e carga desnecessária. O cache é invalidado on-demand via pub/sub quando o flow é republicado.

3.2 Consumo e execução

Para cada FlowStreamUserInputEvent:

1. Carrega SessionState do Redis.
2. Carrega FlowDefinition do cache.
3. Carrega routerVars e flowVars do Redis (se existirem).
4. Executa a máquina de estados: avalia inputActions → content → outputConditions → avança para próximo bloco.
5. Para cada mensagem gerada, publica no stream de saída do canal, escolhido por resolveContentChannel(sessionId) (prefixo do sessionId): ig-* → stream:outgoing-ig, wc-* → stream:outgoing-chat, senão stream:outgoing-meta.
6. Salva SessionState atualizado no Redis (com currentBlockId do bloco em que parou aguardando input).

O engine aceita também FlowStreamHumanSessionEndedEvent — quando um ticket de helpdesk é encerrado, o engine retoma o flow de onde parou (ou no bloco configurado como saída do bloco humanAttendance).

---

Mapa completo de caches Redis envolvidos

Chave	Preenchido por	Lido por	Conteúdo
meta:verify-token:{token}	flow-ai-core	flow-ai-meta-api	{ routerId }
wa:webhook:{phoneNumberId}	flow-ai-core	flow-ai-meta-api	{ phoneNumberId, routerId, appSecret }
ig:webhook:{igUserId}	flow-ai-core	flow-ai-ig-api	{ igUserId, routerId, appSecret }
wa:router:{phoneNumberId}	flow-ai-core	flow-ai-orchestrator	WhatsAppRouterConfig
ig:router:{igUserId}	flow-ai-core	flow-ai-orchestrator	InstagramRouterConfig
wc:channel:{channelId}	flow-ai-core	webchat-gateway, orchestrator	WebChatChannelConfig (TTL 3600s)
session:{sessionId}	flow-ai-orchestrator	flow-ai-engine	SessionState completo
flows	flow-ai-engine (boot)	flow-ai-engine	payload agregado de todos os flows publicados
vars:router:{routerId}	flow-ai-core	flow-ai-engine	Record<string, string> decifrado
vars:flow:{flowId}	flow-ai-core	flow-ai-engine	Record<string, string> decifrado
ig:comment-intent:{igUserId}:{from}	flow-ai-ig-api	flow-ai-orchestrator	destino do gatilho (legado, GETDEL)
ig:comment-destination:{igUserId}:{from}	flow-ai-ig-api	flow-ai-orchestrator	destino pendente do gatilho (GETDEL)

---

Por que HTTP só nas bordas

Uma decisão central da arquitetura é que serviços internos nunca se chamam via HTTP. Toda comunicação interna é via Redis Streams.

As consequências práticas:

• Resiliência: se o orchestrator reiniciar enquanto o meta-api está publicando, nenhuma mensagem se perde — ela fica na stream e é consumida quando o orchestrator voltar.
• Desacoplamento de deploy: cada serviço pode ser atualizado independentemente sem coordination.
• Backpressure natural: se o engine estiver lento, as entries acumulam na stream sem rejeitar mensagens da Meta.
• Observabilidade: todas as mensagens transitando entre serviços passam por um lugar auditável (Redis Streams com pending list).

O custo é complexidade operacional: Redis se torna um ponto de falha crítico e a depuração de problemas exige inspecionar streams, pending lists e session keys — não apenas logs HTTP.

---

Arquivos relevantes

Arquivo	Papel
services/flow-ai-meta-api/src/http/routes/webhook.ts	WhatsApp: validação HMAC, normalização, publicação em stream:incoming
services/flow-ai-meta-api/src/whatsapp/webhook-payload.ts	Tipos wire format do payload Meta
services/flow-ai-ig-api/src/http/routes/webhook.ts	Instagram: verify token (env), HMAC, normalização de DM/comentário
services/flow-ai-webchat-gateway/src/gateway.ts	WebChat: auth Socket.io e publicação em stream:incoming
services/flow-ai-orchestrator/src/consumer.ts	Loop de consumo de stream:incoming
services/flow-ai-orchestrator/src/handlers/incoming.ts	Dedup, sessão, roteamento por canal e caminho de comentário IG
services/flow-ai-orchestrator/src/constants.ts	DEFAULT_SESSION_TTL_MS, META_WINDOW_MS
services/flow-ai-engine/src/consumer.ts	Loop de consumo de stream:flow
services/flow-ai-engine/src/handlers/flow.ts	Execução da máquina de estados
services/flow-ai-engine/src/runtime/publish.ts	resolveContentChannel — escolhe stream:outgoing-* por prefixo do sessionId
packages/flow-ai-types/src/stream-events.ts	IncomingStreamEvent, FlowStreamEvent
packages/flow-ai-types/src/ai-agents.ts	AgentStreamEvent
packages/flow-ai-types/src/session.ts	SessionState, SessionMode, InputObject
packages/flow-ai-redis/src/cache.ts	Helpers de cache: getWaRouter/getIgRouter/getWcChannel, takeIgComment*
packages/flow-ai-redis/src/streams.ts	xadd, xreadgroup, xack
packages/flow-ai-redis/src/constants.ts	STREAMS, GROUPS