DEV Community

Cover image for Implementing the Outbox Pattern with SQLite and using Brighter
Rafael Andrade
Rafael Andrade

Posted on

Implementing the Outbox Pattern with SQLite and using Brighter

#outbox #brighter #sqlite #dotnet

Introduction

This guide demonstrates how to implement the Outbox Pattern with SQLite and .NET 8 using the Brighter library to ensure transactional consistency between database updates and message publishing.

Project

The goal is to process a CreateNewOrder command that publishes two events (OrderPlaced, OrderPaid) only if the transaction succeeds. If an error occurs (e.g., a business rule violation), both the database changes and message publications are rolled back.

Requirement

Messages

For this project, we will need these 3 messages: CreateNewOrder, OrderPlaced and OrderPaid

public class CreateNewOrder() : Command(Guid.NewGuid())
{
    public decimal Value { get; set; }
}

public class OrderPlaced() : Event(Guid.NewGuid())
{
    public string OrderId { get; set; } = string.Empty;
    public decimal Value { get; set; }
}

public class OrderPaid() : Event(Guid.NewGuid())
{
    public string OrderId { get; set; } = string.Empty;
}
Enter fullscreen mode Exit fullscreen mode

Message Mappers

Since only OrderPlaced and OrderPaid events are published to RabbitMQ, we need to implement mappers for them using JSON serialization

public class OrderPlacedMapper : IAmAMessageMapper<OrderPlaced>
{
    public Message MapToMessage(OrderPlaced request)
    {
        var header = new MessageHeader();
        header.Id = request.Id;
        header.TimeStamp = DateTime.UtcNow;
        header.Topic = "order-placed";
        header.MessageType = MessageType.MT_EVENT;
        header.ReplyTo = ""; // Due to a bug in SQLite implementation

        var body = new MessageBody(JsonSerializer.Serialize(request));
        return new Message(header, body);
    }

    public OrderPlaced MapToRequest(Message message)
    {
        return JsonSerializer.Deserialize<OrderPlaced>(message.Body.Bytes)!;
    }
}

public class OrderPaidMapper : IAmAMessageMapper<OrderPaid>
{
    public Message MapToMessage(OrderPaid request)
    {
        var header = new MessageHeader();
        header.Id = request.Id;
        header.TimeStamp = DateTime.UtcNow;
        header.Topic = "order-paid";
        header.MessageType = MessageType.MT_EVENT;
        header.ReplyTo = ""; // Due to a bug in SQLite implementation

        var body = new MessageBody(JsonSerializer.Serialize(request));
        return new Message(header, body);
    }

    public OrderPaid MapToRequest(Message message)
    {
        return JsonSerializer.Deserialize<OrderPaid>(message.Body.Bytes)!;
    }
}
Enter fullscreen mode Exit fullscreen mode

Notes: Due to a bug in the current implementation of outbox in SQLite, it's necessary to set the ReplyTo as empty; it'll be fixed in V10

Request Handlers

For OrderPlaced and OrderPaid we are going to log the received message.

public class OrderPlaceHandler(ILogger<OrderPlaceHandler> logger) : RequestHandlerAsync<OrderPlaced>
{
    public override Task<OrderPlaced> HandleAsync(OrderPlaced command, CancellationToken cancellationToken = default)
    {
        logger.LogInformation("{OrderId} placed with value {OrderValue}", command.OrderId, command.Value);
        return base.HandleAsync(command, cancellationToken);
    }
}

public class OrderPaidHandler(ILogger<OrderPaidHandler> logger) : RequestHandlerAsync<OrderPaid>
{
    public override Task<OrderPaid> HandleAsync(OrderPaid command, CancellationToken cancellationToken = default)
    {
        logger.LogInformation("{OrderId} paid", command.OrderId);
        return base.HandleAsync(command, cancellationToken);
    }
}
Enter fullscreen mode Exit fullscreen mode

Create New Order

The CreateNewOrder handler is going to wait for 10ms to emulate a process, then publish the OrderPlaced, if the value is mod 3 throw an exception (emulation a business error), otherwise publish OrderPaid.

public class CreateNewOrderHandler(IAmACommandProcessor commandProcessor,
    IUnitOfWork unitOfWork,
    ILogger<CreateNewOrderHandler> logger) : RequestHandlerAsync<CreateNewOrder>
{
    public override async Task<CreateNewOrder> HandleAsync(CreateNewOrder command, CancellationToken cancellationToken = default)
    {
        await unitOfWork.BeginTransactionAsync(cancellationToken);
        try
        {
            string id = Guid.NewGuid().ToString();
            logger.LogInformation("Creating a new order: {OrderId}", id);

            await Task.Delay(10, cancellationToken); // emulating an process

            _ = await commandProcessor.DepositPostAsync(new OrderPlaced { OrderId = id, Value = command.Value }, cancellationToken: cancellationToken);
            if (command.Value % 3 == 0)
            {
                throw new InvalidOperationException("invalid value");
            }

            _ = await commandProcessor.DepositPostAsync(new OrderPaid { OrderId = id }, cancellationToken: cancellationToken);

            await unitOfWork.CommitAsync(cancellationToken);
            return await base.HandleAsync(command, cancellationToken);
        }
        catch
        {
            logger.LogError("Invalid data");
            await unitOfWork.RollbackAsync(cancellationToken);
            throw;
        }
    }
}
Enter fullscreen mode Exit fullscreen mode

Key Insight:

  • DepositPostAsync stores messages in the outbox within the same transaction as business data.
  • If an exception occurs (e.g., InvalidOperationException), the transaction rolls back, ensuring no orphaned messages.

Configuring SQLite

To integrate the Outbox Pattern with SQLite, first ensure the outbox_messages table exists.

1. SQL Table Schema 

CREATE TABLE IF NOT EXISTS "outbox_messages"(
  [MessageId] TEXT NOT NULL COLLATE NOCASE,
  [Topic] TEXT NULL,
  [MessageType] TEXT NULL,
  [Timestamp] TEXT NULL,
  [CorrelationId] TEXT NULL,
  [ReplyTo] TEXT NULL,
  [ContentType] TEXT NULL,  
  [Dispatched] TEXT NULL,
  [HeaderBag] TEXT NULL,
  [Body] TEXT NULL,
  CONSTRAINT[PK_MessageId] PRIMARY KEY([MessageId])
);
Enter fullscreen mode Exit fullscreen mode

2. Dependency Injection Setup

Register the outbox and transaction.

services
    .AddServiceActivator(opt => { // Subscription setup (see previous article) })
    .UseSqlOutbox(new SqliteConfiguration(ConnectionString, "outbox_messages"), typeof(SqliteConnectionProvider), ServiceLifetime.Scoped))      
.UseSqliteTransactionConnectionProvider(typeof(SqliteConnectionProvider))
    .UseOutboxSweeper(opt => opt.BatchSize = 10);
Enter fullscreen mode Exit fullscreen mode

Why This Works:

  • UseSqliteOutbox links the outbox to Sqlite.
  • UseOutboxSweeper configures background polling for undelivered messages.

3. Transaction Management

To ensure atomicity between business logic and message publishing in Brighter, implement ISqliteTransactionConnectionProvider and IUnitOfWork for shared transaction context. This guarantees that messages are only stored in the outbox if the database transaction commits successfully.

a. SqliteConnectionProvider 
public class SqliteConnectionProvider(SqliteUnitOfWork sqlConnection) : ISqliteTransactionConnectionProvider
{
    public SqliteConnection GetConnection()
    {
        return sqlConnection.Connection;
    }

    public Task<SqliteConnection> GetConnectionAsync(CancellationToken cancellationToken = default)
    {
        return Task.FromResult(sqlConnection.Connection);
    }

    public SqliteTransaction? GetTransaction()
    {
        return sqlConnection.Transaction;
    }

    public bool HasOpenTransaction => sqlConnection.Transaction != null;
    public bool IsSharedConnection => true;
}
Enter fullscreen mode Exit fullscreen mode
b. Unit of work

And finally we need to create a new interface and implement an interface called IUnitOfWork

public interface IUnitOfWork
{
    Task BeginTransactionAsync(CancellationToken cancellationToken, IsolationLevel isolationLevel = IsolationLevel.Serializable);
    Task CommitAsync(CancellationToken cancellationToken);
    Task RollbackAsync(CancellationToken cancellationToken);
}
Enter fullscreen mode Exit fullscreen mode
c. MySqlUnitOfWork Implementation 
public class SqliteUnitOfWork : IUnitOfWork
{
    public SqliteUnitOfWork(SqliteConfiguration configuration)
    {
        Connection = new(configuration.ConnectionString);
        Connection.Open();
    }

    public SqliteConnection Connection { get; }
    public SqliteTransaction? Transaction { get; private set; }

    public async Task BeginTransactionAsync(IsolationLevel isolationLevel = IsolationLevel.Serializable)
    {
        if (Transaction == null)
        {
            Transaction = (SqliteTransaction)await Connection.BeginTransactionAsync(isolationLevel);
        }
    }

    public async Task CommitAsync(CancellationToken cancellationToken)
    {
        if (Transaction != null)
        {
            await Transaction.CommitAsync(cancellationToken);
        }
    }

    public async Task RollbackAsync(CancellationToken cancellationToken)
    {
        if (Transaction != null)
        {
            await Transaction.RollbackAsync(cancellationToken);
        }
    }
}
Enter fullscreen mode Exit fullscreen mode
d. Register Services in Dependency Injection 
services
    .AddScoped<SqliteUnitOfWork, SqliteUnitOfWork>()
    .TryAddScoped<IUnitOfWork>(provider => provider.GetRequiredService<SqliteUnitOfWork>());
Enter fullscreen mode Exit fullscreen mode

Conclusion

By implementing the Outbox Pattern with Brighter and SQLite, we’ve demonstrated how to achieve transactional consistency between database updates and message publishing. This approach ensures that:

  1. Messages are only published if the transaction commits successfully

    • Using DepositPostAsync, messages like OrderPlaced and OrderPaid are stored in the outbox_messages table within the same transaction as business data. If the handler fails (e.g., due to a simulated error), the transaction rolls back, and no orphaned messages are sent.
    • Brighter's ISqliteTransactionConnectionProvider guarantees that database updates and message deposits share the same transaction.

  2. Fault Tolerance via the Outbox Sweeper

    • The UseOutboxSweeper polls for undelivered messages and retries them until acknowledged by RabbitMQ. This decouples message publishing from the handler’s execution, ensuring reliability even if the broker is temporarily unavailable.

  3. Decoupled Architecture

    • Applications focus on local transactions, while Brighter handles message delivery asynchronously. This avoids tight coupling to the messaging infrastructure and simplifies scalability.

This implementation showcases how Brighter abstracts complexity, enabling developers to focus on business logic while ensuring reliability in distributed systems.

Reference

Top comments (0)