< Summary

Information
Class: Elsa.Workflows.Runtime.KeyValueWorkflowDispatchOutboxStore
Assembly: Elsa.Workflows.Runtime
File(s): /home/runner/work/elsa-core/elsa-core/src/modules/Elsa.Workflows.Runtime/Services/KeyValueWorkflowDispatchOutboxStore.cs
Line coverage
96%
Covered lines: 199
Uncovered lines: 8
Coverable lines: 207
Total lines: 358
Line coverage: 96.1%
Branch coverage
90%
Covered branches: 65
Total branches: 72
Branch coverage: 90.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
SaveAsync()100%11100%
SaveIndexAsync()100%11100%
FindManyAsync()100%11100%
FindManyAsync()100%22100%
DeleteAsync()100%44100%
FindIndexedItemsAsync()90%101089.65%
FindRecoveryItemsAsync()90%101088.46%
FindLegacyItemsAsync()100%1010100%
DeleteIndexesForMissingItemAsync()100%88100%
DeleteUnrecoverableItemAsync()100%44100%
GetItemKey(...)100%11100%
GetLegacyKey(...)100%11100%
GetIndexKey(...)100%11100%
GetIndexByIdKey(...)100%11100%
GetRecoveryKey(...)100%11100%
GetLegacyScanTake(...)75%44100%
IsLegacyScanComplete(...)100%22100%
IsIndexKeyForId(...)71.42%151481.81%
IsRecoverableItemRecord(...)100%66100%

File(s)

/home/runner/work/elsa-core/elsa-core/src/modules/Elsa.Workflows.Runtime/Services/KeyValueWorkflowDispatchOutboxStore.cs

#LineLine coverage
 1using Elsa.KeyValues.Contracts;
 2using Elsa.KeyValues.Entities;
 3using Elsa.KeyValues.Models;
 4using Elsa.Workflows.Runtime.Models;
 5
 6namespace Elsa.Workflows.Runtime;
 7
 8/// <summary>
 9/// Stores workflow dispatch outbox items in the existing key-value store.
 10/// </summary>
 1911public class KeyValueWorkflowDispatchOutboxStore(IKeyValueStore keyValueStore, IPayloadSerializer payloadSerializer) : I
 12{
 13    private const string LegacyKeyPrefix = "Elsa:WorkflowDispatchOutbox:";
 14    private const string ItemKeyPrefix = "Elsa:WorkflowDispatchOutbox:Items:";
 15    private const string IndexKeyPrefix = "Elsa:WorkflowDispatchOutbox:Index:";
 16    private const string IndexByIdKeyPrefix = "Elsa:WorkflowDispatchOutbox:IndexById:";
 17    private const string RecoveryKeyPrefix = "Elsa:WorkflowDispatchOutbox:Recovery:";
 18    private const string StateKeyPrefix = "Elsa:WorkflowDispatchOutbox:State:";
 19    private const string LegacyScanCompletedKey = $"{StateKeyPrefix}LegacyScanCompleted";
 20
 21    /// <inheritdoc />
 22    public async Task SaveAsync(WorkflowDispatchOutboxItem item, CancellationToken cancellationToken = default)
 23    {
 1124        await keyValueStore.SaveAsync(new SerializedKeyValuePair
 1125        {
 1126            Key = GetRecoveryKey(item.Id),
 1127            SerializedValue = item.Id,
 1128            TenantId = item.TenantId
 1129        }, cancellationToken);
 30
 1131        await keyValueStore.SaveAsync(new SerializedKeyValuePair
 1132        {
 1133            Key = GetItemKey(item.Id),
 1134            SerializedValue = payloadSerializer.Serialize(item),
 1135            TenantId = item.TenantId
 1136        }, cancellationToken);
 37
 1138        await SaveIndexAsync(item, cancellationToken);
 1139        await keyValueStore.DeleteAsync(GetRecoveryKey(item.Id), cancellationToken);
 1140    }
 41
 42    private async Task SaveIndexAsync(WorkflowDispatchOutboxItem item, CancellationToken cancellationToken)
 43    {
 1444        var indexKey = GetIndexKey(item);
 45
 1446        await keyValueStore.SaveAsync(new SerializedKeyValuePair
 1447        {
 1448            Key = indexKey,
 1449            SerializedValue = item.Id,
 1450            TenantId = item.TenantId
 1451        }, cancellationToken);
 52
 1453        await keyValueStore.SaveAsync(new SerializedKeyValuePair
 1454        {
 1455            Key = GetIndexByIdKey(item.Id),
 1456            SerializedValue = indexKey,
 1457            TenantId = item.TenantId
 1458        }, cancellationToken);
 1459    }
 60
 61    /// <inheritdoc />
 62    public async Task<IEnumerable<WorkflowDispatchOutboxItem>> FindManyAsync(CancellationToken cancellationToken = defau
 63    {
 664        return await FindManyAsync(maxCount: 0, cancellationToken);
 665    }
 66
 67    /// <inheritdoc />
 68    public async Task<IEnumerable<WorkflowDispatchOutboxItem>> FindManyAsync(int maxCount, CancellationToken cancellatio
 69    {
 1470        var indexedItems = await FindIndexedItemsAsync(maxCount, cancellationToken);
 1471        var recoverableItems = await FindRecoveryItemsAsync(maxCount, cancellationToken);
 1472        var legacyItems = await FindLegacyItemsAsync(maxCount, cancellationToken);
 1473        var items = indexedItems
 1474            .Concat(recoverableItems)
 1475            .Concat(legacyItems)
 2076            .GroupBy(x => x.Id)
 1877            .Select(x => x.First())
 3078            .OrderBy(x => x.CreatedAt);
 79
 1480        if (maxCount > 0)
 881            return items.Take(maxCount).ToList();
 82
 683        return items.ToList();
 1484    }
 85
 86    /// <inheritdoc />
 87    public async Task DeleteAsync(string id, CancellationToken cancellationToken = default)
 88    {
 789        var itemKey = GetItemKey(id);
 790        var record = await keyValueStore.FindAsync(new KeyValueFilter { Key = itemKey }, cancellationToken);
 91
 792        if (record == null)
 93        {
 494            await DeleteIndexesForMissingItemAsync(id, cancellationToken);
 495            await keyValueStore.DeleteAsync(GetRecoveryKey(id), cancellationToken);
 496            await keyValueStore.DeleteAsync(GetLegacyKey(id), cancellationToken);
 497            await keyValueStore.DeleteAsync(itemKey, cancellationToken);
 498            return;
 99        }
 100
 3101        var item = payloadSerializer.Deserialize<WorkflowDispatchOutboxItem>(record.SerializedValue);
 102
 3103        if (item != null)
 104        {
 2105            await keyValueStore.DeleteAsync(GetIndexKey(item), cancellationToken);
 2106            await keyValueStore.DeleteAsync(GetIndexByIdKey(id), cancellationToken);
 107        }
 108        else
 109        {
 1110            await DeleteIndexesForMissingItemAsync(id, cancellationToken);
 111        }
 112
 3113        await keyValueStore.DeleteAsync(GetRecoveryKey(id), cancellationToken);
 3114        await keyValueStore.DeleteAsync(GetLegacyKey(id), cancellationToken);
 3115        await keyValueStore.DeleteAsync(itemKey, cancellationToken);
 7116    }
 117
 118    private async Task<IEnumerable<WorkflowDispatchOutboxItem>> FindIndexedItemsAsync(int maxCount, CancellationToken ca
 119    {
 14120        var indexRecords = (await keyValueStore.FindManyAsync(new KeyValueFilter
 14121        {
 14122            Key = IndexKeyPrefix,
 14123            StartsWith = true,
 14124            OrderByKey = true,
 14125            Take = maxCount > 0 ? maxCount : null
 14126        }, cancellationToken)).ToList();
 23127        var itemKeys = indexRecords.Select(x => GetItemKey(x.SerializedValue)).ToList();
 128
 14129        var itemRecords = itemKeys.Count == 0
 14130            ? Array.Empty<SerializedKeyValuePair>()
 14131            : await keyValueStore.FindManyAsync(new KeyValueFilter
 14132            {
 14133                Keys = itemKeys
 14134            }, cancellationToken);
 23135        var itemRecordLookup = itemRecords.ToDictionary(x => x.Key);
 14136        var items = new List<WorkflowDispatchOutboxItem>();
 137
 46138        foreach (var indexRecord in indexRecords)
 139        {
 9140            var id = indexRecord.SerializedValue;
 9141            var itemKey = GetItemKey(id);
 142
 9143            if (!itemRecordLookup.TryGetValue(itemKey, out var itemRecord))
 144            {
 0145                await keyValueStore.DeleteAsync(indexRecord.Key, cancellationToken);
 0146                await keyValueStore.DeleteAsync(GetIndexByIdKey(id), cancellationToken);
 0147                continue;
 148            }
 149
 9150            var item = payloadSerializer.Deserialize<WorkflowDispatchOutboxItem>(itemRecord.SerializedValue);
 151
 9152            if (item == null)
 153            {
 1154                await DeleteUnrecoverableItemAsync(id, itemKey, indexRecord.Key, recoveryKey: null, cancellationToken: c
 1155                continue;
 156            }
 157
 8158            items.Add(item);
 8159        }
 160
 14161        return items;
 14162    }
 163
 164    private async Task<IEnumerable<WorkflowDispatchOutboxItem>> FindRecoveryItemsAsync(int maxCount, CancellationToken c
 165    {
 14166        var recoveryRecords = (await keyValueStore.FindManyAsync(new KeyValueFilter
 14167        {
 14168            Key = RecoveryKeyPrefix,
 14169            StartsWith = true,
 14170            OrderByKey = true,
 14171            Take = maxCount > 0 ? maxCount : null
 14172        }, cancellationToken)).ToList();
 173
 14174        if (recoveryRecords.Count == 0)
 11175            return [];
 176
 7177        var itemKeys = recoveryRecords.Select(x => GetItemKey(x.SerializedValue)).ToList();
 7178        var itemRecords = (await keyValueStore.FindManyAsync(new KeyValueFilter { Keys = itemKeys }, cancellationToken))
 3179        var items = new List<WorkflowDispatchOutboxItem>();
 180
 14181        foreach (var recoveryRecord in recoveryRecords)
 182        {
 4183            var id = recoveryRecord.SerializedValue;
 184
 4185            if (!itemRecords.TryGetValue(GetItemKey(id), out var itemRecord))
 186            {
 0187                await DeleteIndexesForMissingItemAsync(id, cancellationToken);
 0188                await keyValueStore.DeleteAsync(recoveryRecord.Key, cancellationToken);
 0189                continue;
 190            }
 191
 4192            var item = payloadSerializer.Deserialize<WorkflowDispatchOutboxItem>(itemRecord.SerializedValue);
 193
 4194            if (item == null)
 195            {
 1196                await DeleteUnrecoverableItemAsync(id, itemRecord.Key, indexKey: null, recoveryKey: recoveryRecord.Key, 
 1197                continue;
 198            }
 199
 3200            await SaveIndexAsync(item, cancellationToken);
 3201            await keyValueStore.DeleteAsync(recoveryRecord.Key, cancellationToken);
 3202            items.Add(item);
 3203        }
 204
 3205        return items;
 14206    }
 207
 208    private async Task<IEnumerable<WorkflowDispatchOutboxItem>> FindLegacyItemsAsync(int maxCount, CancellationToken can
 209    {
 14210        var legacyScanCompleted = await keyValueStore.FindAsync(new KeyValueFilter { Key = LegacyScanCompletedKey }, can
 211
 14212        if (legacyScanCompleted != null)
 6213            return [];
 214
 8215        var records = await keyValueStore.FindManyAsync(new KeyValueFilter
 8216        {
 8217            Key = LegacyKeyPrefix,
 8218            StartsWith = true,
 8219            OrderByKey = true,
 8220            Take = GetLegacyScanTake(maxCount)
 8221        }, cancellationToken);
 8222        var recordList = records.ToList();
 223
 8224        var recoverableItems = recordList
 8225            .Where(IsRecoverableItemRecord)
 10226            .Select(x => new
 10227            {
 10228                Record = x,
 10229                Item = payloadSerializer.Deserialize<WorkflowDispatchOutboxItem>(x.SerializedValue)
 10230            })
 10231            .Where(x => x.Item != null)
 8232            .ToList();
 8233        var recoverableItemsToMigrate = maxCount > 0 ? recoverableItems.Take(maxCount).ToList() : recoverableItems;
 8234        var items = new List<WorkflowDispatchOutboxItem>();
 235
 34236        foreach (var recoverableItem in recoverableItemsToMigrate)
 237        {
 9238            var record = recoverableItem.Record;
 9239            var item = recoverableItem.Item!;
 9240            await SaveAsync(item, cancellationToken);
 241
 9242            if (!record.Key.StartsWith(ItemKeyPrefix, StringComparison.Ordinal))
 5243                await keyValueStore.DeleteAsync(record.Key, cancellationToken);
 244
 9245            items.Add(item);
 9246        }
 247
 8248        if (IsLegacyScanComplete(recordList.Count, maxCount))
 249        {
 7250            await keyValueStore.SaveAsync(new SerializedKeyValuePair
 7251            {
 7252                Key = LegacyScanCompletedKey,
 7253                SerializedValue = "true"
 7254            }, cancellationToken);
 255        }
 256
 8257        return items;
 14258    }
 259
 260    private async Task DeleteIndexesForMissingItemAsync(string id, CancellationToken cancellationToken)
 261    {
 6262        var lookupRecord = await keyValueStore.FindAsync(new KeyValueFilter { Key = GetIndexByIdKey(id) }, cancellationT
 263
 6264        if (lookupRecord != null)
 265        {
 5266            if (IsIndexKeyForId(lookupRecord.SerializedValue, id))
 267            {
 3268                await keyValueStore.DeleteAsync(lookupRecord.SerializedValue, cancellationToken);
 3269                await keyValueStore.DeleteAsync(lookupRecord.Key, cancellationToken);
 3270                return;
 271            }
 272
 2273            await keyValueStore.DeleteAsync(lookupRecord.Key, cancellationToken);
 274        }
 275
 3276        var matchingIndexRecords = await keyValueStore.FindManyAsync(new KeyValueFilter
 3277        {
 3278            Key = IndexKeyPrefix,
 3279            StartsWith = true
 3280        }, cancellationToken);
 281
 16282        foreach (var indexRecord in matchingIndexRecords.Where(x => x.SerializedValue == id || x.Key.EndsWith($":{id}", 
 283        {
 3284            await keyValueStore.DeleteAsync(indexRecord.Key, cancellationToken);
 285        }
 6286    }
 287
 288    private async Task DeleteUnrecoverableItemAsync(string id, string itemKey, string? indexKey, string? recoveryKey, Ca
 289    {
 2290        if (indexKey != null)
 291        {
 1292            await keyValueStore.DeleteAsync(indexKey, cancellationToken);
 1293            await keyValueStore.DeleteAsync(GetIndexByIdKey(id), cancellationToken);
 294        }
 295        else
 296        {
 1297            await DeleteIndexesForMissingItemAsync(id, cancellationToken);
 298        }
 299
 2300        await keyValueStore.DeleteAsync(recoveryKey ?? GetRecoveryKey(id), cancellationToken);
 2301        await keyValueStore.DeleteAsync(GetLegacyKey(id), cancellationToken);
 2302        await keyValueStore.DeleteAsync(itemKey, cancellationToken);
 2303    }
 304
 44305    private static string GetItemKey(string id) => $"{ItemKeyPrefix}{id}";
 306
 9307    private static string GetLegacyKey(string id) => $"{LegacyKeyPrefix}{id}";
 308
 16309    private static string GetIndexKey(WorkflowDispatchOutboxItem item) => $"{IndexKeyPrefix}{item.CreatedAt.UtcTicks:D20
 310
 23311    private static string GetIndexByIdKey(string id) => $"{IndexByIdKeyPrefix}{id}";
 312
 30313    private static string GetRecoveryKey(string id) => $"{RecoveryKeyPrefix}{id}";
 314
 315    private static int? GetLegacyScanTake(int maxCount)
 316    {
 16317        if (maxCount <= 0)
 8318            return null;
 319
 8320        return maxCount == int.MaxValue ? int.MaxValue : maxCount + 1;
 321    }
 322
 323    private static bool IsLegacyScanComplete(int recordCount, int maxCount)
 324    {
 8325        var scanTake = GetLegacyScanTake(maxCount);
 8326        return scanTake == null || recordCount < scanTake.Value;
 327    }
 328
 329    private static bool IsIndexKeyForId(string indexKey, string id)
 330    {
 5331        var expectedSuffix = $":{id}";
 332
 5333        if (!indexKey.StartsWith(IndexKeyPrefix, StringComparison.Ordinal) || !indexKey.EndsWith(expectedSuffix, StringC
 2334            return false;
 335
 3336        var ticksStart = IndexKeyPrefix.Length;
 3337        var ticksLength = indexKey.Length - ticksStart - expectedSuffix.Length;
 338
 3339        if (ticksLength != 20)
 0340            return false;
 341
 126342        foreach (var character in indexKey.AsSpan(ticksStart, ticksLength))
 343        {
 60344            if (character is < '0' or > '9')
 0345                return false;
 346        }
 347
 3348        return true;
 349    }
 350
 351    private static bool IsRecoverableItemRecord(SerializedKeyValuePair record)
 352    {
 13353        return !record.Key.StartsWith(IndexKeyPrefix, StringComparison.Ordinal)
 13354               && !record.Key.StartsWith(IndexByIdKeyPrefix, StringComparison.Ordinal)
 13355               && !record.Key.StartsWith(RecoveryKeyPrefix, StringComparison.Ordinal)
 13356               && !record.Key.StartsWith(StateKeyPrefix, StringComparison.Ordinal);
 357    }
 358}
Generated by: ReportGenerator 5.5.11.0
09/26/2026 - 05:55:50
GitHub | reportgenerator.io

Methods/Properties

.ctor(Elsa.KeyValues.Contracts.IKeyValueStore,Elsa.Workflows.IPayloadSerializer)
SaveAsync()
SaveIndexAsync()
FindManyAsync()
FindManyAsync()
DeleteAsync()
FindIndexedItemsAsync()
FindRecoveryItemsAsync()
FindLegacyItemsAsync()
DeleteIndexesForMissingItemAsync()
DeleteUnrecoverableItemAsync()
GetItemKey(System.String)
GetLegacyKey(System.String)
GetIndexKey(Elsa.Workflows.Runtime.Models.WorkflowDispatchOutboxItem)
GetIndexByIdKey(System.String)
GetRecoveryKey(System.String)
GetLegacyScanTake(System.Int32)
IsLegacyScanComplete(System.Int32,System.Int32)
IsIndexKeyForId(System.String,System.String)
IsRecoverableItemRecord(Elsa.KeyValues.Entities.SerializedKeyValuePair)