diff --git a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs index f4f7f50..ba33c3e 100644 --- a/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs +++ b/src/DigitalData.EmailProfiler.Infrastructure/Queue/OutgoingEmailQueue.cs @@ -1,7 +1,6 @@ using System.Text; using System.Text.Json; using DigitalData.EmailProfiler.Application.Common.Interfaces; -using DigitalData.EmailProfiler.Application.EmailSending.Commands; using DigitalData.EmailProfiler.Infrastructure.Messaging; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -16,30 +15,21 @@ namespace DigitalData.EmailProfiler.Infrastructure.Queue; /// Provides message persistence, scalability, and reliability. /// Uses Lazy initialization pattern to avoid blocking constructor. /// -public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable +public sealed class OutgoingEmailQueue(IOptions config, ILogger Logger, IEmailService EmailService) : IOutgoingEmailQueue, IAsyncDisposable { - private readonly ILogger _logger; - private readonly RabbitMqConfiguration _config; - private IConnection? _connection; - private IChannel _channel; - private readonly IEmailService _emailService; - -#pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor - public OutgoingEmailQueue(IOptions config, ILogger logger, IEmailService emailService) -#pragma warning restore CS8618 - { - _logger = logger; - _config = config.Value; - _emailService = emailService; - } - + private readonly RabbitMqConfiguration _config = config.Value; + private IConnection? _connection = null; + private IChannel? _publishChannel = null; // Dedicated channel for publishing + 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(CancellationToken cancellationToken = default) + public async Task InitAsync(CancellationToken stoppingToken = default) { - _logger.LogInformation("Initializing RabbitMQ connection and queues..."); + Logger.LogInformation("Initializing RabbitMQ connection and queues..."); var factory = new ConnectionFactory { @@ -52,36 +42,24 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds) }; - _connection = await factory.CreateConnectionAsync(cancellationToken); - _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken); + _connection = await factory.CreateConnectionAsync(stoppingToken); + // Separate channels: IChannel is not thread-safe; one per role is best practice + _publishChannel = await _connection.CreateChannelAsync(cancellationToken: stoppingToken); + _consumeChannel = await _connection.CreateChannelAsync(cancellationToken: stoppingToken); + + // Topology declaration can use either channel; use publish channel here // Declare Dead Letter Queue (DLQ) exchange - await _channel.ExchangeDeclareAsync( - exchange: _config.DlqExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false); + await _publishChannel.ExchangeDeclareAsync(exchange: _config.DlqExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken); // Declare Dead Letter Queue (DLQ) - await _channel.QueueDeclareAsync( - queue: _config.DlqQueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: null); + await _publishChannel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: stoppingToken); // Bind DLQ to DLQ exchange - await _channel.QueueBindAsync( - queue: _config.DlqQueueName, - exchange: _config.DlqExchangeName, - routingKey: _config.DlqRoutingKey); + await _publishChannel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: stoppingToken); // Declare main exchange (Direct type for routing) - await _channel.ExchangeDeclareAsync( - exchange: _config.ExchangeName, - type: ExchangeType.Direct, - durable: true, - autoDelete: false); + await _publishChannel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: stoppingToken); // Declare main queue (durable for persistence) with DLQ arguments var queueArgs = new Dictionary @@ -90,22 +68,18 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable { "x-dead-letter-routing-key", _config.DlqRoutingKey } }; - await _channel.QueueDeclareAsync( - queue: _config.QueueName, - durable: true, - exclusive: false, - autoDelete: false, - arguments: queueArgs); + await _publishChannel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: stoppingToken); // Bind main queue to exchange with routing key - await _channel.QueueBindAsync( - queue: _config.QueueName, - exchange: _config.ExchangeName, - routingKey: _config.RoutingKey); + await _publishChannel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: stoppingToken); - await StartConsumerAsync(cancellationToken); + // 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); + Logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName); } public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default) @@ -120,7 +94,7 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()) }; - await _channel.BasicPublishAsync( + await _publishChannel!.BasicPublishAsync( exchange: _config.ExchangeName, routingKey: _config.RoutingKey, mandatory: false, @@ -131,16 +105,16 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default) { - var queueInfo = await _channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); + var queueInfo = await _publishChannel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken); return (int)queueInfo.MessageCount; } /// /// Start event-driven consumer that processes messages as they arrive /// - private async Task StartConsumerAsync(CancellationToken cancellationToken) + private async Task StartConsumerAsync(IChannel consumeChannel, CancellationToken cancellationToken) { - var consumer = new AsyncEventingBasicConsumer(_channel); + var consumer = new AsyncEventingBasicConsumer(consumeChannel); consumer.ReceivedAsync += async (sender, args) => { @@ -152,34 +126,34 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable if (oMailEvent is not null) { - _logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); + Logger.LogDebug("Received email message: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); - _logger.LogInformation("Processing outgoing email: To={To}, Subject={Subject}", + 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( + await EmailService.SendEmailAsync( oMailEvent.Recipient, oMailEvent.Subject, oMailEvent.Body, isHtml: oMailEvent.IsHtml); - _logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", + Logger.LogInformation("Email sent successfully: To={To}, Subject={Subject}", oMailEvent.Recipient, oMailEvent.Subject); // Acknowledge message after successful processing - await _channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken); - _logger.LogDebug("Message acknowledged: DeliveryTag={DeliveryTag}", args.DeliveryTag); + await consumeChannel.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, args.CancellationToken); // Don't requeue invalid messages + 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); + 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) @@ -202,25 +176,42 @@ public class OutgoingEmailQueue : IOutgoingEmailQueue, IDisposable // - 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 + await consumeChannel.BasicNackAsync(args.DeliveryTag, false, false, args.CancellationToken); // requeue=false → DLQ } }; // Start consuming messages (event-driven, non-blocking) - await _channel.BasicConsumeAsync( + await consumeChannel.BasicConsumeAsync( queue: _config.QueueName, autoAck: false, consumer: consumer, cancellationToken: cancellationToken); - _logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); + Logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName); } - public void Dispose() + public async ValueTask DisposeAsync() { - _channel?.CloseAsync().GetAwaiter().GetResult(); - _channel?.Dispose(); - _connection?.CloseAsync().GetAwaiter().GetResult(); - _connection?.Dispose(); + await _consumerCts.CancelAsync(); + _consumerCts.Dispose(); + + if (_consumeChannel is not null) + { + await _consumeChannel.CloseAsync(); + await _consumeChannel.DisposeAsync(); + } + + if (_publishChannel is not null) + { + await _publishChannel.CloseAsync(); + await _publishChannel.DisposeAsync(); + } + + if (_connection is not null) + { + await _connection.CloseAsync(); + await _connection.DisposeAsync(); + } } + }