Mongodb implementation

Elsa.sln

@@ -200,6 +200,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.My
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.AspNet.DslWorkflowProvider", "src\samples\aspnet\Elsa.Samples.AspNet.DslWorkflowProvider\Elsa.Samples.AspNet.DslWorkflowProvider.csproj", "{4BC5D3D2-9425-41D7-9804-D145B92520EE}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.MongoDB", "src\modules\Elsa.MongoDB\Elsa.MongoDB.csproj", "{F84DD219-8693-4F1E-8C65-59B673E70DD5}"
+EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "persistence", "persistence", "{9B4F139F-7D26-435C-A561-89E65A67A8E5}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "scripting", "scripting", "{6EF07978-A6D2-40EB-891D-7D70C5F37E76}"
@@ -538,6 +540,10 @@ Global
  {4BC5D3D2-9425-41D7-9804-D145B92520EE}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {4BC5D3D2-9425-41D7-9804-D145B92520EE}.Release|Any CPU.ActiveCfg = Release|Any CPU
  {4BC5D3D2-9425-41D7-9804-D145B92520EE}.Release|Any CPU.Build.0 = Release|Any CPU
+ {F84DD219-8693-4F1E-8C65-59B673E70DD5}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {F84DD219-8693-4F1E-8C65-59B673E70DD5}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {F84DD219-8693-4F1E-8C65-59B673E70DD5}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {F84DD219-8693-4F1E-8C65-59B673E70DD5}.Release|Any CPU.Build.0 = Release|Any CPU
  {4D8F3BB2-709D-481A-8FD3-0DA9762B366D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
  {4D8F3BB2-709D-481A-8FD3-0DA9762B366D}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {4D8F3BB2-709D-481A-8FD3-0DA9762B366D}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -659,5 +665,6 @@ Global
  {A86DF0F0-24E9-470E-984B-9E7332544972} = {9B4F139F-7D26-435C-A561-89E65A67A8E5}
  {A702DD84-B55C-4807-9C28-D8F5AC3CCA78} = {B08B4E00-C2AB-48F3-8389-449F42AEF179}
  {9C9FEB8E-A253-4705-A344-49BFEFC9A516} = {873BFC3E-63C2-4495-A503-5EC05DCD84E4}
+ {F84DD219-8693-4F1E-8C65-59B673E70DD5} = {9B4F139F-7D26-435C-A561-89E65A67A8E5}
  EndGlobalSection
 EndGlobal

src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj

@@ -16,6 +16,7 @@
  <ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore\Elsa.EntityFrameworkCore.csproj" />
  <ProjectReference Include="..\..\modules\Elsa.Environments\Elsa.Environments.csproj" />
  <ProjectReference Include="..\..\modules\Elsa.MassTransit\Elsa.MassTransit.csproj" />
+ <ProjectReference Include="..\..\modules\Elsa.MongoDB\Elsa.MongoDB.csproj" />
  <ProjectReference Include="..\..\modules\Elsa.ProtoActor\Elsa.ProtoActor.csproj" />
  <ProjectReference Include="..\..\modules\Elsa.Identity\Elsa.Identity.csproj" />
  <ProjectReference Include="..\..\modules\Elsa.ProtoActor.Cluster.AzureContainerApps\Elsa.ProtoActor.Cluster.AzureContainerApps.csproj" />

src/bundles/Elsa.WorkflowServer.Web/Program.cs

@@ -5,6 +5,11 @@
 using Elsa.EntityFrameworkCore.Modules.Runtime;
 using Elsa.Http.Handlers;
 using Elsa.JavaScript.Options;
+using Elsa.MongoDB.Extensions;
+using Elsa.MongoDB.Modules.Identity;
+using Elsa.MongoDB.Modules.Labels;
+using Elsa.MongoDB.Modules.Management;
+using Elsa.MongoDB.Modules.Runtime;
 using Elsa.WorkflowServer.Web;
 
 EndpointSecurityOptions.DisableSecurity();
@@ -24,6 +29,7 @@
  .AddTypeAlias<ApiResponse<User>>("ApiResponse[User]")
  .UseIdentity(identity =>
  {
+ identity.UseMongoDb();
  identity.IdentityOptions = options => identitySection.Bind(options);
  identity.TokenOptions = options => identityTokenSection.Bind(options);
  identity.UseConfigurationBasedUserProvider(options => identitySection.Bind(options));
@@ -33,15 +39,15 @@
  .UseDefaultAuthentication()
  .UseWorkflowManagement(management =>
  {
- management.UseEntityFrameworkCore(m => m.UseSqlite(sqliteConnectionString));
+ management.UseMongoDb();
  management.AddVariableType<ApiResponse<User>>("Api");
  management.AddVariableType<User>("Api");
  management.AddVariableType<Support>("Api");
  })
  .UseWorkflowRuntime(runtime =>
  {
- runtime.UseDefaultRuntime(dr => dr.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
- runtime.UseExecutionLogRecords(e => e.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
+ runtime.UseDefaultRuntime(dr => dr.UseMongoDb());
+ runtime.UseExecutionLogRecords(e => e.UseMongoDb());
  runtime.UseAsyncWorkflowStateExporter();
  runtime.UseMassTransitDispatcher();
  })
@@ -50,8 +56,10 @@
  .UseWorkflowsApi(api => api.AddFastEndpointsAssembly<Program>())
  .UseJavaScript()
  .UseLiquid()
+ .UseLabels(options => options.UseMongoDb())
  .UseHttp(http => http.HttpEndpointAuthorizationHandler = sp => sp.GetRequiredService<AllowAnonymousHttpEndpointAuthorizationHandler>())
  .UseEmail(email => email.ConfigureOptions = options => configuration.GetSection("Smtp").Bind(options))
+ .UseMongoDb(options => configuration.GetSection("MongoDb").Bind(options))
  );
 
 services.Configure<JintOptions>(options => options.AllowClrAccess = true);

src/bundles/Elsa.WorkflowServer.Web/appsettings.json

@@ -15,6 +15,10 @@
  "ConnectionStrings": {
  "Sqlite": "Data Source=elsa.sqlite.db;Cache=Shared;"
  },
+ "MongoDb": {
+ "ConnectionString": "mongodb:https://localhost:27017",
+ "DatabaseName": "elsa-workflows"
+ },
  "Smtp": {
  "Host": "localhost",
  "Port": 2525,

src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowDefinitionStore.cs

@@ -193,8 +193,7 @@ private IQueryable<WorkflowDefinition> Paginate(IQueryable<WorkflowDefinition> q
  if (pageArgs?.Limit != null) queryable = queryable.Take(pageArgs.Limit.Value);
  return queryable;
  }
-
-
+
  private class WorkflowDefinitionState
  {
  public WorkflowDefinitionState()

src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/Configurations.cs

@@ -26,18 +26,16 @@ public void Configure(EntityTypeBuilder<WorkflowState> builder)
  builder.Ignore(x => x.Fault);
  builder.Ignore(x => x.Output);
  builder.Property<string>("Data");
- builder.Property<DateTimeOffset>("CreatedAt");
- builder.Property<DateTimeOffset>("UpdatedAt");
  builder.Property(x => x.Status).HasConversion<EnumToStringConverter<WorkflowStatus>>();
  builder.Property(x => x.SubStatus).HasConversion<EnumToStringConverter<WorkflowSubStatus>>();
  builder.HasIndex(x => x.CorrelationId).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.CorrelationId)}");
- builder.HasIndex(x=>x.DefinitionId).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.DefinitionId)}");
+ builder.HasIndex(x=> x.DefinitionId).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.DefinitionId)}");
  builder.HasIndex(x => new { x.Status, x.SubStatus, x.DefinitionId, x.DefinitionVersion }).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.Status)}_{nameof(WorkflowState.SubStatus)}_{nameof(WorkflowState.DefinitionId)}_{nameof(WorkflowState.DefinitionVersion)}");
  builder.HasIndex(x => new { x.Status, x.SubStatus }).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.Status)}_{nameof(WorkflowState.SubStatus)}");
  builder.HasIndex(x => new { x.Status, x.DefinitionId}).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.Status)}_{nameof(WorkflowState.DefinitionId)}");
  builder.HasIndex(x => new { x.Status, x.SubStatus }).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.Status)}_{nameof(WorkflowState.SubStatus)}");
- builder.HasIndex("CreatedAt").HasDatabaseName($"IX_{nameof(WorkflowState)}_CreatedAt");
- builder.HasIndex("UpdatedAt").HasDatabaseName($"IX_{nameof(WorkflowState)}_UpdatedAt");
+ builder.HasIndex(x => x.CreatedAt).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.CreatedAt)}");
+ builder.HasIndex(x => x.UpdatedAt).HasDatabaseName($"IX_{nameof(WorkflowState)}_{nameof(WorkflowState.UpdatedAt)}");
  }
 
  /// <inheritdoc />

src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/WorkflowStateStore.cs

@@ -70,11 +70,10 @@ private async ValueTask<WorkflowState> SaveAsync(RuntimeElsaDbContext dbContext,
  var now = _systemClock.UtcNow;
  var entry = dbContext.Entry(entity);
 
- if (entry.Property<DateTimeOffset>("CreatedAt").CurrentValue == DateTimeOffset.MinValue)
-  entry.Property<DateTimeOffset>("CreatedAt").CurrentValue = now;
+ entity.CreatedAt = entity.CreatedAt == default ? now : entity.CreatedAt;
+ entity.UpdatedAt = now;
 
  entry.Property<string>("Data").CurrentValue = json;
- entry.Property<DateTimeOffset>("UpdatedAt").CurrentValue = now;
  return entity;
  }
 }

src/modules/Elsa.MongoDB/Common/MongoStore.cs

@@ -0,0 +1,170 @@
+using System.Linq.Expressions;
+using Elsa.MongoDB.Extensions;
+using JetBrains.Annotations;
+using MongoDB.Driver;
+using MongoDB.Driver.Linq;
+
+namespace Elsa.MongoDB.Common;
+
+/// <summary>
+/// A generic repository class around MongoDb for accessing documents.
+/// </summary>
+/// <typeparam name="TDocument">The type of the document.</typeparam>
+[PublicAPI]
+public class MongoStore<TDocument> where TDocument : class
+{
+ private readonly IMongoCollection<TDocument> _collection;
+
+ /// <param name="collection"></param>
+ public MongoStore(IMongoCollection<TDocument> collection)
+ {
+ _collection = collection;
+ }
+
+ /// <summary>
+ /// Returns a queryable collection of documents.
+ /// </summary>
+ public IMongoCollection<TDocument> GetCollection() => _collection;
+
+ /// <summary>
+ /// Saves the document.
+ /// </summary>
+ /// <param name="document">The document to save.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public async Task<TDocument> AddAsync(TDocument document, CancellationToken cancellationToken = default)
+ {
+ await _collection.InsertOneAsync(document, new InsertOneOptions(), cancellationToken);
+ return document;
+ }
+
+ /// <summary>
+ /// Saves a list of documents.
+ /// </summary>
+ /// <param name="documents">The documents to save.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public async Task AddManyAsync(IEnumerable<TDocument> documents, CancellationToken cancellationToken = default)
+ {
+ await _collection.InsertManyAsync(documents, new InsertManyOptions(), cancellationToken);
+ }
+
+ /// <summary>
+ /// Saves the document.
+ /// </summary>
+ /// <param name="document">The document to save.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public async Task<TDocument> SaveAsync(TDocument document, CancellationToken cancellationToken = default)
+ {
+ return await _collection.FindOneAndReplaceAsync(document.BuildIdFilter(), document, new FindOneAndReplaceOptions<TDocument>{ ReturnDocument = ReturnDocument.After, IsUpsert = true }, cancellationToken);
+ }
+
+ /// <summary>
+ /// Saves the document.
+ /// </summary>
+ /// <param name="document">The document to save.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public async Task<TDocument> SaveAsync<TResult>(TDocument document, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default)
+ {
+ return await _collection.FindOneAndReplaceAsync(document.BuildExpression(selector), document, new FindOneAndReplaceOptions<TDocument>{ ReturnDocument = ReturnDocument.After, IsUpsert = true }, cancellationToken);
+ }
+
+ /// <summary>
+ /// Saves the specified documents.
+ /// </summary>
+ /// <param name="documents">The documents to save.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ public async Task SaveManyAsync(IEnumerable<TDocument> documents, CancellationToken cancellationToken = default)
+ {
+ var writes = new List<WriteModel<TDocument>>();
+
+ foreach (var document in documents)
+ {
+ var replacement = new ReplaceOneModel<TDocument>(document.BuildIdFilter(), document) { IsUpsert = true };
+ writes.Add(replacement);
+ }
+
+ await _collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
+ }
+
+ /// <summary>
+ /// Finds the document matching the specified predicate
+ /// </summary>
+ /// <param name="predicate">The predicate to use.</param>
+ /// <param name="cancellationToken">The cancellation token.</param>
+ /// <returns>The document if found, otherwise <c>null</c>.</returns>
+ public async Task<TDocument?> FindAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) => 
+ await _collection.AsQueryable().Where(predicate).FirstOrDefaultAsync(cancellationToken);
+
+ /// <summary>
+ /// Finds a single document using a query
+ /// </summary>
+ /// <param name="query">The query to use</param>
+ /// <param name="cancellationToken">The cancellation token</param>
+ /// <returns>The document if found, otherwise <c>null</c></returns>
+ public async Task<TDocument?> FindAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) => 
+ await query(_collection.AsQueryable()).FirstOrDefaultAsync(cancellationToken);
+
+ /// <summary>
+ /// Finds a list of documents matching the specified predicate
+ /// </summary>
+ public async Task<IEnumerable<TDocument>> FindManyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) => 
+ await _collection.AsQueryable().Where(predicate).ToListAsync(cancellationToken);
+
+ /// <summary>
+ /// Queries the database using a query and a selector.
+ /// </summary>
+ public async Task<IEnumerable<TResult>> FindManyAsync<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) => 
+ await query(_collection.AsQueryable()).Select(selector).ToListAsync(cancellationToken);
+
+ /// <summary>
+ /// Finds a list of documents using a query
+ /// </summary>
+ public async Task<IEnumerable<TDocument>> FindManyAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) => 
+ await query(_collection.AsQueryable()).ToListAsync(cancellationToken);
+
+ /// <summary>
+ /// Queries the database using a query and a selector.
+ /// </summary>
+ public async Task<IEnumerable<TResult>> FindMany<TResult>(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, Expression<Func<TDocument, TResult>> selector, CancellationToken cancellationToken = default) => 
+ await query(_collection.AsQueryable()).Select(selector).ToListAsync(cancellationToken);
+
+ /// <summary>
+ /// Counts documents in the collection using a filter.
+ /// </summary>
+ public async Task<long> CountAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default) => 
+ await query(_collection.AsQueryable()).LongCountAsync(cancellationToken);
+
+ /// <summary>
+ /// Checks if any documents exist.
+ /// </summary>
+ public async Task<bool> AnyAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default) => 
+ await _collection.AsQueryable().Where(predicate).AnyAsync(cancellationToken);
+
+ /// <summary>
+ /// Deletes documents using a predicate.
+ /// </summary>
+ /// <returns>The number of documents deleted.</returns>
+ public async Task<int> DeleteWhereAsync(Expression<Func<TDocument, bool>> predicate, CancellationToken cancellationToken = default)
+ {
+ var documentsToDelete = await _collection.AsQueryable().Where(predicate).ToListAsync(cancellationToken);
+ var count = documentsToDelete.Count;
+
+ await _collection.DeleteManyAsync(predicate, cancellationToken);
+
+ return count;
+ }
+
+ /// <summary>
+ /// Deletes documents using a query.
+ /// </summary>
+ /// <returns>The number of documents deleted.</returns>
+ public async Task<int> DeleteWhereAsync(Func<IMongoQueryable<TDocument>, IMongoQueryable<TDocument>> query, CancellationToken cancellationToken = default)
+ {
+ var documentsToDelete = await query(_collection.AsQueryable()).ToListAsync(cancellationToken);
+ var count = documentsToDelete.Count;
+
+ var filter = documentsToDelete.BuildIdFilterForList();
+ await _collection.DeleteManyAsync(filter, cancellationToken);
+
+ return count;
+ }
+}

src/modules/Elsa.MongoDB/Common/PersistenceFeatureBase.cs

@@ -0,0 +1,29 @@
+using Elsa.Features.Abstractions;
+using Elsa.Features.Services;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Elsa.MongoDB.Common;
+
+/// <summary>
+/// Base class for features that configure MongoDb persistence.
+/// </summary>
+public abstract class PersistenceFeatureBase : FeatureBase
+{
+ /// <inheritdoc />
+ protected PersistenceFeatureBase(IModule module) : base(module)
+ {
+ }
+
+ /// <summary>
+ /// Registers an <see cref="MongoStore{TDocument}"/>.
+ /// </summary>
+ /// <typeparam name="TStore">The type of the store.</typeparam>
+ /// <typeparam name="TDocument">The document type of the store.</typeparam>
+ protected void AddStore<TDocument, TStore>() where TDocument : class where TStore : class
+ {
+ Services
+ .AddSingleton<MongoStore<TDocument>>()
+ .AddSingleton<TStore>()
+ ;
+ }
+}

src/modules/Elsa.MongoDB/Elsa.MongoDB.csproj

@@ -0,0 +1,35 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+ <Import Project="..\..\..\common.props" />
+ <Import Project="..\..\..\configureawait.props" />
+
+ <PropertyGroup>
+ <TargetFrameworks>net6.0;net7.0</TargetFrameworks>
+ <Description>
+ Provides MongoDB implementations of various abstractions from various modules.
+ </Description>
+ <PackageTags>elsa module persistence mongo</PackageTags>
+ </PropertyGroup>
+
+ <ItemGroup>
+ <PackageReference Include="MongoDB.Driver" Version="2.19.1" />
+ <PackageReference Include="MongoDB.Driver.Core.Extensions.DiagnosticSources" Version="1.3.0" />
+ <PackageReference Include="MongoDB.Driver.Extensions" Version="2.0.2" />
+ </ItemGroup>
+
+ <ItemGroup>
+ <ProjectReference Include="..\..\common\Elsa.Features\Elsa.Features.csproj" />
+ <ProjectReference Include="..\Elsa.Common\Elsa.Common.csproj" />
+ <ProjectReference Include="..\Elsa.Http\Elsa.Http.csproj" />
+ <ProjectReference Include="..\Elsa.Identity\Elsa.Identity.csproj" />
+ <ProjectReference Include="..\Elsa.Labels\Elsa.Labels.csproj" />
+ <ProjectReference Include="..\Elsa.Workflows.Management\Elsa.Workflows.Management.csproj" />
+ <ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
+ <ProjectReference Include="..\Elsa.Workflows.Sinks\Elsa.Workflows.Sinks.csproj" />
+ </ItemGroup>
+
+ <ItemGroup>
+ <PackageReference Include="Microsoft.Extensions.Options" Version="7.0.1" />
+ </ItemGroup>
+
+</Project>