Merge branch 'S06' into test

src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs

@@ -1,3 +1,4 @@
+using System.Collections.Immutable;
 using System.Text.Json;
 using DotNetCore.CAP;
 using Microsoft.Extensions.Options;
@@ -11,6 +12,7 @@ using SJZY.Track.Application.Helpers;
 using SJZY.Track.Application.Options;
 using SJZY.Track.Application.Track2InterfaceRelay;
 using SJZY.Track.Domain.Aggregates.OrderAggregate;
+using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
 using SJZY.Track.Domain.Enums;
 using Volo.Abp.DependencyInjection;
@@ -25,6 +27,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     private readonly IOrderRepository _orderRepository;
     private readonly ITrackPushCacheService _trackPushCacheService;
     private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
+    private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
     private readonly IBroadcastingService _broadcastingService;
     private List<OrderEntity> _orders = null!;
     private List<TrackPushContentConfigDto> _pushConfigs = null!;
@@ -34,7 +37,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
         ITrackPushCacheService trackPushCacheService,
         ITrackPushRecordEsRepository trackPushRecordEsRepository,
         IOptionsMonitor<InterfaceRelayParamsConfigOptions> options,
-        IBroadcastingService broadcastingService)
+        IBroadcastingService broadcastingService,
+        ITrackDetailsEsRepository trackDetailsEsRepository)
     {
         _httpClientProxy = httpClientProxy;
         _orderRepository = orderRepository;
@@ -42,6 +46,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
         _trackPushRecordEsRepository = trackPushRecordEsRepository;
         _options = options;
         _broadcastingService = broadcastingService;
+        _trackDetailsEsRepository = trackDetailsEsRepository;
     }
 
     [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
@@ -96,130 +101,21 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     private async Task DoPushAsync(TrackPushEvent @event)
     {
         var items = new List<TrackPushInput>();
-        TrackPushContentConfigDto? config;
+        if (@event.OperationMan == null)
-        var isSystemPush = @event.OperationMan == null;
-        if (isSystemPush && !@event.ForcedPush)
         {
-            config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
+            if (@event.OptionType == OptionType.新增)
-            foreach (var order in _orders)
             {
-                if (config == null)
+                await BuiltSystemPushItemsAsync(@event, items);
-                {
+            }
-                    TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
+            else if (@event.OptionType == OptionType.修改)
-                    return;
+            {
-                }
+                await BuiltManualPushItemsAsync(@event, items);
-
-                // 系统不推异常单，只能人工操作
-                if (order.QuestionType == (int) TrackProblemType.Error)
-                {
-                    TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order);
-                    continue;
-                }
-
-                // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序
-                var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
-                if (resultConfig == null)
-                {
-                    TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order);
-                    continue;
-                }
-
-                config = resultConfig;
-
-                // 是否自动推送
-                if (!config.IsAutoPush)
-                {
-                    TrackPushFailedPublicTopic($"没有设置为自动推送，如果要推送请手动推送", @event, order);
-                    continue;
-                }
-
-                // 配置的类型如何和订单类型不一致则不允许推送
-                if (config.OrderType != order.OrderType)
-                {
-                    TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order);
-                    continue;
-                }
-
-                var log = new TrackPushRecordIndex
-                {
-                    OptionType = @event.OptionType.ToString(),
-                    OrderId = order.Id,
-                    OrderNo = order.OrderNo,
-                    TrackStatus = @event.SystemPushInfo.TrackStatus,
-                    TrackCode = @event.TrackCode,
-                    TrackContent = @event.SystemPushInfo.TrackContent,
-                    TrackPlace = @event.SystemPushInfo.TrackPlace,
-                    TrackTime = @event.SystemPushInfo.TrackTime,
-                    TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
-                    CreateUser = GlobalConst.SystemUser,
-                    SubSystemCode = @event.SubSystemCode,
-                    IsToMq = @event.IsToMq,
-                    Extra = @event.SystemPushInfo.Extra
-                };
-                var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
-                item.Log = JsonConvert.SerializeObject(log);
-                items.Add(item);
             }
         }
-        // 手工或强制推送，不会校验已推和顺序问题，因为可能是更新
         else
-
         {
-            config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
+            // 手工推送，不会校验已推和顺序问题，因为可能是更新
-            foreach (var order in _orders)
+            await BuiltManualPushItemsAsync(@event, items);
-            {
-                if (config == null)
-                {
-                    TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
-                    return;
-                }
-
-                if (config is { IsAllPush: true })
-                {
-                    // 非强制推送需要 推断 本次推送 是否可行
-                    if (!@event.ForcedPush)
-                    {
-                        var (resultConfig, nextCode) = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
-                        if (resultConfig == null)
-                        {
-                            TrackPushFailedPublicTopic($"推送顺序异常，下一个要推送的应为[{nextCode}]", @event, order);
-                            continue;
-                        }
-
-                        config = resultConfig;
-                    }
-
-                    // tips:强制推送直接使用配置
-                }
-
-
-                //配置的类型如何和订单类型不一致则不允许推送
-                if (config.OrderType != order.OrderType)
-                {
-                    TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order);
-                    continue;
-                }
-
-                var log = new TrackPushRecordIndex
-                {
-                    OptionType = @event.OptionType.ToString(),
-                    OrderId = order.Id,
-                    OrderNo = order.OrderNo,
-                    TrackStatus = @event.SystemPushInfo.TrackStatus,
-                    TrackCode = @event.TrackCode,
-                    TrackContent = @event.SystemPushInfo.TrackContent,
-                    TrackPlace = @event.SystemPushInfo.TrackPlace,
-                    TrackTime = @event.SystemPushInfo.TrackTime,
-                    TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
-                    CreateUser = @event.OperationMan,
-                    SubSystemCode = @event.SubSystemCode,
-                    IsToMq = @event.IsToMq,
-                    Extra = @event.SystemPushInfo.Extra
-                };
-                var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
-                item.Log = JsonConvert.SerializeObject(log);
-                items.Add(item);
-            }
         }
 
         if (items.Count == 0) return;
@@ -236,12 +132,133 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
             if (r.Code != GlobalConst.SuccessCode)
             {
-                TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast<dynamic>().ToList());
+                TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
-                throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
+                    items.Cast<dynamic>().ToList());
+                throw new BusinessException(
+                    message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
             }
         }
     }
 
+    private async Task BuiltManualPushItemsAsync(TrackPushEvent @event, List<TrackPushInput> items)
+    {
+        // tips: 手工强制推送直接使用配置
+        var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
+        foreach (var order in _orders)
+        {
+            if (config is { IsAllPush: true })
+            {
+                // 非强制推送需要 推断 本次推送 是否可行
+                if (!@event.ForcedPush)
+                {
+                    config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
+                }
+            }
+
+            if (config == null)
+            {
+                TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order);
+                return;
+            }
+
+
+            //配置的类型如何和订单类型不一致则不允许推送
+            if (config.OrderType != order.OrderType)
+            {
+                TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event,
+                    order);
+                continue;
+            }
+
+            var log = new TrackPushRecordIndex
+            {
+                OptionType = @event.OptionType.ToString(),
+                OrderId = order.Id,
+                OrderNo = order.OrderNo,
+                TrackStatus = @event.SystemPushInfo.TrackStatus,
+                TrackCode = @event.TrackCode,
+                TrackContent = @event.SystemPushInfo.TrackContent,
+                TrackPlace = @event.SystemPushInfo.TrackPlace,
+                TrackTime = @event.SystemPushInfo.TrackTime,
+                TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
+                CreateUser = @event.OperationMan,
+                SubSystemCode = @event.SubSystemCode,
+                IsToMq = @event.IsToMq,
+                Extra = @event.SystemPushInfo.Extra
+            };
+            var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
+            item.Log = JsonConvert.SerializeObject(log);
+            items.Add(item);
+        }
+    }
+
+    private async Task BuiltSystemPushItemsAsync(TrackPushEvent @event, List<TrackPushInput> items)
+    {
+        var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
+        foreach (var order in _orders)
+        {
+            if (config == null)
+            {
+                TrackPushFailedPublicTopic($"没有在推送映射配置中找到关于轨迹代码[{@event.TrackCode}]的配置", @event, order);
+                return;
+            }
+
+            // 系统不推异常单，只能人工操作
+            if (order.QuestionType == (int) TrackProblemType.Error)
+            {
+                TrackPushFailedPublicTopic("系统不推异常单，只能人工操作", @event, order);
+                continue;
+            }
+
+            // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序
+            // 如果当前推送code与订单上的最后的code不一致，则考虑推送映射顺序
+            if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode))
+            {
+                config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
+                if (config == null)
+                {
+                    TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order);
+                    continue;
+                }
+            }
+
+            // 是否自动推送
+            if (!config.IsAutoPush)
+            {
+                TrackPushFailedPublicTopic("没有设置为自动推送，如果要推送请手动推送", @event, order);
+                continue;
+            }
+
+            // 配置的类型如何和订单类型不一致则不允许推送
+            if (config.OrderType != order.OrderType)
+            {
+                TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event,
+                    order);
+                continue;
+            }
+
+            var log = new TrackPushRecordIndex
+            {
+                OptionType = @event.OptionType.ToString(),
+                OrderId = order.Id,
+                OrderNo = order.OrderNo,
+                TrackStatus = @event.SystemPushInfo.TrackStatus,
+                TrackCode = @event.TrackCode,
+                TrackContent = @event.SystemPushInfo.TrackContent,
+                TrackPlace = @event.SystemPushInfo.TrackPlace,
+                TrackTime = @event.SystemPushInfo.TrackTime,
+                TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
+                CreateUser = GlobalConst.SystemUser,
+                SubSystemCode = @event.SubSystemCode,
+                IsToMq = @event.IsToMq,
+                Extra = @event.SystemPushInfo.Extra
+            };
+            var item = await BuiltPushItem(config, log, @event.SystemPushInfo.CompareFields);
+            item.Log = JsonConvert.SerializeObject(log);
+            items.Add(item);
+        }
+    }
+
     private async Task DoUpdateAsync(TrackPushEvent @event)
     {
         var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode);
@@ -253,7 +270,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
             //配置的类型如何和订单类型不一致则不允许推送
             if (config.OrderType != order.OrderType)
             {
-                TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event, order);
+                TrackPushFailedPublicTopic($"订单类型{order.OrderType}和配置中的订单类型[{config.OrderType}]不一致，不能推送", @event,
+                    order);
                 continue;
             }
 
@@ -292,8 +310,10 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
             if (r.Code != GlobalConst.SuccessCode)
             {
-                TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId, items.Cast<dynamic>().ToList());
+                TrackPushFailedPublicTopic($"轨迹中台调用接口通失败,RequestId: {r.RequestId}; Msg: {r.Msg}", @event, r.RequestId,
-                throw new BusinessException(message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
+                    items.Cast<dynamic>().ToList());
+                throw new BusinessException(
+                    message: $"requestId: {r.RequestId}; serviceProviderCode:{serviceProviderCode} Msg: {r.Msg}");
             }
         }
     }
@@ -351,104 +371,69 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
         if (config.IsAllPush)
         {
-            var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId);
+            var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true);
-            //找到历史中推送成功的轨迹
-            var trackPushRecords = await _trackPushRecordEsRepository.GetHistorySuccessTrackPushRecordAsync(log.OrderId);
-            //构建标识符，code+比较字段的值(如果提供的比较字段,否则使用轨迹时间)
-            var pushIdentitySuccess = trackPushRecords.Select(x =>
-            {
-                //寻找映射配置
-                var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue;
-                if (string.IsNullOrWhiteSpace(compareFields))
-                    return code;
-
-                //尝试从扩展中提取目标值
-                var extraValue = TextJsonHelper.ExtractValue(x.Extra, compareFields);
-                return code + extraValue;
-            }).ToHashSet();
-
             if (history != default)
             {
                 body = TextJsonHelper.Deserialize<TrackPushInput>(history.PushContent);
-                //todo-test:在历史选出推送成功的历史轨迹
+
+                //在推送历史中，选出当前轨迹明细一致的轨迹
+                var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId);
+                //确定轨迹标识符
+                var trackIdentifies = tracks.Select(x =>
+                    {
+                        //寻找映射配置(转换后的code)
+                        var code = _pushConfigs.First(y => y.TrackCode == x.TrackCode).TrackContentCodeValue;
+                        var timeStr = DateTimeFormat(x.TrackTime, config.TrackTimeZone ?? log.TrackTimeZone, config);
+                        return string.IsNullOrWhiteSpace(compareFields)
+                            ? $"{code}{timeStr}"
+                            : $"{code}{TextJsonHelper.ExtractValue(x.Extra, compareFields)}";
+                    }
+                ).ToImmutableHashSet();
+                //在历史中选取在订单当前轨迹明细中的记录
                 body!.Data = body.Data.Where(x =>
                 {
                     var code = ((JsonElement) x["code"]!).GetString();
-                    if (string.IsNullOrWhiteSpace(compareFields))
+                    var timeStr = ((JsonElement) x["date"]!).GetString();
-                        return pushIdentitySuccess.Contains(code);
 
-                    var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement);
+                    if (string.IsNullOrWhiteSpace(compareFields))
-                    var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null;
+                        return trackIdentifies.Contains($"{code}{timeStr}");
-                    return pushIdentitySuccess.Contains(code + extraValue);
+
+                    if (((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var comparerValueJsonElement))
+                        return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}");
+
+                    return trackIdentifies.Contains($"{code}{null}");
                 }).ToList();
 
-                //添加当前要推送的轨迹
-                body.Data.Add(cur);
 
                 if (config.IsAllPushWithLatest)
                 {
-                    //历史轨迹中同一个code+比较字段联合作为key最新的一个
+                    //同code同比较字段保证唯一
-                    var dict = new Dictionary<string, (string dateStr, object dict)>();
+                    var old = body!.Data
-                    //todo-test:根据code和比较字段值分组，在每组里选出一个创建时间最新的
+                        .Where(x =>
-                    foreach (var x in body.Data)
+                        {
+                            //反序列化出的JsonElement和字典字符串类型做比较
+                            var a = ((JsonElement) x["code"]!).GetString();
+                            var b = cur["code"]!
+                                .ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同
+                            return a == b;
+                        })
+                        .WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
+                        {
+                            if (x["extra"] == null || log.Extra == null) return false;
+                            return TextJsonHelper.EqByJsonElementField(x["extra"]!,
+                                TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
+                        })
+                        .FirstOrDefault();
+
+                    if (old != null)
                     {
-                        var code = string.Empty;
+                        body.Data.Remove(old);
-                        var dateStr = string.Empty;
-                        var key = string.Empty;
-                        if (x["code"] is JsonElement codeEl) code = codeEl.GetString()!;
-                        else code = x["code"]!.ToString();
-                        if (x["date"] is JsonElement dateEl) dateStr = dateEl.GetString()!;
-                        else dateStr = x["date"]!.ToString();
-
-                        if (string.IsNullOrWhiteSpace(compareFields))
-                            key = code;
-                        else
-                        {
-                            var isMatch = ((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var jsonElement);
-                            var extraValue = isMatch ? TextJsonHelper.GetString(jsonElement) : null;
-                            key = code + extraValue;
-                        }
-
-                        //只保留最新的一条数据
-                        if (dict.TryGetValue(key, out var tuple))
-                        {
-                            //如果当前值创建时间大于历史值，则替换历史值
-                            if (DateTime.Parse(dateStr) > DateTime.Parse(tuple.dateStr))
-                                dict[key] = (dateStr, x);
-                        }
-                        else
-                            dict[key] = (dateStr, x);
                     }
-
-                    // //同code同比较字段保证唯一
-                    // var old = body!.Data
-                    //     .Where(x =>
-                    //     {
-                    //         //反序列化出的JsonElement和字典字符串类型做比较
-                    //         var a = ((JsonElement) x["code"]!).GetString();
-                    //         var b = cur["code"]!.ToString(); //tips: 需要要将object类型的字符串转为string类型，否则使用的是引用比较，反序列化出的string一定和本地string引用不同
-                    //         return a == b;
-                    //     }).WhereIf(!compareFields.IsNullOrWhiteSpace(), x =>
-                    //     {
-                    //         if (x["extra"] == null || log.Extra == null) return false;
-                    //         return TextJsonHelper.EqByJsonElementField(x["extra"]!, TextJsonHelper.Deserialize<JsonElement>(log.Extra), compareFields!);
-                    //     })
-                    //     .FirstOrDefault();
-                    //
-                    // if (old != null)
-                    // {
-                    //     body.Data.Remove(old);
-                    // }
-
-                    //选出每个组最新的轨迹
-                    body.Data = dict.Values.Select(x => x.dict).Cast<Dictionary<string, object?>>().ToList();
                 }
 
-                // //添加当前要推送的轨迹
+                // 添加当前要推送的轨迹
-                // body.Data.Add(cur);
+                body.Data.Add(cur);
             }
-
-            //todo-test:按推送映射顺序排序
         }
 
         log.PushContent = TextJsonHelper.Serialize(body)!;
@@ -460,24 +445,28 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     /// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了
     /// 则查询有没有可交换的推送，没有就返回null
     /// </summary>
-    private static (TrackPushContentConfigDto? config, string nextCode) GetValidPushConfigAsync(List<TrackPushContentConfigDto> cache,
+    private static TrackPushContentConfigDto? GetValidPushConfigAsync(List<TrackPushContentConfigDto> cache,
-        string? latestPushTrackCode, string? trackCode)
+        string? latestPushTrackCode, TrackPushEvent @event)
     {
+        var trackCode = @event.TrackCode;
         var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode);
 
+        if (@event.ForcedPush && trackCode == latestPushTrackCode)
+        {
+            //强制推送，允许重复推送当前轨迹
+            return cache[idx];
+        }
+
         if (idx + 1 >= cache.Count) return default; // 推完了
 
         // find next
         var matched = cache[idx + 1].TrackCode == trackCode
             ? cache[idx + 1]
-            :
             // find swap
-            cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode);
+            : cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode);
 
         //没推完，可能不符合推送顺序
-        return matched != null && matched.TrackCode == trackCode
+        return matched != null && matched.TrackCode == trackCode ? matched : default;
-            ? (matched, matched.TrackCode)
-            : (default, matched?.TrackCode ?? cache[idx + 1].TrackCode);
     }
 
     /// <summary>
@@ -501,10 +490,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     }
 
     //轨迹推送失败(被拦截)
-    private void TrackPushFailedPublicTopic(
+    private void TrackPushFailedPublicTopic(string failedMessage, TrackPushEvent @event, OrderEntity order)
-        string failedMessage,
-        TrackPushEvent @event,
-        OrderEntity order)
     {
         _broadcastingService.TrackPushFailedPubilcTopic(new TrackPushRecordDto
         {

src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs

@@ -144,6 +144,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
             return false;
         }
 
+        // todo: 在这里，判断存在性
+
         return true;
     }
 

src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs

@@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Http;
 using SJZY.Track.Application.Contract.Dtos.ProvideServices;
 using SJZY.Track.Application.Contract.Events;
 using SJZY.Track.Application.Contract.Services.ProvideServices;
+using SJZY.Track.Application.Helpers;
 using SJZY.Track.Domain.Aggregates.OrderAggregate;
 using SJZY.Track.Domain.Aggregates.TrackContentAggregate;
 using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
@@ -55,6 +56,7 @@ public partial class TrackService : ApplicationService, ITrackService
     [HttpPost("Track/Upsert")]
     public async Task<UnifyResultDto> UpsertTrackAsync(UpsertTrackInput input)
     {
+        // todo: 代码太长了
         var order = await _orderRepository.GetFirstAsync(x =>
                 x.SubSystemId == input.SubSystemId && x.OrderType == input.OrderType && x.OrderNo == input.OrderNo,
             x => new
@@ -119,6 +121,15 @@ public partial class TrackService : ApplicationService, ITrackService
             .Where(x => x.TrackTime == input.TrackTime)
             .ToList();
 
+        //根据比较字段选出所需的轨迹
+        if (!string.IsNullOrWhiteSpace(input.CompareFields))
+        {
+            var a = input.TrackCode + TextJsonHelper.ExtractValue(input.Extra, input.CompareFields);
+            tracks = tracks.Where(x => string.Equals(a,
+                x.TrackCode + TextJsonHelper.ExtractValue(x.Extra, input.CompareFields!))).ToList();
+        }
+
+        // todo：es落盘延迟是不确定的
         //当为更新操作时发现更新的数目为0进行重试操作
         if (!input.IsAdd && tracks.Count == 0)
         {
@@ -133,6 +144,7 @@ public partial class TrackService : ApplicationService, ITrackService
                         .ToList();
                     if (tracks.Count > 0)
                         success = true;
+                    // todo: 最大重试次数应该是可配置的，当出现的时候应该有告警，而不是直接给调用方，指数退避 2^N
                     await Task.Delay(300);
                 }
                 finally

src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderOperationRecordEsRepository.cs

@@ -16,6 +16,14 @@ public interface IOrderOperationRecordEsRepository
     /// <returns></returns>
     Task AddManyAsync(List<OrderOperationRecordIndex> indexes);
 
+    /// <summary>
+    /// 根据id删除订单操作记录
+    /// </summary>
+    /// <param name="id"></param>
+    /// <param name="indexName"></param>
+    /// <returns></returns>
+    Task DeleteOperationRecordByIdAsync(string id, string indexName);
+
     /// <summary>
     /// 获取订单操作记录
     /// </summary>

src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordEsRepository.cs

@@ -11,6 +11,14 @@ public interface ITrackPushRecordEsRepository
 
     Task AddManyAsync(List<TrackPushRecordIndex> index);
 
+    /// <summary>
+    /// 根据id删除轨迹推送记录
+    /// </summary>
+    /// <param name="id"></param>
+    /// <param name="indexName"></param>
+    /// <returns></returns>
+    Task DeletePushRecordByIdAsync(string id, string indexName);
+
     /// <summary>
     /// 获取轨迹推送记录
     /// </summary>
@@ -22,9 +30,9 @@ public interface ITrackPushRecordEsRepository
     /// 获取最新轨迹推送记录
     /// </summary>
     /// <param name="orderId"></param>
+    /// <param name="isSuccess">只选择成功的记录</param>
     /// <returns></returns>
-    Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId);
+    Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false);
-
 
     /// <summary>
     /// 获取历史中推送成功的记录

src/SJZY.Track.Repository/Repositories/OrderOperationRecordEsRepository.cs

@@ -47,6 +47,18 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract<Ord
         }
     }
 
+    public async Task DeleteOperationRecordByIdAsync(string id, string indexName)
+    {
+        var r = await Context.DeleteAsync<OrderOperationRecordIndex>(id, u => u
+            .Index(indexName)
+        );
+
+        if (!r.IsValid)
+        {
+            throw new InvalidOperationException($"轨迹操作记录失败，id：{indexName}，原因：{r.DebugInformation}。");
+        }
+    }
+
     public async Task<List<OrderOperationRecordIndex>> GetOrderOperationsByIdAsync(int orderId)
     {
         var data = await Context.SearchAsync<OrderOperationRecordIndex>(x => x
@@ -57,6 +69,6 @@ public class OrderOperationRecordEsRepository : ElasticSearchContextAbstract<Ord
             .Size(GlobalConst.NormalLimit)
             .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending)));
 
-        return data.Documents.ToList();
+        return data.ToIdList();
     }
 }

src/SJZY.Track.Repository/Repositories/TrackPushRecordEsRepository.cs

@@ -45,6 +45,18 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
         }
     }
 
+    public async Task DeletePushRecordByIdAsync(string id, string indexName)
+    {
+        var r = await Context.DeleteAsync<TrackPushRecordIndex>(id, u => u
+            .Index(indexName)
+        );
+
+        if (!r.IsValid)
+        {
+            throw new InvalidOperationException($"轨迹推送记录失败，id：{indexName}，原因：{r.DebugInformation}。");
+        }
+    }
+
 
     public async Task<List<TrackPushRecordIndex>> GetTrackPushRecordsAsync(int orderId)
     {
@@ -56,15 +68,24 @@ public class TrackPushRecordEsRepository : ElasticSearchContextAbstract<TrackPus
             .Size(1000)
             .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending)));
 
-        return data.Documents.ToList();
+        return data.ToIdList();
     }
 
-    public async Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId)
+    public async Task<TrackPushRecordIndex?> GetLatestTrackPushRecordAsync(int orderId, bool isSuccess = false)
     {
+        var musts = new List<Func<QueryContainerDescriptor<TrackPushRecordIndex>, QueryContainer>>
+        {
+            mu => mu.Term(t => t.OrderId, orderId),
+        };
+
+        if (isSuccess)
+            musts.Add(mu => mu.Term(t => t.IsSuccess, true));
+
         var data = await Context.SearchAsync<TrackPushRecordIndex>(x => x
             .Index(GlobalConst.TrackPushRecordIndex)
             .Query(q =>
-                q.Term(t => t.OrderId, orderId)
+                q.Bool(b =>
+                    b.Must(musts))
             )
             .Size(1000)
             .Sort(s => s.Field(f => f.CreateTime, SortOrder.Descending)));