diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs index 989634a5f..9992ef931 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs @@ -1,5 +1,6 @@ using Elsa.Common.Contracts; using Elsa.Common.Models; +using Elsa.Extensions; using Elsa.Workflows.Core.Activities; using Elsa.Workflows.Core.Contracts; using Elsa.Workflows.Management.Contracts; @@ -21,7 +22,6 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP private readonly IActivitySerializer _activitySerializer; private readonly IPayloadSerializer _payloadSerializer; private readonly ISystemClock _systemClock; - private readonly IDistributedLockProvider _distributedLockProvider; private readonly IIdentityGraphService _identityGraphService; /// @@ -70,12 +70,7 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP private async Task AddOrUpdateAsync(MaterializedWorkflow materializedWorkflow, CancellationToken cancellationToken = default) { var workflow = materializedWorkflow.Workflow; - - var filter = new WorkflowDefinitionFilter - { - DefinitionId = workflow.Identity.DefinitionId, - VersionOptions = VersionOptions.SpecificVersion(workflow.Version) - }; + var definitionId = workflow.Identity.DefinitionId; // Serialize materializer context. var materializerContext = materializedWorkflow.MaterializerContext; @@ -84,33 +79,93 @@ public class DefaultWorkflowDefinitionStorePopulator : IWorkflowDefinitionStoreP // Serialize the workflow root. var workflowJson = _activitySerializer.Serialize(workflow.Root); - // Check if there's already a workflow definition stored with this workflow. - var existingDefinition = await _workflowDefinitionStore.FindAsync(filter, cancellationToken) ?? new WorkflowDefinition + // Check if there's already a workflow definition stored with this definition ID and version. + var specificVersionFilter = new WorkflowDefinitionFilter + { + DefinitionId = definitionId, + VersionOptions = VersionOptions.SpecificVersion(workflow.Identity.Version) + }; + + var existingDefinitionVersion = await _workflowDefinitionStore.FindAsync(specificVersionFilter, cancellationToken); + + // Setup a list to collect all workflow definitions to be persisted. + var workflowDefinitionsToSave = new HashSet(); + + if(existingDefinitionVersion != null) + workflowDefinitionsToSave.Add(existingDefinitionVersion); + + // If the workflow being added is configured to be the latest version, then we need to reset the current latest version. + if (workflow.Publication.IsLatest) + { + // Reset current latest definitions. + var filter = new WorkflowDefinitionFilter { DefinitionId = definitionId, VersionOptions = VersionOptions.Latest }; + var latestWorkflowDefinitions = (await _workflowDefinitionStore.FindManyAsync(filter, cancellationToken)).ToList(); + + // If the latest definitions contains definitions with the same ID then we need to replace them with the latest workflow definitions. + SyncExistingCopies(latestWorkflowDefinitions, workflowDefinitionsToSave); + + foreach (var latestWorkflowDefinition in latestWorkflowDefinitions) + { + latestWorkflowDefinition.IsLatest = false; + workflowDefinitionsToSave.Add(latestWorkflowDefinition); + } + } + + // If the workflow being added is configured to be the published version, then we need to reset the current published version. + if (workflow.Publication.IsPublished) + { + // Reset current published definitions. + var filter = new WorkflowDefinitionFilter { DefinitionId = definitionId, VersionOptions = VersionOptions.Published }; + var publishedWorkflowDefinitions = (await _workflowDefinitionStore.FindManyAsync(filter, cancellationToken)).ToList(); + + // If the published workflow definitions contains definitions with the same ID as definitions in the latest workflow definitions, then we need to replace them with the latest workflow definitions. + SyncExistingCopies(publishedWorkflowDefinitions, workflowDefinitionsToSave); + + foreach (var publishedWorkflowDefinition in publishedWorkflowDefinitions) + { + publishedWorkflowDefinition.IsPublished = false; + workflowDefinitionsToSave.Add(publishedWorkflowDefinition); + } + } + + var workflowDefinition = existingDefinitionVersion ?? new WorkflowDefinition { DefinitionId = workflow.Identity.DefinitionId, Id = workflow.Identity.Id, Version = workflow.Identity.Version }; + + workflowDefinition.Description = workflow.WorkflowMetadata.Description; + workflowDefinition.Name = workflow.WorkflowMetadata.Name; + workflowDefinition.IsLatest = workflow.Publication.IsLatest; + workflowDefinition.IsPublished = workflow.Publication.IsPublished; + workflowDefinition.IsReadonly = workflow.IsReadonly; + workflowDefinition.CustomProperties = workflow.CustomProperties; + workflowDefinition.Variables = workflow.Variables; + workflowDefinition.Inputs = workflow.Inputs; + workflowDefinition.Outputs = workflow.Outputs; + workflowDefinition.Outcomes = workflow.Outcomes; + workflowDefinition.StringData = workflowJson; + workflowDefinition.CreatedAt = workflow.WorkflowMetadata.CreatedAt == default ? _systemClock.UtcNow : workflow.WorkflowMetadata.CreatedAt; + workflowDefinition.Options = workflow.Options; + workflowDefinition.ProviderName = materializedWorkflow.ProviderName; + workflowDefinition.MaterializerContext = materializerContextJson; + workflowDefinition.MaterializerName = materializedWorkflow.MaterializerName; - existingDefinition.Description = workflow.WorkflowMetadata.Description; - existingDefinition.Name = workflow.WorkflowMetadata.Name; - existingDefinition.IsLatest = workflow.Publication.IsLatest; - existingDefinition.IsPublished = workflow.Publication.IsPublished; - existingDefinition.IsReadonly = workflow.IsReadonly; - existingDefinition.CustomProperties = workflow.CustomProperties; - existingDefinition.Variables = workflow.Variables; - existingDefinition.Inputs = workflow.Inputs; - existingDefinition.Outputs = workflow.Outputs; - existingDefinition.Outcomes = workflow.Outcomes; - existingDefinition.StringData = workflowJson; - existingDefinition.CreatedAt = workflow.WorkflowMetadata.CreatedAt == default ? _systemClock.UtcNow : workflow.WorkflowMetadata.CreatedAt; - existingDefinition.Options = workflow.Options; - existingDefinition.ProviderName = materializedWorkflow.ProviderName; - existingDefinition.MaterializerContext = materializerContextJson; - existingDefinition.MaterializerName = materializedWorkflow.MaterializerName; - - await _workflowDefinitionStore.SaveAsync(existingDefinition, cancellationToken); + workflowDefinitionsToSave.Add(workflowDefinition); + await _workflowDefinitionStore.SaveManyAsync(workflowDefinitionsToSave, cancellationToken); } private async Task IndexTriggersAsync(MaterializedWorkflow workflow, CancellationToken cancellationToken) => await _triggerIndexer.IndexTriggersAsync(workflow.Workflow, cancellationToken); + + /// + /// Syncs the items in the primary list with existing items in the secondary list, even when the object instances are not the same (but their IDs are). + /// + private void SyncExistingCopies(List primary, HashSet secondary) + { + var ids = secondary.Select(x => x.Id).Distinct().ToList(); + var latestWorkflowDefinitions = primary.Where(x => ids.Contains(x.Id)).ToList(); + primary.RemoveAll(x => latestWorkflowDefinitions.Contains(x)); + primary.AddRange(secondary.Where(x => ids.Contains(x.Id))); + } } \ No newline at end of file