hrynco-notification-service/HrynCo.NotificationService.Worker/SendEmailConsumer.cs

namespace HrynCo.NotificationService.Worker;

using HrynCo.NotificationService.Contracts.Messages;
using HrynCo.NotificationService.Worker.Services.EmailProcessing;
using Hrynco.RabbitMq;
using Microsoft.Extensions.Options;

public sealed class SendEmailConsumer : RabbitMqConsumerBase<SendEmailMessage, SendEmailMessageData>
{
    internal const string IncomingQueue = "notification.send-email";

    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<SendEmailConsumer> _logger;

    public SendEmailConsumer(
        IOptionsMonitor<RabbitMqSettings> options,
        IServiceScopeFactory scopeFactory,
        ILogger<SendEmailConsumer> logger)
        : base(options, logger)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;
    }

    protected override string QueueName => IncomingQueue;

    protected override bool TryValidateMessage(
        SendEmailMessage message,
        RabbitMqMessageContext context,
        out string? validationError)
    {
        validationError = SendEmailDeliveryValidator.GetValidationError(
            message,
            context,
            SupportedMessageType.SendEmailV1);
        return validationError is null;
    }

    protected override async Task HandleMessageAsync(
        SendEmailMessage message,
        RabbitMqMessageContext context,
        CancellationToken cancellationToken)
    {
        string payloadCorrelationId = message.CorrelationContext.CorrelationId;
        if (!string.IsNullOrWhiteSpace(context.CorrelationId) &&
            !string.Equals(context.CorrelationId, payloadCorrelationId, StringComparison.Ordinal))
        {
            _logger.LogWarning(
                "Broker and payload correlation IDs differ; payload correlation ID will be used");
        }

        await using AsyncServiceScope scope = _scopeFactory.CreateAsyncScope();
        ISendEmailService service = scope.ServiceProvider.GetRequiredService<ISendEmailService>();
        await service.ProcessAsync(message, cancellationToken);
    }

    protected override async Task HandleMessageRetriesExhaustedAsync(
        SendEmailMessage message,
        RabbitMqMessageContext context,
        Exception exception,
        CancellationToken cancellationToken)
    {
        string deliveryError = NotificationDeliveryErrorFormatter.Format(exception);
        await using AsyncServiceScope scope = _scopeFactory.CreateAsyncScope();
        INotificationResultPublisher resultPublisher =
            scope.ServiceProvider.GetRequiredService<INotificationResultPublisher>();
        await resultPublisher.PublishAsync(message, deliveryError, cancellationToken);
    }
}