MQTT com tipagem segura em TypeScript, MQTT.js e Zod

MQTT com tipagem segura em TypeScript significa duas coisas: o compilador impede que você publique o payload errado no topic errado, e cada mensagem recebida é validada antes de o seu código tratá-la como dado tipado. O MQTT.js 5 traz seus próprios tipos, mas os payloads chegam como Buffer e os topics são strings simples. Este guia fecha essa lacuna com um mapa de topics tipado, tipos de payload inferidos de schemas zod e um pequeno wrapper em volta do cliente MQTT.js.

Os exemplos usam MQTT.js 5, zod 4 e TypeScript 5 em modo strict, contra broker.connectmq.com com topics em testmqtt/ts/....

Install

npm install mqtt zod
npm install -D typescript @types/node

Onde os tipos deixam de ajudar

Código MQTT.js sem tipos compila tranquilamente mesmo quando está errado:

untyped.ts

client.publish("testmqtt/ts/devices/dev-1/telemtry", JSON.stringify({ temp: "21" }));

client.on("message", (topic, payload) => {
  const data = JSON.parse(payload.toString()); // any
  console.log(data.temperature.toFixed(1));     // crashes at runtime
});

Um topic com erro de digitação some em silêncio, uma string onde se esperava número chega ao dashboard, e JSON.parse retorna any, o que desliga a checagem de tipos para tudo o que vem depois. São dois problemas separados: mensagens enviadas precisam de checagem em tempo de compilação, mensagens recebidas precisam de checagem em runtime, porque o broker repassa quaisquer bytes que qualquer cliente envie.

Schemas primeiro, tipos derivados

Escreva cada payload uma única vez como schema zod e deixe o TypeScript inferir o tipo a partir dele. Assim a checagem em runtime e o tipo estático não têm como divergir:

src/schemas.ts

import { z } from "zod";

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(["reboot", "identify", "set-interval"]),
  intervalSec: z.number().int().positive().optional(),
});

export const Status = z.object({
  online: z.boolean(),
  firmware: z.string().optional(),
});

export type Telemetry = z.infer<typeof Telemetry>;

Mantenha os schemas em um pacote que simuladores de firmware, serviços de backend e testes possam importar. Nomes de campos e versionamento são tratados em design de payloads MQTT.

Um mapa de topics tipado

Em seguida, descreva cada topic uma vez: seu padrão com parâmetros nomeados e seu schema. Template literal types conseguem extrair os nomes dos parâmetros da string do padrão, então o compilador sabe que telemetry precisa de um deviceId:

src/topics.ts

import { z } from "zod";
import { Command, Status, Telemetry } from "./schemas";

type ParamNames<T extends string> =
  T extends `${string}{${infer P}}${infer Rest}` ? P | ParamNames<Rest> : never;
type Params<T extends string> = Record<ParamNames<T>, string>;

export const topics = {
  telemetry: { pattern: "testmqtt/ts/devices/{deviceId}/telemetry", schema: Telemetry },
  command: { pattern: "testmqtt/ts/devices/{deviceId}/command", schema: Command },
  status: { pattern: "testmqtt/ts/devices/{deviceId}/status", schema: Status },
} as const satisfies Record<string, { pattern: string; schema: z.ZodType }>;

export type TopicKey = keyof typeof topics;
export type PayloadOf<K extends TopicKey> = z.infer<(typeof topics)[K]["schema"]>;
export type ParamsOf<K extends TopicKey> = Params<(typeof topics)[K]["pattern"]>;

export function topicFor<K extends TopicKey>(key: K, params: ParamsOf<K>): string {
  return topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
    const value = (params as Record<string, string>)[name];
    if (!value || /[+#\/]/.test(value)) {
      throw new Error(`invalid topic parameter ${name}=${value}`);
    }
    return value;
  });
}

export function filterFor(key: TopicKey): string {
  return topics[key].pattern.replace(/\{\w+\}/g, "+");
}

export function matchTopic(topic: string) {
  for (const key of Object.keys(topics) as TopicKey[]) {
    const names: string[] = [];
    const source = topics[key].pattern.replace(/\{(\w+)\}/g, (_, name: string) => {
      names.push(name);
      return "([^/]+)";
    });
    const m = new RegExp(`^${source}$`).exec(topic);
    if (m) {
      const params = Object.fromEntries(names.map((n, i) => [n, m[i + 1]]));
      return { key, params };
    }
  }
  return null;
}

Três detalhes merecem atenção:

as const satisfies … mantém as strings literais dos padrões (para que ParamNames funcione) e ainda verifica que cada entrada tem um padrão e um schema.
topicFor rejeita valores de parâmetro que contenham /, + ou #. Sem essa checagem, um ID de dispositivo como dev-1/# vindo da entrada do usuário transformaria um publish em algo que você não pretendia.
filterFor("telemetry") troca {deviceId} por +, gerando o filtro de assinatura testmqtt/ts/devices/+/telemetry. Os padrões aqui não têm caracteres especiais de regex; escape-os se os seus tiverem. Você pode testar filtros contra topics de exemplo com o topic matcher.

Um wrapper tipado em volta do MqttClient

O wrapper é dono do único listener message. Ele mapeia cada topic recebido de volta para uma chave, faz o parse do JSON, valida com o schema correspondente e só então chama os handlers tipados:

src/typed-mqtt.ts

import type { IClientPublishOptions, MqttClient } from "mqtt";
import type { z } from "zod";
import {
  filterFor, matchTopic, topicFor, topics,
  type ParamsOf, type PayloadOf, type TopicKey,
} from "./topics";

type AnyHandler = (payload: unknown, params: Record<string, string>) => void;

export class TypedMqtt {
  private readonly client: MqttClient;
  private readonly handlers = new Map<TopicKey, AnyHandler[]>();

  constructor(client: MqttClient) {
    this.client = client;
    client.on("message", (topic, buf) => this.dispatch(topic, buf));
  }

  async publish<K extends TopicKey>(
    key: K,
    params: ParamsOf<K>,
    payload: PayloadOf<K>,
    opts: IClientPublishOptions = { qos: 1 },
  ): Promise<void> {
    const schema: z.ZodType = topics[key].schema;
    const data = schema.parse(payload); // strips unknown fields, catches casts
    await this.client.publishAsync(topicFor(key, params), JSON.stringify(data), opts);
  }

  async on<K extends TopicKey>(
    key: K,
    handler: (payload: PayloadOf<K>, params: ParamsOf<K>) => void,
    qos: 0 | 1 | 2 = 1,
  ): Promise<void> {
    const list = this.handlers.get(key) ?? [];
    list.push(handler as unknown as AnyHandler); // the one place types are erased
    this.handlers.set(key, list);
    await this.client.subscribeAsync(filterFor(key), { qos });
  }

  private dispatch(topic: string, buf: Buffer): void {
    const hit = matchTopic(topic);
    if (!hit) return;

    let json: unknown;
    try {
      json = JSON.parse(buf.toString("utf8"));
    } catch {
      console.warn(`[mqtt] not JSON on ${topic}`);
      return;
    }

    const schema: z.ZodType = topics[hit.key].schema;
    const result = schema.safeParse(json);
    if (!result.success) {
      console.warn(`[mqtt] invalid payload on ${topic}`, result.error.issues);
      return;
    }
    for (const h of this.handlers.get(hit.key) ?? []) h(result.data, hit.params);
  }
}

Mensagens inválidas são registradas e descartadas em vez de lançar exceção dentro do event handler, onde ela apareceria longe da causa. Em produção, conte-as ou encaminhe-as para um topic de dead-letter para perceber uma versão de firmware com problema. Os modos de falha comuns estão em erros de parse de JSON no MQTT.

Usando

src/main.ts

import mqtt from "mqtt";
import { TypedMqtt } from "./typed-mqtt";

const client = await mqtt.connectAsync("mqtts://broker.connectmq.com:8883", {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
});
const bus = new TypedMqtt(client);

await bus.on("telemetry", (t, { deviceId }) => {
  // t is { deviceId: string; temperature: number; humidity: number; ts: number }
  console.log(deviceId, t.temperature.toFixed(1), `${t.humidity}%`);
});

await bus.publish(
  "telemetry",
  { deviceId: "dev-1" },
  { deviceId: "dev-1", temperature: 21.4, humidity: 48, ts: Date.now() },
);

await bus.publish("command", { deviceId: "dev-1" }, { action: "set-interval", intervalSec: 30 });

Agora os erros do primeiro exemplo falham em tempo de compilação:

Erro	Mensagem do compilador
`bus.publish("telemtry", …)`	`Argument of type '"telemtry"' is not assignable to parameter of type '"telemetry" \| "command" \| "status"'`
`{ temperature: "21" }`	`Type 'string' is not assignable to type 'number'`
`{ action: "format" }`	`Type '"format"' is not assignable to type '"reboot" \| "identify" \| "set-interval"'`
`bus.publish("status", {}, …)`	`Property 'deviceId' is missing in type '{}' but required in type 'ParamsOf<"status">'`

Adicione linhas @ts-expect-error para esses casos em um arquivo de testes de tipos e rode tsc --noEmit no CI; se alguém afrouxar um schema, o build quebra.

Extras do MQTT 5

Com protocolVersion: 5, você pode colocar a versão do schema nas propriedades do publish, para que os consumidores escolham o parser certo antes de validar:

Publish with MQTT 5 properties

await client.publishAsync("testmqtt/ts/devices/dev-1/telemetry", JSON.stringify(data), {
  qos: 1,
  properties: {
    contentType: "application/json",
    userProperties: { schema: "telemetry/v2" },
  },
});

O lado receptor lê essas propriedades no terceiro argumento do handler message (packet.properties?.userProperties). O gerador de código MQTT produz o trecho de conexão equivalente para outras linguagens se os dispositivos não forem escritos em TypeScript.

Testando a validação

Rode main.ts com npx tsx src/main.ts, abra o cliente MQTT online, conecte em broker.connectmq.com e publique três mensagens em testmqtt/ts/devices/dev-9/telemetry:

{"deviceId":"dev-9","temperature":19.5,"humidity":40,"ts":1760000000000} — o handler registra a mensagem.
{"deviceId":"dev-9","temperature":"hot"} — você vê invalid payload com a lista de issues do zod (expected number, received string e os campos faltando).
not json — você vê not JSON on testmqtt/ts/devices/dev-9/telemetry.

Esses três casos dão um bom teste unitário: entregue ao TypedMqtt um cliente falso, emita eventos message e verifique quais handlers rodaram, como mostrado em testes de MQTT.js com Jest e Vitest.

Enviar payloads de teste ao cliente tipado

Um mapa de topics tipado mais validação de schema na fronteira transforma o MQTT de "quaisquer bytes em qualquer string" em uma interface que você pode refatorar com confiança. Defina topics e schemas uma vez, derive todos os tipos deles, valide tudo o que chega e deixe o compilador pegar o resto.

Perguntas frequentes

O MQTT.js tem tipos TypeScript?

Sim. O MQTT.js 5 é escrito em TypeScript e traz suas próprias declarações, então MqttClient, IClientOptions e IClientPublishOptions estão disponíveis sem pacote @types. Mas os payloads são tipados como Buffer; tipar o conteúdo deles fica por sua conta.

Por que validar payloads MQTT em runtime se já uso TypeScript?

Os tipos do TypeScript somem na compilação e o broker aceita quaisquer bytes em qualquer topic. Bugs de firmware, versões antigas de dispositivos e mensagens de teste manuais chegam como Buffers sem tipo. Um schema zod valida cada mensagem uma vez na fronteira, para que o resto do código possa confiar no tipo.

O TypeScript consegue verificar topics MQTT com wildcards?

Ele verifica os topics que você monta, usando template literal types, então um erro de digitação na chave do topic ou um parâmetro faltando vira erro de compilação. Casar topics recebidos com filtros + e # continua acontecendo em runtime, e é por isso que o wrapper mapeia cada mensagem de volta para uma chave de topic tipada.

MQTT com tipagem segura em TypeScript