Add settings for outbox processor (PollingIntervalSeconds and MaxEventsPerCycle).

Author: runemalm

src/OpenDDD/API/Extensions/OpenDddServiceCollectionExtensions.cs

@@ -50,6 +50,7 @@
 using OpenDDD.Infrastructure.TransactionalOutbox.EfCore;
 using OpenDDD.Infrastructure.TransactionalOutbox.OpenDdd.InMemory;
 using OpenDDD.Infrastructure.TransactionalOutbox.OpenDdd.Postgres;
+using OpenDDD.Infrastructure.TransactionalOutbox.Options;
 using OpenDDD.Infrastructure.Utils;
 
 namespace OpenDDD.API.Extensions
@@ -183,6 +184,11 @@ private static void AddMessaging(this IServiceCollection services, OpenDddOption
 
         private static void AddTransactionalOutbox(this IServiceCollection services)
         {
+            services.Configure<OpenDddOutboxProcessorOptions>(options =>
+            {
+                var config = services.BuildServiceProvider().GetRequiredService<IConfiguration>();
+                config.GetSection("OpenDDD:OutboxProcessor").Bind(options);
+            });
             services.AddHostedService<OutboxProcessor>();
         }
 

src/OpenDDD/Infrastructure/TransactionalOutbox/EfCore/EfCoreOutboxRepository.cs

@@ -40,14 +40,18 @@ public async Task SaveEventAsync<TEvent>(TEvent @event, CancellationToken ct) wh
             await _session.DbContext.SaveChangesAsync(ct);
         }
 
-        public async Task<List<OutboxEntry>> GetPendingEventsAsync(CancellationToken ct)
+        public async Task<List<OutboxEntry>> GetPendingEventsAsync(int? maxCount = null, CancellationToken ct = default)
         {
             await _session.OpenConnectionAsync(ct);
-
-            return await _session.DbContext.Set<OutboxEntry>()
+            
+            IQueryable<OutboxEntry> query = _session.DbContext.Set<OutboxEntry>()
                 .Where(e => e.ProcessedAt == null)
-                .OrderBy(e => e.CreatedAt)
-                .ToListAsync(ct);
+                .OrderBy(e => e.CreatedAt);
+
+            if (maxCount.HasValue)
+                query = query.Take(maxCount.Value);
+
+            return await query.ToListAsync(ct);
         }
 
         public async Task MarkEventAsProcessedAsync(Guid eventId, CancellationToken ct)

src/OpenDDD/Infrastructure/TransactionalOutbox/IOutboxRepository.cs

@@ -5,7 +5,7 @@ namespace OpenDDD.Infrastructure.TransactionalOutbox
     public interface IOutboxRepository
     {
         Task SaveEventAsync<TEvent>(TEvent @event, CancellationToken ct) where TEvent : IEvent;
-        Task<List<OutboxEntry>> GetPendingEventsAsync(CancellationToken ct);
+        Task<List<OutboxEntry>> GetPendingEventsAsync(int? maxCount = null, CancellationToken ct = default);
         Task MarkEventAsProcessedAsync(Guid eventId, CancellationToken ct);
     }
 }

src/OpenDDD/Infrastructure/TransactionalOutbox/OpenDdd/InMemory/InMemoryOpenDddOutboxRepository.cs

@@ -42,10 +42,16 @@ public async Task SaveEventAsync<TEvent>(TEvent @event, CancellationToken ct) wh
             _logger.LogDebug("Added event to in-memory outbox: {EventName}", outboxEntry.EventName);
         }
 
-        public async Task<List<OutboxEntry>> GetPendingEventsAsync(CancellationToken ct)
+        public async Task<List<OutboxEntry>> GetPendingEventsAsync(int? maxCount = null, CancellationToken ct = default)
         {
             var entries = await _session.SelectAllAsync<OutboxEntry>(OutboxTable, ct);
-            return entries.Where(entry => entry.ProcessedAt == null).ToList();
+            var pending = entries
+                .Where(entry => entry.ProcessedAt == null)
+                .OrderBy(entry => entry.CreatedAt);
+
+            return maxCount.HasValue
+                ? pending.Take(maxCount.Value).ToList()
+                : pending.ToList();
         }
 
         public async Task MarkEventAsProcessedAsync(Guid eventId, CancellationToken ct)

src/OpenDDD/Infrastructure/TransactionalOutbox/OpenDdd/Postgres/PostgresOpenDddOutboxRepository.cs

@@ -43,13 +43,23 @@ INSERT INTO outbox_entries (id, event_type, event_name, payload, created_at, pro
             await cmd.ExecuteNonQueryAsync(ct);
         }
 
-        public async Task<List<OutboxEntry>> GetPendingEventsAsync(CancellationToken ct)
+        public async Task<List<OutboxEntry>> GetPendingEventsAsync(int? maxCount = null, CancellationToken ct = default)
         {
             await _session.OpenConnectionAsync(ct);
+            
+            var query = @"
+                SELECT id, event_type, event_name, payload, created_at, processed_at
+                FROM outbox_entries
+                WHERE processed_at IS NULL
+                ORDER BY created_at";
 
-            const string query = "SELECT id, event_type, event_name, payload, created_at, processed_at FROM outbox_entries WHERE processed_at IS NULL ORDER BY created_at;";
+            query = maxCount.HasValue ? query + " LIMIT @maxCount;" : query + ";";
 
             await using var cmd = new NpgsqlCommand(query, _session.Connection, _session.Transaction);
+            
+            if (maxCount.HasValue)
+                cmd.Parameters.AddWithValue("maxCount", maxCount.Value);
+
             await using var reader = await cmd.ExecuteReaderAsync(ct);
 
             var events = new List<OutboxEntry>();

src/OpenDDD/Infrastructure/TransactionalOutbox/Options/OpenDddOutboxProcessorOptions.cs

@@ -0,0 +1,8 @@
+namespace OpenDDD.Infrastructure.TransactionalOutbox.Options
+{
+    public class OpenDddOutboxProcessorOptions
+    {
+        public int PollingIntervalSeconds { get; set; } = 3;
+        public int? MaxEventsPerCycle { get; set; } = null;
+    }
+}

src/OpenDDD/Infrastructure/TransactionalOutbox/OutboxProcessor.cs

@@ -7,6 +7,7 @@
 using OpenDDD.Domain.Model.Helpers;
 using OpenDDD.Infrastructure.Events;
 using OpenDDD.Infrastructure.Persistence.DatabaseSession;
+using OpenDDD.Infrastructure.TransactionalOutbox.Options;
 
 namespace OpenDDD.Infrastructure.TransactionalOutbox
 {
@@ -16,24 +17,29 @@ public class OutboxProcessor : BackgroundService
         private readonly StartupHostedService _startupService;
         private readonly ILogger<OutboxProcessor> _logger;
         private readonly OpenDddOptions _options;
+        private readonly OpenDddOutboxProcessorOptions _processorOptions;
 
         public OutboxProcessor(
             IServiceScopeFactory serviceScopeFactory,
             StartupHostedService startupService,
             ILogger<OutboxProcessor> logger,
-            IOptions<OpenDddOptions> options)
+            IOptions<OpenDddOptions> options,
+            IOptions<OpenDddOutboxProcessorOptions> processorOptions)
         {
             _serviceScopeFactory = serviceScopeFactory ?? throw new ArgumentNullException(nameof(serviceScopeFactory));
             _startupService = startupService;
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _options = options.Value ?? throw new ArgumentNullException(nameof(options));
+            _processorOptions = processorOptions.Value ?? throw new ArgumentNullException(nameof(processorOptions));
         }
 
         protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             _logger.LogInformation("Outbox Processor started.");
+            _logger.LogDebug("Polling interval: {PollingIntervalSeconds}s, Max events per cycle: {MaxEventsPerCycle}",
+                _processorOptions.PollingIntervalSeconds,
+                _processorOptions.MaxEventsPerCycle);
 
-            // Ensure database setup is complete
             _logger.LogInformation("Waiting for database setup to complete before starting outbox processing...");
             await _startupService.StartupCompleted;
             _logger.LogInformation("Database setup completed. Starting outbox processing...");
@@ -56,7 +62,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
                     await databaseSession.OpenConnectionAsync(stoppingToken);
 
-                    var pendingEvents = await outboxRepository.GetPendingEventsAsync(stoppingToken);
+                    var pendingEvents = await outboxRepository
+                        .GetPendingEventsAsync(_processorOptions.MaxEventsPerCycle, stoppingToken);
+
+                    _logger.Log(
+                        pendingEvents.Any() ? LogLevel.Debug : LogLevel.Trace,
+                        "Fetched {EventCount} pending events.", pendingEvents.Count);
 
                     foreach (var outboxEntry in pendingEvents)
                     {
@@ -72,6 +83,8 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
                             await messagingProvider.PublishAsync(topic, outboxEntry.Payload, stoppingToken);
                             await outboxRepository.MarkEventAsProcessedAsync(outboxEntry.Id, stoppingToken);
+                            
+                            _logger.LogDebug("Successfully published and marked event {EventId} as processed.", outboxEntry.Id);
                         }
                         catch (Exception ex)
                         {
@@ -84,7 +97,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                     _logger.LogError(ex, "Unexpected error in Outbox Processor.");
                 }
 
-                await Task.Delay(TimeSpan.FromSeconds(3), stoppingToken);
+                await Task.Delay(TimeSpan.FromSeconds(_processorOptions.PollingIntervalSeconds), stoppingToken);
             }
 
             _logger.LogInformation("Outbox Processor stopping.");