update:调整轨迹推送逻辑

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)]
@@ -206,11 +211,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
             }
 
             // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行，为了约束顺序
-            config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
+            // 如果当前推送code与订单上的最后的code不一致，则考虑推送映射顺序
-            if (config == null)
+            if (!string.Equals(order.LatestPushTrackCode, @event.TrackCode))
             {
-                TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order);
+                config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event);
-                continue;
+                if (config == null)
+                {
+                    TrackPushFailedPublicTopic("推送顺序异常，找不到有效推送配置", @event, order);
+                    continue;
+                }
             }
 
             // 是否自动推送
@@ -362,38 +371,67 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
         if (config.IsAllPush)
         {
-            var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId);
+            var history = await _trackPushRecordEsRepository.GetLatestTrackPushRecordAsync(log.OrderId, true);
             if (history != default)
             {
                 body = TextJsonHelper.Deserialize<TrackPushInput>(history.PushContent);
-                if (config.IsAllPushWithLatest)
-                {
-                    //同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)
+                //在推送历史中，选出当前轨迹明细一致的轨迹
+                var tracks = await _trackDetailsEsRepository.GetListAsync(log.OrderId);
+                //确定轨迹标识符
+                var trackIdentifies = tracks.Select(x =>
                     {
-                        body.Data.Remove(old);
+                        //寻找映射配置(转换后的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();
+                    var timeStr = ((JsonElement) x["date"]!).GetString();
 
-                // //添加当前要推送的轨迹
+                    if (string.IsNullOrWhiteSpace(compareFields))
-                body!.Data.Add(cur);
+                        return trackIdentifies.Contains($"{code}{timeStr}");
+
+                    if (((JsonElement) x["extra"]!).TryGetProperty(compareFields, out var comparerValueJsonElement))
+                        return trackIdentifies.Contains($"{code}{TextJsonHelper.GetString(comparerValueJsonElement)}");
+
+                    return trackIdentifies.Contains($"{code}{null}");
+                }).ToList();
+
+                // if (config.IsAllPushWithLatest)
+                // {
+                //     //同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.Add(cur);
             }
         }
 

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;
@@ -120,6 +121,14 @@ 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)

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)));