Event-Driven Architecture for Node.js Applications

Event-driven architecture decouples components by communicating through events. Publishers emit events without knowing who listens. Subscribers react without knowing who published.

Node.js EventEmitter#

import { EventEmitter } from 'events';

interface UserEvents {
  'user:created': (user: User) => void;
  'user:updated': (user: User, changes: Partial<User>) => void;
  'user:deleted': (userId: string) => void;
}

class TypedEventEmitter extends EventEmitter {
  emit<K extends keyof UserEvents>(
    event: K,
    ...args: Parameters<UserEvents[K]>
  ): boolean {
    return super.emit(event, ...args);
  }

  on<K extends keyof UserEvents>(event: K, listener: UserEvents[K]): this {
    return super.on(event, listener);
  }
}

const userEvents = new TypedEventEmitter();

// Subscribe
userEvents.on('user:created', (user) => {
  console.log(`User created: ${user.id}`);
});

userEvents.on('user:created', async (user) => {
  await sendWelcomeEmail(user.email);
});

// Publish
userEvents.emit('user:created', { id: '1', email: 'test@example.com', name: 'Test' });

Application Event Bus#

type EventHandler<T = any> = (payload: T) => void | Promise<void>;

class EventBus {
  private handlers = new Map<string, EventHandler[]>();

  subscribe<T>(event: string, handler: EventHandler<T>): () => void {
    if (!this.handlers.has(event)) {
      this.handlers.set(event, []);
    }
    this.handlers.get(event)!.push(handler);

    // Return unsubscribe function
    return () => {
      const handlers = this.handlers.get(event);
      if (handlers) {
        const index = handlers.indexOf(handler);
        if (index > -1) handlers.splice(index, 1);
      }
    };
  }

  async publish<T>(event: string, payload: T): Promise<void> {
    const handlers = this.handlers.get(event) || [];

    await Promise.all(
      handlers.map(async (handler) => {
        try {
          await handler(payload);
        } catch (error) {
          console.error(`Error in event handler for ${event}:`, error);
        }
      })
    );
  }
}

// Singleton
export const eventBus = new EventBus();

// Usage
eventBus.subscribe('order:placed', async (order: Order) => {
  await updateInventory(order.items);
});

eventBus.subscribe('order:placed', async (order: Order) => {
  await sendOrderConfirmation(order);
});

await eventBus.publish('order:placed', order);

Domain Events#

interface DomainEvent {
  type: string;
  payload: any;
  occurredAt: Date;
  aggregateId: string;
}

// Event definitions
const OrderEvents = {
  PLACED: 'order.placed',
  CONFIRMED: 'order.confirmed',
  SHIPPED: 'order.shipped',
  DELIVERED: 'order.delivered',
  CANCELLED: 'order.cancelled',
} as const;

interface OrderPlacedEvent extends DomainEvent {
  type: typeof OrderEvents.PLACED;
  payload: {
    orderId: string;
    customerId: string;
    items: OrderItem[];
    total: number;
  };
}

// Event creator
function createEvent<T extends DomainEvent>(
  type: T['type'],
  aggregateId: string,
  payload: T['payload']
): T {
  return {
    type,
    aggregateId,
    payload,
    occurredAt: new Date(),
  } as T;
}

// Order aggregate with events
class Order {
  private events: DomainEvent[] = [];

  constructor(
    public readonly id: string,
    public customerId: string,
    public items: OrderItem[],
    public status: string = 'pending'
  ) {}

  place(): void {
    this.status = 'placed';
    this.events.push(
      createEvent<OrderPlacedEvent>(OrderEvents.PLACED, this.id, {
        orderId: this.id,
        customerId: this.customerId,
        items: this.items,
        total: this.calculateTotal(),
      })
    );
  }

  getUncommittedEvents(): DomainEvent[] {
    return [...this.events];
  }

  clearEvents(): void {
    this.events = [];
  }

  private calculateTotal(): number {
    return this.items.reduce((sum, item) => sum + item.price * item.quantity, 0);
  }
}

Message Queue Integration#

import { Queue, Worker } from 'bullmq';
import Redis from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);

// Event queue
const eventQueue = new Queue('events', { connection: redis });

// Publish to queue
async function publishEvent(event: DomainEvent): Promise<void> {
  await eventQueue.add(event.type, event, {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,
    },
  });
}

// Event handlers registry
const eventHandlers = new Map<string, ((event: DomainEvent) => Promise<void>)[]>();

function registerHandler(
  eventType: string,
  handler: (event: DomainEvent) => Promise<void>
): void {
  if (!eventHandlers.has(eventType)) {
    eventHandlers.set(eventType, []);
  }
  eventHandlers.get(eventType)!.push(handler);
}

// Worker to process events
const eventWorker = new Worker(
  'events',
  async (job) => {
    const event = job.data as DomainEvent;
    const handlers = eventHandlers.get(event.type) || [];

    for (const handler of handlers) {
      await handler(event);
    }
  },
  { connection: redis }
);

// Register handlers
registerHandler(OrderEvents.PLACED, async (event: OrderPlacedEvent) => {
  await inventoryService.reserve(event.payload.items);
});

registerHandler(OrderEvents.PLACED, async (event: OrderPlacedEvent) => {
  await notificationService.sendOrderConfirmation(event.payload.customerId);
});

Event Sourcing#

interface Event {
  id: string;
  type: string;
  aggregateId: string;
  payload: any;
  version: number;
  timestamp: Date;
}

class EventStore {
  constructor(private db: Database) {}

  async save(aggregateId: string, events: Event[], expectedVersion: number): Promise<void> {
    const currentVersion = await this.getVersion(aggregateId);

    if (currentVersion !== expectedVersion) {
      throw new ConcurrencyError(
        `Expected version ${expectedVersion}, but found ${currentVersion}`
      );
    }

    await this.db.transaction(async (tx) => {
      for (let i = 0; i < events.length; i++) {
        await tx.events.create({
          data: {
            ...events[i],
            version: expectedVersion + i + 1,
          },
        });
      }
    });
  }

  async getEvents(aggregateId: string, fromVersion = 0): Promise<Event[]> {
    return this.db.events.findMany({
      where: {
        aggregateId,
        version: { gt: fromVersion },
      },
      orderBy: { version: 'asc' },
    });
  }

  async getVersion(aggregateId: string): Promise<number> {
    const lastEvent = await this.db.events.findFirst({
      where: { aggregateId },
      orderBy: { version: 'desc' },
    });
    return lastEvent?.version ?? 0;
  }
}

// Aggregate rebuilding
class OrderAggregate {
  private state: OrderState = { status: 'draft', items: [] };
  private version = 0;

  apply(event: Event): void {
    switch (event.type) {
      case 'OrderCreated':
        this.state = { ...this.state, id: event.payload.id, customerId: event.payload.customerId };
        break;
      case 'ItemAdded':
        this.state.items.push(event.payload.item);
        break;
      case 'OrderPlaced':
        this.state.status = 'placed';
        break;
    }
    this.version = event.version;
  }

  static async load(eventStore: EventStore, id: string): Promise<OrderAggregate> {
    const events = await eventStore.getEvents(id);
    const aggregate = new OrderAggregate();

    for (const event of events) {
      aggregate.apply(event);
    }

    return aggregate;
  }
}

CQRS Pattern#

// Commands
interface Command {
  type: string;
}

interface PlaceOrderCommand extends Command {
  type: 'PlaceOrder';
  customerId: string;
  items: OrderItem[];
}

// Command handler
class OrderCommandHandler {
  constructor(
    private eventStore: EventStore,
    private eventBus: EventBus
  ) {}

  async handle(command: PlaceOrderCommand): Promise<string> {
    const orderId = crypto.randomUUID();

    const events: Event[] = [
      {
        id: crypto.randomUUID(),
        type: 'OrderPlaced',
        aggregateId: orderId,
        payload: {
          customerId: command.customerId,
          items: command.items,
        },
        version: 1,
        timestamp: new Date(),
      },
    ];

    await this.eventStore.save(orderId, events, 0);

    // Publish for read model updates
    for (const event of events) {
      await this.eventBus.publish(event.type, event);
    }

    return orderId;
  }
}

// Read model (projection)
class OrderReadModel {
  constructor(private db: Database) {}

  async handle(event: Event): Promise<void> {
    switch (event.type) {
      case 'OrderPlaced':
        await this.db.orderViews.create({
          data: {
            id: event.aggregateId,
            customerId: event.payload.customerId,
            itemCount: event.payload.items.length,
            status: 'placed',
            createdAt: event.timestamp,
          },
        });
        break;

      case 'OrderShipped':
        await this.db.orderViews.update({
          where: { id: event.aggregateId },
          data: { status: 'shipped', shippedAt: event.timestamp },
        });
        break;
    }
  }
}

// Query
async function getCustomerOrders(customerId: string): Promise<OrderView[]> {
  return db.orderViews.findMany({
    where: { customerId },
    orderBy: { createdAt: 'desc' },
  });
}

Saga Pattern#

// Orchestration-based saga
class OrderSaga {
  constructor(
    private orderService: OrderService,
    private paymentService: PaymentService,
    private inventoryService: InventoryService,
    private eventBus: EventBus
  ) {}

  async execute(orderId: string): Promise<void> {
    const order = await this.orderService.get(orderId);

    try {
      // Step 1: Reserve inventory
      await this.inventoryService.reserve(order.items);

      // Step 2: Process payment
      await this.paymentService.charge(order.customerId, order.total);

      // Step 3: Confirm order
      await this.orderService.confirm(orderId);

      await this.eventBus.publish('OrderCompleted', { orderId });
    } catch (error) {
      // Compensating transactions
      await this.compensate(orderId, error);
    }
  }

  private async compensate(orderId: string, error: Error): Promise<void> {
    // Reverse inventory reservation
    await this.inventoryService.release(orderId);

    // Refund if payment was made
    await this.paymentService.refund(orderId);

    // Mark order as failed
    await this.orderService.fail(orderId, error.message);

    await this.eventBus.publish('OrderFailed', { orderId, reason: error.message });
  }
}

Best Practices#

Event Design:
✓ Use past tense (OrderPlaced, not PlaceOrder)
✓ Include all relevant data
✓ Make events immutable
✓ Version your events

Implementation:
✓ Handle failures gracefully
✓ Make handlers idempotent
✓ Use transactions for consistency
✓ Monitor event processing

Avoid:
✗ Circular event chains
✗ Business logic in handlers
✗ Blocking event processing
✗ Losing events on failure

Conclusion#

Event-driven architecture enables loose coupling and scalability. Start with in-process events, graduate to message queues as needed, and consider event sourcing for audit trails and complex domains.

Share this article