From fbd6c0c521fb0f699af270dd0275959b8328660c Mon Sep 17 00:00:00 2001 From: TekH Date: Thu, 23 Jul 2026 16:47:14 +0200 Subject: [PATCH] Refactor email processing and RabbitMQ initialization Centralized email processing logic in `RabbitMqEmailQueue` by moving it from `EmailSenderWorker`. Updated `IEmailQueue` to replace `StartConsumerAsync` with `InitAsync`, shifting to an initialization-based model for RabbitMQ. Refactored `RabbitMqEmailQueue` to handle email processing inline, including deserialization, logging, and sending emails via `IEmailService`. Enhanced error handling with detailed logging for failures. Removed lazy initialization (`Lazy`) in favor of explicit initialization via `InitAsync`. Simplified `EmailSenderWorker` by removing `ProcessEmailAsync` and its dependency on `IEmailService`. Updated it to call `EmailQueue.InitAsync` for initialization. Improved logging and error handling for better visibility into email processing and failure scenarios. Updated RabbitMQ acknowledgment and rejection logic to use `args.CancellationToken`. --- .../Workers/EmailSenderWorker.cs | 36 +-------- .../Common/Interfaces/IEmailQueue.cs | 7 +- .../Queue/RabbitMqEmailQueue.cs | 73 ++++++++----------- 3 files changed, 36 insertions(+), 80 deletions(-) diff --git a/src/DigitalData.EmailProfiler.API/Workers/EmailSenderWorker.cs b/src/DigitalData.EmailProfiler.API/Workers/EmailSenderWorker.cs index 289cbc0..24549ba 100644 --- a/src/DigitalData.EmailProfiler.API/Workers/EmailSenderWorker.cs +++ b/src/DigitalData.EmailProfiler.API/Workers/EmailSenderWorker.cs @@ -1,7 +1,5 @@ using DigitalData.EmailProfiler.API.Configurations; -using DigitalData.EmailProfiler.Application.Common.Events; using DigitalData.EmailProfiler.Application.Common.Interfaces; -using DigitalData.EmailProfiler.Application.EmailSending.Commands; using Microsoft.Extensions.Options; namespace DigitalData.EmailProfiler.API.Workers; @@ -13,7 +11,6 @@ namespace DigitalData.EmailProfiler.API.Workers; /// public class EmailSenderWorker( IEmailQueue EmailQueue, - IEmailService EmailService, ILogger Logger, IOptions configuration) : BackgroundService { @@ -32,7 +29,7 @@ public class EmailSenderWorker( try { // Start RabbitMQ consumer (event-driven, non-blocking) - await EmailQueue.StartConsumerAsync(ProcessEmailAsync, stoppingToken); + await EmailQueue.InitAsync(stoppingToken); // Keep worker alive until cancellation await Task.Delay(Timeout.Infinite, stoppingToken); @@ -49,35 +46,4 @@ public class EmailSenderWorker( Logger.LogInformation("EmailSenderWorker stopped"); } - - /// - /// Process single email message (callback from RabbitMQ consumer) - /// NO database operations - only send email and log result - /// - private async Task ProcessEmailAsync(OutgoingEmailEvent outgoingEmailEvent) - { - try - { - Logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}", - outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject); - - // Send email via SMTP (SMTP config is injected in IEmailService via IOptions) - await EmailService.SendEmailAsync( - outgoingEmailEvent.Recipient, - outgoingEmailEvent.Subject, - outgoingEmailEvent.Body, - isHtml: outgoingEmailEvent.IsHtml); - - Logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", - outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject); - } - catch (Exception ex) - { - Logger.LogError(ex, "Failed to send email: To={To}, Subject={Subject}. Moving to DLQ.", - outgoingEmailEvent.Recipient, outgoingEmailEvent.Subject); - - // Re-throw to trigger NACK in RabbitMQ consumer (requeue=false → DLQ) - throw; - } - } } \ No newline at end of file diff --git a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs index 7805e9d..d40d08f 100644 --- a/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Application/Common/Interfaces/IEmailQueue.cs @@ -13,9 +13,8 @@ public interface IEmailQueue Task GetQueueDepthAsync(CancellationToken cancellationToken = default); /// - /// Start event-driven consumer that calls callback when mail received + /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. + /// Called lazily on first use via EnsureInitializedAsync. /// - /// Callback to process received mail - /// Cancellation token - Task StartConsumerAsync(Func onMailReceived, CancellationToken cancellationToken = default); + Task InitAsync(CancellationToken cancellationToken = default); } diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs index 8eef114..d6025c8 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs @@ -22,26 +22,22 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable private readonly RabbitMqConfiguration _config; private IConnection? _connection; private IChannel _channel; - - // Lazy ensures InitAsync is called only ONCE (thread-safe) - private readonly Lazy _initializationTask; + private readonly IEmailService _emailService; #pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor - public RabbitMqEmailQueue(IOptions config, ILogger logger) + public RabbitMqEmailQueue(IOptions config, ILogger logger, IEmailService emailService) #pragma warning restore CS8618 { _logger = logger; _config = config.Value; - - // Lazy initialization - NOT executed until first access (non-blocking constructor) - _initializationTask = new Lazy(InitAsync); + _emailService = emailService; } /// /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Called lazily on first use via EnsureInitializedAsync. /// - private async Task InitAsync() + public async Task InitAsync(CancellationToken cancellationToken = default) { _logger.LogInformation("Initializing RabbitMQ connection and queues..."); @@ -106,23 +102,15 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey); - + + await StartConsumerAsync(cancellationToken); + _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); } - /// - /// Ensure RabbitMQ is initialized before any operation. - /// Thread-safe and guarantees single initialization via Lazy. - /// - private async Task EnsureInitializedAsync() - { - await _initializationTask.Value; - } public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) { - await EnsureInitializedAsync(); // Initialize on first call - var json = JsonSerializer.Serialize(outgoingEmailEvent); var body = Encoding.UTF8.GetBytes(json); @@ -133,7 +121,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) }; - await _channel!.BasicPublishAsync( + await _channel.BasicPublishAsync( exchange: _config.ExchangeName, routingKey: _config.RoutingKey, mandatory: false, @@ -144,9 +132,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable public async Task DequeueAsync(CancellationToken cancellationToken = default) { - await EnsureInitializedAsync(); // Initialize on first call - - var result = await _channel!.BasicGetAsync(_config.QueueName, false, cancellationToken); + var result = await _channel.BasicGetAsync(_config.QueueName, false, cancellationToken); if (result == null) return null; @@ -171,50 +157,55 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) { - await EnsureInitializedAsync(); // Initialize on first call - - var queueInfo = await _channel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); return (int)queueInfo.MessageCount; } /// /// Start event-driven consumer that processes messages as they arrive /// - public async Task StartConsumerAsync( - Func onMailReceived, - CancellationToken cancellationToken = default) + private async Task StartConsumerAsync(CancellationToken cancellationToken) { - await EnsureInitializedAsync(); // Initialize on first call - - var consumer = new AsyncEventingBasicConsumer(_channel!); + var consumer = new AsyncEventingBasicConsumer(_channel); consumer.ReceivedAsync += async (sender, args) => { + OutgoingEmailEvent? oMailEvent = null; try { var json = Encoding.UTF8.GetString(args.Body.ToArray()); - var email = JsonSerializer.Deserialize(json); + oMailEvent = JsonSerializer.Deserialize(json); - if (email != null) + if (oMailEvent is not null) { - _logger.LogDebug("Received email message: To={To}, Subject={Subject}", email.Recipient, email.Subject); + _logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); - // Process message via callback - await onMailReceived(email); + _logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}", + oMailEvent.Recipient, oMailEvent.Subject); + + // Send email via SMTP (SMTP config is injected in IEmailService via IOptions) + await _emailService.SendEmailAsync( + oMailEvent.Recipient, + oMailEvent.Subject, + oMailEvent.Body, + isHtml: oMailEvent.IsHtml); + + _logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", + oMailEvent.Recipient, oMailEvent.Subject); // Acknowledge message after successful processing - await _channel.BasicAckAsync(args.DeliveryTag, false, cancellationToken); + await _channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); _logger.LogDebug("Message acknowledged: DeliveryTag={DeliveryTag}", args.DeliveryTag); } else { _logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); - await _channel.BasicNackAsync(args.DeliveryTag, false, false, cancellationToken); // Don't requeue invalid messages + await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages } } catch (Exception ex) { - _logger.LogError(ex, "Failed to process email message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", args.DeliveryTag); + _logger.LogError(ex, "Failed to process email [To={To}, Subject={Subject}] message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", oMailEvent?.Recipient, oMailEvent?.Subject, args.DeliveryTag); // TODO: Error Reporting Strategy // Option 1: Separate RabbitMQ Queue (emailprofiler.errors) @@ -237,7 +228,7 @@ public class RabbitMqEmailQueue : IEmailQueue, IDisposable // - Real-time alerting via monitoring worker // NO RETRY - All failures move directly to DLQ - await _channel.BasicNackAsync(args.DeliveryTag, false, false, cancellationToken); // requeue=false → DLQ + await _channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ } };