TEAMModelOS/TEAMModelFunction/MonitorServicesBus.cs

using System;
using System.Collections.Generic;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
using Azure.Cosmos;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;
using TEAMModelOS.SDK.DI;
using TEAMModelOS.SDK.Extension;
using TEAMModelOS.SDK;
using TEAMModelOS.SDK.Models;
using TEAMModelOS.SDK.Models.Cosmos;
using TEAMModelOS.SDK.Models.Cosmos.Common;

namespace TEAMModelFunction
{
    public class MonitorServicesBus
    {
        
        private readonly AzureCosmosFactory _azureCosmos;
        private readonly DingDing _dingDing;
        private readonly AzureStorageFactory _azureStorage;
        private readonly AzureRedisFactory _azureRedis;
        private readonly AzureServiceBusFactory _serviceBus;
        public MonitorServicesBus(AzureCosmosFactory azureCosmos, DingDing dingDing, AzureStorageFactory azureStorage   , AzureRedisFactory azureRedis, AzureServiceBusFactory serviceBus)
        {
            _azureCosmos = azureCosmos;
            _dingDing = dingDing;
            _azureStorage = azureStorage;
            _azureRedis = azureRedis;
            _serviceBus = serviceBus;
        }
        [FunctionName("Exam")]
         public async Task Exam([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "exam", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            try
            {
                var json = JsonDocument.Parse(msg);
                json.RootElement.TryGetProperty("id", out JsonElement id);
                json.RootElement.TryGetProperty("progress", out JsonElement progress);
                json.RootElement.TryGetProperty("code", out JsonElement code);
                //Dictionary<string, object> keyValuePairs = mySbMsg.ToObject<Dictionary<string, object>>();
                var client = _azureCosmos.GetCosmosClient();
                ExamInfo exam = await client.GetContainer(Constant.TEAMModelOS, "Common").ReadItemAsync<ExamInfo>(id.ToString(), new PartitionKey($"{code}"));
                exam.progress = progress.ToString();
                await client.GetContainer(Constant.TEAMModelOS, "Common").ReplaceItemAsync(exam, id.ToString(), new PartitionKey($"{code}"));
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,ExamBus()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }

        }
        [FunctionName("Vote")]
        public async Task Vote([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "vote", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            try
            {
                var jsonMsg = JsonDocument.Parse(msg);
                jsonMsg.RootElement.TryGetProperty("id", out JsonElement id);
                jsonMsg.RootElement.TryGetProperty("progress", out JsonElement progress);
                jsonMsg.RootElement.TryGetProperty("code", out JsonElement code);
                var client = _azureCosmos.GetCosmosClient();
                Vote vote = await client.GetContainer(Constant.TEAMModelOS, "Common").ReadItemAsync<Vote>(id.ToString(), new PartitionKey($"{code}"));
                vote.progress = progress.ToString();
                await client.GetContainer(Constant.TEAMModelOS, "Common").ReplaceItemAsync(vote, id.ToString(), new PartitionKey($"{code}"));
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,VoteBus()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }

        }
        [FunctionName("Correct")]
        public async Task Correct([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "correct", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            try
            {
                var jsonMsg = JsonDocument.Parse(msg);
                jsonMsg.RootElement.TryGetProperty("id", out JsonElement id);
                jsonMsg.RootElement.TryGetProperty("progress", out JsonElement progress);
                jsonMsg.RootElement.TryGetProperty("code", out JsonElement code);
                var client = _azureCosmos.GetCosmosClient();
                Correct correct = await client.GetContainer(Constant.TEAMModelOS, "Common").ReadItemAsync<Correct>(id.ToString(), new PartitionKey($"{code}"));
                correct.progress = progress.ToString();
                await client.GetContainer(Constant.TEAMModelOS, "Common").ReplaceItemAsync(correct, id.ToString(), new PartitionKey($"{code}"));
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Correct()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }

        }
        [FunctionName("Survey")]
        public async Task Survey([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "survey", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            try
            {
                var jsonMsg = JsonDocument.Parse(msg);
                jsonMsg.RootElement.TryGetProperty("id", out JsonElement id);
                jsonMsg.RootElement.TryGetProperty("progress", out JsonElement progress);
                jsonMsg.RootElement.TryGetProperty("code", out JsonElement code);
                //Dictionary<string, object> keyValuePairs = mySbMsg.ToObject<Dictionary<string, object>>();
                var client = _azureCosmos.GetCosmosClient();
                Survey survey = await client.GetContainer(Constant.TEAMModelOS, "Common").ReadItemAsync<Survey>(id.ToString(), new PartitionKey($"{code}"));
                survey.progress = progress.ToString();
                await client.GetContainer(Constant.TEAMModelOS, "Common").ReplaceItemAsync(survey, id.ToString(), new PartitionKey($"{code}"));
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,SurveyBus()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }
        }
        [FunctionName("Blob")]
        public async Task Blob([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "blob", Connection = "Azure:ServiceBus:ConnectionString")] string msg) {
            try
            {
               // await _dingDing.SendBotMsg($"ServiceBus,Blob(){msg}", GroupNames.醍摩豆服務運維群組);
                var jsonMsg = JsonDocument.Parse(msg);
                if(jsonMsg.RootElement.TryGetProperty("name", out JsonElement name)&& name.ValueKind==JsonValueKind.String)
                {
                    var client = _azureStorage.GetBlobContainerClient($"{name}");
                    var size = await client.GetBlobsCatalogSize();
                    await _azureRedis.GetRedisClient(8).HashSetAsync($"Blob:Record", new RedisValue($"{name}"), new RedisValue($"{long.Parse($"{size.Item1}")}"));
                    foreach (var key in size.Item2.Keys)
                    {
                        await _azureRedis.GetRedisClient(8).SortedSetRemoveAsync($"Blob:Catalog:{name}", key);
                        await _azureRedis.GetRedisClient(8).SortedSetIncrementAsync($"Blob:Catalog:{name}", key, size.Item2[key].HasValue ? size.Item2[key].Value : 0);
                    }
                    await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Blob() 容器:{name}使用:{size.Item1}，文件分类：{size.Item2.ToJsonString()}",
                            GroupNames.成都开发測試群組);
                }
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Blob()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }
        }
        /// <summary>
        /// 根据容器的根目录刷新redis并获取redis的最新使用情况
        /// </summary>
        /// <param name="msg"></param>
        /// <returns></returns>
        [FunctionName("BlobRoot")]
        public async Task BlobRoot([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "blobroot", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            try
            {

                var jsonMsg = JsonDocument.Parse(msg);
                if (jsonMsg.RootElement.TryGetProperty("name", out JsonElement _name) && _name.ValueKind == JsonValueKind.String
                    && jsonMsg.RootElement.TryGetProperty("root", out JsonElement root) && root.ValueKind == JsonValueKind.String)
                {
                    await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Blob() 容器:触发变更，{jsonMsg.ToJsonString()}",
                      GroupNames.成都开发測試群組);
                    List<Dictionary<string, double?>> list = new List<Dictionary<string, double?>>();
                    string[] uls = System.Web.HttpUtility.UrlDecode($"{root}", Encoding.UTF8).Split("/");
                    string u = !string.IsNullOrEmpty(uls[0]) ? uls[0] : uls[1];
                    string name = $"{_name}";
                    string lockKey = $"Blob:Lock:{name}:{u}";
                    bool exist= await _azureRedis.GetRedisClient(8).KeyExistsAsync(lockKey);
                   
                    if (!exist)
                    {   ///key不存在则正常进行计算
                        bool condition = false;
                        TimeSpan timeSpan = new TimeSpan(DateTimeOffset.UtcNow.AddMinutes(5).Ticks);
                        timeSpan = timeSpan - new TimeSpan(DateTimeOffset.UtcNow.Ticks);
                        //准备处理Blob刷新时间
                        long action = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                        await _azureRedis.GetRedisClient(8).StringSetAsync(lockKey, action, expiry: timeSpan);
                        await RefreshBlob(name, u);
                        //将action 与Redis最新的时间进行比较，如果
                        var rds = await CheckLockKey(lockKey, action);
                        condition = rds.condition;
                        exist = rds.exist;
                        if (condition || !exist) {
                            await RefreshBlob(name, u);
                        }
                        
                        //使用  CancellationToken 
                        //while (condition || !exist)
                        //{

                        //}
                    }
                    else {
                        ///key存在则,则刷新key对应的值
                        TimeSpan timeSpan = new TimeSpan(DateTimeOffset.UtcNow.AddMinutes(5).Ticks);
                        timeSpan = timeSpan - new TimeSpan(DateTimeOffset.UtcNow.Ticks);
                        long now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                        await _azureRedis.GetRedisClient(8).StringSetAsync(lockKey, now, expiry: timeSpan);
                    }
                    await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Blob() 容器:{name}使用:{u}，文件分类：{list.ToJsonString()}",
                        GroupNames.成都开发測試群組);
                }
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-ServiceBus,Blob()\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.醍摩豆服務運維群組);
            }
        }

        private async Task<(bool condition,bool exist)> CheckLockKey(string lockKey,long  nowTime) {
            //Redis的最新时间
            long newestTime = 0;
            RedisValue value = await _azureRedis.GetRedisClient(8).StringGetAsync(lockKey);
            if (value != default && !value.IsNullOrEmpty)
            {
                JsonElement record = value.ToString().ToObject<JsonElement>();
                if (record.TryGetInt64(out newestTime))
                {
                }
            }
            //说明key已经不存在
            if (newestTime == 0)
            {
                return (false, true);
            }
            //说明key存在
            else {
                //说明Redis记录了最新的时间戳
                if (nowTime != newestTime)
                {
                    return (true, false);
                }
                //时间相同，没有被再次记录最新的时间戳
                else
                {
                    await _azureRedis.GetRedisClient(8).KeyDeleteAsync(lockKey);
                    return (false, true);
                }
            }
        }
        private async Task RefreshBlob(string  name ,string  u) {
            long statr = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            var client = _azureStorage.GetBlobContainerClient(name);
            var size = await client.GetBlobsSize(u);
            await _azureRedis.GetRedisClient(8).SortedSetRemoveAsync($"Blob:Catalog:{name}", u);
            await _azureRedis.GetRedisClient(8).SortedSetIncrementAsync($"Blob:Catalog:{name}", u, size.HasValue ? size.Value : 0);
            var scores = await _azureRedis.GetRedisClient(8).SortedSetRangeByRankWithScoresAsync($"Blob:Catalog:{name}");
            double blobsize = 0;
            if (scores != default && scores != null)
            {
                foreach (var score in scores)
                {
                    blobsize = blobsize + score.Score;
                }
            }
            await _azureRedis.GetRedisClient(8).HashSetAsync($"Blob:Record", new RedisValue(name), new RedisValue($"{blobsize}"));
            long end = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            long dis = (end - statr)/1000;
            long timeout = 10;
            if (dis> timeout) {
                await _dingDing.SendBotMsg($"ServiceBus,RefreshBlob:空间计算已经超过{timeout}秒\n容器名：{name}\n文件夹：{u}\n计算时长：{dis}", GroupNames.醍摩豆服務運維群組);
            }
        }
        /// <summary>
        /// 完善课程变更,StuListChange,  originCode是学校编码 则表示名单是学校自定义名单，如果是tmdid则表示醍摩豆的私有名单，scope=school,private。
        /// </summary>
        /// <data msg>
        /// CourseChange
        ///// </data>
        /// <param name="msg"></param>
        /// <returns></returns>
        [FunctionName("StuList")]
        public async Task StuList([ServiceBusTrigger("%Azure:ServiceBus:ActiveTask%", "stulist", Connection = "Azure:ServiceBus:ConnectionString")] string msg)
        {
            var client = _azureCosmos.GetCosmosClient();
            try
            {
              
                var jsonMsg = JsonDocument.Parse(msg);
                StuListChange stuListChange = msg.ToObject<StuListChange>();
                //名单变动修改学生课程关联信息
                //await StuListService.FixStuCourse(client, stuListChange);
                //Vote投票 Survey问卷 Exam评测 Learn学习活动 Homework作业活动
                //名单变动修改学生问卷关联信息
                await StuListService.FixActivity(client, _dingDing, stuListChange, "Survey");
                //名单变动修改学生投票关联信息
                await StuListService.FixActivity(client, _dingDing, stuListChange, "Vote");
                //名单变动修改学生评测关联信息
                await StuListService.FixActivity(client, _dingDing, stuListChange, "Exam");
                //TODO学习活动
                //await FixActivity(client, stuListChange, "Learn");
                //TODO作业活动
                // await FixActivity(client, stuListChange, "Homework");
                //TODO课程名单变动修改学生课程关联信息
                await StuListService.FixStuCourse(client, _dingDing, stuListChange);
            }
            catch (Exception ex)
            {
                await _dingDing.SendBotMsg($"{Environment.GetEnvironmentVariable("Option:Location")}-StuListServiceBus-StuList\n{ex.Message}\n{ex.StackTrace}\n{msg}", GroupNames.成都开发測試群組);
            }
        }


    }
}