From c30586cd9e7998391dc6c5807cc02248ee7dc77f Mon Sep 17 00:00:00 2001 From: MariusVuscanNx <96233009+MariusVuscanNx@users.noreply.github.com> Date: Wed, 8 Mar 2023 10:49:43 +0200 Subject: [PATCH] Implemented workflow authorization (#3773) * Returned 404 when path is not found and 500 when there are multiple workflows with the same path * Return failure response in case of workflow fault * Started workflow authorization implementation * Implemented workflow authorization --- .../Elsa.Http/Activities/HttpEndpoint.cs | 9 +- src/modules/Elsa.Http/Features/HttpFeature.cs | 8 +- ...onBasedHttpEndpointAuthorizationHandler.cs | 14 +-- .../Middleware/WorkflowsMiddleware.cs | 107 ++++++++++++++---- .../Models/AuthorizeHttpEndpointContext.cs | 3 +- .../Models/HttpEndpointBookmarkPayload.cs | 14 ++- .../Elsa.Http/Models/HttpWorkflowResource.cs | 5 - .../Services/ProtoActorWorkflowRuntime.cs | 82 +++++++++++++- .../Models/WorkflowExecutionContext.cs | 65 ++++++----- .../Contracts/IWorkflowRuntime.cs | 42 +++++-- .../Services/DefaultWorkflowRuntime.cs | 87 +++++++++++++- 11 files changed, 339 insertions(+), 97 deletions(-) delete mode 100644 src/modules/Elsa.Http/Models/HttpWorkflowResource.cs diff --git a/src/modules/Elsa.Http/Activities/HttpEndpoint.cs b/src/modules/Elsa.Http/Activities/HttpEndpoint.cs index 6c4ef3812..ab0fa9fd2 100644 --- a/src/modules/Elsa.Http/Activities/HttpEndpoint.cs +++ b/src/modules/Elsa.Http/Activities/HttpEndpoint.cs @@ -142,10 +142,11 @@ public class HttpEndpoint : Trigger // Generate bookmark data for path and selected methods. var path = context.Get(Path); var methods = context.Get(SupportedMethods); - return methods!.Select(x => new HttpEndpointBookmarkPayload(path!, x.ToLowerInvariant()) - { - Policy = Guid.NewGuid().ToString() - }).Cast().ToArray(); + var authorize = context.Get(Authorize); + var policy = context.Get(Policy); + return methods!.Select(x => + new HttpEndpointBookmarkPayload(path!, x.ToLowerInvariant(), authorize, policy)) + .Cast().ToArray(); } private async Task HandleRequestAsync(ActivityExecutionContext context, HttpContext httpContext) diff --git a/src/modules/Elsa.Http/Features/HttpFeature.cs b/src/modules/Elsa.Http/Features/HttpFeature.cs index 339dbaa9a..64fa3f539 100644 --- a/src/modules/Elsa.Http/Features/HttpFeature.cs +++ b/src/modules/Elsa.Http/Features/HttpFeature.cs @@ -42,7 +42,7 @@ public class HttpFeature : FeatureBase /// A delegate that is invoked when authorizing an inbound HTTP request. /// public Func HttpEndpointAuthorizationHandler { get; set; } = ActivatorUtilities.GetServiceOrCreateInstance; - + /// /// A delegate that is invoked when an HTTP workflow faults. /// @@ -70,7 +70,7 @@ public class HttpFeature : FeatureBase typeof(HttpResponse), typeof(HttpRequestHeaders) }, "HTTP"); - + management.AddActivitiesFrom(); }); } @@ -113,9 +113,9 @@ public class HttpFeature : FeatureBase // Add Http endpoint handlers. .AddSingleton(HttpEndpointWorkflowFaultHandler) + .AddSingleton(HttpEndpointAuthorizationHandler) // Add mediator handlers. - .AddNotificationHandlersFrom() - ; + .AddNotificationHandlersFrom(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Http/Handlers/AuthenticationBasedHttpEndpointAuthorizationHandler.cs b/src/modules/Elsa.Http/Handlers/AuthenticationBasedHttpEndpointAuthorizationHandler.cs index bb520de1a..397366943 100644 --- a/src/modules/Elsa.Http/Handlers/AuthenticationBasedHttpEndpointAuthorizationHandler.cs +++ b/src/modules/Elsa.Http/Handlers/AuthenticationBasedHttpEndpointAuthorizationHandler.cs @@ -1,4 +1,3 @@ -using Elsa.Extensions; using Elsa.Http.Contracts; using Elsa.Http.Models; using Microsoft.AspNetCore.Authorization; @@ -18,19 +17,16 @@ public class AuthenticationBasedHttpEndpointAuthorizationHandler : IHttpEndpoint if (identity == null) return false; - + if (identity.IsAuthenticated == false) return false; - - var httpEndpoint = context.Activity; - var expressionExecutionContext = context.ExpressionExecutionContext; - var policyName = httpEndpoint.Policy.TryGet(expressionExecutionContext); - if (string.IsNullOrWhiteSpace(policyName)) + if (string.IsNullOrWhiteSpace(context.Policy)) return identity.IsAuthenticated; - var resource = new HttpWorkflowResource(expressionExecutionContext, httpEndpoint, context.WorkflowInstanceId); - var authorizationResult = await _authorizationService.AuthorizeAsync(user, resource, policyName); + var authorizationResult = await _authorizationService.AuthorizeAsync(user, + new { workflowInstanceId = context.WorkflowInstanceId }, context.Policy!); + return authorizationResult.Succeeded; } } \ No newline at end of file diff --git a/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs b/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs index 25f81a4ef..f1645737e 100644 --- a/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs +++ b/src/modules/Elsa.Http/Middleware/WorkflowsMiddleware.cs @@ -1,16 +1,17 @@ -using System.Net.Mime; -using System.Text.Json; +using Elsa.Http.Contracts; using Elsa.Http.Models; using Elsa.Http.Options; +using Elsa.Workflows.Core.Contracts; using Elsa.Workflows.Core.Helpers; using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Management.Contracts; +using Elsa.Workflows.Runtime.Contracts; using JetBrains.Annotations; using Microsoft.AspNetCore.Http; using Microsoft.Extensions.Options; using System.Net; -using Elsa.Http.Contracts; -using Elsa.Workflows.Management.Contracts; -using Elsa.Workflows.Runtime.Contracts; +using System.Net.Mime; +using System.Text.Json; namespace Elsa.Http.Middleware; @@ -22,11 +23,16 @@ public class WorkflowsMiddleware { private readonly RequestDelegate _next; private readonly IWorkflowRuntime _workflowRuntime; - private readonly IHttpBookmarkProcessor _httpBookmarkProcessor; private readonly IRouteMatcher _routeMatcher; private readonly IRouteTable _routeTable; private readonly IWorkflowInstanceStore _workflowInstanceStore; + private readonly IHttpBookmarkProcessor _httpBookmarkProcessor; private readonly IHttpEndpointWorkflowFaultHandler _httpEndpointWorkflowFaultHandler; + private readonly IHttpEndpointAuthorizationHandler _httpEndpointAuthorizationHandler; + private readonly IBookmarkStore _bookmarkStore; + private readonly ITriggerStore _triggerStore; + private readonly IBookmarkHasher _hasher; + private readonly IBookmarkPayloadSerializer _serializer; private readonly HttpActivityOptions _options; private readonly string _activityTypeName = ActivityTypeNameHelper.GenerateTypeName(); @@ -36,19 +42,29 @@ public class WorkflowsMiddleware public WorkflowsMiddleware( RequestDelegate next, IWorkflowRuntime workflowRuntime, - IHttpBookmarkProcessor httpBookmarkProcessor, IWorkflowInstanceStore workflowInstanceStore, - IHttpEndpointWorkflowFaultHandler httpEndpointWorkflowFaultHandler, IOptions options, + IHttpBookmarkProcessor httpBookmarkProcessor, + IHttpEndpointWorkflowFaultHandler httpEndpointWorkflowFaultHandler, + IHttpEndpointAuthorizationHandler httpEndpointAuthorizationHandler, + IBookmarkStore bookmarkStore, + ITriggerStore triggerStore, + IBookmarkHasher hasher, + IBookmarkPayloadSerializer serializer, IRouteMatcher routeMatcher, IRouteTable routeTable) { _next = next; _workflowRuntime = workflowRuntime; - _httpBookmarkProcessor = httpBookmarkProcessor; _workflowInstanceStore = workflowInstanceStore; - _httpEndpointWorkflowFaultHandler = httpEndpointWorkflowFaultHandler; _options = options.Value; + _httpBookmarkProcessor = httpBookmarkProcessor; + _httpEndpointWorkflowFaultHandler = httpEndpointWorkflowFaultHandler; + _httpEndpointAuthorizationHandler = httpEndpointAuthorizationHandler; + _bookmarkStore = bookmarkStore; + _triggerStore = triggerStore; + _hasher = hasher; + _serializer = serializer; _routeMatcher = routeMatcher; _routeTable = routeTable; } @@ -91,23 +107,29 @@ public class WorkflowsMiddleware var triggerOptions = new TriggerWorkflowsRuntimeOptions(correlationId, input); var cancellationToken = httpContext.RequestAborted; - // Trigger the workflow. - var triggerResult = await _workflowRuntime.TriggerWorkflowsAsync(_activityTypeName, bookmarkPayload, triggerOptions, cancellationToken); + var workflowsQuery = new WorkflowsQuery(_activityTypeName, bookmarkPayload, triggerOptions); + var pendingWorkflows = await _workflowRuntime.FindWorkflowsAsync(workflowsQuery, cancellationToken); - if (await HandleNoWorkflowsFoundAsync(httpContext, triggerResult.TriggeredWorkflows, basePath)) + if (await HandleNoWorkflowsFoundAsync(httpContext, pendingWorkflows, basePath)) return; - if (await HandleMultipleWorkflowsFoundAsync(httpContext, triggerResult.TriggeredWorkflows, cancellationToken)) + if (await HandleMultipleWorkflowsFoundAsync(httpContext, pendingWorkflows, cancellationToken)) return; - if (await HandleWorkflowFaultAsync(httpContext, triggerResult, cancellationToken)) + if (await HandleWorkflowFaultAsync(httpContext, pendingWorkflows.Single(), cancellationToken)) return; + if (await AuthorizeAsync(httpContext, pendingWorkflows.Single(), bookmarkPayload, cancellationToken)) + return; + + var executionResult = await _workflowRuntime.ExecutePendingWorkflowAsync(pendingWorkflows.Single(), input, cancellationToken); + // Process the trigger result by resuming each HTTP bookmark, if any. - await _httpBookmarkProcessor.ProcessBookmarks(triggerResult.TriggeredWorkflows, correlationId, input, cancellationToken); + await _httpBookmarkProcessor.ProcessBookmarks(new List { executionResult }, correlationId, input, cancellationToken); } - private string? GetMatchingRoute(string? path) { + private string? GetMatchingRoute(string? path) + { var matchingRouteQuery = from route in _routeTable @@ -142,9 +164,9 @@ public class WorkflowsMiddleware private string GetPath(HttpContext httpContext) => httpContext.Request.Path.Value.ToLowerInvariant(); - private async Task HandleNoWorkflowsFoundAsync(HttpContext httpContext, ICollection triggeredWorkflows, PathString? basePath) + private async Task HandleNoWorkflowsFoundAsync(HttpContext httpContext, IEnumerable pendingWorkflows, PathString? basePath) { - if (triggeredWorkflows.Any()) + if (pendingWorkflows.Any()) return false; // If a base path was configured, we are sure the requester tried to execute a workflow that doesn't exist. @@ -161,9 +183,9 @@ public class WorkflowsMiddleware return true; } - private async Task HandleMultipleWorkflowsFoundAsync(HttpContext httpContext, ICollection triggeredWorkflows, CancellationToken cancellationToken) + private async Task HandleMultipleWorkflowsFoundAsync(HttpContext httpContext, IEnumerable pendingWorkflows, CancellationToken cancellationToken) { - if (triggeredWorkflows.Count <= 1) + if (pendingWorkflows.ToList().Count <= 1) return false; httpContext.Response.ContentType = "application/json"; @@ -172,16 +194,16 @@ public class WorkflowsMiddleware var responseContent = JsonSerializer.Serialize(new { errorMessage = "The call is ambiguous and matches multiple workflows.", - workflows = triggeredWorkflows + workflows = pendingWorkflows }); await httpContext.Response.WriteAsync(responseContent, cancellationToken); return true; } - private async Task HandleWorkflowFaultAsync(HttpContext httpContext, TriggerWorkflowsResult triggerResult, CancellationToken cancellationToken) + private async Task HandleWorkflowFaultAsync(HttpContext httpContext, CollectedWorkflow pendingWorkflow, CancellationToken cancellationToken) { - var instanceFilter = new WorkflowInstanceFilter { Id = triggerResult.TriggeredWorkflows.Single().InstanceId }; + var instanceFilter = new WorkflowInstanceFilter { Id = pendingWorkflow.WorkflowInstanceId }; var workflowInstance = await _workflowInstanceStore.FindAsync(instanceFilter, cancellationToken); if (workflowInstance is not null @@ -194,4 +216,41 @@ public class WorkflowsMiddleware return false; } + + private async Task AuthorizeAsync( + HttpContext httpContext, + CollectedWorkflow pendingWorkflow, + HttpEndpointBookmarkPayload bookmarkPayload, + CancellationToken cancellationToken = default) + { + var hash = _hasher.Hash(_activityTypeName, bookmarkPayload); + var payload = default(HttpEndpointBookmarkPayload); + + if (pendingWorkflow is CollectedStartableWorkflow) + { + var triggerFilter = new TriggerFilter() { Hash = hash }; + var triggers = (await _triggerStore.FindManyAsync(triggerFilter, cancellationToken)) + .Select(x => _serializer.Deserialize(x.Data!)).ToList(); + payload = triggers.Single(); + } + else + { + var bookmarkFilter = new BookmarkFilter() { Hash = hash }; + var bookmarks = (await _bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)) + .Select(x => _serializer.Deserialize(x.Data!)).ToList(); + payload = bookmarks.Single(); + } + + if (!(payload.Authorize ?? false)) + return false; + + var authorized = await _httpEndpointAuthorizationHandler.AuthorizeAsync(new AuthorizeHttpEndpointContext(httpContext, pendingWorkflow.WorkflowInstanceId, payload.Policy)); + + if (!authorized) + { + httpContext.Response.StatusCode = (int)HttpStatusCode.Unauthorized; + } + + return !authorized; + } } \ No newline at end of file diff --git a/src/modules/Elsa.Http/Models/AuthorizeHttpEndpointContext.cs b/src/modules/Elsa.Http/Models/AuthorizeHttpEndpointContext.cs index 586dbb458..4fb0c25d3 100644 --- a/src/modules/Elsa.Http/Models/AuthorizeHttpEndpointContext.cs +++ b/src/modules/Elsa.Http/Models/AuthorizeHttpEndpointContext.cs @@ -1,6 +1,5 @@ -using Elsa.Expressions.Models; using Microsoft.AspNetCore.Http; namespace Elsa.Http.Models; -public record AuthorizeHttpEndpointContext(ExpressionExecutionContext ExpressionExecutionContext, HttpContext HttpContext, HttpEndpoint Activity, string WorkflowInstanceId); \ No newline at end of file +public record AuthorizeHttpEndpointContext(HttpContext HttpContext, string WorkflowInstanceId, string? Policy = default); \ No newline at end of file diff --git a/src/modules/Elsa.Http/Models/HttpEndpointBookmarkPayload.cs b/src/modules/Elsa.Http/Models/HttpEndpointBookmarkPayload.cs index 6ecd2f096..2d77513f2 100644 --- a/src/modules/Elsa.Http/Models/HttpEndpointBookmarkPayload.cs +++ b/src/modules/Elsa.Http/Models/HttpEndpointBookmarkPayload.cs @@ -1,5 +1,5 @@ -using System.Text.Json.Serialization; using Elsa.Workflows.Core.Attributes; +using System.Text.Json.Serialization; namespace Elsa.Http.Models; @@ -12,11 +12,14 @@ public record HttpEndpointBookmarkPayload public HttpEndpointBookmarkPayload() { } - - public HttpEndpointBookmarkPayload(string path, string method) + + public HttpEndpointBookmarkPayload(string path, string method, + bool? authorize = default, string? policy = default) { Path = path; Method = method; + Authorize = authorize; + Policy = policy; } public string Path @@ -32,5 +35,8 @@ public record HttpEndpointBookmarkPayload } [ExcludeFromHash] - public string Policy { get; set; } + public string? Policy { get; set; } + + [ExcludeFromHash] + public bool? Authorize { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Http/Models/HttpWorkflowResource.cs b/src/modules/Elsa.Http/Models/HttpWorkflowResource.cs deleted file mode 100644 index dfafb1eb0..000000000 --- a/src/modules/Elsa.Http/Models/HttpWorkflowResource.cs +++ /dev/null @@ -1,5 +0,0 @@ -using Elsa.Expressions.Models; - -namespace Elsa.Http.Models; - -public record HttpWorkflowResource(ExpressionExecutionContext ExpressionExecutionContext, HttpEndpoint Activity, string WorkflowInstance); \ No newline at end of file diff --git a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 2b02ec4ab..9adafd3da 100644 --- a/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,4 +1,3 @@ -using System.Text.Json; using Elsa.Common.Models; using Elsa.Extensions; using Elsa.ProtoActor.Extensions; @@ -6,10 +5,10 @@ using Elsa.ProtoActor.Protos; using Elsa.Workflows.Core.Contracts; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.Serialization; -using Elsa.Workflows.Core.Services; using Elsa.Workflows.Core.State; using Elsa.Workflows.Runtime.Contracts; using Proto.Cluster; +using System.Text.Json; namespace Elsa.ProtoActor.Services; @@ -23,6 +22,7 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime private readonly ITriggerStore _triggerStore; private readonly IIdentityGenerator _identityGenerator; private readonly IBookmarkHasher _hasher; + private readonly IWorkflowInstanceFactory _workflowInstanceFactory; /// /// Constructor. @@ -32,13 +32,15 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime SerializerOptionsProvider serializerOptionsProvider, ITriggerStore triggerStore, IIdentityGenerator identityGenerator, - IBookmarkHasher hasher) + IBookmarkHasher hasher, + IWorkflowInstanceFactory workflowInstanceFactory) { _cluster = cluster; _serializerOptionsProvider = serializerOptionsProvider; _triggerStore = triggerStore; _identityGenerator = identityGenerator; _hasher = hasher; + _workflowInstanceFactory = workflowInstanceFactory; } /// @@ -162,6 +164,38 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime return new TriggerWorkflowsResult(results); } + /// + public async Task ExecutePendingWorkflowAsync(CollectedWorkflow collectedWorkflow, IDictionary? input = default, CancellationToken cancellationToken = default) + { + if (collectedWorkflow is CollectedStartableWorkflow collectedStartableWorkflow) + { + var startOptions = new StartWorkflowRuntimeOptions(collectedStartableWorkflow.CorrelationId, input, VersionOptions.Published, + collectedStartableWorkflow.ActivityId, collectedStartableWorkflow.WorkflowInstanceId); + var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions, cancellationToken); + return new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks); + } + else + { + var collectedResumableWorkflow = (collectedWorkflow as CollectedResumableWorkflow)!; + var runtimeOptions = new ResumeWorkflowRuntimeOptions(collectedResumableWorkflow.CorrelationId, Input: input); + var resumeResult = await ResumeWorkflowAsync( + collectedWorkflow.WorkflowInstanceId, + runtimeOptions with { BookmarkId = collectedResumableWorkflow.BookmarkId }, + cancellationToken); + + return new WorkflowExecutionResult(collectedResumableWorkflow.WorkflowInstanceId, resumeResult.Bookmarks); + } + } + + /// + public async Task> FindWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken = default) + { + var startableWorkflows = await CollectStartableWorkflowsAsync(workflowsQuery, cancellationToken); + var resumableWorkflows = await CollectResumableWorkflowsAsync(workflowsQuery, cancellationToken); + var results = startableWorkflows.Concat(resumableWorkflows).ToList(); + return results; + } + /// public async Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) { @@ -277,4 +311,46 @@ public class ProtoActorWorkflowRuntime : IWorkflowRuntime x.ActivityInstanceId, x.AutoBurn, x.CallbackMethodName.NullIfEmpty())); + + private async Task> CollectStartableWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken) + { + var hash = _hasher.Hash(workflowsQuery.ActivityTypeName, workflowsQuery.BookmarkPayload); + var filter = new TriggerFilter { Hash = hash }; + var triggers = await _triggerStore.FindManyAsync(filter, cancellationToken); + var results = new List(); + + foreach (var trigger in triggers) + { + var definitionId = trigger.WorkflowDefinitionId; + var startOptions = new StartWorkflowRuntimeOptions(workflowsQuery.Options.CorrelationId, workflowsQuery.Options.Input, VersionOptions.Published, trigger.ActivityId); + var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions, cancellationToken); + + var workflowInstance = await _workflowInstanceFactory.CreateAsync(definitionId, workflowsQuery.Options.CorrelationId, cancellationToken); + + if (canStartResult.CanStart) + { + results.Add(new CollectedStartableWorkflow(workflowInstance.Id, workflowInstance, workflowsQuery.Options.CorrelationId, trigger.ActivityId, definitionId)); + } + } + + return results; + } + + private async Task> CollectResumableWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken) + { + var hash = _hasher.Hash(workflowsQuery.ActivityTypeName, workflowsQuery.BookmarkPayload); + var client = _cluster.GetNamedBookmarkGrain(hash); + + var request = new ResolveBookmarksRequest + { + ActivityTypeName = workflowsQuery.ActivityTypeName, + CorrelationId = workflowsQuery.Options.CorrelationId.EmptyIfNull() + }; + + var bookmarksResponse = await client.Resolve(request, cancellationToken); + var bookmarks = bookmarksResponse!.Bookmarks; + + var collectedWorkflows = bookmarks.Select(b => new CollectedResumableWorkflow(b.WorkflowInstanceId, default, workflowsQuery.Options.CorrelationId, b.BookmarkId)).ToList(); + return collectedWorkflows; + } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs b/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs index 299fcfe88..a3749b6f0 100644 --- a/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs +++ b/src/modules/Elsa.Workflows.Core/Models/WorkflowExecutionContext.cs @@ -1,10 +1,10 @@ -using System.Collections.ObjectModel; using Elsa.Expressions.Helpers; using Elsa.Expressions.Models; using Elsa.Extensions; using Elsa.Workflows.Core.Contracts; using Elsa.Workflows.Core.Services; using Microsoft.Extensions.DependencyInjection; +using System.Collections.ObjectModel; namespace Elsa.Workflows.Core.Models; @@ -66,57 +66,57 @@ public class WorkflowExecutionContext /// The associated with the execution context. /// public Workflow Workflow { get; } - + /// /// A graph of the workflow structure. /// public ActivityNode Graph { get; } - + /// /// The current status of the workflow. /// public WorkflowStatus Status => GetMainStatus(SubStatus); - + /// /// The current sub status of the workflow. /// public WorkflowSubStatus SubStatus { get; internal set; } - + /// /// The root associated with the execution context. /// public MemoryRegister MemoryRegister { get; } - + /// /// A unique ID of the execution context. /// public string Id { get; set; } - + /// /// An application-specific identifier associated with the execution context. /// public string? CorrelationId { get; set; } - + /// /// A flattened list of s from the . /// public IReadOnlyCollection Nodes => new ReadOnlyCollection(_nodes); - + /// /// A map between activity IDs and s in the workflow graph. /// public IDictionary NodeIdLookup { get; } - + /// /// A map between s and s in the workflow graph. /// public IDictionary NodeActivityLookup { get; } - + /// /// The for the execution context. /// public IActivityScheduler Scheduler { get; } - + /// /// A collection of collected bookmarks during workflow execution. /// @@ -152,22 +152,22 @@ public class WorkflowExecutionContext /// The current delegate to invoke when executing the next activity. /// public ExecuteActivityDelegate? ExecuteDelegate { get; set; } - + /// /// Provides context about the bookmark that was used to resume workflow execution, if any. /// public ResumedBookmarkContext? ResumedBookmarkContext { get; set; } - + /// /// The ID of the activity associated with the trigger that caused this workflow execution, if any. /// public string? TriggerActivityId { get; set; } - + /// /// A that can be used to cancel asynchronous operations. /// public CancellationToken CancellationToken { get; } - + /// /// A list of callbacks that are invoked when the associated child activity completes. /// @@ -191,32 +191,32 @@ public class WorkflowExecutionContext /// Resolves the specified service type from the service provider. /// public T GetRequiredService() where T : notnull => _serviceProvider.GetRequiredService(); - + /// /// Resolves the specified service type from the service provider. /// public object GetRequiredService(Type serviceType) => _serviceProvider.GetRequiredService(serviceType); - + /// /// Resolves the specified service type from the service provider, or creates a new instance if the service type was not found in the service container. /// public T GetOrCreateService() where T : notnull => ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider); - + /// /// Resolves the specified service type from the service provider, or creates a new instance if the service type was not found in the service container. /// public object GetOrCreateService(Type serviceType) => ActivatorUtilities.GetServiceOrCreateInstance(_serviceProvider, serviceType); - + /// /// Resolves the specified service type from the service provider. /// public T? GetService() where T : notnull => _serviceProvider.GetService(); - + /// /// Resolves the specified service type from the service provider. /// public object? GetService(Type serviceType) => _serviceProvider.GetService(serviceType); - + /// /// Resolves multiple implementations of the specified service type from the service provider. /// @@ -257,22 +257,29 @@ public class WorkflowExecutionContext /// Returns the with the specified activity ID from the workflow graph. /// public ActivityNode FindNodeById(string nodeId) => NodeIdLookup[nodeId]; - + /// /// Returns the containing the specified activity from the workflow graph. /// public ActivityNode FindNodeByActivity(IActivity activity) => NodeActivityLookup[activity]; - + /// /// Returns the with the specified ID from the workflow graph. /// public IActivity FindActivityByNodeId(string nodeId) => FindNodeById(nodeId).Activity; + /// + /// + /// + /// + /// + public IActivity FindActivityByActivityId(string activityId) => FindNodeById(NodeIdLookup.Single(n => n.Key.Contains(activityId)).Value.NodeId).Activity; + /// /// Returns a custom property with the specified key from the dictionary. /// public T? GetProperty(string key) => Properties.TryGetValue(key, out var value) ? value.ConvertTo() : default; - + /// /// Sets a custom property with the specified key on the dictionary. /// @@ -317,7 +324,7 @@ public class WorkflowExecutionContext expressionExecutionContext.TransientProperties[ExpressionExecutionContextExtensions.ActivityExecutionContextKey] = activityExecutionContext; return activityExecutionContext; } - + /// /// Removes the specified . /// @@ -330,15 +337,15 @@ public class WorkflowExecutionContext // Remove the context. _activityExecutionContexts.Remove(context); - + // Remove all associated completion callbacks. context.ClearCompletionCallbacks(); - + // Remove all associated variables. var variablePersistenceManager = context.GetRequiredService(); var variables = variablePersistenceManager.GetVariables(context); await variablePersistenceManager.DeleteVariablesAsync(this, variables); - + // Remove all associated bookmarks. Bookmarks.RemoveWhere(x => x.ActivityInstanceId == context.Id); } diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index 6abeec355..2e49a9c1c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -2,6 +2,7 @@ using Elsa.Common.Models; using Elsa.Workflows.Core.Helpers; using Elsa.Workflows.Core.Models; using Elsa.Workflows.Core.State; +using Elsa.Workflows.Management.Entities; namespace Elsa.Workflows.Runtime.Contracts; @@ -14,7 +15,7 @@ public interface IWorkflowRuntime /// Returns a value whether or not the specified workflow definition can create a new instance. /// Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options, CancellationToken cancellationToken); - + /// /// Creates a new workflow instance of the specified definition ID and executes it. /// @@ -32,7 +33,7 @@ public interface IWorkflowRuntime object bookmarkPayload, TriggerWorkflowsRuntimeOptions options, CancellationToken cancellationToken = default); - + /// /// Resumes an existing workflow instance. /// @@ -40,27 +41,44 @@ public interface IWorkflowRuntime /// /// Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options, CancellationToken cancellationToken = default); - + /// /// Resumes all workflows that are bookmarked on the specified activity type. /// Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsRuntimeOptions options, CancellationToken cancellationToken = default); - + /// /// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload. /// Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsRuntimeOptions options, CancellationToken cancellationToken = default); - + + /// + /// Executes a pending workflow. + /// + /// + /// + /// + /// + Task ExecutePendingWorkflowAsync(CollectedWorkflow collectedWorkflow, IDictionary? input = default, CancellationToken cancellationToken = default); + + /// + /// Finds all the workflows that can be started or resumed based on a query model. + /// + /// + /// + /// + Task> FindWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken = default); + /// /// Exports the of the specified workflow instance. /// Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default); - + /// /// Imports the specified . /// Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); - + /// /// Adds and removes bookmarks based on the provided bookmarks diff. /// @@ -78,8 +96,14 @@ public record CanStartWorkflowResult(string? InstanceId, bool CanStart); public record ResumeWorkflowResult(ICollection Bookmarks); public record TriggerWorkflowsRuntimeOptions(string? CorrelationId = default, IDictionary? Input = default); public record TriggerWorkflowsResult(ICollection TriggeredWorkflows); -public record WorkflowExecutionResult(string InstanceId, ICollection Bookmarks); +public record WorkflowExecutionResult(string InstanceId, ICollection Bookmarks, string? ActivityId = null); public record UpdateBookmarksContext(string InstanceId, Diff Diff, string? CorrelationId); +public record WorkflowsQuery(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsRuntimeOptions Options); +public record CollectedWorkflow(string WorkflowInstanceId, WorkflowInstance? WorkflowInstance, string? CorrelationId); +public record CollectedStartableWorkflow(string WorkflowInstanceId, WorkflowInstance? WorkflowInstance, string? CorrelationId, string? ActivityId, string? DefinitionId) + : CollectedWorkflow(WorkflowInstanceId, WorkflowInstance, CorrelationId); +public record CollectedResumableWorkflow(string WorkflowInstanceId, WorkflowInstance? WorkflowInstance, string? CorrelationId, string? BookmarkId) + : CollectedWorkflow(WorkflowInstanceId, WorkflowInstance, CorrelationId); /// /// Contains arguments to use for counting the number of workflow instances. @@ -95,7 +119,7 @@ public class CountRunningWorkflowsArgs /// The workflow definition version to include in the query. /// public int? Version { get; set; } - + /// /// The correlation ID to include in the query. /// diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs index 514d61da4..49abc6583 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs @@ -20,6 +20,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime private readonly IBookmarkStore _bookmarkStore; private readonly IBookmarkHasher _hasher; private readonly IDistributedLockProvider _distributedLockProvider; + private readonly IWorkflowInstanceFactory _workflowInstanceFactory; /// /// Constructor. @@ -31,7 +32,8 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime ITriggerStore triggerStore, IBookmarkStore bookmarkStore, IBookmarkHasher hasher, - IDistributedLockProvider distributedLockProvider) + IDistributedLockProvider distributedLockProvider, + IWorkflowInstanceFactory workflowInstanceFactory) { _workflowHostFactory = workflowHostFactory; _workflowDefinitionService = workflowDefinitionService; @@ -40,6 +42,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime _bookmarkStore = bookmarkStore; _hasher = hasher; _distributedLockProvider = distributedLockProvider; + _workflowInstanceFactory = workflowInstanceFactory; } /// @@ -96,7 +99,7 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime continue; var startResult = await StartWorkflowAsync(definitionId, startOptions, cancellationToken); - results.Add(new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks)); + results.Add(new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks, trigger.ActivityId)); } } @@ -157,6 +160,36 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime return new TriggerWorkflowsResult(results); } + public async Task ExecutePendingWorkflowAsync(CollectedWorkflow collectedWorkflow, IDictionary? input = default, CancellationToken cancellationToken = default) + { + if (collectedWorkflow is CollectedStartableWorkflow collectedStartableWorkflow) + { + var startOptions = new StartWorkflowRuntimeOptions(collectedStartableWorkflow.CorrelationId, input, VersionOptions.Published, + collectedStartableWorkflow.ActivityId, collectedStartableWorkflow.WorkflowInstanceId); + var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions, cancellationToken); + return new WorkflowExecutionResult(startResult.InstanceId, startResult.Bookmarks, collectedStartableWorkflow.ActivityId); + } + else + { + var collectedResumableWorkflow = (collectedWorkflow as CollectedResumableWorkflow)!; + var runtimeOptions = new ResumeWorkflowRuntimeOptions(collectedResumableWorkflow.CorrelationId, Input: input); + var resumeResult = await ResumeWorkflowAsync( + collectedWorkflow.WorkflowInstanceId, + runtimeOptions with { BookmarkId = collectedResumableWorkflow.BookmarkId }, + cancellationToken); + + return new WorkflowExecutionResult(collectedResumableWorkflow.WorkflowInstanceId, resumeResult.Bookmarks); + } + } + + public async Task> FindWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken = default) + { + var startableWorkflows = await CollectStartableWorkflowsAsync(workflowsQuery, cancellationToken); + var resumableWorkflows = await CollectResumableWorkflowsAsync(workflowsQuery, cancellationToken); + var results = startableWorkflows.Concat(resumableWorkflows).ToList(); + return results; + } + /// public async Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => await _workflowStateStore.LoadAsync(workflowInstanceId, cancellationToken); @@ -192,8 +225,11 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime foreach (var bookmark in bookmarks) { var workflowInstanceId = bookmark.WorkflowInstanceId; - var resumeOptions = new ResumeWorkflowRuntimeOptions(runtimeOptions.CorrelationId, bookmark.BookmarkId, Input: runtimeOptions.Input); - var resumeResult = await ResumeWorkflowAsync(workflowInstanceId, resumeOptions, cancellationToken); + + var resumeResult = await ResumeWorkflowAsync( + workflowInstanceId, + runtimeOptions with { BookmarkId = bookmark.BookmarkId }, + cancellationToken); resumedWorkflows.Add(new WorkflowExecutionResult(workflowInstanceId, resumeResult.Bookmarks)); } @@ -221,4 +257,47 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime await _bookmarkStore.DeleteAsync(filter, cancellationToken); } } + + private async Task> CollectStartableWorkflowsAsync( + WorkflowsQuery workflowsQuery, + CancellationToken cancellationToken = default) + { + var results = new List(); + var hash = _hasher.Hash(workflowsQuery.ActivityTypeName, workflowsQuery.BookmarkPayload); + + // Start new workflows. Notice that this happens in a process-synchronized fashion to avoid multiple instances from being created. + var sharedResource = $"{nameof(DefaultWorkflowRuntime)}__StartTriggeredWorkflows__{hash}"; + await using (await _distributedLockProvider.AcquireLockAsync(sharedResource, TimeSpan.FromMinutes(10), cancellationToken)) + { + var filter = new TriggerFilter { Hash = hash }; + var triggers = await _triggerStore.FindManyAsync(filter, cancellationToken); + + foreach (var trigger in triggers) + { + var definitionId = trigger.WorkflowDefinitionId; + var startOptions = new StartWorkflowRuntimeOptions(workflowsQuery.Options.CorrelationId, workflowsQuery.Options.Input, VersionOptions.Published, trigger.ActivityId); + var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions, cancellationToken); + + var workflowInstance = await _workflowInstanceFactory.CreateAsync(definitionId, workflowsQuery.Options.CorrelationId, cancellationToken); + + if (canStartResult.CanStart) + { + results.Add(new CollectedStartableWorkflow(workflowInstance.Id, workflowInstance, workflowsQuery.Options.CorrelationId, trigger.ActivityId, definitionId)); + } + } + } + + return results; + } + + private async Task> CollectResumableWorkflowsAsync(WorkflowsQuery workflowsQuery, CancellationToken cancellationToken = default) + { + var hash = _hasher.Hash(workflowsQuery.ActivityTypeName, workflowsQuery.BookmarkPayload); + var correlationId = workflowsQuery.Options.CorrelationId; + var filter = new BookmarkFilter { Hash = hash, CorrelationId = correlationId }; + var bookmarks = await _bookmarkStore.FindManyAsync(filter, cancellationToken); + + var collectedWorkflows = bookmarks.Select(b => new CollectedResumableWorkflow(b.WorkflowInstanceId, default, correlationId, b.BookmarkId)).ToList(); + return collectedWorkflows; + } } \ No newline at end of file