ledgrab/plans/game-integration/phase-1-event-bus.md

Phase 1: Core Event Bus & Adapter Framework

Status: ✅ Complete Parent plan: PLAN.md Domain: backend

Objective

Build the foundational event infrastructure: the standardized event model, in-process pub/sub bus, adapter ABC, adapter registry, and the YAML-driven mapping adapter. This phase creates the core abstractions that all subsequent phases depend on.

Tasks

• Task 1: Create core/game_integration/__init__.py package
• Task 2: Define standardized event vocabulary as an enum/constant module (core/game_integration/events.py)
  • Categories: health, armor, shield, mana, energy, ammo, gold, fuel, speed (continuous)
  • Categories: kill, death, assist, damage_taken, damage_dealt (combat triggers)
  • Categories: match_start, match_end, round_start, round_end (match flow triggers)
  • Categories: objective_captured, objective_lost, objective_progress (objective)
  • Categories: stunned, blinded, buffed, debuffed (status effects)
  • Categories: team_a, team_b (team affiliation)
  • Each event type has metadata: display name, category, value_type (continuous/trigger), default_range (min, max)
• Task 3: Create GameEvent frozen dataclass (core/game_integration/events.py)
  • Fields: adapter_id, event_type (str from vocabulary), value (float 0.0-1.0), raw_data (dict), timestamp (float, monotonic)
• Task 4: Create GameEventBus (core/game_integration/event_bus.py)
  • Thread-safe pub/sub with threading.Lock
  • publish(event: GameEvent) — dispatches to matching subscribers
  • subscribe(event_type: str, callback) → returns subscription ID
  • subscribe_all(callback) → receives all events (for diagnostics/live monitor)
  • unsubscribe(subscription_id)
  • get_recent_events(limit=50) → returns recent events deque for diagnostics
  • get_stats() → event counts per type, last event timestamp
  • Use collections.deque(maxlen=100) for recent events
• Task 5: Create GameAdapter ABC (core/game_integration/base_adapter.py)
  • Class attributes: ADAPTER_TYPE, DISPLAY_NAME, GAME_NAME, SUPPORTED_EVENTS (list of event type strings)
  • parse_payload(cls, payload: dict, adapter_config: dict, prev_state: dict) -> tuple[list[GameEvent], dict] — returns events + updated state (for diff detection)
  • validate_auth(cls, headers: dict, payload: dict, adapter_config: dict) -> bool
  • get_config_schema(cls) -> dict — returns JSON schema for adapter-specific config fields (for UI auto-generation)
  • get_setup_instructions(cls) -> str — returns markdown setup guide for the game
• Task 6: Create AdapterRegistry (core/game_integration/adapter_registry.py)
  • register(adapter_class), get_adapter(adapter_type), get_all_adapters(), get_available_adapters()
  • Follow EngineRegistry pattern
• Task 7: Create MappingAdapter (core/game_integration/mapping_adapter.py)
  • Parses community YAML adapter files into a generic adapter instance
  • YAML schema: adapter name, game name, protocol (webhook/poll), mappings list
  • Each mapping: source_path (JSONPath-like), event (standard type), min/max (for continuous), trigger mode (on_change/on_increase/on_decrease/on_value)
  • load_adapter_from_yaml(path) -> GameAdapter — factory function
  • validate_adapter_yaml(data: dict) -> list[str] — returns validation errors
• Task 8: Create core/game_integration/adapters/__init__.py package (empty, for Phase 3)
• Task 9: Write unit tests for GameEventBus (publish, subscribe, unsubscribe, thread safety)
• Task 10: Write unit tests for AdapterRegistry (register, get, duplicates)
• Task 11: Write unit tests for MappingAdapter (YAML parsing, payload translation, validation)

Files to Modify/Create

• server/src/wled_controller/core/game_integration/__init__.py — package init, re-exports
• server/src/wled_controller/core/game_integration/events.py — GameEvent dataclass + event vocabulary
• server/src/wled_controller/core/game_integration/event_bus.py — GameEventBus
• server/src/wled_controller/core/game_integration/base_adapter.py — GameAdapter ABC
• server/src/wled_controller/core/game_integration/adapter_registry.py — AdapterRegistry
• server/src/wled_controller/core/game_integration/mapping_adapter.py — MappingAdapter + YAML loader
• server/src/wled_controller/core/game_integration/adapters/__init__.py — empty package
• server/tests/core/test_game_event_bus.py — EventBus tests
• server/tests/core/test_adapter_registry.py — Registry tests
• server/tests/core/test_mapping_adapter.py — MappingAdapter tests

Acceptance Criteria

• GameEvent is frozen/immutable with all required fields
• Event vocabulary covers all standard categories with metadata
• EventBus correctly dispatches events to type-specific and wildcard subscribers
• EventBus is thread-safe (concurrent publish/subscribe doesn't crash)
• AdapterRegistry follows EngineRegistry pattern faithfully
• MappingAdapter can load a YAML file and translate a JSON payload into GameEvents
• MappingAdapter validates YAML schema and reports errors clearly
• All tests pass

Notes

• Use time.monotonic() for timestamps (not time.time()) — monotonic is immune to clock adjustments
• Keep the event vocabulary extensible — new types should just be new entries, no code changes
• The MappingAdapter is key for scaling to many games without writing Python code per game
• prev_state parameter on parse_payload enables diff-based trigger detection (e.g. CS2 kills counter)

Review Checklist

• All tasks completed
• Code follows project conventions (PEP 8, type annotations, immutability)
• No unintended side effects
• Tests pass (new + existing)

Handoff to Next Phase

Completed: All 11 tasks implemented and tested. 55 tests pass, 0 failures. Ruff clean.

What was built

• events.py — GameEvent frozen dataclass + 23-type event vocabulary with EventTypeMetadata (category, value_type, default_range). Helper functions: get_event_vocabulary(), get_event_metadata(), is_known_event_type().
• event_bus.py — GameEventBus with thread-safe pub/sub. Type-specific and wildcard subscriptions, deque(maxlen=100) for recent events, per-type event counts. Callbacks invoked outside lock to prevent deadlocks.
• base_adapter.py — GameAdapter ABC with ADAPTER_TYPE/DISPLAY_NAME/GAME_NAME/SUPPORTED_EVENTS class vars. Abstract methods: parse_payload(), validate_auth(). Default implementations: get_config_schema(), get_setup_instructions().
• adapter_registry.py — AdapterRegistry following EngineRegistry pattern (class-level dict, register/get/list/clear).
• mapping_adapter.py — MappingAdapter (concrete, instance-based) + load_adapter_from_yaml() factory + validate_adapter_yaml() validator. Supports dot-notation JSON paths, 4 trigger modes (on_change/on_increase/on_decrease/on_value), value normalization to 0.0-1.0, header-based auth.
• adapters/__init__.py — empty package ready for Phase 3 built-in adapters.
• __init__.py — re-exports all public API.

Key design decisions

• MappingAdapter is instance-based (not classmethod-based like built-in adapters) because each YAML file creates a unique adapter with its own mappings. The parse_payload/validate_auth methods use # type: ignore[override] for the instance vs classmethod difference.
• prev_state dict keys are source_path strings, values are the last numeric value seen. This enables diff-based trigger detection (e.g., CS2 kill counter increments).
• Non-numeric values in payloads are treated as trigger events with value=1.0.

What Phase 2 needs

• GameEventBus instance should be created as a singleton in dependencies.py and injected via FastAPI Depends().
• AdapterRegistry is ready for built-in adapters to call AdapterRegistry.register(MyAdapter) at import time.
• The GameIntegrationStore (Phase 2) will need to store adapter configs that include adapter_id fields matching what parse_payload expects in adapter_config.