Hermes Agent
a5eb0e7d27
Quality Gate / 🛡️ Lint + Type Check + Tests (push) Failing after 11m21s
- 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)
312 lines
10 KiB
Python
312 lines
10 KiB
Python
#!/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
|