README.md

NetEvolve.Pulse.PostgreSql

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

Features

• Plain ADO.NET: No ORM overhead, direct PostgreSQL access via Npgsql
• Transaction Support: Enlist outbox operations in existing NpgsqlTransaction instances
• Optimized Queries: Uses stored functions with FOR UPDATE SKIP LOCKED 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.PostgreSql

.NET CLI

dotnet add package NetEvolve.Pulse.PostgreSql

PackageReference

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

Database Setup

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

Running the Scripts

The scripts in Scripts/ use psql meta-commands (\if, \set, \gexec), so run them with psql 10 or later. GUI query tools such as pgAdmin or DBeaver cannot execute them unchanged.

Pass the schema and table names as psql variables. Both are optional and default to pulse and the script's table name. The values are used as quoted identifiers, so they are case-sensitive and must match the Schema and TableName options you configure in the application.

psql -v ON_ERROR_STOP=1 -h your-host -d your-database -v schema_name=pulse -v table_name=OutboxMessage -f OutboxMessage.sql
psql -v ON_ERROR_STOP=1 -h your-host -d your-database -v schema_name=pulse -v table_name=IdempotencyKey -f IdempotencyKey.sql
psql -v ON_ERROR_STOP=1 -h your-host -d your-database -v schema_name=pulse -v table_name=CommandDeadLetter -f CommandDeadLetter.sql
psql -v ON_ERROR_STOP=1 -h your-host -d your-database -v schema_name=pulse -v table_name=AuditEntry -f AuditEntry.sql

The scripts are idempotent, and re-running them is the supported upgrade path. Primary keys and indexes are named PK_<schema>_<table> and IX_<schema>_<table>_<columns>. A name that would exceed PostgreSQL's 63-byte identifier limit becomes PK_<md5(schema.table)> or IX_<md5(schema.table)>_<columns> instead, because truncated names could collide and an index would be skipped silently. AuditEntry.sql and CommandDeadLetter.sql create no functions, so several of their tables can share one schema. OutboxMessage.sql and IdempotencyKey.sql create functions that are named per schema and bound to the last table_name the script ran with, so each outbox or idempotency table needs its own schema. OutboxMessage.sql renames the PK_<schema> and IX_<schema>_Status_* names that earlier versions produced instead of adding a second set of indexes.

Schema Script Contents

The script creates:

• The configured schema (default: pulse)
• The OutboxMessage table with two optimized partial indexes

Core functions:

Function	Purpose
get_pending_outbox_messages	Retrieves and locks pending messages for processing (FOR UPDATE SKIP LOCKED)
get_failed_outbox_messages_for_retry	Retrieves failed messages eligible for retry
mark_outbox_message_completed	Marks a message as successfully processed
mark_outbox_message_failed	Marks a message as failed with error details
mark_outbox_message_dead_letter	Moves a message to dead-letter status
delete_completed_outbox_messages	Removes old completed messages older than a given threshold

Management functions:

Function	Purpose
get_dead_letter_outbox_messages	Returns a paginated list of dead-letter messages
get_dead_letter_outbox_message	Returns a single dead-letter message by ID
get_dead_letter_outbox_message_count	Returns the total count of dead-letter messages
replay_outbox_message	Resets a single dead-letter message to Pending
replay_all_dead_letter_outbox_messages	Resets all dead-letter messages to Pending
get_outbox_statistics	Returns message counts grouped by status
get_outbox_messages	Returns a paginated, read-only list of messages in any status, optionally filtered by status
get_outbox_message	Returns a single message by ID, regardless of its status
dismiss_outbox_message	Permanently deletes a single dead-letter message

Upgrading

Every timestamp the outbox functions write (including UpdatedAt) is passed in from the application's injected TimeProvider instead of being taken from the database clock (NOW()). As a result, the signatures of get_pending_outbox_messages, mark_outbox_message_failed, mark_outbox_message_dead_letter, replay_outbox_message and replay_all_dead_letter_outbox_messages changed.

When upgrading from an earlier version, re-run OutboxMessage.sql against every database that hosts the outbox. The script is idempotent: it keeps the table and its data, drops the outdated function overloads and recreates the functions. Deploy the script together with the package upgrade, because the old and new package versions call different function signatures.

get_dead_letter_outbox_messages now orders dead letters with equal UpdatedAt by Id descending, so paging returns every dead letter exactly once. Re-run OutboxMessage.sql to apply the updated function; its signature is unchanged.

The idempotency store reserves keys through fn_reserve_idempotency_key (ON CONFLICT ... DO UPDATE ... WHERE), which also refreshes the created_at of a key that has outlived IdempotencyKeyOptions.TimeToLive. Re-run IdempotencyKey.sql together with the package upgrade; it keeps the table and its data and creates the new function.

Idempotency keys can be up to 450 characters long and are compared case-sensitively. See Idempotency Keys in the NetEvolve.Pulse README.

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)
    .AddPostgreSqlOutbox("Host=localhost;Database=MyDb;Username=postgres;Password=secret;")
);

Using Configuration

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

Registered Services

AddPostgreSqlOutbox(...) registers the following services:

Service	Implementation	Lifetime
IOutboxRepository	PostgreSqlOutboxRepository	Scoped
IOutboxManagement	PostgreSqlOutboxManagement	Scoped
TimeProvider	TimeProvider.System	Singleton (if not already registered)

Dead Letter Management

The IOutboxManagement service is automatically registered when calling AddPostgreSqlOutbox(...). 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.");
    }
}

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 NpgsqlConnection(_connectionString);
        await connection.OpenAsync(ct);
        await using var transaction = await connection.BeginTransactionAsync(ct);

        try
        {
            // Business operation
            await using var cmd = new NpgsqlCommand("INSERT INTO orders ...", connection, transaction);
            await cmd.ExecuteNonQueryAsync(ct);

            // Store event in outbox (same transaction)
            var outbox = new PostgreSqlEventOutbox(
                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;
        }
    }
}

Performance Considerations

Indexing

The default schema includes optimized partial indexes for:

• Pending message polling (Status, CreatedAt) where Status in (0, 3)
• Completed message cleanup (Status, ProcessedAt) where Status = 2

Stored Functions

Operations use stored functions with:

• FOR UPDATE SKIP LOCKED to skip locked rows during concurrent polling
• CTE-based atomic select-and-update for pending message retrieval

Requirements

• .NET 8.0, .NET 9.0, or .NET 10.0
• PostgreSQL 12 or later
• Npgsql 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.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.

---

Note

Made with ❤️ by the NetEvolve Team