build-flow.api/tests/BuildFlow.Workflow.Tests/Reliability/ImportJobRecoveryTests.cs

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.Reliability;

/// <summary>
/// 导入任务的事务进度、崩溃恢复和优雅取消测试。
/// </summary>
[Collection(PostgresWorkflowCollection.Name)]
public class ImportJobRecoveryTests(PostgresWorkflowFixture fixture)
{
    [Fact]
    public async Task RecoveredImportStartsAtCommittedCursorWithoutDuplicatingRows()
    {
        await ResetAsync();
        var filePath = await CreateTempFileAsync();
        try
        {
            var seeded = await SeedImportAsync(filePath, successRows: 1);
            await using var db = fixture.CreateDbContext();
            var processor = CreateProcessor(
                db,
                [
                    ValidRow(2, """{"name":"first"}"""),
                    ValidRow(3, """{"name":"second"}"""),
                ]);

            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(2);
            job.FailedRows.Should().Be(0);
            job.LeaseOwner.Should().BeNull();
            (await verify.FormData.CountAsync(x => x.FormDefinitionId == seeded.FormDefinitionId))
                .Should().Be(2);
        }
        finally
        {
            File.Delete(filePath);
        }
    }

    [Fact]
    public async Task SuccessfulImportClearsLeaseAndPersistsHeartbeat()
    {
        await ResetAsync();
        var filePath = await CreateTempFileAsync();
        try
        {
            var seeded = await SeedImportAsync(filePath);
            await using var db = fixture.CreateDbContext();
            var processor = CreateProcessor(
                db,
                [ValidRow(2, """{"name":"only"}""")]);

            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.LastHeartbeatAt.Should().NotBeNull();
            job.FinishedAt.Should().NotBeNull();
            job.LeaseOwner.Should().BeNull();
            job.LeaseExpiresAt.Should().BeNull();
        }
        finally
        {
            File.Delete(filePath);
        }
    }

    [Fact]
    public async Task GracefulCancellationRequeuesJobAndKeepsCommittedProgress()
    {
        await ResetAsync();
        var filePath = await CreateTempFileAsync();
        using var cts = new CancellationTokenSource();
        try
        {
            var seeded = await SeedImportAsync(filePath);
            var rows = new CancellingRows(
                [
                    ValidRow(2, """{"name":"first"}"""),
                    ValidRow(3, """{"name":"second"}"""),
                ],
                cancelAtIndex: 1,
                cts.Cancel);
            await using var db = fixture.CreateDbContext();
            var processor = CreateProcessor(db, rows);

            var act = () => processor.ProcessAsync(
                seeded.JobId,
                seeded.LeaseOwner,
                TimeSpan.FromMinutes(5),
                1,
                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.SuccessRows.Should().Be(1);
            job.FailedRows.Should().Be(0);
            job.LeaseOwner.Should().BeNull();
            job.LeaseExpiresAt.Should().BeNull();
            (await verify.FormData.CountAsync(x => x.FormDefinitionId == seeded.FormDefinitionId))
                .Should().Be(1);
        }
        finally
        {
            File.Delete(filePath);
        }
    }

    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<SeededImport> SeedImportAsync(
        string filePath,
        int successRows = 0)
    {
        await using var db = fixture.CreateDbContext();
        var formDefinitionId = Guid.NewGuid();
        var jobId = Guid.NewGuid();
        var leaseOwner = "import-worker";
        db.FormDefinitions.Add(new FormDefinition
        {
            Id = formDefinitionId,
            Name = "导入恢复测试表单",
            Code = $"import-{formDefinitionId:N}",
            Version = 1,
            SchemaJson = """{"type":"object","properties":{}}""",
            Status = FormStatus.Published,
        });
        if (successRows > 0)
        {
            db.FormData.Add(new FormData
            {
                Id = Guid.NewGuid(),
                FormDefinitionId = formDefinitionId,
                InstanceId = Guid.NewGuid(),
                DataJson = """{"name":"already-committed"}""",
            });
        }
        db.BuildFlowImportJobs.Add(new BuildFlowImportJob
        {
            Id = jobId,
            FormDefinitionId = formDefinitionId,
            Status = BuildFlowImportJobStatus.Running,
            FileUrl = filePath,
            OriginalFileName = "recovery.xlsx",
            SuccessRows = successRows,
            FailedRows = 0,
            LeaseOwner = leaseOwner,
            LeaseExpiresAt = DateTime.UtcNow.AddMinutes(5),
            AttemptCount = 1,
            LastHeartbeatAt = DateTime.UtcNow,
            StartedAt = DateTime.UtcNow,
        });
        await db.SaveChangesAsync();
        return new SeededImport(jobId, formDefinitionId, leaseOwner);
    }

    private static ExcelParsedRow ValidRow(int rowNumber, string json)
        => new(rowNumber, json, []);

    private static ImportJobProcessor CreateProcessor(
        WorkflowDbContext db,
        IReadOnlyList<ExcelParsedRow> rows)
    {
        var derivation = new FormValueDerivationService([], [], db);
        var aggregates = new FormAggregateRefreshService(db);
        return new ImportJobProcessor(
            db,
            new StubExcelService(rows),
            new ImportBatchWriter(db, derivation, aggregates));
    }

    private static async Task<string> CreateTempFileAsync()
    {
        var path = Path.Combine(Path.GetTempPath(), $"buildflow-import-{Guid.NewGuid():N}.xlsx");
        await File.WriteAllBytesAsync(path, [1, 2, 3]);
        return path;
    }

    private sealed class StubExcelService(IReadOnlyList<ExcelParsedRow> rows)
        : IFormSchemaExcelService
    {
        public byte[] BuildTemplate(string schemaJson, string formName) => [];

        public byte[] BuildExport(
            string schemaJson,
            string formName,
            IEnumerable<string> dataJsonRows) => [];

        public IEnumerable<ExcelParsedRow> Parse(string schemaJson, byte[] xlsxBytes) => rows;
    }

    private sealed class CancellingRows(
        IReadOnlyList<ExcelParsedRow> rows,
        int cancelAtIndex,
        Action cancel) : IReadOnlyList<ExcelParsedRow>
    {
        public int Count => rows.Count;

        public ExcelParsedRow this[int index]
        {
            get
            {
                if (index == cancelAtIndex) cancel();
                return rows[index];
            }
        }

        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();
    }

    private sealed record SeededImport(
        Guid JobId,
        Guid FormDefinitionId,
        string LeaseOwner);
}