using SJZY.Track.Application.Services.Warning.Model; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate; using SJZY.Track.Domain.Aggregates.WarningAggregate.Complete; using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Dtos.TrackDetails; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.Services.Warning.Handlers; public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransientDependency { private readonly IOrderRepository _orderRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IWarningRuleRepository _warningRuleRepository; private readonly IWarningTaskRepository _warningTaskRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; public CompleteWarningHandler(IWarningRuleRepository warningRuleRepository, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderRepository orderRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IWarningTaskRepository warningTaskRepository) { _warningRuleRepository = warningRuleRepository; _trackDetailsEsRepository = trackDetailsEsRepository; _orderRepository = orderRepository; _warningDetailsEsRepository = warningDetailsEsRepository; _warningTaskRepository = warningTaskRepository; } #region todo: 预警信号批处理-删除轨迹 public async Task TrackDeletedHandleAsync(List deletedSignals) { // 1.只处理未删除的订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id)); if (orders.Count == 0) return; //不存在符合条件的订单,直接提交信号丢弃 // 2.查出订单指定状态的轨迹(tips:为了指定状态的轨迹是真的被清空了) var orderIds = orders.Select(x => x.Id).ToHashSet(); var orderIdTrackStatusIdsDict = deletedSignals .Where(x => orderIds.Contains(x.OrderId)) // 过滤掉不必要的信号 .GroupBy(x => x.OrderId) .ToDictionary( x => x.Key, y => y.Select(x => x.DeletedTrack.TrackStatusId) .ToHashSet() ); var orderTracksDict = await _trackDetailsEsRepository.GetOrderTracksDictAsync(new GetOrderTracksDictVO { OrderIds = orderIdSet }); // 构建k:orderId,v:[k:trackStatusId,v:true] var orderIdTrackStatusGroupsDict = orderTracksDict .ToDictionary(x => x.Key, y => y.Value .GroupBy(x => x.TrackStatusId) .ToDictionary(a => a.Key, _ => true) ); // 3.查出预警明细(待处理) var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO { OrderIds = orderIdSet, }); if (warningDetails.Count == 0) return; // 查询规则项目 var ruleItemIds = warningDetails.Select(x => x.CompleteWarningRuleItemId).ToHashSet(); var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds); var ruleItemDict = ruleItems.ToDictionary(x => x.Id); // 4.处理明细 var closeDetailIds = new HashSet(); var updateDetails = new List(); foreach (var detail in warningDetails) { var ruleItem = ruleItemDict.GetValueOrDefault(detail.CompleteWarningRuleItemId); // 规则配置不存在直接关闭 if (ruleItem == null) { closeDetailIds.Add(detail.Id); continue; } var trackStatusIdDict = orderIdTrackStatusGroupsDict.GetValueOrDefault(detail.OrderId); // 订单无任何轨迹,或者不存在检查点状态轨迹 if (trackStatusIdDict == null || !trackStatusIdDict.ContainsKey(detail.CheckPointTrackStatusId)) { closeDetailIds.Add(detail.Id); continue; } // 查看缺失轨迹状态表达式是否有变化,如果有则更新 var trackStatusIds = trackStatusIdDict.Keys.ToHashSet(); var expression = ruleItem.BuildMissingTrackStatusIdExpression(trackStatusIds); if (!string.Equals(detail.MissingTrackStatusIds, expression)) { detail.MissingTrackStatusIds = expression; updateDetails.Add(detail); } // todo:需要处理要求轨迹完整的情况? } if (updateDetails.Count > 0) await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails); if (closeDetailIds.Count == 0) return; // 5.预警信息状态修改为【已关闭】 await _warningDetailsEsRepository.BulkCloseCompleteWarningDetailsAsync(new BulkCloseCompleteWarningDetailsVO { DetailsIds = closeDetailIds }); // 6.关闭预警通知任务 await _warningTaskRepository.CloseWarningNotifyTaskAsync(x => !x.IsClosed && closeDetailIds.Contains(x.WarningDetailsId) && x.WarningType == WarningRuleType.完整预警.GetHashCode() ); } #endregion #region todo: 预警信号批处理-新增轨迹 public async Task TrackAddedHandleAsync(List addedSignals) { // 1.只处理未删除的订单 var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id)); if (orders.Count == 0) return; //不存在符合条件的订单,直接提交信号丢弃 var orderIds = orders.Select(x => x.Id).ToHashSet(); // 2.查询订单每个轨迹状态的最早轨迹 var orderTracksDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds); // 3.以单的维度处理 // todo:优化流程减少N+1 var addDetails = new List(); var updateDetails = new List(); var closeDetailsIds = new HashSet(); foreach (var order in orders) { var trackStatusIdTrackDict = orderTracksDict.GetValueOrDefault(order.Id); if (trackStatusIdTrackDict == null) continue; // 查询规则主体 var rule = await _warningRuleRepository.GetWarningRuleForEnabledAsync(order.SubSystemId, order.CustomId, WarningRuleType.完整预警); if (rule == null) continue; // 查询规则项目 var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(rule.Id); if (ruleItems.Count == 0) continue; // 匹配规则项目(要生成预警的,已完成的) var trackStatusIds = trackStatusIdTrackDict.Keys.ToHashSet(); var hitRuleItems = new List(); var doneRuleItemIds = new HashSet(); foreach (var ruleItem in ruleItems) { if (ruleItem.CanHitRule(order.OrderType, trackStatusIds)) hitRuleItems.Add(ruleItem); if (ruleItem.IsDone(order.OrderType, trackStatusIds)) doneRuleItemIds.Add(ruleItem.Id); } if (hitRuleItems.Count == 0 && doneRuleItemIds.Count == 0) return; var hitRuleItemDict = hitRuleItems.ToDictionary(x => x.Id); // 查出规则已经生成预警信息 var hitRuleItemIds = hitRuleItemDict.Keys.ToHashSet(); var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO { OrderIds = new HashSet(1) { order.Id }, RuleItemIds = new HashSet() .Union(hitRuleItemIds) .Union(doneRuleItemIds) .ToHashSet(), }); // 已完结的预警规则,稍后修改明细状态为【已处理】 closeDetailsIds.UnionWith(warningDetails.Where(x => doneRuleItemIds.Contains(x.CompleteWarningRuleItemId)).Select(x => x.Id)); // 处理已存在的预警明细(命中规则的) var existsWarningDetails = warningDetails.Where(x => hitRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList(); var needUpdateDetails = HandleExistsDetails(trackStatusIds, existsWarningDetails, hitRuleItemDict); updateDetails.AddRange(needUpdateDetails); // 排除已经生成预警信息的规则 hitRuleItemIds.ExceptWith(existsWarningDetails.Select(x => x.CompleteWarningRuleItemId)); // 生成预警明细 if (hitRuleItemIds.Count == 0) continue; var needAddDetails = hitRuleItems .Where(x => hitRuleItemIds.Contains(x.Id)) .Select(x => BuildCompleteWarningDetails(order, trackStatusIds, x, rule.Version)); addDetails.AddRange(needAddDetails); } // 4.预警由于不缺失轨迹变为失效,修改明细状态为【已处理】,并关闭相关预警通知任务 if (closeDetailsIds.Count > 0) await CloseWarningDetailLinksAsync(closeDetailsIds, CompleteWarningStatus.已处理); // 更新预警明细 if (updateDetails.Count > 0) await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails); if (addDetails.Count == 0) return; // 5.批量新增预警明细 await _warningDetailsEsRepository.BulkInsertCompleteWarningDetailsAsync(addDetails); // 6.批量新增预警通知任务 var notifyTasks = BuildWarningNotifyTask(addDetails); await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(notifyTasks); } private static List HandleExistsDetails( HashSet trackStatusIds, List details, Dictionary ruleItemDict ) { var updateDetails = new List(); foreach (var detail in details) { var ruleItem = ruleItemDict[detail.CompleteWarningRuleItemId]; // tips:一定存在 var expression = ruleItem.BuildMissingTrackStatusIdExpression(trackStatusIds); // 忽略状态表达式无变化的 if (string.Equals(detail.MissingTrackStatusIds, expression)) continue; detail.MissingTrackStatusIds = expression; updateDetails.Add(detail); } return updateDetails; } private static CompleteWarningDetailsIndex BuildCompleteWarningDetails(OrderEntity order, HashSet trackStatusIds, CompleteWarningRuleItemEntity ruleItem, string version) { return new CompleteWarningDetailsIndex() { Id = Guid.NewGuid().ToString("N"), WarningId = ruleItem.WarningId, CompleteWarningRuleItemId = ruleItem.Id, Version = version, // NotifyModes = ruleItem.NotifyModes, NotifyUsers = ruleItem.NotifyUsers, CheckPointTrackStatusId = ruleItem.CheckPointTrackStatusId, MissingTrackStatusIds = ruleItem.BuildMissingTrackStatusIdExpression(trackStatusIds), SubSystemId = order.SubSystemId, CustomId = order.CustomId, OrderId = order.Id, OrderNo = order.OrderNo, OrderType = order.OrderType, Status = CompleteWarningStatus.待处理.GetHashCode(), CreateTime = DateTime.Now, ProcessedTime = null }; } private static List BuildWarningNotifyTask(List details) { return details .Select(x => new WarningNotifyTaskEntity( x.OrderId, x.WarningId, x.CompleteWarningRuleItemId, x.Id, WarningRuleType.完整预警.GetHashCode(), x.NotifyUsers ) ).ToList(); } #endregion #region todo-test: 预警主体关闭 public async Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule) { if (rule.IsEnabled) return; // 1.获取所有规则项目 var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(rule.Id); if (ruleItems.Count == 0) return; // 2.关闭预警规则关联的所有信息(包含明细,通知任务) var ruleItemIds = ruleItems.Select(x => x.Id).ToHashSet(); await CloseWarningRuleLinkAsync(ruleItemIds); } #endregion #region todo-test: 规则项目变更 public async Task HandleEditAsync(WarningRuleEntity rule, List postRuleItems) { var addRuleItems = postRuleItems.Where(x => x.Id == 0).ToList(); // 1.找到之前的配置项目 var entities = await _warningRuleRepository.GetCompleteRuleItemsAsync(rule.Id); var oldRuleItems = CompleteWarningRuleItemBO.From(entities); // 1.1之前的配置数量为0,做只新增处理 if (oldRuleItems.Count == 0) { if (addRuleItems.Count > 0) await OnlyAddHandleAsync(rule, addRuleItems); return; } // 1.2之前的配置数量不为0,提交的数量为0,做只删除处理 if (oldRuleItems.Count > 0 && postRuleItems.Count == 0) { await OnlyDeleteHandleAsync(rule, oldRuleItems); return; } // 修改/关注/删除 var updateRuleItems = new List(); var followRuleItems = new List(); var deleteRuleItems = new List(); // 构建id配置项字典 var oldDict = oldRuleItems.ToDictionary(x => x.Id); var updateAndAddRuleItems = postRuleItems.Where(x => x.Id == 0).ToList(); var postDict = postRuleItems .Except(updateAndAddRuleItems) //tips:差异化比较时忽略新增的规则条目 .ToDictionary(x => x.Id); // 2.差异化比较 var ruleItemIds = oldDict.Keys.Concat(postDict.Keys).ToHashSet(); foreach (var ruleItemId in ruleItemIds) { var oldRule = oldDict.GetValueOrDefault(ruleItemId); var postRule = postDict.GetValueOrDefault(ruleItemId); if (oldRule != null && postRule != null) { // 规则属性变更关注(状态减少,时效变更) var (follow, change) = CanNeedFollowAndChange(oldRule, postRule); if (follow) followRuleItems.Add(postRule); //需要关注的 if (change) updateRuleItems.Add(postRule); //更新 continue; } // 规则条目被删除 if (oldRule != null && postRule == null) { followRuleItems.Add(oldRule); //删除一定需要关注 deleteRuleItems.Add(oldRule); //删除 } } // 3.处理变更 await HandleRuleItemsChangeAsync(rule, addRuleItems, updateRuleItems, followRuleItems, deleteRuleItems); } private async Task OnlyDeleteHandleAsync(WarningRuleEntity rule, List bos) { var ruleItemIds = bos.Select(x => x.Id).ToHashSet(); await _warningRuleRepository.DeleteCompleteRuleItemsAsync(ruleItemIds); // 更新规则主体版本号 await _warningRuleRepository.UpdateVersionAsync(rule); // 关闭预警规则关联的所有信息(包含明细,通知任务) await CloseWarningRuleLinkAsync(ruleItemIds); } private async Task OnlyAddHandleAsync(WarningRuleEntity rule, List bos) { var entities = CompleteWarningRuleItemBO.ToEntities(bos); await _warningRuleRepository.BulkInsertCompleteRuleItemsAsync(entities); // 更新规则主体版本号 await _warningRuleRepository.UpdateVersionAsync(rule); } private async Task HandleRuleItemsChangeAsync( WarningRuleEntity rule, List addRuleItems, List updateRuleItems, List followRuleItems, List deleteRuleItems ) { // 新增/修改/关注/删除 实体,稍后做处理 if (addRuleItems.Count > 0) { var addEntities = CompleteWarningRuleItemBO.ToEntities(addRuleItems); await _warningRuleRepository.BulkInsertCompleteRuleItemsAsync(addEntities); } if (updateRuleItems.Count > 0) { var updateEntities = CompleteWarningRuleItemBO.ToEntities(updateRuleItems); await _warningRuleRepository.BulkUpdateCompleteRuleItemsAsync(updateEntities); } // 删除规则条目是必须要关注 if (deleteRuleItems.Count > 0) { var deleteRuleItemIds = deleteRuleItems.Select(x => x.Id).ToHashSet(); await _warningRuleRepository.DeleteCompleteRuleItemsAsync(deleteRuleItemIds); } // 更新规则主体版本号 await _warningRuleRepository.UpdateVersionAsync(rule); if (followRuleItems.Count == 0) return; // 关闭预警规则关联的所有信息(包含明细,通知任务) var ruleItemIds = followRuleItems.Select(x => x.Id).ToHashSet(); await CloseWarningRuleLinkAsync(ruleItemIds); } private static (bool follow, bool change) CanNeedFollowAndChange(CompleteWarningRuleItemBO oldRule, CompleteWarningRuleItemBO postRule) { var defaultResult = (true, true); // 关注的订单类型减少 var oldTypes = oldRule.OrderTypes; var postTypes = postRule.OrderTypes; if (oldTypes.Count > postTypes.Count) //数量减少,一定需要关注 return defaultResult; foreach (var type in oldTypes) { if (!postTypes.Contains(type)) //类型减少,需要关注 return defaultResult; } // 检查点发生变化 if (oldRule.CheckPointTrackStatusId != postRule.CheckPointTrackStatusId) return defaultResult; // 要求的轨迹状态, // 条目内或关系状态变更,比如 案例1:S1/S2,S4 => S1/S3,案例2: S1/S2,S4 => S1,S4 // 条目减少,S1/S2,S4 => S1/S2 var oldStatusIds = oldRule.RequireTrackStatus.SelectMany(x => x).ToHashSet(); var postStatusIds = postRule.RequireTrackStatus.SelectMany(x => x).ToHashSet(); if (oldStatusIds.Count > postStatusIds.Count) //数量减少,一定需要关注 return defaultResult; foreach (var statusId in oldStatusIds) { if (!postStatusIds.Contains(statusId)) //状态减少,需要关注 return defaultResult; } // 通知对象 var oldNotifyUserIds = oldRule.NotifyUsers.Select(x => x.UserId).ToHashSet(); var postNotifyUserIds = postRule.NotifyUsers.Select(x => x.UserId).ToHashSet(); var hasChange = !oldTypes.SetEquals(postTypes) || !oldStatusIds.SetEquals(postStatusIds) || !oldNotifyUserIds.SetEquals(postNotifyUserIds); return (false, hasChange); } #endregion todo: 规则项目变更 // 关闭预警规则关联的信息(预警明细,重试任务,通知任务) private async Task CloseWarningRuleLinkAsync(HashSet ruleItemIds) { if (ruleItemIds.Count == 0) return; // 1.获取所有完整预警明细 var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO { RuleItemIds = ruleItemIds }); if (warningDetails.Count == 0) return; // 2.修改明细状态为【已关闭】,关闭预警通知任务 var detailIds = warningDetails.Select(x => x.Id).ToHashSet(); await CloseWarningDetailLinksAsync(detailIds); } // 关闭预警明细关联的信息(通知任务) private async Task CloseWarningDetailLinksAsync(HashSet detailIds, CompleteWarningStatus status = CompleteWarningStatus.已关闭) { if (detailIds.Count == 0) return; // 1.修改明细状态为【已关闭】 await _warningDetailsEsRepository.BulkCloseCompleteWarningDetailsAsync(new BulkCloseCompleteWarningDetailsVO { DetailsIds = detailIds, SetStatus = status }); // 2.关闭所有相关预警通知任务 await _warningTaskRepository.CloseWarningNotifyTaskAsync(x => !x.IsClosed && detailIds.Contains(x.WarningDetailsId) && x.WarningType == WarningRuleType.完整预警.GetHashCode() ); } }