diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs index 00b332f3d..f47aa314d 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputEvaluation.cs @@ -101,7 +101,7 @@ public static partial class ActivityExecutionContextExtensions // When input is created from an activity provider, there may be no memory block reference. if (memoryReference?.Id != null!) { - // Declare the input memory block on the current context. + // Declare the input memory block in the current context. context.ExpressionExecutionContext.Set(memoryReference, value!); } } @@ -110,12 +110,12 @@ public static partial class ActivityExecutionContextExtensions value = input; } - await StoreInputValueAsync(context, inputDescriptor, value); + await StoreInputValueAsync(context, inputDescriptor, value!); return value; } - private static async Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object? value) + private static Task StoreInputValueAsync(ActivityExecutionContext context, InputDescriptor inputDescriptor, object value) { // Store the serialized input value in the activity state. // Serializing the value ensures we store a copy of the value and not a reference to the input, which may change over time. @@ -128,5 +128,7 @@ public static partial class ActivityExecutionContextExtensions // var filterResult = await manager.RunFiltersAsync(filterContext); context.ActivityState[inputDescriptor.Name] = value; } + + return Task.CompletedTask; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs new file mode 100644 index 000000000..00c6c626e --- /dev/null +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExecutionContextExtensions.InputOutput.cs @@ -0,0 +1,38 @@ +using Elsa.Extensions; +using Elsa.Workflows.Models; + +namespace Elsa.Workflows; + +public static class ActivityExecutionContextExtensions +{ + public static IDictionary GetInputs(this ActivityExecutionContext context) + { + return context.ActivityState!; + } + + public static IDictionary GetOutputs(this ActivityExecutionContext context) + { + var activity = context.Activity; + var expressionExecutionContext = context.ExpressionExecutionContext; + var activityDescriptor = context.ActivityDescriptor; + var outputDescriptors = activityDescriptor.Outputs; + + var outputs = outputDescriptors.ToDictionary(x => x.Name, x => + { + if (x.IsSerializable == false) + return "(not serializable)"; + + var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name); + + if (cachedValue != null) + return cachedValue; + + if (x.ValueGetter(activity) is Output output && context.TryGet(output.MemoryBlockReference(), out var outputValue)) + return outputValue!; + + return null!; + }); + + return outputs; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs index a99cd50b3..b6f033699 100644 --- a/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs +++ b/src/modules/Elsa.Workflows.Core/Extensions/ActivityExtensions.cs @@ -64,7 +64,7 @@ public static class ActivityExtensions /// The activity execution context. /// Name of the output. /// The output value. - public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = default) + public static object? GetOutput(this IActivity activity, ActivityExecutionContext context, string? outputName = null) { var workflowExecutionContext = context.WorkflowExecutionContext; var outputRegister = workflowExecutionContext.GetActivityOutputRegister(); @@ -84,7 +84,7 @@ public static class ActivityExtensions /// The expression execution context. /// Name of the output. /// The output value. - public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = default) + public static object? GetOutput(this IActivity activity, ExpressionExecutionContext context, string? outputName = null) { var activityExecutionContext = context.GetActivityExecutionContext(); diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs index b73e3758e..05b943522 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs @@ -10,12 +10,11 @@ public interface IActivityExecutionMapper /// /// Maps an activity execution context to an activity execution record. /// - Task MapAsync(ActivityExecutionContext source); - + ActivityExecutionRecord Map(ActivityExecutionContext source); + /// - /// Retrieves a dictionary containing the persistable output of an activity execution context. + /// Maps an activity execution context to an activity execution record. /// - /// The activity execution context to extract persistable output from. - /// A dictionary containing the persistable output. - Task> GetPersistableOutputAsync(ActivityExecutionContext context); + [Obsolete( "Use Map instead.", error: false)] + Task MapAsync(ActivityExecutionContext source); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings b/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings index 41b655e3e..fb25136b5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings +++ b/src/modules/Elsa.Workflows.Runtime/Elsa.Workflows.Runtime.csproj.DotSettings @@ -4,6 +4,12 @@ True True True + True + True + True + True + True + True True True True diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs index 86f648e4a..642e449eb 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/PipelineWorkflowsFeatureExtensions.cs @@ -33,6 +33,7 @@ public static class PipelineWorkflowsFeatureExtensions .UseExceptionHandling() .UseExecutionLogging() .UseNotifications() + .UseLogPersistenceModeEvaluation() .UseBackgroundActivityInvoker(); configurePipeline?.Invoke(pipeline); diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs index 7031ca4a1..233e35343 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowsFeatureExtensions.cs @@ -28,6 +28,7 @@ public static class WorkflowsFeatureExtensions workflowsFeature.WithActivityExecutionPipeline(pipeline => pipeline .UseExceptionHandling() + .UseLogPersistenceModeEvaluation() .UseExecutionLogging() .UseNotifications() .UseBackgroundActivityInvoker()); diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index b9a15ffa4..caa7aba26 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -284,6 +284,7 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module) .AddScoped() .AddScoped() .AddScoped, WorkflowExecutionLogRecordExtractor>() + .AddScoped() .AddScoped() .AddScoped() .AddScoped() diff --git a/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs b/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs deleted file mode 100644 index 4770bcbd7..000000000 --- a/src/modules/Elsa.Workflows.Runtime/Instructions/TriggerWorkflowInstruction.cs +++ /dev/null @@ -1 +0,0 @@ -namespace Elsa.Workflows.Runtime.Instructions; diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs new file mode 100644 index 000000000..c587d107f --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Contracts/IActivityPropertyLogPersistenceEvaluator.cs @@ -0,0 +1,19 @@ +namespace Elsa.Workflows.Runtime; + +/// +/// Provides functionality for evaluating log persistence settings for activity properties +/// during the execution of a workflow. +/// +public interface IActivityPropertyLogPersistenceEvaluator +{ + /// + /// Evaluates the log persistence settings for activity properties within the context of a workflow's execution. + /// + Task EvaluateLogPersistenceModesAsync(ActivityExecutionContext context); + + /// + /// Retrieves a dictionary of persistable output values generated during the execution of an activity. + /// + /// A dictionary where the keys represent output property names and the values represent their persistable data. + Task> GetPersistableOutputAsync(ActivityExecutionContext context); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs new file mode 100644 index 000000000..1d6ce8577 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionContextExtensions.cs @@ -0,0 +1,18 @@ +using Elsa.Extensions; + +namespace Elsa.Workflows.Runtime; + +public static class ActivityExecutionContextExtensions +{ + private static object LogPersistenceMapKey { get; } = new(); + + public static ActivityLogPersistenceModeMap GetLogPersistenceModeMap(this ActivityExecutionContext context) + { + return context.TransientProperties.GetValueOrDefault(LogPersistenceMapKey, () => new ActivityLogPersistenceModeMap())!; + } + + public static void SetLogPersistenceModeMap(this ActivityExecutionContext context, ActivityLogPersistenceModeMap map) + { + context.TransientProperties[LogPersistenceMapKey] = map; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs similarity index 61% rename from src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs index 30408d920..988b6d87b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Extensions/ActivityExecutionPipelineBuilderExtensions.cs +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Extensions/ActivityExecutionPipelineBuilderExtensions.cs @@ -1,5 +1,6 @@ using Elsa.Workflows; using Elsa.Workflows.Pipelines.ActivityExecution; +using Elsa.Workflows.Runtime.Middleware; using Elsa.Workflows.Runtime.Middleware.Activities; // ReSharper disable once CheckNamespace @@ -14,4 +15,9 @@ public static class ActivityExecutionPipelineBuilderExtensions /// Installs the . /// public static IActivityExecutionPipelineBuilder UseBackgroundActivityInvoker(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); + + /// + /// Installs the which evaluates log persistence modes during activity execution. + /// + public static IActivityExecutionPipelineBuilder UseLogPersistenceModeEvaluation(this IActivityExecutionPipelineBuilder pipelineBuilder) => pipelineBuilder.UseMiddleware(); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs new file mode 100644 index 000000000..838f22e7d --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Middleware/EvaluateLogPersistenceModesMiddleware.cs @@ -0,0 +1,13 @@ +using Elsa.Workflows.Pipelines.ActivityExecution; + +namespace Elsa.Workflows.Runtime.Middleware; + +public class EvaluateLogPersistenceModesMiddleware(ActivityMiddlewareDelegate next, IActivityPropertyLogPersistenceEvaluator persistenceEvaluator) : IActivityExecutionMiddleware +{ + public async ValueTask InvokeAsync(ActivityExecutionContext context) + { + await next(context); + var persistenceLogMap = await persistenceEvaluator.EvaluateLogPersistenceModesAsync(context); + context.SetLogPersistenceModeMap(persistenceLogMap); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs new file mode 100644 index 000000000..a8472268c --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/ActivityLogPersistenceModeMap.cs @@ -0,0 +1,9 @@ +using Elsa.Workflows.LogPersistence; + +namespace Elsa.Workflows.Runtime; + +public class ActivityLogPersistenceModeMap +{ + public IDictionary Inputs { get; set; } = new Dictionary(); + public IDictionary Outputs { get; set; } = new Dictionary(); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Models/LogPersistenceConfiguration.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceConfiguration.cs similarity index 100% rename from src/modules/Elsa.Workflows.Runtime/Models/LogPersistenceConfiguration.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceConfiguration.cs diff --git a/src/modules/Elsa.Workflows.Runtime/Enums/LogPersistenceEvaluationMode.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceEvaluationMode.cs similarity index 100% rename from src/modules/Elsa.Workflows.Runtime/Enums/LogPersistenceEvaluationMode.cs rename to src/modules/Elsa.Workflows.Runtime/LogPersistence/Models/LogPersistenceEvaluationMode.cs diff --git a/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs new file mode 100644 index 000000000..1f5e2c2e9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/LogPersistence/Services/ActivityPropertyLogPersistenceEvaluator.cs @@ -0,0 +1,230 @@ +using System.Text.Json; +using System.Text.Json.Serialization; +using Elsa.Expressions.Contracts; +using Elsa.Expressions.Models; +using Elsa.Extensions; +using Elsa.Workflows.Activities; +using Elsa.Workflows.LogPersistence; +using Elsa.Workflows.LogPersistence.Strategies; +using Elsa.Workflows.Management.Options; +using Elsa.Workflows.Models; +using Elsa.Workflows.Serialization.Converters; +using Humanizer; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Elsa.Workflows.Runtime; + +/* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity: + * { + * "logPersistenceMode": { + * "default": "default", + * "inputs": { k : v }, + * "outputs": { k: v } + * } + * } + */ + +/* The following JSON structure is expected to be found in the custom properties of the workflow and activity: + * { + * "logPersistenceConfig": { + * "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." }, + * "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }, + * "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } } + * } + * } + */ +public class ActivityPropertyLogPersistenceEvaluator : IActivityPropertyLogPersistenceEvaluator +{ + private readonly IExpressionEvaluator _expressionEvaluator; + private readonly JsonSerializerOptions _jsonOptions; + private readonly IDictionary _strategies; + private readonly IOptions _options; + private readonly ILogger _logger; + + private const string LegacyKey = "logPersistenceMode"; + const string ConfigKey = "logPersistenceConfig"; + + public ActivityPropertyLogPersistenceEvaluator( + ILogPersistenceStrategyService strategyService, + IExpressionDescriptorRegistry expressionDescriptorRegistry, + IExpressionEvaluator expressionEvaluator, + IOptions options, + ILogger logger) + { + _expressionEvaluator = expressionEvaluator; + _options = options; + _logger = logger; + _strategies = strategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x); + _jsonOptions = new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true + }.WithConverters( + new ExpressionJsonConverterFactory(expressionDescriptorRegistry), + new JsonStringEnumConverter(), + new ExpandoObjectConverterFactory()); + } + + public async Task EvaluateLogPersistenceModesAsync(ActivityExecutionContext context) + { + var cancellationToken = context.CancellationToken; + var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken); + var map = new ActivityLogPersistenceModeMap(); + + await EvaluatePropertiesAsync(context, "inputs", context.ActivityDescriptor.Inputs, legacyProps, configProps, defaultMode, map.Inputs, cancellationToken); + await EvaluatePropertiesAsync(context, "outputs", context.ActivityDescriptor.Outputs, legacyProps, configProps, defaultMode, map.Outputs, cancellationToken); + + return map; + } + + public async Task> GetPersistableOutputAsync(ActivityExecutionContext context) + { + var cancellationToken = context.WorkflowExecutionContext.CancellationToken; + var (legacyProps, configProps, defaultMode) = await GetPersistenceDefaultsAsync(context, cancellationToken); + var outputs = context.GetOutputs(); + return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyProps, configProps, defaultMode, cancellationToken); + } + + private async Task<(IDictionary legacyProps, IDictionary configProps, LogPersistenceMode defaultMode)> + GetPersistenceDefaultsAsync(ActivityExecutionContext context, CancellationToken cancellationToken) + { + var legacyProps = context.Activity.CustomProperties.GetValueOrDefault>(LegacyKey, () => new Dictionary())!; + var rootContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); + var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow; + var workflowDefault = await GetDefaultPersistenceModeAsync(rootContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); + var activityDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowDefault, cancellationToken); + var configProps = context.Activity.CustomProperties.GetValueOrDefault>(ConfigKey, () => new Dictionary())!; + return (legacyProps, configProps, activityDefault); + } + + private async Task EvaluatePropertiesAsync( + ActivityExecutionContext context, + string key, + IEnumerable descriptors, + IDictionary legacyConfig, + IDictionary currentConfig, + LogPersistenceMode defaultMode, + IDictionary resultMap, + CancellationToken cancellationToken) + { + var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary())!; + var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary())!; + + foreach (var descriptor in descriptors) + { + resultMap[descriptor.Name] = await EvaluatePropertyModeAsync(context.ExpressionExecutionContext, descriptor, legacySection, currentSection, defaultMode, cancellationToken); + } + } + + private async Task EvaluatePropertyModeAsync( + ExpressionExecutionContext executionContext, + PropertyDescriptor descriptor, + IDictionary legacySection, + IDictionary currentSection, + LogPersistenceMode defaultMode, + CancellationToken cancellationToken) + { + var key = descriptor.Name.Camelize(); + var configObject = currentSection.GetValueOrDefault(key, () => null); + var config = ConvertToConfig(configObject); + if (config != null) + return await EvaluateConfigAsync(config, executionContext, () => defaultMode, cancellationToken); + + var mode = legacySection.GetValueOrDefault(key, () => defaultMode); + return ResolveMode(mode, () => defaultMode); + } + + private async Task> GetPersistablePropertiesAsync( + ActivityExecutionContext context, + IDictionary state, + string key, + IDictionary legacyConfig, + IDictionary currentConfig, + LogPersistenceMode defaultMode, + CancellationToken cancellationToken) + { + var result = new Dictionary(); + var legacySection = legacyConfig.GetValueOrDefault(key, () => new Dictionary()); + var currentSection = currentConfig.GetValueOrDefault(key, () => new Dictionary()); + + foreach (var item in state) + { + var propKey = item.Key.Camelize(); + var configObject = currentSection!.GetValueOrDefault(propKey, () => null); + var config = ConvertToConfig(configObject); + var mode = config != null + ? await EvaluateConfigAsync(config, context.ExpressionExecutionContext, () => defaultMode, cancellationToken) + : legacySection!.GetValueOrDefault(propKey, () => defaultMode); + + if (mode == LogPersistenceMode.Include || (mode == LogPersistenceMode.Inherit && (defaultMode == LogPersistenceMode.Include || defaultMode == LogPersistenceMode.Inherit))) + result.Add(item.Key, item.Value); + } + + return result; + } + + private async Task GetDefaultPersistenceModeAsync( + ExpressionExecutionContext executionContext, + IDictionary properties, + Func defaultFactory, + CancellationToken cancellationToken) + { + var legacyProps = properties.GetValueOrDefault>(LegacyKey, () => new Dictionary()); + var configProps = properties.GetValueOrDefault>(ConfigKey, () => new Dictionary()); + var defaultObj = configProps!.TryGetValue("default", out var val) ? val : null; + if (defaultObj == null) + { + var legacyDefault = legacyProps!.GetValueOrDefault("default", defaultFactory); + return legacyDefault == LogPersistenceMode.Inherit ? defaultFactory() : legacyDefault; + } + + var config = ConvertToConfig(defaultObj); + return await EvaluateConfigAsync(config, executionContext, defaultFactory, cancellationToken); + } + + private async Task EvaluateConfigAsync( + LogPersistenceConfiguration? config, + ExpressionExecutionContext executionContext, + Func defaultFactory, + CancellationToken cancellationToken) + { + if (config?.EvaluationMode == LogPersistenceEvaluationMode.Strategy) + { + var strategyType = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName(); + if (!_strategies.TryGetValue(strategyType, out var strategy)) + return defaultFactory(); + + var strategyContext = new LogPersistenceStrategyContext(cancellationToken); + var mode = await strategy.GetPersistenceModeAsync(strategyContext); + return ResolveMode(mode, defaultFactory); + } + + if (config?.Expression == null) + return defaultFactory(); + + try + { + var mode = await _expressionEvaluator.EvaluateAsync(config.Expression, executionContext); + return ResolveMode(mode, defaultFactory); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Error evaluating log persistence expression"); + return defaultFactory(); + } + } + + private LogPersistenceMode ResolveMode(LogPersistenceMode mode, Func defaultFactory) + { + var m = mode == LogPersistenceMode.Inherit ? defaultFactory() : mode; + return m == LogPersistenceMode.Inherit ? LogPersistenceMode.Include : m; + } + + private LogPersistenceConfiguration? ConvertToConfig(object? value) + { + if (value == null) return null; + if (value is LogPersistenceConfiguration config) return config; + var json = JsonSerializer.Serialize(value); + return JsonSerializer.Deserialize(json, _jsonOptions); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs index b659c8635..ceee49fd3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistActivityExecutionLogMiddleware.cs @@ -12,8 +12,5 @@ public class PersistActivityExecutionLogMiddleware(WorkflowMiddlewareDelegate ne public override async ValueTask InvokeAsync(WorkflowExecutionContext context) { await Next(context); - - // Not used anymore. - //await sink.PersistExecutionLogsAsync(context); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs index 4c2d39534..d7f03fa20 100644 --- a/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs +++ b/src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs @@ -13,8 +13,5 @@ public class PersistWorkflowExecutionLogMiddleware(WorkflowMiddlewareDelegate ne { // Invoke next middleware. await Next(context); - - // Not used anymore. - //await sink.PersistExecutionLogsAsync(context); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs index 7c1a71048..13ff7f8a0 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BackgroundActivityInvoker.cs @@ -15,7 +15,7 @@ public class BackgroundActivityInvoker( IWorkflowDefinitionService workflowDefinitionService, IVariablePersistenceManager variablePersistenceManager, IActivityInvoker activityInvoker, - IActivityExecutionMapper activityExecutionMapper, + IActivityPropertyLogPersistenceEvaluator activityPropertyLogPersistenceEvaluator, WorkflowHeartbeatGeneratorFactory workflowHeartbeatGeneratorFactory, IServiceProvider serviceProvider, ILogger logger) @@ -66,7 +66,7 @@ public class BackgroundActivityInvoker( var completed = activityExecutionContext.GetBackgroundCompleted(); var scheduledActivities = activityExecutionContext.GetBackgroundScheduledActivities().ToList(); var workflowInstanceId = scheduledBackgroundActivity.WorkflowInstanceId; - var outputValues = await activityExecutionMapper.GetPersistableOutputAsync(activityExecutionContext); + var outputValues = await activityPropertyLogPersistenceEvaluator.GetPersistableOutputAsync(activityExecutionContext); var properties = new Dictionary { [scheduledActivitiesKey] = JsonSerializer.Serialize(scheduledActivities), diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs index 4457225f6..8a3b04c69 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs @@ -1,88 +1,20 @@ -using System.Text.Json; -using System.Text.Json.Serialization; -using Elsa.Expressions.Contracts; -using Elsa.Expressions.Models; -using Elsa.Extensions; -using Elsa.Workflows.Activities; using Elsa.Workflows.LogPersistence; -using Elsa.Workflows.LogPersistence.Strategies; -using Elsa.Workflows.Management.Options; -using Elsa.Workflows.Models; using Elsa.Workflows.Runtime.Entities; -using Elsa.Workflows.Serialization.Converters; using Elsa.Workflows.State; -using Humanizer; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime; /// public class DefaultActivityExecutionMapper : IActivityExecutionMapper { - private readonly JsonSerializerOptions _logPersistenceConfigSerializerOptions; - private readonly IOptions _options; - private readonly IExpressionEvaluator _expressionEvaluator; - private readonly ILogger _logger; - private readonly IDictionary _logPersistenceStrategies; - - public DefaultActivityExecutionMapper( - IOptions options, - ILogPersistenceStrategyService logPersistenceStrategyService, - IExpressionEvaluator expressionEvaluator, - IExpressionDescriptorRegistry expressionDescriptorRegistry, - ILogger logger) + public ActivityExecutionRecord Map(ActivityExecutionContext source) { - _options = options; - _expressionEvaluator = expressionEvaluator; - _logger = logger; - _logPersistenceStrategies = logPersistenceStrategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x); - - _logPersistenceConfigSerializerOptions = new JsonSerializerOptions - { - PropertyNameCaseInsensitive = true - }.WithConverters( - new ExpressionJsonConverterFactory(expressionDescriptorRegistry), - new JsonStringEnumConverter(), - new ExpandoObjectConverterFactory()); - } - - private const string LegacyLogPersistenceModeKey = "logPersistenceMode"; - private const string LogPersistenceConfigKey = "logPersistenceConfig"; - - /// - public async Task MapAsync(ActivityExecutionContext source) - { - /* The following legacy JSON structure is expected to be found in the custom properties of the workflow and activity: - * { - * "logPersistenceMode": { - * "default": "default", - * "inputs": { k : v }, - * "outputs": { k: v } - * } - * } - */ - - /* The following JSON structure is expected to be found in the custom properties of the workflow and activity: - * { - * "logPersistenceConfig": { - * "default": { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." }, - * "inputs": { "input1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } }, - * "outputs": { "output1" : { "evaluationMode": "Strategy", "strategyType": "Elsa.Workflows.LogPersistence.Strategies.Inherit, Elsa.Workflows.Core", "expression": "..." } } - * } - * } - */ - - var cancellationToken = source.CancellationToken; - var legacyActivityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary())!; - var rootActivityExecutionContext = source.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); - var workflow = (Workflow?)source.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? source.WorkflowExecutionContext.Workflow; - var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); - var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(source.ExpressionExecutionContext, source.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken); - var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary())!; var payload = GetPayload(source); - var outputs = await GetPersistableOutputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); - var inputs = await GetPersistableInputAsync(source, legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); + var outputs = source.GetOutputs(); + var inputs = source.GetInputs(); + var persistenceMap = source.GetLogPersistenceModeMap(); + var persistableInputs = GetPersistableProperties(inputs, persistenceMap.Inputs); + var persistableOutputs = GetPersistableProperties(outputs, persistenceMap.Outputs); return new() { @@ -92,8 +24,8 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper WorkflowInstanceId = source.WorkflowExecutionContext.Id, ActivityType = source.Activity.Type, ActivityName = source.Activity.Name, - ActivityState = inputs, - Outputs = outputs, + ActivityState = persistableInputs, + Outputs = persistableOutputs, Properties = source.Properties, Payload = payload, Exception = ExceptionState.FromException(source.Exception), @@ -105,162 +37,25 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper }; } - public async Task> GetPersistableOutputAsync(ActivityExecutionContext context) + /// + public Task MapAsync(ActivityExecutionContext source) { - var cancellationToken = context.WorkflowExecutionContext.CancellationToken; - var legacyActivityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); - var rootActivityExecutionContext = context.WorkflowExecutionContext.ActivityExecutionContexts.First(x => x.ParentActivityExecutionContext == null); - var workflow = (Workflow?)context.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? context.WorkflowExecutionContext.Workflow; - var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(rootActivityExecutionContext.ExpressionExecutionContext, workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken); - var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(context.ExpressionExecutionContext, context.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken); - var activityPersistenceProperties = context.Activity.CustomProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary()); - - return await GetPersistableOutputAsync( - context, - legacyActivityPersistenceProperties, - activityPersistenceProperties, - activityPersistencePropertyDefault, - cancellationToken); + return Task.FromResult(Map(source)); } - private async Task> GetPersistableOutputAsync( - ActivityExecutionContext context, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - var outputs = GetOutputs(context); - return await GetPersistablePropertiesAsync(context, outputs, "outputs", legacyActivityPersistenceProperties, activityPersistenceProperties, activityPersistencePropertyDefault, cancellationToken); - } - - private async Task> GetPersistableInputAsync(ActivityExecutionContext context, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - return await GetPersistablePropertiesAsync( - context, - context.ActivityState!, - "inputs", - legacyActivityPersistenceProperties, - activityPersistenceProperties, - activityPersistencePropertyDefault, - cancellationToken); - } - - private async Task> GetPersistablePropertiesAsync( - ActivityExecutionContext context, - IDictionary state, - string key, - IDictionary legacyActivityPersistenceProperties, - IDictionary activityPersistenceProperties, - LogPersistenceMode activityPersistencePropertyDefault, - CancellationToken cancellationToken) - { - return await FilterPropertiesUsingPersistenceMode( - context.ExpressionExecutionContext, - state, - legacyActivityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary())!, - activityPersistenceProperties!.GetValueOrDefault(key, () => new Dictionary())!, - activityPersistencePropertyDefault, - cancellationToken); - } - - private async Task GetDefaultPersistenceModeAsync(ExpressionExecutionContext expressionExecutionContext, IDictionary customProperties, Func defaultFactory, CancellationToken cancellationToken) - { - var legacyProperties = customProperties.GetValueOrDefault>(LegacyLogPersistenceModeKey, () => new Dictionary()); - var properties = customProperties.GetValueOrDefault>(LogPersistenceConfigKey, () => new Dictionary()); - var defaultPersistenceConfigObject = properties?.TryGetValue("default", out var defaultPersistenceConfigObjectValue) == true ? defaultPersistenceConfigObjectValue : null; - - if (defaultPersistenceConfigObject == null) - { - var legacyPersistencePropertyDefault = legacyProperties!.GetValueOrDefault("default", defaultFactory); - - if (legacyPersistencePropertyDefault == LogPersistenceMode.Inherit) - return defaultFactory(); - return legacyPersistencePropertyDefault; - } - - var defaultPersistenceConfig = Convert(defaultPersistenceConfigObject); - return await EvaluateLogPersistenceConfigAsync(defaultPersistenceConfig, expressionExecutionContext, defaultFactory, cancellationToken); - } - - private async Task> FilterPropertiesUsingPersistenceMode( - ExpressionExecutionContext expressionExecutionContext, - IDictionary state, - IDictionary obsoletePersistenceModeConfiguration, - IDictionary persistenceStrategyConfiguration, - LogPersistenceMode defaultLogPersistenceMode, - CancellationToken cancellationToken) + private IDictionary GetPersistableProperties(IDictionary state, IDictionary map) { var result = new Dictionary(); - - foreach (var value in state) + foreach (var stateEntry in state) { - var propKey = value.Key.Camelize(); - var logPersistenceConfigObject = persistenceStrategyConfiguration.GetValueOrDefault(propKey, () => null); - var logPersistenceConfig = Convert(logPersistenceConfigObject); - var mode = logPersistenceConfig != null - ? await EvaluateLogPersistenceConfigAsync(logPersistenceConfig, expressionExecutionContext, () => defaultLogPersistenceMode, cancellationToken) - : obsoletePersistenceModeConfiguration.GetValueOrDefault(propKey, () => defaultLogPersistenceMode); - - if (mode == LogPersistenceMode.Include || mode == LogPersistenceMode.Inherit && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit) - result.Add(value.Key, value.Value); + var mode = map.TryGetValue(stateEntry.Key, out var value) ? value : LogPersistenceMode.Include; + if (mode == LogPersistenceMode.Include) + result.Add(stateEntry.Key, stateEntry.Value); } return result; } - - private LogPersistenceConfiguration? Convert(object? value) - { - if (value == null) - return null; - - if (value is LogPersistenceConfiguration c) - return c; - - var json = JsonSerializer.Serialize(value); - var config = JsonSerializer.Deserialize(json, _logPersistenceConfigSerializerOptions); - - return config; - } - - private async Task EvaluateLogPersistenceConfigAsync(LogPersistenceConfiguration? config, ExpressionExecutionContext executionContext, Func defaultMode, CancellationToken cancellationToken) - { - if (config == null) - return defaultMode(); - - if (config.EvaluationMode == LogPersistenceEvaluationMode.Strategy) - { - var strategyTypeName = config.StrategyType ?? typeof(Inherit).GetSimpleAssemblyQualifiedName(); - var strategy = _logPersistenceStrategies.TryGetValue(strategyTypeName, out var v) ? v : null; - - if (strategy == null) - return defaultMode(); - - var strategyContext = new LogPersistenceStrategyContext(cancellationToken); - var logMode = await strategy.GetPersistenceModeAsync(strategyContext); - return logMode == LogPersistenceMode.Inherit ? defaultMode() : logMode; - } - - if (config.Expression == null) - return defaultMode(); - - var expression = config.Expression; - - try - { - return await _expressionEvaluator.EvaluateAsync(expression, executionContext); - } - catch (Exception e) - { - _logger.LogWarning(e, "Error evaluating log persistence expression"); - return defaultMode(); - } - } - + private static IDictionary GetPayload(ActivityExecutionContext source) { var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : null; @@ -271,30 +66,4 @@ public class DefaultActivityExecutionMapper : IActivityExecutionMapper return payload; } - - private static IDictionary GetOutputs(ActivityExecutionContext source) - { - var activity = source.Activity; - var expressionExecutionContext = source.ExpressionExecutionContext; - var activityDescriptor = source.ActivityDescriptor; - var outputDescriptors = activityDescriptor.Outputs; - - var outputs = outputDescriptors.ToDictionary(x => x.Name, x => - { - if (x.IsSerializable == false) - return "(not serializable)"; - - var cachedValue = activity.GetOutput(expressionExecutionContext, x.Name); - - if (cachedValue != null) - return cachedValue; - - if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue)) - return outputValue; - - return null; - }); - - return outputs; - } } \ No newline at end of file