Cliente MQTT com TypeScript e Tipagem Segura: Topics Tipados e Payloads Validados

Escrever MQTT com TypeScript oferece autocompletar e verificações em tempo de compilação, mas só se seus topics e payloads estiverem de fato tipados. Por padrão, o MQTT.js entrega uma string de topic e um Buffer, e tudo o que vem depois é adivinhação. Neste tutorial você vai definir um arquivo de contrato para topics, QoS e schemas de payload, envolver o MQTT.js em um pequeno cliente tipado e validar cada mensagem em tempo de execução, para que publishers e subscribers não consigam sair de sincronia sem que o compilador ou o validador avisem.

Por que o Código MQTT Quebra em Silêncio

Os brokers MQTT não sabem nem se importam com o que há dentro de um payload. Essa flexibilidade é um ponto forte, mas significa que os bugs mais comuns em sistemas MQTT nunca geram um erro:

Tipos estáticos pegam os dois primeiros em tempo de compilação quando os dois lados compartilham o mesmo código. A validação em tempo de execução pega os dois últimos, porque dados de dispositivos, de outras equipes ou de firmwares antigos nunca passaram pelo seu compilador. Você precisa dos dois.

Configuração do Projeto

mkdir mqtt-typescript-demo && cd mqtt-typescript-demo
npm init -y
npm pkg set type=module
npm install mqtt zod
npm install -D typescript tsx @types/node
npx tsc --init --strict --target es2022 --module nodenext --moduleResolution nodenext

O tsx executa arquivos TypeScript diretamente, e tsc --noEmit verifica os tipos do projeto. O MQTT.js v5 traz suas próprias definições de tipos. O zod fornece schemas que também servem como tipos TypeScript, então você escreve cada formato de payload uma única vez.

Passo 1: O Arquivo de Contrato

Crie src/contracts.ts. Esta é a fonte única da verdade que todo publisher e subscriber importa:

import { z } from "zod";

export const PREFIX = process.env.TOPIC_PREFIX ?? "connectmq/demo/ts-7f3a";

export const Telemetry = z.object({
  deviceId: z.string(),
  temperature: z.number(),
  humidity: z.number().min(0).max(100),
  ts: z.number().int(),
});

export const Command = z.object({
  action: z.enum(["on", "off", "reboot"]),
  requestId: z.string().uuid(),
});

export const Status = z.enum(["online", "offline"]);

interface Contract<S extends z.ZodType, A extends unknown[]> {
  filter: string;
  topic: (...args: A) => string;
  schema: S;
  qos: 0 | 1 | 2;
  retain: boolean;
}

export const contracts = {
  telemetry: {
    filter: `${PREFIX}/devices/+/telemetry`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/telemetry`,
    schema: Telemetry,
    qos: 0,
    retain: false,
  },
  command: {
    filter: `${PREFIX}/devices/+/cmd`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/cmd`,
    schema: Command,
    qos: 1,
    retain: false,
  },
  status: {
    filter: `${PREFIX}/devices/+/status`,
    topic: (deviceId: string) => `${PREFIX}/devices/${deviceId}/status`,
    schema: Status,
    qos: 1,
    retain: true,
  },
} as const satisfies Record<string, Contract<z.ZodType, any[]>>;

export type Contracts = typeof contracts;
export type ContractKey = keyof Contracts;
export type PayloadOf<K extends ContractKey> = z.infer<Contracts[K]["schema"]>;
export type TopicArgs<K extends ContractKey> = Parameters<Contracts[K]["topic"]>;

A combinação as const satisfies faz dois trabalhos. satisfies verifica cada entrada contra o formato Contract, então esquecer um schema ou usar QoS 3 é um erro de compilação. as const mantém os tipos literais exatos, de modo que PayloadOf<"command"> resolve para { action: "on" | "off" | "reboot"; requestId: string } em vez de um objeto vago.

Os construtores de topic são funções, e não strings com placeholders, o que significa que contracts.command.topic(deviceId) é verificado quanto ao número e ao tipo corretos de argumentos, e ninguém mais concatena strings de topic à mão.

Passo 2: Um Cliente MQTT Tipado

Crie src/typed-client.ts:

import mqtt, { IClientOptions, MqttClient } from "mqtt";
import { contracts, ContractKey, PayloadOf, TopicArgs } from "./contracts.js";

type Handler<K extends ContractKey> = (payload: PayloadOf<K>, topic: string) => void;

export function topicMatches(filter: string, topic: string): boolean {
  const f = filter.split("/");
  const t = topic.split("/");
  for (let i = 0; i < f.length; i++) {
    if (f[i] === "#") return true;
    if (i >= t.length) return false;
    if (f[i] !== "+" && f[i] !== t[i]) return false;
  }
  return f.length === t.length;
}

export class TypedMqttClient {
  private readonly client: MqttClient;
  private readonly handlers = new Map<ContractKey, Set<Handler<any>>>();

  constructor(
    url: string,
    options: IClientOptions,
    private readonly onInvalid: (topic: string, error: unknown) => void = (topic, error) =>
      console.warn(`Invalid payload on ${topic}:`, error)
  ) {
    this.client = mqtt.connect(url, { protocolVersion: 5, reconnectPeriod: 2000, ...options });

    this.client.on("connect", () => {
      for (const key of this.handlers.keys()) {
        const c = contracts[key];
        this.client.subscribe(c.filter, { qos: c.qos });
      }
    });

    this.client.on("message", (topic, buffer) => this.dispatch(topic, buffer));
    this.client.on("error", (err) => console.error("MQTT error:", err.message));
  }

  async publish<K extends ContractKey>(key: K, args: TopicArgs<K>, payload: PayloadOf<K>): Promise<void> {
    const c = contracts[key];
    const validated = c.schema.parse(payload); // nunca envie o que nós mesmos rejeitaríamos
    const topic = (c.topic as (...a: unknown[]) => string)(...args);
    await this.client.publishAsync(topic, JSON.stringify(validated), { qos: c.qos, retain: c.retain });
  }

  subscribe<K extends ContractKey>(key: K, handler: Handler<K>): () => void {
    let set = this.handlers.get(key);
    if (!set) {
      set = new Set();
      this.handlers.set(key, set);
      if (this.client.connected) {
        this.client.subscribe(contracts[key].filter, { qos: contracts[key].qos });
      }
    }
    set.add(handler);
    return () => {
      set!.delete(handler);
      if (set!.size === 0) {
        this.handlers.delete(key);
        this.client.unsubscribe(contracts[key].filter);
      }
    };
  }

  async close(): Promise<void> {
    await this.client.endAsync();
  }

  private dispatch(topic: string, buffer: Buffer): void {
    for (const [key, set] of this.handlers) {
      const c = contracts[key];
      if (!topicMatches(c.filter, topic)) continue;
      let raw: unknown;
      try {
        raw = JSON.parse(buffer.toString());
      } catch (err) {
        this.onInvalid(topic, err);
        continue;
      }
      const result = c.schema.safeParse(raw);
      if (!result.success) {
        this.onInvalid(topic, result.error.issues);
        continue;
      }
      for (const handler of set) handler(result.data, topic);
    }
  }
}

Três detalhes tornam isso robusto:

• Os payloads de saída também são validados. O TypeScript já verifica literais, mas valores construídos em tempo de execução (vindos de sensores, formulários ou outros serviços) ainda podem violar restrições como humidity <= 100. schema.parse lança uma exceção antes que uma mensagem inválida saia do processo.
• Os payloads de entrada usam safeParse. Uma mensagem malformada de um dispositivo é reportada por onInvalid, em vez de derrubar o subscriber ou entrar na lógica de negócio como um objeto com o tipo errado.
• As assinaturas são restauradas a cada conexão, usando o QoS de cada contrato, então as reconexões ficam invisíveis para o código que chama.

Passo 3: Use o Cliente

Crie src/demo.ts:

import { randomUUID } from "node:crypto";
import { contracts } from "./contracts.js";
import { TypedMqttClient } from "./typed-client.js";

const mq = new TypedMqttClient("mqtts://broker.connectmq.com:8883", {
  clientId: `ts-demo-${randomUUID().slice(0, 8)}`,
});

// o payload é totalmente tipado: { deviceId: string; temperature: number; humidity: number; ts: number }
mq.subscribe("telemetry", (payload, topic) => {
  console.log(`${payload.deviceId}: ${payload.temperature.toFixed(1)} C (${topic})`);
});

mq.subscribe("command", (cmd) => {
  console.log(`Command ${cmd.action} (${cmd.requestId})`);
});

await mq.publish("status", ["sensor-42"], "online");

const timer = setInterval(() => {
  mq.publish("telemetry", ["sensor-42"], {
    deviceId: "sensor-42",
    temperature: 20 + Math.random() * 5,
    humidity: 40 + Math.random() * 10,
    ts: Date.now(),
  }).catch((err) => console.error("Rejected by schema:", err.message));
}, 2000);

setTimeout(async () => {
  await mq.publish("command", ["sensor-42"], { action: "reboot", requestId: randomUUID() });
}, 5000);

// Estas linhas não compilam, e é exatamente essa a ideia:
// mq.publish("command", ["sensor-42"], { action: "explode", requestId: "x" });
// mq.publish("telemetry", [], { deviceId: "a", temperature: "21", humidity: 1, ts: 0 });

process.on("SIGINT", async () => {
  clearInterval(timer);
  await mq.publish("status", ["sensor-42"], "offline");
  await mq.close();
  process.exit(0);
});

console.log("Telemetry topic example:", contracts.telemetry.topic("sensor-42"));

Execute e verifique os tipos separadamente:

npx tsx src/demo.ts
npx tsc --noEmit

Descomente as duas linhas que "não compilam" e o tsc aponta um valor de enum inválido em action, um argumento de topic faltando e uma string onde era exigido um número. Essa é exatamente a classe de bug que, de outra forma, chega à produção.

Para ver a validação em tempo de execução em ação, publique um payload quebrado pela linha de comando no servidor MQTT público (gratuito), que serve apenas para testes:

mosquitto_pub -h broker.connectmq.com -p 1883 -i ts-bad-7f3a \
  -t connectmq/demo/ts-7f3a/devices/sensor-9/telemetry -m '{"deviceId":"sensor-9","temperature":"hot"}'

A demo registra um aviso Invalid payload listando os campos que falharam, e seu handler nunca é chamado. O servidor MQTT público pode ser lido por qualquer pessoa, então troque ts-7f3a por uma string aleatória sua.

Compartilhando Contratos Entre Serviços e Equipes

O arquivo de contrato tem mais valor quando todos os componentes o importam:

1. Monorepo: coloque contracts.ts em um pacote compartilhado usado pelos serviços de backend, pelo dashboard React e pelas ferramentas de teste.
2. Repositórios separados: publique os contratos como um pacote npm privado e versione-o. Uma mudança incompatível de schema vira um aumento de versão major que os consumidores adotam quando quiserem.
3. Dispositivos que não usam TypeScript: gere JSON Schema a partir dos schemas zod com uma biblioteca de conversão e compartilhe-o com as equipes de firmware, para que o código de ESP32 ou Python possa validar contra a mesma definição.
4. Evolua com segurança: adicione novos campos primeiro como opcionais, implante os subscribers e só então atualize os publishers. Para payloads que mudam completamente de formato, adicione um segmento de versão ao topic.

Testando Seus Contratos

Contratos são código, então teste-os como código. Três tipos de teste dão o maior retorno com o menor esforço:

1. Fixtures de schema. Mantenha uma pasta com payloads reais capturados de dispositivos, incluindo versões antigas de firmware, e verifique se cada um passa no safeParse do seu contrato. Quando alguém tornar um schema mais restritivo, o teste mostra imediatamente quais dispositivos já implantados começariam a falhar na validação.
2. Ida e volta de topics. Para cada contrato, construa um topic com topic() e verifique se topicMatches(filter, builtTopic) é verdadeiro. Isso pega o erro clássico de mudar um construtor sem atualizar o filtro correspondente.
3. Testes no nível de tipos. Arquivos que não devem compilar, verificados com comentários // @ts-expect-error, provam que payloads inválidos continuam sendo rejeitados. Se alguém afrouxar um tipo por acidente, o tsc aponta a diretiva não utilizada.

Os três rodam em milissegundos e não precisam de broker, então pertencem a toda verificação de pull request, e não a um job noturno. Adicione um único teste de integração contra o servidor MQTT público com um prefixo aleatório para verificar todo o caminho de publicação, validação e despacho sobre uma conexão real.

Executando em um Broker Privado

Trocar o servidor MQTT público pelo broker de produção só muda o construtor. No ConnectMQ, crie um broker no portal e passe as credenciais geradas:

const mq = new TypedMqttClient(`mqtts://${process.env.MQTT_HOST}:8883`, {
  clientId: process.env.MQTT_CLIENT_ID!,   // YOUR_CLIENT_ID
  username: process.env.MQTT_USERNAME!,    // YOUR_USERNAME
  password: process.env.MQTT_PASSWORD!,    // YOUR_PASSWORD
});

Um broker MQTT gerenciado cuida do TLS e da operação do broker, enquanto seus contratos tipados cuidam do que trafega por ele. O explorador de topics ao vivo do portal do ConnectMQ é uma forma rápida de confirmar que os topics gerados pelos seus construtores correspondem ao que você espera.

Tutoriais Relacionados

Se você ainda não conhece o cliente por baixo disso, comece com MQTT.js no Node.js. Os mesmos contratos podem sustentar os observables do serviço de MQTT com Angular. Para outras linguagens e suas opções de tipagem, veja o guia de bibliotecas cliente MQTT. A documentação do zod cobre recursos de schema, como refinements e transforms, que você pode adicionar aos seus contratos.

Comece a Usar o ConnectMQ

Execute seus serviços MQTT com tipagem segura em um broker privado em minutos. Cadastre-se em app.connectmq.com, escolha o plano Serverless gratuito, sem cartão de crédito, crie um broker e passe as credenciais para o TypedMqttClient. O plano gratuito suporta MQTT 5 e QoS até 1, o que cobre todos os contratos deste tutorial.

Perguntas frequentes

O MQTT.js suporta TypeScript?

Sim. O MQTT.js v5 é escrito em TypeScript e traz suas próprias declarações, então opções, eventos e pacotes são tipados. Os payloads continuam sendo Buffer, e é por isso que uma camada de contratos como esta é útil.

Por que validar em tempo de execução se o TypeScript já verifica os tipos?

O TypeScript só verifica o código que ele compila. Mensagens de dispositivos, de outros serviços ou de firmwares antigos chegam como bytes sem tipo, então a validação em tempo de execução é a única forma de garantir que o objeto recebido pelo seu handler corresponde ao tipo declarado.

Posso usar algo diferente do zod?

Sim. Qualquer validador que infira tipos TypeScript funciona, como Valibot, TypeBox ou ArkType. A estrutura dos contratos continua a mesma; só mudam as definições de schema e as chamadas de parse.

Como tipar topics com curingas?

Mantenha dois campos por contrato: um filter com + ou # para assinar e um construtor topic() tipado para publicar. Use o filtro apenas do lado do subscriber e extraia os parâmetros da string do topic quando um handler precisar do ID do dispositivo.