Event-Driven Architecture: Design, Deployment, and Hardening Best Practices¶

# Producer example
class EventProducer:
    def __init__(self, broker):
        self.broker = broker

    async def publish(self, topic: str, event: dict):
        """Publish event to topic"""
        await self.broker.publish(topic, event)

# Consumer example
class EventConsumer:
    def __init__(self, broker):
        self.broker = broker

    async def consume(self, topic: str, handler: callable):
        """Consume events from topic"""
        async for event in self.broker.subscribe(topic):
            await handler(event)

# Push model (NATS)
async def push_consumer():
    async def handler(msg):
        await process_message(msg.data)

    await nc.subscribe("events.>", cb=handler)

# Pull model (Kafka)
async def pull_consumer():
    consumer = aiokafka.AIOKafkaConsumer("events")
    async for msg in consumer:
        await process_message(msg.value)

# Event store
class EventStore:
    def __init__(self, db):
        self.db = db

    async def append(self, stream_id: str, event: dict):
        """Append event to stream"""
        await self.db.execute("""
            INSERT INTO events (stream_id, event_type, payload, version)
            VALUES (%s, %s, %s, 
                (SELECT COALESCE(MAX(version), 0) + 1 
                 FROM events WHERE stream_id = %s))
        """, (stream_id, event["type"], json.dumps(event), stream_id))

    async def get_stream(self, stream_id: str) -> List[dict]:
        """Reconstruct state from events"""
        events = await self.db.fetch("""
            SELECT event_type, payload, version
            FROM events
            WHERE stream_id = %s
            ORDER BY version
        """, stream_id)

        return [{"type": e["event_type"], "payload": json.loads(e["payload"])} 
                for e in events]

# At-least-once delivery with idempotency
class IdempotentConsumer:
    def __init__(self, db):
        self.db = db

    async def process(self, event: dict):
        """Process event with idempotency"""
        event_id = event["id"]

        # Check if already processed
        processed = await self.db.fetchval("""
            SELECT 1 FROM processed_events WHERE event_id = %s
        """, event_id)

        if processed:
            return  # Already processed, skip

        # Process event
        await handle_event(event)

        # Mark as processed
        await self.db.execute("""
            INSERT INTO processed_events (event_id, processed_at)
            VALUES (%s, NOW())
        """, event_id)

# Natural key deduplication
class DedupProcessor:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def process(self, event: dict):
        """Process with natural key deduplication"""
        # Use natural key (e.g., user_id + action)
        dedup_key = f"event:{event['user_id']}:{event['action']}"

        # Check if already processed
        if await self.redis.get(dedup_key):
            return  # Duplicate, skip

        # Process event
        await handle_event(event)

        # Mark as processed (TTL for cleanup)
        await self.redis.setex(dedup_key, 86400, "1")

import hashlib

class HashDedupProcessor:
    def __init__(self, redis_client):
        self.redis = redis_client

    def event_hash(self, event: dict) -> str:
        """Generate hash for event"""
        # Normalize event (remove timestamp, etc.)
        normalized = {
            "type": event["type"],
            "payload": event["payload"]
        }
        event_str = json.dumps(normalized, sort_keys=True)
        return hashlib.sha256(event_str.encode()).hexdigest()

    async def process(self, event: dict):
        """Process with hash-based deduplication"""
        event_hash = self.event_hash(event)
        dedup_key = f"event_hash:{event_hash}"

        if await self.redis.get(dedup_key):
            return  # Duplicate

        await handle_event(event)
        await self.redis.setex(dedup_key, 86400, "1")

# Backpressure with Redis Streams
class BackpressureConsumer:
    def __init__(self, redis_client, max_queue_size: int = 1000):
        self.redis = redis_client
        self.max_queue_size = max_queue_size
        self.processing = asyncio.Semaphore(10)  # Max 10 concurrent

    async def consume(self, stream: str, group: str, consumer: str):
        """Consume with backpressure"""
        while True:
            # Check queue size
            queue_size = await self.redis.xlen(stream)

            if queue_size > self.max_queue_size:
                # Backpressure: slow down
                await asyncio.sleep(1.0)
                continue

            # Read messages
            messages = await self.redis.xreadgroup(
                group, consumer, {stream: ">"},
                count=10, block=1000
            )

            # Process with concurrency limit
            for stream_name, msg_list in messages:
                for msg_id, data in msg_list:
                    async with self.processing:
                        await self.process_message(msg_id, data)
                        await self.redis.xack(stream, group, msg_id)

class BoundedQueue:
    def __init__(self, maxsize: int = 1000):
        self.queue = asyncio.Queue(maxsize=maxsize)
        self.dropped = 0

    async def put(self, item):
        """Put item with backpressure"""
        try:
            await asyncio.wait_for(
                self.queue.put(item),
                timeout=0.1
            )
        except asyncio.TimeoutError:
            # Queue full, drop message
            self.dropped += 1
            # Optionally: send to DLQ
            await self.send_to_dlq(item)

class DLQHandler:
    def __init__(self, broker, dlq_topic: str):
        self.broker = broker
        self.dlq_topic = dlq_topic

    async def send_to_dlq(self, event: dict, error: Exception, retry_count: int):
        """Send failed event to DLQ"""
        dlq_event = {
            "original_event": event,
            "error": str(error),
            "retry_count": retry_count,
            "failed_at": datetime.now().isoformat(),
            "dlq_reason": self.classify_error(error)
        }

        await self.broker.publish(self.dlq_topic, dlq_event)

    def classify_error(self, error: Exception) -> str:
        """Classify error for DLQ routing"""
        if isinstance(error, ValueError):
            return "invalid_data"
        elif isinstance(error, TimeoutError):
            return "timeout"
        elif isinstance(error, ConnectionError):
            return "connection_error"
        else:
            return "unknown"

class EventReplayer:
    def __init__(self, event_store):
        self.event_store = event_store

    async def replay_from(self, stream_id: str, from_version: int):
        """Replay events from specific version"""
        events = await self.event_store.get_stream(stream_id)

        # Filter events from version
        events_to_replay = [
            e for e in events 
            if e["version"] >= from_version
        ]

        # Replay in order
        for event in events_to_replay:
            await self.process_event(event)

    async def replay_time_range(self, stream_id: str, start: datetime, end: datetime):
        """Replay events in time range"""
        events = await self.event_store.get_events_in_range(
            stream_id, start, end
        )

        for event in events:
            await self.process_event(event)

class VersionedEvent:
    def __init__(self, event_type: str, version: int, payload: dict):
        self.event_type = event_type
        self.version = version
        self.payload = payload

    def to_dict(self) -> dict:
        return {
            "type": self.event_type,
            "version": self.version,
            "payload": self.payload,
            "schema_version": f"{self.event_type}_v{self.version}"
        }

# Event versioning with schema evolution
class EventVersionHandler:
    def __init__(self):
        self.handlers = {}

    def register_handler(self, event_type: str, version: int, handler: callable):
        """Register handler for specific event version"""
        key = f"{event_type}_v{version}"
        self.handlers[key] = handler

    async def handle(self, event: dict):
        """Handle event with version routing"""
        event_type = event["type"]
        version = event.get("version", 1)
        key = f"{event_type}_v{version}"

        handler = self.handlers.get(key)
        if not handler:
            # Try to find compatible handler
            handler = self.find_compatible_handler(event_type, version)

        if handler:
            await handler(event)
        else:
            raise ValueError(f"No handler for {key}")

class TopicEvolution:
    """Handle topic evolution and migration"""

    def migrate_topic(self, old_topic: str, new_topic: str):
        """Migrate from old topic to new topic"""
        # Strategy 1: Dual-write during migration
        async def dual_write(event: dict):
            await broker.publish(old_topic, event)
            await broker.publish(new_topic, event)

        # Strategy 2: Consumer bridge
        async def bridge_consumer():
            async for event in broker.subscribe(old_topic):
                # Transform if needed
                transformed = self.transform_event(event)
                await broker.publish(new_topic, transformed)

        # Strategy 3: Gradual migration
        # - Start dual-write
        # - Migrate consumers one by one
        # - Stop writing to old topic
        # - Deprecate old topic

class RetryPolicy:
    def __init__(
        self,
        max_retries: int = 3,
        initial_delay: float = 1.0,
        max_delay: float = 60.0,
        multiplier: float = 2.0
    ):
        self.max_retries = max_retries
        self.initial_delay = initial_delay
        self.max_delay = max_delay
        self.multiplier = multiplier

    async def execute_with_retry(self, func: callable, *args, **kwargs):
        """Execute function with exponential backoff retry"""
        delay = self.initial_delay

        for attempt in range(self.max_retries):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    raise  # Last attempt, re-raise

                # Exponential backoff with jitter
                jitter = random.uniform(0, delay * 0.1)
                await asyncio.sleep(delay + jitter)
                delay = min(delay * self.multiplier, self.max_delay)

class RecoverableEventProcessor:
    async def process(self, event: dict):
        """Process recoverable event"""
        try:
            await handle_event(event)
            await self.mark_processed(event["id"])
        except Exception as e:
            await self.send_to_retry_queue(event, e)

class NonRecoverableEventProcessor:
    async def process(self, event: dict):
        """Process non-recoverable event (best effort)"""
        try:
            await handle_event(event)
        except Exception as e:
            # Log and continue (don't retry)
            logger.error(f"Failed to process non-recoverable event: {e}")

# Postgres LISTEN/NOTIFY producer
class PostgresEventProducer:
    def __init__(self, db_pool):
        self.db_pool = db_pool

    async def notify_event(self, channel: str, payload: dict):
        """Send NOTIFY event"""
        async with self.db_pool.acquire() as conn:
            await conn.execute(
                "SELECT pg_notify(%s, %s)",
                channel,
                json.dumps(payload)
            )

# Postgres LISTEN/NOTIFY consumer
class PostgresEventConsumer:
    def __init__(self, db_pool):
        self.db_pool = db_pool
        self.listeners = {}

    async def listen(self, channel: str, handler: callable):
        """Listen for NOTIFY events"""
        async with self.db_pool.acquire() as conn:
            await conn.execute(f"LISTEN {channel}")

            async for notify in conn.notifies():
                payload = json.loads(notify.payload)
                await handler(payload)

# Logical replication consumer
class LogicalReplicationConsumer:
    def __init__(self, connection_string: str, slot_name: str):
        self.conn_string = connection_string
        self.slot_name = slot_name

    async def consume_changes(self, handler: callable):
        """Consume changes via logical replication"""
        conn = await asyncpg.connect(self.conn_string)

        # Create replication slot if not exists
        await conn.execute(f"""
            SELECT * FROM pg_create_logical_replication_slot(
                '{self.slot_name}',
                'pgoutput'
            )
        """)

        # Start replication
        async with conn.cursor(
            f"START_REPLICATION SLOT {self.slot_name} LOGICAL 0/0"
        ) as cur:
            async for change in cur:
                event = self.parse_wal_message(change)
                await handler(event)

# Redis Streams consumer group
class RedisStreamConsumer:
    def __init__(self, redis_client: redis.Redis, stream: str, group: str, consumer: str):
        self.redis = redis_client
        self.stream = stream
        self.group = group
        self.consumer = consumer

    async def consume(self, handler: callable):
        """Consume from stream group"""
        # Create consumer group if not exists
        try:
            await self.redis.xgroup_create(
                self.stream, self.group, id="0", mkstream=True
            )
        except redis.ResponseError as e:
            if "BUSYGROUP" not in str(e):
                raise

        while True:
            # Read messages
            messages = await self.redis.xreadgroup(
                self.group, self.consumer,
                {self.stream: ">"},
                count=10, block=1000
            )

            for stream_name, msg_list in messages:
                for msg_id, data in msg_list:
                    try:
                        await handler(msg_id, data)
                        # Acknowledge message
                        await self.redis.xack(self.stream, self.group, msg_id)
                    except Exception as e:
                        # Handle failure (retry, DLQ, etc.)
                        await self.handle_failure(msg_id, data, e)

class RedisStreamRecovery:
    async def recover_pending_messages(self, stream: str, group: str, consumer: str):
        """Recover pending messages for consumer"""
        # Get pending messages
        pending = await self.redis.xpending_range(
            stream, group, min="-", max="+", count=100,
            consumername=consumer
        )

        for msg in pending:
            msg_id = msg["message_id"]
            idle_time = msg["time_since_delivered"]

            # Claim if idle too long
            if idle_time > 60000:  # 60 seconds
                claimed = await self.redis.xclaim(
                    stream, group, consumer, 60000, msg_id
                )

                if claimed:
                    await self.process_message(msg_id, claimed[0][1])

class MultiWorkerCoordinator:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.worker_id = str(uuid.uuid4())

    async def coordinate_workers(self, stream: str, group: str, num_workers: int):
        """Coordinate multiple workers"""
        # Register worker
        await self.redis.sadd(f"workers:{group}", self.worker_id)
        await self.redis.expire(f"workers:{group}", 60)

        # Heartbeat
        asyncio.create_task(self.heartbeat(group))

        # Consume
        consumer = RedisStreamConsumer(
            self.redis, stream, group, f"worker-{self.worker_id}"
        )
        await consumer.consume(self.handle_message)

    async def heartbeat(self, group: str):
        """Send heartbeat to indicate worker is alive"""
        while True:
            await self.redis.setex(
                f"worker:{self.worker_id}:heartbeat",
                30,
                datetime.now().isoformat()
            )
            await asyncio.sleep(10)

# Kafka to TimescaleDB IoT pipeline
class KafkaTimescaleDBPipeline:
    def __init__(self, kafka_consumer, timescale_client):
        self.kafka = kafka_consumer
        self.timescale = timescale_client

    async def ingest_iot_data(self):
        """Ingest IoT data from Kafka to TimescaleDB"""
        async for msg in self.kafka:
            # Parse message
            data = json.loads(msg.value)

            # Downsample if needed
            if self.should_downsample(data):
                await self.downsample(data)
            else:
                # Insert raw data
                await self.timescale.execute("""
                    INSERT INTO sensor_data (device_id, timestamp, value, location)
                    VALUES (%s, %s, %s, ST_MakePoint(%s, %s))
                """, (
                    data["device_id"],
                    data["timestamp"],
                    data["value"],
                    data["lon"],
                    data["lat"]
                ))

    def should_downsample(self, data: dict) -> bool:
        """Determine if data should be downsampled"""
        # Downsample if older than 1 hour
        data_time = datetime.fromisoformat(data["timestamp"])
        return (datetime.now() - data_time).total_seconds() > 3600

    async def downsample(self, data: dict):
        """Downsample data to hourly aggregates"""
        await self.timescale.execute("""
            INSERT INTO sensor_data_hourly (device_id, hour, avg_value, location)
            VALUES (%s, date_trunc('hour', %s), %s, ST_MakePoint(%s, %s))
            ON CONFLICT (device_id, hour) 
            DO UPDATE SET avg_value = (sensor_data_hourly.avg_value + %s) / 2
        """, (
            data["device_id"],
            data["timestamp"],
            data["value"],
            data["lon"],
            data["lat"],
            data["value"]
        ))

# Kafka with schema registry
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

class SchemaRegistryProducer:
    def __init__(self, schema_registry_url: str):
        self.schema_registry = SchemaRegistryClient({
            "url": schema_registry_url
        })

    async def publish_with_schema(self, topic: str, event: dict, schema: str):
        """Publish event with schema validation"""
        # Get or register schema
        schema_id = await self.get_or_register_schema(topic, schema)

        # Serialize with schema
        serializer = AvroSerializer(
            self.schema_registry,
            schema,
            event
        )

        # Publish
        await self.kafka_producer.produce(topic, serializer)

# Kafka partitioning strategy
class KafkaPartitioner:
    def partition_key(self, event: dict) -> str:
        """Generate partition key for event"""
        # Strategy 1: By user_id (ensures user events in order)
        if "user_id" in event:
            return str(event["user_id"])

        # Strategy 2: By device_id (for IoT)
        if "device_id" in event:
            return str(event["device_id"])

        # Strategy 3: By geographic region
        if "location" in event:
            return self.geohash_partition(event["location"])

        # Default: round-robin
        return None

    def geohash_partition(self, location: dict) -> str:
        """Partition by geohash for geographic distribution"""
        import geohash2
        return geohash2.encode(location["lat"], location["lon"], precision=3)

# NATS for geospatial updates
import nats
from nats.aio.client import Client as NATS

class GeospatialNATSProducer:
    def __init__(self, nc: NATS):
        self.nc = nc

    async def publish_location_update(self, device_id: str, location: dict):
        """Publish real-time location update"""
        subject = f"location.updates.{device_id}"
        payload = json.dumps({
            "device_id": device_id,
            "timestamp": datetime.now().isoformat(),
            "lat": location["lat"],
            "lon": location["lon"],
            "accuracy": location.get("accuracy", 0)
        })

        await self.nc.publish(subject, payload.encode())

class GeospatialNATSConsumer:
    async def subscribe_location_updates(self, device_pattern: str):
        """Subscribe to location updates"""
        async def handler(msg):
            data = json.loads(msg.data.decode())
            await self.update_map(data)

        await self.nc.subscribe(f"location.updates.{device_pattern}", cb=handler)

# NATS command/control
class CommandControl:
    def __init__(self, nc: NATS):
        self.nc = nc

    async def send_command(self, device_id: str, command: dict):
        """Send command to device"""
        subject = f"commands.{device_id}"
        await self.nc.publish(subject, json.dumps(command).encode())

    async def request_response(self, device_id: str, request: dict, timeout: float = 5.0):
        """Request-response pattern"""
        subject = f"requests.{device_id}"
        response = await self.nc.request(
            subject,
            json.dumps(request).encode(),
            timeout=timeout
        )
        return json.loads(response.data.decode())

# NATS for sensor data
class SensorNATS:
    async def publish_sensor_data(self, sensor_id: str, data: dict):
        """Publish sensor data (lightweight)"""
        subject = f"sensors.{sensor_id}.data"
        # Use msgpack for efficiency
        import msgpack
        payload = msgpack.packb(data)
        await self.nc.publish(subject, payload)

# MQTT with QoS and LWT
import paho.mqtt.client as mqtt

class ProductionMQTTClient:
    def __init__(self, broker: str, client_id: str):
        self.client = mqtt.Client(client_id=client_id)
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        self.client.connect(broker, 1883, 60)

    def on_connect(self, client, userdata, flags, rc):
        """Handle connection"""
        if rc == 0:
            # Subscribe with QoS 1 (at-least-once)
            client.subscribe("sensors/+/data", qos=1)
        else:
            raise ConnectionError(f"MQTT connection failed: {rc}")

    def setup_lwt(self, topic: str, payload: str):
        """Setup Last Will and Testament"""
        self.client.will_set(topic, payload, qos=1, retain=False)

    def publish_with_retain(self, topic: str, payload: dict, retain: bool = True):
        """Publish with retain flag for last known state"""
        self.client.publish(
            topic,
            json.dumps(payload),
            qos=1,
            retain=retain
        )

class MQTTHeartbeat:
    def __init__(self, client: mqtt.Client, device_id: str):
        self.client = client
        self.device_id = device_id
        self.heartbeat_topic = f"devices/{device_id}/heartbeat"

    async def start_heartbeat(self, interval: int = 30):
        """Start heartbeat loop"""
        while True:
            self.client.publish(
                self.heartbeat_topic,
                json.dumps({
                    "device_id": self.device_id,
                    "timestamp": datetime.now().isoformat(),
                    "status": "alive"
                }),
                qos=1
            )
            await asyncio.sleep(interval)

# Schema evolution with versioning
class EventSchema:
    def __init__(self, event_type: str, version: int, schema: dict):
        self.event_type = event_type
        self.version = version
        self.schema = schema

    def validate(self, event: dict) -> bool:
        """Validate event against schema"""
        import jsonschema
        try:
            jsonschema.validate(event, self.schema)
            return True
        except jsonschema.ValidationError:
            return False

    def migrate(self, event: dict, target_version: int) -> dict:
        """Migrate event to target version"""
        if self.version == target_version:
            return event

        # Apply migration rules
        migrated = event.copy()
        for v in range(self.version, target_version):
            migrated = self.apply_migration(migrated, v, v + 1)

        return migrated

# Event contract definition
from dataclasses import dataclass
from typing import Optional
from datetime import datetime

@dataclass
class UserCreatedEvent:
    """User created event contract"""
    user_id: str
    username: str
    email: str
    created_at: datetime

    def to_dict(self) -> dict:
        return {
            "type": "user.created",
            "version": 1,
            "payload": {
                "user_id": self.user_id,
                "username": self.username,
                "email": self.email,
                "created_at": self.created_at.isoformat()
            }
        }

    @classmethod
    def from_dict(cls, data: dict) -> 'UserCreatedEvent':
        return cls(
            user_id=data["payload"]["user_id"],
            username=data["payload"]["username"],
            email=data["payload"]["email"],
            created_at=datetime.fromisoformat(data["payload"]["created_at"])
        )

# Python with Protocol Buffers
from events_pb2 import UserCreated, EventEnvelope

class ProtobufEventProducer:
    def publish_user_created(self, user_id: str, username: str, email: str):
        """Publish user created event with protobuf"""
        event = UserCreated(
            user_id=user_id,
            username=username,
            email=email,
            created_at=int(datetime.now().timestamp())
        )

        envelope = EventEnvelope(
            event_type="user.created",
            version=1,
            timestamp=int(datetime.now().timestamp()),
            payload=event.SerializeToString()
        )

        await self.broker.publish("user.events", envelope.SerializeToString())

# Avro schema and serialization
from avro.schema import parse
import avro.io
import io

class AvroEventProducer:
    def __init__(self, schema_json: str):
        self.schema = parse(schema_json)

    def serialize(self, event: dict) -> bytes:
        """Serialize event with Avro"""
        writer = avro.io.DatumWriter(self.schema)
        bytes_writer = io.BytesIO()
        encoder = avro.io.BinaryEncoder(bytes_writer)
        writer.write(event, encoder)
        return bytes_writer.getvalue()

    def deserialize(self, data: bytes) -> dict:
        """Deserialize Avro event"""
        reader = avro.io.DatumReader(self.schema)
        bytes_reader = io.BytesIO(data)
        decoder = avro.io.BinaryDecoder(bytes_reader)
        return reader.read(decoder)

# RDF/OWL event metadata
from rdflib import Graph, Namespace, Literal
from rdflib.namespace import RDF, RDFS

class RDFEventEnvelope:
    def __init__(self):
        self.graph = Graph()
        self.EV = Namespace("http://example.org/events/")

    def create_envelope(self, event: dict) -> str:
        """Create RDF envelope for event"""
        event_id = self.EV[f"event/{uuid.uuid4()}"]

        self.graph.add((event_id, RDF.type, self.EV.Event))
        self.graph.add((event_id, self.EV.eventType, Literal(event["type"])))
        self.graph.add((event_id, self.EV.timestamp, Literal(event["timestamp"])))
        self.graph.add((event_id, self.EV.payload, Literal(json.dumps(event["payload"]))))

        return self.graph.serialize(format="json-ld")

# Event signature for reproducibility
import hashlib
import hmac

class EventSigner:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key.encode()

    def sign_event(self, event: dict) -> str:
        """Sign event for authenticity"""
        # Create canonical representation
        canonical = json.dumps(event, sort_keys=True, separators=(',', ':'))

        # Generate HMAC signature
        signature = hmac.new(
            self.secret_key,
            canonical.encode(),
            hashlib.sha256
        ).hexdigest()

        return signature

    def verify_event(self, event: dict, signature: str) -> bool:
        """Verify event signature"""
        expected = self.sign_event(event)
        return hmac.compare_digest(expected, signature)

# Queue capacity planning
class QueueCapacityPlanner:
    def calculate_queue_size(
        self,
        message_rate: float,  # messages per second
        processing_time: float,  # seconds per message
        peak_multiplier: float = 3.0,
        retention_hours: float = 24.0
    ) -> int:
        """Calculate required queue size"""
        # Base queue size
        base_size = message_rate * processing_time

        # Peak capacity
        peak_size = base_size * peak_multiplier

        # Retention capacity
        retention_size = message_rate * 3600 * retention_hours

        # Total required
        total_size = max(peak_size, retention_size)

        return int(total_size * 1.2)  # 20% buffer