Реалізація Event-Driven Architecture для веб-застосунку

Реалізація Event-Driven Architecture для веб-додатків

Event-Driven Architecture (EDA) — архітектурний стиль, у якому компоненти системи взаємодіють через публікацію та споживання подій. Продюсер публікує подію в брокер і не знає, хто її обробить. Консюмери підписуються на цікаві їм події незалежно. Це радикально знижує зв'язність (coupling) між компонентами.

Коли EDA доречна

• Потрібно сповістити кілька систем про одну подію (новий заказ → інвентаризація + сповіщення + аналітика)
• Обробка займає час і блокування HTTP-запиту недоцільно
• Пікова навантаження, яку потрібно згладити через чергу
• Аудит і історія змін
• Інтеграція з зовнішніми системами через webhooks або CDC

Структура події

interface DomainEvent<T = unknown> {
  id: string;            // UUID — для ідемпотентності
  type: string;          // 'user.registered', 'order.placed'
  version: string;       // '1.0' — для schema evolution
  source: string;        // 'order-service'
  correlationId: string; // наскрізний ID через всі сервіси
  causationId?: string;  // ID події, яка стала причиною
  occurredAt: string;    // ISO 8601
  data: T;
}

// Конкретна подія
interface OrderPlacedEvent extends DomainEvent<{
  orderId: string;
  customerId: string;
  items: Array<{ productId: string; quantity: number; price: number }>;
  total: number;
  shippingAddress: Address;
}> {
  type: 'order.placed';
}

Apache Kafka — основний брокер для EDA

import { Kafka, Partitioners } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'order-service',
  brokers: process.env.KAFKA_BROKERS.split(',')
});

// Продюсер
const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner
});

async function publishOrderPlaced(order: Order): Promise<void> {
  await producer.send({
    topic: 'order.events',
    messages: [{
      key: order.id,  // партиціювання за ID заказу
      value: JSON.stringify({
        id: uuidv4(),
        type: 'order.placed',
        version: '1.0',
        source: 'order-service',
        correlationId: context.correlationId,
        occurredAt: new Date().toISOString(),
        data: {
          orderId: order.id,
          customerId: order.customerId,
          items: order.items,
          total: order.total
        }
      } satisfies OrderPlacedEvent),
      headers: {
        'content-type': 'application/json',
        'schema-version': '1.0'
      }
    }]
  });
}

// Консюмер — Inventory Service
const consumer = kafka.consumer({ groupId: 'inventory-service' });

await consumer.subscribe({ topics: ['order.events'], fromBeginning: false });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString()) as DomainEvent;

    // Ідемпотентність: перевіряємо, чи не обробили це вже
    const processed = await idempotencyRepo.exists(event.id);
    if (processed) return;

    try {
      if (event.type === 'order.placed') {
        await inventoryService.reserveStock(event.data.orderId, event.data.items);
      }
      await idempotencyRepo.mark(event.id);
    } catch (error) {
      // Публікуємо в Dead Letter Topic для аналізу
      await deadLetterProducer.send({
        topic: 'order.events.dlq',
        messages: [{
          value: message.value,
          headers: { 'failure-reason': error.message }
        }]
      });
    }
  }
});

Outbox Pattern — гарантована доставка

Помилка: зберегти в БД, а потім опублікувати в Kafka — ризик втрати події при збої між кроками. Правильно — Transactional Outbox:

// У межах однієї транзакції БД
async function createOrder(dto: CreateOrderDto): Promise<Order> {
  return db.transaction(async (trx) => {
    // 1. Зберігаємо замовлення
    const order = await trx('orders').insert({ ...orderData }).returning('*');

    // 2. Зберігаємо подію в outbox-таблицю (у тій же транзакції!)
    await trx('outbox_events').insert({
      id: uuidv4(),
      aggregate_id: order.id,
      event_type: 'order.placed',
      payload: JSON.stringify(orderPlacedEvent),
      status: 'pending',
      created_at: new Date()
    });

    return order;
  });
}

Окремий Outbox Poller читає pending-події та публікує їх у Kafka:

// Cron job або background worker
async function processOutbox(): Promise<void> {
  const events = await db('outbox_events')
    .where({ status: 'pending' })
    .orderBy('created_at')
    .limit(100)
    .forUpdate()
    .skipLocked();

  for (const event of events) {
    try {
      await kafka.producer.send({
        topic: getTopicForEventType(event.event_type),
        messages: [{ key: event.aggregate_id, value: event.payload }]
      });
      await db('outbox_events')
        .where({ id: event.id })
        .update({ status: 'published', published_at: new Date() });
    } catch {
      await db('outbox_events')
        .where({ id: event.id })
        .update({ retry_count: db.raw('retry_count + 1') });
    }
  }
}

Альтернатива — Debezium CDC: читає WAL PostgreSQL та публікує зміни в Kafka без додаткового коду.

Хореографія проти оркестрації

Event Sourcing як частковий випадок EDA

При Event Sourcing усі зміни стану — це події, які публікуються в event store і в брокер. Проекції будуються на основі цих же подій. EDA та ES хорошо працюють разом.

Терміни реалізації

• EDA для одного сценарію (один продюсер + 2–3 консюмери) — 1–2 тижні
• Outbox Pattern + ідемпотентність + DLQ — ще 1 тиждень
• Повна EDA для 5–10 сервісів з моніторингом — 4–8 тижнів