# nodeledge-mqtt-listener A Portainer / docker compose stack that: 1. **mqtt-listener** — subscribes to MQTT topics on the form `//`, 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) ```bash 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: ```bash pip install -r tools/requirements.txt python tools/publish_test_messages.py --broker localhost:1883 ``` Fetch data from the API: ```bash 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: ```bash # 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>": ""}` — 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: ```json { "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" } ] } ```