From 6110eb561f03b4043a7e9ba78a40874a10ecb36f Mon Sep 17 00:00:00 2001 From: MariusVuscanNx <96233009+MariusVuscanNx@users.noreply.github.com> Date: Fri, 5 Jul 2024 19:35:22 +0300 Subject: [PATCH] Workflow refresh feature (#5713) * Refactor workflow definition events handling and dispatch Removed IDistributedWorkflowDefinitionEventsDispatcher interface and related classes in favor of direct message publishing through MassTransit. Introduced new workflow definition refresh endpoints and services for better modularity and responsiveness. Adjusted caching and notification systems to support the new architecture. * Add comment * Added workflow definition refresh component test * Fixed component tests issue * Fixed id * Fixed test * Rename IsRaisedFromConsumer to IsConsumerExecutionContext Updated the naming for clarity and consistency across the system. This change affects AmbientConsumerScope, related consumers, and condition checks to improve readability and understanding of the consumer context. --------- Co-authored-by: Sipke Schoorstra --- src/bundles/Elsa.Server.Web/Program.cs | 2 +- .../WorkflowDefinitionEventsConsumer.cs | 43 ++++---- ...butedWorkflowDefinitionEventsDispatcher.cs | 49 --------- .../MassTransitWorkflowManagementFeature.cs | 6 - ...dWorkflowDefinitionNotificationsHandler.cs | 70 ++++++++---- .../WorkflowDefinitionVersionsUpdated.cs | 4 - .../Messages/WorkflowDefinitionsRefreshed.cs | 8 ++ .../Services/AmbientConsumerScope.cs | 14 +++ .../MassTransitDistributedEventsDispatcher.cs | 59 ---------- .../WorkflowDefinitions/Refresh/Endpoint.cs | 30 +++++ .../WorkflowDefinitions/Refresh/Models.cs | 6 + .../Contracts/IWorkflowDefinitionRefresher.cs | 11 ++ .../Features/CachingWorkflowRuntimeFeature.cs | 11 +- .../Features/WorkflowRuntimeFeature.cs | 1 + .../Handlers/InvalidateTriggersCache.cs | 25 +++++ .../WorkflowDefinitionsRefreshed.cs | 6 + .../Notifications/WorkflowTriggersIndexed.cs | 2 - .../RefreshWorkflowDefinitionsRequest.cs | 25 +++++ .../RefreshWorkflowDefinitionsResponse.cs | 6 + .../Services/TriggerIndexer.cs | 2 +- .../Services/WorkflowDefinitionRefresher.cs | 55 +++++++++ .../Stores/CachingTriggerStore.cs | 3 +- .../Elsa.Workflows.ComponentTests.csproj | 13 ++- .../Helpers/Fixtures/WorkflowServer.cs | 9 ++ .../Helpers/Services/StaticValueHolder.cs | 6 + .../DynamicEndpointTests.cs | 36 ++++++ .../dynamic-endpoint-http-workflow.json | 104 ++++++++++++++++++ 27 files changed, 433 insertions(+), 173 deletions(-) delete mode 100644 src/modules/Elsa.MassTransit/Contracts/IDistributedWorkflowDefinitionEventsDispatcher.cs create mode 100644 src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsRefreshed.cs create mode 100644 src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs delete mode 100644 src/modules/Elsa.MassTransit/Services/MassTransitDistributedEventsDispatcher.cs create mode 100644 src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs create mode 100644 src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionRefresher.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsRefreshed.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/RefreshWorkflowDefinitionsRequest.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Helpers/Services/StaticValueHolder.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/dynamic-endpoint-http-workflow.json diff --git a/src/bundles/Elsa.Server.Web/Program.cs b/src/bundles/Elsa.Server.Web/Program.cs index 9c41549da..dd1c5475b 100644 --- a/src/bundles/Elsa.Server.Web/Program.cs +++ b/src/bundles/Elsa.Server.Web/Program.cs @@ -49,7 +49,7 @@ const bool useMemoryStores = false; const bool useCaching = true; const bool useReadOnlyMode = false; const DistributedCachingTransport distributedCachingTransport = DistributedCachingTransport.MassTransit; -const MassTransitBroker useMassTransitBroker = MassTransitBroker.Memory; +const MassTransitBroker useMassTransitBroker = MassTransitBroker.RabbitMq; var builder = WebApplication.CreateBuilder(args); var services = builder.Services; diff --git a/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs b/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs index ce3757c72..6fed5a9a5 100644 --- a/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs +++ b/src/modules/Elsa.MassTransit/Consumers/WorkflowDefinitionEventsConsumer.cs @@ -1,27 +1,28 @@ using Elsa.MassTransit.Messages; +using Elsa.MassTransit.Services; +using Elsa.Mediator.Contracts; using Elsa.Workflows.Management.Contracts; using JetBrains.Annotations; using MassTransit; namespace Elsa.MassTransit.Consumers; -/// /// Consumes messages related to workflow definition changes. -/// [PublicAPI] -public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistryUpdater workflowDefinitionActivityRegistryUpdater) : - IConsumer, - IConsumer, - IConsumer, - IConsumer, - IConsumer, - IConsumer, - IConsumer, - IConsumer +public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistryUpdater workflowDefinitionActivityRegistryUpdater, INotificationSender notificationSender) : + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer, + global::MassTransit.IConsumer { /// public Task Consume(ConsumeContext context) - { + { workflowDefinitionActivityRegistryUpdater.RemoveDefinitionFromRegistry(context.Message.Id); return Task.CompletedTask; } @@ -48,9 +49,7 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr public Task Consume(ConsumeContext context) { foreach (var id in context.Message.Ids) - { workflowDefinitionActivityRegistryUpdater.RemoveDefinitionFromRegistry(id); - } return Task.CompletedTask; } @@ -66,9 +65,7 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr public Task Consume(ConsumeContext context) { foreach (var id in context.Message.Ids) - { workflowDefinitionActivityRegistryUpdater.RemoveDefinitionVersionFromRegistry(id); - } return Task.CompletedTask; } @@ -77,14 +74,22 @@ public class WorkflowDefinitionEventsConsumer(IWorkflowDefinitionActivityRegistr public async Task Consume(ConsumeContext context) { foreach (WorkflowDefinitionVersionUpdate definitionUpdate in context.Message.WorkflowDefinitionVersionUpdates) - { await UpdateDefinition(definitionUpdate.Id, definitionUpdate.UsableAsActivity); - } + } + + /// + public async Task Consume(ConsumeContext context) + { + var message = context.Message; + var notification = new Elsa.Workflows.Runtime.Notifications.WorkflowDefinitionsRefreshed(message.WorkflowDefinitionIds); + AmbientConsumerScope.IsConsumerExecutionContext = true; + await notificationSender.SendAsync(notification, context.CancellationToken); + AmbientConsumerScope.IsConsumerExecutionContext = false; } private Task UpdateDefinition(string id, bool usableAsActivity) { - // Once a workflow has been published it should remain in the activity registry unless no longer being marked as an activity. + // Once a workflow has been published, it should remain in the activity registry unless no longer being marked as an activity. if (usableAsActivity) return workflowDefinitionActivityRegistryUpdater.AddToRegistry(id); diff --git a/src/modules/Elsa.MassTransit/Contracts/IDistributedWorkflowDefinitionEventsDispatcher.cs b/src/modules/Elsa.MassTransit/Contracts/IDistributedWorkflowDefinitionEventsDispatcher.cs deleted file mode 100644 index 46a670eec..000000000 --- a/src/modules/Elsa.MassTransit/Contracts/IDistributedWorkflowDefinitionEventsDispatcher.cs +++ /dev/null @@ -1,49 +0,0 @@ -using Elsa.MassTransit.Messages; - -namespace Elsa.MassTransit.Contracts; - -/// -/// Dispatches workflow definition related notifications to refresh the activity registry. -/// -public interface IDistributedWorkflowDefinitionEventsDispatcher -{ - /// - /// Dispatches a workflow definition publication event. - /// - Task DispatchAsync(WorkflowDefinitionPublished request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition retracted event. - /// - Task DispatchAsync(WorkflowDefinitionRetracted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition version retracted event. - /// - Task DispatchAsync(WorkflowDefinitionVersionRetracted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition deleted event. - /// - Task DispatchAsync(WorkflowDefinitionDeleted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definitions deleted event. - /// - Task DispatchAsync(WorkflowDefinitionsDeleted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition version deleted event. - /// - Task DispatchAsync(WorkflowDefinitionVersionDeleted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition versions deleted event. - /// - Task DispatchAsync(WorkflowDefinitionVersionsDeleted request, CancellationToken cancellationToken); - - /// - /// Dispatches a workflow definition versions updated event. - /// - Task DispatchAsync(WorkflowDefinitionVersionsUpdated request, CancellationToken cancellationToken); -} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowManagementFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowManagementFeature.cs index 837c53ecd..77e8eca61 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowManagementFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowManagementFeature.cs @@ -1,11 +1,8 @@ -using System.Diagnostics.CodeAnalysis; using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.MassTransit.Consumers; -using Elsa.MassTransit.Contracts; -using Elsa.MassTransit.Services; using Elsa.Workflows.Management.Features; using Microsoft.Extensions.DependencyInjection; @@ -19,7 +16,6 @@ namespace Elsa.MassTransit.Features; public class MassTransitWorkflowManagementFeature(IModule module) : FeatureBase(module) { /// - [RequiresUnreferencedCode("The assembly containing the specified marker type will be scanned for activity types.")] public override void Configure() { Module.AddMassTransitConsumer("elsa-workflow-definition-updates", true, true); @@ -28,8 +24,6 @@ public class MassTransitWorkflowManagementFeature(IModule module) : FeatureBase( /// public override void Apply() { - Services.AddScoped(); - Services.AddNotificationHandlersFrom(GetType()); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs b/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs index 1c2f38774..12fdef392 100644 --- a/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs +++ b/src/modules/Elsa.MassTransit/Handlers/DistributedWorkflowDefinitionNotificationsHandler.cs @@ -1,12 +1,15 @@ using Elsa.MassTransit.Contracts; +using Elsa.MassTransit.Services; using Elsa.Mediator.Contracts; using Elsa.Workflows.Management.Notifications; +using Elsa.Workflows.Runtime.Notifications; +using MassTransit; using Distributed = Elsa.MassTransit.Messages; namespace Elsa.MassTransit.Handlers; /// Represents a handler for distributed workflow definition notifications. -public class DistributedWorkflowDefinitionNotificationsHandler(IDistributedWorkflowDefinitionEventsDispatcher distributedEventsDispatcher) : +public class DistributedWorkflowDefinitionNotificationsHandler(IBus bus) : INotificationHandler, INotificationHandler, INotificationHandler, @@ -14,44 +17,67 @@ public class DistributedWorkflowDefinitionNotificationsHandler(IDistributedWorkf INotificationHandler, INotificationHandler, INotificationHandler, - INotificationHandler + INotificationHandler, + INotificationHandler { /// - public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionPublished(notification.WorkflowDefinition.Id, - notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + public Task HandleAsync(WorkflowDefinitionPublished notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionPublished(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionRetracted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionRetracted(notification.WorkflowDefinition.Id, - notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + public Task HandleAsync(WorkflowDefinitionRetracted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionRetracted(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionVersionRetracted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionRetracted(notification.WorkflowDefinition.Id, - notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + public Task HandleAsync(WorkflowDefinitionVersionRetracted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionVersionRetracted(notification.WorkflowDefinition.Id, notification.WorkflowDefinition.Options.UsableAsActivity.GetValueOrDefault()), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionDeleted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionDeleted(notification.DefinitionId), cancellationToken); + public Task HandleAsync(WorkflowDefinitionDeleted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionDeleted(notification.DefinitionId), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionsDeleted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionsDeleted(notification.DefinitionIds), cancellationToken); + public Task HandleAsync(WorkflowDefinitionsDeleted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionsDeleted(notification.DefinitionIds), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionVersionDeleted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionDeleted(notification.WorkflowDefinition.Id), cancellationToken); + public Task HandleAsync(WorkflowDefinitionVersionDeleted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionVersionDeleted(notification.WorkflowDefinition.Id), cancellationToken); + } /// - public Task HandleAsync(WorkflowDefinitionVersionsDeleted notification, CancellationToken cancellationToken) => - distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionsDeleted(notification.Ids), cancellationToken); + public Task HandleAsync(WorkflowDefinitionVersionsDeleted notification, CancellationToken cancellationToken) + { + return bus.Publish(new Distributed.WorkflowDefinitionVersionsDeleted(notification.Ids), cancellationToken); + } /// public async Task HandleAsync(WorkflowDefinitionVersionsUpdated notification, CancellationToken cancellationToken) { - var updates = notification.WorkflowDefinitions.Select(x => - new Distributed.WorkflowDefinitionVersionUpdate(x.Id, x.DefinitionId, x.IsPublished, x.Options.UsableAsActivity.GetValueOrDefault())); - await distributedEventsDispatcher.DispatchAsync(new Distributed.WorkflowDefinitionVersionsUpdated(updates), cancellationToken); + var updates = notification.WorkflowDefinitions.Select(x => new Distributed.WorkflowDefinitionVersionUpdate(x.Id, x.DefinitionId, x.IsPublished, x.Options.UsableAsActivity.GetValueOrDefault())); + await bus.Publish(new Distributed.WorkflowDefinitionVersionsUpdated(updates), cancellationToken); + } + + /// + public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken) + { + // Prevent re-entrance. + if (AmbientConsumerScope.IsConsumerExecutionContext) + return Task.CompletedTask; + + var definitionIds = notification.WorkflowDefinitionIds; + var message = new Distributed.WorkflowDefinitionsRefreshed(definitionIds); + return bus.Publish(message, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionVersionsUpdated.cs b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionVersionsUpdated.cs index 014100ee3..2a939cb98 100644 --- a/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionVersionsUpdated.cs +++ b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionVersionsUpdated.cs @@ -1,12 +1,8 @@ namespace Elsa.MassTransit.Messages; -/// /// Represents a distributed message that multiple workflow definition versions have been deleted. -/// public class WorkflowDefinitionVersionsUpdated(IEnumerable workflowDefinitionVersionUpdates) { - /// /// Represents a collection of updates to workflow definition versions. - /// public IEnumerable WorkflowDefinitionVersionUpdates { get; set; } = workflowDefinitionVersionUpdates; } \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsRefreshed.cs b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsRefreshed.cs new file mode 100644 index 000000000..728b89f6b --- /dev/null +++ b/src/modules/Elsa.MassTransit/Messages/WorkflowDefinitionsRefreshed.cs @@ -0,0 +1,8 @@ +namespace Elsa.MassTransit.Messages; + +/// Represents a message that indicates that the specified workflow definitions have been refreshed. +public class WorkflowDefinitionsRefreshed(ICollection workflowDefinitionIds) +{ + /// The workflow definition IDs that have been refreshed. + public ICollection WorkflowDefinitionIds { get; set; } = workflowDefinitionIds; +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs b/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs new file mode 100644 index 000000000..2236cd969 --- /dev/null +++ b/src/modules/Elsa.MassTransit/Services/AmbientConsumerScope.cs @@ -0,0 +1,14 @@ +namespace Elsa.MassTransit.Services; + +/// Holds ambient state about a consumer. +public static class AmbientConsumerScope +{ + private static readonly AsyncLocal IsRaisedFromConsumerState = new(); + + /// Gets or sets a value that indicates if the current code is initiated from a consumer. + public static bool IsConsumerExecutionContext + { + get => IsRaisedFromConsumerState.Value; + set => IsRaisedFromConsumerState.Value = value; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit/Services/MassTransitDistributedEventsDispatcher.cs b/src/modules/Elsa.MassTransit/Services/MassTransitDistributedEventsDispatcher.cs deleted file mode 100644 index 9ea97cf70..000000000 --- a/src/modules/Elsa.MassTransit/Services/MassTransitDistributedEventsDispatcher.cs +++ /dev/null @@ -1,59 +0,0 @@ -using Elsa.MassTransit.Contracts; -using Elsa.MassTransit.Messages; -using MassTransit; - -namespace Elsa.MassTransit.Services; - -/// -/// Dispatches workflow definition related notifications via MassTransit. -/// -public class MassTransitDistributedEventsDispatcher(IBus bus) : IDistributedWorkflowDefinitionEventsDispatcher -{ - /// - public Task DispatchAsync(WorkflowDefinitionPublished request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionRetracted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionVersionRetracted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionDeleted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionsDeleted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionVersionDeleted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionVersionsDeleted request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } - - /// - public Task DispatchAsync(WorkflowDefinitionVersionsUpdated request, CancellationToken cancellationToken) - { - return bus.Publish(request, cancellationToken); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs new file mode 100644 index 000000000..752102397 --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Endpoint.cs @@ -0,0 +1,30 @@ +using Elsa.Abstractions; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Requests; +using JetBrains.Annotations; + +namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh; + +[PublicAPI] +internal class Refresh(IWorkflowDefinitionsRefresher workflowDefinitionsRefresher) : ElsaEndpoint +{ + private const int BatchSize = 10; + + public override void Configure() + { + Post("/actions/workflow-definitions/refresh"); + ConfigurePermissions("actions:workflow-definitions:refresh"); + } + + public override async Task HandleAsync(Request request, CancellationToken cancellationToken) + { + await RefreshWorkflowDefinitionsAsync(request.DefinitionIds, cancellationToken); + await SendOkAsync(cancellationToken); + } + + private async Task RefreshWorkflowDefinitionsAsync(ICollection? definitionIds, CancellationToken cancellationToken) + { + var request = new RefreshWorkflowDefinitionsRequest(definitionIds, BatchSize); + await workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(request, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs new file mode 100644 index 000000000..c5970467b --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Refresh/Models.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Refresh; + +public class Request +{ + public ICollection? DefinitionIds { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionRefresher.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionRefresher.cs new file mode 100644 index 000000000..579c6babb --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowDefinitionRefresher.cs @@ -0,0 +1,11 @@ +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; + +namespace Elsa.Workflows.Runtime.Contracts; + +/// Refreshes all workflows by re-indexing their triggers. +public interface IWorkflowDefinitionsRefresher +{ + /// Refreshes all workflows by re-indexing their triggers. + Task RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/CachingWorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/CachingWorkflowRuntimeFeature.cs index fc521f9ed..9c0e26e73 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/CachingWorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/CachingWorkflowRuntimeFeature.cs @@ -3,14 +3,13 @@ using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Handlers; using Elsa.Workflows.Runtime.Stores; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.Features; -/// /// Installs and configures workflow runtime caching features. -/// [DependsOn(typeof(MemoryCacheFeature))] public class CachingWorkflowRuntimeFeature : FeatureBase { @@ -22,7 +21,11 @@ public class CachingWorkflowRuntimeFeature : FeatureBase /// public override void Apply() { - // Decorators. - Services.Decorate(); + Services + // Decorators. + .Decorate() + + // Handlers. + .AddNotificationHandler(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 827c3920f..f27221b89 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -220,6 +220,7 @@ public class WorkflowRuntimeFeature : FeatureBase .AddScoped() .AddScoped() .AddScoped() + .AddScoped() .AddScoped() .AddScoped() .AddScoped() diff --git a/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs new file mode 100644 index 000000000..e0347168a --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Handlers/InvalidateTriggersCache.cs @@ -0,0 +1,25 @@ +using Elsa.Caching; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Runtime.Notifications; +using Elsa.Workflows.Runtime.Stores; +using JetBrains.Annotations; + +namespace Elsa.Workflows.Runtime.Handlers; + +/// +/// A notification handler that invalidates triggers cache when workflow definitions are refreshed. +/// +/// +/// The class implements the INotificationHandler interface and is responsible for handling WorkflowDefinitionsRefreshed notifications. +/// When a WorkflowDefinitionsRefreshed notification is received, the HandleAsync method is called to invalidate the triggers cache. +/// The cache is invalidated by calling the TriggerTokenAsync method of the ICacheManager passed to the class constructor. +/// +[UsedImplicitly] +public class InvalidateTriggersCache(ICacheManager cacheManager) : INotificationHandler +{ + /// + public Task HandleAsync(WorkflowDefinitionsRefreshed notification, CancellationToken cancellationToken) + { + return cacheManager.TriggerTokenAsync(CachingTriggerStore.CacheInvalidationTokenKey, cancellationToken).AsTask(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsRefreshed.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsRefreshed.cs new file mode 100644 index 000000000..ed8f5d2ac --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowDefinitionsRefreshed.cs @@ -0,0 +1,6 @@ +using Elsa.Mediator.Contracts; + +namespace Elsa.Workflows.Runtime.Notifications; + +/// Published when workflow definitions have been refreshed. +public record WorkflowDefinitionsRefreshed(ICollection WorkflowDefinitionIds) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowTriggersIndexed.cs b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowTriggersIndexed.cs index ec1e03f31..d9e10dda6 100644 --- a/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowTriggersIndexed.cs +++ b/src/modules/Elsa.Workflows.Runtime/Notifications/WorkflowTriggersIndexed.cs @@ -2,8 +2,6 @@ using Elsa.Mediator.Contracts; namespace Elsa.Workflows.Runtime.Notifications; -/// /// Published when the specified workflow's triggers have been indexed. -/// /// Contains the workflow that was indexed and the resulting triggers. public record WorkflowTriggersIndexed(IndexedWorkflowTriggers IndexedWorkflowTriggers) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/RefreshWorkflowDefinitionsRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/RefreshWorkflowDefinitionsRequest.cs new file mode 100644 index 000000000..acd75276f --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Requests/RefreshWorkflowDefinitionsRequest.cs @@ -0,0 +1,25 @@ +namespace Elsa.Workflows.Runtime.Requests; + +/// Represents a request to refresh workflow definitions by re-indexing their triggers. +public class RefreshWorkflowDefinitionsRequest +{ + /// Initializes a new instance of . + public RefreshWorkflowDefinitionsRequest() + { + } + + /// Initializes a new instance of . + public RefreshWorkflowDefinitionsRequest(IEnumerable? definitionIds, int batchSize) + { + BatchSize = batchSize; + DefinitionIds = definitionIds?.ToList(); + } + + /// Gets or sets the batch size for refreshing workflow definitions. + /// The batch size determines the number of definitions to be processed at a time during the refresh process. + public int BatchSize { get; set; } = 10; + + /// Gets or sets the collection of definition IDs for workflow definitions to be refreshed. + /// If not specified, all workflows will be refreshed. + public ICollection? DefinitionIds { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs new file mode 100644 index 000000000..36cfc5360 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Responses/RefreshWorkflowDefinitionsResponse.cs @@ -0,0 +1,6 @@ +using Elsa.Workflows.Management.Entities; + +namespace Elsa.Workflows.Runtime.Responses; + +/// Represents a response to a request to refresh workflow definitions. +public record RefreshWorkflowDefinitionsResponse(ICollection WorkflowDefinitions); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs b/src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs index 57eb30aa5..eea344713 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs @@ -86,7 +86,7 @@ public class TriggerIndexer : ITriggerIndexer // Get current triggers var currentTriggers = await GetCurrentTriggersAsync(workflow.Identity.DefinitionId, cancellationToken).ToList(); - // Collect new triggers **if workflow is published**. + // Collect new triggers **if the workflow is published**. var newTriggers = workflow.Publication.IsPublished ? await GetTriggersInternalAsync(workflow, cancellationToken).ToListAsync(cancellationToken) : new List(0); diff --git a/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs new file mode 100644 index 000000000..39eb938e1 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/WorkflowDefinitionRefresher.cs @@ -0,0 +1,55 @@ +using Elsa.Common.Entities; +using Elsa.Common.Models; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Management.Contracts; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Notifications; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Responses; + +namespace Elsa.Workflows.Runtime.Services; + +/// +public class WorkflowDefinitionsRefresher(IWorkflowDefinitionStore store, ITriggerIndexer triggerIndexer, INotificationSender notificationSender) : IWorkflowDefinitionsRefresher +{ + /// + public async Task RefreshWorkflowDefinitionsAsync(RefreshWorkflowDefinitionsRequest request, CancellationToken cancellationToken) + { + var filter = new WorkflowDefinitionFilter + { + DefinitionIds = request.DefinitionIds, + VersionOptions = VersionOptions.Published + }; + + var currentPage = 0; + var processedWorkflowDefinitions = new List(); + var batchSize = request.BatchSize; + var order = new WorkflowDefinitionOrder(x => x.Id, OrderDirection.Ascending); + + while (!cancellationToken.IsCancellationRequested) + { + var pageArgs = PageArgs.FromPage(currentPage, batchSize); + var definitions = await store.FindManyAsync(filter, order, pageArgs, cancellationToken); + + if (definitions.Items.Count == 0) + break; + + await IndexWorkflowTriggersAsync(definitions.Items, cancellationToken); + processedWorkflowDefinitions.AddRange(definitions.Items); + currentPage++; + } + + var processedWorkflowDefinitionIds = processedWorkflowDefinitions.Select(x => x.Id).ToList(); + var notification = new WorkflowDefinitionsRefreshed(processedWorkflowDefinitionIds); + await notificationSender.SendAsync(notification, cancellationToken); + return new RefreshWorkflowDefinitionsResponse(processedWorkflowDefinitions); + } + + private async Task IndexWorkflowTriggersAsync(IEnumerable definitions, CancellationToken cancellationToken) + { + foreach (var definition in definitions) + await triggerIndexer.IndexTriggersAsync(definition, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs b/src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs index 0ebba4a6e..8490bfe05 100644 --- a/src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs +++ b/src/modules/Elsa.Workflows.Runtime/Stores/CachingTriggerStore.cs @@ -12,7 +12,8 @@ namespace Elsa.Workflows.Runtime.Stores; /// public class CachingTriggerStore(ITriggerStore decoratedStore, ICacheManager cacheManager, IHasher hasher) : ITriggerStore { - private static readonly string CacheInvalidationTokenKey = typeof(CachingTriggerStore).FullName!; + /// The token key to evict all cached triggers. + public static readonly string CacheInvalidationTokenKey = typeof(CachingTriggerStore).FullName!; /// public async ValueTask SaveAsync(StoredTrigger record, CancellationToken cancellationToken = default) diff --git a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj index 0f20fc512..7d745b355 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj +++ b/test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj @@ -1,4 +1,4 @@ - + net8.0 @@ -10,13 +10,13 @@ - + - - - + + + @@ -72,6 +72,9 @@ Always + + Always + diff --git a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs index 49464d7a9..aaaa665b4 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs @@ -6,6 +6,7 @@ using Elsa.Extensions; using Elsa.Identity.Providers; using Elsa.MassTransit.Extensions; using Elsa.Workflows.ComponentTests.Consumers; +using Elsa.Workflows.ComponentTests.Helpers.Services; using Elsa.Workflows.ComponentTests.Services; using FluentStorage; using Hangfire.Annotations; @@ -80,6 +81,14 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl runtime.UseMassTransitDispatcher(); runtime.UseCache(); }); + elsa.UseJavaScript(options => + { + options.AllowClrAccess = true; + options.ConfigureEngine(engine => + { + engine.SetValue("getStaticValue", () => StaticValueHolder.Value); + }); + }); }; } diff --git a/test/component/Elsa.Workflows.ComponentTests/Helpers/Services/StaticValueHolder.cs b/test/component/Elsa.Workflows.ComponentTests/Helpers/Services/StaticValueHolder.cs new file mode 100644 index 000000000..ab1d07c4f --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Helpers/Services/StaticValueHolder.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.ComponentTests.Helpers.Services; + +public static class StaticValueHolder +{ + public static object? Value { get; set; } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs new file mode 100644 index 000000000..f3b8e3df2 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/DynamicEndpointTests.cs @@ -0,0 +1,36 @@ +using Elsa.Workflows.ComponentTests.Helpers.Services; +using Elsa.Workflows.Runtime.Contracts; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowDefinitionRefresh; + +public class DynamicEndpointTests : AppComponentTest +{ + private readonly IWorkflowDefinitionsRefresher _workflowDefinitionsRefresher; + + public DynamicEndpointTests(App app) : base(app) + { + StaticValueHolder.Value = "first-value"; + _workflowDefinitionsRefresher = Scope.ServiceProvider.GetRequiredService(); + } + + [Fact] + public async Task HelloWorldWorkflow_ShouldRespondWithHelloWorld() + { + var client = WorkflowServer.CreateHttpWorkflowClient(); + + var firstResponse = await client.GetStringAsync("first-value"); + + StaticValueHolder.Value = "second-value"; + await _workflowDefinitionsRefresher.RefreshWorkflowDefinitionsAsync(new Runtime.Requests.RefreshWorkflowDefinitionsRequest + { + BatchSize = 10, + DefinitionIds = ["f69f061159adc3ae"] + }, CancellationToken.None); + + var secondResponse = await client.GetStringAsync("second-value"); + + Assert.Equal("", firstResponse); + Assert.Equal("", secondResponse); + } +} \ No newline at end of file diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/dynamic-endpoint-http-workflow.json b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/dynamic-endpoint-http-workflow.json new file mode 100644 index 000000000..29cd97051 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowDefinitionRefresh/dynamic-endpoint-http-workflow.json @@ -0,0 +1,104 @@ +{ + "id": "69cb6deaa8e20599", + "definitionId": "f69f061159adc3ae", + "name": "Workflow 1", + "createdAt": "2024-07-04T13:41:33.1905134+00:00", + "version": 1, + "toolVersion": "3.3.0.0", + "variables": [], + "inputs": [], + "outputs": [], + "outcomes": [], + "customProperties": { + "Elsa:WorkflowContextProviderTypes": [] + }, + "isReadonly": false, + "isSystem": false, + "isLatest": true, + "isPublished": true, + "options": { + "autoUpdateConsumingWorkflows": false + }, + "root": { + "type": "Elsa.Flowchart", + "version": 1, + "id": "1bac532ee993dfdc", + "nodeId": "Workflow1:1bac532ee993dfdc", + "metadata": {}, + "customProperties": { + "source": "FlowchartJsonConverter.cs:45", + "notFoundConnections": [], + "canStartWorkflow": false, + "runAsynchronously": false + }, + "activities": [ + { + "path": { + "typeName": "String", + "expression": { + "type": "JavaScript", + "value": "return getStaticValue() ?? \u0022first-value\u0022;" + } + }, + "supportedMethods": { + "typeName": "String[]", + "expression": { + "type": "Literal", + "value": "[\u0022GET\u0022]" + } + }, + "authorize": { + "typeName": "Boolean", + "expression": { + "type": "Literal", + "value": false + } + }, + "policy": { + "typeName": "String", + "expression": { + "type": "Literal" + } + }, + "requestTimeout": null, + "requestSizeLimit": null, + "fileSizeLimit": null, + "allowedFileExtensions": null, + "blockedFileExtensions": null, + "allowedMimeTypes": null, + "exposeRequestTooLargeOutcome": false, + "exposeFileTooLargeOutcome": false, + "exposeInvalidFileExtensionOutcome": false, + "exposeInvalidFileMimeTypeOutcome": false, + "parsedContent": null, + "files": null, + "routeData": null, + "queryStringData": null, + "headers": null, + "result": null, + "id": "25d3a6465094a796", + "nodeId": "Workflow1:1bac532ee993dfdc:25d3a6465094a796", + "name": "HttpEndpoint1", + "type": "Elsa.HttpEndpoint", + "version": 1, + "customProperties": { + "canStartWorkflow": true, + "runAsynchronously": false + }, + "metadata": { + "designer": { + "position": { + "x": -260, + "y": 460 + }, + "size": { + "width": 176.390625, + "height": 50 + } + } + } + } + ], + "connections": [] + } +} \ No newline at end of file