Josephan.CQRS 3.0.4

Josephan.CQRS

A lightweight, high-performance CQRS library for .NET — built for speed, low allocation, and a flexible pipeline behavior system that supports open-generic and closed-generic behaviors.

Installation

dotnet add package Josephan.CQRS

Core Concept

The pipeline is built on:

IPipelineBehavior<TRequest, TResponse>

and supports:

• Open-generic behaviors — apply to all requests globally
• Closed-generic behaviors — apply to a specific request/response type
• Command-only behaviors — scoped to the write side
• Query-only behaviors — scoped to the read side

Getting Started

Register CQRS

// Single assembly
services.AddCQRS(options =>
{
    options.AddHandlersFromAssemblies(typeof(Program).Assembly);

    options.AddBehavior(typeof(ValidationBehavior<,>));
    options.AddBehavior(typeof(LoggingBehavior<,>));
});

// Multiple assemblies
services.AddCQRS(options =>
{
    options.AddHandlersFromAssemblies(
        typeof(Program).Assembly,
        typeof(OrderModule).Assembly);

    options.AddBehavior(typeof(ValidationBehavior));
    options.AddBehavior(typeof(LoggingBehavior));
});

Commands

Define a Command

public record CreateUserCommand(string Name, string Email) : ICommand;

Command Handler

public class CreateUserCommandHandler : ICommandHandler<CreateUserCommand>
{
    public Task Handle(CreateUserCommand command, CancellationToken ct)
        => Task.CompletedTask;
}

Command with Response

public record CreateUserCommand(string Name) : ICommand<Guid>;

public class CreateUserCommandHandler : ICommandHandler<CreateUserCommand, Guid>
{
    public Task<Guid> Handle(CreateUserCommand command, CancellationToken ct)
        => Task.FromResult(Guid.NewGuid());
}

Queries

Define a Query

public record GetUserQuery(Guid Id) : IQuery<UserDto>;

Query Handler

public class GetUserQueryHandler : IQueryHandler<GetUserQuery, UserDto>
{
    public Task<UserDto> Handle(GetUserQuery query, CancellationToken ct)
        => Task.FromResult(new UserDto(query.Id, "Yousef"));
}

Senders

ISender — Unified API (recommended)

The single entry point for both commands and queries. Best for most applications.

app.MapPost("/users", async (ISender sender) =>
{
    var result = await sender.Send(new CreateUserCommand("Ali", "ali@mail.com"));
    return Results.Ok(result);
});

app.MapGet("/users/{id}", async (ISender sender, Guid id) =>
{
    var result = await sender.Send(new GetUserQuery(id));
    return Results.Ok(result);
});

ICommandSender — Write Side Isolation

Restrict a service to commands only, enforcing CQRS separation at the dependency level.

public class UserService(ICommandSender sender)
{
    public Task<Guid> CreateUser(string name, string email)
        => sender.Send(new CreateUserCommand(name, email));
}

IQuerySender — Read Side Optimization

Restrict a service to queries only, useful for read-optimized layers.

public class UserQueryService(IQuerySender sender)
{
    public Task<UserDto> GetUser(Guid id)
        => sender.Send(new GetUserQuery(id));
}

When to use what?

Pipeline Behaviors

Open-Generic Behavior

Runs for every request through the pipeline.

public class LoggingBehavior<TReq, TRes> : IPipelineBehavior<TReq, TRes>
    where TReq : IRequest<TRes>
{
    public async Task<TRes> Handle(
        TReq request,
        RequestHandlerDelegate<TReq, TRes> next,
        CancellationToken ct)
    {
        Console.WriteLine($"Handling {typeof(TReq).Name}");
        var response = await next(request, ct);
        Console.WriteLine($"Handled {typeof(TReq).Name}");
        return response;
    }
}

Closed-Generic Behavior

Runs only for a specific request/response pair — ideal for caching, auditing, or type-specific logic.

public class CachingBehavior : IPipelineBehavior<GetUserQuery, UserDto>
{
    public async Task<UserDto> Handle(
        GetUserQuery request,
        RequestHandlerDelegate<GetUserQuery, UserDto> next,
        CancellationToken ct)
    {
        return await next(request, ct);
    }
}

Registering Behaviors

Via AddCQRS (recommended)

services.AddCQRS(options =>
{
    options.AddHandlersFromAssemblies(typeof(Program).Assembly);

    // Open-generic — applies to all requests
    options.AddBehavior(typeof(LoggingBehavior<,>));
    options.AddBehavior(typeof(ValidationBehavior<,>));

    // Closed-generic — applies to one specific request
    options.AddBehavior<DummyQuery, string, ShortCircuitBehavior>();

    // Or Simply (Fixed Bug In Previos Versions)
    options.AddBehavior<ShortCircuitBehavior>()

    // Command-only
    options.AddCommandBehavior(typeof(AuditBehavior<,>));

    // Query-only
    options.AddQueryBehavior(typeof(QueryCachingBehavior<,>));
});

Calling AddCQRS from Multiple Layers/Modules

It's safe to call AddCQRS more than once — for example, once from an Application layer module to register behaviors, and once from an Infrastructure layer module to register handlers:

// Application layer
services.AddCQRS(options =>
{
    options.AddBehavior(typeof(LoggingBehavior<,>));
    options.AddCommandBehavior(typeof(AuditBehavior<,>));
});

// Infrastructure layer
services.AddCQRS(options =>
{
    options.AddHandlersFromAssemblies(
        typeof(ApplicationAssemblyMarker).Assembly,
        typeof(InfrastructureAssemblyMarker).Assembly);
});

Handlers, behaviors, and pipeline decoration are all idempotent — no duplicate registrations, no double-wrapped pipelines, no matter how many modules call AddCQRS or in what order.

Via External DI Registration

You can also register behaviors directly through the DI container — useful when integrating with third-party modules, feature flags, or when you want full control outside of AddCQRS.

// Open-generic behavior — applies to all requests
services.AddTransient(
    typeof(IPipelineBehavior<,>),
    typeof(LoggingBehavior<,>));

// Closed-generic behavior — applies to a specific request/response type
services.AddTransient(
    typeof(IPipelineBehavior<GetUserQuery, UserDto>),
    typeof(CachingBehavior));

// Command-only behavior
services.AddTransient(
    typeof(ICommandPipelineBehavior<,>),
    typeof(AuditBehavior<,>));

// Query-only behavior
services.AddTransient(
    typeof(IQueryPipelineBehavior<,>),
    typeof(QueryCachingBehavior<,>));

Note: External registrations are resolved in the order they are registered. Make sure cross-cutting behaviors like logging are registered before more specific ones like validation.

Command-Only Behaviors

Scoped to commands, never runs on queries.

public class AuditBehavior<TReq, TRes> : ICommandPipelineBehavior<TReq, TRes>
    where TReq : IRequest<TRes>
{
    public async Task<TRes> Handle(
        TReq command,
        RequestHandlerDelegate<TReq, TRes> next,
        CancellationToken ct)
        => await next(command, ct);
}

cfg.AddCommandBehavior(typeof(AuditBehavior<,>));

Query-Only Behaviors

Two approaches — via the dedicated interface or via a generic constraint:

Using IQueryPipelineBehavior

public class QueryCachingBehavior<TReq, TRes> : IQueryPipelineBehavior<TReq, TRes>
    where TReq : IRequest<TRes>
{
    public async Task<TRes> Handle(
        TReq query,
        RequestHandlerDelegate<TReq, TRes> next,
        CancellationToken ct)
        => await next(query, ct);
}

cfg.AddQueryBehavior(typeof(QueryCachingBehavior<,>));

Using IQuery<TRes> constraint on IPipelineBehavior

Alternatively, constrain directly on IQuery<TRes> — the pipeline will only invoke this behavior for queries.

public class QueryCachingBehavior<TReq, TRes> : IPipelineBehavior<TReq, TRes>
    where TReq : IQuery<TRes>
{
    public async Task<TRes> Handle(
        TReq query,
        RequestHandlerDelegate<TReq, TRes> next,
        CancellationToken ct)
        => await next(query, ct);
}

Notifications

Notifications let you broadcast an event to multiple handlers — useful for domain events, audit trails, side effects, and decoupled cross-cutting concerns.

Define a Notification

public record UserCreatedNotification(Guid UserId, string Email) : INotification;

Notification Handlers

Multiple handlers can subscribe to the same notification — all of them will be invoked.

public class SendWelcomeEmailHandler : INotificationHandler<UserCreatedNotification>
{
    public Task Handle(UserCreatedNotification notification, CancellationToken ct = default)
    {
        Console.WriteLine($"Sending welcome email to {notification.Email}");
        return Task.CompletedTask;
    }
}

public class AuditUserCreationHandler : INotificationHandler<UserCreatedNotification>
{
    public Task Handle(UserCreatedNotification notification, CancellationToken ct = default)
    {
        Console.WriteLine($"Auditing user creation: {notification.UserId}");
        return Task.CompletedTask;
    }
}

Publishing a Notification

Sequential (default)

Handlers are invoked one after another in registration order. All handlers run even if one throws — exceptions are collected and rethrown as an BroadcastFailedException.

public class UserService(ICommandSender sender, INotificationPublisher publisher)
{
    public async Task<Guid> CreateUser(string name, string email)
    {
        var userId = await sender.Send(new CreateUserCommand(name, email));

        await publisher.PublishAsync(new UserCreatedNotification(userId, email));

        return userId;
    }
}

Parallel

All handlers are invoked concurrently. Exceptions from any handler are collected and rethrown as an BroadcastFailedException.

try
{
    await publisher.PublishAsync(
        new UserCreatedNotification(userId, email),
        NotificationPublishingStrategy.Parallel);
}
catch (BroadcastFailedException ex)
{
    foreach (var typed in ex.TypedExceptions)
    {
        logger.LogError(typed.Exception,
            "Handler {Handler} failed", typed.Type.Name);
    }
}

Publishing Strategies

In Minimal APIs

app.MapPost("/users", async (INotificationPublisher publisher, ISender sender) =>
{
    var userId = await sender.Send(new CreateUserCommand("Ali", "ali@mail.com"));
    await publisher.PublishAsync(new UserCreatedNotification(userId, "ali@mail.com"));
    return Results.Ok(userId);
});

License

MIT © Yousef Yahia