feat: log raw telegram updates in development

raw_updates.py

@@ -0,0 +1,126 @@
+"""Persistent storage for raw incoming Telegram updates (development diagnostics)."""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import closing
+import json
+import logging
+import os
+import re
+import sqlite3
+from pathlib import Path
+from typing import Any, Awaitable, Callable
+
+from aiogram import BaseMiddleware
+from aiogram.types import TelegramObject, Update
+
+logger = logging.getLogger(__name__)
+
+RAW_UPDATES_DB_PATH = os.getenv("RAW_UPDATES_DB_PATH", "/app/data/bot_updates.sqlite3")
+RAW_UPDATE_RETENTION_DAYS = 30
+_REDACTED = "[REDACTED]"
+_SECRET_KEYS = {
+    "authorization", "api_key", "access_token", "bot_token", "token", "secret", "password",
+}
+_SECRET_SUFFIXES = ("_api_key", "_access_token", "_bot_token", "_secret", "_password")
+
+
+def _sanitize(value: Any) -> Any:
+    if isinstance(value, dict):
+        result = {}
+        for key, item in value.items():
+            normalized = str(key).casefold()
+            if normalized in _SECRET_KEYS or normalized.endswith(_SECRET_SUFFIXES):
+                result[key] = _REDACTED
+            else:
+                result[key] = _sanitize(item)
+        return result
+    if isinstance(value, list):
+        return [_sanitize(item) for item in value]
+    if isinstance(value, str):
+        value = re.sub(r"(?i)bearer\s+[^\s\"']+", "Bearer [REDACTED]", value)
+        value = re.sub(r"(?i)(api[_-]?key|access[_-]?token|bot[_-]?token|secret|password)(\s*[=:]\s*)[^\s,;\"']+", r"\1\2[REDACTED]", value)
+        value = re.sub(r"(?i)(?:sk-[A-Za-z0-9_-]{12,}|[0-9]{8,}:[A-Za-z0-9_-]{20,})", _REDACTED, value)
+    return value
+
+
+def _connect(db_path: str | None = None) -> sqlite3.Connection:
+    path = Path(db_path or os.getenv("RAW_UPDATES_DB_PATH", RAW_UPDATES_DB_PATH))
+    path.parent.mkdir(parents=True, exist_ok=True)
+    connection = sqlite3.connect(path, timeout=10)
+    connection.row_factory = sqlite3.Row
+    connection.execute("PRAGMA busy_timeout = 10000")
+    return connection
+
+
+def init_raw_updates_db(db_path: str | None = None) -> None:
+    with closing(_connect(db_path)) as connection, connection:
+        connection.execute(
+            """CREATE TABLE IF NOT EXISTS raw_updates (
+                update_id INTEGER PRIMARY KEY,
+                event_type TEXT NOT NULL,
+                payload TEXT NOT NULL,
+                received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+            )"""
+        )
+        connection.execute(
+            "CREATE INDEX IF NOT EXISTS idx_raw_updates_received_at ON raw_updates(received_at)"
+        )
+        connection.execute(
+            "DELETE FROM raw_updates WHERE received_at < datetime('now', ?)",
+            (f"-{RAW_UPDATE_RETENTION_DAYS} days",),
+        )
+
+
+def save_raw_update(update_id: int, payload: dict[str, Any], db_path: str | None = None) -> None:
+    safe_payload = _sanitize(payload)
+    event_type = next(
+        (key for key, value in safe_payload.items() if key != "update_id" and value is not None),
+        "unknown",
+    )
+    serialized = json.dumps(safe_payload, ensure_ascii=False, separators=(",", ":"))
+    with closing(_connect(db_path)) as connection, connection:
+        connection.execute(
+            "INSERT OR REPLACE INTO raw_updates (update_id, event_type, payload) VALUES (?, ?, ?)",
+            (int(update_id), event_type, serialized),
+        )
+        connection.execute(
+            "DELETE FROM raw_updates WHERE received_at < datetime('now', ?)",
+            (f"-{RAW_UPDATE_RETENTION_DAYS} days",),
+        )
+
+
+def get_raw_update(update_id: int, db_path: str | None = None) -> dict[str, Any] | None:
+    with closing(_connect(db_path)) as connection:
+        row = connection.execute(
+            "SELECT update_id, event_type, payload, received_at FROM raw_updates WHERE update_id = ?",
+            (int(update_id),),
+        ).fetchone()
+    if row is None:
+        return None
+    return {
+        "update_id": row["update_id"],
+        "event_type": row["event_type"],
+        "payload": json.loads(row["payload"]),
+        "received_at": row["received_at"],
+    }
+
+
+class RawUpdateMiddleware(BaseMiddleware):
+    async def __call__(
+        self,
+        handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]],
+        event: TelegramObject,
+        data: dict[str, Any],
+    ) -> Any:
+        if isinstance(event, Update):
+            try:
+                await asyncio.to_thread(
+                    save_raw_update,
+                    event.update_id,
+                    event.model_dump(mode="json"),
+                )
+            except Exception as exc:
+                logger.warning("Raw update persistence failed (%s)", type(exc).__name__)
+        return await handler(event, data)