Excalibur.Outbox.DynamoDb 3.0.0-alpha.208

Excalibur.Outbox.DynamoDb

AWS DynamoDB implementation of the cloud-native outbox pattern for reliable message delivery.

Features

• Transactional outbox pattern with partition-aware storage
• DynamoDB Streams for push-based message processing
• Capacity-aware operations with WCU/RCU tracking
• Automatic TTL cleanup for published messages
• Serverless-friendly - designed for AWS Lambda triggers

Installation

dotnet add package Excalibur.Outbox.DynamoDb

Configuration

services.AddDynamoDbOutboxStore(options =>
{
 options.Connection.Region = "us-east-1";
 options.TableName = "outbox";
 options.DefaultTimeToLiveSeconds = 604800; // 7 days
 options.CreateTableIfNotExists = true;
 options.EnableStreams = true;
});

Or via configuration:

services.AddDynamoDbOutboxStore(configuration.GetSection("DynamoDbOutbox"));

Local Development (DynamoDB Local)

services.AddDynamoDbOutboxStore(options =>
{
 options.Connection.ServiceUrl = "http://localhost:8000";
 options.TableName = "outbox";
 options.Connection.AccessKey = "local";
 options.Connection.SecretKey = "local";
});

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 DynamoDB Streams

// 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 AWS Lambda

For serverless scenarios, use DynamoDB Streams trigger:

public class OutboxProcessor
{
 private readonly IMessageBroker _messageBroker;
 private readonly ICloudNativeOutboxStore _outbox;

 public async Task ProcessStreamRecordsAsync(
 DynamoDBEvent dynamoEvent,
 CancellationToken ct)
 {
 foreach (var record in dynamoEvent.Records)
 {
 if (record.EventName != "INSERT")
 continue;

 var newImage = record.Dynamodb.NewImage;

 // Check if unpublished
 if (newImage.TryGetValue("isPublished", out var isPublished)
 && isPublished.BOOL)
 continue;

 var messageId = newImage["sk"].S;
 var partitionKey = newImage["pk"].S;
 var payload = Convert.FromBase64String(newImage["payload"].S);

 // Publish to message broker
 await _messageBroker.PublishAsync(payload);

 // Mark as published
 await _outbox.MarkAsPublishedAsync(
 messageId,
 new PartitionKey(partitionKey),
 ct);
 }
 }
}

Table Schema

The outbox uses the following DynamoDB table schema:

Streams Configuration

The table is created with streams enabled:

• StreamViewType: NEW_AND_OLD_IMAGES
• Required for change feed subscription

TTL Configuration

• TTL is enabled on the ttl attribute
• Only applied to published messages
• Default retention: 7 days (604800 seconds)

Requirements

• .NET 9.0+
• AWS DynamoDB or DynamoDB Local
• Excalibur.Data.Abstractions for cloud-native interfaces
• Excalibur.Data.DynamoDb for DynamoDB SDK setup

License

See LICENSE files in repository root.