< Summary

Information
Class: Elsa.UserTasks.Persistence.VNext.Repositories.VNextUserTaskRepository
Assembly: Elsa.UserTasks.Persistence.VNext
File(s): /home/runner/work/elsa-core/elsa-core/src/modules/Elsa.UserTasks.Persistence.VNext/Repositories/VNextUserTaskRepository.cs
Line coverage
86%
Covered lines: 174
Uncovered lines: 28
Coverable lines: 202
Total lines: 324
Line coverage: 86.1%
Branch coverage
66%
Covered branches: 172
Total branches: 260
Branch coverage: 66.1%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
.cctor()100%11100%
GetAsync()100%22100%
QueryAsync()87.5%88100%
FindByMaterializationKeyAsync(...)100%11100%
FindByBookmarkIdAsync(...)100%11100%
FindByInvitationTokenHashAsync()100%66100%
SaveAsync()75%4476.92%
AddProjectionAsync()100%2271.42%
AppendEventAsync()100%2280%
TryMutateAsync()66.66%12646.15%
LoadAllAsync()100%22100%
FindByIndexAsync()100%44100%
LoadDocumentAsync()100%22100%
CreateRequest(...)58.33%1212100%
Deserialize(...)50%22100%
DocumentId(...)100%11100%
Matches(...)42.42%6666100%
IsVisible(...)38.23%773466.66%
IsEligible(...)62.5%9880%
NeedsAttention(...)0%110100%
ApplyOrdering(...)100%1414100%
ApplyCursor(...)93.54%706287.5%
CreateCursor(...)100%1010100%
TryReadCursor(...)66.66%6684.61%

File(s)

/home/runner/work/elsa-core/elsa-core/src/modules/Elsa.UserTasks.Persistence.VNext/Repositories/VNextUserTaskRepository.cs

#LineLine coverage
 1using System.Text.Json;
 2using System.Text.Json.Serialization;
 3using Elsa.Persistence.VNext.Document;
 4using Elsa.UserTasks.Contracts;
 5using Elsa.UserTasks.Models;
 6
 7namespace Elsa.UserTasks.Persistence.VNext.Repositories;
 8
 9/// <summary>
 10/// Provider-neutral document implementation. The aggregate is stored as one document while the schema
 11/// provider advertises the same logical units and indexes as the relational providers.
 12/// </summary>
 213public sealed class VNextUserTaskRepository(IDocumentStore documentStore) : IUserTaskRepository
 14{
 15    public const string StorageUnitName = "UserTasks";
 116    private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web)
 117    {
 118        Converters = { new JsonStringEnumConverter() }
 119    };
 20
 21    public async Task<UserTask?> GetAsync(string tenantId, string taskId, CancellationToken cancellationToken = default)
 22    {
 2223        var document = await documentStore.LoadAsync(StorageUnitName, DocumentId(tenantId, taskId), cancellationToken);
 2224        return document is null ? null : Deserialize(document);
 2225    }
 26
 27    public async Task<UserTaskQueryResult> QueryAsync(UserTaskQuery query, CancellationToken cancellationToken = default
 28    {
 10629        var tasks = new List<UserTask>();
 212030        foreach (var status in Enum.GetValues<UserTaskStatus>())
 31        {
 95432            var documents = await documentStore.QueryAsync(
 95433                new DocumentQuery(StorageUnitName, new Dictionary<string, string?>
 95434                {
 95435                    ["TenantId"] = query.TenantId,
 95436                    ["Status"] = status.ToString()
 95437                }), cancellationToken);
 95438            tasks.AddRange(documents.Select(Deserialize));
 39        }
 40
 134641        var filtered = tasks.Where(x => Matches(x, query)).Where(x => query.Scope is null || IsVisible(x, query.Scope)).
 10642        int? totalCount = query.IncludeTotalCount ? filtered.Count : null;
 10643        filtered = ApplyOrdering(filtered, query).ToList();
 10644        filtered = ApplyCursor(filtered, query).ToList();
 10645        var limit = Math.Clamp(query.Limit <= 0 ? 50 : query.Limit, 1, 200);
 10646        var hasMore = filtered.Count > limit;
 10647        var page = filtered.Take(limit).ToList();
 10648        return new UserTaskQueryResult(page, hasMore ? CreateCursor(page[^1], query.Sort) : null, totalCount);
 10649    }
 50
 25551    public Task<UserTask?> FindByMaterializationKeyAsync(string tenantId, string key, CancellationToken cancellationToke
 52
 353    public Task<UserTask?> FindByBookmarkIdAsync(string tenantId, string bookmarkId, CancellationToken cancellationToken
 54
 55    public async Task<(UserTask Task, UserTaskInvitation Invitation)?> FindByInvitationTokenHashAsync(string tokenHash, 
 56    {
 57        // The document store has no cross-tenant secondary index for invitation hashes, so this scans the
 58        // storage unit. Verification is rate limited and rare; correctness matters more than the scan here.
 21759        foreach (var task in await LoadAllAsync(cancellationToken))
 60        {
 10961            var invitation = task.Invitations.FirstOrDefault(x => string.Equals(x.TokenHash, tokenHash, StringComparison
 10762            if (invitation != null)
 163                return (task, invitation);
 64        }
 165        return null;
 266    }
 67
 68    public async Task SaveAsync(UserTask task, int expectedRevision, CancellationToken cancellationToken = default)
 69    {
 670        var existing = await LoadDocumentAsync(task.TenantId, task.Id, cancellationToken);
 671        if (existing is null)
 072            throw new KeyNotFoundException($"User task '{task.Id}' was not found.");
 673        var loaded = existing.Value;
 674        if (loaded.Task.Revision != expectedRevision)
 275            throw new UserTaskRevisionConflictException(task.Id, expectedRevision);
 76
 477        task.Revision = expectedRevision + 1;
 478        task.UpdatedAt = DateTimeOffset.UtcNow;
 79        try
 80        {
 481            await documentStore.SaveAsync(CreateRequest(task, loaded.Document.Version), cancellationToken);
 482        }
 083        catch (DocumentStoreConcurrencyException exception)
 84        {
 85            // A writer won the race between the read above and this save. Same contract as the other
 86            // providers so the caller sees one exception type regardless of the installed store.
 087            throw new UserTaskRevisionConflictException(task.Id, expectedRevision, exception);
 88        }
 489    }
 90
 91    public async Task AddProjectionAsync(UserTask task, CancellationToken cancellationToken = default)
 92    {
 8093        if (await FindByMaterializationKeyAsync(task.TenantId, task.MaterializationKey, cancellationToken) is not null)
 194            return;
 95        try
 96        {
 7997            await documentStore.SaveAsync(CreateRequest(task, expectedVersion: 0), cancellationToken);
 7998        }
 099        catch (DocumentStoreConcurrencyException)
 100        {
 101            // Projection is idempotent when the same aggregate ID was committed concurrently.
 0102        }
 80103    }
 104
 105    public async Task AppendEventAsync(string tenantId, string taskId, UserTaskEvent @event, CancellationToken cancellat
 106    {
 3107        var existing = await LoadDocumentAsync(tenantId, taskId, cancellationToken);
 3108        if (existing is null)
 1109            return;
 110
 111        // The document store has no separate audit stream, so the entry is written back with the aggregate.
 112        // The revision is deliberately left as-is: an audited read must not consume the concurrency token.
 2113        var loaded = existing.Value;
 2114        loaded.Task.Events.Add(@event);
 115        try
 116        {
 2117            await documentStore.SaveAsync(CreateRequest(loaded.Task, loaded.Document.Version), cancellationToken);
 2118        }
 0119        catch (DocumentStoreConcurrencyException)
 120        {
 121            // A concurrent writer won. The audit entry is advisory, so losing this race is not an error.
 0122        }
 3123    }
 124
 125    public async Task<bool> TryMutateAsync(string tenantId, string taskId, int expectedRevision, Func<UserTask, bool> mu
 126    {
 2127        var existing = await LoadDocumentAsync(tenantId, taskId, cancellationToken);
 2128        if (existing is null)
 0129            return false;
 2130        var loaded = existing.Value;
 2131        if (loaded.Task.Revision != expectedRevision || !mutation(loaded.Task))
 2132            return false;
 0133        loaded.Task.Revision = expectedRevision + 1;
 0134        loaded.Task.UpdatedAt = DateTimeOffset.UtcNow;
 135        try
 136        {
 0137            await documentStore.SaveAsync(CreateRequest(loaded.Task, loaded.Document.Version), cancellationToken);
 0138            return true;
 139        }
 0140        catch (DocumentStoreConcurrencyException)
 141        {
 0142            return false;
 143        }
 2144    }
 145
 146    /// <summary>
 147    /// Loads every stored task across tenants. Only the invitation-hash lookup uses this: an anonymous
 148    /// holder presents a secret and no tenant, so the scan cannot be narrowed by an index.
 149    /// </summary>
 150    private async Task<IReadOnlyCollection<UserTask>> LoadAllAsync(CancellationToken cancellationToken)
 151    {
 2152        var tasks = new List<UserTask>();
 40153        foreach (var status in Enum.GetValues<UserTaskStatus>())
 154        {
 18155            var documents = await documentStore.QueryAsync(new DocumentQuery(StorageUnitName, new Dictionary<string, str
 18156            {
 18157                ["Status"] = status.ToString()
 18158            }), cancellationToken);
 18159            tasks.AddRange(documents.Select(Deserialize));
 160        }
 2161        return tasks;
 2162    }
 163
 164    private async Task<UserTask?> FindByIndexAsync(string tenantId, Func<UserTask, bool> predicate, CancellationToken ca
 165    {
 1635166        foreach (var status in Enum.GetValues<UserTaskStatus>())
 167        {
 735168            var documents = await documentStore.QueryAsync(new DocumentQuery(StorageUnitName, new Dictionary<string, str
 735169            {
 735170                ["TenantId"] = tenantId,
 735171                ["Status"] = status.ToString()
 735172            }), cancellationToken);
 735173            var task = documents.Select(Deserialize).FirstOrDefault(predicate);
 735174            if (task is not null)
 3175                return task;
 176        }
 81177        return null;
 84178    }
 179
 180    private async Task<(StoredDocument Document, UserTask Task)?> LoadDocumentAsync(string tenantId, string taskId, Canc
 181    {
 11182        var document = await documentStore.LoadAsync(StorageUnitName, DocumentId(tenantId, taskId), cancellationToken);
 11183        return document is null ? null : (document, Deserialize(document));
 11184    }
 185
 85186    private static SaveDocumentRequest CreateRequest(UserTask task, long expectedVersion) => new(
 85187        StorageUnitName,
 85188        DocumentId(task.TenantId, task.Id),
 85189        JsonSerializer.Serialize(task, JsonOptions),
 85190        new Dictionary<string, string?>
 85191        {
 85192            ["TenantId"] = task.TenantId,
 85193            ["Status"] = task.Status.ToString(),
 85194            ["MaterializationKey"] = task.MaterializationKey,
 85195            ["BookmarkId"] = task.BookmarkId,
 85196            ["TaskType"] = task.TaskType,
 85197            // Every index the schema provider declares must be supplied on save; the store rejects the
 85198            // write outright when one is absent, so an omission here disables the provider entirely
 85199            // rather than merely losing an index.
 85200            ["WorkflowDefinitionId"] = task.WorkflowDefinitionId,
 85201            ["WorkflowInstanceId"] = task.WorkflowInstanceId,
 85202            ["ActivityInstanceId"] = task.ActivityInstanceId,
 85203            ["CreatedAt"] = task.CreatedAt.ToString("O", System.Globalization.CultureInfo.InvariantCulture),
 85204            ["CompletedAt"] = task.CompletedAt?.ToString("O", System.Globalization.CultureInfo.InvariantCulture),
 85205            ["AssigneeProvider"] = task.Assignee?.Provider,
 85206            ["AssigneeType"] = task.Assignee?.Type.ToString(),
 85207            ["AssigneeId"] = task.Assignee?.Id,
 85208            ["HealthSeverity"] = task.HealthSeverity?.ToString(),
 85209            ["Priority"] = task.Priority.ToString(System.Globalization.CultureInfo.InvariantCulture),
 85210            ["DueAt"] = task.DueAt?.ToString("O", System.Globalization.CultureInfo.InvariantCulture)
 85211        }, expectedVersion);
 212
 955213    private static UserTask Deserialize(StoredDocument document) => JsonSerializer.Deserialize<UserTask>(document.Conten
 955214        ?? throw new DocumentStoreValidationException($"Stored User Task document '{document.Id}' could not be deseriali
 215
 118216    private static string DocumentId(string tenantId, string taskId) => $"{tenantId}:{taskId}";
 217
 218    private static bool Matches(UserTask task, UserTaskQuery query)
 219    {
 620220        var search = query.Search?.Trim();
 620221        return task.TenantId == query.TenantId && (query.Statuses.Count == 0 || query.Statuses.Contains(task.Status)) &&
 620222               (!query.OnlyOverdue || task.IsOverdue) && (!query.OnlyWithoutDueDate || task.DueAt is null) &&
 620223               (string.IsNullOrWhiteSpace(query.TaskType) || task.TaskType == query.TaskType) &&
 620224               (!query.PriorityFrom.HasValue || task.Priority >= query.PriorityFrom) && (!query.PriorityTo.HasValue || t
 620225               (!query.DueFrom.HasValue || task.DueAt >= query.DueFrom) && (!query.DueTo.HasValue || task.DueAt <= query
 620226               (string.IsNullOrWhiteSpace(query.WorkflowDefinitionId) || task.WorkflowDefinitionId == query.WorkflowDefi
 620227               (string.IsNullOrWhiteSpace(query.WorkflowInstanceId) || task.WorkflowInstanceId == query.WorkflowInstance
 620228               (string.IsNullOrWhiteSpace(query.Reference) || task.Reference == query.Reference) &&
 620229               (string.IsNullOrWhiteSpace(search) || task.Title.Contains(search, StringComparison.OrdinalIgnoreCase) || 
 230    }
 231
 232    private static bool IsVisible(UserTask task, UserTaskQueryScope scope)
 233    {
 620234        if (!string.Equals(scope.TenantId, task.TenantId, StringComparison.Ordinal) ||
 620235            !string.Equals(scope.Subject.TenantId, task.TenantId, StringComparison.Ordinal) ||
 620236            scope.Groups.Any(group => !string.Equals(group.TenantId, task.TenantId, StringComparison.Ordinal)))
 1237            return false;
 619238        if (scope.ExcludeBlocking && task.HealthSeverity == UserTaskHealthSeverity.Blocking)
 0239            return false;
 240        // Manager-only scopes were already rejected by the policy for non-managers.
 619241        if (scope.RequiresManager)
 0242            return scope.IsManager && (scope.Kind != UserTaskQueryScopeKind.NeedsAttention || NeedsAttention(task));
 243
 619244        var subject = scope.Subject;
 619245        var groups = scope.Groups;
 619246        return scope.Kind switch
 619247        {
 0248            UserTaskQueryScopeKind.Assigned => task.Assignee?.Matches(subject) == true,
 619249            UserTaskQueryScopeKind.Available => task.IsOpen && task.Assignee is null && IsEligible(task, subject, groups
 0250            UserTaskQueryScopeKind.History => task.IsTerminal &&
 0251                                              (task.CompletedBy?.Matches(subject) == true || task.Events.Any(x => x.Acto
 0252            _ => false
 619253        };
 254    }
 255
 256    private static bool IsEligible(UserTask task, ParticipantReference subject, IReadOnlyCollection<ParticipantReference
 257    {
 622258        if (task.ExcludedUsers.Any(x => x.Matches(subject)))
 3259            return false;
 616260        if (task.MembershipResolutionMode == UserTaskMembershipResolutionMode.Snapshot)
 0261            return task.SnapshotMembers.Any(x => x.Matches(subject)) || task.SnapshotGroups.Any(x => groups.Any(x.Matche
 1232262        return task.CandidateUsers.Any(x => x.Matches(subject)) || task.CandidateGroups.Any(x => groups.Any(x.Matches));
 263    }
 264
 265    private static bool NeedsAttention(UserTask task) =>
 0266        task.HealthSeverity == UserTaskHealthSeverity.Blocking
 0267        || task.IsOverdue
 0268        || (task.IsOpen && task.Assignee is null)
 0269        || task.Status is UserTaskStatus.Completing or UserTaskStatus.TimingOut or UserTaskStatus.Cancelling;
 270
 106271    private static IEnumerable<UserTask> ApplyOrdering(IEnumerable<UserTask> tasks, UserTaskQuery query) => query.Sort.T
 106272    {
 312273        "priority" => query.Descending ? tasks.OrderByDescending(x => x.Priority).ThenBy(x => x.Id) : tasks.OrderBy(x =>
 312274        "title" => query.Descending ? tasks.OrderByDescending(x => x.Title).ThenBy(x => x.Id) : tasks.OrderBy(x => x.Tit
 494275        "due" => query.Descending ? tasks.OrderBy(x => x.DueAt == null).ThenByDescending(x => x.DueAt).ThenBy(x => x.Id)
 368276        _ => query.Descending ? tasks.OrderByDescending(x => x.CreatedAt).ThenBy(x => x.Id) : tasks.OrderBy(x => x.Creat
 106277    };
 278
 279    private static IEnumerable<UserTask> ApplyCursor(IEnumerable<UserTask> tasks, UserTaskQuery query)
 280    {
 106281        if (!TryReadCursor(query.Cursor, out var value, out var id))
 41282            return tasks;
 283        return query.Sort.ToLowerInvariant() switch
 284        {
 128285            "priority" when int.TryParse(value, out var priority) => tasks.Where(x => query.Descending ? x.Priority < pr
 112286            "title" => tasks.Where(x => query.Descending ? string.Compare(x.Title, value) < 0 || x.Title == value && str
 30287            "due" when value == "~null" => tasks.Where(x => x.DueAt == null && string.Compare(x.Id, id) > 0),
 112288            "due" when DateTimeOffset.TryParse(value, out var due) => tasks.Where(x => x.DueAt == null || query.Descendi
 132289            _ when DateTimeOffset.TryParse(value, out var created) => tasks.Where(x => query.Descending ? x.CreatedAt < 
 0290            _ => tasks
 291        };
 292    }
 293
 294    private static string CreateCursor(UserTask task, string sort)
 295    {
 66296        var value = sort.ToLowerInvariant() switch
 66297        {
 16298            "priority" => task.Priority.ToString(System.Globalization.CultureInfo.InvariantCulture),
 16299            "title" => task.Title,
 16300            "due" => task.DueAt?.ToString("O", System.Globalization.CultureInfo.InvariantCulture) ?? "~null",
 18301            _ => task.CreatedAt.ToString("O", System.Globalization.CultureInfo.InvariantCulture)
 66302        };
 66303        return Convert.ToBase64String(JsonSerializer.SerializeToUtf8Bytes(new[] { value, task.Id }, JsonOptions)).TrimEn
 304    }
 305
 306    private static bool TryReadCursor(string? cursor, out string value, out string id)
 307    {
 106308        value = id = "";
 106309        if (string.IsNullOrWhiteSpace(cursor))
 40310            return false;
 311        try
 312        {
 66313            var padded = cursor.Replace('-', '+').Replace('_', '/') + new string('=', (4 - cursor.Length % 4) % 4);
 66314            var values = JsonSerializer.Deserialize<string[]>(Convert.FromBase64String(padded), JsonOptions);
 65315            if (values is not [var parsedValue, var parsedId])
 0316                return false;
 65317            value = parsedValue;
 65318            id = parsedId;
 65319            return true;
 320        }
 0321        catch (FormatException) { return false; }
 2322        catch (JsonException) { return false; }
 66323    }
 324}
Generated by: ReportGenerator 5.5.11.0
09/09/2026 - 11:01:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Elsa.Persistence.VNext.Document.IDocumentStore)
.cctor()
GetAsync()
QueryAsync()
FindByMaterializationKeyAsync(System.String,System.String,System.Threading.CancellationToken)
FindByBookmarkIdAsync(System.String,System.String,System.Threading.CancellationToken)
FindByInvitationTokenHashAsync()
SaveAsync()
AddProjectionAsync()
AppendEventAsync()
TryMutateAsync()
LoadAllAsync()
FindByIndexAsync()
LoadDocumentAsync()
CreateRequest(Elsa.UserTasks.Models.UserTask,System.Int64)
Deserialize(Elsa.Persistence.VNext.Document.StoredDocument)
DocumentId(System.String,System.String)
Matches(Elsa.UserTasks.Models.UserTask,Elsa.UserTasks.Models.UserTaskQuery)
IsVisible(Elsa.UserTasks.Models.UserTask,Elsa.UserTasks.Models.UserTaskQueryScope)
IsEligible(Elsa.UserTasks.Models.UserTask,Elsa.UserTasks.Models.ParticipantReference,System.Collections.Generic.IReadOnlyCollection`1<Elsa.UserTasks.Models.ParticipantReference>)
NeedsAttention(Elsa.UserTasks.Models.UserTask)
ApplyOrdering(System.Collections.Generic.IEnumerable`1<Elsa.UserTasks.Models.UserTask>,Elsa.UserTasks.Models.UserTaskQuery)
ApplyCursor(System.Collections.Generic.IEnumerable`1<Elsa.UserTasks.Models.UserTask>,Elsa.UserTasks.Models.UserTaskQuery)
CreateCursor(Elsa.UserTasks.Models.UserTask,System.String)
TryReadCursor(System.String,System.String&,System.String&)