Trixy7/trixy_core/events/persistence/memory.py

# -*- coding: utf-8 -*-
"""
In-Memory Event Store.

Speichert Events im Arbeitsspeicher.
"""

from __future__ import annotations

import asyncio
from datetime import datetime
from typing import TYPE_CHECKING, Any

from trixy_core.events.persistence.store import (
    EventStore,
    StoredEvent,
    EventQuery,
    EventQueryResult,
)

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


class InMemoryEventStore(EventStore):
    """
    Event Store im Arbeitsspeicher.

    Schnell, aber nicht persistent. Ideal für Tests
    und kurzfristiges Event-Tracking.

    Example:
        store = InMemoryEventStore(max_events=10000)

        # Events speichern
        await store.append("click", event_data)

        # Abfragen
        result = await store.query(EventQuery(limit=100))
    """

    def __init__(
        self,
        max_events: int = 100000,
        auto_cleanup: bool = True,
        cleanup_threshold: float = 0.9,
    ) -> None:
        """
        Initialisiert den Store.

        Args:
            max_events: Maximale Anzahl Events.
            auto_cleanup: Automatisches Löschen alter Events.
            cleanup_threshold: Schwelle für Cleanup (0.0-1.0).
        """
        self._events: dict[str, StoredEvent] = {}
        self._sequence_index: list[str] = []  # IDs in Reihenfolge
        self._event_name_index: dict[str, list[str]] = {}  # Event-Name -> IDs
        self._source_index: dict[str, list[str]] = {}  # Source -> IDs

        self._max_events = max_events
        self._auto_cleanup = auto_cleanup
        self._cleanup_threshold = cleanup_threshold
        self._current_sequence = 0
        self._lock = asyncio.Lock()

    @property
    def event_count(self) -> int:
        """Anzahl gespeicherter Events."""
        return len(self._events)

    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.
        """
        async with self._lock:
            # Cleanup wenn nötig
            if self._auto_cleanup and self.event_count >= int(self._max_events * self._cleanup_threshold):
                await self._cleanup()

            self._current_sequence += 1
            stored = StoredEvent.from_event(
                event_name=event_name,
                event_data=event_data,
                sequence=self._current_sequence,
            )

            # In Hauptindex speichern
            self._events[stored.id] = stored

            # In Sequence-Index
            self._sequence_index.append(stored.id)

            # In Event-Name-Index
            if event_name not in self._event_name_index:
                self._event_name_index[event_name] = []
            self._event_name_index[event_name].append(stored.id)

            # In Source-Index
            if stored.source:
                if stored.source not in self._source_index:
                    self._source_index[stored.source] = []
                self._source_index[stored.source].append(stored.id)

            return stored

    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.
        """
        return self._events.get(event_id)

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

        Args:
            query: Die Abfrage.

        Returns:
            EventQueryResult.
        """
        async with self._lock:
            # Kandidaten ermitteln
            if query.event_names:
                candidate_ids: set[str] = set()
                for name in query.event_names:
                    candidate_ids.update(self._event_name_index.get(name, []))
            else:
                candidate_ids = set(self._events.keys())

            # Source-Filter
            if query.source:
                source_ids = set(self._source_index.get(query.source, []))
                candidate_ids &= source_ids

            # Events filtern
            filtered: list[StoredEvent] = []

            for event_id in candidate_ids:
                event = self._events.get(event_id)
                if not event:
                    continue

                # Zeit-Filter
                if query.start_time and event.timestamp < query.start_time:
                    continue
                if query.end_time and event.timestamp > query.end_time:
                    continue

                # Sequenz-Filter
                if query.after_sequence is not None and event.sequence <= query.after_sequence:
                    continue
                if query.before_sequence is not None and event.sequence >= query.before_sequence:
                    continue

                # Metadata-Filter
                if query.metadata_filter:
                    match = True
                    for key, value in query.metadata_filter.items():
                        if event.metadata.get(key) != value:
                            match = False
                            break
                    if not match:
                        continue

                filtered.append(event)

            # Sortieren
            filtered.sort(
                key=lambda e: e.sequence,
                reverse=not query.order_ascending,
            )

            total_count = len(filtered)

            # Paginierung
            start = query.offset
            end = start + query.limit
            paginated = filtered[start:end]

            return EventQueryResult(
                events=paginated,
                total_count=total_count,
                has_more=end < total_count,
                first_sequence=paginated[0].sequence if paginated else None,
                last_sequence=paginated[-1].sequence if paginated else None,
            )

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

        Args:
            query: Optionale Abfrage.

        Returns:
            Anzahl der Events.
        """
        if query is None:
            return self.event_count

        result = await self.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=1000000,  # Großes Limit für Zählung
        ))
        return result.total_count

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

        Args:
            event_id: Die Event-ID.

        Returns:
            True wenn gelöscht.
        """
        async with self._lock:
            event = self._events.pop(event_id, None)
            if not event:
                return False

            # Aus Indices entfernen
            if event_id in self._sequence_index:
                self._sequence_index.remove(event_id)

            if event.event_name in self._event_name_index:
                ids = self._event_name_index[event.event_name]
                if event_id in ids:
                    ids.remove(event_id)

            if event.source and event.source in self._source_index:
                ids = self._source_index[event.source]
                if event_id in ids:
                    ids.remove(event_id)

            return True

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

        Returns:
            Anzahl gelöschter Events.
        """
        async with self._lock:
            count = len(self._events)
            self._events.clear()
            self._sequence_index.clear()
            self._event_name_index.clear()
            self._source_index.clear()
            self._current_sequence = 0
            return count

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

        Returns:
            Anzahl entfernter Events.
        """
        # Älteste 10% entfernen
        to_remove = int(self._max_events * 0.1)
        removed = 0

        for _ in range(to_remove):
            if not self._sequence_index:
                break

            event_id = self._sequence_index[0]
            if await self.delete(event_id):
                removed += 1

        return removed

    def get_event_names(self) -> list[str]:
        """
        Gibt alle Event-Namen zurück.

        Returns:
            Liste der Event-Namen.
        """
        return list(self._event_name_index.keys())

    def get_sources(self) -> list[str]:
        """
        Gibt alle Quellen zurück.

        Returns:
            Liste der Quellen.
        """
        return list(self._source_index.keys())

    def get_stats(self) -> dict[str, Any]:
        """
        Gibt Statistiken zurück.

        Returns:
            Statistik-Dictionary.
        """
        return {
            "total_events": self.event_count,
            "max_events": self._max_events,
            "fill_rate": self.event_count / self._max_events if self._max_events > 0 else 0,
            "current_sequence": self._current_sequence,
            "event_types": len(self._event_name_index),
            "sources": len(self._source_index),
            "events_per_type": {
                name: len(ids)
                for name, ids in self._event_name_index.items()
            },
        }