Encaminhe mensagens MQTT para webhooks e APIs REST

Para encaminhar mensagens MQTT para um webhook ou uma API REST, rode um pequeno consumidor que assina os topics relevantes e transforma cada mensagem em um POST HTTP. A assinatura é a parte fácil. O que deixa um encaminhador pronto para produção é o que acontece quando o lado HTTP se comporta mal: timeouts, 503 Service Unavailable, 429 Too Many Requests, deploys que derrubam conexões por um minuto. Este tutorial constrói um encaminhador em Node.js com MQTT.js 5 que tenta de novo com backoff exponencial e jitter, respeita Retry-After, assina as requisições, envia chaves de idempotência e estaciona mensagens não entregáveis em um topic de dead-letter.

A arquitetura

devices ──MQTT──▶ broker ──▶ forwarder ──HTTPS POST──▶ webhook / REST API
                                 │
                                 └──▶ testmqtt/webhook/dlq (failed deliveries)

Os dispositivos publicam eventos em testmqtt/webhook/<device>/events. O encaminhador assina com QoS 1, envia cada evento como JSON e decide, com base no resultado HTTP, se finaliza, tenta de novo ou manda para dead-letter. Assim os dispositivos não precisam saber de HTTP, e a API receptora não precisa saber de MQTT. Se o receptor for uma função serverless, o mesmo padrão vale; veja MQTT com AWS Lambda para essa variação.

Regras de retentativa que funcionam

Use backoff exponencial com full jitter: espere um tempo aleatório entre 0 e min(cap, base × 2^attempt). Sem o jitter, todas as mensagens que falharam durante uma queda tentam de novo no mesmo instante e derrubam a API outra vez assim que ela volta.

O encaminhador

Instale o MQTT.js (npm install mqtt). O Node 18+ já traz fetch e AbortSignal.timeout, então nenhuma biblioteca HTTP é necessária:

import mqtt from "mqtt";
import crypto from "node:crypto";

const WEBHOOK_URL = process.env.WEBHOOK_URL ?? "http://localhost:3000/hook";
const SECRET = process.env.WEBHOOK_SECRET ?? "change-me";
const MAX_ATTEMPTS = 6;
const CONCURRENCY = 8;
const DLQ_TOPIC = "testmqtt/webhook/dlq";

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "webhook-forwarder-1",
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
});

client.on("connect", () => {
  console.log("conectado");
  client.subscribe("testmqtt/webhook/+/events", { qos: 1 });
});
client.on("error", (err) => console.error("erro mqtt:", err.message));

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const retryable = (status) => status === 408 || status === 429 || status >= 500;

async function deliver(job) {
  const body = JSON.stringify({
    topic: job.topic,
    payload: job.payload.toString("utf8"),
    receivedAt: new Date().toISOString(),
  });
  const signature = crypto.createHmac("sha256", SECRET).update(body).digest("hex");

  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    let retryAfterMs = 0;
    try {
      const res = await fetch(WEBHOOK_URL, {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "idempotency-key": job.id,
          "x-signature-sha256": signature,
        },
        body,
        signal: AbortSignal.timeout(10_000),
      });
      await res.body?.cancel();                    // libera a conexão
      if (res.ok) return { ok: true };
      if (!retryable(res.status)) return { ok: false, reason: `HTTP ${res.status}` };
      retryAfterMs = (Number(res.headers.get("retry-after")) || 0) * 1000;
      console.warn(`tentativa ${attempt}: HTTP ${res.status}`);
    } catch (err) {
      console.warn(`tentativa ${attempt}: ${err.name} ${err.message}`);
    }
    if (attempt < MAX_ATTEMPTS) {
      const backoff = Math.random() * Math.min(30_000, 500 * 2 ** attempt);
      await sleep(Math.max(backoff, retryAfterMs));
    }
  }
  return { ok: false, reason: "tentativas esgotadas" };
}

const queue = [];
let active = 0;

function drain() {
  while (active < CONCURRENCY && queue.length > 0) {
    const job = queue.shift();
    active++;
    deliver(job)
      .then((result) => {
        if (!result.ok) {
          console.error(`enviando para dead-letter ${job.id}: ${result.reason}`);
          client.publish(DLQ_TOPIC, JSON.stringify({
            topic: job.topic,
            payload: job.payload.toString("base64"),
            id: job.id,
            reason: result.reason,
          }), { qos: 1 });
        }
      })
      .finally(() => { active--; drain(); });
  }
}

client.on("message", (topic, payload) => {
  // chave estável: a mesma mensagem reentregue recebe a mesma chave
  const id = crypto.createHash("sha256").update(topic).update(payload).digest("hex").slice(0, 32);
  queue.push({ topic, payload, id });
  drain();
});

O que cada parte faz:

• Timeout em toda requisição. AbortSignal.timeout(10_000) impede que um endpoint travado prenda todos os workers.
• Concorrência limitada. Oito requisições em andamento ao mesmo tempo; ajuste ao que a API aguenta. Sem limite, um pico de tráfego MQTT vira um pico de tráfego HTTP.
• Assinatura HMAC. O receptor recalcula HMAC-SHA256(secret, body) e rejeita requisições que não batem: prática padrão de webhooks.
• Chave de idempotência. Derivada do topic e do payload, então uma mensagem reentregue leva a mesma chave. Isso supõe que os payloads contenham um timestamp ou ID de evento; se dois eventos realmente diferentes puderem ser idênticos byte a byte, coloque um ID no payload e use-o.
• Topic de dead-letter. Mensagens não entregáveis vão para testmqtt/webhook/dlq em vez de sumir. Assine esse topic a partir de um job de alertas ou reprocesse depois de corrigir a API.

Garantias de entrega e seus limites

Por padrão, o MQTT.js confirma uma mensagem QoS 1 assim que o seu handler de message retorna. Por isso a fila em memória acima se perde se o processo cair. Isso é aceitável para muitos webhooks do tipo notificação. Quando não for, você tem duas opções.

Confirmar depois da entrega. Em vez do handler de message e da fila em memória, sobrescreva handleMessage(packet, callback) do MQTT.js; o PUBACK só é enviado quando você chama callback. As mensagens passam a ser processadas uma por vez por cliente, o que também preserva a ordem:

client.handleMessage = (packet, callback) => {
  const id = crypto.createHash("sha256").update(packet.topic).update(packet.payload).digest("hex").slice(0, 32);
  deliver({ topic: packet.topic, payload: packet.payload, id })
    .then((result) => { if (!result.ok) console.error("dead-letter:", id, result.reason); })
    .finally(() => callback());   // PUBACK só agora
};

Combine com clean: false (MQTT 3.1.1) ou com um session expiry interval (MQTT 5) para que o broker enfileire mensagens enquanto o encaminhador está fora. Para isso o broker precisa manter sessões persistentes: use sua própria instância privada de broker, já que o endpoint de teste compartilhado não guarda sessões offline. Veja sessões persistentes no MQTT.

Colocar uma fila durável no meio. Grave as mensagens em Redis Streams, SQS ou uma tabela de banco de dados e deixe workers HTTP consumirem. Mais peças, mas as retentativas sobrevivem a reinícios.

Nos dois casos o resultado é at-least-once, então a API receptora precisa tolerar duplicatas. Lidando com mensagens MQTT duplicadas cobre padrões de deduplicação.

Escalando horizontalmente

Um único encaminhador dá conta de muitos webhooks, porque quase todo o trabalho é esperar o HTTP. Quando precisar de mais, rode várias instâncias com uma assinatura compartilhada como $share/webhook/testmqtt/webhook/+/events. O broker distribui as mensagens entre as instâncias e, se uma cair, a parte dela passa para as outras. Dê a cada instância um client ID único.

Teste as retentativas

Rode um receptor que falha em 40 % das requisições com 503 e registra duplicatas:

import http from "node:http";

const seen = new Set();
http.createServer((req, res) => {
  let body = "";
  req.on("data", (chunk) => (body += chunk));
  req.on("end", () => {
    if (Math.random() < 0.4) {
      res.writeHead(503, { "retry-after": "1" });
      return res.end();
    }
    const key = req.headers["idempotency-key"];
    console.log(seen.has(key) ? "DUPLICADA" : "aceita   ", key, body);
    seen.add(key);
    res.writeHead(204).end();
  });
}).listen(3000, () => console.log("receptor em :3000"));

Inicie node receiver.mjs, depois node forwarder.mjs, e publique uma rajada de eventos:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
    -t testmqtt/webhook/door-1/events -m "{\"event\":\"open\",\"n\":$i,\"ts\":$(date +%s)}"
done

Todos os eventos devem aparecer em algum momento como aceita, com avisos tentativa N: HTTP 503 no log do encaminhador. Depois teste os outros caminhos:

• Mude o receptor para retornar 400: as mensagens precisam ir direto para o topic de dead-letter, sem retentativas. Acompanhe com mosquitto_sub -t testmqtt/webhook/dlq -v ou no cliente MQTT online.
• Pare o receptor por 30 segundos enquanto publica: as retentativas devem se espaçar e dar certo quando ele voltar.

Acompanhe os eventos e o topic de dead-letter no navegador

Um encaminhador de webhooks tem algumas dezenas de linhas, e quase todas tratam de falhas. Classifique as respostas, faça backoff com jitter, envie chaves de idempotência, nunca descarte uma mensagem em silêncio e rode o teste com o receptor que falha antes de confiar tráfego de produção a ele. Se a API for sua, considere também construir o lado HTTP com Express ao lado do encaminhador.

Perguntas frequentes