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

using DotNetCore.CAP;
using Newtonsoft.Json;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.InterfaceRelay.Application.Contracts;
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.TrackPushAggregate;
using SJZY.Track.Domain.Consts;
using SJZY.Track.Domain.Enums;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;

namespace SJZY.Track.Application.EventHandlers;

public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
{
    private readonly IHttpClientProxy<IInterfaceForwardServiceV2> _httpClientProxy;
    private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
    private readonly IOrderRepository _orderRepository;
    private readonly ITrackPushCacheService _trackPushCacheService;
    private readonly ICapPublisher _capPublisher;

    public TrackPushEventHandler(IHttpClientProxy<IInterfaceForwardServiceV2> httpClientProxy,
        ITrackPushRecordsRepository trackPushRecordsRepository,
        IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ICapPublisher capPublisher)
    {
        _httpClientProxy = httpClientProxy;
        _trackPushRecordsRepository = trackPushRecordsRepository;
        _orderRepository = orderRepository;
        _trackPushCacheService = trackPushCacheService;
        _capPublisher = capPublisher;
    }

    [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
    public async Task ConsumerAsync(TrackPushEvent @event)
    {
        var order = await _orderRepository.GetFirstAsync(x =>
            x.Id == @event.OrderId, x => new
        {
            x.SubSystemId, x.LatestPushTrackCode, x.QuestionType, x.CustomId, 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;

        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);
                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);
                break;
            default:
                throw new ArgumentOutOfRangeException();
        }

        if (!log.IsSuccess)
        {
            if (@event.RetryNum < 3)
            {
                // 1, 2, 4min
                await _capPublisher.PublishDelayAsync(
                    TimeSpan.FromSeconds((1 << @event.RetryNum) * 60),
                    GlobalConst.TrackPushSubscribe, @event);
                @event.RetryNum++;
                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)
    {
        var order = new OrderEntity();
        order.Id = dto.OrderId;
        /*
         *  异常：最近一条推送是失败
         */
        if (!dto.IsSuccess)
        {
            order.QuestionType = (int)TrackProblemType.Error;
            order.LatestPushError = dto.Message;
        }
        else
        {
            order.QuestionType = (int)TrackProblemType.None;
            order.LatestPushError = null;
            if (dto.OptionType == OptionType.新增)
            {
                // tips：只有新增会影响推送进度？
                order.LatestPushTrackCode = dto.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
        });
    }

    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 body = new TrackPushInput
        {
            OrderNo = @event.OrderNo,
        };
        body.Data.Add(track);

        if (config.IsAllPush)
        {
            var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(@event.OrderId);
            if (history != default)
            {
                body = JsonConvert.DeserializeObject<TrackPushInput>(history.PushContent);
                body!.Data.Add(track);
            }
        }

        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;
    }

    private async Task DoUpdateAsync(TrackPushEvent @event,
        TrackPushContentConfigDto config, TrackPushRecordIndex log)
    {
        var data = new
        {
            orderNo = @event.OrderNo,
            data = new[]
            {
                new
                {
                    // 推真实的发生地和发生时间
                    location = @event.TrackPlace,
                    date = @event.TrackTime.AddHours(
                        config.TrackTimeZone ?? @event.TrackTimeZone - @event.TrackTimeZone),
                    timeZone = config.TrackTimeZone ?? @event.TrackTimeZone,
                    description = config.TrackEnContentValue,
                    description_cn = config.TrackContentValue,
                    code = config.TrackContentCodeValue,
                    status = config.TrackStatusValue,
                    extra = config.ExpandFieldsValue
                }
            }
        };


        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;
    }
}