Wolverine Integration in Momentum

Wolverine is a powerful messaging framework that serves as the backbone of Momentum's CQRS and messaging infrastructure. It provides in-process messaging, external transport integration, and advanced middleware capabilities.

What is Wolverine?

Wolverine (formerly Jasper) is a .NET messaging framework that provides:

• In-process messaging: CQRS command/query handling
• External transports: Kafka, RabbitMQ, Azure Service Bus integration
• Middleware pipeline: Powerful interceptor pattern
• Message persistence: Reliable message delivery with PostgreSQL
• Code generation: High-performance handlers through source generation

Architecture Overview

Basic Configuration

Wolverine is automatically configured through AddServiceDefaults():

var builder = WebApplication.CreateBuilder(args);

// Automatically configures Wolverine with sensible defaults
builder.AddServiceDefaults();

var app = builder.Build();

// Wolverine commands are supported
await app.RunAsync(args);

Manual Configuration

For more control, configure Wolverine manually:

builder.Services.AddWolverine(opts =>
{
    opts.ServiceName = "MyService";

    // Configure PostgreSQL persistence
    opts.PersistMessagesWithPostgresql(connectionString);

    // Add middleware
    opts.Policies.AddMiddleware<CustomMiddleware>();

    // Configure discovery
    opts.Discovery.IncludeAssembly(typeof(MyDomainAssembly).Assembly);
});

Handler Discovery

Automatic Discovery

Handlers are automatically discovered from domain assemblies:

// Mark assemblies for discovery
[assembly: DomainAssembly(typeof(IAppDomainAssembly))]

// Handlers are automatically found
public static class CreateCashierCommandHandler
{
    public static async Task<(Result<Cashier>, CashierCreated?)> Handle(
        CreateCashierCommand command,
        IMessageBus messaging,
        CancellationToken cancellationToken)
    {
        // Handler logic
    }
}

Handler Patterns

Wolverine supports various handler patterns:

// Static method handlers (recommended)
public static class MyCommandHandler
{
    public static async Task<Result<T>> Handle(MyCommand command, CancellationToken token)
    {
        // Implementation
    }
}

// Instance method handlers
public class MyQueryHandler
{
    private readonly IDbContext _db;

    public MyQueryHandler(IDbContext db)
    {
        _db = db;
    }

    public async Task<Result<T>> Handle(MyQuery query, CancellationToken token)
    {
        // Implementation using _db
    }
}

// Multiple handlers in one class
public static class CashierHandlers
{
    public static Task<Result<Cashier>> Handle(GetCashierQuery query, AppDomainDb db, CancellationToken token)
    {
        // Query handler
    }

    public static Task<(Result<Cashier>, CashierCreated?)> Handle(CreateCashierCommand command, IMessageBus bus, CancellationToken token)
    {
        // Command handler
    }
}

Message Bus Usage

IMessageBus Interface

The IMessageBus interface provides the primary API for messaging:

public interface IMessageBus
{
    // Execute commands/queries
    Task<TResponse> InvokeAsync<TResponse>(IQuery<TResponse> query, CancellationToken cancellationToken);
    Task<TResponse> InvokeAsync<TResponse>(ICommand<TResponse> command, CancellationToken cancellationToken);

    // Execute database commands
    Task<TResponse> InvokeCommandAsync<TResponse>(ICommand<TResponse> command, CancellationToken cancellationToken);

    // Publish events
    Task PublishAsync<T>(T message, CancellationToken cancellationToken);
    Task PublishAsync<T>(T message, DeliveryOptions? options, CancellationToken cancellationToken);
}

Usage Examples

public class CashierController : ControllerBase
{
    private readonly IMessageBus _messageBus;

    public CashierController(IMessageBus messageBus)
    {
        _messageBus = messageBus;
    }

    [HttpPost]
    public async Task<ActionResult<Cashier>> CreateCashier(CreateCashierCommand command)
    {
        var (result, integrationEvent) = await _messageBus.InvokeAsync(command);

        if (result.IsSuccess)
        {
            // Integration event is automatically published if returned from handler
            return CreatedAtAction(nameof(GetCashier), new { id = result.Value.Id }, result.Value);
        }

        return BadRequest(result.Errors);
    }

    [HttpGet("{id}")]
    public async Task<ActionResult<Cashier>> GetCashier(Guid id)
    {
        var query = new GetCashierQuery(GetTenantId(), id);
        var result = await _messageBus.InvokeAsync(query);

        if (result.IsSuccess)
        {
            return Ok(result.Value);
        }

        return NotFound();
    }
}

Middleware Pipeline

Built-in Middleware

Momentum configures several middleware components automatically:

// Automatically configured middleware
opts.Policies.AddMiddleware<RequestPerformanceMiddleware>();        // Performance tracking
opts.Policies.AddMiddleware(typeof(OpenTelemetryInstrumentationMiddleware)); // Tracing
opts.Policies.Add<ExceptionHandlingPolicy>();                      // Error handling
opts.Policies.Add<FluentValidationPolicy>();                       // Validation

Custom Middleware

Create custom middleware for cross-cutting concerns:

public class AuditMiddleware
{
    public async Task InvokeAsync<T>(MessageContext context, T message, Func<Task> next)
    {
        var auditService = context.Services.GetRequiredService<IAuditService>();

        // Before handler execution
        var auditId = await auditService.StartAuditAsync(typeof(T).Name, message);

        try
        {
            await next();

            // After successful execution
            await auditService.CompleteAuditAsync(auditId, success: true);
        }
        catch (Exception ex)
        {
            // After failed execution
            await auditService.CompleteAuditAsync(auditId, success: false, ex.Message);
            throw;
        }
    }
}

// Register middleware
builder.Services.AddWolverine(opts =>
{
    opts.Policies.AddMiddleware<AuditMiddleware>();
});

Conditional Middleware

Apply middleware conditionally:

// Only apply to commands
opts.Policies.AddMiddleware<AuditMiddleware>(
    x => x.HandledMessageType.Implements<ICommand>()
);

// Only apply to specific handlers
opts.Policies.AddMiddleware<CachingMiddleware>(
    x => x.HandledMessageType == typeof(GetCashierQuery)
);

Message Routing

Local Message Routing

Local messages (commands/queries) are routed to handlers within the same process:

// This stays within the process
var result = await messageBus.InvokeAsync(new GetCashierQuery(tenantId, id));

External Message Routing

Integration events are routed to external transports:

// This is published to Kafka automatically when returned from handler
public static async Task<(Result<Cashier>, CashierCreated?)> Handle(
    CreateCashierCommand command,
    IMessageBus messaging,
    CancellationToken cancellationToken)
{
    // ... business logic

    var integrationEvent = new CashierCreated(tenantId, cashier);
    return (result, integrationEvent); // Automatically published
}

// Manual publishing
await messageBus.PublishAsync(new CashierCreated(tenantId, cashier));

Routing Rules

Configure custom routing rules:

builder.Services.AddWolverine(opts =>
{
    // Route all events to Kafka
    opts.PublishAllMessages().ToKafka();

    // Route specific messages to specific topics
    opts.PublishMessage<OrderCreated>().ToKafkaTopic("orders");

    // Route to local queues
    opts.PublishMessage<InternalNotification>().ToLocalQueue("notifications");
});

Persistence and Reliability

PostgreSQL Persistence

Wolverine uses PostgreSQL for reliable message delivery. In a generated service, message persistence reuses the application's registered NpgsqlDataSource — the same data source the transactional outbox middleware opens its transaction on — so the outbox tables always live in the application database:

// Configured automatically by reusing the application's NpgsqlDataSource
var appDataSource = serviceProvider.GetService<NpgsqlDataSource>();
opts.PersistMessagesWithPostgresql(appDataSource, schemaName: "messaging");

When the library runs standalone, without an application NpgsqlDataSource registered in DI, it falls back to the ServiceBus connection string:

// Standalone fallback
var connectionString = configuration.GetConnectionString("ServiceBus");
opts.PersistMessagesWithPostgresql(connectionString, schemaName: "messaging");

Outbox Pattern

The outbox pattern ensures reliable event publishing:

// Automatically configured
opts.Policies.UseDurableOutboxOnAllSendingEndpoints();

// This ensures that if the database transaction succeeds,
// the integration event will be published, even if the service crashes
public static async Task<(Result<Cashier>, CashierCreated?)> Handle(
    CreateCashierCommand command,
    IMessageBus messaging,
    CancellationToken cancellationToken)
{
    // Database operation and event publishing are transactionally consistent
    var dbCommand = CreateInsertCommand(command);
    var insertedCashier = await messaging.InvokeCommandAsync(dbCommand, cancellationToken);

    var result = insertedCashier.ToModel();
    var createdEvent = new CashierCreated(result.TenantId, 0, result);

    return (result, createdEvent); // Published reliably via outbox
}

Message Durability

Configure durability options:

builder.Services.AddWolverine(opts =>
{
    // Use durable local queues for internal messages
    opts.Policies.UseDurableLocalQueues();

    // Apply transactions automatically
    opts.Policies.AutoApplyTransactions();

    // Configure retry policies
    opts.Policies.OnException<SqlException>().Retry(3);
    opts.Policies.OnException<HttpRequestException>().RetryWithCooldown(1.Seconds(), 5.Seconds(), 10.Seconds());
});

Error Handling

Default Failure Policy

With reliable messaging enabled (the default), Momentum installs a transient-failure policy for NpgsqlException (transient) and TimeoutException:

1. Two quick in-process retries with cooldown (50ms, 250ms)
2. Two durable scheduled retries (5s, 30s) — the message is persisted with a future execution time and the listener is released, so a Kafka partition is never blocked by a cooling-down message
3. Dead letter queue

Application-specific rules added in the AddServiceBus configure callback run before these defaults and take precedence:

builder.AddServiceBus(opts =>
{
    // Never retry business rule violations
    opts.OnException<BusinessRuleException>().MoveToErrorQueue();

    // Retry an external dependency harder than the default
    opts.OnException<HttpRequestException>()
        .RetryWithCooldown(1.Seconds(), 5.Seconds())
        .Then.ScheduleRetry(1.Minutes(), 5.Minutes())
        .Then.MoveToErrorQueue();
});

Dead Letter Queue: Inspect and Replay

Failed messages land in the Wolverine persistence schema of the application database (svcbus_{service}.wolverine_dead_letters). To inspect and recover:

# Summarize persisted envelope counts (incoming/outgoing/scheduled/dead-letter)
dotnet run -- storage counts

# Replay dead-lettered messages (marks them replayable; the durability agent re-enqueues them)
dotnet run -- storage replay

# Replay only dead letters caused by a specific exception type
dotnet run -- storage replay --exception-type Npgsql.NpgsqlException

Note: storage clear exists but deletes all persisted envelopes (incoming, outgoing, scheduled, and dead letter) — do not use it to discard dead letters only; use IDeadLetterAdminService.DiscardAsync for targeted removal.

Programmatic access is available through Wolverine's IDeadLetterAdminService (query, discard, replay by message type, exception type, or time range), which is registered automatically with the PostgreSQL message store. Dead-letter rows include the exception type, message, and full envelope, so a fixed bug can be followed by a targeted replay instead of manual reprocessing.

At-Least-Once Delivery on Kafka

Kafka listeners use Wolverine's CommitMode.StoreThenAutoFlush: offsets are stored only after a message is successfully processed (or persisted to the durable inbox) and never ahead of the lowest in-flight offset. Combined with the durable inbox (enabled by default via reliable messaging), this gives at-least-once delivery with duplicate detection by message id.

WARNING

Do not set EnableAutoCommit in the Kafka consumer configuration. An explicit EnableAutoCommit: true suppresses Wolverine's offset management and reverts to librdkafka's store-on-consume behavior, where a crash during message processing loses the message. Momentum logs a warning at startup if this misconfiguration is detected.

Custom Error Handling

Implement custom error handling in middleware:

public class ErrorHandlingMiddleware
{
    private readonly ILogger<ErrorHandlingMiddleware> _logger;

    public ErrorHandlingMiddleware(ILogger<ErrorHandlingMiddleware> logger)
    {
        _logger = logger;
    }

    public async Task InvokeAsync<T>(MessageContext context, T message, Func<Task> next)
    {
        try
        {
            await next();
        }
        catch (BusinessRuleException ex)
        {
            _logger.LogWarning("Business rule violation for {MessageType}: {Message}",
                typeof(T).Name, ex.Message);

            // Don't retry business rule violations
            return;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error processing {MessageType}", typeof(T).Name);
            throw; // Re-throw to trigger retry logic
        }
    }
}

Testing with Wolverine

Unit Testing Handlers

Test handlers independently:

[Test]
public async Task Handle_ValidCommand_ReturnsSuccess()
{
    // Arrange
    var command = new CreateCashierCommand(Guid.NewGuid(), "John Doe", "john@example.com");
    var mockMessaging = new Mock<IMessageBus>();

    mockMessaging
        .Setup(m => m.InvokeCommandAsync(It.IsAny<CreateCashierCommandHandler.DbCommand>(), It.IsAny<CancellationToken>()))
        .ReturnsAsync(new Data.Entities.Cashier { /* ... */ });

    // Act
    var (result, integrationEvent) = await CreateCashierCommandHandler.Handle(
        command, mockMessaging.Object, CancellationToken.None);

    // Assert
    result.IsSuccess.Should().BeTrue();
    integrationEvent.Should().NotBeNull();
}

Integration Testing

Test the full message pipeline:

[Test]
public async Task Should_Process_Command_End_To_End()
{
    // Arrange
    using var testContext = new IntegrationTestContext();
    var messageBus = testContext.GetService<IMessageBus>();

    var command = new CreateCashierCommand(Guid.NewGuid(), "Jane Doe", "jane@example.com");

    // Act
    var (result, integrationEvent) = await messageBus.InvokeAsync(command);

    // Assert
    result.IsSuccess.Should().BeTrue();

    // Verify database state
    var db = testContext.GetService<AppDomainDb>();
    var cashier = await db.Cashiers.FirstOrDefaultAsync(c => c.CashierId == result.Value.Id);
    cashier.Should().NotBeNull();

    // Verify event was published (integration test would check Kafka)
    integrationEvent.Should().NotBeNull();
}

Testing with TestContainers

Use TestContainers for full integration testing:

public class IntegrationTestContext : IDisposable
{
    private readonly IHost _host;
    private readonly PostgreSqlContainer _database;
    private readonly KafkaContainer _kafka;

    public IntegrationTestContext()
    {
        _database = new PostgreSqlBuilder().Build();
        _kafka = new KafkaBuilder().Build();

        // Start containers
        Task.WaitAll(
            _database.StartAsync(),
            _kafka.StartAsync()
        );

        // Configure test host
        var builder = Host.CreateDefaultBuilder()
            .ConfigureServices(services =>
            {
                services.AddWolverine(opts =>
                {
                    opts.PersistMessagesWithPostgresql(_database.GetConnectionString());
                    opts.UseKafka(_kafka.GetBootstrapAddress());
                });
            });

        _host = builder.Build();
        _host.Start();
    }

    public T GetService<T>() where T : notnull
    {
        return _host.Services.GetRequiredService<T>();
    }

    public void Dispose()
    {
        _host?.Dispose();
        _database?.DisposeAsync();
        _kafka?.DisposeAsync();
    }
}

Performance Optimization

Handler Performance

Optimize handler performance:

// Use static handlers to avoid allocation overhead
public static class OptimizedHandler
{
    public static async ValueTask<Result<T>> Handle(MyQuery query, AppDomainDb db, CancellationToken token)
    {
        // Use ValueTask for hot paths
        // Use compiled queries for performance
        var result = await db.Entities
            .Where(e => e.Id == query.Id)
            .FirstOrDefaultAsync(token);

        return result?.ToModel() ?? Result<T>.NotFound();
    }
}

Message Throughput

Configure for high throughput:

builder.Services.AddWolverine(opts =>
{
    // Increase parallelism
    opts.Policies.MaximumParallelization(Environment.ProcessorCount * 2);

    // Optimize local queues
    opts.LocalQueue("high-priority")
        .MaximumParallelization(10)
        .Sequential(); // For ordered processing

    // Configure batch processing
    opts.LocalQueue("batch-processing")
        .ProcessInBatches(50, TimeSpan.FromSeconds(5));
});

Memory Management

Optimize memory usage:

// Use object pools for frequently created objects
public static class PooledHandler
{
    private static readonly ObjectPool<StringBuilder> StringBuilderPool =
        new DefaultObjectPoolProvider().CreateStringBuilderPool();

    public static async Task<Result<string>> Handle(ProcessDataCommand command, CancellationToken token)
    {
        var sb = StringBuilderPool.Get();
        try
        {
            // Use pooled StringBuilder
            sb.AppendLine(command.Data);
            return sb.ToString();
        }
        finally
        {
            StringBuilderPool.Return(sb);
        }
    }
}

Command-Line Tools

Wolverine provides command-line tools for development and operations:

# Check environment and configuration
dotnet run -- check-env

# Generate code artifacts
dotnet run -- codegen

# Database operations
dotnet run -- db-apply      # Apply database migrations
dotnet run -- db-assert     # Verify database state
dotnet run -- db-patch      # Generate migration patch

# Inspect configuration
dotnet run -- describe      # Show current configuration
dotnet run -- resources     # List all resources

# Storage operations
dotnet run -- storage       # Storage-related commands

Usage in Development

These commands are automatically available when using app.RunAsync(args):

var app = builder.Build();

// This enables Wolverine CLI commands
await app.RunAsync(args);

Best Practices

Handler Design

1. Use static methods: Better performance, easier testing
2. Keep handlers focused: One handler per command/query
3. Avoid heavy dependencies: Inject only what you need
4. Use cancellation tokens: Support request cancellation

Error Handling

1. Distinguish error types: Separate business errors from technical errors
2. Use appropriate retry policies: Don't retry business rule violations
3. Log meaningfully: Provide context for debugging
4. Handle poison messages: Configure dead letter queues

Performance

1. Profile handlers: Monitor performance of critical handlers
2. Use appropriate data types: ValueTask for hot paths
3. Minimize allocations: Use object pooling where appropriate
4. Optimize database queries: Use compiled queries for frequently executed queries

Testing

1. Test handlers independently: Unit test business logic
2. Integration test pipelines: Test the full message flow
3. Mock external dependencies: Use mocks for external services
4. Test error scenarios: Verify error handling works correctly

Next Steps

• Learn about Integration Events publishing
• Understand Kafka Configuration for external messaging
• Explore Domain Events for internal messaging
• See Testing for comprehensive testing strategies