track/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs

using System.Diagnostics;
using DotNetCore.CAP;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Dtos.ProvideServices;
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.Consts;
using SJZY.Track.Domain.Queries;
using SJZY.UserCenter.HttpApi.Services;
using Volo.Abp.DependencyInjection;

namespace SJZY.Track.Application.EventHandlers;

public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
{
    private readonly IOrderRepository _orderRepository;
    private readonly IProvideQueries _provideQueries;
    private readonly ITrackDetailsRepository _trackDetailsRepository;
    private readonly ITrackPushCacheService _trackPushCacheService;
    private readonly ICapPublisher _capPublisher;

    public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher,
        IProvideQueries provideQueries, ITrackDetailsRepository trackDetailsRepository,
        ITrackPushCacheService trackPushCacheService)
    {
        _orderRepository = orderRepository;
        _capPublisher = capPublisher;
        _provideQueries = provideQueries;
        _trackDetailsRepository = trackDetailsRepository;
        _trackPushCacheService = trackPushCacheService;
    }

    [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
    public async Task ConsumerAsync(TrackUpsertEvent @event)
    {
        if (@event.Visited.Contains(@event.OrderId))
        {
            LogHelper.Error($"轨迹订单循环依赖预警：{@event.Visited.JoinAsString(",")}");
            return;
        }

        @event.Visited.Add(@event.OrderId);

        // 不同子系统需要做内容映射
        if (@event.UpSubSystemId != 0 && @event.UpSubSystemId != @event.SubSystemId)
        {
            var mapCode = (await _provideQueries
                    .MapTrackCodesAsync(@event.UpSubSystemId, @event.SubSystemId, @event.TrackCode))
                .FirstOrDefault();

            if (mapCode == default)
            {
                // 跨子系统没做映射的轨迹不处理
                return;
            }

            @event.TrackCode = mapCode;
        }

        var track = await _trackDetailsRepository.GetFirstOrDefaultAsync(@event.OrderId, @event.TrackCode);

        // step1、upsert es轨迹
        await UpsertTrackAsync(track, @event);
        Debug.Assert(track != null, nameof(track) + " != null");

        // step2、触发轨迹推送事件（异步）
        await HandleTrackPushAsync(@event, track);

        // step3、触发订单主表更新事件（异步）
        await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
            new TrackChangeAfterUpdateOrderEvent
            {
                OrderId = track.OrderId,
                IsCompleted = track.IsCompleted,
                TrackCode = track.TrackCode,
                TrackContent = track.TrackContent,
                TrackEnContent = track.TrackEnContent,
                TrackPlace = track.TrackPlace,
                TrackTime = track.TrackTime,
                TrackTimeZone = track.TrackTimeZone
            });

        // step4、根据订单关联关系异步向下刷（异步）
        var linkedOrders = await _orderRepository.GetLinkedOrdersAsync(@event.OrderId);
        foreach (var linkedOrder in linkedOrders)
        {
            var nextEvent = @event.DeepCopy();
            nextEvent.OrderId = linkedOrder.OrderId;
            nextEvent.OrderType = linkedOrder.OrderType;
            nextEvent.OrderNo = linkedOrder.OrderNo;
            nextEvent.SubSystemId = linkedOrder.SubSystemId;
            nextEvent.LatestPushTrackContentId = linkedOrder.LatestPushTrackContentId;
            nextEvent.CustomId = linkedOrder.CustomId;
            nextEvent.UpSubSystemId = @event.SubSystemId;
            await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, nextEvent);
        }
    }

    /// <summary>
    /// 不存在则新增轨迹，存在则更新
    /// </summary>
    /// <param name="track"></param>
    /// <param name="event"></param>
    private async Task UpsertTrackAsync(TrackDetailsIndex? track, TrackUpsertEvent @event)
    {
        if (track == default)
        {
            //todo：可以优化
            // 根据轨迹code在订单模板内容中寻找匹配的
            var trackContent = await _provideQueries.MatchTrackContentAsync(@event.OrderId, @event.TrackCode);
            if (trackContent == default)
            {
                // 订单所绑定的模板，没有一条内容可以匹配发生的轨迹
                return;
            }

            track = new TrackDetailsIndex
            {
                IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}",
                OrderId = @event.OrderId,
                OrderNo = @event.OrderNo,
                OrderType = @event.OrderType,
                TrackStatus = trackContent.Status,
                TrackCode = @event.TrackCode,
                TrackContent = trackContent.Content,
                TrackContentId = trackContent.Id,
                IsCompleted = trackContent.IsCompleted,
                TrackEnContent = trackContent.EnContent,
                TrackPlace = @event.TrackPlace,
                TrackTime = @event.TrackTime!.Value,
                TrackTimeZone = @event.TrackTimeZone!.Value,
                IsCanDelete = trackContent.IsCanDelete,
                CreateUser = CurrentUserRequest.CurrentUser.Name,
                CreateTime = DateTime.Now
            };
        }
        else
        {
            track.TrackPlace = @event.TrackPlace;
            track.TrackTime = @event.TrackTime!.Value;
            track.TrackTimeZone = @event.TrackTimeZone!.Value;
            track.UpdateUser = CurrentUserRequest.CurrentUser.Name;
            track.UpdateTime = DateTime.Now;
        }

        await _trackDetailsRepository.UpsertAsync(track);
    }

    /// <summary>
    /// 处理轨迹自动推送
    /// </summary>
    /// <param name="event"></param>
    /// <param name="track"></param>
    private async Task HandleTrackPushAsync(TrackUpsertEvent @event, TrackDetailsIndex track)
    {
        TrackPushContentConfigDto? config;
        OptionType optionType = OptionType.修改;
        // 推送更新
        if (!track.Id.IsNullOrWhiteSpace())
        {
            config = await _trackPushCacheService.GetPushContentConfigAsync(
                @event.SubSystemId, @event.CustomId, track.TrackContentId);
            if (!config.IsCanUpdate) return;
        }
        else
        {
            // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行
            config = await _trackPushCacheService.InferNextPushContentOrNullAsync(
                @event.SubSystemId, @event.CustomId, @event.LatestPushTrackContentId);

            if (config == null || !config.IsAutoPush) return;

            // 当前发生的轨迹就是要推送的
            if (config.TrackContentId != track.TrackContentId) return;

            optionType = OptionType.新增;
        }

        if (config.OrderType != @event.OrderType) return;

        var trackPushEvent = new TrackPushEvent
        {
            OptionType = optionType,
            ServiceProviderCode = config.ServiceProviderCode,
            OrderNo = track.OrderNo,
            TrackContentId = config.TrackContentId,
            TrackStatus = track.TrackStatus,
            TrackCode = track.TrackCode,
            TrackContent = track.TrackContent,
            TrackPlace = track.TrackPlace,
            TrackTime = track.TrackTime.AddHours(config.TrackTimeZone ?? track.TrackTimeZone - track.TrackTimeZone),
            TrackTimeZone = (float)(config.TrackTimeZone ?? track.TrackTimeZone),
            TrackContentCodeValue = config.TrackContentCodeValue,
            TrackContentValue = config.TrackContentValue,
            TrackEnContentValue = config.TrackEnContentValue,
            TrackStatusValue = config.TrackStatusValue,
            ExpandFieldsValue = config.ExpandFieldsValue
        };

        await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
    }
}