src/Services/NotificationService.PushService/Workers/PushWorker.cs - master - VFlov/NotificationService_1

src/Services/NotificationService.PushService/Workers/PushWorker.cs
151 строка6 KB
VFlov
Update
25 дек 2025, 21:02
25 дек 2025, 21:02cf3965d
using NotificationService.Shared.Constants;
using NotificationService.Shared.Interfaces;
using NotificationService.Shared.Models;
using System.Diagnostics;

namespace NotificationService.PushService.Workers;

public class PushWorker : BackgroundService
{
    private readonly ILogger<PushWorker> _logger;
    private readonly IServiceProvider _serviceProvider;

    public PushWorker(ILogger<PushWorker> logger, IServiceProvider serviceProvider)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Push Worker starting...");

        try
        {
            using var scope = _serviceProvider.CreateScope();
            var messageQueue = scope.ServiceProvider.GetRequiredService<IMessageQueue>();
            var repository = scope.ServiceProvider.GetRequiredService<INotificationRepository>();
            var pushSender = scope.ServiceProvider.GetRequiredService<INotificationSender>();

            _logger.LogInformation("Push Worker started successfully, subscribing to queue: {Queue}", QueueNames.PushQueue);

            await messageQueue.SubscribeAsync<NotificationMessage>(
                QueueNames.PushQueue,
                async (message) => await ProcessMessageAsync(message, repository, pushSender, stoppingToken),
                stoppingToken);

            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (Exception ex)
        {
            _logger.LogCritical(ex, "Push Worker failed to start");
            throw;
        }
    }

    private async Task<bool> ProcessMessageAsync(
        NotificationMessage message,
        INotificationRepository repository,
        INotificationSender pushSender,
        CancellationToken cancellationToken)
    {
        var stopwatch = Stopwatch.StartNew();

        try
        {
            if (message == null)
            {
                _logger.LogWarning("Received null message from queue");
                return false;
            }

            _logger.LogInformation(
                "Processing Push notification - Id: {Id}, Recipient: {Recipient}",
                message.Id, message.Recipient);

            var record = await repository.GetByIdAsync(message.Id, cancellationToken);
            if (record == null)
            {
                _logger.LogWarning(
                    "Notification record not found in database - Id: {Id}",
                    message.Id);
                return false;
            }

            record.Status = NotificationStatus.Processing;
            await repository.UpdateAsync(record, cancellationToken);

            _logger.LogDebug("Sending Push notification - Id: {Id}", message.Id);

            var success = await pushSender.SendAsync(message, cancellationToken);

            stopwatch.Stop();

            if (success)
            {
                record.Status = NotificationStatus.Sent;
                record.SentAt = DateTime.UtcNow;
                record.ErrorMessage = null;
                
                _logger.LogInformation(
                    "Push notification sent successfully - Id: {Id}, Duration: {Duration}ms",
                    message.Id, stopwatch.ElapsedMilliseconds);
            }
            else
            {
                record.RetryCount++;
                
                if (record.RetryCount < message.MaxRetries)
                {
                    record.Status = NotificationStatus.Retry;
                    _logger.LogWarning(
                        "Push notification failed, will retry - Id: {Id}, RetryCount: {RetryCount}/{MaxRetries}, Duration: {Duration}ms",
                        message.Id, record.RetryCount, message.MaxRetries, stopwatch.ElapsedMilliseconds);
                }
                else
                {
                    record.Status = NotificationStatus.Failed;
                    record.ErrorMessage = "Failed to send Push notification after maximum retries";
                    
                    _logger.LogError(
                        "Push notification failed permanently - Id: {Id}, MaxRetries: {MaxRetries}, Duration: {Duration}ms",
                        message.Id, message.MaxRetries, stopwatch.ElapsedMilliseconds);
                }
            }

            await repository.UpdateAsync(record, cancellationToken);
            return success;
        }
        catch (Exception ex)
        {
            stopwatch.Stop();
            
            _logger.LogError(ex,
                "Error processing Push notification - Id: {Id}, Duration: {Duration}ms",
                message?.Id, stopwatch.ElapsedMilliseconds);
            
            if (message != null)
            {
                try
                {
                    var record = await repository.GetByIdAsync(message.Id, cancellationToken);
                    if (record != null)
                    {
                        record.Status = NotificationStatus.Failed;
                        record.ErrorMessage = ex.Message;
                        record.RetryCount++;
                        await repository.UpdateAsync(record, cancellationToken);
                    }
                }
                catch (Exception updateEx)
                {
                    _logger.LogError(updateEx,
                        "Failed to update notification record after error - Id: {Id}",
                        message.Id);
                }
            }

            return false;
        }
    }
}