persistence/DD.Persistence/Services/TimestampedValuesService.cs

using DD.Persistence.Extensions;
using DD.Persistence.Models;
using DD.Persistence.Repositories;
using DD.Persistence.Services.Interfaces;
using System.Text.Json;

namespace DD.Persistence.Services;

/// <inheritdoc/>
public class TimestampedValuesService : ITimestampedValuesService
{
    private readonly ITimestampedValuesRepository timestampedValuesRepository;
    private readonly ISchemePropertyRepository dataSchemeRepository;

    /// <inheritdoc/>
    public TimestampedValuesService(ITimestampedValuesRepository timestampedValuesRepository, ISchemePropertyRepository relatedDataRepository)
    {
        this.timestampedValuesRepository = timestampedValuesRepository;
        this.dataSchemeRepository = relatedDataRepository;
    }

    /// <inheritdoc/>
    public async Task<int> AddRange(Guid discriminatorId, IEnumerable<TimestampedValuesDto> dtos, CancellationToken token)
    {
        // ToDo: реализовать без foreach
        foreach (var dto in dtos)
        {
            await CreateDataSchemeIfNotExist(discriminatorId, dto, token);
        }

        var result = await timestampedValuesRepository.AddRange(discriminatorId, dtos, token);

        return result;
    }

    /// <inheritdoc/>
    public async Task<IEnumerable<TimestampedValuesDto>> Get(IEnumerable<Guid> discriminatorIds, DateTimeOffset? geTimestamp, IEnumerable<string>? columnNames, int skip, int take, CancellationToken token)
    {
        var result = await timestampedValuesRepository.Get(discriminatorIds, geTimestamp, columnNames, skip, take, token);

        var dtos = await BindingToDataScheme(result, token);

        if (!columnNames.IsNullOrEmpty())
        {
            dtos = ReduceSetColumnsByNames(dtos, columnNames!).ToList();
        }

        return dtos;
    }

    /// <inheritdoc/>
    public async Task<IEnumerable<TimestampedValuesDto>> GetFirst(Guid discriminatorId, int takeCount, CancellationToken token)
    {
        var result = await timestampedValuesRepository.GetFirst(discriminatorId, takeCount, token);

        var resultBeforeBinding = new[] { KeyValuePair.Create(discriminatorId, result) }
            .ToDictionary();
        var dtos = await BindingToDataScheme(resultBeforeBinding, token);

        return dtos;
    }

    /// <inheritdoc/>
    public async Task<IEnumerable<TimestampedValuesDto>> GetLast(Guid discriminatorId, int takeCount, CancellationToken token)
    {
        var result = await timestampedValuesRepository.GetLast(discriminatorId, takeCount, token);

        var resultBeforeBinding = new[] { KeyValuePair.Create(discriminatorId, result) }
            .ToDictionary();
        var dtos = await BindingToDataScheme(resultBeforeBinding, token);

        return dtos;
    }

    /// <inheritdoc/>
    public async Task<IEnumerable<TimestampedValuesDto>> GetResampledData(
        Guid discriminatorId,
        DateTimeOffset beginTimestamp,
        double intervalSec = 600d,
        int approxPointsCount = 1024,
        CancellationToken token = default)
    {
        var result = await timestampedValuesRepository.GetResampledData(discriminatorId, beginTimestamp, intervalSec, approxPointsCount, token);

        var resultBeforeBinding = new[] { KeyValuePair.Create(discriminatorId, result) }
            .ToDictionary();
        var dtos = await BindingToDataScheme(resultBeforeBinding, token);

        return dtos;
    }

    /// <inheritdoc/>
    public async Task<IEnumerable<TimestampedValuesDto>> GetGtDate(Guid discriminatorId, DateTimeOffset beginTimestamp, CancellationToken token)
    {
        var result = await timestampedValuesRepository.GetGtDate(discriminatorId, beginTimestamp, token);

        var resultBeforeBinding = new[] { KeyValuePair.Create(discriminatorId, result) }
            .ToDictionary();
        var dtos = await BindingToDataScheme(resultBeforeBinding, token);

        return dtos;
    }

    // ToDo: рефакторинг, переименовать (текущее название не отражает суть)
    /// <summary>
    /// Преобразовать результат запроса в набор dto
    /// </summary>
    /// <param name="queryResult"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    private async Task<IEnumerable<TimestampedValuesDto>> BindingToDataScheme(IDictionary<Guid, IEnumerable<(DateTimeOffset Timestamp, object[] Values)>> queryResult, CancellationToken token)
    {
        IEnumerable<TimestampedValuesDto> result = [];
        foreach (var keyValuePair in queryResult)
        {
            var dataScheme = await dataSchemeRepository.Get(keyValuePair.Key, token);
            if (dataScheme is null)
                continue;

            foreach (var (Timestamp, Values) in keyValuePair.Value)
            {
                var dto = new TimestampedValuesDto()
                {
                    Timestamp = Timestamp.ToUniversalTime(),
                    Values = dataScheme
                        .ToDictionary(k => k.PropertyName, v => Values[v.Index])
                };

                result = result.Append(dto);
            }
        }

        return result;
    }

    /// <summary>
    /// Создать схему данных, при отсутствии таковой
    /// </summary>
    /// <param name="discriminatorId">Дискриминатор схемы</param>
    /// <param name="dto">Набор данных, по образу которого будет создана соответствующая схема</param>
    /// <param name="token"></param>
    /// <returns></returns>
    /// <exception cref="InvalidOperationException">Некорректный набор наименований полей</exception>
    private async Task CreateDataSchemeIfNotExist(Guid discriminatorId, TimestampedValuesDto dto, CancellationToken token)
    {
        var valuesList = dto.Values.ToList();
        var properties = valuesList.Select((e, index) => new SchemePropertyDto()
        {
            Index = index,
            PropertyName = e.Key,
            PropertyKind = ((JsonElement)e.Value).ValueKind
        });

        var dataScheme = await dataSchemeRepository.Get(discriminatorId, token);
        if (dataScheme is null)
        {
            dataScheme = new DataSchemeDto(discriminatorId, properties);
            await dataSchemeRepository.AddRange(dataScheme, token);

            return;
        }

        if (!dataScheme.Equals(properties))
        {
            throw new InvalidOperationException($"Для системы {discriminatorId.ToString()} " +
                $"был передан нехарактерный набор данных");
        }
    }

    /// <summary>
    /// Отсеить лишние поля в соответствии с заданным фильтром
    /// </summary>
    /// <param name="dtos"></param>
    /// <param name="fieldNames">Поля, которые необходимо оставить</param>
    /// <returns></returns>
    private static IEnumerable<TimestampedValuesDto> ReduceSetColumnsByNames(IEnumerable<TimestampedValuesDto> dtos, IEnumerable<string> fieldNames)
    {
        var result = dtos.Select(dto =>
        {
            var reducedValues = dto.Values
                .Where(v => fieldNames.Contains(v.Key))
                .ToDictionary();
            dto.Values = reducedValues;

            return dto;
        });

        return result;
    }
}