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 { internal const string IncomingQueue = "notification.send-email"; internal const string SupportedMessageType = "Notification.SendEmail.v1"; private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; public SendEmailConsumer( IOptionsMonitor options, IServiceScopeFactory scopeFactory, ILogger 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); 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(); 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(); await resultPublisher.PublishAsync(message, deliveryError, cancellationToken); } }