Documentation

Pipelines

The extension point: validation, logging, transactions, metrics, multi-tenancy.

The contract

An IPipeline wraps every handler invocation. Four hooks, one per shape of message. Derive from Pipeline and override only what matters — the rest pass straight through.

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;
}

Order

Pipelines run in registration order: the first is the outermost. On the way back, they unwind in reverse.

Text

AddLoggingPipeline()      ┐ in 1st, out 5th
AddTimeoutPipeline()      │ in 2nd, out 4th  ← only acts on messages with [MessageTimeout]
AddEnrichmentPipeline()   │ in 3rd, out 3rd
AddValidationPipeline()   │ in 4th, out 2nd
AddUnitOfWorkPipeline()   ┘ in 5th, out 1st  ← closest to the handler
                              └─ HANDLER

That is the order AddNexo<TContext> assembles, and every position has a reason: logging wraps everything, including the time spent validating; enrichment runs before validation so a rule can validate the author; and the unit of work goes last, so the transaction covers only the handler and its events.

Registering the same pipeline twice does not make it run twice: registration is idempotent, so one extra AddLoggingPipeline() in a library or a test does not duplicate the log.

Example: metrics

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

    public MetricsPipeline(IMeterFactory meters)
    {
        var meter = meters.Create("MyApp.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());

Example: multi-tenancy

A pipeline can reject the message before the handler. Register it before the unit of work, so the refusal does not cost a transaction.

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("No tenant in the request context.");
        }
    }
}

Replacing the result

The hook returns the result, so a pipeline can swap it — that is how you write query caching without touching the 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;
    }
}

Ways to register

services.AddNexo(nexo => nexo
    // By type, with a lifetime (Scoped by default).
    .AddPipeline<TenantPipeline>()
    .AddPipeline<MetricsPipeline>(ServiceLifetime.Singleton)

    // By factory, when construction depends on the scope.
    .AddPipeline(sp => new FeatureFlagPipeline(sp.GetRequiredService<IFeatureManager>()))

    // By instance, when the pipeline has no dependencies at all.
    .AddPipeline(new StopwatchPipeline()));

The built-in pipelines

Pipeline	Package	What it does
`LoggingPipeline`	`Nexo`	A log scope with the message name and id, and the handler's timing. Payload logging off by default.
`MessageTimeoutPipeline`	`Nexo`	Cancels the handler's token at the `[MessageTimeout]` deadline and throws `TimeoutException`. Messages without the attribute pass straight through. On a stream query the budget starts at enumeration and covers the whole stream, including the time between one item and the next.
`MessageEnrichmentPipeline`	`Nexo`	Fills `CreatedBy` from `ICurrentUserAccessor`, respecting a value that is already set.
`UnitOfWorkPipeline`	`Nexo`	Transaction, single save, domain event dispatch and re-entrancy protection.
`ValidationPipeline`	`Nexo.FluentValidation`	Runs every validator for the message before the handler and reports the failures together in `ValidationFailedException`.

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

Not every exception is an incident. A refused validation, a record that does not exist, a business rule, a duplicate key, a concurrency conflict and a cancellation the client asked for are answers, and go to the log at Information (ExpectedFailureLevel), with no error stack trace. Everything else goes out at Error. Adjust what counts as expected with IsExpectedFailure.

A failure inside a nested dispatch — an event handler within a command — travels back out through every level. The stack trace is written once, by the innermost level; the outer ones only record that their message failed.

Telemetry

The mediator publishes one span per message on an ActivitySource and two metrics on a Meter, both named Nexo. With nobody listening, the cost is one check.

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

Signal	Name	Contents
Span	`command CreateProduct`, `query …`, `event …`	Tags `nexo.message.kind`, `nexo.message.type`, `nexo.message.id`; on commands, `nexo.transaction`, `nexo.saved_rows` and `nexo.domain_event_rounds`. Error status with the exception.
Histogram	`nexo.message.duration`	Milliseconds per message, pipelines included, tagged with `nexo.message.kind` and `nexo.message.type`.
Counter	`nexo.message.failures`	Messages whose handling threw.

Spans nest the way the code does: an event dispatched inside a command shows up as a child of the command's span, and EF Core's queries — with its instrumentation turned on — as children of the handler that made them.