Event-Driven Architecture: от теории к production

# ❌ Так мы сделали (и поплатились)
from fastapi import FastAPI, HTTPException
import httpx
 
app = FastAPI()
 
@app.post("/checkout")
async def checkout(order: CheckoutRequest):
    async with httpx.AsyncClient(timeout=5.0) as client:
        # 1. Проверяем пользователя
        user = await client.get(f"http://auth-service/users/{order.user_id}")
        if user.status_code != 200:
            raise HTTPException(401, "User not found")
 
        # 2. Валидируем товары
        catalog = await client.post(
            "http://catalog-service/validate",
            json={"items": order.items}
        )
        if catalog.status_code != 200:
            raise HTTPException(400, "Invalid items")
 
        # 3. Проверяем промокод
        discount = await client.post(
            "http://discount-service/validate",
            json={"code": order.promo_code}
        )
 
        # 4. Резервируем на складе
        inventory = await client.post(
            "http://inventory-service/reserve",
            json={"items": order.items}
        )
        if inventory.status_code != 200:
            raise HTTPException(400, "Out of stock")
 
        # 5. Создаём заказ
        order_response = await client.post(
            "http://order-service/orders",
            json=order.dict()
        )
 
        # 6. Списываем деньги
        payment = await client.post(
            "http://payment-service/charge",
            json={"amount": order.total, "user_id": order.user_id}
        )
        if payment.status_code != 200:
            # О нет! Заказ создан, но оплата не прошла!
            # Нужен rollback... но как?
            raise HTTPException(500, "Payment failed")
 
        # 7. Отправляем email
        await client.post(
            "http://notification-service/send",
            json={"email": user.json()["email"], "order_id": order_response.json()["id"]}
        )
 
        return {"order_id": order_response.json()["id"]}

# ✅ Event-driven подход
from fastapi import FastAPI, HTTPException
from kafka import KafkaProducer
import json
 
app = FastAPI()
 
# Kafka producer (можно Redis Streams, RabbitMQ и т.д.)
producer = KafkaProducer(
    bootstrap_servers='kafka:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
 
@app.post("/checkout")
async def checkout(order: CheckoutRequest):
    # СИНХРОННО (критично для UX)
    # 1. Быстрая валидация
    validate_order_data(order)  # 5ms, в памяти
 
    # 2. Резерв на складе (единственный HTTP-запрос!)
    async with httpx.AsyncClient(timeout=2.0) as client:
        inventory = await client.post(
            "http://inventory-service/reserve",
            json={"items": order.items}
        )
        if inventory.status_code != 200:
            raise HTTPException(400, "Out of stock")
 
    # 3. Создаём заказ в статусе PENDING
    order_id = await db.create_order(order, status="PENDING")
 
    # АСИНХРОННО (через события)
    # Публикуем событие — дальше не наша проблема
    producer.send('orders.created', {
        'order_id': order_id,
        'user_id': order.user_id,
        'items': order.items,
        'total': order.total,
        'timestamp': datetime.utcnow().isoformat()
    })
 
    # Возвращаем результат мгновенно!
    return {
        "order_id": order_id,
        "status": "pending",
        "message": "Order created, payment processing"
    }

# Payment Service (слушает события "orders.created")
from kafka import KafkaConsumer
import json
 
consumer = KafkaConsumer(
    'orders.created',
    bootstrap_servers='kafka:9092',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='payment-service'
)
 
for message in consumer:
    order = message.value
 
    try:
        # Списываем деньги
        payment = await charge_payment(order['user_id'], order['total'])
 
        # Публикуем событие об успехе
        producer.send('payments.completed', {
            'order_id': order['order_id'],
            'payment_id': payment.id,
            'status': 'success'
        })
 
    except PaymentError as e:
        # Публикуем событие об ошибке
        producer.send('payments.failed', {
            'order_id': order['order_id'],
            'reason': str(e)
        })

# Notification Service (слушает события "payments.completed")
consumer = KafkaConsumer(
    'payments.completed',
    bootstrap_servers='kafka:9092',
    group_id='notification-service'
)
 
for message in consumer:
    payment = message.value
 
    # Отправляем email (даже если упадёт — не повлияет на checkout)
    await send_email(
        user_id=payment['user_id'],
        template='order_confirmation',
        data={'order_id': payment['order_id']}
    )

# Publisher
producer.send('user.registered', {
    'event_id': 'evt_123',
    'user_id': 'usr_456',
    'timestamp': '2025-12-26T10:00:00Z'
})
 
# Consumer (должен запросить данные сам)
@app.on_event('user.registered')
async def on_user_registered(event):
    # Делаем HTTP-запрос за данными
    user = await user_service.get_user(event['user_id'])
    await send_welcome_email(user.email)

# Publisher
producer.send('user.registered', {
    'event_id': 'evt_123',
    'user_id': 'usr_456',
    'email': 'user@example.com',
    'name': 'John Doe',
    'created_at': '2025-12-26T10:00:00Z',
    'subscription_plan': 'premium'
})
 
# Consumer (не делает HTTP-запросов!)
@app.on_event('user.registered')
async def on_user_registered(event):
    # Всё уже есть в событии
    await send_welcome_email(
        email=event['email'],
        name=event['name']
    )

# Традиционный подход (CRUD)
class Order:
    id: int
    status: str  # "pending" | "paid" | "shipped" | "completed"
    total: float
 
# При изменении перезаписываем строку в БД
await db.execute("UPDATE orders SET status = 'paid' WHERE id = 123")

# Event Sourcing подход
# События (append-only log)
events = [
    {'type': 'OrderCreated', 'order_id': 123, 'total': 1500, 'timestamp': '2025-12-26T10:00:00Z'},
    {'type': 'PaymentReceived', 'order_id': 123, 'amount': 1500, 'timestamp': '2025-12-26T10:05:00Z'},
    {'type': 'OrderShipped', 'order_id': 123, 'tracking': 'TR123', 'timestamp': '2025-12-26T12:00:00Z'},
]
 
# Состояние = replay всех событий
def get_order_state(order_id):
    events = event_store.get_events(order_id)
    state = {}
    for event in events:
        if event['type'] == 'OrderCreated':
            state['status'] = 'pending'
            state['total'] = event['total']
        elif event['type'] == 'PaymentReceived':
            state['status'] = 'paid'
        elif event['type'] == 'OrderShipped':
            state['status'] = 'shipped'
    return state

# Command side (write)
class CreateOrderCommand:
    user_id: int
    items: List[Item]
 
@app.post("/orders")
async def create_order(cmd: CreateOrderCommand):
    # Валидация
    # Создание событий
    events = [
        OrderCreatedEvent(user_id=cmd.user_id, items=cmd.items),
        InventoryReservedEvent(items=cmd.items)
    ]
    # Сохранение в event store
    await event_store.append(events)
    return {"order_id": events[0].order_id}
 
# Query side (read)
@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    # Читаем из materialized view (быстро!)
    return await read_db.get_order(order_id)
 
# Projector (обновляет read model из событий)
@consumer.on('OrderCreatedEvent')
async def project_order(event):
    await read_db.insert_order({
        'id': event.order_id,
        'user_id': event.user_id,
        'status': 'pending',
        'created_at': event.timestamp
    })

from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json
 
# Producer
producer = AIOKafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode()
)
await producer.start()
await producer.send('orders.created', {'order_id': 123})
 
# Consumer
consumer = AIOKafkaConsumer(
    'orders.created',
    bootstrap_servers='localhost:9092',
    group_id='payment-service',
    value_deserializer=lambda m: json.loads(m.decode())
)
await consumer.start()
async for msg in consumer:
    print(f"Order: {msg.value}")

services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    ports:
      - "9092:9092"
    environment:
      # KRaft mode (без ZooKeeper)
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"
      KAFKA_LISTENERS: "PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      # Форматирование storage (обязательно для KRaft)
      CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"

import aio_pika
import json
 
# Publisher
connection = await aio_pika.connect_robust("amqp://localhost/")
channel = await connection.channel()
 
await channel.default_exchange.publish(
    aio_pika.Message(
        body=json.dumps({'order_id': 123}).encode(),
        delivery_mode=aio_pika.DeliveryMode.PERSISTENT
    ),
    routing_key='orders.created'
)
 
# Consumer
queue = await channel.declare_queue('payment-service-queue', durable=True)
await queue.bind(exchange='events', routing_key='orders.created')
 
async with queue.iterator() as queue_iter:
    async for message in queue_iter:
        async with message.process():
            data = json.loads(message.body.decode())
            print(f"Order: {data}")

import redis.asyncio as redis
import json
 
r = await redis.from_url("redis://localhost")
 
# Producer
await r.xadd(
    'orders:created',
    {'data': json.dumps({'order_id': 123})}
)
 
# Consumer
consumer_group = 'payment-service'
await r.xgroup_create('orders:created', consumer_group, mkstream=True)
 
while True:
    messages = await r.xreadgroup(
        consumer_group,
        'worker-1',
        {'orders:created': '>'},
        count=10
    )
    for stream, msgs in messages:
        for msg_id, data in msgs:
            order = json.loads(data[b'data'])
            print(f"Order: {order}")
            await r.xack('orders:created', consumer_group, msg_id)

# User регистрируется
await db.create_user(user_id=123, email='user@example.com')
producer.send('user.registered', {'user_id': 123})
 
# ...10ms спустя...
 
# Другой сервис пытается получить user
user = await user_service.get_user(123)  # Может вернуть None!

# После создания возвращаем данные сразу
@app.post("/users")
async def create_user(user: UserCreate):
    user_entity = await db.create_user(user)
    producer.send('user.registered', user_entity.dict())
 
    # Возвращаем созданного user
    return user_entity  # Не ждём event processing!

# Frontend polling
POST /users → { "user_id": 123, "status": "pending" }
 
# Frontend polls
GET /users/123 → { "user_id": 123, "status": "pending" }
...
GET /users/123 → { "user_id": 123, "status": "active" }  # Готово!

// Frontend предполагает успех
onSubmit() {
  // Показываем сразу (оптимистично)
  users.add({ id: 123, email: 'user@example.com', status: 'pending' })
 
  // Отправляем на backend
  await api.createUser(...)
 
  // WebSocket/SSE обновит status когда обработается
}

# ❌ НЕ idempotent
@consumer.on('payment.received')
async def on_payment(event):
    # Если event обработается 2 раза → начислим бонусы дважды!
    await db.execute(
        "UPDATE users SET bonus_balance = bonus_balance + 100 WHERE id = ?",
        event['user_id']
    )

# ✅ Idempotent
@consumer.on('payment.received')
async def on_payment(event):
    # Проверяем, обрабатывали ли уже это событие
    processed = await db.fetchval(
        "SELECT 1 FROM processed_events WHERE event_id = ?",
        event['event_id']
    )
    if processed:
        return  # Skip duplicate
 
    # Обрабатываем в транзакции
    async with db.transaction():
        await db.execute(
            "UPDATE users SET bonus_balance = bonus_balance + 100 WHERE id = ?",
            event['user_id']
        )
        await db.execute(
            "INSERT INTO processed_events (event_id, processed_at) VALUES (?, ?)",
            event['event_id'], datetime.utcnow()
        )

# PostgreSQL UPSERT (идемпотентно по дизайну)
await db.execute("""
    INSERT INTO user_bonuses (user_id, payment_id, amount)
    VALUES (?, ?, 100)
    ON CONFLICT (payment_id) DO NOTHING
""", event['user_id'], event['payment_id'])

# Публикуются события
producer.send('order.created', {'order_id': 123, 'status': 'pending'})
producer.send('order.paid', {'order_id': 123, 'status': 'paid'})
 
# Consumer может получить в обратном порядке!
1. order.paid (status = paid)
2. order.created (status = pending)
 
# Результат: Order в неправильном состоянии

# Kafka гарантирует порядок внутри partition
# Отправляем все события одного order в одну partition
await producer.send(
    'orders',
    value={'event': 'order.created', 'order_id': 123},
    key=str(123).encode()  # Partition key = order_id
)

# Добавляем sequence number в event
events = [
    {'order_id': 123, 'event': 'created', 'seq': 1},
    {'order_id': 123, 'event': 'paid', 'seq': 2},
    {'order_id': 123, 'event': 'shipped', 'seq': 3},
]
 
# Consumer проверяет sequence
@consumer.on('order.*')
async def on_order_event(event):
    current_seq = await db.fetchval("SELECT seq FROM orders WHERE id = ?", event['order_id'])
 
    if event['seq'] <= current_seq:
        return  # Старое событие, игнорируем
 
    # Применяем событие
    await apply_event(event)

@consumer.on('payment.received')
async def on_payment(event):
    # Внешний API упал
    await stripe.refund(event['payment_id'])  # Timeout!

from tenacity import retry, stop_after_attempt, wait_exponential
 
@consumer.on('payment.received')
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60)
)
async def on_payment(event):
    try:
        await stripe.refund(event['payment_id'])
    except Exception as e:
        # После 5 попыток → отправляем в DLQ
        if should_send_to_dlq(e):
            await producer.send('dlq.payments', event)
            raise

from aiokafka import AIOKafkaConsumer
 
consumer = AIOKafkaConsumer(
    'payments.received',
    enable_auto_commit=False,  # Manual commit
    max_poll_records=10
)
 
async for msg in consumer:
    try:
        await process_payment(msg.value)
        await consumer.commit()  # Success
    except Exception as e:
        # Send to DLQ
        await producer.send('dlq.payments', msg.value)
        await consumer.commit()  # Commit чтобы не retry
        logger.error(f"Failed to process {msg.value}: {e}")

from prometheus_client import Counter, Histogram, Gauge
 
# Количество событий
events_published = Counter(
    'events_published_total',
    'Total events published',
    ['topic', 'event_type']
)
 
# Latency обработки
event_processing_duration = Histogram(
    'event_processing_duration_seconds',
    'Event processing duration',
    ['consumer', 'topic']
)
 
# Consumer lag (Kafka specific)
consumer_lag = Gauge(
    'kafka_consumer_lag',
    'Consumer lag in messages',
    ['group', 'topic', 'partition']
)
 
# Dead letter queue size
dlq_size = Gauge(
    'dlq_messages_total',
    'Messages in DLQ',
    ['topic']
)

@app.post("/orders")
async def create_order(order: OrderCreate):
    start_time = time.time()
 
    # Создаём заказ
    order_id = await db.create_order(order)
 
    # Публикуем событие
    producer.send('orders.created', {'order_id': order_id})
    events_published.labels(topic='orders.created', event_type='OrderCreated').inc()
 
    # Latency
    event_processing_duration.labels(
        consumer='order-service',
        topic='orders.created'
    ).observe(time.time() - start_time)
 
    return {"order_id": order_id}

# Throughput (events/sec)
rate(events_published_total[1m])
 
# Consumer lag (warning если > 1000)
kafka_consumer_lag > 1000
 
# P99 latency
histogram_quantile(0.99, rate(event_processing_duration_seconds_bucket[5m]))
 
# DLQ growth rate (alert если растёт)
delta(dlq_messages_total[5m]) > 10