diff --git a/src/BuildFlow.Host/BuildFlowHostModule.cs b/src/BuildFlow.Host/BuildFlowHostModule.cs index 424e7d9..9b9ec81 100644 --- a/src/BuildFlow.Host/BuildFlowHostModule.cs +++ b/src/BuildFlow.Host/BuildFlowHostModule.cs @@ -15,6 +15,7 @@ using File.Infrastructure; using IM.Application; using IM.Domain.Common; using IM.Infrastructure; +using Workflow.Domain.Common; using BuildFlow.Platform.Application; using Volo.Abp; using Volo.Abp.Modularity; @@ -86,6 +87,9 @@ public class BuildFlowHostModule : AbpModule // 串联：IM 离线消息 → 平台通知中心站内信（跨模块集成，Host 实现 IImNotificationPublisher 供 IM 调用， // 内部转发为 PublishNotificationCommand 复用 platform 的 fanout 落库 + SignalR 推送） services.AddScoped(); + // 串联：工作流通知 → 平台消息中心铃铛（跨模块集成，Host 实现 IPlatformNotificationPublisher 供 workflow 调用， + // 内部转发为 PublishNotificationCommand 复用 platform 的 fanout 落库 + SignalR 推送） + services.AddScoped(); // 串联：AI 对话 → 触发工作流（跨模块集成，Workflow 实现 IWorkflowTriggerService 供 RAG 调用） services.AddScoped(); diff --git a/src/BuildFlow.Host/Endpoints/file/Buckets/CreateBucketEndpoint.cs b/src/BuildFlow.Host/Endpoints/file/Buckets/CreateBucketEndpoint.cs index 5c3ac25..8eff577 100644 --- a/src/BuildFlow.Host/Endpoints/file/Buckets/CreateBucketEndpoint.cs +++ b/src/BuildFlow.Host/Endpoints/file/Buckets/CreateBucketEndpoint.cs @@ -14,7 +14,6 @@ public class CreateBucketEndpoint(IMediator mediator) : Endpoint +/// 工作流通知 → 平台消息中心发布器。把工作流通知转发为 ， +/// 由 platform 的 PublishNotificationCommandHandler 完成 fanout 落库与 SignalR 铃铛推送。 +/// +/// +/// 跨模块胶水代码：工作流模块的 依赖 +/// （定义在 Workflow.Domain），但工作流模块无法直接访问 +/// platform 模块的 Notification 聚合。本实现位于 Host 层，通过 MediatR 转发，复用 platform 已有链路。 +/// 容错：转发失败不阻断工作流主链路（与 IImNotificationPublisher 同款吞异常降级策略）。 +/// +public class WorkflowNotificationPublisher(IMediator mediator) : IPlatformNotificationPublisher +{ + public async Task PublishAsync( + string type, + string title, + string content, + string source, + string? relatedId, + string? relatedType, + IEnumerable recipientUserIds, + CancellationToken ct = default) + { + var recipients = recipientUserIds as IReadOnlyCollection ?? recipientUserIds.ToList(); + if (recipients.Count == 0) return 0; + + try + { + return await mediator.Send(new PublishNotificationCommand( + Type: type, + Title: title, + Content: content, + Source: source, + RelatedId: relatedId, + RelatedType: relatedType, + RecipientUserIds: recipients, + RecipientRoles: null), ct); + } + catch + { + // 通知是旁路触达，失败不应影响工作流主链路，吞异常降级。 + return 0; + } + } +} diff --git a/src/modules/workflow/Application/Notifications/NotificationService.cs b/src/modules/workflow/Application/Notifications/NotificationService.cs index f7c2742..4b77522 100644 --- a/src/modules/workflow/Application/Notifications/NotificationService.cs +++ b/src/modules/workflow/Application/Notifications/NotificationService.cs @@ -1,5 +1,6 @@ using System.Text.Json; using Microsoft.Extensions.Options; +using Workflow.Domain.Common; using Workflow.Domain.Entities; using Workflow.Infrastructure.Persistence; @@ -8,6 +9,7 @@ namespace Workflow.Application.Notifications; /// /// 通知服务实现。处理任务生命周期事件： /// - 站内信：同步落库 Notification（必达，随调用方事务提交） +/// - 平台消息中心转发：落库同时转发到平台 notifications 表，经 SignalR 铃铛实时推送给用户 /// - Webhook：落库 WebhookDelivery 标 pending，由 WebhookDispatcherService 异步投递 /// /// 接收人解析策略：优先 task.AssigneeId（user: 规则），其次 task.AssigneeRole（role: 规则）。 @@ -15,7 +17,8 @@ namespace Workflow.Application.Notifications; /// public class NotificationService( WorkflowDbContext db, - IOptions options) : INotificationService + IOptions options, + IPlatformNotificationPublisher platformPublisher) : INotificationService { private readonly NotificationOptions _options = options.Value; @@ -89,7 +92,27 @@ public class NotificationService( await db.Notifications.AddRangeAsync(recipientNotifications, ct); } - // 3. Webhook 投递记录（若有配置） + // 3. 转发到平台消息中心（按 userId fanout，复用 SignalR 铃铛实时推送）。 + // 仅转发能确定具体 userId 的通知；role 维度的通知平台侧也需展开，这里暂不转发 role 类。 + var platformRecipients = recipientNotifications + .Where(n => n.RecipientUserId.HasValue) + .Select(n => n.RecipientUserId!.Value) + .Distinct() + .ToList(); + if (platformRecipients.Count > 0) + { + await platformPublisher.PublishAsync( + type: $"workflow-{category}", + title: title, + content: content, + source: "workflow", + relatedId: task.InstanceId.ToString(), + relatedType: "workflow-instance", + recipientUserIds: platformRecipients, + ct: ct); + } + + // 4. Webhook 投递记录（若有配置） await CreateWebhookDeliveryAsync(task, category, title, content, now, ct); } diff --git a/src/modules/workflow/Domain/Common/IPlatformNotificationPublisher.cs b/src/modules/workflow/Domain/Common/IPlatformNotificationPublisher.cs new file mode 100644 index 0000000..13db4a5 --- /dev/null +++ b/src/modules/workflow/Domain/Common/IPlatformNotificationPublisher.cs @@ -0,0 +1,36 @@ +namespace Workflow.Domain.Common; + +/// +/// 平台消息中心通知发布能力。把工作流通知转发到平台消息中心（notifications 表 + SignalR 铃铛推送）， +/// 复用 platform 的 fanout 落库与实时推送链路，避免工作流侧重复实现触达。 +/// +/// +/// 设计动机：工作流模块自带的 wf_notifications 表是孤立的私有通知日志，既不推 SignalR 也不进消息中心。 +/// 本接口在工作流落 wf_notifications 的同时，把同一通知转发到平台消息中心，让用户在铃铛里看到流程通知。 +/// 跨模块解耦：接口定义在工作流 Domain，实现在 Host 层（Host 同时引用 workflow 与 platform，通过 MediatR +/// 发 PublishNotificationCommand）。参照 IM 模块的 IImNotificationPublisher 同款模式。 +/// +public interface IPlatformNotificationPublisher +{ + /// + /// 按接收人 userId 批量发布站内信通知到平台消息中心。 + /// + /// 通知类型，如 "workflow-task-arrived" + /// 通知标题 + /// 通知正文 + /// 来源标识，固定 "workflow" + /// 关联实体 ID（流程实例/任务），用于点击跳转 + /// 关联实体类型，如 "workflow-instance" + /// 接收人 userId 列表 + /// 取消令牌 + /// 实际发布的通知条数（platform 侧 fanout 落库数） + Task PublishAsync( + string type, + string title, + string content, + string source, + string? relatedId, + string? relatedType, + IEnumerable recipientUserIds, + CancellationToken ct = default); +} diff --git a/tests/BuildFlow.Workflow.Tests/Notification/NotificationServiceTests.cs b/tests/BuildFlow.Workflow.Tests/Notification/NotificationServiceTests.cs index 8157240..86677d9 100644 --- a/tests/BuildFlow.Workflow.Tests/Notification/NotificationServiceTests.cs +++ b/tests/BuildFlow.Workflow.Tests/Notification/NotificationServiceTests.cs @@ -2,6 +2,7 @@ using FluentAssertions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Options; using Workflow.Application.Notifications; +using Workflow.Domain.Common; using Workflow.Domain.Entities; using Workflow.Domain.Enums; using Xunit; @@ -65,7 +66,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, Guid.NewGuid())); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskArrivedAsync(task); await db.SaveChangesAsync(); @@ -86,7 +87,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, Guid.NewGuid())); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskArrivedAsync(task); await db.SaveChangesAsync(); @@ -105,7 +106,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, initiatorId)); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskApprovedAsync(task); await db.SaveChangesAsync(); @@ -125,7 +126,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, userId)); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskApprovedAsync(task); await db.SaveChangesAsync(); @@ -142,7 +143,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, initiatorId)); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskRejectedAsync(task); await db.SaveChangesAsync(); @@ -158,7 +159,7 @@ public class NotificationServiceTests db.WorkflowInstances.Add(NewInstance(task.InstanceId, Guid.NewGuid())); await db.SaveChangesAsync(); - var svc = new NotificationService(db, Options.Create(DefaultOptions())); + var svc = new NotificationService(db, Options.Create(DefaultOptions()), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskTimeoutAsync(task, autoApproved: true); await db.SaveChangesAsync(); @@ -184,7 +185,7 @@ public class NotificationServiceTests DefaultUrl = "https://example.com/hook" } }; - var svc = new NotificationService(db, Options.Create(options)); + var svc = new NotificationService(db, Options.Create(options), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskArrivedAsync(task); await db.SaveChangesAsync(); @@ -212,7 +213,7 @@ public class NotificationServiceTests AllowedHosts = ["example.com"] } }; - var svc = new NotificationService(db, Options.Create(options)); + var svc = new NotificationService(db, Options.Create(options), new NoOpPlatformNotificationPublisher()); svc.IsHostAllowed(url).Should().Be(expected); } @@ -234,7 +235,7 @@ public class NotificationServiceTests DefaultUrl = "https://evil.com/hook" } }; - var svc = new NotificationService(db, Options.Create(options)); + var svc = new NotificationService(db, Options.Create(options), new NoOpPlatformNotificationPublisher()); await svc.NotifyTaskArrivedAsync(task); await db.SaveChangesAsync(); @@ -243,3 +244,16 @@ public class NotificationServiceTests (await db.WebhookDeliveries.CountAsync()).Should().Be(0); } } + +/// +/// 测试用 no-op 平台通知发布器：转发逻辑由 Host 层实现，单元测试无需真实推送， +/// 此实现空跑返回 0，避免 NotificationService 构造依赖阻断。 +/// +file sealed class NoOpPlatformNotificationPublisher : IPlatformNotificationPublisher +{ + public Task PublishAsync( + string type, string title, string content, string source, + string? relatedId, string? relatedType, + IEnumerable recipientUserIds, CancellationToken ct = default) + => Task.FromResult(0); +}