Retention Module (#5344)

* Retention module

* Remove unused batch size

* Use ISystemClock instead of DateTime.Now

* Remove TargetFramework

* Use WorkflowInstanceManager

* Update using statement, remove cloneable from WorkflowInstanceFilter.cs

* Update project description and remove unnecessary properties

* Move to BackgroundService

* Generic retention module

---------

Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>

src/modules/Elsa.Retention/CleanupStrategies/DeleteActivityExecutionRecordStrategy.cs

@@ -0,0 +1,37 @@
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Retention.CleanupStrategies;
+
+/// <summary>
+///     Deletes activity execution records.
+/// </summary>
+public class DeleteActivityExecutionRecordStrategy : IDeletionCleanupStrategy<ActivityExecutionRecord>
+{
+    private readonly ILogger<DeleteActivityExecutionRecordStrategy> _logger;
+    private readonly IActivityExecutionStore _store;
+
+    public DeleteActivityExecutionRecordStrategy(IActivityExecutionStore store, ILogger<DeleteActivityExecutionRecordStrategy> logger)
+    {
+        _store = store;
+        _logger = logger;
+    }
+
+    public async Task Cleanup(ICollection<ActivityExecutionRecord> collection)
+    {
+        ActivityExecutionRecordFilter filter = new()
+        {
+            Ids = collection.Select(x => x.Id).ToList()
+        };
+
+        long deletedRecords = await _store.DeleteManyAsync(filter);
+
+        if (deletedRecords != collection.Count)
+        {
+            _logger.LogWarning("Expected to delete {Expected} activity execution records, actually deleted {Actual} activity execution records", collection.Count, deletedRecords);
+        }
+    }
+}

src/modules/Elsa.Retention/CleanupStrategies/DeleteBookmarkStrategy.cs

@@ -0,0 +1,37 @@
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Retention.CleanupStrategies;
+
+/// <summary>
+///     Deletes a collection of bookmarks
+/// </summary>
+public class DeleteBookmarkStrategy : IDeletionCleanupStrategy<StoredBookmark>
+{
+    private readonly ILogger<DeleteBookmarkStrategy> _logger;
+    private readonly IBookmarkStore _store;
+
+    public DeleteBookmarkStrategy(IBookmarkStore store, ILogger<DeleteBookmarkStrategy> logger)
+    {
+        _store = store;
+        _logger = logger;
+    }
+
+    public async Task Cleanup(ICollection<StoredBookmark> collection)
+    {
+        BookmarkFilter bookmarkFilter = new()
+        {
+            BookmarkIds = collection.Select(x => x.Id).ToList()
+        };
+
+        long deletedRecords = await _store.DeleteAsync(bookmarkFilter);
+
+        if (deletedRecords != collection.Count)
+        {
+            _logger.LogWarning("Expected to delete {Expected} bookmarks, actually deleted {Actual} bookmarks", collection.Count, deletedRecords);
+        }
+    }
+}

src/modules/Elsa.Retention/CleanupStrategies/DeleteWorkflowExecutionRecordStrategy.cs

@@ -0,0 +1,37 @@
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+using Microsoft.Extensions.Logging;
+
+namespace Elsa.Retention.CleanupStrategies;
+
+/// <summary>
+///     Deletes <see cref="WorkflowExecutionLogRecord" />
+/// </summary>
+public class DeleteWorkflowExecutionRecordStrategy : IDeletionCleanupStrategy<WorkflowExecutionLogRecord>
+{
+    private readonly ILogger<DeleteWorkflowExecutionRecordStrategy> _logger;
+    private readonly IWorkflowExecutionLogStore _store;
+
+    public DeleteWorkflowExecutionRecordStrategy(IWorkflowExecutionLogStore store, ILogger<DeleteWorkflowExecutionRecordStrategy> logger)
+    {
+        _store = store;
+        _logger = logger;
+    }
+
+    public async Task Cleanup(ICollection<WorkflowExecutionLogRecord> collection)
+    {
+        WorkflowExecutionLogRecordFilter filter = new()
+        {
+            Ids = collection.Select(x => x.Id).ToList()
+        };
+
+        long deletedRecords = await _store.DeleteManyAsync(filter);
+
+        if (deletedRecords != collection.Count)
+        {
+            _logger.LogWarning("Expected to delete {Expected} workflow execution records, actually deleted {Actual} workflow execution records", collection.Count, deletedRecords);
+        }
+    }
+}

src/modules/Elsa.Retention/Collectors/ActivityExecutionRecordCollector.cs

@@ -0,0 +1,36 @@
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+
+namespace Elsa.Retention.Collectors;
+
+/// <summary>
+///     Collects all <see cref="ActivityExecutionRecord" /> related to the <see cref="WorkflowInstance" />
+/// </summary>
+public class ActivityExecutionRecordCollector : IRelatedEntityCollector<ActivityExecutionRecord>
+{
+    private readonly IActivityExecutionStore _store;
+
+    public ActivityExecutionRecordCollector(IActivityExecutionStore store)
+    {
+        _store = store;
+    }
+
+    public async IAsyncEnumerable<ICollection<ActivityExecutionRecord>> GetRelatedEntities(ICollection<WorkflowInstance> workflowInstances)
+    {
+        IEnumerable<WorkflowInstance[]> chunks = workflowInstances.Chunk(5);
+
+        foreach (WorkflowInstance[] chunk in chunks)
+        {
+            ActivityExecutionRecordFilter filter = new()
+            {
+                WorkflowInstanceIds = chunk.Select(x => x.Id).ToArray()
+            };
+
+            IEnumerable<ActivityExecutionRecord> records = await _store.FindManyAsync(filter);
+            yield return records.ToArray();
+        }
+    }
+}

src/modules/Elsa.Retention/Collectors/BookmarkCollector.cs

@@ -0,0 +1,36 @@
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+
+namespace Elsa.Retention.Collectors;
+
+/// <summary>
+///     Collects all <see cref="StoredBookmark" /> related to the <see cref="WorkflowInstance" />
+/// </summary>
+public class BookmarkCollector : IRelatedEntityCollector<StoredBookmark>
+{
+    private readonly IBookmarkStore _store;
+
+    public BookmarkCollector(IBookmarkStore store)
+    {
+        _store = store;
+    }
+
+    public async IAsyncEnumerable<ICollection<StoredBookmark>> GetRelatedEntities(ICollection<WorkflowInstance> workflowInstances)
+    {
+        IEnumerable<WorkflowInstance[]> batches = workflowInstances.Chunk(25);
+
+        foreach (WorkflowInstance[] batch in batches)
+        {
+            BookmarkFilter filter = new()
+            {
+                WorkflowInstanceIds = batch.Select(x => x.Id).ToArray()
+            };
+
+            IEnumerable<StoredBookmark> bookmarks = await _store.FindManyAsync(filter);
+            yield return bookmarks.ToArray();
+        }
+    }
+}

src/modules/Elsa.Retention/Collectors/WorkflowExecutionLogRecordCollector.cs

@@ -0,0 +1,49 @@
+using Elsa.Common.Models;
+using Elsa.Retention.Contracts;
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Filters;
+
+namespace Elsa.Retention.Collectors;
+
+/// <summary>
+///     Collects all <see cref="WorkflowExecutionLogRecord" /> related to the <see cref="WorkflowInstance" />
+/// </summary>
+public class WorkflowExecutionLogRecordCollector : IRelatedEntityCollector<WorkflowExecutionLogRecord>
+{
+    private readonly IWorkflowExecutionLogStore _store;
+
+    public WorkflowExecutionLogRecordCollector(IWorkflowExecutionLogStore store)
+    {
+        _store = store;
+    }
+
+    public async IAsyncEnumerable<ICollection<WorkflowExecutionLogRecord>> GetRelatedEntities(ICollection<WorkflowInstance> workflowInstances)
+    {
+        IEnumerable<WorkflowInstance[]> chunks = workflowInstances.Chunk(25);
+
+        foreach (WorkflowInstance[] chunk in chunks)
+        {
+            WorkflowExecutionLogRecordFilter filter = new()
+            {
+                WorkflowInstanceIds = chunk.Select(x => x.Id).ToArray()
+            };
+
+            PageArgs pageArgs = PageArgs.FromPage(0, 100);
+
+            while (true)
+            {
+                Page<WorkflowExecutionLogRecord> page = await _store.FindManyAsync(filter, pageArgs);
+                yield return page.Items.ToArray();
+
+                if (page.TotalCount <= pageArgs.Offset + page.Items.Count)
+                {
+                    break;
+                }
+
+                pageArgs.Next();
+            }
+        }
+    }
+}

src/modules/Elsa.Retention/Contracts/ICleanupStrategy.cs

@@ -0,0 +1,30 @@
+namespace Elsa.Retention.Contracts;
+
+public interface ICleanupStrategy
+{
+    Task Cleanup(ICollection<object> collection);
+}
+
+/// <summary>
+///     A strategy responsible for cleaning up the <see cref="TEntity" />
+/// </summary>
+/// <typeparam name="TEntity"></typeparam>
+public interface ICleanupStrategy<TEntity> : ICleanupStrategy
+{
+    /// <summary>
+    ///     Cleans up the given entity
+    /// </summary>
+    /// <param name="collection"></param>
+    /// <returns></returns>
+    Task ICleanupStrategy.Cleanup(ICollection<object> collection)
+    {
+        return Cleanup(collection.Cast<TEntity>().ToList());
+    }
+
+    /// <summary>
+    ///     Cleans up the given entities
+    /// </summary>
+    /// <param name="collection"></param>
+    /// <returns></returns>
+    Task Cleanup(ICollection<TEntity> collection);
+}

src/modules/Elsa.Retention/Contracts/IDeletionCleanupStrategy.cs

@@ -0,0 +1,9 @@
+namespace Elsa.Retention.Contracts;
+
+/// <summary>
+///     A strategy responsible for deleting the given entities
+/// </summary>
+/// <typeparam name="TEntity"></typeparam>
+public interface IDeletionCleanupStrategy<TEntity> : ICleanupStrategy<TEntity>
+{
+}

src/modules/Elsa.Retention/Contracts/IRelatedEntityCollector.cs

@@ -0,0 +1,33 @@
+using Elsa.Workflows.Management.Entities;
+
+namespace Elsa.Retention.Contracts;
+
+/// <summary>
+///     A generic variant of <see cref="IRelatedEntityCollector{TEntity}" />
+/// </summary>
+public interface IRelatedEntityCollector
+{
+    IAsyncEnumerable<ICollection<object>> GetRelatedEntitiesGeneric(ICollection<WorkflowInstance> workflowInstances);
+}
+
+/// <summary>
+///     Collects <see cref="TEntity" /> that are related to the workflow instance
+/// </summary>
+/// <typeparam name="TEntity"></typeparam>
+public interface IRelatedEntityCollector<TEntity> : IRelatedEntityCollector where TEntity : class
+{
+    async IAsyncEnumerable<ICollection<object>> IRelatedEntityCollector.GetRelatedEntitiesGeneric(ICollection<WorkflowInstance> workflowInstances)
+    {
+        await foreach (ICollection<TEntity> entity in GetRelatedEntities(workflowInstances).ConfigureAwait(false))
+        {
+            yield return entity.Select(x => (object)x).ToArray();
+        }
+    }
+
+    /// <summary>
+    ///     Collects the entities related to the given workflow instances
+    /// </summary>
+    /// <param name="workflowInstances"></param>
+    /// <returns></returns>
+    IAsyncEnumerable<ICollection<TEntity>> GetRelatedEntities(ICollection<WorkflowInstance> workflowInstances);
+}

src/modules/Elsa.Retention/Contracts/IRetentionPolicy.cs

@@ -0,0 +1,24 @@
+using Elsa.Retention.Models;
+
+namespace Elsa.Retention.Contracts;
+
+/// <summary>
+///     Defines which workflows should be retained and how they are retained
+/// </summary>
+public interface IRetentionPolicy
+{
+    /// <summary>
+    ///     The name of this policy
+    /// </summary>
+    string Name { get; }
+
+    /// <summary>
+    ///     The workflow instance filter
+    /// </summary>
+    Func<IServiceProvider, RetentionWorkflowInstanceFilter> FilterFactory { get; }
+
+    /// <summary>
+    ///     A marker type for which <see cref="ICleanupStrategy{TEntity}" /> has to be used
+    /// </summary>
+    Type CleanupStrategy { get; }
+}

src/modules/Elsa.Retention/Elsa.Retention.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <Description>
+            Provides retention options for workflows.
+        </Description>
+        <PackageTags>elsa module retention archive</PackageTags>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\Elsa.Workflows.Management\Elsa.Workflows.Management.csproj"/>
+        <ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj"/>
+    </ItemGroup>
+</Project>

src/modules/Elsa.Retention/Extensions/ModuleExtensions.cs

@@ -0,0 +1,19 @@
+using Elsa.Features.Services;
+using Elsa.Retention.Feature;
+
+namespace Elsa.Retention.Extensions;
+
+/// <summary>
+///     Provides extensions to install the <see cref="RetentionFeature" /> feature.
+/// </summary>
+public static class ModuleExtensions
+{
+    /// <summary>
+    ///     Install the <see cref="RetentionFeature" /> feature.
+    /// </summary>
+    public static IModule UseRetention(this IModule module, Action<RetentionFeature>? configure = default)
+    {
+        module.Configure(configure);
+        return module;
+    }
+}

src/modules/Elsa.Retention/Extensions/RetentionFeatureExtensions.cs

@@ -0,0 +1,36 @@
+using Elsa.Extensions;
+using Elsa.Retention.Contracts;
+using Elsa.Retention.Feature;
+using Elsa.Retention.Models;
+using Elsa.Retention.Policies;
+
+namespace Elsa.Retention.Extensions;
+
+public static class RetentionFeatureExtensions
+{
+    private static readonly object PoliciesKey = new();
+
+    /// <summary>
+    ///     Adds a policy that will delete workflow instance based on the filter
+    /// </summary>
+    /// <param name="feature"></param>
+    /// <param name="name"></param>
+    /// <param name="filterFactory"></param>
+    /// <returns></returns>
+    public static RetentionFeature AddDeletePolicy(this RetentionFeature feature, string name, Func<IServiceProvider, RetentionWorkflowInstanceFilter> filterFactory)
+    {
+        List<IRetentionPolicy> policies = feature.Module.Properties.GetOrAdd(PoliciesKey, () => new List<IRetentionPolicy>());
+        policies.Add(new DeletionRetentionPolicy(name, filterFactory));
+        return feature;
+    }
+
+    /// <summary>
+    ///     Returns the registered policies
+    /// </summary>
+    /// <param name="feature"></param>
+    /// <returns></returns>
+    public static ICollection<IRetentionPolicy> GetPolicies(this RetentionFeature feature)
+    {
+        return feature.Module.Properties.GetOrAdd(PoliciesKey, () => new List<IRetentionPolicy>());
+    }
+}

src/modules/Elsa.Retention/Extensions/WorkflowInstanceFilterExtensions.cs

@@ -0,0 +1,44 @@
+using Elsa.Workflows;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Management.Models;
+
+namespace Elsa.Retention.Extensions;
+
+public static class WorkflowInstanceFilterExtensions
+{
+    /// <summary>
+    ///     Clone the current filter
+    /// </summary>
+    /// <param name="filter"></param>
+    /// <returns></returns>
+    public static WorkflowInstanceFilter Clone(this WorkflowInstanceFilter filter)
+    {
+        return new WorkflowInstanceFilter
+        {
+            Id = filter.Id,
+            Ids = filter.Ids == null ? null : new List<string>(filter.Ids),
+            Version = filter.Version,
+            CorrelationId = filter.CorrelationId,
+            CorrelationIds = filter.CorrelationIds == null ? null : new List<string>(filter.CorrelationIds),
+            DefinitionId = filter.DefinitionId,
+            DefinitionIds = filter.DefinitionIds == null ? null : new List<string>(filter.DefinitionIds),
+            HasIncidents = filter.HasIncidents,
+            IsSystem = filter.IsSystem,
+            SearchTerm = filter.SearchTerm,
+            TimestampFilters = filter.TimestampFilters?.Select(x => new TimestampFilter
+            {
+                Column = x.Column,
+                Operator = x.Operator,
+                Timestamp = x.Timestamp
+            }).ToList(),
+            WorkflowStatus = filter.WorkflowStatus,
+            WorkflowStatuses = filter.WorkflowStatuses == null ? null : new List<WorkflowStatus>(filter.WorkflowStatuses),
+            DefinitionVersionId = filter.DefinitionVersionId,
+            DefinitionVersionIds = filter.DefinitionVersionIds == null ? null : new List<string>(filter.DefinitionVersionIds),
+            WorkflowSubStatus = filter.WorkflowSubStatus,
+            WorkflowSubStatuses = filter.WorkflowSubStatuses == null ? null : new List<WorkflowSubStatus>(filter.WorkflowSubStatuses),
+            ParentWorkflowInstanceIds =
+                filter.ParentWorkflowInstanceIds == null ? null : new List<string>(filter.ParentWorkflowInstanceIds)
+        };
+    }
+}

src/modules/Elsa.Retention/Feature/RetentionFeature.cs

@@ -0,0 +1,58 @@
+using Elsa.Features.Abstractions;
+using Elsa.Features.Services;
+using Elsa.Retention.CleanupStrategies;
+using Elsa.Retention.Collectors;
+using Elsa.Retention.Contracts;
+using Elsa.Retention.Extensions;
+using Elsa.Retention.HostedServices;
+using Elsa.Retention.Jobs;
+using Elsa.Retention.Options;
+using Elsa.Workflows.Runtime.Entities;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.Retention.Feature;
+
+/// <summary>
+///     The retention features provides automated cleanup of workflow instances
+/// </summary>
+public class RetentionFeature : FeatureBase
+{
+    /// <summary>
+    ///     Create the retention feature
+    /// </summary>
+    /// <param name="module"></param>
+    public RetentionFeature(IModule module) : base(module)
+    {
+    }
+
+    /// <summary>
+    ///     Gets or sets a delegate to configure the retention options.
+    /// </summary>
+    public Action<CleanupOptions> ConfigureCleanupOptions { get; set; } = _ => { };
+
+    /// <inheritdoc cref="FeatureBase" />
+    public override void Apply()
+    {
+        Services.Configure(ConfigureCleanupOptions);
+        Services.AddTransient<CleanupJob>();
+
+        Services.AddScoped<IDeletionCleanupStrategy<StoredBookmark>, DeleteBookmarkStrategy>();
+        Services.AddScoped<IDeletionCleanupStrategy<ActivityExecutionRecord>, DeleteActivityExecutionRecordStrategy>();
+        Services.AddScoped<IDeletionCleanupStrategy<WorkflowExecutionLogRecord>, DeleteWorkflowExecutionRecordStrategy>();
+
+        Services.AddScoped<IRelatedEntityCollector, BookmarkCollector>();
+        Services.AddScoped<IRelatedEntityCollector, ActivityExecutionRecordCollector>();
+        Services.AddScoped<IRelatedEntityCollector, WorkflowExecutionLogRecordCollector>();
+
+        foreach (IRetentionPolicy policy in this.GetPolicies())
+        {
+            Services.AddSingleton(policy);
+        }
+    }
+
+    /// <inheritdoc cref="FeatureBase" />
+    public override void ConfigureHostedServices()
+    {
+        ConfigureHostedService<CleanupHostedService>();
+    }
+}

src/modules/Elsa.Retention/FodyWeavers.xml

@@ -0,0 +1,3 @@
+<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
+    <ConfigureAwait/>
+</Weavers>

src/modules/Elsa.Retention/HostedServices/CleanupHostedService.cs

@@ -0,0 +1,55 @@
+using Elsa.Retention.Jobs;
+using Elsa.Retention.Options;
+using Medallion.Threading;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Retention.HostedServices;
+
+/// <summary>
+///     Periodically wipes workflow instances and their execution logs.
+/// </summary>
+public class CleanupHostedService : BackgroundService
+{
+    private readonly TimeSpan _interval;
+    private readonly ILogger<CleanupHostedService> _logger;
+    private readonly IServiceScopeFactory _serviceScopeFactory;
+
+    /// <summary>
+    ///     Creates new Cleanup hosted service
+    /// </summary>
+    /// <param name="options"></param>
+    /// <param name="serviceScopeFactory"></param>
+    /// <param name="logger"></param>
+    public CleanupHostedService(IOptions<CleanupOptions> options, IServiceScopeFactory serviceScopeFactory, ILogger<CleanupHostedService> logger)
+    {
+        _serviceScopeFactory = serviceScopeFactory;
+        _logger = logger;
+        _interval = options.Value.SweepInterval;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            using IServiceScope scope = _serviceScopeFactory.CreateScope();
+            CleanupJob job = scope.ServiceProvider.GetRequiredService<CleanupJob>();
+
+            IDistributedLockProvider distributedLockProvider = scope.ServiceProvider.GetRequiredService<IDistributedLockProvider>();
+
+            await Task.Delay(_interval, stoppingToken);
+            await using IDistributedSynchronizationHandle handle = await distributedLockProvider.AcquireLockAsync(nameof(CleanupHostedService), cancellationToken: stoppingToken);
+
+            try
+            {
+                await job.ExecuteAsync(stoppingToken);
+            }
+            catch (Exception e)
+            {
+                _logger.LogError(e, "Failed to perform cleanup this time around. Next cleanup attempt will happen in {Interval}", _interval);
+            }
+        }
+    }
+}

src/modules/Elsa.Retention/Jobs/CleanupJob.cs

@@ -0,0 +1,130 @@
+using System.Diagnostics.CodeAnalysis;
+using Elsa.Common.Models;
+using Elsa.Retention.Contracts;
+using Elsa.Retention.Options;
+using Elsa.Workflows.Management;
+using Elsa.Workflows.Management.Entities;
+using Elsa.Workflows.Management.Filters;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace Elsa.Retention.Jobs;
+
+/// <summary>
+///     Deletes all workflow instances that match any of the defined <see cref="IRetentionPolicy" />
+/// </summary>
+[SuppressMessage("Trimming", "IL2055:Either the type on which the MakeGenericType is called can\'t be statically determined, or the type parameters to be used for generic arguments can\'t be statically determined.")]
+public class CleanupJob
+{
+    private readonly ILogger _logger;
+    private readonly CleanupOptions _options;
+    private readonly IServiceProvider _serviceProvider;
+    private readonly IWorkflowInstanceStore _workflowInstanceStore;
+
+    /// <summary>
+    ///     Creates a new cleanup job
+    /// </summary>
+    /// <param name="workflowInstanceStore"></param>
+    /// <param name="options"></param>
+    /// <param name="serviceProvider"></param>
+    /// <param name="logger"></param>
+    public CleanupJob(
+        IWorkflowInstanceStore workflowInstanceStore,
+        IOptions<CleanupOptions> options,
+        IServiceProvider serviceProvider,
+        ILogger<CleanupJob> logger)
+    {
+        _workflowInstanceStore = workflowInstanceStore;
+        _options = options.Value;
+        _serviceProvider = serviceProvider;
+        _logger = logger;
+    }
+
+    /// <summary>
+    ///     Executes the cleanup job
+    /// </summary>
+    /// <param name="cancellationToken"></param>
+    public async Task ExecuteAsync(CancellationToken cancellationToken = default)
+    {
+        await using AsyncServiceScope scope = _serviceProvider.CreateAsyncScope();
+
+        IEnumerable<IRetentionPolicy> policies = scope.ServiceProvider.GetServices<IRetentionPolicy>();
+        Dictionary<Type, object> collectors = GetServices(typeof(IRelatedEntityCollector), typeof(IRelatedEntityCollector<>));
+
+
+        foreach (IRetentionPolicy policy in policies)
+        {
+            WorkflowInstanceFilter filter = policy.FilterFactory(scope.ServiceProvider).Build();
+            PageArgs pageArgs = PageArgs.FromPage(0, _options.PageSize);
+
+            long deletedWorkflowInstances = 0;
+
+            while (true)
+            {
+                Page<WorkflowInstance> page = await _workflowInstanceStore.FindManyAsync(filter, pageArgs, cancellationToken);
+
+                if (page.Items.Count == 0)
+                {
+                    break;
+                }
+
+                foreach (KeyValuePair<Type, object> collectorService in collectors)
+                {
+                    Type cleanupStrategyConcreteType = policy.CleanupStrategy.MakeGenericType(collectorService.Key);
+
+                    IRelatedEntityCollector? collector = collectorService.Value as IRelatedEntityCollector;
+                    ICleanupStrategy? cleanupService = _serviceProvider.GetService(cleanupStrategyConcreteType) as ICleanupStrategy;
+
+                    if (collector == null)
+                    {
+                        _logger.LogWarning("Failed to collect entities of type {Type}", collectorService.Key.Name);
+                        continue;
+                    }
+
+                    if (cleanupService == null)
+                    {
+                        _logger.LogWarning("Failed to clean up {Type} no clean up strategy found that implements {CleanupType}", collectorService.Key.Name, policy.CleanupStrategy.Name);
+                        continue;
+                    }
+
+                    await foreach (ICollection<object> entities in collector.GetRelatedEntitiesGeneric(page.Items).WithCancellation(cancellationToken))
+                    {
+                        await cleanupService.Cleanup(entities);
+                    }
+                }
+
+                deletedWorkflowInstances += await _workflowInstanceStore.DeleteAsync(new WorkflowInstanceFilter
+                {
+                    Ids = page.Items.Select(x => x.Id).ToArray()
+                }, cancellationToken);
+
+                if (page.TotalCount <= page.Items.Count + pageArgs.Offset)
+                {
+                    break;
+                }
+            }
+
+            _logger.LogInformation("Cleaned up {WorkflowInstanceCount} workflow instances through {Policy}", deletedWorkflowInstances, policy.Name);
+        }
+    }
+
+    private Dictionary<Type, object> GetServices(Type baseType, Type openType)
+    {
+        IEnumerable<object?> services = _serviceProvider.GetServices(baseType);
+
+        return services
+            .Where(x => x?.GetType() != null)
+            .Select(service => new
+            {
+                Service = service,
+                GenericArgument = service!.GetType()
+                    .GetInterfaces()
+                    .FirstOrDefault(i => i.IsGenericType && i.GetGenericTypeDefinition() == openType)?
+                    .GetGenericArguments()
+                    .FirstOrDefault()
+            })
+            .Where(x => x.GenericArgument != null)
+            .ToDictionary(x => x.GenericArgument!, x => x.Service)!;
+    }
+}

src/modules/Elsa.Retention/Models/RetentionWorkflowInstanceFilter.cs

@@ -0,0 +1,100 @@
+using Elsa.Workflows;
+using Elsa.Workflows.Management.Filters;
+using Elsa.Workflows.Management.Models;
+
+namespace Elsa.Retention.Models;
+
+/// <summary>
+///     A filter for querying workflow instances.
+/// </summary>
+public class RetentionWorkflowInstanceFilter
+{
+    /// <summary>
+    ///     Filter workflow instances that match the specified search term.
+    /// </summary>
+    public string? SearchTerm { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by definition ID.
+    /// </summary>
+    public string? DefinitionId { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by definition version ID.
+    /// </summary>
+    public string? DefinitionVersionId { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by definition IDs.
+    /// </summary>
+    public ICollection<string>? DefinitionIds { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by correlation ID.
+    /// </summary>
+    public string? CorrelationId { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by correlation IDs.
+    /// </summary>
+    public ICollection<string>? CorrelationIds { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by status.
+    /// </summary>
+    public WorkflowStatus? WorkflowStatus { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by a set of statuses.
+    /// </summary>
+    public ICollection<WorkflowStatus>? WorkflowStatuses { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by sub-status.
+    /// </summary>
+    public WorkflowSubStatus? WorkflowSubStatus { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by a set of sub-status.
+    /// </summary>
+    public ICollection<WorkflowSubStatus>? WorkflowSubStatuses { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by whether they have incidents.
+    /// </summary>
+    public bool? HasIncidents { get; set; }
+
+    /// <summary>
+    ///     Filter on workflows that are system workflows.
+    /// </summary>
+    public bool? IsSystem { get; set; }
+
+    /// <summary>
+    ///     Filter workflow instances by timestamp.
+    /// </summary>
+    public ICollection<TimestampFilter>? TimestampFilters { get; set; }
+
+    /// <summary>
+    ///     Creates a workflow instance filter based on the current filter
+    /// </summary>
+    /// <returns></returns>
+    public WorkflowInstanceFilter Build()
+    {
+        return new WorkflowInstanceFilter
+        {
+            CorrelationId = CorrelationId,
+            CorrelationIds = CorrelationIds,
+            DefinitionId = DefinitionId,
+            DefinitionIds = DefinitionIds,
+            SearchTerm = SearchTerm,
+            WorkflowStatus = WorkflowStatus,
+            WorkflowStatuses = WorkflowStatuses,
+            HasIncidents = HasIncidents,
+            IsSystem = IsSystem,
+            TimestampFilters = TimestampFilters,
+            DefinitionVersionId = DefinitionVersionId,
+            WorkflowSubStatuses = WorkflowSubStatuses,
+            WorkflowSubStatus = WorkflowSubStatus
+        };
+    }
+}

src/modules/Elsa.Retention/Options/CleanupOptions.cs

@@ -0,0 +1,17 @@
+namespace Elsa.Retention.Options;
+
+/// <summary>
+///     Retention options
+/// </summary>
+public class CleanupOptions
+{
+    /// <summary>
+    ///     Controls how often the database is checked for workflow instances and execution log records to remove.
+    /// </summary>
+    public TimeSpan SweepInterval { get; set; } = TimeSpan.FromHours(4);
+
+    /// <summary>
+    ///     Controls the page size of the workflow instance that are retained in a single batch
+    /// </summary>
+    public int PageSize { get; set; } = 25;
+}

src/modules/Elsa.Retention/Policies/DeletionRetentionPolicy.cs

@@ -0,0 +1,21 @@
+using Elsa.Retention.Contracts;
+using Elsa.Retention.Models;
+
+namespace Elsa.Retention.Policies;
+
+/// <summary>
+///     A policy that will delete the workflow instance and its related entities
+/// </summary>
+public class DeletionRetentionPolicy : IRetentionPolicy
+{
+    public DeletionRetentionPolicy(string name, Func<IServiceProvider, RetentionWorkflowInstanceFilter> filter)
+    {
+        Name = name;
+        FilterFactory = filter;
+    }
+
+    public string Name { get; }
+    public Func<IServiceProvider, RetentionWorkflowInstanceFilter> FilterFactory { get; }
+
+    public Type CleanupStrategy => typeof(IDeletionCleanupStrategy<>);
+}