Merge pull request #5708 from elsa-workflows/enh/log_sink

Add sink for persisting workflows execution log records

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

@@ -1,3 +1,4 @@
+using System.Diagnostics.CodeAnalysis;
 using System.Runtime.CompilerServices;
 using Elsa.Common.Models;
 using Elsa.Expressions.Contracts;
@@ -7,8 +8,6 @@ using Elsa.Extensions;
 using Elsa.Workflows.Activities.Flowchart.Attributes;
 using Elsa.Workflows.Attributes;
 using Elsa.Workflows.Contracts;
-using Elsa.Workflows.Exceptions;
-using Elsa.Workflows.UIHints;
 using Elsa.Workflows.Memory;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Options;
@@ -18,8 +17,8 @@ using Elsa.Workflows.Runtime.Models;
 using Elsa.Workflows.Runtime.Requests;
 using Elsa.Workflows.Runtime.UIHints;
 using Elsa.Workflows.Services;
+using Elsa.Workflows.UIHints;
 using JetBrains.Annotations;
-using System.Diagnostics.CodeAnalysis;
 
 namespace Elsa.Workflows.Runtime.Activities;
 
@@ -232,7 +231,6 @@ public class BulkDispatchWorkflows : Activity
         var input = context.WorkflowInput;
         var workflowInstanceId = input["WorkflowInstanceId"].ConvertTo<string>()!;
         var workflowSubStatus = input["WorkflowSubStatus"].ConvertTo<WorkflowSubStatus>();
-        var workflowOutput = input["WorkflowOutput"].ConvertTo<IDictionary<string, object>>();
         var finishedInstancesCount = context.GetProperty<long>(CompletedInstancesCountKey) + 1;
 
         context.SetProperty(CompletedInstancesCountKey, finishedInstancesCount);

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

@@ -3,14 +3,13 @@ using Elsa.Common.Models;
 using Elsa.Extensions;
 using Elsa.Workflows.Attributes;
 using Elsa.Workflows.Contracts;
-using Elsa.Workflows.Exceptions;
-using Elsa.Workflows.UIHints;
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime.Bookmarks;
 using Elsa.Workflows.Runtime.Contracts;
 using Elsa.Workflows.Runtime.Models;
 using Elsa.Workflows.Runtime.Requests;
 using Elsa.Workflows.Runtime.UIHints;
+using Elsa.Workflows.UIHints;
 using JetBrains.Annotations;
 
 namespace Elsa.Workflows.Runtime.Activities;

src/modules/Elsa.Workflows.Runtime/Bookmarks/EventBookmarkPayload.cs

@@ -1,14 +1,23 @@
 namespace Elsa.Workflows.Runtime.Bookmarks;
 
+/// <summary>
+/// The payload for an event bookmark.
+/// </summary>
 public class EventBookmarkPayload
 {
     private readonly string _eventName = default!;
 
+    /// <summary>
+    /// The payload for an event bookmark.
+    /// </summary>
     public EventBookmarkPayload(string eventName)
     {
         EventName = eventName;
     }
 
+    /// <summary>
+    /// The name of the event.
+    /// </summary>
     public string EventName
     {
         get => _eventName;

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

@@ -0,0 +1,12 @@
+namespace Elsa.Workflows.Runtime.Contracts;
+
+/// <summary>
+/// Represents a sink for storing workflow execution log records.
+/// </summary>
+public interface IWorkflowExecutionLogSink
+{
+    /// <summary>
+    /// Persists the execution logs of a workflow.
+    /// </summary>
+    Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken = default);
+}

src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs

@@ -107,6 +107,11 @@ public class WorkflowRuntimeFeature : FeatureBase
     /// A factory that instantiates an <see cref="IBackgroundActivityScheduler"/>.
     /// </summary>
     public Func<IServiceProvider, IBackgroundActivityScheduler> BackgroundActivityScheduler { get; set; } = sp => ActivatorUtilities.CreateInstance<LocalBackgroundActivityScheduler>(sp);
+
+    /// <summary>
+    /// Represents a sink for workflow execution logs.
+    /// </summary>
+    public Func<IServiceProvider, IWorkflowExecutionLogSink> WorkflowExecutionLogSink { get; set; } = sp => sp.GetRequiredService<StoreWorkflowExecutionLogSink>();
     
     /// <summary>
     /// A delegate to configure the <see cref="DistributedLockingOptions"/>.
@@ -206,6 +211,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             .AddScoped(WorkflowCancellationDispatcher)
             .AddScoped(WorkflowExecutionContextStore)
             .AddScoped(RunTaskDispatcher)
+            .AddScoped(WorkflowExecutionLogSink)
             .AddSingleton(BackgroundActivityScheduler)
             .AddSingleton<RandomLongIdentityGenerator>()
             .AddScoped<IBookmarkManager, DefaultBookmarkManager>()
@@ -218,6 +224,7 @@ public class WorkflowRuntimeFeature : FeatureBase
             .AddScoped<ITaskReporter, TaskReporter>()
             .AddScoped<SynchronousTaskDispatcher>()
             .AddScoped<BackgroundTaskDispatcher>()
+            .AddScoped<StoreWorkflowExecutionLogSink>()
             .AddScoped<IEventPublisher, EventPublisher>()
             .AddScoped<IWorkflowInbox, DefaultWorkflowInbox>()
             .AddScoped<IBookmarkUpdater, BookmarkUpdater>()

src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/PersistWorkflowExecutionLogMiddleware.cs

@@ -1,66 +1,19 @@
-using Elsa.Mediator.Contracts;
-using Elsa.Workflows.Contracts;
 using Elsa.Workflows.Pipelines.WorkflowExecution;
 using Elsa.Workflows.Runtime.Contracts;
-using Elsa.Workflows.Runtime.Entities;
-using Elsa.Workflows.Runtime.Notifications;
 
 namespace Elsa.Workflows.Runtime.Middleware.Workflows;
 
 /// <summary>
 /// Takes care of persisting workflow execution log entries.
 /// </summary>
-public class PersistWorkflowExecutionLogMiddleware : WorkflowExecutionMiddleware
+public class PersistWorkflowExecutionLogMiddleware(WorkflowMiddlewareDelegate next, IWorkflowExecutionLogSink sink) : WorkflowExecutionMiddleware(next)
 {
-    private readonly IWorkflowExecutionLogStore _workflowExecutionLogStore;
-    private readonly INotificationSender _notificationSender;
-    private readonly IIdentityGenerator _identityGenerator;
-
-    /// <inheritdoc />
-    public PersistWorkflowExecutionLogMiddleware(
-        WorkflowMiddlewareDelegate next,
-        IWorkflowExecutionLogStore workflowExecutionLogStore,
-        INotificationSender notificationSender,
-        IIdentityGenerator identityGenerator) : base(next)
-    {
-        _workflowExecutionLogStore = workflowExecutionLogStore;
-        _notificationSender = notificationSender;
-        _identityGenerator = identityGenerator;
-    }
-
     /// <inheritdoc />
     public override async ValueTask InvokeAsync(WorkflowExecutionContext context)
     {
         // Invoke next middleware.
         await Next(context);
 
-        // Persist workflow execution log entries.
-        var entries = context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord
-        {
-            Id = _identityGenerator.GenerateId(),
-            ActivityInstanceId = x.ActivityInstanceId,
-            ParentActivityInstanceId = x.ParentActivityInstanceId,
-            ActivityNodeId = x.NodeId,
-            ActivityId = x.ActivityId,
-            ActivityType = x.ActivityType,
-            ActivityTypeVersion = x.ActivityTypeVersion,
-            ActivityName = x.ActivityName,
-            Message = x.Message,
-            EventName = x.EventName,
-            WorkflowDefinitionId = context.Workflow.Identity.DefinitionId,
-            WorkflowDefinitionVersionId = context.Workflow.Identity.Id,
-            WorkflowInstanceId = context.Id,
-            WorkflowVersion = context.Workflow.Version,
-            Source = x.Source,
-            ActivityState = x.ActivityState,
-            Payload = x.Payload,
-            Timestamp = x.Timestamp,
-            Sequence = x.Sequence
-        }).ToList();
-        
-        await _workflowExecutionLogStore.AddManyAsync(entries, context.CancellationTokens.SystemCancellationToken);
-
-        // Publish notification.
-        await _notificationSender.SendAsync(new WorkflowExecutionLogUpdated(context), context.CancellationTokens.SystemCancellationToken);
+        await sink.PersistExecutionLogsAsync(context);
     }
 }

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

@@ -0,0 +1,43 @@
+using Elsa.Mediator.Contracts;
+using Elsa.Workflows.Contracts;
+using Elsa.Workflows.Runtime.Contracts;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Notifications;
+
+namespace Elsa.Workflows.Runtime.Services;
+
+/// <summary>
+/// This implementation saves <see cref="WorkflowExecutionLogRecord"/> directly through the store.
+/// </summary>
+public class StoreWorkflowExecutionLogSink(IWorkflowExecutionLogStore store, IIdentityGenerator identityGenerator, INotificationSender notificationSender) : IWorkflowExecutionLogSink
+{
+    /// <inheritdoc />
+    public async Task PersistExecutionLogsAsync(WorkflowExecutionContext context, CancellationToken cancellationToken)
+    {
+        var records = context.ExecutionLog.Select(x => new WorkflowExecutionLogRecord
+        {
+            Id = identityGenerator.GenerateId(),
+            ActivityInstanceId = x.ActivityInstanceId,
+            ParentActivityInstanceId = x.ParentActivityInstanceId,
+            ActivityNodeId = x.NodeId,
+            ActivityId = x.ActivityId,
+            ActivityType = x.ActivityType,
+            ActivityTypeVersion = x.ActivityTypeVersion,
+            ActivityName = x.ActivityName,
+            Message = x.Message,
+            EventName = x.EventName,
+            WorkflowDefinitionId = context.Workflow.Identity.DefinitionId,
+            WorkflowDefinitionVersionId = context.Workflow.Identity.Id,
+            WorkflowInstanceId = context.Id,
+            WorkflowVersion = context.Workflow.Version,
+            Source = x.Source,
+            ActivityState = x.ActivityState,
+            Payload = x.Payload,
+            Timestamp = x.Timestamp,
+            Sequence = x.Sequence
+        }).ToList();
+        
+        await store.AddManyAsync(records, context.CancellationTokens.SystemCancellationToken);
+        await notificationSender.SendAsync(new WorkflowExecutionLogUpdated(context), context.CancellationTokens.SystemCancellationToken);
+    }
+}