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

using AsbCloudDb.Model;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using System;
using System.Data.Common;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;

namespace AsbCloudInfrastructure.Services
{
#nullable enable
    internal class LimitingParameterBackgroundService : BackgroundService
    {
        private readonly string connectionString;
        private readonly TimeSpan period = TimeSpan.FromHours(1);

        public LimitingParameterBackgroundService(IConfiguration configuration)
        {
            connectionString = configuration.GetConnectionString("DefaultConnection");
        }

        protected override async Task ExecuteAsync(CancellationToken token)
        {
            var timeToStart = DateTime.Now;
            var options = new DbContextOptionsBuilder<AsbCloudDbContext>()
                .UseNpgsql(connectionString)
                .Options;
            while (!token.IsCancellationRequested)
            {
                if (DateTime.Now > timeToStart)
                {
                    timeToStart = DateTime.Now + period;
                    try
                    {
                        using var context = new AsbCloudDbContext(options);
                        var added = await LimitingParameterAsync(context, token);
                        Trace.TraceInformation($"Total limiting parameter complete. Added {added} limiting parameters.");
                    }
                    catch (Exception ex)
                    {
                        Trace.TraceError(ex.Message);
                    }
                    GC.Collect();
                }
                var ms = (int)(timeToStart - DateTime.Now).TotalMilliseconds;
                ms = ms > 100 ? ms : 100;
                await Task.Delay(ms, token).ConfigureAwait(false);
            }
        }

        private static async Task<int> LimitingParameterAsync(IAsbCloudDbContext context, CancellationToken token)
        {
            var lastDetectedDates = await context.LimitingParameter
                .GroupBy(o => o.IdTelemetry)
                .Select(g => new
                {
                    IdTelemetry = g.Key,
                    LastDate = g.Max(o => o.DateEnd)
                })
                .ToListAsync(token);

            var telemetryIds = await context.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 affected = 0;
            foreach (var item in telemetryLastDetectedDates)
            {
                var newLimitingParameters = await GetLimitingParameterAsync(item.IdTelemetry, item.LastDate ?? DateTimeOffset.MinValue, context, token);
                if (newLimitingParameters?.Any() == true)
                {
                    context.LimitingParameter.AddRange(newLimitingParameters);
                    affected += await context.SaveChangesAsync(token);
                }
            }
            return affected;
        }

        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)
                    {
                        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;
        }
    }
#nullable disable
}