From da7bff83b7a48101050dedbb17143145c2a3fee2 Mon Sep 17 00:00:00 2001 From: rosca-sabina <43028000+rosca-sabina@users.noreply.github.com> Date: Sun, 9 Jun 2024 16:18:30 +0300 Subject: [PATCH] Added MongoDB implementation for alteration storage (#5555) --- src/bundles/Elsa.Server.Web/Program.cs | 3 +- src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj | 2 + .../Modules/Alterations/AlterationJobStore.cs | 63 ++++++++++++ .../Alterations/AlterationPlanStore.cs | 95 +++++++++++++++++++ .../Modules/Alterations/CreateIndices.cs | 62 ++++++++++++ .../Documents/AlterationPlanDocument.cs | 20 ++++ .../Modules/Alterations/Extensions.cs | 20 ++++ .../Modules/Alterations/Feature.cs | 31 ++++++ 8 files changed, 295 insertions(+), 1 deletion(-) create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/AlterationJobStore.cs create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/AlterationPlanStore.cs create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/CreateIndices.cs create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/Documents/AlterationPlanDocument.cs create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/Extensions.cs create mode 100644 src/modules/Elsa.MongoDb/Modules/Alterations/Feature.cs diff --git a/src/bundles/Elsa.Server.Web/Program.cs b/src/bundles/Elsa.Server.Web/Program.cs index 3bb7b67f3..f55311e24 100644 --- a/src/bundles/Elsa.Server.Web/Program.cs +++ b/src/bundles/Elsa.Server.Web/Program.cs @@ -16,6 +16,7 @@ using Elsa.Features.Services; using Elsa.Http.Options; using Elsa.MassTransit.Extensions; using Elsa.MongoDb.Extensions; +using Elsa.MongoDb.Modules.Alterations; using Elsa.MongoDb.Modules.Identity; using Elsa.MongoDb.Modules.Management; using Elsa.MongoDb.Modules.Runtime; @@ -285,7 +286,7 @@ services { if (useMongoDb) { - // TODO: alterations.UseMongoDb(); + alterations.UseMongoDb(); } else if (useDapper) { diff --git a/src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj b/src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj index 0899cec24..91fc3bdff 100644 --- a/src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj +++ b/src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj @@ -16,6 +16,8 @@ + + diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationJobStore.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationJobStore.cs new file mode 100644 index 000000000..ba609c4cb --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationJobStore.cs @@ -0,0 +1,63 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Core.Filters; +using Elsa.MongoDb.Common; +using MongoDB.Driver.Linq; + +namespace Elsa.MongoDb.Modules.Alterations; + +/// +/// A MongoDb implementation of . +/// +public class MongoAlterationJobStore : IAlterationJobStore +{ + private readonly MongoDbStore _mongoDbStore; + + /// + /// Constructor. + /// + public MongoAlterationJobStore(MongoDbStore mongoDbStore) + { + _mongoDbStore = mongoDbStore; + } + + /// + public async Task CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + return await _mongoDbStore.CountAsync(queryable => Filter(queryable, filter), cancellationToken); + } + + /// + public async Task FindAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + return await _mongoDbStore.FindAsync(queryable => Filter(queryable, filter), cancellationToken); + } + + /// + public async Task> FindManyAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + return await _mongoDbStore.FindManyAsync(queryable => Filter(queryable, filter), cancellationToken); + } + + /// + public async Task> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + var documents = await _mongoDbStore.FindManyAsync(query => Filter(query, filter), selector => selector.Id, cancellationToken); + return documents; + } + + /// + public async Task SaveAsync(AlterationJob job, CancellationToken cancellationToken = default) + { + await _mongoDbStore.SaveAsync(job, cancellationToken); + } + + /// + public async Task SaveManyAsync(IEnumerable jobs, CancellationToken cancellationToken = default) + { + await _mongoDbStore.SaveManyAsync(jobs.Select(i => i), cancellationToken); + } + + private static IMongoQueryable Filter(IMongoQueryable queryable, AlterationJobFilter filter) => + (filter.Apply(queryable) as IMongoQueryable)!; +} \ No newline at end of file diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationPlanStore.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationPlanStore.cs new file mode 100644 index 000000000..76de65739 --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/AlterationPlanStore.cs @@ -0,0 +1,95 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Core.Filters; +using Elsa.MongoDb.Common; +using Elsa.MongoDb.Modules.Alterations.Documents; +using Microsoft.Extensions.DependencyInjection; +using MongoDB.Driver; +using MongoDB.Driver.Linq; + +namespace Elsa.MongoDb.Modules.Alterations; + +/// +/// A MongoDb implementation of . +/// +public class MongoAlterationPlanStore : IAlterationPlanStore +{ + private readonly MongoDbStore _mongoDbStore; + private readonly IAlterationSerializer _alterationSerializer; + + /// + /// Constructor. + /// + public MongoAlterationPlanStore(IServiceProvider serviceProvider, + IAlterationSerializer alterationSerializer) + { + // Resolved from IServiceProvider instead of injecting through the constructor so AlterationPlanDocument can be internal instead of public. + _mongoDbStore = serviceProvider.GetRequiredService>(); + _alterationSerializer = alterationSerializer; + } + + /// + public async Task CountAsync(AlterationPlanFilter filter, CancellationToken cancellationToken = default) + { + return await _mongoDbStore.CountAsync(queryable => Filter(queryable, filter), cancellationToken); + } + + /// + public async Task FindAsync(AlterationPlanFilter filter, CancellationToken cancellationToken = default) + { + var document = await _mongoDbStore.FindAsync(queryable => Filter(queryable, filter), cancellationToken); + + if (document == null) return null; + + return Map(document); + } + + /// + public async Task SaveAsync(AlterationPlan plan, CancellationToken cancellationToken = default) + { + var document = Map(plan); + + await _mongoDbStore.SaveAsync(document, cancellationToken); + } + + private static IMongoQueryable Filter(IMongoQueryable queryable, AlterationPlanFilter filter) + { + return (Apply(queryable, filter) as IMongoQueryable)!; + } + + private static IQueryable Apply(IQueryable queryable, AlterationPlanFilter filter) + { + if (!string.IsNullOrWhiteSpace(filter.Id)) + queryable = queryable.Where(x => x.Id == filter.Id); + + return queryable; + } + + private AlterationPlan Map(AlterationPlanDocument document) + { + return new AlterationPlan + { + Id = document.Id, + Alterations = _alterationSerializer.DeserializeMany(document.Alterations).ToList(), + WorkflowInstanceFilter = document.WorkflowInstanceFilter, + Status = document.Status, + CreatedAt = document.CreatedAt, + StartedAt = document.StartedAt, + CompletedAt = document.CompletedAt + }; + } + + private AlterationPlanDocument Map(AlterationPlan plan) + { + return new AlterationPlanDocument + { + Id = plan.Id, + Alterations = _alterationSerializer.SerializeMany(plan.Alterations), + WorkflowInstanceFilter = plan.WorkflowInstanceFilter, + Status = plan.Status, + CreatedAt = plan.CreatedAt, + StartedAt = plan.StartedAt, + CompletedAt = plan.CompletedAt + }; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/CreateIndices.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/CreateIndices.cs new file mode 100644 index 000000000..51ed1a48f --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/CreateIndices.cs @@ -0,0 +1,62 @@ +using Elsa.Alterations.Core.Entities; +using Elsa.MongoDb.Helpers; +using Elsa.MongoDb.Modules.Alterations.Documents; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using MongoDB.Driver; + +namespace Elsa.MongoDb.Modules.Alterations; +internal class CreateIndices(IServiceProvider serviceProvider) : IHostedService +{ + public Task StartAsync(CancellationToken cancellationToken) + { + using var scope = serviceProvider.CreateScope(); + + return Task.WhenAll( + CreateAlterationPlanIndices(scope, cancellationToken), + CreateAlterationJobIndices(scope, cancellationToken)); + } + + public Task StopAsync(CancellationToken cancellationToken) + { + return Task.CompletedTask; + } + + private static Task CreateAlterationPlanIndices(IServiceScope serviceScope, CancellationToken cancellationToken) + { + var alterationPlanCollection = serviceScope.ServiceProvider.GetService>(); + if (alterationPlanCollection == null) return Task.CompletedTask; + + return IndexHelpers.CreateAsync( + alterationPlanCollection, + async (collection, indexBuilder) => + await collection.Indexes.CreateManyAsync( + [ + new(indexBuilder.Ascending(x => x.Status)), + new(indexBuilder.Ascending(x => x.CreatedAt)), + new(indexBuilder.Ascending(x => x.StartedAt)), + new(indexBuilder.Ascending(x => x.CompletedAt)) + ], + cancellationToken)); + } + + private static Task CreateAlterationJobIndices(IServiceScope serviceScope, CancellationToken cancellationToken) + { + var alterationJobCollection = serviceScope.ServiceProvider.GetService>(); + if (alterationJobCollection == null) return Task.CompletedTask; + + return IndexHelpers.CreateAsync( + alterationJobCollection, + async (collection, indexBuilder) => + await collection.Indexes.CreateManyAsync( + [ + new(indexBuilder.Ascending(x => x.PlanId)), + new(indexBuilder.Ascending(x => x.WorkflowInstanceId)), + new(indexBuilder.Ascending(x => x.Status)), + new(indexBuilder.Ascending(x => x.CreatedAt)), + new(indexBuilder.Ascending(x => x.StartedAt)), + new(indexBuilder.Ascending(x => x.CompletedAt)) + ], + cancellationToken)); + } +} diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/Documents/AlterationPlanDocument.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/Documents/AlterationPlanDocument.cs new file mode 100644 index 000000000..5bc5e5e6b --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/Documents/AlterationPlanDocument.cs @@ -0,0 +1,20 @@ +using Elsa.Alterations.Core.Enums; +using Elsa.Alterations.Core.Models; + +namespace Elsa.MongoDb.Modules.Alterations.Documents; +internal class AlterationPlanDocument +{ + public string Id { get; init; } = default!; + + public string Alterations { get; init; } = default!; + + public AlterationWorkflowInstanceFilter WorkflowInstanceFilter { get; init; } = default!; + + public AlterationPlanStatus Status { get; init; } = default!; + + public DateTimeOffset CreatedAt { get; init; } = default!; + + public DateTimeOffset? StartedAt { get; init; } = default!; + + public DateTimeOffset? CompletedAt { get; init; } = default!; +} diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/Extensions.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/Extensions.cs new file mode 100644 index 000000000..c946490c9 --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/Extensions.cs @@ -0,0 +1,20 @@ +using Elsa.Alterations.Features; +using JetBrains.Annotations; + +namespace Elsa.MongoDb.Modules.Alterations; + +/// +/// Provides extensions to the feature. +/// +[PublicAPI] +public static class Extensions +{ + /// + /// Configures the to use the . + /// + public static AlterationsFeature UseMongoDb(this AlterationsFeature feature, Action? configure = default) + { + feature.Module.Configure(configure); + return feature; + } +} diff --git a/src/modules/Elsa.MongoDb/Modules/Alterations/Feature.cs b/src/modules/Elsa.MongoDb/Modules/Alterations/Feature.cs new file mode 100644 index 000000000..4a093c688 --- /dev/null +++ b/src/modules/Elsa.MongoDb/Modules/Alterations/Feature.cs @@ -0,0 +1,31 @@ +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Features; +using Elsa.Features.Attributes; +using Elsa.Features.Services; +using Elsa.MongoDb.Common; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.MongoDb.Modules.Alterations; + +[DependsOn(typeof(AlterationsFeature))] +public class MongoAlterationsPersistenceFeature(IModule module) : PersistenceFeatureBase(module) +{ + public override void Configure() + { + Module.Configure(feature => + { + feature.AlterationPlanStoreFactory = sp => sp.GetRequiredService(); + feature.AlterationJobStoreFactory = sp => sp.GetRequiredService(); + }); + } + + public override void Apply() + { + base.Apply(); + + AddCollection("alteration_jobs"); + AddStore(); + + Services.AddHostedService(); + } +}