using System.Text; using BSI.Stack.Rabbit; using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; 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 RabbitClientBus _rabbitClientBus; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; public TrackUpsertEventHandler( IOrderRepository orderRepository, ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository, RabbitClientBus rabbitClientBus, IOrderOperationRecordEsRepository orderOperationRecordEsRepository) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; _rabbitClientBus = rabbitClientBus; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; } [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); // 处理广播事件 PublicTrackTopic(@event, content); if (canNextOrderIds.Count == 0) return; // 根据订单关联关系异步向下刷(异步) await DownByOrderLink(@event, canNextOrderIds); } /// /// 发送轨迹广播 /// /// /// private void PublicTrackTopic(TrackUpsertEvent eto, TrackContentVO? content) { // var trackStatus = await _trackStatusRepository.GetFirstAsync(m => m.Code == content.TrackStatusCode); if (content == null || !content.IsToMq) return; var data = new TrackChangeEto() { EstimatedDeliveryDate = eto.EstimatedDeliveryDate, TrackPlacePostcode = eto.TrackPlacePostcode, TrackPlaceNation = eto.TrackPlaceNation, TrackPlaceNationId = eto.TrackPlaceNationId, TrackPlaceProvince = eto.TrackPlaceProvince, TrackPlaceProvinceId = eto.TrackPlaceProvinceId, TrackPlaceCity = eto.TrackPlaceCity, TrackPlaceCityId = eto.TrackPlaceCityId, Files = eto.Files, Remarks = eto.Remarks, Source = eto.Source, IsVisibility = eto.IsVisibility, TrackTimeZone = eto.TrackTimeZone, TrackTime = eto.TrackTime, TrackPlace = eto.TrackPlace, TrackStatusCode = content.TrackStatusCode, EnStatus = eto.EnStatus, Content = eto.Content, TrackCode = eto.TrackCode, OperationMan = eto.OperationMan, OrderIdentifiers = eto.OrderIdentifiers, CustomId = eto.CustomId, SubSystemId = eto.SubSystemId, Extra = eto.Extra }; if (eto.IsAdd) data.Option = 1; else data.Option = 3; _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); } 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, Extra = @event.Extra } }; 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, false); var insertTracks = new List(); var updateTracks = new List(); foreach (var order in @event.OrderIdentifiers) { TrackDetailsIndex? track; if (!@event.IsAdd) { //同一订单同轨迹code同一轨迹时间如果有多条轨迹明细则不允许修改 var existTracks = tracks.Where(x => x.OrderId == order.OrderId && x.TrackTime == @event.TrackTime).ToArray(); if (existTracks.Length > 1) { // 修改轨迹时,订单存在多条轨迹不允许修改 continue; } track = existTracks.FirstOrDefault(); if (track == null) continue; // 不存在不让改 } else { // 【批量新增】所有可能遇到的场景 // 标识:允许重复,不允许重复,零条,一条,多条; // 场景A:允许重复,零条 => 新增 // 场景B:允许重复,一条 => 新增(约束受限新增,否则跳过新增) // 场景C:允许重复,多条 => 新增(约束受限新增,否则跳过新增) // 场景D:不允许重复,零条 => 新增 // 场景E:不允许重复,一条 => 跳过 // 场景F:不允许重复,(理想不会出现多条) => 跳过 var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToArray(); track = existTracks.FirstOrDefault(); //覆盖场景:B,C if (trackContent.IsRepeatable && existTracks.Length >= 1) { if (existTracks.Any(x => x.TrackPlace == @event.TrackPlace && x.TrackTime == @event.TrackTime)) { //允许重复,出现【轨迹发生地】,【轨迹发生时间】的都重复,则不允许新增 continue; } // 设置null后面执行新增逻辑 track = null; } //不允许重复,并且记录大于等于一条,跳过 //覆盖场景:E,F if (!trackContent.IsRepeatable && existTracks.Length >= 1) { continue; } //其余情况:A,D都为新增操作,无需检查数量,track为null执行新增操作 } if (track == default) { 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)); insertTracks.Add(track); // 写入操作日志 _ = _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 }); } else { //外部轨迹,轨迹发生地,轨迹发生时间,轨迹发生时区,不允许更改 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; updateTracks.Add(track); // 写入操作日志 _ = _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 }); } canNextOrderIds.Add(order.OrderId); } if (updateTracks.Any()) { //批量差异化更新轨迹 await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks); } return insertTracks; } }