nodeledge-mqtt-listener/README.md

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 app tokens stored in the database as SHA-256 digests.
3. db — PostgreSQL 16 storing the messages (JSONB payloads) and the config tables (app_tokens, imei_map).

┌─────────────┐   /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 mqtt_messages + app_tokens + imei_map on first db start
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). Set BOOTSTRAP_APP_TOKEN to a long random value in .env first — it seeds the initially empty app_tokens table on the first authenticated request. Then publish some messages:

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

Fetch data from the API (use the bootstrap token you set):

python tools/smoke_test_api.py --url http://localhost:18080 \
    --token "$BOOTSTRAP_APP_TOKEN"

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, 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. Tokens: set BOOTSTRAP_APP_TOKEN (a long random secret) in the stack's environment. On the first authenticated request after a fresh deploy, the api seeds its SHA-256 digest into the empty app_tokens table. Rotate to real per-app tokens via PUT /configs/app_tokens, then remove the bootstrap variable (it is ignored once the table has rows).
5. Deploy the stack. The db service runs init/01_schema.sql on its first start (an empty volume), creating mqtt_messages, app_tokens and imei_map.
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 the imei_map table and injects the device's settings document's device value as the device name; on no mapping the IMEI itself is used. The listener re-reads the table every 5 seconds, so API edits take effect without a restart.
• Each device's settings document (JSONB) can hold more than the name — e.g. {"device": "Greenhouse sensor A", "location": "Växthus 1"}. The listener only uses device at ingest; the API merges the whole current document into message payloads at read time (see below).
• 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. The device's current settings document (from imei_map) is merged into each item's payload; on duplicate keys the stored payload wins. 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 as JSON. name: app_tokens (secrets are not recoverable — app names with null placeholders) or imei_map (all device settings documents)
PUT/POST /configs/{name}	token	Bulk replace a config in the database (delete + insert, transactional). Body must be a JSON object; per-config schema is validated (imei_map keys must be 15-digit IMEIs, values are settings documents — flat "name" values are promoted to {"device": "name"}; app_tokens must keep the {"tokens": {name: secret}} shape). Invalid payloads → 422, unknown names → 404. app_tokens changes take effect immediately; imei_map changes are picked up by the listener within ~5 seconds
GET /devices	token	All devices, each as {"imei": ..., **settings}
GET /devices/{imei}	token	One device's settings document; unknown IMEI → 404, non-15-digit IMEI → 422
PUT /devices/{imei}	token	Upsert one device's settings document (create or replace). Body is the settings object (must have a non-empty device key; extra keys are arbitrary JSON); a body imei key is tolerated when it matches the URL. Returns the saved document, so a GET response can be edited and PUT straight back
DELETE /devices/{imei}	token	Remove one device (its messages stay); unknown IMEI → 404
GET /tokens	token	App names with a token (secrets are unrecoverable, shown as null)
PUT /tokens/{app}	token	Add or rotate ONE app token (single-row upsert — never touches the other apps' rows). Body: {"token": "secret"}. Takes effect immediately
DELETE /tokens/{app}	token	Remove one app token; unknown name → 404. Deleting the last token locks the API until a new one is added
GET /docs, GET /redoc	none	OpenAPI docs (public)

When the database is unreachable, authenticated endpoints return 503 (the API has no purpose without the db).

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 through the API

Both configs live in the database (app_tokens, imei_map tables). A management frontend should use the per-resource endpoints (/devices/{imei}, /tokens/{app}) — they save one row at a time, so two users editing different devices (or an admin rotating one token) never clobber each other. The whole-document PUT /configs/{name} endpoints remain for bulk imports:

# read the current device settings
TOKEN='your-bootstrap-or-app-token'
curl -H "X-App-Token: $TOKEN" http://localhost:18080/devices

# upsert ONE device with its settings document (the frontend way)
curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \
    -d '{"device": "Greenhouse sensor A", "location": "Växthus 1", "battery_warning": 20}' \
    http://localhost:18080/devices/350123456789012

# bulk import: replace ALL device settings at once (old flat values still accepted)
curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \
    -d '{"350123456789012": {"device": "Greenhouse sensor A", "location": "Växthus 1"}}' \
    http://localhost:18080/configs/imei_map

• app_tokens: "tokens": {"app-name": "secret"} — secrets are stored as SHA-256 digests (never plaintext) and cannot be read back; listing endpoints show only the app names. After an update the API re-authenticates with the new tokens immediately. For single-token rotation prefer PUT /tokens/{app}.
• imei_map: one settings document per IMEI; "device" is required, other keys are arbitrary JSON merged into message payloads at read time. The listener polls the table every ~5 seconds; a temporarily unreachable database never breaks ingestion (the last good mapping stays in effect).

Read-time settings merge: GET /messages joins imei_map and merges the device's current settings document into each item's payload (COALESCE(im.meta, '{}'::jsonb) || m.payload — the stored payload wins on duplicate keys). Historical messages therefore always reflect the latest device settings; the ingest-time imei/device snapshots in the stored payload and columns are unaffected.

Migrating an existing deployment from the JSON files

For a database that already has data (init scripts only run on an empty volume), apply the schema and migrate the JSON configs. The schema script is idempotent and migrates the old imei_map(device TEXT) layout in place: every existing row is converted to {"device": ...} and the legacy column is dropped:

docker compose exec db psql -U nodeledge -d nodeledge -f - < init/01_schema.sql
python tools/migrate_configs_to_db.py --dsn postgresql://nodeledge:PASSWORD@localhost:5432/nodeledge \
    --tokens configs/api/app_tokens.json --imei-map configs/listener/imei_map.json

After migrating, the JSON files are no longer read by the stack and can be deleted.

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",
        "location": "Växthus 1"
      },
      "payload_time": "2026-08-31T09:59:58+00:00"
    }
  ]
}