Learn CQRS: separating commands (writes) from queries (reads) using MediatR, with independent read/write models for scalability.
Command Query Responsibility Segregation (CQRS) separates the write model (commands that change state) from the read model (queries that return data). Implemented with MediatR in .NET, CQRS enables optimized, independently scalable read and write paths.
CQRS splits the system into two separate models:
1// CQRS - Command Query Responsibility Segregation
2/*
3 * Traditional CRUD:
4 * ┌────────────────────────────────────────────┐
5 * │ Single Model │
6 * │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │
7 * │ │Create│ │ Read │ │Update│ │Delete│ │
8 * │ └──────┘ └──────┘ └──────┘ └──────┘ │
9 * │ ▼ │
10 * │ Same Database │
11 * └────────────────────────────────────────────┘
12 *
13 * CQRS Pattern:
14 * ┌────────────────┐ ┌────────────────────┐
15 * │ Commands │ │ Queries │
16 * │ (Write Model) │ │ (Read Model) │
17 * ├────────────────┤ ├────────────────────┤
18 * │ • Create │ │ • GetById │
19 * │ • Update │ │ • GetList │
20 * │ • Delete │ │ • Search │
21 * │ • Business Ops │ │ • Reports │
22 * └───────┬────────┘ └────────┬───────────┘
23 * │ │
24 * ▼ ▼
25 * Write DB Read DB (optional)
26 */
27
28// Benefits of CQRS:
29// 1. Optimized read/write models
30// 2. Scalability - scale reads/writes independently
31// 3. Security - different authorization for reads/writes
32// 4. Performance - read model optimized for queries
33// 5. Flexibility - different storage for reads/writes
34
35// Command - Intent to change state
36public interface ICommand<TResponse> : IRequest<TResponse> { }
37
38// Query - Request for data
39public interface IQuery<TResponse> : IRequest<TResponse> { }
40
41// Command Handler - Processes commands
42public interface ICommandHandler<in TCommand, TResponse>
43 : IRequestHandler<TCommand, TResponse>
44 where TCommand : ICommand<TResponse> { }
45
46// Query Handler - Processes queries
47public interface IQueryHandler<in TQuery, TResponse>
48 : IRequestHandler<TQuery, TResponse>
49 where TQuery : IQuery<TResponse> { }
50
51// Example: Order bounded context with CQRS
52namespace Orders.Application.Commands;
53
54public record CreateOrderCommand(
55 Guid CustomerId,
56 List<OrderItemDto> Items,
57 AddressDto ShippingAddress
58) : ICommand<Result<OrderDto>>;
59
60namespace Orders.Application.Queries;
61
62public record GetOrderByIdQuery(Guid OrderId) : IQuery<OrderDto?>;
63
64public record GetCustomerOrdersQuery(
65 Guid CustomerId,
66 int Page = 1,
67 int PageSize = 10
68) : IQuery<PaginatedList<OrderSummaryDto>>;Commands represent the intent to change state and are processed by handlers:
1// Commands represent intent to change system state
2public record CreateOrderCommand(
3 Guid CustomerId,
4 List<OrderItemDto> Items,
5 AddressDto ShippingAddress
6) : ICommand<Result<OrderDto>>;
7
8public record UpdateOrderCommand(
9 Guid OrderId,
10 List<OrderItemDto> Items,
11 AddressDto? ShippingAddress
12) : ICommand<Result<OrderDto>>;
13
14public record CancelOrderCommand(
15 Guid OrderId,
16 string Reason
17) : ICommand<Result>;
18
19public record SubmitOrderCommand(Guid OrderId) : ICommand<Result>;
20
21// Command Handler with full business logic
22public class CreateOrderCommandHandler
23 : ICommandHandler<CreateOrderCommand, Result<OrderDto>>
24{
25 private readonly IOrderRepository _orderRepository;
26 private readonly ICustomerRepository _customerRepository;
27 private readonly IProductRepository _productRepository;
28 private readonly IUnitOfWork _unitOfWork;
29 private readonly IMapper _mapper;
30 private readonly ILogger<CreateOrderCommandHandler> _logger;
31
32 public CreateOrderCommandHandler(
33 IOrderRepository orderRepository,
34 ICustomerRepository customerRepository,
35 IProductRepository productRepository,
36 IUnitOfWork unitOfWork,
37 IMapper mapper,
38 ILogger<CreateOrderCommandHandler> logger)
39 {
40 _orderRepository = orderRepository;
41 _customerRepository = customerRepository;
42 _productRepository = productRepository;
43 _unitOfWork = unitOfWork;
44 _mapper = mapper;
45 _logger = logger;
46 }
47
48 public async Task<Result<OrderDto>> Handle(
49 CreateOrderCommand request,
50 CancellationToken ct)
51 {
52 _logger.LogInformation(
53 "Creating order for customer {CustomerId}",
54 request.CustomerId);
55
56 // Validate customer exists
57 var customer = await _customerRepository.GetByIdAsync(
58 request.CustomerId, ct);
59 if (customer == null)
60 return Result<OrderDto>.Failure("Customer not found");
61
62 // Create shipping address value object
63 var address = Address.Create(
64 request.ShippingAddress.Street,
65 request.ShippingAddress.City,
66 request.ShippingAddress.State,
67 request.ShippingAddress.ZipCode,
68 request.ShippingAddress.Country);
69
70 // Create order aggregate
71 var order = Order.Create(request.CustomerId, address);
72
73 // Add items with validation
74 foreach (var item in request.Items)
75 {
76 var product = await _productRepository.GetByIdAsync(item.ProductId, ct);
77 if (product == null)
78 return Result<OrderDto>.Failure($"Product {item.ProductId} not found");
79
80 if (!product.HasStock(item.Quantity))
81 return Result<OrderDto>.Failure(
82 $"Insufficient stock for {product.Name}");
83
84 order.AddItem(product.Id, product.Name, product.Price, item.Quantity);
85 }
86
87 // Persist
88 await _orderRepository.AddAsync(order, ct);
89 await _unitOfWork.SaveChangesAsync(ct);
90
91 _logger.LogInformation("Order {OrderId} created", order.Id);
92
93 return Result<OrderDto>.Success(_mapper.Map<OrderDto>(order));
94 }
95}
96
97// Command Validation with FluentValidation
98public class CreateOrderCommandValidator
99 : AbstractValidator<CreateOrderCommand>
100{
101 public CreateOrderCommandValidator()
102 {
103 RuleFor(x => x.CustomerId)
104 .NotEmpty()
105 .WithMessage("Customer ID is required");
106
107 RuleFor(x => x.Items)
108 .NotEmpty()
109 .WithMessage("Order must have at least one item");
110
111 RuleForEach(x => x.Items)
112 .ChildRules(item =>
113 {
114 item.RuleFor(i => i.ProductId).NotEmpty();
115 item.RuleFor(i => i.Quantity).GreaterThan(0);
116 });
117
118 RuleFor(x => x.ShippingAddress).NotNull();
119 RuleFor(x => x.ShippingAddress.Street).NotEmpty();
120 RuleFor(x => x.ShippingAddress.City).NotEmpty();
121 RuleFor(x => x.ShippingAddress.ZipCode).NotEmpty();
122 }
123}Queries return data from optimized read models without modifying state:
1// Queries request data without changing state
2public record GetOrderByIdQuery(Guid OrderId) : IQuery<OrderDto?>;
3
4public record GetCustomerOrdersQuery(
5 Guid CustomerId,
6 int Page = 1,
7 int PageSize = 10,
8 OrderStatus? StatusFilter = null,
9 DateTime? FromDate = null,
10 DateTime? ToDate = null
11) : IQuery<PaginatedList<OrderSummaryDto>>;
12
13public record SearchOrdersQuery(
14 string? SearchTerm,
15 OrderStatus? Status,
16 decimal? MinAmount,
17 decimal? MaxAmount,
18 string? SortBy,
19 bool SortDescending = false,
20 int Page = 1,
21 int PageSize = 20
22) : IQuery<PaginatedList<OrderSummaryDto>>;
23
24// Query Handler - Optimized for reading
25public class GetOrderByIdQueryHandler
26 : IQueryHandler<GetOrderByIdQuery, OrderDto?>
27{
28 private readonly IOrderReadRepository _readRepository;
29 private readonly ICacheService _cache;
30 private readonly IMapper _mapper;
31
32 public GetOrderByIdQueryHandler(
33 IOrderReadRepository readRepository,
34 ICacheService cache,
35 IMapper mapper)
36 {
37 _readRepository = readRepository;
38 _cache = cache;
39 _mapper = mapper;
40 }
41
42 public async Task<OrderDto?> Handle(
43 GetOrderByIdQuery request,
44 CancellationToken ct)
45 {
46 // Try cache first
47 var cacheKey = $"order:{request.OrderId}";
48 var cached = await _cache.GetAsync<OrderDto>(cacheKey, ct);
49 if (cached != null)
50 return cached;
51
52 // Query from read-optimized repository
53 var order = await _readRepository.GetByIdWithDetailsAsync(
54 request.OrderId, ct);
55
56 if (order == null)
57 return null;
58
59 var dto = _mapper.Map<OrderDto>(order);
60
61 // Cache for future requests
62 await _cache.SetAsync(cacheKey, dto, TimeSpan.FromMinutes(5), ct);
63
64 return dto;
65 }
66}
67
68// Search Query Handler with complex filtering
69public class SearchOrdersQueryHandler
70 : IQueryHandler<SearchOrdersQuery, PaginatedList<OrderSummaryDto>>
71{
72 private readonly AppDbContext _context;
73
74 public async Task<PaginatedList<OrderSummaryDto>> Handle(
75 SearchOrdersQuery request,
76 CancellationToken ct)
77 {
78 var query = _context.Orders
79 .AsNoTracking()
80 .Include(o => o.Customer)
81 .AsQueryable();
82
83 // Apply filters
84 if (!string.IsNullOrWhiteSpace(request.SearchTerm))
85 {
86 query = query.Where(o =>
87 o.Customer.Name.Contains(request.SearchTerm) ||
88 o.Id.ToString().Contains(request.SearchTerm));
89 }
90
91 if (request.Status.HasValue)
92 query = query.Where(o => o.Status == request.Status.Value);
93
94 if (request.MinAmount.HasValue)
95 query = query.Where(o => o.TotalAmount.Amount >= request.MinAmount);
96
97 if (request.MaxAmount.HasValue)
98 query = query.Where(o => o.TotalAmount.Amount <= request.MaxAmount);
99
100 // Apply sorting
101 query = request.SortBy?.ToLower() switch
102 {
103 "amount" => request.SortDescending
104 ? query.OrderByDescending(o => o.TotalAmount.Amount)
105 : query.OrderBy(o => o.TotalAmount.Amount),
106 "customer" => request.SortDescending
107 ? query.OrderByDescending(o => o.Customer.Name)
108 : query.OrderBy(o => o.Customer.Name),
109 _ => request.SortDescending
110 ? query.OrderByDescending(o => o.CreatedAt)
111 : query.OrderBy(o => o.CreatedAt)
112 };
113
114 // Project directly to DTO for efficiency
115 return await query
116 .Select(o => new OrderSummaryDto
117 {
118 Id = o.Id,
119 CustomerName = o.Customer.Name,
120 Status = o.Status,
121 TotalAmount = o.TotalAmount.Amount,
122 ItemCount = o.Items.Count,
123 CreatedAt = o.CreatedAt
124 })
125 .ToPaginatedListAsync(request.Page, request.PageSize, ct);
126 }
127}MediatR pipeline behaviors apply logging, validation, and caching to all commands/queries:
1// MediatR Configuration
2public static class DependencyInjection
3{
4 public static IServiceCollection AddApplication(
5 this IServiceCollection services)
6 {
7 // Register MediatR
8 services.AddMediatR(cfg =>
9 {
10 cfg.RegisterServicesFromAssembly(
11 typeof(DependencyInjection).Assembly);
12
13 // Pipeline behaviors (executed in order)
14 cfg.AddBehavior(typeof(IPipelineBehavior<,>),
15 typeof(LoggingBehavior<,>));
16 cfg.AddBehavior(typeof(IPipelineBehavior<,>),
17 typeof(ValidationBehavior<,>));
18 cfg.AddBehavior(typeof(IPipelineBehavior<,>),
19 typeof(PerformanceBehavior<,>));
20 });
21
22 // Register FluentValidation validators
23 services.AddValidatorsFromAssembly(
24 typeof(DependencyInjection).Assembly);
25
26 return services;
27 }
28}
29
30// Logging Behavior - Logs all requests
31public class LoggingBehavior<TRequest, TResponse>
32 : IPipelineBehavior<TRequest, TResponse>
33 where TRequest : IRequest<TResponse>
34{
35 private readonly ILogger<LoggingBehavior<TRequest, TResponse>> _logger;
36
37 public async Task<TResponse> Handle(
38 TRequest request,
39 RequestHandlerDelegate<TResponse> next,
40 CancellationToken cancellationToken)
41 {
42 var requestName = typeof(TRequest).Name;
43
44 _logger.LogInformation(
45 "Handling {RequestName}: {@Request}",
46 requestName, request);
47
48 var response = await next();
49
50 _logger.LogInformation(
51 "Handled {RequestName}: {@Response}",
52 requestName, response);
53
54 return response;
55 }
56}
57
58// Validation Behavior - Validates before handling
59public class ValidationBehavior<TRequest, TResponse>
60 : IPipelineBehavior<TRequest, TResponse>
61 where TRequest : IRequest<TResponse>
62{
63 private readonly IEnumerable<IValidator<TRequest>> _validators;
64
65 public async Task<TResponse> Handle(
66 TRequest request,
67 RequestHandlerDelegate<TResponse> next,
68 CancellationToken cancellationToken)
69 {
70 if (!_validators.Any())
71 return await next();
72
73 var context = new ValidationContext<TRequest>(request);
74
75 var validationResults = await Task.WhenAll(
76 _validators.Select(v =>
77 v.ValidateAsync(context, cancellationToken)));
78
79 var failures = validationResults
80 .SelectMany(r => r.Errors)
81 .Where(f => f != null)
82 .ToList();
83
84 if (failures.Count != 0)
85 throw new ValidationException(failures);
86
87 return await next();
88 }
89}
90
91// Performance Behavior - Monitors slow requests
92public class PerformanceBehavior<TRequest, TResponse>
93 : IPipelineBehavior<TRequest, TResponse>
94 where TRequest : IRequest<TResponse>
95{
96 private readonly Stopwatch _timer = new();
97 private readonly ILogger<PerformanceBehavior<TRequest, TResponse>> _logger;
98
99 public async Task<TResponse> Handle(
100 TRequest request,
101 RequestHandlerDelegate<TResponse> next,
102 CancellationToken cancellationToken)
103 {
104 _timer.Start();
105 var response = await next();
106 _timer.Stop();
107
108 var elapsed = _timer.ElapsedMilliseconds;
109
110 if (elapsed > 500) // Warn if > 500ms
111 {
112 var requestName = typeof(TRequest).Name;
113 _logger.LogWarning(
114 "Long Running Request: {RequestName} ({ElapsedMs}ms) {@Request}",
115 requestName, elapsed, request);
116 }
117
118 return response;
119 }
120}
121
122// Usage in Controller
123[ApiController]
124[Route("api/orders")]
125public class OrdersController : ControllerBase
126{
127 private readonly IMediator _mediator;
128
129 [HttpPost]
130 public async Task<ActionResult<OrderDto>> CreateOrder(
131 [FromBody] CreateOrderCommand command)
132 {
133 var result = await _mediator.Send(command);
134 return result.IsSuccess
135 ? CreatedAtAction(nameof(GetOrder), new { id = result.Value.Id }, result.Value)
136 : BadRequest(result.Error);
137 }
138
139 [HttpGet("{id}")]
140 public async Task<ActionResult<OrderDto>> GetOrder(Guid id)
141 {
142 var result = await _mediator.Send(new GetOrderByIdQuery(id));
143 return result != null ? Ok(result) : NotFound();
144 }
145}Your progress is saved in this browser, on this device only.