Fix MongoDB serialization issues (#6762)

* Ongoing MongoDB work

* Refactor MongoDB serializer configuration and improve type handling

- Added `BsonSerializerHelpers` for streamlined serializer registration with error handling.
- Introduced `ConfigureMongoDbSerializers` hosted service to centralize serializer setup.
- Enhanced `FlowScopeSerializer` to handle additional BSON types.
- Cleaned up and reorganized MongoDB feature implementations, removing redundant code and improving consistency.

* Remove `BsonSerializerHelpers` and update MongoDB serializer registration

- Deleted `BsonSerializerHelpers` as it was redundant.
- Updated `ConfigureMongoDbSerializers` to directly register serializers using `BsonSerializer`.
- Simplified `FlowScopeSerializer` null handling for improved readability.

* Switch persistence provider to Entity Framework Core in `Program.cs`.

* Update src/modules/Elsa.MongoDb/Serializers/FlowScopeSerializer.cs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Configures MongoDB services and options.

Configures MongoDB services, including client and database creation.

Registers default naming strategy and collection naming strategy.

Adds BsonClassMap configuration for KeyValuePair.

* Simplify null check in `FlowScopeSerializer`.

* Allow `FlowScopeSerializer` to handle nullable `FlowScope`.

* Registers FlowScope BSON class map

Registers the `FlowScope` class with BSON to enable proper serialization and deserialization of workflow scopes within MongoDB.

This ensures that workflow scopes, which are used to manage variables within a workflow, are correctly stored and retrieved from the database.

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Sipke Schoorstra 2025-07-04 09:46:20 +02:00 committed by GitHub
parent c4451c4f40
commit 1a554e7e06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 152 additions and 49 deletions

View file

@ -47,11 +47,13 @@ public class Module : IModule
}
/// <inheritdoc />
public T Configure<T>(Action<T>? configure = default) where T : class, IFeature
=> Configure(module => (T)Activator.CreateInstance(typeof(T), module)!, configure);
public T Configure<T>(Action<T>? configure = null) where T : class, IFeature
{
return Configure(module => (T)Activator.CreateInstance(typeof(T), module)!, configure);
}
/// <inheritdoc />
public T Configure<T>(Func<IModule, T> factory, Action<T>? configure = default) where T : class, IFeature
public T Configure<T>(Func<IModule, T> factory, Action<T>? configure = null) where T : class, IFeature
{
if (!_features.TryGetValue(typeof(T), out var feature))
{
@ -81,7 +83,7 @@ public class Module : IModule
/// <inheritdoc />
public IModule ConfigureHostedService(Type hostedServiceType, int priority = 0)
{
_hostedServiceDescriptors.Add(new HostedServiceDescriptor(priority, hostedServiceType));
_hostedServiceDescriptors.Add(new(priority, hostedServiceType));
return this;
}
@ -121,7 +123,7 @@ public class Module : IModule
var ns = "Elsa";
var displayName = type.GetCustomAttribute<DisplayNameAttribute>()?.DisplayName ?? name;
var description = type.GetCustomAttribute<DescriptionAttribute>()?.Description;
registry.Add(new FeatureDescriptor(name, ns, displayName, description));
registry.Add(new(name, ns, displayName, description));
}
Services.AddSingleton<IInstalledFeatureRegistry>(registry);

View file

@ -31,12 +31,12 @@ public interface IModule
/// <summary>
/// Creates and configures a feature of the specified type.
/// </summary>
T Configure<T>(Action<T>? configure = default) where T : class, IFeature;
T Configure<T>(Action<T>? configure = null) where T : class, IFeature;
/// <summary>
/// Creates and configures a feature of the specified type.
/// </summary>
T Configure<T>(Func<IModule, T> factory, Action<T>? configure = default) where T : class, IFeature;
T Configure<T>(Func<IModule, T> factory, Action<T>? configure = null) where T : class, IFeature;
/// <summary>
/// Configures a <see cref="IHostedService"/> using an optional priority to control in which order it will be registered with the service container.

View file

@ -134,10 +134,11 @@ public class MongoDbStore<TDocument>(IMongoCollection<TDocument> collection, ITe
await collection.BulkWriteAsync(writes, cancellationToken: cancellationToken);
}
public async Task UpdatePartialAsync(
public async Task<bool> UpdatePartialAsync(
string id,
IDictionary<string, object> updatedFields,
string primaryKey = nameof(Entity.Id),
bool throwIfNotFound = true,
CancellationToken cancellationToken = default)
{
if (string.IsNullOrEmpty(id))
@ -154,7 +155,14 @@ public class MongoDbStore<TDocument>(IMongoCollection<TDocument> collection, ITe
var updateResult = await collection.UpdateOneAsync(filter, updateDefinition, cancellationToken: cancellationToken);
if (updateResult.MatchedCount == 0)
{
if (!throwIfNotFound)
return false;
throw new InvalidOperationException($"No document found with ID '{id}'.");
}
return updateResult.ModifiedCount > 0;
}
/// <summary>

View file

@ -1,20 +1,16 @@
using System.Text.Json;
using System.Text.Json.Nodes;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Elsa.KeyValues.Entities;
using Elsa.MongoDb.Contracts;
using Elsa.MongoDb.HostedServices;
using Elsa.MongoDb.NamingStrategies;
using Elsa.MongoDb.Options;
using Elsa.MongoDb.Serializers;
using Elsa.Workflows.Memory;
using Elsa.Workflows.Activities.Flowchart.Models;
using Elsa.Workflows.Runtime.Entities;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Options;
using MongoDB.Bson;
using MongoDB.Bson.Serialization;
using MongoDB.Bson.Serialization.Serializers;
using MongoDB.Driver;
namespace Elsa.MongoDb.Features;
@ -27,7 +23,7 @@ public class MongoDbFeature(IModule module) : FeatureBase(module)
/// <summary>
/// The MongoDB connection string.
/// </summary>
public string ConnectionString { get; set; } = default!;
public string ConnectionString { get; set; } = null!;
/// <summary>
/// A delegate that configures MongoDb.
@ -39,6 +35,11 @@ public class MongoDbFeature(IModule module) : FeatureBase(module)
/// </summary>
public Func<IServiceProvider, ICollectionNamingStrategy> CollectionNamingStrategy { get; set; } = sp => sp.GetRequiredService<DefaultNamingStrategy>();
public override void ConfigureHostedServices()
{
Module.ConfigureHostedService<ConfigureMongoDbSerializers>(-10);
}
/// <inheritdoc />
public override void Apply()
{
@ -47,24 +48,13 @@ public class MongoDbFeature(IModule module) : FeatureBase(module)
var mongoUrl = new MongoUrl(ConnectionString);
Services.AddSingleton(sp => CreateMongoClient(sp, mongoUrl));
Services.AddScoped(sp => CreateDatabase(sp, mongoUrl));
Services.TryAddScoped<DefaultNamingStrategy>();
Services.AddScoped(CollectionNamingStrategy);
RegisterSerializers();
RegisterClassMaps();
}
private static void RegisterSerializers()
{
TryRegisterSerializerOrSkipWhenExist(typeof(object), new PolymorphicSerializer());
TryRegisterSerializerOrSkipWhenExist(typeof(Type), new TypeSerializer());
TryRegisterSerializerOrSkipWhenExist(typeof(Variable), new VariableSerializer());
TryRegisterSerializerOrSkipWhenExist(typeof(Version), new VersionSerializer());
TryRegisterSerializerOrSkipWhenExist(typeof(JsonElement), new JsonElementSerializer());
TryRegisterSerializerOrSkipWhenExist(typeof(JsonNode), new JsonNodeBsonConverter());
}
private static void RegisterClassMaps()
{
BsonClassMap.TryRegisterClassMap<StoredBookmark>(cm =>
@ -78,19 +68,14 @@ public class MongoDbFeature(IModule module) : FeatureBase(module)
map.SetIgnoreExtraElements(true); // Needed for missing ID property
map.MapProperty(x => x.Key); // Needed for non-setter property
});
BsonClassMap.TryRegisterClassMap<FlowScope>(map =>
{
map.AutoMap();
map.SetIgnoreExtraElements(true);
});
}
private static void TryRegisterSerializerOrSkipWhenExist(Type type, IBsonSerializer serializer)
{
try
{
BsonSerializer.TryRegisterSerializer(type, serializer);
}
catch (BsonSerializationException ex)
{
}
}
private static IMongoClient CreateMongoClient(IServiceProvider sp, MongoUrl mongoUrl)
{
var options = sp.GetRequiredService<IOptions<MongoDbOptions>>().Value;
@ -99,7 +84,7 @@ public class MongoDbFeature(IModule module) : FeatureBase(module)
// TODO: Uncomment once https://github.com/jbogard/MongoDB.Driver.Core.Extensions.DiagnosticSources/pull/41 is merged and deployed.
//settings.ClusterConfigurator = cb => cb.Subscribe(new DiagnosticsActivityEventSubscriber());
settings.ApplicationName = GetApplicationName(settings);
settings.WriteConcern = options.WriteConcern;
settings.ReadConcern = options.ReadConcern;

View file

@ -5,10 +5,10 @@ using Elsa.Workflows.Runtime.Entities;
namespace Elsa.MongoDb.Helpers;
internal class ExpressionHelpers
internal static class ExpressionHelpers
{
public static readonly Expression<Func<WorkflowDefinition, WorkflowDefinitionSummary>> WorkflowDefinitionSummary =
workflowDefinition => new WorkflowDefinitionSummary
workflowDefinition => new()
{
Id = workflowDefinition.Id,
DefinitionId = workflowDefinition.DefinitionId,
@ -18,11 +18,14 @@ internal class ExpressionHelpers
IsLatest = workflowDefinition.IsLatest,
IsPublished = workflowDefinition.IsPublished,
MaterializerName = workflowDefinition.MaterializerName,
CreatedAt = workflowDefinition.CreatedAt
CreatedAt = workflowDefinition.CreatedAt,
IsReadonly = workflowDefinition.IsReadonly,
ProviderName = workflowDefinition.ProviderName,
ToolVersion = workflowDefinition.ToolVersion
};
public static readonly Expression<Func<WorkflowInstance, WorkflowInstanceSummary>> WorkflowInstanceSummary =
workflowInstance => new WorkflowInstanceSummary
workflowInstance => new()
{
Id = workflowInstance.Id,
DefinitionId = workflowInstance.DefinitionId,
@ -34,16 +37,17 @@ internal class ExpressionHelpers
Name = workflowInstance.Name,
CreatedAt = workflowInstance.CreatedAt,
UpdatedAt = workflowInstance.UpdatedAt,
FinishedAt = workflowInstance.FinishedAt
FinishedAt = workflowInstance.FinishedAt,
IncidentCount = workflowInstance.IncidentCount,
};
public static readonly Expression<Func<WorkflowInstance, WorkflowInstanceId>> WorkflowInstanceId = workflowInstance => new WorkflowInstanceId
public static readonly Expression<Func<WorkflowInstance, WorkflowInstanceId>> WorkflowInstanceId = workflowInstance => new()
{
Id = workflowInstance.Id
};
public static readonly Expression<Func<ActivityExecutionRecord, ActivityExecutionRecordSummary>> ActivityExecutionRecordSummary =
workflowInstance => new ActivityExecutionRecordSummary
workflowInstance => new()
{
Id = workflowInstance.Id,
Status = workflowInstance.Status,
@ -53,6 +57,11 @@ internal class ExpressionHelpers
ActivityTypeVersion = workflowInstance.ActivityTypeVersion,
ActivityName = workflowInstance.ActivityName,
StartedAt = workflowInstance.StartedAt,
HasBookmarks = workflowInstance.HasBookmarks
HasBookmarks = workflowInstance.HasBookmarks,
CompletedAt = workflowInstance.CompletedAt,
AggregateFaultCount = workflowInstance.AggregateFaultCount,
Metadata = workflowInstance.Metadata,
WorkflowInstanceId = workflowInstance.WorkflowInstanceId,
TenantId = workflowInstance.TenantId,
};
}

View file

@ -0,0 +1,43 @@
using System.Text.Json;
using System.Text.Json.Nodes;
using Elsa.MongoDb.Serializers;
using Elsa.Workflows;
using Elsa.Workflows.Activities.Flowchart.Models;
using Elsa.Workflows.Memory;
using JetBrains.Annotations;
using Microsoft.Extensions.Hosting;
using MongoDB.Bson.Serialization.Serializers;
using static MongoDB.Bson.Serialization.BsonSerializer;
namespace Elsa.MongoDb.HostedServices;
/// <summary>
/// A hosted service that configures and registers custom MongoDB serializers for various types.
/// </summary>
/// <remarks>
/// This class implements <see cref="IHostedService"/> and is responsible for registering serializers to handle
/// specific types such as <see cref="object"/>, <see cref="Type"/>, <see cref="Variable"/>, <see cref="Version"/>,
/// <see cref="JsonElement"/>, <see cref="JsonNode"/>, and <see cref="FlowScope"/>.
/// It uses helper methods to register these serializers during the application's startup process.
/// </remarks>
[UsedImplicitly]
public class ConfigureMongoDbSerializers(IPayloadSerializer payloadSerializer) : IHostedService
{
public Task StartAsync(CancellationToken cancellationToken)
{
TryRegisterSerializer(typeof(object), new PolymorphicSerializer());
TryRegisterSerializer(typeof(Type), new TypeSerializer());
TryRegisterSerializer(typeof(Variable), new VariableSerializer());
TryRegisterSerializer(typeof(Version), new VersionSerializer());
TryRegisterSerializer(typeof(JsonElement), new JsonElementSerializer());
TryRegisterSerializer(typeof(JsonNode), new JsonNodeBsonConverter());
TryRegisterSerializer(typeof(FlowScope), new FlowScopeSerializer(payloadSerializer));
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken)
{
return Task.CompletedTask;
}
}

View file

@ -8,6 +8,7 @@ using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Management.Models;
using JetBrains.Annotations;
using Microsoft.Extensions.Logging;
using MongoDB.Driver;
using MongoDB.Driver.Linq;
using Open.Linq.AsyncExtensions;
@ -18,7 +19,7 @@ namespace Elsa.MongoDb.Modules.Management;
/// A MongoDb implementation of <see cref="IWorkflowInstanceStore"/>.
/// </summary>
[UsedImplicitly]
public class MongoWorkflowInstanceStore(MongoDbStore<WorkflowInstance> mongoDbStore) : IWorkflowInstanceStore
public class MongoWorkflowInstanceStore(MongoDbStore<WorkflowInstance> mongoDbStore, ILogger<MongoWorkflowInstanceStore> logger) : IWorkflowInstanceStore
{
/// <inheritdoc />
public async ValueTask<WorkflowInstance?> FindAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default)
@ -130,7 +131,10 @@ public class MongoWorkflowInstanceStore(MongoDbStore<WorkflowInstance> mongoDbSt
[nameof(WorkflowInstance.UpdatedAt)] = value
};
await mongoDbStore.UpdatePartialAsync(workflowInstanceId, props, cancellationToken: cancellationToken);
var updated = await mongoDbStore.UpdatePartialAsync(workflowInstanceId, props, throwIfNotFound: false, cancellationToken: cancellationToken);
if (!updated)
logger.LogDebug("Failed to update the 'UpdatedAt' timestamp for workflow instance with ID '{WorkflowInstanceId}'. This means this workflow does not yet exist in the DB.", workflowInstanceId);
}
/// <inheritdoc />

View file

@ -0,0 +1,52 @@
using Elsa.Workflows;
using Elsa.Workflows.Activities.Flowchart.Models;
using MongoDB.Bson;
using MongoDB.Bson.Serialization;
namespace Elsa.MongoDb.Serializers;
/// <summary>
/// Serializes a <see cref="FlowScope"/>.
/// </summary>
public class FlowScopeSerializer(IPayloadSerializer payloadSerializer) : IBsonSerializer<FlowScope?>
{
/// <inheritdoc />
public Type ValueType => typeof(FlowScope);
void IBsonSerializer.Serialize(BsonSerializationContext context, BsonSerializationArgs args, object value) => Serialize(context, args, (FlowScope)value);
object IBsonSerializer.Deserialize(BsonDeserializationContext context, BsonDeserializationArgs args) => Deserialize(context, args);
/// <inheritdoc />
public void Serialize(BsonSerializationContext context, BsonSerializationArgs args, FlowScope? value)
{
if (value is null)
context.Writer.WriteNull();
else
{
var json = payloadSerializer.Serialize(value);
context.Writer.WriteString(json);
}
}
/// <inheritdoc />
public FlowScope Deserialize(BsonDeserializationContext context, BsonDeserializationArgs args)
{
var reader = context.Reader;
var bsonType = reader.GetCurrentBsonType();
if (bsonType == BsonType.Null)
{
reader.ReadNull();
return new();
}
if(bsonType == BsonType.String)
{
var json = context.Reader.ReadString();
return string.IsNullOrEmpty(json) ? new() : payloadSerializer.Deserialize<FlowScope>(json);
}
return new();
}
}