459 lines
20 KiB
C#
459 lines
20 KiB
C#
using DotNetCore.CAP;
|
||
using NetTaste;
|
||
using SJZY.Expand.ABP.Core.Logger;
|
||
using SJZY.Expand.ABP.Core.PublicMethords;
|
||
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.Dtos.TrackContents;
|
||
using SJZY.Track.Domain.Enums;
|
||
using SJZY.Track.Domain.Queries;
|
||
using Volo.Abp.DependencyInjection;
|
||
using OrderIdentifier = SJZY.Track.Application.Contract.Events.OrderIdentifier;
|
||
|
||
namespace SJZY.Track.Application.CAP.EventHandlers;
|
||
|
||
public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
||
{
|
||
private readonly ICapPublisher _capPublisher;
|
||
private readonly IOrderRepository _orderRepository;
|
||
private readonly IProvideQueries _provideQueries;
|
||
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
|
||
private readonly IBroadcastingService _broadcastingService;
|
||
private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
|
||
|
||
public TrackUpsertEventHandler(
|
||
IOrderRepository orderRepository,
|
||
ICapPublisher capPublisher,
|
||
IProvideQueries provideQueries,
|
||
ITrackDetailsEsRepository trackDetailsEsRepository,
|
||
IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
|
||
IBroadcastingService broadcastingService)
|
||
{
|
||
_orderRepository = orderRepository;
|
||
_capPublisher = capPublisher;
|
||
_provideQueries = provideQueries;
|
||
_trackDetailsEsRepository = trackDetailsEsRepository;
|
||
_orderOperationRecordEsRepository = orderOperationRecordEsRepository;
|
||
_broadcastingService = broadcastingService;
|
||
}
|
||
|
||
[CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
|
||
[CapSubscribe(GlobalConst.TrackUpsertForRefreshSubscribe, Group = GlobalConst.TrackUpsertForRefreshSubscribe)]
|
||
public async Task ConsumerAsync(TrackUpsertEvent @event)
|
||
{
|
||
if (!CanDelete(@event)) return;
|
||
|
||
var canNextOrderIds = new List<int>();
|
||
|
||
var content = await HandleTrackAsync(@event, canNextOrderIds);
|
||
|
||
// 处理广播事件
|
||
if (content != null && (content.IsToMq || @event.IsToMq))
|
||
_broadcastingService.TrackChangePublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode);
|
||
|
||
if (canNextOrderIds.Count == 0) return;
|
||
// 根据订单关联关系异步向下刷（异步）
|
||
await DownByOrderLink(@event, canNextOrderIds);
|
||
}
|
||
|
||
private async Task DownByOrderLink(TrackUpsertEvent @event, List<int> canNextOrderIds)
|
||
{
|
||
// 根据订单关联关系异步向下刷（异步）
|
||
var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(canNextOrderIds.ToArray());
|
||
foreach (var order in @event.OrderIdentifiers)
|
||
{
|
||
// 理论上小包在这不会分组
|
||
foreach (var group in linkedOrders.Where(x => x.OwnerOrderId == order.OrderId)
|
||
.GroupBy(x => new
|
||
{
|
||
x.SubSystemId,
|
||
x.CustomId // 这里按客户分组是为了后续的推送方便
|
||
}))
|
||
{
|
||
var nextEvent = @event.DeepCopy();
|
||
nextEvent.Visited.AddRange(canNextOrderIds);
|
||
nextEvent.SubSystemId = group.Key.SubSystemId;
|
||
nextEvent.CustomId = group.Key.CustomId;
|
||
nextEvent.TrackRefreshData = new TrackRefreshStruct
|
||
{
|
||
ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode,
|
||
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent,
|
||
OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash,
|
||
};
|
||
|
||
// 不同子系统需要做内容映射
|
||
if (@event.SubSystemId != nextEvent.SubSystemId)
|
||
{
|
||
var mapCode = (await _provideQueries
|
||
.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode))
|
||
.FirstOrDefault();
|
||
|
||
if (mapCode == default)
|
||
{
|
||
// 跨子系统没做映射的轨迹不处理
|
||
continue;
|
||
}
|
||
|
||
nextEvent.TrackCode = mapCode;
|
||
}
|
||
|
||
var identifiers = group.Select(x => new OrderIdentifier
|
||
{
|
||
OrderId = x.OrderId,
|
||
OrderNo = x.OrderNo,
|
||
OrderType = x.OrderType,
|
||
}).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);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private bool CanDelete(TrackUpsertEvent @event)
|
||
{
|
||
// 系统创建的轨迹不允许修改
|
||
if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false })
|
||
{
|
||
LogHelper.Warn($"系统创建的轨迹不允许被修改：{@event}", true);
|
||
return false;
|
||
}
|
||
|
||
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
|
||
|
||
var dup = @event.Visited.Intersect(orderIds).ToArray();
|
||
if (dup.Any())
|
||
{
|
||
LogHelper.Warn($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true);
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
private async Task<TrackContentVO?> HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
|
||
{
|
||
var trackContent = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
|
||
|
||
if (trackContent == null)
|
||
{
|
||
// 找不到轨迹内容，直接结束
|
||
LogHelper.Warn($"找不到轨迹内容：{@event}", true);
|
||
return default;
|
||
}
|
||
|
||
//#7001:未来潜在需求，当轨迹内容匹配预设，记录一个标记，保证下刷过程都是使用预设内容
|
||
//虽然在引发事件之前做了判断，引发之后并不一定立即处理，在此过程中轨迹内容仍然可能被修改，所以要二次进行判断。
|
||
//通过来源指示是否为手动操作
|
||
var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial;
|
||
//轨迹内容标识不允许手动创建的
|
||
if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false) return trackContent;
|
||
//轨迹内容标识不允许手动更新的
|
||
if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false) return trackContent;
|
||
|
||
var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent);
|
||
|
||
if (canNextOrderIds.Count == 0) return trackContent;
|
||
|
||
if (insertTracks.Any())
|
||
{
|
||
// 批量add to es
|
||
await _trackDetailsEsRepository.AddManyAsync(insertTracks);
|
||
|
||
// 添加轨迹触发订单拉取状态更新
|
||
await _orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(),
|
||
@event.TrackCode);
|
||
|
||
// 添加轨迹触发预警计算（异步）
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackWarningCalcSubscribe, new TrackWarningCalcEvent
|
||
{
|
||
SubSystemId = @event.SubSystemId,
|
||
CustomId = @event.CustomId,
|
||
OrderIds = insertTracks.Select(x => x.OrderId).ToArray(),
|
||
TrackStatusId = trackContent.TrackStatusId,
|
||
TrackTime = @event.TrackTime,
|
||
});
|
||
}
|
||
|
||
// 触发订单主表更新事件（异步）
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
|
||
new TrackChangeAfterUpdateOrderEvent
|
||
{
|
||
OrderIds = canNextOrderIds.ToArray(),
|
||
IsCompleted = trackContent.IsCompleted,
|
||
TrackContent = trackContent.TrackContent,
|
||
TrackEnContent = trackContent.TrackEnContent,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
DestinationTimeZone = @event.DestinationTimeZone,
|
||
EstimatedDeliveryDate = @event.EstimatedDeliveryDate
|
||
});
|
||
|
||
// 触发轨迹推送事件（异步）
|
||
var trackPushEvent = new TrackPushEvent
|
||
{
|
||
OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改,
|
||
SubSystemId = @event.SubSystemId,
|
||
SubSystemCode = trackContent.SubSystemCode,
|
||
CustomId = @event.CustomId,
|
||
OrderIds = canNextOrderIds.ToArray(),
|
||
TrackCode = @event.TrackCode,
|
||
SystemPushInfo = new SystemPushStruct
|
||
{
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackContent = trackContent.TrackContent,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
CompareFields = @event.CompareFields,
|
||
Extra = @event.Extra
|
||
},
|
||
ForcedPush = @event.ForcedPush,
|
||
IsToMq = (trackContent.IsToMq || @event.IsToMq)
|
||
};
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
|
||
|
||
return trackContent;
|
||
}
|
||
|
||
private async Task<List<TrackDetailsIndex>> UpsertTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds,
|
||
TrackContentVO trackContent)
|
||
{
|
||
var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
|
||
|
||
// todo：条件命名写反了
|
||
// important：轨迹code相同，时间、发生地不同的，可能存在多条！！
|
||
var tracks = await _trackDetailsEsRepository.GetListAsync(orderIds, @event.TrackCode, true);
|
||
|
||
var insertTracks = new List<TrackDetailsIndex>();
|
||
var updateTracks = new List<TrackDetailsIndex>();
|
||
|
||
foreach (var order in @event.OrderIdentifiers)
|
||
{
|
||
// fast path: by update with es id case
|
||
if (!@event.IsAdd && !string.IsNullOrWhiteSpace(@event.EsId))
|
||
{
|
||
var trackById = tracks.FirstOrDefault(x => x.Id == @event.EsId);
|
||
if (trackById != null)
|
||
{
|
||
updateTracks.Add(BuiltUpdateTrack(@event, trackById, order, trackContent));
|
||
}
|
||
|
||
continue;
|
||
}
|
||
|
||
var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList();
|
||
|
||
if (@event.CompareFields.IsNullOrEmpty())
|
||
{
|
||
existTracks = existTracks.Where(x => x.TrackTime == @event.TrackTime).ToList();
|
||
}
|
||
else
|
||
{
|
||
var extraDict = GetExtraDictForTextJson(@event.Extra);
|
||
if (@event.Extra == null ||
|
||
extraDict == null
|
||
|| extraDict.TryGetValue(@event.CompareFields, out var compareValue) == false)
|
||
{
|
||
throw new BusinessException(message: "轨迹比较字段不存在，请检查拓展字段");
|
||
}
|
||
|
||
existTracks = existTracks.Where(x =>
|
||
{
|
||
var oldExtraDict = GetExtraDictForTextJson(x.Extra);
|
||
// tips：注意兼容性，历史轨迹的字段可能为空
|
||
if (x.Extra == null
|
||
|| oldExtraDict == null
|
||
|| !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue))
|
||
{
|
||
return false;
|
||
}
|
||
|
||
return oldValue == compareValue;
|
||
}).ToList();
|
||
}
|
||
|
||
var track = existTracks.FirstOrDefault();
|
||
|
||
if (!@event.IsAdd)
|
||
{
|
||
if (existTracks.Count > 1)
|
||
{
|
||
// 修改轨迹时，订单存在多条轨迹不允许修改
|
||
continue;
|
||
}
|
||
|
||
if (track == null) continue; // 不存在不让改
|
||
|
||
updateTracks.Add(BuiltUpdateTrack(@event, track, order, trackContent));
|
||
}
|
||
else
|
||
{
|
||
// 【批量新增】所有可能遇到的场景
|
||
// 标识：允许重复，不允许重复，零条，一条，多条；
|
||
// 场景A：允许重复，零条 => 新增
|
||
// 场景B：允许重复，一条 => 新增（约束受限新增，否则跳过新增）
|
||
// 场景C：允许重复，多条 => 新增（约束受限新增，否则跳过新增）
|
||
// 场景D：不允许重复，零条 => 新增
|
||
// 场景E：不允许重复，一条 => 跳过
|
||
// 场景F：不允许重复，（理想不会出现多条） => 跳过
|
||
|
||
// 不允许重复轨迹
|
||
if (!trackContent.IsRepeatable && existTracks.Count > 0) continue;
|
||
|
||
insertTracks.Add(BuiltNewTrack(@event, order, trackContent));
|
||
}
|
||
|
||
canNextOrderIds.Add(order.OrderId);
|
||
}
|
||
|
||
if (updateTracks.Any())
|
||
{
|
||
//批量差异化更新轨迹
|
||
await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks);
|
||
}
|
||
|
||
return insertTracks;
|
||
}
|
||
|
||
|
||
private TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order,
|
||
TrackContentVO trackContent)
|
||
{
|
||
//外部轨迹，轨迹发生地，轨迹发生时间，轨迹发生时区，不允许更改
|
||
if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide)
|
||
{
|
||
track.TrackPlace = @event.TrackPlace;
|
||
track.TrackTime = @event.NewTrackTime ?? @event.TrackTime;
|
||
track.TrackTimeZone = @event.TrackTimeZone;
|
||
}
|
||
|
||
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
|
||
? track.TrackTime
|
||
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
|
||
track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser;
|
||
track.UpdateTime = DateTime.Now;
|
||
|
||
//允许二次修改，在修改了轨迹内容后进行修改轨迹操作，如果输入的轨迹内容则以输入为准
|
||
//跨子系统?
|
||
track.TrackContent = @event.Content ?? trackContent.TrackContent;
|
||
track.TrackEnContent = @event.EnStatus ?? trackContent.TrackEnContent;
|
||
//S06新增
|
||
track.IsVisibility = @event.IsVisibility;
|
||
track.Remarks = @event.Remarks ?? GlobalConst.EmptyString;
|
||
track.Files = @event.Files?.Select(x => new FileItem
|
||
{
|
||
OssKey = x.OssKey,
|
||
FileName = x.FileName
|
||
}).ToList();
|
||
track.TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId;
|
||
track.TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId;
|
||
track.TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString;
|
||
track.TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId;
|
||
track.TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString;
|
||
track.DestinationTimeZone = @event.DestinationTimeZone;
|
||
track.EstimatedDeliveryDate = @event.EstimatedDeliveryDate;
|
||
track.Extra = @event.Extra;
|
||
|
||
// 写入操作日志
|
||
_ = _orderOperationRecordEsRepository.AddAsync(new OrderOperationRecordIndex
|
||
{
|
||
OrderId = order.OrderId,
|
||
OrderNo = order.OrderNo,
|
||
OptionType = OrderOperationType.修改轨迹.ToString(),
|
||
Content = track.ToJson(),
|
||
CreateUserId = @event.OperationManId,
|
||
CreateUserName = @event.OperationMan ?? "",
|
||
CreateTime = DateTime.Now
|
||
});
|
||
|
||
return track;
|
||
}
|
||
|
||
private TrackDetailsIndex BuiltNewTrack(TrackUpsertEvent @event, OrderIdentifier order, TrackContentVO trackContent)
|
||
{
|
||
var track = new TrackDetailsIndex
|
||
{
|
||
SubSystemId = @event.SubSystemId,
|
||
OrderId = order.OrderId,
|
||
OrderNo = order.OrderNo,
|
||
OrderType = order.OrderType,
|
||
TrackCode = @event.TrackCode,
|
||
TrackPlace = @event.TrackPlace,
|
||
TrackTime = @event.NewTrackTime ?? @event.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone,
|
||
ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString,
|
||
OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString,
|
||
OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString,
|
||
CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
|
||
CreateTime = DateTime.Now,
|
||
//来源轨迹内容,第一次新增的时候使用轨迹内容表提供的
|
||
TrackStatusId = trackContent.TrackStatusId,
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackStatusCode = trackContent.TrackStatusCode,
|
||
//跨子系统?
|
||
TrackContentId = trackContent.TrackContentId,
|
||
TrackContent = @event.Content ?? trackContent.TrackContent,
|
||
TrackEnContent = @event.EnStatus ?? trackContent.TrackEnContent,
|
||
//S06新增轨迹明细字段
|
||
IsVisibility = @event.IsVisibility,
|
||
Source = @event.Source,
|
||
Remarks = @event.Remarks ?? GlobalConst.EmptyString,
|
||
Files = @event.Files?.Select(x => new FileItem
|
||
{
|
||
OssKey = x.OssKey,
|
||
FileName = x.FileName
|
||
}).ToList(),
|
||
TrackPlaceCity = @event.TrackPlaceCity ?? GlobalConst.EmptyString,
|
||
TrackPlaceCityId = @event.TrackPlaceCityId ?? GlobalConst.NotExistId,
|
||
TrackPlaceProvince = @event.TrackPlaceProvince ?? GlobalConst.EmptyString,
|
||
TrackPlaceProvinceId = @event.TrackPlaceProvinceId ?? GlobalConst.NotExistId,
|
||
TrackPlaceNation = @event.TrackPlaceNation ?? GlobalConst.EmptyString,
|
||
TrackPlaceNationId = @event.TrackPlaceNationId ?? GlobalConst.NotExistId,
|
||
TrackPlacePostcode = @event.TrackPlacePostcode ?? GlobalConst.EmptyString,
|
||
DestinationTimeZone = @event.DestinationTimeZone,
|
||
EstimatedDeliveryDate = @event.EstimatedDeliveryDate,
|
||
Extra = @event.Extra
|
||
};
|
||
|
||
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
|
||
? track.TrackTime
|
||
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
|
||
|
||
// 写入操作日志
|
||
_ = _orderOperationRecordEsRepository.AddAsync(new OrderOperationRecordIndex
|
||
{
|
||
OrderId = order.OrderId,
|
||
OrderNo = order.OrderNo,
|
||
OptionType = OrderOperationType.创建轨迹.ToString(),
|
||
Content = track.ToJson(),
|
||
CreateUserId = @event.OperationManId,
|
||
CreateUserName = @event.OperationMan ?? "",
|
||
CreateTime = DateTime.Now
|
||
});
|
||
|
||
return track;
|
||
}
|
||
|
||
private Dictionary<string, object>? GetExtraDictForTextJson(string? extraStr)
|
||
{
|
||
return string.IsNullOrWhiteSpace(extraStr)
|
||
? null
|
||
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object>>(extraStr);
|
||
}
|
||
}
|