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

Made Updating workflow references part of Publish logic

Browse source
This commit is contained in:
Raymond den Haan 2024-05-29 15:39:15 +02:00 committed by raymonddenhaan
parent de433b39f9
commit fb3b5339f4
16 changed files with 123 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Contracts/IWorkflowDefinitionsApi.cs
View file

@ -73,7 +73,7 @@ public interface IWorkflowDefinitionsApi
/// <param name="request">The request containing the workflow definition to save.</param>
/// <param name="cancellationToken">The cancellation token.</param>
[Post("/workflow-definitions")]
Task<WorkflowDefinition> SaveAsync(SaveWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
Task<SaveWorkflowDefinitionResponse> SaveAsync(SaveWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
/// <summary>
/// Deletes a workflow definition.
@ -99,7 +99,7 @@ public interface IWorkflowDefinitionsApi
/// <param name="cancellationToken">The cancellation token.</param>
[Post("/workflow-definitions/{definitionId}/publish")]
[Headers(MediaTypeNames.Application.Json)]
Task<WorkflowDefinition> PublishAsync(string definitionId, PublishWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
Task<SaveWorkflowDefinitionResponse> PublishAsync(string definitionId, PublishWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
/// <summary>
/// Retracts a workflow definition.

3
src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Responses/BulkPublishWorkflowDefinitionsResponse.cs
View file

@ -6,4 +6,5 @@ namespace Elsa.Api.Client.Resources.WorkflowDefinitions.Responses;
public record BulkPublishWorkflowDefinitionsResponse(
ICollection<string> Published,
ICollection<string> AlreadyPublished,
ICollection<string> NotFound);
ICollection<string> NotFound,
ICollection<string> ConsumingUpdated);

8
src/clients/Elsa.Api.Client/Resources/WorkflowDefinitions/Responses/SaveWorkflowDefinitionResponse.cs Normal file
View file

@ -0,0 +1,8 @@
using Elsa.Api.Client.Resources.WorkflowDefinitions.Models;
namespace Elsa.Api.Client.Resources.WorkflowDefinitions.Responses;
/// <summary>
/// Represents the response for saving a workflow definition.
/// </summary>
public record SaveWorkflowDefinitionResponse(WorkflowDefinition WorkflowDefinition, int ConsumingWorkflowCount);

10
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkPublish/Endpoint.cs
View file

@ -33,6 +33,7 @@ internal class BulkPublish(IWorkflowDefinitionStore store, IWorkflowDefinitionPu
var notFound = new List<string>();
var alreadyPublished = new List<string>();
var skipped = new List<string>();
var consumingUpdated = new List<string>();
var definitions = (await store.FindManyAsync(new WorkflowDefinitionFilter
{
@ -62,10 +63,15 @@ internal class BulkPublish(IWorkflowDefinitionStore store, IWorkflowDefinitionPu
continue;
}
await workflowDefinitionPublisher.PublishAsync(definition, cancellationToken);
var result = await workflowDefinitionPublisher.PublishAsync(definition, cancellationToken);
published.Add(definitionId);
if (result.ConsumingWorkflows?.Any() == true)
{
consumingUpdated.AddRange(result.ConsumingWorkflows.Select(x => x.DefinitionId));
}
}
return new Response(published, alreadyPublished, notFound, skipped);
return new Response(published, alreadyPublished, notFound, skipped, consumingUpdated);
}
}

3
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/BulkPublish/Models.cs
View file

@ -5,10 +5,11 @@ internal class Request
public ICollection<string> DefinitionIds { get; set; } = default!;
}
internal class Response(ICollection<string> published, ICollection<string> alreadyPublished, ICollection<string> notFound, ICollection<string> skipped)
internal class Response(ICollection<string> published, ICollection<string> alreadyPublished, ICollection<string> notFound, ICollection<string> skipped, ICollection<string> consumingUpdated)
{
public ICollection<string> Published { get; } = published;
public ICollection<string> AlreadyPublished { get; } = alreadyPublished;
public ICollection<string> NotFound { get; } = notFound;
public ICollection<string> Skipped { get; } = skipped;
public ICollection<string> ConsumingUpdated { get; } = consumingUpdated;
}

9
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Post/Endpoint.cs
View file

@ -92,9 +92,11 @@ internal class Post(
draft.Outcomes = outcomes;
draft.Options = model.Options ?? new WorkflowOptions();
PublishWorkflowDefinitionResult? result = null;
if (request.Publish.GetValueOrDefault(false))
{
var result = await workflowDefinitionPublisher.PublishAsync(draft, cancellationToken);
result = await workflowDefinitionPublisher.PublishAsync(draft, cancellationToken);
if (!result.Succeeded)
{
@ -110,10 +112,11 @@ internal class Post(
await workflowDefinitionPublisher.SaveDraftAsync(draft, cancellationToken);
}
var response = await linker.MapAsync(draft, cancellationToken);
var mappedDefinition = await linker.MapAsync(draft, cancellationToken);
var response = new Response(mappedDefinition, result?.ConsumingWorkflows?.Count() ?? 0);
if (isNew)
await SendCreatedAtAsync<GetByDefinitionId.GetByDefinitionId>(new { definitionId }, response, cancellation: cancellationToken);
await SendCreatedAtAsync<GetByDefinitionId.GetByDefinitionId>(new { definitionId }, mappedDefinition, cancellation: cancellationToken);
else
{
await HttpContext.Response.WriteAsJsonAsync(response, serializerOptions, cancellationToken);

5
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Post/Models.cs Normal file
View file

@ -0,0 +1,5 @@
using Elsa.Workflows.Api.Models;
namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Post;
internal record Response(LinkedWorkflowDefinitionModel WorkflowDefinition, int ConsumingWorkflowCount);

7
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Publish/Endpoint.cs
View file

@ -55,10 +55,11 @@ internal class Publish(IWorkflowDefinitionStore store, IWorkflowDefinitionPublis
return;
}
await workflowDefinitionPublisher.PublishAsync(definition, cancellationToken);
var response = await linker.MapAsync(definition, cancellationToken);
var result = await workflowDefinitionPublisher.PublishAsync(definition, cancellationToken);
var mappedDefinition = await linker.MapAsync(definition, cancellationToken);
var response = new Response(mappedDefinition, result.ConsumingWorkflows?.Count() ?? 0);
// We do not want to include composite root activities in the response.
var serializerOptions = serializer.GetOptions().Clone();
serializerOptions.Converters.Add(new JsonIgnoreCompositeRootConverterFactory());

6
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Publish/Models.cs
View file

@ -1,6 +1,10 @@
using Elsa.Workflows.Api.Models;
namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Publish;
internal class Request
{
public string DefinitionId { get; set; } = default!;
}
}
internal record Response(LinkedWorkflowDefinitionModel WorkflowDefinition, int ConsumingWorkflowCount);

4
src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/UpdateReferences/Endpoint.cs
View file

@ -10,7 +10,7 @@ using Microsoft.AspNetCore.Authorization;
namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.UpdateReferences;
[PublicAPI]
internal class UpdateReferences(IWorkflowDefinitionStore store, IWorkflowDefinitionManager workflowDefinitionManager, IAuthorizationService authorizationService)
internal class UpdateReferences(IWorkflowDefinitionStore store, IWorkflowDefinitionPublisher workflowDefinitionPublisher, IAuthorizationService authorizationService)
: ElsaEndpoint<Request, Response>
{
public override void Configure()
@ -43,7 +43,7 @@ internal class UpdateReferences(IWorkflowDefinitionStore store, IWorkflowDefinit
return;
}
var affectedWorkflows = await workflowDefinitionManager.UpdateReferencesInConsumingWorkflows(definition, cancellationToken);
var affectedWorkflows = await workflowDefinitionPublisher.UpdateReferencesInConsumingWorkflows(definition, cancellationToken);
var response = new Response(affectedWorkflows.Select(w => w.Name ?? w.DefinitionId));
await SendOkAsync(response, cancellationToken);
}

8
src/modules/Elsa.Workflows.Management/Contracts/IWorkflowDefinitionManager.cs
View file

@ -64,12 +64,4 @@ public interface IWorkflowDefinitionManager
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The new workflow definition.</returns>
Task<WorkflowDefinition> RevertVersionAsync(string definitionId, int version, CancellationToken cancellationToken = default);
/// <summary>
/// Updates all referencing workflow definitions to use the version of the specified workflow definition.
/// </summary>
/// <param name="dependency">The workflow definition to update references for.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The updated workflow definitions.</returns>
Task<IEnumerable<WorkflowDefinition>> UpdateReferencesInConsumingWorkflows(WorkflowDefinition dependency, CancellationToken cancellationToken = default);
}

8
src/modules/Elsa.Workflows.Management/Contracts/IWorkflowDefinitionPublisher.cs
View file

@ -68,4 +68,12 @@ public interface IWorkflowDefinitionPublisher
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The saved workflow definition.</returns>
Task<WorkflowDefinition> SaveDraftAsync(WorkflowDefinition definition, CancellationToken cancellationToken = default);
/// <summary>
/// Updates all referencing workflow definitions to use the version of the specified workflow definition.
/// </summary>
/// <param name="dependency">The workflow definition to update references for.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns>The updated workflow definitions.</returns>
Task<IEnumerable<WorkflowDefinition>> UpdateReferencesInConsumingWorkflows(WorkflowDefinition dependency, CancellationToken cancellationToken = default);
}

4
src/modules/Elsa.Workflows.Management/Models/PublishWorkflowDefinitionResult.cs
View file

@ -1,6 +1,8 @@
using Elsa.Workflows.Management.Entities;
namespace Elsa.Workflows.Management.Models;
/// <summary>
/// Represents the result of publishing a workflow definition.
/// </summary>
public record PublishWorkflowDefinitionResult(bool Succeeded, ICollection<WorkflowValidationError> ValidationErrors);
public record PublishWorkflowDefinitionResult(bool Succeeded, ICollection<WorkflowValidationError> ValidationErrors, IEnumerable<WorkflowDefinition>? ConsumingWorkflows);

57
src/modules/Elsa.Workflows.Management/Services/WorkflowDefinitionManager.cs
View file

@ -1,8 +1,6 @@
using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Management.Activities.WorkflowDefinitionActivity;
using Elsa.Workflows.Management.Contracts;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
@ -17,8 +15,6 @@ public class WorkflowDefinitionManager : IWorkflowDefinitionManager
private readonly INotificationSender _notificationSender;
private readonly IWorkflowDefinitionPublisher _workflowPublisher;
private readonly IIdentityGenerator _identityGenerator;
private readonly IActivitySerializer _activitySerializer;
private readonly IActivityVisitor _activityVisitor;
/// <summary>
/// Constructor.
@ -27,16 +23,12 @@ public class WorkflowDefinitionManager : IWorkflowDefinitionManager
IWorkflowDefinitionStore store,
INotificationSender notificationSender,
IWorkflowDefinitionPublisher workflowPublisher,
IIdentityGenerator identityGenerator,
IActivitySerializer activitySerializer,
IActivityVisitor activityVisitor)
IIdentityGenerator identityGenerator)
{
_store = store;
_notificationSender = notificationSender;
_workflowPublisher = workflowPublisher;
_identityGenerator = identityGenerator;
_activitySerializer = activitySerializer;
_activityVisitor = activityVisitor;
}
/// <inheritdoc />
@ -143,53 +135,6 @@ public class WorkflowDefinitionManager : IWorkflowDefinitionManager
return draft;
}
/// <inheritdoc />
public async Task<IEnumerable<WorkflowDefinition>> UpdateReferencesInConsumingWorkflows(WorkflowDefinition dependency, CancellationToken cancellationToken = default)
{
var updatedWorkflowDefinitions = new List<WorkflowDefinition>();
var workflowDefinitions = (await _store.FindManyAsync(new WorkflowDefinitionFilter
{
VersionOptions = VersionOptions.LatestOrPublished
}, cancellationToken)).ToList();
// Remove the dependency from the list of workflow definitions to consider.
workflowDefinitions = workflowDefinitions.Where(x => x.DefinitionId != dependency.DefinitionId).ToList();
foreach (var definition in workflowDefinitions)
{
var root = _activitySerializer.Deserialize(definition.StringData!);
var graph = await _activityVisitor.VisitAsync(root, cancellationToken);
var flattenedList = graph.Flatten().ToList();
var definitionId = dependency.DefinitionId;
var version = dependency.Version;
var nodes = flattenedList
.Where(x => x.Activity is WorkflowDefinitionActivity workflowDefinitionActivity && workflowDefinitionActivity.WorkflowDefinitionId == definitionId)
.ToList();
foreach (var node in nodes.Where(activity => activity.Activity.Version < version))
{
var activity = (WorkflowDefinitionActivity)node.Activity;
activity.Version = version;
activity.WorkflowDefinitionVersionId = dependency.Id;
if (!updatedWorkflowDefinitions.Contains(definition))
updatedWorkflowDefinitions.Add(definition);
}
if (updatedWorkflowDefinitions.Contains(definition))
{
var serializedData = _activitySerializer.Serialize(root);
definition.StringData = serializedData;
}
}
if (updatedWorkflowDefinitions.Any())
await _store.SaveManyAsync(updatedWorkflowDefinitions, cancellationToken);
return updatedWorkflowDefinitions;
}
private async Task EnsureLastVersionIsLatestAsync(IEnumerable<string> definitionIds, CancellationToken cancellationToken)
{
foreach (var definitionId in definitionIds)

66
src/modules/Elsa.Workflows.Management/Services/WorkflowDefinitionPublisher.cs
View file

@ -1,9 +1,11 @@
using Elsa.Common.Contracts;
using Elsa.Common.Entities;
using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Activities;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Management.Activities.WorkflowDefinitionActivity;
using Elsa.Workflows.Management.Contracts;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
@ -21,6 +23,7 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
private readonly IWorkflowDefinitionStore _workflowDefinitionStore;
private readonly INotificationSender _notificationSender;
private readonly IIdentityGenerator _identityGenerator;
private readonly IActivityVisitor _activityVisitor;
private readonly IActivitySerializer _activitySerializer;
private readonly IRequestSender _requestSender;
private readonly ISystemClock _systemClock;
@ -33,6 +36,7 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
IWorkflowDefinitionStore workflowDefinitionStore,
INotificationSender notificationSender,
IIdentityGenerator identityGenerator,
IActivityVisitor activityVisitor,
IActivitySerializer activitySerializer,
IRequestSender requestSender,
ISystemClock systemClock)
@ -41,6 +45,7 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
_workflowDefinitionStore = workflowDefinitionStore;
_notificationSender = notificationSender;
_identityGenerator = identityGenerator;
_activityVisitor = activityVisitor;
_activitySerializer = activitySerializer;
_requestSender = requestSender;
_systemClock = systemClock;
@ -77,7 +82,7 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
return new PublishWorkflowDefinitionResult(false, new List<WorkflowValidationError>
{
new("Workflow definition not found.")
});
}, null);
return await PublishAsync(definition, cancellationToken);
}
@ -90,7 +95,7 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
var validationErrors = responses.SelectMany(r => r.ValidationErrors).ToList();
if (validationErrors.Any())
return new PublishWorkflowDefinitionResult(false, validationErrors);
return new PublishWorkflowDefinitionResult(false, validationErrors, null);
await _notificationSender.SendAsync(new WorkflowDefinitionPublishing(definition), cancellationToken);
@ -113,7 +118,15 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
await _workflowDefinitionStore.SaveAsync(definition, cancellationToken);
await _notificationSender.SendAsync(new WorkflowDefinitionPublished(definition), cancellationToken);
return new PublishWorkflowDefinitionResult(true, validationErrors);
var consumingWorkflows = new List<WorkflowDefinition>();
if (definition.Options.UsableAsActivity == true)
{
consumingWorkflows.AddRange(await UpdateReferencesInConsumingWorkflows(definition, cancellationToken));
}
return new PublishWorkflowDefinitionResult(true, validationErrors, consumingWorkflows);
}
/// <inheritdoc />
@ -196,6 +209,53 @@ public class WorkflowDefinitionPublisher : IWorkflowDefinitionPublisher
return draft;
}
/// <inheritdoc />
public async Task<IEnumerable<WorkflowDefinition>> UpdateReferencesInConsumingWorkflows(WorkflowDefinition dependency, CancellationToken cancellationToken = default)
{
var updatedWorkflowDefinitions = new List<WorkflowDefinition>();
var workflowDefinitions = (await _workflowDefinitionStore.FindManyAsync(new WorkflowDefinitionFilter
{
VersionOptions = VersionOptions.LatestOrPublished
}, cancellationToken)).ToList();
// Remove the dependency from the list of workflow definitions to consider.
workflowDefinitions = workflowDefinitions.Where(x => x.DefinitionId != dependency.DefinitionId).ToList();
foreach (var definition in workflowDefinitions)
{
var root = _activitySerializer.Deserialize(definition.StringData!);
var graph = await _activityVisitor.VisitAsync(root, cancellationToken);
var flattenedList = graph.Flatten().ToList();
var definitionId = dependency.DefinitionId;
var version = dependency.Version;
var nodes = flattenedList
.Where(x => x.Activity is WorkflowDefinitionActivity workflowDefinitionActivity && workflowDefinitionActivity.WorkflowDefinitionId == definitionId)
.ToList();
foreach (var node in nodes.Where(activity => activity.Activity.Version < version))
{
var activity = (WorkflowDefinitionActivity)node.Activity;
activity.Version = version;
activity.WorkflowDefinitionVersionId = dependency.Id;
if (!updatedWorkflowDefinitions.Contains(definition))
updatedWorkflowDefinitions.Add(definition);
}
if (updatedWorkflowDefinitions.Contains(definition))
{
var serializedData = _activitySerializer.Serialize(root);
definition.StringData = serializedData;
}
}
if (updatedWorkflowDefinitions.Any())
await _workflowDefinitionStore.SaveManyAsync(updatedWorkflowDefinitions, cancellationToken);
return updatedWorkflowDefinitions;
}
private WorkflowDefinition Initialize(WorkflowDefinition definition)
{
if (definition.Id == null!)

8
test/integration/Elsa.Workflows.IntegrationTests/Scenarios/DependencyWorkflowsPublishing/Tests.cs
View file

@ -21,7 +21,7 @@ public class Tests
private readonly IWorkflowDefinitionPublisher _workflowDefinitionPublisher;
private readonly IActivitySerializer _activitySerializer;
private readonly IActivityVisitor _activityVisitor;
private readonly IWorkflowDefinitionManager _workflowManager;
// private readonly IWorkflowDefinitionManager _workflowManager;
/// <summary>
/// Initializes a new instance of the <see cref="Tests"/> class.
@ -34,7 +34,7 @@ public class Tests
.Build();
_workflowDefinitionPublisher = _services.GetRequiredService<IWorkflowDefinitionPublisher>();
_workflowManager = _services.GetRequiredService<IWorkflowDefinitionManager>();
// _workflowManager = _services.GetRequiredService<IWorkflowDefinitionManager>();
_activitySerializer = _services.GetRequiredService<IActivitySerializer>();
_activityVisitor = _services.GetRequiredService<IActivityVisitor>();
}
@ -59,8 +59,8 @@ public class Tests
var childDefinitionV2 = (await _workflowDefinitionPublisher.GetDraftAsync(childDefinitionV1.DefinitionId, VersionOptions.Published))!;
await _workflowDefinitionPublisher.PublishAsync(childDefinitionV2);
// Update consuming workflows to point to the new version of the child workflow.
await _workflowManager.UpdateReferencesInConsumingWorkflows(childDefinitionV2);
// // Update consuming workflows to point to the new version of the child workflow.
// await _workflowDefinitionPublisher.UpdateReferencesInConsumingWorkflows(childDefinitionV2);
// Assert that the parent workflow now points to the new version of the child workflow.
parentDefinition = await _services.GetWorkflowDefinitionAsync("parent", VersionOptions.Latest);