< Summary

Information
Class: Elsa.Workflows.Management.Services.WorkflowReferences
Assembly: Elsa.Workflows.Management
File(s): /home/runner/work/elsa-core/elsa-core/src/modules/Elsa.Workflows.Management/Services/WorkflowReferenceUpdater.cs
Line coverage
100%
Covered lines: 1
Uncovered lines: 0
Coverable lines: 1
Total lines: 229
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
get_ReferencedDefinitionId()100%11100%

File(s)

/home/runner/work/elsa-core/elsa-core/src/modules/Elsa.Workflows.Management/Services/WorkflowReferenceUpdater.cs

#LineLine coverage
 1using System.Runtime.CompilerServices;
 2using Elsa.Common.Models;
 3using Elsa.Extensions;
 4using Elsa.Workflows.Activities;
 5using Elsa.Workflows.Management.Activities.WorkflowDefinitionActivity;
 6using Elsa.Workflows.Management.Entities;
 7using Elsa.Workflows.Management.Models;
 8using Elsa.Workflows.Models;
 9
 10namespace Elsa.Workflows.Management.Services;
 11
 1112internal record WorkflowReferences(string ReferencedDefinitionId, ICollection<string> ReferencingDefinitionIds);
 13
 14internal record UpdatedWorkflowDefinition(WorkflowDefinition Definition, WorkflowGraph NewGraph);
 15
 16public class WorkflowReferenceUpdater(
 17    IWorkflowDefinitionPublisher publisher,
 18    IWorkflowDefinitionService workflowDefinitionService,
 19    IWorkflowDefinitionStore workflowDefinitionStore,
 20    IWorkflowReferenceQuery workflowReferenceQuery,
 21    WorkflowDefinitionActivityDescriptorFactory workflowDefinitionActivityDescriptorFactory,
 22    IActivityRegistry activityRegistry,
 23    IApiSerializer serializer)
 24    : IWorkflowReferenceUpdater
 25{
 26    private bool _isUpdating;
 27
 28    public async Task<UpdateWorkflowReferencesResult> UpdateWorkflowReferencesAsync(
 29        WorkflowDefinition referencedDefinition,
 30        CancellationToken cancellationToken = default)
 31    {
 32        if (_isUpdating ||
 33            referencedDefinition.Options is not { UsableAsActivity: true, AutoUpdateConsumingWorkflows: true })
 34            return new([]);
 35
 36        var allWorkflowReferences = await GetReferencingWorkflowDefinitionIdsAsync(referencedDefinition.DefinitionId, ca
 37        var filteredWorkflowReferences = allWorkflowReferences
 38            .Where(r => r.ReferencingDefinitionIds.Any())
 39            .DistinctBy(r => r.ReferencedDefinitionId)
 40            .ToList();
 41
 42        var referencingIds = filteredWorkflowReferences.SelectMany(r => r.ReferencingDefinitionIds).Distinct().ToList();
 43        var referencedIds = filteredWorkflowReferences.Select(r => r.ReferencedDefinitionId).Distinct().ToList();
 44
 45        var referencingWorkflowGraphs = (await workflowDefinitionService.FindWorkflowGraphsAsync(new()
 46            {
 47                DefinitionIds = referencingIds,
 48                VersionOptions = VersionOptions.Latest,
 49                IsReadonly = false
 50            }, cancellationToken))
 51            .ToDictionary(g => g.Workflow.Identity.DefinitionId);
 52
 53        var referencedWorkflowDefinitionList = (await workflowDefinitionStore.FindManyAsync(new()
 54        {
 55            DefinitionIds = referencedIds,
 56            VersionOptions = VersionOptions.Published,
 57            IsReadonly = false
 58        }, cancellationToken)).ToList();
 59
 60        var referencedWorkflowDefinitionsPublished = referencedWorkflowDefinitionList
 61            .GroupBy(x => x.DefinitionId)
 62            .Select(group =>
 63            {
 64                var publishedVersion = group.FirstOrDefault(x => x.IsPublished);
 65                return publishedVersion ?? group.First();
 66            })
 67            .ToDictionary(d => d.DefinitionId);
 68
 69        var initialPublicationState = new Dictionary<string, bool>();
 70
 71        foreach (var workflowGraph in referencingWorkflowGraphs)
 72            initialPublicationState[workflowGraph.Key] = workflowGraph.Value.Workflow.Publication.IsPublished;
 73
 74        // Add the initially referenced definition
 75        referencedWorkflowDefinitionsPublished[referencedDefinition.DefinitionId] = referencedDefinition;
 76
 77        // Build dependency map for topological sorting
 78        var dependencyMap = filteredWorkflowReferences
 79            .SelectMany(r => r.ReferencingDefinitionIds.Select(id => (id, r.ReferencedDefinitionId)))
 80            .ToLookup(x => x.id, x => x.ReferencedDefinitionId);
 81
 82        // Perform topological sort to ensure dependent workflows are processed in the right order
 83        var sortedWorkflowIds = referencingIds
 84            .TSort(id => dependencyMap[id], true)
 85            // Only process workflows that exist in our referencing workflows dictionary
 86            .Where(id => referencingWorkflowGraphs.ContainsKey(id))
 87            .ToList();
 88
 89        var updatedWorkflows = new Dictionary<string, UpdatedWorkflowDefinition>();
 90
 91        // Create a cache for drafts that we've already created during this operation
 92        var draftCache = new Dictionary<string, WorkflowDefinition>();
 93
 94        foreach (var id in sortedWorkflowIds)
 95        {
 96            if (!referencingWorkflowGraphs.TryGetValue(id, out var graph) || !dependencyMap[id].Any())
 97                continue;
 98
 99            foreach (var refId in dependencyMap[id])
 100            {
 101                var target = referencedWorkflowDefinitionsPublished.GetValueOrDefault(refId);
 102                if (target == null) continue;
 103
 104                var updated = await UpdateWorkflowAsync(graph, target, draftCache, initialPublicationState, cancellation
 105                if (updated == null) continue;
 106
 107                graph = updated.NewGraph;
 108                updatedWorkflows[updated.Definition.DefinitionId] = updated;
 109                referencedWorkflowDefinitionsPublished[id] = updated.Definition;
 110                draftCache[id] = updated.Definition;
 111                referencingWorkflowGraphs[id] = updated.NewGraph;
 112            }
 113        }
 114
 115        _isUpdating = true;
 116        foreach (var updatedWorkflow in updatedWorkflows.Values)
 117        {
 118            var requiresPublication = initialPublicationState.GetValueOrDefault(updatedWorkflow.Definition.DefinitionId)
 119            if (requiresPublication)
 120                await publisher.PublishAsync(updatedWorkflow.Definition, cancellationToken);
 121            else
 122                await publisher.SaveDraftAsync(updatedWorkflow.Definition, cancellationToken);
 123        }
 124
 125        _isUpdating = false;
 126
 127        return new(updatedWorkflows.Select(u => u.Value.Definition));
 128    }
 129
 130    private async IAsyncEnumerable<WorkflowReferences> GetReferencingWorkflowDefinitionIdsAsync(
 131        string definitionId,
 132        [EnumeratorCancellation] CancellationToken cancellationToken,
 133        HashSet<string>? visitedIds = null)
 134    {
 135        visitedIds ??= new();
 136
 137        // If we've already processed this definition ID, skip it to prevent infinite recursion.
 138        if (!visitedIds.Add(definitionId))
 139            yield break;
 140
 141        var refs = (await workflowReferenceQuery.ExecuteAsync(definitionId, cancellationToken)).ToList();
 142        yield return new(definitionId, refs);
 143
 144        foreach (var id in refs)
 145        {
 146            await foreach (var child in GetReferencingWorkflowDefinitionIdsAsync(id, cancellationToken, visitedIds))
 147                yield return child;
 148        }
 149    }
 150
 151    private async Task<UpdatedWorkflowDefinition?> UpdateWorkflowAsync(
 152        WorkflowGraph graph,
 153        WorkflowDefinition target,
 154        Dictionary<string, WorkflowDefinition> draftCache,
 155        Dictionary<string, bool> initialPublicationState,
 156        CancellationToken cancellationToken)
 157    {
 158        var willTargetBePublished = initialPublicationState.GetValueOrDefault(target.DefinitionId, target.IsPublished);
 159        if (!willTargetBePublished)
 160            return null;
 161
 162        var id = graph.Workflow.Identity.DefinitionId;
 163        var draft = await GetOrCreateDraftAsync(id, draftCache, cancellationToken);
 164        if (draft == null) return null;
 165
 166        var newGraph = await workflowDefinitionService.MaterializeWorkflowAsync(draft, cancellationToken);
 167        var outdated = FindActivities(newGraph.Root, target.DefinitionId)
 168            .Where(a => a.WorkflowDefinitionVersionId != target.Id)
 169            .ToList();
 170
 171        if (!outdated.Any()) return null;
 172
 173        foreach (var act in outdated)
 174        {
 175            act.WorkflowDefinitionVersionId = target.Id;
 176            act.Version = target.Version;
 177            act.LatestAvailablePublishedVersionId = target.Id;
 178            act.LatestAvailablePublishedVersion = target.Version;
 179        }
 180
 181        if (newGraph.Root.Activity is Workflow wf)
 182            draft.StringData = serializer.Serialize(wf.Root);
 183
 184        return new(draft, newGraph);
 185    }
 186
 187    private async Task<WorkflowDefinition?> GetOrCreateDraftAsync(
 188        string definitionId,
 189        Dictionary<string, WorkflowDefinition> draftCache,
 190        CancellationToken cancellationToken)
 191    {
 192        // Check if we already have a draft for this workflow
 193        if (draftCache.TryGetValue(definitionId, out var cachedDraft))
 194            return cachedDraft;
 195
 196        // Create or get a draft for this workflow
 197        var draft = await publisher.GetDraftAsync(definitionId, VersionOptions.Latest, cancellationToken);
 198        if (draft == null) return null;
 199
 200        // Store the draft in the cache for potential future use
 201        draftCache[definitionId] = draft;
 202
 203        // Get the current published version of the workflow definition.
 204        var publishedVersion = await workflowDefinitionStore.FindAsync(
 205            WorkflowDefinitionHandle.ByDefinitionId(definitionId, VersionOptions.Published).ToFilter(),
 206            cancellationToken);
 207
 208        // Update the activity registry to be able to materialize the workflow.
 209        var activityDescriptor = workflowDefinitionActivityDescriptorFactory.CreateDescriptor(draft, publishedVersion);
 210        activityRegistry.Add(typeof(WorkflowDefinitionActivityProvider), activityDescriptor);
 211
 212        return draft;
 213    }
 214
 215    private static IEnumerable<WorkflowDefinitionActivity> FindActivities(ActivityNode node, string definitionId)
 216    {
 217        // Do not drill into activities that are WorkflowDefinitionActivity
 218        if (node.Activity is WorkflowDefinitionActivity)
 219            yield break;
 220
 221        foreach (var child in node.Children)
 222        {
 223            if (child.Activity is WorkflowDefinitionActivity activity && activity.WorkflowDefinitionId == definitionId)
 224                yield return activity;
 225            foreach (var grandChildActivity in FindActivities(child, definitionId))
 226                yield return grandChildActivity;
 227        }
 228    }
 229}
Generated by: ReportGenerator 5.5.0.0
11/27/2025 - 20:46:54
GitHub | reportgenerator.io

Methods/Properties

get_ReferencedDefinitionId()