Session 13: CQRS & MediatR Patterns

Mục tiêu: Implement CQRS pattern với MediatR, pipeline behaviors, và Event Sourcing concepts.

1. CQRS - Command Query Responsibility Segregation

Traditional:
┌──────────┐        ┌──────────┐        ┌──────┐
│ API      │──────→ │ Service  │──────→ │  DB  │
│          │←────── │ (CRUD)   │←────── │      │
└──────────┘        └──────────┘        └──────┘
  Read & Write cùng 1 model, cùng 1 path

CQRS:
┌──────────┐   Command   ┌───────────────┐   ┌──────────┐
│          │────────────→ │ Command       │──→│ Write DB │
│   API    │              │ Handler       │   │ (SQL)    │
│          │   Query      ├───────────────┤   ├──────────┤
│          │────────────→ │ Query         │──→│ Read DB  │
└──────────┘              │ Handler       │   │ (Dapper) │
                          └───────────────┘   └──────────┘
  Tách Read & Write → optimize riêng mỗi bên

Tại Sao CQRS?

✅ Read thường >90% traffic → optimize riêng (denormalized, caching, read replica)
✅ Write cần validation, business rules → rich domain model
✅ Scale read/write independently
✅ Mỗi bên dùng model phù hợp nhất

⚠️ Không phải MỌI project đều cần CQRS
⚠️ Complexity tăng: 2 models, eventual consistency (nếu tách DB)
⚠️ Chỉ dùng khi read/write requirements KHÁC NHAU rõ ràng

2. MediatR Implementation

Setup

csharp
// Install: MediatR, MediatR.Extensions.Microsoft.DependencyInjection
builder.Services.AddMediatR(cfg =>
{
    cfg.RegisterServicesFromAssemblyContaining<ApplicationAssemblyMarker>();
    // Pipeline behaviors (thứ tự: ngoài → trong)
    cfg.AddBehavior(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));
    cfg.AddBehavior(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));
    cfg.AddBehavior(typeof(IPipelineBehavior<,>), typeof(TransactionBehavior<,>));
});

Commands (Write)

csharp
// Command: chỉ chứa data cần thiết, KHÔNG chứa logic
public record CreateOrderCommand(
    Guid CustomerId,
    string ShippingStreet,
    string ShippingCity,
    string ShippingPostalCode,
    List<OrderItemDto> Items
) : IRequest<Result<Guid>>;

// Handler: orchestrate use case
public class CreateOrderCommandHandler(
    IOrderRepository orderRepository,
    IProductRepository productRepository,
    IUnitOfWork unitOfWork,
    ILogger<CreateOrderCommandHandler> logger)
    : IRequestHandler<CreateOrderCommand, Result<Guid>>
{
    public async Task<Result<Guid>> Handle(
        CreateOrderCommand request, CancellationToken ct)
    {
        // 1. Validate business rules
        var address = Address.Create(
            request.ShippingStreet, request.ShippingCity, request.ShippingPostalCode);

        // 2. Create aggregate
        var order = Order.Create(request.CustomerId, address);

        // 3. Add items
        foreach (var itemDto in request.Items)
        {
            var product = await productRepository.GetByIdAsync(itemDto.ProductId, ct);
            if (product is null)
                return Result.Failure<Guid>($"Product {itemDto.ProductId} not found");

            order.AddItem(product, itemDto.Quantity);
        }

        // 4. Persist
        await orderRepository.AddAsync(order, ct);
        await unitOfWork.SaveChangesAsync(ct);

        logger.LogInformation("Order {OrderId} created", order.Id);
        return Result.Success(order.Id);
    }
}

Queries (Read)

csharp
// Query: what data to return
public record GetOrderByIdQuery(Guid OrderId) : IRequest<OrderDetailDto?>;

// Handler: optimized for reading
public class GetOrderByIdQueryHandler(IDbConnection connection)
    : IRequestHandler<GetOrderByIdQuery, OrderDetailDto?>
{
    public async Task<OrderDetailDto?> Handle(
        GetOrderByIdQuery request, CancellationToken ct)
    {
        // Dapper: direct SQL, no EF overhead cho reads
        const string sql = """
            SELECT o.Id, o.Status, o.TotalAmount as Total,
                   o.ShippingStreet, o.ShippingCity,
                   c.FullName as CustomerName,
                   c.Email as CustomerEmail
            FROM Orders o
            JOIN Customers c ON c.Id = o.CustomerId
            WHERE o.Id = @OrderId
            """;

        const string itemsSql = """
            SELECT oi.ProductId, oi.ProductName, oi.UnitPrice, oi.Quantity,
                   oi.UnitPrice * oi.Quantity as LineTotal
            FROM OrderItems oi
            WHERE oi.OrderId = @OrderId
            """;

        var order = await connection.QuerySingleOrDefaultAsync<OrderDetailDto>(
            sql, new { request.OrderId });

        if (order is not null)
        {
            order.Items = (await connection.QueryAsync<OrderItemDto>(
                itemsSql, new { request.OrderId })).ToList();
        }

        return order;
    }
}

3. Pipeline Behaviors

Validation Behavior

csharp
public class ValidationBehavior<TRequest, TResponse>(
    IEnumerable<IValidator<TRequest>> validators)
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        if (!validators.Any()) return await next();

        var context = new ValidationContext<TRequest>(request);
        var validationResults = await Task.WhenAll(
            validators.Select(v => v.ValidateAsync(context, ct)));

        var failures = validationResults
            .SelectMany(r => r.Errors)
            .Where(f => f is not null)
            .ToList();

        if (failures.Count > 0)
            throw new ValidationException(failures);

        return await next();
    }
}

// Validator (FluentValidation)
public class CreateOrderCommandValidator : AbstractValidator<CreateOrderCommand>
{
    public CreateOrderCommandValidator()
    {
        RuleFor(x => x.CustomerId).NotEmpty();
        RuleFor(x => x.ShippingStreet).NotEmpty().MaximumLength(200);
        RuleFor(x => x.ShippingCity).NotEmpty().MaximumLength(100);
        RuleFor(x => x.Items).NotEmpty().WithMessage("Order must have items");
        RuleForEach(x => x.Items).ChildRules(item =>
        {
            item.RuleFor(i => i.ProductId).NotEmpty();
            item.RuleFor(i => i.Quantity).GreaterThan(0);
        });
    }
}

Logging Behavior

csharp
public class LoggingBehavior<TRequest, TResponse>(
    ILogger<LoggingBehavior<TRequest, TResponse>> logger)
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var requestName = typeof(TRequest).Name;
        logger.LogInformation("Handling {RequestName}: {@Request}", requestName, request);

        var sw = Stopwatch.StartNew();
        var response = await next();
        sw.Stop();

        if (sw.ElapsedMilliseconds > 500)
        {
            logger.LogWarning("Long running: {RequestName} ({ElapsedMs}ms)",
                requestName, sw.ElapsedMilliseconds);
        }

        return response;
    }
}

Transaction Behavior

csharp
public class TransactionBehavior<TRequest, TResponse>(
    AppDbContext db, ILogger<TransactionBehavior<TRequest, TResponse>> logger)
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        // Chỉ wrap commands trong transaction
        if (request is not ICommand) return await next();

        await using var transaction = await db.Database.BeginTransactionAsync(ct);
        try
        {
            var response = await next();
            await transaction.CommitAsync(ct);
            logger.LogInformation("Transaction committed for {Request}", typeof(TRequest).Name);
            return response;
        }
        catch
        {
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

4. Notifications (Domain Event Handlers)

csharp
// MediatR Notification = Domain Event
public record OrderSubmittedEvent(Guid OrderId, Money Total) : INotification;

// Multiple handlers cho 1 event - loose coupling!
public class SendOrderConfirmationEmail(IEmailService email)
    : INotificationHandler<OrderSubmittedEvent>
{
    public async Task Handle(OrderSubmittedEvent evt, CancellationToken ct)
    {
        await email.SendOrderConfirmationAsync(evt.OrderId);
    }
}

public class UpdateInventory(IInventoryService inventory)
    : INotificationHandler<OrderSubmittedEvent>
{
    public async Task Handle(OrderSubmittedEvent evt, CancellationToken ct)
    {
        await inventory.ReserveItemsAsync(evt.OrderId);
    }
}

public class NotifyWarehouse(IWarehouseService warehouse)
    : INotificationHandler<OrderSubmittedEvent>
{
    public async Task Handle(OrderSubmittedEvent evt, CancellationToken ct)
    {
        await warehouse.EnqueueForPickingAsync(evt.OrderId);
    }
}

// Publish: await mediator.Publish(new OrderSubmittedEvent(orderId, total));
// → Tất cả handlers chạy tuần tự (default) hoặc parallel

5. Result Pattern - No Exceptions for Business Errors

csharp
public class Result
{
    public bool IsSuccess { get; }
    public string? Error { get; }

    protected Result(bool isSuccess, string? error)
    {
        IsSuccess = isSuccess;
        Error = error;
    }

    public static Result Success() => new(true, null);
    public static Result Failure(string error) => new(false, error);
    public static Result<T> Success<T>(T value) => new(value, true, null);
    public static Result<T> Failure<T>(string error) => new(default, false, error);
}

public class Result<T> : Result
{
    public T? Value { get; }

    internal Result(T? value, bool isSuccess, string? error)
        : base(isSuccess, error)
    {
        Value = value;
    }
}

// Usage in handler:
return order is null
    ? Result.Failure<OrderDto>("Order not found")
    : Result.Success(order.ToDto());

// Controller:
var result = await mediator.Send(command);
return result.IsSuccess
    ? Ok(result.Value)
    : BadRequest(new { error = result.Error });

6. Event Sourcing Basics

Event Sourcing: lưu EVENTS thay vì STATE hiện tại

Traditional:
  Orders Table: { Id: 1, Status: "Shipped", Total: 500 }
  → Chỉ biết state HIỆN TẠI, không biết lịch sử

Event Sourcing:
  Events Table:
  1. OrderCreated { orderId: 1, customerId: "C1", timestamp: ... }
  2. ItemAdded { productId: "P1", quantity: 2, price: 200 }
  3. ItemAdded { productId: "P2", quantity: 1, price: 100 }
  4. OrderSubmitted { timestamp: ... }
  5. PaymentReceived { amount: 500, method: "credit" }
  6. OrderShipped { trackingNumber: "TN123" }
  → Rebuild state bằng cách replay events
  → FULL audit trail, time-travel debugging

⚠️ Khi nào dùng Event Sourcing:
✅ Audit trail bắt buộc (finance, insurance, compliance)
✅ Temporal queries ("state ở thời điểm X là gì?")
✅ Complex business workflows (sagas, compensations)

❌ KHÔNG dùng khi:
❌ CRUD đơn giản (over-engineering)
❌ Team chưa quen CQRS
❌ Không cần audit trail

Tổng Kết Session 13

Bài tập: Implement CQRS cho 1 module: Command (EF Core write), Query (Dapper read). Thêm ValidationBehavior + LoggingBehavior. So sánh performance read qua EF vs Dapper.