CQRS & Commands — MediatR-pipeline i Minimal API

CQRS — en mening

Skriv-sidan (commands som ändrar tillstånd) och läs-sidan (queries som returnerar data) hanteras av separata modeller. Greg Young, 2010.

Command-modell

En command är en immutable record med all data som behövs för beslutet — plus en MessageId för idempotens.

public interface ICommand<TResult> : IRequest<TResult>
{
    Guid MessageId { get; }   // idempotency key från klient
}

public sealed record DepositMoneyCommand(
    Guid MessageId,
    AccountId Account,
    decimal Amount,
    string Currency,
    string Reference) : ICommand<DepositMoneyResult>;

public sealed record DepositMoneyResult(int NewVersion, decimal NewBalance);

Handler

public sealed class DepositMoneyHandler
    : IRequestHandler<DepositMoneyCommand, DepositMoneyResult>
{
    private readonly IAccountRepository _repo;

    public DepositMoneyHandler(IAccountRepository repo) => _repo = repo;

    public async Task<DepositMoneyResult> Handle(
        DepositMoneyCommand cmd, CancellationToken ct)
    {
        var account = await _repo.LoadAsync(cmd.Account, ct)
            ?? throw new NotFoundException($"Account {cmd.Account} not found.");

        account.Deposit(new Money(cmd.Amount, cmd.Currency), cmd.Reference);

        await _repo.SaveAsync(account, ct);

        return new DepositMoneyResult(account.Version, account.Balance.Amount);
    }
}

Pipeline behaviors

MediatRs IPipelineBehavior<TReq, TRes> låter oss lägga cross-cutting concerns runt alla handlers utan att klottra ner dem.

1. Loggning

public sealed class LoggingBehavior<TReq, TRes> : IPipelineBehavior<TReq, TRes>
    where TReq : notnull
{
    private readonly ILogger<LoggingBehavior<TReq, TRes>> _log;
    public LoggingBehavior(ILogger<LoggingBehavior<TReq, TRes>> log) => _log = log;

    public async Task<TRes> Handle(TReq req, RequestHandlerDelegate<TRes> next, CancellationToken ct)
    {
        var name = typeof(TReq).Name;
        var sw = Stopwatch.StartNew();
        try
        {
            var result = await next();
            _log.LogInformation("{Command} succeeded in {Ms}ms", name, sw.ElapsedMilliseconds);
            return result;
        }
        catch (Exception ex)
        {
            _log.LogError(ex, "{Command} failed after {Ms}ms", name, sw.ElapsedMilliseconds);
            throw;
        }
    }
}

2. Validering med FluentValidation

public sealed class DepositMoneyValidator : AbstractValidator<DepositMoneyCommand>
{
    public DepositMoneyValidator()
    {
        RuleFor(x => x.Amount).GreaterThan(0);
        RuleFor(x => x.Currency).Length(3);
        RuleFor(x => x.Reference).NotEmpty().MaximumLength(140);
    }
}

public sealed class ValidationBehavior<TReq, TRes> : IPipelineBehavior<TReq, TRes>
    where TReq : notnull
{
    private readonly IEnumerable<IValidator<TReq>> _validators;
    public ValidationBehavior(IEnumerable<IValidator<TReq>> validators) => _validators = validators;

    public async Task<TRes> Handle(TReq req, RequestHandlerDelegate<TRes> next, CancellationToken ct)
    {
        var ctx = new ValidationContext<TReq>(req);
        var failures = (await Task.WhenAll(_validators.Select(v => v.ValidateAsync(ctx, ct))))
            .SelectMany(r => r.Errors).Where(f => f != null).ToList();

        if (failures.Count != 0)
            throw new ValidationException(failures);

        return await next();
    }
}

3. Idempotens

public sealed class IdempotencyBehavior<TReq, TRes> : IPipelineBehavior<TReq, TRes>
    where TReq : ICommand<TRes>
{
    private readonly IIdempotencyStore _store;
    public IdempotencyBehavior(IIdempotencyStore store) => _store = store;

    public async Task<TRes> Handle(TReq req, RequestHandlerDelegate<TRes> next, CancellationToken ct)
    {
        if (await _store.TryGetResultAsync<TRes>(req.MessageId, ct) is { } cached)
            return cached;

        var result = await next();
        await _store.SaveResultAsync(req.MessageId, result, ct);
        return result;
    }
}

IIdempotencyStore kan vara en MSSQL-tabell (MessageId PK, ResultJson, CreatedAt). Klienten genererar MessageId per användarhandling och skickar med samma id vid retry — då får man tillbaka det cachade resultatet i stället för dubbel insättning.

Minimal API-endpoints

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMediatR(cfg => cfg.RegisterServicesFromAssemblyContaining<DepositMoneyCommand>());
builder.Services.AddValidatorsFromAssemblyContaining<DepositMoneyValidator>();

builder.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(LoggingBehavior<,>));
builder.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(ValidationBehavior<,>));
builder.Services.AddTransient(typeof(IPipelineBehavior<,>), typeof(IdempotencyBehavior<,>));

builder.Services.AddScoped<IAccountRepository, AccountRepository>();

var app = builder.Build();

app.MapPost("/accounts/{id:guid}/deposit", async (
    Guid id,
    DepositRequest body,
    IMediator mediator,
    [FromHeader(Name = "Idempotency-Key")] Guid? idempotencyKey) =>
{
    var cmd = new DepositMoneyCommand(
        MessageId: idempotencyKey ?? Guid.NewGuid(),
        Account: new AccountId(id),
        Amount: body.Amount,
        Currency: body.Currency,
        Reference: body.Reference);

    var result = await mediator.Send(cmd);
    return Results.Ok(result);
});

app.MapGet("/accounts/{id:guid}", async (Guid id, IMediator mediator) =>
{
    var query = new GetAccountSummaryQuery(new AccountId(id));
    var dto = await mediator.Send(query);
    return dto is null ? Results.NotFound() : Results.Ok(dto);
});

app.Run();

public sealed record DepositRequest(decimal Amount, string Currency, string Reference);

Queries — läs-sidan

Queries går också genom MediatR men träffar aldrig aggregatet — bara read models.

public sealed record GetAccountSummaryQuery(AccountId Account) : IRequest<AccountSummaryDto?>;

public sealed record AccountSummaryDto(Guid Id, string Currency, decimal Balance, DateTimeOffset UpdatedAt);

public sealed class GetAccountSummaryHandler
    : IRequestHandler<GetAccountSummaryQuery, AccountSummaryDto?>
{
    private readonly IDbConnection _db;
    public GetAccountSummaryHandler(IDbConnection db) => _db = db;

    public Task<AccountSummaryDto?> Handle(GetAccountSummaryQuery q, CancellationToken ct) =>
        _db.QuerySingleOrDefaultAsync<AccountSummaryDto>(
            "SELECT Id, Currency, Balance, UpdatedAt FROM AccountSummary WHERE Id = @Id",
            new { Id = q.Account.Value });
}

Sammanfattning

• Commands ändrar tillstånd och returnerar ett litet resultat. Queries läser och returnerar DTO:er.
• MediatR + pipeline behaviors ger logg/validering/idempotens runt alla handlers.
• Handlers är tunna — affärsregler bor i aggregatet.
• Idempotency-nycklar från klienten gör API:t säkert vid retry.

← Föregående: AggregateRoot Nästa: Event Sourcing →