SimpleOutbox 1.6.0

SimpleOutbox

A simple, standalone Transactional Outbox library for .NET applications using Entity Framework Core and PostgreSQL.

Why SimpleOutbox?

When you need to update your database and publish a message/event, you face the dual-write problem. If the message publish fails after the database commit, you have inconsistent state. SimpleOutbox solves this by storing messages in the same database transaction as your domain changes, then reliably delivering them asynchronously.

Database Transaction
├── Update Order status = "Paid"
├── Insert OutboxMessage (OrderPaidEvent)  ← Same transaction
└── Commit

Background Service (async)
├── SELECT ... FOR UPDATE SKIP LOCKED
├── Deserialize & Process message
└── Mark as Completed

Features

Quick Start

1. Install

dotnet add package SimpleOutbox

2. Configure

// Program.cs
builder.Services.AddSimpleOutbox<MyDbContext>(options =>
{
    options.Schema = "outbox";
    options.MaxRetries = 5;
    options.PollingInterval = TimeSpan.FromSeconds(5);
    
    // Auto-discover handlers from assembly
    options.AddHandlersFromAssembly(typeof(Program).Assembly);
});

3. Apply EF Configuration

public class MyDbContext : DbContext
{
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.ApplyOutboxConfiguration();
    }
}

4. Use It

public class OrderService(MyDbContext db, IOutboxService outbox)
{
    public async Task CreateOrderAsync(Order order, CancellationToken ct)
    {
        db.Orders.Add(order);
        
        // Enqueue event in same transaction (use order.Id for idempotency)
        await outbox.EnqueueAsync(order.Id, new OrderCreatedEvent(order.Id), ct);
        
        // Both committed atomically
        await db.SaveChangesAsync(ct);
    }
}

5. Implement Handler

// Typed handler (recommended)
public class OrderCreatedHandler : IMessageHandler<OrderCreatedEvent>
{
    public async Task HandleAsync(OrderCreatedEvent message, CancellationToken ct)
    {
        // Process the order created event
        await SendWelcomeEmail(message.CustomerEmail, ct);
    }
}

// Or legacy handler for all message types
public class MyOutboxHandler(IMediator mediator) : IOutboxMessageHandler
{
    public async Task HandleAsync(object message, Type messageType, CancellationToken ct)
    {
        await mediator.Send(message, ct);
    }
}

Documentation

API Overview

IOutboxService

public interface IOutboxService
{
    // Enqueue for immediate processing (messageId for idempotency)
    Task EnqueueAsync<T>(Guid messageId, T message, CancellationToken ct = default);
    
    // Schedule for future processing (messageId for idempotency and cancellation)
    Task ScheduleAsync<T>(Guid messageId, T message, TimeSpan delay, CancellationToken ct = default);
    Task ScheduleAsync<T>(Guid messageId, T message, DateTime scheduledFor, CancellationToken ct = default);
    
    // Cancel a scheduled message
    Task<bool> CancelAsync(Guid messageId, CancellationToken ct = default);
}

IMessageHandler<T>

public class OrderCreatedHandler : IMessageHandler<OrderCreatedEvent>
{
    public Task HandleAsync(OrderCreatedEvent message, CancellationToken ct)
    {
        // Process typed message
    }
}

IPartitionedMessage

public record OrderEvent(Guid OrderId) : IPartitionedMessage
{
    public string PartitionKey => $"order:{OrderId}";
    public int MaxConcurrency => 1; // Strict FIFO
}

IMessageMiddleware

public class LoggingMiddleware(ILogger<LoggingMiddleware> logger) : IMessageMiddleware
{
    public async Task InvokeAsync(object message, Type messageType, Func<Task> next, CancellationToken ct)
    {
        logger.LogInformation("Processing {Type}", messageType.Name);
        await next();
        logger.LogInformation("Completed {Type}", messageType.Name);
    }
}

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Your Application                         │
│                                                              │
│   dbContext.Add(order);                                      │
│   await outbox.EnqueueAsync(messageId, new OrderCreatedEvent(...)); │
│   await dbContext.SaveChangesAsync();  // Atomic commit      │
└──────────────────────────┬──────────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────┐
│                  OutboxDeliveryService                       │
│                   (BackgroundService)                        │
│                                                              │
│   1. SELECT ... FOR UPDATE SKIP LOCKED                       │
│   2. Check partition concurrency limits                      │
│   3. Deserialize and execute pipeline:                       │
│      Middlewares -> IMessageHandler<T> (or legacy handler)   │
│   4. Notify observers (telemetry)                            │
│   5. Mark as Completed or schedule retry                     │
│   6. Wait for notification or polling interval               │
└─────────────────────────────────────────────────────────────┘

Database Schema

SimpleOutbox creates a single table in your configured schema (default: outbox):

CREATE TABLE outbox.outbox_messages (
    id              BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    message_id      UUID NOT NULL UNIQUE,
    message_type    VARCHAR(500) NOT NULL,
    payload         TEXT NOT NULL,
    partition_key   VARCHAR(200),
    max_concurrency INT DEFAULT 1,
    created_at      TIMESTAMP DEFAULT NOW(),
    scheduled_for   TIMESTAMP,
    processed_at    TIMESTAMP,
    retry_count     INT DEFAULT 0,
    last_error      TEXT,
    status          VARCHAR(20) DEFAULT 'Pending'
);

Indexes are automatically created for:

• Pending message lookup
• Partition FIFO ordering
• Concurrency control
• Cleanup queries
• Message cancellation

Requirements

• .NET 10.0+
• PostgreSQL 14+
• Entity Framework Core 9.0+

License

MIT License - see LICENSE for details.