persistence/DD.Persistence/Services/TimestampedValuesService.cs

using DD.Persistence.Extensions;
using DD.Persistence.Models;
using DD.Persistence.Models.Common;
using DD.Persistence.Repositories;
using DD.Persistence.Services.Interfaces;

namespace DD.Persistence.Services;

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

    /// <inheritdoc/>
    public TimestampedValuesService(ITimestampedValuesRepository timestampedValuesRepository, IDataSchemeRepository 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)
        {
            var keys = dto.Values.Keys.ToArray();
            await CreateSystemSpecificationIfNotExist(discriminatorId, keys, token);
        }

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

        return result;
    }

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

        var dtos = await Materialize(discriminatorId, result, token);

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

        return dtos;
    }

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

        List<TimestampedValuesDto> dtos = [];
        discriminatorIds.ForEach(async discriminatorId => {
            var materializeDtos = await Materialize(discriminatorId, result, token);
            dtos.AddRange(materializeDtos);
        });

        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 dtos = await Materialize(discriminatorId, result, 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 dtos = await Materialize(discriminatorId, result, 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 dtos = await Materialize(discriminatorId, result, 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 dtos = await Materialize(discriminatorId, result, token);

        return dtos;
    }

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

        return result;
    }

    /// <inheritdoc/>
    public async virtual Task<DatesRangeDto?> GetDatesRange(Guid discriminatorId, CancellationToken token)
    {
        var result = await timestampedValuesRepository.GetDatesRange(discriminatorId, token);

        return result;
    }

    /// <summary>
    /// Преобразовать результат запроса в набор dto
    /// </summary>
    /// <param name="dataSchemeId"></param>
    /// <param name="queryResult"></param>
    /// <param name="token"></param>
    /// <returns></returns>
    private async Task<IEnumerable<TimestampedValuesDto>> Materialize(Guid dataSchemeId, IEnumerable<Tuple<DateTimeOffset, object[]>> queryResult, CancellationToken token)
    {
        var dataScheme = await dataSchemeRepository.Get(dataSchemeId, token);
        if (dataScheme is null)
        {
            return [];
        }

        var dtos = queryResult.Select(entity =>
        {
            var dto = new TimestampedValuesDto()
            {
                Timestamp = entity.Item1.ToUniversalTime()
            };

            var identity = dataScheme!.PropNames;
            var indexedIdentity = identity
                .Select((value, index) => new { index, value });
            dto.Values = indexedIdentity
                .ToDictionary(x => x.value, x => entity.Item2[x.index]);

            return dto;
        });

        return dtos;
    }

    /// <summary>
    /// Создать спецификацию, при отсутствии таковой
    /// </summary>
    /// <param name="discriminatorId">Дискриминатор системы</param>
    /// <param name="fieldNames">Набор наименований полей</param>
    /// <param name="token"></param>
    /// <returns></returns>
    /// <exception cref="InvalidOperationException">Некорректный набор наименований полей</exception>
    private async Task CreateSystemSpecificationIfNotExist(Guid discriminatorId, string[] fieldNames, CancellationToken token)
    {
        var systemSpecification = await dataSchemeRepository.Get(discriminatorId, token);
        if (systemSpecification is null)
        {
            systemSpecification = new DataSchemeDto()
            {
                DiscriminatorId = discriminatorId,
                PropNames = fieldNames
            };
            await dataSchemeRepository.Add(systemSpecification, token);

            return;
        }

        if (!systemSpecification.PropNames.SequenceEqual(fieldNames))
        {
            var expectedFieldNames = string.Join(", ", systemSpecification.PropNames);
            var actualFieldNames = string.Join(", ", fieldNames);
            throw new InvalidOperationException($"Для системы {discriminatorId.ToString()} " +
                $"характерен набор данных: [{expectedFieldNames}], однако был передан набор: [{actualFieldNames}]");
        }
    }

    /// <summary>
    /// Отсеить лишние поля в соответствии с заданным фильтром
    /// </summary>
    /// <param name="dtos"></param>
    /// <param name="fieldNames">Поля, которые необходимо оставить</param>
    /// <returns></returns>
    private 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;
    }
}