5835662074

pending completion

Build # 5835662074

Build Type

Pull #28

github

Committed by

lucaslorentz

Commit Message

Add husky and apply some code fixes

Pull Request Pull Request #28: Add husky and apply code fixes

Run Details

2502 of 2502 new or added lines in 91 files covered. (100.0%)

2297 of 2792 relevant lines covered (82.27%)

141.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.36

/src/LLL.DurableTask.EFCore/OrchestrationDbContextExtensions.cs

﻿using System;
using System.Linq;
using System.Threading.Tasks;
using DurableTask.Core;
using DurableTask.Core.Query;
using LLL.DurableTask.Core;
using LLL.DurableTask.EFCore.Entities;
using Microsoft.EntityFrameworkCore;

namespace LLL.DurableTask.EFCore;

public abstract class OrchestrationDbContextExtensions
{
    public abstract Task Migrate(OrchestrationDbContext dbContext);

    public abstract Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action);

    public async Task<T> WithinTransaction<T>(OrchestrationDbContext dbContext, Func<Task<T>> action)
    {
        T result = default;
        await WithinTransaction(dbContext, async () =>
        {
            result = await action();
        });
        return result;
    }

    public abstract Task<Instance> LockInstanceForUpdate(
        OrchestrationDbContext dbContext,
        string instanceId);

    public abstract Task<Instance> TryLockNextInstanceAsync(
        OrchestrationDbContext dbContext,
        TimeSpan lockTimeout);

    public abstract Task<Instance> TryLockNextInstanceAsync(
        OrchestrationDbContext dbContext,
        string[] queues,
        TimeSpan lockTimeout);

    public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
        OrchestrationDbContext dbContext,
        TimeSpan lockTimeout);

    public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
        OrchestrationDbContext dbContext,
        string[] queues,
        TimeSpan lockTimeout);

    public async Task PurgeOrchestrationHistoryAsync(
        OrchestrationDbContext dbContext,
        DateTime thresholdDateTimeUtc,
        OrchestrationStateTimeRangeFilterType timeRangeFilterType)
    {
        var query = timeRangeFilterType switch
        {
            OrchestrationStateTimeRangeFilterType.OrchestrationCreatedTimeFilter =>
                dbContext.Executions.Where(e => e.CreatedTime < thresholdDateTimeUtc),
            OrchestrationStateTimeRangeFilterType.OrchestrationLastUpdatedTimeFilter =>
                dbContext.Executions.Where(e => e.LastUpdatedTime < thresholdDateTimeUtc),
            OrchestrationStateTimeRangeFilterType.OrchestrationCompletedTimeFilter =>
                dbContext.Executions.Where(e => e.CompletedTime < thresholdDateTimeUtc),
            _ => throw new NotImplementedException()
        };

        await ExecuteDeleteAsync(dbContext, query);
    }

    public async Task<int> PurgeInstanceHistoryAsync(
        OrchestrationDbContext dbContext,
        string instanceId)
    {
        var query = dbContext.Executions.Where(e => e.InstanceId == instanceId);

        return await ExecuteDeleteAsync(dbContext, query);
    }

    public async Task<int> PurgeInstanceHistoryAsync(OrchestrationDbContext dbContext, PurgeInstanceFilter filter)
    {
        var query = dbContext.Executions.Where(e => e.CreatedTime >= filter.CreatedTimeFrom
            && (filter.CreatedTimeTo == null || e.CreatedTime <= filter.CreatedTimeTo)
            && (filter.RuntimeStatus == null || filter.RuntimeStatus.Contains(e.Status)));

        return await ExecuteDeleteAsync(dbContext, query);
    }

    protected virtual Task<int> ExecuteDeleteAsync<T>(OrchestrationDbContext dbContext, IQueryable<T> query)
        where T : class
    {
        return query.ExecuteDeleteAsync();
    }

    public virtual IQueryable<Execution> CreateFilteredQueryable(
        OrchestrationDbContext dbContext,
        OrchestrationQuery query)
    {
        var extendedQuery = query as ExtendedOrchestrationQuery;

        var queryable = dbContext.Executions as IQueryable<Entities.Execution>;

        if (extendedQuery != null && !extendedQuery.IncludePreviousExecutions)
            queryable = queryable.Join(dbContext.Instances, x => x.ExecutionId, x => x.LastExecutionId, (x, y) => x);

        if (!string.IsNullOrEmpty(query.InstanceIdPrefix))
            queryable = queryable.Where(e => e.InstanceId.StartsWith(query.InstanceIdPrefix));

        if (query.CreatedTimeFrom != null)
            queryable = queryable.Where(e => e.CreatedTime >= query.CreatedTimeFrom);

        if (query.CreatedTimeTo != null)
            queryable = queryable.Where(e => e.CreatedTime <= query.CreatedTimeTo);

        if (query.RuntimeStatus != null && query.RuntimeStatus.Any())
            queryable = queryable.Where(e => query.RuntimeStatus.Contains(e.Status));

        if (extendedQuery != null)
        {
            if (!string.IsNullOrEmpty(extendedQuery.NamePrefix))
                queryable = queryable.Where(e => e.Name.StartsWith(extendedQuery.NamePrefix));

            if (extendedQuery.CompletedTimeFrom != null)
                queryable = queryable.Where(e => e.CompletedTime >= extendedQuery.CompletedTimeFrom);

            if (extendedQuery.CompletedTimeTo != null)
                queryable = queryable.Where(e => e.CompletedTime <= extendedQuery.CompletedTimeTo);

            foreach (var kv in extendedQuery.Tags)
            {
                var executionsWithTag = dbContext.Executions
                    .SelectMany(e => e.Tags, (e, t) => new { e, t })
                    .Where(e => e.t.Name == kv.Key && e.t.Value == kv.Value)
                    .Select(x => new { x.e.ExecutionId });

                queryable = queryable.Join(
                    executionsWithTag,
                    x => x.ExecutionId,
                    x => x.ExecutionId,
                    (x, y) => x);
            }
        }

        var continuationToken = EFCoreContinuationToken.Parse(query.ContinuationToken);
        if (continuationToken != null)
        {
            queryable = queryable.Where(i =>
                i.CreatedTime < continuationToken.CreatedTime ||
                i.CreatedTime == continuationToken.CreatedTime && continuationToken.InstanceId.CompareTo(i.InstanceId) < 0);
        }

        return queryable;
    }
}

1	using System;
2	using System.Linq;
3	using System.Threading.Tasks;
4	using DurableTask.Core;
5	using DurableTask.Core.Query;
6	using LLL.DurableTask.Core;
7	using LLL.DurableTask.EFCore.Entities;
8	using Microsoft.EntityFrameworkCore;
9
10	namespace LLL.DurableTask.EFCore;
11
12	public abstract class OrchestrationDbContextExtensions
13	{
14	public abstract Task Migrate(OrchestrationDbContext dbContext);
15
16	public abstract Task WithinTransaction(OrchestrationDbContext dbContext, Func<Task> action);
17
18	public async Task<T> WithinTransaction<T>(OrchestrationDbContext dbContext, Func<Task<T>> action)
19	{
20	T result = default;	1,106✔
21	await WithinTransaction(dbContext, async () =>	1,106✔
22	{	1,106✔
23	result = await action();	2,212✔
24	});	1,106✔
25	return result;	1,106✔
26	}
27
28	public abstract Task<Instance> LockInstanceForUpdate(
29	OrchestrationDbContext dbContext,
30	string instanceId);
31
32	public abstract Task<Instance> TryLockNextInstanceAsync(
33	OrchestrationDbContext dbContext,
34	TimeSpan lockTimeout);
35
36	public abstract Task<Instance> TryLockNextInstanceAsync(
37	OrchestrationDbContext dbContext,
38	string[] queues,
39	TimeSpan lockTimeout);
40
41	public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
42	OrchestrationDbContext dbContext,
43	TimeSpan lockTimeout);
44
45	public abstract Task<ActivityMessage> TryLockNextActivityMessageAsync(
46	OrchestrationDbContext dbContext,
47	string[] queues,
48	TimeSpan lockTimeout);
49
50	public async Task PurgeOrchestrationHistoryAsync(
51	OrchestrationDbContext dbContext,
52	DateTime thresholdDateTimeUtc,
53	OrchestrationStateTimeRangeFilterType timeRangeFilterType)
54	{
55	var query = timeRangeFilterType switch	24✔
56	{	24✔
57	OrchestrationStateTimeRangeFilterType.OrchestrationCreatedTimeFilter =>	24✔
58	dbContext.Executions.Where(e => e.CreatedTime < thresholdDateTimeUtc),	48✔
59	OrchestrationStateTimeRangeFilterType.OrchestrationLastUpdatedTimeFilter =>	24✔
60	dbContext.Executions.Where(e => e.LastUpdatedTime < thresholdDateTimeUtc),	24✔
61	OrchestrationStateTimeRangeFilterType.OrchestrationCompletedTimeFilter =>	24✔
62	dbContext.Executions.Where(e => e.CompletedTime < thresholdDateTimeUtc),	24✔
63	_ => throw new NotImplementedException()	24✔
64	};	24✔
65
66	await ExecuteDeleteAsync(dbContext, query);	24✔
67	}
68
69	public async Task<int> PurgeInstanceHistoryAsync(
70	OrchestrationDbContext dbContext,
71	string instanceId)
72	{
73	var query = dbContext.Executions.Where(e => e.InstanceId == instanceId);	6✔
74
75	return await ExecuteDeleteAsync(dbContext, query);	6✔
76	}
77
78	public async Task<int> PurgeInstanceHistoryAsync(OrchestrationDbContext dbContext, PurgeInstanceFilter filter)
79	{
80	var query = dbContext.Executions.Where(e => e.CreatedTime >= filter.CreatedTimeFrom	×
81	&& (filter.CreatedTimeTo == null \|\| e.CreatedTime <= filter.CreatedTimeTo)	×
82	&& (filter.RuntimeStatus == null \|\| filter.RuntimeStatus.Contains(e.Status)));	×
83
84	return await ExecuteDeleteAsync(dbContext, query);	×
85	}
86
87	protected virtual Task<int> ExecuteDeleteAsync<T>(OrchestrationDbContext dbContext, IQueryable<T> query)
88	where T : class
89	{
90	return query.ExecuteDeleteAsync();	21✔
91	}
92
93	public virtual IQueryable<Execution> CreateFilteredQueryable(
94	OrchestrationDbContext dbContext,
95	OrchestrationQuery query)
96	{
97	var extendedQuery = query as ExtendedOrchestrationQuery;	7✔
98
99	var queryable = dbContext.Executions as IQueryable<Entities.Execution>;	7✔
100
101	if (extendedQuery != null && !extendedQuery.IncludePreviousExecutions)	7✔
102	queryable = queryable.Join(dbContext.Instances, x => x.ExecutionId, x => x.LastExecutionId, (x, y) => x);	7✔
103
104	if (!string.IsNullOrEmpty(query.InstanceIdPrefix))	7✔
105	queryable = queryable.Where(e => e.InstanceId.StartsWith(query.InstanceIdPrefix));	×
106
107	if (query.CreatedTimeFrom != null)	7✔
108	queryable = queryable.Where(e => e.CreatedTime >= query.CreatedTimeFrom);	×
109
110	if (query.CreatedTimeTo != null)	7✔
111	queryable = queryable.Where(e => e.CreatedTime <= query.CreatedTimeTo);	×
112
113	if (query.RuntimeStatus != null && query.RuntimeStatus.Any())	7✔
114	queryable = queryable.Where(e => query.RuntimeStatus.Contains(e.Status));	×
115
116	if (extendedQuery != null)	7✔
117	{
118	if (!string.IsNullOrEmpty(extendedQuery.NamePrefix))	7✔
119	queryable = queryable.Where(e => e.Name.StartsWith(extendedQuery.NamePrefix));	×
120
121	if (extendedQuery.CompletedTimeFrom != null)	7✔
122	queryable = queryable.Where(e => e.CompletedTime >= extendedQuery.CompletedTimeFrom);	×
123
124	if (extendedQuery.CompletedTimeTo != null)	7✔
125	queryable = queryable.Where(e => e.CompletedTime <= extendedQuery.CompletedTimeTo);	×
126
127	foreach (var kv in extendedQuery.Tags)	34✔
128	{
129	var executionsWithTag = dbContext.Executions	10✔
130	.SelectMany(e => e.Tags, (e, t) => new { e, t })	10✔
131	.Where(e => e.t.Name == kv.Key && e.t.Value == kv.Value)	10✔
132	.Select(x => new { x.e.ExecutionId });	10✔
133
134	queryable = queryable.Join(	10✔
135	executionsWithTag,	10✔
136	x => x.ExecutionId,	10✔
137	x => x.ExecutionId,	10✔
138	(x, y) => x);	10✔
139	}
140	}
141
142	var continuationToken = EFCoreContinuationToken.Parse(query.ContinuationToken);	7✔
143	if (continuationToken != null)	7✔
144	{
145	queryable = queryable.Where(i =>	1✔
146	i.CreatedTime < continuationToken.CreatedTime \|\|	1✔
147	i.CreatedTime == continuationToken.CreatedTime && continuationToken.InstanceId.CompareTo(i.InstanceId) < 0);	1✔
148	}
149
150	return queryable;	7✔
151	}
152	}

lucaslorentz / durabletask-extensions / 5835662074

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous