using System.Text.Json; using F9WebAiTest.Application.Common; using F9WebAiTest.Application.Dtos; using F9WebAiTest.Application.Dtos.Notice; using F9WebAiTest.Domain.Common; using F9WebAiTest.Domain.Entities; using F9WebAiTest.Domain.Entities.Notice; using MediatR; namespace F9WebAiTest.Application.Services.Notice; /// /// 业务系统触发消息创建的命令(通用化)。 /// 用法:await _mediator.Send(new CreateNotificationCommand { NotificationTypeCode = "LeaveApproval", ... }); /// public class CreateNotificationCommand : IRequest { public string NotificationTypeCode { get; set; } = ""; // 消息类型编码,如 "LeaveApproval" public string BusinessType { get; set; } = ""; // 业务类型标识 public string BusinessId { get; set; } = ""; // 业务单据ID public string Title { get; set; } = ""; // 标题 public string Content { get; set; } = ""; // 内容 public NotificationPriority Priority { get; set; } = NotificationPriority.Normal; public string BizContext { get; set; } = ""; // 业务上下文(JSON) public List ForceReceivers { get; set; } = new(); // 强制接收人(绕过订阅) public string RoutePath { get; set; } = ""; // 跳转路由(可选,覆盖类型模板) public List AssigneeIds { get; set; } = new(); // 待办处理人(Category=Todo 时生效) public DateTime? DueTime { get; set; } // 待办截止时间 public long CreatorId { get; set; } // 创建人ID public string CreatorName { get; set; } = ""; // 创建人姓名 } /// /// 业务侧发送消息的简洁命令(基于消息类型枚举)。 /// 用法(参照 CQRS 惯例): /// var ok = await mediator.Send(new SendNotificationCommand( /// NotificationTypeEnum.LeaveApproval, /// BusinessId: leaveId.ToString(), /// ForceReceivers: approverIds, /// AssigneeIds: approverIds)); /// if (!ok) throw new F9OperationException("请假审批消息发送失败"); /// public record SendNotificationCommand( NotificationTypeEnum Type, string BusinessId = "", string Title = "", string Content = "", string BizContext = "", NotificationPriority Priority = NotificationPriority.Normal, List ForceReceivers = null, List AssigneeIds = null, DateTime? DueTime = null ) : IRequest; public class SendNotificationCommandHandler : IRequestHandler { private readonly IMediator _mediator; public SendNotificationCommandHandler(IMediator mediator) => _mediator = mediator; public async Task Handle(SendNotificationCommand command, CancellationToken cancellationToken) { // 枚举名即消息类型 Code,复用底层创建命令 var msgId = await _mediator.Send(new CreateNotificationCommand { NotificationTypeCode = command.Type.ToString(), BusinessType = command.Type.ToString(), BusinessId = command.BusinessId ?? "", Title = command.Title ?? "", Content = command.Content ?? "", BizContext = command.BizContext ?? "", Priority = command.Priority, ForceReceivers = command.ForceReceivers ?? new List(), AssigneeIds = command.AssigneeIds ?? new List(), DueTime = command.DueTime }, cancellationToken); return msgId > 0; } } /// /// 创建消息(业务触发的统一入口)。 /// 创建后**立即**确定接收人、写入接收人表并实时推送(已移除后台投递服务,无轮询、无延迟)。 /// 接收人规则:ForceReceivers 非空 → 精确送达;否则 → 按该消息类型的订阅关系广播。 /// 可靠性:接收人幂等(MessageId + UserId 唯一);用户离线时推送无接收者但消息已落库,上线后拉取兜底。 /// public class CreateNotificationCommandHandler : IRequestHandler { private readonly IRepository _typeRepo; private readonly IRepository _msgRepo; private readonly IRepository _todoRepo; private readonly IRepository _subRepo; private readonly IRepository _receiverRepo; private readonly INotificationPusher _pusher; public CreateNotificationCommandHandler( IRepository typeRepo, IRepository msgRepo, IRepository todoRepo, IRepository subRepo, IRepository receiverRepo, INotificationPusher pusher) { _typeRepo = typeRepo; _msgRepo = msgRepo; _todoRepo = todoRepo; _subRepo = subRepo; _receiverRepo = receiverRepo; _pusher = pusher; } public async Task Handle(CreateNotificationCommand cmd, CancellationToken cancellationToken) { var type = await _typeRepo.FirstOrDefaultAsync(t => t.Code == cmd.NotificationTypeCode); if (type == null) throw new ServiceException($"消息类型「{cmd.NotificationTypeCode}」不存在,请先由管理员创建"); if (!type.IsActive) throw new ServiceException($"消息类型「{type.Name}」已关闭订阅"); // 生成跳转路由:命令直接指定 > 类型模板替换业务ID var routePath = !string.IsNullOrWhiteSpace(cmd.RoutePath) ? cmd.RoutePath : (!string.IsNullOrWhiteSpace(type.RouteTemplate) ? type.RouteTemplate.Replace("{businessId}", cmd.BusinessId ?? "") : ""); var msg = new NotificationMessage { NotificationTypeId = type.Id, MessageTypeCode = type.Code, BusinessType = cmd.BusinessType, BusinessId = cmd.BusinessId, Title = cmd.Title, Content = cmd.Content, Priority = cmd.Priority, Status = NotificationStatus.Sent, // 即时送达:无后台队列,创建即投递 BizContext = cmd.BizContext ?? "", ForceReceivers = cmd.ForceReceivers == null || cmd.ForceReceivers.Count == 0 ? "" : JsonSerializer.Serialize(cmd.ForceReceivers), CreatorId = cmd.CreatorId, CreatorName = cmd.CreatorName, Category = type.Category, RoutePath = routePath }; await _msgRepo.AddAsync(msg); await _msgRepo.SaveChangesAsync(); // 待办消息:为每个处理人生成待办事项 if (type.Category == NotificationCategory.Todo && cmd.AssigneeIds != null && cmd.AssigneeIds.Count > 0) { foreach (var assigneeId in cmd.AssigneeIds.Distinct()) { await _todoRepo.AddAsync(new TodoItem { MessageId = msg.Id, AssigneeId = assigneeId, TodoStatus = TodoItemStatus.Pending, DueTime = cmd.DueTime }); } await _todoRepo.SaveChangesAsync(); } // 即时投递 + 实时推送 await DeliverAsync(msg, type, cancellationToken); return msg.Id; } /// 确定接收人 → 写接收人表 → 实时推送(推送失败不影响业务,消息已落库可拉取) private async Task DeliverAsync(NotificationMessage msg, NotificationType type, CancellationToken ct) { // 1. 确定接收人(精确送达 与 订阅广播 二选一) var receiverIds = new HashSet(); var hasForceReceivers = false; if (!string.IsNullOrWhiteSpace(msg.ForceReceivers)) { try { var force = JsonSerializer.Deserialize>(msg.ForceReceivers); if (force != null && force.Count > 0) { hasForceReceivers = true; foreach (var id in force) receiverIds.Add(id); } } catch { /* 忽略非法 JSON */ } } if (!hasForceReceivers && type.IsActive) { var subs = await _subRepo.FindAsync(s => s.NotificationTypeId == msg.NotificationTypeId && s.IsActive); foreach (var s in subs) receiverIds.Add(s.UserId); } if (receiverIds.Count == 0) return; // 2. 写接收人(幂等:MessageId + UserId 唯一) var added = new List(); foreach (var userId in receiverIds) { var exist = await _receiverRepo.FirstOrDefaultAsync(r => r.MessageId == msg.Id && r.UserId == userId); if (exist != null) continue; var receiver = new NotificationReceiver { MessageId = msg.Id, UserId = userId, ReadStatus = NotificationReadStatus.Unread, SendStatus = NotificationSendStatus.NotSent }; await _receiverRepo.AddAsync(receiver); added.Add(receiver); } if (added.Count == 0) return; await _receiverRepo.SaveChangesAsync(); // 3. 实时推送新消息 + 未读统计 // 在线(Hub 有活动连接)→ 推送并标记成功; // 离线 → 不推送、标记 IsOffline,消息已落库,用户上线后从"我的消息"拉取兜底。 var dto = BuildDto(msg, type.Name); foreach (var receiver in added) { try { var delivered = await _pusher.PushNewMessageAsync(receiver.UserId, dto, ct); if (delivered) { receiver.SendStatus = NotificationSendStatus.Success; receiver.SendTime = DateTime.Now; receiver.IsOffline = false; var unread = await _receiverRepo.CountAsync(r => r.UserId == receiver.UserId && r.ReadStatus == NotificationReadStatus.Unread); await _pusher.PushUnreadCountAsync(receiver.UserId, (int)unread, 0, ct); } else { // 用户离线:不推送,等待上线拉取 receiver.SendStatus = NotificationSendStatus.NotSent; receiver.IsOffline = true; } } catch (Exception ex) { // 推送异常(连接刚好断开等):标记失败与离线,消息已持久化,上线后拉取兜底 receiver.SendStatus = NotificationSendStatus.Failed; receiver.IsOffline = true; receiver.ErrorMsg = ex.Message; } await _receiverRepo.UpdateAsync(receiver); } await _receiverRepo.SaveChangesAsync(); } private static NotificationMessageDto BuildDto(NotificationMessage msg, string typeName) => new() { Id = msg.Id, NotificationTypeId = msg.NotificationTypeId, MessageTypeCode = msg.MessageTypeCode, TypeName = typeName, Title = msg.Title, Content = msg.Content, Priority = (int)msg.Priority, BusinessType = msg.BusinessType, BusinessId = msg.BusinessId, Category = (int)msg.Category, RoutePath = msg.RoutePath, CreatedTime = msg.CreatedAt, ReadStatus = 0 }; }