---
name: event-driven-architecture
description: Generic Event-Driven Architecture patterns with Kafka, Dapr, and modern messaging systems. Provides reusable patterns for building scalable, resilient event-driven microservices. Framework-agnostic implementation supporting multiple message brokers, state stores, and event patterns. Follows 2025 best practices for distributed systems.
license: MIT
---

# Event-Driven Architecture with Modern Patterns

This skill provides comprehensive patterns for implementing event-driven microservices using modern messaging systems, distributed runtimes, and cloud-native patterns. It's designed to be framework-agnostic and applicable to any domain requiring event-driven capabilities.

## When to Use This Skill

Use this skill when you need to:
- Build event-driven microservices architecture
- Implement pub/sub patterns with Kafka, RabbitMQ, or cloud services
- Use Dapr for distributed application patterns
- Implement real-time notifications and workflows
- Build recurring task and reminder systems
- Create audit trails and activity logs
- Implement distributed state management
- Build serverless event workflows
- Handle event sourcing and CQRS patterns

## 1. Core Event Architecture

### Base Event Schema

```python
# events/core.py
from pydantic import BaseModel, Field, validator
from datetime import datetime
from typing import Optional, Dict, Any, Union, List
from enum import Enum
from abc import ABC, abstractmethod
import uuid
import json

class EventVersion(str, Enum):
    """Event versioning support"""
    V1_0 = "1.0"
    V1_1 = "1.1"
    V2_0 = "2.0"

class EventPriority(str, Enum):
    """Event priority levels"""
    LOW = "low"
    NORMAL = "normal"
    HIGH = "high"
    CRITICAL = "critical"

class BaseEvent(BaseModel, ABC):
    """Base event schema with common fields"""

    # Core identification
    event_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    event_type: str = Field(..., description="Type of the event")
    event_version: EventVersion = Field(default=EventVersion.V1_0)

    # Metadata
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    correlation_id: Optional[str] = None
    causation_id: Optional[str] = None  # Event that caused this one
    source: str = Field(..., description="Source service/identifier")

    # Context
    user_id: Optional[str] = None
    tenant_id: Optional[str] = None  # Multi-tenant support
    session_id: Optional[str] = None

    # Processing metadata
    priority: EventPriority = Field(default=EventPriority.NORMAL)
    retry_count: int = Field(default=0)
    max_retries: int = Field(default=3)
    delay_until: Optional[datetime] = None  # For delayed processing

    # Schema versioning
    schema_version: str = Field(default="1.0")

    class Config:
        # Allow additional fields for extensibility
        extra = "allow"
        # Use enum values
        use_enum_values = True

    @validator('event_type')
    def validate_event_type(cls, v):
        """Validate event type format"""
        if not v or '.' not in v:
            raise ValueError('event_type must be in format: domain.event_name')
        return v.lower()

    @validator('correlation_id')
    def validate_correlation_id(cls, v, values):
        """Set correlation_id from causation_id if not provided"""
        if not v and 'causation_id' in values:
            return values['causation_id']
        return v

    def to_dict(self) -> Dict[str, Any]:
        """Convert event to dictionary for serialization"""
        return self.model_dump()

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'BaseEvent':
        """Create event from dictionary"""
        return cls(**data)

    def with_context(self, **context) -> 'BaseEvent':
        """Add context to event"""
        for key, value in context.items():
            if hasattr(self, key):
                setattr(self, key, value)
        return self

class DomainEvent(BaseEvent):
    """Domain-specific event"""

    aggregate_id: str = Field(..., description="Aggregate root ID")
    aggregate_type: str = Field(..., description="Aggregate type")
    event_data: Dict[str, Any] = Field(default_factory=dict)

    @validator('aggregate_type')
    def validate_aggregate_type(cls, v):
        """Validate aggregate type"""
        if not v:
            raise ValueError('aggregate_type is required')
        return v.lower()

class IntegrationEvent(BaseEvent):
    """Integration event for cross-service communication"""

    target_services: List[str] = Field(default_factory=list)
    routing_key: Optional[str] = None
    message_format: str = Field(default="json")

    @validator('routing_key')
    def validate_routing_key(cls, v, values):
        """Generate routing key from event_type if not provided"""
        if not v and 'event_type' in values:
            return values['event_type'].replace('.', '/')
        return v

class CommandEvent(BaseEvent):
    """Command event representing an intent"""

    command_type: str = Field(..., description="Type of command")
    command_data: Dict[str, Any] = Field(default_factory=dict)
    expected_version: Optional[int] = None  # For optimistic concurrency

    @validator('command_type')
    def validate_command_type(cls, v):
        """Validate command type"""
        if not v.endswith('.command'):
            v = f"{v}.command"
        return v.lower()

class QueryEvent(BaseEvent):
    """Query event for data retrieval"""

    query_type: str = Field(..., description="Type of query")
    query_params: Dict[str, Any] = Field(default_factory=dict)
    result_topic: Optional[str] = None  # Where to send results

    @validator('query_type')
    def validate_query_type(cls, v):
        """Validate query type"""
        if not v.endswith('.query'):
            v = f"{v}.query"
        return v.lower()
```

### Event Store Pattern

```python
# events/store.py
import asyncio
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any, AsyncIterator
from datetime import datetime, timedelta
import json
import uuid

class EventStore(ABC):
    """Abstract event store interface"""

    @abstractmethod
    async def save_event(self, event: BaseEvent, stream_id: str) -> None:
        """Save event to a stream"""
        pass

    @abstractmethod
    async def get_events(
        self,
        stream_id: str,
        from_version: Optional[int] = None,
        to_version: Optional[int] = None,
        limit: Optional[int] = None
    ) -> AsyncIterator[BaseEvent]:
        """Get events from a stream"""
        pass

    @abstractmethod
    async def get_event_by_id(self, event_id: str) -> Optional[BaseEvent]:
        """Get a specific event by ID"""
        pass

    @abstractmethod
    async def get_events_by_type(
        self,
        event_type: str,
        from_timestamp: Optional[datetime] = None,
        to_timestamp: Optional[datetime] = None
    ) -> AsyncIterator[BaseEvent]:
        """Get events by type"""
        pass

    @abstractmethod
    async def get_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str
    ) -> Optional[Dict[str, Any]]:
        """Get aggregate snapshot"""
        pass

    @abstractmethod
    async def save_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str,
        data: Dict[str, Any],
        version: int
    ) -> None:
        """Save aggregate snapshot"""
        pass

class InMemoryEventStore(EventStore):
    """In-memory event store for testing and development"""

    def __init__(self):
        self._events: Dict[str, List[BaseEvent]] = {}
        self._snapshots: Dict[str, Dict[str, Any]] = {}
        self._type_index: Dict[str, List[str]] = {}

    async def save_event(self, event: BaseEvent, stream_id: str) -> None:
        """Save event to memory"""
        if stream_id not in self._events:
            self._events[stream_id] = []

        self._events[stream_id].append(event)

        # Update type index
        if event.event_type not in self._type_index:
            self._type_index[event.event_type] = []
        self._type_index[event.event_type].append(event.event_id)

    async def get_events(
        self,
        stream_id: str,
        from_version: Optional[int] = None,
        to_version: Optional[int] = None,
        limit: Optional[int] = None
    ) -> AsyncIterator[BaseEvent]:
        """Get events from memory"""
        events = self._events.get(stream_id, [])

        # Apply filters
        if from_version is not None:
            events = events[from_version:]
        if to_version is not None:
            events = events[:to_version]
        if limit is not None:
            events = events[:limit]

        for event in events:
            yield event

    async def get_event_by_id(self, event_id: str) -> Optional[BaseEvent]:
        """Get event by ID from memory"""
        for events in self._events.values():
            for event in events:
                if event.event_id == event_id:
                    return event
        return None

    async def get_events_by_type(
        self,
        event_type: str,
        from_timestamp: Optional[datetime] = None,
        to_timestamp: Optional[datetime] = None
    ) -> AsyncIterator[BaseEvent]:
        """Get events by type from memory"""
        event_ids = self._type_index.get(event_type, [])

        for event_id in event_ids:
            event = await self.get_event_by_id(event_id)
            if event:
                # Apply time filters
                if from_timestamp and event.timestamp < from_timestamp:
                    continue
                if to_timestamp and event.timestamp > to_timestamp:
                    continue
                yield event

    async def get_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str
    ) -> Optional[Dict[str, Any]]:
        """Get snapshot from memory"""
        snapshot_key = f"{aggregate_type}:{aggregate_id}"
        return self._snapshots.get(snapshot_key)

    async def save_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str,
        data: Dict[str, Any],
        version: int
    ) -> None:
        """Save snapshot to memory"""
        snapshot_key = f"{aggregate_type}:{aggregate_id}"
        self._snapshots[snapshot_key] = {
            "data": data,
            "version": version,
            "timestamp": datetime.utcnow()
        }

# PostgreSQL Event Store Implementation
class PostgreSQLEventStore(EventStore):
    """PostgreSQL-based event store"""

    def __init__(self, db_pool):
        self.db_pool = db_pool

    async def initialize_schema(self):
        """Initialize database schema"""
        async with self.db_pool.acquire() as conn:
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS event_store (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    stream_id VARCHAR(255) NOT NULL,
                    stream_version INTEGER NOT NULL,
                    event_id VARCHAR(255) UNIQUE NOT NULL,
                    event_type VARCHAR(255) NOT NULL,
                    event_data JSONB NOT NULL,
                    metadata JSONB,
                    timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
                    INDEX (stream_id, stream_version),
                    INDEX (event_type),
                    INDEX (timestamp)
                );

                CREATE TABLE IF NOT EXISTS snapshots (
                    aggregate_id VARCHAR(255) NOT NULL,
                    aggregate_type VARCHAR(255) NOT NULL,
                    data JSONB NOT NULL,
                    version INTEGER NOT NULL,
                    timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
                    PRIMARY KEY (aggregate_type, aggregate_id)
                );
            """)

    async def save_event(self, event: BaseEvent, stream_id: str) -> None:
        """Save event to PostgreSQL"""
        async with self.db_pool.acquire() as conn:
            # Get next version
            result = await conn.fetchval(
                "SELECT COALESCE(MAX(stream_version), 0) + 1 FROM event_store WHERE stream_id = $1",
                stream_id
            )

            await conn.execute(
                """
                INSERT INTO event_store (
                    stream_id, stream_version, event_id, event_type,
                    event_data, metadata, timestamp
                ) VALUES ($1, $2, $3, $4, $5, $6, $7)
                """,
                stream_id,
                result,
                event.event_id,
                event.event_type,
                event.to_dict(),
                {
                    "correlation_id": event.correlation_id,
                    "causation_id": event.causation_id,
                    "source": event.source,
                    "user_id": event.user_id,
                    "tenant_id": event.tenant_id
                },
                event.timestamp
            )

    async def get_events(
        self,
        stream_id: str,
        from_version: Optional[int] = None,
        to_version: Optional[int] = None,
        limit: Optional[int] = None
    ) -> AsyncIterator[BaseEvent]:
        """Get events from PostgreSQL"""
        query = "SELECT event_data FROM event_store WHERE stream_id = $1"
        params = [stream_id]

        if from_version is not None:
            query += " AND stream_version >= $2"
            params.append(from_version)
            version_offset = 2
        else:
            version_offset = 1

        if to_version is not None:
            query += f" AND stream_version <= ${version_offset + 1}"
            params.append(to_version)
            version_offset += 1

        query += " ORDER BY stream_version"

        if limit is not None:
            query += f" LIMIT ${version_offset + 1}"
            params.append(limit)

        async with self.db_pool.acquire() as conn:
            async for row in conn.cursor(query, *params):
                yield BaseEvent.from_dict(row["event_data"])

    async def get_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str
    ) -> Optional[Dict[str, Any]]:
        """Get snapshot from PostgreSQL"""
        async with self.db_pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT data, version FROM snapshots
                WHERE aggregate_id = $1 AND aggregate_type = $2
                """,
                aggregate_id,
                aggregate_type
            )

            if row:
                return {
                    "data": row["data"],
                    "version": row["version"]
                }
            return None

    async def save_aggregate_snapshot(
        self,
        aggregate_id: str,
        aggregate_type: str,
        data: Dict[str, Any],
        version: int
    ) -> None:
        """Save snapshot to PostgreSQL"""
        async with self.db_pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO snapshots (aggregate_id, aggregate_type, data, version, timestamp)
                VALUES ($1, $2, $3, $4, NOW())
                ON CONFLICT (aggregate_type, aggregate_id)
                DO UPDATE SET data = $3, version = $4, timestamp = NOW()
                """,
                aggregate_id,
                aggregate_type,
                data,
                version
            )
```

## 2. Message Broker Abstraction

### Generic Message Broker Interface

```python
# messaging/broker.py
import asyncio
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, Callable, List, AsyncIterator
from dataclasses import dataclass
from enum import Enum
import json
import logging

logger = logging.getLogger(__name__)

class MessageBrokerType(str, Enum):
    """Supported message broker types"""
    KAFKA = "kafka"
    RABBITMQ = "rabbitmq"
    REDIS = "redis"
    AWS_SQS = "aws_sqs"
    AZURE_SB = "azure_service_bus"
    GCP_PUBSUB = "gcp_pubsub"
    NATS = "nats"
    EMQX = "emqx"

@dataclass
class Message:
    """Generic message representation"""
    topic: str
    key: Optional[str]
    value: Any
    headers: Optional[Dict[str, str]] = None
    partition: Optional[int] = None
    timestamp: Optional[int] = None

    def to_dict(self) -> Dict[str, Any]:
        """Convert message to dictionary"""
        return {
            "topic": self.topic,
            "key": self.key,
            "value": self.value,
            "headers": self.headers or {},
            "partition": self.partition,
            "timestamp": self.timestamp
        }

class MessageBroker(ABC):
    """Abstract message broker interface"""

    @abstractmethod
    async def connect(self) -> None:
        """Establish connection to broker"""
        pass

    @abstractmethod
    async def disconnect(self) -> None:
        """Close connection to broker"""
        pass

    @abstractmethod
    async def publish(self, message: Message) -> None:
        """Publish a message"""
        pass

    @abstractmethod
    async def publish_batch(self, messages: List[Message]) -> None:
        """Publish multiple messages"""
        pass

    @abstractmethod
    async def subscribe(
        self,
        topic: str,
        handler: Callable[[Message], None],
        group_id: Optional[str] = None
    ) -> None:
        """Subscribe to a topic"""
        pass

    @abstractmethod
    async def create_topic(self, topic: str, **kwargs) -> None:
        """Create a topic if it doesn't exist"""
        pass

class KafkaBroker(MessageBroker):
    """Kafka message broker implementation"""

    def __init__(
        self,
        bootstrap_servers: str,
        producer_config: Optional[Dict[str, Any]] = None,
        consumer_config: Optional[Dict[str, Any]] = None
    ):
        self.bootstrap_servers = bootstrap_servers
        self.producer_config = producer_config or {}
        self.consumer_config = consumer_config or {}
        self.producer = None
        self.consumers: List[Any] = []

        # Default configurations
        self.producer_config.update({
            'bootstrap_servers': bootstrap_servers,
            'value_serializer': lambda v: json.dumps(v).encode('utf-8'),
            'key_serializer': lambda k: k.encode('utf-8') if k else None,
            'acks': 'all',
            'retries': 3,
            'retry_backoff_ms': 100
        })

        self.consumer_config.update({
            'bootstrap_servers': bootstrap_servers,
            'value_deserializer': lambda v: json.loads(v.decode('utf-8')),
            'key_deserializer': lambda k: k.decode('utf-8') if k else None,
            'auto_offset_reset': 'earliest',
            'enable_auto_commit': True
        })

    async def connect(self) -> None:
        """Connect to Kafka"""
        from aiokafka import AIOKafkaProducer

        self.producer = AIOKafkaProducer(**self.producer_config)
        await self.producer.start()
        logger.info("Connected to Kafka broker")

    async def disconnect(self) -> None:
        """Disconnect from Kafka"""
        if self.producer:
            await self.producer.stop()

        for consumer in self.consumers:
            await consumer.stop()

        logger.info("Disconnected from Kafka broker")

    async def publish(self, message: Message) -> None:
        """Publish message to Kafka"""
        if not self.producer:
            raise RuntimeError("Producer not initialized")

        await self.producer.send_and_wait(
            topic=message.topic,
            value=message.value,
            key=message.key,
            headers=message.headers,
            partition=message.partition
        )

    async def publish_batch(self, messages: List[Message]) -> None:
        """Publish batch of messages to Ka