Introduce Log Persistence Strategy (#6057)

* Implement log persistence strategy management

Added interfaces, services, and strategies for log persistence. Introduced new endpoint to list available log persistence strategies. Updated configurations and dependency injections accordingly.

* Refactor log record methods to asynchronous

Updated methods for extracting and persisting log records to be asynchronous, enhancing performance and scalability. This change includes modifying interfaces and implementations for better async support in workflow execution logging.

* Remove commented code

* Support nullable values in ActivityState dictionaries

Update ActivityState to support nullable values by changing type to 'IDictionary<string, object?>'. Enhanced DefaultActivityExecutionMapper to handle multiple persistence strategies for logging inputs and outputs.

* Rename ShouldPersistAsync to GetPersistenceModeAsync

Refactor method names for log persistence strategies to improve readability and consistency. Added summary comments for clarification and removed redundant configurations from appsettings.json. Added implicit uses and updated namespaces for better maintainability.

* Refactor activity payload and output retrieval logic

Extract payload and output retrieval into `GetPayload` and `GetOutputs` methods respectively. This modularizes the code for better readability and maintainability, and allows for potential reusability of these methods in other parts of the codebase.

* Add new project reference and update PostgreSQL provider usage

Added a project reference to Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql in the test project file. Also modified the WorkflowServer setup to specify the assembly in the PostgreSQL provider configuration.

* Add agent persistence to WorkflowServer

Integrated agent support and persistence using PostgreSQL in WorkflowServer. This includes adding necessary project references and configuring agents in the workflow server setup.

src/apps/Elsa.Server.Web/Program.cs

@@ -3,7 +3,6 @@ using Elsa.Agents;
 using Elsa.Alterations.Extensions;
 using Elsa.Alterations.MassTransit.Extensions;
 using Elsa.Common.DistributedLocks.Noop;
-using Elsa.Common.Multitenancy;
 using Elsa.Dapper.Extensions;
 using Elsa.Dapper.Services;
 using Elsa.DropIns.Extensions;
@@ -29,8 +28,8 @@ using Elsa.Server.Web.Extensions;
 using Elsa.Server.Web.Filters;
 using Elsa.Tenants.AspNetCore;
 using Elsa.Tenants.Extensions;
-using Elsa.Workflows;
 using Elsa.Workflows.Api;
+using Elsa.Workflows.LogPersistence;
 using Elsa.Workflows.Management.Compression;
 using Elsa.Workflows.Management.Stores;
 using Elsa.Workflows.Runtime.Distributed.Extensions;

src/apps/Elsa.Server.Web/appsettings.json

@@ -2,14 +2,7 @@
   "Logging": {
     "LogLevel": {
       "Default": "Warning",
-      "Elsa": "Warning",
-      "MassTransit": "Warning",
-      "Microsoft.Extensions.Http": "Warning",
-      "Microsoft.Hosting.Lifetime": "Information",
-      "Microsoft.EntityFrameworkCore": "Warning",
-      "Microsoft.AspNetCore": "Warning",
-      "Quartz": "Warning",
-      "System.Net.Http": "Warning"
+      "Microsoft.Hosting.Lifetime": "Information"
     }
   },
   "AllowedHosts": "*",

src/clients/Elsa.Api.Client/Elsa.Api.Client.csproj.DotSettings

@@ -1,2 +1,4 @@
 <wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
-	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=helpers/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=helpers/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=resources_005Clogpersistencestrategies_005Ccontracts/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=resources_005Clogpersistencestrategies_005Cmodels/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/clients/Elsa.Api.Client/Extensions/DependencyInjectionExtensions.cs

@@ -5,6 +5,7 @@ using Elsa.Api.Client.Resources.ActivityExecutions.Contracts;
 using Elsa.Api.Client.Resources.Features.Contracts;
 using Elsa.Api.Client.Resources.Identity.Contracts;
 using Elsa.Api.Client.Resources.IncidentStrategies.Contracts;
+using Elsa.Api.Client.Resources.LogPersistenceStrategies;
 using Elsa.Api.Client.Resources.Scripting.Contracts;
 using Elsa.Api.Client.Resources.StorageDrivers.Contracts;
 using Elsa.Api.Client.Resources.Tasks.Contracts;
@@ -65,6 +66,7 @@ public static class DependencyInjectionExtensions
             services.AddApi<IVariableTypesApi>(builderOptions);
             services.AddApi<IWorkflowActivationStrategiesApi>(builderOptions);
             services.AddApi<IIncidentStrategiesApi>(builderOptions);
+            services.AddApi<ILogPersistenceStrategiesApi>(builderOptions);
             services.AddApi<ILoginApi>(builderOptions);
             services.AddApi<IFeaturesApi>(builderOptions);
             services.AddApi<IJavaScriptApi>(builderOptions);

src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Contracts/ILogPersistenceStrategiesApi.cs

@@ -0,0 +1,19 @@
+using Elsa.Api.Client.Shared.Models;
+using JetBrains.Annotations;
+using Refit;
+
+namespace Elsa.Api.Client.Resources.LogPersistenceStrategies;
+
+/// <summary>
+/// Represents a client API for managing log persistence strategies.
+/// </summary>
+[UsedImplicitly]
+public interface ILogPersistenceStrategiesApi
+{
+    /// <summary>
+    /// Lists log persistence strategies.
+    /// </summary>
+    /// <param name="cancellationToken">The cancellation token.</param>
+    [Get("/descriptors/log-persistence-strategies")]
+    Task<ListResponse<LogPersistenceStrategyDescriptor>> ListAsync(CancellationToken cancellationToken = default);
+}

src/clients/Elsa.Api.Client/Resources/LogPersistenceStrategies/Models/LogPersistenceStrategyDescriptor.cs

@@ -0,0 +1,12 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Api.Client.Resources.LogPersistenceStrategies;
+
+/// <summary>
+/// Represents a log persistence strategy.
+/// </summary>
+/// <param name="TypeName">The .NET type name of the log persistence strategy.</param>
+/// <param name="DisplayName">The display name of the log persistence strategy.</param>
+/// <param name="Description">The description of the log persistence strategy.</param>
+[UsedImplicitly]
+public record LogPersistenceStrategyDescriptor(string TypeName, string DisplayName, string? Description);

src/modules/Elsa.Workflows.Api/Endpoints/LogPersistenceStrategies/List/Endpoint.cs

@@ -0,0 +1,45 @@
+using System.ComponentModel;
+using System.ComponentModel.DataAnnotations;
+using System.Reflection;
+using Elsa.Abstractions;
+using Elsa.Extensions;
+using Elsa.Models;
+using Elsa.Workflows.LogPersistence;
+using Humanizer;
+
+namespace Elsa.Workflows.Api.Endpoints.LogPersistenceStrategies.List;
+
+/// <summary>
+/// Returns list of available <see cref="ILogPersistenceStrategy" /> implementations.
+/// </summary>
+internal class Endpoint(ILogPersistenceStrategyService logPersistenceStrategyService) : ElsaEndpointWithoutRequest<ListResponse<LogPersistenceStrategyDescriptor>>
+{
+    public override void Configure()
+    {
+        Get("/descriptors/log-persistence-strategies");
+        ConfigurePermissions("read:log-persistence-strategies");
+    }
+
+    public override Task<ListResponse<LogPersistenceStrategyDescriptor>> ExecuteAsync(CancellationToken cancellationToken)
+    {
+        var strategies = logPersistenceStrategyService.ListStrategies();
+        var descriptors = strategies.Select(LogPersistenceStrategyDescriptor.FromStrategy).OrderBy(x => x.DisplayName).ToList();
+        var response =new ListResponse<LogPersistenceStrategyDescriptor>(descriptors);
+        return Task.FromResult(response);
+    }
+}
+
+internal record LogPersistenceStrategyDescriptor(string DisplayName, string Description, string TypeName)
+{
+    public static LogPersistenceStrategyDescriptor FromStrategy(ILogPersistenceStrategy strategy)
+    {
+        var type = strategy.GetType();
+        var displayNameAttribute = type.GetCustomAttribute<DisplayNameAttribute>();
+        var descriptionAttribute = type.GetCustomAttribute<DescriptionAttribute>();
+        var displayAttribute = type.GetCustomAttribute<DisplayAttribute>();
+        var displayName = displayNameAttribute?.DisplayName ?? displayAttribute?.Name ?? type.Name.Replace("Strategy", "").Humanize();
+        var description = descriptionAttribute?.Description ?? displayAttribute?.Description ?? "";
+
+        return new LogPersistenceStrategyDescriptor(displayName, description, type.GetSimpleAssemblyQualifiedName());
+    }
+}

src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.cs

@@ -195,7 +195,7 @@ public partial class ActivityExecutionContext : IExecutionContext, IDisposable
     /// <summary>
     /// Stores the evaluated inputs, serialized, for the current activity for historical purposes.
     /// </summary>
-    public IDictionary<string, object> ActivityState { get; set; } = new Dictionary<string, object>();
+    public IDictionary<string, object?> ActivityState { get; set; } = new Dictionary<string, object?>();
 
     /// <summary>
     /// Schedules the specified activity to be executed.

src/modules/Elsa.Workflows.Core/Elsa.Workflows.Core.csproj.DotSettings

@@ -12,5 +12,10 @@
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=contracts/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=enums/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=extensions/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Ccontexts/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Ccontracts/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cenums/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Coptions/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=logpersistence_005Cservices/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=models/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=services/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/modules/Elsa.Workflows.Core/Features/WorkflowsFeature.cs

@@ -9,6 +9,8 @@ using Elsa.Features.Services;
 using Elsa.Workflows.ActivationValidators;
 using Elsa.Workflows.Builders;
 using Elsa.Workflows.IncidentStrategies;
+using Elsa.Workflows.LogPersistence;
+using Elsa.Workflows.LogPersistence.Strategies;
 using Elsa.Workflows.Middleware.Activities;
 using Elsa.Workflows.Middleware.Workflows;
 using Elsa.Workflows.Pipelines.ActivityExecution;
@@ -194,9 +196,16 @@ public class WorkflowsFeature : FeatureBase
             .AddScoped<IUIHintHandler, CheckListUIHintHandler>()
             .AddScoped<IUIHintHandler, JsonEditorUIHintHandler>()
 
-            // Logger state generators
+            // Logger state generators.
             .AddSingleton(WorkflowLoggerStateGenerator)
             .AddSingleton(ActivityLoggerStateGenerator)
+            
+            // Log Persistence Strategies.
+            .AddScoped<ILogPersistenceStrategyService, DefaultLogPersistenceStrategyService>()
+            .AddScoped<ILogPersistenceStrategy, Include>()
+            .AddScoped<ILogPersistenceStrategy, Exclude>()
+            .AddScoped<ILogPersistenceStrategy, Inherit>()
+            .AddScoped<ILogPersistenceStrategy, Configuration>()
 
             // Logging
             .AddLogging();

src/modules/Elsa.Workflows.Core/LogPersistence/Contexts/LogPersistenceStrategyContext.cs

@@ -0,0 +1,3 @@
+namespace Elsa.Workflows.LogPersistence;
+
+public record LogPersistenceStrategyContext(CancellationToken CancellationToken);

src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategy.cs

@@ -0,0 +1,6 @@
+namespace Elsa.Workflows.LogPersistence;
+
+public interface ILogPersistenceStrategy
+{
+    Task<LogPersistenceMode> GetPersistenceModeAsync(LogPersistenceStrategyContext context);
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Contracts/ILogPersistenceStrategyService.cs

@@ -0,0 +1,6 @@
+namespace Elsa.Workflows.LogPersistence;
+
+public interface ILogPersistenceStrategyService
+{
+    IEnumerable<ILogPersistenceStrategy> ListStrategies();
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Enums/LogPersistenceMode.cs

@@ -1,25 +1,25 @@
-using JetBrains.Annotations;
-
-namespace Elsa.Workflows;
-
-/// <summary>
-/// Define the Log Persistence mode to store information
-/// </summary>
-[UsedImplicitly]
-public enum LogPersistenceMode
-{
-    /// <summary>
-    /// Persist using the Parent mode
-    /// </summary>
-    Inherit,
-
-    /// <summary>
-    /// Include property to store
-    /// </summary>
-    Include,
-
-    /// <summary>
-    /// Exclude Property to store
-    /// </summary>
-    Exclude
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.LogPersistence;
+
+/// <summary>
+/// Define the Log Persistence mode to store information
+/// </summary>
+[UsedImplicitly]
+public enum LogPersistenceMode
+{
+    /// <summary>
+    /// Persist using the Parent mode
+    /// </summary>
+    Inherit,
+
+    /// <summary>
+    /// Include property to store
+    /// </summary>
+    Include,
+
+    /// <summary>
+    /// Exclude Property to store
+    /// </summary>
+    Exclude
 }

src/modules/Elsa.Workflows.Core/LogPersistence/Options/LogPersistenceOptions.cs

@@ -0,0 +1,6 @@
+namespace Elsa.Workflows.LogPersistence;
+
+public class LogPersistenceOptions
+{
+    public LogPersistenceMode ConfiguredPersistenceMode { get; set; } = LogPersistenceMode.Inherit;
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Services/DefaultLogPersistenceStrategyService.cs

@@ -0,0 +1,12 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.LogPersistence;
+
+[UsedImplicitly]
+public class DefaultLogPersistenceStrategyService(IEnumerable<ILogPersistenceStrategy> strategies) : ILogPersistenceStrategyService
+{
+    public IEnumerable<ILogPersistenceStrategy> ListStrategies()
+    {
+        return strategies;
+    }
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Configuration.cs

@@ -0,0 +1,16 @@
+using JetBrains.Annotations;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Workflows.LogPersistence.Strategies;
+
+/// <summary>
+/// A log persistence strategy that uses the configured persistence mode.
+/// </summary>
+[UsedImplicitly]
+public class Configuration(IOptions<LogPersistenceOptions> options) : ILogPersistenceStrategy
+{
+    public Task<LogPersistenceMode> GetPersistenceModeAsync(LogPersistenceStrategyContext context)
+    {
+        return Task.FromResult(options.Value.ConfiguredPersistenceMode);
+    }
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Exclude.cs

@@ -0,0 +1,15 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.LogPersistence.Strategies;
+
+/// <summary>
+/// A log persistence strategy that returns <see cref="LogPersistenceMode.Exclude"/>.
+/// </summary>
+[UsedImplicitly]
+public class Exclude : ILogPersistenceStrategy
+{
+    public Task<LogPersistenceMode> GetPersistenceModeAsync(LogPersistenceStrategyContext context)
+    {
+        return Task.FromResult(LogPersistenceMode.Exclude);
+    }
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Include.cs

@@ -0,0 +1,15 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.LogPersistence.Strategies;
+
+/// <summary>
+/// A log persistence strategy that returns <see cref="LogPersistenceMode.Include"/>. 
+/// </summary>
+[UsedImplicitly]
+public class Include : ILogPersistenceStrategy
+{
+    public Task<LogPersistenceMode> GetPersistenceModeAsync(LogPersistenceStrategyContext context)
+    {
+        return Task.FromResult(LogPersistenceMode.Include);
+    }
+}

src/modules/Elsa.Workflows.Core/LogPersistence/Strategies/Inherit.cs

@@ -0,0 +1,15 @@
+using JetBrains.Annotations;
+
+namespace Elsa.Workflows.LogPersistence.Strategies;
+
+/// <summary>
+/// A log persistence strategy that returns <see cref="LogPersistenceMode.Inherit"/>.
+/// </summary>
+[UsedImplicitly]
+public class Inherit : ILogPersistenceStrategy
+{
+    public Task<LogPersistenceMode> GetPersistenceModeAsync(LogPersistenceStrategyContext context)
+    {
+        return Task.FromResult(LogPersistenceMode.Inherit);
+    }
+}

src/modules/Elsa.Workflows.Management/Features/WorkflowManagementFeature.cs

@@ -12,6 +12,7 @@ using Elsa.Features.Abstractions;
 using Elsa.Features.Attributes;
 using Elsa.Features.Services;
 using Elsa.Workflows.Features;
+using Elsa.Workflows.LogPersistence;
 using Elsa.Workflows.Management.Activities.WorkflowDefinitionActivity;
 using Elsa.Workflows.Management.Compression;
 using Elsa.Workflows.Management.Contracts;

src/modules/Elsa.Workflows.Management/Options/ManagementOptions.cs

@@ -1,3 +1,4 @@
+using Elsa.Workflows.LogPersistence;
 using Elsa.Workflows.Management.Models;
 
 namespace Elsa.Workflows.Management.Options;

src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs

@@ -109,7 +109,6 @@ public class BulkDispatchWorkflows : Activity
 
         await foreach (var item in items)
         {
-            //context.DeferTask(async () => await DispatchChildWorkflowAsync(context, item));
             await DispatchChildWorkflowAsync(context, item);
             dispatchedInstancesCount++;
         }

src/modules/Elsa.Workflows.Runtime/Contracts/IActivityExecutonMapper.cs

@@ -10,5 +10,5 @@ public interface IActivityExecutionMapper
     /// <summary>
     /// Maps an activity execution context to an activity execution record.
     /// </summary>
-    ActivityExecutionRecord Map(ActivityExecutionContext source);
+    Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source);
 }

src/modules/Elsa.Workflows.Runtime/Contracts/ILogRecordExtractor.cs

@@ -6,5 +6,5 @@ namespace Elsa.Workflows.Runtime;
 public interface ILogRecordExtractor<T> where T: ILogRecord
 {
     /// Extracts execution logs from a workflow execution context.
-    IEnumerable<T> ExtractLogRecords(WorkflowExecutionContext context);
+    Task<IEnumerable<T>> ExtractLogRecordsAsync(WorkflowExecutionContext context);
 }

src/modules/Elsa.Workflows.Runtime/Entities/ActivityExecutionRecord.cs

@@ -42,7 +42,7 @@ public class ActivityExecutionRecord : Entity, ILogRecord
     /// <summary>
     /// The state of the activity at the time this record is created or last updated.
     /// </summary>
-    public IDictionary<string, object>? ActivityState { get; set; }
+    public IDictionary<string, object?>? ActivityState { get; set; }
     
     /// <summary>
     /// Any additional payload associated with the log record.

src/modules/Elsa.Workflows.Runtime/Services/ActivityExecutionRecordExtractor.cs

@@ -6,9 +6,10 @@ namespace Elsa.Workflows.Runtime;
 public class ActivityExecutionRecordExtractor(IActivityExecutionMapper activityExecutionMapper) : ILogRecordExtractor<ActivityExecutionRecord>
 {
     /// <inheritdoc />
-    public IEnumerable<ActivityExecutionRecord> ExtractLogRecords(WorkflowExecutionContext context)
+    public async Task<IEnumerable<ActivityExecutionRecord>> ExtractLogRecordsAsync(WorkflowExecutionContext context)
     {
         var activityExecutionContexts = context.ActivityExecutionContexts;
-        return activityExecutionContexts.Select(activityExecutionMapper.Map).ToList();
+        var tasks = activityExecutionContexts.Select(activityExecutionMapper.MapAsync).ToList();
+        return await Task.WhenAll(tasks);
     }
 }

src/modules/Elsa.Workflows.Runtime/Services/DefaultActivityExecutionMapper.cs

@@ -1,5 +1,6 @@
 using Elsa.Extensions;
 using Elsa.Workflows.Activities;
+using Elsa.Workflows.LogPersistence;
 using Elsa.Workflows.Management.Options;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime.Entities;
@@ -10,14 +11,24 @@ using Microsoft.Extensions.Options;
 namespace Elsa.Workflows.Runtime;
 
 /// <inheritdoc />
-public class DefaultActivityExecutionMapper(IOptions<ManagementOptions> options) : IActivityExecutionMapper
+public class DefaultActivityExecutionMapper : IActivityExecutionMapper
 {
-    private const string LogPersistenceModeKey = "logPersistenceMode";
+    private readonly IOptions<ManagementOptions> _options;
+    private readonly IDictionary<string, ILogPersistenceStrategy> _logPersistenceStrategies;
+
+    public DefaultActivityExecutionMapper(IOptions<ManagementOptions> options, ILogPersistenceStrategyService logPersistenceStrategyService)
+    {
+        _options = options;
+        _logPersistenceStrategies = logPersistenceStrategyService.ListStrategies().ToDictionary(x => x.GetType().GetSimpleAssemblyQualifiedName(), x => x);
+    }
+
+    private const string LegacyLogPersistenceModeKey = "logPersistenceMode";
+    private const string LogPersistenceStrategyKey = "logPersistenceStrategy";
 
     /// <inheritdoc />
-    public ActivityExecutionRecord Map(ActivityExecutionContext source)
+    public async Task<ActivityExecutionRecord> 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",
@@ -27,42 +38,38 @@ public class DefaultActivityExecutionMapper(IOptions<ManagementOptions> options)
          *  }
          */
 
+        /* The following JSON structure is expected to be found in the custom properties of the workflow and activity:
+         * {
+         *      "logPersistenceStrategy": {
+         *          "default": "null",
+         *          "inputs": { k : v },
+         *          "outputs": { k: v }
+         *          }
+         *  }
+         */
+
+        var cancellationToken = source.WorkflowExecutionContext.CancellationToken;
         var workflow = (Workflow?)source.GetAncestors().FirstOrDefault(x => x.Activity is Workflow)?.Activity ?? source.WorkflowExecutionContext.Workflow;
-        var workflowPersistenceProperty = GetDefaultPersistenceMode(workflow.CustomProperties, () => options.Value.LogPersistenceMode);
-        var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceModeKey, () => new Dictionary<string, object?>());
-        var activityPersistencePropertyDefault = GetDefaultPersistenceMode(source.Activity.CustomProperties, () => workflowPersistenceProperty);
+        var workflowPersistenceProperty = await GetDefaultPersistenceModeAsync(workflow.CustomProperties, () => _options.Value.LogPersistenceMode, cancellationToken);
+        var activityPersistencePropertyDefault = await GetDefaultPersistenceModeAsync(source.Activity.CustomProperties, () => workflowPersistenceProperty, cancellationToken);
+        var legacyActivityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LegacyLogPersistenceModeKey, () => new Dictionary<string, object?>());
+        var activityPersistenceProperties = source.Activity.CustomProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceStrategyKey, () => new Dictionary<string, object?>());
+        var payload = GetPayload(source);
+        var outputs = GetOutputs(source);
 
-        // Get any outcomes that were added to the activity execution context.
-        var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : default;
-        var payload = new Dictionary<string, object>();
+        outputs = await StorePropertyUsingPersistenceMode(
+            outputs,
+            legacyActivityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary<string, object>())!,
+            activityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary<string, object>())!,
+            activityPersistencePropertyDefault,
+            cancellationToken);
 
-        if (outcomes != null)
-            payload.Add("Outcomes", outcomes);
-
-        // Get any outputs that were added to the activity execution context.
-        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 != default)
-                return cachedValue;
-
-            if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue))
-                return outputValue;
-
-            return default;
-        });
-
-        outputs = StorePropertyUsingPersistenceMode(outputs, activityPersistenceProperties!.GetValueOrDefault("outputs", () => new Dictionary<string, object>())!, activityPersistencePropertyDefault);
-        var inputs = StorePropertyUsingPersistenceMode(source.ActivityState, activityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary<string, object>())!, activityPersistencePropertyDefault);
+        var inputs = await StorePropertyUsingPersistenceMode(
+            source.ActivityState, 
+            legacyActivityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary<string, object>())!,
+            activityPersistenceProperties!.GetValueOrDefault("inputs", () => new Dictionary<string, object>())!,
+            activityPersistencePropertyDefault,
+            cancellationToken);
 
         return new ActivityExecutionRecord
         {
@@ -85,32 +92,63 @@ public class DefaultActivityExecutionMapper(IOptions<ManagementOptions> options)
         };
     }
 
-    private static LogPersistenceMode GetDefaultPersistenceMode(IDictionary<string, object> customProperties, Func<LogPersistenceMode> defaultFactory)
+    private async Task<LogPersistenceMode> GetDefaultPersistenceModeAsync(IDictionary<string, object> customProperties, Func<LogPersistenceMode> defaultFactory, CancellationToken cancellationToken)
     {
-        var properties = customProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceModeKey, () => new Dictionary<string, object?>());
-        var persistencePropertyDefault = properties!.GetValueOrDefault("default", defaultFactory);
+        var legacyProperties = customProperties.GetValueOrDefault<IDictionary<string, object?>>(LegacyLogPersistenceModeKey, () => new Dictionary<string, object?>());
+        var properties = customProperties.GetValueOrDefault<IDictionary<string, object?>>(LogPersistenceStrategyKey, () => new Dictionary<string, object?>());
+        var defaultPersistenceStrategyTypeName = properties!.GetValueOrDefault<string>("default");
 
-        if (persistencePropertyDefault == LogPersistenceMode.Inherit)
+        if (defaultPersistenceStrategyTypeName == null)
+        {
+            var legacyPersistencePropertyDefault = legacyProperties!.GetValueOrDefault("default", defaultFactory);
+
+            if (legacyPersistencePropertyDefault == LogPersistenceMode.Inherit)
+                return defaultFactory();
+            return legacyPersistencePropertyDefault;
+        }
+
+        var strategy = _logPersistenceStrategies.TryGetValue(defaultPersistenceStrategyTypeName, out var v) ? v : null;
+
+        if (strategy == null)
             return defaultFactory();
-        return persistencePropertyDefault;
+
+        var strategyContext = new LogPersistenceStrategyContext(cancellationToken);
+        return await strategy.GetPersistenceModeAsync(strategyContext);
     }
 
-    private static Dictionary<string, object?> StorePropertyUsingPersistenceMode(IDictionary<string, object?> state, IDictionary<string, object> persistenceModeConfiguration, LogPersistenceMode defaultLogPersistenceMode)
+    private async Task<Dictionary<string, object?>> StorePropertyUsingPersistenceMode(
+        IDictionary<string, object?> state,
+        IDictionary<string, object> persistenceModeConfiguration,
+        IDictionary<string, object> persistenceStrategyConfiguration,
+        LogPersistenceMode defaultLogPersistenceMode,
+        CancellationToken cancellationToken)
     {
         var result = new Dictionary<string, object?>();
 
         foreach (var value in state)
         {
-            var persistence = persistenceModeConfiguration.GetValueOrDefault(value.Key.Camelize(), () => defaultLogPersistenceMode);
-            if (persistence.Equals(LogPersistenceMode.Include)
-                || (persistence.Equals(LogPersistenceMode.Inherit) && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit))
+            var propKey = value.Key.Camelize();
+            var strategy = persistenceStrategyConfiguration.GetValueOrDefault(propKey) is string strategyTypeName ? _logPersistenceStrategies.TryGetValue(strategyTypeName, out var v) ? v : null : null;
+            var mode = defaultLogPersistenceMode;
+
+            if (strategy != null)
+            {
+                var strategyContext = new LogPersistenceStrategyContext(cancellationToken);
+                mode = await strategy.GetPersistenceModeAsync(strategyContext);
+            }
+            else
+            {
+                mode = persistenceModeConfiguration.GetValueOrDefault(propKey, () => defaultLogPersistenceMode);
+            }
+
+            if (mode == LogPersistenceMode.Include || mode == LogPersistenceMode.Inherit && defaultLogPersistenceMode is LogPersistenceMode.Include or LogPersistenceMode.Inherit)
                 result.Add(value.Key, value.Value);
         }
 
         return result;
     }
 
-    private ActivityStatus GetAggregateStatus(ActivityExecutionContext context)
+    private static ActivityStatus GetAggregateStatus(ActivityExecutionContext context)
     {
         // If any child activity is faulted, the aggregate status is faulted.
         var descendantContexts = context.GetDescendants().ToList();
@@ -120,4 +158,42 @@ public class DefaultActivityExecutionMapper(IOptions<ManagementOptions> options)
 
         return context.Status;
     }
+    
+    private static IDictionary<string, object> GetPayload(ActivityExecutionContext source)
+    {
+        var outcomes = source.JournalData.TryGetValue("Outcomes", out var resultValue) ? resultValue as string[] : default;
+        var payload = new Dictionary<string, object>();
+
+        if (outcomes != null)
+            payload.Add("Outcomes", outcomes);
+        
+        return payload;
+    }
+
+    // Get the outputs of the activity execution context.
+    private static IDictionary<string, object?> 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 != default)
+                return cachedValue;
+
+            if (x.ValueGetter(activity) is Output output && source.TryGet(output.MemoryBlockReference(), out var outputValue))
+                return outputValue;
+
+            return default;
+        });
+        
+        return outputs;
+    }
 }

src/modules/Elsa.Workflows.Runtime/Services/StoreActivityExecutionLogSink.cs

@@ -1,6 +1,7 @@
 using Elsa.Mediator.Contracts;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.Notifications;
+using Open.Linq.AsyncExtensions;
 
 namespace Elsa.Workflows.Runtime.Services;
 
@@ -13,7 +14,7 @@ public class StoreActivityExecutionLogSink(IActivityExecutionStore activityExecu
     /// <inheritdoc />
     public async Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken = default)
     {
-        var records = extractor.ExtractLogRecords(context).ToList();
+        var records = await extractor.ExtractLogRecordsAsync(context).ToList();
         await activityExecutionStore.SaveManyAsync(records, cancellationToken);
         await notificationSender.SendAsync(new ActivityExecutionLogUpdated(context, records), cancellationToken);
     }

src/modules/Elsa.Workflows.Runtime/Services/StoreWorkflowExecutionLogSink.cs

@@ -1,6 +1,7 @@
 using Elsa.Mediator.Contracts;
 using Elsa.Workflows.Runtime.Entities;
 using Elsa.Workflows.Runtime.Notifications;
+using Open.Linq.AsyncExtensions;
 
 namespace Elsa.Workflows.Runtime;
 
@@ -12,7 +13,7 @@ public class StoreWorkflowExecutionLogSink(IWorkflowExecutionLogStore store, ILo
     /// <inheritdoc />
     public async Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken)
     {
-        var records = extractor.ExtractLogRecords(context).ToList();
+        var records = await extractor.ExtractLogRecordsAsync(context).ToList();
         await store.AddManyAsync(records, context.CancellationToken);
         await notificationSender.SendAsync(new WorkflowExecutionLogUpdated(context), context.CancellationToken);
     }

src/modules/Elsa.Workflows.Runtime/Services/WorkflowExecutionLogRecordExtractor.cs

@@ -6,9 +6,9 @@ namespace Elsa.Workflows.Runtime;
 public class WorkflowExecutionLogRecordExtractor(IIdentityGenerator identityGenerator) : ILogRecordExtractor<WorkflowExecutionLogRecord>
 {
     /// <inheritdoc />
-    public IEnumerable<WorkflowExecutionLogRecord> ExtractLogRecords(WorkflowExecutionContext context)
+    public Task<IEnumerable<WorkflowExecutionLogRecord>> ExtractLogRecordsAsync(WorkflowExecutionContext context)
     {
-        return context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord
+        var records = context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord
         {
             Id = identityGenerator.GenerateId(),
             ActivityInstanceId = x.ActivityInstanceId,
@@ -30,5 +30,7 @@ public class WorkflowExecutionLogRecordExtractor(IIdentityGenerator identityGene
             Timestamp = x.Timestamp,
             Sequence = x.Sequence
         });
+        
+        return Task.FromResult(records);
     }
 }

test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj

@@ -21,6 +21,7 @@
     <ItemGroup>
       <ProjectReference Include="..\..\..\src\apps\Elsa.Server.Web\Elsa.Server.Web.csproj" />
       <ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared.Component\Elsa.Testing.Shared.Component.csproj" />
+      <ProjectReference Include="..\..\..\src\modules\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql.csproj" />
       <ProjectReference Include="..\..\..\src\modules\Elsa.AzureServiceBus\Elsa.AzureServiceBus.csproj" />
       <ProjectReference Include="..\..\..\src\modules\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
     </ItemGroup>

test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs

@@ -1,4 +1,5 @@
 using System.Net.Http.Headers;
+using Elsa.Agents;
 using Elsa.Alterations.Extensions;
 using Elsa.AzureServiceBus.ComponentTests.Extensions;
 using Elsa.EntityFrameworkCore.Extensions;
@@ -75,6 +76,8 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
                     http.UseCache();
                 });
                 elsa.UseAzureServiceBus();
+                elsa.UseAgents();
+                elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(typeof(AgentsPostgreSqlProvidersExtensions).Assembly, dbConnectionString)));
             };
         }
 

test/component/Elsa.Workflows.ComponentTests/Elsa.Workflows.ComponentTests.csproj

@@ -17,6 +17,7 @@
         <ProjectReference Include="..\..\..\src\apps\Elsa.Server.Web\Elsa.Server.Web.csproj"/>
         <ProjectReference Include="..\..\..\src\clients\Elsa.Api.Client\Elsa.Api.Client.csproj"/>
         <ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared.Component\Elsa.Testing.Shared.Component.csproj" />
+        <ProjectReference Include="..\..\..\src\modules\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql.csproj" />
         <ProjectReference Include="..\..\..\src\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
     </ItemGroup>
 

test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs

@@ -3,6 +3,7 @@ using System.Reflection;
 using Elsa.Agents;
 using Elsa.Alterations.Extensions;
 using Elsa.Caching;
+using Elsa.EntityFrameworkCore;
 using Elsa.EntityFrameworkCore.Extensions;
 using Elsa.EntityFrameworkCore.Modules.Alterations;
 using Elsa.EntityFrameworkCore.Modules.Identity;
@@ -106,7 +107,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
                     http.UseCache();
                 });
                 elsa.UseAgents();
-                elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString)));
+                elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(typeof(AgentsPostgreSqlProvidersExtensions).Assembly, dbConnectionString)));
             };
         }