MongoDbRepository.Extensions.Outbox.MessageBus 6.4.3

MongoDbRepository.Extensions.Outbox.MessageBus

Bridges your local, database-backed transactional outbox table smoothly with enterprise service brokers while introducing robust distributed idempotency controls.

🌟 Features

• Pluggable Event Dispatch Adapter: Implements IOutboxMessageHandler to intercept raw outbox storage records, safely de-serialize content strings back into concrete integration models, and publish them to an external message broker.
• Type-Safe Idempotency Ledger: Implements a database-backed deduplication layer mapped tightly to the core v6 repository structure using a native ObjectId identity primitive.
• Ecosystem Index Integration: Includes an index provider that hooks into the framework’s AddAutomatedIndexes() tool to build an atomic compound unique key constraint (MessageId + ConsumerName), supporting multiple independent consumer groups safely.

🚀 Quick Start

1. Configure the Broker Routing Layer (Program.cs)

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddMongoDbRepositoryCore(builder.Configuration.GetSection("MongoDb"));

// Bind the outbox string signatures directly to your concrete event contracts
builder.Services.AddMongoDbOutboxMessageBusDispatcher(registry =>
{
    registry.RegisterEvent<OrderCompletedIntegrationEvent>("OrderCompleted");
});

// Automatically provisions the unique ledger constraints out-of-band via assembly scanning
builder.Services.AddAutomatedIndexes();

2. Apply the Duplicate Delivery Guard inside a Broker Consumer

Inject the IIdempotentConsumerGuard to protect downstream execution blocks from network retry delivery duplications:

C#
using MassTransit;
using MongoDbRepository.Extensions.Outbox.MessageBus.Deduplication;

public sealed class OrderCompletedConsumer(IIdempotentConsumerGuard idempotencyGuard) 
    : IConsumer<OrderCompletedIntegrationEvent>
{
    public async Task Consume(ConsumeContext<OrderCompletedIntegrationEvent> context)
    {
        string messageId = context.MessageId?.ToString() ?? context.Message.OrderId.ToString();
        string consumerGroup = nameof(OrderCompletedConsumer);

        // Executes an atomic database-level index check. 
        // Returns false instantly if another container node or pod already handled this message.
        if (!await idempotencyGuard.TryAcquireProcessingLockAsync(messageId, consumerGroup, context.CancellationToken))
        {
            return; // Safe exit point for duplicate deliveries
        }

        // Proceed with your unique business modifications safely here...
    }
}