adas-core/adas-core.Infrastructure/Services/ReceiverService.cs

using adas_core.Application.Services.Interfaces;
using adas_core.Domain.Models;
using adas_core.Domain.Models.AppSettings;
using EasyNetQ;
using EasyNetQ.SystemMessages;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;

namespace adas_core.Infrastructure.Services;

public class ReceiverService
{
    private readonly ILogger<ReceiverService> _logger;
    private readonly RabbitMqSettings _settings;

    private readonly IObservationService _observationService;
    private readonly ITreatmentService _treatmentService;
    private readonly IPatientService _patientsService;
    private readonly IPumpService _pumpService;
    private readonly IRecordingAlertService _recordingAlertService;
    private readonly IRecordingService _recordingService;
    private readonly IAppointmentService _appointmentService;
    private readonly IAlarmService _alarmService;

    private readonly List<string> _queues = [];

    private IBus? _bus;

    public ReceiverService(
        IObservationService observationService,
        ITreatmentService treatmentService,
        IPatientService patientsService,
        IPumpService pumpService,
        IRecordingAlertService recordingAlertService,
        IRecordingService recordingService,
        IAppointmentService appointmentService,
        IAlarmService alarmService,
        IOptions<RabbitMqSettings> settings,
        ILogger<ReceiverService> logger)
    {
        _logger = logger;
        _settings = settings.Value;

        _observationService = observationService;
        _treatmentService = treatmentService;
        _patientsService = patientsService;
        _pumpService = pumpService;
        _recordingAlertService = recordingAlertService;
        _recordingService = recordingService;
        _appointmentService = appointmentService;
        _alarmService = alarmService;

        if (!string.IsNullOrWhiteSpace(_settings.ConnectionString))
        {
            SetQueues();
            TryToConnect();
        }
        else
        {
            _logger.LogError("RabbitMQ connection string NOT configured");
        }
    }

    private void SetQueues()
    {
        AddQueue(_settings.ObservationsQueue);
        AddQueue(_settings.TreatmentsQueue);
        AddQueue(_settings.PatientsQueue);
        AddQueue(_settings.PumpsQueue);
        AddQueue(_settings.AppointmentsQueue);
        AddQueue(_settings.RecordingQueue);
        AddQueue(_settings.RecordingAlertQueue);
        AddQueue(_settings.AlarmObservationQueue);
    }

    private void AddQueue(string? queue)
    {
        if (!string.IsNullOrWhiteSpace(queue))
            _queues.Add(queue!);
    }

    private void TryToConnect()
    {
        try
        {
            DisposeBus();

            var services = new ServiceCollection();
            services.AddEasyNetQ(_settings.ConnectionString);
            var provider = services.BuildServiceProvider();

            _bus = provider.GetRequiredService<IBus>();


            foreach (var queue in _queues)
            {
                RegisterConsumer(queue);
            }

            _logger.LogInformation("RabbitMQ connected. Registered {Count} queues", _queues.Count);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "RabbitMQ connection failed. Retrying in 5 seconds...");

            var timer = new System.Timers.Timer(5000);
            timer.Elapsed += (_, _) =>
            {
                timer.Stop();
                TryToConnect();
            };
            timer.Start();
        }
    }

    private void DisposeBus()
    {
        if (_bus is IDisposable disposable)
        {
            disposable.Dispose();
        }
    }

    private void RegisterConsumer(string queue)
    {
        _bus!.SendReceive.ReceiveAsync<string>(queue, async payload =>
        {
            try
            {
                _logger.LogDebug("Message received from {queue}", queue);

                var service = ResolveService(queue);

                if (service == null)
                {
                    _logger.LogWarning("No service mapped for queue {queue}", queue);
                    return;
                }

                var message = new Message<string>(
                    payload,
                    new MessageProperties()
                );

                await Task.Run(() => TryToParseAndSend(message, service));
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error processing message from {queue}", queue);
                throw; // 🔴 importante: activa retry
            }
        });
    }

    private IApiRequestService? ResolveService(string queue)
    {
        return queue switch
        {
            var q when q == _settings.ObservationsQueue => _observationService,
            var q when q == _settings.TreatmentsQueue => _treatmentService,
            var q when q == _settings.PatientsQueue => _patientsService,
            var q when q == _settings.PumpsQueue => _pumpService,
            var q when q == _settings.RecordingQueue => _recordingService,
            var q when q == _settings.RecordingAlertQueue => _recordingAlertService,
            var q when q == _settings.AlarmObservationQueue => _alarmService,
            var q when q == _settings.AppointmentsQueue => _appointmentService,
            _ => null
        };
    }

    public async Task ProcessErrorQueue(string errorQueueName)
    {
        if (_bus == null)
        {
            _logger.LogError("Cannot process error queue - bus is null");
            return;
        }

        await _bus.SendReceive.ReceiveAsync<Error>(errorQueueName, async err =>
        {
            try
            {
                var queue = err.RoutingKey?.Replace("Key", "");

                if (string.IsNullOrEmpty(queue))
                    return;

                var service = ResolveService(queue);

                if (service == null)
                    return;

                var message = new Message<string>(
                    err.Message,
                    new MessageProperties()
                );

                await Task.Run(() => TryToParseAndSend(message, service));
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error reprocessing message from {queue}", errorQueueName);
                throw;
            }
        });
    }

    /// <summary>
    /// Sends EasyNetQ messages to a service implementing IApiRequestService
    /// Throws exception to trigger retry on failure
    /// </summary>
    private void TryToParseAndSend(IMessage<string> msg, IApiRequestService service)
    {
        try
        {
            _logger.LogDebug("Processing message: {msg}", msg.Body);

            var apiRequest = JsonConvert.DeserializeObject<ApiRequest>(msg.Body);

            if (apiRequest == null)
            {
                _logger.LogError("Invalid ApiRequest JSON: {msg}", msg.Body);
                return;
            }

            if (apiRequest.PatientNumber == "\"\"")
                apiRequest.PatientNumber = string.Empty;

            service.SaveRequest(apiRequest);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error processing request");
            throw;
        }
    }
}