Chapter 04

CQRS Pattern
với MediatR

Tách Command & Query — kiến trúc rõ ràng, dễ scale

CQRS là gì?

Command Query Responsibility Segregation — nguyên tắc tách biệt hoàn toàn thao tác đọc (Query) và thao tác ghi (Command). Mỗi loại có model, handler, và thậm chí có thể dùng database riêng.

❌

CRUD truyền thống

Một model cho cả đọc và ghi
  • Cùng một ProductService xử lý mọi thứ
  • DTO phình to, chứa cả field đọc lẫn ghi
  • Tối ưu đọc ảnh hưởng đến ghi và ngược lại
  • Validation phức tạp khi một model dùng chung
  • Khó scale riêng read/write
✔

CQRS

Tách riêng model đọc và ghi
  • Command handler xử lý ghi, Query handler xử lý đọc
  • DTO tối ưu cho từng use case cụ thể
  • Tối ưu read/write độc lập
  • Validation rõ ràng per command
  • Có thể scale read replica riêng

CQRS Flow với MediatR

Controller
HTTP Request
→
Command
ghi dữ liệu
→
Handler
write DB
Controller
HTTP Request
→
Query
đọc dữ liệu
→
Handler
read DB

Marker Interfaces

Tạo các interface đánh dấu để phân biệt Command và Query — giúp apply behaviors có chọn lọc:

ICommand

: IRequest

Command void (không trả kết quả)

ICommand<T>

: IRequest<T>

Command trả về kết quả (vd: ID mới tạo)

IQuery<T>

: IRequest<T>

Query luôn trả về kết quả

C#
// Marker interfaces
public interface ICommand : IRequest { }
public interface ICommand<TResponse> : IRequest<TResponse> { }
public interface IQuery<TResponse> : IRequest<TResponse> { }

// Marker handler interfaces (optional)
public interface ICommandHandler<TCommand>
    : IRequestHandler<TCommand>
    where TCommand : ICommand { }

public interface ICommandHandler<TCommand, TResponse>
    : IRequestHandler<TCommand, TResponse>
    where TCommand : ICommand<TResponse> { }

public interface IQueryHandler<TQuery, TResponse>
    : IRequestHandler<TQuery, TResponse>
    where TQuery : IQuery<TResponse> { }

Folder Structure

Hai cách tổ chức phổ biến. Khuyến nghị Vertical Slice cho dự án dùng MediatR:

Vertical Slice (Feature-based)

src/Application/
  Features/
    Products/
      Commands/
        CreateProduct.cs ← Command + Handler + Validator cùng file
        UpdateProduct.cs
        DeleteProduct.cs
      Queries/
        GetProductById.cs ← Query + Handler + DTO cùng file
        GetProductsList.cs
    Orders/
      Commands/
      Queries/
  Common/
    Behaviors/ ← Shared pipeline behaviors
    Interfaces/ ← ICommand, IQuery, etc.
Đặt Command + Handler + Validator trong cùng một file. Khi mở file, bạn thấy toàn bộ use case. Đây là ưu điểm lớn nhất của Vertical Slice — co-location.

Command Examples

CreateProduct — Command + Handler + Validator cùng file

C#
// Features/Products/Commands/CreateProduct.cs

public record CreateProductCommand(
    string Name,
    decimal Price,
    string Category
) : ICommand<int>;

public class CreateProductValidator
    : AbstractValidator<CreateProductCommand>
{
    public CreateProductValidator()
    {
        RuleFor(x => x.Name)
            .NotEmpty().MaximumLength(200);
        RuleFor(x => x.Price)
            .GreaterThan(0);
        RuleFor(x => x.Category)
            .NotEmpty();
    }
}

public class CreateProductHandler
    : ICommandHandler<CreateProductCommand, int>
{
    private readonly AppDbContext _db;

    public CreateProductHandler(AppDbContext db) => _db = db;

    public async Task<int> Handle(
        CreateProductCommand request,
        CancellationToken ct)
    {
        var product = new Product
        {
            Name = request.Name,
            Price = request.Price,
            Category = request.Category
        };

        _db.Products.Add(product);
        await _db.SaveChangesAsync(ct);

        return product.Id;
    }
}

UpdateProduct

C#
public record UpdateProductCommand(
    int Id,
    string Name,
    decimal Price
) : ICommand;

public class UpdateProductHandler
    : ICommandHandler<UpdateProductCommand>
{
    private readonly AppDbContext _db;

    public UpdateProductHandler(AppDbContext db) => _db = db;

    public async Task Handle(
        UpdateProductCommand request,
        CancellationToken ct)
    {
        var product = await _db.Products
            .FindAsync(new object[] { request.Id }, ct)
            ?? throw new NotFoundException("Product", request.Id);

        product.Name = request.Name;
        product.Price = request.Price;

        await _db.SaveChangesAsync(ct);
    }
}

Query Examples

GetProductById

C#
// Features/Products/Queries/GetProductById.cs

public record ProductDto(
    int Id, string Name,
    decimal Price, string Category);

public record GetProductByIdQuery(int Id)
    : IQuery<ProductDto>;

public class GetProductByIdHandler
    : IQueryHandler<GetProductByIdQuery, ProductDto>
{
    private readonly AppDbContext _db;

    public GetProductByIdHandler(AppDbContext db) => _db = db;

    public async Task<ProductDto> Handle(
        GetProductByIdQuery request,
        CancellationToken ct)
    {
        var p = await _db.Products
            .AsNoTracking()  // Query = read-only, không cần tracking
            .FirstOrDefaultAsync(x => x.Id == request.Id, ct)
            ?? throw new NotFoundException("Product", request.Id);

        return new ProductDto(p.Id, p.Name, p.Price, p.Category);
    }
}

GetProductsList (with Pagination)

C#
public record GetProductsListQuery(
    string? Category,
    int Page = 1,
    int PageSize = 20
) : IQuery<PagedResult<ProductDto>>;

public record PagedResult<T>(
    List<T> Items,
    int TotalCount,
    int Page,
    int PageSize)
{
    public int TotalPages => (int)Math.Ceiling(TotalCount / (double)PageSize);
    public bool HasNext => Page < TotalPages;
    public bool HasPrev => Page > 1;
}

public class GetProductsListHandler
    : IQueryHandler<GetProductsListQuery, PagedResult<ProductDto>>
{
    private readonly AppDbContext _db;

    public GetProductsListHandler(AppDbContext db) => _db = db;

    public async Task<PagedResult<ProductDto>> Handle(
        GetProductsListQuery request,
        CancellationToken ct)
    {
        var query = _db.Products.AsNoTracking();

        if (!string.IsNullOrEmpty(request.Category))
            query = query.Where(p => p.Category == request.Category);

        var total = await query.CountAsync(ct);

        var items = await query
            .OrderBy(p => p.Name)
            .Skip((request.Page - 1) * request.PageSize)
            .Take(request.PageSize)
            .Select(p => new ProductDto(p.Id, p.Name, p.Price, p.Category))
            .ToListAsync(ct);

        return new PagedResult<ProductDto>(
            items, total, request.Page, request.PageSize);
    }
}

Type-Constrained Behaviors

Dùng generic constraints với marker interfaces để behaviors chỉ apply cho Command hoặc Query:

C#
// Behavior CHỈ apply cho Commands (ghi)
public class TransactionBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : ICommand<TResponse> // ← Constraint!
{
    private readonly AppDbContext _db;

    public TransactionBehavior(AppDbContext db) => _db = db;

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        await using var transaction =
            await _db.Database.BeginTransactionAsync(ct);

        try
        {
            var response = await next();
            await transaction.CommitAsync(ct);
            return response;
        }
        catch
        {
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

// Behavior CHỈ apply cho Queries (đọc)
public class CachingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IQuery<TResponse> // ← Chỉ queries!
{
    private readonly IDistributedCache _cache;

    public CachingBehavior(IDistributedCache cache)
        => _cache = cache;

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var cacheKey = $"{typeof(TRequest).Name}:{
            JsonSerializer.Serialize(request)}";

        var cached = await _cache.GetStringAsync(cacheKey, ct);
        if (cached != null)
            return JsonSerializer.Deserialize<TResponse>(cached)!;

        var response = await next();

        await _cache.SetStringAsync(cacheKey,
            JsonSerializer.Serialize(response),
            new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(5)
            }, ct);

        return response;
    }
}
TransactionBehavior chỉ bọc commands — queries không cần transaction vì chỉ đọc. CachingBehavior chỉ bọc queries — caching commands không có ý nghĩa. Đây chính là sức mạnh của type constraints + marker interfaces.
← Trước Pipeline Behaviors