using adas_core.Application.Services.Interfaces; using adas_core.Domain.Enums; using adas_core.Domain.Models; using adas_core.Domain.Models.GroupedObservations; using adas_core.Domain.Utils; using Microsoft.Extensions.Logging; using MongoDB.Bson; namespace adas_core.Application.Subscriptions; /// /// Provides an implementation of that coordinates subscription handling for grouped observations, integrating caching, client messaging, and logging capabilities. /// /// /// This service depends on an optional cache, a grouped observation service, a logger, and a lazily-initialized client message service to support its subscriber-related operations. /// public class SubscriberGroupedService( ICacheService? cacheService, IGroupedObservationService groupedObservationService, ILogger logger, Lazy clientMessageService) : ISubscriberGroupedService { private List _subscriberGrouped = []; //private static readonly ILogger _logger = Log.ForContext(); public List SubscriberGroupedList { get { lock (_subscriberGrouped) { return _subscriberGrouped; } } set { lock (_subscriberGrouped) { _subscriberGrouped = value; } } } /// /// Returns a thread-safe snapshot of the grouped subscribers as a new list. The internal collection is locked during the copy to ensure consistency and prevent concurrent modification. /// /// A new containing a copy of the grouped subscribers. public List GetGrouped() { lock (_subscriberGrouped) { return _subscriberGrouped.ToList(); } } /// /// Removes all grouped observations associated with the specified patient identifier from the subscriber collection in a thread-safe manner. /// /// The string representation of the patient identifier used to locate and remove the matching grouped observations. public void RemoveGroupedObsByPatientId(string patientId) { lock (_subscriberGrouped) { var itemToRemove = _subscriberGrouped.Where(c => c.PatientId.ToString() == patientId).ToList(); itemToRemove.ForEach(item => _subscriberGrouped.Remove(item)); } } /// /// Check if what clients dont need to get updated with new grouped observations and remove them /// diference between section and box if location is updated doesn't matter if is section or box /// BoxSubscribers don't need to get updated otherwise SectionSubscribers may still need to get updated /// /// objectId on db for Patient /// public void RemoveWsSubscriberByLocation(string patientId, PatientLocation? newLocation) { lock (_subscriberGrouped) { var itemToRemove = _subscriberGrouped.Where(c => c.PatientId.ToString() == patientId).ToList(); foreach (var wsSubscriberGrouped in itemToRemove) { // Make a separate list for wsClients to be removed var wsClientsToRemove = new List(); foreach (var wsClient in wsSubscriberGrouped.WsSubscriber) wsClientsToRemove.Add(wsClient); // Remove the wsClients foreach (var wsClient in wsClientsToRemove) wsSubscriberGrouped.WsSubscriber.Remove(wsClient); CheckEmptySubscriberGroup(wsSubscriberGrouped); } } } /// /// Removes the specified WebSocket subscriber ID from all subscriber groups associated with the given patient ID in a thread-safe manner, and checks whether any resulting group is now empty for further cleanup. /// /// The patient identifier used to locate the subscriber groups to update. /// The WebSocket subscriber ID to remove from the matching subscriber and group collections. public void RemoveWsSubscriberPatientIdAndWsId(string patientId, string wsIdToRemove) { lock (_subscriberGrouped) { var itemToRemove = _subscriberGrouped .Where(c => c.PatientId.ToString() == patientId && c.WsSubscriber.Contains(wsIdToRemove)).ToList(); foreach (var wsSubscriberGrouped in itemToRemove) { // // Make a separate list for wsClients to be removed // var wsClientsToRemove = new List(); // foreach (var wsClient in wsSubscriberGrouped.WsSubscriber) // { // if(wsClient == wsIdToRemove) // wsClientsToRemove.Add(wsClient); // } // // // Remove the wsClients // foreach (var wsClient in wsClientsToRemove) // { // wsSubscriberGrouped.WsSubscriber.Remove(wsClient); // wsSubscriberGrouped.Group.Remove(wsClient); // } wsSubscriberGrouped.WsSubscriber.Remove(wsIdToRemove); wsSubscriberGrouped.Group.Remove(wsIdToRemove); CheckEmptySubscriberGroup(wsSubscriberGrouped); } } } /// /// Check clients using generated grouped observations on disconections /// removing the wole object in case of 0 or /// only the group if any other client is using it /// /// /// Object with properties to generate new grouped observations, clients using those /// grouped obsercations and the generated list of grouped observations /// public List CheckEmptySubscriberGroup(WsSubscriberGrouped? wsSubscriberGrouped) { lock (_subscriberGrouped) { var removed = new List(); if (wsSubscriberGrouped == null) return []; if (wsSubscriberGrouped.WsSubscriber.Count == 0) { //_logger.Debug($"Removing wsSubscriberGrouped from list because client subscriber are empty"); wsSubscriberGrouped.Timer.Stop(); wsSubscriberGrouped.Timer.Dispose(); _subscriberGrouped.Remove(wsSubscriberGrouped); removed.Add( $"ws:{wsSubscriberGrouped.PatientId}-{string.Join(";", wsSubscriberGrouped.Names)}-{wsSubscriberGrouped.Regularity}-{wsSubscriberGrouped.Group.Count}"); cacheService?.DeleteObjectAsync(wsSubscriberGrouped.HashCode); } else { foreach (var kvp in wsSubscriberGrouped.Group.ToList()) if (!wsSubscriberGrouped.WsSubscriber.Any(subscriber => subscriber.Contains(kvp.Key))) { //_logger.Debug($"Removing group {kvp.Key}:{kvp.Value} from list because client subscriber got disconnected but wsSubscriberGrouped still in use with {wsSubscriberGrouped.WsSubscriber.Count()} clients"); wsSubscriberGrouped.Group.Remove(kvp.Key); wsSubscriberGrouped.WsSubscriber.Remove(kvp.Key); removed.Add($"group:{kvp.Key}"); } } return removed; } } /// /// Adds a grouped WebSocket subscriber to the internal collection in a thread-safe manner. /// Uses locking to ensure that concurrent calls do not corrupt the subscriber list. /// /// The grouped WebSocket subscriber instance to be added to the collection. public void AddSubscriberGrouped(WsSubscriberGrouped wsSubscriberGrouped) { lock (_subscriberGrouped) { _subscriberGrouped.Add(wsSubscriberGrouped); } } /// /// Registers a client connection for grouped-field observation updates for a given patient. If an existing grouped subscription matches the patient and field, the connection is added to it; otherwise, a new grouped subscription is created with a callback that processes empty observations. /// /// The grouped field criteria used to locate or create the matching subscription. /// The identifier of the patient whose grouped observation is being subscribed to. /// The optional time zone identifier applied to the new subscription when one is created. /// The unique identifier of the client connection being registered. /// The most recent observation in the group, used to initialize a new subscription. public void CheckOnSubscriptionGroup(GroupedField groupedField, ObjectId patientId, string? timeZoneId, string connectionId, GroupedObservation lastObsInGroup) { var wsSubscriberGrouped = GetGrouped().FirstOrDefault(s => s.Compare(groupedField, patientId)); if (wsSubscriberGrouped != null) { wsSubscriberGrouped.WsSubscriber.Add(connectionId); // Verifica si el diccionario contiene la clave y el valor deseado if (!wsSubscriberGrouped.Group.ContainsKey(connectionId) || wsSubscriberGrouped.Group[connectionId] != connectionId) // Añade el par clave-valor al diccionario wsSubscriberGrouped.Group[connectionId] = groupedField.Group ?? string.Empty; } else { AddSubscriberGrouped( // ReSharper disable once AsyncVoidLambda new WsSubscriberGrouped(patientId, connectionId, timeZoneId, groupedField, lastObsInGroup, async delegate (object? sender, string _) { try { if (sender is WsSubscriberGrouped ws) await AddEmptyObs(ws); } catch (Exception ex) { logger.LogError(ex, "Ocurrió un error al procesar el evento del suscriptor."); throw; } })); } } /// /// Updates the last grouped observation in the subscriber group identified by the specified hash code. /// Performs a thread-safe lookup; if no matching subscriber group is found, the operation is silently skipped. /// /// The hash code used to identify the target subscriber group whose last grouped observation should be updated. /// The new to set as the last grouped observation in the matched group. public void UpdateLastGroupedObsInGroup(string wsgHashCode, GroupedObservation newGroupedObservation) { lock (_subscriberGrouped) { _subscriberGrouped.FirstOrDefault(c => c.HashCode == wsgHashCode)?.UpdateLastGo(newGroupedObservation); } } /// /// Creates the next empty grouped observation for the supplied WebSocket subscriber group and distributes it asynchronously to every subscriber in the group, applying the subscriber-specific group value resolved from the group mapping. /// /// The grouped WebSocket subscriber context whose next empty observation is generated and whose subscribers will receive the notification. private async Task AddEmptyObs(WsSubscriberGrouped ws) { var gobs = await groupedObservationService.CreateNextEmptyObs(ws); UpdateLastGroupedObsInGroup(ws.HashCode, gobs); lock (_subscriberGrouped) { foreach (var sub in ws.WsSubscriber) { gobs.Group = ws.Group.GetValue(sub)!; logger.LogInformation("Sending empty obs on gruped for: {Sub}, with group: {Group}", sub, gobs.Group); _ = clientMessageService.Value.SendAsync(sub, OperationType.GroupedObservation, gobs); } } } }