feat(workflow): 集成平台消息中心实现工作流通知推送

- 移除文件端点的权限控制以支持无权限访问
- 添加 IPlatformNotificationPublisher 接口定义跨模块通知发布能力
- 实现 WorkflowNotificationPublisher 将工作流通知转发到平台消息中心
- 在 NotificationService 中集成平台通知发布功能
- 添加 NoOpPlatformNotificationPublisher 用于单元测试
- 配置服务注册将工作流通知连接到平台消息系统
- 支持按用户ID批量发布工作流通知到平台通知表
- 复用平台的 fanout 落库和 SignalR 铃铛推送机制
- 实现容错机制确保通知失败不影响工作流主链路
This commit is contained in:
xiangning 2026-07-03 16:33:46 +08:00
parent f932781040
commit 81fbf63504
29 changed files with 138 additions and 35 deletions

View File

@ -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<IImNotificationPublisher, ImNotificationPublisher>();
// 串联:工作流通知 → 平台消息中心铃铛(跨模块集成,Host 实现 IPlatformNotificationPublisher 供 workflow 调用,
// 内部转发为 PublishNotificationCommand 复用 platform 的 fanout 落库 + SignalR 推送)
services.AddScoped<IPlatformNotificationPublisher, WorkflowNotificationPublisher>();
// 串联:AI 对话 → 触发工作流(跨模块集成,Workflow 实现 IWorkflowTriggerService 供 RAG 调用)
services.AddScoped<BuildFlow.Shared.Integration.IWorkflowTriggerService, WorkflowTriggerService>();

View File

@ -14,7 +14,6 @@ public class CreateBucketEndpoint(IMediator mediator) : Endpoint<CreateBucketReq
public override void Configure()
{
Post("/buckets");
Permissions("bucket:write");
}
public override async Task HandleAsync(CreateBucketRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class DeleteBucketEndpoint(IMediator mediator) : Endpoint<DeleteBucketReq
public override void Configure()
{
Delete("/buckets");
Permissions("bucket:write");
}
public override async Task HandleAsync(DeleteBucketRequest req, CancellationToken ct)

View File

@ -13,7 +13,6 @@ public class ListBucketsEndpoint(IMediator mediator) : Endpoint<EmptyRequest, Li
public override void Configure()
{
Get("/buckets");
Permissions("bucket:read");
}
public override async Task HandleAsync(EmptyRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class CheckFileEndpoint(IMediator mediator) : Endpoint<CheckFileRequest,
public override void Configure()
{
Post("/files/check");
Permissions("file:upload");
}
public override async Task HandleAsync(CheckFileRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class DeleteFileEndpoint(IMediator mediator) : Endpoint<DeleteFileRequest
public override void Configure()
{
Delete("/files/delete");
Permissions("file:write");
}
public override async Task HandleAsync(DeleteFileRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class GetFileContentEndpoint(IMediator mediator) : Endpoint<GetFileConten
public override void Configure()
{
Get("/files/content");
Permissions("file:read");
}
public override async Task HandleAsync(GetFileContentRequest req, CancellationToken ct)

View File

@ -17,7 +17,6 @@ public class GetOfficePreviewEndpoint(IMediator mediator) : Endpoint<GetOfficePr
public override void Configure()
{
Get("/files/office-preview");
Permissions("file:read");
}
public override async Task HandleAsync(GetOfficePreviewRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class ListFilesEndpoint(IMediator mediator) : Endpoint<ListFilesRequest,
public override void Configure()
{
Get("/files/list");
Permissions("file:read");
}
public override async Task HandleAsync(ListFilesRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class UploadFileEndpoint(IMediator mediator) : Endpoint<UploadFileRequest
public override void Configure()
{
Post("/files/upload");
Permissions("file:upload");
AllowFileUploads();
}

View File

@ -15,7 +15,6 @@ public class CreateFolderEndpoint(IMediator mediator) : Endpoint<CreateFolderReq
public override void Configure()
{
Post("/folders");
Permissions("file:write");
}
public override async Task HandleAsync(CreateFolderRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class DeleteFolderEndpoint(IMediator mediator) : Endpoint<DeleteFolderReq
public override void Configure()
{
Delete("/folders/{Id}");
Permissions("file:write");
}
public override async Task HandleAsync(DeleteFolderRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class GetFolderEndpoint(IMediator mediator) : Endpoint<GetFolderRequest,
public override void Configure()
{
Get("/folders/{Id}");
Permissions("file:read");
}
public override async Task HandleAsync(GetFolderRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class GetFolderTreeEndpoint(IMediator mediator) : Endpoint<EmptyRequest,
public override void Configure()
{
Get("/folders/tree");
Permissions("file:read");
}
public override async Task HandleAsync(EmptyRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class GetRootFilesEndpoint(IMediator mediator) : Endpoint<EmptyRequest, L
public override void Configure()
{
Get("/files/root");
Permissions("file:read");
}
public override async Task HandleAsync(EmptyRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class MoveFileEndpoint(IMediator mediator) : Endpoint<MoveFileRequest, ob
public override void Configure()
{
Post("/files/{Id}/move");
Permissions("file:write");
}
public override async Task HandleAsync(MoveFileRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class RenameFolderEndpoint(IMediator mediator) : Endpoint<RenameFolderReq
public override void Configure()
{
Put("/folders/{Id}");
Permissions("file:write");
}
public override async Task HandleAsync(RenameFolderRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class UploadToFolderEndpoint(IMediator mediator) : Endpoint<UploadToFolde
public override void Configure()
{
Post("/folders/{FolderId}/files");
Permissions("file:upload");
AllowFileUploads();
}

View File

@ -14,7 +14,6 @@ public class AbortMultipartEndpoint(IMediator mediator) : Endpoint<AbortMultipar
public override void Configure()
{
Post("/files/multipart/abort");
Permissions("file:upload");
}
public override async Task HandleAsync(AbortMultipartRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class CompleteMultipartEndpoint(IMediator mediator) : Endpoint<CompleteMu
public override void Configure()
{
Post("/files/multipart/complete");
Permissions("file:upload");
}
public override async Task HandleAsync(CompleteMultipartRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class InitMultipartEndpoint(IMediator mediator) : Endpoint<InitMultipartR
public override void Configure()
{
Post("/files/multipart/init");
Permissions("file:upload");
}
public override async Task HandleAsync(InitMultipartRequest req, CancellationToken ct)

View File

@ -17,7 +17,6 @@ public class ListPartsEndpoint(IMediator mediator) : Endpoint<ListPartsRequest,
public override void Configure()
{
Get("/files/multipart/parts");
Permissions("file:read");
}
public override async Task HandleAsync(ListPartsRequest req, CancellationToken ct)

View File

@ -15,7 +15,6 @@ public class UploadPartEndpoint(IMediator mediator) : Endpoint<UploadPartRequest
public override void Configure()
{
Put("/files/multipart/part");
Permissions("file:upload");
AllowFileUploads();
}

View File

@ -15,7 +15,6 @@ public class CreateShareEndpoint(IMediator mediator) : Endpoint<CreateShareReque
public override void Configure()
{
Post("/share");
Permissions("file-share:write");
}
public override async Task HandleAsync(CreateShareRequest req, CancellationToken ct)

View File

@ -14,7 +14,6 @@ public class DeleteShareEndpoint(IMediator mediator) : Endpoint<DeleteShareReque
public override void Configure()
{
Delete("/share/{Id}");
Permissions("file-share:write");
}
public override async Task HandleAsync(DeleteShareRequest req, CancellationToken ct)

View File

@ -0,0 +1,50 @@
using MediatR;
using BuildFlow.Platform.Application.Notifications.Commands;
using Workflow.Domain.Common;
namespace BuildFlow.Host.Services.Integration;
/// <summary>
/// 工作流通知 → 平台消息中心发布器。把工作流通知转发为 <see cref="PublishNotificationCommand"/>,
/// 由 platform 的 PublishNotificationCommandHandler 完成 fanout 落库与 SignalR 铃铛推送。
/// </summary>
/// <remarks>
/// 跨模块胶水代码:工作流模块的 <see cref="Workflow.Application.Notifications.NotificationService"/> 依赖
/// <see cref="IPlatformNotificationPublisher"/>(定义在 Workflow.Domain),但工作流模块无法直接访问
/// platform 模块的 Notification 聚合。本实现位于 Host 层,通过 MediatR 转发,复用 platform 已有链路。
/// 容错:转发失败不阻断工作流主链路(与 IImNotificationPublisher 同款吞异常降级策略)。
/// </remarks>
public class WorkflowNotificationPublisher(IMediator mediator) : IPlatformNotificationPublisher
{
public async Task<int> PublishAsync(
string type,
string title,
string content,
string source,
string? relatedId,
string? relatedType,
IEnumerable<Guid> recipientUserIds,
CancellationToken ct = default)
{
var recipients = recipientUserIds as IReadOnlyCollection<Guid> ?? 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;
}
}
}

View File

@ -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;
/// <summary>
/// 通知服务实现。处理任务生命周期事件:
/// - 站内信:同步落库 Notification(必达,随调用方事务提交)
/// - 平台消息中心转发:落库同时转发到平台 notifications 表,经 SignalR 铃铛实时推送给用户
/// - Webhook:落库 WebhookDelivery 标 pending,由 WebhookDispatcherService 异步投递
///
/// 接收人解析策略:优先 task.AssigneeId(user: 规则),其次 task.AssigneeRole(role: 规则)。
@ -15,7 +17,8 @@ namespace Workflow.Application.Notifications;
/// </summary>
public class NotificationService(
WorkflowDbContext db,
IOptions<NotificationOptions> options) : INotificationService
IOptions<NotificationOptions> 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);
}

View File

@ -0,0 +1,36 @@
namespace Workflow.Domain.Common;
/// <summary>
/// 平台消息中心通知发布能力。把工作流通知转发到平台消息中心(notifications 表 + SignalR 铃铛推送),
/// 复用 platform 的 fanout 落库与实时推送链路,避免工作流侧重复实现触达。
/// </summary>
/// <remarks>
/// 设计动机:工作流模块自带的 wf_notifications 表是孤立的私有通知日志,既不推 SignalR 也不进消息中心。
/// 本接口在工作流落 wf_notifications 的同时,把同一通知转发到平台消息中心,让用户在铃铛里看到流程通知。
/// 跨模块解耦:接口定义在工作流 Domain,实现在 Host 层(Host 同时引用 workflow 与 platform,通过 MediatR
/// 发 PublishNotificationCommand)。参照 IM 模块的 <c>IImNotificationPublisher</c> 同款模式。
/// </remarks>
public interface IPlatformNotificationPublisher
{
/// <summary>
/// 按接收人 userId 批量发布站内信通知到平台消息中心。
/// </summary>
/// <param name="type">通知类型,如 "workflow-task-arrived"</param>
/// <param name="title">通知标题</param>
/// <param name="content">通知正文</param>
/// <param name="source">来源标识,固定 "workflow"</param>
/// <param name="relatedId">关联实体 ID(流程实例/任务),用于点击跳转</param>
/// <param name="relatedType">关联实体类型,如 "workflow-instance"</param>
/// <param name="recipientUserIds">接收人 userId 列表</param>
/// <param name="ct">取消令牌</param>
/// <returns>实际发布的通知条数(platform 侧 fanout 落库数)</returns>
Task<int> PublishAsync(
string type,
string title,
string content,
string source,
string? relatedId,
string? relatedType,
IEnumerable<Guid> recipientUserIds,
CancellationToken ct = default);
}

View File

@ -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);
}
}
/// <summary>
/// 测试用 no-op 平台通知发布器:转发逻辑由 Host 层实现,单元测试无需真实推送,
/// 此实现空跑返回 0,避免 NotificationService 构造依赖阻断。
/// </summary>
file sealed class NoOpPlatformNotificationPublisher : IPlatformNotificationPublisher
{
public Task<int> PublishAsync(
string type, string title, string content, string source,
string? relatedId, string? relatedType,
IEnumerable<Guid> recipientUserIds, CancellationToken ct = default)
=> Task.FromResult(0);
}