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

using EVCB_OCPP.Domain;
using EVCB_OCPP.Domain.Models.Database;
using EVCB_OCPP.Packet.Messages.Core;
using EVCB_OCPP.WSServer.Helper;
using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using MongoDB.Driver.Core.Connections;
using Newtonsoft.Json;
using OCPPPackage.Profiles;
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace EVCB_OCPP.WSServer.Service;

public interface IMainDbService
{
    Task<string> GetMachineAuthorizationKey(string ChargeBoxId);
    Task<string> GetMachineConfiguration(string ChargeBoxId, string configName);
    Task<string> GetMachineHeartbeatInterval(string ChargeBoxId);
    Task<MachineAndCustomerInfo> GetMachineIdAndCustomerInfo(string ChargeBoxId);
    Task<string> GetMachineSecurityProfile(string ChargeBoxId);
    Task UpdateMachineBasicInfo(string ChargeBoxId, Machine machine);
    Task AddOCMF(OCMF oCMF);
    ValueTask<ConnectorStatus> GetConnectorStatus(string ChargeBoxId, int ConnectorId);
    Task UpdateConnectorStatus(string Id, ConnectorStatus connectorStatus);
    Task AddServerMessage(ServerMessage message);
    Task AddServerMessage(string ChargeBoxId, string OutAction, object OutRequest, string CreatedBy = "", DateTime? CreatedOn = null, string SerialNo = "", string InMessage = "");
}

public class MainDbService : IMainDbService
{
    public MainDbService(
        IDbContextFactory<MainDBContext> contextFactory,
        IMemoryCache memoryCache,
        IConfiguration configuration, 
        ILoggerFactory loggerFactory)
    {
        this.contextFactory = contextFactory;
        this.memoryCache = memoryCache;
        this.loggerFactory = loggerFactory;
        var startupLimit = GetStartupLimit(configuration);
        this.connectionString = configuration.GetConnectionString("MainDBContext");
        this.startupSemaphore = new (startupLimit);

        var opLimit = GetOpLimit(configuration);
        this.opSemaphore = new SemaphoreSlim(opLimit);

        InitUpdateConnectorStatusHandler();
        InitUpdateMachineBasicInfoHandler();
        InitAddServerMessageHandler();
    }

    private readonly IDbContextFactory<MainDBContext> contextFactory;
    private readonly IMemoryCache memoryCache;
    private readonly ILoggerFactory loggerFactory;
    private string connectionString;
    private readonly QueueSemaphore startupSemaphore;
    private readonly SemaphoreSlim opSemaphore;
    private GroupSingleHandler<StatusNotificationParam> statusNotificationHandler;
    private GroupSingleHandler<UpdateMachineBasicInfoParam> updateMachineBasicInfoHandler;
    private GroupSingleHandler<ServerMessage> addServerMessageHandler;

    public async Task<MachineAndCustomerInfo> GetMachineIdAndCustomerInfo(string ChargeBoxId)
    {
        using var semaphoreWrapper = await startupSemaphore.GetToken();
        using var db = contextFactory.CreateDbContext();
        var machine = await db.Machine.Where(x => x.ChargeBoxId == ChargeBoxId && x.IsDelete == false).Select(x => new { x.CustomerId, x.Id }).AsNoTracking().FirstOrDefaultAsync();
        if (machine == null)
        {
            return new MachineAndCustomerInfo(string.Empty, Guid.Empty, "Unknown");
        }
        var customerName = await db.Customer.Where(x => x.Id == machine.CustomerId).Select(x => x.Name).FirstOrDefaultAsync();
        return new MachineAndCustomerInfo(machine.Id, machine.CustomerId, customerName);
    }

    public async Task<string> GetMachineConfiguration(string ChargeBoxId, string configName)
    {
        using var semaphoreWrapper = await startupSemaphore.GetToken();
        using var db = contextFactory.CreateDbContext();
        return await db.MachineConfigurations
            .Where(x => x.ChargeBoxId == ChargeBoxId && x.ConfigureName == configName)
            .Select(x => x.ConfigureSetting).FirstOrDefaultAsync();
    }

    public async Task<string> GetMachineSecurityProfile(string ChargeBoxId)
    {
        return await GetMachineConfiguration(ChargeBoxId, StandardConfiguration.SecurityProfile);
    }

    public async Task<string> GetMachineAuthorizationKey(string ChargeBoxId)
    {
        return await GetMachineConfiguration(ChargeBoxId, StandardConfiguration.AuthorizationKey);
    }

    public async Task<string> GetMachineHeartbeatInterval(string ChargeBoxId)
    {
        return await GetMachineConfiguration(ChargeBoxId, StandardConfiguration.HeartbeatInterval);
    }

    public async Task UpdateMachineBasicInfo(string ChargeBoxId, Machine machine)
    {
        //using var semaphoreWrapper = await startupSemaphore.GetToken();
        //using var db = await contextFactory.CreateDbContextAsync();

        //var _machine = db.Machine.FirstOrDefault(x => x.ChargeBoxId == ChargeBoxId);
        //_machine.ChargeBoxSerialNumber = machine.ChargeBoxSerialNumber;
        //_machine.ChargePointSerialNumber = machine.ChargePointSerialNumber;
        //_machine.ChargePointModel = machine.ChargePointModel;
        //_machine.ChargePointVendor = machine.ChargePointVendor;
        //_machine.FW_CurrentVersion = machine.FW_CurrentVersion;
        //_machine.Iccid = DateTime.UtcNow.ToString("yy-MM-dd HH:mm");
        //_machine.Imsi = machine.Imsi;
        //_machine.MeterSerialNumber = machine.MeterSerialNumber;
        //_machine.MeterType = machine.MeterType;

        //await db.SaveChangesAsync();
        //using var semaphoreWrapper = await startupSemaphore.GetToken();
        await updateMachineBasicInfoHandler.HandleAsync(new UpdateMachineBasicInfoParam(ChargeBoxId, machine));
    }

    public async Task AddOCMF(OCMF oCMF)
    {
        using var db = contextFactory.CreateDbContext();
        db.OCMF.Add(oCMF);
        await db.SaveChangesAsync();
    }

    public async ValueTask<ConnectorStatus> GetConnectorStatus(string ChargeBoxId, int ConnectorId)
    {
        var key = $"{ChargeBoxId}{ConnectorId}";
        if (memoryCache.TryGetValue<ConnectorStatus>(key, out var status))
        {
            return status;
        }
        using var db = contextFactory.CreateDbContext();
        var statusFromDb = await db.ConnectorStatus.Where(x => x.ChargeBoxId == ChargeBoxId
                            && x.ConnectorId == ConnectorId).AsNoTracking().FirstOrDefaultAsync();
        memoryCache.Set(key, statusFromDb);
        return statusFromDb;
    }

    public async Task UpdateConnectorStatus(string Id, ConnectorStatus Status)
    {
        //using var db = await contextFactory.CreateDbContextAsync();

        //ConnectorStatus status = new() { Id = Id };

        //db.ChangeTracker.AutoDetectChangesEnabled = false;
        //db.ConnectorStatus.Attach(status);


        //status.CreatedOn = Status.CreatedOn;
        //status.Status = Status.Status;
        //status.ChargePointErrorCodeId = Status.ChargePointErrorCodeId;
        //status.ErrorInfo = Status.ErrorInfo;
        //status.VendorId = Status.VendorId;
        //status.VendorErrorCode = Status.VendorErrorCode;


        //db.Entry(status).Property(x => x.CreatedOn).IsModified = true;
        //db.Entry(status).Property(x => x.Status).IsModified = true;
        //db.Entry(status).Property(x => x.ChargePointErrorCodeId).IsModified = true;
        //db.Entry(status).Property(x => x.ErrorInfo).IsModified = true;
        //db.Entry(status).Property(x => x.VendorId).IsModified = true;
        //db.Entry(status).Property(x => x.VendorErrorCode).IsModified = true;

        //await db.SaveChangesAsync();
        await statusNotificationHandler.HandleAsync(new StatusNotificationParam(Id, Status));
        var key = $"{Status.ChargeBoxId}{Status.ConnectorId}";
        if (memoryCache.TryGetValue<ConnectorStatus>(key, out _))
        {
            memoryCache.Remove(key);
        }
        memoryCache.Set(key, Status);
        return;
    }

    public Task AddServerMessage(string ChargeBoxId, string OutAction, object OutRequest, string CreatedBy, DateTime? CreatedOn = null, string SerialNo = "", string InMessage = "")
    {
        if (string.IsNullOrEmpty(CreatedBy))
        {
            CreatedBy = "Server";
        }

        if (string.IsNullOrEmpty(SerialNo))
        {
            SerialNo = Guid.NewGuid().ToString();
        }
        var _CreatedOn = CreatedOn ?? DateTime.UtcNow;

        string _OutRequest = "";
        if (OutRequest is not null)
        {
            _OutRequest = JsonConvert.SerializeObject(
                OutRequest,
                new JsonSerializerSettings()
                {
                    NullValueHandling = NullValueHandling.Ignore,
                    Formatting = Formatting.None
                });
        }

        return AddServerMessage(new ServerMessage()
        {
            ChargeBoxId = ChargeBoxId,
            CreatedBy = CreatedBy,
            CreatedOn = _CreatedOn,
            OutAction = OutAction,
            OutRequest = _OutRequest,
            SerialNo = SerialNo,
            InMessage = InMessage
        });
    }

    public Task AddServerMessage(ServerMessage message)
    {
        return addServerMessageHandler.HandleAsync(message);
    }

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

        updateMachineBasicInfoHandler = new GroupSingleHandler<UpdateMachineBasicInfoParam>(
            handleFunc: BundelUpdateMachineBasicInfo,
            logger: loggerFactory.CreateLogger("UpdateMachineBasicInfoHandler"),
            workerCnt: 10);
    }

    private async Task BundelUpdateMachineBasicInfo(IEnumerable<UpdateMachineBasicInfoParam> pams)
    {
        using var db = await contextFactory.CreateDbContextAsync();
        using var trans = await db.Database.BeginTransactionAsync();

        pams = pams.DistinctBy(x => x.ChargeBoxId);

        foreach (var pam in pams)
        {
            var _machine = db.Machine.FirstOrDefault(x => x.ChargeBoxId == pam.ChargeBoxId);
            _machine.ChargeBoxSerialNumber = pam.machine.ChargeBoxSerialNumber;
            _machine.ChargePointSerialNumber = pam.machine.ChargePointSerialNumber;
            _machine.ChargePointModel = pam.machine.ChargePointModel;
            _machine.ChargePointVendor = pam.machine.ChargePointVendor;
            _machine.FW_CurrentVersion = pam.machine.FW_CurrentVersion;
            _machine.Iccid = DateTime.UtcNow.ToString("yy-MM-dd HH:mm");
            _machine.Imsi = pam.machine.Imsi;
            _machine.MeterSerialNumber = pam.machine.MeterSerialNumber;
            _machine.MeterType = pam.machine.MeterType;
        }

        db.SaveChanges();
        trans.Commit();
    }

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

        statusNotificationHandler = new GroupSingleHandler<StatusNotificationParam>(
            handleFunc: BundleUpdateConnectorStatus,
            logger: loggerFactory.CreateLogger("StatusNotificationHandler"),
            workerCnt: 10);
    }

    private async Task BundleUpdateConnectorStatus(IEnumerable<StatusNotificationParam> statusNotifications)
    {
        using var db = await contextFactory.CreateDbContextAsync();
        using var trans = await db.Database.BeginTransactionAsync();

        statusNotifications = statusNotifications.OrderBy(x => x.Status.CreatedOn).DistinctBy(x => x.Id);

        foreach (var param in statusNotifications)
        {
            ConnectorStatus status = new() { Id = param.Id };

            //db.ChangeTracker.AutoDetectChangesEnabled = false;
            db.ConnectorStatus.Attach(status);


            status.CreatedOn = param.Status.CreatedOn;
            status.Status = param.Status.Status;
            status.ChargePointErrorCodeId = param.Status.ChargePointErrorCodeId;
            status.ErrorInfo = param.Status.ErrorInfo;
            status.VendorId = param.Status.VendorId;
            status.VendorErrorCode = param.Status.VendorErrorCode;


            db.Entry(status).Property(x => x.CreatedOn).IsModified = true;
            db.Entry(status).Property(x => x.Status).IsModified = true;
            db.Entry(status).Property(x => x.ChargePointErrorCodeId).IsModified = true;
            db.Entry(status).Property(x => x.ErrorInfo).IsModified = true;
            db.Entry(status).Property(x => x.VendorId).IsModified = true;
            db.Entry(status).Property(x => x.VendorErrorCode).IsModified = true;

            //db.SaveChanges();
        }

        db.SaveChanges();
        trans.Commit();
        db.ChangeTracker.Clear();
    }

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

        addServerMessageHandler = new GroupSingleHandler<ServerMessage>(
            handleFunc: BulkInsertServerMessage,
            logger: loggerFactory.CreateLogger("AddServerMessageHandler"));
    }

    private async Task BundleAddServerMessage(IEnumerable<ServerMessage> messages)
    {
        using var db = await contextFactory.CreateDbContextAsync();
        using var trans = await db.Database.BeginTransactionAsync();

        foreach (var message in messages)
        {
            db.ServerMessage.Add(message);
        }

        db.SaveChanges();
        trans.Commit();
        db.ChangeTracker.Clear();
    }

    private Task BulkInsertServerMessage(IEnumerable<ServerMessage> messages)
    {
        var table = new DataTable();
        table.Columns.Add("ChargeBoxId");
        table.Columns.Add("SerialNo");
        table.Columns.Add("OutAction");
        table.Columns.Add("OutRequest");
        table.Columns.Add("InMessage");
        table.Columns.Add("CreatedOn");
        table.Columns.Add("CreatedBy");
        table.Columns.Add("UpdatedOn");
        table.Columns.Add("ReceivedOn");

        foreach (var param in messages)
        {
            var row = table.NewRow();
            row["ChargeBoxId"] = param.ChargeBoxId;
            row["SerialNo"] = param.SerialNo;
            row["OutAction"] = param.OutAction;
            row["OutRequest"] = param.OutRequest;
            row["InMessage"] = param.InMessage;
            row["CreatedOn"] = param.CreatedOn;
            row["CreatedBy"] = param.CreatedBy;
            row["UpdatedOn"] = param.UpdatedOn;
            row["ReceivedOn"] = param.ReceivedOn;

            table.Rows.Add(row);
        }

        using SqlConnection sqlConnection = new SqlConnection(connectionString);
        sqlConnection.Open();
        using SqlBulkCopy sqlBulkCopy = new SqlBulkCopy(sqlConnection);

        sqlBulkCopy.BatchSize = messages.Count();
        sqlBulkCopy.DestinationTableName = "ServerMessage";

        sqlBulkCopy.ColumnMappings.Add("ChargeBoxId", "ChargeBoxId");
        sqlBulkCopy.ColumnMappings.Add("SerialNo", "SerialNo");
        sqlBulkCopy.ColumnMappings.Add("OutAction", "OutAction");
        sqlBulkCopy.ColumnMappings.Add("OutRequest", "OutRequest");
        sqlBulkCopy.ColumnMappings.Add("InMessage", "InMessage");
        sqlBulkCopy.ColumnMappings.Add("CreatedOn", "CreatedOn");
        sqlBulkCopy.ColumnMappings.Add("CreatedBy", "CreatedBy");
        sqlBulkCopy.ColumnMappings.Add("UpdatedOn", "UpdatedOn");
        sqlBulkCopy.ColumnMappings.Add("ReceivedOn", "ReceivedOn");

        return sqlBulkCopy.WriteToServerAsync(table);
    }

    private int GetStartupLimit(IConfiguration configuration)
    {
        var limitConfig = configuration["MainDbStartupLimit"];
        int limit = 5;
        if (limitConfig != default)
        {
            int.TryParse(limitConfig, out limit);
        }
        return limit;
    }
    private int GetOpLimit(IConfiguration configuration)
    {
        var limitConfig = configuration["MainDbOpLimit"];
        int limit = 500;
        if (limitConfig != default)
        {
            int.TryParse(limitConfig, out limit);
        }
        return limit;
    }
}

public record MachineAndCustomerInfo (string MachineId, Guid CustomerId, string CustomerName);
public record StatusNotificationParam(string Id, ConnectorStatus Status);
public record UpdateMachineBasicInfoParam(string ChargeBoxId, Machine machine);