diff --git a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs index 89026cb..cd334df 100644 --- a/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs +++ b/src/DigitalData.MessagingService.Infrastructure/Queue/OutgoingEmailConsumer.cs @@ -15,14 +15,95 @@ namespace DigitalData.MessagingService.Infrastructure.Queue; /// Provides message persistence, scalability, and reliability. /// Uses Lazy initialization pattern to avoid blocking constructor. /// -public sealed class OutgoingEmailConsumer(IOptions config, ILogger Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable +public sealed class OutgoingEmailConsumer : IAsyncDisposable { - private readonly RabbitMqConfiguration _config = config.Value; + private readonly RabbitMqConfiguration _config; - private readonly Lazy> _lazyChannel = new(CnnFactory.CreateChannelAsync); + private readonly Lazy> _lazyChannel; private readonly AsyncEventingBasicConsumer? consumer; + private readonly Lazy _lazyInit; + + private readonly ILogger? _logger; + + public OutgoingEmailConsumer(IOptions config, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory, ILogger? logger = null) + { + _logger = logger; + _config = config.Value; + + _lazyChannel = new(CnnFactory.CreateChannelAsync); + _lazyInit = new(async () => { + var channel = await _lazyChannel.Value; + + var consumer = new AsyncEventingBasicConsumer(channel); + + consumer.ReceivedAsync += async (sender, args) => + { + OutgoingEmailEvent? oMailEvent = null; + try + { + var json = Encoding.UTF8.GetString(args.Body.ToArray()); + oMailEvent = JsonSerializer.Deserialize(json); + + if (oMailEvent is not null) + { + // Send email via SMTP (SMTP config is injected in IEmailService via IOptions) + await EmailService.SendEmailAsync( + oMailEvent.Recipient, + oMailEvent.Subject, + oMailEvent.Body, + isHtml: oMailEvent.IsHtml); + + // Acknowledge message after successful processing + await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); + } + else + { + logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); + await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages + } + } + catch (Exception ex) + { + 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) + // - Create EmailErrorReport entity { OutgoingEmailEventId, Exception, StackTrace, Timestamp, RetryAttempt } + // - Publish to error queue: await _errorQueue.EnqueueAsync(errorReport) + // - Separate worker processes error queue → Log to DB/File/External monitoring + // + // Option 2: Database Table (TBEMLP_ERROR_LOG) + // - Columns: ERROR_ID, OUTBOX_ID, ERROR_MESSAGE, STACK_TRACE, ERROR_DATE + // - Insert via IErrorLogRepository.CreateAsync(errorLog) + // + // Option 3: External Monitoring Service + // - Sentry: SentrySdk.CaptureException(ex) + // - Application Insights: _telemetryClient.TrackException(ex) + // - Elasticsearch: _elasticClient.IndexDocument(errorLog) + // + // Recommended: Option 1 (RabbitMQ Error Queue) + Option 2 (DB persistence) + // - Fast async error logging (non-blocking) + // - Persistent storage for audit + // - Real-time alerting via monitoring worker + + // NO RETRY - All failures move directly to DLQ + await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ + } + }; + + // Start consuming messages (event-driven, non-blocking) + await channel.BasicConsumeAsync( + queue: _config.QueueName, + autoAck: false, + consumer: consumer, + cancellationToken: CnnFactory.CancellationToken); + + logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); + }); + } + /// /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously. /// Start event-driven consumer that processes messages as they arrive @@ -30,73 +111,10 @@ public sealed class OutgoingEmailConsumer(IOptions config /// public async Task InitAsync() { - var channel = await _lazyChannel.Value; + if (_lazyInit.IsValueCreated) + _logger?.LogWarning("OutgoingEmailConsumer already initialized. InitAsync() called multiple times."); - var consumer = new AsyncEventingBasicConsumer(channel); - - consumer.ReceivedAsync += async (sender, args) => - { - OutgoingEmailEvent? oMailEvent = null; - try - { - var json = Encoding.UTF8.GetString(args.Body.ToArray()); - oMailEvent = JsonSerializer.Deserialize(json); - - if (oMailEvent is not null) - { - // Send email via SMTP (SMTP config is injected in IEmailService via IOptions) - await EmailService.SendEmailAsync( - oMailEvent.Recipient, - oMailEvent.Subject, - oMailEvent.Body, - isHtml: oMailEvent.IsHtml); - - // Acknowledge message after successful processing - await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); - } - else - { - Logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag); - await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // Don't requeue invalid messages - } - } - catch (Exception ex) - { - 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) - // - Create EmailErrorReport entity { OutgoingEmailEventId, Exception, StackTrace, Timestamp, RetryAttempt } - // - Publish to error queue: await _errorQueue.EnqueueAsync(errorReport) - // - Separate worker processes error queue → Log to DB/File/External monitoring - // - // Option 2: Database Table (TBEMLP_ERROR_LOG) - // - Columns: ERROR_ID, OUTBOX_ID, ERROR_MESSAGE, STACK_TRACE, ERROR_DATE - // - Insert via IErrorLogRepository.CreateAsync(errorLog) - // - // Option 3: External Monitoring Service - // - Sentry: SentrySdk.CaptureException(ex) - // - Application Insights: _telemetryClient.TrackException(ex) - // - Elasticsearch: _elasticClient.IndexDocument(errorLog) - // - // Recommended: Option 1 (RabbitMQ Error Queue) + Option 2 (DB persistence) - // - Fast async error logging (non-blocking) - // - Persistent storage for audit - // - Real-time alerting via monitoring worker - - // NO RETRY - All failures move directly to DLQ - await channel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ - } - }; - - // Start consuming messages (event-driven, non-blocking) - await channel.BasicConsumeAsync( - queue: _config.QueueName, - autoAck: false, - consumer: consumer, - cancellationToken: CnnFactory.CancellationToken); - - Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); + await _lazyInit.Value; } public async ValueTask DisposeAsync()