457 lines
20 KiB
C#
457 lines
20 KiB
C#
using System.Text;
|
||
using BSI.Stack.Rabbit;
|
||
using DotNetCore.CAP;
|
||
using SJZY.Expand.ABP.Core.Logger;
|
||
using SJZY.Expand.ABP.Core.PublicMethords;
|
||
using SJZY.Track.Application.Contract.Events;
|
||
using SJZY.Track.Domain.Aggregates.OrderAggregate;
|
||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||
using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
|
||
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 RabbitClientBus _rabbitClientBus;
|
||
private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
|
||
|
||
public TrackUpsertEventHandler(
|
||
IOrderRepository orderRepository,
|
||
ICapPublisher capPublisher,
|
||
IProvideQueries provideQueries,
|
||
ITrackDetailsEsRepository trackDetailsEsRepository,
|
||
RabbitClientBus rabbitClientBus,
|
||
IOrderOperationRecordEsRepository orderOperationRecordEsRepository)
|
||
{
|
||
_orderRepository = orderRepository;
|
||
_capPublisher = capPublisher;
|
||
_provideQueries = provideQueries;
|
||
_trackDetailsEsRepository = trackDetailsEsRepository;
|
||
_rabbitClientBus = rabbitClientBus;
|
||
_orderOperationRecordEsRepository = orderOperationRecordEsRepository;
|
||
}
|
||
|
||
[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);
|
||
|
||
// 处理广播事件
|
||
PublicTrackTopic(@event, content);
|
||
|
||
if (canNextOrderIds.Count == 0) return;
|
||
// 根据订单关联关系异步向下刷（异步）
|
||
await DownByOrderLink(@event, canNextOrderIds);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 发送轨迹广播
|
||
/// </summary>
|
||
/// <param name="eto"></param>
|
||
/// <param name="content"></param>
|
||
private void PublicTrackTopic(TrackUpsertEvent eto, TrackContentVO? content)
|
||
{
|
||
// var trackStatus = await _trackStatusRepository.GetFirstAsync(m => m.Code == content.TrackStatusCode);
|
||
if (content == null || !content.IsToMq)
|
||
return;
|
||
|
||
var data = new TrackChangeEto()
|
||
{
|
||
EstimatedDeliveryDate = eto.EstimatedDeliveryDate,
|
||
TrackPlacePostcode = eto.TrackPlacePostcode,
|
||
TrackPlaceNation = eto.TrackPlaceNation,
|
||
TrackPlaceNationId = eto.TrackPlaceNationId,
|
||
TrackPlaceProvince = eto.TrackPlaceProvince,
|
||
TrackPlaceProvinceId = eto.TrackPlaceProvinceId,
|
||
TrackPlaceCity = eto.TrackPlaceCity,
|
||
TrackPlaceCityId = eto.TrackPlaceCityId,
|
||
Files = eto.Files,
|
||
Remarks = eto.Remarks,
|
||
Source = eto.Source,
|
||
IsVisibility = eto.IsVisibility,
|
||
TrackTimeZone = eto.TrackTimeZone,
|
||
TrackTime = eto.TrackTime,
|
||
TrackPlace = eto.TrackPlace,
|
||
TrackStatusCode = content.TrackStatusCode,
|
||
EnStatus = eto.EnStatus,
|
||
Content = eto.Content,
|
||
TrackCode = eto.TrackCode,
|
||
OperationMan = eto.OperationMan,
|
||
OrderIdentifiers = eto.OrderIdentifiers,
|
||
CustomId = eto.CustomId,
|
||
SubSystemId = eto.SubSystemId
|
||
};
|
||
|
||
if (eto.IsAdd)
|
||
data.Option = 1;
|
||
else
|
||
data.Option = 3;
|
||
|
||
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson()));
|
||
}
|
||
|
||
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;
|
||
}
|
||
|
||
//虽然在引发事件之前做了判断，引发之后并不一定立即处理，在此过程中轨迹内容仍然可能被修改，所以要二次进行判断。
|
||
//通过来源指示是否为手动操作
|
||
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.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.TrackTime,
|
||
TrackTimeZone = @event.TrackTimeZone
|
||
}
|
||
};
|
||
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();
|
||
|
||
// 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)
|
||
{
|
||
TrackDetailsIndex? track;
|
||
if (!@event.IsAdd)
|
||
{
|
||
//同一订单同轨迹code同一轨迹时间如果有多条轨迹明细则不允许修改
|
||
var existTracks = tracks.Where(x => x.OrderId == order.OrderId
|
||
&& x.TrackTime == @event.TrackTime).ToArray();
|
||
if (existTracks.Length > 1)
|
||
{
|
||
// 修改轨迹时，订单存在多条轨迹不允许修改
|
||
continue;
|
||
}
|
||
|
||
track = existTracks.FirstOrDefault();
|
||
|
||
if (track == null) continue; // 不存在不让改
|
||
|
||
//来源与外部服务商不允许修改
|
||
if (track.Source == (int) TrackSourceEnum.ExternalServiceProvide) continue;
|
||
}
|
||
else
|
||
{
|
||
// 【批量新增】所有可能遇到的场景
|
||
// 标识：允许重复，不允许重复，零条，一条，多条；
|
||
// 场景A：允许重复，零条 => 新增
|
||
// 场景B：允许重复，一条 => 新增（约束受限新增，否则跳过新增）
|
||
// 场景C：允许重复，多条 => 新增（约束受限新增，否则跳过新增）
|
||
// 场景D：不允许重复，零条 => 新增
|
||
// 场景E：不允许重复，一条 => 跳过
|
||
// 场景F：不允许重复，（理想不会出现多条） => 跳过
|
||
var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToArray();
|
||
track = existTracks.FirstOrDefault();
|
||
|
||
//覆盖场景:B,C
|
||
if (trackContent.IsRepeatable && existTracks.Length >= 1)
|
||
{
|
||
if (existTracks.Any(x => x.TrackPlace == @event.TrackPlace && x.TrackTime == @event.TrackTime))
|
||
{
|
||
//允许重复，出现【轨迹发生地】，【轨迹发生时间】的都重复，则不允许新增
|
||
continue;
|
||
}
|
||
|
||
// 设置null后面执行新增逻辑
|
||
track = null;
|
||
}
|
||
|
||
//不允许重复，并且记录大于等于一条，跳过
|
||
//覆盖场景:E,F
|
||
if (!trackContent.IsRepeatable && existTracks.Length >= 1)
|
||
{
|
||
continue;
|
||
}
|
||
|
||
//其余情况：A,D都为新增操作，无需检查数量，track为null执行新增操作
|
||
}
|
||
|
||
if (track == default)
|
||
{
|
||
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,
|
||
//来源轨迹内容,第一次新增的时候使用轨迹内容表提供的
|
||
TrackStatus = trackContent.TrackStatus,
|
||
TrackStatusCode = trackContent.TrackStatusCode,
|
||
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,
|
||
};
|
||
|
||
track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
|
||
? track.TrackTime
|
||
: track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));
|
||
insertTracks.Add(track);
|
||
|
||
// 写入操作日志
|
||
_ = _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
|
||
});
|
||
}
|
||
else
|
||
{
|
||
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;
|
||
|
||
updateTracks.Add(track);
|
||
|
||
// 写入操作日志
|
||
_ = _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
|
||
});
|
||
}
|
||
|
||
canNextOrderIds.Add(order.OrderId);
|
||
}
|
||
|
||
if (updateTracks.Any())
|
||
{
|
||
await _trackDetailsEsRepository.UpdateManyAsync(updateTracks);
|
||
}
|
||
|
||
return insertTracks;
|
||
}
|
||
}
|