Event-Driven Architecture: Patterns and Pitfalls

// Events: past tense, factual, immutable
interface OrderCreatedEvent {
  type: "order.created";
  timestamp: string;
  data: {
    orderId: string;
    userId: string;
    items: Array<{ productId: string; quantity: number; price: number }>;
    totalAmount: number;
  };
}
 
// Commands: imperative, directed at a specific handler
interface SendEmailCommand {
  type: "send.email";
  data: {
    to: string;
    template: string;
    variables: Record<string, string>;
  };
}
 
// ❌ Mixing events and commands
// "OrderCreated" event that also tells the email service what to do
interface BadEvent {
  type: "order.created";
  data: {
    orderId: string;
    emailTo: string;           // Why does the order event know about emails?
    emailTemplate: "receipt";  // This couples the producer to the consumer
  };
}
 
// ✅ Clean separation — event carries facts, consumers decide what to do
// Order service publishes: OrderCreatedEvent
// Email service listens, looks up user email, sends receipt
// Inventory service listens, decrements stock
// Analytics service listens, records conversion

type EventHandler<T = unknown> = (event: T) => Promise<void>;
 
class EventBus {
  private handlers = new Map<string, EventHandler[]>();
 
  on<T>(eventType: string, handler: EventHandler<T>): void {
    const existing = this.handlers.get(eventType) ?? [];
    existing.push(handler as EventHandler);
    this.handlers.set(eventType, existing);
  }
 
  async emit<T extends { type: string }>(event: T): Promise<void> {
    const handlers = this.handlers.get(event.type) ?? [];
 
    // Execute all handlers concurrently
    const results = await Promise.allSettled(
      handlers.map(handler => handler(event))
    );
 
    // Log failures without blocking the producer
    results.forEach((result, index) => {
      if (result.status === "rejected") {
        console.error(`Handler ${index} for ${event.type} failed:`, result.reason);
      }
    });
  }
}
 
// Usage
const bus = new EventBus();
 
bus.on<OrderCreatedEvent>("order.created", async (event) => {
  await sendReceiptEmail(event.data.userId, event.data.orderId);
});
 
bus.on<OrderCreatedEvent>("order.created", async (event) => {
  await decrementInventory(event.data.items);
});
 
bus.on<OrderCreatedEvent>("order.created", async (event) => {
  await recordAnalytics("conversion", event.data);
});

// ❌ Assuming immediate consistency
app.post("/orders", async (req, res) => {
  const order = await createOrder(req.body);
  await eventBus.emit({ type: "order.created", data: order });
 
  // Problem: client immediately queries order status
  // but the inventory service hasn't processed the event yet
  res.json({ orderId: order.id, status: "confirmed" });
});
 
// ✅ Acknowledging async processing
app.post("/orders", async (req, res) => {
  const order = await createOrder(req.body);
  await eventBus.emit({ type: "order.created", data: order });
 
  // Return 202 Accepted — processing is asynchronous
  res.status(202).json({
    orderId: order.id,
    status: "processing",
    statusUrl: `/orders/${order.id}/status`, // Client can poll for updates
  });
});
 
// Status endpoint reflects the actual processed state
app.get("/orders/:id/status", async (req, res) => {
  const order = await getOrder(req.params.id);
  res.json({
    orderId: order.id,
    status: order.status,        // "processing" | "confirmed" | "failed"
    inventoryReserved: order.inventoryReserved,
    paymentCaptured: order.paymentCaptured,
    updatedAt: order.updatedAt,
  });
});

// ❌ Non-idempotent handler — double-charges the customer
async function handlePaymentEvent(event: OrderCreatedEvent) {
  await chargeCustomer(event.data.userId, event.data.totalAmount);
  // If this event is delivered twice, the customer is charged twice
}
 
// ✅ Idempotent handler — uses event ID for deduplication
async function handlePaymentEvent(event: OrderCreatedEvent & { id: string }) {
  // Check if we've already processed this event
  const processed = await db.query(
    "SELECT 1 FROM processed_events WHERE event_id = $1",
    [event.id]
  );
 
  if (processed.rows.length > 0) {
    console.log(`Event ${event.id} already processed, skipping`);
    return;
  }
 
  // Process within a transaction
  await db.transaction(async (tx) => {
    await tx.query(
      "INSERT INTO processed_events (event_id, processed_at) VALUES ($1, NOW())",
      [event.id]
    );
    await tx.query(
      "INSERT INTO payments (order_id, amount, status) VALUES ($1, $2, 'captured')",
      [event.data.orderId, event.data.totalAmount]
    );
  });
}

async function processWithRetry(
  event: unknown,
  handler: EventHandler,
  maxRetries: number = 3
): Promise<void> {
  let lastError: Error | undefined;
 
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      await handler(event);
      return;
    } catch (error) {
      lastError = error instanceof Error ? error : new Error(String(error));
      console.warn(`Attempt ${attempt}/${maxRetries} failed:`, lastError.message);
 
      if (attempt < maxRetries) {
        // Exponential backoff
        await new Promise(resolve =>
          setTimeout(resolve, Math.pow(2, attempt) * 1000)
        );
      }
    }
  }
 
  // All retries exhausted — send to dead letter queue
  await deadLetterQueue.push({
    originalEvent: event,
    error: lastError?.message,
    failedAt: new Date().toISOString(),
    attempts: maxRetries,
  });
}