build-flow.api/tests/BuildFlow.Monitoring.Tests/Sync/MonitoringUpsertTests.cs

using FluentAssertions;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Monitoring.Application.Sync;
using Monitoring.Domain.Entities.ExternalSystem;
using Monitoring.Domain.Enums;
using Monitoring.Domain.Services;
using Monitoring.Domain.Values;

namespace BuildFlow.Monitoring.Tests.Sync;

/// <summary>Monitoring 同步记录批量 UPSERT、并发幂等和日志安全集成测试。</summary>
[Collection(PostgresMonitoringCollection.Name)]
public sealed class MonitoringUpsertTests(PostgresMonitoringFixture fixture)
{
    [Fact]
    public async Task Persists1001RecordsInThreeParameterizedUpsertsWithoutRecordSelects()
    {
        var interceptor = new CountingUpsertInterceptor();
        await using var db = fixture.CreateDbContext(interceptor);
        var system = CreateSystem("batch-upsert");
        db.ExternalSystems.Add(system);
        await db.SaveChangesAsync();

        var records = Enumerable.Range(0, 1001)
            .Select(index => new SyncedRecordData(
                $"order-{index}",
                "order",
                $$"""{"value":{{index}}}""",
                DateTime.UtcNow.AddMinutes(index)))
            .ToArray();
        using var services = CreateServices(new StaticExternalConnector(CreateResult(records)));
        var processor = new SyncProcessor(db, services, new RecordingLogger<SyncProcessor>());

        var job = await processor.ExecuteAsync(system, "test", CancellationToken.None);
        var upsertCount = interceptor.UpsertCommandCount;
        var selectCount = interceptor.RecordSelectCount;
        var allParameterized = interceptor.AllUpsertsParameterized;
        db.ChangeTracker.Clear();
        var persistedCount = await db.SyncedRecords.CountAsync(x => x.ExternalSystemId == system.Id);

        job.Status.Should().Be(SyncJobStatusEnum.Succeeded);
        persistedCount.Should().Be(1001);
        upsertCount.Should().Be(3);
        selectCount.Should().Be(0);
        allParameterized.Should().BeTrue();
    }

    [Fact]
    public async Task ConcurrentSynchronizationsUpsertOneCoherentLatestRecord()
    {
        var systemId = Guid.NewGuid();
        await using (var seedDb = fixture.CreateDbContext())
        {
            seedDb.ExternalSystems.Add(CreateSystem("concurrent", systemId));
            await seedDb.SaveChangesAsync();
        }

        var occurredAtA = new DateTime(2026, 7, 23, 1, 0, 0, DateTimeKind.Utc);
        var occurredAtB = occurredAtA.AddMinutes(1);
        var interceptor = new CountingUpsertInterceptor(concurrentSelectParticipants: 2);
        await using var dbA = fixture.CreateDbContext(interceptor);
        await using var dbB = fixture.CreateDbContext(interceptor);
        var systemA = await dbA.ExternalSystems.SingleAsync(x => x.Id == systemId);
        var systemB = await dbB.ExternalSystems.SingleAsync(x => x.Id == systemId);
        using var servicesA = CreateServices(new StaticExternalConnector(CreateResult(
            [new SyncedRecordData("shared-order", "order", """{"source":"A"}""", occurredAtA)])));
        using var servicesB = CreateServices(new StaticExternalConnector(CreateResult(
            [new SyncedRecordData("shared-order", "order", """{"source":"B"}""", occurredAtB)])));
        var processorA = new SyncProcessor(dbA, servicesA, new RecordingLogger<SyncProcessor>());
        var processorB = new SyncProcessor(dbB, servicesB, new RecordingLogger<SyncProcessor>());

        var jobs = await Task.WhenAll(
            processorA.ExecuteAsync(systemA, "concurrent-a", CancellationToken.None),
            processorB.ExecuteAsync(systemB, "concurrent-b", CancellationToken.None));

        jobs.Should().OnlyContain(job => job.Status == SyncJobStatusEnum.Succeeded);
        await using var verifyDb = fixture.CreateDbContext();
        var records = await verifyDb.SyncedRecords
            .Where(x => x.ExternalSystemId == systemId)
            .ToListAsync();
        records.Should().ContainSingle();
        var record = records.Single();
        new[] { jobs[0].Id, jobs[1].Id }.Should().Contain(record.SyncJobId);
        if (record.SyncJobId == jobs[0].Id)
        {
            record.OccurredAt.Should().Be(occurredAtA);
            record.Data.Payload["source"]?.ToString().Should().Be("A");
        }
        else
        {
            record.OccurredAt.Should().Be(occurredAtB);
            record.Data.Payload["source"]?.ToString().Should().Be("B");
        }
    }

    [Fact]
    public async Task SequentialSynchronizationReplacesJobTimeAndPayload()
    {
        await using var db = fixture.CreateDbContext();
        var system = CreateSystem("sequential");
        db.ExternalSystems.Add(system);
        await db.SaveChangesAsync();

        var firstTime = new DateTime(2026, 7, 23, 2, 0, 0, DateTimeKind.Utc);
        var secondTime = firstTime.AddHours(1);
        using var firstServices = CreateServices(new StaticExternalConnector(CreateResult(
            [new SyncedRecordData("updated-order", "order", """{"version":"first"}""", firstTime)])));
        var firstJob = await new SyncProcessor(db, firstServices, new RecordingLogger<SyncProcessor>())
            .ExecuteAsync(system, "first", CancellationToken.None);
        using var secondServices = CreateServices(new StaticExternalConnector(CreateResult(
            [new SyncedRecordData("updated-order", "order", """{"version":"second"}""", secondTime)])));

        var secondJob = await new SyncProcessor(db, secondServices, new RecordingLogger<SyncProcessor>())
            .ExecuteAsync(system, "second", CancellationToken.None);

        db.ChangeTracker.Clear();
        var record = await db.SyncedRecords.SingleAsync(x =>
            x.ExternalSystemId == system.Id && x.ExternalId == "updated-order");
        firstJob.Status.Should().Be(SyncJobStatusEnum.Succeeded);
        secondJob.Status.Should().Be(SyncJobStatusEnum.Succeeded);
        record.SyncJobId.Should().Be(secondJob.Id);
        record.OccurredAt.Should().Be(secondTime);
        record.Data.Payload["version"]?.ToString().Should().Be("second");
    }

    [Fact]
    public async Task InvalidJsonPreservesRawPayloadAndLogsIdentifiersOnly()
    {
        const string rawPayload = "invalid-json-sensitive-payload";
        const string externalId = "invalid-order";
        const string recordType = "order";
        await using var db = fixture.CreateDbContext();
        var system = CreateSystem("invalid-json");
        db.ExternalSystems.Add(system);
        await db.SaveChangesAsync();
        using var services = CreateServices(new StaticExternalConnector(CreateResult(
            [new SyncedRecordData(externalId, recordType, rawPayload, DateTime.UtcNow)])));
        var logger = new RecordingLogger<SyncProcessor>();
        var processor = new SyncProcessor(db, services, logger);

        var job = await processor.ExecuteAsync(system, "invalid-json", CancellationToken.None);

        db.ChangeTracker.Clear();
        var record = await db.SyncedRecords.SingleAsync(x => x.ExternalSystemId == system.Id);
        var warnings = logger.Entries.Where(entry => entry.Level == LogLevel.Warning).ToArray();
        job.Status.Should().Be(SyncJobStatusEnum.Succeeded);
        record.Data.Payload["_raw"]?.ToString().Should().Be(rawPayload);
        warnings.Should().ContainSingle();
        warnings[0].Message.Should().Contain(system.Id.ToString());
        warnings[0].Message.Should().Contain(externalId);
        warnings[0].Message.Should().Contain(recordType);
        warnings[0].Message.Should().NotContain(rawPayload);
    }

    private static ExternalSystemEntity CreateSystem(string name, Guid? id = null)
    {
        var system = new ExternalSystemEntity
        {
            Data = new SystemData
            {
                Name = name,
                SystemType = ExternalSystemTypeEnum.GenericHttp,
                BaseUrl = "https://example.test",
            },
        };
        if (id.HasValue)
        {
            system.Id = id.Value;
        }

        return system;
    }

    private static FetchResult CreateResult(IReadOnlyList<SyncedRecordData> records)
        => new(records.Count, new Dictionary<string, object?>(), records);

    private static ServiceProvider CreateServices(IExternalConnector connector)
        => new ServiceCollection()
            .AddSingleton(connector)
            .AddSingleton<IExternalConnector>(connector)
            .BuildServiceProvider();
}