MVP proposal

Blumchen.sln

@@ -36,13 +36,17 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "pgAdmin", "pgAdmin", "{C050
 		docker\pgAdmin\servers.json = docker\pgAdmin\servers.json
 	EndProjectSection
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "postgres", "postgres", "{8AAAA344-B5FD-48D9-B2BA-379E374448D4}"
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "demo", "demo", "{A4044484-FE08-4399-8239-14AABFA30AD7}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SubscriberWorker", "src\SubscriberWorker\SubscriberWorker.csproj", "{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "UnitTests", "src\UnitTests\UnitTests.csproj", "{B16305B4-8AC3-4435-AADB-D9E2ACAA1C13}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "scripts", "scripts", "{13543764-C1E1-482E-887A-816ABA4CB71C}"
 	ProjectSection(SolutionItems) = preProject
-		docker\postgres\init.sql = docker\postgres\init.sql
+		scripts\autoheal.ps1 = scripts\autoheal.ps1
 	EndProjectSection
 EndProject
-Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "demo", "demo", "{A4044484-FE08-4399-8239-14AABFA30AD7}"
-EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -69,6 +73,14 @@ Global
 		{2BBDA071-FB1C-4D62-A954-B22EA6B1C738}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{2BBDA071-FB1C-4D62-A954-B22EA6B1C738}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{2BBDA071-FB1C-4D62-A954-B22EA6B1C738}.Release|Any CPU.Build.0 = Release|Any CPU
+		{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B16305B4-8AC3-4435-AADB-D9E2ACAA1C13}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B16305B4-8AC3-4435-AADB-D9E2ACAA1C13}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B16305B4-8AC3-4435-AADB-D9E2ACAA1C13}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B16305B4-8AC3-4435-AADB-D9E2ACAA1C13}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -78,7 +90,7 @@ Global
 		{F2878625-0919-4C26-8DC9-58CD8FA34050} = {A4044484-FE08-4399-8239-14AABFA30AD7}
 		{F81E2D5B-FC59-4396-A911-56BE65E4FE80} = {A4044484-FE08-4399-8239-14AABFA30AD7}
 		{C050E9E8-3FB6-4581-953F-31826E385FB4} = {CD59A1A0-F40D-4047-87A3-66C0F1519FA5}
-		{8AAAA344-B5FD-48D9-B2BA-379E374448D4} = {CD59A1A0-F40D-4047-87A3-66C0F1519FA5}
+		{DB58DB36-0366-4ABA-BC06-FCA9BB10EB92} = {A4044484-FE08-4399-8239-14AABFA30AD7}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {9A868C51-0460-4700-AF33-E1A921192614}

README.md

@@ -18,7 +18,7 @@ Main logic is placed in [EventsSubscription](./src/Blumchen/Subscriptions/Subscr
 ```shell
 docker-compose up
 ```
-2. Run(order doesn't matter) Publisher and Subscriber apps, under 'demo' folder, from vs-studio, and follow Publisher instructions.
+2. Run(order doesn't matter) Publisher and (Subscriber or SubscriberWorker)  apps, under 'demo' folder, from vs-studio, and follow Publisher instructions.
 
 ## Testing (against default docker instance)
 

docker-compose.yml

@@ -3,6 +3,7 @@ version: '3.8'  # Specify Docker Compose version
 services:
     postgres:
         image: postgres:15.1-alpine
+        container_name: db
         ports:
             - "5432:5432"
         environment:
@@ -15,9 +16,8 @@ services:
             - "wal_level=logical"
             - "-c"
             - "wal_compression=on"
-        volumes:
-            - ./docker/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
-
+            - "-c"
+            - "max_slot_wal_keep_size=1"   
     pgadmin:
         container_name: pgadmin_container
         image: dpage/pgadmin4
@@ -36,4 +36,3 @@ services:
         depends_on:
             - postgres
         restart: unless-stopped
-

scripts/autoheal.ps1

@@ -0,0 +1,76 @@
+#usage: .\scripts\autoheal.ps1 -SolutionFile .\Blumchen.sln -ComposeFile .\docker-compose.yml
+param(
+  [string]$SolutionFile,
+  [string]$ComposeFile
+)
+
+
+$env:DOCKER_CLI_HINTS=$false #disable docker hints
+Write-Host "Setup infrastrucure"
+
+try { 
+
+	start powershell {
+		docker compose up;
+		Read-Host;
+
+	} 
+
+	Write-Host "Waiting for container readiness..."
+	do
+	{
+		Start-Sleep -s 5
+		$state=$(docker inspect db|ConvertFrom-Json).State
+		$status=$state.Status
+		$exitCode=$state.ExitCode
+		$restart=$state.Restarting
+	}Until(($status -eq "running") -and ($exitCode -eq 0) -and ($restart -eq $false))
+
+	Write-Host "...Done"
+
+	Write-Host "Start subscriber"
+	start powershell {
+		dotnet run --project ./src/SubscriberWorker/SubscriberWorker.csproj
+		Read-Host;
+	}
+
+	Write-Host "Publishing 10 messages to test the subscriptions are working properly: hit ENTER when done!"
+
+	start powershell {
+		dotnet run --project ./src/Publisher/Publisher.csproj -- -c 10 -t \"UserCreated|UserDeleted|UserModified\";
+	}
+
+	Read-Host;
+
+	Write-Host "Start massive insert to force wal segment creation..."
+	start powershell {
+		dotnet run --project ./src/Publisher/Publisher.csproj -- -c 800000 -t "UserSubscribed"
+	}
+
+	Write-Host "Wait for subscribers to auto heal on error...reporting on row insert"
+
+	Start-Sleep -s 15
+	do
+	{
+		docker exec -it db psql -h localhost -U postgres -w -c "select count(*) from outbox;"
+	}Until(Read-Host "Enter to report on counting rows(another key to proceed when done)" "")
+
+	Write-Host "Subscribers resiliency tested :-)"
+	Write-Host "Publishing 10 messages to test the subscriptions are still working properly: hit ENTER when done!"
+
+	start powershell {
+		dotnet run --project ./src/Publisher/Publisher.csproj -- -c 10 -t \"UserCreated|UserDeleted|UserModified\"
+	}
+	Read-Host;
+
+	Write-Host "We're done...: hit ENTER to shut down!"
+
+	Read-Host;
+
+}catch {
+  Write-Host "An error occurred:"
+  Write-Host $_
+}
+finally{
+	docker compose -f $ComposeFile down --rmi local 
+}

src/Blumchen/Blumchen.csproj

@@ -1,7 +1,7 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
-        <VersionPrefix>0.1.0</VersionPrefix>
+        <VersionPrefix>0.1.1</VersionPrefix>
         <TargetFramework>net8.0</TargetFramework>
         <GenerateAssemblyTitleAttribute>true</GenerateAssemblyTitleAttribute>
         <GenerateAssemblyDescriptionAttribute>true</GenerateAssemblyDescriptionAttribute>
@@ -25,20 +25,38 @@
         <SymbolPackageFormat>snupkg</SymbolPackageFormat>
         <RootNamespace>Blumchen</RootNamespace>
     </PropertyGroup>
+
+    <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|AnyCPU'">
+      <NoWarn>1591</NoWarn>
+      <WarningsNotAsErrors></WarningsNotAsErrors>
+    </PropertyGroup>
+
+    <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
+      <NoWarn>1591</NoWarn>
+      <WarningsNotAsErrors></WarningsNotAsErrors>
+    </PropertyGroup>
     <ItemGroup>
     <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleToAttribute">
         <_Parameter1>Tests</_Parameter1>
     </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleToAttribute">
+        <_Parameter1>UnitTests</_Parameter1>
+    </AssemblyAttribute>
+    <AssemblyAttribute Include="System.Runtime.CompilerServices.InternalsVisibleToAttribute">
+        <_Parameter1>Blumchen.DependencyInjection</_Parameter1>
+    </AssemblyAttribute>
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="JetBrains.Annotations" Version="2023.3.0" >
+        <PackageReference Include="JetBrains.Annotations" Version="2024.2.0">
             <PrivateAssets>all</PrivateAssets>
             <ExcludeAssets>none</ExcludeAssets>
             <IncludeAssets>all</IncludeAssets>
         </PackageReference>
+        <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="8.0.0" />
         <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="8.0.1" />
         <PackageReference Include="Npgsql" Version="8.0.3" />
+        <PackageReference Include="Polly.Core" Version="8.4.1" />
     </ItemGroup>
 
 </Project>

src/Blumchen/ConfigurationException.cs

@@ -0,0 +1,3 @@
+namespace Blumchen;
+
+public class ConfigurationException(string message): Exception(message);

src/Blumchen/Database/Run.cs

@@ -1,11 +1,8 @@
 using System.Data;
 using System.Runtime.CompilerServices;
 using Blumchen.Subscriptions.Replication;
-using Blumchen.Subscriptions.ReplicationMessageHandlers;
 using Npgsql;
 
-#pragma warning disable CS1591 // Missing XML comment for publicly visible type or member
-
 namespace Blumchen.Database;
 
 public static class Run
@@ -16,11 +13,11 @@ private static async Task Execute(
         CancellationToken ct)
     {
         await using var command = dataSource.CreateCommand(sql);
-        await command.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+        await command.ExecuteNonQueryAsync(ct);
     }
 
     public static async Task EnsureTableExists(this NpgsqlDataSource dataSource, TableDescriptorBuilder.MessageTable tableDescriptor, CancellationToken ct)
-        => await dataSource.Execute(tableDescriptor.ToString(), ct).ConfigureAwait(false);
+        => await dataSource.Execute(string.Concat("select pg_advisory_xact_lock(12345);", tableDescriptor), ct).ConfigureAwait(false);
 
     public static async Task<bool> Exists(
         this NpgsqlDataSource dataSource,

src/Blumchen/DependencyInjection/ServiceCollectionExtensions.cs

@@ -0,0 +1,23 @@
+using Blumchen.Subscriptions.Replication;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+#pragma warning disable IL2091
+
+namespace Blumchen.DependencyInjection;
+
+public static class ServiceCollectionExtensions
+{
+
+    public static IServiceCollection AddBlumchen<T>(
+        this IServiceCollection service,
+        Func<IServiceProvider, IWorkerOptionsBuilder, IWorkerOptionsBuilder> workerOptions)
+        where T : class, IMessageHandler =>
+        service
+            .AddKeyedSingleton(typeof(T), (provider, _) => workerOptions(provider, new WorkerOptionsBuilder()).Build())
+            .AddHostedService(provider =>
+                new Worker<T>(workerOptions(provider, new WorkerOptionsBuilder()).Build(),
+                    provider.GetRequiredService<ILogger<Worker<T>>>()));
+
+
+}

src/Blumchen/DependencyInjection/Worker.cs

@@ -0,0 +1,48 @@
+using System.Collections.Concurrent;
+using Blumchen.Subscriptions;
+using Blumchen.Subscriptions.Replication;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace Blumchen.DependencyInjection;
+
+public class Worker<T>(
+    WorkerOptions options,
+    ILogger<Worker<T>> logger): BackgroundService where T : class, IMessageHandler
+{
+    private string WorkerName { get; } = $"{nameof(Worker<T>)}<{typeof(T).Name}>";
+    private static readonly ConcurrentDictionary<string, Action<ILogger, string, object[]>> LoggingActions = new(StringComparer.OrdinalIgnoreCase);
+    private static void Notify(ILogger logger, LogLevel level, string template, params object[] parameters)
+    {
+        LoggingActions.GetOrAdd(template,_ => LoggerAction(level, logger.IsEnabled(level)))(logger, template, parameters);
+        return;
+
+        static Action<ILogger, string, object[]> LoggerAction(LogLevel ll, bool enabled) =>
+            (ll, enabled) switch
+            {
+                (LogLevel.Information, true) => (logger, template, parameters) => logger.LogInformation(template, parameters),
+                (LogLevel.Debug, true) => (logger, template, parameters) => logger.LogDebug(template, parameters),
+                (LogLevel.Trace, true) => (logger, template, parameters) => logger.LogTrace(template, parameters),
+                (LogLevel.Warning, true) => (logger, template, parameters) => logger.LogWarning(template, parameters),
+                (LogLevel.Error, true) => (logger, template, parameters) => logger.LogError(template, parameters),
+                (LogLevel.Critical, true) => (logger, template, parameters) => logger.LogCritical(template, parameters),
+                (_, _) => (_, _, _) => { }
+            };
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        await options.OuterPipeline.ExecuteAsync(async token =>
+            await options.InnerPipeline.ExecuteAsync(async ct =>
+        {
+            await using var subscription = new Subscription();
+            await using var cursor = subscription.Subscribe(options.SubscriberOptions, ct)
+                .GetAsyncEnumerator(ct);
+            Notify(logger, LogLevel.Information, "{WorkerName} started", WorkerName);
+            while (await cursor.MoveNextAsync().ConfigureAwait(false) && !ct.IsCancellationRequested)
+                Notify(logger, LogLevel.Trace, "{cursor.Current} processed", cursor.Current);
+        }, token).ConfigureAwait(false), stoppingToken).ConfigureAwait(false);
+        Notify(logger, LogLevel.Information, "{WorkerName} stopped", WorkerName);
+    }
+
+}

src/Blumchen/DependencyInjection/WorkerOptionsBuilder.cs

@@ -0,0 +1,67 @@
+using Blumchen.Subscriber;
+using Blumchen.Subscriptions.Management;
+using Npgsql;
+using Npgsql.Replication;
+using Polly;
+
+namespace Blumchen.DependencyInjection;
+
+public record WorkerOptions(
+    ISubscriberOptions SubscriberOptions,
+    ResiliencePipeline OuterPipeline,
+    ResiliencePipeline InnerPipeline);
+
+public interface IWorkerOptionsBuilder
+{
+    IWorkerOptionsBuilder ResiliencyPipeline(ResiliencePipeline resiliencePipeline);
+    IWorkerOptionsBuilder Subscription(Func<OptionsBuilder, OptionsBuilder>? builder);
+    WorkerOptions Build();
+    IWorkerOptionsBuilder EnableSubscriptionAutoHeal();
+}
+
+internal sealed class WorkerOptionsBuilder: IWorkerOptionsBuilder
+{
+    private ResiliencePipeline? _outerPipeline = default;
+    private Func<string, string, ResiliencePipeline>? _innerPipelineFn = default;
+    private Func<OptionsBuilder, OptionsBuilder>? _builder;
+
+    public IWorkerOptionsBuilder ResiliencyPipeline(ResiliencePipeline resiliencePipeline)
+    {
+        _outerPipeline = resiliencePipeline;
+        return this;
+    }public IWorkerOptionsBuilder Subscription(Func<OptionsBuilder, OptionsBuilder>? builder)
+    {
+        _builder = builder;
+        return this; 
+    }
+
+    public WorkerOptions Build()
+    {
+        ArgumentNullException.ThrowIfNull(_outerPipeline);
+        ArgumentNullException.ThrowIfNull(_builder);
+        var subscriberOptions = _builder(new OptionsBuilder()).Build();
+        return new(subscriberOptions, _outerPipeline,
+            _innerPipelineFn?.Invoke(subscriberOptions.ReplicationOptions.SlotName,subscriberOptions.ConnectionStringBuilder.ConnectionString) ??
+            ResiliencePipeline.Empty
+        );
+    }
+
+    public IWorkerOptionsBuilder EnableSubscriptionAutoHeal()
+    {
+        _innerPipelineFn = (replicationSlotName, connectionString) => new ResiliencePipelineBuilder().AddRetry(new()
+        {
+            ShouldHandle =
+                new PredicateBuilder().Handle<PostgresException>(exception =>
+                    exception.SqlState.Equals("55000", StringComparison.OrdinalIgnoreCase)),
+            MaxRetryAttempts = int.MaxValue,
+            OnRetry = async args =>
+            {
+                await using var conn = new LogicalReplicationConnection(connectionString);
+                await conn.Open(args.Context.CancellationToken);
+                await conn.ReCreate(replicationSlotName, args.Context.CancellationToken).ConfigureAwait(false);
+            },
+        }).Build();
+        return this;
+    }
+}
+