Workflow purger (#402)

Author: danielgerlag

ReleaseNotes/2.1.2.md

@@ -0,0 +1,9 @@
+# Workflow Core 2.1.2
+
+* Adds a feature to purge old workflows from the persistence store.
+
+New `IWorkflowPurger` service that can be injected from the IoC container
+```c#
+Task PurgeWorkflows(WorkflowStatus status, DateTime olderThan)
+```
+Implementations are currently only for SQL Server, Postgres and MongoDB

WorkflowCore.sln

@@ -100,6 +100,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ReleaseNotes", "ReleaseNote
 		ReleaseNotes\1.9.3.md = ReleaseNotes\1.9.3.md
 		ReleaseNotes\2.0.0.md = ReleaseNotes\2.0.0.md
 		ReleaseNotes\2.1.0.md = ReleaseNotes\2.1.0.md
+		ReleaseNotes\2.1.2.md = ReleaseNotes\2.1.2.md
 	EndProjectSection
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample14", "src\samples\WorkflowCore.Sample14\WorkflowCore.Sample14.csproj", "{6BC66637-B42A-4334-ADFB-DBEC9F29D293}"

src/WorkflowCore/Interface/IWorkflowPurger.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Threading.Tasks;
+using WorkflowCore.Models;
+
+namespace WorkflowCore.Interface
+{
+    public interface IWorkflowPurger
+    {
+        Task PurgeWorkflows(WorkflowStatus status, DateTime olderThan);
+    }
+}

src/WorkflowCore/WorkflowCore.csproj

@@ -15,9 +15,9 @@
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
     <Description>Workflow Core is a light weight workflow engine targeting .NET Standard.</Description>
-    <Version>2.1.1</Version>
-    <AssemblyVersion>2.1.1.0</AssemblyVersion>
-    <FileVersion>2.1.1.0</FileVersion>
+    <Version>2.1.2</Version>
+    <AssemblyVersion>2.1.2.0</AssemblyVersion>
+    <FileVersion>2.1.2.0</FileVersion>
     <PackageReleaseNotes></PackageReleaseNotes>
     <PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
   </PropertyGroup>

src/providers/WorkflowCore.Persistence.EntityFramework/Services/WorkflowPurger.cs

@@ -0,0 +1,52 @@
+using System;
+using System.Linq;
+using System.Linq.Dynamic.Core;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+using WorkflowCore.Persistence.EntityFramework.Interfaces;
+using WorkflowCore.Persistence.EntityFramework.Models;
+
+namespace WorkflowCore.Persistence.EntityFramework.Services
+{
+    public class WorkflowPurger : IWorkflowPurger
+    {
+        private readonly IWorkflowDbContextFactory _contextFactory;
+
+        public WorkflowPurger(IWorkflowDbContextFactory contextFactory)
+        {
+            _contextFactory = contextFactory;
+        }
+        
+        public async Task PurgeWorkflows(WorkflowStatus status, DateTime olderThan)
+        {
+            var olderThanUtc = olderThan.ToUniversalTime();
+            using (var db = ConstructDbContext())
+            {
+                var workflows = await db.Set<PersistedWorkflow>().Where(x => x.Status == status && x.CompleteTime < olderThanUtc).ToListAsync();
+                foreach (var wf in workflows)
+                {
+                    foreach (var pointer in wf.ExecutionPointers)
+                    {
+                        foreach (var extAttr in pointer.ExtensionAttributes)
+                        {
+                            db.Remove(extAttr);
+                        }
+
+                        db.Remove(pointer);
+                    }
+                    db.Remove(wf);
+                }
+
+                await db.SaveChangesAsync();
+            }
+        }
+        
+        
+        private WorkflowDbContext ConstructDbContext()
+        {
+            return _contextFactory.Build();
+        }
+    }
+}

src/providers/WorkflowCore.Persistence.EntityFramework/WorkflowCore.Persistence.EntityFramework.csproj

@@ -14,10 +14,10 @@
     <GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute>
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
-    <Version>2.0.1</Version>
+    <Version>2.1.1</Version>
     <Description>Base package for Workflow-core peristence providers using entity framework</Description>
-    <AssemblyVersion>2.0.1.0</AssemblyVersion>
-    <FileVersion>2.0.1.0</FileVersion>
+    <AssemblyVersion>2.1.1.0</AssemblyVersion>
+    <FileVersion>2.1.1.0</FileVersion>
   </PropertyGroup>
 
   <ItemGroup>

src/providers/WorkflowCore.Persistence.MongoDB/ServiceCollectionExtensions.cs

@@ -3,6 +3,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using WorkflowCore.Interface;
 using WorkflowCore.Models;
 using WorkflowCore.Persistence.MongoDB.Services;
 
@@ -18,6 +19,12 @@ public static WorkflowOptions UseMongoDB(this WorkflowOptions options, string mo
                 var db = client.GetDatabase(databaseName);
                 return new MongoPersistenceProvider(db);
             });
+            options.Services.AddTransient<IWorkflowPurger>(sp =>
+            {
+                var client = new MongoClient(mongoUrl);
+                var db = client.GetDatabase(databaseName);
+                return new WorkflowPurger(db);
+            });
             return options;
         }
     }

src/providers/WorkflowCore.Persistence.MongoDB/Services/MongoPersistenceProvider.cs

@@ -13,6 +13,7 @@ namespace WorkflowCore.Persistence.MongoDB.Services
 {
     public class MongoPersistenceProvider : IPersistenceProvider
     {
+        internal const string WorkflowCollectionName = "wfc.workflows";
         private readonly IMongoDatabase _database;
 
         public MongoPersistenceProvider(IMongoDatabase database)
@@ -77,7 +78,7 @@ static void CreateIndexes(MongoPersistenceProvider instance)
             }
         }
 
-        private IMongoCollection<WorkflowInstance> WorkflowInstances => _database.GetCollection<WorkflowInstance>("wfc.workflows");
+        private IMongoCollection<WorkflowInstance> WorkflowInstances => _database.GetCollection<WorkflowInstance>(WorkflowCollectionName);
 
         private IMongoCollection<EventSubscription> EventSubscriptions => _database.GetCollection<EventSubscription>("wfc.subscriptions");
 

src/providers/WorkflowCore.Persistence.MongoDB/Services/WorkflowPurger.cs

@@ -0,0 +1,26 @@
+using System;
+using System.Threading.Tasks;
+using MongoDB.Driver;
+using WorkflowCore.Models;
+using WorkflowCore.Interface;
+
+namespace WorkflowCore.Persistence.MongoDB.Services
+{
+    public class WorkflowPurger : IWorkflowPurger
+    {
+        private readonly IMongoDatabase _database;
+        private IMongoCollection<WorkflowInstance> WorkflowInstances => _database.GetCollection<WorkflowInstance>(MongoPersistenceProvider.WorkflowCollectionName);
+
+
+        public WorkflowPurger(IMongoDatabase database)
+        {
+            _database = database;
+        }
+        
+        public async Task PurgeWorkflows(WorkflowStatus status, DateTime olderThan)
+        {
+            var olderThanUtc = olderThan.ToUniversalTime();
+            await WorkflowInstances.DeleteManyAsync(x => x.Status == status && x.CompleteTime < olderThanUtc);
+        }
+    }
+}

src/providers/WorkflowCore.Persistence.MongoDB/WorkflowCore.Persistence.MongoDB.csproj

@@ -14,10 +14,10 @@
     <GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute>
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
-    <Version>2.0.0</Version>
+    <Version>2.1.0</Version>
     <Description>Provides support to persist workflows running on Workflow Core to a MongoDB database.</Description>
-    <AssemblyVersion>2.0.0.0</AssemblyVersion>
-    <FileVersion>2.0.0.0</FileVersion>
+    <AssemblyVersion>2.1.0.0</AssemblyVersion>
+    <FileVersion>2.1.0.0</FileVersion>
   </PropertyGroup>
 
   <ItemGroup>