NetEvolve.Pulse.SqlServer 0.17.3

NetEvolve.Pulse.SqlServer

SQL Server persistence provider for the Pulse outbox pattern using plain ADO.NET. Provides optimized T-SQL operations with proper transaction support and locking strategies for reliable event delivery in high-throughput scenarios.

Features

• Plain ADO.NET: No ORM overhead, direct SQL Server access via Microsoft.Data.SqlClient
• Transaction Support: Enlist outbox operations in existing SqlTransaction instances
• Optimized Queries: Uses stored procedures with ROWLOCK/READPAST hints for concurrent access
• Dead Letter Management: Built-in support for inspecting, replaying, and monitoring dead-letter messages via IOutboxManagement
• Configurable Schema: Customize schema and table names for multi-tenant scenarios
• Schema Interchangeability: Uses canonical schema compatible with Entity Framework provider

Installation

NuGet Package Manager

Install-Package NetEvolve.Pulse.SqlServer

.NET CLI

dotnet add package NetEvolve.Pulse.SqlServer

PackageReference

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

Database Setup

Before using this provider, execute the schema script to create the required database objects.

Running the Script

sqlcmd utility:

sqlcmd -S your-server -d your-database -i OutboxMessage.sql

SQL Server Management Studio (SSMS):

Enable SQLCMD Mode via Query > SQLCMD Mode (Ctrl+Shift+Q), then execute the script.

Azure Data Studio:

Enable SQLCMD via the query toolbar before executing.

SQLCMD Variables

The script exposes the following configurable variables at the top of OutboxMessage.sql:

To use custom names, change the :setvar values before executing:

:setvar SchemaName "myapp"
:setvar TableName "Events"

Schema Script Contents

The script creates:

• The configured schema (default: [pulse])
• The [OutboxMessage] table with two optimized non-clustered indexes

Core stored procedures:

Management stored procedures:

Quick Start

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

var services = new ServiceCollection();

services.AddPulse(config => config
    .AddOutbox(
        options => options.Schema = "pulse",
        processorOptions => processorOptions.BatchSize = 100)
    .AddSqlServerOutbox("Server=.;Database=MyDb;Integrated Security=true;TrustServerCertificate=true;")
);

Using Configuration

services.AddPulse(config => config
    .AddOutbox()
    .AddSqlServerOutbox(
        sp => sp.GetRequiredService<IConfiguration>().GetConnectionString("Outbox")!,
        options =>
        {
            options.Schema = "messaging";
            options.TableName = "OutboxMessage";
        })
);

Registered Services

AddSqlServerOutbox(...) registers the following services:

Dead Letter Management

The IOutboxManagement service is automatically registered when calling AddSqlServerOutbox(...). It provides operations for inspecting and recovering dead-letter messages, as well as monitoring outbox health.

public class OutboxMonitorService
{
    private readonly IOutboxManagement _management;

    public OutboxMonitorService(IOutboxManagement management) =>
        _management = management;

    public async Task PrintStatisticsAsync(CancellationToken ct)
    {
        var stats = await _management.GetStatisticsAsync(ct);
        Console.WriteLine($"Pending: {stats.Pending}");
        Console.WriteLine($"Processing: {stats.Processing}");
        Console.WriteLine($"Completed: {stats.Completed}");
        Console.WriteLine($"Failed: {stats.Failed}");
        Console.WriteLine($"Dead Letter: {stats.DeadLetter}");
        Console.WriteLine($"Total: {stats.Total}");
    }

    public async Task ReplayAllDeadLettersAsync(CancellationToken ct)
    {
        var replayed = await _management.ReplayAllDeadLetterAsync(ct);
        Console.WriteLine($"Replayed {replayed} dead-letter messages.");
    }
}

Available Operations

Transaction Integration

Manual Transaction Enlistment

public class OrderService
{
    private readonly string _connectionString;
    private readonly IServiceProvider _serviceProvider;

    public async Task CreateOrderAsync(CreateOrderRequest request, CancellationToken ct)
    {
        await using var connection = new SqlConnection(_connectionString);
        await connection.OpenAsync(ct);
        await using var transaction = connection.BeginTransaction();

        try
        {
            // Business operation
            await using var cmd = new SqlCommand("INSERT INTO [Order] ...", connection, transaction);
            await cmd.ExecuteNonQueryAsync(ct);

            // Store event in outbox (same transaction)
            var outbox = new SqlServerEventOutbox(
                connection,
                _serviceProvider.GetRequiredService<IOptions<OutboxOptions>>(),
                TimeProvider.System,
                transaction);

            await outbox.StoreAsync(new OrderCreatedEvent { OrderId = orderId }, ct);

            await transaction.CommitAsync(ct);
        }
        catch
        {
            await transaction.RollbackAsync(ct);
            throw;
        }
    }
}

Using IOutboxTransactionScope

public class UnitOfWork : IOutboxTransactionScope, IAsyncDisposable
{
    private readonly SqlConnection _connection;
    private SqlTransaction? _transaction;

    public async Task BeginTransactionAsync(CancellationToken ct)
    {
        await _connection.OpenAsync(ct);
        _transaction = _connection.BeginTransaction();
    }

    public object? GetCurrentTransaction() => _transaction;

    public async Task CommitAsync(CancellationToken ct)
    {
        if (_transaction is not null)
        {
            await _transaction.CommitAsync(ct);
        }
    }
}

// Register in DI
services.AddScoped<IOutboxTransactionScope, UnitOfWork>();

Schema Customization

To use a custom schema or table name, update the :setvar variables at the top of OutboxMessage.sql before executing, then configure the same names in code:

:setvar SchemaName "myapp"
:setvar TableName "Events"

services.AddPulse(config => config
    .AddOutbox(options =>
    {
        options.Schema = "myapp";      // Default: "pulse"
        options.TableName = "Events";  // Default: "OutboxMessage"
    })
    .AddSqlServerOutbox(connectionString)
);

Performance Considerations

Indexing

The default schema includes optimized indexes for:

• Pending message polling (Status, CreatedAt)
• Completed message cleanup (Status, ProcessedAt)

Stored Procedures

Operations use stored procedures with:

• ROWLOCK for row-level locking
• READPAST to skip locked rows during polling
• SET NOCOUNT ON to reduce network traffic

Batch Processing

Configure batch size based on your throughput requirements:

.AddOutbox(processorOptions: options =>
{
    options.BatchSize = 500;                    // Messages per poll
    options.PollingInterval = TimeSpan.FromSeconds(1);
    options.EnableBatchSending = true;          // Use batch transport
})

Schema Interchangeability

This provider uses the canonical outbox schema, making it fully interchangeable with the Entity Framework provider:

1. Development: Start with Entity Framework for rapid iteration
2. Production: Switch to ADO.NET for maximum performance
3. Mixed: Use both providers against the same database

// Both configurations work with the same database table
.AddSqlServerOutbox(connectionString)
// or
.AddEntityFrameworkOutbox<MyDbContext>()

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• SQL Server 2016 or later (or Azure SQL Database)
• Microsoft.Data.SqlClient for database connectivity
• 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.EntityFramework - Entity Framework 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.