jobe/nodeledge-mqtt-listener
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 Add message delete

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-09-06 22:35:24 +02:00
parent 64847b9ca1
commit f3fc394c0c
7 changed files with 797 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/app.py
+246
View File
@@ -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