diff --git a/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs b/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs deleted file mode 100644 index 3da860753..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorage.cs +++ /dev/null @@ -1,57 +0,0 @@ -using Microsoft.Azure.Documents; -using Microsoft.Azure.Documents.Client; -using Newtonsoft.Json; -using Newtonsoft.Json.Serialization; -using System; -using System.Threading; -using System.Threading.Tasks; - -namespace Elsa.Persistence.DocumentDb -{ - public class DocumentDbStorage - { - private DocumentDbStorageOptions Options { get; } - internal DocumentClient Client { get; } - - public DocumentDbStorage(DocumentDbStorageOptions options) - { - Options = options; - - var settings = new JsonSerializerSettings - { - NullValueHandling = NullValueHandling.Ignore, - DateTimeZoneHandling = DateTimeZoneHandling.Utc, - ContractResolver = new CamelCasePropertyNamesContractResolver - { - NamingStrategy = new CamelCaseNamingStrategy(false, false) - } - }; - - var 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(options.Url, options.Secret, settings, connectionPolicy); - } - - public override string ToString() => $"DocumentDb Database: {Options.DatabaseName}"; - - public async Task GetCollectionAsync(string collectionName, CancellationToken cancellationToken = default) - { - var database = await Client.CreateDatabaseIfNotExistsAsync(new Database { Id = Options.DatabaseName }); - var databaseUri = UriFactory.CreateDatabaseUri(database.Resource.Id); - - var collection = await Client.CreateDocumentCollectionIfNotExistsAsync( - databaseUri, - new DocumentCollection { Id = collectionName }); - - return UriFactory.CreateDocumentCollectionUri(Options.DatabaseName, collection.Resource.Id); - } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs b/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs deleted file mode 100644 index f67fa5019..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/DocumentDbStorageOptions.cs +++ /dev/null @@ -1,62 +0,0 @@ -using Microsoft.Azure.Documents.Client; -using System; -using System.ComponentModel.DataAnnotations; - -namespace Elsa.Persistence.DocumentDb -{ - public class DocumentDbStorageOptions - { - [Required] - public Uri Url { get; set; } - - [Required] - public string Secret { get; set; } - - [Required] - public string DatabaseName { 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/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs b/src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs deleted file mode 100644 index ef1ce0274..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowDefinitionVersionDocument.cs +++ /dev/null @@ -1,44 +0,0 @@ -using Elsa.Models; -using Newtonsoft.Json; -using System.Collections.Generic; - -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 = "isDisabled")] - public bool IsDisabled { get; set; } - - [JsonProperty(PropertyName = "isPublished")] - public bool IsPublished { get; set; } - - [JsonProperty(PropertyName = "isLatest")] - public bool IsLatest { get; set; } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs b/src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs deleted file mode 100644 index c785c6b38..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Documents/WorkflowInstanceDocument.cs +++ /dev/null @@ -1,61 +0,0 @@ -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 = "activities")] - public IDictionary Activities { get; set; } = new Dictionary(); - - [JsonProperty(PropertyName = "variables")] - public Variables Variables { get; set; } - - [JsonProperty(PropertyName = "input")] public Variable? Input { get; set; } - [JsonProperty(PropertyName = "input")] public Variable? Output { get; set; } - - [JsonProperty(PropertyName = "blockingActivities")] - public HashSet BlockingActivities { get; set; } - - [JsonProperty(PropertyName = "scheduledActivities")] - public Stack ScheduledActivities { get; set; } - - [JsonProperty(PropertyName = "executionLog")] - public ICollection ExecutionLog { get; set; } - - [JsonProperty(PropertyName = "fault")] public WorkflowFault Fault { get; set; } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj b/src/providers/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj deleted file mode 100644 index 4f603ca45..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Elsa.Persistence.DocumentDb.csproj +++ /dev/null @@ -1,36 +0,0 @@ - - - - netstandard2.0 - 8.0 - 1.0.0 - Elsa Contributors - - Elsa is a set of workflow libraries and tools that enable super-fast workflowing capabilities in any .NET Core application. - This package provides a CosmosDb persistence provider. - - 2019 - https://github.com/elsa-workflows/elsa-core - https://github.com/elsa-workflows/elsa-core - GitHub - elsa, workflows, cosmosdb - icon.png - enable - - - - - True - - - - - - - - - - - - - diff --git a/src/providers/Elsa.Persistence.DocumentDb/Extensions/ElsaOptionsExtensions.cs b/src/providers/Elsa.Persistence.DocumentDb/Extensions/ElsaOptionsExtensions.cs deleted file mode 100644 index 30c61b90e..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Extensions/ElsaOptionsExtensions.cs +++ /dev/null @@ -1,48 +0,0 @@ -using Elsa.Persistence.DocumentDb.Services; -using Elsa.Extensions; -using Elsa.Mapping; -using Elsa.Persistence.DocumentDb.Mapping; -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.Persistence.DocumentDb.Extensions -{ - public static class ElsaOptionsExtensions - { - public static ElsaOptions UseCosmosDbWorkflowDefinitionStore(this ElsaOptions options, DocumentDbStorageOptions dbOptions) - { - options - .AddCosmosDbProvider(dbOptions) - .UseWorkflowDefinitionStore(sp => sp.GetRequiredService()); - - options.Services.AddSingleton(); - return options; - } - - public static ElsaOptions UseCosmosDbWorkflowInstanceStore(this ElsaOptions options, DocumentDbStorageOptions dbOptions) - { - options - .AddCosmosDbProvider(dbOptions) - .UseWorkflowInstanceStore(sp => sp.GetRequiredService()); - - options.Services.AddSingleton(); - return options; - } - - private static ElsaOptions AddCosmosDbProvider( - this ElsaOptions options, - DocumentDbStorageOptions documentDbOptions) - { - if (options.HasService()) - return options; - - var storage = new DocumentDbStorage(documentDbOptions); - - options.Services - .AddSingleton(storage) - .AddAutoMapperProfile(ServiceLifetime.Singleton) - .AddAutoMapperProfile(ServiceLifetime.Singleton); - - return options; - } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs b/src/providers/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs deleted file mode 100644 index ab04569a2..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Extensions/WorkflowDefinitionDocumentExtensions.cs +++ /dev/null @@ -1,31 +0,0 @@ -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); - else if (version.IsLatest) - query = query.Where(x => x.IsLatest); - else if (version.IsPublished) - query = query.Where(x => x.IsPublished); - else if (version.IsLatestOrPublished) - query = query.Where(x => x.IsPublished || x.IsLatest); - else if (version.AllVersions) - { - // Nothing to filter. - } - else if (version.Version > 0) - query = query.Where(x => x.Version == version.Version); - - return query.OrderByDescending(x => x.Version); - } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs b/src/providers/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs deleted file mode 100644 index 092a7d63d..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Helpers/ClientHelper.cs +++ /dev/null @@ -1,205 +0,0 @@ -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) - { - 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) - { - 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) - { - 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) - { - 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) - { - 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); - } - } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs b/src/providers/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs deleted file mode 100644 index fd609139d..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Helpers/QueryHelper.cs +++ /dev/null @@ -1,24 +0,0 @@ -using Microsoft.Azure.Documents.Linq; -using System.Collections.Generic; -using System.Linq; -using System.Threading.Tasks; - -namespace Elsa.Persistence.DocumentDb.Helpers -{ - internal static class QueryHelper - { - internal static async Task> ToQueryResultAsync(this IQueryable source) - { - var query = source.AsDocumentQuery(); - var results = new List(); - - while (query.HasMoreResults) - { - var nextResults = await Task.Run(async () => await query.ExecuteNextWithRetriesAsync()); - results.AddRange(nextResults); - } - - return results; - } - } -} diff --git a/src/providers/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs b/src/providers/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs deleted file mode 100644 index a6525f243..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Helpers/TimeHelper.cs +++ /dev/null @@ -1,26 +0,0 @@ -using System; -using System.Globalization; - -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; - var 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 var date) - ? date.ToEpoch().ToString(CultureInfo.InvariantCulture) - : s; - } - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs b/src/providers/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs deleted file mode 100644 index df3c3f431..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Mapping/DocumentProfile.cs +++ /dev/null @@ -1,15 +0,0 @@ -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/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs b/src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs deleted file mode 100644 index 01c9b591f..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowDefinitionStore.cs +++ /dev/null @@ -1,114 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using AutoMapper; -using Elsa.Models; -using Elsa.Persistence.DocumentDb.Documents; -using Elsa.Persistence.DocumentDb.Extensions; -using Elsa.Persistence.DocumentDb.Helpers; - -namespace Elsa.Persistence.DocumentDb.Services -{ - public class CosmosDbWorkflowDefinitionStore : IWorkflowDefinitionStore - { - private readonly IMapper mapper; - private readonly DocumentDbStorage storage; - private Uri? collectionUrl; - - public CosmosDbWorkflowDefinitionStore(DocumentDbStorage storage, IMapper mapper) - { - this.storage = storage; - this.mapper = mapper; - collectionUrl = default; - } - - public async Task AddAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) - { - var document = Map(definition); - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - await client.CreateDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken); - return Map(document); - } - - public async Task GetByIdAsync(string id, CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl).Where(c => c.Id == id); - var document = query.FirstOrDefault(); - return Map(document); - } - - public async Task GetByIdAsync(string definitionId, VersionOptions version, CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.DefinitionId == definitionId).WithVersion(version); - var document = query.AsEnumerable().FirstOrDefault(); - return Map(document); - } - - public async Task DeleteAsync(string id, CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var workflowDefinitionDocuments = await client.CreateDocumentQuery(collectionUrl).Where(c => c.DefinitionId == id).ToQueryResultAsync(); - foreach (var record in workflowDefinitionDocuments) - { - await client.DeleteDocumentAsync(record.Id, cancellationToken: cancellationToken); - } - return workflowDefinitionDocuments.Count; - } - - public async Task> ListAsync(VersionOptions version, CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client - .CreateDocumentQuery(collectionUrl) - .WithVersion(version).ToList(); - - return mapper.Map>(query); - } - - public async Task SaveAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) - { - var document = Map(definition); - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - await client.UpsertDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken); - return definition; - } - - public async Task UpdateAsync(WorkflowDefinitionVersion definition, CancellationToken cancellationToken = default) - { - var document = Map(definition); - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - await client.UpsertDocumentWithRetriesAsync(collectionUrl, document, cancellationToken: cancellationToken); - return Map(document); - } - - private async Task GetCollectionUriAsync(CancellationToken cancellationToken) - { - if (collectionUrl == null) - collectionUrl = await storage.GetCollectionAsync("WorkflowDefinitions", cancellationToken); - - return collectionUrl; - } - - private WorkflowDefinitionVersionDocument Map(WorkflowDefinitionVersion source) - { - return mapper.Map(source); - } - - private WorkflowDefinitionVersion Map(WorkflowDefinitionVersionDocument source) - { - return mapper.Map(source); - } - } -} diff --git a/src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs b/src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs deleted file mode 100644 index 4aa641b45..000000000 --- a/src/providers/Elsa.Persistence.DocumentDb/Services/CosmosDbWorkflowInstanceStore.cs +++ /dev/null @@ -1,173 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using AutoMapper; -using Elsa.Extensions; -using Elsa.Models; -using Elsa.Persistence.DocumentDb.Documents; -using Elsa.Persistence.DocumentDb.Helpers; - -namespace Elsa.Persistence.DocumentDb.Services -{ - public class CosmosDbWorkflowInstanceStore : IWorkflowInstanceStore - { - private readonly IMapper mapper; - private readonly DocumentDbStorage storage; - private Uri? collectionUrl; - - public CosmosDbWorkflowInstanceStore(DocumentDbStorage storage, IMapper mapper) - { - this.storage = storage; - this.mapper = mapper; - collectionUrl = default; - } - - public async Task DeleteAsync( - string id, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - await client.DeleteDocumentAsync(id, cancellationToken: cancellationToken); - } - - public async Task GetByCorrelationIdAsync( - string correlationId, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.CorrelationId == correlationId); - var document = query.AsEnumerable().FirstOrDefault(); - return Map(document); - } - - public async Task GetByIdAsync( - string id, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.Id == id); - var document = query.AsEnumerable().FirstOrDefault(); - return Map(document); - } - - public async Task> ListAllAsync(CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client - .CreateDocumentQuery(collectionUrl) - .OrderByDescending(x => x.CreatedAt); - return mapper.Map>(query); - } - public async Task> ListByBlockingActivityTagAsync( - string activityType, - string tag, - string? correlationId = null, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client - .CreateDocumentQuery(collectionUrl) - .Where(x => x.Status == WorkflowStatus.Suspended); - - if (!string.IsNullOrWhiteSpace(correlationId)) - query = query.Where(x => x.CorrelationId == correlationId); - - query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType && y.Tag == tag)); - query = query.OrderByDescending(x => x.CreatedAt); - - var instances = Map(query.ToList()); - return instances.GetBlockingActivities(activityType); - } - - public async Task> ListByBlockingActivityAsync( - string activityType, - string? correlationId = null, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client - .CreateDocumentQuery(collectionUrl) - .Where(x => x.Status == WorkflowStatus.Suspended); - - if (!string.IsNullOrWhiteSpace(correlationId)) - query = query.Where(x => x.CorrelationId == correlationId); - - query = query.Where(x => x.BlockingActivities.Any(y => y.ActivityType == activityType)); - query = query.OrderByDescending(x => x.CreatedAt); - - var instances = Map(query.ToList()); - return instances.GetBlockingActivities(activityType); - } - - public async Task> ListByDefinitionAsync( - string definitionId, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.DefinitionId == definitionId) - .OrderByDescending(x => x.CreatedAt); - return Map(query.ToList()); - } - - public async Task> ListByStatusAsync( - string definitionId, - WorkflowStatus status, - CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.DefinitionId == definitionId && c.Status == status) - .OrderByDescending(x => x.CreatedAt); - return Map(query.ToList()); - } - - public async Task> ListByStatusAsync(WorkflowStatus status, CancellationToken cancellationToken = default) - { - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var query = client.CreateDocumentQuery(collectionUrl) - .Where(c => c.Status == status) - .OrderByDescending(x => x.CreatedAt); - return Map(query.ToList()); - } - - public async Task SaveAsync(WorkflowInstance instance, CancellationToken cancellationToken = default) - { - var document = Map(instance); - var client = storage.Client; - var collectionUrl = await GetCollectionUriAsync(cancellationToken); - var response = await client.UpsertDocumentWithRetriesAsync( - collectionUrl, - document, - cancellationToken: cancellationToken); - - document = (dynamic)response.Resource; - return Map(document); - } - - private async Task GetCollectionUriAsync(CancellationToken cancellationToken) - { - if (collectionUrl == null) - collectionUrl = await storage.GetCollectionAsync("WorkflowInstances", cancellationToken); - - return collectionUrl; - } - - private WorkflowInstanceDocument Map(WorkflowInstance source) => mapper.Map(source); - private WorkflowInstance Map(WorkflowInstanceDocument source) => mapper.Map(source); - - private IEnumerable Map(IEnumerable source) => mapper.Map>(source); - } -} \ No newline at end of file diff --git a/src/providers/Elsa.Persistence.DocumentDb/icon.png b/src/providers/Elsa.Persistence.DocumentDb/icon.png deleted file mode 100644 index f978a2882..000000000 Binary files a/src/providers/Elsa.Persistence.DocumentDb/icon.png and /dev/null differ