track/src/TrackCenter.Application/Messaging/TrackUpsertEventHandler.cs

using Fake.Messaging;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers;
using TrackCenter.Application.Events;
using TrackCenter.Application.Kafka;
using TrackCenter.Application.Queries;
using TrackCenter.Application.Services.Broadcasting;
using TrackCenter.Application.Services.Warning.Signals;
using TrackCenter.Domain.Aggregates.OrderAggregate;
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
using TrackCenter.Domain.Dtos.TrackContents;
using TrackCenter.Repository.Aggregates.OrderAggregate;
using TrackCenter.Repository.Otel;
using OrderIdentifier = TrackCenter.Application.Events.OrderIdentifier;

namespace TrackCenter.Application.Messaging;

public class TrackUpsertEventHandler(
    IOrderRepository orderRepository,
    IMessagePublisher messagePublisher,
    ProvideQueries provideQueries,
    ITrackDetailsEsRepository trackDetailsEsRepository,
    IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
    BroadcastingService broadcastingService,
    WarningSignalPublisher warningSignalPublisher)
    :
        IMessageHandle<TrackUpsertEvent>,
        IMessageHandle<TrackUpsertForRefreshEvent>,
        ITransientDependency
{
    public Task HandleAsync(TrackUpsertEvent message, CancellationToken cancellationToken = default)
        => ConsumerAsync(message);

    public Task HandleAsync(TrackUpsertForRefreshEvent message, CancellationToken cancellationToken = default)
        => ConsumerAsync(message);

    private async Task ConsumerAsync(TrackUpsertEvent @event)
    {
        if (!CanDelete(@event)) return;

        var canNextOrderIds = new List<int>();

        await HandleTrackAsync(@event, canNextOrderIds);

        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.AppKey,
                             x.Customer // 这里按客户分组是为了后续的推送方便
                         }))
            {
                var nextEvent = @event.DeepCopy();
                nextEvent.Visited.AddRange(canNextOrderIds);
                nextEvent.AppKey = group.Key.AppKey;
                nextEvent.Customer = group.Key.Customer;
                nextEvent.TrackRefreshData = new TrackRefreshStruct
                {
                    ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode,
                    OriginTrackContent = @event.TrackRefreshData.OriginTrackContent,
                    OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash,
                };

                // 不同应用需要做内容映射
                if (@event.AppKey != nextEvent.AppKey)
                {
                    var mapCode = (await provideQueries.MapTrackCodesAsync(@event.AppKey, nextEvent.AppKey, @event.TrackCode))
                        .FirstOrDefault();

                    if (mapCode == null)
                    {
                        // 跨应用没做映射的轨迹不处理
                        continue;
                    }

                    nextEvent.TrackCode = mapCode;
                }

                var identifiers = group.Select(x => new OrderIdentifier
                {
                    OrderId = x.OrderId,
                    OrderNo = x.OrderNo,
                    OrderType = x.OrderType,
                });

                foreach (var subOrderIds in identifiers.Chunk(GlobalConst.ChunkSize))
                {
                    var subEvent = nextEvent.DeepCopy();
                    subEvent.OrderIdentifiers = subOrderIds.ToArray();
                    await messagePublisher.PublishAsync(subEvent);
                }
            }
        }
    }

    private static bool CanDelete(TrackUpsertEvent @event)
    {
        // 系统创建的轨迹不允许修改
        if (@event.Visited.IsNullOrEmpty() && @event is { OperationMan: GlobalConst.SystemUser, IsAdd: false })
        {
            Log.Warning($"系统创建的轨迹不允许被修改：{@event}", true);
            return false;
        }

        var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();

        var dup = @event.Visited.Intersect(orderIds).ToArray();
        if (dup.Any())
        {
            Log.Warning($"{(@event.IsAdd ? "添加" : "修改")}轨迹时，发现循环依赖：{dup.JoinAsString(",")}", true);
            return false;
        }

        return true;
    }

    private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
    {
        //#7001:未来潜在需求，当轨迹内容匹配预设，记录一个标记，保证下刷过程都是使用预设内容
        var trackContent = await provideQueries.GetTrackContentAsync(@event.AppKey, @event.TrackCode);

        if (trackContent == null)
        {
            // 找不到轨迹内容，直接结束
            Log.Warning($"找不到轨迹内容：{@event}", true);
            return;
        }

        // 根据轨迹内容标识，确定操作权限
        var manualOperation = @event.Source == (int) TrackSourceEnum.Artificial;
        // 不允许手动创建
        if (manualOperation && @event.IsAdd && trackContent.IsAllowManuallyCreate == false)
        {
            PublicTopicAllFailed(@event, trackContent, "轨迹内容标识不允许手动创建");
            return;
        }

        // 不允许手动更新
        if (manualOperation && !@event.IsAdd && trackContent.IsAllowManuallyUpdate == false)
        {
            PublicTopicAllFailed(@event, trackContent, "轨迹标识内容不允许手动更新");
            return;
        }

        // k:[orderId] v:{EsId,Version,TrackCreateTime}
        var orderIdTrackMetadataDict = new Dictionary<int, TrackMetadata>();

        var trackChangeEtos = new List<TrackChangeEto>();

        var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent, orderIdTrackMetadataDict, trackChangeEtos);

        if (canNextOrderIds.Count == 0)
        {
            await TrackChangePublicTopicAsync(trackContent, trackChangeEtos);
            return;
        }

        if (insertTracks.Any())
        {
            // 批量add to es tips:注意副作用填充版本号
            await trackDetailsEsRepository.AddManyAsync(insertTracks);

            // es新增后填充版本号字典
            insertTracks.ForEach(t => orderIdTrackMetadataDict.Add(t.OrderId, TrackMetadata.Build(t.Id, t.Version, t.CreateTime)));

            // 批量写入订单操作日志
            _ = orderOperationRecordEsRepository.AddManyAsync(insertTracks.Select(t => BuildOrderOperationRecord(@event, t, new OrderIdentifier
            {
                OrderNo = t.OrderNo,
                OrderId = t.OrderId
            })).ToList());

            // 添加轨迹触发订单拉取状态更新
            await orderRepository.UpdatePullStatusToSilenceAsync(insertTracks.Select(x => x.OrderId).ToList(),
                @event.TrackCode);

            // 发布预警信号，稍后将进行预警计算
            warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, insertTracks);

            // 如果存在轨迹标记，则触发订单pod拉取
            if (!string.IsNullOrWhiteSpace(trackContent.TrackFlagId))
            {
                await messagePublisher.PublishAsync(new OrderPodPullEvent
                {
                    AppKey = @event.AppKey,
                    OrderIdEsIdDict = insertTracks.ToDictionary(x => x.OrderId, y => y.Id),
                    TrackFlagIds = trackContent.TrackFlagId.Split(',').Select(int.Parse).ToHashSet()
                });
            }
        }

        // 触发轨迹变更广播
        await TrackChangePublicTopicAsync(trackContent, trackChangeEtos, true);

        // 触发订单主表更新事件（异步）
        await messagePublisher.PublishAsync(new TrackChangeAfterUpdateOrderEvent
        {
            OrderIds = canNextOrderIds.ToArray(),
            IsCompleted = trackContent.IsCompleted,
            TrackStatus = trackContent.TrackStatus,
            TrackContent = trackContent.TrackContent,
            TrackEnContent = trackContent.TrackEnContent,
            TrackPlace = @event.TrackPlace,
            TrackTime = @event.TrackTime,
            TrackTimeZone = @event.TrackTimeZone,
        });

        // 触发轨迹推送事件（异步）
        // tips:同一个轨迹内容，多个订单多条轨迹
        var trackPushEvent = new TrackPushEvent
        {
            OptionType = @event.IsAdd ? OptionType.新增 : OptionType.修改,
            AppKey = @event.AppKey,
            Customer = @event.Customer,
            OrderIdTrackMetadataDict = orderIdTrackMetadataDict,
            TrackCode = @event.TrackCode,
            SystemPushInfo = new SystemPushStruct
            {
                TrackStatusId = trackContent.TrackStatusId,
                TrackStatus = trackContent.TrackStatus,
                TrackContentId = trackContent.TrackContentId,
                TrackContent = trackContent.TrackContent,
                TrackPlace = @event.TrackPlace,
                TrackTime = @event.TrackTime,
                TrackTimeZone = @event.TrackTimeZone,
                Extra = @event.Extra
            },
            ForcedPush = @event.ForcedPush,
            IsToMq = (trackContent.IsToMq || @event.IsToMq)
        };
        await messagePublisher.PublishAsync(trackPushEvent);
    }

    private async Task<List<TrackDetailsIndex>> UpsertTrackAsync(
        TrackUpsertEvent @event,
        List<int> canNextOrderIds,
        TrackContentVO trackContent,
        Dictionary<int, TrackMetadata> orderIdTrackMetadataDict,
        List<TrackChangeEto> trackChangeEtos
    )
    {
        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>();
        var operationRecords = new List<OrderOperationRecordIndex>();
        var traceLog = new Dictionary<string, object>
        {
            { "@event", @event.ToJson() }
        };
        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)
                {
                    var updateTrack = BuiltUpdateTrack(@event, trackById, order, trackContent);
                    updateTracks.Add(updateTrack);
                    AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, updateTrack.Id);
                }
                else
                {
                    var message = $"没有通过传入的EsId:[{@event.EsId}]查询到轨迹信息，更新失败";
                    AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message);
                }

                continue;
            }

            var existTracks = tracks.Where(x => x.OrderId == order.OrderId).ToList();

            // tips:如果没传时区则使用+8兜底进行过滤
            var timezone = Math.Abs(@event.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 0.01f ? 8.0f : @event.TrackTimeZone;

            // 根据时间过滤
            existTracks = existTracks.Where(x =>
                x.GetUtc0TrackTime() == @event.TrackTime.Subtract(TimeSpan.FromHours(timezone))).ToList();

            // 根据扩展信息过滤
            existTracks = existTracks.Where(x => x.Extra == @event.Extra).ToList();

            if (@event.IsAdd)
            {
                if (existTracks.Count > 0)
                {
                    const string message = "订单已经存在不允许重复的轨迹，不允许重复添加";
                    traceLog.Add($"OrderNo-{order.OrderNo}", message);
                    AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message);
                    continue;
                }

                var addTrack = BuiltNewTrack(@event, order, trackContent);
                insertTracks.Add(addTrack);
                AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, addTrack.Id);
            }
            else
            {
                // 不存在不让改
                if (existTracks.Count == 0)
                {
                    const string message = "订单不存在轨迹，不允许修改";
                    traceLog.Add($"OrderNo-{order.OrderNo}", message);
                    AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message);
                    continue;
                }

                if (existTracks.Count > 1)
                {
                    const string message = "订单存在多条轨迹不允许修改";
                    traceLog.Add($"OrderNo-{order.OrderNo}", message);
                    AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, null, message);
                    continue;
                }

                var track = existTracks.First();
                var updateTrack = BuiltUpdateTrack(@event, track, order, trackContent);
                updateTracks.Add(updateTrack);
                operationRecords.Add(BuildOrderOperationRecord(@event, track, order));
                AddTrackChangeEto(trackChangeEtos, @event, trackContent, order.OrderNo, updateTrack.Id);
            }

            canNextOrderIds.Add(order.OrderId);
        }

        ActivityHelper.StartSpanActivity("UpsertTrackAsync", traceLog);

        if (updateTracks.Any())
        {
            // 批量差异化更新轨迹
            await trackDetailsEsRepository.BulkUpdateAsync(updateTracks);
            // 填充轨迹元数据
            updateTracks.ForEach(x => orderIdTrackMetadataDict.Add(x.OrderId, TrackMetadata.Build(x.Id, x.Version, x.CreateTime)));
            // 批量写入操作日志
            _ = orderOperationRecordEsRepository.AddManyAsync(operationRecords);
            // 发布预警信号稍后将进行预警计算
            warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks);
        }

        return insertTracks;
    }

    private static TrackDetailsIndex BuiltUpdateTrack(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order, TrackContentVO trackContent)
    {
        // 非外部轨迹才允许修改 【发生地】【时间】【时区】
        if (track.Source != (int) TrackSourceEnum.ExternalServiceProvide)
        {
            track.TrackPlace = @event.TrackPlace;
            track.TrackTime = @event.TrackTime;
            track.TrackTimeZone = @event.TrackTimeZone;
        }

        // 允许二次修改轨迹内容
        track.TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content;
        track.TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus;

        // 轨迹发生地的更多信息
        track.TrackPlaceCity = GlobalConst.EmptyString;
        track.TrackPlaceCityId = GlobalConst.NotExistId;
        track.TrackPlaceProvince = GlobalConst.EmptyString;
        track.TrackPlaceProvinceId = GlobalConst.NotExistId;
        track.TrackPlaceNation = GlobalConst.EmptyString;
        track.TrackPlaceNationId = GlobalConst.NotExistId;
        track.TrackPlacePostcode = GlobalConst.EmptyString;

        // 签收地时区，签收地时间
        track.DestinationTimeZone = null;
        track.EstimatedDeliveryDate = null;

        // 转换为utc0的时间
        track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
            ? track.TrackTime
            : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));

        // 轨迹标记ids，是否对外，备注，扩展信息，附件
        track.TrackFlagIds = trackContent.TrackFlagId;
        track.IsExternal = @event.IsExternal ?? trackContent.IsExternal;
        track.Remarks = @event.Remarks ?? GlobalConst.EmptyString;
        track.Extra = @event.Extra;
        track.Files = @event.Files?.Select(x => new FileItem
        {
            OssKey = x.OssKey,
            FileName = x.FileName
        }).ToList();

        // 审计属性
        track.UpdateUser = @event.OperationMan ?? GlobalConst.SystemUser;
        track.UpdateTime = DateTime.Now;

        return track;
    }

    private static TrackDetailsIndex BuiltNewTrack(TrackUpsertEvent @event, OrderIdentifier order, TrackContentVO trackContent)
    {
        var track = new TrackDetailsIndex
        {
            // 订单信息
            AppKey = @event.AppKey,
            OrderId = order.OrderId,
            OrderNo = order.OrderNo,
            OrderType = order.OrderType,

            // 轨迹信息
            TrackCode = @event.TrackCode,
            TrackContentId = trackContent.TrackContentId,
            TrackContent = string.IsNullOrWhiteSpace(@event.Content) ? trackContent.TrackContent : @event.Content,
            TrackEnContent = string.IsNullOrWhiteSpace(@event.EnStatus) ? trackContent.TrackEnContent : @event.EnStatus,
            TrackStatusId = trackContent.TrackStatusId,
            TrackStatus = trackContent.TrackStatus,
            TrackStatusCode = trackContent.TrackStatusCode,
            TrackTime = @event.TrackTime,
            TrackTimeZone = @event.TrackTimeZone,
            TrackPlace = @event.TrackPlace,

            // 服务商轨迹信息
            ServiceProviderCode = @event.TrackRefreshData.ServiceProviderCode ?? GlobalConst.EmptyString,
            OriginTrackContent = @event.TrackRefreshData.OriginTrackContent ?? GlobalConst.EmptyString,
            OriginTrackDataHash = @event.TrackRefreshData.OriginTrackDataHash ?? GlobalConst.EmptyString,

            // 轨迹发生地的更多描述
            TrackPlaceCity = GlobalConst.EmptyString,
            TrackPlaceCityId = GlobalConst.NotExistId,
            TrackPlaceProvince = GlobalConst.EmptyString,
            TrackPlaceProvinceId = GlobalConst.NotExistId,
            TrackPlaceNation = GlobalConst.EmptyString,
            TrackPlaceNationId = GlobalConst.NotExistId,
            TrackPlacePostcode = GlobalConst.EmptyString,
            // 轨迹标记ids，轨迹来源，是否对外，备注，扩展信息，附件
            TrackFlagIds = trackContent.TrackFlagId,
            Source = @event.Source,
            IsExternal = @event.IsExternal ?? trackContent.IsExternal,
            Remarks = @event.Remarks ?? GlobalConst.EmptyString,
            Extra = @event.Extra,
            Files = @event.Files?.Select(x => new FileItem
            {
                OssKey = x.OssKey,
                FileName = x.FileName
            }).ToList(),

            // 审计属性
            CreateUser = @event.OperationMan ?? GlobalConst.SystemUser,
            CreateTime = DateTime.Now
        };

        // 转换为utc0的时间
        track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1
            ? track.TrackTime
            : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone));

        return track;
    }

    private static OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order)
    {
        return new OrderOperationRecordIndex
        {
            OrderId = order.OrderId,
            OrderNo = order.OrderNo,
            OptionType = @event.IsAdd ? OrderOperationType.创建轨迹.ToString() : OrderOperationType.修改轨迹.ToString(),
            Content = track.ToJson(),
            CreateUserId = @event.OperationManId,
            CreateUserName = @event.OperationMan ?? "",
            CreateTime = DateTime.Now
        };
    }

    private async Task TrackChangePublicTopicAsync(TrackContentVO trackContent, List<TrackChangeEto> list, bool isDelay = false)
    {
        if (list.Count == 0) return;
        // 理想情况在写入之后1s会刷盘 https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html
        // 增加一个版本号来确定是否真的变更了
        // 没有刷盘成功事件回调
        // 方案1：!!【轮询等待静默刷盘】查出版本号确实发生了变更在发送广播
        // 方案2：考虑强制刷盘（对es压力大）后发送
        // 方案3：采用一个较大延时，不查询直接发
        if (isDelay)
            await Task.Delay(1000);

        foreach (var eto in list)
            broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey);
    }

    private void PublicTopicAllFailed(TrackUpsertEvent @event, TrackContentVO trackContent, string failedReason)
    {
        if (trackContent.IsToMq == false && @event.IsToMq == false) return;
        foreach (var orderNo in @event.OrderIdentifiers.Select(x => x.OrderNo))
        {
            var eto = BuildEto(@event, trackContent, orderNo, null, failedReason);
            broadcastingService.TrackChangePublicTopic(eto, trackContent.AppKey);
        }
    }

    private static void AddTrackChangeEto(List<TrackChangeEto> list, TrackUpsertEvent @event, TrackContentVO trackContent, string orderNo, string? trackEsId = null, string? failedReason = null)
    {
        if (trackContent.IsToMq == false && @event.IsToMq == false) return;
        list.Add(BuildEto(@event, trackContent, orderNo, trackEsId, failedReason));
    }

    private static TrackChangeEto BuildEto(TrackUpsertEvent @event, TrackContentVO trackContent, string orderNo, string? trackEsId = null, string? failedReason = null)
    {
        return new TrackChangeEto
        {
            TrackEsId = trackEsId,
            Option = @event.IsAdd ? 1 : 3,
            IsSuccess = string.IsNullOrWhiteSpace(failedReason),
            FailedReason = failedReason,
            TrackPlacePostcode = null,
            TrackPlaceNation = null,
            TrackPlaceNationId = null,
            TrackPlaceProvince = null,
            TrackPlaceProvinceId = null,
            TrackPlaceCity = null,
            TrackPlaceCityId = null,
            Files = @event.Files,
            Remarks = @event.Remarks,
            Source = @event.Source,
            IsExternal = @event.IsExternal ?? trackContent.IsExternal,
            TrackTimeZone = @event.TrackTimeZone,
            TrackTime = @event.TrackTime,
            TrackPlace = @event.TrackPlace,
            TrackStatusCode = trackContent.TrackStatusCode,
            OrderNo = orderNo,
            EnStatus = @event.EnStatus,
            Content = @event.Content,
            TrackCode = @event.TrackCode,
            OperationMan = @event.OperationMan,
            OrderIdentifiers = @event.OrderIdentifiers,
            Customer = @event.Customer,
            AppKey = @event.AppKey,
            Extra = @event.Extra
        };
    }
}