Event-Driven Architecture Without the PhD — A Practical Guide

// events/emitter.ts — Simple in-process event system
type EventHandler<T = any> = (data: T) => Promise<void>;
 
class EventBus {
  private handlers: Map<string, EventHandler[]> = new Map();
 
  on(event: string, handler: EventHandler) {
    const existing = this.handlers.get(event) || [];
    this.handlers.set(event, [...existing, handler]);
  }
 
  async emit(event: string, data: any) {
    const handlers = this.handlers.get(event) || [];
    // Fire all handlers in parallel — don't await in the request path
    Promise.allSettled(
      handlers.map((handler) => 
        handler(data).catch((err) => {
          console.error(`Handler failed for ${event}:`, err);
          // Log to error tracking, don't crash the request
        })
      )
    );
  }
}
 
export const eventBus = new EventBus();

// events/handlers/order.ts
import { eventBus } from "../emitter";
 
eventBus.on("order.placed", async (order) => {
  await emailService.sendOrderConfirmation(order);
});
 
eventBus.on("order.placed", async (order) => {
  await warehouseService.createShipment(order);
});
 
eventBus.on("order.placed", async (order) => {
  await analyticsService.trackPurchase(order);
});

// Write the event to the database in the SAME transaction as the business data
async function placeOrder(orderData: CreateOrderInput) {
  return await db.transaction(async (tx) => {
    // 1. Create the order
    const order = await tx.orders.create(orderData);
 
    // 2. Write the event to an outbox table (same transaction!)
    await tx.outboxEvents.create({
      eventType: "order.placed",
      payload: JSON.stringify(order),
      createdAt: new Date(),
      processed: false,
    });
 
    return order;
  });
  // Transaction commits → both order AND event are saved atomically
}
 
// Separate worker polls the outbox and publishes events
async function processOutbox() {
  const events = await db.outboxEvents.findMany({
    where: { processed: false },
    orderBy: { createdAt: "asc" },
    take: 100,
  });
 
  for (const event of events) {
    await messageQueue.publish(event.eventType, JSON.parse(event.payload));
    await db.outboxEvents.update({
      where: { id: event.id },
      data: { processed: true, processedAt: new Date() },
    });
  }
}

// BAD: Not idempotent — sends duplicate emails
eventBus.on("order.placed", async (order) => {
  await emailService.send(order.email, "confirmation", order);
});
 
// GOOD: Idempotent — checks before processing
eventBus.on("order.placed", async (order) => {
  const alreadySent = await db.emailLog.exists({
    orderId: order.id,
    template: "confirmation",
  });
  if (alreadySent) return; // Skip duplicate
 
  await emailService.send(order.email, "confirmation", order);
  await db.emailLog.create({
    orderId: order.id,
    template: "confirmation",
    sentAt: new Date(),
  });
});