using System.Collections.Immutable; using System.Text.Json; using DotNetCore.CAP; using Fake.Application; using Fake.DependencyInjection; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Options; using Newtonsoft.Json; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.OTel; using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Events; using TrackCenter.Application.Services; using TrackCenter.Application.Helpers; using TrackCenter.Application.Options; using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.TrackPush; using TrackCenter.Application.Track2InterfaceRelay; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Aggregates.TrackPushAggregate; using TrackCenter.Domain.Enums; using TrackCenter.Domain.Extensions; using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; namespace TrackCenter.Application.CAP.EventHandlers; public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { private readonly IHttpClientProxy _httpClientProxy; private readonly IOptionsMonitor _options; private readonly IOrderRepository _orderRepository; private readonly TrackPushCacheService _trackPushCacheService; private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly BroadcastingService _broadcastingService; private List _orders = null!; private List _pushConfigs = null!; private Dictionary> _orderTypeSortedConfigsDict = null!; //订单类型,根据sort从小到大排序后的配置字典 public TrackPushEventHandler(IHttpClientProxy httpClientProxy, IOrderRepository orderRepository, TrackPushCacheService trackPushCacheService, ITrackPushRecordEsRepository trackPushRecordEsRepository, IOptionsMonitor options, BroadcastingService broadcastingService, ITrackDetailsEsRepository trackDetailsEsRepository) { _httpClientProxy = httpClientProxy; _orderRepository = orderRepository; _trackPushCacheService = trackPushCacheService; _trackPushRecordEsRepository = trackPushRecordEsRepository; _options = options; _broadcastingService = broadcastingService; _trackDetailsEsRepository = trackDetailsEsRepository; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] public async Task ConsumerAsync(TrackPushEvent @event) { // 子系统+客户id找推送映射配置 var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId); if (pushConfigs == null || pushConfigs.Count == 0) return; if (pushConfigs.First().ServiceProviderCode.IsNullOrWhiteSpace()) { throw new BusinessException(GlobalConst.CapNoRetryErrorCode, $"该客户{@event.CustomId}未关联服务商,无法推送"); } _pushConfigs = pushConfigs; // 构建订单类型和排序后的配置字典 _orderTypeSortedConfigsDict = pushConfigs .GroupBy(x => x.OrderType) .ToDictionary(x => x.Key, y => y.OrderBy(item => item.Sort).ToList()); // 查找所有订单 var orderIds = @event.OrderIdEsIdVersionDict.Keys.ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode, OrderType = x.OrderType } ); if (orders.Count == 0) return; _orders = orders; // 指定了强制推送 if (@event.ForcedPush) { await DoPushAsync(@event); return; } switch (@event.OptionType) { case OptionType.无: return; case OptionType.新增: await DoPushAsync(@event); break; case OptionType.删除: break; case OptionType.修改: await DoUpdateAsync(@event); break; default: throw new ArgumentOutOfRangeException(); } } private async Task DoPushAsync(TrackPushEvent @event) { var items = new List(); if (@event.OperationMan == null) { if (@event.OptionType == OptionType.新增) await BuiltSystemPushItemsAsync(@event, items); else if (@event.OptionType == OptionType.修改) await BuiltManualPushItemsAsync(@event, items); } else { // 手工推送,不会校验已推和顺序问题,因为可能是更新 await BuiltManualPushItemsAsync(@event, items); } if (items.Count == 0) return; using (Track2InterfaceRelaySubSystemContext.Use(new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode))) { var asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( _options.CurrentValue.PushTrackInterfaceCode, serviceProviderCode, items.Cast().ToList(), GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); if (r.Code != GlobalConst.SuccessCode) { TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); throw new BusinessException( message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } private async Task BuiltManualPushItemsAsync(TrackPushEvent @event, List items) { foreach (var order in _orders) { // tips:不同订单类型可能会有不同的顺序 if (!_orderTypeSortedConfigsDict.TryGetValue(order.OrderType, out var sortedConfigs)) { TrackPushFailedPublicTopic($"子系统id:[{@event.SubSystemId}],客户id:[{@event.CustomId}],没有订单类型[{order.OrderType}]的推送映射配置", @event, order); continue; } // tips: 手工/强制推送直接使用配置 var config = sortedConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null) { TrackPushFailedPublicTopic($"针对于订单类型[{order.OrderType}]没有找到轨迹代码[{@event.TrackCode}]对应的配置", @event, order); continue; } // 非强制推送需要 判断本次推送 是否可行 if (config.IsAllPush && !@event.ForcedPush) { config = GetValidPushConfigAsync(sortedConfigs, order.LatestPushTrackCode, @event); } if (config == null) { TrackPushFailedPublicTopic($"推送顺序有误,针对于订单类型[{order.OrderType}],轨迹代码[{@event.TrackCode}]", @event, order); return; } var esIdVersion = @event.OrderIdEsIdVersionDict[order.Id]; var log = new TrackPushRecordIndex { // 订单信息 OrderId = order.Id, OrderNo = order.OrderNo, SubSystemCode = @event.SubSystemCode, // 轨迹信息 TrackEsId = esIdVersion.Key, TrackVersion = esIdVersion.Value, TrackStatusId = @event.SystemPushInfo.TrackStatusId, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackContentId = @event.SystemPushInfo.TrackContentId, TrackCode = @event.TrackCode, TrackContent = @event.SystemPushInfo.TrackContent, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CompareFields = @event.SystemPushInfo.CompareFields, Extra = @event.SystemPushInfo.Extra, // 操作信息 OptionType = @event.OptionType.ToString(), CreateUser = @event.OperationMan, IsToMq = @event.IsToMq, }; var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } } private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List items) { foreach (var order in _orders) { // tips:不同订单类型可能会有不同的顺序 if (!_orderTypeSortedConfigsDict.TryGetValue(order.OrderType, out var sortedConfigs)) { TrackPushFailedPublicTopic($"子系统id:[{@event.SubSystemId}],客户id:[{@event.CustomId}],没有订单类型[{order.OrderType}]的推送映射配置", @event, order); continue; } // tips: 手工/强制推送直接使用配置 var config = sortedConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null) { TrackPushFailedPublicTopic($"子系统id:[{@event.SubSystemId}],客户id:[{@event.CustomId}],没有订单类型[{order.OrderType}]的推送映射配置," + $"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order); return; } // 系统不推异常单,只能人工操作 if (order.QuestionType == (int) TrackProblemType.Error) { TrackPushFailedPublicTopic("系统不推异常单,只能人工操作", @event, order); continue; } // 推送新增,需要根据上一次的推送 推断 本次推送 是否可行,为了约束顺序 // 如果当前推送code与订单上的最后的code不一致,则考虑推送映射顺序 if (!@event.ForcedPush && !string.Equals(order.LatestPushTrackCode, @event.TrackCode)) { config = GetValidPushConfigAsync(sortedConfigs, order.LatestPushTrackCode, @event); if (config == null) { TrackPushFailedPublicTopic("推送顺序异常,找不到有效推送配置", @event, order); continue; } } // 是否自动推送 if (!config.IsAutoPush) { TrackPushFailedPublicTopic("没有设置为自动推送,如果要推送请手动推送", @event, order); continue; } var esIdVersion = @event.OrderIdEsIdVersionDict[order.Id]; var log = new TrackPushRecordIndex { //订单信息 OrderId = order.Id, OrderNo = order.OrderNo, SubSystemCode = @event.SubSystemCode, //轨迹信息 TrackEsId = esIdVersion.Key, TrackVersion = esIdVersion.Value, TrackStatusId = @event.SystemPushInfo.TrackStatusId, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackContentId = @event.SystemPushInfo.TrackContentId, TrackCode = @event.TrackCode, TrackContent = @event.SystemPushInfo.TrackContent, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CompareFields = @event.SystemPushInfo.CompareFields, Extra = @event.SystemPushInfo.Extra, //操作信息 OptionType = @event.OptionType.ToString(), CreateUser = GlobalConst.SystemUser, IsToMq = @event.IsToMq, }; var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } } private async Task DoUpdateAsync(TrackPushEvent @event) { var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null || !config.IsCanUpdate) return; var items = new List(); foreach (var order in _orders) { // tips:不同订单类型可能会有不同的顺序 if (!_orderTypeSortedConfigsDict.TryGetValue(order.OrderType, out var sortedConfigs)) { TrackPushFailedPublicTopic($"子系统id:[{@event.SubSystemId}],客户id:[{@event.CustomId}],没有订单类型[{order.OrderType}]的推送映射配置", @event, order); continue; } // tips:更新轨迹要求只能更新已经推过的历史轨迹 config = sortedConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null) { TrackPushFailedPublicTopic($"针对于订单类型[{order.OrderType}]没有找到轨迹代码[{@event.TrackCode}]对应的配置", @event, order); continue; } var esIdVersion = @event.OrderIdEsIdVersionDict[order.Id]; var log = new TrackPushRecordIndex { // 订单信息 OrderId = order.Id, OrderNo = order.OrderNo, SubSystemCode = @event.SubSystemCode, // 轨迹信息 TrackEsId = esIdVersion.Key, TrackVersion = esIdVersion.Value, TrackStatusId = @event.SystemPushInfo.TrackStatusId, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackContentId = @event.SystemPushInfo.TrackContentId, TrackCode = @event.TrackCode, TrackContent = @event.SystemPushInfo.TrackContent, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CompareFields = @event.SystemPushInfo.CompareFields, Extra = @event.SystemPushInfo.Extra, //操作信息 OptionType = @event.OptionType.ToString(), CreateUser = GlobalConst.SystemUser, IsToMq = @event.IsToMq, }; var item = BuiltUpdateItem(config, log); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } if (items.Count == 0) return; using (Track2InterfaceRelaySubSystemContext.Use(new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode))) { var asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); var serviceProviderCode = _pushConfigs.First().ServiceProviderCode; var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( _options.CurrentValue.TrackUpdateInterfaceCode, serviceProviderCode, items.Cast().ToList(), GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); if (r.Code != GlobalConst.SuccessCode) { TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast().ToList()); throw new BusinessException( message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}"); } } } // 构建更新推送请求体 private static TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) { var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone); var trackDateStr = DateTimeFormat(trackDatetime, config.TrackTimeZone ?? log.TrackTimeZone, config); var body = new TrackUpdateInput { OrderNo = log.OrderNo }; var dict = new Dictionary() { // 冗余轨迹id和版本号 { GlobalConst.TrackPushTrackIdDictKey, log.TrackEsId }, { GlobalConst.TrackPushTrackVersionDictKey, log.TrackVersion }, // 推真实的发生地和发生时间(时间是计算后的) { "location", log.TrackPlace }, { "date", trackDateStr }, { "timeZone", config.TrackTimeZone ?? log.TrackTimeZone }, { "description", config.TrackEnContentValue }, { "descriptionCn", config.TrackContentValue }, { "code", config.TrackContentCodeValue }, { "status", config.TrackStatusValue }, { "extra", TextJsonHelper.Deserialize(log.Extra) } }; body.Data.Add(dict); log.PushContent = body.ToJson(); return body; } // 构建新增推送请求体 private async Task BuiltPushItem(TrackPushContentConfigDto config, TrackPushRecordIndex log, string? compareFields) { var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone); var trackDateStr = DateTimeFormat(trackDatetime, config.TrackTimeZone ?? log.TrackTimeZone, config); var body = new TrackPushInput { OrderNo = log.OrderNo }; var cur = new Dictionary { // 冗余轨迹id和版本号 { GlobalConst.TrackPushTrackIdDictKey, log.TrackEsId }, { GlobalConst.TrackPushTrackVersionDictKey, log.TrackVersion }, // 推真实的发生地和发生时间(时间是计算后的) { "date", trackDateStr }, { "timeZone", config.TrackTimeZone ?? log.TrackTimeZone }, { "location", log.TrackPlace }, { "description", config.TrackEnContentValue }, { "descriptionCn", config.TrackContentValue }, { "code", config.TrackContentCodeValue }, { "status", config.TrackStatusValue }, { "extra", TextJsonHelper.Deserialize(log.Extra) } }; body.Data.Add(cur); if (config.IsAllPush) { var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true); if (history != null) { body = TextJsonHelper.Deserialize(history.PushContent); //在推送历史中,选出当前轨迹明细一致的轨迹 var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId); //确定轨迹标识符 var trackIdentifies = tracks.Select(x => { //寻找映射配置(转换后的code) var matchConfig = _orderTypeSortedConfigsDict.TryGetValue(x.OrderType, out var sortedConfigs) ? sortedConfigs.FirstOrDefault(y => y.TrackCode == x.TrackCode) : null; if (matchConfig == null) return null; var code = matchConfig.TrackContentCodeValue; var timeStr = DateTimeFormat(x.TrackTime, config.TrackTimeZone ?? log.TrackTimeZone, config); return string.IsNullOrWhiteSpace(compareFields) ? $"{code}{timeStr}" : $"{code}{TextJsonHelper.ExtractValue(x.Extra, compareFields)}"; }) .Where(x => x != null) .ToImmutableHashSet(); //在历史中选取在订单当前轨迹明细中的记录 body!.Data = body.Data.Where(x => { var code = ((JsonElement) x["code"]!).GetString(); var timeStr = ((JsonElement) x["date"]!).GetString(); if (string.IsNullOrWhiteSpace(compareFields)) return trackIdentifies.Contains($"{code}{timeStr}"); var extra = x.GetValueOrDefault("extra"); if (extra == null) return false; if (((JsonElement) extra).TryGetProperty(compareFields, out var comparerValueJsonElement)) return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}"); return trackIdentifies.Contains($"{code}{null}"); }).ToList(); if (config.IsAllPushWithLatest) { //同code同比较字段保证唯一 var old = body.Data .Where(x => { //反序列化出的JsonElement和字典字符串类型做比较 var a = ((JsonElement) x["code"]!).GetString(); var b = cur["code"]! .ToString(); //tips: 需要要将object类型的字符串转为string类型,否则使用的是引用比较,反序列化出的string一定和本地string引用不同 return a == b; }) .WhereIf(!compareFields.IsNullOrWhiteSpace(), x => { if (x["extra"] == null || log.Extra == null) return false; return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize(log.Extra), compareFields!); }) .FirstOrDefault(); if (old != null) { body.Data.Remove(old); } } // 添加当前要推送的轨迹,到末尾 body.Data.Add(cur); } } log.PushContent = TextJsonHelper.Serialize(body)!; return body; } // 尝试根据推送事件找推送映射配置 private static TrackPushContentConfigDto? GetValidPushConfigAsync(List cache, string? latestPushTrackCode, TrackPushEvent @event) { var trackCode = @event.TrackCode; var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); // 指定了强制推送并且当前轨迹和最后一个推送成功的轨迹相同,允许重复推送 if (@event.ForcedPush && trackCode == latestPushTrackCode) return cache[idx]; if (idx + 1 >= cache.Count) return null; // 推完了 // 命中下一个顺序的配置,否则尝试寻找下一个顺序配置的交换配置 var matched = cache[idx + 1].TrackCode == trackCode ? cache[idx + 1] : cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); // 存在配置并且,并且当前code命中配置 return (matched != null && matched.TrackCode == trackCode) ? matched : null; } // 转换推送时间格式 private static string DateTimeFormat(DateTime dateTime, double timezone, TrackPushContentConfigDto config) { if (config.DateTimeFormatType == (int) DateTimeFormatEnum.Timestamp) return DateTimeHelper.ConvertToUnixTimestamp(dateTime).ToString(); if (config.DateTimeFormatType == (int) DateTimeFormatEnum.DateTimeAndTimezone) { // 只有时间 if (!string.IsNullOrWhiteSpace(config.DateTimeFormat) && string.IsNullOrWhiteSpace(config.TimezoneFormat)) return DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); // 时间和时区 if (!string.IsNullOrWhiteSpace(config.DateTimeFormat) && !string.IsNullOrWhiteSpace(config.TimezoneFormat)) { var zoneStr = DateTimeHelper.TimezoneFormat(timezone, config.TimezoneFormat); var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); return $"{datetimeStr} {zoneStr}"; } } // 默认为 时间+时区 return dateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffK"); } //轨迹推送失败(被拦截) private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order) { using var activity = ActivityHelper.CreateSpanActivity($"TrackPushFailedPublicTopic-{order.OrderNo}"); activity.SetTag("failedMessage", failedMessage); _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { SubSystemCode = @event.SubSystemCode, IsManualPush = @event.OperationMan != null, PushTime = DateTime.Now, //还没请求接口通 RequestId = null, Message = failedMessage, OperationMan = @event.OperationMan, OptionType = @event.OptionType.ToString(), OrderNo = order.OrderNo, IsSuccess = false, //还没到构建推送内容那一步就被拦截 PushContent = string.Empty, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackContent = @event.SystemPushInfo.TrackStatus, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, TrackTime = @event.SystemPushInfo.TrackTime, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackCode = @event.TrackCode, Extra = @event.SystemPushInfo.Extra, }); } //轨迹推送失败(请求接口通) private void TrackPushFailedPublicTopic( string failedMessage, TrackPushEvent @event, Guid requestId, List pushOrUpdateInputs ) { using var activity = ActivityHelper.CreateSpanActivity("TrackPushFailedPublicTopic-All"); activity.SetTag("failedMessage", failedMessage.ToJson()); activity.SetTag("@event", @event.ToJson()); foreach (var pushInput in pushOrUpdateInputs) { _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto { SubSystemCode = @event.SubSystemCode, IsManualPush = @event.OperationMan != null, PushTime = DateTime.Now, OperationMan = @event.OperationMan, OrderNo = pushInput.OrderNo, RequestId = requestId, Message = failedMessage, OptionType = @event.OptionType.ToString(), IsSuccess = false, PushContent = TextJsonHelper.Serialize(pushInput), TrackStatus = @event.SystemPushInfo.TrackStatus, TrackContent = @event.SystemPushInfo.TrackStatus, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, TrackTime = @event.SystemPushInfo.TrackTime, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackCode = @event.TrackCode, Extra = @event.SystemPushInfo.Extra, }); } } }