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.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackContentAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackStatusAggregate; using SJZY.Track.Domain.Enums; 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 IOrderRepository _orderRepository; private readonly ITrackStatusRepository _trackStatusRepository; private readonly ITrackContentRepository _trackContentRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly RabbitClientBus _rabbitClientBus; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; public TrackDeleteEventHandler(IProvideQueries provideQueries, IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackDetailsEsRepository trackDetailsEsRepository, RabbitClientBus rabbitClientBus, ITrackStatusRepository trackStatusRepository, ITrackContentRepository trackContentRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository) { _provideQueries = provideQueries; _orderRepository = orderRepository; _capPublisher = capPublisher; _trackDetailsEsRepository = trackDetailsEsRepository; _rabbitClientBus = rabbitClientBus; _trackStatusRepository = trackStatusRepository; _trackContentRepository = trackContentRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] public async Task ConsumerAsync(TrackDeleteEvent @event) { if (!CanDelete(@event)) return; var trackList = await DeleteTrackAsync(@event); //发布事件 await PublicTrackTopicAsync(@event, 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 PublicTrackTopicAsync(TrackDeleteEvent eto, List trackList) { if (trackList == null) return; var statusList = await _trackStatusRepository.GetListAsync(m => m.SubSystemId == eto.SubSystemId); if (statusList == null) return; var contentList = await _trackContentRepository.GetTrackContentBySyetemIdAsync(eto.SubSystemId); if (contentList == null) return; foreach (var dataItem in trackList.GroupBy(m => new { m.TrackCode, m.OrderNo })) { var item = dataItem.First(); var status = statusList.FirstOrDefault(m => m.Code == item.TrackStatusCode); if (status == null || !status.IsToMq) continue; var content = contentList.FirstOrDefault(m => m.TrackContentCode == item.TrackCode); var data = new TrackChangeEto() { OrderIdentifiers = dataItem.Select(m => new OrderIdentifier { OrderId = m.OrderId, OrderNo = m.OrderNo, OrderType = m.OrderType }).ToArray(), TrackStatusCode = item.TrackStatusCode, TrackCode = item.TrackCode, OperationMan = eto.OperationMan, SubSystemId = eto.SubSystemId, Option = 2 }; _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); } } private async Task> DeleteTrackAsync(TrackDeleteEvent @event) { List orderIds; List tracks; List deleteTracks; if (@event.EsIds != null && @event.EsIds.Any()) //根据明确的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 == TrackSourceEnum.ExternalServiceProvide.GetHashCode())) return deleteTracks; //查出订单下的所有轨迹,以供后面回退使用 tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); // 删除轨迹 foreach (var item in deleteTracks) { await _trackDetailsEsRepository.DeleteTrackByIdAsync(item.Id, item.IndexName); } //清理已经删除过的esid @event.EsIds = Array.Empty(); } else //原有逻辑,根据轨迹code和订单进行删除 { //根据订单查询出所有的轨迹,查所有是因为主单上需要进行回退更新 tracks = await _trackDetailsEsRepository.GetListAsync(@event.OrderIds); //构建标识符集合 var ids = @event.TrackIdentifiers.Select(x => $"{x.TrackCode}{x.TrackTime:G}").ToList(); //根据标识符选出要删除的轨迹明细 deleteTracks = tracks.Where(x => ids.Contains($"{x.TrackCode}{x.TrackTime:G}")).ToList(); if (deleteTracks.Count == 0) return deleteTracks; foreach (var track in deleteTracks) { await _trackDetailsEsRepository.DeleteTrackByIdAsync(track.Id, track.IndexName); } //从被删除轨迹寻找订单id,稍后会更新订单主表 orderIds = deleteTracks.Select(m => m.OrderId).Distinct().ToList(); } //排除掉已经删除的轨迹 tracks = tracks.Except(deleteTracks).ToList(); var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, 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.TrackTime) .FirstOrDefault(x => x.OrderId == order.Id); // 删除全部轨迹导致的重置行为 if (track == default) { if (order.LatestTrackTime == null) continue; // tips:订单是支持从已完成态到未完成态的 order.ResetTrackInfo(); } else { if (order.LatestTrackContent == track.TrackContent) continue; // 删除引发的回退更新 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; } // 写入操作日志 _ = _orderOperationRecordEsRepository.AddAsync(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 }); } await _orderRepository.UpdateRangeAsync(orders, x => new { x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime, x.LatestTrackTimeZone, x.EstimatedDeliveryDate, x.DestinationTimeZone }); return deleteTracks; } }