Incremental work on timers and trigger & bookmark indexing

src/activities/Elsa.Activities.Scheduling/Activities/Delay.cs

@@ -0,0 +1,23 @@
+using System;
+using Elsa.Attributes;
+using Elsa.Contracts;
+using Elsa.Models;
+
+namespace Elsa.Activities.Scheduling;
+
+public class Delay : Activity
+{
+    [Input] public Input<TimeSpan> TimeSpan { get; set; } = default!;
+
+    protected override void Execute(ActivityExecutionContext context)
+    {
+        var timeSpan = context.ExpressionExecutionContext.Get(TimeSpan);
+        var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
+        var resumeAt = clock.UtcNow.Add(timeSpan);
+        var payload = new DelayPayload(resumeAt);
+
+        context.SetBookmark(payload);
+    }
+}
+
+public record DelayPayload(DateTime ResumeAt);

src/activities/Elsa.Activities.Scheduling/Contracts/IJobScheduler.cs

@@ -6,4 +6,5 @@ namespace Elsa.Activities.Scheduling.Contracts;
 public interface IJobScheduler
 {
     Task ScheduleAsync(IJob job, ISchedule schedule, CancellationToken cancellationToken = default);
+    Task ClearAsync(CancellationToken cancellationToken = default);
 }

src/activities/Elsa.Activities.Scheduling/Elsa.Activities.Scheduling.csproj.DotSettings

@@ -1,2 +1,3 @@
 <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/=activities/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=triggers/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/activities/Elsa.Activities.Scheduling/Handlers/ScheduleWorkflowsHandler.cs

@@ -1,3 +1,4 @@
+using System.Linq;
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
@@ -5,6 +6,7 @@ using Elsa.Activities.Scheduling.Contracts;
 using Elsa.Activities.Scheduling.Jobs;
 using Elsa.Activities.Scheduling.Schedules;
 using Elsa.Mediator.Contracts;
+using Elsa.Persistence.Extensions;
 using Elsa.Persistence.Requests;
 using Elsa.Runtime.Notifications;
 
@@ -23,7 +25,11 @@ public class ScheduleWorkflowsHandler : INotificationHandler<TriggerIndexingFini
 
     public async Task HandleAsync(TriggerIndexingFinished notification, CancellationToken cancellationToken)
     {
-        var timerTriggers = await _requestSender.RequestAsync(FindWorkflowTriggers.ForTrigger<Timer>(), cancellationToken);
+        // Unschedule everything.
+        await _jobScheduler.ClearAsync(cancellationToken);
+        
+        // Select all Timer triggers.
+        var timerTriggers = notification.Triggers.Filter<Timer>().ToList();
 
         foreach (var trigger in timerTriggers)
         {

src/activities/Elsa.Activities.Scheduling/Triggers/Timer.cs

@@ -6,6 +6,9 @@ using Elsa.Models;
 
 namespace Elsa.Activities.Scheduling;
 
+/// <summary>
+/// Represents a timer to periodically trigger the workflow it is associated with.
+/// </summary>
 public class Timer : Trigger
 {
     [Input] public Input<TimeSpan> Interval { get; set; } = default!;

src/core/Elsa.Core/Contracts/IHasher.cs

@@ -6,4 +6,5 @@ namespace Elsa.Contracts;
 public interface IHasher
 {
     public string Hash(object value);
+    public string Hash(string value);
 }

src/core/Elsa.Core/Contracts/IPayloadSerializer.cs

@@ -0,0 +1,7 @@
+namespace Elsa.Contracts;
+
+public interface IPayloadSerializer
+{
+    T Deserialize<T>(string json) where T : notnull;
+    string Serialize<T>(T payload) where T : notnull;
+}

src/core/Elsa.Core/Models/ActivityExecutionContext.cs

@@ -47,33 +47,31 @@ public class ActivityExecutionContext
             ScheduleActivity(activity, completionCallback);
     }
 
-    public void SetBookmarks(IEnumerable<object> hashInputs, IDictionary<string, object?>? data = default, ExecuteActivityDelegate? callback = default)
+    public void SetBookmarks(IEnumerable<object> payloads, ExecuteActivityDelegate? callback = default)
     {
-        foreach (var hashInput in hashInputs)
-            SetBookmark(hashInput, data, callback);
+        foreach (var hashInput in payloads)
+            SetBookmark(payloads, callback);
     }
 
     public void SetBookmarks(IEnumerable<Bookmark> bookmarks) => _bookmarks.AddRange(bookmarks);
     public void SetBookmark(Bookmark bookmark) => _bookmarks.Add(bookmark);
 
-    public void SetBookmark(object? hashInput, IDictionary<string, object?>? data = default, ExecuteActivityDelegate? callback = default)
+    public void SetBookmark(object? payload, ExecuteActivityDelegate? callback = default)
     {
         var hasher = GetRequiredService<IHasher>();
-        var hash = hashInput != null ? hasher.Hash(hashInput) : default;
-        SetBookmark(hash, data, callback);
-    }
-
-    public void SetBookmark(string? hash, IDictionary<string, object?>? data = default, ExecuteActivityDelegate? callback = default)
-    {
+        
         var identityGenerator = GetRequiredService<IIdentityGenerator>();
+        var payloadSerializer = GetRequiredService<IPayloadSerializer>();
+        var payloadJson = payload != null ? payloadSerializer.Serialize(payload) : default;
+        var hash = payloadJson != null ? hasher.Hash(payloadJson) : default;
 
         SetBookmark(new Bookmark(
             identityGenerator.GenerateId(),
             Activity.NodeType,
             hash,
+            payloadJson,
             Activity.Id,
             Id,
-            data ?? new Dictionary<string, object?>(),
             callback?.Method.Name));
     }
 
@@ -112,7 +110,7 @@ public class ActivityExecutionContext
         var locationReference = input.LocationReference;
         var value = await evaluator.EvaluateAsync(input, ExpressionExecutionContext);
         locationReference.Set(this, value);
-        return (T?)value;
+        return value;
     }
 
     private RegisterLocation? GetLocation(RegisterLocationReference locationReference) =>

src/core/Elsa.Core/Models/Bookmark.cs

@@ -4,8 +4,8 @@ public record Bookmark(
     string Id,
     string Name,
     string? Hash,
+    string? Payload,
     string ActivityId,
     string ActivityInstanceId,
-    IDictionary<string, object?>? Data = default,
     string? CallbackMethodName = default
 );

src/core/Elsa.Core/Services/Hasher.cs

@@ -7,24 +7,25 @@ namespace Elsa.Services;
 
 public class Hasher : IHasher
 {
+    private readonly IPayloadSerializer _payloadSerializer;
+
+    public Hasher(IPayloadSerializer payloadSerializer)
+    {
+        _payloadSerializer = payloadSerializer;
+    }
+    
     public string Hash(object value)
     {
-        var options = new JsonSerializerOptions
-        {
-            // Enables serialization of ValueTuples, which use fields instead of properties. 
-            IncludeFields = true
-        };
-            
-        var json = JsonSerializer.Serialize(value, options);
+        var json = _payloadSerializer.Serialize(value);
         return Hash(json);
     }
-        
-    private static string Hash(string input)
+    
+    public string Hash(string value)
     {
         using var sha = HashAlgorithm.Create(HashAlgorithmName.SHA256.ToString())!;
-        return Hash(sha, input);
+        return Hash(sha, value);
     }
-
+    
     private static string Hash(HashAlgorithm hashAlgorithm, string input)
     {
         var data = hashAlgorithm.ComputeHash(Encoding.UTF8.GetBytes(input));

src/core/Elsa.Core/Services/PayloadSerializer.cs

@@ -0,0 +1,21 @@
+using System.Text.Json;
+using Elsa.Contracts;
+
+namespace Elsa.Services;
+
+public class PayloadSerializer : IPayloadSerializer
+{
+    private readonly JsonSerializerOptions _settings;
+
+    public PayloadSerializer()
+    {
+        _settings = new JsonSerializerOptions
+        {
+            // Enables serialization of ValueTuples, which use fields instead of properties.
+            IncludeFields = true
+        };
+    }
+
+    public T Deserialize<T>(string json) where T : notnull => JsonSerializer.Deserialize<T>(json, _settings)!;
+    public string Serialize<T>(T payload) where T : notnull => JsonSerializer.Serialize(payload, payload.GetType(), _settings);
+}

src/core/Elsa/Extensions/ServiceCollectionExtensions.cs

@@ -43,6 +43,7 @@ public static class ServiceCollectionExtensions
             .AddSingleton<IHasher, Hasher>()
             .AddSingleton<IIdentityGenerator, IdentityGenerator>()
             .AddSingleton<ISystemClock, SystemClock>()
+            .AddSingleton<IPayloadSerializer, PayloadSerializer>()
 
             // DSL.
             .AddDsl()

src/modules/Elsa.Modules.Quartz/Services/QuartzJobScheduler.cs

@@ -3,6 +3,7 @@ using Elsa.Modules.Quartz.Contracts;
 using Elsa.Modules.Quartz.Jobs;
 using Microsoft.Extensions.Logging;
 using Quartz;
+using Quartz.Impl.Matchers;
 using IElsaJobScheduler = Elsa.Activities.Scheduling.Contracts.IJobScheduler;
 using IElsaJob = Elsa.Activities.Scheduling.Contracts.IJob;
 using IElsaSchedule = Elsa.Activities.Scheduling.Contracts.ISchedule;
@@ -12,6 +13,7 @@ namespace Elsa.Modules.Quartz.Services;
 public class QuartzJobScheduler : IElsaJobScheduler
 {
     public const string JobDataKey = "ElsaJob";
+    public const string GroupKey = "ElsaJobs";
     private readonly IElsaJobSerializer _elsaJobSerializer;
     private readonly ISchedulerFactory _schedulerFactory;
     private readonly ILogger _logger;
@@ -28,7 +30,14 @@ public class QuartzJobScheduler : IElsaJobScheduler
         var quartzTrigger = CreateTrigger(job, schedule);
         await ScheduleJob(quartzTrigger, cancellationToken);
     }
-    
+
+    public async Task ClearAsync(CancellationToken cancellationToken = default)
+    {
+        var scheduler = await _schedulerFactory.GetScheduler(cancellationToken);
+        var jobKeys = await scheduler.GetJobKeys(GroupMatcher<JobKey>.GroupEquals(GroupKey), cancellationToken);
+        await scheduler.DeleteJobs(jobKeys, cancellationToken);
+    }
+
     private async Task ScheduleJob(ITrigger trigger, CancellationToken cancellationToken)
     {
         var scheduler = await _schedulerFactory.GetScheduler(cancellationToken);
@@ -47,8 +56,9 @@ public class QuartzJobScheduler : IElsaJobScheduler
     private ITrigger CreateTrigger(IElsaJob job, IElsaSchedule schedule)
     {
         var jobName = job.GetType().Name;
+        var triggerKey = new TriggerKey(job.JobId, GroupKey);
         var json = _elsaJobSerializer.Serialize(job);
-        var builder = TriggerBuilder.Create().ForJob(jobName).WithIdentity(job.JobId).UsingJobData(JobDataKey, json);
+        var builder = TriggerBuilder.Create().ForJob(jobName).WithIdentity(triggerKey).UsingJobData(JobDataKey, json);
 
         switch (schedule)
         {

src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs

@@ -3,4 +3,4 @@ using Elsa.Persistence.Entities;
 
 namespace Elsa.Persistence.Commands;
 
-public record ReplaceWorkflowTriggers(string WorkflowId, IEnumerable<WorkflowTrigger> WorkflowTriggers) : ICommand;
+public record ReplaceWorkflowTriggers(ICollection<WorkflowTrigger> WorkflowTriggers) : ICommand;

src/persistence/Elsa.Persistence.Abstractions/Elsa.Persistence.Abstractions.csproj

@@ -15,5 +15,9 @@
     <ItemGroup>
       <PackageReference Include="LinqKit" Version="1.1.27" />
     </ItemGroup>
+
+    <ItemGroup>
+      <Folder Include="Services" />
+    </ItemGroup>
     
 </Project>

src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs

@@ -4,11 +4,11 @@ public class WorkflowBookmark : Entity
 {
     public string Name { get; init; } = default!;
     public string? Hash { get; set; }
+    public string? Payload { get; set; }
     public string WorkflowDefinitionId { get; init; } = default!;
     public string WorkflowInstanceId { get; init; } = default!;
     public string CorrelationId { get; init; } = default!;
     public string ActivityId { get; init; } = default!;
     public string ActivityInstanceId { get; init; } = default!;
-    public IDictionary<string, object?>? Data { get; set; }
     public string? CallbackMethodName { get; set; }
 }

src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowTriggerExtensions.cs

@@ -0,0 +1,14 @@
+using Elsa.Contracts;
+using Elsa.Helpers;
+using Elsa.Persistence.Entities;
+
+namespace Elsa.Persistence.Extensions;
+
+public static class WorkflowTriggerExtensions
+{
+    public static IEnumerable<WorkflowTrigger> Filter<T>(this IEnumerable<WorkflowTrigger> triggers) where T : ITrigger
+    {
+        var triggerName = TypeNameHelper.GenerateTypeName<T>();
+        return triggers.Where(x => x.Name == triggerName);
+    }
+}

src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs

@@ -22,6 +22,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
     private readonly IRequestSender _requestSender;
     private readonly ICommandSender _commandSender;
     private readonly IWorkflowStateSerializer _workflowStateSerializer;
+    private readonly IPayloadSerializer _payloadSerializer;
     private readonly IIdentityGenerator _identityGenerator;
     private readonly ISystemClock _clock;
 
@@ -30,6 +31,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
         IRequestSender requestSender,
         ICommandSender commandSender,
         IWorkflowStateSerializer workflowStateSerializer,
+        IPayloadSerializer payloadSerializer,
         IIdentityGenerator identityGenerator,
         ISystemClock clock)
     {
@@ -37,6 +39,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
         _requestSender = requestSender;
         _commandSender = commandSender;
         _workflowStateSerializer = workflowStateSerializer;
+        _payloadSerializer = payloadSerializer;
         _identityGenerator = identityGenerator;
         _clock = clock;
     }
@@ -47,7 +50,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
         var workflow = context.Workflow;
         var (definitionId, version, definitionVersionId) = workflow.Identity;
         var existingWorkflowInstance = await _requestSender.RequestAsync(new FindWorkflowInstance(context.Id), cancellationToken);
-        
+
         // Setup a new workflow instance if no existing instance was found.
         var workflowInstance = existingWorkflowInstance ?? new WorkflowInstance
         {
@@ -63,7 +66,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
 
         // Get a copy of current bookmarks.
         var existingBookmarks = await _requestSender.RequestAsync(new FindWorkflowBookmarks(workflowInstance.Id), cancellationToken);
-        var bookmarksSnapshot = existingBookmarks.Select(x => new Bookmark(x.Id, x.Name, x.Hash, x.ActivityId, x.ActivityInstanceId, x.Data, x.CallbackMethodName)).ToList();
+        var bookmarksSnapshot = existingBookmarks.Select(x => new Bookmark(x.Id, x.Name, x.Hash, x.Payload, x.ActivityId, x.ActivityInstanceId, x.CallbackMethodName)).ToList();
 
         // Exclude the current bookmark that initiated the creation of the workflow context, if any.
         var invokedBookmark = context.Bookmark;
@@ -103,7 +106,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware
             WorkflowInstanceId = workflowInstance.Id,
             CorrelationId = workflowInstance.CorrelationId,
             Hash = x.Hash,
-            Data = x.Data,
+            Payload = x.Payload,
             Name = x.Name,
             ActivityId = x.ActivityId,
             ActivityInstanceId = x.ActivityInstanceId,

src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.Designer.cs

@@ -11,7 +11,7 @@ using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
 namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
 {
     [DbContext(typeof(ElsaDbContext))]
-    [Migration("20220114122659_Initial")]
+    [Migration("20220117205849_Initial")]
     partial class Initial
     {
         protected override void BuildTargetModel(ModelBuilder modelBuilder)
@@ -42,13 +42,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
                     b.Property<string>("Hash")
                         .HasColumnType("TEXT");
 
-                    b.Property<string>("JsonData")
-                        .HasColumnType("TEXT");
-
                     b.Property<string>("Name")
                         .IsRequired()
                         .HasColumnType("TEXT");
 
+                    b.Property<string>("Payload")
+                        .HasColumnType("TEXT");
+
                     b.Property<string>("WorkflowDefinitionId")
                         .IsRequired()
                         .HasColumnType("TEXT");

src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.cs

@@ -16,13 +16,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
                     Id = table.Column<string>(type: "TEXT", nullable: false),
                     Name = table.Column<string>(type: "TEXT", nullable: false),
                     Hash = table.Column<string>(type: "TEXT", nullable: true),
+                    Payload = table.Column<string>(type: "TEXT", nullable: true),
                     WorkflowDefinitionId = table.Column<string>(type: "TEXT", nullable: false),
                     WorkflowInstanceId = table.Column<string>(type: "TEXT", nullable: false),
                     CorrelationId = table.Column<string>(type: "TEXT", nullable: false),
                     ActivityId = table.Column<string>(type: "TEXT", nullable: false),
                     ActivityInstanceId = table.Column<string>(type: "TEXT", nullable: false),
-                    CallbackMethodName = table.Column<string>(type: "TEXT", nullable: true),
-                    JsonData = table.Column<string>(type: "TEXT", nullable: true)
+                    CallbackMethodName = table.Column<string>(type: "TEXT", nullable: true)
                 },
                 constraints: table =>
                 {

src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs

@@ -40,13 +40,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
                     b.Property<string>("Hash")
                         .HasColumnType("TEXT");
 
-                    b.Property<string>("JsonData")
-                        .HasColumnType("TEXT");
-
                     b.Property<string>("Name")
                         .IsRequired()
                         .HasColumnType("TEXT");
 
+                    b.Property<string>("Payload")
+                        .HasColumnType("TEXT");
+
                     b.Property<string>("WorkflowDefinitionId")
                         .IsRequired()
                         .HasColumnType("TEXT");

src/persistence/Elsa.Persistence.EntityFrameworkCore/Configuration/WorkflowBookmarkConfiguration.cs

@@ -8,9 +8,6 @@ namespace Elsa.Persistence.EntityFrameworkCore.Configuration
     {
         public void Configure(EntityTypeBuilder<WorkflowBookmark> builder)
         {
-            builder.Ignore(x => x.Data);
-            builder.Property<string>("JsonData");
-            
             builder.HasIndex(x => x.Name).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.Name)}");
             builder.HasIndex(x => x.Hash).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.Hash)}");
             builder.HasIndex(x => x.ActivityId).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.ActivityId)}");

src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs

@@ -12,7 +12,8 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler<ReplaceWorkflowTri
 
     public async Task<Unit> HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken)
     {
-        await _store.DeleteWhereAsync(x => x.WorkflowDefinitionId == command.WorkflowId, cancellationToken);
+        var ids = command.WorkflowTriggers.Select(x => x.Id).ToList();
+        await _store.DeleteWhereAsync(x => ids.Contains(x.Id), cancellationToken);
         await _store.SaveManyAsync(command.WorkflowTriggers, cancellationToken);
 
         return Unit.Instance;

src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs

@@ -1,3 +1,4 @@
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Elsa.Mediator.Contracts;
@@ -18,9 +19,10 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler<ReplaceWorkflowTri
 
     public Task<Unit> HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken)
     {
-        _store.DeleteWhere(x => x.WorkflowDefinitionId == command.WorkflowId);
+        var ids = command.WorkflowTriggers.Select(x => x.Id).ToList();
+        _store.DeleteWhere(x => ids.Contains(x.Id));
         _store.SaveMany(command.WorkflowTriggers);
-        
+
         return Task.FromResult(Unit.Instance);
     }
 }

src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs

@@ -65,20 +65,15 @@ public class WorkflowOperatorActor : IActor
 
     private ExecuteWorkflowResponse MapResult(ExecuteWorkflowResult result)
     {
-        var bookmarks = result.Bookmarks.Select(x =>
+        var bookmarks = result.Bookmarks.Select(x => new Bookmark
         {
-            var data = x.Data?.ToDictionary(y => y.Key, y => new Json
-            {
-                Text = JsonSerializer.Serialize(y.Value)
-            }) ?? new Dictionary<string, Json>();
-
-            var bookmark = new Bookmark
-            {
-                Id = x.Id
-            };
-
-            bookmark.Data.Add(data);
-            return bookmark;
+            Id = x.Id,
+            Hash = x.Hash,
+            Payload = x.Payload,
+            Name = x.Name,
+            ActivityId = x.ActivityId,
+            ActivityInstanceId = x.ActivityInstanceId,
+            CallbackMethodName = x.CallbackMethodName
         });
 
         var response = new ExecuteWorkflowResponse
@@ -104,9 +99,9 @@ public class WorkflowOperatorActor : IActor
                 bookmarkMessage.Id,
                 bookmarkMessage.Name,
                 bookmarkMessage.Hash,
+                bookmarkMessage.Payload,
                 bookmarkMessage.ActivityId,
                 bookmarkMessage.ActivityInstanceId,
-                null,
                 bookmarkMessage.CallbackMethodName);
 
         return await _workflowEngine.ExecuteAsync(workflow, workflowState, bookmark, cancellationToken);

src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs

@@ -13,7 +13,7 @@ namespace Elsa.Runtime.ProtoActor.Services;
 
 using Bookmark = Elsa.Models.Bookmark;
 using BookmarkMessage = Messages.Bookmark;
-    
+
 public class ProtoActorWorkflowInvoker : IWorkflowInvoker
 {
     private readonly Cluster _cluster;
@@ -56,7 +56,7 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
 
         return new ExecuteWorkflowResult(workflowState, bookmarks);
     }
-        
+
     public async Task<DispatchWorkflowResult> DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
     {
         var (instanceId, bookmark) = request;
@@ -92,7 +92,7 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
 
         return new ExecuteWorkflowResult(workflowState, bookmarks);
     }
-        
+
     private BookmarkMessage? MapBookmark(Bookmark? bookmark)
     {
         if (bookmark == null)
@@ -103,10 +103,10 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
             Id = bookmark.Id,
             Name = bookmark.Name,
             Hash = bookmark.Hash,
+            Payload = bookmark.Payload,
             ActivityId = bookmark.ActivityId,
             ActivityInstanceId = bookmark.ActivityInstanceId,
             CallbackMethodName = bookmark.CallbackMethodName,
-            Data = { }
         };
     }
 
@@ -116,9 +116,9 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker
             bookmarkMessage.Id,
             bookmarkMessage.Name,
             bookmarkMessage.Hash,
+            bookmarkMessage.Payload,
             bookmarkMessage.ActivityId,
             bookmarkMessage.ActivityInstanceId,
-            bookmarkMessage.Data.ToDictionary(x => x.Key, x => JsonSerializer.Deserialize<object>(x.Value.Text)),
             bookmarkMessage.CallbackMethodName
         );
     }

src/runtime/Elsa.Runtime.ProtoActor/messages.proto

@@ -43,9 +43,9 @@ message Bookmark {
   string id = 1;
   string name = 2;
   optional string hash = 3;
-  string activityId = 4;
-  string activityInstanceId = 5;
-  map<string, Json> data = 6;
+  optional string payload = 4;
+  string activityId = 5;
+  string activityInstanceId = 6;
   optional string callbackMethodName = 7;
 }
 

src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs

@@ -1,4 +1,5 @@
 using Elsa.Models;
+using Elsa.Persistence.Entities;
 
 namespace Elsa.Runtime.Contracts;
 
@@ -12,5 +13,5 @@ public interface ITriggerIndexer
     /// <summary>
     /// Indexes triggers of the specified workflow.
     /// </summary>
-    Task IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default);
+    Task<IEnumerable<WorkflowTrigger>> IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default);
 }

src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs

@@ -50,7 +50,7 @@ public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpret
         }
 
         // Resume workflow instance.
-        var bookmark = new Bookmark(workflowBookmark.Id, workflowBookmark.Name, workflowBookmark.Hash, workflowBookmark.ActivityId, workflowBookmark.ActivityInstanceId, workflowBookmark.Data, workflowBookmark.CallbackMethodName);
+        var bookmark = new Bookmark(workflowBookmark.Id, workflowBookmark.Name, workflowBookmark.Hash, workflowBookmark.Payload, workflowBookmark.ActivityId, workflowBookmark.ActivityInstanceId, workflowBookmark.CallbackMethodName);
         var workflowState = workflowInstance.WorkflowState;
         var workflowExecutionResult = await _workflowEngine.ExecuteAsync(workflow, workflowState, bookmark, cancellationToken);
 

src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs

@@ -1,7 +1,6 @@
 using Elsa.Mediator.Contracts;
+using Elsa.Persistence.Entities;
 
 namespace Elsa.Runtime.Notifications;
 
-public class TriggerIndexingFinished : INotification
-{
-}
+public record TriggerIndexingFinished(ICollection<WorkflowTrigger> Triggers) : INotification;

src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs

@@ -58,24 +58,37 @@ public class TriggerIndexer : ITriggerIndexer
         _logger.LogInformation("Indexing workflow triggers");
         stopwatch.Start();
 
+        // Only stream workflows from providers that are not "dynamic" (such as DatabaseWorkflowProvider).
         var workflows = _workflowRegistry.StreamAllAsync(WorkflowRegistry.SkipDynamicProviders, cancellationToken);
+        var collectedTriggers = new List<WorkflowTrigger>();
 
         await foreach (var workflow in workflows.WithCancellation(cancellationToken))
-            await IndexTriggersAsync(workflow, cancellationToken);
+        {
+            var triggers = await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken);
+            collectedTriggers.AddRange(triggers);
+        }
+
+        // Replace triggers for the specified workflow.
+        await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(collectedTriggers), cancellationToken);
 
         stopwatch.Stop();
         _logger.LogInformation("Finished indexing workflow triggers in {ElapsedTime}", stopwatch.Elapsed);
-        await _eventPublisher.PublishAsync(new TriggerIndexingFinished(), cancellationToken);
+
+        // Publish event.
+        await _eventPublisher.PublishAsync(new TriggerIndexingFinished(collectedTriggers), cancellationToken);
     }
 
-    public async Task IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default)
+    public async Task<IEnumerable<WorkflowTrigger>> IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default)
     {
         // Collect new triggers.
         var triggers = await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken);
 
         // Replace triggers for the specified workflow.
-        var definitionId = workflow.Identity.DefinitionId;
-        await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(definitionId, triggers), cancellationToken);
+        await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(triggers), cancellationToken);
+
+        // Publish event.
+        await _eventPublisher.PublishAsync(new TriggerIndexingFinished(triggers), cancellationToken);
+        return triggers;
     }
 
     private async IAsyncEnumerable<WorkflowTrigger> GetTriggersAsync(Workflow workflow, [EnumeratorCancellation] CancellationToken cancellationToken = default)
@@ -132,11 +145,11 @@ public class TriggerIndexer : ITriggerIndexer
         return triggers;
     }
 
-    private async Task<IEnumerable<object>> TryGetPayloadsAsync(ITrigger trigger, TriggerIndexingContext context, CancellationToken cancellationToken)
+    private async Task<ICollection<object>> TryGetPayloadsAsync(ITrigger trigger, TriggerIndexingContext context, CancellationToken cancellationToken)
     {
         try
         {
-            return await trigger.GetPayloadsAsync(context, cancellationToken);
+            return (await trigger.GetPayloadsAsync(context, cancellationToken)).ToList();
         }
         catch (Exception e)
         {