|
| 1 | +# Pandatech MassTransit Outbox |
| 2 | + |
| 3 | +Outbox and inbox pattern implementation for [MassTransit](https://masstransit-project.com/) with **multiple DbContext |
| 4 | +support**. |
| 5 | + |
| 6 | +MassTransit's built-in outbox only supports a single `DbContext`. These packages let you reliably publish and consume |
| 7 | +messages across many modules, each with its own `DbContext` — designed for modular monolith and microservice |
| 8 | +architectures. |
| 9 | + |
| 10 | +| Package | Provider | Concurrency strategy | |
| 11 | +|----------------------------------------|------------|-----------------------------| |
| 12 | +| `Pandatech.MassTransit.PostgresOutbox` | PostgreSQL | `FOR UPDATE SKIP LOCKED` | |
| 13 | +| `Pandatech.MassTransit.SqliteOutbox` | SQLite | Lease-based (`LeasedUntil`) | |
| 14 | + |
| 15 | +Both packages are **wire-compatible** — a service using PostgreSQL for its outbox can publish to a service using SQLite |
| 16 | +for its inbox, and vice versa. |
| 17 | + |
| 18 | +## Features |
| 19 | + |
| 20 | +- **Multiple DbContext support** — each module gets its own `DbContext`, outbox, and inbox |
| 21 | +- **Outbox pattern** — messages are persisted atomically with your domain changes, then published by a background |
| 22 | + service |
| 23 | +- **Inbox pattern** — idempotent message consumption prevents duplicate processing |
| 24 | +- **Background cleanup** — processed messages are automatically removed after a configurable retention period |
| 25 | +- **Zero-allocation logging** — uses `[LoggerMessage]` source generators throughout |
| 26 | +- **Multi-TFM** — targets `net9.0`, and `net10.0` |
| 27 | + |
| 28 | +## Installation |
| 29 | + |
| 30 | +```bash |
| 31 | +# PostgreSQL |
| 32 | +dotnet add package Pandatech.MassTransit.PostgresOutbox |
| 33 | + |
| 34 | +# SQLite |
| 35 | +dotnet add package Pandatech.MassTransit.SqliteOutbox |
| 36 | +``` |
| 37 | + |
| 38 | +## Quick start |
| 39 | + |
| 40 | +The API surface is identical for both providers. Examples below use the PostgreSQL package — replace the namespace with |
| 41 | +`MassTransit.SQLiteOutbox` for SQLite. |
| 42 | + |
| 43 | +### 1. Configure your DbContext |
| 44 | + |
| 45 | +Implement `IOutboxDbContext`, `IInboxDbContext`, or both, and call `ConfigureInboxOutboxEntities` in `OnModelCreating`: |
| 46 | + |
| 47 | +```csharp |
| 48 | +using MassTransit.PostgresOutbox.Abstractions; |
| 49 | +using MassTransit.PostgresOutbox.Extensions; |
| 50 | + |
| 51 | +public class OrdersDbContext : DbContext, IOutboxDbContext, IInboxDbContext |
| 52 | +{ |
| 53 | + public DbSet<OutboxMessage> OutboxMessages { get; set; } |
| 54 | + public DbSet<InboxMessage> InboxMessages { get; set; } |
| 55 | + |
| 56 | + protected override void OnModelCreating(ModelBuilder modelBuilder) |
| 57 | + { |
| 58 | + modelBuilder.ConfigureInboxOutboxEntities(); |
| 59 | + } |
| 60 | +} |
| 61 | +``` |
| 62 | + |
| 63 | +**PostgreSQL only** — enable `UseQueryLocks()` for the `FOR UPDATE SKIP LOCKED` feature: |
| 64 | + |
| 65 | +```csharp |
| 66 | +builder.Services.AddDbContextPool<OrdersDbContext>(options => |
| 67 | + options.UseNpgsql(connectionString) |
| 68 | + .UseQueryLocks()); |
| 69 | +``` |
| 70 | + |
| 71 | +### 2. Register services |
| 72 | + |
| 73 | +```csharp |
| 74 | +using MassTransit.PostgresOutbox.Extensions; |
| 75 | + |
| 76 | +// Registers outbox publisher + outbox cleanup + inbox cleanup background services |
| 77 | +services.AddOutboxInboxServices<OrdersDbContext>(); |
| 78 | +``` |
| 79 | + |
| 80 | +To customize behavior, pass a `Settings` object: |
| 81 | + |
| 82 | +```csharp |
| 83 | +services.AddOutboxInboxServices<OrdersDbContext>(new Settings |
| 84 | +{ |
| 85 | + PublisherTimerPeriod = TimeSpan.FromSeconds(2), |
| 86 | + PublisherBatchCount = 50, |
| 87 | + OutboxRemovalBeforeInDays = 7, |
| 88 | + InboxRemovalBeforeInDays = 7 |
| 89 | +}); |
| 90 | +``` |
| 91 | + |
| 92 | +You can also register services individually: |
| 93 | + |
| 94 | +```csharp |
| 95 | +services.AddOutboxPublisherJob<OrdersDbContext>(); |
| 96 | +services.AddOutboxRemovalJob<OrdersDbContext>(); |
| 97 | +services.AddInboxRemovalJob<OrdersDbContext>(); |
| 98 | +``` |
| 99 | + |
| 100 | +> **SQLite only** — `Settings` has an additional `LeaseDuration` property (default: 5 minutes) that controls how long a |
| 101 | +> message is leased before becoming available for reprocessing after a crash. |
| 102 | +
|
| 103 | +### 3. Publish messages (outbox) |
| 104 | + |
| 105 | +Add your message to the outbox within the same `SaveChangesAsync` call as your domain changes: |
| 106 | + |
| 107 | +```csharp |
| 108 | +dbContext.Orders.Add(new Order |
| 109 | +{ |
| 110 | + Amount = 555, |
| 111 | + CreatedAt = DateTime.UtcNow |
| 112 | +}); |
| 113 | + |
| 114 | +dbContext.AddToOutbox(new OrderCreatedEvent { OrderId = orderId }); |
| 115 | + |
| 116 | +await dbContext.SaveChangesAsync(); |
| 117 | +``` |
| 118 | + |
| 119 | +To add multiple messages at once: |
| 120 | + |
| 121 | +```csharp |
| 122 | +dbContext.AddToOutboxRange(event1, event2, event3); |
| 123 | +await dbContext.SaveChangesAsync(); |
| 124 | +``` |
| 125 | + |
| 126 | +Both methods return the generated outbox message ID(s) for correlation if needed. |
| 127 | + |
| 128 | +The background publisher picks up new messages, publishes them via MassTransit, and marks them as done. |
| 129 | + |
| 130 | +### 4. Consume messages (inbox) |
| 131 | + |
| 132 | +Create a consumer that inherits from `InboxConsumer<TMessage, TDbContext>`: |
| 133 | + |
| 134 | +```csharp |
| 135 | +using MassTransit.PostgresOutbox.Abstractions; |
| 136 | +using Microsoft.EntityFrameworkCore.Storage; |
| 137 | + |
| 138 | +public class OrderCreatedConsumer(IServiceProvider sp) |
| 139 | + : InboxConsumer<OrderCreatedEvent, OrdersDbContext>(sp) |
| 140 | +{ |
| 141 | + protected override async Task ConsumeAsync( |
| 142 | + OrderCreatedEvent message, |
| 143 | + IDbContextTransaction transaction, |
| 144 | + CancellationToken ct) |
| 145 | + { |
| 146 | + // Your idempotent processing logic here. |
| 147 | + // The transaction is managed by InboxConsumer — just do your work. |
| 148 | + } |
| 149 | +} |
| 150 | +``` |
| 151 | + |
| 152 | +The base class handles deduplication (by `MessageId` + `ConsumerId`) and concurrency. In PostgreSQL this uses |
| 153 | +`FOR UPDATE SKIP LOCKED`; in SQLite it uses atomic lease acquisition. |
| 154 | + |
| 155 | +## How it works |
| 156 | + |
| 157 | +### Outbox flow |
| 158 | + |
| 159 | +Your code calls `AddToOutbox()` + `SaveChangesAsync()` → the message is persisted in the `OutboxMessages` table |
| 160 | +atomically with your domain changes → a background `HostedService` polls for new messages, publishes them via |
| 161 | +MassTransit, and marks them as done → a cleanup service deletes old processed messages. |
| 162 | + |
| 163 | +### Inbox flow |
| 164 | + |
| 165 | +MassTransit delivers a message to your `InboxConsumer` → the base class inserts or finds the `InboxMessage` row → |
| 166 | +acquires an exclusive lock (PostgreSQL) or lease (SQLite) → calls your `ConsumeAsync` method → marks the message as done |
| 167 | +and commits → if your code throws, the transaction rolls back and the message is retried. |
| 168 | + |
| 169 | +## Cross-provider compatibility |
| 170 | + |
| 171 | +Both packages serialize messages identically (`System.Text.Json`, same MassTransit header convention), so they are fully |
| 172 | +wire-compatible. A modular monolith can have some modules using PostgreSQL and others using SQLite — messages flow |
| 173 | +seamlessly between them via the shared message broker. |
| 174 | + |
| 175 | +## Settings reference |
| 176 | + |
| 177 | +| Property | Default | Description | |
| 178 | +|---------------------------------|-----------|-------------------------------------------------------| |
| 179 | +| `PublisherTimerPeriod` | 1 second | How often the publisher polls for new outbox messages | |
| 180 | +| `PublisherBatchCount` | 100 | Max messages published per tick | |
| 181 | +| `OutboxRemovalBeforeInDays` | 5 | Days to retain processed outbox messages | |
| 182 | +| `OutboxRemovalTimerPeriod` | 1 day | How often outbox cleanup runs | |
| 183 | +| `InboxRemovalBeforeInDays` | 5 | Days to retain processed inbox messages | |
| 184 | +| `InboxRemovalTimerPeriod` | 1 day | How often inbox cleanup runs | |
| 185 | +| `LeaseDuration` *(SQLite only)* | 5 minutes | How long a message lease is held | |
| 186 | + |
| 187 | +## License |
| 188 | + |
| 189 | +MIT |
0 commit comments