Trixy7/trixy_core/events/dlq/queue.py

# -*- coding: utf-8 -*-
"""
Dead Letter Queue Implementation.

Speichert und verwaltet fehlgeschlagene Events.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, Coroutine
import uuid

if TYPE_CHECKING:
    from trixy_core.events.event_data.base import EventData


class DeadLetterReason(Enum):
    """
    Grund für Ablage in DLQ.
    """

    HANDLER_ERROR = auto()
    """Fehler im Event-Handler."""

    TIMEOUT = auto()
    """Timeout bei Verarbeitung."""

    VALIDATION_FAILED = auto()
    """Validierung fehlgeschlagen."""

    MAX_RETRIES = auto()
    """Maximale Wiederholungen erreicht."""

    REJECTED = auto()
    """Explizit abgelehnt."""

    UNKNOWN = auto()
    """Unbekannter Grund."""


@dataclass
class DeadLetterEntry:
    """
    Eintrag in der Dead Letter Queue.
    """

    id: str
    """Eindeutige ID des Eintrags."""

    event_name: str
    """Name des Events."""

    event_data: dict[str, Any]
    """Serialisierte Event-Daten."""

    reason: DeadLetterReason
    """Grund für DLQ."""

    error_message: str
    """Fehlermeldung."""

    error_type: str = ""
    """Typ des Fehlers."""

    created_at: datetime = field(default_factory=datetime.now)
    """Zeitpunkt der Erstellung."""

    retry_count: int = 0
    """Anzahl der Wiederholungsversuche."""

    last_retry_at: datetime | None = None
    """Zeitpunkt des letzten Retry."""

    next_retry_at: datetime | None = None
    """Geplanter nächster Retry."""

    metadata: dict[str, Any] = field(default_factory=dict)
    """Zusätzliche Metadaten."""

    @classmethod
    def from_event(
        cls,
        event_name: str,
        event_data: "EventData",
        reason: DeadLetterReason,
        error: Exception | None = None,
    ) -> "DeadLetterEntry":
        """
        Erstellt einen Eintrag aus Event-Daten.

        Args:
            event_name: Name des Events.
            event_data: Die Event-Daten.
            reason: Grund für DLQ.
            error: Optionale Exception.

        Returns:
            DeadLetterEntry.
        """
        # Event-Daten serialisieren
        data = {}
        for key, value in event_data.__dict__.items():
            if not key.startswith("_"):
                if isinstance(value, datetime):
                    data[key] = value.isoformat()
                elif hasattr(value, "__dict__"):
                    data[key] = str(value)
                else:
                    data[key] = value

        return cls(
            id=str(uuid.uuid4()),
            event_name=event_name,
            event_data=data,
            reason=reason,
            error_message=str(error) if error else "",
            error_type=type(error).__name__ if error else "",
            metadata={
                "original_event_id": event_data.event_id,
                "original_source": event_data.source,
            },
        )

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "id": self.id,
            "event_name": self.event_name,
            "event_data": self.event_data,
            "reason": self.reason.name,
            "error_message": self.error_message,
            "error_type": self.error_type,
            "created_at": self.created_at.isoformat(),
            "retry_count": self.retry_count,
            "last_retry_at": self.last_retry_at.isoformat() if self.last_retry_at else None,
            "next_retry_at": self.next_retry_at.isoformat() if self.next_retry_at else None,
            "metadata": self.metadata,
        }


@dataclass
class DLQConfig:
    """
    Konfiguration für die Dead Letter Queue.
    """

    max_size: int = 10000
    """Maximale Anzahl Einträge."""

    max_retries: int = 3
    """Maximale Wiederholungsversuche."""

    retry_delay_seconds: float = 60.0
    """Basis-Verzögerung zwischen Retries."""

    auto_retry: bool = True
    """Automatische Wiederholung aktivieren."""

    retention_hours: int = 168  # 1 Woche
    """Aufbewahrungsdauer in Stunden."""

    on_max_retries: Callable[[DeadLetterEntry], Coroutine[Any, Any, None]] | None = None
    """Callback bei erreichten max Retries."""


@dataclass
class DLQStats:
    """
    Statistiken der Dead Letter Queue.
    """

    total_entries: int = 0
    """Gesamtzahl Einträge."""

    pending_retries: int = 0
    """Wartende Retries."""

    max_retried: int = 0
    """Einträge mit max Retries erreicht."""

    by_reason: dict[str, int] = field(default_factory=dict)
    """Einträge nach Grund."""

    by_event: dict[str, int] = field(default_factory=dict)
    """Einträge nach Event-Typ."""

    oldest_entry: datetime | None = None
    """Ältester Eintrag."""

    newest_entry: datetime | None = None
    """Neuester Eintrag."""


class DeadLetterQueue:
    """
    Dead Letter Queue für fehlgeschlagene Events.

    Speichert Events die nicht verarbeitet werden konnten
    und ermöglicht Wiederholungsversuche.

    Example:
        dlq = DeadLetterQueue(DLQConfig(
            max_retries=5,
            retry_delay_seconds=300,
        ))

        # Fehlerhafte Events hinzufügen
        try:
            await process_event(event_data)
        except Exception as e:
            await dlq.add("my_event", event_data, e)

        # Retry-Handler registrieren
        dlq.set_retry_handler(process_event)

        # Automatische Retries starten
        await dlq.start_retry_worker()
    """

    def __init__(
        self,
        config: DLQConfig | None = None,
    ) -> None:
        """
        Initialisiert die DLQ.

        Args:
            config: Konfiguration.
        """
        self._config = config or DLQConfig()
        self._entries: dict[str, DeadLetterEntry] = {}
        self._retry_handler: Callable[[str, dict[str, Any]], Coroutine[Any, Any, None]] | None = None
        self._worker_task: asyncio.Task | None = None
        self._lock = asyncio.Lock()

    @property
    def size(self) -> int:
        """Anzahl der Einträge."""
        return len(self._entries)

    def set_retry_handler(
        self,
        handler: Callable[[str, dict[str, Any]], Coroutine[Any, Any, None]],
    ) -> None:
        """
        Setzt den Handler für Retry-Versuche.

        Args:
            handler: Handler-Funktion (event_name, event_data).
        """
        self._retry_handler = handler

    async def add(
        self,
        event_name: str,
        event_data: "EventData",
        error: Exception | None = None,
        reason: DeadLetterReason = DeadLetterReason.HANDLER_ERROR,
    ) -> DeadLetterEntry:
        """
        Fügt ein Event zur DLQ hinzu.

        Args:
            event_name: Name des Events.
            event_data: Die Event-Daten.
            error: Optionale Exception.
            reason: Grund für DLQ.

        Returns:
            Der erstellte Eintrag.
        """
        async with self._lock:
            # Cleanup wenn voll
            if len(self._entries) >= self._config.max_size:
                await self._cleanup()

            entry = DeadLetterEntry.from_event(
                event_name=event_name,
                event_data=event_data,
                reason=reason,
                error=error,
            )

            # Nächsten Retry planen
            if self._config.auto_retry:
                entry.next_retry_at = datetime.now() + timedelta(
                    seconds=self._config.retry_delay_seconds
                )

            self._entries[entry.id] = entry
            return entry

    async def add_entry(self, entry: DeadLetterEntry) -> None:
        """
        Fügt einen Eintrag direkt hinzu.

        Args:
            entry: Der Eintrag.
        """
        async with self._lock:
            self._entries[entry.id] = entry

    async def get(self, entry_id: str) -> DeadLetterEntry | None:
        """
        Gibt einen Eintrag zurück.

        Args:
            entry_id: Die Eintrags-ID.

        Returns:
            DeadLetterEntry oder None.
        """
        return self._entries.get(entry_id)

    async def remove(self, entry_id: str) -> bool:
        """
        Entfernt einen Eintrag.

        Args:
            entry_id: Die Eintrags-ID.

        Returns:
            True wenn entfernt.
        """
        async with self._lock:
            return self._entries.pop(entry_id, None) is not None

    async def get_pending_retries(self) -> list[DeadLetterEntry]:
        """
        Gibt Einträge zurück, die für Retry bereit sind.

        Returns:
            Liste der Einträge.
        """
        now = datetime.now()
        pending = []

        for entry in self._entries.values():
            if (
                entry.retry_count < self._config.max_retries
                and entry.next_retry_at
                and entry.next_retry_at <= now
            ):
                pending.append(entry)

        return pending

    async def retry(self, entry_id: str) -> bool:
        """
        Führt einen Retry für einen Eintrag durch.

        Args:
            entry_id: Die Eintrags-ID.

        Returns:
            True wenn erfolgreich.
        """
        if not self._retry_handler:
            raise RuntimeError("Kein Retry-Handler registriert")

        entry = await self.get(entry_id)
        if not entry:
            return False

        async with self._lock:
            entry.retry_count += 1
            entry.last_retry_at = datetime.now()

        try:
            await self._retry_handler(entry.event_name, entry.event_data)

            # Erfolg - Eintrag entfernen
            await self.remove(entry_id)
            return True

        except Exception as e:
            async with self._lock:
                entry.error_message = str(e)
                entry.error_type = type(e).__name__

                if entry.retry_count >= self._config.max_retries:
                    entry.next_retry_at = None
                    entry.reason = DeadLetterReason.MAX_RETRIES

                    if self._config.on_max_retries:
                        await self._config.on_max_retries(entry)
                else:
                    # Nächsten Retry planen
                    delay = self._config.retry_delay_seconds * (2 ** entry.retry_count)
                    entry.next_retry_at = datetime.now() + timedelta(seconds=delay)

            return False

    async def retry_all_pending(self) -> tuple[int, int]:
        """
        Führt Retry für alle pending Einträge durch.

        Returns:
            Tupel (erfolgreich, fehlgeschlagen).
        """
        pending = await self.get_pending_retries()
        success = 0
        failed = 0

        for entry in pending:
            if await self.retry(entry.id):
                success += 1
            else:
                failed += 1

        return success, failed

    async def start_retry_worker(
        self,
        interval_seconds: float = 60.0,
    ) -> None:
        """
        Startet den Retry-Worker.

        Args:
            interval_seconds: Prüfintervall in Sekunden.
        """
        if self._worker_task and not self._worker_task.done():
            return

        async def worker():
            while True:
                try:
                    await self.retry_all_pending()
                    await self._cleanup_expired()
                except Exception:
                    pass

                await asyncio.sleep(interval_seconds)

        self._worker_task = asyncio.create_task(worker())

    def stop_retry_worker(self) -> None:
        """Stoppt den Retry-Worker."""
        if self._worker_task:
            self._worker_task.cancel()
            self._worker_task = None

    async def _cleanup(self) -> int:
        """
        Entfernt älteste Einträge.

        Returns:
            Anzahl entfernter Einträge.
        """
        # Älteste 10% entfernen
        to_remove = int(self._config.max_size * 0.1)
        sorted_entries = sorted(
            self._entries.values(),
            key=lambda e: e.created_at,
        )

        removed = 0
        for entry in sorted_entries[:to_remove]:
            del self._entries[entry.id]
            removed += 1

        return removed

    async def _cleanup_expired(self) -> int:
        """
        Entfernt abgelaufene Einträge.

        Returns:
            Anzahl entfernter Einträge.
        """
        cutoff = datetime.now() - timedelta(hours=self._config.retention_hours)
        expired = [
            entry_id
            for entry_id, entry in self._entries.items()
            if entry.created_at < cutoff
        ]

        for entry_id in expired:
            del self._entries[entry_id]

        return len(expired)

    async def clear(self) -> int:
        """
        Löscht alle Einträge.

        Returns:
            Anzahl gelöschter Einträge.
        """
        async with self._lock:
            count = len(self._entries)
            self._entries.clear()
            return count

    def get_stats(self) -> DLQStats:
        """
        Gibt Statistiken zurück.

        Returns:
            DLQStats.
        """
        stats = DLQStats()
        stats.total_entries = len(self._entries)

        now = datetime.now()
        oldest: datetime | None = None
        newest: datetime | None = None

        for entry in self._entries.values():
            # Pending Retries
            if entry.next_retry_at and entry.next_retry_at <= now:
                stats.pending_retries += 1

            # Max Retried
            if entry.retry_count >= self._config.max_retries:
                stats.max_retried += 1

            # Nach Grund
            reason_name = entry.reason.name
            stats.by_reason[reason_name] = stats.by_reason.get(reason_name, 0) + 1

            # Nach Event
            stats.by_event[entry.event_name] = stats.by_event.get(entry.event_name, 0) + 1

            # Ältester/Neuester
            if oldest is None or entry.created_at < oldest:
                oldest = entry.created_at
            if newest is None or entry.created_at > newest:
                newest = entry.created_at

        stats.oldest_entry = oldest
        stats.newest_entry = newest

        return stats

    async def get_by_event(self, event_name: str) -> list[DeadLetterEntry]:
        """
        Gibt Einträge für ein Event zurück.

        Args:
            event_name: Name des Events.

        Returns:
            Liste der Einträge.
        """
        return [
            entry
            for entry in self._entries.values()
            if entry.event_name == event_name
        ]

    async def get_by_reason(self, reason: DeadLetterReason) -> list[DeadLetterEntry]:
        """
        Gibt Einträge nach Grund zurück.

        Args:
            reason: Der Grund.

        Returns:
            Liste der Einträge.
        """
        return [
            entry
            for entry in self._entries.values()
            if entry.reason == reason
        ]

    async def export_all(self) -> list[dict[str, Any]]:
        """
        Exportiert alle Einträge.

        Returns:
            Liste von Dictionaries.
        """
        return [entry.to_dict() for entry in self._entries.values()]