using System.Text;
using FreeRedis;
using Microsoft.Extensions.Options;
using SJZY.Expand.ABP.Core;
using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords;
using SJZY.Track.Application.Contract.Dtos.Forewarning;
using SJZY.Track.Application.Contract.Services;
using SJZY.Track.Application.Options;
using SJZY.Track.Domain.Aggregates.ForewarningAggregate;
using SJZY.Track.Domain.Dtos.Forewarning;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Queries;
using SJZY.UserCenter.HttpApi.Services.Message;
using SJZY.UserCenter.SharedModel.Dtos.Message;
using SJZY.UserCenter.SharedModel.Enums;
using Volo.Abp.DependencyInjection;
using Volo.Abp.Http.Client.DynamicProxying;
namespace SJZY.Track.Application.Services.ForewarningServices;
///
/// 用于处理预警通知
///
public class ForewarningNotifyService : IForewarningNotifyService, ITransientDependency
{
private readonly IForewarningCacheService _forewarningCacheService;
private readonly IOptionsSnapshot _options; // not hot path
private readonly IHttpClientProxy _requestSendMessageService;
private readonly IWarningQueries _warningQueries;
private readonly IWarningRecordEsRepository _warningRecordEsRepository;
public ForewarningNotifyService(IWarningQueries warningQueries,
IWarningRecordEsRepository warningRecordEsRepository,
IHttpClientProxy requestSendMessageService,
IForewarningCacheService forewarningCacheService, IOptionsSnapshot options)
{
_warningQueries = warningQueries;
_warningRecordEsRepository = warningRecordEsRepository;
_requestSendMessageService = requestSendMessageService;
_forewarningCacheService = forewarningCacheService;
_options = options;
}
///
/// 创建预警通知
///
/// 创建预警通知需要的订单信息
public async Task WarningCalculateAsync(WarningDto dto)
{
if (dto.Orders.Count == 0) return;
var rules = await _forewarningCacheService.GetRulesAsync(dto.SubSystemId, dto.CustomerId);
if (rules == null || rules.Count == 0) return;
var adds = new List<(int ruleId, int orderId, int counter, long ts)>();
var dels = new List<(int ruleId, int orderId, int counter)>();
foreach (var order in dto.Orders)
{
// 加左区间
MatchRule(
rules.Where(rule => rule.StartTrackStatusId == dto.TrackStatusId).ToList(),
order
).ForEach(rule =>
{
var interval = (rule.TotalTimeSpan - rule.WarnTimeSpan) / rule.WarnCount;
var startTime = order.TrackTime.AddHours(rule.WarnTimeSpan);
for (var i = 1; i <= rule.WarnCount; i++)
{
var timestamp = GetTimestamp(startTime);
startTime = startTime.AddHours(interval);
// 计算预警时间戳,作为排序依据
adds.Add((rule.Id, order.Id, i, timestamp));
}
});
// 删右区间
MatchRule(
rules.Where(rule => rule.EndTrackStatusId == dto.TrackStatusId).ToList(),
order
).ForEach(rule =>
{
for (var i = 1; i <= rule.WarnCount; i++)
{
dels.Add((rule.Id, order.Id, i));
}
});
}
var redisKey = GetForewarningNotifySetKeyRedisKey(dto.SubSystemId);
long addCounter = 0, delCounter = 0;
// 左区间加入预警队列,【只有第一次发生的才会被加入】
var left = adds.Select(x =>
{
var entry = new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter);
return new ZMember(entry.ToString(), x.ts);
}).ToArray();
if (!left.IsNullOrEmpty())
{
addCounter = await RedisHelper.Client.ZAddNxAsync(redisKey, left);
}
// 右区间移出预警队列,【只要发生了就移除】
var right = dels.Select(x => new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter)).ToList();
if (!right.IsNullOrEmpty())
{
delCounter = await RedisHelper.Client.ZRemAsync(redisKey, right.Select(x => x.ToString()).ToArray());
}
LogHelper.Info($"{nameof(ForewarningNotifyService)} 本次共计产生{addCounter}条预警,移除{delCounter}条预警,触发缘由是{dto}");
await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成
}
public async Task SendNotificationAsync(int subSystemId)
{
var redisKey = GetForewarningNotifySetKeyRedisKey(subSystemId);
var timestamp = GetTimestamp(DateTime.Now);
var set = await RedisHelper.Client.ZRangeByScoreAsync(redisKey, 0, timestamp);
if (set.IsNullOrEmpty())
{
LogHelper.Info($"{nameof(ForewarningNotifyService)} 子系统:{subSystemId} 没有运输时效需要预警");
return 0;
}
var entries = set.Select(x => new WarningNotifyCacheEntry(x)).ToList();
var ruleIds = entries.Select(x => x.RuleId).Distinct().ToList();
var notifyRules = await _warningQueries.GetWarningNotifyRulesAsync(ruleIds);
// 写入es预警记录
var warningRecords = await WriteWarningRecordAsync(entries, notifyRules);
// 移除已处理的预警
await RedisHelper.Client.ZRemRangeByScoreAsync(redisKey, 0, timestamp);
// 发通知
// tips:通知发送失败不应该影响预警记录的写入,可以进入重试队列
await SendCoreAsync(subSystemId, entries, notifyRules);
return warningRecords.Count;
}
private async Task> WriteWarningRecordAsync(List entries,
List notifyRules)
{
var orders = await _warningQueries.GetOrderNoByIdsAsync(entries.Select(x => x.OrderId).ToArray());
var orderNoDict = orders.ToDictionary(x => x.Id, x => x.Name);
var warningRecords = new List();
foreach (var entry in entries)
{
var rule = notifyRules.FirstOrDefault(x => x.RuleId == entry.RuleId);
if (rule == null) continue;
foreach (var notifyUser in rule.NotifyUsers)
{
warningRecords.Add(new WarningRecordIndex
{
SubSystemId = rule.SubSystemId,
SubSystemName = rule.SubSystemName,
WarningName = rule.WarningName,
CustomId = rule.CustomId,
CustomName = rule.CustomName,
RuleId = entry.RuleId,
Rule = rule.ToString(),
Counter = entry.Counter,
IsLast = rule.WarnCount == entry.Counter,
OrderId = entry.OrderId,
OrderNo = orderNoDict.TryGetValue(entry.OrderId, out var orderNo) ? orderNo : string.Empty,
WarningStatus = WarningStatus.UnProcessed,
NotifyUserId = notifyUser.UserId,
NotifyUserName = notifyUser.UserName,
CreateTime = DateTime.Now,
});
}
}
// 批量写入es
await _warningRecordEsRepository.BulkInsertAsync(warningRecords);
return warningRecords;
}
private async Task SendCoreAsync(int subSystemId, List entries,
List notifyRules)
{
var userRules = new Dictionary>();
foreach (var rule in notifyRules)
{
foreach (var notifyUser in rule.NotifyUsers)
{
if (!userRules.ContainsKey(notifyUser.UserId))
{
userRules[notifyUser.UserId] = new List();
}
userRules[notifyUser.UserId].Add(rule);
}
}
// 以人的维度发送通知,异步
var tasks = new List();
foreach (var userRule in userRules)
{
tasks.Add(DoNotifyAsync(subSystemId, userRule.Key, userRule.Value, entries, NotifyMode.SiteInternal));
tasks.Add(DoNotifyAsync(subSystemId, userRule.Key, userRule.Value, entries, NotifyMode.FeiShu));
}
// 等待所有任务完成
await Task.WhenAll(tasks);
}
private string GetForewarningNotifySetKeyRedisKey(int subSystemId) =>
$"{GlobalConst.ForewarningNotifySetKey}:{subSystemId}";
private async Task DoNotifyAsync(int subSystemId, int userId, List rules,
List entries,
NotifyMode mode)
{
/* 以人的维度发送通知
*
预警名称:字节全程时效
预警规则:
{起飞-落地+普货+12345678+美国 时效9,预警时效7,次数3} 命中订单总数:124,详情点击https://middlebusiness.pet.shijizhongyun.com/trajectoryCenter/trajectoryOrder
{清关-提货+普货+12345678+美国 时效4,预警时效3,次数1} 命中订单总数:3250,详情点击https://middlebusiness.pet.shijizhongyun.com/trajectoryCenter/trajectoryOrder
预警名称:字节全程时效
预警规则:
{起飞-落地+普货+12345678+美国 时效9,预警时效7,次数3} 命中订单总数:124,详情点击https://middlebusiness.pet.shijizhongyun.com/trajectoryCenter/trajectoryOrder
{清关-提货+普货+12345678+美国 时效4,预警时效3,次数1} 命中订单总数:3250,详情点击https://middlebusiness.pet.shijizhongyun.com/trajectoryCenter/trajectoryOrder
*/
var notifyContent = new StringBuilder();
foreach (var warningGroup in rules
.Where(x => x.NotifyMode.Split(',').Select(Enum.Parse).Contains(mode))
.GroupBy(x => x.WarningName))
{
notifyContent.AppendLine($"预警名称:{warningGroup.Key}");
notifyContent.AppendLine($"预警规则:");
foreach (var rule in warningGroup.ToList())
{
var cnt = entries.Count(x => x.RuleId == rule.RuleId);
notifyContent.AppendLine(
$"{rule.ToString()} 命中订单总数:{cnt},详情点击{_options.Value.WarningNotifyUrl}?userId={userId}");
}
}
var msgContent = notifyContent.ToString();
if (string.IsNullOrEmpty(msgContent)) return;
var res = await _requestSendMessageService.Service.SendMessageByEmptyTemplateAsync(
new SendMessageByEmptyTemplateInput
{
IsNeedToHandle = true,
MsgSource = MsgSourceEnum.系统提醒,
SystemId = subSystemId,
TemplateCode = "trackWarning",
Channels = new List()
{
mode == NotifyMode.FeiShu ? ChannelEnum.飞书 : ChannelEnum.站内信,
},
MsgTitle = "轨迹运输时效预警",
MsgContent = msgContent,
receivers = new List
{
new()
{
RecipientType = 1,
ReceiverId = userId,
}
},
});
if (res.Code != GlobalConst.SuccessCode)
{
// todo:失败是否上报重试?
LogHelper.Error(
$"{nameof(ForewarningNotifyService)} 发送{mode.ToString()}通知失败,用户:{userId}\n内容:\n{msgContent}\n失败报文:{res.ToJson()}");
}
}
private static List MatchRule(List rules,
WarningOrderDto order)
{
var res = new List();
foreach (var rule in rules)
{
// 规则至少有一个因素,如果订单任一因素不符合则不命中
if (rule.IsTransit != null)
{
if (order.IsTransit != rule.IsTransit) continue;
}
if (!rule.CargoType.IsNullOrEmpty())
{
if (order.CargoType != rule.CargoType) continue;
}
if (!rule.LeavePort.IsNullOrEmpty())
{
if (order.LeavePort != rule.LeavePort) continue;
}
if (!rule.ProductCode.IsNullOrEmpty())
{
if (order.ProductCode != rule.ProductCode) continue;
}
if (!rule.Months.IsNullOrEmpty())
{
if (!order.Month.IsIn(rule.Months)) continue;
}
if (!rule.Areas.IsNullOrEmpty())
{
var inArea = false;
var areas = rule.Areas!.Where(area => order.ArrivalCountry == area.Country).ToList();
foreach (var area in areas)
{
if (!area.LeftPostcode.IsNullOrEmpty() && !area.RightPostcode.IsNullOrEmpty())
{
// tips: 参考小包的邮编匹配,字典序
if (string.CompareOrdinal(order.ArrivalPostCode, area.LeftPostcode) < 0 ||
string.CompareOrdinal(order.ArrivalPostCode, area.RightPostcode) > 0) continue;
}
inArea = true;
break;
}
if (!inArea)
{
foreach (var area in areas)
{
if (!area.City.IsNullOrEmpty() && order.ArrivalCity != area.City) continue;
inArea = true;
break;
}
}
if (!inArea)
{
foreach (var area in areas)
{
if (!area.Province.IsNullOrEmpty() && order.ArrivalProvince != area.Province) continue;
inArea = true;
break;
}
}
if (!inArea) continue;
}
// 命中规则
res.Add(rule);
}
return res;
}
private static long GetTimestamp(DateTime dateTime)
{
return new DateTimeOffset(dateTime).ToUnixTimeSeconds();
}
}