using Fake.Application; using Fake.DependencyInjection; using FeishuNetSdk; using Hangfire; using Microsoft.Extensions.Options; using SJZY.Expand.ABP.Core.Logger; using SJZY.InterfaceRelay.Application.Contracts; using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Domain.Aggregates.WarningAggregate; using TrackCenter.Domain.Aggregates.WarningAggregate.Return; using TrackCenter.Domain.Dtos.Warning; using TrackCenter.Domain.Enums; using TrackCenter.Domain.Extensions; using TrackCenter.Domain.Queries; using SJZY.UserCenter.HttpApi.Services.Message; using TrackCenter.Application.Helpers; using TrackCenter.Application.Options; using TrackCenter.Application.Track2InterfaceRelay; using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; namespace TrackCenter.Application.Hangfire; public interface IWarningTaskService { [AutomaticRetry(Attempts = 1)] Task CallNotifyAsync(WarningNotifyParam param, CancellationToken ct); [AutomaticRetry(Attempts = 3, ExceptOn = new[] { typeof(BusinessException) })] Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity entity, CancellationToken ct); } public class WarningTaskService : IWarningTaskService, ITransientDependency { private readonly IOptionsMonitor _interfaceRelayOptions; private readonly IOptionsMonitor _warningTaskOptions; private readonly IHttpClientProxy _requestSendMessageService; private readonly IHttpClientProxy _interfaceForwardServiceV2; private readonly IFeishuTenantApi _feishuTenantApi; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly IWarningTaskRepository _warningTaskRepository; private readonly IVFsUserQueries _vfsUserQueries; public WarningTaskService( IOptionsMonitor interfaceRelayOptions, IOptionsMonitor warningTaskOptions, IHttpClientProxy interfaceForwardServiceV2, IHttpClientProxy requestSendMessageService, IFeishuTenantApi feishuTenantApi, IWarningDetailsEsRepository warningDetailsEsRepository, IWarningTaskRepository warningTaskRepository, IVFsUserQueries vfsUserQueries) { _interfaceRelayOptions = interfaceRelayOptions; _warningTaskOptions = warningTaskOptions; _interfaceForwardServiceV2 = interfaceForwardServiceV2; _requestSendMessageService = requestSendMessageService; _feishuTenantApi = feishuTenantApi; _warningDetailsEsRepository = warningDetailsEsRepository; _warningTaskRepository = warningTaskRepository; _vfsUserQueries = vfsUserQueries; } #region 用户中心调用,通知(一次) [AutomaticRetry(Attempts = 1)] public async Task CallNotifyAsync(WarningNotifyParam param, CancellationToken ct) { ct.ThrowIfCancellationRequested(); var options = _warningTaskOptions.CurrentValue; // 发送站内信 // https://centerapi.test.shijizhongyun.com/userApi/index.html?urls.primaryName=%E6%B6%88%E6%81%AF%E8%AE%A2%E9%98%85%E5%AE%9E%E4%BE%8B // MsgSubscribeInstance/SendMessageByTemplate var r = await _requestSendMessageService.Service.SendMessageByTemplateAsync(WarningNotifyBuilder.BuildZhanNeiXinMessage(param, options.MiddleSystemId)); if (r.Code != GlobalConst.SuccessCode) throw new BusinessException(message: $"用户中心调用失败:SendMessageByTemplateAsync Msg:{r.Msg}"); // 发送FeiShu卡片信息 var fsUserId = await _vfsUserQueries.GetFirstAsync(x => x.UserCenterId == param.UserId, y => y.FsEmployeeId); if (string.IsNullOrWhiteSpace(fsUserId)) throw new BusinessException(message: "查询用户中心失败:SendMessageByTemplateAsync Msg:未找到用户对应的飞书用户Id"); var res = await _feishuTenantApi.PostImV1MessagesAsync("user_id", WarningNotifyBuilder.BuildFeiShuCard(param, fsUserId)); if (res.Code != 0) throw new BusinessException(message: $"飞书调用失败:SendMessageByTemplateAsync Msg:{res.Msg}"); } #endregion #region 接口通调用,轨迹推送请求(一次) [AutomaticRetry(Attempts = 3, ExceptOn = new[] { typeof(BusinessException) })] //遇到BusinessException就不重试了 public async Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity task, CancellationToken ct) { ct.ThrowIfCancellationRequested(); // 1.构建参数 List items; var isAdd = task.OptionType == OptionType.新增.ToString(); if (isAdd) { var pushInputs = TextJsonHelper.Deserialize>(task.RequestParams)!; items = pushInputs.Cast().ToList(); } else { var updateInputs = TextJsonHelper.Deserialize>(task.RequestParams)!; items = updateInputs.Cast().ToList(); } var retryCountChanged = false; // 2.请求 try { using (Track2InterfaceRelaySubSystemContext.Use(new Track2InterfaceRelaySubSystemContext(task.SubSystemCode))) { var upStreamCode = isAdd ? _interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode : _interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode; var r = await _interfaceForwardServiceV2.Service.BatchForwardThenPushToQueueAsync( upStreamCode: upStreamCode, serviceProviderCode: task.ServiceProviderCode, data: items, resSubscribeName: GlobalConst.TrackPushCompletedSubscribe, attach: Base64Helper.SerializeObject(task.OptionType)); if (r.Code != GlobalConst.SuccessCode) { var msg = $"接口通调用失败:BatchForwardThenPushToQueueAsync" + $"RequestId: {r.RequestId};SubSystemCode:{task.SubSystemCode};" + $" serviceProviderCode:{task.ServiceProviderCode} Msg: {r.Msg}"; // 处理调用接口通失败,直接处于完结状态 await HandlerCallBusinessException(task, msg); retryCountChanged = true; throw new BusinessException(message: msg); } } } catch (Exception e) { LogHelper.Error("[CallTrackRetryPushAsync]" + e, true); throw; } finally { // 3.维护回传预警明细的重试次数 if (!retryCountChanged) { var warningDetailsIds = task.WarningDetailIds.Split(',').ToHashSet(); await _warningDetailsEsRepository.ReturnWarningDetailsRetryIncrementAsync(warningDetailsIds); } } } private async Task HandlerCallBusinessException(ReturnWarningRetryPushTaskEntity task, string msg) { var detailIds = task.WarningDetailIds.Split(',').ToHashSet(); // 1.查询处理(重试中/人工介入)的预警明细 var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO { DetailsIds = detailIds }); // 由于后续都不会在成功了,所以直接根据是否命中kpi考核 var addNotifyTasks = new List(); foreach (var detail in warningDetails) { // 重试次数+1 detail.RetryCount++; if (detail.Status == ReturnWarningStatus.已关闭.GetHashCode()) continue; // 存在kpi考核 if (detail.IsExistsKpi) { detail.Status = ReturnWarningStatus.人工介入.GetHashCode(); addNotifyTasks.Add(BuildWarningNotifyTask(detail)); continue; } // 不存在kpi考核直接关闭 detail.Status = ReturnWarningStatus.已关闭.GetHashCode(); } // 更新预警明细,添加回传报文记录 await _warningDetailsEsRepository.BulkUpdateReturnWarningDetailsAsync(warningDetails); await _warningDetailsEsRepository.InsertReturnMessageRecordAsync(new ReturnMessageRecordIndex { WarningDetailsIds = task.WarningDetailIds, IsSuccess = false, Message = msg, CreateTime = DateTime.Now }); // 添加通知任务 if (addNotifyTasks.Count > 0) await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks); } private static WarningNotifyTaskEntity BuildWarningNotifyTask(ReturnWarningDetailsIndex detail) { return new WarningNotifyTaskEntity( detail.OrderId, detail.WarningId!.Value, detail.ReturnWarningRuleItemId!.Value, detail.Id, WarningRuleType.回传时效预警.GetHashCode(), detail.NotifyUsers! ); } #endregion }