wappler-mqtt-server

A Server Connect (NodeJS) extension for Wappler that lets your server subscribe to MQTT topics in the background and publish messages, using raw MQTT over TCP/TLS (mqtt:// / mqtts://) - no WebSocket needed on the server side (unlike the browser).

Pairs well with the companion App Connect client-side extension (wappler-mqtt) if you built that too, but works completely independently.

The one thing to understand first

A normal Server Connect action runs once per API call and returns a response. Subscribing to MQTT is different - it needs a connection that stays open indefinitely, for as long as your Node server process is running, not just for one request.

This module handles that by:

1. MQTT Subscribe opens a connection and keeps it alive in memory for the life of the Node process (not tied to the API call that started it).
2. Every incoming message gets POSTed as JSON to a Webhook URL you configure - a completely normal Server Connect API endpoint you build yourself with standard steps (Condition, Database Insert, etc). This keeps "what happens on message" fully visual and debuggable in the normal flow editor, instead of hiding it inside custom code.

You need to trigger MQTT Subscribe once when your server starts, not on every request. See "Starting the subscription" below.

Installation

1. In Wappler, go to Project Settings > Extensions > Add Extension > Create New Extension and pick a folder, e.g. src/wappler-mqtt-server.
2. Copy these files into that folder, keeping the structure:

   src/wappler-mqtt-server/
     package.json
     server_connect/
       modules/
         mqtt.js
         mqtt.hjson

3. Open a terminal in Wappler, cd into that folder, and run npm install so the mqtt package gets installed (or let Wappler's dependency install prompt do it when it detects the new package.json).
4. Restart Wappler / reload the project. You should now see MQTT Subscribe, MQTT Unsubscribe, MQTT Publish, and MQTT Status as steps you can add in any Server Connect flow, grouped under MQTT.

Setting it up

1. Build the webhook endpoint first

Create a normal Server Connect API action, e.g. POST /api/mqtt-webhook. It will receive this JSON body on every incoming MQTT message:

{ "topic": "/state/gw/devices/2324248439506/telemetry/vehicle.mileage",
  "message": "90143.074",
  "payload": 90143.074,
  "deviceId": "8439506",
  "field": "vehicle_mileage" }

(payload is auto-parsed from JSON when the raw message is valid JSON, otherwise it's the same as message. deviceId/field are pre-parsed from the topic path server-side, using the Device ID Topic Segment setting on MQTT Subscribe - no string splitting needed in your flow.)

Table design: use one row per (device, field), not one column per field

A SQL table has fixed columns, so it can't dynamically write to a column named after whatever field just arrived - and a device like the one in your logs can report a dozen+ distinct fields (vehicle_mileage, position_latitude, gsm_signal_level, engine_ignition_status...), with more potentially showing up over time. Trying to give every field its own column means constantly changing your schema.

The standard fix for this - an EAV-style table (Entity-Attribute- Value): one row per (device, field) pair, holding whatever the current value is. It scales to any number of fields with zero schema changes.

CREATE TABLE device_telemetry (
  id INT AUTO_INCREMENT PRIMARY KEY,
  device_id VARCHAR(64) NOT NULL,
  field VARCHAR(128) NOT NULL,
  value VARCHAR(255) NULL,
  updated_at DATETIME NOT NULL,
  UNIQUE KEY device_field (device_id, field)
);

Then in your /api/mqtt-webhook endpoint:

1. Add a Database Upsert step on device_telemetry.
2. Match on: device_id = $_POST.deviceId AND field = $_POST.field (this is exactly what the UNIQUE KEY device_field above is for).
3. Set: value = $_POST.payload, updated_at = the current timestamp.

That's it - one single Upsert step handles every field for every device, forever, without ever needing to touch the schema again as new telemetry fields show up.

Displaying it as a flattened dashboard later: query device_telemetry grouped by device_id and either pivot it in SQL (MAX(CASE WHEN field = 'vehicle_mileage' THEN value END) AS mileage, one CASE per field you want as a column) or just loop over the rows per device in your app. Ask if you want help building that pivot query once you're at that stage - it's a separate, standard SQL step from the ingestion side above.

1b. Batching (recommended if you expect meaningful volume)

Calling the webhook once per MQTT message means one HTTP round trip + one DB write per message - fine at low volume, but it'll add up fast with many devices publishing many fields each. Batching buffers messages and sends them to the webhook together as a group, either once a Batch Size is reached or a Batch Interval elapses, whichever comes first - so you get far fewer, larger webhook calls instead of a constant stream of tiny ones.

On MQTT Subscribe, set:

• Batch Size: e.g. 500 - buffer up to 500 messages before sending
• Batch Interval (ms): e.g. 5000 - also send whatever's buffered every 5 seconds even if 500 hasn't been reached yet, so data during a quiet period doesn't sit around indefinitely

With either set above 0, your Webhook URL's request body changes shape - instead of one object, it becomes a JSON array of that same object shape:

[
  { "topic": "/.../vehicle.mileage", "message": "90143.074", "payload": 90143.074, "deviceId": "8439506", "field": "vehicle_mileage" },
  { "topic": "/.../position.latitude", "message": "53.484993", "payload": 53.484993, "deviceId": "8439506", "field": "position_latitude" },
  ...
]

Update your webhook endpoint to loop over this array with a Repeat / For Each step (bound to $_POST), running the same Database Upsert from section 1 inside the loop, once per item. You still get one row per (device, field) combination either way - batching only changes how many messages arrive per webhook call, not what gets saved.

1c. Group By Device (flattening within a batch)

A batch of 10 messages will often span several different devices - multiple fields for the same device arriving close together. Without grouping, your webhook still does one Upsert per field (10 upserts for 10 messages, even if only 3 distinct devices are involved). With Group By Device turned on, messages in the batch that share the same deviceId get merged into one combined object before the webhook is even called - so those same 10 messages become 3 grouped objects (one per device), each holding every field seen for that device within the batch:

[
  {
    "deviceId": "8225378",
    "fields": { "vehicle_mileage": 90143.074, "position_latitude": 53.484993, "position_longitude": -2.908047, "harsh_cornering_event": true },
    "messageCount": 4,
    "topics": [ "...vehicle.mileage", "...position.latitude", "...position.longitude", "...harsh.cornering.event" ],
    "lastReceivedAt": "2026-08-09T15:50:11.571Z"
  },
  { "deviceId": "8439506", "fields": { "...": "..." }, "messageCount": 3, "topics": [ "..." ], "lastReceivedAt": "..." }
]

Now your webhook's Repeat step loops over devices, not raw messages, and each iteration does one Upsert merging the whole fields object into that device's JSONB column in a single query - matching the recommended table from section 1:

INSERT INTO devices (device_id, telemetry, updated_at)
VALUES (:deviceId, :fields::jsonb, now())
ON CONFLICT (device_id) DO UPDATE
SET telemetry = devices.telemetry || :fields::jsonb,
    updated_at = now();

(Bind :fields to the loop item's fields object, cast to JSONB - Postgres's || merge operator handles multiple keys at once just as well as one, so this single query correctly merges every field from the batch into the device's row in one write instead of messageCount separate writes.)

Turn this on together with Batch Size for the biggest win: e.g. Batch Size 10 + Group By Device turns 10 individual field-writes into as few as 1 device-write (if all 10 happened to be the same device) up to 10 (if all 10 were different devices) - realistically somewhere in between, same as the worked example above.

Two more actions that come with batching:

• MQTT Flush - manually sends whatever's currently buffered right away, without waiting for the size/interval trigger. Handy for a "flush now" endpoint, or to call right before planned maintenance.
• MQTT Status now also returns pendingBatchCount - how many messages are currently buffered and not yet sent, useful for monitoring.

Safety nets built in: buffered-but-unsent messages are automatically flushed when you call MQTT Unsubscribe, when you call MQTT Subscribe again with the same Connection Name (replacing the connection), and on a best-effort basis if the Node process receives a graceful shutdown signal (SIGTERM/SIGINT, e.g. from a deploy). A hard crash or kill -9 can still lose whatever was buffered and not yet sent, same as any in-memory buffer - if you truly cannot afford to lose any message ever, keep Batch Interval low (so little sits buffered at once) or don't batch at all.

2. Starting the subscription

Since MQTT Subscribe needs to run once at server start rather than per request, the simplest reliable approach:

• Create a small dedicated API endpoint, e.g. GET /api/mqtt-start, containing only an MQTT Subscribe step with your Broker URL, Topics, Connection Name, and the Webhook URL from step 1.
• After each deploy/restart, hit that URL once (visit it in a browser, curl it, or point an uptime monitor / deploy hook at it).
• Re-running it is safe - MQTT Subscribe replaces any existing connection with the same Connection Name instead of creating a duplicate.

If you want it to start completely automatically on server boot with no manual step, that requires hooking into Express server startup directly (a lower-level "Extending Express" style customization) rather than a Server Connect module action - ask if you want help wiring that up too.

3. Multiple independent subscriptions

Use a different Connection Name for each distinct topic structure/webhook pairing, same pattern as running two MQTT Client components on the client side:

MQTT Subscribe: name="telemetry", topics="/state/gw/devices/+/telemetry/+", webhookurl=".../api/mqtt-webhook-telemetry"
MQTT Subscribe: name="tacho",     topics="/state/gw/devices/+/tacho/+",     webhookurl=".../api/mqtt-webhook-tacho"

Each runs its own persistent connection and posts to its own webhook.

4. Publishing

5. Seeing what values MQTT has actually received

Two ways to inspect real incoming values, from quickest-for-testing to what actually saves them to your database:

Quickest way to just look at the data - call MQTT Status (with its Output toggle on) from any API endpoint or even by just hitting it directly in a browser if it's a GET action. It returns:

{
  "name": "telemetry",
  "exists": true,
  "connected": true,
  "lastTopic": "/state/gw/devices/23234843923432506/telemetry/vehicle.mileage",
  "lastMessage": "90143.074",
  "lastPayload": 90143.074,
  "lastDeviceId": "8439506",
  "lastField": "vehicle_mileage",
  "lastReceivedAt": "2026-08-09T12:55:22.715Z"
}

This only ever shows the single most recent message on that connection - it's for confirming things are actually flowing and seeing the real shape of your data, not for building your permanent save logic on (it's a live snapshot, not a log).

The actual way to save every message to your database - this is what the Webhook URL (step 1 above) is for. Every single message, not just the last one, gets POSTed to that endpoint as { topic, message, payload }. Inside that endpoint you use $_POST.topic, $_POST.message, $_POST.payload exactly like any other API input - drag a Database Insert (or Upsert) step, and bind its column values to $_POST.payload (or specific keys within it, if payload is an object), the same way you'd bind data from any other form submission or API call. Add a Condition step first if you want to branch based on $_POST.topic (e.g. telemetry vs tacho, or per-field routing like the Flatten mode on the client-side extension).

6. Checking connection health

MQTT Status also works as a straightforward health check for a monitoring endpoint - just look at exists/connected and ignore the last* fields if you don't need them.

Notes

• Credentials passed to MQTT Subscribe live in your server-side code/ database, not exposed to the browser - this is the more secure place for broker credentials compared to the client-side extension.
• If your Node server restarts (crash, redeploy, scaling event), the in-memory connection is lost and needs to be re-triggered via your start endpoint - there's no automatic reconnect-on-boot without the Express-hook approach mentioned above.
• QoS 2 support depends on your broker.