nodeledge-mqtt-listener

A Portainer / docker compose stack that:

mqtt-listener — subscribes to MQTT topics on the form /<prefix>/<imei>, injects imei and a device name into each JSON object of the payload (a single object or an array of objects), and stores every object as its own row in PostgreSQL.
api — a FastAPI read API over the database, protected by static app tokens.
db — PostgreSQL 16 storing the messages (JSONB payloads).

┌─────────────┐   /xxx/IMEI      ┌───────────────┐   INSERT    ┌────┐
│ MQTT broker  │ ──────────────► │ mqtt-listener │ ──────────► │ db │
└─────────────┘   JSON payloads  └───────────────┘             └────┘
                                     injects imei + device       ▲
┌─────────────┐   GET /messages   ┌───────────────┐   SELECT      │
│ API clients  │ ──────────────► │     api       │ ──────────────┘
└─────────────┘   X-App-Token    └───────────────┘

Layout

docker-compose.yml     stack definition (paste into Portainer)
.env.example           environment variables (POSTGRES_PASSWORD etc.)
init/01_schema.sql     creates the mqtt_messages table on first db start
configs/listener/imei_map.json        IMEI -> device name mapping
configs/api/app_tokens.json          app tokens for the API
listener/              mqtt -> db service (paho-mqtt, psycopg)
api/                   FastAPI service (uvicorn)
mosquitto/             config for the optional test broker (profile "test")
tools/                 host-side test scripts and pytest suite

Quick start (local, with the test broker)

cp .env.example .env        # then edit POSTGRES_PASSWORD
docker compose --profile test up -d --build

This starts db, mqtt-listener, api and an anonymous Mosquitto broker (ports 1883 → host 2883, API 18080 → host 18080). Publish some messages:

pip install -r tools/requirements.txt
python tools/publish_test_messages.py --broker localhost:1883

Fetch data from the API:

python tools/smoke_test_api.py --url http://localhost:18080 \
    --token replace-me-with-a-long-random-secret-1

Expected smoke test output:

PASS: missing token rejected with 401
PASS: wrong token rejected with 401
PASS: valid token accepted, N message(s) returned
PASS: count endpoint returned {'count': N}

Deploying in Portainer

Create a stack from Stacks → Add stack, choose "Web editor", paste the contents of docker-compose.yml.
Under Environment variables, add at minimum:
- POSTGRES_PASSWORD — required (see the password note in .env.example).
- MQTT_BROKER_HOST — your real broker hostname/IP (not the test broker).
- MQTT_USERNAME / MQTT_PASSWORD — broker credentials, if required.
- Optional: MQTT_TOPIC_FILTER, MQTT_QOS, API_MAX_LIMIT, LOG_LEVEL, ports.
Because the stack references local files (init/01_schema.sql, configs/..., Dockerfiles), the stack files must be present on the Docker host. Two options:
- Recommended: put the repository in a folder on the host (e.g. /opt/nodeledge) and use Stacks → Add stack → Repository pointing at your git repo, or
- Upload the files via "Custom template" including all files.
Secrets: edit configs/api/app_tokens.json and configs/listener/imei_map.json on the host to hold your real values before deploying (the defaults are placeholders).
Deploy the stack. The db service runs init/01_schema.sql on its first start (an empty volume), creating the mqtt_messages table.
Verify: Stacks → your stack → check that db, mqtt-listener and api are all healthy (green).

The broker service only runs when the compose profile test is active (Portainer: enable the test profile in the stack settings), so production deploys never spin up an extra broker.

How the IMEI injection works

The IMEI is the last segment of the topic, e.g. /vivab/350123456789012 → 350123456789012.
It must be exactly 15 digits; anything else (letters, separators, other lengths) is skipped.
It looks the IMEI up in configs/listener/imei_map.json; on a match the device name is injected, otherwise the IMEI itself is used as the device name.
The injected keys are imei and device (configurable via INJECT_IMEI_KEY / INJECT_DEVICE_KEY). Any existing values in the payload are overwritten — the topic is authoritative.
A device timestamp is extracted at ingest time from the first parseable payload key in PAYLOAD_TIME_KEYS (default timestamp,time,ts; ISO 8601 strings, epoch seconds and epoch milliseconds all accepted) and stored in the payload_time column. A missing or unparseable timestamp does not skip the message — payload_time is simply NULL and received_at (server receipt time) remains authoritative.
The payload can be a single JSON object or a JSON array of objects. Each object in an array is injected and stored as its own database row (each with its own payload_time); array elements that are not objects are skipped.
Messages whose payload contains no JSON object at all (invalid JSON, a scalar, an array without objects), or whose last topic segment is not a 15-digit IMEI, are skipped entirely (with a warning log).

API reference

Authentication: pass one of the configured app tokens either as the X-App-Token header or as the app_token query parameter. Missing/invalid tokens get 401.

Endpoint	Auth	Description
`GET /health`	none	Liveness probe (public)
`GET /messages/count`	token	Total number of stored messages
`GET /messages`	token	List messages, newest first. Query parameters: `imei`, `device` (exact-match filters), `from`/`to` (ISO 8601 range on `received_at` — server receipt time; inclusive; naive values treated as UTC), `payload_from`/`payload_to` (same but on `payload_time` — the device timestamp extracted from the JSON; messages without one are excluded; ordering switches to `payload_time DESC`), `limit` (default 100, capped at `API_MAX_LIMIT`), `offset` for paging. Reversed ranges return 422
`GET /configs/{name}`	token	Read a config file's content as JSON. `name`: `app_tokens` or `imei_map`
`PUT/POST /configs/{name}`	token	Replace a config file's content. Body must be a JSON object; per-config schema is validated (imei_map keys must be 15-digit IMEIs; app_tokens must keep the `{"tokens": {name: secret}}` shape). Invalid payloads → 422, unknown names → 404. Writes are atomic. Updating `app_tokens` takes effect immediately; updating `imei_map` is picked up by the listener within ~5 seconds
`GET /docs`, `GET /redoc`	none	OpenAPI docs (public)

Example — everything device Water tank sensor B reported on August 31st, by device clock:

GET /messages?device=Water%20tank%20sensor%20B&payload_from=2026-08-31T00:00:00Z&payload_to=2026-08-31T23:59:59Z

Managing config files through the API

The two JSON config files can be read and replaced over HTTP (authenticated like all other endpoints). This is how to add a device or rotate a token without touching the host filesystem:

# read the current IMEI map
TOKEN='replace-me-with-a-long-random-secret-1'
curl -H "X-App-Token: $TOKEN" http://localhost:18080/configs/imei_map

# replace it (JSON object only; keys must be 15-digit IMEIs)
curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \
  -d '{"350123456789012": "Greenhouse sensor A", "490154203237518": "Water tank sensor B"}' \
  http://localhost:18080/configs/imei_map

app_tokens: {"tokens": {"app-name": "secret"}} — after an update the API re-authenticates with the new tokens immediately.
imei_map: {"<15-digit-IMEI>": "<device name>"} — the listener picks the change up within ~5 seconds (periodic reload); a temporarily invalid file on disk never breaks ingestion (the last good mapping stays in effect).

Example response:

{
  "limit": 100,
  "offset": 0,
  "from": null,
  "to": null,
  "payload_from": "2026-08-31T00:00:00+00:00",
  "payload_to": "2026-08-31T23:59:59+00:00",
  "items": [
    {
      "id": 1,
      "received_at": "2026-08-31T10:00:00.123456+00:00",
      "topic": "/vivab/350123456789012",
      "imei": "350123456789012",
      "device": "Greenhouse sensor A",
      "payload": {
        "temperature": 21.5,
        "timestamp": "2026-08-31T09:59:58+00:00",
        "imei": "350123456789012",
        "device": "Greenhouse sensor A"
      },
      "payload_time": "2026-08-31T09:59:58+00:00"
    }
  ]
}

8.8 KiB Raw Blame History