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,39 @@
         <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="ImTools" Version="4.0.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/LoggerExtensions.cs

@@ -0,0 +1,18 @@
+using Blumchen.Subscriptions.Replication;
+using Microsoft.Extensions.Logging;
+
+namespace Blumchen.DependencyInjection;
+
+internal static partial class LoggerExtensions
+
+{
+    [LoggerMessage(Message = "{workerName} started", Level = LogLevel.Information)]
+    public static partial void ServiceStarted(this ILogger logger, string workerName);
+
+    [LoggerMessage(Message = "{workerName} sopped", Level = LogLevel.Information)]
+    public static partial void ServiceStopped(this ILogger logger, string workerName);
+
+    [LoggerMessage(Message = "{message} processed", Level = LogLevel.Trace)]
+    public static partial void MessageProcessed(this ILogger logger, IEnvelope message);
+
+}

src/Blumchen/DependencyInjection/ServiceCollectionExtensions.cs

@@ -0,0 +1,41 @@
+using Blumchen.Subscriber;
+using Blumchen.Subscriptions.Replication;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using Npgsql;
+
+#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
+            .AddHostedService(provider =>
+                new Worker<T>(workerOptions(provider, new WorkerOptionsBuilder()).Build(),
+                    provider.GetRequiredService<ILogger<Worker<T>>>()));
+
+    public static IServiceCollection AddBlumchen<T>(
+        this IServiceCollection service,
+        string connectionString,
+        Func<IServiceProvider, IConsumes, OptionsBuilder> consumerFn) where T : class, IMessageHandler {
+        return service
+            .AddHostedService(provider =>
+                new Worker<T>(MinimalWorkerOptions(provider, new WorkerOptionsBuilder()).Build(),
+                provider.GetService<ILogger<Worker<T>>>() ?? new NullLogger<Worker<T>>()));
+
+        IWorkerOptionsBuilder MinimalWorkerOptions(IServiceProvider provider, IWorkerOptionsBuilder builder)
+             => builder.Subscription(optionsBuilder => consumerFn(provider, optionsBuilder)
+                .ConnectionString(connectionString)
+                .DataSource(new NpgsqlDataSourceBuilder(connectionString)
+                    .UseLoggerFactory(provider.GetService<ILoggerFactory>()).Build()));
+
+
+    }
+}

src/Blumchen/DependencyInjection/Worker.cs

@@ -0,0 +1,28 @@
+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}>";
+    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);
+            logger.ServiceStarted(WorkerName);
+            while (await cursor.MoveNextAsync().ConfigureAwait(false) && !ct.IsCancellationRequested)
+               logger.MessageProcessed(cursor.Current);
+        }, token).ConfigureAwait(false), stoppingToken).ConfigureAwait(false);
+        logger.ServiceStopped(WorkerName);
+    }
+
+}