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);
}
}
}
}