feat: sys-2 Edge-Queue — Signatur-Prüfung, Dedup, asynchrone Verarbeitung

- webhook_queue.py: WhatsApp X-Hub-Signature-256 + Telegram Webhook-Intake
- queue_handler.py: SQLite-basierte Queue mit Idempotenz (UNIQUE message_id)
- queue_worker.py: Asynchroner Worker für GPU-Verarbeitung
- app.py: Webhooks umgebaut → sofort 200 OK, Worker verarbeitet später
- Redis 8.0 läuft als Infrastruktur (Queue ist SQLite für Persistenz)

.gitea/workflows/quality.yml

@@ -0,0 +1,48 @@
+# BSN-Chatsystem — CI Quality Gate
+# Läuft bei JEDEM Push. Kein --no-verify-Umgehen möglich.
+# Konfiguriert für Python 3.13 + Ruff 0.15 + mypy 2.1
+
+name: Quality Gate
+
+on:
+  push:
+    branches: [main, develop, "feature/**", "fix/**"]
+  pull_request:
+    branches: [main]
+
+jobs:
+  quality:
+    name: "🛡️ Lint + Type Check + Tests"
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - name: 📥 Checkout
+        uses: actions/checkout@v4
+
+      - name: 🐍 Python 3.13
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+
+      - name: 📦 Dependencies
+        run: |
+          pip install ruff mypy pytest pytest-cov
+          # Projekt-spezifische Deps (falls requirements.txt existiert)
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+
+      # ── Linting + Sicherheit ──
+      - name: 🔍 Ruff — Linting + Sicherheit
+        run: ruff check . --output-format=github
+
+      # ── Formatierung ──
+      - name: 🎨 Ruff — Formatierung
+        run: ruff format --check .
+
+      # ── Typ-Prüfung ──
+      - name: 🔎 mypy — Statische Typ-Prüfung
+        run: mypy . --strict
+
+      # ── Tests + Coverage ──
+      - name: 🧪 pytest — Tests + Coverage
+        run: pytest --cov --cov-fail-under=80 --tb=short

app.py

@@ -1020,331 +1020,15 @@ def whatsapp_webhook():
             return challenge, 200
         return "Forbidden", 403
 
-    # POST: Incoming message
+    # POST: Incoming message → Queue (asynchrone Verarbeitung)
     body = request.get_json(force=True, silent=True)
     if not body:
         return "OK", 200
 
-    try:
+    from webhook_queue import whatsapp_intake
-        import threading
+    return whatsapp_intake(request, db)
-        entries = body.get("entry", [])
-        db = get_db()
-
-        for entry in entries:
-            changes = entry.get("changes", [])
-            for change in changes:
-                value = change.get("value", {})
-                contacts = value.get("contacts", [{}])
-                messages = value.get("messages", [])
-
-                for msg in messages:
-                    sender_id = msg.get("from", "unknown")
-                    sender_name = contacts[0].get("profile", {}).get("name", "unknown")
-                    msg_type = msg.get("type", "text")
-                    content = ""
-                    media_path = None
-                    spiel_titel = None
-                    category_override = None  # set to "ausverkauft" for #out
-
-                    if msg_type == "text":
-                        content = msg.get("text", {}).get("body", "")
-                    elif msg_type == "image":
-                        content = msg.get("image", {}).get("caption", "[Bild ohne Text]")
-                        media_id = msg.get("image", {}).get("id")
-                        if media_id:
-                            media_path, _ = download_media(media_id)
-                    elif msg_type == "audio":
-                        content = "[🎤 Sprachnachricht — Transkription läuft...]"
-                        media_id = msg.get("audio", {}).get("id")
-                        if media_id:
-                            media_path, _ = download_media(media_id)
-                    elif msg_type == "video":
-                        content = msg.get("video", {}).get(
-                            "caption", "[🎬 Video — Transkription läuft...]"
-                        )
-                        media_id = msg.get("video", {}).get("id")
-                        if media_id:
-                            media_path, _ = download_media(media_id)
-                    elif msg_type == "document":
-                        content = msg.get("document", {}).get(
-                            "caption", "[Dokument]"
-                        )
-                        media_id = msg.get("document", {}).get("id")
-                        if media_id:
-                            media_path, _ = download_media(media_id)
-                    else:
-                        content = f"[{msg_type}]"
-
-                    # #out detection: set category to "ausverkauft" and extract game title
-                    if "#out" in content.lower():
-                        category_override = "ausverkauft"
-                        idx = content.lower().find("#out")
-                        after = content[idx + 4:].strip()
-                        if after:
-                            spiel_titel = re.split(r'[.!?\n]| - | – | — | ist | war | in ', after, maxsplit=1)[0].strip()[:200]
-
-                    # GDPR: "Meine Daten löschen" trigger
-                    if msg_type == "text" and content.strip().lower() == "meine daten löschen":
-                        _gdpr_delete_sender(db, sender_id)
-                        if META_TOKEN:
-                            try:
-                                send_whatsapp_message(
-                                    sender_id,
-                                    "✅ Alle deine Daten wurden vollständig und endgültig vom Server gelöscht. "
-                                    "Es befinden sich keinerlei personenbezogene Daten mehr von dir bei uns. "
-                                    "Lediglich ein anonymer Vermerk über die erfolgte Löschung bleibt für "
-                                    "unsere Dokumentationspflicht.",
-                                )
-                            except Exception:
-                                pass
-                        continue
-
-                    # FAQ-Trigger: "Keine Antwort" / "Warum keine Nachricht"
-                    faq_keywords = [
-                        "keine antwort", "bekomme keine", "keine nachricht",
-                        "antwort kommt nicht", "warum keine", "keine rückmeldung",
-                        "why no reply", "not receiving", "save contact", "nummer speichern"
-                    ]
-                    if msg_type == "text" and any(kw in content.lower() for kw in faq_keywords):
-                        if META_TOKEN:
-                            try:
-                                send_whatsapp_message(
-                                    sender_id,
-                                    "💡 *Tipp:* Damit unsere Antworten bei dir ankommen, "
-                                    "speichere bitte unsere Nummer in deinem Adressbuch. "
-                                    "WhatsApp filtert Nachrichten von unbekannten Business-Accounts "
-                                    "auf manchen Handys leider automatisch heraus.\n\n"
-                                    "Sobald du uns als Kontakt gespeichert hast, klappt's! ✅"
-                                )
-                            except Exception:
-                                pass
-                        continue
-
-                    # ── Nutzer-Registrierung per WhatsApp ────────────────
-                    pending_codes = getattr(app, '_pending_codes', {})
-                    if msg_type == "text":
-                        clean = content.strip().lower()
-                        # "registrieren" → Code senden
-                        if clean == "registrieren":
-                            import random
-                            code = str(random.randint(100000, 999999))
-                            pending_codes[sender_id] = code
-                            app._pending_codes = pending_codes
-                            if META_TOKEN:
-                                try:
-                                    send_whatsapp_message(sender_id,
-                                        f"🔐 Dein Verifikationscode: *{code}*\n\n"
-                                        f"Antworte einfach mit: code {code}")
-                                except Exception:
-                                    pass
-                            continue
-                        # "code 123456" → verifizieren + User anlegen
-                        if clean.startswith("code ") and len(clean) >= 10:
-                            entered = clean[5:].strip()
-                            if sender_id in pending_codes and pending_codes[sender_id] == entered:
-                                del pending_codes[sender_id]
-                                app._pending_codes = pending_codes
-                                # User anlegen — phone_hash statt Klartext
-                                existing = _lookup_user_by_phone(db, sender_id)
-                                if not existing:
-                                    phone_hash_val, phone_last4_val = _hash_phone(sender_id)
-                                    db.execute(
-                                        "INSERT INTO users (phone, name, verified, phone_hash, phone_last4) VALUES (?,?,1,?,?)",
-                                        (phone_last4_val, sender_name, phone_hash_val, phone_last4_val),
-                                    )
-                                    db.commit()
-                                if META_TOKEN:
-                                    try:
-                                        send_whatsapp_message(sender_id,
-                                            "✅ Du bist jetzt registriert! Dein Profil: "
-                                            "https://chat.datenhimmel.work/profile")
-                                    except Exception:
-                                        pass
-                            else:
-                                if META_TOKEN:
-                                    try:
-                                        send_whatsapp_message(sender_id,
-                                            "❌ Falscher Code. Schick \"registrieren\" für einen neuen.")
-                                    except Exception:
-                                        pass
-                            continue
-
-                    # ── Cross-Channel: User-Profil zuordnen (via phone_hash) ──
-                    user_row = _lookup_user_by_phone(db, sender_id)
-
-                    # Check for existing thread from this sender (5-min window)
-                    recent = db.execute(
-                        """SELECT id, content, media_path, halle FROM submissions
-                           WHERE channel='whatsapp' AND sender_id=? AND status != 'gdpr_deleted'
-                           ORDER BY created_at DESC LIMIT 1""",
-                        (sender_id,),
-                    ).fetchone()
-
-                    is_new_user = db.execute(
-                        "SELECT COUNT(*) FROM submissions WHERE channel='whatsapp' AND sender_id=? AND status != 'gdpr_deleted'",
-                        (sender_id,),
-                    ).fetchone()[0] == 0
-
-                    THREAD_WINDOW_MINUTES = 5
-                    thread_found = False
-                    if recent:
-                        age_check = db.execute(
-                            "SELECT (strftime('%s','now') - strftime('%s',created_at)) < ? FROM submissions WHERE id=?",
-                            (THREAD_WINDOW_MINUTES * 60, recent["id"]),
-                        ).fetchone()
-                        if age_check and age_check[0]:
-                            thread_found = True
-
-                    if thread_found:
-                        # Append to existing thread
-                        prefix = {
-                            "text": "", "image": "\n[🖼️ Bild] ",
-                            "audio": "\n[🎤 Sprachnachricht]",
-                            "video": "\n[🎬 Video] ", "document": "\n[📎 Dokument] ",
-                        }.get(msg_type, f"\n[{msg_type}] ")
-                        new_content = (recent["content"] or "") + prefix + content
-                        # Re-extract halle from combined content
-                        new_halle = extract_halle(new_content) or recent["halle"]
-                        db.execute(
-                            "UPDATE submissions SET content=?, halle=?, created_at=CURRENT_TIMESTAMP WHERE id=?",
-                            (new_content, new_halle, recent["id"]),
-                        )
-                        # Check for #out in appended content — update category + spiel_titel
-                        if "#out" in content.lower():
-                            out_title = None
-                            idx = content.lower().find("#out")
-                            after = content[idx + 4:].strip()
-                            if after:
-                                out_title = re.split(r'[.!?\n]| - | – | — | ist | war | in ', after, maxsplit=1)[0].strip()[:200]
-                            db.execute(
-                                "UPDATE submissions SET category=?, spiel_titel=? WHERE id=?",
-                                ("ausverkauft", out_title, recent["id"]),
-                            )
-                            # Auto-mark matching neuheiten as sold out
-                            if out_title:
-                                mark_ausverkauft_in_neuheiten(db, out_title)
-                        # Merge media paths (keep all, separated by ||)
-                        if media_path:
-                            merged_media = media_path
-                            if recent["media_path"]:
-                                merged_media = recent["media_path"] + "||" + media_path
-                            db.execute(
-                                "UPDATE submissions SET media_path=? WHERE id=?",
-                                (merged_media, recent["id"]),
-                            )
-                    else:
-                        # New thread
-                        halle = extract_halle(content)
-                        cat = category_override or None
-                        db.execute(
-                            """INSERT INTO submissions
-                               (channel, sender_id, sender_name, type, content, media_path, status, halle, category, spiel_titel, user_id)
-                               VALUES (?, ?, ?, ?, ?, ?, 'new', ?, ?, ?, ?)""",
-                            ("whatsapp", sender_id, sender_name, msg_type, content, media_path, halle, cat, spiel_titel,
-                             user_row["id"] if user_row else None),
-                        )
-                        # Auto-mark matching neuheiten as sold out
-                        if cat == "ausverkauft" and spiel_titel:
-                            mark_ausverkauft_in_neuheiten(db, spiel_titel)
-
-                    # ── Admin-Benachrichtigung: Neue Einsendung ──────────
-                    sub_id = recent["id"] if thread_found else db.execute(
-                        "SELECT last_insert_rowid()"
-                    ).fetchone()[0]
-                    content_preview = content.replace("\n", " ")[:80]
-                    type_emoji = {"text": "💬", "image": "🖼️", "audio": "🎤", "video": "🎬"}.get(msg_type, "📎")
-                    notify_admin(
-                        f"{type_emoji} {sender_name}: {content_preview}\n"
-                        f"→ https://chat.datenhimmel.work/admin/{sub_id}"
-                    )
-
-                    # Auto-reply: only on NEW threads, not appends
-                    if not thread_found and META_TOKEN:
-                        try:
-                            if is_new_user:
-                                reply = (
-                                    f"🎲 Hey {sender_name}! Cool, dass du da bist — willkommen in der "
-                                    f"brettspiel-news.de Community!\n\n"
-                                    f"Schick uns einfach, was dich bewegt: 💬 Text, 🎤 Sprachnachrichten, "
-                                    f"🖼️ Fotos oder 🎬 Videos — alles rund um Brettspiele. Wir schauen uns alles an!\n\n"
-                                    f"✍️ Nenn uns deinen Vornamen, wenn du magst — das ist deine Einwilligung, "
-                                    f"dass wir dich als Quelle nennen, falls wir deinen Beitrag verwenden. "
-                                    f"Ohne Namen schreiben wir „anonyme Quelle\".\n\n"
-                                    f"💡 Speicher unsere Nummer im Adressbuch, "
-                                    f"damit unsere Antworten sicher ankommen.\n\n"
-                                    f"🔒 „Meine Daten löschen\" genügt — und alles ist sofort und vollständig weg."
-                                )
-                                send_whatsapp_message(sender_id, reply)
-                                # Thorsten voice: same text but without emojis
-                                tts_text = re.sub(r'[^\x00-\x7F\u00C0-\u00FF\u0100-\u017F\u0180-\u024F\u1E00-\u1EFF\u2018-\u201F\u2013\u2014\u2026\u00A0\u00AB\u00BB\u20AC]', '', reply)
-                                tts_text = re.sub(r' +', ' ', tts_text).strip()
-                                audio_path = generate_thorsten_tts(tts_text)
-                                if audio_path:
-                                    media_id = upload_whatsapp_media(audio_path)
-                                    if media_id:
-                                        send_whatsapp_audio(sender_id, media_id)
-                            else:
-                                reply = (
-                                    f"📨 Danke {sender_name}! Ist angekommen — "
-                                    f"wir kümmern uns drum."
-                                )
-                                send_whatsapp_message(sender_id, reply)
-                        except Exception:
-                            pass
-
-                    # Transcribe audio in background (non-blocking)
-                    if msg_type == "audio" and media_path:
-                        sub_id = recent["id"] if thread_found else db.execute(
-                            "SELECT last_insert_rowid()"
-                        ).fetchone()[0]
-                        threading.Thread(
-                            target=_transcribe_bg,
-                            args=(DB_PATH, sub_id, media_path),
-                            daemon=True,
-                        ).start()
-
-                    # Transcribe video in background
-                    if msg_type == "video" and media_path:
-                        sub_id = recent["id"] if thread_found else db.execute(
-                            "SELECT last_insert_rowid()"
-                        ).fetchone()[0]
-                        threading.Thread(
-                            target=_transcribe_video_bg,
-                            args=(DB_PATH, sub_id, media_path),
-                            daemon=True,
-                        ).start()
-
-                    # Auto-summarize text/image submissions (LLM hall + summary)
-                    if msg_type in ("text", "image") and not thread_found and content and len(content.strip()) > 20:
-                        sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
-                        # Only if regex didn't already find a halle
-                        if not halle:
-                            threading.Thread(
-                                target=auto_process_bg,
-                                args=(DB_PATH, sub_id),
-                                daemon=True,
-                            ).start()
-
-                    # LLM-Triage: 3-Stufen-Klassifikation (nur für neue Threads)
-                    if not thread_found:
-                        sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
-                        threading.Thread(
-                            target=_run_triage_async,
-                            args=(sub_id,),
-                            daemon=True,
-                        ).start()
-
-        db.commit()
-    except Exception as e:
-        print(f"[webhook error] {e}", file=sys.stderr)
-        return "OK", 200
-
-    return "OK", 200
 
 
-@app.route("/admin")
-def admin_dashboard():
-    """Internal admin dashboard — filterable table of all submissions."""
     db = get_db()
 
     channel_filter = request.args.get("channel", "")
@@ -5392,189 +5076,15 @@ def telegram_webhook():
         return jsonify({"status": "off", "reason": "TELEGRAM_TOKEN not set"}), 500
 
     # POST: Incoming message
+    # POST: Incoming message → Queue (asynchrone Verarbeitung)
     body = request.get_json(force=True, silent=True)
     if not body or not TELEGRAM_TOKEN:
         return "OK", 200
 
-    try:
+    from webhook_queue import telegram_intake
-        import threading
-        msg = body.get("message", {})
-        if not msg:
-            return "OK", 200
-
-        chat = msg.get("chat", {})
-        sender_id = str(chat.get("id", "unknown"))
-        sender_name = msg.get("from", {}).get("first_name", "") or ""
-        last_name = msg.get("from", {}).get("last_name", "")
-        if last_name:
-            sender_name = "{} {}".format(sender_name, last_name).strip()
-        if not sender_name:
-            sender_name = sender_id
-
-        msg_type = "text"
-        content = ""
-        media_path = None
-        file_id = None
-
-        if "text" in msg:
-            msg_type = "text"
-            content = msg["text"]
-        elif "photo" in msg:
-            msg_type = "image"
-            content = msg.get("caption", "[Bild ohne Text]")
-            file_id = msg["photo"][-1]["file_id"]
-        elif "video" in msg:
-            msg_type = "video"
-            content = msg.get("caption", "[🎬 Video — Transkription läuft...]")
-            file_id = msg["video"]["file_id"]
-        elif "voice" in msg:
-            msg_type = "audio"
-            content = "[🎤 Sprachnachricht — Transkription läuft...]"
-            file_id = msg["voice"]["file_id"]
-        elif "audio" in msg:
-            msg_type = "audio"
-            content = msg.get("caption", "[🎤 Audio]")
-            file_id = msg["audio"]["file_id"]
-        elif "document" in msg:
-            msg_type = "document"
-            content = msg.get("caption", "[📎 Dokument]")
-            file_id = msg["document"]["file_id"]
-        elif "sticker" in msg:
-            content = "[Sticker: {}]".format(msg['sticker'].get('emoji', ''))
-            msg_type = "document"
-        elif "location" in msg:
-            lat = msg["location"]["latitude"]
-            lon = msg["location"]["longitude"]
-            content = "[📍 Standort: {}, {}]".format(lat, lon)
-            msg_type = "text"
-        else:
-            content = "[Unbekannter Nachrichtentyp]"
-            msg_type = "text"
-
-        if file_id:
-            media_path = _telegram_download_file(file_id)
-
-        # GDPR delete trigger
-        if msg_type == "text" and content.strip().lower() == "meine daten löschen":
-            _gdpr_delete_sender(get_db(), sender_id)
-            if TELEGRAM_TOKEN:
-                import requests as _r
-                try:
-                    _r.post(
-                        "{}/sendMessage".format(TELEGRAM_API),
-                        json={"chat_id": int(sender_id),
-                              "text": "✅ Alle deine Daten wurden vollständig und endgültig vom Server gelöscht. "
-                                      "Es befinden sich keinerlei personenbezogene Daten mehr von dir bei uns."},
-                        timeout=10,
-                    )
-                except Exception:
-                    pass
-            return "OK", 200
-
     db = get_db()
-        spiel_titel = None
+    return telegram_intake(request, db)
-        category_override = None
 
-        if "#out" in content.lower():
-            category_override = "ausverkauft"
-            idx = content.lower().find("#out")
-            after = content[idx + 4:].strip()
-            if after:
-                spiel_titel = re.split(r'[.!?\n]| - | – | — | ist | war | in ', after, maxsplit=1)[0].strip()[:200]
-
-        halle = extract_halle(content)
-
-        THREAD_WINDOW_MINUTES = 5
-        recent = db.execute(
-            """SELECT id, content, media_path, halle FROM submissions
-               WHERE channel='telegram' AND sender_id=? AND status != 'gdpr_deleted'
-               ORDER BY created_at DESC LIMIT 1""",
-            (sender_id,),
-        ).fetchone()
-
-        is_new_user = db.execute(
-            "SELECT COUNT(*) FROM submissions WHERE channel='telegram' AND sender_id=? AND status != 'gdpr_deleted'",
-            (sender_id,),
-        ).fetchone()[0] == 0
-
-        thread_found = False
-        if recent:
-            age_check = db.execute(
-                "SELECT (strftime('%s','now') - strftime('%s',created_at)) < ? FROM submissions WHERE id=?",
-                (THREAD_WINDOW_MINUTES * 60, recent["id"]),
-            ).fetchone()
-            if age_check and age_check[0]:
-                thread_found = True
-
-        if thread_found:
-            prefix = {
-                "text": "", "image": "\n[🖼️ Bild] ",
-                "audio": "\n[🎤 Sprachnachricht]", "video": "\n[🎬 Video] ",
-                "document": "\n[📎 Dokument] ",
-            }.get(msg_type, "\n[{}] ".format(msg_type))
-            new_content = (recent["content"] or "") + prefix + content
-            new_halle = extract_halle(new_content) or recent["halle"]
-            db.execute(
-                "UPDATE submissions SET content=?, halle=?, created_at=CURRENT_TIMESTAMP WHERE id=?",
-                (new_content, new_halle, recent["id"]),
-            )
-            if "#out" in content.lower():
-                out_title = None
-                idx2 = content.lower().find("#out")
-                after2 = content[idx2 + 4:].strip()
-                if after2:
-                    out_title = re.split(r'[.!?\n]| - | – | — | ist | war | in ', after2, maxsplit=1)[0].strip()[:200]
-                db.execute("UPDATE submissions SET category=?, spiel_titel=? WHERE id=?",
-                          ("ausverkauft", out_title, recent["id"]))
-                if out_title:
-                    mark_ausverkauft_in_neuheiten(db, out_title)
-            if media_path:
-                merged = media_path
-                if recent["media_path"]:
-                    merged = recent["media_path"] + "||" + media_path
-                db.execute("UPDATE submissions SET media_path=? WHERE id=?", (merged, recent["id"]))
-        else:
-            cat = category_override or None
-            db.execute(
-                """INSERT INTO submissions
-                   (channel, sender_id, sender_name, type, content, media_path, status, halle, category, spiel_titel)
-                   VALUES (?, ?, ?, ?, ?, ?, 'new', ?, ?, ?)""",
-                ("telegram", sender_id, sender_name, msg_type, content, media_path, halle, cat, spiel_titel),
-            )
-            if cat == "ausverkauft" and spiel_titel:
-                mark_ausverkauft_in_neuheiten(db, spiel_titel)
-
-        if not thread_found:
-            _telegram_autoreply(int(sender_id), sender_name, is_new_user)
-
-        if msg_type == "audio" and media_path:
-            import threading
-            sub_id = recent["id"] if thread_found else db.execute("SELECT last_insert_rowid()").fetchone()[0]
-            threading.Thread(target=_transcribe_bg, args=(DB_PATH, sub_id, media_path), daemon=True).start()
-
-        if msg_type == "video" and media_path:
-            import threading
-            sub_id = recent["id"] if thread_found else db.execute("SELECT last_insert_rowid()").fetchone()[0]
-            threading.Thread(target=_transcribe_video_bg, args=(DB_PATH, sub_id, media_path), daemon=True).start()
-
-        if msg_type in ("text", "image") and not thread_found and content and len(content.strip()) > 20:
-            import threading
-            sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
-            if not halle:
-                threading.Thread(target=auto_process_bg, args=(DB_PATH, sub_id), daemon=True).start()
-
-        # LLM-Triage: 3-Stufen-Klassifikation (nur für neue Threads)
-        if not thread_found:
-            import threading
-            sub_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
-            threading.Thread(target=_run_triage_async, args=(sub_id,), daemon=True).start()
-
-        db.commit()
-    except Exception as e:
-        print("[telegram webhook error] {}".format(e), file=sys.stderr)
-        return "OK", 200
-
-    return "OK", 200
 
 if __name__ == "__main__":
     import threading

queue_handler.py

@@ -0,0 +1,311 @@
+#!/usr/bin/env python3
+"""
+Queue-Handler — SQLite-basierte Submission-Queue mit Idempotenz.
+===============================================================
+
+Architektur:
+  Webhook → Signatur-Prüfung → Queue (sofort 200 OK) → Worker → GPU-Verarbeitung
+
+Eigenschaften:
+  - Persistente Queue in SQLite (kein Redis nötig)
+  - Idempotenz via message_id UNIQUE (WhatsApp message ID / Telegram update_id)
+  - Webhook-Signaturprüfung (WhatsApp X-Hub-Signature-256, Telegram optional)
+  - Status-Tracking (pending → processing → completed/failed)
+
+Nutzung:
+  from queue_handler import QueueHandler
+  qh = QueueHandler(db)
+  qh.enqueue(channel="whatsapp", message_id="wamid.xxx", payload=body)
+"""
+
+import hashlib
+import hmac
+import json
+import logging
+import sqlite3
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+# ── Webhook-Signaturprüfung ──────────────────────────────────────────
+
+def verify_whatsapp_signature(
+    body: bytes,
+    signature_header: str,
+    app_secret: str,
+) -> bool:
+    """
+    Prüft die WhatsApp X-Hub-Signature-256.
+
+    Args:
+        body: Der rohe Request-Body (bytes).
+        signature_header: Der Wert des X-Hub-Signature-256 Headers.
+        app_secret: Der Meta App Secret (aus Environment).
+
+    Returns:
+        True wenn die Signatur gültig ist, sonst False.
+    """
+    if not signature_header or not app_secret:
+        logger.warning("WhatsApp-Signaturprüfung übersprungen — kein Secret/Header")
+        return True  # Im Test-Modus erlauben
+
+    # Format: "sha256=<hex-hash>"
+    if not signature_header.startswith("sha256="):
+        logger.warning("Unbekanntes Signatur-Format: %s", signature_header[:20])
+        return False
+
+    expected = signature_header[7:]  # Nach "sha256="
+    actual = hmac.new(
+        key=app_secret.encode("utf-8"),
+        msg=body,
+        digestmod=hashlib.sha256,
+    ).hexdigest()
+
+    # Constant-time comparison (gegen Timing-Angriffe)
+    if not hmac.compare_digest(actual, expected):
+        logger.warning("WhatsApp-Signatur UNGÜLTIG — echo/replay erkannt")
+        return False
+
+    return True
+
+
+# ── Queue-Implementierung ────────────────────────────────────────────
+
+class QueueHandler:
+    """
+    SQLite-basierte Submission-Queue mit Idempotenz.
+
+    Verwendet die Tabelle `submission_queue` in der bestehenden SQLite-DB.
+    """
+
+    def __init__(self, db: sqlite3.Connection):
+        self.db = db
+        self._ensure_table()
+
+    def _ensure_table(self) -> None:
+        """Erstellt die Queue-Tabelle, falls nicht vorhanden."""
+        self.db.execute("""
+            CREATE TABLE IF NOT EXISTS submission_queue (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                message_id TEXT NOT NULL UNIQUE,
+                channel TEXT NOT NULL CHECK(channel IN ('whatsapp', 'telegram')),
+                payload TEXT NOT NULL,
+                raw_body TEXT,
+                status TEXT NOT NULL DEFAULT 'pending'
+                    CHECK(status IN ('pending', 'processing', 'completed', 'failed')),
+                attempt_count INTEGER NOT NULL DEFAULT 0,
+                max_attempts INTEGER NOT NULL DEFAULT 3,
+                last_error TEXT,
+                created_at TEXT NOT NULL DEFAULT (datetime('now')),
+                started_at TEXT,
+                completed_at TEXT
+            )
+        """)
+        self.db.execute("""
+            CREATE INDEX IF NOT EXISTS idx_queue_status
+            ON submission_queue(status, created_at)
+        """)
+        self.db.commit()
+
+    def enqueue(
+        self,
+        channel: str,
+        message_id: str,
+        payload: dict,
+        raw_body: Optional[bytes] = None,
+    ) -> tuple[bool, str]:
+        """
+        Reiht eine Submission in die Queue ein.
+
+        Args:
+            channel: 'whatsapp' oder 'telegram'
+            message_id: Eindeutige Nachrichten-ID (WhatsApp: wamid.xxx, Telegram: update_id)
+            payload: Das geparste JSON der Nachricht.
+            raw_body: Optionaler roher Body (für Debugging).
+
+        Returns:
+            (success, message) — True = eingereiht, False = Duplikat/Fehler.
+        """
+        payload_json = json.dumps(payload, ensure_ascii=False)
+        raw_bytes = raw_body.decode("utf-8", errors="replace") if raw_body else None
+
+        try:
+            self.db.execute(
+                """
+                INSERT INTO submission_queue (message_id, channel, payload, raw_body)
+                VALUES (?, ?, ?, ?)
+                """,
+                (message_id, channel, payload_json, raw_bytes),
+            )
+            self.db.commit()
+            logger.info(
+                "Submission in Queue: channel=%s message_id=%s",
+                channel,
+                message_id,
+            )
+            return True, "enqueued"
+
+        except sqlite3.IntegrityError:
+            # UNIQUE constraint auf message_id → Duplikat erkannt
+            logger.info(
+                "Duplikat erkannt und verworfen: channel=%s message_id=%s",
+                channel,
+                message_id,
+            )
+            # Prüfen, ob die Nachricht bereits verarbeitet wurde
+            row = self.db.execute(
+                "SELECT status FROM submission_queue WHERE message_id = ?",
+                (message_id,),
+            ).fetchone()
+            if row and row[0] == "completed":
+                return False, "already_processed"
+            return False, "duplicate_pending"
+
+        except Exception as e:
+            logger.exception("Fehler beim Enqueue: %s", e)
+            return False, f"error: {e}"
+
+    def dequeue(self, limit: int = 5) -> list[dict]:
+        """
+        Holt pending Submissions aus der Queue und markiert sie als 'processing'.
+
+        Returns:
+            Liste von Submission-Dicts mit id, message_id, channel, payload.
+        """
+        self.db.execute("BEGIN IMMEDIATE")
+        try:
+            rows = self.db.execute(
+                """
+                SELECT id, message_id, channel, payload, attempt_count
+                FROM submission_queue
+                WHERE status = 'pending'
+                ORDER BY created_at ASC
+                LIMIT ?
+                """,
+                (limit,),
+            ).fetchall()
+
+            if not rows:
+                self.db.commit()
+                return []
+
+            ids = [row[0] for row in rows]
+            placeholders = ",".join("?" for _ in ids)
+            self.db.execute(
+                f"""
+                UPDATE submission_queue
+                SET status = 'processing',
+                    started_at = datetime('now'),
+                    attempt_count = attempt_count + 1
+                WHERE id IN ({placeholders})
+                """,
+                ids,
+            )
+            self.db.commit()
+
+            result = []
+            for row in rows:
+                result.append({
+                    "id": row[0],
+                    "message_id": row[1],
+                    "channel": row[2],
+                    "payload": json.loads(row[3]),
+                    "attempt_count": row[4],
+                })
+
+            if result:
+                logger.info("Dequeued %d submissions", len(result))
+            return result
+
+        except Exception:
+            self.db.rollback()
+            raise
+
+    def mark_completed(self, submission_id: int) -> None:
+        """Markiert eine Submission als erfolgreich verarbeitet."""
+        self.db.execute(
+            """
+            UPDATE submission_queue
+            SET status = 'completed', completed_at = datetime('now')
+            WHERE id = ?
+            """,
+            (submission_id,),
+        )
+        self.db.commit()
+
+    def mark_failed(self, submission_id: int, error: str, requeue: bool = True) -> None:
+        """
+        Markiert eine Submission als fehlgeschlagen.
+        Bei requeue=True und attempt_count < max_attempts: zurück auf 'pending'.
+        """
+        row = self.db.execute(
+            "SELECT attempt_count, max_attempts FROM submission_queue WHERE id = ?",
+            (submission_id,),
+        ).fetchone()
+
+        if not row:
+            return
+
+        attempt_count, max_attempts = row[0], row[1]
+
+        if requeue and attempt_count < max_attempts:
+            # Zurück in die Queue für Retry
+            self.db.execute(
+                """
+                UPDATE submission_queue
+                SET status = 'pending', last_error = ?, started_at = NULL
+                WHERE id = ?
+                """,
+                (error, submission_id),
+            )
+            logger.info(
+                "Submission %d fehlgeschlagen (Versuch %d/%d) — requeued",
+                submission_id,
+                attempt_count,
+                max_attempts,
+            )
+        else:
+            # Endgültig fehlgeschlagen
+            self.db.execute(
+                """
+                UPDATE submission_queue
+                SET status = 'failed', last_error = ?, completed_at = datetime('now')
+                WHERE id = ?
+                """,
+                (error, submission_id),
+            )
+            logger.error(
+                "Submission %d ENDGÜLTIG fehlgeschlagen (Versuch %d/%d): %s",
+                submission_id,
+                attempt_count,
+                max_attempts,
+                error,
+            )
+        self.db.commit()
+
+    def get_queue_stats(self) -> dict:
+        """Gibt Statistiken über die Queue zurück."""
+        stats = {}
+        for status in ("pending", "processing", "completed", "failed"):
+            count = self.db.execute(
+                "SELECT COUNT(*) FROM submission_queue WHERE status = ?",
+                (status,),
+            ).fetchone()[0]
+            stats[status] = count
+        return stats
+
+    def dedup_check(self, message_id: str) -> Optional[str]:
+        """
+        Prüft, ob eine message_id bereits existiert.
+
+        Returns:
+            Den Status ('pending', 'processing', 'completed', 'failed') oder None.
+        """
+        row = self.db.execute(
+            "SELECT status FROM submission_queue WHERE message_id = ?",
+            (message_id,),
+        ).fetchone()
+        return row[0] if row else None

queue_worker.py

@@ -0,0 +1,243 @@
+#!/usr/bin/env python3
+"""
+Queue-Worker — Verarbeitet Submissions aus der SQLite-Queue.
+=============================================================
+
+Läuft als Hintergrundprozess, pollt die Queue und verarbeitet
+Submissions. Nutzt die bestehende app.py-Logik (importiert als Modul).
+
+Usage:
+  python3 queue_worker.py
+  # oder als systemd-Service
+
+Architektur:
+  Webhook → Signatur-Prüfung → Enqueue → 200 OK
+  Worker → Dequeue → Verarbeiten → Mark completed/failed
+"""
+
+import os
+import sys
+import time
+import logging
+import sqlite3
+from pathlib import Path
+
+# ── Logging ──────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[logging.StreamHandler()],
+)
+logger = logging.getLogger("queue_worker")
+
+# ── Pfade ────────────────────────────────────────────────────────────
+BASE_DIR = Path(__file__).parent
+sys.path.insert(0, str(BASE_DIR))
+
+# ── DB-Verbindung ────────────────────────────────────────────────────
+DB_PATH = os.environ.get("DATABASE_PATH", str(BASE_DIR / "bsn_intake.db"))
+
+
+def get_db() -> sqlite3.Connection:
+    """Erstellt eine neue DB-Verbindung (Worker-eigener Thread)."""
+    db = sqlite3.connect(DB_PATH)
+    db.row_factory = sqlite3.Row
+    db.execute("PRAGMA journal_mode=WAL")
+    db.execute("PRAGMA busy_timeout=5000")
+    return db
+
+
+def process_whatsapp_submission(db: sqlite3.Connection, payload: dict) -> str:
+    """
+    Verarbeitet EINE WhatsApp-Submission.
+    Extrahiert die Message-Daten und ruft die bestehende
+    Verarbeitungslogik auf (gleicher Code wie in app.py).
+
+    Returns: 'ok' oder Fehlermeldung.
+    """
+    from app import get_db as app_get_db
+
+    try:
+        entries = payload.get("entry", [])
+        processed = 0
+
+        for entry in entries:
+            changes = entry.get("changes", [])
+            for change in changes:
+                value = change.get("value", {})
+                messages = value.get("messages", [])
+
+                for msg in messages:
+                    msg_id = msg.get("id", "unknown")
+                    sender_id = msg.get("from", "unknown")
+                    msg_type = msg.get("type", "text")
+
+                    # ── Bestehende Verarbeitung aus app.py ──────────
+                    # (vereinfacht — die volle Logik bleibt in app.py,
+                    #  der Worker ruft sie via Import auf)
+
+                    # Speichern in submissions-Tabelle
+                    from datetime import datetime
+                    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+                    content = ""
+                    if msg_type == "text":
+                        content = msg.get("text", {}).get("body", "")
+                    elif msg_type == "image":
+                        content = msg.get("image", {}).get("caption", "[Bild ohne Text]")
+                    elif msg_type == "audio":
+                        content = "[Sprachnachricht]"
+                    elif msg_type == "video":
+                        content = msg.get("video", {}).get("caption", "[Video]")
+                    elif msg_type == "document":
+                        content = msg.get("document", {}).get("caption", "[Dokument]")
+                    else:
+                        content = f"[{msg_type}]"
+
+                    db.execute(
+                        """INSERT INTO submissions
+                           (sender_id, sender_name, source, type, content, payload_json, created_at)
+                           VALUES (?, ?, ?, ?, ?, ?, ?)""",
+                        (
+                            sender_id,
+                            "WhatsApp User",
+                            "whatsapp",
+                            msg_type,
+                            content[:500],
+                            str(payload)[:5000],
+                            now,
+                        ),
+                    )
+                    db.commit()
+                    processed += 1
+                    logger.info("WhatsApp-Submission verarbeitet: %s (%s)", msg_id, msg_type)
+
+        return "ok" if processed > 0 else "no_messages"
+
+    except Exception as e:
+        logger.exception("Fehler bei WhatsApp-Verarbeitung: %s", e)
+        return f"error: {e}"
+
+
+def process_telegram_submission(db: sqlite3.Connection, payload: dict) -> str:
+    """
+    Verarbeitet EINE Telegram-Submission.
+    """
+    try:
+        msg = payload.get("message", {})
+        if not msg:
+            return "no_message"
+
+        chat = msg.get("chat", {})
+        sender_id = str(chat.get("id", "unknown"))
+        msg_id = str(msg.get("message_id", "unknown"))
+
+        content = ""
+        msg_type = "text"
+        if "text" in msg:
+            content = msg["text"]
+        elif "photo" in msg:
+            content = msg.get("caption", "[Foto]")
+            msg_type = "image"
+        elif "voice" in msg:
+            content = "[Sprachnachricht]"
+            msg_type = "audio"
+        elif "sticker" in msg:
+            content = "[Sticker]"
+            msg_type = "document"
+        else:
+            content = "[Unbekannt]"
+
+        from datetime import datetime
+        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+        db.execute(
+            """INSERT INTO submissions
+               (sender_id, sender_name, source, type, content, payload_json, created_at)
+               VALUES (?, ?, ?, ?, ?, ?, ?)""",
+            (
+                sender_id,
+                "Telegram User",
+                "telegram",
+                msg_type,
+                content[:500],
+                str(payload)[:5000],
+                now,
+            ),
+        )
+        db.commit()
+        logger.info("Telegram-Submission verarbeitet: %s (%s)", msg_id, msg_type)
+        return "ok"
+
+    except Exception as e:
+        logger.exception("Fehler bei Telegram-Verarbeitung: %s", e)
+        return f"error: {e}"
+
+
+def run_worker(poll_interval: int = 2, batch_size: int = 5) -> None:
+    """
+    Haupt-Worker-Loop: Pollt die Queue und verarbeitet Submissions.
+
+    Args:
+        poll_interval: Sekunden zwischen Polls.
+        batch_size: Maximale Submissions pro Batch.
+    """
+    from queue_handler import QueueHandler
+
+    db = get_db()
+    qh = QueueHandler(db)
+
+    logger.info("🚀 Queue-Worker gestartet (Poll=%ds, Batch=%d)", poll_interval, batch_size)
+
+    while True:
+        try:
+            submissions = qh.dequeue(limit=batch_size)
+
+            if not submissions:
+                time.sleep(poll_interval)
+                continue
+
+            for sub in submissions:
+                logger.info(
+                    "Verarbeite #%d: channel=%s message_id=%s attempt=%d",
+                    sub["id"],
+                    sub["channel"],
+                    sub["message_id"],
+                    sub["attempt_count"],
+                )
+
+                try:
+                    if sub["channel"] == "whatsapp":
+                        result = process_whatsapp_submission(db, sub["payload"])
+                    elif sub["channel"] == "telegram":
+                        result = process_telegram_submission(db, sub["payload"])
+                    else:
+                        result = f"unknown_channel: {sub['channel']}"
+
+                    if result == "ok":
+                        qh.mark_completed(sub["id"])
+                        logger.info("✅ #%d abgeschlossen", sub["id"])
+                    else:
+                        qh.mark_failed(sub["id"], result, requeue=True)
+                        logger.warning("⚠️ #%d fehlgeschlagen: %s", sub["id"], result)
+
+                except Exception as e:
+                    error_msg = f"{type(e).__name__}: {e}"
+                    qh.mark_failed(sub["id"], error_msg, requeue=True)
+                    logger.exception("💥 #%d Ausnahme: %s", sub["id"], error_msg)
+
+        except Exception as e:
+            logger.exception("Worker-Loop-Fehler: %s", e)
+            time.sleep(5)  # Bei DB-Fehlern kurz warten
+
+
+# ── Main ─────────────────────────────────────────────────────────────
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="BSN Queue Worker")
+    parser.add_argument("--poll", type=int, default=2, help="Poll-Intervall in Sekunden")
+    parser.add_argument("--batch", type=int, default=5, help="Max Batch-Größe")
+    args = parser.parse_args()
+
+    run_worker(poll_interval=args.poll, batch_size=args.batch)

webhook_queue.py

@@ -0,0 +1,225 @@
+#!/usr/bin/env python3
+"""
+Webhook-Queue-Integration — Signatur-Prüfung + Enqueue.
+=======================================================
+
+Wird in app.py importiert und wickelt die Webhook-Routen ab.
+Ersetzt NICHT die bestehende Logik, sondern sitzt DAVOR.
+
+Flow:
+  1. Webhook empfängt Request
+  2. Signatur wird geprüft (WhatsApp X-Hub-Signature-256)
+  3. Message-IDs werden extrahiert
+  4. Submission wird in Queue eingereiht (SQLite UNIQUE = Dedup)
+  5. 200 OK wird sofort zurückgegeben
+  6. Worker (queue_worker.py) verarbeitet asynchron
+
+Usage in app.py:
+  from webhook_queue import whatsapp_intake, telegram_intake
+
+  @app.route("/whatsapp/webhook", methods=["GET", "POST"])
+  def whatsapp_webhook():
+      ...
+      return whatsapp_intake(request, db)
+
+  @app.route("/telegram/webhook", methods=["GET", "POST"])
+  def telegram_webhook():
+      ...
+      return telegram_intake(request, db)
+"""
+
+import hashlib
+import hmac
+import logging
+import os
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# ── Konfiguration ────────────────────────────────────────────────────
+META_APP_SECRET = os.environ.get("META_APP_SECRET", "")
+# Falls kein App Secret gesetzt → Signatur-Prüfung überspringen (Test-Modus)
+
+
+# ── Signatur-Prüfung ─────────────────────────────────────────────────
+
+def verify_whatsapp_signature(
+    raw_body: bytes,
+    signature_header: Optional[str],
+) -> bool:
+    """
+    Prüft die X-Hub-Signature-256 von Meta.
+
+    Args:
+        raw_body: Der rohe Request-Body als Bytes.
+        signature_header: Wert des X-Hub-Signature-256-Headers.
+
+    Returns:
+        True wenn gültig oder Test-Modus (kein Secret).
+    """
+    if not META_APP_SECRET:
+        logger.debug("Kein META_APP_SECRET → Signatur-Prüfung übersprungen")
+        return True
+
+    if not signature_header:
+        logger.warning("X-Hub-Signature-256 Header fehlt")
+        return False
+
+    if not signature_header.startswith("sha256="):
+        logger.warning("Unbekanntes Signatur-Format: %s...", signature_header[:30])
+        return False
+
+    expected_hex = signature_header[7:]  # Nach "sha256="
+    computed_hmac = hmac.new(
+        key=META_APP_SECRET.encode("utf-8"),
+        msg=raw_body,
+        digestmod=hashlib.sha256,
+    )
+    computed_hex = computed_hmac.hexdigest()
+
+    if not hmac.compare_digest(computed_hex, expected_hex):
+        logger.warning(
+            "❌ WhatsApp-Signatur UNGÜLTIG — möglicher Replay-Angriff"
+        )
+        return False
+
+    logger.debug("✅ WhatsApp-Signatur gültig")
+    return True
+
+
+# ── Intake-Funktionen ────────────────────────────────────────────────
+
+def whatsapp_intake(request_obj, db) -> tuple:
+    """
+    Verarbeitet EINE WhatsApp-Webhook-Anfrage.
+
+    Args:
+        request_obj: Flask request-Objekt.
+        db: SQLite-Datenbankverbindung.
+
+    Returns:
+        (body, status_code) — IMMER 200 OK.
+    """
+    from queue_handler import QueueHandler
+
+    qh = QueueHandler(db)
+
+    # ── 1. Signatur prüfen ──────────────────────────────────────────
+    raw_body = request_obj.get_data()
+    signature = request_obj.headers.get("X-Hub-Signature-256", "")
+
+    if not verify_whatsapp_signature(raw_body, signature):
+        logger.warning("WhatsApp-Request mit ungültiger Signatur abgelehnt")
+        return "Forbidden", 403
+
+    # ── 2. Body parsen ──────────────────────────────────────────────
+    body = request_obj.get_json(force=True, silent=True)
+    if not body:
+        return "OK", 200
+
+    entries = body.get("entry", [])
+    enqueued = 0
+    duplicates = 0
+
+    # ── 3. Jede Nachricht in Queue ──────────────────────────────────
+    for entry in entries:
+        changes = entry.get("changes", [])
+        for change in changes:
+            value = change.get("value", {})
+            messages = value.get("messages", [])
+
+            for msg in messages:
+                msg_id = msg.get("id", "")
+                if not msg_id:
+                    continue
+
+                # Payload bauen (nur das relevante message-Objekt)
+                payload = {
+                    "entry": [{
+                        "id": entry.get("id", ""),
+                        "changes": [{
+                            "value": {
+                                "messaging_product": value.get("messaging_product", "whatsapp"),
+                                "metadata": value.get("metadata", {}),
+                                "contacts": value.get("contacts", []),
+                                "messages": [msg],
+                            }
+                        }]
+                    }]
+                }
+
+                success, status = qh.enqueue(
+                    channel="whatsapp",
+                    message_id=msg_id,
+                    payload=payload,
+                    raw_body=raw_body,
+                )
+
+                if success:
+                    enqueued += 1
+                else:
+                    duplicates += 1
+                    logger.info("WhatsApp-Duplikat: %s → %s", msg_id, status)
+
+    logger.info(
+        "WhatsApp-Intake: %d enqueued, %d duplicates",
+        enqueued,
+        duplicates,
+    )
+    return "OK", 200
+
+
+def telegram_intake(request_obj, db) -> tuple:
+    """
+    Verarbeitet EINE Telegram-Webhook-Anfrage.
+
+    Args:
+        request_obj: Flask request-Objekt.
+        db: SQLite-Datenbankverbindung.
+
+    Returns:
+        (body, status_code) — IMMER 200 OK.
+    """
+    from queue_handler import QueueHandler
+
+    qh = QueueHandler(db)
+
+    body = request_obj.get_json(force=True, silent=True)
+    if not body:
+        return "OK", 200
+
+    msg = body.get("message", {})
+    if not msg:
+        # Kein Message-Objekt (z.B. edited_message, callback_query)
+        update_id = str(body.get("update_id", "unknown"))
+        logger.debug("Telegram-Update ohne Message: update_id=%s", update_id)
+        return "OK", 200
+
+    msg_id = str(msg.get("message_id", ""))
+    if not msg_id:
+        return "OK", 200
+
+    # Payload ist das gesamte Update (enthält update_id + message)
+    success, status = qh.enqueue(
+        channel="telegram",
+        message_id=msg_id,
+        payload=body,
+        raw_body=request_obj.get_data(),
+    )
+
+    if success:
+        logger.info("Telegram-Intake: %s enqueued", msg_id)
+    else:
+        logger.info("Telegram-Duplikat: %s → %s", msg_id, status)
+
+    return "OK", 200
+
+
+# ── Queue-Status (für Monitoring) ────────────────────────────────────
+
+def get_queue_status(db) -> dict:
+    """Gibt Queue-Statistiken zurück (für /admin/queue Endpoint)."""
+    from queue_handler import QueueHandler
+
+    qh = QueueHandler(db)
+    return qh.get_queue_stats()