feat: add transactional import batch writer

This commit is contained in:
xiangning 2026-07-24 09:00:09 +08:00
parent 943ae02859
commit 4bedd17741
5 changed files with 221 additions and 0 deletions

View File

@ -0,0 +1,6 @@
using Workflow.Application.BuildFlow.Common.Services;
namespace Workflow.Application.BuildFlow.Imports.Services;
/// <summary>导入批次中的一行及其从零开始的全局行索引。</summary>
public sealed record ImportBatchItem(int RowIndex, ExcelParsedRow Row);

View File

@ -0,0 +1,6 @@
namespace Workflow.Application.BuildFlow.Imports.Services;
/// <summary>导入批次写入结果,包含成功数量和逐行错误。</summary>
public sealed record ImportBatchResult(
int SuccessCount,
IReadOnlyList<ImportRowError> Errors);

View File

@ -0,0 +1,102 @@
using System.Buffers.Binary;
using System.Security.Cryptography;
using Workflow.Application.Form.Aggregates;
using Workflow.Application.Form.Derivation;
using Workflow.Application.Form.Schema;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.BuildFlow.Imports.Services;
/// <summary>
/// 导入批次写入用例。复用预解析 Schema,批量保存表单行,并在批次末统一刷新聚合字段。
/// </summary>
public sealed class ImportBatchWriter(
WorkflowDbContext db,
IFormValueDerivationService derivationService,
IFormAggregateRefreshService aggregateRefreshService)
{
/// <summary>写入一个已受事务保护的导入批次。</summary>
public async Task<ImportBatchResult> WriteAsync(
Guid jobId,
Guid formDefinitionId,
PreparedFormSchema schema,
IReadOnlyList<ImportBatchItem> items,
CancellationToken ct)
{
var errors = new List<ImportRowError>();
var formDataRows = new List<Domain.Entities.FormData>();
var instanceIds = new List<Guid>();
foreach (var item in items)
{
ct.ThrowIfCancellationRequested();
if (item.Row.Errors.Count > 0)
{
errors.Add(new ImportRowError(
item.Row.RowNumber,
item.Row.Errors.Select(x => x.Message).ToList()));
continue;
}
try
{
var instanceId = DeterministicInstanceId(jobId, item.RowIndex);
var derivation = await derivationService.DeriveAsync(
schema,
item.Row.JsonObject,
new FormDerivationContext(formDefinitionId, instanceId, null),
ct);
var validation = FormDataValidator.Validate(schema, derivation.DataJson);
if (!validation.IsValid)
{
errors.Add(new ImportRowError(item.Row.RowNumber, validation.Errors));
continue;
}
formDataRows.Add(new Domain.Entities.FormData
{
Id = Guid.NewGuid(),
FormDefinitionId = formDefinitionId,
InstanceId = instanceId,
DataJson = derivation.DataJson,
});
instanceIds.Add(instanceId);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
errors.Add(new ImportRowError(item.Row.RowNumber, [ex.Message]));
}
}
if (formDataRows.Count > 0)
{
db.FormData.AddRange(formDataRows);
await db.SaveChangesAsync(ct);
var distinctInstanceIds = instanceIds.Distinct().ToArray();
await aggregateRefreshService.RefreshBatchAsync(
formDefinitionId,
distinctInstanceIds,
ct);
await aggregateRefreshService.RefreshAffectedBatchAsync(
formDefinitionId,
distinctInstanceIds,
ct);
}
return new ImportBatchResult(formDataRows.Count, errors);
}
private static Guid DeterministicInstanceId(Guid jobId, int rowIndex)
{
Span<byte> input = stackalloc byte[20];
jobId.TryWriteBytes(input[..16]);
BinaryPrimitives.WriteInt32LittleEndian(input[16..], rowIndex);
Span<byte> hash = stackalloc byte[32];
SHA256.HashData(input, hash);
return new Guid(hash[..16]);
}
}

View File

@ -28,6 +28,7 @@ public class WorkflowApplicationModule : AbpModule
services.AddScoped<IFormAggregateRefreshService, FormAggregateRefreshService>();
services.AddScoped<IBuildFlowButtonActionExecutor, BuildFlowButtonActionExecutor>();
services.AddScoped<IFormSchemaExcelService, FormSchemaExcelService>();
services.AddScoped<ImportBatchWriter>();
services.AddScoped<IImportJobProcessor, Workflow.Application.BuildFlow.Imports.Handlers.ImportJobProcessor>();
// 条件评估器:ProcessEngine 依赖它做条件节点分支判断。无参构造内置对比器责任链。

View File

@ -0,0 +1,106 @@
using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using NSubstitute;
using Workflow.Application.BuildFlow.Common.Services;
using Workflow.Application.BuildFlow.Imports.Services;
using Workflow.Application.Form.Aggregates;
using Workflow.Application.Form.Derivation;
using Workflow.Application.Form.Schema;
using Xunit;
namespace BuildFlow.Workflow.Tests.Performance;
/// <summary>
/// 导入批次写入、错误隔离和确定性实例 ID 测试。
/// </summary>
public class ImportBatchWriterTests
{
[Fact]
public async Task WritesValidRowsAndReturnsInvalidRowsWithoutPerRowAggregateCalls()
{
await using var db = CreateDbContext();
var aggregates = Substitute.For<IFormAggregateRefreshService>();
var writer = new ImportBatchWriter(
db,
new FormValueDerivationService([], []),
aggregates);
var formDefinitionId = Guid.NewGuid();
var prepared = PreparedFormSchema.Create(Schema);
var result = await writer.WriteAsync(
Guid.NewGuid(),
formDefinitionId,
prepared,
[
new ImportBatchItem(0, ValidRow(2, "first")),
new ImportBatchItem(
1,
new ExcelParsedRow(
3,
"{}",
[new ExcelCellError("name", "必填字段为空")])),
new ImportBatchItem(2, ValidRow(4, "third")),
],
CancellationToken.None);
result.SuccessCount.Should().Be(2);
result.Errors.Should().ContainSingle(x => x.Row == 3);
(await db.FormData.CountAsync()).Should().Be(2);
await aggregates.Received(1).RefreshBatchAsync(
formDefinitionId,
Arg.Is<IReadOnlyCollection<Guid>>(ids => ids.Count == 2),
Arg.Any<CancellationToken>());
await aggregates.Received(1).RefreshAffectedBatchAsync(
formDefinitionId,
Arg.Is<IReadOnlyCollection<Guid>>(ids => ids.Count == 2),
Arg.Any<CancellationToken>());
}
[Fact]
public async Task UsesFullJobAndRowIndexForDeterministicInstanceIds()
{
await using var db = CreateDbContext();
var aggregates = Substitute.For<IFormAggregateRefreshService>();
var writer = new ImportBatchWriter(
db,
new FormValueDerivationService([], []),
aggregates);
await writer.WriteAsync(
Guid.Parse("8e6fcacf-3bc1-4d5e-84c4-92be4208855c"),
Guid.NewGuid(),
PreparedFormSchema.Create(Schema),
[
new ImportBatchItem(1, ValidRow(2, "first")),
new ImportBatchItem(65537, ValidRow(3, "second")),
],
CancellationToken.None);
var instanceIds = await db.FormData
.Select(x => x.InstanceId)
.ToListAsync();
instanceIds.Should().OnlyHaveUniqueItems();
}
private static WorkflowDbContext CreateDbContext()
{
var options = new DbContextOptionsBuilder<WorkflowDbContext>()
.UseInMemoryDatabase($"import-batch-writer-{Guid.NewGuid():N}")
.Options;
var db = new WorkflowDbContext(options);
db.SavingChanges += AuditStub.OnSave;
return db;
}
private static ExcelParsedRow ValidRow(int rowNumber, string name)
=> new(rowNumber, $$"""{"name":"{{name}}"}""", []);
private const string Schema = """
{
"type": "object",
"properties": {
"name": { "type": "string", "required": true, "x-component": "Input" }
}
}
""";
}