using System.Collections.Immutable; using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Kafka; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Extensions; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe { private readonly ICapPublisher _capPublisher; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private readonly WarningSignalPublisher _warningSignalPublisher; public TrackDeleteEventHandler( IProvideQueries provideQueries, IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, WarningSignalPublisher warningSignalPublisher ) { _provideQueries = provideQueries; _orderRepository = orderRepository; _capPublisher = capPublisher; _trackDetailsEsRepository = trackDetailsEsRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] public async Task ConsumerAsync(TrackDeleteEvent @event) { if (!CanDelete(@event)) return; var trackList = await DeleteTrackAsync(@event); //当轨迹删除,根据轨迹标记/状态上IsToMq选择进行广播 if (trackList.Count > 0) { _ = PublicTrackTopicAsync(@event, trackList); //引发预警计算 WarningCalc_DeletedTrack(trackList); } // 根据订单关联关系异步向下刷(异步) await DownByOrderLink(@event); } private async Task DownByOrderLink(TrackDeleteEvent @event) { var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderIds); foreach (var orderId in @event.OrderIds) { foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == orderId).GroupBy(x => x.SubSystemId)) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.AddRange(@event.OrderIds); nextEvent.TrackIdentifiers = @event.TrackIdentifiers; nextEvent.SubSystemId = group.Key; // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) { var codeMaps = await _provideQueries.GetTrackCodeMapsAsync( @event.SubSystemId, nextEvent.SubSystemId, @event.TrackIdentifiers.Select(x => x.TrackCode).ToArray() ); if (codeMaps.Count == 0) { LogHelper.Warn($"订单跨子系统,未配置内容映射,请联系管理员:{@event}", true); continue; } //映射轨迹code foreach (var trackIdentity in nextEvent.TrackIdentifiers) { var match = codeMaps.FirstOrDefault(x => x.MapTrackContentCode == trackIdentity.TrackCode); if (match == null) continue; trackIdentity.TrackCode = match.TrackContentCode; } } var identifiers = group.Select(x => x.OrderId).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.OrderIds = identifiers.GetRange(idx, cnt).ToArray(); idx += GlobalConst.ChunkSize; await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, subEvent); } } } } private bool CanDelete(TrackDeleteEvent @event) { if (@event.OrderIds == null) return true; // 订单循环关联检测 var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); if (dup.Any()) { LogHelper.Warn($"删除轨迹时,发现循环依赖:{dup.JoinAsString(",")}", true); return false; } return true; } private async Task> DeleteTrackAsync(TrackDeleteEvent @event) { List orderIds; List tracks; List deleteTracks; // 优先根据EsId查询轨迹 if (@event.EsIds.Length > 0) //根据明确的esid进行删除 { deleteTracks = await _trackDetailsEsRepository.GetDataByIds(@event.EsIds); orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); @event.OrderIds = orderIds.ToArray(); @event.TrackIdentifiers = deleteTracks.Select(x => new TrackIdentifierDto { TrackCode = x.TrackCode, TrackTime = x.TrackTime }).ToList(); // 外部api获取的轨迹不允许删除 if (deleteTracks.Any(m => m.Source == (int) TrackSourceEnum.ExternalServiceProvide)) return deleteTracks; //查出订单下的所有轨迹,以供后面回退使用 tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); // 删除轨迹批量 await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); //清理已经删除过的esid @event.EsIds = Array.Empty(); } else //原有逻辑,根据轨迹code和订单进行删除 { //根据订单查询出所有的轨迹,查所有是因为主单上需要进行回退更新 tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); //比较字段为空则使用时间来作为额外的标识key if (string.IsNullOrWhiteSpace(@event.CompareFields)) { //构建标识符集合 var ids = @event.TrackIdentifiers.Select(x => $"{x.TrackCode}{x.TrackTime:G}").ToHashSet(); //根据标识符选出要删除的轨迹明细 deleteTracks = tracks.Where(x => ids.Contains($"{x.TrackCode}{x.TrackTime:G}")).ToList(); } //否则则使用比较字段,使用扩展字段中的内容进行比较 else { var codes = @event.TrackIdentifiers.Select(x => x.TrackCode).ToImmutableHashSet(); deleteTracks = tracks .Where(x => { if (!codes.Contains(x.TrackCode)) return false; if (string.IsNullOrWhiteSpace(x.Extra)) { throw new BusinessException(message: $"当前轨迹不存在扩展信息,无法利用扩展字段:{@event.CompareFields}"); } var extraDict = TextJsonHelper.DeserializeToDictionary(x.Extra); if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) { throw new BusinessException(message: $"请检查拓展字段,为什么找不到比较字段:{@event.CompareFields}"); } return TextJsonHelper.EqJsonElement(compareValue, @event.CompareValue); }) .ToList(); } if (deleteTracks.Count == 0) return deleteTracks; // 删除轨迹批量 await _trackDetailsEsRepository.DeleteTrackByIdsAsync(deleteTracks.Select(x => x.Id).ToHashSet()); //从被删除轨迹寻找订单id,稍后会更新订单主表 orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); } //排除掉已经删除的轨迹 var deleteTrackIds = deleteTracks.Select(m => m.Id).ToList(); tracks = tracks.Where(x => !deleteTrackIds.Contains(x.Id)).ToList(); var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, LatestTrackStatus = x.LatestTrackStatus, LatestTrackContent = x.LatestTrackContent, LatestTrackEnContent = x.LatestTrackEnContent, LatestTrackPlace = x.LatestTrackPlace, LatestTrackTime = x.LatestTrackTime, LatestTrackTimeZone = x.LatestTrackTimeZone, EstimatedDeliveryDate = x.EstimatedDeliveryDate, DestinationTimeZone = x.DestinationTimeZone }); if (orders.Count == 0) return tracks; foreach (var order in orders) { //找到最新的轨迹数据 var track = tracks?.OrderByDescending(x => x.ZeroTrackTime) .FirstOrDefault(x => x.OrderId == order.Id); // 删除全部轨迹导致的重置行为 if (track == null) { if (order.LatestTrackTime == null) continue; // tips:订单是支持从已完成态到未完成态的 order.ResetTrackInfo(); } else { if (order.LatestTrackContent == track.TrackContent) continue; // 删除引发的回退更新 order.LatestTrackStatus = track.TrackStatus; order.LatestTrackContent = track.TrackContent; order.LatestTrackEnContent = track.TrackEnContent; order.LatestTrackPlace = track.TrackPlace; order.LatestTrackTime = track.TrackTime; order.LatestTrackTimeZone = track.TrackTimeZone; order.EstimatedDeliveryDate = track.EstimatedDeliveryDate; order.DestinationTimeZone = track.DestinationTimeZone; } } //更新订单上冗余轨迹明细的字段 await _orderRepository.UpdateRangeAsync(orders, x => new { x.LatestTrackStatus, x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime, x.LatestTrackTimeZone, x.EstimatedDeliveryDate, x.DestinationTimeZone }); //批量写入订单操作日志 _ = _orderOperationRecordEsRepository.AddManyAsync(orders.Select(order => new OrderOperationRecordIndex { OrderId = order.Id, OrderNo = order.OrderNo, OptionType = OrderOperationType.删除轨迹.ToString(), Content = deleteTracks.Where(m => m.OrderId == order.Id).ToList().ToJson(), CreateUserId = @event.OperationManId, CreateUserName = @event.OperationMan ?? "", CreateTime = DateTime.Now }).ToList()); return deleteTracks; } private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List trackList) { var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray(); var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray(); //从轨迹内容中查询出标记是否需要mq广播 var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes); var toMqDict = contentList .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq); //选出需要广播的轨迹,并按code和订单号分组 var trackGroup = trackList .Where(x => toMqDict[x.SubSystemId + x.TrackCode]) .GroupBy(m => new { m.TrackCode, m.OrderNo }) .ToList(); if (trackGroup.Count == 0) return; //增加延时确保写盘成功 await Task.Delay(1000); foreach (var dataItem in trackGroup) { var item = dataItem.First(); var content = contentList.First(m => m.TrackContentCode == item.TrackCode); //将每个订单的每个轨迹的删除变更发布 foreach (var track in dataItem) { _broadcastingService.TrackChangePublicTopic(new TrackChangeEto { EstimatedDeliveryDate = track.EstimatedDeliveryDate, DestinationTimeZone = track.DestinationTimeZone, TrackPlacePostcode = track.TrackPlacePostcode, TrackPlaceNation = track.TrackPlaceNation, TrackPlaceNationId = track.TrackPlaceNationId, TrackPlaceProvince = track.TrackPlaceProvince, TrackPlaceProvinceId = track.TrackPlaceProvinceId, TrackPlaceCity = track.TrackPlaceCity, TrackPlaceCityId = track.TrackPlaceCityId, Files = track.Files?.Select(x => new FileItemDto() { FileName = x.FileName, OssKey = x.OssKey }).ToList(), Remarks = track.Remarks, Source = track.Source, IsVisibility = track.IsVisibility, IsExternal = track.IsExternal, TrackTimeZone = track.TrackTimeZone, TrackTime = track.TrackTime, TrackPlace = track.TrackPlace, TrackStatusCode = track.TrackStatusCode, EnStatus = track.TrackEnContent, Content = track.TrackContent, TrackCode = track.TrackCode, OperationMan = @event.OperationMan, OrderNo = track.OrderNo, OrderIdentifiers = new[] { new OrderIdentifier { OrderId = track.OrderId, OrderNo = track.OrderNo, OrderType = track.OrderType } }, SubSystemId = @event.SubSystemId, Extra = track.Extra }, content.SubSystemCode); } } } private void WarningCalc_DeletedTrack(List delTracks) { _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); } }