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