Documentação

Pipelines

O ponto de extensão: validação, logging, transação, métricas, multi-tenancy.

O contrato

Um IPipeline envolve toda invocação de handler. Quatro hooks, um por forma de mensagem. Herde de Pipeline e sobrescreva só o que importa — os outros passam direto.

public interface IPipeline
{
    Task OnCommandAsync<TCommand>(
        Func<TCommand, CancellationToken, Task> next, TCommand command, CancellationToken ct)
        where TCommand : class, ICommand;

    Task<TResult> OnCommandAsync<TCommand, TResult>(
        Func<TCommand, CancellationToken, Task<TResult>> next, TCommand command, CancellationToken ct)
        where TCommand : class, ICommand<TResult>;

    Task<TResult> OnQueryAsync<TQuery, TResult>(
        Func<TQuery, CancellationToken, Task<TResult>> next, TQuery query, CancellationToken ct)
        where TQuery : class, IQuery<TResult>;

    Task OnEventAsync<TEvent>(
        Func<TEvent, CancellationToken, Task> next, TEvent @event, CancellationToken ct)
        where TEvent : class, IEvent;
}

Ordem

Pipelines rodam na ordem de registro: o primeiro é o mais externo. Na volta, desenrolam ao contrário.

Texto

AddLoggingPipeline()      ┐ entra 1º, sai 5º
AddTimeoutPipeline()      │ entra 2º, sai 4º  ← só age em mensagens com [MessageTimeout]
AddEnrichmentPipeline()   │ entra 3º, sai 3º
AddValidationPipeline()   │ entra 4º, sai 2º
AddUnitOfWorkPipeline()   ┘ entra 5º, sai 1º  ← encostado no handler
                              └─ HANDLER

Essa é a ordem que AddNexo<TContext> monta, e cada posição tem motivo: o log envolve tudo, incluindo o tempo de validação; o enriquecimento roda antes da validação para que uma regra possa validar o autor; e o unit of work fica por último, para que a transação cubra só o handler e seus eventos.

Registrar o mesmo pipeline duas vezes não o faz rodar duas vezes: o registro é idempotente, então um AddLoggingPipeline() a mais numa biblioteca ou num teste não duplica o log.

Exemplo: métricas

public sealed class MetricsPipeline : Pipeline
{
    private readonly Histogram<double> _duration;

    public MetricsPipeline(IMeterFactory meters)
    {
        var meter = meters.Create("MinhaApp.Cqrs");
        _duration = meter.CreateHistogram<double>("cqrs.handler.duration", "ms");
    }

    public override Task<TResult> OnCommandAsync<TCommand, TResult>(
        Func<TCommand, CancellationToken, Task<TResult>> next,
        TCommand command,
        CancellationToken ct)
        => MeasureAsync("command", typeof(TCommand).Name, () => next(command, ct));

    public override Task<TResult> OnQueryAsync<TQuery, TResult>(
        Func<TQuery, CancellationToken, Task<TResult>> next,
        TQuery query,
        CancellationToken ct)
        => MeasureAsync("query", typeof(TQuery).Name, () => next(query, ct));

    private async Task<TResult> MeasureAsync<TResult>(
        string kind, string name, Func<Task<TResult>> work)
    {
        var start = Stopwatch.GetTimestamp();
        var outcome = "ok";

        try
        {
            return await work();
        }
        catch
        {
            outcome = "error";
            throw;
        }
        finally
        {
            _duration.Record(
                Stopwatch.GetElapsedTime(start).TotalMilliseconds,
                new KeyValuePair<string, object?>("kind", kind),
                new KeyValuePair<string, object?>("name", name),
                new KeyValuePair<string, object?>("outcome", outcome));
        }
    }
}

services.AddNexo(nexo => nexo
    .AddLoggingPipeline()
    .AddPipeline<MetricsPipeline>(ServiceLifetime.Singleton)
    .AddUnitOfWorkPipeline());

Exemplo: multi-tenancy

Um pipeline pode rejeitar a mensagem antes do handler. Registre antes do unit of work, para que a recusa não custe uma transação.

public sealed class TenantPipeline(ITenantContext tenant) : Pipeline
{
    public override Task OnCommandAsync<TCommand>(
        Func<TCommand, CancellationToken, Task> next,
        TCommand command,
        CancellationToken ct)
    {
        Require();
        return next(command, ct);
    }

    public override Task<TResult> OnCommandAsync<TCommand, TResult>(
        Func<TCommand, CancellationToken, Task<TResult>> next,
        TCommand command,
        CancellationToken ct)
    {
        Require();
        return next(command, ct);
    }

    private void Require()
    {
        if (tenant.Current is null)
        {
            throw new InvalidOperationException("Nenhum tenant no contexto da requisição.");
        }
    }
}

Substituindo o resultado

O hook devolve o resultado, então um pipeline pode trocá-lo — é assim que se escreve cache de query sem tocar no handler.

public sealed class QueryCachePipeline(IMemoryCache cache) : Pipeline
{
    public override async Task<TResult> OnQueryAsync<TQuery, TResult>(
        Func<TQuery, CancellationToken, Task<TResult>> next,
        TQuery query,
        CancellationToken ct)
    {
        if (query is not ICacheableQuery cacheable)
        {
            return await next(query, ct);
        }

        if (cache.TryGetValue(cacheable.CacheKey, out TResult? hit))
        {
            return hit!;
        }

        var result = await next(query, ct);
        cache.Set(cacheable.CacheKey, result, cacheable.Ttl);

        return result;
    }
}

Formas de registrar

services.AddNexo(nexo => nexo
    // Por tipo, com tempo de vida (Scoped por padrão).
    .AddPipeline<TenantPipeline>()
    .AddPipeline<MetricsPipeline>(ServiceLifetime.Singleton)

    // Por fábrica, quando a construção depende do escopo.
    .AddPipeline(sp => new FeatureFlagPipeline(sp.GetRequiredService<IFeatureManager>()))

    // Por instância, quando o pipeline não tem dependência nenhuma.
    .AddPipeline(new StopwatchPipeline()));

Os pipelines embutidos

Pipeline	Pacote	O que faz
`LoggingPipeline`	`Nexo`	Escopo de log com nome e id da mensagem, e o tempo do handler. Payload desligado por padrão.
`MessageTimeoutPipeline`	`Nexo`	Cancela o token do handler no prazo de `[MessageTimeout]` e lança `TimeoutException`. Mensagens sem o atributo passam direto. Em stream query o orçamento começa na enumeração e cobre o stream inteiro, inclusive o tempo entre um item e o próximo.
`MessageEnrichmentPipeline`	`Nexo`	Preenche `CreatedBy` a partir do `ICurrentUserAccessor`, respeitando um valor já definido.
`UnitOfWorkPipeline`	`Nexo`	Transação, save único, despacho de eventos de domínio e proteção de reentrância.
`ValidationPipeline`	`Nexo.FluentValidation`	Roda todos os validators da mensagem antes do handler e reporta as falhas juntas em `ValidationFailedException`.

nexo.AddLoggingPipeline(logging =>
{
    logging.Level = LogLevel.Information;
    logging.LogPayload = builder.Environment.IsDevelopment();
    logging.Serializer = message => JsonSerializer.Serialize(message, RedactingOptions);
});

Nem toda exceção é um incidente. Uma validação recusada, um registro que não existe, uma regra de negócio, uma chave duplicada, um conflito de concorrência e um cancelamento pedido pelo cliente são respostas, e saem no log em Information (ExpectedFailureLevel), sem stack trace de erro. O resto sai em Error. Ajuste o que conta como esperado com IsExpectedFailure.

Uma falha dentro de um despacho aninhado — um handler de evento dentro de um comando — atravessa todos os níveis na volta. O stack trace é escrito uma vez, pelo nível mais interno; os de fora registram só que a mensagem deles falhou.

Telemetria

O mediator publica um span por mensagem num ActivitySource e duas métricas num Meter, ambos chamados Nexo. Sem ninguém ouvindo, o custo é uma checagem.

builder.Services.AddOpenTelemetry()
    .WithTracing(tracing => tracing.AddSource("Nexo").AddAspNetCoreInstrumentation())
    .WithMetrics(metrics => metrics.AddMeter("Nexo"));

Sinal	Nome	Conteúdo
Span	`command CreateProduct`, `query …`, `event …`	Tags `nexo.message.kind`, `nexo.message.type`, `nexo.message.id`; em comandos, `nexo.transaction`, `nexo.saved_rows` e `nexo.domain_event_rounds`. Status de erro com a exceção.
Histograma	`nexo.message.duration`	Milissegundos por mensagem, pipelines incluídos, com as tags `nexo.message.kind` e `nexo.message.type`.
Contador	`nexo.message.failures`	Mensagens cujo tratamento lançou.

Os spans aninham como o código: o evento despachado dentro de um comando aparece como filho do span do comando, e as consultas do EF Core — com a instrumentação dele ligada — como filhas do handler que as fez.