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

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

namespace AsbCloudInfrastructure.Services;


class JobHandle
{
    public int Id => Job.Id;
    public JobDto Job { get; set; } = null!;
    public DateTime DateCreate { get; set; } = DateTime.Now;
    public OnJobProgressDelagate? OnProgress { get; set; }
}

public class ReduceSamplingService : IReduceSamplingService
{
    private const string jobResultKeyDeleted = "deleted";
    private const string jobResultKeyTotal = "total";
    private static ReduceSamplingService? instance;
    private readonly string connectionString;
    private const int ratio = 5;
    private readonly List<JobHandle> jobHandlers = new(5);
    private bool isHandling;
    private CancellationTokenSource? cancellationTokenSource;
    private Task? task;

    private ReduceSamplingService(IConfiguration configuration)
    {
        connectionString = configuration.GetConnectionString("DefaultConnection")
             ?? throw new ConfigurationErrorsException("DefaultConnection не определен");
    }

    ~ReduceSamplingService()
    {
        Stop();
    }

    /// <summary>
    /// Get singleton Instance 
    /// </summary>
    /// <param name="configuration"></param>
    /// <returns></returns>
    public static ReduceSamplingService GetInstance(IConfiguration configuration)
    {
        instance ??= new(configuration);
        return instance;
    }

    /// <inheritdoc/>
    public bool TryEnqueueRediceSamplingJob(int idTelemetry, OnJobProgressDelagate? onProgress, out JobDto jobDto)
    {
        lock (jobHandlers)
        {
            var oldJob = jobHandlers.FirstOrDefault(j => j.Id == idTelemetry);
            if (oldJob is not null)
            {
                jobDto = oldJob.Job;
                return false;
            }

            jobDto = new JobDto
            {
                Id = idTelemetry,
                State = JobState.Waiting,
                Results = new(),
            };
            var jobHandler = new JobHandle
            {
                Job = jobDto,
                OnProgress = onProgress,
            };

            jobHandlers.Add(jobHandler);                
        }
        EnsureHandleQueueStarted();
        return true;
    }

    /// <inheritdoc/>
    public JobDto? GetOrDefaultState(int idTelemetry)
    {
        JobHandle? jobHandler;
        lock (jobHandlers)
        {
            jobHandler = jobHandlers.FirstOrDefault(j => j.Id == idTelemetry);
        }
        return jobHandler?.Job;
    }

    /// <inheritdoc/>
    public IEnumerable<JobDto> GetJobs() => jobHandlers.Select(j=>j.Job);

    private bool TryTakeWaitingJob(out JobHandle? job)
    {
        lock (jobHandlers)
        {
            job = jobHandlers.FirstOrDefault(j => j.Job.State == JobState.Waiting);
        }
        return job is not null;
    }

    private void EnsureHandleQueueStarted()
    {
        if (isHandling)
            return;
        isHandling = true;
        cancellationTokenSource = new CancellationTokenSource();
        var token = cancellationTokenSource.Token;
        task = Task.Run(async () => await HandleJobs(token))
            .ContinueWith(_ => isHandling = false);
    }

    private async Task HandleJobs(CancellationToken token)
    {
        while (TryTakeWaitingJob(out JobHandle? jobHandler))
        {
            jobHandler!.Job.State = JobState.Working;
            try
            {
                await RediceSamplingSaubAsync(jobHandler, token);
                await RediceSamplingSpinAsync(jobHandler, token);
                jobHandler.Job.State = JobState.Done;
            }
            catch (Exception exception)
            {
                jobHandler.Job.State = JobState.Fail;
                jobHandler.Job.Results = null;
                jobHandler.Job.Error = exception.Message;
                jobHandler.OnProgress?.Invoke(jobHandler.Job);
            }

            if (!jobHandlers.Any(j => j.Job.State == JobState.Waiting))
            {
                var sw = Stopwatch.StartNew();
                await VacuumAsync(token);
                sw.Stop();
                if (sw.ElapsedMilliseconds < 10_000)
                {
                    var delayMs = 10_000 - (int)sw.ElapsedMilliseconds;
                    await Task.Delay(delayMs, token);
                }
                CleanJobs();
            }
        }
    }

    private void CleanJobs()
    {
        lock (jobHandlers)
        {
            jobHandlers.RemoveAll(j => j.Job.State == JobState.Done || j.Job.State == JobState.Fail);
        }
    }

    private async Task VacuumAsync(CancellationToken token)
    {
        using var db = MakeContext();
        var sqlVacuum = "vacuum (SKIP_LOCKED);";
        await db.Database.ExecuteSqlRawAsync(sqlVacuum, token);
    }

    private void Stop()
    {
        cancellationTokenSource?.Cancel();
        task?.Wait(1_000);
        task = null;
        cancellationTokenSource?.Dispose();
        cancellationTokenSource = null;
    }

    private Task RediceSamplingSaubAsync(JobHandle job, CancellationToken token)
    {
        const int ramLimit = 10 * 1024 * 1024;
        const int dataItemSize = 345; // by profiler
        const int chankSize = ramLimit / dataItemSize; // ~ 90_000
        const double maxWellDepthGap = 0.1;

        var maxDateGapSec = ratio;

        var sqlSelectTemplate =
            "select " +
            "   * " +
            "from " +
            "   (select " +
            "      *, " +
            "      rank() over win1 as row_num, " +
            "      lag(\"date\", 1) over win1 as lag_date, " +
            "      lag(\"mode\", 1) over win1 as lag_mode, " +
            "      lag(mse_state, 1) over win1 as lag_mse_state, " +
            "      lag(well_depth, 1) over win1 as lag_well_depth, " +
            "      lag(id_feed_regulator, 1) over win1 as lag_id_feed_regulator " +
            "   from t_telemetry_data_saub " +
            $"   where id_telemetry = {job.Id} and \"date\" > {{0}}" +
            "   window win1 as (order by \"date\") " +
            "   ) as t_1 " +
            "where " +
            $"   (row_num % {ratio}) = 0 " +
            "   or \"mode\" != lag_mode " +
            $"   or(\"date\" - lag_date) >= interval '{maxDateGapSec} second' " +
            $"   or well_depth - lag_well_depth > {maxWellDepthGap:#0,0#} " +
            "   or mse_state != lag_mse_state " +
            "   or id_feed_regulator != lag_id_feed_regulator " +
            "order by \"date\" ";

        var sqlDeleteTemplate = "delete " +
            "from t_telemetry_data_saub " +
            $"where id_telemetry = {job.Id} and \"date\" between {{0}} and {{1}};";

        return RediceSamplingAsync<TelemetryDataSaub>(
            job,
            chankSize,
            sqlSelectTemplate,
            sqlDeleteTemplate,
            token);
    }

    private Task RediceSamplingSpinAsync(JobHandle job, CancellationToken token)
    {
        const int ramLimit = 10 * 1024 * 1024;
        const int dataItemSize = 345; // by profiler
        const int chankSize = ramLimit / dataItemSize; // ~ 90_000
        var maxDateGapSec = ratio;

        var sqlSelectTemplate =
            "select " +
            "   * " +
            "from " +
            "   (select " +
            "      *, " +
            "      rank() over win1 as row_num, " +
            "      lag(\"date\", 1) over win1 as lag_date, " +
            "      lag(\"mode\", 1) over win1 as lag_mode, " +
            "      lag(state, 1) over win1 as lag_state " +
            "   from t_telemetry_data_spin " +
            $"   where id_telemetry = {job.Id} and \"date\" > {{0}}" +
            "   window win1 as (order by \"date\") " +
            "   ) as t_1 " +
            "where " +
            $"   (row_num % {ratio}) = 0 " +
            "   or \"mode\" != lag_mode " +
            $"   or(\"date\" - lag_date) >= interval '{maxDateGapSec} second' " +
            "   or state != lag_state " +
            "order by \"date\" ";

        var sqlDeleteTemplate = "delete " +
            "from t_telemetry_data_spin " +
            $"where id_telemetry = {job.Id} and \"date\" between {{0}} and {{1}};";

        return RediceSamplingAsync<TelemetryDataSpin>(
            job,
            chankSize,
            sqlSelectTemplate,
            sqlDeleteTemplate,
            token);
    }

    private async Task RediceSamplingAsync<TEntity>(
        JobHandle jobHandle,
        int chankSize,
        string sqlSelectTemplate,
        string sqlDeleteTemplate,
        CancellationToken token)
        where TEntity : class, AsbCloudDb.Model.ITelemetryData
    {
        using var db = MakeContext();
        var dbset = db.Set<TEntity>();

        var deleted = 0;
        var totalCount = await dbset.Where(t => t.IdTelemetry == jobHandle.Id).CountAsync(token);

        var result = jobHandle.Job.Results!;
        if (result[jobResultKeyDeleted] is int previousDeleted)
            deleted += previousDeleted;

        if (result[jobResultKeyTotal] is int previousCount)
            totalCount += previousCount;

        result[jobResultKeyDeleted] = deleted;
        result[jobResultKeyTotal] = totalCount;

        jobHandle.OnProgress?.Invoke(jobHandle.Job);
        var startDate = DateTimeOffset.MinValue;

        do
        {
            var query = dbset
                .FromSqlRaw(sqlSelectTemplate, startDate)
                .AsNoTracking();

            var data = await query
                .Take(chankSize)
                .ToArrayAsync(token);

            var currentDataCount = data.Length;
            if (currentDataCount == 0)
                break;

            var lastDate = data.Last().DateTime;

            var currentDeleted = await db.Database.ExecuteSqlRawAsync(sqlDeleteTemplate, new object[] { startDate, lastDate }.AsEnumerable(), token);
            if (currentDeleted == 0)
                break;

            await db.Database.ExecInsertAsync(dbset, data, token);

            startDate = lastDate;
            deleted += currentDeleted;
            result[jobResultKeyDeleted] = deleted;
            result[jobResultKeyTotal] = totalCount;
            jobHandle.OnProgress?.Invoke(jobHandle.Job);
        } while (true);

        return;
    }

    private AsbCloudDbContext MakeContext()
    {
        var options = new DbContextOptionsBuilder<AsbCloudDbContext>()
            .UseNpgsql(connectionString)
            .Options;

        return new AsbCloudDbContext(options);
    }
}