Event-Driven Architecture¶

MAID uses an event-driven architecture for loose coupling between systems. Events allow different parts of the game to communicate without direct dependencies.

What is an Event?¶

An event is a message that describes something that happened in the game. Events carry data about the occurrence and can be subscribed to by any system or component.

from dataclasses import dataclass
from maid_engine.core.events import Event

@dataclass
class PlayerDamagedEvent(Event):
    player_id: UUID
    damage: int
    source_id: UUID | None = None

Core Concepts¶

EventBus¶

The EventBus is the central hub for event distribution:

from maid_engine.core.events import EventBus

bus = EventBus()

# Subscribe to events
handler_id = bus.subscribe(PlayerDamagedEvent, handle_damage)

# Emit events
await bus.emit(PlayerDamagedEvent(player_id=player.id, damage=10))

# Unsubscribe
bus.unsubscribe(handler_id)

Event Flow¶

System A emits event -> EventBus -> Handler 1
                                 -> Handler 2
                                 -> Handler 3

Multiple handlers can subscribe to the same event type, and they're called in priority order.

Built-in Events¶

MAID Engine provides core lifecycle events:

Entity Events¶

from maid_engine.core.events import (
    EntityCreatedEvent,    # Emitted when an entity is created
    EntityDestroyedEvent,  # Emitted when an entity is destroyed
)

@dataclass
class EntityCreatedEvent(Event):
    entity_id: UUID

@dataclass
class EntityDestroyedEvent(Event):
    entity_id: UUID

Component Events¶

from maid_engine.core.events import (
    ComponentAddedEvent,   # Emitted when a component is added
    ComponentRemovedEvent, # Emitted when a component is removed
)

@dataclass
class ComponentAddedEvent(Event):
    entity_id: UUID
    component_type: str

@dataclass
class ComponentRemovedEvent(Event):
    entity_id: UUID
    component_type: str

Player Events¶

from maid_engine.core.events import (
    PlayerConnectedEvent,    # Player connected to server
    PlayerDisconnectedEvent, # Player disconnected
    PlayerCommandEvent,      # Player entered a command
)

@dataclass
class PlayerConnectedEvent(Event):
    session_id: UUID
    player_id: UUID | None = None

@dataclass
class PlayerDisconnectedEvent(Event):
    session_id: UUID
    player_id: UUID | None = None

@dataclass
class PlayerCommandEvent(Event):
    player_id: UUID
    command: str
    args: list[str]

Room Events¶

from maid_engine.core.events import (
    RoomEnterEvent, # Entity entered a room
    RoomLeaveEvent, # Entity left a room
)

@dataclass
class RoomEnterEvent(Event):
    entity_id: UUID
    room_id: UUID
    from_room_id: UUID | None = None

@dataclass
class RoomLeaveEvent(Event):
    entity_id: UUID
    room_id: UUID
    to_room_id: UUID | None = None

Tick Events¶

from maid_engine.core.events import TickEvent

@dataclass
class TickEvent(Event):
    tick_number: int
    delta: float  # Time since last tick in seconds

Custom Events¶

For ad-hoc events without creating a new class:

from maid_engine.core.events import CustomEvent

await bus.emit(CustomEvent(
    name="achievement_unlocked",
    data={"player_id": player.id, "achievement": "first_kill"}
))

Event Priority¶

Handlers are called in priority order (highest priority first):

from maid_engine.core.events import EventPriority

# High priority - runs first
bus.subscribe(
    DamageEvent,
    validate_damage,
    priority=EventPriority.HIGHEST
)

# Normal priority (default)
bus.subscribe(
    DamageEvent,
    apply_damage,
    priority=EventPriority.NORMAL
)

# Low priority - runs last
bus.subscribe(
    DamageEvent,
    log_damage,
    priority=EventPriority.LOWEST
)

Priority levels:

Priority	Use Case
`HIGHEST`	Validation, blocking checks
`HIGH`	Modifiers, armor reduction
`NORMAL`	Main logic (default)
`LOW`	Secondary effects
`LOWEST`	Logging, statistics

Event Cancellation¶

Events can be cancelled to prevent further processing:

@dataclass
class DamageEvent(Event):
    target_id: UUID
    damage: int

async def check_immunity(event: DamageEvent) -> None:
    target = world.get_entity(event.target_id)
    if target.has_tag("immune"):
        event.cancel()  # Prevents subsequent handlers from running

bus.subscribe(DamageEvent, check_immunity, priority=EventPriority.HIGHEST)

Check cancellation in later handlers:

async def apply_damage(event: DamageEvent) -> None:
    if event.cancelled:
        return  # Skip if cancelled

    # Apply damage...

One-Time Handlers¶

Subscribe to an event only once:

# Handler is automatically unsubscribed after first invocation
bus.subscribe(
    PlayerConnectedEvent,
    welcome_first_player,
    once=True
)

Synchronous Event Queuing¶

When you need to emit from synchronous code:

# Queue event for later processing
bus.emit_sync(PlayerDamagedEvent(player_id=player.id, damage=10))

# Later, process all queued events
count = await bus.process_pending()
print(f"Processed {count} events")

Content Pack Integration¶

Events work seamlessly with content packs:

Registering Event Types¶

class MyContentPack:
    def get_events(self) -> list[type[Event]]:
        """Return custom event types defined by this pack."""
        return [
            SpellCastEvent,
            ItemPickedUpEvent,
            QuestCompletedEvent,
        ]

Handler Source Tracking¶

For hot reload support, track which pack registered handlers:

handler_id = bus.subscribe(
    DamageEvent,
    handle_damage,
    source_pack="my-combat-pack"
)

# Later, unsubscribe all handlers from a pack
removed = bus.unsubscribe_pack("my-combat-pack")

Event Statistics¶

Query event bus state:

# Check if any handlers exist for an event type
if bus.has_handlers(DamageEvent):
    await bus.emit(damage_event)

# Count handlers
total = bus.handler_count()
damage_handlers = bus.handler_count(DamageEvent)

# Check if currently dispatching
if bus.is_dispatching():
    # Removals will be deferred
    pass

Best Practices¶

1. Keep Events Immutable¶

Don't modify event data after creation:

# Good - create new event if needed
event = DamageEvent(target_id=target.id, damage=10)

# Bad - modifying after creation
event.damage = 20  # Avoid!

2. Use Specific Event Types¶

# Good - specific events
class MeleeAttackEvent(Event): ...
class SpellCastEvent(Event): ...
class RangedAttackEvent(Event): ...

# Bad - generic event with type field
class AttackEvent(Event):
    attack_type: str  # "melee", "spell", "ranged"

3. Include Necessary Context¶

# Good - all context included
@dataclass
class ItemPickedUpEvent(Event):
    entity_id: UUID      # Who picked it up
    item_id: UUID        # What was picked up
    room_id: UUID        # Where it happened
    quantity: int = 1    # How many

# Bad - insufficient context
@dataclass
class ItemPickedUpEvent(Event):
    item_id: UUID  # Who? Where? How many?

4. Document Event Contracts¶

@dataclass
class CombatStartedEvent(Event):
    """Emitted when combat begins between entities.

    Handlers should:
    - NOT cancel this event (combat is already initiated)
    - Update UI/notifications for involved entities
    - Start combat timers if applicable

    Attributes:
        initiator_id: Entity that started combat
        target_id: Entity being attacked
        room_id: Room where combat is happening
    """
    initiator_id: UUID
    target_id: UUID
    room_id: UUID

5. Handle Errors Gracefully¶

Event handlers should not crash the game:

async def safe_handler(event: DamageEvent) -> None:
    try:
        entity = world.get_entity(event.target_id)
        if entity is None:
            return  # Entity was destroyed, skip

        health = entity.try_get(HealthComponent)
        if health:
            health.current -= event.damage
    except Exception as e:
        logger.error(f"Error handling damage event: {e}")
        # Don't re-raise - other handlers should still run

Next Steps¶

Writing Event Handlers - Detailed handler patterns
Creating Custom Events - Define your own events