Add SQL and PostgreSQL transports

Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>

Author: zarusz

.github/workflows/build.yml

@@ -160,15 +160,28 @@ jobs:
             docker_services: "mqtt"
 
           # NATS connects to localhost:4222 via the nats_endpoint env var
-          - transport: NATS
-            name: NATS
-            filter: "Category=Integration&Transport=NATS"
-            docker_services: "nats"
-
-          # Outbox – four parallel sub-legs:
-          # 1a. Sql.DbContext OutboxTests – AzureSB + Kafka; SQL Server via local docker
-          - transport: Outbox.SqlDbContext.Tests
-            name: "Outbox (SqlDbContext Tests)"
+          - transport: NATS
+            name: NATS
+            filter: "Category=Integration&Transport=NATS"
+            docker_services: "nats"
+
+          # SQL transports use TestContainers directly
+          - transport: Sql
+            name: SQL Server
+            filter: "Category=Integration&Transport=Sql"
+            test_path: "Tests/SlimMessageBus.Host.Sql.Test"
+            docker_services: ""
+
+          - transport: PostgreSql
+            name: PostgreSQL
+            filter: "Category=Integration&Transport=PostgreSql"
+            test_path: "Tests/SlimMessageBus.Host.PostgreSql.Test"
+            docker_services: ""
+
+          # Outbox – four parallel sub-legs:
+          # 1a. Sql.DbContext OutboxTests – AzureSB + Kafka; SQL Server via local docker
+          - transport: Outbox.SqlDbContext.Tests
+            name: "Outbox (SqlDbContext Tests)"
             filter: "Category=Integration&ClassName=SlimMessageBus.Host.Outbox.Sql.DbContext.Test.OutboxTests"
             test_path: "Tests/SlimMessageBus.Host.Outbox.Sql.DbContext.Test"
             docker_services: "sqldb zookeeper kafka kafka-init"

build/tasks.ps1

@@ -34,6 +34,7 @@ $projects = @(
 	"SlimMessageBus.Host.RabbitMQ",
 	"SlimMessageBus.Host.Sql",
 	"SlimMessageBus.Host.Sql.Common",
+	"SlimMessageBus.Host.PostgreSql",
 	"SlimMessageBus.Host.Nats",
 	"SlimMessageBus.Host.AmazonSQS",
 

docs/provider_postgresql.md

@@ -0,0 +1,51 @@
+# PostgreSQL transport provider for SlimMessageBus <!-- omit in toc -->
+
+Please read the [Introduction](intro.md) before reading this provider documentation.
+
+- [About](#about)
+- [Configuration](#configuration)
+- [How it works](#how-it-works)
+
+## About
+
+The PostgreSQL transport provider allows a shared PostgreSQL database to act as the message broker for collaborating producers and consumers.
+
+This transport is useful for applications that already operate PostgreSQL and do not need a dedicated messaging broker yet.
+
+## Configuration
+
+The configuration is arranged via the `.WithProviderPostgreSql(cfg => {})` method on the message bus builder.
+
+```cs
+services.AddSlimMessageBus(mbb =>
+{
+    mbb.WithProviderPostgreSql(cfg =>
+    {
+       cfg.ConnectionString = "...";
+       cfg.DatabaseSchemaName = "smb";
+       cfg.DatabaseTableName = "messages";
+       cfg.PollDelay = TimeSpan.FromMilliseconds(250);
+       cfg.PollBatchSize = 10;
+    });
+
+    mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue"));
+    mbb.Consume<PingMessage>(x => x.Queue("ping-queue"));
+
+    mbb.Produce<OrderSubmitted>(x => x.DefaultTopic("orders").ToTopic());
+    mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "billing"));
+    mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "shipping"));
+
+    mbb.AddServicesFromAssemblyContaining<PingConsumer>();
+    mbb.AddJsonSerializer();
+});
+```
+
+## How it works
+
+- A messages table stores exchanged messages.
+- A subscriptions table stores durable topic subscriptions configured by consumers.
+- Queue consumers compete for rows using `FOR UPDATE SKIP LOCKED`.
+- Topic publishes create one row per configured subscription.
+- Message rows use a `bigserial` physical key for insert locality and a logical `uuid` message id.
+- The default client-side id generator is sequential-ish for index locality. Random database ids can be selected through `cfg.IdGeneration`.
+- Producers optionally call `pg_notify` after inserting messages. Polling remains the correctness mechanism.

docs/provider_sql.md

@@ -17,42 +17,52 @@ When the application grows over time, and given that SMB is an abstraction, the
 
 ## SQL Compatibility
 
-This transport has been tested on SQL Azure (T-SQL), and should work on most other databases.
-If you see an issue, please raise an github issue.
+This transport targets SQL Server / Azure SQL (T-SQL).
 
 ## Configuration
 
-ToDo: Finish
-
-The configuration is arranged via the `.WithProviderSql(cfg => {})` method on the message bus builder.
-
-```cs
-services.AddSlimMessageBus(mbb =>
-{
-    mbb.WithProviderSql(cfg =>
-    {
-       // ToDo
-    });
-
-    mbb.AddServicesFromAssemblyContaining<PingConsumer>();
-    mbb.AddJsonSerializer();
-});
-```
+The configuration is arranged via the `.WithProviderSql(cfg => {})` method on the message bus builder.
+
+```cs
+services.AddSlimMessageBus(mbb =>
+{
+    mbb.WithProviderSql(cfg =>
+    {
+       cfg.ConnectionString = "...";
+       cfg.DatabaseSchemaName = "smb";
+       cfg.DatabaseTableName = "Messages";
+       cfg.PollDelay = TimeSpan.FromMilliseconds(250);
+       cfg.PollBatchSize = 10;
+    });
+
+    mbb.Produce<PingMessage>(x => x.DefaultQueue("ping-queue"));
+    mbb.Consume<PingMessage>(x => x.Queue("ping-queue"));
+
+    mbb.Produce<OrderSubmitted>(x => x.DefaultTopic("orders").ToTopic());
+    mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "billing"));
+    mbb.Consume<OrderSubmitted>(x => x.Topic("orders", "shipping"));
+
+    mbb.AddServicesFromAssemblyContaining<PingConsumer>();
+    mbb.AddJsonSerializer();
+});
+```
 
 ## How it works
 
 The same SQL database instance is required for all the producers and consumers to collaborate.
 Therefore ensure all of the service instances point to the same database cluster.
 
-- Single table is used to store all the exchanged messages (by default table is called `Messages`).
-- Producers send messages to the messages table.
-  - There are two types of entities (queues, and topics for pub/sub).
-  - In the case of a topic:
-    - Each subscription gets a copy of the message.
-    - Subscription has a lifetime, and can expire after certain idle time. Along with it, all the messages placed on the subscription.
-- Consumers (queue consumers, or subscribers in pub/sub) long poll the table to pick up their respective message.
-  - Queue consumers compete for the message, and ensure only one consumer instance is processing the message.
-  - Topic subscribers complete for the message within the same subscription.
-- In the future we might consider:
-  - Table per each entity, so that we can minimize table locking.
-  - Sessions to ensure order of processing within the same message session ID - similar to how Azure Service Bus feature or Apache Kafka topic-partition works.
+- A messages table is used to store exchanged messages (by default table is called `Messages`).
+- A subscriptions table is used to store durable topic subscriptions configured by consumers.
+- Producers send messages to the messages table.
+  - There are two types of entities (queues, and topics for pub/sub).
+  - In the case of a topic:
+    - Each configured durable subscription gets a copy of the message.
+- Consumers (queue consumers, or subscribers in pub/sub) long poll the table to pick up their respective message.
+  - Queue consumers compete for the message, and ensure only one consumer instance is processing the message.
+  - Topic subscribers compete for the message within the same subscription.
+- Message rows use a clustered `bigint identity` sequence for insert locality and a logical `uniqueidentifier` message id.
+- The default client-side id generator is sequential-ish for SQL Server index locality. Random database ids and database-generated sequential ids can be selected through `cfg.IdGeneration`.
+- In the future we might consider:
+  - Table per each entity, so that we can minimize table locking.
+  - Sessions to ensure order of processing within the same message session ID - similar to how Azure Service Bus feature or Apache Kafka topic-partition works.

src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlConsumerBuilderExtensions.cs

@@ -0,0 +1,42 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public static class PostgreSqlConsumerBuilderExtensions
+{
+    internal const string PropertySubscriptionName = "PostgreSql_SubscriptionName";
+
+    public static ConsumerBuilder<T> Queue<T>(this ConsumerBuilder<T> builder, string queue)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (queue is null) throw new ArgumentNullException(nameof(queue));
+
+        builder.Path(queue);
+        builder.ConsumerSettings.PathKind = PathKind.Queue;
+        return builder;
+    }
+
+    public static ConsumerBuilder<T> Topic<T>(this ConsumerBuilder<T> builder, string topic, string subscriptionName)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (topic is null) throw new ArgumentNullException(nameof(topic));
+        if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
+
+        builder.Topic(topic);
+        builder.ConsumerSettings.PathKind = PathKind.Topic;
+        builder.ConsumerSettings.Properties[PropertySubscriptionName] = subscriptionName;
+        return builder;
+    }
+
+    public static ConsumerBuilder<T> Subscription<T>(this ConsumerBuilder<T> builder, string subscriptionName)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
+
+        builder.ConsumerSettings.Properties[PropertySubscriptionName] = subscriptionName;
+        return builder;
+    }
+
+    internal static string GetSubscriptionName(this AbstractConsumerSettings settings)
+        => settings.Properties.TryGetValue(PropertySubscriptionName, out var value)
+            ? (string)value
+            : settings.Path;
+}

src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlHandlerBuilderExtensions.cs

@@ -0,0 +1,48 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public static class PostgreSqlHandlerBuilderExtensions
+{
+    public static HandlerBuilder<TRequest, TResponse> Queue<TRequest, TResponse>(this HandlerBuilder<TRequest, TResponse> builder, string queue)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (queue is null) throw new ArgumentNullException(nameof(queue));
+
+        builder.Path(queue);
+        builder.ConsumerSettings.PathKind = PathKind.Queue;
+        return builder;
+    }
+
+    public static HandlerBuilder<TRequest> Queue<TRequest>(this HandlerBuilder<TRequest> builder, string queue)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (queue is null) throw new ArgumentNullException(nameof(queue));
+
+        builder.Path(queue);
+        builder.ConsumerSettings.PathKind = PathKind.Queue;
+        return builder;
+    }
+
+    public static HandlerBuilder<TRequest, TResponse> Topic<TRequest, TResponse>(this HandlerBuilder<TRequest, TResponse> builder, string topic, string subscriptionName)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (topic is null) throw new ArgumentNullException(nameof(topic));
+        if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
+
+        builder.Path(topic);
+        builder.ConsumerSettings.PathKind = PathKind.Topic;
+        builder.ConsumerSettings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
+        return builder;
+    }
+
+    public static HandlerBuilder<TRequest> Topic<TRequest>(this HandlerBuilder<TRequest> builder, string topic, string subscriptionName)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (topic is null) throw new ArgumentNullException(nameof(topic));
+        if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
+
+        builder.Path(topic);
+        builder.ConsumerSettings.PathKind = PathKind.Topic;
+        builder.ConsumerSettings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
+        return builder;
+    }
+}

src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlProducerBuilderExtensions.cs

@@ -0,0 +1,30 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public static class PostgreSqlProducerBuilderExtensions
+{
+    public static ProducerBuilder<T> DefaultQueue<T>(this ProducerBuilder<T> producerBuilder, string queue)
+    {
+        if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
+        if (queue is null) throw new ArgumentNullException(nameof(queue));
+
+        producerBuilder.DefaultTopic(queue);
+        producerBuilder.ToQueue();
+        return producerBuilder;
+    }
+
+    public static ProducerBuilder<T> ToTopic<T>(this ProducerBuilder<T> producerBuilder)
+    {
+        if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
+
+        producerBuilder.Settings.PathKind = PathKind.Topic;
+        return producerBuilder;
+    }
+
+    public static ProducerBuilder<T> ToQueue<T>(this ProducerBuilder<T> producerBuilder)
+    {
+        if (producerBuilder is null) throw new ArgumentNullException(nameof(producerBuilder));
+
+        producerBuilder.Settings.PathKind = PathKind.Queue;
+        return producerBuilder;
+    }
+}

src/SlimMessageBus.Host.PostgreSql/Config/PostgreSqlRequestResponseBuilderExtensions.cs

@@ -0,0 +1,26 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public static class PostgreSqlRequestResponseBuilderExtensions
+{
+    public static RequestResponseBuilder ReplyToQueue(this RequestResponseBuilder builder, string queue)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (queue is null) throw new ArgumentNullException(nameof(queue));
+
+        builder.Settings.Path = queue;
+        builder.Settings.PathKind = PathKind.Queue;
+        return builder;
+    }
+
+    public static RequestResponseBuilder ReplyToTopic(this RequestResponseBuilder builder, string topic, string subscriptionName)
+    {
+        if (builder is null) throw new ArgumentNullException(nameof(builder));
+        if (topic is null) throw new ArgumentNullException(nameof(topic));
+        if (subscriptionName is null) throw new ArgumentNullException(nameof(subscriptionName));
+
+        builder.Settings.Path = topic;
+        builder.Settings.PathKind = PathKind.Topic;
+        builder.Settings.Properties[PostgreSqlConsumerBuilderExtensions.PropertySubscriptionName] = subscriptionName;
+        return builder;
+    }
+}

src/SlimMessageBus.Host.PostgreSql/Configuration/PostgreSqlMessageBusBuilderExtensions.cs

@@ -0,0 +1,25 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public static class PostgreSqlMessageBusBuilderExtensions
+{
+    public static MessageBusBuilder WithProviderPostgreSql(this MessageBusBuilder mbb, Action<PostgreSqlMessageBusSettings> configure)
+    {
+        if (mbb == null) throw new ArgumentNullException(nameof(mbb));
+        if (configure == null) throw new ArgumentNullException(nameof(configure));
+
+        var providerSettings = new PostgreSqlMessageBusSettings();
+        configure(providerSettings);
+
+        mbb.PostConfigurationActions.Add(services =>
+        {
+            services.TryAddSingleton(providerSettings);
+            services.TryAddScoped(_ => new NpgsqlConnection(providerSettings.ConnectionString));
+            services.TryAddSingleton<PostgreSqlSequentialGuidGenerator>();
+            services.TryAddScoped<PostgreSqlRepository>();
+            services.Replace(ServiceDescriptor.Scoped<IPostgreSqlRepository>(svp => svp.GetRequiredService<PostgreSqlRepository>()));
+            services.TryAddSingleton<PostgreSqlTemplate>();
+        });
+
+        return mbb.WithProvider(settings => new PostgreSqlMessageBus(settings, providerSettings));
+    }
+}

src/SlimMessageBus.Host.PostgreSql/Configuration/PostgreSqlMessageBusSettings.cs

@@ -0,0 +1,38 @@
+namespace SlimMessageBus.Host.PostgreSql;
+
+public class PostgreSqlMessageBusSettings
+{
+    public string? ConnectionString { get; set; }
+    public string DatabaseSchemaName { get; set; } = "public";
+    public string DatabaseTableName { get; set; } = "messages";
+    public string DatabaseMigrationsTableName { get; set; } = "__EFMigrationsHistory";
+    public TimeSpan? CommandTimeout { get; set; }
+    public IsolationLevel TransactionIsolationLevel { get; set; } = IsolationLevel.ReadCommitted;
+    public TimeSpan PollDelay { get; set; } = TimeSpan.FromMilliseconds(250);
+    public TimeSpan LockDuration { get; set; } = TimeSpan.FromSeconds(30);
+    public int PollBatchSize { get; set; } = 10;
+    public int MaxDeliveryAttempts { get; set; } = 10;
+    public bool NotifyOnPublish { get; set; } = true;
+    public PostgreSqlMessageIdGenerationSettings IdGeneration { get; set; } = new();
+
+    public RetrySettings SchemaCreationRetry { get; set; } = new()
+    {
+        RetryCount = 3,
+        RetryIntervalFactor = 1.2f,
+        RetryInterval = TimeSpan.FromSeconds(2),
+    };
+
+    public RetrySettings OperationRetry { get; set; } = new()
+    {
+        RetryCount = 5,
+        RetryIntervalFactor = 1.5f,
+        RetryInterval = TimeSpan.FromSeconds(2),
+    };
+}
+
+public class RetrySettings
+{
+    public int RetryCount { get; set; }
+    public TimeSpan RetryInterval { get; set; }
+    public float RetryIntervalFactor { get; set; }
+}