EVCB_OCPP.OCPPServer.16J/EVCB_OCPP.WSServer/Service/MeterValueDbService.cs

using Dapper;
using EVCB_OCPP.Domain;
using EVCB_OCPP.WSServer.Helper;
using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Metadata.Internal;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using System.Data;
using System.Data.Common;
using System.Diagnostics;

namespace EVCB_OCPP.WSServer.Service;

public class MeterValueDbService
{
    private readonly IDbContextFactory<MeterValueDBContext> meterValueDbContextFactory;
    private readonly SqlConnectionFactory<MeterValueDBContext> sqlConnectionFactory;
    private readonly ILoggerFactory loggerFactory;
    private readonly MeterValueGroupSingleHandler meterValueGroupSingleHandler;
    private readonly QueueSemaphore insertSemaphore;
    private readonly ILogger logger;
    private readonly Queue<string> _existTables = new();
    private GroupSingleHandler<InsertMeterValueParam> insertMeterValueHandler;

    public MeterValueDbService(
        IDbContextFactory<MeterValueDBContext> meterValueDbContextFactory,
        SqlConnectionFactory<MeterValueDBContext> sqlConnectionFactory,
        ILogger<MeterValueDbService> logger,
        ILoggerFactory loggerFactory,
        IConfiguration configuration
        //, MeterValueGroupSingleHandler meterValueGroupSingleHandler
        )
    {
        this.meterValueDbContextFactory = meterValueDbContextFactory;
        this.sqlConnectionFactory = sqlConnectionFactory;
        this.loggerFactory = loggerFactory;
        //this.meterValueGroupSingleHandler = meterValueGroupSingleHandler;
        //this.meterValueConnectionString = configuration.GetConnectionString("MeterValueDBContext");
        this.logger = logger;

        InitInsertMeterValueHandler();

        var insertLimit = GetInsertLimit(configuration);
        insertSemaphore = new QueueSemaphore(insertLimit);

    }

    public Task InsertAsync(string chargeBoxId, byte connectorId, decimal value, DateTime createdOn
            , int contextId, int formatId, int measurandId, int phaseId
            , int locationId, int unitId, int transactionId)
    {
        var param = new InsertMeterValueParam(chargeBoxId, connectorId, value, createdOn, contextId, formatId, measurandId, phaseId, locationId, unitId, transactionId);

        //return insertMeterValueHandler.HandleAsync(param);
        return InsertAsync(param);
        //return meterValueGroupSingleHandler.HandleAsync(param);
    }

    public Task InsertAsync(InsertMeterValueParam param)
    {
        //return InsertWithEF(param);
        return InsertWithDapper(param);
    }

    public Task InsertBundleAsync(IEnumerable<InsertMeterValueParam> param)
    {
        return BundleInsertWithDapper(param);
    }

    private void InitInsertMeterValueHandler()
    {
        if (insertMeterValueHandler is not null)
        {
            throw new Exception($"{nameof(InitInsertMeterValueHandler)} should only called once");
        }

        insertMeterValueHandler = new GroupSingleHandler<InsertMeterValueParam>(
            //BulkInsertWithBulkCopy,
            BundleInsertWithDapper,
            //loggerFactory.CreateLogger("InsertMeterValueHandler")
            logger,
            workerCnt:1
            );
    }

    private async Task BundleInsertWithStoredProcedure(IEnumerable<InsertMeterValueParam> parms)
    {
        foreach (var param in parms)
        {
            await InsertWithStoredProcedure(param);
        }
    }

    private async Task InsertWithEF(InsertMeterValueParam param)
    {
        using var token = await insertSemaphore.GetToken();
        using var db = await meterValueDbContextFactory.CreateDbContextAsync();

        string sp = "[dbo].[uspInsertMeterValueRecord] @ChargeBoxId, @ConnectorId,@Value,@CreatedOn,@ContextId,@FormatId,@MeasurandId,@PhaseId,@LocationId,@UnitId,@TransactionId";

        List<SqlParameter> parameter = new List<SqlParameter>();
        parameter.AddInsertMeterValueRecordSqlParameters(
            chargeBoxId: param.chargeBoxId
            , connectorId: (byte)param.connectorId
            , value: param.value
            , createdOn: param.createdOn
            , contextId: param.contextId
            , formatId: param.formatId
            , measurandId: param.measurandId
            , phaseId: param.phaseId
            , locationId: param.locationId
            , unitId: param.unitId
            , transactionId: param.transactionId);

        await db.Database.ExecuteSqlRawAsync(sp, parameter.ToArray());
    }

    private async Task InsertWithDapper(InsertMeterValueParam param)
    {
        var watch = Stopwatch.StartNew();
        long t0, t1;

        if (!await GetTableExist(param.createdOn))
        {
            t0 = watch.ElapsedMilliseconds;
            await InsertWithStoredProcedure(param);
            watch.Stop();
            t1 = watch.ElapsedMilliseconds;
            if (t1 > 500)
            {
                logger.LogWarning("MeterValue InsertWithStoredProcedure {0}/{1}", t0, t1);
            }
            return;
        }

        t0 = watch.ElapsedMilliseconds;
        var tableName = GetTableName(param.createdOn);

        await InsertWithNoCheckDapper(tableName, param);

        watch.Stop();
        t1 = watch.ElapsedMilliseconds;
        if(t1 > 700)
        {
            logger.LogWarning("MeterValue Dapper {0}/{1}", t0, t1);
        }
    }

    private async Task BundleInsertWithDapper(IEnumerable<InsertMeterValueParam> parms)
    {
        var watch = Stopwatch.StartNew();
        long t0, t1, t2, t3, t4;

        var parmsList = parms.ToList();
        foreach (var param in parms)
        {
            if (!await GetTableExist(param.createdOn))
            {
                await InsertWithStoredProcedure(param);
                parmsList.Remove(param);
            }
        }

        t0 = watch.ElapsedMilliseconds;
        //logger.LogInformation("MeterValue bundle insert cnt {0}", parmsList.Count);
        var gruopParams = parmsList.GroupBy(x => GetTableName(x.createdOn));

        t1 = watch.ElapsedMilliseconds;
        using SqlConnection sqlConnection = await sqlConnectionFactory.CreateAsync();
        using var trans = await sqlConnection.BeginTransactionAsync();

        t2 = watch.ElapsedMilliseconds;

        List<Task> ExecuteTasks = new List<Task>();
        foreach (var group in gruopParams)
        {
            var tableName = group.Key;
            foreach(var param in group)
            {
                await InsertWithNoCheckDapper(tableName, param, sqlConnection, trans);
            }
        }

        t3 = watch.ElapsedMilliseconds;
        await trans.CommitAsync();

        watch.Stop();
        t4 = watch.ElapsedMilliseconds;
        if (t4 > 500)
        {
            logger.LogWarning("MeterValue Dapper {0}/{1}/{2}/{3}/{4}/{5}", t0, t1, t2, t3, t4, parms.Count());
        }
    }

    private async Task InsertWithNoCheckDapper(string tableName, InsertMeterValueParam data, SqlConnection conn = null, DbTransaction trans = null)
    {
        string command = $"""
                INSERT INTO {tableName} (ConnectorId, Value, CreatedOn, ContextId, FormatId, MeasurandId, PhaseId, LocationId, UnitId, ChargeBoxId, TransactionId)
                VALUES (@ConnectorId, @Value, @CreatedOn, @ContextId, @FormatId, @MeasurandId, @PhaseId, @LocationId, @UnitId, @ChargeBoxId, @TransactionId);
                """;

        SqlConnection connection = conn ?? await sqlConnectionFactory.CreateAsync();

        var parameters = new DynamicParameters();
        parameters.Add("ConnectorId", data.connectorId, DbType.Int16);
        parameters.Add("Value", data.value, DbType.Decimal, precision: 18, scale: 8);
        parameters.Add("CreatedOn", data.createdOn, DbType.DateTime);
        parameters.Add("ContextId", data.contextId, DbType.Int32);
        parameters.Add("FormatId", data.formatId, DbType.Int32);
        parameters.Add("MeasurandId", data.measurandId, DbType.Int32);
        parameters.Add("PhaseId", data.phaseId, DbType.Int32);
        parameters.Add("LocationId", data.locationId, DbType.Int32);
        parameters.Add("UnitId", data.unitId, DbType.Int32);
        parameters.Add("ChargeBoxId", data.chargeBoxId, DbType.String, size: 50);
        parameters.Add("TransactionId", data.transactionId, DbType.Int32);

        await connection.ExecuteAsync(command, parameters, trans);
    }

    private async Task BulkInsertWithBulkCopy(IEnumerable<InsertMeterValueParam> parms)
    {
        var watcher = Stopwatch.StartNew();
        long t0 = 0, t1 = 0, t2 = 0, t3 = 0, t4 = 0;

        var parmsList = parms.ToList();
        foreach (var param in parms)
        {
            if (!await GetTableExist(param.createdOn))
            {
                await InsertWithStoredProcedure(param);
                parmsList.Remove(param);
            }
        }

        //logger.LogInformation("MeterValue bundle insert cnt {0}", parmsList.Count);
        var gruopParams = parmsList.GroupBy(x => GetTableName(x.createdOn));

        t0 = watcher.ElapsedMilliseconds;
        foreach (var group in gruopParams)
        {
            var table = new DataTable();
            table.Columns.Add("ChargeBoxId");
            table.Columns.Add("ConnectorId");
            table.Columns.Add("Value");
            table.Columns.Add("CreatedOn");
            table.Columns.Add("ContextId");
            table.Columns.Add("FormatId");
            table.Columns.Add("MeasurandId");
            table.Columns.Add("PhaseId");
            table.Columns.Add("LocationId");
            table.Columns.Add("UnitId");
            table.Columns.Add("TransactionId");

            foreach (var param in group)
            {
                var row = table.NewRow();
                row["ChargeBoxId"] = param.chargeBoxId;
                row["ConnectorId"] = param.connectorId;
                row["Value"] = param.value;
                row["CreatedOn"] = param.createdOn;
                row["ContextId"] = param.contextId;
                row["FormatId"] = param.formatId;
                row["MeasurandId"] = param.measurandId;
                row["PhaseId"] = param.phaseId;
                row["LocationId"] = param.locationId;
                row["UnitId"] = param.unitId;
                row["TransactionId"] = param.transactionId;

                table.Rows.Add(row);
            }
            t1 = watcher.ElapsedMilliseconds;
            using SqlConnection sqlConnection = await sqlConnectionFactory.CreateAsync();
            using SqlBulkCopy sqlBulkCopy = new SqlBulkCopy(sqlConnection);
            t2 = watcher.ElapsedMilliseconds;
            sqlBulkCopy.BatchSize = group.Count();
            sqlBulkCopy.DestinationTableName = group.Key;

            sqlBulkCopy.ColumnMappings.Add("ChargeBoxId", "ChargeBoxId");
            sqlBulkCopy.ColumnMappings.Add("ConnectorId", "ConnectorId");
            sqlBulkCopy.ColumnMappings.Add("Value", "Value");
            sqlBulkCopy.ColumnMappings.Add("CreatedOn", "CreatedOn");
            sqlBulkCopy.ColumnMappings.Add("ContextId", "ContextId");
            sqlBulkCopy.ColumnMappings.Add("FormatId", "FormatId");
            sqlBulkCopy.ColumnMappings.Add("MeasurandId", "MeasurandId");
            sqlBulkCopy.ColumnMappings.Add("PhaseId", "PhaseId");
            sqlBulkCopy.ColumnMappings.Add("LocationId", "LocationId");
            sqlBulkCopy.ColumnMappings.Add("UnitId", "UnitId");
            sqlBulkCopy.ColumnMappings.Add("TransactionId", "TransactionId");
            t3 = watcher.ElapsedMilliseconds;
            await sqlBulkCopy.WriteToServerAsync(table);
        }
        watcher.Stop();
        t4 = watcher.ElapsedMilliseconds;

        if (t4 > 500)
        {
            logger.LogWarning("MeterValue BulkInsertWithBulkCopy Slow {0}/{1}/{2}/{3}/{4}/{5}", t0,t1,t2,t3,t4, parms.Count());
        }
    }

    private async ValueTask<bool> GetTableExist(DateTime tableDateTime)
    {
        var tableName = GetTableName(tableDateTime);
        if (_existTables.Contains(tableName))
        {
            return true;
        }

        FormattableString checkTableSql = $"SELECT Count(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = {tableName}";

        using var db = await meterValueDbContextFactory.CreateDbContextAsync();
        var resultList = db.Database.SqlQuery<int>(checkTableSql)?.ToList();

        if (resultList is not null && resultList.Count > 0 && resultList[0] > 0)
        {
            _existTables.Enqueue(tableName);
            if (_existTables.Count > 30)
            {
                _existTables.TryDequeue(out _);
            }
            return true;
        }

        return false;
    }

    private async Task InsertWithStoredProcedure(InsertMeterValueParam param)
    {
        using var db = await meterValueDbContextFactory.CreateDbContextAsync();

        string sp = "[dbo].[uspInsertMeterValueRecord] @ChargeBoxId," +
"@ConnectorId,@Value,@CreatedOn,@ContextId,@FormatId,@MeasurandId,@PhaseId,@LocationId,@UnitId,@TransactionId";

        List<SqlParameter> parameter = new List<SqlParameter>();
        parameter.AddInsertMeterValueRecordSqlParameters(
            chargeBoxId: param.chargeBoxId
            , connectorId: (byte)param.connectorId
            , value: param.value
            , createdOn: param.createdOn
            , contextId: param.contextId
            , formatId: param.formatId
            , measurandId: param.measurandId
            , phaseId: param.phaseId
            , locationId: param.locationId
            , unitId: param.unitId
            , transactionId: param.transactionId);

        await db.Database.ExecuteSqlRawAsync(sp, parameter.ToArray());
    }

    private static string GetTableName(DateTime dateTime) 
        => $"ConnectorMeterValueRecord{dateTime:yyMMdd}";

    private int GetInsertLimit(IConfiguration configuration)
    {
        var limitConfig = configuration["MeterValueDbInsertLimit"];
        int limit = 10;
        if (limitConfig != default)
        {
            int.TryParse(limitConfig, out limit);
        }
        return limit;
    }
}

public record InsertMeterValueParam(string chargeBoxId, byte connectorId, decimal value, DateTime createdOn
            , int contextId, int formatId, int measurandId, int phaseId
            , int locationId, int unitId, int transactionId);