elsa-core/src/modules/Elsa.Workflows.Runtime/Services/DefaultBackgroundActivityInvoker.cs

using Elsa.Common.Models;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Management.Contracts;
using Elsa.Workflows.Memory;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime.Contracts;
using Elsa.Workflows.Runtime.Middleware.Activities;
using Elsa.Workflows.Runtime.Models;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Services;
using Microsoft.Extensions.Logging;

namespace Elsa.Workflows.Runtime.Services;

/// <summary>
/// An activity invoker that invokes activities detached from the workflow. This is useful for invoking activities from a background worker.
/// </summary>
public class DefaultBackgroundActivityInvoker : IBackgroundActivityInvoker
{
    private readonly IWorkflowRuntime _workflowRuntime;
    private readonly IWorkflowDispatcher _workflowDispatcher;
    private readonly IWorkflowDefinitionService _workflowDefinitionService;
    private readonly IVariablePersistenceManager _variablePersistenceManager;
    private readonly IActivityInvoker _activityInvoker;
    private readonly IBookmarksPersister _bookmarksPersister;
    private readonly IWorkflowStateExtractor _workflowStateExtractor;
    private readonly IServiceProvider _serviceProvider;
    private readonly ILogger _logger;

    /// <summary>
    /// Initializes a new instance of the <see cref="DefaultBackgroundActivityInvoker"/> class.
    /// </summary>
    public DefaultBackgroundActivityInvoker(
        IWorkflowRuntime workflowRuntime,
        IWorkflowDispatcher workflowDispatcher,
        IWorkflowDefinitionService workflowDefinitionService,
        IVariablePersistenceManager variablePersistenceManager,
        IActivityInvoker activityInvoker,
        IBookmarksPersister bookmarksPersister,
        IWorkflowStateExtractor workflowStateExtractor,
        IServiceProvider serviceProvider,
        ILogger<DefaultBackgroundActivityInvoker> logger)
    {
        _workflowRuntime = workflowRuntime;
        _workflowDispatcher = workflowDispatcher;
        _workflowDefinitionService = workflowDefinitionService;
        _variablePersistenceManager = variablePersistenceManager;
        _activityInvoker = activityInvoker;
        _bookmarksPersister = bookmarksPersister;
        _workflowStateExtractor = workflowStateExtractor;
        _serviceProvider = serviceProvider;
        _logger = logger;
    }

    /// <inheritdoc />
    public async Task ExecuteAsync(ScheduledBackgroundActivity scheduledBackgroundActivity, CancellationToken cancellationToken = default)
    {
        var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId;
        
        var workflowState = await _workflowRuntime.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken);

        if (workflowState == null)
            throw new Exception("Workflow state not found");

        var workflowDefinition = await _workflowDefinitionService.FindAsync(workflowState.DefinitionId, VersionOptions.SpecificVersion(workflowState.DefinitionVersion), cancellationToken);

        if (workflowDefinition == null)
            throw new Exception("Workflow definition not found");

        var workflow = await _workflowDefinitionService.MaterializeWorkflowAsync(workflowDefinition, cancellationToken);
        var workflowExecutionContext = await WorkflowExecutionContext.CreateAsync(_serviceProvider, workflow, workflowState, cancellationTokens: cancellationToken);
        var activityNodeId = scheduledBackgroundActivity.ActivityNodeId;
        var activityExecutionContext = workflowExecutionContext.ActivityExecutionContexts.First(x => x.NodeId == activityNodeId);

        // Load persistent variables for the activity to use.
        await _variablePersistenceManager.LoadVariablesAsync(workflowExecutionContext);

        // Mark the activity as being invoked from a background worker.
        activityExecutionContext.SetIsBackgroundExecution();

        // Invoke the activity.
        await _activityInvoker.InvokeAsync(activityExecutionContext);

        // Capture any activity output produced by the activity (but only if the associated memory block is stored in the workflow itself).
        var outputDescriptors = activityExecutionContext.ActivityDescriptor.Outputs;
        var outputValues = new Dictionary<string, object>();
        var outcomes = activityExecutionContext.GetBackgroundOutcomes().ToList();

        foreach (var outputDescriptor in outputDescriptors)
        {
            var output = (Output?)outputDescriptor.ValueGetter(activityExecutionContext.Activity);

            if (output == null)
                continue;

            var memoryBlockReference = output.MemoryBlockReference();

            if (!activityExecutionContext.ExpressionExecutionContext.TryGetBlock(memoryBlockReference, out var memoryBlock))
                continue;

            var variableMetadata = memoryBlock.Metadata as VariableBlockMetadata;
            var driver = variableMetadata?.StorageDriverType;

            // We only capture output written to the workflow itself. Other drivers like blob storage, etc. will be ignored since the foreground context will be loading those.
            if (driver != typeof(WorkflowStorageDriver))
                continue;

            var outputValue = activityExecutionContext.Get(memoryBlockReference);

            if (outputValue != null)
                outputValues[outputDescriptor.Name] = outputValue;
        }

        // Resume the workflow, passing along the activity output.
        var bookmarkId = scheduledBackgroundActivity.BookmarkId;
        var inputKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutputKey(activityNodeId);
        var outcomesKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityOutcomesKey(activityNodeId);
        var journalDataKey = BackgroundActivityInvokerMiddleware.GetBackgroundActivityJournalDataKey(activityNodeId);

        var dispatchRequest = new DispatchWorkflowInstanceRequest
        {
            InstanceId = workflowInstanceId, 
            BookmarkId = bookmarkId, 
            Properties = new Dictionary<string, object>
            {
                [outcomesKey] = outcomes,
                [inputKey] = outputValues,
                [journalDataKey] = activityExecutionContext.JournalData
            }
        };

        if (cancellationToken.IsCancellationRequested)
        {
            _logger.LogInformation("Background execution for activity {ActivityNodeId} was canceled", activityNodeId);
            return;
        }

        await _workflowDispatcher.DispatchAsync(dispatchRequest, cancellationToken);
    }
}