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. ui — a Vue management frontend (own nginx container) that proxies /api to the api: devices & settings, attribute timelines, tokens, message browsing.
4. db — PostgreSQL 16 storing the messages (JSONB payloads) and the config tables (app_tokens, imei_map, imei_attributes).

┌─────────────┐   /xxx/IMEI      ┌───────────────┐   INSERT    ┌────┐
│ MQTT broker  │ ──────────────► │ mqtt-listener │ ──────────► │ db │
└─────────────┘   JSON payloads  └───────────────┘             └────┘
                                     injects imei + device       ▲
┌─────────────┐  /api (nginx)    ┌───────────────┐   SELECT      │
│ Browser (ui) │ ──────────────► │     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 + imei_attributes on first db start
listener/              mqtt -> db service (paho-mqtt, psycopg)
api/                   FastAPI service (uvicorn)
frontend/              Vue management UI (own nginx container, proxies /api)
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, ui and an anonymous Mosquitto broker (ports 1883 → host 2883, API 18080 → host 18080, UI 18090 → host 18090). 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}

Open the management UI at http://localhost:18090 and paste the bootstrap token when prompted (it is kept in the browser's localStorage; a real login flow will replace this later). Everything the UI does goes through the same authenticated JSON API — there is no separate backend.

Frontend development

The UI is a plain Vue 3 + TypeScript + Vite app in frontend/, served by its own nginx container (it is NOT bundled into the api image). For hot-reload development against a running api:

cd frontend
npm install
npm run dev        # serves the SPA on :5173, proxying /api to localhost:18080

npm run build type-checks (vue-tsc) and builds to frontend/dist. The container build does exactly this — see frontend/README.md.

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 the UI (Tokens screen) or 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, imei_map and imei_attributes.
6. Verify: Stacks → your stack → check that db, mqtt-listener, api and ui are all healthy (green). The UI is reachable on the UI_HOST_PORT (default 18090).

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 UI talks to the exact same endpoints documented below — everything it sends (including the token prompt) is plain X-App-Token requests against /api/..., so this reference applies to both human and machine clients. When running the ui container the API is reachable under the /api prefix (e.g. http://localhost:18090/api/messages); direct api access stays on its own port (default 18080).

• 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 is layered:

• App tokens (X-App-Token header or app_token query param) — for machine clients (scripts, integrations). Unrestricted access; SHA-256 digests live in app_tokens.
• User sessions — for humans via the management UI. POST /auth/login (username/password) returns an opaque Bearer token; only its SHA-256 digest is stored in sessions (30-day server-side expiry). Roles gate writes:

Capability	user	superuser	admin	app token
Read messages/devices/attributes/users	✓	✓	✓	✓
Write devices, settings, attribute timelines, configs, tokens	— 403	✓	✓	✓
Add users	— 403	✓ — only role user	✓ any role	✓
Modify/delete users	— 403	✓ — only user-role accounts	✓ (never the last admin)	✓

The first admin is seeded from BOOTSTRAP_ADMIN_USER/BOOTSTRAP_ADMIN_PASSWORD on the first login attempt against an empty users table (same bootstrap pattern as BOOTSTRAP_APP_TOKEN). Sessions resolve the user's CURRENT role on every request, so role changes apply immediately. Login is throttled: 5 failed attempts per username → 15-minute lockout (429). Missing/invalid credentials get 401; valid credentials but insufficient role get 403.

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. Each item also carries an attributes sibling field with the device's time-varying attributes resolved as of the message's device time (payload_time, received_at as the fallback) — apply the offsets to the payload values yourself (e.g. payload[k] + attributes[k]). 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 /devices/{imei}/attributes	token	Time-varying attributes (e.g. calibration offsets) resolved as of a time: ?at=ISO8601 (default now) returns {key: value} — for each key the version with the greatest effective_from <= at. Keys whose first version lies after at are absent
GET /devices/{imei}/attributes/history	token	The full timeline per key, oldest version first: {key: [{value, effective_from, note}, ...]}
PUT/POST /devices/{imei}/attributes	token	Append one version to a key's timeline. Body: {"key": ..., "value": ..., "effective_from": ISO8601?, "note"?}; effective_from defaults to now, backdating allowed (a sensor swap discovered after the fact becomes a backdated row). Re-using an exact existing (key, effective_from) edits that version in place
DELETE /devices/{imei}/attributes/{key}	token	Delete only the latest version of a key (typo/undo); earlier history stays immutable and the previous value is back in effect. Unknown key → 404
POST /devices/{imei}/rename	superuser+	Backfill-rename the device name on stored messages (both the device column and the payload key): {"device": "New", "from"?: ISO8601, "to"?: ISO8601, "dry_run"?}. Bounds are inclusive on received_at; an absent/null bound means no framing on that end. dry_run: true returns the affected count without writing. No undo — the IMEI is the identity, names are labels
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)
POST /auth/login	none	{"username", "password"} → {"token", "username", "role"}. Opaque Bearer token for the session; 5 failed attempts lock the username for 15 min (429)
GET /auth/me	session or app token	{"username", "role"} (both null for app tokens)
POST /auth/logout	session or app token	Revokes the Bearer session
GET /users	any	All users as [{username, role}] (passwords are unrecoverable)
PUT /users/{username}	superuser+	Create/reset a user: {"password", "role"?} (role default user). Superusers may only assign role user (403 otherwise); admins any role
DELETE /users/{username}	superuser+	Delete a user (their sessions are revoked via cascade). Superusers only user-role accounts; admins any except the last admin

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.

Time-varying attributes (offsets): values that may change over time — calibration offsets, mounting heights, anything recalibrated — live in the imei_attributes table as an append-only timeline (one row per version; the value valid at time T is the row with the greatest effective_from <= T). They deliberately do NOT go into imei_map.meta (which is current-only): a message must be corrected with the value that was in effect when it was measured. GET /messages resolves each message's attributes against the device time (payload_time, received_at as the fallback) via a lateral join and returns them as a sibling attributes field per item — not merged into payload — so consumers apply the corrections themselves and cannot double-apply or retro-bake them. Example: record an offset change discovered after a sensor swap by backdating it, and every message since the swap re-resolves correctly:

curl -X PUT -H "X-App-Token: $TOKEN" -H "Content-Type: application/json" \
    -d '{"key": "temperature_offset", "value": 2.0, "effective_from": "2026-08-15T00:00:00Z", "note": "after sensor swap"}' \
    http://localhost:18080/devices/350123456789012/attributes

The listener never reads imei_attributes; attributes are a read-time correction, and rows may exist for IMEIs without a settings row (no FK, like mqtt_messages).

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

TODO

• Add possability to rename ingested device names in message payloads, e g based on the imei. Must be possible to perform between datetimes. If either field(from or to) is unset, no framing is performed in that end. (→ POST /devices/{imei}/rename, see the endpoint table)
• Separate payload relative settings from auth settings. Sub-sections in menu? (→ sidebar grouped into Data / Access sections)
• Hide features from users not able to use them based on authorization. (→ auth-management links hidden for read-only user role; route guard blocks direct URL access; the API enforces 403 server-side regardless)
• Fix dashboard view, message expanding outside table. Make it look like the table in messages view. (→ payload cell now uses the same collapsed <details> + pretty-printed JSON + attributes block as the Messages view)
• Update tables displaying messages so they also contain a column with the imei, and not only device.
• Add possbility to sort and filter on table headers.
• Remove clickable device links in tables for users that is not allowed to see the device view, or does not have enought authorization to be able to change anything. CLicking as a user ends up on a page with only some text.
• Change datetime localization. How is it handled now, from browser or static? I'm in sweden and not following swedish rules on layout.
• Make it possible to manipulate a messages payload.
• Make it possible, for superusers and admins, to manually inject new messages using json from an "insert message" view. It should allow array of json objects as well as a single message json object.
• Change SESSION_TTL_DAYS to SESSION_TTL_MINUTES. Make SESSION_TTL_MINUTES being read from environmental variable. Default to 5 minutes.
• Add user information. Name, mail address
• Add audit logging. Audit log entries will contain actor(userid+username), timestamp, action type (login, database change etc.), target outcome(successful, failed, or generating a warning), source information (ip address, device ID, and/or location from which the action orginated) and before/after state.
• Add view for viewing audit log