diff --git a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs index 2ae45c492..886160157 100644 --- a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs +++ b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs @@ -20,7 +20,12 @@ public interface IWorkflowInstanceManager /// Finds the first workflow instance that matches the specified filter. /// Task FindAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default); - + + /// + /// Determines whether a workflow instance with the specified ID exists. + /// + Task ExistsAsync(string instanceId, CancellationToken cancellationToken = default); + /// /// Saves the specified workflow instance. /// diff --git a/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs b/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs index 12bc8ca29..f33e43a6d 100644 --- a/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs +++ b/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs @@ -33,6 +33,16 @@ public class WorkflowInstanceManager( return await store.FindAsync(filter, cancellationToken); } + public async Task ExistsAsync(string instanceId, CancellationToken cancellationToken = default) + { + var filter = new WorkflowInstanceFilter + { + Id = instanceId + }; + var count = await store.CountAsync(filter, cancellationToken); + return count > 0; + } + /// public async Task SaveAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto index a23ca9ab7..47088c1d2 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto @@ -15,4 +15,5 @@ service WorkflowInstance { rpc Cancel (Empty) returns (Empty); rpc ExportState(Empty) returns (ExportWorkflowStateResponse); rpc ImportState(ImportWorkflowStateRequest) returns (Empty); + rpc InstanceExists(Empty) returns (bool); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs index 7576d136f..a37cb11b3 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs @@ -85,6 +85,11 @@ public class ProtoActorWorkflowClient : IWorkflowClient await _actorClient.ImportState(request, CreateHeaders(), cancellationToken); } + public Task InstanceExistsAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + private IDictionary CreateHeaders() { var headers = new Dictionary(); diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs new file mode 100644 index 000000000..629322c83 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -0,0 +1,375 @@ +using System.Diagnostics.CodeAnalysis; +using Elsa.Common.Models; +using Elsa.Extensions; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Matches; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Parameters; +using Elsa.Workflows.Runtime.Params; +using Elsa.Workflows.Runtime.ProtoActor.Extensions; +using Elsa.Workflows.Runtime.ProtoActor.ProtoBuf; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Results; +using Elsa.Workflows.State; + +namespace Elsa.Workflows.Runtime.ProtoActor.Services; + +public partial class ProtoActorWorkflowRuntime +{ + /// + public async Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken); + var workflow = workflowGraph!.Workflow; + + var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() + { + Workflow = workflow, + CorrelationId = options?.CorrelationId, + CancellationToken = cancellationToken + }); + + return new CanStartWorkflowResult( + { + CanStart = canStart, + InstanceId = null + }; + } + + /// + public async Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); + var createRequest = new CreateAndRunWorkflowInstanceRequest + { + Properties = options?.Properties, + CorrelationId = options?.CorrelationId, + Input = options?.Input, + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published), + ParentId = options?.ParentWorkflowInstanceId, + TriggerActivityId = options?.TriggerActivityId + }; + var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken); + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); + } + + /// + public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); + var createRequest = new Workflows.Runtime.Messages.CreateAndRunWorkflowInstanceRequest + { + Properties = options?.Properties, + CorrelationId = options?.CorrelationId, + Input = options?.Input, + WorkflowDefinitionHandle = Workflows.Models.WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published), + ParentId = options?.ParentWorkflowInstanceId, + TriggerActivityId = options?.TriggerActivityId + }; + var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken); + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); + } + + /// + public async Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + { + var hash = _hasher.Hash(activityTypeName, bookmarkPayload); + var filter = new TriggerFilter + { + Hash = hash + }; + var systemCancellationToken = options?.CancellationTokens.SystemCancellationToken ?? default; + var triggers = await _triggerStore.FindManyAsync(filter, systemCancellationToken); + var results = new List(); + + foreach (var trigger in triggers) + { + var definitionId = trigger.WorkflowDefinitionId; + + var startOptions = new StartWorkflowRuntimeParams + { + CorrelationId = options?.CorrelationId, + Input = options?.Input, + Properties = options?.Properties, + VersionOptions = VersionOptions.Published, + TriggerActivityId = trigger.ActivityId, + InstanceId = options?.WorkflowInstanceId, + CancellationTokens = options?.CancellationTokens ?? default + }; + + var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions); + + // If we can't start the workflow, don't try it. + if (!canStartResult.CanStart) + continue; + + var startResult = await StartWorkflowAsync(definitionId, startOptions); + results.Add(startResult); + } + + return results; + } + + /// + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default) + { + var request = new ResumeWorkflowRequest + { + InstanceId = workflowInstanceId, + CorrelationId = options?.CorrelationId.EmptyIfNull(), + BookmarkId = options?.BookmarkId.EmptyIfNull(), + ActivityId = options?.ActivityId.EmptyIfNull(), + Input = options?.Input?.SerializeInput(), + Properties = options?.Properties?.SerializeProperties(), + }; + + var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); + var response = await client.Resume(request, options?.CancellationTokens.SystemCancellationToken ?? default); + + return _workflowExecutionResultMapper.Map(response!); + } + + /// + public async Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + { + var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options?.ActivityInstanceId); + var correlationId = options?.CorrelationId; + var workflowInstanceId = options?.WorkflowInstanceId; + var filter = new BookmarkFilter + { + Hash = hash, + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId + }; + var bookmarks = await _bookmarkStore.FindManyAsync(filter, options?.CancellationTokens.SystemCancellationToken ?? default); + + return await ResumeWorkflowsAsync( + bookmarks, + new ResumeWorkflowRuntimeParams + { + CorrelationId = correlationId, + Input = options?.Input, + Properties = options?.Properties, + CancellationTokens = options?.CancellationTokens ?? default + } + ); + } + + /// + public async Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + { + var startedWorkflows = await StartWorkflowsAsync(activityTypeName, bookmarkPayload, options); + var resumedWorkflows = await ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options); + var results = startedWorkflows.Concat(resumedWorkflows).ToList(); + + return new TriggerWorkflowsResult(results); + } + + /// + public async Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) + { + if (match is StartableWorkflowMatch collectedStartableWorkflow) + { + var startOptions = new StartWorkflowRuntimeParams + { + CorrelationId = collectedStartableWorkflow.CorrelationId, + Input = options?.Input, + Properties = options?.Properties, + VersionOptions = VersionOptions.Published, + TriggerActivityId = collectedStartableWorkflow.ActivityId, + InstanceId = collectedStartableWorkflow.WorkflowInstanceId, + CancellationTokens = options?.CancellationTokens ?? default + }; + return await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions); + } + + var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!; + var runtimeOptions = new ResumeWorkflowRuntimeParams + { + CorrelationId = collectedResumableWorkflow.CorrelationId, + Input = options?.Input, + Properties = options?.Properties, + BookmarkId = collectedResumableWorkflow.BookmarkId, + CancellationTokens = options?.CancellationTokens ?? default + }; + var result = await ResumeWorkflowAsync(match.WorkflowInstanceId, runtimeOptions); + + return result!; + } + + /// + public async Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken) + { + var filter = new WorkflowInstanceFilter + { + Id = workflowInstanceId + }; + + var instance = await _workflowInstanceStore.FindAsync(filter, cancellationToken); + if (instance is null) + return new CancellationResult(false, FailureReason.NotFound); + + var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); + var result = await client.Cancel(cancellationToken); + return new CancellationResult(result?.Result ?? false); + } + + /// + public async Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) + { + var startableWorkflows = await FindStartableWorkflowsAsync(filter, cancellationToken); + var resumableWorkflows = await FindResumableWorkflowsAsync(filter, cancellationToken); + var results = startableWorkflows.Concat(resumableWorkflows).ToList(); + return results; + } + + /// + [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.DeserializeAsync(String, CancellationToken)")] + public async Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); + var response = await client.ExportState(new ExportWorkflowStateRequest(), cancellationToken); + var json = response!.SerializedWorkflowState.Text; + var workflowState = await _workflowStateSerializer.DeserializeAsync(json, cancellationToken); + return workflowState; + } + + /// + [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.SerializeAsync(WorkflowState, CancellationToken)")] + public async Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) + { + var client = _cluster.GetNamedWorkflowGrain(workflowState.Id); + var json = await _workflowStateSerializer.SerializeAsync(workflowState, cancellationToken); + + var request = new ImportWorkflowStateRequest + { + SerializedWorkflowState = new Json + { + Text = json + } + }; + + await client.ImportState(request, cancellationToken); + } + + /// + public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) + { + await _bookmarkStore.SaveAsync(bookmark, cancellationToken); + } + + /// + public async Task CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) + { + var filter = new WorkflowInstanceFilter + { + DefinitionId = request.DefinitionId, + Version = request.Version, + CorrelationId = request.CorrelationId, + WorkflowStatus = WorkflowStatus.Running + }; + return await _workflowInstanceStore.CountAsync(filter, cancellationToken); + } + + private async Task> ResumeWorkflowsAsync(IEnumerable bookmarks, ResumeWorkflowRuntimeParams runtimeParams) + { + var resumedWorkflows = new List(); + + foreach (var bookmark in bookmarks) + { + var workflowInstanceId = bookmark.WorkflowInstanceId; + + var newRuntimeOptions = new ResumeWorkflowRuntimeParams + { + CorrelationId = runtimeParams.CorrelationId, + Input = runtimeParams.Input, + Properties = runtimeParams.Properties, + BookmarkId = bookmark.BookmarkId, + ActivityId = runtimeParams.ActivityId, + ActivityNodeId = runtimeParams.ActivityNodeId, + ActivityInstanceId = runtimeParams.ActivityInstanceId, + ActivityHash = runtimeParams.ActivityHash, + CancellationTokens = runtimeParams.CancellationTokens + }; + + var resumeResult = await ResumeWorkflowAsync(workflowInstanceId, newRuntimeOptions); + resumedWorkflows.Add(resumeResult!); + } + + return resumedWorkflows; + } + + private async Task> FindStartableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken) + { + var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.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 StartWorkflowRuntimeParams + { + CorrelationId = workflowsFilter.Options?.CorrelationId, + Input = workflowsFilter.Options.Input, + Properties = workflowsFilter.Options.Properties, + VersionOptions = VersionOptions.Published, + TriggerActivityId = trigger.ActivityId, + CancellationTokens = cancellationToken + }; + + var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions); + var workflowGraph = await _workflowDefinitionService.FindWorkflowGraphAsync(trigger.WorkflowDefinitionVersionId, cancellationToken); + + if (workflowGraph == null) + { + _logger.LogWarning("Workflow version ID {DefinitionVersionId} not found", trigger.WorkflowDefinitionVersionId); + continue; + } + + var workflow = workflowGraph.Workflow; + var createWorkflowInstanceRequest = new CreateWorkflowInstanceRequest + { + Workflow = workflow, + CorrelationId = workflowsFilter.Options.CorrelationId, + WorkflowInstanceId = workflowsFilter.Options?.WorkflowInstanceId, + Input = workflowsFilter.Options?.Input, + Properties = workflowsFilter.Options?.Properties + }; + var workflowInstance = _workflowInstanceFactory.CreateWorkflowInstance(createWorkflowInstanceRequest); + + if (canStartResult.CanStart) + results.Add(new StartableWorkflowMatch(workflowInstance.Id, workflowInstance, workflowsFilter.Options?.CorrelationId, trigger.ActivityId, definitionId, trigger.Payload)); + } + + return results; + } + + private async Task> FindResumableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken) + { + var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload); + var correlationId = workflowsFilter.Options.CorrelationId; + var workflowInstanceId = workflowsFilter.Options.WorkflowInstanceId; + var activityInstanceId = workflowsFilter.Options.ActivityInstanceId; + var filter = new BookmarkFilter + { + Hash = hash, + CorrelationId = correlationId, + WorkflowInstanceId = workflowInstanceId, + ActivityInstanceId = activityInstanceId + }; + var bookmarks = await _bookmarkStore.FindManyAsync(filter, cancellationToken); + var collectedWorkflows = bookmarks.Select(b => new ResumableWorkflowMatch(b.WorkflowInstanceId, default, correlationId, b.BookmarkId, b.Payload)).ToList(); + return collectedWorkflows; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 8ba6c1bd8..ce1266342 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,11 +1,18 @@ +using Elsa.Workflows.Management; using Microsoft.Extensions.DependencyInjection; +using Proto.Cluster; namespace Elsa.Workflows.Runtime.ProtoActor.Services; /// /// Represents a Proto.Actor implementation of the workflows runtime. /// -public class ProtoActorWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime +public partial class ProtoActorWorkflowRuntime( + IServiceProvider serviceProvider, + IWorkflowDefinitionService workflowDefinitionService, + IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, + Cluster cluster, + IIdentityGenerator identityGenerator) : IWorkflowRuntime { /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs index aecfafdd3..8753a76bd 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs @@ -42,4 +42,6 @@ public interface IWorkflowClient /// Imports the specified . /// Task ImportStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); + + Task InstanceExistsAsync(CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index cbef32bcc..5ef752118 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -1,3 +1,13 @@ +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Matches; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Parameters; +using Elsa.Workflows.Runtime.Params; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Results; +using Elsa.Workflows.State; + namespace Elsa.Workflows.Runtime; /// @@ -21,4 +31,101 @@ public interface IWorkflowRuntime /// A new instance. /// The workflow instance itself doesn't have to exist yet. ValueTask CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default); + + + /// + /// Returns a value whether the specified workflow definition can create a new instance. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Creates a new workflow instance of the specified definition ID and executes it. + /// + /// The workflow definition ID to run. + /// Options for starting the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Starts all workflows with triggers matching the specified activity type and bookmark payload. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Tries to start a workflow and returns the result if successful. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Resumes an existing workflow instance. + /// + /// The ID of the workflow instance to resume. + /// Options for resuming the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default); + + /// + /// Resumes all workflows that are bookmarked on the specified activity type. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Executes a pending workflow. + /// + /// A workflow match to execute. + /// Options for executing the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default); + + /// + /// Cancels the execution of a workflow. + /// + /// The ID of the workflow instance to cancel. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + + /// + /// Finds all the workflows that can be started or resumed based on a query model. + /// + /// + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default); + + /// + /// Exports the of the specified workflow instance. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + + /// + /// Imports the specified . + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); + + /// + /// Updates the specified bookmark. + /// + /// The bookmark to update. + /// The cancellation token. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default); + + /// + /// Counts the number of workflow instances based on the provided query args. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs new file mode 100644 index 000000000..438446c5a --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs @@ -0,0 +1,11 @@ +using Elsa.Workflows.Runtime.Options; + +namespace Elsa.Workflows.Runtime.Filters; + +/// +/// A filter for finding workflows to trigger. +/// +/// The activity type name to trigger workflows for. +/// The bookmark payload to trigger workflows for. +/// The options to use when triggering workflows. +public record WorkflowsFilter(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsOptions Options); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs new file mode 100644 index 000000000..ba6f7669b --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs @@ -0,0 +1,6 @@ +using Elsa.Workflows.Management.Entities; + +namespace Elsa.Workflows.Runtime.Matches; + +public record ResumableWorkflowMatch(string WorkflowInstanceId, string? CorrelationId, string? BookmarkId, object? Payload) + : WorkflowMatch(CorrelationId, Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs new file mode 100644 index 000000000..3103b8c52 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs @@ -0,0 +1,4 @@ +namespace Elsa.Workflows.Runtime.Matches; + +public record StartableWorkflowMatch(string? CorrelationId, string? ActivityId, string? DefinitionId, object? Payload) + : WorkflowMatch(CorrelationId, Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs new file mode 100644 index 000000000..541a691c9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs @@ -0,0 +1,5 @@ +using Elsa.Workflows.Management.Entities; + +namespace Elsa.Workflows.Runtime.Matches; + +public record WorkflowMatch(string? CorrelationId, object? Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs index 73d1a9596..8a49e2273 100644 --- a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs @@ -22,6 +22,8 @@ public record RunWorkflowInstanceResponse /// public WorkflowSubStatus SubStatus { get; set; } + public ICollection Bookmarks { get; set; } = new List(); + /// /// Any incidents that occurred during the execution of the workflow instance. /// diff --git a/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs b/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs index e6c72ad91..f9d4ab40c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs +++ b/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs @@ -2,6 +2,7 @@ using Elsa.Workflows.Models; namespace Elsa.Workflows.Runtime.Params; +[Obsolete("This type is obsolete.")] public class ExecuteWorkflowParams { public string? CorrelationId { get; set; } @@ -11,4 +12,5 @@ public class ExecuteWorkflowParams public IDictionary? Properties { get; set; } public string? TriggerActivityId { get; set; } public string? ParentWorkflowInstanceId { get; set; } + public CancellationToken CancellationToken { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs new file mode 100644 index 000000000..9fc36c54f --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs @@ -0,0 +1,22 @@ +namespace Elsa.Workflows.Runtime.Requests; + +/// +/// Contains arguments to use for counting the number of workflow instances. +/// +public class CountRunningWorkflowsRequest +{ + /// + /// The workflow definition ID to include in the query. + /// + public string? DefinitionId { get; set; } + + /// + /// The workflow definition version to include in the query. + /// + public int? Version { get; set; } + + /// + /// The correlation ID to include in the query. + /// + public string? CorrelationId { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs index e3472732f..6745e54f7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs @@ -24,6 +24,8 @@ public record StartWorkflowResponse /// The sub-status of the workflow instance. /// public WorkflowSubStatus? SubStatus { get; set; } + + public ICollection Bookmarks { get; set; } = new List(); /// /// Any incidents that occurred during the execution of the workflow instance. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs index e1c2e79f7..3778ee265 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs @@ -12,7 +12,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio { var workflow = await GetWorkflowAsync(request, cancellationToken); - var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new WorkflowActivationStrategyEvaluationContext + var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() { Workflow = workflow, CorrelationId = request.CorrelationId @@ -41,6 +41,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio WorkflowInstanceId = runWorkflowResponse.WorkflowInstanceId, Status = runWorkflowResponse.Status, SubStatus = runWorkflowResponse.SubStatus, + Bookmarks = runWorkflowResponse.Bookmarks, Incidents = runWorkflowResponse.Incidents }; } @@ -56,7 +57,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(request.WorkflowDefinitionHandle, cancellationToken); if (workflowGraph == null) - throw new WorkflowGraphNotFoundException($"Workflow definition not found.", request.WorkflowDefinitionHandle); + throw new WorkflowGraphNotFoundException("Workflow definition not found.", request.WorkflowDefinitionHandle); return workflowGraph.Workflow; } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs index 6b73d2906..8497c99dc 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs @@ -96,6 +96,11 @@ public class LocalWorkflowClient( await workflowInstanceManager.SaveAsync(workflowInstance, cancellationToken); } + public Task InstanceExistsAsync(CancellationToken cancellationToken = default) + { + return workflowInstanceManager.ExistsAsync(workflowInstanceId, cancellationToken); + } + private async Task RunInstanceAsync(WorkflowInstance workflowInstance, RunWorkflowInstanceRequest request, CancellationToken cancellationToken = default) { var workflowState = workflowInstance.WorkflowState; diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs new file mode 100644 index 000000000..fcd4bb7a4 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -0,0 +1,241 @@ +using Elsa.Common.Models; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Matches; +using Elsa.Workflows.Runtime.Messages; +using Elsa.Workflows.Runtime.Options; +using Elsa.Workflows.Runtime.Parameters; +using Elsa.Workflows.Runtime.Params; +using Elsa.Workflows.Runtime.Requests; +using Elsa.Workflows.Runtime.Results; +using Elsa.Workflows.State; +using Open.Linq.AsyncExtensions; + +namespace Elsa.Workflows.Runtime; + +public partial class LocalWorkflowRuntime +{ + public async Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken); + var workflow = workflowGraph!.Workflow; + + var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() + { + Workflow = workflow, + CorrelationId = options?.CorrelationId, + CancellationToken = cancellationToken + }); + + return new CanStartWorkflowResult( + { + CanStart = canStart, + InstanceId = null + }; + } + + public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); + var createRequest = new CreateAndRunWorkflowInstanceRequest + { + Properties = options?.Properties, + CorrelationId = options?.CorrelationId, + Input = options?.Input, + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published), + ParentId = options?.ParentWorkflowInstanceId, + TriggerActivityId = options?.TriggerActivityId + }; + var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken); + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); + } + + public async Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var metadata = new StimulusMetadata + { + CorrelationId = options?.CorrelationId, + WorkflowInstanceId = options?.WorkflowInstanceId, + Properties = options?.Properties, + ActivityInstanceId = options?.ActivityInstanceId, + Input = options?.Input + }; + var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken); + var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList(); + return results; + } + + public async Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + return await StartWorkflowAsync(definitionId, options); + } + + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var workflowClient = await CreateClientAsync(workflowInstanceId, cancellationToken); + var exists = await workflowClient.InstanceExistsAsync(cancellationToken); + + if (!exists) + return null; + + var runWorkflowRequest = new RunWorkflowInstanceRequest + { + Input = options?.Input, + Properties = options?.Properties, + ActivityHandle = options?.ActivityHandle, + BookmarkId = options?.BookmarkId + }; + + var response = await workflowClient.RunInstanceAsync(runWorkflowRequest, cancellationToken); + + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); + } + + public async Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var metadata = new StimulusMetadata + { + CorrelationId = options?.CorrelationId, + WorkflowInstanceId = options?.WorkflowInstanceId, + Properties = options?.Properties, + ActivityInstanceId = options?.ActivityInstanceId, + Input = options?.Input + }; + var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken); + var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList(); + return results; + } + + public async Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var metadata = new StimulusMetadata + { + CorrelationId = options?.CorrelationId, + WorkflowInstanceId = options?.WorkflowInstanceId, + Properties = options?.Properties, + ActivityInstanceId = options?.ActivityInstanceId, + Input = options?.Input + }; + var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken); + var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList(); + return new(results); + } + + public async Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + if (match is StartableWorkflowMatch collectedStartableWorkflow) + { + var startOptions = new StartWorkflowRuntimeParams + { + CorrelationId = collectedStartableWorkflow.CorrelationId, + Input = options?.Input, + Properties = options?.Properties, + VersionOptions = VersionOptions.Published, + TriggerActivityId = collectedStartableWorkflow.ActivityId, + CancellationToken = cancellationToken + }; + + var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions); + return startResult with + { + TriggeredActivityId = collectedStartableWorkflow.ActivityId + }; + } + + var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!; + var runtimeOptions = new ResumeWorkflowRuntimeParams + { + CorrelationId = collectedResumableWorkflow.CorrelationId, + BookmarkId = collectedResumableWorkflow.BookmarkId, + Input = options?.Input, + Properties = options?.Properties, + CancellationToken = cancellationToken, + }; + + return (await ResumeWorkflowAsync(collectedResumableWorkflow.WorkflowInstanceId, runtimeOptions))!; + } + + public async Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + await client.CancelAsync(cancellationToken); + return new(true); + } + + public async Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) + { + var startableWorkflows = await FindStartableWorkflowsAsync(filter, cancellationToken); + var resumableWorkflows = await FindResumableWorkflowsAsync(filter, cancellationToken); + var results = startableWorkflows.Concat(resumableWorkflows).ToList(); + return results; + } + + public async Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + return await client.ExportStateAsync(cancellationToken); + } + + public async Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) + { + var client = await CreateClientAsync(workflowState.Id, cancellationToken); + await client.ImportStateAsync(workflowState, cancellationToken); + } + + public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) + { + await bookmarkStore.SaveAsync(bookmark, cancellationToken); + } + + public async Task CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) + { + var filter = new WorkflowInstanceFilter + { + DefinitionId = request.DefinitionId, + Version = request.Version, + CorrelationId = request.CorrelationId, + WorkflowStatus = WorkflowStatus.Running + }; + return await workflowInstanceStore.CountAsync(filter, cancellationToken); + } + + private async Task> FindStartableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) + { + var stimulusHash = stimulusHasher.Hash(filter.ActivityTypeName, filter.BookmarkPayload, filter.Options.ActivityInstanceId); + var triggerBoundWorkflows = await triggerBoundWorkflowService.FindManyAsync(stimulusHash, cancellationToken).ToList(); + var correlationId = filter.Options.CorrelationId; + + var query = + from triggerBoundWorkflow in triggerBoundWorkflows + from trigger in triggerBoundWorkflow.Triggers + select new StartableWorkflowMatch(correlationId, trigger.ActivityId, triggerBoundWorkflow.WorkflowGraph.Workflow.Identity.DefinitionId, filter.BookmarkPayload); + + return query.ToList(); + } + + private async Task> FindResumableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken) + { + var bookmarkOptions = new FindBookmarkOptions + { + CorrelationId = filter.Options.CorrelationId, + WorkflowInstanceId = filter.Options.WorkflowInstanceId, + ActivityInstanceId = filter.Options.ActivityInstanceId + }; + var bookmarkBoundWorkflows = await bookmarkBoundWorkflowService.FindManyAsync(filter.ActivityTypeName, filter.BookmarkPayload, bookmarkOptions, cancellationToken).ToList(); + + return ( + from bookmarkBoundWorkflow in bookmarkBoundWorkflows + from bookmark in bookmarkBoundWorkflow.Bookmarks + select new ResumableWorkflowMatch(bookmarkBoundWorkflow.WorkflowInstanceId, bookmark.CorrelationId, bookmark.Id, bookmark.Payload)) + .ToList(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs index dcfa181c7..cfd5309cf 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs @@ -1,3 +1,4 @@ +using Elsa.Workflows.Management; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime; @@ -7,7 +8,19 @@ namespace Elsa.Workflows.Runtime; /// It does not support clustering and is intended for single-node deployments only. /// For distributed deployments, use Proto.Actor or another distributed runtime. /// -public class LocalWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime +public partial class LocalWorkflowRuntime( + IServiceProvider serviceProvider, + IIdentityGenerator identityGenerator, + IWorkflowDefinitionService workflowDefinitionService, + IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, + IStimulusSender stimulusSender, + IBookmarkResumer bookmarkResumer, + IStimulusHasher stimulusHasher, + IWorkflowCanceler workflowCanceler, + IBookmarkStore bookmarkStore, + IWorkflowInstanceStore workflowInstanceStore, + ITriggerBoundWorkflowService triggerBoundWorkflowService, + IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime { /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default)