add support to get multiple workflow instances by a list of ids

src/WorkflowCore/Interface/IPersistenceProvider.cs

@@ -7,7 +7,7 @@ namespace WorkflowCore.Interface
 {
     /// <remarks>
     /// The implemention of this interface will be responsible for
-    /// persisiting running workflow instances to a durable store    
+    /// persisting running workflow instances to a durable store
     /// </remarks>
     public interface IPersistenceProvider
     {
@@ -22,6 +22,8 @@ public interface IPersistenceProvider
 
         Task<WorkflowInstance> GetWorkflowInstance(string Id);
 
+        Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids);
+
         Task<string> CreateEventSubscription(EventSubscription subscription);
 
         Task<IEnumerable<EventSubscription>> GetSubcriptions(string eventName, string eventKey, DateTime asOf);

src/WorkflowCore/Services/DefaultProviders/MemoryPersistenceProvider.cs

@@ -8,22 +8,22 @@
 
 namespace WorkflowCore.Services
 {
-    
+
     public interface ISingletonMemoryProvider : IPersistenceProvider
     {
     }
-    #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
 
     /// <summary>
     /// In-memory implementation of IPersistenceProvider for demo and testing purposes
     /// </summary>
     public class MemoryPersistenceProvider : ISingletonMemoryProvider
-    {        
+    {
         private readonly List<WorkflowInstance> _instances = new List<WorkflowInstance>();
         private readonly List<EventSubscription> _subscriptions = new List<EventSubscription>();
         private readonly List<Event> _events = new List<Event>();
         private readonly List<ExecutionError> _errors = new List<ExecutionError>();
-                
+
         public async Task<string> CreateNewWorkflow(WorkflowInstance workflow)
         {
             lock (_instances)
@@ -61,6 +61,19 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             }
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            lock (_instances)
+            {
+                return _instances.Where(x => ids.Contains(x.Id));
+            }
+        }
+
         public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
         {
             lock (_instances)
@@ -121,7 +134,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)
         }
 
         public void EnsureStoreExists()
-        {            
+        {
         }
 
         public async Task<string> CreateEvent(Event newEvent)
@@ -133,7 +146,7 @@ public async Task<string> CreateEvent(Event newEvent)
                 return newEvent.Id;
             }
         }
-        
+
         public async Task MarkEventProcessed(string id)
         {
             lock (_events)
@@ -197,5 +210,5 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)
         }
     }
 
-    #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
 }

src/WorkflowCore/Services/DefaultProviders/TransientMemoryPersistenceProvider.cs

@@ -36,6 +36,8 @@ public TransientMemoryPersistenceProvider(ISingletonMemoryProvider innerService)
 
         public Task<WorkflowInstance> GetWorkflowInstance(string Id) => _innerService.GetWorkflowInstance(Id);
 
+        public Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids) => _innerService.GetWorkflowInstances(ids);
+
         public Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take) => _innerService.GetWorkflowInstances(status, type, createdFrom, createdTo, skip, take);
 
         public Task MarkEventProcessed(string id) => _innerService.MarkEventProcessed(id);

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

@@ -26,7 +26,7 @@ public EntityFrameworkPersistenceProvider(IWorkflowDbContextFactory contextFacto
             _canCreateDB = canCreateDB;
             _canMigrateDB = canMigrateDB;
         }
-        
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             using (var db = ConstructDbContext())
@@ -96,7 +96,7 @@ public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowSt
                 return result;
             }
         }
-        
+
         public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
         {
             using (var db = ConstructDbContext())
@@ -115,6 +115,26 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             }
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            using (var db = ConstructDbContext())
+            {
+                var uids = ids.Select(i => new Guid(i));
+                var raw = db.Set<PersistedWorkflow>()
+                    .Include(wf => wf.ExecutionPointers)
+                    .ThenInclude(ep => ep.ExtensionAttributes)
+                    .Include(wf => wf.ExecutionPointers)
+                    .Where(x => uids.Contains(x.InstanceId));
+
+                return (await raw.ToListAsync()).Select(i => i.ToWorkflowInstance());
+            }
+        }
+
         public async Task PersistWorkflow(WorkflowInstance workflow)
         {
             using (var db = ConstructDbContext())
@@ -143,7 +163,7 @@ public async Task TerminateSubscription(string eventSubscriptionId)
                 await db.SaveChangesAsync();
             }
         }
-                
+
         public virtual void EnsureStoreExists()
         {
             using (var context = ConstructDbContext())
@@ -283,7 +303,7 @@ public async Task PersistErrors(IEnumerable<ExecutionError> errors)
                 }
             }
         }
-        
+
         private WorkflowDbContext ConstructDbContext()
         {
             return _contextFactory.Build();

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

@@ -26,8 +26,8 @@ public MongoPersistenceProvider(IMongoDatabase database)
         static MongoPersistenceProvider()
         {
             BsonClassMap.RegisterClassMap<WorkflowInstance>(x =>
-            {                
-                x.MapIdProperty(y => y.Id)                    
+            {
+                x.MapIdProperty(y => y.Id)
                     .SetIdGenerator(new StringObjectIdGenerator());
                 x.MapProperty(y => y.Data)
                     .SetSerializer(new DataObjectSerializer());
@@ -48,9 +48,9 @@ static MongoPersistenceProvider()
                     .SetIdGenerator(new StringObjectIdGenerator());
                 x.MapProperty(y => y.EventName);
                 x.MapProperty(y => y.EventKey);
-                x.MapProperty(y => y.StepId);                
+                x.MapProperty(y => y.StepId);
                 x.MapProperty(y => y.WorkflowId);
-                x.MapProperty(y => y.SubscribeAsOf);                
+                x.MapProperty(y => y.SubscribeAsOf);
             });
 
             BsonClassMap.RegisterClassMap<Event>(x =>
@@ -113,7 +113,18 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             var result = await WorkflowInstances.FindAsync(x => x.Id == Id);
             return await result.FirstAsync();
         }
-
+
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var result = await WorkflowInstances.FindAsync(x => ids.Contains(x.Id));
+            return await result.ToListAsync();
+        }
+
         public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(WorkflowStatus? status, string type, DateTime? createdFrom, DateTime? createdTo, int skip, int take)
         {
             IQueryable<WorkflowInstance> result = WorkflowInstances.AsQueryable();
@@ -138,17 +149,17 @@ public async Task<string> CreateEventSubscription(EventSubscription subscription
             await EventSubscriptions.InsertOneAsync(subscription);
             return subscription.Id;
         }
-        
+
         public async Task TerminateSubscription(string eventSubscriptionId)
         {
             await EventSubscriptions.DeleteOneAsync(x => x.Id == eventSubscriptionId);
         }
 
         public void EnsureStoreExists()
         {
-            
-        }                
-                
+
+        }
+
         public async Task<IEnumerable<EventSubscription>> GetSubcriptions(string eventName, string eventKey, DateTime asOf)
         {
             var query = EventSubscriptions
@@ -192,7 +203,7 @@ public async Task<IEnumerable<string>> GetEvents(string eventName, string eventK
             var query = Events
                 .Find(x => x.EventName == eventName && x.EventKey == eventKey && x.EventTime >= asOf)
                 .Project(x => x.Id);
-            
+
             return await query.ToListAsync();
         }
 

src/providers/WorkflowCore.Providers.AWS/Services/DynamoPersistenceProvider.cs

@@ -4,6 +4,7 @@
 using Microsoft.Extensions.Logging;
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using WorkflowCore.Providers.AWS.Interface;
@@ -86,7 +87,7 @@ public async Task<IEnumerable<string>> GetRunnableInstances(DateTime asAt)
                 },
                 ScanIndexForward = true
             };
-            
+
             var response = await _client.QueryAsync(request);
 
             foreach (var item in response.Items)
@@ -117,6 +118,57 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             return response.Item.ToWorkflowInstance();
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var keys = new KeysAndAttributes() { Keys = new List<Dictionary<string, AttributeValue>>() };
+            foreach (var id in ids)
+            {
+                var key = new Dictionary<string, AttributeValue>()
+                {
+                    {
+                        "id", new AttributeValue { S = id }
+                    }
+                };
+                keys.Keys.Add(key);
+            }
+
+            var request = new BatchGetItemRequest
+            {
+                RequestItems = new Dictionary<string, KeysAndAttributes>()
+                {
+                    {
+                        $"{_tablePrefix}-{WORKFLOW_TABLE}", keys
+                    }
+                }
+            };
+
+            var result = new List<Dictionary<string, AttributeValue>>();
+            BatchGetItemResponse response;
+            do
+            {
+                // Making request
+                response = await _client.BatchGetItemAsync(request);
+
+                // Check the response
+                var responses = response.Responses; // Attribute list in the response.
+                foreach (var tableResponse in responses)
+                {
+                    result.AddRange(tableResponse.Value);
+                }
+
+                // Any unprocessed keys? could happen if you exceed ProvisionedThroughput or some other error.
+                Dictionary<string, KeysAndAttributes> unprocessedKeys = response.UnprocessedKeys;
+                request.RequestItems = unprocessedKeys;
+            } while (response.UnprocessedKeys.Count > 0);
+
+            return result.Select(i => i.ToWorkflowInstance());
+        }
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             subscription.Id = Guid.NewGuid().ToString();
@@ -326,4 +378,4 @@ public void EnsureStoreExists()
             _provisioner.ProvisionTables().Wait();
         }
     }
-}
+}

src/providers/WorkflowCore.Providers.Redis/Services/RedisPersistenceProvider.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -75,6 +76,17 @@ public async Task<WorkflowInstance> GetWorkflowInstance(string Id)
             return JsonConvert.DeserializeObject<WorkflowInstance>(raw, _serializerSettings);
         }
 
+        public async Task<IEnumerable<WorkflowInstance>> GetWorkflowInstances(IEnumerable<string> ids)
+        {
+            if (ids == null)
+            {
+                return new List<WorkflowInstance>();
+            }
+
+            var raw = await _redis.HashGetAsync($"{_prefix}.{WORKFLOW_SET}", Array.ConvertAll(ids.ToArray(), x => (RedisValue)x));
+            return raw.Select(r => JsonConvert.DeserializeObject<WorkflowInstance>(r, _serializerSettings));
+        }
+
         public async Task<string> CreateEventSubscription(EventSubscription subscription)
         {
             subscription.Id = Guid.NewGuid().ToString();