adas-core/adas-core.Application/Subscriptions/SubscriberGroupedService.cs

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;

/// <summary>
/// Provides an implementation of <see cref="ISubscriberGroupedService"/> that coordinates subscription handling for grouped observations, integrating caching, client messaging, and logging capabilities.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
public class SubscriberGroupedService(
    ICacheService? cacheService,
    IGroupedObservationService groupedObservationService,
    ILogger<SubscriberGroupedService> logger,
    Lazy<IClientMessageService> clientMessageService)
    : ISubscriberGroupedService
{
    private List<WsSubscriberGrouped> _subscriberGrouped = [];


    //private static readonly ILogger _logger = Log.ForContext<SubscriberGrouped>();

    public List<WsSubscriberGrouped> SubscriberGroupedList
    {
        get
        {
            lock (_subscriberGrouped)
            {
                return _subscriberGrouped;
            }
        }
        set
        {
            lock (_subscriberGrouped)
            {
                _subscriberGrouped = value;
            }
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <returns>A new <see cref="List{WsSubscriberGrouped}"/> containing a copy of the grouped subscribers.</returns>
    public List<WsSubscriberGrouped> GetGrouped()
    {
        lock (_subscriberGrouped)
        {
            return _subscriberGrouped.ToList();
        }
    }

    /// <summary>
    /// Removes all grouped observations associated with the specified patient identifier from the subscriber collection in a thread-safe manner.
    /// </summary>
    /// <param name="patientId">The string representation of the patient identifier used to locate and remove the matching grouped observations.</param>
    public void RemoveGroupedObsByPatientId(string patientId)
    {
        lock (_subscriberGrouped)
        {
            var itemToRemove = _subscriberGrouped.Where(c => c.PatientId.ToString() == patientId).ToList();
            itemToRemove.ForEach(item => _subscriberGrouped.Remove(item));
        }
    }

    /// <summary>
    ///     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
    /// </summary>
    /// <param name="patientId">objectId on db for Patient</param>
    /// <param name="newLocation"></param>
    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<string>();
                foreach (var wsClient in wsSubscriberGrouped.WsSubscriber) wsClientsToRemove.Add(wsClient);

                // Remove the wsClients
                foreach (var wsClient in wsClientsToRemove) wsSubscriberGrouped.WsSubscriber.Remove(wsClient);

                CheckEmptySubscriberGroup(wsSubscriberGrouped);
            }
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="patientId">The patient identifier used to locate the subscriber groups to update.</param>
    /// <param name="wsIdToRemove">The WebSocket subscriber ID to remove from the matching subscriber and group collections.</param>
    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<string>();
                // 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);
            }
        }
    }

    /// <summary>
    ///     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
    /// </summary>
    /// <param name="wsSubscriberGrouped">
    ///     Object with properties to generate new grouped observations, clients using those
    ///     grouped obsercations and the generated list of grouped observations
    /// </param>
    public List<string> CheckEmptySubscriberGroup(WsSubscriberGrouped? wsSubscriberGrouped)
    {
        lock (_subscriberGrouped)
        {
            var removed = new List<string>();
            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;
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="wsSubscriberGrouped">The grouped WebSocket subscriber instance to be added to the collection.</param>
    public void AddSubscriberGrouped(WsSubscriberGrouped wsSubscriberGrouped)
    {
        lock (_subscriberGrouped)
        {
            _subscriberGrouped.Add(wsSubscriberGrouped);
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="groupedField">The grouped field criteria used to locate or create the matching subscription.</param>
    /// <param name="patientId">The identifier of the patient whose grouped observation is being subscribed to.</param>
    /// <param name="timeZoneId">The optional time zone identifier applied to the new subscription when one is created.</param>
    /// <param name="connectionId">The unique identifier of the client connection being registered.</param>
    /// <param name="lastObsInGroup">The most recent observation in the group, used to initialize a new subscription.</param>
    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;
                        }
                    }));
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="wsgHashCode">The hash code used to identify the target subscriber group whose last grouped observation should be updated.</param>
    /// <param name="newGroupedObservation">The new <see cref="GroupedObservation"/> to set as the last grouped observation in the matched group.</param>
    public void UpdateLastGroupedObsInGroup(string wsgHashCode, GroupedObservation newGroupedObservation)
    {
        lock (_subscriberGrouped)
        {
            _subscriberGrouped.FirstOrDefault(c => c.HashCode == wsgHashCode)?.UpdateLastGo(newGroupedObservation);
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <param name="ws">The grouped WebSocket subscriber context whose next empty observation is generated and whose subscribers will receive the notification.</param>
    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);
            }
        }
    }
}