first working commit

Author: MnatsakanMargaryan

src/PandaNuGet/Abstractions/InboxConsumer.cs

@@ -1,10 +1,10 @@
 using EFCore.PostgresExtensions.Enums;
 using EFCore.PostgresExtensions.Extensions;
-using FinHub.Mock1.Box.Entities;
-using FinHub.Mock1.Box.Enums;
-using FinHub.Mock1.Box.Implementations;
 using MassTransit;
+using MassTransit.PostgresOutbox;
 using MassTransit.PostgresOutbox.Abstractions;
+using MassTransit.PostgresOutbox.Entities;
+using MassTransit.PostgresOutbox.Enums;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;

src/PandaNuGet/Extensions/ServiceCollectionExtensions.cs

@@ -1,7 +1,6 @@
 using FinHub.Mock1.Box.Abstractions;
-using FinHub.Mock1.Box.Implementations;
-using FinHub.Mock1.Box.Jobs;
 using MassTransit.PostgresOutbox.Abstractions;
+using MassTransit.PostgresOutbox.Jobs;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 

src/PandaNuGet/Jobs/InboxMessageRemovalService.cs

@@ -0,0 +1,52 @@
+using MassTransit.PostgresOutbox.Abstractions;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MassTransit.PostgresOutbox.Jobs
+{
+    internal class InboxMessageRemovalService<TDbContext> : BackgroundService
+       where TDbContext : DbContext, IInboxDbContext
+    {
+        private readonly int _beforeInDays;
+        private readonly PeriodicTimer _timer;
+        private readonly ILogger<InboxMessageRemovalService<TDbContext>> _logger;
+        private readonly IServiceScopeFactory _serviceScopeFactory;
+
+        public InboxMessageRemovalService(IServiceScopeFactory serviceScopeFactory, ILogger<InboxMessageRemovalService<TDbContext>> logger, Settings settings)
+        {
+            _serviceScopeFactory = serviceScopeFactory;
+            _logger = logger;
+            _timer = new(settings.InboxRemovalTimerPeriod);
+            _beforeInDays = settings.InboxRemovalBeforeInDays;
+        }
+
+        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+        {
+            while (await _timer.WaitForNextTickAsync(cancellationToken))
+            {
+                _logger.LogInformation($"{nameof(InboxMessageRemovalService<TDbContext>)} started iteration");
+
+                using var scope = _serviceScopeFactory.CreateScope();
+                using var dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>();
+
+                try
+                {
+                    var daysBefore = DateTime.UtcNow.AddDays(-_beforeInDays);
+
+                    await dbContext.InboxMessages
+                                   .Where(x => x.State == Enums.MessageState.Done)
+                                   .Where(x => x.UpdatedAt < daysBefore)
+                                   .ExecuteDeleteAsync(cancellationToken);
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, ex.Message);
+                }
+
+                _logger.LogInformation($"{nameof(InboxMessageRemovalService<TDbContext>)} finished iteration");
+            }
+        }
+    }
+}

src/PandaNuGet/Jobs/OutboxMessagePublisherService.cs

@@ -0,0 +1,104 @@
+using EFCore.PostgresExtensions.Enums;
+using EFCore.PostgresExtensions.Extensions;
+using FinHub.Mock1.Box.Abstractions;
+using MassTransit;
+using MassTransit.PostgresOutbox.Enums;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+
+namespace MassTransit.PostgresOutbox.Jobs
+{
+    internal class OutboxMessagePublisherService<TDbContext> : BackgroundService
+       where TDbContext : DbContext, IOutboxDbContext
+    {
+        private readonly int _batchCount;
+        private readonly PeriodicTimer _timer;
+        private readonly ILogger<OutboxMessagePublisherService<TDbContext>> _logger;
+        private readonly IServiceScopeFactory _serviceScopeFactory;
+        private readonly IPublishEndpoint _publishEndpoint;
+
+        public OutboxMessagePublisherService(IPublishEndpoint publishEndpoint,
+           IServiceScopeFactory serviceScopeFactory,
+           ILogger<OutboxMessagePublisherService<TDbContext>> logger, Settings settings)
+        {
+            _publishEndpoint = publishEndpoint;
+            _serviceScopeFactory = serviceScopeFactory;
+            _logger = logger;
+            _timer = new(settings.PublisherTimerPeriod);
+            _batchCount = settings.PublisherBatchCount;
+        }
+
+        protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+        {
+            while (await _timer.WaitForNextTickAsync(cancellationToken))
+            {
+                _logger.LogInformation($"{nameof(OutboxMessagePublisherService<TDbContext>)} started iteration");
+
+                using var scope = _serviceScopeFactory.CreateScope();
+                using var dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>();
+                using var transactionScope = await dbContext.Database.BeginTransactionAsync(System.Data.IsolationLevel.ReadCommitted, cancellationToken);
+
+                try
+                {
+                    var messages = await dbContext.OutboxMessages
+                                                  .Where(x => x.State == MessageState.New)
+                                                  .OrderBy(x => x.CreatedAt)
+                                                  .ForUpdate(LockBehavior.SkipLocked)
+                                                  .Take(_batchCount)
+                                                  .ToListAsync(cancellationToken);
+
+                    if (messages.Count == 0)
+                    {
+                        continue;
+                    }
+
+                    var publishedMessageIds = new List<Guid>(messages.Count);
+
+                    foreach (var message in messages)
+                    {
+                        try
+                        {
+                            var type = Type.GetType(message.Type);
+
+                            var messageObject = JsonConvert.DeserializeObject(message.Payload, type!);
+
+                            await _publishEndpoint.Publish(messageObject!, type!, x => x.Headers.Set(Constants.OutboxMessageId, message.Id), cancellationToken);
+
+                            publishedMessageIds.Add(message.Id);
+                        }
+                        catch (Exception ex)
+                        {
+                            _logger.LogError(ex, ex.Message);
+                        }
+                    }
+
+                    if (publishedMessageIds.Count == 0)
+                    {
+                        continue;
+                    }
+
+                    var utcNow = DateTime.UtcNow;
+
+                    await dbContext.OutboxMessages
+                                   .Where(b => publishedMessageIds.Contains(b.Id))
+                                   .ExecuteUpdateAsync(x => x.SetProperty(m => m.State, MessageState.Done)
+                                                             .SetProperty(m => m.UpdatedAt, utcNow), cancellationToken: cancellationToken);
+
+                    await dbContext.SaveChangesAsync();
+                    await transactionScope.CommitAsync();
+
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, ex.Message);
+                    await transactionScope.RollbackAsync(cancellationToken);
+                }
+
+                _logger.LogInformation($"{nameof(OutboxMessagePublisherService<TDbContext>)} finished iteration");
+            }
+        }
+    }
+}

src/PandaNuGet/Jobs/OutboxMessageRemovalService.cs

@@ -0,0 +1,54 @@
+using FinHub.Mock1.Box.Abstractions;
+using MassTransit.PostgresOutbox;
+using MassTransit.PostgresOutbox.Enums;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MassTransit.PostgresOutbox.Jobs
+{
+   internal class OutboxMessageRemovalService<TDbContext> : BackgroundService
+      where TDbContext : DbContext, IOutboxDbContext
+   {
+      private readonly int _beforeInDays;
+      private readonly PeriodicTimer _timer;
+      private readonly ILogger<OutboxMessageRemovalService<TDbContext>> _logger;
+      private readonly IServiceScopeFactory _serviceScopeFactory;
+
+      public OutboxMessageRemovalService(IServiceScopeFactory serviceScopeFactory, ILogger<OutboxMessageRemovalService<TDbContext>> logger, Settings settings)
+      {
+         _serviceScopeFactory = serviceScopeFactory;
+         _logger = logger;
+         _timer = new(settings.OutboxRemovalTimerPeriod);
+         _beforeInDays = settings.OutboxRemovalBeforeInDays;
+      }
+
+      protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+      {
+         while (await _timer.WaitForNextTickAsync(cancellationToken))
+         {
+            _logger.LogInformation($"{nameof(OutboxMessageRemovalService<TDbContext>)} started iteration");
+
+            using var scope = _serviceScopeFactory.CreateScope();
+            using var dbContext = scope.ServiceProvider.GetRequiredService<TDbContext>();
+
+            try
+            {
+               var daysBefore = DateTime.UtcNow.AddDays(-_beforeInDays);
+
+               await dbContext.OutboxMessages
+                              .Where(x => x.State == MessageState.Done)
+                              .Where(x => x.UpdatedAt < daysBefore)
+                              .ExecuteDeleteAsync(cancellationToken);
+            }
+            catch (Exception ex)
+            {
+               _logger.LogError(ex, ex.Message);
+            }
+
+            _logger.LogInformation($"{nameof(OutboxMessageRemovalService<TDbContext>)} finished iteration");
+         }
+      }
+   }
+}

src/PandaNuGet/MassTransit.PostgresOutbox.csproj

@@ -23,7 +23,11 @@
     </ItemGroup>
 
     <ItemGroup>
+      <PackageReference Include="MassTransit" Version="8.2.0" />
+      <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="8.0.1" />
       <PackageReference Include="Microsoft.EntityFrameworkCore" Version="8.0.3" />
+      <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
+      <PackageReference Include="Pandatech.EFCore.PostgresExtensions" Version="1.0.0" />
     </ItemGroup>
 
 </Project>