nodeledge-mqtt-listener

A Portainer / docker compose stack that:

1. 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.
2. api — a FastAPI read API over the database, protected by static app tokens.
3. 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

1. Create a stack from Stacks → Add stack, choose "Web editor", paste the contents of docker-compose.yml.
2. 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.
3. 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.
4. 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).
5. Deploy the stack. The db service runs init/01_schema.sql on its first start (an empty volume), creating the mqtt_messages table.
6. 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"
    }
  ]
}