DigitalData.MessagingService/src/infrastructure/DigitalData.MessagingService.Infrastructure/Queue/SendingEmailConsumer.cs

using System.Text;
using System.Text.Json;
using DigitalData.MessagingService.Application.Common.Interfaces;
using DigitalData.MessagingService.RabbitMQ;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using DigitalData.MessagingService.Application.Common.Dto;

namespace DigitalData.MessagingService.Infrastructure.Queue;

/// <summary>
/// A single RabbitMQ consumer that processes one email message at a time on its own dedicated channel.
/// Multiple instances run in parallel via <see cref="SendingEmailConsumerPool"/> (competing consumers pattern).
/// Each instance owns exactly one channel — channels are not thread-safe and must not be shared.
/// </summary>
public sealed class SendingEmailConsumer : IAsyncDisposable
{
    private readonly string _queueName;

    private readonly Lazy<Task<IChannel>> _lazyChannel;

    private readonly Lazy<Task> _lazyInit;

    private readonly ILogger<SendingEmailConsumer>? _logger;

    /// <summary>
    /// Transient identifier assigned to this consumer instance at runtime.
    /// A new value is generated each time the application starts or a new consumer is created.
    /// Use this to correlate log entries belonging to the same consumer session across competing instances.
    /// </summary>
    public Guid RuntimeId { get; } = Guid.NewGuid();

    public SendingEmailConsumer(string queueName, IEmailService emailService, RabbitMqConnectionFactory cnnFactory, ILogger<SendingEmailConsumer>? logger = null)
    {
        _logger = logger;
        _queueName = queueName;

        _lazyChannel = new(cnnFactory.CreateChannelAsync);
        _lazyInit = new(async () =>
        {
            var channel = await _lazyChannel.Value;

            // prefetchCount=1 ensures this consumer processes one message at a time before acking.
            // Parallelism comes from running multiple consumer instances, not from within a single channel.
            await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false);

            var consumer = new AsyncEventingBasicConsumer(channel);

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

                    if (oMailEvent is not null)
                    {
                        // Send email via SMTP (SMTP config is injected in IEmailService via IOptions)
                        await emailService.SendEmailAsync(oMailEvent.Mail, args.CancellationToken);

                        // Acknowledge message after successful processing
                        await channel.BasicAckAsync(args.DeliveryTag, false, args.CancellationToken);

                        logger?.LogDebug(
                            "Email successfully sent and acknowledged. RuntimeId={RuntimeId}, Queue={QueueName}, DeliveryTag={DeliveryTag}, To={Recipients}, Subject={Subject}, EventId={EventId}",
                            RuntimeId, _queueName, args.DeliveryTag, oMailEvent.Mail.Recipients, oMailEvent.Mail.Subject, oMailEvent.Id);
                    }
                    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?.Mail.Recipients, oMailEvent?.Mail.Subject, args.DeliveryTag);

                    // TODO: Error Reporting Strategy
                    // Option 1: Separate RabbitMQ Queue (emailprofiler.errors)
                    //   - Create EmailErrorReport entity { SendingEmailEventId, 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: _queueName,
                autoAck: false,
                consumer: consumer,
                cancellationToken: cnnFactory.CancellationToken);

            logger?.LogInformation("RabbitMQ consumer started for queue: {QueueName}", _queueName);
        });
    }

    /// <summary>
    /// Starts the consumer: opens a channel, sets QoS, and registers the event handler.
    /// Called by <see cref="SendingEmailConsumerPool.InitAsync"/>.
    /// </summary>
    public async Task InitAsync()
    {
        if (_lazyInit.IsValueCreated)
            _logger?.LogWarning("SendingEmailConsumer 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();
        }
    }
}