DD.WellWorkover.Cloud/AsbCloudInfrastructure/Services/SAUB/TelemetryDataCache.cs

using AsbCloudDb.Model;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System;
using System.Linq;
using Microsoft.EntityFrameworkCore;
using Mapster;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using AsbCloudInfrastructure.Background;
using System.Threading;
using AsbCloudApp.Data;
using AsbCloudApp.Requests;
using AsbCloudApp.Repositories;

namespace AsbCloudInfrastructure.Services.SAUB
{
    public class TelemetryDataCache<TDto> : ITelemetryDataCache<TDto> where TDto : AsbCloudApp.Data.ITelemetryData
    {
        class TelemetryDataCacheItem
        {
            public TDto FirstByDate { get; init; } = default!;
            public CyclicArray<TDto> LastData { get; init; } = null!;
            public double TimezoneHours { get; init; } = 5;
        }
        
        private const int activeWellCapacity = 12 * 60 * 60;
        private const int doneWellCapacity = 65 * 60;

        // key == idTelemetry
        private readonly ConcurrentDictionary<int, TelemetryDataCacheItem> caches;
        private bool isLoading = false;

        private TelemetryDataCache()
        {
            caches = new();
        }

        private static TelemetryDataCache<TDto>? instance;

        public static TelemetryDataCache<TDto> GetInstance<TEntity>(IServiceProvider provider)
            where TEntity : class, AsbCloudDb.Model.ITelemetryData
        {
            if (instance is null)
            {
                instance = new TelemetryDataCache<TDto>();
                var worker = provider.GetRequiredService<BackgroundWorker>();
                var workId = $"Telemetry cache loading from DB {typeof(TEntity).Name}";
                var work = Work.CreateByDelegate(workId, async (workId, provider, onProgress, token) =>
                {
                    var db = provider.GetRequiredService<IAsbCloudDbContext>();
                    await instance.InitializeCacheFromDBAsync<TEntity>(db, onProgress, token);
                });
                work.Timeout = TimeSpan.FromMinutes(15);
                worker.Enqueue(work);
            }
            return instance;
        }

        /// <summary>
        /// Добавить новые элементы в кеш
        /// </summary>
        /// <param name="idTelemetry"></param>
        /// <param name="range"></param>
        public void AddRange(int idTelemetry, IEnumerable<TDto> range)
        {
            if (!range.Any())
                return;

            range = range.OrderBy(x => x.DateTime);

            foreach (var item in range)
                item.IdTelemetry = idTelemetry;

            TelemetryDataCacheItem cacheItem;
            if (isLoading)
            {
                if (caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? localCacheItem))
                    cacheItem = localCacheItem;
                else
                    return;
            }
            else
            {
                cacheItem = caches.GetOrAdd(idTelemetry, _ => new TelemetryDataCacheItem() 
                { 
                    FirstByDate = range.ElementAt(0),
                    LastData = new CyclicArray<TDto>(activeWellCapacity) 
                });
            }
            
            cacheItem.LastData.AddRange(range);
        }

        /// <summary>
        /// Получить данные из кеша. <br/>
        /// Если dateBegin меньше минимального элемента в кеше, то вернется null.
        /// Даже если intervalSec частично перекрыт данными из кеша.
        /// </summary>
        /// <param name="idTelemetry"></param>
        /// <param name="dateBegin"></param>
        /// <param name="intervalSec"></param>
        /// <param name="approxPointsCount">кол-во элементов до которых эти данные прореживаются</param>
        /// <returns></returns>
        public IEnumerable<TDto>? GetOrDefault(int idTelemetry, DateTime dateBegin, double intervalSec = 600d, int approxPointsCount = 1024)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return null;

            var cacheLastData = cacheItem.LastData;

            if (!cacheLastData.Any() || cacheLastData[0].DateTime > dateBegin)
                return null;

            var dateEnd = dateBegin.AddSeconds(intervalSec);
            var items = cacheLastData
                .Where(i => i.DateTime >= dateBegin && i.DateTime <= dateEnd);

            var ratio = items.Count() / approxPointsCount;
            if (ratio > 1)
                items = items
                    .Where((_, index) => index % ratio == 0);

            return items;
        }

        public IEnumerable<TelemetryDataStatDto> GetStat()
        {
            var result = caches.Select(cacheItem => new TelemetryDataStatDto
            {
                IdTelemetry = cacheItem.Key,
                DateFirst = cacheItem.Value.FirstByDate.DateTime,
                DateLast = cacheItem.Value.LastData[^1].DateTime,
                TimezoneOffsetHours = cacheItem.Value.TimezoneHours,
            });
            return result;
        }

        public virtual TDto? GetLastOrDefault(int idTelemetry)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return default;

            return cacheItem.LastData.LastOrDefault();
        }

        public DatesRangeDto? GetOrDefaultDataDateRange(int idTelemetry)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return null;
            
            if (!cacheItem.LastData.Any())
                return null;

            var from = cacheItem.FirstByDate.DateTime;
            var to = cacheItem.LastData[^1].DateTime;

            return new DatesRangeDto 
            { 
                From = new DateTimeOffset(from, TimeSpan.FromHours(cacheItem.TimezoneHours)),
                To = new DateTimeOffset(to, TimeSpan.FromHours(cacheItem.TimezoneHours)) 
            };
        }

        public DatesRangeDto? GetOrDefaultCachedDateRange(int idTelemetry)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return null;

            if (cacheItem.LastData.Count < 2)
                return null;

            var to = cacheItem.LastData[^1].DateTime;
            var from = cacheItem.LastData[0].DateTime;

            return new DatesRangeDto { From = from, To = to };
        }

        public (TDto First, TDto Last)? GetOrDefaultFirstLast(int idTelemetry)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return null;

            if (!cacheItem.LastData.Any())
                return null;

            var last = cacheItem.LastData[^1];
            var first = cacheItem.FirstByDate;
            return (first, last);
        }

        private async Task InitializeCacheFromDBAsync<TEntity>(IAsbCloudDbContext db, Action<string, double?> onProgress, CancellationToken token)
            where TEntity : class, AsbCloudDb.Model.ITelemetryData
        {
            var defaultTimeout = db.Database.GetCommandTimeout();
            db.Database.SetCommandTimeout(TimeSpan.FromMinutes(5));

            if (isLoading)
                throw new Exception("Multiple cache loading detected.");

            try
            {                
                isLoading = true;
            
                Well[] wells = await db.Set<Well>()
                    .Include(well => well.Telemetry)
                    .Include(well => well.Cluster)
                    .Where(well => well.IdTelemetry != null)
                    .ToArrayAsync(token);

                var count = wells.Length;
                var i = 0d;
                foreach (Well well in wells)
                {
                    var capacity = well.IdState == 1
                        ? activeWellCapacity
                        : doneWellCapacity;

                    var idTelemetry = well.IdTelemetry!.Value;
                    var hoursOffset = well.Timezone.Hours;

                    onProgress($"Loading for well: {well.Cluster?.Caption}/{well.Caption} (capacity:{capacity}) idTelemetry:{idTelemetry}", i++ / count);
                    var cacheItem = await GetOrDefaultCacheDataFromDbAsync<TEntity>(db, idTelemetry, capacity, hoursOffset, token);
                    if (cacheItem is not null)
                        caches.TryAdd(idTelemetry, cacheItem);
                }
            }
            finally
            {
                isLoading = false;
                db.Database.SetCommandTimeout(defaultTimeout);
            }
        }

        private static async Task<TelemetryDataCacheItem?> GetOrDefaultCacheDataFromDbAsync<TEntity>(IAsbCloudDbContext db, int idTelemetry, int capacity, double hoursOffset, CancellationToken token)
            where TEntity : class, AsbCloudDb.Model.ITelemetryData
        {
            var query = db.Set<TEntity>()
                .Where(i => i.IdTelemetry == idTelemetry);

            var firstDbEntity = await query
                .OrderBy(i => i.DateTime)
                .FirstOrDefaultAsync(token);

            if (firstDbEntity is null)
                return default;

            var first = firstDbEntity.Adapt<TDto>();
            first.DateTime = firstDbEntity.DateTime.ToRemoteDateTime(hoursOffset);

            var entities = await query
                .OrderByDescending(i => i.DateTime)
                .Take(capacity)
                .ToArrayAsync(token);

            var dtos = entities
                .AsEnumerable()
                .Reverse()
                .Select(entity =>
                {
                    var dto = entity.Adapt<TDto>();
                    dto.DateTime = entity.DateTime.ToRemoteDateTime(hoursOffset);
                    return dto;
                });

            var cacheItem = new CyclicArray<TDto>(capacity);
            cacheItem.AddRange(dtos);

            var item = new TelemetryDataCacheItem
            {
                FirstByDate = first,
                LastData = cacheItem,
                TimezoneHours = hoursOffset,
            };
            return item;
        }

        public IEnumerable<TDto>? GetOrDefault(int idTelemetry, TelemetryDataRequest request)
        {
            if (!caches.TryGetValue(idTelemetry, out TelemetryDataCacheItem? cacheItem))
                return null;
            
            IEnumerable<TDto> data = cacheItem.LastData;

            if (!data.Any())
                return null;
            
            if (request.GeDate.HasValue)
            {
                var geDate = request.GeDate.Value.ToRemoteDateTime(cacheItem.TimezoneHours);
                if (data.First().DateTime > geDate)
                    return null;

                data = data.Where(d => d.DateTime >= geDate);
            }
            else
            {
                if (request.Order == 0)
                    return null;
            }

            if (request.LeDate.HasValue)
            {
                var leDate = request.LeDate.Value.ToRemoteDateTime(cacheItem.TimezoneHours);
                data = data.Where(d => d.DateTime <= request.LeDate);
            }
            
            if (request.Divider > 1)
                data = data.Where((d) => (((d.DateTime.DayOfYear * 24 + d.DateTime.Hour) * 60 + d.DateTime.Minute) * 60 + d.DateTime.Second) % request.Divider == 0);

            switch (request.Order)
            {
                case 1: // Поздние вперед
                    data = data
                        .OrderByDescending(d => d.DateTime)
                        .Skip(request.Skip)
                        .Take(request.Take)
                        .OrderBy(d => d.DateTime);
                    break;
                default: // Ранние вперед
                    data = data
                        .OrderBy(d => d.DateTime)
                        .Skip(request.Skip)
                        .Take(request.Take);
                    break;
            }

            return data;
        }

        public IEnumerable<int> GetIds(TelemetryDataRequest request)
        {
            var data = caches.Where(i => i.Value.LastData.Count > 0);

            if (request.GeDate.HasValue)
            {
                data = data
                    .Where(item => {
                        var lastItem = item.Value.LastData.Last();
                        var geDate = request.GeDate.Value.ToOffset(TimeSpan.FromHours(item.Value.TimezoneHours));
                        return lastItem.DateTime >= geDate;
                    });
            }

            if (request.LeDate.HasValue)
            {
                data = data
                    .Where(item => {
                        var firstItem = item.Value.LastData.First();
                        var leDate = request.LeDate.Value.ToOffset(TimeSpan.FromHours(item.Value.TimezoneHours));
                        return firstItem.DateTime <= leDate;
                    });
            }

            var telemetryIds = data.Select(item => item.Key);
            return telemetryIds;
        }
    }
}