build-flow/build-flow.api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: reuse prepared form schemas

Browse Source
This commit is contained in:
xiangning 2026-07-24 08:57:03 +08:00
parent 178b441b8a
commit 943ae02859
12 changed files with 268 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/modules/workflow/Application/BuildFlow/Common/Services/ExcelParsedRow.cs Normal file
View File

@ -0,0 +1,7 @@
namespace Workflow.Application.BuildFlow.Common.Services;
/// <summary>解析后的 Excel 行。</summary>
public sealed record ExcelParsedRow(
int RowNumber,
string JsonObject,
IReadOnlyList<ExcelCellError> Errors);

9
src/modules/workflow/Application/BuildFlow/Common/Services/FormSchemaExcelService.cs
View File

@ -53,7 +53,7 @@ public class FormSchemaExcelService : IFormSchemaExcelService
}
/// <inheritdoc />
public IReadOnlyList<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes)
public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes)
{
var fields = ExtractFields(schemaJson);
using var ms = new MemoryStream(xlsxBytes);
@ -62,19 +62,16 @@ public class FormSchemaExcelService : IFormSchemaExcelService
var lastRow = ws.LastRowUsed();
var headerMap = BuildHeaderMap(ws, fields);
var result = new List<ExcelParsedRow>();
if (lastRow is null)
{
return result;
yield break;
}
for (var r = 2; r <= lastRow.RowNumber(); r++)
{
var (json, errors) = BuildRowJson(ws, r, headerMap, fields);
result.Add(new ExcelParsedRow(r, json, errors));
yield return new ExcelParsedRow(r, json, errors);
}
return result;
}
private static IReadOnlyList<FieldSummary> ExtractFields(string schemaJson)

10
src/modules/workflow/Application/BuildFlow/Common/Services/IFormSchemaExcelService.cs
View File

@ -12,12 +12,6 @@ public interface IFormSchemaExcelService
/// <summary>根据 SchemaJson 字段顺序导出已有 FormData 列表为 xlsx 字节流。</summary>
byte[] BuildExport(string schemaJson, string formName, IEnumerable<string> dataJsonRows);
/// <summary>解析上传 xlsx 字节流，按字段顺序返回每行的 JSON 对象字符串。</summary>
IReadOnlyList<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes);
/// <summary>解析上传 xlsx 字节流，按字段顺序惰性返回每行的 JSON 对象字符串。</summary>
IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes);
}
/// <summary>解析后的 Excel 行。</summary>
public sealed record ExcelParsedRow(
int RowNumber,
string JsonObject,
IReadOnlyList<ExcelCellError> Errors);

40
src/modules/workflow/Application/BuildFlow/Imports/Services/ImportJobProcessor.cs
View File

@ -62,25 +62,24 @@ public class ImportJobProcessor(
jobId,
leaseOwner,
leaseDuration,
rows.Count,
ct);
if (nextRowIndex is null) return;
if (nextRowIndex.Value > rows.Count)
{
throw new BusinessException(
ErrorCode.Business,
$"导入任务进度游标 {nextRowIndex.Value} 超过总行数 {rows.Count}");
}
var submitHandler = new SubmitFormDataCommandHandler(
db,
new FormValueDerivationService([], [], db),
new FormAggregateRefreshService(db));
for (var i = nextRowIndex.Value; i < rows.Count; i++)
var rowIndex = 0;
foreach (var row in rows)
{
ct.ThrowIfCancellationRequested();
var row = rows[i];
if (rowIndex < nextRowIndex.Value)
{
rowIndex++;
continue;
}
if (row.Errors.Count > 0)
{
var recorded = await CommitFailedRowAsync(
@ -89,8 +88,10 @@ public class ImportJobProcessor(
leaseDuration,
row.RowNumber,
row.Errors.Select(e => e.Message).ToList(),
rowIndex + 1,
ct);
if (!recorded) return;
rowIndex++;
continue;
}
@ -101,7 +102,7 @@ public class ImportJobProcessor(
leaseOwner,
leaseDuration,
job.FormDefinitionId,
i,
rowIndex,
row.JsonObject,
submitHandler,
ct);
@ -119,12 +120,22 @@ public class ImportJobProcessor(
leaseDuration,
row.RowNumber,
[ex.Message],
rowIndex + 1,
ct);
if (!recorded) return;
}
rowIndex++;
}
await CompleteJobAsync(jobId, leaseOwner, ct);
if (nextRowIndex.Value > rowIndex)
{
throw new BusinessException(
ErrorCode.Business,
$"导入任务进度游标 {nextRowIndex.Value} 超过总行数 {rowIndex}");
}
await CompleteJobAsync(jobId, leaseOwner, rowIndex, ct);
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
@ -141,7 +152,6 @@ public class ImportJobProcessor(
Guid jobId,
string leaseOwner,
TimeSpan leaseDuration,
int totalRows,
CancellationToken ct)
{
db.ChangeTracker.Clear();
@ -153,7 +163,6 @@ public class ImportJobProcessor(
return null;
}
job.TotalRows = totalRows;
RenewLease(job, leaseDuration);
await db.SaveChangesAsync(ct);
await transaction.CommitAsync(ct);
@ -184,6 +193,7 @@ public class ImportJobProcessor(
new SubmitFormDataCommand(formDefinitionId, instanceId, dataJson),
ct);
job.SuccessRows++;
job.TotalRows = Math.Max(job.TotalRows, rowIndex + 1);
RenewLease(job, leaseDuration);
await db.SaveChangesAsync(ct);
await transaction.CommitAsync(ct);
@ -196,6 +206,7 @@ public class ImportJobProcessor(
TimeSpan leaseDuration,
int rowNumber,
IReadOnlyList<string> messages,
int processedRows,
CancellationToken ct)
{
db.ChangeTracker.Clear();
@ -210,6 +221,7 @@ public class ImportJobProcessor(
var errors = DeserializeErrors(job.ErrorJson);
errors.Add(new ImportRowError(rowNumber, messages));
job.FailedRows++;
job.TotalRows = Math.Max(job.TotalRows, processedRows);
job.ErrorJson = JsonSerializer.Serialize(errors, JsonOptions);
RenewLease(job, leaseDuration);
await db.SaveChangesAsync(ct);
@ -220,6 +232,7 @@ public class ImportJobProcessor(
private async Task CompleteJobAsync(
Guid jobId,
string leaseOwner,
int totalRows,
CancellationToken ct)
{
db.ChangeTracker.Clear();
@ -237,6 +250,7 @@ public class ImportJobProcessor(
? BuildFlowImportJobStatus.Failed
: BuildFlowImportJobStatus.PartialSuccess;
job.ErrorMessage = null;
job.TotalRows = totalRows;
job.FinishedAt = DateTime.UtcNow;
job.LastHeartbeatAt = DateTime.UtcNow;
job.LeaseOwner = null;

12
src/modules/workflow/Application/Form/Derivation/FormValueDerivationService.cs
View File

@ -2,6 +2,7 @@ using System.Text.Encodings.Web;
using System.Text.Json;
using System.Text.Json.Nodes;
using Workflow.Application.Form.Relation;
using Workflow.Application.Form.Schema;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
@ -33,15 +34,22 @@ public sealed class FormValueDerivationService : IFormValueDerivationService
_db = db;
}
public async Task<FormDerivationResult> DeriveAsync(
public Task<FormDerivationResult> DeriveAsync(
string schemaJson,
string dataJson,
FormDerivationContext context,
CancellationToken ct)
=> DeriveAsync(PreparedFormSchema.Create(schemaJson), dataJson, context, ct);
public async Task<FormDerivationResult> DeriveAsync(
PreparedFormSchema preparedSchema,
string dataJson,
FormDerivationContext context,
CancellationToken ct)
{
ct.ThrowIfCancellationRequested();
var schema = JsonNode.Parse(schemaJson) as JsonObject ?? new JsonObject();
var schema = preparedSchema.Schema.DeepClone().AsObject();
var data = JsonNode.Parse(dataJson) as JsonObject ?? new JsonObject();
var derivedPaths = new List<string>();

9
src/modules/workflow/Application/Form/Derivation/IFormValueDerivationService.cs
View File

@ -1,3 +1,5 @@
using Workflow.Application.Form.Schema;
namespace Workflow.Application.Form.Derivation;
/// <summary>统一表单提交前派生值管线：计算字段、关联快照、统计字段都在这里收口。</summary>
@ -8,4 +10,11 @@ public interface IFormValueDerivationService
string dataJson,
FormDerivationContext context,
CancellationToken ct);
/// <summary>使用预解析 Schema 执行派生，供批量场景复用字段元数据。</summary>
Task<FormDerivationResult> DeriveAsync(
PreparedFormSchema schema,
string dataJson,
FormDerivationContext context,
CancellationToken ct);
}

25
src/modules/workflow/Application/Form/Schema/FormDataValidator.cs
View File

@ -33,6 +33,31 @@ public static class FormDataValidator
IReadOnlyDictionary<string, string>? nodeFieldPermissions = null)
{
var schemaValidation = SchemaValidator.Validate(schemaJson, allowedComponents);
return Validate(
schemaValidation,
dataJson,
currentNodeKey,
nodeFieldPermissions);
}
/// <summary>使用预解析 Schema 校验表单数据，避免批量场景逐行重复解析。</summary>
public static FormDataValidationResult Validate(
PreparedFormSchema schema,
string dataJson,
string? currentNodeKey = null,
IReadOnlyDictionary<string, string>? nodeFieldPermissions = null)
=> Validate(
schema.Validation,
dataJson,
currentNodeKey,
nodeFieldPermissions);
private static FormDataValidationResult Validate(
SchemaValidationResult schemaValidation,
string dataJson,
string? currentNodeKey,
IReadOnlyDictionary<string, string>? nodeFieldPermissions)
{
if (!schemaValidation.IsValid)
{
return FormDataValidationResult.Invalid(schemaValidation.Errors);

43
src/modules/workflow/Application/Form/Schema/PreparedFormSchema.cs Normal file
View File

@ -0,0 +1,43 @@
using System.Text.Json.Nodes;
namespace Workflow.Application.Form.Schema;
/// <summary>
/// 预解析表单 Schema。批量用例复用字段校验结果和 JSON 结构，避免逐行重复解析。
/// </summary>
public sealed class PreparedFormSchema
{
private PreparedFormSchema(
string schemaJson,
JsonObject schema,
SchemaValidationResult validation)
{
SchemaJson = schemaJson;
Schema = schema;
Validation = validation;
}
/// <summary>原始 Schema JSON。</summary>
public string SchemaJson { get; }
/// <summary>解析后的 Schema JSON 对象。</summary>
public JsonObject Schema { get; }
/// <summary>已提取字段元数据的 Schema 校验结果。</summary>
public SchemaValidationResult Validation { get; }
/// <summary>校验并准备一个可复用的表单 Schema。</summary>
public static PreparedFormSchema Create(string schemaJson)
{
var validation = SchemaValidator.Validate(schemaJson);
if (!validation.IsValid)
{
throw new BusinessException(
ErrorCode.InvalidParam,
$"表单 Schema 无效: {string.Join("; ", validation.Errors)}");
}
var schema = JsonNode.Parse(schemaJson) as JsonObject ?? new JsonObject();
return new PreparedFormSchema(schemaJson, schema, validation);
}
}

4
tests/BuildFlow.Workflow.Tests/BuildFlow/FormSchemaExcelServiceTests.cs
View File

@ -49,7 +49,7 @@ public class FormSchemaExcelServiceTests
using var outMs = new MemoryStream();
wb.SaveAs(outMs);
var rows = _service.Parse(SampleSchema, outMs.ToArray());
var rows = _service.Parse(SampleSchema, outMs.ToArray()).ToList();
rows.Should().HaveCount(1);
var row = rows[0];
@ -73,7 +73,7 @@ public class FormSchemaExcelServiceTests
using var outMs = new MemoryStream();
wb.SaveAs(outMs);
var rows = _service.Parse(SampleSchema, outMs.ToArray());
var rows = _service.Parse(SampleSchema, outMs.ToArray()).ToList();
rows.Should().HaveCount(1);
rows[0].Errors.Should().ContainSingle(e => e.Message.Contains("必填"));

42
tests/BuildFlow.Workflow.Tests/Performance/LazyExcelRowTests.cs Normal file
View File

@ -0,0 +1,42 @@
using ClosedXML.Excel;
using FluentAssertions;
using Workflow.Application.BuildFlow.Common.Services;
using Xunit;
namespace BuildFlow.Workflow.Tests.Performance;
/// <summary>
/// Excel 导入行惰性枚举契约测试。
/// </summary>
public class LazyExcelRowTests
{
[Fact]
public void ParseReturnsLazyEnumerableInsteadOfMaterializedList()
{
var service = new FormSchemaExcelService();
var template = service.BuildTemplate(Schema, "惰性导入");
using var input = new MemoryStream(template);
using var workbook = new XLWorkbook(input);
var worksheet = workbook.Worksheets.First();
worksheet.Cell(2, 1).Value = "first";
worksheet.Cell(3, 1).Value = "second";
worksheet.Cell(4, 1).Value = "third";
using var output = new MemoryStream();
workbook.SaveAs(output);
var rows = service.Parse(Schema, output.ToArray());
rows.Should().NotBeAssignableTo<IReadOnlyList<ExcelParsedRow>>();
rows.Take(1).Should().ContainSingle(x =>
x.RowNumber == 2 && x.JsonObject.Contains("first", StringComparison.Ordinal));
}
private const string Schema = """
{
"type": "object",
"properties": {
"name": { "type": "string", "title": "名称", "required": true, "x-component": "Input" }
}
}
""";
}

89
tests/BuildFlow.Workflow.Tests/Performance/PreparedFormSchemaTests.cs Normal file
View File

@ -0,0 +1,89 @@
using BuildFlow.Shared.Domain.Errors;
using BuildFlow.Shared.Domain.Exceptions;
using FluentAssertions;
using System.Text.Json.Nodes;
using Workflow.Application.Form.Derivation;
using Workflow.Application.Form.Schema;
using Xunit;
namespace BuildFlow.Workflow.Tests.Performance;
/// <summary>
/// 预解析表单 Schema 在批量校验和派生中的复用测试。
/// </summary>
public class PreparedFormSchemaTests
{
[Fact]
public async Task PreparedSchemaCanBeReusedForValidationAndDerivation()
{
var prepared = PreparedFormSchema.Create(ComputedSchema);
var derivation = new FormValueDerivationService([], []);
var firstValidation = FormDataValidator.Validate(
prepared,
"""{"quantity":2,"unitPrice":10}""");
var first = await derivation.DeriveAsync(
prepared,
"""{"quantity":2,"unitPrice":10}""",
new FormDerivationContext(Guid.NewGuid(), Guid.NewGuid(), null),
CancellationToken.None);
var second = await derivation.DeriveAsync(
prepared,
"""{"quantity":3,"unitPrice":5}""",
new FormDerivationContext(Guid.NewGuid(), Guid.NewGuid(), null),
CancellationToken.None);
firstValidation.IsValid.Should().BeTrue();
ReadTotal(first.DataJson).Should().Be(20);
ReadTotal(second.DataJson).Should().Be(15);
prepared.Validation.Fields.Should().HaveCount(3);
}
[Fact]
public void InvalidSchemaCannotBePrepared()
{
var act = () => PreparedFormSchema.Create("""
{
"type": "object",
"properties": {
"supplier": {
"type": "object",
"x-component": "RelationPicker",
"x-field-kind": "relation",
"x-relation": { "idField": "id" }
}
}
}
""");
var exception = act.Should().Throw<BusinessException>().Which;
exception.Code.Should().Be(ErrorCode.InvalidParam);
}
private static decimal ReadTotal(string dataJson)
=> JsonNode.Parse(dataJson)!["total"]!.GetValue<decimal>();
private const string ComputedSchema = """
{
"type": "object",
"properties": {
"quantity": { "type": "number", "x-component": "InputNumber" },
"unitPrice": { "type": "number", "x-component": "InputNumber" },
"total": {
"type": "number",
"x-component": "InputNumber",
"x-field-kind": "computed",
"x-derived": {
"mode": "computed",
"writePolicy": "overwrite",
"expression": {
"operator": "multiply",
"left": { "type": "field", "path": "quantity" },
"right": { "type": "field", "path": "unitPrice" }
}
}
}
}
}
""";
}

11
tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs
View File

@ -201,7 +201,7 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture)
string formName,
IEnumerable<string> dataJsonRows) => [];
public IReadOnlyList<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes) => rows;
public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes) => rows;
}
private sealed class CancellingRows(
@ -220,7 +220,14 @@ public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture)
}
}
public IEnumerator<ExcelParsedRow> GetEnumerator() => rows.GetEnumerator();
public IEnumerator<ExcelParsedRow> GetEnumerator()
{
for (var index = 0; index < rows.Count; index++)
{
if (index == cancelAtIndex) cancel();
yield return rows[index];
}
}
System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
=> GetEnumerator();