DigitalData.MessagingService/src/DigitalData.EmailProfiler.Infrastructure/Queue/RabbitMqEmailQueue.cs

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;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using DigitalData.EmailProfiler.Application.Common.Events;

namespace DigitalData.EmailProfiler.Infrastructure.Queue;

/// <summary>
/// RabbitMQ-based email queue implementation for outgoing emails.
/// Provides message persistence, scalability, and reliability.
/// Uses Lazy<T> initialization pattern to avoid blocking constructor.
/// </summary>
public class RabbitMqEmailQueue : IEmailQueue, IDisposable
{
    private readonly ILogger<RabbitMqEmailQueue> _logger;
    private readonly RabbitMqConfiguration _config;
    private IConnection? _connection;
    private IChannel _channel;
    
    // Lazy<T> ensures InitAsync is called only ONCE (thread-safe)
    private readonly Lazy<Task> _initializationTask;
    
#pragma warning disable CS8618 // channel and connection are initialized in InitAsync, not in constructor
    public RabbitMqEmailQueue(IOptions<RabbitMqConfiguration> config, ILogger<RabbitMqEmailQueue> logger)
#pragma warning restore CS8618
    {
        _logger = logger;
        _config = config.Value;
        
        // Lazy initialization - NOT executed until first access (non-blocking constructor)
        _initializationTask = new Lazy<Task>(InitAsync);
    }
    
    /// <summary>
    /// Initialize RabbitMQ connection, channel, exchanges, and queues asynchronously.
    /// Called lazily on first use via EnsureInitializedAsync.
    /// </summary>
    private async Task InitAsync()
    {
        _logger.LogInformation("Initializing RabbitMQ connection and queues...");
        
        var factory = new ConnectionFactory
        {
            HostName = _config.HostName,
            Port = _config.Port,
            UserName = _config.UserName,
            Password = _config.Password,
            VirtualHost = _config.VirtualHost,
            AutomaticRecoveryEnabled = _config.AutomaticRecoveryEnabled,
            NetworkRecoveryInterval = TimeSpan.FromSeconds(_config.NetworkRecoveryIntervalSeconds)
        };

        _connection = await factory.CreateConnectionAsync();
        _channel = await _connection.CreateChannelAsync();

        // Declare Dead Letter Queue (DLQ) exchange
        await _channel.ExchangeDeclareAsync(
            exchange: _config.DlqExchangeName,
            type: ExchangeType.Direct,
            durable: true,
            autoDelete: false);

        // Declare Dead Letter Queue (DLQ)
        await _channel.QueueDeclareAsync(
            queue: _config.DlqQueueName,
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);

        // Bind DLQ to DLQ exchange
        await _channel.QueueBindAsync(
            queue: _config.DlqQueueName,
            exchange: _config.DlqExchangeName,
            routingKey: _config.DlqRoutingKey);

        // Declare main exchange (Direct type for routing)
        await _channel.ExchangeDeclareAsync(
            exchange: _config.ExchangeName,
            type: ExchangeType.Direct,
            durable: true,
            autoDelete: false);

        // Declare main queue (durable for persistence) with DLQ arguments
        var queueArgs = new Dictionary<string, object?>
        {
            { "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);

        // Bind main queue to exchange with routing key
        await _channel.QueueBindAsync(
            queue: _config.QueueName,
            exchange: _config.ExchangeName,
            routingKey: _config.RoutingKey);
            
        _logger.LogInformation("RabbitMQ initialized successfully: Queue={QueueName}, DLQ={DlqQueueName}", _config.QueueName, _config.DlqQueueName);
    }
    
    /// <summary>
    /// Ensure RabbitMQ is initialized before any operation.
    /// Thread-safe and guarantees single initialization via Lazy<T>.
    /// </summary>
    private async Task EnsureInitializedAsync()
    {
        await _initializationTask.Value;
    }

    public async Task EnqueueAsync(OutgoingEmailEvent outgoingEmailEvent, CancellationToken cancellationToken = default)
    {
        await EnsureInitializedAsync(); // Initialize on first call
        
        var json = JsonSerializer.Serialize(outgoingEmailEvent);
        var body = Encoding.UTF8.GetBytes(json);

        var properties = new BasicProperties
        {
            Persistent = true, // Message persistence
            ContentType = "application/json",
            Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds())
        };

        await _channel!.BasicPublishAsync(
            exchange: _config.ExchangeName,
            routingKey: _config.RoutingKey,
            mandatory: false,
            basicProperties: properties,
            body: body,
            cancellationToken: cancellationToken);
    }

    public async Task<OutgoingEmailEvent?> DequeueAsync(CancellationToken cancellationToken = default)
    {
        await EnsureInitializedAsync(); // Initialize on first call
        
        var result = await _channel!.BasicGetAsync(_config.QueueName, false, cancellationToken);

        if (result == null)
            return null;

        try
        {
            var json = Encoding.UTF8.GetString(result.Body.ToArray());
            var email = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);

            // Acknowledge message after successful deserialization
            await _channel.BasicAckAsync(result.DeliveryTag, false, cancellationToken);

            return email;
        }
        catch
        {
            // Reject and requeue message on error
            await _channel.BasicNackAsync(result.DeliveryTag, false, true, cancellationToken);
            throw;
        }
    }

    public async Task<int> GetQueueDepthAsync(CancellationToken cancellationToken = default)
    {
        await EnsureInitializedAsync(); // Initialize on first call
        
        var queueInfo = await _channel!.QueueDeclarePassiveAsync(_config.QueueName, cancellationToken);
        return (int)queueInfo.MessageCount;
    }

    /// <summary>
    /// Start event-driven consumer that processes messages as they arrive
    /// </summary>
    public async Task StartConsumerAsync(
        Func<OutgoingEmailEvent, Task> onMailReceived,
        CancellationToken cancellationToken = default)
    {
        await EnsureInitializedAsync(); // Initialize on first call
        
        var consumer = new AsyncEventingBasicConsumer(_channel!);

        consumer.ReceivedAsync += async (sender, args) =>
        {
            try
            {
                var json = Encoding.UTF8.GetString(args.Body.ToArray());
                var email = JsonSerializer.Deserialize<OutgoingEmailEvent>(json);

                if (email != null)
                {
                    _logger.LogDebug("Received email message: To={To}, Subject={Subject}", email.Recipient, email.Subject);

                    // Process message via callback
                    await onMailReceived(email);

                    // Acknowledge message after successful processing
                    await _channel.BasicAckAsync(args.DeliveryTag, false, 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, cancellationToken); // Don't requeue invalid messages
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to process email message: DeliveryTag={DeliveryTag}. Moving to DLQ (NO retry).", 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, cancellationToken); // requeue=false → DLQ
            }
        };

        // Start consuming messages (event-driven, non-blocking)
        await _channel.BasicConsumeAsync(
            queue: _config.QueueName,
            autoAck: false,
            consumer: consumer,
            cancellationToken: cancellationToken);

        _logger.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _config.QueueName);
    }

    public void Dispose()
    {
        _channel?.CloseAsync().GetAwaiter().GetResult();
        _channel?.Dispose();
        _connection?.CloseAsync().GetAwaiter().GetResult();
        _connection?.Dispose();
    }
}