feat: QueueHandler mit Redis-Streams, Idempotenz, Dedup

src/queue/handler.py

@@ -0,0 +1,97 @@
+"""QueueHandler mit Redis-Streams – Idempotenz und Dedup via message_id."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+import redis
+import redis.exceptions
+
+from src.models.submission import Submission
+
+
+class QueueHandler:
+    """Redis-Stream-basierte Warteschlange für Submissions."""
+
+    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379) -> None:
+        """Verbindung zu Redis (echt oder fakeredis)."""
+        self.redis: redis.Redis[Any] = redis.Redis(  # type: ignore[type-arg]
+            host=redis_host, port=redis_port, decode_responses=True
+        )
+        self.stream_key: str = "bsn:submissions"
+        self.processed_set: str = "bsn:processed"
+        self.stream_ids_hash: str = "bsn:stream_ids"
+        self.group_name: str = "bsn-workers"
+        self._ensure_consumer_group()
+
+    def _ensure_consumer_group(self) -> None:
+        """Erstellt die Consumer-Group, falls nicht vorhanden."""
+        try:
+            self.redis.xgroup_create(self.stream_key, self.group_name, id="0", mkstream=True)
+        except redis.exceptions.ResponseError as exc:
+            if "BUSYGROUP" not in str(exc):
+                raise
+
+    def enqueue(self, submission: Submission) -> str | None:
+        """Reiht eine Submission in den Stream ein (idempotent via message_id).
+
+        Returns:
+            stream_id bei Erfolg, None wenn bereits verarbeitet (Duplikat).
+        """
+        if self.redis.sismember(self.processed_set, submission.message_id):
+            return None
+
+        data: dict[str, str] = {"submission": submission.model_dump_json()}
+        stream_id: str = cast(str, self.redis.xadd(self.stream_key, data))  # type: ignore[arg-type]
+
+        # Dedup-Merkmal setzen + 24h TTL
+        self.redis.sadd(self.processed_set, submission.message_id)
+        self.redis.expire(self.processed_set, 86400)
+
+        # Mapping für späteres XACK ablegen
+        self.redis.hset(self.stream_ids_hash, submission.message_id, stream_id)
+
+        return stream_id
+
+    def dequeue(self, count: int = 5) -> list[Submission]:
+        """Holt bis zu *count* Submissions aus dem Stream (Consumer-Group).
+
+        Returns:
+            Liste von Submission-Objekten.
+        """
+        results: Any = self.redis.xreadgroup(
+            self.group_name,
+            "bsn-consumer",
+            {self.stream_key: ">"},
+            count=count,
+            block=1000,
+        )
+        submissions: list[Submission] = []
+        if not results:
+            return submissions
+
+        for _stream_name, messages in results:
+            for _msg_id, fields in messages:
+                sub_json: str = cast(str, fields.get("submission", "{}"))
+                submissions.append(Submission.model_validate_json(sub_json))
+
+        return submissions
+
+    def mark_done(self, submission: Submission) -> None:
+        """Bestätigt Verarbeitung: XACK und Eintrag im processed_set."""
+        # Stream-ID aus Redis-Hash holen
+        stream_id: str | None = cast(
+            str | None, self.redis.hget(self.stream_ids_hash, submission.message_id)
+        )
+        if stream_id:
+            self.redis.xack(self.stream_key, self.group_name, stream_id)
+            # Bereinige das Mapping
+            self.redis.hdel(self.stream_ids_hash, submission.message_id)
+
+        # Idempotent in processed_set (falls noch nicht drin)
+        self.redis.sadd(self.processed_set, submission.message_id)
+        self.redis.expire(self.processed_set, 86400)
+
+    def queue_size(self) -> int:
+        """Gibt die Anzahl der Nachrichten im Stream zurück."""
+        return self.redis.xlen(self.stream_key)