using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using DigitalData.MessagingService.RabbitMQ;
using DigitalData.MessagingService.Application.Common.Dto;
using DigitalData.MessagingService.Application.Common.Interfaces;
namespace DigitalData.MessagingService.Publisher;
///
/// 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 SendingEmailPublisher : ISendingEmailPublisher, IAsyncDisposable
{
private readonly RabbitMqConfiguration _config;
private readonly ILogger _logger;
private readonly RabbitMqConnectionFactory _cnnFactory;
private readonly Lazy> _lazyChannel;
public SendingEmailPublisher(IOptions config, ILogger logger, RabbitMqConnectionFactory cnnFactory)
{
_config = config.Value;
_logger = logger;
_cnnFactory = cnnFactory;
_lazyChannel = new(InitChannelAsync);
}
///
/// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
/// Called lazily on first use via EnsureInitializedAsync.
///
private async Task InitChannelAsync()
{
var channel = await _cnnFactory.CreateChannelAsync();
// 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, cancellationToken: _cnnFactory.CancellationToken);
// Declare Dead Letter Queue (DLQ)
await channel.QueueDeclareAsync(queue: _config.DlqQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: _cnnFactory.CancellationToken);
// Bind DLQ to DLQ exchange
await channel.QueueBindAsync(queue: _config.DlqQueueName, exchange: _config.DlqExchangeName, routingKey: _config.DlqRoutingKey, cancellationToken: _cnnFactory.CancellationToken);
// Declare main exchange (Direct type for routing)
await channel.ExchangeDeclareAsync(exchange: _config.ExchangeName, type: ExchangeType.Direct, durable: true, autoDelete: false, cancellationToken: _cnnFactory.CancellationToken);
// Declare main queue (durable for persistence) with DLQ arguments
var queueArgs = new Dictionary
{
{ "x-dead-letter-exchange", _config.DlqExchangeName },
{ "x-dead-letter-routing-key", _config.DlqRoutingKey }
};
await channel.QueueDeclareAsync(queue: _config.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: queueArgs, cancellationToken: _cnnFactory.CancellationToken);
// Bind main queue to exchange with routing key
await channel.QueueBindAsync(queue: _config.QueueName, exchange: _config.ExchangeName, routingKey: _config.RoutingKey, cancellationToken: _cnnFactory.CancellationToken);
_logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
return channel;
}
public async Task EnqueueAsync(SendingEmailEvent sendingEmailEvent, CancellationToken cancellationToken = default)
{
var json = JsonSerializer.Serialize(sendingEmailEvent);
var body = Encoding.UTF8.GetBytes(json);
var properties = new BasicProperties
{
Persistent = true, // Message persistence
ContentType = "application/json",
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
};
var channel = await _lazyChannel.Value;
await channel.BasicPublishAsync(
exchange: _config.ExchangeName,
routingKey: _config.RoutingKey,
mandatory: false,
basicProperties: properties,
body: body,
cancellationToken: cancellationToken);
}
public async Task GetQueueDepthAsync(CancellationToken cancellationToken = default)
{
var channel = await _lazyChannel.Value;
var queueInfo = await channel.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
return (int)queueInfo.MessageCount;
}
public async ValueTask DisposeAsync()
{
if (await _lazyChannel.Value is IChannel channel)
{
await channel.CloseAsync();
await channel.DisposeAsync();
}
}
}