using BuildFlow.Workflow.Tests.Reliability; using FluentAssertions; using Microsoft.EntityFrameworkCore; using Workflow.Application.BuildFlow.Common.Services; using Workflow.Application.BuildFlow.Imports.Handlers; using Workflow.Application.BuildFlow.Imports.Services; using Workflow.Application.Form.Aggregates; using Workflow.Application.Form.Derivation; using Workflow.Domain.Entities; using Workflow.Domain.Enums; using Xunit; using Xunit.Abstractions; namespace BuildFlow.Workflow.Tests.Performance; /// /// 导入任务有界批次事务和取消恢复测试。 /// [Collection(PostgresWorkflowCollection.Name)] public class ImportJobBatchingTests( PostgresWorkflowFixture fixture, ITestOutputHelper output) { [Fact] public async Task FourHundredOneRowsUseThreeBatchCommitsAndOneCompletionCommit() { await ResetAsync(); var filePath = await CreateTempFileAsync(); try { var seeded = await SeedImportAsync(filePath); var rows = Enumerable.Range(0, 401) .Select(index => ValidRow(index + 2, $"row-{index}")) .ToArray(); var transactions = new CountingTransactionInterceptor(); await using var db = fixture.CreateDbContext(transactions); var processor = CreateProcessor(db, rows); await processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), 200, CancellationToken.None); transactions.CommitCount.Should().Be(4); output.WriteLine($"SOP_METRIC import_commit_count={transactions.CommitCount}"); await using var verify = fixture.CreateDbContext(); var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); job.Status.Should().Be(BuildFlowImportJobStatus.Success); job.TotalRows.Should().Be(401); job.SuccessRows.Should().Be(401); job.FailedRows.Should().Be(0); job.LeaseOwner.Should().BeNull(); (await verify.FormData.CountAsync(x => x.FormDefinitionId == seeded.FormDefinitionId)).Should().Be(401); } finally { System.IO.File.Delete(filePath); } } [Fact] public async Task CancellationBeforeRowTwoHundredOneKeepsFirstBatchCommitted() { await ResetAsync(); var filePath = await CreateTempFileAsync(); using var cts = new CancellationTokenSource(); try { var seeded = await SeedImportAsync(filePath); var rows = new CancellingExcelRows( Enumerable.Range(0, 401) .Select(index => ValidRow(index + 2, $"row-{index}")) .ToArray(), cancelAtIndex: 200, cts.Cancel); await using var db = fixture.CreateDbContext(); var processor = CreateProcessor(db, rows); var act = () => processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), 200, cts.Token); await act.Should().ThrowAsync(); await using var verify = fixture.CreateDbContext(); var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); job.Status.Should().Be(BuildFlowImportJobStatus.Pending); job.TotalRows.Should().Be(200); job.SuccessRows.Should().Be(200); job.FailedRows.Should().Be(0); job.LeaseOwner.Should().BeNull(); (await verify.FormData.CountAsync(x => x.FormDefinitionId == seeded.FormDefinitionId)).Should().Be(200); } finally { System.IO.File.Delete(filePath); } } [Fact] public async Task ExistingFormWithNullSchemaUsesEmptyObjectSchema() { await ResetAsync(); var filePath = await CreateTempFileAsync(); try { var seeded = await SeedImportAsync(filePath, schemaJson: null); await using var db = fixture.CreateDbContext(); var processor = CreateProcessor(db, [ValidRow(2, "row")]); await processor.ProcessAsync( seeded.JobId, seeded.LeaseOwner, TimeSpan.FromMinutes(5), 1, CancellationToken.None); await using var verify = fixture.CreateDbContext(); var job = await verify.BuildFlowImportJobs.SingleAsync(x => x.Id == seeded.JobId); job.Status.Should().Be(BuildFlowImportJobStatus.Success); job.SuccessRows.Should().Be(1); } finally { System.IO.File.Delete(filePath); } } private static ImportJobProcessor CreateProcessor( WorkflowDbContext db, IEnumerable rows) { var derivation = new FormValueDerivationService([], [], db); var aggregates = new FormAggregateRefreshService(db); return new ImportJobProcessor( db, new EnumerableFormSchemaExcelService(rows), new ImportBatchWriter(db, derivation, aggregates)); } private async Task ResetAsync() { await using var db = fixture.CreateDbContext(); await db.FormData.ExecuteDeleteAsync(); await db.BuildFlowImportJobs.ExecuteDeleteAsync(); await db.FormDefinitions.ExecuteDeleteAsync(); } private async Task<(Guid JobId, Guid FormDefinitionId, string LeaseOwner)> SeedImportAsync( string filePath, string? schemaJson = Schema) { await using var db = fixture.CreateDbContext(); var formDefinitionId = Guid.NewGuid(); var jobId = Guid.NewGuid(); const string leaseOwner = "batch-import-worker"; db.FormDefinitions.Add(new FormDefinition { Id = formDefinitionId, Name = "批量导入测试表单", Code = $"batch-import-{formDefinitionId:N}", Version = 1, SchemaJson = schemaJson, Status = FormStatus.Published, }); db.BuildFlowImportJobs.Add(new BuildFlowImportJob { Id = jobId, FormDefinitionId = formDefinitionId, Status = BuildFlowImportJobStatus.Running, FileUrl = filePath, OriginalFileName = "batch.xlsx", LeaseOwner = leaseOwner, LeaseExpiresAt = DateTime.UtcNow.AddMinutes(5), AttemptCount = 1, LastHeartbeatAt = DateTime.UtcNow, StartedAt = DateTime.UtcNow, }); await db.SaveChangesAsync(); return (jobId, formDefinitionId, leaseOwner); } private static ExcelParsedRow ValidRow(int rowNumber, string name) => new(rowNumber, $$"""{"name":"{{name}}"}""", []); private static async Task CreateTempFileAsync() { var path = Path.Combine(Path.GetTempPath(), $"buildflow-import-{Guid.NewGuid():N}.xlsx"); await System.IO.File.WriteAllBytesAsync(path, [1, 2, 3]); return path; } private const string Schema = """ { "type": "object", "properties": { "name": { "type": "string", "required": true, "x-component": "Input" } } } """; }