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; on a match the device name is injected, otherwise the IMEI itself is used as the device name. The listener re-reads the table every 5 seconds, so API edits take effect without a restart.
• 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 as JSON. name: app_tokens (secrets are not recoverable — app names with null placeholders) or imei_map (the full mapping)
PUT/POST /configs/{name}	token	Replace a config in the database. 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 transactional (delete + insert). app_tokens changes take effect immediately; imei_map changes are picked up by the listener within ~5 seconds
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) and can be read and replaced over HTTP (authenticated like all other endpoints). This is how to add a device or rotate a token without direct database access:

# read the current IMEI map
TOKEN='your-bootstrap-or-app-token'
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"}} — secrets are stored as SHA-256 digests (never plaintext) and cannot be read back; GET /configs/app_tokens shows only the app names. After an update the API re-authenticates with the new tokens immediately.
• imei_map: {"<15-digit-IMEI>": "<device name>"} — the listener polls the table every ~5 seconds; a temporarily unreachable database never breaks ingestion (the last good mapping stays in effect).

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:

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