# 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)

```bash
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:

```bash
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):

```bash
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:

```bash
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: 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. 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 |
| `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:

```bash
# 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:

```bash
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:

```bash
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:

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