iot/projects/IoTCenter/Services/IoTCenterEventHandler.cs

using Application.Domain.Entities;
using Application.Models;
using Infrastructure.Application.Services.Settings;
using Infrastructure.Data;
using Infrastructure.Events;
using Infrastructure.Extensions;
using IoT.Shared.Services.IoTCenter;
using IoTCenter.Application.Domain;
using Jint;
using Jint.Native;
using Microsoft.AspNetCore.SignalR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Vibrant.InfluxDB.Client;
using Vibrant.InfluxDB.Client.Rows;

namespace IoTCenter.Services
{
    public class IoTCenterEventHandler :
        IEventHander<EntityInsertedEvent<Organ>>,
        IEventHander<EntityUpdatedEvent<Organ>>,
        IEventHander<EntityInsertedEvent<NodeCategory>>,
        IEventHander<EntityUpdatedEvent<NodeCategory>>,
        IEventHander<EntityInsertedEvent<SceneTimer>>,
        IEventHander<EntityUpdatedEvent<SceneTimer>>,
        IEventHander<EntityDeletedEvent<SceneTimer>>,
        IEventHander<EntityInsertedEvent<Node>>,
        IEventHander<EntityUpdatedEvent<Node>>,
        IEventHander<EntityDeletedEvent<Node>>,
        IEventHander<EntityInsertedEvent<Product>>,
        IEventHander<EntityUpdatedEvent<Product>>,
        IEventHander<EntityDeletedEvent<Product>>,
        IEventHander<EntityInsertedEvent<Device>>,
        IEventHander<EntityUpdatedEvent<Device>>,
        IEventHander<EntityDeletedEvent<Device>>,
        IEventHander<EntityInsertedEvent<Data>>,
        IEventHander<EntityUpdatedEvent<Data>>,
        IEventHander<EntityDeletedEvent<Data>>,
        IEventHander<EntityInsertedEvent<Command>>,
        IEventHander<EntityUpdatedEvent<Command>>,
        IEventHander<EntityDeletedEvent<Command>>,
        IEventHander<EntityInsertedEvent<Scene>>,
        IEventHander<EntityUpdatedEvent<Scene>>,
        IEventHander<EntityDeletedEvent<Scene>>,
        IEventHander<NodeClientConnectedEvent>
    {
        private readonly IConfiguration _cfg;
        private readonly ISettingService _settingService;
        private readonly ILogger<IoTCenterEventHandler> _logger;
        private readonly IRepository<Node> _nodeRepo;
        private readonly IRepository<NodeCategory> _nodeCategoryRepo;
        private readonly IRepository<NodeCategoryNode> _nodeCategoryNodeRepo;
        private readonly IRepository<Organ> _organRepo;
        private readonly IRepository<OrganNode> _organNodeRepo;
        private readonly IRepository<Scene> _sceneRepo;
        private readonly IRepository<Device> _deviceRepo;
        private readonly IRepository<SceneTigger> _sceneTiggerRepo;
        private readonly ISceneTiggerService _sceneTiggerService;
        private readonly IHubContext<IoTCenterHub> _hub;
        private readonly IHttpClientFactory _httpClientFactory;

        public IoTCenterEventHandler(IConfiguration cfg,
            ISettingService settingService,
            ILogger<IoTCenterEventHandler> logger,
            IRepository<Node> nodeRepo,
            IRepository<NodeCategory> nodeCategoryRepo,
            IRepository<NodeCategoryNode> nodeCategoryNodeRepo,
            IRepository<Organ> organRepo,
            IRepository<OrganNode> organNodeRepo,
            IRepository<Scene> sceneRepo,
            IRepository<Device> deviceRepo,
            IRepository<SceneTigger> sceneTiggerRepo,
            ISceneTiggerService sceneTiggerService,
            IHubContext<IoTCenterHub> hub,
            IHttpClientFactory httpClientFactory)
        {
            this._cfg = cfg;
            this._settingService = settingService;
            this._logger = logger;
            this._nodeRepo = nodeRepo;
            this._nodeCategoryRepo = nodeCategoryRepo;
            this._nodeCategoryNodeRepo = nodeCategoryNodeRepo;
            this._organRepo = organRepo;
            this._organNodeRepo = organNodeRepo;
            this._sceneRepo = sceneRepo;
            this._deviceRepo = deviceRepo;
            this._sceneTiggerRepo = sceneTiggerRepo;
            this._sceneTiggerService = sceneTiggerService;
            this._hub = hub;
            this._httpClientFactory = httpClientFactory;
        }

        #region timer

        public void Handle(EntityInsertedEvent<SceneTimer> message)
        {
            var timer = message.Data;
            if (_sceneRepo.ReadOnlyTable().Any(o => o.Id == timer.SceneId && o.NodeId == null))
            {
                var url = _cfg.GetConnectionString("JobServer") + "/RecurringJob/AddOrUpdate";
                var id = timer.Id.ToString();
                this.UpdateJobServer(url, new
                {
                    id,
                    url = $"{_cfg.GetConnectionString("JobCallBack")}/{id}",
                    cron = timer.Cron
                });
            }
        }

        public void Handle(EntityUpdatedEvent<SceneTimer> message)
        {
            var timer = message.Data;
            if (_sceneRepo.ReadOnlyTable().Any(o => o.Id == timer.SceneId && o.NodeId == null))
            {
                var url = _cfg.GetConnectionString("JobServer") + "/RecurringJob/AddOrUpdate";
                var id = timer.Id.ToString();
                this.UpdateJobServer(url, new
                {
                    id,
                    url = $"{_cfg.GetConnectionString("JobCallBack")}/{id}",
                    cron = timer.Cron
                });
            }
        }

        public void Handle(EntityDeletedEvent<SceneTimer> message)
        {
            var timer = message.Data;
            if (_sceneRepo.ReadOnlyTable().Any(o => o.Id == timer.SceneId && o.NodeId == null))
            {
                var url = _cfg.GetConnectionString("JobServer") + $"/RecurringJob/Remove/{timer.Id}";
                this.UpdateJobServer(url);
            }
        }

        #endregion timer

        #region Product

        public void Handle(EntityInsertedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Product> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Product> message)
        {
            this.Notify(message);
        }

        #endregion Product

        #region Node

        public void Handle(EntityInsertedEvent<Node> message)
        {
            this.Notify(message);
            this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Node> message)
        {
            this.Notify(message);
            this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityDeletedEvent<Node> message)
        {
            this.Notify(message);
        }

        #endregion Node

        #region organ

        public void Handle(EntityInsertedEvent<Organ> message)
        {
            this.Notify(message);
            this.UpdateOrganNode(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Organ> message)
        {
            this.Notify(message);
            this.UpdateOrganNode(message.Data);
        }

        #endregion organ

        #region nodeCategory

        public void Handle(EntityInsertedEvent<NodeCategory> message)
        {
            this.Notify(message);
            this.UpdateNodeCategory(message.Data);
        }

        public void Handle(EntityUpdatedEvent<NodeCategory> message)
        {
            this.Notify(message);
            this.UpdateNodeCategory(message.Data);
        }

        #endregion nodeCategory

        #region Device

        public void Handle(EntityInsertedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Device> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Device> message)
        {
            this.Notify(message);
        }

        #endregion Device

        #region Data

        public void Handle(EntityInsertedEvent<Data> message)
        {
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityUpdatedEvent<Data> message)
        {
            message.Data.Device = _deviceRepo.ReadOnlyTable().Include(o => o.Node).Where(o => o.Id == message.Data.DeviceId).FirstOrDefault();
            this.Notify(message);
            this.TiggerHandle(message);
            this.LogData(message.Data);
        }

        public void Handle(EntityDeletedEvent<Data> message)
        {
            this.Notify(message);
        }

        #endregion Data

        #region Command

        public void Handle(EntityInsertedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Command> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Command> message)
        {
            this.Notify(message);
        }

        #endregion Command

        #region Scene

        public void Handle(EntityInsertedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityUpdatedEvent<Scene> message)
        {
            this.Notify(message);
        }

        public void Handle(EntityDeletedEvent<Scene> message)
        {
            this.Notify(message);
        }

        #endregion Scene

        private void UpdateJobServer(string url, object data = null)
        {
            try
            {
                this._logger.LogDebug($"update job server");
                var client = _httpClientFactory.CreateClient();
                using var content = new StringContent(JsonConvert.SerializeObject(data), Encoding.UTF8, "application/json");
                var result = client.PostAsync(url, content).Result;
                this._logger.LogDebug($"{result.StatusCode}:{result.Content.ReadAsStringAsync().Result}");
            }
            catch (Exception ex)
            {
                ex.PrintStack();
                this._logger.LogError(ex.ToString());
            }
        }

        private void TiggerHandle(BaseEvent<Data> message)
        {
            try
            {
                foreach (var tigger in this._sceneTiggerService.GetSceneTiggers())
                {
                    var data = message.Data;
                    if (tigger.DataId == data.Id)
                    {
                        try
                        {
                            var condition = tigger.Condition;
                            var value = data.Value;
                            var engine = new Engine().Execute($"function valid(value){{return {condition};}}");
                            var result = engine.Invoke("valid", value);
                            if (result == JsValue.True)
                            {
                                this.TiggerHandle(tigger.Id);
                            }
                        }
                        catch (Exception ex)
                        {
                            this._logger.LogError(ex.ToString());
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
        }

        private void Notify<T>(BaseEvent<T> message)
        {
            Task.Run(() =>
            {
                try
                {
                    _hub.ServerToClient($"{typeof(T).Name}{message.Arg}", message.Data, "page", null);
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private void TiggerHandle(Guid tiggerId)
        {
            this._logger.LogDebug($"global tigger exec:{tiggerId}");
            var tigger = this._sceneTiggerRepo.ReadOnlyTable()
                .Include(o => o.Scene).ThenInclude(o => o.SceneCommands).ThenInclude(o => o.Command).ThenInclude(o => o.Device).ThenInclude(o => o.Node)
                .FirstOrDefault(o => o.Id == tiggerId);
            foreach (var sceneCommand in tigger.Scene.SceneCommands.OrderBy(o => o.Command.DisplayOrder))
            {
                try
                {
                    this._hub.ServerToClient(Methods.ExecCommand, sceneCommand.CommandId, sceneCommand.Command.Device.Node.Number, null);
                    Delay(sceneCommand.Command.Delay);
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            }
        }

        private void LogData(Data data)
        {
            var url = _cfg["influxdb:url"];
            var usr = _cfg["influxdb:usr"];
            var pwd = _cfg["influxdb:pwd"];
            var device = _deviceRepo.ReadOnlyTable().Include(o => o.Node).FirstOrDefault(o => o.Id == data.DeviceId);
            Task.Run(async () =>
            {
                try
                {
                    if (string.IsNullOrEmpty(data.Value))
                    {
                        return;
                    }
                    if (data.Type != DeviceDataType.Int && data.Type != DeviceDataType.Float)
                    {
                        return;
                    }
                    try
                    {
                        var dbName = "iot";
                        var measurementName = "data";
                        using var client = new InfluxClient(new Uri(url), usr, pwd);
                        await client.CreateDatabaseAsync(dbName);
                        var row = new DynamicInfluxRow
                        {
                            Timestamp = DateTimeOffset.FromUnixTimeMilliseconds(data.Timestamp).DateTime
                        };
                        row.Fields.Add("OrganName", device.Node.Name);
                        row.Fields.Add("OrganNumber", device.Node.Number);
                        row.Fields.Add("DeviceNumber", device.Number);
                        row.Fields.Add("DeviceName", device.Name);
                        row.Fields.Add(data.Key, this.GetDataValue(data));
                        await client.WriteAsync(dbName, measurementName, new List<DynamicInfluxRow> { row });
                    }
                    catch (Exception ex)
                    {
                        ex.PrintStack();
                    }
                }
                catch (Exception ex)
                {
                    ex.PrintStack();
                }
            });
        }

        private object GetDataValue(Data model)
        {
            return model.Type switch
            {
                DeviceDataType.Int => Convert.ToInt32(model.Value),

                DeviceDataType.Float => Convert.ToSingle(model.Value),

                _ => model.Value,
            };
        }

        private void Delay(int commandDelay)
        {
            var delay = 0;
            try
            {
                delay = Convert.ToInt32(this._settingService.GetSetting("delay").Value);
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
            if (commandDelay > 0)
            {
                delay += commandDelay;
            }
            if (delay > 0)
            {
                Thread.Sleep(delay);
            }
        }

        public void Handle(NodeClientConnectedEvent message)
        {
            var organ = this._organRepo.Table().FirstOrDefault(o => o.Number == message.OrganNumber);
            if (organ == null)
            {
                organ = new Organ
                {
                    Name = message.OrganNumber,
                    Number = message.OrganNumber,
                    Image = "/images/classroom.png"
                };
                this._organRepo.Add(organ);
                this._organRepo.SaveChanges();
            }
            if (!this._organNodeRepo.Table().Any(o => o.Organ.Number == message.OrganNumber && o.Node.Number == message.NodeNumber))
            {
                var node = this._nodeRepo.Table().FirstOrDefault(o => o.Number == message.NodeNumber);
                if (node != null)
                {
                    this._organNodeRepo.Add(new OrganNode { NodeId = node.Id, OrganId = organ.Id });
                    this._organNodeRepo.SaveChanges();
                }
            }
        }

        private void UpdateOrganNode(Node node)
        {
            try
            {
                var category = this._nodeCategoryRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == node.CategoryNumber);
                if (category != null)
                {
                    var nodeCategory = this._nodeCategoryNodeRepo.Table().FirstOrDefault(o => o.NodeId == node.Id);
                    if (nodeCategory == null)
                    {
                        nodeCategory = new NodeCategoryNode { CategoryId = category.Id, NodeId = node.Id };
                        this._nodeCategoryNodeRepo.Add(nodeCategory);
                        this._nodeCategoryNodeRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
            try
            {
                var organ = this._organRepo.ReadOnlyTable().FirstOrDefault(o => o.Number == node.OrganNumber);
                if (organ != null)
                {
                    var organNode = this._organNodeRepo.Table().FirstOrDefault(o => o.NodeId == node.Id);
                    if (organNode == null)
                    {
                        organNode = new OrganNode { OrganId = organ.Id, NodeId = node.Id };
                        this._organNodeRepo.Add(organNode);
                        this._organNodeRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
        }

        private void UpdateOrganNode(Organ organ)
        {
            try
            {
                var nodes = this._nodeRepo.ReadOnlyTable().Where(o => o.OrganNumber == organ.Number).Select(o => o.Id).ToList();
                foreach (var nodeId in nodes)
                {
                    var organNode = this._organNodeRepo.Table().FirstOrDefault(o => o.NodeId == nodeId);
                    if (organNode == null)
                    {
                        organNode = new OrganNode { OrganId = organ.Id, NodeId = nodeId };
                        this._organNodeRepo.Add(organNode);
                        this._organNodeRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
        }

        private void UpdateNodeCategory(NodeCategory category)
        {
            try
            {
                var nodes = this._nodeRepo.ReadOnlyTable().Where(o => o.CategoryNumber == category.Number).Select(o => o.Id).ToList();
                foreach (var nodeId in nodes)
                {
                    var categoryNode = this._nodeCategoryNodeRepo.Table().FirstOrDefault(o => o.NodeId == nodeId);
                    if (categoryNode == null)
                    {
                        categoryNode = new NodeCategoryNode { CategoryId = category.Id, NodeId = nodeId };
                        this._nodeCategoryNodeRepo.Add(categoryNode);
                        this._nodeCategoryNodeRepo.SaveChanges();
                    }
                }
            }
            catch (Exception ex)
            {
                this._logger.LogError(ex.ToString());
            }
        }
    }
}