MQTT to MongoDB: Save Device Messages with Node.js and Time-Series Collections

Saving MQTT to MongoDB is a natural fit when your devices send JSON: documents go in as they arrive, and time-series collections keep storage compact and queries fast. In this tutorial you will build a Node.js subscriber with MQTT.js that batches device messages from a cloud broker into a MongoDB time-series collection, with automatic expiry and ready-made aggregation queries.

How the Pipeline Fits Together

devices --MQTT--> ConnectMQ broker --MQTT--> Node.js subscriber --insertMany--> MongoDB time-series collection

The broker's job is routing messages; it does not write to databases. The subscriber is a regular MQTT client that you run next to your database, which keeps the design simple and portable. Any fully managed MQTT service works on the left side; this tutorial uses ConnectMQ.

Why Time-Series Collections

A regular collection stores each reading as its own document with its own overhead. A time-series collection groups readings internally by a metadata field and time window, so IoT data takes far less disk space and range queries over time scan less. You also get expireAfterSeconds for automatic deletion of old data. The trade-offs: documents are effectively append-only (updates and deletes are limited), and unique indexes are not supported, so deduplication must happen before insert if you need it.

Step 1: Start MongoDB

docker run -d --name mongo -p 27017:27017 \
  -v mongo-data:/data/db \
  mongo:7

Step 2: Create the Time-Series Collection

Device messages arrive on <prefix>/<site>/<deviceId>/telemetry with a payload such as:

{"ts": "2026-10-16T09:30:00.000Z", "temperature": 23.4, "humidity": 41, "fw": "1.4.2"}

The collection stores each message as one document: a ts time field, a meta object with values that identify the series (site, device), and the measurements. Create it once with a setup script.

setup.mjs:

import { MongoClient } from "mongodb";

const uri = process.env.MONGO_URI ?? "mongodb://localhost:27017";
const client = new MongoClient(uri);
await client.connect();
const db = client.db("iot");

try {
  await db.createCollection("telemetry", {
    timeseries: { timeField: "ts", metaField: "meta", granularity: "seconds" },
    expireAfterSeconds: 60 * 60 * 24 * 90, // keep 90 days
  });
  console.log("created time-series collection iot.telemetry");
} catch (err) {
  if (err.codeName === "NamespaceExists") console.log("collection already exists");
  else throw err;
}

await db.collection("telemetry").createIndex({ "meta.deviceId": 1, ts: -1 });
await client.close();

npm init -y
npm install mqtt mongodb
node setup.mjs

Choose granularity to match how often a single device reports: seconds for sub-minute intervals, minutes for readings every few minutes, hours for sparse data. A good metaField holds values that rarely change for a series; never put the measurements themselves in meta.

Step 3: The MQTT to MongoDB Subscriber

The subscriber keeps the MQTT callback cheap: it validates the message, shapes a document and pushes it into an in-memory buffer. A timer flushes the buffer with insertMany every second, or immediately once 500 documents are waiting.

subscriber.mjs:

import mqtt from "mqtt";
import { MongoClient } from "mongodb";
import { randomBytes } from "node:crypto";

const MQTT_URL = process.env.MQTT_URL ?? "mqtts://broker.connectmq.com:8883";
const PREFIX = process.env.MQTT_TOPIC_PREFIX ?? "connectmq/demo/mongo-9d42a7";
const CLIENT_ID = process.env.MQTT_CLIENT_ID ?? `mongo-sink-${randomBytes(4).toString("hex")}`;
const MONGO_URI = process.env.MONGO_URI ?? "mongodb://localhost:27017";

const BATCH_SIZE = 500;
const FLUSH_MS = 1000;
const MAX_BUFFER = 50_000;

const mongo = new MongoClient(MONGO_URI);
await mongo.connect();
const telemetry = mongo.db("iot").collection("telemetry");

let buffer = [];
let flushing = false;

function toDocument(topic, payload) {
  // <prefix>/<site>/<deviceId>/telemetry
  const parts = topic.split("/");
  if (parts.length < 3 || parts.at(-1) !== "telemetry") return null;
  const deviceId = parts.at(-2);
  const site = parts.at(-3);

  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    return null;
  }
  if (typeof data !== "object" || data === null || Array.isArray(data)) return null;

  const { ts, ...fields } = data;
  const when = ts ? new Date(ts) : new Date();
  if (Number.isNaN(when.getTime())) return null;

  const doc = { ts: when, meta: { site, deviceId }, receivedAt: new Date() };
  for (const [key, value] of Object.entries(fields)) {
    if (key.startsWith("$") || key.includes(".")) continue; // unsafe field names
    if (["number", "string", "boolean"].includes(typeof value)) doc[key] = value;
  }
  return doc;
}

async function flush() {
  if (flushing || buffer.length === 0) return;
  flushing = true;
  const batch = buffer;
  buffer = [];
  try {
    await telemetry.insertMany(batch, { ordered: false });
    console.log(`inserted ${batch.length} documents`);
  } catch (err) {
    console.error("insert failed, re-queueing batch:", err.message);
    buffer = batch.concat(buffer).slice(-MAX_BUFFER);
  } finally {
    flushing = false;
  }
}

const timer = setInterval(flush, FLUSH_MS);

const client = mqtt.connect(MQTT_URL, {
  clientId: CLIENT_ID,
  username: process.env.MQTT_USERNAME,
  password: process.env.MQTT_PASSWORD,
  protocolVersion: 5,
  clean: false,                                   // persistent session
  properties: { sessionExpiryInterval: 1800 },    // broker keeps it for 30 min
  keepalive: 30,
  reconnectPeriod: 2000,
});

client.on("connect", (connack) => {
  console.log(`connected as ${CLIENT_ID}, session present: ${connack.sessionPresent}`);
  client.subscribe(`${PREFIX}/+/+/telemetry`, { qos: 1 }, (err) => {
    if (err) console.error("subscribe failed:", err.message);
  });
});

client.on("message", (topic, payload) => {
  const doc = toDocument(topic, payload);
  if (!doc) return;
  if (buffer.length >= MAX_BUFFER) {
    console.error("buffer full, dropping message");
    return;
  }
  buffer.push(doc);
  if (buffer.length >= BATCH_SIZE) flush();
});

client.on("error", (err) => console.error("mqtt error:", err.message));
client.on("reconnect", () => console.log("reconnecting..."));

async function shutdown() {
  console.log("shutting down");
  clearInterval(timer);
  await client.endAsync();
  await flush();
  await mongo.close();
  process.exit(0);
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

Design Notes

• Batching with insertMany and ordered: false. One round trip per batch instead of per message, and a single bad document does not stop the rest of the batch.
• A bounded buffer. If MongoDB is unreachable, failed batches are re-queued, but the buffer is capped so memory cannot grow without limit.
• A persistent session. clean: false with a session expiry and a fixed client ID tells the broker to queue QoS 1 messages while the subscriber restarts. The public broker keeps sessions for up to 30 minutes. Use a stable MQTT_CLIENT_ID or you get a fresh, empty session every start. More in MQTT persistent sessions.
• Field sanitising. Keys starting with $ or containing dots are skipped, and only primitive values are stored, so a malicious or buggy device cannot inject odd structures into your documents.
• Device timestamps. Using the device's ts keeps buffered offline readings in the right place in time; receivedAt records when the server got them, which helps debug clock problems.

Shaping Documents: One per Message or One per Metric?

The subscriber stores one document per MQTT message, with every measurement as a top-level field. That shape matches how devices publish, keeps related values together, and makes queries such as "temperature and humidity at the same moment" trivial. It works well when each device type sends a stable set of fields.

The alternative is one document per metric: { ts, meta: { deviceId, metric }, value }. It suits fleets where devices report very different measurements, because every document has the same structure and you can index and aggregate on meta.metric generically. The cost is more documents and the need to regroup values that arrived together.

Pick one shape per collection and stick to it. Mixing shapes in a single time-series collection makes aggregations awkward and weakens the storage benefits. If you are unsure, start with one document per message; it is easier to split later than to merge.

Step 4: Test With Simulated Devices

Run the subscriber and publish test readings to the free public MQTT broker. The public area is visible to everyone, so stick to synthetic data:

node subscriber.mjs

# in another terminal
for i in $(seq 1 30); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i mongo-sim-01 -q 1 \
    -t "connectmq/demo/mongo-9d42a7/plant1/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Then look at the data in mongosh:

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

Step 5: Query the Data

Per-minute averages for one device over the last 24 hours:

db.telemetry.aggregate([
  { $match: { "meta.deviceId": "sensor-01", ts: { $gte: new Date(Date.now() - 864e5) } } },
  { $group: {
      _id: { $dateTrunc: { date: "$ts", unit: "minute" } },
      avgTemp: { $avg: "$temperature" },
      maxTemp: { $max: "$temperature" },
      samples: { $sum: 1 }
  } },
  { $sort: { _id: 1 } }
])

Latest reading per device, useful for a fleet overview page:

db.telemetry.aggregate([
  { $sort: { "meta.deviceId": 1, ts: -1 } },
  { $group: { _id: "$meta.deviceId", last: { $first: "$$ROOT" } } }
])

For long-term dashboards, schedule a job that writes hourly rollups into a separate collection with $merge, and let the raw collection expire after 90 days.

Production Checklist

1. Create a private broker in the ConnectMQ portal and set MQTT_USERNAME, MQTT_PASSWORD and MQTT_CLIENT_ID to the generated values; keep mqtts:// on port 8883.
2. Give the subscriber its own credentials. ConnectMQ binds client IDs to issued credentials, and two processes with one ID disconnect each other.
3. Run the subscriber under a process manager (systemd, PM2, Docker with restart policy).
4. Use a MongoDB user with insert rights on iot.telemetry only.
5. Accept that QoS 1 can deliver duplicates; if they matter, deduplicate in the subscriber (for example a short in-memory cache of message IDs) because time-series collections cannot enforce uniqueness.
6. To share the load across several subscribers, use MQTT 5 shared subscriptions ($share/mongo/<prefix>/+/+/telemetry), available on paid Serverless plans and above.
7. Watch insert rate and buffer size in logs or metrics.

Troubleshooting

For a SQL alternative with continuous aggregates, see MQTT to TimescaleDB; for lighter needs, saving MQTT data to SQLite with Python or MQTT to Google Sheets may be enough. The MQTT client libraries guide covers MQTT.js alongside other clients.

Get Started with ConnectMQ

Connect your MongoDB pipeline to a private broker in a few minutes. Sign up at app.connectmq.com for a free Serverless broker, no credit card required, export the generated credentials as environment variables, and use the portal's live topic explorer to check payloads before they reach your database.

Frequently Asked Questions

Can ConnectMQ store MQTT messages in MongoDB automatically?

No, ConnectMQ is the MQTT broker and has no built-in database integrations. Run a subscriber, such as the Node.js service in this tutorial or a Node-RED flow, that connects as an MQTT client and writes to MongoDB.

Should I use a time-series collection or a regular collection for MQTT data?

Use a time-series collection for append-only sensor readings: it stores them compactly and supports automatic expiry. Use a regular collection for data you update in place, such as device configuration or the latest state per device.

How do I avoid losing MQTT messages when MongoDB is down?

Buffer in the subscriber with a bounded queue and retry failed inserts, and use QoS 1 with a persistent MQTT session so the broker holds messages while the subscriber restarts.

What does the metaField do in a MongoDB time-series collection?

It identifies each series, for example site and device ID. MongoDB groups readings with the same metadata together internally, which is why measurements should never go into the metaField.