Store MQTT Messages in MongoDB with Node.js (Time-Series Collections)

To store MQTT messages in MongoDB, run a Node.js consumer that subscribes with MQTT.js, converts each JSON payload into a document and writes batches with insertMany into a time-series collection. MongoDB then groups readings into compressed buckets per device and can delete old data automatically.

This tutorial builds that consumer step by step: collection setup, a batching writer with a bounded buffer, a persistent MQTT session for restarts, and the queries and tests that prove every message landed.

Why a time-series collection

MQTT telemetry has a fixed shape: a timestamp, something that identifies the source, and a few measurements. MongoDB time-series collections (MongoDB 5.0+) are designed for exactly that.

The time-series options you choose at creation:

• timeField — the Date of the reading (ts).
• metaField — what identifies the series (meta.deviceId, meta.topic). Readings with the same meta value are stored together, so keep it stable per device.
• granularity — "seconds" for readings every few seconds; "minutes" or "hours" for slower devices.
• expireAfterSeconds — automatic deletion of old data.

Run MongoDB and create the collection

docker run -d --name mongo -p 27017:27017 mongo:7
npm install mqtt@5 mongodb@6

The consumer creates the collection on first start, so you do not need a separate migration step.

The consumer

import mqtt from "mqtt";
import { MongoClient } from "mongodb";

const MONGO_URL = process.env.MONGO_URL ?? "mongodb://localhost:27017";
const TOPIC = "testmqtt/mongo/+/telemetry";
const BATCH_SIZE = 500;
const FLUSH_MS = 1000;
const MAX_BUFFER = 50_000;

// 1. MongoDB: create the time-series collection once
const mongo = new MongoClient(MONGO_URL);
await mongo.connect();
const db = mongo.db("iot");
if (!(await db.listCollections({ name: "telemetry" }).hasNext())) {
  await db.createCollection("telemetry", {
    timeseries: { timeField: "ts", metaField: "meta", granularity: "seconds" },
    expireAfterSeconds: 60 * 60 * 24 * 30, // keep 30 days
  });
}
const telemetry = db.collection("telemetry");

// 2. Batching writer
let buffer = [];
let flushing = false;

async function flush() {
  if (flushing || buffer.length === 0) return;
  flushing = true;
  const docs = buffer;
  buffer = [];
  try {
    const res = await telemetry.insertMany(docs, { ordered: false });
    console.log(`inserted ${res.insertedCount} docs`);
  } catch (err) {
    if (err.name === "MongoBulkWriteError") {
      console.error(`some documents rejected: ${err.message}`); // bad data: do not retry
    } else {
      console.error(`MongoDB unavailable, re-queueing ${docs.length} docs: ${err.message}`);
      buffer = docs.concat(buffer).slice(-MAX_BUFFER);
    }
  } finally {
    flushing = false;
  }
}
const timer = setInterval(flush, FLUSH_MS);

// 3. MQTT: persistent session, QoS 1
const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  clientId: "mongo-writer-01",
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
  clean: false,
  properties: { sessionExpiryInterval: 3600 },
  reconnectPeriod: 2000,
});

client.on("connect", async (connack) => {
  console.log(`connected, session present: ${connack.sessionPresent}`);
  await client.subscribeAsync(TOPIC, { qos: 1 });
});
client.on("error", (err) => console.error(`mqtt error: ${err.message}`));

client.on("message", (topic, payload) => {
  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    console.warn(`skipping non-JSON payload on ${topic}`);
    return;
  }
  if (typeof data !== "object" || data === null) return;

  const deviceId = topic.split("/")[2];
  const { ts, ...measurements } = data;
  buffer.push({
    ts: typeof ts === "number" ? new Date(ts) : new Date(),
    meta: { deviceId, topic },
    ...measurements,
  });
  if (buffer.length > MAX_BUFFER) buffer.shift(); // protect memory, drop oldest
  if (buffer.length >= BATCH_SIZE) flush();
});

// 4. Graceful shutdown: stop MQTT, then flush what is left
async function shutdown() {
  clearInterval(timer);
  await client.endAsync();
  while (flushing) await new Promise((r) => setTimeout(r, 50));
  await flush();
  await mongo.close();
  process.exit(0);
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

What each part does:

• insertMany with ordered: false sends one request per batch and keeps inserting the remaining documents if one is rejected. One insertOne per message is the most common reason these consumers fall behind.
• The time field is converted to a Date. Devices usually send milliseconds since epoch; a raw number in ts makes the insert fail.
• Error handling separates bad data from outages. A MongoBulkWriteError means MongoDB rejected documents — retrying will not help. Network or server-selection errors re-queue the batch for the next flush.
• MAX_BUFFER caps memory if MongoDB is down for a long time. Dropping the oldest data is a deliberate choice; for zero loss, rely on the broker session described below instead of a bigger buffer.

If you are new to MQTT.js, the JavaScript MQTT guide covers connection options and reconnects, and the code generator produces a starting client for your broker settings.

Restarts, sessions and duplicates

clean: false with sessionExpiryInterval and a fixed clientId makes the broker keep your subscription and queue QoS 1 messages while the consumer is offline — during a deploy, for example. On reconnect the log shows session present: true and the queued messages arrive. Every instance needs its own client ID; generate one with the client ID generator. More background in persistent sessions.

Shared public test endpoints such as broker.connectmq.com do not keep offline sessions, so verify restart behavior on your own private instance.

Two limits to know:

• MQTT.js acknowledges a QoS 1 message as soon as the message event has fired, before the batch is written. If the process crashes, buffered documents are lost. If you need acknowledge-after-write, MQTT.js lets you override client.handleMessage(packet, callback), which sends the PUBACK only after you call callback — but it processes messages one at a time, so it trades throughput for safety.
• Time-series collections do not support unique indexes, so you cannot deduplicate QoS 1 redeliveries with a unique key. If exact-once storage matters, write to a normal collection with a unique index on { "meta.deviceId": 1, seq: 1 } and ignore duplicate-key errors.

Test the pipeline

1. Start MongoDB and run node mqtt-to-mongo.mjs.
2. Subscribe to testmqtt/mongo/# in the online MQTT client to watch traffic, then publish a few readings:

for i in $(seq 1 30); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 -t testmqtt/mongo/sensor-0$((i % 3))/telemetry \
    -m "{\"ts\":$(date +%s000),\"seq\":$i,\"temperature\":$((20 + i % 4)).2,\"battery\":$((100 - i))}"
done

1. Query MongoDB:

use iot
db.telemetry.find().sort({ ts: -1 }).limit(3)

db.telemetry.aggregate([
  { $match: { ts: { $gte: new Date(Date.now() - 3600000) } } },
  { $group: {
      _id: { device: "$meta.deviceId", minute: { $dateTrunc: { date: "$ts", unit: "minute" } } },
      avgTemp: { $avg: "$temperature" },
      n: { $sum: 1 } } },
  { $sort: { "_id.minute": 1 } }
])

Run the shell with docker exec -it mongo mongosh. The counts per device should add up to 30. Then test failure modes: stop MongoDB with docker stop mongo, publish ten messages, start it again and check that the writer re-queued and inserted them.

Watch the MongoDB test topic in your browser

Troubleshooting

To run several writers, subscribe to $share/mongo-writers/testmqtt/mongo/+/telemetry with a distinct client ID per instance; the broker splits the stream between them (see shared subscriptions). Recent MongoDB versions create an index on the meta and time fields of new time-series collections automatically; add db.telemetry.createIndex({ "meta.deviceId": 1, ts: -1 }) yourself on older versions. If your team prefers SQL for analytics, compare this approach with MQTT to PostgreSQL/TimescaleDB.

MongoDB is a natural home for JSON telemetry once you respect three rules: a real Date in the time field, a stable meta per device, and batched writes. Add a persistent session on your own broker and a restart test, and the pipeline will survive deploys without anyone noticing.

Frequently asked questions