NetEvolve.Pulse.EntityFramework 0.84.106

NetEvolve.Pulse.EntityFramework

Provider-agnostic Entity Framework Core persistence for the Pulse outbox pattern. Works with any EF Core database provider (SQL Server, PostgreSQL, SQLite, MySQL, etc.)—you bring your own provider and generate migrations. Features seamless transaction integration and uses the canonical outbox schema for interchangeability with ADO.NET providers.

Features

• Provider Agnostic: Works with SQL Server, PostgreSQL, SQLite, MySQL, and any EF Core provider
• Zero Provider Dependencies: Only depends on Microsoft.EntityFrameworkCore abstractions
• Transaction Integration: Automatic participation in DbContext transactions
• User-Generated Migrations: Full control over database migrations with your chosen provider
• Schema Interchangeability: Uses canonical schema compatible with ADO.NET providers
• LINQ Support: Full Entity Framework query capabilities for advanced scenarios

Installation

NuGet Package Manager

Install-Package NetEvolve.Pulse.EntityFramework

.NET CLI

dotnet add package NetEvolve.Pulse.EntityFramework

PackageReference

<PackageReference Include="NetEvolve.Pulse.EntityFramework" Version="x.x.x" />

Quick Start

1. Implement IOutboxDbContext

using Microsoft.EntityFrameworkCore;
using NetEvolve.Pulse;

public class ApplicationDbContext : DbContext, IOutboxDbContext
{
    public ApplicationDbContext(DbContextOptions<ApplicationDbContext> options)
        : base(options)
    {
    }

    // Your application entities
    public DbSet<Order> Orders => Set<Order>();
    public DbSet<Customer> Customers => Set<Customer>();

    // Required by IOutboxDbContext
    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);

        // The factory automatically selects the right configuration for the active provider
        modelBuilder.ApplyConfiguration(OutboxMessageConfigurationFactory.Create(this));

        // Or with custom options
        // modelBuilder.ApplyConfiguration(OutboxMessageConfigurationFactory.Create(this,
        //     Options.Create(new OutboxOptions { Schema = "myschema" })));
    }
}

2. Generate Migrations

# With your chosen provider (e.g., SQL Server)
dotnet add package Microsoft.EntityFrameworkCore.SqlServer

# Generate migration
dotnet ef migrations add AddOutbox --context ApplicationDbContext

# Apply migration
dotnet ef database update --context ApplicationDbContext

3. Register Services

using Microsoft.Extensions.DependencyInjection;
using NetEvolve.Pulse;

var services = new ServiceCollection();

// Register your DbContext with chosen provider
services.AddDbContext<ApplicationDbContext>(options =>
    options.UseSqlServer(connectionString));

// Register Pulse with Entity Framework outbox
services.AddPulse(config => config
    .AddOutbox(
        options => options.Schema = "pulse",
        processorOptions => processorOptions.BatchSize = 100)
    .AddEntityFrameworkOutbox<ApplicationDbContext>()
);

Transaction Integration

Events stored via IEventOutbox automatically participate in the current DbContext transaction:

public class OrderService
{
    private readonly ApplicationDbContext _context;
    private readonly IEventOutbox _outbox;

    public OrderService(ApplicationDbContext context, IEventOutbox outbox)
    {
        _context = context;
        _outbox = outbox;
    }

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        // Begin transaction
        await using var transaction = await _context.Database.BeginTransactionAsync(ct);

        try
        {
            // Business operation
            var order = new Order { CustomerId = request.CustomerId, Total = request.Total };
            _context.Orders.Add(order);
            await _context.SaveChangesAsync(ct);

            // Store event in outbox (same transaction)
            await _outbox.StoreAsync(new OrderCreatedEvent
            {
                Id = Guid.NewGuid().ToString(),
                OrderId = order.Id,
                CustomerId = order.CustomerId
            }, ct);

            // Commit both business data and event atomically
            await transaction.CommitAsync(ct);
        }
        catch
        {
            // Rollback discards both business data AND the outbox event
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

Event Handlers Sharing the DbContext

When you publish through IMediator.PublishAsync, the outbox writes through the caller's scoped DbContext. That is the same instance your scoped event handlers get. The mediator always runs the outbox handler first and on its own, so it never uses the context concurrently with one of your handlers.

• If two or more of your own handlers for the same event use the context, register UseEventDispatcherFor<MyEvent, SequentialEventDispatcher>() for that event (or UseDefaultEventDispatcher<SequentialEventDispatcher>() for all events). The default ParallelEventDispatcher runs them concurrently, which EF Core does not support.
• Storing an outbox message calls SaveChangesAsync on the shared context. This also saves any changes the caller has made but not yet saved. Wrap the business changes and PublishAsync in one transaction (see above) if they must commit or roll back together.

Multi-Provider Support

OutboxMessageConfigurationFactory.Create(this) automatically picks the right column types and index filter syntax for the active provider. Use it in OnModelCreating once and swap providers without touching the DbContext:

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    base.OnModelCreating(modelBuilder);
    modelBuilder.ApplyConfiguration(OutboxMessageConfigurationFactory.Create(this));
}

SQL Server

services.AddDbContext<ApplicationDbContext>(options =>
    options.UseSqlServer(connectionString));

PostgreSQL

services.AddDbContext<ApplicationDbContext>(options =>
    options.UseNpgsql(connectionString));

SQLite

services.AddDbContext<ApplicationDbContext>(options =>
    options.UseSqlite(connectionString));

MySQL

Supports both the Pomelo community provider and the Oracle official provider:

// Pomelo (recommended community provider)
services.AddDbContext<ApplicationDbContext>(options =>
    options.UseMySql(connectionString, ServerVersion.AutoDetect(connectionString)));

// Oracle MySQL provider
services.AddDbContext<ApplicationDbContext>(options =>
    options.UseMySQL(connectionString));

Other Providers

For providers not listed above, derive from OutboxMessageConfigurationBase directly and return the appropriate filter syntax from PendingMessagesFilter and CompletedMessagesFilter (return null for databases without partial-index support):

internal sealed class MyCustomOutboxMessageConfiguration : OutboxMessageConfigurationBase
{
    public MyCustomOutboxMessageConfiguration(IOptions<OutboxOptions> options)
        : base(options) { }

    // Return null if the database does not support filtered indexes
    protected override string? PendingMessagesFilter => null;
    protected override string? CompletedMessagesFilter => null;
}

// In OnModelCreating:
modelBuilder.ApplyConfiguration(new MyCustomOutboxMessageConfiguration(
    Options.Create(new OutboxOptions())));

Migration Examples

SQL Server Migration Output

// Generated migration snippet
migrationBuilder.EnsureSchema(name: "pulse");

migrationBuilder.CreateTable(
    name: "OutboxMessage",
    schema: "pulse",
    columns: table => new
    {
        Id = table.Column<Guid>(type: "uniqueidentifier", nullable: false),
        EventType = table.Column<string>(type: "nvarchar(500)", maxLength: 500, nullable: false),
        Payload = table.Column<string>(type: "nvarchar(max)", nullable: false),
        CorrelationId = table.Column<string>(type: "nvarchar(100)", maxLength: 100, nullable: true),
        CreatedAt = table.Column<DateTimeOffset>(type: "datetimeoffset", nullable: false),
        UpdatedAt = table.Column<DateTimeOffset>(type: "datetimeoffset", nullable: false),
        ProcessedAt = table.Column<DateTimeOffset>(type: "datetimeoffset", nullable: true),
        RetryCount = table.Column<int>(type: "int", nullable: false, defaultValue: 0),
        Error = table.Column<string>(type: "nvarchar(max)", nullable: true),
        Status = table.Column<int>(type: "int", nullable: false, defaultValue: 0)
    },
    constraints: table =>
    {
        table.PrimaryKey("PK_OutboxMessage", x => x.Id);
    });

migrationBuilder.CreateIndex(
    name: "IX_OutboxMessage_Status_CreatedAt",
    schema: "pulse",
    table: "OutboxMessage",
    columns: new[] { "Status", "CreatedAt" });

Schema Interchangeability

This provider produces the same table structure as the SQL Server ADO.NET provider, allowing you to:

1. Switch providers: Start with EF for development, move to ADO.NET for production
2. Mix providers: Use EF for writes (business transactions) and ADO.NET for reads (background processor)
3. Migrate: Change persistence strategy without data migration

// Both work with the same database table
services.AddPulse(config => config
    .AddOutbox()
    // Choose one:
    .AddEntityFrameworkOutbox<MyDbContext>()
    // or:
    // .AddSqlServerOutbox(connectionString)
);

Performance Considerations

For high-throughput scenarios, consider:

1. Batch processing: Configure OutboxProcessorOptions.BatchSize
2. ADO.NET for polling: Use SQL Server provider for background processor (explicit locking)
3. Separate DbContext: Dedicated context for outbox to avoid change tracker overhead

// High-performance configuration
services.AddPulse(config => config
    .AddOutbox(configureProcessorOptions: options =>
    {
        options.BatchSize = 500;
        options.PollingInterval = TimeSpan.FromMilliseconds(500);
        options.EnableBatchSending = true;
    })
    .AddEntityFrameworkOutbox<ApplicationDbContext>()
);

Idempotency Key Length and Case Sensitivity

• Keys can be up to 450 characters long (IdempotencyKeySchema.MaxLengths.IdempotencyKey). The store and the repository reject a longer key with an ArgumentOutOfRangeException before it touches the database.
• On SQL Server the key column is nvarchar(450) with the collation Latin1_General_100_BIN2, because a clustered index key is limited to 900 bytes. On MySQL it is varchar(500) with the collation utf8mb4_bin. Keys are case-sensitive on every provider.

Existing SQL Server and MySQL databases created with an earlier release need a new migration for these column changes (dotnet ef migrations add IdempotencyKeyExactComparison). Keys longer than 450 characters could never be stored in the old SQL Server column, so no key is truncated. The PostgreSQL (character varying(500)) and SQLite (TEXT) models are unchanged and need no migration.

Processing Lease Reclaim

A message claimed by GetPendingAsync stays in Processing until it is completed or failed. If a worker crashes or shuts down in between, the next pending poll reclaims the message once its UpdatedAt is older than OutboxOptions.ProcessingLeaseTimeout (default: 5 minutes, must be greater than zero).

services.AddPulse(config => config
    .AddEntityFrameworkOutbox<ApplicationDbContext>(options =>
        options.ProcessingLeaseTimeout = TimeSpan.FromMinutes(10))
);

Choose a value well above the longest expected dispatch. A dispatch that runs longer than the lease can be reclaimed by another poller and delivered twice.

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• Entity Framework Core 8.0+ with your chosen database provider
• Microsoft.Extensions.DependencyInjection for service registration
• Microsoft.Extensions.Hosting for the background processor

Related Packages

• NetEvolve.Pulse - Core mediator and outbox abstractions
• NetEvolve.Pulse.Dapr - Dapr pub/sub integration for event dispatch
• NetEvolve.Pulse.Extensibility - Core contracts and abstractions
• NetEvolve.Pulse.SqlServer - SQL Server ADO.NET persistence
• NetEvolve.Pulse.Polly - Polly v8 resilience policies integration

Documentation

For complete documentation, please visit the official documentation.

Contributing

Contributions are welcome! Please read the Contributing Guidelines before submitting a pull request.

Support

• Issues: Report bugs or request features on GitHub Issues
• Documentation: Read the full documentation at https://github.com/dailydevops/pulse

License

This project is licensed under the MIT License - see the LICENSE file for details.