SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Fix import issue with workflow definition store populator

Browse source 
Fixes #4224
This commit is contained in:
Sipke Schoorstra 2023-07-28 13:13:34 +02:00
parent 6cd3bd2d61
commit b6f41aff9b
1 changed files with 82 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

109
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowDefinitionStorePopulator.cs
View file

@ -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;
/// <summary>
@ -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<WorkflowDefinition>();
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);
/// <summary>
/// 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).
/// </summary>
private void SyncExistingCopies(List<WorkflowDefinition> primary, HashSet<WorkflowDefinition> 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)));
}
}