using DotNetCore.CAP; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using TrackCenter.Application.Events; using TrackCenter.Application.Kafka; using TrackCenter.Application.Queries; using TrackCenter.Application.Services.Broadcasting; using TrackCenter.Application.Services.Warning.Signals; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; using TrackCenter.Domain.Dtos.TrackContents; using TrackCenter.Repository.Aggregates.OrderAggregate; using TrackCenter.Repository.Otel; using OrderIdentifier = TrackCenter.Application.Events.OrderIdentifier; namespace TrackCenter.Application.CAP.EventHandlers; public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { private readonly ICapPublisher _capPublisher; private readonly IOrderRepository _orderRepository; private readonly ProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly BroadcastingService _broadcastingService; private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; private readonly WarningSignalPublisher _warningSignalPublisher; public TrackUpsertEventHandler( IOrderRepository orderRepository, ICapPublisher capPublisher, ProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, BroadcastingService broadcastingService, WarningSignalPublisher warningSignalPublisher ) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _warningSignalPublisher = warningSignalPublisher; } [UnitOfWork] [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] [CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)] public virtual 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.AppKey, x.Customer // ????????????????? })) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.AddRange(canNextOrderIds); nextEvent.AppKey = group.Key.AppKey; nextEvent.Customer = group.Key.Customer; nextEvent.TrackRefreshData = new TrackRefreshStruct { ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash, }; // ??????????? if (@event.AppKey != nextEvent.AppKey) { var mapCode = (await _provideQueries.MapTrackCodesAsync(@event.AppKey, nextEvent.AppKey, @event.TrackCode)) .FirstOrDefault(); if (mapCode == null) { // ????????????? continue; } nextEvent.TrackCode = mapCode; } var identifiers = group.Select(x => new OrderIdentifier { OrderId = x.OrderId, OrderNo = x.OrderNo, OrderType = x.OrderType, }); foreach (var subOrderIds in identifiers.Chunk(GlobalConst.ChunkSize)) { var subEvent = nextEvent.DeepCopy(); subEvent.OrderIdentifiers = subOrderIds.ToArray(); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, subEvent); } } } } private static bool CanDelete(TrackUpsertEvent @event) { // ???????????? if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false }) { Log.Warning($"??????????????{@event}", true); return false; } var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var dup = @event.Visited.Intersect(orderIds).ToArray(); if (dup.Any()) { Log.Warning($"{(@event.IsAdd ? "??" : "??")}???????????{dup.JoinAsString(",")}", true); return false; } return true; } private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) { //#7001:?????????????????????????????????????? var trackContent = await _provideQueries.GetTrackContentAsync(@event.AppKey, @event.TrackCode); if (trackContent == null) { // ???????????? Log.Warning($"????????{@event}", true); return; } // ??????????????? var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial; // ??????? if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false) { PublicTopicAllFailed(@event, trackContent, "?????????????"); return; } // ??????? if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false) { PublicTopicAllFailed(@event, trackContent, "?????????????"); return; } // k:[orderId] v:{EsId,Version,TrackCreateTime} var orderIdTrackMetadataDict = new Dictionary(); var trackChangeEtos = new List(); var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent, orderIdTrackMetadataDict, trackChangeEtos); if (canNextOrderIds.Count == 0) { await TrackChangePublicTopicAsync(trackContent, trackChangeEtos); return; } if (insertTracks.Any()) { // ??add to es tips:?????????? await _trackDetailsEsRepository.AddManyAsync(insertTracks); // es?????????? insertTracks.ForEach(t => orderIdTrackMetadataDict.Add(t.OrderId, TrackMetadata.Build(t.Id, t.Version, t.CreateTime))); //?????????? _ = _orderOperationRecordEsRepository.AddManyAsync(insertTracks.Select(t => BuildOrderOperationRecord(@event, t, new OrderIdentifier { OrderNo = t.OrderNo, OrderId = t.OrderId })).ToList()); // ?????????????? await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(), @event.TrackCode); // ???????????????? _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, insertTracks); // ??????????????pod?? if (!string.IsNullOrWhiteSpace(trackContent.TrackFlagId)) { await _capPublisher.PublishAsync(GlobalConst.OrderPodPullSubscribe, new OrderPodPullEvent { AppKey = @event.AppKey, OrderIdEsIdDict = insertTracks.ToDictionary(x => x.OrderId, y => y.Id), TrackFlagIds = trackContent.TrackFlagId.Split(',').Select(int.Parse).ToHashSet() }); } } // ???????? await TrackChangePublicTopicAsync(trackContent, trackChangeEtos, true); // ?????????????? await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe, new TrackChangeAfterUpdateOrderEvent { OrderIds = canNextOrderIds.ToArray(), IsCompleted = trackContent.IsCompleted, TrackStatus = trackContent.TrackStatus, TrackContent = trackContent.TrackContent, TrackEnContent = trackContent.TrackEnContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, }); // 触发轨迹推送事件(异步) // tips:???????????????? var trackPushEvent = new TrackPushEvent { OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, AppKey = @event.AppKey, Customer = @event.Customer, OrderIdTrackMetadataDict = orderIdTrackMetadataDict, TrackCode = @event.TrackCode, SystemPushInfo = new SystemPushStruct { TrackStatusId = trackContent.TrackStatusId, TrackStatus = trackContent.TrackStatus, TrackContentId = trackContent.TrackContentId, TrackContent = trackContent.TrackContent, TrackPlace = @event.TrackPlace, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, Extra = @event.Extra }, ForcedPush = @event.ForcedPush, IsToMq = (trackContent.IsToMq || @event.IsToMq) }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); } private async Task> UpsertTrackAsync( TrackUpsertEvent @event, List canNextOrderIds, TrackContentVO trackContent, Dictionary orderIdTrackMetadataDict, List trackChangeEtos ) { var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); // important:轨迹code相同,时间、发生地不同的,可能存在多条!! var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true); var insertTracks = new List(); var updateTracks = new List(); var operationRecords = new List(); var traceLog = new Dictionary { { "@event", @event.ToJson() } }; foreach (var order in @event.OrderIdentifiers) { // fast path: by update with es id case if (!@event.IsAdd && !string.IsNullOrWhiteSpace(@event.EsId)) { var trackById = tracks.FirstOrDefault(x => x.Id == @event.EsId); if (trackById != null) { var updateTrack = BuiltUpdateTrack(@event, trackById, order, trackContent); updateTracks.Add(updateTrack); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, updateTrack.Id); } else { var message = $"???????EsId:[{@event.EsId}]????????????"; AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); } continue; } var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList(); // tips:??????????8?????? var timezone = Math.Abs(@event.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 0.01f ? 8.0f : @event.TrackTimeZone; // ?????? existTracks = existTracks.Where(x => x.GetUtc0TrackTime() == @event.TrackTime.Subtract(TimeSpan.FromHours(timezone))).ToList(); // ???????? existTracks = existTracks.Where(x => x.Extra == @event.Extra).ToList(); if (@event.IsAdd) { if (existTracks.Count > 0) { const string message = "??????????????????????"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; } var addTrack = BuiltNewTrack(@event, order, trackContent); insertTracks.Add(addTrack); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, addTrack.Id); } else { // ?????? if (existTracks.Count == 0) { const string message = "?????????????"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; } if (existTracks.Count > 1) { const string message = "?????????????"; traceLog.Add($"OrderNo-{order.OrderNo}", message); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message); continue; } var track = existTracks.First(); var updateTrack = BuiltUpdateTrack(@event, track, order, trackContent); updateTracks.Add(updateTrack); operationRecords.Add(BuildOrderOperationRecord(@event, track, order)); AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, updateTrack.Id); } canNextOrderIds.Add(order.OrderId); } ActivityHelper.StartSpanActivity("UpsertTrackAsync", traceLog); if (updateTracks.Any()) { // ????????? await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks); // ???????? updateTracks.ForEach(x => orderIdTrackMetadataDict.Add(x.OrderId, TrackMetadata.Build(x.Id, x.Version, x.CreateTime))); // ???????? _ = _orderOperationRecordEsRepository.AddManyAsync(operationRecords); // ??????????????? _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); } return insertTracks; } private static TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order, TrackContentVO trackContent) { // ?????????? ????????????? if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide) { track.TrackPlace = @event.TrackPlace; track.TrackTime = @event.TrackTime; track.TrackTimeZone = @event.TrackTimeZone; } //?????????? track.TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content; track.TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus; // ?????????? track.TrackPlaceCity = GlobalConst.EmptyString; track.TrackPlaceCityId = GlobalConst.NotExistId; track.TrackPlaceProvince = GlobalConst.EmptyString; track.TrackPlaceProvinceId = GlobalConst.NotExistId; track.TrackPlaceNation = GlobalConst.EmptyString; track.TrackPlaceNationId = GlobalConst.NotExistId; track.TrackPlacePostcode = GlobalConst.EmptyString; // ?????????????????????? track.DestinationTimeZone = null; track.EstimatedDeliveryDate = null; // ???utc0??? track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); // ????ids???????????????? track.TrackFlagIds = trackContent.TrackFlagId; track.IsExternal = @event.IsExternal ?? trackContent.IsExternal; track.Remarks = @event.Remarks ?? GlobalConst.EmptyString; track.Extra = @event.Extra; track.Files = @event.Files?.Select(x => new FileItem { OssKey = x.OssKey, FileName = x.FileName }).ToList(); // ???? track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser; track.UpdateTime = DateTime.Now; return track; } private static TrackDetailsIndex BuiltNewTrack(TrackUpsertEvent @event, OrderIdentifier order, TrackContentVO trackContent) { var track = new TrackDetailsIndex { // ???? AppKey = @event.AppKey, OrderId = order.OrderId, OrderNo = order.OrderNo, OrderType = order.OrderType, // ???? TrackCode = @event.TrackCode, TrackContentId = trackContent.TrackContentId, TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content, TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus, TrackStatusId = trackContent.TrackStatusId, TrackStatus = trackContent.TrackStatus, TrackStatusCode = trackContent.TrackStatusCode, TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone, TrackPlace = @event.TrackPlace, // ??????? ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString, OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString, OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString, // ?????????? TrackPlaceCity = GlobalConst.EmptyString, TrackPlaceCityId = GlobalConst.NotExistId, TrackPlaceProvince = GlobalConst.EmptyString, TrackPlaceProvinceId = GlobalConst.NotExistId, TrackPlaceNation = GlobalConst.EmptyString, TrackPlaceNationId = GlobalConst.NotExistId, TrackPlacePostcode = GlobalConst.EmptyString, // ????ids????????????????????? TrackFlagIds = trackContent.TrackFlagId, Source = @event.Source, IsExternal = @event.IsExternal ?? trackContent.IsExternal, Remarks = @event.Remarks ?? GlobalConst.EmptyString, Extra = @event.Extra, Files = @event.Files?.Select(x => new FileItem { OssKey = x.OssKey, FileName = x.FileName }).ToList(), // ???? CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, CreateTime = DateTime.Now }; // ???utc0??? track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); return track; } private static OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order) { return new OrderOperationRecordIndex { OrderId = order.OrderId, OrderNo = order.OrderNo, OptionType = @event.IsAdd ? OrderOperationType.创建轨迹.ToString() : OrderOperationType.修改轨迹.ToString(), Content = track.ToJson(), CreateUserId = @event.OperationManId, CreateUserName = @event.OperationMan ?? "", CreateTime = DateTime.Now }; } private async Task TrackChangePublicTopicAsync(TrackContentVO trackContent, List list, bool isDelay = false) { if (list.Count == 0) return; //??????????s????https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html //????????????????? //?????????? //??1??!???????????????????????????? //??2?????????es???????? //??3???????????????? if (isDelay) await Task.Delay(1000); foreach (var eto in list) _broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); } private void PublicTopicAllFailed(TrackUpsertEvent @event, TrackContentVO trackContent, string failedReason) { if (trackContent.IsToMq == false && @event.IsToMq == false) return; foreach (var orderNo in @event.OrderIdentifiers.Select(x => x.OrderNo)) { var eto = BuildEto(@event, trackContent, orderNo, null, failedReason); _broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey); } } private static void AddTrackChangeEto(List list, TrackUpsertEvent @event, TrackContentVO trackContent, string orderNo, string? trackEsId = null, string? failedReason = null) { if (trackContent.IsToMq == false && @event.IsToMq == false) return; list.Add(BuildEto(@event, trackContent, orderNo, trackEsId, failedReason)); } private static TrackChangeEto BuildEto(TrackUpsertEvent @event, TrackContentVO trackContent, string orderNo, string? trackEsId = null, string? failedReason = null) { return new TrackChangeEto { TrackEsId = trackEsId, Option = @event.IsAdd ? 1 : 3, IsSuccess = string.IsNullOrWhiteSpace(failedReason), FailedReason = failedReason, TrackPlacePostcode = null, TrackPlaceNation = null, TrackPlaceNationId = null, TrackPlaceProvince = null, TrackPlaceProvinceId = null, TrackPlaceCity = null, TrackPlaceCityId = null, Files = @event.Files, Remarks = @event.Remarks, Source = @event.Source, IsExternal = @event.IsExternal ?? trackContent.IsExternal, TrackTimeZone = @event.TrackTimeZone, TrackTime = @event.TrackTime, TrackPlace = @event.TrackPlace, TrackStatusCode = trackContent.TrackStatusCode, OrderNo = orderNo, EnStatus = @event.EnStatus, Content = @event.Content, TrackCode = @event.TrackCode, OperationMan = @event.OperationMan, OrderIdentifiers = @event.OrderIdentifiers, Customer = @event.Customer, AppKey = @event.AppKey, Extra = @event.Extra }; } }