using System.Diagnostics; using System.Text.Json; using DotNetCore.CAP; using Microsoft.Extensions.Logging; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; using OrderIdentifier = SJZY.Track.Application.Contract.Events.OrderIdentifier; namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { private readonly ICapPublisher _capPublisher; private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private readonly ILogger _logger; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; public TrackUpsertEventHandler( IOrderRepository orderRepository, ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _logger = logger; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] public async Task ConsumerAsync(TrackUpsertEvent @event) { if (!CanDelete(@event)) return; var canNextOrderIds = new List(); var content = await HandleTrackAsync(@event, canNextOrderIds); // 处理广播事件 if (content != null && (content.IsToMq || @event.IsToMq)) _broadcastingService.TrackChangePublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode); if (canNextOrderIds.Count == 0) return; // 根据订单关联关系异步向下刷(异步) await DownByOrderLink(@event, canNextOrderIds); } private async Task DownByOrderLink(TrackUpsertEvent @event, List canNextOrderIds) { // 根据订单关联关系异步向下刷(异步) var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray()); foreach (var order in @event.OrderIdentifiers) { // 理论上小包在这不会分组 foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId) .GroupBy(x => new { x.SubSystemId, x.CustomId // 这里按客户分组是为了后续的推送方便 })) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.AddRange(canNextOrderIds); nextEvent.SubSystemId = group.Key.SubSystemId; nextEvent.CustomId = group.Key.CustomId; nextEvent.TrackRefreshData = new TrackRefreshStruct { ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash, }; // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) { var mapCode = (await _provideQueries .MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) .FirstOrDefault(); if (mapCode == default) { // 跨子系统没做映射的轨迹不处理 continue; } nextEvent.TrackCode = mapCode; } var identifiers = group.Select(x => new OrderIdentifier { OrderId = x.OrderId, OrderNo = x.OrderNo, OrderType = x.OrderType, }).ToList(); var idx = 0; while (idx < identifiers.Count) { var subEvent = nextEvent.DeepCopy(); var cnt = idx + GlobalConst.ChunkSize > identifiers.Count ? identifiers.Count - idx : GlobalConst.ChunkSize; subEvent.OrderIdentifiers = identifiers.GetRange(idx, cnt).ToArray(); idx += GlobalConst.ChunkSize; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); } } } } private bool CanDelete(TrackUpsertEvent @event) { // 系统创建的轨迹不允许修改 if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false }) { LogHelper.Warn($"系统创建的轨迹不允许被修改:{@event}", true); return false; } var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var dup = @event.Visited.Intersect(orderIds).ToArray(); if (dup.Any()) { LogHelper.Warn($"{(@event.IsAdd ? "添加" : "修改")}轨迹时,发现循环依赖:{dup.JoinAsString(",")}", true); return false; } return true; } private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) { var trackContent = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode); if (trackContent == null) { // 找不到轨迹内容,直接结束 LogHelper.Warn($"找不到轨迹内容:{@event}", true); return default; } //#7001:未来潜在需求,当轨迹内容匹配预设,记录一个标记,保证下刷过程都是使用预设内容 //虽然在引发事件之前做了判断,引发之后并不一定立即处理,在此过程中轨迹内容仍然可能被修改,所以要二次进行判断。 //通过来源指示是否为手动操作 var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial; //轨迹内容标识不允许手动创建的 if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false) return trackContent; //轨迹内容标识不允许手动更新的 if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false) return trackContent; var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent); if (canNextOrderIds.Count == 0) return trackContent; if (insertTracks.Any()) { // 批量add to es await _trackDetailsEsRepository.AddManyAsync(insertTracks); // 添加轨迹触发订单拉取状态更新 await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), @event.TrackCode); // 添加轨迹触发预警计算(异步) await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent { SubSystemId = @event.SubSystemId, CustomId = @event.CustomId, OrderIds = insertTracks.Select(x => x.OrderId).ToArray(), TrackStatusId = trackContent.TrackStatusId, TrackTime = @event.TrackTime, }); } // 触发订单主表更新事件(异步) await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { OrderIds = canNextOrderIds.ToArray(), IsCompleted = trackContent.IsCompleted, TrackContent = trackContent.TrackContent, TrackEnContent = trackContent.TrackEnContent, TrackPlace = @event.TrackPlace, TrackTime = @event.NewTrackTime ?? @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, DestinationTimeZone = @event.DestinationTimeZone, EstimatedDeliveryDate = @event.EstimatedDeliveryDate }); // 触发轨迹推送事件(异步) var trackPushEvent = new TrackPushEvent { OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, SubSystemId = @event.SubSystemId, SubSystemCode = trackContent.SubSystemCode, CustomId = @event.CustomId, OrderIds = canNextOrderIds.ToArray(), TrackCode = @event.TrackCode, SystemPushInfo = new SystemPushStruct { TrackStatus = trackContent.TrackStatus, TrackContent = trackContent.TrackContent, TrackPlace = @event.TrackPlace, TrackTime = @event.NewTrackTime ?? @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, CompareFields = @event.CompareFields, Extra = @event.Extra }, ForcedPush = @event.ForcedPush, IsToMq = (trackContent.IsToMq || @event.IsToMq) }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); return trackContent; } private async Task> UpsertTrackAsync(TrackUpsertEvent @event, List canNextOrderIds, TrackContentVO trackContent) { var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); // important:轨迹code相同,时间、发生地不同的,可能存在多条!! var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); var insertTracks = new List(); var updateTracks = new List(); foreach (var order in @event.OrderIdentifiers) { // fast path: by update with es id case if (!@event.IsAdd && !string.IsNullOrWhiteSpace(@event.EsId)) { var trackById = tracks.FirstOrDefault(x => x.Id == @event.EsId); if (trackById != null) { updateTracks.Add(BuiltUpdateTrack(@event, trackById, order, trackContent)); } continue; } var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList(); if (@event.CompareFields.IsNullOrEmpty()) { existTracks = existTracks.Where(x => x.TrackTime == @event.TrackTime).ToList(); } else { Debug.Assert(@event.CompareFields != null, "@event.CompareFields != null"); var extraDict = GetExtraDictForTextJson(@event.Extra); if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) { throw new BusinessException(message: $"请检查拓展字段,为什么找不到比较字段:{@event.CompareFields}"); } existTracks = existTracks.Where(x => { var oldExtraDict = GetExtraDictForTextJson(x.Extra); // tips:注意兼容性,历史轨迹的字段可能为空 if (oldExtraDict == null|| !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue)) { return false; } return EqJsonElementObject(oldValue, compareValue); }).ToList(); } var track = existTracks.FirstOrDefault(); if (!@event.IsAdd) { if (existTracks.Count > 1) { // todo: 所有的event handle 多写log,取缔注释,特别是continue,return,这种跳出的地方,方便排查问题 _logger.LogInformation("订单存在多条轨迹不允许修改:{@event}", @event); continue; } // todo:例如这里 if (track == null) continue; // 不存在不让改 updateTracks.Add(BuiltUpdateTrack(@event, track, order, trackContent)); } else { // 不允许重复轨迹,跳过 if (!trackContent.IsRepeatable && existTracks.Count > 0) continue; insertTracks.Add(BuiltNewTrack(@event, order, trackContent)); } canNextOrderIds.Add(order.OrderId); } if (updateTracks.Any()) { //批量差异化更新轨迹 await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks); } return insertTracks; } private TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order, TrackContentVO trackContent) { //外部轨迹,轨迹发生地,轨迹发生时间,轨迹发生时区,不允许更改 if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide) { track.TrackPlace = @event.TrackPlace; track.TrackTime = @event.NewTrackTime ?? @event.TrackTime; track.TrackTimeZone = @event.TrackTimeZone; } track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser; track.UpdateTime = DateTime.Now; //允许二次修改,在修改了轨迹内容后进行修改轨迹操作,如果输入的轨迹内容则以输入为准 //跨子系统? track.TrackContent = @event.Content ?? trackContent.TrackContent; track.TrackEnContent = @event.EnStatus ?? trackContent.TrackEnContent; //S06新增 track.IsVisibility = @event.IsVisibility; track.Remarks = @event.Remarks ?? GlobalConst.EmptyString; track.Files = @event.Files?.Select(x => new FileItem { OssKey = x.OssKey, FileName = x.FileName }).ToList(); track.TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString; track.TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId; track.TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString; track.TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId; track.TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString; track.TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId; track.TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString; track.DestinationTimeZone = @event.DestinationTimeZone; track.EstimatedDeliveryDate = @event.EstimatedDeliveryDate; track.Extra = @event.Extra; // todo:如果后面写es异常了,这里日志不就不准了,这里应该和写轨迹一样,做成批量的,放到写完轨迹只会写日志 // 写入操作日志 _ = _orderOperationRecordEsRepository.AddAsync(new OrderOperationRecordIndex { OrderId = order.OrderId, OrderNo = order.OrderNo, OptionType = OrderOperationType.修改轨迹.ToString(), Content = track.ToJson(), CreateUserId = @event.OperationManId, CreateUserName = @event.OperationMan ?? "", CreateTime = DateTime.Now }); return track; } private TrackDetailsIndex BuiltNewTrack(TrackUpsertEvent @event, OrderIdentifier order, TrackContentVO trackContent) { var track = new TrackDetailsIndex { SubSystemId = @event.SubSystemId, OrderId = order.OrderId, OrderNo = order.OrderNo, OrderType = order.OrderType, TrackCode = @event.TrackCode, TrackPlace = @event.TrackPlace, TrackTime = @event.NewTrackTime ?? @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString, CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now, //来源轨迹内容,第一次新增的时候使用轨迹内容表提供的 TrackStatusId = trackContent.TrackStatusId, TrackStatus = trackContent.TrackStatus, TrackStatusCode = trackContent.TrackStatusCode, //跨子系统? TrackContentId = trackContent.TrackContentId, TrackContent = @event.Content ?? trackContent.TrackContent, TrackEnContent = @event.EnStatus ?? trackContent.TrackEnContent, //S06新增轨迹明细字段 IsVisibility = @event.IsVisibility, Source = @event.Source, Remarks = @event.Remarks ?? GlobalConst.EmptyString, Files = @event.Files?.Select(x => new FileItem { OssKey = x.OssKey, FileName = x.FileName }).ToList(), TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString, TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId, TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString, TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId, TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString, TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId, TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString, DestinationTimeZone = @event.DestinationTimeZone, EstimatedDeliveryDate = @event.EstimatedDeliveryDate, Extra = @event.Extra }; track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); // todo:如果后面写es异常了,这里日志不就不准了,这里应该和写轨迹一样,做成批量的,放到写完轨迹只会写日志 // 写入操作日志 _ = _orderOperationRecordEsRepository.AddAsync(new OrderOperationRecordIndex { OrderId = order.OrderId, OrderNo = order.OrderNo, OptionType = OrderOperationType.创建轨迹.ToString(), Content = track.ToJson(), CreateUserId = @event.OperationManId, CreateUserName = @event.OperationMan ?? "", CreateTime = DateTime.Now }); return track; } private Dictionary? GetExtraDictForTextJson(string? extraStr) { return string.IsNullOrWhiteSpace(extraStr) ? null : System.Text.Json.JsonSerializer.Deserialize>(extraStr); } private bool EqJsonElementObject(object? a, object? b) { if (a is JsonElement aJsonElement && b is JsonElement bJsonElement) { return aJsonElement.ToString() == bJsonElement.ToString(); } return false; } }