I Got Tired of Untyped MQTT Payloads, So I Built a Library

Enforcing Schema Validation and Type Safety in MQTT Workflows with TypeScript

Current Situation Analysis

In distributed systems leveraging MQTT, the disconnect between the protocol's loose typing and TypeScript's strictness creates significant friction. Engineers frequently encounter runtime failures caused by malformed payloads or mismatched topic structures. The standard mqtt.js interface returns raw Buffer objects, forcing developers to perform manual deserialization and type assertions. This approach shifts error detection from compile-time to runtime, often resulting in silent failures or unhandled exceptions in production environments.

The problem is often overlooked because MQTT is treated purely as a transport layer, while the payload contract is considered an application concern. However, in TypeScript, the topic hierarchy and payload structure are integral to the type system. Without a unified abstraction, teams end up with:

• Stringly-typed topic parsing: Extracting dynamic segments (e.g., deviceId from devices/123/status) requires manual string splitting, which is brittle and prone to index errors.
• Lack of runtime guarantees: A single misbehaving device sending a string instead of a number can crash a handler or corrupt state, as JSON.parse returns any by default.
• Wildcard ambiguity: Subscribing to devices/+/status provides no type information about what the + matched, forcing handlers to re-parse the topic string to determine context.
• Publishing errors: There is no compile-time prevention against publishing to a wildcard pattern, which is invalid in MQTT and leads to broker rejections or unexpected behavior.

Data from production incident reports in IoT projects consistently shows that a significant portion of runtime errors stem from payload deserialization failures and topic parsing bugs, rather than network issues.

WOW Moment: Key Findings

Implementing a typed abstraction layer over mqtt.js fundamentally shifts the error surface. By integrating schema validation (via Zod) and compile-time type inference, teams can eliminate entire categories of bugs before deployment.

Approach	Type Safety	Runtime Validation	Param Extraction	Error Detection
Manual mqtt.js	None (Manual Casts)	None	String Splitting	Runtime
Typed Abstraction (topiq)	Full Inference	Zod Schema	Literal Types	Compile + Runtime

This comparison highlights that a typed approach not only catches structural mismatches at compile time but also enforces data integrity at runtime. The ability to extract topic parameters as typed objects without manual parsing reduces boilerplate and eliminates index-out-of-bounds errors. Furthermore, the distinction between subscription topics (which allow wildcards) and concrete topics (required for publishing) is enforced at the type level, preventing invalid operations.

Core Solution

The solution involves wrapping the MQTT client with a topic-centric API that binds schemas to topic patterns. This enables full type inference for payloads and parameters while providing runtime validation.

Step 1: Define Schemas and Topics

Start by defining Zod schemas for your payloads. Then, create topic definitions that bind the schema to a topic pattern. Dynamic segments are marked with a colon (e.g., :vehicleId).

import { topic, topiq } from 'topiq';
import { z } from 'zod';

// Define the payload schema
const telemetrySchema = z.object({
  speed: z.number().min(0),
  fuelLevel: z.number().min(0).max(100),
  timestamp: z.number(),
});

// Bind schema to topic pattern
// :vehicleId is a dynamic parameter
const telemetryTopic = topic('fleet/:vehicleId/telemetry', telemetrySchema);

Step 2: Initialize the Client

Create the client instance, passing the connection options and the topic definitions. The client uses these definitions to infer types for all subsequent operations.

const fleetClient = topiq(
  { 
    host: 'mqtt.broker.io', 
    tls: true, // Automatically uses mqtts:// on port 8883
  },
  { 
    topics: { 
      telemetry: telemetryTopic 
    } 
  }
);

Step 3: Subscribe with Type Safety

Use the .on() method to subscribe. The callback receives the validated payload and a context object containing the raw topic and extracted parameters. The method returns an unsubscribe function for cleanup.

const unsubscribe = fleetClient.on(telemetryTopic, (payload, context) => {
  // payload: { speed: number; fuelLevel: number; timestamp: number }
  // context.params.vehicleId: string
  // context.topic: string (e.g., "fleet/truck-42/telemetry")
  
  console.log(`Vehicle ${context.params.vehicleId} speed: ${payload.speed}`);
  
  // Business logic here
  updateDashboard(context.params.vehicleId, payload);
});

// Cleanup when component unmounts or service stops
// unsubscribe();

Step 4: Publish with Concrete Topics

To publish, use the .build() method on the topic definition. This generates a concrete topic string with typed parameters. The type system ensures that wildcards cannot be used for publishing.

// build() returns a literal type: "fleet/truck-42/telemetry"
const concreteTopic = telemetryTopic.build({ vehicleId: 'truck-42' });

fleetClient.emit(concreteTopic, {
  speed: 65,
  fuelLevel: 80,
  timestamp: Date.now(),
});

// TypeScript prevents wildcards in emit
// fleetClient.emit("fleet/+/telemetry", data); // ❌ Compile error

Step 5: Stream Processing

For continuous data processing, use the .stream() method. This returns an async iterable that respects AbortSignal for cancellation, eliminating the need for manual event listener management.

const controller = new AbortController();

(async () => {
  for await (const msg of fleetClient.stream(telemetryTopic, controller.signal)) {
    // msg.data: Validated payload
    // msg.params: Extracted parameters
    console.log(`[${msg.params.vehicleId}] ${msg.data.speed} km/h`);
  }
})();

// Stop streaming from anywhere
// controller.abort();

Architecture Rationale

• Zod Integration: Zod provides runtime validation that matches the TypeScript types. If a payload fails validation, the message is silently discarded, preventing crashes. This is critical in IoT where devices may send malformed data.
• Literal Types: The build() method returns precise literal types (e.g., "fleet/truck-42/telemetry") rather than generic string. This allows TypeScript to verify that emitted topics match the expected pattern exactly.
• Tail-Recursive Conditional Types: Under the hood, the library uses tail-recursive TypeScript utilities to extract parameters from arbitrarily deep topic paths. This avoids hitting TypeScript's recursion limits, which can occur with complex topic hierarchies.
• Concrete vs. Topic Types: The type system distinguishes between Topic (allows wildcards, used for subscribe/stream) and ConcreteTopic (no wildcards, used for emit). This enforces MQTT protocol rules at compile time.

Pitfall Guide

1. Wildcard Publishing

   • Explanation: Attempting to publish to a topic with wildcards (e.g., fleet/+/telemetry). MQTT brokers reject this, and the type system should prevent it.
   • Fix: Always use topic.build() or a concrete string literal for emit(). Ensure the topic string contains no + or # characters.

2. Schema Drift

   • Explanation: A device sends a payload with a new field not defined in the schema. By default, Zod may reject this, causing data loss.
   • Fix: Use z.object({ ... }).passthrough() to allow unknown fields, or implement schema versioning. Monitor validation failures to detect drift early.

3. Resource Leaks

   • Explanation: Forgetting to call the unsubscribe function returned by .on(), leading to memory leaks or duplicate handlers.
   • Fix: Store the unsubscribe function and call it in cleanup routines (e.g., useEffect cleanup in React, or finally blocks).

4. Validation Overhead

   • Explanation: Zod validation adds runtime cost. In high-throughput scenarios, this could impact performance.
   • Fix: Benchmark validation overhead. Use safeParse if you need to handle errors manually, or pre-compile schemas. For extreme performance needs, consider stripping validation in production builds, though this is generally discouraged.

5. TLS Misconfiguration

   • Explanation: Using tls: true when the broker requires mutual TLS (mTLS) with client certificates.
   • Fix: Provide explicit certificate paths in the TLS configuration if mTLS is required.

   topiq({
     host: 'broker.io',
     tls: {
       ca: fs.readFileSync('ca.crt'),
       cert: fs.readFileSync('client.crt'),
       key: fs.readFileSync('client.key'),
     },
   }, { topics });
   

6. Ignoring Validation Errors

   • Explanation: Silently discarding invalid messages without logging can mask device issues.
   • Fix: Implement a global error handler or logging mechanism for validation failures to track device health and schema compliance.

7. Type Recursion Limits

   • Explanation: Extremely deep topic paths might hit TypeScript recursion limits in older versions or complex setups.
   • Fix: The library uses tail-recursive types to mitigate this, but keep topic hierarchies reasonably flat. If issues arise, simplify topic structures.

Production Bundle

Action Checklist

• Define Zod schemas for all MQTT topics to enforce payload structure.
• Use topic() to bind schemas to topic patterns with dynamic parameters.
• Initialize the topiq client with connection options and topic definitions.
• Subscribe using .on() and handle the unsubscribe function for cleanup.
• Publish using .emit() with topics generated via .build() to ensure concreteness.
• Use .stream() with AbortController for async iterable processing.
• Configure TLS appropriately (tls: true or explicit certificates).
• Implement logging for validation failures to monitor device behavior.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High Throughput Streaming	.stream() with AbortSignal	Efficient async iteration, easy cancellation	Low
Event-Driven Handlers	.on() callback	Simple integration, automatic cleanup function	Low
Strict Security Requirements	mTLS with certificates	Mutual authentication, enhanced security	High setup
Rapid Prototyping	tls: true flag	Quick encryption without cert management	Low
Dynamic Topic Patterns	topic.build()	Type-safe concrete topic generation	None

Configuration Template

import { topic, topiq } from 'topiq';
import { z } from 'zod';
import fs from 'fs';

// Schemas
const alertSchema = z.object({
  code: z.string(),
  severity: z.enum(['low', 'medium', 'high']),
  message: z.string(),
});

const statusSchema = z.object({
  online: z.boolean(),
  lastSeen: z.number(),
});

// Topics
const alertTopic = topic('fleet/:vehicleId/alert', alertSchema);
const statusTopic = topic('fleet/:vehicleId/status', statusSchema);

// Client
const client = topiq(
  {
    host: process.env.MQTT_HOST!,
    tls: {
      ca: fs.readFileSync(process.env.CA_CERT_PATH!),
      cert: fs.readFileSync(process.env.CLIENT_CERT_PATH!),
      key: fs.readFileSync(process.env.CLIENT_KEY_PATH!),
    },
  },
  {
    topics: {
      alert: alertTopic,
      status: statusTopic,
    },
  }
);

// Handlers
client.on(alertTopic, (data, ctx) => {
  console.warn(`ALERT [${ctx.params.vehicleId}]: ${data.message}`);
});

client.on(statusTopic, (data, ctx) => {
  console.log(`STATUS [${ctx.params.vehicleId}]: ${data.online ? 'Online' : 'Offline'}`);
});

// Emit
client.emit(alertTopic.build({ vehicleId: 'van-12' }), {
  code: 'OIL_LOW',
  severity: 'high',
  message: 'Oil pressure critical',
});

Quick Start Guide

1. Install Dependencies:

   npm install topiq zod
   

2. Define Schema:

   const mySchema = z.object({ value: z.number() });
   

3. Create Topic:

   const myTopic = topic('my/path/:id', mySchema);
   

4. Init Client:

   const c = topiq({ host: 'broker' }, { topics: { myTopic } });
   

5. Subscribe/Emit:

   c.on(myTopic, (data, ctx) => console.log(ctx.params.id, data.value));
   c.emit(myTopic.build({ id: '1' }), { value: 42 });