Cliente MQTT TypeScript con tipos seguros: topics tipados y payloads validados

Escribir MQTT en TypeScript te da autocompletado y comprobaciones en tiempo de compilación, pero solo si tus topics y payloads están realmente tipados. De serie, MQTT.js te entrega una cadena de topic y un Buffer, y todo lo que viene después son conjeturas. En este tutorial definirás un archivo de contratos para topics, QoS y esquemas de payload, envolverás MQTT.js en un pequeño cliente tipado y validarás cada mensaje en tiempo de ejecución, de modo que publicadores y suscriptores no puedan desincronizarse sin que el compilador o el validador te avisen.

Por qué el código MQTT falla en silencio

Los brokers MQTT no saben ni les importa qué hay dentro de un payload. Esa flexibilidad es una ventaja, pero significa que los errores más comunes en los sistemas MQTT nunca lanzan una excepción:

Los tipos estáticos detectan los dos primeros en tiempo de compilación cuando ambos lados comparten el mismo código. La validación en tiempo de ejecución detecta los dos últimos, porque los datos de dispositivos, de otros equipos o de firmware antiguo nunca han pasado por tu compilador. Necesitas ambas cosas.

Configuración del proyecto

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

tsx ejecuta archivos TypeScript directamente y tsc --noEmit comprueba los tipos del proyecto. MQTT.js v5 incluye sus propias definiciones de tipos. zod proporciona esquemas que sirven también como tipos de TypeScript, así que escribes la forma de cada payload una sola vez.

Paso 1: el archivo de contratos

Crea src/contracts.ts. Es la única fuente de verdad que importan todos los publicadores y suscriptores:

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"]>;

La combinación as const satisfies cumple dos funciones. satisfies comprueba cada entrada contra la forma Contract, así que olvidar un esquema o usar QoS 3 es un error de compilación. as const conserva los tipos literales exactos, de modo que PayloadOf<"command"> se resuelve como { action: "on" | "off" | "reboot"; requestId: string } en lugar de un objeto impreciso.

Los constructores de topics son funciones en lugar de cadenas con marcadores, lo que significa que contracts.command.topic(deviceId) se comprueba en número y tipo de argumentos, y nadie vuelve a concatenar cadenas de topic a mano.

Paso 2: un cliente MQTT tipado

Crea 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 enviar lo que nosotros mismos rechazarí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);
    }
  }
}

Tres detalles lo hacen robusto:

• Los payloads salientes también se validan. TypeScript ya comprueba los literales, pero los valores construidos en tiempo de ejecución (a partir de sensores, formularios u otros servicios) todavía pueden violar restricciones como humidity <= 100. schema.parse lanza un error antes de que un mensaje incorrecto salga del proceso.
• Los payloads entrantes usan safeParse. Un mensaje mal formado de un dispositivo se notifica mediante onInvalid en lugar de tumbar al suscriptor o colarse en la lógica de negocio como un objeto con un tipo incorrecto.
• Las suscripciones se restauran en cada conexión, usando el QoS de cada contrato, así que las reconexiones son invisibles para el código que llama.

Paso 3: úsalo

Crea 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)}`,
});

// el payload está 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 líneas no compilan, y esa es precisamente la idea:
// 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"));

Ejecútalo y comprueba los tipos por separado:

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

Descomenta las dos líneas que "no compilan" y tsc informa de un valor de enum no válido para action, un argumento de topic que falta y una cadena donde se necesita un número. Ese es exactamente el tipo de error que, de otro modo, llega a producción.

Para ver la validación en tiempo de ejecución en acción, publica un payload roto desde la línea de comandos en el servidor MQTT público gratuito:

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"}'

La demo registra un aviso Invalid payload con los campos que fallan y tu handler nunca se llama. El área pública la puede leer cualquiera, así que cambia ts-7f3a por tu propia cadena aleatoria.

Compartir contratos entre servicios y equipos

El archivo de contratos es más valioso cuando todos los componentes lo importan:

1. Monorepo: pon contracts.ts en un paquete compartido que usen los servicios de backend, el dashboard de React y las herramientas de prueba.
2. Repositorios separados: publica los contratos como un paquete npm privado y versiónalo. Un cambio de esquema incompatible se convierte en un salto de versión mayor que los consumidores adoptan cuando deciden.
3. Dispositivos sin TypeScript: genera JSON Schema a partir de los esquemas de zod con una librería de conversión y compártelo con los equipos de firmware, para que el código de ESP32 o Python pueda validar contra la misma definición.
4. Evoluciona con seguridad: añade primero los campos nuevos como opcionales, despliega los suscriptores y después actualiza los publicadores. Para payloads que cambian de forma por completo, añade un segmento de versión al topic.

Probar tus contratos

Los contratos son código, así que pruébalos como código. Tres tipos de pruebas dan el mayor valor con el menor esfuerzo:

1. Fixtures de esquema. Mantén una carpeta con payloads reales capturados de dispositivos, incluidas versiones antiguas de firmware, y comprueba que cada uno pasa safeParse para su contrato. Cuando alguien endurece un esquema, la prueba muestra de inmediato qué dispositivos desplegados empezarían a fallar la validación.
2. Ida y vuelta de topics. Para cada contrato, construye un topic con topic() y comprueba que topicMatches(filter, builtTopic) es verdadero. Así detectas el error clásico de cambiar un constructor sin actualizar su filtro.
3. Pruebas a nivel de tipos. Archivos que no deberían compilar, comprobados con comentarios // @ts-expect-error, demuestran que los payloads no válidos siguen rechazándose. Si alguien relaja un tipo por accidente, tsc señala la directiva sin usar.

Las tres se ejecutan en milisegundos y no necesitan broker, así que deberían formar parte de las comprobaciones de cada pull request en lugar de un job nocturno. Añade una única prueba de integración contra el broker público con un prefijo aleatorio para verificar todo el camino de publicación, validación y despacho sobre una conexión real.

Ejecutar contra un broker privado

Pasar del broker público a producción solo cambia el constructor. En ConnectMQ, crea un broker en el portal y pasa las credenciales generadas:

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
});

Un broker MQTT gestionado se encarga de TLS y de la operación del broker, mientras tus contratos tipados se encargan de lo que circula por él. El explorador de topics en vivo del portal de ConnectMQ es una forma rápida de confirmar que los topics generados por tus constructores coinciden con lo que esperas.

Tutoriales relacionados

Si eres nuevo con el cliente subyacente, empieza por MQTT.js en Node.js. Los mismos contratos pueden respaldar los observables del servicio MQTT de Angular. Para otros lenguajes y sus opciones de tipado, consulta la guía de librerías cliente MQTT. La documentación de zod cubre funciones de esquema como refinements y transforms que puedes añadir a tus contratos.

Empieza con ConnectMQ

Ejecuta tus servicios MQTT con tipos seguros en un broker privado en minutos. Regístrate en app.connectmq.com, elige el plan Serverless gratuito sin necesidad de tarjeta de crédito, crea un broker y pasa las credenciales a TypedMqttClient. El plan gratuito admite MQTT 5 y QoS hasta 1, lo que cubre todos los contratos de este tutorial.

Preguntas frecuentes

¿MQTT.js es compatible con TypeScript?

Sí. MQTT.js v5 está escrito en TypeScript e incluye sus propias declaraciones, así que opciones, eventos y paquetes están tipados. Los payloads siguen siendo Buffer, y por eso resulta útil una capa de contratos como esta.

¿Por qué validar en tiempo de ejecución si TypeScript ya comprueba los tipos?

TypeScript solo comprueba el código que compila. Los mensajes de dispositivos, de otros servicios o de firmware antiguo llegan como bytes sin tipo, así que la validación en tiempo de ejecución es la única forma de garantizar que el objeto que recibe tu handler coincide con su tipo declarado.

¿Puedo usar algo distinto de zod?

Sí. Funciona cualquier validador que infiera tipos de TypeScript, como Valibot, TypeBox o ArkType. La estructura del contrato es la misma; solo cambian las definiciones de los esquemas y las llamadas de parseo.

¿Cómo tipo topics con wildcards?

Mantén dos campos por contrato: un filter con + o # para suscribirse y un constructor topic() tipado para publicar. Usa el filtro solo en el lado del suscriptor y extrae los parámetros de la cadena del topic cuando un handler necesite el ID del dispositivo.