Excalibur.Outbox.CosmosDb 3.0.0-alpha.26

Excalibur.Outbox.CosmosDb

Azure Cosmos DB implementation of the cloud-native outbox pattern for reliable message delivery.

Features

• Transactional outbox pattern with partition-aware storage
• Change feed subscription for push-based message processing
• Cost-aware operations with RU tracking
• Automatic TTL cleanup for published messages
• Serverless-friendly - designed for Azure Functions triggers

Installation

dotnet add package Excalibur.Outbox.CosmosDb

Configuration

services.AddCosmosDbOutboxStore(options =>
{
    options.ConnectionString = "AccountEndpoint=...;AccountKey=...";
    options.DatabaseName = "mydb";
    options.ContainerName = "outbox";
    options.DefaultTimeToLiveSeconds = 604800; // 7 days
    options.CreateContainerIfNotExists = true;
});

Or via configuration:

services.AddCosmosDbOutboxStore(configuration.GetSection("CosmosDbOutbox"));

Usage

Adding Messages to the Outbox

public class MyService
{
    private readonly ICloudNativeOutboxStore _outbox;

    public MyService(ICloudNativeOutboxStore outbox)
    {
        _outbox = outbox;
    }

    public async Task PublishEventAsync(OrderPlaced @event, CancellationToken ct)
    {
        var message = new CloudOutboxMessage
        {
            MessageId = Guid.NewGuid().ToString(),
            MessageType = nameof(OrderPlaced),
            Payload = JsonSerializer.SerializeToUtf8Bytes(@event),
            AggregateId = @event.OrderId,
            AggregateType = "Order",
            PartitionKeyValue = @event.OrderId,
            CreatedAt = DateTimeOffset.UtcNow
        };

        var partitionKey = new PartitionKey(@event.OrderId);
        var result = await _outbox.AddAsync(message, partitionKey, ct);

        if (!result.Success)
        {
            // Handle failure
        }
    }
}

Processing Messages with Change Feed

// Subscribe to new outbox messages
var subscription = await _outbox.SubscribeToNewMessagesAsync(
    new ChangeFeedOptions { StartPosition = ChangeFeedStartPosition.Now },
    ct);

await foreach (var change in subscription.ReadChangesAsync(ct))
{
    if (change.Document != null)
    {
        // Publish to message broker
        await _messageBroker.PublishAsync(change.Document);

        // Mark as published
        await _outbox.MarkAsPublishedAsync(
            change.DocumentId,
            change.PartitionKey,
            ct);
    }
}

Using with Azure Functions

For serverless scenarios, use the Cosmos DB change feed trigger:

[FunctionName("OutboxProcessor")]
public async Task ProcessOutboxMessages(
    [CosmosDBTrigger(
        databaseName: "mydb",
        containerName: "outbox",
        Connection = "CosmosDbConnection",
        LeaseContainerName = "leases",
        CreateLeaseContainerIfNotExists = true)]
    IReadOnlyList<OutboxDocument> documents,
    CancellationToken ct)
{
    foreach (var doc in documents.Where(d => !d.IsPublished))
    {
        await _messageBroker.PublishAsync(doc);
        await _outbox.MarkAsPublishedAsync(doc.Id, new PartitionKey(doc.PartitionKey), ct);
    }
}

Document Schema

The outbox uses the following Cosmos DB document schema:

Requirements

• .NET 9.0+
• Azure Cosmos DB account or emulator
• Excalibur.Data.Abstractions for cloud-native interfaces
• Excalibur.Data.CosmosDb for Cosmos DB SDK setup

License

See LICENSE files in repository root.