diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchItem.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchItem.cs new file mode 100644 index 0000000..bf7c62a --- /dev/null +++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchItem.cs @@ -0,0 +1,6 @@ +using Workflow.Application.BuildFlow.Common.Services; + +namespace Workflow.Application.BuildFlow.Imports.Services; + +/// 导入批次中的一行及其从零开始的全局行索引。 +public sealed record ImportBatchItem(int RowIndex, ExcelParsedRow Row); diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchResult.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchResult.cs new file mode 100644 index 0000000..b7c0dae --- /dev/null +++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchResult.cs @@ -0,0 +1,6 @@ +namespace Workflow.Application.BuildFlow.Imports.Services; + +/// 导入批次写入结果，包含成功数量和逐行错误。 +public sealed record ImportBatchResult( + int SuccessCount, + IReadOnlyList Errors); diff --git a/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchWriter.cs b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchWriter.cs new file mode 100644 index 0000000..060f4b1 --- /dev/null +++ b/src/modules/workflow/Application/BuildFlow/Imports/Services/ImportBatchWriter.cs @@ -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; + +/// +/// 导入批次写入用例。复用预解析 Schema，批量保存表单行，并在批次末统一刷新聚合字段。 +/// +public sealed class ImportBatchWriter( + WorkflowDbContext db, + IFormValueDerivationService derivationService, + IFormAggregateRefreshService aggregateRefreshService) +{ + /// 写入一个已受事务保护的导入批次。 + public async Task WriteAsync( + Guid jobId, + Guid formDefinitionId, + PreparedFormSchema schema, + IReadOnlyList items, + CancellationToken ct) + { + var errors = new List(); + var formDataRows = new List(); + var instanceIds = new List(); + + 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 input = stackalloc byte[20]; + jobId.TryWriteBytes(input[..16]); + BinaryPrimitives.WriteInt32LittleEndian(input[16..], rowIndex); + Span hash = stackalloc byte[32]; + SHA256.HashData(input, hash); + return new Guid(hash[..16]); + } +} diff --git a/src/modules/workflow/Application/WorkflowApplicationModule.cs b/src/modules/workflow/Application/WorkflowApplicationModule.cs index b15c7f8..662b282 100644 --- a/src/modules/workflow/Application/WorkflowApplicationModule.cs +++ b/src/modules/workflow/Application/WorkflowApplicationModule.cs @@ -28,6 +28,7 @@ public class WorkflowApplicationModule : AbpModule services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); services.AddScoped(); // 条件评估器：ProcessEngine 依赖它做条件节点分支判断。无参构造内置对比器责任链。 diff --git a/tests/BuildFlow.Workflow.Tests/Performance/ImportBatchWriterTests.cs b/tests/BuildFlow.Workflow.Tests/Performance/ImportBatchWriterTests.cs new file mode 100644 index 0000000..ad2939d --- /dev/null +++ b/tests/BuildFlow.Workflow.Tests/Performance/ImportBatchWriterTests.cs @@ -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; + +/// +/// 导入批次写入、错误隔离和确定性实例 ID 测试。 +/// +public class ImportBatchWriterTests +{ + [Fact] + public async Task WritesValidRowsAndReturnsInvalidRowsWithoutPerRowAggregateCalls() + { + await using var db = CreateDbContext(); + var aggregates = Substitute.For(); + 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>(ids => ids.Count == 2), + Arg.Any()); + await aggregates.Received(1).RefreshAffectedBatchAsync( + formDefinitionId, + Arg.Is>(ids => ids.Count == 2), + Arg.Any()); + } + + [Fact] + public async Task UsesFullJobAndRowIndexForDeterministicInstanceIds() + { + await using var db = CreateDbContext(); + var aggregates = Substitute.For(); + 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() + .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" } + } + } + """; +}