From 8079dd0de131b94b9988f0a132e2efd0eed72cfc Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 20 Mar 2024 10:58:59 +0100 Subject: [PATCH] Add alteration notifications and management (#5097) * Add alteration notifications and management Implement notification publishing for completed alteration plans and add related handlers and management logic to coordinate alterations workflow triggering and completion status updates. * Refactor queue address configuration in MassTransit The initialization of the queue address and mapping for the `RunAlterationJob` has been moved out of the `Apply` method to the module configuration. * Refactor CompleteAlterationPlan activity description The activity summary comment in CompleteAlterationPlan.cs was updated to accurately describe its functionality. The description has been changed from "Submits an alteration plan for execution" to "Marks an alteration plan as completed." --- .../Contracts/IAlterationPlanManager.cs | 10 ++++ .../Notifications/AlterationPlanCompleted.cs | 9 ++++ .../Features/AlterationsMassTransitFeature.cs | 8 +-- .../Activities/CompleteAlterationPlan.cs | 48 ++++++++++++++++++ .../Activities/DispatchAlterationJobs.cs | 11 ++-- .../Activities/GenerateAlterationJobs.cs | 19 ++++--- .../Features/AlterationsFeature.cs | 2 +- .../Handlers/AlterationJobCompletedHandler.cs | 50 ++----------------- .../AlterationPlanCompletedHandler.cs | 26 ++++++++++ .../Services/AlterationPlanManager.cs | 43 ++++++++++++++++ .../Services/DefaultAlterationJobRunner.cs | 2 +- .../Services/DefaultAlterationRunner.cs | 2 +- .../ExecuteAlterationPlanWorkflow.cs | 14 ++++-- 13 files changed, 174 insertions(+), 70 deletions(-) create mode 100644 src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs create mode 100644 src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs create mode 100644 src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs create mode 100644 src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs create mode 100644 src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs diff --git a/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs new file mode 100644 index 000000000..79b72c6d9 --- /dev/null +++ b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs @@ -0,0 +1,10 @@ +using Elsa.Alterations.Core.Entities; + +namespace Elsa.Alterations.Core.Contracts; + +public interface IAlterationPlanManager +{ + Task GetPlanAsync(string planId, CancellationToken cancellationToken = default); + Task GetIsAllJobsCompletedAsync(string planId, CancellationToken cancellationToken = default); + Task CompletePlanAsync(AlterationPlan plan, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs b/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs new file mode 100644 index 000000000..70a13fb5b --- /dev/null +++ b/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs @@ -0,0 +1,9 @@ +using Elsa.Alterations.Core.Entities; +using Elsa.Mediator.Contracts; + +namespace Elsa.Alterations.Core.Notifications; + +/// +/// A notification that is published when an alteration plan is completed. +/// +public record AlterationPlanCompleted(AlterationPlan Plan) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs b/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs index b4faa0832..27cbaf6b6 100644 --- a/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs +++ b/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs @@ -28,15 +28,15 @@ public class MassTransitAlterationsFeature : FeatureBase public override void Configure() { Module.Configure(feature => feature.AlterationJobDispatcherFactory = sp => sp.GetRequiredService()); - Module.AddMassTransitConsumer(); + var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); + var queueAddress = new Uri($"queue:elsa-{queueName}"); + EndpointConvention.Map(queueAddress); + Module.AddMassTransitConsumer(queueName); } /// public override void Apply() { - var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); - var queueAddress = new Uri($"queue:elsa-{queueName}"); - EndpointConvention.Map(queueAddress); Services.AddScoped(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs b/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs new file mode 100644 index 000000000..c4b664e2e --- /dev/null +++ b/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs @@ -0,0 +1,48 @@ +using System.ComponentModel; +using System.Runtime.CompilerServices; +using Elsa.Alterations.Core.Contracts; +using Elsa.Workflows; +using Elsa.Workflows.Attributes; +using Elsa.Workflows.Exceptions; +using Elsa.Workflows.Memory; +using Elsa.Workflows.Models; + +namespace Elsa.Alterations.Activities; + +/// +/// Marks an alteration plan as completed. +/// +[Browsable(false)] +[Activity("Elsa", "Alterations", "Dispatches jobs for the specified Alteration Plan", Kind = ActivityKind.Job)] +public class CompleteAlterationPlan : CodeActivity +{ + /// + public CompleteAlterationPlan(Variable planId, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + { + PlanId = new Input(planId); + } + + /// + public CompleteAlterationPlan([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + { + } + + /// + /// The ID of the alteration plan. + /// + public Input PlanId { get; set; } = default!; + + /// + protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) + { + var cancellationToken = context.CancellationToken; + var planId = context.Get(PlanId)!; + var manager = context.GetRequiredService(); + var plan = await manager.GetPlanAsync(planId, cancellationToken); + + if (plan == null) + throw new FaultException($"Alteration Plan with ID {planId} not found."); + + await manager.CompletePlanAsync(plan, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs b/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs index ac5df8353..495d7ea3d 100644 --- a/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs +++ b/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs @@ -1,20 +1,13 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Alterations.Core.Contracts; -using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Core.Enums; using Elsa.Alterations.Core.Filters; -using Elsa.Common.Contracts; using Elsa.Workflows; using Elsa.Workflows.Attributes; -using Elsa.Workflows.Contracts; using Elsa.Workflows.Exceptions; -using Elsa.Workflows.Management.Contracts; -using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Memory; using Elsa.Workflows.Models; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Filters; namespace Elsa.Alterations.Activities; @@ -72,5 +65,9 @@ public class DispatchAlterationJobs : CodeActivity var alterationJobDispatcher = context.GetRequiredService(); foreach (var jobId in alterationJobIds) await alterationJobDispatcher.DispatchAsync(jobId, cancellationToken); + + // Update status. + plan.Status = AlterationPlanStatus.Running; + await alterationPlanStore.SaveAsync(plan, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs b/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs index 7aa4f7a65..6bf093267 100644 --- a/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs +++ b/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs @@ -6,6 +6,7 @@ using Elsa.Alterations.Core.Enums; using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Models; using Elsa.Common.Contracts; +using Elsa.Extensions; using Elsa.Workflows; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; @@ -24,7 +25,7 @@ namespace Elsa.Alterations.Activities; /// [Browsable(false)] [Activity("Elsa", "Alterations", "Generates jobs for the specified Alteration Plan", Kind = ActivityKind.Job)] -public class GenerateAlterationJobs : CodeActivity +public class GenerateAlterationJobs : CodeActivity { /// public GenerateAlterationJobs([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) @@ -47,10 +48,14 @@ public class GenerateAlterationJobs : CodeActivity { var plan = await GetPlanAsync(context); await UpdatePlanStatusAsync(context, plan); - var workflowInstanceIds = await FindMatchingWorkflowInstanceIdsAsync(context, plan.WorkflowInstanceFilter); - await GenerateJobsAsync(context, plan, workflowInstanceIds); + var workflowInstanceIds = (await FindMatchingWorkflowInstanceIdsAsync(context, plan.WorkflowInstanceFilter)).ToList(); + + if (workflowInstanceIds.Any()) + await GenerateJobsAsync(context, plan, workflowInstanceIds); + + context.SetResult(workflowInstanceIds.Count); } - + private async Task GetPlanAsync(ActivityExecutionContext context) { var cancellationToken = context.CancellationToken; @@ -64,7 +69,7 @@ public class GenerateAlterationJobs : CodeActivity if (plan == null) throw new FaultException($"Alteration Plan with ID {planId} not found."); - + return plan; } @@ -108,10 +113,10 @@ public class GenerateAlterationJobs : CodeActivity workflowInstanceIds.UnionWith(matchingWorkflowInstanceIds); } } - + return workflowInstanceIds; } - + private async Task GenerateJobsAsync(ActivityExecutionContext context, AlterationPlan plan, IEnumerable workflowInstanceIds) { var cancellationToken = context.CancellationToken; diff --git a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs index d33062de7..a4588ee27 100644 --- a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs +++ b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs @@ -1,7 +1,6 @@ using Elsa.Alterations.Core.Contracts; using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Core.Extensions; -using Elsa.Alterations.Core.Services; using Elsa.Alterations.Core.Stores; using Elsa.Alterations.Extensions; using Elsa.Alterations.Services; @@ -60,6 +59,7 @@ public class AlterationsFeature : FeatureBase /// public override void Apply() { + Services.AddScoped(); Services.AddAlterations(); Services.AddAlterationsCore(); Services.AddScoped(); diff --git a/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs index 37313e880..e9b2758e5 100644 --- a/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs +++ b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs @@ -1,14 +1,6 @@ -using Elsa.Alterations.Activities; -using Elsa.Alterations.Bookmarks; using Elsa.Alterations.Core.Contracts; -using Elsa.Alterations.Core.Enums; -using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Notifications; -using Elsa.Common.Contracts; using Elsa.Mediator.Contracts; -using Elsa.Workflows.Helpers; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Requests; using JetBrains.Annotations; namespace Elsa.Alterations.Handlers; @@ -17,53 +9,19 @@ namespace Elsa.Alterations.Handlers; /// Handles notifications and updates the plan status if all jobs are completed. /// [UsedImplicitly] -public class AlterationJobCompletedHandler : INotificationHandler +public class AlterationJobCompletedHandler(IAlterationPlanManager manager) : INotificationHandler { - private readonly IWorkflowDispatcher _workflowDispatcher; - private readonly IAlterationPlanStore _alterationPlanStore; - private readonly IAlterationJobStore _alterationJobStore; - private readonly ISystemClock _systemClock; - - /// - /// Initializes a new instance of the class. - /// - public AlterationJobCompletedHandler(IWorkflowDispatcher workflowDispatcher, IAlterationPlanStore alterationPlanStore, IAlterationJobStore alterationJobStore, ISystemClock systemClock) - { - _workflowDispatcher = workflowDispatcher; - _alterationPlanStore = alterationPlanStore; - _alterationJobStore = alterationJobStore; - _systemClock = systemClock; - } - /// public async Task HandleAsync(AlterationJobCompleted notification, CancellationToken cancellationToken) { var job = notification.Job; var planId = job.PlanId; - var planFilter = new AlterationPlanFilter { Id = planId }; - var plan = (await _alterationPlanStore.FindAsync(planFilter, cancellationToken))!; - - // Check if all jobs are completed. - var jobFilter = new AlterationJobFilter - { - PlanId = planId, - Statuses = new[] { AlterationJobStatus.Pending, AlterationJobStatus.Running } - }; - - var allJobsCompleted = await _alterationJobStore.CountAsync(jobFilter, cancellationToken) == 0; + var plan = (await manager.GetPlanAsync(planId, cancellationToken))!; + var allJobsCompleted = await manager.GetIsAllJobsCompletedAsync(planId, cancellationToken); if(!allJobsCompleted) return; - // Update plan status. - plan.Status = AlterationPlanStatus.Completed; - plan.CompletedAt = _systemClock.UtcNow; - - await _alterationPlanStore.SaveAsync(plan, cancellationToken); - - // Trigger any workflow instances that are waiting for the plan to complete. - var bookmarkPayload = new AlterationPlanCompletedPayload(planId); - var triggerRequest = new DispatchTriggerWorkflowsRequest(ActivityTypeNameHelper.GenerateTypeName(), bookmarkPayload); - await _workflowDispatcher.DispatchAsync(triggerRequest, cancellationToken); + await manager.CompletePlanAsync(plan, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs new file mode 100644 index 000000000..2657cec3c --- /dev/null +++ b/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs @@ -0,0 +1,26 @@ +using Elsa.Alterations.Bookmarks; +using Elsa.Alterations.Core.Notifications; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Helpers; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Requests; +using JetBrains.Annotations; + +namespace Elsa.Alterations.Handlers; + +/// +/// Handles notifications and triggers any workflows that are waiting for the plan to complete. +/// +[UsedImplicitly] +public class AlterationPlanCompletedHandler(IWorkflowDispatcher workflowDispatcher) : INotificationHandler +{ + /// + public async Task HandleAsync(AlterationPlanCompleted notification, CancellationToken cancellationToken) + { + // Trigger any workflow instances that are waiting for the plan to complete. + var planId = notification.Plan.Id; + var bookmarkPayload = new AlterationPlanCompletedPayload(planId); + var triggerRequest = new DispatchTriggerWorkflowsRequest(ActivityTypeNameHelper.GenerateTypeName(), bookmarkPayload); + await workflowDispatcher.DispatchAsync(triggerRequest, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs b/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs new file mode 100644 index 000000000..77d9ba6f5 --- /dev/null +++ b/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs @@ -0,0 +1,43 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Core.Enums; +using Elsa.Alterations.Core.Filters; +using Elsa.Alterations.Core.Notifications; +using Elsa.Common.Contracts; +using Elsa.Mediator.Contracts; + +namespace Elsa.Alterations.Services; + +/// +public class AlterationPlanManager(IAlterationPlanStore planStore, IAlterationJobStore jobStore, ISystemClock systemClock, INotificationSender notificationSender) : IAlterationPlanManager +{ + /// + public async Task GetPlanAsync(string planId, CancellationToken cancellationToken = default) + { + var planFilter = new AlterationPlanFilter { Id = planId }; + return await planStore.FindAsync(planFilter, cancellationToken); + } + + /// + public async Task GetIsAllJobsCompletedAsync(string planId, CancellationToken cancellationToken = default) + { + // Check if all jobs are completed. + var jobFilter = new AlterationJobFilter + { + PlanId = planId, + Statuses = new[] { AlterationJobStatus.Pending, AlterationJobStatus.Running } + }; + + return await jobStore.CountAsync(jobFilter, cancellationToken) == 0; + } + + /// + public async Task CompletePlanAsync(AlterationPlan plan, CancellationToken cancellationToken = default) + { + plan.Status = AlterationPlanStatus.Completed; + plan.CompletedAt = systemClock.UtcNow; + + await planStore.SaveAsync(plan, cancellationToken); + await notificationSender.SendAsync(new AlterationPlanCompleted(plan), cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs b/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs index 36637c384..604baf9c0 100644 --- a/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs +++ b/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs @@ -6,7 +6,7 @@ using Elsa.Alterations.Core.Notifications; using Elsa.Common.Contracts; using Elsa.Mediator.Contracts; -namespace Elsa.Alterations.Core.Services; +namespace Elsa.Alterations.Services; /// public class DefaultAlterationJobRunner : IAlterationJobRunner diff --git a/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs b/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs index 56a5a269c..5cf2df277 100644 --- a/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs +++ b/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs @@ -9,7 +9,7 @@ using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Runtime.Contracts; using Microsoft.Extensions.Logging; -namespace Elsa.Alterations.Core.Services; +namespace Elsa.Alterations.Services; /// public class DefaultAlterationRunner : IAlterationRunner diff --git a/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs b/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs index dd4b83c1c..cd74511a4 100644 --- a/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs +++ b/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs @@ -13,7 +13,7 @@ namespace Elsa.Alterations.Workflows; public class ExecuteAlterationPlanWorkflow : WorkflowBase { internal const string WorkflowDefinitionId = "Elsa.Alterations.ExecuteAlterationPlan"; - + /// protected override void Build(IWorkflowBuilder builder) { @@ -21,6 +21,7 @@ public class ExecuteAlterationPlanWorkflow : WorkflowBase builder.AsSystemWorkflow(); var plan = builder.WithInput("Plan", "The parameters for the new plan"); var planId = builder.WithVariable(); + var jobCount = builder.WithVariable(); builder.Root = new Sequence { @@ -32,8 +33,15 @@ public class ExecuteAlterationPlanWorkflow : WorkflowBase Result = new(planId) }, new Correlate(planId), - new GenerateAlterationJobs(planId), - new DispatchAlterationJobs(planId), + new GenerateAlterationJobs(planId) + { + Result = new(jobCount) + }, + new If(context => jobCount.Get(context) > 0) + { + Then = new DispatchAlterationJobs(planId), + Else = new CompleteAlterationPlan(planId) + }, new AlterationPlanCompleted(planId), } };