diff --git a/Elsa.sln b/Elsa.sln index 8ea9c84cc..24c3411d2 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -19,35 +19,37 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Abstractions", "src\co EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "core", "core", "{35F44BE9-13D0-417C-A01A-F0787BEE6DC3}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Console", "src\activities\Elsa.Activities.Console\Elsa.Activities.Console.csproj", "{B98B80CB-D3A3-437C-A61C-ED82C8C77B2F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Console", "src\activities\Elsa.Activities.Console\Elsa.Activities.Console.csproj", "{B98B80CB-D3A3-437C-A61C-ED82C8C77B2F}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Http", "src\activities\Elsa.Activities.Http\Elsa.Activities.Http.csproj", "{1D63E1B4-2386-4BEC-9090-C3F3CBF25703}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Http", "src\activities\Elsa.Activities.Http\Elsa.Activities.Http.csproj", "{1D63E1B4-2386-4BEC-9090-C3F3CBF25703}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "activities", "activities", "{B43B546E-23F3-46E8-ACB7-D04F05CDA180}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Email", "src\activities\Elsa.Activities.Email\Elsa.Activities.Email.csproj", "{D20FCB88-9DCA-49EA-9CC2-5B94DD935D30}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Email", "src\activities\Elsa.Activities.Email\Elsa.Activities.Email.csproj", "{D20FCB88-9DCA-49EA-9CC2-5B94DD935D30}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Timers", "src\activities\Elsa.Activities.Timers\Elsa.Activities.Timers.csproj", "{E4B71DC4-3E73-49C3-9B4E-CA13909F222D}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Timers", "src\activities\Elsa.Activities.Timers\Elsa.Activities.Timers.csproj", "{E4B71DC4-3E73-49C3-9B4E-CA13909F222D}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.MassTransit", "src\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj", "{C4939482-9447-47D1-B6AF-E9F8C6323841}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.MassTransit", "src\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj", "{C4939482-9447-47D1-B6AF-E9F8C6323841}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "persistence", "persistence", "{76710230-AC6D-4E54-9A06-15F34C155118}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.YesSql", "src\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj", "{4DB0741F-EAD9-4E37-862D-A1DA12462B7D}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.YesSql", "src\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj", "{4DB0741F-EAD9-4E37-862D-A1DA12462B7D}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.UserTask", "src\activities\Elsa.Activities.UserTask\Elsa.Activities.UserTask.csproj", "{BC0F57E2-AB64-4026-B6E3-E03D9133DB26}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.UserTask", "src\activities\Elsa.Activities.UserTask\Elsa.Activities.UserTask.csproj", "{BC0F57E2-AB64-4026-B6E3-E03D9133DB26}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.AutoMapper.Extensions", "src\core\Elsa.AutoMapper.Extensions\Elsa.AutoMapper.Extensions.csproj", "{6CEF6AFB-635D-4FB4-B9EC-0806DD95E1BC}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.AutoMapper.Extensions", "src\core\Elsa.AutoMapper.Extensions\Elsa.AutoMapper.Extensions.csproj", "{6CEF6AFB-635D-4FB4-B9EC-0806DD95E1BC}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.EntityFrameworkCore", "src\persistence\Elsa.Persistence.EntityFrameworkCore\Elsa.Persistence.EntityFrameworkCore.csproj", "{349D4C87-E17F-4F51-B493-1994BF631D37}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.EntityFrameworkCore", "src\persistence\Elsa.Persistence.EntityFrameworkCore\Elsa.Persistence.EntityFrameworkCore.csproj", "{349D4C87-E17F-4F51-B493-1994BF631D37}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "dashboard", "dashboard", "{8B87D79A-F0DE-4F88-81D3-8C8C69108ADB}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Dashboard", "src\dashboard\Elsa.Dashboard\Elsa.Dashboard.csproj", "{493FC3E3-34FE-4F8D-AF37-38A5F5541FBD}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Dashboard", "src\dashboard\Elsa.Dashboard\Elsa.Dashboard.csproj", "{493FC3E3-34FE-4F8D-AF37-38A5F5541FBD}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.MongoDb", "src\persistence\Elsa.Persistence.MongoDb\Elsa.Persistence.MongoDb.csproj", "{82CDD590-2409-4C7E-8AF4-07C7D718D4D8}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.MongoDb", "src\persistence\Elsa.Persistence.MongoDb\Elsa.Persistence.MongoDb.csproj", "{82CDD590-2409-4C7E-8AF4-07C7D718D4D8}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.WorkflowDesigner", "src\dashboard\Elsa.WorkflowDesigner\Elsa.WorkflowDesigner.csproj", "{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.WorkflowDesigner", "src\dashboard\Elsa.WorkflowDesigner\Elsa.WorkflowDesigner.csproj", "{120E4BF9-4A42-4BB9-A0B2-9597160D6E37}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Persistence.DocumentDb", "src\persistence\Elsa.Persistence.DocumentDb\Elsa.Persistence.DocumentDb.csproj", "{39EBECE7-538B-431C-AAB7-901AB16E6D4A}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -111,6 +113,10 @@ Global {120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Debug|Any CPU.Build.0 = Debug|Any CPU {120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Release|Any CPU.ActiveCfg = Release|Any CPU {120E4BF9-4A42-4BB9-A0B2-9597160D6E37}.Release|Any CPU.Build.0 = Release|Any CPU + {39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Debug|Any CPU.Build.0 = Debug|Any CPU + {39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Release|Any CPU.ActiveCfg = Release|Any CPU + {39EBECE7-538B-431C-AAB7-901AB16E6D4A}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -119,9 +125,9 @@ Global {300EE2D5-54C5-46F2-AD03-BB43589EA074} = {35F44BE9-13D0-417C-A01A-F0787BEE6DC3} {3130C574-557C-4DA2-933E-AED7A9EBE48A} = {35F44BE9-13D0-417C-A01A-F0787BEE6DC3} {35F44BE9-13D0-417C-A01A-F0787BEE6DC3} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} - {B43B546E-23F3-46E8-ACB7-D04F05CDA180} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} {B98B80CB-D3A3-437C-A61C-ED82C8C77B2F} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} {1D63E1B4-2386-4BEC-9090-C3F3CBF25703} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} + {B43B546E-23F3-46E8-ACB7-D04F05CDA180} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} {D20FCB88-9DCA-49EA-9CC2-5B94DD935D30} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} {E4B71DC4-3E73-49C3-9B4E-CA13909F222D} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} {C4939482-9447-47D1-B6AF-E9F8C6323841} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} @@ -134,6 +140,7 @@ Global {493FC3E3-34FE-4F8D-AF37-38A5F5541FBD} = {8B87D79A-F0DE-4F88-81D3-8C8C69108ADB} {82CDD590-2409-4C7E-8AF4-07C7D718D4D8} = {76710230-AC6D-4E54-9A06-15F34C155118} {120E4BF9-4A42-4BB9-A0B2-9597160D6E37} = {8B87D79A-F0DE-4F88-81D3-8C8C69108ADB} + {39EBECE7-538B-431C-AAB7-901AB16E6D4A} = {76710230-AC6D-4E54-9A06-15F34C155118} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs new file mode 100644 index 000000000..67ad695ff --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs @@ -0,0 +1,80 @@ +using Microsoft.Azure.Documents; +using Microsoft.Azure.Documents.Client; +using Newtonsoft.Json; +using Newtonsoft.Json.Serialization; +using System; +using System.Threading.Tasks; + +namespace Elsa.Persistence.DocumentDb +{ + public class DocumentDbStorage + { + internal DocumentDbStorageOptions Options { get; } + + internal DocumentClient Client { get; } + + internal Uri CollectionUri { get; private set; } + + public DocumentDbStorage(string url, string authSecret, string database, string collection, DocumentDbStorageOptions options = null) + { + Options = options ?? new DocumentDbStorageOptions(); + Options.DatabaseName = database; + Options.CollectionName = collection; + + JsonSerializerSettings settings = new JsonSerializerSettings + { + NullValueHandling = NullValueHandling.Ignore, + DateTimeZoneHandling = DateTimeZoneHandling.Utc, + ContractResolver = new CamelCasePropertyNamesContractResolver + { + NamingStrategy = new CamelCaseNamingStrategy(false, false) + } + }; + + ConnectionPolicy connectionPolicy = ConnectionPolicy.Default; + connectionPolicy.ConnectionMode = Options.ConnectionMode; + connectionPolicy.ConnectionProtocol = Options.ConnectionProtocol; + connectionPolicy.RequestTimeout = Options.RequestTimeout; + connectionPolicy.RetryOptions = new RetryOptions + { + MaxRetryWaitTimeInSeconds = 10, + MaxRetryAttemptsOnThrottledRequests = 5 + }; + + Client = new DocumentClient(new Uri(url), authSecret, settings, connectionPolicy); + Task task = Client.OpenAsync(); + Task continueTask = task.ContinueWith(t => Initialize(), TaskContinuationOptions.OnlyOnRanToCompletion); + continueTask.Wait(); + } + + /// + /// Return the name of the database + /// + /// + public override string ToString() => $"DoucmentDb Database : {Options.DatabaseName}"; + + private void Initialize() + { + //_logger.Info($"Creating database : {Options.DatabaseName}"); + var databaseTask = Client.CreateDatabaseIfNotExistsAsync(new Database { Id = Options.DatabaseName }); + + // create document collection + Task> collectionTask = databaseTask.ContinueWith(t => + { + //_logger.Info($"Creating document collection : {t.Result.Resource.Id}"); + Uri databaseUri = UriFactory.CreateDatabaseUri(t.Result.Resource.Id); + return Client.CreateDocumentCollectionIfNotExistsAsync(databaseUri, new DocumentCollection { Id = Options.CollectionName }); + }, TaskContinuationOptions.OnlyOnRanToCompletion).Unwrap(); + + var continueTask = collectionTask.ContinueWith(t => + { + CollectionUri = UriFactory.CreateDocumentCollectionUri(Options.DatabaseName, t.Result.Resource.Id); + }, TaskContinuationOptions.OnlyOnRanToCompletion); + continueTask.Wait(); + if (continueTask.IsFaulted || continueTask.IsCanceled) + { + throw new ApplicationException("Unable to setup the storage database", databaseTask.Exception); + } + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs new file mode 100644 index 000000000..70c934e45 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs @@ -0,0 +1,68 @@ +using Microsoft.Azure.Documents.Client; +using System; + +namespace Elsa.Persistence.DocumentDb +{ + public class DocumentDbStorageOptions + { + /// + /// Gets or sets the name of the database. + /// + /// + /// The name of the database. + /// + internal string DatabaseName { get; set; } + + /// + /// Gets or sets the name of the collection. + /// + /// + /// The name of the collection. + /// + internal string CollectionName { get; set; } + + /// + /// Get or sets the request timeout for DocumentDB client. Default value set to 30 seconds + /// + public TimeSpan RequestTimeout { get; set; } + + /// + /// Get or set the interval timespan to process expired entries. Default value 15 minutes + /// Expired items under "locks", "jobs", "lists", "sets", "hashs", "counters/aggregated" will be checked + /// + public TimeSpan ExpirationCheckInterval { get; set; } + + /// + /// Get or sets the interval timespan to aggregated the counters. Default value 1 minute + /// + public TimeSpan CountersAggregateInterval { get; set; } + + /// + /// Gets or sets the interval timespan to poll the queue for processing any new jobs. Default value 2 minutes + /// + public TimeSpan QueuePollInterval { get; set; } + + /// + /// Gets or sets the connection mode for the DocumentDB client. Default value is Direct. + /// + public ConnectionMode ConnectionMode { get; set; } + + /// + /// Gets or sets the connection protocol for the DocumentDB client. Default value is TCP. + /// + public Protocol ConnectionProtocol { get; set; } + + /// + /// Create an instance of AzureDocumentDB Storage option with default values + /// + public DocumentDbStorageOptions() + { + RequestTimeout = TimeSpan.FromSeconds(30); + ExpirationCheckInterval = TimeSpan.FromMinutes(2); + CountersAggregateInterval = TimeSpan.FromMinutes(2); + QueuePollInterval = TimeSpan.FromSeconds(15); + ConnectionMode = ConnectionMode.Direct; + ConnectionProtocol = Protocol.Tcp; + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs b/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs new file mode 100644 index 000000000..dd98a2168 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs @@ -0,0 +1,36 @@ +using Elsa.Models; +using Newtonsoft.Json; +using System; +using System.Collections.Generic; +using System.Text; + +namespace Elsa.Persistence.DocumentDb.Documents +{ + public class WorkflowDefinitionVersionDocument + { + [JsonProperty(PropertyName = "id")] + public string Id { get; set; } + [JsonProperty(PropertyName = "type")] + public string Type { get; } = nameof(WorkflowDefinitionVersionDocument); + [JsonProperty(PropertyName = "definitionId")] + public string DefinitionId { get; set; } + [JsonProperty(PropertyName = "version")] + public int Version { get; set; } + [JsonProperty(PropertyName = "name")] + public string Name { get; set; } + [JsonProperty(PropertyName = "description")] + public string Description { get; set; } + [JsonProperty(PropertyName = "activities")] + public IList Activities { get; set; } + [JsonProperty(PropertyName = "connections")] + public IList Connections { get; set; } + [JsonProperty(PropertyName = "variables")] + public Variables Variables { get; set; } + [JsonProperty(PropertyName = "isSingleton")] + public bool IsSingleton { get; set; } + [JsonProperty(PropertyName = "isPublished")] + public bool IsPublished { get; set; } + [JsonProperty(PropertyName = "isLatest")] + public bool IsLatest { get; set; } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs b/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs new file mode 100644 index 000000000..fde7f87e8 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs @@ -0,0 +1,45 @@ +using Elsa.Models; +using Newtonsoft.Json; +using System; +using System.Collections.Generic; + +namespace Elsa.Persistence.DocumentDb.Documents +{ + public class WorkflowInstanceDocument + { + [JsonProperty(PropertyName = "id")] + public string Id { get; set; } + [JsonProperty(PropertyName = "definitionId")] + public string DefinitionId { get; set; } + [JsonProperty(PropertyName = "type")] + public string Type { get; } = nameof(WorkflowInstanceDocument); + [JsonProperty(PropertyName = "version")] + public int Version { get; set; } + [JsonProperty(PropertyName = "status")] + public WorkflowStatus Status { get; set; } + [JsonProperty(PropertyName = "correlationId")] + public string CorrelationId { get; set; } + [JsonProperty(PropertyName = "createdAt")] + public DateTime CreatedAt { get; set; } + [JsonProperty(PropertyName = "startedAt")] + public DateTime? StartedAt { get; set; } + [JsonProperty(PropertyName = "finishedAt")] + public DateTime? FinishedAt { get; set; } + [JsonProperty(PropertyName = "faultedAt")] + public DateTime? FaultedAt { get; set; } + [JsonProperty(PropertyName = "abortedAt")] + public DateTime? AbortedAt { get; set; } + [JsonProperty(PropertyName = "activites")] + public IDictionary Activities { get; set; } = new Dictionary(); + [JsonProperty(PropertyName = "scopes")] + public Stack Scopes { get; set; } + [JsonProperty(PropertyName = "input")] + public Variables Input { get; set; } + [JsonProperty(PropertyName = "blockingActivities")] + public HashSet BlockingActivities { get; set; } + [JsonProperty(PropertyName = "executionLog")] + public ICollection ExecutionLog { get; set; } + [JsonProperty(PropertyName = "fault")] + public WorkflowFault Fault { get; set; } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj b/src/persistence/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj new file mode 100644 index 000000000..ae591cbd7 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj @@ -0,0 +1,18 @@ + + + + netstandard2.0 + netcoreapp2.2 + latest + 1.0.0 + + + + + + + + + + + diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Extensions/ServiceCollectionExtensions.cs b/src/persistence/Elsa.Persistence.DocumentDb/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 000000000..6f19db4b5 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,32 @@ +using Elsa.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Elsa.Persistence.DocumentDb.Services; +using System; + +namespace Elsa.Persistence.DocumentDb.Extensions +{ + public static class ServiceCollectionExtensions + { + public static IServiceCollection UseAzureCosmosDbAsWorkflowStore( + this IServiceCollection services, string url, string authSecret, string database, string collection, DocumentDbStorageOptions options = null) + { + if (string.IsNullOrEmpty(url)) throw new ArgumentNullException(nameof(url)); + if (string.IsNullOrEmpty(authSecret)) throw new ArgumentNullException(nameof(authSecret)); + + var storage = new DocumentDbStorage(url, authSecret, database, collection, options); + + return services + .AddSingleton(storage); + } + + public static IServiceCollection AddCosmosDbWorkflowInstanceStore(this IServiceCollection services) + { + return services.Replace(ServiceLifetime.Transient); + } + + public static IServiceCollection AddCosmosDbWorkflowDefinitionStore(this IServiceCollection services) + { + return services.Replace(ServiceLifetime.Transient); + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs b/src/persistence/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs new file mode 100644 index 000000000..c2e5a83cf --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs @@ -0,0 +1,28 @@ +using Elsa.Models; +using Elsa.Persistence.DocumentDb.Documents; +using System.Linq; + +namespace Elsa.Persistence.DocumentDb.Extensions +{ + public static class WorkflowDefinitionDocumentExtensions + { + public static IQueryable WithVersion( + this IQueryable query, + VersionOptions version) + { + + if (version.IsDraft) + query = query.Where(x => !x.IsPublished); + if (version.IsLatest) + query = query.OrderByDescending(x => x.Version); + if (version.IsPublished) + query = query.Where(x => x.IsPublished); + if (version.IsLatestOrPublished) + query = query.Where(x => x.IsPublished || x.IsLatest); + if (version.Version > 0) + query = query.Where(x => x.Version == version.Version); + + return query.OrderByDescending(x => x.Version); + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs new file mode 100644 index 000000000..84e5f3447 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs @@ -0,0 +1,146 @@ +using Microsoft.Azure.Documents; +using Microsoft.Azure.Documents.Client; +using Microsoft.Azure.Documents.Linq; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Persistence.DocumentDb.Helpers +{ + internal static class ClientHelper + { + /// + /// Creates a document as an asynchronous operation in the Azure Cosmos DB service. + /// + /// + /// the URI of the document collection to create the document in. + /// the document object. + /// The request options for the request. + /// Disables the automatic id generation, will throw an exception if id is missing. + /// (Optional) representing request cancellation. + /// + internal static Task> CreateDocumentWithRetriesAsync(this DocumentClient client, Uri documentCollectionUri, object document, RequestOptions options = null, bool disableAutomaticIdGeneration = false, CancellationToken cancellationToken = default(CancellationToken)) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.CreateDocumentAsync(documentCollectionUri, document, options, disableAutomaticIdGeneration, cancellationToken)), cancellationToken); + } + + /// + /// Reads a as a generic type T from the Azure Cosmos DB service as an asynchronous operation. + /// + /// + /// + /// A URI to the Document resource to be read. + /// The request options for the request. + /// (Optional) representing request cancellation. + /// + internal static Task> ReadDocumentWithRetriesAsync(this DocumentClient client, Uri documentUri, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken)) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.ReadDocumentAsync(documentUri, options, cancellationToken)), cancellationToken); + } + + /// + /// Upserts a document as an asynchronous operation in the Azure Cosmos DB service. + /// + /// + /// the URI of the document collection to upsert the document in. + /// the document object. + /// The request options for the request. + /// Disables the automatic id generation, will throw an exception if id is missing. + /// (Optional) representing request cancellation. + internal static Task> UpsertDocumentWithRetriesAsync(this DocumentClient client, Uri documentCollectionUri, object document, RequestOptions options = null, bool disableAutomaticIdGeneration = false, CancellationToken cancellationToken = default(CancellationToken)) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.UpsertDocumentAsync(documentCollectionUri, document, options, disableAutomaticIdGeneration, cancellationToken)), cancellationToken); + } + + /// + /// Delete a document as an asynchronous operation from the Azure Cosmos DB service. + /// + /// + /// the URI of the document to delete. + /// The request options for the request. + /// (Optional) representing request cancellation. + internal static Task> DeleteDocumentWithRetriesAsync(this DocumentClient client, Uri documentUri, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken)) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.DeleteDocumentAsync(documentUri, options, cancellationToken)), cancellationToken); + } + + /// + /// Replaces a document as an asynchronous operation in the Azure Cosmos DB service. + /// + /// + /// the URI of the document to be updated. + /// the updated document. + /// The request options for the request. + /// (Optional) representing request cancellation. + /// + internal static Task> ReplaceDocumentWithRetriesAsync(this DocumentClient client, Uri documentUri, object document, RequestOptions options = null, CancellationToken cancellationToken = default(CancellationToken)) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.ReplaceDocumentAsync(documentUri, document, options, cancellationToken)), cancellationToken); + } + + /// + /// Executes a stored procedure against a collection as an asynchronous operation from the Azure Cosmos DB service. + /// + /// + /// + /// the URI of the stored procedure to be executed. + /// the parameters for the stored procedure execution. + /// + internal static Task> ExecuteStoredProcedureWithRetriesAsync(this DocumentClient client, Uri storedProcedureUri, params object[] procedureParams) + { + return Task.Run(async () => await client.ExecuteWithRetries(() => client.ExecuteStoredProcedureAsync(storedProcedureUri, procedureParams))); + } + + /// + /// Execute the function with retries on throttle + /// + internal static async Task> ExecuteNextWithRetriesAsync(this IDocumentQuery query) + { + while (true) + { + TimeSpan timeSpan; + + try + { + return await query.ExecuteNextAsync(); + } + catch (DocumentClientException ex) when (ex.StatusCode != null && (int)ex.StatusCode == 429) + { + timeSpan = ex.RetryAfter; + } + catch (AggregateException ex) when (ex.InnerException is DocumentClientException de && de.StatusCode != null && (int)de.StatusCode == 429) + { + timeSpan = de.RetryAfter; + } + + await Task.Delay(timeSpan); + } + } + + /// + /// Execute the function with retries on throttle + /// + internal static async Task ExecuteWithRetries(this DocumentClient client, Func> function) + { + while (true) + { + TimeSpan timeSpan; + + try + { + return await function(); + } + catch (DocumentClientException ex) when (ex.StatusCode != null && (int)ex.StatusCode == 429) + { + timeSpan = ex.RetryAfter; + } + catch (AggregateException ex) when (ex.InnerException is DocumentClientException de && de.StatusCode != null && (int)de.StatusCode == 429) + { + timeSpan = de.RetryAfter; + } + + await Task.Delay(timeSpan); + } + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs new file mode 100644 index 000000000..37f0e2fc1 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs @@ -0,0 +1,28 @@ +using Microsoft.Azure.Documents.Client; +using Microsoft.Azure.Documents.Linq; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace Elsa.Persistence.DocumentDb.Helpers +{ + internal static class QueryHelper + { + internal static List ToQueryResult(this IQueryable source) + { + IDocumentQuery query = source.AsDocumentQuery(); + List results = new List(); + + while (query.HasMoreResults) + { + Task> task = Task.Run(async () => await query.ExecuteNextWithRetriesAsync()); + task.Wait(); + results.AddRange(task.Result); + } + + return results; + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs new file mode 100644 index 000000000..03860295b --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs @@ -0,0 +1,28 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Text; + +namespace Elsa.Persistence.DocumentDb.Helpers +{ + internal static class TimeHelper + { + private static readonly DateTime epochDateTime = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc); + + internal static int ToEpoch(this DateTime date) + { + if (date.Equals(DateTime.MinValue)) return int.MinValue; + TimeSpan epochTimeSpan = date - epochDateTime; + return (int)epochTimeSpan.TotalSeconds; + } + + internal static DateTime ToDateTime(this int totalSeconds) => epochDateTime.AddSeconds(totalSeconds); + + internal static string TryParseToEpoch(this string s) + { + return DateTime.TryParse(s, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out DateTime date) + ? date.ToEpoch().ToString(CultureInfo.InvariantCulture) + : s; + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs b/src/persistence/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs new file mode 100644 index 000000000..df3c3f431 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs @@ -0,0 +1,15 @@ +using AutoMapper; +using Elsa.Models; +using Elsa.Persistence.DocumentDb.Documents; + +namespace Elsa.Persistence.DocumentDb.Mapping +{ + public class DocumentProfile : Profile + { + public DocumentProfile() + { + CreateMap().ReverseMap(); + CreateMap().ReverseMap(); + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs b/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs new file mode 100644 index 000000000..2644a26db --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs @@ -0,0 +1,94 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AutoMapper; +using Elsa.AutoMapper.Extensions.NodaTime; +using Elsa.Extensions; +using Elsa.Models; +using Elsa.Persistence.DocumentDb.Documents; +using Elsa.Persistence.DocumentDb.Extensions; +using Elsa.Persistence.DocumentDb.Helpers; +using Elsa.Persistence.DocumentDb.Mapping; + +namespace Elsa.Persistence.DocumentDb.Services +{ + public class CosmosDbWorkflowDefinitionStore : IWorkflowDefinitionStore + { + readonly IMapper _mapper; + readonly DocumentDbStorage _storage; + + public CosmosDbWorkflowDefinitionStore(DocumentDbStorage storage) + { + _storage = storage; + var configuration = new MapperConfiguration(cfg => + { + cfg.AddProfile(); + cfg.AddProfile(); + }); + _mapper = configuration.CreateMapper(); + } + + public async Task AddAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) + { + var document = Map(definition); + var client = _storage.Client; + await client.CreateDocumentWithRetriesAsync(_storage.CollectionUri, document); + } + + public async Task DeleteAsync(string id, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var records = client.CreateDocumentQuery(_storage.CollectionUri).Where(c => c.DefinitionId == id).ToQueryResult(); + foreach (var record in records) + { + await client.DeleteDocumentAsync(record.Id); + } + return records.Count; + } + + public Task GetByIdAsync(string id, VersionOptions version, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.DefinitionId == id).WithVersion(version); + var document = query.AsEnumerable().FirstOrDefault(); + return Task.FromResult(Map(document)); + } + + public Task> ListAsync(VersionOptions version, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .WithVersion(version).ToList(); + + return Task.FromResult(_mapper.Map>(query)); + } + + public async Task SaveAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) + { + var document = Map(definition); + var client = _storage.Client; + await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document); + return definition; + } + + public async Task UpdateAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) + { + var document = Map(definition); + var client = _storage.Client; + await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document); + return Map(document); + } + + private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source) + { + return _mapper.Map(source); + } + + private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source) + { + return _mapper.Map(source); + } + } +} diff --git a/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs b/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs new file mode 100644 index 000000000..3598117f0 --- /dev/null +++ b/src/persistence/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs @@ -0,0 +1,114 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AutoMapper; +using Elsa.AutoMapper.Extensions.NodaTime; +using Elsa.Extensions; +using Elsa.Models; +using Elsa.Persistence.DocumentDb.Documents; +using Elsa.Persistence.DocumentDb.Helpers; +using Elsa.Persistence.DocumentDb.Mapping; + +namespace Elsa.Persistence.DocumentDb.Services +{ + public class CosmosDbWorkflowInstanceStore : IWorkflowInstanceStore + { + readonly IMapper _mapper; + readonly DocumentDbStorage _storage; + + public CosmosDbWorkflowInstanceStore(DocumentDbStorage storage) + { + var configuration = new MapperConfiguration(cfg => + { + cfg.AddProfile(); + cfg.AddProfile(); + }); + _mapper = configuration.CreateMapper(); + _storage = storage; + } + + public async Task DeleteAsync(string id, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.Id == id); + await client.DeleteDocumentAsync(id); + } + + public Task GetByCorrelationIdAsync(string correlationId, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.CorrelationId == correlationId); + var document = query.AsEnumerable().FirstOrDefault(); + return Task.FromResult(Map(document)); + } + + public Task GetByIdAsync(string id, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.Id == id); + var document = query.AsEnumerable().FirstOrDefault(); + return Task.FromResult(Map(document)); + } + + public Task> ListAllAsync(CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri); + return Task.FromResult(_mapper.Map>(query)); + } + + public Task> ListByBlockingActivityAsync(string activityType, string correlationId = null, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri).Where(x => x.Status == WorkflowStatus.Executing); + if (!string.IsNullOrWhiteSpace(correlationId)) + { + query = query.Where(x => x.CorrelationId == correlationId); + } + query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType)); + var instances = Map(query.ToList()); + return Task.FromResult(instances.GetBlockingActivities(activityType)); + } + + public Task> ListByDefinitionAsync(string definitionId, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.DefinitionId == definitionId); + return Task.FromResult(Map(query.ToList())); + } + + public Task> ListByStatusAsync(string definitionId, WorkflowStatus status, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.DefinitionId == definitionId && c.Status == status); + return Task.FromResult(Map(query.ToList())); + } + + public Task> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken = default) + { + var client = _storage.Client; + var query = client.CreateDocumentQuery(_storage.CollectionUri) + .Where(c => c.Status == status); + return Task.FromResult(Map(query.ToList())); + } + + public async Task SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) + { + var document = Map(instance); + var client = _storage.Client; + await client.UpsertDocumentWithRetriesAsync(_storage.CollectionUri, document); + } + + private WorkflowInstanceDocument Map(WorkflowInstance source) => _mapper.Map(source); + private WorkflowInstance Map(WorkflowInstanceDocument source) => _mapper.Map(source); + + private IEnumerable Map(IEnumerable source) => + _mapper.Map>(source); + } +}