using System.Text;
using System.Text.Json;
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.Publisher.Abstraction;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
namespace DigitalData.MessagingService.Infrastructure.Queue;
///
/// RabbitMQ-based email queue implementation for outgoing emails.
/// Provides message persistence, scalability, and reliability.
/// Uses Lazy initialization pattern to avoid blocking constructor.
///
public sealed class OutgoingEmailConsumer : IAsyncDisposable
{
private readonly RabbitMqConfiguration _config;
private readonly Lazy> _lazyChannel;
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
/// Called lazily on first use via EnsureInitializedAsync.
///
public async Task InitAsync()
{
if (_lazyInit.IsValueCreated)
_logger?.LogWarning("OutgoingEmailConsumer already initialized. InitAsync() called multiple times.");
await _lazyInit.Value;
}
public async ValueTask DisposeAsync()
{
if (!_lazyChannel.IsValueCreated)
return;
var channel = await _lazyChannel.Value;
if (channel is not null)
{
await channel.CloseAsync();
await channel.DisposeAsync();
}
}
}