This tutorial builds a saga orchestrator in NestJS from scratch — covering module structure, persistence with TypeORM, step execution with dependency injection, compensation handling, and a REST API for triggering and monitoring sagas. NestJS's module system and DI container make it a natural fit for structuring saga step implementations as injectable services.
Project Setup
bash
1nest new saga-demo
2cd saga-demo
3npm install @nestjs/typeorm typeorm pg uuid class-validator class-transformer
4
Step 1: Database Entities
Define the saga state and step records as TypeORM entities.
typescript
1// src/saga/entities/saga-state.entity.ts
2import {
3 Entity, PrimaryColumn, Column, CreateDateColumn, UpdateDateColumn, VersionColumn,
4} from 'typeorm';
5
6('saga_state')
7export class SagaStateEntity {
8 ('uuid')
9 id: string;
10
11 ({ name: 'saga_type' })
12 sagaType: string;
13
14 ({ default: 'running' })
15 status: string;
16
17 ({ name: 'current_step', default: 0 })
18 currentStep: number;
19
20 ({ type: 'jsonb', default: '{}' })
21 context: Record<string, unknown>;
22
23 ({ name: 'completed_steps', type: 'text', array: true, default: '{}' })
24 completedSteps: string[];
25
26 ({ nullable: true })
27 error: string;
28
29 ()
30 version: number;
31
32 ({ name: 'created_at' })
33 createdAt: Date;
34
35 ({ name: 'updated_at' })
36 updatedAt: Date;
37}
38
Step 2: Saga Step Interface
typescript
1// src/saga/interfaces/saga-step.interface.ts
2export interface ISagaStep<TContext = any> {
3 readonly name: string;
4 readonly timeout?: number;
5 readonly retryPolicy?: RetryPolicy;
6 execute(context: TContext): Promise<void>;
7 compensate(context: TContext): Promise<void>;
8}
9
10export interface RetryPolicy {
11 maxAttempts: number;
12 initialBackoffMs: number;
13 multiplier: number;
14 maxBackoffMs: number;
15}
16
Step 3: Saga Store Service
typescript
1// src/saga/saga-store.service.ts
2import { Injectable } from '@nestjs/common';
3import { InjectRepository } from '@nestjs/typeorm';
4import { Repository } from 'typeorm';
5import { SagaStateEntity } from './entities/saga-state.entity';
6
7()
8export class SagaStoreService {
9 constructor(
10 (SagaStateEntity)
11 private readonly repo: Repository<SagaStateEntity>,
12 ) {}
13
14 async save(state: SagaStateEntity): Promise<SagaStateEntity> {
15 return this.repo.save(state);
16 }
17
18 async findById(id: string): Promise<SagaStateEntity | null> {
19 return this.repo.findOneBy({ id });
20 }
21
22 async updateWithLock(state: SagaStateEntity): Promise<void> {
23 const result = await this.repo.update(
24 { id: state.id, version: state.version },
25 {
26 status: state.status,
27 currentStep: state.currentStep,
28 context: state.context,
29 completedSteps: state.completedSteps,
30 error: state.error,
31 },
32 );
33
34 if (result.affected === 0) {
35 throw new Error(`Optimistic lock conflict for saga ${state.id}`);
36 }
37 }
38
39 async findByStatus(status: string, limit = 100): Promise<SagaStateEntity[]> {
40 return this.repo.find({
41 where: { status },
42 order: { createdAt: 'DESC' },
43 take: limit,
44 });
45 }
46
47 async countByStatus(): Promise<Record<string, number>> {
48 const results = await this.repo
49 .createQueryBuilder('s')
50 .select('s.status', 'status')
51 .addSelect('COUNT(*)', 'count')
52 .groupBy('s.status')
53 .getRawMany();
54
55 return Object.fromEntries(results.map(r => [r.status, parseInt(r.count)]));
56 }
57}
58
Step 4: Saga Orchestrator Service
typescript
1// src/saga/saga-orchestrator.service.ts
2import { Injectable, Logger } from '@nestjs/common';
3import { v4 as uuid } from 'uuid';
4import { SagaStoreService } from './saga-store.service';
5import { SagaStateEntity } from './entities/saga-state.entity';
6import { ISagaStep, RetryPolicy } from './interfaces/saga-step.interface';
7
8()
9export class SagaOrchestratorService {
10 private readonly logger = new Logger(SagaOrchestratorService.name);
11
12 constructor(private readonly store: SagaStoreService) {}
13
14 async execute<TContext extends Record<string, unknown>>(
15 sagaType: string,
16 steps: ISagaStep<TContext>[],
17 context: TContext,
18 ): Promise<string> {
19 const state = new SagaStateEntity();
20 state.id = uuid();
21 state.sagaType = sagaType;
22 state.status = 'running';
23 state.currentStep = 0;
24 state.context = context as Record<string, unknown>;
25 state.completedSteps = [];
26
27 await this.store.save(state);
28 this.logger.log(`Saga started: ${state.id} (${sagaType})`);
29
30 await this.executeSteps(state, steps, context);
31 return state.id;
32 }
33
34 private async executeSteps<TContext extends Record<string, unknown>>(
35 state: SagaStateEntity,
36 steps: ISagaStep<TContext>[],
37 context: TContext,
38 ): Promise<void> {
39 for (let i = state.currentStep; i < steps.length; i++) {
40 const step = steps[i];
41
42 this.logger.log(`Step ${step.name}: executing (saga: ${state.id})`);
43
44 try {
45 await this.executeWithRetry(step, context);
46
47 state.currentStep = i + 1;
48 state.completedSteps = [...state.completedSteps, step.name];
49 state.context = { ...context } as Record<string, unknown>;
50 await this.store.updateWithLock(state);
51
52 } catch (error) {
53 this.logger.error(`Step ${step.name} failed: ${(error as Error).message}`);
54
55 state.status = 'compensating';
56 state.error = (error as Error).message;
57 await this.store.updateWithLock(state);
58
59 await this.compensate(state, steps, context);
60 throw error;
61 }
62 }
63
64 state.status = 'completed';
65 await this.store.updateWithLock(state);
66 this.logger.log(`Saga completed: ${state.id}`);
67 }
68
69 private async executeWithRetry<TContext>(
70 step: ISagaStep<TContext>,
71 context: TContext,
72 ): Promise<void> {
73 const policy: RetryPolicy = step.retryPolicy ?? {
74 maxAttempts: 1,
75 initialBackoffMs: 100,
76 multiplier: 2,
77 maxBackoffMs: 5000,
78 };
79
80 let lastError: Error;
81
82 for (let attempt = 0; attempt < policy.maxAttempts; attempt++) {
83 if (attempt > 0) {
84 const backoff = Math.min(
85 policy.initialBackoffMs * Math.pow(policy.multiplier, attempt - 1),
86 policy.maxBackoffMs,
87 );
88 await new Promise(r => setTimeout(r, backoff));
89 }
90
91 try {
92 if (step.timeout) {
93 await Promise.race([
94 step.execute(context),
95 new Promise<never>((_, reject) =>
96 setTimeout(() => reject(new Error(`Timeout after ${step.timeout}ms`)), step.timeout),
97 ),
98 ]);
99 } else {
100 await step.execute(context);
101 }
102 return;
103 } catch (error) {
104 lastError = error as Error;
105 this.logger.warn(`Step ${step.name} attempt ${attempt + 1}/${policy.maxAttempts} failed`);
106 }
107 }
108
109 throw lastError!;
110 }
111
112 private async compensate<TContext extends Record<string, unknown>>(
113 state: SagaStateEntity,
114 steps: ISagaStep<TContext>[],
115 context: TContext,
116 ): Promise<void> {
117 const toCompensate = [...state.completedSteps].reverse();
118
119 for (const stepName of toCompensate) {
120 const step = steps.find(s => s.name === stepName);
121 if (!step) continue;
122
123 this.logger.log(`Compensating ${stepName} (saga: ${state.id})`);
124
125 try {
126 await step.compensate(context);
127 } catch (error) {
128 this.logger.error(`Compensation failed for ${stepName}: ${(error as Error).message}`);
129 }
130 }
131
132 state.status = 'failed';
133 await this.store.updateWithLock(state);
134 }
135}
136
Step 5: Order Saga Implementation
Create concrete step implementations as NestJS injectable services.
typescript
1// src/order/steps/reserve-inventory.step.ts
2import { Injectable, Logger } from '@nestjs/common';
3import { ISagaStep, RetryPolicy } from '../../saga/interfaces/saga-step.interface';
4import { InventoryService } from '../services/inventory.service';
5import { OrderContext } from '../interfaces/order-context.interface';
6
7()
8export class ReserveInventoryStep implements ISagaStep<OrderContext> {
9 readonly name = 'reserve_inventory';
10 readonly timeout = 5000;
11 readonly retryPolicy: RetryPolicy = {
12 maxAttempts: 3,
13 initialBackoffMs: 200,
14 multiplier: 2,
15 maxBackoffMs: 3000,
16 };
17
18 private readonly logger = new Logger(ReserveInventoryStep.name);
19
20 constructor(private readonly inventoryService: InventoryService) {}
21
22 async execute(ctx: OrderContext): Promise<void> {
23 const result = await this.inventoryService.reserve({
24 orderId: ctx.orderId,
25 items: ctx.items,
26 idempotencyKey: `${ctx.orderId}-reserve`,
27 });
28 ctx.reservationId = result.reservationId;
29 this.logger.log(`Inventory reserved: ${result.reservationId}`);
30 }
31
32 async compensate(ctx: OrderContext): Promise<void> {
33 if (!ctx.reservationId) return;
34 await this.inventoryService.release({
35 reservationId: ctx.reservationId,
36 idempotencyKey: `${ctx.orderId}-release`,
37 });
38 this.logger.log(`Inventory released: ${ctx.reservationId}`);
39 }
40}
41
typescript
1// src/order/steps/charge-payment.step.ts
2import { Injectable, Logger } from '@nestjs/common';
3import { ISagaStep } from '../../saga/interfaces/saga-step.interface';
4import { PaymentService } from '../services/payment.service';
5import { OrderContext } from '../interfaces/order-context.interface';
6
7()
8export class ChargePaymentStep implements ISagaStep<OrderContext> {
9 readonly name = 'charge_payment';
10 readonly timeout = 30000;
11 readonly retryPolicy = {
12 maxAttempts: 2,
13 initialBackoffMs: 1000,
14 multiplier: 2,
15 maxBackoffMs: 5000,
16 };
17
18 private readonly logger = new Logger(ChargePaymentStep.name);
19
20 constructor(private readonly paymentService: PaymentService) {}
21
22 async execute(ctx: OrderContext): Promise<void> {
23 const result = await this.paymentService.charge({
24 orderId: ctx.orderId,
25 amount: ctx.totalAmount,
26 paymentMethodId: ctx.paymentMethodId,
27 idempotencyKey: `${ctx.orderId}-charge`,
28 });
29 ctx.chargeId = result.chargeId;
30 this.logger.log(`Payment charged: ${result.chargeId}`);
31 }
32
33 async compensate(ctx: OrderContext): Promise<void> {
34 if (!ctx.chargeId) return;
35 await this.paymentService.refund({
36 chargeId: ctx.chargeId,
37 idempotencyKey: `${ctx.orderId}-refund`,
38 });
39 this.logger.log(`Payment refunded: ${ctx.chargeId}`);
40 }
41}
42
typescript
1// src/order/steps/create-shipment.step.ts
2import { Injectable, Logger } from '@nestjs/common';
3import { ISagaStep } from '../../saga/interfaces/saga-step.interface';
4import { ShippingService } from '../services/shipping.service';
5import { OrderContext } from '../interfaces/order-context.interface';
6
7()
8export class CreateShipmentStep implements ISagaStep<OrderContext> {
9 readonly name = 'create_shipment';
10 readonly timeout = 10000;
11
12 private readonly logger = new Logger(CreateShipmentStep.name);
13
14 constructor(private readonly shippingService: ShippingService) {}
15
16 async execute(ctx: OrderContext): Promise<void> {
17 const result = await this.shippingService.create({
18 orderId: ctx.orderId,
19 address: ctx.shippingAddress,
20 items: ctx.items,
21 idempotencyKey: `${ctx.orderId}-ship`,
22 });
23 ctx.shipmentId = result.shipmentId;
24 ctx.trackingNumber = result.trackingNumber;
25 this.logger.log(`Shipment created: ${result.shipmentId}`);
26 }
27
28 async compensate(ctx: OrderContext): Promise<void> {
29 if (!ctx.shipmentId) return;
30 await this.shippingService.cancel(ctx.shipmentId);
31 this.logger.log(`Shipment cancelled: ${ctx.shipmentId}`);
32 }
33}
34
Need a second opinion on your system design architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallStep 6: Order Saga Service
Wire the steps together into a saga definition.
typescript
1// src/order/order-saga.service.ts
2import { Injectable } from '@nestjs/common';
3import { SagaOrchestratorService } from '../saga/saga-orchestrator.service';
4import { ReserveInventoryStep } from './steps/reserve-inventory.step';
5import { ChargePaymentStep } from './steps/charge-payment.step';
6import { CreateShipmentStep } from './steps/create-shipment.step';
7import { OrderContext } from './interfaces/order-context.interface';
8
9()
10export class OrderSagaService {
11 private readonly steps;
12
13 constructor(
14 private readonly orchestrator: SagaOrchestratorService,
15 private readonly reserveInventory: ReserveInventoryStep,
16 private readonly chargePayment: ChargePaymentStep,
17 private readonly createShipment: CreateShipmentStep,
18 ) {
19 this.steps = [this.reserveInventory, this.chargePayment, this.createShipment];
20 }
21
22 async fulfillOrder(context: OrderContext): Promise<string> {
23 return this.orchestrator.execute('order-fulfillment', this.steps, context);
24 }
25}
26
Step 7: Controller
typescript
1// src/order/order.controller.ts
2import { Controller, Post, Body, Get, Param, HttpCode, HttpStatus } from '@nestjs/common';
3import { IsString, IsNumber, IsArray, ValidateNested, Min } from 'class-validator';
4import { Type } from 'class-transformer';
5import { OrderSagaService } from './order-saga.service';
6import { SagaStoreService } from '../saga/saga-store.service';
7import { OrderContext } from './interfaces/order-context.interface';
8import { v4 as uuid } from 'uuid';
9
10class OrderItemDto {
11 () productId: string;
12 () (1) quantity: number;
13 () (0) price: number;
14}
15
16class CreateOrderDto {
17 () customerId: string;
18 () ({ each: true }) (() => OrderItemDto) items: OrderItemDto[];
19 () paymentMethodId: string;
20 () shippingStreet: string;
21 () shippingCity: string;
22 () shippingState: string;
23 () shippingZip: string;
24 () shippingCountry: string;
25}
26
27('orders')
28export class OrderController {
29 constructor(
30 private readonly orderSaga: OrderSagaService,
31 private readonly sagaStore: SagaStoreService,
32 ) {}
33
34 ()
35 (HttpStatus.CREATED)
36 async createOrder(() dto: CreateOrderDto) {
37 const context: OrderContext = {
38 orderId: uuid(),
39 customerId: dto.customerId,
40 items: dto.items,
41 totalAmount: dto.items.reduce((sum, item) => sum + item.price * item.quantity, 0),
42 paymentMethodId: dto.paymentMethodId,
43 shippingAddress: {
44 street: dto.shippingStreet,
45 city: dto.shippingCity,
46 state: dto.shippingState,
47 zipCode: dto.shippingZip,
48 country: dto.shippingCountry,
49 },
50 };
51
52 try {
53 const sagaId = await this.orderSaga.fulfillOrder(context);
54
55 return {
56 sagaId,
57 orderId: context.orderId,
58 trackingNumber: context.trackingNumber ?? null,
59 status: 'completed',
60 };
61 } catch (error) {
62 return {
63 orderId: context.orderId,
64 status: 'failed',
65 error: (error as Error).message,
66 };
67 }
68 }
69
70 ('saga/:id')
71 async getSagaStatus(('id') id: string) {
72 const state = await this.sagaStore.findById(id);
73 if (!state) return { error: 'Saga not found' };
74 return state;
75 }
76}
77
Step 8: Module Configuration
typescript
1// src/saga/saga.module.ts
2import { Module } from '@nestjs/common';
3import { TypeOrmModule } from '@nestjs/typeorm';
4import { SagaStateEntity } from './entities/saga-state.entity';
5import { SagaStoreService } from './saga-store.service';
6import { SagaOrchestratorService } from './saga-orchestrator.service';
7
8({
9 imports: [TypeOrmModule.forFeature([SagaStateEntity])],
10 providers: [SagaStoreService, SagaOrchestratorService],
11 exports: [SagaStoreService, SagaOrchestratorService],
12})
13export class SagaModule {}
14
typescript
1// src/order/order.module.ts
2import { Module } from '@nestjs/common';
3import { SagaModule } from '../saga/saga.module';
4import { OrderController } from './order.controller';
5import { OrderSagaService } from './order-saga.service';
6import { ReserveInventoryStep } from './steps/reserve-inventory.step';
7import { ChargePaymentStep } from './steps/charge-payment.step';
8import { CreateShipmentStep } from './steps/create-shipment.step';
9import { InventoryService } from './services/inventory.service';
10import { PaymentService } from './services/payment.service';
11import { ShippingService } from './services/shipping.service';
12
13({
14 imports: [SagaModule],
15 controllers: [OrderController],
16 providers: [
17 OrderSagaService,
18 ReserveInventoryStep,
19 ChargePaymentStep,
20 CreateShipmentStep,
21 InventoryService,
22 PaymentService,
23 ShippingService,
24 ],
25})
26export class OrderModule {}
27
Step 9: Monitoring Endpoint
typescript
1// src/saga/saga-monitoring.controller.ts
2import { Controller, Get, Param, Query } from '@nestjs/common';
3import { SagaStoreService } from './saga-store.service';
4
5('admin/sagas')
6export class SagaMonitoringController {
7 constructor(private readonly store: SagaStoreService) {}
8
9 ('stats')
10 async getStats() {
11 return this.store.countByStatus();
12 }
13
14 ('failed')
15 async getFailedSagas(('limit') limit = 20) {
16 return this.store.findByStatus('failed', limit);
17 }
18
19 (':id')
20 async getSaga(('id') id: string) {
21 return this.store.findById(id);
22 }
23}
24
Step 10: Testing
typescript
1// src/order/order-saga.service.spec.ts
2import { Test, TestingModule } from '@nestjs/testing';
3import { SagaOrchestratorService } from '../saga/saga-orchestrator.service';
4import { SagaStoreService } from '../saga/saga-store.service';
5import { OrderSagaService } from './order-saga.service';
6import { ReserveInventoryStep } from './steps/reserve-inventory.step';
7import { ChargePaymentStep } from './steps/charge-payment.step';
8import { CreateShipmentStep } from './steps/create-shipment.step';
9import { OrderContext } from './interfaces/order-context.interface';
10
11describe('OrderSagaService', () => {
12 let service: OrderSagaService;
13 let reserveInventory: ReserveInventoryStep;
14 let chargePayment: ChargePaymentStep;
15
16 const mockStore = {
17 save: jest.fn().mockResolvedValue(undefined),
18 findById: jest.fn(),
19 updateWithLock: jest.fn().mockResolvedValue(undefined),
20 findByStatus: jest.fn(),
21 countByStatus: jest.fn(),
22 };
23
24 beforeEach(async () => {
25 const module: TestingModule = await Test.createTestingModule({
26 providers: [
27 SagaOrchestratorService,
28 OrderSagaService,
29 { provide: SagaStoreService, useValue: mockStore },
30 {
31 provide: ReserveInventoryStep,
32 useValue: {
33 name: 'reserve_inventory',
34 execute: jest.fn().mockImplementation(async (ctx: OrderContext) => {
35 ctx.reservationId = 'res-123';
36 }),
37 compensate: jest.fn().mockResolvedValue(undefined),
38 },
39 },
40 {
41 provide: ChargePaymentStep,
42 useValue: {
43 name: 'charge_payment',
44 execute: jest.fn().mockImplementation(async (ctx: OrderContext) => {
45 ctx.chargeId = 'ch-456';
46 }),
47 compensate: jest.fn().mockResolvedValue(undefined),
48 },
49 },
50 {
51 provide: CreateShipmentStep,
52 useValue: {
53 name: 'create_shipment',
54 execute: jest.fn().mockImplementation(async (ctx: OrderContext) => {
55 ctx.shipmentId = 'ship-789';
56 ctx.trackingNumber = 'TRK001';
57 }),
58 compensate: jest.fn().mockResolvedValue(undefined),
59 },
60 },
61 ],
62 }).compile();
63
64 service = module.get(OrderSagaService);
65 reserveInventory = module.get(ReserveInventoryStep);
66 chargePayment = module.get(ChargePaymentStep);
67 });
68
69 it('executes all steps successfully', async () => {
70 const context: OrderContext = {
71 orderId: 'order-1',
72 customerId: 'cust-1',
73 items: [{ productId: 'prod-1', quantity: 1, price: 29.99 }],
74 totalAmount: 29.99,
75 paymentMethodId: 'pm-1',
76 shippingAddress: { street: '123 Main', city: 'NYC', state: 'NY', zipCode: '10001', country: 'US' },
77 };
78
79 const sagaId = await service.fulfillOrder(context);
80
81 expect(sagaId).toBeDefined();
82 expect(context.reservationId).toBe('res-123');
83 expect(context.chargeId).toBe('ch-456');
84 expect(context.trackingNumber).toBe('TRK001');
85 });
86
87 it('compensates on payment failure', async () => {
88 (chargePayment.execute as jest.Mock).mockRejectedValue(new Error('Payment declined'));
89
90 const context: OrderContext = {
91 orderId: 'order-2',
92 customerId: 'cust-1',
93 items: [{ productId: 'prod-1', quantity: 1, price: 29.99 }],
94 totalAmount: 29.99,
95 paymentMethodId: 'pm-bad',
96 shippingAddress: { street: '123 Main', city: 'NYC', state: 'NY', zipCode: '10001', country: 'US' },
97 };
98
99 await expect(service.fulfillOrder(context)).rejects.toThrow('Payment declined');
100 expect(reserveInventory.compensate).toHaveBeenCalled();
101 });
102});
103
Conclusion
NestJS's dependency injection system maps naturally to saga step implementations. Each step is an injectable service with its own dependencies (inventory service, payment service, shipping service), and the DI container wires everything together at module composition time. This gives you testable steps with mock injection, clean separation of saga orchestration from business logic, and the ability to share services across multiple saga definitions.
The TypeORM entity with @VersionColumn provides optimistic locking for saga state updates without custom SQL. The orchestrator saves state after every step completion, enabling crash recovery by resuming from the last persisted step. Combined with idempotent step implementations, this makes the saga resilient to orchestrator restarts.
For production deployments, add a health check endpoint that queries for stale sagas (running status with updatedAt older than the maximum expected saga duration) and integrate with NestJS's built-in logger for structured log output. The monitoring controller provides the operational visibility needed to diagnose stuck or failed sagas without database access.