From f3fc394c0cddf497e24f455deff989160b8664a4 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Sun, 6 Sep 2026 22:35:24 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A7=20Add=20message=20delete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 9 +- api/app.py | 246 ++++++++++++++++++++ docker-compose.yml | 4 + frontend/src/components/RetentionDialog.vue | 213 +++++++++++++++++ frontend/src/views/Messages.vue | 78 ++++++- init/01_schema.sql | 35 +++ tools/test_listener.py | 215 +++++++++++++++++ 7 files changed, 797 insertions(+), 3 deletions(-) create mode 100644 frontend/src/components/RetentionDialog.vue diff --git a/README.md b/README.md index 72b97ab..97e4a7d 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,7 @@ Authentication is layered: | Read messages/devices/attributes/users | ✓ | ✓ | ✓ | ✓ | | Write devices, settings, attribute timelines, configs, tokens | — 403 | ✓ | ✓ | ✓ | | Import messages | — 403 | ✓ | ✓ | ✓ | +| Delete / restore messages | — 403 | ✓ | ✓ | ✓ | | Add users | — 403 | ✓ — only role `user` | ✓ any role | ✓ | | Modify/delete users | — 403 | ✓ — only `user`-role accounts | ✓ (never the last admin) | ✓ | @@ -128,6 +129,10 @@ The first admin is seeded from `BOOTSTRAP_ADMIN_USER`/`BOOTSTRAP_ADMIN_PASSWORD` | `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. The response carries a **filter-aware `total`** (the count over the same filters as `items`, so pagers can show "page X of Y" and jump to the last page) alongside `limit`/`offset`. Reversed ranges return 422 | | `GET /messages/export` | token | **Export selected messages as stored** (raw rows, no read-time settings merge, no resolved attributes). Query: repeatable `ids` (e.g. `?ids=1&ids=2`; at least one, capped at `API_MAX_LIMIT`). Returns a JSON array of `{id, received_at, topic, imei, device, payload, payload_time}` in the order the ids were given; unknown ids are skipped. Empty/oversized id lists → 422 | | `POST /messages/import` | superuser+ | **Import previously exported messages**: body is the JSON array from `GET /messages/export` (a single message object is accepted too). Rows are inserted as-is (original payload/timestamps/topic; `topic` defaults to `/import/`, absent `received_at` becomes now). The exported **`id` is ignored** — the database assigns fresh BIGSERIAL ids, so re-importing a file (or importing into the source database) never collides. Per-row validation → 422 with `row N` positions; max 10000 rows per request. Returns `{"status": "imported", "imported": n}` | +| `POST /messages/delete` | superuser+ | **Soft-delete the selected messages**: moves them (same ids) into retention where they stay restorable for `RETENTION_DAYS` (env, default 30), then are purged. Body `{"ids": [...]}` (at least one, ≤ 10000; unknown ids skipped). The session username is recorded as `deleted_by` (NULL for app tokens). Returns `{"status": "deleted", "deleted": n}` | +| `GET /messages/retention/count` | token | `{"count": n}` — retained messages right now (the Messages view banner); expired rows are purged first | +| `GET /messages/retention` | token | One page of retained messages (newest deletions first): export shape plus `deleted_at`, `deleted_by`, `expires_at`, with a `total` for the pager. `limit` (default 100, capped at `API_MAX_LIMIT`) and `offset` | +| `POST /messages/retention/restore` | superuser+ | Move retained messages back into `mqtt_messages` **with their original ids**. Body `{"ids": [...]}`; unknown ids (e.g. already purged) are skipped. Returns `{"status": "restored", "restored": n}` | | `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}` | @@ -257,9 +262,9 @@ Example response: - [ ] Change messages in tables to display information about the entire messege, not only the payload. - [X] Fix facet counts on Messages show 1 per device. - [x] Add ability to select how many rows to show in tables except the last messages table in the dashboard. Make it possible to select 10, 25, 50, 100, 200 or manual entry. Default to 25. (→ shared `Pager.vue` footer on Messages (server-side), Devices and Users (client-side via `useTable`); presets 10/25/50/100/200 + "custom…" manual entry clamped to `API_MAX_LIMIT`, default 25, persisted in localStorage; the dashboard's fixed 5-row list is untouched) -- [ ] Add operations button in messages view for import/export or deletion of selected messages. (→ export + import landed: per-row checkboxes + header select-all + "Export JSON" download; superuser+ "Import" dialog (paste/file, array or single object). Delete operation still open) +- [X] Add operations button in messages view for import/export or deletion of selected messages. (→ all three landed: per-row checkboxes + header select-all, "Export JSON" download, superuser+ "Import" dialog (paste/file, array or single object) and superuser+ **Delete** with confirm. The delete moves messages to retention — see the next line) - [x] Export option will export selected messages as they are stored in the database, without added parameters/features. (→ `GET /messages/export?ids=…` returns raw `mqtt_messages` rows — no settings merge, no attributes — and the UI downloads the JSON array; ids keep the caller's order) -- [ ] Delete option will, after warning and user confirmation, move the deleted messages to retention. Where to see and be able to select messages to restore from retention? Rention lasts for 30 days. Add some kind of notification in the message view if messages are retained. Make it possible to view retained messages. Make it possible to recover retained messages. +- [x] Delete option will, after warning and user confirmation, move the deleted messages to retention. Where to see and be able to select messages to restore from retention? Rention lasts for 30 days. Add some kind of notification in the message view if messages are retained. Make it possible to view retained messages in the messages view. Make it possible to recover retained messages. (→ **Delete** button in the selection toolbar (superuser+), confirm dialog, moves rows to `mqtt_messages_retention` keeping their ids; retention period = `RETENTION_DAYS` env (default 30), expired rows purged automatically. A **banner** in the Messages view shows the retained count and opens a **retention dialog**: the retained rows (deleted at/by, purge date, payload) with per-row selection, header select-all and **Restore selected** (superuser+). Viewers see the banner and the dialog but no delete/restore controls) - [ ] Add payload timestamp as column in messages view. The from/to field would be able to select if it is received or payload timestamp that is used for framing the loaded messages. - [x] Add table navigastion buttons for jumping to first or last page. Add manual navigation textfield/dropdown to all tables exept dashboard last messages table. (→ the shared `Pager.vue`: « first / ‹ prev / page-number textfield (clamped, Enter or blur to apply) / › next / » last buttons + "page X of Y" + "n–m of total"; works on Messages (server-side offset/limit) and the client-side tables (Devices/Users via `useTable` pagination). `GET /messages` now returns a filter-aware `total` to power it) - [ ] Add versioning of UI and API, show versions on bottom left. \ No newline at end of file diff --git a/api/app.py b/api/app.py index e60fd11..70c3c3c 100644 --- a/api/app.py +++ b/api/app.py @@ -754,6 +754,144 @@ class DeviceRenameStore: conn.close() +class MessageRetentionPolicy: + """Soft-delete retention for messages: "deleting" MOVES rows into + mqtt_messages_retention (same id), restorable until expires_at, purged + afterwards. All moves are single atomic CTE statements — an id can + never exist in both tables. Expired rows are purged LAZILY on every + read/write below (no scheduler needed; the expires_at index makes it a + cheap indexed delete).""" + + def __init__(self, dsn: str, retention_days: int = 30) -> None: + self._dsn = dsn + self._retention_days = retention_days + + def _purge_expired(self, cur: psycopg.Cursor) -> None: + cur.execute("DELETE FROM mqtt_messages_retention WHERE expires_at < now()") + + def delete(self, ids: list[int], deleted_by: str | None) -> int: + """Move rows out of mqtt_messages into retention. Unknown ids are + skipped (a selection can go stale under live traffic). Returns the + number of rows moved.""" + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + self._purge_expired(cur) + # single atomic move: DELETE ... RETURNING feeds the INSERT + cur.execute( + """ + WITH moved AS ( + DELETE FROM mqtt_messages + WHERE id = ANY(%s) + RETURNING id, received_at, topic, imei, device, + payload, payload_time + ), ins AS ( + INSERT INTO mqtt_messages_retention + (id, received_at, topic, imei, device, payload, + payload_time, deleted_by, expires_at) + SELECT id, received_at, topic, imei, device, payload, + payload_time, %s, + now() + make_interval(days => %s) + FROM moved + RETURNING id + ) + SELECT count(*) FROM ins + """, + (ids, deleted_by, self._retention_days), + ) + return cur.fetchone()[0] + finally: + conn.close() + + def restore(self, ids: list[int]) -> int: + """Move retained rows back into mqtt_messages with their ORIGINAL + ids. Unknown ids are skipped. A plain INSERT (no ON CONFLICT): an + id collision would mean a genuine anomaly and should fail loudly.""" + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + self._purge_expired(cur) + cur.execute( + """ + WITH moved AS ( + DELETE FROM mqtt_messages_retention + WHERE id = ANY(%s) + RETURNING id, received_at, topic, imei, device, + payload, payload_time + ), ins AS ( + INSERT INTO mqtt_messages + (id, received_at, topic, imei, device, payload, + payload_time) + SELECT id, received_at, topic, imei, device, payload, + payload_time + FROM moved + RETURNING id + ) + SELECT count(*) FROM ins + """, + (ids,), + ) + return cur.fetchone()[0] + finally: + conn.close() + + def count(self) -> int: + """Number of retained rows (expired ones purged first).""" + # autocommit: the purge is a write too - on a plain connection it + # would roll back on close (the count would still be correct within + # the connection, but the expired rows would survive). + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + self._purge_expired(cur) + cur.execute("SELECT count(*) FROM mqtt_messages_retention") + return cur.fetchone()[0] + finally: + conn.close() + + def list(self, limit: int, offset: int) -> tuple[list[dict[str, Any]], int]: + """One page of retained rows (newest deletions first) + the total, + for the pager. Expired rows are purged first (persistently - see + count()).""" + conn = psycopg.connect(self._dsn, autocommit=True) + try: + with conn.cursor() as cur: + self._purge_expired(cur) + cur.execute("SELECT count(*) FROM mqtt_messages_retention") + total = cur.fetchone()[0] + cur.execute( + """ + SELECT id, received_at, topic, imei, device, payload, + payload_time, deleted_at, deleted_by, expires_at + FROM mqtt_messages_retention + ORDER BY deleted_at DESC, id DESC + LIMIT %s OFFSET %s + """, + (limit, offset), + ) + rows = cur.fetchall() + finally: + conn.close() + return ( + [ + { + "id": row[0], + "received_at": row[1].isoformat(), + "topic": row[2], + "imei": row[3], + "device": row[4], + "payload": row[5], + "payload_time": row[6].isoformat() if row[6] is not None else None, + "deleted_at": row[7].isoformat(), + "deleted_by": row[8], + "expires_at": row[9].isoformat(), + } + for row in rows + ], + total, + ) + + def _validate_rename_body( data: dict[str, Any], source: str ) -> tuple[str, datetime | None, datetime | None, bool]: @@ -848,6 +986,26 @@ def _validate_attribute_body( return key.strip(), data["value"], effective_from, note +def _validate_id_list(data: object, source: str) -> list[int]: + """Validate a {"ids": [...]} request body for delete/restore: a JSON + object whose "ids" holds a non-empty list of integers, capped at + API_MAX_LIMIT-equivalent 10000. Raises ValueError with a descriptive + message otherwise.""" + if not isinstance(data, dict) or not isinstance(data.get("ids"), list): + raise ValueError(f'{source}: expected JSON on the form {{"ids": [1, 2, ...]}}') + ids = data["ids"] + if not ids: + raise ValueError(f"{source}: \"ids\" must not be empty") + for i, value in enumerate(ids): + if not isinstance(value, int) or isinstance(value, bool): + raise ValueError(f"{source}: ids[{i}] must be an integer") + if len(ids) > 10000: + raise ValueError( + f"{source}: too many ids ({len(ids)}); the limit is 10000" + ) + return ids + + def _validate_import_rows(data: object, source: str) -> list[dict[str, Any]]: """Validate an import body: a JSON array of exported-message objects (the GET /messages/export shape) — a single object is accepted too. @@ -958,6 +1116,7 @@ def create_app( user_store: UserStore | None = None, session_store: SessionStore | None = None, rename_store: DeviceRenameStore | None = None, + retention_policy: MessageRetentionPolicy | None = None, ) -> FastAPI: """Build the API application. @@ -974,6 +1133,7 @@ def create_app( dsn = _env("DB_DSN") max_limit = int(_env("API_MAX_LIMIT", "10000")) + retention_days = int(_env("RETENTION_DAYS", "30")) config = AppConfig(dsn=dsn, max_limit=max_limit) if token_store is None: token_store = TokenStore( @@ -995,6 +1155,8 @@ def create_app( session_store = SessionStore(dsn) if rename_store is None: rename_store = DeviceRenameStore(dsn) + if retention_policy is None: + retention_policy = MessageRetentionPolicy(dsn, retention_days=retention_days) # Brute-force lockout state (in-process, per username). login_attempts: dict[str, dict[str, float]] = {} @@ -1936,6 +2098,90 @@ def create_app( logger.info("imported %d message(s)", imported) return {"status": "imported", "imported": imported} + # --- soft-delete retention --------------------------------------------- + # "Deleting" selected messages MOVES them to mqtt_messages_retention + # (same id, original columns untouched) where they stay restorable for + # RETENTION_DAYS (env, default 30). Expired rows are purged lazily by + # every retention operation below - no scheduler, and an expired row can + # never be seen or restored. Delete/restore are superuser+; the reads + # (banner count, dialog listing) are open to every authenticated role - + # retained messages are the same messages they can already read. + + @app.post( + "/messages/delete", + tags=["data"], + dependencies=[Depends(authenticate), Depends(require_level("superuser"))], + ) + def delete_messages(body: Any = Body(...), request: Request = None) -> dict[str, Any]: + """Move the selected messages to retention. Body: {"ids": [...]} (at + least one, capped at 10000; unknown ids are skipped - a selection can + go stale under live traffic). The session username is recorded as + deleted_by (NULL for app tokens). + + FastAPI injects the real Request for any parameter annotated with + the Request type - the `= None` default is for tests only, never + used at runtime.""" + deleted_by = getattr(request.state, "username", None) if request is not None else None + try: + ids = _validate_id_list(body, "request body") + deleted = retention_policy.delete(ids, deleted_by) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + logger.info("moved %d message(s) to retention (by %s)", deleted, deleted_by) + return {"status": "deleted", "deleted": deleted} + + @app.get( + "/messages/retention/count", tags=["data"], dependencies=[Depends(authenticate)] + ) + def retention_count() -> dict[str, int]: + """Retained rows right now, for the Messages view's banner.""" + try: + return {"count": retention_policy.count()} + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + + @app.get("/messages/retention", tags=["data"], dependencies=[Depends(authenticate)]) + def list_retention( + limit: int = Query(100, gt=0, description="Page size"), + offset: int = Query(0, ge=0, description="Page offset"), + ) -> dict[str, Any]: + """One page of retained messages (newest deletions first), export + shape plus deleted_at/deleted_by/expires_at, with a filter-free + total for the pager.""" + limit = min(limit, config.max_limit) + try: + items, total = retention_policy.list(limit, offset) + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + return {"total": total, "limit": limit, "offset": offset, "items": items} + + @app.post( + "/messages/retention/restore", + tags=["data"], + dependencies=[Depends(authenticate), Depends(require_level("superuser"))], + ) + def restore_messages(body: Any = Body(...)) -> dict[str, Any]: + """Move retained messages back into mqtt_messages with their + ORIGINAL ids. Body: {"ids": [...]}; unknown ids are skipped (e.g. + already purged). Returns {"status": "restored", "restored": n}.""" + try: + ids = _validate_id_list(body, "request body") + restored = retention_policy.restore(ids) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + except psycopg.Error as exc: + raise _db_unavailable(exc) from exc + logger.info("restored %d message(s) from retention", restored) + return {"status": "restored", "restored": restored} + return app diff --git a/docker-compose.yml b/docker-compose.yml index 0e0faf0..58a0beb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -87,6 +87,10 @@ services: BOOTSTRAP_ADMIN_USER: ${BOOTSTRAP_ADMIN_USER:-} BOOTSTRAP_ADMIN_PASSWORD: ${BOOTSTRAP_ADMIN_PASSWORD:-} API_MAX_LIMIT: ${API_MAX_LIMIT:-10000} + # How long soft-deleted messages stay in retention (days) before + # being purged. Env-configurable, no schema impact (expires_at is + # computed per delete). + RETENTION_DAYS: ${RETENTION_DAYS:-30} LOG_LEVEL: ${LOG_LEVEL:-INFO} TZ: ${TZ:-Europe/Stockholm} ports: diff --git a/frontend/src/components/RetentionDialog.vue b/frontend/src/components/RetentionDialog.vue new file mode 100644 index 0000000..e4b2a6d --- /dev/null +++ b/frontend/src/components/RetentionDialog.vue @@ -0,0 +1,213 @@ + + + \ No newline at end of file diff --git a/frontend/src/views/Messages.vue b/frontend/src/views/Messages.vue index ee97d3a..5a840b1 100644 --- a/frontend/src/views/Messages.vue +++ b/frontend/src/views/Messages.vue @@ -1,6 +1,6 @@