Sagas in NestJS CQRS: a practical guide for developers

Sagas in NestJS CQRS: a practical guide for developers. Coordinate long-running processes with event streams.

Long-running business processes rarely complete in a single HTTP request. Your order system needs to reserve inventory, charge payment, update shipping, and send notifications—each step might fail, depend on external APIs, or take minutes to complete. Sagas in NestJS CQRS give you a pattern to orchestrate these workflows reliably, letting your application react to events and emit commands across multiple aggregates without tight coupling.

What is a saga and why it matters

A saga is a special event handler in the CQRS pattern that listens to domain events and emits one or more commands in response. Unlike a simple event handler that performs side effects, a saga coordinates a sequence of commands across bounded contexts. Think of it as a state machine that wakes up when events arrive, makes decisions, and tells other parts of your system what to do next.

In a real-world order processing system, an OrderCreatedEvent arrives. The saga listens, sees that an order exists, and emits a ReserveInventoryCommand. When inventory confirms success with an InventoryReservedEvent, the saga emits ChargePaymentCommand. If payment fails, the saga might emit a ReleaseInventoryCommand to roll back. Each step is decoupled; the order service doesn't know about payment logic, yet the saga orchestrates them together.

Sagas solve the distributed transaction problem without requiring a two-phase commit. Instead of hoping all systems succeed together, you build compensating commands that reverse earlier steps if later ones fail.

A saga listens to events and emits commands

The flow is straightforward: event in, command out. Let's build a concrete example with an order saga that handles the full lifecycle.

import { Injectable } from '@nestjs/common';
import { Saga, ICommand } from '@nestjs/cqrs';
import { map, ofType } from 'rxjs/operators';
import { Observable } from 'rxjs';

import { OrderCreatedEvent } from '../events/order-created.event';
import { InventoryReservedEvent } from '../events/inventory-reserved.event';
import { PaymentChargedEvent } from '../events/payment-charged.event';

import { ReserveInventoryCommand } from '../commands/reserve-inventory.command';
import { ChargePaymentCommand } from '../commands/charge-payment.command';
import { ConfirmOrderCommand } from '../commands/confirm-order.command';

@Injectable()
export class OrderSaga {
  @Saga()
  orderCreated$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(OrderCreatedEvent),
      map(
        (event: OrderCreatedEvent) =>
          new ReserveInventoryCommand(
            event.orderId,
            event.items,
            event.aggregateId,
          ),
      ),
    );
  };

  @Saga()
  inventoryReserved$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(InventoryReservedEvent),
      map(
        (event: InventoryReservedEvent) =>
          new ChargePaymentCommand(
            event.orderId,
            event.amount,
            event.customerId,
          ),
      ),
    );
  };

  @Saga()
  paymentCharged$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(PaymentChargedEvent),
      map(
        (event: PaymentChargedEvent) =>
          new ConfirmOrderCommand(event.orderId),
      ),
    );
  };
}

Each saga method subscribes to a specific event type using ofType, then transforms it into a command using map. The CQRS module automatically publishes the returned command to your command bus. Notice that we pass the orderId through each stage so commands land on the correct aggregate.

This declarative approach scales well. Adding a new step—say, sending a confirmation email after payment—is as simple as adding another @Saga() method that listens to PaymentChargedEvent.

Great for multi-step workflows

Sagas shine when processes span multiple aggregates or external systems. Consider a ride-hailing platform: a rider requests a trip, the system must find an available driver, the driver accepts, the rider confirms, and finally the trip starts. Each transition is an event; each event may trigger a command on a different aggregate.

@Injectable()
export class TripSaga {
  constructor(private eventBus: EventBus) {}

  @Saga()
  tripRequested$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(TripRequestedEvent),
      map(
        (event: TripRequestedEvent) =>
          new SearchDriversCommand(
            event.tripId,
            event.pickupLocation,
            event.dropoffLocation,
          ),
      ),
    );
  };

  @Saga()
  driverFound$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(DriverFoundEvent),
      map(
        (event: DriverFoundEvent) =>
          new NotifyDriverCommand(event.driverId, event.tripId),
      ),
    );
  };

  @Saga()
  driverAccepted$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(DriverAcceptedEvent),
      mergeMap((event: DriverAcceptedEvent) => {
        return [
          new NotifyRiderCommand(event.tripId, event.driverId),
          new StartETACalculationCommand(event.tripId),
        ];
      }),
    );
  };
}

Notice the last saga uses mergeMap instead of map. This allows emitting multiple commands from a single event. When the driver accepts, we notify the rider and start ETA calculation in parallel.

Sagas excel in workflows with conditional branching too. If payment fails, emit a ReleaseInventoryCommand. If the customer is a VIP, apply a discount command. Use filter, switchMap, and other RxJS operators to build sophisticated orchestration logic without cluttering your command or event handlers.

Keep sagas free of side effects

The single most common mistake is treating a saga like an event handler. Sagas should not call external APIs, send emails, or write to a database directly. A saga's job is to listen and command—nothing more.

Why? Because sagas are replayed during recovery. If your system crashes and replays events from the event store, a saga that sent an email on every replay would send duplicate emails. A saga that made an API call would create duplicate records. By keeping sagas pure—pure functions that transform events into commands—you gain idempotency and resilience for free.

Let's look at a bad saga:

// ❌ BAD: Saga contains side effects
@Injectable()
export class BadOrderSaga {
  constructor(private emailService: EmailService) {}

  @Saga()
  orderCreated$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(OrderCreatedEvent),
      tap((event) => {
        // WRONG: Sending email directly in saga
        this.emailService.sendOrderConfirmation(event.customerId);
      }),
      map(
        (event: OrderCreatedEvent) =>
          new ProcessOrderCommand(event.orderId),
      ),
    );
  };
}

If the system restarts and replays events, the email sends again. Instead, delegate side effects to command handlers or event handlers:

// ✅ GOOD: Saga is pure, command handler handles side effects
@Injectable()
export class GoodOrderSaga {
  @Saga()
  orderCreated$ = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(OrderCreatedEvent),
      map(
        (event: OrderCreatedEvent) =>
          new ProcessOrderCommand(event.orderId, event.customerId),
      ),
    );
  };
}

@Injectable()
export class ProcessOrderCommandHandler
  implements ICommandHandler<ProcessOrderCommand> {
  constructor(private emailService: EmailService) {}

  async execute(command: ProcessOrderCommand) {
    // Side effects belong here
    await this.emailService.sendOrderConfirmation(command.customerId);
    // Then emit an event
    this.eventBus.publish(new OrderProcessedEvent(command.orderId));
  }
}

The saga emits a command; the command handler executes business logic and side effects, then publishes an event. If a crash occurs and the saga replays, the command handler may execute again—but because the event store already contains the result, idempotent design ensures no duplicate side effects. This is the correct separation of concerns in CQRS.

Persist state for anything that must survive restarts

Simple sagas are stateless: listen to event A, emit command B, repeat. But what if your saga needs to track state across multiple events? For example, an order that waits for payment, then inventory, then shipping confirmation. If payment arrives before inventory is ready, you need to remember that payment succeeded when inventory finally arrives.

In these cases, use a process manager—a special aggregate that persists saga state. Let's build one:

// Process manager aggregate to track order fulfillment state
@Injectable()
export class OrderFulfillmentProcessManager {
  private state: Map<string, OrderFulfillmentState> = new Map();

  async handle(event: any): Promise<ICommand[]> {
    const orderId = event.orderId;
    let state = this.state.get(orderId) || {
      orderId,
      inventoryReserved: false,
      paymentCharged: false,
      shippingConfirmed: false,
    };

    if (event instanceof InventoryReservedEvent) {
      state.inventoryReserved = true;
    }
    if (event instanceof PaymentChargedEvent) {
      state.paymentCharged = true;
    }
    if (event instanceof ShippingConfirmedEvent) {
      state.shippingConfirmed = true;
    }

    this.state.set(orderId, state);

    const commands: ICommand[] = [];

    // All steps complete—emit final command
    if (
      state.inventoryReserved &&
      state.paymentCharged &&
      state.shippingConfirmed
    ) {
      commands.push(new CompleteOrderCommand(orderId));
      this.state.delete(orderId); // Clean up
    }

    return commands;
  }
}

interface OrderFulfillmentState {
  orderId: string;
  inventoryReserved: boolean;
  paymentCharged: boolean;
  shippingConfirmed: boolean;
}

This in-memory approach works for development but loses state on restart. For production, persist state in a database. Create a table with saga instance ID, state JSON, and timestamp. Load state on startup and update it as events arrive. Many teams use event sourcing for the process manager too—store each state transition as an event, replay on startup.

// Persist saga state to database
@Injectable()
export class OrderFulfillmentRepository {
  constructor(@InjectRepository(ProcessManagerState)
  private repository: Repository<ProcessManagerState>) {}

  async getState(processId: string): Promise<any> {
    const record = await this.repository.findOne({ where: { processId } });
    return record ? JSON.parse(record.stateJson) : null;
  }

  async saveState(processId: string, state: any): Promise<void> {
    await this.repository.upsert(
      {
        processId,
        stateJson: JSON.stringify(state),
        updatedAt: new Date(),
      },
      ['processId'],
    );
  }
}

Register the process manager in your module and inject it into your saga. On each event, load state, update it, check for completion conditions, and persist. This pattern is robust and survives infrastructure failures.

Common mistakes

• Adding side effects directly to sagas: Sagas must be pure. Delegate email, API calls, and database writes to command handlers or event handlers. This ensures idempotency and makes sagas replayable.
• Ignoring failure scenarios: What if a command fails? Emit a compensating command or a saga-specific failure event. Without explicit error handling, failed commands silently disappear and your workflow stalls.
• Forgetting correlation IDs: Pass a correlation ID or aggregate ID through every command and event so sagas can chain related operations. Without it, the saga doesn't know which order the inventory event belongs to.
• Using sagas for simple one-to-one event handlers: Not every event needs a saga. If event A always produces command B with no branching, use a simple event handler. Sagas are for orchestration, not everything.
• Losing state on restart: In-memory saga state disappears when your app crashes. For workflows that span hours or depend on multiple events arriving out of order, persist state in a database or use event sourcing.
• Blocking saga subscriptions: Sagas use RxJS observables. Never use synchronous blocking operations inside map, mergeMap, or switchMap. Use async operators like mergeMap with promises if you must wait for results.

Key takeaways

• Sagas listen to domain events and emit commands, orchestrating workflows across multiple aggregates without tight coupling.
• Use sagas for multi-step workflows, conditional branching, and compensating transactions—not for simple one-to-one event handling.
• Keep sagas pure: no side effects, no API calls, no database writes. Delegate those to command handlers and event handlers.
• Persist saga state in a database or event store for workflows that must survive restarts or depend on events arriving out of order.
• Always pass correlation IDs (aggregate IDs, process IDs) through events and commands so sagas chain the correct operations together.
• Handle failure explicitly: emit compensating commands or failure events when a command fails, so your workflow doesn't silently stall.
• Test sagas using observable testing libraries; verify that events produce the correct commands in the right order.