Added find functions to KeyValueStore

src/modules/Elsa.Dapper/Extensions/ParameterizedQueryBuilderExtensions.cs

@@ -200,6 +200,18 @@ public static class ParameterizedQueryBuilderExtensions
         return query;
     }
 
+    public static ParameterizedQuery StartsWith(this ParameterizedQuery query, string field, bool startsWith, string? value)
+    {
+        if (!startsWith || value == null || string.IsNullOrWhiteSpace(value))
+            return query;
+
+        var searchTermLike = $"{value}%";
+        query.Sql.AppendLine($"and {field} like @SearchTermLike");
+        query.Parameters.Add($"@{field}", searchTermLike);
+
+        return query;
+    }
+
     /// <summary>
     /// Appends an AND clause to the query if the value is not null.
     /// </summary>

src/modules/Elsa.Dapper/Modules/Runtime/Stores/KeyValueStore.cs

@@ -1,9 +1,11 @@
 using Elsa.Dapper.Contracts;
 using Elsa.Dapper.Extensions;
+using Elsa.Dapper.Models;
 using Elsa.Dapper.Modules.Runtime.Records;
 using Elsa.Dapper.Services;
 using Elsa.Workflows.Runtime.Contracts;
 using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Models;
 
 namespace Elsa.Dapper.Modules.Runtime.Stores;
 
@@ -32,18 +34,31 @@ public class DapperKeyValueStore : IKeyValueStore
     }
 
     /// <inheritdoc />
-    public async Task<SerializedKeyValuePair?> GetValue(string key, CancellationToken cancellationToken)
+    public async Task<SerializedKeyValuePair?> FindAsync(KeyValueFilter filter, CancellationToken cancellationToken)
     {
-        var record = await _store.FindAsync(query => query.Is(nameof(KeyValuePairRecord.Key), key), cancellationToken);
+        var record = await _store.FindAsync(q => ApplyFilter(q, filter), cancellationToken);
         return record == null ? null : Map(record);
     }
 
+    public Task<IEnumerable<SerializedKeyValuePair>> FindManyAsync(KeyValueFilter filter, CancellationToken cancellationToken)
+    {
+        throw new NotImplementedException();
+    }
+
     /// <inheritdoc />
     public Task DeleteAsync(string key, CancellationToken cancellationToken)
     {
         return _store.DeleteAsync(query => query.Is(nameof(KeyValuePairRecord.Key), key), cancellationToken);
     }
 
+    private void ApplyFilter(ParameterizedQuery query, KeyValueFilter filter)
+    {
+        query
+            .Is(nameof(KeyValuePairRecord.Key), filter.Key)
+            .In(nameof(KeyValuePairRecord.Key), filter.Keys)
+            .StartsWith(nameof(KeyValuePairRecord.Key), filter.StartsWith, filter.Key);
+    }
+
     private KeyValuePairRecord Map(SerializedKeyValuePair kvp)
     {
         return new()

src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/KeyValueStore.cs

@@ -1,6 +1,7 @@
 using Elsa.EntityFrameworkCore.Common;
 using Elsa.Workflows.Runtime.Contracts;
 using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Models;
 
 namespace Elsa.EntityFrameworkCore.Modules.Runtime;
 
@@ -23,9 +24,15 @@ public class EFCoreKeyValueStore : IKeyValueStore
     }
     
     /// <inheritdoc />
-    public Task<SerializedKeyValuePair?> GetValue(string key, CancellationToken cancellationToken)
+    public Task<SerializedKeyValuePair?> FindAsync(KeyValueFilter filter, CancellationToken cancellationToken)
     {
-        return _store.FindAsync(x => x.Key == key, cancellationToken);
+        return _store.FindAsync(filter.Apply, cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public Task<IEnumerable<SerializedKeyValuePair>> FindManyAsync(KeyValueFilter filter, CancellationToken cancellationToken)
+    {
+        return _store.QueryAsync(filter.Apply, cancellationToken);
     }
 
     /// <inheritdoc />

src/modules/Elsa.MongoDb/Modules/Runtime/KeyValueStore.cs

@@ -1,6 +1,9 @@
 using Elsa.MongoDb.Common;
 using Elsa.Workflows.Runtime.Contracts;
 using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Models;
+using MongoDB.Driver.Linq;
+using Open.Linq.AsyncExtensions;
 
 namespace Elsa.MongoDb.Modules.Runtime;
 
@@ -26,9 +29,15 @@ public class MongoKeyValueStore : IKeyValueStore
     }
 
     /// <inheritdoc />
-    public Task<SerializedKeyValuePair?> GetValue(string key, CancellationToken cancellationToken)
+    public Task<SerializedKeyValuePair?> FindAsync(KeyValueFilter filter, CancellationToken cancellationToken)
     {
-        return _keyValueMongoDbStore.FindAsync(x => x.Key == key, cancellationToken);
+        return _keyValueMongoDbStore.FindAsync(query => Filter(query, filter), cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task<IEnumerable<SerializedKeyValuePair>> FindManyAsync(KeyValueFilter filter, CancellationToken cancellationToken)
+    {
+        return await _keyValueMongoDbStore.FindManyAsync(query => Filter(query, filter), cancellationToken).ToList();
     }
 
     /// <inheritdoc />
@@ -36,4 +45,7 @@ public class MongoKeyValueStore : IKeyValueStore
     {
         return _keyValueMongoDbStore.DeleteWhereAsync(x => x.Key == key, cancellationToken);
     }
+
+    private IMongoQueryable<SerializedKeyValuePair> Filter(IMongoQueryable<SerializedKeyValuePair> queryable, KeyValueFilter filter) =>
+        (filter.Apply(queryable) as IMongoQueryable<SerializedKeyValuePair>)!;
 }

src/modules/Elsa.Workflows.Runtime/Contracts/IKeyValueStore.cs

@@ -1,5 +1,7 @@
+using System.Linq.Expressions;
 using Elsa.Workflows.Runtime.Entities;
- 
+using Elsa.Workflows.Runtime.Models;
+
 namespace Elsa.Workflows.Runtime.Contracts;
 
 /// <summary>
@@ -16,7 +18,12 @@ public interface IKeyValueStore
     /// Retrieves the key value pair from the store.
     /// </summary>
     /// <returns><see cref="SerializedKeyValuePair"/> if the key is found, otherwise null.</returns>
-    Task<SerializedKeyValuePair?> GetValue(string key, CancellationToken cancellationToken);
+    Task<SerializedKeyValuePair?> FindAsync(KeyValueFilter filter, CancellationToken cancellationToken);
+
+    /// <summary>
+    /// Retrieves all key value pairs which match the predicate.
+    /// </summary>
+    Task<IEnumerable<SerializedKeyValuePair>> FindManyAsync(KeyValueFilter filter, CancellationToken cancellationToken);
     
     /// <summary>
     /// If the key is found it deletes the record from the store. 

src/modules/Elsa.Workflows.Runtime/Models/KeyValueFilter.cs

@@ -0,0 +1,40 @@
+using Elsa.Workflows.Runtime.Entities;
+
+namespace Elsa.Workflows.Runtime.Models;
+
+public class KeyValueFilter
+{
+    /// <summary>
+    /// Gets or sets whether the <see cref="Key"/> needs to match the beginning of the key found.
+    /// </summary>
+    public bool StartsWith { get; set; }
+    
+    /// <summary>
+    /// Gets or sets the key to filter for.
+    /// </summary>
+    public string? Key { get; set; }
+    
+    /// <summary>
+    /// Gets or sets the keys to filter for.
+    /// </summary>
+    public ICollection<string>? Keys { get; set; }
+    
+    /// <summary>
+    /// Applies the filter to the specified queryable.
+    /// </summary>
+    /// <param name="queryable">The queryable.</param>
+    /// <returns>The filtered queryable.</returns>
+    public IQueryable<SerializedKeyValuePair> Apply(IQueryable<SerializedKeyValuePair> queryable)
+    {
+        var filter = this;
+        if (filter.Key != null)
+        {
+            queryable = StartsWith
+                ? queryable.Where(x => x.Key.StartsWith(filter.Key))
+                : queryable.Where(x => x.Key == filter.Key);
+        }
+        if (filter.Keys != null) queryable = queryable.Where(x => filter.Keys.Contains(x.Key));
+
+        return queryable;
+    }
+}

src/modules/Elsa.Workflows.Runtime/Stores/MemoryKeyValueStore.cs

@@ -1,6 +1,7 @@
 using Elsa.Common.Services;
 using Elsa.Workflows.Runtime.Contracts;
 using Elsa.Workflows.Runtime.Entities;
+using Elsa.Workflows.Runtime.Models;
 
 namespace Elsa.Workflows.Runtime.Stores;
 
@@ -27,12 +28,19 @@ public class MemoryKeyValueStore : IKeyValueStore
     }
 
     /// <inheritdoc />
-    public Task<SerializedKeyValuePair?> GetValue(string key, CancellationToken cancellationToken)
+    public Task<SerializedKeyValuePair?> FindAsync(KeyValueFilter filter, CancellationToken cancellationToken)
     {
-        var result = _store.Find(x => x.Key == key);
+        var result = _store.Query(filter.Apply).FirstOrDefault();
         return Task.FromResult(result);
     }
 
+    /// <inheritdoc />
+    public Task<IEnumerable<SerializedKeyValuePair>> FindManyAsync(KeyValueFilter filter, CancellationToken cancellationToken)
+    {
+        var result = _store.Query(filter.Apply);
+        return Task.FromResult(result);
+    }
+    
     /// <inheritdoc />
     public Task DeleteAsync(string key, CancellationToken cancellationToken)
     {