Scale to 10M Users: CQRS in NestJS for API Performance

#node #nestjs #cqrs

Scale to 10M Users: CQRS in NestJS for API Performance

In today's digital landscape, building applications that can handle millions of users requires thoughtful architecture decisions. Command Query Responsibility Segregation (CQRS) is one such pattern that helps maintain performance at scale. Let's explore!

What is CQRS and Why Use It?

CQRS splits your application into two models:

Command model: Handles create, update, and delete operations
Query model: Manages read operations

This separation addresses the reality that most applications have asymmetric read/write loads—typically with reads far outnumbering writes.

Setting Up CQRS in NestJS

First, install the required packages:

npm install @nestjs/cqrs uuid kafkajs redis mongodb pg

Directory Structure

src/
├── commands/
│   ├── handlers/
│   ├── impl/
├── queries/
│   ├── handlers/
│   ├── impl/
├── events/
├── models/
├── controllers/
└── app.module.ts

Basic Implementation

Let's start with our app.module.ts:

import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { MongooseModule } from '@nestjs/mongoose';
import { TypeOrmModule } from '@nestjs/typeorm';
import { CommandHandlers } from './commands/handlers';
import { QueryHandlers } from './queries/handlers';
import { EventHandlers } from './events/handlers';
import { Controllers } from './controllers';

@Module({
  imports: [
    CqrsModule,
    MongooseModule.forRoot('mongodb://localhost:27017/cqrs_read'),
    TypeOrmModule.forRoot({
      type: 'postgres',
      host: 'localhost',
      port: 5432,
      username: 'postgres',
      password: 'password',
      database: 'cqrs_write',
      entities: [__dirname + '/**/*.entity{.ts,.js}'],
      synchronize: true,
    }),
  ],
  controllers: [...Controllers],
  providers: [
    ...CommandHandlers,
    ...QueryHandlers,
    ...EventHandlers,
  ],
})
export class AppModule {}

Now, let's define a command:

// commands/impl/create-user.command.ts
export class CreateUserCommand {
  constructor(
    public readonly email: string,
    public readonly name: string,
  ) {}
}

And its handler:

// commands/handlers/create-user.handler.ts
import { CommandHandler, ICommandHandler, EventBus } from '@nestjs/cqrs';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { CreateUserCommand } from '../impl/create-user.command';
import { User } from '../../models/user.entity';
import { UserCreatedEvent } from '../../events/impl/user-created.event';

@CommandHandler(CreateUserCommand)
export class CreateUserHandler implements ICommandHandler<CreateUserCommand> {
  constructor(
    @InjectRepository(User)
    private userRepository: Repository<User>,
    private eventBus: EventBus,
  ) {}

  async execute(command: CreateUserCommand): Promise<User> {
    const { email, name } = command;

    const user = new User();
    user.email = email;
    user.name = name;

    const savedUser = await this.userRepository.save(user);

    // Publish event for read model synchronization
    this.eventBus.publish(new UserCreatedEvent(savedUser.id, email, name));

    return savedUser;
  }
}

Query Side with MongoDB

For read operations, we'll use MongoDB for its superior query performance:

// queries/impl/get-user.query.ts
export class GetUserQuery {
  constructor(public readonly id: string) {}
}

// queries/handlers/get-user.handler.ts
import { QueryHandler, IQueryHandler } from '@nestjs/cqrs';
import { InjectModel } from '@nestjs/mongoose';
import { Model } from 'mongoose';
import { GetUserQuery } from '../impl/get-user.query';
import { UserReadModel } from '../../models/user.read-model';

@QueryHandler(GetUserQuery)
export class GetUserHandler implements IQueryHandler<GetUserQuery> {
  constructor(
    @InjectModel(UserReadModel.name)
    private userModel: Model<UserReadModel>,
  ) {}

  async execute(query: GetUserQuery): Promise<UserReadModel> {
    return this.userModel.findOne({ userId: query.id }).exec();
  }
}

Synchronizing Read/Write Models with Kafka

To keep our read model updated, we'll use Kafka for event sourcing:

// events/handlers/user-created.handler.ts
import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { Inject } from '@nestjs/common';
import { ClientKafka } from '@nestjs/microservices';
import { UserCreatedEvent } from '../impl/user-created.event';

@EventsHandler(UserCreatedEvent)
export class UserCreatedHandler implements IEventHandler<UserCreatedEvent> {
  constructor(
    @Inject('KAFKA_SERVICE') private kafkaClient: ClientKafka
  ) {}

  handle(event: UserCreatedEvent) {
    this.kafkaClient.emit('user-created', {
      id: event.id,
      email: event.email,
      name: event.name,
      timestamp: new Date().toISOString()
    });
  }
}

And our consumer service:

// services/read-model-sync.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { InjectModel } from '@nestjs/mongoose';
import { Model } from 'mongoose';
import { Consumer, Kafka } from 'kafkajs';
import { UserReadModel } from '../models/user.read-model';

@Injectable()
export class ReadModelSyncService implements OnModuleInit {
  private consumer: Consumer;

  constructor(
    @InjectModel(UserReadModel.name)
    private userModel: Model<UserReadModel>,
  ) {
    const kafka = new Kafka({
      clientId: 'read-model-sync',
      brokers: ['localhost:9092'],
    });
    this.consumer = kafka.consumer({ groupId: 'read-model-sync-group' });
  }

  async onModuleInit() {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'user-created', fromBeginning: true });

    await this.consumer.run({
      eachMessage: async ({ topic, message }) => {
        const eventData = JSON.parse(message.value.toString());

        if (topic === 'user-created') {
          await this.userModel.updateOne(
            { userId: eventData.id },
            {
              userId: eventData.id,
              email: eventData.email,
              name: eventData.name,
              updatedAt: new Date()
            },
            { upsert: true }
          );
        }
      },
    });
  }
}