MediatR, CQRS, Pipeline Behaviors — Практическое Руководство

Назначение

Документ объясняет, как реализовать Commands, Queries и Pipeline Behaviors в Pin. Является практическим дополнением к MediatR Pipeline Behaviors с примерами кода, правилами именования и описанием каждого behavior.

---

Request Contracts

Commands И Queries

// Talent.Common.Application

// Базовые маркеры
public interface ICommand<TResult> : IRequest<Result<TResult>> { }
public interface IQuery<TResult> : IRequest<Result<TResult>> { }

// Command с idempotency key
public interface IIdempotentCommand<TResult> : ICommand<TResult>
{
    IdempotencyKey IdempotencyKey { get; }
}

// Command требующий audit record
public interface IAuditedRequest
{
    string AuditAction { get; }
    string? ReasonCode { get; }
}

// Command требующий local transaction
public interface ITransactionalCommand : IMarker { }

// Command публикующий integration events
public interface IProducesIntegrationEvents : IMarker { }

// Query с кешированием
public interface ICacheableQuery<TResult> : IQuery<TResult>
{
    string CacheKey { get; }
    TimeSpan Ttl { get; }
    string? InvalidationTag { get; }
    CachePrivacyLevel PrivacyLevel { get; }
}

// Маркер для authorization behavior
public interface IAuthorizedRequest
{
    string RequiredPermission { get; }
}

---

Порядок Pipeline Behaviors

Порядок behaviors задан в Pin.Kernel.AspNetCore и не меняется без ADR:

1.  CorrelationBehavior          — создаёт request scope и Activity для трасировки
2.  LoggingBehavior              — structured start/finish/failure logs
3.  MetricsBehavior              — duration, count, failures counters
4.  ValidationBehavior           — FluentValidation
5.  AuthorizationBehavior        — actor, permission, tenant/brand scope
6.  IdempotencyBehavior          — replay stored result для IIdempotentCommand
7.  QueryCacheBehavior (pre)     — read cache для ICacheableQuery
8.  TransactionBehavior          — local transaction для ITransactionalCommand
9.  → Handler ←
10. DomainEventDispatchBehavior  — dequeue domain events, map to integration events, publish to outbox
11. AuditBehavior                — write audit в той же transaction boundary
12. QueryCacheBehavior (post)    — сохранить result в cache после успешного read

---

Детали Каждого Behavior

1. CorrelationBehavior

public sealed class CorrelationBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : notnull
{
    public async Task<TResponse> Handle(
        TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        // Создаёт child Activity для request
        using var activity = PinActivitySource.StartActivity(
            name: typeof(TRequest).Name,
            kind: ActivityKind.Internal);

        activity?.SetTag("request.type", typeof(TRequest).Name);
        activity?.SetTag("actor.type", _identityProvider.TryGetCurrent()?.ActorType.ToString());

        return await next();
    }
}

2. LoggingBehavior

public sealed class LoggingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : notnull
{
    public async Task<TResponse> Handle(
        TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var requestName = typeof(TRequest).Name;
        var correlationId = _correlationContext.CorrelationId;

        _logger.LogInformation("Handling {RequestName} {CorrelationId}", requestName, correlationId);

        var sw = Stopwatch.StartNew();
        try
        {
            var response = await next();
            _logger.LogInformation(
                "Handled {RequestName} in {ElapsedMs}ms {CorrelationId}",
                requestName, sw.ElapsedMilliseconds, correlationId);
            return response;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex,
                "Error handling {RequestName} {CorrelationId}",
                requestName, correlationId);
            throw;
        }
    }
}

4. ValidationBehavior

public sealed class ValidationBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : notnull
    where TResponse : IResultBase
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

    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 results = await Task.WhenAll(_validators.Select(v => v.ValidateAsync(context, ct)));
        var failures = results.SelectMany(r => r.Errors).Where(f => f != null).ToList();

        if (failures.Count > 0)
        {
            // Возвращаем typed validation failure без открытия transaction
            return CreateValidationFailure<TResponse>(failures);
        }

        return await next();
    }
}

6. IdempotencyBehavior

public sealed class IdempotencyBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IIdempotentCommand<TResponse>
{
    public async Task<TResponse> Handle(
        TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var key = request.IdempotencyKey;
        var requestHash = _hasher.ComputeHash(request);

        // Проверяем существующий record
        var existing = await _idempotencyStore.Find(key, ct);
        if (existing is { Status: IdempotencyStatus.Completed })
        {
            _logger.LogInformation("Idempotency replay for key {Key}", key.Value);
            return DeserializeReplayResponse<TResponse>(existing.ResponseSnapshot);
        }

        if (existing is { Status: IdempotencyStatus.Started })
            return CreateConflictResult<TResponse>("operation-in-progress");

        if (existing is not null && existing.RequestHash != requestHash)
            return CreateConflictResult<TResponse>("idempotency-conflict");

        // Первый запрос — продолжаем
        return await next();
    }
}

8. TransactionBehavior

public sealed class TransactionBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : ITransactionalCommand
{
    public async Task<TResponse> Handle(
        TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        // Transaction начинается ПОСЛЕ validation и authorization
        await using var transaction = await _transactionBoundary.Begin(ct);
        try
        {
            var response = await next();
            if (response is IResultBase { IsSuccess: true })
                await transaction.Commit(ct);
            else
                await transaction.Rollback(ct);
            return response;
        }
        catch
        {
            await transaction.Rollback(ct);
            throw;
        }
    }
}

10. DomainEventDispatchBehavior

public sealed class DomainEventDispatchBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IProducesIntegrationEvents
{
    public async Task<TResponse> Handle(
        TRequest request, RequestHandlerDelegate<TResponse> next,
        CancellationToken ct)
    {
        var response = await next();

        if (response is IResultBase { IsSuccess: true })
        {
            // Собираем domain events из всех tracked aggregates
            var domainEvents = _dbContext.ChangeTracker.Entries<IAggregateRoot>()
                .SelectMany(e => e.Entity.DequeueDomainEvents())
                .ToList();

            // Маппим в integration events и публикуем через MassTransit Outbox
            foreach (var domainEvent in domainEvents)
            {
                var integrationEvent = _eventMapper.MapToIntegrationEvent(domainEvent);
                if (integrationEvent is not null)
                    await _publishEndpoint.Publish(integrationEvent, ct);
            }
        }

        return response;
    }
}

---

Полный Пример Command Handler

// Feature: Accounting/Withdrawals/ApproveWithdrawal

// Command
public sealed record ApproveWithdrawalCommand(
    WithdrawalRequestId WithdrawalId,
    string? ReasonCode,
    string? Comment
) : ICommand<ApproveWithdrawalResult>,
    IIdempotentCommand<ApproveWithdrawalResult>,
    IAuditedRequest,
    ITransactionalCommand,
    IProducesIntegrationEvents,
    IAuthorizedRequest
{
    public IdempotencyKey IdempotencyKey =>
        IdempotencyKey.For("accounting", "withdrawal.approve", WithdrawalId.ToString());

    public string AuditAction => "accounting.withdrawal.approve";
    public string RequiredPermission => "accounting.withdrawal.approve";
}

// Result
public sealed record ApproveWithdrawalResult(
    WithdrawalRequestId WithdrawalId,
    WithdrawalStatus NewStatus
);

// Validator
public sealed class ApproveWithdrawalCommandValidator
    : AbstractValidator<ApproveWithdrawalCommand>
{
    public ApproveWithdrawalCommandValidator()
    {
        RuleFor(x => x.WithdrawalId).NotEmpty();
        // ReasonCode + Comment validation if required by policy
    }
}

// Handler
public sealed class ApproveWithdrawalCommandHandler
    : ICommandHandler<ApproveWithdrawalCommand, ApproveWithdrawalResult>
{
    private readonly IWithdrawalRepository _repository;
    private readonly IIdentityContextProvider _identityProvider;
    private readonly IDecision CheckPort _decision checkPort;  // application port

    public async Task<Result<ApproveWithdrawalResult>> Handle(
        ApproveWithdrawalCommand command, CancellationToken ct)
    {
        // 1. Load aggregate
        var withdrawal = await _repository.Get(command.WithdrawalId, ct);

        // 2. Application-level policy check (доступ к конкретному игроку)
        var actor = _identityProvider.GetCurrent().ToSnapshot();

        // 3. External data для domain decision (через application port)
        var decision check = await _decision checkPort.Check(withdrawal.UserId, ct);
        if (!decision check.IsEligibleForPayout)
            return Result.Fail(WithdrawalErrors.UserNotEligible);

        // 4. Domain method (core invariants)
        var result = withdrawal.Approve(actor);
        if (result.IsFailed) return result.ToResult<ApproveWithdrawalResult>();

        // 5. Persist (scope, audit, optimistic concurrency — через pipeline)
        await _repository.Update(withdrawal, ct);

        // 6. Domain events dispatch → MassTransit Outbox (через DomainEventDispatchBehavior)
        return Result.Ok(new ApproveWithdrawalResult(withdrawal.Id, withdrawal.Status));
    }
}

---

Полный Пример Query Handler

// Feature: Accounting/Accounts/GetAccountingSummary

// Query
public sealed record GetAccountingSummaryQuery(
    AccountingAccountId AccountingId
) : IQuery<AccountingSummaryView>,
    ICacheableQuery<AccountingSummaryView>,
    IAuthorizedRequest
{
    // Cache key включает scope — user не видит чужой cache
    public string CacheKey =>
        $"accounting:summary:{_tenantId}:{_brandId}:{_userId}:{AccountingId}";

    public TimeSpan Ttl => TimeSpan.FromSeconds(30);
    public string? InvalidationTag => $"accounting:{AccountingId}";
    public CachePrivacyLevel PrivacyLevel => CachePrivacyLevel.UserOwned;
    public string RequiredPermission => "accounting.account.read";
}

// Handler
public sealed class GetAccountingSummaryQueryHandler
    : IQueryHandler<GetAccountingSummaryQuery, AccountingSummaryView>
{
    private readonly IReadRepository<AccountingAccountView, AccountingAccountId> _repository;

    public async Task<Result<AccountingSummaryView>> Handle(
        GetAccountingSummaryQuery query, CancellationToken ct)
    {
        // Repository scope filter автоматически применит UserId/TenantId/BrandId
        var accounting = await _repository.Find(query.AccountingId, ct);
        if (accounting is null) return Result.Fail(AccountingErrors.AccountNotFound);

        return Result.Ok(accounting.ToSummaryView());
    }
}

---

FluentResults — Typed Error Handling

FluentResults используется на всех уровнях — от Domain до API. Ключевой принцип: исключения — для невозможных состояний; Result.Fail — для ожидаемых бизнес-ошибок.

// Talent.Common.Domain/Errors/DomainErrors.cs

// Typed error — стабильный контракт, не string
public sealed class WithdrawalError : Error
{
    public string Code { get; }

    private WithdrawalError(string code, string message) : base(message)
    {
        Code = code;
        WithMetadata("code", code);
    }

    public static readonly WithdrawalError UserNotEligible =
        new("accounting.withdrawal.user-not-eligible", "User is not eligible for withdrawal");

    public static readonly WithdrawalError InsufficientFunds =
        new("accounting.withdrawal.insufficient-funds", "Insufficient accounting balance");

    public static readonly WithdrawalError LimitExceeded =
        new("accounting.withdrawal.limit-exceeded", "Withdrawal limit exceeded");

    public static WithdrawalError InvalidTransition(string from, string to) =>
        new($"accounting.withdrawal.invalid-transition",
            $"Cannot transition from {from} to {to}");
}

// Использование в domain method:
public Result Approve(ActorSnapshot actor)
{
    if (Status != WithdrawalStatus.Requested)
        return Result.Fail(WithdrawalError.InvalidTransition(
            Status.ToString(), WithdrawalStatus.Approved.ToString()));

    Status    = WithdrawalStatus.Approved;
    UpdatedAt = DateTimeOffset.UtcNow;
    RaiseDomainEvent(new WithdrawalApprovedEvent(Id, actor));
    return Result.Ok();
}

// ResultToProblemDetails mapping (в Pin.Kernel.AspNetCore):
public static class ResultMappingExtensions
{
    public static IActionResult ToActionResult<T>(this Result<T> result)
    {
        if (result.IsSuccess) return new OkObjectResult(result.Value);

        var error = result.Errors.FirstOrDefault();
        var code  = error?.Metadata.TryGetValue("code", out var c) == true
            ? c?.ToString()
            : "unknown-error";

        return error switch
        {
            UnauthorizedError  => new ObjectResult(ProblemDetailsFactory.Unauthorized(code!))
                                      { StatusCode = 401 },
            ForbiddenError     => new ObjectResult(ProblemDetailsFactory.Forbidden(code!))
                                      { StatusCode = 403 },
            NotFoundError      => new ObjectResult(ProblemDetailsFactory.NotFound(code!))
                                      { StatusCode = 404 },
            ConflictError      => new ObjectResult(ProblemDetailsFactory.Conflict(code!))
                                      { StatusCode = 409 },
            _                  => new ObjectResult(ProblemDetailsFactory.UnprocessableEntity(code!))
                                      { StatusCode = 422 },
        };
    }
}

Типизированные base errors для инфраструктуры:

// Talent.Common.Application/Results/

public sealed class NotFoundError : Error
{
    public NotFoundError(string entityType, object id)
        : base($"{entityType} with id '{id}' not found")
    {
        WithMetadata("code", $"{entityType.ToLower()}.not-found");
        WithMetadata("entityType", entityType);
        WithMetadata("id", id.ToString());
    }
}

public sealed class UnauthorizedError : Error
{
    public UnauthorizedError(string reason = "unauthenticated")
        : base($"Unauthorized: {reason}")
    {
        WithMetadata("code", reason);
    }
}

public sealed class ForbiddenError : Error
{
    public ForbiddenError(string requiredPermission)
        : base($"Missing permission: {requiredPermission}")
    {
        WithMetadata("code", "forbidden");
        WithMetadata("requiredPermission", requiredPermission);
    }
}

public sealed class ConflictError : Error
{
    public ConflictError(string code, string message) : base(message)
    {
        WithMetadata("code", code);
    }
}

---

FluentValidation Правила

// Именование: {FeatureName}CommandValidator / {FeatureName}QueryValidator

public sealed class CreateWithdrawalCommandValidator
    : AbstractValidator<CreateWithdrawalCommand>
{
    public CreateWithdrawalCommandValidator(IWithdrawalLimitPort limitPort)
    {
        // Синхронные rules — всегда
        RuleFor(x => x.Amount.Amount)
            .GreaterThan(0).WithErrorCode("accounting.withdrawal.amount-positive")
            .LessThanOrEqualTo(MaxWithdrawalAmount).WithErrorCode("accounting.withdrawal.amount-too-large");

        RuleFor(x => x.Amount.CurrencyId).NotEmpty();

        // Async rules — только при необходимости, без DB calls в validator
        // DB calls выполняются в handler после validation
    }
}

Правила:

• Validators не обращаются в БД напрямую — это задача handler.
• Validator использует только простые правила формата/диапазонов.
• Error code — стабильный контракт: <service>.<feature>.<reason>.
• Async rules в validator допустимы только для lookup в памяти или cache (не БД).

---

Регистрация MediatR

// Pin.Accounting.Application/DependencyInjection/AccountingApplicationModule.cs

public static class AccountingApplicationModule
{
    public static IPinApplicationBuilder AddAccountingApplication(
        this IPinApplicationBuilder builder)
    {
        builder
            .AddFeatureAssembly(typeof(ApproveWithdrawalCommand).Assembly)
            // MediatR, validators, Mapster регистрируются через AddFeatureAssembly
            .AddPinMediatR()  // ← регистрирует все pipeline behaviors в правильном порядке
            .AddAccountingPolicies()
            .AddAccountingMapsterConfig();

        return builder;
    }
}

AddPinMediatR() регистрирует behaviors в правильном порядке, сканирует assembly для handlers и validators.

---

Ссылки

• MediatR Pipeline Behaviors
• Result И Error Handling
• Idempotency Standard
• Audit Trail Standard
• Backend Rules

Обратные ссылки

Автоссылки: где используется этот документ.

• index.md (1): Backend SDK — Руководство Разработчика
• 08-testing-patterns.md (1): Тестирование — Практическое Руководство
• 12-resilience-and-audit.md (1): Resilience, FluentValidation и Audit — Практическое Руководство
• 13-code-quality-and-typing.md (1): Code Quality И Strong Typing — Enterprise .NET 10 Standards
• 14-security-patterns.md (1): Security Patterns — Authorization, PII, OWASP
• backend-shared-standards-index.md (1): Backend Shared Standards Index