diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs new file mode 100644 index 000000000..57313ae6c --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkPersister.cs @@ -0,0 +1,16 @@ +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Helpers; +using Elsa.Workflows.Core.Models; + +namespace Elsa.Workflows.Runtime.Contracts; + +/// +/// Persists bookmarks and raises events. +/// +public interface IBookmarksPersister +{ + /// + /// Persists bookmarks and raises events. + /// + Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff diff); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkUpdater.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkUpdater.cs new file mode 100644 index 000000000..3ec2bfa85 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IBookmarkUpdater.cs @@ -0,0 +1,8 @@ +using Elsa.Workflows.Runtime.Requests; + +namespace Elsa.Workflows.Runtime.Contracts; + +public interface IBookmarkUpdater +{ + Task UpdateBookmarksAsync(UpdateBookmarksRequest request, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs new file mode 100644 index 000000000..60fbbd78c --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs @@ -0,0 +1,36 @@ +using Elsa.Mediator; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Core; +using Elsa.Workflows.Core.Helpers; +using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Notifications; +using Elsa.Workflows.Runtime.Requests; + +namespace Elsa.Workflows.Runtime.Services; + +/// +public class BookmarksPersister(IBookmarkUpdater bookmarkUpdater, INotificationSender notificationSender) : IBookmarksPersister +{ + + + /// + public async Task PersistBookmarksAsync(WorkflowExecutionContext context, Diff diff) + { + var cancellationToken = context.CancellationTokens.SystemCancellationToken; + var updateBookmarksContext = new UpdateBookmarksRequest(context.Id, diff, context.CorrelationId); + await bookmarkUpdater.UpdateBookmarksAsync(updateBookmarksContext, cancellationToken); + + // Publish domain event. + await notificationSender.SendAsync(new WorkflowBookmarksIndexed(context, new IndexedWorkflowBookmarks(context.Id, diff.Added, diff.Removed, diff.Unchanged)), cancellationToken); + + // Notify all interested activities that the bookmarks have been persisted. + var activityExecutionContexts = context.ActivityExecutionContexts.Where(x => x.Activity is IBookmarksPersistedHandler && x.Bookmarks.Any()).ToList(); + + foreach (var activityExecutionContext in activityExecutionContexts) + await ((IBookmarksPersistedHandler)activityExecutionContext.Activity).BookmarksPersistedAsync(activityExecutionContext); + + // Publish domain event. + await notificationSender.SendAsync(new WorkflowBookmarksPersisted(context, diff), NotificationStrategy.Background, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs new file mode 100644 index 000000000..edf077b45 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs @@ -0,0 +1,35 @@ +using Elsa.Workflows.Core.Models; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Entities; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Requests; + +namespace Elsa.Workflows.Runtime.Services; + +/// +public class BookmarkUpdater(IBookmarkManager bookmarkManager, IBookmarkStore bookmarkStore) : IBookmarkUpdater +{ + /// + public async Task UpdateBookmarksAsync(UpdateBookmarksRequest request, CancellationToken cancellationToken = default) + { + var instanceId = request.WorkflowInstanceId; + await RemoveBookmarksAsync(instanceId, request.Diff.Removed, cancellationToken); + await StoreBookmarksAsync(instanceId, request.Diff.Added, request.CorrelationId, cancellationToken); + } + + private async Task RemoveBookmarksAsync(string workflowInstanceId, IEnumerable bookmarks, CancellationToken cancellationToken) + { + var matchingHashes = bookmarks.Select(x => x.Hash).ToList(); + var filter = new BookmarkFilter { Hashes = matchingHashes, WorkflowInstanceId = workflowInstanceId }; + await bookmarkManager.DeleteManyAsync(filter, cancellationToken); + } + + private async Task StoreBookmarksAsync(string workflowInstanceId, IEnumerable bookmarks, string? correlationId, CancellationToken cancellationToken) + { + foreach (var bookmark in bookmarks) + { + var storedBookmark = new StoredBookmark(bookmark.Id, bookmark.Name, bookmark.Hash, workflowInstanceId, bookmark.CreatedAt, bookmark.ActivityInstanceId, correlationId, bookmark.Payload); + await bookmarkStore.SaveAsync(storedBookmark, cancellationToken); + } + } +} \ No newline at end of file