DD.WellWorkover.Cloud/AsbCloudInfrastructure/Background/PeriodicWorks/WorkLimitingParameterCalc.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 Microsoft.Extensions.DependencyInjection;

namespace AsbCloudInfrastructure.Background.PeriodicWorks;

public class WorkLimitingParameterCalc : Work
{
    public WorkLimitingParameterCalc()
        : base("Limiting parameter calc")
    {
        Timeout = TimeSpan.FromMinutes(30);
    }

    protected override async Task Action(string id, IServiceProvider services, Action<string, double?> onProgressCallback, CancellationToken token)
    {
        using var db = services.GetRequiredService<IAsbCloudDbContext>();
        db.Database.SetCommandTimeout(TimeSpan.FromMinutes(5));
        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,
                });

        var count = telemetryLastDetectedDates.Count();
        var i = 0d;
        foreach (var item in telemetryLastDetectedDates)
        {
            onProgressCallback($"Start hanling telemetry: {item.IdTelemetry} from {item.LastDate}", i++ / count);
            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;
    }
}