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.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; public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; } [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(); await HandleTrackAsync(@event, canNextOrderIds); 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 { ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, 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; } var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); // important:轨迹code相同,时间、发生地不同的,可能存在多条!! var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); var insertTracks = new List(); foreach (var order in @event.OrderIdentifiers) { TrackDetailsIndex? track; if (!@event.IsAdd) { var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToArray(); if (existTracks.Length > 1) { // 修改轨迹时,订单存在多条轨迹不允许修改 continue; } track = existTracks.FirstOrDefault(); if (track == null) continue; // 不存在不让改 } else { track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId && x.TrackTime == @event.TrackTime && x.TrackPlace == @event.TrackPlace); if (track != null) continue; // 存在不让创建 } 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.TrackTime, TrackTimeZone = @event.TrackTimeZone, ServiceProviderName = @event.TrackRefreshData.ServiceProviderName ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString, CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now, TrackStatus = trackContent.TrackStatus, TrackContent = trackContent.TrackContent, TrackEnContent = trackContent.TrackEnContent }; track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); insertTracks.Add(track); } else { track.TrackPlace = @event.TrackPlace; track.TrackTime = @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; await _trackDetailsEsRepository.UpdateAsync(track); } canNextOrderIds.Add(order.OrderId); } if (canNextOrderIds.Count == 0) return; if (insertTracks.Any()) { // 批量add to es await _trackDetailsEsRepository.AddManyAsync(insertTracks); // add track触发订单拉取状态更新 await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), @event.TrackCode); } // 触发订单主表更新事件(异步) await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { SubSystemId = @event.SubSystemId, CustomId = @event.CustomId, OrderIds = canNextOrderIds.ToArray(), IsCompleted = trackContent.IsCompleted, TrackContent = trackContent.TrackContent, TrackEnContent = trackContent.TrackEnContent, TrackStatusId = trackContent.TrackStatusId, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, TrackCode = @event.TrackCode, DeficiencyTagValue = @event.DeficiencyTagValue }); // 触发轨迹推送事件(异步) var trackPushEvent = new TrackPushEvent { OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, SubSystemId = @event.SubSystemId, CustomId = @event.CustomId, OrderIds = canNextOrderIds.ToArray(), TrackCode = @event.TrackCode, SystemPushInfo = new SystemPushStruct { TrackStatus = trackContent.TrackStatus, TrackContent = trackContent.TrackContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone } }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); } }