event-driven-architecture

What this skill does

# Event-Driven Architecture

## Overview

This skill covers designing and implementing event-driven systems that decouple services through asynchronous message passing. It addresses message broker selection and integration (Kafka, RabbitMQ, SQS/SNS, NATS), event sourcing and CQRS patterns, saga orchestration for distributed transactions, dead letter queues for failure handling, idempotency patterns, event schema evolution, the transactional outbox pattern, and consumer group management.

Use this skill when building microservice architectures, implementing distributed workflows, decoupling services for independent deployment, handling eventual consistency, or replacing synchronous service-to-service calls with asynchronous events.

---

## Core Principles

1. **Events are facts, commands are requests** - Events describe something that happened (`OrderPlaced`, `PaymentReceived`). Commands request an action (`ProcessPayment`, `ShipOrder`). This distinction drives correct system design: events are broadcast, commands are point-to-point.
2. **Idempotency is not optional** - Messages will be delivered at least once (and sometimes more). Every consumer must handle duplicate messages gracefully. Use event IDs, version checks, or database constraints for deduplication.
3. **Schema evolution without breaking consumers** - Event schemas will change. Use backward-compatible evolution (add fields, never remove or rename). Version schemas explicitly and support multiple versions during migration.
4. **Dead letters are not garbage** - Messages that can't be processed go to dead letter queues. These represent bugs, data issues, or edge cases. Monitor DLQs, alert on growth, and build tooling to replay them.
5. **Local transactions, eventual consistency** - Each service owns its data and processes events within local transactions. Cross-service consistency is eventual, not immediate. Design UIs and workflows to handle this.

---

## Key Patterns

### Pattern 1: Event Publishing with Transactional Outbox

**When to use:** When you need to update a database AND publish an event atomically. This prevents the "dual write" problem where the database write succeeds but the event publish fails (or vice versa).

**Implementation:**

```typescript
// The Transactional Outbox Pattern
// 1. Write the event to an outbox table in the SAME database transaction as the business data
// 2. A separate poller/CDC process reads the outbox and publishes to the message broker

interface OutboxEvent {
  id: string;
  aggregateType: string;
  aggregateId: string;
  eventType: string;
  payload: Record<string, unknown>;
  createdAt: Date;
  publishedAt: Date | null;
}

// Step 1: Business operation + outbox write in one transaction
async function placeOrder(order: CreateOrderInput): Promise<Order> {
  return await prisma.$transaction(async (tx) => {
    // Business logic
    const createdOrder = await tx.order.create({
      data: {
        customerId: order.customerId,
        items: { create: order.items },
        total: calculateTotal(order.items),
        status: "PLACED",
      },
      include: { items: true },
    });

    // Write event to outbox (same transaction!)
    await tx.outboxEvent.create({
      data: {
        id: crypto.randomUUID(),
        aggregateType: "Order",
        aggregateId: createdOrder.id,
        eventType: "OrderPlaced",
        payload: {
          orderId: createdOrder.id,
          customerId: createdOrder.customerId,
          items: createdOrder.items,
          total: createdOrder.total,
          placedAt: new Date().toISOString(),
        },
      },
    });

    return createdOrder;
  });
}

// Step 2: Outbox poller publishes events to Kafka/RabbitMQ
async function processOutbox(): Promise<void> {
  const unpublished = await prisma.outboxEvent.findMany({
    where: { publishedAt: null },
    orderBy: { createdAt: "asc" },
    take: 100,
  });

  for (const event of unpublished) {
    try {
      await kafka.producer.send({
        topic: `${event.aggregateType}.${event.eventType}`,
        messages: [
          {
            key: event.aggregateId,
            value: JSON.stringify({
              eventId: event.id,
              eventType: event.eventType,
              aggregateId: event.aggregateId,
              payload: event.payload,
              timestamp: event.createdAt.toISOString(),
            }),
            headers: {
              "event-type": event.eventType,
              "event-id": event.id,
            },
          },
        ],
      });

      await prisma.outboxEvent.update({
        where: { id: event.id },
        data: { publishedAt: new Date() },
      });
    } catch (error) {
      console.error(`Failed to publish outbox event ${event.id}:`, error);
      // Will be retried on next poll cycle
    }
  }
}

// Run poller on interval
setInterval(processOutbox, 5000); // Every 5 seconds
```

**Why:** The dual-write problem is the most common source of data inconsistency in event-driven systems. If you write to the database and then publish an event, the publish can fail after the database write succeeds. The outbox pattern guarantees that if the business data is written, the event will eventually be published.

---

### Pattern 2: Idempotent Event Consumer

**When to use:** Every event consumer. Messages are delivered at least once in all major brokers.

**Implementation:**

```typescript
// Kafka consumer with idempotency
import { Kafka, EachMessagePayload } from "kafkajs";

const kafka = new Kafka({
  clientId: "order-service",
  brokers: [process.env.KAFKA_BROKERS!],
});

const consumer = kafka.consumer({ groupId: "payment-processor" });

interface DomainEvent {
  eventId: string;
  eventType: string;
  aggregateId: string;
  payload: Record<string, unknown>;
  timestamp: string;
}

// Event handler registry
const handlers: Record<string, (event: DomainEvent) => Promise<void>> = {
  OrderPlaced: async (event) => {
    const { orderId, total, customerId } = event.payload as {
      orderId: string;
      total: number;
      customerId: string;
    };

    // Create payment intent (idempotent via orderId)
    await processPayment(orderId, total, customerId);
  },

  PaymentFailed: async (event) => {
    const { orderId } = event.payload as { orderId: string };
    await cancelOrder(orderId);
  },
};

async function startConsumer() {
  await consumer.connect();
  await consumer.subscribe({
    topics: ["Order.OrderPlaced", "Payment.PaymentFailed"],
    fromBeginning: false,
  });

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

      // 1. Idempotency check - have we already processed this event?
      const alreadyProcessed = await prisma.processedEvent.findUnique({
        where: { eventId: event.eventId },
      });

      if (alreadyProcessed) {
        console.log(`Skipping duplicate event: ${event.eventId}`);
        return;
      }

      // 2. Process the event
      const handler = handlers[event.eventType];
      if (!handler) {
        console.warn(`No handler for event type: ${event.eventType}`);
        return;
      }

      try {
        await prisma.$transaction(async (tx) => {
          // Record event as processed (within the same transaction as business logic)
          await tx.processedEvent.create({
            data: {
              eventId: event.eventId,
              eventType: event.eventType,
              processedAt: new Date(),
            },
          });

          // Execute business logic
          await handler(event);
        });
      } catch (error) {
        console.error(`Error processing event ${event.eventId}:`, error);
        // Don't commit offset - message will be redelivered
        throw error;
      }
    },
  });
}
```

**Why:** Kafka, RabbitMQ, and SQS all guarantee at-least-once delivery, meaning consumers will see duplicates during rebalan