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 _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 _queues = []; private IBus? _bus; public ReceiverService( IObservationService observationService, ITreatmentService treatmentService, IPatientService patientsService, IPumpService pumpService, IRecordingAlertService recordingAlertService, IRecordingService recordingService, IAppointmentService appointmentService, IAlarmService alarmService, IOptions settings, ILogger 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(); 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(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( 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(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( err.Message, new MessageProperties() ); await Task.Run(() => TryToParseAndSend(message, service)); } catch (Exception ex) { _logger.LogError(ex, "Error reprocessing message from {queue}", errorQueueName); throw; } }); } /// /// Sends EasyNetQ messages to a service implementing IApiRequestService /// Throws exception to trigger retry on failure /// private void TryToParseAndSend(IMessage msg, IApiRequestService service) { try { _logger.LogDebug("Processing message: {msg}", msg.Body); var apiRequest = JsonConvert.DeserializeObject(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; } } }