diff --git a/src/apps/Elsa.Server.Web/Activities/AuthorizeFlow.cs b/src/apps/Elsa.Server.Web/Activities/AuthorizeFlow.cs new file mode 100644 index 000000000..1761e811f --- /dev/null +++ b/src/apps/Elsa.Server.Web/Activities/AuthorizeFlow.cs @@ -0,0 +1,49 @@ +using Elsa.Extensions; +using Elsa.Workflows; +using Elsa.Workflows.Activities.Flowchart.Attributes; +using Elsa.Workflows.Attributes; + +namespace Elsa.Server.Web.Activities; + +[Activity("Elsa", "Authorization", "Authorizes a flow based on the configured policies.")] +[FlowNode("Authorized", "Unauthorized", "Error")] +public class AuthorizeFlow : Activity +{ + protected override ValueTask ExecuteAsync(ActivityExecutionContext context) + { + var httpContext = context.GetRequiredService().HttpContext; + + if (httpContext == null) + throw new InvalidOperationException("HttpContext is not available. Ensure that the activity is executed within an HTTP request context."); + + var bookmark = context.CreateBookmark(new AuthorizeStimulus(), OnResumeAsync); + var redirectUrl = context.ExpressionExecutionContext.GenerateBookmarkTriggerUrl(bookmark.Id); + + Result.Set(context, redirectUrl); + return ValueTask.CompletedTask; + } + + private async ValueTask OnResumeAsync(ActivityExecutionContext context) + { + if (!context.TryGetWorkflowInput("Answer", out var response)) + { + await context.CompleteActivityWithOutcomesAsync("Unauthorized"); + return; + } + + switch (response) + { + case "Authorized": + await context.CompleteActivityWithOutcomesAsync("Authorized"); + return; + case "Error": + await context.CompleteActivityWithOutcomesAsync("Error"); + return; + default: + await context.CompleteActivityWithOutcomesAsync("Unauthorized"); + break; + } + } +} + +public record AuthorizeStimulus; \ No newline at end of file diff --git a/src/modules/Elsa.Http/Elsa.Http.csproj b/src/modules/Elsa.Http/Elsa.Http.csproj index 810b479a0..227314fad 100644 --- a/src/modules/Elsa.Http/Elsa.Http.csproj +++ b/src/modules/Elsa.Http/Elsa.Http.csproj @@ -1,4 +1,4 @@ - + diff --git a/src/modules/Elsa.Http/Extensions/BookmarkExecutionContextExtensions.cs b/src/modules/Elsa.Http/Extensions/BookmarkExecutionContextExtensions.cs new file mode 100644 index 000000000..c509b1f84 --- /dev/null +++ b/src/modules/Elsa.Http/Extensions/BookmarkExecutionContextExtensions.cs @@ -0,0 +1,79 @@ +using Elsa.Expressions.Models; +using Elsa.Http; +using Elsa.SasTokens.Contracts; +using Elsa.Workflows; +using Elsa.Workflows.Api; +using Elsa.Workflows.Runtime; +using Microsoft.Extensions.Options; + +// ReSharper disable once CheckNamespace +namespace Elsa.Extensions; + +/// +/// Provides extension methods for working with and generating bookmark trigger URLs. +/// +public static class BookmarkExecutionContextExtensions +{ + public static string GenerateBookmarkTriggerUrl(this ActivityExecutionContext context, string bookmarkId, TimeSpan lifetime) => context.ExpressionExecutionContext.GenerateBookmarkTriggerUrl(bookmarkId, lifetime); + public static string GenerateBookmarkTriggerUrl(this ActivityExecutionContext context, string bookmarkId, DateTimeOffset expiresAt) => context.ExpressionExecutionContext.GenerateBookmarkTriggerUrl(bookmarkId, expiresAt); + public static string GenerateBookmarkTriggerUrl(this ActivityExecutionContext context, string bookmarkId) => context.ExpressionExecutionContext.GenerateBookmarkTriggerUrl(bookmarkId); + + /// + /// Generates a URL that can be used to resume a bookmarked workflow. + /// + /// The expression execution context. + /// The ID of the bookmark to resume. + /// The lifetime of the bookmark trigger token. + /// A URL that can be used to resume a bookmarked workflow. + public static string GenerateBookmarkTriggerUrl(this ExpressionExecutionContext context, string bookmarkId, TimeSpan lifetime) + { + var token = context.GenerateBookmarkTriggerTokenInternal(bookmarkId, lifetime); + return context.GenerateBookmarkTriggerUrlInternal(token); + } + + /// + /// Generates a URL that can be used to resume a bookmarked workflow. + /// + /// The expression execution context. + /// The ID of the bookmark to resume. + /// The expiration date of the bookmark trigger token. + /// A URL that can be used to resume a bookmarked workflow. + public static string GenerateBookmarkTriggerUrl(this ExpressionExecutionContext context, string bookmarkId, DateTimeOffset expiresAt) + { + var token = context.GenerateBookmarkTriggerTokenInternal(bookmarkId, expiresAt: expiresAt); + return context.GenerateBookmarkTriggerUrlInternal(token); + } + + /// + /// Generates a URL that can be used to resume a bookmarked workflow. + /// + /// The expression execution context. + /// The ID of the bookmark to resume. + /// A URL that can be used to trigger an event. + public static string GenerateBookmarkTriggerUrl(this ExpressionExecutionContext context, string bookmarkId) + { + var token = context.GenerateBookmarkTriggerTokenInternal(bookmarkId); + return context.GenerateBookmarkTriggerUrlInternal(token); + } + + private static string GenerateBookmarkTriggerUrlInternal(this ExpressionExecutionContext context, string token) + { + var options = context.GetRequiredService>().Value; + var url = $"{options.RoutePrefix}/bookmarks/resume?t={token}"; + var absoluteUrlProvider = context.GetRequiredService(); + return absoluteUrlProvider.ToAbsoluteUrl(url).ToString(); + } + + private static string GenerateBookmarkTriggerTokenInternal(this ExpressionExecutionContext context, string bookmarkId, TimeSpan? lifetime = null, DateTimeOffset? expiresAt = null) + { + var workflowInstanceId = context.GetWorkflowExecutionContext().Id; + var payload = new BookmarkTokenPayload(bookmarkId, workflowInstanceId); + var tokenService = context.GetRequiredService(); + + return lifetime != null + ? tokenService.CreateToken(payload, lifetime.Value) + : expiresAt != null + ? tokenService.CreateToken(payload, expiresAt.Value) + : tokenService.CreateToken(payload); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Http/Extensions/ExpressionExecutionContextExtensions.cs b/src/modules/Elsa.Http/Extensions/EventExpressionExecutionContextExtensions.cs similarity index 88% rename from src/modules/Elsa.Http/Extensions/ExpressionExecutionContextExtensions.cs rename to src/modules/Elsa.Http/Extensions/EventExpressionExecutionContextExtensions.cs index a4eac875f..9f882d7bb 100644 --- a/src/modules/Elsa.Http/Extensions/ExpressionExecutionContextExtensions.cs +++ b/src/modules/Elsa.Http/Extensions/EventExpressionExecutionContextExtensions.cs @@ -1,16 +1,17 @@ using Elsa.Expressions.Models; using Elsa.Http; -using Elsa.Http.Options; using Elsa.SasTokens.Contracts; +using Elsa.Workflows.Api; +using Elsa.Workflows.Runtime; using Microsoft.Extensions.Options; // ReSharper disable once CheckNamespace namespace Elsa.Extensions; /// -/// +/// Provides extension methods for working with . /// -public static class ExpressionExecutionContextExtensions +public static class EventExpressionExecutionContextExtensions { /// /// Generates a URL that can be used to trigger an event. @@ -52,13 +53,13 @@ public static class ExpressionExecutionContextExtensions private static string GenerateEventTriggerUrlInternal(this ExpressionExecutionContext context, string token) { - var options = context.GetRequiredService>().Value; - var url = $"{options.ApiRoutePrefix}/events/trigger?t={token}"; + var options = context.GetRequiredService>().Value; + var url = $"{options.RoutePrefix}/events/trigger?t={token}"; var absoluteUrlProvider = context.GetRequiredService(); return absoluteUrlProvider.ToAbsoluteUrl(url).ToString(); } - private static string GenerateEventTriggerTokenInternal(this ExpressionExecutionContext context, string eventName, TimeSpan? lifetime = default, DateTimeOffset? expiresAt = default) + private static string GenerateEventTriggerTokenInternal(this ExpressionExecutionContext context, string eventName, TimeSpan? lifetime = null, DateTimeOffset? expiresAt = null) { var workflowInstanceId = context.GetWorkflowExecutionContext().Id; var payload = new EventTokenPayload(eventName, workflowInstanceId); diff --git a/src/modules/Elsa.Http/Options/HttpActivityOptions.cs b/src/modules/Elsa.Http/Options/HttpActivityOptions.cs index 2080c0068..1d0c75f49 100644 --- a/src/modules/Elsa.Http/Options/HttpActivityOptions.cs +++ b/src/modules/Elsa.Http/Options/HttpActivityOptions.cs @@ -15,7 +15,7 @@ public class HttpActivityOptions /// /// The base URL of the server. This should be set to the same value at which the Elsa Server is publicly available. It will be used when generating absolute URLs need to be generated by activities such as SendEmail. /// - public Uri BaseUrl { get; set; } = default!; + public Uri BaseUrl { get; set; } = null!; /// /// The prefix used for API routes. diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs index 72b87444c..2b61acc78 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledCronTask.cs @@ -20,6 +20,9 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private readonly ICronParser _cronParser; private readonly IServiceScopeFactory _scopeFactory; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . @@ -32,7 +35,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable _scopeFactory = scopeFactory; _systemClock = systemClock; _logger = logger; - _cancellationTokenSource = new CancellationTokenSource(); + _cancellationTokenSource = new(); Schedule(); } @@ -41,6 +44,13 @@ public class ScheduledCronTask : IScheduledTask, IDisposable public void Cancel() { _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + _cancellationTokenSource.Cancel(); } @@ -54,7 +64,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable var nextOccurence = _cronParser.GetNextOccurrence(_cronExpression); var delay = nextOccurence - now; - if (!adjusted && delay.Milliseconds <= 0) + if (!adjusted && delay <= TimeSpan.Zero) { adjusted = true; continue; @@ -67,7 +77,7 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private void TrySetupTimer(TimeSpan delay) { - if (delay.Milliseconds <= 0) + if (delay <= TimeSpan.Zero) return; try @@ -82,7 +92,10 @@ public class ScheduledCronTask : IScheduledTask, IDisposable private void SetupTimer(TimeSpan delay) { - _timer = new Timer(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -93,8 +106,36 @@ public class ScheduledCronTask : IScheduledTask, IDisposable var commandSender = scope.ServiceProvider.GetRequiredService(); var cancellationToken = _cancellationTokenSource.Token; - if (!cancellationToken.IsCancellationRequested) await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); - if (!cancellationToken.IsCancellationRequested) Schedule(); + + if (!cancellationToken.IsCancellationRequested) + { + try + { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + + _executing = true; + await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } + } + catch (Exception e) + { + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); + } + } + + if (!cancellationToken.IsCancellationRequested) + Schedule(); }; } @@ -102,5 +143,6 @@ public class ScheduledCronTask : IScheduledTask, IDisposable { _timer?.Dispose(); _cancellationTokenSource.Dispose(); + _executionSemaphore.Dispose(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs index 21d7fd380..89a75575d 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledRecurringTask.cs @@ -2,6 +2,7 @@ using Elsa.Common; using Elsa.Mediator.Contracts; using Elsa.Scheduling.Commands; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Timer = System.Timers.Timer; namespace Elsa.Scheduling.ScheduledTasks; @@ -14,24 +15,24 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable private readonly ITask _task; private readonly ISystemClock _systemClock; private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; private readonly TimeSpan _interval; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); private DateTimeOffset _startAt; private Timer? _timer; + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . /// - /// The task to execute. - /// The instant at which to start executing the task. - /// The interval at which to execute the task. - /// The system clock. - /// Scope factory to create the scope and get dependancies. - public ScheduledRecurringTask(ITask task, DateTimeOffset startAt, TimeSpan interval, ISystemClock systemClock, IServiceScopeFactory scopeFactory) + public ScheduledRecurringTask(ITask task, DateTimeOffset startAt, TimeSpan interval, ISystemClock systemClock, IServiceScopeFactory scopeFactory, ILogger logger) { _task = task; _systemClock = systemClock; _scopeFactory = scopeFactory; + _logger = logger; _startAt = startAt; _interval = interval; _cancellationTokenSource = new(); @@ -43,6 +44,13 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable public void Cancel() { _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + _cancellationTokenSource.Cancel(); } @@ -56,7 +64,7 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable var now = _systemClock.UtcNow; var delay = startAt - now; - if (!adjusted && delay.Milliseconds <= 0) + if (!adjusted && delay <= TimeSpan.Zero) { adjusted = true; continue; @@ -71,7 +79,10 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable { if (delay < TimeSpan.Zero) delay = TimeSpan.FromSeconds(1); - _timer = new(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -81,10 +92,35 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable using var scope = _scopeFactory.CreateScope(); var commandSender = scope.ServiceProvider.GetRequiredService(); - var cancellationToken = _cancellationTokenSource.Token; - if (!cancellationToken.IsCancellationRequested) await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); - if (!cancellationToken.IsCancellationRequested) Schedule(); + if (!cancellationToken.IsCancellationRequested) + { + try + { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + _executing = true; + await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } + } + catch (Exception e) + { + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); + } + } + + if (!cancellationToken.IsCancellationRequested) + Schedule(); }; } @@ -92,5 +128,6 @@ public class ScheduledRecurringTask : IScheduledTask, IDisposable { _cancellationTokenSource.Dispose(); _timer?.Dispose(); + _executionSemaphore.Dispose(); } -} +} \ No newline at end of file diff --git a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs index 3d2e9c61e..6979dd410 100644 --- a/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs +++ b/src/modules/Elsa.Scheduling/ScheduledTasks/ScheduledSpecificInstantTask.cs @@ -18,7 +18,10 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable private readonly ILogger _logger; private readonly DateTimeOffset _startAt; private readonly CancellationTokenSource _cancellationTokenSource; + private readonly SemaphoreSlim _executionSemaphore = new(1, 1); private Timer? _timer; + private bool _executing; + private bool _cancellationRequested; /// /// Initializes a new instance of . @@ -30,23 +33,37 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable _scopeFactory = scopeFactory; _logger = logger; _startAt = startAt; - _cancellationTokenSource = new CancellationTokenSource(); + _cancellationTokenSource = new(); Schedule(); } /// - public void Cancel() => _timer?.Dispose(); + public void Cancel() + { + _timer?.Dispose(); + + if (_executing) + { + _cancellationRequested = true; + return; + } + + _cancellationTokenSource.Cancel(); + } private void Schedule() { var now = _systemClock.UtcNow; var delay = _startAt - now; - if (delay.Milliseconds <= 0) + if (delay <= TimeSpan.Zero) delay = TimeSpan.FromMilliseconds(1); - _timer = new Timer(delay.TotalMilliseconds) { Enabled = true }; + _timer = new(delay.TotalMilliseconds) + { + Enabled = true + }; _timer.Elapsed += async (_, _) => { @@ -55,19 +72,31 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable using var scope = _scopeFactory.CreateScope(); var commandSender = scope.ServiceProvider.GetRequiredService(); - var cancellationToken = _cancellationTokenSource.Token; if (!cancellationToken.IsCancellationRequested) { try { + var acquired = await _executionSemaphore.WaitAsync(0, cancellationToken); + if (!acquired) return; + _executing = true; await commandSender.SendAsync(new RunScheduledTask(_task), cancellationToken); + + if (_cancellationRequested) + { + _cancellationRequested = false; + _cancellationTokenSource.Cancel(); + } } catch (Exception e) { - _logger.LogError(e, "Error scheduled task"); + _logger.LogError(e, "Error executing scheduled task"); + } + finally + { + _executing = false; + _executionSemaphore.Release(); } - } }; } @@ -76,5 +105,6 @@ public class ScheduledSpecificInstantTask : IScheduledTask, IDisposable { _cancellationTokenSource.Dispose(); _timer?.Dispose(); + _executionSemaphore.Dispose(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Elsa.Workflows.Api.csproj b/src/modules/Elsa.Workflows.Api/Elsa.Workflows.Api.csproj index 9fc0ccac4..d0efe7f0f 100644 --- a/src/modules/Elsa.Workflows.Api/Elsa.Workflows.Api.csproj +++ b/src/modules/Elsa.Workflows.Api/Elsa.Workflows.Api.csproj @@ -1,4 +1,4 @@ - + diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/Bookmarks/Resume/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/Bookmarks/Resume/Endpoint.cs new file mode 100644 index 000000000..95ddfabd8 --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/Bookmarks/Resume/Endpoint.cs @@ -0,0 +1,89 @@ +using Elsa.Abstractions; +using Elsa.SasTokens.Contracts; +using Elsa.Workflows.Runtime; +using FastEndpoints; +using JetBrains.Annotations; +using Microsoft.AspNetCore.Http; + +namespace Elsa.Workflows.Api.Endpoints.Bookmarks.Resume; + +/// +/// Resumes a bookmarked workflow instance with the bookmark ID specified in the provided SAS token. +/// +[PublicAPI] +internal class Resume(ITokenService tokenService, IBookmarkQueue bookmarkQueue, IPayloadSerializer serializer) : ElsaEndpoint +{ + /// + public override void Configure() + { + Routes("/bookmarks/resume"); + Verbs(Http.GET, Http.POST); + AllowAnonymous(); + } + + /// + public override async Task HandleAsync(Request request, CancellationToken cancellationToken) + { + var token = Query("t")!; + + if (!tokenService.TryDecryptToken(token, out var payload)) + AddError("Invalid token."); + + var input = HttpContext.Request.Method == HttpMethods.Post ? request.Input : GetInputFromQueryString(); + + if (ValidationFailed) + { + await SendErrorsAsync(cancellation: cancellationToken); + return; + } + + await ResumeBookmarkedWorkflowAsync(payload, input, cancellationToken); + + if (!HttpContext.Response.HasStarted) + await SendOkAsync(cancellationToken); + } + + private IDictionary? GetInputFromQueryString() + { + var inputJson = Query("in", false); + if (string.IsNullOrWhiteSpace(inputJson)) + return null; + + try + { + return serializer.Deserialize>(inputJson); + } + catch + { + AddError("Invalid input format. Expected a valid JSON string."); + return null; + } + } + + private async Task ResumeBookmarkedWorkflowAsync(BookmarkTokenPayload tokenPayload, IDictionary? input, CancellationToken cancellationToken) + { + var bookmarkId = tokenPayload.BookmarkId; + var workflowInstanceId = tokenPayload.WorkflowInstanceId; + var item = new NewBookmarkQueueItem + { + BookmarkId = bookmarkId, + WorkflowInstanceId = workflowInstanceId, + Options = new() + { + Input = input + } + }; + await bookmarkQueue.EnqueueAsync(item, cancellationToken); + } +} + +/// +/// The request model for the Resume endpoint. +/// +internal class Request +{ + /// + /// The input to provide to the workflow when resuming. + /// + public IDictionary? Input { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/Events/TriggerPublic/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/Events/TriggerPublic/Endpoint.cs index 6685e7c97..0f3e50a40 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/Events/TriggerPublic/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/Events/TriggerPublic/Endpoint.cs @@ -1,5 +1,4 @@ using Elsa.Abstractions; -using Elsa.Http; using Elsa.SasTokens.Contracts; using Elsa.Workflows.Runtime; using JetBrains.Annotations; diff --git a/src/modules/Elsa.Workflows.Api/Features/WorkflowsApiFeature.cs b/src/modules/Elsa.Workflows.Api/Features/WorkflowsApiFeature.cs index 6fb8a3003..721a36827 100644 --- a/src/modules/Elsa.Workflows.Api/Features/WorkflowsApiFeature.cs +++ b/src/modules/Elsa.Workflows.Api/Features/WorkflowsApiFeature.cs @@ -2,7 +2,6 @@ using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; -using Elsa.Http.Features; using Elsa.SasTokens.Features; using Elsa.Workflows.Api.Constants; using Elsa.Workflows.Api.Requirements; @@ -21,15 +20,9 @@ namespace Elsa.Workflows.Api.Features; [DependsOn(typeof(WorkflowInstancesFeature))] [DependsOn(typeof(WorkflowManagementFeature))] [DependsOn(typeof(WorkflowRuntimeFeature))] -[DependsOn(typeof(HttpFeature))] [DependsOn(typeof(SasTokensFeature))] -public class WorkflowsApiFeature : FeatureBase +public class WorkflowsApiFeature(IModule module) : FeatureBase(module) { - /// - public WorkflowsApiFeature(IModule module) : base(module) - { - } - /// public override void Configure() { @@ -43,7 +36,6 @@ public class WorkflowsApiFeature : FeatureBase Module.AddFastEndpointsFromModule(); Services.AddScoped(); - Services.AddScoped(); Services.Configure(options => { diff --git a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Outcomes.cs b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Outcomes.cs index 9cfa3cfde..c0154ab4b 100644 --- a/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Outcomes.cs +++ b/src/modules/Elsa.Workflows.Core/Activities/Flowchart/Models/Outcomes.cs @@ -4,8 +4,8 @@ namespace Elsa.Workflows.Activities.Flowchart.Models; /// Represents a list of outcomes that can be send when completing an activity. This information is used by . /// /// A list of outcome names. -public record Outcomes(params string[] Names) -{ - public static readonly Outcomes Default = new([null!, "Done"]); - public static readonly Outcomes Empty = new(); -} +public record Outcomes(params string[] Names) +{ + public static readonly Outcomes Default = new(null!, "Done"); + public static readonly Outcomes Empty = new(); +} diff --git a/src/modules/Elsa.Workflows.Core/Services/DefaultWorkflowInstanceVariableReader.cs b/src/modules/Elsa.Workflows.Core/Services/DefaultWorkflowInstanceVariableReader.cs index c3557bf10..3eca2c3a5 100644 --- a/src/modules/Elsa.Workflows.Core/Services/DefaultWorkflowInstanceVariableReader.cs +++ b/src/modules/Elsa.Workflows.Core/Services/DefaultWorkflowInstanceVariableReader.cs @@ -2,11 +2,11 @@ namespace Elsa.Workflows; public class DefaultWorkflowInstanceVariableReader(IVariablePersistenceManager variablePersistenceManager) : IWorkflowInstanceVariableReader { - public async Task> GetVariables(WorkflowExecutionContext workflowExecutionContext, IEnumerable? excludeTags = default, CancellationToken cancellationToken = default) + public async Task> GetVariables(WorkflowExecutionContext workflowExecutionContext, IEnumerable? excludeTags = null, CancellationToken cancellationToken = default) { var workflow = workflowExecutionContext.Workflow; var workflowVariables = workflow.Variables; - var rootWorkflowActivityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.ParentActivityExecutionContext == null); + var rootWorkflowActivityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.FirstOrDefault(x => x.Activity == workflow); if (rootWorkflowActivityExecutionContext == null) return []; @@ -17,7 +17,7 @@ public class DefaultWorkflowInstanceVariableReader(IVariablePersistenceManager v foreach (var workflowVariable in workflowVariables) { var value = workflowVariable.Get(rootWorkflowActivityExecutionContext.ExpressionExecutionContext); - resolvedVariables.Add(new ResolvedVariable(workflowVariable, value)); + resolvedVariables.Add(new(workflowVariable, value)); } return resolvedVariables; diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs index 433c66e00..df63cb374 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowStateExtractor.cs @@ -2,11 +2,12 @@ using Elsa.Extensions; using Elsa.Workflows.Models; using Elsa.Workflows.Services; using Elsa.Workflows.State; +using Microsoft.Extensions.Logging; namespace Elsa.Workflows; /// -public class WorkflowStateExtractor : IWorkflowStateExtractor +public class WorkflowStateExtractor(ILogger logger) : IWorkflowStateExtractor { /// public WorkflowState Extract(WorkflowExecutionContext workflowExecutionContext) @@ -101,7 +102,7 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor workflowExecutionContext.Properties[property.Key] = property.Value; } - private static async Task ApplyActivityExecutionContextsAsync(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) + private async Task ApplyActivityExecutionContextsAsync(WorkflowState state, WorkflowExecutionContext workflowExecutionContext) { var activityExecutionContexts = (await Task.WhenAll(state.ActivityExecutionContexts.Select(async item => await CreateActivityExecutionContextAsync(item)))).Where(x => x != null).Select(x => x!).ToList(); @@ -110,10 +111,14 @@ public class WorkflowStateExtractor : IWorkflowStateExtractor // Reconstruct hierarchy. foreach (var contextState in state.ActivityExecutionContexts.Where(x => !string.IsNullOrWhiteSpace(x.ParentContextId))) { - if (lookup.ContainsKey(contextState.ParentContextId)) + var parentContextId = contextState.ParentContextId; + if (parentContextId == null || !lookup.TryGetValue(parentContextId, out var parentContext)) { - var parentContext = lookup[contextState.ParentContextId!]; - var contextId = contextState.Id; + logger.LogWarning("Parent context with ID '{ParentContextId}' not found for context with ID '{ContextId}'.", parentContextId, contextState.Id); + continue; // Skip if parent context is not found. + } + + var contextId = contextState.Id; if (lookup.TryGetValue(contextId, out var context)) { diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionContextRecordExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionContextRecordExtensions.cs index c2706cc1e..b9a49106c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionContextRecordExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionContextRecordExtensions.cs @@ -16,17 +16,26 @@ public static class ActivityExecutionContextRecordExtensions context.TransientProperties[ActivityExecutionRecordKey] = record; } - public static ActivityExecutionRecord? GetCapturedActivityExecutionRecord(this ActivityExecutionContext context) - { - return context.TransientProperties.TryGetValue(ActivityExecutionRecordKey, out var record) ? (ActivityExecutionRecord?)record : null; - } - public static async Task GetOrMapCapturedActivityExecutionRecordAsync(this ActivityExecutionContext context) { - if(context.TransientProperties.TryGetValue(ActivityExecutionRecordKey, out var record)) - return (ActivityExecutionRecord)record; - var mapper = context.GetRequiredService(); - return await mapper.MapAsync(context); + var record = await mapper.MapAsync(context); + + if (context.TransientProperties.TryGetValue(ActivityExecutionRecordKey, out var capturedRecord)) + { + var serializedSnapshot = ((ActivityExecutionRecord)capturedRecord).SerializedSnapshot!; + + // Take the existing serialized snapshot. + record.SerializedSnapshot = serializedSnapshot; + + // Update the serialized snapshot with the current record's properties. + // This will reflect the latest state of the activity execution context without losing the existing serialized snapshot representing e.g., variable values at the time of the record capture. + serializedSnapshot.HasBookmarks = record.HasBookmarks; + serializedSnapshot.Status = record.Status; + serializedSnapshot.AggregateFaultCount = record.AggregateFaultCount; + serializedSnapshot.CompletedAt = record.CompletedAt; + } + + return record; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/BookmarkTokenPayload.cs b/src/modules/Elsa.Workflows.Runtime/Models/BookmarkTokenPayload.cs new file mode 100644 index 000000000..038959449 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Models/BookmarkTokenPayload.cs @@ -0,0 +1,6 @@ +namespace Elsa.Workflows.Runtime; + +/// +/// Represents the payload for a bookmark token, including the bookmark identifier and the associated workflow instance identifier. +/// +public record BookmarkTokenPayload(string BookmarkId, string WorkflowInstanceId); \ No newline at end of file diff --git a/src/modules/Elsa.Http/Models/EventTokenPayload.cs b/src/modules/Elsa.Workflows.Runtime/Models/EventTokenPayload.cs similarity index 90% rename from src/modules/Elsa.Http/Models/EventTokenPayload.cs rename to src/modules/Elsa.Workflows.Runtime/Models/EventTokenPayload.cs index 22d5fd214..e6981adde 100644 --- a/src/modules/Elsa.Http/Models/EventTokenPayload.cs +++ b/src/modules/Elsa.Workflows.Runtime/Models/EventTokenPayload.cs @@ -1,4 +1,4 @@ -namespace Elsa.Http; +namespace Elsa.Workflows.Runtime; /// /// Represents the payload of an event, serialized as a secured token. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundStimulusDispatcher.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundStimulusDispatcher.cs index 58fad4747..614f63a8d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundStimulusDispatcher.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundStimulusDispatcher.cs @@ -1,5 +1,7 @@ +using Elsa.Common.Multitenancy; using Elsa.Mediator; using Elsa.Mediator.Contracts; +using Elsa.Tenants.Mediator; using Elsa.Workflows.Runtime.Commands; using Elsa.Workflows.Runtime.Requests; using Elsa.Workflows.Runtime.Responses; @@ -9,13 +11,18 @@ namespace Elsa.Workflows.Runtime; /// /// A simple implementation that queues the specified request for delivering stimuli on a non-durable background worker. /// -public class BackgroundStimulusDispatcher(ICommandSender commandSender) : IStimulusDispatcher +public class BackgroundStimulusDispatcher(ICommandSender commandSender, ITenantAccessor tenantAccessor) : IStimulusDispatcher { /// public async Task SendAsync(DispatchStimulusRequest request, CancellationToken cancellationToken = default) { var command = new DispatchStimulusCommand(request); - await commandSender.SendAsync(command, CommandStrategy.Background, cancellationToken); + await commandSender.SendAsync(command, CommandStrategy.Background, CreateHeaders(), cancellationToken); return DispatchStimulusResponse.Empty; } -} \ No newline at end of file + + private IDictionary CreateHeaders() + { + return TenantHeaders.CreateHeaders(tenantAccessor.Tenant?.Id); + } +} diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs index 23e4b0d34..411d85dc6 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs @@ -143,6 +143,7 @@ public class LocalWorkflowClient( SubStatus = workflowState.SubStatus, Incidents = workflowState.Incidents, Output = request.IncludeWorkflowOutput ? new Dictionary(workflowState.Output) : null + Bookmarks = workflowState.Bookmarks }; }