using Microsoft.Extensions.Options; using TrackCenter.Application.Dtos.Warning; using TrackCenter.Application.Options; using TrackCenter.Application.Services.Warning.Model; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Aggregates.WarningAggregate; using TrackCenter.Domain.Dtos.Warning; namespace TrackCenter.Application.Hangfire; /// /// 预警通知聚合器 /// public sealed class WarningNotifyAggregator : ITransientDependency { private readonly IOrderRepository _orderRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IWarningTaskRepository _warningTaskRepository; private readonly IWarningRuleRepository _warningRuleRepository; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly IOptionsMonitor _optionsMonitor; public WarningNotifyAggregator( IWarningTaskRepository warningTaskRepository, IWarningRuleRepository warningRuleRepository, IOrderRepository orderRepository, ITrackDetailsEsRepository trackDetailsEsRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IOptionsMonitor optionsMonitor) { _warningTaskRepository = warningTaskRepository; _warningRuleRepository = warningRuleRepository; _orderRepository = orderRepository; _trackDetailsEsRepository = trackDetailsEsRepository; _warningDetailsEsRepository = warningDetailsEsRepository; _optionsMonitor = optionsMonitor; } // 1.可行性过滤(不符合的统一关闭) // 2.以人的维度聚合通知信息,并生成配置参数NotifyConfig // 3.构建并调度一个调用通知接口的立即执行任务 // 4.绑定这个任务id(可能是多个)和jobId用户跟踪任务状态 public async Task<(List NotifyConfigs, HashSet taskIds)> ScanNotifyAndAggregateAsync() { // 构建查询游标 todo:全量查询需要优化,直接存缓存? // var options = _optionsMonitor.CurrentValue; // var cursor = DateTime.Now.Subtract(TimeSpan.FromHours(options.ScanWarningNotifyTaskHourInterval)); // 1.扫描出可能进行的任务 var tasks = await _warningTaskRepository.ScanNotifyTaskAsync(); // 2.可行性过滤 var filtered = await FilterAsync(tasks); // 3.以人的维度聚合通知构建NotifyConfig return (BuildNotifyConfigs(filtered), tasks.Select(x => x.Id).ToHashSet()); } // 要求订单存在,预警主体存在过滤 private async Task> FilterAsync(List notifyTasks) { // 1.只通知存在的订单 var orderIdSet = notifyTasks.Select(x => x.OrderId).ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id)); if (orders.Count == 0) return Empty.List; var orderDict = orders.ToDictionary(x => x.Id); // 2.过滤掉不存在的订单 var filtered = notifyTasks.Where(x => orderDict.ContainsKey(x.OrderId)).ToList(); // 3.查询规则主体(存在&启用) var warningSet = filtered.Select(x => x.WarningId).ToHashSet(); var warningIds = (await _warningRuleRepository.GetListAsync( x => warningSet.Contains(x.Id) && x.IsEnabled == true, y => y.Id //不回表 )).ToHashSet(); if (warningIds.Count == 0) return Empty.List; // 4.过滤掉规则主体禁用 filtered = filtered.Where(x => warningIds.Contains(x.WarningId)).ToList(); // 5.查询订单每个轨迹状态的最早轨迹,后续过滤规则使用 var orderIds = filtered.Select(x => x.OrderId).ToHashSet(); var orderTrackStatusTrackDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds); var result = new List(filtered.Count); // 6.按预警类型分组,重新验证规则 foreach (var group in filtered.GroupBy(x => x.WarningType)) { switch (group.Key) { case (int) WarningRuleType.运输时效预警: { var filterTasks = await TransportFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict); if (filterTasks.Count > 0) result.AddRange(filterTasks); } break; case (int) WarningRuleType.回传时效预警: { var filterTasks = await ReturnFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict); if (filterTasks.Count > 0) result.AddRange(filterTasks); break; } case (int) WarningRuleType.完整预警: { var filterTasks = await CompleteFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict); if (filterTasks.Count > 0) result.AddRange(filterTasks); break; } } } return result; } private async Task> TransportFilterAsync( List tasks, Dictionary orderDict, Dictionary> orderTrackStatusTrackDict) { // 1.规则场景存在验证 // 场景id附在任务的辅助id上,只有运输时效预警通知才有这个值!! var sceneIds = tasks .Select(x => (Success: int.TryParse(x.AuxiliaryId, out var id), Id: id)) .Where(x => x.Success) .Select(x => x.Id) .ToHashSet(); var scenes = await _warningRuleRepository.GetTransportSceneAsync(x => sceneIds.Contains(x.Id)); if (scenes.Count == 0) return Empty.List; // 1.1过滤掉不存在场景的通知任务 var validSceneIds = scenes.Select(x => x.Id).ToHashSet(); var filtered = tasks .Where(x => x.AuxiliaryId != null && int.TryParse(x.AuxiliaryId, out var id) && validSceneIds.Contains(id)) .ToList(); // 2.场景验证 var sceneDict = scenes.ToDictionary(x => x.Id.ToString(), TransportWarningSceneBO.From); // 2.1场景验证(产品+国家) var validTasks = new List(filtered.Count); foreach (var task in filtered) { if (sceneDict.TryGetValue(task.AuxiliaryId!, out var scene) && orderDict.TryGetValue(task.OrderId, out var order) && scene.CanHitScene(order)) { validTasks.Add(task); } } if (validTasks.Count == 0) return Empty.List; // 3.场景规则存在验证 var ruleItemIds = validTasks.Select(x => x.WarningRuleItemId).ToHashSet(); var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id)); if (ruleItems.Count == 0) return Empty.List; // 过滤掉不存在的场景规则 var ruleItemDict = ruleItems.ToDictionary(x => x.Id); // 使用 Join 替代 Where + Contains 提升性能 filtered = validTasks .Join(ruleItemDict.Keys, task => task.WarningRuleItemId, id => id, (task, _) => task) .ToList(); // 4.场景规则验证 return filtered .GroupBy(x => x.OrderId) .SelectMany(group => { var orderId = group.Key; if (!orderTrackStatusTrackDict.TryGetValue(orderId, out var statusTrack)) return Enumerable.Empty(); var statusIds = statusTrack.Keys.ToHashSet(); return group.Where(task => ruleItemDict.TryGetValue(task.WarningRuleItemId, out var ruleItem) && ruleItem.CanReHit(statusIds)); }) .ToList(); } private async Task> ReturnFilterAsync( List tasks, Dictionary orderDict, Dictionary> orderTrackStatusTrackDict) { // 1.规则条目存在 var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet(); var ruleItems = await _warningRuleRepository.GetReturnRuleItemsAsync(ruleItemIds); if (ruleItems.Count == 0) return Empty.List; var ruleItemDict = ruleItems.ToDictionary(x => x.Id); // 过滤掉不存在规则的通知 var filtered = tasks.Where(x => ruleItemDict.ContainsKey(x.WarningRuleItemId)).ToList(); // 2.检查明细中为人工介入的状态 var detailsIdSet = filtered.Select(x => x.WarningDetailsId).ToHashSet(); var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO { DetailsIds = detailsIdSet, Status = new HashSet(1) { ReturnWarningStatus.人工介入.GetHashCode() }, }); if (warningDetails.Count == 0) return Empty.List; // 过滤掉不为人工介入的预警通知 var detailsIds = warningDetails.Select(x => x.Id).ToHashSet(); return filtered.Where(x => detailsIds.Contains(x.WarningDetailsId)).ToList(); } private async Task> CompleteFilterAsync( List tasks, Dictionary orderDict, Dictionary> orderTrackStatusTrackDict) { // 1.规则条目存在 var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet(); var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds); if (ruleItems.Count == 0) return Empty.List; var ruleItemDict = ruleItems.ToDictionary(x => x.Id); // 过滤掉不存在规则的通知 var filtered = tasks.Where(x => ruleItemDict.ContainsKey(x.WarningRuleItemId)); // 2.验证规则条目 var result = new List(); foreach (var group in filtered.GroupBy(x => x.OrderId)) { var orderId = group.Key; var orderTrackStatusTrack = orderTrackStatusTrackDict.GetValueOrDefault(orderId); // 一个规则状态也不存在 if (orderTrackStatusTrack == null) continue; var order = orderDict[orderId]; var orderTrackStatusIds = orderTrackStatusTrack.Keys.ToHashSet(); foreach (var task in group) { var ruleItem = ruleItemDict[task.WarningRuleItemId]; if (ruleItem.CanHitRule(order.OrderType, orderTrackStatusIds)) result.Add(task); } } return result; } private List BuildNotifyConfigs(List needNotifyTasks) { var result = new List(); if (needNotifyTasks.Count == 0) return result; // 1.以人的维度对不同类别的预警通知分组聚合 // key: userId value: {key:type,value:notify } var userIdNotifyTaskDict = new Dictionary>>(); foreach (var task in needNotifyTasks) { var notifyUsers = TextJsonHelper.Deserialize>(task.NotifyUsers)!; var notifyUserIds = notifyUsers .Select(x => x.UserId.Trim()) .Where(x => !string.IsNullOrWhiteSpace(x)) .ToHashSet(); foreach (var userId in notifyUserIds) { if (!userIdNotifyTaskDict.TryGetValue(userId, out var notifyTask)) { notifyTask = new Dictionary>(); userIdNotifyTaskDict[userId] = notifyTask; } if (!notifyTask.TryGetValue(task.WarningType, out var notify)) { notify = new List(); notifyTask[task.WarningType] = notify; } notify.Add(task); } } // 2.使用消息模板构建通知参数,稍后排队发送通知 foreach (var (userId, typeNotifyTaskDict) in userIdNotifyTaskDict) { //聚合时出现 用户a //接收回传预警,命中规则1,通知方式(飞书,站内信) ,命中规则2,通知方式(飞书) //接收运输预警,命中规则1,通知方式(飞书,站内信) ,命中规则2,通知方式(飞书) //接收完整预警,命中规则1,通知方式(飞书) ,命中规则2,通知方式(飞书,站内信) //此时聚合后,消息为:总共6条预警消息,其中回传xx条,运输xx条,完整xx条 //同时使用飞书和站内信两种方式通知 result.Add(BuildNotifyParam(userId, typeNotifyTaskDict)); } return result; } private WarningNotifyParam BuildNotifyParam(string userId, Dictionary> typeNotifyTaskDict) { var options = _optionsMonitor.CurrentValue; var hours = options.ScanWarningNotifyTaskHourInterval; var param = new WarningNotifyParam { UserId = userId, Hours = hours }; // tips:获取过去x小时的增量 Func hoursNBefore = x => DateTime.Now - x.CreateTime <= TimeSpan.FromHours(hours); if (typeNotifyTaskDict.TryGetValue(WarningRuleType.运输时效预警.GetHashCode(), out var transportTimelinessNotifyTasks)) { param.TransportLastNHoursCount = transportTimelinessNotifyTasks.Count(hoursNBefore); param.TransportCount = transportTimelinessNotifyTasks.Count; } if (typeNotifyTaskDict.TryGetValue(WarningRuleType.回传时效预警.GetHashCode(), out var returnTimelinessNotifyTasks)) { param.ReturnLastNHoursCount = returnTimelinessNotifyTasks.Count(hoursNBefore); param.ReturnCount = returnTimelinessNotifyTasks.Count; } if (typeNotifyTaskDict.TryGetValue(WarningRuleType.完整预警.GetHashCode(), out var completeWarningNotifyTasks)) { param.CompleteLastNHoursCount = completeWarningNotifyTasks.Count(hoursNBefore); param.CompleteCount = completeWarningNotifyTasks.Count; } return param; } }