# -*- 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() }, }