SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Removed EFCore.BulkExtensions and used EF7 bulk operations (#3630)

Browse source 
* Removed EFCore.BulkExtensions and used EF7 bulk operations

* Refactor

* Use bookmarkId instead of hash as unique field

* Reuse GetProperty extension

Co-authored-by: Gürkan Güran <grkngrn@gmail.com>
Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com>
This commit is contained in:
gurkanguran 2023-01-16 19:41:08 +01:00 committed by GitHub
parent d6deed7b73
commit d94cbcd4b1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 128 additions and 147 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/modules/Elsa.Elasticsearch/Options/ElasticsearchOptions.cs
View file

@ -28,11 +28,6 @@ public class ElasticsearchOptions
/// The API key to use when connecting with the Elasticsearch server.
/// </summary>
public string? ApiKey { get; set; }
/// <summary>
/// The interval to attempt a rollover.
/// </summary>
public TimeSpan RolloverInterval { get; set; } = TimeSpan.FromDays(10);
/// <summary>
/// A map between type and index name to use. When no index name is configured for a given type, the name of the type is used.

12
src/modules/Elsa.EntityFrameworkCore.PostgreSql/Elsa.EntityFrameworkCore.PostgreSql.csproj
View file

@ -11,16 +11,8 @@
<PackageTags>elsa module persistence efcore postgresql</PackageTags>
</PropertyGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net6.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.12">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Npgsql.EntityFrameworkCore.PostgreSQL" Version="6.0.8" />
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net7.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.1">
<ItemGroup>
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>

14
src/modules/Elsa.EntityFrameworkCore.SqlServer/Elsa.EntityFrameworkCore.SqlServer.csproj
View file

@ -11,20 +11,12 @@
<PackageTags>elsa module persistence efcore sqlserver</PackageTags>
</PropertyGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net6.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.12">
<ItemGroup>
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="6.0.12" />
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net7.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="7.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="7.0.2" />
</ItemGroup>
<ItemGroup>

15
src/modules/Elsa.EntityFrameworkCore.Sqlite/Elsa.EntityFrameworkCore.Sqlite.csproj
View file

@ -11,19 +11,10 @@
<PackageTags>elsa module persistence efcore sqlite</PackageTags>
</PropertyGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net6.0'">
<PackageReference Include="Microsoft.Data.Sqlite" Version="6.0.12" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="6.0.12" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.12">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net7.0'">
<ItemGroup>
<PackageReference Include="Microsoft.Data.Sqlite" Version="7.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="7.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.1">
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="7.0.2" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>

34
src/modules/Elsa.EntityFrameworkCore/Common/Store.cs
View file

@ -1,5 +1,4 @@
using System.Linq.Expressions;
using EFCore.BulkExtensions;
using Elsa.Common.Entities;
using Elsa.Common.Models;
using Elsa.EntityFrameworkCore.Extensions;
@ -17,18 +16,27 @@ public class Store<TDbContext, TEntity> where TDbContext : DbContext where TEnti
}
public async Task<TDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default) => await _dbContextFactory.CreateDbContextAsync(cancellationToken);
public async Task SaveAsync(TEntity entity, CancellationToken cancellationToken = default) => await SaveAsync(entity, default, cancellationToken);
public async Task SaveAsync(TEntity entity, CancellationToken cancellationToken = default) => await SaveAsync(entity, default, default, cancellationToken);
public async Task SaveAsync(TEntity entity, Expression<Func<TEntity, object>>? uniqueField = default, CancellationToken cancellationToken = default) => await SaveAsync(entity, uniqueField, default, cancellationToken);
public async Task SaveAsync(TEntity entity, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default) => await SaveAsync(entity, default, onSaving, cancellationToken);
public async Task SaveAsync(TEntity entity, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default)
public async Task SaveAsync(TEntity entity, Expression<Func<TEntity, object>>? uniqueField = default, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default)
{
await using var dbContext = await CreateDbContextAsync(cancellationToken);
entity = onSaving?.Invoke(dbContext, entity) ?? entity;
await dbContext.BulkUpsertAsync(new[] {entity}, cancellationToken);
await dbContext.BulkUpsertAsync(new[] {entity}, uniqueField, cancellationToken);
}
public async Task SaveManyAsync(IEnumerable<TEntity> entities, CancellationToken cancellationToken = default) => await SaveManyAsync(entities, default, cancellationToken);
public async Task SaveManyAsync(IEnumerable<TEntity> entities, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default)
public async Task SaveManyAsync(IEnumerable<TEntity> entities, CancellationToken cancellationToken = default) => await SaveManyAsync(entities, default, default, cancellationToken);
public async Task SaveManyAsync(IEnumerable<TEntity> entities, Expression<Func<TEntity, object>>? uniqueField = default, CancellationToken cancellationToken = default) => await SaveManyAsync(entities, uniqueField, default, cancellationToken);
public async Task SaveManyAsync(IEnumerable<TEntity> entities, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default) => await SaveManyAsync(entities, default, onSaving, cancellationToken);
public async Task SaveManyAsync(IEnumerable<TEntity> entities, Expression<Func<TEntity, object>>? uniqueField = default, Func<TDbContext, TEntity, TEntity>? onSaving = default, CancellationToken cancellationToken = default)
{
await using var dbContext = await CreateDbContextAsync(cancellationToken);
var entityList = entities.ToList();
@ -36,7 +44,7 @@ public class Store<TDbContext, TEntity> where TDbContext : DbContext where TEnti
if (onSaving != null)
entityList = entityList.Select(x => onSaving(dbContext, x)).ToList();
await dbContext.BulkUpsertAsync(entityList, cancellationToken);
await dbContext.BulkUpsertAsync(entityList, uniqueField, cancellationToken);
}
public async Task<TEntity?> FindAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default) => await FindAsync(predicate, default, cancellationToken);
@ -111,19 +119,11 @@ public class Store<TDbContext, TEntity> where TDbContext : DbContext where TEnti
return await dbContext.SaveChangesAsync(cancellationToken) == 1;
}
public async Task<int> DeleteManyAsync(IEnumerable<TEntity> entities, CancellationToken cancellationToken = default)
{
await using var dbContext = await CreateDbContextAsync(cancellationToken);
var list = entities.ToList();
await dbContext.BulkDeleteAsync(list, cancellationToken: cancellationToken);
return list.Count;
}
public async Task<int> DeleteWhereAsync(Expression<Func<TEntity, bool>> predicate, CancellationToken cancellationToken = default)
{
await using var dbContext = await CreateDbContextAsync(cancellationToken);
var set = dbContext.Set<TEntity>();
return await set.DeleteWhereAsync(dbContext, predicate, cancellationToken);
return await set.Where(predicate).ExecuteDeleteAsync(cancellationToken);
}
public async Task<IEnumerable<TEntity>> QueryAsync(Func<IQueryable<TEntity>, IQueryable<TEntity>> query, CancellationToken cancellationToken = default) => await QueryAsync(query, default, cancellationToken);

22
src/modules/Elsa.EntityFrameworkCore/Elsa.EntityFrameworkCore.csproj
View file

@ -12,30 +12,18 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="EFCore.BulkExtensions" Version="6.6.5" />
<PackageReference Include="LinqKit" Version="1.2.3" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Sqlite" Version="7.0.2" />
<PackageReference Include="Open.Linq.AsyncExtensions" Version="1.2.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="7.0.0" />
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net6.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="6.0.12" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="6.0.12" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.12">
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.2" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="7.0.2" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.2">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup Condition=" '$(TargetFramework)' == 'net7.0'">
<PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Relational" Version="7.0.1" />
<PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="7.0.1">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.ActivityDefinitions\Elsa.ActivityDefinitions.csproj" />
<ProjectReference Include="..\Elsa.Labels\Elsa.Labels.csproj" />

16
src/modules/Elsa.EntityFrameworkCore/Extensions/DbSetExtensions.cs
View file

@ -1,16 +0,0 @@
using System.Linq.Expressions;
using Microsoft.EntityFrameworkCore;
namespace Elsa.EntityFrameworkCore.Extensions;
/// <summary>
/// Provides extensions to <see cref="DbSet{TEntity}"/>.
/// </summary>
public static class DbSetExtensions
{
/// <summary>
/// Deletes matching results in bulk.
/// </summary>
public static async Task<int> DeleteWhereAsync<T>(this DbSet<T> set, DbContext dbContext, Expression<Func<T, bool>> predicate, CancellationToken cancellationToken = default) where T : class =>
await set.Where(predicate).BulkDeleteAsync(dbContext, cancellationToken);
}

20
src/modules/Elsa.EntityFrameworkCore/Extensions/ExpressionExtensions.cs Normal file
View file

@ -0,0 +1,20 @@
using System.Linq.Expressions;
using System.Reflection;
namespace Elsa.EntityFrameworkCore.Extensions;
public static class ExpressionExtensions
{
public static Expression<Func<TEntity, bool>> BuildContainsExpression<TEntity>(this Func<TEntity, object> uniqueFieldDelegate, IEnumerable<TEntity> entities, PropertyInfo property) where TEntity : class
{
var list = entities.Select(uniqueFieldDelegate.Invoke);
var param = Expression.Parameter(typeof(TEntity));
var body = Expression.Call(
typeof(Enumerable),
"Contains",
new[] {uniqueFieldDelegate.Method.ReturnType},
Expression.Constant(list), Expression.Property(param, property));
return Expression.Lambda<Func<TEntity, bool>>(body, param);
}
}

64
src/modules/Elsa.EntityFrameworkCore/Extensions/QueryableExtensions.cs
View file

@ -1,6 +1,7 @@
using System.Linq.Expressions;
using EFCore.BulkExtensions;
using Elsa.Common.Entities;
using Elsa.Common.Models;
using Elsa.Extensions;
using Microsoft.EntityFrameworkCore;
namespace Elsa.EntityFrameworkCore.Extensions;
@ -10,34 +11,29 @@ namespace Elsa.EntityFrameworkCore.Extensions;
/// </summary>
public static class QueryableExtensions
{
/// <summary>
/// Deletes the matching results in bulk.
/// </summary>
public static async Task<int> BulkDeleteAsync<T>(this IQueryable<T> queryable, DbContext elsaContext, CancellationToken cancellationToken = default) where T : class
{
#if NET7_0_OR_GREATER
return await queryable.ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false);
#else
if (!elsaContext.Database.IsPostgres() && !elsaContext.Database.IsMySql() && !elsaContext.Database.IsOracle())
return await queryable.BatchDeleteAsync(cancellationToken);
// Need this workaround https://github.com/borisdj/EFCore.BulkExtensions/issues/553 is solved.
// Oracle also https://github.com/borisdj/EFCore.BulkExtensions/issues/375
var records = await queryable.ToListAsync(cancellationToken);
foreach (var record in records)
elsaContext.Remove(record);
return records.Count;
#endif
}
/// <summary>
/// Inserts or updates a list of entities in bulk.
/// </summary>
public static async Task BulkUpsertAsync<TDbContext, TEntity>(this TDbContext dbContext, IList<TEntity> entities, CancellationToken cancellationToken = default) where TDbContext : DbContext where TEntity : class =>
await dbContext.BulkInsertOrUpdateAsync(entities, config => { config.EnableShadowProperties = true; }, cancellationToken: cancellationToken);
public static async Task BulkUpsertAsync<TDbContext, TEntity>(this TDbContext dbContext, IList<TEntity> entities, Expression<Func<TEntity, object>>? uniqueFieldExpression = default, CancellationToken cancellationToken = default) where TDbContext : DbContext where TEntity : class
{
uniqueFieldExpression = ResolveUniqueFieldExpression(uniqueFieldExpression);
var uniqueFieldDelegate = uniqueFieldExpression.Compile();
var propertyInfo = uniqueFieldExpression.GetProperty()!;
var set = dbContext.Set<TEntity>();
var lambda = uniqueFieldDelegate.BuildContainsExpression(entities, propertyInfo);
var existingEntities = await set.AsNoTracking().Where(lambda).ToListAsync(cancellationToken);
var entitiesToUpdate = entities.Where(e => existingEntities.Any(ex => uniqueFieldDelegate.Invoke(ex).ToString() == uniqueFieldDelegate.Invoke(e).ToString())).ToList();
var entitiesToInsert = entities.Except(entitiesToUpdate).ToList();
if (entitiesToUpdate.Any())
set.UpdateRange(entitiesToUpdate);
if (entitiesToInsert.Any())
await set.AddRangeAsync(entitiesToInsert, cancellationToken);
await dbContext.SaveChangesAsync(cancellationToken);
}
/// <summary>
/// Returns a paged result from the specified query.
@ -62,4 +58,20 @@ public static class QueryableExtensions
var results = await queryable.ToListAsync();
return Page.Of(results, count);
}
private static Expression<Func<TEntity, object>> ResolveUniqueFieldExpression<TEntity>(Expression<Func<TEntity, object>>? uniqueFieldExpression) where TEntity : class
{
if (uniqueFieldExpression != null) return uniqueFieldExpression;
try
{
uniqueFieldExpression = e => ((Entity)(object)e).Id;
}
catch (Exception)
{
throw new Exception(
"Unique field expression must be passed via BulkUpsertAsync if default object to upsert is not of type Entity.");
}
return uniqueFieldExpression;
}
}

3
src/modules/Elsa.EntityFrameworkCore/Modules/Labels/WorkflowDefinitionLabelStore.cs
View file

@ -18,7 +18,8 @@ public class EFCoreWorkflowDefinitionLabelStore : IWorkflowDefinitionLabelStore
public async Task ReplaceAsync(IEnumerable<WorkflowDefinitionLabel> removed, IEnumerable<WorkflowDefinitionLabel> added, CancellationToken cancellationToken = default)
{
await _store.DeleteManyAsync(removed, cancellationToken);
var idList = removed.Select(r => r.Id);
await _store.DeleteWhereAsync(w => idList.Contains(w.Id), cancellationToken);
await _store.SaveManyAsync(added, cancellationToken);
}

17
src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowDefinitionStore.cs
View file

@ -16,14 +16,19 @@ namespace Elsa.EntityFrameworkCore.Modules.Management;
public class EFCoreWorkflowDefinitionStore : IWorkflowDefinitionStore
{
private readonly Store<ManagementElsaDbContext, WorkflowDefinition> _store;
private readonly Store<ManagementElsaDbContext, WorkflowInstance> _workflowInstanceStore;
private readonly SerializerOptionsProvider _serializerOptionsProvider;
/// <summary>
/// Constructor.
/// </summary>
public EFCoreWorkflowDefinitionStore(Store<ManagementElsaDbContext, WorkflowDefinition> store, SerializerOptionsProvider serializerOptionsProvider)
public EFCoreWorkflowDefinitionStore(
Store<ManagementElsaDbContext, WorkflowDefinition> store,
Store<ManagementElsaDbContext, WorkflowInstance> workflowInstanceStore,
SerializerOptionsProvider serializerOptionsProvider)
{
_store = store;
_workflowInstanceStore = workflowInstanceStore;
_serializerOptionsProvider = serializerOptionsProvider;
}
@ -93,16 +98,16 @@ public class EFCoreWorkflowDefinitionStore : IWorkflowDefinitionStore
public async Task<int> DeleteByDefinitionIdAsync(string definitionId, CancellationToken cancellationToken = default)
{
await using var dbContext = await _store.CreateDbContextAsync(cancellationToken);
await dbContext.WorkflowInstances.DeleteWhereAsync(dbContext, x => x.DefinitionId == definitionId, cancellationToken);
return await dbContext.WorkflowDefinitions.DeleteWhereAsync(dbContext, x => x.DefinitionId == definitionId, cancellationToken);
await _workflowInstanceStore.DeleteWhereAsync(x => x.DefinitionId == definitionId, cancellationToken);
return await _store.DeleteWhereAsync(x => x.DefinitionId == definitionId, cancellationToken);
}
/// <inheritdoc />
public async Task<int> DeleteByDefinitionIdAndVersionAsync(string definitionId, int version, CancellationToken cancellationToken = default)
{
await using var dbContext = await _store.CreateDbContextAsync(cancellationToken);
await dbContext.WorkflowInstances.DeleteWhereAsync(dbContext, x => x.DefinitionId == definitionId && x.Version == version, cancellationToken);
return await dbContext.WorkflowDefinitions.DeleteWhereAsync(dbContext, x => x.DefinitionId == definitionId && x.Version == version, cancellationToken);
await _workflowInstanceStore.DeleteWhereAsync(x => x.DefinitionId == definitionId && x.Version == version, cancellationToken);
return await _store.DeleteWhereAsync(x => x.DefinitionId == definitionId && x.Version == version, cancellationToken);
}
/// <inheritdoc />
@ -110,7 +115,7 @@ public class EFCoreWorkflowDefinitionStore : IWorkflowDefinitionStore
{
var definitionIdList = definitionIds.ToList();
await using var dbContext = await _store.CreateDbContextAsync(cancellationToken);
await dbContext.WorkflowInstances.DeleteWhereAsync(dbContext, x => definitionIdList.Contains(x.DefinitionId), cancellationToken);
await _workflowInstanceStore.DeleteWhereAsync(x => definitionIdList.Contains(x.DefinitionId), cancellationToken);
return await _store.DeleteWhereAsync(x => definitionIdList.Contains(x.DefinitionId), cancellationToken);
}

2
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/BookmarkStore.cs
View file

@ -14,7 +14,7 @@ public class EFCoreBookmarkStore : IBookmarkStore
public EFCoreBookmarkStore(Store<RuntimeElsaDbContext, StoredBookmark> store) => _store = store;
/// <inheritdoc />
public async ValueTask SaveAsync(StoredBookmark record, CancellationToken cancellationToken = default) => await _store.SaveAsync(record, cancellationToken);
public async ValueTask SaveAsync(StoredBookmark record, CancellationToken cancellationToken = default) => await _store.SaveAsync(record, s => s.BookmarkId, cancellationToken);
/// <inheritdoc />
public async ValueTask<IEnumerable<StoredBookmark>> FindByWorkflowInstanceAsync(string workflowInstanceId, CancellationToken cancellationToken = default) =>

2
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/TriggerStore.cs
View file

@ -34,7 +34,7 @@ public class EFCoreTriggerStore : ITriggerStore
/// <inheritdoc />
public async ValueTask ReplaceAsync(IEnumerable<StoredTrigger> removed, IEnumerable<StoredTrigger> added, CancellationToken cancellationToken = default)
{
await _store.DeleteManyAsync(removed, cancellationToken);
await DeleteManyAsync(removed.Select(r => r.Id), cancellationToken);
await _store.SaveManyAsync(added, cancellationToken);
}

41
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/WorkflowStateStore.cs
View file

@ -1,7 +1,7 @@
using System.Text.Json;
using System.Text.Json.Serialization;
using EFCore.BulkExtensions;
using Elsa.Common.Services;
using Elsa.EntityFrameworkCore.Common;
using Elsa.Workflows.Core.Models;
using Elsa.Workflows.Core.Serialization;
using Elsa.Workflows.Core.Serialization.Converters;
@ -17,11 +17,13 @@ public class EFCoreWorkflowStateStore : IWorkflowStateStore
private readonly SerializerOptionsProvider _serializerOptionsProvider;
private readonly ISystemClock _systemClock;
private readonly IDbContextFactory<RuntimeElsaDbContext> _dbContextFactory;
private readonly Store<RuntimeElsaDbContext, WorkflowState> _store;
/// <summary>
/// Constructor.
/// </summary>
public EFCoreWorkflowStateStore(
Store<RuntimeElsaDbContext, WorkflowState> store,
IDbContextFactory<RuntimeElsaDbContext> dbContextFactory,
SerializerOptionsProvider serializerOptionsProvider,
ISystemClock systemClock)
@ -29,26 +31,12 @@ public class EFCoreWorkflowStateStore : IWorkflowStateStore
_serializerOptionsProvider = serializerOptionsProvider;
_systemClock = systemClock;
_dbContextFactory = dbContextFactory;
_store = store;
}
/// <inheritdoc />
public async ValueTask SaveAsync(string id, WorkflowState state, CancellationToken cancellationToken = default)
{
await using var dbContext = await _dbContextFactory.CreateDbContextAsync(cancellationToken);
var options = _serializerOptionsProvider.CreatePersistenceOptions(ReferenceHandler.Preserve);
var json = JsonSerializer.Serialize(state, options);
var now = _systemClock.UtcNow;
var entry = dbContext.Entry(state);
if (entry.Property<DateTimeOffset>("CreatedAt").CurrentValue == DateTimeOffset.MinValue)
entry.Property<DateTimeOffset>("CreatedAt").CurrentValue = now;
entry.Property<string>("Data").CurrentValue = json;
entry.Property<DateTimeOffset>("UpdatedAt").CurrentValue = now;
var entities = new[] { state };
await dbContext.BulkInsertOrUpdateAsync(entities, new BulkConfig { EnableShadowProperties = true }, cancellationToken: cancellationToken);
}
public async ValueTask SaveAsync(string id, WorkflowState state, CancellationToken cancellationToken = default) =>
await _store.SaveAsync(state, Save, cancellationToken: cancellationToken);
/// <inheritdoc />
public async ValueTask<WorkflowState?> LoadAsync(string id, CancellationToken cancellationToken = default)
@ -80,4 +68,21 @@ public class EFCoreWorkflowStateStore : IWorkflowStateStore
return await query.CountAsync(cancellationToken);
}
private WorkflowState Save(RuntimeElsaDbContext dbContext, WorkflowState entity)
{
var options = _serializerOptionsProvider.CreatePersistenceOptions(ReferenceHandler.Preserve);
var json = JsonSerializer.Serialize(entity, options);
var now = _systemClock.UtcNow;
var entry = dbContext.Entry(entity);
if (entry.Property<DateTimeOffset>("CreatedAt").CurrentValue == DateTimeOffset.MinValue)
entry.Property<DateTimeOffset>("CreatedAt").CurrentValue = now;
entry.Property<string>("Data").CurrentValue = json;
entry.Property<DateTimeOffset>("UpdatedAt").CurrentValue = now;
dbContext.Entry(entity).Property("Data").CurrentValue = json;
return entity;
}
}

8
src/modules/Elsa.Workflows.Core/State/WorkflowState.cs
View file

@ -1,4 +1,5 @@
using System.Text.Json.Serialization;
using Elsa.Common.Entities;
using Elsa.Workflows.Core.Models;
using Elsa.Workflows.Core.Serialization.Converters;
@ -7,13 +8,8 @@ namespace Elsa.Workflows.Core.State;
/// <summary>
/// Represents the current state of a workflow.
/// </summary>
public class WorkflowState
public class WorkflowState : Entity
{
/// <summary>
/// A unique workflow instance ID.
/// </summary>
public string Id { get; set; } = default!;
/// <summary>
/// The workflow definition ID.
/// </summary>