Added MongoDB implementation for alteration storage (#5555)

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)
                 {

src/modules/Elsa.MongoDb/Elsa.MongoDb.csproj

@@ -16,6 +16,8 @@
 
     <ItemGroup>
         <ProjectReference Include="..\..\common\Elsa.Features\Elsa.Features.csproj" />
+        <ProjectReference Include="..\Elsa.Alterations.Core\Elsa.Alterations.Core.csproj" />
+        <ProjectReference Include="..\Elsa.Alterations\Elsa.Alterations.csproj" />
         <ProjectReference Include="..\Elsa.Common\Elsa.Common.csproj" />
         <ProjectReference Include="..\Elsa.Http\Elsa.Http.csproj" />
         <ProjectReference Include="..\Elsa.Identity\Elsa.Identity.csproj" />

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;
+
+/// <summary>
+/// A MongoDb implementation of <see cref="IAlterationJobStore"/>.
+/// </summary>
+public class MongoAlterationJobStore : IAlterationJobStore
+{
+    private readonly MongoDbStore<AlterationJob> _mongoDbStore;
+
+    /// <summary>
+    /// Constructor.
+    /// </summary>
+    public MongoAlterationJobStore(MongoDbStore<AlterationJob> mongoDbStore)
+    {
+        _mongoDbStore = mongoDbStore;
+    }
+
+    /// <inheritdoc />
+    public async Task<long> CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
+    {
+        return await _mongoDbStore.CountAsync(queryable => Filter(queryable, filter), cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task<AlterationJob?> FindAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
+    {
+        return await _mongoDbStore.FindAsync(queryable => Filter(queryable, filter), cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task<IEnumerable<AlterationJob>> FindManyAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
+    {
+        return await _mongoDbStore.FindManyAsync(queryable => Filter(queryable, filter), cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task<IEnumerable<string>> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
+    {
+        var documents = await _mongoDbStore.FindManyAsync(query => Filter(query, filter), selector => selector.Id, cancellationToken);
+        return documents;
+    }
+
+    /// <inheritdoc />
+    public async Task SaveAsync(AlterationJob job, CancellationToken cancellationToken = default)
+    {
+        await _mongoDbStore.SaveAsync(job, cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task SaveManyAsync(IEnumerable<AlterationJob> jobs, CancellationToken cancellationToken = default)
+    {
+        await _mongoDbStore.SaveManyAsync(jobs.Select(i => i), cancellationToken);
+    }
+
+    private static IMongoQueryable<AlterationJob> Filter(IMongoQueryable<AlterationJob> queryable, AlterationJobFilter filter) =>
+        (filter.Apply(queryable) as IMongoQueryable<AlterationJob>)!;
+}

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;
+
+/// <summary>
+/// A MongoDb implementation of <see cref="IAlterationPlanStore"/>.
+/// </summary>
+public class MongoAlterationPlanStore : IAlterationPlanStore
+{
+    private readonly MongoDbStore<AlterationPlanDocument> _mongoDbStore;
+    private readonly IAlterationSerializer _alterationSerializer;
+
+    /// <summary>
+    /// Constructor.
+    /// </summary>
+    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<MongoDbStore<AlterationPlanDocument>>();
+        _alterationSerializer = alterationSerializer;
+    }
+
+    /// <inheritdoc />
+    public async Task<long> CountAsync(AlterationPlanFilter filter, CancellationToken cancellationToken = default)
+    {
+        return await _mongoDbStore.CountAsync(queryable => Filter(queryable, filter), cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public async Task<AlterationPlan?> FindAsync(AlterationPlanFilter filter, CancellationToken cancellationToken = default)
+    {
+        var document = await _mongoDbStore.FindAsync(queryable => Filter(queryable, filter), cancellationToken);
+
+        if (document == null) return null;
+
+        return Map(document);
+    }
+
+    /// <inheritdoc />
+    public async Task SaveAsync(AlterationPlan plan, CancellationToken cancellationToken = default)
+    {
+        var document = Map(plan);
+
+        await _mongoDbStore.SaveAsync(document, cancellationToken);
+    }
+
+    private static IMongoQueryable<AlterationPlanDocument> Filter(IMongoQueryable<AlterationPlanDocument> queryable, AlterationPlanFilter filter)
+    {
+        return (Apply(queryable, filter) as IMongoQueryable<AlterationPlanDocument>)!;
+    }
+
+    private static IQueryable<AlterationPlanDocument> Apply(IQueryable<AlterationPlanDocument> 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
+        };
+    }
+}

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<IMongoCollection<AlterationPlanDocument>>();
+        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<IMongoCollection<AlterationJob>>();
+        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));
+    }
+}

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!;
+}

src/modules/Elsa.MongoDb/Modules/Alterations/Extensions.cs

@@ -0,0 +1,20 @@
+using Elsa.Alterations.Features;
+using JetBrains.Annotations;
+
+namespace Elsa.MongoDb.Modules.Alterations;
+
+/// <summary>
+/// Provides extensions to the <see cref="AlterationsFeature"/> feature.
+/// </summary>
+[PublicAPI]
+public static class Extensions
+{
+    /// <summary>
+    /// Configures the <see cref="AlterationsFeature"/> to use the <see cref="MongoAlterationsPersistenceFeature"/>.
+    /// </summary>
+    public static AlterationsFeature UseMongoDb(this AlterationsFeature feature, Action<MongoAlterationsPersistenceFeature>? configure = default)
+    {
+        feature.Module.Configure(configure);
+        return feature;
+    }
+}

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<AlterationsFeature>(feature =>
+        {
+            feature.AlterationPlanStoreFactory = sp => sp.GetRequiredService<MongoAlterationPlanStore>();
+            feature.AlterationJobStoreFactory = sp => sp.GetRequiredService<MongoAlterationJobStore>();
+        });
+    }
+
+    public override void Apply()
+    {
+        base.Apply();
+
+        AddCollection<AlterationJob>("alteration_jobs");
+        AddStore<AlterationJob, MongoAlterationJobStore>();
+
+        Services.AddHostedService<CreateIndices>();
+    }
+}