using System.Text;
using System.Text.Json;
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using DigitalData.MessagingService.Application.Common.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(IOptions config, ILogger Logger, IEmailService EmailService, RabbitMqConnectionFactory CnnFactory) : IAsyncDisposable
{
private readonly RabbitMqConfiguration _config = config.Value;
private IChannel? _consumeChannel = null; // Dedicated channel for consuming
private readonly CancellationTokenSource _consumerCts = new(); // Independent lifetime from InitAsync token
///
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
///
public async Task InitAsync()
{
var connection = await CnnFactory.GetDefaultConnectionAsync();
var stoppingToken = CnnFactory.CancellationToken;
// Separate channels: IChannel is not thread-safe; one per role is best practice
_consumeChannel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);
// Consumer uses its own CancellationToken independent of the startup token,
// so it keeps running after InitAsync completes or its token is cancelled.
// Link stoppingToken so the consumer stops when the host stops.
stoppingToken.Register(_consumerCts.Cancel);
await StartConsumerAsync(_consumeChannel, _consumerCts.Token);
Logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
}
///
/// Start event-driven consumer that processes messages as they arrive
///
private async Task StartConsumerAsync(IChannel consumeChannel, CancellationToken cancellationToken)
{
var consumer = new AsyncEventingBasicConsumer(consumeChannel);
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 consumeChannel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken);
}
else
{
Logger.LogWarning("Failed to deserialize email message: DeliveryTag={DeliveryTag}", args.DeliveryTag);
await consumeChannel.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 consumeChannel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ
}
};
// Start consuming messages (event-driven, non-blocking)
await consumeChannel.BasicConsumeAsync(
queue: _config.QueueName,
autoAck: false,
consumer: consumer,
cancellationToken: cancellationToken);
Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
}
public async ValueTask DisposeAsync()
{
await _consumerCts.CancelAsync();
_consumerCts.Dispose();
if (_consumeChannel is not null)
{
await _consumeChannel.CloseAsync();
await _consumeChannel.DisposeAsync();
}
}
}