build-flow.api/tests/BuildFlow.Workflow.Tests/Performance/ImportJobBatchingTests.cs

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;

namespace BuildFlow.Workflow.Tests.Performance;

/// <summary>
/// 导入任务有界批次事务和取消恢复测试。
/// </summary>
[Collection(PostgresWorkflowCollection.Name)]
public class ImportJobBatchingTests(PostgresWorkflowFixture fixture)
{
    [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);
            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<OperationCanceledException>();
            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<ExcelParsedRow> 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<string> 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" }
      }
    }
    """;
}