using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.TrackPush; 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.Consts; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.EventHandlers; public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ITrackPushCacheService _trackPushCacheService; private readonly ICapPublisher _capPublisher; public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository, ITrackPushCacheService trackPushCacheService) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsRepository = trackDetailsRepository; _trackPushCacheService = trackPushCacheService; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] public async Task ConsumerAsync(TrackUpsertEvent @event) { if (@event.Visited.Contains(@event.OrderId)) { LogHelper.Error($"轨迹订单循环依赖预警:{@event.Visited.JoinAsString(",")}"); return; } var isUnProcessed = false; // 不同子系统需要做内容映射 if (@event.UpSubSystemId != 0 && @event.UpSubSystemId != @event.SubSystemId) { var mapCode = (await _provideQueries .MapTrackCodesAsync(@event.UpSubSystemId, @event.SubSystemId, @event.TrackCode)) .FirstOrDefault(); if (mapCode == default) { // 跨子系统没做映射的轨迹不处理 return; } @event.TrackCode = mapCode; } var track = @event.TrackCode == GlobalConst.EmptyString ? await _trackDetailsRepository.GetFirstOrDefaultAsync(@event.OrderId, @event.TrackTime) : await _trackDetailsRepository.GetFirstOrDefaultAsync(@event.OrderId, @event.TrackCode); if (@event.TrackCode == GlobalConst.EmptyString && track != null) return; // 无法映射的轨迹只做新增 if (@event.IsAdd != null) // 手动操作需要做区分 { if (track == null && !@event.IsAdd!.Value) return; // 不存在不让改 if (track != null && @event.IsAdd!.Value) return; // 存在不让创建 } // step1、upsert es轨迹 track = await UpsertTrackAsync(track, @event); if (track == null || @event.TrackCode == GlobalConst.EmptyString) return; // 模板没配该轨迹代码,不再继续 // step2、触发轨迹推送事件(异步) await HandleTrackPushAsync(@event, track); // step3、触发订单主表更新事件(异步) await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { OrderId = track.OrderId, IsCompleted = track.IsCompleted, TrackCode = track.TrackCode, TrackContent = track.TrackContent, TrackEnContent = track.TrackEnContent, TrackPlace = track.TrackPlace, TrackTime = track.TrackTime, TrackTimeZone = track.TrackTimeZone }); // step4、根据订单关联关系异步向下刷(异步) var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderId); foreach (var linkedOrder in linkedOrders) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.Add(@event.OrderId); nextEvent.OrderId = linkedOrder.OrderId; nextEvent.OrderType = linkedOrder.OrderType; nextEvent.OrderNo = linkedOrder.OrderNo; nextEvent.SubSystemId = linkedOrder.SubSystemId; nextEvent.LatestPushTrackContentId = linkedOrder.LatestPushTrackContentId; nextEvent.CustomId = linkedOrder.CustomId; nextEvent.UpSubSystemId = @event.SubSystemId; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent); } } /// /// 不存在则新增轨迹,存在则更新 /// /// /// private async Task UpsertTrackAsync(TrackDetailsIndex? track, TrackUpsertEvent @event) { if (track == default) { track = new TrackDetailsIndex { IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}", OrderId = @event.OrderId, OrderNo = @event.OrderNo, OrderType = @event.OrderType, TrackCode = @event.TrackCode, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, IsCanDelete = false, OriginTrackContent = @event.OriginTrackContent, OriginTrackData = @event.OriginTrackData, CreateUser = @event.OperationMan, CreateTime = DateTime.Now }; // tips:刷新到无法匹配内容的轨迹引起的插入 if (@event.TrackCode == GlobalConst.EmptyString) { track.TrackStatus = GlobalConst.EmptyString; track.TrackContent = GlobalConst.EmptyString; track.TrackContentId = 0; track.IsCompleted = false; track.TrackEnContent = GlobalConst.EmptyString; } else { //todo:可以优化 // 根据轨迹code在订单模板内容中寻找匹配的 var trackContent = await _provideQueries.MatchTrackContentAsync(@event.OrderId, @event.TrackCode); if (trackContent == default) { // 订单所绑定的模板,没有一条内容可以匹配发生的轨迹 return null; } track = new TrackDetailsIndex { TrackStatus = trackContent.Status, TrackContent = trackContent.Content, TrackContentId = trackContent.Id, IsCompleted = trackContent.IsCompleted, TrackEnContent = trackContent.EnContent, IsCanDelete = trackContent.IsCanDelete }; track.ZeroTrackTime = track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); } } else { // tips:刷新到无法匹配内容的轨迹 相同的发生时间不做更新处理 if (@event.TrackCode == GlobalConst.EmptyString) return track; track.TrackPlace = @event.TrackPlace; track.TrackTime = @event.TrackTime; track.TrackTimeZone = @event.TrackTimeZone; track.UpdateUser = @event.OperationMan ?? "系统"; track.UpdateTime = DateTime.Now; } await _trackDetailsRepository.UpsertAsync(track); return track; } /// /// 处理轨迹自动推送 /// /// /// private async Task HandleTrackPushAsync(TrackUpsertEvent @event, TrackDetailsIndex track) { TrackPushContentConfigDto? config; OptionType optionType = OptionType.修改; // 推送更新 if (!track.Id.IsNullOrWhiteSpace()) { config = await _trackPushCacheService.GetPushContentConfigAsync( @event.SubSystemId, @event.CustomId, track.TrackContentId); if (config == null || !config.IsCanUpdate) return; } else { // 推送新增,需要根据上一次的推送 推断 本次推送 是否可行 config = await _trackPushCacheService.InferNextPushContentOrNullAsync( @event.SubSystemId, @event.CustomId, @event.LatestPushTrackContentId); if (config == null || !config.IsAutoPush) return; // 当前发生的轨迹就是要推送的 if (config.TrackContentId != track.TrackContentId) return; optionType = OptionType.新增; } if (config.OrderType != @event.OrderType) return; var trackPushEvent = new TrackPushEvent { OptionType = optionType, ServiceProviderCode = config.ServiceProviderCode, OrderNo = track.OrderNo, OrderType = track.OrderType, OrderId = track.OrderId, TrackContentId = config.TrackContentId, TrackStatus = track.TrackStatus, TrackCode = track.TrackCode, TrackContent = track.TrackContent, TrackPlace = track.TrackPlace, TrackTime = track.TrackTime.AddHours(config.TrackTimeZone ?? track.TrackTimeZone - track.TrackTimeZone), TrackTimeZone = (float)(config.TrackTimeZone ?? track.TrackTimeZone), TrackContentCodeValue = config.TrackContentCodeValue, TrackContentValue = config.TrackContentValue, TrackEnContentValue = config.TrackEnContentValue, TrackStatusValue = config.TrackStatusValue, ExpandFieldsValue = config.ExpandFieldsValue }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); } }