DD.WellWorkover.Cloud/AsbCloudInfrastructure/Services/LimitingParameterBackgroundService.cs

using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using System;
using System.Data.Common;
using System.Data;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;
using AsbCloudInfrastructure.Background;
using Microsoft.Extensions.DependencyInjection;

namespace AsbCloudInfrastructure.Services
{

    internal static class LimitingParameterCalcWorkFactory
    {
        private const string workId = "Limiting parameter calc";
        private static readonly TimeSpan workPeriod = TimeSpan.FromMinutes(30);

        public static WorkPeriodic MakeWork()
        {
            var workPeriodic = new WorkPeriodic(workId, WorkAction, workPeriod)
            {
                Timeout = TimeSpan.FromMinutes(30)
            };
            return workPeriodic;
        }

        // TODO: Разделить этот акшн на более мелкие части И использовать telemetryServiceData<..> вместо прямого обращения к БД.
        private static async Task WorkAction(string _, IServiceProvider serviceProvider, CancellationToken token)
        {
            using var db = serviceProvider.GetRequiredService<IAsbCloudDbContext>();
            var lastDetectedDates = await db.LimitingParameter
                .GroupBy(o => o.IdTelemetry)
                .Select(g => new
                {
                    IdTelemetry = g.Key,
                    LastDate = g.Max(o => o.DateEnd)
                })
                .ToListAsync(token);

            var telemetryIds = await db.Telemetries
                .Where(t => t.Info != null && t.TimeZone != null)
                .Select(t => t.Id)
                .ToListAsync(token);

            var telemetryLastDetectedDates = telemetryIds
                .GroupJoin(lastDetectedDates,
                    t => t,
                    o => o.IdTelemetry,
                    (outer, inner) => new
                    {
                        IdTelemetry = outer,
                        inner.SingleOrDefault()?.LastDate,
                    });

            foreach (var item in telemetryLastDetectedDates)
            {
                var newLimitingParameters = await GetLimitingParameterAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, db, token);
                if (newLimitingParameters?.Any() == true)
                {
                    db.LimitingParameter.AddRange(newLimitingParameters);
                    await db.SaveChangesAsync(token);
                }
            }
        }

        private static async Task<IEnumerable<LimitingParameter>> GetLimitingParameterAsync(int idTelemetry, DateTimeOffset begin, IAsbCloudDbContext db, CancellationToken token)
        {
            var query = 
                $"select " +
                    $"limiting_parameters.date, limiting_parameters.id_feed_regulator, limiting_parameters.well_depth " +
                $"from ( " +
                    $"select " +
                    $"date, id_feed_regulator, well_depth, " +
                    $"lag(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lag, " +
                    $"lead(id_feed_regulator, 1) over (order by date) as id_feed_regulator_lead " +
                        $"from t_telemetry_data_saub " +
                        $"where id_feed_regulator is not null " + 
                            $"and id_telemetry = {idTelemetry}" +
                            $"and date >= '{begin:u}'" +
                        $"order by date) as limiting_parameters " +
                $"where id_feed_regulator_lag is null " +
                        $"or (id_feed_regulator != id_feed_regulator_lag and id_feed_regulator_lead != id_feed_regulator_lag) " +
                $"order by date;";

            var limitingParameters = new List<LimitingParameter>(32);
            using (var result = await ExecuteReaderAsync(db, query, token))
            {
                LimitingParameter? limitingLast = null;
                while (result.Read())
                {
                    var date = result.GetFieldValue<DateTimeOffset>(0);
                    var idLimiting = result.GetFieldValue<short>(1);
                    var wellDepth = result.GetFieldValue<float>(2);

                    if (limitingLast is null)
                    {
                        limitingLast = new LimitingParameter
                        {
                            DateStart = date,
                            DepthStart = wellDepth,
                            IdFeedRegulator = idLimiting
                        };
                    }

                    if (limitingLast.IdFeedRegulator != idLimiting && limitingLast.DepthStart < wellDepth)
                    {
                        limitingParameters.Add(new LimitingParameter { 
                            IdTelemetry = idTelemetry,
                            IdFeedRegulator = limitingLast.IdFeedRegulator,
                            DateStart = limitingLast.DateStart,
                            DateEnd = date,
                            DepthStart = limitingLast.DepthStart,
                            DepthEnd = wellDepth
                        });   

                        limitingLast = new LimitingParameter
                        {
                            DateStart = date,
                            DepthStart = wellDepth,
                            IdFeedRegulator = idLimiting
                        };
                    }
                }
            }

            return limitingParameters;
        }

        private static async Task<DbDataReader> ExecuteReaderAsync(IAsbCloudDbContext db, string query, CancellationToken token)
        {
            var connection = db.Database.GetDbConnection();
            if (
                connection?.State is null ||
                connection.State == ConnectionState.Broken ||
                connection.State == ConnectionState.Closed)
            {
                await db.Database.OpenConnectionAsync(token);
                connection = db.Database.GetDbConnection();
            }
            using var command = connection.CreateCommand();
            command.CommandText = query;

            var result = await command.ExecuteReaderAsync(token);
            return result;
        }
    }

}