diff --git a/.idea/.idea.SJZY.Track.Api/.idea/.gitignore b/.idea/.idea.SJZY.Track.Api/.idea/.gitignore new file mode 100644 index 00000000..d6752705 --- /dev/null +++ b/.idea/.idea.SJZY.Track.Api/.idea/.gitignore @@ -0,0 +1,13 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Rider ignored files +/projectSettingsUpdater.xml +/.idea.SJZY.Track.Api.iml +/modules.xml +/contentModel.xml +# Editor-based HTTP Client requests +/httpRequests/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/src/SJZY.Track.Application.Contract/Dtos/TrackFlags/GetTrackFlagListDto.cs b/src/SJZY.Track.Application.Contract/Dtos/TrackFlags/GetTrackFlagListDto.cs index 3964caea..505536e1 100644 --- a/src/SJZY.Track.Application.Contract/Dtos/TrackFlags/GetTrackFlagListDto.cs +++ b/src/SJZY.Track.Application.Contract/Dtos/TrackFlags/GetTrackFlagListDto.cs @@ -58,11 +58,11 @@ /// /// 选中的轨迹标记id /// - public List? SelectIds { get; set; } + public List? SelectIds { get; set; } - ///// - ///// 关键字 - ///// - //public string Keywords { get; set; } + /// + /// 关键字 + /// + public string? Keywords { get; set; } } } \ No newline at end of file diff --git a/src/SJZY.Track.Application.Contract/Dtos/TrackPush/TrackPushInput.cs b/src/SJZY.Track.Application.Contract/Dtos/TrackPush/TrackPushInput.cs index 7875d4da..bc1d10bc 100644 --- a/src/SJZY.Track.Application.Contract/Dtos/TrackPush/TrackPushInput.cs +++ b/src/SJZY.Track.Application.Contract/Dtos/TrackPush/TrackPushInput.cs @@ -5,6 +5,17 @@ public class TrackPushInput public string OrderNo { get; set; } = null!; public List Data { get; set; } = new(); + + public string Log { get; set; } +} + +public class TrackUpdateInput +{ + public string OrderNo { get; set; } = null!; + + public List Data { get; set; } = new(); + + public string Log { get; set; } } public class TrackItem diff --git a/src/SJZY.Track.Application.Contract/Events/OrderIdentifier.cs b/src/SJZY.Track.Application.Contract/Events/OrderIdentifier.cs index 49e79b04..158e64b1 100644 --- a/src/SJZY.Track.Application.Contract/Events/OrderIdentifier.cs +++ b/src/SJZY.Track.Application.Contract/Events/OrderIdentifier.cs @@ -5,4 +5,6 @@ public struct OrderIdentifier public int OrderId { get; set; } public string OrderNo { get; set; } public int OrderType { get; set; } + + public int CustomId { get; set; } } \ No newline at end of file diff --git a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs index 43156556..b2724e3d 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs @@ -2,45 +2,58 @@ public class TrackPushEvent { - /// - /// 重试次数 - /// - public int RetryNum { get; set; } - /// /// 操作类型 /// public OptionType OptionType { get; set; } + /// + /// 子系统id + /// + public int SubSystemId { get; set; } + + /// + /// 客户id + /// + public int CustomId { get; set; } + /// /// 订单id /// - public int OrderId { get; set; } - - /// - /// 订单号 - /// - public string OrderNo { get; set; } = null!; - - /// - /// 轨迹状态 - /// - public string TrackStatus { get; set; } = null!; + public int[] OrderIds { get; set; } = null!; /// /// 轨迹代码（我们系统的） /// public string TrackCode { get; set; } = null!; + /// + /// 操作人 + /// + public string? OperationMan { get; set; } + + /// + /// 系统推送信息 + /// + public SystemPushStruct SystemPushInfo { get; set; } +} + +public struct SystemPushStruct +{ + /// + /// 轨迹状态 + /// + public string TrackStatus { get; set; } + /// /// 轨迹内容 /// - public string TrackContent { get; set; } = null!; + public string TrackContent { get; set; } /// /// 轨迹发生地 /// - public string TrackPlace { get; set; } = null!; + public string TrackPlace { get; set; } /// /// 轨迹发生时间 @@ -51,9 +64,4 @@ public class TrackPushEvent /// 轨迹发生时区 /// public float TrackTimeZone { get; set; } - - /// - /// 操作人 - /// - public string? OperationMan { get; set; } } \ No newline at end of file diff --git a/src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs index 3b449bbf..905060ea 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackUpsertEvent.cs @@ -18,9 +18,9 @@ public class TrackUpsertEvent public string? OperationMan { get; set; } /// - /// true新增，false更新，null表示刷新 + /// true新增，false更新 /// - public bool? IsAdd { get; set; } + public bool IsAdd { get; set; } /// /// 轨迹内容代码 diff --git a/src/SJZY.Track.Application.Contract/Services/ITrackPushCacheService.cs b/src/SJZY.Track.Application.Contract/Services/ITrackPushCacheService.cs index 8928ae11..d243a7a0 100644 --- a/src/SJZY.Track.Application.Contract/Services/ITrackPushCacheService.cs +++ b/src/SJZY.Track.Application.Contract/Services/ITrackPushCacheService.cs @@ -19,26 +19,4 @@ public interface ITrackPushCacheService /// /// Task?> SetPushContentConfigsAsync(int subSystemId, int customId); - - /// - /// 获取服务商拉取配置 - /// - /// 子系统id - /// 服务商id - /// 轨迹代码 - /// - Task GetPushContentConfigAsync(int subSystemId, int customId, string trackConde); - - /// - /// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了 - /// 则查询有没有可交换的推送，没有就返回null - /// - /// - /// - /// - /// - /// - Task IsValidPushAsync(int subSystemId, int customId, - string? latestPushTrackCode, - string? trackCode); } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs index 94cbc37d..4c0014c6 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs @@ -57,12 +57,12 @@ public class CapFilter : SubscribeFilter /// public override Task OnSubscribeExecutedAsync(ExecutedContext context) { - context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); - context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); - context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); - context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); - - BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!); + // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); + // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); + // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); + // context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId); + // + // BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!); return Task.CompletedTask; } @@ -86,11 +86,11 @@ public class CapFilter : SubscribeFilter LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}"); - context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); - context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); - context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); - - BuildLog(receiveTime!, receiveName!, arguments, true, messageId!); + // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime); + // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName); + // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments); + // + // BuildLog(receiveTime!, receiveName!, arguments, true, messageId!); return Task.CompletedTask; } diff --git a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs index e9ed2dec..46b87f8b 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs @@ -32,7 +32,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe var dup = @event.Visited.Intersect(@event.OrderIds).ToArray(); if (dup.Any()) { - LogHelper.Error($"删除轨迹订单时，发现循环依赖：{dup.JoinAsString(",")}"); + LogHelper.Error($"删除轨迹时，发现循环依赖：{dup.JoinAsString(",")}"); return; } @@ -49,7 +49,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe nextEvent.Visited.AddRange(@event.OrderIds); nextEvent.TrackCodes = @event.TrackCodes; nextEvent.SubSystemId = group.Key; - nextEvent.OrderIds = group.Select(x => x.OrderId).ToArray(); // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) @@ -60,13 +59,25 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe if (mapCodes.Count == 0) { //throw new BusinessException(message: "订单跨子系统，未配置内容映射，请联系管理员"); - return; + continue; } nextEvent.TrackCodes = mapCodes; } - await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, nextEvent); + var identifiers = group.Select(x => x.OrderId).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.OrderIds = identifiers.GetRange(idx, cnt).ToArray(); + idx += GlobalConst.ChunkSize; + + await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, subEvent); + } } } } diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs new file mode 100644 index 00000000..cfa667a9 --- /dev/null +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs @@ -0,0 +1,85 @@ +using DotNetCore.CAP; +using Newtonsoft.Json; +using SJZY.InterfaceRelay.Application.Contracts; +using SJZY.Track.Domain.Aggregates.OrderAggregate; +using SJZY.Track.Domain.Aggregates.TrackPushAggregate; +using SJZY.Track.Domain.Consts; +using SJZY.Track.Domain.Enums; +using Volo.Abp.DependencyInjection; + +namespace SJZY.Track.Application.EventHandlers; + +public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscribe +{ + private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; + private readonly IOrderRepository _orderRepository; + + public TrackPushCompletedEventHandler(ITrackPushRecordsRepository trackPushRecordsRepository, + IOrderRepository orderRepository) + { + _trackPushRecordsRepository = trackPushRecordsRepository; + _orderRepository = orderRepository; + } + + [CapSubscribe(GlobalConst.TrackPushCompletedSubscribe, Group = GlobalConst.TrackPushSubscribe)] + public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) + { + var log = JsonConvert.DeserializeObject(@event.RequestParams); + if (log == null) throw new BusinessException(message: "轨迹推送返回结果解析异常"); + log.IsSuccess = @event.IsSuccess; + log.Message = @event.Message; + log.RequestId = @event.RequestId; + + // 往es 写推送日志 + await _trackPushRecordsRepository.AddAsync(log); + + // if (!log.IsSuccess) + // { + // // 重试过程中不会去更新订单主表 + // if (@event.RetryNum < 3) + // { + // @event.RetryNum++; + // await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, @event); + // return; + // } + // } + + // 同步去更新订单主表 + await TrackPushAfterUpdateOrderAsync(log); + } + + private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log) + { + var order = new OrderEntity(); + order.Id = log.OrderId; + /* + * 异常：最近一条推送是失败 + */ + if (!log.IsSuccess) + { + order.QuestionType = (int)TrackProblemType.Error; + order.LatestPushError = log.Message; + } + else + { + order.QuestionType = (int)TrackProblemType.None; + order.LatestPushError = null; + if (log.OptionType == OptionType.新增.ToString()) + { + // tips：只有新增会影响推送进度？ + order.LatestPushTrackCode = log.TrackCode; + order.LatestPushTime = DateTime.Now; + await _orderRepository.UpdateColumnAsync(order, x => new + { + x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError + }); + return; + } + } + + await _orderRepository.UpdateColumnAsync(order, x => new + { + x.QuestionType, x.LatestPushError + }); + } +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs index 55d0ed5e..4eb337ba 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs @@ -6,6 +6,7 @@ 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.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Enums; @@ -17,214 +18,176 @@ namespace SJZY.Track.Application.EventHandlers; public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { private readonly IHttpClientProxy _httpClientProxy; - private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; private readonly IOrderRepository _orderRepository; private readonly ITrackPushCacheService _trackPushCacheService; - private readonly ICapPublisher _capPublisher; + private readonly ITrackDetailsRepository _trackDetailsRepository; + private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; + private List _pushConfigs; + private List _orders; public TrackPushEventHandler(IHttpClientProxy httpClientProxy, - ITrackPushRecordsRepository trackPushRecordsRepository, - IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher) + IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, + ITrackDetailsRepository trackDetailsRepository, ITrackPushRecordsRepository trackPushRecordsRepository) { _httpClientProxy = httpClientProxy; - _trackPushRecordsRepository = trackPushRecordsRepository; _orderRepository = orderRepository; _trackPushCacheService = trackPushCacheService; - _capPublisher = capPublisher; + _trackDetailsRepository = trackDetailsRepository; + _trackPushRecordsRepository = trackPushRecordsRepository; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] public async Task ConsumerAsync(TrackPushEvent @event) { - var order = await _orderRepository.GetFirstAsync(x => - x.Id == @event.OrderId, x => new + var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId); + if (pushConfigs == null || pushConfigs.Count == 0) return; + _pushConfigs = pushConfigs; + + var errorType = (int)TrackProblemType.Error; + var orders = await _orderRepository.GetListAsync(x => + @event.OrderIds.Contains(x.Id) && x.QuestionType != errorType, x => new OrderEntity { - x.SubSystemId, x.LatestPushTrackCode, x.QuestionType, x.CustomId, x.OrderType + Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode, + OrderType = x.OrderType }); - if (order == null) return; - - var log = new TrackPushRecordIndex - { - IndexName = $"{GlobalConst.TrackPushRecordsIndex}-{DateTime.Now:yyyy.MM.dd}", - OptionType = @event.OptionType.ToString(), - OrderId = @event.OrderId, - TrackStatus = @event.TrackStatus, - TrackCode = @event.TrackCode, - TrackContent = @event.TrackContent, - TrackPlace = @event.TrackPlace, - TrackTime = @event.TrackTime, - TrackTimeZone = @event.TrackTimeZone, - CreateUser = @event.OperationMan ?? GlobalConst.SystemUser, - CreateTime = DateTime.Now - }; - - TrackPushContentConfigDto? config = null; + if (orders.Count == 0) return; + _orders = orders; switch (@event.OptionType) { case OptionType.无: return; case OptionType.新增: - if (@event.OperationMan == null) - { - if (order.QuestionType != (int)TrackProblemType.Error) - { - // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行 - config = await _trackPushCacheService.IsValidPushAsync(order.SubSystemId, order.CustomId, - order.LatestPushTrackCode, @event.TrackCode); - if (config == null) return; - if (!config.IsAutoPush) return; - if (config.OrderType != order.OrderType) return; - } - } - else - { - // 手工推送单条不推断，直接无脑推 - config = await _trackPushCacheService.GetPushContentConfigAsync(order.SubSystemId, order.CustomId, - @event.TrackCode); - if (config is { IsAllPush: true }) - { - // 推断 本次推送 是否可行 - config = await _trackPushCacheService.IsValidPushAsync(order.SubSystemId, order.CustomId, - order.LatestPushTrackCode, @event.TrackCode); - } - } - - if (config == null) return; - await DoAddAsync(@event, config, log); + await DoPushAsync(@event); break; case OptionType.删除: break; case OptionType.修改: - config = await _trackPushCacheService.GetPushContentConfigAsync(order.SubSystemId, order.CustomId, - @event.TrackCode); - if (config == null || !config.IsCanUpdate) return; - await DoUpdateAsync(@event, config, log); + await DoUpdateAsync(@event); break; default: throw new ArgumentOutOfRangeException(); } - - if (!log.IsSuccess) - { - if (@event.RetryNum < 3) - { - @event.RetryNum++; - // 1, 2, 4min - await _capPublisher.PublishDelayAsync( - TimeSpan.FromSeconds((1 << @event.RetryNum - 1) * RandomHelper.GetRandom(30, 60)), - GlobalConst.TrackPushSubscribe, @event); - return; - } - } - - // 往es 写推送日志 - await _trackPushRecordsRepository.AddAsync(log); - - // 同步去更新订单主表 - await TrackPushAfterUpdateOrderAsync(new TrackPushAfterUpdateOrderDto - { - OptionType = @event.OptionType, - OrderId = @event.OrderId, - TrackCode = @event.TrackCode, - IsSuccess = log.IsSuccess, - Message = log.Message - }); } - private async Task TrackPushAfterUpdateOrderAsync(TrackPushAfterUpdateOrderDto dto) + private async Task DoPushAsync(TrackPushEvent @event) { - var order = new OrderEntity(); - order.Id = dto.OrderId; - /* - * 异常：最近一条推送是失败 - */ - if (!dto.IsSuccess) + var logs = new List(); + TrackPushContentConfigDto? config; + + if (@event.OperationMan == null) { - order.QuestionType = (int)TrackProblemType.Error; - order.LatestPushError = dto.Message; - } - else - { - order.QuestionType = (int)TrackProblemType.None; - order.LatestPushError = null; - if (dto.OptionType == OptionType.新增) + foreach (var order in _orders) { - // tips：只有新增会影响推送进度？ - order.LatestPushTrackCode = dto.TrackCode; - order.LatestPushTime = DateTime.Now; - await _orderRepository.UpdateColumnAsync(order, x => new + // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行 + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); + if (config == null) continue; + if (!config.IsAutoPush) continue; + if (config.OrderType != order.OrderType) continue; + + var log = new TrackPushRecordIndex { - x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError - }); - return; + 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, + }; + await BuiltPushItem(config, log); + logs.Add(log); } } - - await _orderRepository.UpdateColumnAsync(order, x => new + else //手工推送 { - x.QuestionType, x.LatestPushError - }); - } + config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + if (config == null) return; - private async Task DoAddAsync(TrackPushEvent @event, - TrackPushContentConfigDto config, TrackPushRecordIndex log) - { - var track = new TrackItem - { - // 推真实的发生地和发生时间（时间是计算后的） - Location = @event.TrackPlace, - Date = @event.TrackTime.AddHours( - config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone), - TimeZone = config.TrackTimeZone ?? @event.TrackTimeZone, - Description = config.TrackEnContentValue, - DescriptionCn = config.TrackContentValue, - Code = config.TrackContentCodeValue, - Status = config.TrackStatusValue, - Extra = config.ExpandFieldsValue - }; + var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds, @event.TrackCode); + tracks = tracks.DistinctBy(x => x.OrderId).ToList(); // 防止重复推送 - var body = new TrackPushInput - { - OrderNo = @event.OrderNo, - }; - body.Data.Add(track); + if (tracks.Count == 0) return; // 一个能推的轨迹都不存在 - if (config.IsAllPush) - { - var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(@event.OrderId); - if (history != default) + foreach (var track in tracks) { - body = JsonConvert.DeserializeObject(history.PushContent); - body!.Data.Add(track); + var order = _orders.First(x => x.Id == track.OrderId); + if (config is { IsAllPush: true }) + { + // 推断 本次推送 是否可行 + config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); + } + + if (config == null) continue; + + var log = new TrackPushRecordIndex + { + OptionType = @event.OptionType.ToString(), + OrderId = track.OrderId, + OrderNo = order.OrderNo, + TrackStatus = track.TrackStatus, + TrackCode = @event.TrackCode, + TrackContent = track.TrackContent, + TrackPlace = track.TrackPlace, + TrackTime = track.TrackTime, + TrackTimeZone = track.TrackTimeZone, + CreateUser = @event.OperationMan, + }; + await BuiltPushItem(config, log); + logs.Add(log); } } - var res = await _httpClientProxy.Service.ForwardWithSubscribeAsync(InterfaceRelayConst.PushTrackInterfaceCode, - config.ServiceProviderCode, body); - - log.PushContent = body.ToJson(); - - log.IsSuccess = res.Code == InterfaceRelayConst.SuccessCode; - log.RequestId = res.RequestId; - log.Message = res.Msg; + await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.PushTrackInterfaceCode, + _pushConfigs.First().ServiceProviderCode, logs, GlobalConst.TrackPushCompletedSubscribe); } - private async Task DoUpdateAsync(TrackPushEvent @event, - TrackPushContentConfigDto config, TrackPushRecordIndex log) + private async Task DoUpdateAsync(TrackPushEvent @event) + { + var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); + if (config == null || !config.IsCanUpdate) return; + + var logs = new List(); + foreach (var order in _orders) + { + 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, + }; + BuiltUpdateItem(config, log); + logs.Add(log); + } + + await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.TrackUpdateInterfaceCode, + _pushConfigs.First().ServiceProviderCode, logs, GlobalConst.TrackPushCompletedSubscribe); + } + + private void BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) { var data = new { - orderNo = @event.OrderNo, + orderNo = log.OrderNo, data = new[] { new { // 推真实的发生地和发生时间 - location = @event.TrackPlace, - date = @event.TrackTime.AddHours( - config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone), - timeZone = config.TrackTimeZone ?? @event.TrackTimeZone, + location = log.TrackPlace, + date = log.TrackTime.AddHours( + config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone), + timeZone = config.TrackTimeZone ?? log.TrackTimeZone, description = config.TrackEnContentValue, description_cn = config.TrackContentValue, code = config.TrackContentCodeValue, @@ -234,13 +197,66 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe } }; - - var res = await _httpClientProxy.Service.ForwardWithSubscribeAsync(InterfaceRelayConst.TrackUpdateInterfaceCode, - config.ServiceProviderCode, data); - log.PushContent = data.ToJson(); - log.IsSuccess = res.Code == InterfaceRelayConst.SuccessCode; - log.RequestId = res.RequestId; - log.Message = res.Msg; + } + + private async Task BuiltPushItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) + { + var track = new TrackItem + { + // 推真实的发生地和发生时间（时间是计算后的） + Location = log.TrackPlace, + Date = log.TrackTime.AddHours( + config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone), + TimeZone = config.TrackTimeZone ?? log.TrackTimeZone, + Description = config.TrackEnContentValue, + DescriptionCn = config.TrackContentValue, + Code = config.TrackContentCodeValue, + Status = config.TrackStatusValue, + Extra = config.ExpandFieldsValue + }; + + var body = new TrackPushInput + { + OrderNo = log.OrderNo, + }; + body.Data.Add(track); + + if (config.IsAllPush) + { + var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(log.OrderId); + if (history != default) + { + body = JsonConvert.DeserializeObject(history.PushContent); + body!.Data.Add(track); + } + } + + log.PushContent = body.ToJson(); + } + + /// + /// 根据轨迹代码推断下一个（升序）要推送的内容，如果找不到或者已经是最后一个了 + /// 则查询有没有可交换的推送，没有就返回null + /// + /// + /// + /// + /// + TrackPushContentConfigDto? GetValidPushConfigAsync(List cache, + string? latestPushTrackCode, string? trackCode) + { + var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); + + 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); + + return matched != null && matched.TrackCode == trackCode ? matched : default; } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs index 4c132e26..ea34f0e6 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs @@ -36,6 +36,7 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe [CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)] public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { + if (!@event.IsSuccess) return; var attach = Base64Helper.DeSerialize(@event.Attach); if (attach is null) return; @@ -222,15 +223,16 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe // 新增/更新轨迹 var trackUpsertEvent = new TrackUpsertEvent { - OrderIdentifiers = new[] - { + OrderIdentifiers = + [ new OrderIdentifier { OrderId = context.OrderId, OrderNo = context.OrderNo, - OrderType = context.OrderType + OrderType = context.OrderType, + CustomId = context.CustomId } - }, + ], TrackRefreshData = new TrackRefreshStruct { ServiceProviderName = context.ServiceProviderName, @@ -247,6 +249,8 @@ public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe TrackTimeZone = trackTimeZone ?? matchedContentConfig.TrackTimeZoneValue ?? 8 }; + trackUpsertEvent.IsAdd = + !await _trackDetailsRepository.AnyByCodeAsync(context.OrderId, matchedContentConfig.TrackContentCode); await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs index f7746db2..937c0a21 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs @@ -35,7 +35,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var dup = @event.Visited.Intersect(orderIds).ToArray(); if (dup.Any()) { - LogHelper.Error($"删除轨迹订单时，发现循环依赖：{dup.JoinAsString(",")}"); + LogHelper.Error($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}"); return; } @@ -48,17 +48,11 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { // 理论上小包在这不会分组 foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId) - .GroupBy(x => x.SubSystemId)) + .GroupBy(x => new { x.SubSystemId, x.CustomId })) { var nextEvent = @event.DeepCopy(); nextEvent.Visited.AddRange(orderIds); - nextEvent.SubSystemId = group.Key; - nextEvent.OrderIdentifiers = group.Select(x => new OrderIdentifier - { - OrderId = x.OrderId, - OrderNo = x.OrderNo, - OrderType = x.OrderType - }).ToArray(); + nextEvent.SubSystemId = group.Key.SubSystemId; nextEvent.TrackRefreshData = new TrackRefreshStruct { ServiceProviderName = @event.TrackRefreshData.ServiceProviderName, @@ -78,13 +72,31 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (mapCode == default) { // 跨子系统没做映射的轨迹不处理 - return; + continue; } nextEvent.TrackCode = mapCode; } - await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent); + var identifiers = group.Select(x => new OrderIdentifier + { + OrderId = x.OrderId, + OrderNo = x.OrderNo, + OrderType = x.OrderType, + CustomId = x.CustomId + }).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); + } } } } @@ -94,7 +106,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray(); var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode); - TrackDetailsIndex firstTrack = null; + TrackDetailsIndex? firstTrack = null; var insertTracks = new List(); @@ -102,11 +114,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { var track = tracks.FirstOrDefault(x => x.OrderId == order.OrderId); - if (@event.IsAdd != null) - { - if (track == null && !@event.IsAdd!.Value) continue; // 不存在不让改 - if (track != null && @event.IsAdd!.Value) continue; // 存在不让创建 - } + if (track == null && !@event.IsAdd) continue; // 不存在不让改 + if (track != null && @event.IsAdd) continue; // 存在不让创建 if (track == default) { @@ -156,22 +165,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe await _trackDetailsRepository.UpdateAsync(track); } - // step2、触发轨迹推送事件（异步） - var trackPushEvent = new TrackPushEvent - { - OptionType = track.Id.IsNullOrWhiteSpace() ? OptionType.新增 : OptionType.修改, - OrderId = order.OrderId, - OrderNo = order.OrderNo, - TrackStatus = track.TrackStatus, - TrackCode = track.TrackCode, - TrackContent = track.TrackContent, - TrackPlace = track.TrackPlace, - TrackTime = track.TrackTime, - TrackTimeZone = track.TrackTimeZone, - }; - - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); - canNextOrderIds.Add(order.OrderId); firstTrack ??= track; @@ -194,5 +187,27 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe TrackTime = @event.TrackTime, TrackTimeZone = @event.TrackTimeZone }); + + // 触发轨迹推送事件（异步） + foreach (var group in @event.OrderIdentifiers.GroupBy(x => x.CustomId)) + { + var trackPushEvent = new TrackPushEvent + { + OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改, + SubSystemId = @event.SubSystemId, + CustomId = group.Key, + OrderIds = group.Select(x => x.OrderId).ToArray(), + TrackCode = @event.TrackCode, + SystemPushInfo = new SystemPushStruct + { + TrackStatus = firstTrack.TrackStatus, + TrackContent = firstTrack.TrackContent, + TrackPlace = @event.TrackPlace, + TrackTime = @event.TrackTime, + TrackTimeZone = @event.TrackTimeZone + } + }; + await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + } } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index 99912fb3..fba5f4d2 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -12,7 +12,7 @@ - + diff --git a/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs b/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs index f37ff3b5..e9554da2 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/OrderService.Link.cs @@ -72,7 +72,7 @@ public partial class OrderService // 过滤掉不存在的订单 var canLinkOrders = await _orderRepository.GetListAsync(x => x.OrderType == input.LinkOrderType && noLinkOrderNos.Contains(x.OrderNo), - x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId }); + x => new { x.Id, x.OrderNo, x.OrderType, x.SubSystemId, x.CustomId }); if (canLinkOrders.Count == 0) return; @@ -100,7 +100,8 @@ public partial class OrderService { OrderId = x.Id, OrderNo = x.OrderNo, - OrderType = x.OrderType + OrderType = x.OrderType, + CustomId = x.CustomId }).ToArray(), TrackCode = track.TrackCode, TrackPlace = track.TrackPlace, diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs index 7cf315b2..d7f71038 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs @@ -65,7 +65,8 @@ public class TrackPullService : ApplicationService var attach = Base64Helper.SerializeObject(group.Key); // 批量获取轨迹 var _ = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( - "A012", serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, attach); + InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, GlobalConst.TrackRefreshSubscribe, + attach); } return true; @@ -88,32 +89,36 @@ public class TrackPullService : ApplicationService continue; } - var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId); - var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; - var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName; - - foreach (var group in orders.GroupBy(x => - new { x.SubSystemId, x.OrderType })) + var page = 1; + var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page); + while (orders.Any()) { - var data = group.Select(x => new - { - x.OrderNo - }).ToList(); + var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; + var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName; - var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, - serviceProviderName!, - group.Key.SubSystemId, group.Key.OrderType)); - // 批量获取轨迹 - await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( - InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, - GlobalConst.TrackRefreshSubscribe, attach); + foreach (var group in orders.GroupBy(x => + new { x.SubSystemId, x.OrderType })) + { + var data = group.Select(x => new + { + x.OrderNo + }).ToList(); + + var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId, + serviceProviderName!, + group.Key.SubSystemId, group.Key.OrderType)); + // 批量获取轨迹 + await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( + InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data, + GlobalConst.TrackRefreshSubscribe, attach); + } + + orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page); } - if (config.PullInterval < 60) config.PullInterval = 60; await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count); } - return true; } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs index 4a0ef260..8b43e04f 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs @@ -3,7 +3,6 @@ using Microsoft.AspNetCore.Authorization; using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Domain.Aggregates.OrderAggregate; -using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Consts; using SJZY.UserCenter.HttpApi.Services; @@ -17,17 +16,14 @@ namespace SJZY.Track.Application.Services.ProvideServices; [Route("/ProviderService")] public class TrackPushService : ApplicationService { - private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly IOrderRepository _orderRepository; private readonly ICapPublisher _capPublisher; private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; - public TrackPushService(ITrackDetailsRepository trackDetailsRepository - , IOrderRepository orderRepository + public TrackPushService(IOrderRepository orderRepository , ICapPublisher capPublisher , ITrackPushRecordsRepository trackPushRecordsRepository) { - _trackDetailsRepository = trackDetailsRepository; _orderRepository = orderRepository; _capPublisher = capPublisher; _trackPushRecordsRepository = trackPushRecordsRepository; @@ -51,38 +47,17 @@ public class TrackPushService : ApplicationService throw new BusinessException(message: "一个订单都不存在"); } - var tracks = await _trackDetailsRepository.GetListAsync(orderIds, input.TrackCode); - tracks = tracks.DistinctBy(x => x.OrderId).ToList(); - - if (tracks.Count == 0) + var trackPushEvent = new TrackPushEvent { - throw new BusinessException(message: "一个能推的轨迹都不存在"); - } + OptionType = OptionType.新增, + OrderIds = orderIds, + TrackCode = input.TrackCode, + OperationMan = CurrentUserRequest.CurrentUser.Name, + }; - var res = 0; + await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); - foreach (var track in tracks) - { - var trackPushEvent = new TrackPushEvent - { - OptionType = OptionType.新增, - OrderId = track.OrderId, - OrderNo = track.OrderNo, - TrackStatus = track.TrackStatus, - TrackCode = track.TrackCode, - TrackContent = track.TrackContent, - TrackPlace = track.TrackPlace, - TrackTime = track.TrackTime, - TrackTimeZone = track.TrackTimeZone, - OperationMan = CurrentUserRequest.CurrentUser.Name, - }; - - await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); - - res++; - } - - return $"即将推送轨迹代码{input.TrackCode}给{res}个订单"; + return $"即将推送轨迹代码{input.TrackCode}给{orderIds.Length}个订单"; } diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs index 1d20b169..2b820625 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs +++ b/src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs @@ -43,7 +43,7 @@ public partial class TrackService : ApplicationService, ITrackService var order = await _orderRepository.GetFirstAsync(x => x.OrderType == input.OrderType && x.OrderNo == input.OrderNo, x => new { - x.Id, x.SubSystemId, x.OrderNo, x.OrderType + x.Id, x.SubSystemId, x.OrderNo, x.OrderType, x.CustomId }); if (order == default) { @@ -53,15 +53,16 @@ public partial class TrackService : ApplicationService, ITrackService // 新增/更新轨迹 var @event = ObjectMapper.Map(input); @event.SubSystemId = order.SubSystemId; - @event.OrderIdentifiers = new[] - { + @event.OrderIdentifiers = + [ new OrderIdentifier { OrderId = order.Id, OrderNo = order.OrderNo, - OrderType = order.OrderType + OrderType = order.OrderType, + CustomId = order.CustomId } - }; + ]; @event.OperationMan = CurrentUserRequest.CurrentUser.Name; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event); return new UnifyResultDto(true, true); @@ -85,7 +86,7 @@ public partial class TrackService : ApplicationService, ITrackService var orders = await _orderRepository.GetListAsync(x => x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new { - x.Id, x.OrderNo, x.SubSystemId + x.Id, x.OrderNo, x.SubSystemId, x.CustomId }); foreach (var group in orders.GroupBy(x => x.SubSystemId)) @@ -98,7 +99,8 @@ public partial class TrackService : ApplicationService, ITrackService { OrderId = order.Id, OrderNo = order.OrderNo, - OrderType = input.OrderType + OrderType = input.OrderType, + CustomId = order.CustomId }).ToArray(), IsAdd = input.IsAdd, TrackCode = input.TrackCode, @@ -186,7 +188,7 @@ public partial class TrackService : ApplicationService, ITrackService var @event = new TrackDeleteEvent { OrderIds = group.Select(x => x.OrderId).ToArray(), - TrackCodes = codes.Select(x => x.TrackCode).ToList(), + TrackCodes = input.TrackCodes, SubSystemId = orders.First(x => x.Id == group.Key).SubSystemId, OperationMan = CurrentUserRequest.CurrentUser.Name }; diff --git a/src/SJZY.Track.Application/Services/TrackPull/TrackPullService.cs b/src/SJZY.Track.Application/Services/TrackPull/TrackPullService.cs index d23937e9..cd027698 100644 --- a/src/SJZY.Track.Application/Services/TrackPull/TrackPullService.cs +++ b/src/SJZY.Track.Application/Services/TrackPull/TrackPullService.cs @@ -86,6 +86,8 @@ public class TrackPullService : ApplicationService [HttpPost("Edit")] public async Task EditAsync([FromBody] EditTrackPullConfigInput input) { + if (input.PullInterval < 60) input.PullInterval = 60; + if (input.Options != OptionTypeNoNull.删除) { Expression> isAnyExp = x => diff --git a/src/SJZY.Track.Application/Services/TrackPush/TrackPushCacheService.cs b/src/SJZY.Track.Application/Services/TrackPush/TrackPushCacheService.cs index e5d413f4..dfda8279 100644 --- a/src/SJZY.Track.Application/Services/TrackPush/TrackPushCacheService.cs +++ b/src/SJZY.Track.Application/Services/TrackPush/TrackPushCacheService.cs @@ -43,34 +43,6 @@ public class TrackPushCacheService : ITrackPushCacheService, ITransientDependenc return data; } - public async Task GetPushContentConfigAsync(int subSystemId, int customId, - string trackConde) - { - var configs = await GetPushContentConfigsAsync(subSystemId, customId); - return configs?.FirstOrDefault(x => x.TrackCode == trackConde); - } - - public async Task IsValidPushAsync(int subSystemId, int customId, - string? latestPushTrackCode, - string? trackCode) - { - var cache = await GetPushContentConfigsAsync(subSystemId, customId); - if (cache == null || cache.Count < 1) return default; - - var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); - - 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); - - return matched != null && matched.TrackCode == trackCode ? matched : default; - } - private async Task?> DoGetPushContentConfigsAsync(int subSystemId, int customId) { var trackPushConfig = await _trackPushConfigRepository.GetFirstAsync( diff --git a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs index a61e4f95..0d06b2fd 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs @@ -13,6 +13,8 @@ public interface ITrackDetailsRepository Task AnyAsync(int orderId, string originTrackData); + Task AnyByCodeAsync(int orderId, string trackCode); + Task UpdateAsync(TrackDetailsIndex index); Task AddManyAsync(List index); diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordsRepository.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordsRepository.cs index 8783078d..99276b4b 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/ITrackPushRecordsRepository.cs @@ -9,6 +9,8 @@ public interface ITrackPushRecordsRepository /// Task AddAsync(TrackPushRecordIndex index); + Task AddManyAsync(List index); + /// /// 获取轨迹推送记录 /// diff --git a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs index 8f877463..956e241d 100644 --- a/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs +++ b/src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs @@ -17,6 +17,11 @@ public class TrackPushRecordIndex : IdBaseIndex /// public int OrderId { get; set; } + /// + /// 订单号 + /// + public string OrderNo { get; set; } = null!; + /// /// 接口通请求Id /// diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index b855eb70..93fd1c53 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -32,6 +32,11 @@ public partial class GlobalConst /// public const string TrackPushSubscribe = "SJZY.Track.TrackPush"; + /// + /// 轨迹推送完成（接口通回调） + /// + public const string TrackPushCompletedSubscribe = "SJZY.Track.TrackPushCompleted"; + /// /// 轨迹新增/更新/删除后触发更新订单主表 /// diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.cs index ec269ee2..22beb8e1 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.cs @@ -4,4 +4,5 @@ public partial class GlobalConst { public const string SystemUser = "系统"; public const string EmptyString = "--"; + public const int ChunkSize = 100; } \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Dtos/Order/LinkedOrderVO.cs b/src/SJZY.Track.Domain/Dtos/Order/LinkedOrderVO.cs index 5b81e796..23299e30 100644 --- a/src/SJZY.Track.Domain/Dtos/Order/LinkedOrderVO.cs +++ b/src/SJZY.Track.Domain/Dtos/Order/LinkedOrderVO.cs @@ -26,4 +26,9 @@ public class LinkedOrderVO /// 子系统id /// public int SubSystemId { get; set; } + + /// + /// 客户id + /// + public int CustomId { get; set; } } \ No newline at end of file diff --git a/src/SJZY.Track.Domain/Dtos/TrackFlags/GetPageTrackFlagVO.cs b/src/SJZY.Track.Domain/Dtos/TrackFlags/GetPageTrackFlagVO.cs index adbf1a5d..f838b37d 100644 --- a/src/SJZY.Track.Domain/Dtos/TrackFlags/GetPageTrackFlagVO.cs +++ b/src/SJZY.Track.Domain/Dtos/TrackFlags/GetPageTrackFlagVO.cs @@ -18,10 +18,10 @@ namespace SJZY.Track.Domain.Dtos.TrackFlags /// public List? SelectIds { get; set; } - ///// - ///// 关键字 - ///// - //public string Keywords { get; set; } + /// + /// 关键字 + /// + public string? Keywords { get; set; } } public class GetTrackFlagListVO diff --git a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs index 2795df48..7d043a9f 100644 --- a/src/SJZY.Track.Domain/Queries/IProvideQueries.cs +++ b/src/SJZY.Track.Domain/Queries/IProvideQueries.cs @@ -60,6 +60,9 @@ public interface IProvideQueries /// 获取服务商下所有单（未完成的） /// /// + /// + /// /// - Task> FindOrderWithServiceProviderAsync(int serviceProvider); + Task> FindOrderWithServiceProviderAsync(int serviceProvider, int pageIndex, + int pageSize = 1000); } \ No newline at end of file diff --git a/src/SJZY.Track.Domain/SJZY.Track.Domain.csproj b/src/SJZY.Track.Domain/SJZY.Track.Domain.csproj index cb0a715f..3c38d62a 100644 --- a/src/SJZY.Track.Domain/SJZY.Track.Domain.csproj +++ b/src/SJZY.Track.Domain/SJZY.Track.Domain.csproj @@ -8,7 +8,7 @@ - + diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 7f144688..82e4993f 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -113,7 +113,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, .ToListAsync(); } - public async Task> FindOrderWithServiceProviderAsync(int serviceProvider) + public async Task> FindOrderWithServiceProviderAsync(int serviceProvider, + int pageIndex, int pageSize = 1000) { return await Context.Queryable() .RightJoin((a, b) => a.Id == b.OrderId) @@ -128,6 +129,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, ServiceProviderName = c.Name, SubSystemId = a.SubSystemId }) + .Skip((pageIndex - 1) * pageSize) + .Take(pageSize) .ToListAsync(); } } \ No newline at end of file diff --git a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs index 93edf756..85faff5d 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs @@ -78,7 +78,8 @@ AND `a`.`IsDeleted` = 0 AND `b`.`IsDeleted` = 0;"); OrderId = b.Id, OrderNo = b.OrderNo, OrderType = b.OrderType, - SubSystemId = b.SubSystemId + SubSystemId = b.SubSystemId, + CustomId = b.CustomId }) .ToListAsync(); } diff --git a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs index 644beb65..af29d3fd 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs @@ -1,3 +1,4 @@ +using Elasticsearch.Net; using Nest; using SJZY.Expand.ElasticSearch.Services; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -28,6 +29,11 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(x => x + var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd .Field(f => f.TrackEnContent) .Field(f => f.TrackContent) @@ -74,7 +80,12 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract AnyAsync(int orderId, DateTime trackTime) @@ -87,6 +98,11 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 0; } @@ -100,6 +116,29 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract 0; + } + + public async Task AnyByCodeAsync(int orderId, string trackCode) + { + var r = await Context.SearchAsync(x => x + .Source(s => s.Includes(fd => fd.Field(f => f.TrackCode))) + .Query(q => + q.Term(f => f.OrderId, orderId) + && q.Term(f => f.TrackCode.Suffix("keyword"), trackCode) + ) + ); + + if (!r.IsValid) + { + throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。"); + } + return r.Hits.Count > 0; } @@ -107,9 +146,8 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(track.Id, u => u - .Index(track.IndexName) - .Doc(track) - //.Refresh(Refresh.True) // 可选：立即刷新索引使更改立即可见 + .Index(track.IndexName) + .Doc(track) ); if (!r.IsValid) @@ -150,7 +188,7 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract(); - var data = await Context.SearchAsync(x => x + var r = await Context.SearchAsync(x => x .Source(s => s.Includes(fd => fd .Field(f => f.OrderNo) .Field(f => f.TrackStatus) @@ -170,30 +208,41 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract> GetListAsync(int orderId) { - var data = await Context.SearchAsync(x => x + var r = await Context.SearchAsync(x => x .Query(q => q.Term(t => t.OrderId, orderId) ) .Size(1000) .Sort(s => s.Field(f => f.ZeroTrackTime, SortOrder.Descending))); - return data.Documents.ToList(); + if (!r.IsValid) + { + throw new Exception($"轨迹查询异常，原因：{r.DebugInformation}。"); + } + + return r.Documents.ToList(); } public async Task DeleteTrackAsync(int[] orderIds, List trackCodes) { var r = await Context.DeleteByQueryAsync(x => x - .Query( - q => - q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) - && q.Term(t => t.IsCanDelete, true) - && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) - //.Refresh(Refresh.True) // 可选：立即刷新索引使更改立即可见 + .Conflicts(Conflicts.Abort) + .Query( + q => + q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)) + && q.Term(t => t.IsCanDelete, true) + && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))) + .WaitForCompletion() ); if (!r.IsValid) diff --git a/src/SJZY.Track.Repository/Repositories/TrackFlagRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackFlagRepository.cs index d2227924..ed92d283 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackFlagRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackFlagRepository.cs @@ -46,8 +46,9 @@ public class TrackFlagRepository : BasicRepository, ITrackFlagR .LeftJoin((a, b, c) => a.CreateUserId == c.Id && c.IsDeleted == false) .LeftJoin((a, b, c, d) => a.UpdateUserId == d.Id && d.IsDeleted == false) .WhereIF(input.SubSystemId > 0, (a, b, c, d) => a.SubSystemId == input.SubSystemId) + .WhereIF(!input.Keywords.IsNullOrWhiteSpace(), (a, b, c, d) => SqlFunc.Contains(a.Code, input.Keywords)) .OrderByIF(input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => a.CreateTime, OrderByType.Desc) - .OrderByIF(!input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => SqlFunc.IIF(input.SelectIds.Contains(a.Id), SqlFunc.DateAdd(SqlFunc.GetDate(), 1), a.CreateTime), OrderByType.Desc) + .OrderByIF(!input.SelectIds.IsNullOrEmpty(), (a, b, c, d) => SqlFunc.IIF(input.SelectIds!.Contains(a.Id), SqlFunc.DateAdd(SqlFunc.GetDate(), 1), a.CreateTime), OrderByType.Desc) .Select((a, b, c, d) => new GetTrackFlagListVO { Id = a.Id, diff --git a/src/SJZY.Track.Repository/Repositories/TrackPushRecordsRepository.cs b/src/SJZY.Track.Repository/Repositories/TrackPushRecordsRepository.cs index a2745e96..760614e9 100644 --- a/src/SJZY.Track.Repository/Repositories/TrackPushRecordsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/TrackPushRecordsRepository.cs @@ -16,6 +16,8 @@ public class TrackPushRecordsRepository : ElasticSearchContextAbstract request .Index(index.IndexName) @@ -28,6 +30,22 @@ public class TrackPushRecordsRepository : ElasticSearchContextAbstract records) + { + if (!records.Any()) return; + // 按天写入es + var indexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}"; + records.ForEach(x => x.IndexName = indexName); + + var r = await Context.IndexManyAsync(records, indexName); + + if (!r.IsValid) + { + throw new Exception($"{records.Count}条轨迹推送记录写入失败，原因：{r.DebugInformation}。"); + } + } + + public async Task> GetTrackPushRecordsAsync(int orderId) { var data = await Context.SearchAsync(x => x