Trixy7/trixy_core/events/middleware/timing.py

# -*- coding: utf-8 -*-
"""
Timing-Middleware für Performance-Messung.

Misst und protokolliert Event-Verarbeitungszeiten.
"""

from __future__ import annotations

import statistics
import time
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Any, Callable

from trixy_core.events.middleware.base import (
    EventMiddleware,
    MiddlewareContext,
    MiddlewareResult,
    MiddlewareAction,
)

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


@dataclass
class TimingStats:
    """
    Statistiken für Event-Verarbeitungszeiten.
    """

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

    count: int = 0
    """Anzahl der Verarbeitungen."""

    total_ms: float = 0.0
    """Gesamtzeit in Millisekunden."""

    min_ms: float = float("inf")
    """Minimale Zeit."""

    max_ms: float = 0.0
    """Maximale Zeit."""

    last_ms: float = 0.0
    """Letzte Verarbeitungszeit."""

    _times: list[float] = field(default_factory=list, repr=False)
    """Alle Zeiten für Perzentil-Berechnung."""

    @property
    def avg_ms(self) -> float:
        """Durchschnittliche Zeit in Millisekunden."""
        if self.count == 0:
            return 0.0
        return self.total_ms / self.count

    @property
    def p50_ms(self) -> float:
        """50. Perzentil (Median)."""
        if not self._times:
            return 0.0
        return statistics.median(self._times)

    @property
    def p90_ms(self) -> float:
        """90. Perzentil."""
        if not self._times:
            return 0.0
        return self._get_percentile(90)

    @property
    def p99_ms(self) -> float:
        """99. Perzentil."""
        if not self._times:
            return 0.0
        return self._get_percentile(99)

    def _get_percentile(self, percentile: float) -> float:
        """
        Berechnet ein Perzentil.

        Args:
            percentile: Perzentil (0-100).

        Returns:
            Wert am Perzentil.
        """
        if not self._times:
            return 0.0

        sorted_times = sorted(self._times)
        index = (percentile / 100) * (len(sorted_times) - 1)
        lower = int(index)
        upper = min(lower + 1, len(sorted_times) - 1)
        fraction = index - lower

        return sorted_times[lower] * (1 - fraction) + sorted_times[upper] * fraction

    def record(self, duration_ms: float) -> None:
        """
        Zeichnet eine Verarbeitungszeit auf.

        Args:
            duration_ms: Dauer in Millisekunden.
        """
        self.count += 1
        self.total_ms += duration_ms
        self.last_ms = duration_ms
        self.min_ms = min(self.min_ms, duration_ms)
        self.max_ms = max(self.max_ms, duration_ms)

        # Begrenzte Historie für Perzentile
        self._times.append(duration_ms)
        if len(self._times) > 1000:
            self._times = self._times[-500:]

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "event_name": self.event_name,
            "count": self.count,
            "total_ms": round(self.total_ms, 2),
            "avg_ms": round(self.avg_ms, 2),
            "min_ms": round(self.min_ms, 2) if self.min_ms != float("inf") else 0,
            "max_ms": round(self.max_ms, 2),
            "p50_ms": round(self.p50_ms, 2),
            "p90_ms": round(self.p90_ms, 2),
            "p99_ms": round(self.p99_ms, 2),
            "last_ms": round(self.last_ms, 2),
        }


class TimingMiddleware(EventMiddleware):
    """
    Middleware für Event-Timing-Messungen.

    Misst die Verarbeitungszeit jedes Events und sammelt
    Statistiken für Performance-Analyse.

    Example:
        middleware = TimingMiddleware(
            slow_threshold_ms=100,
            on_slow_event=lambda name, ms: print(f"SLOW: {name} took {ms}ms"),
        )
        pipeline.add(middleware)

        # Später: Statistiken abrufen
        stats = middleware.get_stats("my_event")
        print(f"Average: {stats.avg_ms}ms")
    """

    TIMING_KEY = "_timing_start_ns"

    def __init__(
        self,
        name: str = "TimingMiddleware",
        enabled: bool = True,
        slow_threshold_ms: float = 100.0,
        on_slow_event: Callable[[str, float], None] | None = None,
        track_all: bool = True,
        max_events: int = 1000,
    ) -> None:
        """
        Initialisiert die Timing-Middleware.

        Args:
            name: Name der Middleware.
            enabled: Ob aktiv.
            slow_threshold_ms: Schwellenwert für langsame Events (ms).
            on_slow_event: Callback für langsame Events (name, duration_ms).
            track_all: Ob alle Events erfasst werden.
            max_events: Maximale Anzahl erfasster Event-Typen.
        """
        super().__init__(name, enabled)
        self._slow_threshold_ms = slow_threshold_ms
        self._on_slow_event = on_slow_event
        self._track_all = track_all
        self._max_events = max_events

        self._stats: dict[str, TimingStats] = {}
        self._global_stats = TimingStats(event_name="__global__")

    @property
    def slow_threshold_ms(self) -> float:
        """Schwellenwert für langsame Events."""
        return self._slow_threshold_ms

    @slow_threshold_ms.setter
    def slow_threshold_ms(self, value: float) -> None:
        """Setzt den Schwellenwert."""
        self._slow_threshold_ms = value

    def get_stats(self, event_name: str) -> TimingStats | None:
        """
        Gibt die Statistiken für ein Event zurück.

        Args:
            event_name: Name des Events.

        Returns:
            TimingStats oder None.
        """
        return self._stats.get(event_name)

    def get_all_stats(self) -> dict[str, TimingStats]:
        """
        Gibt alle Event-Statistiken zurück.

        Returns:
            Dictionary von Event-Namen zu Statistiken.
        """
        return dict(self._stats)

    def get_global_stats(self) -> TimingStats:
        """
        Gibt die globalen Statistiken zurück.

        Returns:
            Globale TimingStats.
        """
        return self._global_stats

    def get_slowest_events(self, n: int = 10) -> list[tuple[str, float]]:
        """
        Gibt die langsamsten Events zurück.

        Args:
            n: Anzahl der Events.

        Returns:
            Liste von (event_name, avg_ms) Tupeln.
        """
        sorted_events = sorted(
            self._stats.items(),
            key=lambda x: x[1].avg_ms,
            reverse=True,
        )
        return [(name, stats.avg_ms) for name, stats in sorted_events[:n]]

    def reset_stats(self, event_name: str | None = None) -> None:
        """
        Setzt Statistiken zurück.

        Args:
            event_name: Spezifisches Event oder None für alle.
        """
        if event_name:
            if event_name in self._stats:
                del self._stats[event_name]
        else:
            self._stats.clear()
            self._global_stats = TimingStats(event_name="__global__")

    async def before(
        self,
        context: MiddlewareContext,
    ) -> MiddlewareResult:
        """
        Startet die Zeitmessung.

        Args:
            context: Der Middleware-Kontext.

        Returns:
            Ergebnis.
        """
        # Speichere Startzeit im Kontext
        context.set_property(self.TIMING_KEY, time.perf_counter_ns())
        return MiddlewareResult()

    async def after(
        self,
        context: MiddlewareContext,
    ) -> None:
        """
        Beendet die Zeitmessung und zeichnet Statistiken auf.

        Args:
            context: Der Middleware-Kontext.
        """
        start_ns = context.get_property(self.TIMING_KEY)
        if start_ns is None:
            return

        duration_ns = time.perf_counter_ns() - start_ns
        duration_ms = duration_ns / 1_000_000

        # Statistiken aktualisieren
        if self._track_all:
            if context.event_name not in self._stats:
                if len(self._stats) < self._max_events:
                    self._stats[context.event_name] = TimingStats(
                        event_name=context.event_name
                    )

            if context.event_name in self._stats:
                self._stats[context.event_name].record(duration_ms)

        self._global_stats.record(duration_ms)

        # Slow-Event-Callback
        if duration_ms >= self._slow_threshold_ms and self._on_slow_event:
            self._on_slow_event(context.event_name, duration_ms)

        # Speichere Dauer im Kontext für andere Middlewares
        context.set_property("timing_duration_ms", duration_ms)

    def to_dict(self) -> dict[str, Any]:
        """
        Exportiert alle Statistiken als Dictionary.

        Returns:
            Dictionary mit allen Statistiken.
        """
        return {
            "global": self._global_stats.to_dict(),
            "events": {
                name: stats.to_dict()
                for name, stats in self._stats.items()
            },
            "slow_threshold_ms": self._slow_threshold_ms,
            "slowest_events": self.get_slowest_events(5),
        }


class TimeoutMiddleware(EventMiddleware):
    """
    Middleware für Event-Timeouts.

    Bricht Events ab, die zu lange dauern.

    Example:
        middleware = TimeoutMiddleware(
            default_timeout_ms=5000,
            per_event_timeouts={"slow_event": 10000},
        )
    """

    def __init__(
        self,
        name: str = "TimeoutMiddleware",
        enabled: bool = True,
        default_timeout_ms: float = 5000.0,
        per_event_timeouts: dict[str, float] | None = None,
        on_timeout: Callable[[str, float], None] | None = None,
    ) -> None:
        """
        Initialisiert die Timeout-Middleware.

        Args:
            name: Name der Middleware.
            enabled: Ob aktiv.
            default_timeout_ms: Standard-Timeout in Millisekunden.
            per_event_timeouts: Spezifische Timeouts pro Event.
            on_timeout: Callback bei Timeout.
        """
        super().__init__(name, enabled)
        self._default_timeout_ms = default_timeout_ms
        self._per_event_timeouts = per_event_timeouts or {}
        self._on_timeout = on_timeout

    def get_timeout(self, event_name: str) -> float:
        """
        Gibt den Timeout für ein Event zurück.

        Args:
            event_name: Name des Events.

        Returns:
            Timeout in Millisekunden.
        """
        return self._per_event_timeouts.get(event_name, self._default_timeout_ms)

    async def before(
        self,
        context: MiddlewareContext,
    ) -> MiddlewareResult:
        """Speichert Timeout-Informationen."""
        timeout_ms = self.get_timeout(context.event_name)
        context.set_property("_timeout_ms", timeout_ms)
        context.set_property("_timeout_deadline", time.perf_counter_ns() + (timeout_ms * 1_000_000))
        return MiddlewareResult()

    async def after(
        self,
        context: MiddlewareContext,
    ) -> None:
        """Prüft auf Timeout."""
        deadline = context.get_property("_timeout_deadline")
        if deadline is None:
            return

        if time.perf_counter_ns() > deadline:
            timeout_ms = context.get_property("_timeout_ms", 0)
            if self._on_timeout:
                self._on_timeout(context.event_name, context.elapsed_ms)

            # Markiere Event als abgebrochen wegen Timeout
            context.event_data.metadata["timed_out"] = True
            context.event_data.metadata["timeout_ms"] = timeout_ms