Trixy7/trixy_core/events/persistence/store.py

# -*- coding: utf-8 -*-
"""
Abstrakte EventStore-Klasse.

Definiert die Schnittstelle für Event-Speicherung.
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Any, Iterator
import json
import uuid

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


@dataclass
class StoredEvent:
    """
    Ein gespeichertes Event.
    """

    id: str
    """Eindeutige ID."""

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

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

    timestamp: datetime
    """Zeitpunkt des Events."""

    source: str = ""
    """Quelle des Events."""

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

    sequence: int = 0
    """Sequenznummer für Reihenfolge."""

    @classmethod
    def from_event(
        cls,
        event_name: str,
        event_data: "EventData",
        sequence: int = 0,
    ) -> "StoredEvent":
        """
        Erstellt ein StoredEvent aus Event-Daten.

        Args:
            event_name: Name des Events.
            event_data: Die Event-Daten.
            sequence: Sequenznummer.

        Returns:
            StoredEvent.
        """
        # 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=event_data.event_id,
            event_name=event_name,
            data=data,
            timestamp=event_data.timestamp,
            source=event_data.source,
            metadata=dict(event_data.metadata),
            sequence=sequence,
        )

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "id": self.id,
            "event_name": self.event_name,
            "data": self.data,
            "timestamp": self.timestamp.isoformat(),
            "source": self.source,
            "metadata": self.metadata,
            "sequence": self.sequence,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "StoredEvent":
        """
        Erstellt aus einem Dictionary.

        Args:
            data: Das Dictionary.

        Returns:
            StoredEvent.
        """
        return cls(
            id=data["id"],
            event_name=data["event_name"],
            data=data["data"],
            timestamp=datetime.fromisoformat(data["timestamp"]),
            source=data.get("source", ""),
            metadata=data.get("metadata", {}),
            sequence=data.get("sequence", 0),
        )

    def to_json(self) -> str:
        """Konvertiert in JSON."""
        return json.dumps(self.to_dict(), default=str)

    @classmethod
    def from_json(cls, json_str: str) -> "StoredEvent":
        """
        Erstellt aus JSON.

        Args:
            json_str: Der JSON-String.

        Returns:
            StoredEvent.
        """
        return cls.from_dict(json.loads(json_str))


@dataclass
class EventQuery:
    """
    Abfrage für gespeicherte Events.
    """

    event_names: list[str] | None = None
    """Filter nach Event-Namen."""

    start_time: datetime | None = None
    """Start-Zeitpunkt."""

    end_time: datetime | None = None
    """End-Zeitpunkt."""

    source: str | None = None
    """Filter nach Quelle."""

    metadata_filter: dict[str, Any] | None = None
    """Filter nach Metadaten."""

    limit: int = 1000
    """Maximale Anzahl Ergebnisse."""

    offset: int = 0
    """Offset für Paginierung."""

    order_ascending: bool = True
    """Sortierreihenfolge."""

    after_sequence: int | None = None
    """Events nach dieser Sequenznummer."""

    before_sequence: int | None = None
    """Events vor dieser Sequenznummer."""


@dataclass
class EventQueryResult:
    """
    Ergebnis einer Event-Abfrage.
    """

    events: list[StoredEvent]
    """Gefundene Events."""

    total_count: int
    """Gesamtzahl (ohne Limit)."""

    has_more: bool
    """Ob weitere Events existieren."""

    first_sequence: int | None = None
    """Erste Sequenznummer."""

    last_sequence: int | None = None
    """Letzte Sequenznummer."""


class EventStore(ABC):
    """
    Abstrakte Basisklasse für Event-Speicherung.

    Speichert Events für Replay, Audit und Analyse.

    Example:
        store = InMemoryEventStore()

        # Event speichern
        await store.append("user_login", event_data)

        # Events abfragen
        query = EventQuery(
            event_names=["user_login"],
            start_time=datetime.now() - timedelta(hours=1),
        )
        result = await store.query(query)

        for event in result.events:
            print(f"{event.timestamp}: {event.event_name}")
    """

    @abstractmethod
    async def append(
        self,
        event_name: str,
        event_data: "EventData",
    ) -> StoredEvent:
        """
        Speichert ein Event.

        Args:
            event_name: Name des Events.
            event_data: Die Event-Daten.

        Returns:
            Das gespeicherte Event.
        """
        pass

    @abstractmethod
    async def get(self, event_id: str) -> StoredEvent | None:
        """
        Gibt ein Event nach ID zurück.

        Args:
            event_id: Die Event-ID.

        Returns:
            StoredEvent oder None.
        """
        pass

    @abstractmethod
    async def query(self, query: EventQuery) -> EventQueryResult:
        """
        Führt eine Abfrage aus.

        Args:
            query: Die Abfrage.

        Returns:
            EventQueryResult.
        """
        pass

    @abstractmethod
    async def count(self, query: EventQuery | None = None) -> int:
        """
        Zählt Events.

        Args:
            query: Optionale Abfrage.

        Returns:
            Anzahl der Events.
        """
        pass

    @abstractmethod
    async def delete(self, event_id: str) -> bool:
        """
        Löscht ein Event.

        Args:
            event_id: Die Event-ID.

        Returns:
            True wenn gelöscht.
        """
        pass

    @abstractmethod
    async def clear(self) -> int:
        """
        Löscht alle Events.

        Returns:
            Anzahl gelöschter Events.
        """
        pass

    async def get_latest(
        self,
        event_name: str | None = None,
        count: int = 1,
    ) -> list[StoredEvent]:
        """
        Gibt die neuesten Events zurück.

        Args:
            event_name: Optionaler Event-Name-Filter.
            count: Anzahl der Events.

        Returns:
            Liste der Events.
        """
        query = EventQuery(
            event_names=[event_name] if event_name else None,
            limit=count,
            order_ascending=False,
        )
        result = await self.query(query)
        return result.events

    async def get_by_sequence_range(
        self,
        start: int,
        end: int,
    ) -> list[StoredEvent]:
        """
        Gibt Events in einem Sequenzbereich zurück.

        Args:
            start: Start-Sequenz (inklusiv).
            end: End-Sequenz (exklusiv).

        Returns:
            Liste der Events.
        """
        query = EventQuery(
            after_sequence=start - 1,
            before_sequence=end,
            limit=end - start,
        )
        result = await self.query(query)
        return result.events

    async def stream(
        self,
        query: EventQuery | None = None,
    ) -> Iterator[StoredEvent]:
        """
        Streamt Events.

        Args:
            query: Optionale Abfrage.

        Yields:
            StoredEvent Objekte.
        """
        query = query or EventQuery()
        offset = 0
        batch_size = 100

        while True:
            batch_query = EventQuery(
                event_names=query.event_names,
                start_time=query.start_time,
                end_time=query.end_time,
                source=query.source,
                metadata_filter=query.metadata_filter,
                limit=batch_size,
                offset=offset,
                order_ascending=query.order_ascending,
            )

            result = await self.query(batch_query)

            for event in result.events:
                yield event

            if not result.has_more:
                break

            offset += batch_size